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/scanner.rs

@@ -16,8 +16,8 @@ use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_scanner_metrics_report;
 use crate::auth::{check_key_valid, get_session_token};
-use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
+use crate::startup_background::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use chrono::Utc;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};

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/lib.rs

@@ -88,7 +88,6 @@ pub mod inspect;
 pub(crate) mod kms_deletion_gate;
 pub mod license;
 pub mod memory_observability;
-pub mod module_switches;
 pub mod profiling;
 #[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))]
 pub mod protocols;

rustfs/src/module_switches.rs

@@ -1,68 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Layer-neutral module switches (backlog#1834).
-//!
-//! Whether the scanner, heal, audit and notify modules are on is read from the
-//! infra layer (storage helpers, node-service RPC) and from the interface layer
-//! (admin handlers), but the switches used to live in `startup_background`
-//! (composition) and `server` (interface). Every lower-layer read was therefore
-//! an upward edge that had to be baselined by the layer-dependency guard.
-//!
-//! The env-derived scanner/heal predicates and the audit/notify state cells now
-//! live here, at the bottom of the layer order, so those reads are ordinary
-//! downward edges. Resolving the audit/notify state still needs server-side
-//! configuration, so `server::refresh_audit_module_enabled` and its notify twin
-//! keep that logic and publish the result through the setters below.
-
-use rustfs_utils::get_env_bool_with_aliases;
-use std::sync::atomic::{AtomicBool, Ordering};
-
-pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
-pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
-pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
-pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
-
-static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
-static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
-
-/// Whether the data scanner is enabled, defaulting to on.
-pub(crate) fn scanner_enabled_from_env() -> bool {
-    get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
-}
-
-/// Whether background heal is enabled, defaulting to on.
-pub(crate) fn heal_enabled_from_env() -> bool {
-    get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
-}
-
-/// Last published audit-module state.
-pub fn is_audit_module_enabled() -> bool {
-    AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
-}
-
-/// Publish the audit-module state resolved by `server::refresh_audit_module_enabled`.
-pub(crate) fn set_audit_module_enabled(enabled: bool) {
-    AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
-}
-
-/// Last published notify-module state.
-pub fn is_notify_module_enabled() -> bool {
-    NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
-}
-
-/// Publish the notify-module state resolved by `server::refresh_notify_module_enabled`.
-pub(crate) fn set_notify_module_enabled(enabled: bool) {
-    NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
-}

rustfs/src/server/audit.rs

@@ -19,8 +19,11 @@ use super::{
 use crate::runtime_sources::AppContext;
 use rustfs_audit::{AuditError, AuditResult, audit_system, init_audit_system, system::AuditSystemState};
 use std::collections::HashSet;
+use std::sync::atomic::{AtomicBool, Ordering};
 use tracing::{info, warn};
 
+static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
+
 fn server_config_from_context() -> Option<rustfs_config::server_config::Config> {
     runtime_sources::current_server_config()
 }
@@ -34,11 +37,13 @@ fn server_config_for_context(context: Option<&AppContext>) -> Option<rustfs_conf
 
 pub fn refresh_audit_module_enabled() -> bool {
     let enabled = resolve_audit_module_state().enabled;
-    crate::module_switches::set_audit_module_enabled(enabled);
+    AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
     enabled
 }
 
-pub use crate::module_switches::is_audit_module_enabled;
+pub fn is_audit_module_enabled() -> bool {
+    AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
+}
 
 fn has_any_persisted_audit_targets(config: &rustfs_config::server_config::Config) -> bool {
     for &subsystem in rustfs_config::audit::AUDIT_SUB_SYSTEMS {

rustfs/src/server/event.rs

@@ -34,6 +34,7 @@ use tokio::time::{Instant, MissedTickBehavior};
 use tokio_util::sync::CancellationToken;
 use tracing::{info, instrument, warn};
 
+static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
 static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false);
 static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false);
 static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new();
@@ -69,11 +70,13 @@ fn should_reconcile_bucket_notification_rules(runtime_changed: bool, notify_enab
 
 pub fn refresh_notify_module_enabled() -> bool {
     let enabled = resolve_notify_module_state().enabled;
-    crate::module_switches::set_notify_module_enabled(enabled);
+    NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
     enabled
 }
 
-pub use crate::module_switches::is_notify_module_enabled;
+pub fn is_notify_module_enabled() -> bool {
+    NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
+}
 
 pub(crate) use crate::shared_types::convert_ecstore_object_info;
 
@@ -168,7 +171,7 @@ pub(crate) async fn reconcile_event_notifier_from_store(
         let transition_system = system.clone();
         let transition_store = store.clone();
         let transition = with_refreshed_notify_module_state_from(store.clone(), move |resolution| async move {
-            crate::module_switches::set_notify_module_enabled(resolution.enabled);
+            NOTIFY_MODULE_ENABLED.store(resolution.enabled, Ordering::Relaxed);
             let read_store = transition_store.clone();
             let config_system = transition_system.clone();
             with_server_config_read_lock(transition_store, move || async move {

rustfs/src/startup_background.rs

@@ -12,20 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
 use crate::storage_api::startup::background::{ECStore, set_workload_admission_snapshot_provider};
 use crate::workload_admission::RustFsWorkloadAdmissionSnapshotProvider;
 use rustfs_concurrency::WorkloadAdmissionSnapshotProvider;
 use rustfs_heal::{
     create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager_with_workload_provider,
 };
+use rustfs_utils::get_env_bool_with_aliases;
 use std::{io::Result, sync::Arc};
 use tracing::{debug, info};
 
+pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
+pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
+pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
+pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
 const LOG_COMPONENT_MAIN: &str = "main";
 const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured";
 
+pub(crate) fn scanner_enabled_from_env() -> bool {
+    get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
+}
+
+pub(crate) fn heal_enabled_from_env() -> bool {
+    get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
+}
+
 pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Result<bool> {
     let _ = create_ahm_services_cancel_token();
 

rustfs/src/storage/helper.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::module_switches::{is_audit_module_enabled, is_notify_module_enabled};
+use crate::server::{is_audit_module_enabled, is_notify_module_enabled};
 use crate::shared_types::convert_ecstore_object_info;
 use crate::storage::access::{ReqInfo, request_context_from_req};
 use crate::storage::request_context::RequestContext;

rustfs/src/storage/rpc/node_service/heal.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::module_switches::{heal_enabled_from_env, scanner_enabled_from_env};
+use crate::startup_background::{heal_enabled_from_env, scanner_enabled_from_env};
 use crate::storage::storage_api::runtime_sources_consumer::EndpointServerPools;
 use jiff::Timestamp;
 use rmp_serde::Deserializer;

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()

scripts/layer-dependency-baseline.txt

@@ -16,7 +16,11 @@
 #   cycle|left_layer<->right_layer
 cycle|app<->infra
 cycle|app<->interface
+cycle|composition<->infra
+cycle|composition<->interface
 cycle|infra<->interface
+dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::ENV_SCANNER_ENABLED
+dep|rustfs/src/admin/handlers/scanner.rs|interface->composition|crate::startup_background::scanner_enabled_from_env
 dep|rustfs/src/app/admin_usecase.rs|app->interface|crate::server::collect_dependency_readiness_report
 dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_bucket_meta_hook
 dep|rustfs/src/app/bucket_usecase.rs|app->interface|crate::admin::handlers::site_replication::site_replication_delete_bucket_hook
@@ -25,6 +29,8 @@ dep|rustfs/src/cluster_snapshot.rs|infra->interface|crate::server::snapshot_depe
 dep|rustfs/src/runtime_sources.rs|infra->app|crate::app::context
 dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::server::cors
 dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::storage::ecfs::ListObjectUnorderedQuery
+dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_audit_module_enabled
+dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::is_notify_module_enabled
 dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_audit_module_enabled
 dep|rustfs/src/storage/helper.rs|infra->interface|crate::server::refresh_notify_module_enabled
 dep|rustfs/src/storage/rpc/http_service.rs|infra->interface|crate::server::RPC_PREFIX
@@ -34,3 +40,5 @@ dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::servic
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::config::reload_runtime_config_snapshot
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::admin::service::site_replication::reload_site_replication_runtime_state
 dep|rustfs/src/storage/rpc/node_service.rs|infra->interface|crate::server::MODULE_SWITCHES_SIGNAL_SUBSYSTEM
+dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::heal_enabled_from_env
+dep|rustfs/src/storage/rpc/node_service/heal.rs|infra->composition|crate::startup_background::scanner_enabled_from_env