feat: add DuckDB, Trino, Dremio & Spark support to CI and CLI

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

@@ -96,6 +96,10 @@ jobs:
             databricks_catalog,
             athena,
             clickhouse,
+            duckdb,
+            trino,
+            dremio,
+            spark,
           ]
     uses: ./.github/workflows/test-warehouse.yml
     with:

.github/workflows/test-warehouse.yml

@@ -15,6 +15,9 @@ on:
           - spark
           - athena
           - clickhouse
+          - duckdb
+          - trino
+          - dremio
       elementary-ref:
         type: string
         required: false
@@ -83,6 +86,46 @@ jobs:
           path: dbt-data-reliability
           ref: ${{ inputs.dbt-data-reliability-ref }}
 
+      # ── Seed cache: compute key & restore volumes BEFORE starting services ──
+      # This ensures Docker volumes are populated before containers initialize.
+      - name: Compute seed cache key
+        id: seed-cache-key
+        if: inputs.warehouse-type == 'postgres' || inputs.warehouse-type == 'clickhouse' || inputs.warehouse-type == 'duckdb'
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          # Cache key is a hash of seed-related files so that cache busts when
+          # the data generation script, dbt project config, or seed schemas change.
+          SEED_HASH=$(
+            {
+              cat generate_data.py \
+                  dbt_project.yml \
+                  docker-compose.yml \
+                  ${{ github.workspace }}/elementary/tests/profiles/profiles.yml.j2
+              echo "dbt_version=${{ inputs.dbt-version || '' }}"
+            } | sha256sum | head -c 16
+          )
+          echo "seed-hash=$SEED_HASH" >> "$GITHUB_OUTPUT"
+
+      - name: Restore seed cache
+        id: seed-cache
+        if: steps.seed-cache-key.outputs.seed-hash
+        uses: actions/cache@v4
+        with:
+          path: /tmp/seed-cache-${{ inputs.warehouse-type }}
+          key: seed-${{ inputs.warehouse-type }}-${{ steps.seed-cache-key.outputs.seed-hash }}
+
+      - name: Restore cached seed data into Docker volumes
+        if: steps.seed-cache.outputs.cache-hit == 'true' && inputs.warehouse-type != 'duckdb'
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: bash ci/restore_seed_cache.sh "${{ inputs.warehouse-type }}"
+
+      - name: Restore cached DuckDB seed
+        if: steps.seed-cache.outputs.cache-hit == 'true' && inputs.warehouse-type == 'duckdb'
+        run: |
+          cp /tmp/seed-cache-duckdb/elementary_test.duckdb /tmp/elementary_test.duckdb
+          echo "DuckDB seed cache restored."
+
+      # ── Start warehouse services ──────────────────────────────────────────
       - name: Start Postgres
         if: inputs.warehouse-type == 'postgres'
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
@@ -93,20 +136,43 @@ jobs:
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
         run: docker compose up -d clickhouse
 
+      - name: Start Trino
+        if: inputs.warehouse-type == 'trino'
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          docker compose up -d --wait trino
+
+      - name: Start Dremio
+        if: inputs.warehouse-type == 'dremio'
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          # Start Dremio services in detached mode with healthchecks, then
+          # run the setup container separately.  Using --exit-code-from would
+          # imply --abort-on-container-exit, killing all services when the
+          # setup container finishes.
+          docker compose up -d --wait dremio dremio-minio nessie
+          docker compose run --rm dremio-setup
+
+      - name: Start Spark
+        if: inputs.warehouse-type == 'spark'
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          docker compose up -d --build --wait spark-thrift
+
       - name: Setup Python
         uses: actions/setup-python@v5
         with:
           python-version: "3.10"
 
       - name: Install Spark requirements
         if: inputs.warehouse-type == 'spark'
-        run: sudo apt-get install python-dev libsasl2-dev gcc
+        run: sudo apt-get install -y python3-dev libsasl2-dev gcc
 
       - name: Install dbt
         run: >
           pip install
           "dbt-core${{ inputs.dbt-version && format('=={0}', inputs.dbt-version) }}"
-          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || (inputs.warehouse-type == 'athena' && 'athena-community') || inputs.warehouse-type }}${{ inputs.dbt-version && format('~={0}', inputs.dbt-version) }}"
+          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || (inputs.warehouse-type == 'athena' && 'athena-community') || (inputs.warehouse-type == 'dremio' && 'dremio') || inputs.warehouse-type }}${{ (inputs.warehouse-type == 'spark' && '[PyHive]') || '' }}${{ inputs.dbt-version && format('~={0}', inputs.dbt-version) }}"
 
       - name: Install Elementary
         run: |
