coordinode-lsm-tree 7.4.1

use super::*;
use crate::fs::StdFs;
use test_log::test;

/// A carried frame too short for a header is a malformed copy; one whose
/// header is long enough but not a block header keeps the decoder's answer.
#[test]
fn carried_header_refuses_a_short_or_foreign_frame() {
    assert!(matches!(
        carried_header(&[0u8; 2]),
        Err(crate::Error::InvalidHeader(
            "carried block header is cut short"
        ))
    ));
    assert!(matches!(
        carried_header(&[0u8; 64]),
        Err(crate::Error::InvalidHeader("Block"))
    ));
}

/// The encoded bytes of a `linked_blob_files` section holding `links`, as the
/// writer lays them out.
fn encode_links(links: &[LinkedFile]) -> Vec<u8> {
    let narrow = |n: usize| -> u64 {
        let Ok(n) = u64::try_from(n) else {
            panic!("{n} fits a u64");
        };
        n
    };
    let Ok(count) = u32::try_from(links.len()) else {
        panic!("a test section holds few links");
    };
    let mut out = count.to_le_bytes().to_vec();
    for link in links {
        out.extend_from_slice(&link.blob_file_id.to_le_bytes());
        out.extend_from_slice(&narrow(link.len).to_le_bytes());
        out.extend_from_slice(&link.bytes.to_le_bytes());
        out.extend_from_slice(&link.on_disk_bytes.to_le_bytes());
        for key in [&link.first_key, &link.last_key] {
            let Ok(len) = u16::try_from(key.len()) else {
                panic!("a test key is short");
            };
            out.extend_from_slice(&len.to_le_bytes());
            out.extend_from_slice(key);
        }
    }
    out
}

fn sample_links() -> Vec<LinkedFile> {
    vec![
        LinkedFile {
            blob_file_id: 3,
            len: 2,
            bytes: 200,
            on_disk_bytes: 120,
            first_key: b"apple".into(),
            last_key: b"banana".into(),
        },
        LinkedFile {
            blob_file_id: 9,
            len: 1,
            bytes: 50,
            on_disk_bytes: 40,
            first_key: b"cherry".into(),
            last_key: b"cherry".into(),
        },
    ]
}

/// What the writer lays out, the parser reads back record for record.
#[test]
fn linked_blob_files_round_trip_with_their_key_spans() -> crate::Result<()> {
    let links = sample_links();
    let encoded = encode_links(&links);
    assert_eq!(parse_linked_blob_files(&encoded)?, links);
    Ok(())
}

/// The section size kept up to date as entries register is exactly the bytes
/// the links encode to, whether a file's last key grows or shrinks, and none
/// once the links are taken.
#[test]
fn linked_blob_files_track_their_section_len_as_entries_register() {
    let mut linked = LinkedBlobFiles::default();
    assert_eq!(linked.section_len(), 0);
    for (file, key) in [
        (3, "apple"),
        (9, "banana-split"),
        (3, "cherry-pie"),
        (9, "date"),
        (3, "elderberry"),
        (5, "fig"),
    ] {
        linked.register(file, 10, 7, &UserKey::from(key.as_bytes()), true);
        let links: Vec<LinkedFile> = linked.files.values().cloned().collect();
        assert_eq!(linked.section_len(), encode_links(&links).len() as u64);
    }
    assert_eq!(linked.len(), 3);
    assert_eq!(linked.take().count(), 3);
    assert_eq!(linked.section_len(), 0);
}

/// A borrowed reference links its file without counting toward it: dropping
/// the table charges a file only for the objects the table owns, while the
/// link alone keeps the file from being removed or relocated under it.
#[test]
fn a_borrowed_reference_links_its_file_without_counts() {
    let mut linked = LinkedBlobFiles::default();
    linked.register(4, 100, 60, &UserKey::from("a"), false);
    linked.register(4, 30, 20, &UserKey::from("b"), true);
    linked.register(4, 100, 60, &UserKey::from("c"), false);
    linked.register(8, 50, 40, &UserKey::from("d"), false);
    let mut links: Vec<LinkedFile> = linked.take().collect();
    links.sort_by_key(|link| link.blob_file_id);
    assert_eq!(
        links,
        vec![
            LinkedFile {
                blob_file_id: 4,
                bytes: 30,
                on_disk_bytes: 20,
                len: 1,
                first_key: UserKey::from("a"),
                last_key: UserKey::from("c"),
            },
            LinkedFile {
                blob_file_id: 8,
                bytes: 0,
                on_disk_bytes: 0,
                len: 0,
                first_key: UserKey::from("d"),
                last_key: UserKey::from("d"),
            },
        ]
    );
    // A borrowed-only record survives the section's round trip.
    assert_eq!(
        parse_linked_blob_files(&encode_links(&links)).ok(),
        Some(links)
    );
}

/// The records must fill the section: bytes left past the last record mean
/// the count was forged down (or the section grew), not a shorter list.
#[test]
fn linked_blob_files_reject_bytes_past_the_records() {
    let mut encoded = encode_links(&sample_links());
    encoded.extend_from_slice(&[0; 3]);
    assert!(matches!(
        parse_linked_blob_files(&encoded),
        Err(crate::Error::InvalidHeader(
            "linked_blob_files: bytes remain past the declared records"
        ))
    ));
}

/// A key length that reaches past the section is corrupt, not a key to read
/// from whatever follows.
#[test]
fn linked_blob_files_reject_a_key_running_past_the_section() {
    let mut encoded = encode_links(&sample_links());
    let last_key_len_at = encoded.len() - b"cherry".len() - 2;
    let Some(len) = encoded.get_mut(last_key_len_at..last_key_len_at + 2) else {
        panic!("the last key's length is within the section");
    };
    len.copy_from_slice(&100u16.to_le_bytes());
    assert!(matches!(
        parse_linked_blob_files(&encoded),
        Err(crate::Error::InvalidHeader(
            "linked_blob_files: a key runs past the section"
        ))
    ));
}

/// A forged count far beyond what the bytes can hold fails before anything is
/// reserved for it.
#[test]
fn linked_blob_files_reject_a_count_the_section_cannot_hold() {
    let mut encoded = encode_links(&sample_links());
    let Some(count) = encoded.get_mut(..4) else {
        panic!("the count is within the section");
    };
    count.copy_from_slice(&u32::MAX.to_le_bytes());
    assert!(matches!(
        parse_linked_blob_files(&encoded),
        Err(crate::Error::InvalidHeader(
            "linked_blob_files: declared record count exceeds section size"
        ))
    ));
}

/// An input stamped while the clock read zero has no age, and an output that
/// takes any of its data has none either: the newest age of the other inputs
/// would make data that never expires eligible for a TTL.
#[test]
fn an_input_with_no_age_leaves_the_output_without_one() {
    let key = |n: u32| UserKey::from(n.to_be_bytes().as_slice());
    let inputs = vec![
        InputAge {
            min: key(0),
            max: key(10),
            created_at: 0,
        },
        InputAge {
            min: key(5),
            max: key(20),
            created_at: 1_000,
        },
    ];
    let mut sweep = AgeSweep::new(inputs, crate::comparator::default_comparator());
    assert_eq!(sweep.age_of(&key(0), &key(8)), Some(0), "both inputs met");
    assert_eq!(
        sweep.age_of(&key(15), &key(20)),
        Some(1_000),
        "only the dated input met"
    );
}

