rustfs/scripts/table-catalog/engine_compatibility.py

#!/usr/bin/env python3
"""Machine-readable Iceberg engine compatibility helpers for RustFS S3 Tables."""

from __future__ import annotations

import argparse
import json
import re
import shlex
import urllib.parse
from collections import OrderedDict
from io import StringIO
from typing import Any

DEFAULT_PYICEBERG_VERSION = "0.10.0"
DEFAULT_SPARK_VERSION = "3.5.4"
DEFAULT_ICEBERG_VERSION = "1.7.1"
DEFAULT_SCALA_VERSION = "2.12"
DEFAULT_TRINO_VERSION = "477"
DEFAULT_DUCKDB_VERSION = "1.3.2"
DEFAULT_SNOWFLAKE_CLIENT_VERSION = "operator-recorded"
DEFAULT_DATABEND_VERSION = "operator-recorded"
DEFAULT_TRINO_SERVER = "http://127.0.0.1:8080"
DEFAULT_DATABEND_DSN = "databend://root:@127.0.0.1:8000/default"

VENDOR_SPARK_PROFILES: dict[str, dict[str, str]] = {
    "rustfs": {
        "catalog_uri": "{endpoint}/iceberg",
        "warehouse": "{warehouse}",
        "rest_signing_name": "s3",
        "s3_endpoint": "{endpoint}",
        "s3_path_style_access": "true",
    },
    "rustfs-compat": {
        "catalog_uri": "{endpoint}/_iceberg",
        "warehouse": "{warehouse}",
        "rest_signing_name": "s3tables",
        "s3_endpoint": "{endpoint}",
        "s3_path_style_access": "true",
    },
    "aws-s3tables": {
        "catalog_uri": "https://s3tables.{region}.amazonaws.com/iceberg",
        "warehouse": "arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}",
        "rest_signing_name": "s3tables",
    },
    "minio-aistor": {
        "catalog_uri": "{endpoint}/_iceberg",
        "warehouse": "{warehouse}",
        "rest_signing_name": "s3tables",
        "s3_endpoint": "{endpoint}",
        "s3_path_style_access": "true",
    },
    "cloudflare-r2-data-catalog": {
        "catalog_uri": "{catalog_uri}",
        "warehouse": "{warehouse_name}",
        "rest_signing_name": "s3",
    },
    "oss-tables": {
        "catalog_uri": "{endpoint}/iceberg",
        "warehouse": "{warehouse}",
        "rest_signing_name": "s3",
    },
}


def scenario(name: str, status: str, evidence: str) -> dict[str, str]:
    return {
        "name": name,
        "status": status,
        "evidence": evidence,
    }


def engine_compatibility_matrix() -> list[dict[str, Any]]:
    return [
        {
            "client": "PyIceberg",
            "status": "automated-smoke",
            "entrypoint": "scripts/table-catalog/pyiceberg_smoke.py",
            "scenarios": [
                scenario("create-namespace", "automated", "PyIceberg catalog.create_namespace_if_not_exists/create_namespace"),
                scenario("create-table", "automated", "PyIceberg catalog.create_table"),
                scenario("append", "automated", "PyIceberg table.append with PyArrow rows"),
                scenario("reload-table", "automated", "PyIceberg catalog.load_table after append"),
                scenario("scan", "automated", "PyIceberg table.scan().to_arrow"),
                scenario("drop-table", "automated-with-cleanup", "PyIceberg catalog.drop_table when --cleanup or --replace is set"),
                scenario("commit-conflict", "direct-rest-probe-required", "catalog commit conflict remains a follow-up live probe"),
            ],
        },
        {
            "client": "Spark Iceberg REST catalog",
            "status": "manual-live-harness",
            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
                scenario("create-namespace", "manual-live-harness", "CREATE NAMESPACE IF NOT EXISTS"),
                scenario("create-table", "manual-live-harness", "CREATE TABLE USING iceberg"),
                scenario("append", "manual-live-harness", "INSERT INTO"),
                scenario("reload-table", "manual-live-harness", "REFRESH TABLE and SELECT COUNT"),
                scenario("drop-table", "manual-live-harness", "DROP TABLE and optional DROP NAMESPACE"),
                scenario("commit-conflict", "manual-validation-required", "requires a two-writer Spark or REST conflict harness"),
            ],
        },
        {
            "client": "Trino Iceberg REST catalog",
            "status": "manual-live-read-probe",
            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
                scenario("catalog-load", "manual-live-probe", "REST catalog properties generated by the live conformance helper"),
                scenario("read-table", "manual-live-probe", "SELECT from a table created by PyIceberg or Spark"),
                scenario("write-table", "not-claimed", "Trino write compatibility is not claimed by this harness"),
            ],
        },
        {
            "client": "DuckDB Iceberg",
            "status": "manual-live-read-probe",
            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
                scenario("metadata-read", "manual-live-probe", "read a supplied Iceberg metadata location through DuckDB iceberg_scan"),
                scenario("read-table", "manual-live-probe", "read-path verification only"),
                scenario("write-table", "not-claimed", "DuckDB write/commit compatibility is not claimed"),
            ],
        },
        {
            "client": "StarRocks Iceberg REST catalog",
            "status": "documented-read-path",
            "entrypoint": "scripts/table-catalog/README.md",
            "scenarios": [
                scenario("catalog-load", "manual-validation-required", "REST catalog configuration reference"),
                scenario("read-table", "manual-validation-required", "external catalog read-path verification only"),
                scenario("write-table", "not-claimed", "StarRocks write/commit compatibility is not claimed"),
            ],
        },
        {
            "client": "Snowflake Open Catalog / Iceberg integrations",
            "status": "manual-reference-probe",
            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
                scenario("external-volume", "manual-reference-probe", "operator-adapted external volume/catalog integration SQL template"),
                scenario("catalog-load", "not-claimed", "reference only until a repeatable external integration harness exists"),
            ],
        },
        {
            "client": "Databend",
            "status": "manual-live-s3-stage-probe",
            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
                scenario("s3-data-plane-read", "manual-live-probe", "S3 stage/data-plane read probe for table data files"),
                scenario("iceberg-rest-catalog", "not-claimed", "Databend REST catalog integration is not claimed"),
            ],
        },
    ]


