coordinode-lsm-tree 5.12.0

use super::*;
use crate::{AbstractTree, Config, MAX_SEQNO, SequenceNumberCounter};
use std::sync::Arc;
use test_log::test;

#[test]
fn leveled_empty_levels() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let strategy = Arc::new(Strategy::default());
    tree.compact(strategy, 0)?;

    assert_eq!(0, tree.table_count());
    Ok(())
}

#[test]
fn leveled_l0_below_limit() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    for i in 0..3u8 {
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.flush_active_memtable(0)?;
    }

    let before = tree.table_count();
    assert_eq!(3, before);

    let strategy = Arc::new(Strategy::default());
    tree.compact(strategy, 0)?;

    assert_eq!(before, tree.table_count());

    Ok(())
}

#[test]
fn leveled_intra_l0_compaction() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Flush 3 overlapping memtables with distinct values (below configured l0_threshold=4)
    for i in 0..3u8 {
        tree.insert("a", [b'v', i].as_slice(), u64::from(i));
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.insert("z", [b'v', i].as_slice(), u64::from(i));
        tree.flush_active_memtable(0)?;
    }

    assert_eq!(3, tree.table_count());
    assert!(
        tree.l0_run_count() > 1,
        "L0 should have multiple overlapping runs"
    );

    let strategy = Arc::new(
        Strategy::default()
            .with_l0_threshold(4)
            .with_table_target_size(128 * 1024 * 1024),
    );
    tree.compact(strategy, 0)?;

    // Intra-L0 compaction should consolidate runs within L0
    assert_eq!(
        1,
        tree.l0_run_count(),
        "L0 should have exactly 1 run after intra-L0 compaction"
    );
    assert_eq!(
        1,
        tree.table_count(),
        "Tables should be merged into 1 after intra-L0 compaction"
    );

    // All data must still be readable with correct values
    for i in 0..3u8 {
        assert!(tree.get([b'k', i].as_slice(), MAX_SEQNO)?.is_some());
    }
    // Latest visible versions should be the last written values
    assert_eq!(
        tree.get("a", MAX_SEQNO)?.as_deref(),
        Some([b'v', 2].as_slice()),
    );
    assert_eq!(
        tree.get("z", MAX_SEQNO)?.as_deref(),
        Some([b'v', 2].as_slice()),
    );

    // Verify data stayed in L0 (not pushed to L1)
    assert!(
        tree.current_version().level(1).is_none_or(|l| l.is_empty()),
        "L1 should remain empty after intra-L0 compaction"
    );

    Ok(())
}

#[test]
fn leveled_intra_l0_preserves_newer_run_ordering() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Flush 2 overlapping memtables (below l0_threshold=4)
    tree.insert("key", "old_1", 0);
    tree.flush_active_memtable(0)?;
    tree.insert("key", "old_2", 1);
    tree.flush_active_memtable(0)?;

    assert_eq!(2, tree.l0_run_count());

    // Intra-L0 compaction merges the 2 runs
    let strategy = Arc::new(
        Strategy::default()
            .with_l0_threshold(4)
            .with_table_target_size(128 * 1024 * 1024),
    );
    tree.compact(strategy, 0)?;
    assert_eq!(1, tree.l0_run_count());

    // Flush a newer memtable AFTER compaction — this run must be searched first
    tree.insert("key", "newest", 2);
    tree.flush_active_memtable(0)?;

    assert_eq!(2, tree.l0_run_count());

    // The newest flush must win: merged (older) run is appended, newer run is at front
    assert_eq!(
        tree.get("key", MAX_SEQNO)?.as_deref(),
        Some(b"newest".as_slice()),
        "newer L0 run must be found before merged (older) run"
    );

    Ok(())
}

