diff --git a/src/mito2/src/engine/index_build_test.rs b/src/mito2/src/engine/index_build_test.rs
index a00b4798a03..672c5e472d5 100644
--- a/src/mito2/src/engine/index_build_test.rs
+++ b/src/mito2/src/engine/index_build_test.rs
@@ -1147,7 +1147,10 @@ async fn test_consecutive_schema_changes_publish_latest_index_generation() {
     let files = current_file_metas(&engine, region_id).await;
     assert_eq!(files.len(), 1);
     assert!(
-        files[0].is_index_consistent_with_region(&version.metadata.column_metadatas),
+        files[0].is_index_consistent_with_region(
+            &version.metadata.column_metadatas,
+            &version.options.index_options,
+        ),
         "the published index must match the latest schema generation, file: {:?}, metadata: {:?}",
         files[0],
         version.metadata
@@ -1166,10 +1169,10 @@ async fn test_consecutive_schema_changes_publish_latest_index_generation() {
     let reopened = engine.get_region(region_id).unwrap();
     let reopened_files = current_file_metas(&engine, region_id).await;
     assert_eq!(reopened_files.len(), 1);
-    assert!(
-        reopened_files[0]
-            .is_index_consistent_with_region(&reopened.version().metadata.column_metadatas)
-    );
+    assert!(reopened_files[0].is_index_consistent_with_region(
+        &reopened.version().metadata.column_metadatas,
+        &reopened.version().options.index_options,
+    ));
 }
 
 #[tokio::test]
@@ -1250,7 +1253,184 @@ async fn test_unrelated_schema_change_retries_stale_flush_index_build() {
     let region = engine.get_region(region_id).unwrap();
     let files = current_file_metas(&engine, region_id).await;
     assert_eq!(files.len(), 1);
-    assert!(files[0].is_index_consistent_with_region(&region.version().metadata.column_metadatas));
+    let version = region.version();
+    assert!(files[0].is_index_consistent_with_region(
+        &version.metadata.column_metadatas,
+        &version.options.index_options,
+    ));
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1);
+}
+
+#[tokio::test]
+async fn test_manual_build_waits_for_committed_publication_to_apply() {
+    let mut env =
+        TestEnv::with_prefix("test_manual_build_waits_for_committed_publication_to_apply_").await;
+    let gate = Arc::new(IndexPublicationGate::new(
+        IndexPublicationPhase::AfterManifestCommit,
+    ));
+    let engine = Arc::new(
+        env.create_engine_with(
+            async_build_mode_config(true),
+            None,
+            Some(gate.clone()),
+            None,
+        )
+        .await,
+    );
+
+    let region_id = RegionId::new(1, 1);
+    env.get_schema_metadata_manager()
+        .register_region_table_info(
+            region_id.table_id(),
+            "test_table",
+            "test_catalog",
+            "test_schema",
+            None,
+            env.get_kv_backend(),
+        )
+        .await;
+
+    let request = CreateRequestBuilder::new().build_with_index();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    put_and_flush(&engine, region_id, &column_schemas, 0..20).await;
+    // The flush build has committed its index to the manifest, but the worker
+    // has not applied it to the version.
+    tokio::time::timeout(Duration::from_secs(10), gate.wait_entered(1))
+        .await
+        .expect("flush index build did not commit the manifest");
+
+    let engine_for_build = engine.clone();
+    let mut manual_build = tokio::spawn(async move {
+        engine_for_build
+            .handle_request(
+                region_id,
+                RegionRequest::BuildIndex(RegionBuildIndexRequest::default()),
+            )
+            .await
+    });
+    assert!(
+        tokio::time::timeout(Duration::from_millis(300), &mut manual_build)
+            .await
+            .is_err(),
+        "manual build returned before the committed index was applied"
+    );
+
+    gate.release(1);
+    tokio::time::timeout(Duration::from_secs(10), manual_build)
+        .await
+        .expect("manual index build request did not finish")
+        .expect("manual index build task panicked")
+        .unwrap();
+
+    let version = engine.get_region(region_id).unwrap().version();
+    let files = version
+        .ssts
+        .levels()
+        .iter()
+        .flat_map(|level| level.files.values())
+        .collect::<Vec<_>>();
+    assert_eq!(files.len(), 1);
+    assert!(files[0].meta_ref().is_index_consistent_with_region(
+        &version.metadata.column_metadatas,
+        &version.options.index_options,
+    ));
+}
+
+#[tokio::test]
+async fn test_manual_build_waits_for_schema_change_retry() {
+    let mut env = TestEnv::with_prefix("test_manual_build_waits_for_schema_change_retry_").await;
+    let gate = Arc::new(IndexPublicationGate::new(
+        IndexPublicationPhase::BeforeManifestCommit,
+    ));
+    let engine = Arc::new(
+        env.create_engine_with(
+            async_build_mode_config(false),
+            None,
+            Some(gate.clone()),
+            None,
+        )
+        .await,
+    );
+
+    let region_id = RegionId::new(1, 1);
+    env.get_schema_metadata_manager()
+        .register_region_table_info(
+            region_id.table_id(),
+            "test_table",
+            "test_catalog",
+            "test_schema",
+            None,
+            env.get_kv_backend(),
+        )
+        .await;
+
+    let request = CreateRequestBuilder::new().build_with_index();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    put_and_flush(&engine, region_id, &column_schemas, 0..20).await;
+
+    let engine_for_build = engine.clone();
+    let manual_build = tokio::spawn(async move {
+        engine_for_build
+            .handle_request(
+                region_id,
+                RegionRequest::BuildIndex(RegionBuildIndexRequest::default()),
+            )
+            .await
+    });
+    tokio::time::timeout(Duration::from_secs(10), gate.wait_entered(1))
+        .await
+        .expect("manual index build did not reach manifest publication");
+
+    // An unrelated schema change makes the manual publication stale.
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Alter(RegionAlterRequest {
+                kind: AlterKind::AddColumns {
+                    columns: vec![AddColumn {
+                        column_metadata: ColumnMetadata {
+                            column_schema: ColumnSchema::new(
+                                "field_1",
+                                ConcreteDataType::float64_datatype(),
+                                true,
+                            ),
+                            semantic_type: SemanticType::Field,
+                            column_id: 3,
+                        },
+                        location: None,
+                    }],
+                },
+            }),
+        )
+        .await
+        .unwrap();
+
+    gate.release(1);
+    tokio::time::timeout(Duration::from_secs(10), gate.wait_entered(2))
+        .await
+        .expect("stale manual index build was not retried");
+    // The caller waits for the retry rather than the aborted attempt.
+    assert!(!manual_build.is_finished());
+
+    gate.release(1);
+    tokio::time::timeout(Duration::from_secs(10), manual_build)
+        .await
+        .expect("manual index build request did not finish")
+        .expect("manual index build task panicked")
+        .unwrap();
+
     let scanner = engine
         .scanner(region_id, ScanRequest::default())
         .await
@@ -1339,7 +1519,10 @@ async fn test_schema_change_uses_manifest_file_generation() {
     let files = current_file_metas(&engine, region_id).await;
     assert_eq!(files.len(), 1);
     assert!(
-        files[0].is_index_consistent_with_region(&version.metadata.column_metadatas),
+        files[0].is_index_consistent_with_region(
+            &version.metadata.column_metadatas,
+            &version.options.index_options,
+        ),
         "the latest index definition was not published"
     );
 }
@@ -1593,36 +1776,31 @@ async fn test_index_build_type_manual_duplicate_in_flight() {
     assert_eq!(gate.finish_count(), 0);
     assert_eq!(gate.abort_count(), 0);
 
-    // Issue the second manual BuildIndex for the same region/file.
-    // Since the file is already in building_files (from the first manual build),
-    // schedule_build detects the duplicate and calls on_index_build_abort.
-    let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default());
-    engine.handle_request(region_id, request).await.unwrap();
-
-    // The second request should have been aborted as duplicate.
-    assert_eq!(gate.abort_count(), 1, "duplicate request should be aborted");
-    assert_eq!(gate.begin_count(), 2, "no new begin for duplicate");
-    assert_eq!(gate.finish_count(), 0, "first build hasn't finished yet");
-
-    // Release the gate to let the first manual build proceed.
-    gate.release_begin();
-    tokio::time::timeout(std::time::Duration::from_secs(5), gate.wait_finish(1))
+    // The second manual BuildIndex for the same file waits for the active build
+    // instead of returning before the index exists.
+    let received = gate.recv_count();
+    let engine_clone = engine.clone();
+    let second_handle = tokio::spawn(async move {
+        let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default());
+        engine_clone.handle_request(region_id, request).await
+    });
+    tokio::time::timeout(Duration::from_secs(5), gate.wait_recv(received + 1))
         .await
-        .unwrap(); // first manual build completes
-
-    // Final counts: only one successful build, one aborted duplicate.
-    assert_eq!(gate.begin_count(), 2); // flush + first manual
-    assert_eq!(gate.finish_count(), 1); // first manual only
-    assert_eq!(gate.abort_count(), 1); // second manual duplicate abort
-
-    // Await the first build's handle_request to ensure it completed cleanly.
-    tokio::time::timeout(std::time::Duration::from_secs(5), first_handle)
-        .await
-        .unwrap()
-        .unwrap()
         .unwrap();
 
-    // Verify exactly one SST and one index file.
+    // Release the first manual build and the second one queued behind it.
+    gate.release_begin();
+    gate.release_begin();
+    for handle in [first_handle, second_handle] {
+        tokio::time::timeout(Duration::from_secs(5), handle)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+    }
+
+    // The queued build starts from the published index and skips the rebuild.
+    assert_eq!(gate.finish_count(), 1);
     let scanner = engine
         .scanner(region_id, ScanRequest::default())
         .await
@@ -1631,6 +1809,162 @@ async fn test_index_build_type_manual_duplicate_in_flight() {
     assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1);
 }
 
