feat: widen dbt-core compatibility range

Author: eakmanrq

.github/workflows/pr.yaml

@@ -8,6 +8,8 @@ on:
 concurrency:
   group: 'pr-${{ github.event.pull_request.number }}'
   cancel-in-progress: true
+permissions:
+  contents: read
 jobs:
   test-vscode:
     env:
@@ -66,3 +68,52 @@ jobs:
           name: playwright-report
           path: vscode/extension/playwright-report/
           retention-days: 30
+  test-dbt-versions:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        dbt-version: ["1.3.0", "1.4.0", "1.5.0", "1.6.0", "1.7.0", "1.8.0", "1.9.0", "1.10.0"]
+    steps:
+      - uses: actions/checkout@v5
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.10'
+      - name: Install SQLMesh dev dependencies
+        run: |
+          python -m venv .venv
+          source .venv/bin/activate
+          if [[ "${{ matrix.dbt-version }}" == "1.10.0" ]]; then
+            # For 1.10.0: only add version to dbt-core, remove versions from all adapter packages
+            sed -i -E 's/"(dbt-core)[^"]*"/"\1~=${{ matrix.dbt-version }}"/g' pyproject.toml
+            # Remove version constraints from all dbt adapter packages
+            sed -i -E 's/"(dbt-(bigquery|duckdb|snowflake|athena-community|clickhouse|databricks|redshift|trino))[^"]*"/"\1"/g' pyproject.toml
+          else
+            # For other versions: apply version to all dbt packages
+            sed -i -E 's/"(dbt-[^">=<~!]+)[^"]*"/"\1~=${{ matrix.dbt-version }}"/g' pyproject.toml
+          fi
+          make install-dev
+          pip install pydantic>=2.0.0 --force-reinstall
+          if [[ "${{ matrix.dbt-version }}" < "1.5.0" ]]; then
+            pip install jsonschema<4.17.0 --force-reinstall
+          fi
+      - name: Run dbt tests
+        # We can't run slow tests across all engines due to tests requiring DuckDB and old versions
+        # of DuckDB require a version of DuckDB we no longer support
+        run: |
+          source .venv/bin/activate
+          make dbt-fast-test
+      - name: Remove versions sections for dbt < 1.5.0
+        run: |
+          if [[ "${{ matrix.dbt-version }}" < "1.5.0" ]]; then
+            echo "DBT version is ${{ matrix.dbt-version }}, removing versions sections..."
+            sed -i '/^    versions:/,/^[^ ]/{ /^    versions:/d; /^      - v:/d; }' examples/sushi_dbt/models/schema.yml
+          else
+            echo "DBT version is ${{ matrix.dbt-version }}, keeping versions sections"
+          fi
+      - name: Test SQLMesh info in sushi_dbt
+        working-directory: ./examples/sushi_dbt
+        run: |
+          source ../../.venv/bin/activate
+          sqlmesh info --skip-connection

Makefile

@@ -93,6 +93,9 @@ engine-test:
 dbt-test:
 	pytest -n auto -m "dbt and not cicdonly"
 
+dbt-fast-test:
+	pytest -n auto -m "dbt and fast"
+
 github-test:
 	pytest -n auto -m "github"
 

pyproject.toml

