move validate logic back to data_source

Signed-off-by: Cody Lin <codyl@twitter.com>
feast-dev · feast-ci-bot · Jun 21, 2021 · Jun 7, 2021 · Jun 8, 2021 · Jun 8, 2021
commit 6b3975a24da2cae221d5676680e5d42d6f49cd71
@@ -20,6 +20,7 @@
 
 from feast import type_map
 from feast.data_format import FileFormat, StreamFormat
+from feast.errors import DataSourceNotFoundException
 from feast.protos.feast.core.DataSource_pb2 import DataSource as DataSourceProto
 from feast.value_type import ValueType
 
@@ -519,6 +520,46 @@ def to_proto(self) -> DataSourceProto:
         """
         raise NotImplementedError
 
+    def validate(self):
+        """
+        Validates the underlying data source.
+        """
+        raise NotImplementedError
+
+    def _infer_event_timestamp_column(self, ts_column_type_regex_pattern):
+        ERROR_MSG_PREFIX = "Unable to infer DataSource event_timestamp_column"
+        USER_GUIDANCE = "Please specify event_timestamp_column explicitly."
+
+        if isinstance(self, FileSource) or isinstance(self, BigQuerySource):
+            event_timestamp_column, matched_flag = None, False
+            for col_name, col_datatype in self.get_table_column_names_and_types():
+                if re.match(ts_column_type_regex_pattern, col_datatype):
+                    if matched_flag:
+                        raise TypeError(
+                            f"""
+                            {ERROR_MSG_PREFIX} due to multiple possible columns satisfying
+                            the criteria. {USER_GUIDANCE}
+                            """
+                        )
+                    matched_flag = True
+                    event_timestamp_column = col_name
+            if matched_flag:
+                return event_timestamp_column
+            else:
+                raise TypeError(
+                    f"""
+                    {ERROR_MSG_PREFIX} due to an absence of columns that satisfy the criteria.
+                     {USER_GUIDANCE}
+                    """
+                )
+        else:
+            raise TypeError(
+                f"""
+                {ERROR_MSG_PREFIX} because this DataSource currently does not support this inference.
+                 {USER_GUIDANCE}
+                """
+            )
+
 
 class FileSource(DataSource):
     def __init__(
@@ -615,6 +656,10 @@ def to_proto(self) -> DataSourceProto:
 
         return data_source_proto
 
+    def validate(self):
+        # TODO: validate a FileSource
+        pass
+
     @staticmethod
     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
         return type_map.pa_to_feast_value_type
@@ -692,6 +737,17 @@ def to_proto(self) -> DataSourceProto:
 
         return data_source_proto
 
+    def validate(self):
+        if not self.query:
+            from google.api_core.exceptions import NotFound
+            from google.cloud import bigquery
+
+            client = bigquery.Client()
+            try:
+                client.get_table(self.table_ref)
+            except NotFound:
+                raise DataSourceNotFoundException(self.table_ref)
+
     def get_table_query_string(self) -> str:
         """Returns a string that can directly be used to reference this table in SQL"""
         if self.table_ref:

@@ -12,7 +12,6 @@
 from click.exceptions import BadParameter
 
 from feast import Entity, FeatureTable
-from feast.errors import DataSourceNotFoundException
 from feast.feature_view import FeatureView
 from feast.inference import (
     infer_entity_value_type_from_feature_views,
@@ -164,6 +163,7 @@ def apply_total(repo_config: RepoConfig, repo_path: Path):
     update_data_sources_with_inferred_event_timestamp_col(data_sources)
     for view in repo.feature_views:
         view.infer_features_from_input_source()
+        data_source.validate()
 
     tables_to_delete = []
     for registry_table in registry.list_feature_tables(project=project):

@@ -80,9 +80,11 @@ def test_missing_bq_source_fail() -> None:
 
         repo_example = repo_path / "example.py"
         repo_example.write_text(
-            (Path(__file__).parent / "example_feature_repo_with_missing_bq_source.py").read_text()
+            (
+                Path(__file__).parent / "example_feature_repo_with_missing_bq_source.py"
+            ).read_text()
         )
 
         returncode, output = runner.run_with_output(["apply"], cwd=repo_path)
         assert returncode == 1
-        assert "DataSourceNotFoundException" in output
+        assert b"DataSourceNotFoundException" in output