feat: Enable materialization for ODFV Transform on Write (feast-dev#5459)

franciscojavierarceo · web-flow · commit 3d17892fe493 · 2025-06-23T20:06:45.000+05:30
* feat: Enable materialization for ODFV Transform on Write

Signed-off-by: Francisco Javier Arceo &lt;farceo@redhat.com&gt;
diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py
@@ -656,7 +656,7 @@ def _make_inferences(
     def _get_feature_views_to_materialize(
         self,
         feature_views: Optional[List[str]],
-    ) -> List[FeatureView]:
+    ) -> List[Union[FeatureView, OnDemandFeatureView]]:
         """
         Returns the list of feature views that should be materialized.
 
@@ -669,34 +669,53 @@ def _get_feature_views_to_materialize(
             FeatureViewNotFoundException: One of the specified feature views could not be found.
             ValueError: One of the specified feature views is not configured for materialization.
         """
-        feature_views_to_materialize: List[FeatureView] = []
+        feature_views_to_materialize: List[Union[FeatureView, OnDemandFeatureView]] = []
 
         if feature_views is None:
-            feature_views_to_materialize = utils._list_feature_views(
+            regular_feature_views = utils._list_feature_views(
                 self._registry, self.project, hide_dummy_entity=False
             )
-            feature_views_to_materialize = [
-                fv for fv in feature_views_to_materialize if fv.online
-            ]
+            feature_views_to_materialize.extend(
+                [fv for fv in regular_feature_views if fv.online]
+            )
             stream_feature_views_to_materialize = self._list_stream_feature_views(
                 hide_dummy_entity=False
             )
-            feature_views_to_materialize += [
-                sfv for sfv in stream_feature_views_to_materialize if sfv.online
-            ]
+            feature_views_to_materialize.extend(
+                [sfv for sfv in stream_feature_views_to_materialize if sfv.online]
+            )
+            on_demand_feature_views_to_materialize = self.list_on_demand_feature_views()
+            feature_views_to_materialize.extend(
+                [
+                    odfv
+                    for odfv in on_demand_feature_views_to_materialize
+                    if odfv.write_to_online_store
+                ]
+            )
         else:
             for name in feature_views:
+                feature_view: Union[FeatureView, OnDemandFeatureView]
                 try:
                     feature_view = self._get_feature_view(name, hide_dummy_entity=False)
                 except FeatureViewNotFoundException:
-                    feature_view = self._get_stream_feature_view(
-                        name, hide_dummy_entity=False
-                    )
+                    try:
+                        feature_view = self._get_stream_feature_view(
+                            name, hide_dummy_entity=False
+                        )
+                    except FeatureViewNotFoundException:
+                        feature_view = self.get_on_demand_feature_view(name)
 
-                if not feature_view.online:
+                if hasattr(feature_view, "online") and not feature_view.online:
                     raise ValueError(
                         f"FeatureView {feature_view.name} is not configured to be served online."
                     )
+                elif (
+                    hasattr(feature_view, "write_to_online_store")
+                    and not feature_view.write_to_online_store
+                ):
+                    raise ValueError(
+                        f"OnDemandFeatureView {feature_view.name} is not configured for write_to_online_store."
+                    )
                 feature_views_to_materialize.append(feature_view)
 
         return feature_views_to_materialize
@@ -1312,6 +1331,8 @@ def materialize_incremental(
         )
         # TODO paging large loads
         for feature_view in feature_views_to_materialize:
+            if isinstance(feature_view, OnDemandFeatureView):
+                continue
             start_date = feature_view.most_recent_end_time
             if start_date is None:
                 if feature_view.ttl is None:
@@ -1340,7 +1361,7 @@ def tqdm_builder(length):
                 return tqdm(total=length, ncols=100)
 
             start_date = utils.make_tzaware(start_date)
-            end_date = utils.make_tzaware(end_date)
+            end_date = utils.make_tzaware(end_date) or _utc_now()
 
             provider.materialize_single_feature_view(
                 config=self.config,
@@ -1351,13 +1372,13 @@ def tqdm_builder(length):
                 project=self.project,
                 tqdm_builder=tqdm_builder,
             )
-
-            self._registry.apply_materialization(
-                feature_view,
-                self.project,
-                start_date,
-                end_date,
-            )
+            if not isinstance(feature_view, OnDemandFeatureView):
+                self._registry.apply_materialization(
+                    feature_view,
+                    self.project,
+                    start_date,
+                    end_date,
+                )
 
     def materialize(
         self,
diff --git a/sdk/python/feast/infra/passthrough_provider.py b/sdk/python/feast/infra/passthrough_provider.py
@@ -420,17 +420,24 @@ def ingest_df_to_offline_store(self, feature_view: FeatureView, table: pa.Table)
     def materialize_single_feature_view(
         self,
         config: RepoConfig,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         start_date: datetime,
         end_date: datetime,
         registry: BaseRegistry,
         project: str,
         tqdm_builder: Callable[[int], tqdm],
     ) -> None:
+        if isinstance(feature_view, OnDemandFeatureView):
+            if not feature_view.write_to_online_store:
+                raise ValueError(
+                    f"OnDemandFeatureView {feature_view.name} does not have write_to_online_store enabled"
+                )
+            return
         assert (
             isinstance(feature_view, BatchFeatureView)
             or isinstance(feature_view, StreamFeatureView)
             or isinstance(feature_view, FeatureView)
+            or isinstance(feature_view, OnDemandFeatureView)
         ), f"Unexpected type for {feature_view.name}: {type(feature_view)}"
         task = MaterializationTask(
             project=project,
diff --git a/sdk/python/feast/infra/provider.py b/sdk/python/feast/infra/provider.py
@@ -217,7 +217,7 @@ def ingest_df_to_offline_store(
     def materialize_single_feature_view(
         self,
         config: RepoConfig,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         start_date: datetime,
         end_date: datetime,
         registry: BaseRegistry,
diff --git a/sdk/python/feast/infra/registry/base_registry.py b/sdk/python/feast/infra/registry/base_registry.py
@@ -16,7 +16,7 @@
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 
 from google.protobuf.json_format import MessageToJson
 from google.protobuf.message import Message
@@ -432,7 +432,7 @@ def list_all_feature_views(
     @abstractmethod
     def apply_materialization(
         self,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         project: str,
         start_date: datetime,
         end_date: datetime,
diff --git a/sdk/python/feast/infra/registry/registry.py b/sdk/python/feast/infra/registry/registry.py
@@ -16,7 +16,7 @@
 from enum import Enum
 from pathlib import Path
 from threading import Lock
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Union
 from urllib.parse import urlparse
 
 from google.protobuf.internal.containers import RepeatedCompositeFieldContainer
@@ -529,7 +529,7 @@ def get_data_source(
 
     def apply_materialization(
         self,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         project: str,
         start_date: datetime,
         end_date: datetime,
diff --git a/sdk/python/feast/infra/registry/remote.py b/sdk/python/feast/infra/registry/remote.py
@@ -356,7 +356,7 @@ def list_feature_views(
 
     def apply_materialization(
         self,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         project: str,
         start_date: datetime,
         end_date: datetime,
diff --git a/sdk/python/feast/infra/registry/snowflake.py b/sdk/python/feast/infra/registry/snowflake.py
@@ -992,7 +992,7 @@ def list_permissions(
 
     def apply_materialization(
         self,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         project: str,
         start_date: datetime,
         end_date: datetime,
diff --git a/sdk/python/feast/infra/registry/sql.py b/sdk/python/feast/infra/registry/sql.py
@@ -702,7 +702,7 @@ def apply_validation_reference(
 
     def apply_materialization(
         self,
-        feature_view: FeatureView,
+        feature_view: Union[FeatureView, OnDemandFeatureView],
         project: str,
         start_date: datetime,
         end_date: datetime,
diff --git a/sdk/python/tests/unit/test_on_demand_python_transformation.py b/sdk/python/tests/unit/test_on_demand_python_transformation.py
