coordinode-lsm-tree 7.4.2

use super::{
    COL_SEQNO, COL_USER_KEY, COL_VALUE, COL_VALUE_TYPE, Column, ColumnBatch, RowPageColumns,
    TypeTag, column_batch_into_entries, column_batch_to_entries, entries_to_column_batch,
    frame_value_cells, frame_value_cells_nullable, page_match_entries, unframe_value_cells,
    unframe_value_cells_nullable, unframe_value_cells_with_defaults,
    validate_columnar_ingest_batch,
};
use crate::{Slice, ValueType, config::ColumnEncoding, key::InternalKey, value::InternalValue};

#[test]
fn fixed_only_value_framing_has_no_overhead_and_round_trips() {
    // Fixed-width cells are stored verbatim (width recoverable from the tag),
    // so the framed blob is the bare concatenation: zero per-cell overhead.
    let tags = [TypeTag::Fixed(4), TypeTag::Fixed(4)];
    let blob = frame_value_cells(&[
        (TypeTag::Fixed(4), &[1, 0, 0, 0][..]),
        (TypeTag::Fixed(4), &[2, 0, 0, 0][..]),
    ])
    .expect("frame");
    assert_eq!(blob, vec![1, 0, 0, 0, 2, 0, 0, 0], "no length prefixes");
    let cells = unframe_value_cells(&blob, &tags).expect("unframe");
    assert_eq!(cells, vec![&[1, 0, 0, 0][..], &[2, 0, 0, 0][..]]);
}

#[test]
fn mixed_value_framing_round_trips() {
    let tags = [TypeTag::Bytes, TypeTag::Fixed(1), TypeTag::Bytes];
    let blob = frame_value_cells(&[
        (TypeTag::Bytes, b"abc"),
        (TypeTag::Fixed(1), &[7][..]),
        (TypeTag::Bytes, b""),
    ])
    .expect("frame");
    let cells = unframe_value_cells(&blob, &tags).expect("unframe");
    assert_eq!(cells, vec![&b"abc"[..], &[7][..], &b""[..]]);
}

#[test]
fn empty_value_framing_round_trips() {
    let blob = frame_value_cells(&[]).expect("frame");
    assert_eq!(blob, Vec::<u8>::new());
    assert_eq!(
        unframe_value_cells(&blob, &[]).expect("unframe"),
        Vec::<&[u8]>::new()
    );
}

#[test]
fn unframe_rejects_truncated_blob() {
    // A fixed(4) tag over a 2-byte blob is truncated.
    assert!(unframe_value_cells(&[1, 2], &[TypeTag::Fixed(4)]).is_err());
    // A bytes tag whose declared length runs past the end.
    let mut blob = 8u32.to_le_bytes().to_vec();
    blob.extend_from_slice(b"abc"); // 3 bytes present, length claims 8
    assert!(unframe_value_cells(&blob, &[TypeTag::Bytes]).is_err());
}

#[test]
fn unframe_rejects_trailing_bytes() {
    // One fixed(2) cell, but the blob carries an extra byte the tags do not
    // cover: the blob and the tag list disagree.
    assert!(unframe_value_cells(&[1, 2, 3], &[TypeTag::Fixed(2)]).is_err());
}

#[test]
fn frame_rejects_fixed_cell_length_mismatch() {
    // A fixed(4) tag with a 3-byte cell would misframe (and shift later cells),
    // so framing rejects it.
    assert!(frame_value_cells(&[(TypeTag::Fixed(4), &[1, 2, 3][..])]).is_err());
}

#[test]
fn nullable_framing_round_trips_mixed_present_and_null() {
    let tags = [TypeTag::Fixed(4), TypeTag::Bytes, TypeTag::Fixed(2)];
    let blob = frame_value_cells_nullable(&[
        (TypeTag::Fixed(4), Some(&[1, 0, 0, 0][..])),
        (TypeTag::Bytes, None),
        (TypeTag::Fixed(2), Some(&[9, 9][..])),
    ])
    .expect("frame");
    // Presence bitmap (1 byte): bits 0 and 2 set, bit 1 clear -> 0b101 = 5.
    assert_eq!(blob[0], 0b0000_0101, "presence bitmap marks the null cell");
    assert_eq!(
        unframe_value_cells_nullable(&blob, &tags).expect("unframe"),
        vec![Some(&[1, 0, 0, 0][..]), None, Some(&[9, 9][..])],
    );
}

#[test]
fn nullable_framing_all_null_and_all_present() {
    let tags = [TypeTag::Bytes, TypeTag::Fixed(1)];
    let all_null = frame_value_cells_nullable(&[(TypeTag::Bytes, None), (TypeTag::Fixed(1), None)])
        .expect("frame");
    assert_eq!(all_null, vec![0u8], "bitmap only, no bodies");
    assert_eq!(
        unframe_value_cells_nullable(&all_null, &tags).expect("unframe"),
        vec![None, None],
    );

    let all_present = frame_value_cells_nullable(&[
        (TypeTag::Bytes, Some(&b"ab"[..])),
        (TypeTag::Fixed(1), Some(&[7][..])),
    ])
    .expect("frame");
    assert_eq!(
        unframe_value_cells_nullable(&all_present, &tags).expect("unframe"),
        vec![Some(&b"ab"[..]), Some(&[7][..])],
    );
}

#[test]
fn nullable_framing_rejects_truncated_and_trailing() {
    // A present fixed(4) cell whose bytes are missing after the bitmap.
    let truncated = alloc::vec![0b0000_0001u8, 1, 2]; // bit 0 set, only 2 of 4 bytes
    assert!(unframe_value_cells_nullable(&truncated, &[TypeTag::Fixed(4)]).is_err());
    // A valid all-null encoding with an extra trailing byte.
    let mut trailing = frame_value_cells_nullable(&[(TypeTag::Bytes, None)]).expect("frame");
    trailing.push(0);
    assert!(unframe_value_cells_nullable(&trailing, &[TypeTag::Bytes]).is_err());
}

#[test]
fn nullable_framing_rejects_a_fixed_cell_length_mismatch() {
    // A Fixed(4) tag with a 3-byte present cell would misframe (and shift
    // later cells), so the nullable framing rejects it like the non-nullable
    // path does.
    assert!(
        frame_value_cells_nullable(&[(TypeTag::Fixed(4), Some(&[1, 2, 3][..]))]).is_err(),
        "a present fixed cell whose length differs from its tag width must be rejected",
    );
}

#[test]
fn unframe_with_defaults_substitutes_for_null_cells() {
    let blob = frame_value_cells_nullable(&[
        (TypeTag::Fixed(2), Some(&[7, 7][..])),
        (TypeTag::Fixed(2), None),
    ])
    .expect("frame");
    let cells = unframe_value_cells_with_defaults(
        &blob,
        &[
            (TypeTag::Fixed(2), &[7, 7][..]),
            (TypeTag::Fixed(2), &[0, 0][..]),
        ],
    )
    .expect("unframe with defaults");
    assert_eq!(cells, vec![&[7, 7][..], &[0, 0][..]]);
}

#[test]
fn first_user_key_rejects_a_non_key_first_column() {
    // A first column that is not the non-null bytes user-key column is
    // rejected before any low-level offset read.
    let fixed_first = ColumnBatch {
        row_count: 1,
        columns: alloc::vec![Column {
            column_id: COL_USER_KEY,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: alloc::vec![0, 0, 0, 0].into(),
        }],
    };
    assert!(fixed_first.first_user_key().is_err());

    let missing = ColumnBatch {
        row_count: 1,
        columns: alloc::vec![Column {
            column_id: 99,
            type_tag: TypeTag::Bytes,
            validity: None,
            data: alloc::vec![0, 0, 0, 0, 0, 0, 0, 0].into(),
        }],
    };
    assert!(missing.first_user_key().is_err());
}

