gnitz-expr 0.1.6

//! Resolved-addressing reads, driven through a non-engine [`crate::BatchView`].

use crate::SchemaColumn;
use std::cmp::Ordering;

use crate::test_support::TestView;
use crate::SchemaDescriptor;
use crate::{cmp_order_keys, order_bits, order_locators, ColumnLocator, OrderLocator, RowRanking};
use gnitz_wire::{FixedInt, OrderKey, RowSource, ScalarKind, TypeCode};

/// Every locator read of every key type, from a key column and from a payload
/// slot holding the same values, against the value's own encoding.
#[test]
fn every_locator_read_agrees_with_the_values_encoding() {
    const PATTERNS: &[u128] = &[
        0,
        1,
        2,
        0x7F,
        0x80,
        0xFF,
        0x100,
        i64::MAX as u128,
        1 << 63,
        u64::MAX as u128,
        1 << 64,
        i128::MAX as u128,
        1 << 127,
        u128::MAX,
    ];
    let opk = |native: u128, tc: TypeCode| {
        let mut out = vec![0u8; tc.wire_stride()];
        gnitz_wire::store_opk(&mut out, native, tc.is_signed_int());
        out
    };
    for &tc in TypeCode::ALL.iter().filter(|t| t.is_pk_eligible()) {
        let w = tc.wire_stride();
        let schema = SchemaDescriptor::new(
            &[
                SchemaColumn::new(TypeCode::U16, false), // key column 0
                SchemaColumn::new(tc, false),            // key column 1, at byte 2
                SchemaColumn::new(TypeCode::I64, true),  // payload slot 0
                SchemaColumn::new(tc, true),             // payload slot 1
            ],
            &[0, 1],
        );
        let natives: Vec<Vec<u8>> = PATTERNS.iter().map(|p| p.to_le_bytes()[..w].to_vec()).collect();
        let mut v = TestView::for_schema(&schema, PATTERNS.len());
        for (row, &p) in PATTERNS.iter().enumerate() {
            v.set_native(&schema, row, 1, p);
            v.set_native(&schema, row, 3, p);
            v.set_null_word(row, u64::from(row % 2 == 0) | u64::from(row % 3 == 0) << 1);
        }
        let (from_pk, from_payload) = (schema.locate(1), schema.locate(3));
        assert!(matches!(from_pk, ColumnLocator::Pk { byte_off: 2, .. }), "{from_pk:?}");
        for loc in [from_pk, from_payload] {
            let arm = if loc == from_pk { "pk" } else { "payload" };
            for (row, native) in natives.iter().enumerate() {
                let label = format!("{tc} {arm} row {row}");
                assert_eq!(
                    loc.opk_image(&v, row),
                    gnitz_wire::widen_pk_be(&opk(PATTERNS[row], tc)),
                    "{label}"
                );
                assert_eq!(loc.native_le_bytes(&v, row, &mut [0u8; 16]), &native[..], "{label}");
                assert_eq!(loc.is_null(&v, row), loc == from_payload && row % 3 == 0, "{label}");
                let Some(fi) = FixedInt::from_type_code(tc) else {
                    continue;
                };
                assert_eq!(loc.decode_i64(&v, row, fi), fi.decode_le_i64(native), "{label}");
                let kind = ScalarKind::Int(fi);
                assert_eq!(
                    order_bits(&loc, &v, row, kind),
                    kind.order_image(fi.decode_le_i64(native) as u64),
                    "{label}"
                );
            }
            for i in 0..natives.len() {
                for j in 0..natives.len() {
                    assert_eq!(
                        loc.cmp_non_null(&v, i, &v, j),
                        loc.opk_image(&v, i).cmp(&loc.opk_image(&v, j)),
                        "{tc} {arm} rows {i}, {j}"
                    );
                }
            }
        }
    }
    // The at-rest image itself: big-endian with the sign bit flipped, so `-1`
    // lands just below `i64::MAX` and `i64::MIN` at all-zero.
    assert_eq!(
        opk(-1i64 as u128, TypeCode::I64),
        [0x7F, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]
    );
    assert_eq!(opk(i64::MIN as u128, TypeCode::I64), [0; 8]);
}