/// The swept age of each output in an ascending chain must be what checking
/// every input gives, for disjoint inputs, overlapping ones (an L0 table
/// spanning many others), and outputs before, between and after them.
#[test]
fn swept_ages_match_every_input_checked() {
    let key = |n: u32| UserKey::from(n.to_be_bytes().as_slice());
    let mut inputs: Vec<InputAge> = (0..40u32)
        .map(|i| InputAge {
            min: key(i * 10),
            max: key(i * 10 + 5),
            created_at: u128::from(i * 7 % 13),
        })
        .collect();
    // Wide tables overlapping many disjoint ones, as L0 tables over a level.
    inputs.push(InputAge {
        min: key(3),
        max: key(390),
        created_at: 5,
    });
    inputs.push(InputAge {
        min: key(150),
        max: key(160),
        created_at: 100,
    });
    let comparator = crate::comparator::default_comparator();
    // The newest age of the inputs met, or none (zero) when one of them has none.
    let naive = |first: &[u8], last: &[u8]| {
        let met: Vec<u128> = inputs
            .iter()
            .filter(|input| {
                comparator.compare(&input.min, last) != core::cmp::Ordering::Greater
                    && comparator.compare(&input.max, first) != core::cmp::Ordering::Less
            })
            .map(|input| input.created_at)
            .collect();
        if met.contains(&0) {
            Some(0)
        } else {
            met.into_iter().max()
        }
    };
    // One chain per output width, each output starting after the last ended.
    for width in [0u32, 1, 4, 9, 30, 200] {
        let mut sweep = AgeSweep::new(inputs.clone(), comparator.clone());
        let mut first = 0u32;
        while first < 420 {
            let (lo, hi) = (key(first), key(first + width));
            assert_eq!(
                sweep.age_of(&lo, &hi),
                naive(&lo, &hi),
                "output {lo:?}..={hi:?} of width {width}"
            );
            first += width + 1;
        }
    }
}

#[test]
fn finish_rejects_a_delete_bitmap_without_a_zone_map() -> crate::Result<()> {
    // The positional mask resolves each block's start row from the zone map,
    // so a segment that marks deletes must also carry one. The writer must
    // reject the misconfiguration at finish() rather than emit an SST that
    // then fails to open.
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path, 1, 0, Arc::new(StdFs))?;
    writer.write(InternalValue::from_components(
        b"a",
        b"v",
        1,
        ValueType::Value,
    ))?;
    // Mark a delete, but never enable the zone map.
    writer.delete_bitmap_mut().insert(0);
    match writer.finish() {
        Ok(_) => panic!("must reject a delete-bitmap without a zone map"),
        Err(err) => assert!(
            matches!(err, crate::Error::InvalidHeader(_)),
            "expected an InvalidHeader error, got {err:?}",
        ),
    }
    Ok(())
}

#[test]
fn table_writer_count() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path, 1, 0, Arc::new(StdFs))?;

    assert_eq!(0, writer.meta.key_count);
    assert_eq!(0, writer.chunk_size);

    writer.write(InternalValue::from_components(
        b"a",
        b"a",
        0,
        ValueType::Value,
    ))?;
    assert_eq!(1, writer.meta.key_count);
    assert_eq!(2, writer.chunk_size);

    writer.write(InternalValue::from_components(
        b"b",
        b"b",
        0,
        ValueType::Value,
    ))?;
    assert_eq!(2, writer.meta.key_count);
    assert_eq!(4, writer.chunk_size);

    writer.write(InternalValue::from_components(
        b"c",
        b"c",
        0,
        ValueType::Value,
    ))?;
    assert_eq!(3, writer.meta.key_count);
    assert_eq!(6, writer.chunk_size);

    writer.spill_block()?;
    assert_eq!(0, writer.chunk_size);

    Ok(())
}

/// A shard scheme whose parity trailer for a block would exceed the payload
/// hard cap (256 MiB) must be rejected at WRITE time: the out-of-band
/// verifier bounds the trailer at that cap before reserving its buffer, so a
/// writer that could emit a larger one would produce an SST the verifier
/// falsely flags as corrupt. Rejecting the write keeps the cap an invariant —
/// every SST in existence stays within the verifier's supported envelope.
#[cfg(feature = "page_ecc")]
#[test]
fn writer_rejects_a_block_whose_parity_exceeds_the_hard_cap() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    // RS(1,255): every payload byte amplified 255x into parity. Keep the
    // payload JUST above the 256 MiB cap (1_100_000 × 255 ≈ 268 MiB) so a
    // regression of the pre-encoding guard allocates barely past the cap
    // instead of half a gigabyte destabilizing CI.
    let mut writer = Writer::new(path, 1, 0, Arc::new(StdFs))?
        .use_ecc(Some(crate::table::block::EccParams::try_new(1, 255)?));
    let write_result = writer.write(InternalValue::from_components(
        b"k".as_slice(),
        vec![0xABu8; 1_100_000],
        0,
        ValueType::Value,
    ));
    let result = match write_result {
        Ok(()) => writer.finish().map(|_| ()),
        Err(e) => Err(e),
    };
    assert!(
        matches!(result, Err(crate::Error::FeatureUnsupported(_))),
        "an over-cap parity trailer must fail the write loudly, got {result:?}",
    );
    Ok(())
}

#[test]
#[should_panic(expected = "index block restart interval must be greater than zero")]
fn writer_rejects_zero_index_block_restart_interval() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    let _writer = writer.use_index_block_restart_interval(0);
}

#[test]
#[should_panic(expected = "data block restart interval must be greater than zero")]
fn writer_rejects_zero_data_block_restart_interval() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    let _writer = writer.use_data_block_restart_interval(0);
}

#[test]
#[should_panic(expected = "data block restart interval must be configured before writing starts")]
fn writer_rejects_data_block_restart_interval_change_after_write() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let mut writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    if let Err(e) = writer.write(InternalValue::from_components(
        b"a",
        b"v",
        0,
        ValueType::Value,
    )) {
        panic!("write should succeed: {e}");
    }
    let _writer = writer.use_data_block_restart_interval(2);
}

/// The table records one hash index ratio for all its blocks, and the hashes
/// a block's index buckets by are kept from its first item on, so the ratio
/// is set before the first write, not between them.
#[test]
#[should_panic(expected = "data block hash ratio must be configured before writing starts")]
fn writer_rejects_data_block_hash_ratio_change_after_write() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let mut writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    if let Err(e) = writer.write(InternalValue::from_components(
        b"a",
        b"v",
        0,
        ValueType::Value,
    )) {
        panic!("write should succeed: {e}");
    }
    let _writer = writer.use_data_block_hash_ratio(1.33);
}

#[test]
#[should_panic(expected = "index block restart interval must be configured before writing starts")]
fn writer_rejects_index_block_restart_interval_change_after_write() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let mut writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    if let Err(e) = writer.write(InternalValue::from_components(
        b"a",
        b"v",
        0,
        ValueType::Value,
    )) {
        panic!("write should succeed: {e}");
    }
    let _writer = writer.use_index_block_restart_interval(2);
}

#[test]
#[should_panic(expected = "partitioned index must be configured before writing starts")]
fn writer_rejects_partitioned_index_switch_after_write() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let mut writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    if let Err(e) = writer.write(InternalValue::from_components(
        b"a",
        b"v",
        0,
        ValueType::Value,
    )) {
        panic!("write should succeed: {e}");
    }
    let _writer = writer.use_partitioned_index();
}

#[test]
fn writer_meta_partition_size_is_chainable_with_full_index_writer() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("full-index");
    let mut writer = Writer::new(path, 1, 0, Arc::new(StdFs))?.use_meta_partition_size(8_192);

    writer.write(InternalValue::from_components(
        b"k",
        b"v",
        0,
        ValueType::Value,
    ))?;
    writer.spill_block()?;

    Ok(())
}

#[test]
#[should_panic(expected = "partitioned filter must be configured before writing starts")]
fn writer_rejects_partitioned_filter_switch_after_write() {
    let dir = match tempfile::tempdir() {
        Ok(dir) => dir,
        Err(e) => panic!("tempdir should be created: {e}"),
    };
    let path = dir.path().join("1");
    let mut writer = match Writer::new(path, 1, 0, Arc::new(StdFs)) {
        Ok(writer) => writer,
        Err(e) => panic!("writer should be created: {e}"),
    };
    if let Err(e) = writer.write(InternalValue::from_components(
        b"a",
        b"v",
        0,
        ValueType::Value,
    )) {
        panic!("write should succeed: {e}");
    }
    let _writer = writer.use_partitioned_filter();
}