#[test]
fn first_user_key_returns_the_first_row_key() {
    // A two-row key column: the first key is read without decoding the batch.
    let mut data = Vec::new();
    for off in [0u32, 2, 5] {
        data.extend_from_slice(&off.to_le_bytes());
    }
    data.extend_from_slice(b"k0k11");
    let batch = ColumnBatch {
        row_count: 2,
        columns: alloc::vec![Column {
            column_id: COL_USER_KEY,
            type_tag: TypeTag::Bytes,
            validity: None,
            data: data.into(),
        }],
    };
    assert_eq!(batch.first_user_key().expect("first key"), Some(&b"k0"[..]));
}

fn entry(user_key: &[u8], seqno: u64, value_type: ValueType, value: &[u8]) -> InternalValue {
    InternalValue {
        key: InternalKey {
            user_key: Slice::from(user_key),
            seqno,
            value_type,
        },
        value: Slice::from(value),
    }
}

fn assert_entries_eq(a: &[InternalValue], b: &[InternalValue]) {
    assert_eq!(a.len(), b.len(), "entry count mismatch");
    for (x, y) in a.iter().zip(b) {
        assert_eq!(x.key.user_key.as_ref(), y.key.user_key.as_ref());
        assert_eq!(x.key.seqno, y.key.seqno);
        assert!(x.key.value_type == y.key.value_type, "value_type mismatch");
        assert_eq!(x.value.as_ref(), y.value.as_ref());
    }
}

/// The expression a batch's one column was encoded as: the column header is
/// `row_count` (4) + `column_count` (4) + id (2) + type (1) + width (1) +
/// validity flag (1), then the values' length and the values.
fn only_column_expression(batch: &ColumnBatch) -> super::Expression {
    let encoded = batch.encode().expect("encode");
    let column = &batch.columns[0];
    let mut rest = &encoded[13..];
    let len =
        usize::try_from(crate::table::column_page::take_varint(&mut rest, 5).expect("length"))
            .expect("a length a page holds");
    super::Values::parse(column.type_tag, batch.row_count, &rest[..len])
        .expect("parse")
        .describe()
        .expect("describe")
}

#[test]
fn a_seqno_column_is_encoded_by_its_ordinals_and_round_trips() {
    // A u64 number column of nearby values packs into a few bits a row by its
    // ordinals, a repeat and a decrease included, and decodes exactly.
    let seqnos: [u64; 5] = [100, 105, 105, 200, 199];
    let data: Vec<u8> = seqnos.iter().flat_map(|s| s.to_le_bytes()).collect();
    let batch = ColumnBatch {
        row_count: 5,
        columns: vec![Column {
            column_id: COL_SEQNO,
            type_tag: TypeTag::Number(crate::table::columnar::Number::U64_LE),
            validity: None,
            data: data.clone().into(),
        }],
    };
    assert!(
        matches!(
            only_column_expression(&batch),
            super::Expression::Ordinals(_)
        ),
        "a narrow range of u64 numbers is encoded by its ordinals",
    );
    let decoded = ColumnBatch::decode(&batch.encode().expect("encode").into()).expect("decode");
    assert_eq!(decoded.columns[0].data, data, "the column must round-trip");
}

#[test]
fn an_opaque_fixed_column_is_never_encoded_by_ordinals() {
    // An opaque fixed column has no number type, so it has no ordinals: its
    // bytes are not integers the engine may reinterpret, however narrow they
    // look. Distinct rows leave it plain.
    let data: Vec<u8> = (0u64..8).flat_map(|i| (i * 3).to_le_bytes()).collect();
    let batch = ColumnBatch {
        row_count: 8,
        columns: vec![Column {
            column_id: 99,
            type_tag: TypeTag::Fixed(8),
            validity: None,
            data: data.clone().into(),
        }],
    };
    assert_eq!(only_column_expression(&batch), super::Expression::Plain);
    let decoded = ColumnBatch::decode(&batch.encode().expect("encode").into()).expect("decode");
    assert_eq!(decoded.columns[0].data, data);
}

#[test]
fn from_column_batch_rejects_a_zero_row_batch() {
    // A zero-row columnar group is corrupt (the writer never spills empty);
    // reconstructing it must error, not panic in the row encoder.
    let empty = entries_to_column_batch(&[]).expect("transpose");
    assert!(crate::table::data_block::DataBlock::from_column_batch([empty], 16, &mut 0).is_err());
}

/// A group with no row pages at all is refused just the same: there is no
/// row to rebuild a block from.
#[test]
fn from_column_batch_rejects_a_group_without_row_pages() {
    assert!(crate::table::data_block::DataBlock::from_column_batch([], 16, &mut 0).is_err());
}

#[test]
fn decode_projected_decodes_only_the_wanted_columns() {
    // Project just the user-key column: the result carries ONLY that column,
    // proving the value / seqno / value-type columns were never decoded.
    let entries = vec![
        entry(b"alpha", 10, ValueType::Value, b"v1"),
        entry(b"bravo", 9, ValueType::Value, b"v2"),
    ];
    let bytes = entries_to_column_batch(&entries)
        .expect("transpose")
        .encode()
        .expect("encode");

    let projected = ColumnBatch::decode_projected(&bytes.clone().into(), &[COL_USER_KEY])
        .expect("decode_projected");
    assert_eq!(projected.row_count, 2);
    assert_eq!(
        projected.columns.len(),
        1,
        "only the projected column is decoded"
    );
    assert_eq!(projected.columns[0].column_id, COL_USER_KEY);
    assert!(
        !projected.columns.iter().any(|c| c.column_id == COL_VALUE),
        "a key-only projection must not decode the value column"
    );

    // Projecting every column equals a full decode.
    let all = [COL_USER_KEY, COL_SEQNO, COL_VALUE_TYPE, COL_VALUE];
    assert_eq!(
        ColumnBatch::decode_projected(&bytes.clone().into(), &all).expect("decode_projected all"),
        ColumnBatch::decode(&bytes.into()).expect("decode"),
    );
}

/// Whether `col`'s bytes live inside `block`'s allocation (a zero-copy view)
/// rather than in a detached copy.
fn is_view_of(col: &Column, block: &Slice) -> bool {
    let start = block.as_ptr() as usize;
    let end = start + block.len();
    let p = col.data.as_ptr() as usize;
    p >= start && p < end
}

#[test]
fn decode_projected_detaches_a_narrow_projection_from_a_value_heavy_block() {
    // Keys are a few bytes, values are large: a key-only projection covers a
    // sliver of the block. Served as a view it would keep the whole block
    // (values included) alive per batch, so it must come back detached; a
    // full decode covers the block and keeps the zero-copy view. The two
    // values differ, so the value column is stored in its own layout rather
    // than as a constant, and is served as a view.
    let entries = vec![
        entry(b"a", 2, ValueType::Value, &[0xABu8; 4096]),
        entry(b"b", 1, ValueType::Value, &[0xCDu8; 4096]),
    ];
    let block: Slice = entries_to_column_batch(&entries)
        .expect("transpose")
        .encode()
        .expect("encode")
        .into();

    let narrow =
        ColumnBatch::decode_projected(&block, &[COL_USER_KEY]).expect("key-only projection");
    assert!(
        !is_view_of(&narrow.columns[0], &block),
        "a key-only projection of a value-heavy block must not pin the block",
    );

    let full = ColumnBatch::decode(&block).expect("full decode");
    let value_col = full
        .columns
        .iter()
        .find(|c| c.column_id == COL_VALUE)
        .expect("value column");
    assert!(
        is_view_of(value_col, &block),
        "a full decode covers the block and stays zero-copy",
    );
}

#[test]
fn intrinsic_transpose_round_trips_entries() {
    // A mix of value kinds, including a tombstone (empty value) and a merge
    // operand, exercises the value column's variable width and the
    // value-type column.
    let entries = vec![
        entry(b"alpha", 10, ValueType::Value, b"v1"),
        entry(b"bravo", 9, ValueType::Tombstone, b""),
        entry(b"charlie", 8, ValueType::MergeOperand, b"+1"),
    ];
    let batch = entries_to_column_batch(&entries).expect("transpose");
    assert_eq!(batch.row_count, 3);
    assert_eq!(batch.columns.len(), 4, "four intrinsic columns");

    // Direct reconstruction.
    let back = column_batch_to_entries(&batch).expect("untranspose");
    assert_entries_eq(&entries, &back);

    // And through the block encode / decode, so the transpose composes with
    // the on-disk columnar format.
    let bytes = batch.encode().expect("encode");
    let decoded = ColumnBatch::decode(&bytes.into()).expect("decode");
    let back2 = column_batch_to_entries(&decoded).expect("untranspose decoded");
    assert_entries_eq(&entries, &back2);
}

