Brought back FVProjections without exposing it ever to user-facing API

Signed-off-by: David Y Liu <davidyliuliu@gmail.com>

protos/feast/core/FeatureService.proto

@@ -6,9 +6,7 @@ option java_outer_classname = "FeatureServiceProto";
 option java_package = "feast.proto.core";
 
 import "google/protobuf/timestamp.proto";
-import "feast/core/OnDemandFeatureView.proto";
-import "feast/core/FeatureTable.proto";
-import "feast/core/FeatureView.proto";
+import "feast/core/FeatureViewProjection.proto";
 
 message FeatureService {
   // User-specified specifications of this feature service.
@@ -19,24 +17,15 @@ message FeatureService {
 }
 
 message FeatureServiceSpec {
-  // was previously used with 'repeated FeatureViewProjection features'
-  reserved 3;
-
   // Name of the Feature Service. Must be unique. Not updated.
   string name = 1;
 
   // Name of Feast project that this Feature Service belongs to.
   string project = 2;
 
-  // List of features that this feature service encapsulates.
-  // Stored as a list of references to other features views and the features from those views.
-  repeated string features = 6;
-
-  repeated FeatureTable feature_tables = 7;
-
-  repeated FeatureView feature_views = 8;
-
-  repeated OnDemandFeatureView on_demand_feature_views = 9;
+  // Represents a projection that's to be applied on top of the FeatureView.
+  // Contains data like a name alias or the features to use from a FeatureView.
+  repeated FeatureViewProjection features = 3;
 
   // User defined metadata
   map<string,string> tags = 4;
@@ -53,4 +42,4 @@ message FeatureServiceMeta {
   // Time where this Feature Service is last updated
   google.protobuf.Timestamp last_updated_timestamp = 2;
 
-}
+}

sdk/python/feast/feature_service.py

@@ -5,6 +5,7 @@
 
 from feast.feature_table import FeatureTable
 from feast.feature_view import FeatureView
+from feast.feature_view_projection import FeatureViewProjection
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.protos.feast.core.FeatureService_pb2 import (
     FeatureService as FeatureServiceProto,
@@ -31,10 +32,7 @@ class FeatureService:
     """
 
     name: str
-    features: List[str]
-    feature_tables: List[FeatureTable]
-    feature_views: List[FeatureView]
-    on_demand_feature_views: List[OnDemandFeatureView]
+    feature_view_projections: List[FeatureViewProjection]
     tags: Dict[str, str]
     description: Optional[str] = None
     created_timestamp: Optional[datetime] = None
@@ -55,27 +53,20 @@ def __init__(
             ValueError: If one of the specified features is not a valid type.
         """
         self.name = name
-        self.features = []
-        self.feature_tables, self.feature_views, self.on_demand_feature_views = (
-            [],
-            [],
-            [],
-        )
+        self.feature_view_projections = []
 
         for feature_grouping in features:
-            if isinstance(feature_grouping, FeatureTable):
-                self.feature_tables.append(feature_grouping)
+            if isinstance(feature_grouping, FeatureTable) or isinstance(
+                feature_grouping, OnDemandFeatureView
+            ):
+                self.feature_view_projections.append(
+                    FeatureViewProjection.from_definition(feature_grouping)
+                )
             elif isinstance(feature_grouping, FeatureView):
-                self.feature_views.append(feature_grouping)
-            elif isinstance(feature_grouping, OnDemandFeatureView):
-                self.on_demand_feature_views.append(feature_grouping)
+                self.feature_view_projections.append(feature_grouping.projection)
             else:
                 raise ValueError(f"Unexpected type: {type(feature_grouping)}")
 
-            self.features.extend(
-                [f"{feature_grouping.name}:{f.name}" for f in feature_grouping.features]
-            )
-
         self.tags = tags or {}
         self.description = description
         self.created_timestamp = None
@@ -108,13 +99,15 @@ def __eq__(self, other):
     def from_proto(feature_service_proto: FeatureServiceProto):
         """
         Converts a FeatureServiceProto to a FeatureService object.
-
         Args:
             feature_service_proto: A protobuf representation of a FeatureService.
         """
         fs = FeatureService(
             name=feature_service_proto.spec.name,
-            features=[],
+            features=[
+                FeatureViewProjection.from_proto(projection)
+                for projection in feature_service_proto.spec.features
+            ],
             tags=dict(feature_service_proto.spec.tags),
             description=(
                 feature_service_proto.spec.description
@@ -123,20 +116,6 @@ def from_proto(feature_service_proto: FeatureServiceProto):
             ),
         )
 
-        fs.features = [feature for feature in feature_service_proto.spec.features]
-        fs.feature_tables = [
-            FeatureTable.from_proto(table)
-            for table in feature_service_proto.spec.feature_tables
-        ]
-        fs.feature_views = [
-            FeatureView.from_proto(view)
-            for view in feature_service_proto.spec.feature_views
-        ]
-        fs.on_demand_feature_views = [
-            OnDemandFeatureView.from_proto(view)
-            for view in feature_service_proto.spec.on_demand_feature_views
-        ]
-
         if feature_service_proto.meta.HasField("created_timestamp"):
             fs.created_timestamp = (
                 feature_service_proto.meta.created_timestamp.ToDatetime()
@@ -151,7 +130,6 @@ def from_proto(feature_service_proto: FeatureServiceProto):
     def to_proto(self) -> FeatureServiceProto:
         """
         Converts a FeatureService to its protobuf representation.
-
         Returns:
             A FeatureServiceProto protobuf.
         """
@@ -161,13 +139,11 @@ def to_proto(self) -> FeatureServiceProto:
 
         spec = FeatureServiceSpec(
             name=self.name,
-            features=self.features,
-            feature_tables=[table.to_proto() for table in self.feature_tables],
-            feature_views=[view.to_proto() for view in self.feature_views],
-            on_demand_feature_views=[
-                view.to_proto() for view in self.on_demand_feature_views
+            features=[
+                projection.to_proto() for projection in self.feature_view_projections
             ],
         )
+
         if self.tags:
             spec.tags.update(self.tags)
         if self.description:

sdk/python/feast/feature_store.py

@@ -316,12 +316,18 @@ def _get_features(
         if not _features:
             raise ValueError("No features specified for retrieval")
 
-        _feature_refs: List[str]
+        _feature_refs = []
         if isinstance(_features, FeatureService):
-            _feature_refs = self.get_feature_service(_features.name).features
-            if _features.features != _feature_refs:
-                raise ValueError(
-                    "FeatureService object that's passed in is inconsistent with version from Registry."
+            feature_service_from_registry = self.get_feature_service(_features.name)
+            if feature_service_from_registry != _features:
+                warnings.warn(
+                    "The FeatureService object that has been passed in as an argument is"
+                    "inconsistent with the version from Registry. Potentially a newer version"
+                    "of the FeatureService has been applied to the registry."
+                )
+            for projection in feature_service_from_registry.features:
+                _feature_refs.extend(
+                    [f"{projection.name}:{f.name}" for f in projection.features]
                 )
         else:
             assert isinstance(_features, list)
@@ -1031,14 +1037,18 @@ def _get_feature_views_to_use(
         }
 
         if isinstance(features, FeatureService):
-            for fv in features.feature_views:
-                if fv.name not in all_feature_views:
+            for fv_name, projection in {
+                projection.name: projection
+                for projection in features.feature_view_projections
+            }.items():
+                if fv_name in all_feature_views:
+                    all_feature_views[fv_name].set_projection(projection)
+                else:
                     raise ValueError(
-                        f"{fv.name} used in the FeatureService is not in the registry."
+                        f"The provided feature service {features.name} contains a reference to a feature view"
+                        f"{fv_name} which doesn't exist. Please make sure that you have created the feature view"
+                        f'{fv_name} and that you have registered it by running "apply".'
                     )
-                all_feature_views[fv.name] = all_feature_views[fv.name].get_projection(
-                    fv
-                )
 
         return [*all_feature_views.values()]
 

sdk/python/feast/feature_view.py

@@ -23,6 +23,7 @@
 from feast.data_source import DataSource
 from feast.errors import RegistryInferenceFailure
 from feast.feature import Feature
+from feast.feature_view_projection import FeatureViewProjection
 from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
 from feast.protos.feast.core.FeatureView_pb2 import (
     FeatureViewMeta as FeatureViewMetaProto,
@@ -73,6 +74,7 @@ class FeatureView:
     online: bool
     input: DataSource
     batch_source: DataSource
+    projection: FeatureViewProjection
     stream_source: Optional[DataSource] = None
     created_timestamp: Optional[datetime] = None
     last_updated_timestamp: Optional[datetime] = None
@@ -140,6 +142,8 @@ def __init__(
         self.created_timestamp: Optional[datetime] = None
         self.last_updated_timestamp: Optional[datetime] = None
 
+        self.projection = FeatureViewProjection.from_definition(self)
+
     def __repr__(self):
         items = (f"{k} = {v}" for k, v in self.__dict__.items())
         return f"<{self.__class__.__name__}({', '.join(items)})>"
@@ -158,17 +162,9 @@ def __getitem__(self, item):
             if feature.name in item:
                 referenced_features.append(feature)
 
-        return FeatureView(
-            name=self.name,
-            entities=self.entities,
-            ttl=self.ttl,
-            input=self.input,
-            batch_source=self.batch_source,
-            stream_source=self.stream_source,
-            features=referenced_features,
-            tags=self.tags,
-            online=self.online,
-        )
+        self.projection.features = referenced_features
+
+        return self
 
     def __eq__(self, other):
         if not isinstance(other, FeatureView):
@@ -292,6 +288,10 @@ def from_proto(cls, feature_view_proto: FeatureViewProto):
             stream_source=stream_source,
         )
 
+        # FeatureViewProjections are not saved in the FeatureView proto.
+        # Create the default projection.
+        feature_view.projection = FeatureViewProjection.from_definition(feature_view)
+
         if feature_view_proto.meta.HasField("created_timestamp"):
             feature_view.created_timestamp = (
                 feature_view_proto.meta.created_timestamp.ToDatetime()
@@ -389,47 +389,14 @@ def infer_features_from_batch_source(self, config: RepoConfig):
                     f"Could not infer Features for the FeatureView named {self.name}.",
                 )
 
-    def get_projection(self, feature_view):
-        """
-        Produces a copy of this FeatureView (self) with specific fields modified according to the FeatureView object
-        that's passed in as the argument. This allows users to make modifications to a FeatureView object
-        (e.g. the name) and then projecting those changes onto the corresponding actual FeatureView from the registry.
-        Currently all FeatureViews that are used must be registered and this method enables modifying those FeatureViews
-        while still making sure we're pulling the most up-to-date FeatureView from the registry to modify from.
-
-        Currently, only `FeatureView.features` is the field that's replaced the features from feature_view.
-
-        Args:
-            feature_view: The FeatureView object that's likely not registered and has the modified fields that should
-            be projected onto a copy of this FeatureView (self).
+    def set_projection(self, feature_view_projection: FeatureViewProjection):
+        assert feature_view_projection.name == self.name
 
-        Returns:
-            A copy of this FeatureView (self) with some of its fields modified according to the feature_view argument.
-
-        """
-        if not isinstance(feature_view, FeatureView):
-            raise TypeError(
-                "A projection can only be created from a passed in FeatureView."
-            )
-
-        features_to_use = []
-        features_dict = {feature.name: feature for feature in self.features}
-        for feature in feature_view.features:
+        for feature in feature_view_projection.features:
             if feature not in self.features:
                 raise ValueError(
-                    "There are features in the passed in FeatureView object that are not in the FeatureView object"
-                    "of the same name from the registry."
+                    f"The projection for {self.name} cannot be applied because it contains {feature.name} which the"
+                    "FeatureView doesn't have."
                 )
-            features_to_use.append(features_dict[feature.name])
 
-        return FeatureView(
-            name=self.name,
-            entities=self.entities,
-            ttl=self.ttl,
-            input=self.input,
-            batch_source=self.batch_source,
-            stream_source=self.stream_source,
-            features=features_to_use,
-            tags=self.tags,
-            online=self.online,
-        )
+        self.projection = feature_view_projection