feat(table-catalog): validate Iceberg metadata graphs (#5758)

* feat(table-catalog): validate Iceberg metadata graphs

* fix(table-catalog): scope Iceberg graph validation

* fix(table-catalog): preserve commit validation semantics

---------

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

Cargo.lock

@@ -9017,6 +9017,7 @@ dependencies = [
  "const-str",
  "datafusion",
  "flatbuffers",
+ "flate2",
  "futures",
  "futures-lite",
  "futures-util",

rustfs/Cargo.toml

@@ -267,6 +267,7 @@ tower-http = { workspace = true, features = ["trace", "compression-full", "cors"
 apache-avro = { workspace = true }
 bytes = { workspace = true, features = ["serde"] }
 chrono = { workspace = true, features = ["serde"] }
+flate2 = { workspace = true }
 flatbuffers.workspace = true
 rmp-serde.workspace = true
 quick-xml.workspace = true

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

@@ -1954,28 +1954,15 @@ fn validate_view_location_in_bucket(bucket: &str, location: &str) -> S3Result<()
 }
 
 fn metadata_table_uuid(metadata: &serde_json::Value) -> S3Result<&str> {
-    metadata
-        .get("table-uuid")
-        .and_then(serde_json::Value::as_str)
-        .filter(|uuid| !uuid.is_empty())
-        .ok_or_else(|| s3_error!(InvalidRequest, "table metadata is missing table-uuid"))
+    crate::table_catalog::table_metadata_uuid(metadata).map_err(catalog_store_error)
 }
 
 fn metadata_format_version(metadata: &serde_json::Value) -> S3Result<u16> {
-    let version = metadata
-        .get("format-version")
-        .and_then(serde_json::Value::as_u64)
-        .filter(|version| *version > 0)
-        .ok_or_else(|| s3_error!(InvalidRequest, "table metadata is missing format-version"))?;
-    u16::try_from(version).map_err(|_| s3_error!(InvalidRequest, "table metadata format-version is too large"))
+    crate::table_catalog::table_metadata_format_version(metadata).map_err(catalog_store_error)
 }
 
 fn metadata_table_location(metadata: &serde_json::Value) -> S3Result<&str> {
-    metadata
-        .get("location")
-        .and_then(serde_json::Value::as_str)
-        .filter(|location| !location.is_empty())
-        .ok_or_else(|| s3_error!(InvalidRequest, "table metadata is missing location"))
+    crate::table_catalog::table_metadata_location(metadata).map_err(catalog_store_error)
 }
 
 fn validate_metadata_table_location_in_bucket(bucket: &str, metadata: &serde_json::Value) -> S3Result<()> {
@@ -1991,6 +1978,15 @@ fn validate_metadata_view_location_in_bucket(bucket: &str, metadata: &serde_json
 fn validate_metadata_matches_current_metadata(
     current_metadata: &serde_json::Value,
     target_metadata: &serde_json::Value,
+) -> S3Result<()> {
+    crate::table_catalog::validate_supported_table_metadata(current_metadata).map_err(catalog_store_error)?;
+    crate::table_catalog::validate_supported_table_metadata(target_metadata).map_err(catalog_store_error)?;
+    validate_metadata_identity_matches_current_metadata(current_metadata, target_metadata)
+}
+
+fn validate_metadata_identity_matches_current_metadata(
+    current_metadata: &serde_json::Value,
+    target_metadata: &serde_json::Value,
 ) -> S3Result<()> {
     let expected_table_uuid = metadata_table_uuid(current_metadata)?;
     metadata_format_version(current_metadata)?;
@@ -2031,6 +2027,7 @@ fn adopt_registered_metadata_identity(
     entry: &mut crate::table_catalog::TableEntry,
     metadata: &serde_json::Value,
 ) -> S3Result<()> {
+    crate::table_catalog::validate_supported_table_metadata(metadata).map_err(catalog_store_error)?;
     entry.table_uuid = metadata_table_uuid(metadata)?.to_string();
     entry.format_version = metadata_format_version(metadata)?;
     entry.warehouse_location = metadata_table_location(metadata)?.to_string();
@@ -2112,20 +2109,42 @@ fn table_entry_from_create_table_request(
     namespace: &crate::table_catalog::Namespace,
     request: CreateTableRequest,
 ) -> S3Result<(crate::table_catalog::TableEntry, serde_json::Value)> {
-    if request.stage_create {
+    let CreateTableRequest {
+        name,
+        location,
+        schema,
+        partition_spec,
+        write_order,
+        stage_create,
+        mut properties,
+    } = request;
+    if stage_create {
         return Err(s3_error!(NotImplemented, "stage-create is not supported"));
     }
 
-    let table = crate::table_catalog::IdentifierSegment::parse(request.name)
+    let table = crate::table_catalog::IdentifierSegment::parse(name)
         .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
     let table_id = Uuid::new_v4().to_string();
     let table_uuid = Uuid::new_v4().to_string();
-    let warehouse_location = request.location.unwrap_or_else(|| format!("s3://{bucket}/tables/{table_id}"));
+    let format_version = match properties.remove("format-version") {
+        Some(version) => version
+            .parse::<u16>()
+            .map_err(|_| s3_error!(InvalidRequest, "format-version property must be an integer"))?,
+        None => 2,
+    };
+    if !(1..=2).contains(&format_version) {
+        return Err(iceberg_rest_error(
+            ICEBERG_ERROR_UNSUPPORTED_OPERATION,
+            StatusCode::NOT_ACCEPTABLE,
+            format!("unsupported Iceberg table format-version: {format_version}"),
+        ));
+    }
+    let warehouse_location = location.unwrap_or_else(|| format!("s3://{bucket}/tables/{table_id}"));
     validate_table_location_in_bucket(bucket, &warehouse_location)?;
     let metadata_location =
         crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
 
-    let mut entry = crate::table_catalog::TableEntry {
+    let entry = crate::table_catalog::TableEntry {
         version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
         table_bucket: bucket.to_string(),
         namespace: namespace.public_name(),
@@ -2133,28 +2152,17 @@ fn table_entry_from_create_table_request(
         table_id,
         table_uuid,
         format: "ICEBERG".to_string(),
-        format_version: 2,
+        format_version,
         warehouse_location,
         metadata_location,
         version_token: format!("token-{}", Uuid::new_v4()),
         generation: 1,
         state: crate::table_catalog::TableCatalogEntryState::Active,
-        properties: request.properties,
+        properties,
         created_at: None,
         updated_at: None,
     };
-    let metadata = initial_table_metadata_json(
-        &entry,
-        request.schema,
-        request.partition_spec,
-        request.write_order,
-        entry.properties.clone(),
-    )?;
-    entry.format_version = metadata
-        .get("format-version")
-        .and_then(serde_json::Value::as_u64)
-        .and_then(|version| u16::try_from(version).ok())
-        .unwrap_or(2);
+    let metadata = initial_table_metadata_json(&entry, schema, partition_spec, write_order, entry.properties.clone())?;
     Ok((entry, metadata))
 }
 
@@ -2254,11 +2262,10 @@ fn initial_table_metadata_json(
         .and_then(serde_json::Value::as_i64)
         .ok_or_else(|| s3_error!(InvalidRequest, "sort order-id must be an integer"))?;
 
-    Ok(serde_json::json!({
+    let mut metadata = serde_json::json!({
         "format-version": entry.format_version,
         "table-uuid": entry.table_uuid,
         "location": entry.warehouse_location,
-        "last-sequence-number": 0,
         "last-updated-ms": current_time_millis(),
         "last-column-id": last_column_id,
         "schemas": [schema],
@@ -2273,7 +2280,13 @@ fn initial_table_metadata_json(
         "snapshot-log": [],
         "metadata-log": [],
         "refs": {}
-    }))
+    });
+    if entry.format_version == 2 {
+        metadata_object_mut(&mut metadata)?.insert("last-sequence-number".to_string(), serde_json::Value::from(0));
+    }
+    crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?;
+    crate::table_catalog::validate_supported_table_metadata(&metadata).map_err(catalog_store_error)?;
+    Ok(metadata)
 }
 
 fn initial_view_metadata_json(
@@ -2564,6 +2577,9 @@ fn apply_table_commit_updates(
         }
     }
 
+    if metadata.get("format-version").is_some() {
+        crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata).map_err(catalog_store_error)?;
+    }
     append_previous_metadata_log(&mut metadata, previous_metadata_location)?;
     metadata_object_mut(&mut metadata)?.insert("last-updated-ms".to_string(), serde_json::Value::from(current_time_millis()));
     Ok(metadata)
@@ -2631,6 +2647,7 @@ fn apply_view_commit_updates(
         }
     }
 
+    crate::table_catalog::validate_view_metadata_references(&metadata).map_err(catalog_store_error)?;
     append_previous_metadata_log(&mut metadata, previous_metadata_location)?;
     metadata_object_mut(&mut metadata)?.insert("last-updated-ms".to_string(), serde_json::Value::from(current_time_millis()));
     Ok(metadata)
@@ -2701,10 +2718,29 @@ fn apply_upgrade_format_version_update(metadata: &mut serde_json::Value, update:
     let current = metadata
         .get("format-version")
         .and_then(serde_json::Value::as_i64)
-        .unwrap_or_default();
+        .ok_or_else(|| s3_error!(InvalidRequest, "current table metadata is missing format-version"))?;
+    if !(1..=2).contains(&version) {
+        return Err(iceberg_rest_error(
+            ICEBERG_ERROR_UNSUPPORTED_OPERATION,
+            StatusCode::NOT_ACCEPTABLE,
+            format!("unsupported Iceberg table format-version: {version}"),
+        ));
+    }
     if version < current {
         return Err(s3_error!(InvalidRequest, "format-version cannot be downgraded"));
     }
+    if current == 1
+        && version == 2
+        && let Some(snapshots) = metadata.get_mut("snapshots").and_then(serde_json::Value::as_array_mut)
+    {
+        for snapshot in snapshots {
+            snapshot
+                .as_object_mut()
+                .ok_or_else(|| s3_error!(InvalidRequest, "snapshot must be an object"))?
+                .entry("sequence-number".to_string())
+                .or_insert_with(|| serde_json::Value::from(0));
+        }
+    }
     metadata_object_mut(metadata)?.insert("format-version".to_string(), serde_json::Value::from(version));
     Ok(())
 }
@@ -3297,12 +3333,14 @@ where
             crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestFile,
         )?;
         let manifest_object = metadata_backend
-            .read_object(bucket, &manifest_key)
+            .read_object_limited(bucket, &manifest_key, crate::table_catalog::TABLE_MANIFEST_AVRO_MAX_SIZE)
             .await
             .map_err(catalog_store_error)?
             .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest object is missing"))?;
-        let file_references =
-            crate::table_catalog::data_file_references_from_manifest_avro(&manifest_object.data).map_err(catalog_store_error)?;
+        let file_references = crate::table_catalog::decode_manifest_avro_async(manifest_object.data)
+            .await
+            .map_err(catalog_store_error)?
+            .references;
         let mut references = Vec::with_capacity(file_references.len());
         for mut reference in file_references {
             if reference.snapshot_id.is_none() {
@@ -3353,15 +3391,14 @@ where
             crate::table_catalog::TableMetadataMaintenanceObjectKind::ManifestList,
         )?;
         let manifest_list_object = metadata_backend
-            .read_object(bucket, &manifest_list_key)
+            .read_object_limited(bucket, &manifest_list_key, crate::table_catalog::TABLE_MANIFEST_AVRO_MAX_SIZE)
             .await
             .map_err(catalog_store_error)?
             .ok_or_else(|| s3_error!(InvalidRequest, "snapshot manifest-list object is missing"))?;
-        let references = crate::table_catalog::manifest_list_references_from_manifest_list_avro(&manifest_list_object.data)
-            .map_err(catalog_store_error)?;
-        if references.is_empty() {
-            return Err(s3_error!(InvalidRequest, "snapshot manifest-list must reference at least one manifest"));
-        }
+        let references = crate::table_catalog::decode_manifest_list_avro_async(manifest_list_object.data)
+            .await
+            .map_err(catalog_store_error)?
+            .references;
         return Ok(references
             .into_iter()
             .map(|reference| SnapshotManifestLocation {
@@ -3375,9 +3412,6 @@ where
     let Some(manifests) = snapshot.get("manifests").and_then(serde_json::Value::as_array) else {
         return Err(s3_error!(InvalidRequest, "snapshot manifest-list is required"));
     };
-    if manifests.is_empty() {
-        return Err(s3_error!(InvalidRequest, "snapshot manifests must reference at least one manifest"));
-    }
     manifests
         .iter()
         .map(|manifest| {
@@ -3838,6 +3872,9 @@ where
     let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?;
     validate_metadata_table_location_in_bucket(bucket, &metadata)?;
     adopt_registered_metadata_identity(&mut entry, &metadata)?;
+    let table = crate::table_catalog::IdentifierSegment::parse(entry.table.clone())
+        .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
+    validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table, &entry, None, &metadata).await?;
     store
         .register_table(entry.clone())
         .await
@@ -3912,21 +3949,51 @@ async fn read_table_metadata_json(
     bucket: &str,
     metadata_location: &str,
 ) -> S3Result<serde_json::Value> {
-    let Some(object) = metadata_backend
-        .read_object(bucket, metadata_location)
+    let Some(metadata) = crate::table_catalog::read_table_metadata_value(metadata_backend, bucket, metadata_location)
         .await
         .map_err(catalog_store_error)?
     else {
         return Err(s3_error!(InvalidRequest, "table metadata object not found: {metadata_location}"));
     };
-    let metadata = serde_json::from_slice::<serde_json::Value>(&object.data)
-        .map_err(|err| s3_error!(InvalidRequest, "failed to parse table metadata JSON: {}", err))?;
-    if !metadata.is_object() {
-        return Err(s3_error!(InvalidRequest, "table metadata JSON must be an object"));
-    }
     Ok(metadata)
 }
 
+async fn validate_table_metadata_snapshot_graph<B>(
+    metadata_backend: &B,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    table: &crate::table_catalog::IdentifierSegment,
+    entry: &crate::table_catalog::TableEntry,
+    current_metadata: Option<&serde_json::Value>,
+    metadata: &serde_json::Value,
+) -> S3Result<()>
+where
+    B: crate::table_catalog::TableCatalogObjectBackend,
+{
+    validate_table_metadata_snapshot_graph_result(metadata_backend, bucket, namespace, table, entry, current_metadata, metadata)
+        .await
+        .map_err(catalog_store_error)
+}
+
+async fn validate_table_metadata_snapshot_graph_result<B>(
+    metadata_backend: &B,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    table: &crate::table_catalog::IdentifierSegment,
+    entry: &crate::table_catalog::TableEntry,
+    current_metadata: Option<&serde_json::Value>,
+    metadata: &serde_json::Value,
+) -> crate::table_catalog::TableCatalogStoreResult<()>
+where
+    B: crate::table_catalog::TableCatalogObjectBackend,
+{
+    let mut target_entry = entry.clone();
+    target_entry.warehouse_location = crate::table_catalog::table_metadata_location(metadata)?.to_string();
+    let context =
+        crate::table_catalog::TableSnapshotGraphValidationContext::new(metadata_backend, bucket, namespace, table, &target_entry);
+    crate::table_catalog::validate_table_snapshot_changes(&context, current_metadata, metadata).await
+}
+
 async fn list_tables_response<S>(
     store: &S,
     bucket: &str,
@@ -4205,6 +4272,16 @@ where
     let target_metadata = read_table_metadata_json(metadata_backend, bucket, &metadata_location).await?;
     validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
     validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
+    validate_table_metadata_snapshot_graph(
+        metadata_backend,
+        bucket,
+        namespace,
+        &table_name,
+        &current,
+        Some(&current_metadata),
+        &target_metadata,
+    )
+    .await?;
     let commit_request = crate::table_catalog::TableCommitRequest {
         table_bucket: bucket.to_string(),
         namespace: namespace.public_name(),
@@ -4255,6 +4332,16 @@ where
     let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.new_metadata_location).await?;
     validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
     validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
+    validate_table_metadata_snapshot_graph(
+        metadata_backend,
+        bucket,
+        namespace,
+        &table_name,
+        &current,
+        Some(&current_metadata),
+        &target_metadata,
+    )
+    .await?;
     let result = store.commit_table(request).await.map_err(catalog_store_error)?;
     Ok(commit_table_response_from_result(result, target_metadata))
 }
@@ -4285,7 +4372,21 @@ where
     let previous_metadata_location = table_metadata_location_for_client(bucket, &current.metadata_location);
     let next_metadata = apply_table_commit_updates(current_metadata, &request.updates, &previous_metadata_location)?;
     validate_metadata_table_location_in_bucket(bucket, &next_metadata)?;
-    validate_metadata_matches_current_metadata(&expected_metadata, &next_metadata)?;
+    validate_metadata_identity_matches_current_metadata(&expected_metadata, &next_metadata)?;
+    validate_table_metadata_snapshot_graph_result(
+        metadata_backend,
+        bucket,
+        namespace,
+        &table_name,
+        &current,
+        Some(&expected_metadata),
+        &next_metadata,
+    )
+    .await
+    .map_err(|err| match err {
+        crate::table_catalog::TableCatalogStoreError::Invalid(message) => s3_error!(InvalidRequest, "{}", message),
+        err => catalog_store_error(err),
+    })?;
     validate_table_snapshot_commit_conflicts(
         metadata_backend,
         bucket,
@@ -4296,6 +4397,7 @@ where
         &request.updates,
     )
     .await?;
+    validate_metadata_matches_current_metadata(&expected_metadata, &next_metadata)?;
     let (commit_id, metadata_file_token) = standard_commit_ids(request.commit_id);
     let next_generation = current.generation.saturating_add(1);
     let next_metadata_location = crate::table_catalog::default_table_metadata_file_path(
@@ -4501,6 +4603,16 @@ where
     validate_metadata_table_location_in_bucket(bucket, &next_metadata)?;
     let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
         .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
+    validate_table_metadata_snapshot_graph(
+        metadata_backend,
+        bucket,
+        namespace,
+        &table_name,
+        &current,
+        Some(&current_metadata),
+        &next_metadata,
+    )
+    .await?;
     let (commit_id, metadata_file_token) = standard_commit_ids(None);
     let next_generation = current.generation.saturating_add(1);
     let next_metadata_location = crate::table_catalog::default_table_metadata_file_path(
@@ -4992,6 +5104,8 @@ where
     let target_metadata = read_table_metadata_json(metadata_backend, bucket, &request.metadata_location).await?;
     validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
     let external_table_uuid = validate_external_catalog_metadata_uuid(request.external_table_uuid.as_deref(), &target_metadata)?;
+    let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
+        .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
 
     let (action, table_response) = if let Some(current) = store
         .load_table(bucket, &namespace.public_name(), table)
@@ -5009,6 +5123,16 @@ where
         let current_metadata = read_table_metadata_json(metadata_backend, bucket, &current.metadata_location).await?;
         validate_metadata_table_location_in_bucket(bucket, &current_metadata)?;
         validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
+        validate_table_metadata_snapshot_graph(
+            metadata_backend,
+            bucket,
+            namespace,
+            &table_name,
+            &current,
+            Some(&current_metadata),
+            &target_metadata,
+        )
+        .await?;
         let result = store
             .commit_table(crate::table_catalog::TableCommitRequest {
                 table_bucket: bucket.to_string(),
@@ -5046,6 +5170,8 @@ where
             },
         )?;
         adopt_registered_metadata_identity(&mut entry, &target_metadata)?;
+        validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table_name, &entry, None, &target_metadata)
+            .await?;
         store.register_table(entry.clone()).await.map_err(catalog_store_error)?;
         (
             EXTERNAL_CATALOG_ACTION_REGISTERED.to_string(),
@@ -5085,6 +5211,9 @@ where
         let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?;
         validate_metadata_table_location_in_bucket(bucket, &metadata)?;
         adopt_registered_metadata_identity(&mut entry, &metadata)?;
+        let table_name = crate::table_catalog::IdentifierSegment::parse(table.to_string())
+            .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?;
+        validate_table_metadata_snapshot_graph(metadata_backend, bucket, namespace, &table_name, &entry, None, &metadata).await?;
         if let Some(existing) = store
             .load_table(bucket, &namespace.public_name(), table)
             .await
@@ -5140,6 +5269,16 @@ where
         let target_metadata = read_table_metadata_json(metadata_backend, bucket, &metadata_location).await?;
         validate_metadata_table_location_in_bucket(bucket, &target_metadata)?;
         validate_metadata_matches_current_metadata(&current_metadata, &target_metadata)?;
+        validate_table_metadata_snapshot_graph(
+            metadata_backend,
+            bucket,
+            namespace,
+            &table_name,
+            &current,
+            None,
+            &target_metadata,
+        )
+        .await?;
         let commit_request = crate::table_catalog::TableCommitRequest {
             table_bucket: bucket.to_string(),
             namespace: namespace.public_name(),

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

@@ -1635,6 +1635,36 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
     assert_eq!(request.name, "events");
 }
 
+#[test]
+fn create_table_request_honors_supported_format_version_property() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events_v1",
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        },
+        "properties": {"format-version": "1"}
+    }))
+    .expect("v1 create table request should parse");
+
+    let (entry, metadata) =
+        table_entry_from_create_table_request("warehouse", &namespace, request).expect("v1 table metadata should be created");
+
+    assert_eq!(entry.format_version, 1);
+    assert_eq!(metadata["format-version"], 1);
+    assert!(!entry.properties.contains_key("format-version"));
+    assert!(metadata["properties"].get("format-version").is_none());
+    assert!(metadata.get("schema").is_some());
+    assert!(metadata.get("partition-spec").is_some());
+    assert!(metadata.get("schemas").is_some());
+    assert_eq!(metadata["current-schema-id"], 0);
+    assert!(metadata.get("partition-specs").is_some());
+    assert!(metadata.get("sort-orders").is_some());
+    assert!(metadata.get("last-sequence-number").is_none());
+}
+
 #[test]
 fn commit_table_request_accepts_standard_iceberg_rest_shape() {
     let request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({
@@ -1676,6 +1706,47 @@ fn standard_commit_ids_generate_metadata_hash_for_non_uuid_client_id() {
     assert_eq!(metadata_file_token, table_catalog_path_hash("commit-1"));
 }
 
+#[test]
+fn format_upgrade_assigns_v1_snapshot_sequences_and_rejects_v3() {
+    let mut metadata = serde_json::json!({
+        "format-version": 1,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id",
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        },
+        "partition-spec": [],
+        "snapshots": [{
+            "snapshot-id": 10,
+            "timestamp-ms": 1,
+            "manifests": ["s3://warehouse/tables/table-id/metadata/manifest-10.avro"],
+            "summary": {"operation": "append"}
+        }]
+    });
+
+    apply_upgrade_format_version_update(
+        &mut metadata,
+        &serde_json::json!({"action": "upgrade-format-version", "format-version": 2}),
+    )
+    .expect("v1 metadata should upgrade to v2");
+    crate::table_catalog::synchronize_table_metadata_version_fields(&mut metadata)
+        .expect("upgraded metadata fields should synchronize");
+
+    assert_eq!(metadata["snapshots"][0]["sequence-number"], 0);
+    crate::table_catalog::validate_supported_table_metadata(&metadata).expect("upgraded metadata should satisfy the v2 contract");
+
+    let error = apply_upgrade_format_version_update(
+        &mut metadata,
+        &serde_json::json!({"action": "upgrade-format-version", "format-version": 3}),
+    )
+    .expect_err("format version 3 is not supported");
+    assert_eq!(error.status_code(), Some(StatusCode::NOT_ACCEPTABLE));
+}
+
 #[tokio::test]
 async fn create_table_response_writes_initial_metadata_for_standard_request() {
     let store = TestTableCatalogStore::default();
@@ -2272,12 +2343,7 @@ async fn standard_commit_accepts_legacy_catalog_uuid_when_current_metadata_match
         .put_json(
             "warehouse",
             current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "metadata-table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "properties": {}
-            }),
+            test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -2340,26 +2406,13 @@ async fn metadata_location_api_accepts_legacy_catalog_uuid_when_target_matches_c
         .put_json(
             "warehouse",
             current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "metadata-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     let next_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json";
-    metadata_backend
-        .put_json(
-            "warehouse",
-            next_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "metadata-table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "last-sequence-number": 2
-            }),
-        )
-        .await;
+    let mut next_metadata = test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id");
+    next_metadata["last-sequence-number"] = serde_json::Value::from(2);
+    metadata_backend.put_json("warehouse", next_location, next_metadata).await;
 
     let updated = update_table_metadata_location_response(
         &store,
@@ -2567,11 +2620,16 @@ async fn table_metadata_maintenance_helper_commits_snapshot_expiration() {
                 "last-sequence-number": 2,
                 "last-updated-ms": 2000,
                 "last-column-id": 1,
-                "schemas": [],
+                "schemas": [{
+                    "type": "struct",
+                    "schema-id": 0,
+                    "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+                }],
                 "current-schema-id": 0,
-                "partition-specs": [],
+                "partition-specs": [{"spec-id": 0, "fields": []}],
                 "default-spec-id": 0,
-                "sort-orders": [],
+                "last-partition-id": 999,
+                "sort-orders": [{"order-id": 0, "fields": []}],
                 "default-sort-order-id": 0,
                 "current-snapshot-id": 20,
                 "metadata-log": [],
@@ -2588,13 +2646,17 @@ async fn table_metadata_maintenance_helper_commits_snapshot_expiration() {
                 "snapshots": [
                     {
                         "snapshot-id": 10,
+                        "sequence-number": 1,
                         "timestamp-ms": 1000,
-                        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro"
+                        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro",
+                        "summary": {"operation": "append"}
                     },
                     {
                         "snapshot-id": 20,
+                        "sequence-number": 2,
                         "timestamp-ms": 2000,
-                        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-20.avro"
+                        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-20.avro",
+                        "summary": {"operation": "append"}
                     }
                 ],
                 "refs": {
@@ -3038,11 +3100,7 @@ async fn external_catalog_bridge_sync_registers_missing_table_from_snapshot() {
         .put_json(
             bucket,
             &metadata_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -3103,30 +3161,12 @@ async fn external_catalog_bridge_sync_commits_existing_table_pointer() {
     let current_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
     let next_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
     seed_object_table_for_metadata_maintenance(&store, &backend, bucket, &namespace, &table, current_location.clone()).await;
-    backend
-        .put_json(
-            bucket,
-            &current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "last-sequence-number": 1
-            }),
-        )
-        .await;
-    backend
-        .put_json(
-            bucket,
-            &next_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "last-sequence-number": 2
-            }),
-        )
-        .await;
+    let mut current_metadata = test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id");
+    current_metadata["last-sequence-number"] = serde_json::Value::from(1);
+    backend.put_json(bucket, &current_location, current_metadata).await;
+    let mut next_metadata = test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id");
+    next_metadata["last-sequence-number"] = serde_json::Value::from(2);
+    backend.put_json(bucket, &next_location, next_metadata).await;
 
     let synced = sync_external_catalog_bridge_response(
         &store,
@@ -3181,22 +3221,14 @@ async fn external_catalog_bridge_sync_conflicts_leave_pointer_unchanged() {
         .put_json(
             bucket,
             &current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     backend
         .put_json(
             bucket,
             &next_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "different-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("different-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -5283,6 +5315,16 @@ impl TestTableCatalogObjectBackend {
         self.put_json_with_mod_time(bucket, object, value, None).await;
     }
 
+    async fn put_gzip_json(&self, bucket: &str, object: &str, value: serde_json::Value) {
+        use std::io::Write;
+
+        let data = serde_json::to_vec(&value).expect("metadata JSON should serialize");
+        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+        encoder.write_all(&data).expect("metadata JSON should compress");
+        self.put_bytes(bucket, object, encoder.finish().expect("metadata gzip stream should finish"))
+            .await;
+    }
+
     async fn put_json_with_mod_time(
         &self,
         bucket: &str,
@@ -5302,6 +5344,33 @@ impl TestTableCatalogObjectBackend {
     }
 }
 
+fn test_table_metadata_json(table_uuid: &str, location: &str) -> serde_json::Value {
+    serde_json::json!({
+        "format-version": 2,
+        "table-uuid": table_uuid,
+        "location": location,
+        "last-sequence-number": 0,
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schemas": [{
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        }],
+        "current-schema-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "default-spec-id": 0,
+        "last-partition-id": 999,
+        "sort-orders": [{"order-id": 0, "fields": []}],
+        "default-sort-order-id": 0,
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": [],
+        "refs": {}
+    })
+}
+
 fn test_snapshot_object_key(bucket: &str, location: &str) -> String {
     crate::table_catalog::table_catalog_object_key_from_location(bucket, location)
         .expect("test snapshot object location should be valid")
@@ -5310,12 +5379,20 @@ fn test_snapshot_object_key(bucket: &str, location: &str) -> String {
 fn test_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, sequence_number, snapshot_id))
+        .map(|manifest_path| (*manifest_path, 0, sequence_number, snapshot_id))
         .collect::<Vec<_>>();
-    test_manifest_list_avro_entries(&manifests)
+    test_manifest_list_avro_entries_with_partition_specs(&manifests)
 }
 
 fn test_manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
+    let manifests = manifests
+        .iter()
+        .map(|(manifest_path, sequence_number, snapshot_id)| (*manifest_path, 0, *sequence_number, *snapshot_id))
+        .collect::<Vec<_>>();
+    test_manifest_list_avro_entries_with_partition_specs(&manifests)
+}
+
+fn test_manifest_list_avro_entries_with_partition_specs(manifests: &[(&str, i32, i64, i64)]) -> Vec<u8> {
     let schema = apache_avro::Schema::parse_str(
         r#"
             {
@@ -5323,23 +5400,43 @@ fn test_manifest_list_avro_entries(manifests: &[(&str, i64, i64)]) -> Vec<u8> {
               "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": "added_snapshot_id", "type": "long"}
+                {"name": "min_sequence_number", "type": "long"},
+                {"name": "added_snapshot_id", "type": "long"},
+                {"name": "added_files_count", "type": "int"},
+                {"name": "existing_files_count", "type": "int"},
+                {"name": "deleted_files_count", "type": "int"},
+                {"name": "added_rows_count", "type": "long"},
+                {"name": "existing_rows_count", "type": "long"},
+                {"name": "deleted_rows_count", "type": "long"}
               ]
             }
             "#,
     )
     .expect("manifest list avro schema should parse");
     let mut writer = apache_avro::Writer::new(&schema, Vec::new());
-    for (manifest_path, sequence_number, snapshot_id) in manifests {
+    for (manifest_path, partition_spec_id, sequence_number, snapshot_id) in manifests {
         writer
             .append(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)),
+                ("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::Int(1)),
+                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
+                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
             ]))
             .expect("manifest list record should append");
     }
@@ -5547,11 +5644,14 @@ async fn seed_test_manifest_data_files(
     }
 }
 
-async fn create_standard_events_table(
-    store: &TestTableCatalogStore,
+async fn create_standard_events_table<S>(
+    store: &S,
     metadata_backend: &TestTableCatalogObjectBackend,
     namespace: &crate::table_catalog::Namespace,
-) -> RestLoadTableResponse {
+) -> RestLoadTableResponse
+where
+    S: crate::table_catalog::TableCatalogStore + ?Sized,
+{
     ensure_table_bucket_entry(store, "warehouse", true)
         .await
         .expect("table bucket entry should be seeded");
@@ -6217,11 +6317,7 @@ async fn table_helpers_call_catalog_store() {
         .put_json(
             "warehouse",
             metadata_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     let register = register_table_response(
@@ -6262,17 +6358,10 @@ async fn table_helpers_call_catalog_store() {
         .expect("table should exist");
     let next_metadata_location =
         ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json";
+    let mut next_metadata = test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id");
+    next_metadata["last-sequence-number"] = serde_json::Value::from(2);
     metadata_backend
-        .put_json(
-            "warehouse",
-            next_metadata_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "last-sequence-number": 2
-            }),
-        )
+        .put_json("warehouse", next_metadata_location, next_metadata)
         .await;
     let commit = commit_table_response(
         &store,
@@ -6344,11 +6433,7 @@ async fn register_table_response_adopts_metadata_table_uuid() {
         .put_json(
             "warehouse",
             metadata_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "metadata-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("metadata-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -6467,11 +6552,7 @@ async fn metadata_location_api_loads_and_updates_current_pointer() {
         .put_json(
             "warehouse",
             current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": table_uuid,
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json(&table_uuid, "s3://warehouse/tables/table-id"),
         )
         .await;
     let current = get_table_metadata_location_response(&store, "warehouse", &namespace, "events")
@@ -6486,11 +6567,7 @@ async fn metadata_location_api_loads_and_updates_current_pointer() {
         .put_json(
             "warehouse",
             next_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": table_uuid,
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json(&table_uuid, "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -6515,6 +6592,156 @@ async fn metadata_location_api_loads_and_updates_current_pointer() {
     assert_ne!(updated.version_token, current.version_token);
 }
 
+#[tokio::test]
+async fn metadata_location_api_accepts_gzip_table_metadata() {
+    let store = TestTableCatalogStore::default();
+    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
+    let current = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should exist");
+    let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
+    let target_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json.gz");
+    metadata_backend
+        .put_gzip_json("warehouse", &target_location, created.metadata)
+        .await;
+
+    let updated = update_table_metadata_location_response(
+        &store,
+        &metadata_backend,
+        "warehouse",
+        &namespace,
+        "events",
+        UpdateTableMetadataLocationRequest {
+            metadata_location: table_metadata_location_for_client("warehouse", &target_location),
+            version_token: current.version_token,
+            commit_id: Some("gzip-metadata".to_string()),
+            idempotency_key: None,
+        },
+    )
+    .await
+    .expect("gzip table metadata should commit");
+
+    assert_eq!(
+        updated.metadata_location,
+        table_metadata_location_for_client("warehouse", &target_location)
+    );
+}
+
+#[tokio::test]
+async fn metadata_location_api_validates_snapshot_graph_before_commit() {
+    let store = TestTableCatalogStore::default();
+    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
+    let current = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should exist");
+    let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
+    let target_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002-graph.metadata.json");
+    let manifest_list = format!("{}/metadata/snap-10.avro", current.warehouse_location);
+    let data_file = format!("{}/data/part-10.parquet", current.warehouse_location);
+    let mut target_metadata = created.metadata;
+    target_metadata["last-sequence-number"] = serde_json::Value::from(1);
+    target_metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 1,
+        "timestamp-ms": 10,
+        "manifest-list": manifest_list,
+        "summary": {"operation": "append"}
+    }]);
+    target_metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    target_metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    metadata_backend
+        .put_json("warehouse", &target_location, target_metadata)
+        .await;
+    let request = || UpdateTableMetadataLocationRequest {
+        metadata_location: table_metadata_location_for_client("warehouse", &target_location),
+        version_token: current.version_token.clone(),
+        commit_id: Some("graph-commit".to_string()),
+        idempotency_key: Some("graph-replay".to_string()),
+    };
+
+    let error = update_table_metadata_location_response(&store, &metadata_backend, "warehouse", &namespace, "events", request())
+        .await
+        .expect_err("missing manifest-list must fail before pointer publication");
+    assert_eq!(error.message(), Some("snapshot manifest-list object is missing"));
+    let unchanged = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should remain present");
+    assert_eq!(unchanged.metadata_location, current.metadata_location);
+
+    seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await;
+    update_table_metadata_location_response(&store, &metadata_backend, "warehouse", &namespace, "events", request())
+        .await
+        .expect("complete snapshot graph should commit");
+}
+
+#[tokio::test]
+async fn metadata_location_api_validates_relocated_snapshot_graph_under_target_warehouse() {
+    let metadata_backend = TestTableCatalogObjectBackend::default();
+    let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone());
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let created = create_standard_events_table(&store, &metadata_backend, &namespace).await;
+    let current = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should exist");
+    let table = crate::table_catalog::IdentifierSegment::parse("events").expect("table should parse");
+    let target_location =
+        crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002-relocated.metadata.json");
+    let target_warehouse = "s3://warehouse/tables/relocated-table-id";
+    let manifest_list = format!("{target_warehouse}/metadata/snap-10.avro");
+    let data_file = format!("{target_warehouse}/data/part-10.parquet");
+    let mut target_metadata = created.metadata;
+    target_metadata["location"] = serde_json::Value::String(target_warehouse.to_string());
+    target_metadata["last-sequence-number"] = serde_json::Value::from(1);
+    target_metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 1,
+        "timestamp-ms": 10,
+        "manifest-list": manifest_list,
+        "summary": {"operation": "append"}
+    }]);
+    target_metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    target_metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    metadata_backend
+        .put_json("warehouse", &target_location, target_metadata)
+        .await;
+    seed_test_snapshot_manifest(&metadata_backend, "warehouse", &manifest_list, 10, 1, &[(&data_file, 0, 1, 10, 1)]).await;
+
+    update_table_metadata_location_response(
+        &store,
+        &metadata_backend,
+        "warehouse",
+        &namespace,
+        "events",
+        UpdateTableMetadataLocationRequest {
+            metadata_location: table_metadata_location_for_client("warehouse", &target_location),
+            version_token: current.version_token,
+            commit_id: Some("relocate-graph".to_string()),
+            idempotency_key: None,
+        },
+    )
+    .await
+    .expect("snapshot objects under the target warehouse should validate");
+
+    let relocated = store
+        .load_table("warehouse", "analytics", "events")
+        .await
+        .expect("table lookup should succeed")
+        .expect("table should exist");
+    assert_eq!(relocated.warehouse_location, target_warehouse);
+}
+
 #[tokio::test]
 async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() {
     let store = TestTableCatalogStore::default();
@@ -6535,21 +6762,25 @@ async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() {
     .await
     .expect("namespace should be created");
     let current_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json";
-    store
-        .register_table(
-            table_entry_from_register_request(
-                "warehouse",
-                &namespace,
-                RegisterTableRequest {
-                    name: "events".to_string(),
-                    metadata_location: current_location.to_string(),
-                    overwrite: false,
-                },
-            )
-            .expect("table entry should build"),
+    let entry = table_entry_from_register_request(
+        "warehouse",
+        &namespace,
+        RegisterTableRequest {
+            name: "events".to_string(),
+            metadata_location: current_location.to_string(),
+            overwrite: false,
+        },
+    )
+    .expect("table entry should build");
+    let table_uuid = entry.table_uuid.clone();
+    store.register_table(entry).await.expect("table should register");
+    metadata_backend
+        .put_json(
+            "warehouse",
+            current_location,
+            test_table_metadata_json(&table_uuid, "s3://warehouse/tables/table-id"),
         )
-        .await
-        .expect("table should register");
+        .await;
     let current = get_table_metadata_location_response(&store, "warehouse", &namespace, "events")
         .await
         .expect("metadata location should load");
@@ -6558,11 +6789,7 @@ async fn metadata_location_api_rejects_invalid_target_metadata_before_commit() {
         .put_json(
             "warehouse",
             invalid_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://other-warehouse/tables/table-id"
-            }),
+            test_table_metadata_json(&table_uuid, "s3://other-warehouse/tables/table-id"),
         )
         .await;
 
@@ -6617,11 +6844,7 @@ async fn metadata_location_api_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             "warehouse",
             current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     register_table_response(
@@ -6646,11 +6869,7 @@ async fn metadata_location_api_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             "warehouse",
             mismatched_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "other-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("other-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -6707,11 +6926,7 @@ async fn catalog_import_and_rollback_use_register_and_commit_paths() {
         .put_json(
             bucket,
             &imported_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -6757,18 +6972,9 @@ async fn catalog_import_and_rollback_use_register_and_commit_paths() {
     );
 
     let rollback_location = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00002.metadata.json");
-    backend
-        .put_json(
-            bucket,
-            &rollback_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id",
-                "last-sequence-number": 2
-            }),
-        )
-        .await;
+    let mut rollback_metadata = test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id");
+    rollback_metadata["last-sequence-number"] = serde_json::Value::from(2);
+    backend.put_json(bucket, &rollback_location, rollback_metadata).await;
     let rollback = rollback_table_response(
         &store,
         &backend,
@@ -6815,11 +7021,7 @@ async fn rollback_rejects_invalid_target_metadata_before_commit() {
         .put_json(
             bucket,
             &current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     catalog_import_response(
@@ -6908,11 +7110,7 @@ async fn rollback_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             bucket,
             &current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     catalog_import_response(
@@ -6940,11 +7138,7 @@ async fn rollback_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             bucket,
             &mismatched_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "other-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("other-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 
@@ -6998,11 +7192,7 @@ async fn legacy_commit_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             "warehouse",
             current_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
     register_table_response(
@@ -7029,11 +7219,7 @@ async fn legacy_commit_rejects_mismatched_table_uuid_before_commit() {
         .put_json(
             "warehouse",
             mismatched_location,
-            serde_json::json!({
-                "format-version": 2,
-                "table-uuid": "other-table-uuid",
-                "location": "s3://warehouse/tables/table-id"
-            }),
+            test_table_metadata_json("other-table-uuid", "s3://warehouse/tables/table-id"),
         )
         .await;
 

rustfs/src/table_catalog/iceberg/manifest.rs

@@ -12,11 +12,15 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::io::Read;
+
 use super::super::*;
 
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct ManifestDataFileReference {
     pub location: String,
+    pub format_version: u16,
+    pub content_id: Option<i32>,
     pub content: ManifestDataFileContent,
     pub object_kind: TableMetadataMaintenanceObjectKind,
     pub entry_status: Option<i32>,
@@ -39,9 +43,29 @@ pub(crate) enum ManifestDataFileContent {
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct ManifestListReference {
     pub manifest_path: String,
+    pub format_version: u16,
+    pub manifest_length: Option<u64>,
     pub partition_spec_id: Option<i32>,
+    pub content: Option<i32>,
     pub sequence_number: Option<i64>,
+    pub min_sequence_number: Option<i64>,
     pub added_snapshot_id: Option<i64>,
+    pub added_files_count: Option<u64>,
+    pub existing_files_count: Option<u64>,
+    pub deleted_files_count: Option<u64>,
+    pub added_rows_count: Option<u64>,
+    pub existing_rows_count: Option<u64>,
+    pub deleted_rows_count: Option<u64>,
+}
+
+pub(crate) struct DecodedManifestList {
+    pub references: Vec<ManifestListReference>,
+    pub decoded_size: usize,
+}
+
+pub(crate) struct DecodedManifest {
+    pub references: Vec<ManifestDataFileReference>,
+    pub decoded_size: usize,
 }
 
 pub(crate) fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<String>> {
@@ -54,10 +78,27 @@ pub(crate) fn manifest_paths_from_manifest_list_avro(data: &[u8]) -> TableCatalo
 pub(crate) fn manifest_list_references_from_manifest_list_avro(
     data: &[u8],
 ) -> TableCatalogStoreResult<Vec<ManifestListReference>> {
+    Ok(decode_manifest_list_avro(data)?.references)
+}
+
+pub(crate) fn decode_manifest_list_avro(data: &[u8]) -> TableCatalogStoreResult<DecodedManifestList> {
+    if data.len() > TABLE_MANIFEST_AVRO_MAX_SIZE {
+        return Err(TableCatalogStoreError::Invalid(format!(
+            "manifest list exceeds the maximum size of {TABLE_MANIFEST_AVRO_MAX_SIZE} bytes"
+        )));
+    }
+    let decoded_size = validate_avro_container(data)?;
     let reader = apache_avro::Reader::new(data)
         .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")?;
     let mut manifest_paths = Vec::new();
     for value in reader {
+        if manifest_paths.len() >= TABLE_MANIFEST_AVRO_MAX_RECORDS {
+            return Err(TableCatalogStoreError::Invalid(format!(
+                "manifest list exceeds the maximum record count of {TABLE_MANIFEST_AVRO_MAX_RECORDS}"
+            )));
+        }
         let value =
             value.map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest list record: {err}")))?;
         let manifest_path = avro_record_field(&value, "manifest_path")
@@ -65,12 +106,45 @@ pub(crate) fn manifest_list_references_from_manifest_list_avro(
             .ok_or_else(|| TableCatalogStoreError::Invalid("manifest list entry missing manifest_path".to_string()))?;
         manifest_paths.push(ManifestListReference {
             manifest_path: manifest_path.to_string(),
+            format_version,
+            manifest_length: avro_record_field(&value, "manifest_length")
+                .and_then(avro_i64_value)
+                .and_then(|value| u64::try_from(value).ok()),
             partition_spec_id: avro_record_field(&value, "partition_spec_id").and_then(avro_i32_value),
+            content: avro_record_field(&value, "content").and_then(avro_i32_value),
             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()),
         });
     }
-    Ok(manifest_paths)
+    Ok(DecodedManifestList {
+        references: manifest_paths,
+        decoded_size,
+    })
+}
+
+pub(crate) async fn decode_manifest_list_avro_async(data: Vec<u8>) -> TableCatalogStoreResult<DecodedManifestList> {
+    tokio::task::spawn_blocking(move || decode_manifest_list_avro(&data))
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("manifest-list parser task failed: {err}")))?
 }
 
 pub(crate) fn file_references_from_manifest_avro(
@@ -83,27 +157,55 @@ pub(crate) fn file_references_from_manifest_avro(
 }
 
 pub(crate) fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<Vec<ManifestDataFileReference>> {
+    Ok(decode_manifest_avro(data)?.references)
+}
+
+pub(crate) fn decode_manifest_avro(data: &[u8]) -> TableCatalogStoreResult<DecodedManifest> {
+    if data.len() > TABLE_MANIFEST_AVRO_MAX_SIZE {
+        return Err(TableCatalogStoreError::Invalid(format!(
+            "manifest exceeds the maximum size of {TABLE_MANIFEST_AVRO_MAX_SIZE} bytes"
+        )));
+    }
+    let decoded_size = validate_avro_container(data)?;
     let reader = apache_avro::Reader::new(data)
         .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 mut files = Vec::new();
     for value in reader {
+        if files.len() >= TABLE_MANIFEST_AVRO_MAX_RECORDS {
+            return Err(TableCatalogStoreError::Invalid(format!(
+                "manifest exceeds the maximum record count of {TABLE_MANIFEST_AVRO_MAX_RECORDS}"
+            )));
+        }
         let value = value.map_err(|err| TableCatalogStoreError::Invalid(format!("failed to read manifest record: {err}")))?;
         let data_file = avro_record_field(&value, "data_file")
             .ok_or_else(|| TableCatalogStoreError::Invalid("manifest entry missing data_file".to_string()))?;
         let file_path = avro_record_field(data_file, "file_path")
             .and_then(avro_string_value)
             .ok_or_else(|| TableCatalogStoreError::Invalid("manifest data file missing file_path".to_string()))?;
-        let content = avro_record_field(data_file, "content")
-            .and_then(avro_i32_value)
-            .ok_or_else(|| TableCatalogStoreError::Invalid("manifest data file missing content".to_string()))?;
-        let (content, object_kind) = match content {
+        let content_id = match avro_record_field(data_file, "content") {
+            Some(content) => Some(
+                avro_i32_value(content)
+                    .ok_or_else(|| TableCatalogStoreError::Invalid("manifest data file content must be an int".to_string()))?,
+            ),
+            None => None,
+        };
+        let content_value = content_id.unwrap_or(0);
+        let (content, object_kind) = match content_value {
             0 => (ManifestDataFileContent::Data, TableMetadataMaintenanceObjectKind::DataFile),
             1 => (ManifestDataFileContent::PositionDelete, TableMetadataMaintenanceObjectKind::DeleteFile),
             2 => (ManifestDataFileContent::EqualityDelete, TableMetadataMaintenanceObjectKind::DeleteFile),
-            _ => continue,
+            _ => {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "manifest data file has unsupported content value {content_value}"
+                )));
+            }
         };
         files.push(ManifestDataFileReference {
             location: file_path.to_string(),
+            format_version,
+            content_id,
             content,
             object_kind,
             entry_status: avro_record_field(&value, "status").and_then(avro_i32_value),
@@ -122,7 +224,188 @@ pub(crate) fn data_file_references_from_manifest_avro(data: &[u8]) -> TableCatal
             sort_order_id: avro_record_field(data_file, "sort_order_id").and_then(avro_i32_value),
         });
     }
-    Ok(files)
+    Ok(DecodedManifest {
+        references: files,
+        decoded_size,
+    })
+}
+
+pub(crate) async fn decode_manifest_avro_async(data: Vec<u8>) -> TableCatalogStoreResult<DecodedManifest> {
+    tokio::task::spawn_blocking(move || decode_manifest_avro(&data))
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("manifest parser task failed: {err}")))?
+}
+
+#[derive(Clone, Copy)]
+enum AvroContainerCodec {
+    Null,
+    Deflate,
+}
+
+fn validate_avro_container(data: &[u8]) -> TableCatalogStoreResult<usize> {
+    if !data.starts_with(b"Obj\x01") {
+        return Err(TableCatalogStoreError::Invalid("Avro container has invalid header magic".to_string()));
+    }
+    let mut offset = 4;
+    let mut entry_count = 0usize;
+    let mut codec = None;
+    loop {
+        let block_count = read_avro_long(data, &mut offset)?;
+        if block_count == 0 {
+            break;
+        }
+        let (block_count, expected_end) = if block_count < 0 {
+            let block_count = block_count
+                .checked_neg()
+                .and_then(|count| usize::try_from(count).ok())
+                .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header block count is invalid".to_string()))?;
+            let block_size = read_avro_long(data, &mut offset)?;
+            let block_size = usize::try_from(block_size)
+                .map_err(|_| TableCatalogStoreError::Invalid("Avro header block size is invalid".to_string()))?;
+            let expected_end = offset
+                .checked_add(block_size)
+                .filter(|end| *end <= data.len())
+                .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header block exceeds the object size".to_string()))?;
+            (block_count, Some(expected_end))
+        } else {
+            let block_count = usize::try_from(block_count)
+                .map_err(|_| TableCatalogStoreError::Invalid("Avro header block count is invalid".to_string()))?;
+            (block_count, None)
+        };
+        entry_count = entry_count
+            .checked_add(block_count)
+            .filter(|count| *count <= TABLE_MANIFEST_AVRO_MAX_HEADER_ENTRIES)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header has too many metadata entries".to_string()))?;
+        for _ in 0..block_count {
+            let key = read_avro_bytes(data, &mut offset)?;
+            let value = read_avro_bytes(data, &mut offset)?;
+            if key == b"avro.codec" {
+                codec = Some(value);
+            }
+        }
+        if expected_end.is_some_and(|expected_end| offset != expected_end) {
+            return Err(TableCatalogStoreError::Invalid(
+                "Avro header block size does not match its contents".to_string(),
+            ));
+        }
+    }
+    let marker_end = offset
+        .checked_add(16)
+        .filter(|end| *end <= data.len())
+        .ok_or_else(|| TableCatalogStoreError::Invalid("Avro container is missing its sync marker".to_string()))?;
+    let sync_marker: [u8; 16] = data[offset..marker_end]
+        .try_into()
+        .map_err(|_| TableCatalogStoreError::Invalid("Avro container has an invalid sync marker".to_string()))?;
+    offset = marker_end;
+    let codec = match codec.unwrap_or(b"null") {
+        b"null" => AvroContainerCodec::Null,
+        b"deflate" => AvroContainerCodec::Deflate,
+        codec => {
+            return Err(TableCatalogStoreError::Unsupported(format!(
+                "Avro codec {} is not supported for table commit validation",
+                String::from_utf8_lossy(codec)
+            )));
+        }
+    };
+
+    let mut record_count = 0usize;
+    let mut decoded_size = 0usize;
+    while offset < data.len() {
+        let block_count = read_avro_long(data, &mut offset)?;
+        let block_count = usize::try_from(block_count)
+            .ok()
+            .filter(|count| *count > 0)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro data block count is invalid".to_string()))?;
+        record_count = record_count
+            .checked_add(block_count)
+            .filter(|count| *count <= TABLE_MANIFEST_AVRO_MAX_RECORDS)
+            .ok_or_else(|| {
+                TableCatalogStoreError::Invalid(format!(
+                    "Avro container exceeds the maximum record count of {TABLE_MANIFEST_AVRO_MAX_RECORDS}"
+                ))
+            })?;
+        let block_size = read_avro_long(data, &mut offset)?;
+        let block_size = usize::try_from(block_size)
+            .map_err(|_| TableCatalogStoreError::Invalid("Avro data block size is invalid".to_string()))?;
+        let block_end = offset
+            .checked_add(block_size)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro data block exceeds the object size".to_string()))?;
+        let block_marker_end = block_end
+            .checked_add(16)
+            .filter(|marker_end| *marker_end <= data.len())
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro data block exceeds the object size".to_string()))?;
+        let remaining_decoded_size = TABLE_MANIFEST_AVRO_MAX_DECODED_SIZE
+            .checked_sub(decoded_size)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro decoded data exceeds the commit limit".to_string()))?;
+        let block_decoded_size = avro_block_decoded_size(codec, &data[offset..block_end], remaining_decoded_size)?;
+        decoded_size = decoded_size
+            .checked_add(block_decoded_size)
+            .filter(|size| *size <= TABLE_MANIFEST_AVRO_MAX_DECODED_SIZE)
+            .ok_or_else(|| {
+                TableCatalogStoreError::Invalid(format!(
+                    "Avro decoded data exceeds the maximum size of {TABLE_MANIFEST_AVRO_MAX_DECODED_SIZE} bytes"
+                ))
+            })?;
+        if data[block_end..block_marker_end] != sync_marker {
+            return Err(TableCatalogStoreError::Invalid("Avro data block has an invalid sync marker".to_string()));
+        }
+        offset = block_marker_end;
+    }
+    Ok(decoded_size)
+}
+
+fn avro_block_decoded_size(codec: AvroContainerCodec, block: &[u8], remaining_size: usize) -> TableCatalogStoreResult<usize> {
+    match codec {
+        AvroContainerCodec::Null => Ok(block.len()),
+        AvroContainerCodec::Deflate => {
+            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 decoder = flate2::read::DeflateDecoder::new(block);
+            let decoded_size = std::io::copy(&mut decoder.take(limit), &mut std::io::sink())
+                .map_err(|err| TableCatalogStoreError::Invalid(format!("failed to decompress Avro deflate block: {err}")))?;
+            usize::try_from(decoded_size)
+                .map_err(|_| TableCatalogStoreError::Invalid("Avro decoded data size is invalid".to_string()))
+        }
+    }
+}
+
+fn read_avro_long(data: &[u8], offset: &mut usize) -> TableCatalogStoreResult<i64> {
+    let mut raw = 0u64;
+    for shift in (0..=63).step_by(7) {
+        let byte = *data
+            .get(*offset)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header is truncated".to_string()))?;
+        *offset = offset
+            .checked_add(1)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header offset overflowed".to_string()))?;
+        if shift == 63 && byte & 0xfe != 0 {
+            return Err(TableCatalogStoreError::Invalid("Avro long is out of range".to_string()));
+        }
+        raw |= u64::from(byte & 0x7f) << shift;
+        if byte & 0x80 == 0 {
+            let magnitude =
+                i64::try_from(raw >> 1).map_err(|_| TableCatalogStoreError::Invalid("Avro long is out of range".to_string()))?;
+            let sign = if raw & 1 == 0 { 0 } else { -1 };
+            return Ok(magnitude ^ sign);
+        }
+    }
+    Err(TableCatalogStoreError::Invalid("Avro long is out of range".to_string()))
+}
+
+fn read_avro_bytes<'a>(data: &'a [u8], offset: &mut usize) -> TableCatalogStoreResult<&'a [u8]> {
+    let length = read_avro_long(data, offset)?;
+    let length = usize::try_from(length)
+        .map_err(|_| TableCatalogStoreError::Invalid("Avro header byte string length is invalid".to_string()))?;
+    let end = offset
+        .checked_add(length)
+        .filter(|end| *end <= data.len())
+        .ok_or_else(|| TableCatalogStoreError::Invalid("Avro header byte string is truncated".to_string()))?;
+    let value = &data[*offset..end];
+    *offset = end;
+    Ok(value)
 }
 
 fn avro_record_field<'a>(value: &'a apache_avro::types::Value, name: &str) -> Option<&'a apache_avro::types::Value> {
@@ -135,6 +418,20 @@ fn avro_record_field<'a>(value: &'a apache_avro::types::Value, name: &str) -> Op
         .find_map(|(field_name, field_value)| (field_name == name).then_some(avro_non_union_value(field_value)))
 }
 
+fn avro_record_format_version(schema: &apache_avro::Schema, v2_fields: &[&str], label: &str) -> TableCatalogStoreResult<u16> {
+    let apache_avro::Schema::Record(record) = schema else {
+        return Err(TableCatalogStoreError::Invalid(format!("{label} Avro schema must be a record")));
+    };
+    let present = v2_fields.iter().filter(|field| record.lookup.contains_key(**field)).count();
+    match present {
+        0 => Ok(1),
+        count if count == v2_fields.len() => Ok(2),
+        _ => Err(TableCatalogStoreError::Invalid(format!(
+            "{label} Avro schema contains an incomplete set of Iceberg v2 fields"
+        ))),
+    }
+}
+
 fn avro_record_value_fields(value: &apache_avro::types::Value) -> Option<Vec<(String, apache_avro::types::Value)>> {
     let value = avro_non_union_value(value);
     let apache_avro::types::Value::Record(fields) = value else {

rustfs/src/table_catalog/iceberg/metadata.rs

@@ -14,8 +14,30 @@
 
 use super::super::*;
 
+pub(crate) async fn read_table_metadata_value<B>(
+    backend: &B,
+    table_bucket: &str,
+    metadata_location: &str,
+) -> TableCatalogStoreResult<Option<serde_json::Value>>
+where
+    B: TableCatalogObjectBackend,
+{
+    let Some(object) = backend
+        .read_object_limited(table_bucket, metadata_location, TABLE_METADATA_JSON_MAX_SIZE)
+        .await?
+    else {
+        return Ok(None);
+    };
+    let metadata_location = metadata_location.to_string();
+    tokio::task::spawn_blocking(move || decode_table_metadata_json(&metadata_location, &object.data))
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))?
+        .map(Some)
+}
+
 pub(crate) fn metadata_log_locations(
     current_metadata: &serde_json::Value,
+    table_bucket: &str,
     namespace: &Namespace,
     table: &IdentifierSegment,
 ) -> BTreeSet<String> {
@@ -25,11 +47,15 @@ pub(crate) fn metadata_log_locations(
     };
 
     for entry in metadata_log {
-        let Some(metadata_location) = entry.get("metadata-file").and_then(serde_json::Value::as_str) else {
+        let Some(metadata_location) = entry
+            .get("metadata-file")
+            .and_then(serde_json::Value::as_str)
+            .and_then(|location| table_catalog_object_key_from_location(table_bucket, location))
+        else {
             continue;
         };
-        if is_valid_table_metadata_location(namespace, table, metadata_location) {
-            locations.insert(metadata_location.to_string());
+        if is_valid_table_metadata_location(namespace, table, &metadata_location) {
+            locations.insert(metadata_location);
         }
     }
 
@@ -57,14 +83,15 @@ where
         if !is_valid_table_metadata_location(namespace, table, metadata_location) {
             continue;
         }
-        let Some(metadata_object) = backend.read_object(table_bucket, metadata_location).await? else {
-            continue;
-        };
-        let Ok(metadata) = serde_json::from_slice::<serde_json::Value>(&metadata_object.data) else {
-            continue;
-        };
-        if metadata_contains_protected_snapshot_ref(&metadata, &protected_snapshot_ids) {
-            retained.insert(metadata_location.clone());
+        match read_table_metadata_value(backend, table_bucket, metadata_location).await {
+            Ok(Some(metadata)) if metadata_contains_protected_snapshot_ref(&metadata, &protected_snapshot_ids) => {
+                retained.insert(metadata_location.clone());
+            }
+            Ok(Some(_)) | Ok(None) => {}
+            Err(TableCatalogStoreError::Invalid(_)) => {
+                retained.insert(metadata_location.clone());
+            }
+            Err(err) => return Err(err),
         }
     }
     Ok(retained)

rustfs/src/table_catalog/identifier.rs

@@ -289,7 +289,7 @@ pub(crate) fn is_valid_view_metadata_location(namespace: &Namespace, view: &Iden
 pub(crate) fn is_valid_table_metadata_file_name(metadata_file_name: &str) -> bool {
     if metadata_file_name.is_empty()
         || metadata_file_name.len() > TABLE_METADATA_FILE_NAME_MAX_LEN
-        || !metadata_file_name.ends_with(".json")
+        || !(metadata_file_name.ends_with(".json") || metadata_file_name.ends_with(".json.gz"))
         || metadata_file_name.contains("..")
         || metadata_file_name.contains('%')
         || metadata_file_name.contains('/')

rustfs/src/table_catalog/maintenance/planner.rs

@@ -348,10 +348,18 @@ where
             "compaction manifest list must be inside the table metadata directory".to_string(),
         ));
     }
-    let Some(manifest_list_object) = backend.read_object(table_bucket, &manifest_list_key).await? else {
+    let Some(manifest_list_object) = backend
+        .read_object_limited(table_bucket, &manifest_list_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+        .await?
+    else {
         return Err(TableCatalogStoreError::NotFound(format!("compaction manifest list {manifest_list_key}")));
     };
-    let manifest_paths = manifest_paths_from_manifest_list_avro(&manifest_list_object.data)?;
+    let manifest_paths = decode_manifest_list_avro_async(manifest_list_object.data)
+        .await?
+        .references
+        .into_iter()
+        .map(|reference| reference.manifest_path)
+        .collect::<Vec<_>>();
     let mut planning = CompactionManifestPlanning::default();
     for manifest_location in manifest_paths {
         let Some(manifest_key) = table_catalog_object_key_from_location(table_bucket, &manifest_location) else {
@@ -366,10 +374,13 @@ where
                 "compaction manifest must be inside the table metadata directory".to_string(),
             ));
         }
-        let Some(manifest_object) = backend.read_object(table_bucket, &manifest_key).await? else {
+        let Some(manifest_object) = backend
+            .read_object_limited(table_bucket, &manifest_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+            .await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!("compaction manifest {manifest_key}")));
         };
-        for reference in data_file_references_from_manifest_avro(&manifest_object.data)? {
+        for reference in decode_manifest_avro_async(manifest_object.data).await?.references {
             if reference.object_kind != TableMetadataMaintenanceObjectKind::DataFile {
                 record_compaction_row_level_delete_file(
                     backend,
@@ -552,12 +563,15 @@ where
             "compaction manifest list must be inside the table metadata directory".to_string(),
         ));
     }
-    let Some(manifest_list_object) = backend.read_object(table_bucket, &manifest_list_key).await? else {
+    let Some(manifest_list_object) = backend
+        .read_object_limited(table_bucket, &manifest_list_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+        .await?
+    else {
         return Err(TableCatalogStoreError::NotFound(format!("compaction manifest list {manifest_list_key}")));
     };
 
     let mut data_files = Vec::new();
-    for manifest_reference in manifest_list_references_from_manifest_list_avro(&manifest_list_object.data)? {
+    for manifest_reference in decode_manifest_list_avro_async(manifest_list_object.data).await?.references {
         let Some(manifest_key) = table_catalog_object_key_from_location(table_bucket, &manifest_reference.manifest_path) else {
             return Err(TableCatalogStoreError::Invalid(
                 "compaction manifest must be inside the table bucket".to_string(),
@@ -570,10 +584,13 @@ where
                 "compaction manifest must be inside the table metadata directory".to_string(),
             ));
         }
-        let Some(manifest_object) = backend.read_object(table_bucket, &manifest_key).await? else {
+        let Some(manifest_object) = backend
+            .read_object_limited(table_bucket, &manifest_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+            .await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!("compaction manifest {manifest_key}")));
         };
-        for reference in data_file_references_from_manifest_avro(&manifest_object.data)? {
+        for reference in decode_manifest_avro_async(manifest_object.data).await?.references {
             if reference.object_kind != TableMetadataMaintenanceObjectKind::DataFile {
                 return Err(TableCatalogStoreError::Invalid(
                     "compaction currently does not support delete files".to_string(),

rustfs/src/table_catalog/maintenance/recovery.rs

@@ -97,36 +97,20 @@ where
     .await?;
 
     for metadata_location in retained_metadata_locations {
-        let Some(metadata_object) = backend.read_object(table_bucket, metadata_location).await? else {
-            insert_referenced_object_report(
-                &mut reports,
-                metadata_location.clone(),
-                TableMetadataMaintenanceObjectKind::MetadataFile,
-                TableMetadataMaintenanceObjectState::ManualReviewRequired,
-                TableMetadataMaintenanceReason::UnreadableMetadata,
-            );
-            continue;
+        let metadata = match read_table_metadata_value(backend, table_bucket, metadata_location).await {
+            Ok(Some(metadata)) => metadata,
+            Ok(None) | Err(TableCatalogStoreError::Invalid(_)) => {
+                insert_referenced_object_report(
+                    &mut reports,
+                    metadata_location.clone(),
+                    TableMetadataMaintenanceObjectKind::MetadataFile,
+                    TableMetadataMaintenanceObjectState::ManualReviewRequired,
+                    TableMetadataMaintenanceReason::UnreadableMetadata,
+                );
+                continue;
+            }
+            Err(err) => return Err(err),
         };
-        let Ok(metadata) = serde_json::from_slice::<serde_json::Value>(&metadata_object.data) else {
-            insert_referenced_object_report(
-                &mut reports,
-                metadata_location.clone(),
-                TableMetadataMaintenanceObjectKind::MetadataFile,
-                TableMetadataMaintenanceObjectState::ManualReviewRequired,
-                TableMetadataMaintenanceReason::UnreadableMetadata,
-            );
-            continue;
-        };
-        if !metadata.is_object() {
-            insert_referenced_object_report(
-                &mut reports,
-                metadata_location.clone(),
-                TableMetadataMaintenanceObjectKind::MetadataFile,
-                TableMetadataMaintenanceObjectState::ManualReviewRequired,
-                TableMetadataMaintenanceReason::UnreadableMetadata,
-            );
-            continue;
-        }
         metadata_maintenance_referenced_object_reports_for_metadata(
             backend,
             table_bucket,
@@ -233,9 +217,9 @@ where
         );
         return Ok(());
     };
-    if table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &manifest_list_key)
-        != Some(TableMetadataMaintenanceObjectKind::ManifestList)
-    {
+    if !table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &manifest_list_key).is_some_and(|kind| {
+        table_maintenance_object_kind_matches_reference(&kind, &TableMetadataMaintenanceObjectKind::ManifestList)
+    }) {
         insert_referenced_object_report(
             reports,
             manifest_list_key,
@@ -253,21 +237,36 @@ where
         TableMetadataMaintenanceReason::ManifestList,
     );
 
-    let Some(manifest_list_object) = backend.read_object(table_bucket, &manifest_list_key).await? else {
-        mark_referenced_object_manual_review(
-            reports,
-            &manifest_list_key,
-            TableMetadataMaintenanceReason::UnsupportedManifestAvro,
-        );
-        return Ok(());
+    let manifest_list_object = match backend
+        .read_object_limited(table_bucket, &manifest_list_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+        .await
+    {
+        Ok(Some(manifest_list_object)) => manifest_list_object,
+        Ok(None) | Err(TableCatalogStoreError::Invalid(_)) => {
+            mark_referenced_object_manual_review(
+                reports,
+                &manifest_list_key,
+                TableMetadataMaintenanceReason::UnsupportedManifestAvro,
+            );
+            return Ok(());
+        }
+        Err(err) => return Err(err),
     };
-    let Ok(manifest_paths) = manifest_paths_from_manifest_list_avro(&manifest_list_object.data) else {
-        mark_referenced_object_manual_review(
-            reports,
-            &manifest_list_key,
-            TableMetadataMaintenanceReason::UnsupportedManifestAvro,
-        );
-        return Ok(());
+    let manifest_paths = match decode_manifest_list_avro_async(manifest_list_object.data).await {
+        Ok(decoded) => decoded
+            .references
+            .into_iter()
+            .map(|reference| reference.manifest_path)
+            .collect::<Vec<_>>(),
+        Err(TableCatalogStoreError::Invalid(_)) => {
+            mark_referenced_object_manual_review(
+                reports,
+                &manifest_list_key,
+                TableMetadataMaintenanceReason::UnsupportedManifestAvro,
+            );
+            return Ok(());
+        }
+        Err(err) => return Err(err),
     };
     for manifest_location in manifest_paths {
         metadata_maintenance_referenced_manifest_file(
@@ -307,9 +306,9 @@ where
         );
         return Ok(());
     };
-    if table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &manifest_key)
-        != Some(TableMetadataMaintenanceObjectKind::ManifestFile)
-    {
+    if !table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &manifest_key).is_some_and(|kind| {
+        table_maintenance_object_kind_matches_reference(&kind, &TableMetadataMaintenanceObjectKind::ManifestFile)
+    }) {
         insert_referenced_object_report(
             reports,
             manifest_key,
@@ -327,13 +326,28 @@ where
         TableMetadataMaintenanceReason::ManifestFile,
     );
 
-    let Some(manifest_object) = backend.read_object(table_bucket, &manifest_key).await? else {
-        mark_referenced_object_manual_review(reports, &manifest_key, TableMetadataMaintenanceReason::UnsupportedManifestAvro);
-        return Ok(());
+    let manifest_object = match backend
+        .read_object_limited(table_bucket, &manifest_key, TABLE_MANIFEST_AVRO_MAX_SIZE)
+        .await
+    {
+        Ok(Some(manifest_object)) => manifest_object,
+        Ok(None) | Err(TableCatalogStoreError::Invalid(_)) => {
+            mark_referenced_object_manual_review(reports, &manifest_key, TableMetadataMaintenanceReason::UnsupportedManifestAvro);
+            return Ok(());
+        }
+        Err(err) => return Err(err),
     };
-    let Ok(file_references) = file_references_from_manifest_avro(&manifest_object.data) else {
-        mark_referenced_object_manual_review(reports, &manifest_key, TableMetadataMaintenanceReason::UnsupportedManifestAvro);
-        return Ok(());
+    let file_references = match decode_manifest_avro_async(manifest_object.data).await {
+        Ok(decoded) => decoded
+            .references
+            .into_iter()
+            .map(|reference| (reference.location, reference.object_kind))
+            .collect::<Vec<_>>(),
+        Err(TableCatalogStoreError::Invalid(_)) => {
+            mark_referenced_object_manual_review(reports, &manifest_key, TableMetadataMaintenanceReason::UnsupportedManifestAvro);
+            return Ok(());
+        }
+        Err(err) => return Err(err),
     };
     for (file_location, object_kind) in file_references {
         let Some(file_key) = table_catalog_object_key_from_location(table_bucket, &file_location) else {
@@ -346,7 +360,9 @@ where
             );
             continue;
         };
-        if table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &file_key) != Some(object_kind.clone()) {
+        if !table_maintenance_object_kind(namespace, table, warehouse_object_prefix, &file_key)
+            .is_some_and(|kind| table_maintenance_object_kind_matches_reference(&kind, &object_kind))
+        {
             insert_referenced_object_report(
                 reports,
                 file_key,
@@ -455,6 +471,23 @@ pub(crate) fn table_maintenance_object_kind(
     None
 }
 
+pub(crate) fn table_maintenance_object_kind_matches_reference(
+    actual: &TableMetadataMaintenanceObjectKind,
+    referenced: &TableMetadataMaintenanceObjectKind,
+) -> bool {
+    match referenced {
+        TableMetadataMaintenanceObjectKind::ManifestList | TableMetadataMaintenanceObjectKind::ManifestFile => matches!(
+            actual,
+            TableMetadataMaintenanceObjectKind::ManifestList | TableMetadataMaintenanceObjectKind::ManifestFile
+        ),
+        TableMetadataMaintenanceObjectKind::DeleteFile => matches!(
+            actual,
+            TableMetadataMaintenanceObjectKind::DataFile | TableMetadataMaintenanceObjectKind::DeleteFile
+        ),
+        _ => actual == referenced,
+    }
+}
+
 fn table_maintenance_metadata_object_kind(
     metadata_prefix: &str,
     object_location: &str,

rustfs/src/table_catalog/mod.rs

@@ -96,6 +96,15 @@ pub(crate) const ENV_TABLE_CATALOG_BACKING: &str = "RUSTFS_TABLE_CATALOG_BACKING
 pub(crate) const TABLE_CATALOG_BACKING_OBJECT: &str = "object";
 pub(crate) const TABLE_CATALOG_BACKING_DURABLE_STRONG: &str = "durable-strong";
 pub(crate) const TABLE_METADATA_FILE_NAME_MAX_LEN: usize = 128;
+pub(crate) const TABLE_METADATA_JSON_MAX_SIZE: usize = 50 * 1024 * 1024;
+pub(crate) const TABLE_MANIFEST_AVRO_MAX_SIZE: usize = 128 * 1024 * 1024;
+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_AVRO_BYTES: usize = 512 * 1024 * 1024;
+const TABLE_COMMIT_MAX_FILE_REFERENCES: usize = 1_000_000;
+const TABLE_COMMIT_OBJECT_VALIDATION_CONCURRENCY: usize = 16;
 pub const TABLE_RESERVED_PREFIX: &str = BUCKET_TABLE_RESERVED_PREFIX;
 const WAREHOUSE_ROOT: &str = "warehouses";
 const NAMESPACE_ROOT: &str = "namespaces";

rustfs/src/table_catalog/store/mod.rs

@@ -195,6 +195,10 @@ pub(crate) trait TableCatalogStore: Send + Sync {
         scan_table_data_plane_resource_for_object(self, table_bucket, object).await
     }
 
+    /// Atomically advances a validated table metadata pointer.
+    ///
+    /// Callers publishing client-supplied Iceberg metadata must validate its logical shape and the physical graph of
+    /// newly introduced or changed snapshots before invoking this persistence boundary.
     async fn commit_table(&self, request: TableCommitRequest) -> TableCatalogStoreResult<TableCommitResult>;
 
     async fn drop_table(&self, table_bucket: &str, namespace: &str, table: &str) -> TableCatalogStoreResult<()>;
@@ -269,10 +273,40 @@ pub(crate) enum TableCatalogPutPrecondition {
 pub(crate) trait TableCatalogObjectBackend: Clone + Send + Sync + 'static {
     async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>>;
 
+    async fn read_object_limited(
+        &self,
+        bucket: &str,
+        object: &str,
+        max_size: usize,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        let result = self.read_object(bucket, object).await?;
+        if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
+            return Err(TableCatalogStoreError::Invalid(format!(
+                "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
+            )));
+        }
+        Ok(result)
+    }
+
     async fn read_object_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
         self.read_object(bucket, object).await
     }
 
+    async fn read_object_unlocked_limited(
+        &self,
+        bucket: &str,
+        object: &str,
+        max_size: usize,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        let result = self.read_object_unlocked(bucket, object).await?;
+        if result.as_ref().is_some_and(|object| object.data.len() > max_size) {
+            return Err(TableCatalogStoreError::Invalid(format!(
+                "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
+            )));
+        }
+        Ok(result)
+    }
+
     async fn object_metadata(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObjectMetadata>> {
         Ok(self
             .read_object(bucket, object)
@@ -1115,7 +1149,18 @@ where
     S: TableCatalogStorage,
 {
     async fn read_object(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
-        self.read_object_with_options(bucket, object, ObjectOptions::default()).await
+        self.read_object_with_options(bucket, object, ObjectOptions::default(), None)
+            .await
+    }
+
+    async fn read_object_limited(
+        &self,
+        bucket: &str,
+        object: &str,
+        max_size: usize,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        self.read_object_with_options(bucket, object, ObjectOptions::default(), Some(max_size))
+            .await
     }
 
     async fn read_object_unlocked(&self, bucket: &str, object: &str) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
@@ -1126,6 +1171,25 @@ where
                 no_lock: true,
                 ..Default::default()
             },
+            None,
+        )
+        .await
+    }
+
+    async fn read_object_unlocked_limited(
+        &self,
+        bucket: &str,
+        object: &str,
+        max_size: usize,
+    ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
+        self.read_object_with_options(
+            bucket,
+            object,
+            ObjectOptions {
+                no_lock: true,
+                ..Default::default()
+            },
+            Some(max_size),
         )
         .await
     }
@@ -1286,12 +1350,22 @@ where
         bucket: &str,
         object: &str,
         opts: ObjectOptions,
+        max_size: Option<usize>,
     ) -> TableCatalogStoreResult<Option<TableCatalogObject>> {
         let info = match self.store.get_object_info(bucket, object, &opts).await {
             Ok(info) => info,
             Err(err) if is_missing_storage_error(&err) => return Ok(None),
             Err(err) => return Err(storage_error_to_catalog("read catalog object info", err)),
         };
+        if let Some(max_size) = max_size {
+            let object_size = usize::try_from(info.size)
+                .map_err(|_| TableCatalogStoreError::Invalid(format!("catalog object {bucket}/{object} has an invalid size")))?;
+            if object_size > max_size {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
+                )));
+            }
+        }
         let mut reader = match self
             .store
             .get_object_reader(bucket, object, None, HeaderMap::new(), &opts)
@@ -1302,11 +1376,21 @@ where
             Err(err) => return Err(storage_error_to_catalog("read catalog object", err)),
         };
         let mut data = Vec::new();
-        reader
-            .stream
-            .read_to_end(&mut data)
-            .await
-            .map_err(|err| TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}")))?;
+        if let Some(max_size) = max_size {
+            let read_limit = u64::try_from(max_size.saturating_add(1)).unwrap_or(u64::MAX);
+            reader.stream.take(read_limit).read_to_end(&mut data).await.map_err(|err| {
+                TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}"))
+            })?;
+            if data.len() > max_size {
+                return Err(TableCatalogStoreError::Invalid(format!(
+                    "catalog object {bucket}/{object} exceeds the maximum size of {max_size} bytes"
+                )));
+            }
+        } else {
+            reader.stream.read_to_end(&mut data).await.map_err(|err| {
+                TableCatalogStoreError::Internal(format!("failed to read catalog object {bucket}/{object}: {err}"))
+            })?;
+        }
         Ok(Some(TableCatalogObject {
             data,
             etag: info.etag,

rustfs/src/table_catalog/store/object.rs

@@ -2445,21 +2445,13 @@ where
             ));
         }
 
-        let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else {
+        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
                 entry.metadata_location
             )));
         };
-        let current_metadata = serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location))
-        })?;
-        if !current_metadata.is_object() {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "current metadata {} must be a JSON object",
-                entry.metadata_location
-            )));
-        }
 
         Ok(table_snapshot_expiration_report(
             table_bucket,
@@ -2497,21 +2489,13 @@ where
             ));
         }
 
-        let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else {
+        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
                 entry.metadata_location
             )));
         };
-        let current_metadata = serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location))
-        })?;
-        if !current_metadata.is_object() {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "current metadata {} must be a JSON object",
-                entry.metadata_location
-            )));
-        }
 
         table_compaction_planning_report(&self.backend, table_bucket, &namespace, &table, &entry, &current_metadata, config).await
     }
@@ -2541,21 +2525,13 @@ where
             ));
         }
 
-        let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else {
+        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
                 entry.metadata_location
             )));
         };
-        let current_metadata = serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location))
-        })?;
-        if !current_metadata.is_object() {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "current metadata {} must be a JSON object",
-                entry.metadata_location
-            )));
-        }
         let mut report =
             table_compaction_planning_report(&self.backend, table_bucket, &namespace, &table, &entry, &current_metadata, config)
                 .await?;
@@ -2769,18 +2745,20 @@ where
         let current_metadata_status =
             if is_valid_table_metadata_location(&parsed_namespace, &parsed_table, &current_metadata_location) {
                 retained.insert(current_metadata_location.clone());
-                match self.backend.read_object(table_bucket, &current_metadata_location).await? {
-                    Some(current_metadata_object) => {
-                        match serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data) {
-                            Ok(current_metadata) if current_metadata.is_object() => {
-                                retained.extend(metadata_log_locations(&current_metadata, &parsed_namespace, &parsed_table));
-                                current_metadata_for_refs = Some(current_metadata);
-                                TableMetadataPointerStatus::Valid
-                            }
-                            Ok(_) | Err(_) => TableMetadataPointerStatus::InvalidJson,
-                        }
+                match read_table_metadata_value(&self.backend, table_bucket, &current_metadata_location).await {
+                    Ok(Some(current_metadata)) => {
+                        retained.extend(metadata_log_locations(
+                            &current_metadata,
+                            table_bucket,
+                            &parsed_namespace,
+                            &parsed_table,
+                        ));
+                        current_metadata_for_refs = Some(current_metadata);
+                        TableMetadataPointerStatus::Valid
                     }
-                    None => TableMetadataPointerStatus::MissingObject,
+                    Ok(None) => TableMetadataPointerStatus::MissingObject,
+                    Err(TableCatalogStoreError::Invalid(_)) => TableMetadataPointerStatus::InvalidJson,
+                    Err(err) => return Err(err),
                 }
             } else {
                 TableMetadataPointerStatus::InvalidLocation
@@ -2859,25 +2837,17 @@ where
             ));
         }
 
-        let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else {
+        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
                 entry.metadata_location
             )));
         };
-        let current_metadata = serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location))
-        })?;
-        if !current_metadata.is_object() {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "current metadata {} must be a JSON object",
-                entry.metadata_location
-            )));
-        }
 
         let mut retained = BTreeSet::new();
         let mut maintenance_reasons = BTreeMap::<String, BTreeSet<TableMetadataMaintenanceReason>>::new();
-        for metadata_location in metadata_log_locations(&current_metadata, &namespace, &table) {
+        for metadata_location in metadata_log_locations(&current_metadata, table_bucket, &namespace, &table) {
             retained.insert(metadata_location.clone());
             insert_metadata_maintenance_reason(
                 &mut maintenance_reasons,
@@ -2945,7 +2915,7 @@ where
                 metadata_location.clone(),
                 TableMetadataMaintenanceReason::NoCurrentReachability,
             );
-            let Some(candidate_object) = self.backend.read_object(table_bucket, metadata_location).await? else {
+            let Some(candidate_object) = self.backend.object_metadata(table_bucket, metadata_location).await? else {
                 insert_metadata_maintenance_reason(
                     &mut maintenance_reasons,
                     metadata_location.clone(),
@@ -3290,23 +3260,15 @@ where
         }
         let warehouse_object_prefix = table_warehouse_object_prefix(&entry).ok();
 
-        let Some(current_metadata_object) = self.backend.read_object(table_bucket, &entry.metadata_location).await? else {
+        let Some(current_metadata) = read_table_metadata_value(&self.backend, table_bucket, &entry.metadata_location).await?
+        else {
             return Err(TableCatalogStoreError::NotFound(format!(
                 "current metadata object {}",
                 entry.metadata_location
             )));
         };
-        let current_metadata = serde_json::from_slice::<serde_json::Value>(&current_metadata_object.data).map_err(|err| {
-            TableCatalogStoreError::Invalid(format!("failed to parse current metadata {}: {err}", entry.metadata_location))
-        })?;
-        if !current_metadata.is_object() {
-            return Err(TableCatalogStoreError::Invalid(format!(
-                "current metadata {} must be a JSON object",
-                entry.metadata_location
-            )));
-        }
 
-        let mut protected = metadata_log_locations(&current_metadata, &namespace, &table);
+        let mut protected = metadata_log_locations(&current_metadata, table_bucket, &namespace, &table);
         protected.insert(entry.metadata_location.clone());
         protected.extend(report.retained_metadata_locations.iter().cloned());
         protected.extend(
@@ -3336,7 +3298,7 @@ where
                     "cleanup candidate {metadata_location} is retained by current metadata"
                 )));
             }
-            let Some(candidate_object) = self.backend.read_object(table_bucket, metadata_location).await? else {
+            let Some(candidate_object) = self.backend.object_metadata(table_bucket, metadata_location).await? else {
                 continue;
             };
             if !planned_deletable_locations.contains(metadata_location.as_str()) {
@@ -3927,7 +3889,7 @@ where
         }
         let Some(new_metadata_object) = self
             .backend
-            .read_object(&request.table_bucket, &request.new_metadata_location)
+            .read_object_limited(&request.table_bucket, &request.new_metadata_location, TABLE_METADATA_JSON_MAX_SIZE)
             .await?
         else {
             return table_commit_result(
@@ -3943,8 +3905,13 @@ where
                 ))),
             );
         };
-        let next_warehouse_location =
-            table_metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &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)
+        })
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
 
         let has_existing_commit = existing_commit.is_some();
         let mut staged_commit_log = existing_commit.unwrap_or_else(|| CommitLogEntry {
@@ -4190,7 +4157,7 @@ where
         }
         let Some(new_metadata_object) = self
             .backend
-            .read_object(&request.table_bucket, &request.new_metadata_location)
+            .read_object_limited(&request.table_bucket, &request.new_metadata_location, TABLE_METADATA_JSON_MAX_SIZE)
             .await?
         else {
             return Err(TableCatalogStoreError::NotFound(format!(
@@ -4198,8 +4165,13 @@ where
                 request.new_metadata_location
             )));
         };
-        let next_warehouse_location =
-            view_metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &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 || {
+            view_metadata_warehouse_location(&table_bucket, &metadata_location, &new_metadata_object)
+        })
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("view metadata parser task failed: {err}")))??;
 
         let mut next = current;
         next.metadata_location = request.new_metadata_location;

rustfs/src/table_catalog/store/strong.rs

@@ -1380,7 +1380,7 @@ where
 
         let Some(new_metadata_object) = self
             .object_backend
-            .read_object(&request.table_bucket, &request.new_metadata_location)
+            .read_object_limited(&request.table_bucket, &request.new_metadata_location, TABLE_METADATA_JSON_MAX_SIZE)
             .await?
         else {
             return table_commit_result(
@@ -1396,8 +1396,13 @@ where
                 ))),
             );
         };
