feast-dev
@@ -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 %}
+
@@ -1940,6 +1940,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.
 
@@ -1971,6 +1972,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.
@@ -2072,11 +2078,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()
 

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

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

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

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

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

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

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

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

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