refactor: Centralize filter_by_created_timestamp support behind a sto…

…re capability flag Simplification pass over the feature, reviewed again by Codex: - Declare support via OfflineStore.supports_filter_by_created_timestamp (default False) and check it once in the passthrough provider via ensure_filter_by_created_timestamp_supported, replacing the three per-store NotImplementedError guards. Unsupported stores can no longer silently ignore the flag. - The hybrid store re-checks the resolved child store before delegating. - Standardize every supporting store on an explicit filter_by_created_timestamp parameter instead of kwargs.get lookups. - Document the flag in the OfflineStore.get_historical_features docstring alongside start_date/end_date. - Return the dask created-timestamp filter lazily so the mask fuses into the following _drop_duplicates persist instead of forcing an extra materialization. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: David <david-adeniji@hotmail.co.uk>
ntkathole · Jul 31, 2026 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026
commit 81a44e3ef865b50ae60b3ceffc7bc01416cd6a4c
@@ -137,6 +137,8 @@ def project_id_exists(cls, v, values, **kwargs):
 
 
 class BigQueryOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -273,6 +275,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs: Any,
     ) -> RetrievalJob:
         # TODO: Add entity_df validation in order to fail before interacting with BigQuery
@@ -388,7 +391,7 @@ def query_generator() -> Iterator[str]:
                 entity_df_columns=entity_schema_keys,
                 query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN,
                 full_feature_names=full_feature_names,
-                filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False),
+                filter_by_created_timestamp=filter_by_created_timestamp,
             )
 
             try:

@@ -66,6 +66,8 @@ class AthenaOfflineStoreConfig(FeastConfigBaseModel):
 
 
 class AthenaOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

@@ -39,6 +39,8 @@ class ClickhouseOfflineStoreConfig(ClickhouseConfig):
 
 
 class ClickhouseOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def get_historical_features(
         config: RepoConfig,
@@ -48,6 +50,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs,
     ) -> RetrievalJob:
         assert isinstance(config.offline_store, ClickhouseOfflineStoreConfig)
@@ -123,7 +126,7 @@ def query_generator() -> Iterator[str]:
                     entity_df_columns=entity_schema.keys(),
                     query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN,
                     full_feature_names=full_feature_names,
-                    filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False),
+                    filter_by_created_timestamp=filter_by_created_timestamp,
                 )
                 yield query
             finally:

@@ -64,6 +64,8 @@ class CouchbaseColumnarOfflineStoreConfig(FeastConfigBaseModel):
 
 
 class CouchbaseColumnarOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

@@ -118,6 +118,8 @@ class MsSqlServerOfflineStoreConfig(FeastConfigBaseModel):
 
 
 class MsSqlServerOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

@@ -486,6 +486,8 @@ def _oracle_try_execute_ddl(con, ddl: str) -> None:
 
 
 class OracleOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -521,6 +523,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs,
     ) -> RetrievalJob:
         if not feature_views:
@@ -554,7 +557,7 @@ def get_historical_features(
             full_feature_names=full_feature_names,
             data_source_reader=_build_data_source_reader(config, con=con),
             data_source_writer=_build_data_source_writer(config, con=con),
-            filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False),
+            filter_by_created_timestamp=filter_by_created_timestamp,
         )
 
     @staticmethod

@@ -75,6 +75,8 @@ class PostgreSQLOfflineStoreConfig(PostgreSQLConfig):
 
 
 class PostgreSQLOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -135,6 +137,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs,
     ) -> RetrievalJob:
         assert isinstance(config.offline_store, PostgreSQLOfflineStoreConfig)
@@ -222,7 +225,7 @@ def query_generator() -> Iterator[str]:
                     use_cte=use_cte,
                     start_date=start_date,
                     end_date=end_date,
-                    filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False),
+                    filter_by_created_timestamp=filter_by_created_timestamp,
                 )
             finally:
                 # Only cleanup if we created a table

