From d47545921a1769c540e0b7edef55b596005dba52 Mon Sep 17 00:00:00 2001 From: Lout Philipps Date: Fri, 18 Sep 2026 17:46:32 +0200 Subject: [PATCH] feat(indexing): gate pipeline spreading behind an environment flag Co-Authored-By: Codex --- .../src/actors/cooperative_indexing.rs | 4 + .../quickwit-indexing/src/actors/indexer.rs | 87 ++++++++++++++----- 2 files changed, 71 insertions(+), 20 deletions(-) diff --git a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs index 556779673d7..474827977f5 100644 --- a/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs +++ b/quickwit/quickwit-indexing/src/actors/cooperative_indexing.rs @@ -109,6 +109,10 @@ impl CooperativeIndexingCycle { Duration::from_millis(initial_sleep_millis) } + pub fn commit_timeout(&self) -> Duration { + self.schedule.commit_timeout() + } + pub async fn cooperative_indexing_period(&self) -> CooperativeIndexingPeriod { let t_wake = Instant::now(); let permit = Semaphore::acquire_owned(self.indexing_permits.clone()) diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 742c94c25c1..b85474b12a1 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -16,6 +16,7 @@ use std::collections::hash_map::Entry; use std::num::NonZeroU32; use std::ops::RangeInclusive; use std::sync::Arc; +use std::time::Duration; use anyhow::Context; use async_trait::async_trait; @@ -47,11 +48,13 @@ use tantivy::tokenizer::TokenizerManager; use tantivy::{DateTime, DocId, IndexBuilder, IndexSettings}; use tokio::runtime::Handle; use tokio::sync::Semaphore; +use tokio::time::Instant; use tracing::{Span, debug, info_span, warn}; use ulid::Ulid; use super::IndexSerializer; use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod}; +use super::pipeline_schedule::PipelineSchedule; use crate::docs_clustering::{DocIdClusterer, Fingerprinter}; use crate::metrics::SPLIT_BUILDERS; use crate::models::{ @@ -62,11 +65,52 @@ use crate::models::{ // Random partition ID used to gather partitions exceeding the maximum number of partitions. pub(crate) const OTHER_PARTITION_ID: u64 = 3264326757911759461u64; +const ENABLE_SPREAD_INDEXING_PIPELINES_ENV_KEY: &str = "QW_ENABLE_SPREAD_INDEXING_PIPELINES"; + #[derive(Debug)] struct CommitTimeout { workbench_id: Ulid, } +enum IndexingMode { + Normal(Duration), + Cooperative(CooperativeIndexingCycle), + Spread(PipelineSchedule), +} + +impl IndexingMode { + fn new( + pipeline_id: &IndexingPipelineId, + commit_timeout: Duration, + cooperative_indexing_permits_opt: Option>, + enable_spread_indexing_pipelines: bool, + ) -> Self { + if let Some(permits) = cooperative_indexing_permits_opt { + return Self::Cooperative(CooperativeIndexingCycle::new( + pipeline_id, + commit_timeout, + permits, + )); + } + + if enable_spread_indexing_pipelines { + return Self::Spread(PipelineSchedule::new(pipeline_id, commit_timeout)); + } + + Self::Normal(commit_timeout) + } + + fn commit_timeout(&self, now: Instant) -> Duration { + match self { + Self::Normal(commit_timeout) => *commit_timeout, + // in cooperative indexing mode, the nudging is done only when the actor mailbox is + // empty + Self::Cooperative(cycle) => cycle.commit_timeout(), + Self::Spread(schedule) => schedule.nudged_commit_timeout(now), + } + } +} + #[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)] pub struct IndexerCounters { /// Number of splits that were emitted by the indexer. @@ -100,7 +144,7 @@ struct IndexerState { tokenizer_manager: TokenizerManager, max_num_partitions: NonZeroU32, index_settings: IndexSettings, - cooperative_indexing_opt: Option, + indexing_mode: IndexingMode, indexing_io_throughput_limiter_opt: Option, } @@ -202,7 +246,7 @@ impl IndexerState { ); let indexing_span = info_span!(parent: batch_parent_span.id(), "indexer"); let cooperative_indexing_period = - if let Some(cooperative_indexing) = &self.cooperative_indexing_opt { + if let IndexingMode::Cooperative(cooperative_indexing) = &self.indexing_mode { Some( ctx.protect_future(cooperative_indexing.cooperative_indexing_period()) .await, @@ -261,10 +305,8 @@ impl IndexerState { let commit_timeout_message = CommitTimeout { workbench_id: indexing_workbench.workbench_id, }; - ctx.schedule_self_msg( - self.indexing_settings.commit_timeout(), - commit_timeout_message, - ); + let commit_timeout = self.indexing_mode.commit_timeout(Instant::now()); + ctx.schedule_self_msg(commit_timeout, commit_timeout_message); *indexing_workbench_opt = Some(indexing_workbench); } let current_indexing_workbench = indexing_workbench_opt.as_mut().context( @@ -404,11 +446,15 @@ impl Actor for Indexer { } async fn initialize(&mut self, ctx: &ActorContext) -> Result<(), ActorExitStatus> { - if let Some(cooperative_indexing_cycle) = &self.indexer_state.cooperative_indexing_opt { - let initial_sleep_duration = cooperative_indexing_cycle.initial_sleep_duration(); - ctx.pause(); - ctx.schedule_self_msg(initial_sleep_duration, Command::Resume); - } + let initial_sleep_duration = match &self.indexer_state.indexing_mode { + // when spreading indexing pipelines is enabled, start indexing immediately and nudge + // pipelines on further commit timeout. This way, we are using all available CPU + // right away instead of idling on the first indexing cycle + IndexingMode::Normal(_) | IndexingMode::Spread(_) => return Ok(()), + IndexingMode::Cooperative(cycle) => cycle.initial_sleep_duration(), + }; + ctx.pause(); + ctx.schedule_self_msg(initial_sleep_duration, Command::Resume); Ok(()) } @@ -551,14 +597,15 @@ impl Indexer { // A configured fingerprinter supplies the mapping when the split is finalized. manual_doc_id_mapping: fingerprinter_opt.is_some(), }; - let cooperative_indexing_opt: Option = - cooperative_indexing_permits_opt.map(|cooperative_indexing_permits| { - CooperativeIndexingCycle::new( - &pipeline_id, - indexing_settings.commit_timeout(), - cooperative_indexing_permits, - ) - }); + let indexing_mode = IndexingMode::new( + &pipeline_id, + indexing_settings.commit_timeout(), + cooperative_indexing_permits_opt, + quickwit_common::get_bool_from_env_cached!( + ENABLE_SPREAD_INDEXING_PIPELINES_ENV_KEY, + false, + ), + ); Self { indexer_state: IndexerState { pipeline_id, @@ -572,7 +619,7 @@ impl Indexer { tokenizer_manager: tokenizer_manager.tantivy_manager().clone(), index_settings, max_num_partitions: doc_mapper.max_num_partitions(), - cooperative_indexing_opt, + indexing_mode, indexing_io_throughput_limiter_opt, }, index_serializer_mailbox,