Skip to content

fix: set maturity on parquet splits #6399

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
mattmkim wants to merge 5 commits into
base: main
Choose a base branch
from
Open

fix: set maturity on parquet splits #6399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
af08ca9
set maturity on parquet splits
mattmkim May 6, 2026
ed3ca35
linter
mattmkim May 6, 2026
4cb24ac
Merge remote-tracking branch 'origin/main' into matthew.kim/byoc-metr…
mattmkim May 29, 2026
32496e2
fixes
mattmkim May 29, 2026
a890db9
fixes
mattmkim May 29, 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 quickwit/Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

4 changes: 4 additions & 0 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/indexing_service_impl.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,9 @@ impl IndexingService {
let partition_key = RoutingExpr::new(partition_key_str).map_err(|error| {
IndexingError::Internal(format!("failed to parse partition_key: {error}"))
})?;
let parquet_merge_policy = crate::merge_policy::parquet_merge_policy_from_settings(
&index_config.indexing_settings,
);

// Spawn the Parquet merge pipeline (or reuse an existing one for this
// index). The planner mailbox is wired into the MetricsPipeline's
Expand Down Expand Up @@ -97,6 +100,7 @@ impl IndexingService {
use_sketch_processors,
partition_key,
max_num_partitions: index_config.doc_mapping.max_num_partitions,
parquet_merge_policy,
parquet_merge_planner_mailbox_opt: Some(merge_planner_mailbox),
};
let pipeline = MetricsPipeline::new(pipeline_params);
Expand Down
3 changes: 3 additions & 0 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_doc_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -591,6 +591,9 @@ mod tests {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
let (uploader_mailbox, _uploader_handle) = universe.spawn_builder().spawn(uploader);

Expand Down
6 changes: 6 additions & 0 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_e2e_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -177,6 +177,9 @@ async fn test_metrics_pipeline_e2e() {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
let (uploader_mailbox, _uploader_handle) = universe.spawn_builder().spawn(uploader);

Expand Down Expand Up @@ -525,6 +528,9 @@ async fn test_sketch_pipeline_e2e() {
ram_storage.clone(),
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
let (uploader_mailbox, _uploader_handle) = universe.spawn_builder().spawn(uploader);

Expand Down
9 changes: 9 additions & 0 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_indexer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -823,6 +823,9 @@ mod tests {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
universe.spawn_builder().spawn(uploader)
}
Expand All @@ -849,6 +852,9 @@ mod tests {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
universe.spawn_builder().spawn(uploader)
}
Expand Down Expand Up @@ -1614,6 +1620,9 @@ mod tests {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);

Expand Down
17 changes: 12 additions & 5 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ use quickwit_parquet_engine::split::ParquetSplitMetadata;
use quickwit_proto::metastore::{MetastoreService, MetastoreServiceClient};
use quickwit_proto::types::IndexUid;
use quickwit_storage::Storage;
use time::OffsetDateTime;
use tokio::sync::Semaphore;
use tracing::{debug, error, info, instrument};

Expand Down Expand Up @@ -302,6 +303,7 @@ impl ParquetMergePipeline {
self.params.storage.clone(),
sequencer_mailbox,
self.params.max_concurrent_split_uploads,
self.params.merge_policy.clone(),
);
let (merge_uploader_mailbox, merge_uploader_handle) = ctx
.spawn_actor()
Expand Down Expand Up @@ -420,15 +422,16 @@ impl ParquetMergePipeline {
Ok(())
}

/// Fetch published Parquet splits from the metastore for merge planning.
/// Fetch immature published Parquet splits from the metastore for merge planning.
///
/// On first spawn, uses the initial splits provided by the IndexingService
/// (avoids per-pipeline metastore queries when many pipelines start).
/// On subsequent spawns (after crash/respawn), queries the metastore
/// directly to recover splits that were in-flight during the crash.
///
/// The planner's `record_splits_if_necessary` filters out mature splits,
/// so we don't need to filter here.
/// The planner re-checks maturity before recording splits, so this DB-side
/// filter is an optimization and crash-recovery guard, not the only line of
/// defense.
async fn fetch_immature_splits(
&mut self,
ctx: &ActorContext<Self>,
Expand All @@ -441,7 +444,8 @@ impl ParquetMergePipeline {
// Dispatch to the correct RPC based on whether this is a metrics or
// sketches index — they use separate Postgres tables.
let index_uid = self.params.index_uid.clone();
let query = quickwit_metastore::ListParquetSplitsQuery::for_index(index_uid.clone());
let query = quickwit_metastore::ListParquetSplitsQuery::for_index(index_uid.clone())
.retain_immature(OffsetDateTime::now_utc());
let is_sketch = quickwit_common::is_sketches_index(&index_uid.index_id);
let records = if is_sketch {
let list_request = quickwit_proto::metastore::ListSketchSplitsRequest::try_from_query(
Expand Down Expand Up @@ -628,7 +632,7 @@ mod tests {
use quickwit_actors::{ActorExitStatus, Universe};
use quickwit_common::temp_dir::TempDirectory;
use quickwit_parquet_engine::merge::policy::{
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig,
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig, ParquetSplitMaturity,
};
use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};
use quickwit_proto::metastore::{MetastoreServiceClient, MockMetastoreService};
Expand Down Expand Up @@ -680,6 +684,9 @@ mod tests {
.sort_fields("metric_name|host|timestamp_secs/V2")
.window_start_secs(0)
.window_duration_secs(3600)
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.build()
}

Expand Down
5 changes: 4 additions & 1 deletion ...quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline_multi_metric_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ use quickwit_common::temp_dir::TempDirectory;
use quickwit_common::test_utils::wait_until_predicate;
use quickwit_metastore::StageParquetSplitsRequestExt;
use quickwit_parquet_engine::merge::policy::{
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig,
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig, ParquetSplitMaturity,
};
use quickwit_parquet_engine::sorted_series::SORTED_SERIES_COLUMN;
use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};
Expand Down Expand Up @@ -126,6 +126,9 @@ fn make_multi_metric_split_metadata(
.sort_fields(table_config.effective_sort_fields())
.window_start_secs(0)
.window_duration_secs(900)
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.rg_partition_prefix_len(prefix_len);
for metric in metric_names {
builder = builder.add_metric_name(*metric);
Expand Down
8 changes: 7 additions & 1 deletion quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_pipeline_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ use quickwit_common::temp_dir::TempDirectory;
use quickwit_common::test_utils::wait_until_predicate;
use quickwit_metastore::StageParquetSplitsRequestExt;
use quickwit_parquet_engine::merge::policy::{
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig,
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig, ParquetSplitMaturity,
};
use quickwit_parquet_engine::sorted_series::SORTED_SERIES_COLUMN;
use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};
Expand Down Expand Up @@ -165,6 +165,9 @@ pub(super) fn make_test_split_metadata(
.sort_fields(table_config.effective_sort_fields())
.window_start_secs(0)
.window_duration_secs(900)
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.add_metric_name(metric_name)
.build()
}
Expand All @@ -188,6 +191,9 @@ pub(super) fn make_test_sketch_split_metadata(
.sort_fields(table_config.effective_sort_fields())
.window_start_secs(0)
.window_duration_secs(900)
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.add_metric_name(metric_name)
.build()
}
Expand Down
27 changes: 14 additions & 13 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_merge_planner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -259,21 +259,16 @@ impl ParquetMergePlanner {
fn record_splits_if_necessary(&mut self, splits: Vec<ParquetSplitMetadata>) {
let now = std::time::SystemTime::now();
for split in splits {
match self
if let ParquetSplitMaturity::Mature = self
.merge_policy
.split_maturity(split.size_bytes, split.num_merge_ops)
{
ParquetSplitMaturity::Mature => continue,
ParquetSplitMaturity::Immature {
maturation_period, ..
} => {
// A split that has lived past its maturation period is
// effectively mature — no further merges needed. This
// mirrors the Tantivy merge planner's `is_mature(now)`.
if split.created_at + maturation_period <= now {
continue;
}
}
// This can happen if the merge policy changed (e.g. decreased
// target_split_size_bytes or max_merge_ops).
continue;
}
if split.is_mature(now) {
continue;
Comment on lines +270 to +271
Copy link
Copy Markdown

@chatgpt-codex-connector chatgpt-codex-connector Bot May 29, 2026

Choose a reason for hiding this comment

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

P2 Badge Recompute maturity for legacy parquet splits

When a cluster restarts after upgrading, already-published parquet splits serialized before this new maturity field are deserialized with the field's default (Mature), so this split.is_mature(now) check discards them even if the merge policy check immediately above classifies their size/merge count as still immature. This hits the exact upgrade path where existing splits were created before the uploader started populating maturity, leaving those small legacy splits out of compaction; use the policy-derived maturation period for missing/defaulted metadata or otherwise distinguish legacy records instead of treating them as mature here.

Useful? React with 👍 / 👎.

}
if !self.acknowledge_split(split.split_id.as_str()) {
continue;
Expand Down Expand Up @@ -435,7 +430,7 @@ mod tests {

use quickwit_actors::Universe;
use quickwit_parquet_engine::merge::policy::{
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig,
ConstWriteAmplificationParquetMergePolicy, ParquetMergePolicyConfig, ParquetSplitMaturity,
};
use quickwit_parquet_engine::split::{ParquetSplitId, ParquetSplitMetadata, TimeRange};

Expand All @@ -453,6 +448,9 @@ mod tests {
.sort_fields("metric_name|host|timestamp_secs/V2")
.window_start_secs(0)
.window_duration_secs(3600)
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.num_merge_ops(num_merge_ops)
.build()
}
Expand Down Expand Up @@ -690,6 +688,9 @@ mod tests {
.num_rows(100)
.size_bytes(size_bytes)
.sort_fields("metric_name|host|timestamp_secs/V2")
.maturity(ParquetSplitMaturity::Immature {
maturation_period: Duration::from_secs(3600),
})
.num_merge_ops(num_merge_ops)
.build();

Expand Down
3 changes: 3 additions & 0 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_packager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -284,6 +284,9 @@ mod tests {
ram_storage,
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);
universe.spawn_builder().spawn(uploader)
}
Expand Down
39 changes: 37 additions & 2 deletions quickwit/quickwit-indexing/src/actors/parquet_pipeline/parquet_uploader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ use quickwit_common::spawn_named_task;
use quickwit_dst::events::merge_pipeline::{MergePipelineEvent, record_merge_pipeline_event};
use quickwit_metastore::StageParquetSplitsRequestExt;
use quickwit_metrics::{gauge, label_values};
use quickwit_parquet_engine::merge::policy::ParquetMergePolicy;
use quickwit_parquet_engine::split::{ParquetSplitKind, ParquetSplitMetadata};
use quickwit_proto::metastore::{MetastoreService, MetastoreServiceClient};
use quickwit_storage::Storage;
Expand Down Expand Up @@ -95,6 +96,7 @@ pub struct ParquetUploader {
split_store: Arc<dyn Storage>,
sequencer_mailbox: Mailbox<Sequencer<Publisher>>,
max_concurrent_uploads: usize,
merge_policy: Arc<dyn ParquetMergePolicy>,
counters: UploaderCounters,
}

Expand All @@ -106,13 +108,15 @@ impl ParquetUploader {
split_store: Arc<dyn Storage>,
sequencer_mailbox: Mailbox<Sequencer<Publisher>>,
max_concurrent_uploads: usize,
merge_policy: Arc<dyn ParquetMergePolicy>,
) -> Self {
Self {
uploader_type,
metastore,
split_store,
sequencer_mailbox,
max_concurrent_uploads,
merge_policy,
counters: Default::default(),
}
}
Expand Down Expand Up @@ -227,13 +231,14 @@ impl Handler<ParquetSplitBatch> for ParquetUploader {
// Clone what we need for the async task
let metastore = self.metastore.clone();
let split_store = self.split_store.clone();
let merge_policy = self.merge_policy.clone();
let counters = self.counters.clone();

let output_dir = batch.output_dir;
let checkpoint_delta_opt = batch.checkpoint_delta_opt;
let publish_lock = batch.publish_lock;
let publish_token_opt = batch.publish_token_opt;
let splits = batch.splits;
let mut splits = batch.splits;
let replaced_split_ids = batch.replaced_split_ids;
let merge_task_opt = batch._merge_task_opt;
// Hold the scratch directory alive until the upload task completes.
Expand All @@ -257,6 +262,11 @@ impl Handler<ParquetSplitBatch> for ParquetUploader {
return;
}

for split in &mut splits {
split.maturity =
merge_policy.split_maturity(split.size_bytes, split.num_merge_ops);
}

// Stage splits in metastore based on split type
let stage_result =
stage_splits(metastore.clone(), index_uid.clone(), &splits).await;
Expand Down Expand Up @@ -444,7 +454,20 @@ mod tests {
let mut mock_metastore = MockMetastoreService::new();
mock_metastore
.expect_stage_metrics_splits()
.withf(|request| request.index_uid().index_id == "test-index")
.withf(|request| {
if request.index_uid().index_id != "test-index" {
return false;
}
let splits = request.deserialize_splits_metadata().unwrap();
matches!(
splits.as_slice(),
[split]
if matches!(
split.maturity,
quickwit_parquet_engine::merge::policy::ParquetSplitMaturity::Immature { .. }
)
)
})
.times(1)
.returning(|_| Ok(EmptyResponse {}));

Expand All @@ -455,6 +478,9 @@ mod tests {
ram_storage.clone(),
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);

let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);
Expand Down Expand Up @@ -544,6 +570,9 @@ mod tests {
ram_storage.clone(),
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);

let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);
Expand Down Expand Up @@ -631,6 +660,9 @@ mod tests {
ram_storage.clone(),
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);

let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);
Expand Down Expand Up @@ -697,6 +729,9 @@ mod tests {
ram_storage.clone(),
sequencer_mailbox,
4,
crate::merge_policy::parquet_merge_policy_from_settings(
&quickwit_config::IndexingSettings::default(),
),
);

let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);
Expand Down
Loading
Loading