diff --git a/site-docs/docs/cli/compile.md b/site-docs/docs/cli/compile.md index 7bfe628..3dda0ad 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, promotion +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..528c8a2 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 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 | 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 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 +- `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 - [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..9c66ac6 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 promoted 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..e08a4ca --- /dev/null +++ b/site-docs/docs/guides/strict-mode.md @@ -0,0 +1,93 @@ +# 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_`, 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 `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_` 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 + +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_ CLONE db.schema.orders + +INSERT INTO db.schema.orders__clair_ +SELECT * FROM ( ... ) + +-- tests run here + +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 seeding clone; promotion is unchanged. + +## 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 +``` + +## 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 | 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: + +``` +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. 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 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 + +- [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..dc89d41 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,27 @@ 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. 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: """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 +481,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 +493,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 +506,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..c39ce47 100644 --- a/src/clair/core/compiler.py +++ b/src/clair/core/compiler.py @@ -13,6 +13,12 @@ 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_drop_staging_statement, + build_promote_statement, + 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 +92,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 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) + + 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.append( + build_promote_statement( + trouve.type, + staging_name=staging_name, + target_name=target_name, + resolved_sql=trouve.compiled.resolved_sql, + ) + ) + statements.append(build_drop_staging_statement(trouve.type, staging_name)) + return statements + + def write_compile_output( dag: ClairDag, selected: list[str], @@ -93,6 +139,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 +151,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 +222,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..e099a77 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_clone_statement, + build_drop_staging_statement, + build_promote_statement, + 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,81 @@ def _run_df_fn_trouve( ) +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: 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 + 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: + return ( + query_ids, + query_urls, + f"strict mode: tests failed, {target_name} left unchanged " + f"(rejected candidate retained at {staging_name})", + ) + + assert trouve.compiled is not None + 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, "" + + 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 +370,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 +417,134 @@ 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 + + # 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 - 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 + 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 - duration = time.monotonic() - start + 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) - # 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]) + if not statements: + continue - if all_succeeded: + # 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 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_retain( + 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: + # 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) 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..4bca618 --- /dev/null +++ b/src/clair/core/strict.py @@ -0,0 +1,152 @@ +"""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_``, 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 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 +names their SQL references. +""" + +from __future__ import annotations + +from clair.exceptions import ClairError +from clair.trouves.trouve import TrouveType + + +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 + + +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. 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_" 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_statement( + trouve_type: TrouveType, + staging_name: str, + target_name: str, + resolved_sql: str = "", +) -> 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. + resolved_sql: The Trouve's resolved SQL; required for VIEW promotion. + + Returns: + A single SQL statement. + """ + if trouve_type == TrouveType.VIEW: + # 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} 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: drop the promoted 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..9720b3f --- /dev/null +++ b/tests/unit/test_strict.py @@ -0,0 +1,620 @@ +"""Tests for strict mode -- build into staging, test, then promote.""" + +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_clone_statement, + build_drop_staging_statement, + build_promote_statement, + 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): + """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"] + + 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_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", + ) + assert "CREATE OR REPLACE TABLE db.s.t CLONE db.s.t__staging COPY GRANTS" in statement + + 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", + ) + assert "IF NOT EXISTS" not in statement + assert "SWAP WITH" not in statement + assert "RENAME TO" not in statement + + 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", + resolved_sql="SELECT 1 AS id", + ) + assert "CREATE OR REPLACE VIEW db.s.v COPY GRANTS AS" in statement + assert "SELECT 1 AS id" in 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_drop_staging_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"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() + 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_retain_the_candidate_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 + # 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_promotion_is_identical_when_the_target_does_not_exist(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, + )) + + 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_retains_whatever_was_built(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 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="CLONE") + + 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, + )) + + 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) + 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_promoted(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"CREATE OR REPLACE TABLE db.s.orders CLONE {staging} COPY GRANTS" in sql + for sql in executed + ) + + def test_failing_tests_retain_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 staging in results[0].error + assert not any("DROP TABLE" 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_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"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( + 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 + )