def normalized_endpoint(endpoint: str) -> str:
    return endpoint.rstrip("/")


def normalized_rest_path(rest_path: str) -> str:
    stripped = rest_path.strip()
    if not stripped:
        raise ValueError("REST catalog path cannot be empty")
    if not stripped.startswith("/"):
        stripped = f"/{stripped}"
    return stripped.rstrip("/")


def vendor_profile_context(
    *,
    endpoint: str,
    warehouse: str,
    region: str,
    account_id: str,
    table_bucket: str,
    catalog_uri: str | None,
    warehouse_name: str | None,
) -> dict[str, str]:
    endpoint = normalized_endpoint(endpoint)
    return {
        "account_id": account_id,
        "catalog_uri": (catalog_uri or f"{endpoint}/iceberg").rstrip("/"),
        "endpoint": endpoint,
        "region": region,
        "table_bucket": table_bucket,
        "warehouse": warehouse,
        "warehouse_name": warehouse_name or warehouse,
    }


def vendor_profile_value(profile: str, key: str, context: dict[str, str]) -> str:
    try:
        template = VENDOR_SPARK_PROFILES[profile][key]
    except KeyError as err:
        raise ValueError(f"unknown vendor profile field: {profile}.{key}") from err
    return template.format(**context).rstrip("/")


def spark_catalog_config(
    *,
    endpoint: str,
    warehouse: str,
    access_key: str,
    secret_key: str,
    region: str,
    catalog_name: str,
    rest_path: str,
    rest_signing_name: str,
) -> OrderedDict[str, str]:
    endpoint = normalized_endpoint(endpoint)
    rest_path = normalized_rest_path(rest_path)
    prefix = f"spark.sql.catalog.{catalog_name}"
    return OrderedDict(
        [
            (prefix, "org.apache.iceberg.spark.SparkCatalog"),
            (f"{prefix}.type", "rest"),
            (f"{prefix}.uri", f"{endpoint}{rest_path}"),
            (f"{prefix}.warehouse", warehouse),
            (f"{prefix}.io-impl", "org.apache.iceberg.aws.s3.S3FileIO"),
            (f"{prefix}.s3.endpoint", endpoint),
            (f"{prefix}.s3.path-style-access", "true"),
            (f"{prefix}.rest.sigv4-enabled", "true"),
            (f"{prefix}.rest.signing-name", rest_signing_name),
            (f"{prefix}.rest.signing-region", region),
            (f"{prefix}.s3.access-key-id", access_key),
            (f"{prefix}.s3.secret-access-key", secret_key),
        ]
    )


def spark_vendor_catalog_config(
    *,
    profile: str,
    endpoint: str,
    warehouse: str,
    access_key: str,
    secret_key: str,
    region: str,
    catalog_name: str,
    account_id: str,
    table_bucket: str,
    catalog_uri: str | None,
    warehouse_name: str | None,
    rest_path: str | None = None,
    rest_signing_name: str | None = None,
) -> OrderedDict[str, str]:
    context = vendor_profile_context(
        endpoint=endpoint,
        warehouse=warehouse,
        region=region,
        account_id=account_id,
        table_bucket=table_bucket,
        catalog_uri=catalog_uri,
        warehouse_name=warehouse_name,
    )
    profile_defaults = VENDOR_SPARK_PROFILES.get(profile)
    if profile_defaults is None:
        raise ValueError(f"unknown vendor profile: {profile}")
    configured_catalog_uri = vendor_profile_value(profile, "catalog_uri", context)
    if rest_path is not None:
        configured_catalog_uri = f"{normalized_endpoint(endpoint)}{normalized_rest_path(rest_path)}"
    config = spark_catalog_config(
        endpoint=endpoint,
        warehouse=vendor_profile_value(profile, "warehouse", context),
        access_key=access_key,
        secret_key=secret_key,
        region=region,
        catalog_name=catalog_name,
        rest_path="/iceberg",
        rest_signing_name=rest_signing_name or profile_defaults["rest_signing_name"],
    )
    prefix = f"spark.sql.catalog.{catalog_name}"
    config[f"{prefix}.uri"] = configured_catalog_uri
    if "s3_endpoint" in profile_defaults:
        config[f"{prefix}.s3.endpoint"] = vendor_profile_value(profile, "s3_endpoint", context)
    else:
        config.pop(f"{prefix}.s3.endpoint", None)
        config.pop(f"{prefix}.s3.access-key-id", None)
        config.pop(f"{prefix}.s3.secret-access-key", None)
    if "s3_path_style_access" in profile_defaults:
        config[f"{prefix}.s3.path-style-access"] = profile_defaults["s3_path_style_access"]
    else:
        config.pop(f"{prefix}.s3.path-style-access", None)
    return config