#[test]
fn leveled_l0_reached_limit() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    for i in 0..4u8 {
        // NOTE: Tables need to overlap
        tree.insert("a", "v", 0);
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.insert("z", "v", 0);
        tree.flush_active_memtable(0)?;
    }

    assert_eq!(4, tree.table_count());

    let strategy = Arc::new(Strategy::default());
    tree.compact(strategy, 0)?;

    assert_eq!(1, tree.table_count());

    Ok(())
}

#[test]
fn compaction_debt_flags_l0_over_threshold_then_clears() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;
    let strategy = Arc::new(Strategy::default());

    // Empty tree: no level is above target, so nothing is pending.
    assert_eq!(tree.compaction_debt(&*strategy), 0);

    // Four overlapping L0 runs reach the default L0 threshold (4): the whole of
    // L0 is pending a merge into L1, so the debt estimate is non-zero.
    for i in 0..4u8 {
        tree.insert("a", "v", 0);
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.insert("z", "v", 0);
        tree.flush_active_memtable(0)?;
    }
    assert!(
        tree.compaction_debt(&*strategy) > 0,
        "L0 at the file threshold should report pending compaction debt",
    );

    // After compaction the runs collapse into a single L1 table below target,
    // clearing the debt.
    tree.compact(strategy, 0)?;
    assert_eq!(tree.compaction_debt(&*Arc::new(Strategy::default())), 0);

    Ok(())
}

#[test]
fn leveled_l0_reached_limit_disjoint() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    for i in 0..4u8 {
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.flush_active_memtable(0)?;
    }

    assert_eq!(4, tree.table_count());

    let strategy = Arc::new(Strategy::default());
    tree.compact(strategy, 0)?;

    assert_eq!(4, tree.table_count());

    Ok(())
}

#[test]
fn leveled_l0_reached_limit_disjoint_l1() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    for i in 0..4 {
        // NOTE: Tables need to overlap
        tree.insert("a", "v", i);
        tree.insert("b", "v", i);
        tree.flush_active_memtable(0)?;
    }

    let fifo = Arc::new(Strategy::default());
    tree.compact(fifo, 0)?;

    assert_eq!(1, tree.table_count());

    for i in 0..4u8 {
        tree.insert([b'k', i].as_slice(), "v", 0);
        tree.flush_active_memtable(0)?;
    }

    assert_eq!(5, tree.table_count());

    let strategy = Arc::new(Strategy::default());
    tree.compact(strategy, 0)?;

    assert_eq!(5, tree.table_count());

    Ok(())
}

#[test]
#[expect(clippy::unwrap_used)]
fn leveled_sequential_inserts() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let strategy = Arc::new(Strategy {
        target_size: 1,
        ..Default::default()
    });

    let mut table_count = 0;

    for k in 0u64..30 {
        table_count += 1;

        tree.insert(k.to_be_bytes(), "", 0);
        tree.flush_active_memtable(0)?;

        assert_eq!(table_count, tree.table_count());
        tree.compact(strategy.clone(), 0)?;
        assert_eq!(table_count, tree.table_count());

        for idx in 1..=5 {
            assert_eq!(
                0,
                tree.current_version().level(idx).unwrap().len(),
                "no tables should be in intermediary level (L{idx})",
            );
        }
    }

    Ok(())
}

// --- Dynamic Leveling Tests ---

#[test]
fn dynamic_leveling_empty() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Dynamic leveling on an empty tree should behave like static (DoNothing)
    let strategy = Arc::new(Strategy::default().with_dynamic_level_bytes(true));
    tree.compact(strategy, 0)?;

    assert_eq!(0, tree.table_count());
    Ok(())
}

