fix(table-catalog): assign fresh schema IDs on create (#6146)

* fix(table-catalog): assign fresh schema IDs on create

* fix(table-catalog): accept negative create schema IDs

---------

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

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -668,6 +668,60 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     parts_metadata: &[FileInfo],
     disks: &[Option<DiskStore>],
 ) -> Option<&'static str> {
+    if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
+        return Some(reason);
+    }
+
+    let Ok(erasure) = coding::Erasure::try_new_with_options(
+        candidate.erasure.data_blocks,
+        candidate.erasure.parity_blocks,
+        candidate.erasure.block_size,
+        candidate.uses_legacy_checksum,
+    ) else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+    };
+    let data_files =
+        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
+            disks.get(index).is_some_and(Option::is_some)
+        }) {
+            Ok(data_files) => data_files,
+            Err(reason) => return Some(reason),
+        };
+
+    let Some(part) = candidate.parts.first() else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+    };
+    let Ok(object_size) = usize::try_from(candidate.size) else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
+    };
+    let checksum_info = candidate.erasure.get_checksum_info(part.number);
+    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
+        HashAlgorithm::HighwayHash256SLegacy
+    } else {
+        checksum_info.algorithm
+    };
+    let read_length = inline_erasure_shard_file_offset(
+        0,
+        object_size,
+        object_size,
+        candidate.erasure.block_size,
+        erasure.data_shards,
+        candidate.uses_legacy_checksum,
+    );
+    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
+    let Ok(mut readers) =
+        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
+    else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+    };
+
+    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
+        Some(body) if body.len() == object_size => None,
+        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+    }
+}
+
+fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
     // `inline_data` excludes remote objects; this diagnostic reports them separately.
     if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