def quote_spark_identifier(identifier: str) -> str:
    if not identifier or "`" in identifier or "\n" in identifier or "\r" in identifier:
        raise ValueError("Spark identifier must be non-empty and must not contain backticks or newlines")
    return f"`{identifier}`"


def quote_double_identifier(identifier: str) -> str:
    if not identifier or '"' in identifier or "\n" in identifier or "\r" in identifier:
        raise ValueError("SQL identifier must be non-empty and must not contain double quotes or newlines")
    return f'"{identifier}"'


def spark_table_identifier(catalog_name: str, namespace: str, table: str) -> str:
    return ".".join(
        [
            catalog_name,
            quote_spark_identifier(namespace),
            quote_spark_identifier(table),
        ]
    )


def spark_sql_smoke(
    *,
    catalog_name: str,
    namespace: str,
    table: str,
    cleanup: bool = False,
) -> str:
    namespace_identifier = f"{catalog_name}.{quote_spark_identifier(namespace)}"
    table_identifier = spark_table_identifier(catalog_name, namespace, table)
    statements = [
        f"CREATE NAMESPACE IF NOT EXISTS {namespace_identifier};",
        f"DROP TABLE IF EXISTS {table_identifier};",
        f"CREATE TABLE {table_identifier} (id BIGINT, payload STRING) USING iceberg;",
        f"INSERT INTO {table_identifier} VALUES (1, 'alpha'), (2, 'beta');",
        f"REFRESH TABLE {table_identifier};",
        f"SELECT COUNT(*) AS row_count FROM {table_identifier};",
    ]
    if cleanup:
        statements.extend(
            [
                f"DROP TABLE IF EXISTS {table_identifier};",
                f"DROP NAMESPACE IF EXISTS {namespace_identifier};",
            ]
        )
    return "\n".join(statements) + "\n"


def shell_join(parts: list[str]) -> str:
    return " ".join(shlex.quote(part) for part in parts)


def sql_string(value: str) -> str:
    return "'" + value.replace("'", "''") + "'"


def safe_file_segment(value: str) -> str:
    safe = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-")
    return safe or "value"


def spark_packages(*, spark_version: str, iceberg_version: str, scala_version: str) -> str:
    spark_minor = ".".join(spark_version.split(".")[:2])
    return ",".join(
        [
            f"org.apache.iceberg:iceberg-spark-runtime-{spark_minor}_{scala_version}:{iceberg_version}",
            f"org.apache.iceberg:iceberg-aws-bundle:{iceberg_version}",
        ]
    )


def spark_sql_command(*, packages: str, config: OrderedDict[str, str], sql_file: str) -> str:
    parts = ["spark-sql", "--packages", packages]
    for key, value in config.items():
        parts.extend(["--conf", f"{key}={value}"])
    parts.extend(["-f", sql_file])
    return shell_join(parts)


def trino_catalog_properties(
    *,
    endpoint: str,
    warehouse: str,
    access_key: str,
    secret_key: str,
    region: str,
    rest_path: str,
    rest_signing_name: str,
) -> str:
    endpoint = normalized_endpoint(endpoint)
    rest_path = normalized_rest_path(rest_path)
    properties = OrderedDict(
        [
            ("connector.name", "iceberg"),
            ("iceberg.catalog.type", "rest"),
            ("iceberg.rest-catalog.uri", f"{endpoint}{rest_path}"),
            ("iceberg.rest-catalog.warehouse", warehouse),
            ("iceberg.rest-catalog.security", "SIGV4"),
            ("iceberg.rest-catalog.signing-name", rest_signing_name),
            ("iceberg.security", "read_only"),
            ("fs.native-s3.enabled", "true"),
            ("s3.endpoint", endpoint),
            ("s3.path-style-access", "true"),
            ("s3.aws-access-key", access_key),
            ("s3.aws-secret-key", secret_key),
        ]
    )
    return "\n".join(f"{key}={value}" for key, value in properties.items()) + "\n"


def trino_sql_probe(*, namespace: str, table: str) -> str:
    return f"SELECT COUNT(*) AS row_count FROM {quote_double_identifier(namespace)}.{quote_double_identifier(table)};"


def trino_command(*, server: str, catalog_name: str, namespace: str, table: str) -> str:
    return shell_join(
        [
            "trino",
            "--server",
            server,
            "--catalog",
            catalog_name,
            "--schema",
            namespace,
            "--execute",
            trino_sql_probe(namespace=namespace, table=table),
        ]
    )


def default_metadata_location(*, warehouse: str) -> str:
    return f"s3://{warehouse}/tables/<table-id>/metadata/<current>.metadata.json"


def duckdb_sql_probe(
    *,
    endpoint: str,
    access_key: str,
    secret_key: str,
    region: str,
    metadata_location: str,
) -> str:
    parsed = re.match(r"^(https?)://(.+)$", normalized_endpoint(endpoint))
    if not parsed:
        raise ValueError("DuckDB probe endpoint must include http:// or https://")
    scheme, endpoint_without_scheme = parsed.groups()
    statements = [
        "INSTALL httpfs;",
        "LOAD httpfs;",
        "INSTALL iceberg;",
        "LOAD iceberg;",
        f"SET s3_endpoint={sql_string(endpoint_without_scheme)};",
        f"SET s3_access_key_id={sql_string(access_key)};",
        f"SET s3_secret_access_key={sql_string(secret_key)};",
        f"SET s3_region={sql_string(region)};",
        "SET s3_url_style='path';",
        f"SET s3_use_ssl={'true' if scheme == 'https' else 'false'};",
        f"SELECT COUNT(*) AS row_count FROM iceberg_scan({sql_string(metadata_location)});",
    ]
    return "\n".join(statements) + "\n"


