feat(ingest/s3): add table filtering

- add tables_filter_pattern to path_spec
- add table filtering to S3Source().s3_browser()

Author: eagle-25

metadata-ingestion/src/datahub/ingestion/source/data_lake_common/path_spec.py

@@ -11,7 +11,7 @@
 from pydantic.fields import Field
 from wcmatch import pathlib
 
-from datahub.configuration.common import ConfigModel
+from datahub.configuration.common import AllowDenyPattern, ConfigModel
 from datahub.ingestion.source.aws.s3_util import is_s3_uri
 from datahub.ingestion.source.azure.abs_utils import is_abs_uri
 from datahub.ingestion.source.gcs.gcs_utils import is_gcs_uri
@@ -145,6 +145,11 @@ class Config:
         description="Include hidden folders in the traversal (folders starting with . or _",
     )
 
+    tables_filter_pattern: AllowDenyPattern = Field(
+        default=AllowDenyPattern.allow_all(),
+        description="The tables_filter_pattern configuration field uses regular expressions to filter the tables part of the Pathspec for ingestion, allowing fine-grained control over which tables are included or excluded based on specified patterns. The default setting allows all tables.",
+    )
+
     def is_path_hidden(self, path: str) -> bool:
         # Split the path into directories and filename
         dirs, filename = os.path.split(path)
@@ -177,6 +182,12 @@ def allowed(self, path: str, ignore_ext: bool = False) -> bool:
                 ):
                     return False
         logger.debug(f"{path} is not excluded")
+
+        table_name, _ = self.extract_table_name_and_path(path)
+        if not self.tables_filter_pattern.allowed(table_name):
+            return False
+        logger.debug(f"{path} is passed table name check")
+
         ext = os.path.splitext(path)[1].strip(".")
 
         if not ignore_ext:
@@ -218,6 +229,15 @@ def dir_allowed(self, path: str) -> bool:
                     exclude_path.rstrip("/"), flags=pathlib.GLOBSTAR
                 ):
                     return False
+
+        file_name_pattern = self.include.rsplit("/", 1)[1]
+        table_name, _ = self.extract_table_name_and_path(
+            os.path.join(path, file_name_pattern)
+        )
+        if not self.tables_filter_pattern.allowed(table_name):
+            return False
+        logger.debug(f"{path} is passed table name check")
+
         return True
 
     @classmethod

metadata-ingestion/src/datahub/ingestion/source/s3/source.py

@@ -945,6 +945,17 @@ def s3_browser(self, path_spec: PathSpec, sample_size: int) -> Iterable[BrowsePa
                     for f in list_folders(
                         bucket_name, f"{folder}", self.source_config.aws_config
                     ):
+                        table_path = self.create_s3_path(bucket_name, f)
+                        table_name, _ = path_spec.extract_table_name_and_path(
+                            table_path
+                        )
+                        if not path_spec.tables_filter_pattern.allowed(table_name):
+                            logger.debug(
+                                f"Table '{table_name}' not allowed and skipping"
+                            )
+                            self.report.report_file_dropped(table_path)
+                            continue
+
                         dirs_to_process = []
                         logger.info(f"Processing folder: {f}")
                         if path_spec.traversal_method == FolderTraversalMethod.ALL: