feat(duckdb): append and upsert

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
 

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):

adapters/duckdb/continuo_duckdb_adapter/application/ports.py

@@ -6,9 +6,11 @@
 from contextlib import AbstractContextManager
 from typing import TYPE_CHECKING
 
-from ..domain.columns import ColumnDefinition
-from ..domain.identifiers import Identifier, QualifiedTable
-from ..domain.layout import TableLayout
+from continuo_engine_contract.port import LiveColumn  # type: ignore[import-untyped]
+
+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]
@@ -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: ...
 
@@ -73,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: ...

adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py

@@ -10,16 +10,21 @@
 
 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 WarehouseAdapter  # type: ignore[import-untyped]
+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]
 
-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]
@@ -42,7 +47,7 @@ def _columns(raw: list[dict]) -> list[ColumnDefinition]:
     return [ColumnDefinition.from_mapping(entry) for entry in raw]
 
 
-class LakeWarehouse(WarehouseAdapter):
+class LakeWarehouse(WarehouseAdapter, AppendCapable, UpsertCapable):
     """WarehouseAdapter implemented over a LakeGateway."""
 
     def __init__(
@@ -159,21 +164,38 @@ 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,
         table: str,
         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.
 
         Validation comes first, so a bad config or type emits no statement. The
         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)
@@ -200,6 +222,41 @@ 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 _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()

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)
 

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")

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_.:/\-]+")
 
@@ -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
 
@@ -212,6 +220,38 @@ 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.
+        """
+        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
+        )
+        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)

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,

adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py

@@ -19,12 +19,13 @@
 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
-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,
@@ -40,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]
@@ -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))
 
@@ -273,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()

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)
 
@@ -95,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")