#[test]
fn dynamic_leveling_basic_compaction() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Insert enough overlapping data to trigger L0→L1 compaction
    for i in 0..4u8 {
        tree.insert("a", "v", u64::from(i));
        tree.insert([b'k', i].as_slice(), "v", u64::from(i));
        tree.insert("z", "v", u64::from(i));
        tree.flush_active_memtable(u64::from(i))?;
    }

    assert_eq!(4, tree.table_count());

    let strategy = Arc::new(Strategy::default().with_dynamic_level_bytes(true));
    tree.compact(strategy, 4)?;

    // Compaction should still work — data should be merged
    assert!(tree.table_count() < 4, "tables should be compacted");

    // All data should be readable
    for i in 0..4u8 {
        assert!(tree.get([b'k', i].as_slice(), MAX_SEQNO)?.is_some());
    }

    Ok(())
}

#[test]
fn dynamic_leveling_data_integrity() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let strategy = Arc::new(
        Strategy::default()
            .with_dynamic_level_bytes(true)
            .with_table_target_size(1), // tiny tables to force many small output tables under dynamic leveling
    );

    // Insert and compact multiple rounds
    for round in 0..5u64 {
        for k in 0..4u8 {
            tree.insert(
                "a",
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.insert(
                [b'k', k].as_slice(),
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.insert(
                "z",
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.flush_active_memtable(round * 4 + u64::from(k))?;
        }
        tree.compact(strategy.clone(), (round + 1) * 4)?;
    }

    // All data should be readable with latest values
    for k in 0..4u8 {
        let val = tree.get([b'k', k].as_slice(), MAX_SEQNO)?;
        assert!(val.is_some(), "key k{k} should exist");
        assert_eq!(val.as_deref(), Some(b"r4".as_slice()));
    }

    Ok(())
}

// --- Multi-Level Compaction Tests ---

#[test]
fn multi_level_no_skip_when_l1_has_room() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Insert enough to trigger L0→L1 but L1 should be under its target
    for i in 0..4u8 {
        tree.insert("a", "v", u64::from(i));
        tree.insert([b'k', i].as_slice(), "v", u64::from(i));
        tree.insert("z", "v", u64::from(i));
        tree.flush_active_memtable(u64::from(i))?;
    }

    let strategy = Arc::new(Strategy::default().with_multi_level(true));
    tree.compact(strategy, 4)?;

    // L1 has room — normal L0→L1 compaction, not multi-level skip
    // Data should still be compacted and readable
    assert!(tree.table_count() < 4);

    // Verify data went to L1 (not skipped to L2) — L2 should be empty
    assert!(
        tree.current_version().level(2).is_none_or(|l| l.is_empty()),
        "L2 should remain empty when L1 has room (no multi-level skip)",
    );

    for i in 0..4u8 {
        assert!(tree.get([b'k', i].as_slice(), MAX_SEQNO)?.is_some());
    }

    Ok(())
}

#[test]
fn multi_level_data_integrity() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let strategy = Arc::new(
        Strategy::default()
            .with_multi_level(true)
            .with_table_target_size(1), // tiny tables to stress test
    );

    // Run multiple rounds of inserts + compaction
    for round in 0..8u64 {
        for k in 0..4u8 {
            tree.insert(
                "a",
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.insert(
                [b'k', k].as_slice(),
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.insert(
                "z",
                format!("r{round}").as_bytes(),
                round * 4 + u64::from(k),
            );
            tree.flush_active_memtable(round * 4 + u64::from(k))?;
        }
        tree.compact(strategy.clone(), (round + 1) * 4)?;
    }

    // All keys readable with latest values
    for k in 0..4u8 {
        let val = tree.get([b'k', k].as_slice(), MAX_SEQNO)?;
        assert!(val.is_some(), "key k{k} should exist");
        assert_eq!(val.as_deref(), Some(b"r7".as_slice()));
    }

    Ok(())
}

// --- Coverage: get_config, get_name, builder methods ---

#[test]
fn leveled_get_name() {
    use crate::compaction::CompactionStrategy;
    let strategy = Strategy::default();
    assert_eq!(strategy.get_name(), "LeveledCompaction");
}

#[test]
fn leveled_get_config_includes_new_fields() {
    use crate::compaction::CompactionStrategy;
    let strategy = Strategy::default()
        .with_dynamic_level_bytes(true)
        .with_multi_level(true);

    let config = strategy.get_config();

    let keys: Vec<_> = config.iter().map(|(k, _)| k.as_ref()).collect();
    assert!(
        keys.iter().any(|k| k == b"leveled_dynamic"),
        "should have leveled_dynamic key",
    );
    assert!(
        keys.iter().any(|k| k == b"leveled_multi_level"),
        "should have leveled_multi_level key",
    );
    assert!(
        keys.iter().any(|k| k == b"leveled_l0_threshold"),
        "should have leveled_l0_threshold key",
    );
    assert!(
        keys.iter().any(|k| k == b"leveled_target_size"),
        "should have leveled_target_size key",
    );
    assert!(
        keys.iter().any(|k| k == b"leveled_level_ratio_policy"),
        "should have leveled_level_ratio_policy key",
    );
}

#[test]
fn leveled_builder_chaining() {
    use crate::compaction::CompactionStrategy;
    // Exercise all builder methods to cover their code paths
    let strategy = Strategy::default()
        .with_l0_threshold(8)
        .with_table_target_size(32 * 1024 * 1024)
        .with_level_ratio_policy(vec![8.0, 10.0])
        .with_dynamic_level_bytes(true)
        .with_multi_level(true);

    let config = strategy.get_config();
    assert!(!config.is_empty());
    assert_eq!(strategy.get_name(), "LeveledCompaction");
}

// --- Coverage: dynamic leveling with enough data to exercise backward computation ---

#[test]
fn dynamic_leveling_multiple_levels() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let strategy = Arc::new(
        Strategy::default()
            .with_dynamic_level_bytes(true)
            .with_l0_threshold(4)
            .with_table_target_size(1), // tiny tables → many levels get populated
    );

    let mut seqno = 0u64;

    // Multiple rounds of flush + compact to push data through levels
    for _round in 0..12 {
        for _k in 0..4 {
            tree.insert("a", "val", seqno);
            tree.insert(format!("key_{seqno}").as_bytes(), "val", seqno);
            tree.insert("z", "val", seqno);
            tree.flush_active_memtable(seqno)?;
            seqno += 1;
        }
        // Run compaction multiple times per round to propagate through levels
        for _ in 0..3 {
            tree.compact(strategy.clone(), seqno)?;
        }
    }

    // All keys should be readable
    for s in 0..seqno {
        assert!(
            tree.get(format!("key_{s}").as_bytes(), MAX_SEQNO)?
                .is_some(),
            "key_{s} should exist",
        );
    }

    Ok(())
}

// --- Coverage: multi-level compaction with oversized L1 ---

#[test]
fn multi_level_skip_fires_when_l1_oversized() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Step 1: Fill L1 with overlapping data using normal leveled compaction.
    // Use default target_size (64MiB) so data stays in L1 instead of
    // trivial-moving to Lmax (which happens with target_size=1).
    let leveled = Arc::new(Strategy::default().with_l0_threshold(4));

    let mut seqno = 0u64;

    for _round in 0..3 {
        for _k in 0..4 {
            tree.insert("a", "val", seqno);
            tree.insert(format!("k_{seqno}").as_bytes(), "val", seqno);
            tree.insert("z", "val", seqno);
            tree.flush_active_memtable(seqno)?;
            seqno += 1;
        }
        tree.compact(leveled.clone(), seqno)?;
    }

    // Step 2: Switch to tiny target_size so L1 target (target_size * l0_threshold
    // = 64 * 4 = 256 bytes) is small enough that any real table in L1 causes
    // l1_score > 1.0, but L0 with 8+ tables still wins overall scoring
    // (L0 score = tables/threshold = 8/4 = 2.0 > l1_score ≈ 1.x).
    // This makes level_idx_with_highest_score == 0 AND l1_score > 1.0,
    // which is the condition for multi-level skip to fire.
    let multi = Arc::new(
        Strategy::default()
            .with_multi_level(true)
            .with_table_target_size(64)
            .with_l0_threshold(4),
    );

    // Flush 8 overlapping tables so L0 scores 2.0 (8/4)
    for _k in 0..8 {
        tree.insert("a", "val", seqno);
        tree.insert(format!("k_{seqno}").as_bytes(), "val", seqno);
        tree.insert("z", "val", seqno);
        tree.flush_active_memtable(seqno)?;
        seqno += 1;
    }

    // Single compact should trigger multi-level: L0 wins scoring,
    // L1 is oversized (real table > 256 byte target) → L0+L1→L2 skip
    let result = tree.compact(multi, seqno)?;

    // compact() now returns CompactionResult so we can assert the merge path fired
    assert_eq!(
        result.action,
        crate::compaction::CompactionAction::Merged,
        "multi-level skip should produce a Merged action",
    );
    assert!(
        result.dest_level.is_some_and(|lvl| lvl >= 2),
        "multi-level skip should target L2+, got {:?}",
        result.dest_level,
    );

    // Data MUST have propagated beyond L1 into deeper levels.
    let version = tree.current_version();
    let has_deep_data =
        (2..version.level_count()).any(|idx| version.level(idx).is_some_and(|l| !l.is_empty()));
    assert!(
        has_deep_data,
        "data should exist in L2+ after compaction with multi_level",
    );

    // All data should be readable
    for s in 0..seqno {
        assert!(
            tree.get(format!("k_{s}").as_bytes(), MAX_SEQNO)?.is_some(),
            "key k_{s} should exist",
        );
    }

    Ok(())
}