/// `order_bits` of a float payload slot is the order image of the slot's bits.
#[test]
fn order_bits_reads_a_float_slot_as_its_image() {
    const FLOATS: [f64; 3] = [-1.5, 0.0, f64::NAN];
    for (kind, tc) in [(ScalarKind::F32, TypeCode::F32), (ScalarKind::F64, TypeCode::F64)] {
        let schema = SchemaDescriptor::new(
            &[SchemaColumn::new(TypeCode::U64, false), SchemaColumn::new(tc, false)],
            &[0],
        );
        let mut v = TestView::for_schema(&schema, FLOATS.len());
        let bits = |f: f64| match kind {
            ScalarKind::F32 => (f as f32).to_bits() as u64,
            _ => f.to_bits(),
        };
        for (row, &f) in FLOATS.iter().enumerate() {
            v.set_native(&schema, row, 1, bits(f) as u128);
        }
        let loc = schema.locate(1);
        for (row, &f) in FLOATS.iter().enumerate() {
            assert_eq!(order_bits(&loc, &v, row, kind), kind.order_image(bits(f)), "{tc} {f}");
        }
    }
}

/// Under `total` the identity tiebreak follows the written keys: PK columns in
/// PK-list order (here the reverse of column order), then payload columns in slot
/// order, all ASC NULLS FIRST.
#[test]
fn order_locators_append_the_identity_tiebreak_in_pk_list_order() {
    // `PRIMARY KEY (c3, c0)`.
    let schema = SchemaDescriptor::new(
        &[
            SchemaColumn::new(TypeCode::U32, false),
            SchemaColumn::new(TypeCode::String, true),
            SchemaColumn::new(TypeCode::F64, false),
            SchemaColumn::new(TypeCode::I64, false),
        ],
        &[3, 0],
    );
    let written = OrderKey { col: 2, desc: true, nulls_first: false };
    let c0 = ColumnLocator::Pk {
        byte_off: 8,
        size: 4,
        type_code: TypeCode::U32,
    };
    let c1 = ColumnLocator::Payload {
        slot: 0,
        size: 16,
        type_code: TypeCode::String,
    };
    let c2 = ColumnLocator::Payload {
        slot: 1,
        size: 8,
        type_code: TypeCode::F64,
    };
    let c3 = ColumnLocator::Pk {
        byte_off: 0,
        size: 8,
        type_code: TypeCode::I64,
    };
    let asc = |loc| OrderLocator { loc, desc: false, nulls_first: true };
    let own = OrderLocator { loc: c2, desc: true, nulls_first: false };
    assert_eq!(order_locators(&[written], &schema, false), vec![own]);
    assert_eq!(
        order_locators(&[written], &schema, true),
        vec![own, asc(c3), asc(c0), asc(c1), asc(c2)],
    );
}

/// NULL placement is absolute: `nulls_first` alone decides it, whatever the
/// direction, and two NULLs tie.
#[test]
fn null_placement_ignores_the_direction() {
    let schema = SchemaDescriptor::new(
        &[
            SchemaColumn::new(TypeCode::U64, false),
            SchemaColumn::new(TypeCode::I64, true),
        ],
        &[0],
    );
    let mut v = TestView::for_schema(&schema, 3);
    v.set_int(1, 0, 5);
    v.set_null(0, 0);
    v.set_null(2, 0);
    let loc = schema.locate(1);
    for nulls_first in [false, true] {
        let want = if nulls_first { Ordering::Less } else { Ordering::Greater };
        for desc in [false, true] {
            let keys = [OrderLocator { loc, desc, nulls_first }];
            assert_eq!(
                cmp_order_keys(&keys, &v, 0, 1),
                want,
                "nulls_first={nulls_first} desc={desc}"
            );
            assert_eq!(
                cmp_order_keys(&keys, &v, 1, 0),
                want.reverse(),
                "nulls_first={nulls_first} desc={desc}"
            );
            assert_eq!(cmp_order_keys(&keys, &v, 0, 2), Ordering::Equal, "two NULLs tie");
        }
    }
}

/// A PK key never reads the null word: with every bit set, it still orders by
/// value, where a NULL arm would call the rows tied.
#[test]
fn a_pk_key_never_takes_the_null_arm() {
    let schema = SchemaDescriptor::new(&[SchemaColumn::new(TypeCode::U64, false)], &[0]);
    // Keys 1 and 2.
    let mut v = TestView::for_schema(&schema, 2);
    for row in 0..2 {
        v.set_null_word(row, u64::MAX);
    }
    let loc = schema.locate(0);
    for desc in [false, true] {
        let keys = [OrderLocator { loc, desc, nulls_first: true }];
        let want = if desc { Ordering::Greater } else { Ordering::Less };
        assert_eq!(cmp_order_keys(&keys, &v, 0, 1), want, "desc={desc}");
    }
}