/// A group whose tag a group of the table already took is encoded under
/// another one: tags stay unique within the table.
#[cfg(feature = "columnar")]
#[test]
fn a_group_tag_already_taken_is_skipped() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_created_at(1_000);
    let batch = crate::table::columnar::ColumnBatch {
        row_count: 0,
        columns: alloc::vec![],
    };
    let tag = writer.next_group_tag(&batch, b"k");
    writer.group_tags.insert(tag);
    let next = writer.next_group_tag(&batch, b"k");
    assert_ne!(next, 0, "a group is never tagged zero");
    assert!(
        !writer.group_tags.contains(&next),
        "a taken tag was handed out"
    );
    Ok(())
}

/// The tags of two tables of one id and one age, the groups written given as
/// `(key, value)` rows, one group each, sorted.
#[cfg(feature = "columnar")]
fn tags_of(
    dir: &std::path::Path,
    name: &str,
    groups: &[(&[u8], &[u8])],
) -> crate::Result<Vec<u64>> {
    use crate::comparator::default_comparator;
    use crate::table::columnar::entries_to_column_batch;

    let mut writer = Writer::new(dir.join(name), 1, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_created_at(1_000);
    for (seqno, (key, value)) in (1..).zip(groups) {
        let batch = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
            key.to_vec(),
            value.to_vec(),
            seqno,
            ValueType::Value,
        )])?;
        writer.write_columnar_block_verbatim(&batch, &default_comparator())?;
    }
    let mut tags: Vec<u64> = writer.group_tags.iter().copied().collect();
    tags.sort_unstable();
    Ok(tags)
}

/// Two tables of one id and one age, as two trees on one fixed clock write,
/// tag every group apart when its rows differ, so a page carried to the other
/// table is refused by its stamp, and alike when the rows are the same, so
/// the same writes give the same bytes. A later group counts as much as the
/// first: the same first group followed by different ones must not repeat
/// the later tags.
#[cfg(feature = "columnar")]
#[test]
fn group_tags_differ_for_different_content_at_one_id_and_age() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let dir = dir.path();
    let a = tags_of(dir, "a", &[(b"apple", b"v"), (b"cherry", b"x")])?;
    let a_again = tags_of(dir, "a2", &[(b"apple", b"v"), (b"cherry", b"x")])?;
    let other_key = tags_of(dir, "b", &[(b"banana", b"v"), (b"cherry", b"x")])?;
    let other_value = tags_of(dir, "c", &[(b"apple", b"w"), (b"cherry", b"x")])?;
    let other_later = tags_of(dir, "d", &[(b"apple", b"v"), (b"cherry", b"y")])?;

    assert_eq!(a.len(), 2, "both groups were encoded");
    assert_eq!(a, a_again, "the same groups took other tags");
    for (other, what) in [
        (&other_key, "a different first key"),
        (&other_value, "a different first value"),
        (&other_later, "a different later group"),
    ] {
        let shared = a.iter().filter(|tag| other.contains(tag)).count();
        assert!(shared < 2, "{what} left every tag shared: {a:?} {other:?}");
    }
    Ok(())
}

/// Two groups whose columns hold the same bytes under different types are
/// different groups, and take different tags.
#[cfg(feature = "columnar")]
#[test]
fn group_tags_differ_for_columns_of_another_type() -> crate::Result<()> {
    use crate::table::columnar::{Column, ColumnBatch, Number, TypeTag};

    let dir = tempfile::tempdir()?;
    let batch = |type_tag| ColumnBatch {
        row_count: 1,
        columns: alloc::vec![Column {
            column_id: 0,
            type_tag,
            validity: None,
            data: crate::Slice::from(7u64.to_le_bytes().to_vec()),
        }],
    };
    let tag_of = |name: &str, type_tag| -> crate::Result<u64> {
        let mut writer = Writer::new(dir.path().join(name), 1, 0, Arc::new(StdFs))?
            .use_columnar(true)
            .use_created_at(1_000);
        Ok(writer.next_group_tag(&batch(type_tag), b"k"))
    };

    let fixed = tag_of("fixed", TypeTag::Fixed(8))?;
    let number = tag_of("number", TypeTag::Number(Number::U64_LE))?;
    assert_ne!(fixed, number, "one tag for columns of different types");
    Ok(())
}

/// A block re-emitted through the verbatim columnar path can hold several MVCC
/// versions of one user key (same key, descending seqno). Unlike bulk ingest,
/// that path must NOT reject equal user keys — only strictly-unique keys are an
/// ingest contract. Regression for the verbatim salvage re-emit path.
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_block_verbatim_accepts_mvcc_duplicate_keys() -> crate::Result<()> {
    use crate::comparator::default_comparator;
    use crate::table::columnar::entries_to_column_batch;

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let cmp = default_comparator();
    let mut writer = Writer::new(path, 1, 0, Arc::new(StdFs))?.use_columnar(true);

    // Two MVCC versions of "dup" (valid block order: user key ascending, seqno
    // descending within a key) — NOT strictly unique.
    let entries = alloc::vec![
        InternalValue::from_components(b"dup".to_vec(), b"v3".to_vec(), 3, ValueType::Value),
        InternalValue::from_components(b"dup".to_vec(), b"v1".to_vec(), 1, ValueType::Value),
    ];
    let batch = entries_to_column_batch(&entries)?;
    writer.write_columnar_block_verbatim(&batch, &cmp)?;
    assert!(
        writer.finish()?.is_some(),
        "the verbatim block writes and finishes"
    );
    Ok(())
}

/// When the same user key spans two consecutive direct blocks, the boundary
/// must keep the internal order `(user_key asc, seqno desc)`: a second block
/// whose FIRST version of the shared key carries a seqno >= the previous
/// block's LAST version is a tampered / malformed block and must be rejected,
/// exactly like an in-block inversion. A correctly descending boundary still
/// passes.
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_block_verbatim_rejects_an_equal_key_boundary_seqno_inversion() -> crate::Result<()>
{
    use crate::comparator::default_comparator;
    use crate::table::columnar::entries_to_column_batch;

    let dir = tempfile::tempdir()?;
    let cmp = default_comparator();

    // Inverted boundary: block 1 ends with ("dup", 5); block 2 begins with
    // ("dup", 6) — a NEWER version sorting after an older one.
    let mut writer = Writer::new(dir.path().join("inv"), 1, 0, Arc::new(StdFs))?.use_columnar(true);
    let first = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"dup".to_vec(),
        b"v5".to_vec(),
        5,
        ValueType::Value,
    )])?;
    writer.write_columnar_block_verbatim(&first, &cmp)?;
    let inverted = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"dup".to_vec(),
        b"v6".to_vec(),
        6,
        ValueType::Value,
    )])?;
    assert!(
        writer
            .write_columnar_block_verbatim(&inverted, &cmp)
            .is_err(),
        "an equal-key boundary whose seqno does not decrease is rejected",
    );

    // Valid boundary: the shared key's versions keep strictly decreasing
    // across the block edge.
    let mut ok_writer =
        Writer::new(dir.path().join("ok"), 1, 0, Arc::new(StdFs))?.use_columnar(true);
    let first = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"dup".to_vec(),
        b"v5".to_vec(),
        5,
        ValueType::Value,
    )])?;
    ok_writer.write_columnar_block_verbatim(&first, &cmp)?;
    let descending = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"dup".to_vec(),
        b"v3".to_vec(),
        3,
        ValueType::Value,
    )])?;
    ok_writer.write_columnar_block_verbatim(&descending, &cmp)?;
    assert!(
        ok_writer.finish()?.is_some(),
        "a correctly descending equal-key boundary still writes and finishes",
    );
    Ok(())
}

/// The columnar bulk-ingest contract is enforced by `write_columnar_batch`:
/// a row-mode writer, a non-zero per-row seqno, or non-increasing keys are each
/// rejected before any block is written.
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_enforces_the_ingest_contract() -> crate::Result<()> {
    use crate::comparator::default_comparator;
    use crate::table::columnar::entries_to_column_batch;

    let dir = tempfile::tempdir()?;
    let cmp = default_comparator();
    let good = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"a".to_vec(),
        b"v".to_vec(),
        0,
        ValueType::Value,
    )])?;

    // 1. Row-mode writer (no `use_columnar`) rejects a columnar batch.
    let mut row_writer = Writer::new(dir.path().join("row"), 1, 0, Arc::new(StdFs))?;
    assert!(
        row_writer.write_columnar_batch(&good, &cmp).is_err(),
        "a columnar batch on a row-mode writer is rejected",
    );

    // 2. A non-zero per-row seqno is rejected (ingest assigns the seqno).
    let mut w2 = Writer::new(dir.path().join("s"), 1, 0, Arc::new(StdFs))?.use_columnar(true);
    let nonzero = entries_to_column_batch(&alloc::vec![InternalValue::from_components(
        b"a".to_vec(),
        b"v".to_vec(),
        7,
        ValueType::Value,
    )])?;
    assert!(
        w2.write_columnar_batch(&nonzero, &cmp).is_err(),
        "a non-zero per-row seqno is rejected on bulk ingest",
    );

    // 3. Non-increasing keys within the batch are rejected.
    let mut w3 = Writer::new(dir.path().join("k"), 1, 0, Arc::new(StdFs))?.use_columnar(true);
    let unsorted = entries_to_column_batch(&alloc::vec![
        InternalValue::from_components(b"b".to_vec(), b"v".to_vec(), 0, ValueType::Value),
        InternalValue::from_components(b"a".to_vec(), b"v".to_vec(), 0, ValueType::Value),
    ])?;
    assert!(
        w3.write_columnar_batch(&unsorted, &cmp).is_err(),
        "non-increasing keys are rejected on bulk ingest",
    );
    Ok(())
}

/// A columnar batch is written and registered before its locator slots are
/// folded in; the estimates the table rotates on must still count them.
#[cfg(feature = "columnar")]
#[test]
fn a_columnar_batch_leaves_the_estimates_current() -> crate::Result<()> {
    use crate::comparator::default_comparator;
    use crate::config::{LocatorPolicyEntry, LocatorPrecision};
    use crate::table::columnar::entries_to_column_batch;

    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_locator(LocatorPolicyEntry::Enabled {
            precision: LocatorPrecision::Entry,
            block_id_bits: None,
            slot_bits: None,
        });
    // One row group: its slots are the table's first, so they grow the
    // locator's width from nothing.
    let entries: alloc::vec::Vec<InternalValue> = (0..300u32)
        .map(|i| {
            InternalValue::from_components(
                format!("key{i:06}").into_bytes(),
                b"v".to_vec(),
                0,
                ValueType::Value,
            )
        })
        .collect();
    writer.write_columnar_batch(&entries_to_column_batch(&entries)?, &default_comparator())?;
    assert_eq!(writer.meta.data_block_count, 1);
    let (held, hint) = (writer.held_state_bytes(), writer.output_size_hint());
    writer.refresh_state_estimates();
    assert_eq!(held, writer.held_state_bytes(), "held state");
    assert_eq!(hint, writer.output_size_hint(), "size hint");
    Ok(())
}

/// Columnar bulk ingest with an `Entry`-precision locator records a per-key
/// locator slot for every distinct key (the per-entry-index arm of the direct
/// block accounting).
#[cfg(feature = "columnar")]
#[test]
fn write_columnar_batch_records_entry_precision_locator() -> crate::Result<()> {
    use crate::comparator::default_comparator;
    use crate::config::{LocatorPolicyEntry, LocatorPrecision};
    use crate::table::columnar::entries_to_column_batch;

    let dir = tempfile::tempdir()?;
    let cmp = default_comparator();
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_columnar(true)
        .use_locator(LocatorPolicyEntry::Enabled {
            precision: LocatorPrecision::Entry,
            block_id_bits: None,
            slot_bits: None,
        });
    let entries: alloc::vec::Vec<InternalValue> = (0..8u32)
        .map(|i| {
            InternalValue::from_components(
                format!("key{i:03}").into_bytes(),
                b"v".to_vec(),
                0,
                ValueType::Value,
            )
        })
        .collect();
    let batch = entries_to_column_batch(&entries)?;
    writer.write_columnar_batch(&batch, &cmp)?;

    // Validate the recorded locator slots, not just that finish() succeeds. All
    // eight keys land in the single direct block (ordinal 0), and Entry
    // precision records each key's row index as its slot, so the accumulated
    // triples are exactly `(hash64(key), 0, row)`.
    assert_eq!(
        writer.locators.len(),
        entries.len(),
        "one locator slot per distinct key",
    );
    for (row, recorded) in writer.locators.iter().enumerate() {
        let key = format!("key{row:03}");
        assert_eq!(
            *recorded,
            (crate::hash::hash64(key.as_bytes()), 0, row as u64),
            "key at row {row} maps to (hash, direct block 0, slot == row index)",
        );
    }

    assert!(
        writer.finish()?.is_some(),
        "the columnar SST with an entry-precision locator finishes",
    );
    Ok(())
}

/// The default must stay the codec's own behaviour: whoever asks for level 22
/// is asking for ratio, and gets it without opting into anything. Pinned here
/// because the cost of the opposite default is silent: the output stays valid,
/// only slightly larger, so nothing else would notice the flip.
#[cfg(zstd_any)]
#[test]
fn writer_keeps_the_two_pass_seed_on_by_default_across_subwriter_swaps() -> crate::Result<()> {
    assert!(
        crate::runtime_config::RuntimeConfig::default().zstd_two_pass_seed,
        "the runtime config ships with the seed on",
    );

    let dir = tempfile::tempdir()?;
    let writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?;
    assert!(
        writer.zstd_two_pass_seed,
        "a fresh table writer starts with the seed on",
    );

    // Selecting another index or filter layout replaces the subwriter, and the
    // default has to survive that too.
    let writer = writer.use_partitioned_index().use_partitioned_filter();
    assert!(
        writer.zstd_two_pass_seed,
        "swapping subwriters leaves the default in force",
    );

    Ok(())
}

/// The layouts whose finish-time sections are estimated differently.
#[derive(Clone, Copy, Debug)]
enum StateLayout {
    Full,
    Partitioned,
    Locator,
}

fn state_writer(path: crate::path::PathBuf, layout: StateLayout) -> crate::Result<Writer> {
    let writer = Writer::new(path, 1, 0, Arc::new(StdFs))?;
    Ok(match layout {
        StateLayout::Full => writer,
        StateLayout::Partitioned => writer.use_partitioned_index().use_partitioned_filter(),
        StateLayout::Locator => writer.use_locator(crate::config::LocatorPolicyEntry::Enabled {
            precision: crate::config::LocatorPrecision::Restart,
            block_id_bits: None,
            slot_bits: None,
        }),
    })
}

/// Writes `n` keys with `value_len`-byte values and spills the last block.
fn write_keys(writer: &mut Writer, n: u32, value_len: usize) -> crate::Result<()> {
    let value = alloc::vec![0x5a; value_len];
    for i in 0..n {
        writer.write(InternalValue::from_components(
            format!("key{i:010}").into_bytes(),
            value.clone(),
            0,
            ValueType::Value,
        ))?;
    }
    writer.spill_block()
}

/// A writer that has seen no key holds no state for `finish` and expects it to
/// append nothing past what is on disk.
#[test]
fn a_fresh_writer_holds_no_state_for_finish() -> crate::Result<()> {
    for layout in [
        StateLayout::Full,
        StateLayout::Partitioned,
        StateLayout::Locator,
    ] {
        let dir = tempfile::tempdir()?;
        let writer = state_writer(dir.path().join("1"), layout)?;
        assert_eq!(writer.held_state_bytes(), 0, "{layout:?}");
        assert_eq!(writer.finish_metadata_bytes, 0, "{layout:?}");
    }
    Ok(())
}

/// A table rotates on the size hint before `finish` has written its filter,
/// index and locator, so the hint must already count them: it lands close to
/// the finished file, which only adds the meta block and the table of contents.
#[test]
fn the_size_hint_before_finish_is_close_to_the_finished_table() -> crate::Result<()> {
    for layout in [
        StateLayout::Full,
        StateLayout::Partitioned,
        StateLayout::Locator,
    ] {
        let dir = tempfile::tempdir()?;
        let path = dir.path().join("1");
        let mut writer = state_writer(path.clone(), layout)?;
        write_keys(&mut writer, 50_000, 8)?;
        let hint = writer.output_size_hint();
        let data = *writer.meta.file_pos;
        assert!(
            hint > data,
            "{layout:?}: the hint counts the sections to come"
        );
        writer.finish()?;
        let size = std::fs::metadata(&path)?.len();
        assert!(
            hint * 10 >= size * 9 && hint * 10 <= size * 11,
            "{layout:?}: hint {hint} for a {size}-byte table",
        );
    }
    Ok(())
}

/// A table rotates on the estimates as they stand after its last block, so
/// they must already count that block's index entry and section entries:
/// recomputing them changes nothing.
#[test]
fn the_estimates_count_the_block_just_written() -> crate::Result<()> {
    for layout in [
        StateLayout::Full,
        StateLayout::Partitioned,
        StateLayout::Locator,
    ] {
        let dir = tempfile::tempdir()?;
        let mut writer = state_writer(dir.path().join("1"), layout)?
            .use_zone_map(true)
            .use_seqno_in_index(true);
        write_keys(&mut writer, 1_000, 8)?;
        let (held, hint) = (writer.held_state_bytes(), writer.output_size_hint());
        writer.refresh_state_estimates();
        assert_eq!(held, writer.held_state_bytes(), "{layout:?}: held state");
        assert_eq!(hint, writer.output_size_hint(), "{layout:?}: size hint");
    }
    Ok(())
}

/// The state a table holds for `finish` grows with its keys, not with its
/// bytes: the same keys with values a hundred times larger hold about the same
/// state, while their rows are twenty times larger. Only the index grows with the data,
/// by one entry per block.
#[test]
fn held_state_follows_the_keys_not_the_value_bytes() -> crate::Result<()> {
    for layout in [
        StateLayout::Full,
        StateLayout::Partitioned,
        StateLayout::Locator,
    ] {
        let dir = tempfile::tempdir()?;
        let mut small = state_writer(dir.path().join("1"), layout)?;
        let mut large = state_writer(dir.path().join("2"), layout)?;
        write_keys(&mut small, 20_000, 4)?;
        write_keys(&mut large, 20_000, 400)?;
        let (small_state, large_state) = (small.held_state_bytes(), large.held_state_bytes());
        assert!(
            small_state > 20_000 * 8,
            "{layout:?}: {small_state} bytes for 20 000 keys"
        );
        assert!(
            large_state < 2 * small_state,
            "{layout:?}: {large_state} bytes held for 400-byte values, {small_state} for 4-byte",
        );
        // 13-byte keys: rows of 17 and 413 bytes.
        assert!(
            large.output_size_hint() > 10 * small.output_size_hint(),
            "{layout:?}: the data grew with the values",
        );
    }
    Ok(())
}

/// Writes `n` keys of `key_len` bytes with 8-byte values and spills the last
/// block.
fn write_long_keys(writer: &mut Writer, n: u32, key_len: usize) -> crate::Result<()> {
    for i in 0..n {
        writer.write(InternalValue::from_components(
            format!("{i:0key_len$}").into_bytes(),
            b"value---".to_vec(),
            0,
            ValueType::Value,
        ))?;
    }
    writer.spill_block()
}

/// The size hint before `finish` lands within 10% of the finished table.
fn assert_hint_matches_the_table(writer: Writer, path: &std::path::Path) -> crate::Result<()> {
    let hint = writer.output_size_hint();
    writer.finish()?;
    let size = std::fs::metadata(path)?.len();
    assert!(
        hint * 10 >= size * 9 && hint * 10 <= size * 11,
        "hint {hint} for a {size}-byte table",
    );
    Ok(())
}

/// The top-level index is written twice, at its place and mirrored at the
/// tail. Small blocks under long keys make it a large share of the table, so
/// the hint lands near the finished size only if both copies are counted.
#[test]
fn the_size_hint_counts_the_mirrored_index() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?.use_data_block_size(64);
    write_long_keys(&mut writer, 5_000, 200)?;
    assert_hint_matches_the_table(writer, &path)
}

/// A prefix shared by every key hashes to the same token once per key. The
/// filter is built from distinct tokens, so the estimates, counted from every
/// buffered token, stay above what `finish` builds and writes.
#[test]
fn a_prefix_shared_by_every_key_is_built_once() -> crate::Result<()> {
    struct UpToColon;
    impl crate::prefix::PrefixExtractor for UpToColon {
        fn name(&self) -> &'static str {
            "up-to-colon"
        }

        fn prefixes<'a>(&self, key: &'a [u8]) -> Box<dyn Iterator<Item = &'a [u8]> + 'a> {
            let end = key.iter().position(|b| *b == b':').map_or(0, |i| i + 1);
            Box::new(key.get(..end).into_iter())
        }
    }
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?
        .use_prefix_extractor(Some(Arc::new(UpToColon)));
    for i in 0..20_000u32 {
        writer.write(InternalValue::from_components(
            format!("p:{i:08}").into_bytes(),
            b"v".to_vec(),
            0,
            ValueType::Value,
        ))?;
    }
    writer.spill_block()?;
    let hint = writer.output_size_hint();
    writer.finish()?;
    let size = std::fs::metadata(&path)?.len();
    assert!(hint >= size, "hint {hint} for a {size}-byte table");
    Ok(())
}

/// Every table ends with sections `finish` always writes: two copies of the
/// meta block, the version byte, the 4 KiB separator between them, the table
/// of contents and the trailer. A table of a few keys is mostly these. The
/// hint is not below the table, and above it by the 10% the other estimates
/// allow plus the filter's bound: a small filter is counted as if it built
/// every layer.
#[test]
fn the_size_hint_counts_the_tail_every_table_writes() -> crate::Result<()> {
    for key_len in [8, 1_000] {
        let dir = tempfile::tempdir()?;
        let path = dir.path().join("1");
        let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?;
        write_long_keys(&mut writer, 3, key_len)?;
        let hint = writer.output_size_hint();
        writer.finish()?;
        let size = std::fs::metadata(&path)?.len();
        let slack = crate::config::BloomConstructionPolicy::default().filter_size_bound(3) as u64;
        assert!(
            hint >= size && hint * 10 <= size * 11 + slack * 10,
            "hint {hint} for a {size}-byte table, filter slack {slack}",
        );
    }
    Ok(())
}

/// A table names the extractor its filter indexes in both copies of its meta
/// block, and a parameterised extractor's name has no bound: the hint counts
/// it, and the filter's other meta entries, before the filter is built.
#[test]
fn the_size_hint_counts_the_extractor_name_the_meta_records() -> crate::Result<()> {
    struct LongName(String);
    impl crate::prefix::PrefixExtractor for LongName {
        fn name(&self) -> &str {
            &self.0
        }

        fn prefixes<'a>(&self, key: &'a [u8]) -> Box<dyn Iterator<Item = &'a [u8]> + 'a> {
            Box::new(key.get(..1).into_iter())
        }
    }

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?
        .use_prefix_extractor(Some(Arc::new(LongName("x".repeat(64 * 1_024)))));
    write_long_keys(&mut writer, 3, 8)?;
    let hint = writer.output_size_hint();
    writer.finish()?;
    let size = std::fs::metadata(&path)?.len();
    assert!(hint >= size, "hint {hint} for a {size}-byte table");

    // Without a filter the meta names no extractor, and the hint counts
    // none: an unbounded name it never writes would make every table look
    // full long before it is.
    let path = dir.path().join("2");
    let mut writer = Writer::new(path.clone(), 2, 0, Arc::new(StdFs))?
        .use_bloom_policy(crate::config::BloomConstructionPolicy::BitsPerKey(0.0))
        .use_prefix_extractor(Some(Arc::new(LongName("x".repeat(64 * 1_024)))));
    write_long_keys(&mut writer, 3, 8)?;
    let hint = writer.output_size_hint();
    writer.finish()?;
    let size = std::fs::metadata(&path)?.len();
    assert!(
        hint >= size && hint * 10 <= size * 11,
        "hint {hint} for a {size}-byte table without a filter",
    );

    // A partitioned filter indexes no prefix and names no extractor either.
    // Its other estimates run above a table this small, so what is checked is
    // that the hint does not carry the name.
    let path = dir.path().join("3");
    let mut writer = Writer::new(path.clone(), 3, 0, Arc::new(StdFs))?
        .use_partitioned_filter()
        .use_prefix_extractor(Some(Arc::new(LongName("x".repeat(64 * 1_024)))));
    write_long_keys(&mut writer, 3, 8)?;
    let hint = writer.output_size_hint();
    writer.finish()?;
    let size = std::fs::metadata(&path)?.len();
    assert!(
        hint >= size && hint - size < 64 * 1_024,
        "hint {hint} for a {size}-byte table with a partitioned filter",
    );
    Ok(())
}

/// A table may close before its first block is cut, when what the multi-writer
/// adds at rotation fills it: the tail is counted from the first key on.
#[test]
fn the_size_hint_counts_the_tail_before_a_block_is_cut() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?;
    writer.write(InternalValue::from_components(
        b"key".to_vec(),
        b"v".to_vec(),
        0,
        ValueType::Value,
    ))?;
    assert_eq!(writer.meta.data_block_count, 0);
    assert!(
        writer.output_size_hint() > FIXED_TAIL_LEN,
        "hint {} before the first block",
        writer.output_size_hint(),
    );
    Ok(())
}

/// A large block gathers filter and locator state for many keys before it is
/// cut; the estimates count that state within the block, not only once the
/// block is written.
#[test]
fn the_estimates_follow_the_keys_of_a_block_not_yet_cut() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer =
        Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?.use_data_block_size(4 << 20);
    for i in 0..20_000u32 {
        writer.write(InternalValue::from_components(
            format!("key{i:06}").into_bytes(),
            b"v".to_vec(),
            0,
            ValueType::Value,
        ))?;
    }
    assert_eq!(writer.meta.data_block_count, 0);
    // One 8-byte hash per key, less the keys since the last refresh.
    assert!(
        writer.held_state_bytes() >= 19_000 * 8,
        "{} held for 20 000 keys",
        writer.held_state_bytes(),
    );
    Ok(())
}

/// `finish` encodes each section into a buffer it keeps to the end and frames
/// it into another, while the section itself is still held: a large section
/// counts three times at that point.
#[test]
fn the_held_state_counts_the_section_encoding_buffers() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(256)
        .use_zone_map(true);
    write_long_keys(&mut writer, 5_000, 200)?;
    assert!(
        writer.held_state_bytes() >= 3 * writer.zone_map_bytes,
        "{} held for {} bytes of zone-map bounds",
        writer.held_state_bytes(),
        writer.zone_map_bytes,
    );
    Ok(())
}

/// A delete bitmap stores each touched chunk with its index, kind and count,
/// so sparse deletes cost more than their row count: both estimates count the
/// bitmap as it will be encoded.
#[test]
fn the_estimates_count_a_sparse_delete_bitmap_as_encoded() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?.use_zone_map(true);
    write_keys(&mut writer, 100, 8)?;
    let (held, hint) = (writer.held_state_bytes(), writer.output_size_hint());
    for chunk in 0..1_000 {
        writer
            .delete_bitmap_mut()
            .insert(chunk * crate::table::delete_bitmap::CHUNK_ROWS);
    }
    writer.refresh_state_estimates();
    let encoded = writer.delete_bitmap.encode().len() as u64;
    assert!(
        writer.output_size_hint() - hint >= encoded,
        "the hint grew by {} for a {encoded}-byte bitmap",
        writer.output_size_hint() - hint,
    );
    assert!(writer.held_state_bytes() - held >= encoded);
    Ok(())
}

/// `finish` encodes the delete bitmap once and keeps the bytes for the meta's
/// content hash, while it copies them into the block buffer and seals that
/// into a frame: with its containers, four copies are live at once.
#[cfg(feature = "encryption")]
#[test]
fn the_held_state_counts_the_retained_delete_bitmap_encoding() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_zone_map(true)
        .use_encryption(Some(Arc::new(crate::encryption::Aes256GcmProvider::new(
            &[7; 32],
        ))));
    write_keys(&mut writer, 100, 8)?;
    for chunk in 0..10_000 {
        writer
            .delete_bitmap_mut()
            .insert(chunk * crate::table::delete_bitmap::CHUNK_ROWS);
    }
    writer.refresh_state_estimates();
    let encoded = writer.delete_bitmap.encode().len() as u64;
    let containers = writer.delete_bitmap.heap_len_bound();
    assert!(
        writer.held_state_bytes() >= containers + 3 * encoded,
        "the held state is {} for a {encoded}-byte bitmap in {containers} bytes",
        writer.held_state_bytes(),
    );
    Ok(())
}

/// Zone-map entries own copies of each block's bounds; under long keys those
/// dominate the section, and both estimates have to count them.
#[test]
fn the_estimates_count_the_zone_map_bounds() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(256)
        .use_zone_map(true);
    write_long_keys(&mut writer, 5_000, 200)?;
    let blocks = writer.zone_map_section.len() as u64;
    assert!(
        writer.held_state_bytes() > blocks * 2 * 200,
        "{} bytes held for {blocks} blocks of 200-byte bounds",
        writer.held_state_bytes(),
    );
    assert_hint_matches_the_table(writer, &path)
}

/// Under page ECC every block `finish` writes carries a parity trailer, which
/// the hint has to count: long keys over small blocks make the index a large
/// share of the table.
#[cfg(feature = "page_ecc")]
#[test]
fn the_size_hint_counts_the_parity_of_the_blocks_to_come() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(64)
        .use_ecc(Some(crate::table::block::EccParams::RS_4_2));
    write_long_keys(&mut writer, 5_000, 200)?;
    assert_hint_matches_the_table(writer, &path)
}

/// A table of exactly `2^k` blocks fits explicit `k`-bit block ids, so its
/// locator section is written and both estimates count it, although the next
/// block ordinal no longer fits.
#[test]
fn a_locator_filled_to_its_last_block_id_is_counted() -> crate::Result<()> {
    let locator = crate::config::LocatorPolicyEntry::Enabled {
        precision: crate::config::LocatorPrecision::Block,
        block_id_bits: Some(1),
        slot_bits: None,
    };
    let dir = tempfile::tempdir()?;
    let mut with = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?.use_locator(locator);
    let mut without = Writer::new(dir.path().join("2"), 1, 0, Arc::new(StdFs))?;
    for writer in [&mut with, &mut without] {
        for block in 0..2 {
            for i in 0..10 {
                writer.write(InternalValue::from_components(
                    format!("key{block}{i:02}").into_bytes(),
                    b"value---".to_vec(),
                    0,
                    ValueType::Value,
                ))?;
            }
            writer.spill_block()?;
        }
    }
    assert_eq!(with.meta.data_block_count, 2);
    assert!(
        with.held_state_bytes() > without.held_state_bytes(),
        "{} held with the locator, {} without",
        with.held_state_bytes(),
        without.held_state_bytes(),
    );
    assert!(with.output_size_hint() > without.output_size_hint());
    Ok(())
}