// --- Coverage: multi-level L2 overlap uses per-range, not aggregate (#72) ---

#[test]
fn multi_level_sparse_keyspace_data_integrity() -> crate::Result<()> {
    // Regression test for #72: when L0/L1 tables have narrow, disjoint key
    // ranges, the per-table L2 overlap query avoids pulling in gap-filling
    // L2 tables that the old aggregate-range approach would include.
    //
    // Each flush writes keys from ONE narrow range so the flushed table's
    // key_range is tight (e.g. [a,d] or [x,z]), not the full keyspace.
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    let mut seqno = 0u64;

    // Phase 1: Fill L1 with overlapping data using default target (64 MiB).
    // L1 target = 256 MiB >> our data, so it stays in L1.
    let leveled = Arc::new(Strategy::default().with_l0_threshold(4));
    for _round in 0..3 {
        for _k in 0..4 {
            tree.insert("a", "val", seqno);
            tree.insert(format!("k_{seqno}").as_bytes(), "val", seqno);
            tree.insert("z", "val", seqno);
            tree.flush_active_memtable(seqno)?;
            seqno += 1;
        }
        tree.compact(leveled.clone(), seqno)?;
    }

    // Phase 2: Switch to tiny target so L1 becomes "oversized" relative
    // to the new target (64 * 4 = 256 bytes). Flush disjoint L0 tables
    // with narrow key ranges to exercise the per-range overlap selection.
    let multi = Arc::new(
        Strategy::default()
            .with_multi_level(true)
            .with_table_target_size(64)
            .with_l0_threshold(4),
    );

    for _k in 0..8 {
        tree.insert("a", "val", seqno);
        tree.insert(format!("k_{seqno}").as_bytes(), "val", seqno);
        tree.insert("z", "val", seqno);
        tree.flush_active_memtable(seqno)?;
        seqno += 1;
    }

    let result = tree.compact(multi, seqno)?;

    // Verify the compaction produced a merge targeting L2+
    assert_eq!(
        result.action,
        crate::compaction::CompactionAction::Merged,
        "compaction should produce a Merged action",
    );
    assert!(
        result.dest_level.is_some_and(|lvl| lvl >= 2),
        "compaction should target L2+, got {:?}",
        result.dest_level,
    );

    // Verify data propagated beyond L1 into deeper levels.
    let version = tree.current_version();
    let has_deep_data =
        (2..version.level_count()).any(|idx| version.level(idx).is_some_and(|l| !l.is_empty()));
    assert!(
        has_deep_data,
        "data should exist in L2+ after compaction with multi_level",
    );

    // All data should be readable
    for s in 0..seqno {
        assert!(
            tree.get(format!("k_{s}").as_bytes(), MAX_SEQNO)?.is_some(),
            "key k_{s} should exist",
        );
    }

    Ok(())
}

