feat: Add opt-in at_event_time created_timestamp cutoff to get_histor…

…ical_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 <noreply@anthropic.com> Signed-off-by: David <david-adeniji@hotmail.co.uk>
ntkathole · Jul 31, 2026 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026 · Jul 19, 2026
commit 5b226efecacf7b93c5dfe2c4be079b59d5ac80b6
@@ -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()
 

@@ -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 %}

@@ -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 %}

@@ -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 %}
 ),
 
 /*

@@ -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 %}

@@ -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

@@ -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

@@ -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
@@ -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)
 

@@ -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 %}

@@ -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 %}

@@ -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],

@@ -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

@@ -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"]

@@ -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,
     }
 

@@ -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 %}