/// A compaction's lineage lists its inputs and stays in the writer to the end;
/// the meta block then copies it into the parameters, the encoded ids and the
/// meta entry, so the held state counts it once held and again encoded.
#[test]
fn the_held_state_counts_the_lineage_and_its_meta_encoding() -> crate::Result<()> {
    const INPUTS: u64 = 100_000;
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_lineage(Some((0..INPUTS).collect()));
    writer.write(InternalValue::from_components(
        b"key".to_vec(),
        b"v".to_vec(),
        0,
        ValueType::Value,
    ))?;
    writer.spill_block()?;
    let ids = INPUTS * 8;
    assert!(
        writer.held_state_bytes() >= 4 * ids,
        "{} held for a lineage of {ids} bytes",
        writer.held_state_bytes(),
    );
    Ok(())
}

/// Parallel compression on a private four-worker pool, the inline threshold
/// derived from the block size.
#[cfg(feature = "parallel")]
fn four_workers() -> crate::Result<super::ParallelCompression> {
    Ok(super::ParallelCompression {
        spawner: Arc::new(super::RayonSpawner::with_threads(4)?),
        threads: 4,
        inline_below: None,
    })
}

/// Blocks in flight on the parallel pipeline are counted by the frames they
/// will be written as, not by their payload: once drained, the bytes they
/// take on disk stay within what the estimate counted for them.
#[cfg(feature = "parallel")]
#[test]
fn blocks_in_flight_are_counted_by_their_frames() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_compression(crate::CompressionType::None)
        .use_parallel_compression(four_workers()?);
    for block in 0..3u32 {
        for i in 0..10u32 {
            writer.write(InternalValue::from_components(
                format!("key{block}{i:02}").into_bytes(),
                b"value---".to_vec(),
                0,
                ValueType::Value,
            ))?;
        }
        writer.spill_block()?;
    }
    assert_eq!(*writer.meta.file_pos, 0, "all three blocks are in flight");
    let counted = writer.output_size_hint() - writer.finish_metadata_bytes;
    for _ in 0..3 {
        writer.drain_one_parallel()?;
    }
    assert_eq!(writer.meta.data_block_count, 3);
    assert!(
        *writer.meta.file_pos <= counted,
        "{} bytes written for {counted} counted in flight",
        *writer.meta.file_pos,
    );
    Ok(())
}

/// Blocks in flight on the parallel pipeline hold their encoded payload and
/// then their frame until they are written: the held state counts them.
#[cfg(feature = "parallel")]
#[test]
fn blocks_in_flight_count_toward_the_held_state() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_compression(crate::CompressionType::None)
        .use_parallel_compression(four_workers()?);
    for block in 0..3u32 {
        for i in 0..10u32 {
            writer.write(InternalValue::from_components(
                format!("key{block}{i:02}").into_bytes(),
                vec![7u8; 1_000],
                0,
                ValueType::Value,
            ))?;
        }
        writer.spill_block()?;
    }
    assert_eq!(*writer.meta.file_pos, 0, "all three blocks are in flight");
    assert!(
        writer.held_state_bytes() >= 3 * 10 * 1_000,
        "{} held with three 10 KB blocks in flight",
        writer.held_state_bytes(),
    );
    Ok(())
}

/// A block both compressed and encrypted on the parallel pipeline holds its
/// payload, the compressed bytes and the sealed frame at once: the held state
/// counts all three.
#[cfg(all(feature = "parallel", feature = "encryption", feature = "lz4"))]
#[test]
fn blocks_in_flight_count_their_compression_and_encryption_buffers() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let held = |name: &str, encrypted: bool| -> crate::Result<u64> {
        let encryption: Option<Arc<dyn crate::encryption::EncryptionProvider>> =
            encrypted.then(|| Arc::new(crate::encryption::Aes256GcmProvider::new(&[7; 32])) as _);
        let mut writer = Writer::new(dir.path().join(name), 1, 0, Arc::new(StdFs))?
            .use_data_block_compression(crate::CompressionType::Lz4)
            .use_encryption(encryption)
            .use_parallel_compression(four_workers()?);
        for block in 0..3u32 {
            for i in 0..10u32 {
                writer.write(InternalValue::from_components(
                    format!("key{block}{i:02}").into_bytes(),
                    vec![7u8; 1_000],
                    0,
                    ValueType::Value,
                ))?;
            }
            writer.spill_block()?;
        }
        assert_eq!(*writer.meta.file_pos, 0, "all three blocks are in flight");
        Ok(writer.held_state_bytes())
    };
    let sealed = held("1", true)?;
    let plain = held("2", false)?;
    // Sealing keeps each block's compressed bytes besides its frame: about
    // 10 KB more per block than compressing alone.
    assert!(
        sealed >= plain + 3 * 10_000,
        "{sealed} held sealed against {plain} compressed alone",
    );
    Ok(())
}

/// Within a block not yet cut, an entry-precise locator records a slot per key,
/// so its section needs slot bits a block-precise one does not: the estimates
/// count the open block's slots, not only those of blocks already cut.
#[test]
fn the_estimates_count_the_locator_slots_of_a_block_not_yet_cut() -> crate::Result<()> {
    let locator = |precision| crate::config::LocatorPolicyEntry::Enabled {
        precision,
        block_id_bits: None,
        slot_bits: None,
    };
    let dir = tempfile::tempdir()?;
    let hint = |name: &str, precision| -> crate::Result<u64> {
        let mut writer = Writer::new(dir.path().join(name), 1, 0, Arc::new(StdFs))?
            .use_data_block_size(4 << 20)
            .use_locator(locator(precision));
        for i in 0..20_000u32 {
            writer.write(InternalValue::from_components(
                format!("key{i:06}").into_bytes(),
                b"v".to_vec(),
                0,
                ValueType::Value,
            ))?;
        }
        assert_eq!(writer.meta.data_block_count, 0);
        Ok(writer.output_size_hint())
    };
    let entry = hint("1", crate::config::LocatorPrecision::Entry)?;
    let block = hint("2", crate::config::LocatorPrecision::Block)?;
    // Slots up to 19 999 take 15 bits a key; count at least 14.
    assert!(
        entry >= block + 20_000 * 14 / 8,
        "entry-precise hint {entry} against block-precise {block}",
    );
    Ok(())
}

/// Only a key's newest version gets a locator entry, so blocks holding older
/// versions alone add no block id. Explicit widths that fit every recorded id
/// keep the locator however many such blocks follow.
#[test]
fn blocks_of_older_versions_do_not_outgrow_the_locator() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?.use_locator(
        crate::config::LocatorPolicyEntry::Enabled {
            precision: crate::config::LocatorPrecision::Block,
            block_id_bits: Some(1),
            slot_bits: None,
        },
    );
    // One key, its versions spread over four blocks.
    for block in 0..4_u64 {
        for version in 0..10 {
            writer.write(InternalValue::from_components(
                b"key".to_vec(),
                b"value---".to_vec(),
                100 - (block * 10 + version),
                ValueType::Value,
            ))?;
        }
        writer.spill_block()?;
    }
    assert_eq!(writer.meta.data_block_count, 4);
    assert_eq!(writer.locators.len(), 1, "the locator was dropped");
    Ok(())
}

/// Explicit locator widths too narrow for the table skip its section at
/// `finish`. The widths only grow with the table, so once they no longer fit
/// the writer holds nothing for the locator and charges nothing for it.
#[test]
fn a_locator_too_narrow_for_the_table_stops_holding_state() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut narrow = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(64)
        .use_locator(crate::config::LocatorPolicyEntry::Enabled {
            precision: crate::config::LocatorPrecision::Block,
            block_id_bits: Some(1),
            slot_bits: None,
        });
    let mut plain =
        Writer::new(dir.path().join("2"), 1, 0, Arc::new(StdFs))?.use_data_block_size(64);
    write_keys(&mut narrow, 5_000, 8)?;
    write_keys(&mut plain, 5_000, 8)?;
    assert!(
        narrow.locators.is_empty(),
        "{} triples held",
        narrow.locators.len()
    );
    assert_eq!(narrow.held_state_bytes(), plain.held_state_bytes());
    assert_eq!(narrow.output_size_hint(), plain.output_size_hint());
    Ok(())
}

