feat: Move data source validation entrypoint to offline store (#4197)

* move validate_data_source entrypoint to offline store

Signed-off-by: tokoko <togurg14@freeuni.edu.ge>

* add validate_data_source to foo provider

Signed-off-by: tokoko <togurg14@freeuni.edu.ge>

---------

Signed-off-by: tokoko <togurg14@freeuni.edu.ge>

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

@@ -351,3 +351,17 @@ def offline_write_batch(
                 to show progress.
         """
         raise NotImplementedError
+
+    @staticmethod
+    def validate_data_source(
+        config: RepoConfig,
+        data_source: DataSource,
+    ):
+        """
+        Validates the underlying data source.
+
+        Args:
+            config: Configuration object used to configure a feature store.
+            data_source: DataSource object that needs to be validated
+        """
+        data_source.validate(config=config)

sdk/python/feast/infra/passthrough_provider.py

@@ -7,6 +7,7 @@
 
 from feast import importer
 from feast.batch_feature_view import BatchFeatureView
+from feast.data_source import DataSource
 from feast.entity import Entity
 from feast.feature_logging import FeatureServiceLoggingSource
 from feast.feature_service import FeatureService
@@ -383,3 +384,10 @@ def retrieve_feature_service_logs(
             start_date=make_tzaware(start_date),
             end_date=make_tzaware(end_date),
         )
+
+    def validate_data_source(
+        self,
+        config: RepoConfig,
+        data_source: DataSource,
+    ):
+        self.offline_store.validate_data_source(config=config, data_source=data_source)

sdk/python/feast/infra/provider.py

@@ -8,6 +8,7 @@
 from tqdm import tqdm
 
 from feast import FeatureService, errors
+from feast.data_source import DataSource
 from feast.entity import Entity
 from feast.feature_view import FeatureView
 from feast.importer import import_class
@@ -352,6 +353,21 @@ def retrieve_online_documents(
         """
         pass
 
+    @abstractmethod
+    def validate_data_source(
+        self,
+        config: RepoConfig,
+        data_source: DataSource,
+    ):
+        """
+        Validates the underlying data source.
+
+        Args:
+            config: Configuration object used to configure a feature store.
+            data_source: DataSource object that needs to be validated
+        """
+        pass
+
 
 def get_provider(config: RepoConfig) -> Provider:
     if "." not in config.provider:

sdk/python/feast/repo_operations.py

@@ -205,10 +205,11 @@ def plan(repo_config: RepoConfig, repo_path: Path, skip_source_validation: bool)
     project, registry, repo, store = _prepare_registry_and_repo(repo_config, repo_path)
 
     if not skip_source_validation:
+        provider = store._get_provider()
         data_sources = [t.batch_source for t in repo.feature_views]
         # Make sure the data source used by this feature view is supported by Feast
         for data_source in data_sources:
-            data_source.validate(store.config)
+            provider.validate_data_source(store.config, data_source)
 
     registry_diff, infra_diff, _ = store.plan(repo)
     click.echo(registry_diff.to_string())
@@ -282,10 +283,11 @@ def apply_total_with_repo_instance(
     skip_source_validation: bool,
 ):
     if not skip_source_validation:
+        provider = store._get_provider()
         data_sources = [t.batch_source for t in repo.feature_views]
         # Make sure the data source used by this feature view is supported by Feast
         for data_source in data_sources:
-            data_source.validate(store.config)
+            provider.validate_data_source(store.config, data_source)
 
     # For each object in the registry, determine whether it should be kept or deleted.
     (

sdk/python/tests/foo_provider.py

@@ -7,6 +7,7 @@
 from tqdm import tqdm
 
 from feast import Entity, FeatureService, FeatureView, RepoConfig
+from feast.data_source import DataSource
 from feast.infra.offline_stores.offline_store import RetrievalJob
 from feast.infra.provider import Provider
 from feast.infra.registry.base_registry import BaseRegistry
@@ -130,3 +131,10 @@ def retrieve_online_documents(
         ]
     ]:
         return []
+
+    def validate_data_source(
+        self,
+        config: RepoConfig,
+        data_source: DataSource,
+    ):
+        pass