Support dbt 2.0 / Fusion in the edr CLI

.github/workflows/test-all-warehouses.yml

@@ -132,3 +132,36 @@ jobs:
       CI_SLACK_WEBHOOK: ${{ secrets.CI_SLACK_WEBHOOK }}
       CI_SLACK_TOKEN: ${{ secrets.CI_SLACK_TOKEN }}
       AWS_OIDC_ROLE_ARN: ${{ secrets.AWS_OIDC_ROLE_ARN }}
+
+  # dbt 2.0 (Fusion engine) targets, limited to Fusion-supported warehouses.
+  # 'fusion' installs the standalone binary from the 'dbt' PyPI package;
+  # a 2.x version installs the 'dbt-core' Python package.
+  # dbt 2.0 is still a prerelease; these jobs are separate from 'test' so they
+  # can be treated as informational (not required checks) while it stabilizes.
+  test-dbt2:
+    needs: [check-fork-status, approve-fork]
+    permissions:
+      contents: read
+      id-token: write
+    if: |
+      ! cancelled() &&
+      needs.check-fork-status.result == 'success' &&
+      needs.check-fork-status.outputs.should_skip != 'true' &&
+      (needs.check-fork-status.outputs.is_fork != 'true' || needs.approve-fork.result == 'success')
+    strategy:
+      fail-fast: false
+      matrix:
+        dbt-version: [fusion, 2.0.0b2]
+        warehouse-type: [snowflake, bigquery, databricks_catalog]
+    uses: ./.github/workflows/test-warehouse.yml
+    with:
+      warehouse-type: ${{ matrix.warehouse-type }}
+      elementary-ref: ${{ inputs.elementary-ref || ((github.event_name == 'pull_request_target' || github.event_name == 'pull_request') && github.event.pull_request.head.sha) || '' }}
+      dbt-data-reliability-ref: ${{ inputs.dbt-data-reliability-ref }}
+      dbt-version: ${{ matrix.dbt-version }}
+      generate-data: ${{ inputs.generate-data || false }}
+    secrets:
+      CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS }}
+      CI_SLACK_WEBHOOK: ${{ secrets.CI_SLACK_WEBHOOK }}
+      CI_SLACK_TOKEN: ${{ secrets.CI_SLACK_TOKEN }}
+      AWS_OIDC_ROLE_ARN: ${{ secrets.AWS_OIDC_ROLE_ARN }}

.github/workflows/test-warehouse.yml

@@ -32,7 +32,7 @@ on:
       dbt-version:
         type: string
         required: false
-        description: dbt's version to test with
+        description: dbt's version to test with ('fusion' for the dbt Fusion binary)
       generate-data:
         type: boolean
         required: false
@@ -80,6 +80,7 @@ env:
 jobs:
   test:
     runs-on: ubuntu-latest
+    timeout-minutes: 90
     permissions:
       contents: read
       id-token: write
@@ -121,11 +122,24 @@ jobs:
               ;;
           esac
 
-          if [ -n "$DBT_VERSION" ] && ! [[ "$DBT_VERSION" =~ ^[0-9]+(\.[0-9]+){1,2}([a-zA-Z0-9._+-]+)?$ ]]; then
+          if [ -n "$DBT_VERSION" ] && [ "$DBT_VERSION" != "fusion" ] && ! [[ "$DBT_VERSION" =~ ^[0-9]+(\.[0-9]+){1,2}([a-zA-Z0-9._+-]+)?$ ]]; then
             echo "Unsupported dbt version: $DBT_VERSION" >&2
             exit 1
           fi
 
+      - name: Remove dbt 1.x-only configs from E2E dbt project
+        if: inputs.dbt-version == 'fusion' || startsWith(inputs.dbt-version, '2.')
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          # +root_path is a dbt-dremio (1.x-only) config that dbt 2.0 rejects;
+          # Fusion doesn't support Dremio anyway.
+          sed -i '/+root_path: elementary/d' dbt_project.yml
+
+          # dbt 2.0 can't match an unrendered env_var package path against
+          # package-lock.yml, which breaks package context resolution (e.g.
+          # 'elementary' macros in run-operation), so render it upfront.
+          sed -i "s|{{ env_var('ELEMENTARY_DBT_PACKAGE_PATH') }}|$ELEMENTARY_DBT_PACKAGE_PATH|" packages.yml
+
       # ── Seed cache: compute key & restore volumes BEFORE starting services ──
       # This ensures Docker volumes are populated before containers initialize.
       - name: Compute seed cache key
@@ -214,7 +228,8 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v6
         with:
-          python-version: "3.10"
+          # dbt 2.x (the Fusion engine) requires Python >= 3.11.
+          python-version: ${{ (inputs.dbt-version == 'fusion' || startsWith(inputs.dbt-version, '2.')) && '3.11' || '3.10' }}
 
       - name: Install Spark requirements
         if: inputs.warehouse-type == 'spark'
@@ -231,6 +246,19 @@ jobs:
       - name: Install dbt
         if: inputs.warehouse-type != 'vertica'
         run: |
+          if [ "$DBT_VERSION" = "fusion" ]; then
+            # The Fusion binary is installed after Elementary (see 'Install dbt Fusion')
+            # so that dbt-core, pulled in as a dependency of elementary, can be removed first.
+            echo "Skipping dbt installation for the fusion target"
+            exit 0
+          fi
+
+          if [[ "$DBT_VERSION" == 2* ]]; then
+            # dbt-core 2.x is the Fusion engine with adapters built in.
+            pip install --pre "dbt-core==$DBT_VERSION"
+            exit 0
+          fi
+
           DBT_CORE_SPEC="dbt-core"
           DBT_ADAPTER="$WAREHOUSE_TYPE"
           DBT_ADAPTER_EXTRA=""
@@ -270,7 +298,9 @@ jobs:
           # For Vertica, dbt-vertica is already installed with --no-deps above;
           # using ".[vertica]" would re-resolve dbt-vertica's deps and downgrade
           # dbt-core to ~=1.8.  Install elementary without the adapter extra.
