diff --git a/CHANGELOG.md b/CHANGELOG.md index f8a708b..fc26b00 100644 --- a/CHANGELOG.md +++ b/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 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/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/README.md b/contract/README.md index 88be9a9..68f5c7f 100644 --- a/contract/README.md +++ b/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-` 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 diff --git a/contract/continuo_engine_contract/keys.py b/contract/continuo_engine_contract/keys.py new file mode 100644 index 0000000..1ccdfa4 --- /dev/null +++ b/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" + ) 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/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/testing/conformance.py b/contract/continuo_engine_contract/testing/conformance.py new file mode 100644 index 0000000..e98cc6c --- /dev/null +++ b/contract/continuo_engine_contract/testing/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.testing.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..b9592da 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" }] @@ -13,6 +13,12 @@ classifiers = [ "Programming Language :: Python :: 3.14", ] +[project.optional-dependencies] +# 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"] + [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..16f34bc --- /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.port import ( + AppendCapable, + LiveColumn, + UpsertCapable, + WarehouseAdapter, +) +from continuo_engine_contract.testing.conformance import ( + AppendConformance, + IntrospectionConformance, + UpsertConformance, + canonical, +) + + +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_keys.py b/contract/tests/test_contract_keys.py new file mode 100644 index 0000000..cb19e0f --- /dev/null +++ b/contract/tests/test_contract_keys.py @@ -0,0 +1,95 @@ +"""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 + + +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"]) diff --git a/contract/tests/test_contract_packaging.py b/contract/tests/test_contract_packaging.py new file mode 100644 index 0000000..00f1ce7 --- /dev/null +++ b/contract/tests/test_contract_packaging.py @@ -0,0 +1,26 @@ +"""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/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) + assert re.search(r"Requires-Dist: pyarrow==25\.0\.1; extra == ['\"]testing['\"]", metadata) 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 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 d239f80..dbfbd5b 100644 --- a/uv.lock +++ b/uv.lock @@ -203,14 +203,25 @@ test = [ [[package]] name = "continuo-engine-contract" -version = "0.7.3" +version = "0.8.0" source = { editable = "contract" } 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"