chore: Refactor StreamFeatureViewMeta to FeatureViewMeta and dedupe (#2915)

* chore: Refactor StreamFeatureViewMeta to FeatureViewMeta and deduplicate code

Signed-off-by: Achal Shah <[email protected]>

* ttl_duration

Signed-off-by: Achal Shah <[email protected]>

* nits

Signed-off-by: Achal Shah <[email protected]>

Author: achals

protos/feast/core/StreamFeatureView.proto

@@ -34,7 +34,7 @@ import "feast/core/Aggregation.proto";
 message StreamFeatureView {
     // User-specified specifications of this feature view.
     StreamFeatureViewSpec spec = 1;
-    StreamFeatureViewMeta meta = 2;
+    FeatureViewMeta meta = 2;
 }
 
 // Next available id: 17
@@ -90,13 +90,3 @@ message StreamFeatureViewSpec {
     string timestamp_field = 16;
 }
 
-message StreamFeatureViewMeta {
-    // Time where this Feature View is created
-    google.protobuf.Timestamp created_timestamp = 1;
-
-    // Time where this Feature View is last updated
-    google.protobuf.Timestamp last_updated_timestamp = 2;
-
-    // List of pairs (start_time, end_time) for which this feature view has been materialized.
-    repeated MaterializationInterval materialization_intervals = 3;
-}

sdk/python/feast/feature_view.py

@@ -407,21 +407,8 @@ def to_proto(self) -> FeatureViewProto:
         Returns:
             A FeatureViewProto protobuf.
         """
-        meta = FeatureViewMetaProto(materialization_intervals=[])
-        if self.created_timestamp:
-            meta.created_timestamp.FromDatetime(self.created_timestamp)
-        if self.last_updated_timestamp:
-            meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp)
-        for interval in self.materialization_intervals:
-            interval_proto = MaterializationIntervalProto()
-            interval_proto.start_time.FromDatetime(interval[0])
-            interval_proto.end_time.FromDatetime(interval[1])
-            meta.materialization_intervals.append(interval_proto)
-
-        ttl_duration = None
-        if self.ttl is not None:
-            ttl_duration = Duration()
-            ttl_duration.FromTimedelta(self.ttl)
+        meta = self.to_proto_meta()
+        ttl_duration = self.get_ttl_duration()
 
         batch_source_proto = self.batch_source.to_proto()
         batch_source_proto.data_source_class_type = f"{self.batch_source.__class__.__module__}.{self.batch_source.__class__.__name__}"
@@ -447,6 +434,26 @@ def to_proto(self) -> FeatureViewProto:
 
         return FeatureViewProto(spec=spec, meta=meta)
 
+    def to_proto_meta(self):
+        meta = FeatureViewMetaProto(materialization_intervals=[])
+        if self.created_timestamp:
+            meta.created_timestamp.FromDatetime(self.created_timestamp)
+        if self.last_updated_timestamp:
+            meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp)
+        for interval in self.materialization_intervals:
+            interval_proto = MaterializationIntervalProto()
+            interval_proto.start_time.FromDatetime(interval[0])
+            interval_proto.end_time.FromDatetime(interval[1])
+            meta.materialization_intervals.append(interval_proto)
+        return meta
+
+    def get_ttl_duration(self):
+        ttl_duration = None
+        if self.ttl is not None:
+            ttl_duration = Duration()
+            ttl_duration.FromTimedelta(self.ttl)
+        return ttl_duration
+
     @classmethod
     def from_proto(cls, feature_view_proto: FeatureViewProto):
         """

sdk/python/feast/stream_feature_view.py

@@ -6,7 +6,6 @@
 from typing import Dict, List, Optional, Tuple, Union
 
 import dill
-from google.protobuf.duration_pb2 import Duration
 from typeguard import typechecked
 
 from feast import utils
@@ -16,18 +15,12 @@
 from feast.feature_view import FeatureView
 from feast.field import Field
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
-from feast.protos.feast.core.FeatureView_pb2 import (
-    MaterializationInterval as MaterializationIntervalProto,
-)
 from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
     UserDefinedFunction as UserDefinedFunctionProto,
 )
 from feast.protos.feast.core.StreamFeatureView_pb2 import (
     StreamFeatureView as StreamFeatureViewProto,
 )
-from feast.protos.feast.core.StreamFeatureView_pb2 import (
-    StreamFeatureViewMeta as StreamFeatureViewMetaProto,
-)
 from feast.protos.feast.core.StreamFeatureView_pb2 import (
     StreamFeatureViewSpec as StreamFeatureViewSpecProto,
 )
@@ -170,23 +163,8 @@ def __hash__(self) -> int:
         return super().__hash__()
 
     def to_proto(self):
-        meta = StreamFeatureViewMetaProto(materialization_intervals=[])
-        if self.created_timestamp:
-            meta.created_timestamp.FromDatetime(self.created_timestamp)
-
-        if self.last_updated_timestamp:
-            meta.last_updated_timestamp.FromDatetime(self.last_updated_timestamp)
-
-        for interval in self.materialization_intervals:
-            interval_proto = MaterializationIntervalProto()
-            interval_proto.start_time.FromDatetime(interval[0])
-            interval_proto.end_time.FromDatetime(interval[1])
-            meta.materialization_intervals.append(interval_proto)
-
-        ttl_duration = None
-        if self.ttl is not None:
-            ttl_duration = Duration()
-            ttl_duration.FromTimedelta(self.ttl)
+        meta = self.to_proto_meta()
+        ttl_duration = self.get_ttl_duration()
 
         batch_source_proto = None
         if self.batch_source: