Fixes #24795: dbt snapshot model patch (#26874)

* Fix dbt snapshot metadata ingestion

* address comments

* address comments

* checkstyle

Author: SumanMaharana

ingestion/src/metadata/ingestion/source/database/dbt/dbt_utils.py

@@ -29,6 +29,7 @@
     DbtCommonEnum,
     RawQueriesEnum,
 )
+from metadata.ingestion.source.database.dbt.models import SnapshotNodeLocation
 from metadata.utils import entity_link
 from metadata.utils.logger import ingestion_logger
 
@@ -786,6 +787,30 @@ def get_data_model_path(manifest_node):
     return datamodel_path
 
 
+def get_snapshot_effective_schema_and_database(
+    manifest_node: Any,
+) -> SnapshotNodeLocation:
+    """
+    For snapshot nodes, config.target_schema and config.target_database
+    override manifest_node.schema_ and manifest_node.database respectively.
+    Returns a SnapshotNodeLocation with the resolved schema and database.
+    """
+    effective_schema: str = manifest_node.schema_
+    effective_database: Optional[str] = manifest_node.database
+    if hasattr(manifest_node, "config") and manifest_node.config:
+        if (
+            hasattr(manifest_node.config, "target_schema")
+            and manifest_node.config.target_schema
+        ):
+            effective_schema = manifest_node.config.target_schema
+        if (
+            hasattr(manifest_node.config, "target_database")
+            and manifest_node.config.target_database
+        ):
+            effective_database = manifest_node.config.target_database
+    return SnapshotNodeLocation(schema_=effective_schema, database=effective_database)
+
+
 def find_entity_by_type_and_fqn(
     metadata: OpenMetadata, entity_type: str, entity_fqn: str
 ) -> Optional[Any]:

ingestion/src/metadata/ingestion/source/database/dbt/metadata.py

@@ -98,6 +98,7 @@
     get_dbt_model_name,
     get_dbt_raw_query,
     get_manifest_column_name,
+    get_snapshot_effective_schema_and_database,
     validate_custom_property_value,
 )
 from metadata.ingestion.source.database.dbt.models import DbtMeta
@@ -615,8 +616,8 @@ def yield_dbt_tags(
                     if model_tags:
                         dbt_tags_list.extend(self.filter_tags(model_tags))
 
-                    # Add the tags from the columns
-                    for _, column in manifest_node.columns.items():
+                    # snapshot nodes may have columns=None (columns are inferred at runtime)
+                    for _, column in (manifest_node.columns or {}).items():
                         column_tags = column.tags
                         if column_tags:
                             dbt_tags_list.extend(self.filter_tags(column_tags))
@@ -903,10 +904,21 @@ def yield_data_models(
 
                     model_name = get_dbt_model_name(manifest_node)
 
-                    # Filter the dbt models based on filter patterns
+                    # snapshots can redirect output to a different schema/database via config.target_schema/target_database
+                    if resource_type == "snapshot":
+                        location = get_snapshot_effective_schema_and_database(
+                            manifest_node
+                        )
+                        node_schema = location.schema_
+                        node_database = location.database
+                    else:
+                        node_schema = manifest_node.schema_
+                        node_database = manifest_node.database
+
+                    # Filter the dbt models based on filter patterns using effective schema/database
                     filter_model = self.is_filtered(
-                        database_name=get_corrected_name(manifest_node.database),
-                        schema_name=get_corrected_name(manifest_node.schema_),
+                        database_name=get_corrected_name(node_database),
+                        schema_name=get_corrected_name(node_schema),
                         table_name=model_name,
                     )
                     if filter_model.is_filtered:
@@ -931,13 +943,12 @@ def yield_data_models(
                             )
                             or []
                         )
-
                     table_fqn = fqn.build(
                         self.metadata,
                         entity_type=Table,
                         service_name=self.config.serviceName,
-                        database_name=get_corrected_name(manifest_node.database),
-                        schema_name=get_corrected_name(manifest_node.schema_),
+                        database_name=get_corrected_name(node_database),
+                        schema_name=get_corrected_name(node_schema),
                         table_name=model_name,
                     )
 
@@ -1017,9 +1028,24 @@ def parse_upstream_nodes(self, manifest_entities, dbt_node):
                     parent_node = manifest_entities[node]
                     table_name = get_dbt_model_name(parent_node)
 
+                    parent_resource_type = getattr(
+                        parent_node.resource_type,
+                        "value",
+                        parent_node.resource_type,
+                    )
+                    if parent_resource_type == "snapshot":
+                        parent_location = get_snapshot_effective_schema_and_database(
+                            parent_node
+                        )
+                        parent_database = parent_location.database
+                        parent_schema = parent_location.schema_
+                    else:
+                        parent_database = parent_node.database
+                        parent_schema = parent_node.schema_
+
                     filter_model = self.is_filtered(
-                        database_name=get_corrected_name(parent_node.database),
-                        schema_name=get_corrected_name(parent_node.schema_),
+                        database_name=get_corrected_name(parent_database),
+                        schema_name=get_corrected_name(parent_schema),
                         table_name=table_name,
                     )
                     if filter_model.is_filtered:
@@ -1036,8 +1062,8 @@ def parse_upstream_nodes(self, manifest_entities, dbt_node):
                             self.metadata,
                             entity_type=Table,
                             service_name=self.config.serviceName,
-                            database_name=get_corrected_name(parent_node.database),
-                            schema_name=get_corrected_name(parent_node.schema_),
+                            database_name=get_corrected_name(parent_database),
+                            schema_name=get_corrected_name(parent_schema),
                             table_name=table_name,
                         )
 