-        let next_warehouse_location =
-            table_metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &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)
+        })
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("table metadata parser task failed: {err}")))??;
 
         let cas_started = Instant::now();
         let prepared_result = {
@@ -1540,7 +1545,7 @@ where
         }
         let Some(new_metadata_object) = self
             .object_backend
-            .read_object(&request.table_bucket, &request.new_metadata_location)
+            .read_object_limited(&request.table_bucket, &request.new_metadata_location, TABLE_METADATA_JSON_MAX_SIZE)
             .await?
         else {
             return Err(TableCatalogStoreError::NotFound(format!(
@@ -1548,8 +1553,13 @@ where
                 request.new_metadata_location
             )));
         };
-        let next_warehouse_location =
-            view_metadata_warehouse_location(&request.table_bucket, &request.new_metadata_location, &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 || {
+            view_metadata_warehouse_location(&table_bucket, &metadata_location, &new_metadata_object)
+        })
+        .await
+        .map_err(|err| TableCatalogStoreError::Internal(format!("view metadata parser task failed: {err}")))??;
 
         let key = Self::table_key(&request.table_bucket, &namespace, &view);
         let (snapshot, precondition, next) = {

rustfs/src/table_catalog/tests.rs

@@ -631,6 +631,10 @@ fn object_cleanup_report<'a>(
 }
 
 fn manifest_list_avro_bytes(manifest_paths: &[&str]) -> Vec<u8> {
+    manifest_list_avro_bytes_with_spec(manifest_paths, 0)
+}
+
+fn manifest_list_avro_bytes_with_spec(manifest_paths: &[&str], partition_spec_id: i32) -> Vec<u8> {
     let schema = apache_avro::Schema::parse_str(
         r#"
             {
@@ -638,9 +642,18 @@ fn manifest_list_avro_bytes(manifest_paths: &[&str]) -> Vec<u8> {
               "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": "added_snapshot_id", "type": "long"}
+                {"name": "min_sequence_number", "type": "long"},
+                {"name": "added_snapshot_id", "type": "long"},
+                {"name": "added_files_count", "type": "int"},
+                {"name": "existing_files_count", "type": "int"},
+                {"name": "deleted_files_count", "type": "int"},
+                {"name": "added_rows_count", "type": "long"},
+                {"name": "existing_rows_count", "type": "long"},
+                {"name": "deleted_rows_count", "type": "long"}
               ]
             }
             "#,
@@ -654,15 +667,773 @@ fn manifest_list_avro_bytes(manifest_paths: &[&str]) -> Vec<u8> {
                     "manifest_path".to_string(),
                     apache_avro::types::Value::String((*manifest_path).to_string()),
                 ),
-                ("partition_spec_id".to_string(), apache_avro::types::Value::Int(0)),
+                ("manifest_length".to_string(), apache_avro::types::Value::Long(1)),
+                ("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(7)),
+                ("min_sequence_number".to_string(), apache_avro::types::Value::Long(7)),
                 ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(20)),
+                ("added_files_count".to_string(), apache_avro::types::Value::Int(1)),
+                ("existing_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("deleted_files_count".to_string(), apache_avro::types::Value::Int(0)),
+                ("added_rows_count".to_string(), apache_avro::types::Value::Long(1)),
+                ("existing_rows_count".to_string(), apache_avro::types::Value::Long(0)),
+                ("deleted_rows_count".to_string(), apache_avro::types::Value::Long(0)),
             ]))
             .expect("manifest list record should append");
     }
     writer.into_inner().expect("manifest list avro bytes should flush")
 }
 
+fn v1_manifest_list_avro_bytes(manifest_path: &str) -> 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": "added_snapshot_id", "type": "long"}
+          ]
+        }
+        "#,
+    )
+    .expect("v1 manifest list schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+    writer
+        .append(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)),
+            ("partition_spec_id".to_string(), apache_avro::types::Value::Int(0)),
+            ("added_snapshot_id".to_string(), apache_avro::types::Value::Long(10)),
+        ]))
+        .expect("v1 manifest list record should append");
+    writer.into_inner().expect("v1 manifest list should flush")
+}
+
+fn v1_manifest_avro_bytes(data_file_path: &str) -> Vec<u8> {
+    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": "file_format", "type": "string"},
+                  {"name": "partition", "type": {"type": "record", "name": "partition", "fields": []}},
+                  {"name": "record_count", "type": "long"},
+                  {"name": "file_size_in_bytes", "type": "long"}
+                ]
+              }
+            }
+          ]
+        }
+        "#,
+    )
+    .expect("v1 manifest schema should parse");
+    let mut writer = apache_avro::Writer::new(&schema, Vec::new());
+    writer
+        .append(apache_avro::types::Value::Record(vec![
+            ("status".to_string(), apache_avro::types::Value::Int(1)),
+            ("snapshot_id".to_string(), apache_avro::types::Value::Long(10)),
+            (
+                "data_file".to_string(),
+                apache_avro::types::Value::Record(vec![
+                    ("file_path".to_string(), apache_avro::types::Value::String(data_file_path.to_string())),
+                    ("file_format".to_string(), apache_avro::types::Value::String("PARQUET".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)),
+                ]),
+            ),
+        ]))
+        .expect("v1 manifest record should append");
+    writer.into_inner().expect("v1 manifest should flush")
+}
+
+fn table_metadata_json_for_validation() -> serde_json::Value {
+    serde_json::json!({
+        "format-version": 2,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id",
+        "last-sequence-number": 0,
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schemas": [{
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        }],
+        "current-schema-id": 0,
+        "partition-specs": [{"spec-id": 0, "fields": []}],
+        "default-spec-id": 0,
+        "last-partition-id": 999,
+        "sort-orders": [{"order-id": 0, "fields": []}],
+        "default-sort-order-id": 0,
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": [],
+        "refs": {}
+    })
+}
+
+#[test]
+fn iceberg_metadata_validation_accepts_complete_v1_and_v2_shapes() {
+    validate_supported_table_metadata(&table_metadata_json_for_validation())
+        .expect("complete Iceberg v2 metadata should validate");
+
+    let v1 = serde_json::json!({
+        "format-version": 1,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id",
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        },
+        "partition-spec": [],
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": []
+    });
+    validate_supported_table_metadata(&v1).expect("complete Iceberg v1 metadata should validate");
+}
+
+#[test]
+fn iceberg_metadata_validation_rejects_incomplete_v2_and_dangling_references() {
+    let metadata = table_metadata_json_for_validation();
+    for field in [
+        "last-updated-ms",
+        "last-column-id",
+        "last-sequence-number",
+        "schemas",
+        "current-schema-id",
+        "partition-specs",
+        "default-spec-id",
+        "last-partition-id",
+        "sort-orders",
+        "default-sort-order-id",
+    ] {
+        let mut incomplete = metadata.clone();
+        incomplete
+            .as_object_mut()
+            .expect("metadata should be an object")
+            .remove(field);
+        validate_supported_table_metadata(&incomplete).expect_err("missing required Iceberg v2 metadata fields must be rejected");
+    }
+
+    let mut dangling = metadata;
+    dangling["snapshots"] = serde_json::json!([{"snapshot-id": 10, "schema-id": 7}]);
+    dangling["current-snapshot-id"] = serde_json::Value::from(10);
+    dangling["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 11}});
+    let error = validate_supported_table_metadata(&dangling).expect_err("dangling snapshot references must be rejected");
+    assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+}
+
+#[test]
+fn iceberg_metadata_validation_enforces_snapshot_and_ref_contracts() {
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(1);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 1,
+        "timestamp-ms": 1,
+        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro",
+        "summary": {"operation": "append"},
+        "schema-id": 0
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    validate_supported_table_metadata(&metadata).expect("complete snapshot and main ref should validate");
+
+    metadata["snapshots"][0]["summary"]["operation"] = serde_json::Value::from("rewrite-manifests");
+    validate_supported_table_metadata(&metadata).expect("Iceberg snapshot operations are extensible strings");
+
+    let mut empty_operation = metadata.clone();
+    empty_operation["snapshots"][0]["summary"]["operation"] = serde_json::Value::from("");
+    validate_supported_table_metadata(&empty_operation).expect_err("snapshot operation must not be empty");
+
+    let mut missing_timestamp = metadata.clone();
+    missing_timestamp["snapshots"][0]
+        .as_object_mut()
+        .expect("snapshot should be an object")
+        .remove("timestamp-ms");
+    validate_supported_table_metadata(&missing_timestamp).expect_err("snapshot timestamp must be required");
+
+    let mut mismatched_main = metadata.clone();
+    mismatched_main["snapshots"]
+        .as_array_mut()
+        .expect("snapshots should be an array")
+        .push(serde_json::json!({
+            "snapshot-id": 11,
+            "sequence-number": 1,
+            "timestamp-ms": 2,
+            "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-11.avro",
+            "summary": {"operation": "append"}
+        }));
+    mismatched_main["refs"]["main"]["snapshot-id"] = serde_json::Value::from(11);
+    validate_supported_table_metadata(&mismatched_main).expect_err("main ref must point to current-snapshot-id");
+
+    let mut invalid_tag = metadata;
+    invalid_tag["refs"]["release"] = serde_json::json!({
+        "type": "tag",
+        "snapshot-id": 10,
+        "min-snapshots-to-keep": 2
+    });
+    validate_supported_table_metadata(&invalid_tag).expect_err("tags must reject branch-only retention fields");
+}
+
+#[test]
+fn iceberg_metadata_validation_bounds_snapshot_sequence_numbers() {
+    let mut v2 = table_metadata_json_for_validation();
+    v2["last-sequence-number"] = serde_json::Value::from(1);
+    v2["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 2,
+        "timestamp-ms": 1,
+        "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-10.avro",
+        "summary": {"operation": "append"}
+    }]);
+    let error = validate_supported_table_metadata(&v2).expect_err("a v2 snapshot sequence must not exceed the table sequence");
+    assert_eq!(
+        error,
+        TableCatalogStoreError::Invalid(
+            "Iceberg v2 snapshot sequence-number must be between zero and last-sequence-number".to_string()
+        )
+    );
+
+    let mut v1 = serde_json::json!({
+        "format-version": 1,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id",
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schema": {"type": "struct", "fields": []},
+        "partition-spec": [],
+        "snapshots": [{
+            "snapshot-id": 10,
+            "sequence-number": 1,
+            "timestamp-ms": 1,
+            "manifests": []
+        }]
+    });
+    let error = validate_supported_table_metadata(&v1).expect_err("v1 metadata must not carry a non-zero snapshot sequence");
+    assert_eq!(
+        error,
+        TableCatalogStoreError::Invalid("Iceberg v1 snapshot sequence-number must be zero when present".to_string())
+    );
+    v1["snapshots"][0]["sequence-number"] = serde_json::Value::from(0);
+    validate_supported_table_metadata(&v1).expect("a zero v1 compatibility sequence should validate");
+}
+
+#[test]
+fn iceberg_metadata_decoder_accepts_both_gzip_file_name_conventions() {
+    use std::io::Write;
+
+    let metadata = table_metadata_json_for_validation();
+    let data = serde_json::to_vec(&metadata).expect("metadata should serialize");
+    let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
+    encoder.write_all(&data).expect("metadata should compress");
+    let compressed = encoder.finish().expect("gzip stream should finish");
+
+    for metadata_location in ["v1.gz.metadata.json", "v1.metadata.json.gz"] {
+        let decoded = decode_table_metadata_json(metadata_location, &compressed)
+            .expect("Iceberg gzip metadata naming conventions should decode");
+        assert_eq!(decoded, metadata);
+    }
+}
+
+#[test]
+fn metadata_log_locations_normalize_same_bucket_s3_uris() {
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let metadata_location = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json");
+    let metadata = serde_json::json!({
+        "metadata-log": [
+            {"timestamp-ms": 1, "metadata-file": format!("s3://warehouse/{metadata_location}")},
+            {"timestamp-ms": 2, "metadata-file": format!("s3://other/{metadata_location}")}
+        ]
+    });
+
+    assert_eq!(
+        metadata_log_locations(&metadata, "warehouse", &namespace, &table),
+        BTreeSet::from([metadata_location])
+    );
+}
+
+#[test]
+fn iceberg_metadata_version_synchronization_builds_complete_v2_shape() {
+    let mut metadata = serde_json::json!({
+        "format-version": 2,
+        "table-uuid": "table-uuid",
+        "location": "s3://warehouse/tables/table-id",
+        "last-updated-ms": 1,
+        "last-column-id": 1,
+        "schema": {
+            "type": "struct",
+            "schema-id": 7,
+            "fields": [{"id": 1, "name": "id", "required": true, "type": "long"}]
+        },
+        "partition-spec": [],
+        "properties": {},
+        "snapshots": [],
+        "snapshot-log": [],
+        "metadata-log": []
+    });
+
+    synchronize_table_metadata_version_fields(&mut metadata).expect("v1 fields should synchronize to v2");
+    validate_supported_table_metadata(&metadata).expect("synchronized metadata should satisfy the v2 contract");
+    assert_eq!(metadata["current-schema-id"], 7);
+    assert_eq!(metadata["default-spec-id"], 0);
+    assert_eq!(metadata["default-sort-order-id"], 0);
+    assert_eq!(metadata["last-sequence-number"], 0);
+    assert!(metadata.get("schema").is_none());
+    assert!(metadata.get("partition-spec").is_none());
+}
+
+#[test]
+fn iceberg_manifest_validation_accepts_deflate_and_rejects_unknown_content() {
+    let schema = apache_avro::Schema::parse_str(
+        r#"
+        {
+          "type": "record",
+          "name": "manifest_file",
+          "fields": [
+            {"name": "manifest_path", "type": "string"},
+            {"name": "partition_spec_id", "type": "int"}
+          ]
+        }
+        "#,
+    )
+    .expect("manifest list schema should parse");
+    let mut writer = apache_avro::Writer::with_codec(&schema, Vec::new(), apache_avro::Codec::Deflate(Default::default()));
+    writer
+        .append(apache_avro::types::Value::Record(vec![
+            (
+                "manifest_path".to_string(),
+                apache_avro::types::Value::String("s3://warehouse/tables/table-id/metadata/manifest.avro".to_string()),
+            ),
+            ("partition_spec_id".to_string(), apache_avro::types::Value::Int(0)),
+        ]))
+        .expect("compressed manifest list record should append");
+    let compressed = writer.into_inner().expect("compressed manifest list should flush");
+    let references = manifest_list_references_from_manifest_list_avro(&compressed)
+        .expect("deflate-compressed manifest lists should be supported with bounded decoding");
+    assert_eq!(references.len(), 1);
+    assert_eq!(references[0].manifest_path, "s3://warehouse/tables/table-id/metadata/manifest.avro");
+
+    let unknown_content = manifest_avro_bytes_with_status(&[("s3://warehouse/tables/table-id/data/part.parquet", 3, 1)]);
+    let error = data_file_references_from_manifest_avro(&unknown_content)
+        .expect_err("unknown Iceberg manifest content values must be rejected");
+    assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_graph_rejects_unknown_partition_specs() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = TableEntry {
+        version: TABLE_CATALOG_ENTRY_VERSION,
+        table_bucket: "warehouse".to_string(),
+        namespace: namespace.public_name(),
+        table: table.as_str().to_string(),
+        table_id: "table-id".to_string(),
+        table_uuid: "table-uuid".to_string(),
+        format: "ICEBERG".to_string(),
+        format_version: 2,
+        warehouse_location: "s3://warehouse/tables/table-id".to_string(),
+        metadata_location: "tables/table-id/metadata/v1.metadata.json".to_string(),
+        version_token: "token-v1".to_string(),
+        generation: 1,
+        state: TableCatalogEntryState::Active,
+        properties: BTreeMap::new(),
+        created_at: None,
+        updated_at: None,
+    };
+    let manifest_list_location = "s3://warehouse/tables/table-id/metadata/snap-10.avro";
+    let manifest_location = "s3://warehouse/tables/table-id/metadata/manifest-10.avro";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/snap-10.avro",
+            manifest_list_avro_bytes_with_spec(&[manifest_location], 7),
+        )
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": manifest_list_location,
+        "summary": {"operation": "append"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    let error = validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect_err("manifest partition specs absent from table metadata must be rejected");
+    assert_eq!(
+        error,
+        TableCatalogStoreError::Invalid("snapshot manifest references missing partition spec 7".to_string())
+    );
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_graph_allows_missing_deleted_files() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest_list_location = "s3://warehouse/tables/table-id/metadata/snap-10.avro";
+    let manifest_location = "s3://warehouse/tables/table-id/metadata/manifest-10.avro";
+    let deleted_data_location = "s3://warehouse/tables/table-id/data/deleted.parquet";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/snap-10.avro",
+            manifest_list_avro_bytes(&[manifest_location]),
+        )
+        .await;
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/manifest-10.avro",
+            manifest_avro_bytes_with_status(&[(deleted_data_location, 0, 2)]),
+        )
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": manifest_list_location,
+        "summary": {"operation": "delete"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("deleted manifest entries may reference files that have already been removed");
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_graph_accepts_empty_manifest_lists() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest_list_location = "s3://warehouse/tables/table-id/metadata/snap-empty.avro";
+    backend
+        .seed_object("warehouse", "tables/table-id/metadata/snap-empty.avro", manifest_list_avro_bytes(&[]))
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": manifest_list_location,
+        "summary": {"operation": "append"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("an empty Iceberg snapshot may have an empty manifest list");
+}
+
+#[tokio::test]
+async fn iceberg_v2_snapshot_graph_accepts_reused_v1_manifests() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest_list_location = "s3://warehouse/tables/table-id/metadata/snap-v1.avro";
+    let manifest_location = "s3://warehouse/tables/table-id/metadata/manifest-v1.avro";
+    let data_location = "s3://warehouse/tables/table-id/data/v1.parquet";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/snap-v1.avro",
+            v1_manifest_list_avro_bytes(manifest_location),
+        )
+        .await;
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/manifest-v1.avro",
+            v1_manifest_avro_bytes(data_location),
+        )
+        .await;
+    backend
+        .seed_object("warehouse", "tables/table-id/data/v1.parquet", vec![1])
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": manifest_list_location,
+        "summary": {"operation": "append"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("v2 tables may retain v1 manifest lists and manifests after upgrade");
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_change_validation_skips_unchanged_history() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let historical_manifest_list = "s3://warehouse/tables/table-id/metadata/snap-10.avro";
+    let new_manifest_list = "s3://warehouse/tables/table-id/metadata/snap-11.avro";
+    backend
+        .seed_object("warehouse", "tables/table-id/metadata/snap-11.avro", manifest_list_avro_bytes(&[]))
+        .await;
+
+    let mut current_metadata = table_metadata_json_for_validation();
+    current_metadata["last-sequence-number"] = serde_json::Value::from(7);
+    current_metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": historical_manifest_list,
+        "summary": {"operation": "append"}
+    }]);
+    current_metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    current_metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let mut next_metadata = current_metadata.clone();
+    next_metadata
+        .get_mut("snapshots")
+        .and_then(serde_json::Value::as_array_mut)
+        .expect("snapshots should be an array")
+        .push(serde_json::json!({
+            "snapshot-id": 11,
+            "sequence-number": 7,
+            "timestamp-ms": 2,
+            "manifest-list": new_manifest_list,
+            "summary": {"operation": "append"}
+        }));
+    next_metadata["current-snapshot-id"] = serde_json::Value::from(11);
+    next_metadata["refs"]["main"]["snapshot-id"] = serde_json::Value::from(11);
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, Some(&current_metadata), &next_metadata)
+        .await
+        .expect("an unchanged historical snapshot must not be reread during commit validation");
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_registration_validates_only_active_snapshot_heads() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    backend
+        .seed_object("warehouse", "tables/table-id/metadata/snap-11.avro", manifest_list_avro_bytes(&[]))
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([
+        {
+            "snapshot-id": 10,
+            "sequence-number": 7,
+            "timestamp-ms": 1,
+            "manifest-list": "s3://warehouse/tables/table-id/metadata/missing-history.avro",
+            "summary": {"operation": "append"}
+        },
+        {
+            "snapshot-id": 11,
+            "sequence-number": 7,
+            "timestamp-ms": 2,
+            "manifest-list": "s3://warehouse/tables/table-id/metadata/snap-11.avro",
+            "summary": {"operation": "append"}
+        }
+    ]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(11);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 11}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("registration should validate active snapshot heads without traversing all history");
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_graph_counts_shared_manifests_once() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest_list = "s3://warehouse/tables/table-id/metadata/shared-list.avro";
+    let manifest = "s3://warehouse/tables/table-id/metadata/shared-manifest.avro";
+    let data_file = "s3://warehouse/tables/table-id/data/shared.parquet";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/shared-list.avro",
+            manifest_list_avro_bytes(&[manifest]),
+        )
+        .await;
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/shared-manifest.avro",
+            manifest_avro_bytes(&[(data_file, 0)]),
+        )
+        .await;
+    backend
+        .seed_object("warehouse", "tables/table-id/data/shared.parquet", vec![1])
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::Value::Array(
+        (0..=TABLE_COMMIT_MAX_MANIFESTS)
+            .map(|index| {
+                let snapshot_id = i64::try_from(index + 1).expect("snapshot id should fit in i64");
+                serde_json::json!({
+                    "snapshot-id": snapshot_id,
+                    "sequence-number": 7,
+                    "timestamp-ms": snapshot_id,
+                    "manifest-list": manifest_list,
+                    "summary": {"operation": "append"}
+                })
+            })
+            .collect(),
+    );
+    let current_snapshot_id = i64::try_from(TABLE_COMMIT_MAX_MANIFESTS + 1).expect("snapshot id should fit in i64");
+    metadata["current-snapshot-id"] = serde_json::Value::from(current_snapshot_id);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": current_snapshot_id}});
+    let current_metadata = table_metadata_json_for_validation();
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, Some(&current_metadata), &metadata)
+        .await
+        .expect("shared manifest objects must consume the commit budget only once");
+}
+
+#[tokio::test]
+async fn iceberg_v2_snapshot_graph_accepts_embedded_v2_manifests() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest = "s3://warehouse/tables/table-id/metadata/embedded.avro";
+    let data_file = "s3://warehouse/tables/table-id/data/embedded.parquet";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/embedded.avro",
+            manifest_avro_bytes(&[(data_file, 0)]),
+        )
+        .await;
+    backend
+        .seed_object("warehouse", "tables/table-id/data/embedded.parquet", vec![1])
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifests": [manifest],
+        "summary": {"operation": "append"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("embedded v2 manifests should use the enclosing snapshot sequence bound");
+}
+
+#[tokio::test]
+async fn iceberg_snapshot_graph_accepts_delete_files_in_data_directory() {
+    let backend = TestCatalogObjectBackend::default();
+    let namespace = Namespace::parse("analytics").expect("namespace should parse");
+    let table = IdentifierSegment::parse("events").expect("table should parse");
+    let entry = test_table_entry("warehouse", &namespace, &table, "tables/table-id/metadata/v1.metadata.json".to_string());
+    let manifest_list = "s3://warehouse/tables/table-id/metadata/list-10.avro";
+    let manifest = "s3://warehouse/tables/table-id/metadata/snap-delete-manifest.avro";
+    let delete_file = "s3://warehouse/tables/table-id/data/position-deletes.parquet";
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/list-10.avro",
+            manifest_list_avro_bytes(&[manifest]),
+        )
+        .await;
+    backend
+        .seed_object(
+            "warehouse",
+            "tables/table-id/metadata/snap-delete-manifest.avro",
+            manifest_avro_bytes(&[(delete_file, 1)]),
+        )
+        .await;
+    backend
+        .seed_object("warehouse", "tables/table-id/data/position-deletes.parquet", vec![1])
+        .await;
+    let mut metadata = table_metadata_json_for_validation();
+    metadata["last-sequence-number"] = serde_json::Value::from(7);
+    metadata["snapshots"] = serde_json::json!([{
+        "snapshot-id": 10,
+        "sequence-number": 7,
+        "timestamp-ms": 1,
+        "manifest-list": manifest_list,
+        "summary": {"operation": "delete"}
+    }]);
+    metadata["current-snapshot-id"] = serde_json::Value::from(10);
+    metadata["refs"] = serde_json::json!({"main": {"type": "branch", "snapshot-id": 10}});
+    let context = TableSnapshotGraphValidationContext::new(&backend, "warehouse", &namespace, &table, &entry);
+
+    validate_table_snapshot_changes(&context, None, &metadata)
+        .await
+        .expect("manifest content should identify delete files stored under the data directory");
+}
+
+#[tokio::test]
+async fn catalog_object_limited_reads_reject_oversized_results() {
+    let backend = TestCatalogObjectBackend::default();
+    backend.seed_object("warehouse", "metadata.json", vec![0; 5]).await;
+
+    let error = backend
+        .read_object_limited("warehouse", "metadata.json", 4)
+        .await
+        .expect_err("bounded catalog reads must reject oversized objects");
+    assert!(matches!(error, TableCatalogStoreError::Invalid(_)));
+}
+
 fn manifest_avro_bytes(files: &[(&str, i32)]) -> Vec<u8> {
     manifest_avro_bytes_with_status(
         &files
@@ -8436,6 +9207,8 @@ fn table_metadata_file_path_stays_under_metadata_boundary() {
 #[test]
 fn table_metadata_file_name_validation_rejects_unsafe_names() {
     assert!(is_valid_table_metadata_file_name("00001.metadata.json"));
+    assert!(is_valid_table_metadata_file_name("00001.gz.metadata.json"));
+    assert!(is_valid_table_metadata_file_name("00001.metadata.json.gz"));
     assert!(is_valid_table_metadata_file_name("v1-4f2c_metadata.json"));
 
     assert!(!is_valid_table_metadata_file_name(""));