fix: Fix issue with apply feature view

Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>

Author: ntkathole

sdk/python/feast/feature_view.py

@@ -520,14 +520,17 @@ def get_ttl_duration(self):
         return ttl_duration
 
     @classmethod
-    def from_proto(cls, feature_view_proto: FeatureViewProto) -> "FeatureView":
-        return cls._from_proto_internal(feature_view_proto, seen={})
+    def from_proto(
+        cls, feature_view_proto: FeatureViewProto, skip_udf: bool = False
+    ) -> "FeatureView":
+        return cls._from_proto_internal(feature_view_proto, seen={}, skip_udf=skip_udf)
 
     @classmethod
     def _from_proto_internal(
         cls,
         feature_view_proto: FeatureViewProto,
         seen: Dict[str, Union[None, "FeatureView"]],
+        skip_udf: bool = False,
     ) -> "FeatureView":
         """
         Creates a feature view from a protobuf representation of a feature view.
@@ -561,7 +564,7 @@ def _from_proto_internal(
         )
         source_views = [
             FeatureView._from_proto_internal(
-                FeatureViewProto(spec=view_spec, meta=None), seen
+                FeatureViewProto(spec=view_spec, meta=None), seen, skip_udf=skip_udf
             )
             for view_spec in feature_view_proto.spec.source_views
         ]
@@ -581,22 +584,23 @@ def _from_proto_internal(
             )
             transformation = None
 
-            if feature_transformation_proto.HasField("user_defined_function"):
-                udf_proto = feature_transformation_proto.user_defined_function
-                if udf_proto.mode:
-                    try:
-                        transformation_class = get_transformation_class_from_type(
-                            udf_proto.mode
-                        )
-                        transformation = transformation_class.from_proto(udf_proto)
-                    except (ValueError, KeyError):
+            if not skip_udf:
+                if feature_transformation_proto.HasField("user_defined_function"):
+                    udf_proto = feature_transformation_proto.user_defined_function
+                    if udf_proto.mode:
+                        try:
+                            transformation_class = get_transformation_class_from_type(
+                                udf_proto.mode
+                            )
+                            transformation = transformation_class.from_proto(udf_proto)
+                        except (ValueError, KeyError):
+                            transformation = PythonTransformation.from_proto(udf_proto)
+                    else:
                         transformation = PythonTransformation.from_proto(udf_proto)
-                else:
-                    transformation = PythonTransformation.from_proto(udf_proto)
-            elif feature_transformation_proto.HasField("substrait_transformation"):
-                transformation = SubstraitTransformation.from_proto(
-                    feature_transformation_proto.substrait_transformation
-                )
+                elif feature_transformation_proto.HasField("substrait_transformation"):
+                    transformation = SubstraitTransformation.from_proto(
+                        feature_transformation_proto.substrait_transformation
+                    )
 
             mode: Union[TransformationMode, str]
             if feature_view_proto.spec.mode:

sdk/python/feast/on_demand_feature_view.py

@@ -140,7 +140,7 @@ class OnDemandFeatureView(BaseFeatureView):
     features: List[Field]
     source_feature_view_projections: dict[str, FeatureViewProjection]
     source_request_sources: dict[str, RequestSource]
-    feature_transformation: Transformation
+    feature_transformation: Optional[Transformation]
     mode: str
     description: str
     tags: dict[str, str]
@@ -259,9 +259,12 @@ def __init__(  # noqa: C901
                 features.append(field)
 
         self.features = features
-        self.feature_transformation = (
-            feature_transformation or self.get_feature_transformation()
-        )
+        if feature_transformation is not None:
+            self.feature_transformation = feature_transformation
+        elif self.udf is not None:
+            self.feature_transformation = self.get_feature_transformation()
+        else:
+            self.feature_transformation = None
         self.write_to_online_store = write_to_online_store
         self.singleton = singleton
         if self.singleton and self.mode != "python":
@@ -578,11 +581,13 @@ def from_proto(
             A OnDemandFeatureView object based on the on-demand feature view protobuf.
         """
         # Parse sources from proto
-        sources = cls._parse_sources_from_proto(on_demand_feature_view_proto)
+        sources = cls._parse_sources_from_proto(
+            on_demand_feature_view_proto, skip_udf=skip_udf
+        )
 
-        # Parse transformation from proto
+        # Parse transformation from proto (skip UDF deserialization if requested)
         transformation = cls._parse_transformation_from_proto(
-            on_demand_feature_view_proto
+            on_demand_feature_view_proto, skip_udf=skip_udf
         )
 
         # Parse optional fields with defaults
