From 3c2aac0a55900721ee3d9d93937fb2f3d15a8ee4 Mon Sep 17 00:00:00 2001 From: Yingwen Date: Thu, 24 Sep 2026 09:39:21 +0000 Subject: [PATCH] feat: add manual series index reconciliation (#9323) * feat(mito): support manual series index reconciliation Signed-off-by: evenyag * feat(storage): route series index build requests Signed-off-by: evenyag * feat(admin): add BUILD_SERIES_INDEX Signed-off-by: evenyag * test: specify compaction type in series index fixtures Signed-off-by: evenyag * test: correct series index SQL fixtures and error assertions Signed-off-by: evenyag * test(compat): preserve legacy index rebuild across upgrades Signed-off-by: evenyag * test(sql): cover series index admin validation Signed-off-by: evenyag * refactor(mito): bound series index maintenance queue Signed-off-by: evenyag * docs: remove series index how-to guide Signed-off-by: evenyag * test: remove index build upgrade compatibility case Signed-off-by: evenyag * fix: address series index reconciliation review feedback Signed-off-by: evenyag * fix: bound manual series index reconciliation admission Signed-off-by: evenyag * chore: update greptime-proto to merged index build options Signed-off-by: evenyag --------- Signed-off-by: evenyag --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/common/function/src/admin.rs | 3 +- .../function/src/admin/build_index_table.rs | 85 ++- src/metric-engine/src/engine.rs | 31 + src/mito2/src/engine/index_build_test.rs | 57 +- src/mito2/src/series_index/task.rs | 568 ++++++++++++++++-- src/mito2/src/series_index/tests.rs | 137 ++++- src/mito2/src/worker.rs | 16 +- src/mito2/src/worker/handle_rebuild_index.rs | 61 +- src/operator/src/request.rs | 45 +- src/store-api/src/region_request.rs | 39 +- src/table/src/requests.rs | 4 +- tests-integration/src/standalone.rs | 19 +- tests-integration/tests/build_series_index.rs | 148 +++++ tests-integration/tests/main.rs | 1 + 16 files changed, 1111 insertions(+), 107 deletions(-) create mode 100644 tests-integration/tests/build_series_index.rs diff --git a/Cargo.lock b/Cargo.lock index fca6847c3db..2cae7d03322 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5996,7 +5996,7 @@ dependencies = [ [[package]] name = "greptime-proto" version = "0.1.0" -source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=3d232f406fad574e8c630fb1c304fe2ebb013c9c#3d232f406fad574e8c630fb1c304fe2ebb013c9c" +source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=9533078d6fd4bdaddbca0750cb9bf7ad64b2e568#9533078d6fd4bdaddbca0750cb9bf7ad64b2e568" dependencies = [ "prost 0.14.1", "prost-types 0.14.1", diff --git a/Cargo.toml b/Cargo.toml index 5cdc3cf63a2..5dc37afceba 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -160,7 +160,7 @@ fs2 = "0.4" fst = "0.4.7" futures = "0.3" futures-util = "0.3" -greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "3d232f406fad574e8c630fb1c304fe2ebb013c9c" } +greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "9533078d6fd4bdaddbca0750cb9bf7ad64b2e568" } hex = "0.4" hostname = "0.4.0" http = "1" diff --git a/src/common/function/src/admin.rs b/src/common/function/src/admin.rs index ff8172ebb5a..79182dd74e9 100644 --- a/src/common/function/src/admin.rs +++ b/src/common/function/src/admin.rs @@ -34,7 +34,7 @@ use reconcile_catalog::ReconcileCatalogFunction; use reconcile_database::ReconcileDatabaseFunction; use reconcile_table::ReconcileTableFunction; -use crate::admin::build_index_table::BuildIndexFunction; +use crate::admin::build_index_table::{BuildIndexFunction, BuildSeriesIndexFunction}; use crate::admin::discard_unflushed_data::DiscardUnflushedDataFunction; use crate::flush_flow::FlushFlowFunction; use crate::function_registry::FunctionRegistry; @@ -53,6 +53,7 @@ impl AdminFunction { registry.register(GcRegionsFunction::factory()); registry.register(GcTableFunction::factory()); registry.register(BuildIndexFunction::factory()); + registry.register(BuildSeriesIndexFunction::factory()); registry.register(FlushFlowFunction::factory()); registry.register(ReconcileCatalogFunction::factory()); registry.register(ReconcileDatabaseFunction::factory()); diff --git a/src/common/function/src/admin/build_index_table.rs b/src/common/function/src/admin/build_index_table.rs index 155f198c792..24026aead8d 100644 --- a/src/common/function/src/admin/build_index_table.rs +++ b/src/common/function/src/admin/build_index_table.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use api::v1::region::build_index_request; use arrow::datatypes::DataType as ArrowDataType; use common_error::ext::BoxedError; use common_macro::admin_fn; @@ -38,6 +39,45 @@ pub(crate) async fn build_index( table_mutation_handler: &TableMutationHandlerRef, query_ctx: &QueryContextRef, params: &[ValueRef<'_>], +) -> Result { + build_index_impl( + table_mutation_handler, + query_ctx, + params, + "build_index", + None, + ) + .await +} + +/// Reconciles series indexes on every physical data-region leader and waits for publication. +#[admin_fn( + name = BuildSeriesIndexFunction, + display_name = build_series_index, + sig_fn = build_index_signature, + ret = uint64 +)] +pub(crate) async fn build_series_index( + table_mutation_handler: &TableMutationHandlerRef, + query_ctx: &QueryContextRef, + params: &[ValueRef<'_>], +) -> Result { + build_index_impl( + table_mutation_handler, + query_ctx, + params, + "build_series_index", + Some(build_index_request::Options::SeriesIndex(Default::default())), + ) + .await +} + +async fn build_index_impl( + table_mutation_handler: &TableMutationHandlerRef, + query_ctx: &QueryContextRef, + params: &[ValueRef<'_>], + function: &str, + options: Option, ) -> Result { ensure!( params.len() == 1, @@ -51,7 +91,7 @@ pub(crate) async fn build_index( let ValueRef::String(table_name) = params[0] else { return UnsupportedInputDataTypeSnafu { - function: "build_index", + function, datatypes: params.iter().map(|v| v.data_type()).collect::>(), } .fail(); @@ -64,6 +104,7 @@ pub(crate) async fn build_index( let affected_rows = table_mutation_handler .build_index( BuildIndexTableRequest { + options, catalog_name, schema_name, table_name, @@ -78,3 +119,45 @@ pub(crate) async fn build_index( fn build_index_signature() -> Signature { Signature::uniform(1, vec![ArrowDataType::Utf8], Volatility::Immutable) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::function::FunctionContext; + + #[tokio::test] + async fn test_build_series_index_arguments() { + let ctx = FunctionContext::mock(); + let handler = ctx.state.table_mutation_handler.as_ref().unwrap(); + for params in [ + vec![], + vec![ValueRef::Int32(1)], + vec![ValueRef::Null], + vec![ValueRef::String("a"), ValueRef::String("b")], + ] { + assert!( + build_index_impl( + handler, + &ctx.query_ctx, + ¶ms, + "build_series_index", + Some(build_index_request::Options::SeriesIndex(Default::default())) + ) + .await + .is_err() + ); + } + assert_eq!( + Value::UInt64(42), + build_index_impl( + handler, + &ctx.query_ctx, + &[ValueRef::String("greptime.public.test")], + "build_series_index", + Some(build_index_request::Options::SeriesIndex(Default::default())), + ) + .await + .unwrap() + ); + } +} diff --git a/src/metric-engine/src/engine.rs b/src/metric-engine/src/engine.rs index 18d6ff6f197..ec180839526 100644 --- a/src/metric-engine/src/engine.rs +++ b/src/metric-engine/src/engine.rs @@ -614,6 +614,37 @@ mod test { TestEnv, build_rows, create_logical_region_request, row_schema_with_tags, }; + #[tokio::test] + async fn test_build_series_index_forwarding() { + use api::v1::region::build_index_request; + use store_api::region_request::RegionBuildIndexRequest; + + let env = TestEnv::new().await; + env.init_metric_region().await; + let request = || { + RegionRequest::BuildIndex(RegionBuildIndexRequest { + options: Some(build_index_request::Options::SeriesIndex(Default::default())), + }) + }; + // Default Mito config disables series indexes: reaching that error proves forwarding. + let error = env + .metric() + .handle_request(env.default_physical_region_id(), request()) + .await + .unwrap_err(); + let error = format!("{error:?}"); + assert!(error.contains("series index is disabled"), "{error}"); + let error = env + .metric() + .handle_request(env.default_logical_region_id(), request()) + .await + .unwrap_err(); + assert_eq!( + common_error::status_code::StatusCode::Unsupported, + error.status_code() + ); + } + #[tokio::test] async fn test_discard_unflushed_data_only() { let env = TestEnv::new().await; diff --git a/src/mito2/src/engine/index_build_test.rs b/src/mito2/src/engine/index_build_test.rs index d2d59b038ff..6d2241846b4 100644 --- a/src/mito2/src/engine/index_build_test.rs +++ b/src/mito2/src/engine/index_build_test.rs @@ -563,7 +563,7 @@ async fn run_index_publication_compaction_race(phase: IndexPublicationPhase, gc_ engine_for_build .handle_request( region_id, - RegionRequest::BuildIndex(RegionBuildIndexRequest {}), + RegionRequest::BuildIndex(RegionBuildIndexRequest::default()), ) .await }); @@ -705,7 +705,7 @@ async fn test_index_build_uses_physical_file_region_and_logical_manifest_region( engine .handle_request( target_region_id, - RegionRequest::BuildIndex(RegionBuildIndexRequest {}), + RegionRequest::BuildIndex(RegionBuildIndexRequest::default()), ) .await .unwrap(); @@ -1391,7 +1391,7 @@ async fn test_index_build_type_manual_basic() { assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 0); // Trigger manual index build task and make sure index file is built without flush or compaction. - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); listener.wait_finish(1).await; let scanner = engine @@ -1403,7 +1403,7 @@ async fn test_index_build_type_manual_basic() { assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1); // Test idempotency: Second manual index build request on the same file. - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); reopen_region(&engine, region_id, table_dir.clone(), true, HashMap::new()).await; let scanner = engine @@ -1416,7 +1416,7 @@ async fn test_index_build_type_manual_basic() { assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1); // Test idempotency again: Third manual index build request to further verify. - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); reopen_region(&engine, region_id, table_dir.clone(), true, HashMap::new()).await; let scanner = engine @@ -1475,7 +1475,7 @@ async fn test_index_build_type_manual_consistency() { assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1); // Check index build task for consistent file will be skipped. - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); // Reopen the region to ensure the task wasn't skipped due to insufficient time. reopen_region(&engine, region_id, table_dir.clone(), true, HashMap::new()).await; @@ -1505,7 +1505,7 @@ async fn test_index_build_type_manual_consistency() { // SyncColumns won't trigger index build. assert_listener_counts(&listener, 1, 1); - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); listener.wait_finish(2).await; // previous 1 + new 1 // Because the file is inconsistent, new index build task is triggered. @@ -1581,7 +1581,7 @@ async fn test_index_build_type_manual_duplicate_in_flight() { // so we must spawn it in a separate task. let engine_clone = engine.clone(); let first_handle = tokio::spawn(async move { - let request = RegionRequest::BuildIndex(RegionBuildIndexRequest {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine_clone.handle_request(region_id, request).await }); @@ -1596,7 +1596,7 @@ async fn test_index_build_type_manual_duplicate_in_flight() { // 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 {}); + let request = RegionRequest::BuildIndex(RegionBuildIndexRequest::default()); engine.handle_request(region_id, request).await.unwrap(); // The second request should have been aborted as duplicate. @@ -1679,7 +1679,7 @@ async fn test_reopen_waits_for_active_index_build_of_previous_incarnation() { engine_for_old_build .handle_request( region_id, - RegionRequest::BuildIndex(RegionBuildIndexRequest {}), + RegionRequest::BuildIndex(RegionBuildIndexRequest::default()), ) .await }); @@ -1707,7 +1707,7 @@ async fn test_reopen_waits_for_active_index_build_of_previous_incarnation() { engine_for_new_build .handle_request( region_id, - RegionRequest::BuildIndex(RegionBuildIndexRequest {}), + RegionRequest::BuildIndex(RegionBuildIndexRequest::default()), ) .await }); @@ -1879,3 +1879,38 @@ async fn test_index_build_type_compact_abort_race() { assert_eq!(scanner.num_files(), 1); assert_eq!(num_of_index_files(&engine, &scanner, region_id).await, 1); } + +#[tokio::test] +async fn test_build_series_index_rejects_non_metric_region() { + let mut env = TestEnv::with_prefix("series-unsupported-schema").await; + let engine = env + .create_engine(MitoConfig { + experimental_enable_series_index: true, + ..Default::default() + }) + .await; + let region_id = RegionId::new(1, 1); + engine + .handle_request( + region_id, + RegionRequest::Create(CreateRequestBuilder::new().build()), + ) + .await + .unwrap(); + let error = engine + .handle_request( + region_id, + RegionRequest::BuildIndex(RegionBuildIndexRequest { + options: Some(api::v1::region::build_index_request::Options::SeriesIndex( + Default::default(), + )), + }), + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("sparse metric metadata"), + "{error}" + ); + engine.stop().await.unwrap(); +} diff --git a/src/mito2/src/series_index/task.rs b/src/mito2/src/series_index/task.rs index 5579cf87ce4..6ce340827a1 100644 --- a/src/mito2/src/series_index/task.rs +++ b/src/mito2/src/series_index/task.rs @@ -20,32 +20,52 @@ use std::time::Duration; use common_telemetry::{info, warn}; use object_store::ObjectStore; -use tokio::sync::Notify; -use tokio::sync::mpsc::UnboundedReceiver; +use snafu::ensure; +use tokio::sync::mpsc::{self, Receiver, Sender, UnboundedReceiver}; +use tokio::sync::oneshot; use tokio::task::JoinHandle; -use tokio::time::{Instant, MissedTickBehavior}; +use tokio::time::Instant; +use crate::error::{InvalidRequestSnafu, RegionBusySnafu, Result, WorkerStoppedSnafu}; use crate::metrics::{ SERIES_INDEX_CAPACITY_DEFERRED, SERIES_INDEX_DISK_BYTES, SERIES_INDEX_RECONCILE_TOTAL, }; -use crate::region::{RegionLeaderState, RegionMapRef, RegionRoleState}; -use crate::series_index::maintenance::reconcile_series_indexes; +use crate::read::series_candidate::is_sparse_metric_metadata; +use crate::region::{MitoRegionRef, RegionLeaderState, RegionMapRef, RegionRoleState}; +use crate::series_index::maintenance::{ReconcileStats, reconcile_series_indexes}; use crate::series_index::purger::{IndexFilePurger, PurgeRequest, run_index_purge_task}; use crate::time_provider::TimeProviderRef; -/// Shared lifecycle state for a worker's series-index task. +/// Commands serialized with periodic series-index maintenance. +#[derive(Debug)] +pub(crate) enum SeriesIndexCommand { + Wake, + Reconcile { + region: MitoRegionRef, + sender: oneshot::Sender>, + }, + Stop, +} + +/// Shared lifecycle state and command sender for a worker's series-index task. #[derive(Debug)] pub(crate) struct SeriesIndexTaskState { + worker_id: u32, running: AtomicBool, - notify: Notify, + sender: Sender, } impl SeriesIndexTaskState { - pub(crate) fn new() -> Self { - Self { - running: AtomicBool::new(true), - notify: Notify::new(), - } + pub(crate) fn new(worker_id: u32, channel_size: usize) -> (Self, Receiver) { + let (sender, receiver) = mpsc::channel(channel_size); + ( + Self { + worker_id, + running: AtomicBool::new(true), + sender, + }, + receiver, + ) } pub(crate) fn is_running(&self) -> bool { @@ -53,17 +73,37 @@ impl SeriesIndexTaskState { } pub(crate) fn wake(&self) { - self.notify.notify_one(); + // Queued work already wakes the task. Periodic maintenance still runs even + // if the queued command only reconciles one region. + if self.is_running() && self.sender.capacity() == self.sender.max_capacity() { + let _ = self.sender.try_send(SeriesIndexCommand::Wake); + } + } + + /// Returns a receiver for build completion, or RegionBusy if the queue is full. + pub(crate) fn try_reconcile( + &self, + region: MitoRegionRef, + ) -> Result>> { + ensure!(self.is_running(), WorkerStoppedSnafu { id: self.worker_id }); + let region_id = region.region_id; + let (sender, receiver) = oneshot::channel(); + self.sender + .try_send(SeriesIndexCommand::Reconcile { region, sender }) + .map_err(|error| match error { + mpsc::error::TrySendError::Full(_) => RegionBusySnafu { region_id }.build(), + mpsc::error::TrySendError::Closed(_) => { + WorkerStoppedSnafu { id: self.worker_id }.build() + } + })?; + Ok(receiver) } pub(crate) fn stop(&self) { - self.running.store(false, Ordering::Release); - // Retain a permit if maintenance has not started waiting yet. - self.notify.notify_one(); - } - - pub(crate) async fn notified(&self) { - self.notify.notified().await; + if self.running.swap(false, Ordering::AcqRel) { + // A full queue already wakes the task, which checks the running flag. + let _ = self.sender.try_send(SeriesIndexCommand::Stop); + } } } @@ -74,6 +114,7 @@ pub(crate) fn spawn_series_index_tasks( store: ObjectStore, regions: RegionMapRef, state: Arc, + receiver: Receiver, bucket_width: Duration, purger: IndexFilePurger, purge_receiver: UnboundedReceiver, @@ -100,6 +141,7 @@ pub(crate) fn spawn_series_index_tasks( max_size, reported_usage: 0, state, + receiver, interval, time_provider, enable_range_index, @@ -120,6 +162,7 @@ struct SeriesIndexTask { reported_usage: u64, worker_id: u32, state: Arc, + receiver: Receiver, interval: Duration, time_provider: TimeProviderRef, enable_range_index: bool, @@ -131,21 +174,98 @@ impl SeriesIndexTask { let worker_id = self.worker_id; info!("Start series-index background task, worker: {worker_id}"); let interval = self.time_provider.wait_duration(self.interval); - let mut timer = tokio::time::interval_at(Instant::now() + interval, interval); - // Schedule future ticks from a late tick rather than the original cadence. - timer.set_missed_tick_behavior(MissedTickBehavior::Delay); + let mut deadline = Instant::now() + interval; while self.state.is_running() { - tokio::select! { - _ = self.state.notified() => {} - _ = timer.tick() => {} - } - if self.state.is_running() { + // Check before receiving so a busy channel cannot starve periodic maintenance. + if Instant::now() >= deadline { self.maintain().await; + deadline = Instant::now() + interval; + continue; + } + match tokio::time::timeout_at(deadline, self.receiver.recv()).await { + Err(_) => continue, + Ok(None | Some(SeriesIndexCommand::Stop)) => break, + Ok(Some(SeriesIndexCommand::Wake)) => { + if self.state.is_running() { + self.maintain().await; + deadline = Instant::now() + interval; + } + } + Ok(Some(SeriesIndexCommand::Reconcile { region, sender })) => { + let result = self.reconcile_manual(region).await; + let _ = sender.send(result); + } + } + } + self.state.stop(); + self.receiver.close(); + while let Some(command) = self.receiver.recv().await { + if let SeriesIndexCommand::Reconcile { sender, .. } = command { + let _ = sender.send(WorkerStoppedSnafu { id: worker_id }.fail()); } } info!("Stop series-index background task, worker: {worker_id}"); } + /// A manual request must still refer to the same writable region instance. + fn validate_manual_region(&self, region: &MitoRegionRef) -> Result<()> { + ensure!( + self.state.is_running(), + WorkerStoppedSnafu { id: self.worker_id } + ); + let current = self.regions.writable_non_staging_region(region.region_id)?; + ensure!( + Arc::ptr_eq(¤t, region), + InvalidRequestSnafu { + region_id: region.region_id, + reason: "region was replaced during series-index reconciliation", + } + ); + ensure!( + is_sparse_metric_metadata(®ion.version().metadata), + InvalidRequestSnafu { + region_id: region.region_id, + reason: "series indexes require sparse metric metadata", + } + ); + Ok(()) + } + + async fn reconcile_manual(&mut self, region: MitoRegionRef) -> Result { + self.validate_manual_region(®ion)?; + self.refresh_usage(); + let result = self.reconcile_region(region.clone()).await; + self.refresh_usage(); + let result = result?; + self.validate_manual_region(®ion)?; + Ok(result) + } + + async fn reconcile_region(&self, region: MitoRegionRef) -> Result { + // Full capacity defers builds, but cleanup must still reclaim published usage. + let allow_builds = self.disk_usage.load(Ordering::Relaxed) < self.max_size; + if !allow_builds { + SERIES_INDEX_CAPACITY_DEFERRED.inc(); + } + let result = reconcile_series_indexes( + self.worker_id, + self.store.clone(), + region, + self.bucket_width, + self.time_provider.current_time_millis(), + self.purger.clone(), + self.enable_range_index, + allow_builds, + ) + .await; + if result.is_err() { + SERIES_INDEX_RECONCILE_TOTAL + .with_label_values(&["failure"]) + .inc(); + } + result + } + /// Each worker contributes only its open regions, refreshed at maintenance boundaries. fn refresh_usage(&mut self) { let usage = self @@ -185,26 +305,7 @@ impl SeriesIndexTask { ) { continue; } - // Full capacity defers builds, but cleanup must still reclaim published usage. - let allow_builds = self.disk_usage.load(Ordering::Relaxed) < self.max_size; - if !allow_builds { - SERIES_INDEX_CAPACITY_DEFERRED.inc(); - } - if let Err(error) = reconcile_series_indexes( - self.worker_id, - self.store.clone(), - region.clone(), - self.bucket_width, - self.time_provider.current_time_millis(), - self.purger.clone(), - self.enable_range_index, - allow_builds, - ) - .await - { - SERIES_INDEX_RECONCILE_TOTAL - .with_label_values(&["failure"]) - .inc(); + if let Err(error) = self.reconcile_region(region.clone()).await { warn!(error; "Failed to reconcile series indexes, worker: {}, region: {}", self.worker_id, region.region_id); } self.refresh_usage(); @@ -220,8 +321,10 @@ impl Drop for SeriesIndexTask { #[cfg(test)] mod tests { + use object_store::layers::mock::{self, MockLayerBuilder, oio}; use object_store::services::Memory; - use store_api::region_engine::RegionEngine; + use store_api::region_engine::{RegionEngine, RegionRole}; + use tokio::sync::Notify; use super::*; use crate::region::RegionMap; @@ -230,6 +333,331 @@ mod tests { use crate::series_index::tests::prepare_region; use crate::test_util::TestEnv; + struct BlockingCatalogWriter { + inner: oio::Writer, + entered: Arc, + release: Arc, + } + + impl mock::Write for BlockingCatalogWriter { + async fn write(&mut self, buffer: mock::Buffer) -> mock::Result<()> { + self.inner.write(buffer).await + } + + async fn close(&mut self) -> mock::Result { + self.entered.notify_one(); + self.release.notified().await; + self.inner.close().await + } + + async fn abort(&mut self) -> mock::Result<()> { + self.inner.abort().await + } + } + + #[derive(Clone, Copy, Debug)] + enum LifecycleChange { + Remove, + Demote, + Stop, + } + + fn new_task(region: MitoRegionRef) -> (SeriesIndexTask, UnboundedReceiver) { + let store = ObjectStore::new(Memory::default()).unwrap(); + let (purger, purge_receiver) = series_index_channel(store.clone()); + let regions = Arc::new(RegionMap::default()); + regions.insert_region(region); + let (state, receiver) = SeriesIndexTaskState::new(0, 1); + ( + SeriesIndexTask { + store, + regions, + bucket_width: Duration::from_secs(100), + purger, + disk_usage: Arc::default(), + max_size: u64::MAX, + reported_usage: 0, + worker_id: 0, + state: Arc::new(state), + receiver, + interval: Duration::from_secs(3600), + time_provider: Arc::new(crate::time_provider::StdTimeProvider), + enable_range_index: true, + }, + purge_receiver, + ) + } + + #[test] + fn test_coalesce_wakeups_and_stop() { + let (state, mut receiver) = SeriesIndexTaskState::new(0, 2); + state.wake(); + state.wake(); + assert!(matches!( + receiver.try_recv().unwrap(), + SeriesIndexCommand::Wake + )); + assert!(receiver.try_recv().is_err()); + state.wake(); + assert!(matches!( + receiver.try_recv().unwrap(), + SeriesIndexCommand::Wake + )); + state.stop(); + state.stop(); + state.wake(); + assert!(matches!( + receiver.try_recv().unwrap(), + SeriesIndexCommand::Stop + )); + assert!(receiver.try_recv().is_err()); + } + + #[rstest::rstest] + #[case(true)] + #[case(false)] + #[tokio::test] + async fn test_manual_reconcile_publishes_before_reply(#[case] range_enabled: bool) { + let mut env = TestEnv::with_prefix("series-manual").await; + let (engine, region) = prepare_region(&mut env).await; + let (mut task, _purge_receiver) = new_task(region.clone()); + task.enable_range_index = range_enabled; + let state = task.state.clone(); + let store = task.store.clone(); + let first = state.try_reconcile(region.clone()).unwrap(); + state.wake(); + assert_eq!(1, task.receiver.len()); + assert!(matches!( + state.try_reconcile(region.clone()), + Err(crate::error::Error::RegionBusy { .. }) + )); + let handle = tokio::spawn(task.run()); + let first = tokio::time::timeout(Duration::from_secs(10), first) + .await + .unwrap() + .unwrap() + .unwrap(); + // Admission recovers after completion; both builds reuse the same publication. + let second = state.try_reconcile(region.clone()).unwrap(); + let second = tokio::time::timeout(Duration::from_secs(10), second) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(1, first.built_series); + assert_eq!(if range_enabled { 4 } else { 0 }, first.built_range); + assert_eq!(0, second.built_series); + assert_eq!(0, second.built_range); + assert_eq!(1, region.series_index_version().series_indexes.len()); + assert!( + store + .exists(&series_catalog_path(region.region_id)) + .await + .unwrap() + ); + assert_eq!( + range_enabled, + store + .exists(&range_catalog_path(region.region_id)) + .await + .unwrap() + ); + state.stop(); + handle.await.unwrap(); + assert!(state.try_reconcile(region).is_err()); + engine.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_manual_reconcile_respects_capacity_and_refreshes_usage() { + let mut env = TestEnv::with_prefix("series-manual-capacity").await; + let (engine, region) = prepare_region(&mut env).await; + let (mut task, _purge_receiver) = new_task(region.clone()); + task.max_size = 0; + task.reconcile_manual(region.clone()).await.unwrap(); + assert_eq!(0, region.series_index_version().disk_usage()); + assert_eq!(0, task.disk_usage.load(Ordering::Relaxed)); + + task.max_size = u64::MAX; + task.reconcile_manual(region.clone()).await.unwrap(); + let bytes = region.series_index_version().disk_usage(); + assert!(bytes > 0); + assert_eq!(bytes, task.disk_usage.load(Ordering::Relaxed)); + let usage = task.disk_usage.clone(); + drop(task); + assert_eq!(0, usage.load(Ordering::Relaxed)); + engine.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_shutdown_fails_queued_reconciliation() { + let mut env = TestEnv::with_prefix("series-manual-stop").await; + let (engine, region) = prepare_region(&mut env).await; + let (task, _purge_receiver) = new_task(region.clone()); + let state = task.state.clone(); + let result = state.try_reconcile(region.clone()).unwrap(); + assert!(matches!( + state.try_reconcile(region.clone()), + Err(crate::error::Error::RegionBusy { .. }) + )); + state.stop(); + assert!(matches!( + state.try_reconcile(region.clone()), + Err(crate::error::Error::WorkerStopped { .. }) + )); + tokio::time::timeout(Duration::from_secs(10), async { + task.run().await; + assert!(matches!( + result.await.unwrap(), + Err(crate::error::Error::WorkerStopped { .. }) + )); + }) + .await + .unwrap(); + assert!(region.series_index_version().series_indexes.is_empty()); + engine.stop().await.unwrap(); + } + + #[tokio::test] + async fn test_manual_reconcile_rejects_closed_region() { + let mut env = TestEnv::with_prefix("series-manual-closed").await; + let (engine, region) = prepare_region(&mut env).await; + let (task, _purge_receiver) = new_task(region.clone()); + let result = task.state.try_reconcile(region.clone()).unwrap(); + task.regions.remove_region(region.region_id); + let state = task.state.clone(); + let handle = tokio::spawn(task.run()); + assert!(matches!( + result.await.unwrap(), + Err(crate::error::Error::RegionNotFound { .. }) + )); + assert!(region.series_index_version().series_indexes.is_empty()); + state.stop(); + handle.await.unwrap(); + engine.stop().await.unwrap(); + } + + #[rstest::rstest] + #[case::remove(LifecycleChange::Remove)] + #[case::demote(LifecycleChange::Demote)] + #[case::stop(LifecycleChange::Stop)] + #[tokio::test] + async fn test_manual_reconcile_rejects_lifecycle_change_during_io( + #[case] change: LifecycleChange, + ) { + let mut env = TestEnv::with_prefix("series-manual-lifecycle").await; + let (engine, region) = prepare_region(&mut env).await; + let (mut task, _purge_receiver) = new_task(region.clone()); + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let writer_entered = entered.clone(); + let writer_release = release.clone(); + let catalog_path = series_catalog_path(region.region_id); + let layer = MockLayerBuilder::default() + .writer_factory(Arc::new(move |path, _, inner| { + if path == catalog_path { + Box::new(BlockingCatalogWriter { + inner, + entered: writer_entered.clone(), + release: writer_release.clone(), + }) + } else { + inner + } + })) + .build() + .unwrap(); + task.store = task.store.clone().layer(layer); + let state = task.state.clone(); + let regions = task.regions.clone(); + let mut result = state.try_reconcile(region.clone()).unwrap(); + let handle = tokio::spawn(task.run()); + tokio::time::timeout(Duration::from_secs(10), entered.notified()) + .await + .unwrap(); + assert!(matches!( + result.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + )); + + // A canceled queued caller still occupies capacity while catalog I/O is blocked. + drop(state.try_reconcile(region.clone()).unwrap()); + for _ in 0..3 { + assert!(matches!( + state.try_reconcile(region.clone()), + Err(crate::error::Error::RegionBusy { .. }) + )); + } + + // The initial validation passed; invalidate the request during catalog I/O. + match change { + LifecycleChange::Remove => { + regions.remove_region(region.region_id).unwrap(); + } + LifecycleChange::Demote => region.set_role(RegionRole::Follower), + LifecycleChange::Stop => state.stop(), + } + release.notify_one(); + let result = tokio::time::timeout(Duration::from_secs(10), result) + .await + .unwrap() + .unwrap(); + state.stop(); + tokio::time::timeout(Duration::from_secs(10), handle) + .await + .unwrap() + .unwrap(); + engine.stop().await.unwrap(); + + match change { + LifecycleChange::Remove => assert!(matches!( + result, + Err(crate::error::Error::RegionNotFound { .. }) + )), + LifecycleChange::Demote => assert!(matches!( + result, + Err(crate::error::Error::RegionState { .. }) + )), + LifecycleChange::Stop => assert!(matches!( + result, + Err(crate::error::Error::WorkerStopped { .. }) + )), + } + } + + #[tokio::test] + async fn test_manual_traffic_does_not_postpone_periodic_maintenance() { + let mut env = TestEnv::with_prefix("series-manual-deadline").await; + let (engine, region) = prepare_region(&mut env).await; + region.switch_state_to_staging(RegionLeaderState::Writable); + region + .manifest_ctx + .exit_staging(region.region_id, RegionRoleState::Follower) + .unwrap(); + let (mut task, _purge_receiver) = new_task(region.clone()); + task.interval = Duration::from_millis(20); + let state = task.state.clone(); + let handle = tokio::spawn(task.run()); + // Manual builds on followers fail. Only a periodic sweep can publish coverage. + tokio::time::timeout(Duration::from_secs(10), async { + while region.series_index_version().series_indexes.is_empty() { + assert!( + state + .try_reconcile(region.clone()) + .unwrap() + .await + .unwrap() + .is_err() + ); + } + }) + .await + .unwrap(); + state.stop(); + handle.await.unwrap(); + engine.stop().await.unwrap(); + } + #[rstest::rstest] #[case::follower(RegionRoleState::Follower, true)] #[case::writable(RegionRoleState::Leader(RegionLeaderState::Writable), true)] @@ -254,6 +682,7 @@ mod tests { let (purger, _receiver) = series_index_channel(store.clone()); let regions = Arc::new(RegionMap::default()); regions.insert_region(region.clone()); + let (state, receiver) = SeriesIndexTaskState::new(0, 2); let mut task = SeriesIndexTask { store: store.clone(), regions, @@ -263,11 +692,16 @@ mod tests { max_size: u64::MAX, reported_usage: 0, worker_id: 0, - state: Arc::new(SeriesIndexTaskState::new()), + state: Arc::new(state), + receiver, interval: Duration::from_secs(3600), time_provider: Arc::new(crate::time_provider::StdTimeProvider), enable_range_index: true, }; + assert_eq!( + role == RegionRoleState::Leader(RegionLeaderState::Writable), + task.validate_manual_region(®ion).is_ok(), + ); task.maintain().await; assert_eq!( builds, @@ -332,6 +766,7 @@ mod tests { regions.insert_region(region.clone()); let clock = Arc::new(MockTimeProvider::new(0)); let usage = Arc::new(AtomicU64::new(0)); + let (state, command_receiver) = SeriesIndexTaskState::new(0, 1); let mut task = SeriesIndexTask { store: store.clone(), regions, @@ -341,7 +776,8 @@ mod tests { max_size: u64::MAX, reported_usage: 0, worker_id: 0, - state: Arc::new(SeriesIndexTaskState::new()), + state: Arc::new(state), + receiver: command_receiver, interval: Duration::from_secs(3600), time_provider: clock.clone(), enable_range_index, @@ -465,19 +901,23 @@ mod tests { let (engine, region) = prepare_region(&mut env).await; let usage = Arc::new(AtomicU64::new(0)); let store = ObjectStore::new(Memory::default()).unwrap(); - let make_task = || SeriesIndexTask { - store: store.clone(), - regions: Arc::new(RegionMap::default()), - bucket_width: Duration::from_secs(100), - purger: series_index_channel(store.clone()).0, - disk_usage: usage.clone(), - max_size: 1024, - reported_usage: 0, - worker_id: 0, - state: Arc::new(SeriesIndexTaskState::new()), - interval: Duration::from_secs(3600), - time_provider: Arc::new(crate::time_provider::StdTimeProvider), - enable_range_index: true, + let make_task = || { + let (state, command_receiver) = SeriesIndexTaskState::new(0, 1); + SeriesIndexTask { + store: store.clone(), + regions: Arc::new(RegionMap::default()), + bucket_width: Duration::from_secs(100), + purger: series_index_channel(store.clone()).0, + disk_usage: usage.clone(), + max_size: 1024, + reported_usage: 0, + worker_id: 0, + state: Arc::new(state), + receiver: command_receiver, + interval: Duration::from_secs(3600), + time_provider: Arc::new(crate::time_provider::StdTimeProvider), + enable_range_index: true, + } }; let mut task = make_task(); task.regions.insert_region(region.clone()); diff --git a/src/mito2/src/series_index/tests.rs b/src/mito2/src/series_index/tests.rs index 8ee898277bc..631bdc75b87 100644 --- a/src/mito2/src/series_index/tests.rs +++ b/src/mito2/src/series_index/tests.rs @@ -52,12 +52,23 @@ pub(crate) async fn prepare_region_with_timestamps( env: &mut TestEnv, timestamps: &[i64], ) -> (MitoEngine, MitoRegionRef) { - let engine = env - .create_engine(MitoConfig { + prepare_region_with_config( + env, + timestamps, + MitoConfig { min_compaction_interval: Duration::from_secs(3600), ..Default::default() - }) - .await; + }, + ) + .await +} + +async fn prepare_region_with_config( + env: &mut TestEnv, + timestamps: &[i64], + config: MitoConfig, +) -> (MitoEngine, MitoRegionRef) { + let engine = env.create_engine(config).await; let metadata = Arc::new(sst_region_metadata_with_encoding( PrimaryKeyEncoding::Sparse, )); @@ -834,7 +845,8 @@ async fn test_maintenance_wakeup_and_timer(#[case] enable_range_index: bool) { region.version_control.alter_options(options); let store = ObjectStore::new(Memory::default()).unwrap(); let (purger, receiver) = series_index_channel(store.clone()); - let state = Arc::new(super::task::SeriesIndexTaskState::new()); + let (state, commands) = super::task::SeriesIndexTaskState::new(0, 2); + let state = Arc::new(state); let clock = Arc::new(crate::time_provider::mock::MockTimeProvider::new(0)); // A notification issued before the task starts must also trigger maintenance. state.wake(); @@ -845,6 +857,7 @@ async fn test_maintenance_wakeup_and_timer(#[case] enable_range_index: bool) { store, regions, state.clone(), + commands, Duration::from_secs(100), purger, receiver, @@ -917,13 +930,15 @@ async fn test_drop_catalogs_and_retained_snapshot_after_task_stop() { let control = load_version_control(&store, region_id, &purger).await; let snapshot = control.current(); assert_eq!(&entry, snapshot.series_indexes[&entry.index_uuid].entry()); - let state = Arc::new(super::task::SeriesIndexTaskState::new()); + let (state, commands) = super::task::SeriesIndexTaskState::new(0, 2); + let state = Arc::new(state); state.stop(); super::task::spawn_series_index_tasks( 0, store.clone(), Arc::new(RegionMap::default()), state, + commands, Duration::from_secs(100), purger, receiver, @@ -951,3 +966,113 @@ async fn test_drop_catalogs_and_retained_snapshot_after_task_stop() { .await .unwrap(); } + +#[rstest::rstest] +#[case(true)] +#[case(false)] +#[tokio::test] +async fn test_build_series_index_region_request(#[case] enabled: bool) { + use api::v1::region::build_index_request; + use store_api::region_request::RegionBuildIndexRequest; + + let mut env = TestEnv::with_prefix("series-manual-request").await; + let config = MitoConfig { + experimental_enable_series_index: enabled, + experimental_series_index_maintenance_interval: Duration::from_secs(3600), + ..Default::default() + }; + let (engine, region) = + prepare_region_with_config(&mut env, &[1000, 2000, 3000, 4000], config).await; + for _ in 0..2 { + let result = engine + .handle_request( + region.region_id, + RegionRequest::BuildIndex(RegionBuildIndexRequest { + options: Some(build_index_request::Options::SeriesIndex(Default::default())), + }), + ) + .await; + if enabled { + assert_eq!(0, result.unwrap().affected_rows); + assert!(!region.series_index_version().series_indexes.is_empty()); + } else { + assert!( + result + .unwrap_err() + .to_string() + .contains("series index is disabled") + ); + } + } + engine.stop().await.unwrap(); +} + +#[rstest::rstest] +#[case::build(false)] +#[case::catalog(true)] +#[tokio::test] +async fn test_manual_reconcile_returns_storage_failure(#[case] catalog_failure: bool) { + let mut env = TestEnv::with_prefix("series-manual-failure").await; + let (engine, region) = prepare_region(&mut env).await; + let catalog = series_catalog_path(region.region_id); + let fail_once = Arc::new(std::sync::atomic::AtomicBool::new(true)); + let fail = fail_once.clone(); + let layer = MockLayerBuilder::default() + .writer_factory(Arc::new(move |path, _, inner| { + let target = if catalog_failure { + path == catalog + } else { + path.contains("/series/") + }; + Box::new(FailingSeriesWriter { + inner, + fail: target && fail.load(std::sync::atomic::Ordering::Relaxed), + }) + })) + .build() + .unwrap(); + let store = ObjectStore::new(Memory::default()).unwrap().layer(layer); + let (purger, purge_receiver) = series_index_channel(store.clone()); + let (state, receiver) = super::task::SeriesIndexTaskState::new(0, 2); + let state = Arc::new(state); + let regions = Arc::new(RegionMap::default()); + regions.insert_region(region.clone()); + let task = super::task::spawn_series_index_tasks( + 0, + store, + regions, + state.clone(), + receiver, + Duration::from_secs(100), + purger, + purge_receiver, + Arc::default(), + u64::MAX, + Duration::from_secs(3600), + Arc::new(StdTimeProvider), + false, + ); + assert!( + state + .try_reconcile(region.clone()) + .unwrap() + .await + .unwrap() + .is_err() + ); + assert!(region.series_index_version().series_indexes.is_empty()); + fail_once.store(false, std::sync::atomic::Ordering::Relaxed); + assert_eq!( + 1, + state + .try_reconcile(region) + .unwrap() + .await + .unwrap() + .unwrap() + .built_series + ); + state.stop(); + task.await.unwrap(); + engine.stop().await.unwrap(); +} diff --git a/src/mito2/src/worker.rs b/src/mito2/src/worker.rs index 0f67d8d2a1b..2eda4fff49f 100644 --- a/src/mito2/src/worker.rs +++ b/src/mito2/src/worker.rs @@ -605,16 +605,21 @@ impl WorkerStarter { let (sender, receiver) = mpsc::channel(self.config.worker_channel_size); let running = Arc::new(AtomicBool::new(true)); - let series_index_task_state = self - .series_index_store - .as_ref() - .map(|_| Arc::new(SeriesIndexTaskState::new())); + let (series_index_task_state, series_index_receiver) = if self.series_index_store.is_some() + { + let (state, receiver) = + SeriesIndexTaskState::new(self.id, self.config.worker_channel_size); + (Some(Arc::new(state)), Some(receiver)) + } else { + (None, None) + }; let mut series_index_purger = None; let series_index_handle = self .series_index_store .clone() .zip(series_index_task_state.clone()) - .map(|(store, state)| { + .zip(series_index_receiver) + .map(|((store, state), receiver)| { let (purger, purge_receiver) = series_index_channel(store.clone()); series_index_purger = Some(purger.clone()); spawn_series_index_tasks( @@ -622,6 +627,7 @@ impl WorkerStarter { store, regions.clone(), state, + receiver, self.config.experimental_series_index_bucket_width, purger, purge_receiver, diff --git a/src/mito2/src/worker/handle_rebuild_index.rs b/src/mito2/src/worker/handle_rebuild_index.rs index 7b203e9af04..59f42ef4232 100644 --- a/src/mito2/src/worker/handle_rebuild_index.rs +++ b/src/mito2/src/worker/handle_rebuild_index.rs @@ -17,13 +17,14 @@ use std::collections::HashMap; use std::sync::Arc; +use api::v1::region::build_index_request; use common_telemetry::{debug, error, warn}; use store_api::region_request::RegionBuildIndexRequest; use store_api::storage::{FileId, RegionId}; use tokio::sync::mpsc; use crate::cache::CacheStrategy; -use crate::error::Result; +use crate::error::{InvalidRequestSnafu, Result, WorkerStoppedSnafu}; use crate::manifest::action::RegionEdit; use crate::metrics::INDEX_PUBLICATION_STALE_TOTAL; use crate::region::version::VersionRef; @@ -96,18 +97,56 @@ impl RegionWorkerLoop { pub(crate) async fn handle_build_index_request( &mut self, region_id: RegionId, - _req: RegionBuildIndexRequest, + req: RegionBuildIndexRequest, sender: OptionOutputTx, ) { - self.handle_rebuild_index( - BuildIndexRequest { - region_id, - build_type: IndexBuildType::Manual, - file_metas: Vec::new(), - }, - sender, - ) - .await; + match req.options { + Some(build_index_request::Options::SeriesIndex(_)) => { + self.handle_build_series_index_request(region_id, sender); + } + None | Some(build_index_request::Options::SstIndex(_)) => { + self.handle_rebuild_index( + BuildIndexRequest { + region_id, + build_type: IndexBuildType::Manual, + file_metas: Vec::new(), + }, + sender, + ) + .await; + } + } + } + + /// Submits manual reconciliation without blocking the region worker on index I/O. + fn handle_build_series_index_request(&self, region_id: RegionId, mut sender: OptionOutputTx) { + let Some(region) = self.regions.writable_region_or(region_id, &mut sender) else { + return; + }; + let Some(state) = &self.series_index_task_state else { + sender.send( + InvalidRequestSnafu { + region_id, + reason: "series index is disabled; enable experimental_enable_series_index", + } + .fail(), + ); + return; + }; + let receiver = match state.try_reconcile(region) { + Ok(receiver) => receiver, + Err(error) => { + sender.send(Err(error)); + return; + } + }; + let worker_id = self.id; + common_runtime::spawn_global(async move { + let result = receiver + .await + .unwrap_or_else(|_| WorkerStoppedSnafu { id: worker_id }.fail()); + sender.send(result.map(|_| 0)); + }); } pub(crate) async fn handle_rebuild_index( diff --git a/src/operator/src/request.rs b/src/operator/src/request.rs index b46a6b657a6..ce3b3568651 100644 --- a/src/operator/src/request.rs +++ b/src/operator/src/request.rs @@ -18,7 +18,7 @@ use api::helper::to_pb_time_unit; use api::v1::region::region_request::Body as RegionRequestBody; use api::v1::region::{ BuildIndexRequest, CompactRequest, CompactionTimeRange, FlushRequest, RegionRequestHeader, - TruncateRequest, Unflushed, truncate_request, + TruncateRequest, Unflushed, build_index_request, truncate_request, }; use catalog::CatalogManagerRef; use common_catalog::build_db_string; @@ -107,6 +107,30 @@ impl Requester { request: BuildIndexTableRequest, ctx: QueryContextRef, ) -> Result { + if matches!( + request.options, + Some(build_index_request::Options::SeriesIndex(_)) + ) { + let table = self + .catalog_manager + .table( + &request.catalog_name, + &request.schema_name, + &request.table_name, + None, + ) + .await + .context(CatalogSnafu)?; + let table = table.with_context(|| TableNotFoundSnafu { + table_name: common_catalog::format_full_table_name( + &request.catalog_name, + &request.schema_name, + &request.table_name, + ), + })?; + let info = table.table_info(); + ensure_build_series_index_supported(&info.meta.engine, info.is_physical_table())?; + } let partitions = &self .get_table_partition_info( &request.catalog_name, @@ -121,6 +145,7 @@ impl Requester { .map(|partition| { RegionRequestBody::BuildIndex(BuildIndexRequest { region_id: partition.id.into(), + options: request.options, }) }) .collect(); @@ -404,6 +429,16 @@ impl Requester { } } +fn ensure_build_series_index_supported(engine: &str, is_physical_table: bool) -> Result<()> { + ensure!( + engine == METRIC_ENGINE && is_physical_table, + NotSupportedSnafu { + feat: "building series indexes requires a physical metric table", + } + ); + Ok(()) +} + fn ensure_discard_unflushed_supported(engine: &str, is_physical_table: bool) -> Result<()> { ensure!( engine != METRIC_ENGINE || is_physical_table, @@ -422,6 +457,14 @@ mod tests { use super::*; + #[test] + fn test_build_series_index_requires_physical_metric_table() { + ensure_build_series_index_supported(METRIC_ENGINE, true).unwrap(); + assert!(ensure_build_series_index_supported(METRIC_ENGINE, false).is_err()); + assert!(ensure_build_series_index_supported("mito", false).is_err()); + assert!(ensure_build_series_index_supported("mito", true).is_err()); + } + #[test] fn test_to_pb_compaction_time_range_normalizes_mixed_units() { let range = TimestampRange::new( diff --git a/src/store-api/src/region_request.rs b/src/store-api/src/region_request.rs index 113a2393ccc..762bf80c4b3 100644 --- a/src/store-api/src/region_request.rs +++ b/src/store-api/src/region_request.rs @@ -26,7 +26,8 @@ use api::v1::region::{ AlterRequest, AlterRequests, BuildIndexRequest, BulkInsertRequest, CleanUpRequest as PbCleanUpRequest, CloseRequest, CompactRequest, CreateRequest, CreateRequests, DeleteRequests, DropRequest, DropRequests, FlushRequest, InsertRequests, - OpenRequest, TruncateRequest, alter_request, compact_request, region_request, truncate_request, + OpenRequest, TruncateRequest, alter_request, build_index_request, compact_request, + region_request, truncate_request, }; use api::v1::{ self, Analyzer, ArrowIpc, FulltextBackend as PbFulltextBackend, Option as PbOption, Rows, @@ -442,7 +443,9 @@ fn make_region_build_index(index: BuildIndexRequest) -> Result, +} /// Truncate region request. #[derive(Debug)] @@ -2054,6 +2060,33 @@ mod tests { use super::*; use crate::metadata::RegionMetadataBuilder; + #[test] + fn test_build_index_options_round_trip() { + use prost::Message; + + for options in [ + None, + Some(build_index_request::Options::SstIndex(Default::default())), + Some(build_index_request::Options::SeriesIndex(Default::default())), + ] { + let request = BuildIndexRequest { + region_id: 42, + options, + }; + let decoded = BuildIndexRequest::decode(request.encode_to_vec().as_slice()).unwrap(); + let requests = make_region_build_index(decoded).unwrap(); + assert_eq!(1, requests.len()); + assert_eq!(RegionId::from_u64(42), requests[0].0); + let RegionRequest::BuildIndex(request) = &requests[0].1 else { + panic!("expected build-index request"); + }; + assert_eq!(options, request.options); + } + // The legacy wire message contains only region_id (field 1). + let legacy = BuildIndexRequest::decode(&[0x08, 42][..]).unwrap(); + assert!(legacy.options.is_none()); + } + #[test] fn test_make_region_puts_preserves_skip_wal() { let region_id = RegionId::new(42, 3); diff --git a/src/table/src/requests.rs b/src/table/src/requests.rs index 2c519b32a5e..899438edf52 100644 --- a/src/table/src/requests.rs +++ b/src/table/src/requests.rs @@ -30,7 +30,7 @@ use datatypes::prelude::VectorRef; use datatypes::schema::{ ColumnDefaultConstraint, ColumnSchema, FulltextOptions, Schema, SkippingIndexOptions, }; -use greptime_proto::v1::region::compact_request; +use greptime_proto::v1::region::{build_index_request, compact_request}; use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; use store_api::metric_engine_consts::{ @@ -805,6 +805,8 @@ pub struct FlushTableRequest { #[derive(Debug, Clone, Default)] pub struct BuildIndexTableRequest { + /// The index build mode. Absent options select SST indexes. + pub options: Option, pub catalog_name: String, pub schema_name: String, pub table_name: String, diff --git a/tests-integration/src/standalone.rs b/tests-integration/src/standalone.rs index 6fc8ce231af..3fa1d394489 100644 --- a/tests-integration/src/standalone.rs +++ b/tests-integration/src/standalone.rs @@ -91,6 +91,7 @@ pub struct GreptimeDbStandaloneBuilder { experimental_metric_export: bool, logical_batcher: Option, table_batcher: BatcherOptions, + mito_config: Option, } impl GreptimeDbStandaloneBuilder { @@ -113,9 +114,17 @@ impl GreptimeDbStandaloneBuilder { experimental_metric_export: false, logical_batcher: None, table_batcher: BatcherOptions::default(), + mito_config: None, } } + /// Overrides the Mito configuration for this test instance. + #[must_use] + pub fn with_mito_config(mut self, config: mito2::config::MitoConfig) -> Self { + self.mito_config = Some(config); + self + } + /// Enables experimental Metric export for the standalone test instance. #[must_use] pub fn with_experimental_metric_export(mut self) -> Self { @@ -386,7 +395,7 @@ impl GreptimeDbStandaloneBuilder { let (procedure_manager, event_recorder_handle) = standalone::build_procedure_manager(kv_backend.clone(), procedure_config); - let standalone_opts = StandaloneOptions { + let mut standalone_opts = StandaloneOptions { storage: opts.storage, procedure: procedure_config, metadata_store: kv_backend_config, @@ -408,6 +417,14 @@ impl GreptimeDbStandaloneBuilder { ..StandaloneOptions::default() }; + if let Some(config) = &self.mito_config { + for engine in &mut standalone_opts.region_engine { + if let datanode::config::RegionEngineConfig::Mito(mito) = engine { + *mito = config.clone(); + } + } + } + self.build_with( kv_backend, guard, diff --git a/tests-integration/tests/build_series_index.rs b/tests-integration/tests/build_series_index.rs new file mode 100644 index 00000000000..8c0e91aa35b --- /dev/null +++ b/tests-integration/tests/build_series_index.rs @@ -0,0 +1,148 @@ +// Copyright 2023 Greptime Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use datanode::config::RegionEngineConfig; +use frontend::instance::Instance; +use mito2::config::MitoConfig; +use tests_integration::cluster::GreptimeDbClusterBuilder; +use tests_integration::standalone::GreptimeDbStandaloneBuilder; +use tests_integration::test_util::{ + StorageType, create_tmp_dir_and_datanode_opts, execute_sql, try_execute_sql, +}; + +fn config() -> MitoConfig { + MitoConfig { + experimental_enable_series_index: true, + experimental_series_index_maintenance_interval: Duration::from_secs(3600), + ..Default::default() + } +} + +async fn exercise_admin(frontend: &Arc) { + execute_sql( + frontend, + r#" + CREATE TABLE series_physical (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) + PARTITION ON COLUMNS (host) (host < 'm', host >= 'm') + ENGINE = metric WITH ( + physical_metric_table = 'true', 'compaction.type' = 'twcs', + 'compaction.twcs.time_window' = '1h', + 'compaction.twcs.active_window.trigger_file_num' = '100', + 'compaction.twcs.inactive_window.trigger_file_num' = '100' + ) + "#, + ) + .await; + execute_sql( + frontend, + r#" + CREATE TABLE series_logical (ts TIMESTAMP TIME INDEX, host STRING PRIMARY KEY, val DOUBLE) + ENGINE = metric WITH (on_physical_table = 'series_physical') + "#, + ) + .await; + // Reconciliation requires at least four SSTs in each bucket. + for timestamp in [1000, 2000, 3000, 4000] { + execute_sql( + frontend, + &format!( + "INSERT INTO series_logical (ts, host, val) VALUES ({timestamp}, 'a', 1), ({timestamp}, 'z', 2)" + ), + ) + .await; + execute_sql(frontend, "ADMIN FLUSH_TABLE('series_physical')").await; + } + for _ in 0..2 { + let output = execute_sql(frontend, "ADMIN BUILD_SERIES_INDEX('series_physical')").await; + assert!(output.data.pretty_print().await.contains("| 0")); + } + // The function must never silently expand a logical table to its shared physical table. + let error = try_execute_sql(frontend, "ADMIN BUILD_SERIES_INDEX('series_logical')") + .await + .unwrap_err(); + assert!(format!("{error:?}").contains("physical metric table")); + for sql in [ + "ADMIN BUILD_SERIES_INDEX()", + "ADMIN BUILD_SERIES_INDEX(1)", + "ADMIN BUILD_SERIES_INDEX('missing')", + ] { + assert!(try_execute_sql(frontend, sql).await.is_err(), "{sql}"); + } + execute_sql(frontend, "CREATE TABLE ordinary (ts TIMESTAMP TIME INDEX)").await; + assert!( + try_execute_sql(frontend, "ADMIN BUILD_SERIES_INDEX('ordinary')") + .await + .is_err() + ); + execute_sql(frontend, "ADMIN BUILD_INDEX('ordinary')").await; + let output = execute_sql(frontend, "SELECT count(*) AS n FROM series_logical").await; + assert!(output.data.pretty_print().await.contains("| 8")); +} + +fn catalog_count(home: &str) -> usize { + let root = Path::new(home).join("series_index"); + std::fs::read_dir(root) + .unwrap() + .map(|entry| entry.unwrap().path()) + .filter(|path| path.join("series-index.json").is_file()) + .count() +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_build_series_index_standalone() { + let standalone = GreptimeDbStandaloneBuilder::new("build-series-index") + .with_mito_config(config()) + .build() + .await; + exercise_admin(standalone.fe_instance()).await; + assert_eq!(2, catalog_count(&standalone.opts.storage.data_home)); +} + +#[tokio::test(flavor = "multi_thread")] +async fn test_build_series_index_distributed() { + let builder = GreptimeDbClusterBuilder::new("build-series-index").await; + let mut options = Vec::new(); + let mut guards = Vec::new(); + for id in 1..=2 { + let (mut opts, guard) = create_tmp_dir_and_datanode_opts( + StorageType::File, + vec![], + &format!("build-series-index-{id}"), + Default::default(), + Default::default(), + ); + opts.node_id = Some(id); + for engine in &mut opts.region_engine { + if let RegionEngineConfig::Mito(mito) = engine { + *mito = config(); + } + } + options.push(opts); + guards.push(guard); + } + let cluster = builder.build_with(options, false, guards).await; + exercise_admin(cluster.fe_instance()).await; + assert_eq!( + 2, + cluster + .datanode_options + .iter() + .map(|opts| catalog_count(&opts.storage.data_home)) + .sum::() + ); +} diff --git a/tests-integration/tests/main.rs b/tests-integration/tests/main.rs index 278a7ba3851..f84b954e23c 100644 --- a/tests-integration/tests/main.rs +++ b/tests-integration/tests/main.rs @@ -15,6 +15,7 @@ #![recursion_limit = "256"] mod admin_function_event; +mod build_series_index; mod database_ddl_event; mod dict_groupby_sst; mod event_recorder_test_util;