From d9915510630b3e3ba75f26cbb53f1ace8ef0d230 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 14:12:49 +0200 Subject: [PATCH 1/6] feat(engine-contract): ensure_unique_key and DuplicateKeyError Groups an Arrow table by its primary key and raises DuplicateKeyError (a ValueError) naming the first repeated keys and their counts. Upsert calls it before any engine sees the rows, because no engine handles repeated keys in one write consistently. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- contract/continuo_engine_contract/keys.py | 74 +++++++++++++++++++++++ contract/tests/test_contract_keys.py | 68 +++++++++++++++++++++ 2 files changed, 142 insertions(+) create mode 100644 contract/continuo_engine_contract/keys.py create mode 100644 contract/tests/test_contract_keys.py diff --git a/contract/continuo_engine_contract/keys.py b/contract/continuo_engine_contract/keys.py new file mode 100644 index 0000000..7543ea0 --- /dev/null +++ b/contract/continuo_engine_contract/keys.py @@ -0,0 +1,74 @@ +"""The primary-key invariant every upsert relies on, checked once for every engine. + +An upsert whose own rows repeat a key cannot be written consistently: DuckLake's +MERGE inserts both rows of a repeated new key and keeps the last row of a +repeated existing one, Trino fails only when two rows match one existing row, +and postgres's ON CONFLICT fails within one statement but keeps the last row +across statements. ``UpsertCapable.upsert`` therefore calls +:func:`ensure_unique_key` before any engine sees the rows. +""" +from __future__ import annotations + +from collections.abc import Sequence +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + import pyarrow # type: ignore[import-untyped] + +# How many repeated keys the error message names before summarizing the rest. +_SHOWN_KEYS = 5 + + +class DuplicateKeyError(ValueError): + """The rows to upsert repeat a primary key; nothing was written.""" + + +def ensure_unique_key(data: "pyarrow.Table", primary_key: Sequence[str]) -> None: + """Raise :class:`DuplicateKeyError` if two rows of *data* share a *primary_key* value. + + Groups the rows by the key columns. NULL counts as a value here; the + contract requires every key column to be ``nullable: false`` and the + harness's ``conform()`` rejects NULLs in such columns before this runs. + The message names the first repeated keys, in order of first appearance, + with their row counts. Only methods of *data* are called, so this module + needs no pyarrow import. + + Raises: + ValueError: If *primary_key* is empty or names a column *data* lacks. + DuplicateKeyError: If a key value appears in more than one row. + """ + key = list(primary_key) + if not key: + raise ValueError("primary_key must name at least one column") + missing = [name for name in key if name not in data.column_names] + if missing: + raise ValueError( + f"primary key column(s) {missing} are not in the data's columns {data.column_names}" + ) + if data.num_rows < 2: + return + # Positional names: a key column called "count_all" must not collide with + # the aggregate's own output column. + positional = [f"k{index}" for index in range(len(key))] + counts = ( + data.select(key) + .rename_columns(positional) + .group_by(positional, use_threads=False) + .aggregate([([], "count_all")]) + ) + if counts.num_rows == data.num_rows: + return + repeated = [row for row in counts.to_pylist() if row["count_all"] > 1] + shown = "; ".join( + f"({', '.join(str(row[name]) for name in positional)}) appears {row['count_all']} times" + for row in repeated[:_SHOWN_KEYS] + ) + more = ( + f"; and {len(repeated) - _SHOWN_KEYS} more repeated key(s)" + if len(repeated) > _SHOWN_KEYS + else "" + ) + raise DuplicateKeyError( + f"primary key ({', '.join(key)}) is not unique in the rows to write: " + f"{shown}{more}; nothing was written" + ) diff --git a/contract/tests/test_contract_keys.py b/contract/tests/test_contract_keys.py new file mode 100644 index 0000000..72f0e44 --- /dev/null +++ b/contract/tests/test_contract_keys.py @@ -0,0 +1,68 @@ +"""ensure_unique_key: the primary-key invariant every upsert checks first.""" +import datetime +from decimal import Decimal + +import pyarrow as pa +import pytest + +from continuo_engine_contract.keys import DuplicateKeyError, ensure_unique_key + + +def test_duplicate_key_error_is_a_value_error(): + assert issubclass(DuplicateKeyError, ValueError) + + +def test_a_unique_single_column_key_passes(): + ensure_unique_key(pa.table({"id": [1, 2, 3], "v": ["a", "a", "a"]}), ["id"]) + + +def test_a_unique_composite_key_passes(): + ensure_unique_key(pa.table({"d": [1, 1, 2], "b": ["EUR", "USD", "EUR"]}), ("d", "b")) + + +def test_empty_and_single_row_tables_pass(): + ensure_unique_key(pa.table({"id": pa.array([], pa.int64())}), ["id"]) + ensure_unique_key(pa.table({"id": [1]}), ["id"]) + + +def test_a_repeated_key_raises_naming_the_key_and_its_count(): + data = pa.table({ + "rate_date": [datetime.date(2026, 10, 8)] * 2 + [datetime.date(2026, 10, 9)], + "base": ["EUR", "EUR", "EUR"], + "rate": pa.array([Decimal("1.1")] * 3, pa.decimal128(12, 6)), + }) + with pytest.raises( + DuplicateKeyError, + match=( + r"^primary key \(rate_date, base\) is not unique in the rows to write: " + r"\(2026-10-08, EUR\) appears 2 times; nothing was written$" + ), + ): + ensure_unique_key(data, ["rate_date", "base"]) + + +def test_repeated_keys_are_listed_in_order_of_first_appearance_and_capped_at_five(): + ids = [7, 7, 3, 3, 3] + [n for n in range(10, 16) for _ in range(2)] + with pytest.raises(DuplicateKeyError) as caught: + ensure_unique_key(pa.table({"id": ids}), ["id"]) + message = str(caught.value) + assert message.index("(7) appears 2 times") < message.index("(3) appears 3 times") + assert "(12) appears 2 times" in message + assert "(13) appears" not in message + assert "; and 3 more repeated key(s); nothing was written" in message + + +def test_a_key_column_named_like_the_aggregate_still_works(): + with pytest.raises(DuplicateKeyError, match=r"\(1\) appears 2 times"): + ensure_unique_key(pa.table({"count_all": [1, 1]}), ["count_all"]) + + +def test_an_empty_key_is_rejected(): + with pytest.raises(ValueError, match="at least one column"): + ensure_unique_key(pa.table({"id": [1]}), []) + + +def test_a_key_column_missing_from_the_data_is_rejected(): + with pytest.raises(ValueError, match=r"primary key column\(s\) \['quote'\]") as caught: + ensure_unique_key(pa.table({"base": ["EUR"]}), ["base", "quote"]) + assert type(caught.value) is ValueError From ed03da667c8a09176f4e1666176e4d6d68a9b665 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 14:16:04 +0200 Subject: [PATCH 2/6] feat(engine-contract): opt-in write capabilities and ensure_table primary_key TableIntrospection, AppendCapable and UpsertCapable are abstract interfaces an adapter declares by inheritance; LiveColumn is what table_columns reports. UpsertCapable.upsert is a template method that checks key uniqueness before the engine-specific _upsert. ensure_table gains an optional keyword-only primary_key that the harness passes only for upsert nodes. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 7 ++ contract/README.md | 11 +- contract/continuo_engine_contract/port.py | 100 ++++++++++++++++- contract/tests/test_port.py | 124 +++++++++++++++++++++- docs/boundary-contract.md | 5 +- 5 files changed, 239 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f8a708b..6bd2d1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,13 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `primary_key` (authored order) for `upsert`; both are part of `config_hash`. Entries of nodes that do not set them, and their hashes, are unchanged. +- `continuo-engine-contract`: opt-in write capabilities on top of + `WarehouseAdapter` — `TableIntrospection` (`table_columns`, returning + `LiveColumn(name, nullable)`; `LookupError` for a missing table), + `AppendCapable` (`append`) and `UpsertCapable` (`upsert`, which runs + `ensure_unique_key` and then the engine's `_upsert`). `ensure_table` takes an + optional keyword-only `primary_key`. `continuo_engine_contract.keys` holds + `ensure_unique_key` and `DuplicateKeyError(ValueError)`. ### Changed diff --git a/contract/README.md b/contract/README.md index 88be9a9..42766c7 100644 --- a/contract/README.md +++ b/contract/README.md @@ -1,11 +1,12 @@ # continuo-engine-contract The contract shared by the Continuo python runtime — the node harness and the -validation runner alike — and every engine adapter library. Five modules: +validation runner alike — and every engine adapter library. Six modules: | Module | What it holds | |---|---| -| `continuo_engine_contract.port` | The `WarehouseAdapter` port — one port covering both the data plane (`fetch`/`ensure_table`/`load`) and validation (engine-specific empty-table DDL) — plus entry-point discovery | +| `continuo_engine_contract.port` | The `WarehouseAdapter` port — one port covering both the data plane (`fetch`/`ensure_table`/`load`) and validation (engine-specific empty-table DDL) — the opt-in write capabilities (`TableIntrospection`, `AppendCapable`, `UpsertCapable`, with `LiveColumn`), plus entry-point discovery | +| `continuo_engine_contract.keys` | `ensure_unique_key` and `DuplicateKeyError`: the primary-key check `UpsertCapable.upsert` runs before any engine sees the rows | | `continuo_engine_contract.types` | The SQL column-type grammar that DDL-emitting methods validate against | | `continuo_engine_contract.sql` | `ensure_single_read`, the parse-backed gate `check_binds` runs before it lets a declared read near an engine | | `continuo_engine_contract.config` | `ensure_known_keys`, the fail-closed gate `build_empty_from_columns` runs before it lets a node's physical-layout `config` block reach DDL | @@ -40,6 +41,12 @@ implementations are required to run rather than reinvent: `sql.ensure_single_read` before letting a declared read reach an engine, and `config.ensure_known_keys` before building any DDL from a node's physical-layout `config` block, to reject keys this engine doesn't recognize. +Three capability interfaces are opt-in on top of the port. An adapter that +also inherits `AppendCapable` (`table_columns`, `append`) or `UpsertCapable` +(`table_columns`, `_upsert`; the port's own `upsert` checks key uniqueness +first) can run `append` / `upsert` nodes; an adapter that inherits neither +still runs every `replace` node. `ensure_table` takes an optional +keyword-only `primary_key`, passed only for upsert nodes. Because the port *is* the contract, this package is very unlikely to need changes — new warehouse support is a new adapter package implementing the diff --git a/contract/continuo_engine_contract/port.py b/contract/continuo_engine_contract/port.py index c1dc8d3..ea198a7 100644 --- a/contract/continuo_engine_contract/port.py +++ b/contract/continuo_engine_contract/port.py @@ -1,4 +1,4 @@ -"""The warehouse-engine adapter port and entry-point discovery. +"""The warehouse-engine adapter port, its opt-in write capabilities, and discovery. One port covers both roles an engine plays in Continuo: @@ -9,6 +9,11 @@ * **python-node data-plane I/O** — the runtime harness reads declared inputs and writes a node's result (``fetch``, ``ensure_table``, ``load``). +Writing modes beyond replace are opt-in capabilities an adapter declares by +inheritance (``TableIntrospection``, ``AppendCapable``, ``UpsertCapable``): an +adapter that declares none still runs every ``replace`` node, and one that +declares a capability must implement all of it to be instantiated. + Both roles are served by the same connection to the same warehouse, so they are one implementation per engine rather than two. Engine packages register a single ``continuo_engine.adapters`` entry point; each image installs exactly one, so @@ -17,9 +22,13 @@ from __future__ import annotations from abc import ABC, abstractmethod +from collections.abc import Sequence +from dataclasses import dataclass from importlib.metadata import entry_points from typing import TYPE_CHECKING +from continuo_engine_contract.keys import ensure_unique_key + ENTRY_POINT_GROUP = "continuo_engine.adapters" if TYPE_CHECKING: # pragma: no cover @@ -126,7 +135,13 @@ def fetch(self, sql: str) -> "pyarrow.Table": @abstractmethod def ensure_table( - self, schema: str, table: str, columns: list[dict], *, config: dict + self, + schema: str, + table: str, + columns: list[dict], + *, + config: dict, + primary_key: Sequence[str] = (), ) -> None: """CREATE TABLE IF NOT EXISTS with the typed DDL compiled from columns. @@ -146,6 +161,13 @@ def ensure_table( recognize, and MUST validate config values before interpolating anything into DDL — the same discipline :meth:`build_empty_from_columns` follows. + + *primary_key* names an upsert node's key columns, spelled exactly as in + *columns*. The harness passes it only for ``upsert`` nodes, so an + adapter that is not :class:`UpsertCapable` never receives it. An engine + that can enforce a key declares it on the table it creates (postgres: + ``PRIMARY KEY``); an engine without constraints ignores it. Like + *config*, it applies only when this call creates the table. """ @abstractmethod @@ -157,6 +179,80 @@ def close(self) -> None: """Release the underlying connection.""" +@dataclass(frozen=True) +class LiveColumn: + """One column of an existing table, as the engine reports it.""" + + name: str + nullable: bool + + +class TableIntrospection(ABC): + """Reads the shape of an existing table. + + The base of every write capability: before an append or upsert the + harness compares the live table with the node's declared columns. + """ + + @abstractmethod + def table_columns(self, schema: str, table: str) -> list[LiveColumn]: + """Columns of an existing table. Raises if the table does not exist. + + Raises ``LookupError`` naming ``schema.table`` when it does not exist. + Column order is not part of the contract; callers compare by name. + """ + + +class AppendCapable(TableIntrospection): + """An engine that can add rows to a table and keep the rows already there. + + Declare it by inheritance, e.g. ``class MyAdapter(WarehouseAdapter, + AppendCapable)``; the harness runs a ``mode_type: append`` node only on an + adapter that is an instance of this class. + """ + + @abstractmethod + def append(self, schema: str, table: str, data: "pyarrow.Table") -> None: + """Atomically insert every row of data. All rows or none. + + The insert names *data*'s columns, so their order need not match the + table's. A 0-row *data* writes nothing. + """ + + +class UpsertCapable(TableIntrospection): + """An engine that can insert new keys and overwrite existing ones in one write. + + Declare it by inheritance. :meth:`upsert` is a template method: the key + invariant is checked here, once for every engine, and the rows then go to + the engine's :meth:`_upsert`. + """ + + def upsert( + self, schema: str, table: str, data: "pyarrow.Table", primary_key: Sequence[str] + ) -> None: + """Atomically insert rows with a new key and overwrite rows whose key exists. + + Rows whose key is not in data are untouched. *primary_key* names + columns of *data* exactly as spelled there. + + Raises: + DuplicateKeyError: If *data* repeats a key; nothing is written. + """ + ensure_unique_key(data, primary_key) + self._upsert(schema, table, data, primary_key) + + @abstractmethod + def _upsert( + self, schema: str, table: str, data: "pyarrow.Table", primary_key: Sequence[str] + ) -> None: + """Engine-specific write. Keys in data are already unique. + + All rows or none; a 0-row *data* writes nothing. When every column is + in the key there is nothing to overwrite: insert only the new keys. + """ + + def discover_adapter() -> tuple[str, type[WarehouseAdapter]]: """Return ``(engine_name, adapter_class)`` from the single installed plugin. diff --git a/contract/tests/test_port.py b/contract/tests/test_port.py index 38880bb..c2d76c3 100644 --- a/contract/tests/test_port.py +++ b/contract/tests/test_port.py @@ -5,14 +5,24 @@ real ``importlib.metadata`` machinery — no monkeypatching of ``entry_points``, so the tests exercise actual entry-point loading rather than an assumed return shape. """ +import dataclasses import subprocess import textwrap from pathlib import Path +import pyarrow as pa import pytest -from continuo_engine_contract.port import ENTRY_POINT_GROUP, WarehouseAdapter +from continuo_engine_contract.keys import DuplicateKeyError +from continuo_engine_contract.port import ( + ENTRY_POINT_GROUP, + AppendCapable, + LiveColumn, + TableIntrospection, + UpsertCapable, + WarehouseAdapter, +) CONTRACT_ROOT = Path(__file__).resolve().parent.parent # contract/ @@ -181,12 +191,122 @@ def test_ensure_table_signature_declares_config_keyword_only(): sig = inspect.signature(WarehouseAdapter.ensure_table) params = list(sig.parameters) - assert params == ["self", "schema", "table", "columns", "config"] + assert params == ["self", "schema", "table", "columns", "config", "primary_key"] config = sig.parameters["config"] assert config.kind is inspect.Parameter.KEYWORD_ONLY assert config.default is inspect.Parameter.empty +def test_ensure_table_accepts_an_optional_keyword_only_primary_key(): + """The harness passes primary_key only for upsert nodes, so it must default to ().""" + import inspect + + primary_key = inspect.signature(WarehouseAdapter.ensure_table).parameters["primary_key"] + assert primary_key.kind is inspect.Parameter.KEYWORD_ONLY + assert primary_key.default == () + + +def test_capabilities_declare_exactly_their_own_abstract_methods(): + """No capability ships a method that exists but raises 'unsupported'.""" + assert TableIntrospection.__abstractmethods__ == frozenset({"table_columns"}) + assert AppendCapable.__abstractmethods__ == frozenset({"table_columns", "append"}) + assert UpsertCapable.__abstractmethods__ == frozenset({"table_columns", "_upsert"}) + + +def test_a_partially_implemented_append_capability_cannot_be_instantiated(): + class NoIntrospection(AppendCapable): + def append(self, schema, table, data): + ... + + with pytest.raises(TypeError, match="table_columns"): + NoIntrospection() # type: ignore[abstract] + + +def test_a_partially_implemented_upsert_capability_cannot_be_instantiated(): + class NoEngineWrite(UpsertCapable): + def table_columns(self, schema, table): + return [] + + with pytest.raises(TypeError, match="_upsert"): + NoEngineWrite() # type: ignore[abstract] + + +class _RecordingUpsert(UpsertCapable): + def __init__(self): + self.writes = [] + + def table_columns(self, schema, table): + return [] + + def _upsert(self, schema, table, data, primary_key): + self.writes.append((schema, table, data, tuple(primary_key))) + + +def test_upsert_hands_unique_rows_to_the_engine_write(): + adapter = _RecordingUpsert() + data = pa.table({"id": [1, 2], "v": ["a", "b"]}) + adapter.upsert("s", "t", data, ["id"]) + assert adapter.writes == [("s", "t", data, ("id",))] + + +def test_upsert_checks_the_key_before_the_engine_sees_any_row(): + adapter = _RecordingUpsert() + with pytest.raises(DuplicateKeyError): + adapter.upsert("s", "t", pa.table({"id": [1, 1]}), ["id"]) + assert adapter.writes == [] + + +class _PortOnly(WarehouseAdapter): + @classmethod + def required_env(cls): + return [] + + @classmethod + def from_env(cls): + return cls() + + def ensure_schema(self, schema): + ... + + def drop_schema(self, schema): + ... + + def build_empty_from_sql(self, schema, table, compiled_sql): + ... + + def clone_empty_from_prod(self, candidate_schema, prod_schema, table): + ... + + def build_empty_from_columns(self, schema, table, columns, config): + ... + + def check_binds(self, sql): + ... + + def fetch(self, sql): + ... + + def ensure_table(self, schema, table, columns, *, config, primary_key=()): + ... + + def load(self, schema, table, data): + ... + + def close(self): + ... + + +def test_capabilities_are_opt_in(): + adapter = _PortOnly() + assert not isinstance(adapter, (TableIntrospection, AppendCapable, UpsertCapable)) + + +def test_live_column_is_a_frozen_value(): + assert LiveColumn("id", False) == LiveColumn(name="id", nullable=False) + with pytest.raises(dataclasses.FrozenInstanceError): + LiveColumn("id", False).name = "x" # type: ignore[misc] + + # --- Real installed-adapter discovery via throwaway venvs -------------------- _FIXTURE_MODULE = '''\ diff --git a/docs/boundary-contract.md b/docs/boundary-contract.md index 83d9f11..0d175cc 100644 --- a/docs/boundary-contract.md +++ b/docs/boundary-contract.md @@ -351,8 +351,9 @@ executor runs it as a Kubernetes Job: 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: - `ensure_table(self, schema, table, columns, - *, config)` — required and keyword-only. Adapters implement exactly that + `ensure_table(self, schema, table, columns, *, config, primary_key=())` — + `config` required and keyword-only; `primary_key` (contract 0.8.0) optional + and keyword-only, passed by the harness only for upsert nodes. Adapters implement exactly that signature; callers pass `{}` when a node declares no physical layout. - **Early `config` tripwire**: `ensure_table` is called only *after* the script has run and its result has been conformed, so a bad `config` would From 4ee033071293c26a9ff000bf5e8f4700e52e515b Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 14:19:31 +0200 Subject: [PATCH 3/6] feat(engine-contract): conformance suite for the write capabilities IntrospectionConformance, AppendConformance and UpsertConformance hold the spec's cases (accumulation, zero rows, atomicity across statements, overwrite, idempotent rerun, repeated keys, every column in the key) plus quoting, column order, first-run upsert and the FX key types. A reference in-memory adapter runs them here, with two broken adapters as negative controls. Shipped in the wheel; pytest and pyarrow come with the testing extra. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 5 + contract/README.md | 3 +- .../continuo_engine_contract/conformance.py | 306 ++++++++++++++++++ contract/pyproject.toml | 6 + contract/tests/test_conformance_suite.py | 168 ++++++++++ contract/tests/test_contract_packaging.py | 25 ++ uv.lock | 13 +- 7 files changed, 524 insertions(+), 2 deletions(-) create mode 100644 contract/continuo_engine_contract/conformance.py create mode 100644 contract/tests/test_conformance_suite.py create mode 100644 contract/tests/test_contract_packaging.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bd2d1e..9900f4b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,11 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `ensure_unique_key` and then the engine's `_upsert`). `ensure_table` takes an optional keyword-only `primary_key`. `continuo_engine_contract.keys` holds `ensure_unique_key` and `DuplicateKeyError(ValueError)`. +- `continuo_engine_contract.conformance`: a pytest conformance suite for the + write capabilities (`IntrospectionConformance`, `AppendConformance`, + `UpsertConformance`). Each shipped adapter's integration suite runs it + against its real engine; a bring-your-own adapter subclasses the same + classes. Install with `continuo-engine-contract[testing]`. ### Changed diff --git a/contract/README.md b/contract/README.md index 42766c7..fac0804 100644 --- a/contract/README.md +++ b/contract/README.md @@ -1,7 +1,7 @@ # continuo-engine-contract The contract shared by the Continuo python runtime — the node harness and the -validation runner alike — and every engine adapter library. Six modules: +validation runner alike — and every engine adapter library. Seven modules: | Module | What it holds | |---|---| @@ -11,6 +11,7 @@ validation runner alike — and every engine adapter library. Six modules: | `continuo_engine_contract.sql` | `ensure_single_read`, the parse-backed gate `check_binds` runs before it lets a declared read near an engine | | `continuo_engine_contract.config` | `ensure_known_keys`, the fail-closed gate `build_empty_from_columns` runs before it lets a node's physical-layout `config` block reach DDL | | `continuo_engine_contract.result` | The sentinel-framed result-block wire format the runner emits and the k8s-controller parses | +| `continuo_engine_contract.conformance` | The pytest conformance suite for the write capabilities — `IntrospectionConformance`, `AppendConformance`, `UpsertConformance` — that an adapter's integration tests subclass, providing `adapter` and `schema` fixtures. Needs the `testing` extra | Each `continuo-python-runtime-` library depends on this package and implements the port. diff --git a/contract/continuo_engine_contract/conformance.py b/contract/continuo_engine_contract/conformance.py new file mode 100644 index 0000000..d0e524d --- /dev/null +++ b/contract/continuo_engine_contract/conformance.py @@ -0,0 +1,306 @@ +"""Conformance tests for the engine contract's write capabilities. + +One test class per capability. An adapter's integration suite subclasses the +classes for exactly the capabilities it declares and runs them against its +real engine; a bring-your-own adapter can do the same:: + + from continuo_engine_contract.conformance import AppendConformance + + class TestMyEngineAppend(AppendConformance): + pass + +The subclass's module (or a conftest.py above it) provides two fixtures: + +``adapter`` + A connected adapter instance that declares the capability under test. +``schema`` + A schema name the tests may create tables in (``ensure_table`` creates + it). The fixture drops it after the test. + +Rows are read back through ``adapter.fetch`` with ANSI double-quoted +identifiers; an engine that quotes differently overrides +:meth:`ConformanceBase.read_rows`. Comparisons raise ``AssertionError`` with +both sides in the message, so no pytest assertion rewriting is needed. + +Needs the ``testing`` extra: ``pip install "continuo-engine-contract[testing]"``. +""" +from __future__ import annotations + +import datetime +import uuid +from collections.abc import Iterable, Sequence +from decimal import Decimal +from typing import Any + +import pyarrow as pa # type: ignore[import-untyped] +import pytest + +from continuo_engine_contract.keys import DuplicateKeyError +from continuo_engine_contract.port import LiveColumn + +# More rows than one insert statement carries in any shipped adapter (postgres +# pages 1000 rows, trino batches 500), so an atomicity case spans statements. +BULK_ROWS = 1200 + +COLUMNS = [ + {"name": "id", "type": "BIGINT", "nullable": False}, + {"name": "name", "type": "VARCHAR(20)", "nullable": False}, + {"name": "score", "type": "BIGINT", "nullable": True}, +] +NAMES = ("id", "name", "score") +ARROW = pa.schema([("id", pa.int64()), ("name", pa.string()), ("score", pa.int64())]) + +# Every column in the key: an upsert has nothing to overwrite. +KEY_ONLY_COLUMNS = [ + {"name": "a", "type": "BIGINT", "nullable": False}, + {"name": "b", "type": "VARCHAR(20)", "nullable": False}, +] +KEY_ONLY_ARROW = pa.schema([("a", pa.int64()), ("b", pa.string())]) + +# Names that need quoting: a reserved word, mixed case, a percent sign. +AWKWARD_COLUMNS = [ + {"name": "order", "type": "BIGINT", "nullable": False}, + {"name": "Rate", "type": "VARCHAR(20)", "nullable": False}, + {"name": "pct%", "type": "BIGINT", "nullable": True}, +] +AWKWARD_NAMES = ("order", "Rate", "pct%") +AWKWARD_ARROW = pa.schema([("order", pa.int64()), ("Rate", pa.string()), ("pct%", pa.int64())]) + +# The shape of the first upsert node (daily FX rates). +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_NAMES = ("rate_date", "base", "rate") +FX_ARROW = pa.schema( + [("rate_date", pa.date32()), ("base", pa.string()), ("rate", pa.decimal128(12, 6))] +) + + +def quote(name: str) -> str: + """*name* as an ANSI double-quoted identifier.""" + return '"' + name.replace('"', '""') + '"' + + +def table_of(schema: Any, rows: Sequence[Sequence[Any]]) -> Any: + """An Arrow table of *rows* (tuples in *schema*'s column order).""" + return pa.Table.from_pylist([dict(zip(schema.names, row)) for row in rows], schema=schema) + + +def bulk_rows(start: int) -> list[tuple[Any, ...]]: + """BULK_ROWS valid ``ARROW`` rows with ids from *start*, then one whose NOT NULL name is NULL.""" + rows: list[tuple[Any, ...]] = [(i, f"n{i}", i) for i in range(start, start + BULK_ROWS)] + rows.append((start + BULK_ROWS, None, 0)) + return rows + + +def canonical(rows: Iterable[Sequence[Any]]) -> list[tuple[Any, ...]]: + """*rows* as tuples in a deterministic order, for comparing a table's contents.""" + return sorted((tuple(row) for row in rows), key=repr) + + +class ConformanceBase: + """Helpers shared by the capability classes (fixtures: see the module docstring).""" + + def new_table( + self, + adapter: Any, + schema: str, + columns: list[dict[str, Any]], + primary_key: Sequence[str] = (), + ) -> str: + """Create a fresh table through ``ensure_table`` and return its name.""" + table = f"conf_{uuid.uuid4().hex[:12]}" + if primary_key: + adapter.ensure_table(schema, table, columns, config={}, primary_key=primary_key) + else: + adapter.ensure_table(schema, table, columns, config={}) + return table + + def read_rows( + self, adapter: Any, schema: str, table: str, names: Sequence[str] + ) -> list[tuple[Any, ...]]: + """Every row of ``schema.table`` as a tuple in *names* order, canonically sorted.""" + select = ", ".join(f"{quote(name)} AS c{index}" for index, name in enumerate(names)) + data = adapter.fetch(f"SELECT {select} FROM {quote(schema)}.{quote(table)}") + return canonical( + tuple(row[f"c{index}"] for index in range(len(names))) for row in data.to_pylist() + ) + + def expect_rows( + self, + adapter: Any, + schema: str, + table: str, + names: Sequence[str], + expected: Iterable[Sequence[Any]], + ) -> None: + """Fail unless ``schema.table`` holds exactly *expected*, in any order.""" + actual = self.read_rows(adapter, schema, table, names) + wanted = canonical(expected) + if actual != wanted: + raise AssertionError(f"{schema}.{table} holds {actual}, expected {wanted}") + + +class IntrospectionConformance(ConformanceBase): + """For an adapter that is a ``TableIntrospection`` (every write capability is).""" + + def test_table_columns_reports_names_and_nullability(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS) + reported = sorted(adapter.table_columns(schema, table), key=lambda column: column.name) + expected = [LiveColumn("id", False), LiveColumn("name", False), LiveColumn("score", True)] + if reported != expected: + raise AssertionError(f"table_columns reported {reported}, expected {expected}") + + def test_table_columns_of_a_missing_table_raises_lookup_error(self, adapter, schema): + self.new_table(adapter, schema, COLUMNS) # the schema exists; only the table is missing + with pytest.raises(LookupError): + adapter.table_columns(schema, f"conf_missing_{uuid.uuid4().hex[:8]}") + + +class AppendConformance(ConformanceBase): + """For an adapter that is an ``AppendCapable``.""" + + def test_append_into_an_empty_table(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS) + adapter.append(schema, table, table_of(ARROW, [(1, "a", 10), (2, "b", None)])) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10), (2, "b", None)]) + + def test_two_appends_accumulate(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS) + adapter.append(schema, table, table_of(ARROW, [(1, "a", 10)])) + adapter.append(schema, table, table_of(ARROW, [(2, "b", 20), (3, "c", 30)])) + self.expect_rows( + adapter, schema, table, NAMES, [(1, "a", 10), (2, "b", 20), (3, "c", 30)] + ) + + def test_append_of_zero_rows_writes_nothing(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS) + adapter.append(schema, table, table_of(ARROW, [(1, "a", 10)])) + adapter.append(schema, table, table_of(ARROW, [])) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10)]) + + def test_append_is_all_rows_or_none(self, adapter, schema): + """More rows than one insert statement carries; the last violates NOT NULL.""" + table = self.new_table(adapter, schema, COLUMNS) + adapter.append(schema, table, table_of(ARROW, [(1, "a", 10)])) + with pytest.raises(Exception): + adapter.append(schema, table, table_of(ARROW, bulk_rows(100))) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10)]) + + def test_append_names_its_columns(self, adapter, schema): + """The data's column order need not match the table's.""" + table = self.new_table(adapter, schema, COLUMNS) + reordered = pa.table({ + "score": pa.array([10], pa.int64()), + "name": pa.array(["a"], pa.string()), + "id": pa.array([1], pa.int64()), + }) + adapter.append(schema, table, reordered) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10)]) + + def test_append_quotes_awkward_identifiers(self, adapter, schema): + table = self.new_table(adapter, schema, AWKWARD_COLUMNS) + adapter.append(schema, table, table_of(AWKWARD_ARROW, [(1, "x", 5), (2, "y", None)])) + self.expect_rows(adapter, schema, table, AWKWARD_NAMES, [(1, "x", 5), (2, "y", None)]) + + +class UpsertConformance(ConformanceBase): + """For an adapter that is an ``UpsertCapable``. Tables are made by ``ensure_table`` with the key.""" + + def test_upsert_inserts_new_keys_into_a_table_created_in_the_same_run(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert(schema, table, table_of(ARROW, [(1, "a", 10), (2, "b", None)]), ["id"]) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10), (2, "b", None)]) + + def test_upsert_overwrites_existing_keys_and_leaves_other_rows_untouched(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert( + schema, table, table_of(ARROW, [(1, "a", 10), (2, "b", 20), (3, "c", 30)]), ["id"] + ) + adapter.upsert(schema, table, table_of(ARROW, [(2, "B", None), (4, "d", 40)]), ["id"]) + self.expect_rows( + adapter, schema, table, NAMES, + [(1, "a", 10), (2, "B", None), (3, "c", 30), (4, "d", 40)], + ) + + def test_upsert_rerun_is_idempotent(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + rows = [(1, "a", 10), (2, "b", 20)] + adapter.upsert(schema, table, table_of(ARROW, rows), ["id"]) + adapter.upsert(schema, table, table_of(ARROW, rows), ["id"]) + self.expect_rows(adapter, schema, table, NAMES, rows) + + def test_upsert_with_a_repeated_key_raises_and_writes_nothing(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert(schema, table, table_of(ARROW, [(1, "a", 10)]), ["id"]) + with pytest.raises(DuplicateKeyError): + adapter.upsert( + schema, table, table_of(ARROW, [(1, "x", 1), (5, "y", 1), (5, "z", 2)]), ["id"] + ) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10)]) + + def test_upsert_when_every_column_is_in_the_key(self, adapter, schema): + table = self.new_table(adapter, schema, KEY_ONLY_COLUMNS, primary_key=["a", "b"]) + adapter.upsert(schema, table, table_of(KEY_ONLY_ARROW, [(1, "x"), (1, "y")]), ["a", "b"]) + adapter.upsert(schema, table, table_of(KEY_ONLY_ARROW, [(1, "x"), (2, "x")]), ["a", "b"]) + self.expect_rows(adapter, schema, table, ("a", "b"), [(1, "x"), (1, "y"), (2, "x")]) + + def test_upsert_is_all_rows_or_none(self, adapter, schema): + """Updates of existing keys plus more rows than one statement; the last violates NOT NULL.""" + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert(schema, table, table_of(ARROW, [(1, "a", 10), (2, "b", 20)]), ["id"]) + rows = [(1, "changed", 1), (2, "changed", 2)] + bulk_rows(100) + with pytest.raises(Exception): + adapter.upsert(schema, table, table_of(ARROW, rows), ["id"]) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10), (2, "b", 20)]) + + def test_upsert_of_zero_rows_writes_nothing(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert(schema, table, table_of(ARROW, [(1, "a", 10)]), ["id"]) + adapter.upsert(schema, table, table_of(ARROW, []), ["id"]) + self.expect_rows(adapter, schema, table, NAMES, [(1, "a", 10)]) + + def test_upsert_names_its_columns(self, adapter, schema): + table = self.new_table(adapter, schema, COLUMNS, primary_key=["id"]) + adapter.upsert(schema, table, table_of(ARROW, [(1, "a", 10)]), ["id"]) + reordered = pa.table({ + "score": pa.array([11, 20], pa.int64()), + "name": pa.array(["A", "b"], pa.string()), + "id": pa.array([1, 2], pa.int64()), + }) + adapter.upsert(schema, table, reordered, ["id"]) + self.expect_rows(adapter, schema, table, NAMES, [(1, "A", 11), (2, "b", 20)]) + + def test_upsert_quotes_awkward_identifiers(self, adapter, schema): + key = ["order", "Rate"] + table = self.new_table(adapter, schema, AWKWARD_COLUMNS, primary_key=key) + adapter.upsert(schema, table, table_of(AWKWARD_ARROW, [(1, "x", 5), (1, "y", 6)]), key) + adapter.upsert(schema, table, table_of(AWKWARD_ARROW, [(1, "x", 50), (2, "x", None)]), key) + self.expect_rows( + adapter, schema, table, AWKWARD_NAMES, [(1, "x", 50), (1, "y", 6), (2, "x", None)] + ) + + def test_upsert_with_a_date_and_text_key_and_a_decimal_value(self, adapter, schema): + key = ["rate_date", "base"] + day1, day2 = datetime.date(2026, 10, 8), datetime.date(2026, 10, 9) + table = self.new_table(adapter, schema, FX_COLUMNS, primary_key=key) + adapter.upsert( + schema, table, + table_of(FX_ARROW, [(day1, "EUR", Decimal("1.081200")), (day1, "GBP", Decimal("0.861000"))]), + key, + ) + adapter.upsert( + schema, table, + table_of(FX_ARROW, [(day1, "EUR", Decimal("1.090000")), (day2, "EUR", None)]), + key, + ) + self.expect_rows( + adapter, schema, table, FX_NAMES, + [ + (day1, "EUR", Decimal("1.090000")), + (day1, "GBP", Decimal("0.861000")), + (day2, "EUR", None), + ], + ) diff --git a/contract/pyproject.toml b/contract/pyproject.toml index 6bffd9c..62a18af 100644 --- a/contract/pyproject.toml +++ b/contract/pyproject.toml @@ -13,6 +13,12 @@ classifiers = [ "Programming Language :: Python :: 3.14", ] +[project.optional-dependencies] +# The conformance suite (continuo_engine_contract.conformance) for adapter test +# suites, this repo's and bring-your-own: pytest, plus pyarrow to build the rows. +# Pinned exactly, like every in-repo dependency. +testing = ["pytest==9.1.1", "pyarrow==25.0.1"] + [build-system] requires = ["hatchling"] build-backend = "hatchling.build" diff --git a/contract/tests/test_conformance_suite.py b/contract/tests/test_conformance_suite.py new file mode 100644 index 0000000..b671039 --- /dev/null +++ b/contract/tests/test_conformance_suite.py @@ -0,0 +1,168 @@ +"""The conformance suite, run against an in-memory reference adapter. + +Proves the suite's own cases agree with each other before any engine runs them, +shows a bring-your-own adapter how to use it, and checks with two broken +adapters that the suite really catches a non-atomic append and an upsert that +loses rows. +""" +import pytest + +from continuo_engine_contract.conformance import ( + AppendConformance, + IntrospectionConformance, + UpsertConformance, + canonical, +) +from continuo_engine_contract.port import ( + AppendCapable, + LiveColumn, + UpsertCapable, + WarehouseAdapter, +) + + +class _MemoryWarehouse(WarehouseAdapter, AppendCapable, UpsertCapable): + """A reference adapter over Python lists: NOT NULL enforced, every write all-or-nothing.""" + + def __init__(self): + self.tables: dict[tuple[str, str], dict] = {} + + @classmethod + def required_env(cls): + return [] + + @classmethod + def from_env(cls): + return cls() + + def ensure_schema(self, schema): + ... + + def drop_schema(self, schema): + for key in [key for key in self.tables if key[0] == schema]: + del self.tables[key] + + def build_empty_from_sql(self, schema, table, compiled_sql): + ... + + def clone_empty_from_prod(self, candidate_schema, prod_schema, table): + ... + + def build_empty_from_columns(self, schema, table, columns, config): + ... + + def check_binds(self, sql): + ... + + def fetch(self, sql): + raise AssertionError("the memory adapter is read through read_rows, never fetch") + + def ensure_table(self, schema, table, columns, *, config, primary_key=()): + self.tables.setdefault( + (schema, table), {"columns": [dict(c) for c in columns], "rows": []} + ) + + def load(self, schema, table, data): + self.tables[(schema, table)]["rows"] = self._rows(schema, table, data.to_pylist()) + + def close(self): + ... + + def table_columns(self, schema, table): + if (schema, table) not in self.tables: + raise LookupError(f"table {schema}.{table} does not exist") + return [ + LiveColumn(c["name"], c.get("nullable", True)) + for c in self.tables[(schema, table)]["columns"] + ] + + def _rows(self, schema, table, rows): + """*rows* projected onto the table's columns; ValueError on a NULL in a NOT NULL column.""" + columns = self.tables[(schema, table)]["columns"] + for row in rows: + for column in columns: + if not column.get("nullable", True) and row.get(column["name"]) is None: + raise ValueError(f"NOT NULL violated: {column['name']}") + return [{c["name"]: row.get(c["name"]) for c in columns} for row in rows] + + def append(self, schema, table, data): + rows = self._rows(schema, table, data.to_pylist()) # validate everything first + self.tables[(schema, table)]["rows"].extend(rows) + + def _upsert(self, schema, table, data, primary_key): + rows = self._rows(schema, table, data.to_pylist()) + stored = self.tables[(schema, table)]["rows"] + index = {tuple(row[k] for k in primary_key): i for i, row in enumerate(stored)} + for row in rows: + key = tuple(row[k] for k in primary_key) + if key in index: + stored[index[key]] = row + else: + index[key] = len(stored) + stored.append(row) + + +class _LeakyAppendWarehouse(_MemoryWarehouse): + """Appends row by row, so a failing row leaves the earlier ones behind.""" + + def append(self, schema, table, data): + for row in data.to_pylist(): + self.tables[(schema, table)]["rows"].extend(self._rows(schema, table, [row])) + + +class _ReplacingUpsertWarehouse(_MemoryWarehouse): + """Upserts by replacing the whole table, losing rows whose key is not in the data.""" + + def _upsert(self, schema, table, data, primary_key): + self.tables[(schema, table)]["rows"] = self._rows(schema, table, data.to_pylist()) + + +class _ReadsMemory: + """Reads the memory adapter's rows directly instead of through fetch.""" + + def read_rows(self, adapter, schema, table, names): + rows = adapter.tables[(schema, table)]["rows"] + return canonical(tuple(row[name] for name in names) for row in rows) + + +@pytest.fixture +def adapter(): + return _MemoryWarehouse() + + +@pytest.fixture +def schema(adapter): + yield "conf" + adapter.drop_schema("conf") + + +class TestMemoryIntrospection(_ReadsMemory, IntrospectionConformance): + pass + + +class TestMemoryAppend(_ReadsMemory, AppendConformance): + pass + + +class TestMemoryUpsert(_ReadsMemory, UpsertConformance): + pass + + +class _MemoryAppendCase(_ReadsMemory, AppendConformance): + pass + + +class _MemoryUpsertCase(_ReadsMemory, UpsertConformance): + pass + + +def test_the_suite_catches_a_non_atomic_append(): + with pytest.raises(AssertionError, match="holds"): + _MemoryAppendCase().test_append_is_all_rows_or_none(_LeakyAppendWarehouse(), "conf") + + +def test_the_suite_catches_an_upsert_that_loses_other_rows(): + with pytest.raises(AssertionError, match="holds"): + _MemoryUpsertCase().test_upsert_overwrites_existing_keys_and_leaves_other_rows_untouched( + _ReplacingUpsertWarehouse(), "conf" + ) diff --git a/contract/tests/test_contract_packaging.py b/contract/tests/test_contract_packaging.py new file mode 100644 index 0000000..8a646b3 --- /dev/null +++ b/contract/tests/test_contract_packaging.py @@ -0,0 +1,25 @@ +"""The published wheel ships the conformance suite and declares the testing extra.""" +import re +import subprocess +import zipfile +from pathlib import Path + +CONTRACT_ROOT = Path(__file__).resolve().parent.parent # contract/ + + +def test_wheel_ships_the_conformance_suite_and_the_testing_extra(tmp_path): + subprocess.run( + ["uv", "build", "--wheel", "--out-dir", str(tmp_path), str(CONTRACT_ROOT)], + check=True, + capture_output=True, + ) + (wheel,) = tmp_path.glob("continuo_engine_contract-*.whl") + with zipfile.ZipFile(wheel) as archive: + names = set(archive.namelist()) + (metadata_name,) = [n for n in names if n.endswith(".dist-info/METADATA")] + metadata = archive.read(metadata_name).decode() + assert "continuo_engine_contract/conformance.py" in names + assert "continuo_engine_contract/keys.py" in names + assert "Provides-Extra: testing" in metadata + assert re.search(r"Requires-Dist: pytest==9\.1\.1; extra == ['\"]testing['\"]", metadata) + assert re.search(r"Requires-Dist: pyarrow==25\.0\.1; extra == ['\"]testing['\"]", metadata) diff --git a/uv.lock b/uv.lock index d239f80..8da7800 100644 --- a/uv.lock +++ b/uv.lock @@ -209,8 +209,19 @@ dependencies = [ { name = "sqlglot" }, ] +[package.optional-dependencies] +testing = [ + { name = "pyarrow" }, + { name = "pytest" }, +] + [package.metadata] -requires-dist = [{ name = "sqlglot", specifier = "==30.17.0" }] +requires-dist = [ + { name = "pyarrow", marker = "extra == 'testing'", specifier = "==25.0.1" }, + { name = "pytest", marker = "extra == 'testing'", specifier = "==9.1.1" }, + { name = "sqlglot", specifier = "==30.17.0" }, +] +provides-extras = ["testing"] [[package]] name = "continuo-postgres-adapter" From 6cd18085211a881acbf1ada0da78f73948e03dfa Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 14:22:38 +0200 Subject: [PATCH 4/6] chore(engine-contract): version 0.8.0 and pins The write capabilities are a new contract surface the adapters will subclass, so the contract moves to 0.8.0 and every exact pin (runtime, three adapters, three image requirement files) moves with it. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 3 +++ adapters/duckdb/pyproject.toml | 2 +- adapters/postgres/pyproject.toml | 2 +- adapters/trino/pyproject.toml | 2 +- contract/pyproject.toml | 2 +- image-requirements-duckdb.txt | 2 +- image-requirements-postgres.txt | 2 +- image-requirements-trino.txt | 2 +- pyproject.toml | 2 +- uv.lock | 2 +- 10 files changed, 12 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9900f4b..a224d6c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,9 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). the trove classifiers and `setup-python` inputs track its major.minor. A new consistency guard (`tests/test_python_version_consistency.py`) scans every `pyproject.toml`, `Dockerfile*` and workflow and fails on any drift. +- `continuo-engine-contract` 0.7.3 -> 0.8.0 (the write capabilities above). + The runtime and the three adapters pin it exactly, as do the engine images' + `image-requirements-*.txt`. - **Breaking for newly copied templates.** The `template/` release workflow calls continuo's public release API instead of an unauthenticated webhook: - `RELEASE_ENDPOINT` keeps its name but is now the origin of continuo's diff --git a/adapters/duckdb/pyproject.toml b/adapters/duckdb/pyproject.toml index cf40088..1a10d63 100644 --- a/adapters/duckdb/pyproject.toml +++ b/adapters/duckdb/pyproject.toml @@ -10,7 +10,7 @@ dependencies = [ # Pinned exactly for the same reason as the postgres adapter: the # [tool.uv.sources] redirect below is uv-only and never reaches built # metadata, and this adapter subclasses the contract's port directly. - "continuo-engine-contract==0.7.3", + "continuo-engine-contract==0.8.0", "duckdb==1.5.6", "pyarrow==25.0.1", ] diff --git a/adapters/postgres/pyproject.toml b/adapters/postgres/pyproject.toml index e90ee39..172e05c 100644 --- a/adapters/postgres/pyproject.toml +++ b/adapters/postgres/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ # of this adapter would otherwise accept whatever contract release is # newest. This adapter subclasses the contract's WarehouseAdapter port # directly, so the pair must stay the tested one. - "continuo-engine-contract==0.7.3", + "continuo-engine-contract==0.8.0", "psycopg2-binary==2.9.12", "pyarrow==25.0.1", ] diff --git a/adapters/trino/pyproject.toml b/adapters/trino/pyproject.toml index 74b6c86..635bca8 100644 --- a/adapters/trino/pyproject.toml +++ b/adapters/trino/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ # of this adapter would otherwise accept whatever contract release is # newest. This adapter subclasses the contract's WarehouseAdapter port # directly, so the pair must stay the tested one. - "continuo-engine-contract==0.7.3", + "continuo-engine-contract==0.8.0", "trino==0.339.0", "pyarrow==25.0.1", ] diff --git a/contract/pyproject.toml b/contract/pyproject.toml index 62a18af..ca5a0ac 100644 --- a/contract/pyproject.toml +++ b/contract/pyproject.toml @@ -1,5 +1,5 @@ [project] -version = "0.7.3" +version = "0.8.0" name = "continuo-engine-contract" description = "The contract a Continuo warehouse engine must satisfy: the WarehouseAdapter port and the result-block wire format." authors = [{ name = "Simone Carolini" }] diff --git a/image-requirements-duckdb.txt b/image-requirements-duckdb.txt index b6c66fa..2aacb75 100644 --- a/image-requirements-duckdb.txt +++ b/image-requirements-duckdb.txt @@ -1,3 +1,3 @@ continuo-python-runtime==0.8.0 continuo-duckdb-adapter==0.1.0 -continuo-engine-contract==0.7.3 +continuo-engine-contract==0.8.0 diff --git a/image-requirements-postgres.txt b/image-requirements-postgres.txt index b61ec8f..74954d8 100644 --- a/image-requirements-postgres.txt +++ b/image-requirements-postgres.txt @@ -1,3 +1,3 @@ continuo-python-runtime==0.8.0 continuo-postgres-adapter==0.2.2 -continuo-engine-contract==0.7.3 +continuo-engine-contract==0.8.0 diff --git a/image-requirements-trino.txt b/image-requirements-trino.txt index 31ee21a..8813359 100644 --- a/image-requirements-trino.txt +++ b/image-requirements-trino.txt @@ -1,3 +1,3 @@ continuo-python-runtime==0.8.0 continuo-trino-adapter==0.2.1 -continuo-engine-contract==0.7.3 +continuo-engine-contract==0.8.0 diff --git a/pyproject.toml b/pyproject.toml index dc9c82b..4a0936b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,7 +19,7 @@ dependencies = [ # directly. Both packages ship from this repo on one v* tag, so a new # contract version never exists without a new runtime version to match it; # an exact pin therefore costs nothing and keeps the tested pair together. - "continuo-engine-contract==0.7.3", + "continuo-engine-contract==0.8.0", "boto3==1.43.85", "pyarrow==25.0.1", "PyYAML==6.0.3", diff --git a/uv.lock b/uv.lock index 8da7800..dbfbd5b 100644 --- a/uv.lock +++ b/uv.lock @@ -203,7 +203,7 @@ test = [ [[package]] name = "continuo-engine-contract" -version = "0.7.3" +version = "0.8.0" source = { editable = "contract" } dependencies = [ { name = "sqlglot" }, From 28b6402fe49859e5a648b4ab573874381183ce76 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:15:10 +0200 Subject: [PATCH 5/6] fix(engine-contract): reject floating-point primary keys in ensure_unique_key Arrow groups 0.0 and -0.0 as distinct, so the uniqueness check passed rows a warehouse would treat as one key, letting a duplicate reach _upsert. The check cannot import pyarrow to normalize, so it now rejects floating-point key columns outright, with a regression test; decimal keys are unaffected. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- contract/continuo_engine_contract/keys.py | 15 ++++++++++++- contract/tests/test_contract_keys.py | 27 +++++++++++++++++++++++ 2 files changed, 41 insertions(+), 1 deletion(-) diff --git a/contract/continuo_engine_contract/keys.py b/contract/continuo_engine_contract/keys.py index 7543ea0..1ccdfa4 100644 --- a/contract/continuo_engine_contract/keys.py +++ b/contract/continuo_engine_contract/keys.py @@ -34,7 +34,9 @@ def ensure_unique_key(data: "pyarrow.Table", primary_key: Sequence[str]) -> None needs no pyarrow import. Raises: - ValueError: If *primary_key* is empty or names a column *data* lacks. + ValueError: If *primary_key* is empty, names a column *data* lacks, or + includes a floating-point column (a warehouse treats ``0.0`` and + ``-0.0`` as one key, which this check cannot see). DuplicateKeyError: If a key value appears in more than one row. """ key = list(primary_key) @@ -45,6 +47,17 @@ def ensure_unique_key(data: "pyarrow.Table", primary_key: Sequence[str]) -> None raise ValueError( f"primary key column(s) {missing} are not in the data's columns {data.column_names}" ) + floating = [ + name + for name in key + if str(data.schema.field(name).type) in {"halffloat", "float", "double"} + ] + if floating: + raise ValueError( + f"primary key column(s) {floating} are floating-point; floating-point " + "columns cannot be a primary key because the warehouse treats values like " + "0.0 and -0.0 as equal while this check would see them as distinct" + ) if data.num_rows < 2: return # Positional names: a key column called "count_all" must not collide with diff --git a/contract/tests/test_contract_keys.py b/contract/tests/test_contract_keys.py index 72f0e44..cb19e0f 100644 --- a/contract/tests/test_contract_keys.py +++ b/contract/tests/test_contract_keys.py @@ -66,3 +66,30 @@ def test_a_key_column_missing_from_the_data_is_rejected(): with pytest.raises(ValueError, match=r"primary key column\(s\) \['quote'\]") as caught: ensure_unique_key(pa.table({"base": ["EUR"]}), ["base", "quote"]) assert type(caught.value) is ValueError + + +def test_a_floating_point_key_is_rejected(): + # Arrow groups 0.0 and -0.0 as distinct; a warehouse treats them as one key. + data = pa.table({"id": pa.array([0.0, -0.0], pa.float64())}) + with pytest.raises(ValueError, match="floating-point") as caught: + ensure_unique_key(data, ["id"]) + assert type(caught.value) is ValueError + + +def test_a_float32_key_is_rejected(): + data = pa.table({"id": pa.array([1.0, 2.0], pa.float32())}) + with pytest.raises(ValueError, match="floating-point") as caught: + ensure_unique_key(data, ["id"]) + assert type(caught.value) is ValueError + + +def test_a_composite_key_with_a_float_column_is_rejected(): + data = pa.table({"base": ["EUR", "USD"], "rate": pa.array([1.5, 2.5], pa.float64())}) + with pytest.raises(ValueError, match=r"column\(s\) \['rate'\] are floating-point") as caught: + ensure_unique_key(data, ["base", "rate"]) + assert type(caught.value) is ValueError + + +def test_a_decimal_key_is_allowed(): + data = pa.table({"k": pa.array([Decimal("1.0"), Decimal("2.0")], pa.decimal128(12, 6))}) + ensure_unique_key(data, ["k"]) From eae4b8f184cef60cf0f2389028b22adb222f562e Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Sat, 10 Oct 2026 15:49:39 +0200 Subject: [PATCH 6/6] refactor(engine-contract): isolate reusable conformance tests Signed-off-by: Simone Carolini --- CHANGELOG.md | 8 ++--- CLAUDE.md | 30 +++++++++++++++++++ contract/README.md | 2 +- .../testing/__init__.py | 4 +++ .../{ => testing}/conformance.py | 2 +- contract/pyproject.toml | 2 +- contract/tests/test_conformance_suite.py | 12 ++++---- contract/tests/test_contract_packaging.py | 3 +- 8 files changed, 49 insertions(+), 14 deletions(-) create mode 100644 contract/continuo_engine_contract/testing/__init__.py rename contract/continuo_engine_contract/{ => testing}/conformance.py (99%) diff --git a/CHANGELOG.md b/CHANGELOG.md index a224d6c..fc26b00 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,11 +25,11 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). `ensure_unique_key` and then the engine's `_upsert`). `ensure_table` takes an optional keyword-only `primary_key`. `continuo_engine_contract.keys` holds `ensure_unique_key` and `DuplicateKeyError(ValueError)`. -- `continuo_engine_contract.conformance`: a pytest conformance suite for the +- `continuo_engine_contract.testing.conformance`: a pytest conformance suite for the write capabilities (`IntrospectionConformance`, `AppendConformance`, - `UpsertConformance`). Each shipped adapter's integration suite runs it - against its real engine; a bring-your-own adapter subclasses the same - classes. Install with `continuo-engine-contract[testing]`. + `UpsertConformance`). Adapter integration suites subclass the reusable + classes and run them against their real engine, providing `adapter` and + `schema` fixtures. Install with `continuo-engine-contract[testing]`. ### Changed diff --git a/CLAUDE.md b/CLAUDE.md index 14eda13..c5ff07f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -21,6 +21,36 @@ Every user-facing change keeps `CHANGELOG.md` current. Two steps, both required: If the changelog is missing a shipped version, backfill it from that tag's merged PRs before adding anything new. +## Engine adapter conformance + +`contract/continuo_engine_contract/testing/conformance.py` +(`continuo_engine_contract.testing.conformance`) is the shared, published +conformance suite for engine adapter authors. Its dependencies are installed +with `continuo-engine-contract[testing]`. + +- Every adapter MUST satisfy the engine contract and pass the shared + conformance classes for every capability it declares: `IntrospectionConformance` + for `TableIntrospection`, `AppendConformance` for `AppendCapable`, and + `UpsertConformance` for `UpsertCapable`. Append and upsert adapters must also + run the introspection class. An adapter need not declare capabilities its + engine cannot support. +- Each adapter's integration suite MUST subclass the applicable classes, + supply connected `adapter` and temporary `schema` fixtures, and run them + against its real warehouse in CI. In-memory reference tests alone do not + establish an adapter's conformance. Do not skip or weaken shared checks to + accommodate an adapter that violates the contract. +- Every new or changed engine-independent contract requirement, including + bug regressions, MUST have coverage in this shared testing module. Extend + the applicable conformance class, or add a class for a new capability, so + all adapters declaring it are checked by the same cases. Exercise the + reusable cases in `contract/tests/test_conformance_suite.py` as well. +- Engine-specific implementation details, such as SQL generation, driver + behavior and authentication, belong in the owning adapter's tests. Add a + shared case when the behavior expresses a requirement of the engine + contract; unrelated runtime features do not belong in this module. +- Before completing an adapter or contract change, run `contract/tests` and + the affected adapter's unit and real-engine integration suites in Docker. + ## Releasing A release is one `chore(release):` commit that bumps versions and updates the diff --git a/contract/README.md b/contract/README.md index fac0804..68f5c7f 100644 --- a/contract/README.md +++ b/contract/README.md @@ -11,7 +11,7 @@ validation runner alike — and every engine adapter library. Seven modules: | `continuo_engine_contract.sql` | `ensure_single_read`, the parse-backed gate `check_binds` runs before it lets a declared read near an engine | | `continuo_engine_contract.config` | `ensure_known_keys`, the fail-closed gate `build_empty_from_columns` runs before it lets a node's physical-layout `config` block reach DDL | | `continuo_engine_contract.result` | The sentinel-framed result-block wire format the runner emits and the k8s-controller parses | -| `continuo_engine_contract.conformance` | The pytest conformance suite for the write capabilities — `IntrospectionConformance`, `AppendConformance`, `UpsertConformance` — that an adapter's integration tests subclass, providing `adapter` and `schema` fixtures. Needs the `testing` extra | +| `continuo_engine_contract.testing.conformance` | The pytest conformance suite for the write capabilities — `IntrospectionConformance`, `AppendConformance`, `UpsertConformance` — that an adapter's integration tests subclass, providing `adapter` and `schema` fixtures. Needs the `testing` extra | Each `continuo-python-runtime-` library depends on this package and implements the port. diff --git a/contract/continuo_engine_contract/testing/__init__.py b/contract/continuo_engine_contract/testing/__init__.py new file mode 100644 index 0000000..981d950 --- /dev/null +++ b/contract/continuo_engine_contract/testing/__init__.py @@ -0,0 +1,4 @@ +"""Reusable test support for warehouse adapter authors. + +Install ``continuo-engine-contract[testing]`` to use the conformance suite. +""" diff --git a/contract/continuo_engine_contract/conformance.py b/contract/continuo_engine_contract/testing/conformance.py similarity index 99% rename from contract/continuo_engine_contract/conformance.py rename to contract/continuo_engine_contract/testing/conformance.py index d0e524d..e98cc6c 100644 --- a/contract/continuo_engine_contract/conformance.py +++ b/contract/continuo_engine_contract/testing/conformance.py @@ -4,7 +4,7 @@ classes for exactly the capabilities it declares and runs them against its real engine; a bring-your-own adapter can do the same:: - from continuo_engine_contract.conformance import AppendConformance + from continuo_engine_contract.testing.conformance import AppendConformance class TestMyEngineAppend(AppendConformance): pass diff --git a/contract/pyproject.toml b/contract/pyproject.toml index ca5a0ac..b9592da 100644 --- a/contract/pyproject.toml +++ b/contract/pyproject.toml @@ -14,7 +14,7 @@ classifiers = [ ] [project.optional-dependencies] -# The conformance suite (continuo_engine_contract.conformance) for adapter test +# The conformance suite (continuo_engine_contract.testing.conformance) for adapter test # suites, this repo's and bring-your-own: pytest, plus pyarrow to build the rows. # Pinned exactly, like every in-repo dependency. testing = ["pytest==9.1.1", "pyarrow==25.0.1"] diff --git a/contract/tests/test_conformance_suite.py b/contract/tests/test_conformance_suite.py index b671039..16f34bc 100644 --- a/contract/tests/test_conformance_suite.py +++ b/contract/tests/test_conformance_suite.py @@ -7,18 +7,18 @@ """ import pytest -from continuo_engine_contract.conformance import ( - AppendConformance, - IntrospectionConformance, - UpsertConformance, - canonical, -) from continuo_engine_contract.port import ( AppendCapable, LiveColumn, UpsertCapable, WarehouseAdapter, ) +from continuo_engine_contract.testing.conformance import ( + AppendConformance, + IntrospectionConformance, + UpsertConformance, + canonical, +) class _MemoryWarehouse(WarehouseAdapter, AppendCapable, UpsertCapable): diff --git a/contract/tests/test_contract_packaging.py b/contract/tests/test_contract_packaging.py index 8a646b3..00f1ce7 100644 --- a/contract/tests/test_contract_packaging.py +++ b/contract/tests/test_contract_packaging.py @@ -18,7 +18,8 @@ def test_wheel_ships_the_conformance_suite_and_the_testing_extra(tmp_path): names = set(archive.namelist()) (metadata_name,) = [n for n in names if n.endswith(".dist-info/METADATA")] metadata = archive.read(metadata_name).decode() - assert "continuo_engine_contract/conformance.py" in names + assert "continuo_engine_contract/testing/__init__.py" in names + assert "continuo_engine_contract/testing/conformance.py" in names assert "continuo_engine_contract/keys.py" in names assert "Provides-Extra: testing" in metadata assert re.search(r"Requires-Dist: pytest==9\.1\.1; extra == ['\"]testing['\"]", metadata)