gnitz-zset 0.1.1

use super::*;
use crate::repr::{BatchBuilder, MappedShard};
use crate::schema::{SchemaDescriptor, TypeCode};
use crate::test_support::{make_batch, make_schema_pk_u64_payload_string, make_schema_u64_i64, make_string_batch};

/// Any change to the written bytes, the writer's own or a dependency's, needs a
/// `SHARD_EPOCH` bump.
#[test]
fn shard_bytes_are_pinned() {
    const PINNED: (u64, u64) = (27, 8259605191670743709);
    let int = SchemaColumn::new(TypeCode::I64, false);
    let schema = SchemaDescriptor::new(
        &[
            SchemaColumn::new(TypeCode::U64, false),
            int,
            int,
            SchemaColumn::new(TypeCode::String, false),
            SchemaColumn::new(TypeCode::String, false),
            SchemaColumn::new(TypeCode::I64, true),
            int,
        ],
        &[0],
    );
    let mut b = BatchBuilder::new(&schema);
    for i in 0..64i64 {
        b.begin_row((i * 3 + 1) as u128, 1 + i % 2);
        // A narrow range packs as FoR, a full-range one stays Raw.
        b.put_int((1_000_000 + i * 3) as u128);
        b.put_int((i.wrapping_mul(0x0123_4567_89AB_CDEF) ^ (i << 60)) as u128);
        // Five values, inline and on the heap, make a dictionary.
        b.put_string(&"shard".repeat(1 + (i % 5) as usize));
        // No value twice, inline and on the heap, is its lengths.
        b.put_string(&format!("{i:0w$}", w = 6 + i as usize % 12));
        // One row in eight holds a value: the others' cells are left out.
        b.put_opt_int((i % 8 == 0).then_some((1 << 40) + i as u128));
        b.put_int(7);
        b.end_row();
    }
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("golden.db");
    b.finish()
        .write_as_shard(path.to_str().unwrap(), ShardWriteOpts::default())
        .unwrap();
    let mut bytes = std::fs::read(&path).unwrap();

    let spans = spans_of(&bytes);
    let encodings: Vec<Encoding> = spans.iter().map(|s| s.encoding).collect();
    use Encoding::*;
    assert_eq!(
        encodings,
        [Raw, TwoValue, TwoValue, For, Raw, Dict, Seq, Sparse, Constant, Raw, Raw],
        "the batch covers every encoding"
    );
    assert!(spans.last().unwrap().size > 0, "and a PK filter");

    // Both vary with things other than the writer: the system schema, the path.
    write_u64_le(&mut bytes, OFF_VERSION, 0);
    write_u64_le(&mut bytes, OFF_DESC_CHECKSUM, 0);
    assert_eq!(
        (SHARD_EPOCH, gnitz_wire::checksum(&bytes)),
        PINNED,
        "shard bytes changed: bump SHARD_EPOCH and re-pin"
    );
}

/// A replicated relayout hard-links one shard inode into several stores, so a
/// write at an existing name must leave that file untouched.
#[test]
fn write_refuses_an_existing_path_and_leaves_it_intact() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("taken.db");
    let path = path.to_str().unwrap();
    let schema = make_schema_u64_i64();
    make_batch(&schema, &[(1, 1, 10)])
        .write_as_shard(path, ShardWriteOpts::default())
        .unwrap();
    let before = std::fs::read(path).unwrap();

    let other = make_batch(&schema, &[(2, 1, 20)]);
    assert_eq!(
        other.write_as_shard(path, ShardWriteOpts::default()),
        Err(StorageError::Io(libc::EEXIST))
    );
    assert_eq!(std::fs::read(path).unwrap(), before);
}

