diff --git a/.github/actions/prebuild-wheel/action.yml b/.github/actions/prebuild-wheel/action.yml new file mode 100644 index 000000000..a4568d67f --- /dev/null +++ b/.github/actions/prebuild-wheel/action.yml @@ -0,0 +1,33 @@ +name: 'Pre-build DQX wheel' +description: > + Builds the DQX wheel once per CI job and exports the absolute path as DQX_PREBUILT_WHEEL. + Tests that use PrebuiltWheels copy this artifact instead of running `pip wheel` per-test, + which avoids JFrog auth failures after the 1h OIDC token TTL expires mid-suite. +runs: + using: "composite" + steps: + - name: Build wheel + shell: bash + run: | + # Bump version per CI run so cluster library installs and pip caches + # don't reuse stale bytes. Delegate the scheme to blueprint so it + # matches what blueprint would compute per-test + NEW_VERSION="$(UV_FROZEN=1 uv run --quiet python -c ' + from databricks.labs.blueprint.wheels import ProductInfo + from databricks.labs.dqx.config import WorkspaceConfig + print(ProductInfo.from_class(WorkspaceConfig).version()) + ')" + test -n "$NEW_VERSION" || { echo "failed to compute unreleased version"; exit 1; } + + VERSION_FILE="src/databricks/labs/dqx/__about__.py" + sed -i "s/^__version__ = .*/__version__ = \"${NEW_VERSION}\"/" "$VERSION_FILE" + + make build + + # Restore so later steps see a clean tree + git checkout -- "$VERSION_FILE" + + WHEEL_PATH="$(find dist -maxdepth 1 -name '*.whl' -print -quit)" + test -n "$WHEEL_PATH" || { echo "no wheel produced in dist/"; ls -la dist/; exit 1; } + + printf '%s=%s\n' 'DQX_PREBUILT_WHEEL' "$PWD/$WHEEL_PATH" >> "$GITHUB_ENV" diff --git a/.github/workflows/acceptance.yml b/.github/workflows/acceptance.yml index 5ac381aed..29efc6f62 100644 --- a/.github/workflows/acceptance.yml +++ b/.github/workflows/acceptance.yml @@ -45,8 +45,8 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env - - name: Run unit tests and generate test coverage report - run: make test + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel # Integration tests are run from within tests/integration folder. # Create .coveragerc with correct relative path to source code. @@ -63,7 +63,7 @@ jobs: # and generate code coverage for modules defined in .coveragerc # Run 10 tests in parallel: https://github.com/databrickslabs/sandbox/blob/main/acceptance/ecosystem/pytest_run.py - name: Run integration tests and generate test coverage report - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -73,16 +73,15 @@ jobs: ARM_CLIENT_ID: ${{ secrets.ARM_CLIENT_ID }} ARM_TENANT_ID: ${{ secrets.ARM_TENANT_ID }} COVERAGE_FILE: ${{ github.workspace }}/.coverage # make sure the coverage report is preserved - - name: Merge coverage reports and convert them to XML run: make combine-coverage - # Recursively search the entire workspace directory for all coverage reports. - # All uploaded test coverage reports will be used even if publish is done multiple time. - name: Publish test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: integration integration_serverless: needs: not-a-fork @@ -104,6 +103,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Integration tests are run from within tests/integration folder. # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for integration tests @@ -116,7 +118,7 @@ jobs: EOF - name: Run integration tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -127,15 +129,15 @@ jobs: ARM_TENANT_ID: ${{ secrets.ARM_TENANT_ID }} DATABRICKS_SERVERLESS_COMPUTE_ID: ${{ env.DATABRICKS_SERVERLESS_COMPUTE_ID }} COVERAGE_FILE: ${{ github.workspace }}/.coverage # make sure the coverage report is preserved - - name: Merge coverage reports and convert them to XML run: make combine-coverage - # collects all coverage reports - name: Publish test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: integration-serverless e2e: needs: not-a-fork @@ -155,12 +157,15 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Required for DAB (Databricks Asset Bundle) e2e tests - name: Install Databricks CLI uses: databricks/setup-cli@596b0a354ba14aa59921aca1b02bd67c2b0a81a5 # v0.297.2 - name: Run e2e tests - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -191,12 +196,15 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Required for DAB (Databricks Asset Bundle) e2e tests - name: Install Databricks CLI uses: databricks/setup-cli@596b0a354ba14aa59921aca1b02bd67c2b0a81a5 # v0.297.2 - name: Run e2e tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h diff --git a/.github/workflows/anomaly.yml b/.github/workflows/anomaly.yml index 70710d85f..91b7f86d1 100644 --- a/.github/workflows/anomaly.yml +++ b/.github/workflows/anomaly.yml @@ -43,6 +43,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for anomaly tests run: | @@ -54,7 +57,7 @@ jobs: EOF - name: Run anomaly integration tests - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 3h @@ -68,17 +71,16 @@ jobs: MLFLOW_TRACKING_URI: databricks MLFLOW_REGISTRY_URI: databricks-uc MLFLOW_HTTP_REQUEST_TIMEOUT: "600" - MLFLOW_HTTP_REQUEST_MAX_RETRIES: "10" - + MLFLOW_HTTP_REQUEST_MAX_RETRIES: "9" - name: Merge coverage reports and convert them to XML run: make combine-coverage - # Recursively search the entire workspace directory for all coverage reports. - # All uploaded test coverage reports will be used even if publish is done multiple time. - name: Publish test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: anomaly anomaly-tests-serverless: needs: not-a-fork @@ -98,6 +100,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for anomaly tests run: | @@ -109,7 +114,7 @@ jobs: EOF - name: Run anomaly integration tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 3h @@ -124,13 +129,13 @@ jobs: MLFLOW_TRACKING_URI: databricks MLFLOW_REGISTRY_URI: databricks-uc MLFLOW_HTTP_REQUEST_TIMEOUT: "600" - MLFLOW_HTTP_REQUEST_MAX_RETRIES: "10" - + MLFLOW_HTTP_REQUEST_MAX_RETRIES: "9" - name: Merge coverage reports and convert them to XML run: make combine-coverage - # collects all coverage reports - name: Publish test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: anomaly-serverless diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml index 8926f9498..7367934f8 100644 --- a/.github/workflows/nightly.yml +++ b/.github/workflows/nightly.yml @@ -33,6 +33,9 @@ jobs: - name: Run unit tests and generate test coverage report run: make test + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Integration tests are run from within tests/integration folder. # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for integration tests @@ -47,7 +50,7 @@ jobs: # Run tests from `tests/integration` as defined in .codegen.json # and generate code coverage for modules defined in .coveragerc - name: Run integration tests and generate test coverage report - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -58,15 +61,22 @@ jobs: ARM_CLIENT_ID: ${{ secrets.ARM_CLIENT_ID }} ARM_TENANT_ID: ${{ secrets.ARM_TENANT_ID }} COVERAGE_FILE: ${{ github.workspace }}/.coverage # make sure the coverage report is preserved - - name: Merge coverage reports and convert them to XML run: make combine-coverage - # collects all coverage reports - - name: Publish test coverage + - name: Publish unit test coverage + uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 + with: + use_oidc: true + files: coverage-unit.xml + flags: unit + + - name: Publish integration test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: integration integration_serverless: environment: tool @@ -87,6 +97,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Integration tests are run from within tests/integration folder. # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for integration tests @@ -99,7 +112,7 @@ jobs: EOF - name: Run integration tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -111,15 +124,15 @@ jobs: ARM_TENANT_ID: ${{ secrets.ARM_TENANT_ID }} DATABRICKS_SERVERLESS_COMPUTE_ID: ${{ env.DATABRICKS_SERVERLESS_COMPUTE_ID }} COVERAGE_FILE: ${{ github.workspace }}/.coverage # make sure the coverage report is preserved - - name: Merge coverage reports and convert them to XML run: make combine-coverage - # collects all coverage reports - name: Publish test coverage uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: integration-serverless anomaly-tests: environment: tool @@ -138,6 +151,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for anomaly tests run: | @@ -149,7 +165,7 @@ jobs: EOF - name: Run anomaly integration tests - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 3h @@ -164,8 +180,7 @@ jobs: MLFLOW_TRACKING_URI: databricks MLFLOW_REGISTRY_URI: databricks-uc MLFLOW_HTTP_REQUEST_TIMEOUT: "600" - MLFLOW_HTTP_REQUEST_MAX_RETRIES: "10" - + MLFLOW_HTTP_REQUEST_MAX_RETRIES: "9" - name: Merge coverage reports and convert them to XML run: make combine-coverage @@ -173,6 +188,8 @@ jobs: uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: anomaly anomaly-tests-serverless: environment: tool @@ -191,6 +208,9 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Create .coveragerc with correct relative path to source code. - name: Prepare code coverage configuration for anomaly tests run: | @@ -202,7 +222,7 @@ jobs: EOF - name: Run anomaly integration tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 3h @@ -218,8 +238,7 @@ jobs: MLFLOW_TRACKING_URI: databricks MLFLOW_REGISTRY_URI: databricks-uc MLFLOW_HTTP_REQUEST_TIMEOUT: "600" - MLFLOW_HTTP_REQUEST_MAX_RETRIES: "10" - + MLFLOW_HTTP_REQUEST_MAX_RETRIES: "9" - name: Merge coverage reports and convert them to XML run: make combine-coverage @@ -227,6 +246,8 @@ jobs: uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 with: use_oidc: true + files: coverage-combined.xml + flags: anomaly-serverless e2e: environment: tool @@ -245,12 +266,15 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Required for DAB (Databricks Asset Bundle) e2e tests - name: Install Databricks CLI uses: databricks/setup-cli@596b0a354ba14aa59921aca1b02bd67c2b0a81a5 # v0.297.2 - name: Run e2e tests - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h @@ -280,12 +304,15 @@ jobs: - name: Setup environment uses: ./.github/actions/setup-env + - name: Pre-build DQX wheel + uses: ./.github/actions/prebuild-wheel + # Required for DAB (Databricks Asset Bundle) e2e tests - name: Install Databricks CLI uses: databricks/setup-cli@596b0a354ba14aa59921aca1b02bd67c2b0a81a5 # v0.297.2 - name: Run e2e tests on serverless cluster - uses: databrickslabs/sandbox/acceptance@3313d06ce86227537b3f37f5974f7eecb2a8e59a # acceptance/v0.4.4 + uses: databrickslabs/sandbox/acceptance@83461e5dd7021feabb1a9ca3ee10d6f46b72092a # acceptance/v0.4.6 with: vault_uri: ${{ secrets.VAULT_URI }} timeout: 2h diff --git a/.github/workflows/push.yml b/.github/workflows/push.yml index 63c677078..2bdecbd8b 100644 --- a/.github/workflows/push.yml +++ b/.github/workflows/push.yml @@ -27,7 +27,10 @@ jobs: - run: echo "Not a fork PR, proceeding" ci: - needs: not-a-fork + # Use a job-level `if:` instead of `needs: not-a-fork` so that for fork PRs the matrix + # still expands and each leaf (ci (3.10), ci (3.11), ci (3.12)) reports its own + # "skipped" status. + if: github.event_name != 'pull_request' || !github.event.pull_request.head.repo.fork strategy: fail-fast: false matrix: @@ -49,6 +52,16 @@ jobs: - name: Run unit tests run: make test + # Publish unit test coverage only from the 3.12 leg to avoid duplicate + # uploads. Codecov merges unit + integration reports via the `flags` + # dimension; acceptance.yml publishes the integration coverage. + - name: Publish unit test coverage + if: matrix.pyVersion == '3.12' + uses: codecov/codecov-action@b9fd7d16f6d7d1b5d2bec1a2887e65ceed900238 # v4.6.0 + with: + use_oidc: true + flags: unit + fmt: needs: not-a-fork runs-on: diff --git a/AGENTS.md b/AGENTS.md index aba0da143..c982ea198 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -263,7 +263,7 @@ Use [Google Style Python Docstrings](https://sphinxcontrib-napoleon.readthedocs. | Need | Import | Usage | |---|---|---| -| Module logger | `import logging` | `logger = logging.getLogger(__name__)` | +| Module logger | `import logging` | `logger = logging.getLogger(__name__)`; use f-strings for log messages (e.g. `logger.warning(f"value={x}")`), not `%`-style args | | CLI entrypoint logger | `blueprint.entrypoint.get_logger` | `logger = get_logger(__file__)` | | Package logger setup | `blueprint.logger.install_logger` | call once in `__init__.py` | | Parallel tasks | `blueprint.parallel.Threads` | `Threads.strict("label", tasks)` | diff --git a/demos/dqx_demo_summary_metrics.py b/demos/dqx_demo_summary_metrics.py index 9fca8bb0d..5ebcc6444 100644 --- a/demos/dqx_demo_summary_metrics.py +++ b/demos/dqx_demo_summary_metrics.py @@ -111,6 +111,7 @@ # MAGIC - The number of input rows # MAGIC - The number of rows with warnings # MAGIC - The number of rows with errors +# MAGIC - Per-check error and warning counts (as a JSON breakdown per check name) # MAGIC - User-defined custom metrics # MAGIC # MAGIC Summary metrics can be tracked for both streaming and batch datasets. To track summary metrics, pass a `DQMetricsObserver` when creating the `DQEngine`. diff --git a/docs/dqx/docs/guide/summary_metrics.mdx b/docs/dqx/docs/guide/summary_metrics.mdx index 4a72ae6eb..1d2c403ac 100644 --- a/docs/dqx/docs/guide/summary_metrics.mdx +++ b/docs/dqx/docs/guide/summary_metrics.mdx @@ -12,7 +12,7 @@ DQX provides comprehensive functionality to capture and store aggregate statisti ## Overview -Summary metrics in DQX capture both **built-in metrics** (automatically calculated) and **custom metrics** (user-defined SQL expressions) during data quality checking. These metrics are collected using Spark's built-in [Observation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Observation.html) functionality and can be persisted to tables for historical analysis. +Summary metrics in DQX capture **built-in metrics** (automatically calculated), **per-check metrics** (error and warning counts broken down by check name), and **custom metrics** (user-defined SQL expressions) during data quality checking. These metrics are collected using Spark's built-in [Observation](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.Observation.html) functionality and can be persisted to tables for historical analysis. ### Built-in Metrics @@ -24,6 +24,40 @@ DQX automatically captures the following built-in metrics for every data quality | `error_row_count` | `int` | Number of rows that failed error-level checks | | `warning_row_count` | `int` | Number of rows that triggered warning-level checks | | `valid_row_count` | `int` | Number of rows that passed all checks (have no errors and warnings) | +| `check_metrics` | `string` (JSON) | Per-check breakdown of error and warning counts, stored as a JSON array of structs: `[{"check_name": "...", "error_count": N, "warning_count": N}, ...]`. Automatically included when checks are applied. | + +### Per-Check Metrics + +When checks are applied, DQX automatically includes a `check_metrics` metric that provides a per-check breakdown of error and warning counts. This makes it possible to identify which specific checks are failing without querying the row-level `_errors` and `_warnings` columns. + +The `check_metrics` value is a JSON-serialized array of structs, with one entry per check: + +```json +[ + {"check_name": "id_is_not_null", "error_count": 5, "warning_count": 0}, + {"check_name": "name_is_not_null_and_not_empty", "error_count": 0, "warning_count": 3} +] +``` + +Each entry contains: +- `check_name` — the name of the check (either explicitly set via `name` in the rule definition, or auto-derived from the check function and arguments) +- `error_count` — number of rows where this check triggered an error +- `warning_count` — number of rows where this check triggered a warning + +When persisted to the metrics table, `check_metrics` is stored as a single row with `metric_name = 'check_metrics'` and `metric_value` containing the JSON string. You can parse it with Spark's `from_json` or `json_tuple` functions for analysis: + +```sql +SELECT + run_id, + check.check_name, + check.error_count, + check.warning_count +FROM main.analytics.dq_metrics +LATERAL VIEW explode( + from_json(metric_value, 'array>') +) AS check +WHERE metric_name = 'check_metrics' +``` ### Custom Metrics @@ -82,6 +116,7 @@ Metrics are not directly accessible from the returned Spark Observation when dat print(f"Error row count: {metrics['error_row_count']}") print(f"Warning row count: {metrics['warning_row_count']}") print(f"Valid row count: {metrics['valid_row_count']}") + print(f"Check metrics: {metrics['check_metrics']}") # per-check error/warning counts as JSON ``` @@ -368,6 +403,7 @@ Pass custom metrics as Spark SQL expressions when creating the `DQMetricsObserve print(f"Error row count: {metrics['error_row_count']}") print(f"Warning row count: {metrics['warning_row_count']}") print(f"Valid row count: {metrics['valid_row_count']}") + print(f"Check metrics: {metrics['check_metrics']}") # per-check error/warning counts as JSON print(f"Total check errors: {metrics['total_check_errors']}") print(f"Total check warnings: {metrics['total_check_warnings']}") ``` @@ -438,7 +474,7 @@ Summary metrics can be written and centralized in a delta table. The metrics tab | `checks_location` | `STRING` | Location where checks are stored (table name or file path), if known. | | `rule_set_fingerprint`| `STRING` | SHA-256 fingerprint of the rule set used for this run. Enables correlation with checks storage and results. Populated automatically when applying checks or saving results. | | `metric_name` | `STRING` | Name of the metric (e.g., 'input_row_count'). | -| `metric_value` | `STRING` | Value of the metric (stored as string). | +| `metric_value` | `STRING` | Value of the metric. All values are stored as strings — numeric metrics (e.g. `input_row_count`) are string-encoded integers, and `check_metrics` is a JSON string: `[{"check_name": "...", "error_count": N, "warning_count": N}, ...]`. Cast to the appropriate type when querying (e.g. `CAST(metric_value AS INT)` for counts, or parse JSON for `check_metrics`). | | `run_time` | `TIMESTAMP` | Run timestamp when the summary metrics were calculated. | | `error_column_name` | `STRING` | Name of the error column in the output or quarantine table containing per row quality checking results (default: '_errors'). | | `warning_column_name` | `STRING` | Name of the warning column in the output or quarantine table containing per row quality checking results (default: '_warnings'). | diff --git a/docs/dqx/docs/reference/table_schemas.mdx b/docs/dqx/docs/reference/table_schemas.mdx index 6a2cfc38f..88fa95194 100644 --- a/docs/dqx/docs/reference/table_schemas.mdx +++ b/docs/dqx/docs/reference/table_schemas.mdx @@ -115,8 +115,8 @@ Summary metrics are optional. When enabled (see [Summary Metrics](/docs/guide/su | `quarantine_location` | STRING | Quarantine table location, if used. | | `checks_location` | STRING | Where checks were loaded from (table or file). | | `rule_set_fingerprint` | STRING | SHA-256 hash of the rule set; links to checks table. | -| `metric_name` | STRING | Metric name (e.g. `input_row_count`, `error_row_count`). | -| `metric_value` | STRING | Metric value (stored as string). | +| `metric_name` | STRING | Metric name (e.g. `input_row_count`, `error_row_count`, `check_metrics`). | +| `metric_value` | STRING | All values are stored as strings. Numeric metrics are string-encoded integers (cast with `CAST(metric_value AS INT)`). For `check_metrics`, this is a JSON string: `[{"check_name": "...", "error_count": N, "warning_count": N}, ...]`. | | `run_time` | TIMESTAMP | When the run completed / check applied. | | `error_column_name` | STRING | Name of the error column (default: `_errors`). | | `warning_column_name` | STRING | Name of the warning column (default: `_warnings`). | diff --git a/pyproject.toml b/pyproject.toml index edbf17094..e91a8c028 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -95,9 +95,18 @@ requires = [ ] build-backend = "hatchling.build" -[tool.hatch.build] +# Tell hatchling where the package lives and how to lay it out in the wheel. +# Hatchling's auto-detect looks for `src/databricks_labs_dqx/` (project name) and finds nothing — +# this is a PEP 420 namespace package at `src/databricks/labs/dqx/`. +# `packages` points it at the namespace root; `sources` strips `src/` so imports resolve as +# `databricks.labs.dqx.*` in the installed wheel +[tool.hatch.build.targets.wheel] sources = ["src"] -include = ["src", "README.md", "LICENSE", "NOTICE"] +packages = ["src/databricks"] + +# Restrict the sdist to source code and top-level metadata +[tool.hatch.build.targets.sdist] +only-include = ["src", "README.md", "LICENSE", "NOTICE"] [tool.hatch.version] path = "src/databricks/labs/dqx/__about__.py" diff --git a/src/databricks/labs/dqx/engine.py b/src/databricks/labs/dqx/engine.py index 1982e3e4e..e48d78aa0 100644 --- a/src/databricks/labs/dqx/engine.py +++ b/src/databricks/labs/dqx/engine.py @@ -165,7 +165,11 @@ def apply_checks( ref_dfs, rule_set_fingerprint=rule_set_fingerprint, ) - observed_result = self._observe_metrics(result_df) + + # Duplicate check names are intentionally preserved — if two rules share a name, + # check_metrics will report each occurrence separately so the user can spot the overlap. + check_names = [check.name for check in checks] + observed_result = self._observe_metrics(result_df, check_names) if isinstance(observed_result, tuple): observed_df, observation = observed_result @@ -519,12 +523,15 @@ def _create_results_array( return result_df - def _observe_metrics(self, df: DataFrame) -> DataFrame | tuple[DataFrame, Observation]: + def _observe_metrics( + self, df: DataFrame, check_names: list[str] | None = None + ) -> DataFrame | tuple[DataFrame, Observation]: """ Adds Spark observable metrics to the input DataFrame. Args: df: Input DataFrame + check_names: Optional list of check names to include per-check metrics. Returns: The unmodified DataFrame with observed metrics and the corresponding Spark Observation @@ -532,7 +539,7 @@ def _observe_metrics(self, df: DataFrame) -> DataFrame | tuple[DataFrame, Observ if not self.observer: return df - metric_exprs = [F.expr(metric_statement) for metric_statement in self.observer.metrics] + metric_exprs = [F.expr(m) for m in self.observer.get_metrics(check_names)] if not metric_exprs: return df diff --git a/src/databricks/labs/dqx/metrics_observer.py b/src/databricks/labs/dqx/metrics_observer.py index 62e51ef94..1ec156c0b 100644 --- a/src/databricks/labs/dqx/metrics_observer.py +++ b/src/databricks/labs/dqx/metrics_observer.py @@ -81,23 +81,56 @@ def id(self) -> str: """ return self.id_overwrite or str(uuid4()) - @cached_property - def metrics(self) -> list[str]: + def get_metrics(self, check_names: list[str] | None = None) -> list[str]: """ Gets the observer metrics as Spark SQL expressions. + Args: + check_names: Optional list of check names from the applied quality rules. + When provided, a per-check breakdown (``check_metrics``) is included. + Returns: - A list of Spark SQL expressions defining the observer metrics (both default and custom). + A list of Spark SQL expressions defining the observer metrics (default, per-check, and custom). """ - default_metrics = [ + metrics = [ "count(1) as input_row_count", f"count(case when {self._error_column_name} is not null then 1 end) as error_row_count", f"count(case when {self._warning_column_name} is not null then 1 end) as warning_row_count", f"count(case when {self._error_column_name} is null and {self._warning_column_name} is null then 1 end) as valid_row_count", ] + if check_names: + metrics.append(self._build_check_metrics_expr(check_names)) if self.custom_metrics: - default_metrics.extend(self.custom_metrics) - return default_metrics + metrics.extend(self.custom_metrics) + return metrics + + def _build_check_metrics_expr(self, check_names: list[str]) -> str: + """Build a single SQL expression that produces a JSON-serialized check_metrics value. + + The metric is stored as one row with metric_name ``check_metrics`` and + metric_value containing a JSON array of structs:: + + [{"check_name": "my_check", "error_count": 5, "warning_count": 2}, ...] + + Args: + check_names: List of check names to include in the expression. + + Returns: + A Spark SQL expression string aliased as ``check_metrics``. + """ + elements: list[str] = [] + for check_name in check_names: + check_name_escaped = check_name.replace("'", "''") + err = self._error_column_name + warn = self._warning_column_name + elements.append( + f"named_struct(" + f"'check_name', '{check_name_escaped}', " + f"'error_count', count(case when exists({err}, x -> x.name = '{check_name_escaped}') then 1 end), " + f"'warning_count', count(case when exists({warn}, x -> x.name = '{check_name_escaped}') then 1 end)" + f")" + ) + return f"to_json(array({', '.join(elements)})) as check_metrics" @property def observation(self) -> Observation: diff --git a/tests/conftest.py b/tests/conftest.py index f1af6c9cd..187ccf5a5 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2,8 +2,6 @@ import logging import os import re -import subprocess -import sys from collections.abc import Callable, Generator from pathlib import Path from dataclasses import dataclass, replace @@ -40,13 +38,13 @@ logger = logging.getLogger(__name__) -class VerboseWheels(WheelsV2): - """Test-only WheelsV2 that captures `pip wheel` output and logs it on failure. +class PrebuiltWheels(WheelsV2): + """Test-only WheelsV2 that copies a pre-built wheel when `DQX_PREBUILT_WHEEL` is set. - Upstream's `verbose=True` sets stdout/stderr to `subprocess.STDOUT` which is invalid - as a stdout value (raises OSError: Bad file descriptor). This override uses pipes so - the real pip error surfaces in CI logs — which is what we need to diagnose root - causes of transient wheel build failures (e.g. PyPI mirror issues). + CI builds the wheel once via the prebuild-wheel action while the JFrog token is fresh, + then points each pytest invocation at that artifact so per-test `pip wheel` calls do + not fail after the 1h OIDC token TTL expires mid-suite. Local development falls through + to upstream's standard `pip wheel` behaviour. """ def _build_wheel( @@ -57,20 +55,12 @@ def _build_wheel( no_deps: bool = True, dirs_exist_ok: bool = False, ): - del verbose # always capture in tests; fresh pipes avoid fd inheritance issues - checkout_root = self._product_info.checkout_root() - if self._product_info.is_git_checkout() and self._product_info.is_unreleased_version(): - checkout_root = self._copy_root_to(tmp_dir, dirs_exist_ok=dirs_exist_ok) - self._override_version_to_unreleased(checkout_root) - args = [sys.executable, "-m", "pip", "wheel", "--wheel-dir", tmp_dir, checkout_root.as_posix()] - if no_deps: - args.append("--no-deps") - result = subprocess.run(args, capture_output=True, text=True, check=False) - if result.returncode != 0: - logger.error( - f"pip wheel failed (rc={result.returncode}); stdout={result.stdout!r}; stderr={result.stderr!r}" - ) - raise subprocess.CalledProcessError(result.returncode, args, output=result.stdout, stderr=result.stderr) + prebuilt = os.environ.get("DQX_PREBUILT_WHEEL") + if not prebuilt: + return super()._build_wheel(tmp_dir, verbose=verbose, no_deps=no_deps, dirs_exist_ok=dirs_exist_ok) + src = Path(prebuilt) + dst = Path(tmp_dir) / src.name + dst.write_bytes(src.read_bytes()) return Path(tmp_dir).glob("*.whl") @@ -94,7 +84,7 @@ def get_schema_validation_rules(rules: list[dict[str, Any]]) -> list[dict[str, A @pytest.fixture(scope="session") def debug_env_name(): - return "ws2" # Specify the name of the debug environment from ~/.databricks/debug-env.json + return "ws" # Specify the name of the debug environment from ~/.databricks/debug-env.json @pytest.fixture @@ -300,7 +290,7 @@ def workflows_deployment(self) -> WorkflowDeployment: self.installation, self.install_state, self.workspace_client, - VerboseWheels(self.installation, self.product_info), + PrebuiltWheels(self.installation, self.product_info), self.product_info, self.tasks, ) diff --git a/tests/e2e/notebooks/save_dataframe_as_table_notebook.py b/tests/e2e/notebooks/save_dataframe_as_table_notebook.py index 52cda3f46..22e5046ed 100644 --- a/tests/e2e/notebooks/save_dataframe_as_table_notebook.py +++ b/tests/e2e/notebooks/save_dataframe_as_table_notebook.py @@ -60,7 +60,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by_missing_spark_config( logger.removeHandler(handler) result_df = spark.table(output_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) test_save_streaming_dataframe_in_table_with_cluster_by_missing_spark_config() @@ -82,7 +82,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by(): save_dataframe_as_table(streaming_input_df, output_config).awaitTermination() result_df = spark.table(output_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) table_detail = spark.sql(f"DESCRIBE DETAIL {output_table_name}").collect()[0] assert table_detail["clusteringColumns"] == ["a"] diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 8b46c4bb8..745402411 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1,9 +1,8 @@ import logging import os import threading -from datetime import datetime, timedelta, timezone +from datetime import datetime, timezone from io import BytesIO -from subprocess import CalledProcessError from typing import Any from unittest.mock import patch from pyspark.sql import DataFrame, functions as F @@ -11,7 +10,6 @@ from chispa import assert_df_equality as _chispa_assert_df_equality # type: ignore import pytest -from databricks.sdk.retries import retried from databricks.sdk.service.workspace import ImportFormat from databricks.labs.blueprint.installation import Installation from databricks.labs.pytester.fixtures.baseline import factory @@ -43,11 +41,6 @@ FINGERPRINT_FIELDS = ["rule_fingerprint", "rule_set_fingerprint"] -def _install_with_retry(installation_service) -> None: - # WheelsV2 shells out to `pip wheel`; retry transient CalledProcessError (network / PyPI hiccups). - retried(on=[CalledProcessError], timeout=timedelta(minutes=5))(installation_service.run)() - - def _strip_fingerprints_from_result_column(df: DataFrame, col_name: str) -> DataFrame: """Remove fingerprint fields from a result array column (_errors or _warnings).""" @@ -203,7 +196,7 @@ def setup_workflows(ws, spark, installation_ctx, make_schema, make_table, make_r pytest.skip() def create(_spark, **kwargs): - _install_with_retry(installation_ctx.installation_service) + installation_ctx.installation_service.run() quarantine = False if "quarantine" in kwargs and kwargs["quarantine"]: @@ -236,7 +229,7 @@ def setup_serverless_workflows(ws, spark, serverless_installation_ctx, make_sche pytest.skip() def create(_spark, **kwargs): - _install_with_retry(serverless_installation_ctx.installation_service) + serverless_installation_ctx.installation_service.run() quarantine = False if "quarantine" in kwargs and kwargs["quarantine"]: @@ -283,7 +276,7 @@ def create(_spark, **kwargs): installation_ctx.config.profiler_override_clusters["default"] = cluster_id installation_ctx.config.quality_checker_override_clusters["default"] = cluster_id installation_ctx.config.e2e_override_clusters["default"] = cluster_id - _install_with_retry(installation_ctx.installation_service) + installation_ctx.installation_service.run() quarantine = False if "quarantine" in kwargs and kwargs["quarantine"]: @@ -341,7 +334,7 @@ def setup_workflows_with_custom_folder( pytest.skip() def create(_spark, **kwargs): - _install_with_retry(installation_ctx_custom_install_folder.installation_service) + installation_ctx_custom_install_folder.installation_service.run() quarantine = False if "quarantine" in kwargs and kwargs["quarantine"]: diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index e3ed75034..02d632982 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -5675,7 +5675,7 @@ def test_apply_checks_all_row_checks_as_yaml_with_streaming(ws, make_schema, mak expected_schema, ) - assert_df_equality(checked_df, expected, ignore_nullable=True) + assert_df_equality(checked_df.sort("col2"), expected.sort("col2"), ignore_nullable=True) def test_apply_checks_all_row_geo_checks_as_yaml_with_streaming( @@ -5795,7 +5795,7 @@ def test_apply_checks_all_row_geo_checks_as_yaml_with_streaming( expected_schema, ) - assert_df_equality(checked_df, expected, ignore_nullable=True) + assert_df_equality(checked_df.sort("col3"), expected.sort("col3"), ignore_nullable=True) def test_apply_checks_all_checks_as_yaml(ws, spark): diff --git a/tests/integration/test_apply_checks_and_save_in_table.py b/tests/integration/test_apply_checks_and_save_in_table.py index 3f0514b92..efad921b9 100644 --- a/tests/integration/test_apply_checks_and_save_in_table.py +++ b/tests/integration/test_apply_checks_and_save_in_table.py @@ -764,7 +764,7 @@ def test_apply_checks_and_save_in_table_streaming_write(ws, spark, make_schema, ], schema=expected_schema, ) - assert_df_equality(actual_df, expected_df, ignore_nullable=True) + assert_df_equality(actual_df.sort("a"), expected_df.sort("a"), ignore_nullable=True) def test_apply_checks_and_save_in_tables(ws, spark, make_schema, make_random, make_directory): @@ -933,7 +933,7 @@ def test_apply_checks_and_save_in_tables_streaming_write( ], schema=expected_schema, ) - assert_df_equality(actual_df, expected_df, ignore_nullable=True) + assert_df_equality(actual_df.sort("a"), expected_df.sort("a"), ignore_nullable=True) def test_apply_checks_and_save_in_tables_multiple_tables(ws, spark, make_schema, make_random, make_directory): @@ -1697,10 +1697,14 @@ def test_apply_checks_and_save_in_tables_for_patterns_with_quarantine( schema=expected_schema, ) - assert_df_equality(spark.table(output_tables[0]), expected_valid_df1, ignore_nullable=True) - assert_df_equality(spark.table(output_tables[1]), expected_valid_df2, ignore_nullable=True) - assert_df_equality(spark.table(quarantine_tables[0]), expected_quarantine_df1, ignore_nullable=True) - assert_df_equality(spark.table(quarantine_tables[1]), expected_quarantine_df2, ignore_nullable=True) + assert_df_equality(spark.table(output_tables[0]).sort("a"), expected_valid_df1.sort("a"), ignore_nullable=True) + assert_df_equality(spark.table(output_tables[1]).sort("a"), expected_valid_df2.sort("a"), ignore_nullable=True) + assert_df_equality( + spark.table(quarantine_tables[0]).sort("a"), expected_quarantine_df1.sort("a"), ignore_nullable=True + ) + assert_df_equality( + spark.table(quarantine_tables[1]).sort("a"), expected_quarantine_df2.sort("a"), ignore_nullable=True + ) def test_apply_checks_and_save_in_tables_for_patterns_with_exclude_patterns( @@ -1790,8 +1794,8 @@ def test_apply_checks_and_save_in_tables_for_patterns_with_exclude_patterns( schema=expected_schema, ) - assert_df_equality(spark.table(output_table), expected_valid_df, ignore_nullable=True) - assert_df_equality(spark.table(quarantine_table), expected_quarantine_df, ignore_nullable=True) + assert_df_equality(spark.table(output_table).sort("a"), expected_valid_df.sort("a"), ignore_nullable=True) + assert_df_equality(spark.table(quarantine_table).sort("a"), expected_quarantine_df.sort("a"), ignore_nullable=True) # 1 input + 1 output + 1 quarantine + 2 existing tables = ws.tables.list_summaries(catalog_name=catalog_name, schema_name_pattern=schema.name) assert len(list(tables)) == 5, "Tables count mismatch" @@ -1964,10 +1968,14 @@ def test_apply_checks_and_save_in_tables_for_patterns_with_custom_suffix( schema=expected_schema, ) - assert_df_equality(spark.table(output_tables[0]), expected_valid_df1, ignore_nullable=True) - assert_df_equality(spark.table(output_tables[1]), expected_valid_df2, ignore_nullable=True) - assert_df_equality(spark.table(quarantine_tables[0]), expected_quarantine_df1, ignore_nullable=True) - assert_df_equality(spark.table(quarantine_tables[1]), expected_quarantine_df2, ignore_nullable=True) + assert_df_equality(spark.table(output_tables[0]).sort("a"), expected_valid_df1.sort("a"), ignore_nullable=True) + assert_df_equality(spark.table(output_tables[1]).sort("a"), expected_valid_df2.sort("a"), ignore_nullable=True) + assert_df_equality( + spark.table(quarantine_tables[0]).sort("a"), expected_quarantine_df1.sort("a"), ignore_nullable=True + ) + assert_df_equality( + spark.table(quarantine_tables[1]).sort("a"), expected_quarantine_df2.sort("a"), ignore_nullable=True + ) def test_apply_checks_and_save_in_tables_with_patterns_and_custom_functions( diff --git a/tests/integration/test_io.py b/tests/integration/test_io.py index 0cce340d2..8b78c8e4c 100644 --- a/tests/integration/test_io.py +++ b/tests/integration/test_io.py @@ -48,7 +48,7 @@ def test_read_input_data_no_input_format(spark, make_schema, make_volume): input_config = InputConfig(location=input_location) actual_df = read_input_data(spark, input_config) - assert_df_equality(actual_df, input_df) + assert_df_equality(actual_df, input_df, ignore_row_order=True) def test_read_invalid_input_location(spark): @@ -84,7 +84,7 @@ def test_read_input_data_from_table(spark, make_schema, make_random): input_df.write.format("delta").saveAsTable(input_location) result_df = read_input_data(spark, input_config) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_read_input_data_from_table_with_schema_and_spark_options(spark, make_schema, make_random): @@ -105,7 +105,7 @@ def test_read_input_data_from_table_with_schema_and_spark_options(spark, make_sc input_df_ver1.write.format("delta").insertInto(input_location) result_df = read_input_data(spark, input_config) - assert_df_equality(input_df_ver0, result_df) + assert_df_equality(input_df_ver0, result_df, ignore_row_order=True) def test_read_input_data_from_workspace_file(spark, make_schema, make_volume): @@ -121,7 +121,7 @@ def test_read_input_data_from_workspace_file(spark, make_schema, make_volume): input_df.write.format("delta").saveAsTable(input_location) result_df = read_input_data(spark, input_config) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_read_input_data_from_workspace_file_with_spark_options(spark, make_schema, make_volume): @@ -140,7 +140,7 @@ def test_read_input_data_from_workspace_file_with_spark_options(spark, make_sche input_df_ver1.write.format("delta").insertInto(input_location) result_df = read_input_data(spark, input_config) - assert_df_equality(input_df_ver0, result_df) + assert_df_equality(input_df_ver0, result_df, ignore_row_order=True) def test_read_input_data_from_workspace_file_in_csv_format(spark, make_schema, make_volume): @@ -162,7 +162,7 @@ def test_read_input_data_from_workspace_file_in_csv_format(spark, make_schema, m result_df = read_input_data(spark, input_config) - assert_df_equality(input_df_ver0, result_df) + assert_df_equality(input_df_ver0, result_df, ignore_row_order=True) def test_save_dataframe_as_table(spark, make_schema, make_random): @@ -177,7 +177,7 @@ def test_save_dataframe_as_table(spark, make_schema, make_random): save_dataframe_as_table(input_df, output_config) result_df = spark.table(table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) output_config.mode = "append" output_config.options = {"mergeSchema": "true"} @@ -203,7 +203,7 @@ def test_save_dataframe_as_table_with_partition_by(spark, make_schema, make_rand save_dataframe_as_table(input_df, output_config) result_df = spark.table(table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) table_detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0] assert table_detail["partitionColumns"] == ["a"] @@ -220,7 +220,7 @@ def test_save_dataframe_as_table_with_cluster_by(spark, make_schema, make_random save_dataframe_as_table(input_df, output_config) result_df = spark.table(table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) table_detail = spark.sql(f"DESCRIBE DETAIL {table_name}").collect()[0] assert table_detail["clusteringColumns"] == ["a"] @@ -245,12 +245,12 @@ def test_save_streaming_dataframe_in_table(spark, make_schema, make_random, make save_dataframe_as_table(streaming_input_df, output_config).awaitTermination() result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) save_dataframe_as_table(streaming_input_df, output_config).awaitTermination() result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) # no new records + assert_df_equality(input_df, result_df, ignore_row_order=True) # no new records def test_save_streaming_dataframe_in_table_with_partition_by(spark, make_schema, make_random, make_volume): @@ -272,7 +272,7 @@ def test_save_streaming_dataframe_in_table_with_partition_by(spark, make_schema, save_dataframe_as_table(streaming_input_df, output_config).awaitTermination() result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) table_detail = spark.sql(f"DESCRIBE DETAIL {result_table_name}").collect()[0] assert table_detail["partitionColumns"] == ["a"] @@ -312,7 +312,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by_missing_env_var( ) result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_save_streaming_dataframe_in_table_with_cluster_by_invalid_env_var( @@ -351,7 +351,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by_invalid_env_var( ) result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_save_streaming_dataframe_in_table_with_cluster_by_serverless_env( @@ -392,7 +392,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by_serverless_env( ) result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_save_streaming_dataframe_in_table_with_cluster_by_unsupported_env( @@ -433,7 +433,7 @@ def test_save_streaming_dataframe_in_table_with_cluster_by_unsupported_env( ) result_df = spark.table(result_table_name) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_save_batch_dataframe_to_path(spark, make_schema, make_volume, make_random): @@ -481,7 +481,7 @@ def test_save_streaming_dataframe_to_path(spark, make_schema, make_volume, make_ query.awaitTermination() result_df = spark.read.format("delta").load(volume_path) - assert_df_equality(input_df, result_df) + assert_df_equality(input_df, result_df, ignore_row_order=True) def test_save_dataframe_invalid_location(spark): diff --git a/tests/integration/test_quality_checker_workflow_with_metrics.py b/tests/integration/test_quality_checker_workflow_with_metrics.py index 0c712a219..790cc0a86 100644 --- a/tests/integration/test_quality_checker_workflow_with_metrics.py +++ b/tests/integration/test_quality_checker_workflow_with_metrics.py @@ -15,6 +15,10 @@ ) _WORKFLOW_RULE_SET_FINGERPRINT = compute_rule_set_fingerprint_by_metadata(WORKFLOW_CHECKS) +_WORKFLOW_CHECK_METRICS_VALUE = ( + '[{"check_name":"id_is_not_null","error_count":1,"warning_count":0},' + '{"check_name":"name_is_not_null_and_not_empty","error_count":2,"warning_count":0}]' +) def test_quality_checker_workflow_with_metrics(spark, setup_workflows_with_metrics, expected_quality_checking_output): @@ -84,6 +88,21 @@ def test_quality_checker_workflow_with_metrics(spark, setup_workflows_with_metri "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": RUN_ID, + "run_name": "dqx", + "input_location": run_config.input_config.location, + "output_location": run_config.output_config.location, + "quarantine_location": None, + "checks_location": checks_location, + "rule_set_fingerprint": _WORKFLOW_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": _WORKFLOW_CHECK_METRICS_VALUE, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -180,6 +199,21 @@ def test_quality_checker_workflow_with_quarantine_and_metrics( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": RUN_ID, + "run_name": "dqx", + "input_location": run_config.input_config.location, + "output_location": run_config.output_config.location, + "quarantine_location": run_config.quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": _WORKFLOW_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": _WORKFLOW_CHECK_METRICS_VALUE, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -257,7 +291,7 @@ def test_e2e_workflow_with_metrics(ws, spark, setup_workflows_with_metrics, expe ) actual_metrics_df = ( spark.table(run_config.metrics_config.location) - .where("metric_name NOT IN ('error_row_count', 'warning_row_count', 'valid_row_count')") + .where("metric_name NOT IN ('error_row_count', 'warning_row_count', 'valid_row_count', 'check_metrics')") .orderBy("metric_name") ) assert_df_equality(expected_metrics_df, actual_metrics_df) @@ -379,7 +413,7 @@ def test_custom_metrics_in_workflow_for_all_run_configs( ) actual_metrics_df = ( spark.table(run_config.metrics_config.location) - .where("metric_name NOT IN ('error_row_count', 'warning_row_count', 'valid_row_count')") + .where("metric_name NOT IN ('error_row_count', 'warning_row_count', 'valid_row_count', 'check_metrics')") .orderBy("metric_name") ) assert_df_equality(expected_metrics_df, actual_metrics_df) @@ -476,6 +510,21 @@ def test_quality_checker_workflow_with_streaming_quarantine_and_metrics( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": RUN_ID, + "run_name": "dqx", + "input_location": run_config.input_config.location, + "output_location": run_config.output_config.location, + "quarantine_location": run_config.quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": _WORKFLOW_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": _WORKFLOW_CHECK_METRICS_VALUE, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = ( @@ -582,6 +631,21 @@ def test_quality_checker_workflow_with_continuous_streaming_quarantine_and_metri "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": RUN_ID, + "run_name": "dqx", + "input_location": run_config.input_config.location, + "output_location": run_config.output_config.location, + "quarantine_location": run_config.quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": _WORKFLOW_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": _WORKFLOW_CHECK_METRICS_VALUE, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = ( @@ -684,6 +748,21 @@ def test_quality_checker_workflow_with_quarantine_and_metrics_for_patterns( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": RUN_ID, + "run_name": "dqx", + "input_location": run_config.input_config.location, + "output_location": output_location, + "quarantine_location": quarantine_location, + "checks_location": checks_location, + "rule_set_fingerprint": _WORKFLOW_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": _WORKFLOW_CHECK_METRICS_VALUE, + "run_time": RUN_TIME, + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( diff --git a/tests/integration/test_save_and_load_checks_from_table.py b/tests/integration/test_save_and_load_checks_from_table.py index 06794226f..745f920a3 100644 --- a/tests/integration/test_save_and_load_checks_from_table.py +++ b/tests/integration/test_save_and_load_checks_from_table.py @@ -164,7 +164,7 @@ def test_save_checks_to_table_with_unresolved_for_each_column(ws, make_schema, m expected_checks_df = spark.createDataFrame(expected_raw_checks, TEST_CHECKS_TABLE_SCHEMA) - assert_df_equality(checks_df, expected_checks_df, ignore_nullable=True) + assert_df_equality(checks_df.sort("name"), expected_checks_df.sort("name"), ignore_nullable=True) def test_load_checks_from_table_saved_from_dict_with_unresolved_for_each_column(ws, make_schema, make_random, spark): diff --git a/tests/integration/test_save_results_in_table.py b/tests/integration/test_save_results_in_table.py index 71b0838bb..49a25c0e6 100644 --- a/tests/integration/test_save_results_in_table.py +++ b/tests/integration/test_save_results_in_table.py @@ -34,8 +34,8 @@ def test_save_results_in_table(ws, spark, make_schema, make_random): output_df_loaded = spark.table(output_table) quarantine_df_loaded = spark.table(quarantine_table) - assert_df_equality(output_df, output_df_loaded) - assert_df_equality(quarantine_df, quarantine_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df, quarantine_df_loaded, ignore_row_order=True) output_config.mode = "append" output_config.options = {"overwriteSchema": "true"} @@ -49,8 +49,8 @@ def test_save_results_in_table(ws, spark, make_schema, make_random): quarantine_config=quarantine_config, ) - assert_df_equality(output_df.union(output_df), output_df_loaded) - assert_df_equality(quarantine_df.union(quarantine_df), quarantine_df_loaded) + assert_df_equality(output_df.union(output_df), output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df.union(quarantine_df), quarantine_df_loaded, ignore_row_order=True) def test_save_results_in_table_only_output(ws, spark, make_schema, make_random): @@ -71,7 +71,7 @@ def test_save_results_in_table_only_output(ws, spark, make_schema, make_random): output_df_loaded = spark.table(output_table) - assert_df_equality(output_df, output_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) def test_save_results_in_table_only_quarantine(ws, spark, make_schema, make_random): @@ -120,8 +120,8 @@ def test_save_results_in_table_in_user_installation(ws, spark, installation_ctx, output_df_loaded = spark.table(output_table) quarantine_df_loaded = spark.table(quarantine_table) - assert_df_equality(output_df, output_df_loaded) - assert_df_equality(quarantine_df, quarantine_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df, quarantine_df_loaded, ignore_row_order=True) def test_save_results_in_table_in_user_installation_only_output(ws, spark, installation_ctx, make_schema, make_random): @@ -147,7 +147,7 @@ def test_save_results_in_table_in_user_installation_only_output(ws, spark, insta ) output_df_loaded = spark.table(output_table) - assert_df_equality(output_df, output_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) def test_save_results_in_table_in_user_installation_only_quarantine( @@ -209,8 +209,8 @@ def test_save_results_in_table_in_user_installation_output_table_provided( output_df_loaded = spark.table(output_table) quarantine_df_loaded = spark.table(quarantine_table) - assert_df_equality(output_df, output_df_loaded) - assert_df_equality(quarantine_df, quarantine_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df, quarantine_df_loaded, ignore_row_order=True) def test_save_results_in_table_in_user_installation_quarantine_table_provided( @@ -244,8 +244,8 @@ def test_save_results_in_table_in_user_installation_quarantine_table_provided( output_df_loaded = spark.table(output_table) quarantine_df_loaded = spark.table(quarantine_table) - assert_df_equality(output_df, output_df_loaded) - assert_df_equality(quarantine_df, quarantine_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df, quarantine_df_loaded, ignore_row_order=True) def test_save_results_in_table_in_user_installation_missing_output_and_quarantine_table( @@ -326,8 +326,8 @@ def test_save_results_in_table_in_custom_folder_installation( output_df_loaded = spark.table(output_table) quarantine_df_loaded = spark.table(quarantine_table) - assert_df_equality(output_df, output_df_loaded) - assert_df_equality(quarantine_df, quarantine_df_loaded) + assert_df_equality(output_df, output_df_loaded, ignore_row_order=True) + assert_df_equality(quarantine_df, quarantine_df_loaded, ignore_row_order=True) def test_save_streaming_results_in_table(ws, spark, make_schema, make_random, make_volume): @@ -359,4 +359,4 @@ def test_save_streaming_results_in_table(ws, spark, make_schema, make_random, ma ) output_df_loaded = spark.table(output_table) - assert_df_equality(input_df, output_df_loaded) + assert_df_equality(input_df, output_df_loaded, ignore_row_order=True) diff --git a/tests/integration/test_summary_metrics.py b/tests/integration/test_summary_metrics.py index 9805aa6cf..d2e32a6df 100644 --- a/tests/integration/test_summary_metrics.py +++ b/tests/integration/test_summary_metrics.py @@ -1,3 +1,4 @@ +import json import time from datetime import datetime @@ -39,6 +40,12 @@ ] TEST_CHECKS_RULE_SET_FINGERPRINT = compute_rule_set_fingerprint_by_metadata(TEST_CHECKS) TEST_OBSERVER_NAME = "test_observer" +# Expected check_metrics JSON value for TEST_CHECKS with standard 4-row test data +# (row 3 has id=None → error, row 4 has name=None → warning) +TEST_CHECK_METRICS_VALUE = ( + '[{"check_name":"id_is_not_null","error_count":1,"warning_count":0},' + '{"check_name":"name_is_not_null_and_not_empty","error_count":0,"warning_count":1}]' +) def test_observer_custom_column_names(ws, spark): @@ -54,11 +61,12 @@ def test_observer_custom_column_names(ws, spark): observer = DQMetricsObserver(name="test_observer") _ = DQEngine(workspace_client=ws, spark=spark, extra_params=engine_params, observer=observer) - assert f"count(case when {errors_column} is not null then 1 end) as error_row_count" in observer.metrics - assert f"count(case when {warnings_column} is not null then 1 end) as warning_row_count" in observer.metrics + metrics = observer.get_metrics() + assert f"count(case when {errors_column} is not null then 1 end) as error_row_count" in metrics + assert f"count(case when {warnings_column} is not null then 1 end) as warning_row_count" in metrics assert ( f"count(case when {errors_column} is null and {warnings_column} is null then 1 end) as valid_row_count" - in observer.metrics + in metrics ) @@ -80,14 +88,25 @@ def test_observer_metrics_before_action(ws, spark, apply_checks_method): if apply_checks_method == DQEngine.apply_checks: checks = deserialize_checks(TEST_CHECKS) - _, observation = dq_engine.apply_checks(test_df, checks) + checked_df, observation = dq_engine.apply_checks(test_df, checks) elif apply_checks_method == DQEngine.apply_checks_by_metadata: - _, observation = dq_engine.apply_checks_by_metadata(test_df, TEST_CHECKS) + checked_df, observation = dq_engine.apply_checks_by_metadata(test_df, TEST_CHECKS) else: raise ValueError("Invalid 'apply_checks_method' used for testing observable metrics.") - actual_metrics = observation.get - assert actual_metrics == {} + # Read metrics BEFORE any action — must be empty. + assert observation.get == {} + # Trigger the action so Spark Connect can complete the observation's lifecycle and + # release server-side state. Leaving the attached DataFrame GC'd without executing + # it has been observed to destabilize the shared session for subsequent tests. + checked_df.count() + assert observation.get == { + "input_row_count": 4, + "error_row_count": 1, + "warning_row_count": 1, + "valid_row_count": 2, + "check_metrics": TEST_CHECK_METRICS_VALUE, + } @pytest.mark.parametrize("apply_checks_method", [DQEngine.apply_checks, DQEngine.apply_checks_by_metadata]) @@ -115,14 +134,14 @@ def test_observer_metrics(ws, spark, apply_checks_method): raise ValueError("Invalid 'apply_checks_method' used for testing observable metrics.") checked_df.count() # Trigger an action to get the metrics - expected_metrics = { + actual_metrics = observation.get + assert actual_metrics == { "input_row_count": 4, "error_row_count": 1, "warning_row_count": 1, "valid_row_count": 2, + "check_metrics": TEST_CHECK_METRICS_VALUE, } - actual_metrics = observation.get - assert actual_metrics == expected_metrics @pytest.mark.parametrize("apply_checks_method", [DQEngine.apply_checks, DQEngine.apply_checks_by_metadata]) @@ -188,16 +207,16 @@ def test_observer_custom_metrics(ws, spark, apply_checks_method): raise ValueError("Invalid 'apply_checks_method' used for testing observable metrics.") checked_df.count() # Trigger an action to get the metrics - expected_metrics = { + actual_metrics = observation.get + assert actual_metrics == { "input_row_count": 4, "error_row_count": 1, "warning_row_count": 1, "valid_row_count": 2, + "check_metrics": TEST_CHECK_METRICS_VALUE, "avg_error_age": 35.0, "total_warning_salary": 55000, } - actual_metrics = observation.get - assert actual_metrics == expected_metrics @pytest.mark.parametrize( @@ -283,6 +302,21 @@ def test_save_summary_metrics(ws, spark, make_schema, make_random): actual_metrics_df = spark.table(metrics_config.location).orderBy("metric_name") expected_metrics = [ + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": observer_name, + "input_location": input_config.location, + "output_location": output_config.location, + "quarantine_location": quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": None, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, { "run_id": EXTRA_PARAMS.run_id_overwrite, "run_name": observer_name, @@ -497,6 +531,21 @@ def test_save_summary_metrics_custom_metrics_and_params(ws, spark_keep_alive, ma "warning_column_name": "dq_warnings", "user_metadata": user_metadata, }, + { + "run_id": extra_params_custom.run_id_overwrite, + "run_name": observer_name, + "input_location": input_config.location, + "output_location": output_config.location, + "quarantine_location": quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": None, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(extra_params_custom.run_time_overwrite), + "error_column_name": "dq_errors", + "warning_column_name": "dq_warnings", + "user_metadata": user_metadata, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -677,6 +726,21 @@ def test_save_summary_metrics_with_streaming_and_custom_params(ws, spark, make_s "warning_column_name": "dq_warnings", "user_metadata": user_metadata, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": TEST_OBSERVER_NAME, + "input_location": input_config.location, + "output_location": output_config.location, + "quarantine_location": quarantine_config.location, + "checks_location": checks_location, + "rule_set_fingerprint": None, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "dq_errors", + "warning_column_name": "dq_warnings", + "user_metadata": user_metadata, + }, ] expected_metrics_df = ( @@ -1143,6 +1207,21 @@ def test_observer_metrics_output(skip_if_classic_compute, apply_checks_method, s "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_observer", + "input_location": input_table_name, + "output_location": output_table_name, + "quarantine_location": None, + "checks_location": checks_location, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -1307,6 +1386,21 @@ def test_observer_metrics_output_with_quarantine( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_observer", + "input_location": input_table_name, + "output_location": output_table_name, + "quarantine_location": quarantine_table_name, + "checks_location": checks_location, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -1432,6 +1526,21 @@ def test_save_results_in_table_batch_with_metrics( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_save_batch_observer", + "input_location": None, + "output_location": output_table_name, + "quarantine_location": quarantine_table_name, + "checks_location": None, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -1702,6 +1811,21 @@ def test_save_results_in_table_streaming_with_metrics( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_save_batch_observer", + "input_location": None, + "output_location": output_config.location, + "quarantine_location": quarantine_config.location, + "checks_location": None, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = ( @@ -1876,6 +2000,21 @@ def test_streaming_observer_metrics_output(apply_checks_method, spark, ws, make_ "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_streaming_observer", + "input_location": input_config.location, + "output_location": output_config.location, + "quarantine_location": None, + "checks_location": checks_location, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = spark.createDataFrame(expected_metrics, schema=OBSERVATION_TABLE_SCHEMA).orderBy( @@ -2051,6 +2190,21 @@ def test_streaming_observer_metrics_output_and_quarantine( "warning_column_name": "_warnings", "user_metadata": None, }, + { + "run_id": EXTRA_PARAMS.run_id_overwrite, + "run_name": "test_streaming_observer_with_quarantine", + "input_location": input_table_name, + "output_location": output_table_name, + "quarantine_location": quarantine_table_name, + "checks_location": checks_location, + "rule_set_fingerprint": TEST_CHECKS_RULE_SET_FINGERPRINT, + "metric_name": "check_metrics", + "metric_value": TEST_CHECK_METRICS_VALUE, + "run_time": datetime.fromisoformat(EXTRA_PARAMS.run_time_overwrite), + "error_column_name": "_errors", + "warning_column_name": "_warnings", + "user_metadata": None, + }, ] expected_metrics_df = ( @@ -2408,3 +2562,126 @@ def test_streaming_observer_metrics_output_and_quarantine_with_empty_checks( assert ( spark.table(quarantine_config.location).count() == 0 ), f"Quarantine table {quarantine_config.location} has {spark.table(quarantine_config.location).count()} rows" + + +@pytest.mark.parametrize("apply_checks_method", [DQEngine.apply_checks, DQEngine.apply_checks_by_metadata]) +def test_observer_check_metrics(ws, spark, apply_checks_method): + """Test that per-check metrics are included as a compact JSON check_metrics value.""" + observer = DQMetricsObserver(name="test_observer") + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + test_df = spark.createDataFrame( + [ + [1, "Alice", 30, 50000], + [2, "Bob", 25, 45000], + [None, "Charlie", 35, 60000], + [4, None, 28, 55000], + ], + TEST_SCHEMA, + ) + + if apply_checks_method == DQEngine.apply_checks: + checks = deserialize_checks(TEST_CHECKS) + checked_df, observation = dq_engine.apply_checks(test_df, checks) + elif apply_checks_method == DQEngine.apply_checks_by_metadata: + checked_df, observation = dq_engine.apply_checks_by_metadata(test_df, TEST_CHECKS) + else: + raise ValueError("Invalid 'apply_checks_method' used for testing observable metrics.") + + checked_df.count() # Trigger an action to get the metrics + actual_metrics = observation.get + + # Default metrics + assert actual_metrics["input_row_count"] == 4 + assert actual_metrics["error_row_count"] == 1 + assert actual_metrics["warning_row_count"] == 1 + assert actual_metrics["valid_row_count"] == 2 + + # Per-check metrics as compact JSON + check_metrics = json.loads(actual_metrics["check_metrics"]) + assert check_metrics == [ + {"check_name": "id_is_not_null", "error_count": 1, "warning_count": 0}, + {"check_name": "name_is_not_null_and_not_empty", "error_count": 0, "warning_count": 1}, + ] + + +@pytest.mark.parametrize("apply_checks_method", [DQEngine.apply_checks, DQEngine.apply_checks_by_metadata]) +def test_observer_check_metrics_with_auto_derived_names(ws, spark, apply_checks_method): + """Test that check_metrics uses the correct auto-derived name when check.name is not provided.""" + checks_without_names = [ + { + "criticality": "error", + "check": {"function": "is_not_null", "arguments": {"column": "id"}}, + }, + { + "criticality": "warn", + "check": {"function": "is_not_null_and_not_empty", "arguments": {"column": "name"}}, + }, + ] + + observer = DQMetricsObserver(name="test_observer") + dq_engine = DQEngine(workspace_client=ws, spark=spark, observer=observer, extra_params=EXTRA_PARAMS) + + test_df = spark.createDataFrame( + [ + [1, "Alice", 30, 50000], + [2, "Bob", 25, 45000], + [None, "Charlie", 35, 60000], + [4, None, 28, 55000], + ], + TEST_SCHEMA, + ) + + if apply_checks_method == DQEngine.apply_checks: + checks = deserialize_checks(checks_without_names) + checked_df, observation = dq_engine.apply_checks(test_df, checks) + elif apply_checks_method == DQEngine.apply_checks_by_metadata: + checked_df, observation = dq_engine.apply_checks_by_metadata(test_df, checks_without_names) + else: + raise ValueError("Invalid 'apply_checks_method' used for testing observable metrics.") + + checked_df.count() + actual_metrics = observation.get + + assert actual_metrics["input_row_count"] == 4 + + # Auto-derived names come from the check condition alias (e.g. is_not_null("id") -> "id_is_null") + check_metrics = json.loads(actual_metrics["check_metrics"]) + auto_derived_names = [m["check_name"] for m in check_metrics] + assert len(auto_derived_names) == 2 + # Verify that names were auto-derived (not empty) and match the check condition aliases + assert auto_derived_names == ["id_is_null", "name_is_null_or_empty"] + + +def test_observer_check_metrics_change_between_runs(ws, spark): + """Test that check_metrics reflect the correct checks when the rule set changes between runs.""" + test_df = spark.createDataFrame( + [ + [1, "Alice", 30, 50000], + [2, "Bob", 25, 45000], + [None, "Charlie", 35, 60000], + [4, None, 28, 55000], + ], + TEST_SCHEMA, + ) + + # First run: both checks + observer1 = DQMetricsObserver(name="test_observer") + dq_engine1 = DQEngine(workspace_client=ws, spark=spark, observer=observer1, extra_params=EXTRA_PARAMS) + checked_df1, observation1 = dq_engine1.apply_checks_by_metadata(test_df, TEST_CHECKS) + checked_df1.count() + _assert_check_metrics(observation1.get, ["id_is_not_null", "name_is_not_null_and_not_empty"]) + + # Second run: reduced checks — a fresh observer/engine picks up the new rule set + observer2 = DQMetricsObserver(name="test_observer") + dq_engine2 = DQEngine(workspace_client=ws, spark=spark, observer=observer2, extra_params=EXTRA_PARAMS) + checked_df2, observation2 = dq_engine2.apply_checks_by_metadata(test_df, [TEST_CHECKS[0]]) + checked_df2.count() + _assert_check_metrics(observation2.get, ["id_is_not_null"]) + + +def _assert_check_metrics(actual_metrics: dict, expected_check_names: list[str]) -> None: + """Assert that check_metrics contains exactly the expected check names.""" + check_metrics = json.loads(actual_metrics["check_metrics"]) + actual_names = [m["check_name"] for m in check_metrics] + assert actual_names == expected_check_names diff --git a/tests/unit/test_observer.py b/tests/unit/test_observer.py index db54832cf..cdd56457e 100644 --- a/tests/unit/test_observer.py +++ b/tests/unit/test_observer.py @@ -7,96 +7,122 @@ def test_dq_observer_default_initialization(): - """Test DQMetricsObserver default initialization.""" observer = DQMetricsObserver() assert observer.name == "dqx" assert observer.custom_metrics is None - - expected_default_metrics = [ - "count(1) as input_row_count", - "count(case when _errors is not null then 1 end) as error_row_count", - "count(case when _warnings is not null then 1 end) as warning_row_count", - "count(case when _errors is null and _warnings is null then 1 end) as valid_row_count", - ] - assert observer.metrics == expected_default_metrics + assert observer.get_metrics() == _default_metrics() def test_dq_observer_with_custom_metrics(): - """Test DQMetricsObserver with custom metrics.""" - custom_metrics = ["avg(age) as avg_age", "count(case when age > 65 then 1 end) as senior_count"] - - observer = DQMetricsObserver(name="custom_observer", custom_metrics=custom_metrics) + custom = ["avg(age) as avg_age", "count(case when age > 65 then 1 end) as senior_count"] + observer = DQMetricsObserver(name="custom_observer", custom_metrics=custom) assert observer.name == "custom_observer" - assert observer.custom_metrics == custom_metrics - - expected_metrics = [ - "count(1) as input_row_count", - "count(case when _errors is not null then 1 end) as error_row_count", - "count(case when _warnings is not null then 1 end) as warning_row_count", - "count(case when _errors is null and _warnings is null then 1 end) as valid_row_count", - "avg(age) as avg_age", - "count(case when age > 65 then 1 end) as senior_count", - ] - assert observer.metrics == expected_metrics + assert observer.get_metrics() == _default_metrics() + custom def test_dq_observer_empty_custom_metrics(): - """Test DQMetricsObserver with empty custom metrics list.""" observer = DQMetricsObserver(custom_metrics=[]) - - expected_default_metrics = [ - "count(1) as input_row_count", - "count(case when _errors is not null then 1 end) as error_row_count", - "count(case when _warnings is not null then 1 end) as warning_row_count", - "count(case when _errors is null and _warnings is null then 1 end) as valid_row_count", - ] - assert observer.metrics == expected_default_metrics + assert observer.get_metrics() == _default_metrics() def test_dq_observer_run_id_uniqueness(): - observer1 = DQMetricsObserver() - observer2 = DQMetricsObserver() - assert observer1.id != observer2.id + assert DQMetricsObserver().id != DQMetricsObserver().id def test_dq_observer_id_overwrite(): - run_id_overwrite = "1" - observer = DQMetricsObserver(id_overwrite=run_id_overwrite) - assert observer.id == run_id_overwrite + observer = DQMetricsObserver(id_overwrite="1") + assert observer.id == "1" def test_dq_observer_default_column_names(): - """Test that DQMetricsObserver uses correct default column names.""" observer = DQMetricsObserver() - errors_column = DefaultColumnNames.ERRORS.value - warnings_column = DefaultColumnNames.WARNINGS.value - - assert f"count(case when {errors_column} is not null then 1 end) as error_row_count" in observer.metrics - assert f"count(case when {warnings_column} is not null then 1 end) as warning_row_count" in observer.metrics - assert ( - f"count(case when {errors_column} is null and {warnings_column} is null then 1 end) as valid_row_count" - in observer.metrics - ) + err = DefaultColumnNames.ERRORS.value + warn = DefaultColumnNames.WARNINGS.value + assert observer.get_metrics() == _default_metrics(err, warn) def test_dq_observer_custom_column_names(): - """Test that DQMetricsObserver uses correct default column names.""" observer = DQMetricsObserver() - errors_column = "my_errors" - warnings_column = "my_warnings" - observer.set_column_names(error_column_name=errors_column, warning_column_name=warnings_column) - - assert f"count(case when {errors_column} is not null then 1 end) as error_row_count" in observer.metrics - assert f"count(case when {warnings_column} is not null then 1 end) as warning_row_count" in observer.metrics - assert ( - f"count(case when {errors_column} is null and {warnings_column} is null then 1 end) as valid_row_count" - in observer.metrics - ) + observer.set_column_names(error_column_name="my_errors", warning_column_name="my_warnings") + assert observer.get_metrics() == _default_metrics("my_errors", "my_warnings") def test_dq_observer_observation_property(): - """Test that the observation property creates a Spark Observation.""" - observer = DQMetricsObserver(name="test_obs") - observation = observer.observation - assert observation is not None + observation = DQMetricsObserver(name="test_obs").observation assert isinstance(observation, Observation | SparkConnectObservation) + + +def test_get_metrics_without_check_names(): + observer = DQMetricsObserver() + assert observer.get_metrics() == _default_metrics() + + +def test_get_metrics_with_checks_single(): + observer = DQMetricsObserver() + metrics = observer.get_metrics(["id_is_not_null"]) + expected = _default_metrics() + [_check_metrics_expr(["id_is_not_null"])] + assert metrics == expected + + +def test_get_metrics_with_checks_multiple(): + checks = ["id_is_not_null", "name_is_not_empty", "age_in_range"] + observer = DQMetricsObserver() + metrics = observer.get_metrics(checks) + expected = _default_metrics() + [_check_metrics_expr(checks)] + assert metrics == expected + + +def test_get_metrics_with_checks_ordering_with_custom(): + custom = ["avg(age) as avg_age"] + observer = DQMetricsObserver(custom_metrics=custom) + metrics = observer.get_metrics(["my_check"]) + expected = _default_metrics() + [_check_metrics_expr(["my_check"])] + custom + assert metrics == expected + + +def test_get_metrics_with_checks_uses_custom_column_names(): + observer = DQMetricsObserver() + observer.set_column_names(error_column_name="dq_errors", warning_column_name="dq_warnings") + metrics = observer.get_metrics(["my_check"]) + expected = _default_metrics("dq_errors", "dq_warnings") + [ + _check_metrics_expr(["my_check"], "dq_errors", "dq_warnings") + ] + assert metrics == expected + + +def test_get_metrics_with_checks_escapes_single_quotes(): + observer = DQMetricsObserver() + metrics = observer.get_metrics(["it's_valid"]) + expected = _default_metrics() + [_check_metrics_expr(["it's_valid"])] + assert metrics == expected + + +def test_get_metrics_with_checks_empty_list(): + observer = DQMetricsObserver() + metrics = observer.get_metrics([]) + assert metrics == _default_metrics() + + +def test_get_metrics_idempotent(): + """Verifies that repeated calls with the same args return equal results.""" + observer = DQMetricsObserver() + assert observer.get_metrics() == observer.get_metrics() + assert observer.get_metrics(["a"]) == observer.get_metrics(["a"]) + + +def _default_metrics(err="_errors", warn="_warnings"): + return [ + "count(1) as input_row_count", + f"count(case when {err} is not null then 1 end) as error_row_count", + f"count(case when {warn} is not null then 1 end) as warning_row_count", + f"count(case when {err} is null and {warn} is null then 1 end) as valid_row_count", + ] + + +def _check_metrics_expr(check_names, err="_errors", warn="_warnings"): + # Note: this helper derives expected values from production code, so it validates structural + # properties (ordering, column-name propagation) but not SQL correctness. The integration + # tests in test_summary_metrics.py run the generated SQL against real Spark to cover that. + observer = DQMetricsObserver() + observer.set_column_names(error_column_name=err, warning_column_name=warn) + return observer.get_metrics(check_names)[len(_default_metrics()) :][-1]