From 5b226efecacf7b93c5dfe2c4be079b59d5ac80b6 Mon Sep 17 00:00:00 2001 From: David Date: Sun, 19 Jul 2026 18:38:08 +0100 Subject: [PATCH 01/10] feat: Add opt-in at_event_time created_timestamp cutoff to get_historical_features When a feature view has a created_timestamp_column, it is currently used only as a dedup tiebreaker in point-in-time joins, so retrieval can serve feature values whose created_timestamp is after the entity row's event timestamp (backfills, late corrections). This leaks future information into training data and makes training sets non-reproducible. Add an opt-in at_event_time flag (default False) to get_historical_features that adds a created_timestamp <= entity_timestamp predicate to the point-in-time join, so retrieval reflects what was known as of each entity row's timestamp. Implemented for the SQL template stores (BigQuery, Redshift, Spark, Trino, Athena, Postgres, ClickHouse, Couchbase), the ibis-based stores (DuckDB, MSSQL, Oracle) and the dask store. Snowflake, Remote and Ray raise NotImplementedError when the flag is set. Fixes #6615 Co-Authored-By: Claude Fable 5 Signed-off-by: David --- sdk/python/feast/feature_store.py | 12 +- .../feast/infra/offline_stores/bigquery.py | 5 + .../contrib/athena_offline_store/athena.py | 6 + .../clickhouse_offline_store/clickhouse.py | 5 + .../couchbase_offline_store/couchbase.py | 7 + .../contrib/mssql_offline_store/mssql.py | 2 + .../contrib/oracle_offline_store/oracle.py | 1 + .../postgres_offline_store/postgres.py | 7 + .../contrib/ray_offline_store/ray.py | 4 + .../contrib/spark_offline_store/spark.py | 5 + .../contrib/trino_offline_store/trino.py | 5 + sdk/python/feast/infra/offline_stores/dask.py | 25 +++ .../feast/infra/offline_stores/duckdb.py | 2 + sdk/python/feast/infra/offline_stores/ibis.py | 11 + .../infra/offline_stores/offline_utils.py | 2 + .../feast/infra/offline_stores/redshift.py | 6 + .../feast/infra/offline_stores/remote.py | 4 + .../feast/infra/offline_stores/snowflake.py | 6 + .../offline_stores/test_at_event_time.py | 199 ++++++++++++++++++ 19 files changed, 313 insertions(+), 1 deletion(-) create mode 100644 sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index cabca1490b5..2e2bb0deeb2 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1871,6 +1871,7 @@ def get_historical_features( full_feature_names: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, + at_event_time: bool = False, ) -> RetrievalJob: """Enrich an entity dataframe with historical feature values for either training or batch scoring. @@ -1902,6 +1903,13 @@ def get_historical_features( Required when entity_df is not provided. end_date (Optional[datetime]): End date for the timestamp range when retrieving features without entity_df. Required when entity_df is not provided. By default, the current time is used. + at_event_time (bool): If True, only feature values that were already created as of each entity row's + event timestamp are considered: rows whose created timestamp (the ``created_timestamp_column`` of + the batch source) is later than the entity row's event timestamp are excluded. This makes retrieval + reflect what was known at the event time, so late-arriving or backfilled values cannot leak into + training data. Feature views without a ``created_timestamp_column`` are unaffected. Defaults to + False, which preserves the existing behavior of serving the latest known value for the event + timestamp window. Returns: RetrievalJob which can be used to materialize the results. @@ -2003,11 +2011,13 @@ def get_historical_features( provider = self._get_provider() # Optional kwargs - kwargs = {} + kwargs: Dict[str, Any] = {} if start_date is not None: kwargs["start_date"] = start_date if end_date is not None: kwargs["end_date"] = end_date + if at_event_time: + kwargs["at_event_time"] = at_event_time _retrieval_start = time.monotonic() diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index dc5eaf40cac..22d196d8254 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -388,6 +388,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, + at_event_time=kwargs.get("at_event_time", False), ) try: @@ -1734,6 +1735,10 @@ def arrow_schema_to_bq_schema(arrow_schema: pyarrow.Schema) -> List[SchemaField] AND subquery.event_timestamp >= Timestamp_sub(entity_dataframe.entity_timestamp, interval {{ featureview.ttl }} second) {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} + {% for entity in featureview.entities %} AND subquery.{{ entity }} = entity_dataframe.{{ entity }} {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py index d287367cc8d..3b8041fc530 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py +++ b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py @@ -195,6 +195,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, AthenaOfflineStoreConfig) for fv in feature_views: @@ -252,6 +253,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, + at_event_time=at_event_time, ) try: @@ -651,6 +653,10 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} + {% for entity in featureview.entities %} AND subquery.{{ entity }} = entity_dataframe.{{ entity }} {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py index 79958960c85..f8f1e436cda 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py +++ b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py @@ -123,6 +123,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, + at_event_time=kwargs.get("at_event_time", False), ) yield query finally: @@ -539,6 +540,10 @@ def _append_alias(field_names: List[str], alias: str) -> List[str]: {% if featureview.ttl == 0 %}{% else %} AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - interval {{ featureview.ttl }} second {% endif %} + + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} ), /* diff --git a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py index 54921e9515e..bc6c11be4de 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py +++ b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py @@ -136,6 +136,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: """ Retrieve historical features using point-in-time joins. @@ -197,6 +198,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, + at_event_time=at_event_time, ) yield query finally: @@ -481,6 +483,7 @@ def build_point_in_time_query( entity_df_columns: KeysView[str], query_template: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for Couchbase Columnar""" template = Environment(loader=BaseLoader()).from_string(source=query_template) @@ -507,6 +510,7 @@ def build_point_in_time_query( "featureviews": feature_view_query_contexts, "full_feature_names": full_feature_names, "final_output_feature_names": final_output_feature_names, + "at_event_time": at_event_time, } query = template.render(template_context) @@ -620,6 +624,9 @@ def _get_entity_schema( {% if featureview.ttl == 0 %}{% else %} AND date_diff_str(entity_dataframe.entity_timestamp, subquery.event_timestamp, "second") <= {{ featureview.ttl }} {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} {% for entity in featureview.entities %} AND subquery.`{{ entity }}` = entity_dataframe.`{{ entity }}` {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py index 4821aa8dcb6..3fab83ded8a 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py +++ b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py @@ -151,6 +151,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: # TODO avoid this conversion if type(entity_df) == str: @@ -168,6 +169,7 @@ def get_historical_features( data_source_reader=_build_data_source_reader(config), data_source_writer=_build_data_source_writer(config), event_expire_timestamp_fn=mssql_event_expire_timestamp_fn, + at_event_time=at_event_time, ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py index 25517fa74c3..6b719865732 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py +++ b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py @@ -554,6 +554,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), + at_event_time=kwargs.get("at_event_time", False), ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py index d2bbeb90133..03315aa7560 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py +++ b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py @@ -222,6 +222,7 @@ def query_generator() -> Iterator[str]: use_cte=use_cte, start_date=start_date, end_date=end_date, + at_event_time=kwargs.get("at_event_time", False), ) finally: # Only cleanup if we created a table @@ -693,6 +694,7 @@ def build_point_in_time_query( use_cte: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, + at_event_time: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for PostgreSQL""" template = Environment(loader=BaseLoader()).from_string(source=query_template) @@ -723,6 +725,7 @@ def build_point_in_time_query( "use_cte": use_cte, "start_date": start_date, "end_date": end_date, + "at_event_time": at_event_time, } query = template.render(template_context) @@ -965,6 +968,10 @@ def _get_entity_schema( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} + {% for entity in featureview.entities %} AND subquery."{{ entity }}" = entity_dataframe."{{ entity }}" {% endfor %} 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 index 5230797d94b..b44513b319b 100644 --- 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,6 +2313,10 @@ def get_historical_features( full_feature_names: bool = False, **kwargs: Any, ) -> RetrievalJob: + if kwargs.get("at_event_time"): + raise NotImplementedError( + "at_event_time is not yet supported by the Ray offline store." + ) store = RayOfflineStore() store._init_ray(config) diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index 27200a05bc4..ac411468b8d 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py +++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py @@ -330,6 +330,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, + at_event_time=kwargs.get("at_event_time", False), ) return SparkRetrievalJob( @@ -1810,6 +1811,10 @@ def _cast_data_frame( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} + {% for entity in featureview.entities %} AND subquery.{{ entity }} = entity_dataframe.{{ entity }} {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py index 0f77d6e18fc..9b88d4a96f2 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py @@ -356,6 +356,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> TrinoRetrievalJob: assert isinstance(config.offline_store, TrinoOfflineStoreConfig) for fv in feature_views: @@ -417,6 +418,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, + at_event_time=at_event_time, ) return TrinoRetrievalJob( @@ -648,6 +650,9 @@ def _get_entity_df_event_timestamp_range( {% if featureview.ttl == 0 %}{% else %} AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - interval '{{ featureview.ttl }}' second {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} {% for entity in featureview.entities %} AND subquery.{{ entity }} = entity_dataframe.{{ entity }} {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index 1a68cdfb69f..9ce2bb4a01a 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -314,6 +314,13 @@ def evaluate_historical_retrieval(): timestamp_field, ) + if kwargs.get("at_event_time", False) and created_timestamp_column: + df_to_join = _filter_created_timestamp( + df_to_join, + created_timestamp_column, + entity_df_event_timestamp_col, + ) + df_to_join = _drop_duplicates( df_to_join, all_join_keys, @@ -1183,6 +1190,24 @@ def _filter_ttl( return df_to_join +def _filter_created_timestamp( + df_to_join: dd.DataFrame, + created_timestamp_column: str, + entity_df_event_timestamp_col: str, +) -> dd.DataFrame: + # Only keep feature values that were already created as of the entity event timestamp + df_to_join = df_to_join[ + # do not drop entity rows if one of the sources returns NaNs + df_to_join[created_timestamp_column].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, all_join_keys: List[str], diff --git a/sdk/python/feast/infra/offline_stores/duckdb.py b/sdk/python/feast/infra/offline_stores/duckdb.py index 1d3872d6096..3108f60d794 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -595,6 +595,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: return get_historical_features_ibis( config=config, @@ -608,6 +609,7 @@ def get_historical_features( data_source_writer=_write_data_source, staging_location=config.offline_store.staging_location, staging_location_endpoint_override=config.offline_store.staging_location_endpoint_override, + at_event_time=at_event_time, ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/ibis.py b/sdk/python/feast/infra/offline_stores/ibis.py index 3aebc4e903e..62f8478a83b 100644 --- a/sdk/python/feast/infra/offline_stores/ibis.py +++ b/sdk/python/feast/infra/offline_stores/ibis.py @@ -153,6 +153,7 @@ def get_historical_features_ibis( staging_location: Optional[str] = None, staging_location_endpoint_override: Optional[str] = None, event_expire_timestamp_fn=None, + at_event_time: bool = False, ) -> RetrievalJob: entity_schema = _get_entity_schema( entity_df=entity_df, @@ -235,6 +236,7 @@ def read_fv( ], event_timestamp_col=event_timestamp_col, event_expire_timestamp_fn=event_expire_timestamp_fn, + at_event_time=at_event_time, ) odfvs = OnDemandFeatureView.get_requested_odfvs(feature_refs, project, registry) @@ -379,6 +381,7 @@ def point_in_time_join( feature_tables: List[Tuple[Table, str, str, Dict[str, str], List[str], timedelta]], event_timestamp_col="event_timestamp", event_expire_timestamp_fn=None, + at_event_time: bool = False, ): # TODO handle ttl all_entities = [event_timestamp_col] @@ -434,6 +437,14 @@ def point_in_time_join( feature_table[timestamp_field] <= entity_table[event_timestamp_col], ) + if at_event_time and created_timestamp_field: + predicates.append( + feature_table[created_timestamp_field].cast( + dt.Timestamp(timezone="UTC") + ) + <= entity_table[event_timestamp_col] + ) + if ttl: predicates.append( feature_table["event_expire_timestamp"] diff --git a/sdk/python/feast/infra/offline_stores/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index cd2b05a9a60..4de465caaa8 100644 --- a/sdk/python/feast/infra/offline_stores/offline_utils.py +++ b/sdk/python/feast/infra/offline_stores/offline_utils.py @@ -200,6 +200,7 @@ def build_point_in_time_query( entity_df_columns: KeysView[str], query_template: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for Bigquery and Redshift""" env = Environment(loader=BaseLoader()) @@ -228,6 +229,7 @@ def build_point_in_time_query( ), "featureviews": [asdict(context) for context in feature_view_query_contexts], "full_feature_names": full_feature_names, + "at_event_time": at_event_time, "final_output_feature_names": final_output_feature_names, } diff --git a/sdk/python/feast/infra/offline_stores/redshift.py b/sdk/python/feast/infra/offline_stores/redshift.py index ec708ccf798..111daeee435 100644 --- a/sdk/python/feast/infra/offline_stores/redshift.py +++ b/sdk/python/feast/infra/offline_stores/redshift.py @@ -220,6 +220,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, RedshiftOfflineStoreConfig) for fv in feature_views: @@ -278,6 +279,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, + at_event_time=at_event_time, ) try: @@ -1311,6 +1313,10 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if at_event_time and featureview.created_timestamp_column %} + AND subquery.created_timestamp <= entity_dataframe.entity_timestamp + {% endif %} + {% for entity in featureview.entities %} AND subquery.{{ entity }} = entity_dataframe.{{ entity }} {% endfor %} diff --git a/sdk/python/feast/infra/offline_stores/remote.py b/sdk/python/feast/infra/offline_stores/remote.py index a32774ad085..b88f7d2275d 100644 --- a/sdk/python/feast/infra/offline_stores/remote.py +++ b/sdk/python/feast/infra/offline_stores/remote.py @@ -218,6 +218,10 @@ def get_historical_features( full_feature_names: bool = False, **kwargs, ) -> RemoteRetrievalJob: + if kwargs.get("at_event_time"): + raise NotImplementedError( + "at_event_time is not yet supported by the remote offline store." + ) assert isinstance(config.offline_store, RemoteOfflineStoreConfig) client = build_arrow_flight_client( diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 05ed82a1aae..1a4e8abfbea 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -298,7 +298,13 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + at_event_time: bool = False, ) -> RetrievalJob: + if at_event_time: + raise NotImplementedError( + "at_event_time is not yet supported by the Snowflake offline store: the ASOF JOIN " + "used for point-in-time retrieval cannot express a created_timestamp cutoff." + ) assert isinstance(config.offline_store, SnowflakeOfflineStoreConfig) for fv in feature_views: assert isinstance(fv.batch_source, SnowflakeSource) diff --git a/sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py b/sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py new file mode 100644 index 00000000000..f35c01fe2f3 --- /dev/null +++ b/sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py @@ -0,0 +1,199 @@ +from datetime import timedelta +from unittest.mock import MagicMock + +import dask.dataframe as dd +import ibis +import pandas as pd + +from feast.entity import Entity +from feast.feature_view import FeatureView, Field +from feast.infra.offline_stores import dask as dask_mod +from feast.infra.offline_stores import offline_utils +from feast.infra.offline_stores.bigquery import ( + MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, +) +from feast.infra.offline_stores.dask import ( + DaskOfflineStore, + DaskOfflineStoreConfig, +) +from feast.infra.offline_stores.file_source import FileSource +from feast.infra.offline_stores.ibis import point_in_time_join +from feast.infra.offline_stores.offline_utils import FeatureViewQueryContext +from feast.repo_config import RepoConfig +from feast.types import Float32, ValueType + +AT_EVENT_TIME_PREDICATE = ( + "subquery.created_timestamp <= entity_dataframe.entity_timestamp" +) + + +def _query_context(created_timestamp_column): + return FeatureViewQueryContext( + name="driver_stats", + ttl=86400, + entities=["driver_id"], + features=["conv_rate"], + field_mapping={}, + timestamp_field="event_timestamp", + created_timestamp_column=created_timestamp_column, + table_subquery="`project`.`dataset`.`table`", + entity_selections=["driver_id AS driver_id"], + min_event_timestamp="2025-01-01T00:00:00", + max_event_timestamp="2025-01-02T00:00:00", + date_partition_column=None, + timestamp_field_type=None, + ) + + +def _render(created_timestamp_column, **kwargs): + return offline_utils.build_point_in_time_query( + [_query_context(created_timestamp_column)], + left_table_query_string="entity_df_table", + entity_df_event_timestamp_col="event_timestamp", + entity_df_columns={"driver_id": None, "event_timestamp": None}.keys(), + query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, + full_feature_names=False, + **kwargs, + ) + + +def test_at_event_time_adds_created_timestamp_cutoff_to_query(): + query = _render("created_ts", at_event_time=True) + assert AT_EVENT_TIME_PREDICATE in query + + +def test_at_event_time_defaults_to_false_and_leaves_query_unchanged(): + assert AT_EVENT_TIME_PREDICATE not in _render("created_ts") + assert _render("created_ts") == _render("created_ts", at_event_time=False) + + +def test_at_event_time_has_no_effect_without_created_timestamp_column(): + query = _render(None, at_event_time=True) + assert AT_EVENT_TIME_PREDICATE not in query + + +class TestDaskAtEventTime: + def _run(self, at_event_time, monkeypatch): + src = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T10:00:00Z", + "2025-01-01T10:00:00Z", # same event ts, created after the entity ts + ] + ), + "created_ts": pd.to_datetime( + [ + "2025-01-01T12:00:00Z", # created before the entity ts + "2025-01-03T00:00:00Z", # created after the entity ts + ] + ), + "conv_rate": [0.4, 0.6], + } + ) + ddf = dd.from_pandas(src, npartitions=1) + monkeypatch.setattr(dask_mod, "_read_datasource", lambda ds, repo_path: ddf) + + repo_config = RepoConfig( + project="test_project", + registry="test_registry", + provider="local", + offline_store=DaskOfflineStoreConfig(type="dask"), + ) + fv = FeatureView( + name="driver_stats", + entities=[ + Entity( + name="driver_id", + join_keys=["driver_id"], + value_type=ValueType.INT64, + ) + ], + schema=[Field(name="conv_rate", dtype=Float32)], + source=FileSource( + path="dummy.parquet", # not read in this test + timestamp_field="event_timestamp", + created_timestamp_column="created_ts", + ), + ttl=timedelta(days=7), + ) + registry = MagicMock() + registry.list_on_demand_feature_views.return_value = [] + + entity_df = pd.DataFrame( + { + "driver_id": [1], + "event_timestamp": pd.to_datetime(["2025-01-02T00:00:00Z"]), + } + ) + + job = DaskOfflineStore.get_historical_features( + config=repo_config, + feature_views=[fv], + feature_refs=["driver_stats:conv_rate"], + entity_df=entity_df, + registry=registry, + project="test_project", + full_feature_names=False, + at_event_time=at_event_time, + ) + return job.to_df() + + def test_default_serves_latest_created_value(self, monkeypatch): + df = self._run(False, monkeypatch) + assert df["conv_rate"].tolist() == [0.6] + + def test_at_event_time_excludes_values_created_after_entity_timestamp( + self, monkeypatch + ): + df = self._run(True, monkeypatch) + assert df["conv_rate"].tolist() == [0.4] + + +class TestIbisAtEventTime: + def _run(self, at_event_time): + entity_table = ibis.memtable( + pd.DataFrame( + { + "driver_id": [1], + "event_timestamp": pd.to_datetime(["2025-01-02T00:00:00Z"]), + } + ) + ) + feature_table = ibis.memtable( + pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + ["2025-01-01T10:00:00Z", "2025-01-01T10:00:00Z"] + ), + "created_ts": pd.to_datetime( + ["2025-01-01T12:00:00Z", "2025-01-03T00:00:00Z"] + ), + "conv_rate": [0.4, 0.6], + } + ) + ) + res = point_in_time_join( + entity_table=entity_table, + feature_tables=[ + ( + feature_table, + "event_timestamp", + "created_ts", + {"driver_id": "driver_id"}, + ["conv_rate"], + None, + ) + ], + event_timestamp_col="event_timestamp", + at_event_time=at_event_time, + ).execute() + return res + + def test_default_serves_latest_created_value(self): + assert self._run(False)["conv_rate"].tolist() == [0.6] + + def test_at_event_time_excludes_values_created_after_entity_timestamp(self): + assert self._run(True)["conv_rate"].tolist() == [0.4] From aa0fb798e2b1c83dd0652af23dd4ce1228765257 Mon Sep 17 00:00:00 2001 From: David Date: Sun, 19 Jul 2026 18:53:50 +0100 Subject: [PATCH 02/10] test: Add universal integration test and docs for at_event_time Adds a universal offline store integration test covering at_event_time (default returns backfilled values, at_event_time=True only returns values created at or before the entity timestamp, stores without support skip via NotImplementedError) and documents the flag on the point-in-time joins concept page. Co-Authored-By: Claude Fable 5 Signed-off-by: David --- .../concepts/point-in-time-joins.md | 25 +++++ .../test_universal_historical_retrieval.py | 99 +++++++++++++++++++ 2 files changed, 124 insertions(+) diff --git a/docs/getting-started/concepts/point-in-time-joins.md b/docs/getting-started/concepts/point-in-time-joins.md index 55672209005..ff849821c86 100644 --- a/docs/getting-started/concepts/point-in-time-joins.md +++ b/docs/getting-started/concepts/point-in-time-joins.md @@ -62,3 +62,28 @@ Below is the resulting joined training dataframe. It contains both the original Three feature rows were successfully joined to the entity dataframe rows. The first row in the entity dataframe was older than the earliest feature rows in the feature view and could not be joined. The last row in the entity dataframe was outside of the TTL window \(the event happened 11 hours after the feature row\) and also couldn't be joined. +## Retrieving features as of the event time + +By default, point-in-time joins only constrain the feature's event timestamp. If a data source also has a `created_timestamp_column`, it is used to deduplicate rows that share an event timestamp \(the row with the highest created timestamp wins\), but it is not otherwise filtered. This means a value that was backfilled or corrected *after* an entity dataframe timestamp can still be returned for it. + +To restrict retrieval to feature values that were already available at each entity row's timestamp, pass `at_event_time=True`: + +```python +training_df = store.get_historical_features( + entity_df=entity_df, + features = [ + 'driver_hourly_stats:trips_today', + 'driver_hourly_stats:earnings_today' + ], + at_event_time=True, +) +``` + +This adds a `created_timestamp <= entity_timestamp` condition to the join, so each entity dataframe row only sees feature values whose created timestamp is at or before its own timestamp. This is useful to keep backfilled values from leaking into training data, and to reproduce what the online store would have served at each event time \(assuming the created timestamp reflects when the value became available online\). + +A few things to keep in mind: + +* The flag has no effect on feature views whose data source does not set a `created_timestamp_column`. +* Rows with a NULL created timestamp are excluded when the flag is enabled, so the column should be non-null. +* Not all offline stores support this flag yet; unsupported stores raise an error rather than silently ignoring it. + diff --git a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py index af0de479f3e..a0ffd7b81ff 100644 --- a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py +++ b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py @@ -939,3 +939,102 @@ def test_odfv_projection(environment, universal_data_sources, full_feature_names assert unrequested_feature_2 not in actual_df2.columns, ( f"Unrequested ODFV feature '{unrequested_feature_2}' should NOT be in the result" ) + + +@pytest.mark.integration +@pytest.mark.universal_offline_stores +def test_historical_features_at_event_time(environment): + store = environment.feature_store + + now = datetime.now().replace(microsecond=0, second=0, minute=0) + tomorrow = now + timedelta(days=1) + day_after_tomorrow = now + timedelta(days=2) + + entity_df = pd.DataFrame( + data=[ + {"driver_id": 1001, "event_timestamp": tomorrow}, + {"driver_id": 1002, "event_timestamp": tomorrow}, + ] + ) + + driver_stats_df = pd.DataFrame( + data=[ + # Values that were already created at the entity timestamp + { + "driver_id": 1001, + "avg_daily_trips": 10, + "event_timestamp": now, + "created": now, + }, + { + "driver_id": 1002, + "avg_daily_trips": 30, + "event_timestamp": now, + "created": now, + }, + # Backfilled values for the same event timestamps, created after the entity timestamp + { + "driver_id": 1001, + "avg_daily_trips": 20, + "event_timestamp": now, + "created": day_after_tomorrow, + }, + { + "driver_id": 1002, + "avg_daily_trips": 40, + "event_timestamp": now, + "created": day_after_tomorrow, + }, + ] + ) + + driver_stats_data_source = environment.data_source_creator.create_data_source( + df=driver_stats_df, + destination_name=f"test_driver_stats_{int(time.time_ns())}_{random.randint(1000, 9999)}", + timestamp_field="event_timestamp", + created_timestamp_column="created", + ) + + driver = Entity(name="driver", join_keys=["driver_id"]) + driver_fv = FeatureView( + name="driver_stats", + entities=[driver], + schema=[Field(name="avg_daily_trips", dtype=Int32)], + source=driver_stats_data_source, + ) + + store.apply([driver, driver_fv]) + + # By default the backfilled values win the dedup, even though they were + # created after the entity timestamp + actual_df = store.get_historical_features( + entity_df=entity_df, + features=["driver_stats:avg_daily_trips"], + full_feature_names=False, + ).to_df() + expected_df = pd.DataFrame( + data=[ + {"driver_id": 1001, "event_timestamp": tomorrow, "avg_daily_trips": 20}, + {"driver_id": 1002, "event_timestamp": tomorrow, "avg_daily_trips": 40}, + ] + ) + validate_dataframes(expected_df, actual_df, sort_by=["driver_id"]) + + # With at_event_time=True only values created at or before the entity + # timestamp are served + try: + actual_df = store.get_historical_features( + entity_df=entity_df, + features=["driver_stats:avg_daily_trips"], + full_feature_names=False, + at_event_time=True, + ).to_df() + except NotImplementedError: + pytest.skip("The offline store does not support at_event_time") + expected_df = pd.DataFrame( + data=[ + {"driver_id": 1001, "event_timestamp": tomorrow, "avg_daily_trips": 10}, + {"driver_id": 1002, "event_timestamp": tomorrow, "avg_daily_trips": 30}, + ] + ) + validate_dataframes(expected_df, actual_df, sort_by=["driver_id"]) From d47bf4f61426ec48aa05c5d68400da495a1e7989 Mon Sep 17 00:00:00 2001 From: David Date: Sun, 19 Jul 2026 19:09:58 +0100 Subject: [PATCH 03/10] refactor: Rename at_event_time to filter_by_created_timestamp and address review feedback Rename the flag to filter_by_created_timestamp to match the repo's mechanical flag naming and sit alongside created_timestamp_column. Review fixes: - HybridOfflineStore now forwards optional kwargs to the delegated store instead of raising TypeError for supported underlying stores. - The dask filter now excludes feature rows with a null created timestamp (consistent with the SQL predicate) while still keeping unmatched entity rows from the left join. - The integration test only treats NotImplementedError from get_historical_features itself as an unsupported-store skip. - Add template-render tests for all SQL dialects and a dask null created-timestamp test. Co-Authored-By: Claude Fable 5 Signed-off-by: David --- .../concepts/point-in-time-joins.md | 4 +- sdk/python/feast/feature_store.py | 8 +- .../feast/infra/offline_stores/bigquery.py | 4 +- .../contrib/athena_offline_store/athena.py | 6 +- .../clickhouse_offline_store/clickhouse.py | 4 +- .../couchbase_offline_store/couchbase.py | 10 +- .../contrib/mssql_offline_store/mssql.py | 4 +- .../contrib/oracle_offline_store/oracle.py | 2 +- .../postgres_offline_store/postgres.py | 8 +- .../contrib/ray_offline_store/ray.py | 4 +- .../contrib/spark_offline_store/spark.py | 4 +- .../contrib/trino_offline_store/trino.py | 6 +- sdk/python/feast/infra/offline_stores/dask.py | 15 +- .../feast/infra/offline_stores/duckdb.py | 4 +- .../offline_stores/hybrid_offline_store.py | 2 + sdk/python/feast/infra/offline_stores/ibis.py | 8 +- .../infra/offline_stores/offline_utils.py | 4 +- .../feast/infra/offline_stores/redshift.py | 6 +- .../feast/infra/offline_stores/remote.py | 4 +- .../feast/infra/offline_stores/snowflake.py | 6 +- .../test_universal_historical_retrieval.py | 13 +- ...py => test_filter_by_created_timestamp.py} | 131 +++++++++++++----- 22 files changed, 161 insertions(+), 96 deletions(-) rename sdk/python/tests/unit/infra/offline_stores/{test_at_event_time.py => test_filter_by_created_timestamp.py} (58%) diff --git a/docs/getting-started/concepts/point-in-time-joins.md b/docs/getting-started/concepts/point-in-time-joins.md index ff849821c86..6ed5f09cd57 100644 --- a/docs/getting-started/concepts/point-in-time-joins.md +++ b/docs/getting-started/concepts/point-in-time-joins.md @@ -66,7 +66,7 @@ Three feature rows were successfully joined to the entity dataframe rows. The fi By default, point-in-time joins only constrain the feature's event timestamp. If a data source also has a `created_timestamp_column`, it is used to deduplicate rows that share an event timestamp \(the row with the highest created timestamp wins\), but it is not otherwise filtered. This means a value that was backfilled or corrected *after* an entity dataframe timestamp can still be returned for it. -To restrict retrieval to feature values that were already available at each entity row's timestamp, pass `at_event_time=True`: +To restrict retrieval to feature values that were already available at each entity row's timestamp, pass `filter_by_created_timestamp=True`: ```python training_df = store.get_historical_features( @@ -75,7 +75,7 @@ training_df = store.get_historical_features( 'driver_hourly_stats:trips_today', 'driver_hourly_stats:earnings_today' ], - at_event_time=True, + filter_by_created_timestamp=True, ) ``` diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index 2e2bb0deeb2..ae02a54bda6 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1871,7 +1871,7 @@ def get_historical_features( full_feature_names: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: """Enrich an entity dataframe with historical feature values for either training or batch scoring. @@ -1903,7 +1903,7 @@ def get_historical_features( Required when entity_df is not provided. end_date (Optional[datetime]): End date for the timestamp range when retrieving features without entity_df. Required when entity_df is not provided. By default, the current time is used. - at_event_time (bool): If True, only feature values that were already created as of each entity row's + filter_by_created_timestamp (bool): If True, only feature values that were already created as of each entity row's event timestamp are considered: rows whose created timestamp (the ``created_timestamp_column`` of the batch source) is later than the entity row's event timestamp are excluded. This makes retrieval reflect what was known at the event time, so late-arriving or backfilled values cannot leak into @@ -2016,8 +2016,8 @@ def get_historical_features( kwargs["start_date"] = start_date if end_date is not None: kwargs["end_date"] = end_date - if at_event_time: - kwargs["at_event_time"] = at_event_time + if filter_by_created_timestamp: + kwargs["filter_by_created_timestamp"] = filter_by_created_timestamp _retrieval_start = time.monotonic() diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index 22d196d8254..17c36642f57 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -388,7 +388,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, - at_event_time=kwargs.get("at_event_time", False), + filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False), ) try: @@ -1735,7 +1735,7 @@ def arrow_schema_to_bq_schema(arrow_schema: pyarrow.Schema) -> List[SchemaField] AND subquery.event_timestamp >= Timestamp_sub(entity_dataframe.entity_timestamp, interval {{ featureview.ttl }} second) {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py index 3b8041fc530..630c48d9e9e 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py +++ b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py @@ -195,7 +195,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, AthenaOfflineStoreConfig) for fv in feature_views: @@ -253,7 +253,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, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) try: @@ -653,7 +653,7 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py index f8f1e436cda..f654aadf039 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py +++ b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py @@ -123,7 +123,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, - at_event_time=kwargs.get("at_event_time", False), + filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False), ) yield query finally: @@ -541,7 +541,7 @@ def _append_alias(field_names: List[str], alias: str) -> List[str]: AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - interval {{ featureview.ttl }} second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} ), diff --git a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py index bc6c11be4de..d794ae7b8a1 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py +++ b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py @@ -136,7 +136,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: """ Retrieve historical features using point-in-time joins. @@ -198,7 +198,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, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) yield query finally: @@ -483,7 +483,7 @@ def build_point_in_time_query( entity_df_columns: KeysView[str], query_template: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for Couchbase Columnar""" template = Environment(loader=BaseLoader()).from_string(source=query_template) @@ -510,7 +510,7 @@ def build_point_in_time_query( "featureviews": feature_view_query_contexts, "full_feature_names": full_feature_names, "final_output_feature_names": final_output_feature_names, - "at_event_time": at_event_time, + "filter_by_created_timestamp": filter_by_created_timestamp, } query = template.render(template_context) @@ -624,7 +624,7 @@ def _get_entity_schema( {% if featureview.ttl == 0 %}{% else %} AND date_diff_str(entity_dataframe.entity_timestamp, subquery.event_timestamp, "second") <= {{ featureview.ttl }} {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} {% for entity in featureview.entities %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py index 3fab83ded8a..f4692e37a23 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py +++ b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py @@ -151,7 +151,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: # TODO avoid this conversion if type(entity_df) == str: @@ -169,7 +169,7 @@ def get_historical_features( data_source_reader=_build_data_source_reader(config), data_source_writer=_build_data_source_writer(config), event_expire_timestamp_fn=mssql_event_expire_timestamp_fn, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py index 6b719865732..a566915ad10 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py +++ b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py @@ -554,7 +554,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), - at_event_time=kwargs.get("at_event_time", False), + filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False), ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py index 03315aa7560..894b2b4f05a 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py +++ b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py @@ -222,7 +222,7 @@ def query_generator() -> Iterator[str]: use_cte=use_cte, start_date=start_date, end_date=end_date, - at_event_time=kwargs.get("at_event_time", False), + filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False), ) finally: # Only cleanup if we created a table @@ -694,7 +694,7 @@ def build_point_in_time_query( use_cte: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for PostgreSQL""" template = Environment(loader=BaseLoader()).from_string(source=query_template) @@ -725,7 +725,7 @@ def build_point_in_time_query( "use_cte": use_cte, "start_date": start_date, "end_date": end_date, - "at_event_time": at_event_time, + "filter_by_created_timestamp": filter_by_created_timestamp, } query = template.render(template_context) @@ -968,7 +968,7 @@ def _get_entity_schema( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} 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 index b44513b319b..39aa441e061 100644 --- 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,9 +2313,9 @@ def get_historical_features( full_feature_names: bool = False, **kwargs: Any, ) -> RetrievalJob: - if kwargs.get("at_event_time"): + if kwargs.get("filter_by_created_timestamp"): raise NotImplementedError( - "at_event_time is not yet supported by the Ray offline store." + "filter_by_created_timestamp is not yet supported by the Ray offline store." ) store = RayOfflineStore() store._init_ray(config) diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index ac411468b8d..3188aa673ec 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py +++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py @@ -330,7 +330,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, - at_event_time=kwargs.get("at_event_time", False), + filter_by_created_timestamp=kwargs.get("filter_by_created_timestamp", False), ) return SparkRetrievalJob( @@ -1811,7 +1811,7 @@ def _cast_data_frame( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py index 9b88d4a96f2..fbde2c4c670 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py @@ -356,7 +356,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> TrinoRetrievalJob: assert isinstance(config.offline_store, TrinoOfflineStoreConfig) for fv in feature_views: @@ -418,7 +418,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, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) return TrinoRetrievalJob( @@ -650,7 +650,7 @@ def _get_entity_df_event_timestamp_range( {% if featureview.ttl == 0 %}{% else %} AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - interval '{{ featureview.ttl }}' second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} {% for entity in featureview.entities %} diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index 9ce2bb4a01a..39ba9e3f77b 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -314,9 +314,13 @@ def evaluate_historical_retrieval(): timestamp_field, ) - if kwargs.get("at_event_time", False) and created_timestamp_column: + if ( + kwargs.get("filter_by_created_timestamp", False) + and created_timestamp_column + ): df_to_join = _filter_created_timestamp( df_to_join, + timestamp_field, created_timestamp_column, entity_df_event_timestamp_col, ) @@ -1192,13 +1196,16 @@ def _filter_ttl( def _filter_created_timestamp( df_to_join: dd.DataFrame, + timestamp_field: str, created_timestamp_column: str, entity_df_event_timestamp_col: str, ) -> dd.DataFrame: - # Only keep feature values that were already created as of the entity event 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[ - # do not drop entity rows if one of the sources returns NaNs - df_to_join[created_timestamp_column].isna() + df_to_join[timestamp_field].isna() | ( df_to_join[created_timestamp_column] <= df_to_join[entity_df_event_timestamp_col] diff --git a/sdk/python/feast/infra/offline_stores/duckdb.py b/sdk/python/feast/infra/offline_stores/duckdb.py index 3108f60d794..335e7841707 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -595,7 +595,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: return get_historical_features_ibis( config=config, @@ -609,7 +609,7 @@ def get_historical_features( data_source_writer=_write_data_source, staging_location=config.offline_store.staging_location, staging_location_endpoint_override=config.offline_store.staging_location_endpoint_override, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py b/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py index a52f560952a..50312f1187e 100644 --- a/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py +++ b/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py @@ -101,6 +101,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + **kwargs, ) -> RetrievalJob: # TODO: Multiple data sources can be supported when feature store use compute engine # for getting historical features @@ -128,6 +129,7 @@ def get_historical_features( registry, project, full_feature_names, + **kwargs, ) @staticmethod diff --git a/sdk/python/feast/infra/offline_stores/ibis.py b/sdk/python/feast/infra/offline_stores/ibis.py index 62f8478a83b..87f07c61dc7 100644 --- a/sdk/python/feast/infra/offline_stores/ibis.py +++ b/sdk/python/feast/infra/offline_stores/ibis.py @@ -153,7 +153,7 @@ def get_historical_features_ibis( staging_location: Optional[str] = None, staging_location_endpoint_override: Optional[str] = None, event_expire_timestamp_fn=None, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: entity_schema = _get_entity_schema( entity_df=entity_df, @@ -236,7 +236,7 @@ def read_fv( ], event_timestamp_col=event_timestamp_col, event_expire_timestamp_fn=event_expire_timestamp_fn, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) odfvs = OnDemandFeatureView.get_requested_odfvs(feature_refs, project, registry) @@ -381,7 +381,7 @@ def point_in_time_join( feature_tables: List[Tuple[Table, str, str, Dict[str, str], List[str], timedelta]], event_timestamp_col="event_timestamp", event_expire_timestamp_fn=None, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ): # TODO handle ttl all_entities = [event_timestamp_col] @@ -437,7 +437,7 @@ def point_in_time_join( feature_table[timestamp_field] <= entity_table[event_timestamp_col], ) - if at_event_time and created_timestamp_field: + if filter_by_created_timestamp and created_timestamp_field: predicates.append( feature_table[created_timestamp_field].cast( dt.Timestamp(timezone="UTC") diff --git a/sdk/python/feast/infra/offline_stores/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index 4de465caaa8..7ccaf965c9c 100644 --- a/sdk/python/feast/infra/offline_stores/offline_utils.py +++ b/sdk/python/feast/infra/offline_stores/offline_utils.py @@ -200,7 +200,7 @@ def build_point_in_time_query( entity_df_columns: KeysView[str], query_template: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> str: """Build point-in-time query between each feature view table and the entity dataframe for Bigquery and Redshift""" env = Environment(loader=BaseLoader()) @@ -229,7 +229,7 @@ def build_point_in_time_query( ), "featureviews": [asdict(context) for context in feature_view_query_contexts], "full_feature_names": full_feature_names, - "at_event_time": at_event_time, + "filter_by_created_timestamp": filter_by_created_timestamp, "final_output_feature_names": final_output_feature_names, } diff --git a/sdk/python/feast/infra/offline_stores/redshift.py b/sdk/python/feast/infra/offline_stores/redshift.py index 111daeee435..5677ce84e84 100644 --- a/sdk/python/feast/infra/offline_stores/redshift.py +++ b/sdk/python/feast/infra/offline_stores/redshift.py @@ -220,7 +220,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, RedshiftOfflineStoreConfig) for fv in feature_views: @@ -279,7 +279,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, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) try: @@ -1313,7 +1313,7 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} - {% if at_event_time and featureview.created_timestamp_column %} + {% if filter_by_created_timestamp and featureview.created_timestamp_column %} AND subquery.created_timestamp <= entity_dataframe.entity_timestamp {% endif %} diff --git a/sdk/python/feast/infra/offline_stores/remote.py b/sdk/python/feast/infra/offline_stores/remote.py index b88f7d2275d..a02c4ec2764 100644 --- a/sdk/python/feast/infra/offline_stores/remote.py +++ b/sdk/python/feast/infra/offline_stores/remote.py @@ -218,9 +218,9 @@ def get_historical_features( full_feature_names: bool = False, **kwargs, ) -> RemoteRetrievalJob: - if kwargs.get("at_event_time"): + if kwargs.get("filter_by_created_timestamp"): raise NotImplementedError( - "at_event_time is not yet supported by the remote offline store." + "filter_by_created_timestamp is not yet supported by the remote offline store." ) assert isinstance(config.offline_store, RemoteOfflineStoreConfig) diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 1a4e8abfbea..46ceb79c12e 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -298,11 +298,11 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, - at_event_time: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: - if at_event_time: + if filter_by_created_timestamp: raise NotImplementedError( - "at_event_time is not yet supported by the Snowflake offline store: the ASOF JOIN " + "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." ) assert isinstance(config.offline_store, SnowflakeOfflineStoreConfig) diff --git a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py index a0ffd7b81ff..d9f1b5c2477 100644 --- a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py +++ b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py @@ -943,7 +943,7 @@ def test_odfv_projection(environment, universal_data_sources, full_feature_names @pytest.mark.integration @pytest.mark.universal_offline_stores -def test_historical_features_at_event_time(environment): +def test_historical_features_filter_by_created_timestamp(environment): store = environment.feature_store now = datetime.now().replace(microsecond=0, second=0, minute=0) @@ -1020,17 +1020,18 @@ def test_historical_features_at_event_time(environment): ) validate_dataframes(expected_df, actual_df, sort_by=["driver_id"]) - # With at_event_time=True only values created at or before the entity + # With filter_by_created_timestamp=True only values created at or before the entity # timestamp are served try: - actual_df = store.get_historical_features( + job = store.get_historical_features( entity_df=entity_df, features=["driver_stats:avg_daily_trips"], full_feature_names=False, - at_event_time=True, - ).to_df() + filter_by_created_timestamp=True, + ) except NotImplementedError: - pytest.skip("The offline store does not support at_event_time") + pytest.skip("The offline store does not support filter_by_created_timestamp") + actual_df = job.to_df() expected_df = pd.DataFrame( data=[ {"driver_id": 1001, "event_timestamp": tomorrow, "avg_daily_trips": 10}, diff --git a/sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py similarity index 58% rename from sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py rename to sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py index f35c01fe2f3..b23af9ba665 100644 --- a/sdk/python/tests/unit/infra/offline_stores/test_at_event_time.py +++ b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py @@ -4,6 +4,7 @@ import dask.dataframe as dd import ibis import pandas as pd +import pytest from feast.entity import Entity from feast.feature_view import FeatureView, Field @@ -22,7 +23,7 @@ from feast.repo_config import RepoConfig from feast.types import Float32, ValueType -AT_EVENT_TIME_PREDICATE = ( +CREATED_TIMESTAMP_PREDICATE = ( "subquery.created_timestamp <= entity_dataframe.entity_timestamp" ) @@ -45,53 +46,89 @@ def _query_context(created_timestamp_column): ) -def _render(created_timestamp_column, **kwargs): +def _render( + created_timestamp_column, + query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, + **kwargs, +): return offline_utils.build_point_in_time_query( [_query_context(created_timestamp_column)], left_table_query_string="entity_df_table", entity_df_event_timestamp_col="event_timestamp", entity_df_columns={"driver_id": None, "event_timestamp": None}.keys(), - query_template=MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN, + query_template=query_template, full_feature_names=False, **kwargs, ) -def test_at_event_time_adds_created_timestamp_cutoff_to_query(): - query = _render("created_ts", at_event_time=True) - assert AT_EVENT_TIME_PREDICATE in query +def test_filter_by_created_timestamp_adds_created_timestamp_cutoff_to_query(): + query = _render("created_ts", filter_by_created_timestamp=True) + assert CREATED_TIMESTAMP_PREDICATE in query -def test_at_event_time_defaults_to_false_and_leaves_query_unchanged(): - assert AT_EVENT_TIME_PREDICATE not in _render("created_ts") - assert _render("created_ts") == _render("created_ts", at_event_time=False) +def test_filter_by_created_timestamp_defaults_to_false_and_leaves_query_unchanged(): + assert CREATED_TIMESTAMP_PREDICATE not in _render("created_ts") + assert _render("created_ts") == _render( + "created_ts", filter_by_created_timestamp=False + ) -def test_at_event_time_has_no_effect_without_created_timestamp_column(): - query = _render(None, at_event_time=True) - assert AT_EVENT_TIME_PREDICATE not in query +def test_filter_by_created_timestamp_has_no_effect_without_created_timestamp_column(): + query = _render(None, filter_by_created_timestamp=True) + assert CREATED_TIMESTAMP_PREDICATE not in query -class TestDaskAtEventTime: - def _run(self, at_event_time, monkeypatch): - src = pd.DataFrame( - { - "driver_id": [1, 1], - "event_timestamp": pd.to_datetime( - [ - "2025-01-01T10:00:00Z", - "2025-01-01T10:00:00Z", # same event ts, created after the entity ts - ] - ), - "created_ts": pd.to_datetime( - [ - "2025-01-01T12:00:00Z", # created before the entity ts - "2025-01-03T00:00:00Z", # created after the entity ts - ] - ), - "conv_rate": [0.4, 0.6], - } - ) +SQL_TEMPLATE_STORE_MODULES = [ + "feast.infra.offline_stores.redshift", + "feast.infra.offline_stores.contrib.spark_offline_store.spark", + "feast.infra.offline_stores.contrib.trino_offline_store.trino", + "feast.infra.offline_stores.contrib.athena_offline_store.athena", + "feast.infra.offline_stores.contrib.postgres_offline_store.postgres", + "feast.infra.offline_stores.contrib.clickhouse_offline_store.clickhouse", + "feast.infra.offline_stores.contrib.couchbase_offline_store.couchbase", +] + + +@pytest.mark.parametrize("module_name", SQL_TEMPLATE_STORE_MODULES) +def test_all_sql_templates_gate_the_created_timestamp_cutoff(module_name): + module = pytest.importorskip(module_name) + template = module.MULTIPLE_FEATURE_VIEW_POINT_IN_TIME_JOIN + + with_cutoff = _render( + "created_ts", query_template=template, filter_by_created_timestamp=True + ) + assert CREATED_TIMESTAMP_PREDICATE in with_cutoff + + assert CREATED_TIMESTAMP_PREDICATE not in _render( + "created_ts", query_template=template + ) + assert CREATED_TIMESTAMP_PREDICATE not in _render( + None, query_template=template, filter_by_created_timestamp=True + ) + + +class TestDaskFilterByCreatedTimestamp: + def _run(self, filter_by_created_timestamp, monkeypatch, src=None): + if src is None: + src = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + [ + "2025-01-01T10:00:00Z", + "2025-01-01T10:00:00Z", # same event ts, created after the entity ts + ] + ), + "created_ts": pd.to_datetime( + [ + "2025-01-01T12:00:00Z", # created before the entity ts + "2025-01-03T00:00:00Z", # created after the entity ts + ] + ), + "conv_rate": [0.4, 0.6], + } + ) ddf = dd.from_pandas(src, npartitions=1) monkeypatch.setattr(dask_mod, "_read_datasource", lambda ds, repo_path: ddf) @@ -136,7 +173,7 @@ def _run(self, at_event_time, monkeypatch): registry=registry, project="test_project", full_feature_names=False, - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ) return job.to_df() @@ -144,15 +181,31 @@ def test_default_serves_latest_created_value(self, monkeypatch): df = self._run(False, monkeypatch) assert df["conv_rate"].tolist() == [0.6] - def test_at_event_time_excludes_values_created_after_entity_timestamp( + def test_filter_by_created_timestamp_excludes_values_created_after_entity_timestamp( self, monkeypatch ): df = self._run(True, monkeypatch) assert df["conv_rate"].tolist() == [0.4] + def test_filter_by_created_timestamp_excludes_rows_with_null_created_timestamp( + self, monkeypatch + ): + src = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + ["2025-01-01T10:00:00Z", "2025-01-01T10:00:00Z"] + ), + "created_ts": pd.to_datetime(["2025-01-01T12:00:00Z", pd.NaT]), + "conv_rate": [0.4, 0.6], + } + ) + df = self._run(True, monkeypatch, src=src) + assert df["conv_rate"].tolist() == [0.4] + -class TestIbisAtEventTime: - def _run(self, at_event_time): +class TestIbisFilterByCreatedTimestamp: + def _run(self, filter_by_created_timestamp): entity_table = ibis.memtable( pd.DataFrame( { @@ -188,12 +241,14 @@ def _run(self, at_event_time): ) ], event_timestamp_col="event_timestamp", - at_event_time=at_event_time, + filter_by_created_timestamp=filter_by_created_timestamp, ).execute() return res def test_default_serves_latest_created_value(self): assert self._run(False)["conv_rate"].tolist() == [0.6] - def test_at_event_time_excludes_values_created_after_entity_timestamp(self): + def test_filter_by_created_timestamp_excludes_values_created_after_entity_timestamp( + self, + ): assert self._run(True)["conv_rate"].tolist() == [0.4] From 81a44e3ef865b50ae60b3ceffc7bc01416cd6a4c Mon Sep 17 00:00:00 2001 From: David Date: Sun, 19 Jul 2026 19:30:32 +0100 Subject: [PATCH 04/10] refactor: Centralize filter_by_created_timestamp support behind a store 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 Signed-off-by: David --- sdk/python/feast/infra/offline_stores/bigquery.py | 5 ++++- .../contrib/athena_offline_store/athena.py | 2 ++ .../clickhouse_offline_store/clickhouse.py | 5 ++++- .../contrib/couchbase_offline_store/couchbase.py | 2 ++ .../contrib/mssql_offline_store/mssql.py | 2 ++ .../contrib/oracle_offline_store/oracle.py | 5 ++++- .../contrib/postgres_offline_store/postgres.py | 5 ++++- .../contrib/ray_offline_store/ray.py | 4 ---- .../contrib/spark_offline_store/spark.py | 5 ++++- .../contrib/trino_offline_store/trino.py | 2 ++ sdk/python/feast/infra/offline_stores/dask.py | 15 +++++++-------- sdk/python/feast/infra/offline_stores/duckdb.py | 2 ++ .../infra/offline_stores/hybrid_offline_store.py | 5 +++++ .../feast/infra/offline_stores/offline_store.py | 14 ++++++++++++++ sdk/python/feast/infra/offline_stores/redshift.py | 2 ++ sdk/python/feast/infra/offline_stores/remote.py | 4 ---- .../feast/infra/offline_stores/snowflake.py | 9 +++------ sdk/python/feast/infra/passthrough_provider.py | 2 ++ .../test_filter_by_created_timestamp.py | 13 +++++++++++++ 19 files changed, 76 insertions(+), 27 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/bigquery.py b/sdk/python/feast/infra/offline_stores/bigquery.py index 17c36642f57..a7c8c41094d 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery.py +++ b/sdk/python/feast/infra/offline_stores/bigquery.py @@ -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: diff --git a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py index 630c48d9e9e..d8fae6bf19b 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py +++ b/sdk/python/feast/infra/offline_stores/contrib/athena_offline_store/athena.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py index f654aadf039..9e353151746 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py +++ b/sdk/python/feast/infra/offline_stores/contrib/clickhouse_offline_store/clickhouse.py @@ -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: diff --git a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py index d794ae7b8a1..7ff7ed7225f 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py +++ b/sdk/python/feast/infra/offline_stores/contrib/couchbase_offline_store/couchbase.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py index f4692e37a23..fd3e2cb4a34 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py +++ b/sdk/python/feast/infra/offline_stores/contrib/mssql_offline_store/mssql.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py index a566915ad10..0aa657c69c9 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py +++ b/sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py @@ -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 diff --git a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py index 894b2b4f05a..44270356b2d 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py +++ b/sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py @@ -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 index 39aa441e061..5230797d94b 100644 --- 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) diff --git a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index 3188aa673ec..6fb605972c2 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py +++ b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py @@ -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( diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py index fbde2c4c670..c9d4119f94f 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index 39ba9e3f77b..02ec1dacb01 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -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,8 +1203,9 @@ 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] @@ -1212,8 +1213,6 @@ def _filter_created_timestamp( ) ] - return df_to_join.persist() - def _drop_duplicates( df_to_join: dd.DataFrame, diff --git a/sdk/python/feast/infra/offline_stores/duckdb.py b/sdk/python/feast/infra/offline_stores/duckdb.py index 335e7841707..e3421d5cbb3 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py b/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py index 50312f1187e..2da5c2f2701 100644 --- a/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py +++ b/sdk/python/feast/infra/offline_stores/hybrid_offline_store.py @@ -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, diff --git a/sdk/python/feast/infra/offline_stores/offline_store.py b/sdk/python/feast/infra/offline_stores/offline_store.py index e8cf5e5c796..9d4092d6799 100644 --- a/sdk/python/feast/infra/offline_stores/offline_store.py +++ b/sdk/python/feast/infra/offline_stores/offline_store.py @@ -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. diff --git a/sdk/python/feast/infra/offline_stores/redshift.py b/sdk/python/feast/infra/offline_stores/redshift.py index 5677ce84e84..1717cbaee79 100644 --- a/sdk/python/feast/infra/offline_stores/redshift.py +++ b/sdk/python/feast/infra/offline_stores/redshift.py @@ -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 index a02c4ec2764..a32774ad085 100644 --- 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( diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 46ceb79c12e..dd4b55c9325 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -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) diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py index 93f38567376..d33f3be9b69 100644 --- a/sdk/python/feast/infra/passthrough_provider.py +++ b/sdk/python/feast/infra/passthrough_provider.py @@ -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, diff --git a/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py index b23af9ba665..e1353709172 100644 --- a/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py +++ b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py @@ -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: From 6d1cf1d553129fec2e2e1fbfa19d3051ccd73206 Mon Sep 17 00:00:00 2001 From: David Date: Tue, 21 Jul 2026 18:09:30 +0100 Subject: [PATCH 05/10] chore: Trim comments to the non-obvious constraints Co-Authored-By: Claude Fable 5 Signed-off-by: David --- sdk/python/feast/infra/offline_stores/dask.py | 8 +++----- sdk/python/feast/infra/offline_stores/snowflake.py | 5 ++--- .../offline_store/test_universal_historical_retrieval.py | 5 +---- 3 files changed, 6 insertions(+), 12 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index 02ec1dacb01..807c6a9c6f9 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -1200,11 +1200,9 @@ def _filter_created_timestamp( created_timestamp_column: str, entity_df_event_timestamp_col: str, ) -> dd.DataFrame: - # 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. Returned lazily so the mask - # fuses into the following _drop_duplicates persist. + # timestamp_field is NaN only for unmatched entity rows from the left join, + # which must be kept; null created timestamps fail the comparison and are + # excluded, matching the SQL predicate. Kept lazy to fuse into the next persist. return df_to_join[ df_to_join[timestamp_field].isna() | ( diff --git a/sdk/python/feast/infra/offline_stores/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index dd4b55c9325..84b829617f8 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -299,9 +299,8 @@ def get_historical_features( project: str, full_feature_names: bool = False, ) -> RetrievalJob: - # 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). + # supports_filter_by_created_timestamp stays False: the ASOF JOIN cannot + # express a created_timestamp cutoff. assert isinstance(config.offline_store, SnowflakeOfflineStoreConfig) for fv in feature_views: assert isinstance(fv.batch_source, SnowflakeSource) diff --git a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py index d9f1b5c2477..6a17fc47bbb 100644 --- a/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py +++ b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py @@ -1005,8 +1005,7 @@ def test_historical_features_filter_by_created_timestamp(environment): store.apply([driver, driver_fv]) - # By default the backfilled values win the dedup, even though they were - # created after the entity timestamp + # Default: the backfilled values win the dedup actual_df = store.get_historical_features( entity_df=entity_df, features=["driver_stats:avg_daily_trips"], @@ -1020,8 +1019,6 @@ def test_historical_features_filter_by_created_timestamp(environment): ) validate_dataframes(expected_df, actual_df, sort_by=["driver_id"]) - # With filter_by_created_timestamp=True only values created at or before the entity - # timestamp are served try: job = store.get_historical_features( entity_df=entity_df, From 6d237e5453477f541cada8614ce8da6c80ad31ce Mon Sep 17 00:00:00 2001 From: David Date: Tue, 21 Jul 2026 18:16:49 +0100 Subject: [PATCH 06/10] docs: Tighten filter_by_created_timestamp docstring Co-Authored-By: Claude Fable 5 Signed-off-by: David --- sdk/python/feast/feature_store.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index fc5c7613ec7..587ca1415bf 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1914,13 +1914,11 @@ def get_historical_features( Required when entity_df is not provided. end_date (Optional[datetime]): End date for the timestamp range when retrieving features without entity_df. Required when entity_df is not provided. By default, the current time is used. - filter_by_created_timestamp (bool): If True, only feature values that were already created as of each entity row's - event timestamp are considered: rows whose created timestamp (the ``created_timestamp_column`` of - the batch source) is later than the entity row's event timestamp are excluded. This makes retrieval - reflect what was known at the event time, so late-arriving or backfilled values cannot leak into - training data. Feature views without a ``created_timestamp_column`` are unaffected. Defaults to - False, which preserves the existing behavior of serving the latest known value for the event - timestamp window. + filter_by_created_timestamp (bool): If True, exclude feature values whose created timestamp + (the batch source's ``created_timestamp_column``) is later than the entity row's event + timestamp, so retrieval only reflects what was known at the event time and backfilled + values cannot leak into training data. Feature views without a + ``created_timestamp_column`` are unaffected. Defaults to False. Returns: RetrievalJob which can be used to materialize the results. From 90c0bb4232e45936283e6c70d45ad9937f3d85c0 Mon Sep 17 00:00:00 2001 From: David Date: Tue, 21 Jul 2026 18:19:29 +0100 Subject: [PATCH 07/10] docs: Condense filter_by_created_timestamp caveats into a hint Co-Authored-By: Claude Fable 5 Signed-off-by: David --- docs/getting-started/concepts/point-in-time-joins.md | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/docs/getting-started/concepts/point-in-time-joins.md b/docs/getting-started/concepts/point-in-time-joins.md index 6ed5f09cd57..9e385774327 100644 --- a/docs/getting-started/concepts/point-in-time-joins.md +++ b/docs/getting-started/concepts/point-in-time-joins.md @@ -81,9 +81,7 @@ training_df = store.get_historical_features( This adds a `created_timestamp <= entity_timestamp` condition to the join, so each entity dataframe row only sees feature values whose created timestamp is at or before its own timestamp. This is useful to keep backfilled values from leaking into training data, and to reproduce what the online store would have served at each event time \(assuming the created timestamp reflects when the value became available online\). -A few things to keep in mind: - -* The flag has no effect on feature views whose data source does not set a `created_timestamp_column`. -* Rows with a NULL created timestamp are excluded when the flag is enabled, so the column should be non-null. -* Not all offline stores support this flag yet; unsupported stores raise an error rather than silently ignoring it. +{% hint style="info" %} +Rows with a NULL created timestamp are excluded when the flag is enabled, so the column should be non-null. Not all offline stores support this flag yet; unsupported stores raise an error rather than silently ignoring it. +{% endhint %} From 48c4468d1ce678d2addbc0c23c00a661b06caa51 Mon Sep 17 00:00:00 2001 From: David Date: Fri, 31 Jul 2026 10:57:33 +0100 Subject: [PATCH 08/10] fix: Keep entity rows whose candidate versions are all future-created The Dask cutoff filtered rows after the left join, so an entity whose every candidate version was created after its timestamp lost all of its rows and disappeared from the result. get_historical_features must preserve entity-dataframe cardinality and return null features instead. Blank the feature-view columns rather than dropping the row, which leaves it indistinguishable from an unmatched left join. _drop_duplicates already sorts nulls first and keeps the last row, so a valid version still wins where one exists and a blanked row survives only when nothing else does. The unit tests now set fv.entity_columns. Without it the derived join keys are empty and _merge silently cross joins, which hid the per-entity behaviour because the existing cases all used a single entity row. Signed-off-by: David --- sdk/python/feast/infra/offline_stores/dask.py | 41 +++++++---- .../test_filter_by_created_timestamp.py | 69 ++++++++++++++++--- 2 files changed, 89 insertions(+), 21 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index 807c6a9c6f9..b8a19ce211e 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -3,7 +3,7 @@ import uuid from datetime import date, datetime, timezone from pathlib import Path -from typing import Any, Callable, Dict, List, Literal, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Literal, Optional, Set, Tuple, Union import dask import dask.dataframe as dd @@ -302,6 +302,9 @@ def evaluate_historical_retrieval(): if non_entity_mode: current_join_keys = [] + # Snapshot before the merge blends the two sides together. + entity_df_columns = list(entity_df_with_features.columns) + df_to_join = _merge( entity_df_with_features, df_to_join, current_join_keys ) @@ -318,11 +321,14 @@ def evaluate_historical_retrieval(): ) if filter_by_created_timestamp and created_timestamp_column: - df_to_join = _filter_created_timestamp( + df_to_join = _apply_created_timestamp_cutoff( df_to_join, timestamp_field, created_timestamp_column, entity_df_event_timestamp_col, + # Join keys too: in non-entity mode they come from the + # feature view side. + set(entity_df_columns) | set(join_keys), ) df_to_join = _drop_duplicates( @@ -1194,22 +1200,31 @@ def _filter_ttl( return df_to_join -def _filter_created_timestamp( +def _apply_created_timestamp_cutoff( df_to_join: dd.DataFrame, timestamp_field: str, created_timestamp_column: str, entity_df_event_timestamp_col: str, + preserved_columns: Set[str], ) -> dd.DataFrame: - # timestamp_field is NaN only for unmatched entity rows from the left join, - # which must be kept; null created timestamps fail the comparison and are - # excluded, matching the SQL predicate. Kept lazy to fuse into the next 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] - ) - ] + # Blanking rather than dropping preserves entity-dataframe cardinality when every + # candidate is too new. A blanked row looks like an unmatched left join, which + # _drop_duplicates already resolves (nulls sort first, keep="last"). The isna() term + # keeps a matched row whose source timestamp is null, as _filter_ttl does. + too_new = ~df_to_join[timestamp_field].isna() & ~( + df_to_join[created_timestamp_column] + <= df_to_join[entity_df_event_timestamp_col] + ) + + # One assign, not a per-column loop: chained assignments make optimization + # super-linear in column count. Lazy so it fuses into the next persist. + return df_to_join.assign( + **{ + column: df_to_join[column].mask(too_new) + for column in df_to_join.columns + if column not in preserved_columns + } + ) def _drop_duplicates( diff --git a/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py index e1353709172..e5eff80a412 100644 --- a/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py +++ b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py @@ -21,7 +21,7 @@ from feast.infra.offline_stores.ibis import point_in_time_join from feast.infra.offline_stores.offline_utils import FeatureViewQueryContext from feast.repo_config import RepoConfig -from feast.types import Float32, ValueType +from feast.types import Float32, Int64, ValueType CREATED_TIMESTAMP_PREDICATE = ( "subquery.created_timestamp <= entity_dataframe.entity_timestamp" @@ -122,7 +122,7 @@ class _UnsupportedStore(OfflineStore): class TestDaskFilterByCreatedTimestamp: - def _run(self, filter_by_created_timestamp, monkeypatch, src=None): + def _run(self, filter_by_created_timestamp, monkeypatch, src=None, entity_df=None): if src is None: src = pd.DataFrame( { @@ -168,15 +168,19 @@ def _run(self, filter_by_created_timestamp, monkeypatch, src=None): ), ttl=timedelta(days=7), ) + # Without this the derived join keys are empty and _merge cross joins, which + # hides per-entity bugs. apply()/infer_features() sets it in real usage. + fv.entity_columns = [Field(name="driver_id", dtype=Int64)] registry = MagicMock() registry.list_on_demand_feature_views.return_value = [] - entity_df = pd.DataFrame( - { - "driver_id": [1], - "event_timestamp": pd.to_datetime(["2025-01-02T00:00:00Z"]), - } - ) + if entity_df is None: + entity_df = pd.DataFrame( + { + "driver_id": [1], + "event_timestamp": pd.to_datetime(["2025-01-02T00:00:00Z"]), + } + ) job = DaskOfflineStore.get_historical_features( config=repo_config, @@ -216,6 +220,55 @@ def test_filter_by_created_timestamp_excludes_rows_with_null_created_timestamp( df = self._run(True, monkeypatch, src=src) assert df["conv_rate"].tolist() == [0.4] + def test_entity_row_is_kept_when_every_candidate_is_future_created( + self, monkeypatch + ): + src = pd.DataFrame( + { + "driver_id": [1, 1], + "event_timestamp": pd.to_datetime( + ["2025-01-01T10:00:00Z", "2025-01-01T11:00:00Z"] + ), + "created_ts": pd.to_datetime( + ["2025-01-03T00:00:00Z", "2025-01-04T00:00:00Z"] + ), + "conv_rate": [0.4, 0.6], + } + ) + df = self._run(True, monkeypatch, src=src) + assert len(df) == 1 + assert df["driver_id"].tolist() == [1] + assert df["conv_rate"].isna().all() + + def test_entity_rows_without_a_valid_version_do_not_borrow_from_others( + self, monkeypatch + ): + src = pd.DataFrame( + { + "driver_id": [1, 2], + "event_timestamp": pd.to_datetime( + ["2025-01-01T10:00:00Z", "2025-01-01T10:00:00Z"] + ), + "created_ts": pd.to_datetime( + ["2025-01-01T12:00:00Z", "2025-01-03T00:00:00Z"] + ), + "conv_rate": [0.4, 0.6], + } + ) + entity_df = pd.DataFrame( + { + "driver_id": [1, 2], + "event_timestamp": pd.to_datetime( + ["2025-01-02T00:00:00Z", "2025-01-02T00:00:00Z"] + ), + } + ) + df = self._run(True, monkeypatch, src=src, entity_df=entity_df) + by_driver = df.set_index("driver_id")["conv_rate"] + assert len(df) == 2 + assert by_driver.loc[1] == pytest.approx(0.4) + assert pd.isna(by_driver.loc[2]) + class TestIbisFilterByCreatedTimestamp: def _run(self, filter_by_created_timestamp): From adbdebeb28c8b98d3d9e42af5723af2ccf7f8f42 Mon Sep 17 00:00:00 2001 From: David Date: Fri, 31 Jul 2026 10:57:43 +0100 Subject: [PATCH 09/10] refactor: Normalize created timestamp on read, not in the join predicate The cutoff predicate cast created_timestamp to UTC while the event-timestamp comparison beside it did not. read_fv normalizes the event timestamp when it reads the source and left the created timestamp alone, so the predicate was compensating for a missing normalization at the one comparison site. Normalize both on read instead. The predicate then needs no cast and matches its neighbour. deduplicate() orders by created_timestamp_column regardless of the cutoff flag, so the normalization is unconditional rather than gated on it; casting a column that is already tz-aware compiles away, so this leaves the emitted query unchanged for tz-aware sources and retires the "mutate only if tz-naive" TODO. Signed-off-by: David --- sdk/python/feast/infra/offline_stores/ibis.py | 23 +++++++++++-------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/ibis.py b/sdk/python/feast/infra/offline_stores/ibis.py index 87f07c61dc7..c4080fd8ba0 100644 --- a/sdk/python/feast/infra/offline_stores/ibis.py +++ b/sdk/python/feast/infra/offline_stores/ibis.py @@ -188,13 +188,18 @@ def read_fv( fv_table = fv_table.rename({new_name: old_name}) timestamp_field = feature_view.batch_source.timestamp_field + created_timestamp_column = feature_view.batch_source.created_timestamp_column + + # deduplicate() orders by created_timestamp_column regardless of the cutoff, so + # normalize both unconditionally. An already-tz-aware cast compiles away. + utc_columns = [timestamp_field] + if created_timestamp_column: + utc_columns.append(created_timestamp_column) - # TODO mutate only if tz-naive fv_table = fv_table.mutate( **{ - timestamp_field: fv_table[timestamp_field].cast( - dt.Timestamp(timezone="UTC") - ) + column: fv_table[column].cast(dt.Timestamp(timezone="UTC")) + for column in utc_columns } ) @@ -220,8 +225,8 @@ def read_fv( return ( fv_table, - feature_view.batch_source.timestamp_field, - feature_view.batch_source.created_timestamp_column, + timestamp_field, + created_timestamp_column, feature_view.projection.join_key_map or {e.name: e.name for e in feature_view.entity_columns}, feature_refs, @@ -439,10 +444,8 @@ def point_in_time_join( if filter_by_created_timestamp and created_timestamp_field: predicates.append( - feature_table[created_timestamp_field].cast( - dt.Timestamp(timezone="UTC") - ) - <= entity_table[event_timestamp_col] + feature_table[created_timestamp_field] + <= entity_table[event_timestamp_col], ) if ttl: From 3d2ea3e0d43d18f0606659d1132402be640f8412 Mon Sep 17 00:00:00 2001 From: David Date: Fri, 31 Jul 2026 11:19:30 +0100 Subject: [PATCH 10/10] Shorten comments in the created timestamp cutoff Signed-off-by: David --- sdk/python/feast/infra/offline_stores/dask.py | 11 +++++------ sdk/python/feast/infra/offline_stores/ibis.py | 4 ++-- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/sdk/python/feast/infra/offline_stores/dask.py b/sdk/python/feast/infra/offline_stores/dask.py index b8a19ce211e..c9233c77d60 100644 --- a/sdk/python/feast/infra/offline_stores/dask.py +++ b/sdk/python/feast/infra/offline_stores/dask.py @@ -1207,17 +1207,16 @@ def _apply_created_timestamp_cutoff( entity_df_event_timestamp_col: str, preserved_columns: Set[str], ) -> dd.DataFrame: - # Blanking rather than dropping preserves entity-dataframe cardinality when every - # candidate is too new. A blanked row looks like an unmatched left join, which - # _drop_duplicates already resolves (nulls sort first, keep="last"). The isna() term - # keeps a matched row whose source timestamp is null, as _filter_ttl does. + # Versions created after the entity timestamp. The isna() term leaves rows with a + # null source timestamp untouched, matching _filter_ttl. too_new = ~df_to_join[timestamp_field].isna() & ~( df_to_join[created_timestamp_column] <= df_to_join[entity_df_event_timestamp_col] ) - # One assign, not a per-column loop: chained assignments make optimization - # super-linear in column count. Lazy so it fuses into the next persist. + # Blank instead of drop, so the entity row survives when every candidate is too new. + # _drop_duplicates then prefers a real match (nulls sort first, keep="last"). + # Single assign, left lazy: a per-column loop is super-linear to optimize. return df_to_join.assign( **{ column: df_to_join[column].mask(too_new) diff --git a/sdk/python/feast/infra/offline_stores/ibis.py b/sdk/python/feast/infra/offline_stores/ibis.py index c4080fd8ba0..e0aadd63b4a 100644 --- a/sdk/python/feast/infra/offline_stores/ibis.py +++ b/sdk/python/feast/infra/offline_stores/ibis.py @@ -190,8 +190,8 @@ def read_fv( timestamp_field = feature_view.batch_source.timestamp_field created_timestamp_column = feature_view.batch_source.created_timestamp_column - # deduplicate() orders by created_timestamp_column regardless of the cutoff, so - # normalize both unconditionally. An already-tz-aware cast compiles away. + # deduplicate() orders by created_timestamp_column whether or not the cutoff is + # on, so this cannot be gated on it. Casting an already-UTC column is a no-op. utc_columns = [timestamp_field] if created_timestamp_column: utc_columns.append(created_timestamp_column)