Implement feature_store._apply_diffs to handle registry and infra diffs

sdk/python/feast/diff/FcoDiff.py

@@ -5,6 +5,7 @@
 from feast.diff.property_diff import PropertyDiff, TransitionType
 from feast.entity import Entity
 from feast.feature_service import FeatureService
+from feast.feature_view import DUMMY_ENTITY_NAME
 from feast.protos.feast.core.Entity_pb2 import Entity as EntityProto
 from feast.protos.feast.core.FeatureService_pb2 import (
     FeatureService as FeatureServiceProto,
@@ -16,26 +17,16 @@
 from feast.protos.feast.core.RequestFeatureView_pb2 import (
     RequestFeatureView as RequestFeatureViewProto,
 )
-from feast.registry import FeastObjectType, Registry
+from feast.registry import FEAST_OBJECT_TYPES, FeastObjectType, Registry
 from feast.repo_contents import RepoContents
 
-FEAST_OBJECT_TYPE_TO_STR = {
-    FeastObjectType.ENTITY: "entity",
-    FeastObjectType.FEATURE_VIEW: "feature view",
-    FeastObjectType.ON_DEMAND_FEATURE_VIEW: "on demand feature view",
-    FeastObjectType.REQUEST_FEATURE_VIEW: "request feature view",
-    FeastObjectType.FEATURE_SERVICE: "feature service",
-}
-
-FEAST_OBJECT_TYPES = FEAST_OBJECT_TYPE_TO_STR.keys()
-
 Fco = TypeVar("Fco", Entity, BaseFeatureView, FeatureService)
 
 
 @dataclass
 class FcoDiff(Generic[Fco]):
     name: str
-    fco_type: str
+    fco_type: FeastObjectType
     current_fco: Fco
     new_fco: Fco
     fco_property_diffs: List[PropertyDiff]
@@ -52,6 +43,28 @@ def __init__(self):
     def add_fco_diff(self, fco_diff: FcoDiff):
         self.fco_diffs.append(fco_diff)
 
+    def to_string(self):
+        from colorama import Fore, Style
+
+        log_string = ""
+
+        message_action_map = {
+            TransitionType.CREATE: ("Created", Fore.GREEN),
+            TransitionType.DELETE: ("Deleted", Fore.RED),
+            TransitionType.UNCHANGED: ("Unchanged", Fore.LIGHTBLUE_EX),
+            TransitionType.UPDATE: ("Updated", Fore.YELLOW),
+        }
+        for fco_diff in self.fco_diffs:
+            if fco_diff.name == DUMMY_ENTITY_NAME:
+                continue
+            action, color = message_action_map[fco_diff.transition_type]
+            log_string += f"{action} {fco_diff.fco_type.value} {Style.BRIGHT + color}{fco_diff.name}{Style.RESET_ALL}\n"
+            if fco_diff.transition_type == TransitionType.UPDATE:
+                for _p in fco_diff.fco_property_diffs:
+                    log_string += f"\t{_p.property_name}: {Style.BRIGHT + color}{_p.val_existing}{Style.RESET_ALL} -> {Style.BRIGHT + Fore.LIGHTGREEN_EX}{_p.val_declared}{Style.RESET_ALL}\n"
+
+        return log_string
+
 
 def tag_objects_for_keep_delete_update_add(
     existing_objs: Iterable[Fco], desired_objs: Iterable[Fco]
@@ -93,7 +106,9 @@ def tag_proto_objects_for_keep_delete_add(
 FIELDS_TO_IGNORE = {"project"}
 
 
-def diff_registry_objects(current: Fco, new: Fco, object_type: str) -> FcoDiff:
+def diff_registry_objects(
+    current: Fco, new: Fco, object_type: FeastObjectType
+) -> FcoDiff:
     current_proto = current.to_proto()
     new_proto = new.to_proto()
     assert current_proto.DESCRIPTOR.full_name == new_proto.DESCRIPTOR.full_name
@@ -145,30 +160,12 @@ def extract_objects_for_keep_delete_update_add(
     objs_to_update = {}
     objs_to_add = {}
 
-    registry_object_type_to_objects: Dict[FeastObjectType, List[Any]]
-    registry_object_type_to_objects = {
-        FeastObjectType.ENTITY: registry.list_entities(project=current_project),
-        FeastObjectType.FEATURE_VIEW: registry.list_feature_views(
-            project=current_project
-        ),
-        FeastObjectType.ON_DEMAND_FEATURE_VIEW: registry.list_on_demand_feature_views(
-            project=current_project
-        ),
-        FeastObjectType.REQUEST_FEATURE_VIEW: registry.list_request_feature_views(
-            project=current_project
-        ),
-        FeastObjectType.FEATURE_SERVICE: registry.list_feature_services(
-            project=current_project
-        ),
-    }
-    registry_object_type_to_repo_contents: Dict[FeastObjectType, Set[Any]]
-    registry_object_type_to_repo_contents = {
-        FeastObjectType.ENTITY: desired_repo_contents.entities,
-        FeastObjectType.FEATURE_VIEW: desired_repo_contents.feature_views,
-        FeastObjectType.ON_DEMAND_FEATURE_VIEW: desired_repo_contents.on_demand_feature_views,
-        FeastObjectType.REQUEST_FEATURE_VIEW: desired_repo_contents.request_feature_views,
-        FeastObjectType.FEATURE_SERVICE: desired_repo_contents.feature_services,
-    }
+    registry_object_type_to_objects: Dict[
+        FeastObjectType, List[Any]
+    ] = FeastObjectType.get_objects_from_registry(registry, current_project)
+    registry_object_type_to_repo_contents: Dict[
+        FeastObjectType, Set[Any]
+    ] = FeastObjectType.get_objects_from_repo_contents(desired_repo_contents)
 
     for object_type in FEAST_OBJECT_TYPES:
         (
@@ -221,7 +218,7 @@ def diff_between(
             diff.add_fco_diff(
                 FcoDiff(
                     name=e.name,
-                    fco_type=FEAST_OBJECT_TYPE_TO_STR[object_type],
+                    fco_type=object_type,
                     current_fco=None,
                     new_fco=e,
                     fco_property_diffs=[],
@@ -232,7 +229,7 @@ def diff_between(
             diff.add_fco_diff(
                 FcoDiff(
                     name=e.name,
-                    fco_type=FEAST_OBJECT_TYPE_TO_STR[object_type],
+                    fco_type=object_type,
                     current_fco=e,
                     new_fco=None,
                     fco_property_diffs=[],
@@ -241,10 +238,56 @@ def diff_between(
             )
         for e in objects_to_update:
             current_obj = [_e for _e in objects_to_keep if _e.name == e.name][0]
-            diff.add_fco_diff(
-                diff_registry_objects(
-                    current_obj, e, FEAST_OBJECT_TYPE_TO_STR[object_type]
-                )
-            )
+            diff.add_fco_diff(diff_registry_objects(current_obj, e, object_type))
 
     return diff
+
+
+def apply_diff_to_registry(
+    registry: Registry, registry_diff: RegistryDiff, project: str, commit: bool = True
+):
+    """
+    Applies the given diff to the given Feast project in the registry.
+
+    Args:
+        registry: The registry to be updated.
+        registry_diff: The diff to apply.
+        project: Feast project to be updated.
+        commit: Whether the change should be persisted immediately
+    """
+    for fco_diff in registry_diff.fco_diffs:
+        # There is no need to delete the FCO on an update, since applying the new FCO
+        # will automatically delete the existing FCO.
+        if fco_diff.transition_type == TransitionType.DELETE:
+            if fco_diff.fco_type == FeastObjectType.ENTITY:
+                registry.delete_entity(fco_diff.current_fco.name, project, commit=False)
+            elif fco_diff.fco_type == FeastObjectType.FEATURE_SERVICE:
+                registry.delete_feature_service(
+                    fco_diff.current_fco.name, project, commit=False
+                )
+            elif fco_diff.fco_type in [
+                FeastObjectType.FEATURE_VIEW,
+                FeastObjectType.ON_DEMAND_FEATURE_VIEW,
+                FeastObjectType.REQUEST_FEATURE_VIEW,
+            ]:
+                registry.delete_feature_view(
+                    fco_diff.current_fco.name, project, commit=False,
+                )
+
+        if fco_diff.transition_type in [
+            TransitionType.CREATE,
+            TransitionType.UPDATE,
+        ]:
+            if fco_diff.fco_type == FeastObjectType.ENTITY:
+                registry.apply_entity(fco_diff.new_fco, project, commit=False)
+            elif fco_diff.fco_type == FeastObjectType.FEATURE_SERVICE:
+                registry.apply_feature_service(fco_diff.new_fco, project, commit=False)
+            elif fco_diff.fco_type in [
+                FeastObjectType.FEATURE_VIEW,
+                FeastObjectType.ON_DEMAND_FEATURE_VIEW,
+                FeastObjectType.REQUEST_FEATURE_VIEW,
+            ]:
+                registry.apply_feature_view(fco_diff.new_fco, project, commit=False)
+
+    if commit:
+        registry.commit()

sdk/python/feast/diff/infra_diff.py

@@ -60,7 +60,24 @@ def update(self):
                 infra_object.update()
 
     def to_string(self):
-        pass
+        from colorama import Fore, Style
+
+        log_string = ""
+
+        message_action_map = {
+            TransitionType.CREATE: ("Created", Fore.GREEN),
+            TransitionType.DELETE: ("Deleted", Fore.RED),
+            TransitionType.UNCHANGED: ("Unchanged", Fore.LIGHTBLUE_EX),
+            TransitionType.UPDATE: ("Updated", Fore.YELLOW),
+        }
+        for infra_object_diff in self.infra_object_diffs:
+            action, color = message_action_map[infra_object_diff.transition_type]
+            log_string += f"{action} {infra_object_diff.infra_object_type} {Style.BRIGHT + color}{infra_object_diff.name}{Style.RESET_ALL}\n"
+            if infra_object_diff.transition_type == TransitionType.UPDATE:
+                for _p in infra_object_diff.infra_object_property_diffs:
+                    log_string += f"\t{_p.property_name}: {Style.BRIGHT + color}{_p.val_existing}{Style.RESET_ALL} -> {Style.BRIGHT + Fore.LIGHTGREEN_EX}{_p.val_declared}{Style.RESET_ALL}\n"
+
+        return log_string
 
 
 def tag_infra_proto_objects_for_keep_delete_add(