diff --git a/README.md b/README.md index 83e0f6512d..01c1be89f4 100644 --- a/README.md +++ b/README.md @@ -24,7 +24,7 @@ All state is persisted as append-only tables in the warehouse alongside your dat - **Cursor-based incremental processing.** Automatic gap detection and resume, with microbatch mode for large ranges. No external checkpoint to maintain. - **Ingestion and Python nodes.** Load external data with Python `@loader` functions, and run `@task`, `@asset`, and `@check` nodes as first-class members of the same DAG as your SQL models. -See the [documentation](https://docs.sqlbuild.com) for the full feature set, including providers, lifecycle hooks, Python macros, UDFs, custom materializations, data diffs, zero-copy cloning, and virtual environments. To coordinate dbt and SQLBuild projects, see the [dbt compatibility guide](https://docs.sqlbuild.com/concepts/dbt-compatibility/overview). +See the [documentation](https://docs.sqlbuild.com) for the full feature set, including providers, lifecycle hooks, Python macros, UDFs, custom materializations, data diffs, zero-copy cloning, and virtual environments. This repository also includes a focused guide to [model and raw-query diffs](docs/query-diffs.md). To coordinate dbt and SQLBuild projects, see the [dbt compatibility guide](https://docs.sqlbuild.com/concepts/dbt-compatibility/overview). ## Quick start diff --git a/docs/query-diffs.md b/docs/query-diffs.md new file mode 100644 index 0000000000..b2baeab5cb --- /dev/null +++ b/docs/query-diffs.md @@ -0,0 +1,120 @@ +# Query and model data diffs + +`sqb diff` compares either selected models or two literal read-only SQL queries. Both input modes +use the same schema inspection, row-count reporting, key validation, deterministic sampling, +numeric tolerance, excluded-column, evidence, JSON, and exit-code behavior. + +## Compare models + +Model mode keeps the `FROM:TO` range and selectors: + +```bash +sqb diff prod:dev --full --select orders +``` + +By default, row identity comes from each model's `unique_key`. Repeated `--key` arguments override +that configuration for one invocation: + +```bash +sqb diff prod:dev --full --select order_lines \ + --key order_id \ + --key line_id +``` + +For a model without a stable row key, opt in to exact full-row multiset comparison with +`--unkeyed`. Duplicate rows retain their multiplicity; SQLBuild never guesses a key. + +```bash +sqb diff prod:dev --full --select daily_order_totals --unkeyed +``` + +## Compare raw queries + +Query mode omits `FROM:TO` because each query identifies its own input. Supply each side inline or +from a UTF-8 file, and choose keyed or unkeyed comparison explicitly: + +```bash +sqb diff \ + --left-query 'SELECT order_id, amount_cents FROM analytics.orders_before' \ + --right-query 'SELECT order_id, amount_cents FROM analytics.orders_after' \ + --key order_id +``` + +```bash +sqb diff \ + --left-query-file queries/orders_before.sql \ + --right-query-file queries/orders_after.sql \ + --key account_id \ + --key order_id +``` + +Raw inputs must contain exactly one read-only `SELECT`, `WITH`, or `VALUES` expression. SQLBuild +does not expand `ref()` or other SQLBuild templates in these inputs. The active target connection +executes both sides; use `--target NAME` to select another configured connection. + +Query mode performs a full comparison when no comparison mode flag is present. `--schema-only` is +also available. `--bounded` is model-specific; filter both raw queries directly when a bounded +query comparison is needed. + +### Sampling, tolerances, and exclusions + +Keyed query comparisons support deterministic union-key sampling and the same bounded mismatch +examples as model comparisons: + +```bash +sqb diff \ + --left-query-file queries/orders_before.sql \ + --right-query-file queries/orders_after.sql \ + --key order_id \ + --sample-rows 50000 \ + --sample-seed 7 +``` + +Use repeated invocation overrides where needed: + +```bash +sqb diff \ + --left-query-file queries/orders_before.sql \ + --right-query-file queries/orders_after.sql \ + --key order_id \ + --exclude-column loaded_at \ + --tolerance amount_cents:absolute=1 \ + --tolerance conversion_rate:relative=0.001 +``` + +`--exhaustive` disables inherited or requested sampling. Unkeyed comparison is always exhaustive +and exact, so it does not accept sampling or numeric tolerances. + +## Stable query results and cleanup + +SQLBuild materializes each raw query once in the selected target's default schema before comparing +it. Artifacts use a reserved run-owned name and are dropped immediately when the command finishes, +including after comparison errors. + +Each successfully created artifact publishes an immutable `query_diff_artifact` fingerprint. The +fingerprint is ownership evidence for crash recovery only: running and comparing the queries never +reads historical fingerprints, and fingerprints do not represent planned, running, complete, or +cleaned workflow states. + +If a process stops before immediate cleanup, the next query diff and `sqb janitor` can remove an +expired artifact only when its exact physical identity and reserved name match its ownership +fingerprint. A matching name without ownership evidence is reported for manual investigation and +is not deleted automatically. + +The recovery TTL defaults to 24 hours and accepts SQLBuild fixed-duration syntax: + +```toml +[diff] +query_artifact_ttl = "24h" +``` + +The TTL is a crash-recovery backstop, not the normal cleanup path. Historical ownership inspection +problems are reported but do not prevent the current queries from running. Failure to publish +ownership for a newly created artifact, or failure to remove a current-run artifact, is an execution +error. + +## Structured output and exit behavior + +Use `--json-output PATH` in either mode. Query results include `input_kind = "query"`; model results +use `input_kind = "model"`. Exit code `0` means no differences were found. Exit code `1` means +differences were found or the comparison could not execute successfully. diff --git a/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md b/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md index 061e470793..3eabe15f72 100644 --- a/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md +++ b/src/sqlbuild/.agents/skills/sqlbuild/SKILL.md @@ -7753,10 +7753,12 @@ Compare schemas and data between targets or virtual environments to validate cha SQLBuild can compare schemas and row-level data between two build contexts. This lets you validate that changes produce the expected results before promoting them. -`sqb diff FROM:TO` compares: +`sqb diff` supports two input modes: -- **two targets** (e.g. `prod:dev`) in direct mode, or -- **two virtual environments** (VDEs) when [virtual environments](/concepts/virtual-environments) are enabled. +- **model mode:** `sqb diff FROM:TO --select ...` compares two targets (for example `prod:dev`) in + direct mode, or two virtual environments when virtual environments are enabled; +- **raw-query mode:** inline SQL or SQL files identify the two sides and execute through the active + target connection or an explicit `--target`. The mechanics below are identical for both; only what `FROM` and `TO` refer to changes. @@ -7770,9 +7772,30 @@ for that execution connection still fail before warehouse inspection. sqb diff prod:dev --full --select customer_status_snapshot ``` +Raw-query mode omits `FROM:TO` and requires repeated `--key` values or explicit `--unkeyed` for a +row comparison. Inputs must each be one literal read-only `SELECT`, `WITH`, or `VALUES` expression; +SQLBuild templates such as `ref()` are not expanded. + +```bash +sqb diff \ + --left-query 'SELECT order_id, amount_cents FROM analytics.orders_before' \ + --right-query 'SELECT order_id, amount_cents FROM analytics.orders_after' \ + --key order_id + +sqb diff \ + --left-query-file queries/orders_before.sql \ + --right-query-file queries/orders_after.sql \ + --target dev \ + --key account_id \ + --key order_id +``` + +Query mode defaults to a full comparison and also supports `--schema-only`. It does not accept +`--bounded`; put the desired filter in both SQL inputs instead. + ### Comparison modes -Every diff requires exactly one mode: +Model mode requires exactly one mode. Raw-query mode defaults to full when no mode is supplied: #### Full diff @@ -7864,7 +7887,9 @@ side. ### Row matching -Rows are matched between the two sides using the model's `unique_key`. Models without a `unique_key` can use schema-only diff but cannot run full or bounded row comparisons. +Rows are matched using the model's `unique_key` or repeated invocation-level `--key` overrides. +For inputs without stable row identity, `--unkeyed` performs an exhaustive exact multiset +comparison in which duplicate rows retain their multiplicity. SQLBuild never guesses a key. The diff output categorises rows as: - **Equal** - same key, same values on both sides @@ -7896,6 +7921,10 @@ Tolerance rules support: Tolerances can be set per-column (`by_column`) or per-type (`by_type`). +Use repeated `--tolerance COLUMN:absolute=VALUE` or `--tolerance COLUMN:relative=VALUE` arguments +to override per-column rules for one model or raw-query invocation. Unkeyed comparison is exact and +does not accept tolerances. + ### Excluded columns Columns that are expected to differ between the two sides (like timestamps or context-specific values) can be excluded from the row comparison: @@ -7910,6 +7939,29 @@ MODEL ( Excluded columns are still shown in the schema comparison but skipped during row-level diffing. A column cannot be in both `row_diff_exclude_columns` and `unique_key`. +Use repeated `--exclude-column COLUMN` arguments for invocation-level exclusions. + +### Raw-query artifacts and crash recovery + +SQLBuild materializes each raw query once in the active target schema, compares the stable results, +and drops both artifacts in a `finally` path. Every successfully created artifact appends an +immutable `query_diff_artifact` fingerprint containing ownership and expiry metadata. Query +execution and comparison never depend on historical fingerprints; they are ownership evidence for +later cleanup, not workflow status. + +Before a query diff, and independently through `sqb janitor`, SQLBuild removes an expired artifact +only when its exact physical identity and reserved name match its fingerprint. A reserved-name +relation without matching ownership evidence is reported and retained for manual investigation. +Configure the crash-recovery TTL, which defaults to 24 hours, in `sqlbuild_project.toml`: + +```toml +[diff] +query_artifact_ttl = "24h" +``` + +Historical ownership inspection failures are warnings and do not prevent the current query pair +from running. Current-run ownership publication and immediate cleanup remain mandatory. + ### Verbose output Add `--verbose` or `-v` to see more example rows for mismatches and side-only rows: @@ -7955,7 +8007,7 @@ observed schema for each model before starting its row comparison. ### Selectors -Diff requires `--select` in the current version. You can use any selector syntax: +Model diff requires `--select` in the current version. You can use any selector syntax: ```bash # Diff a single model @@ -7970,7 +8022,9 @@ sqb diff prod:dev --full --select tag:acceptance ### Exit codes -`sqb diff` returns exit code `0` when all selected models have no differences, and `1` when any model has schema or row differences. This makes it usable in CI pipelines as a validation gate. +`sqb diff` returns exit code `0` when no differences are found, and `1` when any input has schema or +row differences or the comparison cannot execute successfully. This makes it usable in CI pipelines +as a validation gate. ## Execution Observability @@ -14448,23 +14502,30 @@ destination. Managed physical sources may bootstrap a destination that has not b Source: `cli/diff.mdx` -Compare schemas and data between targets or virtual environments. +Compare schemas and data between targets, virtual environments, or two literal read-only queries. -Compares schemas and optionally row-level data between two build contexts: two targets (e.g. `prod:dev`) in direct mode, or two virtual environments when virtual mode is enabled. See [Data Diffs](/concepts/diff) for detailed usage. +Compares schemas and optionally row-level data between two build contexts, or materialized results +from two literal read-only SQL inputs. See [Data Diffs](/concepts/diff) for detailed usage. ### Usage ```bash sqb diff : [flags] +sqb diff (--left-query SQL | --left-query-file PATH) \ + (--right-query SQL | --right-query-file PATH) (--key COLUMN... | --unkeyed) [flags] ``` -The first argument is a positional `FROM:TO` range. Exactly one mode is required: `--full`, `--schema-only`, or `--bounded `. +Model mode uses a positional `FROM:TO` range and requires exactly one of `--full`, `--schema-only`, +or `--bounded `. Raw-query mode does not accept `FROM:TO`; it defaults to full comparison, +also supports `--schema-only`, and does not support `--bounded`. In direct mode, `FROM` and `TO` are configured target names. Their database/schema namespaces remain authoritative, while the `TO` target's named connection executes the complete comparison and must be able to read both namespaces. -Full and bounded row comparisons require the model to define `unique_key`. Bounded mode uses the model's cursor and falls back to a full row comparison when no cursor is configured. +Full and bounded keyed comparisons use the model's `unique_key` unless repeated `--key` arguments +override it. `--unkeyed` opts into exact multiset comparison. Bounded mode uses the model's cursor +and falls back to a full row comparison when no cursor is configured. ### Flags @@ -14482,6 +14543,15 @@ Full and bounded row comparisons require the model to define `unique_key`. Bound | `--json-output` | Write structured comparison scope, coverage, and results to a JSON file | | `--max-models` | Fail when the selected scope contains more models than this limit | | `--max-columns` | Fail when either side of a model has more columns than this limit | +| `--left-query` | Literal read-only SQL for the left side | +| `--left-query-file` | UTF-8 SQL file for the left side | +| `--right-query` | Literal read-only SQL for the right side | +| `--right-query-file` | UTF-8 SQL file for the right side | +| `--target` | Target connection for raw-query mode; defaults to the active target | +| `--key` | Row identity column; repeat for a composite key or to override model configuration | +| `--unkeyed` | Compare exact full-row multiplicities without a key | +| `--exclude-column` | Column to omit from row comparison; repeat as needed | +| `--tolerance` | Per-column numeric rule such as `amount:absolute=0.01`; repeat as needed | | `--no-sql-analysis` | Disable compile-time SQL analysis (`--no-sql-validation` is an alias) | | `--select`, `-s` | Select specific models to diff (required in v1) | | `--exclude` | Exclude specific models from diffing | @@ -14503,11 +14573,19 @@ sqb diff prod:dev --bounded 14d --sample-rows 50000 --sample-seed 7 --select ord # Force exhaustive comparison despite inherited sampling defaults sqb diff prod:dev --bounded 14d --exhaustive --select order_lines + +# Compare two raw queries with a composite key +sqb diff \ + --left-query-file queries/orders_before.sql \ + --right-query-file queries/orders_after.sql \ + --key account_id \ + --key order_id ``` ### Exit codes -Returns `0` when all selected models have no differences, `1` when any model has schema or row differences. +Returns `0` when no differences are found. Returns `1` when any input has schema or row differences +or the comparison cannot execute successfully. ## lineage diff --git a/src/sqlbuild/adapter/contract/classes/base_adapter.py b/src/sqlbuild/adapter/contract/classes/base_adapter.py index f855c2a042..979ba8a255 100644 --- a/src/sqlbuild/adapter/contract/classes/base_adapter.py +++ b/src/sqlbuild/adapter/contract/classes/base_adapter.py @@ -118,6 +118,90 @@ def supports_concurrent_microbatch_dml(self) -> bool: return False + def diff_unkeyed_rows( + self, + *, + connection: Any, + left: str, + right: str, + excluded_columns: tuple[str, ...] = (), + cursor_column: str | None = None, + start_cursor: CursorValue | None = None, + end_cursor: CursorValue | None = None, + ) -> RowDiffResult: + """Compare complete-row multiplicities without a unique key.""" + + left_columns: tuple[ColumnInfo, ...] = self.describe_relation( + connection=connection, relation=left + ) + excluded_names: frozenset[str] = frozenset(column.lower() for column in excluded_columns) + compared_columns: tuple[ColumnInfo, ...] = tuple( + column for column in left_columns if column.name.lower() not in excluded_names + ) + if not compared_columns: + raise AdapterUserError(message="unkeyed row diff requires at least one compared column") + column_list: str = ", ".join( + self.render_identifier(column.name) for column in compared_columns + ) + multiplicity_column: str = "__sqlbuild_count" + compared_names: frozenset[str] = frozenset( + column.name.lower() for column in compared_columns + ) + while multiplicity_column.lower() in compared_names: + multiplicity_column += "_" + equal_conditions: str = " AND ".join( + self.build_row_diff_equal_expression( + column=self.render_identifier(column.name), + column_info=column, + tolerances=None, + ) + for column in compared_columns + ) + cursor_filter: str = self.build_cursor_filter( + cursor_column=cursor_column, + start_cursor=start_cursor, + end_cursor=end_cursor, + ) + where_clause: str = f" WHERE {cursor_filter}" if cursor_filter else "" + sql: str = ( + f"WITH __left AS (SELECT {column_list}, COUNT(*) AS {multiplicity_column} " + f"FROM {left}{where_clause} GROUP BY {column_list}), " + f"__right AS (SELECT {column_list}, COUNT(*) AS {multiplicity_column} " + f"FROM {right}{where_clause} GROUP BY {column_list}), " + "__joined AS (SELECT " + f"COALESCE(__left.{multiplicity_column}, 0) AS __left_count, " + f"COALESCE(__right.{multiplicity_column}, 0) AS __right_count " + f"FROM __left FULL OUTER JOIN __right ON {equal_conditions}) " + "SELECT " + "COALESCE(SUM(__left_count), 0), " + "COALESCE(SUM(__right_count), 0), " + "COALESCE(SUM(CASE WHEN __left_count < __right_count " + "THEN __left_count ELSE __right_count END), 0), " + "COALESCE(SUM(CASE WHEN __left_count > __right_count " + "THEN __left_count - __right_count ELSE 0 END), 0), " + "COALESCE(SUM(CASE WHEN __right_count > __left_count " + "THEN __right_count - __left_count ELSE 0 END), 0) " + "FROM __joined" + ) + row: tuple[Any, ...] = self.execute(connection=connection, sql=sql).fetchone() + left_count: int = int(row[0]) + right_count: int = int(row[1]) + equal_count: int = int(row[2]) + left_only_count: int = int(row[3]) + right_only_count: int = int(row[4]) + population_count: int = equal_count + left_only_count + right_only_count + return RowDiffResult( + left_count=left_count, + right_count=right_count, + joined_count=population_count, + equal_count=equal_count, + unequal_count=0, + left_only_count=left_only_count, + right_only_count=right_only_count, + population_count=population_count, + compared_count=population_count, + ) + def physical_relation_generation( self, *, diff --git a/src/sqlbuild/adapter/contract/classes/diff.py b/src/sqlbuild/adapter/contract/classes/diff.py index 4e35d62c28..852a3d438c 100644 --- a/src/sqlbuild/adapter/contract/classes/diff.py +++ b/src/sqlbuild/adapter/contract/classes/diff.py @@ -47,6 +47,21 @@ def diff_rows( """Compare row-level data between two relations.""" ... + @abstractmethod + def diff_unkeyed_rows( + self, + *, + connection: Any, + left: str, + right: str, + excluded_columns: tuple[str, ...] = (), + cursor_column: str | None = None, + start_cursor: CursorValue | None = None, + end_cursor: CursorValue | None = None, + ) -> RowDiffResult: + """Compare exact full-row multiplicities without a unique key.""" + ... + @abstractmethod def count_rows( self, diff --git a/src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py b/src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py index 5795f43478..0d9fc65fb9 100644 --- a/src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py +++ b/src/sqlbuild/adapter/contract/classes/duckdb_backed_adapter.py @@ -21,6 +21,7 @@ ) from sqlbuild.adapter.contract.classes.observed_connection import ObservedConnection from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import ( DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE, @@ -74,7 +75,7 @@ from sqlbuild.sql_values.models import SqlValue -class DuckDbBackedAdapter(BaseAdapter): +class DuckDbBackedAdapter(UnkeyedDiffMixin, BaseAdapter): """Shared adapter implementation for DuckDB-backed connections.""" def get_columns_for_relations( @@ -1175,9 +1176,7 @@ def describe_relation(self, *, connection: Any, relation: str) -> tuple[ColumnIn def query_column_names(self, *, connection: Any, sql: str) -> tuple[str, ...]: """Return DuckDB query column names using DESCRIBE SELECT.""" - cursor: Any = self.execute( - connection=connection, sql=f"DESCRIBE SELECT * FROM ({sql}) AS __describe_source" - ) + cursor: Any = self.execute(connection=connection, sql=f"DESCRIBE {sql}") return tuple(str(row[0]) for row in cursor.fetchall()) def get_relation_max_cursor( diff --git a/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py b/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py new file mode 100644 index 0000000000..b79f47a922 --- /dev/null +++ b/src/sqlbuild/adapter/contract/classes/unkeyed_diff.py @@ -0,0 +1,36 @@ +"""Reusable exact unkeyed diff implementation for first-class adapters.""" + +from __future__ import annotations + +from typing import Any, cast + +from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.adapter.contract.models import CursorValue, RowDiffResult + + +class UnkeyedDiffMixin: + """Expose shared multiset comparison below the base contract boundary.""" + + def diff_unkeyed_rows( + self, + *, + connection: Any, + left: str, + right: str, + excluded_columns: tuple[str, ...] = (), + cursor_column: str | None = None, + start_cursor: CursorValue | None = None, + end_cursor: CursorValue | None = None, + ) -> RowDiffResult: + """Compare exact full-row multiplicities without a unique key.""" + + return BaseAdapter.diff_unkeyed_rows( + cast(BaseAdapter, self), + connection=connection, + left=left, + right=right, + excluded_columns=excluded_columns, + cursor_column=cursor_column, + start_cursor=start_cursor, + end_cursor=end_cursor, + ) diff --git a/src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py b/src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py index 3833a7720c..28b1f00a0c 100644 --- a/src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py +++ b/src/sqlbuild/adapters/bigquery/classes/bigquery_adapter.py @@ -20,6 +20,7 @@ ) from sqlbuild.adapter.contract.classes.microbatch import MicrobatchMixin from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE from sqlbuild.adapter.contract.exceptions import ( AdapterUserError, @@ -102,7 +103,7 @@ from sqlbuild.sql_values.types import SqlValueKind -class BigQueryAdapter(MicrobatchMixin, BaseAdapter): +class BigQueryAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter): """BigQuery adapter backed by google-cloud-bigquery.""" def get_columns_for_relations( diff --git a/src/sqlbuild/adapters/databricks/classes/databricks_adapter.py b/src/sqlbuild/adapters/databricks/classes/databricks_adapter.py index 258d35c1b7..66ba6f9379 100644 --- a/src/sqlbuild/adapters/databricks/classes/databricks_adapter.py +++ b/src/sqlbuild/adapters/databricks/classes/databricks_adapter.py @@ -20,6 +20,7 @@ ) from sqlbuild.adapter.contract.classes.microbatch import MicrobatchMixin from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE from sqlbuild.adapter.contract.exceptions import AdapterUserError from sqlbuild.adapter.contract.main.normalize_seed_csv_value import normalize_seed_csv_value @@ -88,7 +89,7 @@ from sqlbuild.sql_values.models import SqlValue -class DatabricksAdapter(MicrobatchMixin, BaseAdapter): +class DatabricksAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter): """Databricks adapter backed by databricks-sql-connector.""" def get_columns_for_relations( diff --git a/src/sqlbuild/adapters/postgres/classes/postgres_adapter.py b/src/sqlbuild/adapters/postgres/classes/postgres_adapter.py index 0daa4bf982..f77aad4ff0 100644 --- a/src/sqlbuild/adapters/postgres/classes/postgres_adapter.py +++ b/src/sqlbuild/adapters/postgres/classes/postgres_adapter.py @@ -30,6 +30,7 @@ ) from sqlbuild.adapter.contract.classes.microbatch import MicrobatchMixin from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import ( DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE, @@ -85,7 +86,7 @@ from sqlbuild.sql_values.models import SqlValue -class PostgresAdapter(MicrobatchMixin, BaseAdapter): +class PostgresAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter): """PostgreSQL adapter backed by psycopg.""" def get_columns_for_relations( diff --git a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py index c6e65fda2c..3e8a477b37 100644 --- a/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py +++ b/src/sqlbuild/adapters/snowflake/classes/snowflake_adapter.py @@ -19,6 +19,7 @@ ) from sqlbuild.adapter.contract.classes.microbatch import MicrobatchMixin from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE from sqlbuild.adapter.contract.exceptions import AdapterUserError from sqlbuild.adapter.contract.main.normalize_seed_csv_value import normalize_seed_csv_value @@ -96,7 +97,7 @@ _BULK_COLUMN_RELATION_CHUNK_SIZE: int = 200 -class SnowflakeAdapter(MicrobatchMixin, BaseAdapter): +class SnowflakeAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter): """Snowflake adapter backed by snowflake-connector-python.""" adapter_name: ClassVar[str] = BuiltinAdapter.SNOWFLAKE.value diff --git a/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py b/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py index 5db513ed0a..09755ece86 100644 --- a/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py +++ b/src/sqlbuild/adapters/sqlserver/classes/sqlserver_adapter.py @@ -26,6 +26,7 @@ ) from sqlbuild.adapter.contract.classes.microbatch import MicrobatchMixin from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.classes.unkeyed_diff import UnkeyedDiffMixin from sqlbuild.adapter.contract.constants import DIFF_LEFT_SIDE, DIFF_RIGHT_SIDE from sqlbuild.adapter.contract.exceptions import ( AdapterUserError, @@ -87,7 +88,7 @@ from sqlbuild.sql_values.types import SqlValueKind -class SqlServerAdapter(MicrobatchMixin, BaseAdapter): +class SqlServerAdapter(MicrobatchMixin, UnkeyedDiffMixin, BaseAdapter): """Microsoft SQL Server adapter backed by pymssql.""" def get_columns_for_relations( diff --git a/src/sqlbuild/cli/commands/_helpers/diff/execution.py b/src/sqlbuild/cli/commands/_helpers/diff/execution.py index e0d47809e6..804fe903f2 100644 --- a/src/sqlbuild/cli/commands/_helpers/diff/execution.py +++ b/src/sqlbuild/cli/commands/_helpers/diff/execution.py @@ -3,9 +3,14 @@ from __future__ import annotations import sys +from datetime import UTC, datetime +from pathlib import Path from typing import Any from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.adapter.contract.models import RelationLookup +from sqlbuild.adapter.contract.types import BuiltinAdapter +from sqlbuild.adapter.relations.main.relation_lookup import build_relation_lookup from sqlbuild.cli.commands._helpers.planning.external_refs import ( resolve_external_sql_reference_resolver, ) @@ -14,21 +19,37 @@ resolve_project_connection_config, resolve_target_connection_config, ) +from sqlbuild.cli.commands.constants import READ_ONLY_QUERY_ROOT_KEYS from sqlbuild.cli.commands.exceptions import CliUserError from sqlbuild.cli.commands.models import ( DiffCommandRequest, DiffInvocation, DirectDiffPreparation, + QueryDiffPreparation, VirtualDiffPreparation, VirtualDiffRunOutcome, ) from sqlbuild.cli.progress.classes.connection_progress_reporter import ConnectionProgressReporter from sqlbuild.cli.progress.classes.planning_progress_reporter import PlanningProgressReporter +from sqlbuild.compiler.compile.main.effective_runtime import build_effective_runtime_config +from sqlbuild.compiler.compile.main.effective_target_namespace import ( + build_effective_target_namespace, +) from sqlbuild.compiler.pipeline.main.diff import run_diff_pipeline +from sqlbuild.compiler.sql_analysis.main.import_polyglot_sql import import_polyglot_sql +from sqlbuild.cursor_algebra.models import Duration +from sqlbuild.executor.diff.classes.query_artifact_lifecycle import QueryDiffArtifactLifecycle +from sqlbuild.executor.diff.main.config import parse_cli_tolerance_overrides from sqlbuild.executor.diff.main.execute import execute_diff +from sqlbuild.executor.diff.main.execute_query import ( + execute_query_diff as execute_prepared_query_diff, +) from sqlbuild.executor.diff.models import ( DiffExecutionOptions, DiffExecutionResult, + ModelDiffResult, + QueryDiffArtifact, + QueryDiffArtifactCleanupResult, RowDiffSamplingOverride, ) from sqlbuild.presentation.main.supports_color import supports_color @@ -45,6 +66,8 @@ def prepare_direct_diff( ) -> DirectDiffPreparation: """Resolve direct target diff adapter, compiled projects, and limits.""" + if request.from_name is None or request.to_name is None: + raise CliUserError("model diff requires FROM:TO", code="C224") from_target: str = request.from_name to_target: str = request.to_name if from_target not in invocation.discovered_inputs.project_config.targets: @@ -123,17 +146,381 @@ def execute_direct_diff( max_models=request.max_models, max_columns=request.max_columns, sampling_override=_sampling_override(request=request), + unique_key_override=request.unique_key_override, + unkeyed=request.unkeyed, + excluded_columns_override=request.excluded_columns_override, + tolerance_overrides=parse_cli_tolerance_overrides( + values=request.tolerance_overrides + ), ), ) finally: preparation.adapter.close(connection) +def prepare_query_diff( + *, request: DiffCommandRequest, invocation: DiffInvocation +) -> QueryDiffPreparation: + """Resolve raw SQL, active target namespace, adapter, and connection.""" + + effective_adapter_name: str = resolve_effective_adapter_name( + project_config=invocation.discovered_inputs.project_config, + local_config=invocation.discovered_inputs.local_config, + ) + adapter: BaseAdapter = resolve_adapter( + adapter_name=effective_adapter_name, + project_dir=invocation.effective_project_dir, + ) + connection_config: dict[str, object] = resolve_project_connection_config( + discovered_inputs=invocation.discovered_inputs, + project_dir=invocation.effective_project_dir, + selected_target=request.selected_target, + cli_vars=request.cli_vars, + ) + connection_database: object | None = connection_config.get("database") + if effective_adapter_name in { + BuiltinAdapter.DUCKDB.value, + BuiltinAdapter.MOTHERDUCK.value, + }: + connection_database = None + target_name, _target, database, schema = build_effective_target_namespace( + discovered_inputs=invocation.discovered_inputs, + selected_target=request.selected_target, + connection_database=connection_database, + connection_schema=connection_config.get("schema"), + default_database=adapter.default_database(), + default_schema=adapter.default_schema(), + ) + if schema is None: + raise CliUserError( + "raw-query diff requires an active target schema", + code="C227", + help="configure the selected target schema before running the comparison", + ) + left_sql: str = _resolve_query_input( + inline=request.left_query, + file=request.left_query_file, + label="left", + ) + right_sql: str = _resolve_query_input( + inline=request.right_query, + file=request.right_query_file, + label="right", + ) + left_sql = _validate_read_only_query(sql=left_sql, label="left", adapter=adapter) + right_sql = _validate_read_only_query(sql=right_sql, label="right", adapter=adapter) + _runtime_target, _effective_vars, run_id = build_effective_runtime_config( + discovered_inputs=invocation.discovered_inputs, + selected_target=request.selected_target, + cli_vars=request.cli_vars, + ) + return QueryDiffPreparation( + adapter=adapter, + connection_config=connection_config, + left_sql=left_sql, + right_sql=right_sql, + selected_target=target_name, + database=database, + schema=schema, + run_id=run_id, + artifact_ttl=invocation.discovered_inputs.project_config.diff.query_artifact_ttl, + effective_max_column_examples=_effective_max_examples( + explicit_value=request.max_column_examples, + verbose=request.verbose, + ), + effective_max_row_only_examples=_effective_max_examples( + explicit_value=request.max_row_only_examples, + verbose=request.verbose, + ), + ) + + +def execute_query_diff( + *, request: DiffCommandRequest, preparation: QueryDiffPreparation +) -> DiffExecutionResult: + """Materialize, compare, and clean one raw-query pair.""" + + adapter: BaseAdapter = preparation.adapter + print("Query diff connection START", file=sys.stderr) + try: + connection: Any = adapter.connect(preparation.connection_config) + except Exception: + print("Query diff connection ERROR", file=sys.stderr) + raise + print("Query diff connection OK", file=sys.stderr) + left: QueryDiffArtifact = QueryDiffArtifactLifecycle.build( + adapter=adapter, + run_id=preparation.run_id, + side="left", + database=preparation.database, + schema=preparation.schema, + ) + right: QueryDiffArtifact = QueryDiffArtifactLifecycle.build( + adapter=adapter, + run_id=preparation.run_id, + side="right", + database=preparation.database, + schema=preparation.schema, + ) + created: list[QueryDiffArtifact] = [] + cleanup_errors: list[Exception] = [] + try: + print("Query diff cleanup START", file=sys.stderr) + try: + cleanup_result: QueryDiffArtifactCleanupResult = ( + QueryDiffArtifactLifecycle.cleanup_expired( + adapter=adapter, + connection=connection, + database=preparation.database, + schema=preparation.schema, + now=datetime.now(UTC), + ) + ) + except Exception as error: + print( + "Query diff cleanup WARNING recovery cleanup could not inspect historical " + f"ownership evidence: {error}", + file=sys.stderr, + ) + else: + print( + f"Query diff cleanup OK ({len(cleanup_result.cleaned)} expired)", + file=sys.stderr, + ) + for relation in cleanup_result.untracked: + print( + "Query diff cleanup WARNING reserved-name relation lacks matching " + f"ownership evidence: {relation.name}", + file=sys.stderr, + ) + artifact_lookup: RelationLookup = build_relation_lookup( + adapter=adapter, + connection=connection, + locations=tuple( + (artifact.database, artifact.schema, artifact.name) for artifact in (left, right) + ), + ) + for artifact in (left, right): + if artifact_lookup.exists( + database=artifact.database, + schema=artifact.schema, + name=artifact.name, + ): + raise CliUserError( + f"query-diff artifact already exists: {artifact.relation}", + code="C245", + help="wait for its TTL cleanup or investigate the reserved-name relation", + ) + ttl: Duration | None = Duration.parse(preparation.artifact_ttl) + if ttl is None: + raise CliUserError("diff.query_artifact_ttl must be a positive duration", code="C228") + expires_at: datetime = ttl.add_to(datetime.now(UTC)) + _preflight_query_columns( + adapter=adapter, + connection=connection, + sql=preparation.left_sql, + label="left", + max_columns=request.max_columns, + ) + _preflight_query_columns( + adapter=adapter, + connection=connection, + sql=preparation.right_sql, + label="right", + max_columns=request.max_columns, + ) + print("Query diff materialization START", file=sys.stderr) + try: + QueryDiffArtifactLifecycle.materialize( + adapter=adapter, + connection=connection, + artifact=left, + sql=preparation.left_sql, + expires_at=expires_at, + ) + created.append(left) + QueryDiffArtifactLifecycle.materialize( + adapter=adapter, + connection=connection, + artifact=right, + sql=preparation.right_sql, + expires_at=expires_at, + ) + created.append(right) + except Exception: + print("Query diff materialization ERROR", file=sys.stderr) + raise + print( + f"Query diff materialization OK ({preparation.schema}; expires " + f"{expires_at.isoformat()})", + file=sys.stderr, + ) + print("Query diff comparison START", file=sys.stderr) + try: + model_result: ModelDiffResult = execute_prepared_query_diff( + adapter=adapter, + connection=connection, + left_relation=left.relation, + right_relation=right.relation, + options=DiffExecutionOptions( + schema_only=request.schema_only, + collect_samples=not request.schema_only, + max_column_examples=preparation.effective_max_column_examples, + max_row_only_examples=preparation.effective_max_row_only_examples, + max_columns=request.max_columns, + sampling_override=_sampling_override(request=request), + unique_key_override=request.unique_key_override, + unkeyed=request.unkeyed, + excluded_columns_override=request.excluded_columns_override, + tolerance_overrides=parse_cli_tolerance_overrides( + values=request.tolerance_overrides + ), + ), + ) + except Exception: + print("Query diff comparison ERROR", file=sys.stderr) + raise + print("Query diff comparison OK", file=sys.stderr) + return DiffExecutionResult(model_results=(model_result,)) + finally: + active_error: BaseException | None = sys.exc_info()[1] + print("Query diff immediate cleanup START", file=sys.stderr) + for artifact in reversed(created): + try: + QueryDiffArtifactLifecycle.cleanup( + adapter=adapter, + connection=connection, + artifact=artifact, + ) + except Exception as error: + cleanup_errors.append(error) + if cleanup_errors: + print( + f"Query diff immediate cleanup ERROR ({len(cleanup_errors)} failed)", + file=sys.stderr, + ) + else: + print( + f"Query diff immediate cleanup OK ({len(created)} removed)", + file=sys.stderr, + ) + adapter.close(connection) + if cleanup_errors and active_error is None: + raise cleanup_errors[0] + + +def _resolve_query_input(*, inline: str | None, file: Path | None, label: str) -> str: + if inline is not None: + return inline + if file is None: + raise CliUserError(f"raw-query diff requires a {label} query", code="C229") + if not file.exists() or not file.is_file(): + raise CliUserError(f"{label} query file does not exist: {file}", code="C230") + try: + return file.read_text(encoding="utf-8") + except (OSError, UnicodeError) as error: + raise CliUserError( + f"{label} query file could not be read as UTF-8: {file}", code="C231" + ) from error + + +def _validate_read_only_query(*, sql: str, label: str, adapter: BaseAdapter) -> str: + normalized_sql: str = sql.strip() + if normalized_sql.endswith(";"): + normalized_sql = normalized_sql[:-1].rstrip() + if not normalized_sql: + raise CliUserError(f"{label} query must not be empty", code="C232") + polyglot: Any | None = import_polyglot_sql() + if polyglot is None: + raise CliUserError("raw-query diff requires the bundled SQL parser", code="C233") + try: + expression: Any = polyglot.parse_one( + normalized_sql, + dialect=adapter.sql_analysis_dialect() or "generic", + ) + except Exception as error: + raise CliUserError( + f"{label} query is not one valid SQL statement: {error}", code="C234" + ) from error + if str(getattr(expression, "key", "")).lower() not in READ_ONLY_QUERY_ROOT_KEYS: + raise CliUserError( + f"{label} query must be one read-only SELECT, WITH, or VALUES expression", + code="C235", + ) + mutating_keys: frozenset[str] = frozenset( + { + "alter", + "command", + "copy", + "create", + "delete", + "drop", + "execute", + "grant", + "insert", + "load_data", + "merge", + "pragma", + "revoke", + "set", + "transaction", + "truncate", + "uncache", + "update", + "use", + } + ) + if any(str(getattr(node, "key", "")).lower() in mutating_keys for node in expression.walk()): + raise CliUserError( + f"{label} query must not contain data-changing or administrative statements", + code="C240", + ) + expression_args: object = getattr(expression, "args", {}) + if isinstance(expression_args, dict) and ( + expression_args.get("into") is not None or bool(expression_args.get("locks")) + ): + raise CliUserError( + f"{label} query must not write a relation or acquire row locks", + code="C244", + ) + return normalized_sql + + +def _preflight_query_columns( + *, + adapter: BaseAdapter, + connection: Any, + sql: str, + label: str, + max_columns: int | None, +) -> None: + columns: tuple[str, ...] = adapter.query_column_names(connection=connection, sql=sql) + if not columns: + raise CliUserError(f"{label} query returned no columns", code="C237") + normalized: tuple[str, ...] = tuple(column.lower() for column in columns) + duplicates: tuple[str, ...] = tuple( + sorted({column for column in normalized if normalized.count(column) > 1}) + ) + if duplicates: + raise CliUserError( + f"{label} query returns duplicate column names: {', '.join(duplicates)}", + code="C238", + help="add unique aliases before comparing query results", + ) + if max_columns is not None and len(columns) > max_columns: + raise CliUserError( + f"{label} query has {len(columns)} columns, exceeding --max-columns {max_columns}", + code="C239", + ) + + def prepare_virtual_diff( *, request: DiffCommandRequest, invocation: DiffInvocation ) -> VirtualDiffPreparation: """Resolve virtual diff adapter, connection, and sample limits.""" + if request.from_name is None or request.to_name is None: + raise CliUserError("virtual diff requires FROM:TO", code="C236") effective_adapter_name: str = resolve_effective_adapter_name( project_config=invocation.discovered_inputs.project_config, local_config=invocation.discovered_inputs.local_config, @@ -204,6 +591,10 @@ def execute_virtual_diff( max_models=request.max_models, max_columns=request.max_columns, sampling_override=_sampling_override(request=request), + unique_key_override=request.unique_key_override, + unkeyed=request.unkeyed, + excluded_columns_override=request.excluded_columns_override, + tolerance_overrides=parse_cli_tolerance_overrides(values=request.tolerance_overrides), allow_partial_diff=request.allow_partial_diff, cli_vars=request.cli_vars, external_sql_reference_resolver=resolve_external_sql_reference_resolver( diff --git a/src/sqlbuild/cli/commands/_helpers/diff/invocation.py b/src/sqlbuild/cli/commands/_helpers/diff/invocation.py index 6dc8426b55..dc3e4abfc9 100644 --- a/src/sqlbuild/cli/commands/_helpers/diff/invocation.py +++ b/src/sqlbuild/cli/commands/_helpers/diff/invocation.py @@ -21,7 +21,7 @@ def resolve_diff_invocation(*, request: DiffCommandRequest) -> DiffInvocation: project_dir=effective_project_dir ) is_virtual_mode: bool = discovered_inputs.project_config.settings.virtual_environments - if not request.select and not is_virtual_mode: + if not request.select and not is_virtual_mode and not is_query_diff_request(request=request): raise CliUserError("diff requires --select in v1", code="C204") return DiffInvocation( effective_project_dir=effective_project_dir, @@ -31,10 +31,11 @@ def resolve_diff_invocation(*, request: DiffCommandRequest) -> DiffInvocation: def _validate_diff_request(*, request: DiffCommandRequest) -> None: + query_mode: bool = is_query_diff_request(request=request) selected_modes: int = ( int(request.full) + int(request.schema_only) + int(request.bounded is not None) ) - if selected_modes != 1: + if selected_modes != 1 and not (query_mode and selected_modes == 0): raise CliUserError( "diff requires exactly one of --full, --schema-only, or --bounded", code="C201", @@ -51,3 +52,81 @@ def _validate_diff_request(*, request: DiffCommandRequest) -> None: raise CliUserError("diff --max-models must be positive", code="C210") if request.max_columns is not None and request.max_columns <= 0: raise CliUserError("diff --max-columns must be positive", code="C211") + if request.unkeyed and request.unique_key_override: + raise CliUserError("diff --unkeyed cannot be combined with --key", code="C226") + if request.unkeyed and (request.sample_rows is not None or request.sample_seed is not None): + raise CliUserError("unkeyed diff does not support sampling", code="C222") + if request.unkeyed and request.tolerance_overrides: + raise CliUserError("unkeyed diff does not support tolerances", code="C223") + if any(not key.strip() for key in request.unique_key_override): + raise CliUserError("diff --key values must not be empty", code="C241") + normalized_keys: tuple[str, ...] = tuple(key.lower() for key in request.unique_key_override) + if len(set(normalized_keys)) != len(normalized_keys): + raise CliUserError("diff --key values must be unique", code="C242") + if query_mode: + _validate_query_diff_request(request=request) + else: + _validate_model_diff_request(request=request) + + +def is_query_diff_request(*, request: DiffCommandRequest) -> bool: + """Return whether any raw-query input was supplied.""" + + return any( + value is not None + for value in ( + request.left_query, + request.left_query_file, + request.right_query, + request.right_query_file, + ) + ) + + +def _validate_query_diff_request(*, request: DiffCommandRequest) -> None: + if request.from_name is not None or request.to_name is not None: + raise CliUserError("raw-query diff does not accept FROM:TO", code="C212") + if request.select or request.exclude: + raise CliUserError("raw-query diff does not accept model selectors", code="C213") + if request.bounded is not None: + raise CliUserError( + "raw-query diff does not support --bounded; filter both raw queries explicitly", + code="C214", + ) + if request.allow_partial_diff: + raise CliUserError("raw-query diff does not use --allow-partial-diff", code="C215") + if request.max_models is not None: + raise CliUserError("raw-query diff does not accept --max-models", code="C243") + if request.left_query is None and request.left_query_file is None: + raise CliUserError("raw-query diff requires a left query", code="C216") + if request.left_query is not None and request.left_query_file is not None: + raise CliUserError( + "raw-query diff accepts either --left-query or --left-query-file, not both", + code="C217", + ) + if request.right_query is None and request.right_query_file is None: + raise CliUserError("raw-query diff requires a right query", code="C218") + if request.right_query is not None and request.right_query_file is not None: + raise CliUserError( + "raw-query diff accepts either --right-query or --right-query-file, not both", + code="C219", + ) + if request.schema_only: + if request.unkeyed or request.unique_key_override: + raise CliUserError( + "schema-only raw-query diff does not accept --key or --unkeyed", + code="C220", + ) + return + if bool(request.unique_key_override) == request.unkeyed: + raise CliUserError( + "raw-query diff requires one or more --key values or explicit --unkeyed", + code="C221", + ) + + +def _validate_model_diff_request(*, request: DiffCommandRequest) -> None: + if request.from_name is None or request.to_name is None: + raise CliUserError("model diff requires FROM:TO", code="C224") + if request.selected_target is not None: + raise CliUserError("model diff does not accept --target", code="C225") diff --git a/src/sqlbuild/cli/commands/_helpers/diff/json_output.py b/src/sqlbuild/cli/commands/_helpers/diff/json_output.py index 679e090533..c7a8c141f9 100644 --- a/src/sqlbuild/cli/commands/_helpers/diff/json_output.py +++ b/src/sqlbuild/cli/commands/_helpers/diff/json_output.py @@ -39,10 +39,12 @@ def write_diff_json_output( def _model_payload(*, model: ModelDiffResult) -> dict[str, object]: rows: RowDiffResult | None = model.row_result payload: dict[str, object] = { + "input_kind": model.input_kind, "name": model.name, "from_relation": model.left_relation, "to_relation": model.right_relation, "unique_key": list(model.unique_key), + "unkeyed": model.unkeyed, "excluded_columns": list(model.excluded_columns), "schema": { "added_columns": [column.name for column in model.schema_result.added_columns], @@ -70,8 +72,8 @@ def _model_payload(*, model: ModelDiffResult) -> dict[str, object]: payload["sampling"] = { "configured_row_limit": rows.sampling.row_limit if rows.sampling is not None else None, "seed": rows.sampling.seed if rows.sampling is not None else None, - "population_keys": rows.population_count, - "compared_keys": rows.compared_count, + "population_keys": None if model.unkeyed else rows.population_count, + "compared_keys": None if model.unkeyed else rows.compared_count, "coverage_ratio": ( rows.compared_count / rows.population_count if rows.population_count else 1.0 ), diff --git a/src/sqlbuild/cli/commands/_helpers/diff/output.py b/src/sqlbuild/cli/commands/_helpers/diff/output.py index 30ec893e64..e35593c298 100644 --- a/src/sqlbuild/cli/commands/_helpers/diff/output.py +++ b/src/sqlbuild/cli/commands/_helpers/diff/output.py @@ -17,6 +17,7 @@ RowDiffSampleRow, SchemaDiffResult, ) +from sqlbuild.executor.diff.constants import DIFF_INPUT_KIND_QUERY from sqlbuild.executor.diff.models import DiffExecutionResult, ModelDiffResult _WIDTH: int = 110 @@ -56,7 +57,14 @@ def render_diff_output( console.print( Text(f"{from_label} vs {to_label}", style=_RICH_OBJECT_STYLE, justify="center") ) - console.print(Text(f"selected models: {len(result.model_results):,}", justify="center")) + input_label: str = ( + "query pairs" + if all(item.input_kind == DIFF_INPUT_KIND_QUERY for item in result.model_results) + else "models" + ) + console.print( + Text(f"selected {input_label}: {len(result.model_results):,}", justify="center") + ) console.print() model_result: ModelDiffResult for index, model_result in enumerate(result.model_results): @@ -181,7 +189,10 @@ def _render_overview(*, model_result: ModelDiffResult, mode_label: str) -> Rende table: Table = Table.grid(padding=(0, 2)) table.add_column(style=_RICH_OBJECT_STYLE, justify="right") table.add_column() - table.add_row("Model", model_result.name) + table.add_row( + "Input" if model_result.input_kind == DIFF_INPUT_KIND_QUERY else "Model", + model_result.name, + ) table.add_row("Key", _primary_key_label(model_result)) table.add_row("Comparison", _mode_display(model_result=model_result, mode_label=mode_label)) if model_result.bounded_fallback: @@ -189,7 +200,10 @@ def _render_overview(*, model_result: ModelDiffResult, mode_label: str) -> Rende if model_result.excluded_columns: table.add_row("Excluded", ", ".join(model_result.excluded_columns)) if model_result.row_result is not None: - table.add_row("Scope", _row_scope_label(rows=model_result.row_result)) + table.add_row( + "Scope", + _row_scope_label(rows=model_result.row_result, unkeyed=model_result.unkeyed), + ) tolerance_label: str | None = _tolerance_label(model_result.row_result) if tolerance_label is not None: table.add_row("Tolerances", tolerance_label) @@ -227,7 +241,9 @@ def _tolerance_label(row_result: RowDiffResult | None) -> str | None: return ", ".join(labels) -def _row_scope_label(*, rows: RowDiffResult) -> str: +def _row_scope_label(*, rows: RowDiffResult, unkeyed: bool = False) -> str: + if unkeyed: + return f"exhaustive ({rows.population_count:,} multiset rows)" if rows.sampling is None: return f"exhaustive ({rows.population_count:,} union keys)" if rows.is_exhaustive: diff --git a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py b/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py index ad9886d8b1..38be21541c 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py +++ b/src/sqlbuild/cli/commands/_helpers/entry/dispatch.py @@ -356,9 +356,10 @@ def dispatch_cli_command(*, args: CliNamespace, handlers: CliEntrypointHandlers) ) ) if args.command == CliCommand.DIFF: - from_name: str - to_name: str - from_name, to_name = parse_diff_name_range(args.target_range) + from_name: str | None = None + to_name: str | None = None + if args.target_range is not None: + from_name, to_name = parse_diff_name_range(args.target_range) return handlers.run_diff( DiffCommandRequest( project_dir=project_dir, @@ -382,6 +383,15 @@ def dispatch_cli_command(*, args: CliNamespace, handlers: CliEntrypointHandlers) verbose=args.verbose, cli_vars=args.vars, allow_partial_diff=args.allow_partial_diff, + selected_target=args.target, + left_query=args.left_query, + left_query_file=args.left_query_file, + right_query=args.right_query, + right_query_file=args.right_query_file, + unique_key_override=tuple(args.key), + unkeyed=args.unkeyed, + excluded_columns_override=tuple(args.exclude_column), + tolerance_overrides=tuple(args.tolerance), ) ) if args.command == CliCommand.RECONCILE: diff --git a/src/sqlbuild/cli/commands/_helpers/entry/parser.py b/src/sqlbuild/cli/commands/_helpers/entry/parser.py index 65ac994937..12294e40be 100644 --- a/src/sqlbuild/cli/commands/_helpers/entry/parser.py +++ b/src/sqlbuild/cli/commands/_helpers/entry/parser.py @@ -348,7 +348,7 @@ def _add_data_parsers( clone_parser.add_argument("--event-output", type=Path, default=None, help=argparse.SUPPRESS) diff_parser: argparse.ArgumentParser = subparsers.add_parser(CliCommand.DIFF) - diff_parser.add_argument("target_range", metavar="FROM:TO") + diff_parser.add_argument("target_range", nargs="?", metavar="FROM:TO") _add_sql_analysis_override(diff_parser) diff_parser.add_argument("--full", action="store_true", default=False) diff_parser.add_argument("--schema-only", action="store_true", default=False) @@ -362,6 +362,37 @@ def _add_data_parsers( diff_parser.add_argument("--max-models", type=int, default=None) diff_parser.add_argument("--max-columns", type=int, default=None) diff_parser.add_argument("--allow-partial-diff", action="store_true", default=False) + diff_parser.add_argument( + "--target", + default=None, + help="target connection for raw-query diff (defaults to the active target)", + ) + diff_parser.add_argument("--left-query", default=None, help="literal left-side SQL query") + diff_parser.add_argument( + "--left-query-file", type=Path, default=None, help="path to left-side SQL query" + ) + diff_parser.add_argument("--right-query", default=None, help="literal right-side SQL query") + diff_parser.add_argument( + "--right-query-file", type=Path, default=None, help="path to right-side SQL query" + ) + diff_parser.add_argument( + "--key", action="append", default=[], help="row identity column; repeat for composite keys" + ) + diff_parser.add_argument( + "--unkeyed", + action="store_true", + default=False, + help="compare exact full-row multiplicities without a key", + ) + diff_parser.add_argument( + "--exclude-column", action="append", default=[], help="column to omit from row comparison" + ) + diff_parser.add_argument( + "--tolerance", + action="append", + default=[], + help="numeric tolerance as COLUMN:absolute=VALUE or COLUMN:relative=VALUE", + ) _ = add_execution_json_output_arg(diff_parser) _ = add_select_args(diff_parser) _ = add_vars_args(diff_parser) diff --git a/src/sqlbuild/cli/commands/_helpers/janitor_output/output.py b/src/sqlbuild/cli/commands/_helpers/janitor_output/output.py index 3e602f3762..9f479ef0b7 100644 --- a/src/sqlbuild/cli/commands/_helpers/janitor_output/output.py +++ b/src/sqlbuild/cli/commands/_helpers/janitor_output/output.py @@ -11,6 +11,7 @@ JanitorExpiredLockCandidate, JanitorExpiredVirtualEnvironmentCandidate, JanitorPlan, + JanitorQueryDiffArtifactCandidate, JanitorSkippedRelation, JanitorSkippedSchema, JanitorStateBackupCandidate, @@ -53,6 +54,12 @@ def _write_plan_summary(*, plan: JanitorPlan, stream: TextIO, style: CliStyle) - label="reported objects" if plan.direct_mode else "eligible for deletion", items=plan.candidates, ) + _write_count_row( + stream=stream, + style=style, + label="query artifacts", + items=plan.query_diff_artifact_candidates, + ) _write_count_row( stream=stream, style=style, @@ -107,6 +114,15 @@ def _write_count_row( stream.write(f" {label:<22} {rendered}\n") +def _write_query_artifact_candidates(*, plan: JanitorPlan, stream: TextIO, style: CliStyle) -> None: + if not plan.query_diff_artifact_candidates: + return + stream.write(f"\n{style.success('Eligible expired query artifacts')}\n") + query_artifact: JanitorQueryDiffArtifactCandidate + for query_artifact in plan.query_diff_artifact_candidates: + stream.write(f" {style.object_name(query_artifact.key.display_name())}\n") + + def write_plan(*, plan: JanitorPlan, stream: TextIO, use_color: bool = False) -> None: """Write a janitor preview.""" @@ -152,6 +168,8 @@ def write_plan(*, plan: JanitorPlan, stream: TextIO, use_color: bool = False) -> f" {style.object_name(candidate.key.display_name())}{style.muted(reason)}\n" ) + _write_query_artifact_candidates(plan=plan, stream=stream, style=style) + if plan.checkpoint_candidates: stream.write(f"\n{style.success('Eligible checkpoints')}\n") checkpoint_candidate: JanitorCheckpointCandidate @@ -238,9 +256,13 @@ def confirmation_text(plan: JanitorPlan) -> str: + len(plan.virtual_state_prune_candidates) ) if state_candidate_count == 0: - deletion_count: int = 0 if plan.direct_mode else len(plan.candidates) + deletion_count: int = (0 if plan.direct_mode else len(plan.candidates)) + len( + plan.query_diff_artifact_candidates + ) return f"delete {deletion_count} objects from {environment_label(plan)}" - physical_deletion_count: int = 0 if plan.direct_mode else len(plan.candidates) + physical_deletion_count: int = (0 if plan.direct_mode else len(plan.candidates)) + len( + plan.query_diff_artifact_candidates + ) deletion_count: int = physical_deletion_count + state_candidate_count return f"delete {deletion_count} items from {environment_label(plan)}" diff --git a/src/sqlbuild/cli/commands/_helpers/janitor_output/outputs.py b/src/sqlbuild/cli/commands/_helpers/janitor_output/outputs.py index c256843f98..8aaedb868b 100644 --- a/src/sqlbuild/cli/commands/_helpers/janitor_output/outputs.py +++ b/src/sqlbuild/cli/commands/_helpers/janitor_output/outputs.py @@ -38,6 +38,7 @@ def janitor_plan_has_work(planning_result: JanitorPlanningResult) -> bool: plan: JanitorPlan = planning_result.plan return bool( (plan.candidates and not plan.direct_mode) + or plan.query_diff_artifact_candidates or plan.checkpoint_candidates or plan.detached_virtual_environment_candidates or plan.expired_virtual_environment_candidates @@ -65,14 +66,18 @@ def confirm_janitor_plan(*, planning_result: JanitorPlanningResult) -> bool: plan.virtual_state_prune_candidates ) if state_candidate_count or prune_count: - physical_deletion_count: int = 0 if plan.direct_mode else len(plan.candidates) + physical_deletion_count: int = (0 if plan.direct_mode else len(plan.candidates)) + len( + plan.query_diff_artifact_candidates + ) deletion_count: int = physical_deletion_count + state_candidate_count + prune_count sys.stdout.write( f"Janitor will delete {deletion_count} items from {environment_label(plan)}.\n" ) else: sys.stdout.write( - f"Janitor will delete {len(plan.candidates)} objects from {environment_label(plan)}.\n" + "Janitor will delete " + f"{len(plan.candidates) + len(plan.query_diff_artifact_candidates)} objects " + f"from {environment_label(plan)}.\n" ) if plan.retention_days == 0: sys.stdout.write("Retention: disabled (0 days)\n") @@ -112,6 +117,7 @@ def _deleted_message(*, result: JanitorExecutionResult) -> str: + len(result.deleted_state_backups) + len(result.deleted_expired_locks) ) + deleted_object_count: int = len(result.deleted) + len(result.deleted_query_diff_artifacts) pruned_state_count: int = len(result.pruned_direct_state) + len(result.pruned_virtual_state) non_checkpoint_state_count: int = deleted_state_count - len(result.deleted_checkpoints) if non_checkpoint_state_count or pruned_state_count: @@ -119,12 +125,12 @@ def _deleted_message(*, result: JanitorExecutionResult) -> str: "state tables" if result.pruned_virtual_state else "direct state tables" ) return ( - f"Deleted {len(result.deleted)} objects, deleted {deleted_state_count} " + f"Deleted {deleted_object_count} objects, deleted {deleted_state_count} " f"state items, and pruned {pruned_state_count} {pruned_state_label}." ) if result.deleted_checkpoints: return ( - f"Deleted {len(result.deleted)} objects and " + f"Deleted {deleted_object_count} objects and " f"{len(result.deleted_checkpoints)} checkpoints." ) - return f"Deleted {len(result.deleted)} objects." + return f"Deleted {deleted_object_count} objects." diff --git a/src/sqlbuild/cli/commands/_helpers/janitor_runtime/planning.py b/src/sqlbuild/cli/commands/_helpers/janitor_runtime/planning.py index 117311a754..4b0249cadb 100644 --- a/src/sqlbuild/cli/commands/_helpers/janitor_runtime/planning.py +++ b/src/sqlbuild/cli/commands/_helpers/janitor_runtime/planning.py @@ -206,6 +206,7 @@ def _janitor_candidate_count(plan: JanitorPlan) -> int: len(candidates) for candidates in ( plan.candidates, + plan.query_diff_artifact_candidates, plan.checkpoint_candidates, plan.detached_virtual_environment_candidates, plan.expired_virtual_environment_candidates, diff --git a/src/sqlbuild/cli/commands/classes/cli_namespace.py b/src/sqlbuild/cli/commands/classes/cli_namespace.py index 3f70ba6476..3e7dee920d 100644 --- a/src/sqlbuild/cli/commands/classes/cli_namespace.py +++ b/src/sqlbuild/cli/commands/classes/cli_namespace.py @@ -108,6 +108,14 @@ "full": False, "schema_only": False, "allow_partial_diff": False, + "left_query": None, + "left_query_file": None, + "right_query": None, + "right_query_file": None, + "key": [], + "unkeyed": False, + "exclude_column": [], + "tolerance": [], "reconcile_command": None, "reconcile_model": None, "reconcile_physical_relation": None, @@ -267,6 +275,14 @@ class CliNamespace: full: bool schema_only: bool allow_partial_diff: bool + left_query: str | None + left_query_file: Path | None + right_query: str | None + right_query_file: Path | None + key: list[str] + unkeyed: bool + exclude_column: list[str] + tolerance: list[str] reconcile_command: str | None reconcile_model: str | None reconcile_physical_relation: str | None diff --git a/src/sqlbuild/cli/commands/constants.py b/src/sqlbuild/cli/commands/constants.py index b3ef37aa11..c27ac51d19 100644 --- a/src/sqlbuild/cli/commands/constants.py +++ b/src/sqlbuild/cli/commands/constants.py @@ -94,3 +94,6 @@ SCOPE_GLOBAL_SUMMARY: str = "summary" SCOPE_GLOBAL_ALL: str = "all" SCOPE_DEFAULT_PAGE_SIZE: int = 100 +READ_ONLY_QUERY_ROOT_KEYS: frozenset[str] = frozenset( + {"except", "intersect", "select", "union", "values"} +) diff --git a/src/sqlbuild/cli/commands/main/inspection/_diff.py b/src/sqlbuild/cli/commands/main/inspection/_diff.py index a58d3be511..2b9cea5ccc 100644 --- a/src/sqlbuild/cli/commands/main/inspection/_diff.py +++ b/src/sqlbuild/cli/commands/main/inspection/_diff.py @@ -4,11 +4,16 @@ from sqlbuild.cli.commands._helpers.diff.execution import ( execute_direct_diff, + execute_query_diff, execute_virtual_diff, prepare_direct_diff, + prepare_query_diff, prepare_virtual_diff, ) -from sqlbuild.cli.commands._helpers.diff.invocation import resolve_diff_invocation +from sqlbuild.cli.commands._helpers.diff.invocation import ( + is_query_diff_request, + resolve_diff_invocation, +) from sqlbuild.cli.commands._helpers.diff.outputs import ( resolve_diff_exit_code, write_direct_diff_output, @@ -18,6 +23,7 @@ DiffCommandRequest, DiffInvocation, DirectDiffPreparation, + QueryDiffPreparation, VirtualDiffPreparation, VirtualDiffRunOutcome, ) @@ -28,6 +34,31 @@ def run_diff(request: DiffCommandRequest) -> int: """Execute the diff command.""" invocation: DiffInvocation = resolve_diff_invocation(request=request) + if is_query_diff_request(request=request): + query_preparation: QueryDiffPreparation = prepare_query_diff( + request=request, + invocation=invocation, + ) + query_result: DiffExecutionResult = execute_query_diff( + request=request, + preparation=query_preparation, + ) + write_direct_diff_output( + request=request, + preparation=DirectDiffPreparation( + from_target="left query", + to_target="right query", + adapter=query_preparation.adapter, + left_project=None, + right_project=None, + selected_names=("raw queries",), + connection_config=query_preparation.connection_config, + effective_max_column_examples=query_preparation.effective_max_column_examples, + effective_max_row_only_examples=query_preparation.effective_max_row_only_examples, + ), + result=query_result, + ) + return resolve_diff_exit_code(query_result) if invocation.is_virtual_mode: virtual_preparation: VirtualDiffPreparation = prepare_virtual_diff( request=request, diff --git a/src/sqlbuild/cli/commands/models.py b/src/sqlbuild/cli/commands/models.py index f9c22dd059..865f42cd65 100644 --- a/src/sqlbuild/cli/commands/models.py +++ b/src/sqlbuild/cli/commands/models.py @@ -671,8 +671,8 @@ class DiffCommandRequest: project_dir: Path | None no_color: bool no_sql_validation: bool - from_name: str - to_name: str + from_name: str | None + to_name: str | None full: bool schema_only: bool bounded: str | None @@ -689,6 +689,15 @@ class DiffCommandRequest: verbose: bool = False cli_vars: dict[str, object] | None = None allow_partial_diff: bool = False + selected_target: str | None = None + left_query: str | None = None + left_query_file: Path | None = None + right_query: str | None = None + right_query_file: Path | None = None + unique_key_override: tuple[str, ...] = () + unkeyed: bool = False + excluded_columns_override: tuple[str, ...] = () + tolerance_overrides: tuple[str, ...] = () @dataclass(frozen=True) @@ -715,6 +724,23 @@ class DirectDiffPreparation: effective_max_row_only_examples: int +@dataclass(frozen=True) +class QueryDiffPreparation: + """Resolved raw-query diff adapter, connection, SQL, and limits.""" + + adapter: BaseAdapter + connection_config: dict[str, object] + left_sql: str + right_sql: str + selected_target: str | None + database: str | None + schema: str + run_id: str + artifact_ttl: str + effective_max_column_examples: int + effective_max_row_only_examples: int + + @dataclass(frozen=True) class VirtualDiffPreparation: """Resolved virtual diff adapter, connection, reporters, and sample limits.""" diff --git a/src/sqlbuild/compiler/discovery/_helpers/yml/project.py b/src/sqlbuild/compiler/discovery/_helpers/yml/project.py index e8e4bffee7..45f144c496 100644 --- a/src/sqlbuild/compiler/discovery/_helpers/yml/project.py +++ b/src/sqlbuild/compiler/discovery/_helpers/yml/project.py @@ -48,6 +48,7 @@ CursorsConfig, DbtConfig, DefaultsConfig, + DiffConfig, FutureCursorsConfig, JanitorConfig, LifecycleEventSinkFilterConfig, @@ -127,6 +128,7 @@ def load_project_config(*, project_dir: Path) -> ProjectConfig: payload=payload.get("microbatches"), file_path=file_path ) defaults: DefaultsConfig = _load_defaults(payload=payload.get("defaults"), file_path=file_path) + diff: DiffConfig = _load_diff(payload=payload.get("diff"), file_path=file_path) materialization_defaults: MaterializationDefaultsConfig = _load_materialization_defaults( payload=payload.get("materialization_defaults"), file_path=file_path ) @@ -167,6 +169,7 @@ def load_project_config(*, project_dir: Path) -> ProjectConfig: cursors=cursors, microbatches=microbatches, defaults=defaults, + diff=diff, materialization_defaults=materialization_defaults, path_defaults=path_defaults, vars=vars_map, @@ -179,6 +182,24 @@ def load_project_config(*, project_dir: Path) -> ProjectConfig: ) +def _load_diff(*, payload: object, file_path: Path) -> DiffConfig: + mapping: dict[str, object] = _coerce_mapping(payload=payload, label="diff", file_path=file_path) + key: str = "query_artifact_ttl" + _validate_allowed_keys( + mapping=mapping, + allowed_keys=frozenset({key}), + label="diff", + file_path=file_path, + ) + value: str = _optional_str(payload=mapping, key=key) or "24h" + duration: Duration | None = Duration.parse(value) + if duration is None or duration.has_calendar_component: + raise ProjectConfigError( + f"{file_path} diff.{key} must be a positive fixed duration such as '24h'" + ) + return DiffConfig(query_artifact_ttl=value) + + def _load_sinks(*, payload: object, file_path: Path) -> SinksConfig: if payload is None: return SinksConfig() diff --git a/src/sqlbuild/compiler/fingerprints/constants.py b/src/sqlbuild/compiler/fingerprints/constants.py index 6d03781c9a..2fb473e3dc 100644 --- a/src/sqlbuild/compiler/fingerprints/constants.py +++ b/src/sqlbuild/compiler/fingerprints/constants.py @@ -30,6 +30,7 @@ NODE_TYPE_ASSET: str = "asset" NODE_TYPE_CHECK: str = "check" NODE_TYPE_HOOK: str = "hook" +NODE_TYPE_QUERY_DIFF_ARTIFACT: str = "query_diff_artifact" FINGERPRINT_COLUMNS: tuple[str, ...] = ( COLUMN_NODE_TYPE, diff --git a/src/sqlbuild/executor/diff/_helpers/config.py b/src/sqlbuild/executor/diff/_helpers/config.py index 313d9543a0..0ec3179925 100644 --- a/src/sqlbuild/executor/diff/_helpers/config.py +++ b/src/sqlbuild/executor/diff/_helpers/config.py @@ -65,6 +65,42 @@ def parse_row_diff_tolerances( return RowDiffTolerances(by_type=by_type, by_column=by_column) +def parse_cli_tolerance_overrides(*, values: tuple[str, ...]) -> RowDiffTolerances: + """Parse repeated COLUMN:KIND=VALUE CLI tolerance specifications.""" + + raw_by_column: dict[str, dict[str, str]] = {} + for value in values: + column_and_kind, separator, amount = value.partition("=") + column, kind_separator, kind = column_and_kind.partition(":") + if ( + not separator + or not kind_separator + or not column + or kind not in ROW_DIFF_TOLERANCE_KEYS + or not amount + ): + raise ExecutorInputError( + "diff --tolerance must use COLUMN:absolute=VALUE or COLUMN:relative=VALUE", + code="X408", + ) + raw_by_column.setdefault(column, {})[kind] = amount + return parse_row_diff_tolerances(raw={"by_column": raw_by_column}) + + +def merge_row_diff_tolerances( + *, base: RowDiffTolerances, override: RowDiffTolerances | None +) -> RowDiffTolerances: + """Overlay invocation tolerance rules on compiled model rules.""" + + if override is None: + return base + return RowDiffTolerances( + by_type={**base.by_type, **override.by_type}, + by_column={**base.by_column, **override.by_column}, + sampling=base.sampling, + ) + + def _parse_tolerance_section( *, raw: object, diff --git a/src/sqlbuild/executor/diff/_helpers/execution.py b/src/sqlbuild/executor/diff/_helpers/execution.py index f866365939..bba65814b8 100644 --- a/src/sqlbuild/executor/diff/_helpers/execution.py +++ b/src/sqlbuild/executor/diff/_helpers/execution.py @@ -7,6 +7,7 @@ from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.adapter.contract.models import ( + ColumnInfo, RowDiffCoverage, RowDiffResult, RowDiffSampling, @@ -16,6 +17,7 @@ from sqlbuild.errors.contracts.exceptions import ExecutorInputError from sqlbuild.executor.diff._helpers.bounds import resolve_bounded_cursors from sqlbuild.executor.diff._helpers.config import ( + merge_row_diff_tolerances, parse_row_diff_tolerances, resolve_row_diff_sampling, ) @@ -24,6 +26,7 @@ get_unique_key, qualified_name, ) +from sqlbuild.executor.diff.constants import DIFF_INPUT_KIND_QUERY from sqlbuild.executor.diff.models import DiffExecutionOptions, ModelDiffResult @@ -41,7 +44,11 @@ def execute_model_diff( left_relation: str = qualified_name(adapter=adapter, model=left_model) right_relation: str = qualified_name(adapter=adapter, model=right_model) unique_key: tuple[str, ...] = () - if _model_has_unique_key(right_model): + if options.unkeyed: + unique_key = () + elif options.unique_key_override: + unique_key = options.unique_key_override + elif _model_has_unique_key(right_model): unique_key = get_unique_key(right_model) schema_result: SchemaDiffResult = adapter.diff_schema( connection=connection, @@ -83,7 +90,8 @@ def execute_model_diff( unique_key=unique_key, options=options, ) - unique_key = get_unique_key(right_model) if not unique_key else unique_key + if not unique_key and not options.unkeyed: + unique_key = get_unique_key(right_model) return ModelDiffResult( name=name, left_relation=left_relation, @@ -96,9 +104,243 @@ def execute_model_diff( right_only_key_samples=right_only_key_samples, bounded_fallback=bounded_fallback, excluded_columns=excluded_columns, + unkeyed=options.unkeyed, ) +def execute_query_diff( + *, + adapter: BaseAdapter, + connection: Any, + left_relation: str, + right_relation: str, + options: DiffExecutionOptions, +) -> ModelDiffResult: + """Execute one prepared raw-query relation comparison.""" + + unique_key: tuple[str, ...] = options.unique_key_override + schema_result: SchemaDiffResult = adapter.diff_schema( + connection=connection, + left=left_relation, + right=right_relation, + ) + observed_columns: int = max( + schema_result.left_column_count, + schema_result.right_column_count, + ) + if options.max_columns is not None and observed_columns > options.max_columns: + raise ExecutorInputError( + f"query results have {observed_columns} columns, exceeding --max-columns " + f"{options.max_columns}", + code="X305", + ) + result: ModelDiffResult = ModelDiffResult( + name="raw queries", + left_relation=left_relation, + right_relation=right_relation, + unique_key=unique_key, + schema_result=schema_result, + excluded_columns=options.excluded_columns_override, + input_kind=DIFF_INPUT_KIND_QUERY, + unkeyed=options.unkeyed, + ) + if options.schema_only or _schema_differs(schema_result=schema_result): + return result + left_columns: tuple[ColumnInfo, ...] = adapter.describe_relation( + connection=connection, + relation=left_relation, + ) + right_columns: tuple[ColumnInfo, ...] = adapter.describe_relation( + connection=connection, + relation=right_relation, + ) + _validate_query_diff_columns( + left_columns=left_columns, + right_columns=right_columns, + unique_key=unique_key, + excluded_columns=options.excluded_columns_override, + tolerance_columns=tuple( + options.tolerance_overrides.by_column if options.tolerance_overrides is not None else () + ), + unkeyed=options.unkeyed, + ) + canonical_names: dict[str, str] = {column.name.lower(): column.name for column in left_columns} + unique_key = tuple(canonical_names[key.lower()] for key in unique_key) + excluded_columns: tuple[str, ...] = tuple( + canonical_names[column.lower()] for column in options.excluded_columns_override + ) + tolerance_overrides: RowDiffTolerances = options.tolerance_overrides or RowDiffTolerances() + tolerances: RowDiffTolerances = replace( + tolerance_overrides, + by_column={ + canonical_names[column.lower()]: tolerance + for column, tolerance in tolerance_overrides.by_column.items() + }, + ) + result = replace( + result, + unique_key=unique_key, + excluded_columns=excluded_columns, + ) + left_coverage: RowDiffCoverage = adapter.inspect_row_diff_coverage( + connection=connection, + relation=left_relation, + ) + right_coverage: RowDiffCoverage = adapter.inspect_row_diff_coverage( + connection=connection, + relation=right_relation, + ) + if options.unkeyed: + row_result: RowDiffResult = adapter.diff_unkeyed_rows( + connection=connection, + left=left_relation, + right=right_relation, + excluded_columns=excluded_columns, + ) + return replace( + result, + row_result=replace( + row_result, + left_coverage=left_coverage, + right_coverage=right_coverage, + ), + ) + sampling: RowDiffSampling | None = resolve_row_diff_sampling( + raw_row_limit=None, + raw_seed=None, + override=options.sampling_override, + label="raw-query diff", + ) + tolerances = replace(tolerances, sampling=sampling) + row_result = adapter.diff_rows( + connection=connection, + left=left_relation, + right=right_relation, + unique_key=unique_key, + excluded_columns=excluded_columns, + tolerances=tolerances, + ) + row_result = replace( + row_result, + left_coverage=left_coverage, + right_coverage=right_coverage, + ) + unequal_row_samples: tuple[Any, ...] = () + left_only_key_samples: tuple[tuple[tuple[str, object], ...], ...] = () + right_only_key_samples: tuple[tuple[tuple[str, object], ...], ...] = () + if options.collect_samples and row_result.unequal_count > 0: + unequal_row_samples = adapter.sample_unequal_rows( + connection=connection, + left=left_relation, + right=right_relation, + unique_key=unique_key, + excluded_columns=excluded_columns, + tolerances=tolerances, + limit=options.max_column_examples * 5, + ) + if options.collect_samples and row_result.left_only_count > 0: + left_only_key_samples = adapter.sample_side_only_rows( + connection=connection, + left=left_relation, + right=right_relation, + unique_key=unique_key, + side="left", + limit=options.max_row_only_examples, + sampling=sampling, + ) + if options.collect_samples and row_result.right_only_count > 0: + right_only_key_samples = adapter.sample_side_only_rows( + connection=connection, + left=left_relation, + right=right_relation, + unique_key=unique_key, + side="right", + limit=options.max_row_only_examples, + sampling=sampling, + ) + return replace( + result, + row_result=row_result, + unequal_row_samples=tuple(unequal_row_samples), + left_only_key_samples=left_only_key_samples, + right_only_key_samples=right_only_key_samples, + ) + + +def _schema_differs(*, schema_result: SchemaDiffResult) -> bool: + return bool( + schema_result.added_columns + or schema_result.removed_columns + or schema_result.type_changed_columns + ) + + +def _validate_query_diff_columns( + *, + left_columns: tuple[ColumnInfo, ...], + right_columns: tuple[ColumnInfo, ...], + unique_key: tuple[str, ...], + excluded_columns: tuple[str, ...], + tolerance_columns: tuple[str, ...], + unkeyed: bool, +) -> None: + left_names: dict[str, str] = {column.name.lower(): column.name for column in left_columns} + right_names: dict[str, str] = {column.name.lower(): column.name for column in right_columns} + for key in unique_key: + if key.lower() not in left_names or key.lower() not in right_names: + raise ExecutorInputError( + f"raw-query diff key column '{key}' is missing from one or both results", + code="X307", + ) + missing_exclusions: tuple[str, ...] = tuple( + column + for column in excluded_columns + if column.lower() not in left_names or column.lower() not in right_names + ) + if missing_exclusions: + raise ExecutorInputError( + "raw-query diff excluded columns are missing from one or both results: " + + ", ".join(missing_exclusions), + code="X308", + ) + missing_tolerances: tuple[str, ...] = tuple( + column + for column in tolerance_columns + if column.lower() not in left_names or column.lower() not in right_names + ) + if missing_tolerances: + raise ExecutorInputError( + "raw-query diff tolerance columns are missing from one or both results: " + + ", ".join(missing_tolerances), + code="X311", + ) + excluded_column_names: frozenset[str] = frozenset(item.lower() for item in excluded_columns) + intersection: tuple[str, ...] = tuple( + key for key in unique_key if key.lower() in excluded_column_names + ) + if intersection: + raise ExecutorInputError( + "raw-query diff excluded columns intersect --key: " + ", ".join(intersection), + code="X309", + ) + tolerance_column_names: frozenset[str] = frozenset(item.lower() for item in tolerance_columns) + tolerance_key_intersection: tuple[str, ...] = tuple( + key for key in unique_key if key.lower() in tolerance_column_names + ) + if tolerance_key_intersection: + raise ExecutorInputError( + "raw-query diff tolerance columns intersect --key: " + + ", ".join(tolerance_key_intersection), + code="X312", + ) + excluded_names: frozenset[str] = frozenset(column.lower() for column in excluded_columns) + if unkeyed and all(column.name.lower() in excluded_names for column in left_columns): + raise ExecutorInputError( + "unkeyed raw-query diff requires at least one compared column", + code="X310", + ) + + def _execute_model_row_diff( *, adapter: BaseAdapter, @@ -117,8 +359,14 @@ def _execute_model_row_diff( bool, tuple[str, ...], ]: - resolved_unique_key: tuple[str, ...] = unique_key or get_unique_key(right_model) - excluded_columns: tuple[str, ...] = get_row_diff_exclude_columns(right_model) + resolved_unique_key: tuple[str, ...] = unique_key + if not resolved_unique_key and not options.unkeyed: + resolved_unique_key = get_unique_key(right_model) + excluded_columns: tuple[str, ...] = tuple( + dict.fromkeys( + (*get_row_diff_exclude_columns(right_model), *options.excluded_columns_override) + ) + ) intersection: tuple[str, ...] = tuple( key for key in resolved_unique_key if key in excluded_columns ) @@ -136,17 +384,6 @@ def _execute_model_row_diff( model=right_model, bounded=options.bounded, ) - tolerances: RowDiffTolerances = parse_row_diff_tolerances( - raw=right_model.config.values.get("row_diff_tolerances"), - label=f"model '{name}' row_diff_tolerances", - ) - sampling: RowDiffSampling | None = resolve_row_diff_sampling( - raw_row_limit=right_model.config.values.get("row_diff_sample_rows"), - raw_seed=right_model.config.values.get("row_diff_sample_seed"), - override=options.sampling_override, - label=f"model '{name}'", - ) - tolerances = replace(tolerances, sampling=sampling) left_coverage: RowDiffCoverage = adapter.inspect_row_diff_coverage( connection=connection, relation=left_relation, @@ -161,6 +398,46 @@ def _execute_model_row_diff( start_cursor=start_cursor, end_cursor=end_cursor, ) + if options.unkeyed: + row_result: RowDiffResult = adapter.diff_unkeyed_rows( + connection=connection, + left=left_relation, + right=right_relation, + excluded_columns=excluded_columns, + cursor_column=cursor_column, + start_cursor=start_cursor, + end_cursor=end_cursor, + ) + return ( + replace( + row_result, + left_coverage=left_coverage, + right_coverage=right_coverage, + cursor_column=cursor_column, + cursor_start=start_cursor.value if start_cursor is not None else None, + cursor_end=end_cursor.value if end_cursor is not None else None, + ), + (), + (), + (), + bounded_fallback, + excluded_columns, + ) + tolerances: RowDiffTolerances = parse_row_diff_tolerances( + raw=right_model.config.values.get("row_diff_tolerances"), + label=f"model '{name}' row_diff_tolerances", + ) + tolerances = merge_row_diff_tolerances( + base=tolerances, + override=options.tolerance_overrides, + ) + sampling: RowDiffSampling | None = resolve_row_diff_sampling( + raw_row_limit=right_model.config.values.get("row_diff_sample_rows"), + raw_seed=right_model.config.values.get("row_diff_sample_seed"), + override=options.sampling_override, + label=f"model '{name}'", + ) + tolerances = replace(tolerances, sampling=sampling) row_result: RowDiffResult = adapter.diff_rows( connection=connection, left=left_relation, diff --git a/src/sqlbuild/executor/diff/_helpers/query_artifacts.py b/src/sqlbuild/executor/diff/_helpers/query_artifacts.py new file mode 100644 index 0000000000..094d1547cf --- /dev/null +++ b/src/sqlbuild/executor/diff/_helpers/query_artifacts.py @@ -0,0 +1,309 @@ +"""Run-owned raw-query diff artifact lifecycle.""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import UTC, datetime +from typing import Any + +from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.adapter.contract.classes.statement_recorder import StatementRecorder +from sqlbuild.adapter.contract.models import ColumnInfo, RelationInfo, RelationLookup +from sqlbuild.adapter.relations.main.relation_lookup import build_relation_lookup +from sqlbuild.compiler.fingerprints.constants import ( + FINGERPRINT_TABLE_NAME, + NODE_TYPE_QUERY_DIFF_ARTIFACT, +) +from sqlbuild.compiler.fingerprints.main.compute_query_hash import compute_query_hash +from sqlbuild.compiler.fingerprints.main.read import read_latest_fingerprints +from sqlbuild.compiler.fingerprints.main.write import write_fingerprint +from sqlbuild.compiler.fingerprints.models import Fingerprint, FingerprintSet +from sqlbuild.errors.contracts.exceptions import ExecutorInputError +from sqlbuild.executor.diff.models import ( + QueryDiffArtifact, + QueryDiffArtifactCleanupResult, + QueryDiffArtifactInspection, +) + +_ARTIFACT_PATTERN: re.Pattern[str] = re.compile( + r"^__sqlbuild_query_diff_(?P\d{8}T\d{6}Z_[0-9a-f]{12})_(?Pleft|right)$", + flags=re.IGNORECASE, +) +_ARTIFACT_VERSION: int = 1 + + +def build_query_diff_artifact( + *, + adapter: BaseAdapter, + run_id: str, + side: str, + database: str | None, + schema: str, +) -> QueryDiffArtifact: + """Build one collision-resistant artifact identity in the active target schema.""" + + name: str = f"__sqlbuild_query_diff_{run_id}_{side}" + relation: str | None = adapter.render_qualified_name( + database=database, + schema=schema, + name=name, + ) + if relation is None: + raise ExecutorInputError("query diff requires an active target schema", code="X306") + return QueryDiffArtifact( + side=side, + run_id=run_id, + database=database, + schema=schema, + name=name, + relation=relation, + node_name=f"{run_id}:{side}", + ) + + +def is_query_diff_artifact_name(name: str) -> bool: + """Return whether a relation uses the strict reserved query-diff name format.""" + + return _ARTIFACT_PATTERN.fullmatch(name) is not None + + +def materialize_query_diff_artifact( + *, + adapter: BaseAdapter, + connection: Any, + artifact: QueryDiffArtifact, + sql: str, + expires_at: datetime, +) -> None: + """Materialize a query and publish fingerprint ownership or clean up on failure.""" + + recorder: StatementRecorder = StatementRecorder() + adapter.create_table_as( + connection=connection, + destination=artifact.relation, + sql=sql, + statement_recorder=recorder, + ) + try: + columns: tuple[ColumnInfo, ...] = adapter.describe_relation( + connection=connection, + relation=artifact.relation, + ) + definition_hash: str = compute_query_hash(sql) + metadata_json: str = json.dumps( + { + "artifact_version": _ARTIFACT_VERSION, + "expires_at": expires_at.astimezone(UTC).isoformat(), + "side": artifact.side, + }, + sort_keys=True, + separators=(",", ":"), + ) + write_fingerprint( + connection=connection, + execute=adapter.execute, + database=artifact.database, + schema=artifact.schema, + fingerprint=Fingerprint( + node_type=NODE_TYPE_QUERY_DIFF_ARTIFACT, + node_name=artifact.node_name, + target_database=artifact.database, + target_schema=artifact.schema, + target_name=artifact.name, + run_id=artifact.run_id, + definition_hash=definition_hash, + version_hash=definition_hash, + schema_fingerprint=_schema_fingerprint(columns=columns), + definition="", + metadata_json=metadata_json, + ts=datetime.now(UTC), + ), + render_qualified_name=adapter.render_qualified_name, + render_framework_type=adapter.render_framework_type, + render_create_table_sql=adapter.render_create_fingerprint_table_sql, + render_create_index_sqls=adapter.render_create_fingerprint_index_sqls, + ) + except Exception: + adapter.drop( + connection=connection, + destination=artifact.relation, + statement_recorder=recorder, + ) + raise + + +def cleanup_query_diff_artifact( + *, + adapter: BaseAdapter, + connection: Any, + artifact: QueryDiffArtifact, +) -> None: + """Drop one artifact while retaining immutable ownership evidence.""" + + adapter.drop( + connection=connection, + destination=artifact.relation, + statement_recorder=StatementRecorder(), + ) + + +def cleanup_expired_query_diff_artifacts( + *, + adapter: BaseAdapter, + connection: Any, + database: str | None, + schema: str, + now: datetime, +) -> QueryDiffArtifactCleanupResult: + """Remove expired physical artifacts with matching ownership fingerprints.""" + + inspection: QueryDiffArtifactInspection = inspect_query_diff_artifacts( + adapter=adapter, + connection=connection, + database=database, + schema=schema, + now=now, + ) + cleaned: list[str] = [] + for artifact in inspection.expired: + cleanup_query_diff_artifact( + adapter=adapter, + connection=connection, + artifact=artifact, + ) + cleaned.append(artifact.relation) + return QueryDiffArtifactCleanupResult( + cleaned=tuple(cleaned), + untracked=inspection.untracked, + ) + + +def inspect_query_diff_artifacts( + *, + adapter: BaseAdapter, + connection: Any, + database: str | None, + schema: str, + now: datetime, +) -> QueryDiffArtifactInspection: + """Classify strict-name query artifacts using immutable fingerprint ownership.""" + + relations: tuple[RelationInfo, ...] = adapter.list_relations( + connection=connection, + database=database, + schemas=(schema,), + ) + candidates: tuple[RelationInfo, ...] = tuple( + relation for relation in relations if _ARTIFACT_PATTERN.fullmatch(relation.name) + ) + if not candidates: + return QueryDiffArtifactInspection() + lookup: RelationLookup = build_relation_lookup( + adapter=adapter, + connection=connection, + locations=((database, schema, FINGERPRINT_TABLE_NAME),), + ) + fingerprints: FingerprintSet = read_latest_fingerprints( + connection=connection, + execute=adapter.execute, + table_exists=lookup.exists( + database=database, + schema=schema, + name=FINGERPRINT_TABLE_NAME, + ), + database=database, + schema=schema, + render_qualified_name=adapter.render_qualified_name, + render_read_latest_sql=adapter.render_read_latest_fingerprints_sql, + ) + owned: dict[str, Fingerprint] = { + fingerprint.target_name.lower(): fingerprint + for (node_type, _), fingerprint in (fingerprints.fingerprints_by_identity or {}).items() + if node_type == NODE_TYPE_QUERY_DIFF_ARTIFACT and fingerprint.target_name is not None + } + expired: list[QueryDiffArtifact] = [] + untracked: list[RelationInfo] = [] + for relation in candidates: + fingerprint: Fingerprint | None = owned.get(relation.name.lower()) + match: re.Match[str] | None = _ARTIFACT_PATTERN.fullmatch(relation.name) + if match is None: + continue + artifact: QueryDiffArtifact = build_query_diff_artifact( + adapter=adapter, + run_id=match.group("run_id"), + side=match.group("side").lower(), + database=database, + schema=schema, + ) + metadata: dict[str, object] | None = ( + _fingerprint_metadata(fingerprint=fingerprint) if fingerprint is not None else None + ) + if ( + fingerprint is None + or metadata is None + or not _fingerprint_matches_artifact( + fingerprint=fingerprint, + metadata=metadata, + relation=relation, + artifact=artifact, + ) + ): + untracked.append(relation) + continue + expires_at: datetime | None = _metadata_expiry(metadata=metadata) + if expires_at is None: + untracked.append(relation) + continue + if expires_at <= now: + expired.append(artifact) + return QueryDiffArtifactInspection( + expired=tuple(expired), + untracked=tuple(untracked), + ) + + +def _schema_fingerprint(*, columns: tuple[ColumnInfo, ...]) -> str: + payload: str = "\n".join(f"{column.name}:{column.type}" for column in columns) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def _fingerprint_matches_artifact( + *, + fingerprint: Fingerprint, + metadata: dict[str, object], + relation: RelationInfo, + artifact: QueryDiffArtifact, +) -> bool: + return ( + fingerprint.node_type == NODE_TYPE_QUERY_DIFF_ARTIFACT + and fingerprint.node_name.lower() == artifact.node_name.lower() + and fingerprint.run_id.lower() == artifact.run_id.lower() + and (fingerprint.target_database or "").lower() == (relation.database or "").lower() + and (fingerprint.target_schema or "").lower() == (relation.schema or "").lower() + and (fingerprint.target_name or "").lower() == relation.name.lower() + and metadata.get("artifact_version") == _ARTIFACT_VERSION + and metadata.get("side") == artifact.side + ) + + +def _fingerprint_metadata(*, fingerprint: Fingerprint) -> dict[str, object] | None: + try: + payload: object = json.loads(fingerprint.metadata_json) + if not isinstance(payload, dict): + return None + return {str(key): value for key, value in payload.items()} + except (ValueError, TypeError, json.JSONDecodeError): + return None + + +def _metadata_expiry(*, metadata: dict[str, object]) -> datetime | None: + raw: object = metadata.get("expires_at") + if not isinstance(raw, str): + return None + try: + parsed: datetime = datetime.fromisoformat(raw) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) diff --git a/src/sqlbuild/executor/diff/classes/query_artifact_lifecycle.py b/src/sqlbuild/executor/diff/classes/query_artifact_lifecycle.py new file mode 100644 index 0000000000..73f32405da --- /dev/null +++ b/src/sqlbuild/executor/diff/classes/query_artifact_lifecycle.py @@ -0,0 +1,105 @@ +"""Public query-diff artifact lifecycle service.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.executor.diff._helpers.query_artifacts import ( + build_query_diff_artifact, + cleanup_expired_query_diff_artifacts, + cleanup_query_diff_artifact, + inspect_query_diff_artifacts, + is_query_diff_artifact_name, + materialize_query_diff_artifact, +) +from sqlbuild.executor.diff.models import ( + QueryDiffArtifact, + QueryDiffArtifactCleanupResult, + QueryDiffArtifactInspection, +) + + +class QueryDiffArtifactLifecycle: + """Create, inspect, and clean run-owned query-diff artifacts.""" + + @staticmethod + def build( + *, + adapter: BaseAdapter, + run_id: str, + side: str, + database: str | None, + schema: str, + ) -> QueryDiffArtifact: + return build_query_diff_artifact( + adapter=adapter, + run_id=run_id, + side=side, + database=database, + schema=schema, + ) + + @staticmethod + def materialize( + *, + adapter: BaseAdapter, + connection: Any, + artifact: QueryDiffArtifact, + sql: str, + expires_at: datetime, + ) -> None: + materialize_query_diff_artifact( + adapter=adapter, + connection=connection, + artifact=artifact, + sql=sql, + expires_at=expires_at, + ) + + @staticmethod + def cleanup(*, adapter: BaseAdapter, connection: Any, artifact: QueryDiffArtifact) -> None: + cleanup_query_diff_artifact( + adapter=adapter, + connection=connection, + artifact=artifact, + ) + + @staticmethod + def cleanup_expired( + *, + adapter: BaseAdapter, + connection: Any, + database: str | None, + schema: str, + now: datetime, + ) -> QueryDiffArtifactCleanupResult: + return cleanup_expired_query_diff_artifacts( + adapter=adapter, + connection=connection, + database=database, + schema=schema, + now=now, + ) + + @staticmethod + def inspect( + *, + adapter: BaseAdapter, + connection: Any, + database: str | None, + schema: str, + now: datetime, + ) -> QueryDiffArtifactInspection: + return inspect_query_diff_artifacts( + adapter=adapter, + connection=connection, + database=database, + schema=schema, + now=now, + ) + + @staticmethod + def is_artifact_name(name: str) -> bool: + return is_query_diff_artifact_name(name) diff --git a/src/sqlbuild/executor/diff/constants.py b/src/sqlbuild/executor/diff/constants.py index fb712402c1..5387092604 100644 --- a/src/sqlbuild/executor/diff/constants.py +++ b/src/sqlbuild/executor/diff/constants.py @@ -2,6 +2,8 @@ ROW_DIFF_ABSOLUTE_TOLERANCE_KEY: str = "absolute" ROW_DIFF_RELATIVE_TOLERANCE_KEY: str = "relative" +DIFF_INPUT_KIND_MODEL: str = "model" +DIFF_INPUT_KIND_QUERY: str = "query" ROW_DIFF_TOLERANCE_KEYS: frozenset[str] = frozenset( (ROW_DIFF_ABSOLUTE_TOLERANCE_KEY, ROW_DIFF_RELATIVE_TOLERANCE_KEY) ) diff --git a/src/sqlbuild/executor/diff/main/config.py b/src/sqlbuild/executor/diff/main/config.py new file mode 100644 index 0000000000..6177f157ad --- /dev/null +++ b/src/sqlbuild/executor/diff/main/config.py @@ -0,0 +1,12 @@ +"""Public diff invocation configuration entrypoints.""" + +from sqlbuild.adapter.contract.models import RowDiffTolerances +from sqlbuild.executor.diff._helpers.config import ( + parse_cli_tolerance_overrides as _parse_cli_tolerance_overrides, +) + + +def parse_cli_tolerance_overrides(*, values: tuple[str, ...]) -> RowDiffTolerances: + """Parse repeated invocation-level per-column tolerance rules.""" + + return _parse_cli_tolerance_overrides(values=values) diff --git a/src/sqlbuild/executor/diff/main/execute_query.py b/src/sqlbuild/executor/diff/main/execute_query.py new file mode 100644 index 0000000000..aed9182687 --- /dev/null +++ b/src/sqlbuild/executor/diff/main/execute_query.py @@ -0,0 +1,28 @@ +"""Execute one prepared query diff.""" + +from __future__ import annotations + +from typing import Any + +from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter +from sqlbuild.executor.diff._helpers.execution import execute_query_diff as _execute_query_diff +from sqlbuild.executor.diff.models import DiffExecutionOptions, ModelDiffResult + + +def execute_query_diff( + *, + adapter: BaseAdapter, + connection: Any, + left_relation: str, + right_relation: str, + options: DiffExecutionOptions, +) -> ModelDiffResult: + """Execute one prepared raw-query relation comparison.""" + + return _execute_query_diff( + adapter=adapter, + connection=connection, + left_relation=left_relation, + right_relation=right_relation, + options=options, + ) diff --git a/src/sqlbuild/executor/diff/models.py b/src/sqlbuild/executor/diff/models.py index c51f6e82c2..aa6c80b7d3 100644 --- a/src/sqlbuild/executor/diff/models.py +++ b/src/sqlbuild/executor/diff/models.py @@ -4,7 +4,43 @@ from dataclasses import dataclass, field -from sqlbuild.adapter.contract.models import RowDiffResult, RowDiffSampleRow, SchemaDiffResult +from sqlbuild.adapter.contract.models import ( + RelationInfo, + RowDiffResult, + RowDiffSampleRow, + RowDiffTolerances, + SchemaDiffResult, +) +from sqlbuild.executor.diff.constants import DIFF_INPUT_KIND_MODEL + + +@dataclass(frozen=True) +class QueryDiffArtifact: + """One physical query result and its immutable ownership identity.""" + + side: str + run_id: str + database: str | None + schema: str + name: str + relation: str + node_name: str + + +@dataclass(frozen=True) +class QueryDiffArtifactInspection: + """Expired owned artifacts and reserved-name relations requiring manual review.""" + + expired: tuple[QueryDiffArtifact, ...] = () + untracked: tuple[RelationInfo, ...] = () + + +@dataclass(frozen=True) +class QueryDiffArtifactCleanupResult: + """Reconciliation outcome for one target schema.""" + + cleaned: tuple[str, ...] = () + untracked: tuple[RelationInfo, ...] = () @dataclass(frozen=True) @@ -28,6 +64,10 @@ class DiffExecutionOptions: max_models: int | None = None max_columns: int | None = None sampling_override: RowDiffSamplingOverride = field(default_factory=RowDiffSamplingOverride) + unique_key_override: tuple[str, ...] = () + unkeyed: bool = False + excluded_columns_override: tuple[str, ...] = () + tolerance_overrides: RowDiffTolerances | None = None @dataclass(frozen=True) @@ -47,6 +87,8 @@ class ModelDiffResult: ) bounded_fallback: bool = False excluded_columns: tuple[str, ...] = field(default_factory=tuple) + input_kind: str = DIFF_INPUT_KIND_MODEL + unkeyed: bool = False @dataclass(frozen=True) diff --git a/src/sqlbuild/executor/janitor/_helpers/classification.py b/src/sqlbuild/executor/janitor/_helpers/classification.py index a8145b64ff..1675314741 100644 --- a/src/sqlbuild/executor/janitor/_helpers/classification.py +++ b/src/sqlbuild/executor/janitor/_helpers/classification.py @@ -15,6 +15,8 @@ is_scenario_artifact_physical_name, ) from sqlbuild.compiler.source_freshness.constants import SOURCE_FRESHNESS_TABLE_NAME +from sqlbuild.executor.diff.classes.query_artifact_lifecycle import QueryDiffArtifactLifecycle +from sqlbuild.executor.diff.models import QueryDiffArtifactInspection from sqlbuild.executor.janitor._helpers.plan import ( collect_desired_keys, collect_source_schemas, @@ -28,6 +30,7 @@ from sqlbuild.executor.janitor.models import ( JanitorDeleteCandidate, JanitorDirectStatePruneCandidate, + JanitorQueryDiffArtifactCandidate, JanitorRelationClassification, JanitorRelationKey, JanitorSkippedRelation, @@ -35,6 +38,47 @@ ) +def collect_query_diff_artifact_candidates( + *, + adapter: BaseAdapter, + connection: Any, + target_schemas: set[tuple[str | None, str]], + now: datetime, +) -> tuple[tuple[JanitorQueryDiffArtifactCandidate, ...], tuple[JanitorSkippedRelation, ...]]: + """Collect expired owned query artifacts and report untracked reserved names.""" + + candidates: list[JanitorQueryDiffArtifactCandidate] = [] + skipped: list[JanitorSkippedRelation] = [] + for database, schema in sorted(target_schemas, key=lambda value: (value[0] or "", value[1])): + inspection: QueryDiffArtifactInspection = QueryDiffArtifactLifecycle.inspect( + adapter=adapter, + connection=connection, + database=database, + schema=schema, + now=now, + ) + for artifact in inspection.expired: + candidates.append( + JanitorQueryDiffArtifactCandidate( + key=JanitorRelationKey( + database=artifact.database, + schema=artifact.schema, + name=artifact.name, + ) + ) + ) + for relation in inspection.untracked: + key: JanitorRelationKey = build_relation_key(relation) + skipped.append( + JanitorSkippedRelation( + key=key, + relation=relation, + reason="reserved query-diff name lacks matching ownership evidence", + ) + ) + return tuple(candidates), tuple(skipped) + + def gather_janitor_warehouse_facts( *, project: CompiledProject, @@ -187,6 +231,8 @@ def _relation_skip_reason( effective_exclude_patterns: tuple[str, ...], delete_tracked_only: bool, ) -> str | None: + if QueryDiffArtifactLifecycle.is_artifact_name(relation_key.name): + return "query-diff artifacts use fingerprint expiry cleanup" if relation_key in protected_relation_keys: return protection_reasons.get( relation_key, diff --git a/src/sqlbuild/executor/janitor/main/execute.py b/src/sqlbuild/executor/janitor/main/execute.py index 0f2f82448c..e603f5094a 100644 --- a/src/sqlbuild/executor/janitor/main/execute.py +++ b/src/sqlbuild/executor/janitor/main/execute.py @@ -17,6 +17,7 @@ JanitorExpiredLockCandidate, JanitorExpiredVirtualEnvironmentCandidate, JanitorPlan, + JanitorQueryDiffArtifactCandidate, JanitorStateBackupCandidate, JanitorVirtualStatePruneCandidate, ) @@ -86,6 +87,24 @@ def _execute_janitor_plan( if_exists=True, statement_recorder=recorder, ) + query_artifact_candidate: JanitorQueryDiffArtifactCandidate + for query_artifact_candidate in plan.query_diff_artifact_candidates: + with OperationLifecycle(operation_kind="janitor", operation_name="janitor_cleanup_action"): + query_artifact_destination: str | None = adapter.render_qualified_name( + database=query_artifact_candidate.key.database, + schema=query_artifact_candidate.key.schema, + name=query_artifact_candidate.key.name, + ) + adapter.drop( + connection=connection, + destination=( + query_artifact_destination + if query_artifact_destination is not None + else query_artifact_candidate.key.display_name() + ), + if_exists=True, + statement_recorder=recorder, + ) deleted_checkpoints: tuple[JanitorCheckpointCandidate, ...] = apply_janitor_deletions( candidates=plan.checkpoint_candidates, delete=delete_checkpoint ) @@ -118,6 +137,7 @@ def _execute_janitor_plan( ) return JanitorExecutionResult( deleted=() if plan.direct_mode else plan.candidates, + deleted_query_diff_artifacts=plan.query_diff_artifact_candidates, deleted_checkpoints=deleted_checkpoints, deleted_detached_virtual_environments=deleted_detached_virtual_environments, deleted_expired_virtual_environments=deleted_expired_virtual_environments, @@ -130,6 +150,7 @@ def _execute_janitor_plan( def _janitor_action_count(plan: JanitorPlan) -> int: relation_count: int = 0 if plan.direct_mode else len(plan.candidates) + relation_count += len(plan.query_diff_artifact_candidates) return relation_count + sum( len(candidates) for candidates in ( diff --git a/src/sqlbuild/executor/janitor/main/plan.py b/src/sqlbuild/executor/janitor/main/plan.py index 9e8874ec5f..ae7f92df3e 100644 --- a/src/sqlbuild/executor/janitor/main/plan.py +++ b/src/sqlbuild/executor/janitor/main/plan.py @@ -7,8 +7,10 @@ from sqlbuild.adapter.contract.classes.base_adapter import BaseAdapter from sqlbuild.compiler.compile.models import CompiledProject +from sqlbuild.executor.diff.classes.query_artifact_lifecycle import QueryDiffArtifactLifecycle from sqlbuild.executor.janitor._helpers.classification import ( collect_direct_state_prune_candidates, + collect_query_diff_artifact_candidates, gather_janitor_warehouse_facts, ) from sqlbuild.executor.janitor._helpers.plan import collect_target_schemas @@ -50,11 +52,29 @@ def build_janitor_plan( target_schemas: set[tuple[str | None, str | None]] = set(managed_target_schemas) target_schemas.update((key.database, key.schema) for key in scope.protected_relation_keys) target_schemas.update((key.database, key.schema) for key in scope.scan_relation_keys) + query_artifact_schemas: set[tuple[str | None, str]] = { + (database, schema) for database, schema in target_schemas if schema is not None + } + if project.effective_target_schema is not None: + query_artifact_schemas.add( + (project.effective_target_database, project.effective_target_schema) + ) + now: datetime = datetime.now(UTC) + ( + query_diff_artifact_candidates, + query_diff_artifact_skipped, + ) = collect_query_diff_artifact_candidates( + adapter=adapter, + connection=connection, + target_schemas=query_artifact_schemas, + now=now, + ) if not target_schemas: return JanitorPlan( target_name=project.effective_target_name, retention_days=retention_days, direct_mode=direct.enabled, + query_diff_artifact_candidates=query_diff_artifact_candidates, checkpoint_candidates=state.checkpoint_candidates, detached_virtual_environment_candidates=(state.detached_virtual_environment_candidates), expired_virtual_environment_candidates=(state.expired_virtual_environment_candidates), @@ -62,6 +82,8 @@ def build_janitor_plan( expired_lock_candidates=state.expired_lock_candidates, virtual_state_prune_candidates=state.virtual_state_prune_candidates, direct_state_prune_candidates=(), + skipped_relations=query_diff_artifact_skipped, + scanned_schema_count=len(query_artifact_schemas), age_metadata_supported=adapter.supports_relation_age_metadata(), ) @@ -84,7 +106,6 @@ def build_janitor_plan( ) ) inspection.completed(metadata={"item_count": len(target_schemas)}) - now: datetime = datetime.now(UTC) age_supported: bool = adapter.supports_relation_age_metadata() schemas: JanitorSchemaClassification = classify_target_schemas( target_schemas=target_schemas, @@ -104,6 +125,7 @@ def build_janitor_plan( retention_days=retention_days, direct_mode=direct.enabled, candidates=schemas.candidates, + query_diff_artifact_candidates=query_diff_artifact_candidates, checkpoint_candidates=state.checkpoint_candidates, detached_virtual_environment_candidates=state.detached_virtual_environment_candidates, expired_virtual_environment_candidates=state.expired_virtual_environment_candidates, @@ -111,9 +133,16 @@ def build_janitor_plan( expired_lock_candidates=state.expired_lock_candidates, virtual_state_prune_candidates=state.virtual_state_prune_candidates, direct_state_prune_candidates=direct_state_prune_candidates, - skipped_relations=schemas.skipped_relations, + skipped_relations=( + *tuple( + skipped + for skipped in schemas.skipped_relations + if not QueryDiffArtifactLifecycle.is_artifact_name(skipped.key.name) + ), + *query_diff_artifact_skipped, + ), skipped_schemas=schemas.skipped_schemas, blocked_schemas=schemas.blocked_schemas, - scanned_schema_count=len(target_schemas), + scanned_schema_count=len(target_schemas | set(query_artifact_schemas)), age_metadata_supported=age_supported, ) diff --git a/src/sqlbuild/executor/janitor/models.py b/src/sqlbuild/executor/janitor/models.py index 0c9fe52165..ebbcdebae1 100644 --- a/src/sqlbuild/executor/janitor/models.py +++ b/src/sqlbuild/executor/janitor/models.py @@ -37,6 +37,13 @@ class JanitorDeleteCandidate: age_timestamp: datetime | None +@dataclass(frozen=True) +class JanitorQueryDiffArtifactCandidate: + """One expired, fingerprint-owned query-diff artifact eligible for deletion.""" + + key: JanitorRelationKey + + @dataclass(frozen=True) class JanitorCheckpointCandidate: """One retained-history checkpoint eligible for pruning.""" @@ -231,6 +238,9 @@ class JanitorPlan: retention_days: int direct_mode: bool = False candidates: tuple[JanitorDeleteCandidate, ...] = field(default_factory=tuple) + query_diff_artifact_candidates: tuple[JanitorQueryDiffArtifactCandidate, ...] = field( + default_factory=tuple + ) checkpoint_candidates: tuple[JanitorCheckpointCandidate, ...] = field(default_factory=tuple) detached_virtual_environment_candidates: tuple[ JanitorDetachedVirtualEnvironmentCandidate, ... @@ -258,6 +268,9 @@ class JanitorExecutionResult: """Result from deleting janitor candidates.""" deleted: tuple[JanitorDeleteCandidate, ...] = field(default_factory=tuple) + deleted_query_diff_artifacts: tuple[JanitorQueryDiffArtifactCandidate, ...] = field( + default_factory=tuple + ) deleted_checkpoints: tuple[JanitorCheckpointCandidate, ...] = field(default_factory=tuple) deleted_detached_virtual_environments: tuple[ JanitorDetachedVirtualEnvironmentCandidate, ... diff --git a/src/sqlbuild/spec/contracts/models.py b/src/sqlbuild/spec/contracts/models.py index 9885b0a2c9..6e4c5bee93 100644 --- a/src/sqlbuild/spec/contracts/models.py +++ b/src/sqlbuild/spec/contracts/models.py @@ -282,6 +282,13 @@ class DefaultsConfig: post_hooks: object | None = None +@dataclass(frozen=True) +class DiffConfig: + """Project-wide data-diff configuration.""" + + query_artifact_ttl: str = "24h" + + @dataclass(frozen=True) class JanitorConfig: """Janitor command defaults.""" @@ -382,6 +389,7 @@ class ProjectConfig: cursors: CursorsConfig = field(default_factory=CursorsConfig) microbatches: MicrobatchesConfig = field(default_factory=MicrobatchesConfig) defaults: DefaultsConfig = field(default_factory=DefaultsConfig) + diff: DiffConfig = field(default_factory=DiffConfig) materialization_defaults: MaterializationDefaultsConfig = field( default_factory=MaterializationDefaultsConfig ) diff --git a/src/sqlbuild/virtual/diff/_helpers/diff.py b/src/sqlbuild/virtual/diff/_helpers/diff.py index aadeca193b..d7d5aa7b60 100644 --- a/src/sqlbuild/virtual/diff/_helpers/diff.py +++ b/src/sqlbuild/virtual/diff/_helpers/diff.py @@ -298,6 +298,10 @@ def execute_virtual_diff_between_relations( max_models=options.max_models, max_columns=options.max_columns, sampling_override=options.sampling_override, + unique_key_override=options.unique_key_override, + unkeyed=options.unkeyed, + excluded_columns_override=options.excluded_columns_override, + tolerance_overrides=options.tolerance_overrides, ), ) finally: diff --git a/src/sqlbuild/virtual/diff/models.py b/src/sqlbuild/virtual/diff/models.py index 6787817c8b..a8fc8974cc 100644 --- a/src/sqlbuild/virtual/diff/models.py +++ b/src/sqlbuild/virtual/diff/models.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field +from sqlbuild.adapter.contract.models import RowDiffTolerances from sqlbuild.compiler.references.types import ExternalSqlReferenceResolver from sqlbuild.executor.diff.models import RowDiffSamplingOverride from sqlbuild.virtual.planner.models import VirtualPlanSemantics @@ -29,6 +30,10 @@ class VirtualDiffOptions: max_models: int | None = None max_columns: int | None = None sampling_override: RowDiffSamplingOverride = field(default_factory=RowDiffSamplingOverride) + unique_key_override: tuple[str, ...] = () + unkeyed: bool = False + excluded_columns_override: tuple[str, ...] = () + tolerance_overrides: RowDiffTolerances | None = None allow_partial_diff: bool = False cli_vars: dict[str, object] | None = None external_sql_reference_resolver: ExternalSqlReferenceResolver | None = None diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py index e7862fc5e6..7f1a24082d 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/diff/_test_types.py @@ -15,6 +15,15 @@ class DiffCommandE2ETestCase: mutation_sql: tuple[str, ...] = field(default_factory=tuple) +@dataclass(frozen=True) +class SingleDiffE2ETestCase: + """One named standalone diff behavior case.""" + + description: str + expected_exit_code: int + expected_fragment: str + + @dataclass(frozen=True) class DiffKeyFailureE2ETestCase: description: str diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_diff.py b/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_diff.py index 8d44e4920b..234c6ea1bd 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_diff.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/diff/test_diff.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import cast +import duckdb import pytest from tests.e2e.src.sqlbuild.cli.commands.main.diff._test_types import ( @@ -15,6 +16,7 @@ DiffKeyFailureE2ETestCase, DiffSamplingPrecedenceE2ETestCase, DiffSamplingSeedE2ETestCase, + SingleDiffE2ETestCase, VirtualDiffE2ETestCase, ) from tests.e2e.src.sqlbuild.cli.commands.main.diff.helpers import ( @@ -433,6 +435,35 @@ expected_exit_code=1, expected_stderr_fragments=("model 'daily_revenue' requires unique_key for row diff",), ), + DiffCommandE2ETestCase( + description="cli key enables model without declared unique key", + command=( + "--no-color", + "diff", + "prod:dev", + "--full", + "--key", + "revenue_date", + "--select", + "daily_revenue", + ), + expected_exit_code=0, + expected_stdout_fragments=("revenue_date", "No changed columns."), + ), + DiffCommandE2ETestCase( + description="explicit unkeyed mode compares model row multiplicities", + command=( + "--no-color", + "diff", + "prod:dev", + "--full", + "--unkeyed", + "--select", + "orders_snapshot", + ), + expected_exit_code=0, + expected_stdout_fragments=("", "multiset rows"), + ), DiffCommandE2ETestCase( description="environment schemas auto-create through real cli build flow", command=( @@ -474,6 +505,404 @@ def test_given_diff_project_when_running_diff_then_behavior_matches_expected( assert fragment in result.stderr, result.stdout + result.stderr +@pytest.mark.parametrize( + "test_case", + [ + DiffCommandE2ETestCase( + description="inline keyed queries report changed values", + mutation_sql=("UPDATE dev.orders_snapshot SET amount_cents = 999 WHERE order_id = 1",), + command=( + "--no-color", + "diff", + "--left-query", + "SELECT order_id, amount_cents FROM prod.orders_snapshot", + "--right-query", + "SELECT order_id, amount_cents FROM dev.orders_snapshot", + "--key", + "order_id", + ), + expected_exit_code=1, + expected_stdout_fragments=( + "selected query pairs: 1", + "order_id=1 | 100 -> 999", + ), + expected_stderr_fragments=( + "Query diff materialization OK", + "Query diff immediate cleanup OK", + ), + ), + DiffCommandE2ETestCase( + description="inline keyed queries support deterministic sampling", + mutation_sql=("UPDATE dev.orders_snapshot SET amount_cents = amount_cents + 10",), + command=( + "--no-color", + "diff", + "--left-query", + "SELECT order_id, amount_cents FROM prod.orders_snapshot", + "--right-query", + "SELECT order_id, amount_cents FROM dev.orders_snapshot", + "--key", + "order_id", + "--sample-rows", + "2", + "--sample-seed", + "7", + ), + expected_exit_code=1, + expected_stdout_fragments=("sampled 2 of 3 union keys", "seed 7"), + ), + DiffCommandE2ETestCase( + description="unkeyed queries preserve duplicate multiplicity", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT * FROM (VALUES (1), (1)) AS orders(order_id)", + "--right-query", + "SELECT * FROM (VALUES (1)) AS orders(order_id)", + "--unkeyed", + ), + expected_exit_code=1, + expected_stdout_fragments=("left query only", "1"), + ), + DiffCommandE2ETestCase( + description="unkeyed queries reserve internal multiplicity aliases safely", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT 0 AS __sqlbuild_count", + "--right-query", + "SELECT CAST(NULL AS INTEGER) AS __sqlbuild_count WHERE FALSE", + "--unkeyed", + ), + expected_exit_code=1, + expected_stdout_fragments=("left query only", "1"), + ), + DiffCommandE2ETestCase( + description="schema drift reports findings without invalid row comparison", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT 1 AS order_id", + "--right-query", + "SELECT 1 AS order_id, 100 AS amount_cents", + "--key", + "order_id", + ), + expected_exit_code=1, + expected_stdout_fragments=("schema differences: 1", "added columns: 1"), + ), + DiffCommandE2ETestCase( + description="query tolerance suppresses numeric differences", + mutation_sql=( + "UPDATE dev.orders_snapshot SET amount_cents = amount_cents + 5 WHERE order_id = 1", + ), + command=( + "--no-color", + "diff", + "--left-query", + "SELECT order_id, amount_cents FROM prod.orders_snapshot", + "--right-query", + "SELECT order_id, amount_cents FROM dev.orders_snapshot", + "--key", + "order_id", + "--tolerance", + "amount_cents:absolute=5", + ), + expected_exit_code=0, + expected_stdout_fragments=("amount_cents absolute=5", "No changed columns."), + ), + DiffCommandE2ETestCase( + description="query exclusions suppress ignored column differences", + mutation_sql=("UPDATE dev.orders_snapshot SET status = 'fulfilled'",), + command=( + "--no-color", + "diff", + "--left-query", + "SELECT order_id, status FROM prod.orders_snapshot", + "--right-query", + "SELECT order_id, status FROM dev.orders_snapshot", + "--key", + "order_id", + "--exclude-column", + "status", + ), + expected_exit_code=0, + expected_stdout_fragments=("Excluded", "status", "No changed columns."), + ), + DiffCommandE2ETestCase( + description="query composite keys preserve row identity", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT customer_id, order_id, amount_cents FROM prod.orders_snapshot", + "--right-query", + "SELECT customer_id, order_id, amount_cents FROM dev.orders_snapshot", + "--key", + "customer_id", + "--key", + "order_id", + ), + expected_exit_code=0, + expected_stdout_fragments=("customer_id, order_id", "No changed columns."), + ), + DiffCommandE2ETestCase( + description="read-only set queries are accepted", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT 1 AS order_id UNION ALL SELECT 2 AS order_id", + "--right-query", + "SELECT 1 AS order_id UNION ALL SELECT 2 AS order_id", + "--key", + "order_id", + ), + expected_exit_code=0, + expected_stdout_fragments=("No changed columns.",), + ), + DiffCommandE2ETestCase( + description="query key integrity rejects duplicate identities", + command=( + "--no-color", + "diff", + "--left-query", + "SELECT * FROM (VALUES (1, 100), (1, 200)) AS orders(order_id, amount_cents)", + "--right-query", + "SELECT * FROM (VALUES (1, 100)) AS orders(order_id, amount_cents)", + "--key", + "order_id", + ), + expected_exit_code=1, + expected_stderr_fragments=("contains duplicate unique_key values",), + ), + ], + ids=lambda case: case.description, +) +def test_given_raw_queries_when_running_diff_then_comparison_matches_model_diff_capabilities( + test_case: DiffCommandE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + build_both_environments(project_dir=project_dir) + for mutation_sql in test_case.mutation_sql: + execute_duckdb(db_path=project_dir / "diff.duckdb", sql=mutation_sql) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=test_case.command, + project_dir=project_dir, + ) + + assert result.returncode == test_case.expected_exit_code, result.stdout + result.stderr + for fragment in test_case.expected_stdout_fragments: + assert fragment in result.stdout, result.stdout + result.stderr + for fragment in test_case.expected_stderr_fragments: + assert fragment in result.stderr, result.stdout + result.stderr + connection: duckdb.DuckDBPyConnection = duckdb.connect(str(project_dir / "diff.duckdb")) + try: + artifact_row: tuple[int] | None = connection.execute( + "SELECT COUNT(*) FROM duckdb_tables() WHERE table_name LIKE '__sqlbuild_query_diff_%'" + ).fetchone() + fingerprint_row: tuple[int] | None = connection.execute( + "SELECT COUNT(*) FROM dev._sqlbuild_fingerprints " + "WHERE node_type = 'query_diff_artifact'" + ).fetchone() + finally: + connection.close() + assert artifact_row == (0,) + assert fingerprint_row == (2,) + + +@pytest.mark.parametrize( + "test_case", + [ + SingleDiffE2ETestCase( + description="query files target and JSON output", + expected_exit_code=0, + expected_fragment="no_differences_found", + ) + ], + ids=lambda case: case.description, +) +def test_given_query_files_target_and_json_output_when_running_diff_then_all_inputs_are_honored( + test_case: SingleDiffE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + build_both_environments(project_dir=project_dir) + (project_dir / "left.sql").write_text( + "SELECT order_id, amount_cents FROM prod.orders_snapshot;\n", + encoding="utf-8", + ) + (project_dir / "right.sql").write_text( + "SELECT order_id, amount_cents FROM dev.orders_snapshot;\n", + encoding="utf-8", + ) + json_path: Path = project_dir / "query-diff.json" + + result: subprocess.CompletedProcess[str] = run_sqb( + command=( + "--no-color", + "diff", + "--left-query-file", + str(project_dir / "left.sql"), + "--right-query-file", + str(project_dir / "right.sql"), + "--target", + "prod", + "--key", + "order_id", + "--json-output", + str(json_path), + ), + project_dir=project_dir, + ) + + assert result.returncode == test_case.expected_exit_code, result.stdout + result.stderr + payload: dict[str, object] = cast(dict[str, object], json.loads(json_path.read_text())) + assert payload["status"] == test_case.expected_fragment + models: list[dict[str, object]] = cast(list[dict[str, object]], payload["models"]) + assert models[0]["input_kind"] == "query" + connection: duckdb.DuckDBPyConnection = duckdb.connect(str(project_dir / "diff.duckdb")) + try: + prod_fingerprint_row: tuple[int] | None = connection.execute( + "SELECT COUNT(*) FROM prod._sqlbuild_fingerprints " + "WHERE node_type = 'query_diff_artifact'" + ).fetchone() + dev_fingerprint_row: tuple[int] | None = connection.execute( + "SELECT COUNT(*) FROM dev._sqlbuild_fingerprints " + "WHERE node_type = 'query_diff_artifact'" + ).fetchone() + finally: + connection.close() + assert prod_fingerprint_row == (2,) + assert dev_fingerprint_row == (0,) + + +@pytest.mark.parametrize( + "test_case", + [ + SingleDiffE2ETestCase( + description="mutating CTE rejection", + expected_exit_code=1, + expected_fragment="must not contain data-changing or administrative statements", + ) + ], + ids=lambda case: case.description, +) +def test_given_mutating_cte_when_running_query_diff_then_validation_rejects_it( + test_case: SingleDiffE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=( + "--no-color", + "diff", + "--left-query", + "WITH changed AS (DELETE FROM orders RETURNING *) SELECT * FROM changed", + "--right-query", + "SELECT 1 AS order_id", + "--key", + "order_id", + ), + project_dir=project_dir, + ) + + assert result.returncode == test_case.expected_exit_code + assert test_case.expected_fragment in result.stderr + + +@pytest.mark.parametrize( + "test_case", + [ + SingleDiffE2ETestCase( + description="duplicate query column rejection", + expected_exit_code=1, + expected_fragment="left query returns duplicate column names: order_id", + ) + ], + ids=lambda case: case.description, +) +def test_given_duplicate_query_columns_when_running_query_diff_then_preflight_rejects_them( + test_case: SingleDiffE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=( + "--no-color", + "diff", + "--left-query", + "SELECT 1 AS order_id, 100 AS order_id", + "--right-query", + "SELECT 1 AS order_id, 100 AS amount_cents", + "--key", + "order_id", + ), + project_dir=project_dir, + ) + + assert result.returncode == test_case.expected_exit_code + assert test_case.expected_fragment in result.stderr + + +@pytest.mark.parametrize( + "test_case", + [ + SingleDiffE2ETestCase( + description="historical fingerprint isolation", + expected_exit_code=0, + expected_fragment="recovery cleanup could not inspect historical ownership evidence", + ) + ], + ids=lambda case: case.description, +) +def test_given_unreadable_historical_fingerprint_when_running_query_diff_then_comparison_continues( + test_case: SingleDiffE2ETestCase, + tmp_path: Path, +) -> None: + project_dir: Path = prepare_diff_project(tmp_path) + build_both_environments(project_dir=project_dir) + execute_duckdb( + db_path=project_dir / "diff.duckdb", + sql=( + "CREATE TABLE dev.__sqlbuild_query_diff_20260917T120000Z_a1b2c3d4e5f6_left " + "AS SELECT 1 AS order_id" + ), + ) + execute_duckdb( + db_path=project_dir / "diff.duckdb", + sql=( + "UPDATE dev._sqlbuild_fingerprints SET definition_b64 = 'not-base64' " + "WHERE node_type = 'model'" + ), + ) + + result: subprocess.CompletedProcess[str] = run_sqb( + command=( + "--no-color", + "diff", + "--left-query", + "SELECT order_id FROM prod.orders_snapshot", + "--right-query", + "SELECT order_id FROM dev.orders_snapshot", + "--key", + "order_id", + ), + project_dir=project_dir, + ) + + assert result.returncode == test_case.expected_exit_code, result.stdout + result.stderr + assert test_case.expected_fragment in result.stderr + assert "No changed columns." in result.stdout + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/integration/src/sqlbuild/executor/diff/_test_types.py b/tests/integration/src/sqlbuild/executor/diff/_test_types.py new file mode 100644 index 0000000000..944e1cd4ad --- /dev/null +++ b/tests/integration/src/sqlbuild/executor/diff/_test_types.py @@ -0,0 +1,11 @@ +"""Test types for query-diff artifact integration coverage.""" + +from dataclasses import dataclass + + +@dataclass(frozen=True) +class QueryArtifactCleanupTestCase: + """One query-artifact cleanup integration case.""" + + description: str + expected_fingerprint_count: int diff --git a/tests/integration/src/sqlbuild/executor/diff/test_query_artifact_cleanup.py b/tests/integration/src/sqlbuild/executor/diff/test_query_artifact_cleanup.py new file mode 100644 index 0000000000..8a57f16b61 --- /dev/null +++ b/tests/integration/src/sqlbuild/executor/diff/test_query_artifact_cleanup.py @@ -0,0 +1,119 @@ +"""Integration coverage for query-diff artifact ownership and crash recovery.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import duckdb +import pytest + +from sqlbuild.adapters.duckdb.classes.duckdb_adapter import DuckDbAdapter +from sqlbuild.compiler.compile.models import CompiledProject +from sqlbuild.compiler.fingerprints.constants import NODE_TYPE_QUERY_DIFF_ARTIFACT +from sqlbuild.executor.diff.classes.query_artifact_lifecycle import QueryDiffArtifactLifecycle +from sqlbuild.executor.diff.models import QueryDiffArtifact +from sqlbuild.executor.janitor.main.execute import execute_janitor_plan +from sqlbuild.executor.janitor.main.plan import build_janitor_plan +from sqlbuild.executor.janitor.models import ( + JanitorDirectModeSettings, + JanitorExecutionResult, + JanitorPlan, +) +from tests.integration.src.sqlbuild.executor.diff._test_types import ( + QueryArtifactCleanupTestCase, +) + + +@pytest.mark.parametrize( + "test_case", + [ + QueryArtifactCleanupTestCase( + description="owned expiry and untracked protection", + expected_fingerprint_count=1, + ) + ], + ids=lambda case: case.description, +) +def test_given_expired_owned_and_untracked_artifacts_when_janitor_runs_then_only_owned_artifact_is_deleted( + test_case: QueryArtifactCleanupTestCase, +) -> None: + adapter: DuckDbAdapter = DuckDbAdapter() + connection: duckdb.DuckDBPyConnection = duckdb.connect(":memory:") + owned: QueryDiffArtifact = QueryDiffArtifactLifecycle.build( + adapter=adapter, + run_id="20260918T142301Z_a1b2c3d4e5f6", + side="left", + database=None, + schema="main", + ) + untracked: QueryDiffArtifact = QueryDiffArtifactLifecycle.build( + adapter=adapter, + run_id="20260918T142302Z_b1c2d3e4f5a6", + side="right", + database=None, + schema="main", + ) + try: + QueryDiffArtifactLifecycle.materialize( + adapter=adapter, + connection=connection, + artifact=owned, + sql="SELECT 1 AS order_id", + expires_at=datetime.now(UTC) - timedelta(hours=1), + ) + normalized_owned_name: str = owned.name.lower() + connection.execute(f"ALTER TABLE {owned.relation} RENAME TO {normalized_owned_name}") + connection.execute(f"CREATE TABLE {untracked.relation} AS SELECT 2 AS order_id") + project: CompiledProject = CompiledProject( + run_id="run-1", + effective_target_name="dev", + effective_connection={}, + effective_vars={}, + effective_target_schema="main", + ) + + plan: JanitorPlan = build_janitor_plan( + project=project, + adapter=adapter, + connection=connection, + retention_days=30, + direct_settings=JanitorDirectModeSettings(enabled=True), + ) + + assert tuple(candidate.key.name for candidate in plan.query_diff_artifact_candidates) == ( + normalized_owned_name, + ) + assert any( + skipped.key.name == untracked.name + and "lacks matching ownership evidence" in skipped.reason + for skipped in plan.skipped_relations + ) + + result: JanitorExecutionResult = execute_janitor_plan( + plan=plan, + adapter=adapter, + connection=connection, + ) + + assert tuple(candidate.key.name for candidate in result.deleted_query_diff_artifacts) == ( + normalized_owned_name, + ) + assert not adapter.relation_exists( + connection=connection, + database=None, + schema="main", + name=normalized_owned_name, + ) + assert adapter.relation_exists( + connection=connection, + database=None, + schema="main", + name=untracked.name, + ) + fingerprint_row: tuple[int] | None = connection.execute( + "SELECT COUNT(*) FROM main._sqlbuild_fingerprints WHERE node_type = ?", + [NODE_TYPE_QUERY_DIFF_ARTIFACT], + ).fetchone() + assert fingerprint_row == (test_case.expected_fingerprint_count,) + finally: + connection.close() diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/_test_types.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/_test_types.py index 0d21d1d8e5..619c23f7d2 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/_test_types.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/_test_types.py @@ -25,3 +25,13 @@ class VersionFlagTestCase: description: str argv: tuple[str, ...] expected_exit_code: int + + +@dataclass(frozen=True) +class QueryDiffParsingTestCase: + description: str + argv: tuple[str, ...] + expected_left_query: str | None + expected_right_query: str | None + expected_keys: tuple[str, ...] + expected_unkeyed: bool diff --git a/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/test_parsing.py b/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/test_parsing.py index c59a3d3acf..c459f51aa6 100644 --- a/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/test_parsing.py +++ b/tests/unit/src/sqlbuild/cli/commands/_helpers/entry/test_parsing.py @@ -11,11 +11,67 @@ from sqlbuild.cli.commands.models import ParsedCliInvocation from tests.unit.src.sqlbuild.cli.commands._helpers.entry._test_types import ( AuditConcurrencyParsingTestCase, + QueryDiffParsingTestCase, VerboseCommandTestCase, VersionFlagTestCase, ) +@pytest.mark.parametrize( + "test_case", + [ + QueryDiffParsingTestCase( + description="inline composite keyed query diff", + argv=( + "diff", + "--left-query", + "SELECT order_id FROM old_orders", + "--right-query", + "SELECT order_id FROM new_orders", + "--key", + "account_id", + "--key", + "order_id", + ), + expected_left_query="SELECT order_id FROM old_orders", + expected_right_query="SELECT order_id FROM new_orders", + expected_keys=("account_id", "order_id"), + expected_unkeyed=False, + ), + QueryDiffParsingTestCase( + description="inline unkeyed query diff", + argv=( + "diff", + "--left-query", + "VALUES (1)", + "--right-query", + "VALUES (1)", + "--unkeyed", + ), + expected_left_query="VALUES (1)", + expected_right_query="VALUES (1)", + expected_keys=(), + expected_unkeyed=True, + ), + ], + ids=lambda case: case.description, +) +def test_given_raw_query_diff_arguments_when_parsing_then_inputs_remain_distinct( + test_case: QueryDiffParsingTestCase, +) -> None: + parsed: ParsedCliInvocation = parse_cli_invocation( + argv=test_case.argv, + parser=build_cli_parser(), + ) + + assert parsed.args is not None + assert parsed.args.target_range is None + assert parsed.args.left_query == test_case.expected_left_query + assert parsed.args.right_query == test_case.expected_right_query + assert tuple(parsed.args.key) == test_case.expected_keys + assert parsed.args.unkeyed is test_case.expected_unkeyed + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py index d66f7d41b9..5534915eac 100644 --- a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py +++ b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/_test_types.py @@ -50,6 +50,15 @@ class EventExporterDiscoveryTestCase: expected_name: str +@dataclass(frozen=True) +class DiffConfigTestCase: + """One project-level diff configuration case.""" + + description: str + expected_query_artifact_ttl: str + expected_error_fragment: str | None = None + + @dataclass(frozen=True) class CommandOutputSinkDiscoveryTestCase: description: str diff --git a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py index ad1a9bafd4..e61a2e0496 100644 --- a/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py +++ b/tests/unit/src/sqlbuild/compiler/discovery/_helpers/test_yml_project.py @@ -16,6 +16,7 @@ from tests.unit.src.sqlbuild.compiler.discovery._helpers._test_types import ( ColumnContractModeConfigErrorTestCase, ColumnContractModeConfigTestCase, + DiffConfigTestCase, EffectiveBatchDefaultTestCase, FutureCursorConfigErrorTestCase, FutureCursorConfigTestCase, @@ -39,6 +40,51 @@ ) +@pytest.mark.parametrize( + "test_case", + [DiffConfigTestCase(description="positive fixed TTL", expected_query_artifact_ttl="6h")], + ids=lambda case: case.description, +) +def test_given_query_artifact_ttl_when_loading_project_then_diff_config_is_typed( + test_case: DiffConfigTestCase, + tmp_path: Path, +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text( + 'name = "demo"\nadapter = "duckdb"\n[diff]\n' + f'query_artifact_ttl = "{test_case.expected_query_artifact_ttl}"\n', + encoding="utf-8", + ) + + config: ProjectConfig = load_project_config(project_dir=tmp_path) + + assert config.diff.query_artifact_ttl == test_case.expected_query_artifact_ttl + + +@pytest.mark.parametrize( + "test_case", + [ + DiffConfigTestCase( + description="zero TTL", + expected_query_artifact_ttl="0h", + expected_error_fragment="query_artifact_ttl must be a positive", + ) + ], + ids=lambda case: case.description, +) +def test_given_invalid_query_artifact_ttl_when_loading_project_then_error_is_actionable( + test_case: DiffConfigTestCase, + tmp_path: Path, +) -> None: + (tmp_path / "sqlbuild_project.toml").write_text( + 'name = "demo"\nadapter = "duckdb"\n[diff]\n' + f'query_artifact_ttl = "{test_case.expected_query_artifact_ttl}"\n', + encoding="utf-8", + ) + + with pytest.raises(ProjectConfigError, match=test_case.expected_error_fragment or ""): + _ = load_project_config(project_dir=tmp_path) + + @pytest.mark.parametrize( "test_case", [ diff --git a/tests/unit/src/sqlbuild/executor/diff/_test_types.py b/tests/unit/src/sqlbuild/executor/diff/_test_types.py index e355df25e3..7b380c92d7 100644 --- a/tests/unit/src/sqlbuild/executor/diff/_test_types.py +++ b/tests/unit/src/sqlbuild/executor/diff/_test_types.py @@ -11,6 +11,15 @@ class ParseRowDiffTolerancesTestCase: expected_result: RowDiffTolerances +@dataclass(frozen=True) +class CliToleranceTestCase: + """One CLI tolerance parsing case.""" + + description: str + values: tuple[str, ...] + expected_result: RowDiffTolerances + + @dataclass(frozen=True) class ParseRowDiffTolerancesErrorTestCase: description: str diff --git a/tests/unit/src/sqlbuild/executor/diff/test_config.py b/tests/unit/src/sqlbuild/executor/diff/test_config.py index 194514d5fd..2880938ca0 100644 --- a/tests/unit/src/sqlbuild/executor/diff/test_config.py +++ b/tests/unit/src/sqlbuild/executor/diff/test_config.py @@ -7,11 +7,13 @@ from sqlbuild.adapter.contract.models import RowDiffSampling, RowDiffTolerance, RowDiffTolerances from sqlbuild.errors.contracts.exceptions import ExecutorInputError from sqlbuild.executor.diff._helpers.config import ( + parse_cli_tolerance_overrides, parse_row_diff_tolerances, resolve_row_diff_sampling, ) from sqlbuild.executor.diff.models import RowDiffSamplingOverride from tests.unit.src.sqlbuild.executor.diff._test_types import ( + CliToleranceTestCase, ParseRowDiffTolerancesErrorTestCase, ParseRowDiffTolerancesTestCase, ResolveRowDiffSamplingErrorTestCase, @@ -19,6 +21,32 @@ ) +@pytest.mark.parametrize( + "test_case", + [ + CliToleranceTestCase( + description="absolute and relative rules for one column", + values=("amount:absolute=0.01", "amount:relative=0.001"), + expected_result=RowDiffTolerances( + by_column={ + "amount": RowDiffTolerance( + absolute=Decimal("0.01"), + relative=Decimal("0.001"), + ) + } + ), + ) + ], + ids=lambda case: case.description, +) +def test_given_cli_tolerance_specs_when_parsing_then_combines_rules_per_column( + test_case: CliToleranceTestCase, +) -> None: + result: RowDiffTolerances = parse_cli_tolerance_overrides(values=test_case.values) + + assert result == test_case.expected_result + + @pytest.mark.parametrize( "test_case", [