rudb_kernels/

compare.rs

//! Comparing values, which is where SQL's three-valued logic actually lives.
//!
//! Six of the eight comparisons return null when either side is null, and the other two never do.
//! That is not a detail: `WHERE a = b` drops a row where either is null and `WHERE a IS NOT
//! DISTINCT FROM b` keeps the row where both are, and the binder produces the second one for `IS
//! NULL` and for a `USING` join under some rewrites. One enum with the null rule attached to the
//! variant is what stops that difference from being re-decided in every operator.
//!
//! The comparison enum here is this crate's own rather than `rudb_plan`'s, because the plan sits
//! nine ranks above the kernels and a kernel that imports a plan type is a kernel that cannot be
//! called from anywhere else. The executor maps one to the other, which is four lines it writes
//! once.
//!
//! Float comparison is DuckDB's rather than IEEE's. Two NaNs are equal, NaN sorts above every
//! number, and negative zero equals zero. IEEE says the first is false and that a NaN comparison is
//! unordered, which would make `GROUP BY` over a column with a NaN in it produce a group nothing
//! can ever find again and make a sort's result depend on the order the rows arrived in.
//!
//! # How the vectorized path is put together
//!
//! `spec/engine/03-data-plane.md` opens with this file as the example of what layer one is for.
//! What it used to be was a loop from zero to length calling `value_at` on both sides, comparing
//! two owned `Value`s and pushing into a `Vec<Value>` that a second pass then walked to pack into a
//! vector. On a varchar column that is a heap allocation and a memcpy per row per side, plus a
//! match on the operator inside the loop that the compiler has no way to hoist.
//!
//! What it is now is three decisions taken once per vector and then a loop that does one thing.
//!
//! The first decision is the form pair. Flat against flat, flat against constant and dictionary
//! against constant each get a hand written path, because those three are what a filter on a scan
//! actually produces. Constant on the left is the same code with the comparison turned around,
//! which [`Comparison::swapped`] does, so there is one loop rather than two. Everything else falls
//! through to the row at a time path, which is still here, is still correct, and now increments a
//! counter in [`crate::fallback`] on the way past so that a combination worth specializing shows up
//! as a number rather than as an opinion.
//!
//! The second decision is the physical type, which a macro turns into one loop per layout. Fifteen
//! layouts by three form pairs by eight operators written out by hand is how a wrong answer gets
//! in, and it is also four thousand lines nobody reads.
//!
//! The third decision is the operator, hoisted out of the loop once. The eight operators
//! become eight monomorphized loops over the same ordering, each with a comparison against a
//! constant `Ordering` in it, which is what makes the body a compare and a store.
//!
//! Validity gets its three cases used rather than collapsed. Two all valid sides skip the mask
//! entirely and produce an all valid result. Either side all invalid, on one of the six ordinary
//! comparisons, is every answer null without reading the data at all, which is a real case because
//! it is what a constant `NULL` in a predicate is.
//!
//! A conjunct that is not the first one does not need every row. [`refine`] is the same three
//! decisions with the output position mapped through the selection the conjuncts before it left, so
//! every loop in this file serves the threaded path without being written twice. The mapping is a
//! generic parameter rather than a function in a field, because an index mapping the compiler cannot
//! see through is an indirect call in a loop that is otherwise three instructions.
//!
//! Strings resolve from the four byte prefix in the view. Two views whose prefixes differ are in
//! that order, which holds because the payload past the end of a short string is zero and zero is
//! the least byte, so prefix order is byte order whenever the prefixes are not equal. On `hits` the
//! columns that carry the file are `URL` and `Referer`, and a filter on either of them is now a
//! four byte compare on almost every row instead of a `String` being built to be thrown away.
//!
//! # What is still slow here
//!
//! The index into each side goes through a closure so that the same macro serves flat, constant and
//! dictionary, which means the bounds check on each access survives. That is a known cost and it is
//! next to nothing beside the allocation it replaced, but it is the reason this file will not hit
//! the one nanosecond per row target on its own. The way out is a slice narrowed to the vector
//! length on the identity path, and that wants the benchmark suite to exist first so that the
//! change is a number rather than a belief.

use std::borrow::Cow;
use std::cell::RefCell;
use std::cmp::Ordering;
use std::sync::Arc;

use rudb_common::{Error, LogicalType, Result, Value, interval_micros};
use rudb_vector::{
    Coded, Data, Form, Packed, Selection, StringColumn, StringView, Validity, Vector,
};

use crate::fallback::{self, Kernel};
use crate::logic::is_true;
use crate::number::{approximate, integral};
use crate::peel::{self, Found};
use crate::prepare::Held;
use crate::shape::{first, identity, nulls_of, single};

/// Which comparison.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Comparison {
    /// `=`, null if either side is null.
    Equal,
    /// `<>`, null if either side is null.
    NotEqual,
    /// `<`, null if either side is null.
    Less,
    /// `<=`, null if either side is null.
    LessOrEqual,
    /// `>`, null if either side is null.
    Greater,
    /// `>=`, null if either side is null.
    GreaterOrEqual,
    /// `IS DISTINCT FROM`, which is total and never null.
    DistinctFrom,
    /// `IS NOT DISTINCT FROM`, which is total and never null.
    NotDistinctFrom,
}

impl Comparison {
    /// Whether this comparison treats null as a value rather than as an absence.
    #[must_use]
    pub fn is_total(self) -> bool {
        matches!(self, Self::DistinctFrom | Self::NotDistinctFrom)
    }

    /// The comparison that means the same thing with the two sides exchanged.
    ///
    /// This is what halves the number of specialized loops. A constant on the left against a
    /// column on the right is the column against the constant with the inequality turned around,
    /// and writing it that way means the column against constant loop is written once and tested
    /// once rather than twice with a chance of the second one being subtly wrong.
    #[must_use]
    pub fn swapped(self) -> Self {
        match self {
            Self::Less => Self::Greater,
            Self::LessOrEqual => Self::GreaterOrEqual,
            Self::Greater => Self::Less,
            Self::GreaterOrEqual => Self::LessOrEqual,
            same => same,
        }
    }

    /// Whether this comparison is true of two sides that sit in this order.
    ///
    /// For the six ordinary comparisons only. The two total ones read a null as a value and an
    /// `Ordering` has no way to say which side was null, so there is nothing sensible to return for
    /// them and they answer false rather than pretending.
    #[must_use]
    fn holds(self, order: Ordering) -> bool {
        match self {
            Self::Equal => order == Ordering::Equal,
            Self::NotEqual => order != Ordering::Equal,
            Self::Less => order == Ordering::Less,
            Self::LessOrEqual => order != Ordering::Greater,
            Self::Greater => order == Ordering::Greater,
            Self::GreaterOrEqual => order != Ordering::Less,
            Self::DistinctFrom | Self::NotDistinctFrom => false,
        }
    }
}

/// Compares two vectors of the same length, producing a `BOOLEAN` vector.
///
/// # Errors
///
/// If the two sides are not the same length, or if the two types cannot be compared.
pub fn compare(op: Comparison, left: &Vector, right: &Vector) -> Result<Vector> {
    compare_prepared(op, left, right, None)
}

/// [`compare`], with the constant side already turned into the column the loops read it through.
///
/// The same body and the same answer. A caller that built the plan knows which side is a literal
/// and can hand a [`Held`] built once for the query, which saves the allocations that building it
/// per chunk costs. A caller that has no plan in front of it passes `None` and nothing changes.
///
/// # Errors
///
/// The same ones [`compare`] gives.
pub fn compare_prepared(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    held: Option<&Held>,
) -> Result<Vector> {
    if left.len() != right.len() {
        return Err(Error::internal(format!(
            "a comparison of a {} row vector with a {} row one",
            left.len(),
            right.len()
        )));
    }
    let len = left.len();
    if left.form() == Form::Constant && right.form() == Form::Constant && len > 0 {
        let single = compare_values(op, &left.try_value_at(0)?, &right.try_value_at(0)?)?;
        return Ok(Vector::constant(LogicalType::Boolean, single, len));
    }

    let (left_valid, right_valid) = (nulls_of(left), nulls_of(right));
    // Either side entirely null, on one of the six ordinary comparisons, is every answer null and
    // the data is never read. This is not a corner case: a `NULL` literal in a predicate is a
    // constant vector whose validity is exactly this, and so is a column the scan knows is empty.
    if !op.is_total()
        && (left_valid == Validity::AllInvalid || right_valid == Validity::AllInvalid)
        && len > 0
    {
        return boolean(vec![false; len], Validity::AllInvalid, len);
    }

    if let Some(answers) = external_text_literal(op, left, right, len, identity, held)? {
        let validity = left_valid.and(&right_valid, len);
        return boolean(blank_the_nulls(answers, &validity), validity, len);
    }
    if let Some(answers) =
        specialized(op, left, right, &left_valid, &right_valid, len, identity, held)
    {
        let validity =
            if op.is_total() { Validity::AllValid } else { left_valid.and(&right_valid, len) };
        return boolean(blank_the_nulls(answers, &validity), validity, len);
    }

    fallback::record(Kernel::Compare, left.form(), right.form());
    let mut values = Vec::with_capacity(len);
    // row at a time: the path recorded on the line above, which exists to be correct for a pair of
    // forms no specialization covers and counts itself so that pair shows up in the report.
    for index in 0..len {
        values.push(compare_values(op, &left.try_value_at(index)?, &right.try_value_at(index)?)?);
    }
    Vector::from_values(LogicalType::Boolean, &values)
}

/// The rows the comparison is true on, which is [`compare_prepared`] and then
/// [`crate::select::selection`] without the flag vector in between.
///
/// The first conjunct of a filter asks exactly this, and going the long way round built a
/// `BOOLEAN` vector, wrote false into every null of it, and then had `selection` take the vector
/// apart again to find the booleans it had just been handed. Here the answers go straight to the
/// loop that picks the rows, and a null is dropped by the validity the way `selection` drops a
/// null flag, so the rows are the same ones. A pair of forms with no loop of its own goes the long
/// way, which is where it is counted.
///
/// # Errors
///
/// The same ones [`compare`] gives.
pub fn select_prepared(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    held: Option<&Held>,
) -> Result<Selection> {
    if left.len() != right.len() {
        return Err(Error::internal(format!(
            "a comparison of a {} row vector with a {} row one",
            left.len(),
            right.len()
        )));
    }
    let len = left.len();
    if len == 0 {
        return Ok(Selection::empty());
    }
    if left.form() == Form::Constant && right.form() == Form::Constant {
        let single = compare_values(op, &left.try_value_at(0)?, &right.try_value_at(0)?)?;
        return Ok(if is_true(&single) { Selection::identity(len) } else { Selection::empty() });
    }
    let (left_valid, right_valid) = (nulls_of(left), nulls_of(right));
    if !op.is_total() && (left_valid == Validity::AllInvalid || right_valid == Validity::AllInvalid)
    {
        return Ok(Selection::empty());
    }
    // A selection holds `u32` rows, and a chunk is nowhere near that, but the long way checks it
    // and so this does too rather than casting past it.
    if u32::try_from(len).is_ok() {
        if let Some(answers) = external_text_literal(op, left, right, len, identity, held)? {
            return Ok(crate::select::picked_flat(
                &answers[..len],
                &left_valid.and(&right_valid, len),
            ));
        }
        if left_valid == Validity::AllValid && right_valid == Validity::AllValid {
            if let Some(kept) = flat_kept(op, left, right, None, held) {
                return Ok(kept);
            }
            if let Some(kept) = packed_kept(op, left, right, None) {
                return Ok(kept);
            }
        }
        if let Some(answers) =
            specialized(op, left, right, &left_valid, &right_valid, len, identity, held)
        {
            let validity =
                if op.is_total() { Validity::AllValid } else { left_valid.and(&right_valid, len) };
            return Ok(crate::select::picked_flat(&answers[..len], &validity));
        }
    }
    Ok(crate::select::selection(&compare_prepared(op, left, right, held)?, len))
}

/// The rows of a flat integer column with no nulls that hold against a literal or against another
/// such column, in one pass.
///
/// The common shape of a scan's first filter, `AdvEngineID <> 0` on ClickBench Q1, and of the date
/// checks in TPC-H q4, q12 and q21, `l_commitdate < l_receiptdate`. The general path gets there in
/// two passes: an ordering per row into a run of booleans, through index closures the loop cannot
/// see past, and then [`crate::select::picked`] or [`narrowed`] over the booleans. Here the two
/// sides are compared in place and the row goes straight into the selection, so the loop is two
/// loads, a compare and a store, and it no longer writes a byte per row that is read once and
/// thrown away. `rows` is the rows the conjuncts before this one kept, or `None` for every row.
/// Floats are left to the general path, because their order is DuckDB's rather than the machine's.
fn flat_kept(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    rows: Option<&[u32]>,
    held: Option<&Held>,
) -> Option<Selection> {
    if op.is_total() || left.logical_type() != right.logical_type() {
        return None;
    }
    let len = left.len();
    let data = left.data()?;
    macro_rules! ordered {
        ($values:expr, $side:expr) => {{
            let (values, side) = ($values, $side);
            match op {
                Comparison::Equal => flat_where(rows, values, side, |a, b| a == b),
                Comparison::NotEqual => flat_where(rows, values, side, |a, b| a != b),
                Comparison::Less => flat_where(rows, values, side, |a, b| a < b),
                Comparison::LessOrEqual => flat_where(rows, values, side, |a, b| a <= b),
                Comparison::Greater => flat_where(rows, values, side, |a, b| a > b),
                Comparison::GreaterOrEqual => flat_where(rows, values, side, |a, b| a >= b),
                Comparison::DistinctFrom | Comparison::NotDistinctFrom => return None,
            }
        }};
    }
    macro_rules! layouts {
        ($($variant:ident),+ $(,)?) => {
            if let Some(others) = right.data() {
                match (data, others) {
                    $(
                        (Data::$variant(values), Data::$variant(others)) => {
                            let (values, others) = (values.get(..len)?, others.get(..len)?);
                            Some(ordered!(values, Side::Column(others)))
                        }
                    )+
                    _ => None,
                }
            } else {
                let column = readied(held, left.logical_type(), right.constant_value()?)?;
                match (data, column.data()?) {
                    $(
                        (Data::$variant(values), Data::$variant(wanted)) => {
                            let (values, wanted) = (values.get(..len)?, *wanted.first()?);
                            Some(ordered!(values, Side::Constant(wanted)))
                        }
                    )+
                    _ => None,
                }
            }
        };
    }
    layouts!(Int8, Int16, Int32, Int64, Int128, UInt8, UInt16, UInt32, UInt64, UInt128)
}

/// The right side of a comparison [`flat_where`] runs, another column or one value for every row.
#[derive(Clone, Copy)]
enum Side<'a, T> {
    Column(&'a [T]),
    Constant(T),
}

/// The rows of `values` for which `test` holds against `side`, out of `rows` when there are some.
///
/// With no `rows` this hands [`kept_in_blocks`] the slices themselves, so a block's flags come from
/// zipping 64 values with no index to check, which the compiler turns into vector compares. Through
/// a closure over a row number each row was a load behind a bounds check and a compare on its own.
fn flat_where<T: Copy>(
    rows: Option<&[u32]>,
    values: &[T],
    side: Side<'_, T>,
    test: impl Fn(T, T) -> bool,
) -> Selection {
    let len = values.len();
    match (rows, side) {
        (Some(rows), Side::Constant(wanted)) => {
            kept_where(Some(rows), len, |row| test(values[row], wanted))
        }
        (Some(rows), Side::Column(others)) => {
            kept_where(Some(rows), len, |row| test(values[row], others[row]))
        }
        (None, Side::Constant(wanted)) => kept_in_blocks(
            len,
            |base, flags| {
                for (flag, &value) in flags.iter_mut().zip(&values[base..base + 64]) {
                    *flag = u8::from(test(value, wanted));
                }
            },
            |row| test(values[row], wanted),
        ),
        (None, Side::Column(others)) => {
            let others = &others[..len];
            kept_in_blocks(
                len,
                |base, flags| {
                    let pairs = values[base..base + 64].iter().zip(&others[base..base + 64]);
                    for (flag, (&value, &other)) in flags.iter_mut().zip(pairs) {
                        *flag = u8::from(test(value, other));
                    }
                },
                |row| test(values[row], others[row]),
            )
        }
    }
}

/// The rows where one bit packed column with no nulls holds against another, in one pass.
///
/// This is `l_commitdate < l_receiptdate` and `l_shipdate < l_commitdate` in q4, q12 and q21 as a
/// stored table hands them up. [`packed_against_packed`] reads each code on its own, working out
/// the word and the straddle every time, and writes a flag per row for a second pass to pick the
/// rows out of. Here both sides are unpacked in blocks by [`Packed::unpack`], where the width is a
/// constant, and the comparison writes the selection directly. The two bases are folded into one
/// difference added to the right side, so the loop compares two `i64` rather than two `i128`. When
/// the conjuncts before this one left only a few rows, the codes of just those rows are read one at
/// a time instead, since unpacking the whole vector would read far more than it keeps.
///
/// `None` for anything but two straight packed runs of the same type, for codes too wide to leave
/// room for the difference, and for two ranges that do not meet, which the general path answers
/// without reading a code.
fn packed_kept(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    rows: Option<&[u32]>,
) -> Option<Selection> {
    if op.is_total() || left.logical_type() != right.logical_type() {
        return None;
    }
    let (one, other) = (left.packed_parts()?, right.packed_parts()?);
    if one.width() > 61 || other.width() > 61 {
        return None;
    }
    if one.ceiling() < other.base() || other.ceiling() < one.base() {
        return None;
    }
    // A row holds when `one.base() + a op other.base() + b`, which is `a op b + shift`.
    let shift = i64::try_from(other.base() - one.base()).ok()?;
    if shift.unsigned_abs() > 1 << 61 {
        return None;
    }
    let len = left.len();
    let dense = rows.is_none_or(|rows| rows.len() * 8 >= len);
    macro_rules! ordered {
        ($at:expr, $rows:expr) => {{
            let at = $at;
            match op {
                Comparison::Equal => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a == b
                }),
                Comparison::NotEqual => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a != b
                }),
                Comparison::Less => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a < b
                }),
                Comparison::LessOrEqual => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a <= b
                }),
                Comparison::Greater => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a > b
                }),
                Comparison::GreaterOrEqual => kept_where($rows, len, |row| {
                    let (a, b) = at(row);
                    a >= b
                }),
                Comparison::DistinctFrom | Comparison::NotDistinctFrom => return None,
            }
        }};
    }
    #[expect(
        clippy::cast_possible_wrap,
        reason = "both widths are at most 61 bits, so a code is well inside an i64"
    )]
    let kept = if dense {
        // Unpacked into two runs of `i64` with the difference already on the right, and then
        // compared as two flat columns are, so that a block of 64 is a zip the compiler does in
        // vector registers. Compared through a closure over the row, each row reloaded both
        // vectors behind the flag it had just stored, which could have been either of them for
        // all the compiler knew, and that was a fifth of q12.
        // Codes of 30 bits or fewer, moved by less than 2^30, fit an `i32` on both sides, and a
        // vector register holds twice as many of those and compares them signed in one step,
        // where a signed compare of `i64` lanes is put together out of several on SSE2. Either
        // way the codes go straight into their lanes a block at a time, rather than into a run of
        // `u64` zeroed first and then walked again into a second run, which was 6% of q12.
        if one.width() <= 30 && other.width() <= 30 && shift.unsigned_abs() < 1 << 30 {
            #[expect(
                clippy::cast_possible_truncation,
                reason = "a code is below 2^30 and the shift is below 2^30 either way"
            )]
            let (a, b) = {
                let (mut a, mut b) = (Vec::new(), Vec::new());
                one.unpack_mapped(0, len, &mut a, |code| code as i32);
                other.unpack_mapped(0, len, &mut b, |code| (code as i64 + shift) as i32);
                (a, b)
            };
            flat_by(op, rows, &a, &b)?
        } else {
            let (mut a, mut b) = (Vec::new(), Vec::new());
            one.unpack_mapped(0, len, &mut a, |code| code as i64);
            other.unpack_mapped(0, len, &mut b, |code| code as i64 + shift);
            flat_by(op, rows, &a, &b)?
        }
    } else {
        ordered!(|row: usize| (one.code(row) as i64, other.code(row) as i64 + shift), rows)
    };
    Some(kept)
}

