fix: Update Feast object metadata in the registry (feast-dev#4257)

Author: msistla96

sdk/python/feast/feature_view.py

@@ -311,6 +311,16 @@ def with_join_key_map(self, join_key_map: Dict[str, str]):
 
         return cp
 
+    def update_materialization_intervals(
+        self, existing_materialization_intervals: List[Tuple[datetime, datetime]]
+    ):
+        if (
+            len(existing_materialization_intervals) > 0
+            and len(self.materialization_intervals) == 0
+        ):
+            for interval in existing_materialization_intervals:
+                self.materialization_intervals.append((interval[0], interval[1]))
+
     def to_proto(self) -> FeatureViewProto:
         """
         Converts a feature view object to its protobuf representation.

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

@@ -29,7 +29,19 @@
 from feast.infra.infra_object import Infra
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.project_metadata import ProjectMetadata
+from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto
+from feast.protos.feast.core.FeatureService_pb2 import (
+    FeatureService as FeatureServiceProto,
+)
+from feast.protos.feast.core.FeatureView_pb2 import FeatureView as FeatureViewProto
+from feast.protos.feast.core.OnDemandFeatureView_pb2 import (
+    OnDemandFeatureView as OnDemandFeatureViewProto,
+)
 from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
+from feast.protos.feast.core.SavedDataset_pb2 import SavedDataset as SavedDatasetProto
+from feast.protos.feast.core.StreamFeatureView_pb2 import (
+    StreamFeatureView as StreamFeatureViewProto,
+)
 from feast.saved_dataset import SavedDataset, ValidationReference
 from feast.stream_feature_view import StreamFeatureView
 from feast.transformation.pandas_transformation import PandasTransformation
@@ -705,3 +717,19 @@ def to_dict(self, project: str) -> Dict[str, List[Any]]:
                 self._message_to_sorted_dict(infra_object.to_proto())
             )
         return registry_dict
+
+    @staticmethod
+    def deserialize_registry_values(serialized_proto, feast_obj_type) -> Any:
+        if feast_obj_type == Entity:
+            return EntityProto.FromString(serialized_proto)
+        if feast_obj_type == SavedDataset:
+            return SavedDatasetProto.FromString(serialized_proto)
+        if feast_obj_type == FeatureView:
+            return FeatureViewProto.FromString(serialized_proto)
+        if feast_obj_type == StreamFeatureView:
+            return StreamFeatureViewProto.FromString(serialized_proto)
+        if feast_obj_type == OnDemandFeatureView:
+            return OnDemandFeatureViewProto.FromString(serialized_proto)
+        if feast_obj_type == FeatureService:
+            return FeatureServiceProto.FromString(serialized_proto)
+        return None

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

@@ -265,9 +265,13 @@ def apply_entity(self, entity: Entity, project: str, commit: bool = True):
                 existing_entity_proto.spec.name == entity_proto.spec.name
                 and existing_entity_proto.spec.project == project
             ):
+                entity.created_timestamp = (
+                    existing_entity_proto.meta.created_timestamp.ToDatetime()
+                )
+                entity_proto = entity.to_proto()
+                entity_proto.spec.project = project
                 del self.cached_registry_proto.entities[idx]
                 break
-
         self.cached_registry_proto.entities.append(entity_proto)
         if commit:
             self.commit()
@@ -346,6 +350,11 @@ def apply_feature_service(
                 == feature_service_proto.spec.name
                 and existing_feature_service_proto.spec.project == project
             ):
+                feature_service.created_timestamp = (
+                    existing_feature_service_proto.meta.created_timestamp.ToDatetime()
+                )
+                feature_service_proto = feature_service.to_proto()
+                feature_service_proto.spec.project = project
                 del registry.feature_services[idx]
         registry.feature_services.append(feature_service_proto)
         if commit:
@@ -421,6 +430,18 @@ def apply_feature_view(
                 ):
                     return
                 else:
+                    existing_feature_view = type(feature_view).from_proto(
+                        existing_feature_view_proto
+                    )
+                    feature_view.created_timestamp = (
+                        existing_feature_view.created_timestamp
+                    )
+                    if isinstance(feature_view, (FeatureView, StreamFeatureView)):
+                        feature_view.update_materialization_intervals(
+                            existing_feature_view.materialization_intervals
+                        )
+                    feature_view_proto = feature_view.to_proto()
+                    feature_view_proto.spec.project = project
                     del existing_feature_views_of_same_type[idx]
                     break
 
@@ -660,6 +681,17 @@ def apply_saved_dataset(
                 existing_saved_dataset_proto.spec.name == saved_dataset_proto.spec.name
                 and existing_saved_dataset_proto.spec.project == project
             ):
+                saved_dataset.created_timestamp = (
+                    existing_saved_dataset_proto.meta.created_timestamp.ToDatetime()
+                )
+                saved_dataset.min_event_timestamp = (
+                    existing_saved_dataset_proto.meta.min_event_timestamp.ToDatetime()
+                )
+                saved_dataset.max_event_timestamp = (
+                    existing_saved_dataset_proto.meta.max_event_timestamp.ToDatetime()
+                )
+                saved_dataset_proto = saved_dataset.to_proto()
+                saved_dataset_proto.spec.project = project
                 del self.cached_registry_proto.saved_datasets[idx]
                 break
 

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

@@ -296,6 +296,7 @@ def apply_materialization(
         start_date_timestamp.FromDatetime(start_date)
         end_date_timestamp.FromDatetime(end_date)
 
+        # TODO: for this to work for stream feature views, ApplyMaterializationRequest needs to be updated
         request = RegistryServer_pb2.ApplyMaterializationRequest(
             feature_view=feature_view.to_proto(),
             project=project,

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

@@ -713,6 +713,24 @@ def _apply_object(
                 obj.last_updated_timestamp = update_datetime
 
             if row:
+                if proto_field_name in [
+                    "entity_proto",
+                    "saved_dataset_proto",
+                    "feature_view_proto",
+                    "feature_service_proto",
+                ]:
+                    deserialized_proto = self.deserialize_registry_values(
+                        row._mapping[proto_field_name], type(obj)
+                    )
+                    obj.created_timestamp = (
+                        deserialized_proto.meta.created_timestamp.ToDatetime()
+                    )
+                    if isinstance(obj, (FeatureView, StreamFeatureView)):
+                        obj.update_materialization_intervals(
+                            type(obj)
+                            .from_proto(deserialized_proto)
+                            .materialization_intervals
+                        )
                 values = {
                     proto_field_name: obj.to_proto().SerializeToString(),
                     "last_updated_timestamp": update_time,

sdk/python/feast/registry_server.py

@@ -3,6 +3,7 @@
 
 import grpc
 from google.protobuf.empty_pb2 import Empty
+from pytz import utc
 
 from feast import FeatureStore
 from feast.data_source import DataSource
@@ -313,10 +314,10 @@ def ApplyMaterialization(
             feature_view=FeatureView.from_proto(request.feature_view),
             project=request.project,
             start_date=datetime.fromtimestamp(
-                request.start_date.seconds + request.start_date.nanos / 1e9
+                request.start_date.seconds + request.start_date.nanos / 1e9, tz=utc
             ),
             end_date=datetime.fromtimestamp(
-                request.end_date.seconds + request.end_date.nanos / 1e9
+                request.end_date.seconds + request.end_date.nanos / 1e9, tz=utc
             ),
             commit=request.commit,
         )