-          if [ "$WAREHOUSE_TYPE" = "vertica" ]; then
+          # For dbt 2.x / fusion, adapters are built into the engine, and the 1.x
+          # adapter extras would downgrade dbt-core.
+          if [ "$WAREHOUSE_TYPE" = "vertica" ] || [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; then
             pip install "."
           else
             EXTRA="$WAREHOUSE_TYPE"
@@ -280,6 +310,18 @@ jobs:
             pip install ".[$EXTRA]"
           fi
 
+      - name: Install dbt Fusion
+        if: inputs.dbt-version == 'fusion'
+        run: |
+          # Remove dbt-core (pulled in as a dependency of elementary) so the
+          # environment matches a binary-only Fusion installation, then install
+          # the Fusion binary from the 'dbt' package.
+          # Note: without --pre, 'pip install dbt' resolves to the unrelated legacy
+          # dbt Cloud CLI package, hence the explicit >=2 prerelease spec.
+          pip uninstall -y dbt-core
+          pip install --pre "dbt>=2.0.0rc1"
+          dbt --version
+
       - name: Write dbt profiles
         env:
           CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
@@ -308,6 +350,15 @@ jobs:
 
           echo "SCHEMA_NAME=$SCHEMA_NAME" >> "$GITHUB_ENV"
 
+          # The seeds schema is shared between jobs on the same warehouse, and
+          # the dbt 2.x jobs run in parallel with the dbt 1.x job on the same
+          # warehouses - give each dbt version its own seeds schema to avoid
+          # concurrent-write conflicts (e.g. Delta transaction conflicts).
+          if [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; then
+            SAFE_DBT_VERSION=$(echo -n "$DBT_VERSION" | sed 's/[^a-zA-Z0-9]/_/g')
+            echo "TEST_SEEDS_SCHEMA=test_seeds_$SAFE_DBT_VERSION" >> "$GITHUB_ENV"
+          fi
+
           python "${{ github.workspace }}/elementary/tests/profiles/generate_profiles.py" \
             --template "${{ github.workspace }}/elementary/tests/profiles/profiles.yml.j2" \
             --output ~/.dbt/profiles.yml \
@@ -377,22 +428,20 @@ jobs:
           fi
           dbt run --target "$WAREHOUSE_TYPE" "${EXTRA_ARGS[@]}" || true
 
-          # Validate run_results.json: only error_model should be non-success
-          jq -e '
-            [.results[] | select(.status != "success") | .unique_id]
-            | length == 1 and .[0] == "model.elementary_integration_tests.error_model"
-          ' target/run_results.json > /dev/null
-          jq_exit=$?
-
-          if [ $jq_exit -eq 0 ]; then
+          # Validate run_results.json: only error_model should fail (statuses like
+          # "skipped" or dbt 2.0's "no-op" for ephemeral models are not failures)
+          if jq -e '
+            [.results[] | select(.status == "error" or .status == "fail") | .unique_id]
+            == ["model.elementary_integration_tests.error_model"]
+          ' target/run_results.json > /dev/null; then
             echo "✅ Validation passed: only error_model failed."
           else
+            jq_exit=$?
             echo "❌ Validation failed. Unexpected failures:"
-            jq '[.results[] | select(.status != "success") | .unique_id] | join(", ")' target/run_results.json
+            jq '[.results[] | select(.status == "error" or .status == "fail") | .unique_id] | join(", ")' target/run_results.json
+            exit $jq_exit
           fi
 
-          exit $jq_exit
-
       - name: Test e2e dbt project
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
         continue-on-error: true

.pre-commit-config.yaml

@@ -73,9 +73,12 @@ repos:
         name: Verify internal dbt project package lock
         entry: |
           bash -c '
-            dbt_version=$(python -c "from dbt.version import __version__; print(__version__)");
-            required_version="1.8";
-            if [[ $(echo -e "$dbt_version\n$required_version" | sort -V | tail -1) == "$dbt_version" ]]; then
+            dbt_version=$(python -c "from importlib.metadata import version; print(version(\"dbt-core\"))" 2>/dev/null || echo "");
+            if [[ -z "$dbt_version" ]]; then
+              echo "Skipping hook, dbt-core is not installed.";
+            elif [[ "$dbt_version" == 2* ]]; then
+              echo "Skipping hook, dbt-core version is $dbt_version (>= 2.0).";
+            elif [[ $(echo -e "$dbt_version\n1.8" | sort -V | tail -1) == "$dbt_version" ]]; then
               dbt deps --lock --project-dir elementary/monitor/dbt_project && git diff --quiet elementary/monitor/dbt_project/package-lock.yml || (echo "Changes detected in package lock file!" && exit 1);
             else
               echo "Skipping hook, dbt version is $dbt_version (< 1.8).";

dev-requirements.txt

@@ -1,6 +1,9 @@
 pytest
 pytest-parametrization>=2022.2.1
 pre-commit
+# Used by tests/profiles/generate_profiles.py; with dbt 2.x / Fusion it is no
+# longer pulled in transitively by dbt-core.
+jinja2
 # Pinned below 1.16 because mypy >=1.16 crashes with INTERNAL ERROR on this codebase.
 mypy<1.16
 deepdiff

elementary/clients/dbt/command_line_dbt_runner.py

@@ -17,12 +17,20 @@
 from elementary.clients.dbt.dbt_log import parse_dbt_output
 from elementary.clients.dbt.transient_errors import is_transient_error
 from elementary.exceptions.exceptions import DbtCommandError, DbtLsCommandError
-from elementary.monitor.dbt_project_utils import is_dbt_package_up_to_date
+from elementary.monitor.dbt_project_utils import (
+    CLI_DBT_PROJECT_PATH,
+    is_dbt_package_up_to_date,
+)
 from elementary.utils.env_vars import is_debug
 from elementary.utils.log import get_logger
 
 logger = get_logger(__name__)
 
+# Directory for the internal dbt project's compiled SQL and run artifacts.
+# Translated to dbt's standard DBT_TARGET_PATH when running the internal project
+# (it can't be set via 'target-path' in dbt_project.yml, which dbt 2.0 rejects).
+EDR_INTERNAL_TARGET_PATH_ENV_VAR = "EDR_INTERNAL_TARGET_PATH"
+
 # Retry configuration for transient errors.
 _TRANSIENT_MAX_RETRIES = 3
 _TRANSIENT_WAIT_MULTIPLIER = 10  # seconds
@@ -86,12 +94,22 @@ def __init__(
         )
         self.adapter_type = self._get_adapter_type()
         self.raise_on_failure = raise_on_failure
-        self.env_vars = env_vars
+        self.env_vars = self._add_internal_target_path_env_var(env_vars)
         if force_dbt_deps:
             self.deps()
         elif run_deps_if_needed:
             self._run_deps_if_needed()
 
+    def _add_internal_target_path_env_var(
+        self, env_vars: Optional[Dict[str, str]]
+    ) -> Optional[Dict[str, str]]:
+        internal_target_path = os.getenv(EDR_INTERNAL_TARGET_PATH_ENV_VAR)
+        if internal_target_path and os.path.abspath(
+            self.project_dir
+        ) == os.path.abspath(CLI_DBT_PROJECT_PATH):
+            return {**(env_vars or {}), "DBT_TARGET_PATH": internal_target_path}
+        return env_vars
+
     def _get_adapter_type(self) -> Optional[str]:
         """Resolve the adapter type from ``profiles.yml``.
 

elementary/clients/dbt/dbt2_runner.py

@@ -0,0 +1,11 @@
+from elementary.clients.dbt.dbt_installation import get_dbt2_binary_path
+from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner
+
+
+class Dbt2Runner(SubprocessDbtRunner):
+    """Runner for dbt 2.0 (the Fusion engine), which is distributed as a
+    standalone binary (via the `dbt` PyPI package, the `dbt-core` 2.x package
+    or the standalone installer) and has no importable Python API."""
+
+    def _get_dbt_command_name(self) -> str:
+        return get_dbt2_binary_path()

elementary/clients/dbt/dbt_fusion_runner.py

@@ -1,14 +1,4 @@
-import os
+# Kept for backward compatibility; use Dbt2Runner instead.
+from elementary.clients.dbt.dbt2_runner import Dbt2Runner as DbtFusionRunner
 
-from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner
-
-DBT_FUSION_PATH = os.getenv("DBT_FUSION_PATH", "~/.local/bin/dbt")
-
-
-class DbtFusionRunner(SubprocessDbtRunner):
-    def _get_dbt_command_name(self) -> str:
-        return os.path.expanduser(DBT_FUSION_PATH)
-
-    def _run_deps_if_needed(self):
-        # Currently we don't support auto-updating deps for dbt fusion
-        return
+__all__ = ["DbtFusionRunner"]

elementary/clients/dbt/dbt_installation.py

@@ -0,0 +1,61 @@
+import os
+import shutil
+from importlib import metadata
+from typing import Optional
+
+from packaging import version
+
+DBT_FUSION_PATH_ENV_VAR = "DBT_FUSION_PATH"
+DEFAULT_DBT_FUSION_PATH = "~/.local/bin/dbt"
+
+
+def _get_package_version(package_name: str) -> Optional[version.Version]:
+    try:
+        return version.Version(metadata.version(package_name))
+    except (metadata.PackageNotFoundError, version.InvalidVersion):
+        return None
+
+
+def get_dbt_core_version() -> Optional[version.Version]:
+    """Version of the installed `dbt-core` package, or None if not installed."""
+    return _get_package_version("dbt-core")
+
+
+def get_dbt_package_version() -> Optional[version.Version]:
+    """Version of the installed `dbt` package, or None if not installed.
+
+    From 2.0, the `dbt` package on PyPI ships the dbt (Fusion) binary as a
+    platform wheel with no importable Python module.
+    """
+    return _get_package_version("dbt")
+
+
+def is_dbt2_binary_available() -> bool:
+    env_path = os.getenv(DBT_FUSION_PATH_ENV_VAR)
+    if env_path and os.path.exists(os.path.expanduser(env_path)):
+        return True
+
+    dbt_package_version = get_dbt_package_version()
+    if dbt_package_version is not None and dbt_package_version.major >= 2:
+        return True
+    return os.path.exists(os.path.expanduser(DEFAULT_DBT_FUSION_PATH))
+
+
+def get_dbt2_binary_path() -> str:
+    env_path = os.getenv(DBT_FUSION_PATH_ENV_VAR)
+    if env_path:
+        return os.path.expanduser(env_path)
+
+    # When only dbt-core 1.x is installed, the `dbt` executable on PATH is its
+    # entrypoint, so it can't be trusted to be the dbt 2.0 binary.
+    dbt_core_version = get_dbt_core_version()
+    dbt_package_version = get_dbt_package_version()
+    dbt2_installed_via_pip = (
+        dbt_package_version is not None and dbt_package_version.major >= 2
+    ) or (dbt_core_version is not None and dbt_core_version.major >= 2)
+    if dbt2_installed_via_pip or dbt_core_version is None:
+        which_path = shutil.which("dbt")
+        if which_path:
+            return which_path
+
+    return os.path.expanduser(DEFAULT_DBT_FUSION_PATH)