+#[tokio::test]
+async fn test_index_build_type_manual_waits_for_in_flight_flush_build() {
+    let mut env =
+        TestEnv::with_prefix("test_index_build_type_manual_waits_for_in_flight_flush_build_").await;
+    let gate = Arc::new(GateIndexBuildListener::default());
+    let engine = Arc::new(
+        env.create_engine_with(
+            async_build_mode_config(false), // Flush-triggered builds create no index.
+            None,
+            Some(gate.clone()),
+            None,
+        )
+        .await,
+    );
+
+    let region_id = RegionId::new(1, 1);
+    env.get_schema_metadata_manager()
+        .register_region_table_info(
+            region_id.table_id(),
+            "test_table",
+            "test_catalog",
+            "test_schema",
+            None,
+            env.get_kv_backend(),
+        )
+        .await;
+
+    let request = CreateRequestBuilder::new().build_with_index();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    put_and_flush(&engine, region_id, &column_schemas, 10..20).await;
+    tokio::time::timeout(Duration::from_secs(5), gate.wait_begin(1))
+        .await
+        .unwrap();
+
+    // The manual build arrives while the no-op flush build of the same file is
+    // active. Returning at this point would leave the file without an index.
+    let received = gate.recv_count();
+    let engine_clone = engine.clone();
+    let manual_handle = tokio::spawn(async move {
+        let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default());
+        engine_clone.handle_request(region_id, request).await
+    });
+    tokio::time::timeout(Duration::from_secs(5), gate.wait_recv(received + 1))
+        .await
+        .unwrap();
+
+    gate.release_begin();
+    gate.release_begin();
+    tokio::time::timeout(Duration::from_secs(5), manual_handle)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+
+    // The index is visible to scans as soon as the manual build returns.
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    assert_eq!(scanner.num_files(), 1);
+    assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1);
+}
+
+#[tokio::test]
+async fn test_manual_build_treats_compacted_files_as_done() {
+    let mut env = TestEnv::with_prefix("test_manual_build_treats_compacted_files_as_done_").await;
+    let gate = Arc::new(GateIndexBuildListener::default());
+    let mut config = async_build_mode_config(false);
+    config.inverted_index.create_on_compaction = Mode::Disable;
+    config.fulltext_index.create_on_compaction = Mode::Disable;
+    config.bloom_filter_index.create_on_compaction = Mode::Disable;
+    // The builds of both SSTs must block at the gate together; the default
+    // limit depends on the CPU count and can be 1.
+    config.max_background_index_builds = 2;
+    let engine = Arc::new(
+        env.create_engine_with(config, None, Some(gate.clone()), None)
+            .await,
+    );
+
+    let region_id = RegionId::new(1, 1);
+    env.get_schema_metadata_manager()
+        .register_region_table_info(
+            region_id.table_id(),
+            "test_table",
+            "test_catalog",
+            "test_schema",
+            None,
+            env.get_kv_backend(),
+        )
+        .await;
+
+    let request = CreateRequestBuilder::new().build_with_index();
+    let column_schemas = rows_schema(&request);
+    engine
+        .handle_request(region_id, RegionRequest::Create(request))
+        .await
+        .unwrap();
+    put_and_flush(&engine, region_id, &column_schemas, 10..20).await;
+    put_and_flush(&engine, region_id, &column_schemas, 20..30).await;
+    // The flush builds create no index.
+    tokio::time::timeout(Duration::from_secs(5), gate.wait_begin(2))
+        .await
+        .unwrap();
+    gate.release_begin();
+    gate.release_begin();
+
+    let engine_for_build = engine.clone();
+    let manual_build = tokio::spawn(async move {
+        engine_for_build
+            .handle_request(
+                region_id,
+                RegionRequest::BuildIndex(RegionBuildIndexRequest::default()),
+            )
+            .await
+    });
+    tokio::time::timeout(Duration::from_secs(5), gate.wait_begin(4))
+        .await
+        .unwrap();
+
+    // Compaction replaces both SSTs while their manual builds are blocked.
+    engine
+        .handle_request(
+            region_id,
+            RegionRequest::Compact(RegionCompactRequest {
+                options: compact_request::Options::StrictWindow(StrictWindow {
+                    window_seconds: 3600,
+                }),
+                parallelism: None,
+                time_range: None,
+            }),
+        )
+        .await
+        .unwrap();
+    for _ in 0..3 {
+        gate.release_begin();
+    }
+
+    // The build covers the SSTs that existed when it was requested, so the
+    // unindexed compaction output does not fail it.
+    tokio::time::timeout(Duration::from_secs(5), manual_build)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    let scanner = engine
+        .scanner(region_id, ScanRequest::default())
+        .await
+        .unwrap();
+    assert_eq!(scanner.num_files(), 1);
+    assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 0);
+}
+
 #[tokio::test]
 async fn test_reopen_waits_for_active_index_build_of_previous_incarnation() {
     let mut env =
@@ -1723,12 +2057,18 @@ async fn test_reopen_waits_for_active_index_build_of_previous_incarnation() {
     assert_eq!(gate.begin_count(), 2);
     assert_eq!(gate.abort_count(), aborts_before_rebuild);
 
+    // The closed incarnation cannot publish its build, so its caller must not
+    // be told the index exists.
     gate.release_begin();
-    tokio::time::timeout(Duration::from_secs(10), old_build)
+    let err = tokio::time::timeout(Duration::from_secs(10), old_build)
         .await
         .expect("old index build request did not finish")
         .expect("old index build task panicked")
-        .unwrap();
+        .unwrap_err();
+    assert!(
+        err.to_string().contains("Index build did not complete"),
+        "{err:?}"
+    );
     tokio::time::timeout(Duration::from_secs(10), gate.wait_begin(3))
         .await
         .expect("reopened index build did not start after the old build stopped");
@@ -2058,14 +2398,16 @@ async fn test_sync_flush_indexes_encoded_bulk_parts(
         return;
     }
 
-    let metadata = engine.get_region(region_id).unwrap().metadata();
+    let version = engine.get_region(region_id).unwrap().version();
     let indexed = async {
         loop {
             let files = current_file_metas(&engine, region_id).await;
-            if files
-                .iter()
-                .all(|f| f.is_index_consistent_with_region(&metadata.column_metadatas))
-            {
+            if files.iter().all(|f| {
+                f.is_index_consistent_with_region(
+                    &version.metadata.column_metadatas,
+                    &version.options.index_options,
+                )
+            }) {
                 return files;
             }
             tokio::time::sleep(Duration::from_millis(20)).await;
diff --git a/src/mito2/src/error.rs b/src/mito2/src/error.rs
index ab781f790f4..d7ecdd88b1b 100644
--- a/src/mito2/src/error.rs
+++ b/src/mito2/src/error.rs
@@ -775,6 +775,18 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display(
+        "Index build did not complete for files {:?} in region {}",
+        file_ids,
+        region_id
+    ))]
+    IndexBuildIncomplete {
+        region_id: RegionId,
+        file_ids: Vec<FileId>,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Failed to convert value"))]
     ConvertValue {
         source: datatypes::error::Error,
@@ -1561,6 +1573,7 @@ impl ErrorExt for Error {
                 StatusCode::StorageUnavailable
             }
             FlushRegion { source, .. } | BuildIndexAsync { source, .. } => source.status_code(),
+            IndexBuildIncomplete { .. } => StatusCode::IllegalState,
             RegionDropped { .. } => StatusCode::Cancelled,
             RegionClosed { .. } => StatusCode::Cancelled,
             StaleCompactionExecution { .. } => StatusCode::Cancelled,
diff --git a/src/mito2/src/request.rs b/src/mito2/src/request.rs
index 5b9aff4f82d..2e81fbc4396 100644
--- a/src/mito2/src/request.rs
+++ b/src/mito2/src/request.rs
@@ -65,7 +65,7 @@ use crate::metrics::COMPACTION_ELAPSED_TOTAL;
 use crate::region::MitoRegionRef;
 use crate::region::options::RegionOptions;
 use crate::sst::file::FileMeta;
-use crate::sst::index::IndexBuildType;
+use crate::sst::index::{IndexBuildType, ResultMpscSender};
 use crate::wal::EntryId;
 use crate::wal::entry_distributor::WalEntryReceiver;
 
@@ -1113,8 +1113,6 @@ pub(crate) enum BackgroundNotify {
     IndexBuildFinished(IndexBuildFinished),
     /// Index build has been stopped (aborted or succeeded).
     IndexBuildStopped(IndexBuildStopped),
-    /// Index build has failed.
-    IndexBuildFailed(IndexBuildFailed),
     /// An index build must be retried against the latest schema generation.
     IndexBuildRetry(BuildIndexRequest),
     /// Compaction has finished.
@@ -1196,6 +1194,8 @@ impl FlushFailed {
 pub(crate) struct IndexBuildFinished {
     pub(crate) manifest_version: ManifestVersion,
     pub(crate) file_meta: FileMeta,
+    /// Waiters of the build, notified after the worker applies the edit.
+    pub(crate) result_senders: Vec<ResultMpscSender>,
 }
 
 /// Notifies an index build job has been stopped.
@@ -1204,12 +1204,6 @@ pub(crate) struct IndexBuildStopped {
     pub(crate) file_id: FileId,
 }
 
-/// Notifies an index build job has failed.
-#[derive(Debug)]
-pub(crate) struct IndexBuildFailed {
-    pub(crate) err: Arc<Error>,
-}
-
 /// A local unit event carries only identity, never an input reservation lease.
 #[derive(Debug)]
 pub(crate) enum CompactionUnitNotification {
@@ -1419,6 +1413,8 @@ pub(crate) struct BuildIndexRequest {
     pub(crate) build_type: IndexBuildType,
     /// files need to build index, empty means all.
     pub(crate) file_metas: Vec<FileMeta>,
+    /// Waiters carried over from a previous build of the same files.
+    pub(crate) result_senders: Vec<ResultMpscSender>,
 }
 
 #[derive(Debug)]
diff --git a/src/mito2/src/sst/file.rs b/src/mito2/src/sst/file.rs
index 5f4dcb28752..28b6218cd7b 100644
--- a/src/mito2/src/sst/file.rs
+++ b/src/mito2/src/sst/file.rs
@@ -37,6 +37,7 @@ use store_api::storage::{ColumnId, FileId, IndexVersion, RegionId};
 use crate::access_layer::AccessLayerRef;
 use crate::cache::CacheManagerRef;
 use crate::cache::file_cache::{FileType, IndexKey};
+use crate::region::options::IndexOptions;
 use crate::sst::file_purger::FilePurgerRef;
 use crate::sst::index::fulltext_index::FULLTEXT_BLOOM_VERSION;
 use crate::sst::location;
@@ -456,56 +457,73 @@ impl FileMeta {
         self.index_file_size
     }
 
-    /// Check whether the file index is consistent with the given region metadata.
-    pub fn is_index_consistent_with_region(&self, metadata: &[ColumnMetadata]) -> bool {
+    /// Check whether the file index is consistent with the given region metadata
+    /// and index options.
+    pub fn is_index_consistent_with_region(
+        &self,
+        metadata: &[ColumnMetadata],
+        index_options: &IndexOptions,
+    ) -> bool {
         let id_to_indexes = self
             .indexes
             .iter()
             .map(|index| (index.column_id, index))
             .collect::<std::collections::HashMap<_, _>>();
+        let inverted_options = &index_options.inverted_index;
         for column in metadata {
-            if !column.column_schema.is_indexed() {
+            let schema = &column.column_schema;
+            // The index builder never creates these inverted indexes, so expecting
+            // them would make every manual build rebuild the file again.
+            let expects_inverted = schema.is_inverted_indexed()
+                && inverted_options.segment_row_count > 0
+                && !inverted_options
+                    .ignore_column_ids
+                    .contains(&column.column_id);
+            let expects_fulltext = schema.is_fulltext_indexed();
+            let expects_skipping = schema.is_skipping_indexed();
+            if !expects_inverted && !expects_fulltext && !expects_skipping {
                 continue;
             }
-            if let Some(index) = id_to_indexes.get(&column.column_id) {
-                let indexes = &index.created_indexes;
-                if column.column_schema.is_inverted_indexed()
-                    && !indexes.contains(&IndexType::InvertedIndex)
-                {
-                    return false;
-                }
-                if column.column_schema.is_fulltext_indexed()
-                    && !indexes.contains(&IndexType::FulltextIndex)
-                {
-                    return false;
-                }
-                if column
-                    .column_schema
-                    .fulltext_options()
-                    .ok()
-                    .flatten()
-                    .is_some_and(|options| {
-                        options.enable && options.backend == FulltextBackend::Bloom
-                    })
-                    && index.fulltext_bloom_version < FULLTEXT_BLOOM_VERSION
-                {
-                    debug!(
-                        "Fulltext bloom index is outdated, region_id: {}, file_id: {}, column_id: {}, version: {}, current: {}",
-                        self.region_id,
-                        self.file_id,
-                        column.column_id,
-                        index.fulltext_bloom_version,
-                        FULLTEXT_BLOOM_VERSION
-                    );
-                    return false;
-                }
-                if column.column_schema.is_skipping_indexed()
-                    && !indexes.contains(&IndexType::BloomFilterIndex)
-                {
-                    return false;
-                }
-            } else {
+            let missing_index = |index_type: IndexType| {
+                debug!(
+                    "File index is missing, region_id: {}, file_id: {}, column_id: {}, index_type: {:?}",
+                    self.region_id, self.file_id, column.column_id, index_type
+                );
+                false
+            };
+            let Some(index) = id_to_indexes.get(&column.column_id) else {
+                debug!(
+                    "File has no index for column, region_id: {}, file_id: {}, column_id: {}",
+                    self.region_id, self.file_id, column.column_id
+                );
                 return false;
+            };
+            let indexes = &index.created_indexes;
+            if expects_inverted && !indexes.contains(&IndexType::InvertedIndex) {
+                return missing_index(IndexType::InvertedIndex);
+            }
+            if expects_fulltext && !indexes.contains(&IndexType::FulltextIndex) {
+                return missing_index(IndexType::FulltextIndex);
+            }
+            if schema
+                .fulltext_options()
+                .ok()
+                .flatten()
+                .is_some_and(|options| options.enable && options.backend == FulltextBackend::Bloom)
+                && index.fulltext_bloom_version < FULLTEXT_BLOOM_VERSION
+            {
+                debug!(
+                    "Fulltext bloom index is outdated, region_id: {}, file_id: {}, column_id: {}, version: {}, current: {}",
+                    self.region_id,
+                    self.file_id,
+                    column.column_id,
+                    index.fulltext_bloom_version,
+                    FULLTEXT_BLOOM_VERSION
+                );
+                return false;
+            }
+            if expects_skipping && !indexes.contains(&IndexType::BloomFilterIndex) {
+                return missing_index(IndexType::BloomFilterIndex);
             }
         }
         true
@@ -1379,7 +1397,7 @@ mod tests {
             ..Default::default()
         };
         let region_meta = vec![new_column_meta(1, "tag1", true, false, false)];
-        assert!(file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 2: Superset match. File has more indexes than required.
         file_meta.indexes = vec![ColumnIndexMetadata {
@@ -1391,7 +1409,7 @@ mod tests {
             ..Default::default()
         }];
         let region_meta = vec![new_column_meta(1, "tag1", true, false, false)];
-        assert!(file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 3: Missing index type. File has the column but lacks the required index type.
         file_meta.indexes = vec![ColumnIndexMetadata {
@@ -1400,7 +1418,7 @@ mod tests {
             ..Default::default()
         }];
         let region_meta = vec![new_column_meta(1, "tag1", true, true, false)]; // Requires fulltext too
-        assert!(!file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(!file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 4: Missing column. Region requires an index on a column not in the file's index list.
         file_meta.indexes = vec![ColumnIndexMetadata {
@@ -1409,7 +1427,7 @@ mod tests {
             ..Default::default()
         }];
         let region_meta = vec![new_column_meta(1, "tag1", true, false, false)]; // Requires index on column 1
-        assert!(!file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(!file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 5: No indexes required by region. Should always be consistent.
         file_meta.indexes = vec![ColumnIndexMetadata {
@@ -1418,12 +1436,12 @@ mod tests {
             ..Default::default()
         }];
         let region_meta = vec![new_column_meta(1, "tag1", false, false, false)]; // No index required
-        assert!(file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 6: Empty file indexes. Region requires an index.
         file_meta.indexes = vec![];
         let region_meta = vec![new_column_meta(1, "tag1", true, false, false)];
-        assert!(!file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(!file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
 
         // Case 7: Multiple columns, one is inconsistent.
         file_meta.indexes = vec![
@@ -1442,7 +1460,17 @@ mod tests {
             new_column_meta(1, "tag1", true, false, false),
             new_column_meta(2, "tag2", false, true, true), // Requires Fulltext and BloomFilter
         ];
-        assert!(!file_meta.is_index_consistent_with_region(&region_meta));
+        assert!(!file_meta.is_index_consistent_with_region(&region_meta, &IndexOptions::default()));
+
+        // Case 8: Inverted indexes the builder skips by index options are not required.
+        file_meta.indexes = vec![];
+        let region_meta = vec![new_column_meta(1, "tag1", true, false, false)];
+        let mut index_options = IndexOptions::default();
+        index_options.inverted_index.ignore_column_ids = vec![1];
+        assert!(file_meta.is_index_consistent_with_region(&region_meta, &index_options));
+        let mut index_options = IndexOptions::default();
+        index_options.inverted_index.segment_row_count = 0;
+        assert!(file_meta.is_index_consistent_with_region(&region_meta, &index_options));
     }
 
     #[test]
@@ -1475,10 +1503,13 @@ mod tests {
 
         let bloom = fulltext_column(FulltextBackend::Bloom);
         // Written before the version was recorded, i.e. with the old bloom blob.
-        assert!(!file_meta(0).is_index_consistent_with_region(&bloom));
-        assert!(file_meta(FULLTEXT_BLOOM_VERSION).is_index_consistent_with_region(&bloom));
+        assert!(!file_meta(0).is_index_consistent_with_region(&bloom, &IndexOptions::default()));
+        assert!(
+            file_meta(FULLTEXT_BLOOM_VERSION)
+                .is_index_consistent_with_region(&bloom, &IndexOptions::default())
+        );
         // Tantivy indexes carry no bloom version.
         let tantivy = fulltext_column(FulltextBackend::Tantivy);
-        assert!(file_meta(0).is_index_consistent_with_region(&tantivy));
+        assert!(file_meta(0).is_index_consistent_with_region(&tantivy, &IndexOptions::default()));
     }
 }
diff --git a/src/mito2/src/sst/index.rs b/src/mito2/src/sst/index.rs
index 6b90d337202..ff734e3241b 100644
--- a/src/mito2/src/sst/index.rs
+++ b/src/mito2/src/sst/index.rs
@@ -30,6 +30,7 @@ pub(crate) mod store;
 use std::cmp::Ordering;
 use std::collections::{HashMap, HashSet};
 use std::num::NonZeroUsize;
+use std::ops::ControlFlow;
 use std::sync::Arc;
 
 use bloom_filter::creator::BloomFilterIndexer;
@@ -66,8 +67,8 @@ use crate::region::{
     IndexBuildSource, IndexPublication, IndexPublicationStale, ManifestContextRef,
 };
 use crate::request::{
-    BackgroundNotify, BuildIndexRequest, IndexBuildFailed, IndexBuildFinished, IndexBuildStopped,
-    WorkerRequest, WorkerRequestWithTime,
+    BackgroundNotify, BuildIndexRequest, IndexBuildFinished, IndexBuildStopped, WorkerRequest,
+    WorkerRequestWithTime,
 };
 use crate::schedule::scheduler::{Job, SchedulerRef};
 use crate::sst::file::{
@@ -649,6 +650,15 @@ pub enum IndexBuildOutcome {
     Aborted(String),
 }
 
+/// How an index build task reports its result.
+enum IndexBuildReport {
+    /// The task sends the outcome to its waiters.
+    Done(IndexBuildOutcome),
+    /// The worker or a retry task owns the waiters, so the task must not
+    /// report to them.
+    Deferred,
+}
+
 /// Mpsc output result sender.
 pub type ResultMpscSender = Sender<Result<IndexBuildOutcome>>;
 
@@ -668,6 +678,8 @@ pub struct IndexBuildTask {
     pub(crate) target_region_metadata: RegionMetadataRef,
     /// The manifest state this build is based on.
     pub(crate) source: IndexBuildSource,
+    /// Index options of the target region when the task was created.
+    pub(crate) index_options: IndexOptions,
     pub reason: IndexBuildType,
     pub access_layer: AccessLayerRef,
     pub(crate) listener: WorkerListener,
@@ -680,8 +692,9 @@ pub struct IndexBuildTask {
     pub indexer_builder: Arc<dyn IndexerBuilder + Send + Sync>,
     /// Request sender to notify the region worker.
     pub(crate) request_sender: Sender<WorkerRequestWithTime>,
-    /// Index build result sender.
-    pub(crate) result_sender: ResultMpscSender,
+    /// Senders of callers waiting for this build. Coalesced tasks and
+    /// schema-change retries hand their senders over to the surviving task.
+    pub(crate) result_senders: Vec<ResultMpscSender>,
 }
 
 impl std::fmt::Debug for IndexBuildTask {
@@ -699,17 +712,20 @@ impl std::fmt::Debug for IndexBuildTask {
 impl IndexBuildTask {
     /// Notify the caller the job is success.
     pub async fn on_success(&self, outcome: IndexBuildOutcome) {
-        let _ = self.result_sender.send(Ok(outcome)).await;
+        for sender in &self.result_senders {
+            let _ = sender.send(Ok(outcome.clone())).await;
+        }
     }
 
     /// Send index build error to waiter.
     pub async fn on_failure(&self, err: Arc<Error>) {
-        let _ = self
-            .result_sender
-            .send(Err(err.clone()).context(BuildIndexAsyncSnafu {
-                region_id: self.region_id,
-            }))
-            .await;
+        for sender in &self.result_senders {
+            let _ = sender
+                .send(Err(err.clone()).context(BuildIndexAsyncSnafu {
+                    region_id: self.region_id,
+                }))
+                .await;
+        }
     }
 
     fn into_index_build_job(mut self, version_control: VersionControlRef) -> Job {
@@ -720,10 +736,7 @@ impl IndexBuildTask {
 
     async fn do_index_build(&mut self, version_control: VersionControlRef) {
         self.listener
-            .on_index_build_begin(RegionFileId::new(
-                self.source.file_meta.region_id,
-                self.source.file_meta.file_id,
-            ))
+            .on_index_build_begin(self.region_file_id())
             .await;
         let result = if self.reason == IndexBuildType::Compact {
             let mut task = self.clone();
@@ -740,7 +753,13 @@ impl IndexBuildTask {
             self.index_build(version_control).await
         };
         match result {
-            Ok(outcome) => self.on_success(outcome).await,
+            Ok(IndexBuildReport::Done(outcome)) => {
+                if let IndexBuildOutcome::Aborted(reason) = &outcome {
+                    debug!("Index build aborted: {}", reason);
+                }
+                self.on_success(outcome).await
+            }
+            Ok(IndexBuildReport::Deferred) => {}
             Err(e) => {
                 warn!(
                     e; "Index build task failed, region: {}, file_id: {}",
@@ -749,11 +768,23 @@ impl IndexBuildTask {
                 self.on_failure(e.into()).await
             }
         }
+        self.notify_worker(BackgroundNotify::IndexBuildStopped(IndexBuildStopped {
+            file_id: self.source.file_meta.file_id,
+        }))
+        .await;
+    }
+
+    fn region_file_id(&self) -> RegionFileId {
+        RegionFileId::new(
+            self.source.file_meta.region_id,
+            self.source.file_meta.file_id,
+        )
+    }
+
+    async fn notify_worker(&self, notify: BackgroundNotify) {
         let worker_request = WorkerRequest::Background {
             region_id: self.region_id,
-            notify: BackgroundNotify::IndexBuildStopped(IndexBuildStopped {
-                file_id: self.source.file_meta.file_id,
-            }),
+            notify,
         };
         let _ = self
             .request_sender
@@ -761,6 +792,25 @@ impl IndexBuildTask {
             .await;
     }
 
+    /// Builds an aborted outcome without a listener event. Builds that create
+    /// no index stop here, and the listener observes them as neither finished
+    /// nor aborted.
+    fn aborted(&self, reason: &str) -> IndexBuildReport {
+        IndexBuildReport::Done(IndexBuildOutcome::Aborted(format!(
+            "{reason}, region: {}, file_id: {}",
+            self.region_id, self.source.file_meta.file_id
+        )))
+    }
+
+    async fn abort_and_notify(
+        &self,
+        region_file_id: RegionFileId,
+        reason: &str,
+    ) -> IndexBuildReport {
+        self.listener.on_index_build_abort(region_file_id).await;
+        self.aborted(reason)
+    }
+
     // Checks if the SST file still exists in object store and version to avoid conflict with compaction.
     async fn check_sst_file_exists(&self, version_control: &VersionControlRef) -> bool {
         let file_id = self.source.file_meta.file_id;
@@ -796,27 +846,93 @@ impl IndexBuildTask {
     async fn index_build(
         &mut self,
         version_control: VersionControlRef,
-    ) -> Result<IndexBuildOutcome> {
+    ) -> Result<IndexBuildReport> {
+        // Taken before `prepare_source` replaces the source, and used for the
+        // index files and listener events of this build.
+        let region_file_id = self.region_file_id();
+        if let ControlFlow::Break(report) =
+            self.prepare_source(region_file_id, &version_control).await
+        {
+            return Ok(report);
+        }
         let new_index_version = self
             .source
             .file_meta
             .index_version()
             .map_or(0, |version| version + 1);
+        let (mut indexer, index_output) =
+            match self.create_index(region_file_id, new_index_version).await? {
+                ControlFlow::Continue(created) => created,
+                ControlFlow::Break(report) => return Ok(report),
+            };
 
-        // Check SST file existence before building index to avoid failure of parquet reader.
+        // Check SST file existence again after building index.
         if !self.check_sst_file_exists(&version_control).await {
-            self.listener
-                .on_index_build_abort(RegionFileId::new(
-                    self.source.file_meta.region_id,
-                    self.source.file_meta.file_id,
-                ))
-                .await;
-            return Ok(IndexBuildOutcome::Aborted(format!(
-                "SST file not found during index build, region: {}, file_id: {}",
-                self.region_id, self.source.file_meta.file_id
-            )));
+            // Calls abort to clean up index files.
+            indexer.abort().await;
+            return Ok(self
+                .abort_and_notify(region_file_id, "SST file not found during index build")
+                .await);
         }
 
+        self.publish(region_file_id, index_output, new_index_version)
+            .await
+    }
+
+    /// Checks that the SST still exists and moves the build source to the
+    /// file's current manifest state.
+    async fn prepare_source(
+        &mut self,
+        region_file_id: RegionFileId,
+        version_control: &VersionControlRef,
+    ) -> ControlFlow<IndexBuildReport> {
+        // Check SST file existence before building index to avoid failure of parquet reader.
+        if !self.check_sst_file_exists(version_control).await {
+            return ControlFlow::Break(
+                self.abort_and_notify(region_file_id, "SST file not found during index build")
+                    .await,
+            );
+        }
+
+        // A task may have waited behind another build of the same SST. Build on
+        // the index that build published, otherwise this publication is stale.
+        let current_file_meta = self
+            .manifest_ctx
+            .manifest()
+            .await
+            .files
+            .get(&self.source.file_meta.file_id)
+            .cloned();
+        let Some(current_file_meta) = current_file_meta else {
+            return ControlFlow::Break(
+                self.abort_and_notify(
+                    region_file_id,
+                    "SST file not found in manifest during index build",
+                )
+                .await,
+            );
+        };
+        if self.reason == IndexBuildType::Manual
+            && current_file_meta.is_index_consistent_with_region(
+                &self.target_region_metadata.column_metadatas,
+                &self.index_options,
+            )
+        {
+            return ControlFlow::Break(
+                self.abort_and_notify(region_file_id, "Index is already up to date")
+                    .await,
+            );
+        }
+        self.source.file_meta = current_file_meta;
+        ControlFlow::Continue(())
+    }
+
+    /// Reads the SST and creates the index files.
+    async fn create_index(
+        &self,
+        region_file_id: RegionFileId,
+        new_index_version: u64,
+    ) -> Result<ControlFlow<IndexBuildReport, (Indexer, IndexOutput)>> {
         let mut parquet_reader = self
             .access_layer
             .read_sst(self.file.clone()) // use the latest file handle instead of creating a new one
@@ -833,12 +949,6 @@ impl IndexBuildTask {
                 .filter(|size| *size > 0)
         });
 
-        // Use the same file_id but with new version for index file.
-        let region_file_id = RegionFileId::new(
-            self.source.file_meta.region_id,
-            self.source.file_meta.file_id,
-        );
-        let index_file_id = region_file_id.file_id();
         let mut indexer = self
             .indexer_builder
             .build(region_file_id, new_index_version, row_group_size)
@@ -846,7 +956,7 @@ impl IndexBuildTask {
         // The build type or index options may leave nothing to build. Return before reading
         // any row; an empty indexer finishes with no output, which publishes nothing.
         if indexer.is_empty() {
-            return Ok(IndexBuildOutcome::Finished);
+            return Ok(ControlFlow::Break(self.aborted("No index to build")));
         }
 
         if let Some(mut parquet_reader) = parquet_reader.take() {
@@ -866,103 +976,100 @@ impl IndexBuildTask {
         }
         let index_output = indexer.finish().await;
 
-        if index_output.file_size > 0 {
-            // Check SST file existence again after building index.
-            if !self.check_sst_file_exists(&version_control).await {
-                // Calls abort to clean up index files.
-                indexer.abort().await;
+        if index_output.file_size == 0 {
+            // The indexer logs and drops index creation failures.
+            return Ok(ControlFlow::Break(self.aborted("No index was created")));
+        }
+        Ok(ControlFlow::Continue((indexer, index_output)))
+    }
+
+    /// Uploads the index files and commits them to the manifest.
+    async fn publish(
+        &self,
+        region_file_id: RegionFileId,
+        index_output: IndexOutput,
+        new_index_version: u64,
+    ) -> Result<IndexBuildReport> {
+        // Use the same file_id but with new version for index file.
+        self.maybe_upload_index_file(
+            index_output.clone(),
+            region_file_id.file_id(),
+            new_index_version,
+        )
+        .await?;
+
+        self.listener
+            .on_index_build_before_manifest_commit(region_file_id)
+            .await;
+
+        // The commit may have reached storage even on error, so the uploaded
+        // index file is left for GC instead of being deleted here.
+        match self
+            .update_manifest(index_output, new_index_version)
+            .await?
+        {
+            IndexPublication::Committed {
+                manifest_version,
+                file_meta,
+            } => {
                 self.listener
-                    .on_index_build_abort(RegionFileId::new(
-                        self.source.file_meta.region_id,
-                        self.source.file_meta.file_id,
-                    ))
+                    .on_index_build_manifest_committed(region_file_id)
                     .await;
-                return Ok(IndexBuildOutcome::Aborted(format!(
-                    "SST file not found during index build, region: {}, file_id: {}",
-                    self.region_id, self.source.file_meta.file_id
-                )));
-            }
-
-            // Upload index file if write cache is enabled.
-            self.maybe_upload_index_file(index_output.clone(), index_file_id, new_index_version)
-                .await?;
-
-            self.listener
-                .on_index_build_before_manifest_commit(region_file_id)
-                .await;
-
-            let worker_request = match self.update_manifest(index_output, new_index_version).await {
-                Ok(IndexPublication::Committed {
+                self.notify_worker(BackgroundNotify::IndexBuildFinished(IndexBuildFinished {
                     manifest_version,
                     file_meta,
-                }) => {
-                    self.listener
-                        .on_index_build_manifest_committed(region_file_id)
-                        .await;
-                    let index_build_finished = IndexBuildFinished {
-                        manifest_version,
-                        file_meta,
-                    };
-                    WorkerRequest::Background {
-                        region_id: self.region_id,
-                        notify: BackgroundNotify::IndexBuildFinished(index_build_finished),
-                    }
-                }
-                Ok(IndexPublication::Stale(stale)) => {
-                    INDEX_PUBLICATION_STALE_TOTAL
-                        .with_label_values(&["manifest_commit"])
-                        .inc();
-                    let index_id = RegionIndexId::new(region_file_id, new_index_version);
-                    // If no successor publishes the same version, GC collects the
-                    // remote artifact. Repartition requires GC, while local mode
-                    // accepts this narrow orphan window when GC is disabled.
-                    cleanup_stale_index_caches(
-                        index_id,
-                        &self.access_layer,
-                        self.cache_manager.as_ref(),
-                        self.write_cache.as_ref(),
-                    )
-                    .await;
-                    self.listener.on_index_build_abort(region_file_id).await;
-                    if stale == IndexPublicationStale::SchemaChanged {
-                        // An index-unrelated schema change also invalidates the
-                        // generation fence. Retry to avoid leaving the SST
-                        // unindexed; per-SST coalescing limits this to one active
-                        // and one pending task, so repeated changes cannot storm.
-                        let retry = BuildIndexRequest {
-                            region_id: self.region_id,
-                            build_type: IndexBuildType::SchemaChange,
-                            file_metas: vec![self.source.file_meta.clone()],
-                        };
-                        let worker_request = WorkerRequest::Background {
-                            region_id: self.region_id,
-                            notify: BackgroundNotify::IndexBuildRetry(retry),
-                        };
-                        let _ = self
-                            .request_sender
-                            .send(WorkerRequestWithTime::new(worker_request))
-                            .await;
-                    }
-                    return Ok(IndexBuildOutcome::Aborted(format!(
-                        "Index build source changed before publication, region: {}, file_id: {}",
-                        self.region_id, self.source.file_meta.file_id
-                    )));
-                }
-                Err(e) => {
-                    let err = Arc::new(e);
-                    WorkerRequest::Background {
-                        region_id: self.region_id,
-                        notify: BackgroundNotify::IndexBuildFailed(IndexBuildFailed { err }),
-                    }
-                }
-            };
+                    result_senders: self.result_senders.clone(),
+                }))
+                .await;
+                Ok(IndexBuildReport::Deferred)
+            }
+            IndexPublication::Stale(stale) => Ok(self
+                .handle_stale_publication(region_file_id, stale, new_index_version)
+                .await),
+        }
+    }
 
-            let _ = self
-                .request_sender
-                .send(WorkerRequestWithTime::new(worker_request))
+    async fn handle_stale_publication(
+        &self,
+        region_file_id: RegionFileId,
+        stale: IndexPublicationStale,
+        new_index_version: u64,
+    ) -> IndexBuildReport {
+        INDEX_PUBLICATION_STALE_TOTAL
+            .with_label_values(&["manifest_commit"])
+            .inc();
+        // If no successor publishes the same version, GC collects the
+        // remote artifact. Repartition requires GC, while local mode
+        // accepts this narrow orphan window when GC is disabled.
+        cleanup_stale_index_caches(
+            RegionIndexId::new(region_file_id, new_index_version),
+            &self.access_layer,
+            self.cache_manager.as_ref(),
+            self.write_cache.as_ref(),
+        )
+        .await;
+        if stale != IndexPublicationStale::SchemaChanged {
+            return self
+                .abort_and_notify(
+                    region_file_id,
+                    "Index build source changed before publication",
+                )
                 .await;
         }
-        Ok(IndexBuildOutcome::Finished)
+
+        self.listener.on_index_build_abort(region_file_id).await;
+        // An index-unrelated schema change also invalidates the generation
+        // fence. Retry to avoid leaving the SST unindexed; per-SST coalescing
+        // limits this to one active and one pending task, so repeated changes
+        // cannot storm.
+        self.notify_worker(BackgroundNotify::IndexBuildRetry(BuildIndexRequest {
+            region_id: self.region_id,
+            build_type: IndexBuildType::SchemaChange,
+            file_metas: vec![self.source.file_meta.clone()],
+            result_senders: self.result_senders.clone(),
+        }))
+        .await;
+        IndexBuildReport::Deferred
     }
 
     async fn maybe_upload_index_file(
@@ -1131,19 +1238,26 @@ struct PendingIndexBuilds {
 }
 
 impl PendingIndexBuilds {
-    fn insert(&mut self, pending: PendingIndexBuild) -> Option<IndexBuildTask> {
+    /// Inserts a pending task and returns the task it replaces or that is dropped.
+    ///
+    /// The surviving task takes over the result senders of the returned task, so
+    /// callers waiting on either task wait for the build that actually runs.
+    fn insert(&mut self, mut pending: PendingIndexBuild) -> Option<IndexBuildTask> {
         let file_id = pending.task.source.file_meta.file_id;
         match self.tasks.entry(file_id) {
             std::collections::hash_map::Entry::Vacant(entry) => {
                 entry.insert(pending);
                 None
             }
-            std::collections::hash_map::Entry::Occupied(mut entry)
-                if pending.supersedes(entry.get()) =>
-            {
-                Some(entry.insert(pending).task)
+            std::collections::hash_map::Entry::Occupied(mut entry) => {
+                let survivor = entry.get_mut();
+                if pending.supersedes(survivor) {
+                    std::mem::swap(survivor, &mut pending);
+                }
+                let senders = std::mem::take(&mut pending.task.result_senders);
+                survivor.task.result_senders.extend(senders);
+                Some(pending.task)
             }
-            std::collections::hash_map::Entry::Occupied(_) => Some(pending.task),
         }
     }
 
@@ -1227,7 +1341,7 @@ impl IndexBuildScheduler {
         &mut self,
         version_control: &VersionControlRef,
         task: IndexBuildTask,
-    ) -> Result<()> {
+    ) {
         let status = self
             .region_status
             .entry(task.region_id)
@@ -1238,7 +1352,13 @@ impl IndexBuildScheduler {
             task.source.file_meta.region_id,
             task.source.file_meta.file_id,
         );
-        let can_wait_for_active = status.retiring || task.reason == IndexBuildType::SchemaChange;
+        // A manual build must not report success while an active build that may
+        // not create the requested indexes is still running, so it waits as well.
+        let can_wait_for_active = status.retiring
+            || matches!(
+                task.reason,
+                IndexBuildType::SchemaChange | IndexBuildType::Manual
+            );
         let (rejected, coalesced) =
             if status.building_files.contains(&file_id) && !can_wait_for_active {
                 (Some(task), None)
@@ -1285,7 +1405,6 @@ impl IndexBuildScheduler {
         if should_schedule {
             self.schedule_next_build_batch();
         }
-        Ok(())
     }
 
     /// Schedule tasks until reaching the files limit or no more tasks.
@@ -1370,25 +1489,6 @@ impl IndexBuildScheduler {
         self.schedule_next_build_batch();
     }
 
-    pub(crate) async fn on_failure(&mut self, region_id: RegionId, err: Arc<Error>) {
-        let Some(status) = self.region_status.get_mut(&region_id) else {
-            error!(err; "Index build failed after scheduler state was removed, region: {}", region_id);
-            return;
-        };
-        if status.retiring {
-            error!(err; "Index build from a retiring region incarnation failed, region: {}", region_id);
-            return;
-        }
-        error!(
-            err; "Index build scheduler encountered failure for region {}, failing pending tasks.",
-            region_id
-        );
-        status.fail_pending(err).await;
-        if status.building_files.is_empty() {
-            self.region_status.remove(&region_id);
-        }
-    }
-
     /// Notifies the scheduler that the region is dropped.
     pub(crate) async fn on_region_dropped(&mut self, region_id: RegionId) {
         self.retire_region(
@@ -1470,6 +1570,19 @@ mod tests {
         with_skipping_bloom: bool,
     }
 
+    /// Receives the published index that the worker applies before notifying waiters.
+    async fn recv_index_build_finished(
+        rx: &mut mpsc::Receiver<WorkerRequestWithTime>,
+    ) -> IndexBuildFinished {
+        match rx.recv().await.unwrap().request {
+            WorkerRequest::Background {
+                notify: BackgroundNotify::IndexBuildFinished(finished),
+                ..
+            } => finished,
+            other => panic!("Expect index build finished, got: {other:?}"),
+        }
+    }
+
     async fn seed_manifest_file(manifest_ctx: &ManifestContextRef, file_meta: &FileMeta) {
         manifest_ctx
             .update_manifest(
@@ -1918,14 +2031,12 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
 
         // Schedule the build task and check result.
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task).await;
         match result_rx.recv().await.unwrap() {
             Ok(outcome) => {
                 if outcome == IndexBuildOutcome::Finished {
@@ -1986,7 +2097,7 @@ mod tests {
 
         // Create mock task.
         let (tx, mut rx) = mpsc::channel(4);
-        let (result_tx, mut result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
+        let (result_tx, _result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
         let task = IndexBuildTask {
             region_id,
             file,
@@ -2004,21 +2115,11 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
 
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
-
-        // The task should finish successfully.
-        match result_rx.recv().await.unwrap() {
-            Ok(outcome) => {
-                assert_eq!(outcome, IndexBuildOutcome::Finished);
-            }
-            _ => panic!("Expect finished result"),
-        }
+        scheduler.schedule_build(&version_control, task).await;
 
         // A notification should be sent to the worker to update the manifest.
         let worker_req = rx.recv().await.unwrap().request;
@@ -2098,8 +2199,8 @@ mod tests {
         let file = FileHandle::new(file_meta.clone(), file_purger.clone());
 
         // Create mock task.
-        let (tx, _rx) = mpsc::channel(4);
-        let (result_tx, mut result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
+        let (tx, mut rx) = mpsc::channel(4);
+        let (result_tx, _result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
         let task = IndexBuildTask {
             region_id,
             file,
@@ -2117,13 +2218,11 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
 
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task).await;
 
         let puffin_path = location::index_file_path(
             env.access_layer.table_dir(),
@@ -2151,13 +2250,8 @@ mod tests {
             );
         }
 
-        // The task should finish successfully.
-        match result_rx.recv().await.unwrap() {
-            Ok(outcome) => {
-                assert_eq!(outcome, IndexBuildOutcome::Finished);
-            }
-            _ => panic!("Expect finished result"),
-        }
+        // The worker is notified to apply the published index.
+        recv_index_build_finished(&mut rx).await;
 
         // The index file should exist after the task finishes.
         assert!(
@@ -2228,24 +2322,26 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
 
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task).await;
 
-        // The task should finish successfully.
-        match result_rx.recv().await.unwrap() {
-            Ok(outcome) => {
-                assert_eq!(outcome, IndexBuildOutcome::Finished);
+        // No index needs to be built, so the SST is not read.
+        assert!(matches!(
+            result_rx.recv().await.unwrap(),
+            Ok(IndexBuildOutcome::Aborted(reason)) if reason.contains("No index to build")
+        ));
+
+        // Only the stop notification reaches the worker.
+        assert!(matches!(
+            rx.recv().await.unwrap().request,
+            WorkerRequest::Background {
+                notify: BackgroundNotify::IndexBuildStopped(_),
+                ..
             }
-            _ => panic!("Expect finished result"),
-        }
-
-        // No index is built, so no notification should be sent to the worker.
-        let _ = rx.recv().await.is_none();
+        ));
     }
 
     #[tokio::test]
@@ -2306,8 +2402,8 @@ mod tests {
         let file = FileHandle::new(file_meta.clone(), file_purger.clone());
 
         // Create mock task.
-        let (tx, mut _rx) = mpsc::channel(4);
-        let (result_tx, mut result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
+        let (tx, mut rx) = mpsc::channel(4);
+        let (result_tx, _result_rx) = mpsc::channel::<Result<IndexBuildOutcome>>(4);
         let task = IndexBuildTask {
             region_id,
             file,
@@ -2325,21 +2421,13 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
 
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task).await;
 
-        // The task should finish successfully.
-        match result_rx.recv().await.unwrap() {
-            Ok(outcome) => {
-                assert_eq!(outcome, IndexBuildOutcome::Finished);
-            }
-            _ => panic!("Expect finished result"),
-        }
+        recv_index_build_finished(&mut rx).await;
 
         // The write cache should contain the uploaded index file.
         let index_key = IndexKey::new(
@@ -2400,7 +2488,8 @@ mod tests {
             file_purger,
             indexer_builder,
             request_sender: tx,
-            result_sender: result_tx,
+            index_options: IndexOptions::default(),
+            result_senders: vec![result_tx],
         };
         (task, result_rx)
     }
@@ -2432,10 +2521,7 @@ mod tests {
             IndexBuildType::Manual,
         )
         .await;
-        scheduler
-            .schedule_build(&version_control, active)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, active).await;
         assert_eq!(job_scheduler.num_jobs(), 1);
 
         let (mut older, mut older_rx) = create_mock_task_for_schedule_with_result(
@@ -2446,10 +2532,7 @@ mod tests {
         )
         .await;
         older.source.schema_version = 1;
-        scheduler
-            .schedule_build(&version_control, older)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, older).await;
 
         let (mut latest, mut latest_rx) = create_mock_task_for_schedule_with_result(
             &env,
@@ -2459,27 +2542,23 @@ mod tests {
         )
         .await;
         latest.source.schema_version = 2;
-        scheduler
-            .schedule_build(&version_control, latest)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, latest).await;
 
-        let replaced = tokio::time::timeout(std::time::Duration::from_secs(5), older_rx.recv())
-            .await
-            .expect("replaced pending task result sender was not completed")
-            .expect("replaced pending task result channel closed");
-        assert!(matches!(
-            replaced,
-            Ok(IndexBuildOutcome::Aborted(reason)) if reason.contains("coalesced")
-        ));
-        assert!(matches!(
-            latest_rx.try_recv(),
-            Err(mpsc::error::TryRecvError::Empty)
-        ));
+        // The replaced task's caller waits for the surviving build instead of
+        // being told the replaced build has completed.
+        for rx in [&mut older_rx, &mut latest_rx] {
+            assert!(matches!(
+                rx.try_recv(),
+                Err(mpsc::error::TryRecvError::Empty)
+            ));
+        }
 
         let status = &scheduler.region_status[&region_id];
         assert_eq!(status.building_files.len(), 1);
         assert_eq!(status.pending_tasks.len(), 1);
+        let survivor = &status.pending_tasks.tasks[&file_id].task;
+        assert_eq!(survivor.source.schema_version, 2);
+        assert_eq!(survivor.result_senders.len(), 2);
         assert_eq!(job_scheduler.num_jobs(), 1);
 
         scheduler.on_task_stopped(region_id, file_id);
@@ -2517,10 +2596,7 @@ mod tests {
         )
         .await;
 
-        scheduler
-            .schedule_build(&version_control, task)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task).await;
 
         let result = tokio::time::timeout(std::time::Duration::from_secs(5), result_rx.recv())
             .await
@@ -2563,12 +2639,7 @@ mod tests {
         // Test 1: Basic scheduling
         let task1 =
             create_mock_task_for_schedule(&env, file_id1, region_id, IndexBuildType::Flush).await;
-        assert!(
-            scheduler
-                .schedule_build(&version_control, task1)
-                .await
-                .is_ok()
-        );
+        scheduler.schedule_build(&version_control, task1).await;
         assert!(scheduler.region_status.contains_key(&region_id));
         let status = scheduler.region_status.get(&region_id).unwrap();
         assert_eq!(status.building_files.len(), 1);
@@ -2577,20 +2648,14 @@ mod tests {
         // Test 2: Duplicate file scheduling (should be skipped)
         let task1_dup =
             create_mock_task_for_schedule(&env, file_id1, region_id, IndexBuildType::Flush).await;
-        scheduler
-            .schedule_build(&version_control, task1_dup)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task1_dup).await;
         let status = scheduler.region_status.get(&region_id).unwrap();
         assert_eq!(status.building_files.len(), 1); // Still only one
 
         // Test 3: Fill up to limit (2 building tasks)
         let task2 =
             create_mock_task_for_schedule(&env, file_id2, region_id, IndexBuildType::Flush).await;
-        scheduler
-            .schedule_build(&version_control, task2)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task2).await;
         let status = scheduler.region_status.get(&region_id).unwrap();
         assert_eq!(status.building_files.len(), 2); // Reached limit
         assert_eq!(status.pending_tasks.len(), 0);
@@ -2605,18 +2670,9 @@ mod tests {
         let task5 =
             create_mock_task_for_schedule(&env, file_id5, region_id, IndexBuildType::Manual).await;
 
-        scheduler
-            .schedule_build(&version_control, task3)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task4)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task5)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task3).await;
+        scheduler.schedule_build(&version_control, task4).await;
+        scheduler.schedule_build(&version_control, task5).await;
 
         let status = scheduler.region_status.get(&region_id).unwrap();
         assert_eq!(status.building_files.len(), 2); // Still at limit
@@ -2651,53 +2707,6 @@ mod tests {
 
         // Region should be removed when all tasks complete
         assert!(!scheduler.region_status.contains_key(&region_id));
-
-        // Test 8: A build failure keeps active leases without retiring the region
-        let task6 =
-            create_mock_task_for_schedule(&env, file_id1, region_id, IndexBuildType::Flush).await;
-        let task7 =
-            create_mock_task_for_schedule(&env, file_id2, region_id, IndexBuildType::Flush).await;
-        let task8 =
-            create_mock_task_for_schedule(&env, file_id3, region_id, IndexBuildType::Manual).await;
-
-        scheduler
-            .schedule_build(&version_control, task6)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task7)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task8)
-            .await
-            .unwrap();
-
-        assert!(scheduler.region_status.contains_key(&region_id));
-        let status = scheduler.region_status.get(&region_id).unwrap();
-        assert_eq!(status.building_files.len(), 2);
-        assert_eq!(status.pending_tasks.len(), 1);
-
-        scheduler
-            .on_failure(
-                region_id,
-                Arc::new(
-                    crate::error::UnexpectedSnafu {
-                        reason: "index build failed".to_string(),
-                    }
-                    .build(),
-                ),
-            )
-            .await;
-        let status = scheduler.region_status.get(&region_id).unwrap();
-        assert!(!status.retiring);
-        assert_eq!(status.building_files.len(), 2);
-        assert!(status.pending_tasks.is_empty());
-
-        scheduler.on_task_stopped(region_id, file_id1);
-        assert!(scheduler.region_status.contains_key(&region_id));
-        scheduler.on_task_stopped(region_id, file_id2);
-        assert!(!scheduler.region_status.contains_key(&region_id));
     }
 
     /// Helper to set up a scheduler with files_limit=1 and 3 scheduled tasks,
@@ -2780,18 +2789,9 @@ mod tests {
         )
         .await;
 
-        scheduler
-            .schedule_build(&version_control, task1)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task2)
-            .await
-            .unwrap();
-        scheduler
-            .schedule_build(&version_control, task3)
-            .await
-            .unwrap();
+        scheduler.schedule_build(&version_control, task1).await;
+        scheduler.schedule_build(&version_control, task2).await;
+        scheduler.schedule_build(&version_control, task3).await;
 
         // Verify: 1 building + 2 pending.
         assert!(scheduler.region_status.contains_key(&region_id));
@@ -2900,20 +2900,12 @@ mod tests {
             .await;
             scheduler
                 .schedule_build(&version_control, reopened_task)
-                .await
-                .unwrap();
+                .await;
             let status = scheduler.region_status.get(&region_id).unwrap();
             assert!(status.retiring);
             assert_eq!(status.building_files.len(), 1);
             assert_eq!(status.pending_tasks.len(), 1);
 
-            // A manifest error from the old incarnation must not discard the
-            // reopened task before the old build reports stopped.
-            scheduler
-                .on_failure(region_id, Arc::new(RegionClosedSnafu { region_id }.build()))
-                .await;
-            assert_eq!(scheduler.region_status[&region_id].pending_tasks.len(), 1);
-
             scheduler.on_task_stopped(region_id, building_file_id);
             let status = scheduler.region_status.get(&region_id).unwrap();
             assert!(!status.retiring);
diff --git a/src/mito2/src/worker.rs b/src/mito2/src/worker.rs
index 556f4a0757f..c92070ac4e7 100644
--- a/src/mito2/src/worker.rs
+++ b/src/mito2/src/worker.rs
@@ -1435,9 +1435,6 @@ impl<S: LogStore> RegionWorkerLoop<S> {
             BackgroundNotify::IndexBuildStopped(req) => {
                 self.handle_index_build_stopped(region_id, req).await
             }
-            BackgroundNotify::IndexBuildFailed(req) => {
-                self.handle_index_build_failed(region_id, req).await
-            }
             BackgroundNotify::IndexBuildRetry(req) => {
                 self.handle_rebuild_index(req, OptionOutputTx::new(None))
                     .await
diff --git a/src/mito2/src/worker/handle_compaction.rs b/src/mito2/src/worker/handle_compaction.rs
index 50147584bcd..269cf258340 100644
--- a/src/mito2/src/worker/handle_compaction.rs
+++ b/src/mito2/src/worker/handle_compaction.rs
@@ -79,6 +79,7 @@ impl<S> RegionWorkerLoop<S> {
                                     region_id,
                                     build_type: IndexBuildType::Compact,
                                     file_metas: edit.files_to_add,
+                                    result_senders: Vec::new(),
                                 },
                                 OptionOutputTx::new(None),
                             )
@@ -245,6 +246,7 @@ impl<S> RegionWorkerLoop<S> {
                     region_id,
                     build_type: IndexBuildType::Compact,
                     file_metas: index_build_file_metas,
+                    result_senders: Vec::new(),
                 },
                 OptionOutputTx::new(None),
             )
diff --git a/src/mito2/src/worker/handle_flush.rs b/src/mito2/src/worker/handle_flush.rs
index 88d911dccc3..f09efe81e47 100644
--- a/src/mito2/src/worker/handle_flush.rs
+++ b/src/mito2/src/worker/handle_flush.rs
@@ -443,10 +443,13 @@ impl<S: LogStore> RegionWorkerLoop<S> {
         if self.config.index.build_mode == IndexBuildMode::Sync {
             // Sync flush builds indexes while writing SSTs, except for encoded bulk parts,
             // which are uploaded as-is. Build the missing indexes for those files here.
-            let metadata = region.metadata();
+            let version = region.version();
             index_build_file_metas.retain(|meta| {
                 request.unindexed_files.contains(&meta.file_id)
-                    && !meta.is_index_consistent_with_region(&metadata.column_metadatas)
+                    && !meta.is_index_consistent_with_region(
+                        &version.metadata.column_metadatas,
+                        &version.options.index_options,
+                    )
             });
         }
 
@@ -456,6 +459,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                     region_id,
                     build_type: IndexBuildType::Flush,
                     file_metas: index_build_file_metas,
+                    result_senders: Vec::new(),
                 },
                 OptionOutputTx::new(None),
             )
diff --git a/src/mito2/src/worker/handle_manifest.rs b/src/mito2/src/worker/handle_manifest.rs
index 8aac67dc96f..373080aee90 100644
--- a/src/mito2/src/worker/handle_manifest.rs
+++ b/src/mito2/src/worker/handle_manifest.rs
@@ -197,6 +197,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
                     region_id: region.region_id,
                     build_type: IndexBuildType::SchemaChange,
                     file_metas: Vec::new(),
+                    result_senders: Vec::new(),
                 },
                 OptionOutputTx::new(None),
             )
diff --git a/src/mito2/src/worker/handle_rebuild_index.rs b/src/mito2/src/worker/handle_rebuild_index.rs
index 17ba1ece1c8..927a5866a80 100644
--- a/src/mito2/src/worker/handle_rebuild_index.rs
+++ b/src/mito2/src/worker/handle_rebuild_index.rs
@@ -18,21 +18,21 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use api::v1::region::build_index_request;
-use common_telemetry::{debug, error, warn};
+use common_telemetry::{debug, warn};
 use store_api::metadata::RegionMetadataRef;
 use store_api::region_request::RegionBuildIndexRequest;
 use store_api::storage::{FileId, RegionId};
 use tokio::sync::mpsc;
 
 use crate::cache::CacheStrategy;
-use crate::error::{InvalidRequestSnafu, Result, WorkerStoppedSnafu};
+use crate::error::{
+    IndexBuildIncompleteSnafu, InvalidRequestSnafu, RegionStateSnafu, Result, WorkerStoppedSnafu,
+};
 use crate::manifest::action::RegionEdit;
 use crate::metrics::INDEX_PUBLICATION_STALE_TOTAL;
 use crate::region::options::IndexOptions;
-use crate::region::{IndexBuildSource, MitoRegionRef};
-use crate::request::{
-    BuildIndexRequest, IndexBuildFailed, IndexBuildFinished, IndexBuildStopped, OptionOutputTx,
-};
+use crate::region::{IndexBuildSource, MitoRegionRef, RegionLeaderState, RegionRoleState};
+use crate::request::{BuildIndexRequest, IndexBuildFinished, IndexBuildStopped, OptionOutputTx};
 use crate::sst::file::{FileHandle, FileMeta, RegionFileId, RegionIndexId};
 use crate::sst::index::{
     IndexBuildOutcome, IndexBuildTask, IndexBuildType, IndexerBuilderImpl, ResultMpscSender,
@@ -57,7 +57,7 @@ impl<S> RegionWorkerLoop<S> {
         file: FileHandle,
         file_meta: FileMeta,
         build_type: IndexBuildType,
-        result_sender: ResultMpscSender,
+        result_senders: Vec<ResultMpscSender>,
     ) -> IndexBuildTask {
         let access_layer = region.access_layer.clone();
 
@@ -90,6 +90,7 @@ impl<S> RegionWorkerLoop<S> {
             file: file.clone(),
             target_region_metadata: snapshot.metadata.clone(),
             source: IndexBuildSource::new(file_meta, snapshot.metadata.schema_version),
+            index_options: snapshot.index_options.clone(),
             reason: build_type,
             access_layer: access_layer.clone(),
             listener: self.listener.clone(),
@@ -99,7 +100,7 @@ impl<S> RegionWorkerLoop<S> {
             file_purger: file.file_purger(),
             request_sender: self.sender.clone(),
             indexer_builder: indexer_builder_ref.clone(),
-            result_sender,
+            result_senders,
         }
     }
 
@@ -120,6 +121,7 @@ impl<S> RegionWorkerLoop<S> {
                         region_id,
                         build_type: IndexBuildType::Manual,
                         file_metas: Vec::new(),
+                        result_senders: Vec::new(),
                     },
                     sender,
                 )
@@ -168,6 +170,18 @@ impl<S> RegionWorkerLoop<S> {
         let Some(region) = self.regions.writable_region_or(region_id, &mut sender) else {
             return;
         };
+        // Staging regions are writable but must not publish indexes.
+        if region.should_abort_index() {
+            sender.send(
+                RegionStateSnafu {
+                    region_id,
+                    state: region.state(),
+                    expect: RegionRoleState::Leader(RegionLeaderState::Writable),
+                }
+                .fail(),
+            );
+            return;
+        }
 
         // Only keep the parts of the version that index builds need, so the
         // memtables and SST version are not pinned across the awaits below.
@@ -204,13 +218,20 @@ impl<S> RegionWorkerLoop<S> {
 
         let build_tasks = if request.file_metas.is_empty() {
             // If no specific files are provided, find files whose index is inconsistent with the region metadata.
+            // Check the version rather than the manifest: a publication committed
+            // to the manifest but not yet applied is still invisible to queries.
+            // Its build still holds the file, so the manual task waits for it.
             all_files
                 .values()
+                .filter(|file| {
+                    !file.meta_ref().is_index_consistent_with_region(
+                        &snapshot.metadata.column_metadatas,
+                        &snapshot.index_options,
+                    )
+                })
                 .filter_map(|file| {
                     let file_meta = current_file_meta(file.meta_ref().file_id)?;
-                    (!file_meta
-                        .is_index_consistent_with_region(&snapshot.metadata.column_metadatas))
-                    .then(|| (file.clone(), file_meta))
+                    Some((file.clone(), file_meta))
                 })
                 .collect::<Vec<_>>()
         } else {
@@ -234,9 +255,13 @@ impl<S> RegionWorkerLoop<S> {
             return;
         }
 
-        let num_tasks = build_tasks.len();
-        let (tx, mut rx) = mpsc::channel::<Result<IndexBuildOutcome>>(num_tasks);
-
+        let file_ids = build_tasks
+            .iter()
+            .map(|(file, _)| file.meta_ref().file_id)
+            .collect::<Vec<_>>();
+        let build_type = request.build_type.clone();
+        // Each task holds a sender and reports at most once, so sends never block.
+        let (tx, mut rx) = mpsc::channel::<Result<IndexBuildOutcome>>(build_tasks.len());
         for (file_handle, file_meta) in build_tasks {
             debug!(
                 "Scheduling index build for region {}, file_id {}",
@@ -244,6 +269,7 @@ impl<S> RegionWorkerLoop<S> {
                 file_handle.meta_ref().file_id
             );
 
+            // `set_region_role` can change the state while tasks are scheduled.
             if region.should_abort_index() {
                 warn!(
                     "Region {} is in state {:?}, abort index rebuild process for file_id {}",
@@ -254,29 +280,79 @@ impl<S> RegionWorkerLoop<S> {
                 break;
             }
 
+            let mut result_senders = request.result_senders.clone();
+            result_senders.push(tx.clone());
             let task = self.new_index_build_task(
                 &region,
                 &snapshot,
-                file_handle.clone(),
+                file_handle,
                 file_meta,
                 request.build_type.clone(),
-                tx.clone(),
+                result_senders,
             );
-            let _ = self
-                .index_build_scheduler
+            self.index_build_scheduler
                 .schedule_build(&region.version_control, task)
                 .await;
         }
+        drop(tx);
+
         // Wait for all index build tasks to finish and notify the caller.
         common_runtime::spawn_global(async move {
-            for _ in 0..num_tasks {
-                if let Some(Err(e)) = rx.recv().await {
-                    warn!(e; "Index build task failed for region: {}", region_id);
-                    sender.send(Err(e));
-                    return;
+            let mut abort_reasons = Vec::new();
+            // The channel closes once every task has reported or been dropped.
+            while let Some(result) = rx.recv().await {
+                match result {
+                    Ok(IndexBuildOutcome::Finished) => {}
+                    Ok(IndexBuildOutcome::Aborted(reason)) => abort_reasons.push(reason),
+                    Err(e) => {
+                        warn!(e; "Index build task failed for region: {}", region_id);
+                        sender.send(Err(e));
+                        return;
+                    }
                 }
             }
-            sender.send(Ok(0));
+
+            // The build covers the SSTs that existed when it was requested. A file
+            // removed meanwhile, e.g. replaced by compaction, counts as done: its
+            // replacement is indexed according to the compaction index options,
+            // as any later compaction would be.
+            let manifest = region.manifest_ctx.manifest().await;
+            let (present, removed): (Vec<_>, Vec<_>) = file_ids
+                .into_iter()
+                .partition(|file_id| manifest.files.contains_key(file_id));
+            if !removed.is_empty() && build_type == IndexBuildType::Manual {
+                warn!(
+                    "Files were removed during the manual index build, their replacements follow the compaction index options, region: {}, files: {:?}",
+                    region_id, removed
+                );
+            }
+            let incomplete = present
+                .into_iter()
+                .filter(|file_id| {
+                    !manifest.files[file_id].is_index_consistent_with_region(
+                        &snapshot.metadata.column_metadatas,
+                        &snapshot.index_options,
+                    )
+                })
+                .collect::<Vec<_>>();
+            let result = if incomplete.is_empty() {
+                Ok(0)
+            } else {
+                // Background builds have no caller, and files they leave
+                // unindexed by configuration are expected.
+                if build_type == IndexBuildType::Manual {
+                    warn!(
+                        "Manual index build did not complete, region: {}, files: {:?}, aborted: {:?}",
+                        region_id, incomplete, abort_reasons
+                    );
+                }
+                IndexBuildIncompleteSnafu {
+                    region_id,
+                    file_ids: incomplete,
+                }
+                .fail()
+            };
+            sender.send(result);
         });
     }
 
@@ -324,7 +400,10 @@ impl<S> RegionWorkerLoop<S> {
         }
         drop(manager);
 
-        if !is_current {
+        let outcome = if is_current {
+            self.listener.on_index_build_finish(region_file_id).await;
+            IndexBuildOutcome::Finished
+        } else {
             INDEX_PUBLICATION_STALE_TOTAL
                 .with_label_values(&["worker_apply"])
                 .inc();
@@ -340,21 +419,18 @@ impl<S> RegionWorkerLoop<S> {
             )
             .await;
             self.listener.on_index_build_abort(region_file_id).await;
-            return;
+            IndexBuildOutcome::Aborted(format!(
+                "Index publication was superseded before worker apply, region: {}, file_id: {}",
+                region_id, file_meta.file_id
+            ))
+        };
+        // Notify waiters after the version is updated, so queries issued after a
+        // manual build returns can use the new index. `try_send` keeps the worker
+        // loop from waiting on a caller; a missing result is caught by the
+        // waiter's manifest check.
+        for sender in &request.result_senders {
+            let _ = sender.try_send(Ok(outcome.clone()));
         }
-
-        self.listener.on_index_build_finish(region_file_id).await;
-    }
-
-    pub(crate) async fn handle_index_build_failed(
-        &mut self,
-        region_id: RegionId,
-        request: IndexBuildFailed,
-    ) {
-        error!(request.err; "Index build failed for region: {}", region_id);
-        self.index_build_scheduler
-            .on_failure(region_id, request.err.clone())
-            .await;
     }
 
     pub(crate) async fn handle_index_build_stopped(
diff --git a/tests/cases/standalone/common/function/admin/build_index_table_restart.result b/tests/cases/standalone/common/function/admin/build_index_table_restart.result
index 62b0a1c084a..0004d00ed1b 100644
--- a/tests/cases/standalone/common/function/admin/build_index_table_restart.result
+++ b/tests/cases/standalone/common/function/admin/build_index_table_restart.result
@@ -91,10 +91,6 @@ ADMIN BUILD_INDEX('build_index_restart_test');
 | 0                                             |
 +-----------------------------------------------+
 
--- ALTER may already schedule an async schema-change rebuild. If ADMIN BUILD_INDEX
--- races with it, the manual task can be duplicate-aborted immediately while the
--- schema-change task is still finishing, so wait before checking query visibility.
--- SQLNESS SLEEP 1s
 -- Fulltext index built, verify via fulltext query
 SELECT msg FROM build_index_restart_test WHERE MATCHES(msg, 'fox') ORDER BY ts;
 
@@ -135,6 +131,8 @@ AND index_type LIKE 'fulltext%';
 | 1                         |
 +---------------------------+
 
+-- Distributed region statistics come from datanode heartbeats.
+-- SQLNESS SLEEP 1s
 -- Fulltext index size is reported after the build
 SELECT index_size FROM INFORMATION_SCHEMA.REGION_STATISTICS
 WHERE table_id = (
@@ -200,7 +198,6 @@ ADMIN BUILD_INDEX('build_index_restart_test');
 | 0                                             |
 +-----------------------------------------------+
 
--- SQLNESS SLEEP 1s
 -- Fulltext query still works after idempotent rebuild
 SELECT msg FROM build_index_restart_test WHERE MATCHES(msg, 'fox') ORDER BY ts;
 
diff --git a/tests/cases/standalone/common/function/admin/build_index_table_restart.sql b/tests/cases/standalone/common/function/admin/build_index_table_restart.sql
index 745e98eee2c..f13ede93be0 100644
--- a/tests/cases/standalone/common/function/admin/build_index_table_restart.sql
+++ b/tests/cases/standalone/common/function/admin/build_index_table_restart.sql
@@ -45,10 +45,6 @@ ALTER TABLE build_index_restart_test MODIFY COLUMN msg SET FULLTEXT INDEX;
 
 ADMIN BUILD_INDEX('build_index_restart_test');
 
--- ALTER may already schedule an async schema-change rebuild. If ADMIN BUILD_INDEX
--- races with it, the manual task can be duplicate-aborted immediately while the
--- schema-change task is still finishing, so wait before checking query visibility.
--- SQLNESS SLEEP 1s
 -- Fulltext index built, verify via fulltext query
 SELECT msg FROM build_index_restart_test WHERE MATCHES(msg, 'fox') ORDER BY ts;
 
@@ -66,6 +62,8 @@ WHERE table_id = (
 )
 AND index_type LIKE 'fulltext%';
 
+-- Distributed region statistics come from datanode heartbeats.
+-- SQLNESS SLEEP 1s
 -- Fulltext index size is reported after the build
 SELECT index_size FROM INFORMATION_SCHEMA.REGION_STATISTICS
 WHERE table_id = (
@@ -96,7 +94,6 @@ AND index_type LIKE 'fulltext%';
 -- Run ADMIN BUILD_INDEX again (idempotency)
 ADMIN BUILD_INDEX('build_index_restart_test');
 
--- SQLNESS SLEEP 1s
 -- Fulltext query still works after idempotent rebuild
 SELECT msg FROM build_index_restart_test WHERE MATCHES(msg, 'fox') ORDER BY ts;