@@ -117,21 +183,29 @@ jobs:
         env:
           CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
         run: |
-          # Schema name = py_<YYMMDD_HHMMSS>_<branch≤19>_<8-char hash>
-          # The hash prevents collisions across concurrent jobs; the branch
-          # keeps it human-readable; the timestamp helps with stale schema
-          # cleanup and ensures each CI run gets a unique schema.
-          #
-          # Budget (PostgreSQL 63-char limit):
-          #   py_(3) + timestamp(13) + _(1) + branch(≤19) + _(1) + hash(8) = 45
-          #   + _elementary(11) + _gw7(4) = 60
-          CONCURRENCY_GROUP="tests_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}_${BRANCH_NAME}"
-          SHORT_HASH=$(echo -n "$CONCURRENCY_GROUP" | sha256sum | head -c 8)
-          SAFE_BRANCH=$(echo "${BRANCH_NAME}" | awk '{print tolower($0)}' | sed "s/[^a-z0-9]/_/g; s/__*/_/g" | head -c 19)
-          DATE_STAMP=$(date -u +%y%m%d_%H%M%S)
-          SCHEMA_NAME="py_${DATE_STAMP}_${SAFE_BRANCH}_${SHORT_HASH}"
-
-          echo "Schema name: $SCHEMA_NAME (branch='${BRANCH_NAME}', timestamp=${DATE_STAMP}, hash of concurrency group)"
+          # Docker-based adapters use ephemeral containers, so a fixed schema
+          # name is safe (the concurrency group prevents parallel collisions).
+          # This enables caching the seeded database state between runs.
+          IS_DOCKER=false
+          case "${{ inputs.warehouse-type }}" in
+            postgres|clickhouse|trino|dremio|duckdb|spark) IS_DOCKER=true ;;
+          esac
+
+          if [ "$IS_DOCKER" = "true" ]; then
+            SCHEMA_NAME="elementary_tests"
+            echo "Schema name: $SCHEMA_NAME (fixed for Docker adapter '${{ inputs.warehouse-type }}')"
+          else
+            # Cloud adapters: unique schema per run to avoid collisions.
+            # Schema name = py_<YYMMDD_HHMMSS>_<branch≤19>_<8-char hash>
+            CONCURRENCY_GROUP="tests_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}_${BRANCH_NAME}"
+            SHORT_HASH=$(echo -n "$CONCURRENCY_GROUP" | sha256sum | head -c 8)
+            SAFE_BRANCH=$(echo "${BRANCH_NAME}" | awk '{print tolower($0)}' | sed "s/[^a-z0-9]/_/g; s/__*/_/g" | head -c 19)
+            DATE_STAMP=$(date -u +%y%m%d_%H%M%S)
+            SCHEMA_NAME="py_${DATE_STAMP}_${SAFE_BRANCH}_${SHORT_HASH}"
+            echo "Schema name: $SCHEMA_NAME (branch='${BRANCH_NAME}', timestamp=${DATE_STAMP}, hash of concurrency group)"
+          fi
+
+          echo "SCHEMA_NAME=$SCHEMA_NAME" >> "$GITHUB_ENV"
 
           python "${{ github.workspace }}/elementary/tests/profiles/generate_profiles.py" \
             --template "${{ github.workspace }}/elementary/tests/profiles/profiles.yml.j2" \
@@ -160,17 +234,42 @@ jobs:
         run: |
           dbt deps
 
+      - name: Generate seed data
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        if: steps.seed-cache.outputs.cache-hit != 'true'
+        run: python generate_data.py
+
+      - name: Seed e2e dbt project (external)
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        if: steps.seed-cache.outputs.cache-hit != 'true' && (inputs.warehouse-type == 'dremio' || inputs.warehouse-type == 'spark')
+        run: python load_seeds_external.py "${{ inputs.warehouse-type }}" "$SCHEMA_NAME" data
+
       - name: Seed e2e dbt project
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
-        if: inputs.warehouse-type == 'postgres' || inputs.warehouse-type == 'clickhouse' || inputs.generate-data
+        if: steps.seed-cache.outputs.cache-hit != 'true' && inputs.warehouse-type != 'dremio' && inputs.warehouse-type != 'spark'
+        run: dbt seed -f --target "${{ inputs.warehouse-type }}"
+
+      - name: Save seed cache from Docker volumes
+        if: steps.seed-cache.outputs.cache-hit != 'true' && (inputs.warehouse-type == 'postgres' || inputs.warehouse-type == 'clickhouse')
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: bash ci/save_seed_cache.sh "${{ inputs.warehouse-type }}"
+
+      - name: Save DuckDB seed cache
+        if: steps.seed-cache.outputs.cache-hit != 'true' && inputs.warehouse-type == 'duckdb'
         run: |
