lsm_tree/compaction/planned/mod.rs
1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026-present, Dmitry Prudnikov
3
4//! Compaction planned for the throughput queries get over a compaction round.
5//!
6//! The goal is the queries served, rather than the least write or read
7//! amplification (Wang et al., "Rethinking The Compaction Policies in
8//! LSM-trees", SIGMOD 2025). Compactions and queries draw on the same CPU and
9//! I/O, so a merge is an investment: it costs the queries that run while it
10//! does, and it pays them back in a faster read for as long as its output
11//! lasts. The strategy lays the tree out in three levels by what merging them
12//! buys:
13//!
14//! - **top** (L0): flushed runs, never merged among themselves. Once they
15//! hold a unit `S` of bytes they go down together as one run of the main
16//! level.
17//! - **main** (L1): runs of a unit or more. Between two units coming down, the
18//! newest runs may be merged into one, as the round's plan says.
19//! - **last**: one run. Once `R` units have come down in a round, the main
20//! level's runs move as they are to L2, which ends the round, and are merged
21//! from there into the last. That merge holds neither the top nor the main
22//! level, so the next round's units come down while it runs.
23//!
24//! The plan of a round is worked out by dynamic programming over what the
25//! merges cost and what they buy (see [`dp`]): early in the round a merge
26//! serves many queries and is worth more, late in it few, so the plan merges
27//! eagerly first and grows lazy. A unit is the last level's size over
28//! `C · R`, so the tree grows into larger units rather than more levels.
29//!
30//! What the plan is worked out from is a [`Model`] of the machine and the
31//! reads the tables recorded: the share of them that are range scans. The
32//! same tree, model and reads give the same plan, however often and in
33//! whichever process it is solved.
34//!
35//! # Compaction budget
36//!
37//! The engine does not cap how fast a tree is written:
38//! [`Config::write_rate_limit`](crate::Config::write_rate_limit) paces writes
39//! only while the tree slows them down for compaction to catch up. What it
40//! does cap, when the deployment asks, is compaction:
41//! [`Config::compaction_rate_limit`](crate::Config::compaction_rate_limit), or
42//! a limiter shared by the trees of one device, gives compaction its share of
43//! the device. That share is the budget, and the deployment sets it: how much
44//! of the device its reads and writes need is not something the tree can see.
45//!
46//! Every written byte is compacted again before it settles, so a budget that
47//! pays for less than the write rate times the write amplification leaves
48//! compaction behind and runs pile up. This strategy reads the budget and
49//! plans a tree it can keep: a round costs the size ratio plus two units of
50//! compaction per unit written, so the size ratio is cut to what the budget
51//! over [`Model::write_rate`] pays for, down to one. Set the model's write
52//! rate to what the tree takes in.
53//!
54//! A [write budget](crate::write_budget) on the medium of the last level caps
55//! the budget too: over the long run compaction may write the allowance's rate,
56//! less the write rate when L0 is on the same budget and its flushes spend
57//! from it. The
58//! plan then keeps a tree that budget pays for, rather than one whose merges
59//! the budget's gate would hold back.
60//!
61//! One measurement, made with `db_bench planned-sustained` (3M rows of
62//! 16-byte keys and 100-byte values, uncompressed, written at 8 MiB/s with
63//! point reads and scans between flushes, two compaction threads per tree;
64//! three runs, planned against [`Leveled`](crate::compaction::Leveled) at the
65//! same budget):
66//!
67//! | Budget | Queries/s | Scan p99 | Point read p99 | Compaction thread time | Mean runs |
68//! |---|---|---|---|---|---|
69//! | 8× the write rate | +7% to +14% | −11% to −21% | −8% to +7% | about a third | 4.7–5.0 vs 6.6–7.0 |
70//! | 2× the write rate | +6% to +13% | −5% to +10% | +7% to +18% | about half | 11.9–12.1 vs 16.0–17.1 |
71//!
72//! At eight times the write rate the planner wrote more compaction bytes per
73//! written byte than leveled (7.6–8.0 vs 7.0–7.3), at two times fewer
74//! (3.6–4.1 vs 4.5–4.6). These are figures of one machine and one workload:
75//! another device, key and value size, read mix or budget can move each of
76//! them either way. Measure the workload that runs, the bench above included.
77//!
78//! # A model calibrated from observation
79//!
80//! The [`Model`] is set, not learned. A model moved toward what a tree
81//! observes (its merges' rate, the filters' false positive rate, the speed
82//! queries keep beside a merge, a scan's cost per run against a point read's)
83//! was measured against it with `db_bench planned-sustained`: 1M rows, four
84//! read mixes (even points and 100-key scans, Zipf-hot points with short
85//! scans, 1000-key scans, 32 runs in L0), both budgets, three runs, each
86//! model judged on steps it was not fitted to. It won every run on two
87//! (Zipf-hot 2× by 12% to 129%, 32 L0 runs 8× by 13% to 80%), lost every run
88//! on one (Zipf-hot 8× by 24% to 46%), and went either way on the other five.
89//! Its plan changed 4 to 27 times a run where the set model's never did, and
90//! both models' predicted query cost erred by 40% to 110% on the held-out
91//! steps: the main level's run count does not explain what a query costs.
92//! Queries beside a merge kept their speed, so the share they keep fitted to
93//! its bound of one. The set model is kept. With 32 runs in L0, leveled
94//! served the most queries in two runs of three at either budget.
95
96pub mod dp;
97
98use super::{Choice, CompactionStrategy, Input as CompactionInput};
99use crate::compaction::state::CompactionState;
100use crate::config::Config;
101use crate::table::Table;
102use crate::version::Version;
103use crate::{HashSet, TableId};
104use alloc::sync::Arc;
105use alloc::vec::Vec;
106use portable_atomic::{AtomicU64, Ordering};
107
108#[cfg(feature = "std")]
109use parking_lot::Mutex;
110// no-std: spin mirrors parking_lot's Mutex API without an allocator.
111#[cfg(not(feature = "std"))]
112use spin::Mutex;
113
114#[doc(hidden)]
115pub const NAME: &str = "PlannedCompaction";
116
117/// The unit runs a round takes by default.
118///
119/// Measured on sustained writes with point reads and scans under a compaction
120/// rate limit both strategies keep up with: 10 units served more queries than
121/// leveled at no worse p99, where 25 left an ended round's runs to raise the
122/// scan p99 by a sixth.
123pub const DEFAULT_ROUND: u32 = 10;
124
125/// The level an ended round's runs wait in to go into the last level.
126const DRAIN: usize = 2;
127#[expect(clippy::cast_possible_truncation, reason = "a small constant level")]
128const DRAIN_LEVEL: u8 = DRAIN as u8;
129
130/// What the planner knows of the machine a tree runs on.
131#[derive(Clone, Copy, Debug, PartialEq)]
132pub struct Model {
133 /// Bytes per second the tree takes in, which its flushes write. With a
134 /// compaction limiter shared by several trees, the rate they all take in
135 /// together: the budget is theirs together, so each plans what the shared
136 /// budget pays for over the shared writes.
137 pub write_rate: u64,
138 /// Bytes per second a merge in the main level encodes. The tree's
139 /// compaction rate limit caps it, when one is set: what a merge may use
140 /// is what the shared budget leaves it. Over [`Self::write_rate`] it also
141 /// bounds the size ratio a round can afford: a round costs the ratio plus
142 /// two units of compaction per unit written.
143 pub encode_rate: u64,
144 /// Bytes per second a merge copies what it does not encode.
145 pub copy_rate: u64,
146 /// The share of a merge's output it copies rather than encodes, as the
147 /// work reports of past merges measure it.
148 pub copied_share: f64,
149 /// The share of their speed queries keep while a merge in the main level
150 /// runs.
151 pub merge_query_share: f64,
152 /// The filters' false positive rate: what a point read pays per run its
153 /// key is not in.
154 pub false_positive_rate: f64,
155}
156
157impl Default for Model {
158 /// A write stream of 16 MiB/s, merges encoding at 256 MiB/s and copying
159 /// at 1 GiB/s, nothing copied, queries at half speed during a merge, and
160 /// filters of 1% false positives.
161 fn default() -> Self {
162 Self {
163 write_rate: 16 * 1_024 * 1_024,
164 encode_rate: 256 * 1_024 * 1_024,
165 copy_rate: 1_024 * 1_024 * 1_024,
166 copied_share: 0.0,
167 merge_query_share: 0.5,
168 false_positive_rate: 0.01,
169 }
170 }
171}
172
173/// Compaction planned for query throughput over a compaction round.
174///
175/// # Examples
176///
177/// ```
178/// use lsm_tree::compaction::Planned;
179///
180/// // Rounds of ten units; the last level holds three times the main.
181/// let strategy = Planned::default().with_round(10).with_size_ratio(3);
182/// # let _ = strategy;
183/// ```
184pub struct Strategy {
185 /// Unit runs per round (`R`).
186 round: u32,
187 /// The last level's size over the main level's at the end of a round
188 /// (`C`).
189 size_ratio: u32,
190 /// The smallest unit, and the target size of the tables written.
191 target_size: u64,
192 /// The L0 table count that sends the top level down whatever its size.
193 l0_file_threshold: u16,
194 model: Model,
195 /// The share of reads that are scans, when set rather than counted.
196 scan_share: Option<f64>,
197 /// The compaction budget the last choice saw, for the debt reported
198 /// between choices.
199 budget_seen: AtomicU64,
200 /// The plan last worked out, with what it was worked out from.
201 plan: Mutex<Option<(dp::Inputs, Arc<dp::Plan>)>>,
202}
203
204impl Default for Strategy {
205 fn default() -> Self {
206 Self {
207 round: DEFAULT_ROUND,
208 size_ratio: 3,
209 target_size: 64 * 1_024 * 1_024,
210 l0_file_threshold: super::leveled::DEFAULT_L0_FILE_THRESHOLD,
211 model: Model::default(),
212 scan_share: None,
213 budget_seen: AtomicU64::new(0),
214 plan: Mutex::new(None),
215 }
216 }
217}
218
219impl Strategy {
220 /// Sets the unit runs a round takes before it ends in the last level
221 /// (default [`DEFAULT_ROUND`], at least one and at most
222 /// [`dp::MAX_ROUND`]).
223 ///
224 /// A round rewrites the last level once per `R` units, `C` units per unit
225 /// written whatever `R` is; what `R` sets is how many runs an ended round
226 /// leaves to read until it is merged into the last, and how large a unit
227 /// is, the last level over `C · R`.
228 #[must_use]
229 pub fn with_round(mut self, units: u32) -> Self {
230 self.round = units.clamp(1, dp::MAX_ROUND);
231 self
232 }
233
234 /// Sets the last level's size over the main level's at the end of a
235 /// round (default 3, at most [`dp::MAX_SIZE_RATIO`]). A model whose
236 /// compaction rate cannot pay for it gets a smaller one (see
237 /// [`Model::encode_rate`]).
238 #[must_use]
239 pub fn with_size_ratio(mut self, ratio: u32) -> Self {
240 self.size_ratio = ratio.clamp(1, dp::MAX_SIZE_RATIO);
241 self
242 }
243
244 /// Sets the target size of the tables written, which is also the
245 /// smallest unit (default 64 MiB).
246 #[must_use]
247 pub fn with_table_target_size(mut self, bytes: u64) -> Self {
248 self.target_size = bytes.max(1);
249 self
250 }
251
252 /// Sets the L0 table count that sends the top level down whatever its
253 /// size (default as [`Leveled`](super::Leveled)'s).
254 #[must_use]
255 pub fn with_l0_file_threshold(mut self, tables: u16) -> Self {
256 self.l0_file_threshold = tables.max(1);
257 self
258 }
259
260 /// Sets what the planner knows of the machine.
261 #[must_use]
262 pub fn with_model(mut self, model: Model) -> Self {
263 self.model = model;
264 self
265 }
266
267 /// Plans for `share` of the reads being range scans (clamped to `0..=1`)
268 /// instead of the share the tree's tables counted: a stand-in for a
269 /// workload known ahead, or not yet seen. `None` reads the tables again.
270 #[must_use]
271 pub fn with_scan_share(mut self, share: Option<f64>) -> Self {
272 self.scan_share = share.map(|share| share.clamp(0.0, 1.0));
273 self
274 }
275
276 /// Unit runs per round.
277 const fn round(&self) -> u32 {
278 self.round
279 }
280
281 /// The last level's size over the main level's at the end of a round, as
282 /// far as the model's compaction rate pays for it.
283 ///
284 /// A round writes each unit down once and rewrites it into the last level
285 /// with `C` more, `C + 2` units of compaction per unit written: a rate
286 /// below that falls behind every round, and what it cannot do piles up in
287 /// L0. The ratio shrinks to what the rate pays, down to 1, never past the
288 /// one set. The rate is the model's, capped by `budget`, the bytes per
289 /// second the tree's compactions may use (`0` for no limit).
290 fn size_ratio(&self, budget: u64) -> u32 {
291 let paid = within_budget(self.model.encode_rate, budget) / self.model.write_rate.max(1);
292 // Below two units per unit written no ratio is paid for; the floor of
293 // one is the cheapest round there is.
294 let paid = u32::try_from(paid.saturating_sub(2)).unwrap_or(u32::MAX);
295 paid.clamp(1, self.size_ratio)
296 }
297
298 /// The unit of a tree whose last level holds `last_bytes`, compacting
299 /// within `budget`.
300 fn unit(&self, last_bytes: u64, budget: u64) -> u64 {
301 (last_bytes / (u64::from(self.size_ratio(budget)) * u64::from(self.round())))
302 .max(self.target_size)
303 }
304
305 /// The plan for `inputs`, solved once for the same inputs.
306 fn plan(&self, inputs: dp::Inputs) -> Arc<dp::Plan> {
307 let mut cached = self.plan.lock();
308 if let Some((solved_for, plan)) = cached.as_ref()
309 && *solved_for == inputs
310 {
311 return Arc::clone(plan);
312 }
313 let plan = Arc::new(dp::Plan::solve(&inputs));
314 *cached = Some((inputs, Arc::clone(&plan)));
315 plan
316 }
317
318 /// What the plan for `version` is worked out from, with `budget` bytes
319 /// per second the tree's compactions may use, `0` for no limit.
320 fn inputs(&self, version: &Version, unit: u64, budget: u64) -> dp::Inputs {
321 let scan_share = self.scan_share.unwrap_or_else(|| {
322 // Queries as the tree served them: a lookup whatever the filters
323 // answered, a scan, row or projected, however many runs it reached.
324 let (points, scans) = version.read_mix().counts();
325 #[expect(
326 clippy::cast_precision_loss,
327 reason = "a share of read counts, approximate by nature"
328 )]
329 let scans = scans as f64;
330 // With no read seen there is nothing to plan for: no merge in the
331 // main level pays, and none runs.
332 #[expect(
333 clippy::cast_precision_loss,
334 reason = "a share of read counts, approximate by nature"
335 )]
336 let points = points as f64;
337 if scans + points == 0.0 {
338 0.0
339 } else {
340 scans / (scans + points)
341 }
342 });
343 let model = self.model;
344 // A copied group is read through the same limiter as an encoded one.
345 let encode_rate = within_budget(model.encode_rate, budget).max(1);
346 let copy_rate = within_budget(model.copy_rate, budget).max(1);
347 #[expect(
348 clippy::cast_precision_loss,
349 reason = "seconds of a unit, approximate by nature"
350 )]
351 let (unit, write_rate, copy_rate, encode_rate) = (
352 unit as f64,
353 model.write_rate.max(1) as f64,
354 copy_rate as f64,
355 encode_rate as f64,
356 );
357 let copied = model.copied_share.clamp(0.0, 1.0);
358 dp::Inputs {
359 round: self.round(),
360 size_ratio: self.size_ratio(budget),
361 arrival_secs: unit / write_rate,
362 merge_secs: unit * (1.0 - copied) / encode_rate + unit * copied / copy_rate,
363 merge_query_share: model.merge_query_share.clamp(0.0, 1.0),
364 scan_share,
365 false_positive_rate: model.false_positive_rate.clamp(0.0, 1.0),
366 }
367 }
368}
369
370/// The plan a [`Strategy`] would follow for one version, and what it was
371/// worked out from.
372#[derive(Clone, Debug)]
373pub struct PlanView {
374 /// What the plan is worked out from.
375 pub inputs: dp::Inputs,
376 /// The plan of the round.
377 pub plan: Arc<dp::Plan>,
378 /// The unit runs that have come down in the round so far.
379 pub units: u32,
380}
381
382impl Strategy {
383 /// The plan this strategy would follow for `version` under `config`'s
384 /// compaction budget, worked out as [`CompactionStrategy::choose`] works it
385 /// out but choosing nothing: for comparing models side by side on one
386 /// tree. `None` for a tree with no main level apart from the last, which
387 /// has no plan.
388 ///
389 /// # Examples
390 ///
391 /// ```
392 /// # use lsm_tree::{AbstractTree, Config, SequenceNumberCounter};
393 /// use lsm_tree::compaction::Planned;
394 ///
395 /// # let folder = tempfile::tempdir()?;
396 /// # let tree = Config::new(folder.path(), SequenceNumberCounter::default(), SequenceNumberCounter::default()).open()?;
397 /// let strategy = Planned::default();
398 /// let view = strategy
399 /// .plan_for(&tree.current_version(), tree.tree_config())
400 /// .expect("a tree has a main level apart from the last");
401 /// // Nothing has come down yet: the round is at its start.
402 /// assert_eq!(view.units, 0);
403 /// # Ok::<(), lsm_tree::Error>(())
404 /// ```
405 #[must_use]
406 pub fn plan_for(&self, version: &Version, config: &Config) -> Option<PlanView> {
407 let last = version.level_count() - 1;
408 if last < DRAIN {
409 return None;
410 }
411 let budget = self.budget_for(config, last);
412 let unit = self.unit(
413 version.level(last).map_or(0, crate::version::Level::size),
414 budget,
415 );
416 let bytes: u64 = version.level(1).map_or(0, |main| {
417 main_runs(main).iter().map(|(_, bytes)| bytes).sum()
418 });
419 let inputs = self.inputs(version, unit, budget);
420 Some(PlanView {
421 inputs,
422 plan: self.plan(inputs),
423 units: u32::try_from(bytes / unit).unwrap_or(u32::MAX),
424 })
425 }
426
427 /// The tree's compaction budget in bytes per second, `0` for no limit:
428 /// the tighter of its rate limit and the write budget of the medium of
429 /// `last`, its last level.
430 fn budget_for(&self, config: &Config, last: usize) -> u64 {
431 #[expect(clippy::cast_possible_truncation, reason = "level index below 255")]
432 tighter(
433 budget(config),
434 endurance_budget(config, last as u8, self.model.write_rate).unwrap_or(0),
435 )
436 }
437}
438
439/// The bytes a merge may encode or copy per second: the model's rate, capped
440/// by the tree's compaction `budget` when it has one (`0` for none).
441fn within_budget(model_rate: u64, budget: u64) -> u64 {
442 match budget {
443 0 => model_rate,
444 budget => model_rate.min(budget),
445 }
446}
447
448/// The tree's compaction budget in bytes per second, `0` for no limit: the
449/// live rate of its limiter. A limiter shared by several trees gives each the
450/// whole rate, which is why [`Model::write_rate`] is then their writes
451/// together.
452fn budget(config: &Config) -> u64 {
453 config
454 .live_compaction_rate_limiter
455 .as_ref()
456 .or(config.compaction_rate_limiter.as_ref())
457 .map_or(config.compaction_rate_limit, |limiter| limiter.rate())
458}
459
460/// The bytes per second compaction may write over the long run on the medium
461/// of `level`, `None` without a write budget there: the budget's allowance
462/// rate, less `write_rate` when L0, which the flushes write, is on the same
463/// budget and spends from it. An allowance the writes alone use up leaves
464/// compaction one byte a second, the least a budget can say, since `0` reads
465/// as no limit.
466fn endurance_budget(config: &Config, level: u8, write_rate: u64) -> Option<u64> {
467 let (_, fs) = config.tables_folder_for_level(level);
468 let budget = fs.write_budget()?;
469 let allowance = budget.allowance();
470 let per_sec = u128::from(allowance.bytes_per_period()) * 1_000_000_000
471 / allowance.period().as_nanos().max(1);
472 let per_sec = u64::try_from(per_sec).unwrap_or(u64::MAX);
473 let (_, l0_fs) = config.tables_folder_for_level(0);
474 let flushes_here = l0_fs
475 .write_budget()
476 .is_some_and(|l0| Arc::ptr_eq(l0, budget));
477 if !flushes_here {
478 return Some(per_sec.max(1));
479 }
480 // Below the write rate nothing is left: none, not an error.
481 Some(per_sec.saturating_sub(write_rate).max(1))
482}
483
484/// The tighter of two budgets in bytes per second, `0` standing for none.
485fn tighter(a: u64, b: u64) -> u64 {
486 match (a, b) {
487 (0, other) | (other, 0) => other,
488 (a, b) => a.min(b),
489 }
490}
491
492/// The runs of the main level newest first, each with its bytes.
493fn main_runs(main: &crate::version::Level) -> Vec<(Vec<TableId>, u64)> {
494 main.iter()
495 .map(|run| {
496 let bytes: u64 = run.iter().map(Table::file_size).sum();
497 (run.iter().map(Table::id).collect(), bytes)
498 })
499 .collect()
500}
501
502impl CompactionStrategy for Strategy {
503 fn get_name(&self) -> &'static str {
504 NAME
505 }
506
507 fn pending_compaction_bytes(&self, version: &Version) -> u64 {
508 use crate::version::Level;
509
510 let last = version.level_count() - 1;
511 // A tree of one level has no compaction to owe.
512 if last == 0 {
513 return 0;
514 }
515 let last_bytes = version.level(last).map_or(0, Level::size);
516 // The budget the last choice saw: the version alone does not say it.
517 let unit = self.unit(last_bytes, self.budget_seen.load(Ordering::Relaxed));
518 let top_level = version.level(0);
519 let top = top_level.map_or(0, Level::size);
520 // The top level past its unit owes the bytes past it; one past its
521 // table count owes all of it, as `choose` sends it all down then.
522 let top_owed = if top_level
523 .is_some_and(|level| level.table_count() >= usize::from(self.l0_file_threshold))
524 {
525 top
526 } else {
527 top.saturating_sub(unit)
528 };
529 // A tree of two levels owes only its top level: the main level is the
530 // last, which nothing drains.
531 if last < DRAIN {
532 return top_owed;
533 }
534 let main = version.level(1).map_or(0, Level::size);
535 let draining: u64 = (DRAIN..last)
536 .filter_map(|index| version.level(index))
537 .map(Level::size)
538 .sum();
539 // The top level past its unit, the main level past its round, and an
540 // ended round still to go into the last: a level within its bound
541 // owes nothing, not less than nothing.
542 top_owed + main.saturating_sub(unit * u64::from(self.round())) + draining
543 }
544
545 fn next_merge_level(&self, version: &Version, dest_level: u8) -> Option<usize> {
546 // A run of the main level is next merged whole with others, or into
547 // the last level whole: no boundary of another level is worth an
548 // output's cut.
549 let _ = (version, dest_level);
550 None
551 }
552
553 fn choose(&self, version: &Version, config: &Config, state: &CompactionState) -> Choice {
554 let last = version.level_count() - 1;
555 // A tree of one level has no level to compact into.
556 if last == 0 {
557 return Choice::DoNothing;
558 }
559 let hidden = state.hidden_set();
560 let busy = |index: usize| version.level_is_busy(index, hidden);
561 #[expect(
562 clippy::expect_used,
563 reason = "levels 0 to the last exist in every version, and the last is past 0"
564 )]
565 let (top, main, last_level) = (
566 version.level(0).expect("L0"),
567 version.level(1).expect("L1"),
568 version.level(last).expect("the last level"),
569 );
570 let budget = self.budget_for(config, last);
571 self.budget_seen.store(budget, Ordering::Relaxed);
572 let unit = self.unit(last_level.size(), budget);
573 let runs = main_runs(main);
574 // The units come down so far, counted by bytes: runs the L0 table
575 // count sent down before they held a unit count for what they hold.
576 // Counted from the version, not from arrivals kept beside it, so the
577 // plan is the same in every process: a round ends once the main level
578 // holds a round of units, as a leveled level within its target never
579 // goes down; overwrites a main-level merge collapses stay there,
580 // merged as reads ask, and tombstones, which only the last level drops,
581 // fill the round.
582 let units = runs.iter().map(|(_, bytes)| bytes).sum::<u64>() / unit;
583 #[expect(clippy::cast_possible_truncation, reason = "level index below 255")]
584 let last_index = last as u8;
585
586 let top_due = !top.is_empty()
587 && (top.size() >= unit || top.table_count() >= usize::from(self.l0_file_threshold));
588
589 // A tree of two levels has no main level apart from the last: the top
590 // level goes into it whole, merged with what it holds.
591 if last < DRAIN {
592 if top_due && !busy(0) && !busy(1) {
593 let mut table_ids = top.list_ids();
594 table_ids.extend(main.list_ids());
595 return Choice::Merge(CompactionInput {
596 table_ids,
597 dest_level: last_index,
598 canonical_level: last_index,
599 target_size: self.target_size,
600 });
601 }
602 return Choice::DoNothing;
603 }
604
605 // The top level holds a unit: it comes down as one run. First, as the
606 // cheapest work and the one that keeps flushes from piling up.
607 if !busy(0) && !busy(1) && top_due {
608 let choice = CompactionInput {
609 table_ids: top.list_ids(),
610 dest_level: 1,
611 canonical_level: 1,
612 target_size: self.target_size,
613 };
614 let cmp = config.comparator.as_ref();
615 let key_range = top.aggregate_key_range_cmp(cmp);
616 let overlaps = main
617 .iter()
618 .any(|run| !run.get_overlapping_cmp(&key_range, cmp).is_empty());
619 if top.run_count() == 1 && !overlaps {
620 return Choice::Move(choice);
621 }
622 return Choice::Merge(choice);
623 }
624
625 // An ended round's runs, and anything else left between the main and
626 // the last level, go into the last. The merge holds neither the top
627 // nor the main level, so units keep coming down while it runs.
628 let draining: HashSet<TableId> = (DRAIN..last)
629 .filter_map(|index| version.level(index))
630 .flat_map(crate::version::Level::list_ids)
631 .collect();
632 if draining.is_empty() {
633 // The round is over: its runs move out of the main level as they
634 // are, kept apart and in age order, so the next round starts on an
635 // empty one.
636 if units >= u64::from(self.round()) && !busy(1) && !busy(DRAIN) {
637 // A tree of three levels has no level between the main and
638 // the last to wait in: the round merges into the last at once.
639 if last == DRAIN {
640 let mut table_ids = main.list_ids();
641 table_ids.extend(last_level.list_ids());
642 return Choice::Merge(CompactionInput {
643 table_ids,
644 dest_level: last_index,
645 canonical_level: last_index,
646 target_size: self.target_size,
647 });
648 }
649 return Choice::Move(CompactionInput {
650 table_ids: main.list_ids(),
651 dest_level: DRAIN_LEVEL,
652 canonical_level: DRAIN_LEVEL,
653 target_size: self.target_size,
654 });
655 }
656 } else if (DRAIN..=last).all(|index| !busy(index)) {
657 let mut table_ids = draining;
658 table_ids.extend(last_level.list_ids());
659 return Choice::Merge(CompactionInput {
660 table_ids,
661 dest_level: last_index,
662 canonical_level: last_index,
663 target_size: self.target_size,
664 });
665 }
666
667 if busy(1) {
668 return Choice::DoNothing;
669 }
670 // The merge the plan makes after this many units came down.
671 let plan = self.plan(self.inputs(version, unit, budget));
672 let wanted = u64::from(plan.merge_after(u32::try_from(units).unwrap_or(u32::MAX)));
673 if wanted > 1 {
674 // At most a round of units, each at most the last level's bytes.
675 let wanted_bytes = wanted * unit;
676 let mut table_ids = HashSet::default();
677 let (mut merged_bytes, mut merged_runs) = (0u64, 0usize);
678 for (ids, run_bytes) in &runs {
679 if merged_bytes >= wanted_bytes {
680 break;
681 }
682 table_ids.extend(ids.iter().copied());
683 merged_bytes += run_bytes;
684 merged_runs += 1;
685 }
686 if merged_runs > 1 {
687 return Choice::Merge(CompactionInput {
688 table_ids,
689 dest_level: 1,
690 canonical_level: 1,
691 target_size: self.target_size,
692 });
693 }
694 }
695 Choice::DoNothing
696 }
697}
698
699#[cfg(test)]
700#[expect(clippy::expect_used, reason = "test code")]
701mod tests;