test(table-catalog): automate DuckDB REST conformance (#6750)

* test(table-catalog): automate DuckDB REST conformance

* fix(table-catalog): protect DuckDB smoke tables

docs/architecture/s3-tables-support-matrix.md

@@ -43,7 +43,7 @@ catalog extension.
 | PyIceberg | Automated | Creates namespace and table, appends rows, reloads, scans, probes metadata-location, refs, views, maintenance, diagnostics, and optional catalog-vended table credentials with an exact-prefix data-plane scope check. |
 | Spark Iceberg REST catalog | Manual/live harness | RustFS can generate pinned Spark/Iceberg package inputs, REST catalog properties, SQL, run commands, expected `row_count=2`, and a CI opt-in gate for namespace creation, table creation, append, refresh, count, and cleanup. Live Spark execution and commit-conflict probing are still manual validation items unless explicitly enabled in the runner. |
 | Trino Iceberg REST catalog | Manual/live harness | RustFS can generate catalog properties and a read-only `SELECT COUNT(*)` command for a table created by PyIceberg or Spark. Write compatibility is not claimed. |
-| DuckDB Iceberg | Manual/live harness | RustFS can generate the read-only `iceberg_scan` path using an operator-supplied current metadata location and a generic signed Iceberg REST Catalog profile for `/iceberg` or `/_iceberg`. The REST profile disables staged create, post-create metadata updates, multi-table commit, client-side file removal, and purge-on-drop. Write and commit compatibility remain not claimed until repeatable live evidence is automated. |
+| DuckDB Iceberg 1.5.5 | Automated | `duckdb_smoke.py` verifies the metadata-location read path and generic REST Catalog single-table create, insert, update, delete, merge, schema evolution, snapshots, concurrent writers, normal drop, PyIceberg cross-read, `/iceberg` with `s3` signing, and `/_iceberg` with `s3tables` signing. Staged create, purge-on-drop, and format v3 are verified as fail-closed boundaries. DuckDB's endpoint-disabled two-table mode is exercised without claiming cross-table atomicity. AWS `ENDPOINT_TYPE S3_TABLES` and catalog-vended credential integration are not claimed. |
 | StarRocks Iceberg REST catalog | Documented, not automated | External catalog read-path reference only. Write compatibility is not claimed. |
 | Databend | Manual/live harness | RustFS can generate an S3 stage read probe for table data files. RustFS does not claim Databend Iceberg REST Catalog integration yet. |
 | Snowflake Open Catalog / Iceberg integrations | Generated harness | RustFS can generate an operator-adapted external volume/catalog SQL template. Live RustFS interoperability is not claimed. |
@@ -52,10 +52,10 @@ catalog extension.
 
 | Area | Status | Current RustFS claim |
 |---|---|---|
-| Live conformance evidence | Manual/live harness | `engine_compatibility.py --print-live-evidence-schema` defines the required evidence schema and claim promotion boundaries. `pyiceberg_smoke.py --live-evidence-output` writes a validated PyIceberg evidence record after a successful live smoke run. |
+| Live conformance evidence | Automated for PyIceberg and DuckDB | `engine_compatibility.py --print-live-evidence-schema` defines the required evidence schema and claim promotion boundaries. `pyiceberg_smoke.py --live-evidence-output` and `duckdb_smoke.py --live-evidence-output` write validated client evidence records after successful live smoke runs. |
 | Production operations guide | Generated harness | `engine_compatibility.py --print-operations-guide` records command, evidence, pass criteria, and fail-closed signals for live conformance, durable backing cutover, maintenance, recovery, permissions, credential vending, and unsupported-claim governance. |
 | Vendor compatibility gap audit | Generated harness | `engine_compatibility.py --print-vendor-audit` records provider source URLs, catalog path and warehouse shapes, signing/auth models, error/permission/maintenance validation categories, and not-claimed boundaries for AWS S3 Tables, MinIO AIStor Tables, Cloudflare R2 Data Catalog, and Alibaba OSS Tables. |
-| Client claim promotion | Documented, not automated | PyIceberg remains the automated claim. Spark can be promoted only with recorded manual/live evidence; Trino and DuckDB read probes do not promote write compatibility; Snowflake and vendor profiles remain reference-only without repeatable live evidence. |
+| Client claim promotion | Automated for scoped clients | PyIceberg and DuckDB claims remain bounded by their repeatable smoke entrypoints and recorded versions. Spark can be promoted only with recorded manual/live evidence; Trino remains read-only; Snowflake and vendor profiles remain reference-only without repeatable live evidence. |
 
 ## Catalog API Matrix
 
@@ -242,6 +242,7 @@ compatibility claims:
 ```bash
 python3 scripts/table-catalog/test_pyiceberg_smoke.py
 python3 scripts/table-catalog/test_engine_compatibility.py
+python3 scripts/table-catalog/test_duckdb_smoke.py
 python3 scripts/table-catalog/test_failure_coverage.py
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-client-matrix
 python3 scripts/table-catalog/pyiceberg_smoke.py --print-engine-compatibility
@@ -305,9 +306,9 @@ Use conservative release wording that matches the matrix.
 Acceptable wording:
 
 > RustFS includes a core Iceberg REST Catalog-based S3 Tables implementation
-> with PyIceberg smoke coverage, table-aware S3 data-plane policy checks,
+> with PyIceberg and DuckDB smoke coverage, table-aware S3 data-plane policy checks,
 > controlled maintenance, catalog recovery diagnostics, manual conformance
-> input for Spark, Trino, DuckDB, Databend, and Snowflake, production-failure
+> input for Spark, Trino, Databend, and Snowflake, production-failure
 > probe harnesses, disaster-recovery and scale/fault rehearsal probes, and a
 > machine-readable production operations evidence guide.
 

scripts/table-catalog/README.md

@@ -163,10 +163,10 @@ python3 scripts/table-catalog/engine_compatibility.py --print-live-evidence-sche
 ```
 
 Use these outputs when updating release notes, PR descriptions, or follow-up
-work items. They are intentionally conservative: only PyIceberg is automated by
+work items. They are intentionally conservative: PyIceberg and DuckDB have
-this script today. Spark has a repeatable manual/live harness with pinned
+separate automated smoke entrypoints. Spark has a repeatable manual/live harness
-client package inputs, generated configuration, generated SQL, expected
+with pinned client package inputs, generated configuration, generated SQL,
-results, and a CI opt-in gate. Trino, DuckDB, Databend, and Snowflake now have
+expected results, and a CI opt-in gate. Trino, Databend, and Snowflake have
 generated manual probe inputs, but they remain opt-in and do not promote write
 or full vendor interoperability claims.
 
@@ -311,7 +311,7 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces:
 | PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, metadata-location, refs, views, maintenance, diagnostics, optional catalog-vended table credentials with exact-prefix data-plane scope probe |
 | Spark Iceberg REST catalog | Manual/live harness | pinned Spark and Iceberg package inputs, configuration, SQL, run command, expected row count, and cleanup can be generated for a running RustFS endpoint; CI execution is opt-in |
 | Trino Iceberg REST catalog | Manual/live read probe | generated catalog properties and a read-only SELECT probe for a table created by PyIceberg or Spark; no write compatibility claim yet |
-| DuckDB Iceberg | Manual/live harness | generated metadata-location read SQL plus generic Iceberg REST Catalog attach SQL for `/iceberg` with `s3` signing or `/_iceberg` with `s3tables` signing; write compatibility still requires repeatable live evidence |
+| DuckDB Iceberg | Automated smoke target | metadata-location read plus generic REST Catalog single-table DDL, DML, schema evolution, snapshots, `/iceberg` and `/_iceberg` signing, fail-closed unsupported boundaries, endpoint-disabled non-atomic multi-table mode, concurrent writers, and PyIceberg cross-read |
 | StarRocks Iceberg REST catalog | Documented, not automated | external catalog read-path reference only |
 | Databend | Manual/live S3 stage probe | generated S3 stage read probe for table data files; Iceberg REST catalog integration is not claimed |
 | Snowflake/Open Catalog integrations | Manual reference probe | generated external volume/catalog SQL template; live RustFS interoperability is not claimed |
@@ -538,8 +538,35 @@ keep DuckDB within RustFS's claimed single-table REST surface. Do not replace
 the explicit endpoint with DuckDB `ENDPOINT_TYPE S3_TABLES`; that shortcut is
 for AWS S3 Tables endpoint and warehouse shapes.
 
-This profile is generated conformance input. It does not promote DuckDB write
+Run the repeatable DuckDB 1.5.5 smoke against an already running RustFS:
-compatibility until the repeatable live smoke records passing evidence.
+
+```bash
+python3 scripts/table-catalog/duckdb_smoke.py \
+  --duckdb /path/to/duckdb \
+  --endpoint http://127.0.0.1:9000 \
+  --bucket rustfs-duckdb-smoke \
+  --namespace duckdb_smoke \
+  --table events \
+  --cleanup \
+  --rustfs-build rustfs-v1.0.0-rc.4 \
+  --git-sha "$(git rev-parse HEAD)" \
+  --catalog-backing object \
+  --live-evidence-output /tmp/rustfs-duckdb-live-evidence.json
+```
+
+The script requires the same PyIceberg, PyArrow, and boto3 dependencies as the
+PyIceberg smoke because it verifies both cross-engine directions. It creates an
+isolated namespace, keeps the final verified table at two rows for the shared
+evidence contract, and cleans all smoke tables only when `--cleanup` is set.
+It refuses to remove pre-existing suffixed smoke tables unless `--replace` is
+set explicitly. Cleanup preserves a namespace that existed before the run.
+
+The automated claim is limited to DuckDB 1.5.5, static S3 credentials, and the
+single-table scenarios exercised by this script. It does not claim DuckDB's AWS
+`S3_TABLES` shortcut, staged create, purge-on-drop, format v3, multi-table
+atomicity, or catalog-vended credential integration. The smoke verifies that
+DuckDB can run a two-table transaction with its multi-table commit endpoint
+disabled, but each table remains an independent RustFS commit.
 
 ## Spark Manual/Live Harness
 
@@ -629,8 +656,9 @@ engines that are not run by default in RustFS CI:
 - Trino: catalog properties and a read-only `SELECT COUNT(*)` command for a
   table already created by PyIceberg or Spark. Trino write compatibility is not
   claimed.
-- DuckDB: `httpfs` and `iceberg` SQL using an operator-supplied current Iceberg
+- DuckDB: a legacy `httpfs` and `iceberg` read probe using an operator-supplied
-  metadata location. DuckDB write and commit compatibility are not claimed.
+  current Iceberg metadata location. The separate `duckdb_smoke.py` entrypoint
+  owns the automated generic REST Catalog single-table read/write claim.
 - Databend: an S3 stage read probe for Parquet data files under the table
   warehouse. Databend Iceberg REST Catalog integration is not claimed.
 - Snowflake: an operator-adapted external volume/catalog integration SQL

scripts/table-catalog/duckdb_smoke.py

@@ -0,0 +1,653 @@
+#!/usr/bin/env python3
+"""DuckDB Iceberg REST Catalog smoke test for RustFS S3 Tables."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import time
+import urllib.parse
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import engine_compatibility
+import pyiceberg_smoke
+
+
+DEFAULT_DUCKDB_VERSION = engine_compatibility.DEFAULT_DUCKDB_VERSION
+IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,47}$")
+
+
+@dataclass(frozen=True)
+class DuckDBExecution:
+    returncode: int
+    batches: list[list[dict[str, Any]]]
+    stdout: str
+    stderr: str
+
+
+@dataclass(frozen=True)
+class DuckDBSmokeResult:
+    client_version: str
+    metadata_location: str
+    row_count: int
+    cleanup_result: str
+    checks: dict[str, str]
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    run_id = str(int(time.time()))
+    parser = argparse.ArgumentParser(description="Run DuckDB Iceberg REST Catalog conformance against RustFS.")
+    parser.add_argument("--endpoint", default=os.getenv("RUSTFS_ENDPOINT", "http://127.0.0.1:9000"))
+    parser.add_argument("--access-key", default=os.getenv("RUSTFS_ACCESS_KEY", "rustfsadmin"))
+    parser.add_argument("--secret-key", default=os.getenv("RUSTFS_SECRET_KEY", "rustfsadmin"))
+    parser.add_argument("--region", default=os.getenv("RUSTFS_REGION", "us-east-1"))
+    parser.add_argument("--bucket", default=os.getenv("RUSTFS_TABLE_BUCKET", "rustfs-duckdb-smoke"))
+    parser.add_argument("--namespace", default=os.getenv("RUSTFS_TABLE_NAMESPACE", f"duckdb_smoke_{run_id}"))
+    parser.add_argument("--table", default=os.getenv("RUSTFS_TABLE_NAME", "events"))
+    parser.add_argument("--duckdb", default=os.getenv("DUCKDB_BIN", "duckdb"))
+    parser.add_argument("--duckdb-version", default=DEFAULT_DUCKDB_VERSION)
+    parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "60")))
+    parser.add_argument("--cleanup", action="store_true")
+    parser.add_argument("--replace", action="store_true", help="Drop existing smoke tables with matching identifiers first.")
+    parser.add_argument("--insecure", action="store_true")
+    parser.add_argument("--live-evidence-output")
+    parser.add_argument("--rustfs-build", default=os.getenv("RUSTFS_BUILD", "operator-recorded"))
+    parser.add_argument("--git-sha", default=os.getenv("RUSTFS_GIT_SHA", "operator-recorded"))
+    parser.add_argument("--catalog-backing", default=os.getenv("RUSTFS_TABLE_CATALOG_BACKING", "operator-recorded"))
+    parser.add_argument("--operator", default=os.getenv("USER", "operator-recorded"))
+    parser.add_argument("--run-timestamp-utc")
+    args = parser.parse_args(argv)
+    for label, value in [("namespace", args.namespace), ("table", args.table)]:
+        if not IDENTIFIER_RE.fullmatch(value):
+            parser.error(f"{label} must start with a letter and contain at most 48 ASCII letters, digits, or underscores")
+    return args
+
+
+def duckdb_path(value: str) -> str:
+    resolved = shutil.which(value)
+    if resolved is None:
+        raise RuntimeError(f"DuckDB executable was not found: {value}")
+    return resolved
+
+
+def duckdb_client_version(executable: str, timeout: float) -> str:
+    process = subprocess.run(
+        [executable, "-csv", "-noheader", "-c", "SELECT version();"],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+    if process.returncode != 0:
+        raise RuntimeError(f"DuckDB version probe failed: {process.stderr.strip()}")
+    version = process.stdout.strip().removeprefix("v")
+    if not version:
+        raise RuntimeError("DuckDB version probe returned an empty version")
+    return version
+
+
+def parse_duckdb_json(stdout: str) -> list[list[dict[str, Any]]]:
+    batches: list[list[dict[str, Any]]] = []
+    decoder = json.JSONDecoder()
+    offset = 0
+    while offset < len(stdout):
+        while offset < len(stdout) and stdout[offset].isspace():
+            offset += 1
+        if offset == len(stdout):
+            break
+        value, offset = decoder.raw_decode(stdout, offset)
+        if not isinstance(value, list) or any(not isinstance(row, dict) for row in value):
+            raise RuntimeError("DuckDB JSON output did not contain row objects")
+        batches.append(value)
+    return batches
+
+
+def run_duckdb(executable: str, sql: str, timeout: float) -> DuckDBExecution:
+    process = subprocess.run(
+        [executable, "-json", "-c", sql],
+        capture_output=True,
+        text=True,
+        timeout=timeout,
+        check=False,
+    )
+    batches = parse_duckdb_json(process.stdout) if process.returncode == 0 else []
+    return DuckDBExecution(process.returncode, batches, process.stdout, process.stderr)
+
+
+def require_duckdb_success(execution: DuckDBExecution, label: str) -> None:
+    if execution.returncode != 0:
+        message = execution.stderr.strip() or execution.stdout.strip()
+        raise RuntimeError(f"DuckDB {label} failed: {message}")
+
+
+def require_duckdb_error(execution: DuckDBExecution, label: str, expected: str) -> None:
+    if execution.returncode == 0:
+        raise RuntimeError(f"DuckDB {label} unexpectedly succeeded")
+    message = f"{execution.stdout}\n{execution.stderr}"
+    if expected not in message:
+        raise RuntimeError(f"DuckDB {label} failed without expected error {expected!r}: {message.strip()}")
+
+
+def batches_with_column(execution: DuckDBExecution, column: str) -> list[list[dict[str, Any]]]:
+    return [batch for batch in execution.batches if batch and column in batch[0]]
+
+
+def table_name(base: str, suffix: str) -> str:
+    return f"{base}_{suffix}"
+
+
+def table_identifier(catalog: str, namespace: str, table: str) -> str:
+    return ".".join(
+        [
+            engine_compatibility.quote_double_identifier(catalog),
+            engine_compatibility.quote_double_identifier(namespace),
+            engine_compatibility.quote_double_identifier(table),
+        ]
+    )
+
+
+def profile_sql(
+    args: argparse.Namespace,
+    *,
+    catalog: str,
+    table: str,
+    rest_path: str = "/iceberg",
+    signing_name: str = "s3",
+) -> str:
+    return engine_compatibility.duckdb_rest_catalog_sql(
+        endpoint=args.endpoint,
+        warehouse=args.bucket,
+        access_key=args.access_key,
+        secret_key=args.secret_key,
+        region=args.region,
+        catalog_name=catalog,
+        namespace=args.namespace,
+        table=table,
+        rest_path=rest_path,
+        rest_signing_name=signing_name,
+    )
+
+
+def attach_sql(
+    args: argparse.Namespace,
+    *,
+    catalog: str,
+    rest_path: str,
+    signing_name: str,
+    compatibility_options: bool,
+    purge_requested: bool = False,
+) -> str:
+    options = [
+        "  TYPE iceberg",
+        f"  ENDPOINT {engine_compatibility.sql_string(f'{args.endpoint.rstrip('/')}{rest_path}')}",
+        "  AUTHORIZATION_TYPE 'sigv4'",
+        "  SECRET 'rustfs_s3'",
+        f"  SIGV4_REGION {engine_compatibility.sql_string(args.region)}",
+        f"  SIGV4_SERVICE {engine_compatibility.sql_string(signing_name)}",
+        "  ACCESS_DELEGATION_MODE 'none'",
+    ]
+    if compatibility_options:
+        options.extend(
+            [
+                "  STAGE_CREATE_TABLES false",
+                "  SKIP_CREATE_TABLE_METADATA_UPDATES true",
+                "  DISABLE_MULTI_TABLE_COMMIT true",
+                "  REMOVE_FILES_ON_DELETE false",
+                f"  PURGE_REQUESTED {'true' if purge_requested else 'false'}",
+                "  SUPPORT_NESTED_NAMESPACES false",
+            ]
+        )
+    rendered_options = ",\n".join(options)
+    return (
+        f"ATTACH {engine_compatibility.sql_string(args.bucket)} "
+        f"AS {engine_compatibility.quote_double_identifier(catalog)} (\n{rendered_options}\n);\n"
+    )
+
+
+def canonical_positive_sql(args: argparse.Namespace, seed_table: str, write_table: str, purge_table: str, drop_table: str) -> str:
+    catalog = "rustfs_duckdb"
+    namespace = ".".join(
+        [
+            engine_compatibility.quote_double_identifier(catalog),
+            engine_compatibility.quote_double_identifier(args.namespace),
+        ]
+    )
+    write_identifier = table_identifier(catalog, args.namespace, write_table)
+    purge_identifier = table_identifier(catalog, args.namespace, purge_table)
+    drop_identifier = table_identifier(catalog, args.namespace, drop_table)
+    return profile_sql(args, catalog=catalog, table=seed_table) + "\n".join(
+        [
+            f"CREATE SCHEMA IF NOT EXISTS {namespace};",
+            f"CREATE TABLE {write_identifier} (id BIGINT, payload VARCHAR);",
+            f"INSERT INTO {write_identifier} VALUES (10, 'ten'), (20, 'twenty');",
+            f"UPDATE {write_identifier} SET payload = 'TWENTY' WHERE id = 20;",
+            f"DELETE FROM {write_identifier} WHERE id = 10;",
+            f"ALTER TABLE {write_identifier} ADD COLUMN category VARCHAR;",
+            f"INSERT INTO {write_identifier} VALUES (30, 'thirty', 'new');",
+            f"MERGE INTO {write_identifier} AS target",
+            "USING (VALUES (20, 'twenty-merged', 'merged'), (40, 'forty', 'inserted')) AS source(id, payload, category)",
+            "ON target.id = source.id",
+            "WHEN MATCHED THEN UPDATE SET payload = source.payload, category = source.category",
+            "WHEN NOT MATCHED THEN INSERT (id, payload, category) VALUES (source.id, source.payload, source.category);",
+            f"DELETE FROM {write_identifier} WHERE id = 30;",
+            f"SELECT id, payload, category FROM {write_identifier} ORDER BY id;",
+            f"SELECT count(*) AS snapshot_count FROM iceberg_snapshots({write_identifier});",
+            f"CREATE TABLE {drop_identifier} (id BIGINT);",
+            f"INSERT INTO {drop_identifier} VALUES (1);",
+            f"DROP TABLE {drop_identifier};",
+            f"CREATE TABLE {purge_identifier} (id BIGINT);",
+            f"INSERT INTO {purge_identifier} VALUES (1);",
+            f"SELECT count(*) AS row_count FROM {write_identifier};",
+        ]
+    ) + "\n"
+
+
+def alias_sql(args: argparse.Namespace, write_table: str) -> str:
+    catalog = "rustfs_compat"
+    identifier = table_identifier(catalog, args.namespace, write_table)
+    return profile_sql(args, catalog=catalog, table=write_table, rest_path="/_iceberg", signing_name="s3tables") + "\n".join(
+        [
+            f"INSERT INTO {identifier} VALUES (50, 'fifty', 'compat');",
+            f"SELECT count(*) AS alias_row_count FROM {identifier};",
+            f"DELETE FROM {identifier} WHERE id = 50;",
+            f"SELECT count(*) AS alias_final_row_count FROM {identifier};",
+        ]
+    ) + "\n"
+
+
+def concurrent_insert_sql(args: argparse.Namespace, catalog: str, write_table: str, row_id: int) -> str:
+    identifier = table_identifier(catalog, args.namespace, write_table)
+    return profile_sql(args, catalog=catalog, table=write_table) + f"INSERT INTO {identifier} VALUES ({row_id}, 'writer-{row_id}', 'concurrent');\n"
+
+
+def multi_table_sql(args: argparse.Namespace, seed_table: str, write_table: str, purge_table: str) -> str:
+    catalog = "multi_table"
+    first = table_identifier(catalog, args.namespace, write_table)
+    second = table_identifier(catalog, args.namespace, purge_table)
+    return profile_sql(args, catalog=catalog, table=seed_table) + "\n".join(
+        [
+            "BEGIN TRANSACTION;",
+            f"INSERT INTO {first} VALUES (999, 'multi-a', 'non-atomic');",
+            f"INSERT INTO {second} VALUES (999);",
+            "COMMIT;",
+        ]
+    ) + "\n"
+
+
+def negative_sql(args: argparse.Namespace, *, kind: str, seed_table: str, write_table: str, purge_table: str) -> str:
+    bootstrap = f"bootstrap_{kind}"
+    sql = profile_sql(args, catalog=bootstrap, table=seed_table)
+    sql += f"DETACH {engine_compatibility.quote_double_identifier(bootstrap)};\n"
+    if kind == "stage-create":
+        catalog = "stage_default"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=False,
+        )
+        sql += f"CREATE TABLE {table_identifier(catalog, args.namespace, table_name(args.table, 'stage'))} (id BIGINT);\n"
+        return sql
+    if kind == "purge":
+        catalog = "purge_requested"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=True,
+            purge_requested=True,
+        )
+        sql += f"DROP TABLE {table_identifier(catalog, args.namespace, purge_table)};\n"
+        return sql
+    if kind == "format-v3":
+        catalog = "format_v3"
+        sql += attach_sql(
+            args,
+            catalog=catalog,
+            rest_path="/iceberg",
+            signing_name="s3",
+            compatibility_options=True,
+        )
+        identifier = table_identifier(catalog, args.namespace, table_name(args.table, "v3"))
+        sql += f"CREATE TABLE {identifier} (id BIGINT) WITH ('format-version' = '3');\n"
+        return sql
+    raise ValueError(f"unknown negative DuckDB smoke kind: {kind}")
+
+
+def pyiceberg_args(args: argparse.Namespace) -> argparse.Namespace:
+    return argparse.Namespace(
+        profile="rustfs",
+        endpoint=args.endpoint,
+        access_key=args.access_key,
+        secret_key=args.secret_key,
+        region=args.region,
+        bucket=args.bucket,
+        warehouse=None,
+        table_bucket=None,
+        account_id="000000000000",
+        warehouse_name=None,
+        catalog_uri=None,
+        namespace=args.namespace,
+        table=args.table,
+        catalog_name="rustfs_duckdb_pyiceberg",
+        rest_path="/iceberg",
+        rest_signing_name="s3",
+        require_vended_credentials=False,
+        timeout=args.timeout,
+        insecure=args.insecure,
+    )
+
+
+def prepare_smoke_tables(catalog: Any, namespace: str, tables: list[str], replace: bool) -> None:
+    existing = [table for table in tables if pyiceberg_smoke.table_exists(catalog, (namespace, table))]
+    if existing and not replace:
+        identifiers = ", ".join(f"{namespace}.{table}" for table in existing)
+        raise RuntimeError(f"DuckDB smoke tables already exist: {identifiers}; rerun with --replace to remove them")
+    for table in existing:
+        catalog.drop_table((namespace, table))
+
+
+def seed_pyiceberg_table(catalog: Any, args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps, table: str) -> None:
+    identifier = (args.namespace, table)
+    schema = deps.pyarrow.schema(
+        [
+            deps.pyarrow.field("id", deps.pyarrow.int64(), nullable=False),
+            deps.pyarrow.field("payload", deps.pyarrow.string(), nullable=False),
+        ]
+    )
+    created = catalog.create_table(identifier, schema=schema)
+    created.append(
+        deps.pyarrow.Table.from_pylist(
+            [{"id": 1, "payload": "alpha"}, {"id": 2, "payload": "beta"}],
+            schema=schema,
+        )
+    )
+
+
+def pyiceberg_rows(catalog: Any, namespace: str, table: str) -> list[dict[str, Any]]:
+    rows = catalog.load_table((namespace, table)).scan().to_arrow().to_pylist()
+    return sorted(rows, key=lambda row: row["id"])
+
+
+def run_concurrent_inserts(executable: str, args: argparse.Namespace, write_table: str) -> str:
+    probes = [("writer_a", 60), ("writer_b", 70)]
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        futures = [
+            executor.submit(run_duckdb, executable, concurrent_insert_sql(args, catalog, write_table, row_id), args.timeout)
+            for catalog, row_id in probes
+        ]
+    executions = [future.result() for future in futures]
+    retried = False
+    for (catalog, row_id), execution in zip(probes, executions, strict=True):
+        if execution.returncode == 0:
+            continue
+        error_text = f"{execution.stdout}\n{execution.stderr}".lower()
+        if not any(marker in error_text for marker in ["409", "conflict", "version token"]):
+            require_duckdb_success(execution, f"concurrent writer {row_id}")
+        retried = True
+        retry = run_duckdb(executable, concurrent_insert_sql(args, f"{catalog}_retry", write_table, row_id), args.timeout)
+        require_duckdb_success(retry, f"concurrent writer retry {row_id}")
+    return "passed-with-serial-retry" if retried else "passed-concurrently"
+
+
+def cleanup_tables(catalog: Any, namespace: str, tables: list[str], *, drop_namespace: bool) -> str:
+    cleanup_errors: list[str] = []
+    for table in tables:
+        try:
+            pyiceberg_smoke.drop_table_if_present(catalog, (namespace, table))
+        except Exception as error:
+            cleanup_errors.append(f"{table}: {error}")
+    if drop_namespace:
+        try:
+            catalog.drop_namespace(namespace)
+        except Exception as error:
+            cleanup_errors.append(f"namespace: {error}")
+    if cleanup_errors:
+        raise RuntimeError("DuckDB smoke cleanup failed: " + "; ".join(cleanup_errors))
+    return "dropped-tables-and-namespace" if drop_namespace else "dropped-tables-preserved-existing-namespace"
+
+
+def run_smoke(args: argparse.Namespace, deps: pyiceberg_smoke.RuntimeDeps) -> DuckDBSmokeResult:
+    executable = duckdb_path(args.duckdb)
+    client_version = duckdb_client_version(executable, args.timeout)
+    if client_version != args.duckdb_version:
+        raise RuntimeError(f"expected DuckDB {args.duckdb_version}, found {client_version}")
+
+    endpoint = pyiceberg_smoke.normalized_endpoint(args.endpoint)
+    pyiceberg_smoke.ensure_local_proxy_bypass(endpoint)
+    pyiceberg_smoke.ensure_aws_env(args.access_key, args.secret_key, args.region)
+    iceberg_args = pyiceberg_args(args)
+    pyiceberg_smoke.ensure_bucket(iceberg_args, deps)
+    pyiceberg_smoke.enable_table_bucket(iceberg_args, deps)
+
+    seed_table = table_name(args.table, "seed")
+    write_table = table_name(args.table, "write")
+    purge_table = table_name(args.table, "purge")
+    drop_table = table_name(args.table, "drop")
+    stage_table = table_name(args.table, "stage")
+    v3_table = table_name(args.table, "v3")
+    smoke_tables = [seed_table, write_table, purge_table, drop_table, stage_table, v3_table]
+    catalog = deps.load_catalog(iceberg_args.catalog_name, **pyiceberg_smoke.catalog_properties(iceberg_args))
+    pyiceberg_smoke.install_rustfs_rest_sigv4_adapter(catalog, iceberg_args, deps)
+    namespace_preexisting = bool(catalog.namespace_exists(args.namespace))
+    prepare_smoke_tables(catalog, args.namespace, smoke_tables, args.replace)
+    pyiceberg_smoke.ensure_namespace(catalog, args.namespace)
+    seed_pyiceberg_table(catalog, args, deps, seed_table)
+
+    checks: dict[str, str] = {}
+    cleanup_result = "not-requested"
+    metadata_location = "operator-recorded"
+    try:
+        positive = run_duckdb(
+            executable,
+            canonical_positive_sql(args, seed_table, write_table, purge_table, drop_table),
+            args.timeout,
+        )
+        require_duckdb_success(positive, "canonical REST catalog lifecycle")
+        row_count_batches = batches_with_column(positive, "row_count")
+        if len(row_count_batches) < 2 or row_count_batches[0][0]["row_count"] != 2 or row_count_batches[-1][0]["row_count"] != 2:
+            raise RuntimeError("DuckDB canonical REST catalog row counts did not remain at 2")
+        result_batches = batches_with_column(positive, "id")
+        expected_rows = [
+            {"id": 20, "payload": "twenty-merged", "category": "merged"},
+            {"id": 40, "payload": "forty", "category": "inserted"},
+        ]
+        if not result_batches or result_batches[-1] != expected_rows:
+            raise RuntimeError(f"DuckDB canonical DML returned unexpected rows: {result_batches[-1] if result_batches else []}")
+        snapshot_batches = batches_with_column(positive, "snapshot_count")
+        if not snapshot_batches or snapshot_batches[-1][0]["snapshot_count"] < 1:
+            raise RuntimeError("DuckDB snapshot metadata probe returned no snapshots")
+        if catalog.table_exists((args.namespace, drop_table)):
+            raise RuntimeError("DuckDB DROP TABLE did not remove the catalog entry")
+        checks["canonical_rest_catalog"] = "pass"
+        checks["single_table_ddl_dml"] = "pass"
+        checks["schema_evolution"] = "pass"
+        checks["snapshot_metadata"] = "pass"
+
+        if pyiceberg_rows(catalog, args.namespace, write_table) != expected_rows:
+            raise RuntimeError("PyIceberg did not observe DuckDB-created table rows")
+        checks["pyiceberg_cross_read"] = "pass"
+
+        alias = run_duckdb(executable, alias_sql(args, write_table), args.timeout)
+        require_duckdb_success(alias, "s3tables compatibility alias")
+        alias_counts = batches_with_column(alias, "alias_row_count")
+        alias_final_counts = batches_with_column(alias, "alias_final_row_count")
+        if not alias_counts or alias_counts[-1][0]["alias_row_count"] != 3:
+            raise RuntimeError("DuckDB compatibility alias insert did not produce row_count=3")
+        if not alias_final_counts or alias_final_counts[-1][0]["alias_final_row_count"] != 2:
+            raise RuntimeError("DuckDB compatibility alias cleanup did not restore row_count=2")
+        checks["s3tables_alias"] = "pass"
+
+        negatives = [
+            ("stage-create", "stage-create is not supported"),
+            ("purge", "purgeRequested=true is not supported"),
+            ("format-v3", "unsupported Iceberg table format-version: 3"),
+        ]
+        for kind, expected_error in negatives:
+            execution = run_duckdb(
+                executable,
+                negative_sql(
+                    args,
+                    kind=kind,
+                    seed_table=seed_table,
+                    write_table=write_table,
+                    purge_table=purge_table,
+                ),
+                args.timeout,
+            )
+            require_duckdb_error(execution, kind, expected_error)
+            checks[kind] = "failed-closed"
+        if catalog.table_exists((args.namespace, stage_table)) or catalog.table_exists((args.namespace, v3_table)):
+            raise RuntimeError("a failed DuckDB create probe left a catalog table behind")
+        if not catalog.table_exists((args.namespace, purge_table)):
+            raise RuntimeError("purgeRequested=true removed a table despite the expected failure")
+
+        multi_table = run_duckdb(
+            executable,
+            multi_table_sql(args, seed_table, write_table, purge_table),
+            args.timeout,
+        )
+        require_duckdb_success(multi_table, "multi-table endpoint-disabled mode")
+        if not any(row["id"] == 999 for row in pyiceberg_rows(catalog, args.namespace, write_table)):
+            raise RuntimeError("DuckDB multi-table endpoint-disabled mode did not commit the first table")
+        if not any(row["id"] == 999 for row in pyiceberg_rows(catalog, args.namespace, purge_table)):
+            raise RuntimeError("DuckDB multi-table endpoint-disabled mode did not commit the second table")
+        multi_cleanup_catalog = "cleanup_multi_table"
+        multi_cleanup = profile_sql(args, catalog=multi_cleanup_catalog, table=seed_table) + "\n".join(
+            [
+                f"DELETE FROM {table_identifier(multi_cleanup_catalog, args.namespace, write_table)} WHERE id = 999;",
+                f"DELETE FROM {table_identifier(multi_cleanup_catalog, args.namespace, purge_table)} WHERE id = 999;",
+            ]
+        )
+        require_duckdb_success(
+            run_duckdb(executable, multi_cleanup, args.timeout),
+            "multi-table endpoint-disabled cleanup",
+        )
+        checks["multi_table_endpoint_disabled"] = "pass-single-table-atomicity-only"
+
+        checks["concurrent_writers"] = run_concurrent_inserts(executable, args, write_table)
+        concurrent_rows = pyiceberg_rows(catalog, args.namespace, write_table)
+        if [row["id"] for row in concurrent_rows] != [20, 40, 60, 70]:
+            raise RuntimeError(f"concurrent DuckDB writers produced unexpected rows: {concurrent_rows}")
+        cleanup_catalog = "cleanup_concurrency"
+        cleanup_identifier = table_identifier(cleanup_catalog, args.namespace, write_table)
+        cleanup_sql = profile_sql(args, catalog=cleanup_catalog, table=write_table) + "\n".join(
+            [
+                f"DELETE FROM {cleanup_identifier} WHERE id IN (60, 70);",
+                f"SELECT count(*) AS final_row_count FROM {cleanup_identifier};",
+            ]
+        )
+        cleanup_execution = run_duckdb(executable, cleanup_sql, args.timeout)
+        require_duckdb_success(cleanup_execution, "concurrency cleanup")
+        final_batches = batches_with_column(cleanup_execution, "final_row_count")
+        if not final_batches or final_batches[-1][0]["final_row_count"] != 2:
+            raise RuntimeError("DuckDB concurrency cleanup did not restore row_count=2")
+
+        final_table = catalog.load_table((args.namespace, write_table))
+        final_rows = sorted(final_table.scan().to_arrow().to_pylist(), key=lambda row: row["id"])
+        if final_rows != expected_rows:
+            raise RuntimeError(f"final PyIceberg cross-read returned unexpected rows: {final_rows}")
+        metadata_location = pyiceberg_smoke.table_metadata_location(final_table) or "operator-recorded"
+        if metadata_location == "operator-recorded":
+            response = pyiceberg_smoke.signed_rest_request(
+                argparse.Namespace(**{**vars(iceberg_args), "table": write_table}),
+                deps,
+                "GET",
+                f"/iceberg/v1/{urllib.parse.quote(args.bucket, safe='')}/namespaces/"
+                f"{urllib.parse.quote(args.namespace, safe='')}/tables/{urllib.parse.quote(write_table, safe='')}",
+            )
+            metadata_location = response.get("metadata-location", "operator-recorded")
+        if metadata_location == "operator-recorded":
+            raise RuntimeError("DuckDB smoke could not resolve the final metadata location")
+        metadata_scan = run_duckdb(
+            executable,
+            engine_compatibility.duckdb_sql_probe(
+                endpoint=args.endpoint,
+                access_key=args.access_key,
+                secret_key=args.secret_key,
+                region=args.region,
+                metadata_location=metadata_location,
+            ),
+            args.timeout,
+        )
+        require_duckdb_success(metadata_scan, "metadata-location scan")
+        metadata_counts = batches_with_column(metadata_scan, "row_count")
+        if not metadata_counts or metadata_counts[-1][0]["row_count"] != 2:
+            raise RuntimeError("DuckDB metadata-location scan did not return row_count=2")
+        checks["metadata_location_scan"] = "pass"
+    finally:
+        if args.cleanup:
+            cleanup_result = cleanup_tables(
+                catalog,
+                args.namespace,
+                smoke_tables,
+                drop_namespace=not namespace_preexisting,
+            )
+
+    return DuckDBSmokeResult(client_version, metadata_location, 2, cleanup_result, checks)
+
+
+def current_utc_timestamp() -> str:
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def write_live_evidence(args: argparse.Namespace, result: DuckDBSmokeResult) -> None:
+    if not args.live_evidence_output:
+        return
+    command = pyiceberg_smoke.redacted_command(sys.argv)
+    record = engine_compatibility.live_conformance_evidence_record(
+        client_name="DuckDB Iceberg",
+        client_version=result.client_version,
+        scenario="rest-catalog-single-table-read-write-cross-engine-negative-boundaries",
+        rustfs_build=args.rustfs_build,
+        git_sha=args.git_sha,
+        catalog_backing=args.catalog_backing,
+        endpoint=args.endpoint,
+        warehouse=args.bucket,
+        rest_path="/iceberg",
+        namespace=args.namespace,
+        table=table_name(args.table, "write"),
+        metadata_location=result.metadata_location,
+        run_timestamp_utc=args.run_timestamp_utc or current_utc_timestamp(),
+        operator=args.operator,
+        expected_status="pass",
+        observed_status="pass",
+        row_count=result.row_count,
+        cleanup_result=result.cleanup_result,
+        claim="automated-rest-catalog-smoke",
+        command=command,
+    )
+    document = {
+        "live_conformance_evidence": record,
+        "checks": result.checks,
+        "validation": engine_compatibility.validate_live_conformance_evidence(record),
+    }
+    Path(args.live_evidence_output).write_text(json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        deps = pyiceberg_smoke.load_runtime_deps()
+        result = run_smoke(args, deps)
+        write_live_evidence(args, result)
+        print(json.dumps({"status": "pass", "row_count": result.row_count, "checks": result.checks}, sort_keys=True))
+        return 0
+    except Exception as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/table-catalog/engine_compatibility.py

@@ -53,7 +53,7 @@ LIVE_EVIDENCE_ALLOWED_CLAIMS = OrderedDict(
         ("PyIceberg", ["automated-smoke"]),
         ("Spark Iceberg REST catalog", ["manual-live-verified"]),
         ("Trino Iceberg REST catalog", ["manual-live-read-verified"]),
-        ("DuckDB Iceberg", ["manual-live-read-verified"]),
+        ("DuckDB Iceberg", ["manual-live-read-verified", "automated-rest-catalog-smoke"]),
         ("Databend", ["manual-live-s3-stage-verified"]),
         ("Snowflake Open Catalog / Iceberg integrations", ["reference-only"]),
     ]
@@ -155,13 +155,15 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
         },
         {
             "client": "DuckDB Iceberg",
-            "status": "manual-live-harness",
+            "status": "automated-smoke",
-            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
+            "entrypoint": "scripts/table-catalog/duckdb_smoke.py",
             "scenarios": [
-                scenario("metadata-read", "manual-live-probe", "read a supplied Iceberg metadata location through DuckDB iceberg_scan"),
+                scenario("metadata-read", "automated", "read the final metadata location through DuckDB iceberg_scan"),
-                scenario("catalog-attach", "generated-harness", "attach RustFS as a generic signed Iceberg REST catalog"),
+                scenario("catalog-attach", "automated", "attach `/iceberg` with s3 signing and `/_iceberg` with s3tables signing"),
-                scenario("read-table", "manual-live-probe", "read an existing table through the attached catalog"),
+                scenario("read-table", "automated", "read a PyIceberg-created table through the attached catalog"),
-                scenario("write-table", "manual-validation-required", "generated SQL does not promote DuckDB write compatibility"),
+                scenario("write-table", "automated", "exercise single-table DDL, DML, schema evolution, snapshots, and PyIceberg cross-read"),
+                scenario("unsupported-boundaries", "automated", "verify staged create, purge, and format v3 fail closed"),
+                scenario("multi-table-mode", "automated", "verify DuckDB can avoid the multi-table commit endpoint without claiming cross-table atomicity"),
             ],
         },
         {
@@ -682,11 +684,11 @@ def live_conformance_evidence(
                     OrderedDict(
                         [
                             ("client", "DuckDB Iceberg"),
-                            ("scenario", "iceberg-scan-current-metadata-location"),
+                            ("scenario", "rest-catalog-single-table-read-write-cross-engine-negative-boundaries"),
                             ("expected_status", "pass"),
                             ("expected_row_count", 2),
-                            ("claim_after_pass", "manual-live-read-verified"),
+                            ("claim_after_pass", "automated-rest-catalog-smoke"),
-                            ("write_claim_after_pass", "not-claimed"),
+                            ("write_claim_after_pass", "single-table-automated-smoke"),
                         ]
                     ),
                     OrderedDict(
@@ -712,9 +714,9 @@ def live_conformance_evidence(
             (
                 "promotion_rules",
                 [
-                    "Keep PyIceberg as the only automated claim unless the run is executed by CI or a repeatable operator job.",
+                    "Keep PyIceberg and DuckDB automated claims tied to their repeatable smoke entrypoints and recorded client versions.",
                     "Promote Spark only to manual-live-verified when the exact RustFS build, Spark version, Iceberg version, SQL output, and row_count are recorded.",
-                    "Do not promote Trino or DuckDB write compatibility from read probes; write compatibility remains not-claimed.",
+                    "Keep Trino write compatibility not-claimed after its read probe and do not broaden DuckDB beyond the automated single-table scenarios.",
                     "Do not promote Snowflake or vendor catalog interoperability from a generated template without a repeatable live run.",
                     "Treat manual-live failures as compatibility findings and keep the previous public claim boundary.",
                 ],
@@ -1624,7 +1626,7 @@ def live_conformance_harness(
                     OrderedDict(
                         [
                             ("name", "DuckDB Iceberg"),
-                            ("status", "manual-live-harness"),
+                            ("status", "automated-smoke"),
                             ("version", duckdb_version),
                             ("metadata_location", metadata_location),
                             ("sql_file", "/tmp/rustfs-s3tables-duckdb-read.sql"),
@@ -1641,8 +1643,8 @@ def live_conformance_harness(
                                 "rest_catalog_expected",
                                 "generic Iceberg REST ATTACH returns row_count=2 for an existing RustFS table",
                             ),
-                            ("rest_catalog_write_compatibility", "manual-live-validation-required"),
+                            ("rest_catalog_write_compatibility", "single-table-automated-smoke"),
-                            ("write_compatibility", "not-claimed"),
+                            ("write_compatibility", "single-table-automated-smoke"),
                         ]
                     ),
                     OrderedDict(

scripts/table-catalog/pyiceberg_smoke.py

@@ -167,9 +167,9 @@ CLIENT_MATRIX: list[dict[str, str]] = [
     },
     {
         "client": "DuckDB Iceberg",
-        "status": "manual-live-read-probe",
+        "status": "automated-smoke",
-        "coverage": "generated httpfs/iceberg SQL using an operator-supplied current metadata location; write/commit is not claimed",
+        "coverage": "metadata-location read plus generic REST catalog single-table DDL, DML, schema evolution, snapshots, canonical and compatibility signing, negative boundaries, endpoint-disabled multi-table mode, concurrent writers, and PyIceberg cross-read",
-        "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
+        "entrypoint": "scripts/table-catalog/duckdb_smoke.py",
     },
     {
         "client": "Databend",

scripts/table-catalog/test_duckdb_smoke.py

@@ -0,0 +1,220 @@
+#!/usr/bin/env python3
+"""Unit tests for the RustFS DuckDB REST Catalog smoke helper."""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+import duckdb_smoke
+
+
+class DuckDBSmokeTest(unittest.TestCase):
+    def args(self) -> argparse.Namespace:
+        return argparse.Namespace(
+            endpoint="http://127.0.0.1:9000",
+            access_key="rustfsadmin",
+            secret_key="rustfsadmin",
+            region="us-east-1",
+            bucket="rustfs-duckdb-smoke",
+            namespace="duckdb_smoke",
+            table="events",
+            duckdb="duckdb",
+            duckdb_version="1.5.5",
+            timeout=60.0,
+            cleanup=True,
+            replace=False,
+            insecure=False,
+            live_evidence_output=None,
+            rustfs_build="rustfs-test",
+            git_sha="abc123",
+            catalog_backing="object",
+            operator="test-operator",
+            run_timestamp_utc="2026-08-27T00:00:00Z",
+        )
+
+    def test_parse_args_rejects_unsafe_identifiers(self) -> None:
+        with contextlib.redirect_stderr(io.StringIO()):
+            with self.assertRaises(SystemExit):
+                duckdb_smoke.parse_args(["--namespace", "bad-name"])
+
+    def test_duckdb_client_version_removes_v_prefix(self) -> None:
+        process = SimpleNamespace(returncode=0, stdout="v1.5.5\n", stderr="")
+        with mock.patch.object(duckdb_smoke.subprocess, "run", return_value=process):
+            self.assertEqual(duckdb_smoke.duckdb_client_version("duckdb", 10), "1.5.5")
+
+    def test_parse_duckdb_json_accepts_multiple_query_batches(self) -> None:
+        batches = duckdb_smoke.parse_duckdb_json(
+            '[{"row_count":2}]\n[{"id":20},\n{"id":40}]\n'
+        )
+
+        self.assertEqual(batches[0][0]["row_count"], 2)
+        self.assertEqual([row["id"] for row in batches[1]], [20, 40])
+
+    def test_run_duckdb_does_not_parse_json_for_failed_process(self) -> None:
+        process = SimpleNamespace(returncode=1, stdout="not-json", stderr="expected failure")
+        with mock.patch.object(duckdb_smoke.subprocess, "run", return_value=process):
+            execution = duckdb_smoke.run_duckdb("duckdb", "SELECT 1", 10)
+
+        self.assertEqual(execution.returncode, 1)
+        self.assertEqual(execution.batches, [])
+
+    def test_concurrent_retry_markers_do_not_accept_generic_commit_errors(self) -> None:
+        args = self.args()
+        generic_failure = duckdb_smoke.DuckDBExecution(1, [], "", "commit failed: permission denied")
+        with mock.patch.object(duckdb_smoke, "run_duckdb", return_value=generic_failure):
+            with self.assertRaisesRegex(RuntimeError, "permission denied"):
+                duckdb_smoke.run_concurrent_inserts("duckdb", args, "events_write")
+
+    def test_canonical_sql_covers_single_table_lifecycle(self) -> None:
+        sql = duckdb_smoke.canonical_positive_sql(
+            self.args(),
+            "events_seed",
+            "events_write",
+            "events_purge",
+            "events_drop",
+        )
+
+        self.assertIn("STAGE_CREATE_TABLES false", sql)
+        self.assertIn("SKIP_CREATE_TABLE_METADATA_UPDATES true", sql)
+        self.assertIn("CREATE TABLE", sql)
+        self.assertIn("INSERT INTO", sql)
+        self.assertIn("UPDATE", sql)
+        self.assertIn("DELETE FROM", sql)
+        self.assertIn("MERGE INTO", sql)
+        self.assertIn("ALTER TABLE", sql)
+        self.assertIn("iceberg_snapshots", sql)
+        self.assertNotIn("DROP TABLE IF EXISTS", sql)
+        self.assertIn('DROP TABLE "rustfs_duckdb"."duckdb_smoke"."events_drop"', sql)
+
+    def test_prepare_smoke_tables_refuses_existing_identifiers_without_replace(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.side_effect = lambda identifier: identifier[1] == "events_write"
+
+        with self.assertRaisesRegex(RuntimeError, "duckdb_smoke.events_write"):
+            duckdb_smoke.prepare_smoke_tables(
+                catalog,
+                "duckdb_smoke",
+                ["events_seed", "events_write"],
+                replace=False,
+            )
+
+        catalog.drop_table.assert_not_called()
+
+    def test_prepare_smoke_tables_replaces_only_existing_identifiers_when_requested(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.side_effect = lambda identifier: identifier[1] == "events_write"
+
+        duckdb_smoke.prepare_smoke_tables(
+            catalog,
+            "duckdb_smoke",
+            ["events_seed", "events_write"],
+            replace=True,
+        )
+
+        catalog.drop_table.assert_called_once_with(("duckdb_smoke", "events_write"))
+
+    def test_cleanup_preserves_a_preexisting_namespace(self) -> None:
+        catalog = mock.Mock()
+        catalog.table_exists.return_value = False
+
+        result = duckdb_smoke.cleanup_tables(
+            catalog,
+            "duckdb_smoke",
+            ["events_seed", "events_write"],
+            drop_namespace=False,
+        )
+
+        self.assertEqual(result, "dropped-tables-preserved-existing-namespace")
+        catalog.drop_namespace.assert_not_called()
+
+    def test_alias_sql_uses_s3tables_signing(self) -> None:
+        sql = duckdb_smoke.alias_sql(self.args(), "events_write")
+
+        self.assertIn("ENDPOINT 'http://127.0.0.1:9000/_iceberg'", sql)
+        self.assertIn("SIGV4_SERVICE 's3tables'", sql)
+        self.assertIn("INSERT INTO", sql)
+        self.assertIn("alias_final_row_count", sql)
+
+    def test_boundary_sql_records_required_compatibility_options(self) -> None:
+        args = self.args()
+        stage_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="stage-create",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        stage_attach = stage_sql.split('DETACH "bootstrap_stage-create";', 1)[1]
+        self.assertNotIn("STAGE_CREATE_TABLES false", stage_attach)
+        self.assertIn("CREATE TABLE", stage_attach)
+
+        purge_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="purge",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("PURGE_REQUESTED true", purge_sql)
+        self.assertIn('DROP TABLE "purge_requested"."duckdb_smoke"."events_purge"', purge_sql)
+
+        v3_sql = duckdb_smoke.negative_sql(
+            args,
+            kind="format-v3",
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("'format-version' = '3'", v3_sql)
+
+        multi_sql = duckdb_smoke.multi_table_sql(
+            args,
+            seed_table="events_seed",
+            write_table="events_write",
+            purge_table="events_purge",
+        )
+        self.assertIn("DISABLE_MULTI_TABLE_COMMIT true", multi_sql)
+        self.assertIn("BEGIN TRANSACTION", multi_sql)
+        self.assertIn("'non-atomic'", multi_sql)
+
+    def test_pyiceberg_args_use_canonical_catalog(self) -> None:
+        args = duckdb_smoke.pyiceberg_args(self.args())
+
+        self.assertEqual(args.rest_path, "/iceberg")
+        self.assertEqual(args.rest_signing_name, "s3")
+        self.assertEqual(args.bucket, "rustfs-duckdb-smoke")
+
+    def test_live_evidence_records_automated_duckdb_claim(self) -> None:
+        args = self.args()
+        result = duckdb_smoke.DuckDBSmokeResult(
+            client_version="1.5.5",
+            metadata_location="s3://rustfs-duckdb-smoke/metadata/00001.json",
+            row_count=2,
+            cleanup_result="dropped-tables-and-namespace",
+            checks={"canonical_rest_catalog": "pass"},
+        )
+        with tempfile.TemporaryDirectory() as temp_dir:
+            output = Path(temp_dir) / "evidence.json"
+            args.live_evidence_output = str(output)
+            with mock.patch.object(duckdb_smoke.sys, "argv", ["duckdb_smoke.py", "--secret-key", "secret"]):
+                duckdb_smoke.write_live_evidence(args, result)
+            document = json.loads(output.read_text(encoding="utf-8"))
+
+        evidence = document["live_conformance_evidence"]
+        self.assertEqual(evidence["client_name"], "DuckDB Iceberg")
+        self.assertEqual(evidence["claim"], "automated-rest-catalog-smoke")
+        self.assertIn("--secret-key '<redacted>'", evidence["command"])
+        self.assertNotIn("--secret-key secret", evidence["command"])
+        self.assertEqual(document["validation"]["status"], "accepted")
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/table-catalog/test_engine_compatibility.py

@@ -42,10 +42,13 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertContainsScenario(trino, "catalog-load", "manual-live-probe")
 
         duckdb = by_client["DuckDB Iceberg"]
-        self.assertEqual(duckdb["status"], "manual-live-harness")
+        self.assertEqual(duckdb["status"], "automated-smoke")
-        self.assertContainsScenario(duckdb, "metadata-read", "manual-live-probe")
+        self.assertEqual(duckdb["entrypoint"], "scripts/table-catalog/duckdb_smoke.py")
-        self.assertContainsScenario(duckdb, "catalog-attach", "generated-harness")
+        self.assertContainsScenario(duckdb, "metadata-read", "automated")
-        self.assertContainsScenario(duckdb, "write-table", "manual-validation-required")
+        self.assertContainsScenario(duckdb, "catalog-attach", "automated")
+        self.assertContainsScenario(duckdb, "write-table", "automated")
+        self.assertContainsScenario(duckdb, "unsupported-boundaries", "automated")
+        self.assertContainsScenario(duckdb, "multi-table-mode", "automated")
 
     def test_spark_config_uses_rustfs_rest_catalog_and_s3fileio(self) -> None:
         config = engine_compatibility.spark_catalog_config(
@@ -450,7 +453,7 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertEqual(trino["write_compatibility"], "not-claimed")
 
         duckdb = by_client["DuckDB Iceberg"]
-        self.assertEqual(duckdb["status"], "manual-live-harness")
+        self.assertEqual(duckdb["status"], "automated-smoke")
         self.assertEqual(duckdb["version"], "1.5.5")
         self.assertIn("LOAD httpfs", duckdb["sql"])
         self.assertIn("LOAD iceberg", duckdb["sql"])
@@ -458,8 +461,8 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertIn("ATTACH 'rustfs-s3table-smoke'", duckdb["rest_catalog_sql"])
         self.assertIn("STAGE_CREATE_TABLES false", duckdb["rest_catalog_sql"])
         self.assertIn("SKIP_CREATE_TABLE_METADATA_UPDATES true", duckdb["rest_catalog_sql"])
-        self.assertEqual(duckdb["rest_catalog_write_compatibility"], "manual-live-validation-required")
+        self.assertEqual(duckdb["rest_catalog_write_compatibility"], "single-table-automated-smoke")
-        self.assertEqual(duckdb["write_compatibility"], "not-claimed")
+        self.assertEqual(duckdb["write_compatibility"], "single-table-automated-smoke")
 
         snowflake = by_client["Snowflake Open Catalog / Iceberg integrations"]
         self.assertEqual(snowflake["status"], "manual-reference-probe")
@@ -503,7 +506,8 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertEqual(table_by_client["PyIceberg"]["claim_after_pass"], "automated-smoke")
         self.assertEqual(table_by_client["Spark Iceberg REST catalog"]["claim_after_pass"], "manual-live-verified")
         self.assertEqual(table_by_client["Trino Iceberg REST catalog"]["write_claim_after_pass"], "not-claimed")
-        self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "not-claimed")
+        self.assertEqual(table_by_client["DuckDB Iceberg"]["claim_after_pass"], "automated-rest-catalog-smoke")
+        self.assertEqual(table_by_client["DuckDB Iceberg"]["write_claim_after_pass"], "single-table-automated-smoke")
         self.assertIn("manual-live", " ".join(evidence["promotion_rules"]))
         self.assertIn("not-claimed", " ".join(evidence["promotion_rules"]))
 
@@ -592,6 +596,10 @@ class EngineCompatibilityTest(unittest.TestCase):
         self.assertIn("metadata_location", schema["required_fields"])
         self.assertIn("claim", schema["required_fields"])
         self.assertEqual(schema["claim_promotion"]["Trino Iceberg REST catalog"], ["manual-live-read-verified"])
+        self.assertEqual(
+            schema["claim_promotion"]["DuckDB Iceberg"],
+            ["manual-live-read-verified", "automated-rest-catalog-smoke"],
+        )
 
     def test_production_operations_guide_covers_release_boundaries(self) -> None:
         guide = engine_compatibility.production_operations_guide(