/// One end of a range a filter asks for: the comparison that sets it, the literal it compares
/// against, and the column that literal was built into early, if one was.
#[derive(Clone, Copy, Debug)]
pub struct Bound<'a> {
    /// `>` or `>=` for the low end, `<` or `<=` for the high one.
    pub op: Comparison,
    /// The literal, in the column's own type.
    pub value: &'a Value,
    /// What the step that holds the literal built for it, see [`Held`].
    pub held: Option<&'a Held>,
}

/// The rows of `column` that lie between `low` and `high`, out of `live` when there are some, in one
/// pass over the column.
///
/// `l_shipdate >= date '1994-01-01' and l_shipdate < date '1995-01-01'` is two conjuncts, and each
/// of them on its own keeps most of the rows it is given, so running them one after the other pays
/// for a selection of half the chunk that the second one then walks again. Together they keep a
/// seventh of it. A value is in the range when its distance above the low end, read unsigned, is at
/// most the width of the range, which is one subtraction and one compare a row and a loop the
/// compiler does in vector registers. TPC-H q6 asks this twice and q4, q5, q10, q12, q14 and q15
/// ask it once. See `spec/perf/65-one-range-one-pass.md`.
///
/// `None` for anything this has no loop for, a column with nulls, a form other than flat or bit
/// packed, a type wider than 64 bits, a literal that is null or of another type, so that the caller
/// runs the two comparisons the way it would have.
#[must_use]
pub fn select_range(
    column: &Vector,
    low: Bound<'_>,
    high: Bound<'_>,
    live: Option<&[u32]>,
) -> Option<Selection> {
    let len = column.len();
    if len == 0 || u32::try_from(len).is_err() || nulls_of(column) != Validity::AllValid {
        return None;
    }
    // Asked before the literals are built, since building one is an allocation.
    let integral = column.packed_parts().is_some()
        || matches!(
            column.data(),
            Some(
                Data::Int8(_)
                    | Data::Int16(_)
                    | Data::Int32(_)
                    | Data::Int64(_)
                    | Data::UInt8(_)
                    | Data::UInt16(_)
                    | Data::UInt32(_)
                    | Data::UInt64(_)
            )
        );
    if !integral {
        return None;
    }
    let ty = column.logical_type();
    let literal = |bound: Bound<'_>| -> Option<i128> {
        if bound.value.is_null() {
            return None;
        }
        let single = readied(bound.held, ty, bound.value)?;
        if single.logical_type() != ty {
            return None;
        }
        Some(match single.data()? {
            Data::Int8(values) => i128::from(*values.first()?),
            Data::Int16(values) => i128::from(*values.first()?),
            Data::Int32(values) => i128::from(*values.first()?),
            Data::Int64(values) => i128::from(*values.first()?),
            Data::UInt8(values) => i128::from(*values.first()?),
            Data::UInt16(values) => i128::from(*values.first()?),
            Data::UInt32(values) => i128::from(*values.first()?),
            Data::UInt64(values) => i128::from(*values.first()?),
            _ => return None,
        })
    };
    // Both ends inclusive from here on.
    let low = match low.op {
        Comparison::GreaterOrEqual => literal(low)?,
        Comparison::Greater => literal(low)? + 1,
        _ => return None,
    };
    let high = match high.op {
        Comparison::LessOrEqual => literal(high)?,
        Comparison::Less => literal(high)? - 1,
        _ => return None,
    };
    if let Some(packed) = column.packed_parts() {
        return packed_range(&packed, len, low, high, live);
    }
    macro_rules! flat {
        ($($variant:ident: $signed:ty => $unsigned:ty),+ $(,)?) => {
            match column.data()? {
                $(
                    Data::$variant(values) => {
                        let values = values.get(..len)?;
                        let low = low.max(i128::from(<$signed>::MIN));
                        let high = high.min(i128::from(<$signed>::MAX));
                        if low > high {
                            return Some(Selection::empty());
                        }
                        #[expect(
                            clippy::cast_possible_truncation,
                            clippy::cast_sign_loss,
                            reason = "both ends were clamped to the type, and the distance is read \
                                      unsigned on purpose"
                        )]
                        let (low, span) = (low as $signed, (high - low) as $unsigned);
                        // Allowed rather than expected, since an unsigned type has no sign to lose.
                        #[allow(clippy::cast_sign_loss, reason = "the distance is read unsigned")]
                        let within = |value: $signed| value.wrapping_sub(low) as $unsigned <= span;
                        Some(match live {
                            Some(rows) => kept_where(Some(rows), len, |row| within(values[row])),
                            None => kept_in_blocks(
                                len,
                                |base, flags| {
                                    for (flag, &value) in
                                        flags.iter_mut().zip(&values[base..base + 64])
                                    {
                                        *flag = u8::from(within(value));
                                    }
                                },
                                |row| within(values[row]),
                            ),
                        })
                    }
                )+
                _ => None,
            }
        };
    }
    flat!(
        Int8: i8 => u8,
        Int16: i16 => u16,
        Int32: i32 => u32,
        Int64: i64 => u64,
        UInt8: u8 => u8,
        UInt16: u16 => u16,
        UInt32: u32 => u32,
        UInt64: u64 => u64,
    )
}

/// [`select_range`] on a bit packed column, in code space, where the range moves down by the base
/// and is cut to the codes the width can hold, or `None` for a width and base whose top code does
/// not fit an `i128`.
fn packed_range(
    packed: &Packed<'_>,
    len: usize,
    low: i128,
    high: i128,
    live: Option<&[u32]>,
) -> Option<Selection> {
    let (low, high) = (low.max(packed.base()), high.min(ceiling_of(packed)?));
    if low > high {
        return Some(Selection::empty());
    }
    #[expect(
        clippy::cast_possible_truncation,
        clippy::cast_sign_loss,
        reason = "both ends are inside the codes the width holds, so both differences fit a u64"
    )]
    let (from, span) = ((low - packed.base()) as u64, (high - low) as u64);
    let within = |code: u64| code.wrapping_sub(from) <= span;
    Some(match live {
        Some(rows) => kept_where(Some(rows), len, |row| within(packed.code(row))),
        None => kept_in_blocks(
            len,
            |base, flags| {
                let mut codes = [0_u64; 64];
                packed.unpack(base, &mut codes);
                for (flag, &code) in flags.iter_mut().zip(&codes) {
                    *flag = u8::from(within(code));
                }
            },
            |row| within(packed.code(row)),
        ),
    })
}

/// The rows where `a` stands in `op` to `b`, row by row, or none for the two operators that are
/// about nulls, which two flat runs of codes know nothing of.
fn flat_by<T: Copy + PartialOrd>(
    op: Comparison,
    rows: Option<&[u32]>,
    a: &[T],
    b: &[T],
) -> Option<Selection> {
    let side = Side::Column(b);
    Some(match op {
        Comparison::Equal => flat_where(rows, a, side, |x, y| x == y),
        Comparison::NotEqual => flat_where(rows, a, side, |x, y| x != y),
        Comparison::Less => flat_where(rows, a, side, |x, y| x < y),
        Comparison::LessOrEqual => flat_where(rows, a, side, |x, y| x <= y),
        Comparison::Greater => flat_where(rows, a, side, |x, y| x > y),
        Comparison::GreaterOrEqual => flat_where(rows, a, side, |x, y| x >= y),
        Comparison::DistinctFrom | Comparison::NotDistinctFrom => return None,
    })
}

/// The rows out of `rows`, or out of every row below `len` when there is no `rows`, that `held`
/// keeps, written without a branch on the answer.
#[inline]
fn kept_where(rows: Option<&[u32]>, len: usize, held: impl Fn(usize) -> bool) -> Selection {
    let Some(rows) = rows else {
        let fill = |base: usize, flags: &mut [u8; 64]| {
            for (bit, flag) in flags.iter_mut().enumerate() {
                *flag = u8::from(held(base + bit));
            }
        };
        return kept_in_blocks(len, fill, &held);
    };
    let mut kept = 0;
    let mut out = vec![0_u32; rows.len()];
    for &row in rows {
        out[kept] = row;
        kept += usize::from(held(row as usize));
    }
    out.truncate(kept);
    Selection::from_indices(out)
}

/// [`kept_where`] over every row below `len`, a block of 64 rows at a time.
///
/// The branchless loop writes each row and then moves the end of the answer by the flag, so every
/// row waits on the one before it through the count, and the compare can't be done in vector
/// registers. ClickBench Q1, `AdvEngineID <> 0` over a SMALLINT, spent 1.5ns a row there, and the
/// buffer the size of the chunk it zeroed first was another few percent. Here a block's 64 flags are
/// worked out on their own, which the compiler does in vector registers, and folded into a mask.
/// This is how ClickHouse filters a column: a block the mask says is empty is skipped, a full one
/// is copied as a range, and only a mixed block walks its set bits. `fill` writes the flags of the
/// 64 rows from the one it is given, and `held` answers for the rows after the last whole block.
#[expect(
    clippy::cast_possible_truncation,
    reason = "the caller checked that the row count fits in a u32 before getting here"
)]
#[inline]
pub(crate) fn kept_in_blocks(
    len: usize,
    fill: impl Fn(usize, &mut [u8; 64]),
    held: impl Fn(usize) -> bool,
) -> Selection {
    let mut out = Vec::with_capacity(len);
    let mut base = 0;
    while base + 64 <= len {
        let mut flags = [0_u8; 64];
        fill(base, &mut flags);
        let mut mask = 0_u64;
        for (lane, eight) in flags.chunks_exact(8).enumerate() {
            let bytes = u64::from_le_bytes(eight.try_into().unwrap_or_default());
            // Each byte is 0 or 1, and the multiply gathers byte i into bit 56 + i.
            mask |= (bytes.wrapping_mul(0x0102_0408_1020_4080) >> 56) << (lane * 8);
        }
        if mask == u64::MAX {
            out.extend(base as u32..(base + 64) as u32);
        } else {
            while mask != 0 {
                out.push((base + mask.trailing_zeros() as usize) as u32);
                mask &= mask - 1;
            }
        }
        base += 64;
    }
    out.extend((base..len).filter(|&index| held(index)).map(|index| index as u32));
    Selection::from_indices(out)
}

/// The rows of `kept` the comparison also keeps.
///
/// This is [`compare`] for a conjunct that is not the first one. A filter with four conjuncts
/// evaluated the obvious way runs all four over every row, so on TPC-H Q6, where each conjunct
/// passes about a fifth of the rows and the four together pass about two percent, the last conjunct
/// does fifty times the work it needs to. Handing it the rows the earlier ones kept is the whole
/// difference, and it is a difference that grows with the number of conjuncts rather than washing
/// out.
///
/// The answer is the rows of `kept`, in the order `kept` has them, for which the comparison is true.
/// Null is not true, so a row whose either side is null is dropped on the six ordinary comparisons,
/// which is the same rule [`crate::select::selection`] applies to a flag vector and the reason both
/// of them are a kernel rather than a line at the call site.
///
/// # Errors
///
/// If the two sides are not the same length, or if a position in `kept` is past the end of them.
pub fn refine(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    kept: &Selection,
) -> Result<Selection> {
    refine_prepared(op, left, right, kept, None)
}

/// [`refine`], with the constant side already built, for the reason [`compare_prepared`] gives.
///
/// This is the one that gains the most from it. A conjunct after the first reads the rows the ones
/// before it kept, so the loop can be eleven rows long while the setup is the same size it would be
/// for a full chunk.
///
/// # Errors
///
/// The same ones [`refine`] gives.
pub fn refine_prepared(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    kept: &Selection,
    held: Option<&Held>,
) -> Result<Selection> {
    if left.len() != right.len() {
        return Err(Error::internal(format!(
            "a comparison of a {} row vector with a {} row one",
            left.len(),
            right.len()
        )));
    }
    let len = left.len();
    // One vectorized pass over a run of `u32` before any of the loops below index with them, which
    // is what turns a caller's mistake into this message rather than into a panic from inside a
    // macro generated loop eight frames down.
    if !rudb_vector::below(kept.indices(), len) {
        return Err(Error::internal(format!("a selection past the end of a {len} row vector")));
    }
    if kept.is_empty() {
        return Ok(Selection::empty());
    }
    if left.form() == Form::Constant && right.form() == Form::Constant {
        let single = compare_values(op, &left.try_value_at(0)?, &right.try_value_at(0)?)?;
        return Ok(if is_true(&single) { kept.clone() } else { Selection::empty() });
    }

    let (left_valid, right_valid) = (nulls_of(left), nulls_of(right));
    if !op.is_total() && (left_valid == Validity::AllInvalid || right_valid == Validity::AllInvalid)
    {
        return Ok(Selection::empty());
    }

    let rows = kept.indices();
    if left_valid == Validity::AllValid && right_valid == Validity::AllValid {
        if let Some(kept) = flat_kept(op, left, right, Some(rows), held) {
            return Ok(kept);
        }
        if let Some(kept) = packed_kept(op, left, right, Some(rows)) {
            return Ok(kept);
        }
    }
    let map = |slot: usize| rows[slot] as usize;
    if let Some(answers) = external_text_literal(op, left, right, kept.len(), map, held)? {
        return Ok(narrowed(&answers, rows, |slot| {
            let row = rows[slot] as usize;
            left_valid.is_valid(row) && right_valid.is_valid(row)
        }));
    }
    if let Some(answers) =
        specialized(op, left, right, &left_valid, &right_valid, kept.len(), map, held)
    {
        // A total comparison has the nulls in the answer already, and two all valid sides have no
        // null to drop, so both of those get the loop with nothing in it but the flag.
        if op.is_total() || (left_valid == Validity::AllValid && right_valid == Validity::AllValid)
        {
            return Ok(narrowed(&answers, rows, |_| true));
        }
        // A bit at a time rather than a word at a time, which is the one place this path gives up
        // something `compare` has. The rows are scattered by construction, so the two mask reads for
        // one row are in different words as often as not and a word oriented loop would reread them.
        return Ok(narrowed(&answers, rows, |slot| {
            let row = rows[slot] as usize;
            left_valid.is_valid(row) && right_valid.is_valid(row)
        }));
    }

    fallback::record(Kernel::Compare, left.form(), right.form());
    let mut out = Vec::with_capacity(kept.len());
    // row at a time: the path recorded on the line above, for a pair of forms no specialization
    // covers, reading only the rows the conjuncts before this one kept.
    for &row in rows {
        let index = row as usize;
        if is_true(&compare_values(op, &left.try_value_at(index)?, &right.try_value_at(index)?)?) {
            out.push(row);
        }
    }
    Ok(Selection::from_indices(out))
}

/// One of the six ordinary comparisons between storage-backed text and a literal, without
/// constructing row values.
///
/// Where the comparison is an equality, the column is a dictionary that shares its values and the
/// caller brought the literal it was built with, this decides once per distinct value instead of
/// once per row. See the `peel` module. Everything else reads the column a row at a time, which is
/// still better than the general path because it never builds a value.
///
/// An ordering comparison gets none of that and is here anyway, because what the general path costs
/// on a text column is not the comparison. It is `try_value_at`, which allocates a `String` a row so
/// that `compare_values` has a `Value` to look at, and then drops it. Reading the bytes where they
/// lie and comparing those is the same answer with neither the allocation nor the dispatch, and the
/// caller this matters most to is the top N: it asks every chunk whether any row can still beat the
/// worst candidate it holds, and the constant it asks about changes as it goes, so nothing is memoized
/// and the row loop is the whole of it.
fn external_text_literal<M>(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    len: usize,
    map: M,
    held: Option<&Held>,
) -> Result<Option<Vec<bool>>>
where
    M: Fn(usize) -> usize + Copy,
{
    if op.is_total()
        || left.logical_type() != &LogicalType::Varchar
        || right.logical_type() != &LogicalType::Varchar
    {
        return Ok(None);
    }
    let (column, literal, swapped) = match (left.constant_value(), right.constant_value()) {
        (None, Some(Value::Varchar(literal))) if left.positions().is_some() => {
            (left, literal.as_bytes(), false)
        }
        (Some(Value::Varchar(literal)), None) if right.positions().is_some() => {
            (right, literal.as_bytes(), true)
        }
        _ => return Ok(None),
    };
    // The column is on the left from here on, so an inequality written the other way round is
    // turned around once rather than once a row.
    let op = if swapped { op.swapped() } else { op };
    let same = op == Comparison::Equal;
    if matches!(op, Comparison::Equal | Comparison::NotEqual) {
        // The literal has to be the one the memo was filled against, which it is when the caller
        // took both from the same comparison node. A caller that gets it wrong is slow rather than
        // wrong, which is the rule the rest of `Held` keeps.
        if let Some(held) = held.filter(|held| held.text() == Some(literal)) {
            // A dictionary that came with its sorted order answers this without reading any value
            // more than the search does, so try that before filling a memo one value at a time.
            if let Some(found) = held.lookup().find(column, literal) {
                return Ok(Some(against_code(column, found?, len, map, same)?));
            }
            let decide = |dictionary: &Vector, code: usize| -> Result<bool> {
                let found = if literal.is_empty() {
                    dictionary.try_bytes_len_at(code)?.is_some_and(|length| length == 0)
                } else {
                    dictionary.try_bytes_at(code)?.is_some_and(|bytes| bytes == literal)
                };
                Ok(found)
            };
            if let Some(answers) = held.peel().answer(column, len, map, decide) {
                let mut answers = answers?;
                if !same {
                    for answer in &mut answers {
                        *answer = !*answer;
                    }
                }
                return Ok(Some(answers));
            }
        }
        let mut answers = Vec::with_capacity(len);
        for slot in 0..len {
            let row = map(slot);
            // An empty literal is settled by the length alone, which for a dictionary is one load
            // of two offsets rather than a walk to wherever the value's bytes live.
            let equal = if literal.is_empty() {
                column.try_bytes_len_at(row)?.is_some_and(|length| length == 0)
            } else {
                column.try_bytes_at(row)?.is_some_and(|bytes| bytes == literal)
            };
            answers.push(equal == same);
        }
        return Ok(Some(answers));
    }
    if let Some(answers) = by_rank(op, column, literal, len, map)? {
        return Ok(Some(answers));
    }
    let mut answers = Vec::with_capacity(len);
    for slot in 0..len {
        let row = map(slot);
        // A null row's bytes are whatever the column left there, and the answer for it is thrown
        // away by the caller, which blanks every position the validity says is null. Equal is what
        // is written there because it is the cheapest thing to write and it is never read.
        let order = match column.try_bytes_at(row)? {
            Some(bytes) => bytes.cmp(literal),
            None => Ordering::Equal,
        };
        answers.push(op.holds(order));
    }
    Ok(Some(answers))
}

/// An inequality against a literal answered out of the dictionary's sorted order, or `None` when
/// there is no order to answer it from.
///
/// The equality path has resolved a literal to a code and compared integers since the sorted order
/// was written, and this is the same trick for the other four comparisons. One binary search puts the
/// literal at a rank boundary, the inverse of the order turns each row's code into its rank, and then
/// the whole comparison is one `usize` against another. No value is read and no bytes are compared.
///
/// The caller this was written for is the top N. `ORDER BY <varchar> LIMIT 10` asks every chunk
/// whether any row in it can still beat the tenth best candidate, which is a comparison of the key
/// column against a literal that changes as the query runs, so there is nothing to memoize and the
/// row loop is the whole of the operator. On ClickBench 25, which is that query over `SearchPhrase`,
/// the top N was 0.377 of the 0.791 milliseconds the pipeline's operators spent per instance, against
/// 0.097 for the same query ordered by a timestamp. The gap was the byte compare.
///
/// The filter gets it too, for `WHERE <varchar> < 'literal'` and for both halves of a `BETWEEN`.
///
/// Two things have to be true and the source decides both. It has to know its own order, and it has
/// to be willing to hand the order back inverted, which is a `u32` per value it builds once and keeps.
/// A source that would rather not answers `None` to one of them and the byte loop below runs instead,
/// which is what every in memory column does today.
fn by_rank<M>(
    op: Comparison,
    column: &Vector,
    literal: &[u8],
    len: usize,
    map: M,
) -> Result<Option<Vec<bool>>>
where
    M: Fn(usize) -> usize,
{
    let Some((codes, dictionary)) = column.shared_dictionary_parts() else { return Ok(None) };
    let Some(ranks) = dictionary.ranks() else { return Ok(None) };
    let Some(order) = dictionary.code_ranks() else { return Ok(None) };
    let (below, equal) = peel::below(dictionary, ranks, literal)?;
    // The two boundaries `peel::below` describes, picked by which side of the literal the comparison
    // wants and whether the literal itself counts as being on that side.
    let cut = match op {
        Comparison::Less | Comparison::GreaterOrEqual => below,
        _ => below + usize::from(equal),
    };
    let under = matches!(op, Comparison::Less | Comparison::LessOrEqual);
    let mut answers = Vec::with_capacity(len);
    // row at a time: the comparison is the loop, and a row of it is two loads and an integer compare.
    for slot in 0..len {
        let code = *codes
            .get(map(slot))
            .ok_or_else(|| Error::internal("a ranked row is past the end of its codes"))?;
        let rank = *order
            .get(code as usize)
            .ok_or_else(|| Error::internal("a ranked code is past the end of its dictionary"))?;
        answers.push(((rank as usize) < cut) == under);
    }
    Ok(Some(answers))
}

