feat(table-catalog): finalize Iceberg REST behavior (#6072)

* feat(table-catalog): finalize Iceberg REST behavior

* fix(table-catalog): address REST finalization regressions

* test(table-catalog): expect REST commit conflicts

* test(table-catalog): avoid serialized view test deadlocks

* fix(table-catalog): adapt shared test backend

* fix(table-catalog): enforce Iceberg metadata invariants

* fix(table-catalog): preserve manifest length in test

* test(table-catalog): use valid metadata fixtures

* test(table-catalog): seed manifests before manifest lists

* fix(table-catalog): restore validation gates

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>

Cargo.lock

@@ -278,6 +278,7 @@ checksum = "312c1ea69e5fe9966e0029fb95aca8790100b85aff4f0d3b00a9337c74069a9c"
 dependencies = [
  "bigdecimal",
  "bon",
+ "crc32fast",
  "digest 0.11.3",
  "log",
  "miniz_oxide 0.9.1",
@@ -289,9 +290,11 @@ dependencies = [
  "serde",
  "serde_bytes",
  "serde_json",
+ "snap",
  "strum",
  "thiserror 2.0.20",
  "uuid",
+ "zstd",
 ]
 
 [[package]]
@@ -9200,6 +9203,7 @@ dependencies = [
  "serial_test",
  "sha2 0.11.0",
  "shadow-rs",
+ "snap",
  "socket2",
  "subtle",
  "sysinfo",

Cargo.toml

@@ -171,7 +171,7 @@ tower = { version = "0.5.3" }
 tower-http = { version = "0.7.0" }
 
 # Serialization and Data Formats
-apache-avro = "0.22.0"
+apache-avro = { version = "0.22.0", features = ["snappy", "zstandard"] }
 bytes = { version = "1.12.1" }
 bytesize = "2.7.0"
 byteorder = "1.5.0"

docs/architecture/s3-tables-support-matrix.md

@@ -61,17 +61,17 @@ catalog extension.
 
 | Area | Status | Covered behavior |
 |---|---|---|
-| Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and route capabilities. |
+| Catalog config | Supported | `GET /v1/config` advertises RustFS catalog defaults and only the supported OpenAPI REST paths in `endpoints`. RustFS administration, maintenance, migration, diagnostics, refs, and metadata-location extensions remain available but are not presented as standard Iceberg REST endpoints. |
 | Table bucket discovery | Supported | `PUT` and `GET /v1/buckets/{warehouse}` enable and inspect table bucket state. |
 | Namespaces | Supported | Create, list, load, existence check, and drop namespace routes are registered on both catalog prefixes. List responses support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Namespace identifiers are limited to 512 ASCII characters so persisted paths and stateless continuation tokens remain bounded. |
-| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. |
-| Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. Standard commits preserve the normal commit-token file name and use an immutable-table-scoped fallback when rename followed by source-name reuse would otherwise collide at the same generation and commit ID. |
+| Tables | Supported | Create, register, list, load, existence check, commit, metadata-location get/update, and drop table routes are registered on both catalog prefixes. Table and view listings support Iceberg REST `pageSize`/`pageToken` pagination with context-bound tokens and bounded catalog-store reads. Commit identifiers must match the URL resource; unknown requirements, updates, and snapshot operations fail as bad requests; staged create, register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. Standard statistics, partition statistics, and schema/spec cleanup updates are accepted. |
+| Commit CAS | Supported | Single-table commits validate base metadata, expected version token, referenced object existence, warehouse scope, and Iceberg commit requirements before advancing the current metadata pointer. Externally supplied metadata transitions preserve monotonic column, partition, and sequence assignment watermarks and immutable definitions for retained schemas, partition specs, sort orders, and snapshots. Standard commits preserve the normal commit-token file name and use an immutable-table-scoped fallback when rename followed by source-name reuse would otherwise collide at the same generation and commit ID. The catalog does not advertise `idempotency-key-lifetime`; clients must treat standard mutation-wide `Idempotency-Key` semantics as unsupported. |
 | Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose staged/finalization gaps and repair safe idempotency gaps without moving the table pointer. |
 | Snapshot refs | Supported | Refs can be listed, created or replaced, and deleted through catalog commits. `main` is protected and refs with explicit retention require forced delete. |
-| Iceberg views | Supported | Basic create, list, load, replace, existence check, and drop routes persist view metadata with view-scoped authorization. |
-| Table credentials endpoint | Supported | Returns an empty `storage-credentials` list by default. Returns table-scoped temporary credentials only when credential vending is enabled. |
+| Iceberg views | Supported | Basic create, list, load, replace, existence check, and drop routes persist view metadata with view-scoped authorization. Replace identifiers must match the URL resource, `schema-id: -1` resolves to the last added schema, one commit timestamp is used consistently, and only Iceberg view format version 1 is accepted. |
+| Table credentials endpoint | Supported | Returns an empty `storage-credentials` list by default. Returns table-scoped temporary credentials only when credential vending is enabled. Credential responses set `Cache-Control: no-store, private`, `Pragma: no-cache`, and `Expires: 0`. |
 | Catalog diagnostics and export | Supported | Exposes recovery state, consistency state, backing manifest, recoverable commit-log WAL state, strong backing migration target, single-active-writer policy, and scale validation matrix. |
-| Catalog import and rollback | Supported | Import/register and rollback use catalog validation and commit paths rather than direct pointer mutation. |
+| Catalog import and rollback | Supported | Import/register and online rollback use catalog validation and commit paths rather than direct pointer mutation. Online rollback accepts only a forward-safe metadata target that preserves assignment watermarks and retained definitions. Restoring an older target that lowers those watermarks is an offline disaster-recovery operation and requires every writer to be stopped. |
 | External catalog bridge | Supported operator path | Operator-supplied metadata pointer sync/import is supported for external catalog identity boundaries. Online vendor SDK polling and policy mirroring are not claimed. |
 | Multi-table transactions | Not claimed | RustFS currently claims single-table commit atomicity only. |
 

rustfs/Cargo.toml

@@ -278,6 +278,8 @@ rustfs-signer.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
 serde_urlencoded = { workspace = true }
+snap.workspace = true
+zstd.workspace = true
 
 # Cryptography and Security
 rustls = { workspace = true, default-features = false, features = ["aws-lc-rs", "logging", "tls12", "prefer-post-quantum", "std"] }
@@ -355,7 +357,6 @@ rcgen = { workspace = true }
 rustfs-test-utils.workspace = true
 # diagnose_e2e fixtures (archives are generated in-test, never checked in)
 zip = { workspace = true }
-zstd = { workspace = true }
 # Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
 # the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
 rustfs-ecstore = { workspace = true, features = ["test-util"] }

rustfs/src/admin/handlers/table_catalog/credentials.rs

@@ -29,6 +29,6 @@ impl Operation for RestLoadCredentialsHandler {
         let issuer = IamTableCredentialIssuer::from_request(&req)?;
         let response =
             load_credentials_response(&store, &warehouse, &namespace, &table, &issuer, Some(&principal.credentials)).await?;
-        build_json_response(StatusCode::OK, &response)
+        build_sensitive_json_response(StatusCode::OK, &response)
     }
 }

rustfs/src/admin/handlers/table_catalog/table.rs

@@ -125,7 +125,9 @@ impl Operation for RestLoadTableHandler {
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
+        let snapshot_selection = rest_table_snapshot_selection_from_query(&req.uri)?;
+        let mut response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?;
+        apply_rest_table_snapshot_selection(&mut response.metadata, snapshot_selection);
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -158,7 +160,7 @@ impl Operation for RestCommitTableHandler {
         let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
         install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
-        let request = read_json_body::<RestCommitTableRequest>(std::mem::take(&mut req.input)).await?;
+        let request = read_rest_commit_table_request(std::mem::take(&mut req.input)).await?;
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
@@ -178,6 +180,14 @@ impl Operation for RestDropTableHandler {
         let table = table_name_from_params(&params)?;
         let resource = TableCatalogResource::table(&warehouse, &namespace, &table);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableAction).await?;
+        let purge_requested = rest_purge_requested_from_query(&req.uri)?;
+        if purge_requested {
+            return Err(iceberg_rest_error(
+                ICEBERG_ERROR_UNSUPPORTED_OPERATION,
+                StatusCode::NOT_ACCEPTABLE,
+                "purgeRequested=true is not supported",
+            ));
+        }
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
         let store = table_catalog_store_from_extensions(&req.extensions)?;
         drop_table_in_store(&store, &warehouse, &namespace, &table).await?;

rustfs/src/admin/handlers/table_catalog/view.rs

@@ -43,8 +43,9 @@ impl Operation for RestCreateViewHandler {
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
         let table_bucket_enabled = table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
+        let publication_backend = TableCommitObjectBackend::preauthorized(metadata_backend);
         let response =
-            create_view_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
+            create_view_response(&store, &publication_backend, &warehouse, &namespace, request, table_bucket_enabled).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }
@@ -87,17 +88,20 @@ pub struct RestReplaceViewHandler {}
 
 #[async_trait::async_trait]
 impl Operation for RestReplaceViewHandler {
-    async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+    async fn call(&self, mut req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
         let namespace = namespace_from_params(&params)?;
         let view = view_name_from_params(&params)?;
         let resource = TableCatalogResource::view(&warehouse, &namespace, &view);
-        authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        let principal = authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?;
+        install_table_catalog_s3_request_info(&mut req, &principal)?;
         ensure_table_bucket_enabled_from_extensions(&req.extensions, &warehouse).await?;
-        let request = read_json_body::<RestCommitViewRequest>(req.input).await?;
+        let request = read_rest_commit_view_request(std::mem::take(&mut req.input)).await?;
         let metadata_backend = table_catalog_backend_from_extensions(&req.extensions)?;
         let store = table_catalog_store_from_backend(metadata_backend.clone())?;
-        let response = replace_view_response(&store, &metadata_backend, &warehouse, &namespace, &view, request).await?;
+        let commit_backend = TableCommitObjectBackend::for_request(metadata_backend, req);
+        let result = replace_view_response(&store, &commit_backend, &warehouse, &namespace, &view, request).await;
+        let response = commit_backend.finish(result).await?;
         build_json_response(StatusCode::OK, &response)
     }
 }

rustfs/src/storage/access.rs

@@ -1478,7 +1478,7 @@ async fn retain_table_data_plane_publication_guard<T>(
         .map_err(|err| s3_error!(InternalError, "failed to acquire table publication guard: {}", err))?;
     let mut state = retained.state.lock();
     state.keys.insert(key);
-    state.guards.push(guard);
+    state.guards.push(Box::new(guard));
     drop(state);
     req.extensions.insert(retained);
     Ok(())

rustfs/src/table_catalog/iceberg/manifest.rs

@@ -16,6 +16,8 @@ use std::io::Read;
 
 use super::super::*;
 
+const AVRO_ZSTANDARD_MAX_WINDOW_LOG: u32 = 27;
+
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct ManifestDataFileReference {
     pub location: String,
@@ -66,6 +68,7 @@ pub(crate) struct DecodedManifestList {
 pub(crate) struct DecodedManifest {
     pub references: Vec<ManifestDataFileReference>,
     pub decoded_size: usize,
+    pub partition_spec_id: Option<i32>,
 }
 
 pub(crate) fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<String>> {
@@ -92,6 +95,25 @@ pub(crate) fn decode_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<
         .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list Avro: {err}")))?;
     let format_version =
         avro_record_format_version(reader.writer_schema(), &["sequence_number", "min_sequence_number"], "manifest list")?;
+    if format_version == 2 {
+        let apache_avro::Schema::Record(record) = reader.writer_schema() else {
+            return Err(TableCatalogStoreError::Invalid("manifest list Avro schema must be a record".to_string()));
+        };
+        for field in [
+            "added_files_count",
+            "existing_files_count",
+            "deleted_files_count",
+            "added_rows_count",
+            "existing_rows_count",
+            "deleted_rows_count",
+        ] {
+            if !record.lookup.contains_key(field) {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "Iceberg v2 manifest list Avro schema is missing {field}"
+                )));
+            }
+        }
+    }
     let mut manifest_paths = Vec::new();
     for value in reader {
         if manifest_paths.len() >= TABLE_MANIFEST_AVRO_MAX_RECORDS {
@@ -115,24 +137,12 @@ pub(crate) fn decode_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<
             sequence_number: avro_record_field(&value, "sequence_number").and_then(avro_i64_value),
             min_sequence_number: avro_record_field(&value, "min_sequence_number").and_then(avro_i64_value),
             added_snapshot_id: avro_record_field(&value, "added_snapshot_id").and_then(avro_i64_value),
-            added_files_count: avro_record_field(&value, "added_files_count")
-                .and_then(avro_i32_value)
-                .and_then(|value| u64::try_from(value).ok()),
-            existing_files_count: avro_record_field(&value, "existing_files_count")
-                .and_then(avro_i32_value)
-                .and_then(|value| u64::try_from(value).ok()),
-            deleted_files_count: avro_record_field(&value, "deleted_files_count")
-                .and_then(avro_i32_value)
-                .and_then(|value| u64::try_from(value).ok()),
-            added_rows_count: avro_record_field(&value, "added_rows_count")
-                .and_then(avro_i64_value)
-                .and_then(|value| u64::try_from(value).ok()),
-            existing_rows_count: avro_record_field(&value, "existing_rows_count")
-                .and_then(avro_i64_value)
-                .and_then(|value| u64::try_from(value).ok()),
-            deleted_rows_count: avro_record_field(&value, "deleted_rows_count")
-                .and_then(avro_i64_value)
-                .and_then(|value| u64::try_from(value).ok()),
+            added_files_count: avro_nullable_non_negative_i32(&value, "added_files_count", "manifest list")?,
+            existing_files_count: avro_nullable_non_negative_i32(&value, "existing_files_count", "manifest list")?,
+            deleted_files_count: avro_nullable_non_negative_i32(&value, "deleted_files_count", "manifest list")?,
+            added_rows_count: avro_nullable_non_negative_i64(&value, "added_rows_count", "manifest list")?,
+            existing_rows_count: avro_nullable_non_negative_i64(&value, "existing_rows_count", "manifest list")?,
+            deleted_rows_count: avro_nullable_non_negative_i64(&value, "deleted_rows_count", "manifest list")?,
         });
     }
     Ok(DecodedManifestList {
@@ -171,6 +181,17 @@ pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Decod
         .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest Avro: {err}")))?;
     let format_version =
         avro_record_format_version(reader.writer_schema(), &["sequence_number", "file_sequence_number"], "manifest")?;
+    let partition_spec_id = reader
+        .user_metadata()
+        .get("partition-spec-id")
+        .map(|value| {
+            std::str::from_utf8(value)
+                .ok()
+                .and_then(|value| value.parse::<i32>().ok())
+                .filter(|value| *value >= 0)
+                .ok_or_else(|| TableCatalogStoreError::Invalid("manifest partition-spec-id metadata is invalid".to_string()))
+        })
+        .transpose()?;
     let mut files = Vec::new();
     for value in reader {
         if files.len() >= TABLE_MANIFEST_AVRO_MAX_RECORDS {
@@ -202,6 +223,9 @@ pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Decod
                 )));
             }
         };
+        let partition = avro_record_field(data_file, "partition")
+            .and_then(avro_record_value_fields)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("manifest data file partition must be a record".to_string()))?;
         files.push(ManifestDataFileReference {
             location: file_path.to_string(),
             format_version,
@@ -218,15 +242,14 @@ pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Decod
             file_size_bytes: avro_record_field(data_file, "file_size_in_bytes")
                 .and_then(avro_i64_value)
                 .and_then(|value| u64::try_from(value).ok()),
-            partition: avro_record_field(data_file, "partition")
-                .and_then(avro_record_value_fields)
-                .unwrap_or_default(),
+            partition,
             sort_order_id: avro_record_field(data_file, "sort_order_id").and_then(avro_i32_value),
         });
     }
     Ok(DecodedManifest {
         references: files,
         decoded_size,
+        partition_spec_id,
     })
 }
 
