feat: Updating protos to separate transformation

protos/feast/core/OnDemandFeatureView.proto

@@ -27,6 +27,7 @@ import "feast/core/FeatureView.proto";
 import "feast/core/FeatureViewProjection.proto";
 import "feast/core/Feature.proto";
 import "feast/core/DataSource.proto";
+import "feast/core/Transformation.proto";
 
 message OnDemandFeatureView {
     // User-specified specifications of this feature view.
@@ -49,9 +50,11 @@ message OnDemandFeatureViewSpec {
     map<string, OnDemandSource> sources = 4;
 
     oneof transformation {
-        UserDefinedFunction user_defined_function = 5;
-        OnDemandSubstraitTransformation on_demand_substrait_transformation = 9;
+        UserDefinedFunction user_defined_function = 5 [deprecated = true];
+        OnDemandSubstraitTransformation on_demand_substrait_transformation = 9 [deprecated = true];
     }
+    // Oneof with {user_defined_function, on_demand_substrait_transformation}
+    FeatureTransformationV2 feature_transformation = 10;
 
     // Description of the on demand feature view.
     string description = 6;
@@ -61,6 +64,7 @@ message OnDemandFeatureViewSpec {
 
     // Owner of the on demand feature view.
     string owner = 8;
+    string mode = 11;
 }
 
 message OnDemandFeatureViewMeta {
@@ -81,6 +85,8 @@ message OnDemandSource {
 
 // Serialized representation of python function.
 message UserDefinedFunction {
+    option deprecated = true;
+
     // The function name
     string name = 1;
 
@@ -92,5 +98,7 @@ message UserDefinedFunction {
 }
 
 message OnDemandSubstraitTransformation {
+    option deprecated = true;
+
     bytes substrait_plan = 1;
-}
+}

protos/feast/core/StreamFeatureView.proto

@@ -29,6 +29,7 @@ import "feast/core/FeatureView.proto";
 import "feast/core/Feature.proto";
 import "feast/core/DataSource.proto";
 import "feast/core/Aggregation.proto";
+import "feast/core/Transformation.proto";
 
 message StreamFeatureView {
     // User-specified specifications of this feature view.
@@ -77,7 +78,8 @@ message StreamFeatureViewSpec {
     bool online = 12;
 
     // Serialized function that is encoded in the streamfeatureview
-    UserDefinedFunction user_defined_function = 13;
+    UserDefinedFunction user_defined_function = 13 [deprecated = true];
+
 
     // Mode of execution
     string mode = 14;
@@ -87,5 +89,8 @@ message StreamFeatureViewSpec {
 
     // Timestamp field for aggregation
     string timestamp_field = 16;
+
+    // Oneof with {user_defined_function, on_demand_substrait_transformation}
+    FeatureTransformationV2 feature_transformation = 17;
 }
 

protos/feast/core/Transformation.proto

@@ -0,0 +1,33 @@
+syntax = "proto3";
+package feast.core;
+
+option go_package = "github.com/feast-dev/feast/go/protos/feast/core";
+option java_outer_classname = "FeatureTransformationProto";
+option java_package = "feast.proto.core";
+
+import "google/protobuf/duration.proto";
+
+// Serialized representation of python function.
+message UserDefinedFunctionV2 {
+    // The function name
+    string name = 1;
+
+    // The python-syntax function body (serialized by dill)
+    bytes body = 2;
+
+    // The string representation of the udf
+    string body_text = 3;
+}
+
+// A feature transformation executed as a user-defined function
+message FeatureTransformationV2 {
+  // Note this Transformation starts at 5 for backwards compatibility
+  oneof transformation {
+    UserDefinedFunctionV2 user_defined_function = 1;
+    OnDemandSubstraitTransformationV2 on_demand_substrait_transformation = 2;
+  }
+}
+
+message OnDemandSubstraitTransformationV2 {
+  bytes substrait_plan = 1;
+}

sdk/python/feast/diff/registry_diff.py

@@ -1,3 +1,4 @@
+import warnings
 from dataclasses import dataclass
 from typing import Any, Dict, Iterable, List, Optional, Set, Tuple, TypeVar, cast
 
@@ -144,11 +145,25 @@ def diff_registry_objects(
             if _field.name in FIELDS_TO_IGNORE:
                 continue
             elif getattr(current_spec, _field.name) != getattr(new_spec, _field.name):
-                if _field.name == "user_defined_function":
+                # TODO: Delete "transformation" after we've safely deprecated it from the proto
+                if _field.name in ["transformation", "feature_transformation"]:
+                    warnings.warn(
+                        "transformation will be deprecated in the future please use feature_transformation instead.",
+                        DeprecationWarning,
+                    )
                     current_spec = cast(OnDemandFeatureViewSpec, current_spec)
                     new_spec = cast(OnDemandFeatureViewSpec, new_spec)
-                    current_udf = current_spec.user_defined_function
-                    new_udf = new_spec.user_defined_function
+                    # Check if the old proto is populated and use that if it is
+                    deprecated_udf = current_spec.user_defined_function
+                    feature_transformation_udf = (
+                        current_spec.feature_transformation.user_defined_function
+                    )
+                    current_udf = (
+                        deprecated_udf
+                        if deprecated_udf.body_text != ""
+                        else feature_transformation_udf
+                    )
+                    new_udf = new_spec.feature_transformation.user_defined_function
                     for _udf_field in current_udf.DESCRIPTOR.fields:
                         if _udf_field.name == "body":
                             continue

sdk/python/feast/infra/registry/base_registry.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import json
+import warnings
 from abc import ABC, abstractmethod
 from collections import defaultdict
 from datetime import datetime
@@ -662,10 +663,16 @@ 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"][
+            # We are logging a warning because the registry object may be read from a proto that is not updated
+            # i.e., we have to submit dual writes but in order to ensure the read behavior succeeds we have to load
+            # both objects to compare any changes in the registry
+            warnings.warn(
+                "We will be deprecating the usage of spec.userDefinedFunction in a future release please upgrade cautiously.",
+                DeprecationWarning,
+            )
+            odfv_dict["spec"]["featureTransformation"]["userDefinedFunction"][
                 "body"
-            ] = on_demand_feature_view.transformation.udf_string
+            ] = on_demand_feature_view.feature_transformation.udf_string
             registry_dict["onDemandFeatureViews"].append(odfv_dict)
         for request_feature_view in sorted(
             self.list_request_feature_views(project=project),
@@ -684,6 +691,7 @@ def to_dict(self, project: str) -> Dict[str, List[Any]]:
                 "body"
             ] = stream_feature_view.udf_string
             registry_dict["streamFeatureViews"].append(sfv_dict)
+
         for saved_dataset in sorted(
             self.list_saved_datasets(project=project), key=lambda item: item.name
         ):

sdk/python/feast/on_demand_feature_view.py

@@ -27,6 +27,12 @@
     OnDemandFeatureViewSpec,
     OnDemandSource,
 )
+from feast.protos.feast.core.Transformation_pb2 import (
+    FeatureTransformationV2 as FeatureTransformationProto,
+)
+from feast.protos.feast.core.Transformation_pb2 import (
+    UserDefinedFunctionV2 as UserDefinedFunctionProto,
+)
 from feast.type_map import (
     feast_value_type_to_pandas_type,
     python_type_to_feast_value_type,
@@ -63,6 +69,7 @@ class OnDemandFeatureView(BaseFeatureView):
     source_feature_view_projections: Dict[str, FeatureViewProjection]
     source_request_sources: Dict[str, RequestSource]
     transformation: Union[OnDemandPandasTransformation]
+    feature_transformation: Union[OnDemandPandasTransformation]
     description: str
     tags: Dict[str, str]
     owner: str
@@ -83,6 +90,7 @@ def __init__(  # noqa: C901
         udf: Optional[FunctionType] = None,
         udf_string: str = "",
         transformation: Optional[Union[OnDemandPandasTransformation]] = None,
+        feature_transformation: Optional[Union[OnDemandPandasTransformation]] = None,
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         owner: str = "",
@@ -101,6 +109,7 @@ def __init__(  # noqa: C901
                 dataframes as inputs.
             udf_string (deprecated): The source code version of the udf (for diffing and displaying in Web UI)
             transformation: The user defined transformation.
+            feature_transformation: The user defined transformation.
             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
@@ -139,6 +148,7 @@ def __init__(  # noqa: C901
                 ] = odfv_source.projection
 
         self.transformation = transformation
+        self.feature_transformation = self.transformation
 
     @property
     def proto_class(self) -> Type[OnDemandFeatureViewProto]:
@@ -151,6 +161,7 @@ def __copy__(self):
             sources=list(self.source_feature_view_projections.values())
             + list(self.source_request_sources.values()),
             transformation=self.transformation,
+            feature_transformation=self.transformation,
             description=self.description,
             tags=self.tags,
             owner=self.owner,
@@ -172,6 +183,7 @@ def __eq__(self, other):
             != other.source_feature_view_projections
             or self.source_request_sources != other.source_request_sources
             or self.transformation != other.transformation
+            or self.feature_transformation != other.feature_transformation
         ):
             return False
 
@@ -205,16 +217,19 @@ def to_proto(self) -> OnDemandFeatureViewProto:
                 request_data_source=request_sources.to_proto()
             )
 
-        spec = OnDemandFeatureViewSpec(
-            name=self.name,
-            features=[feature.to_proto() for feature in self.features],
-            sources=sources,
+        feature_transformation = FeatureTransformationProto(
             user_defined_function=self.transformation.to_proto()
             if type(self.transformation) == OnDemandPandasTransformation
             else None,
-            on_demand_substrait_transformation=self.transformation.to_proto()  # type: ignore
+            on_demand_substrait_transformation=self.transformation.to_proto()
             if type(self.transformation) == OnDemandSubstraitTransformation
-            else None,
+            else None,  # type: ignore
+        )
+        spec = OnDemandFeatureViewSpec(
+            name=self.name,
+            features=[feature.to_proto() for feature in self.features],
+            sources=sources,
+            feature_transformation=feature_transformation,
             description=self.description,
             tags=self.tags,
             owner=self.owner,
@@ -254,18 +269,37 @@ def from_proto(cls, on_demand_feature_view_proto: OnDemandFeatureViewProto):
                 )
 
         if (
-            on_demand_feature_view_proto.spec.WhichOneof("transformation")
+            on_demand_feature_view_proto.spec.feature_transformation.WhichOneof(
+                "transformation"
+            )
             == "user_defined_function"
+            and on_demand_feature_view_proto.spec.feature_transformation.user_defined_function.body_text
+            != ""
         ):
             transformation = OnDemandPandasTransformation.from_proto(
-                on_demand_feature_view_proto.spec.user_defined_function
+                on_demand_feature_view_proto.spec.feature_transformation.user_defined_function
             )
         elif (
-            on_demand_feature_view_proto.spec.WhichOneof("transformation")
+            on_demand_feature_view_proto.spec.feature_transformation.WhichOneof(
+                "transformation"
+            )
             == "on_demand_substrait_transformation"
         ):
             transformation = OnDemandSubstraitTransformation.from_proto(
-                on_demand_feature_view_proto.spec.on_demand_substrait_transformation
+                on_demand_feature_view_proto.spec.feature_transformation.on_demand_substrait_transformation
+            )
+        elif (
+            hasattr(on_demand_feature_view_proto.spec, "user_defined_function")
+            and on_demand_feature_view_proto.spec.feature_transformation.user_defined_function.body_text
+            == ""
+        ):
+            backwards_compatible_udf = UserDefinedFunctionProto(
+                name=on_demand_feature_view_proto.spec.user_defined_function.name,
+                body=on_demand_feature_view_proto.spec.user_defined_function.body,
+                body_text=on_demand_feature_view_proto.spec.user_defined_function.body_text,
+            )
+            transformation = OnDemandPandasTransformation.from_proto(
+                user_defined_function_proto=backwards_compatible_udf,
             )
         else:
             raise Exception("At least one transformation type needs to be provided")

sdk/python/feast/on_demand_pandas_transformation.py

@@ -3,8 +3,8 @@
 import dill
 import pandas as pd
 
-from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
-    UserDefinedFunction as UserDefinedFunctionProto,
+from feast.protos.feast.core.Transformation_pb2 import (
+    UserDefinedFunctionV2 as UserDefinedFunctionProto,
 )
 
 

sdk/python/feast/on_demand_substrait_transformation.py

@@ -2,8 +2,8 @@
 import pyarrow
 import pyarrow.substrait as substrait  # type: ignore # noqa
 
-from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
-    OnDemandSubstraitTransformation as OnDemandSubstraitTransformationProto,
+from feast.protos.feast.core.Transformation_pb2 import (
+    OnDemandSubstraitTransformationV2 as OnDemandSubstraitTransformationProto,
 )