/// A shard carries no dead heap bytes: a run whose fold left dead spans behind
/// writes only the spans its rows reference, and every row reads back.
#[test]
fn a_shard_drops_its_runs_dead_heap() {
    use crate::test_support::{make_schema_pk_u64_payload_string, make_string_batch, map_shard, read_strings};
    use gnitz_wire::RowSource;
    let (x, y) = ([b'x'; 20], [b'y'; 20]);
    let a = make_string_batch(&[(1, 1, &x), (2, 1, &y)]);
    let run = a.merged_consolidated(&make_string_batch(&[(1, -1, &x)]), &make_schema_pk_u64_payload_string());
    assert!(run.dead_heap > 0, "premise: the fold left dead bytes");

    let dir = tempfile::tempdir().unwrap();
    let shard = map_shard(&dir.path().join("s.db"), &run);
    assert_eq!(shard.blob().len(), y.len(), "only the referenced span reaches disk");
    assert_eq!(read_strings(&shard.slice_to_owned_batch(0, 1)), [y.to_vec()]);
}

/// The heap of the shard `rows` are written as, and its string region's encoding.
fn written_strings(rows: &[(u64, i64, &[u8])]) -> (usize, Encoding) {
    use gnitz_wire::RowSource;
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("s.db");
    let batch = make_string_batch(rows);
    batch
        .write_as_shard(path.to_str().unwrap(), ShardWriteOpts::default())
        .unwrap();
    let image = std::fs::read(&path).unwrap();
    let shard = MappedShard::open(path.to_str().unwrap(), batch.schema()).unwrap();
    for (row, &(_, _, want)) in rows.iter().enumerate() {
        assert_eq!(gnitz_wire::payload_bytes(&shard, row, 0), want, "row {row}");
    }
    (shard.blob().len(), spans_of(&image)[REG_PAYLOAD_START].encoding)
}

/// A long value several rows hold reaches the heap once, whichever image the
/// column takes.
#[test]
fn a_repeated_long_string_reaches_the_heap_once() {
    let (x, y) = ([b'x'; 20], [b'y'; 33]);
    // Three rows: a dictionary would not be the smaller image.
    let few: Vec<(u64, i64, &[u8])> = vec![(1, 1, &x), (2, 1, &y), (3, 1, &x)];
    assert_eq!(written_strings(&few), (x.len() + y.len(), Encoding::Raw));
    let many: Vec<(u64, i64, &[u8])> = (1..=40)
        .map(|pk| (pk, 1, if pk % 3 == 0 { &y[..] } else { &x[..] }))
        .collect();
    assert_eq!(written_strings(&many), (x.len() + y.len(), Encoding::Dict));
    let one: Vec<(u64, i64, &[u8])> = (1..=40).map(|pk| (pk, 1, &y[..])).collect();
    assert_eq!(written_strings(&one), (y.len(), Encoding::Constant));
}

/// A packed heap holds no dead bytes either: a run whose fold left dead spans
/// behind writes the values its rows still hold.
#[test]
fn a_packed_shard_drops_its_runs_dead_heap() {
    use crate::test_support::map_shard;
    use gnitz_wire::RowSource;
    let (x, y) = ([b'x'; 20], [b'y'; 20]);
    let a = make_string_batch(&[(1, 1, &x), (2, 1, &y), (3, 1, &y), (4, 1, &y)]);
    let run = a.merged_consolidated(&make_string_batch(&[(1, -1, &x)]), &make_schema_pk_u64_payload_string());
    assert!(run.dead_heap > 0, "premise: the fold left dead bytes");

    let dir = tempfile::tempdir().unwrap();
    let shard = map_shard(&dir.path().join("s.db"), &run);
    assert_eq!(shard.blob().len(), y.len());
    assert_eq!(shard.row_count(), 3);
}

/// `2 * values` cells holding each of `values` strings of `width` bytes twice,
/// and their heap.
fn each_value_twice(values: usize, width: usize) -> (Vec<[u8; 16]>, Vec<u8>) {
    let mut heap = Vec::new();
    let cells = (0..2 * values)
        .map(|i| gnitz_wire::encode_german_string(format!("{:0width$}", i % values).as_bytes(), &mut heap))
        .collect();
    (cells, heap)
}