@@ -240,6 +263,8 @@ pub(crate) async fn decode_manifest_avro_async(data: Vec<u8>) -> TableCatalogSto
 enum AvroContainerCodec {
     Null,
     Deflate,
+    Snappy,
+    Zstandard,
 }
 
 fn validate_avro_container(data: &[u8]) -> TableCatalogStoreResult<usize> {
@@ -300,6 +325,8 @@ fn validate_avro_container(data: &[u8]) -> TableCatalogStoreResult<usize> {
     let codec = match codec.unwrap_or(b"null") {
         b"null" => AvroContainerCodec::Null,
         b"deflate" => AvroContainerCodec::Deflate,
+        b"snappy" => AvroContainerCodec::Snappy,
+        b"zstandard" => AvroContainerCodec::Zstandard,
         codec => {
             return Err(TableCatalogStoreError::Unsupported(format!(
                 "Avro codec {} is not supported for table commit validation",
@@ -369,6 +396,34 @@ fn avro_block_decoded_size(codec: AvroContainerCodec, block: &[u8], remaining_si
             usize::try_from(decoded_size)
                 .map_err(|_| TableCatalogStoreError::Invalid("Avro decoded data size is invalid".to_string()))
         }
+        AvroContainerCodec::Snappy => {
+            let data_end = block
+                .len()
+                .checked_sub(4)
+                .ok_or_else(|| TableCatalogStoreError::Invalid("Avro snappy block is missing its checksum".to_string()))?;
+            let decoded_size = snap::raw::decompress_len(&block[..data_end])
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to inspect Avro snappy block: {err}")))?;
+            if decoded_size > remaining_size {
+                return Err(TableCatalogStoreError::Invalid("Avro decoded data exceeds the commit limit".to_string()));
+            }
+            Ok(decoded_size)
+        }
+        AvroContainerCodec::Zstandard => {
+            let limit = remaining_size
+                .checked_add(1)
+                .ok_or_else(|| TableCatalogStoreError::Invalid("Avro decoded data size limit overflowed".to_string()))?;
+            let limit = u64::try_from(limit)
+                .map_err(|_| TableCatalogStoreError::Invalid("Avro decoded data size limit is invalid".to_string()))?;
+            let mut decoder = zstd::stream::read::Decoder::new(block)
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to decompress Avro zstandard block: {err}")))?;
+            decoder
+                .window_log_max(AVRO_ZSTANDARD_MAX_WINDOW_LOG)
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to bound Avro zstandard window: {err}")))?;
+            let decoded_size = std::io::copy(&mut decoder.take(limit), &mut std::io::sink())
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to decompress Avro zstandard block: {err}")))?;
+            usize::try_from(decoded_size)
+                .map_err(|_| TableCatalogStoreError::Invalid("Avro decoded data size is invalid".to_string()))
+        }
     }
 }
 
@@ -445,6 +500,40 @@ fn avro_record_value_fields(value: &apache_avro::types::Value) -> Option<Vec<(St
     )
 }
 
+fn avro_nullable_non_negative_i32(
+    value: &apache_avro::types::Value,
+    field: &str,
+    label: &str,
+) -> TableCatalogStoreResult<Option<u64>> {
+    let Some(value) = avro_record_field(value, field) else {
+        return Ok(None);
+    };
+    match avro_non_union_value(value) {
+        apache_avro::types::Value::Null => Ok(None),
+        apache_avro::types::Value::Int(value) => u64::try_from(*value)
+            .map(Some)
+            .map_err(|_| TableCatalogStoreError::Invalid(format!("{label} field {field} must be a non-negative int"))),
+        _ => Err(TableCatalogStoreError::Invalid(format!("{label} field {field} must be a nullable int"))),
+    }
+}
+
+fn avro_nullable_non_negative_i64(
+    value: &apache_avro::types::Value,
+    field: &str,
+    label: &str,
+) -> TableCatalogStoreResult<Option<u64>> {
+    let Some(value) = avro_record_field(value, field) else {
+        return Ok(None);
+    };
+    match avro_non_union_value(value) {
+        apache_avro::types::Value::Null => Ok(None),
+        apache_avro::types::Value::Long(value) => u64::try_from(*value)
+            .map(Some)
+            .map_err(|_| TableCatalogStoreError::Invalid(format!("{label} field {field} must be a non-negative long"))),
+        _ => Err(TableCatalogStoreError::Invalid(format!("{label} field {field} must be a nullable long"))),
+    }
+}
+
 pub(crate) fn avro_non_union_value(value: &apache_avro::types::Value) -> &apache_avro::types::Value {
     match value {
         apache_avro::types::Value::Union(_, inner) => avro_non_union_value(inner),
@@ -472,3 +561,127 @@ fn avro_i64_value(value: &apache_avro::types::Value) -> Option<i64> {
         _ => None,
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rejects_v2_manifest_lists_without_required_count_fields() {
+        let schema = apache_avro::Schema::parse_str(
+            r#"{
+                "type": "record",
+                "name": "manifest_file",
+                "fields": [
+                    {"name": "manifest_path", "type": "string"},
+                    {"name": "manifest_length", "type": "long"},
+                    {"name": "partition_spec_id", "type": "int"},
+                    {"name": "content", "type": "int"},
+                    {"name": "sequence_number", "type": "long"},
+                    {"name": "min_sequence_number", "type": "long"},
+                    {"name": "added_snapshot_id", "type": "long"}
+                ]
+            }"#,
+        )
+        .expect("incomplete manifest-list schema should parse");
+        let data = apache_avro::Writer::new(&schema, Vec::new())
+            .expect("manifest-list writer should initialize")
+            .into_inner()
+            .expect("manifest-list bytes should flush");
+
+        let error = match decode_manifest_list_avro(&data) {
+            Ok(_) => panic!("v2 count fields must be declared in the writer schema"),
+            Err(error) => error,
+        };
+        assert_eq!(
+            error,
+            TableCatalogStoreError::Invalid("Iceberg v2 manifest list Avro schema is missing added_files_count".to_string())
+        );
+    }
+
+    #[test]
+    fn rejects_negative_nullable_manifest_list_counts() {
+        let value = apache_avro::types::Value::Record(vec![
+            ("added_files_count".to_string(), apache_avro::types::Value::Int(-1)),
+            ("added_rows_count".to_string(), apache_avro::types::Value::Long(-1)),
+        ]);
+
+        assert_eq!(
+            avro_nullable_non_negative_i32(&value, "added_files_count", "manifest list")
+                .expect_err("negative file counts must be rejected"),
+            TableCatalogStoreError::Invalid("manifest list field added_files_count must be a non-negative int".to_string())
+        );
+        assert_eq!(
+            avro_nullable_non_negative_i64(&value, "added_rows_count", "manifest list")
+                .expect_err("negative row counts must be rejected"),
+            TableCatalogStoreError::Invalid("manifest list field added_rows_count must be a non-negative long".to_string())
+        );
+    }
+
+    #[test]
+    fn rejects_manifest_partition_with_non_record_schema() {
+        let schema = apache_avro::Schema::parse_str(
+            r#"{
+                "type": "record",
+                "name": "manifest_entry",
+                "fields": [
+                    {"name": "status", "type": "int"},
+                    {"name": "snapshot_id", "type": "long"},
+                    {
+                        "name": "data_file",
+                        "type": {
+                            "type": "record",
+                            "name": "data_file",
+                            "fields": [
+                                {"name": "file_path", "type": "string"},
+                                {"name": "record_count", "type": "long"},
+                                {"name": "file_size_in_bytes", "type": "long"},
+                                {"name": "partition", "type": "string"}
+                            ]
+                        }
+                    }
+                ]
+            }"#,
+        )
+        .expect("manifest schema should parse");
+        let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                ("status".to_string(), apache_avro::types::Value::Int(1)),
+                ("snapshot_id".to_string(), apache_avro::types::Value::Long(1)),
+                (
+                    "data_file".to_string(),
+                    apache_avro::types::Value::Record(vec![
+                        (
+                            "file_path".to_string(),
+                            apache_avro::types::Value::String("s3://warehouse/tables/table-id/data/file.parquet".to_string()),
+                        ),
+                        ("record_count".to_string(), apache_avro::types::Value::Long(1)),
+                        ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
+                        ("partition".to_string(), apache_avro::types::Value::String("not-a-record".to_string())),
+                    ]),
+                ),
+            ]))
+            .expect("manifest record should append");
+        let data = writer.into_inner().expect("manifest bytes should flush");
+
+        let error = match decode_manifest_avro(&data) {
+            Ok(_) => panic!("manifest partitions must preserve their record shape"),
+            Err(error) => error,
+        };
+        assert_eq!(
+            error,
+            TableCatalogStoreError::Invalid("manifest data file partition must be a record".to_string())
+        );
+    }
+
+    #[test]
+    fn rejects_oversized_zstandard_windows() {
+        // Non-single-segment frame with a 2^28-byte window and one empty final block.
+        let compressed = [0x28, 0xb5, 0x2f, 0xfd, 0x00, 0x90, 0x01, 0x00, 0x00];
+
+        let error = avro_block_decoded_size(AvroContainerCodec::Zstandard, &compressed, TABLE_MANIFEST_AVRO_MAX_DECODED_SIZE)
+            .expect_err("zstandard windows larger than the manifest decode budget must be rejected");
+        assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+    }
+}

rustfs/src/table_catalog/mod.rs

@@ -111,8 +111,12 @@ const TABLE_MANIFEST_AVRO_MAX_DECODED_SIZE: usize = 128 * 1024 * 1024;
 const TABLE_MANIFEST_AVRO_MAX_RECORDS: usize = 1_000_000;
 const TABLE_MANIFEST_AVRO_MAX_HEADER_ENTRIES: usize = 1_024;
 const TABLE_COMMIT_MAX_MANIFESTS: usize = 10_000;
+const TABLE_COMMIT_MAX_MANIFEST_TRAVERSALS: usize = 20_000;
 const TABLE_COMMIT_MAX_AVRO_BYTES: usize = 512 * 1024 * 1024;
 const TABLE_COMMIT_MAX_FILE_REFERENCES: usize = 1_000_000;
+const TABLE_COMMIT_MAX_STATISTICS_OBJECTS: usize = 1_024;
+const TABLE_COMMIT_MAX_STATISTICS_BYTES: usize = 512 * 1024 * 1024;
+const TABLE_STATISTICS_FILE_MAX_SIZE: usize = 128 * 1024 * 1024;
 pub(crate) const TABLE_COMMIT_OBJECT_VALIDATION_CONCURRENCY: usize = 16;
 pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;
 const WAREHOUSE_ROOT: &str = "warehouses";

rustfs/src/table_catalog/store/migration.rs

@@ -216,7 +216,7 @@ where
         Ok(fence)
     }
 