/// A batch of the intrinsic columns and one caller sub-column has as many
/// columns as a whole-value batch, yet its rows cannot be laid out again as
/// it lays them out: rebuilt as whole values, the sub-column's id, type and
/// validity would be lost.
#[test]
fn a_batch_of_one_caller_sub_column_does_not_round_trip() {
    let entries = vec![entry(b"alpha", 10, ValueType::Value, b"v1")];
    let mut batch = entries_to_column_batch(&entries).expect("transpose");
    batch.columns[3] = Column {
        column_id: 3,
        type_tag: TypeTag::Number(crate::table::column_type::Number::U64_LE),
        validity: None,
        data: Slice::from(7u64.to_le_bytes().to_vec()),
    };
    let rows = column_batch_to_entries(&batch).expect("untranspose");
    assert!(!super::rows_round_trip(&batch));
    assert!(
        super::transpose_like(&batch, &rows)
            .expect("transpose")
            .is_none()
    );
    // The whole-value layout itself still round-trips.
    let whole = entries_to_column_batch(&entries).expect("transpose");
    assert!(super::rows_round_trip(&whole));
}

#[test]
fn intrinsic_untranspose_rejects_wrong_layout() {
    // A batch whose columns are not the four intrinsic columns is refused.
    let bad = ColumnBatch {
        row_count: 1,
        columns: vec![Column {
            column_id: 0,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: vec![0; 4].into(),
        }],
    };
    assert!(column_batch_to_entries(&bad).is_err());
}

#[test]
fn intrinsic_untranspose_rejects_nullable_intrinsic() {
    // Intrinsic fields are never null; a validity bitmap on one marks a
    // malformed batch even if it is otherwise well-framed.
    let mut batch =
        entries_to_column_batch(&[entry(b"k", 1, ValueType::Value, b"v")]).expect("transpose");
    batch.columns[0].validity = Some(vec![0b1]); // one valid row, but nullable
    assert!(column_batch_to_entries(&batch).is_err());
}

#[test]
fn intrinsic_untranspose_rejects_empty_key() {
    // An empty user key violates the engine's key invariant.
    let batch =
        entries_to_column_batch(&[entry(b"", 1, ValueType::Value, b"v")]).expect("transpose");
    assert!(column_batch_to_entries(&batch).is_err());
}

#[test]
fn intrinsic_untranspose_rejects_huge_row_count() {
    // A hand-built batch claiming billions of rows but carrying tiny columns
    // is rejected by per-column validation before any allocation.
    let bad = ColumnBatch {
        row_count: u32::MAX,
        columns: vec![
            Column {
                column_id: 0,
                type_tag: TypeTag::Bytes,
                validity: None,
                data: 0u32.to_le_bytes().to_vec().into(),
            },
            Column {
                column_id: 1,
                type_tag: TypeTag::Fixed(8),
                validity: None,
                data: Vec::new().into(),
            },
            Column {
                column_id: 2,
                type_tag: TypeTag::Fixed(1),
                validity: None,
                data: Vec::new().into(),
            },
            Column {
                column_id: 3,
                type_tag: TypeTag::Bytes,
                validity: None,
                data: 0u32.to_le_bytes().to_vec().into(),
            },
        ],
    };
    assert!(column_batch_to_entries(&bad).is_err());
}

#[test]
fn untranspose_frames_multiple_value_subcolumns() {
    // A consumer batch: the three intrinsic columns plus two value
    // sub-columns (a fixed-4 and a variable-width bytes). Each row's
    // reconstructed value is the framed concat of its two sub-cells, which
    // unframe_value_cells reverses.
    let mut batch = entries_to_column_batch(&[
        entry(b"k0", 1, ValueType::Value, b"ignored"),
        entry(b"k1", 2, ValueType::Value, b"ignored"),
    ])
    .expect("transpose");
    // Replace the single opaque value column with two value sub-columns.
    batch.columns.pop();
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Fixed(4),
        validity: None,
        data: vec![1, 0, 0, 0, 2, 0, 0, 0].into(), // row 0 = 1, row 1 = 2
    });
    let mut bytes_data = Vec::new();
    for off in [0u32, 2, 5] {
        bytes_data.extend_from_slice(&off.to_le_bytes());
    }
    bytes_data.extend_from_slice(b"aabbb"); // row 0 = "aa", row 1 = "bbb"
    batch.columns.push(Column {
        column_id: 4,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: bytes_data.into(),
    });

    let entries = column_batch_to_entries(&batch).expect("untranspose");
    assert_eq!(entries.len(), 2);
    assert_eq!(entries[0].key.user_key.as_ref(), b"k0");
    assert_eq!(entries[1].key.seqno, 2);

    let tags = [TypeTag::Fixed(4), TypeTag::Bytes];
    assert_eq!(
        unframe_value_cells(entries[0].value.as_ref(), &tags).expect("unframe 0"),
        vec![&[1, 0, 0, 0][..], &b"aa"[..]],
    );
    assert_eq!(
        unframe_value_cells(entries[1].value.as_ref(), &tags).expect("unframe 1"),
        vec![&[2, 0, 0, 0][..], &b"bbb"[..]],
    );
}

#[test]
fn untranspose_rejects_value_subcolumn_id_collisions() {
    // Projection selects value sub-columns by id, so a value sub-column that
    // reuses an intrinsic id (0/1/2) or duplicates another value id would make
    // projected results ambiguous. The untranspose must reject both.
    let make = |value_cols: Vec<Column>| -> ColumnBatch {
        let mut batch = entries_to_column_batch(&[entry(b"k0", 1, ValueType::Value, b"ignored")])
            .expect("transpose");
        batch.columns.pop(); // drop the single opaque value column
        batch.columns.extend(value_cols);
        batch
    };

    // A value sub-column reusing COL_USER_KEY (0).
    let collide_intrinsic = make(vec![Column {
        column_id: COL_USER_KEY,
        type_tag: TypeTag::Fixed(4),
        validity: None,
        data: vec![0, 0, 0, 0].into(),
    }]);
    assert!(
        column_batch_to_entries(&collide_intrinsic).is_err(),
        "a value sub-column reusing an intrinsic id must be rejected",
    );

    // Two value sub-columns sharing id 5.
    let duplicate = make(vec![
        Column {
            column_id: 5,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: vec![0, 0, 0, 0].into(),
        },
        Column {
            column_id: 5,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: vec![1, 0, 0, 0].into(),
        },
    ]);
    assert!(
        column_batch_to_entries(&duplicate).is_err(),
        "duplicate value sub-column ids must be rejected",
    );
}

#[test]
fn validate_ingest_rejects_an_invalid_value_type() {
    // The eager ingest validation must reject a malformed value-type byte on
    // the submitting call, not defer it to the flush-time decode.
    let mut batch =
        entries_to_column_batch(&[entry(b"k0", 0, ValueType::Value, b"v")]).expect("transpose");
    // Column bytes are an immutable view now — rebuild the poisoned column.
    let mut poisoned = batch.columns[2].data.to_vec();
    poisoned[0] = 99; // not a valid ValueType tag
    batch.columns[2].data = poisoned.into();
    assert!(
        validate_columnar_ingest_batch(&batch, &crate::comparator::default_comparator()).is_err(),
        "an invalid value-type tag must be rejected during eager validation",
    );
}