/// The rows that compare true against the value already known to sit at `rank`, or `None`.
///
/// The `by_rank` path with the search taken out. That path searches because it is handed a literal
/// and has to find out where the literal sits, and the search is not cheap: about nineteen probes of
/// a dictionary of a hundred thousand, and a probe that cannot settle on the eight bytes the file
/// stores per rank has to read a value, which decodes the block the value sits in. On ClickBench 25
/// that search and the block decoding under it were most of the query.
///
/// A caller that already knows the rank pays none of it. The one this was written for is the top N,
/// whose bound is not a literal from the query at all: it is a value that came out of this same
/// dictionary, carried by a row that arrived with its code, so its rank was known the moment it was
/// kept and nothing has to be found. See `crate::topn`.
///
/// `None` when the column is not a dictionary over `dictionary`, when that dictionary will not hand
/// its order back inverted, or for a comparison that is not one of the four inequalities. The
/// identity check is the whole of what makes this safe to offer: a rank means nothing except against
/// the dictionary it was read out of, so the caller hands that dictionary over and this refuses
/// rather than trusting it.
///
/// Null rows are dropped, which is what [`crate::select::selection`] does with a null flag and
/// therefore what the caller would have got by going the long way round.
#[must_use]
pub fn select_against_rank(
    op: Comparison,
    column: &Vector,
    dictionary: &Arc<Vector>,
    rank: u32,
    rows: usize,
) -> Option<Selection> {
    let (codes, values) = column.shared_dictionary_parts()?;
    if !Arc::ptr_eq(values, dictionary) {
        return None;
    }
    let order = values.code_ranks()?;
    let validity = column.validity();
    let live = !validity.has_nulls(rows);
    let rows = rows.min(codes.len());
    let mut kept = vec![0_u32; rows];
    let mut count = 0;
    // row at a time: the comparison is the loop, and a row of it is two loads and an integer
    // compare. Every slot writes its row at the current length and only a kept slot moves the
    // length on, for the reason `narrowed` gives.
    for (row, &code) in codes.iter().enumerate().take(rows) {
        let at = *order.get(code as usize)?;
        let held = match op {
            Comparison::Less => at < rank,
            Comparison::LessOrEqual => at <= rank,
            Comparison::Greater => at > rank,
            Comparison::GreaterOrEqual => at >= rank,
            _ => return None,
        };
        kept[count] = u32::try_from(row).ok()?;
        // A single `&` rather than `&&`, because the short circuit would put back the branch.
        count += usize::from(held & (live || validity.is_valid(row)));
    }
    kept.truncate(count);
    Some(Selection::from_indices(kept))
}

/// The rank of the value one row of a dictionary column holds, when that is knowable.
///
/// The other half of [`select_against_rank`]. A caller that wants to compare against a row's value
/// later without searching for it asks for this when the row goes by and keeps the answer.
///
/// `None` for a column that is not a dictionary over a source that knows its order, for a row that
/// is null, and for a code the order does not cover. The caller treats all three the same way, by
/// giving up on ranks and doing what it did before.
#[must_use]
pub fn rank_at(column: &Vector, row: usize) -> Option<(Arc<Vector>, u32)> {
    let (codes, values) = column.shared_dictionary_parts()?;
    if !column.validity().is_valid(row) {
        return None;
    }
    let order = values.code_ranks()?;
    let code = *codes.get(row)?;
    let rank = *order.get(code as usize)?;
    Some((Arc::clone(values), rank))
}

/// The rank of one row's value in a dictionary the caller is already holding.
///
/// [`rank_at`] for a caller that has a dictionary in hand and wants to know where one row sits in
/// it, which is a top n rejecting a row against a candidate it kept earlier. It answers the question
/// without the clone of the `Arc` that `rank_at` has to make, because a reject runs on every row
/// that reaches the operator and an atomic increment per row is not nothing.
///
/// The identity check is what makes a rank mean anything, and it is the same one
/// [`select_against_rank`] makes: a rank is a position in one dictionary and says nothing at all
/// about any other, so this refuses a column that is not over the dictionary it was given.
///
/// `None` in every case `rank_at` answers `None` in, and additionally for a column over a different
/// dictionary. The caller reads the value and compares it the old way.
#[must_use]
pub fn rank_within(column: &Vector, row: usize, dictionary: &Arc<Vector>) -> Option<u32> {
    let (codes, values) = column.shared_dictionary_parts()?;
    if !Arc::ptr_eq(values, dictionary) || !column.validity().is_valid(row) {
        return None;
    }
    let order = values.code_ranks()?;
    let code = *codes.get(row)?;
    Some(*order.get(code as usize)?)
}

/// Every row's answer once the literal has been resolved to a code, or to nothing.
///
/// This is the whole point of storing a dictionary's sorted order. The comparison is a `u32`
/// against a `u32` and it never touches the payload, so a filter on a text column costs what a
/// filter on an integer column costs. A literal the dictionary does not hold is decided for the
/// whole chunk without looking at the codes at all, because a code that is in the dictionary cannot
/// be the one that is not.
fn against_code<M>(
    column: &Vector,
    found: Found,
    len: usize,
    map: M,
    same: bool,
) -> Result<Vec<bool>>
where
    M: Fn(usize) -> usize,
{
    let Found::At(wanted) = found else { return Ok(vec![!same; len]) };
    let (codes, _) = column
        .shared_dictionary_parts()
        .ok_or_else(|| Error::internal("a resolved literal lost the codes it was resolved for"))?;
    // Every row the map can name is a row of the column, which both callers check before they get
    // here, so codes as long as the column settle the bound for the whole chunk at once. Asking per
    // row put an error path and a push in a loop that is otherwise one compare, and that loop was
    // fifteen instructions a row on ClickBench 28 rather than one.
    if codes.len() < column.len() {
        return Err(Error::internal("a compared column is longer than its codes"));
    }
    // row at a time: the comparison is the loop. Nothing here reads a value or allocates.
    Ok((0..len).map(|slot| (codes[map(slot)] == wanted) == same).collect())
}

/// The positions of `rows` whose answer is true and whose row is live, without a branch per row.
///
/// The same shape as the loop in `crate::select` and for the same reason: which rows a filter keeps is
/// what the data decides rather than what the code does, so the branch is unpredictable by
/// construction and a mispredict is worth more than the rest of the loop put together. Every slot
/// writes its row at the current length and only a slot that is kept moves the length on.
fn narrowed<L: Fn(usize) -> bool>(answers: &[bool], rows: &[u32], live: L) -> Selection {
    let mut out = vec![0_u32; answers.len()];
    let mut count = 0;
    for (slot, &answer) in answers.iter().enumerate() {
        out[count] = rows[slot];
        // A single `&` rather than `&&`, because the short circuit would put back the branch.
        count += usize::from(answer & live(slot));
    }
    out.truncate(count);
    Selection::from_indices(out)
}

/// A `BOOLEAN` vector from a run of answers and the validity that says which of them count.
fn boolean(answers: Vec<bool>, validity: Validity, len: usize) -> Result<Vector> {
    // An empty vector has no null to record, and `Vector::from_values` normalizes the empty mask it
    // builds to all valid, so saying the same here is what keeps an empty specialized result the
    // same vector as the oracle's rather than merely the same length.
    let validity = if len == 0 { Validity::AllValid } else { validity.normalize(len) };
    Ok(Vector::flat(LogicalType::Boolean, Data::Bool(answers.into()))?.with_validity(validity))
}

/// A false in every position the validity says is null.
///
/// The comparison at a null position read whatever the zero the null was stored as compared to,
/// which is a defined value and a meaningless one. Writing false there costs one pass over a run
/// of bytes, only when there are nulls at all, and it buys the property that a specialized result
/// is the same vector as the row at a time result rather than merely the same answer. A test that
/// can compare two vectors with `==` is a much better test than one that has to walk them.
fn blank_the_nulls(mut answers: Vec<bool>, validity: &Validity) -> Vec<bool> {
    if let Validity::Mask(mask) = validity {
        for (index, answer) in answers.iter_mut().enumerate() {
            if !mask.get(index) {
                *answer = false;
            }
        }
    }
    answers
}

/// The answers for a form pair this file has a loop for, or `None` to say it has not.
///
/// `map` turns an output position into the row of `left` and `right` it is the answer for, and
/// `len` is how many output positions there are. [`compare`] passes [`identity`] and the length of
/// its operands, which is every row. [`refine`] passes the selection it was handed and the size of
/// it, which is how a conjunct after the first reads only the rows the conjuncts before it kept.
///
/// A generic parameter rather than a `fn(usize) -> usize` in a field, for the reason
/// `spec/engine/03-data-plane.md` records as the first performance lesson of this layer: an index
/// mapping the compiler cannot see through is an indirect call per row, and one of those in a loop
/// that is otherwise three instructions is the whole loop.
#[expect(
    clippy::too_many_arguments,
    reason = "two sides, two validities, the operator, the length, the index mapping and the \
              literal that was built early, all of which the branches below need"
)]
fn specialized<M>(
    op: Comparison,
    left: &Vector,
    right: &Vector,
    left_valid: &Validity,
    right_valid: &Validity,
    len: usize,
    map: M,
    held: Option<&Held>,
) -> Option<Vec<bool>>
where
    M: Fn(usize) -> usize + Copy,
{
    // Across representations is the fallback's job. `INTEGER` against `BIGINT` reaches the same
    // answer through `numeric_order`, and a specialized loop that assumed the two runs had the same
    // layout would compare a four byte column against an eight byte one position by position.
    if left.logical_type() != right.logical_type() {
        return None;
    }
    // A bit string's bytes do not sort the way its bits do, so ordering two of them is the
    // fallback's, which asks `bit::cmp`. Equality is still the bytes, since a bit string has one
    // layout.
    let equality = matches!(
        op,
        Comparison::Equal
            | Comparison::NotEqual
            | Comparison::DistinctFrom
            | Comparison::NotDistinctFrom
    );
    if *left.logical_type() == LogicalType::Bit && !equality {
        return None;
    }

    // Where each side keeps its values and how a row of it is reached, which is what turns flat,
    // dictionary and run length into one branch below rather than nine. See [`Through`].
    let (one, other) = (through(left), through(right));

    if let (Some(one), Some(other)) = (&one, &other) {
        return match (one, other) {
            (Through::Direct(a), Through::Direct(b)) => {
                dispatch(op, len, a, map, b, map, left_valid, right_valid, map)
            }
            (Through::Coded(codes, a), Through::Direct(b)) => dispatch(
                op,
                len,
                a,
                |slot| codes[map(slot)] as usize,
                b,
                map,
                left_valid,
                right_valid,
                map,
            ),
            (Through::Direct(a), Through::Coded(codes, b)) => dispatch(
                op,
                len,
                a,
                map,
                b,
                |slot| codes[map(slot)] as usize,
                left_valid,
                right_valid,
                map,
            ),
            (Through::Coded(left_codes, a), Through::Coded(right_codes, b)) => dispatch(
                op,
                len,
                a,
                |slot| left_codes[map(slot)] as usize,
                b,
                |slot| right_codes[map(slot)] as usize,
                left_valid,
                right_valid,
                map,
            ),
        };
    }
    // A bit packed column against a literal, which is the pair the form was added for. The literal
    // is turned into a code once and then the loop compares codes, so nothing is unpacked at all,
    // and a literal outside what the width can hold answers the whole vector without a bit of it
    // being read. Only the six comparisons that go null on a null side come here, because the other
    // two want the null rule inside the loop and this loop does not have it.
    if !op.is_total() {
        if let (Some(packed), Some(value)) = (left.packed_parts(), right.constant_value()) {
            let wanted = exact(held, left.logical_type(), value)?;
            return Some(packed_against(op, &packed, wanted, len, map));
        }
        if let (Some(value), Some(packed)) = (left.constant_value(), right.packed_parts()) {
            let wanted = exact(held, right.logical_type(), value)?;
            return Some(packed_against(op.swapped(), &packed, wanted, len, map));
        }
        if let (Some(one), Some(other)) = (packing(left), packing(right)) {
            return packings_against_each_other(op, &one, &other, len, map);
        }
        // A bit packed column against a flat one, which is the pair a clustered table hands the
        // filter. Inside a narrow partition a date column takes few enough distinct values to pack
        // and the column beside it does not, so the better encoding the clustering buys is what
        // used to send the comparison to the row at a time path. Both sides are read where they
        // are, the packed one as its base plus its code and the flat one as the number it already
        // is. On TPC-H this is `l_commitdate < l_receiptdate` on the clustered file, which is q4
        // and q21.
        if let (Some(one), Some(other)) = (packing(left), &other) {
            return packing_against_run(op, &one, other, len, map);
        }
        if let (Some(one), Some(other)) = (&one, packing(right)) {
            return packing_against_run(op.swapped(), &other, one, len, map);
        }
    }
    if let (Some(one), Some(value)) = (&one, right.constant_value()) {
        let column = readied(held, left.logical_type(), value)?;
        let other = column.data()?;
        return match one {
            Through::Direct(a) => {
                dispatch(op, len, a, map, other, first, left_valid, right_valid, map)
            }
            Through::Coded(codes, a) => dispatch(
                op,
                len,
                a,
                |slot| codes[map(slot)] as usize,
                other,
                first,
                left_valid,
                right_valid,
                map,
            ),
        };
    }
    if let (Some(value), Some(other)) = (left.constant_value(), &other) {
        // The same loop with the comparison turned around, rather than a second loop.
        let column = readied(held, right.logical_type(), value)?;
        let one = column.data()?;
        return match other {
            Through::Direct(b) => {
                dispatch(op.swapped(), len, b, map, one, first, right_valid, left_valid, map)
            }
            Through::Coded(codes, b) => dispatch(
                op.swapped(),
                len,
                b,
                |slot| codes[map(slot)] as usize,
                one,
                first,
                right_valid,
                left_valid,
                map,
            ),
        };
    }
    // A compressed column against a literal, tested in the code space the column is already in.
    // Only equality, because a symbol code says nothing about where its symbol sorts, so an ordering
    // comparison has to decompress and does. Equality does not: compressing is a function of the
    // table and the bytes, so two strings have the same codes exactly when they are the same string.
    if matches!(op, Comparison::Equal | Comparison::NotEqual) {
        if let (Some(coded), Some(value)) = (left.coded_parts(), right.constant_value()) {
            let wanted = encoded(&coded, held, left.logical_type(), value)?;
            return Some(coded_against(op, &coded, &wanted, len, map));
        }
        if let (Some(value), Some(coded)) = (left.constant_value(), right.coded_parts()) {
            let wanted = encoded(&coded, held, right.logical_type(), value)?;
            return Some(coded_against(op, &coded, &wanted, len, map));
        }
    }
    // A string column against another one or against a literal, with the views read where they are.
    // It catches the string view form, whose bytes live in an arena the vector shares and so has no
    // data slice for the branches above to find, and it catches the flat form as well so that the
    // two cannot be compared by two different loops. The order itself is the one `view_order`
    // writes down either way.
    if let (Some((one, one_arena)), Some((other, other_arena))) =
        (left.text_parts(), right.text_parts())
    {
        return Some(sweep(
            op,
            len,
            |index| view_order(one.get(map(index)), one_arena, other.get(map(index)), other_arena),
            left_valid,
            right_valid,
            map,
        ));
    }
    // A string column against a literal. The literal becomes one view before the loop starts, so
    // every row is a four byte prefix against the same four bytes and the payload is only read for
    // the rows the prefix could not settle.
    if let (Some((one, one_arena)), Some(value)) = (left.text_parts(), right.constant_value()) {
        let column = readied(held, left.logical_type(), value)?;
        let (other, other_arena) = column.text_parts()?;
        let wanted = other.first();
        return Some(sweep(
            op,
            len,
            |index| view_order(one.get(map(index)), one_arena, wanted, other_arena),
            left_valid,
            right_valid,
            map,
        ));
    }
    if let (Some(value), Some((other, other_arena))) = (left.constant_value(), right.text_parts()) {
        // The same loop with the comparison turned around, rather than a second loop.
        let column = readied(held, right.logical_type(), value)?;
        let (one, one_arena) = column.text_parts()?;
        let wanted = one.first();
        return Some(sweep(
            op.swapped(),
            len,
            |index| view_order(other.get(map(index)), other_arena, wanted, one_arena),
            right_valid,
            left_valid,
            map,
        ));
    }
    if let (Some((codes, values)), Some(value)) = (left.positions(), right.constant_value()) {
        let one = values.data()?;
        let column = readied(held, left.logical_type(), value)?;
        let other = column.data()?;
        let at = |index: usize| codes[map(index)] as usize;
        return dispatch(op, len, one, at, other, first, left_valid, right_valid, map);
    }
    if let (Some(value), Some((codes, values))) = (left.constant_value(), right.positions()) {
        let other = values.data()?;
        let column = readied(held, right.logical_type(), value)?;
        let one = column.data()?;
        let at = |index: usize| codes[map(index)] as usize;
        return dispatch(op.swapped(), len, other, at, one, first, right_valid, left_valid, map);
    }
    // A dictionary against a flat column. This pair had no loop until the kernel table put a number
    // on what that cost, which on `server3` was 83 nanoseconds a row against 1.2 for the dictionary
    // against constant pair beside it, on the same data and the same operator. It is not a rare
    // shape either: it is what a filtered column compared against an unfiltered one is, which is
    // every conjunct after the first.
    if let (Some((codes, values)), Some(other)) = (left.positions(), right.data()) {
        let one = values.data()?;
        let at = |index: usize| codes[map(index)] as usize;
        return dispatch(op, len, one, at, other, map, left_valid, right_valid, map);
    }
    if let (Some(one), Some((codes, values))) = (left.data(), right.positions()) {
        let other = values.data()?;
        let at = |index: usize| codes[map(index)] as usize;
        return dispatch(op.swapped(), len, other, at, one, map, right_valid, left_valid, map);
    }
    None
}

/// A literal as the whole number it is, and `None` for one that is not a whole number.
///
/// It goes through [`readied`] rather than reading the [`Value`] apart, so that a literal written
/// as `900` against a `SMALLINT` column is narrowed by the same cast path every other comparison
/// narrows it with. Reading the value apart here would be a second cast path with its own rounding
/// and its own overflow rule, which is how two comparisons of the same literal end up disagreeing.
fn exact(held: Option<&Held>, ty: &LogicalType, value: &Value) -> Option<i128> {
    let column = readied(held, ty, value)?;
    let data = column.data()?;
    data.signed_at(0).or_else(|| data.unsigned_at(0).and_then(|value| i128::try_from(value).ok()))
}

/// A literal in the code space a compressed column is in, and `None` for one with no bytes.
///
/// It goes through [`readied`] for the reason [`exact`] does: the literal is narrowed to the column
/// type by the same path every other comparison narrows it with, rather than by a second reading of
/// the [`Value`] that could disagree with the first.
fn encoded(
    coded: &Coded<'_>,
    held: Option<&Held>,
    ty: &LogicalType,
    value: &Value,
) -> Option<Vec<u8>> {
    let column = readied(held, ty, value)?;
    let (views, arena) = column.text_parts()?;
    Some(coded.encode(views.first()?.bytes_in(arena)?))
}