diff --git a/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py b/sdk/python/feast/infra/offline_stores/contrib/ray_offline_store/ray.py
@@ -2313,10 +2313,6 @@ def get_historical_features(
         full_feature_names: bool = False,
         **kwargs: Any,
     ) -> RetrievalJob:
-        if kwargs.get("filter_by_created_timestamp"):
-            raise NotImplementedError(
-                "filter_by_created_timestamp is not yet supported by the Ray offline store."
-            )
         store = RayOfflineStore()
         store._init_ray(config)
 

@@ -93,6 +93,8 @@ class SparkFeatureViewQueryContext(offline_utils.FeatureViewQueryContext):
 
 
 class SparkOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -172,6 +174,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs,
     ) -> RetrievalJob:
         assert isinstance(config.offline_store, SparkOfflineStoreConfig)
@@ -330,7 +333,7 @@ def get_historical_features(
             entity_df_columns=entity_schema_keys,
             query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN,
             full_feature_names=full_feature_names,
-            filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False),
+            filter_by_created_timestamp=filter_by_created_timestamp,
         )
 
         return SparkRetrievalJob(

@@ -298,6 +298,8 @@ def metadata(self) -> Optional[RetrievalMetadata]:
 
 
 class TrinoOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

@@ -141,6 +141,8 @@ def supports_remote_storage_export(self) -> bool:
 
 
 class DaskOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def get_historical_features(
         config: RepoConfig,
@@ -150,6 +152,7 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
+        filter_by_created_timestamp: bool = False,
         **kwargs,
     ) -> RetrievalJob:
         assert isinstance(config.offline_store, DaskOfflineStoreConfig)
@@ -314,10 +317,7 @@ def evaluate_historical_retrieval():
                     timestamp_field,
                 )
 