#[test]
fn untranspose_reconstructs_nullable_value_subcolumns() {
    // Two rows, two value sub-columns; the second (fixed-4) is nullable with
    // row 1 absent. Each row's reconstructed value is a presence-bitmap frame
    // that unframe_value_cells_nullable reverses to the original cells.
    let mut batch = entries_to_column_batch(&[
        InternalValue::from_components(b"k0", b"ignored", 0, ValueType::Value),
        InternalValue::from_components(b"k1", b"ignored", 0, ValueType::Value),
    ])
    .expect("transpose");
    batch.columns.pop();
    // Bytes sub-column (always present): "aa", "bbb".
    let mut bytes_data = Vec::new();
    for off in [0u32, 2, 5] {
        bytes_data.extend_from_slice(&off.to_le_bytes());
    }
    bytes_data.extend_from_slice(b"aabbb");
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: bytes_data.into(),
    });
    // Fixed-4 sub-column, row 0 present (=1), row 1 null. validity bit 0 set.
    batch.columns.push(Column {
        column_id: 4,
        type_tag: TypeTag::Fixed(4),
        validity: Some(alloc::vec![0b0000_0001]),
        data: alloc::vec![1, 0, 0, 0, 0, 0, 0, 0].into(),
    });

    let entries = column_batch_to_entries(&batch).expect("untranspose");
    let tags = [TypeTag::Bytes, TypeTag::Fixed(4)];
    assert_eq!(
        unframe_value_cells_nullable(entries[0].value.as_ref(), &tags).expect("row 0"),
        vec![Some(&b"aa"[..]), Some(&[1, 0, 0, 0][..])],
    );
    assert_eq!(
        unframe_value_cells_nullable(entries[1].value.as_ref(), &tags).expect("row 1"),
        vec![Some(&b"bbb"[..]), None],
        "row 1's fixed sub-cell is absent",
    );
}

#[test]
fn untranspose_reconstructs_a_nullable_bytes_subcolumn() {
    // A nullable variable-width (`Bytes`) value sub-column: row 0 present,
    // row 1 null. The null row must still carry valid offset-table entries
    // (here equal consecutive offsets => an empty span) because `validate`
    // frames every row regardless of the validity bitmap; the validity bit,
    // not the offsets, is what marks the row absent.
    let mut batch = entries_to_column_batch(&[
        InternalValue::from_components(b"k0", b"ignored", 0, ValueType::Value),
        InternalValue::from_components(b"k1", b"ignored", 0, ValueType::Value),
    ])
    .expect("transpose");
    batch.columns.pop();
    // Offsets [0, 2, 2]: row 0 spans 0..2 ("hi"), row 1 spans 2..2 (empty).
    let mut data = Vec::new();
    for off in [0u32, 2, 2] {
        data.extend_from_slice(&off.to_le_bytes());
    }
    data.extend_from_slice(b"hi");
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Bytes,
        validity: Some(alloc::vec![0b0000_0001]), // row 0 present, row 1 null
        data: data.into(),
    });

    let entries = column_batch_to_entries(&batch).expect("untranspose");
    let tags = [TypeTag::Bytes];
    assert_eq!(
        unframe_value_cells_nullable(entries[0].value.as_ref(), &tags).expect("row 0"),
        vec![Some(&b"hi"[..])],
    );
    assert_eq!(
        unframe_value_cells_nullable(entries[1].value.as_ref(), &tags).expect("row 1"),
        vec![None],
        "the null bytes row reconstructs as absent, not as an empty cell",
    );
}

#[test]
fn concat_joins_fixed_bytes_and_nullable_columns() {
    // Build a two-row consumer batch with a fixed-4 (id 3), a bytes (id 4),
    // and a nullable fixed-2 (id 5) value sub-column.
    let build = |keys: [&[u8]; 2],
                 fixed: [u32; 2],
                 bytes: [&[u8]; 2],
                 nf2: [Option<[u8; 2]>; 2]|
     -> ColumnBatch {
        let mut batch = entries_to_column_batch(&[
            InternalValue::from_components(keys[0], b"x", 0, ValueType::Value),
            InternalValue::from_components(keys[1], b"x", 0, ValueType::Value),
        ])
        .expect("transpose");
        batch.columns.pop();
        batch.columns.push(Column {
            column_id: 3,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: fixed
                .iter()
                .flat_map(|v| v.to_le_bytes())
                .collect::<Vec<u8>>()
                .into(),
        });
        let mut bytes_data = Vec::new();
        let mut acc = 0u32;
        bytes_data.extend_from_slice(&acc.to_le_bytes());
        for c in bytes {
            acc += u32::try_from(c.len()).expect("bytes cell length fits u32");
            bytes_data.extend_from_slice(&acc.to_le_bytes());
        }
        for c in bytes {
            bytes_data.extend_from_slice(c);
        }
        batch.columns.push(Column {
            column_id: 4,
            type_tag: TypeTag::Bytes,
            validity: None,
            data: bytes_data.into(),
        });
        let mut nf2_data = Vec::new();
        let mut nf2_valid = 0u8;
        for (i, v) in nf2.iter().enumerate() {
            nf2_data.extend_from_slice(&v.unwrap_or([0, 0]));
            if v.is_some() {
                nf2_valid |= 1 << i;
            }
        }
        batch.columns.push(Column {
            column_id: 5,
            type_tag: TypeTag::Fixed(2),
            validity: Some(alloc::vec![nf2_valid]),
            data: nf2_data.into(),
        });
        batch
    };

    let a = build([b"k0", b"k1"], [1, 2], [b"a", b"bb"], [Some([9, 9]), None]);
    let b = build(
        [b"k2", b"k3"],
        [3, 4],
        [b"ccc", b"dddd"],
        [Some([7, 7]), None],
    );
    let a = ColumnBatch::concat(vec![a, b]).expect("concat");
    assert_eq!(a.row_count, 4);

    let entries = column_batch_to_entries(&a).expect("untranspose combined");
    let tags = [TypeTag::Fixed(4), TypeTag::Bytes, TypeTag::Fixed(2)];
    let check =
        |entry: &InternalValue, key: &[u8], fixed: [u8; 4], bytes: &[u8], nf2: Option<[u8; 2]>| {
            assert_eq!(entry.key.user_key.as_ref(), key);
            let cells = unframe_value_cells_nullable(entry.value.as_ref(), &tags).expect("unframe");
            assert_eq!(cells[0], Some(&fixed[..]));
            assert_eq!(cells[1], Some(bytes));
            assert_eq!(cells[2], nf2.as_ref().map(|n| &n[..]));
        };
    check(&entries[0], b"k0", [1, 0, 0, 0], b"a", Some([9, 9]));
    check(&entries[1], b"k1", [2, 0, 0, 0], b"bb", None);
    check(&entries[2], b"k2", [3, 0, 0, 0], b"ccc", Some([7, 7]));
    check(&entries[3], b"k3", [4, 0, 0, 0], b"dddd", None);
}

fn sample_batch() -> ColumnBatch {
    // 3 rows: a fixed u32 column (row 1 null) and a variable-width Bytes
    // column (all valid). Bytes data = offset array [0,2,2,5] + payload.
    let bytes_offsets: Vec<u8> = [0u32, 2, 2, 5]
        .iter()
        .flat_map(|o| o.to_le_bytes())
        .collect();
    let mut bytes_data = bytes_offsets;
    bytes_data.extend_from_slice(b"hi"); // row 0
    // row 1 empty (offset 2..2)
    bytes_data.extend_from_slice(b"abc"); // row 2
    ColumnBatch {
        row_count: 3,
        columns: vec![
            Column {
                column_id: 1,
                type_tag: TypeTag::Fixed(4),
                validity: Some(vec![0b0000_0101]), // rows 0 and 2 valid, row 1 null
                data: vec![10, 0, 0, 0, 0, 0, 0, 0, 30, 0, 0, 0].into(),
            },
            Column {
                column_id: 2,
                type_tag: TypeTag::Bytes,
                validity: None,
                data: bytes_data.into(),
            },
        ],
    }
}

/// `batch` cut into row pages of `cuts` rows each, every column sliced with
/// [`Column::rows`], as the writer cuts a row group.
fn split_into_row_pages(batch: &ColumnBatch, cuts: &[u32]) -> Vec<ColumnBatch> {
    let mut start = 0;
    cuts.iter()
        .map(|&rows| {
            let end = start + rows;
            let columns = batch
                .columns
                .iter()
                .map(|c| c.rows(batch.row_count, start, end).expect("rows"))
                .collect();
            start = end;
            ColumnBatch {
                row_count: rows,
                columns,
            }
        })
        .collect()
}