-    pub(super) async fn acquire_table_bucket_registry_write_permit(&self) -> TableCatalogStoreResult<Box<dyn Send>> {
+    pub(super) async fn acquire_table_bucket_registry_write_permit(&self) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let fence_path = self.paths.backing_migration_global_fence_path();
         let lock_path = self.paths.backing_migration_global_fence_lock_path();
         let guard = self.backend.acquire_read_lock(self.catalog_bucket(), &lock_path).await?;
@@ -235,7 +235,7 @@ where
     pub(super) async fn acquire_object_backed_catalog_write_permit(
         &self,
         table_bucket: &str,
-    ) -> TableCatalogStoreResult<Box<dyn Send>> {
+    ) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let lock_path = self.paths.backing_migration_fence_lock_path(table_bucket);
         let guard = self.backend.acquire_read_lock(self.catalog_bucket(), &lock_path).await?;
         if self.read_backing_migration_fence(table_bucket).await?.is_some() {
@@ -290,7 +290,7 @@ where
     async fn collect_bucket_snapshot_with_locks(
         &self,
         table_bucket: &str,
-        guards: &mut Vec<Box<dyn Send>>,
+        guards: &mut Vec<TableCatalogLockGuard>,
     ) -> TableCatalogStoreResult<StrongTableCatalogBucketSnapshot> {
         let bucket_path = self.paths.table_bucket_entry_path(table_bucket);
         guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_path).await?);

rustfs/src/table_catalog/store/mod.rs

@@ -262,6 +262,29 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
     async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()>;
 
+    async fn create_view_with_publication(
+        &self,
+        entry: ViewEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
+        publication.begin_table_bucket(&entry.table_bucket).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation requires a table-bucket publication fence".to_string(),
+            ));
+        }
+        publication
+            .prepare(&entry.table_bucket, &entry.namespace, &entry.view)
+            .await?;
+        if !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.view) {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation requires a view publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        self.create_view(entry).await
+    }
+
     async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<ViewEntry>>;
 
     async fn list_views_page(
@@ -283,6 +306,32 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
     async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult<ViewCommitResult>;
 
+    async fn replace_view_with_publication(
+        &self,
+        request: ViewCommitRequest,
+        table_bucket_fence_required: bool,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<ViewCommitResult> {
+        if table_bucket_fence_required {
+            publication.begin_table_bucket(&request.table_bucket).await?;
+            if !publication.holds_table_bucket(&request.table_bucket) {
+                return Err(TableCatalogStoreError::Internal(
+                    "view replacement requires a table-bucket publication fence".to_string(),
+                ));
+            }
+        }
+        publication
+            .prepare(&request.table_bucket, &request.namespace, &request.view)
+            .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.view) {
+            return Err(TableCatalogStoreError::Internal(
+                "view replacement requires a view publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        self.replace_view(request).await
+    }
+
     async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()>;
 
     async fn get_commit_by_id(
@@ -338,7 +387,7 @@ struct TableCommitLockPublication<'a, B> {
 struct TableCommitLockPublicationState {
     table_bucket: Option<String>,
     table: Option<(String, String, String)>,
-    guards: Vec<Box<dyn Send>>,
+    guards: Vec<TableCatalogLockGuard>,
 }
 
 impl<'a, B> TableCommitLockPublication<'a, B> {
@@ -409,15 +458,17 @@ where
     }
 
     fn holds_table_bucket(&self, table_bucket: &str) -> bool {
-        self.state.lock().table_bucket.as_deref() == Some(table_bucket)
+        let state = self.state.lock();
+        state.table_bucket.as_deref() == Some(table_bucket) && state.guards.iter().all(|guard| !guard.is_lock_lost())
     }
 
     fn holds_table(&self, table_bucket: &str, namespace: &str, table: &str) -> bool {
-        self.state
-            .lock()
+        let state = self.state.lock();
+        state
             .table
             .as_ref()
             .is_some_and(|held| held.0 == table_bucket && held.1 == namespace && held.2 == table)
+            && state.guards.iter().all(|guard| !guard.is_lock_lost())
     }
 
     fn complete(&self) {
@@ -438,6 +489,32 @@ pub(crate) struct TableCatalogObjectMetadata {
     pub mod_time: Option<OffsetDateTime>,
 }
 
+pub(crate) struct TableCatalogLockGuard {
+    _guard: Box<dyn Send>,
+    lock_lost: Option<Arc<rustfs_lock::distributed_lock::LockLostSignal>>,
+}
+
+impl TableCatalogLockGuard {
+    pub(crate) fn stable(guard: impl Send + 'static) -> Self {
+        Self {
+            _guard: Box::new(guard),
+            lock_lost: None,
+        }
+    }
+
+    fn namespace(guard: rustfs_lock::NamespaceLockGuard) -> Self {
+        let lock_lost = guard.lock_lost_signal();
+        Self {
+            _guard: Box::new(guard),
+            lock_lost,
+        }
+    }
+
+    pub(crate) fn is_lock_lost(&self) -> bool {
+        self.lock_lost.as_ref().is_some_and(|signal| signal.is_lost())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct TableCatalogObjectListPage {
     pub objects: Vec<String>,
@@ -588,11 +665,11 @@ pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
         Ok(TableCatalogObjectListPage { objects, is_truncated })
     }
 
-    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         self.acquire_write_lock(bucket, object).await
     }
 
-    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>>;
+    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard>;
 
     async fn begin_table_bucket_commit_publication(&self, _table_bucket: &str) -> TableCatalogStoreResult<()> {
         Ok(())
@@ -1169,6 +1246,17 @@ where
         }
     }
 
+    async fn create_view_with_publication(
+        &self,
+        entry: ViewEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
+        match self {
+            Self::ObjectBacked(store) => store.create_view_with_publication(entry, publication).await,
+            Self::DurableStrong(store) => store.create_view_with_publication(entry, publication).await,
+        }
+    }
+
     async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<ViewEntry>> {
         match self {
             Self::ObjectBacked(store) => store.list_views(table_bucket, namespace).await,
@@ -1203,6 +1291,26 @@ where
         }
     }
 
+    async fn replace_view_with_publication(
+        &self,
+        request: ViewCommitRequest,
+        table_bucket_fence_required: bool,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<ViewCommitResult> {
+        match self {
+            Self::ObjectBacked(store) => {
+                store
+                    .replace_view_with_publication(request, table_bucket_fence_required, publication)
+                    .await
+            }
+            Self::DurableStrong(store) => {
+                store
+                    .replace_view_with_publication(request, table_bucket_fence_required, publication)
+                    .await
+            }
+        }
+    }
+
     async fn drop_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult<()> {
         match self {
             Self::ObjectBacked(store) => store.drop_view(table_bucket, namespace, view).await,
@@ -1686,7 +1794,7 @@ where
         })
     }
 
-    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let lock = self
             .store
             .new_ns_lock(bucket, object)
@@ -1696,10 +1804,10 @@ where
             .get_write_lock(get_lock_acquire_timeout())
             .await
             .map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog table lock: {err}")))?;
-        Ok(Box::new(guard))
+        Ok(TableCatalogLockGuard::namespace(guard))
     }
 
-    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let lock = self
             .store
             .new_ns_lock(bucket, object)
@@ -1709,7 +1817,7 @@ where
             .get_read_lock(get_lock_acquire_timeout())
             .await
             .map_err(|err| TableCatalogStoreError::Internal(format!("failed to acquire catalog migration lock: {err}")))?;
-        Ok(Box::new(guard))
+        Ok(TableCatalogLockGuard::namespace(guard))
     }
 }
 

rustfs/src/table_catalog/store/object.rs

@@ -1036,6 +1036,20 @@ where
             .await
     }
 
+    async fn restore_table_warehouse_index_after_failed_drop(&self, entry: &TableEntry, reason: &'static str) {
+        if let Err(err) = self.reserve_table_warehouse_index(entry).await {
+            tracing::warn!(
+                table_bucket = %entry.table_bucket,
+                namespace = %entry.namespace,
+                table = %entry.table,
+                table_id = %entry.table_id,
+                reason,
+                error = %err,
+                "failed to restore table warehouse index after table drop stopped"
+            );
+        }
+    }
+
     async fn delete_table_warehouse_index_if_changed(&self, current: &TableEntry, next: &TableEntry) {
         let Ok(current_index) = table_warehouse_index_entry(current) else {
             return;
@@ -1316,6 +1330,15 @@ where
         }
         self.ensure_table_warehouse_prefix_available(&entry).await?;
         let reservation = self.reserve_table_warehouse_index(&entry).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket)
+            || !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.table)
+        {
+            self.delete_created_table_warehouse_index(&entry, reservation, "table publication fence lost")
+                .await;
+            return Err(TableCatalogStoreError::Internal(
+                "table registration publication fence was lost before catalog update".to_string(),
+            ));
+        }
         let result = self
             .write_entry_unlocked(self.catalog_bucket(), &table_path, &entry, precondition)
             .await;
@@ -1327,7 +1350,25 @@ where
     }
 
     async fn write_view_entry(&self, entry: ViewEntry, precondition: TableCatalogPutPrecondition) -> TableCatalogStoreResult<()> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        self.write_view_entry_with_publication(entry, precondition, &publication)
