diff --git a/crates/dbsp/src/circuit/dbsp_handle.rs b/crates/dbsp/src/circuit/dbsp_handle.rs index 7fed588398a..aa762e386cd 100644 --- a/crates/dbsp/src/circuit/dbsp_handle.rs +++ b/crates/dbsp/src/circuit/dbsp_handle.rs @@ -3587,10 +3587,19 @@ pub(crate) mod tests { fn test_is_compaction_complete() { use crate::circuit::GlobalNodeId; use crate::circuit::metadata::{MetaItem, SPINE_BATCHES_COUNT}; + use crate::trace::spine_async::MIN_LEVEL0_MERGE_BATCHES; use crate::utils::Tup2; use std::time::Duration; - const BATCHES: usize = 30; + // Stay below the number of batches that makes a spine merge on its own, + // so that every batch is still waiting when compaction is requested. + // Feeding more than that races the background merger, which is free to + // reduce a spine to a single batch and leave compaction nothing to do. + const BATCHES: i32 = MIN_LEVEL0_MERGE_BATCHES as i32 - 1; + + // Small batches keep every batch in level 0, where + // `MIN_LEVEL0_MERGE_BATCHES` applies. 500 records also spread over + // both workers, so both spines hold `BATCHES` batches. const RECORDS_PER_BATCH: i32 = 500; let (mut dbsp, input_handle) = @@ -3602,7 +3611,7 @@ pub(crate) mod tests { .unwrap(); // Feed enough batches to give each spine more than one batch to merge. - for batch in 0..BATCHES as i32 { + for batch in 0..BATCHES { let mut tuples: Vec<_> = (0..RECORDS_PER_BATCH) .map(|r| Tup2(batch * RECORDS_PER_BATCH + r, Tup2(r, 1))) .collect(); diff --git a/crates/dbsp/src/trace/spine_async.rs b/crates/dbsp/src/trace/spine_async.rs index 79b204d7f1c..a9b879c3b50 100644 --- a/crates/dbsp/src/trace/spine_async.rs +++ b/crates/dbsp/src/trace/spine_async.rs @@ -112,6 +112,13 @@ static LEVEL_NAMES: [&str; MAX_LEVELS] = [ /// Configurable via `dev_tweaks.max_level0_batch_size_records`. pub(crate) const MAX_LEVEL0_BATCH_SIZE_RECORDS: u16 = 14_999; +/// Number of loose level-0 batches at which the spine starts a background +/// merge, i.e., `MERGE_COUNTS[0].start()`. +/// +/// A spine holding fewer level-0 batches than this merges nothing until +/// something else asks it to, such as a compaction request or memory pressure. +pub(crate) const MIN_LEVEL0_MERGE_BATCHES: usize = 8; + fn scope_tokio_merger_locals( worker_index: usize, buffer_cache: Arc, @@ -264,7 +271,7 @@ where /// The minimum number of batches to merge is key to performance. The /// maximum number seems much less important. const MERGE_COUNTS: [RangeInclusive; MAX_LEVELS] = [ - 8..=128, + MIN_LEVEL0_MERGE_BATCHES..=128, 8..=64, 3..=64, 3..=64,