diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index a49c3520d..063fd3a41 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/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. diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index cbeb08b84..771a9ea40 100644 --- a/scripts/table-catalog/README.md +++ b/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 -this script today. Spark has a repeatable manual/live harness with pinned -client package inputs, generated configuration, generated SQL, expected -results, and a CI opt-in gate. Trino, DuckDB, Databend, and Snowflake now have +work items. They are intentionally conservative: PyIceberg and DuckDB have +separate automated smoke entrypoints. Spark has a repeatable manual/live harness +with pinned client package inputs, generated configuration, generated SQL, +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 -compatibility until the repeatable live smoke records passing evidence. +Run the repeatable DuckDB 1.5.5 smoke against an already running RustFS: + +```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 - metadata location. DuckDB write and commit compatibility are not claimed. +- DuckDB: a legacy `httpfs` and `iceberg` read probe using an operator-supplied + 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 diff --git a/scripts/table-catalog/duckdb_smoke.py b/scripts/table-catalog/duckdb_smoke.py new file mode 100755 index 000000000..b4089fd0f --- /dev/null +++ b/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()) diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index 31a64adf4..e394a660e 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/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", - "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance", + "status": "automated-smoke", + "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("catalog-attach", "generated-harness", "attach RustFS as a generic signed Iceberg REST catalog"), - scenario("read-table", "manual-live-probe", "read an existing table through the attached catalog"), - scenario("write-table", "manual-validation-required", "generated SQL does not promote DuckDB write compatibility"), + scenario("metadata-read", "automated", "read the final metadata location through DuckDB iceberg_scan"), + scenario("catalog-attach", "automated", "attach `/iceberg` with s3 signing and `/_iceberg` with s3tables signing"), + scenario("read-table", "automated", "read a PyIceberg-created table through the attached catalog"), + 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"), - ("write_claim_after_pass", "not-claimed"), + ("claim_after_pass", "automated-rest-catalog-smoke"), + ("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"), - ("write_compatibility", "not-claimed"), + ("rest_catalog_write_compatibility", "single-table-automated-smoke"), + ("write_compatibility", "single-table-automated-smoke"), ] ), OrderedDict( diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index f11d81f28..8c055c9ff 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -167,9 +167,9 @@ CLIENT_MATRIX: list[dict[str, str]] = [ }, { "client": "DuckDB Iceberg", - "status": "manual-live-read-probe", - "coverage": "generated httpfs/iceberg SQL using an operator-supplied current metadata location; write/commit is not claimed", - "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance", + "status": "automated-smoke", + "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/duckdb_smoke.py", }, { "client": "Databend", diff --git a/scripts/table-catalog/test_duckdb_smoke.py b/scripts/table-catalog/test_duckdb_smoke.py new file mode 100644 index 000000000..cba18393f --- /dev/null +++ b/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 ''", evidence["command"]) + self.assertNotIn("--secret-key secret", evidence["command"]) + self.assertEqual(document["validation"]["status"], "accepted") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 0d0307bd3..2577e2b21 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/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.assertContainsScenario(duckdb, "metadata-read", "manual-live-probe") - self.assertContainsScenario(duckdb, "catalog-attach", "generated-harness") - self.assertContainsScenario(duckdb, "write-table", "manual-validation-required") + self.assertEqual(duckdb["status"], "automated-smoke") + self.assertEqual(duckdb["entrypoint"], "scripts/table-catalog/duckdb_smoke.py") + self.assertContainsScenario(duckdb, "metadata-read", "automated") + 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["write_compatibility"], "not-claimed") + self.assertEqual(duckdb["rest_catalog_write_compatibility"], "single-table-automated-smoke") + 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(