From 7c9bbf64dd013b9c416ccac7264feb2eb8885271 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:10:55 +0200 Subject: [PATCH 1/6] feat(harness): shape_diff between a live table and output_columns A pure comparison: names case-insensitively (exact spelling paired first), nullability exactly, order and types ignored. ShapeDiff lists missing and unexpected columns and nullability mismatches, and renders one message naming all of them and the remedy. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- continuo_python_runtime/shape.py | 95 ++++++++++++++++++++++++++++++++ tests/test_shape.py | 60 ++++++++++++++++++++ 2 files changed, 155 insertions(+) create mode 100644 continuo_python_runtime/shape.py create mode 100644 tests/test_shape.py 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/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" + ) From 93db2ec73a23fc1b068b42025479253c14a5e137 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:14:28 +0200 Subject: [PATCH 2/6] feat(harness): write each node with its mode_type _WRITERS maps replace, append and upsert to the capability they need and the write call. The capability is checked right after the config tripwire, so an engine that cannot append fails before the script runs. The upsert key goes to ensure_table and upsert as written; DuplicateKeyError becomes ConformError, any other write failure stays LoadError. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 6 + README.md | 55 +++++++- continuo_python_runtime/harness.py | 97 +++++++++++--- docs/boundary-contract.md | 7 +- tests/conftest.py | 48 +++++++ tests/test_harness.py | 39 ------ tests/test_harness_write_modes.py | 203 +++++++++++++++++++++++++++++ tests/test_kind_registries.py | 6 +- 8 files changed, 401 insertions(+), 60 deletions(-) create mode 100644 tests/test_harness_write_modes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 87fa621..8e27fa0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,12 @@ 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. See "Write modes" in the README. ### Changed diff --git a/README.md b/README.md index 5b8ddd1..cfda8b0 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,54 @@ 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. +- 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 +310,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`, 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/continuo_python_runtime/harness.py b/continuo_python_runtime/harness.py index 0d797a3..010ec76 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,17 @@ 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, +) logger = logging.getLogger("continuo_python_runtime.harness") @@ -232,6 +243,57 @@ 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 run_node( env: Mapping[str, str], adapter: Any = None, reader: CsvSourceReader | None = None ) -> int: @@ -265,16 +327,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,8 +338,9 @@ 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) + table =_PRODUCERS[node.kind](node, active_adapter, app_root, reader) conformed = conform(table, node.output_columns, node.extra_columns) @@ -295,11 +348,23 @@ 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) + 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/docs/boundary-contract.md b/docs/boundary-contract.md index 2bfe6af..5515be2 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -349,7 +349,12 @@ 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. 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..e8e50c9 --- /dev/null +++ b/tests/test_harness_write_modes.py @@ -0,0 +1,203 @@ +"""Harness write modes: writer dispatch, capability check and error mapping.""" +import json + +import yaml + +from continuo_engine_contract.port import 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" + ) 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 From d1cfeadabd4db91b3f53435fdb20d7546b583cfc Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:18:13 +0200 Subject: [PATCH 3/6] feat(harness): check the live table's shape before append and upsert After ensure_table, an append or upsert compares the target table's columns with output_columns (names case-insensitive, nullability exact, order and types ignored) and fails with one LoadError naming every difference. It inspects TARGET_SCHEMA.TABLE_NAME, the table actually written. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 4 ++- README.md | 10 +++++- continuo_python_runtime/harness.py | 23 +++++++++++- docs/boundary-contract.md | 5 ++- tests/test_harness_write_modes.py | 58 +++++++++++++++++++++++++++++- 5 files changed, 95 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e27fa0..87e2d13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,7 +50,9 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `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. See "Write modes" in the README. + 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 cfda8b0..35d242b 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,14 @@ produced, never how it is checked or written. 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 @@ -311,7 +319,7 @@ text. | `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, 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`, or the DDL/INSERT failed at the warehouse during the write. | Warehouse connectivity/permissions, the engine image, or the target table. | +| `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/continuo_python_runtime/harness.py b/continuo_python_runtime/harness.py index 010ec76..99dc4ea 100644 --- a/continuo_python_runtime/harness.py +++ b/continuo_python_runtime/harness.py @@ -45,6 +45,7 @@ LoadError, ScriptError, ) +from continuo_python_runtime.shape import shape_diff logger = logging.getLogger("continuo_python_runtime.harness") @@ -294,6 +295,24 @@ def _writer_for(adapter: Any, node: Node) -> _Writer: 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: @@ -340,7 +359,7 @@ def run_node( _validate_config_early(active_adapter, node) writer = _writer_for(active_adapter, node) - table =_PRODUCERS[node.kind](node, active_adapter, app_root, reader) + table = _PRODUCERS[node.kind](node, active_adapter, app_root, reader) conformed = conform(table, node.output_columns, node.extra_columns) @@ -359,6 +378,8 @@ def run_node( ) 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 diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md index 5515be2..5501bf8 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -353,7 +353,10 @@ executor runs it as a Kubernetes Job: `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. An `upsert` whose rows + 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 diff --git a/tests/test_harness_write_modes.py b/tests/test_harness_write_modes.py index e8e50c9..046e3a1 100644 --- a/tests/test_harness_write_modes.py +++ b/tests/test_harness_write_modes.py @@ -3,7 +3,7 @@ import yaml -from continuo_engine_contract.port import WarehouseAdapter +from continuo_engine_contract.port import LiveColumn, WarehouseAdapter from continuo_python_runtime.harness import run_node from tests.conftest import FakeAppendAdapter, FakeUpsertAdapter, FakeWarehouseAdapter @@ -201,3 +201,59 @@ def _upsert(self, schema, table, data, primary_key): 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" + ) From 5038f9df2d7c92f1b5ec89257f15d910dddf24f3 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:20:18 +0200 Subject: [PATCH 4/6] test(postgres): append and upsert end to end through run_node Two append runs keep both runs' rows; two upsert runs overwrite by (rate_date, base) and leave the PRIMARY KEY in place; an append into a table missing a declared column fails with LoadError naming it and writes nothing. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../test_integration_runtime_postgres.py | 108 ++++++++++++++++++ 1 file changed, 108 insertions(+) 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 From bede12347082e05dd3d1113df7ef0c12431e66b1 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:22:45 +0200 Subject: [PATCH 5/6] test(duckdb): append and upsert end to end through run_node Two append runs keep both runs' rows; two upsert runs overwrite by (rate_date, base) on the DuckLake stack. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../tests/test_integration_duckdb_runtime.py | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) 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")}, + ] From f37b28c40ec7ed614ce2c1453ef22223935ca0da Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 18:25:09 +0200 Subject: [PATCH 6/6] test(trino): append and upsert end to end through run_node Two append runs keep both runs' rows; two upsert runs overwrite by (rate_date, base) on the Trino plus Iceberg stack. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../tests/test_integration_runtime_trino.py | 86 +++++++++++++++++++ 1 file changed, 86 insertions(+) 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")), + ]