+            .await
+    }
+
+    async fn write_view_entry_with_publication(
+        &self,
+        entry: ViewEntry,
+        precondition: TableCatalogPutPrecondition,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
         validate_view_entry_version_and_id(&entry)?;
+        publication.begin_table_bucket(&entry.table_bucket).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation requires a table-bucket publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
         self.require_table_bucket(&entry.table_bucket).await?;
         let namespace = parse_namespace_for_store(&entry.namespace)?;
         let view = parse_table_for_store(&entry.view)?;
@@ -1353,6 +1394,17 @@ where
                 entry.table_bucket, entry.namespace, entry.view
             )));
         }
+        // Preserve catalog -> publication -> object lock order across rolling upgrades.
+        publication
+            .prepare(&entry.table_bucket, &entry.namespace, &entry.view)
+            .await?;
+        if !publication.holds_table_bucket(&entry.table_bucket)
+            || !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.view)
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation publication fence was lost before catalog update".to_string(),
+            ));
+        }
         self.write_entry_unlocked(self.catalog_bucket(), &view_path, &entry, precondition)
             .await
     }
@@ -4493,16 +4545,16 @@ where
         validate_commit_metadata_digest(&request, &new_metadata_object)?;
         let table_bucket = request.table_bucket.clone();
         let metadata_location = request.new_metadata_location.clone();