def duckdb_command() -> str:
    return shell_join(["duckdb", "-c", ".read /tmp/rustfs-s3tables-duckdb-read.sql"])


def snowflake_sql_template(*, endpoint: str, warehouse: str, rest_path: str, namespace: str, table: str) -> str:
    catalog_uri = f"{normalized_endpoint(endpoint)}{normalized_rest_path(rest_path)}"
    statements = [
        "-- Reference template: replace credentials and storage integration names before running.",
        "CREATE OR REPLACE EXTERNAL VOLUME rustfs_s3tables_volume",
        f"  STORAGE_LOCATIONS = ((NAME = 'rustfs' STORAGE_PROVIDER = 'S3' STORAGE_BASE_URL = {sql_string(f's3://{warehouse}/')}));",
        "CREATE OR REPLACE CATALOG INTEGRATION rustfs_s3tables_catalog",
        "  CATALOG_SOURCE = ICEBERG_REST",
        "  TABLE_FORMAT = ICEBERG",
        f"  REST_CONFIG = (CATALOG_URI = {sql_string(catalog_uri)})",
        "  ENABLED = TRUE;",
        "CREATE OR REPLACE ICEBERG TABLE rustfs_s3tables_probe",
        f"  CATALOG = 'rustfs_s3tables_catalog' CATALOG_NAMESPACE = {sql_string(namespace)} CATALOG_TABLE_NAME = {sql_string(table)}",
        "  EXTERNAL_VOLUME = 'rustfs_s3tables_volume';",
        "SELECT COUNT(*) AS row_count FROM rustfs_s3tables_probe;",
    ]
    return "\n".join(statements) + "\n"


def databend_sql_probe(*, endpoint: str, warehouse: str, access_key: str, secret_key: str) -> str:
    statements = [
        "CREATE STAGE IF NOT EXISTS rustfs_s3tables_stage",
        f"  URL = {sql_string(f's3://{warehouse}/tables/')}",
        "  CONNECTION = (",
        f"    ACCESS_KEY_ID = {sql_string(access_key)}",
        f"    SECRET_ACCESS_KEY = {sql_string(secret_key)}",
        f"    ENDPOINT_URL = {sql_string(normalized_endpoint(endpoint))}",
        "    ENABLE_VIRTUAL_HOST_STYLE = 'false'",
        "  );",
        "SELECT COUNT(*) AS row_count",
        "FROM @rustfs_s3tables_stage",
        "  (PATTERN => '.*[.]parquet', FILE_FORMAT => (TYPE => PARQUET));",
    ]
    return "\n".join(statements) + "\n"


def databend_command(*, dsn: str) -> str:
    return shell_join(["databend-sql", "--dsn", dsn, "-f", "/tmp/rustfs-s3tables-databend-stage.sql"])


def rest_api_prefix(rest_path: str) -> str:
    prefix = normalized_rest_path(rest_path)
    if not prefix.endswith("/v1"):
        prefix = f"{prefix}/v1"
    return prefix


def rest_path_segment(value: str) -> str:
    return urllib.parse.quote(value, safe="")


def warehouse_catalog_path(*, warehouse: str, rest_path: str, suffix: str = "") -> str:
    return f"{rest_api_prefix(rest_path)}/{rest_path_segment(warehouse)}{suffix}"


def table_catalog_path(*, warehouse: str, namespace: str, table: str, rest_path: str, suffix: str = "") -> str:
    return (
        f"{rest_api_prefix(rest_path)}/{rest_path_segment(warehouse)}"
        f"/namespaces/{rest_path_segment(namespace)}/tables/{rest_path_segment(table)}{suffix}"
    )