@@ -52,7 +52,7 @@ bigframes = ["bigframes>=1.32.0"]
 clickhouse = ["clickhouse-connect"]
 databricks = ["databricks-sql-connector[pyarrow]"]
 dev = [
-    "agate==1.7.1",
+    "agate",
     "beautifulsoup4",
     "clickhouse-connect",
     "cryptography",

sqlmesh/dbt/loader.py

@@ -188,8 +188,11 @@ def _load_projects(self) -> t.List[Project]:
 
             self._projects.append(project)
 
-            if project.context.target.database != (self.context.default_catalog or ""):
-                raise ConfigError("Project default catalog does not match context default catalog")
+            context_default_catalog = self.context.default_catalog or ""
+            if project.context.target.database != context_default_catalog:
+                raise ConfigError(
+                    f"Project default catalog ('{project.context.target.database}') does not match context default catalog ('{context_default_catalog}')."
+                )
             for path in project.project_files:
                 self._track_file(path)
 

sqlmesh/dbt/manifest.py

@@ -15,7 +15,12 @@
 from sqlmesh.utils.conversions import make_serializable
 
 # Override the file name to prevent dbt commands from invalidating the cache.
-dbt_constants.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
+if hasattr(dbt_constants, "PARTIAL_PARSE_FILE_NAME"):
+    dbt_constants.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
+else:
+    from dbt.parser import manifest as dbt_manifest
+
+    dbt_manifest.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"
 
 import jinja2
 from dbt.adapters.factory import register_adapter, reset_adapters
@@ -378,11 +383,17 @@ def _load_on_run_start_end(self) -> None:
 
                 if "on-run-start" in node.tags:
                     self._on_run_start_per_package[node.package_name][node_name] = HookConfig(
-                        sql=sql, index=node.index or 0, path=node_path, dependencies=dependencies
+                        sql=sql,
+                        index=getattr(node, "index", None) or 0,
+                        path=node_path,
+                        dependencies=dependencies,
                     )
                 else:
                     self._on_run_end_per_package[node.package_name][node_name] = HookConfig(
-                        sql=sql, index=node.index or 0, path=node_path, dependencies=dependencies
+                        sql=sql,
+                        index=getattr(node, "index", None) or 0,
+                        path=node_path,
+                        dependencies=dependencies,
                     )
 
     @property
@@ -598,6 +609,9 @@ def _macro_references(
     manifest: Manifest, node: t.Union[ManifestNode, Macro]
 ) -> t.Set[MacroReference]:
     result: t.Set[MacroReference] = set()
+    if not hasattr(node, "depends_on"):
+        return result
+
     for macro_node_id in node.depends_on.macros:
         if not macro_node_id:
             continue
@@ -613,18 +627,20 @@ def _macro_references(
 
 def _refs(node: ManifestNode) -> t.Set[str]:
     if DBT_VERSION >= (1, 5, 0):
-        result = set()
+        result: t.Set[str] = set()
+        if not hasattr(node, "refs"):
+            return result
         for r in node.refs:
-            ref_name = f"{r.package}.{r.name}" if r.package else r.name
+            ref_name = f"{r.package}.{r.name}" if r.package else r.name  # type: ignore
             if getattr(r, "version", None):
-                ref_name = f"{ref_name}_v{r.version}"
+                ref_name = f"{ref_name}_v{r.version}"  # type: ignore
             result.add(ref_name)
         return result
     return {".".join(r) for r in node.refs}  # type: ignore
 
 
 def _sources(node: ManifestNode) -> t.Set[str]:
-    return {".".join(s) for s in node.sources}
+    return {".".join(s) for s in getattr(node, "sources", [])}
 
 
 def _model_node_id(model_name: str, package: str) -> str:

sqlmesh/dbt/relation.py

@@ -1,7 +1,7 @@
 from sqlmesh.dbt.util import DBT_VERSION
 
 
-if DBT_VERSION < (1, 8, 0):
-    from dbt.contracts.relation import *  # type: ignore  # noqa: F403
-else:
+if DBT_VERSION >= (1, 8, 0):
     from dbt.adapters.contracts.relation import *  # type: ignore # noqa: F403
+else:
+    from dbt.contracts.relation import *  # type: ignore  # noqa: F403

sqlmesh/dbt/seed.py

@@ -5,11 +5,13 @@
 
 import agate
 
-try:
+from sqlmesh.dbt.util import DBT_VERSION
+
+if DBT_VERSION >= (1, 8, 0):
     from dbt_common.clients import agate_helper  # type: ignore
 
     SUPPORTS_DELIMITER = True
-except ImportError:
+else:
     from dbt.clients import agate_helper  # type: ignore
 
     SUPPORTS_DELIMITER = False
@@ -90,31 +92,33 @@ def to_sqlmesh(
         )
 
 
-class Integer(agate_helper.Integer):
-    def cast(self, d: t.Any) -> t.Optional[int]:
-        if isinstance(d, str):
-            # The dbt's implementation doesn't support coercion of strings to integers.
-            if d.strip().lower() in self.null_values:
-                return None
-            try:
-                return int(d)
-            except ValueError:
-                raise agate.exceptions.CastError('Can not parse value "%s" as Integer.' % d)
-        return super().cast(d)
-
-    def jsonify(self, d: t.Any) -> str:
-        return d
-
-
-agate_helper.Integer = Integer  # type: ignore
-
-
 AGATE_TYPE_MAPPING = {
-    agate_helper.Integer: exp.DataType.build("int"),
     agate_helper.Number: exp.DataType.build("double"),
     agate_helper.ISODateTime: exp.DataType.build("datetime"),
     agate.Date: exp.DataType.build("date"),
     agate.DateTime: exp.DataType.build("datetime"),
     agate.Boolean: exp.DataType.build("boolean"),
     agate.Text: exp.DataType.build("text"),
 }
+
+
+if DBT_VERSION >= (1, 7, 0):
+
+    class Integer(agate_helper.Integer):
+        def cast(self, d: t.Any) -> t.Optional[int]:
+            if isinstance(d, str):
+                # The dbt's implementation doesn't support coercion of strings to integers.
+                if d.strip().lower() in self.null_values:
+                    return None
+                try:
+                    return int(d)
+                except ValueError:
+                    raise agate.exceptions.CastError('Can not parse value "%s" as Integer.' % d)
+            return super().cast(d)
+
+        def jsonify(self, d: t.Any) -> str:
+            return d
+
+    agate_helper.Integer = Integer  # type: ignore
+
+    AGATE_TYPE_MAPPING[agate_helper.Integer] = exp.DataType.build("int")

sqlmesh/dbt/util.py

@@ -20,10 +20,10 @@ def _get_dbt_version() -> t.Tuple[int, int, int]:
 
 DBT_VERSION = _get_dbt_version()
 
-if DBT_VERSION < (1, 8, 0):
-    from dbt.clients.agate_helper import table_from_data_flat, empty_table, as_matrix  # type: ignore  # noqa: F401
-else:
+if DBT_VERSION >= (1, 8, 0):
     from dbt_common.clients.agate_helper import table_from_data_flat, empty_table, as_matrix  # type: ignore  # noqa: F401
+else:
+    from dbt.clients.agate_helper import table_from_data_flat, empty_table, as_matrix  # type: ignore  # noqa: F401
 
 
 def pandas_to_agate(df: pd.DataFrame) -> agate.Table:

tests/dbt/conftest.py

@@ -7,6 +7,7 @@
 from sqlmesh.core.context import Context
 from sqlmesh.dbt.context import DbtContext
 from sqlmesh.dbt.project import Project
+from sqlmesh.dbt.target import PostgresConfig
 
 
 @pytest.fixture()
@@ -25,3 +26,16 @@ def render(value: str) -> str:
         return render
 
     return create_renderer
+
+
+@pytest.fixture()
+def dbt_dummy_postgres_config() -> PostgresConfig:
+    return PostgresConfig(  # type: ignore
+        name="postgres",
+        host="host",
+        user="user",
+        password="password",
+        dbname="dbname",
+        port=5432,
+        schema="schema",
+    )

tests/dbt/test_adapter.py

@@ -23,6 +23,7 @@
 pytestmark = pytest.mark.dbt
 
 
+@pytest.mark.slow
 def test_adapter_relation(sushi_test_project: Project, runtime_renderer: t.Callable):
     context = sushi_test_project.context
     assert context.target
@@ -96,6 +97,7 @@ def test_adapter_relation(sushi_test_project: Project, runtime_renderer: t.Calla
     assert engine_adapter.table_exists("foo.bar__backup")
 
 
+@pytest.mark.slow
 def test_bigquery_get_columns_in_relation(
     sushi_test_project: Project,
     runtime_renderer: t.Callable,
@@ -135,6 +137,7 @@ def test_bigquery_get_columns_in_relation(
 
 
 @pytest.mark.cicdonly
+@pytest.mark.slow
 def test_normalization(
     sushi_test_project: Project, runtime_renderer: t.Callable, mocker: MockerFixture
 ):
@@ -232,6 +235,7 @@ def test_normalization(
     adapter_mock.drop_table.assert_has_calls([call(relation_bla_bob)])
 
 
+@pytest.mark.slow
 def test_adapter_dispatch(sushi_test_project: Project, runtime_renderer: t.Callable):
     context = sushi_test_project.context
     renderer = runtime_renderer(context)
@@ -244,6 +248,7 @@ def test_adapter_dispatch(sushi_test_project: Project, runtime_renderer: t.Calla
 
 
 @pytest.mark.parametrize("project_dialect", ["duckdb", "bigquery"])
+@pytest.mark.slow
 def test_adapter_map_snapshot_tables(
     sushi_test_project: Project,
     runtime_renderer: t.Callable,
@@ -320,6 +325,7 @@ def test_quote_as_configured():
     adapter.quote_as_configured("foo", "database") == "foo"
 
 
+@pytest.mark.slow
 def test_adapter_get_relation_normalization(
     sushi_test_project: Project, runtime_renderer: t.Callable
 ):
@@ -352,6 +358,7 @@ def test_adapter_get_relation_normalization(
         )
 
 
+@pytest.mark.slow
 def test_adapter_expand_target_column_types(
     sushi_test_project: Project, runtime_renderer: t.Callable, mocker: MockerFixture
 ):