fix: Fix on demand feature view output in feast plan + Web UI crash (feast-dev#3057)

* fix: Fix on demand feature view output in feast plan + Web UI crash with ODFV

Signed-off-by: Danny Chiao <[email protected]>

* lint

Signed-off-by: Danny Chiao <[email protected]>

* fix tests

Signed-off-by: Danny Chiao <[email protected]>

Signed-off-by: Danny Chiao <[email protected]>

Author: adchia

protos/feast/core/OnDemandFeatureView.proto

@@ -83,4 +83,7 @@ message UserDefinedFunction {
 
     // The python-syntax function body (serialized by dill)
     bytes body = 2;
+
+    // The string representation of the udf
+    string body_text = 3;
 }

sdk/python/feast/diff/registry_diff.py

@@ -17,6 +17,7 @@
 from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
     OnDemandFeatureView as OnDemandFeatureViewProto,
 )
+from feast.protos.feast.core.OnDemandFeatureView_pb2 import OnDemandFeatureViewSpec
 from feast.protos.feast.core.RequestFeatureView_pb2 import (
     RequestFeatureView as RequestFeatureViewProto,
 )
@@ -137,19 +138,39 @@ def diff_registry_objects(
     else:
         current_spec = current_proto.spec
         new_spec = new_proto.spec
-    if current_spec != new_spec:
+    if current != new:
         for _field in current_spec.DESCRIPTOR.fields:
             if _field.name in FIELDS_TO_IGNORE:
                 continue
-            if getattr(current_spec, _field.name) != getattr(new_spec, _field.name):
-                transition = TransitionType.UPDATE
-                property_diffs.append(
-                    PropertyDiff(
-                        _field.name,
-                        getattr(current_spec, _field.name),
-                        getattr(new_spec, _field.name),
+            elif getattr(current_spec, _field.name) != getattr(new_spec, _field.name):
+                if _field.name == "user_defined_function":
+                    current_spec = cast(OnDemandFeatureViewSpec, current_proto)
+                    new_spec = cast(OnDemandFeatureViewSpec, new_proto)
+                    current_udf = current_spec.user_defined_function
+                    new_udf = new_spec.user_defined_function
+                    for _udf_field in current_udf.DESCRIPTOR.fields:
+                        if _udf_field.name == "body":
+                            continue
+                        if getattr(current_udf, _udf_field.name) != getattr(
+                            new_udf, _udf_field.name
+                        ):
+                            transition = TransitionType.UPDATE
+                            property_diffs.append(
+                                PropertyDiff(
+                                    _field.name + "." + _udf_field.name,
+                                    getattr(current_udf, _udf_field.name),
+                                    getattr(new_udf, _udf_field.name),
+                                )
+                            )
+                else:
+                    transition = TransitionType.UPDATE
+                    property_diffs.append(
+                        PropertyDiff(
+                            _field.name,
+                            getattr(current_spec, _field.name),
+                            getattr(new_spec, _field.name),
+                        )
                     )
-                )
     return FeastObjectDiff(
         name=new_spec.name,
         feast_object_type=object_type,

sdk/python/feast/on_demand_feature_view.py

@@ -59,12 +59,12 @@ class OnDemandFeatureView(BaseFeatureView):
             maintainer.
     """
 
-    # TODO(adchia): remove inputs from proto and declaration
     name: str
     features: List[Field]
     source_feature_view_projections: Dict[str, FeatureViewProjection]
     source_request_sources: Dict[str, RequestSource]
     udf: FunctionType
+    udf_string: str
     description: str
     tags: Dict[str, str]
     owner: str
@@ -83,6 +83,7 @@ def __init__(  # noqa: C901
             ]
         ],
         udf: FunctionType,
+        udf_string: str = "",
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         owner: str = "",
@@ -99,6 +100,7 @@ def __init__(  # noqa: C901
                 which will refer to them by name.
             udf: The user defined transformation function, which must take pandas
                 dataframes as inputs.
+            udf_string: The source code version of the udf (for diffing and displaying in Web UI)
             description (optional): A human-readable description.
             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
             owner (optional): The owner of the on demand feature view, typically the email
@@ -125,6 +127,7 @@ def __init__(  # noqa: C901
                 ] = odfv_source.projection
 
         self.udf = udf  # type: ignore
+        self.udf_string = udf_string
 
     @property
     def proto_class(self) -> Type[OnDemandFeatureViewProto]:
@@ -137,6 +140,7 @@ def __copy__(self):
             sources=list(self.source_feature_view_projections.values())
             + list(self.source_request_sources.values()),
             udf=self.udf,
+            udf_string=self.udf_string,
             description=self.description,
             tags=self.tags,
             owner=self.owner,
@@ -157,6 +161,7 @@ def __eq__(self, other):
             self.source_feature_view_projections
             != other.source_feature_view_projections
             or self.source_request_sources != other.source_request_sources
+            or self.udf_string != other.udf_string
             or self.udf.__code__.co_code != other.udf.__code__.co_code
         ):
             return False
@@ -198,6 +203,7 @@ def to_proto(self) -> OnDemandFeatureViewProto:
             user_defined_function=UserDefinedFunctionProto(
                 name=self.udf.__name__,
                 body=dill.dumps(self.udf, recurse=True),
+                body_text=self.udf_string,
             ),
             description=self.description,
             tags=self.tags,
@@ -250,6 +256,7 @@ def from_proto(cls, on_demand_feature_view_proto: OnDemandFeatureViewProto):
             udf=dill.loads(
                 on_demand_feature_view_proto.spec.user_defined_function.body
             ),
+            udf_string=on_demand_feature_view_proto.spec.user_defined_function.body_text,
             description=on_demand_feature_view_proto.spec.description,
             tags=dict(on_demand_feature_view_proto.spec.tags),
             owner=on_demand_feature_view_proto.spec.owner,
@@ -438,6 +445,7 @@ def mainify(obj):
             obj.__module__ = "__main__"
 
     def decorator(user_function):
+        udf_string = dill.source.getsource(user_function)
         mainify(user_function)
         on_demand_feature_view_obj = OnDemandFeatureView(
             name=user_function.__name__,
@@ -447,6 +455,7 @@ def decorator(user_function):
             description=description,
             tags=tags,
             owner=owner,
+            udf_string=udf_string,
         )
         functools.update_wrapper(
             wrapper=on_demand_feature_view_obj, wrapped=user_function

sdk/python/feast/registry.py

@@ -24,7 +24,6 @@
 from typing import Any, Dict, List, Optional
 from urllib.parse import urlparse
 
-import dill
 from google.protobuf.internal.containers import RepeatedCompositeFieldContainer
 from google.protobuf.json_format import MessageToJson
 from proto import Message
@@ -732,9 +731,10 @@ def to_dict(self, project: str) -> Dict[str, List[Any]]:
             key=lambda on_demand_feature_view: on_demand_feature_view.name,
         ):
             odfv_dict = self._message_to_sorted_dict(on_demand_feature_view.to_proto())
-            odfv_dict["spec"]["userDefinedFunction"]["body"] = dill.source.getsource(
-                on_demand_feature_view.udf
-            )
+
+            odfv_dict["spec"]["userDefinedFunction"][
+                "body"
+            ] = on_demand_feature_view.udf_string
             registry_dict["onDemandFeatureViews"].append(odfv_dict)
         for request_feature_view in sorted(
             self.list_request_feature_views(project=project),

sdk/python/tests/integration/feature_repos/universal/feature_views.py

@@ -69,6 +69,7 @@ def conv_rate_plus_100_feature_view(
         schema=[] if infer_features else _features,
         sources=sources,
         udf=conv_rate_plus_100,
+        udf_string="raw udf source",
     )
 
 
@@ -106,6 +107,7 @@ def similarity_feature_view(
         sources=sources,
         schema=[] if infer_features else _fields,
         udf=similarity,
+        udf_string="similarity raw udf",
     )
 
 

sdk/python/tests/unit/test_on_demand_feature_view.py

@@ -55,6 +55,7 @@ def test_hash():
             Field(name="output2", dtype=Float32),
         ],
         udf=udf1,
+        udf_string="udf1 source code",
     )
     on_demand_feature_view_2 = OnDemandFeatureView(
         name="my-on-demand-feature-view",
@@ -64,6 +65,7 @@ def test_hash():
             Field(name="output2", dtype=Float32),
         ],
         udf=udf1,
+        udf_string="udf1 source code",
     )
     on_demand_feature_view_3 = OnDemandFeatureView(
         name="my-on-demand-feature-view",
@@ -73,6 +75,7 @@ def test_hash():
             Field(name="output2", dtype=Float32),
         ],
         udf=udf2,
+        udf_string="udf2 source code",
     )
     on_demand_feature_view_4 = OnDemandFeatureView(
         name="my-on-demand-feature-view",
@@ -82,6 +85,7 @@ def test_hash():
             Field(name="output2", dtype=Float32),
         ],
         udf=udf2,
+        udf_string="udf2 source code",
         description="test",
     )