Merge branch 'master' into fix/replace-double-quotes-with-single-quotes

Author: arbiv

.github/workflows/test-warehouse.yml

@@ -26,16 +26,6 @@ on:
         type: string
         required: false
         description: dbt's version to test with
-      should-run-tests:
-        type: boolean
-        required: false
-        default: true
-        description: Whether to run E2E tests
-      clear-tests:
-        type: boolean
-        required: false
-        default: true
-        description: Whether to clean test environment
       generate-data:
         type: boolean
         required: false
@@ -47,10 +37,6 @@ on:
       warehouse-type:
         type: string
         required: true
-      should-run-tests:
-        type: boolean
-        required: false
-        default: true
       elementary-ref:
         type: string
         required: false
@@ -60,19 +46,16 @@ on:
       dbt-version:
         type: string
         required: false
-      clear-tests:
-        type: boolean
-        required: false
-        default: true
       generate-data:
         type: boolean
         required: false
         default: false
 
 env:
   BRANCH_NAME: ${{ github.head_ref || github.ref_name }}
-  DBT_PKG_INTEG_TESTS_DIR: ${{ github.workspace }}/dbt-data-reliability/integration_tests/deprecated_tests
-  ELMENTARY_INTERNAL_DBT_PKG_DIR: ${{ github.workspace }}/elementary/elementary/monitor/dbt_project
+  ELEMENTARY_DBT_PACKAGE_PATH: ${{ github.workspace }}/dbt-data-reliability
+  CLI_INTERNAL_DBT_PKG_DIR: ${{ github.workspace }}/elementary/elementary/monitor/dbt_project
+  E2E_DBT_PROJECT_DIR: ${{ github.workspace }}/elementary/tests/e2e_dbt_project
 
 jobs:
   # PRs from forks require approval, specifically with the "pull_request_target" event as it contains repo secrets.
@@ -117,12 +100,12 @@ jobs:
 
       - name: Start Postgres
         if: inputs.warehouse-type == 'postgres'
-        working-directory: ${{ env.DBT_PKG_INTEG_TESTS_DIR }}
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
         run: docker compose up -d postgres
 
       # - name: Start Clickhouse
       #   if: inputs.warehouse-type == 'clickhouse'
-      #   working-directory: ${{ env.DBT_PKG_INTEG_TESTS_DIR }}
+      #   working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
       #   run: docker compose up -d clickhouse
 
       - name: Setup Python
@@ -138,8 +121,7 @@ jobs:
         run: >
           pip install
           "dbt-core${{ inputs.dbt-version && format('=={0}', inputs.dbt-version) }}"
-          # TODO: remove the <1.10.2 once we have a fix for https://github.com/elementary-data/elementary/issues/1931
-          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks<1.10.2,') || inputs.warehouse-type }}${{ inputs.dbt-version && format('~={0}', inputs.dbt-version) }}"
+          "dbt-${{ (inputs.warehouse-type == 'databricks_catalog' && 'databricks') || inputs.warehouse-type }}${{ inputs.dbt-version && format('~={0}', inputs.dbt-version) }}"
 
       - name: Install Elementary
         run: |
@@ -170,12 +152,46 @@ jobs:
           rm -rf "$DBT_PKGS_PATH/elementary"
           ln -vs "$GITHUB_WORKSPACE/dbt-data-reliability" "$DBT_PKGS_PATH/elementary"
 
-      - name: Run dbt package integration tests
-        if: github.event_name != 'workflow_dispatch' || inputs.should-run-tests
-        working-directory: ${{ env.DBT_PKG_INTEG_TESTS_DIR }}
+      - name: Run deps for E2E dbt project
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        env:
+          ELEMENTARY_DBT_PACKAGE_PATH: ${{ env.ELEMENTARY_DBT_PACKAGE_PATH }}
         run: |
           dbt deps
-          python run_e2e_tests.py -t "${{ inputs.warehouse-type }}" -g "${{ inputs.warehouse-type == 'postgres' || inputs.generate-data }}" --clear-tests "${{ inputs.clear-tests }}"
+
+      - name: Seed e2e dbt project
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        if: inputs.warehouse-type == 'postgres' || inputs.generate-data
+        run: |
+          python generate_data.py
+          dbt seed -f --target "${{ inputs.warehouse-type }}"
+
+      - name: Run e2e dbt project
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        run: |
+          dbt run --target "${{ inputs.warehouse-type }}" || 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
+            echo "✅ Validation passed: only error_model failed."
+          else
+            echo "❌ Validation failed. Unexpected failures:"
+            jq '[.results[] | select(.status != "success") | .unique_id] | join(", ")' target/run_results.json
+          fi
+
+          exit $jq_exit
+
+      - name: Test e2e dbt project
+        working-directory: ${{ env.E2E_DBT_PROJECT_DIR }}
+        continue-on-error: true
+        run: |
+          dbt test --target "${{ inputs.warehouse-type }}"
 
       - name: Run help
         run: edr --help
@@ -187,12 +203,12 @@ jobs:
           edr monitor
           -t "${{ inputs.warehouse-type }}"
           --group-by table
