From 436d07bf2dcf69cc3e5c5736f1bfe24a0ce4dff1 Mon Sep 17 00:00:00 2001 From: dennis zhuang Date: Wed, 21 Oct 2026 19:03:56 +0000 Subject: [PATCH] fix(mito): return from manual index builds only after the index is published (#9643) * fix(mito): return from manual index builds only after the index is published A manual build_index could report success while the index was missing: the request was rejected or coalesced behind another build, the manifest commit failed, the indexer dropped an error, or the region was staging. Manual builds now wait for active builds of the same SST, keep their waiters across coalescing and schema-change retries, are notified after the worker applies the edit, and verify the files against the manifest before returning. The consistency check also honors inverted_index.ignore_column_ids and a zero segment row count, so such SSTs are no longer rebuilt on every call. Signed-off-by: Dennis Zhuang * test(mito): cover manual index build waiting for a schema-change retry Signed-off-by: Dennis Zhuang * fix(mito): select manual index build files by the applied version Signed-off-by: Dennis Zhuang * chore(mito): log why index builds are skipped or aborted Signed-off-by: Dennis Zhuang * refactor(mito): split IndexBuildTask::index_build into phases Signed-off-by: Dennis Zhuang * fix(mito): stop scheduling index builds when the region state changes Signed-off-by: Dennis Zhuang * chore(mito): document region file id and silent aborts in index builds Signed-off-by: Dennis Zhuang * fix(mito): log why a manual index build is incomplete Signed-off-by: Dennis Zhuang * fix(mito): expose incomplete index builds and treat compacted files as done Signed-off-by: Dennis Zhuang * test(mito): pin index build concurrency in the compaction test Signed-off-by: Dennis Zhuang --------- Signed-off-by: Dennis Zhuang --- src/mito2/src/engine/index_build_test.rs | 424 +++++++++-- src/mito2/src/error.rs | 13 + src/mito2/src/request.rs | 14 +- src/mito2/src/sst/file.rs | 133 ++-- src/mito2/src/sst/index.rs | 672 +++++++++--------- src/mito2/src/worker.rs | 3 - src/mito2/src/worker/handle_compaction.rs | 2 + src/mito2/src/worker/handle_flush.rs | 8 +- src/mito2/src/worker/handle_manifest.rs | 1 + src/mito2/src/worker/handle_rebuild_index.rs | 154 +++- .../admin/build_index_table_restart.result | 7 +- .../admin/build_index_table_restart.sql | 7 +- 12 files changed, 943 insertions(+), 495 deletions(-) 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(®ion.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::>(); + 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, + #[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, } /// 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, -} - /// 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, + /// Waiters carried over from a previous build of the same files. + pub(crate) result_senders: Vec, } #[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::>(); + 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(®ion_meta)); + assert!(file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(!file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(!file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(!file_meta.is_index_consistent_with_region(®ion_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(®ion_meta)); + assert!(!file_meta.is_index_consistent_with_region(®ion_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(®ion_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(®ion_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>; @@ -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, /// Request sender to notify the region worker. pub(crate) request_sender: Sender, - /// 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, } 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) { - 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 { + ) -> Result { + // 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 { + // 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> { 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 { + // 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 { + /// 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 { 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) { - let Some(status) = self.region_status.get_mut(®ion_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(®ion_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, + ) -> 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::>(4); + let (result_tx, _result_rx) = mpsc::channel::>(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::>(4); + let (tx, mut rx) = mpsc::channel(4); + let (result_tx, _result_rx) = mpsc::channel::>(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::>(4); + let (tx, mut rx) = mpsc::channel(4); + let (result_tx, _result_rx) = mpsc::channel::>(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[®ion_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(®ion_id)); let status = scheduler.region_status.get(®ion_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(®ion_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(®ion_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(®ion_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(®ion_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(®ion_id)); - let status = scheduler.region_status.get(®ion_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(®ion_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(®ion_id)); - scheduler.on_task_stopped(region_id, file_id2); - assert!(!scheduler.region_status.contains_key(®ion_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(®ion_id)); @@ -2900,20 +2900,12 @@ mod tests { .await; scheduler .schedule_build(&version_control, reopened_task) - .await - .unwrap(); + .await; let status = scheduler.region_status.get(®ion_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[®ion_id].pending_tasks.len(), 1); - scheduler.on_task_stopped(region_id, building_file_id); let status = scheduler.region_status.get(®ion_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 RegionWorkerLoop { 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 RegionWorkerLoop { region_id, build_type: IndexBuildType::Compact, file_metas: edit.files_to_add, + result_senders: Vec::new(), }, OptionOutputTx::new(None), ) @@ -245,6 +246,7 @@ impl RegionWorkerLoop { 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 RegionWorkerLoop { 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 RegionWorkerLoop { 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 RegionWorkerLoop { 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 RegionWorkerLoop { file: FileHandle, file_meta: FileMeta, build_type: IndexBuildType, - result_sender: ResultMpscSender, + result_senders: Vec, ) -> IndexBuildTask { let access_layer = region.access_layer.clone(); @@ -90,6 +90,7 @@ impl RegionWorkerLoop { 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 RegionWorkerLoop { 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 RegionWorkerLoop { region_id, build_type: IndexBuildType::Manual, file_metas: Vec::new(), + result_senders: Vec::new(), }, sender, ) @@ -168,6 +170,18 @@ impl RegionWorkerLoop { 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 RegionWorkerLoop { 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::>() } else { @@ -234,9 +255,13 @@ impl RegionWorkerLoop { return; } - let num_tasks = build_tasks.len(); - let (tx, mut rx) = mpsc::channel::>(num_tasks); - + let file_ids = build_tasks + .iter() + .map(|(file, _)| file.meta_ref().file_id) + .collect::>(); + 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::>(build_tasks.len()); for (file_handle, file_meta) in build_tasks { debug!( "Scheduling index build for region {}, file_id {}", @@ -244,6 +269,7 @@ impl RegionWorkerLoop { 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 RegionWorkerLoop { break; } + let mut result_senders = request.result_senders.clone(); + result_senders.push(tx.clone()); let task = self.new_index_build_task( ®ion, &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(®ion.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::>(); + 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 RegionWorkerLoop { } 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 RegionWorkerLoop { ) .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;