-        let next_warehouse_location = tokio::task::spawn_blocking(move || {
-            table_metadata_warehouse_location(&table_bucket, &metadata_location, &new_metadata_object)
+        let next_metadata_state = tokio::task::spawn_blocking(move || {
+            table_metadata_commit_state(&table_bucket, &metadata_location, &new_metadata_object)
         })
         .await
         .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
-        if next_warehouse_location
+        let warehouse_relocation = next_metadata_state
+            .warehouse_location
             .as_ref()
-            .is_some_and(|warehouse_location| warehouse_location != &current.warehouse_location)
-            && !publication.holds_table_bucket(&request.table_bucket)
-        {
+            .is_some_and(|warehouse_location| warehouse_location != &current.warehouse_location);
+        if warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket) {
             return table_commit_result(
                 &request.table_bucket,
                 &request.namespace,
@@ -4537,9 +4589,12 @@ where
 
         let mut next = current.clone();
         next.metadata_location = staged_commit_log.new_metadata_location.clone();
-        if let Some(warehouse_location) = next_warehouse_location {
+        if let Some(warehouse_location) = next_metadata_state.warehouse_location {
             next.warehouse_location = warehouse_location;
         }
+        if let Some(format_version) = next_metadata_state.format_version {
+            next.format_version = format_version;
+        }
         next.version_token = staged_commit_log.new_version_token.clone();
         next.generation = current.generation.saturating_add(1);
         if next.warehouse_location != current.warehouse_location {
@@ -4585,6 +4640,24 @@ where
             );
         }
 
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table)
+            || (warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket))
+        {
+            self.delete_created_table_warehouse_index(&next, reservation, "table publication fence lost")
+                .await;
+            return table_commit_result(
+                &request.table_bucket,
+                &request.namespace,
+                &request.table,
+                &request.commit_id,
+                &request.operation,
+                commit_started,
+                Err(TableCatalogStoreError::Internal(
+                    "table commit publication fence was lost before pointer update".to_string(),
+                )),
+            );
+        }
+
         let cas_started = Instant::now();
         let cas_result = self
             .write_entry_unlocked(
@@ -4662,20 +4735,21 @@ where
             )));
         };
         self.delete_owned_table_warehouse_index_for_drop(&entry).await?;
+        if !publication.holds_table_bucket(table_bucket)
+            || !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str())
+        {
+            self.restore_table_warehouse_index_after_failed_drop(&entry, "table publication fence lost")
+                .await;
+            return Err(TableCatalogStoreError::Internal(
+                "table drop publication fence was lost before catalog update".to_string(),
+            ));
+        }
         if let Err(err) = self.backend.delete_object_unlocked(self.catalog_bucket(), &object).await {
             match self.read_table_with_etag_unlocked(table_bucket, &namespace, &table).await {
                 Ok(None) => return Ok(()),
                 Ok(Some((current, _))) if current == entry => {
-                    if let Err(restore_err) = self.reserve_table_warehouse_index(&entry).await {
-                        tracing::warn!(
-                            table_bucket = %entry.table_bucket,
-                            namespace = %entry.namespace,
-                            table = %entry.table,
-                            table_id = %entry.table_id,
-                            error = %restore_err,
-                            "failed to restore table warehouse index after table entry delete failure"
-                        );
-                    }
+                    self.restore_table_warehouse_index_after_failed_drop(&entry, "table entry delete failed")
+                        .await;
                 }
                 Ok(Some(_)) => {
                     return Err(TableCatalogStoreError::Internal(format!(
@@ -4703,6 +4777,15 @@ where
         self.write_view_entry(entry, TableCatalogPutPrecondition::IfAbsent).await
     }
 
+    async fn create_view_with_publication(
+        &self,
+        entry: ViewEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
+        self.write_view_entry_with_publication(entry, TableCatalogPutPrecondition::IfAbsent, publication)
+            .await
+    }
+
     async fn list_views(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<Vec<ViewEntry>> {
         let namespace = parse_namespace_for_store(namespace)?;
         let mut entries = Vec::new();
@@ -4757,8 +4840,26 @@ where
     }
 
     async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult<ViewCommitResult> {
+        let publication = TableCommitLockPublication::new(&self.backend);
+        self.replace_view_with_publication(request, true, &publication).await
+    }
+
+    async fn replace_view_with_publication(
+        &self,
+        request: ViewCommitRequest,
+        table_bucket_fence_required: bool,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<ViewCommitResult> {
         let namespace = parse_namespace_for_store(&request.namespace)?;
         let view = parse_table_for_store(&request.view)?;
+        if table_bucket_fence_required {
+            publication.begin_table_bucket(&request.table_bucket).await?;
+            if !publication.holds_table_bucket(&request.table_bucket) {
+                return Err(TableCatalogStoreError::Internal(
+                    "view replacement requires a table-bucket publication fence".to_string(),
+                ));
+            }
+        }
         let _migration_guard = self.acquire_object_backed_catalog_write_permit(&request.table_bucket).await?;
         let namespace_path = self.paths.namespace_entry_path(&request.table_bucket, &namespace);
         let _namespace_guard = self
@@ -4767,6 +4868,16 @@ where
             .await?;
         let view_path = self.paths.view_entry_path(&request.table_bucket, &namespace, &view);
         let _guard = self.backend.acquire_write_lock(self.catalog_bucket(), &view_path).await?;
+        // Preserve catalog -> publication -> object lock order across rolling upgrades.
+        publication
+            .prepare(&request.table_bucket, &request.namespace, &request.view)
+            .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.view) {
+            return Err(TableCatalogStoreError::Internal(
+                "view replacement requires a table publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
         let Some((current, current_etag)) = self
             .read_view_with_etag_unlocked(&request.table_bucket, &namespace, &view)
             .await?
@@ -4814,6 +4925,14 @@ where
         })
         .await
         .map_err(|err| TableCatalogStoreError::Internal(format!("view metadata parser task failed: {err}")))??;
+        let warehouse_relocation = next_warehouse_location
+            .as_deref()
+            .is_some_and(|location| location != current.warehouse_location);
+        if warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "view warehouse relocation requires a table-bucket publication fence".to_string(),
+            ));
+        }
 
         let mut next = current;
         next.metadata_location = request.new_metadata_location;
@@ -4822,13 +4941,40 @@ where
         }
         next.version_token = format!("token-{}", Uuid::new_v4());
         next.generation = next.generation.saturating_add(1);
-        self.write_entry_unlocked(
-            self.catalog_bucket(),
-            &view_path,
-            &next,
-            TableCatalogPutPrecondition::IfMatch(current_etag),
-        )
-        .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.view)
+            || ((table_bucket_fence_required || warehouse_relocation) && !publication.holds_table_bucket(&request.table_bucket))
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "view replacement publication fence was lost before catalog update".to_string(),
+            ));
+        }
+        let write_result = self
+            .write_entry_unlocked(
+                self.catalog_bucket(),
+                &view_path,
+                &next,
+                TableCatalogPutPrecondition::IfMatch(current_etag),
+            )
+            .await;
+        if let Err(err) = write_result {
+            match self
+                .read_view_with_etag_unlocked(&request.table_bucket, &namespace, &view)
+                .await
+            {
+                Ok(Some((persisted, _))) if persisted == next => {}
+                Ok(_) => return Err(err),
+                Err(read_err) => {
+                    tracing::warn!(
+                        table_bucket = %request.table_bucket,
+                        namespace = %request.namespace,
+                        view = %request.view,
+                        error = %read_err,
+                        "failed to verify view state after an ambiguous catalog update"
+                    );
+                    return Err(err);
+                }
+            }
+        }
         Ok(ViewCommitResult { view: next })
     }
 