// --- Coverage: dynamic fallback to static when tree is small ---

#[test]
fn dynamic_leveling_fallback_to_static() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // With large target_size, dynamic L1 target will be smaller than
    // level_base_size, triggering the static fallback path.
    let strategy = Arc::new(
        Strategy::default()
            .with_dynamic_level_bytes(true)
            .with_table_target_size(64 * 1024 * 1024) // 64 MiB — large
            .with_l0_threshold(4),
    );

    // Flush a small amount of data — dynamic targets will be tiny
    // compared to level_base_size (64M * 4 = 256M), so the dynamic path
    // falls back to static targets. Exercising this code path ensures
    // compute_level_targets covers the fallback branch.
    for i in 0..4u8 {
        tree.insert("a", "v", u64::from(i));
        tree.insert([b'k', i].as_slice(), "v", u64::from(i));
        tree.insert("z", "v", u64::from(i));
        tree.flush_active_memtable(u64::from(i))?;
    }

    let result = tree.compact(strategy, 4)?;
    assert_ne!(
        result.action,
        crate::compaction::CompactionAction::Nothing,
        "compaction should have done work",
    );

    // Data should be compacted and readable
    for i in 0..4u8 {
        assert!(tree.get([b'k', i].as_slice(), MAX_SEQNO)?.is_some());
    }

    Ok(())
}

