diff --git a/docs/getting-started/concepts/point-in-time-joins.md b/docs/getting-started/concepts/point-in-time-joins.md index 55672209005..9e385774327 100644 --- a/docs/getting-started/concepts/point-in-time-joins.md +++ b/docs/getting-started/concepts/point-in-time-joins.md @@ -62,3 +62,26 @@ 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 `filter_by_created_timestamp=True`: + +```python +training_df = store.get_historical_features( + entity_df=entity_df, + features = [ + 'driver_hourly_stats:trips_today', + 'driver_hourly_stats:earnings_today' + ], + filter_by_created_timestamp=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\). + +{% 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 %} + diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index ba110c1e0a4..587ca1415bf 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -1882,6 +1882,7 @@ def get_historical_features( full_feature_names: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: """Enrich an entity dataframe with historical feature values for either training or batch scoring. @@ -1913,6 +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, 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. @@ -2014,11 +2020,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 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 dc5eaf40cac..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,6 +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=filter_by_created_timestamp, ) try: @@ -1734,6 +1738,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 filter_by_created_timestamp 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..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, @@ -195,6 +197,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, AthenaOfflineStoreConfig) for fv in feature_views: @@ -252,6 +255,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=filter_by_created_timestamp, ) try: @@ -651,6 +655,10 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if filter_by_created_timestamp 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..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,6 +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=filter_by_created_timestamp, ) yield query finally: @@ -539,6 +543,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 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 54921e9515e..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, @@ -136,6 +138,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: """ Retrieve historical features using point-in-time joins. @@ -197,6 +200,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=filter_by_created_timestamp, ) yield query finally: @@ -481,6 +485,7 @@ def build_point_in_time_query( entity_df_columns: KeysView[str], query_template: str, full_feature_names: 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) @@ -507,6 +512,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, + "filter_by_created_timestamp": filter_by_created_timestamp, } query = template.render(template_context) @@ -620,6 +626,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 filter_by_created_timestamp 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..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, @@ -151,6 +153,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: # TODO avoid this conversion if type(entity_df) == str: @@ -168,6 +171,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, + 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 25517fa74c3..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,6 +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=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 d2bbeb90133..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,6 +225,7 @@ def query_generator() -> Iterator[str]: use_cte=use_cte, start_date=start_date, end_date=end_date, + filter_by_created_timestamp=filter_by_created_timestamp, ) finally: # Only cleanup if we created a table @@ -693,6 +697,7 @@ def build_point_in_time_query( use_cte: bool = False, start_date: Optional[datetime] = None, end_date: Optional[datetime] = None, + 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) @@ -723,6 +728,7 @@ def build_point_in_time_query( "use_cte": use_cte, "start_date": start_date, "end_date": end_date, + "filter_by_created_timestamp": filter_by_created_timestamp, } query = template.render(template_context) @@ -965,6 +971,10 @@ def _get_entity_schema( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if filter_by_created_timestamp 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/spark_offline_store/spark.py b/sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py index 44590287ff6..7e0a03e69bb 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) @@ -332,6 +335,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=filter_by_created_timestamp, ) return SparkRetrievalJob( @@ -1792,6 +1796,10 @@ def _cast_data_frame( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if filter_by_created_timestamp 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..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, @@ -356,6 +358,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> TrinoRetrievalJob: assert isinstance(config.offline_store, TrinoOfflineStoreConfig) for fv in feature_views: @@ -417,6 +420,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=filter_by_created_timestamp, ) return TrinoRetrievalJob( @@ -648,6 +652,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 filter_by_created_timestamp 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..807c6a9c6f9 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,6 +317,14 @@ def evaluate_historical_retrieval(): timestamp_field, ) + if filter_by_created_timestamp and created_timestamp_column: + df_to_join = _filter_created_timestamp( + df_to_join, + timestamp_field, + created_timestamp_column, + entity_df_event_timestamp_col, + ) + df_to_join = _drop_duplicates( df_to_join, all_join_keys, @@ -1183,6 +1194,24 @@ def _filter_ttl( return df_to_join +def _filter_created_timestamp( + df_to_join: dd.DataFrame, + timestamp_field: str, + created_timestamp_column: str, + entity_df_event_timestamp_col: 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] + ) + ] + + 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 27940feeb2f..f09d0a9f982 100644 --- a/sdk/python/feast/infra/offline_stores/duckdb.py +++ b/sdk/python/feast/infra/offline_stores/duckdb.py @@ -496,6 +496,8 @@ class DuckDBOfflineStoreConfig(FeastConfigBaseModel): class DuckDBOfflineStore(OfflineStore): + supports_filter_by_created_timestamp = True + @staticmethod def pull_latest_from_table_or_query( config: RepoConfig, @@ -531,6 +533,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: return get_historical_features_ibis( config=config, @@ -544,6 +547,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, + 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..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] @@ -101,6 +103,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 @@ -120,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, @@ -128,6 +134,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 3aebc4e903e..87f07c61dc7 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, + filter_by_created_timestamp: 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, + filter_by_created_timestamp=filter_by_created_timestamp, ) 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, + filter_by_created_timestamp: 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 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] + ) + if ttl: predicates.append( feature_table["event_expire_timestamp"] 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/offline_utils.py b/sdk/python/feast/infra/offline_stores/offline_utils.py index cd2b05a9a60..7ccaf965c9c 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, + 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()) @@ -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, + "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 ec708ccf798..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, @@ -220,6 +222,7 @@ def get_historical_features( registry: BaseRegistry, project: str, full_feature_names: bool = False, + filter_by_created_timestamp: bool = False, ) -> RetrievalJob: assert isinstance(config.offline_store, RedshiftOfflineStoreConfig) for fv in feature_views: @@ -278,6 +281,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=filter_by_created_timestamp, ) try: @@ -1311,6 +1315,10 @@ def _get_entity_df_event_timestamp_range( AND subquery.event_timestamp >= entity_dataframe.entity_timestamp - {{ featureview.ttl }} * interval '1' second {% endif %} + {% if filter_by_created_timestamp 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/snowflake.py b/sdk/python/feast/infra/offline_stores/snowflake.py index 05ed82a1aae..84b829617f8 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake.py +++ b/sdk/python/feast/infra/offline_stores/snowflake.py @@ -299,6 +299,8 @@ def get_historical_features( project: str, full_feature_names: bool = False, ) -> RetrievalJob: + # 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/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/integration/offline_store/test_universal_historical_retrieval.py b/sdk/python/tests/integration/offline_store/test_universal_historical_retrieval.py index af0de479f3e..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 @@ -939,3 +939,100 @@ 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_filter_by_created_timestamp(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]) + + # Default: the backfilled values win the dedup + 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"]) + + try: + job = store.get_historical_features( + entity_df=entity_df, + features=["driver_stats:avg_daily_trips"], + full_feature_names=False, + filter_by_created_timestamp=True, + ) + except NotImplementedError: + 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}, + {"driver_id": 1002, "event_timestamp": tomorrow, "avg_daily_trips": 30}, + ] + ) + validate_dataframes(expected_df, actual_df, sort_by=["driver_id"]) 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 new file mode 100644 index 00000000000..e1353709172 --- /dev/null +++ b/sdk/python/tests/unit/infra/offline_stores/test_filter_by_created_timestamp.py @@ -0,0 +1,267 @@ +from datetime import timedelta +from unittest.mock import MagicMock + +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 +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 + +CREATED_TIMESTAMP_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, + 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=query_template, + full_feature_names=False, + **kwargs, + ) + + +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_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_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 + + +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 + ) + + +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: + 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, + filter_by_created_timestamp=filter_by_created_timestamp, + ) + 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_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 TestIbisFilterByCreatedTimestamp: + def _run(self, filter_by_created_timestamp): + 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", + 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_filter_by_created_timestamp_excludes_values_created_after_entity_timestamp( + self, + ): + assert self._run(True)["conv_rate"].tolist() == [0.4]