rustfs/src/table_catalog/store/strong.rs

@@ -554,7 +554,7 @@ where
 
     // Ordinary mutations hold the global migration read lock before the local write lock; migration takes the
     // write side before invoking its dedicated snapshot mutation methods.
-    async fn acquire_snapshot_write_permit(&self) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_snapshot_write_permit(&self) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         let lock_path = TableCatalogObjectPaths::default().backing_migration_global_fence_lock_path();
         self.object_backend.acquire_read_lock(RUSTFS_META_BUCKET, &lock_path).await
     }
@@ -1775,7 +1775,7 @@ where
         request: &TableCommitRequest,
         namespace: &Namespace,
         table: &IdentifierSegment,
-        next_warehouse_location: Option<String>,
+        next_metadata_state: TableMetadataCommitState,
     ) -> TableCatalogStoreResult<TableCommitResult> {
         let key = Self::table_key(&request.table_bucket, namespace, table);
         let current = Self::validate_new_table_commit_locked(state, &key, request)?;
@@ -1802,9 +1802,12 @@ where
 
         let mut next = current;
         next.metadata_location = commit_log.new_metadata_location.clone();
-        if let Some(warehouse_location) = next_warehouse_location {
+        if let Some(warehouse_location) = next_metadata_state.warehouse_location {
             next.warehouse_location = warehouse_location;
         }
+        if let Some(format_version) = next_metadata_state.format_version {
+            next.format_version = format_version;
+        }
         Self::ensure_table_warehouse_prefix_available_locked(state, &next, &key)?;
         next.version_token = commit_log.new_version_token.clone();
         next.generation = next.generation.saturating_add(1);
@@ -2170,6 +2173,7 @@ where
         let _write_guard = self.write_lock.lock().await;
         self.hydrate_state().await?;
         let key = Self::table_key(&entry.table_bucket, &namespace, &table);
+        let publication_identity = (entry.table_bucket.clone(), entry.namespace.clone(), entry.table.clone());
         let (snapshot, precondition, postcondition) = {
             let state = self.state.lock().await;
             Self::require_table_bucket_in_state(&state, &entry.table_bucket)?;
@@ -2198,6 +2202,13 @@ where
                 StrongSnapshotWritePostcondition::TablePresent(entry),
             )
         };
+        if !publication.holds_table_bucket(&publication_identity.0)
+            || !publication.holds_table(&publication_identity.0, &publication_identity.1, &publication_identity.2)
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "table registration publication fence was lost before snapshot update".to_string(),
+            ));
+        }
         self.finalize_snapshot_write(snapshot, precondition, postcondition).await
     }
 
@@ -2479,9 +2490,15 @@ where
             let result = match prepared_result {
                 Ok((result, Some((snapshot, precondition)))) => {
                     let postcondition = Self::commit_write_postcondition(&request.table_bucket, &result.commit_log);
-                    self.finalize_snapshot_write(snapshot, precondition, postcondition)
-                        .await
-                        .map(|_| result)
+                    if !publication.holds_table(&request.table_bucket, &request.namespace, &request.table) {
+                        Err(TableCatalogStoreError::Internal(
+                            "table commit publication fence was lost before snapshot update".to_string(),
+                        ))
+                    } else {
+                        self.finalize_snapshot_write(snapshot, precondition, postcondition)
+                            .await
+                            .map(|_| result)
+                    }
                 }
                 Ok((result, None)) => Ok(result),
                 Err(err) => Err(err),
@@ -2519,8 +2536,8 @@ where
         validate_commit_metadata_digest(&request, &new_metadata_object)?;
         let table_bucket = request.table_bucket.clone();
         let metadata_location = request.new_metadata_location.clone();
-        let next_warehouse_location = tokio::task::spawn_blocking(move || {
-            table_metadata_warehouse_location(&table_bucket, &metadata_location, &new_metadata_object)
+        let next_metadata_state = tokio::task::spawn_blocking(move || {
+            table_metadata_commit_state(&table_bucket, &metadata_location, &new_metadata_object)
         })
         .await
         .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
@@ -2539,11 +2556,11 @@ where
                     ))
                 })?
         };
-        if next_warehouse_location
+        let warehouse_relocation = next_metadata_state
+            .warehouse_location
             .as_ref()
-            .is_some_and(|warehouse_location| warehouse_location != &current_warehouse_location)
-            && !publication.holds_table_bucket(&request.table_bucket)
-        {
+            .is_some_and(|warehouse_location| warehouse_location != &current_warehouse_location);
+        if warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket) {
             return table_commit_result(
                 &request.table_bucket,
                 &request.namespace,
@@ -2561,7 +2578,7 @@ where
         let prepared_result = {
             let state = self.state.lock().await;
             let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state);
-            match Self::apply_commit_locked(&mut draft_state, &request, &namespace, &table, next_warehouse_location) {
+            match Self::apply_commit_locked(&mut draft_state, &request, &namespace, &table, next_metadata_state) {
                 Ok(result) => Self::snapshot_from_mutated_state_locked(&mut draft_state, self.snapshot_write_version)
                     .map(|snapshot| (result, snapshot, precondition)),
                 Err(err) => Err(err),
@@ -2570,7 +2587,16 @@ where
         let result = match prepared_result {
             Ok((result, snapshot, precondition)) => {
                 let postcondition = Self::commit_write_postcondition(&request.table_bucket, &result.commit_log);
-                match self.finalize_snapshot_write(snapshot, precondition, postcondition).await {
+                let snapshot_result = if publication.holds_table(&request.table_bucket, &request.namespace, &request.table)
+                    && (!warehouse_relocation || publication.holds_table_bucket(&request.table_bucket))
+                {
+                    self.finalize_snapshot_write(snapshot, precondition, postcondition).await
+                } else {
+                    Err(TableCatalogStoreError::Internal(
+                        "table commit publication fence was lost before snapshot update".to_string(),
+                    ))
+                };
+                match snapshot_result {
                     Ok(()) => Ok(result),
                     Err(err) => {
                         let replay = {
@@ -2647,13 +2673,26 @@ where
                 },
             )
         };
+        if !publication.holds_table_bucket(table_bucket)
+            || !publication.holds_table(table_bucket, &namespace.public_name(), table.as_str())
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "table drop publication fence was lost before snapshot update".to_string(),
+            ));
+        }
         self.finalize_snapshot_write(snapshot, precondition, postcondition).await
     }
 
     async fn create_view(&self, entry: ViewEntry) -> TableCatalogStoreResult<()> {
-        let _migration_guard = self.acquire_snapshot_write_permit().await?;
-        let _write_guard = self.write_lock.lock().await;
-        self.hydrate_state().await?;
+        let publication = TableCommitLockPublication::new(&self.object_backend);
+        self.create_view_with_publication(entry, &publication).await
+    }
+
+    async fn create_view_with_publication(
+        &self,
+        entry: ViewEntry,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<()> {
         validate_view_entry_version_and_id(&entry)?;
         validate_view_warehouse_location(&entry.table_bucket, &entry.warehouse_location)?;
         let namespace = parse_namespace_for_store(&entry.namespace)?;
@@ -2663,7 +2702,26 @@ where
                 "view metadata location must be inside the view metadata directory".to_string(),
             ));
         }
+        publication.begin_table_bucket(&entry.table_bucket).await?;
+        if !publication.holds_table_bucket(&entry.table_bucket) {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation requires a table-bucket publication fence".to_string(),
+            ));
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
+        let _migration_guard = self.acquire_snapshot_write_permit().await?;
+        publication
+            .prepare(&entry.table_bucket, &entry.namespace, &entry.view)
+            .await?;
+        if !publication.holds_table(&entry.table_bucket, &entry.namespace, &entry.view) {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation requires a table publication fence".to_string(),
+            ));
+        }
+        let _write_guard = self.write_lock.lock().await;
+        self.hydrate_state().await?;
         let key = Self::table_key(&entry.table_bucket, &namespace, &view);