/// Every way of cutting a batch into row pages joins back into the batch it
/// was cut from: the nullable fixed column keeps its null, the bytes column
/// its empty cell, whichever page they land on. A reader that joins a group's
/// row pages depends on this, and so does every page the writer cuts.
#[test]
fn a_batch_cut_into_row_pages_joins_back_unchanged() {
    let batch = sample_batch();
    for cuts in [&[3][..], &[1, 2], &[2, 1], &[1, 1, 1]] {
        let pages = split_into_row_pages(&batch, cuts);
        assert_eq!(pages.iter().map(|p| p.row_count).collect::<Vec<_>>(), cuts,);
        let joined = ColumnBatch::concat(pages).expect("concat");
        assert_eq!(joined, batch, "cut {cuts:?} must join back to the batch");
    }
}

/// A row page holds exactly its rows: the middle row of the sample, null in
/// the fixed column and empty in the bytes column, is a page of its own that
/// says so.
#[test]
fn a_row_page_of_one_row_holds_that_row() {
    let batch = sample_batch();
    let pages = split_into_row_pages(&batch, &[1, 1, 1]);
    let middle = pages.get(1).expect("three pages");
    let fixed = middle.columns.first().expect("fixed column");
    assert_eq!(fixed.validity.as_deref(), Some(&[0u8][..]), "row 1 is null");
    assert_eq!(&fixed.data[..], &[0, 0, 0, 0]);
    let bytes = middle.columns.get(1).expect("bytes column");
    assert_eq!(&bytes.data[..], &[0, 0, 0, 0, 0, 0, 0, 0], "one empty cell");
}

/// A row range past the column, or reversed, is refused rather than clamped:
/// a clamped page would hold fewer rows than the directory says it does.
#[test]
fn column_rows_refuses_a_range_outside_the_column() {
    let batch = sample_batch();
    for col in &batch.columns {
        assert!(col.rows(3, 2, 4).is_err(), "end past the rows");
        assert!(col.rows(3, 2, 1).is_err(), "reversed range");
        assert!(col.rows(3, 0, 3).is_ok(), "the whole column");
    }
}

/// Row pages close once they reach the page size, counting a fixed column's
/// width and a bytes cell's length plus its offset: the sample's rows weigh
/// 10, 8 and 11 bytes.
#[test]
fn row_page_cuts_close_a_page_once_it_reaches_the_size() {
    let batch = sample_batch();
    let cuts = |size| batch.row_page_cuts(size).expect("cuts");
    assert_eq!(
        cuts(0),
        [1, 1, 1],
        "a zero size puts every row on its own page"
    );
    assert_eq!(cuts(10), [1, 2], "the first row alone reaches 10 bytes");
    assert_eq!(cuts(18), [2, 1], "the first two rows reach 18 bytes");
    assert_eq!(cuts(1_000), [3], "a size no row reaches keeps one page");
}

/// Pages that disagree on their columns cannot be one group, and no pages
/// are no group: both are refused rather than joined into something else.
#[test]
fn concat_refuses_no_pages_and_pages_with_other_columns() {
    assert!(ColumnBatch::concat(Vec::new()).is_err(), "no row pages");

    let batch = sample_batch();
    let mut pages = split_into_row_pages(&batch, &[1, 2]);
    pages
        .get_mut(1)
        .expect("two pages")
        .columns
        .first_mut()
        .expect("a column")
        .column_id = 9;
    assert!(ColumnBatch::concat(pages).is_err(), "a renamed column");

    let mut pages = split_into_row_pages(&batch, &[1, 2]);
    pages.get_mut(1).expect("two pages").columns.pop();
    assert!(ColumnBatch::concat(pages).is_err(), "a missing column");
}

#[test]
fn columnar_batch_round_trips() {
    let batch = sample_batch();
    let encoded = batch.encode().expect("encode");
    let decoded = ColumnBatch::decode(&encoded.into()).expect("decode");
    assert_eq!(decoded, batch, "columnar batch must survive a round-trip");
}

#[test]
fn columnar_decode_rejects_truncated_payload() {
    let encoded = sample_batch().encode().expect("encode");
    // Drop the last byte: the final column's data is now short.
    let truncated = &encoded[..encoded.len() - 1];
    assert!(ColumnBatch::decode(&truncated.to_vec().into()).is_err());
}

#[test]
fn columnar_encode_rejects_fixed_width_length_mismatch() {
    // A fixed(4) column claiming 3 rows but carrying 8 bytes (= 2 rows) is
    // rejected at encode, so no non-round-trippable payload is produced.
    let bad = ColumnBatch {
        row_count: 3,
        columns: vec![Column {
            column_id: 1,
            type_tag: TypeTag::Fixed(4),
            validity: None,
            data: vec![0; 8].into(),
        }],
    };
    assert!(bad.encode().is_err());
}

/// Where the sample batch's fields lie: the first column's header starts at
/// 8 (after `row_count` and `column_count`), so its validity flag is at 12,
/// its values length at 13, its one-byte bitmap at 14 and its operator at 15,
/// then its 12 plain bytes. The second column starts at 28, its width at 31,
/// and its values, encoded by their lengths, open at 34 with the lengths'
/// FFOR: operator 35, base 36, bit width 37.
#[test]
fn sample_batch_encodes_as_the_layout_the_offset_tests_assume() {
    let encoded = sample_batch().encode().expect("encode");
    assert_eq!(encoded[12], 1, "the first column is nullable");
    assert_eq!(encoded[15], 0, "the opaque fixed column is plain");
    assert_eq!(encoded[31], 0, "the bytes column's width");
    assert_eq!(encoded[34], 7, "the bytes column is encoded by its lengths");
    assert_eq!(encoded[37], 2, "lengths 2, 0 and 3 pack in two bits");
}

#[test]
fn columnar_decode_rejects_an_unknown_operator() {
    let mut encoded = sample_batch().encode().expect("encode");
    encoded[15] = 0xFF;
    assert!(ColumnBatch::decode(&encoded.clone().into()).is_err());
}

#[test]
fn columnar_encode_rejects_zero_width_fixed_column() {
    // Fixed(0) is publicly constructible but cannot round-trip; encode must
    // reject it rather than emit bytes decode would refuse.
    let bad = ColumnBatch {
        row_count: 1,
        columns: vec![Column {
            column_id: 1,
            type_tag: TypeTag::Fixed(0),
            validity: None,
            data: Vec::new().into(),
        }],
    };
    assert!(bad.encode().is_err());
}

#[test]
fn columnar_encode_rejects_wrong_validity_length() {
    // 1 row needs a 1-byte bitmap; a 2-byte bitmap is rejected.
    let bad = ColumnBatch {
        row_count: 1,
        columns: vec![Column {
            column_id: 1,
            type_tag: TypeTag::Fixed(1),
            validity: Some(vec![0, 0]),
            data: vec![7].into(),
        }],
    };
    assert!(bad.encode().is_err());
}

#[test]
fn columnar_encode_rejects_validity_padding_bits() {
    // 1 row: only bit 0 is meaningful; a set padding bit (0xFF) is rejected.
    let bad = ColumnBatch {
        row_count: 1,
        columns: vec![Column {
            column_id: 1,
            type_tag: TypeTag::Fixed(1),
            validity: Some(vec![0xFF]),
            data: vec![7].into(),
        }],
    };
    assert!(bad.encode().is_err());
}

#[test]
fn columnar_decode_rejects_a_bit_width_past_64() {
    let mut encoded = sample_batch().encode().expect("encode");
    encoded[37] = 65;
    assert!(ColumnBatch::decode(&encoded.clone().into()).is_err());
}

#[test]
fn columnar_decode_rejects_trailing_bytes() {
    let mut encoded = sample_batch().encode().expect("encode");
    encoded.push(0); // one byte past the last declared column
    assert!(ColumnBatch::decode(&encoded.clone().into()).is_err());
}

