gnitz-zset 0.1.7

//! The payload row order — the (PK, payload) total order's second term — and the
//! per-schema choice between its two comparators, cached on the descriptor at
//! construction and read through [`with_payload_cmp!`].

use std::cmp::Ordering;

use super::SchemaDescriptor;
use gnitz_wire::RowSource;
use gnitz_wire::{cmp_col_window, null_word_get, read_unsigned_exact};

/// Compare two rows in the full (PK, payload) order.
pub(crate) fn compare_full_rows<A: RowSource, B: RowSource>(
    schema: &SchemaDescriptor,
    src_a: &A,
    row_a: usize,
    src_b: &B,
    row_b: usize,
) -> Ordering {
    src_a
        .get_pk_bytes(row_a)
        .cmp(src_b.get_pk_bytes(row_b))
        .then_with(|| compare_rows(schema, src_a, row_a, src_b, row_b))
}

/// Compare two rows from any [`RowSource`] implementations by payload columns.
///
/// Null words are read once per row outside the column loop.
#[inline]
pub(crate) fn compare_rows<A: RowSource, B: RowSource>(
    schema: &SchemaDescriptor,
    src_a: &A,
    row_a: usize,
    src_b: &B,
    row_b: usize,
) -> Ordering {
    let null_word_a = src_a.get_null_word(row_a);
    let null_word_b = src_b.get_null_word(row_b);
    // Both blob arenas are loop-invariant, and only the German-string arm of
    // `cmp_col_window` reads them, so they are hoisted out of the column loop.
    let (blob_a, blob_b) = (src_a.blob(), src_b.blob());

    for (payload_col, col) in schema.payload_columns() {
        let null_a = null_word_get(null_word_a, payload_col);
        let null_b = null_word_get(null_word_b, payload_col);
        if null_a && null_b {
            continue;
        }
        if null_a {
            return Ordering::Less;
        }
        if null_b {
            return Ordering::Greater;
        }

        let cs = col.size();
        let ord = cmp_col_window(
            src_a.get_col_ptr(row_a, payload_col, cs),
            blob_a,
            src_b.get_col_ptr(row_b, payload_col, cs),
            blob_b,
            col.type_code,
        );

        if ord != Ordering::Equal {
            return ord;
        }
    }

    Ordering::Equal
}

/// A payload row order. Generic over both sources, so one selected order serves
/// every source pairing a merge seat puts under it.
pub(crate) trait PayloadOrder: Copy {
    fn compare<A: RowSource, B: RowSource>(
        self,
        schema: &SchemaDescriptor,
        a: &A,
        ra: usize,
        b: &B,
        rb: usize,
    ) -> Ordering;
}

/// The order of a schema whose [`SchemaDescriptor::payload_is_fixed_int_nonnull`].
#[derive(Clone, Copy)]
pub(crate) struct FixedIntNonnull;
/// The order of every other schema.
#[derive(Clone, Copy)]
pub(crate) struct Generic;

impl PayloadOrder for FixedIntNonnull {
    /// Each column read as an order-preserving u64: zero-extend, then flip the
    /// sign bit for signed columns. Same order as [`compare_rows`], with no
    /// null-bitmap read and no per-column type dispatch.
    #[inline]
    fn compare<A: RowSource, B: RowSource>(self, s: &SchemaDescriptor, a: &A, ra: usize, b: &B, rb: usize) -> Ordering {
        debug_assert!(
            s.payload_is_fixed_int_nonnull(),
            "FixedIntNonnull on a non-fixedint or nullable schema",
        );
        for (payload_col, col) in s.payload_columns() {
            let cs = col.size();
            // `cs*8-1 ∈ {7,15,31,63}`: `FixedIntNonnull` admits only
            // `is_fixed_int` type codes, which are exactly the 1/2/4/8-byte
            // widths. `is_signed` is 0 for unsigned columns, so the XOR is then
            // a no-op; for signed ones it flips the MSB, putting two's-complement
            // negatives below non-negatives.
            let sign_flip = (col.is_signed() as u64) << (cs * 8 - 1);
            // `get_col_ptr` returns exactly `cs` bytes.
            let av = read_unsigned_exact(a.get_col_ptr(ra, payload_col, cs)) ^ sign_flip;
            let bv = read_unsigned_exact(b.get_col_ptr(rb, payload_col, cs)) ^ sign_flip;
            let ord = av.cmp(&bv);
            if ord != Ordering::Equal {
                return ord;
            }
        }
        Ordering::Equal
    }
}

impl PayloadOrder for Generic {
    #[inline]
    fn compare<A: RowSource, B: RowSource>(self, s: &SchemaDescriptor, a: &A, ra: usize, b: &B, rb: usize) -> Ordering {
        compare_rows(s, a, ra, b, rb)
    }
}

/// `f(args..., payload)` with the schema's [`PayloadOrder`] appended: one
/// monomorphization of `f` per order.
macro_rules! with_payload_cmp {
    ($schema:expr, $f:path $(, $arg:expr)* $(,)?) => {
        if $schema.payload_is_fixed_int_nonnull() {
            $f($($arg,)* $crate::schema::payload_order::FixedIntNonnull)
        } else {
            $f($($arg,)* $crate::schema::payload_order::Generic)
        }
    };
}
pub(crate) use with_payload_cmp;

#[cfg(test)]
#[path = "tests/payload_order.rs"]
mod tests;