+        let publication_identity = (entry.table_bucket.clone(), entry.namespace.clone(), entry.view.clone());
         let (snapshot, precondition, postcondition) = {
             let state = self.state.lock().await;
             Self::require_table_bucket_in_state(&state, &entry.table_bucket)?;
@@ -2682,6 +2740,13 @@ where
                 StrongSnapshotWritePostcondition::ViewPresent(entry),
             )
         };
+        if !publication.holds_table_bucket(&publication_identity.0)
+            || !publication.holds_table(&publication_identity.0, &publication_identity.1, &publication_identity.2)
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "view creation publication fence was lost before snapshot update".to_string(),
+            ));
+        }
         self.finalize_snapshot_write(snapshot, precondition, postcondition).await
     }
 
@@ -2751,11 +2816,38 @@ where
     }
 
     async fn replace_view(&self, request: ViewCommitRequest) -> TableCatalogStoreResult<ViewCommitResult> {
+        let publication = TableCommitLockPublication::new(&self.object_backend);
+        self.replace_view_with_publication(request, true, &publication).await
+    }
+
+    async fn replace_view_with_publication(
+        &self,
+        request: ViewCommitRequest,
+        table_bucket_fence_required: bool,
+        publication: &(dyn TableCommitPublication + Sync),
+    ) -> TableCatalogStoreResult<ViewCommitResult> {
+        if table_bucket_fence_required {
+            publication.begin_table_bucket(&request.table_bucket).await?;
+            if !publication.holds_table_bucket(&request.table_bucket) {
+                return Err(TableCatalogStoreError::Internal(
+                    "view replacement requires a table-bucket publication fence".to_string(),
+                ));
+            }
+        }
+        let _publication_completion = TableCommitPublicationCompletion::new(publication);
         let _migration_guard = self.acquire_snapshot_write_permit().await?;
-        let write_guard = self.write_lock.lock().await;
-        self.hydrate_state().await?;
         let namespace = parse_namespace_for_store(&request.namespace)?;
         let view = parse_table_for_store(&request.view)?;
+        publication
+            .prepare(&request.table_bucket, &request.namespace, &request.view)
+            .await?;
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.view) {
+            return Err(TableCatalogStoreError::Internal(
+                "view replacement requires a table publication fence".to_string(),
+            ));
+        }
+        let write_guard = self.write_lock.lock().await;
+        self.hydrate_state().await?;
         let key = Self::table_key(&request.table_bucket, &namespace, &view);
         let expected_view_id = {
             let state = self.state.lock().await;
@@ -2798,7 +2890,7 @@ where
 
         let _write_guard = self.write_lock.lock().await;
         self.hydrate_state().await?;
-        let (snapshot, precondition, next, postcondition) = {
+        let (snapshot, precondition, next, postcondition, warehouse_relocation) = {
             let state = self.state.lock().await;
             Self::ensure_identifier_is_unambiguous_locked(&state, &key)?;
             let Some(current) = state.views.get(&key).cloned() else {
@@ -2828,6 +2920,14 @@ where
                     "current view metadata location does not match expected location".to_string(),
                 ));
             }
+            let warehouse_relocation = next_warehouse_location
+                .as_deref()
+                .is_some_and(|location| location != current.warehouse_location);
+            if warehouse_relocation && !publication.holds_table_bucket(&request.table_bucket) {
+                return Err(TableCatalogStoreError::Internal(
+                    "view warehouse relocation requires a table-bucket publication fence".to_string(),
+                ));
+            }
 
             let mut next = current;
             next.metadata_location = request.new_metadata_location;
@@ -2843,8 +2943,16 @@ where
                 precondition,
                 next.clone(),
                 StrongSnapshotWritePostcondition::ViewPresent(next),
+                warehouse_relocation,
             )
         };
+        if !publication.holds_table(&request.table_bucket, &request.namespace, &request.view)
+            || ((table_bucket_fence_required || warehouse_relocation) && !publication.holds_table_bucket(&request.table_bucket))
+        {
+            return Err(TableCatalogStoreError::Internal(
+                "view replacement publication fence was lost before snapshot update".to_string(),
+            ));
+        }
         self.finalize_snapshot_write(snapshot, precondition, postcondition).await?;
         Ok(ViewCommitResult { view: next })
     }

rustfs/src/table_catalog/test_support.rs

@@ -55,23 +55,35 @@ pub(crate) fn table_metadata_json(table_uuid: &str, location: &str) -> serde_jso
     })
 }
 
-pub(crate) fn manifest_list_avro_bytes(manifest_paths: &[&str], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
-    let manifests = manifest_paths
-        .iter()
-        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
-        .collect::<Vec<_>>();
-    manifest_list_avro_entries_with_partition_specs(&manifests)
-}
-
-pub(crate) fn manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
+pub(crate) fn manifest_list_avro_bytes(manifests: &[(&str, usize)], sequence_number: i64, snapshot_id: i64) -> Vec<u8> {
     let manifests = manifests
         .iter()
-        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
+        .map(|(manifest_path, manifest_length)| (*manifest_path, *manifest_length, 0, sequence_number, snapshot_id))
         .collect::<Vec<_>>();
     manifest_list_avro_entries_with_partition_specs(&manifests)
 }
 
-pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
+pub(crate) fn manifest_list_avro_entries(manifests: &[(&str, usize, i64, i64)]) -> Vec<u8> {
+    let manifests = manifests
+        .iter()
+        .map(|(manifest_path, manifest_length, sequence_number, snapshot_id)| {
+            (*manifest_path, *manifest_length, 0, *sequence_number, *snapshot_id)
+        })
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, usize, i32, i64, i64)]) -> Vec<u8> {
+    let manifests = manifests
+        .iter()
+        .map(|(path, length, spec_id, sequence_number, snapshot_id)| {
+            (*path, *length, *spec_id, 0, *sequence_number, *snapshot_id)
+        })
+        .collect::<Vec<_>>();
+    manifest_list_avro_entries_with_content(&manifests)
+}
+
+pub(crate) fn manifest_list_avro_entries_with_content(manifests: &[(&str, usize, i32, i32, i64, i64)]) -> Vec<u8> {
     let schema = apache_avro::Schema::parse_str(
         r#"
             {
@@ -97,16 +109,19 @@ pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str
     )
     .expect("manifest list avro schema should parse");
     let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
-    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
+    for (manifest_path, manifest_length, partition_spec_id, content, sequence_number, snapshot_id) in manifests {
         writer
             .append_value(apache_avro::types::Value::Record(vec![
                 (
                     "manifest_path".to_string(),
                     apache_avro::types::Value::String((*manifest_path).to_string()),
                 ),
-                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
+                (
+                    "manifest_length".to_string(),
+                    apache_avro::types::Value::Long(i64::try_from(*manifest_length).expect("test manifest length should fit")),
+                ),
                 ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
-                ("content".to_string(), apache_avro::types::Value::Int(0)),
+                ("content".to_string(), apache_avro::types::Value::Int(*content)),
                 ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
                 ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
                 ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
@@ -122,7 +137,86 @@ pub(crate) fn manifest_list_avro_entries_with_partition_specs(manifests: &[(&str
     writer.into_inner().expect("manifest list avro bytes should flush")
 }
 
+pub(crate) fn manifest_list_avro_entries_with_nullable_counts(manifests: &[(&str, usize, i32, i64, i64)]) -> Vec<u8> {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+            {
+              "type": "record",
+              "name": "manifest_file",
+              "fields": [
+                {"name": "manifest_path", "type": "string"},
+                {"name": "manifest_length", "type": "long"},
+                {"name": "partition_spec_id", "type": "int"},
+                {"name": "content", "type": "int"},
+                {"name": "sequence_number", "type": "long"},
+                {"name": "min_sequence_number", "type": "long"},
+                {"name": "added_snapshot_id", "type": "long"},
+                {"name": "added_files_count", "type": ["null", "int"], "default": null},
+                {"name": "existing_files_count", "type": ["null", "int"], "default": null},
+                {"name": "deleted_files_count", "type": ["null", "int"], "default": null},
+                {"name": "added_rows_count", "type": ["null", "long"], "default": null},
+                {"name": "existing_rows_count", "type": ["null", "long"], "default": null},
+                {"name": "deleted_rows_count", "type": ["null", "long"], "default": null}
+              ]
+            }
+            "#,
+    )
+    .expect("manifest list avro schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest list writer should initialize");
+    for (manifest_path, manifest_length, partition_spec_id, sequence_number, snapshot_id) in manifests {
+        writer
+            .append_value(apache_avro::types::Value::Record(vec![
+                (
+                    "manifest_path".to_string(),
+                    apache_avro::types::Value::String((*manifest_path).to_string()),
+                ),
+                (
+                    "manifest_length".to_string(),
+                    apache_avro::types::Value::Long(i64::try_from(*manifest_length).expect("test manifest length should fit")),
+                ),
+                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(*partition_spec_id)),
+                ("content".to_string(), apache_avro::types::Value::Int(0)),
+                ("sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(*sequence_number)),
+                ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(*snapshot_id)),
+                (
+                    "added_files_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+                (
+                    "existing_files_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+                (
+                    "deleted_files_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+                (
+                    "added_rows_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+                (
+                    "existing_rows_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+                (
+                    "deleted_rows_count".to_string(),
+                    apache_avro::types::Value::Union(0, Box::new(apache_avro::types::Value::Null)),
+                ),
+            ]))
+            .expect("manifest list record should append");
+    }
+    writer.into_inner().expect("manifest list avro bytes should flush")
+}
+
 pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u8> {
+    manifest_avro_bytes_with_partition_spec(files, None)
+}
+
+pub(crate) fn manifest_avro_bytes_with_partition_spec(
+    files: &[(&str, i32, i32, i64, i64)],
+    partition_spec_id: Option<i32>,
+) -> Vec<u8> {
     let schema = apache_avro::Schema::parse_str(
         r#"
             {
@@ -141,6 +235,7 @@ pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u
                     "fields": [
                       {"name": "content", "type": "int"},
                       {"name": "file_path", "type": "string"},
+                      {"name": "partition", "type": {"type": "record", "name": "partition", "fields": []}},
                       {"name": "record_count", "type": "long"},
                       {"name": "file_size_in_bytes", "type": "long"}
                     ]
@@ -152,6 +247,11 @@ pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u
     )
     .expect("manifest avro schema should parse");
     let mut writer = apache_avro::Writer::new(&schema, Vec::new()).expect("manifest writer should initialize");
+    if let Some(partition_spec_id) = partition_spec_id {
+        writer
+            .add_user_metadata("partition-spec-id".to_string(), partition_spec_id.to_string())
+            .expect("manifest partition spec metadata should write");
+    }
     for (file_path, content, status, snapshot_id, sequence_number) in files {
         writer
             .append_value(apache_avro::types::Value::Record(vec![
@@ -164,6 +264,7 @@ pub(crate) fn manifest_avro_bytes(files: &[(&str, i32, i32, i64, i64)]) -> Vec<u
                     apache_avro::types::Value::Record(vec![
                         ("content".to_string(), apache_avro::types::Value::Int(*content)),
                         ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("partition".to_string(), apache_avro::types::Value::Record(Vec::new())),
                         ("record_count".to_string(), apache_avro::types::Value::Long(1)),
                         ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
                     ]),
@@ -200,6 +301,7 @@ pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i
                     "fields": [
                       {"name": "content", "type": "int"},
                       {"name": "file_path", "type": "string"},
+                      {"name": "partition", "type": {"type": "record", "name": "partition", "fields": []}},
                       {"name": "record_count", "type": "long"},
                       {"name": "file_size_in_bytes", "type": "long"}
                     ]
@@ -223,6 +325,7 @@ pub(crate) fn manifest_avro_bytes_with_nullable_sequences(files: &[(&str, i32, i
                     apache_avro::types::Value::Record(vec![
                         ("content".to_string(), apache_avro::types::Value::Int(*content)),
                         ("file_path".to_string(), apache_avro::types::Value::String((*file_path).to_string())),
+                        ("partition".to_string(), apache_avro::types::Value::Record(Vec::new())),
                         ("record_count".to_string(), apache_avro::types::Value::Long(1)),
                         ("file_size_in_bytes".to_string(), apache_avro::types::Value::Long(1)),
                     ]),
@@ -284,7 +387,7 @@ pub(crate) struct BlockingObjectPublication {
     backend: TestCatalogObjectBackend,
     object: String,
     started: Arc<tokio::sync::Notify>,
-    guard: Arc<parking_lot::Mutex<Option<Box<dyn Send>>>>,
+    guard: Arc<parking_lot::Mutex<Option<TableCatalogLockGuard>>>,
 }
 
 impl BlockingObjectPublication {
@@ -812,7 +915,7 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
             .collect())
     }
 
-    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_write_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         self.lock_attempts.lock().await.push((bucket.to_string(), object.to_string()));
         {
             let mut state = self.state.lock().await;
@@ -828,10 +931,10 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
                 .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
                 .clone()
         };
-        Ok(Box::new(lock.write_owned().await))
+        Ok(TableCatalogLockGuard::stable(lock.write_owned().await))
     }
 
-    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Box<dyn Send>> {
+    async fn acquire_read_lock(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<TableCatalogLockGuard> {
         // The admin fake implemented only acquire_write_lock, so the trait's
         // default read->write delegation made read acquisitions observable in
         // lock_attempts as well; keep that (backlog#1837 PR2).
@@ -850,7 +953,7 @@ impl TableCatalogObjectBackend for TestCatalogObjectBackend {
                 .or_insert_with(|| std::sync::Arc::new(tokio::sync::RwLock::new(())))
                 .clone()
         };
-        Ok(Box::new(lock.read_owned().await))
+        Ok(TableCatalogLockGuard::stable(lock.read_owned().await))
     }
 }
 
@@ -1165,6 +1268,8 @@ pub(crate) struct TestTableCatalogStore {
     pub(crate) fail_put_table_bucket: tokio::sync::Mutex<bool>,
     pub(crate) register_table_pause: Option<TestCatalogPublishPause>,
     pub(crate) commit_table_pause: Option<TestCatalogPublishPause>,
+    pub(crate) create_view_pause: Option<TestCatalogPublishPause>,
+    pub(crate) replace_view_pause: Option<TestCatalogPublishPause>,
 }
 
 #[async_trait::async_trait]
@@ -1473,6 +1578,10 @@ impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore {
                 entry.table_bucket, entry.namespace
             )));
         }
+        if let Some(pause) = &self.create_view_pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
         self.views.lock().await.push(entry);
         Ok(())
     }
@@ -1531,6 +1640,10 @@ impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore {
                 "current view metadata location does not match expected location".to_string(),
             ));
         }
+        if let Some(pause) = &self.replace_view_pause {
+            pause.started.notify_one();
+            pause.release.notified().await;
+        }
         let mut next = current;
         next.metadata_location = request.new_metadata_location;
         next.version_token = "token-view-committed".to_string();

scripts/table-catalog/failure_coverage.py

@@ -132,18 +132,8 @@ def failure_probe_plan(warehouse: str, namespace: str, table: str, rest_path: st
                 "expected-version-token": "stale-token-from-previous-load",
                 "expected-metadata-location": "current-metadata-location-from-load-table",
                 "new-metadata-location": f"s3://{warehouse}/tables/table-id/metadata/conflict_probe.metadata.json",
-                "requirements": [
-                    {
-                        "type": "assert-current-snapshot-id",
-                        "snapshot-id": 0,
-                    }
-                ],
-                "updates": [
-                    {
-                        "action": "set-current-schema",
-                        "schema-id": 0,
-                    }
-                ],
+                "requirements": [],
+                "updates": [],
             },
         ),
         probe_step(
@@ -157,6 +147,8 @@ def failure_probe_plan(warehouse: str, namespace: str, table: str, rest_path: st
                 "expected-version-token": "current-version-token-from-load-table",
                 "expected-metadata-location": "current-metadata-location-from-load-table",
                 "new-metadata-location": f"s3://{warehouse}/tables/table-id/metadata/does_not_exist.metadata.json",
+                "requirements": [],
+                "updates": [],
             },
         ),
         probe_step(

scripts/table-catalog/test_failure_coverage.py

@@ -48,6 +48,8 @@ class FailureCoverageTest(unittest.TestCase):
         self.assertIn("expected-version-token", by_name["stale-token-commit-conflict"]["body"])
         self.assertIn("expected-metadata-location", by_name["stale-token-commit-conflict"]["body"])
         self.assertIn("new-metadata-location", by_name["stale-token-commit-conflict"]["body"])
+        self.assertEqual(by_name["stale-token-commit-conflict"]["body"]["requirements"], [])
+        self.assertEqual(by_name["stale-token-commit-conflict"]["body"]["updates"], [])
         self.assertNotIn("base", by_name["stale-token-commit-conflict"]["body"])
         self.assertEqual(
             by_name["diagnostics-after-finalization-gap"]["path"],
@@ -56,6 +58,8 @@ class FailureCoverageTest(unittest.TestCase):
         self.assertEqual(by_name["diagnostics-after-finalization-gap"]["method"], "GET")
         self.assertEqual(by_name["recovery-repairs-idempotency-index"]["method"], "POST")
         self.assertIn("does_not_exist.metadata.json", json.dumps(by_name["missing-metadata-object-rejected"]))
+        self.assertEqual(by_name["missing-metadata-object-rejected"]["body"]["requirements"], [])
+        self.assertEqual(by_name["missing-metadata-object-rejected"]["body"]["updates"], [])
         self.assertNotIn("base", by_name["missing-metadata-object-rejected"]["body"])
 
     def test_cli_prints_failure_matrix_and_probe_plan(self) -> None: