From 2c6881695f5943978768ed8aac2aa1a87c834245 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Fri, 19 Jun 2026 08:49:59 +0800 Subject: [PATCH] test(table-catalog): add Iceberg compatibility and failure coverage (#3581) * test(table-catalog): add Iceberg client compatibility coverage * test(table-catalog): add production failure coverage * fix(table-catalog): generate valid failure probe requests --------- Co-authored-by: Henry Guo --- scripts/table-catalog/README.md | 106 ++++++- scripts/table-catalog/engine_compatibility.py | 264 ++++++++++++++++++ scripts/table-catalog/failure_coverage.py | 251 +++++++++++++++++ scripts/table-catalog/pyiceberg_smoke.py | 15 + .../test_engine_compatibility.py | 107 +++++++ .../table-catalog/test_failure_coverage.py | 90 ++++++ scripts/table-catalog/test_pyiceberg_smoke.py | 28 ++ 7 files changed, 853 insertions(+), 8 deletions(-) create mode 100644 scripts/table-catalog/engine_compatibility.py create mode 100644 scripts/table-catalog/failure_coverage.py create mode 100644 scripts/table-catalog/test_engine_compatibility.py create mode 100644 scripts/table-catalog/test_failure_coverage.py diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 9b0153a6b..e7c34072b 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -123,6 +123,8 @@ PyIceberg, PyArrow, or boto3: ```bash python3 scripts/table-catalog/pyiceberg_smoke.py --print-client-matrix +python3 scripts/table-catalog/pyiceberg_smoke.py --print-engine-compatibility +python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-failure-coverage python3 scripts/table-catalog/pyiceberg_smoke.py --print-vendor-profiles python3 scripts/table-catalog/pyiceberg_smoke.py --print-unsupported-inventory python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-readiness @@ -130,8 +132,41 @@ python3 scripts/table-catalog/pyiceberg_smoke.py --print-production-readiness 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; other engines are documented until a repeatable harness is -added. +this script today. Spark now has generated configuration and SQL smoke input; +other engines are documented until a repeatable harness is added. + +The standalone engine helper prints the same compatibility matrix and can also +generate Spark REST catalog input without importing PyIceberg: + +```bash +python3 scripts/table-catalog/engine_compatibility.py --print-engine-matrix +python3 scripts/table-catalog/engine_compatibility.py --print-spark-config +python3 scripts/table-catalog/engine_compatibility.py --print-spark-sql --cleanup +``` + +The production failure helper records the negative coverage required before +calling a release production-ready and can generate REST probe steps for a live +RustFS endpoint: + +```bash +python3 scripts/table-catalog/failure_coverage.py --print-failure-matrix +python3 scripts/table-catalog/failure_coverage.py \ + --warehouse rustfs-s3table-smoke \ + --namespace smoke \ + --table events \ + --rest-path /iceberg \ + --print-failure-probes +``` + +The generated probe plan covers stale-token commit conflicts, missing metadata +object rejection, diagnostics/recovery for finalization gaps, maintenance stale +plan rejection, and external catalog sync conflicts. These steps are meant to be +run against a prepared live table and should be recorded with the exact RustFS +build and client versions used. + +`--rest-path` defaults to `/iceberg` and generated probe paths include the +mounted catalog prefix, for example `/iceberg/v1/{warehouse}/...`. Use +`--rest-path /_iceberg` to generate paths for the compatibility alias. The smoke test also probes catalog-backed advanced Iceberg surfaces: @@ -153,12 +188,41 @@ The smoke test also probes catalog-backed advanced Iceberg surfaces: | Client | Current status | Claim | |---|---|---| | 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-ready | create/load/append/reload should be verified against a running RustFS endpoint | +| Spark Iceberg REST catalog | Generated smoke harness | configuration and SQL can be generated for create namespace, create table, append, reload, count, and cleanup against a running RustFS endpoint | | Trino Iceberg REST catalog | Documented, not automated | no write compatibility claim yet | | DuckDB Iceberg | Documented, not automated | read-path reference only | +| StarRocks Iceberg REST catalog | Documented, not automated | external catalog read-path reference only | | Databend | Documented, not automated | S3 data-plane reference only; Iceberg REST catalog integration is not claimed | | Snowflake/Open Catalog integrations | Documented, not automated | reference only | +## Production Failure Coverage + +Production failure coverage is tracked separately from positive client +conformance. Positive smoke tests prove a client can create and use a table; +failure probes prove RustFS does not silently advance table state when something +goes wrong. + +The current failure matrix covers: + +- stale commit tokens and stale base metadata returning a conflict without + advancing the table pointer +- post-CAS finalization gaps surfacing through diagnostics and safe recovery + repair without pointer movement +- missing metadata, manifest, data, or delete objects failing closed before a + commit or maintenance operation advances state +- concurrent writers producing a single winning CAS and retryable conflicts +- table catalog and ordinary S3 object permission denials preventing data-plane + bypass +- stale maintenance plans failing closed before object deletion or catalog + commit +- external catalog sync conflicts leaving pointer, token, and generation + unchanged +- backing migration remaining blocked until WAL/recovery replay is clean + +Do not promote a failure case from `probe-required` or `load-test-required` to +an automated claim until the live probe or stress harness is repeatable and its +RustFS build, client version, and expected response shape are recorded. + ## Vendor Profile References | Profile | Catalog shape | Signing name | Credential model | RustFS claim | @@ -221,10 +285,21 @@ The TTL is clamped to the supported short-lived range by the server. ## Spark Manual Baseline Spark validation should use the same RustFS endpoint and warehouse bucket as the -PyIceberg smoke test. The exact package version should be recorded in the client -matrix after each run. +PyIceberg smoke test. The exact Spark and Iceberg package versions should be +recorded in the client matrix after each run. -Minimum configuration shape: +Generate the configuration properties: + +```bash +python3 scripts/table-catalog/engine_compatibility.py \ + --endpoint http://127.0.0.1:9000 \ + --warehouse rustfs-s3table-smoke \ + --access-key rustfsadmin \ + --secret-key rustfsadmin \ + --print-spark-config +``` + +The generated configuration shape is: ```properties spark.sql.catalog.rustfs=org.apache.iceberg.spark.SparkCatalog @@ -237,7 +312,22 @@ spark.sql.catalog.rustfs.s3.path-style-access=true spark.sql.catalog.rustfs.rest.sigv4-enabled=true spark.sql.catalog.rustfs.rest.signing-name=s3 spark.sql.catalog.rustfs.rest.signing-region=us-east-1 +spark.sql.catalog.rustfs.s3.access-key-id=rustfsadmin +spark.sql.catalog.rustfs.s3.secret-access-key=rustfsadmin ``` -Until Spark is automated, do not claim Spark support beyond a manually verified -run with the exact Spark and Iceberg versions recorded. +Generate the SQL smoke input: + +```bash +python3 scripts/table-catalog/engine_compatibility.py \ + --catalog-name rustfs \ + --namespace smoke \ + --table events \ + --print-spark-sql \ + --cleanup +``` + +The generated SQL covers namespace creation, table creation, append, refresh, +count, and optional cleanup. Until Spark execution is automated in CI, do not +claim Spark support beyond a manually verified run with the exact Spark and +Iceberg versions recorded. diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py new file mode 100644 index 000000000..da894dd32 --- /dev/null +++ b/scripts/table-catalog/engine_compatibility.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +"""Machine-readable Iceberg engine compatibility helpers for RustFS S3 Tables.""" + +from __future__ import annotations + +import argparse +import json +from collections import OrderedDict +from io import StringIO +from typing import Any + + +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": "generated-smoke-harness", + "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-spark-sql", + "scenarios": [ + scenario("create-namespace", "generated-spark-sql", "CREATE NAMESPACE IF NOT EXISTS"), + scenario("create-table", "generated-spark-sql", "CREATE TABLE USING iceberg"), + scenario("append", "generated-spark-sql", "INSERT INTO"), + scenario("reload-table", "generated-spark-sql", "REFRESH TABLE and SELECT COUNT"), + scenario("drop-table", "generated-spark-sql", "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": "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", "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": "documented-read-path", + "entrypoint": "scripts/table-catalog/README.md", + "scenarios": [ + scenario("catalog-load", "manual-validation-required", "REST catalog extension/configuration reference"), + scenario("read-table", "manual-validation-required", "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": "reference-only", + "entrypoint": "scripts/table-catalog/README.md", + "scenarios": [ + scenario("catalog-load", "not-claimed", "reference only until a repeatable external integration harness exists"), + ], + }, + { + "client": "Databend", + "status": "s3-data-plane-reference", + "entrypoint": "scripts/table-catalog/README.md", + "scenarios": [ + scenario("s3-data-plane-read", "manual-validation-required", "S3 stage/data-plane reference only"), + 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 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 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 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 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("--namespace", default="smoke") + parser.add_argument("--table", default="events") + parser.add_argument("--catalog-name", default="rustfs") + parser.add_argument("--rest-path", default="/iceberg") + parser.add_argument("--rest-signing-name", default="s3") + parser.add_argument("--cleanup", action="store_true") + parser.add_argument("--print-engine-matrix", 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_catalog_config( + endpoint=args.endpoint, + warehouse=args.warehouse, + access_key=args.access_key, + secret_key=args.secret_key, + region=args.region, + catalog_name=args.catalog_name, + rest_path=args.rest_path, + rest_signing_name=args.rest_signing_name, + ) + }, + 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() diff --git a/scripts/table-catalog/failure_coverage.py b/scripts/table-catalog/failure_coverage.py new file mode 100644 index 000000000..d8063321d --- /dev/null +++ b/scripts/table-catalog/failure_coverage.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Machine-readable production failure coverage helpers for RustFS S3 Tables.""" + +from __future__ import annotations + +import argparse +import json +from io import StringIO +from typing import Any + + +def failure_case( + case: str, + coverage_status: str, + expected_behavior: str, + evidence: str, +) -> dict[str, str]: + return { + "case": case, + "coverage_status": coverage_status, + "expected_behavior": expected_behavior, + "evidence": evidence, + } + + +def production_failure_matrix() -> list[dict[str, str]]: + return [ + failure_case( + "commit-cas-conflict", + "server-tests-plus-live-probe-required", + "conflict-without-pointer-advance", + "standard commit validation rejects stale expected token or base metadata before advancing the table pointer", + ), + failure_case( + "post-cas-finalization-gap", + "diagnostics-and-recovery-probe-required", + "recoverable-diagnostics-and-idempotency-repair", + "catalog diagnostics and recovery routes must report staged/finalization gaps and repair idempotency indexes without moving the pointer", + ), + failure_case( + "missing-referenced-object", + "server-tests-plus-live-probe-required", + "fail-closed-before-pointer-advance", + "commit validation and maintenance reachability must reject missing metadata, manifest, data, or delete objects", + ), + failure_case( + "concurrent-writer-stress", + "load-test-required", + "single-winner-cas-and-retryable-conflicts", + "multiple writers against one table should leave one committed pointer and retryable conflicts for stale writers", + ), + failure_case( + "permission-negative", + "server-tests-plus-live-probe-required", + "deny-without-data-plane-bypass", + "table catalog actions and ordinary S3 object actions must both enforce table-scoped permissions", + ), + failure_case( + "maintenance-stale-plan", + "server-tests-plus-live-probe-required", + "manual-review-or-conflict-without-delete", + "maintenance delete/commit requests must re-read current pointer and fail closed when the plan is stale", + ), + failure_case( + "external-bridge-conflict", + "server-tests-plus-live-probe-required", + "sync-conflict-without-pointer-advance", + "external catalog sync must not advance pointer/token/generation on UUID mismatch or stale external metadata token", + ), + failure_case( + "backing-migration-blocked", + "diagnostics-probe-required", + "blocked-until-recovery-and-replay-are-clean", + "catalog export and diagnostics must expose WAL/recovery blockers before any strong backing cutover", + ), + ] + + +def catalog_prefix(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}" + stripped = stripped.rstrip("/") + if not stripped.endswith("/v1"): + stripped = f"{stripped}/v1" + return stripped + + +def table_path(warehouse: str, namespace: str, table: str, suffix: str = "", rest_path: str = "/iceberg") -> str: + base = f"{catalog_prefix(rest_path)}/{warehouse}/namespaces/{namespace}/tables/{table}" + return f"{base}{suffix}" + + +def probe_step( + name: str, + method: str, + path: str, + expected_status: str, + expected_behavior: str, + body: dict[str, Any] | None = None, +) -> dict[str, Any]: + step: dict[str, Any] = { + "name": name, + "method": method, + "path": path, + "expected_status": expected_status, + "expected_behavior": expected_behavior, + } + if body is not None: + step["body"] = body + return step + + +def failure_probe_plan(warehouse: str, namespace: str, table: str, rest_path: str = "/iceberg") -> list[dict[str, Any]]: + table_endpoint = table_path(warehouse, namespace, table, rest_path=rest_path) + return [ + probe_step( + "stale-token-commit-conflict", + "POST", + table_endpoint, + "409", + "stale expected version token is rejected and current metadata pointer remains unchanged", + { + "identifier": {"namespace": [namespace], "name": table}, + "expected-version-token": "stale-token-from-previous-load", + "expected-metadata-location": "current-metadata-location-from-load-table", + "new-metadata-location": f"s3://{warehouse}/tables/table-id/metadata/conflict_probe.metadata.json", + "requirements": [ + { + "type": "assert-current-snapshot-id", + "snapshot-id": 0, + } + ], + "updates": [ + { + "action": "set-current-schema", + "schema-id": 0, + } + ], + }, + ), + probe_step( + "missing-metadata-object-rejected", + "POST", + table_endpoint, + "400", + "new metadata object must exist before the catalog pointer can advance", + { + "identifier": {"namespace": [namespace], "name": table}, + "expected-version-token": "current-version-token-from-load-table", + "expected-metadata-location": "current-metadata-location-from-load-table", + "new-metadata-location": f"s3://{warehouse}/tables/table-id/metadata/does_not_exist.metadata.json", + }, + ), + probe_step( + "diagnostics-after-finalization-gap", + "GET", + table_path(warehouse, namespace, table, "/catalog/diagnostics", rest_path), + "200", + "diagnostics should surface recoverable commit-log/idempotency gaps with operator actions", + ), + probe_step( + "recovery-repairs-idempotency-index", + "POST", + table_path(warehouse, namespace, table, "/catalog/recovery", rest_path), + "200", + "recovery should repair stale or missing idempotency indexes without moving the table pointer", + { + "mode": "safe-repair", + }, + ), + probe_step( + "maintenance-stale-plan-rejected", + "POST", + table_path(warehouse, namespace, table, "/maintenance/metadata", rest_path), + "409", + "stale maintenance plans must not delete or commit after the current pointer changes", + { + "dry-run": False, + "expected-metadata-location": "stale-metadata-location", + }, + ), + probe_step( + "external-sync-conflict", + "POST", + table_path(warehouse, namespace, table, "/catalog/import", rest_path), + "409", + "external catalog sync conflicts must leave pointer, token, and generation unchanged", + { + "external-version-token": "stale-external-token", + "metadata-location": f"s3://{warehouse}/external/metadata/metadata.json", + }, + ), + ] + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Print RustFS S3 Tables production failure coverage helpers.") + parser.add_argument("--warehouse", default="rustfs-s3table-smoke") + parser.add_argument("--namespace", default="smoke") + parser.add_argument("--table", default="events") + parser.add_argument("--rest-path", default="/iceberg") + parser.add_argument("--print-failure-matrix", action="store_true") + parser.add_argument("--print-failure-probes", 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_failure_matrix: + print_json({"production_failure_coverage": production_failure_matrix()}, output) + printed = True + if args.print_failure_probes: + print_json( + { + "failure_probe_plan": failure_probe_plan( + warehouse=args.warehouse, + namespace=args.namespace, + table=args.table, + rest_path=args.rest_path, + ) + }, + output, + ) + printed = True + if not printed: + print_json({"production_failure_coverage": production_failure_matrix()}, output) + + +def main() -> None: + run(parse_args()) + + +if __name__ == "__main__": + main() diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index f94817c46..5bc80ef45 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -16,6 +16,9 @@ import urllib.request from dataclasses import dataclass from typing import Any +import engine_compatibility +import failure_coverage + DEFAULT_PROFILE = "rustfs" CATALOG_VENDED_PROFILE = "rustfs-vended-credentials" VENDED_CREDENTIAL_PROFILES = {CATALOG_VENDED_PROFILE} @@ -298,6 +301,12 @@ def parse_args() -> argparse.Namespace: ) parser.add_argument("--insecure", action="store_true", help="Disable TLS verification for HTTPS endpoints.") parser.add_argument("--print-client-matrix", action="store_true", help="Print the current client conformance matrix as JSON and exit.") + parser.add_argument("--print-engine-compatibility", action="store_true", help="Print the current Iceberg engine compatibility matrix as JSON and exit.") + parser.add_argument( + "--print-production-failure-coverage", + action="store_true", + help="Print the current production failure coverage matrix as JSON and exit.", + ) parser.add_argument("--print-vendor-profiles", action="store_true", help="Print vendor connection profile references as JSON and exit.") parser.add_argument("--print-unsupported-inventory", action="store_true", help="Print unsupported capability inventory as JSON and exit.") parser.add_argument( @@ -721,6 +730,12 @@ def printed_metadata(args: argparse.Namespace) -> bool: if args.print_client_matrix: print_json_document({"client_matrix": client_matrix()}) printed = True + if args.print_engine_compatibility: + print_json_document({"engine_compatibility": engine_compatibility.engine_compatibility_matrix()}) + printed = True + if args.print_production_failure_coverage: + print_json_document({"production_failure_coverage": failure_coverage.production_failure_matrix()}) + printed = True if args.print_vendor_profiles: print_json_document({"vendor_profiles": vendor_profiles()}) printed = True diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py new file mode 100644 index 000000000..496fd66b0 --- /dev/null +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Unit tests for the RustFS table catalog engine compatibility helper.""" + +from __future__ import annotations + +import json +import unittest + +import engine_compatibility + + +class EngineCompatibilityTest(unittest.TestCase): + def test_matrix_records_required_engine_scenarios_without_overclaiming(self) -> None: + matrix = engine_compatibility.engine_compatibility_matrix() + by_client = {entry["client"]: entry for entry in matrix} + + self.assertIn("PyIceberg", by_client) + self.assertIn("Spark Iceberg REST catalog", by_client) + self.assertIn("Trino Iceberg REST catalog", by_client) + self.assertIn("DuckDB Iceberg", by_client) + self.assertIn("StarRocks Iceberg REST catalog", by_client) + + pyiceberg = by_client["PyIceberg"] + self.assertEqual(pyiceberg["status"], "automated-smoke") + self.assertContainsScenario(pyiceberg, "create-namespace", "automated") + self.assertContainsScenario(pyiceberg, "create-table", "automated") + self.assertContainsScenario(pyiceberg, "append", "automated") + self.assertContainsScenario(pyiceberg, "reload-table", "automated") + self.assertContainsScenario(pyiceberg, "drop-table", "automated-with-cleanup") + + spark = by_client["Spark Iceberg REST catalog"] + self.assertEqual(spark["status"], "generated-smoke-harness") + self.assertContainsScenario(spark, "create-namespace", "generated-spark-sql") + self.assertContainsScenario(spark, "create-table", "generated-spark-sql") + self.assertContainsScenario(spark, "append", "generated-spark-sql") + self.assertContainsScenario(spark, "reload-table", "generated-spark-sql") + self.assertContainsScenario(spark, "drop-table", "generated-spark-sql") + self.assertContainsScenario(spark, "commit-conflict", "manual-validation-required") + + trino = by_client["Trino Iceberg REST catalog"] + self.assertEqual(trino["status"], "documented-read-path") + self.assertContainsScenario(trino, "catalog-load", "manual-validation-required") + + def test_spark_config_uses_rustfs_rest_catalog_and_s3fileio(self) -> None: + config = engine_compatibility.spark_catalog_config( + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + rest_path="/iceberg", + rest_signing_name="s3", + ) + + self.assertEqual(config["spark.sql.catalog.rustfs"], "org.apache.iceberg.spark.SparkCatalog") + self.assertEqual(config["spark.sql.catalog.rustfs.type"], "rest") + self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "http://127.0.0.1:9000/iceberg") + self.assertEqual(config["spark.sql.catalog.rustfs.warehouse"], "rustfs-s3table-smoke") + self.assertEqual(config["spark.sql.catalog.rustfs.io-impl"], "org.apache.iceberg.aws.s3.S3FileIO") + self.assertEqual(config["spark.sql.catalog.rustfs.s3.endpoint"], "http://127.0.0.1:9000") + self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3") + + def test_spark_sql_smoke_covers_lifecycle_append_reload_and_cleanup(self) -> None: + sql = engine_compatibility.spark_sql_smoke( + catalog_name="rustfs", + namespace="sales", + table="orders", + cleanup=True, + ) + + self.assertIn("CREATE NAMESPACE IF NOT EXISTS rustfs.`sales`", sql) + self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql) + self.assertIn("CREATE TABLE rustfs.`sales`.`orders`", sql) + self.assertIn("INSERT INTO rustfs.`sales`.`orders`", sql) + self.assertIn("REFRESH TABLE rustfs.`sales`.`orders`", sql) + self.assertIn("SELECT COUNT(*) AS row_count FROM rustfs.`sales`.`orders`", sql) + self.assertIn("DROP TABLE IF EXISTS rustfs.`sales`.`orders`", sql) + self.assertIn("DROP NAMESPACE IF EXISTS rustfs.`sales`", sql) + + def test_spark_sql_rejects_unsafe_identifiers(self) -> None: + with self.assertRaisesRegex(ValueError, "Spark identifier"): + engine_compatibility.spark_sql_smoke( + catalog_name="rustfs", + namespace="sales`prod", + table="orders", + ) + + def test_cli_prints_machine_readable_engine_matrix(self) -> None: + payload = engine_compatibility.cli_json(["--print-engine-matrix"]) + document = json.loads(payload) + + self.assertIn("engine_compatibility", document) + self.assertTrue(document["engine_compatibility"]) + + def assertContainsScenario(self, entry: dict[str, object], name: str, status: str) -> None: + scenarios = entry.get("scenarios") + self.assertIsInstance(scenarios, list) + for scenario in scenarios: + if isinstance(scenario, dict) and scenario.get("name") == name: + self.assertEqual(scenario.get("status"), status) + return + self.fail(f"missing scenario {name!r} in {entry.get('client')!r}") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/table-catalog/test_failure_coverage.py b/scripts/table-catalog/test_failure_coverage.py new file mode 100644 index 000000000..9543d8c20 --- /dev/null +++ b/scripts/table-catalog/test_failure_coverage.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +"""Unit tests for RustFS table catalog production failure coverage helpers.""" + +from __future__ import annotations + +import json +import unittest + +import failure_coverage + + +class FailureCoverageTest(unittest.TestCase): + def test_matrix_tracks_production_failure_boundaries_without_overclaiming(self) -> None: + matrix = failure_coverage.production_failure_matrix() + by_case = {entry["case"]: entry for entry in matrix} + + self.assertIn("commit-cas-conflict", by_case) + self.assertIn("post-cas-finalization-gap", by_case) + self.assertIn("missing-referenced-object", by_case) + self.assertIn("maintenance-stale-plan", by_case) + self.assertIn("external-bridge-conflict", by_case) + self.assertIn("backing-migration-blocked", by_case) + self.assertIn("permission-negative", by_case) + + self.assertEqual(by_case["commit-cas-conflict"]["expected_behavior"], "conflict-without-pointer-advance") + self.assertEqual(by_case["post-cas-finalization-gap"]["expected_behavior"], "recoverable-diagnostics-and-idempotency-repair") + for entry in matrix: + self.assertIn("coverage_status", entry) + self.assertIn("evidence", entry) + self.assertNotEqual(entry["coverage_status"], "claimed-automated-live") + + def test_failure_probe_plan_generates_rest_negative_steps(self) -> None: + plan = failure_coverage.failure_probe_plan( + warehouse="lake", + namespace="sales", + table="orders", + ) + by_name = {step["name"]: step for step in plan} + + self.assertEqual( + by_name["stale-token-commit-conflict"]["path"], + "/iceberg/v1/lake/namespaces/sales/tables/orders", + ) + self.assertEqual(by_name["stale-token-commit-conflict"]["method"], "POST") + self.assertEqual(by_name["stale-token-commit-conflict"]["expected_status"], "409") + self.assertIn("expected-version-token", by_name["stale-token-commit-conflict"]["body"]) + self.assertIn("expected-metadata-location", by_name["stale-token-commit-conflict"]["body"]) + self.assertIn("new-metadata-location", by_name["stale-token-commit-conflict"]["body"]) + self.assertNotIn("base", by_name["stale-token-commit-conflict"]["body"]) + self.assertEqual( + by_name["diagnostics-after-finalization-gap"]["path"], + "/iceberg/v1/lake/namespaces/sales/tables/orders/catalog/diagnostics", + ) + self.assertEqual(by_name["diagnostics-after-finalization-gap"]["method"], "GET") + self.assertEqual(by_name["recovery-repairs-idempotency-index"]["method"], "POST") + self.assertIn("does_not_exist.metadata.json", json.dumps(by_name["missing-metadata-object-rejected"])) + self.assertNotIn("base", by_name["missing-metadata-object-rejected"]["body"]) + + def test_cli_prints_failure_matrix_and_probe_plan(self) -> None: + matrix_payload = failure_coverage.cli_json(["--print-failure-matrix"]) + matrix_document = json.loads(matrix_payload) + + self.assertIn("production_failure_coverage", matrix_document) + self.assertTrue(matrix_document["production_failure_coverage"]) + + probe_payload = failure_coverage.cli_json( + [ + "--warehouse", + "lake", + "--namespace", + "sales", + "--table", + "orders", + "--rest-path", + "/_iceberg", + "--print-failure-probes", + ] + ) + probe_document = json.loads(probe_payload) + + self.assertIn("failure_probe_plan", probe_document) + self.assertTrue(probe_document["failure_probe_plan"]) + self.assertEqual( + probe_document["failure_probe_plan"][0]["path"], + "/_iceberg/v1/lake/namespaces/sales/tables/orders", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index e92486e20..d2e24123e 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -3,6 +3,7 @@ from __future__ import annotations +import json import os import re import sys @@ -766,12 +767,39 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn("status", entry) self.assertIn("validation", entry) + def test_print_engine_compatibility_outputs_machine_readable_matrix(self) -> None: + args = self.parse_with_args(["--print-engine-compatibility"]) + + stdout = StringIO() + with redirect_stdout(stdout): + self.assertTrue(pyiceberg_smoke.printed_metadata(args)) + + document = json.loads(stdout.getvalue()) + self.assertIn("engine_compatibility", document) + clients = {entry["client"] for entry in document["engine_compatibility"]} + self.assertIn("PyIceberg", clients) + self.assertIn("Spark Iceberg REST catalog", clients) + + def test_print_production_failure_coverage_outputs_machine_readable_matrix(self) -> None: + args = self.parse_with_args(["--print-production-failure-coverage"]) + + stdout = StringIO() + with redirect_stdout(stdout): + self.assertTrue(pyiceberg_smoke.printed_metadata(args)) + + document = json.loads(stdout.getvalue()) + self.assertIn("production_failure_coverage", document) + cases = {entry["case"] for entry in document["production_failure_coverage"]} + self.assertIn("commit-cas-conflict", cases) + self.assertIn("post-cas-finalization-gap", cases) + def test_published_table_catalog_docs_do_not_use_internal_roadmap_labels(self) -> None: readme = (SCRIPT_DIR / "README.md").read_text(encoding="utf-8") self.assertIsNone(INTERNAL_ROADMAP_LABEL_RE.search(readme)) self.assertIsNone(INTERNAL_ROADMAP_LABEL_RE.search(str(pyiceberg_smoke.unsupported_inventory()))) self.assertIsNone(INTERNAL_ROADMAP_LABEL_RE.search(str(pyiceberg_smoke.production_readiness_inventory()))) + self.assertIsNone(INTERNAL_ROADMAP_LABEL_RE.search(str(pyiceberg_smoke.failure_coverage.production_failure_matrix()))) if __name__ == "__main__":