@@ -1060,7 +1086,8 @@ def parse_data_model_columns(
         Method to parse the DBT columns
         """
         columns = []
-        manifest_columns = manifest_node.columns
+        # snapshot nodes default columns to None; treat as empty to avoid AttributeError
+        manifest_columns = manifest_node.columns or {}
         for key, manifest_column in manifest_columns.items():
             try:
                 logger.debug(f"Processing DBT column: {key}")

ingestion/src/metadata/ingestion/source/database/dbt/models.py

@@ -47,3 +47,10 @@ class DbtMetaOpenmetadata(BaseModel):
 
 class DbtMeta(BaseModel):
     openmetadata: Optional[DbtMetaOpenmetadata] = None
+
+
+class SnapshotNodeLocation(BaseModel):
+    """Resolved schema and database for a dbt snapshot node after applying config overrides."""
+
+    schema_: str
+    database: Optional[str] = None

ingestion/tests/unit/resources/datasets/manifest_snapshot.json

@@ -0,0 +1,95 @@
+{
+  "metadata": {
+    "dbt_schema_version": "https://schemas.getdbt.com/dbt/manifest/v12.json",
+    "dbt_version": "1.8.0",
+    "generated_at": "2024-01-01T00:00:00.000000Z",
+    "invocation_id": "test-snapshot-invocation",
+    "env": {},
+    "project_name": "jaffle_shop",
+    "project_id": "test-project"
+  },
+  "nodes": {
+    "snapshot.jaffle_shop.snap_orders": {
+      "resource_type": "snapshot",
+      "database": "dev",
+      "schema": "jaffle_shop",
+      "name": "snap_orders",
+      "alias": "snap_orders",
+      "package_name": "jaffle_shop",
+      "path": "snap_orders.sql",
+      "original_file_path": "snapshots/snap_orders.sql",
+      "unique_id": "snapshot.jaffle_shop.snap_orders",
+      "fqn": ["jaffle_shop", "snap_orders"],
+      "checksum": {"name": "sha256", "checksum": "abc123"},
+      "config": {
+        "enabled": true,
+        "alias": null,
+        "schema": null,
+        "database": null,
+        "tags": [],
+        "meta": {},
+        "group": null,
+        "materialized": "snapshot",
+        "incremental_strategy": null,
+        "batch_size": null,
+        "lookback": 1,
+        "begin": null,
+        "persist_docs": {},
+        "post-hook": [],
+        "pre-hook": [],
+        "quoting": {},
+        "column_types": {},
+        "full_refresh": null,
+        "unique_key": "order_id",
+        "on_schema_change": "ignore",
+        "on_configuration_change": "apply",
+        "grants": {},
+        "packages": [],
+        "docs": {"show": true, "node_color": null},
+        "contract": {"enforced": false, "alias_types": true},
+        "event_time": null,
+        "concurrent_batches": null,
+        "strategy": "timestamp",
+        "target_schema": "snapshots",
+        "target_database": null,
+        "updated_at": "updated_at",
+        "check_cols": null,
+        "snapshot_meta_column_names": null,
+        "dbt_valid_to_current": null
+      },
+      "tags": [],
+      "description": "Snapshot of orders table for SCD Type 2",
+      "columns": null,
+      "meta": {},
+      "group": null,
+      "docs": {"show": true, "node_color": null},
+      "patch_path": null,
+      "build_path": null,
+      "created_at": 1704067200.0,
+      "relation_name": "\"dev\".\"snapshots\".\"snap_orders\"",
+      "raw_code": "SELECT * FROM {{ ref('stg_orders') }}",
+      "compiled_code": "SELECT * FROM dev.jaffle_shop.stg_orders",
+      "language": "sql",
+      "depends_on": {
+        "macros": [],
+        "nodes": ["model.jaffle_shop.stg_orders"]
+      },
+      "refs": [{"name": "stg_orders", "package": null, "version": null}],
+      "sources": []
+    }
+  },
+  "sources": {},
+  "exposures": {},
+  "metrics": {},
+  "groups": {},
+  "selectors": {},
+  "disabled": {},
+  "docs": {},
+  "parent_map": {},
+  "child_map": {},
+  "group_map": {},
+  "saved_queries": {},
+  "semantic_models": {},
+  "unit_tests": {},
+  "macros": {}
+}