/// A page with bytes after its column is refused while it is parsed, before
/// its validity bitmap or values are copied out: the refusal costs no copy,
/// and none is counted.
#[test]
fn column_page_decode_refused_for_a_tail_copies_nothing() {
    use crate::table::column_page::{PageId, PageStamp};

    let batch = sample_batch();
    let nullable = batch.columns.first().expect("the nullable fixed column");
    let stamp = PageStamp {
        group_tag: 7,
        id: PageId {
            column_id: nullable.column_id,
            part: 0,
        },
        row_page: 0,
    };
    let mut page = nullable
        .encode_page(batch.row_count, stamp, ColumnEncoding::Auto)
        .expect("encode page");
    page.push(0); // one byte past the page's column
    let mut copied = 0usize;
    let decoded = Column::decode_page(
        &page.into(),
        batch.row_count,
        stamp,
        &mut copied,
        &mut super::DecodeBudget::default(),
    );
    assert!(
        matches!(decoded, Err(crate::Error::InvalidHeader(m)) if m.contains("trailing bytes")),
        "trailing bytes must be refused, got {decoded:?}",
    );
    assert_eq!(copied, 0, "the refusal came before any copy");
}

/// A group whose pages decode to more bytes than any row group a writer cuts
/// is refused before the column is built: 200 rows of one 64 KiB value
/// encode as a constant of one value, while their layout is 12.5 MiB, past
/// twice the 4 MiB a group is cut at beyond the page's own bytes. Without the
/// bound a small forged page makes a read allocate what its rows decode to.
/// The bound is the group's, not the page's: three pages of 64 such rows are
/// each a page a writer could cut, but no writer puts all three in one group,
/// so the third is refused. One page of 64 rows decodes.
#[test]
fn column_page_decoding_past_what_a_writer_cuts_is_refused() {
    use crate::table::column_page::{PageId, PageStamp};

    let value = vec![7u8; 64 * 1024];
    let bytes_column = |rows: u32| Column {
        column_id: 3,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: super::build_bytes_column((0..rows).map(|_| value.as_slice()))
            .expect("build")
            .into(),
    };
    let stamp = PageStamp {
        group_tag: 7,
        id: PageId {
            column_id: 3,
            part: 0,
        },
        row_page: 0,
    };
    let page_of = |rows: u32| -> crate::Slice {
        let page = bytes_column(rows)
            .encode_page(rows, stamp, ColumnEncoding::Auto)
            .expect("encode page");
        assert!(page.len() < 2 * value.len(), "the rows encode as one value");
        page.into()
    };

    let refused = |result: crate::Result<Column>| {
        matches!(result, Err(crate::Error::DecompressedSizeTooLarge { .. }))
    };

    let forged = page_of(200);
    Column::parse_page(&forged, 200, stamp)
        .expect("a parse builds nothing, so it has nothing to refuse");
    let mut copied = 0usize;
    assert!(
        refused(Column::decode_page(
            &forged,
            200,
            stamp,
            &mut copied,
            &mut super::DecodeBudget::default(),
        )),
        "a page past the bound must be refused",
    );
    assert_eq!(copied, 0, "refused before anything was built");

    // 64 rows reach the 4 MiB a group is cut at, so a writer never puts more
    // of these rows in one group.
    let cut = page_of(64);
    let mut group = super::DecodeBudget::default();
    for page in 0..2 {
        let column = Column::decode_page(&cut, 64, stamp, &mut copied, &mut group)
            .unwrap_or_else(|e| panic!("page {page} of the group: {e:?}"));
        assert_eq!(column.data.len(), 65 * 4 + 64 * value.len());
    }
    assert!(
        refused(Column::decode_page(
            &cut,
            64,
            stamp,
            &mut copied,
            &mut group
        )),
        "the third page takes the group past the bound",
    );
    Column::decode_page(
        &cut,
        64,
        stamp,
        &mut copied,
        &mut super::DecodeBudget::default(),
    )
    .expect("the same page in a group of its own");
}

/// A block whose second row carries a bad value-type tag is refused only after
/// the first row's value was rebuilt from its sub-columns: that rebuild was a
/// copy and is counted, like a read a later check refuses.
#[test]
fn columnar_entries_refused_after_rebuilding_a_value_still_count_it() {
    let mut batch = entries_to_column_batch(&[
        entry(b"k0", 1, ValueType::Value, b"ignored"),
        entry(b"k1", 2, ValueType::Value, b"ignored"),
    ])
    .expect("transpose");
    batch.columns.pop();
    batch.columns.push(Column {
        column_id: 3,
        type_tag: TypeTag::Fixed(4),
        validity: None,
        data: vec![1, 0, 0, 0, 2, 0, 0, 0].into(),
    });
    let mut bytes_data = Vec::new();
    for off in [0u32, 2, 5] {
        bytes_data.extend_from_slice(&off.to_le_bytes());
    }
    bytes_data.extend_from_slice(b"aabbb");
    batch.columns.push(Column {
        column_id: 4,
        type_tag: TypeTag::Bytes,
        validity: None,
        data: bytes_data.into(),
    });
    let vt_col = batch.columns.get_mut(2).expect("value-type column");
    vt_col.data = vec![u8::from(ValueType::Value), 0xEE].into();

    let mut gathered = 0usize;
    let refused = crate::table::data_block::DataBlock::column_batch_entries([batch], &mut gathered);
    assert!(refused.is_err(), "the bad value-type tag must be refused");
    let first_row = frame_value_cells(&[
        (TypeTag::Fixed(4), &[1, 0, 0, 0][..]),
        (TypeTag::Bytes, b"aa"),
    ])
    .expect("frame");
    assert_eq!(
        gathered,
        first_row.len(),
        "the first row's value was rebuilt before the refusal",
    );
}

/// The `Bytes` offset accumulator's overflow guard, exercised WITHOUT
/// materializing a multi-GiB payload: the arithmetic is driven directly at the
/// u32 boundary.
#[test]
fn advance_bytes_offset_rejects_a_u32_offset_overflow() {
    // A repeated gather can push the accumulated total past u32::MAX; the
    // `checked_add` guard rejects it (a tiny `value_len`, nothing allocated).
    assert!(matches!(
        super::advance_bytes_offset(u32::MAX - 3, 10),
        Err(crate::Error::DecompressedSizeTooLarge { .. }),
    ));
    // A single value longer than u32::MAX trips the `u32::try_from` guard. Only
    // reachable where usize is wider than u32 (64-bit); on a 32-bit target usize
    // IS u32, so a length can never exceed it.
    #[cfg(target_pointer_width = "64")]
    assert!(matches!(
        super::advance_bytes_offset(0, (u32::MAX as usize) + 1),
        Err(crate::Error::DecompressedSizeTooLarge { .. }),
    ));
    // A normal advance within the u32 ceiling succeeds.
    assert_eq!(
        super::advance_bytes_offset(100, 50).expect("within the u32 offset ceiling"),
        150,
    );
}

/// The framed column is sized by one pass over the cells and filled by a
/// second, so a source that yields differently the second time must be
/// refused rather than leave part of the buffer unwritten.
#[test]
fn frame_bytes_column_refuses_cells_that_change_between_passes() {
    let calls = core::cell::Cell::new(0u32);
    let short: &[&[u8]] = &[b"a", b"b"];
    let long: &[&[u8]] = &[b"a", b"bbb"];
    let fewer: &[&[u8]] = &[b"a"];
    for second in [long, short, fewer] {
        calls.set(0);
        let framed = super::frame_bytes_column(2, || {
            calls.set(calls.get() + 1);
            if calls.get() == 1 { long } else { second }.iter().copied()
        });
        if core::ptr::eq(second, long) {
            let framed = framed.expect("identical passes frame the cells");
            let mut expected = Vec::new();
            for off in [0u32, 1, 4] {
                expected.extend_from_slice(&off.to_le_bytes());
            }
            expected.extend_from_slice(b"abbb");
            assert_eq!(&*framed, &expected[..]);
        } else {
            assert!(
                matches!(framed, Err(crate::Error::InvalidHeader(_))),
                "a second pass yielding other cells must be refused",
            );
        }
    }
    // Declared count and yielded cells disagree on the first pass.
    assert!(super::frame_bytes_column(3, || short.iter().copied()).is_err());
}

