Skip to main content

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}