From 98409da2da9178e9c173667654113e872ef89b5a Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:34:56 +0200 Subject: [PATCH 1/8] test(duckdb): pin that DuckLake reports NOT NULL through introspection The write-modes shape check compares nullability; duckdb_columns() and information_schema both report the NOT NULL that ensure_table declared on the stack's DuckLake (postgres catalog, MinIO data). Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- ...st_integration_duckdb_nullability_probe.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) create mode 100644 adapters/duckdb/tests/test_integration_duckdb_nullability_probe.py diff --git a/adapters/duckdb/tests/test_integration_duckdb_nullability_probe.py b/adapters/duckdb/tests/test_integration_duckdb_nullability_probe.py new file mode 100644 index 0000000..35ab5ba --- /dev/null +++ b/adapters/duckdb/tests/test_integration_duckdb_nullability_probe.py @@ -0,0 +1,35 @@ +"""Probe (spec section 8): DuckLake reports NOT NULL through introspection. + +table_columns reads duckdb_columns(); the run-time shape check compares the +nullability it reports with the contract. Pinned against the compose stack's +DuckLake (postgres catalog, MinIO data). +""" +import pytest + +pytestmark = pytest.mark.integration + +COLS = [ + {"name": "id", "type": "BIGINT", "nullable": False}, + {"name": "label", "type": "VARCHAR(10)", "nullable": True}, +] + + +def test_duckdb_columns_reports_not_null_on_ducklake(adapter, schema): + adapter.ensure_table(schema, "probe", COLS, config={}) + rows = adapter.fetch( + "SELECT column_name, is_nullable FROM duckdb_columns() " + f"WHERE database_name = 'lake' AND schema_name = '{schema}' AND table_name = 'probe' " + "ORDER BY column_index" + ).to_pylist() + assert rows == [ + {"column_name": "id", "is_nullable": False}, + {"column_name": "label", "is_nullable": True}, + ] + + +def test_information_schema_agrees(adapter, schema, columns_of): + adapter.ensure_table(schema, "probe", COLS, config={}) + assert [(name, nullable) for name, _type, nullable in columns_of(schema, "probe")] == [ + ("id", "NO"), + ("label", "YES"), + ] From 606d5ee17cf503d47200bbaf854b6bf5527e29ee Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:38:05 +0200 Subject: [PATCH 2/8] feat(duckdb): table_columns through LakeGateway.columns LakeWarehouse declares TableIntrospection. The new LakeGateway.columns reads duckdb_columns() with the same pre-filter-then-confirm identifier matching as table_exists, and table_columns raises LookupError for a missing table. The conformance suite runs against the DuckLake stack from here on. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../application/ports.py | 6 ++++ .../application/warehouse.py | 19 ++++++++++-- .../infrastructure/ddl.py | 8 +++++ .../infrastructure/session.py | 14 +++++++++ adapters/duckdb/tests/conftest.py | 5 +++ .../tests/test_adapter_duckdb_runtime.py | 12 +++++++ .../duckdb/tests/test_infra_duckdb_session.py | 31 +++++++++++++++++++ .../tests/test_infra_duckdb_settings_ddl.py | 8 +++++ .../test_integration_duckdb_conformance.py | 15 +++++++++ 9 files changed, 116 insertions(+), 2 deletions(-) create mode 100644 adapters/duckdb/tests/test_integration_duckdb_conformance.py diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py index 49a74d6..8d31523 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py @@ -6,6 +6,8 @@ from contextlib import AbstractContextManager from typing import TYPE_CHECKING +from continuo_engine_contract.port import LiveColumn # type: ignore[import-untyped] + from ..domain.columns import ColumnDefinition from ..domain.identifiers import Identifier, QualifiedTable from ..domain.layout import TableLayout @@ -35,6 +37,10 @@ def drop_schema_cascade(self, schema: Identifier) -> None: @abstractmethod def table_exists(self, table: QualifiedTable) -> bool: ... + @abstractmethod + def columns(self, table: QualifiedTable) -> "list[LiveColumn]": + """Name and nullability of every column of *table*, in table order; [] when absent.""" + @abstractmethod def drop_table_if_exists(self, table: QualifiedTable) -> None: ... diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py index 6ef39e3..e5b1ead 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py @@ -13,7 +13,11 @@ from collections.abc import Callable from typing import TYPE_CHECKING, Any -from continuo_engine_contract.port import WarehouseAdapter # type: ignore[import-untyped] +from continuo_engine_contract.port import ( # type: ignore[import-untyped] + LiveColumn, + TableIntrospection, + WarehouseAdapter, +) from continuo_engine_contract.sql import ensure_single_read # type: ignore[import-untyped] from ..domain.columns import ColumnDefinition, column_types @@ -42,7 +46,7 @@ def _columns(raw: list[dict]) -> list[ColumnDefinition]: return [ColumnDefinition.from_mapping(entry) for entry in raw] -class LakeWarehouse(WarehouseAdapter): +class LakeWarehouse(WarehouseAdapter, TableIntrospection): """WarehouseAdapter implemented over a LakeGateway.""" def __init__( @@ -159,6 +163,17 @@ def validate_config(cls, config: dict[str, Any] | None, column_names: list[str]) """ TableLayout.from_config(config, {name: None for name in column_names}) + def table_columns(self, schema: str, table: str) -> list[LiveColumn]: + """Columns of ``schema.table`` as the DuckLake catalog reports them. + + Raises: + LookupError: If the table does not exist. + """ + columns = self._gateway.columns(QualifiedTable.of(schema, table)) + if not columns: + raise LookupError(f"table {schema}.{table} does not exist") + return columns + def ensure_table( self, schema: str, diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py index 7bbb4f4..2b2074e 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py @@ -149,6 +149,14 @@ class DdlRenderer: "AND lower(table_name) = lower(?)" ) + # Same pre-filter-then-confirm shape as TABLE_EXISTS_QUERY: the session keeps + # only the rows whose schema and table pass ``same_identifier``. + COLUMNS_QUERY = ( + "SELECT schema_name, table_name, column_name, is_nullable FROM duckdb_columns() " + "WHERE lower(database_name) = lower(?) AND lower(schema_name) = lower(?) " + "AND lower(table_name) = lower(?) ORDER BY column_index" + ) + def __init__(self, catalog: str) -> None: self._catalog = catalog diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py index 99f3f31..d6e460a 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py @@ -19,6 +19,7 @@ from typing import TYPE_CHECKING import duckdb +from continuo_engine_contract.port import LiveColumn # type: ignore[import-untyped] from ..application.ports import LakeConflictError, LakeGateway from ..domain.columns import ColumnDefinition @@ -220,6 +221,19 @@ def table_exists(self, table: QualifiedTable) -> bool: for schema, name in rows ) + def columns(self, table: QualifiedTable) -> list[LiveColumn]: + with self._guarded(): + rows = self._run( + DdlRenderer.COLUMNS_QUERY, + [self._renderer.catalog_name, table.schema.name, table.table.name], + ).fetchall() + return [ + LiveColumn(name=name, nullable=bool(nullable)) + for schema, table_name, name, nullable in rows + if same_identifier(schema, table.schema.name) + and same_identifier(table_name, table.table.name) + ] + def drop_table_if_exists(self, table: QualifiedTable) -> None: self._run(self._renderer.drop_table_if_exists(table)) diff --git a/adapters/duckdb/tests/conftest.py b/adapters/duckdb/tests/conftest.py index 2548653..8c1cbad 100644 --- a/adapters/duckdb/tests/conftest.py +++ b/adapters/duckdb/tests/conftest.py @@ -35,6 +35,7 @@ def __init__(self) -> None: self.conflicts: dict[str, int] = {} self.fail_on: dict[str, Exception] = {} self.fetch_result = None + self.live_columns: list = [] def names(self) -> list[str]: return [call[0] for call in self.calls] @@ -67,6 +68,10 @@ def table_exists(self, table): self._do("table_exists", table) return (table.schema.name, table.table.name) in self.existing + def columns(self, table): + self._do("columns", table) + return list(self.live_columns) + def drop_table_if_exists(self, table): self._do("drop_table_if_exists", table) diff --git a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py index 0ab23e8..ae8b593 100644 --- a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py +++ b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py @@ -1,6 +1,7 @@ """Runtime use cases, against the recording FakeLakeGateway (no engine).""" import pyarrow as pa import pytest +from continuo_engine_contract.port import LiveColumn from continuo_duckdb_adapter.application.ports import LakeConflictError from continuo_duckdb_adapter.domain.columns import ColumnDefinition @@ -104,3 +105,14 @@ def test_load_rolls_back_when_the_insert_fails(warehouse, gateway): with pytest.raises(RuntimeError): warehouse.load("s", "t", pa.table({"id": [1]})) assert gateway.names() == ["begin", "delete_all", "insert_arrow", "rollback"] + + +def test_table_columns_returns_the_gateways_columns(warehouse, gateway): + gateway.live_columns = [LiveColumn("id", False), LiveColumn("ts", True)] + assert warehouse.table_columns("s", "t") == [LiveColumn("id", False), LiveColumn("ts", True)] + assert gateway.calls == [("columns", T)] + + +def test_table_columns_of_a_missing_table_raises_lookup_error(warehouse, gateway): + with pytest.raises(LookupError, match=r"s\.t"): + warehouse.table_columns("s", "t") diff --git a/adapters/duckdb/tests/test_infra_duckdb_session.py b/adapters/duckdb/tests/test_infra_duckdb_session.py index f3f0ce2..4757b81 100644 --- a/adapters/duckdb/tests/test_infra_duckdb_session.py +++ b/adapters/duckdb/tests/test_infra_duckdb_session.py @@ -6,6 +6,7 @@ import duckdb import pytest +from continuo_engine_contract.port import LiveColumn from continuo_duckdb_adapter.application.ports import LakeConflictError from continuo_duckdb_adapter.domain.identifiers import Identifier, QualifiedTable @@ -456,3 +457,33 @@ def test_table_exists_does_not_fold_non_ascii_case(): def test_table_exists_is_false_when_the_catalog_returns_nothing(): session, _ = make_session(FakeConnection()) assert not session.table_exists(T) + + +def test_columns_keeps_only_the_rows_of_the_same_table_by_duckdb_casing(): + connection = FakeConnection() + connection.rows = [ + ("Analytics", "Orders", "id", False), + ("Analytics", "Orders", "note", True), + ("analytics", "Other", "x", True), + ] + session, _ = make_session(connection) + assert session.columns(QualifiedTable.of("analytics", "orders")) == [ + LiveColumn("id", False), + LiveColumn("note", True), + ] + assert connection.statements == [DdlRenderer.COLUMNS_QUERY] + + +def test_columns_of_a_missing_table_is_empty(): + session, _ = make_session(FakeConnection()) + assert session.columns(T) == [] + + +def test_columns_errors_are_redacted(): + session, connection = make_session() + connection.errors["SELECT schema_name, table_name, column_name"] = leaky( + duckdb.IOException, PASSWORD + ) + with pytest.raises(duckdb.IOException) as caught: + session.columns(T) + assert PASSWORD not in str(caught.value) diff --git a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py index 29d8a46..a8ab913 100644 --- a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py +++ b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py @@ -293,3 +293,11 @@ def test_table_exists_query_prefilters_with_lower_and_binds_every_name(): "WHERE lower(database_name) = lower(?) AND lower(schema_name) = lower(?) " "AND lower(table_name) = lower(?)" ) + + +def test_columns_query_prefilters_with_lower_and_keeps_table_order(): + assert DdlRenderer.COLUMNS_QUERY == ( + "SELECT schema_name, table_name, column_name, is_nullable FROM duckdb_columns() " + "WHERE lower(database_name) = lower(?) AND lower(schema_name) = lower(?) " + "AND lower(table_name) = lower(?) ORDER BY column_index" + ) diff --git a/adapters/duckdb/tests/test_integration_duckdb_conformance.py b/adapters/duckdb/tests/test_integration_duckdb_conformance.py new file mode 100644 index 0000000..2c1c758 --- /dev/null +++ b/adapters/duckdb/tests/test_integration_duckdb_conformance.py @@ -0,0 +1,15 @@ +"""The engine-contract conformance suite against a real DuckLake (tests/smoke/duckdb-stack). + +The `adapter` and `schema` fixtures come from this directory's conftest.py. +Run the stack first: + docker compose -f tests/smoke/duckdb-stack/docker-compose.yml up -d --wait +""" +import pytest + +from continuo_engine_contract.conformance import IntrospectionConformance + +pytestmark = pytest.mark.integration + + +class TestDuckDBIntrospection(IntrospectionConformance): + pass From dcc65d0511c4f294e135a942916b6acbc2195c94 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:40:29 +0200 Subject: [PATCH 3/8] feat(duckdb): append LakeWarehouse declares AppendCapable: insert_arrow inside one gateway transaction, nothing for zero rows. Passes the append conformance cases on the DuckLake stack. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../application/warehouse.py | 18 ++++++++++++++++-- .../tests/test_adapter_duckdb_runtime.py | 18 ++++++++++++++++++ .../test_integration_duckdb_conformance.py | 6 +++++- 3 files changed, 39 insertions(+), 3 deletions(-) diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py index e5b1ead..c149216 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py @@ -14,8 +14,8 @@ from typing import TYPE_CHECKING, Any from continuo_engine_contract.port import ( # type: ignore[import-untyped] + AppendCapable, LiveColumn, - TableIntrospection, WarehouseAdapter, ) from continuo_engine_contract.sql import ensure_single_read # type: ignore[import-untyped] @@ -46,7 +46,7 @@ def _columns(raw: list[dict]) -> list[ColumnDefinition]: return [ColumnDefinition.from_mapping(entry) for entry in raw] -class LakeWarehouse(WarehouseAdapter, TableIntrospection): +class LakeWarehouse(WarehouseAdapter, AppendCapable): """WarehouseAdapter implemented over a LakeGateway.""" def __init__( @@ -215,6 +215,20 @@ def load(self, schema: str, table: str, data: "pa.Table") -> None: logger.info("loading %d row(s) into %s.%s", data.num_rows, schema, table) self._gateway.insert_arrow(target, data) + def append(self, schema: str, table: str, data: "pa.Table") -> None: + """Atomically insert every row of *data*; earlier rows stay (one transaction). + + The insert names *data*'s columns. A 0-row *data* opens no transaction, + so DuckLake records no snapshot for it. + """ + if not data.num_rows: + logger.info("append of 0 rows into %s.%s: nothing to write", schema, table) + return + target = QualifiedTable.of(schema, table) + with self._gateway.transaction(): + logger.info("appending %d row(s) into %s.%s", data.num_rows, schema, table) + self._gateway.insert_arrow(target, data) + def close(self) -> None: """Release the underlying connection.""" self._gateway.close() diff --git a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py index ae8b593..a39a7c7 100644 --- a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py +++ b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py @@ -116,3 +116,21 @@ def test_table_columns_returns_the_gateways_columns(warehouse, gateway): def test_table_columns_of_a_missing_table_raises_lookup_error(warehouse, gateway): with pytest.raises(LookupError, match=r"s\.t"): warehouse.table_columns("s", "t") + + +def test_append_inserts_in_one_transaction_without_clearing(warehouse, gateway): + data = pa.table({"id": [1, 2]}) + warehouse.append("s", "t", data) + assert gateway.calls == [("begin",), ("insert_arrow", T, data), ("commit",)] + + +def test_append_of_zero_rows_touches_nothing(warehouse, gateway): + warehouse.append("s", "t", pa.table({"id": pa.array([], pa.int32())})) + assert gateway.calls == [] + + +def test_append_rolls_back_when_the_insert_fails(warehouse, gateway): + gateway.fail_on["insert_arrow"] = RuntimeError("NOT NULL violated") + with pytest.raises(RuntimeError): + warehouse.append("s", "t", pa.table({"id": [1]})) + assert gateway.names() == ["begin", "insert_arrow", "rollback"] diff --git a/adapters/duckdb/tests/test_integration_duckdb_conformance.py b/adapters/duckdb/tests/test_integration_duckdb_conformance.py index 2c1c758..e735cb7 100644 --- a/adapters/duckdb/tests/test_integration_duckdb_conformance.py +++ b/adapters/duckdb/tests/test_integration_duckdb_conformance.py @@ -6,10 +6,14 @@ """ import pytest -from continuo_engine_contract.conformance import IntrospectionConformance +from continuo_engine_contract.conformance import AppendConformance, IntrospectionConformance pytestmark = pytest.mark.integration class TestDuckDBIntrospection(IntrospectionConformance): pass + + +class TestDuckDBAppend(AppendConformance): + pass From fc03db2ce8eb0618fa8ff4c20bb13f6f0bbe415c Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:42:54 +0200 Subject: [PATCH 4/8] feat(duckdb): MERGE from a registered Arrow view DdlRenderer.merge_from renders MERGE INTO the target on the key, overwriting the other columns of a matched row and inserting the rest, with no WHEN MATCHED branch when every column is in the key. LakeGateway.merge_arrow runs it from a registered view that is always unregistered; errors stay redacted. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../application/ports.py | 6 +++++ .../infrastructure/ddl.py | 26 +++++++++++++++++++ .../infrastructure/session.py | 10 +++++++ adapters/duckdb/tests/conftest.py | 3 +++ .../duckdb/tests/test_infra_duckdb_session.py | 23 ++++++++++++++++ .../tests/test_infra_duckdb_settings_ddl.py | 21 +++++++++++++++ 6 files changed, 89 insertions(+) diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py index 8d31523..7c05028 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py @@ -79,5 +79,11 @@ def delete_all(self, table: QualifiedTable) -> None: ... def insert_arrow(self, table: QualifiedTable, data: "pyarrow.Table") -> None: """Append *data* in its own column order.""" + @abstractmethod + def merge_arrow( + self, table: QualifiedTable, data: "pyarrow.Table", key: Sequence[str] + ) -> None: + """Insert *data*'s rows with a new *key*, overwrite rows whose key exists.""" + @abstractmethod def close(self) -> None: ... diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py index 2b2074e..4327b10 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py @@ -220,6 +220,32 @@ def insert_select(self, table: QualifiedTable, column_names: Sequence[str], sour cols = ", ".join(quote_identifier(name) for name in column_names) return f"INSERT INTO {self.table_ref(table)} ({cols}) SELECT {cols} FROM {quote_identifier(source)}" + def merge_from( + self, table: QualifiedTable, column_names: Sequence[str], key: Sequence[str], source: str + ) -> str: + """MERGE *source* into *table* on *key*: overwrite a matched row's other columns, insert the rest. + + With every column in the key there is nothing to overwrite, so the + statement has no WHEN MATCHED branch. Every column is named, so the + source's column order does not matter. + """ + key_set = set(key) + on = " AND ".join( + f"t.{quote_identifier(name)} = s.{quote_identifier(name)}" for name in key + ) + updates = [name for name in column_names if name not in key_set] + matched = "" + if updates: + matched = " WHEN MATCHED THEN UPDATE SET " + ", ".join( + f"{quote_identifier(name)} = s.{quote_identifier(name)}" for name in updates + ) + cols = ", ".join(quote_identifier(name) for name in column_names) + values = ", ".join(f"s.{quote_identifier(name)}" for name in column_names) + return ( + f"MERGE INTO {self.table_ref(table)} AS t USING {quote_identifier(source)} AS s " + f"ON ({on}){matched} WHEN NOT MATCHED THEN INSERT ({cols}) VALUES ({values})" + ) + @staticmethod def _partition(key: PartitionKey) -> str: column = quote_identifier(key.column.name) diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py index d6e460a..d44911a 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py @@ -287,6 +287,16 @@ def insert_arrow(self, table: QualifiedTable, data: "pa.Table") -> None: with contextlib.suppress(duckdb.Error): self._con.unregister(view) + def merge_arrow(self, table: QualifiedTable, data: "pa.Table", key: Sequence[str]) -> None: + view = f"__continuo_merge_{uuid.uuid4().hex}" + with self._guarded(): + self._con.register(view, data) + try: + self._run(self._renderer.merge_from(table, data.schema.names, key, view)) + finally: + with contextlib.suppress(duckdb.Error): + self._con.unregister(view) + def close(self) -> None: try: self._con.close() diff --git a/adapters/duckdb/tests/conftest.py b/adapters/duckdb/tests/conftest.py index 8c1cbad..b6eddc5 100644 --- a/adapters/duckdb/tests/conftest.py +++ b/adapters/duckdb/tests/conftest.py @@ -100,6 +100,9 @@ def delete_all(self, table): def insert_arrow(self, table, data): self._do("insert_arrow", table, data) + def merge_arrow(self, table, data, key): + self._do("merge_arrow", table, data, tuple(key)) + def close(self): self._do("close") diff --git a/adapters/duckdb/tests/test_infra_duckdb_session.py b/adapters/duckdb/tests/test_infra_duckdb_session.py index 4757b81..47dd1c1 100644 --- a/adapters/duckdb/tests/test_infra_duckdb_session.py +++ b/adapters/duckdb/tests/test_infra_duckdb_session.py @@ -487,3 +487,26 @@ def test_columns_errors_are_redacted(): with pytest.raises(duckdb.IOException) as caught: session.columns(T) assert PASSWORD not in str(caught.value) + + +def test_merge_arrow_registers_merges_and_unregisters(): + import pyarrow as pa + + session, connection = make_session() + session.merge_arrow(T, pa.table({"id": [1], "v": ["a"]}), ["id"]) + register, merge, unregister = connection.statements + view = register.removeprefix("register ") + assert view.startswith("__continuo_merge_") + assert merge == DdlRenderer("lake").merge_from(T, ["id", "v"], ["id"], view) + assert unregister == f"unregister {view}" + + +def test_merge_arrow_errors_are_redacted_and_the_view_is_still_unregistered(): + import pyarrow as pa + + session, connection = make_session() + connection.errors["MERGE"] = leaky(duckdb.IOException, PASSWORD) + with pytest.raises(duckdb.IOException) as caught: + session.merge_arrow(T, pa.table({"id": [1]}), ["id"]) + assert PASSWORD not in str(caught.value) + assert connection.statements[-1].startswith("unregister __continuo_merge_") diff --git a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py index a8ab913..7fffc63 100644 --- a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py +++ b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py @@ -301,3 +301,24 @@ def test_columns_query_prefilters_with_lower_and_keeps_table_order(): "WHERE lower(database_name) = lower(?) AND lower(schema_name) = lower(?) " "AND lower(table_name) = lower(?) ORDER BY column_index" ) + + +def test_merge_from_overwrites_non_key_columns_and_inserts_the_rest(): + assert R.merge_from(T, ["id", "v%", "Rate"], ["id"], "src") == ( + 'MERGE INTO "lake"."s"."t" AS t USING "src" AS s ON (t."id" = s."id")' + ' WHEN MATCHED THEN UPDATE SET "v%" = s."v%", "Rate" = s."Rate"' + ' WHEN NOT MATCHED THEN INSERT ("id", "v%", "Rate") VALUES (s."id", s."v%", s."Rate")' + ) + + +def test_merge_from_with_every_column_in_the_key_has_no_matched_branch(): + assert R.merge_from(T, ["a", "b"], ["a", "b"], "src") == ( + 'MERGE INTO "lake"."s"."t" AS t USING "src" AS s ON (t."a" = s."a" AND t."b" = s."b")' + ' WHEN NOT MATCHED THEN INSERT ("a", "b") VALUES (s."a", s."b")' + ) + + +def test_merge_from_quotes_a_reserved_word_key(): + assert R.merge_from(T, ["order", "x"], ["order"], 'we"ird').startswith( + 'MERGE INTO "lake"."s"."t" AS t USING "we""ird" AS s ON (t."order" = s."order")' + ) From f5135a8e4fc371cfcc41839060d63eed1900e574 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Fri, 9 Oct 2026 17:45:36 +0200 Subject: [PATCH 5/8] feat(duckdb): upsert through MERGE LakeWarehouse declares UpsertCapable: a MERGE from the registered Arrow data inside one gateway transaction, nothing for zero rows. ensure_table accepts the key and ignores it because DuckLake has no constraints. Passes the upsert conformance cases on the DuckLake stack. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- CHANGELOG.md | 4 +++ .../application/warehouse.py | 32 ++++++++++++++++-- .../tests/test_adapter_duckdb_runtime.py | 33 +++++++++++++++++++ .../test_integration_duckdb_conformance.py | 10 +++++- 4 files changed, 76 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8884755..5f3a9b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,10 @@ follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). column is in the key) and `table_columns`; `ensure_table` declares an upsert key as the table's `PRIMARY KEY`. `load` now shares the insert path (no behavior change). +- `continuo-duckdb-adapter`: `append` (one transaction), `upsert` (`MERGE` + from the registered Arrow data, one transaction) and `table_columns`. + `ensure_table` accepts and ignores an upsert key: DuckLake has no + constraints, so the key is guaranteed by the write path. ### Changed diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py index c149216..cb79848 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py @@ -10,12 +10,13 @@ import logging import time -from collections.abc import Callable +from collections.abc import Callable, Sequence from typing import TYPE_CHECKING, Any from continuo_engine_contract.port import ( # type: ignore[import-untyped] AppendCapable, LiveColumn, + UpsertCapable, WarehouseAdapter, ) from continuo_engine_contract.sql import ensure_single_read # type: ignore[import-untyped] @@ -46,7 +47,7 @@ def _columns(raw: list[dict]) -> list[ColumnDefinition]: return [ColumnDefinition.from_mapping(entry) for entry in raw] -class LakeWarehouse(WarehouseAdapter, AppendCapable): +class LakeWarehouse(WarehouseAdapter, AppendCapable, UpsertCapable): """WarehouseAdapter implemented over a LakeGateway.""" def __init__( @@ -181,6 +182,7 @@ def ensure_table( columns: list[dict[str, Any]], *, config: dict[str, Any], + primary_key: Sequence[str] = (), ) -> None: """Create the table if absent; layout is applied only when this call creates it. @@ -188,7 +190,12 @@ def ensure_table( existence check and the create share one transaction; a concurrent creator makes one side fail with a conflict, and the retry then finds the table and returns. + + *primary_key* is accepted and ignored: DuckLake rejects PRIMARY KEY and + UNIQUE constraints, so an upsert node's key is guaranteed by ``upsert`` + (the port's duplicate check, then MERGE), not by the table. """ + del primary_key defs = _columns(columns) layout = TableLayout.from_config(config, column_types(defs)) self.ensure_schema(schema) @@ -229,6 +236,27 @@ def append(self, schema: str, table: str, data: "pa.Table") -> None: logger.info("appending %d row(s) into %s.%s", data.num_rows, schema, table) self._gateway.insert_arrow(target, data) + def _upsert( + self, schema: str, table: str, data: "pa.Table", primary_key: Sequence[str] + ) -> None: + """MERGE *data* into ``schema.table`` on *primary_key*, in one transaction. + + DuckLake's MERGE does not detect a key repeated within *data* (it + inserts both rows of a new key and keeps the last of an existing one); + the port's ``upsert`` has already rejected that. A 0-row *data* opens + no transaction. + """ + if not data.num_rows: + logger.info("upsert of 0 rows into %s.%s: nothing to write", schema, table) + return + target = QualifiedTable.of(schema, table) + with self._gateway.transaction(): + logger.info( + "upserting %d row(s) into %s.%s on (%s)", + data.num_rows, schema, table, ", ".join(primary_key), + ) + self._gateway.merge_arrow(target, data, list(primary_key)) + def close(self) -> None: """Release the underlying connection.""" self._gateway.close() diff --git a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py index a39a7c7..b99a778 100644 --- a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py +++ b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py @@ -1,6 +1,7 @@ """Runtime use cases, against the recording FakeLakeGateway (no engine).""" import pyarrow as pa import pytest +from continuo_engine_contract.keys import DuplicateKeyError from continuo_engine_contract.port import LiveColumn from continuo_duckdb_adapter.application.ports import LakeConflictError @@ -134,3 +135,35 @@ def test_append_rolls_back_when_the_insert_fails(warehouse, gateway): with pytest.raises(RuntimeError): warehouse.append("s", "t", pa.table({"id": [1]})) assert gateway.names() == ["begin", "insert_arrow", "rollback"] + + +def test_ensure_table_accepts_and_ignores_a_primary_key(warehouse, gateway): + warehouse.ensure_table("s", "t", COLS, config={}, primary_key=("id",)) + keyed = list(gateway.calls) + gateway.calls.clear() + warehouse.ensure_table("s", "t", COLS, config={}) + assert gateway.calls == keyed + + +def test_upsert_merges_in_one_transaction(warehouse, gateway): + data = pa.table({"id": [1, 2], "v": ["a", "b"]}) + warehouse.upsert("s", "t", data, ["id"]) + assert gateway.calls == [("begin",), ("merge_arrow", T, data, ("id",)), ("commit",)] + + +def test_upsert_of_zero_rows_touches_nothing(warehouse, gateway): + warehouse.upsert("s", "t", pa.table({"id": pa.array([], pa.int32())}), ["id"]) + assert gateway.calls == [] + + +def test_upsert_with_a_repeated_key_touches_nothing(warehouse, gateway): + with pytest.raises(DuplicateKeyError): + warehouse.upsert("s", "t", pa.table({"id": [1, 1]}), ["id"]) + assert gateway.calls == [] + + +def test_upsert_rolls_back_when_the_merge_fails(warehouse, gateway): + gateway.fail_on["merge_arrow"] = RuntimeError("NOT NULL violated") + with pytest.raises(RuntimeError): + warehouse.upsert("s", "t", pa.table({"id": [1]}), ["id"]) + assert gateway.names() == ["begin", "merge_arrow", "rollback"] diff --git a/adapters/duckdb/tests/test_integration_duckdb_conformance.py b/adapters/duckdb/tests/test_integration_duckdb_conformance.py index e735cb7..c67b975 100644 --- a/adapters/duckdb/tests/test_integration_duckdb_conformance.py +++ b/adapters/duckdb/tests/test_integration_duckdb_conformance.py @@ -6,7 +6,11 @@ """ import pytest -from continuo_engine_contract.conformance import AppendConformance, IntrospectionConformance +from continuo_engine_contract.conformance import ( + AppendConformance, + IntrospectionConformance, + UpsertConformance, +) pytestmark = pytest.mark.integration @@ -17,3 +21,7 @@ class TestDuckDBIntrospection(IntrospectionConformance): class TestDuckDBAppend(AppendConformance): pass + + +class TestDuckDBUpsert(UpsertConformance): + pass From 0497fc47f2e9c48a44e1301aee2c2e70319463eb Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Sat, 10 Oct 2026 23:27:12 +0200 Subject: [PATCH 6/8] fix(duckdb): import conformance from the testing subpackage The engine-contract conformance suite moved to continuo_engine_contract.testing.conformance in #75; the duckdb integration test still imported the pre-move path and failed collection once #75 merged to main. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- adapters/duckdb/tests/test_integration_duckdb_conformance.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/adapters/duckdb/tests/test_integration_duckdb_conformance.py b/adapters/duckdb/tests/test_integration_duckdb_conformance.py index c67b975..3887eda 100644 --- a/adapters/duckdb/tests/test_integration_duckdb_conformance.py +++ b/adapters/duckdb/tests/test_integration_duckdb_conformance.py @@ -6,7 +6,7 @@ """ import pytest -from continuo_engine_contract.conformance import ( +from continuo_engine_contract.testing.conformance import ( AppendConformance, IntrospectionConformance, UpsertConformance, From 00bdb5749886fde6d09fc7844cae9d336875d63b Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Sun, 11 Oct 2026 10:04:02 +0200 Subject: [PATCH 7/8] fix(duckdb): guard merge_from against an empty key An empty key renders `ON ()`, invalid SQL. UpsertCapable.upsert already rejects an empty key through ensure_unique_key before any engine is called, so this is unreachable in the port flow; the guard makes the renderer self-defending if it is ever called directly. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../duckdb/continuo_duckdb_adapter/infrastructure/ddl.py | 6 ++++++ adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py | 5 +++++ 2 files changed, 11 insertions(+) diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py index 4327b10..d38cfb1 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py @@ -229,6 +229,12 @@ def merge_from( statement has no WHEN MATCHED branch. Every column is named, so the source's column order does not matter. """ + if not key: + # An empty key renders ``ON ()`` — invalid SQL. The port's + # ``ensure_unique_key`` already rejects an empty key before any + # engine is reached; this keeps the renderer self-defending if it + # is ever called directly. + raise ValueError("merge_from needs a non-empty key") key_set = set(key) on = " AND ".join( f"t.{quote_identifier(name)} = s.{quote_identifier(name)}" for name in key diff --git a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py index 7fffc63..0ab8047 100644 --- a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py +++ b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py @@ -322,3 +322,8 @@ def test_merge_from_quotes_a_reserved_word_key(): assert R.merge_from(T, ["order", "x"], ["order"], 'we"ird').startswith( 'MERGE INTO "lake"."s"."t" AS t USING "we""ird" AS s ON (t."order" = s."order")' ) + + +def test_merge_from_rejects_an_empty_key(): + with pytest.raises(ValueError, match="non-empty key"): + R.merge_from(T, ["id", "v"], [], "src") From 0525ffdbfece0d25db8dd4214b0acf677a656608 Mon Sep 17 00:00:00 2001 From: Simone Carolini Date: Sun, 11 Oct 2026 12:43:46 +0200 Subject: [PATCH 8/8] refactor(duckdb): absolute intra-package imports The duckdb adapter was the only package in the repo using relative intra-package imports (27 of them). The runtime, the engine contract and the postgres/trino adapters all use absolute imports. Standardise the duckdb adapter to match: every `from .` / `from ..` becomes `from continuo_duckdb_adapter...`. No behaviour change; imports resolve identically. Signed-off-by: Simone Carolini Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0137E9XmyUXhwPTRrtNfivDw --- .../duckdb/continuo_duckdb_adapter/adapter.py | 6 +++--- .../continuo_duckdb_adapter/application/ports.py | 6 +++--- .../application/warehouse.py | 8 ++++---- .../continuo_duckdb_adapter/domain/columns.py | 2 +- .../continuo_duckdb_adapter/domain/layout.py | 4 ++-- .../infrastructure/ddl.py | 8 ++++---- .../infrastructure/extensions.py | 4 ++-- .../infrastructure/session.py | 16 ++++++++-------- 8 files changed, 27 insertions(+), 27 deletions(-) diff --git a/adapters/duckdb/continuo_duckdb_adapter/adapter.py b/adapters/duckdb/continuo_duckdb_adapter/adapter.py index f6348a5..8e16125 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/adapter.py +++ b/adapters/duckdb/continuo_duckdb_adapter/adapter.py @@ -6,9 +6,9 @@ """ from __future__ import annotations -from .application.warehouse import LakeWarehouse -from .infrastructure.session import DuckLakeSession -from .infrastructure.settings import REQUIRED_ENV, DuckLakeSettings +from continuo_duckdb_adapter.application.warehouse import LakeWarehouse +from continuo_duckdb_adapter.infrastructure.session import DuckLakeSession +from continuo_duckdb_adapter.infrastructure.settings import REQUIRED_ENV, DuckLakeSettings class DuckDBAdapter(LakeWarehouse): diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py index 7c05028..316ca9f 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/ports.py @@ -8,9 +8,9 @@ from continuo_engine_contract.port import LiveColumn # type: ignore[import-untyped] -from ..domain.columns import ColumnDefinition -from ..domain.identifiers import Identifier, QualifiedTable -from ..domain.layout import TableLayout +from continuo_duckdb_adapter.domain.columns import ColumnDefinition +from continuo_duckdb_adapter.domain.identifiers import Identifier, QualifiedTable +from continuo_duckdb_adapter.domain.layout import TableLayout if TYPE_CHECKING: # pragma: no cover import pyarrow # type: ignore[import-untyped] diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py index cb79848..347fb5e 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py +++ b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py @@ -21,10 +21,10 @@ ) from continuo_engine_contract.sql import ensure_single_read # type: ignore[import-untyped] -from ..domain.columns import ColumnDefinition, column_types -from ..domain.identifiers import Identifier, QualifiedTable -from ..domain.layout import TableLayout -from .ports import LakeConflictError, LakeGateway +from continuo_duckdb_adapter.domain.columns import ColumnDefinition, column_types +from continuo_duckdb_adapter.domain.identifiers import Identifier, QualifiedTable +from continuo_duckdb_adapter.domain.layout import TableLayout +from continuo_duckdb_adapter.application.ports import LakeConflictError, LakeGateway if TYPE_CHECKING: # pragma: no cover import pyarrow as pa # type: ignore[import-untyped] diff --git a/adapters/duckdb/continuo_duckdb_adapter/domain/columns.py b/adapters/duckdb/continuo_duckdb_adapter/domain/columns.py index 9e831d6..f056e57 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/domain/columns.py +++ b/adapters/duckdb/continuo_duckdb_adapter/domain/columns.py @@ -8,7 +8,7 @@ from continuo_engine_contract.types import validate_column_type # type: ignore[import-untyped] -from .identifiers import Identifier +from continuo_duckdb_adapter.domain.identifiers import Identifier _TEMPORAL = re.compile(r"^(TIMESTAMP|DATE)\Z", re.IGNORECASE | re.ASCII) diff --git a/adapters/duckdb/continuo_duckdb_adapter/domain/layout.py b/adapters/duckdb/continuo_duckdb_adapter/domain/layout.py index e7bf965..8b97f28 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/domain/layout.py +++ b/adapters/duckdb/continuo_duckdb_adapter/domain/layout.py @@ -13,8 +13,8 @@ from continuo_engine_contract.config import ensure_known_keys # type: ignore[import-untyped] -from .columns import is_temporal_type -from .identifiers import Identifier +from continuo_duckdb_adapter.domain.columns import is_temporal_type +from continuo_duckdb_adapter.domain.identifiers import Identifier ENGINE = "duckdb" KNOWN_CONFIG_KEYS: tuple[str, ...] = ("partitioned_by", "sorted_by") diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py index d38cfb1..e716ebd 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/ddl.py @@ -11,10 +11,10 @@ import re from collections.abc import Sequence -from ..domain.columns import ColumnDefinition -from ..domain.identifiers import Identifier, QualifiedTable -from ..domain.layout import PartitionKey, SortKey -from .settings import DuckLakeSettings +from continuo_duckdb_adapter.domain.columns import ColumnDefinition +from continuo_duckdb_adapter.domain.identifiers import Identifier, QualifiedTable +from continuo_duckdb_adapter.domain.layout import PartitionKey, SortKey +from continuo_duckdb_adapter.infrastructure.settings import DuckLakeSettings _LIBPQ_BARE = re.compile(r"[A-Za-z0-9_.:/\-]+") diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/extensions.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/extensions.py index 860e69f..da46584 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/extensions.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/extensions.py @@ -10,8 +10,8 @@ import duckdb -from .settings import DuckLakeSettings -from .ddl import ( +from continuo_duckdb_adapter.infrastructure.settings import DuckLakeSettings +from continuo_duckdb_adapter.infrastructure.ddl import ( disable_extension_autoinstall, load_extension, set_extension_directory, diff --git a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py index d44911a..98dbabc 100644 --- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py +++ b/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py @@ -21,11 +21,11 @@ import duckdb from continuo_engine_contract.port import LiveColumn # type: ignore[import-untyped] -from ..application.ports import LakeConflictError, LakeGateway -from ..domain.columns import ColumnDefinition -from ..domain.identifiers import Identifier, QualifiedTable -from ..domain.layout import TableLayout -from .ddl import ( +from continuo_duckdb_adapter.application.ports import LakeConflictError, LakeGateway +from continuo_duckdb_adapter.domain.columns import ColumnDefinition +from continuo_duckdb_adapter.domain.identifiers import Identifier, QualifiedTable +from continuo_duckdb_adapter.domain.layout import TableLayout +from continuo_duckdb_adapter.infrastructure.ddl import ( BEGIN, BEGIN_READ_ONLY, COMMIT, @@ -41,9 +41,9 @@ set_temp_directory, use_catalog, ) -from .extensions import extensions_to_load -from .passfile import Passfile, open_passfile -from .settings import DuckLakeSettings +from continuo_duckdb_adapter.infrastructure.extensions import extensions_to_load +from continuo_duckdb_adapter.infrastructure.passfile import Passfile, open_passfile +from continuo_duckdb_adapter.infrastructure.settings import DuckLakeSettings if TYPE_CHECKING: # pragma: no cover import pyarrow as pa # type: ignore[import-untyped]