test(table-catalog): extend live engine conformance probes (#4184)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
2026-08-09 06:39:25 +00:00 · 2026-07-02 16:34:37 +08:00
parent b21b27dddd
commit 14c0fca576
5 changed files with 345 additions and 41 deletions
@@ -15,6 +15,12 @@ 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": {
@@ -95,21 +101,21 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
        },
        {
            "client": "Trino Iceberg REST catalog",
-            "status": "documented-read-path",
-            "entrypoint": "scripts/table-catalog/README.md",
+            "status": "manual-live-read-probe",
+            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "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("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": "documented-read-path",
-            "entrypoint": "scripts/table-catalog/README.md",
+            "status": "manual-live-read-probe",
+            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
-                scenario("catalog-load", "manual-validation-required", "REST catalog extension/configuration reference"),
-                scenario("read-table", "manual-validation-required", "read-path verification only"),
+                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"),
            ],
        },
@@ -125,18 +131,19 @@ def engine_compatibility_matrix() -> list[dict[str, Any]]:
        },
        {
            "client": "Snowflake Open Catalog / Iceberg integrations",
-            "status": "reference-only",
-            "entrypoint": "scripts/table-catalog/README.md",
+            "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": "s3-data-plane-reference",
-            "entrypoint": "scripts/table-catalog/README.md",
+            "status": "manual-live-s3-stage-probe",
+            "entrypoint": "scripts/table-catalog/engine_compatibility.py --print-live-conformance",
            "scenarios": [
-                scenario("s3-data-plane-read", "manual-validation-required", "S3 stage/data-plane reference only"),
+                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"),
            ],
        },
@@ -280,6 +287,12 @@ def quote_spark_identifier(identifier: str) -> str:
    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(
        [
@@ -321,6 +334,10 @@ 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"
@@ -344,6 +361,133 @@ def spark_sql_command(*, packages: str, config: OrderedDict[str, str], 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 live_conformance_harness(
    *,
    endpoint: str,
@@ -360,10 +504,18 @@ def live_conformance_harness(
    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,
@@ -377,6 +529,30 @@ def live_conformance_harness(
    )
    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",
@@ -430,6 +606,7 @@ def live_conformance_harness(
                    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),
@@ -439,6 +616,7 @@ def live_conformance_harness(
                    OrderedDict(
                        [
                            ("name", "Spark Iceberg REST catalog"),
+                            ("status", "manual-live-harness"),
                            ("spark_version", spark_version),
                            ("iceberg_version", iceberg_version),
                            ("scala_version", scala_version),
@@ -468,6 +646,54 @@ def live_conformance_harness(
                            ("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"),
+                        ]
+                    ),
                ],
            ),
        ]
@@ -495,6 +721,13 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    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")
@@ -562,6 +795,13 @@ def run(args: argparse.Namespace, output: StringIO | None = None) -> None:
                    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,
                )
            },