@@ -705,51 +759,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     }
-
-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-    };
-    let data_files =
-        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
-            disks.get(index).is_some_and(Option::is_some)
-        }) {
-            Ok(data_files) => data_files,
-            Err(reason) => return Some(reason),
-        };
-
-    let Some(part) = candidate.parts.first() else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    };
-    let checksum_info = candidate.erasure.get_checksum_info(part.number);
-    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
-        HashAlgorithm::HighwayHash256SLegacy
-    } else {
-        checksum_info.algorithm
-    };
-    let read_length = inline_erasure_shard_file_offset(
-        0,
-        object_size,
-        object_size,
-        candidate.erasure.block_size,
-        erasure.data_shards,
-        candidate.uses_legacy_checksum,
-    );
-    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
-    let Ok(mut readers) =
-        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
-    else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
-    };
-
-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => None,
-        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
-    }
+    None
 }
 
 fn data_read_inline_missing_shards_are_pending(
@@ -2446,6 +2456,7 @@ impl SetDisks {
         let bucket: Arc<str> = Arc::from(bucket);
         let object: Arc<str> = Arc::from(object);
         let version_id: Arc<str> = Arc::from(version_id);
+        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
             let disk = disk.clone();
             let task_opts = opts;
@@ -2453,10 +2464,14 @@ impl SetDisks {
             let bucket = bucket.clone();
             let object = object.clone();
             let version_id = version_id.clone();
+            let slowtail_fault = slowtail_fault.clone();
             tokio::spawn(async move {
                 let response_start = observe.then(Instant::now);
                 let result = if let Some(disk) = disk {
                     Self::record_read_version_call(&object, disk_index);
+                    if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
+                        tokio::time::sleep(delay).await;
+                    }
                     disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                         .await
                 } else {
@@ -2552,6 +2567,7 @@ impl SetDisks {
         let mut scheduled_count = 0usize;
         let mut force_full_wait = false;
         let mut final_miss_reason_override = None;
+        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let spawn_read_version =
             |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                 let task_opts = opts;
@@ -2559,6 +2575,7 @@ impl SetDisks {
                 let bucket = bucket.clone();
                 let object = object.clone();
                 let version_id = version_id.clone();
+                let slowtail_fault = slowtail_fault.clone();
                 join_set.spawn(async move {
                     let response_start = Instant::now();
                     let result = if let Some(disk) = disk {
@@ -2567,6 +2584,9 @@ impl SetDisks {
                         Self::record_read_version_call(&object, index);
                         #[cfg(test)]
                         Self::read_version_fanout_barrier(&object, index).await;
+                        if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
+                            tokio::time::sleep(delay).await;
+                        }
                         disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                             .await
                     } else {
@@ -2600,6 +2620,14 @@ impl SetDisks {
                     Ok(file_info) => {
                         observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
                         accumulator.observe_file_info(&file_info);
+                        if bounded_fanout
+                            && read_data
+                            && !force_full_wait
+                            && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
+                        {
+                            force_full_wait = true;
+                            final_miss_reason_override.get_or_insert(reason);
+                        }
                         if let Some(slot) = ress.get_mut(index) {
                             *slot = file_info;
                         }
@@ -5744,6 +5772,130 @@ mod tests {
         (dirs, disks)
     }
 
+    #[test]
+    fn metadata_slowtail_fault_delay_parses_and_filters_request() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            || {
+                assert_eq!(
+                    get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
+                    Some(Duration::from_millis(25))
+                );
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
+            },
+        );
+    }
+
+    #[test]
+    fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
+            ],
+            || {
+                assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
+            },
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
+        const DISKS: usize = 4;
+        let bucket = "metadata-slowtail-fault-bucket";
+        let object = "objects/metadata-slowtail-fault-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
+
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            async {
+                let read_without_data =
+                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
+                tokio::time::timeout(Duration::from_millis(100), read_without_data)
+                    .await
+                    .expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
+                    .expect("metadata fanout without read_data should resolve");
+
+                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
+                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
+                ));
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
+                        .await
+                        .is_err(),
+                    "data-read metadata fanout must wait for the injected slow read_version response"
+                );
+                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
+                    .await
+                    .expect("injected slowtail should eventually complete")
+                    .expect("data-read metadata fanout should resolve");
+                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
+                assert!(errs.iter().all(Option::is_none));
+                assert_eq!(diagnostics.total_responses(), DISKS);
+            },
+        )
+        .await;
+
+        drop(dirs);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
+        const DISKS: usize = 4;
+        let bucket = "metadata-slowtail-early-stop-bucket";
+        let object = "objects/metadata-slowtail-early-stop-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
+
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
+                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            async {
+                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
+                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
+                ));
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
+                        .await
+                        .is_err(),
+                    "early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
+                );
+                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
+                    .await
+                    .expect("injected early-stop slowtail should eventually complete")
+                    .expect("early-stop metadata fanout should resolve");
+                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
+                assert!(errs.iter().all(Option::is_none));
+                assert_eq!(diagnostics.total_responses(), DISKS);
+            },
+        )
+        .await;
+
+        drop(dirs);
+    }
+
     /// Demo / regression guard for the backlog#1325 per-disk call counters.
     ///
     /// The metadata fan-out issues each `read_version` inside its own
@@ -7091,7 +7243,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
+    async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
         const DISKS: usize = 4;
         let bucket = "bounded-data-get-hedge-bucket";
         let object = "bounded-data-get-hedge-object";
@@ -7122,7 +7274,7 @@ mod tests {
                     }
                 })
                 .await
-                .expect("bounded data-read fanout should hedge by starting the spare disk");
+                .expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
 
                 let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
                 assert!(
@@ -7138,7 +7290,7 @@ mod tests {
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "bounded data-read fanout should issue the paused disk plus one spare hedge"
+                    "bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7165,16 +7317,42 @@ mod tests {
                 ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
             ],
             async {
+                let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
                 let calls = disk_call_counters::observe(object);
-                let (parts_metadata, errs, diagnostics) =
-                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
+                let disks_for_read = disks.clone();
+                let mut read = tokio::spawn(async move {
+                    SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
                         .await
-                        .expect("default data-read metadata should resolve");
+                });
+
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
+                    .await
+                    .expect("default bounded non-inline read should schedule the paused metadata task");
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
+                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect(
+                    "default bounded non-inline read should immediately force full fanout after the first non-inline response",
+                );
+
+                let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
+                assert!(
+                    pending.is_err(),
+                    "default non-inline data reads must not return before the paused metadata response"
+                );
+                barrier.release();
+                let (parts_metadata, errs, diagnostics) = read
+                    .await
+                    .expect("metadata read task should not panic")
+                    .expect("default data-read metadata should resolve");
 
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
+                    "default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);

crates/ecstore/src/set_disk/mod.rs

@@ -174,15 +174,14 @@ use std::future::Future;
 use std::hash::{BuildHasher, Hash, Hasher};
 use std::mem::{self};
 use std::pin::Pin;
-use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, OnceLock};
 use std::task::{Context, Poll};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
     path::Path,
-    sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
@@ -715,7 +714,12 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
 const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
 
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
-const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
+const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
+
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
 
 // --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
 
@@ -1121,7 +1125,10 @@ mod prepared_get_object_metadata_tests {
 
         assert_eq!(object_size, payload.len() as i64);
         assert_eq!(restored, payload);
-        assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
+        assert_eq!(
+            calls_total, 4,
+            "default production inline GET should schedule the initial bounded quorum plus one hedge"
+        );
         assert_eq!(
             recorder.histogram_values(
                 "rustfs_io_get_object_metadata_fanout_scheduled",
@@ -1695,6 +1702,95 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
     }
 }
 
+#[derive(Debug)]
+struct GetMetadataSlowtailFaultConfig {
+    delay: Duration,
+    disks: Arc<[usize]>,
+    bucket: Option<String>,
+    object_prefix: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+struct GetMetadataSlowtailFaultRequest {
+    delay: Duration,
+    disks: Arc<[usize]>,
+}
+
+impl GetMetadataSlowtailFaultRequest {
+    fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
+        self.disks.contains(&disk_index).then_some(self.delay)
+    }
+}
+
+fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
+    let mut disks = Vec::new();
+    for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
+        let Ok(index) = item.parse::<usize>() else {
+            return None;
+        };
+        if !disks.contains(&index) {
+            disks.push(index);
+        }
+    }
+    (!disks.is_empty()).then_some(disks)
+}
+
+fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
+    let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
+    if delay_ms == 0 {
+        return None;
+    }
+    let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
+    let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
+        .ok()
+        .filter(|value| !value.is_empty());
+    let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
+        .ok()
+        .filter(|value| !value.is_empty());
+    Some(GetMetadataSlowtailFaultConfig {
+        delay: Duration::from_millis(delay_ms),
+        disks: Arc::from(disks.into_boxed_slice()),
+        bucket,
+        object_prefix,
+    })
+}
+
+fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
+    if !read_data {
+        return None;
+    }
+
+    #[cfg(test)]
+    let config = load_get_metadata_slowtail_fault_config();
+    #[cfg(test)]
+    let config = config.as_ref()?;
+    #[cfg(not(test))]
+    let config = ({
+        static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
+        CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
+    })?;
+
+    if let Some(expected_bucket) = &config.bucket
+        && expected_bucket != bucket
+    {
+        return None;
+    }
+    if let Some(expected_prefix) = &config.object_prefix
+        && !object.starts_with(expected_prefix)
+    {
+        return None;
+    }
+    Some(GetMetadataSlowtailFaultRequest {
+        delay: config.delay,
+        disks: config.disks.clone(),
+    })
+}
+
+#[cfg(test)]
+fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
+    get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
+}
+
 /// Check if multipart reads prefetch the next part's bitrot reader setup
 /// while the current part decodes (backlog#870).
 ///

crates/ecstore/src/set_disk/read.rs

@@ -3937,7 +3937,7 @@ mod tests {
     }
 
     #[test]
-    fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
+    fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3946,20 +3946,20 @@ mod tests {
             ],
             || {
                 assert!(is_get_metadata_data_read_early_stop_enabled());
-                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
-        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
-            assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
+        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
+            assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
         });
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
             ],
             || {
                 assert!(!is_get_metadata_data_read_early_stop_enabled());
-                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
     }

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

@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
     let CreateTableRequest {
         name,
         location,
-        schema,
-        partition_spec,
-        write_order,
+        mut schema,
+        mut partition_spec,
+        mut write_order,
         stage_create,
         mut properties,
     } = request;
@@ -3031,6 +3031,9 @@ fn table_entry_from_create_table_request(
     let metadata_location =
         crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
 
+    crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
+        .map_err(catalog_store_error)?;
+
     let entry = crate::table_catalog::TableEntry {
         version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
         table_bucket: bucket.to_string(),

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

@@ -1873,6 +1873,66 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
     assert_eq!(request.name, "events");
 }
 
+#[test]
+fn create_table_assigns_positive_ids_to_spark_schema() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [
+                {"id": 0, "name": "id", "required": false, "type": "long"},
+                {"id": 1, "name": "payload", "required": false, "type": "string"}
+            ]
+        },
+        "partition-spec": {"spec-id": 0, "fields": []},
+        "properties": {"owner": "spark"}
+    }))
+    .expect("Spark create table request should parse");
+
+    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect("catalog should assign positive field IDs");
+
+    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
+    assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
+    assert_eq!(metadata["last-column-id"], 2);
+}
+
+#[test]
+fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "identifier-field-ids": [-1],
+            "fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
+        },
+        "partition-spec": {
+            "fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
+        },
+        "write-order": {
+            "fields": [{
+                "source-id": -1,
+                "transform": "identity",
+                "direction": "asc",
+                "null-order": "nulls-first"
+            }]
+        }
+    }))
+    .expect("create table request with a negative temporary field ID should parse");
+
+    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect("catalog should replace the negative temporary field ID");
+
+    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
+    assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
+    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["last-column-id"], 1);
+}
+
 #[test]
 fn create_table_request_honors_supported_format_version_property() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -1990,6 +2050,142 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
     assert_eq!(updated["default-sort-order-id"], 0);
 }
 
+#[test]
+fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 41,
+            "identifier-field-ids": [0],
+            "fields": [
+                {"id": 0, "name": "id", "required": true, "type": "long"},
+                {
+                    "id": 10,
+                    "name": "details",
+                    "required": false,
+                    "type": {
+                        "type": "struct",
+                        "fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
+                    }
+                },
+                {
+                    "id": 20,
+                    "name": "tags",
+                    "required": false,
+                    "type": {
+                        "type": "list",
+                        "element-id": 21,
+                        "element-required": false,
+                        "element": "string"
+                    }
+                },
+                {
+                    "id": 30,
+                    "name": "attributes",
+                    "required": false,
+                    "type": {
+                        "type": "map",
+                        "key-id": 31,
+                        "key": "string",
+                        "value-id": 32,
+                        "value-required": false,
+                        "value": {
+                            "type": "struct",
+                            "fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
+                        }
+                    }
+                }
+            ]
+        },
+        "partition-spec": {
+            "spec-id": 42,
+            "fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
+        },
+        "write-order": {
+            "order-id": 43,
+            "fields": [{
+                "source-id": 11,
+                "transform": "identity",
+                "direction": "asc",
+                "null-order": "nulls-first"
+            }]
+        }
+    }))
+    .expect("create table request should parse");
+
+    let (_, metadata) =
+        table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
+
+    let schema = &metadata["schemas"][0];
+    assert_eq!(schema["fields"][0]["id"], 1);
+    assert_eq!(schema["fields"][1]["id"], 2);
+    assert_eq!(schema["fields"][2]["id"], 3);
+    assert_eq!(schema["fields"][3]["id"], 4);
+    assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
+    assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
+    assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
+    assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
+    assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
+    assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
+    assert_eq!(metadata["last-column-id"], 9);
+    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
+}
+
+#[test]
+fn create_table_rejects_duplicate_temporary_schema_field_ids() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "fields": [
+                {"id": 0, "name": "id", "required": false, "type": "long"},
+                {"id": 0, "name": "payload", "required": false, "type": "string"}
+            ]
+        }
+    }))
+    .expect("create table request should parse");
+
+    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect_err("duplicate temporary field IDs must be rejected");
+
+    assert_eq!(error.message(), Some("duplicate create schema field id 0"));
+}
+
+#[test]
+fn create_table_rejects_excessive_schema_nesting() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let mut field_type = serde_json::Value::from("long");
+    for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
+        field_type = serde_json::json!({
+            "type": "list",
+            "element-id": element_id,
+            "element-required": false,
+            "element": field_type
+        });
+    }
+    let request = CreateTableRequest {
+        name: "events".to_string(),
+        location: None,
+        schema: serde_json::json!({
+            "type": "struct",
+            "fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
+        }),
+        partition_spec: None,
+        write_order: None,
+        stage_create: false,
+        properties: BTreeMap::new(),
+    };
+
+    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect_err("excessively nested create schemas must be rejected");
+
+    assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
+}
+
 #[test]
 fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2247,7 +2443,11 @@ fn create_table_counts_collection_ids_in_last_column_id() {
     let (_, metadata) =
         table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
 
-    assert_eq!(metadata["last-column-id"], 9);
+    let schema = &metadata["schemas"][0];
+    assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
+    assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
+    assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
+    assert_eq!(metadata["last-column-id"], 5);
 }
 
 #[test]

