diff --git a/docs/architecture/s3-tables-support-matrix.md b/docs/architecture/s3-tables-support-matrix.md index 1251e10af..84ee1f475 100644 --- a/docs/architecture/s3-tables-support-matrix.md +++ b/docs/architecture/s3-tables-support-matrix.md @@ -30,10 +30,10 @@ catalog extension. | S3 object data plane | Supported | Data, metadata, manifest, and delete files remain ordinary S3 objects, with table-aware policy checks for table warehouse paths. | | Table bucket enablement | Supported | A regular RustFS bucket can be enabled for table catalog use and then addressed as the REST catalog warehouse. | | Catalog-vended table credentials | Automated when enabled | Disabled by default. When enabled, the credentials endpoint returns short-lived table-scoped S3 credentials. | -| AWS S3 Tables endpoint shape | Reference only | The AWS profile is recorded for comparison, not as a RustFS parity claim. | -| MinIO AIStor Tables profile | Reference only plus RustFS alias smoke | RustFS exposes the alias shape, but does not claim all AIStor private extensions. | -| Cloudflare R2 Data Catalog profile | Reference only | Kept as an interoperability reference. Live RustFS compatibility is not claimed. | -| Alibaba OSS Tables profile | Reference only | Kept as an interoperability reference. Live RustFS compatibility is not claimed. | +| AWS S3 Tables endpoint shape | Profile generator | Generates the AWS catalog URI and S3 Tables warehouse ARN shape for migration docs. Full AWS S3 Tables API parity is not claimed. | +| MinIO AIStor Tables profile | Profile generator plus RustFS alias smoke | RustFS exposes the alias shape, but does not claim all AIStor private extensions. | +| Cloudflare R2 Data Catalog profile | Profile generator | Generates the catalog URI and warehouse-name shape for migration docs. Live RustFS interoperability is not claimed. | +| Alibaba OSS Tables profile | Profile generator | Generates provider endpoint and warehouse shapes for migration docs. Live RustFS interoperability is not claimed. | ## Client And Engine Matrix @@ -157,8 +157,15 @@ python3 scripts/table-catalog/test_failure_coverage.py python3 scripts/table-catalog/pyiceberg_smoke.py --print-client-matrix python3 scripts/table-catalog/pyiceberg_smoke.py --print-engine-compatibility 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-production-readiness python3 scripts/table-catalog/engine_compatibility.py --print-spark-config +python3 scripts/table-catalog/engine_compatibility.py \ + --profile aws-s3tables \ + --region us-east-1 \ + --account-id 123456789012 \ + --table-bucket analytics \ + --print-spark-config python3 scripts/table-catalog/failure_coverage.py \ --warehouse rustfs-s3table-smoke \ --namespace smoke \ diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 4ff6c8285..caf3ea3bf 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -138,12 +138,54 @@ 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. +Vendor profiles are machine-readable connection references. They include the +catalog URI shape, warehouse shape, signing name, credential model, namespace +model, pagination model, and explicit not-claimed behavior. They are useful for +building migration docs without turning provider references into compatibility +claims. The `vendor_profiles` object lists every supported reference template; +the `selected_vendor_profile` object renders the active profile with the +provided endpoint, account, bucket, and warehouse arguments. + +Spark vendor config generation only emits S3-compatible data-plane endpoint, +path-style, and static S3 credential properties for profiles that explicitly use +the supplied endpoint as object storage, such as RustFS and MinIO AIStor. AWS S3 +Tables, Cloudflare R2 Data Catalog, and OSS Tables reference profiles leave +provider object I/O endpoint and credential selection to the provider-specific +Spark/Iceberg runtime configuration instead of inheriting RustFS local defaults. + +Generate an AWS S3 Tables-style reference profile: + +```bash +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --profile aws-s3tables \ + --region us-east-1 \ + --account-id 123456789012 \ + --table-bucket analytics \ + --print-vendor-profiles +``` + +Generate a Cloudflare R2 Data Catalog-style reference profile: + +```bash +python3 scripts/table-catalog/pyiceberg_smoke.py \ + --profile cloudflare-r2-data-catalog \ + --catalog-uri https://example.account.r2.cloudflarestorage.com/catalog \ + --warehouse-name analytics \ + --print-vendor-profiles +``` + 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 \ + --profile aws-s3tables \ + --region us-east-1 \ + --account-id 123456789012 \ + --table-bucket analytics \ + --print-spark-config python3 scripts/table-catalog/engine_compatibility.py --print-spark-sql --cleanup ``` @@ -228,15 +270,15 @@ RustFS build, client version, and expected response shape are recorded. ## Vendor Profile References -| Profile | Catalog shape | Signing name | Credential model | RustFS claim | +| Profile | Catalog shape | Warehouse shape | Signing name | RustFS claim | |---|---|---|---|---| -| `rustfs` | `{endpoint}/iceberg` | `s3` | static S3 credentials | automated smoke target | -| `rustfs-compat` | `{endpoint}/_iceberg` | `s3tables` by default | static S3 credentials | compatibility smoke target | -| `rustfs-vended-credentials` | `{endpoint}/iceberg` | `s3` | catalog-vended table credentials after table creation | automated credential smoke target when server vending is enabled | -| `aws-s3tables` | `https://s3tables.{region}.amazonaws.com/iceberg` | `s3tables` | AWS IAM/session credentials | reference only | -| `minio-aistor` | `{endpoint}/_iceberg` | `s3tables` | policy-scoped S3 credentials | reference only | -| `cloudflare-r2-data-catalog` | catalog URI returned by R2 | `s3` | catalog-vended credentials | reference only | -| `oss-tables` | provider REST endpoint | `s3` | SigV4 S3FileIO credentials | reference only | +| `rustfs` | `{endpoint}/iceberg` | `{bucket}` | `s3` | automated smoke target | +| `rustfs-compat` | `{endpoint}/_iceberg` | `{bucket}` | `s3tables` by default | compatibility smoke target | +| `rustfs-vended-credentials` | `{endpoint}/iceberg` | `{bucket}` | `s3` | automated credential smoke target when server vending is enabled | +| `aws-s3tables` | `https://s3tables.{region}.amazonaws.com/iceberg` | `arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}` | `s3tables` | profile generator only; full AWS S3 Tables API parity is not claimed | +| `minio-aistor` | `{endpoint}/_iceberg` | `{warehouse}` | `s3tables` | profile generator plus RustFS alias smoke; full AIStor extension parity is not claimed | +| `cloudflare-r2-data-catalog` | catalog URI returned by R2 | `{warehouse_name}` | `s3` | profile generator only; live RustFS interoperability is not claimed | +| `oss-tables` | provider REST endpoint | `{warehouse}` | `s3` | profile generator only; live RustFS interoperability is not claimed | ## Unsupported Inventory diff --git a/scripts/table-catalog/engine_compatibility.py b/scripts/table-catalog/engine_compatibility.py index da894dd32..186277df9 100644 --- a/scripts/table-catalog/engine_compatibility.py +++ b/scripts/table-catalog/engine_compatibility.py @@ -9,6 +9,45 @@ from collections import OrderedDict from io import StringIO from typing import Any +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 { @@ -110,6 +149,36 @@ def normalized_rest_path(rest_path: str) -> str: 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, @@ -142,6 +211,62 @@ def spark_catalog_config( ) +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") @@ -192,11 +317,16 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace: 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", default="/iceberg") - parser.add_argument("--rest-signing-name", default="s3") + parser.add_argument("--rest-path") + parser.add_argument("--rest-signing-name") 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") @@ -226,13 +356,18 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None: if args.print_spark_config: print_json( { - "spark_config": spark_catalog_config( + "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, ) diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index 5bc80ef45..aab3e2aa2 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -30,7 +30,7 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = ( TABLE_MAINTENANCE_CONFIG_VERSION = 1 IDENTIFIER_SEGMENT_MAX_LEN = 64 -PROFILE_DEFAULTS: dict[str, dict[str, str]] = { +PROFILE_DEFAULTS: dict[str, dict[str, Any]] = { "rustfs": { "catalog_uri": "{endpoint}/iceberg", "rest_path": "/iceberg", @@ -38,6 +38,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{bucket}", "credential_mode": "static-s3-credentials", "status": "automated-smoke-target", + "compatibility_stage": "automated", + "catalog_uri_shape": "{endpoint}/iceberg", + "warehouse_shape": "{bucket}", + "namespace_model": "single-level", + "pagination_model": "rustfs", + "not_claimed": [], }, "rustfs-compat": { "catalog_uri": "{endpoint}/_iceberg", @@ -46,6 +52,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{bucket}", "credential_mode": "static-s3-credentials", "status": "compatibility-smoke-target", + "compatibility_stage": "automated", + "catalog_uri_shape": "{endpoint}/_iceberg", + "warehouse_shape": "{bucket}", + "namespace_model": "single-level", + "pagination_model": "rustfs", + "not_claimed": ["full MinIO AIStor private extension parity"], }, CATALOG_VENDED_PROFILE: { "catalog_uri": "{endpoint}/iceberg", @@ -54,6 +66,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{bucket}", "credential_mode": "catalog-vended-temporary-credentials", "status": "automated-credential-smoke-target", + "compatibility_stage": "automated-when-enabled", + "catalog_uri_shape": "{endpoint}/iceberg", + "warehouse_shape": "{bucket}", + "namespace_model": "single-level", + "pagination_model": "rustfs", + "not_claimed": ["no-long-term-data-credential bootstrap"], }, "aws-s3tables": { "catalog_uri": "https://s3tables.{region}.amazonaws.com/iceberg", @@ -62,6 +80,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}", "credential_mode": "aws-iam-or-session-credentials", "status": "reference-only", + "compatibility_stage": "reference-only", + "catalog_uri_shape": "https://s3tables.{region}.amazonaws.com/iceberg", + "warehouse_shape": "arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}", + "namespace_model": "single-level", + "pagination_model": "vendor-specific", + "not_claimed": ["full AWS S3 Tables API parity", "AWS IAM/Lake Formation policy parity"], }, "minio-aistor": { "catalog_uri": "{endpoint}/_iceberg", @@ -70,6 +94,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{warehouse}", "credential_mode": "policy-scoped-s3-credentials", "status": "reference-only", + "compatibility_stage": "reference-only-plus-rustfs-alias", + "catalog_uri_shape": "{endpoint}/_iceberg", + "warehouse_shape": "{warehouse}", + "namespace_model": "single-level", + "pagination_model": "vendor-specific", + "not_claimed": ["full MinIO AIStor private extension parity"], }, "cloudflare-r2-data-catalog": { "catalog_uri": "{catalog_uri}", @@ -78,6 +108,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{warehouse_name}", "credential_mode": "catalog-vended", "status": "reference-only", + "compatibility_stage": "reference-only", + "catalog_uri_shape": "{catalog_uri}", + "warehouse_shape": "{warehouse_name}", + "namespace_model": "provider-defined", + "pagination_model": "vendor-specific", + "not_claimed": ["live RustFS interoperability", "Cloudflare R2 catalog API parity"], }, "oss-tables": { "catalog_uri": "{endpoint}/iceberg", @@ -86,6 +122,12 @@ PROFILE_DEFAULTS: dict[str, dict[str, str]] = { "warehouse": "{warehouse}", "credential_mode": "sigv4-s3fileio-credentials", "status": "reference-only", + "compatibility_stage": "reference-only", + "catalog_uri_shape": "{endpoint}/iceberg", + "warehouse_shape": "{warehouse}", + "namespace_model": "provider-defined", + "pagination_model": "vendor-specific", + "not_claimed": ["live RustFS interoperability", "Alibaba OSS Tables API parity"], }, } @@ -234,10 +276,32 @@ def env_or_none(key: str) -> str | None: return value -def vendor_profiles() -> dict[str, dict[str, str]]: +def vendor_profiles() -> dict[str, dict[str, Any]]: return {name: values.copy() for name, values in PROFILE_DEFAULTS.items()} +def selected_vendor_profile(args: argparse.Namespace) -> dict[str, Any]: + defaults = PROFILE_DEFAULTS[args.profile] + not_claimed = defaults.get("not_claimed", []) + if not isinstance(not_claimed, list): + raise ValueError("profile not_claimed field must be a list") + return { + "name": args.profile, + "status": defaults["status"], + "compatibility_stage": defaults["compatibility_stage"], + "catalog_uri": profile_catalog_uri(args), + "catalog_uri_shape": defaults["catalog_uri_shape"], + "warehouse": profile_warehouse(args), + "warehouse_shape": defaults["warehouse_shape"], + "rest_path": args.rest_path, + "rest_signing_name": args.rest_signing_name, + "credential_mode": defaults["credential_mode"], + "namespace_model": defaults["namespace_model"], + "pagination_model": defaults["pagination_model"], + "not_claimed": not_claimed.copy(), + } + + def client_matrix() -> list[dict[str, str]]: return [entry.copy() for entry in CLIENT_MATRIX] @@ -271,6 +335,43 @@ def apply_profile_defaults(args: argparse.Namespace) -> argparse.Namespace: return args +def profile_template_context(args: argparse.Namespace) -> dict[str, str]: + endpoint = normalized_endpoint(args.endpoint) + warehouse = args.warehouse or args.bucket + table_bucket = args.table_bucket or args.bucket + warehouse_name = args.warehouse_name or warehouse + catalog_uri = args.catalog_uri or f"{endpoint}{args.rest_path}" + return { + "account_id": args.account_id, + "bucket": args.bucket, + "catalog_uri": catalog_uri, + "endpoint": endpoint, + "region": args.region, + "table_bucket": table_bucket, + "warehouse": warehouse, + "warehouse_name": warehouse_name, + } + + +def formatted_profile_value(args: argparse.Namespace, key: str) -> str: + template = PROFILE_DEFAULTS[args.profile][key] + if not isinstance(template, str): + raise ValueError(f"profile field {key} is not a string") + return template.format(**profile_template_context(args)) + + +def profile_catalog_uri(args: argparse.Namespace) -> str: + if args.catalog_uri: + return args.catalog_uri.rstrip("/") + return formatted_profile_value(args, "catalog_uri").rstrip("/") + + +def profile_warehouse(args: argparse.Namespace) -> str: + if args.warehouse: + return args.warehouse + return formatted_profile_value(args, "warehouse") + + def parse_args() -> argparse.Namespace: run_id = str(int(time.time())) parser = argparse.ArgumentParser(description="Run a PyIceberg smoke test against RustFS table catalog APIs.") @@ -280,6 +381,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--secret-key", default=os.getenv("RUSTFS_SECRET_KEY", "rustfsadmin")) parser.add_argument("--region", default=os.getenv("RUSTFS_REGION", os.getenv("AWS_REGION", "us-east-1"))) parser.add_argument("--bucket", default=os.getenv("RUSTFS_TABLE_BUCKET", "rustfs-s3table-smoke")) + parser.add_argument("--warehouse", default=env_or_none("RUSTFS_TABLE_WAREHOUSE")) + parser.add_argument("--table-bucket", default=env_or_none("RUSTFS_TABLE_BUCKET_NAME")) + parser.add_argument("--account-id", default=os.getenv("RUSTFS_TABLE_ACCOUNT_ID", "000000000000")) + parser.add_argument("--warehouse-name", default=env_or_none("RUSTFS_TABLE_WAREHOUSE_NAME")) + parser.add_argument("--catalog-uri", default=env_or_none("RUSTFS_TABLE_CATALOG_URI")) parser.add_argument("--namespace", default=os.getenv("RUSTFS_TABLE_NAMESPACE", f"smoke{run_id}")) parser.add_argument("--table", default=os.getenv("RUSTFS_TABLE_NAME", f"events{run_id}")) parser.add_argument("--catalog-name", default=os.getenv("RUSTFS_TABLE_CATALOG_NAME", "rustfs")) @@ -699,11 +805,12 @@ def verify_vended_credential_data_plane_scope( def catalog_properties(args: argparse.Namespace, storage_credential: StorageCredential | None = None) -> dict[str, str]: endpoint = normalized_endpoint(args.endpoint) + warehouse = profile_warehouse(args) properties = { "type": "rest", - "uri": f"{endpoint}{args.rest_path}", - "warehouse": args.bucket, - "prefix": args.bucket, + "uri": profile_catalog_uri(args), + "warehouse": warehouse, + "prefix": warehouse, "py-io-impl": "pyiceberg.io.pyarrow.PyArrowFileIO", "s3.endpoint": endpoint, "s3.access-key-id": args.access_key, @@ -737,7 +844,12 @@ def printed_metadata(args: argparse.Namespace) -> bool: 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()}) + print_json_document( + { + "selected_vendor_profile": selected_vendor_profile(args), + "vendor_profiles": vendor_profiles(), + } + ) printed = True if args.print_unsupported_inventory: print_json_document({"unsupported_inventory": unsupported_inventory()}) diff --git a/scripts/table-catalog/test_engine_compatibility.py b/scripts/table-catalog/test_engine_compatibility.py index 496fd66b0..274e20873 100644 --- a/scripts/table-catalog/test_engine_compatibility.py +++ b/scripts/table-catalog/test_engine_compatibility.py @@ -61,6 +61,77 @@ class EngineCompatibilityTest(unittest.TestCase): 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_vendor_config_formats_aws_s3tables_profile(self) -> None: + config = engine_compatibility.spark_vendor_catalog_config( + profile="aws-s3tables", + endpoint="https://s3tables.us-east-1.amazonaws.com", + warehouse="ignored", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + account_id="123456789012", + table_bucket="analytics", + catalog_uri=None, + warehouse_name=None, + ) + + self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg") + self.assertEqual( + config["spark.sql.catalog.rustfs.warehouse"], + "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics", + ) + self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables") + self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config) + + def test_spark_vendor_config_formats_cloudflare_catalog_profile(self) -> None: + config = engine_compatibility.spark_vendor_catalog_config( + profile="cloudflare-r2-data-catalog", + endpoint="https://example.r2.cloudflarestorage.com", + warehouse="ignored", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="auto", + catalog_name="rustfs", + account_id="000000000000", + table_bucket="ignored", + catalog_uri="https://catalog.example.com/iceberg", + warehouse_name="analytics", + ) + + self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://catalog.example.com/iceberg") + self.assertEqual(config["spark.sql.catalog.rustfs.warehouse"], "analytics") + self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3") + self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config) + + def test_spark_vendor_config_keeps_endpoint_for_s3_compatible_profiles(self) -> None: + config = engine_compatibility.spark_vendor_catalog_config( + profile="minio-aistor", + endpoint="https://minio.example.com", + warehouse="analytics", + access_key="rustfsadmin", + secret_key="rustfsadmin", + region="us-east-1", + catalog_name="rustfs", + account_id="000000000000", + table_bucket="analytics", + catalog_uri=None, + warehouse_name=None, + ) + + self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://minio.example.com/_iceberg") + self.assertEqual(config["spark.sql.catalog.rustfs.s3.endpoint"], "https://minio.example.com") + self.assertEqual(config["spark.sql.catalog.rustfs.s3.path-style-access"], "true") + self.assertEqual(config["spark.sql.catalog.rustfs.s3.access-key-id"], "rustfsadmin") + self.assertEqual(config["spark.sql.catalog.rustfs.s3.secret-access-key"], "rustfsadmin") + self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables") + def test_spark_sql_smoke_covers_lifecycle_append_reload_and_cleanup(self) -> None: sql = engine_compatibility.spark_sql_smoke( catalog_name="rustfs", @@ -93,6 +164,52 @@ class EngineCompatibilityTest(unittest.TestCase): self.assertIn("engine_compatibility", document) self.assertTrue(document["engine_compatibility"]) + def test_cli_prints_vendor_spark_config(self) -> None: + payload = engine_compatibility.cli_json( + [ + "--print-spark-config", + "--profile", + "aws-s3tables", + "--region", + "us-east-1", + "--account-id", + "123456789012", + "--table-bucket", + "analytics", + ] + ) + document = json.loads(payload) + config = document["spark_config"] + + self.assertEqual(config["spark.sql.catalog.rustfs.uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg") + self.assertEqual( + config["spark.sql.catalog.rustfs.warehouse"], + "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics", + ) + self.assertEqual(config["spark.sql.catalog.rustfs.rest.signing-name"], "s3tables") + self.assertNotIn("spark.sql.catalog.rustfs.s3.endpoint", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.path-style-access", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.access-key-id", config) + self.assertNotIn("spark.sql.catalog.rustfs.s3.secret-access-key", config) + + def test_cli_spark_config_keeps_explicit_rest_overrides(self) -> None: + payload = engine_compatibility.cli_json( + [ + "--print-spark-config", + "--endpoint", + "http://127.0.0.1:9000/", + "--rest-path", + "/_iceberg", + "--rest-signing-name", + "s3tables", + ] + ) + document = json.loads(payload) + config = document["spark_config"] + + 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 assertContainsScenario(self, entry: dict[str, object], name: str, status: str) -> None: scenarios = entry.get("scenarios") self.assertIsInstance(scenarios, list) diff --git a/scripts/table-catalog/test_pyiceberg_smoke.py b/scripts/table-catalog/test_pyiceberg_smoke.py index d2e24123e..5ea7a067e 100644 --- a/scripts/table-catalog/test_pyiceberg_smoke.py +++ b/scripts/table-catalog/test_pyiceberg_smoke.py @@ -74,6 +74,56 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertEqual(profiles["minio-aistor"]["rest_signing_name"], "s3tables") self.assertEqual(profiles["cloudflare-r2-data-catalog"]["credential_mode"], "catalog-vended") + def test_vendor_profiles_publish_migration_boundaries(self) -> None: + profiles = pyiceberg_smoke.vendor_profiles() + + aws = profiles["aws-s3tables"] + self.assertEqual(aws["compatibility_stage"], "reference-only") + self.assertEqual(aws["warehouse_shape"], "arn:aws:s3tables:{region}:{account_id}:bucket/{table_bucket}") + self.assertEqual(aws["namespace_model"], "single-level") + self.assertEqual(aws["pagination_model"], "vendor-specific") + self.assertIn("full AWS S3 Tables API parity", aws["not_claimed"]) + + cloudflare = profiles["cloudflare-r2-data-catalog"] + self.assertEqual(cloudflare["catalog_uri_shape"], "{catalog_uri}") + self.assertEqual(cloudflare["credential_mode"], "catalog-vended") + self.assertIn("live RustFS interoperability", cloudflare["not_claimed"]) + + def test_aws_reference_profile_formats_s3tables_warehouse_arn(self) -> None: + args = self.parse_with_args([ + "--profile", + "aws-s3tables", + "--endpoint", + "https://s3tables.us-east-1.amazonaws.com", + "--region", + "us-east-1", + "--account-id", + "123456789012", + "--table-bucket", + "analytics", + ]) + + properties = pyiceberg_smoke.catalog_properties(args) + + self.assertEqual(properties["uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg") + self.assertEqual(properties["warehouse"], "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics") + self.assertEqual(properties["rest.signing-name"], "s3tables") + + def test_cloudflare_reference_profile_uses_catalog_uri_and_warehouse_name(self) -> None: + args = self.parse_with_args([ + "--profile", + "cloudflare-r2-data-catalog", + "--catalog-uri", + "https://catalog.example.com/iceberg", + "--warehouse-name", + "analytics", + ]) + + properties = pyiceberg_smoke.catalog_properties(args) + + self.assertEqual(properties["uri"], "https://catalog.example.com/iceberg") + self.assertEqual(properties["warehouse"], "analytics") + def test_vended_profile_requires_catalog_credentials_and_keeps_canonical_path(self) -> None: args = self.parse_with_args([ "--profile", @@ -793,6 +843,51 @@ class PyIcebergSmokeConfigTest(unittest.TestCase): self.assertIn("commit-cas-conflict", cases) self.assertIn("post-cas-finalization-gap", cases) + def test_print_vendor_profiles_outputs_migration_boundaries(self) -> None: + args = self.parse_with_args(["--print-vendor-profiles"]) + + stdout = StringIO() + with redirect_stdout(stdout): + self.assertTrue(pyiceberg_smoke.printed_metadata(args)) + + document = json.loads(stdout.getvalue()) + self.assertIn("vendor_profiles", document) + aws = document["vendor_profiles"]["aws-s3tables"] + self.assertEqual(aws["rest_signing_name"], "s3tables") + self.assertEqual(aws["compatibility_stage"], "reference-only") + self.assertIn("full AWS S3 Tables API parity", aws["not_claimed"]) + + selected = document["selected_vendor_profile"] + self.assertEqual(selected["name"], "rustfs") + self.assertEqual(selected["catalog_uri"], "http://127.0.0.1:9000/iceberg") + self.assertEqual(selected["warehouse"], "rustfs-s3table-smoke") + + def test_print_vendor_profiles_renders_selected_aws_profile(self) -> None: + args = self.parse_with_args( + [ + "--profile", + "aws-s3tables", + "--region", + "us-east-1", + "--account-id", + "123456789012", + "--table-bucket", + "analytics", + "--print-vendor-profiles", + ] + ) + + stdout = StringIO() + with redirect_stdout(stdout): + self.assertTrue(pyiceberg_smoke.printed_metadata(args)) + + document = json.loads(stdout.getvalue()) + selected = document["selected_vendor_profile"] + self.assertEqual(selected["name"], "aws-s3tables") + self.assertEqual(selected["catalog_uri"], "https://s3tables.us-east-1.amazonaws.com/iceberg") + self.assertEqual(selected["warehouse"], "arn:aws:s3tables:us-east-1:123456789012:bucket/analytics") + self.assertEqual(selected["rest_signing_name"], "s3tables") + def test_published_table_catalog_docs_do_not_use_internal_roadmap_labels(self) -> None: readme = (SCRIPT_DIR / "README.md").read_text(encoding="utf-8")