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/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 49a74d6..316ca9f 100644
--- a/adapters/duckdb/continuo_duckdb_adapter/application/ports.py
+++ b/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: ...
diff --git a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py b/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py
index 6ef39e3..347fb5e 100644
--- a/adapters/duckdb/continuo_duckdb_adapter/application/warehouse.py
+++ b/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,6 +164,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,
@@ -166,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.
 
@@ -173,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)
@@ -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()
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 7bbb4f4..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_.:/\-]+")
 
@@ -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)
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 99f3f31..98dbabc 100644
--- a/adapters/duckdb/continuo_duckdb_adapter/infrastructure/session.py
+++ b/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()
diff --git a/adapters/duckdb/tests/conftest.py b/adapters/duckdb/tests/conftest.py
index 2548653..b6eddc5 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)
 
@@ -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")
 
diff --git a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py
index 0ab23e8..b99a778 100644
--- a/adapters/duckdb/tests/test_adapter_duckdb_runtime.py
+++ b/adapters/duckdb/tests/test_adapter_duckdb_runtime.py
@@ -1,6 +1,8 @@
 """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
 from continuo_duckdb_adapter.domain.columns import ColumnDefinition
@@ -104,3 +106,64 @@ 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")
+
+
+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"]
+
+
+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_infra_duckdb_session.py b/adapters/duckdb/tests/test_infra_duckdb_session.py
index f3f0ce2..47dd1c1 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,56 @@ 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)
+
+
+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 29d8a46..0ab8047 100644
--- a/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py
+++ b/adapters/duckdb/tests/test_infra_duckdb_settings_ddl.py
@@ -293,3 +293,37 @@ 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"
+    )
+
+
+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")'
+    )
+
+
+def test_merge_from_rejects_an_empty_key():
+    with pytest.raises(ValueError, match="non-empty key"):
+        R.merge_from(T, ["id", "v"], [], "src")
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..3887eda
--- /dev/null
+++ b/adapters/duckdb/tests/test_integration_duckdb_conformance.py
@@ -0,0 +1,27 @@
+"""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.testing.conformance import (
+    AppendConformance,
+    IntrospectionConformance,
+    UpsertConformance,
+)
+
+pytestmark = pytest.mark.integration
+
+
+class TestDuckDBIntrospection(IntrospectionConformance):
+    pass
+
+
+class TestDuckDBAppend(AppendConformance):
+    pass
+
+
+class TestDuckDBUpsert(UpsertConformance):
+    pass
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"),
+    ]