fix: Don't prevent apply from running given duplicate empty names in data sources. Also fix repeated apply of Spark data source. (#2415)

* fix: Print more warning statements on requirement for data sources to have name in future, but don't prevent apply from running if there are duplicate empty data sources. Also attach class type when applying data sources so repeated feast apply commands properly work for Spark

Signed-off-by: Danny Chiao <[email protected]>

* typo

Signed-off-by: Danny Chiao <[email protected]>

* typo

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* More tests

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* revert

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

* fix

Signed-off-by: Danny Chiao <[email protected]>

Author: adchia

protos/feast/core/SavedDataset.proto

@@ -24,7 +24,6 @@ option go_package = "github.com/feast-dev/feast/go/protos/feast/core";
 
 import "google/protobuf/timestamp.proto";
 import "feast/core/DataSource.proto";
-import "feast/core/FeatureService.proto";
 
 message SavedDatasetSpec {
   // Name of the dataset. Must be unique since it's possible to overwrite dataset by name

sdk/python/feast/cli.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import logging
+import warnings
 from datetime import datetime
 from pathlib import Path
 from typing import List, Optional
@@ -151,6 +152,11 @@ def data_source_describe(ctx: click.Context, name: str):
         print(e)
         exit(1)
 
+    warnings.warn(
+        "Describing data sources will only work properly if all data sources have names or table names specified. "
+        "Starting Feast 0.21, data source unique names will be required to encourage data source discovery.",
+        RuntimeWarning,
+    )
     print(
         yaml.dump(
             yaml.safe_load(str(data_source)), default_flow_style=False, sort_keys=False
@@ -173,6 +179,11 @@ def data_source_list(ctx: click.Context):
 
     from tabulate import tabulate
 
+    warnings.warn(
+        "Listing data sources will only work properly if all data sources have names or table names specified. "
+        "Starting Feast 0.21, data source unique names will be required to encourage data source discovery",
+        RuntimeWarning,
+    )
     print(tabulate(table, headers=["NAME", "CLASS"], tablefmt="plain"))
 
 

sdk/python/feast/data_source.py

@@ -17,6 +17,8 @@
 from abc import ABC, abstractmethod
 from typing import Any, Callable, Dict, Iterable, Optional, Tuple
 
+from google.protobuf.json_format import MessageToJson
+
 from feast import type_map
 from feast.data_format import StreamFormat
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
@@ -192,6 +194,9 @@ def __init__(
     def __hash__(self):
         return hash((id(self), self.name))
 
+    def __str__(self):
+        return str(MessageToJson(self.to_proto()))
+
     def __eq__(self, other):
         if not isinstance(other, DataSource):
             raise TypeError("Comparisons should only involve DataSource class objects.")

sdk/python/feast/diff/registry_diff.py

@@ -60,6 +60,9 @@ def to_string(self):
                 continue
             if feast_object_diff.transition_type == TransitionType.UNCHANGED:
                 continue
+            if feast_object_diff.feast_object_type == FeastObjectType.DATA_SOURCE:
+                # TODO(adchia): Print statements out starting in Feast 0.21
+                continue
             action, color = message_action_map[feast_object_diff.transition_type]
             log_string += f"{action} {feast_object_diff.feast_object_type.value} {Style.BRIGHT + color}{feast_object_diff.name}{Style.RESET_ALL}\n"
             if feast_object_diff.transition_type == TransitionType.UPDATE:
@@ -78,8 +81,11 @@ def to_string(self):
 def tag_objects_for_keep_delete_update_add(
     existing_objs: Iterable[FeastObject], desired_objs: Iterable[FeastObject]
 ) -> Tuple[Set[FeastObject], Set[FeastObject], Set[FeastObject], Set[FeastObject]]:
-    existing_obj_names = {e.name for e in existing_objs}
-    desired_obj_names = {e.name for e in desired_objs}
+    # TODO(adchia): Remove the "if X.name" condition when data sources are forced to have names
+    existing_obj_names = {e.name for e in existing_objs if e.name}
+    desired_objs = [obj for obj in desired_objs if obj.name]
+    existing_objs = [obj for obj in existing_objs if obj.name]
+    desired_obj_names = {e.name for e in desired_objs if e.name}
 
     objs_to_add = {e for e in desired_objs if e.name not in existing_obj_names}
     objs_to_update = {e for e in desired_objs if e.name in existing_obj_names}

sdk/python/feast/feature_store.py

@@ -2063,14 +2063,17 @@ def _validate_feature_views(feature_views: List[BaseFeatureView]):
 def _validate_data_sources(data_sources: List[DataSource]):
     """Verify data sources have case-insensitively unique names"""
     ds_names = set()
-    for fv in data_sources:
-        case_insensitive_ds_name = fv.name.lower()
+    for ds in data_sources:
+        case_insensitive_ds_name = ds.name.lower()
         if case_insensitive_ds_name in ds_names:
-            raise ValueError(
-                f"More than one data source with name {case_insensitive_ds_name} found. "
-                f"Please ensure that all data source names are case-insensitively unique. "
-                f"It may be necessary to ignore certain files in your feature repository by using a .feastignore file."
-            )
+            if case_insensitive_ds_name.strip():
+                warnings.warn(
+                    f"More than one data source with name {case_insensitive_ds_name} found. "
+                    f"Please ensure that all data source names are case-insensitively unique. "
+                    f"It may be necessary to ignore certain files in your feature repository by using a .feastignore "
+                    f"file. Starting in Feast 0.21, unique names (perhaps inferred from the table name) will be "
+                    f"required in data sources to encourage data source discovery"
+                )
         else:
             ds_names.add(case_insensitive_ds_name)
 

sdk/python/feast/infra/offline_stores/bigquery_source.py

@@ -64,7 +64,7 @@ def __init__(
             else:
                 warnings.warn(
                     (
-                        "Starting in Feast 0.21, Feast will require either a name for a data source (if using query) or `table`."
+                        f"Starting in Feast 0.21, Feast will require either a name for a data source (if using query) or `table`: {self.query}"
                     ),
                     DeprecationWarning,
                 )

sdk/python/feast/infra/offline_stores/redshift_source.py

@@ -41,6 +41,12 @@ def __init__(
             query (optional): The query to be executed to obtain the features.
             name (optional): Name for the source. Defaults to the table_ref if not specified.
         """
+        # The default Redshift schema is named "public".
+        _schema = "public" if table and not schema else schema
+        self.redshift_options = RedshiftOptions(
+            table=table, schema=_schema, query=query
+        )
+
         if table is None and query is None:
             raise ValueError('No "table" argument provided.')
         _name = name
@@ -50,7 +56,8 @@ def __init__(
             else:
                 warnings.warn(
                     (
-                        "Starting in Feast 0.21, Feast will require either a name for a data source (if using query) or `table`."
+                        f"Starting in Feast 0.21, Feast will require either a name for a data source (if using query) "
+                        f"or `table`: {self.query}"
                     ),
                     DeprecationWarning,
                 )
@@ -63,13 +70,6 @@ def __init__(
             date_partition_column,
         )
 
-        # The default Redshift schema is named "public".
-        _schema = "public" if table and not schema else schema
-
-        self.redshift_options = RedshiftOptions(
-            table=table, schema=_schema, query=query
-        )
-
     @staticmethod
     def from_proto(data_source: DataSourceProto):
         """

sdk/python/feast/infra/offline_stores/snowflake_source.py

@@ -44,6 +44,12 @@ def __init__(
         """
         if table is None and query is None:
             raise ValueError('No "table" argument provided.')
+        # The default Snowflake schema is named "PUBLIC".
+        _schema = "PUBLIC" if (database and table and not schema) else schema
+
+        self.snowflake_options = SnowflakeOptions(
+            database=database, schema=_schema, table=table, query=query
+        )
 
         # If no name, use the table as the default name
         _name = name
@@ -53,7 +59,8 @@ def __init__(
             else:
                 warnings.warn(
                     (
-                        "Starting in Feast 0.21, Feast will require either a name for a data source (if using query) or `table`."
+                        f"Starting in Feast 0.21, Feast will require either a name for a data source (if using query) "
+                        f"or `table`: {self.query}"
                     ),
                     DeprecationWarning,
                 )
@@ -66,13 +73,6 @@ def __init__(
             date_partition_column,
         )
 
-        # The default Snowflake schema is named "PUBLIC".
-        _schema = "PUBLIC" if (database and table and not schema) else schema
-
-        self.snowflake_options = SnowflakeOptions(
-            database=database, schema=_schema, table=table, query=query
-        )
-
     @staticmethod
     def from_proto(data_source: DataSourceProto):
         """

sdk/python/feast/registry.py

@@ -314,11 +314,13 @@ def apply_data_source(
             commit: Whether to immediately commit to the registry
         """
         registry = self._prepare_registry_for_changes()
-
         for idx, existing_data_source_proto in enumerate(registry.data_sources):
             if existing_data_source_proto.name == data_source.name:
                 del registry.data_sources[idx]
         data_source_proto = data_source.to_proto()
+        data_source_proto.data_source_class_type = (
+            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__}"

sdk/python/feast/repo_operations.py

@@ -188,10 +188,8 @@ def extract_objects_for_apply_delete(project, registry, repo):
     return (
         all_to_apply,
         all_to_delete,
-        set(
-            objs_to_add[FeastObjectType.FEATURE_VIEW].union(
-                objs_to_update[FeastObjectType.FEATURE_VIEW]
-            )
+        set(objs_to_add[FeastObjectType.FEATURE_VIEW]).union(
+            set(objs_to_update[FeastObjectType.FEATURE_VIEW])
         ),
         objs_to_delete[FeastObjectType.FEATURE_VIEW],
     )