wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-08 07:12:55 +08:00
Code Issues Packages Projects Releases Wiki Activity

ENH: add ffill checkpointing to blaze core loader

Browse Source
This commit is contained in:
Joe Jevnik
2016-05-26 18:36:18 -04:00
parent 4c2f0e86eb
commit c8cf5a6761
3 changed files with 282 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/pipeline/test_blaze.py
+136 -30
View File
@@ -26,12 +26,12 @@ from zipline.pipeline.engine import SimplePipelineEngine
from zipline.pipeline.loaders.blaze import (
from_blaze,
BlazeLoader,
NoDeltasWarning,
NoMetaDataWarning,
)
from zipline.pipeline.loaders.blaze.core import (
NonPipelineField,
no_deltas_rules,
)
from zipline.testing import parameter_space
from zipline.testing.fixtures import WithAssetFinder
from zipline.utils.numpy_utils import (
float64_dtype,
@@ -112,7 +112,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
self.assertEqual(ds.__name__, name)
@@ -129,7 +130,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
ds,
@@ -141,7 +143,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
value = from_blaze(
expr.value,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
self.assertEqual(value.name, 'value')
@@ -153,7 +156,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr.value,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
value,
@@ -162,7 +166,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
).value,
value,
@@ -173,7 +178,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
),
value.dataset,
@@ -196,32 +202,49 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
)
self.assertIn("'asof_date'", str(e.exception))
self.assertIn(repr(str(expr.dshape.measure)), str(e.exception))
def test_auto_deltas(self):
@parameter_space(deltas={True, False}, checkpoints={True, False})
def test_auto_metadata(self, deltas, checkpoints):
select_level = op.getitem(('ignore', 'raise'))
m = {'ds': self.df}
if deltas:
m['ds_deltas'] = pd.DataFrame(columns=self.df.columns),
if checkpoints:
m['ds_checkpoints'] = pd.DataFrame(columns=self.df.columns),
expr = bz.data(
{'ds': self.df,
'ds_deltas': pd.DataFrame(columns=self.df.columns)},
dshape=var * Record((
('ds', self.dshape.measure),
('ds_deltas', self.dshape.measure),
)),
m,
dshape=var * Record((k, self.dshape.measure) for k in m),
)
loader = BlazeLoader()
ds = from_blaze(
expr.ds,
loader=loader,
missing_values=self.missing_values,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
)
self.assertEqual(len(loader), 1)
exprdata = loader[ds]
self.assertTrue(exprdata.expr.isidentical(expr.ds))
self.assertTrue(exprdata.deltas.isidentical(expr.ds_deltas))
if deltas:
self.assertTrue(exprdata.deltas.isidentical(expr.ds_deltas))
else:
self.assertIsNone(exprdata.deltas)
if checkpoints:
self.assertTrue(
exprdata.checkpoints.isidentical(expr.ds_checkpoints),
)
else:
self.assertIsNone(exprdata.checkpoints)
def test_auto_deltas_fail_warn(self):
@parameter_space(deltas={True, False}, checkpoints={True, False})
def test_auto_metadata_fail_warn(self, deltas, checkpoints):
select_level = op.getitem(('ignore', 'warn'))
with warnings.catch_warnings(record=True) as ws:
warnings.simplefilter('always')
loader = BlazeLoader()
@@ -229,22 +252,31 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.warn,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
missing_values=self.missing_values,
)
self.assertEqual(len(ws), 1)
w = ws[0].message
self.assertIsInstance(w, NoDeltasWarning)
self.assertIn(str(expr), str(w))
self.assertEqual(len(ws), deltas + checkpoints)
def test_auto_deltas_fail_raise(self):
for w in ws:
w = w.message
self.assertIsInstance(w, NoMetaDataWarning)
self.assertIn(str(expr), str(w))
@parameter_space(deltas={True, False}, checkpoints={True, False})
def test_auto_metadata_fail_raise(self, deltas, checkpoints):
if not (deltas or checkpoints):
# not a real case
return
select_level = op.getitem(('ignore', 'raise'))
loader = BlazeLoader()
expr = bz.data(self.df, dshape=self.dshape)
with self.assertRaises(ValueError) as e:
from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.raise_,
no_deltas_rule=select_level(deltas),
no_checkpoints_rule=select_level(checkpoints),
)
self.assertIn(str(expr), str(e.exception))
@@ -261,7 +293,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=self.garbage_loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
)
with self.assertRaises(AttributeError):
ds.a
@@ -550,6 +583,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
deltas=None,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
with self.assertRaises(TypeError):
@@ -558,6 +592,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
deltas=None,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
deltas = bz.data(
@@ -570,6 +605,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
deltas=deltas,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
with self.assertRaises(TypeError):
@@ -578,6 +614,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
deltas=deltas,
loader=self.garbage_loader,
missing_values=self.missing_values,
no_checkpoints_rule='ignore',
)
def _test_id(self, df, dshape, expected, finder, add):
@@ -586,7 +623,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -617,7 +655,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
loader=loader,
no_deltas_rule=no_deltas_rules.ignore,
no_deltas_rule='ignore',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -1044,6 +1083,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
def _run_pipeline(self,
expr,
deltas,
checkpoints,
expected_views,
expected_output,
finder,
@@ -1056,8 +1096,10 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
ds = from_blaze(
expr,
deltas,
checkpoints,
loader=loader,
no_deltas_rule=no_deltas_rules.raise_,
no_deltas_rule='raise',
no_checkpoints_rule='ignore',
missing_values=self.missing_values,
)
p = Pipeline()
@@ -1070,7 +1112,11 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
window_length = window_length_
def compute(self, today, assets, out, data):
assert_array_almost_equal(data, expected_views[today])
assert_array_almost_equal(
data,
expected_views[today],
err_msg=str(today),
)
out[:] = compute_fn(data)
p.add(TestFactor(), 'value')
@@ -1142,6 +1188,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1194,6 +1241,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1237,6 +1285,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1311,6 +1360,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1371,6 +1421,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
self._run_pipeline(
expr,
deltas,
None,
expected_views,
expected_output,
finder,
@@ -1380,3 +1431,58 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
window_length=3,
compute_fn=op.itemgetter(-1),
)
def test_checkpoints(self):
dates = pd.Timestamp('2014-01-01'), pd.Timestamp('2014-01-04')
baseline = pd.DataFrame({
'value': [-1.0, 1.0],
'asof_date': dates,
'timestamp': dates,
})
checkpoints_ts = pd.Timestamp('2014-01-02')
checkpoints = pd.DataFrame({
'value': [0.0],
'asof_date': checkpoints_ts,
'timestamp': checkpoints_ts,
})
asset_info = asset_infos[0][0]
nassets = len(asset_info)
expected_views = keymap(pd.Timestamp, {
'2014-01-03': repeat_last_axis(
np.array([0.0]),
nassets,
),
'2014-01-04': repeat_last_axis(
np.array([1.0]),
nassets,
),
})
with tmp_asset_finder(equities=asset_info) as finder:
expected_output = pd.DataFrame(
list(concatv([0.0] * nassets, [1.0] * nassets)),
index=pd.MultiIndex.from_product((
sorted(expected_views.keys()),
finder.retrieve_all(asset_info.index),
)),
columns=('value',),
)
self._run_pipeline(
bz.data(baseline, name='expr', dshape=self.macro_dshape),
None,
bz.data(
checkpoints,
name='expr_checkpoints',
dshape=self.macro_dshape,
),
expected_views,
expected_output,
finder,
calendar=pd.date_range('2014-01-01', '2014-01-04'),
start=checkpoints_ts + pd.Timedelta('1 days'),
end=dates[-1],
window_length=1,
compute_fn=op.itemgetter(-1),
)
zipline/pipeline/loaders/blaze/__init__.py
+2 -2
View File
@@ -1,6 +1,6 @@
from .core import (
BlazeLoader,
NoDeltasWarning,
NoMetaDataWarning,
from_blaze,
global_loader,
)
@@ -9,5 +9,5 @@ __all__ = (
'BlazeLoader',
'from_blaze',
'global_loader',
'NoDeltasWarning',
'NoMetaDataWarning',
)
zipline/pipeline/loaders/blaze/core.py
+144 -63
View File
@@ -127,6 +127,7 @@ from __future__ import division, absolute_import
from abc import ABCMeta, abstractproperty
from collections import namedtuple, defaultdict
from copy import copy
import datetime
from functools import partial
from itertools import count
import warnings
@@ -169,7 +170,6 @@ from zipline.pipeline.loaders.utils import (
from zipline.pipeline.term import NotSpecified
from zipline.lib.adjusted_array import AdjustedArray, can_represent_dtype
from zipline.lib.adjustment import Float64Overwrite
from zipline.utils.enum import enum
from zipline.utils.input_validation import (
expect_element,
ensure_timezone,
@@ -196,7 +196,7 @@ is_invalid_deltas_node = complement(flip(isinstance, valid_deltas_node_types))
get__name__ = op.attrgetter('__name__')
class ExprData(namedtuple('ExprData', 'expr deltas odo_kwargs')):
class ExprData(namedtuple('ExprData', 'expr deltas checkpoints odo_kwargs')):
"""A pair of expressions and data resources. The expresions will be
computed using the resources as the starting scope.
@@ -206,14 +206,17 @@ class ExprData(namedtuple('ExprData', 'expr deltas odo_kwargs')):
The baseline values.
deltas : Expr, optional
The deltas for the data.
checkpoints : Expr, optional
The forward fill checkpoints for the data.
odo_kwargs : dict, optional
The keyword arguments to forward to the odo calls internally.
"""
def __new__(cls, expr, deltas=None, odo_kwargs=None):
def __new__(cls, expr, deltas=None, checkpoints=None, odo_kwargs=None):
return super(ExprData, cls).__new__(
cls,
expr,
deltas,
checkpoints,
odo_kwargs or {},
)
@@ -224,6 +227,7 @@ class ExprData(namedtuple('ExprData', 'expr deltas odo_kwargs')):
return super(ExprData, cls).__repr__(cls(
str(self.expr),
str(self.deltas),
str(self.checkpoint),
self.odo_kwargs,
))
@@ -411,58 +415,66 @@ def _check_datetime_field(name, measure):
)
class NoDeltasWarning(UserWarning):
"""Warning used to signal that no deltas could be found and none
were provided.
class NoMetaDataWarning(UserWarning):
"""Warning used to signal that no deltas or checkpoints could be found and
none were provided.
Parameters
----------
expr : Expr
The expression that was searched.
field : {'deltas', 'checkpoints'}
The field that was looked up.
"""
def __init__(self, expr):
def __init__(self, expr, field):
self._expr = expr
self._field = field
def __str__(self):
return 'No deltas could be inferred from expr: %s' % self._expr
return 'No %s could be inferred from expr: %s' % (
self._field,
self._expr,
)
no_deltas_rules = enum('warn', 'raise_', 'ignore')
no_metadata_rules = frozenset({'warn', 'raise', 'ignore'})
def get_deltas(expr, deltas, no_deltas_rule):
"""Find the correct deltas for the expression.
def _get_metadata(field, expr, metadata_expr, no_metadata_rule):
"""Find the correct metadata expression for the expression.
Parameters
----------
field : {'deltas', 'checkpoints'}
The kind of metadata expr to lookup.
expr : Expr
The baseline expression.
deltas : Expr, 'auto', or None
The deltas argument. If this is 'auto', then the deltas table will
metadata_expr : Expr, 'auto', or None
The metadata argument. If this is 'auto', then the metadata table will
be searched for by walking up the expression tree. If this cannot be
reflected, then an action will be taken based on the
``no_deltas_rule``.
no_deltas_rule : no_deltas_rule
How to handle the case where deltas='auto' but no deltas could be
found.
``no_metadata_rule``.
no_metadata_rule : {'warn', 'raise', 'ignore'}
How to handle the case where the metadata_expr='auto' but no expr
could be found.
Returns
-------
deltas : Expr or None
metadata : Expr or None
The deltas table to use.
"""
if isinstance(deltas, bz.Expr) or deltas != 'auto':
return deltas
if isinstance(metadata_expr, bz.Expr) or metadata_expr != 'auto':
return metadata_expr
try:
return expr._child[(expr._name or '') + '_deltas']
return expr._child['_'.join(((expr._name or ''), field))]
except (ValueError, AttributeError):
if no_deltas_rule == no_deltas_rules.raise_:
if no_metadata_rule == 'raise':
raise ValueError(
"no deltas table could be reflected for %s" % expr
"no %s table could be reflected for %s" % (field, expr)
)
elif no_deltas_rule == no_deltas_rules.warn:
warnings.warn(NoDeltasWarning(expr))
elif no_metadata_rule == 'warn':
warnings.warn(NoMetaDataWarning(expr, field), stacklevel=4)
return None
@@ -502,26 +514,37 @@ def _ensure_timestamp_field(dataset_expr, deltas):
return dataset_expr, deltas
@expect_element(no_deltas_rule=no_deltas_rules)
@expect_element(
no_deltas_rule=no_metadata_rules,
no_checkpoints_rule=no_metadata_rules,
)
def from_blaze(expr,
deltas='auto',
checkpoints='auto',
loader=None,
resources=None,
odo_kwargs=None,
missing_values=None,
no_deltas_rule=no_deltas_rules.warn):
no_deltas_rule='warn',
no_checkpoints_rule='warn'):
"""Create a Pipeline API object from a blaze expression.
Parameters
----------
expr : Expr
The blaze expression to use.
deltas : Expr or 'auto', optional
deltas : Expr, 'auto' or None, optional
The expression to use for the point in time adjustments.
If the string 'auto' is passed, a deltas expr will be looked up
by stepping up the expression tree and looking for another field
with the name of ``expr`` + '_deltas'. If None is passed, no deltas
will be used.
with the name of ``expr._name`` + '_deltas'. If None is passed, no
deltas will be used.
deltas : Expr, 'auto' or None, optional
The expression to use for the forward fill checkpoints.
If the string 'auto' is passed, a checkpoints expr will be looked up
by stepping up the expression tree and looking for another field
with the name of ``expr._name`` + '_checkpoints'. If None is passed,
no checkpoints will be used.
loader : BlazeLoader, optional
The blaze loader to attach this pipeline dataset to. If None is passed,
the global blaze loader is used.
@@ -533,11 +556,16 @@ def from_blaze(expr,
missing_values : dict[str -> any], optional
A dict mapping column names to missing values for those columns.
Missing values are required for integral columns.
no_deltas_rule : no_deltas_rule
no_deltas_rule : {'warn', 'raise', 'ignore'}, optional
What should happen if ``deltas='auto'`` but no deltas can be found.
'warn' says to raise a warning but continue.
'raise' says to raise an exception if no deltas can be found.
'ignore' says take no action and proceed with no deltas.
no_checkpoints_rule : {'warn', 'raise', 'ignore'}, optional
What should happen if ``checkpoints='auto'`` but no checkpoints can be
found. 'warn' says to raise a warning but continue.
'raise' says to raise an exception if no deltas can be found.
'ignore' says take no action and proceed with no deltas.
Returns
-------
@@ -548,13 +576,28 @@ def from_blaze(expr,
is passed, a ``BoundColumn`` on the dataset that would be constructed
from passing the parent is returned.
"""
deltas = get_deltas(expr, deltas, no_deltas_rule)
if deltas is not None:
deltas = _get_metadata(
'deltas',
expr,
deltas,
no_deltas_rule,
)
checkpoints = _get_metadata(
'checkpoints',
expr,
checkpoints,
no_checkpoints_rule,
)
if 'auto' in {deltas, checkpoints}:
invalid_nodes = tuple(filter(is_invalid_deltas_node, expr._subterms()))
if invalid_nodes:
raise TypeError(
'expression with deltas may only contain (%s) nodes,'
'expression with %s may only contain (%s) nodes,'
" found: %s" % (
' or '.join(
['deltas'] if deltas is not None else [] +
['checkpoints'] if checkpoints is not None else [],
),
', '.join(map(get__name__, valid_deltas_node_types)),
', '.join(
set(map(compose(get__name__, type), invalid_nodes)),
@@ -632,6 +675,9 @@ def from_blaze(expr,
bind_expression_to_resources(deltas, resources)
if deltas is not None else
None,
bind_expression_to_resources(checkpoints, resources)
if checkpoints is not None else
None,
odo_kwargs=odo_kwargs,
)
if single_column is not None:
@@ -669,10 +715,12 @@ def overwrite_novel_deltas(baseline, deltas, dates):
) <= 1
novel_deltas = deltas.loc[novel_idx]
non_novel_deltas = deltas.loc[~novel_idx]
return sort_values(pd.concat(
cat = pd.concat(
(baseline, novel_deltas),
ignore_index=True,
), TS_FIELD_NAME), non_novel_deltas
)
sort_values(cat, TS_FIELD_NAME, inplace=True)
return cat, non_novel_deltas
def overwrite_from_dates(asof, dense_dates, sparse_dates, asset_idx, value):
@@ -822,6 +870,31 @@ def adjustments_from_deltas_with_sids(dense_dates,
return dict(adjustments) # no subclasses of dict
def _checkpoint_ts(lower_dt):
"""Given a lower time bound for a query, get the date in the checkpoint
table to query for.
Parameters
----------
lower_dt : datetime
The lower time bound for the query.
Returns
-------
checkpoint_ts : pd.Timestamp
The date in the checkpoint table to query for.
"""
date = lower_dt.date()
return pd.Timestamp.combine(
date.replace(
day=1,
month=(date.month - 2) % 12 + 1,
year=date.year - 1 if date.month == 1 else date.year,
),
datetime.time(0),
).tz_localize('US/Eastern')
class BlazeLoader(dict):
"""A PipelineLoader for datasets constructed with ``from_blaze``.
@@ -873,13 +946,14 @@ class BlazeLoader(dict):
except ValueError:
raise AssertionError('all columns must come from the same dataset')
expr, deltas, odo_kwargs = self[dataset]
expr, deltas, checkpoints, odo_kwargs = self[dataset]
have_sids = SID_FIELD_NAME in expr.fields
asset_idx = pd.Series(index=assets, data=np.arange(len(assets)))
assets = list(map(int, assets)) # coerce from numpy.int64
added_query_fields = [AD_FIELD_NAME, TS_FIELD_NAME] + (
[SID_FIELD_NAME] if have_sids else []
)
colnames = added_query_fields + list(map(getname, columns))
data_query_time = self._data_query_time
data_query_tz = self._data_query_tz
@@ -890,30 +964,15 @@ class BlazeLoader(dict):
data_query_tz,
)
def where(e):
"""Create the query to run against the resources.
Parameters
----------
e : Expr
The baseline or deltas expression.
Returns
-------
q : Expr
The query to run.
"""
return e[
(e[TS_FIELD_NAME] <= upper_dt)
][added_query_fields + list(map(getname, columns))]
def collect_expr(e):
def collect_expr(e, lower):
"""Materialize the expression as a dataframe.
Parameters
----------
e : Expr
The baseline or deltas expression.
lower : datetime
The lower time bound to query.
Returns
-------
@@ -925,17 +984,39 @@ class BlazeLoader(dict):
This can return more data than needed. The in memory reindex will
handle this.
"""
df = odo(where(e), pd.DataFrame, **odo_kwargs)
df.sort(TS_FIELD_NAME, inplace=True) # sort for the groupby later
return df
predicate = e[TS_FIELD_NAME] <= upper_dt
if lower is not None:
predicate &= e[TS_FIELD_NAME] >= lower
materialized_expr = collect_expr(expr)
return odo(e[predicate][colnames], pd.DataFrame, **odo_kwargs)
if checkpoints is not None:
ts = checkpoints[TS_FIELD_NAME]
checkpoints_ts = odo(ts[ts <= lower_dt].max(), pd.Timestamp)
if pd.isnull(checkpoints_ts):
materialized_checkpoints = pd.DataFrame(columns=colnames)
lower = lower_dt
else:
materialized_checkpoints = odo(
checkpoints[ts == checkpoints_ts][colnames],
pd.DataFrame,
**odo_kwargs
)
lower = checkpoints_ts
else:
materialized_checkpoints = pd.DataFrame(columns=colnames)
lower = None
materialized_expr = collect_expr(expr, lower)
if materialized_checkpoints is not None:
materialized_expr = pd.concat((
materialized_checkpoints,
materialized_expr,
))
materialized_deltas = (
collect_expr(deltas)
collect_expr(deltas, lower)
if deltas is not None else
pd.DataFrame(
columns=added_query_fields + list(map(getname, columns)),
)
pd.DataFrame(columns=colnames)
)
# It's not guaranteed that assets returned by the engine will contain