diff --git a/CHANGELOG.md b/CHANGELOG.md index 87fa621..87e2d13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,14 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `table_columns` reads `information_schema`. `ensure_table` accepts an upsert key (no constraint DDL) and rejects it with `format_version: 1`. `load` now shares the stage helpers (no behavior change). +- The harness writes each node with its `mode_type`: `load` for `replace`, + `append`, or `upsert` with the node's `primary_key` (also passed to + `ensure_table`). An adapter without the needed capability fails with + `LoadError: does not support ` before the script runs; an + `upsert` whose rows repeat a key fails with `ConformError` and writes + nothing. Before an `append` or `upsert` the live table's columns and + nullability are compared with `output_columns`; any difference is a + `LoadError` naming each one. See "Write modes" in the README. ### Changed diff --git a/README.md b/README.md index 5b8ddd1..35d242b 100644 --- a/README.md +++ b/README.md @@ -53,7 +53,8 @@ engine adapters, the validation runner, and the node harness. The former `continuo-validation` repository was merged in — there is no longer a separate validation-side port, adapter class, entry-point group, or image. One `WarehouseAdapter` per engine serves both the data plane (`fetch` / -`ensure_table` / `load`) and validation (`ensure_schema` / `drop_schema` / +`ensure_table` / `load`, plus the opt-in `table_columns` / `append` / `upsert` +capabilities) and validation (`ensure_schema` / `drop_schema` / `build_empty_from_sql` / `build_empty_from_columns` / `clone_empty_from_prod` / `check_binds`), and one image per engine runs both roles. @@ -175,6 +176,62 @@ produced, never how it is checked or written. script reads with `os.environ[...]`. The contract never holds a secret value or env var name; the Secret itself is created in the cluster by the operator. +## Write modes + +`mode_type:` chooses how the harness writes a node's table on each run: + +| `mode_type` | Each run | Rows from earlier runs | +| --- | --- | --- | +| `replace` (default) | empties the table, inserts this run's rows | gone | +| `append` | inserts this run's rows | kept | +| `upsert` | inserts rows with a new key, overwrites rows whose key exists | kept unless overwritten; never deleted | + +```yaml +- schema: fx + table: fx_rates + kind: python-api + mode_type: upsert + primary_key: [rate_date, base, quote] + output_columns: + - {name: rate_date, type: DATE, nullable: false} + - {name: base, type: VARCHAR(3), nullable: false} + - {name: quote, type: VARCHAR(3), nullable: false} + - {name: rate, type: DECIMAL(12,6)} +``` + +- `python-csv` nodes only `replace`. +- `primary_key` is required for `upsert` and rejected otherwise: a non-empty + list of output columns declared `nullable: false` (a NULL key never + matches), each spelled exactly as in `output_columns`, names unique + case-insensitively. A Postgres `replace` table that + wants uniqueness uses `config.indexes` with `unique: true`. +- Every write is atomic: all of the run's rows or none. An `append` or + `upsert` run with zero rows writes nothing (a `replace` run with zero rows + still empties the table). +- `append` is a plain insert: a run that committed but was reported failed + and is retried stores its rows twice. `upsert` is idempotent. +- An `upsert` run whose own rows repeat a key fails with `ConformError` and + writes nothing. +- Before an `append` or `upsert` the harness compares the live table with + `output_columns` — names case-insensitively, nullability exactly; order and + types are not compared — and fails with `LoadError` naming every missing + column, unexpected column and nullability mismatch. Changing an `append` or + `upsert` table (columns, types, nullability, key or mode) needs a new table + name (`_v2`) under a new contract; continuo rejects the release otherwise. + Adding a nullable column is no exception: the existing rows cannot be + backfilled. +- The installed engine must support the mode; all three shipped adapters do. + A bring-your-own adapter declares `AppendCapable` / `UpsertCapable` from + `continuo-engine-contract` (and can run its conformance suite); without them + an `append`/`upsert` node fails with `LoadError: does not support + ` before its script runs. +- Engines: on Postgres the key is the table's `PRIMARY KEY`, created with the + table; DuckLake and Trino/Iceberg have no constraints, so the key is + guaranteed by the write path only. On Trino an `upsert` table needs Iceberg + `format_version` 2 (the default). +- `full_refresh` stays refused for every python node: a rebuild would destroy + an accumulating table. + ## The script API A node script is a Python file with exactly one required entry point: @@ -261,8 +318,8 @@ text. | `ContractError` | Contract missing or invalid, node not found for `NODE_ID`, or the declared script is missing/unreachable. | The contract yaml or the `script:` path. | | `ReadError` | `ctx.read()` was called with an undeclared name, or a declared read failed at the warehouse. | The `reads:` map, or the upstream query/warehouse access. | | `ScriptError` | `run()` raised, has no callable `run`, or returned a value that isn't Arrow-convertible. | The node script. | -| `ConformError` | Structural mismatch (extra/missing/duplicate columns), a strict-cast failure, a not-null violation, or a VARCHAR/CHAR overflow. | The script's output shape, or the `output_columns` declaration. | -| `LoadError` | Adapter construction failed, or the DDL/INSERT failed at the warehouse during the write. | Warehouse connectivity/permissions, or the target table. | +| `ConformError` | Structural mismatch (extra/missing/duplicate columns), a strict-cast failure, a not-null violation, a VARCHAR/CHAR overflow, or an `upsert` run whose rows repeat a primary key. | The script's output shape, or the `output_columns` declaration. | +| `LoadError` | Adapter construction failed, the installed engine does not support the node's `mode_type`, the live table's shape differs from `output_columns` (append/upsert), or the DDL/INSERT failed at the warehouse during the write. | Warehouse connectivity/permissions, the engine image, or the target table. | ## Engine selection diff --git a/adapters/duckdb/tests/test_integration_duckdb_runtime.py b/adapters/duckdb/tests/test_integration_duckdb_runtime.py index ecb1b78..fb0d153 100644 --- a/adapters/duckdb/tests/test_integration_duckdb_runtime.py +++ b/adapters/duckdb/tests/test_integration_duckdb_runtime.py @@ -251,3 +251,81 @@ def test_non_ascii_names_that_differ_only_in_case_are_two_tables(adapter, schema "ORDER BY table_name" ).to_pylist() assert sorted(row["table_name"] for row in names) == ["É", "é"] + + +# --- write modes end to end ---------------------------------------------------- + +_FX_COLUMNS = [ + {"name": "rate_date", "type": "DATE", "nullable": False}, + {"name": "base", "type": "VARCHAR(3)", "nullable": False}, + {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True}, +] + +_FX_SCRIPT = ( + "import datetime\n" + "from decimal import Decimal\n\n" + "import pyarrow as pa\n\n" + "ROWS = {rows!r}\n\n\n" + "def run(ctx):\n" + " return pa.table({{\n" + " 'rate_date': pa.array([datetime.date.fromisoformat(r[0]) for r in ROWS], pa.date32()),\n" + " 'base': pa.array([r[1] for r in ROWS], pa.string()),\n" + " 'rate': pa.array([Decimal(r[2]) for r in ROWS], pa.decimal128(12, 6)),\n" + " }})\n" +) + + +def _fx_run_env(repo, schema, run, rows, **mode): + """A python-api fx node returning *rows*, with a script file unique to *run*. + + One file per run: rewriting the same file within a second at the same size + would let Python reuse its cached bytecode and return the previous rows. + """ + (repo / "contracts").mkdir(exist_ok=True) + (repo / "scripts").mkdir(exist_ok=True) + script = f"scripts/fx_{run}.py" + (repo / script).write_text(_FX_SCRIPT.format(rows=rows)) + node = { + "schema": schema, "table": "fx_rates", "owner": "m", "schedule": "hourly", + "criticality": "SECONDARY", "kind": "python-api", "script": script, + "output_columns": _FX_COLUMNS, **mode, + } + (repo / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]})) + return { + "NODE_ID": f"python-api.svc.{schema}.fx_rates", "TABLE_NAME": "fx_rates", + "TARGET_SCHEMA": schema, "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo), + } + + +def test_run_node_append_keeps_the_rows_of_earlier_runs(adapter_factory, schema, scalar, tmp_path): + env = _fx_run_env(tmp_path, schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append") + assert run_node(env, adapter=adapter_factory()) == 0 + env = _fx_run_env( + tmp_path, schema, "run2", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-09", "EUR", "1.090000")], + mode_type="append", + ) + assert run_node(env, adapter=adapter_factory()) == 0 + assert scalar(f'SELECT count(*) AS n FROM "{schema}"."fx_rates"') == 3 + + +def test_run_node_upsert_overwrites_by_key(adapter, adapter_factory, schema, tmp_path): + key = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]} + env = _fx_run_env( + tmp_path, schema, "run1", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-08", "GBP", "0.861000")], **key, + ) + assert run_node(env, adapter=adapter_factory()) == 0 + env = _fx_run_env( + tmp_path, schema, "run2", + [("2026-10-08", "EUR", "1.090000"), ("2026-10-09", "EUR", "1.100000")], **key, + ) + assert run_node(env, adapter=adapter_factory()) == 0 + rows = adapter.fetch( + f'SELECT rate_date, base, rate FROM "{schema}"."fx_rates" ORDER BY rate_date, base' + ).to_pylist() + assert rows == [ + {"rate_date": date(2026, 10, 8), "base": "EUR", "rate": Decimal("1.090000")}, + {"rate_date": date(2026, 10, 8), "base": "GBP", "rate": Decimal("0.861000")}, + {"rate_date": date(2026, 10, 9), "base": "EUR", "rate": Decimal("1.100000")}, + ] diff --git a/adapters/postgres/tests/test_integration_runtime_postgres.py b/adapters/postgres/tests/test_integration_runtime_postgres.py index d9ba586..9cf242b 100644 --- a/adapters/postgres/tests/test_integration_runtime_postgres.py +++ b/adapters/postgres/tests/test_integration_runtime_postgres.py @@ -3,6 +3,7 @@ Run the stack first: docker compose -f tests/smoke/postgres-stack/docker-compose.yml up -d --wait """ +import datetime import decimal import os import uuid @@ -382,3 +383,110 @@ def test_run_node_csv_kind_loads_minio_csv_into_postgres(clean_schema, csv_minio rows = cur.fetchall() conn.close() assert rows == [(1, 10.5), (2, 20.0), (3, 5.25)] + + +# --- write modes end to end ---------------------------------------------------- + +_FX_COLUMNS = [ + {"name": "rate_date", "type": "DATE", "nullable": False}, + {"name": "base", "type": "VARCHAR(3)", "nullable": False}, + {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True}, +] + +_FX_SCRIPT = ( + "import datetime\n" + "from decimal import Decimal\n\n" + "import pyarrow as pa\n\n" + "ROWS = {rows!r}\n\n\n" + "def run(ctx):\n" + " return pa.table({{\n" + " 'rate_date': pa.array([datetime.date.fromisoformat(r[0]) for r in ROWS], pa.date32()),\n" + " 'base': pa.array([r[1] for r in ROWS], pa.string()),\n" + " 'rate': pa.array([Decimal(r[2]) for r in ROWS], pa.decimal128(12, 6)),\n" + " }})\n" +) + + +def _fx_run_env(repo, schema, run, rows, **mode): + """A python-api fx node returning *rows*, with a script file unique to *run*. + + One file per run: rewriting the same file within a second at the same size + would let Python reuse its cached bytecode and return the previous rows. + """ + (repo / "contracts").mkdir(exist_ok=True) + (repo / "scripts").mkdir(exist_ok=True) + script = f"scripts/fx_{run}.py" + (repo / script).write_text(_FX_SCRIPT.format(rows=rows)) + node = { + "schema": schema, "table": "fx_rates", "owner": "m", "schedule": "hourly", + "criticality": "SECONDARY", "kind": "python-api", "script": script, + "output_columns": _FX_COLUMNS, **mode, + } + (repo / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]})) + return { + "NODE_ID": f"python-api.svc.{schema}.fx_rates", "TABLE_NAME": "fx_rates", + "TARGET_SCHEMA": schema, "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo), + } + + +@pytest.mark.integration +def test_run_node_append_keeps_the_rows_of_earlier_runs(clean_schema, tmp_path): + env = _fx_run_env(tmp_path, clean_schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append") + assert run_node(env, adapter=_adapter()) == 0 + env = _fx_run_env( + tmp_path, clean_schema, "run2", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-09", "EUR", "1.090000")], + mode_type="append", + ) + assert run_node(env, adapter=_adapter()) == 0 + assert _count(clean_schema, "fx_rates") == 3 + + +@pytest.mark.integration +def test_run_node_upsert_overwrites_by_key_and_keeps_the_constraint(clean_schema, tmp_path): + """Review Focus 4 and 5 through the whole harness.""" + key = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]} + env = _fx_run_env( + tmp_path, clean_schema, "run1", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-08", "GBP", "0.861000")], **key, + ) + assert run_node(env, adapter=_adapter()) == 0 + env = _fx_run_env( + tmp_path, clean_schema, "run2", + [("2026-10-08", "EUR", "1.090000"), ("2026-10-09", "EUR", "1.100000")], **key, + ) + assert run_node(env, adapter=_adapter()) == 0 + conn = _conn() + with conn.cursor() as cur: + cur.execute( + f'SELECT rate_date, base, rate FROM "{clean_schema}"."fx_rates" ORDER BY rate_date, base' + ) + rows = cur.fetchall() + cur.execute( + "SELECT kcu.column_name FROM information_schema.table_constraints tc " + "JOIN information_schema.key_column_usage kcu " + "ON kcu.constraint_name = tc.constraint_name AND kcu.table_schema = tc.table_schema " + "AND kcu.table_name = tc.table_name " + "WHERE tc.table_schema = %s AND tc.table_name = %s " + "AND tc.constraint_type = 'PRIMARY KEY' ORDER BY kcu.ordinal_position", + (clean_schema, "fx_rates"), + ) + key_columns = [row[0] for row in cur.fetchall()] + conn.close() + assert rows == [ + (datetime.date(2026, 10, 8), "EUR", decimal.Decimal("1.090000")), + (datetime.date(2026, 10, 8), "GBP", decimal.Decimal("0.861000")), + (datetime.date(2026, 10, 9), "EUR", decimal.Decimal("1.100000")), + ] + assert key_columns == ["rate_date", "base"] + + +@pytest.mark.integration +def test_run_node_append_into_a_drifted_table_fails_naming_the_column(clean_schema, tmp_path, capsys): + _seed(clean_schema, "fx_rates", "rate_date date NOT NULL, base varchar(3) NOT NULL", "") + env = _fx_run_env(tmp_path, clean_schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append") + assert run_node(env, adapter=_adapter()) == 1 + out = capsys.readouterr().out + assert '"message":"LoadError: table runtime_it.fx_rates does not match' in out + assert "missing column(s) ['rate']" in out + assert _count(clean_schema, "fx_rates") == 0 diff --git a/adapters/trino/tests/test_integration_runtime_trino.py b/adapters/trino/tests/test_integration_runtime_trino.py index 1473994..533b0e4 100644 --- a/adapters/trino/tests/test_integration_runtime_trino.py +++ b/adapters/trino/tests/test_integration_runtime_trino.py @@ -10,6 +10,7 @@ deliberately raises. The in-repo test therefore selects the named ``trino`` entry point; runner images separately enforce the exactly-one-plugin invariant. """ +import datetime import decimal import os import uuid @@ -18,9 +19,11 @@ import pyarrow as pa import pytest +import yaml import trino +from continuo_python_runtime.harness import run_node from continuo_trino_adapter.adapter import TrinoAdapter TRINO_ENV = { @@ -368,3 +371,86 @@ def test_entry_point_resolves_to_this_adapter(): eps = [ep for ep in entry_points(group="continuo_engine.adapters") if ep.name == "trino"] assert len(eps) == 1 assert eps[0].load() is TrinoAdapter + + +# --- write modes end to end ---------------------------------------------------- + +_FX_COLUMNS = [ + {"name": "rate_date", "type": "DATE", "nullable": False}, + {"name": "base", "type": "VARCHAR(3)", "nullable": False}, + {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True}, +] + +_FX_SCRIPT = ( + "import datetime\n" + "from decimal import Decimal\n\n" + "import pyarrow as pa\n\n" + "ROWS = {rows!r}\n\n\n" + "def run(ctx):\n" + " return pa.table({{\n" + " 'rate_date': pa.array([datetime.date.fromisoformat(r[0]) for r in ROWS], pa.date32()),\n" + " 'base': pa.array([r[1] for r in ROWS], pa.string()),\n" + " 'rate': pa.array([Decimal(r[2]) for r in ROWS], pa.decimal128(12, 6)),\n" + " }})\n" +) + + +def _fx_run_env(repo, schema, run, rows, **mode): + """A python-api fx node returning *rows*, with a script file unique to *run*. + + One file per run: rewriting the same file within a second at the same size + would let Python reuse its cached bytecode and return the previous rows. + """ + (repo / "contracts").mkdir(exist_ok=True) + (repo / "scripts").mkdir(exist_ok=True) + script = f"scripts/fx_{run}.py" + (repo / script).write_text(_FX_SCRIPT.format(rows=rows)) + node = { + "schema": schema, "table": "fx_rates", "owner": "m", "schedule": "hourly", + "criticality": "SECONDARY", "kind": "python-api", "script": script, + "output_columns": _FX_COLUMNS, **mode, + } + (repo / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]})) + return { + "NODE_ID": f"python-api.svc.{schema}.fx_rates", "TABLE_NAME": "fx_rates", + "TARGET_SCHEMA": schema, "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo), + } + + +@pytest.mark.integration +def test_run_node_append_keeps_the_rows_of_earlier_runs(clean_schema, tmp_path): + env = _fx_run_env(tmp_path, clean_schema, "run1", [("2026-10-08", "EUR", "1.081200")], mode_type="append") + assert run_node(env, adapter=_adapter()) == 0 + env = _fx_run_env( + tmp_path, clean_schema, "run2", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-09", "EUR", "1.090000")], + mode_type="append", + ) + assert run_node(env, adapter=_adapter()) == 0 + assert _count(clean_schema, "fx_rates") == 3 + + +@pytest.mark.integration +def test_run_node_upsert_overwrites_by_key(clean_schema, tmp_path): + key = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]} + env = _fx_run_env( + tmp_path, clean_schema, "run1", + [("2026-10-08", "EUR", "1.081200"), ("2026-10-08", "GBP", "0.861000")], **key, + ) + assert run_node(env, adapter=_adapter()) == 0 + env = _fx_run_env( + tmp_path, clean_schema, "run2", + [("2026-10-08", "EUR", "1.090000"), ("2026-10-09", "EUR", "1.100000")], **key, + ) + assert run_node(env, adapter=_adapter()) == 0 + a = _adapter() + rows = a._execute( + f'SELECT rate_date, base, rate FROM iceberg."{clean_schema}"."fx_rates" ' + "ORDER BY rate_date, base" + ) + a.close() + assert [tuple(row) for row in rows] == [ + (datetime.date(2026, 10, 8), "EUR", decimal.Decimal("1.090000")), + (datetime.date(2026, 10, 8), "GBP", decimal.Decimal("0.861000")), + (datetime.date(2026, 10, 9), "EUR", decimal.Decimal("1.100000")), + ] diff --git a/continuo_python_runtime/harness.py b/continuo_python_runtime/harness.py index 0d797a3..99dc4ea 100644 --- a/continuo_python_runtime/harness.py +++ b/continuo_python_runtime/harness.py @@ -3,7 +3,8 @@ ``run_node`` is the sole write sink for a python node. It resolves the node from the contract, loads and executes its script inside a :class:`~continuo_python_runtime.context.RunContext`, conforms the result to -the declared schema, and writes it through the runtime adapter. Exactly one +the declared schema, and writes it through the runtime adapter with the +node's ``mode_type``: ``load`` (replace), ``append`` or ``upsert``. Exactly one sentinel-framed result block is printed to stdout per run; every other diagnostic goes to stderr via ``logging``. """ @@ -17,11 +18,15 @@ import sys import uuid from collections.abc import Callable, Mapping +from dataclasses import dataclass from pathlib import Path from types import ModuleType from typing import Any +from continuo_engine_contract.keys import DuplicateKeyError # type: ignore[import-untyped] from continuo_engine_contract.port import ( # type: ignore[import-untyped] + AppendCapable, + UpsertCapable, discover_adapter, ) from continuo_engine_contract.result import result_block # type: ignore[import-untyped] @@ -29,11 +34,18 @@ from continuo_python_runtime.conform import conform, to_arrow from continuo_python_runtime.context import RunContext from continuo_python_runtime.contract.loader import load_contract_dir -from continuo_python_runtime.contract.model import DEFAULT_MODE_TYPE, Node +from continuo_python_runtime.contract.model import Node from continuo_python_runtime.contract.paths import resolve_script_path from continuo_python_runtime.csv_loader import produce_csv from continuo_python_runtime.csv_source import CsvSourceReader -from continuo_python_runtime.errors import ContractError, HarnessError, LoadError, ScriptError +from continuo_python_runtime.errors import ( + ConformError, + ContractError, + HarnessError, + LoadError, + ScriptError, +) +from continuo_python_runtime.shape import shape_diff logger = logging.getLogger("continuo_python_runtime.harness") @@ -232,6 +244,75 @@ def _produce_from_csv( } +def _write_replace(adapter: Any, schema: str, table: str, data: Any, key: tuple[str, ...]) -> None: + adapter.load(schema, table, data) + + +def _write_append(adapter: Any, schema: str, table: str, data: Any, key: tuple[str, ...]) -> None: + adapter.append(schema, table, data) + + +def _write_upsert(adapter: Any, schema: str, table: str, data: Any, key: tuple[str, ...]) -> None: + adapter.upsert(schema, table, data, key) + + +@dataclass(frozen=True) +class _Writer: + """How one ``mode_type`` is written. + + ``capability`` is the engine-contract interface the adapter must declare + (``None``: every ``WarehouseAdapter`` can, through ``load``). Every + capability extends ``TableIntrospection``, so a capability-gated write is + also a shape-checked one. + """ + + capability: type | None + write: Callable[[Any, str, str, Any, tuple[str, ...]], None] + + +# One entry per mode_type, beside _PRODUCERS: a new mode is one capability +# interface in the engine contract plus one entry here. +_WRITERS: dict[str, _Writer] = { + "replace": _Writer(capability=None, write=_write_replace), + "append": _Writer(capability=AppendCapable, write=_write_append), + "upsert": _Writer(capability=UpsertCapable, write=_write_upsert), +} + + +def _writer_for(adapter: Any, node: Node) -> _Writer: + """The node's writer, if the adapter declares the capability its mode needs. + + Called right after the config tripwire, before the script runs or the + warehouse is touched: an engine that cannot append fails in the first + second, not after the whole run. + + Raises: + LoadError: `` does not support ``. + """ + writer = _WRITERS[node.mode_type] + if writer.capability is not None and not isinstance(adapter, writer.capability): + raise LoadError(f"{type(adapter).__name__} does not support {node.mode_type}") + return writer + + +def _check_shape(adapter: Any, node: Node, schema: str, table: str) -> None: + """Fail unless the live target table still has the node's declared shape. + + Runs before an append or upsert only: those keep earlier rows, so a table + whose columns or nullability drifted (changed outside the contract, or + created before the release gate recorded shapes) must not be written. It + inspects the table actually written -- ``TARGET_SCHEMA.TABLE_NAME`` -- not + ``node.relation``. + + Raises: + LoadError: Naming every missing column, unexpected column and + nullability mismatch. + """ + diff = shape_diff(node.output_columns, adapter.table_columns(schema, table)) + if diff: + raise LoadError(diff.message(f"{schema}.{table}")) + + def run_node( env: Mapping[str, str], adapter: Any = None, reader: CsvSourceReader | None = None ) -> int: @@ -265,16 +346,6 @@ def run_node( nodes = load_contract_dir(contract_dir, check_reads=False) node = select_node(nodes, node_id) - # Interim: the harness writes every node with load() (replace) until the - # per-mode write dispatch lands. Until then, reject append/upsert before - # the script runs or anything is written, so an accepted contract cannot - # silently replace a table it meant to append to or upsert into. - if node.mode_type != DEFAULT_MODE_TYPE: - raise LoadError( - f"write mode {node.mode_type!r} is not supported by the harness yet " - f"(node {target_schema}.{table_name}); only {DEFAULT_MODE_TYPE!r} is available" - ) - if adapter is not None: active_adapter = adapter else: @@ -286,6 +357,7 @@ def run_node( raise LoadError(f"adapter construction failed: {exc}") from exc _validate_config_early(active_adapter, node) + writer = _writer_for(active_adapter, node) table = _PRODUCERS[node.kind](node, active_adapter, app_root, reader) @@ -295,11 +367,25 @@ def run_node( {"name": c.name, "type": c.type, "nullable": c.nullable} for c in node.output_columns ] + # Spelled exactly as in output_columns (the loader enforces it). Empty + # for every mode but upsert, so primary_key reaches ensure_table only + # for upsert nodes. + key = node.primary_key try: - active_adapter.ensure_table(target_schema, table_name, columns, config=node.config) - active_adapter.load(target_schema, table_name, conformed) + if key: + active_adapter.ensure_table( + target_schema, table_name, columns, config=node.config, primary_key=key + ) + else: + active_adapter.ensure_table(target_schema, table_name, columns, config=node.config) + if writer.capability is not None: + _check_shape(active_adapter, node, target_schema, table_name) + writer.write(active_adapter, target_schema, table_name, conformed, key) except HarnessError: raise + except DuplicateKeyError as exc: + # The script's rows repeat a key: its output is at fault, not the warehouse. + raise ConformError(str(exc)) from exc except Exception as exc: raise LoadError(f"failed to write {target_schema}.{table_name}: {exc}") from exc diff --git a/continuo_python_runtime/shape.py b/continuo_python_runtime/shape.py new file mode 100644 index 0000000..c8c3d66 --- /dev/null +++ b/continuo_python_runtime/shape.py @@ -0,0 +1,95 @@ +"""Compare a live table's columns with a node's declared output columns. + +The harness runs this before every append or upsert: those modes keep earlier +rows, so writing into a table whose shape drifted from the contract would mix +two shapes in one table. Names are compared case-insensitively (Trino folds +names, DuckDB matches them case-insensitively) and nullability exactly. Order +is ignored, because every insert names its columns; types are ignored, +because engines report them in their own spelling (Trino and DuckDB drop a +VARCHAR's length) and there is no engine-to-contract type mapping. Type +changes are the release gate's job. +""" +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from typing import NamedTuple + +from continuo_engine_contract.port import LiveColumn # type: ignore[import-untyped] + +from continuo_python_runtime.contract.model import Column + + +class NullabilityDiff(NamedTuple): + """A column whose nullability differs: as declared vs as the live table has it.""" + + name: str + declared: bool + live: bool + + +def _yaml_bool(value: bool) -> str: + return "true" if value else "false" + + +@dataclass(frozen=True) +class ShapeDiff: + """What separates a live table from the declared columns; falsy when nothing does.""" + + missing: tuple[str, ...] = () + unexpected: tuple[str, ...] = () + nullability: tuple[NullabilityDiff, ...] = () + + def __bool__(self) -> bool: + return bool(self.missing or self.unexpected or self.nullability) + + def message(self, target: str) -> str: + """One line naming every difference, for a LoadError about *target* (``schema.table``).""" + parts: list[str] = [] + if self.missing: + parts.append(f"missing column(s) {list(self.missing)}") + if self.unexpected: + parts.append(f"unexpected column(s) {list(self.unexpected)}") + parts.extend( + f"column {diff.name} is nullable: {_yaml_bool(diff.live)} in the table " + f"but nullable: {_yaml_bool(diff.declared)} in the contract" + for diff in self.nullability + ) + return ( + f"table {target} does not match the contract's output_columns: {'; '.join(parts)}. " + "An append or upsert table keeps its rows, so its shape cannot change: " + "write to a new table (a _v2 name) under a new contract" + ) + + +def shape_diff( + contract_columns: Sequence[Column], live_columns: Sequence[LiveColumn] +) -> ShapeDiff: + """Compare the declared columns with a live table's. + + Each declared column is paired with a live column of the exact same name + if there is one, else with one whose name matches case-insensitively, so + two live columns that differ only in case pair with the right declared + column. + """ + remaining = list(live_columns) + missing: list[str] = [] + nullability: list[NullabilityDiff] = [] + for declared in contract_columns: + live = next((column for column in remaining if column.name == declared.name), None) + if live is None: + folded = declared.name.casefold() + live = next( + (column for column in remaining if column.name.casefold() == folded), None + ) + if live is None: + missing.append(declared.name) + continue + remaining.remove(live) + if live.nullable != declared.nullable: + nullability.append(NullabilityDiff(declared.name, declared.nullable, live.nullable)) + return ShapeDiff( + missing=tuple(missing), + unexpected=tuple(column.name for column in remaining), + nullability=tuple(nullability), + ) diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md index 2bfe6af..5501bf8 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -349,7 +349,15 @@ executor runs it as a Kubernetes Job: - **Sole write sink**: user scripts return a dataframe; the harness runs `conform()` (strict Arrow cast per §7.1), then calls `WarehouseAdapter.ensure_table` — applying the node's `config` physical - layout on create, see §13.1 — and performs the only INSERT. Scripts get + layout on create, see §13.1, and for an `upsert` node its + `primary_key` — and performs the only write, chosen by `mode_type`: `load` + (replace), `append` or `upsert`. `append` and `upsert` need the adapter to + declare `AppendCapable` / `UpsertCapable`; the harness checks that right after + selecting the node and fails with `LoadError` otherwise. Before an `append` or + `upsert` it compares the live target table (`TARGET_SCHEMA.TABLE_NAME`) with + `output_columns` — names case-insensitively, nullability exactly, order and + types ignored — and fails with `LoadError` on any difference. An `upsert` whose rows + repeat a key fails with `ConformError` and writes nothing. Scripts get a read-only connection surface. The harness calls `ensure_table(..., config=...)` as a keyword unconditionally, and the abstract `WarehouseAdapter` port in `continuo-engine-contract` declares it: diff --git a/tests/conftest.py b/tests/conftest.py index 4575eaf..310d1ca 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -11,6 +11,7 @@ import pytest import yaml +from continuo_engine_contract.port import AppendCapable, LiveColumn, UpsertCapable from continuo_python_runtime.contract.model import Column, Node @@ -133,6 +134,53 @@ def close(self): self.closed = True +class FakeAppendAdapter(FakeWarehouseAdapter, AppendCapable): + """FakeWarehouseAdapter that declares AppendCapable. + + ``live_columns`` is what table_columns reports; None reports exactly the + columns ensure_table received (a table the harness just created). + """ + + def __init__(self, tables=None, live_columns=None): + super().__init__(tables) + self.live_columns = live_columns + self.introspected = None + self.appended = None + + def table_columns(self, schema, table): + self.introspected = (schema, table) + if self.live_columns is not None: + return list(self.live_columns) + return [LiveColumn(c["name"], c["nullable"]) for c in self.ensured[2]] + + def append(self, schema, table, data): + self.appended = (schema, table, data) + + +class FakeUpsertAdapter(FakeWarehouseAdapter, UpsertCapable): + """FakeWarehouseAdapter that declares UpsertCapable; upsert() is the port's own template method.""" + + def __init__(self, tables=None, live_columns=None): + super().__init__(tables) + self.live_columns = live_columns + self.introspected = None + self.ensured_primary_key = None + self.upserted = None + + def ensure_table(self, schema, table, columns, *, config, primary_key=()): + super().ensure_table(schema, table, columns, config=config) + self.ensured_primary_key = tuple(primary_key) + + def table_columns(self, schema, table): + self.introspected = (schema, table) + if self.live_columns is not None: + return list(self.live_columns) + return [LiveColumn(c["name"], c["nullable"]) for c in self.ensured[2]] + + def _upsert(self, schema, table, data, primary_key): + self.upserted = (schema, table, data, tuple(primary_key)) + + @pytest.fixture def harness_repo(tmp_path): """Create a minimal contract repository with scripts and YAML files.""" diff --git a/tests/test_harness.py b/tests/test_harness.py index e952115..0c4a71a 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -715,42 +715,3 @@ def test_python_api_missing_env_key_is_a_script_error_naming_the_key(tmp_path, c assert run_node(_api_env(repo), adapter=FakeWarehouseAdapter()) == 1 out = capsys.readouterr().out assert "ScriptError" in out and "FX_API_KEY" in out - - -# --- non-replace write modes are rejected until the per-mode dispatch lands --- - - -def _repo_with_mode(tmp_path, marker, **mode_fields): - repo = _repo_with_config(tmp_path, {}, marker) - doc = yaml.safe_load((repo / "contracts" / "t.yml").read_text()) - del doc["nodes"][0]["config"] - doc["nodes"][0].update(mode_fields) - (repo / "contracts" / "t.yml").write_text(yaml.safe_dump(doc)) - return repo - - -@pytest.mark.parametrize( - "mode_fields", - [ - {"mode_type": "append"}, - {"mode_type": "upsert", "primary_key": ["id"]}, - ], - ids=["append", "upsert"], -) -def test_run_node_rejects_non_replace_modes_before_writing_anything( - tmp_path, capsys, mode_fields -): - """load() replaces the table, so an accepted append/upsert contract must not reach it.""" - marker = tmp_path / "script-ran" - repo = _repo_with_mode(tmp_path, marker, **mode_fields) - ad = FakeWarehouseAdapter({"select id from analytics.a": pa.table({"id": [1]})}) - assert run_node(_env(repo), adapter=ad) == 1 - out = capsys.readouterr().out - assert out.count("===CONTINUO_VALIDATION_RESULT_BEGIN===") == 1 - body = _sentinel_body(out) - assert body["status"] == "error" - assert body["message"].startswith("LoadError:") - assert "not supported by the harness yet" in body["message"] - assert mode_fields["mode_type"] in body["message"] - assert not marker.exists(), "script ran despite an unsupported write mode" - assert ad.ensured is None and ad.loaded is None diff --git a/tests/test_harness_write_modes.py b/tests/test_harness_write_modes.py new file mode 100644 index 0000000..046e3a1 --- /dev/null +++ b/tests/test_harness_write_modes.py @@ -0,0 +1,259 @@ +"""Harness write modes: writer dispatch, capability check and error mapping.""" +import json + +import yaml + +from continuo_engine_contract.port import LiveColumn, WarehouseAdapter + +from continuo_python_runtime.harness import run_node +from tests.conftest import FakeAppendAdapter, FakeUpsertAdapter, FakeWarehouseAdapter + +FX_COLUMNS = [ + {"name": "rate_date", "type": "DATE", "nullable": False}, + {"name": "base", "type": "VARCHAR(3)", "nullable": False}, + {"name": "rate", "type": "DECIMAL(12,6)", "nullable": True}, +] + + +def _script(dates: str, bases: str, prelude: str = "") -> str: + return ( + "import datetime\n" + "import pathlib\n" + "from decimal import Decimal\n\n" + "import pyarrow as pa\n\n\n" + "def run(ctx):\n" + f"{prelude}" + " return pa.table({\n" + f" 'rate_date': pa.array({dates}, pa.date32()),\n" + f" 'base': {bases},\n" + " 'rate': pa.array([Decimal('1.081200'), Decimal('0.861000')], pa.decimal128(12, 6)),\n" + " })\n" + ) + + +SAME_DAY = "[datetime.date(2026, 10, 8), datetime.date(2026, 10, 8)]" +TWO_KEYS = _script(SAME_DAY, "['EUR', 'GBP']") +REPEATED_KEY = _script(SAME_DAY, "['EUR', 'EUR']") +NULL_KEY = _script("[None, datetime.date(2026, 10, 8)]", "['EUR', 'GBP']") +UPSERT = {"mode_type": "upsert", "primary_key": ["rate_date", "base"]} + + +def _repo(tmp_path, script=TWO_KEYS, **node_extra): + tmp_path.mkdir(parents=True, exist_ok=True) + (tmp_path / "contracts").mkdir() + (tmp_path / "scripts").mkdir() + (tmp_path / "scripts" / "fx.py").write_text(script) + node = { + "schema": "fx", "table": "fx_rates", "owner": "m", "schedule": "hourly", + "criticality": "SECONDARY", "kind": "python-api", "script": "scripts/fx.py", + "output_columns": FX_COLUMNS, **node_extra, + } + (tmp_path / "contracts" / "fx.yml").write_text(yaml.safe_dump({"nodes": [node]})) + return tmp_path + + +def _env(repo, **over): + return { + "NODE_ID": "python-api.svc.fx.fx_rates", "TABLE_NAME": "fx_rates", + "TARGET_SCHEMA": "fx", "CONTRACT_DIR": str(repo / "contracts"), "APP_ROOT": str(repo), + **over, + } + + +def _message(out: str) -> str: + return json.loads(out.split("BEGIN===\n")[1].split("\n===CONTINUO")[0])["message"] + + +def test_a_replace_node_still_writes_through_load(tmp_path): + adapter = FakeWarehouseAdapter() + assert run_node(_env(_repo(tmp_path)), adapter=adapter) == 0 + assert adapter.loaded[:2] == ("fx", "fx_rates") + assert adapter.loaded[2].num_rows == 2 + + +def test_an_append_node_appends_and_never_loads(tmp_path): + adapter = FakeAppendAdapter() + assert run_node(_env(_repo(tmp_path, mode_type="append")), adapter=adapter) == 0 + assert adapter.appended[:2] == ("fx", "fx_rates") + assert adapter.appended[2].num_rows == 2 + assert adapter.loaded is None + + +def test_an_upsert_node_passes_its_key_to_ensure_table_and_upsert(tmp_path): + adapter = FakeUpsertAdapter() + assert run_node(_env(_repo(tmp_path, **UPSERT)), adapter=adapter) == 0 + assert adapter.ensured_primary_key == ("rate_date", "base") + assert adapter.upserted[:2] == ("fx", "fx_rates") + assert adapter.upserted[3] == ("rate_date", "base") + assert adapter.loaded is None + + +def test_append_without_the_capability_fails_before_the_script_runs(tmp_path, capsys): + marker = tmp_path / "ran" + script = _script( + SAME_DAY, "['EUR', 'GBP']", prelude=f" pathlib.Path({str(marker)!r}).write_text('ran')\n" + ) + adapter = FakeWarehouseAdapter() + assert run_node(_env(_repo(tmp_path, script=script, mode_type="append")), adapter=adapter) == 1 + assert _message(capsys.readouterr().out) == "LoadError: FakeWarehouseAdapter does not support append" + assert not marker.exists() + assert adapter.ensured is None and adapter.loaded is None + + +def test_upsert_on_an_append_only_adapter_is_a_load_error(tmp_path, capsys): + adapter = FakeAppendAdapter() + assert run_node(_env(_repo(tmp_path, **UPSERT)), adapter=adapter) == 1 + assert _message(capsys.readouterr().out) == "LoadError: FakeAppendAdapter does not support upsert" + assert adapter.ensured is None and adapter.appended is None + + +class _PortOnlyAdapter(WarehouseAdapter): + """A bring-your-own adapter that implements the port and no capability.""" + + def __init__(self): + self.calls = [] + + @classmethod + def required_env(cls): + return [] + + @classmethod + def from_env(cls): + return cls() + + def ensure_schema(self, schema): + self.calls.append("ensure_schema") + + def drop_schema(self, schema): + self.calls.append("drop_schema") + + def build_empty_from_sql(self, schema, table, compiled_sql): + self.calls.append("build_empty_from_sql") + + def clone_empty_from_prod(self, candidate_schema, prod_schema, table): + self.calls.append("clone_empty_from_prod") + + def build_empty_from_columns(self, schema, table, columns, config): + self.calls.append("build_empty_from_columns") + + def check_binds(self, sql): + self.calls.append("check_binds") + + def fetch(self, sql): + self.calls.append("fetch") + + def ensure_table(self, schema, table, columns, *, config): + self.calls.append("ensure_table") + + def load(self, schema, table, data): + self.calls.append("load") + + def close(self): + self.calls.append("close") + + +def test_a_port_only_byo_adapter_still_replaces_and_cannot_append(tmp_path, capsys): + replacing = _PortOnlyAdapter() + assert run_node(_env(_repo(tmp_path / "a")), adapter=replacing) == 0 + assert replacing.calls == ["ensure_table", "load", "close"] + + appending = _PortOnlyAdapter() + capsys.readouterr() + assert run_node(_env(_repo(tmp_path / "b", mode_type="append")), adapter=appending) == 1 + assert _message(capsys.readouterr().out) == "LoadError: _PortOnlyAdapter does not support append" + assert appending.calls == ["close"] + + +def test_a_repeated_key_is_a_conform_error_and_nothing_is_written(tmp_path, capsys): + adapter = FakeUpsertAdapter() + assert run_node(_env(_repo(tmp_path, script=REPEATED_KEY, **UPSERT)), adapter=adapter) == 1 + assert _message(capsys.readouterr().out).startswith( + "ConformError: primary key (rate_date, base) is not unique in the rows to write: " + "(2026-10-08, EUR) appears 2 times" + ) + assert adapter.upserted is None + + +def test_a_null_key_is_a_conform_error_before_ensure_table(tmp_path, capsys): + adapter = FakeUpsertAdapter() + assert run_node(_env(_repo(tmp_path, script=NULL_KEY, **UPSERT)), adapter=adapter) == 1 + assert _message(capsys.readouterr().out) == ( + "ConformError: column rate_date is declared nullable: false but contains nulls" + ) + assert adapter.ensured is None and adapter.upserted is None + + +def test_an_engine_error_during_upsert_is_a_load_error(tmp_path, capsys): + class Deadlocking(FakeUpsertAdapter): + def _upsert(self, schema, table, data, primary_key): + raise RuntimeError("deadlock detected") + + assert run_node(_env(_repo(tmp_path, **UPSERT)), adapter=Deadlocking()) == 1 + assert _message(capsys.readouterr().out) == "LoadError: failed to write fx.fx_rates: deadlock detected" + + +def test_a_value_error_from_the_engine_is_not_taken_for_a_repeated_key(tmp_path, capsys): + class Overflowing(FakeUpsertAdapter): + def _upsert(self, schema, table, data, primary_key): + raise ValueError("numeric field overflow") + + assert run_node(_env(_repo(tmp_path, **UPSERT)), adapter=Overflowing()) == 1 + assert _message(capsys.readouterr().out) == ( + "LoadError: failed to write fx.fx_rates: numeric field overflow" + ) + + +def test_append_into_a_table_missing_a_declared_column_fails_naming_it(tmp_path, capsys): + adapter = FakeAppendAdapter(live_columns=[LiveColumn("rate_date", False), LiveColumn("base", False)]) + assert run_node(_env(_repo(tmp_path, mode_type="append")), adapter=adapter) == 1 + assert _message(capsys.readouterr().out) == ( + "LoadError: table fx.fx_rates does not match the contract's output_columns: " + "missing column(s) ['rate']. An append or upsert table keeps its rows, so its shape " + "cannot change: write to a new table (a _v2 name) under a new contract" + ) + assert adapter.appended is None + + +def test_upsert_reports_every_shape_difference_at_once(tmp_path, capsys): + adapter = FakeUpsertAdapter(live_columns=[ + LiveColumn("rate_date", True), LiveColumn("base", False), + LiveColumn("rate", True), LiveColumn("old_rate", True), + ]) + assert run_node(_env(_repo(tmp_path, **UPSERT)), adapter=adapter) == 1 + message = _message(capsys.readouterr().out) + assert "unexpected column(s) ['old_rate']" in message + assert "column rate_date is nullable: true in the table but nullable: false in the contract" in message + assert adapter.upserted is None + + +def test_the_shape_check_inspects_the_target_table_not_the_contract_relation(tmp_path, capsys): + adapter = FakeAppendAdapter() + env = _env(_repo(tmp_path, mode_type="append"), TARGET_SCHEMA="candidate_fx") + assert run_node(env, adapter=adapter) == 0 + assert adapter.introspected == ("candidate_fx", "fx_rates") + + +def test_a_matching_shape_in_another_case_and_order_is_written(tmp_path, capsys): + adapter = FakeAppendAdapter(live_columns=[ + LiveColumn("RATE", True), LiveColumn("Base", False), LiveColumn("RATE_DATE", False), + ]) + assert run_node(_env(_repo(tmp_path, mode_type="append")), adapter=adapter) == 0 + assert adapter.appended is not None + + +def test_replace_never_inspects_the_table(tmp_path, capsys): + adapter = FakeAppendAdapter(live_columns=[]) + assert run_node(_env(_repo(tmp_path)), adapter=adapter) == 0 + assert adapter.introspected is None + assert adapter.loaded is not None + + +def test_a_table_columns_failure_is_a_load_error(tmp_path, capsys): + class Vanished(FakeAppendAdapter): + def table_columns(self, schema, table): + raise LookupError(f"table {schema}.{table} does not exist") + + assert run_node(_env(_repo(tmp_path, mode_type="append")), adapter=Vanished()) == 1 + assert _message(capsys.readouterr().out) == ( + "LoadError: failed to write fx.fx_rates: table fx.fx_rates does not exist" + ) diff --git a/tests/test_kind_registries.py b/tests/test_kind_registries.py index 1039955..d4f0aa8 100644 --- a/tests/test_kind_registries.py +++ b/tests/test_kind_registries.py @@ -7,7 +7,7 @@ from continuo_python_runtime import harness from continuo_python_runtime.contract import merge from continuo_python_runtime.contract.kinds import RULES -from continuo_python_runtime.contract.model import KINDS +from continuo_python_runtime.contract.model import KINDS, MODE_TYPES def test_every_kind_has_rules(): @@ -20,3 +20,7 @@ def test_every_kind_has_hash_inputs(): def test_every_kind_has_a_producer(): assert set(harness._PRODUCERS) == KINDS + + +def test_every_mode_has_a_writer(): + assert set(harness._WRITERS) == MODE_TYPES diff --git a/tests/test_shape.py b/tests/test_shape.py new file mode 100644 index 0000000..cf31433 --- /dev/null +++ b/tests/test_shape.py @@ -0,0 +1,60 @@ +"""shape_diff: the live table against the declared output columns.""" +from continuo_engine_contract.port import LiveColumn + +from continuo_python_runtime.contract.model import Column +from continuo_python_runtime.shape import NullabilityDiff, ShapeDiff, shape_diff + +DECLARED = ( + Column("rate_date", "DATE", nullable=False), + Column("base", "VARCHAR(3)", nullable=False), + Column("rate", "DECIMAL(12,6)"), +) + + +def test_a_matching_table_has_no_difference(): + diff = shape_diff( + DECLARED, [LiveColumn("rate_date", False), LiveColumn("base", False), LiveColumn("rate", True)] + ) + assert not diff + assert diff == ShapeDiff() + + +def test_column_order_is_ignored(): + live = [LiveColumn("rate", True), LiveColumn("base", False), LiveColumn("rate_date", False)] + assert not shape_diff(DECLARED, live) + + +def test_names_compare_case_insensitively(): + live = [LiveColumn("RATE_DATE", False), LiveColumn("Base", False), LiveColumn("rate", True)] + assert not shape_diff(DECLARED, live) + + +def test_missing_unexpected_and_nullability_are_all_reported(): + diff = shape_diff( + DECLARED, + [LiveColumn("rate_date", True), LiveColumn("base", False), LiveColumn("old_rate", True)], + ) + assert diff.missing == ("rate",) + assert diff.unexpected == ("old_rate",) + assert diff.nullability == (NullabilityDiff("rate_date", declared=False, live=True),) + assert diff + + +def test_the_exact_spelling_is_paired_first(): + declared = (Column("Rate", "BIGINT", nullable=False), Column("rate", "BIGINT")) + assert not shape_diff(declared, [LiveColumn("rate", True), LiveColumn("Rate", False)]) + + +def test_the_message_names_every_difference_and_the_remedy(): + diff = ShapeDiff( + missing=("rate",), + unexpected=("old_rate",), + nullability=(NullabilityDiff("rate_date", declared=False, live=True),), + ) + assert diff.message("fx.fx_rates") == ( + "table fx.fx_rates does not match the contract's output_columns: " + "missing column(s) ['rate']; unexpected column(s) ['old_rate']; " + "column rate_date is nullable: true in the table but nullable: false in the contract. " + "An append or upsert table keeps its rows, so its shape cannot change: " + "write to a new table (a _v2 name) under a new contract" + )