#[test]
fn multi_level_with_both_flags() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let tree = Config::new(
        dir.path(),
        SequenceNumberCounter::default(),
        SequenceNumberCounter::default(),
    )
    .open()?;

    // Test that both dynamic + multi_level work together
    let strategy = Arc::new(
        Strategy::default()
            .with_dynamic_level_bytes(true)
            .with_multi_level(true)
            .with_table_target_size(1),
    );

    let mut seqno = 0u64;

    for _round in 0..10 {
        for _k in 0..4 {
            tree.insert("a", "val", seqno);
            tree.insert(format!("key_{seqno}").as_bytes(), "val", seqno);
            tree.insert("z", "val", seqno);
            tree.flush_active_memtable(seqno)?;
            seqno += 1;
        }
        for _ in 0..3 {
            tree.compact(strategy.clone(), seqno)?;
        }
    }

    // All data readable
    for s in 0..seqno {
        assert!(
            tree.get(format!("key_{s}").as_bytes(), MAX_SEQNO)?
                .is_some(),
            "key_{s} should exist",
        );
    }

    Ok(())
}

const MIB: u64 = 1_024 * 1_024;

/// The index the streaming ranking chooses among `candidates`, offered in order.
fn rank_by_promoted_ratio(candidates: &[MergeCost], overshoot: u64, slack: u64) -> Option<usize> {
    let mut ranking = MergeRanking::new(overshoot, slack);
    for (index, cost) in candidates.iter().enumerate() {
        ranking.offer(*cost, index);
    }
    ranking.finish().map(|((_, _, index), _)| index)
}