@@ -643,7 +648,7 @@ def from_proto(
 
     @classmethod
     def _parse_sources_from_proto(
-        cls, proto: OnDemandFeatureViewProto
+        cls, proto: OnDemandFeatureViewProto, skip_udf: bool = False
     ) -> List[OnDemandSourceType]:
         """Parse and convert sources from the protobuf representation."""
         sources: List[OnDemandSourceType] = []
@@ -652,7 +657,9 @@ def _parse_sources_from_proto(
 
             if source_type == "feature_view":
                 sources.append(
-                    FeatureView.from_proto(on_demand_source.feature_view).projection
+                    FeatureView.from_proto(
+                        on_demand_source.feature_view, skip_udf=skip_udf
+                    ).projection
                 )
             elif source_type == "feature_view_projection":
                 sources.append(
@@ -673,9 +680,12 @@ def _parse_sources_from_proto(
 
     @classmethod
     def _parse_transformation_from_proto(
-        cls, proto: OnDemandFeatureViewProto
-    ) -> Transformation:
+        cls, proto: OnDemandFeatureViewProto, skip_udf: bool = False
+    ) -> Optional[Transformation]:
         """Parse and convert the transformation from the protobuf representation."""
+        if skip_udf:
+            return None
+
         feature_transformation = proto.spec.feature_transformation
         transformation_type = feature_transformation.WhichOneof("transformation")
         mode = proto.spec.mode
@@ -898,6 +908,7 @@ def transform_arrow(
         pa_table, columns_to_cleanup = self._preprocess_arrow_table(pa_table)
 
         # Apply the transformation
+        assert self.feature_transformation is not None
         transformed_table = self.feature_transformation.transform_arrow(
             pa_table, self.features
         )
@@ -983,6 +994,7 @@ def transform_dict(
         )
 
         # Apply the appropriate transformation based on mode
+        assert self.feature_transformation is not None
         if self.singleton and self.mode == "python":
             output_dict = self.feature_transformation.transform_singleton(
                 preprocessed_dict
@@ -1024,6 +1036,7 @@ def _preprocess_feature_dict(
         return preprocessed_dict, columns_to_cleanup
 
     def infer_features(self) -> None:
+        assert self.feature_transformation is not None
         random_input = self._construct_random_input(singleton=self.singleton)
         inferred_features = self.feature_transformation.infer_features(
             random_input=random_input, singleton=self.singleton

sdk/python/feast/registry_server.py

@@ -342,22 +342,35 @@ def ApplyFeatureView(
         self, request: RegistryServer_pb2.ApplyFeatureViewRequest, context
     ):
         feature_view_type = request.WhichOneof("base_feature_view")
+
         if feature_view_type == "feature_view":
-            feature_view = FeatureView.from_proto(request.feature_view)
+            feature_view_meta = FeatureView.from_proto(
+                request.feature_view, skip_udf=True
+            )
         elif feature_view_type == "on_demand_feature_view":
-            feature_view = OnDemandFeatureView.from_proto(
-                request.on_demand_feature_view
+            feature_view_meta = OnDemandFeatureView.from_proto(
+                request.on_demand_feature_view, skip_udf=True
             )
         elif feature_view_type == "stream_feature_view":
-            feature_view = StreamFeatureView.from_proto(request.stream_feature_view)
+            feature_view_meta = StreamFeatureView.from_proto(
+                request.stream_feature_view, skip_udf=True
+            )
 
         assert_permissions_to_update(
-            resource=feature_view,
-            # Will replace with the new get_any_feature_view method later
+            resource=feature_view_meta,
             getter=self.proxied_registry.get_feature_view,
             project=request.project,
         )
 
+        if feature_view_type == "feature_view":
+            feature_view = FeatureView.from_proto(request.feature_view)
+        elif feature_view_type == "on_demand_feature_view":
+            feature_view = OnDemandFeatureView.from_proto(
+                request.on_demand_feature_view
+            )
+        elif feature_view_type == "stream_feature_view":
+            feature_view = StreamFeatureView.from_proto(request.stream_feature_view)
+
         (
             self.proxied_registry.apply_feature_view(
                 feature_view=feature_view,

sdk/python/feast/stream_feature_view.py

@@ -320,7 +320,7 @@ def to_proto(self):
         return StreamFeatureViewProto(spec=spec, meta=meta)
 
     @classmethod
-    def from_proto(cls, sfv_proto):
+    def from_proto(cls, sfv_proto, skip_udf: bool = False):
         batch_source = (
             DataSource.from_proto(sfv_proto.spec.batch_source)
             if sfv_proto.spec.HasField("batch_source")
@@ -333,7 +333,7 @@ def from_proto(cls, sfv_proto):
         )
         udf = (
             dill.loads(sfv_proto.spec.user_defined_function.body)
-            if sfv_proto.spec.HasField("user_defined_function")
+            if sfv_proto.spec.HasField("user_defined_function") and not skip_udf
             else None
         )
         udf_string = (