/// Past a dictionary's worth of values a column whose cells cost less than a
/// second copy of each value keeps a cell per row, over a heap that holds each
/// value once.
#[test]
fn more_values_than_a_dictionary_holds_stay_raw_over_a_shared_heap() {
    let values = DICT_MAX_ENTRIES + 1;
    let (cells, src_heap) = each_value_twice(values, 40);
    let mut heap = Vec::new();
    let (encoding, image) = pack_string_column(&cells, &src_heap, &mut heap);
    assert_eq!(encoding, Encoding::Raw);
    assert_eq!(heap.len() * 2, src_heap.len(), "each value once");
    let packed = image.as_chunks::<16>().0;
    assert_eq!(packed.len(), cells.len());
    for (i, (cell, src)) in packed.iter().zip(&cells).enumerate() {
        assert_eq!(
            gnitz_wire::german_string_content(cell, &heap),
            gnitz_wire::german_string_content(src, &src_heap),
            "row {i}"
        );
    }
    assert_eq!(packed[..values], packed[values..], "a value's rows share its cell");
}

/// The same column of values a cell outweighs is its lengths, every row's value
/// on the heap.
#[test]
fn more_short_values_than_a_dictionary_holds_are_their_lengths() {
    let (cells, src_heap) = each_value_twice(DICT_MAX_ENTRIES + 1, 20);
    let mut heap = vec![0u8; 5];
    let (encoding, image) = pack_string_column(&cells, &src_heap, &mut heap);
    assert_eq!(encoding, Encoding::Seq);
    assert_eq!(
        heap.len() - 5,
        src_heap.len(),
        "every row's value, and nothing a refused image left"
    );
    let mut packed = vec![0u8; cells.len() * 16];
    SeqImage::parse(&image, cells.len())
        .unwrap()
        .decode(&heap, 0, &mut packed);
    for (i, (cell, src)) in packed.as_chunks::<16>().0.iter().zip(&cells).enumerate() {
        assert_eq!(
            gnitz_wire::german_string_content(cell, &heap),
            gnitz_wire::german_string_content(src, &src_heap),
            "row {i}"
        );
    }
}

/// A column of values that never repeat is its lengths once those are the
/// smaller region.
#[test]
fn distinct_strings_are_their_lengths() {
    let distinct = |n: u64| -> Vec<(u64, i64, Vec<u8>)> {
        (1..=n)
            .map(|pk| (pk, 1, format!("{pk:0w$}", w = 8 + pk as usize % 9).into_bytes()))
            .collect()
    };
    let written = |rows: &[(u64, i64, Vec<u8>)]| {
        let rows: Vec<(u64, i64, &[u8])> = rows.iter().map(|(pk, w, s)| (*pk, *w, &s[..])).collect();
        written_strings(&rows)
    };
    let long = |rows: &[(u64, i64, Vec<u8>)]| rows.iter().map(|r| r.2.len()).filter(|&len| len > 12).sum::<usize>();
    let (two, forty) = (distinct(2), distinct(40));
    assert_eq!(written(&two), (long(&two), Encoding::Raw));
    assert_eq!(written(&forty), (long(&forty), Encoding::Seq));
    assert_eq!(written(&distinct(1)).1, Encoding::Constant);
}

/// The sample finds a value repeated across the column and one repeated only
/// next to itself, and nothing in a column of distinct values.
#[test]
fn the_sample_sees_spread_and_adjacent_repeats() {
    let n = 4 * SAMPLE_RUN * SAMPLE_RUNS;
    let column = |value: &dyn Fn(usize) -> usize| {
        let mut heap = Vec::new();
        let cells: Vec<[u8; 16]> = (0..n)
            .map(|i| gnitz_wire::encode_german_string(format!("value-{:012}", value(i)).as_bytes(), &mut heap))
            .collect();
        sample_repeats(n, |row| gnitz_wire::german_string_content(&cells[row], &heap))
    };
    assert!(!column(&|i| i), "distinct");
    assert!(column(&|i| i % 1000), "a thousand values, spread");
    assert!(column(&|i| i / 2), "each value in two adjacent rows");
}