/// A compressed column against a literal, tested without decompressing a row of it.
///
/// The comparison is a byte slice against a byte slice, which is what it would have been on the
/// strings, over half as many bytes and with no decompression before it. A row whose codes are a
/// different length is settled by the length alone, which on a column of URLs is most of them.
fn coded_against<M>(
    op: Comparison,
    coded: &Coded<'_>,
    wanted: &[u8],
    len: usize,
    map: M,
) -> Vec<bool>
where
    M: Fn(usize) -> usize + Copy,
{
    let same = op == Comparison::Equal;
    let mut answers = Vec::with_capacity(len);
    for row in 0..len {
        answers.push((coded.row(map(row)) == Some(wanted)) == same);
    }
    answers
}

/// A bit packed column against a literal, compared in the code space the column is already in.
///
/// The translation is one subtraction done once. After it the loop is a shift, a mask and a compare
/// of two `u64`, which is what the flat loop would have been doing anyway minus the unpacking, so
/// the form costs nothing on the operation a filter spends most of its time in.
///
/// The operator is a match out here and a separate loop under each arm rather than one loop calling
/// a function it was handed. A function pointer is an indirect call per row, and a loop with one in
/// it is a loop the compiler will not widen, so the compare that should have been a few rows at a
/// time was one row at a time with a call in the middle. On a sample of the TPC-H q20 filter, which
/// is two comparisons of a packed date column over six million rows, this pair of loops was two
/// thirds of everything the scan did.
///
/// The codes go into a vector each thread keeps and the answers are built by the sweep rather than
/// laid out as a run of `false` for it to write over. Both of those were an allocation and a run of
/// zeroes per chunk per conjunct, and a filter over six million rows is three thousand chunks, so
/// the allocator and the zeroing were four of the twenty two instructions a row this path costs.
fn packed_against<M>(
    op: Comparison,
    packed: &Packed<'_>,
    wanted: i128,
    len: usize,
    map: M,
) -> Vec<bool>
where
    M: Fn(usize) -> usize + Copy,
{
    let Some(code) = packed.code_of(wanted) else {
        // The literal is outside the range the width can hold, so every row answers the same way
        // and the answer is arithmetic on two numbers rather than a pass over the column.
        let above = wanted > packed.ceiling();
        let same = match op {
            Comparison::Equal | Comparison::NotDistinctFrom => false,
            Comparison::NotEqual | Comparison::DistinctFrom => true,
            Comparison::Less | Comparison::LessOrEqual => above,
            Comparison::Greater | Comparison::GreaterOrEqual => !above,
        };
        return vec![same; len];
    };
    thread_local! {
        static CODES: RefCell<Vec<u64>> = const { RefCell::new(Vec::new()) };
    }
    // Taken out of the thread's slot and put back rather than borrowed for the body, because a body
    // in a closure that both arms of a borrow call is a body that stops being inlined, and the sweep
    // under each arm is the loop this whole path exists to keep tight.
    let mut held = CODES.with_borrow_mut(std::mem::take);
    // Unpacked in bulk first, so that the loop under each arm is a compare of two numbers. See
    // [`Packed::unpack`].
    packed.codes_into(map, len, &mut held);
    let codes = &held[..len];
    /// One pass over the rows with the comparison inlined into it.
    macro_rules! sweep {
        ($test:expr) => {{
            let test = $test;
            codes.iter().map(|&found| test(found, code)).collect()
        }};
    }
    let answers: Vec<bool> = match op {
        Comparison::Equal | Comparison::NotDistinctFrom => sweep!(|found, want| found == want),
        Comparison::NotEqual | Comparison::DistinctFrom => sweep!(|found, want| found != want),
        Comparison::Less => sweep!(|found, want| found < want),
        Comparison::LessOrEqual => sweep!(|found, want| found <= want),
        Comparison::Greater => sweep!(|found, want| found > want),
        Comparison::GreaterOrEqual => sweep!(|found, want| found >= want),
    };
    CODES.with_borrow_mut(|slot| *slot = held);
    answers
}

/// Two bit packed columns against each other, which is the pair a stored table produces.
///
/// A packed vector's value is its base plus its code, so a row is compared by adding each side's
/// base to each side's code and comparing the two sums. Nothing is unpacked into a vector of its
/// own and no `Value` is built, which is what the row at a time path underneath this was doing for
/// both sides of every row.
///
/// The two bases are almost never the same number, since each one is the smallest value in its own
/// column, so there is no shortcut in comparing the codes directly. What there is instead is that
/// the ranges may not overlap at all: a column whose largest value is below the other's smallest
/// answers every row the same way, and that is decided here out of four numbers with no bit of
/// either column being read.
///
/// `None` for a pair whose base plus width does not fit, which sends it to the row at a time path
/// rather than wrapping. A `DECIMAL(38)` column can be packed and its base can sit near the end of
/// the range, and the loop adds without asking so the asking happens once out here.
///
/// Each side takes its own index mapping, which is what lets a dictionary over a packed run of
/// values reach this with its codes as the mapping. A stored column of dates is that as often as it
/// is a packed run on its own, and from here the two are one loop with a different index.
///
/// On TPC-H this is `l_commitdate < l_receiptdate` and `l_shipdate < l_commitdate`, two packed date
/// columns of six million rows, which is q4, q12 and q21.
fn packed_against_packed<L, R, M>(
    op: Comparison,
    left: &Packed<'_>,
    at_left: L,
    right: &Packed<'_>,
    at_right: R,
    len: usize,
    map: M,
) -> Option<Vec<bool>>
where
    L: Fn(usize) -> usize,
    R: Fn(usize) -> usize,
    M: Fn(usize) -> usize + Copy,
{
    let (low, high) = (left.base(), ceiling_of(left)?);
    let (other_low, other_high) = (right.base(), ceiling_of(right)?);
    if high < other_low || other_high < low {
        // The two ranges are disjoint, so every row of the left column is on the same side of every
        // row of the right one and the answer is arithmetic on four numbers.
        let below = high < other_low;
        let same = match op {
            Comparison::Equal | Comparison::NotDistinctFrom => false,
            Comparison::NotEqual | Comparison::DistinctFrom => true,
            Comparison::Less | Comparison::LessOrEqual => below,
            Comparison::Greater | Comparison::GreaterOrEqual => !below,
        };
        return Some(vec![same; len]);
    }
    // The operator is a match out here with a loop under each arm, the same way `packed_against`
    // takes it and for the same reason.
    let mut answers = vec![false; len];
    /// One pass over the rows with the comparison inlined into it.
    macro_rules! sweep {
        ($test:expr) => {{
            let test = $test;
            for (slot, answer) in answers.iter_mut().enumerate() {
                let row = map(slot);
                *answer = test(
                    low + i128::from(left.code(at_left(row))),
                    other_low + i128::from(right.code(at_right(row))),
                );
            }
        }};
    }
    match op {
        Comparison::Equal | Comparison::NotDistinctFrom => sweep!(|one, other| one == other),
        Comparison::NotEqual | Comparison::DistinctFrom => sweep!(|one, other| one != other),
        Comparison::Less => sweep!(|one, other| one < other),
        Comparison::LessOrEqual => sweep!(|one, other| one <= other),
        Comparison::Greater => sweep!(|one, other| one > other),
        Comparison::GreaterOrEqual => sweep!(|one, other| one >= other),
    }
    Some(answers)
}

/// A side whose values are a packed run, either its own or one a dictionary points into.
///
/// A dictionary of few distinct numbers over a wide range is what our own storage writes for a
/// column like a date, and [`through`] answers `None` for it because a packed run is not a run of
/// `Data` to index. This is the same question asked of the form underneath.
enum Packing<'a> {
    /// The run is this vector's own and row `n` is code `n` of it.
    Straight(Packed<'a>),
    /// The run belongs to a dictionary and row `n` is the code the run names for it.
    Coded(Cow<'a, [u32]>, Packed<'a>),
}

/// How `vector` reaches a packed run, or `None` when it does not have one.
fn packing(vector: &Vector) -> Option<Packing<'_>> {
    if let Some(packed) = vector.packed_parts() {
        return Some(Packing::Straight(packed));
    }
    let (codes, values) = vector.positions()?;
    Some(Packing::Coded(codes, values.packed_parts()?))
}

/// The four ways a pair of packed runs can be indexed, resolved once out here so that the loop
/// underneath is monomorphized on both mappings rather than calling through a pointer per row.
fn packings_against_each_other<M>(
    op: Comparison,
    left: &Packing<'_>,
    right: &Packing<'_>,
    len: usize,
    map: M,
) -> Option<Vec<bool>>
where
    M: Fn(usize) -> usize + Copy,
{
    match (left, right) {
        (Packing::Straight(one), Packing::Straight(other)) => {
            packed_against_packed(op, one, identity, other, identity, len, map)
        }
        (Packing::Straight(one), Packing::Coded(codes, other)) => {
            packed_against_packed(op, one, identity, other, |row| codes[row] as usize, len, map)
        }
        (Packing::Coded(codes, one), Packing::Straight(other)) => {
            packed_against_packed(op, one, |row| codes[row] as usize, other, identity, len, map)
        }
        (Packing::Coded(codes, one), Packing::Coded(others, other)) => packed_against_packed(
            op,
            one,
            |row| codes[row] as usize,
            other,
            |row| others[row] as usize,
            len,
            map,
        ),
    }
}

/// A packed run against a run of whole numbers, with the four ways to index the pair resolved here.
///
/// The answer is for `packed op run`, so a caller with the flat side on the left hands the swapped
/// operator rather than a second loop. It is the same arrangement [`packings_against_each_other`]
/// has and it is here for the same reason: the mapping is picked once out here so that the loop
/// underneath is monomorphized on it rather than calling through a pointer per row.
fn packing_against_run<M>(
    op: Comparison,
    packed: &Packing<'_>,
    run: &Through<'_>,
    len: usize,
    map: M,
) -> Option<Vec<bool>>
where
    M: Fn(usize) -> usize + Copy,
{
    match (packed, run) {
        (Packing::Straight(one), Through::Direct(other)) => {
            packed_against_flat(op, one, identity, other, identity, len, map)
        }
        (Packing::Straight(one), Through::Coded(codes, other)) => {
            packed_against_flat(op, one, identity, other, |row| codes[row] as usize, len, map)
        }
        (Packing::Coded(codes, one), Through::Direct(other)) => {
            packed_against_flat(op, one, |row| codes[row] as usize, other, identity, len, map)
        }
        (Packing::Coded(codes, one), Through::Coded(others, other)) => packed_against_flat(
            op,
            one,
            |row| codes[row] as usize,
            other,
            |row| others[row] as usize,
            len,
            map,
        ),
    }
}

/// A bit packed column against a flat one, with neither side turned into the other.
///
/// The packed side is its base plus its code and the flat side is the number it already holds, so
/// the loop is an add, a widen and a compare, and nothing is unpacked into a vector of its own. The
/// row at a time path underneath this was building a [`Value`] for both sides of every row.
///
/// `None` for a layout whose values do not all fit in an `i128`, and for a packed side whose base
/// plus width does not fit either, which is the question [`ceiling_of`] asks. The loop adds without
/// asking so the asking happens once out here, the same way [`packed_against_packed`] does it.
///
/// This is the pair a clustered table produces. Sorting lineitem by month of `l_shipdate` leaves
/// `l_commitdate` with about a hundred distinct values inside a partition, which packs, while
/// `l_receiptdate` arrives flat, and `l_commitdate < l_receiptdate` is q4, q12 and q21.
fn packed_against_flat<L, R, M>(
    op: Comparison,
    packed: &Packed<'_>,
    at_packed: L,
    flat: &Data,
    at_flat: R,
    len: usize,
    map: M,
) -> Option<Vec<bool>>
where
    L: Fn(usize) -> usize,
    R: Fn(usize) -> usize,
    M: Fn(usize) -> usize + Copy,
{
    ceiling_of(packed)?;
    let base = packed.base();
    macro_rules! layouts {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match flat {
                $(
                    Data::$variant(run) => Some(numbers_compared(op, len, |slot| {
                        let row = map(slot);
                        (
                            base + i128::from(packed.code(at_packed(row))),
                            i128::from(run[at_flat(row)]),
                        )
                    })),
                )+
                // A layout with no whole number in it, which is every string, float and interval
                // column, plus `UBIGINT`'s wider sibling whose largest value has no `i128`.
                _ => None,
            }
        };
    }
    rudb_vector::for_each_layout!(exact, layouts)
}

/// One pass over the rows with the comparison inlined into it, over a pair of numbers a row.
///
/// The operator is a match out here and a separate loop under each arm rather than one loop calling
/// a function it was handed, for the reason [`packed_against`] gives: a function pointer is an
/// indirect call per row and a loop with one in it is a loop the compiler will not widen.
fn numbers_compared<P>(op: Comparison, len: usize, pair: P) -> Vec<bool>
where
    P: Fn(usize) -> (i128, i128),
{
    let mut answers = vec![false; len];
    /// One pass over the rows with `$test` as the body.
    macro_rules! sweep {
        ($test:expr) => {{
            let test = $test;
            for (slot, answer) in answers.iter_mut().enumerate() {
                let (one, other) = pair(slot);
                *answer = test(one, other);
            }
        }};
    }
    match op {
        Comparison::Equal | Comparison::NotDistinctFrom => sweep!(|one, other| one == other),
        Comparison::NotEqual | Comparison::DistinctFrom => sweep!(|one, other| one != other),
        Comparison::Less => sweep!(|one, other| one < other),
        Comparison::LessOrEqual => sweep!(|one, other| one <= other),
        Comparison::Greater => sweep!(|one, other| one > other),
        Comparison::GreaterOrEqual => sweep!(|one, other| one >= other),
    }
    answers
}

/// The largest value a packed vector can hold, or `None` if that number does not exist.
///
/// [`Packed::ceiling`] adds without asking, which is right where the caller has already put a
/// literal through [`Packed::code_of`] and so knows the base and the width are a pair that works.
/// A loop that adds a code to a base for every row has not asked anything yet, so it asks here.
fn ceiling_of(packed: &Packed<'_>) -> Option<i128> {
    let mask = u64::MAX >> (u64::BITS - packed.width());
    packed.base().checked_add(i128::from(mask))
}

/// Where a side keeps its values, for the forms that reach them through a run of them.
///
/// A flat vector holds its values itself and row `n` is at position `n`. A dictionary holds them
/// somewhere else and a code per row, so row `n` is at position `codes[n]` of that. A run length
/// vector is the same shape with the run index standing in for the code, which is why both come
/// from `Vector::positions` rather than from an accessor per form: from here there is no difference
/// between them and writing two branches would only be two chances to write one of them wrong.
///
/// The point of naming it is that the loop underneath does not change. A comparison of two
/// dictionary columns is the flat loop with a different index mapping, so it costs one extra load
/// per row per coded side against a path that was allocating a `Value` per row and calling the
/// generic comparison on it. On TPC-H that path was `l_commitdate < l_receiptdate`, two dictionary
/// encoded date columns of six million rows, which is q4, q12 and q21.
enum Through<'a> {
    /// The values are this vector's own and row `n` is at position `n`.
    Direct(&'a Data),
    /// The values are somewhere else and row `n` is at the position this run names for it.
    Coded(Cow<'a, [u32]>, &'a Data),
}

/// How `vector` reaches its values, or `None` for a form that does not reach them through a run.
///
/// A constant, a bit packed column and a compressed one all answer `None` here, and each has a
/// branch of its own further down [`specialized`] that does something better than reading a run
/// would. A dictionary whose values are themselves not flat answers `None` too, because then there
/// is no run to index and the only thing left is the row at a time path.
fn through(vector: &Vector) -> Option<Through<'_>> {
    if let Some(data) = vector.data() {
        return Some(Through::Direct(data));
    }
    let (codes, values) = vector.positions()?;
    Some(Through::Coded(codes, values.data()?))
}

/// One loop per physical layout, generated rather than written out.
///
/// The two index closures are what let the same body serve flat against flat, a column against a
/// constant and a dictionary against a constant. `identity` on both sides is the first, `first` on
/// the right is the second, and the codes on the left are the third.
#[expect(
    clippy::too_many_arguments,
    reason = "two sides with an index each, the operator, the length and two validities, all of \
              which the loop needs and none of which is worth a struct that exists for one call"
)]
fn dispatch<L, R, V>(
    op: Comparison,
    len: usize,
    left: &Data,
    at_left: L,
    right: &Data,
    at_right: R,
    left_valid: &Validity,
    right_valid: &Validity,
    at_valid: V,
) -> Option<Vec<bool>>
where
    L: Fn(usize) -> usize,
    R: Fn(usize) -> usize,
    V: Fn(usize) -> usize,
{
    macro_rules! layouts {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match (left, right) {
                $(
                    (Data::$variant(one), Data::$variant(other)) => Some(sweep(
                        op,
                        len,
                        |index| one[at_left(index)].cmp(&other[at_right(index)]),
                        left_valid,
                        right_valid,
                        &at_valid,
                    )),
                )+
                // Floats have their own order, which is DuckDB's rather than IEEE's, and the
                // widening on a `f32` is free because the comparison is against another `f32`.
                (Data::Float32(one), Data::Float32(other)) => Some(sweep(
                    op,
                    len,
                    |index| {
                        float_order(
                            f64::from(one[at_left(index)]),
                            f64::from(other[at_right(index)]),
                        )
                    },
                    left_valid,
                    right_valid,
                    &at_valid,
                )),
                (Data::Float64(one), Data::Float64(other)) => Some(sweep(
                    op,
                    len,
                    |index| float_order(one[at_left(index)], other[at_right(index)]),
                    left_valid,
                    right_valid,
                    &at_valid,
                )),
                // An interval is three counts and the order is over the one length they add up to,
                // so this is not the derived order of the triple and cannot be generated above.
                (Data::Interval(one), Data::Interval(other)) => Some(sweep(
                    op,
                    len,
                    |index| {
                        let (months, days, micros) = one[at_left(index)];
                        let (bm, bd, bu) = other[at_right(index)];
                        interval_micros(months, days, micros).cmp(&interval_micros(bm, bd, bu))
                    },
                    left_valid,
                    right_valid,
                    &at_valid,
                )),
                (Data::Varlen(one), Data::Varlen(other)) => Some(sweep(
                    op,
                    len,
                    |index| string_order(one, at_left(index), other, at_right(index)),
                    left_valid,
                    right_valid,
                    &at_valid,
                )),
                _ => None,
            }
        };
    }
    rudb_vector::for_each_layout!(ordered, layouts)
}

/// The one row column for a constant, either the one that was built early or one built here.
///
/// Borrowed when a caller handed one over for this side and this value, owned when it did not, and
/// the loop below cannot tell the two apart. `None` is a type with no column layout, which is what
/// sends the whole comparison to the row at a time path.
fn readied<'a>(held: Option<&'a Held>, ty: &LogicalType, value: &Value) -> Option<Cow<'a, Vector>> {
    match held {
        Some(held) if held.matches(ty, value) => Some(Cow::Borrowed(held.single())),
        _ => Some(Cow::Owned(single(ty, value)?)),
    }
}

/// Two strings in byte order, resolved from the four byte prefix where it can be.
///
/// The lemma this rests on is that prefix order is byte order whenever the two prefixes differ. A
/// view pads a string shorter than four bytes with zeros, zero is the least byte, and byte order
/// says a string is less than any string that extends it, so padding compares the same way the
/// missing bytes would have. When the prefixes are equal the payload settles it, which for an
/// inline string is the same sixteen bytes already loaded and for a long one is a block read.
fn string_order(
    left: &StringColumn,
    at_left: usize,
    right: &StringColumn,
    at_right: usize,
) -> Ordering {
    view_order(left.views().get(at_left), left.arena(), right.views().get(at_right), right.arena())
}