#[test]
fn columnar_decode_rejects_huge_column_count() {
    // row_count = 0, column_count = u32::MAX, but no column bytes follow.
    let mut payload = 0u32.to_le_bytes().to_vec();
    payload.extend_from_slice(&u32::MAX.to_le_bytes());
    assert!(ColumnBatch::decode(&payload.clone().into()).is_err());
}

#[test]
fn columnar_decode_rejects_non_boolean_validity_flag() {
    let mut encoded = sample_batch().encode().expect("encode");
    encoded[12] = 2;
    assert!(ColumnBatch::decode(&encoded.clone().into()).is_err());
}

#[test]
fn columnar_decode_rejects_non_zero_bytes_width() {
    let mut encoded = sample_batch().encode().expect("encode");
    encoded[31] = 5;
    assert!(ColumnBatch::decode(&encoded.clone().into()).is_err());
}

#[test]
fn column_batch_into_entries_rejects_an_empty_key_row() {
    // A corrupt block can frame an empty key (offset table [0, 0], no payload) —
    // a byte layout the structural validator accepts. The scan reconstruction
    // must reject it rather than emit an empty-keyed entry.
    let mut batch =
        entries_to_column_batch(&[entry(b"k", 5, ValueType::Value, b"v")]).expect("valid batch");
    let key_col = batch.columns.get_mut(0).expect("key column");
    key_col.data = alloc::vec![0u8; 8].into();
    let err = column_batch_into_entries(batch, &mut 0).expect_err("empty key must be rejected");
    assert!(
        matches!(err, crate::Error::InvalidHeader(m) if m.contains("user key is empty")),
        "expected an empty-key InvalidHeader, got {err:?}",
    );
}

/// Runs the point read over `batch` as one row page, its columns written as
/// the pages a writer writes and read back as a lookup reads them: the key
/// search first, then the matcher over the run it found.
fn match_as_pages(
    batch: &ColumnBatch,
    needle: &[u8],
    deletes: Option<(&crate::table::delete_bitmap::DeleteBitmap, u32)>,
    copied: &mut usize,
) -> crate::Result<Vec<InternalValue>> {
    match_as_pages_within(
        batch,
        needle,
        deletes,
        copied,
        &mut super::DecodeBudget::default(),
    )
}

/// [`match_as_pages`] charging what the lookup holds decoded to `budget`.
fn match_as_pages_within(
    batch: &ColumnBatch,
    needle: &[u8],
    deletes: Option<(&crate::table::delete_bitmap::DeleteBitmap, u32)>,
    copied: &mut usize,
    budget: &mut super::DecodeBudget,
) -> crate::Result<Vec<InternalValue>> {
    use crate::table::column_page::{PageId, PageStamp};

    let stamp = |column_id| PageStamp {
        group_tag: 1,
        id: PageId { column_id, part: 0 },
        row_page: 0,
    };
    let pages = batch
        .columns
        .iter()
        .map(|c| {
            c.encode_page(batch.row_count, stamp(c.column_id), ColumnEncoding::Auto)
                .map(Slice::from)
        })
        .collect::<crate::Result<Vec<_>>>()?;
    let columns = pages
        .iter()
        .zip(&batch.columns)
        .map(|(page, c)| {
            budget.parsed(Column::parse_page(
                page,
                batch.row_count,
                stamp(c.column_id),
            )?)
        })
        .collect::<crate::Result<Vec<_>>>()?;
    let rows = batch.row_count;
    // The key pages are searched on their own; the matcher reads the rest.
    let mut columns: Vec<_> = columns;
    let key = columns.remove(0);
    let keys = budget.rows(key.values, TypeTag::Bytes, rows)?;
    let run = super::key_rows(
        rows,
        needle,
        &crate::comparator::default_comparator(),
        |row| keys.get(TypeTag::Bytes, rows, row),
    )?;
    let mut out = Vec::new();
    // A miss on a page that has rows reads nothing more, as in a lookup.
    if run.is_empty() && rows > 0 {
        return Ok(out);
    }
    page_match_entries(
        RowPageColumns {
            ordinal: 0,
            start: 0,
            rows,
            columns,
        },
        run,
        needle,
        deletes,
        copied,
        budget,
        &mut out,
    )?;
    Ok(out)
}

/// A lookup holds decoded only what a group a writer cut lets it hold: each
/// column it prepares for row reads, other than a layout or a constant read
/// in place, holds up to a row's worth per row, and the group's budget counts
/// them over every column and page. A forged group of many rows and columns
/// encoded that way would otherwise make a lookup of one row hold gigabytes.
/// Here 64 rows of one key under varied values: the seqnos by their ordinals
/// and the values by their lengths each hold the page's 64 rows, so a budget
/// of 100 rows refuses the lookup and one of 1000 serves it.
#[test]
fn a_lookup_holds_decoded_no_more_rows_than_its_group_budget() {
    let entries: Vec<_> = (0..64u32)
        .map(|i| {
            entry(
                b"dup",
                u64::from(1_000 - i),
                ValueType::Value,
                &alloc::vec![b'v'; (i % 7) as usize],
            )
        })
        .collect();
    let batch = entries_to_column_batch(&entries).expect("batch");
    let budget = |allowance| super::DecodeBudget {
        allowance,
        built: 0,
        held: 0,
    };
    let refused = match_as_pages_within(&batch, b"dup", None, &mut 0, &mut budget(100));
    assert!(
        matches!(refused, Err(crate::Error::DecompressedSizeTooLarge { .. })),
        "a lookup past the budget must be refused, got {refused:?}",
    );
    let found = match_as_pages_within(&batch, b"dup", None, &mut 0, &mut budget(1_000))
        .expect("a lookup within the budget");
    assert_eq!(found.len(), 64, "every version is found");
}

#[test]
fn page_match_entries_finds_every_version_of_a_key_and_nothing_else() {
    // Three keys, the middle one in two versions: a lookup returns exactly its
    // versions, newest first, and a key between or past them returns none.
    let batch = entries_to_column_batch(&[
        entry(b"a", 9, ValueType::Value, b"va"),
        entry(b"m", 7, ValueType::Value, b"new"),
        entry(b"m", 4, ValueType::Tombstone, b""),
        entry(b"z", 2, ValueType::Value, b"vz"),
    ])
    .expect("batch");
    let found = match_as_pages(&batch, b"m", None, &mut 0).expect("lookup");
    assert_entries_eq(
        &found,
        &[
            entry(b"m", 7, ValueType::Value, b"new"),
            entry(b"m", 4, ValueType::Tombstone, b""),
        ],
    );
    for absent in [&b"b"[..], b"zz", b"0"] {
        assert!(
            match_as_pages(&batch, absent, None, &mut 0)
                .expect("lookup")
                .is_empty(),
            "no row holds {absent:?}",
        );
    }
}

#[test]
fn page_match_entries_rejects_an_empty_key_row() {
    // The point-read matcher applies the same non-empty-key invariant as the
    // scan path: a matched empty key in a corrupt block is an error, not a hit.
    let mut batch =
        entries_to_column_batch(&[entry(b"k", 5, ValueType::Value, b"v")]).expect("valid batch");
    let key_col = batch.columns.get_mut(0).expect("key column");
    key_col.data = alloc::vec![0u8; 8].into();
    let err =
        match_as_pages(&batch, b"", None, &mut 0).expect_err("matched empty key must be rejected");
    assert!(
        matches!(err, crate::Error::InvalidHeader(m) if m.contains("user key is empty")),
        "expected an empty-key InvalidHeader, got {err:?}",
    );
}