/// `RowRanking` against the comparator it stands in for, over `keys` on `v`: `sorted` is the
/// stable comparator sort, and under a total order `keep_smallest` keeps that sort's prefix.
fn assert_ranks_as_the_comparator(keys: &[OrderLocator], total: bool, v: &TestView, label: &str) {
    let n = v.row_count();
    let mut want: Vec<u32> = (0..n as u32).collect();
    want.sort_by(|&a, &b| cmp_order_keys(keys, v, a as usize, b as usize));
    assert_eq!(RowRanking::new(keys, v).sorted(), want, "{label}");
    if !total {
        return;
    }
    for k in [0, 1, n / 3, n - 1, n, n + 1] {
        let mut ranking = RowRanking::new(keys, v);
        ranking.keep_smallest(k);
        let mut kept: Vec<u32> = ranking.rows().collect();
        kept.sort_unstable();
        let mut prefix = want[..k.min(n)].to_vec();
        prefix.sort_unstable();
        assert_eq!(kept, prefix, "{label} k={k}");
        assert_eq!(ranking.sorted(), want[..k.min(n)], "{label} k={k}");
    }
}

/// A leading key of every type, as a payload column with NULLs on both sides of a tie and, where
/// the type allows, as a key column: in both directions and NULL placements, alone (its ties
/// left in input order) and under the identity tiebreak.
#[test]
fn a_ranking_orders_rows_as_the_comparator_does() {
    const ROWS: usize = 48;
    // Values that tie, that differ only in the low half of 16 bytes, and that cross each
    // width's sign bit.
    const PATTERNS: &[u128] = &[
        0,
        1,
        0x7F,
        0x80,
        0xFF,
        0x7FFF,
        0x8000,
        1 << 31,
        i64::MAX as u128,
        1 << 63,
        u64::MAX as u128,
        1 << 64,
        (1 << 64) + 1,
        i128::MAX as u128,
        1 << 127,
        u128::MAX,
    ];
    const STRINGS: &[&[u8]] = &[
        b"",
        b"a",
        b"a\0",
        b"abcdefgh",
        b"abcdefgh\0",
        b"abcdefghi",
        b"abcdefghijklmnop",
        b"abcdefghijklmnoq",
        b"b",
    ];
    let mut st = 0x9E3779B97F4A7C15u64;
    let mut rng = move || crate::test_support::xorshift(&mut st) as usize;
    for &tc in TypeCode::ALL {
        // `(key column 0, the leading key, a payload column)`, the leading key a payload column.
        let schema = SchemaDescriptor::new(
            &[
                SchemaColumn::new(TypeCode::U64, false),
                SchemaColumn::new(tc, true),
                SchemaColumn::new(TypeCode::I64, false),
            ],
            &[0],
        );
        let mut v = TestView::for_schema(&schema, ROWS);
        for row in 0..ROWS {
            match tc.is_german_string() {
                true => v.set_string(row, 0, STRINGS[rng() % STRINGS.len()]),
                false => v.set_native(&schema, row, 1, PATTERNS[rng() % PATTERNS.len()]),
            }
            if rng() % 5 == 0 {
                v.set_null(row, 0);
            }
            v.set_int(row, 1, (rng() % 3) as i64);
        }
        let mut cases = vec![(schema, v, 1u16)];
        if tc.is_pk_eligible() {
            // The leading key the first of two key columns, the second telling its ties apart.
            let schema = SchemaDescriptor::new(
                &[
                    SchemaColumn::new(tc, false),
                    SchemaColumn::new(TypeCode::U16, false),
                    SchemaColumn::new(TypeCode::I64, true),
                ],
                &[0, 1],
            );
            let mut v = TestView::for_schema(&schema, ROWS);
            for row in 0..ROWS {
                v.set_native(&schema, row, 0, PATTERNS[rng() % PATTERNS.len()]);
            }
            cases.push((schema, v, 0));
        }
        for (schema, v, col) in &cases {
            for (desc, nulls_first) in [(false, false), (false, true), (true, false), (true, true)] {
                let label = format!("{tc} column {col} desc={desc} nulls_first={nulls_first}");
                let first = OrderKey { col: *col, desc, nulls_first };
                assert_ranks_as_the_comparator(&order_locators(&[first], schema, false), false, v, &label);
                assert_ranks_as_the_comparator(&order_locators(&[first], schema, true), true, v, &label);
                // A second written key, ranked by its own image inside the first's ties.
                let second = OrderKey { col: 2, desc: !desc, nulls_first };
                assert_ranks_as_the_comparator(&order_locators(&[first, second], schema, false), false, v, &label);
            }
        }
    }
}