Merge branch 'master' into slack-rick-text-refactor

Author: haritamar

.github/workflows/assign-devin-prs.yml

@@ -0,0 +1,32 @@
+name: Auto-assign Devin PRs
+
+on:
+  pull_request:
+    types: [opened]
+
+jobs:
+  assign:
+    if: github.actor == 'devin-ai-integration[bot]'
+    runs-on: ubuntu-latest
+    permissions:
+      pull-requests: write
+      issues: write
+    steps:
+      - name: Extract and assign requesting user
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const body = context.payload.pull_request.body || '';
+            const match = body.match(/Requested by[:\s]*(?:@(\w[\w-]*)|[\w][\w\s]*\(@(\w[\w-]*)\))/);
+            const user = match?.[1] || match?.[2];
+            if (user) {
+              await github.rest.issues.addAssignees({
+                owner: context.repo.owner,
+                repo: context.repo.repo,
+                issue_number: context.issue.number,
+                assignees: [user]
+              });
+              console.log(`Assigned PR #${context.issue.number} to @${user}`);
+            } else {
+              console.log('Could not determine requesting user from PR description');
+            }

.github/workflows/cleanup-stale-schemas.yml

@@ -0,0 +1,85 @@
+name: Cleanup stale CI schemas
+
+on:
+  schedule:
+    # Daily at 03:00 UTC
+    - cron: "0 3 * * *"
+  workflow_dispatch:
+    inputs:
+      max-age-hours:
+        type: string
+        required: false
+        default: "24"
+        description: Drop schemas older than this many hours
+
+env:
+  # Re-use the dbt-data-reliability integration-test project so we get the
+  # cleanup macro (drop_stale_ci_schemas) without duplicating it.
+  TESTS_DIR: ${{ github.workspace }}/dbt-data-reliability/integration_tests
+
+jobs:
+  cleanup:
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        warehouse-type:
+          - snowflake
+          - bigquery
+          - redshift
+          - databricks_catalog
+          - athena
+    steps:
+      - name: Checkout dbt package
+        uses: actions/checkout@v4
+        with:
+          repository: elementary-data/dbt-data-reliability
+          path: dbt-data-reliability
+
+      - name: Setup Python
+        uses: actions/setup-python@v6
+        with:
+          python-version: "3.10"
+          cache: "pip"
+
+      - name: Install dbt
+        run: >
+          pip install
+          "dbt-core"
+          "dbt-${{ (matrix.warehouse-type == 'databricks_catalog' && 'databricks') || (matrix.warehouse-type == 'athena' && 'athena-community') || matrix.warehouse-type }}"
+
+      - name: Write dbt profiles
+        env:
+          CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
+        run: |
+          if [ -z "$CI_WAREHOUSE_SECRETS" ]; then
+            echo "::error::Missing required secret: CI_WAREHOUSE_SECRETS"
+            exit 1
+          fi
+          # The cleanup job doesn't create schemas, but generate_profiles.py
+          # requires --schema-name.  Use a dummy value.
+          python "${{ github.workspace }}/dbt-data-reliability/integration_tests/profiles/generate_profiles.py" \
+            --template "${{ github.workspace }}/dbt-data-reliability/integration_tests/profiles/profiles.yml.j2" \
+            --output ~/.dbt/profiles.yml \
+            --schema-name "cleanup_placeholder"
+
+      - name: Install dbt deps
+        working-directory: ${{ env.TESTS_DIR }}/dbt_project
+        run: dbt deps
+
+      - name: Symlink local elementary package
+        run: ln -sfn ${{ github.workspace }}/dbt-data-reliability ${{ env.TESTS_DIR }}/dbt_project/dbt_packages/elementary
+
+      - name: Drop stale CI schemas
+        working-directory: ${{ env.TESTS_DIR }}/dbt_project
+        env:
+          MAX_AGE_HOURS: ${{ inputs.max-age-hours || '24' }}
+        run: |
+          if ! [[ "$MAX_AGE_HOURS" =~ ^[0-9]+$ ]]; then
+            echo "::error::max-age-hours must be a non-negative integer"
+            exit 1
+          fi
+          ARGS=$(printf '{"prefixes":["py_"],"max_age_hours":%s}' "$MAX_AGE_HOURS")
+          dbt run-operation drop_stale_ci_schemas \
+            --args "$ARGS" \
+            -t "${{ matrix.warehouse-type }}"

.github/workflows/release.yml

@@ -9,7 +9,27 @@ env:
   IMAGE_NAME: ${{ github.repository }}
 
 jobs:
+  validate-packages-yml:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Elementary
+        uses: actions/checkout@v4
+
+      - name: Validate dbt-data-reliability is not a git hash reference
+        run: |
+          PACKAGES_FILE="./elementary/monitor/dbt_project/packages.yml"
+          if grep -q 'git: https://github.com/elementary-data/dbt-data-reliability.git' "$PACKAGES_FILE"; then
+            echo "::error::packages.yml contains a git hash reference for dbt-data-reliability. Releases must use a proper package version (e.g. 'package: elementary-data/elementary' with a 'version:' field). Please update packages.yml before releasing."
+            exit 1
+          fi
+          if ! grep -q 'package: elementary-data/elementary' "$PACKAGES_FILE"; then
+            echo "::error::packages.yml does not contain a proper package reference for elementary-data/elementary. Please update packages.yml before releasing."
+            exit 1
+          fi
+          echo "packages.yml validation passed - using proper package version reference."
+
   publish-to-pypi:
+    needs: validate-packages-yml
     runs-on: ubuntu-latest
     steps:
       - name: Checkout Elementary
@@ -39,6 +59,7 @@ jobs:
           password: ${{ secrets.PYPI_PASS }}
 
   build-and-push-docker-image:
+    needs: validate-packages-yml
     runs-on: ubuntu-latest
     permissions:
       contents: read
@@ -77,6 +98,7 @@ jobs:
           labels: ${{ steps.meta.outputs.labels }}
 
   merge-to-docs:
+    needs: validate-packages-yml
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4

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

@@ -5,13 +5,15 @@ on:
     branches: ["master"]
     paths:
       - elementary/**
+      - tests/**
       - .github/**
       - pyproject.toml
   # For fork PRs - requires approval before running (has access to secrets)
   pull_request_target:
     branches: ["master"]
     paths:
       - elementary/**
+      - tests/**
       - .github/**
       - pyproject.toml
   workflow_dispatch:
@@ -103,19 +105,3 @@ jobs:
       dbt-version: ${{ matrix.dbt-version }}
       generate-data: ${{ inputs.generate-data || false }}
     secrets: inherit
-
-  notify_failures:
-    name: Notify Slack
-    secrets: inherit
-    needs: [test]
-    if: |
-      always() &&
-      ! cancelled() &&
-      ! contains(needs.test.result, 'success') &&
-      ! contains(needs.test.result, 'cancelled') &&
-      github.event_name != 'pull_request_target'
-    uses: ./.github/workflows/notify_slack.yml
-    with:
-      result: "failure"
-      run_id: ${{ github.run_id }}
-      workflow_name: "Test all warehouse platforms"

.github/workflows/test-warehouse.yml

@@ -65,7 +65,8 @@ jobs:
       run:
         working-directory: elementary
     concurrency:
-      # This is what eventually defines the schema name in the data platform.
+      # Serialises runs for the same warehouse × dbt-version × branch.
+      # The schema name is derived from a hash of this group (see "Write dbt profiles").
       group: tests_${{ inputs.warehouse-type }}_dbt_${{ inputs.dbt-version }}_${{ github.head_ref || github.ref_name }}
       cancel-in-progress: true
     steps:
@@ -114,12 +115,28 @@ jobs:
 
       - name: Write dbt profiles
         env:
-          PROFILES_YML: ${{ secrets.CI_PROFILES_YML }}
+          CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }}
         run: |
-          mkdir -p ~/.dbt
-          DBT_VERSION=$(pip show dbt-core | grep -i version | awk '{print $2}' | sed 's/\.//g')
-          UNDERSCORED_REF_NAME=$(echo "${{ inputs.warehouse-type }}_dbt_${DBT_VERSION}_${BRANCH_NAME}" | awk '{print tolower($0)}' | head -c 40 | sed "s/[-\/]/_/g")
-          echo "$PROFILES_YML" | base64 -d | sed "s/<SCHEMA_NAME>/py_$UNDERSCORED_REF_NAME/g" > ~/.dbt/profiles.yml
+          # 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)"
+
+          python "${{ github.workspace }}/elementary/tests/profiles/generate_profiles.py" \
+            --template "${{ github.workspace }}/elementary/tests/profiles/profiles.yml.j2" \
+            --output ~/.dbt/profiles.yml \
+            --schema-name "$SCHEMA_NAME"
 
       - name: Run Python package unit tests
         run: pytest -vv tests/unit --warehouse-type ${{ inputs.warehouse-type }}
@@ -260,3 +277,10 @@ jobs:
 
       - name: Run Python package e2e tests
         run: pytest -vv tests/e2e --warehouse-type ${{ inputs.warehouse-type }}
+
+      - name: Drop test schemas
+        if: always()
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        continue-on-error: true
+        run: |
+          dbt run-operation elementary_integration_tests.drop_test_schemas --target "${{ inputs.warehouse-type }}"

elementary/cli/cli.py

@@ -24,9 +24,10 @@ def get_log_path(ctx):
         ctx_args = ctx.args
         target_path_flag = "--target-path"
         target_path = ctx_args[ctx_args.index(target_path_flag) + 1]
-    finally:
-        os.makedirs(os.path.abspath(target_path), exist_ok=True)
-        return os.path.join(target_path, "edr.log")
+    except (ValueError, IndexError):
+        pass
+    os.makedirs(os.path.abspath(target_path), exist_ok=True)
+    return os.path.join(target_path, "edr.log")
 
 
 def get_quiet_logs(ctx):

elementary/clients/dbt/api_dbt_runner.py

@@ -27,7 +27,6 @@ class APIDbtRunner(CommandLineDbtRunner):
     def _inner_run_command(
         self,
         dbt_command_args: List[str],
-        capture_output: bool,
         quiet: bool,
         log_output: bool,
         log_format: str,
@@ -50,15 +49,24 @@ def collect_dbt_command_logs(event):
             with with_chdir(self.project_dir):
                 res: dbtRunnerResult = dbt.invoke(dbt_command_args)
         output = "\n".join(dbt_logs) or None
+        # Surface the exception text so that transient-error detection in
+        # _inner_run_command_with_retries can match against it.  The dbt
+        # Python API doesn't write to stderr, so we repurpose that field
+        # for the exception string (analogous to how SubprocessDbtRunner
+        # captures subprocess stderr).
+        exception_text = str(res.exception) if res.exception else None
         if self.raise_on_failure and not res.success:
             raise DbtCommandError(
                 base_command_args=dbt_command_args,
-                err_msg=(str(res.exception) if res.exception else output),
+                err_msg=(exception_text or output),
                 logs=[DbtLog.from_log_line(log) for log in dbt_logs],
             )
 
         return APIDbtCommandResult(
-            success=res.success, output=output, stderr=None, result_obj=res
+            success=res.success,
+            output=output,
+            stderr=exception_text,
+            result_obj=res,
         )
 
     def _parse_ls_command_result(