#[test]
fn page_match_entries_fails_closed_when_a_delete_mask_position_overflows() {
    // Two rows share a key, so both match the needle. With `block_start_row` at
    // u32::MAX, the second matched row's position (start + 1) overflows u32. A
    // masked point-read lookup must fail closed (error) like the scan path, never
    // treat the overflow as "not deleted" and silently expose the row.
    let batch = entries_to_column_batch(&[
        entry(b"dup", 5, ValueType::Value, b"v0"),
        entry(b"dup", 3, ValueType::Value, b"v1"),
    ])
    .expect("two-row same-key batch");
    let bitmap = crate::table::delete_bitmap::DeleteBitmap::new();
    let err = match_as_pages(&batch, b"dup", Some((&bitmap, u32::MAX)), &mut 0)
        .expect_err("an overflowing delete-mask position must fail closed");
    assert!(
        matches!(err, crate::Error::InvalidHeader(m) if m.contains("position exceeds u32::MAX")),
        "expected an overflow InvalidHeader, got {err:?}",
    );
}

#[test]
fn page_match_entries_counts_the_rows_it_copied_before_a_later_row_fails() {
    // Two versions of one key; the second carries an invalid value-type byte.
    // The first version's key and value were already copied out of the columns
    // when the second is refused, and those copies must still be reported.
    let mut batch = entries_to_column_batch(&[
        entry(b"dup", 5, ValueType::Value, b"v0"),
        entry(b"dup", 3, ValueType::Value, b"v1"),
    ])
    .expect("two-row same-key batch");
    let vt_col = batch.columns.get_mut(2).expect("value-type column");
    let mut vt = vt_col.data.to_vec();
    vt[1] = 0xFF;
    vt_col.data = vt.into();
    let mut copied = 0;
    let err = match_as_pages(&batch, b"dup", None, &mut copied)
        .expect_err("the invalid value type must be refused");
    assert!(
        matches!(err, crate::Error::InvalidTag(("ValueType", 0xFF))),
        "expected an invalid value-type tag, got {err:?}",
    );
    assert_eq!(
        copied,
        b"dup".len() + b"v0".len(),
        "the first version was copied before the second failed",
    );
}

#[test]
fn page_match_entries_refuses_a_run_outside_its_page() {
    // The run comes from the key search over the same page; one that names
    // rows the page does not have (or none) is an inconsistency to refuse,
    // never rows to read past the page.
    use crate::table::column_page::{PageId, PageStamp};

    let batch = entries_to_column_batch(&[
        entry(b"k", 5, ValueType::Value, b"v0"),
        entry(b"k", 3, ValueType::Value, b"v1"),
    ])
    .expect("batch");
    let stamp = |column_id| PageStamp {
        group_tag: 1,
        id: PageId { column_id, part: 0 },
        row_page: 0,
    };
    let pages = batch
        .columns
        .iter()
        .map(|c| {
            Slice::from(
                c.encode_page(2, stamp(c.column_id), ColumnEncoding::Auto)
                    .expect("page"),
            )
        })
        .collect::<Vec<_>>();
    for run in [1..3, 2..2, 0..0] {
        // Every column but the key's, as a lookup's second read takes them.
        let columns = pages
            .iter()
            .zip(&batch.columns)
            .skip(1)
            .map(|(page, c)| Column::parse_page(page, 2, stamp(c.column_id)).expect("parse"))
            .collect();
        let page = RowPageColumns {
            ordinal: 0,
            start: 0,
            rows: 2,
            columns,
        };
        let err = page_match_entries(
            page,
            run.clone(),
            b"k",
            None,
            &mut 0,
            &mut super::DecodeBudget::default(),
            &mut Vec::new(),
        )
        .expect_err("a run outside the page must be refused");
        assert!(
            matches!(err, crate::Error::InvalidHeader(m) if m.contains("key run outside")),
            "run {run:?}: expected a run InvalidHeader, got {err:?}",
        );
    }
}

#[test]
fn page_match_entries_rejects_a_zero_row_block() {
    // A zero-row columnar block is malformed (the writer never emits one). The
    // point-read matcher must fail closed on it, not return an empty match that
    // load_columnar_point_block would turn into an absent-key miss, hiding the
    // on-disk corruption.
    let batch = entries_to_column_batch(&[]).expect("zero-row batch");
    let err =
        match_as_pages(&batch, b"x", None, &mut 0).expect_err("a zero-row block must fail closed");
    assert!(
        matches!(err, crate::Error::InvalidHeader(m) if m.contains("empty reconstructed data block")),
        "expected an empty-block InvalidHeader, got {err:?}",
    );
}

#[test]
fn zone_stats_records_every_ordered_column_and_omits_opaque_ones() {
    // A columnar block's zone map must carry one entry PER ordered column (the
    // user-key and value Bytes columns, and the seqno number column), each
    // with the min / max of its comparable encoding over its rows — not just a
    // single synthetic key-range column. The opaque fixed value-type column has
    // no order and is omitted, so a block-skip over it stays conservative.
    let batch = entries_to_column_batch(&[
        entry(b"alpha", 3, ValueType::Value, b"mval"),
        entry(b"gamma", 2, ValueType::Value, b"aval"),
        entry(b"omega", 1, ValueType::Value, b"zval"),
    ])
    .expect("three-row batch");

    let stats = batch.zone_stats();

    // User key = 0, seqno = 1 and value = 3, in column order; the value-type
    // column (2) is absent.
    let ids: Vec<u32> = stats.iter().map(|s| s.column_id).collect();
    assert_eq!(
        ids,
        vec![
            u32::from(COL_USER_KEY),
            u32::from(COL_SEQNO),
            u32::from(COL_VALUE)
        ],
        "only the ordered columns get a zone-map entry, in column order",
    );
    // The seqno range in its comparable (big-endian) encoding.
    assert_eq!(stats[1].min, 1u64.to_be_bytes());
    assert_eq!(stats[1].max, 3u64.to_be_bytes());

    let key_stat = &stats[0];
    assert_eq!(
        key_stat.min, b"alpha",
        "user-key min is the first sorted key"
    );
    assert_eq!(
        key_stat.max, b"omega",
        "user-key max is the last sorted key"
    );
    assert_eq!(key_stat.row_count, 3);
    assert_eq!(key_stat.null_count, 0);
    // Bytes wire tag (see `TypeTag::to_wire`).
    assert_eq!(key_stat.type_tag, 1);

    let val_stat = &stats[2];
    assert_eq!(
        val_stat.min, b"aval",
        "value min is the byte-wise minimum value, independent of key order",
    );
    assert_eq!(
        val_stat.max, b"zval",
        "value max is the byte-wise maximum value",
    );
    assert_eq!(val_stat.row_count, 3);
    assert_eq!(val_stat.null_count, 0);

    // The value column's range enables a non-key block-skip: a predicate whose
    // value band is entirely below the block minimum proves the block holds no
    // match, which the old single-column (key-only) map could never do.
    let below = crate::table::columnar_predicate::ColumnRangePredicate {
        column_id: COL_VALUE,
        lower: None,
        upper: Some(b"aaaa".to_vec()),
        apply: crate::table::columnar_predicate::PredicateApply::Filter,
    };
    assert!(
        below.can_skip_block(&stats),
        "a value-column predicate below the block's value range skips the block",
    );
}

#[test]
fn bytes_column_span_names_each_row_and_refuses_a_row_past_the_payload() {
    use super::{bytes_column_row, bytes_column_span};

    // Two rows, "ab" and "c": offsets 0, 2, 3 then the payload.
    let mut data = Vec::new();
    for offset in [0u32, 2, 3] {
        data.extend_from_slice(&offset.to_le_bytes());
    }
    data.extend_from_slice(b"abc");
    assert_eq!(bytes_column_span(&data, 2, 0).expect("row 0"), 12..14);
    assert_eq!(bytes_column_row(&data, 2, 1).expect("row 1"), b"c");

    // An offset past the payload names bytes the column does not hold: the
    // span is refused rather than cut short or wrapped.
    let mut forged = data.clone();
    forged[8..12].copy_from_slice(&9u32.to_le_bytes());
    assert!(bytes_column_span(&forged, 2, 1).is_err());
    // So is an offset that runs backwards.
    let mut backwards = data;
    backwards[4..8].copy_from_slice(&3u32.to_le_bytes());
    backwards[8..12].copy_from_slice(&2u32.to_le_bytes());
    assert!(bytes_column_span(&backwards, 2, 1).is_err());
}