feat(table-catalog): harden table row-level commit conflicts (#3517)

* feat(table-catalog): harden snapshot conflict validation * feat: harden table row-level commit conflicts * fix: accept additional row-level snapshot shapes * test(table-catalog): expand smoke catalog probes * fix(table-catalog): satisfy row-level clippy checks --------- Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com> Co-authored-by: houseme <housemecn@gmail.com>
2026-07-26 08:18:18 +00:00 · 2026-06-18 07:32:20 +08:00
parent 24fa03e04b
commit 604379d62f
5 changed files with 1831 additions and 53 deletions
@@ -921,7 +921,7 @@ fn table_warehouse_object_prefix_from_location(table_bucket: &str, warehouse_loc
    normalize_warehouse_object_prefix(object_prefix)
 }

-fn table_warehouse_object_prefix(entry: &TableEntry) -> TableCatalogStoreResult<String> {
+pub(crate) fn table_warehouse_object_prefix(entry: &TableEntry) -> TableCatalogStoreResult<String> {
    table_warehouse_object_prefix_from_location(&entry.table_bucket, &entry.warehouse_location)
 }

@@ -4035,7 +4035,9 @@ fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResul
        .collect())
 }

-fn manifest_list_references_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<ManifestListReference>> {
+pub(crate) fn manifest_list_references_from_manifest_list_avro(
+    data: &[u8],
+) -> TableCatalogStoreResult<Vec<ManifestListReference>> {
    let reader = apache_avro::Reader::new(data)
        .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list Avro: {err}")))?;
    let mut manifest_paths = Vec::new();
@@ -4061,7 +4063,7 @@ fn file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Ve
        .collect())
 }

-fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<ManifestDataFileReference>> {
+pub(crate) fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<ManifestDataFileReference>> {
    let reader = apache_avro::Reader::new(data)
        .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest Avro: {err}")))?;
    let mut files = Vec::new();
@@ -4136,7 +4138,7 @@ fn avro_i64_value(value: &apache_avro::types::Value) -> Option<i64> {
    }
 }

-fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) -> Option<String> {
+pub(crate) fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) -> Option<String> {
    let object = if let Some(location) = location.strip_prefix("s3://") {
        let (bucket, object) = location.split_once('/')?;
        if bucket != table_bucket {
@@ -4159,7 +4161,7 @@ fn table_catalog_object_key_from_location(table_bucket: &str, location: &str) ->
    Some(object.to_string())
 }

-fn table_maintenance_object_kind(
+pub(crate) fn table_maintenance_object_kind(
    namespace: &Namespace,
    table: &IdentifierSegment,
    warehouse_object_prefix: Option<&str>,
@@ -4707,21 +4709,23 @@ struct CompactedDataFile {
    file_sequence_number: i64,
 }

-struct ManifestDataFileReference {
-    location: String,
-    object_kind: TableMetadataMaintenanceObjectKind,
-    entry_status: Option<i32>,
-    snapshot_id: Option<i64>,
-    sequence_number: Option<i64>,
-    file_sequence_number: Option<i64>,
-    record_count: Option<u64>,
-    file_size_bytes: Option<u64>,
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct ManifestDataFileReference {
+    pub location: String,
+    pub object_kind: TableMetadataMaintenanceObjectKind,
+    pub entry_status: Option<i32>,
+    pub snapshot_id: Option<i64>,
+    pub sequence_number: Option<i64>,
+    pub file_sequence_number: Option<i64>,
+    pub record_count: Option<u64>,
+    pub file_size_bytes: Option<u64>,
 }

-struct ManifestListReference {
-    manifest_path: String,
-    sequence_number: Option<i64>,
-    added_snapshot_id: Option<i64>,
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct ManifestListReference {
+    pub manifest_path: String,
+    pub sequence_number: Option<i64>,
+    pub added_snapshot_id: Option<i64>,
 }

 struct CompactionManifestListSummary<'a> {
@@ -32,6 +32,9 @@ The smoke test covers:
 - create namespace and table
 - append two rows through PyIceberg
 - reload and scan the table
+- probe direct REST catalog endpoints for metadata-location, table refs,
+  Iceberg views, maintenance config, metadata maintenance, worker run, and
+  catalog diagnostics
 - optionally drop the table and namespace

 The default profile uses the canonical RustFS catalog URI:
@@ -104,6 +107,15 @@ temporary credentials:
 - `PutObject` and `GetObject` to the same bucket outside that prefix must be
  rejected.

+The direct REST catalog probes run by default after the PyIceberg append and
+scan. For deployments that intentionally expose only the core Iceberg REST
+Catalog table path, skip those probes explicitly:
+
+```bash
+python3 scripts/table-catalog/pyiceberg_smoke.py \
+  --skip-catalog-api-probes
+```
+
 ## Machine-Readable Inventories

 The script can print the current conformance inventories without importing
@@ -120,20 +132,23 @@ work items. They are intentionally conservative: only PyIceberg is automated by
 this script today; other engines are documented until a repeatable harness is
 added.

-RustFS also exposes catalog-backed advanced Iceberg surfaces that are not part
-of the PyIceberg append smoke path yet:
+The smoke test also probes catalog-backed advanced Iceberg surfaces:

 - table refs can be listed, created or replaced, and deleted through catalog
  commits; refs with explicit retention policy require a forced delete, and
  `main` cannot be deleted
 - Iceberg views support basic create, list, load, replace, existence check, and
  drop routes with persisted view metadata and view-scoped authorization
+- metadata maintenance supports safe dry-run planning and controlled worker
+  execution checks
+- catalog diagnostics exposes the table recovery and consistency state used by
+  operators

 ## Client Matrix

 | Client | Current status | Claim |
 |---|---|---|
-| PyIceberg | Automated smoke target | create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe |
+| 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 |
 | Trino Iceberg REST catalog | Documented, not automated | no write compatibility claim yet |
 | DuckDB Iceberg | Documented, not automated | read-path reference only |
@@ -163,6 +178,7 @@ current unsupported inventory is:
 - snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports
 - automatic maintenance scheduling: external scheduler hook supported through the worker run endpoint; built-in periodic scheduling is not claimed
 - compaction rewrite: controlled run-once support for unpartitioned Parquet binpack through metadata maintenance; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed
+- row-level delete/update/merge commits: standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item
 - external catalog bridges: metadata import/register is supported, but Polaris/Glue/DLF/Hive synchronization is unsupported
 - multi-table transactions: not a short-term production claim

@@ -24,6 +24,7 @@ REQUIRED_STORAGE_CREDENTIAL_KEYS = (
    "s3.secret-access-key",
    "s3.session-token",
 )
+TABLE_MAINTENANCE_CONFIG_VERSION = 1

 PROFILE_DEFAULTS: dict[str, dict[str, str]] = {
    "rustfs": {
@@ -88,7 +89,7 @@ CLIENT_MATRIX: list[dict[str, str]] = [
    {
        "client": "PyIceberg",
        "status": "automated",
-        "coverage": "create namespace, create table, append, reload, scan, optional catalog-vended table credentials with exact-prefix data-plane scope probe",
+        "coverage": "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",
        "entrypoint": "scripts/table-catalog/pyiceberg_smoke.py",
    },
    {
@@ -156,6 +157,12 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [
        "roadmap_area": "snapshot-maintenance",
        "expected_behavior": "metadata maintenance can plan binpack candidates and commit a safe unpartitioned Parquet rewrite through the catalog; built-in periodic scheduling, sort compaction, delete-file rewrite, and row-level compaction are not claimed",
    },
+    {
+        "capability": "row-level-delete-update-merge",
+        "status": "catalog-conflict-validation-supported",
+        "roadmap_area": "row-level-conflict-detection",
+        "expected_behavior": "standard catalog commit validates append, overwrite, delete, and replace snapshot manifests for table-warehouse scope, referenced object existence, current-live-file deletes, and stale add/delete conflicts; end-to-end SQL DML client coverage remains a compatibility validation item",
+    },
    {
        "capability": "external-catalog-bridge",
        "status": "metadata-import-only",
@@ -189,6 +196,15 @@ class StorageCredential:
    config: dict[str, str]


+class RestRequestError(RuntimeError):
+    def __init__(self, method: str, path: str, status_code: int, response_body: str) -> None:
+        super().__init__(f"{method} {path} failed with HTTP {status_code}: {response_body}")
+        self.method = method
+        self.path = path
+        self.status_code = status_code
+        self.response_body = response_body
+
+
 def env_or_none(key: str) -> str | None:
    value = os.getenv(key)
    if value is None or value == "":
@@ -252,6 +268,11 @@ def parse_args() -> argparse.Namespace:
    parser.add_argument("--timeout", type=float, default=float(os.getenv("RUSTFS_TABLE_SMOKE_TIMEOUT", "20")))
    parser.add_argument("--cleanup", action="store_true", help="Drop the smoke table and namespace before exiting.")
    parser.add_argument("--replace", action="store_true", help="Drop an existing table with the same identifier first.")
+    parser.add_argument(
+        "--skip-catalog-api-probes",
+        action="store_true",
+        help="Skip direct REST probes for metadata-location, refs, views, maintenance, and diagnostics endpoints.",
+    )
    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-vendor-profiles", action="store_true", help="Print vendor connection profile references as JSON and exit.")
@@ -362,7 +383,25 @@ def signed_rest_request(args: argparse.Namespace, deps: RuntimeDeps, method: str
            return json.loads(response_data.decode("utf-8"))
    except urllib.error.HTTPError as error:
        response_body = error.read().decode("utf-8", errors="replace")
-        raise RuntimeError(f"{method} {path} failed with HTTP {error.code}: {response_body}") from error
+        raise RestRequestError(method, path, error.code, response_body) from error
+
+
+def signed_rest_request_expect_error(
+    args: argparse.Namespace,
+    deps: RuntimeDeps,
+    method: str,
+    path: str,
+    body: dict[str, Any] | None = None,
+    expected_statuses: set[int] | None = None,
+) -> RestRequestError:
+    try:
+        signed_rest_request(args, deps, method, path, body)
+    except RestRequestError as error:
+        if expected_statuses is not None and error.status_code not in expected_statuses:
+            expected = ", ".join(str(status) for status in sorted(expected_statuses))
+            raise RuntimeError(f"{method} {path} failed with HTTP {error.status_code}, expected one of: {expected}") from error
+        return error
+    raise RuntimeError(f"{method} {path} unexpectedly succeeded")


 def ensure_bucket(args: argparse.Namespace, deps: RuntimeDeps) -> None:
@@ -406,6 +445,45 @@ def credentials_endpoint_path(args: argparse.Namespace) -> str:
    return f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/tables/{encoded_table}/credentials"


+def table_endpoint_path(args: argparse.Namespace, suffix: str = "") -> str:
+    encoded_bucket = urllib.parse.quote(args.bucket, safe="")
+    encoded_namespace = urllib.parse.quote(args.namespace, safe="")
+    encoded_table = urllib.parse.quote(args.table, safe="")
+    return f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/tables/{encoded_table}{suffix}"
+
+
+def table_ref_endpoint_path(args: argparse.Namespace, ref_name: str | None = None) -> str:
+    path = table_endpoint_path(args, "/refs")
+    if ref_name is None:
+        return path
+    return f"{path}/{urllib.parse.quote(ref_name, safe='')}"
+
+
+def view_endpoint_path(args: argparse.Namespace, view_name: str | None = None) -> str:
+    encoded_bucket = urllib.parse.quote(args.bucket, safe="")
+    encoded_namespace = urllib.parse.quote(args.namespace, safe="")
+    path = f"{args.rest_path}/v1/{encoded_bucket}/namespaces/{encoded_namespace}/views"
+    if view_name is None:
+        return path
+    return f"{path}/{urllib.parse.quote(view_name, safe='')}"
+
+
+def default_maintenance_config() -> dict[str, Any]:
+    return {
+        "version": TABLE_MAINTENANCE_CONFIG_VERSION,
+        "retain-recent-metadata-files": 2,
+        "delete-enabled": False,
+        "background-enabled": False,
+        "worker-paused": True,
+        "worker-lease-timeout-seconds": 60,
+        "max-retry-attempts": 0,
+        "retry-initial-backoff-seconds": 5,
+        "retry-max-backoff-seconds": 300,
+        "quarantine-enabled": False,
+        "quarantine-retention-seconds": 0,
+    }
+
+
 def storage_credential_from_response(response: dict[str, Any]) -> StorageCredential:
    credentials = response.get("storage-credentials")
    if not isinstance(credentials, list) or not credentials:
@@ -692,29 +770,155 @@ def cleanup_catalog(catalog: Any, identifier: tuple[str, str]) -> None:
            print(f"warning: failed to drop namespace {identifier[0]}: {error}", file=sys.stderr)


+def current_snapshot_id_from_table_response(response: dict[str, Any]) -> int:
+    metadata = response.get("metadata")
+    if not isinstance(metadata, dict):
+        raise RuntimeError("REST loadTable response did not include metadata")
+    snapshot_id = metadata.get("current-snapshot-id")
+    if not isinstance(snapshot_id, int):
+        raise RuntimeError("REST loadTable response did not include a current snapshot id")
+    return snapshot_id
+
+
+def run_metadata_location_probe(args: argparse.Namespace, deps: RuntimeDeps, table_response: dict[str, Any]) -> None:
+    metadata_location = table_response.get("metadata-location")
+    if not isinstance(metadata_location, str) or not metadata_location:
+        raise RuntimeError("REST loadTable response did not include metadata-location")
+    location_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/metadata-location"))
+    if location_response.get("metadata-location") != metadata_location:
+        raise RuntimeError("metadata-location endpoint does not match loadTable metadata-location")
+    if not location_response.get("version-token"):
+        raise RuntimeError("metadata-location endpoint did not include version-token")
+
+
+def run_table_ref_probe(args: argparse.Namespace, deps: RuntimeDeps, snapshot_id: int) -> None:
+    ref_name = f"smoke-{safe_probe_segment(args.namespace, args.table)}"
+    ref_body = {
+        "snapshot-id": snapshot_id,
+        "type": "tag",
+        "max-ref-age-ms": 86_400_000,
+    }
+    signed_rest_request(args, deps, "PUT", table_ref_endpoint_path(args, ref_name), ref_body)
+    refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args))
+    if refs.get("refs", {}).get(ref_name, {}).get("snapshot-id") != snapshot_id:
+        raise RuntimeError("table refs endpoint did not return the smoke tag")
+    signed_rest_request_expect_error(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), expected_statuses={400})
+    signed_rest_request(args, deps, "DELETE", table_ref_endpoint_path(args, ref_name), {"force": True})
+    refs = signed_rest_request(args, deps, "GET", table_ref_endpoint_path(args))
+    if ref_name in refs.get("refs", {}):
+        raise RuntimeError("table refs endpoint still returned the deleted smoke tag")
+
+
+def smoke_view_request(args: argparse.Namespace, view_name: str, version_id: int, sql: str) -> dict[str, Any]:
+    return {
+        "name": view_name,
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [
+                {"id": 1, "name": "id", "required": True, "type": "long"},
+                {"id": 2, "name": "payload", "required": False, "type": "string"},
+            ],
+        },
+        "view-version": {
+            "version-id": version_id,
+            "schema-id": 0,
+            "timestamp-ms": int(time.time() * 1000),
+            "summary": {"operation": "replace"},
+            "representations": [
+                {
+                    "type": "sql",
+                    "sql": sql,
+                    "dialect": "spark",
+                }
+            ],
+        },
+        "properties": {"rustfs.smoke": "true", "rustfs.smoke.table": args.table},
+    }
+
+
+def run_view_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
+    view_name = f"{args.table}_smoke_view"
+    create_body = smoke_view_request(args, view_name, 1, f"SELECT id, payload FROM {args.namespace}.{args.table}")
+    signed_rest_request(args, deps, "POST", view_endpoint_path(args), create_body)
+    try:
+        views = signed_rest_request(args, deps, "GET", view_endpoint_path(args))
+        identifiers = views.get("identifiers", [])
+        if not any(identifier.get("name") == view_name for identifier in identifiers if isinstance(identifier, dict)):
+            raise RuntimeError("listViews response did not include the smoke view")
+        loaded = signed_rest_request(args, deps, "GET", view_endpoint_path(args, view_name))
+        metadata_location = loaded.get("metadata-location")
+        if not isinstance(metadata_location, str) or not metadata_location:
+            raise RuntimeError("loadView response did not include metadata-location")
+    finally:
+        try:
+            signed_rest_request(args, deps, "DELETE", view_endpoint_path(args, view_name))
+        except RestRequestError as error:
+            if error.status_code != 404:
+                raise
+
+
+def run_maintenance_probe(args: argparse.Namespace, deps: RuntimeDeps) -> None:
+    config_path = table_endpoint_path(args, "/maintenance/config")
+    signed_rest_request(args, deps, "PUT", config_path, default_maintenance_config())
+    config = signed_rest_request(args, deps, "GET", config_path)
+    if config.get("version") != TABLE_MAINTENANCE_CONFIG_VERSION:
+        raise RuntimeError("maintenance config endpoint did not persist the smoke config")
+
+    report = signed_rest_request(
+        args,
+        deps,
+        "POST",
+        table_endpoint_path(args, "/maintenance/metadata"),
+        {"retain-recent-metadata-files": 1, "delete": False},
+    )
+    job = report.get("job")
+    if not isinstance(job, dict) or not job.get("job-id"):
+        raise RuntimeError("maintenance metadata endpoint did not return a job id")
+    signed_rest_request(args, deps, "GET", table_endpoint_path(args, f"/maintenance/jobs/{job['job-id']}"))
+
+    worker = signed_rest_request(args, deps, "POST", table_endpoint_path(args, "/maintenance/worker/run"), {})
+    worker_job = worker.get("job")
+    expected_statuses = {"DISABLED", "PAUSED", "FAILED", "SUCCESSFUL", "RUNNING"}
+    if not isinstance(worker_job, dict) or worker_job.get("status") not in expected_statuses:
+        raise RuntimeError("maintenance worker endpoint did not return a stable job status")
+
+
+def run_catalog_api_probes(args: argparse.Namespace, deps: RuntimeDeps) -> None:
+    table_response = signed_rest_request(args, deps, "GET", table_endpoint_path(args))
+    snapshot_id = current_snapshot_id_from_table_response(table_response)
+    run_metadata_location_probe(args, deps, table_response)
+    run_table_ref_probe(args, deps, snapshot_id)
+    run_view_probe(args, deps)
+    run_maintenance_probe(args, deps)
+    diagnostics = signed_rest_request(args, deps, "GET", table_endpoint_path(args, "/catalog/diagnostics"))
+    if not isinstance(diagnostics, dict) or not diagnostics:
+        raise RuntimeError("catalog diagnostics endpoint returned an empty response")
+
+
 def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:
    endpoint = normalized_endpoint(args.endpoint)
    ensure_local_proxy_bypass(endpoint)
    ensure_aws_env(args.access_key, args.secret_key, args.region)

-    print(f"[1/9] ensuring S3 bucket {args.bucket}")
+    print(f"[1/10] ensuring S3 bucket {args.bucket}")
    ensure_bucket(args, deps)

-    print(f"[2/9] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1")
+    print(f"[2/10] enabling RustFS table bucket {args.bucket} through {args.rest_path}/v1")
    enable_table_bucket(args, deps)

-    print(f"[3/9] loading PyIceberg REST catalog at {endpoint}{args.rest_path}")
+    print(f"[3/10] loading PyIceberg REST catalog at {endpoint}{args.rest_path}")
    catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args))
    install_rustfs_rest_sigv4_adapter(catalog, args, deps)
    identifier = table_identifier(args)

    if args.replace:
-        print(f"[4/9] replacing existing table {'.'.join(identifier)} if present")
+        print(f"[4/10] replacing existing table {'.'.join(identifier)} if present")
        drop_table_if_present(catalog, identifier)
    else:
-        print(f"[4/9] table {'.'.join(identifier)} is available")
+        print(f"[4/10] table {'.'.join(identifier)} is available")

-    print(f"[5/9] creating namespace and table {'.'.join(identifier)}")
+    print(f"[5/10] creating namespace and table {'.'.join(identifier)}")
    ensure_namespace(catalog, args.namespace)
    arrow_schema = deps.pyarrow.schema(
        [
@@ -726,10 +930,10 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:

    storage_credential = None
    if args.require_vended_credentials:
-        print(f"[6/9] loading table-scoped storage credentials")
+        print(f"[6/10] loading table-scoped storage credentials")
        storage_credential = load_table_storage_credential(args, deps)
        print(f"credential scope: {storage_credential.prefix or 'not reported'}")
-        print(f"[7/9] verifying table-scoped data-plane access")
+        print(f"[7/10] verifying table-scoped data-plane access")
        try:
            expected_table_location = table_warehouse_location(created_table)
        except RuntimeError:
@@ -738,10 +942,10 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:
        catalog = deps.load_catalog(args.catalog_name, **catalog_properties(args, storage_credential=storage_credential))
        install_rustfs_rest_sigv4_adapter(catalog, args, deps)
    else:
-        print(f"[6/9] using configured S3 credentials for data-plane operations")
-        print(f"[7/9] skipping vended credential data-plane scope probe")
+        print(f"[6/10] using configured S3 credentials for data-plane operations")
+        print(f"[7/10] skipping vended credential data-plane scope probe")

-    print(f"[8/9] appending rows through PyIceberg")
+    print(f"[8/10] appending rows through PyIceberg")
    table = catalog.load_table(identifier)
    rows = deps.pyarrow.Table.from_pylist(
        [
@@ -752,12 +956,18 @@ def run_smoke(args: argparse.Namespace, deps: RuntimeDeps) -> None:
    )
    table.append(rows)

-    print(f"[9/9] reloading and scanning table")
+    print(f"[9/10] reloading and scanning table")
    loaded = catalog.load_table(identifier)
    scanned = loaded.scan().to_arrow()
    if scanned.num_rows != 2:
        raise RuntimeError(f"expected 2 rows after append, got {scanned.num_rows}")

+    if args.skip_catalog_api_probes:
+        print("[10/10] skipping direct REST catalog API probes")
+    else:
+        print("[10/10] probing direct REST catalog APIs")
+        run_catalog_api_probes(args, deps)
+
    if args.cleanup:
        print(f"cleanup: dropping table and namespace {'.'.join(identifier)}")
        cleanup_catalog(catalog, identifier)
@@ -100,6 +100,165 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
            "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/credentials",
        )

+    def test_table_catalog_endpoint_paths_encode_identifier_components(self) -> None:
+        args = self.parse_with_args([
+            "--bucket",
+            "lake bucket",
+            "--namespace",
+            "sales.analytics",
+            "--table",
+            "orders table",
+        ])
+
+        self.assertEqual(
+            pyiceberg_smoke.table_endpoint_path(args),
+            "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table",
+        )
+        self.assertEqual(
+            pyiceberg_smoke.table_endpoint_path(args, "/metadata-location"),
+            "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/metadata-location",
+        )
+        self.assertEqual(
+            pyiceberg_smoke.table_ref_endpoint_path(args, "release/2026"),
+            "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/tables/orders%20table/refs/release%2F2026",
+        )
+        self.assertEqual(
+            pyiceberg_smoke.view_endpoint_path(args, "orders view"),
+            "/iceberg/v1/lake%20bucket/namespaces/sales.analytics/views/orders%20view",
+        )
+
+    def test_default_maintenance_config_is_safe_for_smoke_runs(self) -> None:
+        config = pyiceberg_smoke.default_maintenance_config()
+
+        self.assertEqual(config["version"], 1)
+        self.assertFalse(config["delete-enabled"])
+        self.assertFalse(config["background-enabled"])
+        self.assertTrue(config["worker-paused"])
+        self.assertEqual(config["max-retry-attempts"], 0)
+
+    def test_expected_error_helper_returns_matching_rest_error(self) -> None:
+        args = self.parse_with_args([])
+        expected = pyiceberg_smoke.RestRequestError("DELETE", "/path", 400, "bad request")
+
+        with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=expected):
+            returned = pyiceberg_smoke.signed_rest_request_expect_error(
+                args,
+                mock.Mock(),
+                "DELETE",
+                "/path",
+                expected_statuses={400},
+            )
+
+        self.assertIs(returned, expected)
+
+    def test_expected_error_helper_rejects_wrong_status_or_success(self) -> None:
+        args = self.parse_with_args([])
+        wrong_status = pyiceberg_smoke.RestRequestError("DELETE", "/path", 404, "missing")
+
+        with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=wrong_status):
+            with self.assertRaisesRegex(RuntimeError, "expected one of"):
+                pyiceberg_smoke.signed_rest_request_expect_error(
+                    args,
+                    mock.Mock(),
+                    "DELETE",
+                    "/path",
+                    expected_statuses={400},
+                )
+
+        with mock.patch.object(pyiceberg_smoke, "signed_rest_request", return_value={}):
+            with self.assertRaisesRegex(RuntimeError, "unexpectedly succeeded"):
+                pyiceberg_smoke.signed_rest_request_expect_error(args, mock.Mock(), "DELETE", "/path")
+
+    def test_current_snapshot_id_is_read_from_rest_load_table_response(self) -> None:
+        self.assertEqual(
+            pyiceberg_smoke.current_snapshot_id_from_table_response({"metadata": {"current-snapshot-id": 42}}),
+            42,
+        )
+
+        with self.assertRaisesRegex(RuntimeError, "metadata"):
+            pyiceberg_smoke.current_snapshot_id_from_table_response({})
+        with self.assertRaisesRegex(RuntimeError, "current snapshot id"):
+            pyiceberg_smoke.current_snapshot_id_from_table_response({"metadata": {}})
+
+    def test_catalog_api_probe_exercises_extended_rest_surfaces(self) -> None:
+        args = self.parse_with_args(["--namespace", "sales", "--table", "orders"])
+        deps = mock.Mock()
+        calls: list[tuple[str, str, object]] = []
+        table_path = pyiceberg_smoke.table_endpoint_path(args)
+        metadata_location_path = pyiceberg_smoke.table_endpoint_path(args, "/metadata-location")
+        refs_path = pyiceberg_smoke.table_ref_endpoint_path(args)
+        view_path = pyiceberg_smoke.view_endpoint_path(args)
+        config_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/config")
+        maintenance_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/metadata")
+        worker_path = pyiceberg_smoke.table_endpoint_path(args, "/maintenance/worker/run")
+        diagnostics_path = pyiceberg_smoke.table_endpoint_path(args, "/catalog/diagnostics")
+
+        def fake_signed_request(
+            _args: object,
+            _deps: object,
+            method: str,
+            path: str,
+            body: object = None,
+        ) -> dict[str, object]:
+            calls.append((method, path, body))
+            if (method, path) == ("GET", table_path):
+                return {"metadata-location": "s3://lake/tables/id/metadata/v2.metadata.json", "metadata": {"current-snapshot-id": 7}}
+            if (method, path) == ("GET", metadata_location_path):
+                return {"metadata-location": "s3://lake/tables/id/metadata/v2.metadata.json", "version-token": "token-2"}
+            if method == "PUT" and path.startswith(f"{refs_path}/"):
+                return {}
+            if (method, path) == ("GET", refs_path):
+                if sum(1 for call in calls if call[:2] == ("GET", refs_path)) == 1:
+                    return {"refs": {"smoke-sales-orders": {"snapshot-id": 7}}}
+                return {"refs": {}}
+            if method == "DELETE" and path.startswith(f"{refs_path}/"):
+                return {}
+            if (method, path) == ("POST", view_path):
+                return {}
+            if (method, path) == ("GET", view_path):
+                return {"identifiers": [{"name": "orders_smoke_view"}]}
+            if (method, path) == ("GET", f"{view_path}/orders_smoke_view"):
+                return {"metadata-location": "s3://lake/views/orders_smoke_view/metadata/v1.json"}
+            if (method, path) == ("DELETE", f"{view_path}/orders_smoke_view"):
+                return {}
+            if (method, path) == ("PUT", config_path):
+                return {}
+            if (method, path) == ("GET", config_path):
+                return {"version": 1}
+            if (method, path) == ("POST", maintenance_path):
+                return {"job": {"job-id": "job-1"}}
+            if (method, path) == ("GET", pyiceberg_smoke.table_endpoint_path(args, "/maintenance/jobs/job-1")):
+                return {"job": {"job-id": "job-1", "status": "SUCCESSFUL"}}
+            if (method, path) == ("POST", worker_path):
+                return {"job": {"status": "PAUSED"}}
+            if (method, path) == ("GET", diagnostics_path):
+                return {"status": "ok"}
+            raise AssertionError(f"unexpected REST request: {method} {path}")
+
+        with mock.patch.object(pyiceberg_smoke, "signed_rest_request", side_effect=fake_signed_request):
+            with mock.patch.object(
+                pyiceberg_smoke,
+                "signed_rest_request_expect_error",
+                return_value=pyiceberg_smoke.RestRequestError("DELETE", f"{refs_path}/smoke-sales-orders", 400, "retained"),
+            ) as expect_error:
+                pyiceberg_smoke.run_catalog_api_probes(args, deps)
+
+        expect_error.assert_called_once()
+        self.assertIn(("GET", metadata_location_path, None), calls)
+        self.assertIn(("GET", diagnostics_path, None), calls)
+        self.assertIn(("POST", worker_path, {}), calls)
+
+    def test_smoke_view_request_uses_stable_iceberg_view_shape(self) -> None:
+        args = self.parse_with_args(["--namespace", "sales", "--table", "orders"])
+
+        request = pyiceberg_smoke.smoke_view_request(args, "orders_view", 7, "SELECT id FROM sales.orders")
+
+        self.assertEqual(request["name"], "orders_view")
+        self.assertEqual(request["schema"]["type"], "struct")
+        self.assertEqual(request["view-version"]["version-id"], 7)
+        self.assertEqual(request["view-version"]["representations"][0]["dialect"], "spark")
+        self.assertEqual(request["properties"]["rustfs.smoke.table"], "orders")
+
    def test_catalog_properties_can_use_vended_storage_credentials(self) -> None:
        args = self.parse_with_args([
            "--access-key",
@@ -313,7 +472,7 @@ class PyIcebergSmokeConfigTest(unittest.TestCase):
        capabilities = {entry["capability"] for entry in inventory}

        self.assertIn("credential-vending", capabilities)
-        self.assertIn("iceberg-views", capabilities)
+        self.assertIn("row-level-delete-update-merge", capabilities)
        self.assertIn("background-maintenance-worker", capabilities)
        for entry in inventory:
            self.assertIn("status", entry)