working state

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>
feast-dev · feast-ci-bot · Aug 19, 2022 · Aug 5, 2022 · Aug 10, 2022 · Aug 10, 2022
commit 011d1e0750c55d464067f26a3b7d6b515edac5d5
@@ -297,6 +297,7 @@ def from_proto(data_source: DataSourceProto) -> Any:
             raise ValueError("Could not identify the source type being added.")
 
         if data_source_type == DataSourceProto.SourceType.CUSTOM_SOURCE:
+            data_source.data_source_class_type = "feast.infra.offline_stores.contrib.mssql_offline_store.mssqlserver_source.MsSqlServerSource"
             cls = get_data_source_class_from_type(data_source.data_source_class_type)
             return cls.from_proto(data_source)
 

@@ -564,9 +564,7 @@ def _validate_all_feature_views(
                 "This API is stable, but the functionality does not scale well for offline retrieval",
                 RuntimeWarning,
             )
-
         set_usage_attribute("odfv", bool(odfvs_to_update))
-
         _validate_feature_views(
             [
                 *views_to_update,

diff --git a/sdk/python/feast/feature_view.py b/sdk/python/feast/feature_view.py
@@ -351,6 +351,7 @@ def from_proto(cls, feature_view_proto: FeatureViewProto):
         Returns:
             A FeatureViewProto object based on the feature view protobuf.
         """
+        print(feature_view_proto.spec.batch_source)
         batch_source = DataSource.from_proto(feature_view_proto.spec.batch_source)
         stream_source = (
             DataSource.from_proto(feature_view_proto.spec.stream_source)

@@ -38,7 +38,7 @@ def __init__(self, config: RepoConfig):
             else None
         )
 
-    @log_exceptions_and_usage(registry="az")
+    #@log_exceptions_and_usage(registry="az")
     def update_infra(
         self,
         project: str,
@@ -59,7 +59,7 @@ def update_infra(
                 partial=partial,
             )
 
-    @log_exceptions_and_usage(registry="az")
+    #@log_exceptions_and_usage(registry="az")
     def teardown_infra(
         self,
         project: str,
@@ -69,7 +69,7 @@ def teardown_infra(
         if self.online_store:
             self.online_store.teardown(self.repo_config, tables, entities)
 
-    @log_exceptions_and_usage(registry="az")
+    #@log_exceptions_and_usage(registry="az")
     def online_write_batch(
         self,
         config: RepoConfig,
@@ -82,7 +82,7 @@ def online_write_batch(
         if self.online_store:
             self.online_store.online_write_batch(config, table, data, progress)
 
-    @log_exceptions_and_usage(sampler=RatioSampler(ratio=0.001), registry="az")
+    #@log_exceptions_and_usage(sampler=RatioSampler(ratio=0.001), registry="az")
     def online_read(
         self,
         config: RepoConfig,
@@ -159,7 +159,7 @@ def materialize_single_feature_view(
                     lambda x: pbar.update(x),
                 )
 
-    @log_exceptions_and_usage(registry="az")
+    #@log_exceptions_and_usage(registry="az")
     def get_historical_features(
         self,
         config: RepoConfig,

@@ -27,7 +27,6 @@
 from feast import errors
 from feast.data_source import DataSource
 
-from .mssqlserver_source import MsSqlServerSource
 from feast.feature_view import FeatureView
 from feast.infra.offline_stores.file_source import SavedDatasetFileStorage
 from feast.infra.offline_stores.offline_store import (
@@ -45,7 +44,6 @@
 from feast.repo_config import FeastBaseModel, RepoConfig
 from feast.saved_dataset import SavedDatasetStorage
 from feast import FileSource
-from feast.usage import log_exceptions_and_usage
 from feast.utils import _get_requested_feature_views_to_features_dict
 
 EntitySchema = Dict[str, np.dtype]
@@ -55,8 +53,8 @@ class MsSqlServerOfflineStoreConfig(FeastBaseModel):
     """Offline store config for SQL Server"""
 
     type: Literal[
-        "feast_azure_provider.mssqlserver.MsSqlServerOfflineStore"
-    ] = "feast_azure_provider.mssqlserver.MsSqlServerOfflineStore"
+        "mssql"
+    ] = "mssql"
     """ Offline store type selector"""
 
     connection_string: StrictStr = "mssql+pyodbc://sa:yourStrong(!)Password@localhost:1433/feast_test?driver=ODBC+Driver+17+for+SQL+Server"
@@ -74,7 +72,7 @@ def _make_engine(self, config: RepoConfig = None) -> Session:
         return self._engine
 
     @staticmethod
-    @log_exceptions_and_usage(offline_store="mssql")
+    #@log_exceptions_and_usage(offline_store="mssql")
     def pull_latest_from_table_or_query(
         self,
         config: RepoConfig,
@@ -87,10 +85,6 @@ def pull_latest_from_table_or_query(
         end_date: datetime,
     ) -> RetrievalJob:
         assert type(data_source).__name__ == "MsSqlServerSource"
-        assert (
-            config.offline_store.type
-            == "feast_azure_provider.mssqlserver.MsSqlServerOfflineStore"
-        )
         from_expression = data_source.get_table_query_string().replace("`", "")
 
         partition_by_join_key_string = ", ".join(join_key_columns)
@@ -125,7 +119,7 @@ def pull_latest_from_table_or_query(
         )
 
     @staticmethod
-    @log_exceptions_and_usage(offline_store="mssql")
+    #@log_exceptions_and_usage(offline_store="mssql")
     def pull_all_from_table_or_query(
         self,
         config: RepoConfig,
@@ -137,10 +131,6 @@ def pull_all_from_table_or_query(
         end_date: datetime,
     ) -> RetrievalJob:
         assert type(data_source).__name__ == "MsSqlServerSource"
-        assert (
-            config.offline_store.type
-            == "feast_azure_provider.mssqlserver.MsSqlServerOfflineStore"
-        )
         from_expression = data_source.get_table_query_string().replace("`", "")
         timestamps = [event_timestamp_column]
         field_string = ", ".join(join_key_columns + feature_name_columns + timestamps)
@@ -163,7 +153,7 @@ def pull_all_from_table_or_query(
         )
 
     @staticmethod
-    @log_exceptions_and_usage(offline_store="mssql")
+    #@log_exceptions_and_usage(offline_store="mssql")
     def get_historical_features(
         self,
         config: RepoConfig,
@@ -437,7 +427,7 @@ def get_feature_view_query_context(
         else:
             ttl_seconds = 0
 
-        assert isinstance(feature_view.source, MsSqlServerSource)
+        #assert isinstance(feature_view.source, MsSqlServerSource)
 
         event_timestamp_column = feature_view.source.event_timestamp_column
         created_timestamp_column = feature_view.source.created_timestamp_column

@@ -12,7 +12,7 @@
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
 from feast.value_type import ValueType
 from feast.repo_config import RepoConfig
-
+from feast.infra.offline_stores.contrib.mssql_offline_store.mssql import MsSqlServerOfflineStoreConfig
 
 class MsSqlServerOptions:
     """
@@ -165,6 +165,7 @@ def mssqlserver_options(self, sqlserver_options):
     def from_proto(data_source: DataSourceProto):
         options = json.loads(data_source.custom_options.configuration)
         return MsSqlServerSource(
+            name=data_source.name,
             field_mapping=dict(data_source.field_mapping),
             table_ref=options["table_ref"],
             connection_str=options["connection_string"],
@@ -183,7 +184,7 @@ def to_proto(self) -> DataSourceProto:
         data_source_proto.timestamp_field = self.timestamp_field
         data_source_proto.created_timestamp_column = self.created_timestamp_column
         data_source_proto.date_partition_column = self.date_partition_column
-
+        data_source_proto.name = self.name
         return data_source_proto
 
     def get_table_query_string(self) -> str:
@@ -193,20 +194,22 @@ def get_table_query_string(self) -> str:
     def validate(self, config: RepoConfig):
         # As long as the query gets successfully executed, or the table exists,
         # the data source is validated. We don't need the results though.
-        # self.get_table_column_names_and_types()
+        self.get_table_column_names_and_types(config)
         return None
 
     @staticmethod
     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
         return type_map.mssqlserver_to_feast_value_type
 
-    def get_table_column_names_and_types(self) -> Iterable[Tuple[str, str]]:
-        conn = create_engine(self._connection_str)
+    def get_table_column_names_and_types(self, config: RepoConfig) -> Iterable[Tuple[str, str]]:
+        assert isinstance(config.offline_store, MsSqlServerOfflineStoreConfig)
+        conn = create_engine(config.offline_store.connection_string)
+        self._mssqlserver_options.connection_str = config.offline_store.connection_string
         name_type_pairs = []
         database, table_name = self.table_ref.split(".")
         columns_query = f"""
-            SELECT COLUMN_NAME, DATA_TYPE FROM {database}.INFORMATION_SCHEMA.COLUMNS
-            WHERE TABLE_NAME = '{table_name}'
+            SELECT COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS
+            WHERE TABLE_NAME = '{table_name} and table_schema = {database}'
         """
         table_schema = pandas.read_sql(columns_query, conn)
         name_type_pairs.extend(

@@ -25,6 +25,7 @@
     "gcp": "feast.infra.gcp.GcpProvider",
     "aws": "feast.infra.aws.AwsProvider",
     "local": "feast.infra.local.LocalProvider",
+    "azure": "feast.infra.contrib.azure_provider.AzureProvider",
 }
 
 

@@ -57,10 +57,11 @@
 REGISTRY_SCHEMA_VERSION = "1"
 
 REGISTRY_STORE_CLASS_FOR_TYPE = {
-    "GCSRegistryStore": "feast.infra.registry.gcs.GCSRegistryStore",
-    "S3RegistryStore": "feast.infra.registry.s3.S3RegistryStore",
+    "GCSRegistryStore": "feast.infra.gcp.GCSRegistryStore",
+    "S3RegistryStore": "feast.infra.aws.S3RegistryStore",
     "FileRegistryStore": "feast.infra.registry.file.FileRegistryStore",
-    "PostgreSQLRegistryStore": "feast.infra.registry.contrib.postgres.postgres_registry_store.PostgreSQLRegistryStore",
+    "PostgreSQLRegistryStore": "feast.infra.registry_stores.contrib.postgres.registry_store.PostgreSQLRegistryStore",
+    "AzureRegistryStore": "feast.infra.registry_stores.contrib.azure.registry_store.AzBlobRegistryStore"
 }
 
 REGISTRY_STORE_CLASS_FOR_SCHEME = {
@@ -322,9 +323,6 @@ def apply_data_source(
             f"{data_source.__class__.__module__}.{data_source.__class__.__name__}"
         )
         data_source_proto.project = project
-        data_source_proto.data_source_class_type = (
-            f"{data_source.__class__.__module__}.{data_source.__class__.__name__}"
-        )
         registry.data_sources.append(data_source_proto)
         if commit:
             self.commit()

@@ -61,7 +61,7 @@
     "trino": "feast.infra.offline_stores.contrib.trino_offline_store.trino.TrinoOfflineStore",
     "postgres": "feast.infra.offline_stores.contrib.postgres_offline_store.postgres.PostgreSQLOfflineStore",
     "athena": "feast.infra.offline_stores.contrib.athena_offline_store.athena.AthenaOfflineStore",
-    "mssql": "feast.infra.offline_stores.contrib.mssql_offline_store.msql.MsSqlServerOfflineStore"
+    "mssql": "feast.infra.offline_stores.contrib.mssql_offline_store.mssql.MsSqlServerOfflineStore"
 }
 
 FEATURE_SERVER_CONFIG_CLASS_FOR_TYPE = {