-                if (
-                    kwargs.get("filter_by_created_timestamp", False)
-                    and created_timestamp_column
-                ):
+                if filter_by_created_timestamp and created_timestamp_column:
                     df_to_join = _filter_created_timestamp(
                         df_to_join,
                         timestamp_field,
@@ -1203,17 +1203,16 @@ def _filter_created_timestamp(
     # Only keep feature values that were already created as of the entity event
     # timestamp. Rows without a timestamp are unmatched entity rows from the
     # left join and are kept; feature rows with a null created timestamp are
-    # excluded, consistent with the SQL predicate.
-    df_to_join = df_to_join[
+    # excluded, consistent with the SQL predicate. Returned lazily so the mask
+    # fuses into the following _drop_duplicates persist.
+    return df_to_join[
         df_to_join[timestamp_field].isna()
         | (
             df_to_join[created_timestamp_column]
             <= df_to_join[entity_df_event_timestamp_col]
         )
     ]
 
-    return df_to_join.persist()
-
 
 def _drop_duplicates(
     df_to_join: dd.DataFrame,

@@ -560,6 +560,8 @@ class DuckDBOfflineStoreConfig(FeastConfigBaseModel):
 
 
 class DuckDBOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

@@ -31,6 +31,8 @@ class OfflineStoresWithConfig(FeastConfigBaseModel):
 
 
 class HybridOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     _instance: Optional["HybridOfflineStore"] = None
     _initialized: bool
     offline_stores: Dict[str, OfflineStore]
@@ -121,6 +123,9 @@ def get_historical_features(
         store = HybridOfflineStore()._get_offline_store_for_feature_view(
             feature_views[0], config
         )
+        # The hybrid store only supports the flag if the store it delegates to does.
+        if kwargs.get("filter_by_created_timestamp"):
+            store.ensure_filter_by_created_timestamp_supported()
         return store.get_historical_features(
             config,
             feature_views,

@@ -451,6 +451,16 @@ class OfflineStore(ABC):
     the SnowflakeOfflineStore can handle SnowflakeSources but not FileSources.
     """
 
+    supports_filter_by_created_timestamp: bool = False
+    """Whether get_historical_features supports the filter_by_created_timestamp flag."""
+
+    def ensure_filter_by_created_timestamp_supported(self) -> None:
+        """Raises NotImplementedError if this store does not support filter_by_created_timestamp."""
+        if not self.supports_filter_by_created_timestamp:
+            raise NotImplementedError(
+                f"filter_by_created_timestamp is not supported by {type(self).__name__}"
+            )
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,
@@ -513,6 +523,10 @@ def get_historical_features(
         Keyword Args:
             start_date: Start date for the timestamp range when retrieving features without entity_df.
             end_date: End date for the timestamp range when retrieving features without entity_df. By default, the current time is used.
+            filter_by_created_timestamp: If True, only feature values whose created timestamp (the
+                ``created_timestamp_column`` of the batch source) is at or before the entity row's event
+                timestamp are considered. Only passed through when a store declares
+                ``supports_filter_by_created_timestamp``.
 
         Returns:
             A RetrievalJob that can be executed to get the features.

@@ -105,6 +105,8 @@ def require_cluster_and_user_or_workgroup(self):
 
 
 class RedshiftOfflineStore(OfflineStore):
+    supports_filter_by_created_timestamp = True
+
     @staticmethod
     def pull_latest_from_table_or_query(
         config: RepoConfig,

diff --git a/sdk/python/feast/infra/offline_stores/remote.py b/sdk/python/feast/infra/offline_stores/remote.py
@@ -218,10 +218,6 @@ def get_historical_features(
         full_feature_names: bool = False,
         **kwargs,
     ) -> RemoteRetrievalJob:
-        if kwargs.get("filter_by_created_timestamp"):
-            raise NotImplementedError(
-                "filter_by_created_timestamp is not yet supported by the remote offline store."
-            )
         assert isinstance(config.offline_store, RemoteOfflineStoreConfig)
 
         client = build_arrow_flight_client(

@@ -298,13 +298,10 @@ def get_historical_features(
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
-        filter_by_created_timestamp: bool = False,
     ) -> RetrievalJob:
-        if filter_by_created_timestamp:
-            raise NotImplementedError(
-                "filter_by_created_timestamp is not yet supported by the Snowflake offline store: the ASOF JOIN "
-                "used for point-in-time retrieval cannot express a created_timestamp cutoff."
-            )
+        # filter_by_created_timestamp is not supported: the ASOF JOIN used for
+        # point-in-time retrieval cannot express a created_timestamp cutoff
+        # (supports_filter_by_created_timestamp stays False).
         assert isinstance(config.offline_store, SnowflakeOfflineStoreConfig)
         for fv in feature_views:
             assert isinstance(fv.batch_source, SnowflakeSource)

@@ -478,6 +478,8 @@ def get_historical_features(
         full_feature_names: bool,
         **kwargs,
     ) -> RetrievalJob:
+        if kwargs.get("filter_by_created_timestamp"):
+            self.offline_store.ensure_filter_by_created_timestamp_supported()
         job = self.offline_store.get_historical_features(
             config=config,
             feature_views=feature_views,

@@ -108,6 +108,19 @@ def test_all_sql_templates_gate_the_created_timestamp_cutoff(module_name):
     )
 
 
+def test_stores_declare_filter_by_created_timestamp_support():
+    from feast.infra.offline_stores.offline_store import OfflineStore
+
+    class _UnsupportedStore(OfflineStore):
+        pass
+
+    with pytest.raises(NotImplementedError, match="filter_by_created_timestamp"):
+        _UnsupportedStore().ensure_filter_by_created_timestamp_supported()
+
+    assert DaskOfflineStore.supports_filter_by_created_timestamp
+    DaskOfflineStore().ensure_filter_by_created_timestamp_supported()
+
+
 class TestDaskFilterByCreatedTimestamp:
     def _run(self, filter_by_created_timestamp, monkeypatch, src=None):
         if src is None: