feat: Add Snowflake Registry (#3363)

* feat: Add Snowflake Registry

Signed-off-by: Miles Adkins <miles.adkins@snowflake.com>

* Fix custom registry store parsing

Signed-off-by: Danny Chiao <danny@tecton.ai>

* fix test

Signed-off-by: Danny Chiao <danny@tecton.ai>

---------

Signed-off-by: Miles Adkins <miles.adkins@snowflake.com>
Signed-off-by: Danny Chiao <danny@tecton.ai>
Co-authored-by: Danny Chiao <danny@tecton.ai>

docs/tutorials/azure/notebooks/src/score.py

@@ -6,9 +6,11 @@
 import json
 import joblib
 from feast import FeatureStore, RepoConfig
-from feast.infra.registry.registry import RegistryConfig
+from feast.repo_config import RegistryConfig
 
-from feast.infra.offline_stores.contrib.mssql_offline_store.mssql import MsSqlServerOfflineStoreConfig
+from feast.infra.offline_stores.contrib.mssql_offline_store.mssql import (
+    MsSqlServerOfflineStoreConfig,
+)
 from feast.infra.online_stores.redis import RedisOnlineStoreConfig, RedisOnlineStore
 
 
@@ -73,4 +75,4 @@ def run(raw_data):
         y_hat = model.predict(data)
         return y_hat.tolist()
     else:
-        return 0.0
+        return 0.0

sdk/python/feast/errors.py

@@ -132,6 +132,11 @@ def __init__(self):
         super().__init__("Provider is not set, but is required")
 
 
+class FeastRegistryNotSetError(Exception):
+    def __init__(self):
+        super().__init__("Registry is not set, but is required")
+
+
 class FeastFeatureServerTypeSetError(Exception):
     def __init__(self, feature_server_type: str):
         super().__init__(
@@ -146,6 +151,13 @@ def __init__(self, feature_server_type: str):
         )
 
 
+class FeastRegistryTypeInvalidError(Exception):
+    def __init__(self, registry_type: str):
+        super().__init__(
+            f"Feature server type was set to {registry_type}, but this type is invalid"
+        )
+
+
 class FeastModuleImportError(Exception):
     def __init__(self, module_name: str, class_name: str):
         super().__init__(

sdk/python/feast/feature_store.py

@@ -164,9 +164,15 @@ def __init__(
                 self.repo_path, utils.get_default_yaml_file_path(self.repo_path)
             )
 
-        registry_config = self.config.get_registry_config()
+        registry_config = self.config.registry
         if registry_config.registry_type == "sql":
             self._registry = SqlRegistry(registry_config, self.config.project, None)
+        elif registry_config.registry_type == "snowflake.registry":
+            from feast.infra.registry.snowflake import SnowflakeRegistry
+
+            self._registry = SnowflakeRegistry(
+                registry_config, self.config.project, None
+            )
         else:
             r = Registry(self.config.project, registry_config, repo_path=self.repo_path)
             r._initialize_registry(self.config.project)
@@ -209,7 +215,7 @@ def refresh_registry(self):
         greater than 0, then once the cache becomes stale (more time than the TTL has passed), a new cache will be
         downloaded synchronously, which may increase latencies if the triggering method is get_online_features().
         """
-        registry_config = self.config.get_registry_config()
+        registry_config = self.config.registry
         registry = Registry(
             self.config.project, registry_config, repo_path=self.repo_path
         )

sdk/python/feast/infra/materialization/snowflake_engine.py

@@ -132,6 +132,10 @@ def update(
             # if the stage already exists,
             # assumes that the materialization functions have been deployed
             if f"feast_{project}" in stage_list["name"].tolist():
+                click.echo(
+                    f"Materialization functions for {Style.BRIGHT + Fore.GREEN}{project}{Style.RESET_ALL} already detected."
+                )
+                click.echo()
                 return None
 
             click.echo(

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

@@ -9,7 +9,6 @@
     EntityNotFoundException,
     FeatureServiceNotFoundException,
     FeatureViewNotFoundException,
-    OnDemandFeatureViewNotFoundException,
     SavedDatasetNotFound,
     ValidationReferenceNotFound,
 )
@@ -98,7 +97,7 @@ def get_on_demand_feature_view(
             and on_demand_feature_view.spec.name == name
         ):
             return OnDemandFeatureView.from_proto(on_demand_feature_view)
-    raise OnDemandFeatureViewNotFoundException(name, project=project)
+    raise FeatureViewNotFoundException(name, project=project)
 
 
 def get_data_source(
@@ -138,10 +137,6 @@ def get_validation_reference(
     raise ValidationReferenceNotFound(name, project=project)
 
 
-def list_validation_references(registry_proto: RegistryProto):
-    return registry_proto.validation_references
-
-
 def list_feature_services(
     registry_proto: RegistryProto, project: str, allow_cache: bool = False
 ) -> List[FeatureService]:
@@ -215,13 +210,25 @@ def list_data_sources(registry_proto: RegistryProto, project: str) -> List[DataS
 
 
 def list_saved_datasets(
-    registry_proto: RegistryProto, project: str, allow_cache: bool = False
+    registry_proto: RegistryProto, project: str
 ) -> List[SavedDataset]:
-    return [
-        SavedDataset.from_proto(saved_dataset)
-        for saved_dataset in registry_proto.saved_datasets
-        if saved_dataset.spec.project == project
-    ]
+    saved_datasets = []
+    for saved_dataset in registry_proto.saved_datasets:
+        if saved_dataset.project == project:
+            saved_datasets.append(SavedDataset.from_proto(saved_dataset))
+    return saved_datasets
+
+
+def list_validation_references(
+    registry_proto: RegistryProto, project: str
+) -> List[ValidationReference]:
+    validation_references = []
+    for validation_reference in registry_proto.validation_references:
+        if validation_reference.project == project:
+            validation_references.append(
+                ValidationReference.from_proto(validation_reference)
+            )
+    return validation_references
 
 
 def list_project_metadata(

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

@@ -174,6 +174,10 @@ def __new__(
             from feast.infra.registry.sql import SqlRegistry
 
             return SqlRegistry(registry_config, project, repo_path)
+        elif registry_config and registry_config.registry_type == "snowflake.registry":
+            from feast.infra.registry.snowflake import SnowflakeRegistry
+
+            return SnowflakeRegistry(registry_config, project, repo_path)
         else:
             return super(Registry, cls).__new__(cls)
 
@@ -731,7 +735,7 @@ def list_validation_references(
         registry_proto = self._get_registry_proto(
             project=project, allow_cache=allow_cache
         )
-        return proto_registry_utils.list_validation_references(registry_proto)
+        return proto_registry_utils.list_validation_references(registry_proto, project)
 
     def delete_validation_reference(self, name: str, project: str, commit: bool = True):
         registry_proto = self._prepare_registry_for_changes(project)