feat(ingest/bigquery): Respect dataset and table patterns when ingest…

…ing lineage via catalog api (datahub-project#10080)
sleeperdeep · Jun 25, 2024 · 581af25 · 581af25
1 parent d0e0b2d
commit 581af25
Showing 1 changed file with 15 additions and 11 deletions.
diff --git a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py
@@ -517,19 +517,23 @@ def lineage_via_catalog_lineage_api(
                     ]
                 )
 
-            # Convert project tables to <project_id>.<dataset_id>.<table_id> format
-            project_table_names = list(
-                map(
-                    lambda table: "{}.{}.{}".format(
-                        table.project, table.dataset_id, table.table_id
-                    ),
-                    project_tables,
-                )
-            )
-
             lineage_map: Dict[str, Set[LineageEdge]] = {}
             curr_date = datetime.now()
-            for table in project_table_names:
+            for project_table in project_tables:
+                # Convert project table to <project_id>.<dataset_id>.<table_id> format
+                table = f"{project_table.project}.{project_table.dataset_id}.{project_table.table_id}"
+
+                if not is_schema_allowed(
+                    self.config.dataset_pattern,
+                    schema_name=project_table.dataset_id,
+                    db_name=project_table.project,
+                    match_fully_qualified_schema_name=self.config.match_fully_qualified_names,
+                ) or not self.config.table_pattern.allowed(table):
+                    self.report.num_skipped_lineage_entries_not_allowed[
+                        project_table.project
+                    ] += 1
+                    continue
+
                 logger.info("Creating lineage map for table %s", table)
                 upstreams = set()
                 downstream_table = lineage_v1.EntityReference()