-          python generate_data.py
-          dbt seed -f --target "${{ inputs.warehouse-type }}"
+          mkdir -p /tmp/seed-cache-duckdb
+          cp /tmp/elementary_test.duckdb /tmp/seed-cache-duckdb/elementary_test.duckdb
+          echo "DuckDB seed cache saved."
 
       - name: Run e2e dbt project
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
         run: |
-          dbt run --target "${{ inputs.warehouse-type }}" || true
+          # Dremio needs single-threaded execution to avoid Nessie catalog race conditions
+          EXTRA_ARGS=()
+          if [ "${{ inputs.warehouse-type }}" = "dremio" ]; then
+            EXTRA_ARGS+=(--threads 1)
+          fi
+          dbt run --target "${{ inputs.warehouse-type }}" "${EXTRA_ARGS[@]}" || true
 
           # Validate run_results.json: only error_model should be non-success
           jq -e '
@@ -192,7 +291,12 @@ jobs:
         working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
         continue-on-error: true
         run: |
-          dbt test --target "${{ inputs.warehouse-type }}"
+          # Dremio needs single-threaded execution to avoid Nessie catalog race conditions
+          EXTRA_ARGS=()
+          if [ "${{ inputs.warehouse-type }}" = "dremio" ]; then
+            EXTRA_ARGS+=(--threads 1 --exclude tag:ephemeral_model)
+          fi
+          dbt test --target "${{ inputs.warehouse-type }}" "${EXTRA_ARGS[@]}"
 
       - name: Run help
         run: edr --help

elementary/clients/dbt/transient_errors.py

@@ -100,6 +100,15 @@
         "connection timed out",
         "broken pipe",
     ),
+    "spark": (
+        "thrift transport is closed",
+        "could not connect to any thrift server",
+        "connection refused",
+    ),
+    "duckdb": (
+        # DuckDB runs in-process; transient errors are rare.
+        # Common patterns (connection reset, broken pipe) are in _COMMON.
+    ),
 }
 
 # Pre-computed union of all adapter-specific patterns for the fallback path

elementary/monitor/dbt_project/dbt_project.yml

@@ -28,9 +28,17 @@ clean-targets: # directories to be removed by `dbt clean`
 
 # Configuring models
 # Full documentation: https://docs.getdbt.com/docs/configuring-models
+dispatch:
+  - macro_namespace: elementary
+    search_order: ["elementary_cli", "elementary"]
+
 vars:
   edr_cli_run: true
 
+models:
+  elementary_cli:
+    +file_format: "{{ 'delta' if target.type == 'spark' else none }}"
+
 quoting:
   database: "{{ env_var('DATABASE_QUOTING', 'None') | as_native }}"
   schema: "{{ env_var('SCHEMA_QUOTING', 'None')  | as_native }}"

elementary/monitor/dbt_project/macros/alerts/population/model_alerts.sql

@@ -52,7 +52,7 @@
         select * from {{ ref('elementary', 'dbt_models') }}
     ),
 
