Skip to content

Add an option to spread indexing pipelines over time #6655

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
loutPhilipps wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Add an option to spread indexing pipelines over time #6655

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
aefc04d
Spread indexing pipelines
loutPhilipps Aug 3, 2026
0dfc0f4
Update comments
loutPhilipps Aug 3, 2026
9402a36
Rename file
loutPhilipps Aug 3, 2026
e84a525
Update tests
loutPhilipps Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/configuration/node-config.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,7 @@ This section contains the configuration options for an indexer. The split store
| `enable_otlp_endpoint` | If true, enables the OpenTelemetry exporter endpoint to ingest logs and traces via the OpenTelemetry Protocol (OTLP). | `false` |
| `cpu_capacity` | Advisory parameter used by the control plane. The value can expressed be in threads (e.g. `2`) or in term of millicpus (`2000m`). The control plane will attempt to schedule indexing pipelines on the different nodes proportionally to the cpu capacity advertised by the indexer. It is NOT used as a limit. All pipelines will be scheduled regardless of whether the cluster has sufficient capacity or not. The control plane does not attempt to spread the work equally when the load is well below the `cpu_capacity`. Users who need a balanced load on all of their indexer nodes can set the `cpu_capacity` to an arbitrarily low value as long as they keep it proportional to the number of threads available. | `num threads available` |
| `enable_cooperative_indexing` | Enable sharing resources more efficiently when the number of indexes actively written to is significantly higher than the number of cores but might decrease the overall indexing throughput. | `false` |
| `enable_spread_indexing_pipelines` | Enable spreading indexing pipelines in time to make sure they are not in sync. | `false` |

@fulmicoton-dd fulmicoton-dd Aug 7, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can we make that an environment flag? (the goal is just to make it as light as possible).


Example:

Expand Down
10 changes: 10 additions & 0 deletions quickwit/quickwit-config/src/node_config/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -210,6 +210,10 @@ pub struct IndexerConfig {
pub enable_otlp_endpoint: bool,
#[serde(default = "IndexerConfig::default_enable_cooperative_indexing")]
pub enable_cooperative_indexing: bool,
/// Spreads the indexing pipelines of a node uniformly in time, so that they don't all use
/// the same resources at the same time. Implied by `enable_cooperative_indexing`.
#[serde(default = "IndexerConfig::default_enable_spread_indexing_pipelines")]
pub enable_spread_indexing_pipelines: bool,
#[serde(default = "IndexerConfig::default_cpu_capacity")]
pub cpu_capacity: CpuCapacity,
/// If true, run Parquet merges through the streaming column-major engine
Expand All @@ -229,6 +233,10 @@ impl IndexerConfig {
false
}

fn default_enable_spread_indexing_pipelines() -> bool {
false
}

fn default_enable_otlp_endpoint() -> bool {
#[cfg(any(test, feature = "testsuite"))]
{
Expand Down Expand Up @@ -269,6 +277,7 @@ impl IndexerConfig {
use quickwit_proto::indexing::PIPELINE_FULL_CAPACITY;
let indexer_config = IndexerConfig {
enable_cooperative_indexing: false,
enable_spread_indexing_pipelines: false,
enable_otlp_endpoint: true,
split_store_max_num_bytes: ByteSize::mb(1),
split_store_max_num_splits: 3,
Expand All @@ -286,6 +295,7 @@ impl Default for IndexerConfig {
fn default() -> Self {
Self {
enable_cooperative_indexing: Self::default_enable_cooperative_indexing(),
enable_spread_indexing_pipelines: Self::default_enable_spread_indexing_pipelines(),
enable_otlp_endpoint: Self::default_enable_otlp_endpoint(),
split_store_max_num_bytes: Self::default_split_store_max_num_bytes(),
split_store_max_num_splits: Self::default_split_store_max_num_splits(),
Expand Down
1 change: 1 addition & 0 deletions quickwit/quickwit-config/src/node_config/serialize.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -850,6 +850,7 @@ mod tests {
merge_concurrency: NonZeroUsize::new(2).unwrap(),
cpu_capacity: IndexerConfig::default_cpu_capacity(),
enable_cooperative_indexing: false,
enable_spread_indexing_pipelines: false,
max_merge_write_throughput: Some(ByteSize::mb(100)),
parquet_merge_use_streaming_engine: true,
}
Expand Down
62 changes: 35 additions & 27 deletions quickwit/quickwit-indexing/src/actors/indexer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ use tracing::{Span, debug, info_span, warn};
use ulid::Ulid;

use super::IndexSerializer;
use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod};
use super::indexing_cycle::{IndexingCycle, IndexingPeriod};
use crate::docs_clustering::{DocIdClusterer, Fingerprinter};
use crate::metrics::SPLIT_BUILDERS;
use crate::models::{
Expand Down Expand Up @@ -84,7 +84,8 @@ pub struct IndexerCounters {
pub num_doc_batches_in_workbench: u64,

/// Metrics describing the load and indexing performance of the
/// pipeline. This is only updated for cooperative indexers.
/// pipeline. This is only updated for cooperative indexers or
/// when indexing pipeline spreading is enabled.
pub pipeline_metrics_opt: Option<PipelineMetrics>,
}

Expand All @@ -100,7 +101,7 @@ struct IndexerState {
tokenizer_manager: TokenizerManager,
max_num_partitions: NonZeroU32,
index_settings: IndexSettings,
cooperative_indexing_opt: Option<CooperativeIndexingCycle>,
indexing_cycle_opt: Option<IndexingCycle>,
}

impl IndexerState {
Expand Down Expand Up @@ -199,15 +200,11 @@ impl IndexerState {
workbench_id=%workbench_id,
);
let indexing_span = info_span!(parent: batch_parent_span.id(), "indexer");
let cooperative_indexing_period =
if let Some(cooperative_indexing) = &self.cooperative_indexing_opt {
Some(
ctx.protect_future(cooperative_indexing.cooperative_indexing_period())
.await,
)
} else {
None
};
let indexing_period_opt = if let Some(indexing_cycle) = &self.indexing_cycle_opt {
Some(ctx.protect_future(indexing_cycle.indexing_period()).await)
} else {
None
};

let last_delete_opstamp_request = LastDeleteOpstampRequest {
index_uid: Some(self.pipeline_id.index_uid.clone()),
Expand Down Expand Up @@ -239,7 +236,7 @@ impl IndexerState {
publish_lock,
last_delete_opstamp,
memory_usage: GaugeGuard::new(&IN_FLIGHT_INDEX_WRITER, 0.0),
cooperative_indexing_period,
indexing_period_opt,
split_builders_guard,
};
Ok(workbench)
Expand Down Expand Up @@ -290,7 +287,7 @@ impl IndexerState {
.get_or_create_workbench(indexing_workbench_opt, ctx)
.await?;
if publish_lock.is_dead() {
// Release indexing permit early.
// Release indexing permit early if there is one.
indexing_workbench_opt.take();
return Ok(());
}
Expand Down Expand Up @@ -366,7 +363,7 @@ struct IndexingWorkbench {
// Number of bytes declared as used by tantivy.
memory_usage: GaugeGuard,
split_builders_guard: GaugeGuard,
cooperative_indexing_period: Option<CooperativeIndexingPeriod>,
indexing_period_opt: Option<IndexingPeriod>,
}

pub struct Indexer {
Expand Down Expand Up @@ -402,8 +399,8 @@ impl Actor for Indexer {
}

async fn initialize(&mut self, ctx: &ActorContext<Self>) -> Result<(), ActorExitStatus> {
if let Some(cooperative_indexing_cycle) = &self.indexer_state.cooperative_indexing_opt {
let initial_sleep_duration = cooperative_indexing_cycle.initial_sleep_duration();
if let Some(indexing_cycle) = &self.indexer_state.indexing_cycle_opt {
let initial_sleep_duration = indexing_cycle.initial_sleep_duration();
ctx.pause();
ctx.schedule_self_msg(initial_sleep_duration, Command::Resume);
}
Expand All @@ -418,9 +415,7 @@ impl Actor for Indexer {
return Ok(());
};

let Some(cooperative_indexing_period) =
indexing_workbench.cooperative_indexing_period.take()
else {
let Some(indexing_period) = indexing_workbench.indexing_period_opt.take() else {
return Ok(());
};

Expand All @@ -430,9 +425,9 @@ impl Actor for Indexer {
.map(|split| split.split_attrs.uncompressed_docs_size_in_bytes)
.sum::<u64>();

// This also drops the indexing permit.
// This also drops the indexing permit, if any.
let (sleep_duration, pipeline_metrics) =
cooperative_indexing_period.end_of_work(uncompressed_num_bytes);
indexing_period.end_of_work(uncompressed_num_bytes);

self.counters.pipeline_metrics_opt = Some(pipeline_metrics);

Expand Down Expand Up @@ -532,6 +527,7 @@ impl Indexer {
indexing_directory: TempDirectory,
indexing_settings: IndexingSettings,
cooperative_indexing_permits_opt: Option<Arc<Semaphore>>,
spread_indexing_pipelines: bool,
index_serializer_mailbox: Mailbox<IndexSerializer>,
fingerprinter_opt: Option<Fingerprinter>,
) -> Self {
Expand All @@ -548,12 +544,12 @@ 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<CooperativeIndexingCycle> =
cooperative_indexing_permits_opt.map(|cooperative_indexing_permits| {
CooperativeIndexingCycle::new(
let indexing_cycle_opt: Option<IndexingCycle> =
(spread_indexing_pipelines || cooperative_indexing_permits_opt.is_some()).then(|| {
IndexingCycle::new(
Comment thread
loutPhilipps marked this conversation as resolved.
&pipeline_id,
indexing_settings.commit_timeout(),
cooperative_indexing_permits,
cooperative_indexing_permits_opt,
)
});
Comment on lines +547 to 554

@loutPhilipps loutPhilipps Aug 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is where we enable the IndexingCycle, either if spreading pipelines or if cooperative indexing is enabled

Self {
Expand All @@ -569,7 +565,7 @@ impl Indexer {
tokenizer_manager: tokenizer_manager.tantivy_manager().clone(),
index_settings,
max_num_partitions: doc_mapper.max_num_partitions(),
cooperative_indexing_opt,
indexing_cycle_opt,
},
index_serializer_mailbox,
indexing_workbench_opt: None,
Expand Down Expand Up @@ -778,6 +774,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -919,6 +916,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -997,6 +995,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1081,6 +1080,7 @@ mod tests {
indexing_directory,
indexing_settings,
Some(Arc::new(Semaphore::new(1))),
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1170,6 +1170,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1238,6 +1239,7 @@ mod tests {
TempDirectory::for_test(),
IndexingSettings::for_test(),
None,
false,
index_serializer_mailbox,
Some(Fingerprinter::new(
&serde_json::from_value::<DocsClusteringConfig>(serde_json::json!([
Expand Down Expand Up @@ -1358,6 +1360,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1457,6 +1460,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1530,6 +1534,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1604,6 +1609,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1670,6 +1676,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down Expand Up @@ -1740,6 +1747,7 @@ mod tests {
indexing_directory,
indexing_settings,
None,
false,
index_serializer_mailbox,
None,
);
Expand Down
Loading
Loading