/// The same comparison written against a view and the arena behind it rather than against a column.
///
/// Both forms that hold strings come through here, so a flat varchar column and a string view column
/// order a pair of rows the same way and there is no second copy of the prefix rule to drift from
/// this one.
fn view_order(
    one: Option<&StringView>,
    one_arena: &[u8],
    other: Option<&StringView>,
    other_arena: &[u8],
) -> Ordering {
    let (Some(one), Some(other)) = (one, other) else {
        return Ordering::Equal;
    };
    let (prefix, against) = (one.prefix(), other.prefix());
    if prefix != against {
        return prefix.cmp(&against);
    }
    // Bytes rather than `StringColumn::get`, which validates UTF-8. Everything in a column was
    // pushed from a `&str` so the validation cannot fail, and on a URL column, where every row
    // shares the `http` prefix and the payload therefore decides every comparison, it was the
    // larger half of the per row cost.
    let bytes = one.bytes_in(one_arena).unwrap_or_default();
    let against_bytes = other.bytes_in(other_arena).unwrap_or_default();
    bytes.cmp(against_bytes)
}

/// The answers for one ordering, with the operator decided once rather than once per row.
///
/// This is where the match on the operator gets hoisted. Each arm calls a generic `fill` with a
/// different predicate, so the compiler produces eight loops whose bodies are an ordering against a
/// constant, rather than one loop with a branch table in it.
fn sweep<O, V>(
    op: Comparison,
    len: usize,
    order_at: O,
    left_valid: &Validity,
    right_valid: &Validity,
    at_valid: V,
) -> Vec<bool>
where
    O: Fn(usize) -> Ordering,
    V: Fn(usize) -> usize,
{
    let mut answers = vec![false; len];
    match op {
        Comparison::Equal => fill(&mut answers, order_at, |o| o == Ordering::Equal),
        Comparison::NotEqual => fill(&mut answers, order_at, |o| o != Ordering::Equal),
        Comparison::Less => fill(&mut answers, order_at, |o| o == Ordering::Less),
        Comparison::LessOrEqual => fill(&mut answers, order_at, |o| o != Ordering::Greater),
        Comparison::Greater => fill(&mut answers, order_at, |o| o == Ordering::Greater),
        Comparison::GreaterOrEqual => fill(&mut answers, order_at, |o| o != Ordering::Less),
        Comparison::DistinctFrom => {
            total(&mut answers, order_at, left_valid, right_valid, at_valid);
            for answer in &mut answers {
                *answer = !*answer;
            }
        }
        Comparison::NotDistinctFrom => {
            total(&mut answers, order_at, left_valid, right_valid, at_valid);
        }
    }
    answers
}

/// One loop, one predicate, no branch on the operator.
#[inline]
fn fill<O, H>(answers: &mut [bool], order_at: O, held: H)
where
    O: Fn(usize) -> Ordering,
    H: Fn(Ordering) -> bool,
{
    for (index, answer) in answers.iter_mut().enumerate() {
        *answer = held(order_at(index));
    }
}

/// `IS NOT DISTINCT FROM`, which reads validity as data rather than as an absence.
///
/// Two nulls are the same value here and a null against anything else is not, which is the whole
/// difference between this and `=`. The all valid case is checked once so that the common shape,
/// which is a total comparison inside a join on columns that happen not to be nullable, does not
/// pay for two validity lookups per row.
fn total<O, V>(
    answers: &mut [bool],
    order_at: O,
    left_valid: &Validity,
    right_valid: &Validity,
    at_valid: V,
) where
    O: Fn(usize) -> Ordering,
    V: Fn(usize) -> usize,
{
    if *left_valid == Validity::AllValid && *right_valid == Validity::AllValid {
        fill(answers, order_at, |o| o == Ordering::Equal);
        return;
    }
    for (index, answer) in answers.iter_mut().enumerate() {
        let row = at_valid(index);
        *answer = match (left_valid.is_valid(row), right_valid.is_valid(row)) {
            (true, true) => order_at(index) == Ordering::Equal,
            (false, false) => true,
            _ => false,
        };
    }
}

/// Compares two values, producing `TRUE`, `FALSE` or `NULL`.
///
/// # Errors
///
/// If the two types cannot be compared, which after binding means one of them is a nested type.
pub fn compare_values(op: Comparison, left: &Value, right: &Value) -> Result<Value> {
    if op.is_total() {
        let same = match (left.is_null(), right.is_null()) {
            (true, true) => true,
            (true, false) | (false, true) => false,
            (false, false) => order(left, right)? == Ordering::Equal,
        };
        return Ok(Value::Boolean(match op {
            Comparison::NotDistinctFrom => same,
            _ => !same,
        }));
    }
    if left.is_null() || right.is_null() {
        return Ok(Value::Null);
    }
    let ordering = order(left, right)?;
    let held = match op {
        Comparison::Equal => ordering == Ordering::Equal,
        Comparison::NotEqual => ordering != Ordering::Equal,
        Comparison::Less => ordering == Ordering::Less,
        Comparison::LessOrEqual => ordering != Ordering::Greater,
        Comparison::Greater => ordering == Ordering::Greater,
        Comparison::GreaterOrEqual => ordering != Ordering::Less,
        Comparison::DistinctFrom | Comparison::NotDistinctFrom => {
            return Err(Error::internal("a total comparison reached the ordered path"));
        }
    };
    Ok(Value::Boolean(held))
}

/// The order of two values, neither of which is null.
///
/// This is the one place the sort order of a type is written down. `ORDER BY`, `GROUP BY`, a merge
/// join and a min or max aggregate all reach it, and a type that ordered differently in two of
/// those would produce a query whose answer depends on which operator the optimizer picked.
///
/// # Errors
///
/// If either value is null, which is the caller's mistake rather than a comparison, or if the
/// types have no order between them.
pub fn order(left: &Value, right: &Value) -> Result<Ordering> {
    match (left, right) {
        (Value::Null, _) | (_, Value::Null) => {
            Err(Error::internal("a null reached the ordering path"))
        }
        (Value::Boolean(a), Value::Boolean(b)) => Ok(a.cmp(b)),
        (Value::Varchar(a), Value::Varchar(b)) => Ok(a.as_bytes().cmp(b.as_bytes())),
        (Value::Blob(a), Value::Blob(b)) => Ok(a.cmp(b)),
        (Value::Bit(a), Value::Bit(b)) => Ok(rudb_common::bit::cmp(a, b)),
        (Value::Date(a), Value::Date(b)) => Ok(a.cmp(b)),
        // A zoned value orders with its own kind and by the same rule, since both of them are the
        // count of microseconds from a fixed point and the zone is about printing.
        (Value::Time(a), Value::Time(b))
        | (Value::TimeTz(a), Value::TimeTz(b))
        | (Value::Timestamp(a), Value::Timestamp(b))
        | (Value::TimestampTz(a), Value::TimestampTz(b)) => Ok(a.cmp(b)),
        (
            Value::Interval { months: am, days: ad, micros: au },
            Value::Interval { months: bm, days: bd, micros: bu },
        ) => Ok(interval_micros(*am, *ad, *au).cmp(&interval_micros(*bm, *bd, *bu))),
        (Value::List { values: a, .. }, Value::List { values: b, .. }) => list_order(a, b),
        // Two structs of one type order field by field in the order the type lists them, with a
        // null field above everything the way a null element is in a list, which is the pin's.
        (Value::Struct(a), Value::Struct(b)) => {
            for ((_, one), (_, other)) in a.iter().zip(b) {
                let ordering = order_with_nulls(one, other, false)?;
                if ordering != Ordering::Equal {
                    return Ok(ordering);
                }
            }
            Ok(a.len().cmp(&b.len()))
        }
        // A map is a list of key and value structs underneath on the pin and orders as one, entry
        // by entry and key before value, with the shorter one first when one runs out.
        (Value::Map { entries: a, .. }, Value::Map { entries: b, .. }) => {
            for ((key, value), (other_key, other_value)) in a.iter().zip(b) {
                let ordering = order_with_nulls(key, other_key, false)?.then(order_with_nulls(
                    value,
                    other_value,
                    false,
                )?);
                if ordering != Ordering::Equal {
                    return Ok(ordering);
                }
            }
            Ok(a.len().cmp(&b.len()))
        }
        _ => numeric_order(left, right),
    }
}

/// Two lists in the order DuckDB puts them in, which is the order two words are in a dictionary.
///
/// The first element the two disagree on decides, and if neither ran out of elements before that
/// happened then the shorter one is the smaller one. So `[] < [1]` and `[1, 2] < [1, 2, 3]`, and a
/// list is never equal to a longer list that starts with it.
///
/// A null element is not an absence here, it is the largest value there is, which is the one place
/// a comparison inside a list disagrees with the same comparison outside one. `[1, NULL] > [1, 2]`
/// is true on the pin and `[NULL] < [1]` is false, and two nulls in the same position are equal,
/// which is what makes `[1, NULL] = [1, NULL]` true while `NULL = NULL` is null. That is
/// [`order_with_nulls`] with nulls last, so it is the sort's rule and not a second one written here.
///
/// A null list, as opposed to a list with a null in it, never reaches this. It is handled by the
/// caller the way a null of any other type is, which is why `NULL::INT[] = [1]` is null.
fn list_order(left: &[Value], right: &[Value]) -> Result<Ordering> {
    for (one, other) in left.iter().zip(right) {
        let ordering = order_with_nulls(one, other, false)?;
        if ordering != Ordering::Equal {
            return Ok(ordering);
        }
    }
    Ok(left.len().cmp(&right.len()))
}

/// The order of two numbers, which is the case that has to work across representations.
fn numeric_order(left: &Value, right: &Value) -> Result<Ordering> {
    if let (Some(a), Some(b)) = (integral(left), integral(right)) {
        return Ok(a.cmp(&b));
    }
    if let (
        Value::Decimal { unscaled: a, scale: sa, .. },
        Value::Decimal { unscaled: b, scale: sb, .. },
    ) = (left, right)
        && sa == sb
    {
        return Ok(a.cmp(b));
    }
    match (approximate(left), approximate(right)) {
        (Some(a), Some(b)) => Ok(float_order(a, b)),
        _ => Err(Error::not_implemented(format!(
            "comparing {} with {}",
            left.logical_type(),
            right.logical_type()
        ))),
    }
}

/// DuckDB's float order: NaN is equal to itself and above everything else, and zero has one place.
pub(crate) fn float_order(left: f64, right: f64) -> Ordering {
    if left == right {
        return Ordering::Equal;
    }
    match (left.is_nan(), right.is_nan()) {
        (true, true) => Ordering::Equal,
        (true, false) => Ordering::Greater,
        (false, true) => Ordering::Less,
        (false, false) => left.partial_cmp(&right).unwrap_or(Ordering::Equal),
    }
}

