From c914fbef8e6362a51cca1801b340a0fcdf7261c0 Mon Sep 17 00:00:00 2001 From: OmerBaddour Date: Sat, 1 Aug 2026 11:25:35 -0400 Subject: [PATCH 1/2] feat: add strict mode -- test before publishing A table can only be tested once it has been materialized, so a normal run writes production first and finds out afterwards. `clair run --strict` closes that window. Per Trouve: - Build into `__clair_strict_`, a sibling in the same schema (incremental Trouves first zero-copy CLONE the target so the INSERT/MERGE has a base to apply to). - Run the Trouve's tests against that staging object. - On pass, promote: `ALTER TABLE ... SWAP WITH ...` for tables, which is metadata-only and does not scale with table size; `CREATE OR REPLACE VIEW` for views. A target that does not exist yet is RENAMEd into place instead. - On fail, drop the staging object. The target is untouched, the node is reported as FAILED, and its dependents are skipped. Promotion happens per node, immediately after its tests, so dependents always read their upstreams under the real names their SQL references. `clair compile --strict` writes the same plan to _clairtifacts/ for review. `--strict` is rejected alongside `--no-test`, since it gates on tests. Co-Authored-By: Claude Opus 5 --- site-docs/docs/cli/compile.md | 7 +- site-docs/docs/cli/run.md | 11 +- site-docs/docs/guides/data-quality-tests.md | 8 + site-docs/docs/guides/index.md | 1 + site-docs/docs/guides/strict-mode.md | 69 +++ site-docs/mkdocs.yml | 1 + src/clair/cli/main.py | 39 +- src/clair/core/compiler.py | 51 +- src/clair/core/runner.py | 256 ++++++--- src/clair/core/strict.py | 139 +++++ src/clair/core/test_runner.py | 15 +- src/clair/trouves/trouve.py | 20 +- tests/unit/test_strict.py | 605 ++++++++++++++++++++ 13 files changed, 1139 insertions(+), 83 deletions(-) create mode 100644 site-docs/docs/guides/strict-mode.md create mode 100644 src/clair/core/strict.py create mode 100644 tests/unit/test_strict.py diff --git a/site-docs/docs/cli/compile.md b/site-docs/docs/cli/compile.md index 7bfe628..49f6e62 100644 --- a/site-docs/docs/cli/compile.md +++ b/site-docs/docs/cli/compile.md @@ -3,7 +3,7 @@ Resolve the DAG and write generated SQL to `_clairtifacts/`. No Snowflake connection is made. ```bash -clair compile [--project PATH] [--env NAME] [--select PATTERN]... [--run-mode MODE] +clair compile [--project PATH] [--env NAME] [--select PATTERN]... [--run-mode MODE] [--strict] ``` ## Example @@ -17,6 +17,9 @@ clair compile --project . --env prod # Compile only the orders schema clair compile --project . --select='refined.orders.*' + +# Show the strict-mode plan: staging build, test checkpoint, swap +clair compile --project . --strict ``` ## What it does @@ -53,9 +56,11 @@ _clairtifacts/ | `--env` | optional | Environment name. Required if you want routing applied to generated SQL. | | `--select` | all | Glob pattern to filter Trouves. Repeat to union patterns. | | `--run-mode` | `full_refresh` | `full_refresh` or `incremental`. Affects which SQL variant is generated. | +| `--strict` | `false` | Emit the [strict-mode](../guides/strict-mode.md) plan instead of a direct write to the target. | ## See also - [DAG](../concepts/dag.md) - [Selectors](../guides/selectors.md) +- [Strict Mode](../guides/strict-mode.md) - [clair clean](clean.md) diff --git a/site-docs/docs/cli/run.md b/site-docs/docs/cli/run.md index c3dbf2f..ab36770 100644 --- a/site-docs/docs/cli/run.md +++ b/site-docs/docs/cli/run.md @@ -3,7 +3,7 @@ Execute Trouves against Snowflake in topological dependency order, then run data quality tests. ```bash -clair run [--project PATH] [--env NAME] [--select PATTERN]... [--run-mode MODE] [--no-test] [--sample] +clair run [--project PATH] [--env NAME] [--select PATTERN]... [--run-mode MODE] [--no-test] [--sample] [--strict] ``` ## Example @@ -20,6 +20,9 @@ clair run --project . --env prod --run-mode full_refresh # Skip tests clair run --project . --env dev --no-test + +# Only publish a Trouve once its tests pass +clair run --project . --env prod --strict ``` ## Execution order @@ -32,6 +35,8 @@ SOURCE Trouves pass through (no SQL is executed against them). After each successful TABLE or VIEW, attached tests run automatically. If any test fails, the run exits with a non-zero status code. Use `--no-test` to skip tests. +By default the target is written before its tests run, so a failing test means bad data is already in production. `--strict` builds each Trouve into a run-scoped staging object, tests it there, and swaps it into the real name only if every test passes — see [Strict Mode](../guides/strict-mode.md). It cannot be combined with `--no-test`. + ## Flags | Flag | Default | Description | @@ -42,14 +47,16 @@ After each successful TABLE or VIEW, attached tests run automatically. If any te | `--run-mode` | `full_refresh` | `full_refresh` or `incremental`. Overrides each Trouve's `run_config`. | | `--no-test` | `false` | Skip data quality tests | | `--sample` | `false` | Run tests against `SELECT TOP 1000 *` (skips `TestRowCount`) | +| `--strict` | `false` | Build into a staging object, test, then swap into place. Incompatible with `--no-test`. | ## Exit codes - `0` — all Trouves succeeded and all tests passed -- `1` — one or more Trouves failed, or one or more tests failed +- `1` — one or more Trouves failed, or one or more tests failed (under `--strict`, a Trouve whose tests failed is itself reported as failed and its target is left unchanged) ## See also - [Selectors](../guides/selectors.md) - [Incrementality](../guides/incrementality.md) - [Data Quality Tests](../guides/data-quality-tests.md) +- [Strict Mode](../guides/strict-mode.md) diff --git a/site-docs/docs/guides/data-quality-tests.md b/site-docs/docs/guides/data-quality-tests.md index dd675a2..933dfce 100644 --- a/site-docs/docs/guides/data-quality-tests.md +++ b/site-docs/docs/guides/data-quality-tests.md @@ -140,3 +140,11 @@ clair test --project . --env dev --sample | `TestUniqueColumns` | `columns: list[str]` (min 2) | No | See also: [Tests API reference](../reference/tests-api.md). + +## Testing before publishing + +Tests run after a Trouve has been materialized — that is the only point at which there is a table to query. By default that means a failing test tells you production is already wrong. [Strict Mode](strict-mode.md) inverts this: each Trouve is built into a staging object, tested there, and swapped into its real name only if every test passes. + +```bash +clair run --project . --env prod --strict +``` diff --git a/site-docs/docs/guides/index.md b/site-docs/docs/guides/index.md index 459df0e..15c5f75 100644 --- a/site-docs/docs/guides/index.md +++ b/site-docs/docs/guides/index.md @@ -5,6 +5,7 @@ How-to guides for common clair tasks. - **[Pandas-Native Transformations](pandas-native.md)** — write pipeline steps as Python functions using pandas - **[Incrementality](incrementality.md)** — APPEND and UPSERT strategies for large tables - **[Data Quality Tests](data-quality-tests.md)** — attach tests to Trouves +- **[Strict Mode](strict-mode.md)** — publish a Trouve only after its tests pass - **[Selectors](selectors.md)** — run only a subset of your project - **[Routing Policies](routing.md)** — remap Snowflake targets per environment - **[Per-Database & Schema Config](per-database-schema-config.md)** — warehouse and role overrides per directory diff --git a/site-docs/docs/guides/strict-mode.md b/site-docs/docs/guides/strict-mode.md new file mode 100644 index 0000000..d3b0653 --- /dev/null +++ b/site-docs/docs/guides/strict-mode.md @@ -0,0 +1,69 @@ +# Strict Mode + +A table can only be tested once it has been materialized. In a normal run that means bad data lands in production first and the tests tell you about it afterwards — the table is already wrong, and anything reading it has already read the wrong numbers. + +Strict mode closes that window. Every Trouve is built into a run-scoped staging object, tested there, and only swapped into its real name once every test passes. + +```bash +clair run --project . --env prod --strict +``` + +## What happens per Trouve + +1. **Build into staging.** The Trouve is materialized as `
__clair_strict_`, a sibling object in the same database and schema. +2. **Test the staging object.** The Trouve's data quality tests run against the staging object, not the target. +3. **Promote on pass.** For a `TABLE`, clair issues `ALTER TABLE ... SWAP WITH ...` — a metadata-only operation whose cost does not scale with table size — then drops the superseded copy. For a `VIEW`, the target is recreated with `CREATE OR REPLACE VIEW`. +4. **Discard on fail.** The staging object is dropped and the target is left exactly as it was. The Trouve is reported as a failure and everything downstream of it is skipped. + +Promotion happens immediately after each node's tests, before the next node starts. Dependents therefore always read their upstreams under the real names their SQL references — you never write `__clair_strict_` anywhere yourself. + +## Incremental Trouves + +An incremental Trouve applies changes on top of state that already exists, so the staging object needs that state before the `INSERT` or `MERGE` can run. clair seeds it with a zero-copy clone: + +```sql +-- strict: clone target into staging so incremental statements have a base +CREATE OR REPLACE TABLE db.schema.orders__clair_strict_ CLONE db.schema.orders + +INSERT INTO db.schema.orders__clair_strict_ +SELECT * FROM ( ... ) + +-- tests run here + +ALTER TABLE db.schema.orders__clair_strict_ SWAP WITH db.schema.orders +DROP TABLE IF EXISTS db.schema.orders__clair_strict_ +``` + +Snowflake clones are metadata-only, so seeding the staging table is constant-time no matter how large the target is. If the target does not exist yet, clair falls back to a full refresh for that Trouve and skips the clone; the staging object is then renamed into place rather than swapped. + +## Seeing the plan + +`clair compile --strict` writes the full plan to `_clairtifacts/`, including the clone, the staging build, a comment marking where tests run, and the promotion: + +```bash +clair compile --project . --strict +clair compile --project . --strict --run-mode incremental +``` + +## Failure modes + +| What failed | Target | Staging object | +|-------------|--------|----------------| +| The build itself | Untouched | Dropped | +| A data quality test | Untouched | Dropped | +| The swap or rename | Untouched | **Retained** — it holds tested data and is the only record of the run | + +In all three cases the Trouve is reported as `FAILED`, its dependents are skipped, and `clair run` exits `1`. + +## Costs and caveats + +- **Storage.** Each Trouve briefly holds two copies. For a full refresh that is a real second copy for the duration of the node; for an incremental build the clone shares micro-partitions with the target and only diverges as rows change. +- **Tests are required.** `--strict` cannot be combined with `--no-test` — the whole point is to gate promotion on tests. A Trouve with no tests attached still goes through staging and is promoted; strict mode does not manufacture coverage you have not written. +- **Name length.** The suffix adds 47 characters to the table component. Snowflake caps identifiers at 255, so a Trouve whose name is within 47 characters of the cap fails at the naming step with a clear error rather than mid-run. +- **`--sample`.** Sampling applies to the staging object, so `--strict --sample` gates promotion on a `TOP 1000` check rather than the full table. + +## See also + +- [Data Quality Tests](data-quality-tests.md) +- [Incrementality](incrementality.md) +- [clair run](../cli/run.md) diff --git a/site-docs/mkdocs.yml b/site-docs/mkdocs.yml index 2c17bb8..acd7f5f 100644 --- a/site-docs/mkdocs.yml +++ b/site-docs/mkdocs.yml @@ -78,6 +78,7 @@ nav: - Pandas-Native Transformations: guides/pandas-native.md - Incrementality: guides/incrementality.md - Data Quality Tests: guides/data-quality-tests.md + - Strict Mode: guides/strict-mode.md - Selectors: guides/selectors.md - Routing Policies: guides/routing.md - Per-Database & Schema Config: guides/per-database-schema-config.md diff --git a/src/clair/cli/main.py b/src/clair/cli/main.py index b536cf3..c9c5353 100644 --- a/src/clair/cli/main.py +++ b/src/clair/cli/main.py @@ -252,7 +252,13 @@ def _require(prompt_text: str, **kwargs) -> str: default="full_refresh", help="Run mode: full_refresh recreates all tables; incremental applies only new data.", ) -def compile_cmd(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: str | None, run_mode: str) -> None: +@click.option( + "--strict", + is_flag=True, + default=False, + help="Show the strict-mode plan: build into a run-scoped staging object, test, then swap.", +) +def compile_cmd(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: str | None, run_mode: str, strict: bool) -> None: """Compile the project and show generated SQL (no Snowflake connection).""" project_root = Path(project).resolve() run_mode_enum = RunMode(run_mode) @@ -301,7 +307,7 @@ def _on_node_compiled(node_info: CompiledNodeInfo) -> None: artifact_file = artifacts_dir / "/".join(parts[:-1]) / f"{parts[-1]}{extension}" logger.info("compile.node", trouve=node_info.name, dependencies=node_info.dependencies, artifact_file=str(artifact_file)) - write_compile_output(dag, selected, project_root, on_node_compiled=_on_node_compiled, run_mode=run_mode_enum, run_id=run_id) + write_compile_output(dag, selected, project_root, on_node_compiled=_on_node_compiled, run_mode=run_mode_enum, run_id=run_id, strict=strict) logger.info("compile.complete", run_id=run_id, artifacts_dir=str(artifacts_dir)) except ClairError as e: @@ -429,12 +435,26 @@ def docs(project: str, port: int, host: str, no_browser: bool) -> None: default=False, help="Run post-run tests against a sample of each Trouve (skips row count tests).", ) -def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: str | None, run_mode: str, no_test: bool, sample: bool) -> None: +@click.option( + "--strict", + is_flag=True, + default=False, + help=( + "Build each Trouve into a run-scoped staging object, test it, and only " + "promote it into its real name if every test passes. Tables are promoted " + "with a constant-time SWAP; a failing Trouve leaves its target untouched." + ), +) +def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: str | None, run_mode: str, no_test: bool, sample: bool, strict: bool) -> None: """Run Trouves against Snowflake, then run data quality tests.""" project_root = Path(project).resolve() run_mode_enum = RunMode(run_mode) run_id = uuid6.uuid7().hex + if strict and no_test: + logger.error("run.error", error="--strict cannot be combined with --no-test: strict mode promotes a Trouve only after its tests pass") + sys.exit(1) + try: # Load environment env_name, environment = load_environment(env) @@ -460,7 +480,7 @@ def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: st return recompile_for_selection(discovered, set(selected)) - write_compile_output(dag, selected, project_root, run_mode=run_mode_enum, run_id=run_id) + write_compile_output(dag, selected, project_root, run_mode=run_mode_enum, run_id=run_id, strict=strict) # Warn if account_locator is missing (query URLs will be incomplete) if not environment.account_locator: @@ -472,8 +492,12 @@ def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: st test_failures: list[str] = [] - def on_node_success(node_name: str) -> bool: - node_test_results = run_tests(dag, [node_name], adapter, use_sample=sample) + def on_node_success(node_name: str, physical_name: str) -> bool: + node_test_results = run_tests( + dag, [node_name], adapter, + use_sample=sample, + physical_names={node_name: physical_name}, + ) passed = all(r.passed for r in node_test_results) if not passed: test_failures.append(node_name) @@ -481,13 +505,14 @@ def on_node_success(node_name: str) -> bool: try: total = len(selected) - logger.info("run.start", run_id=run_id, env=env_name, project=str(project_root), trouves=total, run_mode=run_mode) + logger.info("run.start", run_id=run_id, env=env_name, project=str(project_root), trouves=total, run_mode=run_mode, strict=strict) results = list(run_project( dag, selected, adapter, run_mode=run_mode_enum, run_id=run_id, after_node_success=on_node_success if not no_test else None, + strict=strict, )) counts = Counter(r.status for r in results) diff --git a/src/clair/core/compiler.py b/src/clair/core/compiler.py index 08fbd2f..9516b05 100644 --- a/src/clair/core/compiler.py +++ b/src/clair/core/compiler.py @@ -13,6 +13,11 @@ from clair.core.dag import ClairDag from clair.core.discovery import ARTIFACTS_DIR_NAME from clair.core.runner import resolve_effective_mode +from clair.core.strict import ( + build_clone_statement, + build_promote_statements, + strict_staging_name, +) from clair.exceptions import CompileError from clair.trouves.run_config import RunMode from clair.trouves.trouve import ExecutionType, Trouve, TrouveType @@ -86,6 +91,46 @@ def render(self) -> str: return "\n".join(parts) +def build_statements( + trouve: Trouve, + run_mode: RunMode, + run_id: str, + strict: bool = False, +) -> list[str]: + """Build the SQL statements for one Trouve, as they would be executed. + + Under strict mode the plan is shown end to end: the optional clone, the build + into the staging object, and the promotion that follows a passing test run. + The runner decides between SWAP and RENAME by checking whether the target + exists; compile has no connection, so it shows the SWAP form. + """ + effective_mode = resolve_effective_mode(trouve, run_mode) + + if not strict: + return trouve.build_sql(effective_mode, run_id=run_id) + + target_name = trouve.full_name + staging_name = strict_staging_name(target_name, run_id) + + statements: list[str] = [] + if effective_mode == RunMode.INCREMENTAL: + statements.append(build_clone_statement(target_name, staging_name)) + statements.extend(trouve.build_sql(effective_mode, run_id=run_id, target_name=staging_name)) + + assert trouve.compiled is not None + statements.append("-- strict: tests run against the staging object here") + statements.extend( + build_promote_statements( + trouve.type, + staging_name=staging_name, + target_name=target_name, + target_exists=True, + resolved_sql=trouve.compiled.resolved_sql, + ) + ) + return statements + + def write_compile_output( dag: ClairDag, selected: list[str], @@ -93,6 +138,7 @@ def write_compile_output( on_node_compiled: Callable[[CompiledNodeInfo], None] = lambda _: None, run_mode: RunMode = RunMode.FULL_REFRESH, run_id: str = "", + strict: bool = False, ) -> CompileOutput: """Write compiled SQL to _clairtifacts// and return a structured output. @@ -104,6 +150,8 @@ def write_compile_output( to disk, allowing callers to stream output. run_mode: The run mode to use when generating SQL statements. run_id: UUIDv7 hex string identifying this compile run. + strict: When True, emit the full strict-mode plan (staging build, test + checkpoint, promotion) rather than a direct write to the target. Returns: A CompileOutput with structured data and a .render() method. @@ -173,8 +221,7 @@ def write_compile_output( artifact_path.parent.mkdir(parents=True, exist_ok=True) artifact_path.write_text(artifact_content) elif trouve.compiled.execution_type == ExecutionType.SNOWFLAKE: - effective_mode = resolve_effective_mode(trouve, run_mode) - statements = trouve.build_sql(effective_mode, run_id=run_id) + statements = build_statements(trouve, run_mode, run_id, strict=strict) node_info = CompiledNodeInfo( name=name, diff --git a/src/clair/core/runner.py b/src/clair/core/runner.py index 291a092..7dde7e4 100644 --- a/src/clair/core/runner.py +++ b/src/clair/core/runner.py @@ -15,6 +15,13 @@ from clair.adapters.base import WarehouseAdapter from clair.core.dag import ClairDag, get_executable_nodes +from clair.core.strict import ( + build_cleanup_statement, + build_clone_statement, + build_promote_statements, + strict_staging_name, +) +from clair.exceptions import ClairError, RunError from clair.trouves.run_config import RunMode from clair.trouves.trouve import Trouve, TrouveType @@ -175,9 +182,17 @@ def resolve_effective_mode(trouve: Trouve, cli_run_mode: RunMode) -> RunMode: def _run_df_fn_trouve( trouve: Trouve, adapter: WarehouseAdapter, + target_full_name: str | None = None, ) -> RunResult: """Execute a df_fn Trouve: fetch inputs, transform, write output. + Args: + trouve: The compiled df_fn Trouve to run. + adapter: Connected warehouse adapter. + target_full_name: Object to write into, overriding the Trouve's routed + name. Strict mode uses this to write into a staging table. The + returned RunResult always reports the routed name. + Returns a RunResult with SUCCESS or FAILURE status. """ start = time.monotonic() @@ -224,13 +239,14 @@ def _run_df_fn_trouve( # 4. Write the result to Snowflake full_name = trouve.full_name - name_parts = full_name.split(".") + write_target = target_full_name or full_name + name_parts = write_target.split(".") if len(name_parts) != 3: duration = time.monotonic() - start return RunResult( full_name=full_name, status=RunStatus.FAILURE, - error=f"Cannot parse full_name '{full_name}' into database.schema.table", + error=f"Cannot parse full_name '{write_target}' into database.schema.table", duration_seconds=duration, ) @@ -239,7 +255,7 @@ def _run_df_fn_trouve( try: query_result = adapter.write_dataframe( dataframe=result_dataframe, - full_name=full_name, + full_name=write_target, database_name=database_name, schema_name=schema_name, table_name=table_name, @@ -249,7 +265,7 @@ def _run_df_fn_trouve( return RunResult( full_name=full_name, status=RunStatus.FAILURE, - error=f"Failed to write DataFrame to {full_name}: {write_error}", + error=f"Failed to write DataFrame to {write_target}: {write_error}", duration_seconds=duration, ) @@ -272,13 +288,73 @@ def _run_df_fn_trouve( ) +def _promote_or_discard( + trouve: Trouve, + adapter: WarehouseAdapter, + staging_name: str, + target_name: str, + tests_passed: bool, +) -> tuple[list[str], list[str], str]: + """Finish a strict-mode node: swap the staging object in, or drop it. + + Returns: + (query_ids, query_urls, error). The error is empty when the staging object + was promoted successfully. + """ + query_ids: list[str] = [] + query_urls: list[str] = [] + + def _execute(statement: str) -> str: + query_result = adapter.execute(statement) + if query_result.query_id: + query_ids.append(query_result.query_id) + if query_result.query_url: + query_urls.append(query_result.query_url) + return "" if query_result.success else (query_result.error or "unknown error") + + if not tests_passed: + cleanup_error = _execute(build_cleanup_statement(trouve.type, staging_name)) + error = ( + f"strict mode: tests failed, {target_name} left unchanged " + f"(staging object {staging_name} discarded)" + ) + if cleanup_error: + error += f"; staging cleanup also failed: {cleanup_error}" + return query_ids, query_urls, error + + database_name, schema_name, table_name = target_name.split(".") + target_exists = adapter.table_exists(database_name, schema_name, table_name) + + assert trouve.compiled is not None + for statement in build_promote_statements( + trouve.type, + staging_name=staging_name, + target_name=target_name, + target_exists=target_exists, + resolved_sql=trouve.compiled.resolved_sql, + ): + promote_error = _execute(statement) + if promote_error: + # Leave the staging object in place: it holds tested data and is the + # only record of what this run produced. + return ( + query_ids, + query_urls, + f"strict mode: tests passed but promotion failed: {promote_error} " + f"(staging object {staging_name} retained)", + ) + + return query_ids, query_urls, "" + + def run_project( dag: ClairDag, selected: list[str], adapter: WarehouseAdapter, run_mode: RunMode = RunMode.FULL_REFRESH, run_id: str = "", - after_node_success: Callable[[str], bool] | None = None, + after_node_success: Callable[[str, str], bool] | None = None, + strict: bool = False, ) -> Iterator[RunResult]: """Execute selected Trouves in topological order, yielding each result as it completes. @@ -286,9 +362,19 @@ def run_project( then continues with unrelated branches. after_node_success: optional callback invoked after each successful node, before - the next node runs. Return False to treat the node as failed for downstream - dependency purposes (circuit breaker for eager testing). + the next node runs. Called with (node_name, physical_name) where + physical_name is the object that was actually written -- the staging object + under strict mode, the routed name otherwise. Return False to treat the node + as failed for downstream dependency purposes (circuit breaker for eager + testing). + strict: when True, materialize each node into a run-scoped staging object, let + after_node_success test it, and only then promote it into its real name. A + node whose tests fail leaves its target untouched and is reported as a + FAILURE. Requires after_node_success. """ + if strict and after_node_success is None: + raise RunError("strict mode requires tests; after_node_success must be provided") + all_executable = get_executable_nodes(dag) to_run = [name for name in all_executable if name in selected] @@ -323,91 +409,133 @@ def run_project( skip_reasons.setdefault(desc, name) continue + routed_name = trouve.compiled.full_name + if trouve.type != TrouveType.SOURCE: - assert trouve.compiled is not None - routed_parts = trouve.compiled.full_name.split(".") + routed_parts = routed_name.split(".") if len(routed_parts) >= 2: adapter.execute(f"CREATE DATABASE IF NOT EXISTS {routed_parts[0]}") adapter.execute(f"CREATE SCHEMA IF NOT EXISTS {routed_parts[0]}.{routed_parts[1]}") - # Branch: df_fn Trouves use fetch/transform/write instead of SQL execution - if trouve.df_fn is not None: - logger.info("run.node.start", trouve=name, effective_mode="full_refresh") - result = _run_df_fn_trouve(trouve, adapter) - yield result - - if result.status == RunStatus.SUCCESS: - logger.info("run.node.success", trouve=name, duration_seconds=round(result.duration_seconds, 3)) - if after_node_success is not None and not after_node_success(name): - for desc in nx.descendants(dag, name): - skip_reasons.setdefault(desc, name) - else: - logger.warning("run.node.failure", trouve=name, duration_seconds=round(result.duration_seconds, 3), error=result.error) + # Strict mode materializes into a run-scoped sibling object; the real name + # is only written once the tests against that object have passed. + staging_name: str | None = None + if strict: + try: + staging_name = strict_staging_name(routed_name, run_id) + except ClairError as naming_error: + logger.warning("run.node.failure", trouve=name, error=str(naming_error)) + yield RunResult( + full_name=name, + status=RunStatus.FAILURE, + error=str(naming_error), + ) for desc in nx.descendants(dag, name): skip_reasons.setdefault(desc, name) - continue - - effective_mode = resolve_effective_mode(trouve, run_mode) - # Incremental fallback: if target table doesn't exist yet, run full refresh - if effective_mode == RunMode.INCREMENTAL: - assert trouve.compiled is not None - routed_parts = trouve.compiled.full_name.split(".") - if len(routed_parts) == 3 and not adapter.table_exists(routed_parts[0], routed_parts[1], routed_parts[2]): - logger.info("run.node.incremental_fallback", trouve=name, reason="table_not_found") - effective_mode = RunMode.FULL_REFRESH - - logger.info("run.node.start", trouve=name, effective_mode=effective_mode.value) - statements = trouve.build_sql(effective_mode, run_id) - - if not statements: - continue + continue + write_target = staging_name or routed_name - start = time.monotonic() - last_result = None - all_succeeded = True - failed_at = None query_ids: list[str] = [] query_urls: list[str] = [] + statements: list[str] | None = None - for stmt_idx, stmt in enumerate(statements): - query_result = adapter.execute(stmt) - last_result = query_result - if query_result.query_id: - query_ids.append(query_result.query_id) - if query_result.query_url: - query_urls.append(query_result.query_url) - if not query_result.success: - all_succeeded = False - failed_at = stmt_idx - break + # Branch: df_fn Trouves use fetch/transform/write instead of SQL execution + if trouve.df_fn is not None: + logger.info("run.node.start", trouve=name, effective_mode="full_refresh", target=write_target) + df_result = _run_df_fn_trouve(trouve, adapter, target_full_name=write_target) + report_name = df_result.full_name + duration = df_result.duration_seconds + query_ids.extend(df_result.query_ids) + query_urls.extend(df_result.query_urls) + materialized = df_result.status == RunStatus.SUCCESS + error = df_result.error + else: + report_name = name - duration = time.monotonic() - start + effective_mode = resolve_effective_mode(trouve, run_mode) + # Incremental fallback: if target table doesn't exist yet, run full refresh + if effective_mode == RunMode.INCREMENTAL: + routed_parts = routed_name.split(".") + if len(routed_parts) == 3 and not adapter.table_exists(routed_parts[0], routed_parts[1], routed_parts[2]): + logger.info("run.node.incremental_fallback", trouve=name, reason="table_not_found") + effective_mode = RunMode.FULL_REFRESH - # UPSERT cleanup: if MERGE (stmt index 1) failed, still drop staging (stmt index 2) - if not all_succeeded and len(statements) == 3 and failed_at == 1: - adapter.execute(statements[2]) + logger.info("run.node.start", trouve=name, effective_mode=effective_mode.value, target=write_target) + statements = trouve.build_sql(effective_mode, run_id, target_name=write_target) - if all_succeeded: + if not statements: + continue + + # An incremental build needs its prior state; a zero-copy clone puts it + # in the staging table in constant time. + if staging_name is not None and effective_mode == RunMode.INCREMENTAL: + statements = [build_clone_statement(routed_name, staging_name)] + statements + + start = time.monotonic() + last_result = None + materialized = True + failed_at = None + + for stmt_idx, stmt in enumerate(statements): + query_result = adapter.execute(stmt) + last_result = query_result + if query_result.query_id: + query_ids.append(query_result.query_id) + if query_result.query_url: + query_urls.append(query_result.query_url) + if not query_result.success: + materialized = False + failed_at = stmt_idx + break + + duration = time.monotonic() - start + + # UPSERT cleanup: if the MERGE failed, still drop the merge staging table. + # UPSERT always emits (create staging, merge, drop staging) as the last + # three statements -- strict mode may prepend a clone before them. + if not materialized and len(statements) >= 3: + merge_index = len(statements) - 2 + if failed_at == merge_index: + adapter.execute(statements[merge_index + 1]) + + error = "" if materialized else ((last_result.error if last_result else "") or "") + + # Strict mode: test the staging object, then swap it in or throw it away. + if staging_name is not None and materialized: + assert after_node_success is not None + tests_passed = after_node_success(name, staging_name) + promote_ids, promote_urls, strict_error = _promote_or_discard( + trouve, adapter, staging_name, routed_name, tests_passed + ) + query_ids.extend(promote_ids) + query_urls.extend(promote_urls) + if strict_error: + materialized = False + error = strict_error + elif staging_name is not None and not materialized: + adapter.execute(build_cleanup_statement(trouve.type, staging_name)) + + if materialized: logger.info("run.node.success", trouve=name, duration_seconds=round(duration, 3), query_ids=query_ids) yield RunResult( - full_name=name, + full_name=report_name, status=RunStatus.SUCCESS, query_ids=query_ids, query_urls=query_urls, duration_seconds=duration, ) - if after_node_success is not None and not after_node_success(name): + # Non-strict: tests run after the target has already been written. + if not strict and after_node_success is not None and not after_node_success(name, routed_name): for desc in nx.descendants(dag, name): skip_reasons.setdefault(desc, name) else: - assert last_result is not None - logger.warning("run.node.failure", trouve=name, duration_seconds=round(duration, 3), error=last_result.error, query_ids=query_ids) + logger.warning("run.node.failure", trouve=name, duration_seconds=round(duration, 3), error=error, query_ids=query_ids) yield RunResult( - full_name=name, + full_name=report_name, status=RunStatus.FAILURE, query_ids=query_ids, query_urls=query_urls, - error=last_result.error or "", + error=error, sql=statements, duration_seconds=duration, ) diff --git a/src/clair/core/strict.py b/src/clair/core/strict.py new file mode 100644 index 0000000..43b7c07 --- /dev/null +++ b/src/clair/core/strict.py @@ -0,0 +1,139 @@ +"""Strict mode -- materialize into a run-scoped staging object, test, then promote. + +A table can only be tested once it has been materialized, so a plain run leaves a +window where the production object holds untested data. Strict mode closes that +window: + +1. Materialize the Trouve into ``
__clair_strict_``, a sibling object + in the same schema. For incremental Trouves the current target is first + zero-copy cloned into that name so the incremental statements have a base to + apply to. +2. Run the Trouve's data quality tests against the staging object. +3. On pass, promote the staging object into the real name. For tables this is + ``ALTER TABLE ... SWAP WITH ...``, a metadata-only operation whose cost does + not scale with table size. +4. On failure, drop the staging object. The production object is never touched. + +Downstream Trouves are unaffected: promotion happens immediately after each node's +tests, so by the time a dependent runs, its upstreams already resolve to the real +names their SQL references. +""" + +from __future__ import annotations + +from clair.exceptions import ClairError +from clair.trouves.trouve import TrouveType + + +STRICT_SUFFIX = "__clair_strict_" + +# Snowflake's maximum identifier length. +MAX_IDENTIFIER_LENGTH = 255 + + +class StrictNamingError(ClairError): + """Raised when the strict staging name would exceed Snowflake's identifier limit.""" + + +def strict_staging_name(full_name: str, run_id: str) -> str: + """Return the run-scoped staging name for a Trouve's routed full_name. + + The suffix is appended to the table component only, so the staging object + lives in the same database and schema as its target -- a requirement for + ``ALTER TABLE ... SWAP WITH ...``. + + Args: + full_name: Routed "database.schema.table" name of the target object. + run_id: UUIDv7 hex string identifying this clair run. + + Returns: + The staging "database.schema.table__clair_strict_" name. + + Raises: + StrictNamingError: If the staging table identifier exceeds 255 characters. + """ + parts = full_name.split(".") + if len(parts) != 3: + raise StrictNamingError( + f"Cannot derive a strict staging name from '{full_name}': " + "expected database.schema.table" + ) + + database_name, schema_name, table_name = parts + staging_table_name = f"{table_name}{STRICT_SUFFIX}{run_id}" + + if len(staging_table_name) > MAX_IDENTIFIER_LENGTH: + raise StrictNamingError( + f"Strict staging name '{staging_table_name}' is " + f"{len(staging_table_name)} chars (max {MAX_IDENTIFIER_LENGTH}). " + f"Shorten the name of '{full_name}' or run without --strict." + ) + + return f"{database_name}.{schema_name}.{staging_table_name}" + + +def build_clone_statement(target_name: str, staging_name: str) -> str: + """Return the zero-copy CLONE that seeds an incremental build's staging table. + + Snowflake clones are metadata-only, so this is constant-time regardless of + how large the target table is. + """ + return ( + f"-- strict: clone target into staging so incremental statements have a base\n" + f"CREATE OR REPLACE TABLE {staging_name} CLONE {target_name}" + ) + + +def build_promote_statements( + trouve_type: TrouveType, + staging_name: str, + target_name: str, + target_exists: bool, + resolved_sql: str = "", +) -> list[str]: + """Return the statements that promote a tested staging object into its real name. + + Args: + trouve_type: TABLE or VIEW. SOURCE Trouves are never materialized. + staging_name: Routed name of the staging object holding tested data. + target_name: Routed name the object should end up under. + target_exists: Whether the target already exists in the warehouse. + resolved_sql: The Trouve's resolved SQL; required for VIEW promotion. + + Returns: + Ordered list of SQL statements to execute. + """ + if trouve_type == TrouveType.VIEW: + # Views have no SWAP. CREATE OR REPLACE VIEW is itself atomic and, being + # metadata-only, costs nothing -- the staging view proved the SQL is valid + # and its results pass the tests. + return [ + f"-- strict: promote tested view\n" + f"CREATE OR REPLACE VIEW {target_name} AS (\n{resolved_sql.strip()}\n)", + f"-- strict: drop staging view\n" + f"DROP VIEW IF EXISTS {staging_name}", + ] + + if not target_exists: + return [ + f"-- strict: promote tested table (target did not exist)\n" + f"ALTER TABLE {staging_name} RENAME TO {target_name}" + ] + + return [ + # SWAP is a metadata-only operation: O(1) in the size of either table. + f"-- strict: swap tested staging table into place\n" + f"ALTER TABLE {staging_name} SWAP WITH {target_name}", + # After the swap, the staging name holds the previous target contents. + f"-- strict: drop the superseded table\n" + f"DROP TABLE IF EXISTS {staging_name}", + ] + + +def build_cleanup_statement(trouve_type: TrouveType, staging_name: str) -> str: + """Return the statement that discards a staging object after a failed build or test.""" + object_type = "VIEW" if trouve_type == TrouveType.VIEW else "TABLE" + return ( + f"-- strict: discard untested staging object\n" + f"DROP {object_type} IF EXISTS {staging_name}" + ) diff --git a/src/clair/core/test_runner.py b/src/clair/core/test_runner.py index 01e3a65..b2c354c 100644 --- a/src/clair/core/test_runner.py +++ b/src/clair/core/test_runner.py @@ -122,6 +122,7 @@ def run_tests( selected: list[str], adapter: WarehouseAdapter, use_sample: bool = False, + physical_names: dict[str, str] | None = None, ) -> list[TestResult]: """Execute data quality tests for selected Trouves. @@ -137,11 +138,16 @@ def run_tests( use_sample: When True, enable per-Trouve native sampling via ``trouve.sample()`` and skip tests not meaningful on sampled data (e.g. ``TestRowCount``). + physical_names: Optional mapping of node name -> object to query instead + of the Trouve's routed name. Strict mode uses this to test a + staging table before it is promoted. Results still report the + routed name so output is stable across modes. Returns: List of TestResult, one per test executed. """ results: list[TestResult] = [] + physical_names = physical_names or {} for name in selected: trouve = dag.get_trouve(name) @@ -156,6 +162,7 @@ def run_tests( assert trouve.compiled is not None routed_name = trouve.compiled.full_name + queried_name = physical_names.get(name, routed_name) # Skip tests that are meaningless on sampled data. if use_sample and not test.is_run_with_sample: @@ -168,11 +175,13 @@ def run_tests( continue try: - sql = test.to_sql(routed_name) + sql = test.to_sql(queried_name) if use_sample: - sample_subquery = trouve.sample() - pattern = re.compile(re.escape(f"FROM {routed_name}"), re.IGNORECASE) + # sample() is written against the Trouve's own full_name; point + # it at the object actually under test. + sample_subquery = trouve.sample().replace(routed_name, queried_name) + pattern = re.compile(re.escape(f"FROM {queried_name}"), re.IGNORECASE) sql = pattern.sub(f"FROM {sample_subquery}", sql) query_result = adapter.execute(sql) diff --git a/src/clair/trouves/trouve.py b/src/clair/trouves/trouve.py index b4f1915..30b7962 100644 --- a/src/clair/trouves/trouve.py +++ b/src/clair/trouves/trouve.py @@ -131,12 +131,21 @@ def sample(self) -> str: assert self.compiled is not None, "sample() requires a compiled Trouve" return f"(SELECT TOP 1000 * FROM {self.compiled.full_name})" - def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]: + def build_sql( + self, + effective_mode: RunMode, + run_id: str, + target_name: str | None = None, + ) -> list[str]: """Generate the SQL statements to materialize this Trouve. Args: effective_mode: The resolved run mode (caller determines this). run_id: Unique identifier for this clair run invocation. + target_name: Object to write into, overriding ``full_name``. Used by + strict mode to build into a run-scoped staging table. References + to *upstream* Trouves inside the SQL are unaffected -- they always + resolve to their real names. Returns: Ordered list of SQL statements to execute. Empty for SOURCE Trouves. @@ -153,16 +162,17 @@ def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]: return [] resolved_sql = self.compiled.resolved_sql.strip() + write_target = target_name or self.full_name if effective_mode == RunMode.FULL_REFRESH: object_type = "TABLE" if self.type == TrouveType.TABLE else "VIEW" return [ - f"CREATE OR REPLACE {object_type} {self.full_name} AS (\n{resolved_sql}\n)" + f"CREATE OR REPLACE {object_type} {write_target} AS (\n{resolved_sql}\n)" ] if self.run_config.incremental_mode == IncrementalMode.APPEND: return [ - f"INSERT INTO {self.full_name}\nSELECT * FROM (\n{resolved_sql}\n)" + f"INSERT INTO {write_target}\nSELECT * FROM (\n{resolved_sql}\n)" ] # UPSERT @@ -171,6 +181,8 @@ def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]: "upsert mode requires columns to be defined on the Trouve" ) + # Derived from full_name, not write_target, so that strict mode's staging + # suffix is not stacked on top of this one. staging_name = f"{self.full_name}__clair_staging_{run_id}" all_col_names = [c.name for c in self.columns] unique_keys = set(self.run_config.primary_key_columns or []) @@ -198,7 +210,7 @@ def build_sql(self, effective_mode: RunMode, run_id: str) -> list[str]: ) stmt_2 = ( f"-- [2/3] merge into target\n" - f"MERGE INTO {self.full_name} AS {TARGET}\n" + f"MERGE INTO {write_target} AS {TARGET}\n" f"USING {staging_name} AS {SOURCE}\n" f"ON {join_condition}\n" f"WHEN MATCHED THEN UPDATE SET {update_clause}\n" diff --git a/tests/unit/test_strict.py b/tests/unit/test_strict.py new file mode 100644 index 0000000..0a11611 --- /dev/null +++ b/tests/unit/test_strict.py @@ -0,0 +1,605 @@ +"""Tests for strict mode -- build into staging, test, then swap.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock + +import pandas as pd +import pytest + +from clair.adapters.base import QueryResult, WarehouseAdapter +from clair.core.compiler import build_statements +from clair.core.dag import build_dag, get_executable_nodes +from clair.core.runner import RunStatus, run_project +from clair.core.strict import ( + MAX_IDENTIFIER_LENGTH, + STRICT_SUFFIX, + StrictNamingError, + build_cleanup_statement, + build_clone_statement, + build_promote_statements, + strict_staging_name, +) +from clair.exceptions import RunError +from clair.trouves.config import ResolvedConfig +from clair.trouves.run_config import IncrementalMode, RunConfig, RunMode +from clair.trouves.trouve import CompiledAttributes, ExecutionType, Trouve, TrouveType + + +RUN_ID = "0195aabbccddeeff0011223344556677" + + +def _compile( + trouve: Trouve, + full_name: str, + imports: list[str] | None = None, + execution_type: ExecutionType = ExecutionType.SNOWFLAKE, +) -> Trouve: + trouve.compiled = CompiledAttributes( + full_name=full_name, + logical_name=full_name, + resolved_sql=trouve.sql, + file_path=Path(f"/fake/{full_name.replace('.', '/')}.py"), + module_name=full_name, + imports=imports or [], + config=ResolvedConfig(), + execution_type=execution_type, + ) + return trouve + + +def _make_adapter( + fail_on: str | None = None, + target_exists: bool = True, +) -> tuple[WarehouseAdapter, list[str]]: + """Return a mock adapter plus the list of SQL it was asked to execute.""" + executed: list[str] = [] + adapter = MagicMock(spec=WarehouseAdapter) + counter = 0 + + def mock_execute(sql: str) -> QueryResult: + nonlocal counter + counter += 1 + executed.append(sql) + query_id = f"qid-{counter:04d}" + success = fail_on is None or fail_on not in sql + return QueryResult( + query_id=query_id, + query_url=f"https://test/#/query/{query_id}", + success=success, + error=None if success else f"Simulated failure for {fail_on}", + ) + + adapter.execute.side_effect = mock_execute + adapter.table_exists.return_value = target_exists + return adapter, executed + + +class TestStrictStagingName: + def test_suffix_applied_to_table_component_only(self): + staging = strict_staging_name("db.schema.orders", RUN_ID) + assert staging == f"db.schema.orders{STRICT_SUFFIX}{RUN_ID}" + + def test_staging_shares_database_and_schema_with_target(self): + """SWAP WITH requires both objects to live in the same schema.""" + staging = strict_staging_name("analytics.revenue.daily", RUN_ID) + assert staging.split(".")[:2] == ["analytics", "revenue"] + + def test_run_id_makes_concurrent_runs_disjoint(self): + first = strict_staging_name("db.schema.orders", "aaaa") + second = strict_staging_name("db.schema.orders", "bbbb") + assert first != second + + def test_rejects_name_that_is_not_three_parts(self): + with pytest.raises(StrictNamingError, match="database.schema.table"): + strict_staging_name("db.schema", RUN_ID) + + def test_rejects_identifier_over_snowflake_limit(self): + long_table = "x" * MAX_IDENTIFIER_LENGTH + with pytest.raises(StrictNamingError, match="max 255"): + strict_staging_name(f"db.schema.{long_table}", RUN_ID) + + def test_accepts_identifier_at_the_limit(self): + budget = MAX_IDENTIFIER_LENGTH - len(STRICT_SUFFIX) - len(RUN_ID) + staging = strict_staging_name(f"db.schema.{'x' * budget}", RUN_ID) + assert len(staging.split(".")[2]) == MAX_IDENTIFIER_LENGTH + + +class TestPromoteStatements: + def test_existing_table_is_swapped_then_old_copy_dropped(self): + statements = build_promote_statements( + TrouveType.TABLE, + staging_name="db.s.t__staging", + target_name="db.s.t", + target_exists=True, + ) + assert "ALTER TABLE db.s.t__staging SWAP WITH db.s.t" in statements[0] + assert "DROP TABLE IF EXISTS db.s.t__staging" in statements[1] + + def test_missing_table_is_renamed_into_place(self): + statements = build_promote_statements( + TrouveType.TABLE, + staging_name="db.s.t__staging", + target_name="db.s.t", + target_exists=False, + ) + assert len(statements) == 1 + assert "ALTER TABLE db.s.t__staging RENAME TO db.s.t" in statements[0] + + def test_view_is_recreated_and_staging_dropped(self): + statements = build_promote_statements( + TrouveType.VIEW, + staging_name="db.s.v__staging", + target_name="db.s.v", + target_exists=True, + resolved_sql="SELECT 1 AS id", + ) + assert "CREATE OR REPLACE VIEW db.s.v" in statements[0] + assert "SELECT 1 AS id" in statements[0] + assert "DROP VIEW IF EXISTS db.s.v__staging" in statements[1] + + def test_cleanup_uses_matching_object_type(self): + assert "DROP TABLE IF EXISTS db.s.t" in build_cleanup_statement( + TrouveType.TABLE, "db.s.t" + ) + assert "DROP VIEW IF EXISTS db.s.v" in build_cleanup_statement( + TrouveType.VIEW, "db.s.v" + ) + + def test_clone_is_zero_copy(self): + statement = build_clone_statement("db.s.t", "db.s.t__staging") + assert "CREATE OR REPLACE TABLE db.s.t__staging CLONE db.s.t" in statement + + +class TestBuildSqlTargetOverride: + def test_full_refresh_writes_into_override(self): + trouve = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.orders") + statements = trouve.build_sql(RunMode.FULL_REFRESH, RUN_ID, target_name="db.s.staging") + assert "CREATE OR REPLACE TABLE db.s.staging" in statements[0] + assert "db.s.orders" not in statements[0] + + def test_append_inserts_into_override(self): + trouve = _compile( + Trouve( + sql="SELECT 1 AS id", + run_config=RunConfig( + run_mode=RunMode.INCREMENTAL, incremental_mode=IncrementalMode.APPEND + ), + ), + "db.s.orders", + ) + statements = trouve.build_sql(RunMode.INCREMENTAL, RUN_ID, target_name="db.s.staging") + assert statements[0].startswith("INSERT INTO db.s.staging") + + def test_upsert_merges_into_override_without_stacking_staging_suffixes(self): + from clair.trouves.column import Column, ColumnType + + trouve = _compile( + Trouve( + sql="SELECT 1 AS id, 2 AS amount", + columns=[ + Column(name="id", type=ColumnType.NUMBER), + Column(name="amount", type=ColumnType.NUMBER), + ], + run_config=RunConfig( + run_mode=RunMode.INCREMENTAL, + incremental_mode=IncrementalMode.UPSERT, + primary_key_columns=["id"], + ), + ), + "db.s.orders", + ) + statements = trouve.build_sql(RunMode.INCREMENTAL, RUN_ID, target_name="db.s.strict") + assert "MERGE INTO db.s.strict" in statements[1] + # The merge staging table derives from the real name, not the override, + # so the two suffixes never stack. + assert f"db.s.orders__clair_staging_{RUN_ID}" in statements[0] + + def test_omitting_override_preserves_previous_behaviour(self): + trouve = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.orders") + assert trouve.build_sql(RunMode.FULL_REFRESH, RUN_ID) == trouve.build_sql( + RunMode.FULL_REFRESH, RUN_ID, target_name=None + ) + + +def _single_table_dag( + trouve_type: TrouveType = TrouveType.TABLE, + run_config: RunConfig | None = None, +): + kwargs = {"sql": "SELECT 1 AS id", "type": trouve_type} + if run_config is not None: + kwargs["run_config"] = run_config + trouve = _compile(Trouve(**kwargs), "db.s.orders") + dag = build_dag([trouve]) + return dag, get_executable_nodes(dag) + + +class TestStrictRunner: + def test_build_targets_staging_and_promotes_after_passing_tests(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter() + tested: list[str] = [] + + def on_success(node_name: str, physical_name: str) -> bool: + tested.append(physical_name) + return True + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=on_success, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert results[0].status == RunStatus.SUCCESS + # Tests ran against the staging object, not the target. + assert tested == [staging] + assert any(f"CREATE OR REPLACE TABLE {staging}" in sql for sql in executed) + assert any(f"ALTER TABLE {staging} SWAP WITH db.s.orders" in sql for sql in executed) + + def test_target_is_never_written_before_tests_pass(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter() + sql_at_test_time: list[list[str]] = [] + + def on_success(node_name: str, physical_name: str) -> bool: + sql_at_test_time.append(list(executed)) + return True + + list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=on_success, strict=True, + )) + + assert not any( + "CREATE OR REPLACE TABLE db.s.orders AS" in sql for sql in sql_at_test_time[0] + ) + + def test_failing_tests_drop_staging_and_leave_target_untouched(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter() + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: False, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert results[0].status == RunStatus.FAILURE + assert "tests failed" in results[0].error + assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) + assert not any("SWAP WITH" in sql for sql in executed) + + def test_missing_target_is_renamed_rather_than_swapped(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter(target_exists=False) + + list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + assert any("RENAME TO db.s.orders" in sql for sql in executed) + assert not any("SWAP WITH" in sql for sql in executed) + + def test_failed_materialization_cleans_up_staging(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter(fail_on="CREATE OR REPLACE TABLE") + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert results[0].status == RunStatus.FAILURE + assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) + + def test_failed_promotion_retains_staging_and_reports_failure(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter(fail_on="SWAP WITH") + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + assert results[0].status == RunStatus.FAILURE + assert "promotion failed" in results[0].error + assert "retained" in results[0].error + + def test_incremental_clones_target_into_staging_first(self): + dag, selected = _single_table_dag( + run_config=RunConfig( + run_mode=RunMode.INCREMENTAL, incremental_mode=IncrementalMode.APPEND + ) + ) + adapter, executed = _make_adapter() + + list(run_project( + dag, selected, adapter, + run_mode=RunMode.INCREMENTAL, run_id=RUN_ID, + after_node_success=lambda _n, _p: True, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + clone_index = next(i for i, sql in enumerate(executed) if "CLONE db.s.orders" in sql) + insert_index = next(i for i, sql in enumerate(executed) if sql.startswith(f"INSERT INTO {staging}")) + assert clone_index < insert_index + + def test_incremental_fallback_to_full_refresh_skips_the_clone(self): + dag, selected = _single_table_dag( + run_config=RunConfig( + run_mode=RunMode.INCREMENTAL, incremental_mode=IncrementalMode.APPEND + ) + ) + adapter, executed = _make_adapter(target_exists=False) + + list(run_project( + dag, selected, adapter, + run_mode=RunMode.INCREMENTAL, run_id=RUN_ID, + after_node_success=lambda _n, _p: True, strict=True, + )) + + assert not any("CLONE" in sql for sql in executed) + + def test_view_is_created_in_staging_then_replaced_at_target(self): + dag, selected = _single_table_dag(trouve_type=TrouveType.VIEW) + adapter, executed = _make_adapter() + + list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert any(f"CREATE OR REPLACE VIEW {staging}" in sql for sql in executed) + assert any("CREATE OR REPLACE VIEW db.s.orders" in sql for sql in executed) + assert any(f"DROP VIEW IF EXISTS {staging}" in sql for sql in executed) + + def test_downstream_is_skipped_when_upstream_tests_fail(self): + upstream = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.upstream") + downstream = _compile( + Trouve(sql="SELECT * FROM db.s.upstream"), + "db.s.downstream", + imports=["db.s.upstream"], + ) + dag = build_dag([upstream, downstream]) + adapter, _ = _make_adapter() + + results = list(run_project( + dag, get_executable_nodes(dag), adapter, + run_id=RUN_ID, + after_node_success=lambda node_name, _p: node_name != "db.s.upstream", + strict=True, + )) + + by_name = {r.full_name: r for r in results} + assert by_name["db.s.upstream"].status == RunStatus.FAILURE + assert by_name["db.s.downstream"].status == RunStatus.SKIPPED + assert by_name["db.s.downstream"].skipped_by == "db.s.upstream" + + def test_downstream_reads_the_promoted_name_of_its_upstream(self): + """Promotion happens per node, so dependents never see a staging name.""" + upstream = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.upstream") + downstream = _compile( + Trouve(sql="SELECT * FROM db.s.upstream"), + "db.s.downstream", + imports=["db.s.upstream"], + ) + dag = build_dag([upstream, downstream]) + adapter, executed = _make_adapter() + + list(run_project( + dag, get_executable_nodes(dag), adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + downstream_staging = strict_staging_name("db.s.downstream", RUN_ID) + build_sql = next(sql for sql in executed if f"CREATE OR REPLACE TABLE {downstream_staging}" in sql) + assert "FROM db.s.upstream" in build_sql + assert STRICT_SUFFIX not in build_sql.split("FROM")[1] + + def test_strict_without_tests_is_rejected(self): + dag, selected = _single_table_dag() + adapter, _ = _make_adapter() + + with pytest.raises(RunError, match="strict mode requires tests"): + list(run_project(dag, selected, adapter, run_id=RUN_ID, strict=True)) + + def test_naming_failure_fails_the_node_rather_than_the_run(self): + long_name = "x" * MAX_IDENTIFIER_LENGTH + trouve = _compile(Trouve(sql="SELECT 1 AS id"), f"db.s.{long_name}") + dag = build_dag([trouve]) + adapter, executed = _make_adapter() + + results = list(run_project( + dag, get_executable_nodes(dag), adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + assert results[0].status == RunStatus.FAILURE + assert "max 255" in results[0].error + assert not any("CREATE OR REPLACE TABLE" in sql for sql in executed) + + +class TestStrictRunnerPandas: + def test_dataframe_is_written_to_staging_then_swapped(self): + def transform() -> pd.DataFrame: + return pd.DataFrame({"id": [1, 2]}) + + trouve = _compile( + Trouve(df_fn=transform), + "db.s.orders", + execution_type=ExecutionType.PANDAS, + ) + dag = build_dag([trouve]) + adapter, executed = _make_adapter() + adapter.write_dataframe = MagicMock( + return_value=QueryResult(query_id="w1", query_url="u1", success=True) + ) + + results = list(run_project( + dag, get_executable_nodes(dag), adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert results[0].status == RunStatus.SUCCESS + # Reported under the real name even though the write went to staging. + assert results[0].full_name == "db.s.orders" + assert adapter.write_dataframe.call_args.kwargs["full_name"] == staging + assert adapter.write_dataframe.call_args.kwargs["table_name"] == staging.split(".")[2] + assert any(f"ALTER TABLE {staging} SWAP WITH db.s.orders" in sql for sql in executed) + + def test_failing_tests_discard_the_staging_table(self): + def transform() -> pd.DataFrame: + return pd.DataFrame({"id": [1, 2]}) + + trouve = _compile( + Trouve(df_fn=transform), + "db.s.orders", + execution_type=ExecutionType.PANDAS, + ) + dag = build_dag([trouve]) + adapter, executed = _make_adapter() + adapter.write_dataframe = MagicMock( + return_value=QueryResult(query_id="w1", query_url="u1", success=True) + ) + + results = list(run_project( + dag, get_executable_nodes(dag), adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: False, strict=True, + )) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert results[0].status == RunStatus.FAILURE + assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) + + +class TestNonStrictUnchanged: + def test_target_is_written_directly_and_tested_afterwards(self): + dag, selected = _single_table_dag() + adapter, executed = _make_adapter() + tested: list[str] = [] + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, + after_node_success=lambda _n, physical: (tested.append(physical) or True), + )) + + assert results[0].status == RunStatus.SUCCESS + assert tested == ["db.s.orders"] + assert any("CREATE OR REPLACE TABLE db.s.orders" in sql for sql in executed) + assert not any(STRICT_SUFFIX in sql for sql in executed) + + def test_failing_tests_do_not_fail_the_node_itself(self): + """Without strict mode the write already happened; only downstream is cut off.""" + dag, selected = _single_table_dag() + adapter, _ = _make_adapter() + + results = list(run_project( + dag, selected, adapter, + run_id=RUN_ID, after_node_success=lambda _n, _p: False, + )) + + assert results[0].status == RunStatus.SUCCESS + + +class TestPhysicalNameOverride: + """run_tests can be pointed at a staging object while still reporting the real name.""" + + def _dag_with_test(self): + from clair.trouves.test import TestNotNull + + trouve = _compile( + Trouve(sql="SELECT 1 AS id", tests=[TestNotNull(column="id")]), + "db.s.orders", + ) + return build_dag([trouve]) + + def test_sql_queries_the_override(self): + from clair.core.test_runner import run_tests + + dag = self._dag_with_test() + adapter, executed = _make_adapter() + + run_tests(dag, ["db.s.orders"], adapter, physical_names={"db.s.orders": "db.s.staging"}) + + assert any("db.s.staging" in sql for sql in executed) + assert not any("FROM db.s.orders" in sql for sql in executed) + + def test_results_still_report_the_routed_name(self): + from clair.core.test_runner import run_tests + + dag = self._dag_with_test() + adapter, _ = _make_adapter() + + results = run_tests( + dag, ["db.s.orders"], adapter, physical_names={"db.s.orders": "db.s.staging"} + ) + + assert [r.full_name for r in results] == ["db.s.orders"] + + def test_sampling_applies_to_the_override(self): + from clair.core.test_runner import run_tests + + dag = self._dag_with_test() + adapter, executed = _make_adapter() + + run_tests( + dag, ["db.s.orders"], adapter, + use_sample=True, + physical_names={"db.s.orders": "db.s.staging"}, + ) + + assert any("SELECT TOP 1000 * FROM db.s.staging" in sql for sql in executed) + assert not any("db.s.orders" in sql for sql in executed) + + +class TestStrictCliGuard: + def test_strict_with_no_test_is_rejected(self): + import structlog + from click.testing import CliRunner + + from clair.cli.main import cli + + try: + result = CliRunner().invoke(cli, ["run", "--strict", "--no-test"]) + finally: + # The CLI binds structlog to the runner's stdout/stderr, which are + # closed on exit; reset so later tests log to real streams. + structlog.reset_defaults() + + assert result.exit_code == 1 + + +class TestStrictCompilePlan: + def test_plan_shows_staging_build_test_checkpoint_and_swap(self): + trouve = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.orders") + statements = build_statements(trouve, RunMode.FULL_REFRESH, RUN_ID, strict=True) + + staging = strict_staging_name("db.s.orders", RUN_ID) + assert f"CREATE OR REPLACE TABLE {staging}" in statements[0] + assert "tests run against the staging object" in statements[1] + assert f"ALTER TABLE {staging} SWAP WITH db.s.orders" in statements[2] + + def test_incremental_plan_starts_with_a_clone(self): + trouve = _compile( + Trouve( + sql="SELECT 1 AS id", + run_config=RunConfig( + run_mode=RunMode.INCREMENTAL, incremental_mode=IncrementalMode.APPEND + ), + ), + "db.s.orders", + ) + statements = build_statements(trouve, RunMode.INCREMENTAL, RUN_ID, strict=True) + assert "CLONE db.s.orders" in statements[0] + + def test_non_strict_plan_is_the_plain_build(self): + trouve = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.orders") + assert build_statements(trouve, RunMode.FULL_REFRESH, RUN_ID) == trouve.build_sql( + RunMode.FULL_REFRESH, RUN_ID + ) From 4b0c48f589f5b659444e1f2b0663c56a78f8ad50 Mon Sep 17 00:00:00 2001 From: OmerBaddour Date: Sun, 2 Aug 2026 20:11:11 -0400 Subject: [PATCH 2/2] fix: promote with COPY GRANTS clone, retain rejected candidates Verified against a live Snowflake account that ALTER TABLE ... SWAP WITH attaches grants to the object rather than the name: after a swap, the production name carries only the staging object's grants and any privilege granted directly on the target is silently revoked. Promotion is now: CREATE OR REPLACE TABLE CLONE COPY GRANTS which is equally metadata-only, copies every privilege except OWNERSHIP from the object being replaced, and -- because COPY GRANTS falls back to the clone source when there is no object to replace -- removes the SWAP-vs-RENAME branch entirely. Views get CREATE OR REPLACE VIEW ... COPY GRANTS, which fixes the same defect on the view path. Also: - Staging objects are no longer dropped on failure. A rejected candidate is the only copy of what the run produced and reproducing it means re-running everything upstream, so it is retained and its name reported in the error. Staging is dropped only after a successful promotion. - Shorten the suffix from __clair_strict_ to __clair_ (47 -> 40 chars of identifier budget). - Record at the length check that Snowflake's 255-char limit applies per object name, not per fully-qualified path -- a 767-char database.schema.table is legal. Co-Authored-By: Claude Opus 5 --- site-docs/docs/cli/compile.md | 2 +- site-docs/docs/cli/run.md | 6 +- site-docs/docs/guides/data-quality-tests.md | 2 +- site-docs/docs/guides/strict-mode.md | 56 ++++++++--- src/clair/cli/main.py | 5 +- src/clair/core/compiler.py | 13 +-- src/clair/core/runner.py | 75 ++++++++------ src/clair/core/strict.py | 105 +++++++++++--------- tests/unit/test_strict.py | 95 ++++++++++-------- 9 files changed, 211 insertions(+), 148 deletions(-) diff --git a/site-docs/docs/cli/compile.md b/site-docs/docs/cli/compile.md index 49f6e62..3dda0ad 100644 --- a/site-docs/docs/cli/compile.md +++ b/site-docs/docs/cli/compile.md @@ -18,7 +18,7 @@ clair compile --project . --env prod # Compile only the orders schema clair compile --project . --select='refined.orders.*' -# Show the strict-mode plan: staging build, test checkpoint, swap +# Show the strict-mode plan: staging build, test checkpoint, promotion clair compile --project . --strict ``` diff --git a/site-docs/docs/cli/run.md b/site-docs/docs/cli/run.md index ab36770..528c8a2 100644 --- a/site-docs/docs/cli/run.md +++ b/site-docs/docs/cli/run.md @@ -35,7 +35,7 @@ SOURCE Trouves pass through (no SQL is executed against them). After each successful TABLE or VIEW, attached tests run automatically. If any test fails, the run exits with a non-zero status code. Use `--no-test` to skip tests. -By default the target is written before its tests run, so a failing test means bad data is already in production. `--strict` builds each Trouve into a run-scoped staging object, tests it there, and swaps it into the real name only if every test passes — see [Strict Mode](../guides/strict-mode.md). It cannot be combined with `--no-test`. +By default the target is written before its tests run, so a failing test means bad data is already in production. `--strict` builds each Trouve into a run-scoped staging object, tests it there, and promotes it into the real name only if every test passes — see [Strict Mode](../guides/strict-mode.md). It cannot be combined with `--no-test`. ## Flags @@ -47,12 +47,12 @@ By default the target is written before its tests run, so a failing test means b | `--run-mode` | `full_refresh` | `full_refresh` or `incremental`. Overrides each Trouve's `run_config`. | | `--no-test` | `false` | Skip data quality tests | | `--sample` | `false` | Run tests against `SELECT TOP 1000 *` (skips `TestRowCount`) | -| `--strict` | `false` | Build into a staging object, test, then swap into place. Incompatible with `--no-test`. | +| `--strict` | `false` | Build into a staging object, test, then promote into place. Incompatible with `--no-test`. | ## Exit codes - `0` — all Trouves succeeded and all tests passed -- `1` — one or more Trouves failed, or one or more tests failed (under `--strict`, a Trouve whose tests failed is itself reported as failed and its target is left unchanged) +- `1` — one or more Trouves failed, or one or more tests failed (under `--strict`, a Trouve whose tests failed is itself reported as failed, its target is left unchanged, and the rejected candidate is retained for inspection) ## See also diff --git a/site-docs/docs/guides/data-quality-tests.md b/site-docs/docs/guides/data-quality-tests.md index 933dfce..9c66ac6 100644 --- a/site-docs/docs/guides/data-quality-tests.md +++ b/site-docs/docs/guides/data-quality-tests.md @@ -143,7 +143,7 @@ See also: [Tests API reference](../reference/tests-api.md). ## Testing before publishing -Tests run after a Trouve has been materialized — that is the only point at which there is a table to query. By default that means a failing test tells you production is already wrong. [Strict Mode](strict-mode.md) inverts this: each Trouve is built into a staging object, tested there, and swapped into its real name only if every test passes. +Tests run after a Trouve has been materialized — that is the only point at which there is a table to query. By default that means a failing test tells you production is already wrong. [Strict Mode](strict-mode.md) inverts this: each Trouve is built into a staging object, tested there, and promoted into its real name only if every test passes. ```bash clair run --project . --env prod --strict diff --git a/site-docs/docs/guides/strict-mode.md b/site-docs/docs/guides/strict-mode.md index d3b0653..e08a4ca 100644 --- a/site-docs/docs/guides/strict-mode.md +++ b/site-docs/docs/guides/strict-mode.md @@ -10,12 +10,20 @@ clair run --project . --env prod --strict ## What happens per Trouve -1. **Build into staging.** The Trouve is materialized as `
__clair_strict_`, a sibling object in the same database and schema. +1. **Build into staging.** The Trouve is materialized as `
__clair_`, a sibling object in the same database and schema. 2. **Test the staging object.** The Trouve's data quality tests run against the staging object, not the target. -3. **Promote on pass.** For a `TABLE`, clair issues `ALTER TABLE ... SWAP WITH ...` — a metadata-only operation whose cost does not scale with table size — then drops the superseded copy. For a `VIEW`, the target is recreated with `CREATE OR REPLACE VIEW`. -4. **Discard on fail.** The staging object is dropped and the target is left exactly as it was. The Trouve is reported as a failure and everything downstream of it is skipped. +3. **Promote on pass.** For a `TABLE`, clair issues `CREATE OR REPLACE TABLE CLONE COPY GRANTS` — a metadata-only operation whose cost does not scale with table size — then drops the staging copy. For a `VIEW`, the target is recreated with `CREATE OR REPLACE VIEW ... COPY GRANTS`. +4. **Keep the candidate on fail.** The target is left exactly as it was, and the staging object is deliberately **not** dropped — see [When something fails](#when-something-fails). The Trouve is reported as a failure and everything downstream of it is skipped. -Promotion happens immediately after each node's tests, before the next node starts. Dependents therefore always read their upstreams under the real names their SQL references — you never write `__clair_strict_` anywhere yourself. +Promotion happens immediately after each node's tests, before the next node starts. Dependents therefore always read their upstreams under the real names their SQL references — you never write `__clair_` anywhere yourself. + +## Grants + +Promotion uses `COPY GRANTS`, and it is not optional. Snowflake attaches privileges to the *object*, not to the name: an `ALTER TABLE ... SWAP WITH ...` carries a table's grants away under the staging name and leaves the production name holding only whatever the staging object was created with. Any privilege granted directly on a target would be silently revoked on every run. + +`COPY GRANTS` copies every privilege except `OWNERSHIP` from the object being replaced — or, when the target does not yet exist, from the clone source, which is why promotion needs no special case for a first run. + +`OWNERSHIP` is the exception: it lands on the role executing the run. If a production object is owned by some other role, strict mode changes its owner. ## Incremental Trouves @@ -23,18 +31,18 @@ An incremental Trouve applies changes on top of state that already exists, so th ```sql -- strict: clone target into staging so incremental statements have a base -CREATE OR REPLACE TABLE db.schema.orders__clair_strict_ CLONE db.schema.orders +CREATE OR REPLACE TABLE db.schema.orders__clair_ CLONE db.schema.orders -INSERT INTO db.schema.orders__clair_strict_ +INSERT INTO db.schema.orders__clair_ SELECT * FROM ( ... ) -- tests run here -ALTER TABLE db.schema.orders__clair_strict_ SWAP WITH db.schema.orders -DROP TABLE IF EXISTS db.schema.orders__clair_strict_ +CREATE OR REPLACE TABLE db.schema.orders CLONE db.schema.orders__clair_ COPY GRANTS +DROP TABLE IF EXISTS db.schema.orders__clair_ ``` -Snowflake clones are metadata-only, so seeding the staging table is constant-time no matter how large the target is. If the target does not exist yet, clair falls back to a full refresh for that Trouve and skips the clone; the staging object is then renamed into place rather than swapped. +Snowflake clones are metadata-only, so seeding the staging table is constant-time no matter how large the target is. If the target does not exist yet, clair falls back to a full refresh for that Trouve and skips the seeding clone; promotion is unchanged. ## Seeing the plan @@ -45,21 +53,37 @@ clair compile --project . --strict clair compile --project . --strict --run-mode incremental ``` -## Failure modes +## When something fails + +Nothing is thrown away. A staging object is dropped only after it has been successfully promoted. | What failed | Target | Staging object | |-------------|--------|----------------| -| The build itself | Untouched | Dropped | -| A data quality test | Untouched | Dropped | -| The swap or rename | Untouched | **Retained** — it holds tested data and is the only record of the run | +| The build itself | Untouched | Retained, if it got far enough to be created | +| A data quality test | Untouched | Retained — the rejected candidate | +| The promotion | Untouched | Retained — it holds tested data and is the only record of the run | + +In all three cases the Trouve is reported as `FAILED`, its dependents are skipped, `clair run` exits `1`, and the error names the staging object so you can query it directly: -In all three cases the Trouve is reported as `FAILED`, its dependents are skipped, and `clair run` exits `1`. +``` +db.schema.orders ... FAILED (2.4s) + Error: strict mode: tests failed, db.schema.orders left unchanged + (rejected candidate retained at db.schema.orders__clair_) +``` + +This is the fastest path to a diagnosis. The candidate is the exact data that failed the test, and rebuilding it otherwise means re-running everything upstream of it. + +The cost is that failed runs accumulate objects. Staging tables created by a full refresh are real copies; a retained incremental clone starts out sharing micro-partitions with its target but diverges — and therefore starts costing real storage — as the target changes. Drop them once you are done with them: + +```sql +SHOW TABLES LIKE '%__clair_%' IN SCHEMA db.schema; +``` ## Costs and caveats -- **Storage.** Each Trouve briefly holds two copies. For a full refresh that is a real second copy for the duration of the node; for an incremental build the clone shares micro-partitions with the target and only diverges as rows change. +- **Storage.** Each Trouve briefly holds two copies. For a full refresh that is a real second copy for the duration of the node; for an incremental build the clone shares micro-partitions with the target and only diverges as rows change. Objects left behind by failures persist until you drop them. - **Tests are required.** `--strict` cannot be combined with `--no-test` — the whole point is to gate promotion on tests. A Trouve with no tests attached still goes through staging and is promoted; strict mode does not manufacture coverage you have not written. -- **Name length.** The suffix adds 47 characters to the table component. Snowflake caps identifiers at 255, so a Trouve whose name is within 47 characters of the cap fails at the naming step with a clear error rather than mid-run. +- **Name length.** The suffix adds 40 characters to the table component. Snowflake caps each identifier at 255 — the limit is per object name, not per fully-qualified path — so a Trouve whose table name is within 40 characters of the cap fails at the naming step with a clear error rather than mid-run. - **`--sample`.** Sampling applies to the staging object, so `--strict --sample` gates promotion on a `TOP 1000` check rather than the full table. ## See also diff --git a/src/clair/cli/main.py b/src/clair/cli/main.py index c9c5353..dc89d41 100644 --- a/src/clair/cli/main.py +++ b/src/clair/cli/main.py @@ -441,8 +441,9 @@ def docs(project: str, port: int, host: str, no_browser: bool) -> None: default=False, help=( "Build each Trouve into a run-scoped staging object, test it, and only " - "promote it into its real name if every test passes. Tables are promoted " - "with a constant-time SWAP; a failing Trouve leaves its target untouched." + "promote it into its real name if every test passes. Promotion is a " + "constant-time clone that carries existing grants. A failing Trouve " + "leaves its target untouched and its rejected candidate in place." ), ) def run(select: tuple[str, ...], exclude: tuple[str, ...], project: str, env: str | None, run_mode: str, no_test: bool, sample: bool, strict: bool) -> None: diff --git a/src/clair/core/compiler.py b/src/clair/core/compiler.py index 9516b05..c39ce47 100644 --- a/src/clair/core/compiler.py +++ b/src/clair/core/compiler.py @@ -15,7 +15,8 @@ from clair.core.runner import resolve_effective_mode from clair.core.strict import ( build_clone_statement, - build_promote_statements, + build_drop_staging_statement, + build_promote_statement, strict_staging_name, ) from clair.exceptions import CompileError @@ -101,8 +102,8 @@ def build_statements( Under strict mode the plan is shown end to end: the optional clone, the build into the staging object, and the promotion that follows a passing test run. - The runner decides between SWAP and RENAME by checking whether the target - exists; compile has no connection, so it shows the SWAP form. + The plan shown is the passing path -- a failing test run stops after the build + and leaves the staging object in place. """ effective_mode = resolve_effective_mode(trouve, run_mode) @@ -119,15 +120,15 @@ def build_statements( assert trouve.compiled is not None statements.append("-- strict: tests run against the staging object here") - statements.extend( - build_promote_statements( + statements.append( + build_promote_statement( trouve.type, staging_name=staging_name, target_name=target_name, - target_exists=True, resolved_sql=trouve.compiled.resolved_sql, ) ) + statements.append(build_drop_staging_statement(trouve.type, staging_name)) return statements diff --git a/src/clair/core/runner.py b/src/clair/core/runner.py index 7dde7e4..e099a77 100644 --- a/src/clair/core/runner.py +++ b/src/clair/core/runner.py @@ -16,9 +16,9 @@ from clair.adapters.base import WarehouseAdapter from clair.core.dag import ClairDag, get_executable_nodes from clair.core.strict import ( - build_cleanup_statement, build_clone_statement, - build_promote_statements, + build_drop_staging_statement, + build_promote_statement, strict_staging_name, ) from clair.exceptions import ClairError, RunError @@ -288,14 +288,18 @@ def _run_df_fn_trouve( ) -def _promote_or_discard( +def _promote_or_retain( trouve: Trouve, adapter: WarehouseAdapter, staging_name: str, target_name: str, tests_passed: bool, ) -> tuple[list[str], list[str], str]: - """Finish a strict-mode node: swap the staging object in, or drop it. + """Finish a strict-mode node: promote the staging object, or leave it for inspection. + + A rejected candidate is never dropped. It is the only copy of what the run + produced, and rebuilding it means re-running every upstream Trouve -- so it is + left in place and its name reported, ready to be queried directly. Returns: (query_ids, query_urls, error). The error is empty when the staging object @@ -313,36 +317,40 @@ def _execute(statement: str) -> str: return "" if query_result.success else (query_result.error or "unknown error") if not tests_passed: - cleanup_error = _execute(build_cleanup_statement(trouve.type, staging_name)) - error = ( + return ( + query_ids, + query_urls, f"strict mode: tests failed, {target_name} left unchanged " - f"(staging object {staging_name} discarded)" + f"(rejected candidate retained at {staging_name})", ) - if cleanup_error: - error += f"; staging cleanup also failed: {cleanup_error}" - return query_ids, query_urls, error - - database_name, schema_name, table_name = target_name.split(".") - target_exists = adapter.table_exists(database_name, schema_name, table_name) assert trouve.compiled is not None - for statement in build_promote_statements( - trouve.type, - staging_name=staging_name, - target_name=target_name, - target_exists=target_exists, - resolved_sql=trouve.compiled.resolved_sql, - ): - promote_error = _execute(statement) - if promote_error: - # Leave the staging object in place: it holds tested data and is the - # only record of what this run produced. - return ( - query_ids, - query_urls, - f"strict mode: tests passed but promotion failed: {promote_error} " - f"(staging object {staging_name} retained)", - ) + promote_error = _execute( + build_promote_statement( + trouve.type, + staging_name=staging_name, + target_name=target_name, + resolved_sql=trouve.compiled.resolved_sql, + ) + ) + if promote_error: + return ( + query_ids, + query_urls, + f"strict mode: tests passed but promotion failed: {promote_error} " + f"(candidate retained at {staging_name})", + ) + + # The target now holds the tested data, so the staging copy is redundant. + # A failure here is untidy, not incorrect -- the node still succeeded. + drop_result = adapter.execute(build_drop_staging_statement(trouve.type, staging_name)) + if not drop_result.success: + logger.warning( + "run.node.staging_drop_failed", + trouve=target_name, + staging=staging_name, + error=drop_result.error, + ) return query_ids, query_urls, "" @@ -500,11 +508,11 @@ def run_project( error = "" if materialized else ((last_result.error if last_result else "") or "") - # Strict mode: test the staging object, then swap it in or throw it away. + # Strict mode: test the staging object, then promote it or leave it be. if staging_name is not None and materialized: assert after_node_success is not None tests_passed = after_node_success(name, staging_name) - promote_ids, promote_urls, strict_error = _promote_or_discard( + promote_ids, promote_urls, strict_error = _promote_or_retain( trouve, adapter, staging_name, routed_name, tests_passed ) query_ids.extend(promote_ids) @@ -513,7 +521,8 @@ def run_project( materialized = False error = strict_error elif staging_name is not None and not materialized: - adapter.execute(build_cleanup_statement(trouve.type, staging_name)) + # Whatever the build managed to produce is worth keeping around. + error = f"{error} (strict staging object left at {staging_name} if it was created)" if materialized: logger.info("run.node.success", trouve=name, duration_seconds=round(duration, 3), query_ids=query_ids) diff --git a/src/clair/core/strict.py b/src/clair/core/strict.py index 43b7c07..4bca618 100644 --- a/src/clair/core/strict.py +++ b/src/clair/core/strict.py @@ -4,15 +4,15 @@ window where the production object holds untested data. Strict mode closes that window: -1. Materialize the Trouve into ``
__clair_strict_``, a sibling object - in the same schema. For incremental Trouves the current target is first - zero-copy cloned into that name so the incremental statements have a base to - apply to. +1. Materialize the Trouve into ``
__clair_``, a sibling object in the + same schema. For incremental Trouves the current target is first zero-copy + cloned into that name so the incremental statements have a base to apply to. 2. Run the Trouve's data quality tests against the staging object. -3. On pass, promote the staging object into the real name. For tables this is - ``ALTER TABLE ... SWAP WITH ...``, a metadata-only operation whose cost does - not scale with table size. -4. On failure, drop the staging object. The production object is never touched. +3. On pass, promote the staging object into the real name with + ``CREATE OR REPLACE TABLE CLONE COPY GRANTS`` -- a + metadata-only operation whose cost does not scale with table size. +4. On failure, leave the staging object in place. The production object is never + touched, and the rejected candidate can be queried directly to find out why. Downstream Trouves are unaffected: promotion happens immediately after each node's tests, so by the time a dependent runs, its upstreams already resolve to the real @@ -25,9 +25,16 @@ from clair.trouves.trouve import TrouveType -STRICT_SUFFIX = "__clair_strict_" +STRICT_SUFFIX = "__clair_" # Snowflake's maximum identifier length. +# +# Verified against a live account: the limit applies to each object name +# individually, not to the fully-qualified path. A 255-character table name is +# accepted; 256 is rejected with "Object name '...' exceeds maximum length limit +# of 255 characters"; and a 767-character database.schema.table (255 per +# component) creates and queries without complaint. Only the table component +# grows under strict mode, so that is the only one checked below. MAX_IDENTIFIER_LENGTH = 255 @@ -39,15 +46,16 @@ def strict_staging_name(full_name: str, run_id: str) -> str: """Return the run-scoped staging name for a Trouve's routed full_name. The suffix is appended to the table component only, so the staging object - lives in the same database and schema as its target -- a requirement for - ``ALTER TABLE ... SWAP WITH ...``. + lives in the same database and schema as its target. Promotion clones rather + than swaps, so this is a convention rather than a hard requirement -- but it + keeps a rejected candidate next to the table it was meant to become. Args: full_name: Routed "database.schema.table" name of the target object. run_id: UUIDv7 hex string identifying this clair run. Returns: - The staging "database.schema.table__clair_strict_" name. + The staging "database.schema.table__clair_" name. Raises: StrictNamingError: If the staging table identifier exceeds 255 characters. @@ -84,56 +92,61 @@ def build_clone_statement(target_name: str, staging_name: str) -> str: ) -def build_promote_statements( +def build_promote_statement( trouve_type: TrouveType, staging_name: str, target_name: str, - target_exists: bool, resolved_sql: str = "", -) -> list[str]: - """Return the statements that promote a tested staging object into its real name. +) -> str: + """Return the statement that promotes a tested staging object into its real name. + + ``COPY GRANTS`` is what makes this safe to run against a production object. + Without it, privileges granted directly on the target are lost: they are + attached to the object, not the name, so ``ALTER TABLE ... SWAP WITH`` carries + them off under the staging name and leaves the production name bare. With + ``COPY GRANTS``, Snowflake copies every privilege except OWNERSHIP from the + object being replaced -- or, when the target does not exist yet, from the + clone source. That covers both cases without a branch. + + OWNERSHIP is the one privilege that does not carry over; it lands on the role + executing the run. ``SWAP`` behaves the same way, so this is not a regression, + but a target owned by some other role will change hands. Args: trouve_type: TABLE or VIEW. SOURCE Trouves are never materialized. staging_name: Routed name of the staging object holding tested data. target_name: Routed name the object should end up under. - target_exists: Whether the target already exists in the warehouse. resolved_sql: The Trouve's resolved SQL; required for VIEW promotion. Returns: - Ordered list of SQL statements to execute. + A single SQL statement. """ if trouve_type == TrouveType.VIEW: - # Views have no SWAP. CREATE OR REPLACE VIEW is itself atomic and, being - # metadata-only, costs nothing -- the staging view proved the SQL is valid - # and its results pass the tests. - return [ + # Views cannot be cloned into place the way tables can, but CREATE OR + # REPLACE VIEW is itself atomic and metadata-only -- the staging view + # proved the SQL is valid and that its results pass the tests. + return ( f"-- strict: promote tested view\n" - f"CREATE OR REPLACE VIEW {target_name} AS (\n{resolved_sql.strip()}\n)", - f"-- strict: drop staging view\n" - f"DROP VIEW IF EXISTS {staging_name}", - ] - - if not target_exists: - return [ - f"-- strict: promote tested table (target did not exist)\n" - f"ALTER TABLE {staging_name} RENAME TO {target_name}" - ] - - return [ - # SWAP is a metadata-only operation: O(1) in the size of either table. - f"-- strict: swap tested staging table into place\n" - f"ALTER TABLE {staging_name} SWAP WITH {target_name}", - # After the swap, the staging name holds the previous target contents. - f"-- strict: drop the superseded table\n" - f"DROP TABLE IF EXISTS {staging_name}", - ] - - -def build_cleanup_statement(trouve_type: TrouveType, staging_name: str) -> str: - """Return the statement that discards a staging object after a failed build or test.""" + f"CREATE OR REPLACE VIEW {target_name} COPY GRANTS AS (\n" + f"{resolved_sql.strip()}\n)" + ) + + # A clone is metadata-only: O(1) in the size of the staging table. + return ( + f"-- strict: promote tested table\n" + f"CREATE OR REPLACE TABLE {target_name} CLONE {staging_name} COPY GRANTS" + ) + + +def build_drop_staging_statement(trouve_type: TrouveType, staging_name: str) -> str: + """Return the statement that drops a staging object after a successful promotion. + + Only ever used on the success path. A staging object left behind by a failed + build or a failed test is deliberately retained -- it is the only copy of the + rejected candidate, and reproducing it means re-running everything upstream. + """ object_type = "VIEW" if trouve_type == TrouveType.VIEW else "TABLE" return ( - f"-- strict: discard untested staging object\n" + f"-- strict: drop the promoted staging object\n" f"DROP {object_type} IF EXISTS {staging_name}" ) diff --git a/tests/unit/test_strict.py b/tests/unit/test_strict.py index 0a11611..9720b3f 100644 --- a/tests/unit/test_strict.py +++ b/tests/unit/test_strict.py @@ -1,4 +1,4 @@ -"""Tests for strict mode -- build into staging, test, then swap.""" +"""Tests for strict mode -- build into staging, test, then promote.""" from __future__ import annotations @@ -16,9 +16,9 @@ MAX_IDENTIFIER_LENGTH, STRICT_SUFFIX, StrictNamingError, - build_cleanup_statement, build_clone_statement, - build_promote_statements, + build_drop_staging_statement, + build_promote_statement, strict_staging_name, ) from clair.exceptions import RunError @@ -82,7 +82,7 @@ def test_suffix_applied_to_table_component_only(self): assert staging == f"db.schema.orders{STRICT_SUFFIX}{RUN_ID}" def test_staging_shares_database_and_schema_with_target(self): - """SWAP WITH requires both objects to live in the same schema.""" + """A rejected candidate should sit next to the table it was meant to become.""" staging = strict_staging_name("analytics.revenue.daily", RUN_ID) assert staging.split(".")[:2] == ["analytics", "revenue"] @@ -107,43 +107,40 @@ def test_accepts_identifier_at_the_limit(self): class TestPromoteStatements: - def test_existing_table_is_swapped_then_old_copy_dropped(self): - statements = build_promote_statements( + def test_table_is_cloned_into_place_carrying_grants(self): + statement = build_promote_statement( TrouveType.TABLE, staging_name="db.s.t__staging", target_name="db.s.t", - target_exists=True, ) - assert "ALTER TABLE db.s.t__staging SWAP WITH db.s.t" in statements[0] - assert "DROP TABLE IF EXISTS db.s.t__staging" in statements[1] + assert "CREATE OR REPLACE TABLE db.s.t CLONE db.s.t__staging COPY GRANTS" in statement - def test_missing_table_is_renamed_into_place(self): - statements = build_promote_statements( + def test_table_promotion_does_not_depend_on_the_target_existing(self): + """COPY GRANTS copies from the replaced table, or the clone source if there is none.""" + statement = build_promote_statement( TrouveType.TABLE, staging_name="db.s.t__staging", target_name="db.s.t", - target_exists=False, ) - assert len(statements) == 1 - assert "ALTER TABLE db.s.t__staging RENAME TO db.s.t" in statements[0] + assert "IF NOT EXISTS" not in statement + assert "SWAP WITH" not in statement + assert "RENAME TO" not in statement - def test_view_is_recreated_and_staging_dropped(self): - statements = build_promote_statements( + def test_view_is_recreated_carrying_grants(self): + statement = build_promote_statement( TrouveType.VIEW, staging_name="db.s.v__staging", target_name="db.s.v", - target_exists=True, resolved_sql="SELECT 1 AS id", ) - assert "CREATE OR REPLACE VIEW db.s.v" in statements[0] - assert "SELECT 1 AS id" in statements[0] - assert "DROP VIEW IF EXISTS db.s.v__staging" in statements[1] + assert "CREATE OR REPLACE VIEW db.s.v COPY GRANTS AS" in statement + assert "SELECT 1 AS id" in statement - def test_cleanup_uses_matching_object_type(self): - assert "DROP TABLE IF EXISTS db.s.t" in build_cleanup_statement( + def test_drop_staging_uses_matching_object_type(self): + assert "DROP TABLE IF EXISTS db.s.t" in build_drop_staging_statement( TrouveType.TABLE, "db.s.t" ) - assert "DROP VIEW IF EXISTS db.s.v" in build_cleanup_statement( + assert "DROP VIEW IF EXISTS db.s.v" in build_drop_staging_statement( TrouveType.VIEW, "db.s.v" ) @@ -235,7 +232,10 @@ def on_success(node_name: str, physical_name: str) -> bool: # Tests ran against the staging object, not the target. assert tested == [staging] assert any(f"CREATE OR REPLACE TABLE {staging}" in sql for sql in executed) - assert any(f"ALTER TABLE {staging} SWAP WITH db.s.orders" in sql for sql in executed) + assert any( + f"CREATE OR REPLACE TABLE db.s.orders CLONE {staging} COPY GRANTS" in sql + for sql in executed + ) def test_target_is_never_written_before_tests_pass(self): dag, selected = _single_table_dag() @@ -255,7 +255,7 @@ def on_success(node_name: str, physical_name: str) -> bool: "CREATE OR REPLACE TABLE db.s.orders AS" in sql for sql in sql_at_test_time[0] ) - def test_failing_tests_drop_staging_and_leave_target_untouched(self): + def test_failing_tests_retain_the_candidate_and_leave_target_untouched(self): dag, selected = _single_table_dag() adapter, executed = _make_adapter() @@ -267,10 +267,13 @@ def test_failing_tests_drop_staging_and_leave_target_untouched(self): staging = strict_staging_name("db.s.orders", RUN_ID) assert results[0].status == RunStatus.FAILURE assert "tests failed" in results[0].error - assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) - assert not any("SWAP WITH" in sql for sql in executed) + # The rejected candidate is the whole point: keep it, and say where it is. + assert staging in results[0].error + assert not any("DROP TABLE" in sql for sql in executed) + # db.s.orders is a prefix of the staging name, so match the promotion exactly. + assert not any("CREATE OR REPLACE TABLE db.s.orders CLONE" in sql for sql in executed) - def test_missing_target_is_renamed_rather_than_swapped(self): + def test_promotion_is_identical_when_the_target_does_not_exist(self): dag, selected = _single_table_dag() adapter, executed = _make_adapter(target_exists=False) @@ -279,10 +282,13 @@ def test_missing_target_is_renamed_rather_than_swapped(self): run_id=RUN_ID, after_node_success=lambda _n, _p: True, strict=True, )) - assert any("RENAME TO db.s.orders" in sql for sql in executed) - assert not any("SWAP WITH" in sql for sql in executed) + staging = strict_staging_name("db.s.orders", RUN_ID) + assert any( + f"CREATE OR REPLACE TABLE db.s.orders CLONE {staging} COPY GRANTS" in sql + for sql in executed + ) - def test_failed_materialization_cleans_up_staging(self): + def test_failed_materialization_retains_whatever_was_built(self): dag, selected = _single_table_dag() adapter, executed = _make_adapter(fail_on="CREATE OR REPLACE TABLE") @@ -293,11 +299,12 @@ def test_failed_materialization_cleans_up_staging(self): staging = strict_staging_name("db.s.orders", RUN_ID) assert results[0].status == RunStatus.FAILURE - assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) + assert staging in results[0].error + assert not any("DROP TABLE" in sql for sql in executed) def test_failed_promotion_retains_staging_and_reports_failure(self): dag, selected = _single_table_dag() - adapter, executed = _make_adapter(fail_on="SWAP WITH") + adapter, executed = _make_adapter(fail_on="CLONE") results = list(run_project( dag, selected, adapter, @@ -341,7 +348,10 @@ def test_incremental_fallback_to_full_refresh_skips_the_clone(self): after_node_success=lambda _n, _p: True, strict=True, )) - assert not any("CLONE" in sql for sql in executed) + staging = strict_staging_name("db.s.orders", RUN_ID) + # The promotion clone still runs; what must not happen is seeding staging + # from a target that does not exist yet. + assert not any(f"CREATE OR REPLACE TABLE {staging} CLONE" in sql for sql in executed) def test_view_is_created_in_staging_then_replaced_at_target(self): dag, selected = _single_table_dag(trouve_type=TrouveType.VIEW) @@ -424,7 +434,7 @@ def test_naming_failure_fails_the_node_rather_than_the_run(self): class TestStrictRunnerPandas: - def test_dataframe_is_written_to_staging_then_swapped(self): + def test_dataframe_is_written_to_staging_then_promoted(self): def transform() -> pd.DataFrame: return pd.DataFrame({"id": [1, 2]}) @@ -450,9 +460,12 @@ def transform() -> pd.DataFrame: assert results[0].full_name == "db.s.orders" assert adapter.write_dataframe.call_args.kwargs["full_name"] == staging assert adapter.write_dataframe.call_args.kwargs["table_name"] == staging.split(".")[2] - assert any(f"ALTER TABLE {staging} SWAP WITH db.s.orders" in sql for sql in executed) + assert any( + f"CREATE OR REPLACE TABLE db.s.orders CLONE {staging} COPY GRANTS" in sql + for sql in executed + ) - def test_failing_tests_discard_the_staging_table(self): + def test_failing_tests_retain_the_staging_table(self): def transform() -> pd.DataFrame: return pd.DataFrame({"id": [1, 2]}) @@ -474,7 +487,8 @@ def transform() -> pd.DataFrame: staging = strict_staging_name("db.s.orders", RUN_ID) assert results[0].status == RunStatus.FAILURE - assert any(f"DROP TABLE IF EXISTS {staging}" in sql for sql in executed) + assert staging in results[0].error + assert not any("DROP TABLE" in sql for sql in executed) class TestNonStrictUnchanged: @@ -576,14 +590,15 @@ def test_strict_with_no_test_is_rejected(self): class TestStrictCompilePlan: - def test_plan_shows_staging_build_test_checkpoint_and_swap(self): + def test_plan_shows_staging_build_test_checkpoint_and_promotion(self): trouve = _compile(Trouve(sql="SELECT 1 AS id"), "db.s.orders") statements = build_statements(trouve, RunMode.FULL_REFRESH, RUN_ID, strict=True) staging = strict_staging_name("db.s.orders", RUN_ID) assert f"CREATE OR REPLACE TABLE {staging}" in statements[0] assert "tests run against the staging object" in statements[1] - assert f"ALTER TABLE {staging} SWAP WITH db.s.orders" in statements[2] + assert f"CREATE OR REPLACE TABLE db.s.orders CLONE {staging} COPY GRANTS" in statements[2] + assert f"DROP TABLE IF EXISTS {staging}" in statements[3] def test_incremental_plan_starts_with_a_clone(self): trouve = _compile(