feat(engine-contract): write capabilities, ensure_unique_key, conformance suite (contract 0.8.0)

CHANGELOG.md

@@ -18,6 +18,18 @@ 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)`.
+- `continuo_engine_contract.testing.conformance`: a pytest conformance suite for the
+  write capabilities (`IntrospectionConformance`, `AppendConformance`,
+  `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
 
@@ -27,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

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

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",
 ]

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",
 ]

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",
 ]

contract/README.md

@@ -1,15 +1,17 @@
 # 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. Seven 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 |
 | `continuo_engine_contract.result` | The sentinel-framed result-block wire format the runner emits and the k8s-controller parses |
+| `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-<engine>` library depends on this package and
 implements the port.
@@ -40,6 +42,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

contract/continuo_engine_contract/keys.py

@@ -0,0 +1,87 @@
+"""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, 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)
+    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}"
+        )
+    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
+    # 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"
+    )

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.
 

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.
+"""