/// The order of two values with nulls in it, for a sort key.
///
/// A sort has to put nulls somewhere and SQL lets the query say where, so this takes the answer
/// rather than deciding it.
///
/// # Errors
///
/// If the two types have no order between them.
pub fn order_with_nulls(left: &Value, right: &Value, nulls_first: bool) -> Result<Ordering> {
    match (left.is_null(), right.is_null()) {
        (true, true) => Ok(Ordering::Equal),
        (true, false) => Ok(if nulls_first { Ordering::Less } else { Ordering::Greater }),
        (false, true) => Ok(if nulls_first { Ordering::Greater } else { Ordering::Less }),
        (false, false) => order(left, right),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn compared(op: Comparison, left: Value, right: Value) -> Value {
        compare_values(op, &left, &right).expect("these types compare")
    }

    /// Every comparison, so that a test that sweeps them cannot quietly miss one.
    const EVERY: [Comparison; 8] = [
        Comparison::Equal,
        Comparison::NotEqual,
        Comparison::Less,
        Comparison::LessOrEqual,
        Comparison::Greater,
        Comparison::GreaterOrEqual,
        Comparison::DistinctFrom,
        Comparison::NotDistinctFrom,
    ];

    /// The row at a time path, kept as the oracle rather than deleted.
    ///
    /// `spec/engine/03-data-plane.md` is explicit that the slow path becomes the thing the fast
    /// path is checked against. This is that, written out here so that a test can call it on a pair
    /// of vectors whose forms the fast path does specialize.
    fn oracle(op: Comparison, left: &Vector, right: &Vector) -> Vector {
        let values: Vec<Value> = (0..left.len())
            .map(|index| {
                compare_values(op, &left.value_at(index), &right.value_at(index))
                    .expect("the oracle is only asked about types that compare")
            })
            .collect();
        Vector::from_values(LogicalType::Boolean, &values).expect("booleans")
    }

    /// Asserts that the specialized path and the oracle produce the same vector, not merely the
    /// same answers. Same vector means the same data, the same validity representation and the
    /// same false in every null position, which is a much stronger statement and is free to check.
    fn agrees(op: Comparison, left: &Vector, right: &Vector) {
        let fast = compare(op, left, right).expect("compares");
        let slow = oracle(op, left, right);
        assert_eq!(fast, slow, "{op:?} on a {:?} against a {:?}", left.form(), right.form());
    }

    /// Every row in blocks of 64 is the plain filter, over empty, full and mixed blocks, with and
    /// without a tail, and through a flat column against a literal and against another column.
    #[test]
    fn rows_kept_in_blocks_are_the_rows_a_plain_filter_keeps() {
        let shapes: [fn(usize) -> bool; 5] = [
            |_| false,
            |_| true,
            |row| row % 7 == 3,
            |row| (row / 64) % 3 == 1,
            |row| (row / 64) % 2 == 0 || row % 61 == 0,
        ];
        for len in [0, 1, 63, 64, 65, 128, 1000, 2048] {
            for held in shapes {
                let blocks = kept_where(None, len, held);
                let plain: Vec<u32> = (0..len)
                    .filter(|&row| held(row))
                    .map(|row| u32::try_from(row).expect("small"))
                    .collect();
                assert_eq!(blocks.indices(), plain.as_slice(), "{len} rows");
            }
        }
        let values: Vec<Value> =
            (0..2000).map(|row| Value::SmallInt(if row % 50 == 0 { 3 } else { 0 })).collect();
        let column = Vector::from_values(LogicalType::SmallInt, &values).expect("smallints");
        let zero = Vector::constant(LogicalType::SmallInt, Value::SmallInt(0), 2000);
        let others: Vec<Value> = (0..2000).map(|row| Value::SmallInt(row % 5)).collect();
        let others = Vector::from_values(LogicalType::SmallInt, &others).expect("smallints");
        for op in [Comparison::NotEqual, Comparison::Equal, Comparison::Less] {
            agrees_and_selects(op, &column, &zero);
            agrees_and_selects(op, &column, &others);
        }
    }

    /// [`agrees`], and the rows picked straight from the answers are the rows the flag vector
    /// gives. Kept apart from `agrees` because a pair of forms with no loop goes the long way here
    /// too, and the tests that count the long way would count it twice.
    fn agrees_and_selects(op: Comparison, left: &Vector, right: &Vector) {
        agrees(op, left, right);
        let slow = oracle(op, left, right);
        let picked = select_prepared(op, left, right, None).expect("selects");
        assert_eq!(
            picked.indices(),
            crate::select::selection(&slow, slow.len()).indices(),
            "{op:?} selected on a {:?} against a {:?}",
            left.form(),
            right.form()
        );
    }

    /// A small deterministic generator, because a property test with no seed is a test that fails
    /// on somebody else's machine and passes on yours.
    struct Rng(u64);

    impl Rng {
        fn next(&mut self) -> u64 {
            self.0 ^= self.0 << 13;
            self.0 ^= self.0 >> 7;
            self.0 ^= self.0 << 17;
            self.0
        }

        fn below(&mut self, bound: u64) -> u64 {
            self.next() % bound
        }
    }

    #[test]
    fn an_ordinary_comparison_is_null_when_either_side_is() {
        assert_eq!(compared(Comparison::Equal, Value::Integer(1), Value::Null), Value::Null);
        assert_eq!(compared(Comparison::Less, Value::Null, Value::Integer(1)), Value::Null);
    }

    #[test]
    fn a_total_comparison_is_never_null() {
        assert_eq!(
            compared(Comparison::NotDistinctFrom, Value::Null, Value::Null),
            Value::Boolean(true)
        );
        assert_eq!(
            compared(Comparison::NotDistinctFrom, Value::Integer(1), Value::Null),
            Value::Boolean(false)
        );
        assert_eq!(
            compared(Comparison::DistinctFrom, Value::Integer(1), Value::Null),
            Value::Boolean(true)
        );
    }

    #[test]
    fn a_string_compares_by_bytes() {
        assert_eq!(
            compared(Comparison::Less, Value::Varchar("a".into()), Value::Varchar("b".into())),
            Value::Boolean(true)
        );
        assert_eq!(
            compared(Comparison::Less, Value::Varchar("Z".into()), Value::Varchar("a".into())),
            Value::Boolean(true)
        );
    }

    /// The reason this crate does not use `f64::partial_cmp` directly. A NaN that compared
    /// unordered would make a group by produce a group nothing can find again.
    #[test]
    fn two_nans_are_one_value_and_they_sort_above_the_numbers() {
        assert_eq!(
            compared(Comparison::Equal, Value::Double(f64::NAN), Value::Double(f64::NAN)),
            Value::Boolean(true)
        );
        assert_eq!(
            compared(Comparison::Greater, Value::Double(f64::NAN), Value::Double(1e300)),
            Value::Boolean(true)
        );
    }

    #[test]
    fn zero_has_one_value_however_it_is_signed() {
        assert_eq!(
            compared(Comparison::Equal, Value::Double(0.0), Value::Double(-0.0)),
            Value::Boolean(true)
        );
    }

    /// An interval is three counts and two of them that are the same length are one value, at
    /// thirty days to a month and twenty four hours to a day, which is what upstream answers. The
    /// three counts are still kept apart, because adding a month to a date is not adding thirty
    /// days to it, so these pairs are equal and print differently.
    #[test]
    fn two_intervals_of_the_same_length_are_one_value() {
        let day = Value::Interval { months: 0, days: 1, micros: 0 };
        let hours = Value::Interval { months: 0, days: 0, micros: 86_400_000_000 };
        let month = Value::Interval { months: 1, days: 0, micros: 0 };
        let thirty = Value::Interval { months: 0, days: 30, micros: 0 };
        let long_day = Value::Interval { months: 0, days: 0, micros: 90_000_000_000 };
        assert_eq!(compared(Comparison::Equal, day.clone(), hours), Value::Boolean(true));
        assert_eq!(compared(Comparison::Equal, month, thirty), Value::Boolean(true));
        assert_eq!(compared(Comparison::Greater, long_day, day), Value::Boolean(true));
    }

    #[test]
    fn a_number_compares_the_same_however_it_is_stored() {
        assert_eq!(
            compared(Comparison::Equal, Value::Integer(3), Value::BigInt(3)),
            Value::Boolean(true)
        );
        assert_eq!(
            compared(Comparison::Less, Value::Integer(3), Value::Double(3.5)),
            Value::Boolean(true)
        );
    }

    #[test]
    fn nulls_go_where_the_query_asked_for_them() {
        assert_eq!(
            order_with_nulls(&Value::Null, &Value::Integer(1), true).expect("orders"),
            Ordering::Less
        );
        assert_eq!(
            order_with_nulls(&Value::Null, &Value::Integer(1), false).expect("orders"),
            Ordering::Greater
        );
    }

    #[test]
    fn two_constant_vectors_cost_one_comparison() {
        let left = Vector::constant(LogicalType::Integer, Value::Integer(1), 512);
        let right = Vector::constant(LogicalType::Integer, Value::Integer(2), 512);
        let result = compare(Comparison::Less, &left, &right).expect("compares");
        assert_eq!(result.form(), Form::Constant);
        assert_eq!(result.value_at(500), Value::Boolean(true));
    }

    #[test]
    fn a_comparison_of_two_vectors_is_one_answer_per_row() {
        let left = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(5), Value::Null],
        )
        .expect("three rows");
        let right = Vector::constant(LogicalType::Integer, Value::Integer(3), 3);
        let result = compare(Comparison::Greater, &left, &right).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(false));
        assert_eq!(result.value_at(1), Value::Boolean(true));
        assert_eq!(result.value_at(2), Value::Null);
    }

    #[test]
    fn two_vectors_of_different_lengths_are_caught() {
        let left = Vector::constant(LogicalType::Integer, Value::Integer(1), 4);
        let right = Vector::constant(LogicalType::Integer, Value::Integer(1), 5);
        let error = compare(Comparison::Equal, &left, &right).expect_err("ragged");
        assert!(error.message().contains("4 row vector"), "{error}");
    }

    #[test]
    fn turning_a_comparison_around_is_what_the_other_side_would_have_said() {
        for op in EVERY {
            let left = Value::Integer(3);
            let right = Value::Integer(7);
            assert_eq!(
                compare_values(op, &left, &right).expect("compares"),
                compare_values(op.swapped(), &right, &left).expect("compares"),
                "{op:?}"
            );
        }
    }

    /// The whole point of the rewrite, stated as a property. Every operator, every physical
    /// layout, every form pair the fast path claims, against the row at a time oracle.
    #[test]
    fn every_specialized_path_agrees_with_the_row_at_a_time_path() {
        let mut rng = Rng(0x5eed_1234_9876_4321);
        let types: [LogicalType; 11] = [
            LogicalType::Boolean,
            LogicalType::TinyInt,
            LogicalType::SmallInt,
            LogicalType::Integer,
            LogicalType::BigInt,
            LogicalType::HugeInt,
            LogicalType::UInteger,
            LogicalType::Float,
            LogicalType::Double,
            LogicalType::Varchar,
            LogicalType::Interval,
        ];
        for ty in &types {
            for nulls in [0u64, 1, 3] {
                let len = 37;
                let make = |rng: &mut Rng| {
                    let values: Vec<Value> = (0..len)
                        .map(|_| {
                            if nulls > 0 && rng.below(nulls + 1) == 0 {
                                Value::Null
                            } else {
                                sample(ty, rng)
                            }
                        })
                        .collect();
                    Vector::from_values(ty.clone(), &values).expect("a flat vector")
                };
                let left = make(&mut rng);
                let right = make(&mut rng);
                let literal = sample(ty, &mut rng);
                let constant = Vector::constant(ty.clone(), literal, len);
                let null_constant = Vector::constant(ty.clone(), Value::Null, len);
                let codes: Vec<u32> =
                    (0..len).map(|_| rng.below(left.len() as u64) as u32).collect();
                let dictionary =
                    Vector::dictionary(codes, left.clone()).expect("codes are in range");
                // Runs over the same values, with the last one cut short so that a run boundary
                // does not land on the end of the vector.
                let ends: Vec<u32> = (1..=left.len())
                    .map(|run| ((run * len) / left.len()).max(run) as u32)
                    .collect();
                let runs =
                    Vector::runs(ends.clone(), left.clone()).expect("one value for each run");
                // A second pair over the other column's values, so that a coded side against a
                // coded side is two different runs of values reached through two different sets of
                // codes rather than one set read twice.
                let other_codes: Vec<u32> =
                    (0..len).map(|_| rng.below(right.len() as u64) as u32).collect();
                let other_dictionary =
                    Vector::dictionary(other_codes, right.clone()).expect("codes are in range");
                let other_runs = Vector::runs(ends, right.clone()).expect("one value for each run");

                for op in EVERY {
                    agrees_and_selects(op, &left, &right);
                    agrees_and_selects(op, &left, &constant);
                    agrees_and_selects(op, &constant, &left);
                    agrees_and_selects(op, &left, &null_constant);
                    agrees_and_selects(op, &null_constant, &left);
                    agrees_and_selects(op, &dictionary, &constant);
                    agrees_and_selects(op, &constant, &dictionary);
                    // The dictionary against a flat column, which reads a null from either side and
                    // from the dictionary's values as well, so it is the pair with the most ways to
                    // disagree with the oracle and the one that got a loop last.
                    agrees_and_selects(op, &dictionary, &right);
                    agrees_and_selects(op, &right, &dictionary);
                    // The same four pairings for run length, which reaches the same loops through
                    // the same accessor, so what is being checked is that the positions it works
                    // out are the positions the row at a time path reads.
                    agrees_and_selects(op, &runs, &constant);
                    agrees_and_selects(op, &constant, &runs);
                    agrees_and_selects(op, &runs, &right);
                    agrees_and_selects(op, &right, &runs);
                    // Both sides reached through codes, which is the pair TPC-H actually hits:
                    // `l_commitdate < l_receiptdate` is two dictionary encoded columns of one
                    // table. A null here can be in four places at once, the two dictionaries' own
                    // masks and the two value vectors, and the answer has to be null if it is in
                    // any of them.
                    agrees_and_selects(op, &dictionary, &other_dictionary);
                    agrees_and_selects(op, &runs, &other_runs);
                    agrees_and_selects(op, &dictionary, &other_runs);
                    agrees_and_selects(op, &runs, &other_dictionary);
                }
            }
        }
    }

    /// The rows of a selection the row at a time path keeps, which is what [`refine`] has to say.
    fn refined(op: Comparison, left: &Vector, right: &Vector, kept: &Selection) -> Selection {
        let mut out = Vec::new();
        for &row in kept.indices() {
            let index = row as usize;
            let answer = compare_values(op, &left.value_at(index), &right.value_at(index))
                .expect("the oracle is only asked about types that compare");
            if is_true(&answer) {
                out.push(row);
            }
        }
        Selection::from_indices(out)
    }

    fn threads(op: Comparison, left: &Vector, right: &Vector, kept: &Selection) {
        let fast = refine(op, left, right, kept).expect("compares");
        assert_eq!(
            fast,
            refined(op, left, right, kept),
            "{op:?} on a {:?} against a {:?} over {} rows",
            left.form(),
            right.form(),
            kept.len()
        );
    }

    /// Threading a selection through a comparison is the same rows as comparing everything and
    /// then keeping the ones that were already kept. Every operator, every form pair that has a
    /// loop, at four densities of selection, against the row at a time path.
    #[test]
    fn a_threaded_comparison_keeps_what_the_row_at_a_time_path_keeps() {
        let mut rng = Rng(0x5eed_4321_1234_9876);
        let types = [LogicalType::Integer, LogicalType::Double, LogicalType::Varchar];
        for ty in &types {
            for nulls in [0u64, 1, 3] {
                let len = 37;
                let make = |rng: &mut Rng| {
                    let values: Vec<Value> = (0..len)
                        .map(|_| {
                            if nulls > 0 && rng.below(nulls + 1) == 0 {
                                Value::Null
                            } else {
                                sample(ty, rng)
                            }
                        })
                        .collect();
                    Vector::from_values(ty.clone(), &values).expect("a flat vector")
                };
                let left = make(&mut rng);
                let right = make(&mut rng);
                let constant = Vector::constant(ty.clone(), sample(ty, &mut rng), len);
                let null_constant = Vector::constant(ty.clone(), Value::Null, len);
                let codes: Vec<u32> =
                    (0..len).map(|_| rng.below(left.len() as u64) as u32).collect();
                let dictionary =
                    Vector::dictionary(codes, left.clone()).expect("codes are in range");

                // Everything, every third row, a handful including the last one, and nothing,
                // which is the state a conjunct chain reaches as soon as one conjunct rejects a
                // whole chunk and is the case where the loop below must not read anything at all.
                let selections = [
                    Selection::identity(len),
                    Selection::from_indices((0..len as u32).filter(|row| row % 3 == 0).collect()),
                    Selection::from_indices(vec![2, 5, 6, 17, 36]),
                    Selection::empty(),
                ];
                for op in EVERY {
                    for kept in &selections {
                        threads(op, &left, &right, kept);
                        threads(op, &left, &constant, kept);
                        threads(op, &constant, &left, kept);
                        threads(op, &left, &null_constant, kept);
                        threads(op, &null_constant, &left, kept);
                        threads(op, &constant, &null_constant, kept);
                        threads(op, &dictionary, &constant, kept);
                        threads(op, &constant, &dictionary, kept);
                        threads(op, &dictionary, &right, kept);
                        threads(op, &right, &dictionary, kept);
                    }
                }
            }
        }
    }

    /// Two conjuncts threaded one after the other are the rows both of them keep, which is the
    /// property the whole filter path rests on. The second comparison sees the rows the first one
    /// left and never looks at the others.
    #[test]
    fn a_second_conjunct_reads_only_what_the_first_one_left() {
        let numbers: Vec<Value> = (0..64).map(|row| Value::Integer(row % 10)).collect();
        let column = Vector::from_values(LogicalType::Integer, &numbers).expect("a flat vector");
        let three = Vector::constant(LogicalType::Integer, Value::Integer(3), 64);
        let seven = Vector::constant(LogicalType::Integer, Value::Integer(7), 64);

        let first = refine(Comparison::Greater, &column, &three, &Selection::identity(64))
            .expect("compares");
        let both = refine(Comparison::Less, &column, &seven, &first).expect("compares");

        let expected: Vec<u32> = (0..64)
            .filter(|row| {
                let value = row % 10;
                value > 3 && value < 7
            })
            .collect();
        assert_eq!(both.indices(), expected.as_slice());
        assert!(both.len() < first.len(), "the second conjunct narrowed the selection");
    }

    /// A null is not a true, so a threaded comparison drops the row rather than keeping it with an
    /// unknown answer. This is the rule that makes `WHERE a < 5` leave out the rows where `a` is
    /// null, and it is the one a branchless loop gets wrong if the validity is left out of it.
    #[test]
    fn a_null_row_is_not_kept_by_an_ordinary_comparison_and_is_by_a_total_one() {
        let column = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Null, Value::Integer(3), Value::Null],
        )
        .expect("four rows");
        let cut = Vector::constant(LogicalType::Integer, Value::Integer(2), 4);
        let all = Selection::identity(4);
        assert_eq!(
            refine(Comparison::Less, &column, &cut, &all).expect("compares").indices(),
            &[0]
        );
        // The total comparison has an answer at every row, so the two nulls are kept here.
        let nulls = Vector::constant(LogicalType::Integer, Value::Null, 4);
        assert_eq!(
            refine(Comparison::NotDistinctFrom, &column, &nulls, &all).expect("compares").indices(),
            &[1, 3]
        );
    }

    #[test]
    fn a_selection_past_the_end_is_caught() {
        let column = Vector::constant(LogicalType::Integer, Value::Integer(1), 4);
        let past = Selection::from_indices(vec![0, 4]);
        let error = refine(Comparison::Equal, &column, &column, &past).expect_err("out of range");
        assert!(error.message().contains("4 row vector"), "{error}");
    }

    /// One value of a type, for the generator above.
    fn sample(ty: &LogicalType, rng: &mut Rng) -> Value {
        match ty {
            LogicalType::Boolean => Value::Boolean(rng.below(2) == 1),
            LogicalType::TinyInt => Value::TinyInt(rng.below(7) as i8 - 3),
            LogicalType::SmallInt => Value::SmallInt(rng.below(11) as i16 - 5),
            LogicalType::Integer => Value::Integer(rng.below(9) as i32 - 4),
            LogicalType::BigInt => Value::BigInt(rng.below(9) as i64 - 4),
            LogicalType::HugeInt => Value::HugeInt(i128::from(rng.below(9)) - 4),
            LogicalType::UInteger => Value::UInteger(rng.below(9) as u32),
            // A NaN and a negative zero in the pool on purpose, because DuckDB's float order is
            // not IEEE's and the fast path has to reach the same answer the oracle does.
            LogicalType::Float => Value::Float(match rng.below(5) {
                0 => f32::NAN,
                1 => -0.0,
                other => other as f32 - 2.0,
            }),
            LogicalType::Double => Value::Double(match rng.below(5) {
                0 => f64::NAN,
                1 => -0.0,
                other => other as f64 - 2.0,
            }),
            // The same length written three ways and two lengths that are close to it, because an
            // interval that compares as a triple gets every pair here wrong and one that compares
            // as a length gets them right.
            LogicalType::Interval => match rng.below(6) {
                0 => Value::Interval { months: 0, days: 1, micros: 0 },
                1 => Value::Interval { months: 0, days: 0, micros: 86_400_000_000 },
                2 => Value::Interval { months: 1, days: -29, micros: 86_400_000_000 },
                3 => Value::Interval { months: 1, days: 0, micros: 0 },
                4 => Value::Interval { months: 0, days: 0, micros: 90_000_000_000 },
                _ => Value::Interval { months: -1, days: 0, micros: 0 },
            },
            // Short, at the inline limit, over it, and sharing a prefix with each other, which is
            // where a comparison that trusts the prefix too far goes wrong.
            LogicalType::Varchar => Value::Varchar(
                match rng.below(6) {
                    0 => "",
                    1 => "ab",
                    2 => "abc",
                    3 => "abcdefghijkl",
                    4 => "abcdefghijklm",
                    _ => "abcdefghijklmnopqrstuvwxyz",
                }
                .to_owned(),
            ),
            other => panic!("the generator has no values for {other}"),
        }
    }

    /// The prefix lemma, written as a test because the whole string path rests on it. A view pads
    /// a short string with zeros, so prefix order has to agree with byte order on every pair where
    /// the prefixes differ, including the pairs where one string is shorter than four bytes.
    #[test]
    fn prefix_order_is_byte_order_whenever_the_prefixes_differ() {
        let words =
            ["", "a", "ab", "abc", "abcd", "abcde", "b", "abcdefghijklmnop", "abcdefghijklmnoq"];
        let mut column = StringColumn::new();
        for word in words {
            column.push(word);
        }
        for (i, one) in words.iter().enumerate() {
            for (j, other) in words.iter().enumerate() {
                assert_eq!(
                    string_order(&column, i, &column, j),
                    one.as_bytes().cmp(other.as_bytes()),
                    "{one:?} against {other:?}"
                );
            }
        }
    }

    /// A dictionary is compared once per distinct value, not once per row, and it has to reach the
    /// same answer including for the nulls it keeps in the vector it points at.
    #[test]
    fn a_dictionary_against_a_constant_reads_its_nulls_from_the_values() {
        let values = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Null, Value::Integer(9)],
        )
        .expect("three values");
        let dictionary =
            Vector::dictionary(vec![0, 1, 2, 1, 0], values).expect("codes are in range");
        let constant = Vector::constant(LogicalType::Integer, Value::Integer(5), 5);
        let result = compare(Comparison::Less, &dictionary, &constant).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(true));
        assert_eq!(result.value_at(1), Value::Null);
        assert_eq!(result.value_at(2), Value::Boolean(false));
        assert_eq!(result.value_at(3), Value::Null);
        assert_eq!(result.value_at(4), Value::Boolean(true));
    }

    /// An ordering comparison on a text column reads the bytes where they lie rather than building a
    /// `Value` a row.
    ///
    /// The assertion that matters is the counter at the end. The answers were already right through
    /// the row at a time path, and what was wrong was the cost: `ORDER BY <varchar> LIMIT 10` asks
    /// every chunk whether any row in it can still beat the worst candidate the top N holds, and that
    /// question used to allocate a `String` for every row of every chunk. On the ClickBench file that
    /// was eighteen times the CPU of the same query with an integer sort key.
    #[test]
    fn an_ordering_on_text_against_a_literal_does_not_fall_back() {
        let before = fallback::count(Kernel::Compare, Form::Dictionary, Form::Constant);
        let values = Vector::from_values(
            LogicalType::Varchar,
            &[Value::Varchar("apple".into()), Value::Null, Value::Varchar("pear".into())],
        )
        .expect("three values");
        let column = Vector::dictionary(vec![0, 1, 2, 0], values).expect("codes are in range");
        let cut = Vector::constant(LogicalType::Varchar, Value::Varchar("melon".into()), 4);
        let result = compare(Comparison::Less, &column, &cut).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(true));
        assert_eq!(result.value_at(1), Value::Null);
        assert_eq!(result.value_at(2), Value::Boolean(false));
        assert_eq!(result.value_at(3), Value::Boolean(true));
        // The literal on the left, which is the same question with the comparison turned around, and
        // it has to be turned around once rather than once a row.
        let other = compare(Comparison::Greater, &cut, &column).expect("compares");
        assert_eq!(other.value_at(0), Value::Boolean(true));
        assert_eq!(other.value_at(1), Value::Null);
        assert_eq!(other.value_at(2), Value::Boolean(false));
        // The selection entry point, which is the one the filter and the top N reach.
        let kept = refine(Comparison::GreaterOrEqual, &column, &cut, &Selection::identity(4))
            .expect("refines");
        assert_eq!(kept.indices(), [2]);
        assert_eq!(fallback::count(Kernel::Compare, Form::Dictionary, Form::Constant), before);
    }

    /// Two dictionary columns compared with each other, which is the pair the TPC-H suite spends
    /// the most row at a time calls on and had no loop until it was counted.
    ///
    /// `l_commitdate < l_receiptdate` in q4, q12 and q21 is two dictionary encoded date columns of
    /// six million rows each, and every one of those rows was going through the generic comparison
    /// on a pair of freshly allocated `Value`s. The shape here is the same one: different codes,
    /// different values, and a null reached through the codes rather than sitting at the row.
    #[test]
    fn two_dictionary_columns_compared_with_each_other_do_not_fall_back() {
        let before = fallback::count(Kernel::Compare, Form::Dictionary, Form::Dictionary);
        let dates = |values: &[Value]| {
            Vector::from_values(LogicalType::Date, values).expect("a flat date column")
        };
        // 10, 20, 10, 30 against 15, null, 5, 15.
        let committed = Vector::dictionary(
            vec![0, 1, 0, 2],
            dates(&[Value::Date(10), Value::Date(20), Value::Date(30)]),
        )
        .expect("codes are in range");
        let received = Vector::dictionary(
            vec![0, 1, 2, 0],
            dates(&[Value::Date(15), Value::Null, Value::Date(5)]),
        )
        .expect("codes are in range");
        let result = compare(Comparison::Less, &committed, &received).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(true), "10 < 15");
        assert_eq!(result.value_at(1), Value::Null, "20 against a null");
        assert_eq!(result.value_at(2), Value::Boolean(false), "10 against 5");
        assert_eq!(result.value_at(3), Value::Boolean(false), "30 against 15");
        // The selection entry point, which is the one the filter reaches and the one q4 is made of.
        let kept = refine(Comparison::Less, &committed, &received, &Selection::identity(4))
            .expect("refines");
        assert_eq!(kept.indices(), [0]);
        assert_eq!(fallback::count(Kernel::Compare, Form::Dictionary, Form::Dictionary), before);
    }

    /// A form pair with no loop is answered correctly and counted, which is the whole contract of
    /// the fallback counter. Sequence against a column is the one this file leaves out on purpose.
    #[test]
    fn a_form_pair_with_no_loop_is_still_right_and_says_so() {
        // The counters are per thread in a test build, so this reads its own and nothing else's.
        let before = fallback::count(Kernel::Compare, Form::Sequence, Form::Flat);
        let sequence = Vector::sequence(10, 1, 4);
        let flat = Vector::from_values(
            LogicalType::BigInt,
            &[Value::BigInt(9), Value::BigInt(11), Value::BigInt(12), Value::Null],
        )
        .expect("four rows");
        let result = compare(Comparison::Less, &sequence, &flat).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(false));
        assert_eq!(result.value_at(1), Value::Boolean(false));
        assert_eq!(result.value_at(2), Value::Boolean(false));
        assert_eq!(result.value_at(3), Value::Null);
        assert!(fallback::count(Kernel::Compare, Form::Sequence, Form::Flat) > before);
    }

    /// The reason `Vector::dictionary` composes rather than stacks, stated as the thing that breaks
    /// if it stops.
    ///
    /// Every loop in this file reaches for the values behind the codes with `Vector::data`, and a
    /// dictionary pointing at a dictionary has no data to hand back, so a second filter over an
    /// already filtered chunk used to turn every one of these kernels off and drop the comparison
    /// onto the row at a time path. Measured on server3 over a chunk of two numeric columns that was
    /// selected twice, that was 3.5 nanoseconds a row becoming 104, and a third and fourth level
    /// cost nothing more because the first one had already given up everything there was to give.
    #[test]
    fn a_second_level_of_codes_does_not_turn_the_loops_off() {
        let before = fallback::count(Kernel::Compare, Form::Dictionary, Form::Constant);
        let values = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(5), Value::Integer(9)],
        )
        .expect("three rows");
        let once = Vector::dictionary(vec![2, 1, 0], values).expect("codes are in range");
        let twice = Vector::dictionary(vec![1, 2], once).expect("codes are in range");
        let cut = Vector::constant(LogicalType::Integer, Value::Integer(4), 2);
        let result = compare(Comparison::Greater, &twice, &cut).expect("compares");
        assert_eq!(result.value_at(0), Value::Boolean(true));
        assert_eq!(result.value_at(1), Value::Boolean(false));
        assert_eq!(fallback::count(Kernel::Compare, Form::Dictionary, Form::Constant), before);
    }

    /// Either side all null, on one of the six ordinary comparisons, is every answer null without
    /// the data being read. The vector this produces has to be the one the oracle produces, which
    /// is a flat run of falses under an all invalid validity rather than a constant.
    #[test]
    fn a_side_that_is_entirely_null_answers_without_reading_the_other() {
        let nulls = Vector::constant(LogicalType::Integer, Value::Null, 6);
        let flat = Vector::from_values(
            LogicalType::Integer,
            &[
                Value::Integer(1),
                Value::Integer(2),
                Value::Integer(3),
                Value::Integer(4),
                Value::Integer(5),
                Value::Integer(6),
            ],
        )
        .expect("six rows");
        agrees(Comparison::Less, &nulls, &flat);
        agrees(Comparison::Equal, &flat, &nulls);
        assert_eq!(
            compare(Comparison::Less, &nulls, &flat).expect("compares").validity(),
            &Validity::AllInvalid
        );
    }

    /// An empty vector is not a special case anywhere, and the easiest way to keep it that way is
    /// to say so in a test rather than to find out from a panic in an operator.
    #[test]
    fn an_empty_comparison_is_an_empty_answer() {
        let left = Vector::from_values(LogicalType::Integer, &[]).expect("no rows");
        let right = Vector::constant(LogicalType::Integer, Value::Integer(1), 0);
        let result = compare(Comparison::Equal, &left, &right).expect("compares");
        assert_eq!(result.len(), 0);
    }

    /// Six strings, three of them sharing a prefix, and a null, which is the column the two tests
    /// below read.
    fn words() -> Vector {
        Vector::from_values(
            LogicalType::Varchar,
            &[
                Value::Varchar("http://a".into()),
                Value::Varchar("http://b".into()),
                Value::Null,
                Value::Varchar("ab".into()),
                Value::Varchar("http://a".into()),
                Value::Varchar("z".into()),
            ],
        )
        .expect("six rows")
    }

    /// A literal built early answers what a literal built per chunk answers.
    ///
    /// Every operator and both entry points, because the whole claim of the prepared literal is
    /// that it changes nothing, and the string column is the one where it changes the most work:
    /// what it carries is the four byte prefix the comparison resolves almost every row from.
    #[test]
    fn a_literal_built_early_answers_what_one_built_here_answers() {
        let column = words();
        let value = Value::Varchar("http://b".into());
        let constant = Vector::constant(LogicalType::Varchar, value.clone(), column.len());
        let held = Held::of(&LogicalType::Varchar, &value).expect("a varchar has a column");
        let kept = Selection::from_indices(vec![0, 1, 3, 5]);
        for op in [
            Comparison::Equal,
            Comparison::NotEqual,
            Comparison::Less,
            Comparison::LessOrEqual,
            Comparison::Greater,
            Comparison::GreaterOrEqual,
            Comparison::DistinctFrom,
            Comparison::NotDistinctFrom,
        ] {
            let prepared = compare_prepared(op, &column, &constant, Some(&held)).expect("compares");
            assert_eq!(prepared, compare(op, &column, &constant).expect("compares"), "{op:?}");
            // And with the literal on the left, which is the same loop turned around.
            let flipped = compare_prepared(op, &constant, &column, Some(&held)).expect("compares");
            assert_eq!(flipped, compare(op, &constant, &column).expect("compares"), "{op:?}");
            let refined =
                refine_prepared(op, &column, &constant, &kept, Some(&held)).expect("refines");
            assert_eq!(refined, refine(op, &column, &constant, &kept).expect("refines"), "{op:?}");
        }
    }

    /// A literal built for something else is ignored rather than believed.
    ///
    /// The caller in `rudb-exec` takes the value out of the step it hands the answer back with, so
    /// this cannot happen there, and the kernel is public. A wrong answer is a much worse failure
    /// than a column built per chunk, so the check is a value comparison per chunk and this is what
    /// says it works.
    #[test]
    fn a_literal_built_for_another_value_is_ignored() {
        let column = words();
        let constant = Vector::constant(LogicalType::Varchar, Value::Varchar("z".into()), 6);
        let wrong = Held::of(&LogicalType::Varchar, &Value::Varchar("ab".into()))
            .expect("a varchar has a column");
        let answer = compare_prepared(Comparison::Equal, &column, &constant, Some(&wrong))
            .expect("compares");
        assert_eq!(answer, compare(Comparison::Equal, &column, &constant).expect("compares"));
        // And one built for another type, which is what a comparison across two types would hand
        // over if the caller took it from the wrong side.
        let other = Held::of(&LogicalType::Integer, &Value::Integer(1)).expect("an integer column");
        let answer = compare_prepared(Comparison::Equal, &column, &constant, Some(&other))
            .expect("compares");
        assert_eq!(answer, compare(Comparison::Equal, &column, &constant).expect("compares"));
    }

    /// A range answered in one pass keeps the rows the two comparisons keep one after the other,
    /// flat and bit packed, strict and not, from every row and from some, and for ends that cut
    /// past the column's values or meet nowhere.
    #[test]
    fn a_range_keeps_the_rows_its_two_comparisons_keep() {
        let values: Vec<i32> = (0..1000).map(|row| 700 + (row * 37) % 600).collect();
        let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into()))
            .expect("integers are an i32 layout");
        let packed = flat.bit_packed().expect("packs");
        let some = Selection::from_predicate(1000, |row| row % 3 != 1);
        let ends = [i32::MIN, -5, 699, 700, 701, 900, 1299, 1300, 5000, i32::MAX];
        for column in [&flat, &packed] {
            for low in ends {
                for high in ends {
                    for (lower, upper) in [
                        (Comparison::GreaterOrEqual, Comparison::LessOrEqual),
                        (Comparison::Greater, Comparison::Less),
                    ] {
                        let (low, high) = (Value::Integer(low), Value::Integer(high));
                        let at = |value: &Value| {
                            Vector::constant(LogicalType::Integer, value.clone(), 1000)
                        };
                        let (one, other) = (at(&low), at(&high));
                        let bounds = (
                            Bound { op: lower, value: &low, held: None },
                            Bound { op: upper, value: &high, held: None },
                        );
                        let first = select_prepared(lower, column, &one, None).expect("compares");
                        let both = refine(upper, column, &other, &first).expect("refines");
                        let ranged = select_range(column, bounds.0, bounds.1, None)
                            .expect("a column with no nulls has a range");
                        assert_eq!(
                            ranged.indices(),
                            both.indices(),
                            "{low} {lower:?}, {high} {upper:?}"
                        );
                        let first = refine(lower, column, &one, &some).expect("refines");
                        let both = refine(upper, column, &other, &first).expect("refines");
                        let ranged = select_range(column, bounds.0, bounds.1, Some(some.indices()))
                            .expect("a column with no nulls has a range");
                        assert_eq!(
                            ranged.indices(),
                            both.indices(),
                            "{low} {lower:?}, {high} {upper:?} of some"
                        );
                    }
                }
            }
        }
    }

    /// Nulls, a literal of another type and a string column are left to the two comparisons.
    #[test]
    fn a_range_it_has_no_loop_for_is_none() {
        let low = Value::Integer(1);
        let high = Value::Integer(5);
        fn bound(op: Comparison, value: &Value) -> Bound<'_> {
            Bound { op, value, held: None }
        }
        let nulls = Vector::from_values(LogicalType::Integer, &[Value::Integer(2), Value::Null])
            .expect("integers");
        let range = |column: &Vector, low: &Value, high: &Value| {
            select_range(
                column,
                bound(Comparison::GreaterOrEqual, low),
                bound(Comparison::Less, high),
                None,
            )
        };
        assert!(range(&nulls, &low, &high).is_none());
        let plain =
            Vector::from_values(LogicalType::Integer, &[Value::Integer(2), Value::Integer(9)])
                .expect("integers");
        assert_eq!(range(&plain, &low, &high).expect("a range").indices(), &[0]);
        assert!(range(&plain, &Value::Null, &high).is_none());
        assert!(range(&words(), &low, &high).is_none());
        let wrong = select_range(
            &plain,
            bound(Comparison::Less, &low),
            bound(Comparison::Less, &high),
            None,
        );
        assert!(wrong.is_none(), "a low end has to be > or >=");
    }

    /// A bit packed column against a literal is compared in code space, which has to reach the
    /// oracle's answer on all eight comparisons and with the literal on either side.
    #[test]
    fn a_packed_column_against_a_constant_answers_what_the_oracle_answers() {
        let values: Vec<i32> = (0..64).map(|row| 1000 + (row * 37) % 500).collect();
        let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into()))
            .expect("integers are an i32 layout");
        let packed = flat.bit_packed().expect("a five hundred wide range packs");
        assert_eq!(packed.form(), Form::BitPacked);
        for literal in [999, 1000, 1200, 1499, 1500, 2000] {
            let constant = Vector::constant(LogicalType::Integer, Value::Integer(literal), 64);
            for op in EVERY {
                agrees(op, &packed, &constant);
                agrees(op, &constant, &packed);
            }
        }
    }

    /// The nulls of a packed column live in its validity rather than in its bits, so a comparison
    /// has to blank them the way it blanks a flat column's, and the bits under them are whatever
    /// the packing wrote there.
    #[test]
    fn a_packed_column_with_nulls_answers_what_the_oracle_answers() {
        let values: Vec<i32> = (0..32).map(|row| 40 + row * 3).collect();
        let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into()))
            .expect("integers are an i32 layout")
            .with_validity(Validity::from_iter(32, |row| row % 5 != 0));
        let packed = flat.bit_packed().expect("packs");
        let constant = Vector::constant(LogicalType::Integer, Value::Integer(80), 32);
        for op in EVERY {
            agrees(op, &packed, &constant);
        }
    }

    /// A literal the width cannot hold answers every row without a bit being read, and the answer
    /// still has to be the one the oracle gives.
    #[test]
    fn a_literal_outside_the_packed_range_answers_the_whole_vector_at_once() {
        let before = fallback::count(Kernel::Compare, Form::BitPacked, Form::Constant);
        let values: Vec<i32> = (0..16).map(|row| 500 + row).collect();
        let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.into()))
            .expect("integers are an i32 layout");
        let packed = flat.bit_packed().expect("packs");
        let literals = [-1, 0, 499, 516, 100_000];
        for literal in literals {
            let constant = Vector::constant(LogicalType::Integer, Value::Integer(literal), 16);
            for op in EVERY {
                agrees(op, &packed, &constant);
            }
        }
        // The six ordinary comparisons have a loop for this pair and the two that never go null do
        // not, because those want the null rule inside the loop and the code space loop does not
        // carry one. They take the row at a time path and count themselves, which is the counter
        // doing its job rather than a gap being hidden.
        let total = EVERY.iter().filter(|op| op.is_total()).count();
        assert_eq!(
            fallback::count(Kernel::Compare, Form::BitPacked, Form::Constant) - before,
            (literals.len() * total) as u64,
            "only the two total comparisons fall through"
        );
    }

    /// The conjunct path reads the rows an earlier conjunct kept, so the code space loop has to be
    /// reached through the selection rather than through the row number.
    #[test]
    fn refining_a_selection_over_a_packed_column_keeps_the_same_rows() {
        let values: Vec<i32> = (0..64).map(|row| 200 + (row * 11) % 128).collect();
        let flat = Vector::flat(LogicalType::Integer, Data::Int32(values.clone().into()))
            .expect("integers are an i32 layout");
        let packed = flat.bit_packed().expect("packs");
        let kept = Selection::from_predicate(64, |row| row % 3 == 0);
        let constant = Vector::constant(LogicalType::Integer, Value::Integer(260), 64);
        let packed_rows = refine(Comparison::Greater, &packed, &constant, &kept).expect("refines");
        let flat_rows = refine(Comparison::Greater, &flat, &constant, &kept).expect("refines");
        assert_eq!(packed_rows.indices(), flat_rows.indices());
        assert!(!packed_rows.is_empty(), "the literal is inside the range");
    }

    /// Two packed columns whose ranges overlap, which is the pair a stored table produces and the
    /// pair `l_commitdate < l_receiptdate` is. The two bases differ, so the answer has to come out
    /// of the sums rather than out of the codes, and it has to be the oracle's answer.
    #[test]
    fn two_packed_columns_against_each_other_answer_what_the_oracle_answers() {
        let before = fallback::count(Kernel::Compare, Form::BitPacked, Form::BitPacked);
        let one: Vec<i32> = (0..64).map(|row| 9000 + (row * 37) % 500).collect();
        let other: Vec<i32> = (0..64).map(|row| 9200 + (row * 53) % 400).collect();
        let left = Vector::flat(LogicalType::Integer, Data::Int32(one.into()))
            .expect("integers are an i32 layout")
            .bit_packed()
            .expect("a five hundred wide range packs");
        let right = Vector::flat(LogicalType::Integer, Data::Int32(other.into()))
            .expect("integers are an i32 layout")
            .bit_packed()
            .expect("a four hundred wide range packs");
        assert_eq!(left.form(), Form::BitPacked);
        assert_eq!(right.form(), Form::BitPacked);
        assert_ne!(
            left.packed_parts().expect("packed").base(),
            right.packed_parts().expect("packed").base(),
            "the two bases are the two column minimums and this test wants them apart"
        );
        for op in EVERY {
            agrees(op, &left, &right);
            agrees(op, &right, &left);
        }
        // The two total comparisons want the null rule inside the loop and the code space loop does
        // not carry one, so those are the only ones that count themselves, the same way they do for
        // a packed column against a literal.
        let total = EVERY.iter().filter(|op| op.is_total()).count();
        assert_eq!(
            fallback::count(Kernel::Compare, Form::BitPacked, Form::BitPacked) - before,
            (total * 2) as u64,
            "only the two total comparisons fall through"
        );
    }

    #[test]
    fn two_long_packed_columns_sliced_off_a_word_answer_what_the_oracle_answers() {
        // Past one block of 64 with a ragged tail, and sliced so that the rows do not start on a
        // word, which are the two ways the unpacked runs could come out misaligned.
        let one: Vec<i32> = (0..1000).map(|row| 9000 + (row * 37) % 500).collect();
        let other: Vec<i32> = (0..1000).map(|row| 9200 + (row * 53) % 400).collect();
        let packed = |values: Vec<i32>| {
            Vector::flat(LogicalType::Integer, Data::Int32(values.into()))
                .expect("integers are an i32 layout")
                .bit_packed()
                .expect("a narrow range packs")
        };
        let (left, right) = (packed(one), packed(other));
        for (at, len) in [(0, 1000), (7, 900), (70, 129)] {
            let (left, right) = (
                left.slice(at, len).expect("inside the vector"),
                right.slice(at, len).expect("inside the vector"),
            );
            for op in EVERY {
                agrees(op, &left, &right);
                agrees(op, &right, &left);
            }
        }
    }

    /// The rows the oracle's flags keep out of `kept`, or out of every row.
    fn wanted(op: Comparison, left: &Vector, right: &Vector, kept: Option<&Selection>) -> Vec<u32> {
        let flags = oracle(op, left, right);
        let every = crate::select::selection(&flags, flags.len());
        match kept {
            None => every.indices().to_vec(),
            Some(kept) => {
                kept.indices().iter().copied().filter(|row| every.indices().contains(row)).collect()
            }
        }
    }

    /// Two date columns of a thousand rows, flat and packed, compared the way a filter's first
    /// conjunct and a later one compare them. A thousand rows is whole blocks of sixty four and a
    /// tail, and the later conjunct is asked over a dense selection, which unpacks, and a sparse
    /// one, which reads a code at a time.
    #[test]
    fn two_columns_compared_in_one_pass_keep_the_rows_the_oracle_keeps() {
        let one: Vec<i32> = (0..1000).map(|row| 9000 + (row * 37) % 500).collect();
        let other: Vec<i32> = (0..1000).map(|row| 9200 + (row * 53) % 400).collect();
        let flat = |values: Vec<i32>| {
            Vector::flat(LogicalType::Date, Data::Int32(values.into())).expect("dates are i32")
        };
        let (left, right) = (flat(one), flat(other));
        let (packed_left, packed_right) =
            (left.bit_packed().expect("packs"), right.bit_packed().expect("packs"));
        assert_eq!(packed_left.form(), Form::BitPacked);
        assert_eq!(packed_right.form(), Form::BitPacked);
        let dense = Selection::from_predicate(1000, |row| row % 3 != 0);
        let sparse = Selection::from_predicate(1000, |row| row % 97 == 5);
        for op in EVERY {
            for (a, b) in [(&left, &right), (&right, &left), (&packed_left, &packed_right)] {
                let picked = select_prepared(op, a, b, None).expect("selects");
                assert_eq!(picked.indices(), wanted(op, a, b, None), "{op:?} {:?}", a.form());
                for kept in [&dense, &sparse] {
                    let refined = refine(op, a, b, kept).expect("refines");
                    assert_eq!(refined.indices(), wanted(op, a, b, Some(kept)), "{op:?}");
                }
            }
        }
        let refined = refine(Comparison::Less, &packed_left, &packed_right, &sparse).expect("ok");
        assert!(!refined.is_empty(), "the sparse rows reach both answers");
    }

    /// A literal against a flat column in a later conjunct, which now picks its rows in the same
    /// pass that compares them.
    #[test]
    fn a_flat_column_refined_against_a_literal_keeps_the_rows_the_oracle_keeps() {
        let values: Vec<i64> = (0..700).map(|row| (row * 7919) % 1000).collect();
        let column = Vector::flat(LogicalType::BigInt, Data::Int64(values.into())).expect("i64");
        let literal = Vector::constant(LogicalType::BigInt, Value::BigInt(500), 700);
        let kept = Selection::from_predicate(700, |row| row % 4 == 1);
        for op in EVERY {
            let refined = refine(op, &column, &literal, &kept).expect("refines");
            assert_eq!(refined.indices(), wanted(op, &column, &literal, Some(&kept)), "{op:?}");
        }
    }

    /// A column whose largest value is below the other's smallest answers every row the same way,
    /// and the answer still has to be the one the oracle gives on all eight comparisons.
    #[test]
    fn two_packed_columns_whose_ranges_do_not_overlap_answer_the_whole_vector_at_once() {
        let one: Vec<i32> = (0..32).map(|row| 100 + row).collect();
        let other: Vec<i32> = (0..32).map(|row| 500 + row * 2).collect();
        let low = Vector::flat(LogicalType::Integer, Data::Int32(one.into()))
            .expect("integers are an i32 layout")
            .bit_packed()
            .expect("packs");
        let high = Vector::flat(LogicalType::Integer, Data::Int32(other.into()))
            .expect("integers are an i32 layout")
            .bit_packed()
            .expect("packs");
        for op in EVERY {
            agrees(op, &low, &high);
            agrees(op, &high, &low);
        }
    }

    /// The nulls of a packed column live in its validity rather than in its bits, so a pair of them
    /// has to blank the rows either side is null in, and the bits under those rows are whatever the
    /// packing wrote there.
    #[test]
    fn two_packed_columns_with_nulls_answer_what_the_oracle_answers() {
        let one: Vec<i32> = (0..32).map(|row| 40 + row * 3).collect();
        let other: Vec<i32> = (0..32).map(|row| 60 + row * 2).collect();
        let left = Vector::flat(LogicalType::Integer, Data::Int32(one.into()))
            .expect("integers are an i32 layout")
            .with_validity(Validity::from_iter(32, |row| row % 5 != 0))
            .bit_packed()
            .expect("packs");
        let right = Vector::flat(LogicalType::Integer, Data::Int32(other.into()))
            .expect("integers are an i32 layout")
            .with_validity(Validity::from_iter(32, |row| row % 3 != 0))
            .bit_packed()
            .expect("packs");
        for op in EVERY {
            agrees(op, &left, &right);
        }
    }

    /// A dictionary over a packed run on either side or both, which is what a stored date column
    /// comes back as and is the pair q04 and q21 fall through on. The nulls are in three places at
    /// once here, the outer mask, the dictionary's values and the rows a code repeats, and the
    /// oracle sees all three.
    #[test]
    fn a_dictionary_over_a_packed_run_answers_what_the_oracle_answers() {
        let before = fallback::count(Kernel::Compare, Form::Dictionary, Form::Dictionary);
        let distinct = |start: i32, step: i32, nulls: usize| {
            let values: Vec<i32> = (0..24).map(|row| start + row * step).collect();
            Vector::flat(LogicalType::Date, Data::Int32(values.into()))
                .expect("dates are an i32 layout")
                .with_validity(Validity::from_iter(24, |row| row % nulls != 0))
                .bit_packed()
                .expect("packs")
        };
        let codes =
            |seed: usize| -> Vec<u32> { (0..64).map(|row| ((row * seed) % 24) as u32).collect() };
        let one = Vector::dictionary(codes(7), distinct(9_000, 3, 5)).expect("codes are in range");
        let other =
            Vector::dictionary(codes(5), distinct(9_020, 2, 7)).expect("codes are in range");
        let straight = Vector::flat(
            LogicalType::Date,
            Data::Int32((0..64).map(|row| 9_010 + row).collect::<Vec<i32>>().into()),
        )
        .expect("dates are an i32 layout")
        .bit_packed()
        .expect("packs");
        assert_eq!(one.form(), Form::Dictionary);
        assert_eq!(straight.form(), Form::BitPacked);
        for op in EVERY {
            agrees(op, &one, &other);
            agrees(op, &one, &straight);
            agrees(op, &straight, &other);
        }
        // Three pairs, and only the two total comparisons of each are left to count themselves.
        let total = EVERY.iter().filter(|op| op.is_total()).count();
        assert_eq!(
            fallback::count(Kernel::Compare, Form::Dictionary, Form::Dictionary) - before,
            total as u64,
            "only the two total comparisons fall through"
        );
    }

    /// A bit packed column against a flat one, which is the pair a clustered table hands the filter.
    ///
    /// Sorting lineitem by month of `l_shipdate` leaves `l_commitdate` with few enough distinct
    /// values inside a partition to pack, while `l_receiptdate` arrives flat, so the better encoding
    /// the clustering buys was what turned `l_commitdate < l_receiptdate` into a `Value` a side a
    /// row. All four ways the two sides can be indexed are here, because a stored date column comes
    /// back as a dictionary over a packed run about as often as it comes back as a packed run on its
    /// own.
    #[test]
    fn a_packed_column_against_a_flat_one_answers_what_the_oracle_answers() {
        let before = fallback::count(Kernel::Compare, Form::BitPacked, Form::Flat);
        let dates = |start: i32, step: i32, nulls: usize| {
            Vector::flat(
                LogicalType::Date,
                Data::Int32((0..64).map(|row| start + row * step).collect::<Vec<i32>>().into()),
            )
            .expect("dates are an i32 layout")
            .with_validity(Validity::from_iter(64, |row| row % nulls != 0))
        };
        let codes =
            |seed: usize| -> Vec<u32> { (0..64).map(|row| ((row * seed) % 64) as u32).collect() };
        let packed = dates(9_000, 3, 5).bit_packed().expect("packs");
        let flat = dates(9_040, 2, 7);
        let over_packed = Vector::dictionary(codes(7), packed.clone()).expect("codes are in range");
        let over_flat = Vector::dictionary(codes(11), flat.clone()).expect("codes are in range");
        assert_eq!(packed.form(), Form::BitPacked);
        assert_eq!(flat.form(), Form::Flat);
        // The ranges overlap, which is what makes the comparison a pass over the rows rather than
        // arithmetic on four numbers, and this test wants the pass.
        assert!(packed.packed_parts().expect("packed").base() < 9_040 + 63 * 2);
        for op in EVERY {
            agrees(op, &packed, &flat);
            agrees(op, &flat, &packed);
            agrees(op, &packed, &over_flat);
            agrees(op, &over_packed, &flat);
            agrees(op, &over_packed, &over_flat);
        }
        // Only the two total comparisons are left to count themselves, the same way they are for a
        // pair of packed columns, and of the five pairs above only the first is packed against flat.
        let total = EVERY.iter().filter(|op| op.is_total()).count();
        assert_eq!(
            fallback::count(Kernel::Compare, Form::BitPacked, Form::Flat) - before,
            total as u64,
            "only the two total comparisons fall through"
        );
    }

    /// The conjunct path reads the rows an earlier conjunct kept, so a pair of packed columns has
    /// to be reached through the selection rather than through the row number, which is what q12
    /// does with its two date comparisons one after the other.
    #[test]
    fn refining_a_selection_over_two_packed_columns_keeps_the_same_rows() {
        let one: Vec<i32> = (0..64).map(|row| 200 + (row * 11) % 128).collect();
        let other: Vec<i32> = (0..64).map(|row| 240 + (row * 17) % 96).collect();
        let left = Vector::flat(LogicalType::Integer, Data::Int32(one.clone().into()))
            .expect("integers are an i32 layout");
        let right = Vector::flat(LogicalType::Integer, Data::Int32(other.clone().into()))
            .expect("integers are an i32 layout");
        let kept = Selection::from_predicate(64, |row| row % 3 == 0);
        let packed_rows = refine(
            Comparison::Less,
            &left.bit_packed().expect("packs"),
            &right.bit_packed().expect("packs"),
            &kept,
        )
        .expect("refines");
        let flat_rows = refine(Comparison::Less, &left, &right, &kept).expect("refines");
        assert_eq!(packed_rows.indices(), flat_rows.indices());
        assert!(!packed_rows.is_empty(), "the two ranges overlap");
    }

    /// A column of URLs, which is the shape the string view form exists for: a shared prefix that
    /// the four bytes in the view cannot settle, and payloads long enough to be in the arena.
    fn urls(count: usize) -> Vector {
        let mut rng = Rng(0x5eed_1234);
        let values: Vec<Value> = (0..count)
            .map(|_| {
                let host = rng.below(6);
                let path = rng.below(40);
                Value::Varchar(format!("http://example{host}.test/a/rather/long/path/{path}"))
            })
            .collect();
        Vector::from_values(LogicalType::Varchar, &values).expect("strings")
    }

    #[test]
    fn a_string_view_column_against_a_literal_answers_what_the_oracle_answers() {
        let before = fallback::count(Kernel::Compare, Form::StringView, Form::Constant);
        let shared = urls(64).shared_text().expect("shares");
        assert_eq!(shared.form(), Form::StringView);
        let literals = ["http://example3.test/a/rather/long/path/7", "a", "zzz", ""];
        for literal in literals {
            let value = Value::Varchar(literal.to_owned());
            let constant = Vector::constant(LogicalType::Varchar, value, 64);
            for op in EVERY {
                agrees(op, &shared, &constant);
                agrees(op, &constant, &shared);
            }
        }
        assert_eq!(
            fallback::count(Kernel::Compare, Form::StringView, Form::Constant),
            before,
            "the form has a loop of its own for every comparison"
        );
    }

    #[test]
    fn a_string_view_column_against_another_one_answers_what_the_oracle_answers() {
        let shared = urls(48).shared_text().expect("shares");
        let other = urls(48).shared_text().expect("shares");
        let flat = urls(48);
        for op in EVERY {
            agrees(op, &shared, &other);
            agrees(op, &shared, &flat);
            agrees(op, &flat, &shared);
        }
    }

    #[test]
    fn the_nulls_of_a_string_view_column_are_the_nulls_the_oracle_sees() {
        let shared = urls(32)
            .with_validity(Validity::from_iter(32, |row| row % 4 != 1))
            .shared_text()
            .expect("shares");
        let constant =
            Vector::constant(LogicalType::Varchar, Value::Varchar("http://example3".into()), 32);
        for op in EVERY {
            agrees(op, &shared, &constant);
        }
    }

    #[test]
    fn a_compressed_column_against_a_literal_answers_what_the_oracle_answers() {
        let before = fallback::count(Kernel::Compare, Form::Fsst, Form::Constant);
        let flat = urls(64);
        let coded = flat.clone().compressed().expect("compresses");
        assert_eq!(coded.form(), Form::Fsst);
        let present = match coded.value_at(9) {
            Value::Varchar(text) => text,
            other => panic!("a string column reads back strings, not {other:?}"),
        };
        for literal in [present.as_str(), "http://example3.test/nothing/like/it", ""] {
            let value = Value::Varchar(literal.to_owned());
            let constant = Vector::constant(LogicalType::Varchar, value, 64);
            for op in EVERY {
                agrees(op, &coded, &constant);
                agrees(op, &constant, &coded);
            }
        }
        // Equality has a loop in code space and the six comparisons that need an order do not,
        // because a symbol code says nothing about where its symbol sorts. Those decompress a row at
        // a time and count themselves, which is the counter doing its job rather than a gap hiding.
        let ordered = EVERY.len() - 2;
        assert_eq!(
            fallback::count(Kernel::Compare, Form::Fsst, Form::Constant) - before,
            (3 * ordered) as u64,
            "only the comparisons that need an order fall through"
        );
    }

    /// Equality in code space is only right if compressing is a function, so the same string always
    /// has the same codes and two different strings never do. This is that claim as a test.
    #[test]
    fn every_row_of_a_compressed_column_matches_itself_and_nothing_else() {
        let flat = urls(48);
        let coded = flat.clone().compressed().expect("compresses");
        for row in 0..48 {
            let constant = Vector::constant(LogicalType::Varchar, flat.value_at(row), 48);
            let equal = compare(Comparison::Equal, &coded, &constant).expect("compares");
            for other in 0..48 {
                let want = flat.value_at(other) == flat.value_at(row);
                assert_eq!(
                    equal.value_at(other),
                    Value::Boolean(want),
                    "row {row} against {other}"
                );
            }
        }
    }

    #[test]
    fn the_nulls_of_a_compressed_column_are_the_nulls_the_oracle_sees() {
        let coded = urls(32)
            .with_validity(Validity::from_iter(32, |row| row % 3 != 0))
            .compressed()
            .expect("compresses");
        let value = coded.value_at(1);
        let constant = Vector::constant(LogicalType::Varchar, value, 32);
        for op in EVERY {
            agrees(op, &coded, &constant);
        }
    }

    /// The two forms hold the same strings in two different places, so a filter over either one has
    /// to keep the same rows. This is the differential check that the arena being shared changed
    /// nothing about what a comparison means.
    #[test]
    fn a_filter_over_either_string_form_keeps_the_same_rows() {
        let flat = urls(96);
        let shared = flat.clone().shared_text().expect("shares");
        let constant =
            Vector::constant(LogicalType::Varchar, Value::Varchar("http://example3".into()), 96);
        let kept = Selection::from_predicate(96, |row| row % 5 != 0);
        for op in EVERY {
            let over_flat = refine(op, &flat, &constant, &kept).expect("refines");
            let over_shared = refine(op, &shared, &constant, &kept).expect("refines");
            assert_eq!(over_shared.indices(), over_flat.indices(), "{op:?}");
        }
    }

    /// The peeled path and the row at a time oracle on the same rows, on both spellings and on
    /// both entry points. A dictionary that shares its values is what a native scan hands over, so
    /// this is the shape every `WHERE URL <> ''` in ClickBench arrives in.
    #[test]
    fn a_comparison_peeled_over_a_shared_dictionary_answers_what_the_oracle_answers() {
        let words = ["", "one", "two", "", "three"];
        let values: Vec<Value> = words.iter().map(|text| Value::Varchar((*text).into())).collect();
        let values =
            Arc::new(Vector::from_values(LogicalType::Varchar, &values).expect("a vector of text"));
        let codes = vec![0, 1, 3, 2, 0, 4, 1, 0];
        let column = Vector::stable_dictionary(codes.clone(), values).expect("codes are in range");
        same_as_the_oracle(&column);
    }

    /// Values a storage reader would hand over, which answer one at a time and which know the
    /// order the writer sorted them into.
    #[derive(Debug)]
    struct Filed {
        values: Vec<Vec<u8>>,
        order: Vec<u32>,
        /// [`Self::order`] turned round, built on the first ask the way a reader's is.
        ranked: std::sync::OnceLock<Option<Vec<u32>>>,
    }

    impl rudb_vector::TextSource for Filed {
        fn len(&self) -> usize {
            self.values.len()
        }

        fn bytes_at(&self, index: usize) -> Result<Option<&[u8]>> {
            Ok(self.values.get(index).map(Vec::as_slice))
        }

        fn footprint(&self) -> usize {
            self.values.iter().map(Vec::len).sum()
        }

        fn ranks(&self) -> Option<usize> {
            Some(self.order.len())
        }

        fn compare_rank(&self, rank: usize, wanted: &[u8]) -> Result<Ordering> {
            // Plain bytes rather than the head the native format compares first, because what this
            // test is about is the answer the comparison gives and not how few reads it took.
            Ok(self.values[self.order[rank] as usize].as_slice().cmp(wanted))
        }

        fn code_at_rank(&self, rank: usize) -> Result<u32> {
            Ok(self.order[rank])
        }

        fn code_ranks(&self) -> Option<&[u32]> {
            self.ranked
                .get_or_init(|| {
                    let mut ranks = vec![0; self.order.len()];
                    for (rank, &code) in self.order.iter().enumerate() {
                        ranks[code as usize] = rank as u32;
                    }
                    Some(ranks)
                })
                .as_deref()
        }
    }

    /// The same comparison over a dictionary whose values came out of a file with their order, so
    /// the literal is resolved to a code by search rather than compared against every value.
    #[test]
    fn a_comparison_against_a_sorted_dictionary_answers_what_the_oracle_answers() {
        // Distinct, which is what a source promises by answering with an order at all, and which
        // a global dictionary is by construction.
        let (column, _) = filed(&["", "one", "two", "four", "three"], vec![0, 1, 3, 2, 0, 4, 1, 0]);
        same_as_the_oracle(&column);
    }

    /// A dictionary column over values that arrived from a file with their sorted order, handed
    /// back with the dictionary so a caller can check what it is holding a rank against.
    fn filed(words: &[&str], codes: Vec<u32>) -> (Vector, Arc<Vector>) {
        let values: Vec<Vec<u8>> = words.iter().map(|text| text.as_bytes().to_vec()).collect();
        let mut order = (0..values.len() as u32).collect::<Vec<_>>();
        order.sort_by(|&left, &right| values[left as usize].cmp(&values[right as usize]));
        let dictionary = Arc::new(
            Vector::external_text(
                LogicalType::Varchar,
                Arc::new(Filed { values, order, ranked: std::sync::OnceLock::new() }),
            )
            .expect("a filed vector"),
        );
        let column =
            Vector::stable_dictionary(codes, Arc::clone(&dictionary)).expect("codes are in range");
        (column, dictionary)
    }

    /// The top N's path. A bound whose rank is already known is compared without a search, and what
    /// it keeps is what comparing against the same value the long way round keeps.
    #[test]
    fn a_comparison_against_a_known_rank_keeps_what_a_search_for_it_keeps() {
        let (column, dictionary) =
            filed(&["", "one", "two", "four", "three"], vec![0, 1, 3, 2, 0, 4, 1, 0]);
        let rows = column.len();
        for row in 0..rows {
            let (held, rank) = rank_at(&column, row).expect("a row of a ranked dictionary");
            assert!(Arc::ptr_eq(&held, &dictionary), "the dictionary it came from");
            let value = column.try_value_at(row).expect("a value");
            for op in [
                Comparison::Less,
                Comparison::LessOrEqual,
                Comparison::Greater,
                Comparison::GreaterOrEqual,
            ] {
                let against = Vector::constant(LogicalType::Varchar, value.clone(), rows);
                let flags = compare(op, &column, &against).expect("the search path answers");
                let wanted = crate::select::selection(&flags, rows);
                let got = select_against_rank(op, &column, &dictionary, rank, rows)
                    .expect("the rank path answers");
                assert_eq!(got.indices(), wanted.indices(), "row {row} under {op:?}");
            }
        }
    }

    /// A rank is a position in one dictionary and means nothing in another, so a rank offered
    /// against the wrong one is declined rather than answered out of the wrong order.
    #[test]
    fn a_rank_offered_against_another_dictionary_is_declined() {
        let (column, dictionary) = filed(&["one", "two"], vec![0, 1]);
        let (other, _) = filed(&["one", "two"], vec![1, 0]);
        assert!(select_against_rank(Comparison::Less, &column, &dictionary, 1, 2).is_some());
        assert!(select_against_rank(Comparison::Less, &other, &dictionary, 1, 2).is_none());
        let flat = Vector::constant(LogicalType::Varchar, Value::Varchar("one".into()), 2);
        assert!(select_against_rank(Comparison::Less, &flat, &dictionary, 1, 2).is_none());
        assert!(rank_at(&flat, 0).is_none(), "a column with no dictionary has no ranks");
        assert!(rank_within(&other, 0, &dictionary).is_none(), "another dictionary says nothing");
        assert!(rank_within(&flat, 0, &dictionary).is_none(), "no dictionary says nothing");
    }

    /// The top N's reject. Two rows of the same column ordered by their ranks come out in the order
    /// their values come out in, which is the whole of what the rank comparison assumes.
    #[test]
    fn two_ranks_in_one_dictionary_order_their_values() {
        let (column, dictionary) =
            filed(&["", "one", "two", "four", "three"], vec![0, 1, 3, 2, 0, 4, 1, 0]);
        let rows = column.len();
        for left in 0..rows {
            for right in 0..rows {
                let here = rank_within(&column, left, &dictionary).expect("a ranked row");
                let there = rank_within(&column, right, &dictionary).expect("a ranked row");
                let values = (
                    column.try_value_at(left).expect("a value"),
                    column.try_value_at(right).expect("a value"),
                );
                let wanted = order(&values.0, &values.1).expect("two strings compare");
                assert_eq!(here.cmp(&there), wanted, "rows {left} and {right}");
                let (held, rank) = rank_at(&column, left).expect("a ranked row");
                assert!(Arc::ptr_eq(&held, &dictionary), "the dictionary it came from");
                assert_eq!(rank, here, "the same rank whichever way it is asked for");
            }
        }
    }

    /// Every equality and inequality against a handful of literals, whole and narrowed, checked
    /// against the row at a time path. `missing` is in here because a dictionary that does not
    /// hold the literal is decided for the whole chunk and that is its own arm of the code.
    fn same_as_the_oracle(column: &Vector) {
        for literal in ["", "one", "missing", "zzz"] {
            for op in [
                Comparison::Equal,
                Comparison::NotEqual,
                Comparison::Less,
                Comparison::LessOrEqual,
                Comparison::Greater,
                Comparison::GreaterOrEqual,
            ] {
                let value = Value::Varchar(literal.to_owned());
                let held = Held::of(&LogicalType::Varchar, &value).expect("text has a column");
                let right = Vector::constant(LogicalType::Varchar, value.clone(), column.len());
                let wanted = oracle(op, column, &right);
                let got = compare_prepared(op, column, &right, Some(&held))
                    .expect("the peeled path answers");
                assert_eq!(got, wanted, "{literal:?} under {op:?}");
                let held = Held::of(&LogicalType::Varchar, &value).expect("text has a column");
                let picked = select_prepared(op, column, &right, Some(&held))
                    .expect("the peeled path selects");
                assert_eq!(
                    picked.indices(),
                    crate::select::selection(&wanted, column.len()).indices(),
                    "{literal:?} under {op:?}, selected"
                );
                // A fresh memo for the selection, since the one above belongs to that call's node.
                let held = Held::of(&LogicalType::Varchar, &value).expect("text has a column");
                let kept = Selection::from_predicate(column.len(), |row| row % 3 != 1);
                let refined = refine_prepared(op, column, &right, &kept, Some(&held))
                    .expect("the peeled path narrows");
                let wanted: Vec<u32> = kept
                    .indices()
                    .iter()
                    .copied()
                    .filter(|&row| is_true(&wanted.value_at(row as usize)))
                    .collect();
                assert_eq!(refined.indices(), wanted, "{literal:?} under {op:?}, narrowed");
            }
        }
    }
}