feat(adapter): fail loud on dbt v2/Fusion (v20) manifests

recce/adapter/dbt_adapter/__init__.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import os
+import re
 import time
 import uuid
 from contextlib import contextmanager
@@ -28,6 +29,7 @@
 from recce.exceptions import (
     DuckDBExternalAccessBlocked,
     RecceException,
+    UnsupportedDbtSchemaError,
     is_duckdb_external_access_blocked,
 )
 from recce.util.cll import CLLPerformanceTracking, cll, get_cll_cache
@@ -51,6 +53,11 @@
     print("Error: dbt module not found. Please install it by running:")
     print("pip install dbt-core dbt-<adapter>")
     raise e
+
+try:
+    from dbt.artifacts.exceptions import IncompatibleSchemaError
+except ImportError:  # dbt < 1.8 kept it under dbt.exceptions
+    from dbt.exceptions import IncompatibleSchemaError
 from watchdog.events import FileSystemEventHandler
 from watchdog.observers import Observer
 
@@ -229,13 +236,36 @@ def as_manifest(m: WritableManifest) -> Manifest:
         return result
 
 
+# Highest schema versions dbt 1.x ever emits; dbt v2 / Fusion jumps straight to
+# v20, so anything above these is a Fusion artifact. Do not tie the ceiling to
+# what the installed dbt is compatible with: under dbt 1.6 a v12 manifest is
+# already incompatible, but it comes from dbt 1.x and should surface dbt's own
+# version-mismatch error, not be reported as Fusion.
+_DBT1X_MAX_SCHEMA = {"manifest": 12, "catalog": 1}
+
+
+def _guard_unsupported_schema(artifact: str, found_version_url):
+    """Raise UnsupportedDbtSchemaError if `found_version_url` is a dbt v2 / Fusion
+    schema (above the dbt 1.x ceiling); no-op otherwise."""
+    # dbt_schema_version looks like "https://schemas.getdbt.com/dbt/manifest/v12.json"
+    match = re.search(r"/v(\d+)\.json", str(found_version_url))
+    found = int(match.group(1)) if match else None
+    if found is not None and found > _DBT1X_MAX_SCHEMA[artifact]:
+        raise UnsupportedDbtSchemaError(artifact, found)
+
+
 @track_timing(record_size=True)
 def load_manifest(path: str = None, data: dict = None):
     if path is not None:
         if not os.path.isfile(path):
             return None
-        return WritableManifest.read_and_check_versions(path)
+        try:
+            return WritableManifest.read_and_check_versions(path)
+        except IncompatibleSchemaError as e:
+            _guard_unsupported_schema("manifest", e.found)
+            raise
     if data is not None:
+        _guard_unsupported_schema("manifest", (data.get("metadata") or {}).get("dbt_schema_version"))
         return WritableManifest.upgrade_schema_version(data)
 
 
@@ -244,8 +274,13 @@ def load_catalog(path: str = None, data: dict = None):
     if path is not None:
         if not os.path.isfile(path):
             return None
-        return CatalogArtifact.read_and_check_versions(path)
+        try:
+            return CatalogArtifact.read_and_check_versions(path)
+        except IncompatibleSchemaError as e:
+            _guard_unsupported_schema("catalog", e.found)
+            raise
     if data is not None:
+        _guard_unsupported_schema("catalog", (data.get("metadata") or {}).get("dbt_schema_version"))
         return CatalogArtifact.upgrade_schema_version(data)
 
 

recce/exceptions.py

@@ -21,6 +21,20 @@ def __str__(self):
         return super().__str__()
 
 
+class UnsupportedDbtSchemaError(RecceException):
+    """Raised when a dbt artifact's schema is newer than the bundled dbt 1.x
+    supports, i.e. a dbt v2 / Fusion artifact. See _DBT1X_MAX_SCHEMA in the dbt
+    adapter for the version ceiling."""
+
+    def __init__(self, artifact: str, found_version: int):
+        message = (
+            f"dbt v2 / Fusion {artifact}s (schema v{found_version}) are not yet "
+            f"supported by Recce. Recce supports dbt 1.x artifacts (manifest schema up "
+            f"to v12, catalog v1). Re-generate the {artifact} with dbt 1.x."
+        )
+        super().__init__(message, is_raise=True)
+
+
 class DuckDBExternalAccessBlocked(RecceException):
     """Raised when DuckDB rejects a query because external access is disabled."""
 

tests/test_dbt.py

@@ -1,12 +1,23 @@
+import json
 import os
+import tempfile
 from unittest import TestCase
 from unittest.mock import MagicMock
 
+import pytest
+
 from recce.adapter.dbt_adapter import DbtAdapter, load_catalog, load_manifest
+from recce.exceptions import UnsupportedDbtSchemaError
 
 current_dir = os.path.dirname(os.path.abspath(__file__))
 
 
+def _fusion_artifact(kind: str) -> dict:
+    return {
+        "metadata": {"dbt_schema_version": f"https://schemas.getdbt.com/dbt/{kind}/v20.json"},
+    }
+
+
 class TestAdapterLineage(TestCase):
     def setUp(self) -> None:
         self.manifest = load_manifest(path=os.path.join(current_dir, "manifest.json"))
@@ -34,3 +45,71 @@ def test_load_lineage_with_catalog(self):
         lineage = dbt_adapter.get_lineage()
         assert lineage is not None
         assert len(lineage["nodes"]["model.jaffle_shop.orders"]["columns"]) == 9
+
+
+class TestFusionManifestFailLoud(TestCase):
+    """A v20 (dbt v2 / Fusion) artifact must fail loud with a Recce-branded message."""
+
+    def _assert_friendly(self, exc_info):
+        msg = str(exc_info.value)
+        assert "Fusion" in msg
+        assert "v20" in msg
+        assert "not yet supported" in msg
+
+    def test_load_manifest_v20_data_fails_loud(self):
+        with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
+            load_manifest(data=_fusion_artifact("manifest"))
+        self._assert_friendly(exc_info)
+
+    def test_load_manifest_v20_path_fails_loud(self):
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
+            json.dump(_fusion_artifact("manifest"), f)
+            path = f.name
+        try:
+            with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
+                load_manifest(path=path)
+            self._assert_friendly(exc_info)
+        finally:
+            os.unlink(path)
+
+    def test_load_catalog_v20_data_fails_loud(self):
+        with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
+            load_catalog(data=_fusion_artifact("catalog"))
+        self._assert_friendly(exc_info)
+
+    def test_load_catalog_v20_path_fails_loud(self):
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
+            json.dump(_fusion_artifact("catalog"), f)
+            path = f.name
+        try:
+            with pytest.raises(UnsupportedDbtSchemaError) as exc_info:
+                load_catalog(path=path)
+            self._assert_friendly(exc_info)
+        finally:
+            os.unlink(path)
+
+    def test_v12_not_flagged_as_fusion(self):
+        # A v12 manifest / v1 catalog is dbt 1.x, not Fusion — the guard must not fire,
+        # regardless of which dbt version Recce is running against.
+        from recce.adapter.dbt_adapter import _guard_unsupported_schema
+
+        _guard_unsupported_schema("manifest", "https://schemas.getdbt.com/dbt/manifest/v12.json")
+        _guard_unsupported_schema("catalog", "https://schemas.getdbt.com/dbt/catalog/v1.json")
+
+    def test_old_incompatible_artifact_keeps_dbt_error(self):
+        # v1 / v0 sit below the 1.x ceiling but dbt still rejects them as too old —
+        # the guard must stay silent, not mislabel them as Fusion.
+        from recce.adapter.dbt_adapter import IncompatibleSchemaError
+
+        for loader, kind, version in [(load_manifest, "manifest", 1), (load_catalog, "catalog", 0)]:
+            artifact = {
+                "metadata": {"dbt_schema_version": f"https://schemas.getdbt.com/dbt/{kind}/v{version}.json"},
+            }
+            with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
+                json.dump(artifact, f)
+                path = f.name
+            try:
+                with pytest.raises(IncompatibleSchemaError):
+                    loader(path=path)
+            finally:
+                os.unlink(path)