From a96a512166564cf9c40af4b83e7138dcb48c914d Mon Sep 17 00:00:00 2001 From: Tamas Nemeth Date: Thu, 26 Oct 2023 18:46:10 +0200 Subject: [PATCH] fix(ingest/bigquery): Fixing lineage filter query (#9114) --- .../ingestion/source/bigquery_v2/bigquery_config.py | 1 + .../datahub/ingestion/source/bigquery_v2/lineage.py | 8 ++++++-- .../src/datahub/ingestion/source/bigquery_v2/usage.py | 10 ++++++---- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/bigquery_config.py b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/bigquery_config.py index 6203192769750..f762d451849ab 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/bigquery_config.py +++ b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/bigquery_config.py @@ -309,6 +309,7 @@ def backward_compatibility_configs_set(cls, values: Dict) -> Dict: "dataset_pattern is not set but schema_pattern is set, using schema_pattern as dataset_pattern. schema_pattern will be deprecated, please use dataset_pattern instead." ) values["dataset_pattern"] = schema_pattern + dataset_pattern = schema_pattern elif ( dataset_pattern != AllowDenyPattern.allow_all() and schema_pattern != AllowDenyPattern.allow_all() diff --git a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py index aa462435b8105..e9acf5ea86044 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py +++ b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/lineage.py @@ -20,6 +20,7 @@ from google.cloud.datacatalog import lineage_v1 from google.cloud.logging_v2.client import Client as GCPLoggingClient +from datahub.configuration.pattern_utils import is_schema_allowed from datahub.emitter import mce_builder from datahub.emitter.mcp import MetadataChangeProposalWrapper from datahub.ingestion.api.workunit import MetadataWorkUnit @@ -683,8 +684,11 @@ def _create_lineage_map( self.report.num_skipped_lineage_entries_missing_data[e.project_id] += 1 continue - if not self.config.dataset_pattern.allowed( - destination_table.table_identifier.dataset + if not is_schema_allowed( + self.config.dataset_pattern, + destination_table.table_identifier.dataset, + destination_table.table_identifier.project_id, + self.config.match_fully_qualified_names, ) or not self.config.table_pattern.allowed( destination_table.table_identifier.get_table_name() ): diff --git a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/usage.py b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/usage.py index 7fc38991e5928..65b559550ffc5 100644 --- a/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/usage.py +++ b/metadata-ingestion/src/datahub/ingestion/source/bigquery_v2/usage.py @@ -21,6 +21,7 @@ import humanfriendly +from datahub.configuration.pattern_utils import is_schema_allowed from datahub.configuration.time_window_config import ( BaseTimeWindowConfig, get_time_bucket, @@ -335,10 +336,11 @@ def get_time_window(self) -> Tuple[datetime, datetime]: def _is_table_allowed(self, table_ref: Optional[BigQueryTableRef]) -> bool: return ( table_ref is not None - and self.config.dataset_pattern.allowed( - f"{table_ref.table_identifier.project_id}.{table_ref.table_identifier.dataset}" - if self.config.match_fully_qualified_names - else table_ref.table_identifier.dataset + and is_schema_allowed( + self.config.dataset_pattern, + table_ref.table_identifier.dataset, + table_ref.table_identifier.project_id, + self.config.match_fully_qualified_names, ) and self.config.table_pattern.allowed(str(table_ref.table_identifier)) )