def live_conformance_evidence(
    *,
    warehouse: str,
    namespace: str,
    table: str,
    rest_path: str,
    metadata_location: str,
    cleanup: bool,
) -> OrderedDict[str, Any]:
    return OrderedDict(
        [
            ("result", "operator-recorded"),
            (
                "required_run_metadata",
                [
                    "rustfs_build",
                    "git_sha",
                    "catalog_backing",
                    "endpoint",
                    "warehouse",
                    "rest_path",
                    "namespace",
                    "table",
                    "metadata_location",
                    "client_name",
                    "client_version",
                    "run_timestamp_utc",
                    "operator",
                    "expected_status",
                    "observed_status",
                    "row_count",
                    "cleanup_result",
                ],
            ),
            (
                "result_table_template",
                [
                    OrderedDict(
                        [
                            ("client", "PyIceberg"),
                            ("scenario", "create-append-reload-scan-direct-rest-probes"),
                            ("expected_status", "pass"),
                            ("expected_row_count", 2),
                            ("claim_after_pass", "automated-smoke"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("client", "Spark Iceberg REST catalog"),
                            ("scenario", "create-namespace-create-table-append-refresh-count-cleanup"),
                            ("expected_status", "pass"),
                            ("expected_row_count", 2),
                            ("claim_after_pass", "manual-live-verified"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("client", "Trino Iceberg REST catalog"),
                            ("scenario", "read-count-existing-table"),
                            ("expected_status", "pass"),
                            ("expected_row_count", 2),
                            ("claim_after_pass", "manual-live-read-verified"),
                            ("write_claim_after_pass", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("client", "DuckDB Iceberg"),
                            ("scenario", "iceberg-scan-current-metadata-location"),
                            ("expected_status", "pass"),
                            ("expected_row_count", 2),
                            ("claim_after_pass", "manual-live-read-verified"),
                            ("write_claim_after_pass", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("client", "Databend"),
                            ("scenario", "s3-stage-read-table-data-files"),
                            ("expected_status", "pass"),
                            ("claim_after_pass", "manual-live-s3-stage-verified"),
                            ("iceberg_rest_catalog_claim_after_pass", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("client", "Snowflake Open Catalog / Iceberg integrations"),
                            ("scenario", "operator-adapted-external-volume-reference"),
                            ("expected_status", "operator-recorded"),
                            ("claim_after_pass", "reference-only"),
                            ("live_rustfs_interoperability_after_pass", "not-claimed"),
                        ]
                    ),
                ],
            ),
            (
                "promotion_rules",
                [
                    "Keep PyIceberg as the only automated claim unless the run is executed by CI or a repeatable operator job.",
                    "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.",
                    "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.",
                ],
            ),
            (
                "target",
                OrderedDict(
                    [
                        ("warehouse", warehouse),
                        ("namespace", namespace),
                        ("table", table),
                        ("rest_path", rest_path),
                        ("metadata_location", metadata_location),
                        ("cleanup", cleanup),
                    ]
                ),
            ),
        ]
    )


def production_operations_guide(
    *,
    endpoint: str,
    warehouse: str,
    namespace: str,
    table: str,
    rest_path: str,
) -> OrderedDict[str, Any]:
    endpoint = normalized_endpoint(endpoint)
    table_base = table_catalog_path(warehouse=warehouse, namespace=namespace, table=table, rest_path=rest_path)
    pyiceberg_smoke_command = shell_join(
        [
            "python3",
            "scripts/table-catalog/pyiceberg_smoke.py",
            "--profile",
            "rustfs",
            "--endpoint",
            endpoint,
            "--bucket",
            warehouse,
            "--namespace",
            namespace,
            "--table",
            table,
            "--rest-path",
            rest_path,
            "--replace",
            "--cleanup",
        ]
    )
    live_conformance_command = shell_join(
        [
            "python3",
            "scripts/table-catalog/engine_compatibility.py",
            "--endpoint",
            endpoint,
            "--warehouse",
            warehouse,
            "--namespace",
            namespace,
            "--table",
            table,
            "--rest-path",
            rest_path,
            "--print-live-conformance",
            "--cleanup",
        ]
    )
    return OrderedDict(
        [
            ("claim_boundary", "Iceberg REST Catalog S3 Tables implementation"),
            (
                "target",
                OrderedDict(
                    [
                        ("endpoint", endpoint),
                        ("warehouse", warehouse),
                        ("namespace", namespace),
                        ("table", table),
                        ("rest_path", rest_path),
                    ]
                ),
            ),
            (
                "sections",
                [
                    OrderedDict(
                        [
                            ("name", "live-client-conformance"),
                            ("objective", "Record repeatable client evidence before expanding engine compatibility claims."),
                            (
                                "commands",
                                [
                                    pyiceberg_smoke_command,
                                    live_conformance_command,
                                ],
                            ),
                            (
                                "required_evidence",
                                [
                                    "RustFS build and git SHA",
                                    "catalog backing mode",
                                    "client name and version",
                                    "generated command or SQL",
                                    "expected and observed status",
                                    "row_count or response status",
                                    "current metadata location",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "PyIceberg append/reload/scan returns row_count=2",
                                    "Spark manual/live run returns row_count=2 before cleanup",
                                    "read-only engines do not claim write compatibility",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "missing run metadata",
                                    "manual-live run fails",
                                    "client writes are attempted through a read-only probe",
                                ],
                            ),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "catalog-backing-cutover"),
                            ("objective", "Verify durable backing readiness before selecting a non-object-backed catalog mode."),
                            ("commands", [f"GET {warehouse_catalog_path(warehouse=warehouse, rest_path=rest_path, suffix='/catalog/migration')}"]),
                            (
                                "required_evidence",
                                [
                                    "migration dry-run response",
                                    "commit recovery blockers",
                                    "idempotency index readiness",
                                    "warehouse prefix index readiness",
                                    "rollback configuration",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "blockers is empty",
                                    "recommended actions are completed",
                                    "object-backed catalog backup exists before cutover",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "non-empty blockers",
                                    "recoverable commit gaps require repair",
                                    "warehouse prefix index is missing or stale",
                                ],
                            ),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "maintenance-operations"),
                            ("objective", "Validate controlled maintenance without claiming a continuous in-process scheduler."),
                            (
                                "commands",
                                [
                                    f"POST {table_base}/maintenance/metadata",
                                    f"GET {table_base}/maintenance/jobs/{{job}}",
                                ],
                            ),
                            (
                                "required_evidence",
                                [
                                    "dry-run report before delete or rewrite",
                                    "current pointer re-read",
                                    "audit-events for planning and worker transitions",
                                    "quarantine status for failed jobs",
                                    "safe-window configuration",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "unreachable object deletion candidates pass safety-window checks",
                                    "rewrite groups stay partition-local and sort-order-local",
                                    "delete-file or row-level maintenance remains manual review",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "stale maintenance plan",
                                    "quarantine required",
                                    "row-level delete file rewrite requested",
                                    "missing referenced object",
                                ],
                            ),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "recovery-and-disaster-rehearsal"),
                            ("objective", "Exercise operator recovery paths without moving the table pointer unexpectedly."),
                            (
                                "commands",
                                [
                                    f"GET {table_base}/catalog/diagnostics",
                                    f"POST {table_base}/catalog/recovery",
                                    "python3 scripts/table-catalog/failure_coverage.py --print-disaster-recovery-rehearsal",
                                ],
                            ),
                            (
                                "required_evidence",
                                [
                                    "catalog export baseline",
                                    "diagnostics recovery status",
                                    "safe repair result",
                                    "rollback or import response when used",
                                    "post-recovery loadTable result",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "safe repair does not move the current pointer",
                                    "rollback/import uses normal catalog validation",
                                    "post-recovery loadTable returns the expected metadata location",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "manual-review diagnostics",
                                    "stale rollback or import conflict",
                                    "post-recovery data-plane policy failure",
                                ],
                            ),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "permissions-and-credentials"),
                            ("objective", "Verify table policy and credential vending cannot bypass table warehouse scope."),
                            (
                                "commands",
                                [
                                    "python3 scripts/table-catalog/pyiceberg_smoke.py --profile rustfs-vended-credentials --replace --cleanup",
                                    f"GET {table_base}/credentials",
                                ],
                            ),
                            (
                                "required_evidence",
                                [
                                    "catalog principal permissions",
                                    "vended credential expiration",
                                    "warehouse prefix scope",
                                    "inside-prefix S3 put/head/get/delete result",
                                    "outside-prefix deny result",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "default credentials response is empty unless vending is enabled",
                                    "vended credentials work inside the exact table prefix",
                                    "vended credentials are denied outside the exact table prefix",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "long-lived storage secret returned by catalog",
                                    "prefix mismatch",
                                    "outside-prefix object access succeeds",
                                ],
                            ),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "unsupported-claim-governance"),
                            ("objective", "Keep release language aligned with verified RustFS behavior."),
                            (
                                "commands",
                                [
                                    "python3 scripts/table-catalog/pyiceberg_smoke.py --print-unsupported-inventory",
                                    "python3 scripts/table-catalog/engine_compatibility.py --print-engine-matrix",
                                ],
                            ),
                            (
                                "not_claimed",
                                [
                                    "full AWS S3 Tables control-plane API parity",
                                    "full MinIO AIStor private extension parity",
                                    "full Cloudflare R2 Data Catalog interoperability",
                                    "active-active multi-region table writes",
                                    "multi-table transactions",
                                    "built-in SQL query execution",
                                    "Delta Lake or Hudi table format support",
                                    "end-to-end SQL row-level DML validation",
                                ],
                            ),
                            (
                                "required_evidence",
                                [
                                    "support matrix status label",
                                    "unsupported inventory entry",
                                    "live evidence before any promotion",
                                ],
                            ),
                            (
                                "pass_criteria",
                                [
                                    "public claim wording matches support matrix status",
                                    "reference profiles stay reference-only without live evidence",
                                ],
                            ),
                            (
                                "fail_closed_signals",
                                [
                                    "vendor parity wording without recorded live evidence",
                                    "manual probe documented as automated support",
                                ],
                            ),
                        ]
                    ),
                ],
            ),
        ]
    )


def live_conformance_harness(
    *,
    endpoint: str,
    warehouse: str,
    access_key: str,
    secret_key: str,
    region: str,
    catalog_name: str,
    namespace: str,
    table: str,
    rest_path: str,
    rest_signing_name: str,
    pyiceberg_version: str,
    spark_version: str,
    iceberg_version: str,
    scala_version: str,
    trino_version: str = DEFAULT_TRINO_VERSION,
    trino_server: str = DEFAULT_TRINO_SERVER,
    duckdb_version: str = DEFAULT_DUCKDB_VERSION,
    snowflake_client_version: str = DEFAULT_SNOWFLAKE_CLIENT_VERSION,
    databend_version: str = DEFAULT_DATABEND_VERSION,
    databend_dsn: str = DEFAULT_DATABEND_DSN,
    metadata_location: str | None = None,
    cleanup: bool = False,
) -> OrderedDict[str, Any]:
    endpoint = normalized_endpoint(endpoint)
    rest_path = normalized_rest_path(rest_path)
    metadata_location = metadata_location or default_metadata_location(warehouse=warehouse)
    sql_file = f"/tmp/rustfs-s3tables-{safe_file_segment(namespace)}-{safe_file_segment(table)}-spark.sql"
    spark_config = spark_catalog_config(
        endpoint=endpoint,
        warehouse=warehouse,
        access_key=access_key,
        secret_key=secret_key,
        region=region,
        catalog_name=catalog_name,
        rest_path=rest_path,
        rest_signing_name=rest_signing_name,
    )
    sql = spark_sql_smoke(catalog_name=catalog_name, namespace=namespace, table=table, cleanup=cleanup)
    packages = spark_packages(spark_version=spark_version, iceberg_version=iceberg_version, scala_version=scala_version)
    trino_properties = trino_catalog_properties(
        endpoint=endpoint,
        warehouse=warehouse,
        access_key=access_key,
        secret_key=secret_key,
        region=region,
        rest_path=rest_path,
        rest_signing_name=rest_signing_name,
    )
    duckdb_sql = duckdb_sql_probe(
        endpoint=endpoint,
        access_key=access_key,
        secret_key=secret_key,
        region=region,
        metadata_location=metadata_location,
    )
    snowflake_sql = snowflake_sql_template(
        endpoint=endpoint,
        warehouse=warehouse,
        rest_path=rest_path,
        namespace=namespace,
        table=table,
    )
    databend_sql = databend_sql_probe(endpoint=endpoint, warehouse=warehouse, access_key=access_key, secret_key=secret_key)
    pyiceberg_command = shell_join(
        [
            "python3",
            "scripts/table-catalog/pyiceberg_smoke.py",
            "--endpoint",
            endpoint,
            "--access-key",
            access_key,
            "--secret-key",
            secret_key,
            "--bucket",
            warehouse,
            "--namespace",
            namespace,
            "--table",
            table,
            "--rest-path",
            rest_path,
            "--rest-signing-name",
            rest_signing_name,
            "--replace",
            *([] if not cleanup else ["--cleanup"]),
        ]
    )
    return OrderedDict(
        [
            ("mode", "manual-or-ci-optional"),
            ("ci_gate", "RUSTFS_TABLE_CATALOG_LIVE_CONFORMANCE=1"),
            (
                "prerequisites",
                [
                    "RustFS endpoint is reachable",
                    "table bucket can be created or reused",
                    "Python, PyIceberg, PyArrow, boto3, Spark, and Iceberg Spark runtime packages are available",
                ],
            ),
            (
                "expected_results",
                OrderedDict(
                    [
                        ("namespace", namespace),
                        ("table", table),
                        ("row_count", 2),
                        ("cleanup", cleanup),
                    ]
                ),
            ),
            (
                "evidence",
                live_conformance_evidence(
                    warehouse=warehouse,
                    namespace=namespace,
                    table=table,
                    rest_path=rest_path,
                    metadata_location=metadata_location,
                    cleanup=cleanup,
                ),
            ),
            (
                "clients",
                [
                    OrderedDict(
                        [
                            ("name", "PyIceberg"),
                            ("status", "automated-smoke"),
                            ("version", pyiceberg_version),
                            ("install", shell_join(["python3", "-m", "pip", "install", f"pyiceberg[pyarrow]=={pyiceberg_version}", "boto3"])),
                            ("command", pyiceberg_command),
                            ("expected", "append two rows, reload the table, scan row_count=2, and run direct REST catalog probes"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "Spark Iceberg REST catalog"),
                            ("status", "manual-live-harness"),
                            ("spark_version", spark_version),
                            ("iceberg_version", iceberg_version),
                            ("scala_version", scala_version),
                            ("packages", packages),
                            ("spark_config", spark_config),
                            ("sql_file", sql_file),
                            (
                                "prepare_sql",
                                shell_join(
                                    [
                                        "python3",
                                        "scripts/table-catalog/engine_compatibility.py",
                                        "--print-spark-sql",
                                        "--namespace",
                                        namespace,
                                        "--table",
                                        table,
                                        "--catalog-name",
                                        catalog_name,
                                        *([] if not cleanup else ["--cleanup"]),
                                    ]
                                )
                                + f" > {shlex.quote(sql_file)}",
                            ),
                            ("sql", sql),
                            ("command", spark_sql_command(packages=packages, config=spark_config, sql_file=sql_file)),
                            ("expected", "Spark SQL SELECT COUNT(*) returns row_count=2 before optional cleanup"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "Trino Iceberg REST catalog"),
                            ("status", "manual-live-read-probe"),
                            ("version", trino_version),
                            ("catalog_file", f"/tmp/{safe_file_segment(catalog_name)}.properties"),
                            ("catalog_properties", trino_properties),
                            ("sql", trino_sql_probe(namespace=namespace, table=table)),
                            ("command", trino_command(server=trino_server, catalog_name=catalog_name, namespace=namespace, table=table)),
                            ("expected", "SELECT COUNT(*) returns row_count=2 for a table already written by PyIceberg or Spark"),
                            ("write_compatibility", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "DuckDB Iceberg"),
                            ("status", "manual-live-read-probe"),
                            ("version", duckdb_version),
                            ("metadata_location", metadata_location),
                            ("sql_file", "/tmp/rustfs-s3tables-duckdb-read.sql"),
                            ("sql", duckdb_sql),
                            ("command", duckdb_command()),
                            ("expected", "iceberg_scan returns row_count=2 when metadata_location points at the current Iceberg metadata JSON"),
                            ("write_compatibility", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "Snowflake Open Catalog / Iceberg integrations"),
                            ("status", "manual-reference-probe"),
                            ("client_version", snowflake_client_version),
                            ("sql", snowflake_sql),
                            ("expected", "operator-adapted reference query returns row_count=2 after external volume and catalog integration setup"),
                            ("live_rustfs_interoperability", "not-claimed"),
                        ]
                    ),
                    OrderedDict(
                        [
                            ("name", "Databend"),
                            ("status", "manual-live-s3-stage-probe"),
                            ("version", databend_version),
                            ("sql_file", "/tmp/rustfs-s3tables-databend-stage.sql"),
                            ("sql", databend_sql),
                            ("command", databend_command(dsn=databend_dsn)),
                            ("expected", "S3 stage read probe returns row_count for Parquet data files under the table warehouse"),
                            ("iceberg_rest_catalog", "not-claimed"),
                        ]
                    ),
                ],
            ),
        ]
    )


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Print RustFS S3 Tables Iceberg engine compatibility helpers.")
    parser.add_argument("--endpoint", default="http://127.0.0.1:9000")
    parser.add_argument("--access-key", default="rustfsadmin")
    parser.add_argument("--secret-key", default="rustfsadmin")
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--warehouse", default="rustfs-s3table-smoke")
    parser.add_argument("--profile", choices=sorted(VENDOR_SPARK_PROFILES), default="rustfs")
    parser.add_argument("--account-id", default="000000000000")
    parser.add_argument("--table-bucket", default="rustfs-s3table-smoke")
    parser.add_argument("--catalog-uri")
    parser.add_argument("--warehouse-name")
    parser.add_argument("--namespace", default="smoke")
    parser.add_argument("--table", default="events")
    parser.add_argument("--catalog-name", default="rustfs")
    parser.add_argument("--rest-path")
    parser.add_argument("--rest-signing-name")
    parser.add_argument("--pyiceberg-version", default=DEFAULT_PYICEBERG_VERSION)
    parser.add_argument("--spark-version", default=DEFAULT_SPARK_VERSION)
    parser.add_argument("--iceberg-version", default=DEFAULT_ICEBERG_VERSION)
    parser.add_argument("--scala-version", default=DEFAULT_SCALA_VERSION)
    parser.add_argument("--trino-version", default=DEFAULT_TRINO_VERSION)
    parser.add_argument("--trino-server", default=DEFAULT_TRINO_SERVER)
    parser.add_argument("--duckdb-version", default=DEFAULT_DUCKDB_VERSION)
    parser.add_argument("--snowflake-client-version", default=DEFAULT_SNOWFLAKE_CLIENT_VERSION)
    parser.add_argument("--databend-version", default=DEFAULT_DATABEND_VERSION)
    parser.add_argument("--databend-dsn", default=DEFAULT_DATABEND_DSN)
    parser.add_argument("--metadata-location")
    parser.add_argument("--cleanup", action="store_true")
    parser.add_argument("--print-engine-matrix", action="store_true")
    parser.add_argument("--print-live-conformance", action="store_true")
    parser.add_argument("--print-operations-guide", action="store_true")
    parser.add_argument("--print-spark-config", action="store_true")
    parser.add_argument("--print-spark-sql", action="store_true")
    return parser.parse_args(argv)