/// The runner-up the ranking reports is the cheapest candidate it did not
/// choose: the overall cheapest when the bound excluded it, the second
/// cheapest otherwise.
#[test]
fn the_ranking_reports_the_cheapest_candidate_it_did_not_choose() {
    let within = cost(20 * MIB, 60 * MIB);
    let cheaper_past_bound = cost(1_000 * MIB, 100 * MIB);
    let dearer = cost(10 * MIB, 90 * MIB);
    let mut ranking = MergeRanking::new(20 * MIB, 64 * MIB);
    for (index, c) in [dearer, cheaper_past_bound, within].into_iter().enumerate() {
        ranking.offer(c, index);
    }
    let Some(((_, _, chosen), runner_up)) = ranking.finish() else {
        panic!("candidates were offered");
    };
    assert_eq!(chosen, 2);
    assert_eq!(runner_up, Some(cheaper_past_bound));

    let mut ranking = MergeRanking::new(u64::MAX, 0);
    for (index, c) in [dearer, within, cheaper_past_bound].into_iter().enumerate() {
        ranking.offer(c, index);
    }
    let Some(((_, _, chosen), runner_up)) = ranking.finish() else {
        panic!("candidates were offered");
    };
    assert_eq!(chosen, 2);
    assert_eq!(runner_up, Some(within));
}

fn cost(promoted: u64, pulled_in: u64) -> MergeCost {
    MergeCost {
        promoted,
        total: promoted + pulled_in,
    }
}

/// The smallest merge is not the cheapest way to pay down a level: a merge
/// promoting 10 MiB into 90 MiB rewrites ten bytes per promoted byte, one
/// promoting 100 MiB into 150 MiB rewrites 2.5. Ranking by total input alone
/// took the first, and has to run it ten times to retire what the second
/// retires once.
#[test]
fn a_merge_is_chosen_by_bytes_rewritten_per_promoted_byte() {
    let small = cost(10 * MIB, 90 * MIB);
    let large = cost(100 * MIB, 150 * MIB);
    assert_eq!(
        rank_by_promoted_ratio(&[small, large], 100 * MIB, 64 * MIB),
        Some(1),
        "the candidate that rewrites less per promoted byte wins",
    );
}

/// A candidate that promotes far more than the level owes is passed over for
/// one within the debt, however cheap its ratio, so a level is not asked to
/// push down much more than its overshoot.
#[test]
fn a_merge_promoting_far_past_the_overshoot_loses_to_one_within_it() {
    let within = cost(20 * MIB, 60 * MIB);
    let far_past = cost(1_000 * MIB, 100 * MIB);
    assert!(
        far_past.cheaper_than(within),
        "the oversized candidate has the better ratio"
    );
    assert_eq!(
        rank_by_promoted_ratio(&[far_past, within], 20 * MIB, 64 * MIB),
        Some(1),
    );
}

/// When the debt is smaller than any candidate, the cheapest one overall is
/// still taken: the level has to make progress.
#[test]
fn a_level_whose_debt_is_below_every_candidate_still_merges() {
    let a = cost(200 * MIB, 600 * MIB);
    let b = cost(300 * MIB, 300 * MIB);
    assert_eq!(rank_by_promoted_ratio(&[a, b], MIB, 0), Some(1));
    assert_eq!(rank_by_promoted_ratio(&[], MIB, 0), None);
}

/// The ratios are compared exactly at byte counts where a cross-product of two
/// `u64`s would wrap: `u64::MAX / 3 * 2` promoted against `u64::MAX / 2`
/// rewritten per side, where the rational answer is known.
#[test]
fn ratios_are_compared_exactly_near_the_top_of_the_range() {
    let big = u64::MAX / 4;
    // 1 rewritten byte per promoted byte against a hair over 1.
    let exact = MergeCost {
        promoted: big,
        total: big,
    };
    let slightly_worse = MergeCost {
        promoted: big - 1,
        total: big,
    };
    assert!(
        (u128::from(exact.total) * u128::from(slightly_worse.promoted)) > u128::from(u64::MAX),
        "the cross-product is past what a u64 holds",
    );
    assert_eq!(
        rank_by_promoted_ratio(&[slightly_worse, exact], u64::MAX, 0),
        Some(1)
    );
    assert_eq!(
        rank_by_promoted_ratio(&[exact, slightly_worse], u64::MAX, 0),
        Some(0)
    );
}

