Specification: In-Memory Registry

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "feast-affirm"
-version = "0.28+affirm147"
+version = "0.28+affirm155"
 description = "Feast - Affirm"
 authors = ["Francisco Arceo", "Ross Briden", "Maks Stachowiak"]
 readme = "README.md"

sdk/python/feast/errors.py

@@ -5,6 +5,7 @@
 from feast.field import Field
 
 
+
 class DataSourceNotFoundException(Exception):
     def __init__(self, path):
         super().__init__(
@@ -118,6 +119,13 @@ def __init__(self, name: str, project: str):
         )
 
 
+class DuplicateValidationReference(Exception):
+    def __init__(self, name: str, project) -> None:
+        super(DuplicateValidationReference, self).__init__(
+            f"Duplication validation reference {name} for project {project}."
+        )
+
+
 class FeastProviderLoginError(Exception):
     """Error class that indicates a user has not authenticated with their provider."""
 
@@ -411,3 +419,35 @@ def __init__(self):
 class PushSourceNotFoundException(Exception):
     def __init__(self, push_source_name: str):
         super().__init__(f"Unable to find push source '{push_source_name}'.")
+
+
+class RegistryNotBuiltException(Exception):
+    def __init__(self, registry_name: str) -> None:
+        super(RegistryNotBuiltException, self).__init__(f"Registry {registry_name} must be built before being queried.")
+
+
+class EntityNameCollisionException(Exception):
+    def __init__(self, entity_name: str, project: str) -> None:
+        super(EntityNameCollisionException, self).__init__(f"Duplicate entity {entity_name} for project {project}.")
+
+
+class FeatureServiceNameCollisionException(Exception):
+    def __init__(self, service_name: str, project: str) -> None:
+        super(FeatureServiceNameCollisionException, self).__init__(
+            f"Duplicate feature service {service_name} for project {project}."
+        )
+
+
+class MissingInfraObjectException(Exception):
+    def __init__(self, project: str) -> None:
+        super(MissingInfraObjectException, self).__init__(f"No infra objects found for project {project}.")
+
+
+class SavedDatasetCollisionException(Exception):
+    def __init__(self, project: str, name: str) -> None:
+        super(SavedDatasetCollisionException, self).__init__(f"Duplicated saved dataset {name} for project {project}")
+
+
+class MissingProjectMetadataException(Exception):
+    def __init__(self, project: str) -> None:
+        super(MissingProjectMetadataException, self).__init__(f"No project metadata for project {project}")

sdk/python/feast/feature_store.py

@@ -81,6 +81,7 @@
 from feast.infra.registry.base_registry import BaseRegistry
 from feast.infra.registry.registry import Registry
 from feast.infra.registry.sql import SqlRegistry
+from feast.infra.registry.memory import MemoryRegistry
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.online_response import OnlineResponse
 from feast.protos.feast.serving.ServingService_pb2 import (
@@ -167,16 +168,24 @@ def __init__(
                 self.repo_path, utils.get_default_yaml_file_path(self.repo_path)
             )
 
+        self._provider = get_provider(self.config)
+
+        # RB: ordering matters here because `apply_total` assumes a constructed `FeatureStore` instance
         registry_config = self.config.get_registry_config()
         if registry_config.registry_type == "sql":
             self._registry = SqlRegistry(registry_config, None, is_feast_apply=is_feast_apply)
+        elif registry_config.registry_type == "memory":
+            self._registry = MemoryRegistry(registry_config, repo_path, is_feast_apply=is_feast_apply)
+
+            # RB: MemoryRegistry is stateless, meaning we'll need to call `apply` with each new FeatureStore instance
+            if not is_feast_apply:
+                from feast.repo_operations import apply_total
+                apply_total(repo_config=self.config, repo_path=self.repo_path, skip_source_validation=False, store=self)
         else:
             r = Registry(registry_config, repo_path=self.repo_path)
             r._initialize_registry(self.config.project)
             self._registry = r
 
-        self._provider = get_provider(self.config)
-
     @log_exceptions
     def version(self) -> str:
         """Returns the version of the current Feast SDK/CLI."""
@@ -212,6 +221,10 @@ def refresh_registry(self):
         downloaded synchronously, which may increase latencies if the triggering method is get_online_features().
         """
         registry_config = self.config.get_registry_config()
+
+        # RB: MemoryRegistry is a cache
+        if registry_config.registry_type == "memory":
+            return
         registry = Registry(registry_config, repo_path=self.repo_path)
         registry.refresh(self.config.project)