def print_json(document: Any, output: StringIO | None = None) -> None:
    text = json.dumps(document, indent=2, sort_keys=True)
    if output is None:
        print(text)
    else:
        output.write(f"{text}\n")


def cli_json(argv: list[str]) -> str:
    output = StringIO()
    run(parse_args(argv), output)
    return output.getvalue()


def run(args: argparse.Namespace, output: StringIO | None = None) -> None:
    printed = False
    if args.print_engine_matrix:
        print_json({"engine_compatibility": engine_compatibility_matrix()}, output)
        printed = True
    if args.print_spark_config:
        print_json(
            {
                "spark_config": spark_vendor_catalog_config(
                    profile=args.profile,
                    endpoint=args.endpoint,
                    warehouse=args.warehouse,
                    access_key=args.access_key,
                    secret_key=args.secret_key,
                    region=args.region,
                    catalog_name=args.catalog_name,
                    account_id=args.account_id,
                    table_bucket=args.table_bucket,
                    catalog_uri=args.catalog_uri,
                    warehouse_name=args.warehouse_name,
                    rest_path=args.rest_path,
                    rest_signing_name=args.rest_signing_name,
                )
            },
            output,
        )
        printed = True
    if args.print_live_conformance:
        print_json(
            {
                "live_conformance": live_conformance_harness(
                    endpoint=args.endpoint,
                    warehouse=args.warehouse,
                    access_key=args.access_key,
                    secret_key=args.secret_key,
                    region=args.region,
                    catalog_name=args.catalog_name,
                    namespace=args.namespace,
                    table=args.table,
                    rest_path=args.rest_path or "/iceberg",
                    rest_signing_name=args.rest_signing_name or "s3",
                    pyiceberg_version=args.pyiceberg_version,
                    spark_version=args.spark_version,
                    iceberg_version=args.iceberg_version,
                    scala_version=args.scala_version,
                    trino_version=args.trino_version,
                    trino_server=args.trino_server,
                    duckdb_version=args.duckdb_version,
                    snowflake_client_version=args.snowflake_client_version,
                    databend_version=args.databend_version,
                    databend_dsn=args.databend_dsn,
                    metadata_location=args.metadata_location,
                    cleanup=args.cleanup,
                )
            },
            output,
        )
        printed = True
    if args.print_operations_guide:
        print_json(
            {
                "production_operations_guide": production_operations_guide(
                    endpoint=args.endpoint,
                    warehouse=args.warehouse,
                    namespace=args.namespace,
                    table=args.table,
                    rest_path=args.rest_path or "/iceberg",
                )
            },
            output,
        )
        printed = True
    if args.print_spark_sql:
        sql = spark_sql_smoke(
            catalog_name=args.catalog_name,
            namespace=args.namespace,
            table=args.table,
            cleanup=args.cleanup,
        )
        if output is None:
            print(sql, end="")
        else:
            output.write(sql)
        printed = True
    if not printed:
        print_json({"engine_compatibility": engine_compatibility_matrix()}, output)


def main() -> None:
    run(parse_args())


if __name__ == "__main__":
    main()