rustfs/src/table_catalog/iceberg/validation.rs

@@ -19,6 +19,7 @@ use futures::{StreamExt, TryStreamExt, stream};
 use super::super::*;
 
 const ICEBERG_MAX_USER_FIELD_ID: i32 = i32::MAX - 200;
+pub(crate) const ICEBERG_MAX_SCHEMA_NESTING_DEPTH: usize = 128;
 
 fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option<usize>) -> TableCatalogStoreResult<String> {
     let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix);
@@ -1361,6 +1362,188 @@ fn validate_iceberg_schema(schema: &serde_json::Value, label: &str) -> TableCata
     Ok(validate_iceberg_schema_fields(schema, label)?.field_ids)
 }
 
+pub(crate) fn assign_fresh_create_schema_ids(
+    schema: &mut serde_json::Value,
+    partition_spec: Option<&mut serde_json::Value>,
+    sort_order: Option<&mut serde_json::Value>,
+) -> TableCatalogStoreResult<()> {
+    let mut assigner = FreshCreateSchemaIdAssigner::new();
+    assigner.assign_schema(schema)?;
+    assigner.remap_identifier_field_ids(schema)?;
+    if let Some(partition_spec) = partition_spec {
+        assigner.remap_source_ids(partition_spec, "partition spec")?;
+    }
+    if let Some(sort_order) = sort_order {
+        assigner.remap_source_ids(sort_order, "sort order")?;
+    }
+    Ok(())
+}
+
+struct FreshCreateSchemaIdAssigner {
+    next_id: i32,
+    old_to_new: BTreeMap<i32, i32>,
+}
+
+impl FreshCreateSchemaIdAssigner {
+    fn new() -> Self {
+        Self {
+            next_id: 1,
+            old_to_new: BTreeMap::new(),
+        }
+    }
+
+    fn assign_schema(&mut self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
+        let schema = schema
+            .as_object_mut()
+            .ok_or_else(|| TableCatalogStoreError::Invalid("create schema must be a JSON object".to_string()))?;
+        if schema.get("type").and_then(serde_json::Value::as_str) != Some("struct") {
+            return Err(TableCatalogStoreError::Invalid("create schema type must be struct".to_string()));
+        }
+        let fields = schema
+            .get_mut("fields")
+            .and_then(serde_json::Value::as_array_mut)
+            .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be an array".to_string()))?;
+        self.assign_struct_fields(fields, 0)
+    }
+
+    fn assign_struct_fields(&mut self, fields: &mut [serde_json::Value], depth: usize) -> TableCatalogStoreResult<()> {
+        for field in fields.iter_mut() {
+            let field = field
+                .as_object_mut()
+                .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
+            self.assign_object_id(field, "id", "create schema field id")?;
+        }
+        for field in fields {
+            let field = field
+                .as_object_mut()
+                .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?;
+            let field_type = field
+                .get_mut("type")
+                .ok_or_else(|| TableCatalogStoreError::Invalid("create schema field type is required".to_string()))?;
+            self.assign_type_ids(field_type, depth)?;
+        }
+        Ok(())
+    }
+
+    fn assign_type_ids(&mut self, field_type: &mut serde_json::Value, depth: usize) -> TableCatalogStoreResult<()> {
+        if field_type.is_string() {
+            return Ok(());
+        }
+        if depth >= ICEBERG_MAX_SCHEMA_NESTING_DEPTH {
+            return Err(TableCatalogStoreError::Invalid(
+                "create schema exceeds the maximum nesting depth".to_string(),
+            ));
+        }
+        let nested_depth = depth + 1;
+        let field_type = field_type.as_object_mut().ok_or_else(|| {
+            TableCatalogStoreError::Invalid("create schema field type must be a string or JSON object".to_string())
+        })?;
+        match field_type.get("type").and_then(serde_json::Value::as_str) {
+            Some("struct") => {
+                let fields = field_type
+                    .get_mut("fields")
+                    .and_then(serde_json::Value::as_array_mut)
+                    .ok_or_else(|| TableCatalogStoreError::Invalid("create schema struct fields must be an array".to_string()))?;
+                self.assign_struct_fields(fields, nested_depth)
+            }
+            Some("list") => {
+                self.assign_object_id(field_type, "element-id", "create schema list element-id")?;
+                let element = field_type
+                    .get_mut("element")
+                    .ok_or_else(|| TableCatalogStoreError::Invalid("create schema list element is required".to_string()))?;
+                self.assign_type_ids(element, nested_depth)
+            }
+            Some("map") => {
+                self.assign_object_id(field_type, "key-id", "create schema map key-id")?;
+                self.assign_object_id(field_type, "value-id", "create schema map value-id")?;
+                let key = field_type
+                    .get_mut("key")
+                    .ok_or_else(|| TableCatalogStoreError::Invalid("create schema map key is required".to_string()))?;
+                self.assign_type_ids(key, nested_depth)?;
+                let value = field_type
+                    .get_mut("value")
+                    .ok_or_else(|| TableCatalogStoreError::Invalid("create schema map value is required".to_string()))?;
+                self.assign_type_ids(value, nested_depth)
+            }
+            _ => Err(TableCatalogStoreError::Invalid(
+                "create schema contains an unsupported field type".to_string(),
+            )),
+        }
+    }
+
+    fn assign_object_id(
+        &mut self,
+        object: &mut serde_json::Map<String, serde_json::Value>,
+        field: &str,
+        label: &str,
+    ) -> TableCatalogStoreResult<()> {
+        let old_id = required_i32_value(object, field, label)?;
+        let entry = match self.old_to_new.entry(old_id) {
+            std::collections::btree_map::Entry::Occupied(_) => {
+                return Err(TableCatalogStoreError::Invalid(format!("duplicate create schema field id {old_id}")));
+            }
+            std::collections::btree_map::Entry::Vacant(entry) => entry,
+        };
+        let new_id = self.next_id;
+        if new_id > ICEBERG_MAX_USER_FIELD_ID {
+            return Err(TableCatalogStoreError::Invalid(
+                "create schema exceeds the available Iceberg field ID range".to_string(),
+            ));
+        }
+        self.next_id = new_id.checked_add(1).ok_or_else(|| {
+            TableCatalogStoreError::Invalid("create schema exceeds the available Iceberg field ID range".to_string())
+        })?;
+        entry.insert(new_id);
+        object.insert(field.to_string(), serde_json::Value::from(new_id));
+        Ok(())
+    }
+
+    fn remap_identifier_field_ids(&self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> {
+        let Some(identifier_field_ids) = schema
+            .as_object_mut()
+            .and_then(|schema| schema.get_mut("identifier-field-ids"))
+        else {
+            return Ok(());
+        };
+        let identifier_field_ids = identifier_field_ids
+            .as_array_mut()
+            .ok_or_else(|| TableCatalogStoreError::Invalid("create schema identifier-field-ids must be an array".to_string()))?;
+        for field_id in identifier_field_ids {
+            let old_id = required_i32(field_id, "create schema identifier field id")?;
+            let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
+                TableCatalogStoreError::Invalid(format!(
+                    "create schema identifier field id {old_id} does not reference a schema field"
+                ))
+            })?;
+            *field_id = serde_json::Value::from(*new_id);
+        }
+        Ok(())
+    }
+
+    fn remap_source_ids(&self, value: &mut serde_json::Value, label: &str) -> TableCatalogStoreResult<()> {
+        let value = value
+            .as_object_mut()
+            .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} must be a JSON object")))?;
+        let Some(fields) = value.get_mut("fields") else {
+            return Ok(());
+        };
+        let fields = fields
+            .as_array_mut()
+            .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be an array")))?;
+        for field in fields {
+            let field = field
+                .as_object_mut()
+                .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be JSON objects")))?;
+            let old_id = required_i32_value(field, "source-id", &format!("{label} source-id"))?;
+            let new_id = self.old_to_new.get(&old_id).ok_or_else(|| {
+                TableCatalogStoreError::Invalid(format!("{label} source-id {old_id} does not reference the create schema"))
+            })?;
+            field.insert("source-id".to_string(), serde_json::Value::from(*new_id));
+        }
+        Ok(())
+    }
+}
+
 fn validate_iceberg_schema_fields(schema: &serde_json::Value, label: &str) -> TableCatalogStoreResult<IcebergSchemaFields> {
     let schema = schema
         .as_object()