diff --git a/.github/workflows/test-all-warehouses.yml b/.github/workflows/test-all-warehouses.yml index 4c1d04403..d672724d8 100644 --- a/.github/workflows/test-all-warehouses.yml +++ b/.github/workflows/test-all-warehouses.yml @@ -132,3 +132,43 @@ 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; + # '2.x' installs the latest 'dbt-core' 2.x Python package (including + # pre-releases); an explicit 2.x version pins that package version. + # 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.x] + warehouse-type: [snowflake, bigquery, databricks_catalog] + exclude: + # dbt-core 2.0.0b2 (the latest 2.x on PyPI) hangs on Snowflake even + # with threads: 1 (dbt-labs/dbt-fusion#410); the fusion target covers + # Snowflake until a dbt-core 2.x with a newer engine is released. + - dbt-version: 2.x + warehouse-type: snowflake + 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 }} diff --git a/.github/workflows/test-warehouse.yml b/.github/workflows/test-warehouse.yml index 0e34c0c2e..c059e0fe0 100644 --- a/.github/workflows/test-warehouse.yml +++ b/.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,31 @@ 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" != "2.x" ] && ! [[ "$DBT_VERSION" =~ ^[0-9]+(\.[0-9]+){1,2}([a-zA-Z0-9._+-]+)?$ ]]; then echo "Unsupported dbt version: $DBT_VERSION" >&2 exit 1 fi + # dbt 2.x / Fusion has adapters built into the engine and doesn't + # support third-party adapters such as dbt-vertica. + if [ "$WAREHOUSE_TYPE" = "vertica" ] && { [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; }; then + echo "dbt version '$DBT_VERSION' is not supported with the vertica warehouse" >&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 +235,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 +253,26 @@ 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. + if [ "$DBT_VERSION" = "2.x" ]; then + # Track the latest 2.x release, including pre-releases (which + # order below 2.0.0, hence the >=2.0.0a0 lower bound). + pip install --pre "dbt-core>=2.0.0a0,<3" + else + pip install --pre "dbt-core==$DBT_VERSION" + fi + dbt --version + exit 0 + fi + DBT_CORE_SPEC="dbt-core" DBT_ADAPTER="$WAREHOUSE_TYPE" DBT_ADAPTER_EXTRA="" @@ -270,7 +312,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 +324,20 @@ 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 a version pin, 'pip install dbt' resolves to the + # unrelated legacy dbt Cloud CLI package, hence the explicit 2.x spec. + pip uninstall -y dbt-core + # Pinned for reproducible CI runs (rc205, for example, generates + # broken BigQuery temp-table DDL: "Invalid timestamp: '{}'"). + pip install "dbt==2.0.0rc212" + dbt --version + - name: Write dbt profiles env: CI_WAREHOUSE_SECRETS: ${{ secrets.CI_WAREHOUSE_SECRETS || '' }} @@ -308,10 +366,28 @@ 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 + + # The dbt 2.x (Fusion) Snowflake driver can hang with concurrent + # connections (see dbt-labs/dbt-fusion#410); threads: 1 is the + # documented mitigation. + PROFILE_ARGS=() + if { [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; } && [ "$WAREHOUSE_TYPE" = "snowflake" ]; then + PROFILE_ARGS+=(--var snowflake_threads=1) + fi + 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" + --schema-name "$SCHEMA_NAME" \ + "${PROFILE_ARGS[@]}" - name: Run Python package unit tests run: pytest -vv tests/unit --warehouse-type "$WAREHOUSE_TYPE" @@ -375,24 +451,29 @@ jobs: if [ "$WAREHOUSE_TYPE" = "dremio" ]; then EXTRA_ARGS+=(--threads 1) fi + # The dbt 2.x Snowflake driver hangs with concurrent connections + # (dbt-labs/dbt-fusion#410); pass --threads 1 explicitly on top of + # the profile-level setting as an extra safeguard. + if { [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; } && + [ "$WAREHOUSE_TYPE" = "snowflake" ]; then + EXTRA_ARGS+=(--threads 1) + 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 @@ -402,6 +483,13 @@ jobs: if [ "$WAREHOUSE_TYPE" = "dremio" ]; then EXTRA_ARGS+=(--threads 1 --exclude tag:ephemeral_model) fi + # The dbt 2.x Snowflake driver hangs with concurrent connections + # (dbt-labs/dbt-fusion#410); pass --threads 1 explicitly on top of + # the profile-level setting as an extra safeguard. + if { [ "$DBT_VERSION" = "fusion" ] || [[ "$DBT_VERSION" == 2* ]]; } && + [ "$WAREHOUSE_TYPE" = "snowflake" ]; then + EXTRA_ARGS+=(--threads 1) + fi dbt test --target "$WAREHOUSE_TYPE" "${EXTRA_ARGS[@]}" - name: Run help diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 6937d09f2..e0e9af5e0 100644 --- a/.pre-commit-config.yaml +++ b/.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)."; diff --git a/dev-requirements.txt b/dev-requirements.txt index 89f6c709e..c7b57d966 100644 --- a/dev-requirements.txt +++ b/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 diff --git a/elementary/clients/dbt/command_line_dbt_runner.py b/elementary/clients/dbt/command_line_dbt_runner.py index a450e429a..fe626cf87 100644 --- a/elementary/clients/dbt/command_line_dbt_runner.py +++ b/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``. diff --git a/elementary/clients/dbt/dbt2_runner.py b/elementary/clients/dbt/dbt2_runner.py new file mode 100644 index 000000000..c2a9546bd --- /dev/null +++ b/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() diff --git a/elementary/clients/dbt/dbt_fusion_runner.py b/elementary/clients/dbt/dbt_fusion_runner.py index 408054cc1..cdd87aa57 100644 --- a/elementary/clients/dbt/dbt_fusion_runner.py +++ b/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"] diff --git a/elementary/clients/dbt/dbt_installation.py b/elementary/clients/dbt/dbt_installation.py new file mode 100644 index 000000000..084f7bc1b --- /dev/null +++ b/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) diff --git a/elementary/clients/dbt/factory.py b/elementary/clients/dbt/factory.py index e03bc9bb0..3b71f64a7 100644 --- a/elementary/clients/dbt/factory.py +++ b/elementary/clients/dbt/factory.py @@ -2,19 +2,22 @@ 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.dbt2_runner import Dbt2Runner +from elementary.clients.dbt.dbt_installation import ( + get_dbt_core_version, + is_dbt2_binary_available, +) from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner -DBT_VERSION = version.Version(dbt_version_string) - class RunnerMethod(Enum): SUBPROCESS = "subprocess" API = "api" + DBT2 = "dbt2" + # Legacy alias for DBT2 (dbt 2.0 is the Fusion engine). FUSION = "fusion" @@ -52,20 +55,30 @@ def get_dbt_runner_method() -> RunnerMethod: if runner_method: return RunnerMethod(runner_method) - if DBT_VERSION >= version.Version("1.5.0"): - return RunnerMethod.API + dbt_core_version = get_dbt_core_version() + if dbt_core_version is not None: + if dbt_core_version.major >= 2: + return RunnerMethod.DBT2 + if dbt_core_version >= version.Version("1.5.0"): + return RunnerMethod.API + return RunnerMethod.SUBPROCESS + + if is_dbt2_binary_available(): + return RunnerMethod.DBT2 + 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 + # Import it internally since it will fail if dbt-core is not installed + # or its 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 + elif runner_method in (RunnerMethod.DBT2, RunnerMethod.FUSION): + return Dbt2Runner else: raise ValueError(f"Invalid runner method: {runner_method}") diff --git a/elementary/monitor/dbt_project/dbt_project.yml b/elementary/monitor/dbt_project/dbt_project.yml index 8c35fc852..dcf554f9f 100644 --- a/elementary/monitor/dbt_project/dbt_project.yml +++ b/elementary/monitor/dbt_project/dbt_project.yml @@ -20,7 +20,9 @@ snapshot-paths: ["snapshots"] packages-install-path: "{{ env_var('DBT_PACKAGES_FOLDER', 'dbt_packages') }}" -target-path: "{{ env_var('EDR_INTERNAL_TARGET_PATH', 'target') }}" # directory which will store compiled SQL files +# NOTE: 'target-path' is intentionally not set here since dbt 2.0 (Fusion) rejects it. +# The EDR_INTERNAL_TARGET_PATH env var is instead translated to DBT_TARGET_PATH by the CLI +# when running this project. clean-targets: # directories to be removed by `dbt clean` - "{{ env_var('EDR_INTERNAL_TARGET_PATH', 'target') }}" - "{{ env_var('DBT_PACKAGES_FOLDER', 'dbt_packages') }}" diff --git a/pyproject.toml b/pyproject.toml index 4d907c393..f7733132a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,9 @@ packages = [{include = "elementary"}] [tool.poetry.dependencies] python = ">=3.10,<3.14" click = ">=7.0,<9.0" -dbt-core = ">=1.8,<2.0.0" +# From 2.0, dbt-core is the Fusion engine (requires Python >=3.11); adapters are +# built into the engine, so the adapter extras below are only relevant for 1.x. +dbt-core = ">=1.8,<3.0.0" requests = ">=2.28.1,<3.0.0" urllib3 = ">=2.7.0" # transitive dependency via requests, which caps the major itself idna = ">=3.15" # transitive dependency via requests, floored to address CVE-2026-45409 (GHSA-65pc-fj4g-8rjx) ReDoS in idna.encode() diff --git a/tests/e2e_dbt_project/dbt_project.yml b/tests/e2e_dbt_project/dbt_project.yml index eae2d060a..c0b4689b9 100644 --- a/tests/e2e_dbt_project/dbt_project.yml +++ b/tests/e2e_dbt_project/dbt_project.yml @@ -10,7 +10,6 @@ seed-paths: ["data"] macro-paths: ["macros"] snapshot-paths: ["snapshots"] -target-path: "target" # directory which will store compiled SQL files clean-targets: # directories to be removed by `dbt clean` - "target" - "dbt_packages" @@ -19,12 +18,15 @@ clean-targets: # directories to be removed by `dbt clean` vars: days_back: 30 debug_logs: "{{ env_var('DBT_EDR_DEBUG', False) }}" - custom_run_started_at: "{{ modules.datetime.datetime.utcfromtimestamp(0) }}" + custom_run_started_at: "1970-01-01 00:00:00" clean_elementary_temp_tables: false disable_dbt_artifacts_autoupload: true seeds: - +schema: test_seeds + # The seeds schema is shared between CI jobs on the same warehouse (the + # generate_schema_name macro doesn't prefix it with the target schema), so + # jobs that may run concurrently must use a distinct name. + +schema: "{{ env_var('TEST_SEEDS_SCHEMA', 'test_seeds') }}" models: elementary_integration_tests: @@ -32,5 +34,12 @@ models: elementary: +schema: elementary - +root_path: elementary +file_format: "{{ 'delta' if target.type == 'spark' else none }}" + # dbt-dremio config; rejected by dbt 2.0 (Fusion), which doesn't support + # Dremio - the CI strips it for dbt 2.x targets. + +root_path: elementary +# This flag and the 'arguments:' syntax in the schema files require +# dbt-core >= 1.10.5 (or dbt 2.x), so that's the minimum for running the E2E +# project (the elementary package itself still supports dbt-core >= 1.8). +flags: + require_generic_test_arguments_property: true diff --git a/tests/e2e_dbt_project/external_seeders/spark.py b/tests/e2e_dbt_project/external_seeders/spark.py index 3d9be0e72..a139aded2 100644 --- a/tests/e2e_dbt_project/external_seeders/spark.py +++ b/tests/e2e_dbt_project/external_seeders/spark.py @@ -10,10 +10,11 @@ class SparkExternalSeeder(ExternalSeeder): """Load seeds into Spark via PyHive from CSV files mounted in the container.""" - # dbt_project.yml sets ``+schema: test_seeds`` for seeds and the default - # ``generate_schema_name`` macro returns that verbatim, so the actual seed - # schema is always ``test_seeds`` regardless of the target schema name. - SEED_SCHEMA = "test_seeds" + # dbt_project.yml sets ``+schema`` for seeds (TEST_SEEDS_SCHEMA env var, + # defaulting to ``test_seeds``) and the ``generate_schema_name`` macro + # returns that verbatim, so the actual seed schema matches it regardless + # of the target schema name. + DEFAULT_SEED_SCHEMA = "test_seeds" @staticmethod def _q(name: str) -> str: @@ -23,7 +24,7 @@ def _q(name: str) -> str: def load(self) -> None: failures: list[str] = [] q = self._q - seed_schema = self.SEED_SCHEMA + seed_schema = os.environ.get("TEST_SEEDS_SCHEMA", self.DEFAULT_SEED_SCHEMA) print( f"\n=== Loading Spark seeds via external CSV tables " f"(schema={seed_schema}) ===" @@ -39,14 +40,14 @@ def load(self) -> None: host = os.environ.get("SPARK_HOST", "127.0.0.1") port = int(os.environ.get("SPARK_PORT", "10000")) - print(f"Connecting to Spark Thrift at {host}:{port}...") + print(f"Connecting to Spark Thrift at {host}:{port}...") # noqa: E231 conn = None cursor = None try: conn = hive.Connection(host=host, port=port, username="dbt") cursor = conn.cursor() print(f"Creating schema '{seed_schema}'...") - cursor.execute(f"CREATE DATABASE IF NOT EXISTS `{seed_schema}`") + cursor.execute(f"CREATE DATABASE IF NOT EXISTS {q(seed_schema)}") for subdir, csv_path, table_name in self.iter_seed_csvs(): fname = os.path.basename(csv_path) diff --git a/tests/e2e_dbt_project/load_seeds_external.py b/tests/e2e_dbt_project/load_seeds_external.py index b4fe28068..567d12f34 100644 --- a/tests/e2e_dbt_project/load_seeds_external.py +++ b/tests/e2e_dbt_project/load_seeds_external.py @@ -37,10 +37,10 @@ def main(adapter: str, schema_name: str, data_dir: str) -> None: \b ADAPTER Target warehouse adapter (dremio | spark). SCHEMA_NAME Target schema / namespace for the seed tables. - NOTE: Spark ignores this value and always uses the fixed - schema defined in SparkExternalSeeder.SEED_SCHEMA (currently - "test_seeds") because the generate_schema_name macro returns - that name verbatim. + NOTE: Spark ignores this value and uses the seeds schema + (the TEST_SEEDS_SCHEMA env var, defaulting to "test_seeds") + because the generate_schema_name macro returns that name + verbatim. DATA_DIR Path to the directory containing training/ and validation/ CSVs. """ seeder_cls = SEEDERS[adapter] diff --git a/tests/e2e_dbt_project/macros/generic_tests/test_uniques.sql b/tests/e2e_dbt_project/macros/generic_tests/test_uniques.sql new file mode 100644 index 000000000..a23282398 --- /dev/null +++ b/tests/e2e_dbt_project/macros/generic_tests/test_uniques.sql @@ -0,0 +1,4 @@ +{# A test that always errors at runtime (used on error_model's missing column). #} +{%- test uniques(model, column_name) -%} + select {{ column_name }} from {{ model }} group by {{ column_name }} having count(*) > 1 +{%- endtest -%} diff --git a/tests/e2e_dbt_project/models/schema.yml b/tests/e2e_dbt_project/models/schema.yml index c1eeb09b6..27ffda9e7 100644 --- a/tests/e2e_dbt_project/models/schema.yml +++ b/tests/e2e_dbt_project/models/schema.yml @@ -4,20 +4,19 @@ models: - name: one config: tags: "{{ var('one_tags', []) }}" - meta: - owner: "{{ var('one_owner', none) }}" + meta: + owner: "{{ var('one_owner', none) }}" columns: - name: col1 tests: - accepted_values: - meta: - owner: "@elon" - values: [2, 3] + config: + meta: + owner: "@elon" + arguments: + values: [2, 3] - name: any_type_column_anomalies - meta: - owner: ["@edr"] - subscribers: "@egk" description: > This is a very weird description with breaklines @@ -25,324 +24,420 @@ models: and even a string like this 'wow'. You know, these $##$34#@#!^ can also be helpful WDYT? config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at + owner: ["@edr"] + subscribers: "@egk" tests: - elementary.volume_anomalies: - time_bucket: - period: hour - count: 4 - meta: - description: > - This is a very weird description - with breaklines - and comma, - and even a string like this 'wow'. You know, these $##$34#@#!^ can also be helpful - WDYT? config: severity: warn - tags: ["table_anomalies"] + meta: + description: > + This is a very weird description + with breaklines + and comma, + and even a string like this 'wow'. You know, these $##$34#@#!^ can also be helpful + WDYT? + tags: ["table_anomalies"] + arguments: + time_bucket: + period: hour + count: 4 - elementary.volume_anomalies: - time_bucket: - period: week - count: 1 config: severity: warn - where: 1=1 - tags: ["table_anomalies"] + where: 1=1 + tags: ["table_anomalies"] + arguments: + time_bucket: + period: week + count: 1 - elementary.all_columns_anomalies: - tags: ["all_any_type_columns_anomalies", "column_anomalies"] - #This here is to simulate a long test name as test params are part of the test name - exclude_regexp: ".*column1|column2|column3|column4|column5|column6|column7|column8|column9|column10|column11|column12|column13|column14|column15|column16|column17.*" + config: + tags: ["all_any_type_columns_anomalies", "column_anomalies"] + arguments: + exclude_regexp: ".*column1|column2|column3|column4|column5|column6|column7|column8|column9|column10|column11|column12|column13|column14|column15|column16|column17.*" - generic_test_on_model: - tags: ["regular_tests"] + config: + tags: ["regular_tests"] - elementary.all_columns_anomalies: - anomaly_direction: "drop" - where: 1=1 - tags: ["directional_anomalies", "drop"] + config: + where: 1=1 + tags: ["directional_anomalies", "drop"] + arguments: + anomaly_direction: "drop" - elementary.all_columns_anomalies: - anomaly_direction: "spike" - tags: ["directional_anomalies", "spike"] + config: + tags: ["directional_anomalies", "spike"] + arguments: + anomaly_direction: "spike" - name: no_timestamp_anomalies - meta: - owner: "elon@elementary-data.com, or@elementary-data.com" - subscribers: ["elon@elementary-data.com"] - description: This is a description. description: We use this model to test anomalies when there is no timestamp column tests: - elementary.volume_anomalies: - tags: ["no_timestamp"] + config: + tags: ["no_timestamp"] columns: - name: "null_count_str" tests: - elementary.column_anomalies: - tags: ["no_timestamp"] - where: 1=1 - column_anomalies: - - null_count + config: + tags: ["no_timestamp"] + where: 1=1 + arguments: + column_anomalies: + - null_count + config: + meta: + owner: "elon@elementary-data.com, or@elementary-data.com" + subscribers: ["elon@elementary-data.com"] + description: This is a description. - name: dimension_anomalies - meta: - owner: "egk" - subscribers: "elon, egk" description: We use this model to test dimension anomalies tests: - elementary.dimension_anomalies: - tags: ["dimension_anomalies", "should_fail"] - alias: "dimension_anomalies_platform" - timestamp_column: updated_at - where: 1=1 - dimensions: - - platform + config: + tags: ["dimension_anomalies", "should_fail"] + alias: "dimension_anomalies_platform" + where: 1=1 + arguments: + timestamp_column: updated_at + dimensions: + - platform - elementary.dimension_anomalies: - alias: "dimension_anomalies_platform_where_expression" - tags: ["dimension_anomalies"] - timestamp_column: updated_at - dimensions: - - platform - where_expression: "platform = 'android'" + config: + alias: "dimension_anomalies_platform_where_expression" + tags: ["dimension_anomalies"] + arguments: + timestamp_column: updated_at + dimensions: + - platform + where_expression: "platform = 'android'" - elementary.dimension_anomalies: - alias: "dimension_anomalies_platform_new_dimension" - tags: ["dimension_anomalies"] - timestamp_column: updated_at - dimensions: - - platform - where_expression: "platform = 'windows'" + config: + alias: "dimension_anomalies_platform_new_dimension" + tags: ["dimension_anomalies"] + arguments: + timestamp_column: updated_at + dimensions: + - platform + where_expression: "platform = 'windows'" - elementary.dimension_anomalies: - alias: "dimension_anomalies_platform_new_dimension_no_timestamp" - tags: ["dimension_anomalies"] - dimensions: - - platform - where_expression: "platform = 'windows'" + config: + alias: "dimension_anomalies_platform_new_dimension_no_timestamp" + tags: ["dimension_anomalies"] + arguments: + dimensions: + - platform + where_expression: "platform = 'windows'" - elementary.dimension_anomalies: - tags: ["dimension_anomalies", "should_fail"] - alias: "dimension_anomalies_platform_version" - timestamp_column: updated_at - dimensions: - - platform - - version + config: + tags: ["dimension_anomalies", "should_fail"] + alias: "dimension_anomalies_platform_version" + arguments: + timestamp_column: updated_at + dimensions: + - platform + - version - elementary.dimension_anomalies: - anomaly_direction: "spike" - tags: ["directional_anomalies", "spike"] - timestamp_column: updated_at - dimensions: - - platform + config: + tags: ["directional_anomalies", "spike"] + arguments: + anomaly_direction: "spike" + timestamp_column: updated_at + dimensions: + - platform - elementary.dimension_anomalies: - anomaly_direction: "drop" - tags: ["directional_anomalies", "drop"] - timestamp_column: updated_at - dimensions: - - platform + config: + tags: ["directional_anomalies", "drop"] + arguments: + anomaly_direction: "drop" + timestamp_column: updated_at + dimensions: + - platform - elementary.dimension_anomalies: - dimensions: - - platform - tags: ["dimension_anomalies"] - alias: "dimension_anomalies_no_timestamp" - + config: + tags: ["dimension_anomalies"] + alias: "dimension_anomalies_no_timestamp" + arguments: + dimensions: + - platform + config: + meta: + owner: "egk" + subscribers: "elon, egk" - name: error_model description: We use this model to create error runs and tests - meta: - owner: ["elon@elementary-data.com", "@elon", "egk"] config: tags: ["error_model"] + meta: + owner: ["elon@elementary-data.com", "@elon", "egk"] columns: - name: "missing_column" tests: - uniques: - tags: ["error_test", "regular_tests"] + config: + tags: ["error_test", "regular_tests"] - name: backfill_days_column_anomalies config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at columns: - name: "min_length" tests: - elementary.column_anomalies: - column_anomalies: - - min_length - - max_length - tags: ["backfill_days"] + config: + tags: ["backfill_days"] + arguments: + column_anomalies: + - min_length + - max_length - elementary.column_anomalies: - backfill_days: 7 - column_anomalies: - - min_length - - max_length - tags: ["backfill_days"] + config: + tags: ["backfill_days"] + arguments: + backfill_days: 7 + column_anomalies: + - min_length + - max_length - name: string_column_anomalies - meta: - owner: "@or" - tags: ["marketing"] config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at + owner: "@or" + tags: ["marketing"] tests: - elementary.freshness_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.event_freshness_anomalies: - tags: ["event_freshness_anomalies"] - event_timestamp_column: occurred_at - update_timestamp_column: updated_at + config: + tags: ["event_freshness_anomalies"] + arguments: + event_timestamp_column: occurred_at + update_timestamp_column: updated_at - elementary.all_columns_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] + config: + tags: ["string_column_anomalies", "column_anomalies"] - elementary.schema_changes: - where: 1=1 - tags: ["schema_changes"] + config: + where: 1=1 + tags: ["schema_changes"] columns: - name: "min_length" tests: - relationships: - tags: ["regular_tests"] - to: source('training', 'string_column_anomalies_training') - field: max_length + config: + tags: ["regular_tests"] + arguments: + to: source('training', 'string_column_anomalies_training') + field: max_length - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] - column_anomalies: - - min_length - - max_length - - missing_count + config: + tags: ["string_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - min_length + - max_length + - missing_count - name: max_length tests: - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] + config: + tags: ["string_column_anomalies", "column_anomalies"] - name: average_length tests: - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] - column_anomalies: - - average_length - - null_count + config: + tags: ["string_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - average_length + - null_count - name: missing_count tests: - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] + config: + tags: ["string_column_anomalies", "column_anomalies"] - name: missing_percent tests: - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] + config: + tags: ["string_column_anomalies", "column_anomalies"] - name: updated_at tests: - elementary.column_anomalies: - tags: ["string_column_anomalies", "column_anomalies"] + config: + tags: ["string_column_anomalies", "column_anomalies"] - name: numeric_column_anomalies config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at tests: - elementary.volume_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.volume_anomalies: - anomaly_direction: "drop" - tags: ["directional_anomalies", "drop"] + config: + tags: ["directional_anomalies", "drop"] + arguments: + anomaly_direction: "drop" - elementary.volume_anomalies: - anomaly_direction: "spike" - tags: ["directional_anomalies", "spike"] + config: + tags: ["directional_anomalies", "spike"] + arguments: + anomaly_direction: "spike" - elementary.freshness_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.event_freshness_anomalies: - tags: ["event_freshness_anomalies"] - event_timestamp_column: occurred_at - update_timestamp_column: updated_at + config: + tags: ["event_freshness_anomalies"] + arguments: + event_timestamp_column: occurred_at + update_timestamp_column: updated_at - elementary.schema_changes: - tags: ["schema_changes"] + config: + tags: ["schema_changes"] - elementary.all_columns_anomalies: - tags: ["all_numeric_columns_anomalies"] - column_anomalies: - - average_length - - null_count + config: + tags: ["all_numeric_columns_anomalies"] + arguments: + column_anomalies: + - average_length + - null_count columns: - name: min_val tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - min + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - min - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - max + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - max - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - average + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - average - name: max_val tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - min + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - min - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - max + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - max - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - average + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - average - elementary.column_anomalies: - column_anomalies: - - average - anomaly_direction: "spike" - tags: ["directional_anomalies", "spike"] + config: + tags: ["directional_anomalies", "spike"] + arguments: + column_anomalies: + - average + anomaly_direction: "spike" - elementary.column_anomalies: - column_anomalies: - - average - anomaly_direction: "drop" - tags: ["directional_anomalies", "drop"] + config: + tags: ["directional_anomalies", "drop"] + arguments: + column_anomalies: + - average + anomaly_direction: "drop" - name: average tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - min + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - min - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - max + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - max - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] - column_anomalies: - - average + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - average - name: zero_count tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] - name: zero_percent tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] - name: updated_at tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] - name: variance tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] - name: standard_deviation tests: - elementary.column_anomalies: - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] - name: sum_val tests: - elementary.column_anomalies: - column_anomalies: - - sum - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - sum - name: copy_numeric_column_anomalies config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at tests: - elementary.all_columns_anomalies: - column_anomalies: - - zero_count - tags: ["numeric_column_anomalies", "column_anomalies"] + config: + tags: ["numeric_column_anomalies", "column_anomalies"] + arguments: + column_anomalies: + - zero_count - name: groups columns: @@ -353,15 +448,19 @@ models: - name: group_c tests: - elementary.schema_changes: - tags: ["schema_changes"] + config: + tags: ["schema_changes"] - elementary.schema_changes_from_baseline: - fail_on_added: true - where: 1=1 - tags: ["schema_changes_from_baseline"] + config: + where: 1=1 + tags: ["schema_changes_from_baseline"] + arguments: + fail_on_added: true - elementary.schema_changes_from_baseline: - tags: ["schema_changes_from_baseline", "error_test"] - enforce_types: true - + config: + tags: ["schema_changes_from_baseline", "error_test"] + arguments: + enforce_types: true - name: stats_players columns: - name: player @@ -372,121 +471,132 @@ models: data_type: INTEGER tests: - elementary.schema_changes: - tags: ["schema_changes"] + config: + tags: ["schema_changes"] - elementary.schema_changes_from_baseline: - tags: ["schema_changes_from_baseline"] + config: + tags: ["schema_changes_from_baseline"] - elementary.schema_changes_from_baseline: - tags: ["schema_changes_from_baseline"] - enforce_types: true - + config: + tags: ["schema_changes_from_baseline"] + arguments: + enforce_types: true - name: stats_team tests: - elementary.schema_changes: - tags: ["schema_changes"] + config: + tags: ["schema_changes"] - name: users_per_day_weekly_seasonal config: - elementary: - backfill_days: 14 + meta: + elementary: + backfill_days: 14 tests: - elementary.volume_anomalies: - alias: day_of_week_volume_anomalies_no_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 + config: + alias: day_of_week_volume_anomalies_no_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 - elementary.volume_anomalies: - alias: day_of_week_volume_anomalies_with_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 - seasonality: day_of_week + config: + alias: day_of_week_volume_anomalies_with_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 + seasonality: day_of_week - elementary.volume_anomalies: - alias: hour_of_week_volume_anomalies_no_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 - time_bucket: - period: hour - count: 1 + config: + alias: hour_of_week_volume_anomalies_no_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 + time_bucket: + period: hour + count: 1 - elementary.volume_anomalies: - alias: hour_of_week_volume_anomalies_with_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 - time_bucket: - period: hour - count: 1 - seasonality: hour_of_week - + config: + alias: hour_of_week_volume_anomalies_with_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 + time_bucket: + period: hour + count: 1 + seasonality: hour_of_week - name: users_per_hour_daily_seasonal tests: - elementary.volume_anomalies: - alias: hour_of_day_volume_anomalies_no_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 - time_bucket: - period: hour - count: 1 + config: + alias: hour_of_day_volume_anomalies_no_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 + time_bucket: + period: hour + count: 1 - elementary.volume_anomalies: - alias: hour_of_day_volume_anomalies_with_seasonality - timestamp_column: "updated_at" - tags: ["seasonality_volume", "table_anomalies"] - sensitivity: 2 - time_bucket: - period: hour - count: 1 - seasonality: hour_of_day - + config: + alias: hour_of_day_volume_anomalies_with_seasonality + tags: ["seasonality_volume", "table_anomalies"] + arguments: + timestamp_column: "updated_at" + sensitivity: 2 + time_bucket: + period: hour + count: 1 + seasonality: hour_of_day - name: ephemeral_model config: - elementary: - timestamp_column: updated_at + meta: + elementary: + timestamp_column: updated_at tests: - elementary.volume_anomalies: - tags: ["ephemeral_model", "error_test"] + config: + tags: ["ephemeral_model", "error_test"] - elementary.all_columns_anomalies: - tags: ["ephemeral_model", "error_test"] + config: + tags: ["ephemeral_model", "error_test"] - elementary.freshness_anomalies: - where: 1=1 - tags: ["ephemeral_model", "error_test"] + config: + where: 1=1 + tags: ["ephemeral_model", "error_test"] - elementary.schema_changes: - tags: ["ephemeral_model", "error_test"] + config: + tags: ["ephemeral_model", "error_test"] - name: config_levels_test_and_model - tags: ["config_levels"] config: - elementary: - min_training_set_size: 22 - days_back: 100 - backfill_days: 10 - anomaly_direction: "drop" - anomaly_sensitivity: 4 - where_expression: "true" - timestamp_column: "updated_at" - time_bucket: - period: hour - count: 4 + meta: + elementary: + min_training_set_size: 22 + days_back: 100 + backfill_days: 10 + anomaly_direction: "drop" + anomaly_sensitivity: 4 + where_expression: "true" + timestamp_column: "updated_at" + time_bucket: + period: hour + count: 4 + tags: ["config_levels"] tests: - config_levels: - tags: ["config_levels"] - alias: "test_level_config" - min_training_set_size: 18 - days_back: 5 - backfill_days: 5 - seasonality: "day_of_week" - anomaly_direction: "spike" - anomaly_sensitivity: 5 - where_expression: "1=1" - timestamp_column: "occurred_at" - time_bucket: - period: day - count: 1 - expected_config: ## Test level expected config - seasonality: "day_of_week" + config: + tags: ["config_levels"] + alias: "test_level_config" + arguments: min_training_set_size: 18 - days_back: 35 ## *7 because of seasonality + days_back: 5 backfill_days: 5 + seasonality: "day_of_week" anomaly_direction: "spike" anomaly_sensitivity: 5 where_expression: "1=1" @@ -494,120 +604,127 @@ models: time_bucket: period: day count: 1 + expected_config: ## Test level expected config + seasonality: "day_of_week" + min_training_set_size: 18 + days_back: 35 ## *7 because of seasonality + backfill_days: 5 + anomaly_direction: "spike" + anomaly_sensitivity: 5 + where_expression: "1=1" + timestamp_column: "occurred_at" + time_bucket: + period: day + count: 1 - config_levels: - tags: ["config_levels"] - alias: "model_level_config" - expected_config: ## Model level expected config - min_training_set_size: 22 - seasonality: null - days_back: 100 - backfill_days: 10 - anomaly_direction: "drop" - anomaly_sensitivity: 4 - where_expression: "true" - timestamp_column: "updated_at" - time_bucket: - period: hour - count: 4 + config: + tags: ["config_levels"] + alias: "model_level_config" + arguments: + expected_config: ## Model level expected config + min_training_set_size: 22 + seasonality: + days_back: 100 + backfill_days: 10 + anomaly_direction: "drop" + anomaly_sensitivity: 4 + where_expression: "true" + timestamp_column: "updated_at" + time_bucket: + period: hour + count: 4 - name: config_levels_project - tags: ["config_levels"] tests: - config_levels: - tags: ["config_levels"] - alias: "project_level_config" - expected_config: ## Project level expected config - min_training_set_size: 14 - seasonality: null - days_back: 30 - backfill_days: 2 - anomaly_direction: both - anomaly_sensitivity: 3 - where_expression: null - timestamp_column: null - time_bucket: - period: day - count: 1 + config: + tags: ["config_levels"] + alias: "project_level_config" + arguments: + expected_config: ## Project level expected config + min_training_set_size: 14 + seasonality: + days_back: 30 + backfill_days: 2 + anomaly_direction: both + anomaly_sensitivity: 3 + where_expression: + timestamp_column: + time_bucket: + period: day + count: 1 + config: + tags: ["config_levels"] sources: - name: training - schema: "{{ target.schema if target.type == 'dremio' else 'test_seeds' }}" + schema: "{{ target.schema if target.type == 'dremio' else env_var('TEST_SEEDS_SCHEMA', 'test_seeds') }}" tables: - name: users_per_hour_daily_seasonal_training columns: - name: "user_id" tests: - relationships: - tags: ["regular_tests"] - to: source('training', 'users_per_day_weekly_seasonal_training') - field: user_id + config: + tags: ["regular_tests"] + arguments: + to: source('training', 'users_per_day_weekly_seasonal_training') + field: user_id - name: any_type_column_anomalies_training - meta: - owner: ["@edr", "egk"] - freshness: - error_after: - count: 1 - period: minute - loaded_at_field: updated_at tests: - elementary.volume_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.freshness_anomalies: - tags: ["table_anomalies", "error_test"] + config: + tags: ["table_anomalies", "error_test"] - elementary.event_freshness_anomalies: - tags: ["event_freshness_anomalies"] - event_timestamp_column: occurred_at + config: + tags: ["event_freshness_anomalies"] + arguments: + event_timestamp_column: occurred_at + config: + freshness: + error_after: + count: 1 + period: minute + loaded_at_field: updated_at + meta: + owner: ["@edr", "egk"] - name: string_column_anomalies_training - meta: - owner: "@edr" - elementary: - timestamp_column: updated_at - freshness: - error_after: - count: 1 - period: minute - loaded_at_field: no_such_column tests: - elementary.volume_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.freshness_anomalies: - tags: ["table_anomalies"] + config: + tags: ["table_anomalies"] - elementary.event_freshness_anomalies: - tags: ["event_freshness_anomalies"] - event_timestamp_column: occurred_at - update_timestamp_column: updated_at + config: + tags: ["event_freshness_anomalies"] + arguments: + event_timestamp_column: occurred_at + update_timestamp_column: updated_at + config: + freshness: + error_after: + count: 1 + period: minute + loaded_at_field: no_such_column + meta: + owner: "@edr" + elementary: + timestamp_column: updated_at - name: numeric_column_anomalies_training - meta: - elementary: - min_training_set_size: 22 - days_back: 100 - backfill_days: 10 - anomaly_direction: "drop" - anomaly_sensitivity: 4 - where_expression: "true" - timestamp_column: "updated_at" - time_bucket: - period: hour - count: 4 tests: - config_levels: - tags: ["config_levels"] - alias: "test_level_config" - min_training_set_size: 18 - days_back: 5 - backfill_days: 5 - seasonality: "day_of_week" - anomaly_direction: "spike" - anomaly_sensitivity: 5 - where_expression: "1=1" - timestamp_column: "occurred_at" - time_bucket: - period: day - count: 1 - expected_config: ## Test level expected config - seasonality: "day_of_week" + config: + tags: ["config_levels"] + alias: "test_level_config" + arguments: min_training_set_size: 18 - days_back: 35 ## *7 because of seasonality + days_back: 5 backfill_days: 5 + seasonality: "day_of_week" anomaly_direction: "spike" anomaly_sensitivity: 5 where_expression: "1=1" @@ -615,43 +732,73 @@ sources: time_bucket: period: day count: 1 + expected_config: ## Test level expected config + seasonality: "day_of_week" + min_training_set_size: 18 + days_back: 35 ## *7 because of seasonality + backfill_days: 5 + anomaly_direction: "spike" + anomaly_sensitivity: 5 + where_expression: "1=1" + timestamp_column: "occurred_at" + time_bucket: + period: day + count: 1 - config_levels: - tags: ["config_levels"] - alias: "model_level_config" - expected_config: ## Model level expected config - min_training_set_size: 22 - seasonality: null - days_back: 100 - backfill_days: 10 - anomaly_direction: "drop" - anomaly_sensitivity: 4 - where_expression: "true" - timestamp_column: "updated_at" - time_bucket: - period: hour - count: 4 + config: + tags: ["config_levels"] + alias: "model_level_config" + arguments: + expected_config: ## Model level expected config + min_training_set_size: 22 + seasonality: + days_back: 100 + backfill_days: 10 + anomaly_direction: "drop" + anomaly_sensitivity: 4 + where_expression: "true" + timestamp_column: "updated_at" + time_bucket: + period: hour + count: 4 + config: + meta: + elementary: + min_training_set_size: 22 + days_back: 100 + backfill_days: 10 + anomaly_direction: "drop" + anomaly_sensitivity: 4 + where_expression: "true" + timestamp_column: "updated_at" + time_bucket: + period: hour + count: 4 - name: users_per_day_weekly_seasonal_training - name: validation - schema: "{{ target.schema if target.type == 'dremio' else 'test_seeds' }}" + schema: "{{ target.schema if target.type == 'dremio' else env_var('TEST_SEEDS_SCHEMA', 'test_seeds') }}" tables: - name: users_per_hour_daily_seasonal_validation - name: any_type_column_anomalies_validation - meta: - owner: "hello, world" - freshness: - warn_after: - count: 1 - period: minute - loaded_at_field: updated_at tests: - elementary.all_columns_anomalies: - tags: ["elementary_source"] + config: + tags: ["elementary_source"] columns: - name: null_count_int tests: - generic_test_on_column: - tags: ["regular_tests"] + config: + tags: ["regular_tests"] + config: + freshness: + warn_after: + count: 1 + period: minute + loaded_at_field: updated_at + meta: + owner: "hello, world" - name: users_per_day_weekly_seasonal_validation exposures: @@ -667,8 +814,9 @@ exposures: owner: name: Complete Nonsense email: fake@fakerson.com - tags: - - marketing + config: + tags: + - marketing - name: weekly_jaffle_metrics type: dashboard @@ -682,15 +830,15 @@ exposures: owner: name: Claire from Data email: data@jaffleshop.com - tags: - - hack - - the - - planet - meta: - platform: Tableau - workbook: By the Week - path: ByTheWeek/Jaffles - + config: + tags: + - hack + - the + - planet + meta: + platform: Tableau + workbook: By the Week + path: ByTheWeek/Jaffles - name: monthly_jaffle_metrics type: dashboard maturity: high @@ -703,15 +851,15 @@ exposures: owner: name: Claire from Data email: data@jaffleshop.com - tags: - - hack - - the - - planet - meta: - platform: Looker - workbook: By the Month - path: ByTheMonth/Jaffles - + config: + tags: + - hack + - the + - planet + meta: + platform: Looker + workbook: By the Month + path: ByTheMonth/Jaffles - name: daily_jaffle_metrics type: dashboard maturity: high @@ -724,11 +872,12 @@ exposures: owner: name: Claire from Data email: data@jaffleshop.com - tags: - - hack - - the - - planet - meta: - platform: bi.tool - workbook: By the Day - path: ByTheDay/Jaffles + config: + tags: + - hack + - the + - planet + meta: + platform: bi.tool + workbook: By the Day + path: ByTheDay/Jaffles diff --git a/tests/profiles/generate_profiles.py b/tests/profiles/generate_profiles.py index 1ea3115c5..b992673b2 100644 --- a/tests/profiles/generate_profiles.py +++ b/tests/profiles/generate_profiles.py @@ -71,11 +71,18 @@ def _yaml_inline(value: Any) -> str: show_default=True, help="Name of the env-var holding the base64-encoded JSON secrets blob.", ) +@click.option( + "--var", + "extra_vars", + multiple=True, + help="Extra template variable as KEY=VALUE (may be repeated).", +) def main( template: Path, output: Path, schema_name: str, secrets_json_env: str, + extra_vars: tuple[str, ...], ) -> None: """Render a Jinja2 profiles template into a dbt profiles.yml file. @@ -117,6 +124,14 @@ def main( err=True, ) + for extra_var in extra_vars: + key, sep, value = extra_var.partition("=") + if not sep or not key: + raise click.ClickException( + f"Invalid --var {extra_var!r}, expected KEY=VALUE" + ) + context[key.lower()] = value + # ── Render ────────────────────────────────────────────────────────── # When secrets are loaded, use StrictUndefined so typos in secret keys # fail fast. For docker-only runs (no secrets) use _NullUndefined so diff --git a/tests/profiles/profiles.yml.j2 b/tests/profiles/profiles.yml.j2 index 4bb0b4609..e095be485 100644 --- a/tests/profiles/profiles.yml.j2 +++ b/tests/profiles/profiles.yml.j2 @@ -112,7 +112,7 @@ elementary_tests: database: {{ snowflake_database | toyaml }} warehouse: {{ snowflake_warehouse | toyaml }} schema: {{ schema_name }} - threads: 4 + threads: {{ snowflake_threads | default(4) }} bigquery: &bigquery type: bigquery diff --git a/tests/tests_with_db/conftest.py b/tests/tests_with_db/conftest.py index 831d0d441..704ac9ebd 100644 --- a/tests/tests_with_db/conftest.py +++ b/tests/tests_with_db/conftest.py @@ -4,9 +4,9 @@ import env import pytest -from dbt.version import __version__ as dbt_version from packaging import version +from elementary.clients.dbt.dbt_installation import get_dbt_core_version from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner DBT_PROJECT_PATH = Path(__file__).parent / "dbt_project" @@ -60,7 +60,8 @@ def requires_dbt_version(request): required_version = request.node.get_closest_marker("requires_dbt_version").args[ 0 ] - if version.parse(dbt_version) < version.parse(required_version): + dbt_version = get_dbt_core_version() + if dbt_version is not None and dbt_version < version.parse(required_version): pytest.skip( "Test requires dbt version {} or above, but {} is installed.".format( required_version, dbt_version diff --git a/tests/unit/clients/dbt_runner/test_factory.py b/tests/unit/clients/dbt_runner/test_factory.py new file mode 100644 index 000000000..d78bc386c --- /dev/null +++ b/tests/unit/clients/dbt_runner/test_factory.py @@ -0,0 +1,142 @@ +from typing import Optional +from unittest import mock + +import pytest +from packaging import version + +from elementary.clients.dbt import factory +from elementary.clients.dbt.dbt2_runner import Dbt2Runner +from elementary.clients.dbt.dbt_installation import ( + get_dbt2_binary_path, + is_dbt2_binary_available, +) +from elementary.clients.dbt.factory import ( + RunnerMethod, + get_dbt_runner_class, + get_dbt_runner_method, +) +from elementary.clients.dbt.subprocess_dbt_runner import SubprocessDbtRunner + + +def _mock_installation( + monkeypatch, + dbt_core_version: Optional[str] = None, + dbt2_binary_available: bool = False, +): + monkeypatch.setattr( + factory, + "get_dbt_core_version", + lambda: version.Version(dbt_core_version) if dbt_core_version else None, + ) + monkeypatch.setattr( + factory, "is_dbt2_binary_available", lambda: dbt2_binary_available + ) + + +@pytest.mark.parametrize( + "dbt_core_version,dbt2_binary_available,expected_method", + [ + ("1.8.0", False, RunnerMethod.API), + ("1.10.5", True, RunnerMethod.API), + ("1.4.0", False, RunnerMethod.SUBPROCESS), + ("2.0.0b2", False, RunnerMethod.DBT2), + (None, True, RunnerMethod.DBT2), + (None, False, RunnerMethod.SUBPROCESS), + ], +) +def test_get_dbt_runner_method_auto_detection( + monkeypatch, dbt_core_version, dbt2_binary_available, expected_method +): + monkeypatch.delenv("DBT_RUNNER_METHOD", raising=False) + _mock_installation(monkeypatch, dbt_core_version, dbt2_binary_available) + assert get_dbt_runner_method() == expected_method + + +@pytest.mark.parametrize( + "env_value,expected_method", + [ + ("subprocess", RunnerMethod.SUBPROCESS), + ("api", RunnerMethod.API), + ("dbt2", RunnerMethod.DBT2), + ("fusion", RunnerMethod.FUSION), + ], +) +def test_get_dbt_runner_method_env_override(monkeypatch, env_value, expected_method): + monkeypatch.setenv("DBT_RUNNER_METHOD", env_value) + assert get_dbt_runner_method() == expected_method + + +def test_get_dbt_runner_class(): + assert get_dbt_runner_class(RunnerMethod.SUBPROCESS) is SubprocessDbtRunner + assert get_dbt_runner_class(RunnerMethod.DBT2) is Dbt2Runner + assert get_dbt_runner_class(RunnerMethod.FUSION) is Dbt2Runner + + +@mock.patch("elementary.clients.dbt.dbt_installation.shutil.which") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_package_version") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_core_version") +def test_dbt2_runner_uses_path_binary_when_no_dbt_core( + mock_get_dbt_core_version, mock_get_dbt_package_version, mock_which, monkeypatch +): + monkeypatch.delenv("DBT_FUSION_PATH", raising=False) + mock_get_dbt_core_version.return_value = None + mock_get_dbt_package_version.return_value = None + mock_which.return_value = "/some/venv/bin/dbt" + + assert get_dbt2_binary_path() == "/some/venv/bin/dbt" + + +@mock.patch("elementary.clients.dbt.dbt_installation.shutil.which") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_package_version") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_core_version") +def test_dbt2_runner_uses_path_binary_when_dbt2_pip_installed_alongside_dbt_core_1x( + mock_get_dbt_core_version, mock_get_dbt_package_version, mock_which, monkeypatch +): + monkeypatch.delenv("DBT_FUSION_PATH", raising=False) + mock_get_dbt_core_version.return_value = version.Version("1.10.0") + mock_get_dbt_package_version.return_value = version.Version("2.0.0rc212") + mock_which.return_value = "/some/venv/bin/dbt" + + assert get_dbt2_binary_path() == "/some/venv/bin/dbt" + + +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_package_version") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_core_version") +def test_dbt2_runner_ignores_path_binary_when_dbt_core_1x( + mock_get_dbt_core_version, mock_get_dbt_package_version, monkeypatch +): + monkeypatch.delenv("DBT_FUSION_PATH", raising=False) + mock_get_dbt_core_version.return_value = version.Version("1.10.0") + mock_get_dbt_package_version.return_value = None + + assert get_dbt2_binary_path().endswith("/.local/bin/dbt") + + +def test_dbt2_runner_honors_dbt_fusion_path_env_var(monkeypatch): + monkeypatch.setenv("DBT_FUSION_PATH", "/custom/path/dbt") + + assert get_dbt2_binary_path() == "/custom/path/dbt" + + +@mock.patch("elementary.clients.dbt.dbt_installation.os.path.exists") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_package_version") +def test_dbt2_binary_available_when_dbt_fusion_path_env_var_points_to_binary( + mock_get_dbt_package_version, mock_exists, monkeypatch +): + monkeypatch.setenv("DBT_FUSION_PATH", "/custom/path/dbt") + mock_get_dbt_package_version.return_value = None + mock_exists.side_effect = lambda path: path == "/custom/path/dbt" + + assert is_dbt2_binary_available() + + +@mock.patch("elementary.clients.dbt.dbt_installation.os.path.exists") +@mock.patch("elementary.clients.dbt.dbt_installation.get_dbt_package_version") +def test_dbt2_binary_not_available_when_nothing_installed( + mock_get_dbt_package_version, mock_exists, monkeypatch +): + monkeypatch.delenv("DBT_FUSION_PATH", raising=False) + mock_get_dbt_package_version.return_value = None + mock_exists.return_value = False + + assert not is_dbt2_binary_available() diff --git a/tests/unit/clients/dbt_runner/test_retry_logic.py b/tests/unit/clients/dbt_runner/test_retry_logic.py index 3a1b16d97..47fec36af 100644 --- a/tests/unit/clients/dbt_runner/test_retry_logic.py +++ b/tests/unit/clients/dbt_runner/test_retry_logic.py @@ -8,6 +8,15 @@ from elementary.clients.dbt.command_line_dbt_runner import _TRANSIENT_MAX_RETRIES from elementary.exceptions.exceptions import DbtCommandError +# The dbt Python API only exists in dbt-core 1.x; with dbt-core 2.x or a +# binary-only Fusion installation the api_dbt_runner module can't be imported. +try: + import elementary.clients.dbt.api_dbt_runner # noqa: F401 + + HAS_DBT_PYTHON_API = True +except ImportError: + HAS_DBT_PYTHON_API = False + # Patch tenacity wait to zero so tests don't block on exponential backoff. _ZERO_WAIT = mock.patch( "elementary.clients.dbt.command_line_dbt_runner._TRANSIENT_WAIT_MULTIPLIER", 0 @@ -219,6 +228,9 @@ def _make_api_runner(**kwargs): return APIDbtRunner(**defaults) +@pytest.mark.skipif( + not HAS_DBT_PYTHON_API, reason="The dbt Python API is not available" +) @_ZERO_WAIT class TestAPIDbtRunnerTransientDetection: """Test that APIDbtRunner surfaces exception text for transient error detection.