Fix: MCE validation error

metaphor/common/sql/utils.py

@@ -0,0 +1,13 @@
+from typing import List
+
+from metaphor.models.metadata_change_event import QueriedDataset
+
+
+def is_valid_queried_datasets(datasets: List[QueriedDataset]) -> bool:
+    """
+    database and schema must exist and not an empty string
+    """
+    for dataset in datasets:
+        if not dataset.database or not dataset.schema or not dataset.table:
+            return False
+    return True

metaphor/postgresql/extractor.py

@@ -2,13 +2,6 @@
 from datetime import datetime, timezone
 from typing import Collection, Dict, Iterator, List, Optional, Tuple
 
-from metaphor.common.fieldpath import build_schema_field
-from metaphor.common.sql.process_query.process_query import process_query
-from metaphor.common.sql.table_level_lineage.table_level_lineage import (
-    extract_table_level_lineage,
-)
-from metaphor.common.utils import md5_digest, safe_float, start_of_day
-
 try:
     import asyncpg
     import boto3
@@ -19,8 +12,15 @@
 from metaphor.common.base_extractor import BaseExtractor
 from metaphor.common.entity_id import dataset_normalized_name
 from metaphor.common.event_util import ENTITY_TYPES
+from metaphor.common.fieldpath import build_schema_field
 from metaphor.common.logger import get_logger
 from metaphor.common.models import to_dataset_statistics
+from metaphor.common.sql.process_query.process_query import process_query
+from metaphor.common.sql.table_level_lineage.table_level_lineage import (
+    extract_table_level_lineage,
+)
+from metaphor.common.sql.utils import is_valid_queried_datasets
+from metaphor.common.utils import md5_digest, safe_float, start_of_day
 from metaphor.models.crawler_run_metadata import Platform
 from metaphor.models.metadata_change_event import (
     DataPlatform,
@@ -483,6 +483,12 @@
         if not tll.sources and not tll.targets:
             return None
 
+        # Skip if parsed sources or targets has invalid data.
+        if not is_valid_queried_datasets(tll.sources) or not is_valid_queried_datasets(
+            tll.targets
+        ):
+            return None
+
         sql_hash = md5_digest(query.encode("utf-8"))
         sql: Optional[str] = query
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "metaphor-connectors"
-version = "0.14.97"
+version = "0.14.98"
 license = "Apache-2.0"
 description = "A collection of Python-based 'connectors' that extract metadata from various sources to ingest into the Metaphor app."
 authors = ["Metaphor <[email protected]>"]