@@ -1361,9 +1361,9 @@ def get_historical_features(
             all_on_demand_feature_views,
         ) = utils._get_feature_views_to_use(self.registry, self.project, features)
 
-        # TODO(achal): _group_feature_refs returns the on demand feature views, but it's not passed into the provider.
-        # This is a weird interface quirk - we should revisit the `get_historical_features` to
-        # pass in the on demand feature views as well.
+        # TODO: This is an interim API change. We should revisit the Provider/OfflineStore
+        # historical retrieval interface to avoid ad-hoc argument additions (e.g. passing ODFVs
+        # separately) and define a clearer long-term contract.
 
         # Deliberately disable writing to online store for ODFVs during historical retrieval
         # since it's not applicable in this context.
@@ -1374,7 +1374,6 @@ def get_historical_features(
 
         for odfv in all_on_demand_feature_views:
             odfv.write_to_online_store = False
-
         fvs, odfvs = utils._group_feature_refs(
             _feature_refs,
             all_feature_views,
@@ -1409,6 +1408,7 @@ def get_historical_features(
         job = provider.get_historical_features(
             self.config,
             feature_views,
+            on_demand_feature_views,
             _feature_refs,
             entity_df,
             self.registry,

@@ -465,13 +465,15 @@ def get_historical_features(
         self,
         config: RepoConfig,
         feature_views: List[Union[FeatureView, OnDemandFeatureView]],
+        on_demand_feature_views: List[OnDemandFeatureView],
         feature_refs: List[str],
         entity_df: Optional[Union[pd.DataFrame, str]],
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool,
         **kwargs,
     ) -> RetrievalJob:
+        del on_demand_feature_views
         job = self.offline_store.get_historical_features(
             config=config,
             feature_views=feature_views,

@@ -250,6 +250,7 @@ def get_historical_features(
         self,
         config: RepoConfig,
         feature_views: List[Union[FeatureView, OnDemandFeatureView]],
+        on_demand_feature_views: List[OnDemandFeatureView],
         feature_refs: List[str],
         entity_df: Optional[Union[pd.DataFrame, str]],
         registry: BaseRegistry,
@@ -263,6 +264,7 @@ def get_historical_features(
         Args:
             config: The config for the current feature store.
             feature_views: A list containing all feature views that are referenced in the entity rows.
+            on_demand_feature_views: The on demand feature views requested as part of the retrieval.
             feature_refs: The features to be retrieved.
             entity_df: A collection of rows containing all entity columns on which features need to be joined,
                 as well as the timestamp column used for point-in-time joins. Either a pandas dataframe can be

@@ -26,6 +26,7 @@
     ProviderAsyncMethods,
     SupportedAsyncMethods,
 )
+from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.online_response import OnlineResponse
 from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto
 from feast.protos.feast.types.Value_pb2 import RepeatedValue
@@ -98,12 +99,14 @@ def get_historical_features(
         self,
         config: RepoConfig,
         feature_views: List[FeatureView],
+        on_demand_feature_views: List[OnDemandFeatureView],
         feature_refs: List[str],
         entity_df: Union[pandas.DataFrame, str],
         registry: BaseRegistry,
         project: str,
         full_feature_names: bool = False,
     ) -> RetrievalJob:
+        del on_demand_feature_views
         return RetrievalJob()
 
     def online_read(

@@ -0,0 +1,63 @@
+from types import SimpleNamespace
+
+from feast.feature_store import FeatureStore
+from feast.infra.offline_stores.offline_store import RetrievalJob
+
+
+class _CapturingProvider:
+    def __init__(self):
+        self.captured_on_demand_feature_views = None
+
+    def get_historical_features(
+        self,
+        config,
+        feature_views,
+        on_demand_feature_views,
+        feature_refs,
+        entity_df,
+        registry,
+        project,
+        full_feature_names,
+        **kwargs,
+    ):
+        self.captured_on_demand_feature_views = on_demand_feature_views
+        return RetrievalJob()
+
+
+def test_feature_store_passes_on_demand_feature_views_to_provider(monkeypatch):
+    provider = _CapturingProvider()
+
+    store = FeatureStore.__new__(FeatureStore)
+    store.config = SimpleNamespace(project="test_project", coerce_tz_aware=False)
+    store._registry = object()
+    store._get_provider = lambda: provider
+
+    fv = object()
+    odfv = object()
+
+    monkeypatch.setattr(
+        "feast.utils._get_features",
+        lambda registry, project, features, allow_cache=False: ["odfv:feat"],
+    )
+    monkeypatch.setattr(
+        "feast.utils._get_feature_views_to_use",
+        lambda registry, project, features, allow_cache=False, hide_dummy_entity=True: (
+            [fv],
+            [odfv],
+        ),
+    )
+    monkeypatch.setattr(
+        "feast.utils._group_feature_refs",
+        lambda feature_refs, feature_views, on_demand_feature_views: (
+            [(fv, [])],
+            [(odfv, [])],
+        ),
+    )
+    monkeypatch.setattr(
+        "feast.utils._validate_feature_refs",
+        lambda *args, **kwargs: None,
+    )
+
+    store.get_historical_features(entity_df="SELECT 1", features=["odfv:feat"])
+
+    assert provider.captured_on_demand_feature_views == [odfv]