-          --project-dir "${{ env.DBT_PKG_INTEG_TESTS_DIR }}"
+          --project-dir "${{ env.E2E_DBT_PROJECT_DIR }}"
           --project-profile-target "${{ inputs.warehouse-type }}"
           --slack-webhook "$SLACK_WEBHOOK"
 
       - name: Validate alerts statuses were updated
-        working-directory: ${{ env.ELMENTARY_INTERNAL_DBT_PKG_DIR }}
+        working-directory: ${{ env.CLI_INTERNAL_DBT_PKG_DIR }}
         run: |
           dbt deps
           dbt run-operation validate_alert_statuses_are_updated -t "${{ inputs.warehouse-type }}"
@@ -201,7 +217,7 @@ jobs:
         run: >
           edr monitor report
           -t "${{ inputs.warehouse-type }}"
-          --project-dir "${{ env.DBT_PKG_INTEG_TESTS_DIR }}"
+          --project-dir "${{ env.E2E_DBT_PROJECT_DIR }}"
           --project-profile-target "${{ inputs.warehouse-type }}"
 
       - name: Set report artifact name
@@ -230,7 +246,7 @@ jobs:
         run: >
           edr monitor send-report
           -t "${{ inputs.warehouse-type }}"
-          --project-dir "${{ env.DBT_PKG_INTEG_TESTS_DIR }}"
+          --project-dir "${{ env.E2E_DBT_PROJECT_DIR }}"
           --project-profile-target "${{ inputs.warehouse-type }}"
           --slack-file-name "report_${{ inputs.warehouse-type }}_${{ env.BRANCH_NAME }}.html"
           --slack-token "$SLACK_TOKEN"
@@ -259,5 +275,4 @@ jobs:
           path: elementary/edr_target/edr.log
 
       - name: Run Python package e2e tests
-        if: github.event_name != 'workflow_dispatch' || inputs.should-run-tests
         run: pytest -vv tests/e2e --warehouse-type ${{ inputs.warehouse-type }}

.gitignore

@@ -96,4 +96,4 @@ venv/
 
 # elementary outputs
 edr_target/
-tests/tests_with_db/dbt_project/dbt_packages/
+**/dbt_packages/

docs/_snippets/quickstart-package-install.mdx

@@ -39,7 +39,7 @@ Some packages we recommend you check out: [dbt_utils](https://github.com/dbt-lab
     ```yml packages.yml
     packages:
       - package: elementary-data/elementary
-        version: 0.19.4
+        version: 0.20.0
         ## Docs: https://docs.elementary-data.com
     ```
   </Step>

elementary/clients/dbt/api_dbt_runner.py

@@ -57,7 +57,9 @@ def collect_dbt_command_logs(event):
                 logs=[DbtLog.from_log_line(log) for log in dbt_logs],
             )
 
-        return APIDbtCommandResult(success=res.success, output=output, result_obj=res)
+        return APIDbtCommandResult(
+            success=res.success, output=output, stderr=None, result_obj=res
+        )
 
     def _parse_ls_command_result(
         self, select: Optional[str], result: DbtCommandResult

elementary/clients/dbt/command_line_dbt_runner.py

@@ -25,6 +25,7 @@
 class DbtCommandResult:
     success: bool
     output: Optional[str]
+    stderr: Optional[str]
 
 
 class CommandLineDbtRunner(BaseDbtRunner):
@@ -190,16 +191,23 @@ def run_operation(
         log_pattern = (
             RAW_EDR_LOGS_PATTERN if return_raw_edr_logs else MACRO_RESULT_PATTERN
         )
-        if capture_output and result.output is not None:
-            for log in parse_dbt_output(result.output):
-                if log_errors and log.level == "error":
-                    logger.error(log.msg)
-                    continue
-
-                if log.msg:
-                    match = log_pattern.match(log.msg)
-                    if match:
-                        run_operation_results.append(match.group(1))
+        if capture_output:
+            if result.output is not None:
+                for log in parse_dbt_output(result.output):
+                    if log_errors and log.level == "error":
+                        logger.error(log.msg)
+                        continue
+
+                    if log.msg:
+                        match = log_pattern.match(log.msg)
+                        if match:
+                            run_operation_results.append(match.group(1))
+
+            if result.stderr is not None and log_errors:
+                for log in parse_dbt_output(result.stderr):
+                    if log.level == "error":
+                        logger.error(log.msg)
+                        continue
 
         return run_operation_results
 

elementary/clients/dbt/dbt_fusion_runner.py

@@ -0,0 +1,14 @@
+import os
+
+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

elementary/clients/dbt/factory.py

@@ -1,25 +1,21 @@
 import os
+from enum import Enum
 from typing import Any, Dict, Optional, Type
 
 from dbt.version import __version__ as dbt_version_string
 from packaging import version
 
 from elementary.clients.dbt.command_line_dbt_runner import CommandLineDbtRunner
+from elementary.clients.dbt.dbt_fusion_runner import DbtFusionRunner
+from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner
 
 DBT_VERSION = version.Version(dbt_version_string)
 
-RUNNER_CLASS: Type[CommandLineDbtRunner]
-if (
-    DBT_VERSION >= version.Version("1.5.0")
-    and os.getenv("DBT_RUNNER_METHOD") != "subprocess"
-):
-    from elementary.clients.dbt.api_dbt_runner import APIDbtRunner
 
-    RUNNER_CLASS = APIDbtRunner
-else:
-    from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner
-
-    RUNNER_CLASS = SubprocessDbtRunner
+class RunnerMethod(Enum):
+    SUBPROCESS = "subprocess"
+    API = "api"
+    FUSION = "fusion"
 
 
 def create_dbt_runner(
@@ -33,8 +29,11 @@ def create_dbt_runner(
     allow_macros_without_package_prefix: bool = False,
     run_deps_if_needed: bool = True,
     force_dbt_deps: bool = False,
+    runner_method: Optional[RunnerMethod] = None,
 ) -> CommandLineDbtRunner:
-    return RUNNER_CLASS(
+    runner_method = runner_method or get_dbt_runner_method()
+    runner_class = get_dbt_runner_class(runner_method)
+    return runner_class(
         project_dir=project_dir,
         profiles_dir=profiles_dir,
         target=target,
@@ -46,3 +45,27 @@ def create_dbt_runner(
         run_deps_if_needed=run_deps_if_needed,
         force_dbt_deps=force_dbt_deps,
     )
+
+
+def get_dbt_runner_method() -> RunnerMethod:
+    runner_method = os.getenv("DBT_RUNNER_METHOD")
+    if runner_method:
+        return RunnerMethod(runner_method)
+
+    if DBT_VERSION >= version.Version("1.5.0"):
+        return RunnerMethod.API
+    return RunnerMethod.SUBPROCESS
+
+
+def get_dbt_runner_class(runner_method: RunnerMethod) -> Type[CommandLineDbtRunner]:
+    if runner_method == RunnerMethod.API:
+        # Import it internally since it will fail if the dbt version is below 1.5.0
+        from elementary.clients.dbt.api_dbt_runner import APIDbtRunner
+
+        return APIDbtRunner
+    elif runner_method == RunnerMethod.SUBPROCESS:
+        return SubprocessDbtRunner
+    elif runner_method == RunnerMethod.FUSION:
+        return DbtFusionRunner
+    else:
+        raise ValueError(f"Invalid runner method: {runner_method}")

elementary/clients/dbt/subprocess_dbt_runner.py

@@ -26,16 +26,17 @@ def _inner_run_command(
     ) -> DbtCommandResult:
         try:
             result = subprocess.run(
-                ["dbt"] + dbt_command_args,
+                [self._get_dbt_command_name()] + dbt_command_args,
                 check=self.raise_on_failure,
                 capture_output=capture_output or quiet,
                 env=self._get_command_env(),
                 cwd=self.project_dir,
             )
             success = result.returncode == 0
             output = result.stdout.decode() if result.stdout else None
+            stderr = result.stderr.decode() if result.stderr else None
 
-            return DbtCommandResult(success=success, output=output)
+            return DbtCommandResult(success=success, output=output, stderr=stderr)
         except subprocess.CalledProcessError as err:
             logs = (
                 list(parse_dbt_output(err.output.decode(), log_format))
@@ -49,6 +50,9 @@ def _inner_run_command(
                 base_command_args=dbt_command_args, logs=logs, err=err
             )
 
+    def _get_dbt_command_name(self) -> str:
+        return "dbt"
+
     def _parse_ls_command_result(
         self, select: Optional[str], result: DbtCommandResult
     ) -> List[str]:

elementary/config/config.py

@@ -71,6 +71,7 @@ def __init__(
         azure_container_name: Optional[str] = None,
         report_url: Optional[str] = None,
         teams_webhook: Optional[str] = None,
+        maximum_columns_in_alert_samples: Optional[int] = None,
         env: str = DEFAULT_ENV,
         run_dbt_deps_if_needed: Optional[bool] = None,
         project_name: Optional[str] = None,
@@ -103,6 +104,12 @@ def __init__(
             False,
         )
 
+        self.maximum_columns_in_alert_samples = self._first_not_none(
+            maximum_columns_in_alert_samples,
+            config.get("maximum_columns_in_alert_samples"),
+            4,
+        )
+
         self.timezone = self._first_not_none(
             timezone,
             config.get("timezone"),

elementary/messages/formats/adaptive_cards.py

@@ -144,7 +144,8 @@ def format_table_block(block: TableBlock) -> Dict[str, Any]:
                         "items": [
                             {
                                 "type": "TextBlock",
-                                "text": str(cell),
+                                "text": str(cell) if cell is not None else "",
+                                "wrap": True,
                             }
                         ],
                     }
@@ -256,4 +257,5 @@ def format_adaptive_card(message: MessageBody, version: str = "1.5") -> Dict[str
         "type": "AdaptiveCard",
         "body": format_adaptive_card_body(message),
         "version": version,
+        "msteams": {"width": "Full"},
     }