diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 60d2b9a5d..267699926 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -14,6 +14,7 @@ catalog extension. | Label | Meaning | |---|---| | Automated | Covered by a runnable RustFS script or server test. | +| Manual/live harness | RustFS can generate pinned client package inputs, commands, expected outputs, and CI opt-in gates for a live endpoint, but the live run is not enabled by default in CI. | | Generated harness | RustFS can generate client configuration or probe input, but live execution is not automated in CI. | | Supported | Implemented server-side and covered by focused RustFS tests. | | Preview / controlled | Implemented behind explicit operator action or a run-once endpoint. No automatic background claim is made. | @@ -40,7 +41,7 @@ catalog extension. | Client or engine | Status | Current RustFS claim | |---|---|---| | PyIceberg | Automated | Creates namespace and table, appends rows, reloads, scans, probes metadata-location, refs, views, maintenance, diagnostics, and optional catalog-vended table credentials with an exact-prefix data-plane scope check. | -| Spark Iceberg REST catalog | Generated harness | RustFS can generate Spark REST catalog properties and SQL for namespace creation, table creation, append, refresh, count, and cleanup. Live Spark execution and commit-conflict probing are manual validation items. | +| Spark Iceberg REST catalog | Manual/live harness | RustFS can generate pinned Spark/Iceberg package inputs, REST catalog properties, SQL, run commands, expected `row_count=2`, and a CI opt-in gate for namespace creation, table creation, append, refresh, count, and cleanup. Live Spark execution and commit-conflict probing are still manual validation items unless explicitly enabled in the runner. | | Trino Iceberg REST catalog | Documented, not automated | Read-path configuration reference only. Write compatibility is not claimed. | | DuckDB Iceberg | Documented, not automated | Read-path reference only. Write and commit compatibility are not claimed. | | StarRocks Iceberg REST catalog | Documented, not automated | External catalog read-path reference only. Write compatibility is not claimed. | @@ -187,6 +188,7 @@ python3 scripts/table-catalog/engine_compatibility.py \ --account-id 123456789012 \ --table-bucket analytics \ --print-spark-config +python3 scripts/table-catalog/engine_compatibility.py --print-live-conformance --cleanup python3 scripts/table-catalog/failure_coverage.py \ --warehouse rustfs-s3table-smoke \ --namespace smoke \ @@ -202,8 +204,8 @@ Acceptable wording: > RustFS includes a core Iceberg REST Catalog-based S3 Tables implementation > with PyIceberg smoke coverage, table-aware S3 data-plane policy checks, -> controlled maintenance, catalog recovery diagnostics, and generated Spark and -> production-failure probe harnesses. +> controlled maintenance, catalog recovery diagnostics, Spark manual/live +> conformance input, and production-failure probe harnesses. Do not claim: diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 9400da176..e63ee85c5 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -135,8 +135,10 @@ 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. Spark now has generated configuration and SQL smoke input; -other engines are documented until a repeatable harness is added. +this script today. Spark has a repeatable manual/live harness with pinned +client package inputs, generated configuration, generated SQL, expected +results, and a CI opt-in gate; other engines are documented until a repeatable +harness is added. Vendor profiles are machine-readable connection references. They include the catalog URI shape, warehouse shape, signing name, credential model, namespace @@ -187,6 +189,7 @@ python3 scripts/table-catalog/engine_compatibility.py \ --table-bucket analytics \ --print-spark-config python3 scripts/table-catalog/engine_compatibility.py --print-spark-sql --cleanup +python3 scripts/table-catalog/engine_compatibility.py --print-live-conformance --cleanup ``` The production failure helper records the negative coverage required before @@ -233,7 +236,7 @@ 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 | Generated smoke harness | configuration and SQL can be generated for create namespace, create table, append, reload, count, and cleanup against a running RustFS endpoint | +| Spark Iceberg REST catalog | Manual/live harness | pinned Spark and Iceberg package inputs, configuration, SQL, run command, expected row count, and cleanup can be generated for a running RustFS endpoint; CI execution is opt-in | | Trino Iceberg REST catalog | 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 | @@ -327,11 +330,37 @@ RUSTFS_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS=900 The TTL is clamped to the supported short-lived range by the server. -## Spark Manual Baseline +## Spark Manual/Live Harness Spark validation should use the same RustFS endpoint and warehouse bucket as the -PyIceberg smoke test. The exact Spark and Iceberg package versions should be -recorded in the client matrix after each run. +PyIceberg smoke test. The harness records default pinned client package inputs, +the exact command shape, and the expected row count. It is manual by default and +should only run in CI when explicitly gated with +`RUSTFS_TABLE_CATALOG_LIVE_CONFORMANCE=1`. + +Generate the full live harness document: + +```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 \ + --pyiceberg-version 0.10.0 \ + --spark-version 3.5.4 \ + --iceberg-version 1.7.1 \ + --print-live-conformance \ + --cleanup +``` + +The output includes: + +- PyIceberg install and smoke commands +- Spark package coordinates for `iceberg-spark-runtime` and `iceberg-aws-bundle` +- Spark REST catalog properties +- generated Spark SQL +- a `spark-sql` command using the generated properties +- expected `row_count=2` before optional cleanup Generate the configuration properties: @@ -373,6 +402,7 @@ python3 scripts/table-catalog/engine_compatibility.py \ ``` 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. +count, and optional cleanup. Until Spark execution is enabled in CI through the +explicit live-conformance gate, do not claim Spark support beyond a manually +verified run with the exact RustFS build, Spark version, Iceberg version, and +expected output recorded. diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index 186277df9..46ba0ae96 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/scripts/table-catalog/engine_compatibility.py @@ -5,10 +5,17 @@ from __future__ import annotations import argparse import json +import re +import shlex 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" + VENDOR_SPARK_PROFILES: dict[str, dict[str, str]] = { "rustfs": { "catalog_uri": "{endpoint}/iceberg", @@ -75,14 +82,14 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]: }, { "client": "Spark Iceberg REST catalog", - "status": "generated-smoke-harness", - "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-spark-sql", + "status": "manual-live-harness", + "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance", "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("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"), ], }, @@ -310,6 +317,163 @@ def spark_sql_smoke( return "\n".join(statements) + "\n" +def shell_join(parts: list[str]) -> str: + return " ".join(shlex.quote(part) for part in parts) + + +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 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, + cleanup: bool = False, +) -> OrderedDict[str, Any]: + endpoint = normalized_endpoint(endpoint) + rest_path = normalized_rest_path(rest_path) + 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) + 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), + ] + ), + ), + ( + "clients", + [ + OrderedDict( + [ + ("name", "PyIceberg"), + ("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"), + ("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"), + ] + ), + ], + ), + ] + ) + + 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") @@ -327,8 +491,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: 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("--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-spark-config", action="store_true") parser.add_argument("--print-spark-sql", action="store_true") return parser.parse_args(argv) @@ -375,6 +544,30 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None: 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, + cleanup=args.cleanup, + ) + }, + output, + ) + printed = True if args.print_spark_sql: sql = spark_sql_smoke( catalog_name=args.catalog_name, diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 274e20873..faec78d72 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -29,12 +29,12 @@ class EngineCompatibilityTest(unittest.TestCase): 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.assertEqual(spark["status"], "manual-live-harness") + self.assertContainsScenario(spark, "create-namespace", "manual-live-harness") + self.assertContainsScenario(spark, "create-table", "manual-live-harness") + self.assertContainsScenario(spark, "append", "manual-live-harness") + self.assertContainsScenario(spark, "reload-table", "manual-live-harness") + self.assertContainsScenario(spark, "drop-table", "manual-live-harness") self.assertContainsScenario(spark, "commit-conflict", "manual-validation-required") trino = by_client["Trino Iceberg REST catalog"] @@ -210,6 +210,86 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "http://127.0.0.1:9000/_iceberg") self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables") + def test_live_conformance_harness_pins_clients_and_records_commands(self) -> None: + harness = engine_compatibility.live_conformance_harness( + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + namespace="smoke", + table="events", + rest_path="/iceberg", + rest_signing_name="s3", + pyiceberg_version="0.10.0", + spark_version="3.5.4", + iceberg_version="1.7.1", + scala_version="2.12", + cleanup=True, + ) + + self.assertEqual(harness["mode"], "manual-or-ci-optional") + self.assertEqual(harness["ci_gate"], "RUSTFS_TABLE_CATALOG_LIVE_CONFORMANCE=1") + self.assertEqual(harness["expected_results"]["row_count"], 2) + self.assertIn("RustFS endpoint is reachable", harness["prerequisites"]) + + by_client = {client["name"]: client for client in harness["clients"]} + pyiceberg = by_client["PyIceberg"] + self.assertEqual(pyiceberg["version"], "0.10.0") + self.assertIn("pyiceberg[pyarrow]==0.10.0", pyiceberg["install"]) + self.assertIn("scripts/table-catalog/pyiceberg_smoke.py", pyiceberg["command"]) + self.assertIn("--replace", pyiceberg["command"]) + self.assertIn("--cleanup", pyiceberg["command"]) + + spark = by_client["Spark Iceberg REST catalog"] + self.assertEqual(spark["spark_version"], "3.5.4") + self.assertEqual(spark["iceberg_version"], "1.7.1") + self.assertIn("iceberg-spark-runtime-3.5_2.12:1.7.1", spark["packages"]) + self.assertIn("iceberg-aws-bundle:1.7.1", spark["packages"]) + self.assertIn("spark-sql", spark["command"]) + self.assertIn("spark.sql.catalog.rustfs.uri=http://127.0.0.1:9000/iceberg", spark["command"]) + self.assertIn("SELECT COUNT(*) AS row_count", spark["sql"]) + + def test_cli_prints_live_conformance_harness(self) -> None: + payload = engine_compatibility.cli_json( + [ + "--print-live-conformance", + "--pyiceberg-version", + "0.10.0", + "--spark-version", + "3.5.4", + "--iceberg-version", + "1.7.1", + "--cleanup", + ] + ) + document = json.loads(payload) + + self.assertEqual(document["live_conformance"]["mode"], "manual-or-ci-optional") + self.assertEqual(document["live_conformance"]["expected_results"]["row_count"], 2) + + def test_live_conformance_harness_sanitizes_sql_file_path(self) -> None: + harness = engine_compatibility.live_conformance_harness( + endpoint="http://127.0.0.1:9000", + warehouse="rustfs-s3table-smoke", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + namespace="sales/prod", + table="orders 2026", + rest_path="/iceberg", + rest_signing_name="s3", + pyiceberg_version="0.10.0", + spark_version="3.5.4", + iceberg_version="1.7.1", + scala_version="2.12", + ) + + spark = {client["name"]: client for client in harness["clients"]}["Spark Iceberg REST catalog"] + self.assertEqual(spark["sql_file"], "/tmp/rustfs-s3tables-sales-prod-orders-2026-spark.sql") + def assertContainsScenario(self, entry: dict[str, object], name: str, status: str) -> None: scenarios = entry.get("scenarios") self.assertIsInstance(scenarios, list)