lsm_tree/compaction/leveled/mod.rs
1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2024-present, fjall-rs
3// Copyright (c) 2026-present, Dmitry Prudnikov
4
5#[cfg(test)]
6#[allow(
7 clippy::unwrap_used,
8 clippy::indexing_slicing,
9 clippy::useless_vec,
10 clippy::unnecessary_map_or,
11 reason = "test code"
12)]
13mod test;
14
15#[cfg(test)]
16#[expect(clippy::unwrap_used, clippy::indexing_slicing, reason = "test code")]
17mod picker_tests;
18
19#[cfg(test)]
20#[expect(clippy::expect_used, reason = "test code")]
21mod primitive_tests;
22
23mod primitives;
24
25pub use primitives::{Granularity, Layout, LevelPolicy, Movement, Trigger};
26
27use super::{Choice, CompactionStrategy, Input as CompactionInput};
28use crate::{
29 HashSet, TableId,
30 compaction::state::{CompactionState, hidden_set::HiddenSet},
31 config::Config,
32 table::{Table, util::aggregate_run_key_range},
33 version::{Level, Version, run::Ranged},
34};
35#[cfg(not(feature = "std"))]
36use alloc::vec::Vec;
37
38/// A merge candidate's cost: the bytes it moves down a level (`promoted`) and
39/// every byte it reads and rewrites to do so (`total`, the promoted bytes
40/// plus those it pulls in from the level below), and the reads its tables
41/// served (`heat`).
42#[derive(Clone, Copy, Debug, PartialEq, Eq)]
43struct MergeCost {
44 promoted: u64,
45 total: u64,
46 heat: u64,
47}
48
49impl MergeCost {
50 /// Whether `self` rewrites fewer bytes per promoted byte than `other`.
51 ///
52 /// Cross-multiplied in `u128`: two `u64` byte counts multiply into at
53 /// most 128 bits, so the comparison is exact for every size a level can
54 /// hold, where dividing first would lose the low bits.
55 fn cheaper_than(self, other: Self) -> bool {
56 u128::from(self.total) * u128::from(other.promoted)
57 < u128::from(other.total) * u128::from(self.promoted)
58 }
59
60 /// Whether `movement` prefers `self` to `other`: the least rewrite per
61 /// promoted byte, or the most reads per byte rewritten with the least
62 /// rewrite breaking a tie. Cross-multiplied in `u128` as
63 /// [`Self::cheaper_than`] is.
64 fn better_than(self, other: Self, movement: Movement) -> bool {
65 match movement {
66 Movement::LeastRewrite { .. } => self.cheaper_than(other),
67 Movement::Hottest => {
68 let ours = u128::from(self.heat) * u128::from(other.total);
69 let theirs = u128::from(other.heat) * u128::from(self.total);
70 ours > theirs || (ours == theirs && self.cheaper_than(other))
71 }
72 }
73 }
74}
75
76/// Streams merge candidates and keeps the one that pays down a level's debt
77/// best by the level's [`Movement`]: by default at the least rewriting per
78/// promoted byte.
79///
80/// Ranking by the total input alone prefers a small merge that promotes little
81/// into a dense level below, which has to be repeated many times to retire the
82/// same debt; the ratio ranks how much each promoted byte costs. Among the
83/// candidates that promote no more than `overshoot + slack`, the best one
84/// wins, so a level is not asked to push down far more than it owes; when none
85/// is that small, the best overall still makes progress. Ties go to the
86/// earliest candidate, so the choice is the same on every call.
87///
88/// Only the two leaders are held, whatever the number of candidates offered.
89struct MergeRanking<T> {
90 movement: Movement,
91 bound: u64,
92 offered: usize,
93 /// The cheapest candidate within the bound, with its offer index.
94 within: Option<(usize, MergeCost, T)>,
95 /// The cheapest candidate overall, with its offer index.
96 any: Option<(usize, MergeCost, T)>,
97 /// The cost of the cheapest candidate after `any`.
98 second: Option<MergeCost>,
99}
100
101impl<T: Copy> MergeRanking<T> {
102 fn new(overshoot: u64, slack: u64, movement: Movement) -> Self {
103 Self {
104 movement,
105 // A bound past `u64::MAX` admits every candidate, which is what
106 // clamping to `u64::MAX` means: no promoted byte count exceeds it.
107 bound: overshoot.saturating_add(slack),
108 offered: 0,
109 within: None,
110 any: None,
111 second: None,
112 }
113 }
114
115 fn offer(&mut self, cost: MergeCost, candidate: T) {
116 let index = self.offered;
117 let movement = self.movement;
118 self.offered += 1;
119 if cost.promoted <= self.bound
120 && self
121 .within
122 .as_ref()
123 .is_none_or(|(_, best, _)| cost.better_than(*best, movement))
124 {
125 self.within = Some((index, cost, candidate));
126 }
127 match &self.any {
128 Some((_, best, _)) if !cost.better_than(*best, movement) => {
129 if self
130 .second
131 .is_none_or(|second| cost.better_than(second, movement))
132 {
133 self.second = Some(cost);
134 }
135 }
136 previous => {
137 self.second = previous.as_ref().map(|(_, best, _)| *best);
138 self.any = Some((index, cost, candidate));
139 }
140 }
141 }
142
143 /// The chosen candidate with its offer index and cost, and the cost of the
144 /// cheapest candidate not chosen.
145 fn finish(self) -> Option<((usize, MergeCost, T), Option<MergeCost>)> {
146 match (self.within, self.any) {
147 (Some(within), Some(any)) if within.0 != any.0 => Some((within, Some(any.1))),
148 (Some(chosen), _) | (None, Some(chosen)) => Some((chosen, self.second)),
149 (None, None) => None,
150 }
151 }
152
153 fn offered(&self) -> usize {
154 self.offered
155 }
156}
157
158/// Tries to find the most optimal compaction set from one level into the other.
159///
160/// Scans all runs in both levels to handle transient multi-run states from
161/// multi-level compaction (#108). See #122 Part 3.
162fn pick_minimal_compaction(
163 curr_level: &Level,
164 next_level: &Level,
165 hidden_set: &HiddenSet,
166 overshoot: u64,
167 table_base_size: u64,
168 movement: Movement,
169 cmp: &dyn crate::comparator::UserComparator,
170) -> Option<(HashSet<TableId>, bool)> {
171 let promotion_slack = match movement {
172 Movement::LeastRewrite { slack } => slack.unwrap_or(table_base_size),
173 Movement::Hottest => table_base_size,
174 };
175 if let Some(ids) = largest_trivial_move(curr_level, next_level, hidden_set, cmp) {
176 return Some((ids, true));
177 }
178
179 // NOTE: Look for merges
180 // Iterate windows across all runs in next_level, pull in from all runs
181 // in curr_level.
182 if next_level.is_empty() {
183 return None;
184 }
185
186 // Candidates are ranked as they are enumerated, holding only a window's
187 // place and its cost for the leaders; the winner's tables are collected at
188 // the end.
189 let mut ranking = MergeRanking::new(overshoot, promotion_slack, movement);
190 let cap = 50 * table_base_size;
191 for (run_index, run) in next_level.iter().enumerate() {
192 let window_bytes = prefix_sums(run.iter().map(Table::file_size));
193 let window_hidden =
194 prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id()))));
195 let window_heat = prefix_sums(run.iter().map(Table::heat));
196 let pull_ins: Vec<PullIn> = curr_level
197 .iter()
198 .map(|curr_run| PullIn::new(curr_run, run, hidden_set, cmp))
199 .collect();
200 // Windows by width, then by start, the order that decides ties. Table
201 // sizes vary, so a window past the cap leaves the others of its width
202 // to be looked at; once every window of a width is past it, so is
203 // every wider one, which holds one of them.
204 for width in 1..=run.len() {
205 let mut within_cap = false;
206 for start in 0..=(run.len() - width) {
207 let end = start + width;
208 #[expect(
209 clippy::indexing_slicing,
210 reason = "start < end <= run.len(), and the sums hold run.len() + 1 entries"
211 )]
212 let (next_level_size, next_hidden, next_heat) = (
213 window_bytes[end] - window_bytes[start],
214 window_hidden[end] - window_hidden[start],
215 window_heat[end] - window_heat[start],
216 );
217 if next_level_size > cap {
218 continue;
219 }
220 within_cap = true;
221 if next_hidden > 0 {
222 continue;
223 }
224 let mut promoted = 0;
225 let mut heat = next_heat;
226 let mut blocked = false;
227 for pull_in in &pull_ins {
228 let (bytes, hidden, reads) = pull_in.contained(start, end - 1);
229 promoted += bytes;
230 heat += reads;
231 blocked |= hidden > 0;
232 }
233 if promoted == 0 || blocked {
234 continue;
235 }
236 ranking.offer(
237 MergeCost {
238 promoted,
239 total: promoted + next_level_size,
240 heat,
241 },
242 (run_index, start, end),
243 );
244 }
245 if !within_cap {
246 break;
247 }
248 }
249 }
250
251 let offered = ranking.offered();
252 let ((chosen, cost, (run_index, start, end)), runner_up) = ranking.finish()?;
253 log::debug!(
254 "leveled: merge candidate {chosen} of {offered} chosen at {cost:?}, runner-up \
255 {runner_up:?} (overshoot {overshoot}, slack {promotion_slack})",
256 );
257 #[expect(
258 clippy::expect_used,
259 reason = "the ranking holds a window it was offered from these runs"
260 )]
261 let window = next_level
262 .iter()
263 .nth(run_index)
264 .and_then(|run| run.get(start..end))
265 .expect("the chosen window lies in its run");
266 let key_range = aggregate_run_key_range(window);
267 let contained: HashSet<_> = curr_level
268 .iter()
269 .flat_map(|run| run.get_contained_cmp(&key_range, cmp))
270 .map(Table::id)
271 .collect();
272 let pulled_in = contained.len();
273 let curr_ids = with_overlapping_runs(curr_level, contained, hidden_set, cmp)?;
274 let mut below: HashSet<_> = window.iter().map(Table::id).collect();
275 // Every table below within the span of the merge goes into it too, from
276 // every run there and whether or not the level above added any: the
277 // window may lie in an older run of the level below, and its output
278 // becomes that level's front run, so a newer version left out under the
279 // span would read as older than what the merge wrote. The tables taken
280 // of each run lie side by side, none left between them.
281 if curr_ids.len() > pulled_in || next_level.run_count() > 1 {
282 #[expect(clippy::expect_used, reason = "the window holds a table")]
283 let span = span_of(
284 curr_level
285 .iter()
286 .flat_map(|run| run.iter())
287 .filter(|table| curr_ids.contains(&table.id()))
288 .chain(window.iter()),
289 cmp,
290 )
291 .expect("a span over the window");
292 for table in next_level
293 .iter()
294 .flat_map(|run| run.get_overlapping_cmp(&span, cmp))
295 {
296 if below.insert(table.id()) && hidden_set.is_hidden(table.id()) {
297 return None;
298 }
299 }
300 below = with_overlapping_runs(next_level, below, hidden_set, cmp)?;
301 }
302 let mut ids = below;
303 ids.extend(curr_ids);
304 Some((ids, false))
305}
306
307/// The key range from the least key of `tables` to the greatest.
308fn span_of<'a>(
309 tables: impl Iterator<Item = &'a Table>,
310 cmp: &dyn crate::comparator::UserComparator,
311) -> Option<crate::KeyRange> {
312 use core::cmp::Ordering;
313
314 let mut span: Option<(crate::UserKey, crate::UserKey)> = None;
315 for table in tables {
316 let range = table.key_range();
317 span = Some(match span {
318 None => (range.min().clone(), range.max().clone()),
319 Some((min, max)) => (
320 if cmp.compare(range.min(), &min) == Ordering::Less {
321 range.min().clone()
322 } else {
323 min
324 },
325 if cmp.compare(range.max(), &max) == Ordering::Greater {
326 range.max().clone()
327 } else {
328 max
329 },
330 ),
331 });
332 }
333 span.map(crate::KeyRange::new)
334}
335
336/// `ids`, tables of `level`, with every table of its runs within their span,
337/// taken in until none is left out; `None` when one is hidden.
338///
339/// The runs of a level hold versions of one key in more than one of them,
340/// and a read stops at the first level that holds the key: a merge that
341/// took the newer version down and left the older above would have reads
342/// return the older. The span, not only the keys shared, so that the tables
343/// taken of each run lie side by side, as a merge reads a run, none left
344/// between them.
345fn with_overlapping_runs(
346 level: &Level,
347 mut ids: HashSet<TableId>,
348 hidden_set: &HiddenSet,
349 cmp: &dyn crate::comparator::UserComparator,
350) -> Option<HashSet<TableId>> {
351 // The tables of one run share no key.
352 if level.run_count() <= 1 {
353 return Some(ids);
354 }
355 loop {
356 let Some(span) = span_of(
357 level
358 .iter()
359 .flat_map(|run| run.iter())
360 .filter(|table| ids.contains(&table.id())),
361 cmp,
362 ) else {
363 return Some(ids);
364 };
365 let mut grew = false;
366 for table in level
367 .iter()
368 .flat_map(|run| run.get_overlapping_cmp(&span, cmp))
369 {
370 if ids.insert(table.id()) {
371 if hidden_set.is_hidden(table.id()) {
372 return None;
373 }
374 grew = true;
375 }
376 }
377 if !grew {
378 return Some(ids);
379 }
380 }
381}
382
383/// The longest window of one run of `curr_level` that overlaps no table of
384/// `next_level`, shares no key with the other runs of `curr_level` and holds no
385/// hidden table, the leftmost of that length and from the first run that has
386/// one: a move that rewrites nothing.
387///
388/// A window's key range is its tables' ranges and the gaps between them, so it
389/// overlaps the level below exactly when one of its tables does or one of its
390/// gaps does. The windows that qualify are therefore the stretches of
391/// neighbouring tables that each qualify with clear gaps between them, and
392/// the longest such stretch is found in one pass. A key of a table the other
393/// runs hold as well stays where it is: a read stops at the first level that
394/// holds it, and the version left above may be the older one.
395fn largest_trivial_move(
396 curr_level: &Level,
397 next_level: &Level,
398 hidden_set: &HiddenSet,
399 cmp: &dyn crate::comparator::UserComparator,
400) -> Option<HashSet<TableId>> {
401 use core::ops::Bound;
402
403 let free_of = |runs: &[alloc::sync::Arc<crate::version::run::Run<Table>>],
404 skip: Option<usize>,
405 bounds: (Bound<&crate::Slice>, Bound<&crate::Slice>)| {
406 runs.iter().enumerate().all(|(index, run)| {
407 Some(index) == skip
408 || run
409 .range_overlap_indexes_cmp::<crate::Slice, _, _>(&bounds, cmp)
410 .is_none()
411 })
412 };
413 let clear = |bounds| free_of(next_level, None, bounds);
414 for (run_index, curr_run) in curr_level.iter().enumerate() {
415 // The longest stretch so far, as (start, length), and the current one.
416 let mut best: Option<(usize, usize)> = None;
417 let (mut stretch_start, mut stretch_len) = (0, 0);
418 let mut previous: Option<&Table> = None;
419 for (index, table) in curr_run.iter().enumerate() {
420 let range = table.key_range();
421 let bounds = (Bound::Included(range.min()), Bound::Included(range.max()));
422 let movable = !hidden_set.is_hidden(table.id())
423 && clear(bounds)
424 && free_of(curr_level, Some(run_index), bounds);
425 if !movable {
426 stretch_len = 0;
427 previous = None;
428 continue;
429 }
430 let joins = previous.is_some_and(|previous| {
431 clear((
432 Bound::Excluded(previous.key_range().max()),
433 Bound::Excluded(range.min()),
434 ))
435 });
436 if joins {
437 stretch_len += 1;
438 } else {
439 (stretch_start, stretch_len) = (index, 1);
440 }
441 previous = Some(table);
442 if best.is_none_or(|(_, len)| stretch_len > len) {
443 best = Some((stretch_start, stretch_len));
444 }
445 }
446 if let Some((start, len)) = best {
447 return curr_run
448 .get(start..start + len)
449 .map(|window| window.iter().map(Table::id).collect());
450 }
451 }
452 None
453}
454
455/// A window of one run of `level` to go into a tiered level below, which
456/// takes nothing of that level in, none of its tables hidden.
457///
458/// Every window rewrites only its own bytes, so least rewrite takes the
459/// tables holding `overshoot` bytes from the first one not hidden, in the
460/// first run that has one: a compaction under way on the front of a run
461/// leaves the rest to go. The hottest takes the window with the most reads
462/// per byte, ranked as a merge into a leveled level ranks them.
463fn window_over(
464 level: &Level,
465 hidden_set: &HiddenSet,
466 overshoot: u64,
467 movement: Movement,
468 table_base_size: u64,
469) -> Option<HashSet<TableId>> {
470 let (run, start, end) = match movement {
471 Movement::LeastRewrite { .. } => level.iter().find_map(|run| {
472 let start = run
473 .iter()
474 .position(|table| !hidden_set.is_hidden(table.id()))?;
475 let mut bytes = 0u64;
476 let mut end = start;
477 for table in run.iter().skip(start) {
478 if hidden_set.is_hidden(table.id()) {
479 break;
480 }
481 end += 1;
482 bytes += table.file_size();
483 if bytes >= overshoot {
484 break;
485 }
486 }
487 Some((run, start, end))
488 })?,
489 Movement::Hottest => {
490 let mut ranking = MergeRanking::new(overshoot, table_base_size, movement);
491 let cap = 50 * table_base_size;
492 for (run_index, run) in level.iter().enumerate() {
493 let bytes = prefix_sums(run.iter().map(Table::file_size));
494 let hidden =
495 prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id()))));
496 let heat = prefix_sums(run.iter().map(Table::heat));
497 // As a merge into a leveled level looks at them: a window past
498 // the cap ends a width only once every window of it is.
499 for width in 1..=run.len() {
500 let mut within_cap = false;
501 for start in 0..=(run.len() - width) {
502 let end = start + width;
503 #[expect(
504 clippy::indexing_slicing,
505 reason = "start < end <= run.len(), and the sums hold run.len() + 1 entries"
506 )]
507 let (size, hidden, heat) = (
508 bytes[end] - bytes[start],
509 hidden[end] - hidden[start],
510 heat[end] - heat[start],
511 );
512 if size > cap {
513 continue;
514 }
515 within_cap = true;
516 if hidden > 0 {
517 continue;
518 }
519 ranking.offer(
520 MergeCost {
521 promoted: size,
522 total: size,
523 heat,
524 },
525 (run_index, start, end),
526 );
527 }
528 if !within_cap {
529 break;
530 }
531 }
532 }
533 let ((_, _, (run_index, start, end)), _) = ranking.finish()?;
534 (level.iter().nth(run_index)?, start, end)
535 }
536 };
537 let ids: HashSet<_> = run.get(start..end)?.iter().map(Table::id).collect();
538 (!ids.is_empty()).then_some(ids)
539}
540
541/// Prefix sums: entry `i` is the sum of the first `i` items.
542fn prefix_sums(items: impl Iterator<Item = u64>) -> Vec<u64> {
543 let mut sums = Vec::with_capacity(items.size_hint().0 + 1);
544 sums.push(0);
545 let mut total = 0;
546 for item in items {
547 // Byte sizes or counts of one run's tables, bounded by the disk.
548 total += item;
549 sums.push(total);
550 }
551 sums
552}
553
554/// The tables of one run of the level a merge promotes from that each window
555/// of one run of the level below would pull in, priced in constant time.
556///
557/// A window pulls in the tables its key range contains. In a run, sorted by
558/// key with disjoint ranges, those are the tables starting at or after the
559/// window's min key and ending at or before its max key: a contiguous stretch
560/// whose ends move only forward as the window's ends do.
561struct PullIn {
562 /// For each table of the window run, the first table of this run starting
563 /// at or after its min key.
564 first_from: Vec<usize>,
565 /// For each table of the window run, the number of tables of this run
566 /// ending at or before its max key.
567 end_through: Vec<usize>,
568 /// Prefix sums of this run's table sizes, of its hidden tables and of the
569 /// reads its tables served.
570 bytes: Vec<u64>,
571 hidden: Vec<u64>,
572 reads: Vec<u64>,
573}
574
575impl PullIn {
576 fn new(
577 run: &[Table],
578 window_run: &[Table],
579 hidden_set: &HiddenSet,
580 cmp: &dyn crate::comparator::UserComparator,
581 ) -> Self {
582 use core::cmp::Ordering;
583
584 let mut first_from = Vec::with_capacity(window_run.len());
585 let mut at = 0;
586 for table in window_run {
587 let min = table.key_range().min();
588 while run
589 .get(at)
590 .is_some_and(|t| cmp.compare(t.key_range().min(), min) == Ordering::Less)
591 {
592 at += 1;
593 }
594 first_from.push(at);
595 }
596 let mut end_through = Vec::with_capacity(window_run.len());
597 at = 0;
598 for table in window_run {
599 let max = table.key_range().max();
600 while run
601 .get(at)
602 .is_some_and(|t| cmp.compare(t.key_range().max(), max) != Ordering::Greater)
603 {
604 at += 1;
605 }
606 end_through.push(at);
607 }
608 Self {
609 first_from,
610 end_through,
611 bytes: prefix_sums(run.iter().map(Table::file_size)),
612 hidden: prefix_sums(run.iter().map(|t| u64::from(hidden_set.is_hidden(t.id())))),
613 reads: prefix_sums(run.iter().map(Table::heat)),
614 }
615 }
616
617 /// The bytes, the hidden tables and the reads of what the window from
618 /// the window run's table `first` to its table `last` pulls in.
619 #[expect(
620 clippy::indexing_slicing,
621 reason = "first and last index the window run, and the stretch ends lie in 0..=run.len()"
622 )]
623 fn contained(&self, first: usize, last: usize) -> (u64, u64, u64) {
624 let (from, through) = (self.first_from[first], self.end_through[last]);
625 if from < through {
626 (
627 self.bytes[through] - self.bytes[from],
628 self.hidden[through] - self.hidden[from],
629 self.reads[through] - self.reads[from],
630 )
631 } else {
632 (0, 0, 0)
633 }
634 }
635}
636
637#[doc(hidden)]
638pub const NAME: &str = "LeveledCompaction";
639
640/// The default of [`Strategy::with_l0_file_threshold`].
641pub const DEFAULT_L0_FILE_THRESHOLD: u16 = 500;
642
643/// Leveled compaction strategy (LCS)
644///
645/// When a level reaches some threshold size, parts of it are merged into overlapping tables in the next level.
646///
647/// Each level Ln for n >= 2 can have up to `level_base_size * ratio^(n - 1)` tables.
648///
649/// LCS suffers from comparatively high write amplification, but has decent read amplification and great space amplification (~1.1x).
650///
651/// LCS is the recommended compaction strategy to use.
652///
653/// More info here: <https://fjall-rs.github.io/post/lsm-leveling/>
654#[derive(Clone)]
655pub struct Strategy {
656 /// L0 run count that triggers a merge into L1.
657 l0_threshold: u8,
658
659 /// L0 table count that triggers a merge into L1 whatever the run count.
660 l0_file_threshold: u16,
661
662 /// The target table size as disk (possibly compressed).
663 target_size: u64,
664
665 /// Size ratio between levels of the LSM tree (a.k.a fanout, growth rate)
666 level_ratio_policy: Vec<f32>,
667
668 /// When true, dynamically sizes levels based on the actual data in
669 /// the last non-empty level, reducing space amplification to ~1.1x.
670 ///
671 /// Same as `level_compaction_dynamic_level_bytes` in `RocksDB`.
672 ///
673 /// Default = false (static leveling).
674 dynamic: bool,
675
676 /// When true, enables multi-level compaction: if L0→L1 is chosen but
677 /// L1 is already oversized, compacts L0+L1→L2 directly in one pass
678 /// to avoid a write-then-rewrite cycle.
679 ///
680 /// Default = false.
681 multi_level: bool,
682
683 /// How many bytes past a level's overshoot a merge may promote and still
684 /// be preferred on its cost per promoted byte. `None` = one target table
685 /// size, the smallest step a merge can take.
686 promotion_slack: Option<u64>,
687
688 /// The choices set for a level in place of the leveled ones, by level.
689 policies: [Option<LevelPolicy>; 7],
690}
691
692impl Default for Strategy {
693 fn default() -> Self {
694 Self {
695 l0_threshold: 4,
696 l0_file_threshold: DEFAULT_L0_FILE_THRESHOLD,
697 target_size:/* 64 MiB */ 64 * 1_024 * 1_024,
698 level_ratio_policy: vec![10.0],
699 dynamic: false,
700 multi_level: false,
701 promotion_slack: None,
702 policies: [None; 7],
703 }
704 }
705}
706
707impl Strategy {
708 /// Sets the growth ratio between levels.
709 ///
710 /// Same as `set_max_bytes_for_level_multiplier` in `RocksDB`.
711 ///
712 /// Default = [10.0]
713 #[must_use]
714 pub fn with_level_ratio_policy(mut self, policy: Vec<f32>) -> Self {
715 self.level_ratio_policy = policy;
716 self
717 }
718
719 /// Sets the L0 threshold.
720 ///
721 /// When the number of runs in L0 reaches this threshold, L0 is merged
722 /// into L1. A run's tables have disjoint key ranges, so a read pays at
723 /// most one table per run: the run count bounds L0's read amplification,
724 /// where the table count does not. A flush larger than
725 /// [`Config::table_target_size`](crate::Config::table_target_size) writes
726 /// one run of several tables and counts once.
727 ///
728 /// Plays the role of `level0_file_num_compaction_trigger` in `RocksDB`,
729 /// whose flushes write one table each, so files and runs coincide there.
730 ///
731 /// Default = 4; `0` is taken as `1`, since no run count is below zero.
732 #[must_use]
733 pub fn with_l0_threshold(mut self, threshold: u8) -> Self {
734 self.l0_threshold = threshold.max(1);
735 self
736 }
737
738 /// Sets the L0 table count that triggers a merge into L1 whatever the run
739 /// count.
740 ///
741 /// A guard against fan-out: every L0 table costs a file handle, a
742 /// manifest entry and per-table metadata in memory, however few runs
743 /// hold them.
744 ///
745 /// Keep it above the run threshold times the tables one flush writes
746 /// (memtable size over
747 /// [`Config::table_target_size`](crate::Config::table_target_size)):
748 /// below that it fires before the run trigger, on every few flushes,
749 /// and L0 merges into L1 as often as a file-count trigger would make it.
750 /// The default, with the default run threshold of 4, leaves room for a
751 /// memtable of 125 tables of the target (about 8 GiB at 64 MiB).
752 ///
753 /// Default = 500; `0` is taken as `1`, since no table count is below zero.
754 ///
755 /// # Examples
756 ///
757 /// ```
758 /// use lsm_tree::compaction::Leveled;
759 ///
760 /// let strategy = Leveled::default().with_l0_file_threshold(200);
761 /// # let _ = strategy;
762 /// ```
763 #[must_use]
764 pub fn with_l0_file_threshold(mut self, threshold: u16) -> Self {
765 self.l0_file_threshold = threshold.max(1);
766 self
767 }
768
769 /// Whether L0's shape alone calls for a merge into L1: its run count or
770 /// its table count has reached its threshold.
771 fn l0_triggered(&self, l0: &Level) -> bool {
772 // L0 is never the last level of a tree that compacts.
773 shape_triggered(self.policy(0, 0, usize::MAX).trigger, l0)
774 }
775}
776
777/// Whether a level's shape alone calls for compacting it: it is triggered by
778/// its shape, and its run count or its table count has reached its count.
779/// A count of `0` is taken as `1`, since no count is below zero.
780fn shape_triggered(trigger: Trigger, level: &Level) -> bool {
781 match trigger {
782 Trigger::Shape { runs, tables } => {
783 level.run_count() >= usize::from(runs.max(1))
784 || level.table_count() >= usize::from(tables.max(1))
785 }
786 Trigger::Size | Trigger::Never => false,
787 }
788}
789
790/// A level's score under a shape trigger of `runs` and `tables`: how far past
791/// the nearer of the two counts it is.
792fn shape_score((runs, tables): (u8, u16), level: &Level) -> f64 {
793 #[expect(
794 clippy::cast_precision_loss,
795 reason = "counts of runs and tables are far below 2^52"
796 )]
797 let (run_count, table_count) = (level.run_count() as f64, level.table_count() as f64);
798 (run_count / f64::from(runs.max(1))).max(table_count / f64::from(tables.max(1)))
799}
800
801/// The fewest bytes `level`, past a shape trigger of `runs` and `tables`, can
802/// send down by windows to hold fewer runs than `runs` and fewer tables than
803/// `tables`: at least its smallest runs that many too many, and at least its
804/// smallest tables that many too many.
805fn shape_window_owed((runs, tables): (u8, u16), level: &Level) -> u64 {
806 /// The `count` smallest of `sizes`, summed.
807 fn smallest(mut sizes: Vec<u64>, count: usize) -> u64 {
808 sizes.sort_unstable();
809 sizes.iter().take(count).sum()
810 }
811 // How many below the count, the trigger firing at it: none while it
812 // holds fewer.
813 let excess = |held: usize, limit: usize| {
814 let limit = limit.max(1);
815 if held >= limit { held + 1 - limit } else { 0 }
816 };
817 let run_bytes = smallest(
818 level
819 .iter()
820 .map(|run| run.iter().map(Table::file_size).sum())
821 .collect(),
822 excess(level.run_count(), usize::from(runs)),
823 );
824 let table_bytes = smallest(
825 level
826 .iter()
827 .flat_map(|run| run.iter())
828 .map(Table::file_size)
829 .collect(),
830 excess(level.table_count(), usize::from(tables)),
831 );
832 run_bytes.max(table_bytes)
833}
834
835impl Strategy {
836 /// Sets the table target size on disk (possibly compressed).
837 ///
838 /// Same as `target_file_size_base` in `RocksDB`.
839 ///
840 /// Default = 64 MiB
841 #[must_use]
842 pub fn with_table_target_size(mut self, bytes: u64) -> Self {
843 self.target_size = bytes;
844 self
845 }
846
847 /// Enables dynamic level sizing based on actual data in the last level.
848 ///
849 /// When enabled, level target sizes are computed top-down from the actual
850 /// size of the last non-empty level, divided by the ratio at each step.
851 /// This reduces space amplification to ~1.1x while keeping write
852 /// amplification comparable to static leveling.
853 ///
854 /// Same as `level_compaction_dynamic_level_bytes` in `RocksDB`.
855 ///
856 /// Default = false
857 #[must_use]
858 pub fn with_dynamic_level_bytes(mut self, enabled: bool) -> Self {
859 self.dynamic = enabled;
860 self
861 }
862
863 /// Enables multi-level compaction optimization.
864 ///
865 /// When L0→L1 compaction is selected but L1 already exceeds its target
866 /// size, this option allows compacting L0+L1 directly into L2 in one
867 /// pass, avoiding the write-then-rewrite cycle that would otherwise
868 /// occur.
869 ///
870 /// Default = false
871 #[must_use]
872 pub fn with_multi_level(mut self, enabled: bool) -> Self {
873 self.multi_level = enabled;
874 self
875 }
876
877 /// Sets how many bytes past a level's overshoot a merge may promote and
878 /// still be preferred.
879 ///
880 /// A leveled merge is chosen by the bytes it rewrites per byte it moves
881 /// down a level, among the candidates that promote no more than the
882 /// level's overshoot plus this slack; when none is that small, the
883 /// cheapest candidate overall is taken, so the level still makes progress.
884 ///
885 /// Default = the table target size.
886 ///
887 /// # Examples
888 ///
889 /// ```
890 /// use lsm_tree::compaction::Leveled;
891 ///
892 /// let strategy = Leveled::default().with_promotion_slack(128 * 1_024 * 1_024);
893 /// # let _ = strategy;
894 /// ```
895 #[must_use]
896 pub fn with_promotion_slack(mut self, bytes: u64) -> Self {
897 self.promotion_slack = Some(bytes);
898 self
899 }
900
901 /// Sets the four choices for level `level` in place of the leveled ones:
902 /// when it compacts, how it holds its data, how much one compaction out
903 /// of it takes and which data that is. A level index past the tree's last
904 /// level is ignored.
905 ///
906 /// The leveled defaults are: L0 triggered by its shape (see
907 /// [`Self::with_l0_threshold`] and [`Self::with_l0_file_threshold`]),
908 /// tiered and compacted whole (through L1 when
909 /// [`Self::with_multi_level`] is on); every other level leveled,
910 /// triggered by its size, and compacted a window at a time chosen by the
911 /// least rewrite per promoted byte (see [`Self::with_promotion_slack`]);
912 /// the last level never triggered.
913 ///
914 /// # Examples
915 ///
916 /// ```
917 /// use lsm_tree::compaction::{Granularity, Layout, LevelPolicy, Leveled, Movement, Trigger};
918 ///
919 /// // L1 accumulates runs and goes down whole once it holds four.
920 /// let strategy = Leveled::default().with_level_policy(
921 /// 1,
922 /// LevelPolicy {
923 /// trigger: Trigger::Shape { runs: 4, tables: u16::MAX },
924 /// layout: Layout::Tiered,
925 /// granularity: Granularity::Level { through_full_next: false },
926 /// movement: Movement::LeastRewrite { slack: None },
927 /// },
928 /// );
929 /// # let _ = strategy;
930 /// ```
931 #[must_use]
932 pub fn with_level_policy(mut self, level: u8, policy: LevelPolicy) -> Self {
933 if let Some(slot) = self.policies.get_mut(usize::from(level)) {
934 *slot = Some(policy);
935 }
936 self
937 }
938
939 /// The first level below L0 holding data, the physical level that plays
940 /// L1, and the levels it lies below L1: while the tree is small its data
941 /// starts deeper, and moves up a level once every level holding it is
942 /// past its target. Tables `state` hides are left out of those sizes.
943 fn canonical_l1(&self, version: &Version, state: &CompactionState) -> (usize, usize, usize) {
944 #[expect(clippy::map_unwrap_or)]
945 let first_non_empty_level = version
946 .iter_levels()
947 .enumerate()
948 .skip(1)
949 .find(|(_, lvl)| !lvl.is_empty())
950 .map(|(idx, _)| idx)
951 .unwrap_or_else(|| version.level_count() - 1);
952
953 let mut canonical_l1_idx = first_non_empty_level;
954
955 // Number of levels we have to shift to get from the actual level idx to the canonical
956 let mut level_shift = canonical_l1_idx - 1;
957
958 if canonical_l1_idx > 1 && version.iter_levels().skip(1).any(|lvl| !lvl.is_empty()) {
959 let need_new_l1 = version
960 .iter_levels()
961 .enumerate()
962 .skip(1)
963 .filter(|(_, lvl)| !lvl.is_empty())
964 .all(|(idx, level)| {
965 let level_size = level
966 .iter()
967 .flat_map(|x| x.iter())
968 // NOTE: Take bytes that are already being compacted into account,
969 // otherwise we may be overcompensating
970 .filter(|x| !state.hidden_set().is_hidden(x.id()))
971 .map(Table::file_size)
972 .sum::<u64>();
973
974 #[expect(
975 clippy::cast_possible_truncation,
976 reason = "level index is bounded by level count (7, technically 255)"
977 )]
978 let target_size = self.level_target_size((idx - level_shift) as u8);
979
980 level_size > target_size
981 });
982
983 // Move up L1 one level if all current levels are at capacity
984 if need_new_l1 {
985 canonical_l1_idx -= 1;
986 level_shift -= 1;
987 }
988 }
989 (first_non_empty_level, canonical_l1_idx, level_shift)
990 }
991
992 /// The choices for the level at `physical`, `level_shift` levels below
993 /// its place while the tree is small, of a tree whose last level is
994 /// `last`: the ones set for its place, or the leveled ones. A level is
995 /// set by its place, L1 being the first level holding data below L0.
996 fn policy(&self, physical: usize, level_shift: usize, last: usize) -> LevelPolicy {
997 let level = if physical == 0 {
998 0
999 } else {
1000 physical.saturating_sub(level_shift).max(1)
1001 };
1002 let policy = self
1003 .policies
1004 .get(level)
1005 .copied()
1006 .flatten()
1007 .unwrap_or_else(|| self.leveled_policy(level));
1008 // The last level only receives compactions, whatever its place.
1009 if physical == last && physical > 0 {
1010 return LevelPolicy {
1011 trigger: Trigger::Never,
1012 ..policy
1013 };
1014 }
1015 policy
1016 }
1017
1018 /// The leveled choices for the level at place `level`.
1019 fn leveled_policy(&self, level: usize) -> LevelPolicy {
1020 LevelPolicy {
1021 trigger: if level == 0 {
1022 Trigger::Shape {
1023 runs: self.l0_threshold,
1024 tables: self.l0_file_threshold,
1025 }
1026 } else {
1027 Trigger::Size
1028 },
1029 layout: if level == 0 {
1030 Layout::Tiered
1031 } else {
1032 Layout::Leveled
1033 },
1034 granularity: if level == 0 {
1035 Granularity::Level {
1036 through_full_next: self.multi_level,
1037 }
1038 } else {
1039 Granularity::Window
1040 },
1041 movement: Movement::LeastRewrite {
1042 slack: self.promotion_slack,
1043 },
1044 }
1045 }
1046
1047 /// Calculates the size of L1.
1048 fn level_base_size(&self) -> u64 {
1049 self.target_size * u64::from(self.l0_threshold)
1050 }
1051
1052 /// Calculates the level target size.
1053 ///
1054 /// L1 = `level_base_size`
1055 ///
1056 /// L2 = `level_base_size * ratio`
1057 ///
1058 /// L3 = `level_base_size * ratio * ratio`
1059 ///
1060 /// ...
1061 fn level_target_size(&self, canonical_level_idx: u8) -> u64 {
1062 assert!(
1063 canonical_level_idx >= 1,
1064 "level_target_size does not apply to L0",
1065 );
1066
1067 if canonical_level_idx == 1 {
1068 // u64::from(self.target_size)
1069 self.level_base_size()
1070 } else {
1071 #[expect(
1072 clippy::cast_precision_loss,
1073 reason = "precision loss is acceptable for level size calculations"
1074 )]
1075 let mut size = self.level_base_size() as f32;
1076
1077 // NOTE: Minus 2 because |{L0, L1}|
1078 for idx in 0..=(canonical_level_idx - 2) {
1079 let ratio = self
1080 .level_ratio_policy
1081 .get(usize::from(idx))
1082 .copied()
1083 .unwrap_or_else(|| self.level_ratio_policy.last().copied().unwrap_or(10.0));
1084
1085 size *= ratio;
1086 }
1087
1088 #[expect(
1089 clippy::cast_possible_truncation,
1090 clippy::cast_sign_loss,
1091 reason = "size is always positive and will never even come close to u64::MAX"
1092 )]
1093 {
1094 size as u64
1095 }
1096 }
1097 }
1098
1099 /// Computes level target sizes for all 7 levels.
1100 ///
1101 /// In static mode, uses the standard exponential formula.
1102 /// In dynamic mode, derives targets from the actual size of the last
1103 /// non-empty level, dividing backwards by the ratio at each step.
1104 /// Falls back to static mode when the tree is small (dynamic L1 target
1105 /// would be less than `level_base_size`).
1106 fn compute_level_targets(
1107 &self,
1108 version: &Version,
1109 level_shift: usize,
1110 state: &CompactionState,
1111 ) -> [u64; 7] {
1112 let mut targets = [u64::MAX; 7];
1113
1114 targets[0] = self.l0_size_target();
1115
1116 if self.dynamic {
1117 // Find the last non-empty level (Lmax) and its actual size.
1118 // Iterate forward and keep track of the last non-empty level
1119 // since iter_levels() does not support DoubleEndedIterator.
1120 let mut lmax_idx = None;
1121
1122 for (idx, lvl) in version.iter_levels().enumerate().skip(1) {
1123 if !lvl.is_empty() {
1124 lmax_idx = Some(idx);
1125 }
1126 }
1127
1128 if let Some(lmax_idx) = lmax_idx {
1129 #[expect(
1130 clippy::expect_used,
1131 reason = "lmax_idx was found by iterating levels, so it must exist"
1132 )]
1133 let lmax_level = version.level(lmax_idx).expect("level should exist");
1134
1135 let lmax_size: u64 = lmax_level
1136 .iter()
1137 .flat_map(|run| run.iter())
1138 .filter(|table| !state.hidden_set().is_hidden(table.id()))
1139 .map(Table::file_size)
1140 .sum();
1141
1142 // Work backwards from Lmax
1143 if let Some(slot) = targets.get_mut(lmax_idx) {
1144 *slot = lmax_size;
1145 }
1146
1147 #[expect(
1148 clippy::cast_precision_loss,
1149 reason = "precision loss is acceptable for level size calculations"
1150 )]
1151 let mut current_target = lmax_size as f64;
1152
1153 // Only backfill down to the effective L1 (accounting for
1154 // level_shift), not to physical level 1, so we don't
1155 // overwrite slots below the shifted canonical L1.
1156 let dynamic_l1_idx = level_shift + 1;
1157
1158 for idx in (dynamic_l1_idx..lmax_idx).rev() {
1159 let canonical = idx - level_shift;
1160 // In the forward formula, target(k+1)/target(k) = ratio[k-1],
1161 // so backwards: target(k) = target(k+1) / ratio[k-1]
1162 // Clamp-to-zero: the ratio for the first dynamic level reuses
1163 // index 0 rather than underflowing to a phantom level.
1164 let ratio_idx = canonical.saturating_sub(1);
1165 let ratio = f64::from(
1166 self.level_ratio_policy
1167 .get(ratio_idx)
1168 .copied()
1169 .unwrap_or_else(|| {
1170 self.level_ratio_policy.last().copied().unwrap_or(10.0)
1171 }),
1172 );
1173
1174 // Guard against invalid ratios (zero, negative, NaN, infinite).
1175 // Fall back to static targets instead of leaving partial
1176 // dynamic targets with u64::MAX in lower-level slots.
1177 if !ratio.is_finite() || ratio <= 0.0 {
1178 return self.compute_static_targets(level_shift);
1179 }
1180
1181 current_target /= ratio;
1182
1183 #[expect(
1184 clippy::cast_possible_truncation,
1185 clippy::cast_sign_loss,
1186 reason = "target is always positive"
1187 )]
1188 if let Some(slot) = targets.get_mut(idx) {
1189 *slot = current_target as u64;
1190 }
1191 }
1192
1193 // Fallback: if dynamic L1 target is too small, use static.
1194 // Compare the shifted L1 slot, not physical slot 1.
1195 let static_l1 = self.level_base_size();
1196 if targets.get(dynamic_l1_idx).copied().unwrap_or(0) < static_l1 {
1197 return self.compute_static_targets(level_shift);
1198 }
1199
1200 return targets;
1201 }
1202 }
1203
1204 self.compute_static_targets(level_shift)
1205 }
1206
1207 /// Computes static (exponential) level targets.
1208 /// L0's size target: none (`u64::MAX`) for an L0 triggered by its shape,
1209 /// which no size serves; L1's base size for one set to trigger by size, as
1210 /// L0's size score in `RocksDB`'s leveled compaction measures it.
1211 fn l0_size_target(&self) -> u64 {
1212 if self.policy(0, 0, usize::MAX).trigger == Trigger::Size {
1213 self.level_base_size()
1214 } else {
1215 u64::MAX
1216 }
1217 }
1218
1219 fn compute_static_targets(&self, level_shift: usize) -> [u64; 7] {
1220 let mut targets = [u64::MAX; 7];
1221 targets[0] = self.l0_size_target();
1222
1223 for (idx, slot) in targets.iter_mut().enumerate().skip(1) {
1224 if idx <= level_shift {
1225 continue; // stays at u64::MAX
1226 }
1227 #[expect(
1228 clippy::cast_possible_truncation,
1229 reason = "level index is bounded by level count (7)"
1230 )]
1231 {
1232 *slot = self.level_target_size((idx - level_shift) as u8);
1233 }
1234 }
1235
1236 targets
1237 }
1238}
1239
1240impl CompactionStrategy for Strategy {
1241 fn get_name(&self) -> &'static str {
1242 NAME
1243 }
1244
1245 fn get_config(&self) -> Vec<crate::KvPair> {
1246 vec![
1247 (
1248 crate::UserKey::from("leveled_l0_threshold"),
1249 crate::UserValue::from(self.l0_threshold.to_le_bytes()),
1250 ),
1251 (
1252 crate::UserKey::from("leveled_l0_file_threshold"),
1253 crate::UserValue::from(self.l0_file_threshold.to_le_bytes()),
1254 ),
1255 (
1256 crate::UserKey::from("leveled_target_size"),
1257 crate::UserValue::from(self.target_size.to_le_bytes()),
1258 ),
1259 (
1260 crate::UserKey::from("leveled_level_ratio_policy"),
1261 crate::UserValue::from({
1262 use crate::io::{LittleEndian, WriteBytesExt};
1263
1264 let mut v = vec![];
1265
1266 #[expect(
1267 clippy::expect_used,
1268 clippy::cast_possible_truncation,
1269 reason = "writing into Vec should not fail; policies have length of 255 max"
1270 )]
1271 v.write_u8(self.level_ratio_policy.len() as u8)
1272 .expect("cannot fail");
1273
1274 for &f in &self.level_ratio_policy {
1275 #[expect(clippy::expect_used, reason = "writing into Vec should not fail")]
1276 v.write_f32::<LittleEndian>(f).expect("cannot fail");
1277 }
1278
1279 v
1280 }),
1281 ),
1282 (
1283 crate::UserKey::from("leveled_dynamic"),
1284 crate::UserValue::from([u8::from(self.dynamic)]),
1285 ),
1286 (
1287 crate::UserKey::from("leveled_multi_level"),
1288 crate::UserValue::from([u8::from(self.multi_level)]),
1289 ),
1290 ]
1291 }
1292
1293 fn pending_compaction_bytes(&self, version: &Version) -> u64 {
1294 // Read-only snapshot: an empty compaction state (nothing hidden) so the
1295 // targets reflect the full current footprint.
1296 let state = CompactionState::default();
1297 let last = version.level_count() - 1;
1298 // A level's place and target are those `choose` finds for it: while
1299 // the tree is small its data lies deeper, and the level playing L1
1300 // there is held to L1's target, not to that of its physical index.
1301 let (_, _, level_shift) = self.canonical_l1(version, &state);
1302 let targets = self.compute_level_targets(version, level_shift, &state);
1303 let mut debt = 0u64;
1304 for (idx, level) in version.iter_levels().enumerate() {
1305 // Each term is at most one level's on-disk size, so the running sum is
1306 // bounded by the tree's total footprint (disk capacity) and cannot
1307 // overflow u64 — plain addition.
1308 let policy = self.policy(idx, level_shift, last);
1309 match policy.trigger {
1310 // A shape-triggered level, once its runs or its tables reach
1311 // their count, the same predicate `choose` acts on, owes what
1312 // it sends down to fall back below them: all of it when it
1313 // goes down whole, and at least its fewest bytes that take it
1314 // below both counts when it goes down by windows.
1315 trigger @ Trigger::Shape { runs, tables } => {
1316 if shape_triggered(trigger, level) {
1317 debt += match policy.granularity {
1318 Granularity::Level { .. } => level.size(),
1319 Granularity::Window => shape_window_owed((runs, tables), level),
1320 };
1321 }
1322 }
1323 // A level set never to trigger owes nothing: no compaction of
1324 // this strategy drains it, and a debt nothing pays would hold
1325 // writes back for good. The last level owes the bytes past its
1326 // target, as it does in a static tree.
1327 Trigger::Never if idx != last => {}
1328 // A level triggered by size owes the bytes past its target; L0
1329 // has one only when it is set so.
1330 Trigger::Size | Trigger::Never => {
1331 // max(0, size - target): the bytes above this level's
1332 // target are the work that must be rewritten downward; a
1333 // level past the targets there are has none.
1334 let target = targets.get(idx).copied().unwrap_or(u64::MAX);
1335 debt += level.size().saturating_sub(target);
1336 }
1337 }
1338 }
1339 debt
1340 }
1341
1342 #[expect(clippy::too_many_lines)]
1343 fn choose(&self, version: &Version, config: &Config, state: &CompactionState) -> Choice {
1344 // Tie the invariant to the caller-supplied `Config::level_count`
1345 // rather than a literal so a tree explicitly configured with a
1346 // non-default level count does not panic here. The assertion still
1347 // fires when a `Version` carries a different count than the
1348 // `Config` driving this compactor invocation — that mismatch is a
1349 // real bug (Version and Config disagree).
1350 assert!(
1351 version.level_count() == usize::from(config.level_count),
1352 "leveled compaction requires version.level_count() == config.level_count (got version={}, config={})",
1353 version.level_count(),
1354 config.level_count,
1355 );
1356 let cmp = config.comparator.as_ref();
1357 let last = version.level_count() - 1;
1358
1359 // Trivial move into Lmax: free, so taken whatever L0's trigger says,
1360 // unless L0 is set never to start a compaction.
1361 'trivial_lmax: {
1362 if self.policy(0, 0, last).trigger == Trigger::Never {
1363 break 'trivial_lmax;
1364 }
1365 #[expect(
1366 clippy::expect_used,
1367 reason = "level 0 is guaranteed to exist in a valid version"
1368 )]
1369 let l0 = version.level(0).expect("first level should exist");
1370
1371 if !l0.is_empty() && l0.is_disjoint() {
1372 let lmax_index = version.level_count() - 1;
1373
1374 if (1..lmax_index).any(|idx| {
1375 #[expect(
1376 clippy::expect_used,
1377 reason = "levels within level_count are guaranteed to exist"
1378 )]
1379 let level = version.level(idx).expect("level should exist");
1380 !level.is_empty()
1381 }) {
1382 // There are intermediary levels with data, cannot trivially move to Lmax
1383 break 'trivial_lmax;
1384 }
1385
1386 #[expect(
1387 clippy::expect_used,
1388 reason = "lmax_index is derived from level_count so level is guaranteed to exist"
1389 )]
1390 let lmax = version.level(lmax_index).expect("last level should exist");
1391
1392 if !lmax
1393 .aggregate_key_range_cmp(cmp)
1394 .overlaps_with_key_range_cmp(&l0.aggregate_key_range_cmp(cmp), cmp)
1395 {
1396 return Choice::Move(CompactionInput {
1397 table_ids: l0.list_ids(),
1398 #[expect(
1399 clippy::cast_possible_truncation,
1400 reason = "level count is at most 7, fits in u8"
1401 )]
1402 dest_level: lmax_index as u8,
1403 canonical_level: 1,
1404 target_size: self.target_size,
1405 });
1406 }
1407 }
1408 }
1409
1410 let (first_non_empty_level, canonical_l1_idx, level_shift) =
1411 self.canonical_l1(version, state);
1412
1413 // Trivial move into L1: free, so taken whatever L0's trigger says,
1414 // unless L0 is set never to start a compaction.
1415 'trivial: {
1416 let first_level = version.l0();
1417 let target_level_idx = first_non_empty_level.min(canonical_l1_idx);
1418
1419 if self.policy(0, level_shift, last).trigger == Trigger::Never {
1420 break 'trivial;
1421 }
1422
1423 if first_level.run_count() == 1 {
1424 if version.level_is_busy(0, state.hidden_set())
1425 || version.level_is_busy(target_level_idx, state.hidden_set())
1426 {
1427 break 'trivial;
1428 }
1429
1430 let Some(target_level) = &version.level(target_level_idx) else {
1431 break 'trivial;
1432 };
1433
1434 if target_level.run_count() != 1 {
1435 break 'trivial;
1436 }
1437
1438 let key_range = first_level.aggregate_key_range_cmp(cmp);
1439
1440 // Get overlapping tables in next level
1441 let get_overlapping = target_level
1442 .iter()
1443 .flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
1444 .map(Table::id)
1445 .next();
1446
1447 if get_overlapping.is_none() && first_level.is_disjoint() {
1448 #[expect(
1449 clippy::cast_possible_truncation,
1450 reason = "level index is bounded by level count (7)"
1451 )]
1452 return Choice::Move(CompactionInput {
1453 table_ids: first_level.list_ids(),
1454 dest_level: target_level_idx as u8,
1455 canonical_level: 1,
1456 target_size: self.target_size,
1457 });
1458 }
1459 }
1460 }
1461
1462 // Intra-L0 compaction: merge multiple L0 runs into a single run within
1463 // L0 while L0 is a few tables. It rewrites the whole of L0, so it pays
1464 // only on a small L0: on runs of many tables it would fire after every
1465 // flush, keep the run count under the trigger and never let L0 go
1466 // down, rewriting an ever larger L0 each time.
1467 // Only an L0 triggered by its run count compacts within itself: the
1468 // merge keeps its runs few.
1469 if let Trigger::Shape { runs, .. } = self.policy(0, level_shift, last).trigger {
1470 let first_level = version.l0();
1471
1472 // A triggered L0 goes down instead: rewriting it here would clear
1473 // the debt (a file guard below the run threshold) without moving
1474 // anything to L1. `l0_triggered` reads L0's policy, the one set
1475 // for it when one is, so a table count set lower than the
1476 // defaults is the one this tests.
1477 if first_level.run_count() > 1
1478 && first_level.table_count() < usize::from(runs.max(1))
1479 && !self.l0_triggered(first_level)
1480 && !version.level_is_busy(0, state.hidden_set())
1481 {
1482 return Choice::Merge(CompactionInput {
1483 table_ids: first_level.list_ids(),
1484 dest_level: 0,
1485 canonical_level: 0,
1486 target_size: self.target_size,
1487 });
1488 }
1489 }
1490
1491 // Compute level targets (supports both static and dynamic modes)
1492 let level_targets = self.compute_level_targets(version, level_shift, state);
1493
1494 // Scoring
1495 let mut scores = [(/* score */ 0.0, /* overshoot */ 0u64); 7];
1496
1497 {
1498 // TODO(weak-tombstone-rewrite): incorporate `Table::weak_tombstone_count` and
1499 // `Table::weak_tombstone_reclaimable` when computing level scores so rewrite
1500 // decisions can prioritize tables that would free the most reclaimable values.
1501
1502 for (idx, level) in version.iter_levels().enumerate() {
1503 let trigger = self.policy(idx, level_shift, last).trigger;
1504 // A level triggered by its shape scores by it, whether or not
1505 // it is empty, as L0 always has.
1506 if let Trigger::Shape { runs, tables } = trigger {
1507 if shape_triggered(trigger, level)
1508 && let Some(slot) = scores.get_mut(idx)
1509 {
1510 *slot = (shape_score((runs, tables), level), 0);
1511 }
1512 continue;
1513 }
1514 // The last level, and any level set so, receives compactions
1515 // and never starts one. L0 reaches here only set to trigger by
1516 // size, against the target its policy gives it.
1517 if trigger == Trigger::Never || level.is_empty() {
1518 continue;
1519 }
1520
1521 let level_size = level
1522 .iter()
1523 .flat_map(|x| x.iter())
1524 // NOTE: Take bytes that are already being compacted into account,
1525 // otherwise we may be overcompensating
1526 .filter(|x| !state.hidden_set().is_hidden(x.id()))
1527 .map(Table::file_size)
1528 .sum::<u64>();
1529
1530 // NOTE: We check for level length above
1531 #[expect(clippy::indexing_slicing)]
1532 let target_size = level_targets[idx];
1533
1534 #[expect(
1535 clippy::indexing_slicing,
1536 reason = "idx is from iter_levels().enumerate() so always < 7 = scores.len()"
1537 )]
1538 if level_size > target_size {
1539 #[expect(
1540 clippy::cast_precision_loss,
1541 reason = "precision loss is acceptable for scoring calculations"
1542 )]
1543 let score = level_size as f64 / target_size as f64;
1544 scores[idx] = (score, level_size - target_size);
1545
1546 // NOTE: Force a trivial move
1547 if version
1548 .level(idx + 1)
1549 .is_some_and(|next_level| next_level.is_empty())
1550 {
1551 scores[idx] = (99.99, 999);
1552 }
1553 }
1554 }
1555 }
1556
1557 // Choose compaction
1558 #[expect(clippy::expect_used, reason = "highest score is expected to exist")]
1559 let (level_idx_with_highest_score, (score, overshoot_bytes)) = scores
1560 .into_iter()
1561 .enumerate()
1562 .max_by(|(_, (score_a, _)), (_, (score_b, _))| {
1563 score_a
1564 .partial_cmp(score_b)
1565 .unwrap_or(core::cmp::Ordering::Equal)
1566 })
1567 .expect("should have highest score somewhere");
1568
1569 if score < 1.0 {
1570 return Choice::DoNothing;
1571 }
1572
1573 // We choose L0->L1 compaction, the whole of L0
1574 let l0_policy = self.policy(0, level_shift, last);
1575 if let (0, Granularity::Level { through_full_next }) =
1576 (level_idx_with_highest_score, l0_policy.granularity)
1577 {
1578 let Some(first_level) = version.level(0) else {
1579 return Choice::DoNothing;
1580 };
1581
1582 if version.level_is_busy(0, state.hidden_set())
1583 || version.level_is_busy(canonical_l1_idx, state.hidden_set())
1584 {
1585 return Choice::DoNothing;
1586 }
1587
1588 let Some(target_level) = &version.level(canonical_l1_idx) else {
1589 return Choice::DoNothing;
1590 };
1591
1592 let mut table_ids = first_level.list_ids();
1593
1594 let key_range = first_level.aggregate_key_range_cmp(cmp);
1595
1596 // Get overlapping tables in next level
1597 let target_level_overlapping_table_ids: Vec<_> = target_level
1598 .iter()
1599 .flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
1600 .map(Table::id)
1601 .collect();
1602
1603 // A leveled L1 takes the merge into its run; a tiered one takes
1604 // its output as a run of its own.
1605 let l1_tiered =
1606 self.policy(canonical_l1_idx, level_shift, last).layout == Layout::Tiered;
1607 if !l1_tiered {
1608 table_ids.extend(&target_level_overlapping_table_ids);
1609 }
1610
1611 // Multi-level compaction: if L1 is already oversized, skip it
1612 // and compact L0+L1 directly into L2 in one pass.
1613 // NOTE: Currently triggers on pre-compaction L1 score. A future
1614 // improvement could use projected post-compaction bytes to also
1615 // catch cases where L1 is close to its target and this batch
1616 // would push it over.
1617 if through_full_next && !l1_tiered {
1618 let l1_score = scores.get(canonical_l1_idx).map_or(0.0, |(s, _)| *s);
1619 let l2_idx = canonical_l1_idx + 1;
1620
1621 if l1_score > 1.0
1622 && l2_idx < version.level_count()
1623 && !version.level_is_busy(l2_idx, state.hidden_set())
1624 && let Some(l2) = version.level(l2_idx)
1625 {
1626 // Include ALL L1 tables (we're emptying L1 into L2)
1627 table_ids.extend(target_level.list_ids());
1628
1629 // Include overlapping L2 tables — query per merged
1630 // interval instead of one coarse aggregate (#72).
1631 // An aggregate across disjoint tables (e.g. [a,d] and
1632 // [x,z] → [a,z]) covers gaps and pulls in L2 tables
1633 // that don't actually overlap any input table.
1634 //
1635 // Merge input key ranges into disjoint intervals first
1636 // to reduce redundant queries when L0 tables overlap
1637 // (#122 Part 2). Sort by comparator-min, then coalesce.
1638 {
1639 let mut input_ranges: Vec<_> = target_level
1640 .iter()
1641 .chain(first_level.iter())
1642 .flat_map(|run| run.iter())
1643 .map(|t| t.key_range().clone())
1644 .collect();
1645 input_ranges.sort_by(|a, b| cmp.compare(a.min(), b.min()));
1646
1647 let merged = crate::KeyRange::merge_sorted_cmp(input_ranges, cmp);
1648
1649 for run in l2.iter() {
1650 for interval in &merged {
1651 for l2t in run.get_overlapping_cmp(interval, cmp) {
1652 table_ids.insert(Table::id(l2t));
1653 }
1654 }
1655 }
1656 }
1657
1658 #[expect(
1659 clippy::cast_possible_truncation,
1660 reason = "level index is bounded by level count (7)"
1661 )]
1662 return Choice::Merge(CompactionInput {
1663 table_ids,
1664 dest_level: l2_idx as u8,
1665 canonical_level: 2,
1666 target_size: self.target_size,
1667 });
1668 }
1669 }
1670
1671 #[expect(
1672 clippy::cast_possible_truncation,
1673 reason = "level index is bounded by level count (7, technically 255)"
1674 )]
1675 let choice = CompactionInput {
1676 table_ids,
1677 dest_level: canonical_l1_idx as u8,
1678 canonical_level: 1,
1679 target_size: self.target_size,
1680 };
1681
1682 if target_level_overlapping_table_ids.is_empty() && first_level.is_disjoint() {
1683 return Choice::Move(choice);
1684 }
1685 return Choice::Merge(choice);
1686 }
1687
1688 // We choose L1+ compaction instead, or a window of L0
1689
1690 // L0 goes into the canonical L1, a deeper level into the one below.
1691 let next_level_idx = if level_idx_with_highest_score == 0 {
1692 canonical_l1_idx
1693 } else {
1694 level_idx_with_highest_score + 1
1695 };
1696 // NOTE: Level count is 255 max
1697 #[expect(clippy::cast_possible_truncation)]
1698 let (next_level_index, canonical_level) =
1699 (next_level_idx as u8, (next_level_idx - level_shift) as u8);
1700
1701 let Some(level) = version.level(level_idx_with_highest_score) else {
1702 return Choice::DoNothing;
1703 };
1704
1705 let Some(next_level) = version.level(next_level_idx) else {
1706 return Choice::DoNothing;
1707 };
1708
1709 let policy = self.policy(level_idx_with_highest_score, level_shift, last);
1710 let next_tiered = self.policy(next_level_idx, level_shift, last).layout == Layout::Tiered;
1711 // A window into a tiered level below goes down as a run of its own,
1712 // nothing of that level taken in: the window of this one that moves
1713 // as it is, when one does.
1714 let tiered_window = if policy.granularity == Granularity::Window && next_tiered {
1715 match window_over(
1716 level,
1717 state.hidden_set(),
1718 overshoot_bytes,
1719 policy.movement,
1720 self.target_size,
1721 ) {
1722 Some(ids) => {
1723 let Some(ids) = with_overlapping_runs(level, ids, state.hidden_set(), cmp)
1724 else {
1725 return Choice::DoNothing;
1726 };
1727 Some(ids)
1728 }
1729 None => None,
1730 }
1731 } else {
1732 None
1733 };
1734 let (table_ids, can_trivial_move) = match (tiered_window, policy.granularity) {
1735 // Moved only where nothing below overlaps it, which a move may not
1736 // pass under; merged into a run of its own otherwise.
1737 (Some(ids), _) => {
1738 if version.level_is_busy(next_level_idx, state.hidden_set()) {
1739 return Choice::DoNothing;
1740 }
1741 let overlaps = level
1742 .iter()
1743 .flat_map(|run| run.iter())
1744 .filter(|table| ids.contains(&table.id()))
1745 .any(|table| {
1746 next_level
1747 .iter()
1748 .any(|run| !run.get_overlapping_cmp(table.key_range(), cmp).is_empty())
1749 });
1750 (ids, !overlaps)
1751 }
1752 // A window of a leveled level below: pick_minimal_compaction scans
1753 // all runs in both levels, handling transient multi-run states from
1754 // multi-level compaction (#108, #122).
1755 (None, Granularity::Window) if !next_tiered => {
1756 let Some(picked) = pick_minimal_compaction(
1757 level,
1758 next_level,
1759 state.hidden_set(),
1760 overshoot_bytes,
1761 self.target_size,
1762 policy.movement,
1763 cmp,
1764 ) else {
1765 return Choice::DoNothing;
1766 };
1767 picked
1768 }
1769 // The whole level, with what it overlaps of a leveled level
1770 // below; into a tiered one it goes as a run of its own, which
1771 // joins a run there it overlaps nothing of: the shape trigger
1772 // counts runs as a read meets them. A window
1773 // into a tiered level comes here only when no part of this one
1774 // moves as it is.
1775 (None, Granularity::Window | Granularity::Level { .. }) => {
1776 if version.level_is_busy(level_idx_with_highest_score, state.hidden_set())
1777 || version.level_is_busy(next_level_idx, state.hidden_set())
1778 {
1779 return Choice::DoNothing;
1780 }
1781 // The level below already past its target goes on down with
1782 // this one in the same merge, as L0 does into a full L1,
1783 // rather than being rewritten here and again when it is
1784 // chosen.
1785 if policy.granularity
1786 == (Granularity::Level {
1787 through_full_next: true,
1788 })
1789 && !next_tiered
1790 && scores
1791 .get(next_level_idx)
1792 .is_some_and(|(score, _)| *score > 1.0)
1793 && let Some(below) = version.level(next_level_idx + 1)
1794 && !version.level_is_busy(next_level_idx + 1, state.hidden_set())
1795 // An empty level below takes the next one as a trivial
1796 // move instead, which rewrites nothing.
1797 && !below.is_empty()
1798 {
1799 let mut ids = level.list_ids();
1800 ids.extend(next_level.list_ids());
1801 let mut ranges: Vec<_> = level
1802 .iter()
1803 .chain(next_level.iter())
1804 .flat_map(|run| run.iter())
1805 .map(|table| table.key_range().clone())
1806 .collect();
1807 ranges.sort_by(|a, b| cmp.compare(a.min(), b.min()));
1808 for interval in &crate::KeyRange::merge_sorted_cmp(ranges, cmp) {
1809 for run in below.iter() {
1810 ids.extend(
1811 run.get_overlapping_cmp(interval, cmp).iter().map(Table::id),
1812 );
1813 }
1814 }
1815 #[expect(
1816 clippy::cast_possible_truncation,
1817 reason = "level index is bounded by level count (7, technically 255)"
1818 )]
1819 return Choice::Merge(CompactionInput {
1820 table_ids: ids,
1821 dest_level: (next_level_idx + 1) as u8,
1822 canonical_level: canonical_level + 1,
1823 target_size: self.target_size,
1824 });
1825 }
1826 let key_range = level.aggregate_key_range_cmp(cmp);
1827 let overlapping: Vec<TableId> = next_level
1828 .iter()
1829 .flat_map(|run| run.get_overlapping_cmp(&key_range, cmp))
1830 .map(Table::id)
1831 .collect();
1832 let mut ids = level.list_ids();
1833 if !next_tiered {
1834 ids.extend(&overlapping);
1835 }
1836 (ids, overlapping.is_empty())
1837 }
1838 };
1839
1840 let choice = CompactionInput {
1841 table_ids,
1842 dest_level: next_level_index,
1843 canonical_level,
1844 target_size: self.target_size,
1845 };
1846
1847 if can_trivial_move && level.is_disjoint() {
1848 return Choice::Move(choice);
1849 }
1850 Choice::Merge(choice)
1851 }
1852}