feat: widen dbt-core compatibility range (#5211)

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,61 @@ 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 uv
+        uses: astral-sh/setup-uv@v6
+      - name: Install SQLMesh dev dependencies
+        run: |
+          uv venv .venv
+          source .venv/bin/activate
+          sed -i 's/"pydantic>=2.0.0"/"pydantic"/g' pyproject.toml
+          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
+          UV=1 make install-dev
+          uv pip install pydantic>=2.0.0 --reinstall
+      - 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: Test SQLMesh info in sushi_dbt
+        working-directory: ./examples/sushi_dbt
+        run: |
+          source ../../.venv/bin/activate
+          sed -i 's/target: in_memory/target: postgres/g' profiles.yml
+          if [[ $(echo -e "${{ matrix.dbt-version }}\n1.5.0" | sort -V | head -n1) == "${{ matrix.dbt-version }}" ]] && [[ "${{ matrix.dbt-version }}" != "1.5.0" ]]; then
+            echo "DBT version is ${{ matrix.dbt-version }} (< 1.5.0), removing version parameters..."
+            sed -i -e 's/, version=1) }}/) }}/g' -e 's/, v=1) }}/) }}/g' models/top_waiters.sql
+          else
+            echo "DBT version is ${{ matrix.dbt-version }} (>= 1.5.0), keeping version parameters"
+          fi
+          sqlmesh info --skip-connection

Makefile

@@ -1,10 +1,16 @@
 .PHONY: docs
 
+ifdef UV
+    PIP := uv pip
+else
+    PIP := pip3
+endif
+
 install-dev:
-	pip3 install -e ".[dev,web,slack,dlt,lsp]" ./examples/custom_materializations
+	$(PIP) install -e ".[dev,web,slack,dlt,lsp]" ./examples/custom_materializations
 
 install-doc:
-	pip3 install -r ./docs/requirements.txt
+	$(PIP) install -r ./docs/requirements.txt
 
 install-pre-commit:
 	pre-commit install
@@ -22,16 +28,16 @@ doc-test:
 	python -m pytest --doctest-modules sqlmesh/core sqlmesh/utils
 
 package:
-	pip3 install build && python3 -m build
+	$(PIP) install build && python3 -m build
 
 publish: package
-	pip3 install twine && python3 -m twine upload dist/*
+	$(PIP) install twine && python3 -m twine upload dist/*
 
 package-tests:
-	pip3 install build && cp pyproject.toml tests/sqlmesh_pyproject.toml && python3 -m build tests/
+	$(PIP) install build && cp pyproject.toml tests/sqlmesh_pyproject.toml && python3 -m build tests/
 
 publish-tests: package-tests
-	pip3 install twine && python3 -m twine upload -r tobiko-private tests/dist/*
+	$(PIP) install twine && python3 -m twine upload -r tobiko-private tests/dist/*
 
 docs-serve:
 	mkdocs serve
@@ -93,6 +99,9 @@ engine-test:
 dbt-test:
 	pytest -n auto -m "dbt and not cicdonly"
 
+dbt-fast-test:
+	pytest -n auto -m "dbt and fast" --retries 3
+
 github-test:
 	pytest -n auto -m "github"
 
@@ -109,7 +118,7 @@ guard-%:
 	fi
 
 engine-%-install:
-	pip3 install -e ".[dev,web,slack,lsp,${*}]" ./examples/custom_materializations
+	$(PIP) install -e ".[dev,web,slack,lsp,${*}]" ./examples/custom_materializations
 
 engine-docker-%-up:
 	docker compose -f ./tests/core/engine_adapter/integration/docker/compose.${*}.yaml up -d
@@ -157,11 +166,11 @@ snowflake-test: guard-SNOWFLAKE_ACCOUNT guard-SNOWFLAKE_WAREHOUSE guard-SNOWFLAK
 	pytest -n auto -m "snowflake" --retries 3 --junitxml=test-results/junit-snowflake.xml
 
 bigquery-test: guard-BIGQUERY_KEYFILE engine-bigquery-install
-	pip install -e ".[bigframes]"
+	$(PIP) install -e ".[bigframes]"
 	pytest -n auto -m "bigquery" --retries 3 --junitxml=test-results/junit-bigquery.xml
 
 databricks-test: guard-DATABRICKS_CATALOG guard-DATABRICKS_SERVER_HOSTNAME guard-DATABRICKS_HTTP_PATH guard-DATABRICKS_ACCESS_TOKEN guard-DATABRICKS_CONNECT_VERSION engine-databricks-install
-	pip install 'databricks-connect==${DATABRICKS_CONNECT_VERSION}'
+	$(PIP) install 'databricks-connect==${DATABRICKS_CONNECT_VERSION}'
 	pytest -n auto -m "databricks" --retries 3 --junitxml=test-results/junit-databricks.xml
 
 redshift-test: guard-REDSHIFT_HOST guard-REDSHIFT_USER guard-REDSHIFT_PASSWORD guard-REDSHIFT_DATABASE engine-redshift-install

examples/sushi_dbt/profiles.yml

@@ -3,6 +3,14 @@ sushi:
     in_memory:
       type: duckdb
       schema: sushi
+    postgres:
+      type: postgres
+      host: "host"
+      user: "user"
+      password: "password"
+      dbname: "dbname"
+      port: 5432
+      schema: sushi
     duckdb:
       type: duckdb
       path: 'local.duckdb'

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

@@ -13,10 +13,17 @@
 
 from dbt import constants as dbt_constants, flags
 
+from sqlmesh.dbt.util import DBT_VERSION
 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 DBT_VERSION >= (1, 6, 0):
+    dbt_constants.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"  # type: ignore
+else:
+    from dbt.parser import manifest as dbt_manifest  # type: ignore
+
+    dbt_manifest.PARTIAL_PARSE_FILE_NAME = "sqlmesh_partial_parse.msgpack"  # type: ignore
 
 import jinja2
 from dbt.adapters.factory import register_adapter, reset_adapters
@@ -379,11 +386,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
@@ -599,6 +612,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
@@ -614,18 +630,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
@@ -95,31 +97,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",
+    )