-    snapshots as (
+    snapshots_data as (
         select * from {{ ref('elementary', 'dbt_snapshots') }}
     ),
 
@@ -71,7 +71,7 @@
     artifacts_meta as (
         select unique_id, meta from models
         union all
-        select unique_id, meta from snapshots
+        select unique_id, meta from snapshots_data
         union all
         select unique_id, meta from seeds
     ),

elementary/monitor/dbt_project/macros/alerts/population/source_freshness_alerts.sql

@@ -97,7 +97,7 @@
       {% if error_after_column_exists %}
         results.error_after,
         results.warn_after,
-        results.filter,
+        results.{{ elementary.escape_reserved_keywords('filter') }},
       {% endif %}
       results.error,
       sources.database_name,

elementary/monitor/dbt_project/macros/can_upload_source_freshness.sql

@@ -2,10 +2,10 @@
     {% set counter_query %}
         with invocations as (
             select invocation_id
-            from {{ ref("elementary", "dbt_source_freshness_results") }}
+            from {{ ref("dbt_source_freshness_results", package="elementary") }}
             where {{ elementary.edr_datediff(elementary.edr_cast_as_timestamp('generated_at'), elementary.edr_current_timestamp(), 'day') }} < {{ days_back }}
         )
-        select count(*) as count
+        select count(*) as {{ elementary.escape_reserved_keywords('count') }}
         from invocations
         where invocation_id = {{ elementary.edr_quote(invocation_id) }}
     {% endset %}

elementary/monitor/dbt_project/macros/get_adapter_type_and_unique_id.sql

@@ -17,3 +17,7 @@
 {% macro athena__get_adapter_unique_id() %}
     {{ return(target.s3_staging_dir) }}
 {% endmacro %}
+
+{% macro duckdb__get_adapter_unique_id() %}
+    {{ return(target.path) }}
+{% endmacro %}

elementary/monitor/dbt_project/macros/get_models_latest_invocation.sql

@@ -2,16 +2,17 @@
   {% set query %}
     with ordered_run_results as (
       select
-        *,
-        row_number() over (partition by unique_id order by run_results.generated_at desc) as row_number
-      from {{ ref("elementary", "dbt_run_results") }} run_results
-      join {{ ref("elementary", "dbt_models") }} using (unique_id)
+        run_results.unique_id,
+        run_results.invocation_id,
+        row_number() over (partition by run_results.unique_id order by run_results.generated_at desc) as {{ elementary.escape_reserved_keywords('row_number') }}
+      from {{ ref("dbt_run_results", package="elementary") }} run_results
+      join {{ ref("dbt_models", package="elementary") }} models on run_results.unique_id = models.unique_id
     ),
 
     latest_run_results as (
-      select *
+      select unique_id, invocation_id
       from ordered_run_results
-      where row_number = 1
+      where {{ elementary.escape_reserved_keywords('row_number') }} = 1
     )
 
     select unique_id, invocation_id from latest_run_results

elementary/monitor/dbt_project/macros/get_models_latest_invocations_data.sql

@@ -1,35 +1,35 @@
 {% macro get_models_latest_invocations_data() %}
-  {% set invocations_relation = ref("elementary", "dbt_invocations") %}
+  {% set invocations_relation = ref("dbt_invocations", package="elementary") %}
   {% set column_exists = elementary.column_exists_in_relation(invocations_relation, 'job_url') %}
 
   {% set query %}
     with ordered_run_results as (
       select
-        *,
-        row_number() over (partition by unique_id order by run_results.generated_at desc) as row_number
-      from {{ ref("elementary", "dbt_run_results") }} run_results
-      join {{ ref("elementary", "dbt_models") }} using (unique_id)
+        run_results.invocation_id,
+        row_number() over (partition by run_results.unique_id order by run_results.generated_at desc) as {{ elementary.escape_reserved_keywords('row_number') }}
+      from {{ ref("dbt_run_results", package="elementary") }} run_results
+      join {{ ref("dbt_models", package="elementary") }} models on run_results.unique_id = models.unique_id
     ),
 
     latest_models_invocations as (
       select distinct invocation_id
       from ordered_run_results
-      where row_number = 1
+      where {{ elementary.escape_reserved_keywords('row_number') }} = 1
     )
 
     select
-      invocation_id,
-      command,
-      selected,
-      full_refresh,
+      invocations.invocation_id,
+      invocations.command,
+      invocations.selected,
+      invocations.full_refresh,
       {% if column_exists %}
-        job_url,
+        invocations.job_url,
       {% endif %}
-      job_name,
-      job_id,
-      orchestrator
+      invocations.job_name,
+      invocations.job_id,
+      invocations.orchestrator
     from {{ invocations_relation }} invocations
-    join latest_models_invocations using (invocation_id)
+    join latest_models_invocations on invocations.invocation_id = latest_models_invocations.invocation_id
   {% endset %}
   {% set result = elementary.run_query(query) %}
   {% do return(elementary.agate_to_dicts(result)) %}

elementary/monitor/dbt_project/macros/get_result_rows_agate.sql

@@ -7,7 +7,7 @@
   select
     elementary_test_results_id,
     result_row
-  from {{ ref("elementary", "test_result_rows") }}
+  from {{ ref("test_result_rows", package="elementary") }}
   where {{ elementary.edr_datediff(elementary.edr_cast_as_timestamp('detected_at'), elementary.edr_current_timestamp(), 'day') }} < {{ days_back }}
   {% if valid_ids_query %}
     and elementary_test_results_id in ({{ valid_ids_query }})
@@ -25,7 +25,7 @@
   select
     elementary_test_results_id,
     result_row
-  from {{ ref("elementary", "test_result_rows") }}
+  from {{ ref("test_result_rows", package="elementary") }}
   where detected_at > {{ elementary.edr_timeadd('day', -1 * days_back, elementary.edr_current_timestamp()) }}
   {% if valid_ids_query %}
     and elementary_test_results_id in ({{ valid_ids_query }})