diff --git a/docs/source/whatsnew/1.0.2.txt b/docs/source/whatsnew/1.0.2.txt
new file mode 100644
index 00000000..348dc076
--- /dev/null
+++ b/docs/source/whatsnew/1.0.2.txt
@@ -0,0 +1,23 @@
+Release 1.0.2
+-------------
+
+:Release: 1.0.2
+:Date: TBD
+
+Enhancements
+~~~~~~~~~~~~
+
+- Adds forward fill checkpoint tables for the blaze core loader. This allow the
+  loader to more efficiently forward fill the data by capping the lower date it
+  must search for when querying data. The checkpoints should have novel deltas
+  applied (:issue:`1276`).
+
+Bug Fixes
+~~~~~~~~~
+
+None
+
+Documentation
+~~~~~~~~~~~~~
+
+None
diff --git a/tests/pipeline/test_blaze.py b/tests/pipeline/test_blaze.py
index 742a6b70..3a26cb7e 100644
--- a/tests/pipeline/test_blaze.py
+++ b/tests/pipeline/test_blaze.py
@@ -21,16 +21,21 @@ from toolz.curried import operator as op
 
 from zipline.assets.synthetic import make_simple_equity_info
 from zipline.pipeline import Pipeline, CustomFactor
-from zipline.pipeline.data import DataSet, BoundColumn
+from zipline.pipeline.data import DataSet, BoundColumn, Column
 from zipline.pipeline.engine import SimplePipelineEngine
 from zipline.pipeline.loaders.blaze import (
     from_blaze,
     BlazeLoader,
-    NoDeltasWarning,
+    NoMetaDataWarning,
 )
 from zipline.pipeline.loaders.blaze.core import (
+    ExprData,
     NonPipelineField,
-    no_deltas_rules,
+)
+from zipline.testing import (
+    ZiplineTestCase,
+    parameter_space,
+    tmp_asset_finder,
 )
 from zipline.testing.fixtures import WithAssetFinder
 from zipline.utils.numpy_utils import (
@@ -38,7 +43,7 @@ from zipline.utils.numpy_utils import (
     int64_dtype,
     repeat_last_axis,
 )
-from zipline.testing import tmp_asset_finder, ZiplineTestCase
+from zipline.testing.predicates import assert_equal, assert_isidentical
 
 nameof = op.attrgetter('name')
 dtypeof = op.attrgetter('dtype')
@@ -54,6 +59,7 @@ asset_infos = (
         pd.Timestamp('2015'),
     ),),
 )
+simple_asset_info = asset_infos[0][0]
 with_extra_sid = parameterized.expand(asset_infos)
 with_ignore_sid = parameterized.expand(
     product(chain.from_iterable(asset_infos), [True, False])
@@ -106,13 +112,21 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
         cls.garbage_loader = BlazeLoader()
         cls.missing_values = {'int_value': 0}
 
+        cls.value_dshape = dshape("""var * {
+            sid: ?int64,
+            value: float64,
+            asof_date: datetime,
+            timestamp: datetime,
+        }""")
+
     def test_tabular(self):
         name = 'expr'
         expr = bz.data(self.df, name=name, dshape=self.dshape)
         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 +143,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 +156,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 +169,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 +179,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 +191,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,
@@ -184,9 +203,8 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
         expr = bz.data(
             self.df.loc[:, ['sid', 'value', 'timestamp']],
             name='expr',
-            dshape="""
-            var * {
-                sid: ?int64,
+            dshape="""var * {
+                sid: int64,
                 value: float64,
                 timestamp: datetime,
             }""",
@@ -196,32 +214,150 @@ 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):
+    def test_missing_timestamp(self):
         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),
-            )),
+            self.df.loc[:, ['sid', 'value', 'asof_date']],
+            name='expr',
+            dshape="""var * {
+                sid: int64,
+                value: float64,
+                asof_date: datetime,
+            }""",
+        )
+
+        loader = BlazeLoader()
+
+        from_blaze(
+            expr,
+            loader=loader,
+            no_deltas_rule='ignore',
+            no_checkpoints_rule='ignore',
+        )
+
+        self.assertEqual(len(loader), 1)
+        exprdata, = loader.values()
+
+        assert_isidentical(
+            exprdata.expr,
+            bz.transform(expr, timestamp=expr.asof_date),
+        )
+
+    def test_from_blaze_no_resources_dataset_expr(self):
+        expr = bz.symbol('expr', self.dshape)
+
+        with self.assertRaises(ValueError) as e:
+            from_blaze(
+                expr,
+                loader=self.garbage_loader,
+                no_deltas_rule='ignore',
+                no_checkpoints_rule='ignore',
+                missing_values=self.missing_values,
+            )
+        assert_equal(
+            str(e.exception),
+            'no resources provided to compute expr',
+        )
+
+    @parameter_space(metadata={'deltas', 'checkpoints'})
+    def test_from_blaze_no_resources_metadata_expr(self, metadata):
+        expr = bz.data(self.df, name='expr', dshape=self.dshape)
+        metadata_expr = bz.symbol('metadata', self.dshape)
+
+        with self.assertRaises(ValueError) as e:
+            from_blaze(
+                expr,
+                loader=self.garbage_loader,
+                no_deltas_rule='ignore',
+                no_checkpoints_rule='ignore',
+                missing_values=self.missing_values,
+                **{metadata: metadata_expr}
+            )
+        assert_equal(
+            str(e.exception),
+            'no resources provided to compute %s' % metadata,
+        )
+
+    def test_from_blaze_mixed_resources_dataset_expr(self):
+        expr = bz.data(self.df, name='expr', dshape=self.dshape)
+
+        with self.assertRaises(ValueError) as e:
+            from_blaze(
+                expr,
+                resources={expr: self.df},
+                loader=self.garbage_loader,
+                no_deltas_rule='ignore',
+                no_checkpoints_rule='ignore',
+                missing_values=self.missing_values,
+            )
+        assert_equal(
+            str(e.exception),
+            'explicit and implicit resources provided to compute expr',
+        )
+
+    @parameter_space(metadata={'deltas', 'checkpoints'})
+    def test_from_blaze_mixed_resources_metadata_expr(self, metadata):
+        expr = bz.symbol('expr', self.dshape)
+        metadata_expr = bz.data(self.df, name=metadata, dshape=self.dshape)
+
+        with self.assertRaises(ValueError) as e:
+            from_blaze(
+                expr,
+                resources={metadata_expr: self.df},
+                loader=self.garbage_loader,
+                no_deltas_rule='ignore',
+                no_checkpoints_rule='ignore',
+                missing_values=self.missing_values,
+                **{metadata: metadata_expr}
+            )
+        assert_equal(
+            str(e.exception),
+            'explicit and implicit resources provided to compute %s' %
+            metadata,
+        )
+
+    @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(
+            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 +365,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 +406,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
@@ -540,45 +686,71 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
         )
 
     def test_complex_expr(self):
-        expr = bz.data(self.df, dshape=self.dshape)
+        expr = bz.data(self.df, dshape=self.dshape, name='expr')
         # put an Add in the table
         expr_with_add = bz.transform(expr, value=expr.value + 1)
 
-        # Test that we can have complex expressions with no deltas
+        # test that we can have complex expressions with no metadata
         from_blaze(
             expr_with_add,
             deltas=None,
+            checkpoints=None,
             loader=self.garbage_loader,
             missing_values=self.missing_values,
+            no_checkpoints_rule='ignore',
         )
 
-        with self.assertRaises(TypeError):
+        with self.assertRaises(TypeError) as e:
+            # test that we cannot create a single column from a non field
             from_blaze(
                 expr.value + 1,  # put an Add in the column
                 deltas=None,
+                checkpoints=None,
                 loader=self.garbage_loader,
                 missing_values=self.missing_values,
+                no_checkpoints_rule='ignore',
             )
+        assert_equal(
+            str(e.exception),
+            "expression 'expr.value + 1' was array-like but not a simple field"
+            " of some larger table",
+        )
 
         deltas = bz.data(
             pd.DataFrame(columns=self.df.columns),
             dshape=self.dshape,
+            name='deltas',
+        )
+        checkpoints = bz.data(
+            pd.DataFrame(columns=self.df.columns),
+            dshape=self.dshape,
+            name='checkpoints',
         )
-        with self.assertRaises(TypeError):
-            from_blaze(
-                expr_with_add,
-                deltas=deltas,
-                loader=self.garbage_loader,
-                missing_values=self.missing_values,
-            )
 
-        with self.assertRaises(TypeError):
+        # test that we can have complex expressions with explicit metadata
+        from_blaze(
+            expr_with_add,
+            deltas=deltas,
+            checkpoints=checkpoints,
+            loader=self.garbage_loader,
+            missing_values=self.missing_values,
+        )
+
+        with self.assertRaises(TypeError) as e:
+            # test that we cannot create a single column from a non field
+            # even with explicit metadata
             from_blaze(
                 expr.value + 1,
                 deltas=deltas,
+                checkpoints=checkpoints,
                 loader=self.garbage_loader,
                 missing_values=self.missing_values,
             )
+        assert_equal(
+            str(e.exception),
+            "expression 'expr.value + 1' was array-like but not a simple field"
+            " of some larger table",
+        )
 
     def _test_id(self, df, dshape, expected, finder, add):
         expr = bz.data(df, name='expr', dshape=dshape)
@@ -586,7 +758,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 +790,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()
@@ -809,13 +983,12 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
                    Equity(66 [B])      2
                    Equity(67 [C])      2
         """
-        asset_info = asset_infos[0][0]
-        nassets = len(asset_info)
+        nassets = len(simple_asset_info)
         expected = pd.DataFrame(
             list(concatv([0] * nassets, [1] * nassets, [2] * nassets)),
             index=pd.MultiIndex.from_product((
                 self.macro_df.timestamp,
-                self.asset_finder.retrieve_all(asset_info.index),
+                self.asset_finder.retrieve_all(simple_asset_info.index),
             )),
             columns=('value',),
         )
@@ -907,15 +1080,14 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
         fields = OrderedDict(self.macro_dshape.measure.fields)
         fields['other'] = fields['value']
 
-        asset_info = asset_infos[0][0]
-        with tmp_asset_finder(equities=asset_info) as finder:
+        with tmp_asset_finder(equities=simple_asset_info) as finder:
             expected = pd.DataFrame(
                 np.array([[0, 1],
                           [1, 2],
                           [2, 3]]).repeat(3, axis=0),
                 index=pd.MultiIndex.from_product((
                     df.timestamp,
-                    finder.retrieve_all(asset_info.index),
+                    finder.retrieve_all(simple_asset_info.index),
                 )),
                 columns=('value', 'other'),
             ).sort_index(axis=1)
@@ -1044,6 +1216,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
     def _run_pipeline(self,
                       expr,
                       deltas,
+                      checkpoints,
                       expected_views,
                       expected_output,
                       finder,
@@ -1056,8 +1229,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 +1245,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 +1321,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             self._run_pipeline(
                 expr,
                 deltas,
+                None,
                 expected_views,
                 expected_output,
                 finder,
@@ -1194,6 +1374,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             self._run_pipeline(
                 expr,
                 deltas,
+                None,
                 expected_views,
                 expected_output,
                 finder,
@@ -1205,7 +1386,6 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             )
 
     def test_deltas_macro(self):
-        asset_info = asset_infos[0][0]
         expr = bz.data(self.macro_df, name='expr', dshape=self.macro_dshape)
         deltas = bz.data(
             self.macro_df.iloc[:-1],
@@ -1218,18 +1398,18 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             timestamp=deltas.timestamp + timedelta(days=1),
         )
 
-        nassets = len(asset_info)
+        nassets = len(simple_asset_info)
         expected_views = keymap(pd.Timestamp, {
             '2014-01-02': repeat_last_axis(np.array([10.0, 1.0]), nassets),
             '2014-01-03': repeat_last_axis(np.array([11.0, 2.0]), nassets),
         })
 
-        with tmp_asset_finder(equities=asset_info) as finder:
+        with tmp_asset_finder(equities=simple_asset_info) as finder:
             expected_output = pd.DataFrame(
                 list(concatv([10] * nassets, [11] * nassets)),
                 index=pd.MultiIndex.from_product((
                     sorted(expected_views.keys()),
-                    finder.retrieve_all(asset_info.index),
+                    finder.retrieve_all(simple_asset_info.index),
                 )),
                 columns=('value',),
             )
@@ -1237,6 +1417,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             self._run_pipeline(
                 expr,
                 deltas,
+                None,
                 expected_views,
                 expected_output,
                 finder,
@@ -1311,6 +1492,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             self._run_pipeline(
                 expr,
                 deltas,
+                None,
                 expected_views,
                 expected_output,
                 finder,
@@ -1322,7 +1504,6 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             )
 
     def test_novel_deltas_macro(self):
-        asset_info = asset_infos[0][0]
         base_dates = pd.DatetimeIndex([
             pd.Timestamp('2014-01-01'),
             pd.Timestamp('2014-01-04')
@@ -1340,7 +1521,7 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             timestamp=deltas.timestamp + timedelta(days=1),
         )
 
-        nassets = len(asset_info)
+        nassets = len(simple_asset_info)
         expected_views = keymap(pd.Timestamp, {
             '2014-01-03': repeat_last_axis(
                 np.array([10.0, 10.0, 10.0]),
@@ -1359,18 +1540,19 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
             # omitting the 4th and 5th to simulate a weekend
             pd.Timestamp('2014-01-06'),
         ])
-        with tmp_asset_finder(equities=asset_info) as finder:
+        with tmp_asset_finder(equities=simple_asset_info) as finder:
             expected_output = pd.DataFrame(
                 list(concatv([10] * nassets, [11] * nassets)),
                 index=pd.MultiIndex.from_product((
                     sorted(expected_views.keys()),
-                    finder.retrieve_all(asset_info.index),
+                    finder.retrieve_all(simple_asset_info.index),
                 )),
                 columns=('value',),
             )
             self._run_pipeline(
                 expr,
                 deltas,
+                None,
                 expected_views,
                 expected_output,
                 finder,
@@ -1380,3 +1562,242 @@ class BlazeToPipelineTestCase(WithAssetFinder, ZiplineTestCase):
                 window_length=3,
                 compute_fn=op.itemgetter(-1),
             )
+
+    def _test_checkpoints_macro(self, checkpoints, ffilled_value=-1.0):
+        """Simple checkpoints test that accepts a checkpoints dataframe and
+        the expected value for 2014-01-03 for macro datasets.
+
+        The underlying data has value -1.0 on 2014-01-01 and 1.0 on 2014-01-04.
+
+        Parameters
+        ----------
+        checkpoints : pd.DataFrame
+            The checkpoints data.
+        ffilled_value : float, optional
+            The value to be read on the third, if not provided, it will be the
+            value in the base data that will be naturally ffilled there.
+        """
+        dates = pd.Timestamp('2014-01-01'), pd.Timestamp('2014-01-04')
+        baseline = pd.DataFrame({
+            'value': [-1.0, 1.0],
+            'asof_date': dates,
+            'timestamp': dates,
+        })
+
+        nassets = len(simple_asset_info)
+        expected_views = keymap(pd.Timestamp, {
+            '2014-01-03': repeat_last_axis(
+                np.array([ffilled_value]),
+                nassets,
+            ),
+            '2014-01-04': repeat_last_axis(
+                np.array([1.0]),
+                nassets,
+            ),
+        })
+
+        with tmp_asset_finder(equities=simple_asset_info) as finder:
+            expected_output = pd.DataFrame(
+                list(concatv([ffilled_value] * nassets, [1.0] * nassets)),
+                index=pd.MultiIndex.from_product((
+                    sorted(expected_views.keys()),
+                    finder.retrieve_all(simple_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=pd.Timestamp('2014-01-03'),
+                end=dates[-1],
+                window_length=1,
+                compute_fn=op.itemgetter(-1),
+            )
+
+    def test_checkpoints_macro(self):
+        ffilled_value = 0.0
+
+        checkpoints_ts = pd.Timestamp('2014-01-02')
+        checkpoints = pd.DataFrame({
+            'value': [ffilled_value],
+            'asof_date': checkpoints_ts,
+            'timestamp': checkpoints_ts,
+        })
+
+        self._test_checkpoints_macro(checkpoints, ffilled_value)
+
+    def test_empty_checkpoints_macro(self):
+        empty_checkpoints = pd.DataFrame({
+            'value': [],
+            'asof_date': [],
+            'timestamp': [],
+        })
+
+        self._test_checkpoints_macro(empty_checkpoints)
+
+    def test_checkpoints_out_of_bounds_macro(self):
+        # provide two checkpoints, one before the data in the base table
+        # and one after, these should not affect the value on the third
+        dates = pd.to_datetime(['2013-12-31', '2014-01-05'])
+        checkpoints = pd.DataFrame({
+            'value': [-2, 2],
+            'asof_date': dates,
+            'timestamp': dates,
+        })
+
+        self._test_checkpoints_macro(checkpoints)
+
+    def _test_checkpoints(self, checkpoints, ffilled_values=None):
+        """Simple checkpoints test that accepts a checkpoints dataframe and
+        the expected value for 2014-01-03.
+
+        The underlying data has value -1.0 on 2014-01-01 and 1.0 on 2014-01-04.
+
+        Parameters
+        ----------
+        checkpoints : pd.DataFrame
+            The checkpoints data.
+        ffilled_value : float, optional
+            The value to be read on the third, if not provided, it will be the
+            value in the base data that will be naturally ffilled there.
+        """
+        nassets = len(simple_asset_info)
+
+        dates = pd.to_datetime(['2014-01-01', '2014-01-04'])
+        dates_repeated = np.tile(dates, nassets)
+        values = np.arange(nassets) + 1
+        values = np.hstack((values[::-1], values))
+        baseline = pd.DataFrame({
+            'sid': np.tile(simple_asset_info.index, 2),
+            'value': values,
+            'asof_date': dates_repeated,
+            'timestamp': dates_repeated,
+        })
+
+        if ffilled_values is None:
+            ffilled_values = baseline.value.iloc[:nassets]
+
+        updated_values = baseline.value.iloc[nassets:]
+
+        expected_views = keymap(pd.Timestamp, {
+            '2014-01-03': [ffilled_values],
+            '2014-01-04': [updated_values],
+        })
+
+        with tmp_asset_finder(equities=simple_asset_info) as finder:
+            expected_output = pd.DataFrame(
+                list(concatv(ffilled_values, updated_values)),
+                index=pd.MultiIndex.from_product((
+                    sorted(expected_views.keys()),
+                    finder.retrieve_all(simple_asset_info.index),
+                )),
+                columns=('value',),
+            )
+
+            self._run_pipeline(
+                bz.data(baseline, name='expr', dshape=self.value_dshape),
+                None,
+                bz.data(
+                    checkpoints,
+                    name='expr_checkpoints',
+                    dshape=self.value_dshape,
+                ),
+                expected_views,
+                expected_output,
+                finder,
+                calendar=pd.date_range('2014-01-01', '2014-01-04'),
+                start=pd.Timestamp('2014-01-03'),
+                end=dates[-1],
+                window_length=1,
+                compute_fn=op.itemgetter(-1),
+            )
+
+    def test_checkpoints(self):
+        nassets = len(simple_asset_info)
+        ffilled_values = (np.arange(nassets, dtype=np.float64) + 1) * 10
+        dates = [pd.Timestamp('2014-01-02')] * nassets
+        checkpoints = pd.DataFrame({
+            'sid': simple_asset_info.index,
+            'value': ffilled_values,
+            'asof_date': dates,
+            'timestamp': dates,
+        })
+
+        self._test_checkpoints(checkpoints, ffilled_values)
+
+    def test_empty_checkpoints(self):
+        checkpoints = pd.DataFrame({
+            'sid': [],
+            'value': [],
+            'asof_date': [],
+            'timestamp': [],
+        })
+
+        self._test_checkpoints(checkpoints)
+
+    def test_checkpoints_out_of_bounds(self):
+        nassets = len(simple_asset_info)
+        # provide two sets of checkpoints, one before the data in the base
+        # table and one after, these should not affect the value on the third
+        dates = pd.to_datetime(['2013-12-31', '2014-01-05'])
+        dates_repeated = np.tile(dates, nassets)
+        ffilled_values = (np.arange(nassets) + 2) * 10
+        ffilled_values = np.hstack((ffilled_values[::-1], ffilled_values))
+        checkpoints = pd.DataFrame({
+            'sid': np.tile(simple_asset_info.index, 2),
+            'value': ffilled_values,
+            'asof_date': dates_repeated,
+            'timestamp': dates_repeated,
+        })
+
+        self._test_checkpoints(checkpoints)
+
+
+class MiscTestCase(ZiplineTestCase):
+    def test_exprdata_repr(self):
+        strd = set()
+
+        class BadRepr(object):
+            """A class which cannot be repr'd.
+            """
+            def __init__(self, name):
+                self._name = name
+
+            def __repr__(self):  # pragma: no cover
+                raise AssertionError('ayy')
+
+            def __str__(self):
+                strd.add(self)
+                return self._name
+
+        assert_equal(
+            repr(ExprData(
+                expr=BadRepr('expr'),
+                deltas=BadRepr('deltas'),
+                checkpoints=BadRepr('checkpoints'),
+                odo_kwargs={'a': 'b'},
+            )),
+            "ExprData(expr='expr', deltas='deltas',"
+            " checkpoints='checkpoints', odo_kwargs={'a': 'b'})",
+        )
+
+    def test_blaze_loader_repr(self):
+        assert_equal(repr(BlazeLoader()), '<BlazeLoader: {}>')
+
+    def test_blaze_loader_lookup_failure(self):
+        class D(DataSet):
+            c = Column(dtype='float64')
+
+        with self.assertRaises(KeyError) as e:
+            BlazeLoader()(D.c)
+        assert_equal(str(e.exception), 'D.c::float64')
diff --git a/zipline/pipeline/loaders/blaze/__init__.py b/zipline/pipeline/loaders/blaze/__init__.py
index ec88396e..5f283b8f 100644
--- a/zipline/pipeline/loaders/blaze/__init__.py
+++ b/zipline/pipeline/loaders/blaze/__init__.py
@@ -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',
 )
diff --git a/zipline/pipeline/loaders/blaze/core.py b/zipline/pipeline/loaders/blaze/core.py
index 6fbe3a73..12832616 100644
--- a/zipline/pipeline/loaders/blaze/core.py
+++ b/zipline/pipeline/loaders/blaze/core.py
@@ -1,4 +1,5 @@
-"""Blaze integration with the Pipeline API.
+"""
+Blaze integration with the Pipeline API.
 
 For an overview of the blaze project, see blaze.pydata.org
 
@@ -81,6 +82,18 @@ actually 3. By pulling our data into these two tables and not silently updating
 our original table we can run our pipelines using the information we would
 have had on that day, and we can prevent lookahead bias in the pipelines.
 
+
+Another optional expression that may be provided is ``checkpoints``. The
+``checkpoints`` expression is used when doing a forward fill query to cap the
+lower date that must be searched. This expression has the same shape as the
+``baseline`` and ``deltas`` expressions but should be downsampled with novel
+deltas applied. For example, imagine we had one data point per asset per day
+for some dataset. We could dramatically speed up our queries by pre populating
+a downsampled version which has the most recently known value at the start of
+each month. Then, when we query, we only must look back at most one month
+before the start of the pipeline query to provide enough data to forward fill
+correctly.
+
 Conversion from Blaze to the Pipeline API
 -----------------------------------------
 
@@ -169,7 +182,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 +208,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 +218,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 +239,7 @@ class ExprData(namedtuple('ExprData', 'expr deltas odo_kwargs')):
         return super(ExprData, cls).__repr__(cls(
             str(self.expr),
             str(self.deltas),
+            str(self.checkpoints),
             self.odo_kwargs,
         ))
 
@@ -240,7 +256,7 @@ class InvalidField(with_metaclass(ABCMeta)):
         The shape of the field.
     """
     @abstractproperty
-    def error_format(self):
+    def error_format(self):  # pragma: no cover
         raise NotImplementedError('error_format')
 
     def __init__(self, field, type_):
@@ -265,14 +281,6 @@ class NonPipelineField(InvalidField):
     )
 
 
-class NotPipelineCompatible(TypeError):
-    """Exception used to indicate that a dshape is not Pipeline API
-    compatible.
-    """
-    def __str__(self):
-        return "'%s' is a non Pipeline API compatible type'" % self.args
-
-
 _new_names = ('BlazeDataSet_%d' % n for n in count())
 
 
@@ -352,7 +360,7 @@ def new_dataset(expr, deltas, missing_values):
 
     # unicode is a name error in py3 but the branch is only hit
     # when we are in python 2.
-    if PY2 and isinstance(name, unicode):  # noqa
+    if PY2 and isinstance(name, unicode):  # pragma: no cover # noqa
         name = name.encode('utf-8')
 
     return type(name, (DataSet,), columns)
@@ -411,62 +419,90 @@ 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
-        The deltas table to use.
+    metadata : Expr or None
+        The deltas or metadata table to use.
     """
-    if isinstance(deltas, bz.Expr) or deltas != 'auto':
-        return deltas
+    if isinstance(metadata_expr, bz.Expr) or metadata_expr is None:
+        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
 
 
-def _ensure_timestamp_field(dataset_expr, deltas):
+def _ad_as_ts(expr):
+    """Duplicate the asof_date column as the timestamp column.
+
+    Parameters
+    ----------
+    expr : Expr or None
+        The expression to change the columns of.
+
+    Returns
+    -------
+    transformed : Expr or None
+        The transformed expression or None if ``expr`` is None.
+    """
+    return (
+        None
+        if expr is None else
+        bz.transform(expr, **{TS_FIELD_NAME: expr[AD_FIELD_NAME]})
+    )
+
+
+def _ensure_timestamp_field(dataset_expr, deltas, checkpoints):
     """Verify that the baseline and deltas expressions have a timestamp field.
 
     If there is not a ``TS_FIELD_NAME`` on either of the expressions, it will
@@ -479,6 +515,8 @@ def _ensure_timestamp_field(dataset_expr, deltas):
         The baseline expression.
     deltas : Expr or None
         The deltas expression if any was provided.
+    checkpoints : Expr or None
+        The checkpoints expression if any was provided.
 
     Returns
     -------
@@ -491,37 +529,45 @@ def _ensure_timestamp_field(dataset_expr, deltas):
             dataset_expr,
             **{TS_FIELD_NAME: dataset_expr[AD_FIELD_NAME]}
         )
-        if deltas is not None:
-            deltas = bz.transform(
-                deltas,
-                **{TS_FIELD_NAME: deltas[AD_FIELD_NAME]}
-            )
+        deltas = _ad_as_ts(deltas)
+        checkpoints = _ad_as_ts(checkpoints)
     else:
         _check_datetime_field(TS_FIELD_NAME, measure)
 
-    return dataset_expr, deltas
+    return dataset_expr, deltas, checkpoints
 
 
-@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 +579,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,19 +599,34 @@ 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:
+    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 auto %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)),
                     ),
                 ),
             )
+    deltas = _get_metadata(
+        'deltas',
+        expr,
+        deltas,
+        no_deltas_rule,
+    )
+    checkpoints = _get_metadata(
+        'checkpoints',
+        expr,
+        checkpoints,
+        no_checkpoints_rule,
+    )
 
     # Check if this is a single column out of a dataset.
     if bz.ndim(expr) != 1:
@@ -606,7 +672,11 @@ def from_blaze(expr,
             ),
         )
     _check_datetime_field(AD_FIELD_NAME, measure)
-    dataset_expr, deltas = _ensure_timestamp_field(dataset_expr, deltas)
+    dataset_expr, deltas, checkpoints = _ensure_timestamp_field(
+        dataset_expr,
+        deltas,
+        checkpoints,
+    )
 
     if deltas is not None and (sorted(deltas.dshape.measure.fields) !=
                                sorted(measure.fields)):
@@ -616,10 +686,20 @@ def from_blaze(expr,
                 deltas.dshape.measure,
             ),
         )
+    if (checkpoints is not None and
+        (sorted(checkpoints.dshape.measure.fields) !=
+         sorted(measure.fields))):
+        raise TypeError(
+            'baseline measure != checkpoints measure:\n%s != %s' % (
+                measure,
+                checkpoints.dshape.measure,
+            ),
+        )
 
     # Ensure that we have a data resource to execute the query against.
-    _check_resources('dataset_expr', dataset_expr, resources)
+    _check_resources('expr', dataset_expr, resources)
     _check_resources('deltas', deltas, resources)
+    _check_resources('checkpoints', checkpoints, resources)
 
     # Create or retrieve the Pipeline API dataset.
     if missing_values is None:
@@ -632,6 +712,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 +752,13 @@ 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
+        copy=False,
+    )
+    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):
@@ -859,6 +945,12 @@ class BlazeLoader(dict):
             return self
         raise KeyError(column)
 
+    def __repr__(self):
+        return '<%s: %s>' % (
+            type(self).__name__,
+            super(BlazeLoader, self).__repr__(),
+        )
+
     def load_adjusted_array(self, columns, dates, assets, mask):
         return dict(
             concat(map(
@@ -873,13 +965,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 +983,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 +1003,43 @@ 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)
-        materialized_deltas = (
-            collect_expr(deltas)
-            if deltas is not None else
-            pd.DataFrame(
-                columns=added_query_fields + list(map(getname, columns)),
+            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 = None
+            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,
+                ),
+                ignore_index=True,
+                copy=False,
             )
+        materialized_deltas = (
+            collect_expr(deltas, lower)
+            if deltas is not None else
+            pd.DataFrame(columns=colnames)
         )
 
         # It's not guaranteed that assets returned by the engine will contain
diff --git a/zipline/testing/predicates.py b/zipline/testing/predicates.py
index 3bc35346..ed1c472c 100644
--- a/zipline/testing/predicates.py
+++ b/zipline/testing/predicates.py
@@ -380,6 +380,12 @@ def assert_timestamp_and_datetime_equal(result,
     )
 
 
+def assert_isidentical(result, expected, msg=''):
+    assert result.isidentical(expected), (
+        '%s%s is not identical to %s' % (_fmt_msg(msg), result, expected)
+    )
+
+
 try:
     # pull the dshape cases in
     from datashape.util.testing import assert_dshape_equal