almost got everything working and type validation behaving

Signed-off-by: Francisco Javier Arceo <farceo@redhat.com>
haoxu0 · Sep 23, 2024 · Sep 17, 2024 · Sep 18, 2024 · Sep 18, 2024 · Sep 18, 2024
commit 44066d2afff7c4c52ebd7a9a5fac6636773809fb
@@ -18,6 +18,7 @@
 from google.protobuf.json_format import MessageToJson
 from google.protobuf.message import Message
 
+from feast.data_source import DataSource
 from feast.feature_view_projection import FeatureViewProjection
 from feast.field import Field
 from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
@@ -65,6 +66,7 @@ def __init__(
         description: str = "",
         tags: Optional[Dict[str, str]] = None,
         owner: str = "",
+        source: Optional[DataSource] = None,
     ):
         """
         Creates a BaseFeatureView object.
@@ -76,7 +78,8 @@ def __init__(
             tags (optional): A dictionary of key-value pairs to store arbitrary metadata.
             owner (optional): The owner of the base feature view, typically the email of the
                 primary maintainer.
-
+            source (optional): The source of data for this group of features. May be a stream source, or a batch source.
+                If a stream source, the source should contain a batch_source for backfills & batch materialization.
         Raises:
             ValueError: A field mapping conflicts with an Entity or a Feature.
         """
@@ -90,6 +93,9 @@ def __init__(
         self.created_timestamp = None
         self.last_updated_timestamp = None
 
+        if source:
+            self.source = source
+
     @property
     @abstractmethod
     def proto_class(self) -> Type[Message]:
@@ -155,6 +161,7 @@ def __eq__(self, other):
             or self.description != other.description
             or self.tags != other.tags
             or self.owner != other.owner
+            or self.source != other.source
         ):
             return False
 

@@ -206,6 +206,7 @@ def __init__(
             description=description,
             tags=tags,
             owner=owner,
+            source=source,
         )
         self.online = online
         self.materialization_intervals = []
@@ -429,7 +430,9 @@ def from_proto(cls, feature_view_proto: FeatureViewProto):
 
         # FeatureViewProjections are not saved in the FeatureView proto.
         # Create the default projection.
-        feature_view.projection = FeatureViewProjection.from_definition(feature_view)
+        feature_view.projection = FeatureViewProjection.from_feature_view_definition(
+            feature_view
+        )
 
         if feature_view_proto.meta.HasField("created_timestamp"):
             feature_view.created_timestamp = (

@@ -10,6 +10,7 @@
 
 if TYPE_CHECKING:
     from feast.base_feature_view import BaseFeatureView
+    from feast.feature_view import FeatureView
 
 
 @dataclass
@@ -53,17 +54,17 @@ def name_to_use(self):
     def to_proto(self) -> FeatureViewProjectionProto:
         batch_source = None
         if getattr(self, "batch_source", None):
-            if type(self.batch_source).__name__ == "DataSource":
-                batch_source = self.batch_source
-            else:
+            if isinstance(self.batch_source, DataSource):
                 batch_source = self.batch_source.to_proto()
+            else:
+                batch_source = self.batch_source
         feature_reference_proto = FeatureViewProjectionProto(
             feature_view_name=self.name,
             feature_view_name_alias=self.name_alias or "",
             join_key_map=self.join_key_map,
-            timestamp_field=self.timestamp_field,
-            date_partition_column=self.date_partition_column,
-            created_timestamp_column=self.created_timestamp_column,
+            timestamp_field=self.timestamp_field or "",
+            date_partition_column=self.date_partition_column or "",
+            created_timestamp_column=self.created_timestamp_column or "",
             batch_source=batch_source,
         )
         for feature in self.features:
@@ -90,8 +91,25 @@ def from_proto(proto: FeatureViewProjectionProto):
         return feature_view_projection
 
     @staticmethod
-    def from_definition(base_feature_view: "BaseFeatureView"):
+    def from_feature_view_definition(feature_view: "FeatureView"):
         # TODO need to implement this for StreamFeatureViews
+        if getattr(feature_view, "batch_source", None):
+            return FeatureViewProjection(
+                name=feature_view.name,
+                name_alias=None,
+                features=feature_view.features,
+                desired_features=[],
+                timestamp_field=feature_view.batch_source.created_timestamp_column
+                or None,
+                created_timestamp_column=feature_view.batch_source.created_timestamp_column
+                or None,
+                date_partition_column=feature_view.batch_source.date_partition_column
+                or None,
+                batch_source=feature_view.batch_source or None,
+            )
+
+    @staticmethod
+    def from_definition(base_feature_view: "BaseFeatureView"):
         if getattr(base_feature_view, "batch_source", None):
             return FeatureViewProjection(
                 name=base_feature_view.name,
@@ -106,12 +124,13 @@ def from_definition(base_feature_view: "BaseFeatureView"):
                 or None,
                 batch_source=base_feature_view.batch_source or None,
             )
-        return FeatureViewProjection(
-            name=base_feature_view.name,
-            name_alias=None,
-            features=base_feature_view.features,
-            desired_features=[],
-        )
+        else:
+            return FeatureViewProjection(
+                name=base_feature_view.name,
+                name_alias=None,
+                features=base_feature_view.features,
+                desired_features=[],
+            )
 
     def get_feature(self, feature_name: str) -> Field:
         try:

@@ -203,7 +203,7 @@ def _infer_features_and_entities(
         run_inference_for_features: Whether to run inference for features.
         config: The config for the current feature store.
     """
-    entity_columns = []
+    entity_columns: list[str] = []
     if isinstance(fv, OnDemandFeatureView):
         columns_to_exclude = set()
         for (

@@ -235,9 +235,9 @@ def __eq__(self, other):
                 "Comparisons should only involve OnDemandFeatureView class objects."
             )
 
-        if not super().__eq__(other):
-            return False
-
+        # Note, no longer evaluating the base feature view layer as ODFVs can have
+        # multiple datasources and a base_feature_view only has one source
+        # though maybe that shouldn't be true
         if (
             self.source_feature_view_projections
             != other.source_feature_view_projections

@@ -344,7 +344,9 @@ def _group_feature_refs(
 
     # on demand view to on demand view proto
     on_demand_view_index = {
-        view.projection.name_to_use(): view for view in all_on_demand_feature_views
+        view.projection.name_to_use(): view
+        for view in all_on_demand_feature_views
+        if view.projection
     }
 
     # view name to feature names

@@ -209,8 +209,9 @@ def test_apply_feature_view_with_inline_batch_source(
         test_feature_store.apply([entity, driver_fv])
 
         fvs = test_feature_store.list_batch_feature_views()
+        dfv = fvs[0]
         assert len(fvs) == 1
-        assert fvs[0] == driver_fv
+        assert dfv == driver_fv
 
         ds = test_feature_store.list_data_sources()
         assert len(ds) == 1