/// A table of range tombstones alone whose caller gave no coverage takes its
/// key range from the tombstones in byte order: the least start and the
/// greatest end, whatever order they were written in.
#[test]
fn a_table_of_tombstones_alone_covers_them_without_a_given_range() -> crate::Result<()> {
    use crate::range_tombstone::RangeTombstone;

    let dir = tempfile::tempdir()?;
    let path = dir.path().join("1");
    let mut writer = Writer::new(path.clone(), 1, 0, Arc::new(StdFs))?;
    for (start, end) in [(b"c", b"d"), (b"a", b"b"), (b"b", b"z")] {
        writer.write_range_tombstone(RangeTombstone::new(
            UserKey::from(start as &[u8]),
            UserKey::from(end as &[u8]),
            3,
        ));
    }
    let Some((id, checksum)) = writer.finish()? else {
        panic!("a table of tombstones alone is written");
    };
    let table = crate::Table::recover(crate::table::RecoverParams::new(
        path,
        checksum,
        id,
        Arc::new(StdFs),
        crate::comparator::default_comparator(),
        Arc::new(crate::Cache::with_capacity_bytes(64 * 1_024)),
    ))?;
    assert_eq!(table.metadata.key_range.min().as_ref(), b"a");
    assert_eq!(table.metadata.key_range.max().as_ref(), b"z");
    assert_eq!(table.range_tombstones().len(), 3);
    Ok(())
}

/// A key whose versions run from a block appended as it is into rows written
/// after it keeps its own hash in the hash index of the block those rows form:
/// every version reads back at its own snapshot.
#[test]
fn a_key_running_from_a_verbatim_block_into_written_rows_keeps_its_hash() -> crate::Result<()> {
    use crate::coding::Decode;
    use crate::table::block::Header;

    const VERSIONS: u64 = 60;
    let key = b"the-one-key";
    let version = |seqno: u64| {
        InternalValue::from_components(
            key.to_vec(),
            format!("value-{seqno:04}").into_bytes(),
            seqno,
            ValueType::Value,
        )
    };
    let hashed = |path: crate::path::PathBuf| -> crate::Result<Writer> {
        Ok(Writer::new(path, 1, 0, Arc::new(StdFs))?
            .use_data_block_size(256)
            .use_data_block_hash_ratio(2.0))
    };
    let recover = |path: crate::path::PathBuf, id, checksum| {
        crate::Table::recover(crate::table::RecoverParams::new(
            path,
            checksum,
            id,
            Arc::new(StdFs),
            crate::comparator::default_comparator(),
            Arc::new(crate::Cache::with_capacity_bytes(1 << 20)),
        ))
    };

    let dir = tempfile::tempdir()?;
    let source = dir.path().join("source");
    let mut writer = hashed(source.clone())?;
    for seqno in (1..=VERSIONS).rev() {
        writer.write(version(seqno))?;
    }
    let Some((id, checksum)) = writer.finish()? else {
        panic!("the source table is written");
    };
    let table = recover(source.clone(), id, checksum)?;
    let Some(first) = table.data_block_handles().find_map(Result::ok) else {
        panic!("the source table has a data block");
    };
    let last_copied = first.seqno();
    assert!(
        last_copied > 1,
        "the key's versions run past the first block"
    );
    let offset = *first.as_ref().offset();

    let bytes = std::fs::read(&source)?;
    let Some(frame) = bytes.get(usize::try_from(offset).unwrap_or(usize::MAX)..) else {
        panic!("the first block lies within the file");
    };
    let header = Header::decode_from(&mut &frame[..])?;
    let Some(raw) = frame.get(..header.on_disk_size_with(None) as usize) else {
        panic!("the first block's frame lies within the file");
    };

    let dest = dir.path().join("dest");
    let mut writer = hashed(dest.clone())?;
    let copied: Vec<InternalValue> = (last_copied..=VERSIONS).rev().map(version).collect();
    writer.append_verbatim_data_block(
        (
            raw,
            VerbatimSource {
                table_id: id,
                offset,
            },
        ),
        header,
        Vec::new(),
        &copied,
        None,
        &crate::comparator::default_comparator(),
    )?;
    for seqno in (1..last_copied).rev() {
        writer.write(version(seqno))?;
    }
    let Some((id, checksum)) = writer.finish()? else {
        panic!("the copy is written");
    };

    let table = recover(dest, id, checksum)?;
    let hash = crate::hash::hash64(key);
    for seqno in 1..=VERSIONS {
        let got = table.get(key, seqno + 1, hash)?;
        assert_eq!(
            got.map(|v| v.key.seqno),
            Some(seqno),
            "the version at seqno {seqno} reads back",
        );
    }
    Ok(())
}

/// Every byte prefix of a key is a token.
struct AllPrefixes;

impl crate::prefix::PrefixExtractor for AllPrefixes {
    fn name(&self) -> &'static str {
        "all-prefixes"
    }

    fn prefixes<'a>(&self, key: &'a [u8]) -> Box<dyn Iterator<Item = &'a [u8]> + 'a> {
        Box::new((1..=key.len()).filter_map(|end| key.get(..end)))
    }
}

/// A prefix extractor can register thousands of hashes per key, so the
/// filter's state grows far faster than one hash a key: the estimates follow
/// that growth between the refreshes the key count sets.
#[test]
fn the_estimates_follow_a_filter_growing_by_many_hashes_a_key() -> crate::Result<()> {
    const KEYS: u8 = 20;
    const LEN: usize = 5_000;
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(1 << 20)
        .use_prefix_extractor(Some(Arc::new(AllPrefixes)));
    for i in 0..KEYS {
        // Keys differ in their first byte, so no prefix repeats.
        let mut key = alloc::vec![i];
        key.resize(LEN, 0x5a);
        writer.write(InternalValue::from_components(
            key,
            b"v".to_vec(),
            0,
            ValueType::Value,
        ))?;
    }
    let hashes = u64::from(KEYS) * LEN as u64 * core::mem::size_of::<u64>() as u64;
    assert!(
        writer.held_state_bytes() >= hashes,
        "held {} for {hashes} bytes of prefix hashes",
        writer.held_state_bytes(),
    );
    Ok(())
}

/// A block that outgrows explicit slot widths drops the locator only when it
/// is cut; until then its triples stay allocated and are held.
#[test]
fn a_forming_block_past_the_slot_width_still_holds_its_locators() -> crate::Result<()> {
    let dir = tempfile::tempdir()?;
    let mut writer = Writer::new(dir.path().join("1"), 1, 0, Arc::new(StdFs))?
        .use_data_block_size(1 << 20)
        .use_locator(crate::config::LocatorPolicyEntry::Enabled {
            precision: crate::config::LocatorPrecision::Entry,
            block_id_bits: None,
            slot_bits: Some(2),
        });
    let mut plain =
        Writer::new(dir.path().join("2"), 1, 0, Arc::new(StdFs))?.use_data_block_size(1 << 20);
    for i in 0..5_000_u32 {
        for w in [&mut writer, &mut plain] {
            w.write(InternalValue::from_components(
                format!("key{i:010}").into_bytes(),
                b"value---".to_vec(),
                0,
                ValueType::Value,
            ))?;
        }
    }
    writer.refresh_state_estimates();
    plain.refresh_state_estimates();
    assert_eq!(writer.meta.data_block_count, 0, "the block was cut");
    let triples = (writer.locators.capacity() * core::mem::size_of::<(u64, u64, u64)>()) as u64;
    assert!(triples > 0, "the locator stopped collecting");
    assert!(
        writer.held_state_bytes() >= plain.held_state_bytes() + triples,
        "held {} against {} without a locator and {triples} bytes of triples",
        writer.held_state_bytes(),
        plain.held_state_bytes(),
    );
    Ok(())
}