/// Equal ratios keep the earliest candidate, so the same version and hidden
/// set give the same table set on every call.
#[test]
fn equal_ratios_keep_the_earliest_candidate() {
    let a = cost(10 * MIB, 20 * MIB);
    let b = cost(20 * MIB, 40 * MIB);
    for _ in 0..3 {
        assert_eq!(rank_by_promoted_ratio(&[a, b], u64::MAX, 0), Some(0));
        assert_eq!(rank_by_promoted_ratio(&[b, a], u64::MAX, 0), Some(0));
    }
}

/// Flushes `keys` with `value_len`-byte values as one table and moves it from
/// L0 to `level`, returning its id.
fn table_at_level(
    tree: &crate::AnyTree,
    seqno: &SequenceNumberCounter,
    keys: &[&str],
    value_len: usize,
    level: u8,
) -> crate::Result<TableId> {
    for key in keys {
        tree.insert(*key, vec![b'v'; value_len], seqno.next());
    }
    tree.flush_active_memtable(0)?;
    let Some(id) = tree
        .current_version()
        .level(0)
        .and_then(|l0| l0.iter().flat_map(|run| run.iter()).map(Table::id).max())
    else {
        panic!("the flush wrote a table to L0");
    };
    tree.compact(Arc::new(crate::compaction::MoveDown(0, level)), 0)?;
    Ok(id)
}

/// The picker itself, over real levels shaped like the issue's example: a
/// dense region of L2 that a small L1 table overlaps, and a sparser region a
/// large L1 table overlaps. Merging the small table rewrites 100 KiB to promote
/// 10; merging the large one rewrites 250 KiB to promote 100. Ranking by total
/// input took the first.
#[test]
fn the_picker_chooses_the_merge_with_the_least_rewriting_per_promoted_byte() -> crate::Result<()> {
    const KIB: usize = 1_024;
    let dir = tempfile::tempdir()?;
    let seqno = SequenceNumberCounter::default();
    let tree = Config::new(dir.path(), seqno.clone(), SequenceNumberCounter::default())
        .data_block_compression_policy(crate::config::CompressionPolicy::disabled())
        .open()?;

    let dense = ["a0", "a1", "a2", "a3", "a4", "a5", "a6", "a7", "a8", "a9"];
    let sparse = ["b0", "b1", "b2", "b3", "b4", "b5", "b6", "b7", "b8", "b9"];
    table_at_level(&tree, &seqno, &dense, 9 * KIB, 2)?;
    let sparse_l2 = table_at_level(&tree, &seqno, &sparse, 15 * KIB, 2)?;
    table_at_level(&tree, &seqno, &["a3", "a4"], 5 * KIB, 1)?;
    let large_l1 = table_at_level(
        &tree,
        &seqno,
        &["b1", "b2", "b3", "b4", "b5", "b6", "b7", "b8"],
        12 * KIB + KIB / 2,
        1,
    )?;

    let version = tree.current_version();
    let (Some(l1), Some(l2)) = (version.level(1), version.level(2)) else {
        panic!("the fixture fills L1 and L2");
    };
    let Some((chosen, _)) = pick_minimal_compaction(
        l1,
        l2,
        &HiddenSet::default(),
        u64::MAX,
        64 * 1_024 * 1_024,
        64 * 1_024 * 1_024,
        &crate::comparator::DefaultUserComparator,
    ) else {
        panic!("a merge candidate exists");
    };
    assert_eq!(
        chosen,
        [sparse_l2, large_l1].into_iter().collect::<HashSet<_>>(),
        "the large table over the sparse region is the cheaper merge per promoted byte",
    );
    Ok(())
}