quatzal-storage 0.1.0

// SPDX-License-Identifier: Apache-2.0
//! UACE-FR-1.9 / UACE-FR-1.10: batched row writes and reads. `UACE-NFR-33` measured the
//! product surface at ~540 rows/sec and traced it to `Engine::put`'s one-fsync-per-row --
//! the group-commit deferral `CLAUDE.md` has carried since Phase 1, promoted to the
//! binding constraint once the vector tier stopped being the bottleneck. Real (tempdir)
//! I/O, per this crate's mock-boundary rule: durability claims are only meaningful
//! against the real stack.

use tempfile::TempDir;
use quatzal_schema::{Row, Value};
use quatzal_storage::Engine;

fn row(i: u32) -> Row {
    Row::new(format!("k-{i}").into_bytes())
        .with_scalar("n", Value::I64(i as i64))
        .with_vector(vec![i as f32, 0.5])
}

/// `UACE-FR-1.9`: a batch lands atomically-per-shard and every row is durable -- one WAL
/// append plus one fsync covers the whole batch, and reopening the engine finds all of it.
#[test]
fn put_batch_is_durable_across_reopen() {
    let tmp = TempDir::new().unwrap();
    {
        let engine = Engine::open(tmp.path(), Some(4)).unwrap();
        engine.put_batch((0..500).map(row).collect()).unwrap();
        // Readable immediately.
        let fetched = engine.get(b"k-250").unwrap().expect("present");
        assert_eq!(fetched.scalars["n"], Value::I64(250));
    }
    // Reopen: WAL replay must reconstruct everything the batch acknowledged.
    let engine = Engine::open(tmp.path(), Some(4)).unwrap();
    assert_eq!(
        engine.scan().unwrap().len(),
        500,
        "every batched row survives a reopen"
    );
    let fetched = engine.get(b"k-499").unwrap().expect("present");
    assert_eq!(fetched.vector, Some(vec![499.0, 0.5]));
}

/// `UACE-FR-1.9`: batched and single-row writes interleave correctly -- later writes to
/// the same key win regardless of which path wrote them.
#[test]
fn put_batch_and_put_interleave_with_newest_wins() {
    let tmp = TempDir::new().unwrap();
    let engine = Engine::open(tmp.path(), Some(2)).unwrap();

    engine.put_batch((0..10).map(row).collect()).unwrap();
    engine
        .put(Row::new(b"k-5".to_vec()).with_scalar("n", Value::I64(555)))
        .unwrap();
    assert_eq!(
        engine.get(b"k-5").unwrap().unwrap().scalars["n"],
        Value::I64(555)
    );

    engine
        .put_batch(vec![
            Row::new(b"k-5".to_vec()).with_scalar("n", Value::I64(999)),
        ])
        .unwrap();
    assert_eq!(
        engine.get(b"k-5").unwrap().unwrap().scalars["n"],
        Value::I64(999)
    );

    let empty: Vec<Row> = Vec::new();
    engine
        .put_batch(empty)
        .expect("an empty batch is a no-op, not an error");
}

/// `UACE-FR-1.10`: `get_batch` returns rows positionally, `None` for absent keys, in one
/// round trip per shard rather than one per key -- the cost `UACE-NFR-33` identified in
/// provenance lookups (k sequential channel round-trips per search).
#[test]
fn get_batch_returns_rows_positionally() {
    let tmp = TempDir::new().unwrap();
    let engine = Engine::open(tmp.path(), Some(4)).unwrap();
    engine.put_batch((0..50).map(row).collect()).unwrap();
    engine.delete(b"k-7").unwrap();

    let keys: Vec<Vec<u8>> = vec![
        b"k-3".to_vec(),
        b"missing".to_vec(),
        b"k-49".to_vec(),
        b"k-7".to_vec(),
        b"k-0".to_vec(),
    ];
    let got = engine.get_batch(&keys).unwrap();
    assert_eq!(
        got.len(),
        keys.len(),
        "one slot per requested key, in order"
    );
    assert_eq!(got[0].as_ref().unwrap().scalars["n"], Value::I64(3));
    assert!(got[1].is_none(), "absent key -> None");
    assert_eq!(got[2].as_ref().unwrap().scalars["n"], Value::I64(49));
    assert!(got[3].is_none(), "deleted key -> None");
    assert_eq!(got[4].as_ref().unwrap().scalars["n"], Value::I64(0));

    assert!(engine.get_batch(&[]).unwrap().is_empty());
}

/// `UACE-FR-1.10`: `get_batch` agrees with `get` for every key -- one code path's answer
/// can never diverge from the other's.
#[test]
fn get_batch_agrees_with_point_reads() {
    let tmp = TempDir::new().unwrap();
    let engine = Engine::open(tmp.path(), Some(3)).unwrap();
    engine.put_batch((0..100).map(row).collect()).unwrap();

    let keys: Vec<Vec<u8>> = (0..100).map(|i| format!("k-{i}").into_bytes()).collect();
    let batched = engine.get_batch(&keys).unwrap();
    for (i, key) in keys.iter().enumerate() {
        let single = engine.get(key).unwrap();
        assert_eq!(
            single.map(|r| r.scalars["n"].clone()),
            batched[i].as_ref().map(|r| r.scalars["n"].clone()),
            "get_batch and get disagree on {}",
            String::from_utf8_lossy(key)
        );
    }
}