rudb-kernels 0.3.41

//! The aggregate accumulators.
//!
//! One of these per group per aggregate, so a `GROUP BY` over a million distinct keys with three
//! aggregates in it holds three million of them. That is the reason the state is an enum of small
//! fixed cases rather than a boxed trait object: the hash table is going to hold these inline and a
//! pointer chase per row per aggregate is a cost that shows up on every grouped query there is.
//!
//! Null is skipped by every aggregate. `sum` over a column that is entirely null is null and not
//! zero, `count(x)` counts the rows where `x` is not null, and `count(*)` counts rows without
//! looking at anything. Those three are not variations on a theme, they are three different
//! questions, and the reason `count(*)` is a separate function rather than `count` with a star
//! argument is so the executor never has to work out which one it was handed.
//!
//! # How the vectorized path is put together
//!
//! The other four kernel files take a vector and give one back. This one has state, so the batch
//! interface is [`Accumulator::update_run`], which folds a whole vector into the running state in
//! one pass. What it can do in one pass depends on the aggregate, and the four shapes are worth
//! naming because they are not the same problem.
//!
//! `count` and `count(*)` do not read the data at all. A count of rows is the row count and a count
//! of values is the number of bits set in the validity mask, so those two are answered from the
//! mask whatever the form and whatever the type, which is why they are the only ones that come back
//! before the form dispatch.
//!
//! A whole sum reads the data and ignores the order. Rows that are null are masked out with a
//! conditional move rather than a branch, and the accumulator is an `i128` so that nothing narrower
//! than a `HUGEINT` can overflow inside one vector and the check only has to happen once, where the
//! vector's total meets the running total.
//!
//! A floating point sum reads the data and does not ignore the order, because floating point
//! addition is not associative and the answer this has to reach is the one the row at a time loop
//! reaches. So that loop stays sequential and gives up the vectorization the whole sum gets. It is
//! still about fifty times faster than building a `Value` per row, and an answer that is fast and
//! different from the reference is not an answer.
//!
//! `min` and `max` read the data to find which row won and then ask the vector for that one row.
//! One `Value` per vector instead of one per row, and one call into the comparison kernel instead
//! of one per row.
//!
//! # The grouped form
//!
//! [`update_run`](Accumulator::update_run) folds a vector into one accumulator, which is what an
//! ungrouped aggregate wants and is no use at all to a `GROUP BY`, where the rows of one vector
//! belong to as many different accumulators as there are groups in it. [`update_scattered`] is the
//! grouped form: one vector, one slot per row saying which accumulator that row belongs to, and one
//! pass that reads the run once and folds each value into the accumulator its row points at.
//!
//! It cannot reduce the way the ungrouped form does, because two adjacent rows are usually two
//! different groups and there is nothing to add up before the scatter. What it removes is everything
//! else: the `Value` built per row, which for a string column is a malloc, the match on which
//! aggregate this is, and the match on which layout the column is in. All three of those are decided
//! once per vector here and none of them per row.

use std::sync::Arc;

use rudb_common::{Error, LogicalType, Result, Value};
use rudb_vector::{Data, Form, Validity, Vector};

use crate::compare::order;
use crate::fallback::{self, Kernel};
use crate::number::{fit, integral, pow10, rescale};
use crate::shape::{identity, nulls_of};

/// Where a row that belongs to no accumulator points.
///
/// A row a `FILTER` threw away and a row that went to a spill file both have nothing to update, and
/// the caller says so by pointing them here rather than by handing over a second mask. One sentinel
/// rather than a second buffer, because the slots are written per row anyway and the test is a
/// compare against a constant.
pub const NOWHERE: usize = usize::MAX;

/// A running aggregate.
#[derive(Debug, Clone)]
pub struct Accumulator {
    state: State,
}

/// The part of an aggregate return type its final value needs.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Return {
    TinyInt,
    SmallInt,
    Integer,
    BigInt,
    HugeInt,
    UTinyInt,
    USmallInt,
    UInteger,
    UBigInt,
    UHugeInt,
    Float,
    Double,
    Decimal(u8),
    /// Min and max return their held value and do not inspect the declared type.
    Other,
}

impl Return {
    fn new(ty: &LogicalType) -> Self {
        match ty {
            LogicalType::TinyInt => Self::TinyInt,
            LogicalType::SmallInt => Self::SmallInt,
            LogicalType::Integer => Self::Integer,
            LogicalType::BigInt => Self::BigInt,
            LogicalType::HugeInt => Self::HugeInt,
            LogicalType::UTinyInt => Self::UTinyInt,
            LogicalType::USmallInt => Self::USmallInt,
            LogicalType::UInteger => Self::UInteger,
            LogicalType::UBigInt => Self::UBigInt,
            LogicalType::UHugeInt => Self::UHugeInt,
            LogicalType::Float => Self::Float,
            LogicalType::Double => Self::Double,
            LogicalType::Decimal { width, .. } => Self::Decimal(*width),
            _ => Self::Other,
        }
    }

    fn logical(self) -> LogicalType {
        match self {
            Self::TinyInt => LogicalType::TinyInt,
            Self::SmallInt => LogicalType::SmallInt,
            Self::Integer => LogicalType::Integer,
            Self::BigInt => LogicalType::BigInt,
            Self::HugeInt => LogicalType::HugeInt,
            Self::UTinyInt => LogicalType::UTinyInt,
            Self::USmallInt => LogicalType::USmallInt,
            Self::UInteger => LogicalType::UInteger,
            Self::UBigInt => LogicalType::UBigInt,
            Self::UHugeInt => LogicalType::UHugeInt,
            Self::Float => LogicalType::Float,
            Self::Double => LogicalType::Double,
            Self::Decimal(width) => LogicalType::Decimal { width, scale: 0 },
            Self::Other => LogicalType::Null,
        }
    }
}

/// Which aggregate.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Kind {
    CountStar,
    Count,
    Sum,
    Avg,
    Min,
    Max,
}

/// What the aggregate has seen so far.
#[derive(Debug, Clone)]
enum State {
    /// A row count, for `count` and `count(*)`.
    Counted { count: i64, star: bool },
    /// A whole running total and whether anything landed in it.
    Whole { total: i128, seen: bool, returns: Return },
    /// A running total in floating point, and the count `avg` divides by.
    Real { total: f64, seen: i64, kind: Kind, returns: Return },
    /// A running total for `avg`, exact while every value folded in is a whole number.
    ///
    /// `avg` over an integer column has to add the column up exactly and divide once at the end.
    /// Adding into a double as it goes gives a different number: each addition past `2^53` rounds,
    /// and the roundings do not cancel. `AVG(UserID)` over ten thousand rows of the benchmark file
    /// came out `435091026172918.3` that way where duckdb says `435091026172920.25`, which is the
    /// sum divided once. So the total is an `i128` and the division is the only rounding.
    ///
    /// `exact` goes false the first time a value is not a whole number, or the first time the total
    /// would overflow, and from then on `real` carries it. A column of doubles therefore lands on
    /// the same additions in the same order as before, which is what the float path has to keep.
    // The exact and floating totals are mutually exclusive. The floating total's bits occupy the
    // same word as the exact i128 after `exact` becomes false, cutting every AVG state by 16 bytes.
    Mean { total: i128, seen: i64, exact: bool, returns: Return },
    /// A running total at a fixed decimal scale.
    Scaled { total: i128, scale: u8, seen: bool, returns: Return },
    /// The smallest or largest value so far.
    // Box the one state whose scalar representation is much wider than every numeric aggregate.
    // Most ClickBench groups contain count/sum/avg states and should not each pay for a 64-byte
    // Value they never hold. Min and max allocate only after they see their first non-null value.
    Extreme { held: Option<Box<Extremum>>, least: bool },
}

/// What a min or a max is holding.
#[derive(Debug, Clone)]
enum Extremum {
    /// A value copied out of the column it came from.
    Held(Value),
    /// A position in a dictionary that knows its sorted order, kept in place of the value there.
    ///
    /// A grouped min over a column of strings does a comparison per row, and comparing two strings
    /// means fetching both out of a payload the size of the column. A dictionary that sorted its
    /// values when it was written already knows which of any two of them is smaller, so the
    /// comparison becomes two integers and a string is fetched once for the group that answers with
    /// it rather than once for every row that fails to.
    ///
    /// The dictionary is held rather than borrowed, and that is what makes the `Arc::ptr_eq` in
    /// [`Extremum::offer`] a sound way to ask whether two of these rank against each other. Two
    /// dictionaries number their values differently, so a rank from one means nothing to the other,
    /// and keeping this one alive is what stops a second from turning up at the same address and
    /// being taken for it.
    Ranked { dictionary: Arc<Vector>, code: u32, rank: u32 },
}

impl Extremum {
    /// The value this holds, read out of the dictionary when that is where it still is.
    fn value(&self) -> Result<Value> {
        match self {
            Self::Held(value) => Ok(value.clone()),
            Self::Ranked { dictionary, code, .. } => dictionary.try_value_at(*code as usize),
        }
    }

    /// Replaces a rank with the value it stands for, so that something without ranks can compare.
    fn settle(&mut self) -> Result<&mut Value> {
        if let Self::Ranked { .. } = self {
            let value = self.value()?;
            *self = Self::Held(value);
        }
        match self {
            Self::Held(value) => Ok(value),
            Self::Ranked { .. } => {
                Err(Error::internal("a settled extreme kept its rank".to_string()))
            }
        }
    }

    /// Offers one dictionary row, and keeps it when it beats what is already here.
    ///
    /// The whole point of the fast path is the first arm: same dictionary, so one integer compare
    /// and, on a win, two stores. Anything else settles into a value and compares the long way,
    /// which is what a mix of dictionaries or a mix of forms in one aggregate costs.
    fn offer(&mut self, dictionary: &Arc<Vector>, code: u32, rank: u32, least: bool) -> Result<()> {
        if let Self::Ranked { dictionary: mine, code: held_code, rank: held_rank } = self {
            if Arc::ptr_eq(mine, dictionary) {
                if if least { rank < *held_rank } else { rank > *held_rank } {
                    *held_code = code;
                    *held_rank = rank;
                }
                return Ok(());
            }
        }
        let candidate = dictionary.try_value_at(code as usize)?;
        let ordering = order(&candidate, self.settle()?)?;
        if if least { ordering.is_lt() } else { ordering.is_gt() } {
            *self = Self::Held(candidate);
        }
        Ok(())
    }
}

impl Accumulator {
    /// The value of a COUNT state, and `None` for every other aggregate.
    #[must_use]
    pub fn counted(&self) -> Option<i64> {
        match self.state {
            State::Counted { count, .. } => Some(count),
            _ => None,
        }
    }

    /// Finish an exact integer SUM held in a compact grouped state.
    #[must_use]
    pub fn exact_sum(total: i128, seen: bool, returns: &LogicalType) -> Self {
        Self { state: State::Whole { total, seen, returns: Return::new(returns) } }
    }

    /// Finish an exact integer AVG held in a compact grouped state.
    #[must_use]
    pub fn exact_avg(total: i128, seen: i64, returns: &LogicalType) -> Self {
        Self { state: State::Mean { total, seen, exact: true, returns: Return::new(returns) } }
    }

    fn kind(&self) -> Kind {
        match self.state {
            State::Counted { star, .. } => {
                if star {
                    Kind::CountStar
                } else {
                    Kind::Count
                }
            }
            State::Whole { .. } | State::Scaled { .. } => Kind::Sum,
            State::Real { kind, .. } => kind,
            State::Mean { .. } => Kind::Avg,
            State::Extreme { least, .. } => {
                if least {
                    Kind::Min
                } else {
                    Kind::Max
                }
            }
        }
    }

    fn returns(&self) -> Return {
        match self.state {
            State::Counted { .. } => Return::BigInt,
            State::Whole { returns, .. }
            | State::Real { returns, .. }
            | State::Mean { returns, .. }
            | State::Scaled { returns, .. } => returns,
            State::Extreme { .. } => Return::Other,
        }
    }

    /// A fresh accumulator for a named aggregate returning `returns`.
    ///
    /// # Errors
    ///
    /// If the name is not an aggregate this crate implements.
    pub fn new(name: &str, returns: &LogicalType) -> Result<Self> {
        let kind = match name {
            "count_star" => Kind::CountStar,
            "count" => Kind::Count,
            "sum" => Kind::Sum,
            "avg" => Kind::Avg,
            "min" => Kind::Min,
            "max" => Kind::Max,
            other => {
                return Err(Error::not_implemented(format!("the {other} aggregate")));
            }
        };
        let scale = match returns {
            LogicalType::Decimal { scale, .. } => *scale,
            _ => 0,
        };
        let returns = Return::new(returns);
        let state = match kind {
            Kind::CountStar | Kind::Count => {
                State::Counted { count: 0, star: kind == Kind::CountStar }
            }
            Kind::Avg => State::Mean { total: 0, seen: 0, exact: true, returns },
            Kind::Min | Kind::Max => State::Extreme { held: None, least: kind == Kind::Min },
            Kind::Sum => match returns {
                Return::Decimal(_) => State::Scaled { total: 0, scale, seen: false, returns },
                Return::Float | Return::Double => {
                    State::Real { total: 0.0, seen: 0, kind, returns }
                }
                _ => State::Whole { total: 0, seen: false, returns },
            },
        };
        Ok(Self { state })
    }

    /// Folds one row in.
    ///
    /// # Errors
    ///
    /// If the argument count is wrong for the aggregate, if the value is not one the aggregate can
    /// accumulate, or if a whole running total overflows.
    pub fn update(&mut self, args: &[Value]) -> Result<()> {
        if self.kind() == Kind::CountStar {
            if let State::Counted { count, .. } = &mut self.state {
                *count += 1;
            }
            return Ok(());
        }
        let value = match args {
            [only] => only,
            _ => {
                return Err(Error::internal(format!("an aggregate over {} arguments", args.len())));
            }
        };
        if value.is_null() {
            return Ok(());
        }
        match &mut self.state {
            State::Counted { count, .. } => *count += 1,
            State::Whole { total, seen, .. } => {
                let whole = integral(value).ok_or_else(|| not_narrow(value))?;
                *total = total.checked_add(whole).ok_or_else(overflowed)?;
                *seen = true;
            }
            State::Real { total, seen, .. } => {
                *total += approximate_or_error(value)?;
                *seen += 1;
            }
            State::Mean { total, seen, exact, .. } => {
                match integral(value)
                    .filter(|_| *exact)
                    .and_then(|number| total.checked_add(number))
                {
                    Some(sum) => *total = sum,
                    None => {
                        // The first value that is not whole, or the first one that would overflow.
                        // What was counted exactly so far comes across as one conversion, and the
                        // rest of the column is added the way it always was.
                        let real = if *exact { exactly(*total) } else { mean_real(*total) };
                        *total = mean_bits(real + approximate_or_error(value)?);
                        *exact = false;
                    }
                }
                *seen += 1;
            }
            State::Scaled { total, scale, seen, .. } => {
                let unscaled = at_scale(value, *scale).ok_or_else(|| not_narrow(value))?;
                *total = total.checked_add(unscaled).ok_or_else(overflowed)?;
                *seen = true;
            }
            State::Extreme { held, least } => {
                let replace = match held {
                    None => true,
                    Some(current) => {
                        let ordering = order(value, current.settle()?)?;
                        if *least { ordering.is_lt() } else { ordering.is_gt() }
                    }
                };
                if replace {
                    *held = Some(Box::new(Extremum::Held(value.clone())));
                }
            }
        }
        Ok(())
    }

    /// Folds a whole vector in, in one pass over the data and without building a [`Value`] per row.
    ///
    /// `rows` is how many rows to fold, which is the row count of the chunk rather than the capacity
    /// of the vectors in it. `count(*)` takes no argument and reads nothing but that number, and
    /// every other aggregate here takes exactly one vector.
    ///
    /// A shape the one pass form does not cover falls through to [`Accumulator::update`] per row and
    /// records itself in [`crate::fallback`], so this always reaches the answer the row at a time
    /// loop reaches and never a different one. That is not a slogan about floating point: the
    /// running total below is carried into the vector loop rather than restarted at zero, precisely
    /// so that the additions happen in the same order and round the same way.
    ///
    /// # Errors
    ///
    /// The same errors [`Accumulator::update`] raises, for the same reasons.
    pub fn update_run(&mut self, args: &[Vector], rows: usize) -> Result<()> {
        if self.kind() == Kind::CountStar {
            if let State::Counted { count, .. } = &mut self.state {
                *count += i64::try_from(rows).map_err(|_| overlong())?;
            }
            return Ok(());
        }
        let input = match args {
            [only] => only,
            _ => {
                return Err(Error::internal(format!("an aggregate over {} arguments", args.len())));
            }
        };
        if input.len() < rows {
            return Err(Error::internal(format!(
                "an aggregate handed {rows} rows and a vector of {}",
                input.len()
            )));
        }
        if self.folded(input, rows)? {
            return Ok(());
        }
        // An aggregate reads one vector, so its form goes in both halves of the report rather than
        // leaving a column of zeros next to every row of it.
        fallback::record(Kernel::Aggregate, input.form(), input.form());
        // row at a time: the path recorded on the line above, which exists to be correct for an
        // aggregate `folded` does not cover and counts itself so that aggregate shows up.
        for row in 0..rows {
            let value = input.try_value_at(row)?;
            self.update(std::slice::from_ref(&value))?;
        }
        Ok(())
    }

    /// Folds a vector in in one pass, or says this is a shape the one pass form does not cover.
    ///
    /// # Errors
    ///
    /// If a whole running total overflows where the row at a time loop would also have overflowed.
    fn folded(&mut self, input: &Vector, rows: usize) -> Result<bool> {
        let nulls = nulls_of(input);
        // Both counts are answered by the mask on its own, whatever the form is and whatever the
        // type is, so they come back before there is any question of which loop to run.
        if let State::Counted { count, .. } = &mut self.state {
            *count += i64::try_from(nulls.count_valid(rows)).map_err(|_| overlong())?;
            return Ok(true);
        }
        let least = self.kind() == Kind::Min;
        let want = match (&self.state, input.logical_type()) {
            (State::Whole { .. }, _) => Want::Whole,
            (State::Real { total, .. }, ty) => {
                Want::Real { scale: decimal_scale(ty), from: *total }
            }
            // An exact mean over an integer column is read the way a sum is and divided at the end.
            // Anything else is the float path, carried on from wherever the total is now, which for
            // a mean that was exact until this vector is the exact total converted once.
            (State::Mean { exact: true, .. }, ty) if ty.is_integer() => Want::Whole,
            (State::Mean { total, exact, .. }, ty) => {
                let from = if *exact { exactly(*total) } else { mean_real(*total) };
                Want::Real { scale: decimal_scale(ty), from }
            }
            // A total at the scale the column is already held at is a sum of the raw unscaled
            // integers and nothing else, which is the case every real query is in, because the sum
            // of a `DECIMAL(15, 2)` column is declared at scale two. An integer summed into a
            // decimal total, or a decimal at some other scale, needs a rescale per row that the row
            // at a time path already does correctly, so those go that way and the counter says
            // whether that was the wrong call.
            (State::Scaled { scale, .. }, LogicalType::Decimal { scale: held, .. })
                if held == scale =>
            {
                Want::Whole
            }
            (State::Scaled { .. }, _) => return Ok(false),
            (State::Extreme { .. }, _) => Want::Extreme(least),
            (State::Counted { .. }, _) => return Ok(false),
        };
        let Some(contribution) = gather(input, rows, &nulls, want) else {
            return Ok(false);
        };
        let live = nulls.count_valid(rows);
        match (&mut self.state, contribution) {
            (
                State::Whole { total, seen, .. } | State::Scaled { total, seen, .. },
                Contribution::Whole(sum),
            ) => {
                *total = total.checked_add(sum).ok_or_else(overflowed)?;
                *seen |= live > 0;
            }
            (
                State::Real { total, seen, .. },
                Contribution::Real { total: carried, seen: added },
            ) => {
                *total = carried;
                *seen += added;
            }
            (State::Mean { total, seen, .. }, Contribution::Whole(sum)) => {
                // An overflow here is not an error the way it is for a sum, because the row at a
                // time loop answers an overflowing mean in floating point rather than raising. So
                // this hands the vector back and that loop folds it in, state untouched.
                let Some(sum) = total.checked_add(sum) else { return Ok(false) };
                *total = sum;
                *seen += i64::try_from(live).map_err(|_| overlong())?;
            }
            (
                State::Mean { total, seen, exact, .. },
                Contribution::Real { total: carried, seen: added },
            ) => {
                *total = mean_bits(carried);
                *exact = false;
                *seen += added;
            }
            (State::Extreme { held, .. }, Contribution::Extreme(Some(index))) => {
                // One `Value` for the whole vector and one call into the comparison kernel, rather
                // than one of each per row. The row that won is found on the numbers.
                let candidate = input.try_value_at(index)?;
                let replace = match held {
                    None => true,
                    Some(current) => {
                        let ordering = order(&candidate, current.settle()?)?;
                        if least { ordering.is_lt() } else { ordering.is_gt() }
                    }
                };
                if replace {
                    *held = Some(Box::new(Extremum::Held(candidate)));
                }
            }
            (State::Extreme { .. }, Contribution::Extreme(None)) => {}
            // The `want` above picks the contribution, so the pairs left over are ones that cannot
            // be built. Falling through costs a slow loop and a wrong answer costs a lot more.
            _ => return Ok(false),
        }
        Ok(true)
    }

    /// Folds another accumulator over the same aggregate into this one.
    ///
    /// This is what lets one aggregate run on more than one thread. Each thread builds its own
    /// table, and where two of them found the same group there are two states for it, so the merge
    /// needs a way to add one state to another that does not go back to the rows either of them was
    /// built from. That is this, and it is the whole of what an aggregate has to provide to be
    /// parallel, because the key side of the merge is the same probe the fold already does.
    ///
    /// # A sum of doubles is not associative
    ///
    /// Two threads that split a column of doubles add it up in a different order than one thread
    /// does, and floating point addition does not care that the numbers are the same. So `sum` and
    /// `avg` over a `FLOAT` or a `DOUBLE` can answer differently depending on how the work was
    /// divided. That is not a bug being introduced here, it is the arithmetic, and it is the reason
    /// `avg` over integers keeps an exact `i128` total and divides once, which makes the common case
    /// of the two give the same answer whatever the division was.
    ///
    /// # Errors
    ///
    /// [`rudb_common::ErrorCode::Internal`] if the two states are not the same aggregate over the
    /// same type, which is a caller merging two tables that did not come from the same operator.
    /// [`rudb_common::ErrorCode::OutOfRange`] if a whole running total overflows, which is the same
    /// answer the row at a time path gives to the same sum.
    pub fn combine(&mut self, other: &Self) -> Result<()> {
        match (&mut self.state, &other.state) {
            (State::Counted { count, star }, State::Counted { count: added, star: same })
                if star == same =>
            {
                *count += added;
            }
            (State::Whole { total, seen, .. }, State::Whole { total: added, seen: any, .. }) => {
                *total = total.checked_add(*added).ok_or_else(overflowed)?;
                *seen |= any;
            }
            (
                State::Scaled { total, scale, seen, .. },
                State::Scaled { total: added, scale: same, seen: any, .. },
            ) if scale == same => {
                *total = total.checked_add(*added).ok_or_else(overflowed)?;
                *seen |= any;
            }
            (State::Real { total, seen, .. }, State::Real { total: added, seen: more, .. }) => {
                *total += added;
                *seen += more;
            }
            (
                State::Mean { total, seen, exact, .. },
                State::Mean { total: added, seen: more, exact: whole, .. },
            ) => {
                // Both sides exact and the sum still fitting is the case worth keeping exact,
                // because it is `AVG` over an integer column and it is what gives the same answer
                // however the rows were divided. Anything else falls to the floating total, and it
                // falls once rather than per row, so the side that was exact is converted here and
                // the two are added as doubles.
                let both = if *exact && *whole { total.checked_add(*added) } else { None };
                match both {
                    Some(sum) => *total = sum,
                    None => {
                        let here = if *exact { exactly(*total) } else { mean_real(*total) };
                        let there = if *whole { exactly(*added) } else { mean_real(*added) };
                        *total = mean_bits(here + there);
                        *exact = false;
                    }
                }
                *seen += more;
            }
            (State::Extreme { held, least }, State::Extreme { held: candidate, least: same })
                if least == same =>
            {
                if let Some(candidate) = candidate {
                    match (held.as_deref_mut(), candidate.as_ref()) {
                        (None, _) => *held = Some(candidate.clone()),
                        // Two workers over one column hold ranks out of the one dictionary, so the
                        // merge that brings their tables together compares integers too.
                        (
                            Some(Extremum::Ranked { dictionary, rank, .. }),
                            Extremum::Ranked { dictionary: theirs, rank: other, .. },
                        ) if Arc::ptr_eq(dictionary, theirs) => {
                            if if *least { other < rank } else { other > rank } {
                                *held = Some(candidate.clone());
                            }
                        }
                        (Some(current), _) => {
                            let value = candidate.value()?;
                            let ordering = order(&value, current.settle()?)?;
                            if if *least { ordering.is_lt() } else { ordering.is_gt() } {
                                *held = Some(candidate.clone());
                            }
                        }
                    }
                }
            }
            (here, there) => {
                return Err(Error::internal(format!(
                    "combining a {here:?} aggregate state with a {there:?} one, which are not the \
                     same aggregate over the same type"
                )));
            }
        }
        Ok(())
    }

    /// The aggregate's answer.
    ///
    /// # Errors
    ///
    /// If the running total does not fit the declared return type.
    pub fn finish(&self) -> Result<Value> {
        match &self.state {
            State::Counted { count, .. } => Ok(Value::BigInt(*count)),
            State::Whole { total, seen, .. } => {
                if !seen {
                    return Ok(Value::Null);
                }
                let returns = self.returns().logical();
                fit(*total, &returns).ok_or_else(|| {
                    Error::out_of_range(format!("a sum of {total} does not fit in {}", returns))
                })
            }
            State::Real { total, seen, .. } => {
                if *seen == 0 {
                    return Ok(Value::Null);
                }
                #[expect(
                    clippy::cast_precision_loss,
                    reason = "the count of rows in one group is well inside the exact range"
                )]
                let answer = if self.kind() == Kind::Avg { total / *seen as f64 } else { *total };
                if self.returns() == Return::Float {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "a declared FLOAT result is a FLOAT"
                    )]
                    return Ok(Value::Float(answer as f32));
                }
                Ok(Value::Double(answer))
            }
            State::Mean { total, seen, exact, .. } => {
                if *seen == 0 {
                    return Ok(Value::Null);
                }
                let total = if *exact { exactly(*total) } else { mean_real(*total) };
                #[expect(
                    clippy::cast_precision_loss,
                    reason = "the count of rows in one group is well inside the exact range"
                )]
                let answer = total / *seen as f64;
                if self.returns() == Return::Float {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "a declared FLOAT result is a FLOAT"
                    )]
                    return Ok(Value::Float(answer as f32));
                }
                Ok(Value::Double(answer))
            }
            State::Scaled { total, scale, seen, .. } => {
                if !seen {
                    return Ok(Value::Null);
                }
                let width = match self.returns() {
                    Return::Decimal(width) => width,
                    _ => rudb_common::MAX_DECIMAL_WIDTH,
                };
                Ok(Value::Decimal { unscaled: *total, width, scale: *scale })
            }
            State::Extreme { held, .. } => held.as_deref().map_or(Ok(Value::Null), Extremum::value),
        }
    }

    /// Finishes an integer sum after adding one constant for every nonnull input row.
    pub fn finish_offset(&self, offset: i64, rows: i64) -> Result<Value> {
        let Value::HugeInt(total) = self.finish()? else {
            return Ok(Value::Null);
        };
        let added = i128::from(offset).checked_mul(i128::from(rows)).ok_or_else(overflowed)?;
        Ok(Value::HugeInt(total.checked_add(added).ok_or_else(overflowed)?))
    }
}

/// Folds one vector into many accumulators, each row into the one its slot points at.
///
/// `states` is the caller's flat array of accumulators, `stride` is how many aggregates there are
/// per group and `offset` is which of them this call is, so the accumulator a row belongs to is at
/// `slots[row] * stride + offset`. That is the layout a hash aggregate already keeps, one run of
/// accumulators per group, so nothing is copied to call this and the slots are the probe results the
/// operator has in hand. A slot of [`NOWHERE`] is a row that contributes to nothing.
///
/// A shape the one pass form does not cover falls through to [`Accumulator::update`] per row and
/// records itself in [`crate::fallback`], exactly as [`Accumulator::update_run`] does, so this
/// always reaches the answer the row at a time loop reaches. The rows are walked in order in every
/// path, which is what keeps a floating point total inside one group adding in the same order and
/// rounding the same way.
///
/// # Errors
///
/// The same errors [`Accumulator::update`] raises, and an internal error if the slots are shorter
/// than the rows or if the vector is.
pub fn update_scattered(
    states: &mut [Accumulator],
    slots: &[usize],
    stride: usize,
    offset: usize,
    input: Option<&Vector>,
    rows: usize,
) -> Result<()> {
    if states.is_empty() {
        return Ok(());
    }
    if slots.len() < rows {
        return Err(Error::internal(format!(
            "an aggregate handed {rows} rows and {} slots",
            slots.len()
        )));
    }
    let Some(first) = states.get(offset) else {
        return Err(Error::internal(format!(
            "an aggregate at {offset} of {} accumulators",
            states.len()
        )));
    };
    let kind = first.kind();
    // Which way a min or a max runs, taken before the loops below borrow the states they update.
    let extreme = match first.state {
        State::Extreme { least, .. } => Some(least),
        _ => None,
    };
    let into = Where { slots, stride, offset };
    // `count(*)` reads nothing, so it never asks for the argument it does not have.
    if kind == Kind::CountStar {
        for row in 0..rows {
            let Some(index) = into.index(row) else { continue };
            if let State::Counted { count, .. } = &mut states[index].state {
                *count += 1;
            }
        }
        return Ok(());
    }
    let Some(input) = input else {
        return Err(Error::internal("an aggregate over 0 arguments".to_string()));
    };
    if input.len() < rows {
        return Err(Error::internal(format!(
            "an aggregate handed {rows} rows and a vector of {}",
            input.len()
        )));
    }
    let nulls = nulls_of(input);
    // Every aggregate here skips nulls, so a vector that is entirely null contributes nothing to
    // anything whatever the type is and whatever the form is.
    if matches!(nulls, Validity::AllInvalid) {
        return Ok(());
    }
    if let Some(least) = extreme {
        if ranked_extremes(states, into, input, rows, &nulls, least)? {
            return Ok(());
        }
    }
    if extreme.is_some()
        && input.logical_type() == &LogicalType::Varchar
        && matches!(input.form(), Form::Flat | Form::Dictionary | Form::StringView | Form::Rle)
    {
        // Bytes rather than text, because the comparison below is a comparison of bytes and asking
        // for a `&str` would check that every row of the column is UTF-8 on the way past. That check
        // is the whole column read again: on `SELECT MIN(Referer) ... GROUP BY` over the million row
        // ClickBench sample it was 210 million instructions, a tenth of the query, to produce a
        // `&str` that was turned straight back into the bytes it came from. It is paid where it
        // means something instead, which is the row that becomes a group's answer.
        //
        // row at a time: each input belongs to one group; only a new extreme owns its text.
        for row in 0..rows {
            if !nulls.is_valid(row) {
                continue;
            }
            let Some(index) = into.index(row) else { continue };
            let bytes = input.try_bytes_at(row)?.ok_or_else(|| {
                Error::internal("a valid varchar row had no borrowed text".to_string())
            })?;
            let State::Extreme { held, least } = &mut states[index].state else {
                return Err(Error::internal("a string extreme into another state".to_string()));
            };
            match held {
                Some(current) => {
                    let Value::Varchar(previous) = current.settle()? else {
                        return Err(Error::internal(
                            "a varchar extreme held another type".to_string(),
                        ));
                    };
                    let better = if *least {
                        bytes < previous.as_bytes()
                    } else {
                        bytes > previous.as_bytes()
                    };
                    if better {
                        // Written over the string the group already holds rather than into a new
                        // one, because a group is made once and its answer is replaced many times.
                        previous.clear();
                        previous.push_str(utf8(bytes)?);
                    }
                }
                None => {
                    let text = Value::Varchar(utf8(bytes)?.to_owned());
                    *held = Some(Box::new(Extremum::Held(text)));
                }
            }
        }
        return Ok(());
    }
    let Some(first) = states.get(offset) else {
        return Err(Error::internal(format!(
            "an aggregate at {offset} of {} accumulators",
            states.len()
        )));
    };
    let feed = feed_of(first, input.logical_type());
    if let Some(feed) = feed {
        if spread(states, into, input, rows, &nulls, feed)? {
            return Ok(());
        }
    }
    // An aggregate reads one vector, so its form goes in both halves of the report.
    fallback::record(Kernel::Aggregate, input.form(), input.form());
    // row at a time: the path recorded on the line above, which exists to be correct for a column
    // `spread` does not cover and counts itself so that column shows up.
    for row in 0..rows {
        let Some(index) = into.index(row) else { continue };
        let value = input.try_value_at(row)?;
        states[index].update(std::slice::from_ref(&value))?;
    }
    Ok(())
}

/// A grouped min or max over a dictionary that sorted its values when it was written.
///
/// The win is that no string is read. Each row turns into the rank of its code, which is one load
/// out of a map four bytes wide per distinct value, and a group keeps the rank it has rather than
/// the text, so a column whose payload is sixty six megabytes over four hundred thousand values is
/// never touched until the groups are finished. Against the byte comparison below it, on ClickBench
/// query 28 over the million row file, this is the difference between a fetch out of a dictionary
/// block per row for eight hundred thousand rows and one fetch per group for ninety five thousand
/// groups.
///
/// `false` when the input is not a dictionary, or is one that does not know its order, and the
/// caller then takes whichever of the slower paths fits. Nothing here decides an answer differently
/// from those, only more cheaply: a rank order is the byte order of the values by the promise
/// [`rudb_vector::TextSource::ranks`] makes.
fn ranked_extremes(
    states: &mut [Accumulator],
    into: Where<'_>,
    input: &Vector,
    rows: usize,
    nulls: &Validity,
    least: bool,
) -> Result<bool> {
    let Some((codes, dictionary)) = input.shared_dictionary_parts() else { return Ok(false) };
    let Some(ranks) = dictionary.code_ranks() else { return Ok(false) };
    if codes.len() < rows {
        return Ok(false);
    }
    // row at a time: a scatter is per row by definition, since two adjacent rows are usually two
    // different groups and there is nothing to reduce before it.
    for (row, &code) in codes.iter().enumerate().take(rows) {
        if !nulls.is_valid(row) {
            continue;
        }
        let Some(index) = into.index(row) else { continue };
        // Giving up here leaves the rows already offered in the groups that took them, and that is
        // harmless: offering a row to a min twice reaches the same min as offering it once, so the
        // path the caller falls back to reads the whole vector again and lands in the same place.
        let Some(&rank) = ranks.get(code as usize) else { return Ok(false) };
        let State::Extreme { held, .. } = &mut states[index].state else {
            return Err(Error::internal("a ranked extreme into another state".to_string()));
        };
        match held {
            Some(current) => current.offer(dictionary, code, rank, least)?,
            None => {
                let kept = Extremum::Ranked { dictionary: dictionary.clone(), code, rank };
                *held = Some(Box::new(kept));
            }
        }
    }
    Ok(true)
}

/// The text a row holds, checked once for the row that is going to be kept.
///
/// The same message [`Vector::try_text_at`] gives, because it is the same failure and a caller
/// cannot tell which of the two read the column.
fn utf8(bytes: &[u8]) -> Result<&str> {
    std::str::from_utf8(bytes)
        .map_err(|error| Error::conversion(format!("invalid UTF-8 in VARCHAR: {error}")))
}

/// Which accumulator a row belongs to.
#[derive(Clone, Copy)]
struct Where<'w> {
    slots: &'w [usize],
    stride: usize,
    offset: usize,
}

impl Where<'_> {
    /// The accumulator this row folds into, or none if it folds into nothing.
    fn index(self, row: usize) -> Option<usize> {
        let slot = self.slots[row];
        (slot != NOWHERE).then(|| slot * self.stride + self.offset)
    }
}

/// What one run of values is read as on the way into many accumulators.
#[derive(Clone, Copy)]
enum Feed {
    /// A count of the rows that are not null, which reads the mask and not the data.
    Counted,
    /// An exact number per row.
    Whole,
    /// A number per row in floating point, at the scale a decimal column is held at.
    Real { scale: u8 },
    /// A number per row against the best that group has seen, the smallest one if true.
    Extreme(bool),
}

/// How a column is read for a call, or none if the one pass form does not cover it.
///
/// This is [`Accumulator::folded`]'s `want` with the running total left out, because there is no one
/// running total here. It has to make the same choices for the same reasons, so the arms are in the
/// same order and the comments there are the comments here.
fn feed_of(first: &Accumulator, ty: &LogicalType) -> Option<Feed> {
    match (&first.state, ty) {
        (State::Counted { .. }, _) => Some(Feed::Counted),
        (State::Whole { .. }, _) => Some(Feed::Whole),
        (State::Scaled { scale, .. }, LogicalType::Decimal { scale: held, .. })
            if held == scale =>
        {
            Some(Feed::Whole)
        }
        (State::Scaled { .. }, _) => None,
        (State::Mean { .. }, ty) if ty.is_integer() => Some(Feed::Whole),
        (State::Mean { .. } | State::Real { .. }, ty) => {
            Some(Feed::Real { scale: decimal_scale(ty) })
        }
        // A number is compared as a number and anything else is compared the way the comparison
        // kernel says, which a run of `i128` cannot do for a float, a string or a date.
        (State::Extreme { .. }, ty) if ty.is_integer() => {
            Some(Feed::Extreme(first.kind() == Kind::Min))
        }
        (State::Extreme { .. }, _) => None,
    }
}

/// One pass over a vector, folding each row into the accumulator it belongs to.
fn spread(
    states: &mut [Accumulator],
    into: Where<'_>,
    input: &Vector,
    rows: usize,
    nulls: &Validity,
    feed: Feed,
) -> Result<bool> {
    // Both counts are answered by the mask on its own, whatever the form and whatever the type, so
    // they come back before there is any question of which loop to run.
    if matches!(feed, Feed::Counted) {
        for row in 0..rows {
            if !nulls.is_valid(row) {
                continue;
            }
            let Some(index) = into.index(row) else { continue };
            if let State::Counted { count, .. } = &mut states[index].state {
                *count += 1;
            }
        }
        return Ok(true);
    }
    match input.form() {
        Form::Flat => {
            let Some(data) = input.data() else { return Ok(false) };
            if data.len() < rows {
                return Ok(false);
            }
            let run = Run { input, data, rows, nulls };
            scatter(states, into, &run, identity, feed)
        }
        Form::Dictionary | Form::Rle => {
            let Some((codes, values)) = input.positions() else { return Ok(false) };
            if codes.len() < rows {
                return Ok(false);
            }
            let Some(data) = values.data() else { return Ok(false) };
            let run = Run { input, data, rows, nulls };
            // Every code is inside the dictionary because `Vector::dictionary` checks that on the
            // way in, so the gather below indexes without a bound of its own.
            scatter(states, into, &run, |index| codes[index] as usize, feed)
        }
        // A constant and a sequence both have a closed form per group that is better than any loop,
        // and neither is what a scan of a column produces, so both wait for the counter to ask.
        _ => Ok(false),
    }
}

/// One vector to read, in the shape the three loops below all want it.
struct Run<'r> {
    input: &'r Vector,
    data: &'r Data,
    rows: usize,
    nulls: &'r Validity,
}

fn scatter<M: Fn(usize) -> usize>(
    states: &mut [Accumulator],
    into: Where<'_>,
    run: &Run<'_>,
    at: M,
    feed: Feed,
) -> Result<bool> {
    match feed {
        Feed::Counted => Ok(true),
        Feed::Whole => whole_into(states, into, run, at),
        Feed::Real { scale } => real_into(states, into, run, at, scale),
        Feed::Extreme(least) => extreme_into(states, into, run, at, least),
    }
}

/// An exact number per row into the running total of the group that row belongs to.
fn whole_into<M: Fn(usize) -> usize>(
    states: &mut [Accumulator],
    into: Where<'_>,
    run: &Run<'_>,
    at: M,
) -> Result<bool> {
    macro_rules! each {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match run.data {
                $(Data::$variant(values) => {
                    for row in 0..run.rows {
                        if !run.nulls.is_valid(row) {
                            continue;
                        }
                        let Some(index) = into.index(row) else { continue };
                        fold_whole(&mut states[index], i128::from(values[at(row)]))?;
                    }
                })+
                // A total of hugeints can overflow inside one group, and then the overflow is the
                // answer rather than a detail, so both of those go the row at a time way.
                _ => return Ok(false),
            }
        };
    }
    rudb_vector::for_each_layout!(narrow, each);
    Ok(true)
}

/// One exact number into one accumulator, which is [`Accumulator::update`] with the `Value` gone.
fn fold_whole(into: &mut Accumulator, number: i128) -> Result<()> {
    match &mut into.state {
        State::Whole { total, seen, .. } | State::Scaled { total, seen, .. } => {
            *total = total.checked_add(number).ok_or_else(overflowed)?;
            *seen = true;
        }
        State::Mean { total, seen, exact, .. } => {
            match total.checked_add(number).filter(|_| *exact) {
                Some(sum) => *total = sum,
                None => {
                    let real = if *exact { exactly(*total) } else { mean_real(*total) };
                    *total = mean_bits(real + exactly(number));
                    *exact = false;
                }
            }
            *seen += 1;
        }
        // `feed_of` chose this loop off the state, so the states left over cannot be here.
        other => {
            return Err(Error::internal(format!("an exact total into {other:?}")));
        }
    }
    Ok(())
}

/// A number per row in floating point into the running total of the group that row belongs to.
#[expect(
    clippy::cast_precision_loss,
    reason = "a wide integer past 2^53 losing digits is what a double is, and this is the float path"
)]
fn real_into<M: Fn(usize) -> usize>(
    states: &mut [Accumulator],
    into: Where<'_>,
    run: &Run<'_>,
    at: M,
    scale: u8,
) -> Result<bool> {
    let factor = pow10(scale) as f64;
    let scaled = scale != 0;
    macro_rules! each {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match run.data {
                $(Data::$variant(values) => each!(@run values, |number| number as f64),)+
                Data::Float32(values) => each!(@run values, f64::from),
                Data::Float64(values) => each!(@run values, |number: f64| number),
                _ => return Ok(false),
            }
        };
        (@run $values:expr, $convert:expr) => {{
            let values = $values;
            let convert = $convert;
            for row in 0..run.rows {
                if !run.nulls.is_valid(row) {
                    continue;
                }
                let Some(index) = into.index(row) else { continue };
                let number = convert(values[at(row)]);
                fold_real(&mut states[index], if scaled { number / factor } else { number });
            }
        }};
    }
    rudb_vector::for_each_layout!(integer, each);
    Ok(true)
}

/// One approximate number into one accumulator.
fn fold_real(into: &mut Accumulator, number: f64) {
    match &mut into.state {
        State::Real { total, seen, .. } => {
            *total += number;
            *seen += 1;
        }
        State::Mean { total, seen, exact, .. } => {
            // The first value that is not whole. What was counted exactly so far comes across as
            // one conversion, and the rest of the group is added the way the float path adds.
            let real = if *exact { exactly(*total) } else { mean_real(*total) };
            *total = mean_bits(real + number);
            *exact = false;
            *seen += 1;
        }
        // `feed_of` chose this loop off the state, so the states left over cannot be here.
        _ => {}
    }
}

/// A number per row against the best the group it belongs to has seen.
fn extreme_into<M: Fn(usize) -> usize>(
    states: &mut [Accumulator],
    into: Where<'_>,
    run: &Run<'_>,
    at: M,
    least: bool,
) -> Result<bool> {
    macro_rules! each {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match run.data {
                $(Data::$variant(values) => {
                    // row at a time: a scatter is per row by definition, since two adjacent rows
                    // are usually two different groups and there is nothing to reduce before it.
                    // The `Value` below is built on a win rather than on a row, which is the part
                    // that makes this loop worth having over the one it replaced.
                    for row in 0..run.rows {
                        if !run.nulls.is_valid(row) {
                            continue;
                        }
                        let Some(index) = into.index(row) else { continue };
                        let number = i128::from(values[at(row)]);
                        let State::Extreme { held, .. } = &mut states[index].state else {
                            return Err(Error::internal("an extreme into a total".to_string()));
                        };
                        let replace = match held {
                            None => true,
                            Some(current) => {
                                let current: &Value = current.settle()?;
                                let mark = integral(current).ok_or_else(|| not_narrow(current))?;
                                if least { number < mark } else { number > mark }
                            }
                        };
                        // The `Value` is built on a win and not per row, which for a column that
                        // arrives sorted is once and for a column that arrives shuffled is about
                        // the harmonic number of the rows in the group.
                        if replace {
                            let value = run.input.try_value_at(row)?;
                            *held = Some(Box::new(Extremum::Held(value)));
                        }
                    }
                })+
                _ => return Ok(false),
            }
        };
    }
    rudb_vector::for_each_layout!(narrow, each);
    Ok(true)
}

fn not_narrow(value: &Value) -> Error {
    Error::not_implemented(format!("summing a {}", value.logical_type()))
}

/// An exact total as the double a mean divides, which is the one rounding `avg` over whole numbers
/// is allowed to do and is where duckdb does it too.
#[expect(
    clippy::cast_precision_loss,
    reason = "a total past 2^53 rounding once here is the definition of a double result"
)]
fn exactly(total: i128) -> f64 {
    total as f64
}

fn mean_bits(total: f64) -> i128 {
    i128::from(total.to_bits())
}

fn mean_real(bits: i128) -> f64 {
    f64::from_bits(bits as u64)
}

fn approximate_or_error(value: &Value) -> Result<f64> {
    crate::number::approximate(value).ok_or_else(|| not_narrow(value))
}

/// A value as an unscaled integer at a fixed scale.
fn at_scale(value: &Value, scale: u8) -> Option<i128> {
    match *value {
        Value::Decimal { unscaled, scale: held, .. } => rescale(unscaled, held, scale),
        _ => integral(value).and_then(|whole| whole.checked_mul(pow10(scale))),
    }
}

fn overflowed() -> Error {
    Error::out_of_range("Overflow in the running total of a sum".to_string())
}

fn overlong() -> Error {
    Error::out_of_range("more rows in one vector than a count can hold".to_string())
}

/// The scale a type holds its numbers at, which is zero for everything that is not a decimal.
fn decimal_scale(ty: &LogicalType) -> u8 {
    match *ty {
        LogicalType::Decimal { scale, .. } => scale,
        _ => 0,
    }
}

/// What one vector has to be read for.
#[derive(Clone, Copy)]
enum Want {
    /// A total of exact numbers, read at whatever scale they are already held at.
    Whole,
    /// A total in floating point, carried on from what is already there rather than restarted.
    Real { scale: u8, from: f64 },
    /// The row that wins, the smallest one if true and the largest one if false.
    Extreme(bool),
}

/// What one vector contributes.
enum Contribution {
    Whole(i128),
    Real { total: f64, seen: i64 },
    Extreme(Option<usize>),
}

/// Reads a vector once, whichever form it is in.
fn gather(input: &Vector, rows: usize, nulls: &Validity, want: Want) -> Option<Contribution> {
    match input.form() {
        Form::Flat => {
            let data = input.data()?;
            if data.len() < rows {
                return None;
            }
            // `DIRECT` says the mapping is the identity, which the loops below turn into a slice
            // of the first `rows` values. That is the only way the compiler gets to see that an
            // index cannot be out of range, and a bounds check per row was most of what this cost.
            collect::<true, _>(data, identity, rows, nulls, want)
        }
        Form::Dictionary | Form::Rle => {
            let (codes, values) = input.positions()?;
            // Cut to the rows wanted rather than checked against them, so that the loops below can
            // see that a row number is inside the codes and drop the check they would otherwise do
            // on every row. The codes are the half of the read that can say this: which value a
            // code lands on is a number out of the data and nothing knows it is in range until it
            // has been looked at.
            let codes = codes.get(..rows)?;
            let Some(data) = values.data() else {
                // A dictionary whose values sit in a file rather than in a run of memory, which is
                // what a scan of a native column hands over. There is nothing for the loops below
                // to read, but an extreme is decided on the bytes and the bytes can be asked for
                // one code at a time, so that much still works here.
                return match want {
                    Want::Extreme(least) => {
                        extreme_bytes(values, codes, nulls, least).map(Contribution::Extreme)
                    }
                    _ => None,
                };
            };
            if let (Want::Whole, Validity::AllValid) = (want, nulls) {
                if let Some(total) = tally(data, codes) {
                    return Some(Contribution::Whole(total));
                }
            }
            collect::<false, _>(data, |index| codes[index] as usize, rows, nulls, want)
        }
        Form::BitPacked => {
            let packed = input.packed_parts()?;
            match want {
                Want::Whole => {
                    let mut total = 0_i128;
                    for row in 0..rows {
                        if nulls.is_valid(row) {
                            total += packed.base() + i128::from(packed.code(row));
                        }
                    }
                    Some(Contribution::Whole(total))
                }
                Want::Real { from, .. } => {
                    let mut total = from;
                    let mut seen = 0_i64;
                    for row in 0..rows {
                        if nulls.is_valid(row) {
                            total += (packed.base() + i128::from(packed.code(row))) as f64;
                            seen += 1;
                        }
                    }
                    Some(Contribution::Real { total, seen })
                }
                Want::Extreme(least) => {
                    let mut found: Option<(usize, u64)> = None;
                    for row in 0..rows {
                        if !nulls.is_valid(row) {
                            continue;
                        }
                        let code = packed.code(row);
                        if found.is_none_or(
                            |(_, held)| {
                                if least { code < held } else { code > held }
                            },
                        ) {
                            found = Some((row, code));
                        }
                    }
                    Some(Contribution::Extreme(found.map(|(row, _)| row)))
                }
            }
        }
        // A constant folds in as one value repeated and a sequence as an arithmetic series, and
        // both have a closed form that is better than any loop. Neither is what a scan of a column
        // produces, so both wait for the counter to ask for them.
        _ => None,
    }
}

/// The row that wins an extreme over a dictionary whose values are not a run in memory.
///
/// `MIN` and `MAX` over a text column of a native file used to fall all the way back to the row at
/// a time path, because the dictionary a scan hands over keeps its payload in the file and so has
/// no `Data` to read. That path built one `Value::Varchar` per row, which is an allocation and a
/// copy of the string for every row of the column, and it was about ninety nanoseconds a row.
///
/// Winning is decided on the bytes and nothing else, which [`order`](crate::compare::order) says
/// for both of the types that arrive this way, so this compares the candidate against the bytes of
/// the row that is winning and never builds a value at all. The winner's bytes are kept here rather
/// than read again per row, since reading them can mean going back to the file.
///
/// The check against the code that is already winning is what makes this cheap on real data. A
/// column read out of a file repeats its codes, so most rows never reach the comparison.
///
/// `None` means this declines and the caller falls back, which is what happens for a type that does
/// not order on its bytes and for a code that lands on a dictionary entry that is null.
fn extreme_bytes(
    values: &Vector,
    codes: &[u32],
    nulls: &Validity,
    least: bool,
) -> Option<Option<usize>> {
    if !matches!(values.logical_type(), LogicalType::Varchar | LogicalType::Blob) {
        return None;
    }
    if let Some(ranks) = values.code_ranks() {
        return extreme_ranked(ranks, codes, nulls, least);
    }
    let mut winner: Option<(usize, u32)> = None;
    let mut best: Vec<u8> = Vec::new();
    // row at a time: a dictionary in a file answers one code at a time and there is no run to read.
    for (row, &code) in codes.iter().enumerate() {
        if !nulls.is_valid(row) {
            continue;
        }
        if let Some((_, held)) = winner {
            if held == code {
                continue;
            }
        }
        let candidate = values.try_bytes_at(code as usize).ok()??;
        let ahead = match winner {
            None => true,
            Some(_) => {
                let ordering = candidate.cmp(best.as_slice());
                if least { ordering.is_lt() } else { ordering.is_gt() }
            }
        };
        if ahead {
            best.clear();
            best.extend_from_slice(candidate);
            winner = Some((row, code));
        }
    }
    Some(winner.map(|(row, _)| row))
}

/// The row that wins an extreme, out of the sorted order the file wrote beside the dictionary.
///
/// This is [`extreme_bytes`] over a dictionary that knows which of any two of its values is smaller,
/// and then no value is read at all. The loop above reads one out of the payload for every row whose
/// code is not the code already winning, which on a column like `Referer` is nearly every row,
/// because eight hundred thousand rows there hold four hundred thousand distinct values and a repeat
/// almost never lands next to the thing it repeats. Here a row costs a load out of a map four bytes
/// wide per distinct value and a comparison of two integers.
///
/// A tie keeps the earlier row, which is what the byte loop does, and two rows tie here exactly when
/// they hold the same value, so the row the caller goes on to read is the same row either way.
///
/// `None` declines, for a code with no rank against it, and the caller falls back.
fn extreme_ranked(
    ranks: &[u32],
    codes: &[u32],
    nulls: &Validity,
    least: bool,
) -> Option<Option<usize>> {
    let mut winner: Option<(usize, u32)> = None;
    // row at a time: a code is a number out of the data, so which value it lands on is not known
    // for any row until that row has been read.
    for (row, &code) in codes.iter().enumerate() {
        if !nulls.is_valid(row) {
            continue;
        }
        let &rank = ranks.get(code as usize)?;
        if winner.is_none_or(|(_, held)| if least { rank < held } else { rank > held }) {
            winner = Some((row, rank));
        }
    }
    Some(winner.map(|(row, _)| row))
}

/// The widest dictionary [`tally`] copies, and a power of two.
///
/// The copy is per vector and the gather it speeds up is per row, so a dictionary wide enough that
/// copying it costs more than the fifteen hundred or so rows of a vector is one to leave alone.
/// `UserID` has about half a million distinct values, which is where the idea stops working
/// entirely, and the cut is well below that because the copy also has to stay in the first level
/// cache to be worth anything.
const TALLY_LIMIT: usize = 256;

/// The total of a dictionary encoded column, read through a padded copy of the dictionary.
///
/// The loop this replaces is `values[codes[row]]` for every row, and its cost is not the two loads.
/// It is that a code is data, so nothing knows it is inside the dictionary until it has been read,
/// so there is a bounds check and a branch in the middle of the loop on every row. That is about
/// thirteen instructions a value with no unrolling, and on a ten column integer scan of ClickBench's
/// `hits` it was the largest single thing in the program at twenty three percent.
///
/// So the dictionary is copied into an array whose length is a power of two known at compile time,
/// and the code is masked with that length instead of checked against it. Now the index is inside
/// the array by construction, the compiler can see it, the check and the branch are gone and the
/// loop unrolls. About seven instructions a value, and no branch in it that can be mispredicted.
///
/// The array is sized to the dictionary by a ladder rather than fixed at [`TALLY_LIMIT`], because it
/// has to be zeroed before the copy and a two entry dictionary should not pay for a two hundred and
/// fifty six entry one.
///
/// # Why not count the codes instead
///
/// Because it is slower, which is not what the instruction count says. Counting how often each code
/// appears and then taking one product per dictionary entry moves the multiplications off the per
/// row path entirely and measured six percent fewer instructions than the version here. It also
/// measured nine percent slower on one thread and twenty eight percent slower on thirty two, because
/// the row loop becomes a read modify write against memory at an address that comes out of the data.
/// Four counters per entry were not enough to keep consecutive rows off the same address, and every
/// repeat waits for the previous store to forward. A load has no such problem, and the loop here is
/// loads.
///
/// `None` when the dictionary is too wide to copy, and for a layout with no fixed width. A code past
/// the end of the dictionary reads one of the zeros the copy was padded with rather than raising,
/// and cannot happen: `Vector::dictionary` refuses one on the way in, which is the same invariant
/// the gather this replaces was already relying on to index without a bound of its own.
fn tally(data: &Data, codes: &[u32]) -> Option<i128> {
    // Each rung covers dictionaries up to its own size, and every rung is a power of two so that the
    // mask below is the bounds check.
    match data.len() {
        0..=8 => tally_into::<8>(data, codes),
        9..=32 => tally_into::<32>(data, codes),
        33..=128 => tally_into::<128>(data, codes),
        129..=TALLY_LIMIT => tally_into::<TALLY_LIMIT>(data, codes),
        _ => None,
    }
}

/// [`tally`] with the rung it decided on, `SLOTS` entries wide.
fn tally_into<const SLOTS: usize>(data: &Data, codes: &[u32]) -> Option<i128> {
    macro_rules! padded {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match data {
                $(Data::$variant(values) => padded!(@run values, $zero),)+
                _ => return None,
            }
        };
        (@run $values:expr, $zero:expr) => {{
            let values = $values.as_slice();
            if values.len() > SLOTS {
                return None;
            }
            let mut table = [$zero; SLOTS];
            table[..values.len()].copy_from_slice(values);
            let mut total: i128 = 0;
            // row at a time: this is the loop the whole function is. Masked rather than bounds
            // checked, which is the point: `SLOTS` is a power of two and is the length of `table`,
            // so the index is inside it by construction and the compiler can see that without any
            // unsafe code.
            for &code in codes {
                total += i128::from(table[code as usize & (SLOTS - 1)]);
            }
            total
        }};
    }
    Some(rudb_vector::for_each_layout!(narrow, padded))
}

/// The first `rows` values as one slice, when the mapping into them is the identity.
///
/// This is the whole of what `DIRECT` buys. A loop written as `values[at(index)]` has to check the
/// index against the length on every row, because nothing in the loop tells the compiler that `at`
/// answers something in range, and on a scan that check was most of what the sum cost. A loop over
/// a slice has the length in hand before it starts, so there is no check, the trip count is known
/// and the addition can go four at a time.
///
/// `None` when the mapping is not the identity, and also when the run is shorter than the rows
/// asked for, which is a caller whose data and length disagree and which goes the careful way
/// rather than panicking.
fn straight<const DIRECT: bool, T>(values: &[T], rows: usize) -> Option<&[T]> {
    if DIRECT { values.get(..rows) } else { None }
}

/// `DIRECT` says `at` is the identity, which is the flat case and is the one worth writing twice.
fn collect<const DIRECT: bool, M: Fn(usize) -> usize>(
    data: &Data,
    at: M,
    rows: usize,
    nulls: &Validity,
    want: Want,
) -> Option<Contribution> {
    match want {
        Want::Whole => whole_sum::<DIRECT, M>(data, at, rows, nulls).map(Contribution::Whole),
        Want::Real { scale, from } => real_sum::<DIRECT, M>(data, at, rows, nulls, scale, from),
        Want::Extreme(least) => {
            extreme::<DIRECT, M>(data, at, rows, nulls, least).map(Contribution::Extreme)
        }
    }
}

/// The total of the rows that are not null, as an exact number.
///
/// The accumulator is an `i128` and the widest thing read into it is sixty four bits, so a vector
/// would have to be about 2^63 rows long before its own total could overflow. That is what lets the
/// only overflow check be the one where this total meets the running total, which in turn is what
/// lets the loop vectorize at all.
fn whole_sum<const DIRECT: bool, M: Fn(usize) -> usize>(
    data: &Data,
    at: M,
    rows: usize,
    nulls: &Validity,
) -> Option<i128> {
    macro_rules! summed {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match data {
                $(Data::$variant(values) => summed!(@run values),)+
                // A total of hugeints can overflow inside one vector, and then the overflow is the
                // answer rather than a detail. The `narrow` group is exactly the widths where it
                // cannot, so both hugeints are out of it and both go the row at a time way, which is
                // the way that raises.
                _ => return None,
            }
        };
        (@run $values:expr) => {{
            let values = $values.as_slice();
            let run = straight::<DIRECT, _>(values, rows);
            let mut total: i128 = 0;
            match nulls {
                Validity::AllValid => match run {
                    Some(run) => {
                        for &value in run {
                            total += i128::from(value);
                        }
                    }
                    None => {
                        for index in 0..rows {
                            total += i128::from(values[at(index)]);
                        }
                    }
                },
                Validity::AllInvalid => {}
                Validity::Mask(mask) => {
                    // A word of the mask at a time, and a conditional move rather than a branch
                    // inside it, because the rows a filter leaves behind are in no pattern a branch
                    // predictor is going to learn.
                    for start in (0..rows).step_by(64) {
                        let word = mask.word(start / 64);
                        for index in start..(start + 64).min(rows) {
                            let number = match run {
                                Some(run) => i128::from(run[index]),
                                None => i128::from(values[at(index)]),
                            };
                            total += if word >> (index - start) & 1 == 1 { number } else { 0 };
                        }
                    }
                }
            }
            total
        }};
    }
    Some(rudb_vector::for_each_layout!(narrow, summed))
}

/// The running total carried through the rows that are not null, in floating point.
///
/// Sequential on purpose. Floating point addition is not associative, so four accumulators or a
/// reassociation would give an answer that is close to the one the row at a time loop gives rather
/// than the same one, and this file's whole job is to be the thing the fast paths are checked
/// against. What it does buy is the `Value` per row, the enum match per row and the null check per
/// row, and that is most of the cost.
#[expect(
    clippy::cast_precision_loss,
    reason = "a wide integer past 2^53 losing digits is what a double is, and this is the float path"
)]
fn real_sum<const DIRECT: bool, M: Fn(usize) -> usize>(
    data: &Data,
    at: M,
    rows: usize,
    nulls: &Validity,
    scale: u8,
    from: f64,
) -> Option<Contribution> {
    let factor = pow10(scale) as f64;
    let scaled = scale != 0;
    let all = i64::try_from(rows).ok()?;
    // Every integer arm converts with `as`, which for the widths below `2^53` is the same value
    // `f64::from` gives and for the ones above it is the rounding this whole function is about.
    // Splitting the list in two so that the narrow half could say `from` would be two lists that
    // produce the same code, which is two chances to put a width in the wrong one.
    macro_rules! added {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match data {
                $(Data::$variant(values) => added!(@run values, |number| number as f64),)+
                Data::Float32(values) => added!(@run values, f64::from),
                Data::Float64(values) => added!(@run values, |number: f64| number),
                _ => return None,
            }
        };
        (@run $values:expr, $convert:expr) => {{
            let values = $values.as_slice();
            let convert = $convert;
            let run = straight::<DIRECT, _>(values, rows);
            let mut total = from;
            let mut seen: i64 = 0;
            // Which rows count is decided once for the vector rather than once per row. A match on
            // the validity enum inside the loop is three quarters of a nanosecond a row on top of
            // an addition that takes one, which is a thing worth finding out by measuring.
            match nulls {
                Validity::AllValid => {
                    match run {
                        Some(run) => {
                            for &value in run {
                                let number = convert(value);
                                total += if scaled { number / factor } else { number };
                            }
                        }
                        None => {
                            for index in 0..rows {
                                let number = convert(values[at(index)]);
                                total += if scaled { number / factor } else { number };
                            }
                        }
                    }
                    seen = all;
                }
                Validity::AllInvalid => {}
                Validity::Mask(mask) => {
                    for start in (0..rows).step_by(64) {
                        let word = mask.word(start / 64);
                        for index in start..(start + 64).min(rows) {
                            if word >> (index - start) & 1 == 0 {
                                continue;
                            }
                            let number = match run {
                                Some(run) => convert(run[index]),
                                None => convert(values[at(index)]),
                            };
                            total += if scaled { number / factor } else { number };
                            seen += 1;
                        }
                    }
                }
            }
            (total, seen)
        }};
    }
    let (total, seen) = rudb_vector::for_each_layout!(integer, added);
    Some(Contribution::Real { total, seen })
}

/// Which row holds the smallest or largest number, or none if every row is null.
fn extreme<const DIRECT: bool, M: Fn(usize) -> usize>(
    data: &Data,
    at: M,
    rows: usize,
    nulls: &Validity,
    least: bool,
) -> Option<Option<usize>> {
    macro_rules! best {
        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
            match data {
                $(Data::$variant(values) => best!(@run values),)+
                // A float orders NaN the way the comparison kernel says rather than the way the
                // hardware does, and a string extreme is a comparison of bytes rather than of
                // numbers. Both are worth a loop of their own and neither gets a wrong one here.
                // The two hugeints are out because the seed and the running best are both `i128`.
                _ => return None,
            }
        };
        (@run $values:expr) => {{
            let values = $values.as_slice();
            let run = straight::<DIRECT, _>(values, rows);
            // The winner is a row number and a number, not an `Option` of a pair. Carrying the
            // option into the loop puts a discriminant test on every row, and the first row is the
            // only row that needs one, so the seed is the first row that is not null and the loop
            // starts after it.
            let mut held = usize::MAX;
            let mut mark: i128 = 0;
            match nulls {
                Validity::AllValid => match run {
                    Some(run) if !run.is_empty() => {
                        mark = i128::from(run[0]);
                        held = 0;
                        for (index, &value) in run.iter().enumerate().skip(1) {
                            let number = i128::from(value);
                            let win = if least { number < mark } else { number > mark };
                            if win {
                                mark = number;
                                held = index;
                            }
                        }
                    }
                    Some(_) => {}
                    None => {
                        if rows > 0 {
                            mark = i128::from(values[at(0)]);
                            held = 0;
                            for index in 1..rows {
                                let number = i128::from(values[at(index)]);
                                let win = if least { number < mark } else { number > mark };
                                if win {
                                    mark = number;
                                    held = index;
                                }
                            }
                        }
                    }
                },
                Validity::AllInvalid => {}
                Validity::Mask(mask) => {
                    for start in (0..rows).step_by(64) {
                        let word = mask.word(start / 64);
                        for index in start..(start + 64).min(rows) {
                            if word >> (index - start) & 1 == 0 {
                                continue;
                            }
                            let number = match run {
                                Some(run) => i128::from(run[index]),
                                None => i128::from(values[at(index)]),
                            };
                            let win = if least { number < mark } else { number > mark };
                            if held == usize::MAX || win {
                                mark = number;
                                held = index;
                            }
                        }
                    }
                }
            }
            (held != usize::MAX).then_some(held)
        }};
    }
    Some(rudb_vector::for_each_layout!(narrow, best))
}

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

    #[test]
    fn a_grouped_accumulator_does_not_carry_a_full_logical_type() {
        // A LogicalType can own a nested schema. Keeping one in every aggregate state cost more
        // than a hundred MiB on ClickBench q33 before the return was narrowed to Return.
        assert!(size_of::<Accumulator>() <= 32, "{} bytes", size_of::<Accumulator>());
    }

    fn run(name: &str, returns: &LogicalType, rows: &[Value]) -> Value {
        let mut accumulator = Accumulator::new(name, returns).expect("a known aggregate");
        for row in rows {
            accumulator.update(std::slice::from_ref(row)).expect("accumulates");
        }
        accumulator.finish().expect("finishes")
    }

    #[test]
    fn count_star_counts_rows_and_count_counts_values() {
        let mut stars = Accumulator::new("count_star", &LogicalType::BigInt).expect("known");
        for _ in 0..3 {
            stars.update(&[]).expect("no arguments");
        }
        assert_eq!(stars.finish().expect("finishes"), Value::BigInt(3));
        let counted = run(
            "count",
            &LogicalType::BigInt,
            &[Value::Integer(1), Value::Null, Value::Integer(3)],
        );
        assert_eq!(counted, Value::BigInt(2));
    }

    /// The distinction that makes `sum` over an empty group different from `count` over one.
    #[test]
    fn a_sum_of_nothing_is_null_and_a_count_of_nothing_is_zero() {
        assert_eq!(run("sum", &LogicalType::HugeInt, &[]), Value::Null);
        assert_eq!(run("sum", &LogicalType::HugeInt, &[Value::Null]), Value::Null);
        assert_eq!(run("count", &LogicalType::BigInt, &[]), Value::BigInt(0));
        assert_eq!(run("count_star", &LogicalType::BigInt, &[]), Value::BigInt(0));
    }

    #[test]
    fn a_sum_of_integers_accumulates_wider_than_it_reads() {
        let rows = vec![Value::Integer(i32::MAX); 4];
        let total = run("sum", &LogicalType::HugeInt, &rows);
        assert_eq!(total, Value::HugeInt(i128::from(i32::MAX) * 4));
    }

    #[test]
    fn an_average_divides_by_the_rows_it_saw_rather_than_the_rows_there_were() {
        let average =
            run("avg", &LogicalType::Double, &[Value::Integer(1), Value::Null, Value::Integer(3)]);
        assert_eq!(average, Value::Double(2.0));
    }

    /// Four whole numbers that are all past 2^53, so the two ways of averaging them differ.
    ///
    /// The last one is what the benchmark's `UserID` column is made of and is the reason this test
    /// exists: `AVG(UserID)` came out `435091026172918.3` here where duckdb said
    /// `435091026172920.25`.
    const WIDE: [i64; 4] = [435090932899640449, 435090932899640450, 1000003, 999999999999999999];

    /// The mean of [`WIDE`] the way duckdb computes it, which is the sum and then one division.
    fn wide_mean() -> f64 {
        exactly(WIDE.iter().map(|&number| i128::from(number)).sum()) / 4.0
    }

    fn wide_values() -> Vec<Value> {
        WIDE.iter().map(|&number| Value::BigInt(number)).collect()
    }

    #[test]
    fn an_average_of_whole_numbers_adds_them_up_exactly_and_divides_once() {
        // Adding these into a double as they arrive rounds at every step and the roundings do not
        // cancel, so the running answer is off in the last digit. The assertion that the two ways
        // disagree is there because without it this test would pass on a build that never fixed
        // anything.
        let mut running = 0.0_f64;
        for value in wide_values() {
            running += crate::number::approximate(&value).expect("a number");
        }
        assert_ne!(running / 4.0, wide_mean(), "the two ways of averaging have to differ here");
        assert_eq!(run("avg", &LogicalType::Double, &wide_values()), Value::Double(wide_mean()));
    }

    #[test]
    fn the_vector_path_averages_whole_numbers_exactly_as_well() {
        let values = wide_values();
        let vector = Vector::from_values(LogicalType::BigInt, &values).expect("a vector of these");
        let mut accumulator = Accumulator::new("avg", &LogicalType::Double).expect("a known one");
        accumulator.update_run(std::slice::from_ref(&vector), values.len()).expect("folds them in");
        assert_eq!(accumulator.finish().expect("finishes"), Value::Double(wide_mean()));
    }

    /// A column that is not whole numbers is added the way it always was, in order, in a double.
    #[test]
    fn an_average_of_doubles_is_the_running_total_the_float_path_produces() {
        let rows = [Value::Double(1e17), Value::Double(1.0), Value::Double(3.0)];
        let mut running = 0.0_f64;
        for value in &rows {
            running += crate::number::approximate(value).expect("a number");
        }
        assert_eq!(run("avg", &LogicalType::Double, &rows), Value::Double(running / 3.0));
    }

    #[test]
    fn min_and_max_skip_nulls_and_keep_the_value_rather_than_a_number() {
        let smallest = run(
            "min",
            &LogicalType::Varchar,
            &[Value::Varchar("b".into()), Value::Null, Value::Varchar("a".into())],
        );
        assert_eq!(smallest, Value::Varchar("a".into()));
        let largest = run(
            "max",
            &LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(7), Value::Integer(3)],
        );
        assert_eq!(largest, Value::Integer(7));
    }

    #[test]
    fn a_decimal_sums_at_its_own_scale() {
        let ty = LogicalType::decimal(10, 2).expect("a legal decimal");
        let total = run(
            "sum",
            &ty,
            &[
                Value::Decimal { unscaled: 250, width: 10, scale: 2 },
                Value::Decimal { unscaled: 125, width: 10, scale: 2 },
            ],
        );
        assert_eq!(total, Value::Decimal { unscaled: 375, width: 10, scale: 2 });
    }

    #[test]
    fn an_aggregate_nobody_has_written_says_which_one() {
        let error = Accumulator::new("median", &LogicalType::Double)
            .expect_err("median is not written yet");
        assert!(error.message().contains("the median aggregate"), "{error}");
    }

    /// The row at a time path, which is the answer the one pass path has to reach.
    fn row_at_a_time(name: &str, returns: &LogicalType, batches: &[Vector]) -> Result<Value> {
        let mut accumulator = Accumulator::new(name, returns)?;
        for batch in batches {
            for row in 0..batch.len() {
                let value = batch.value_at(row);
                accumulator.update(std::slice::from_ref(&value))?;
            }
        }
        accumulator.finish()
    }

    fn a_vector_at_a_time(name: &str, returns: &LogicalType, batches: &[Vector]) -> Result<Value> {
        let mut accumulator = Accumulator::new(name, returns)?;
        for batch in batches {
            accumulator.update_run(std::slice::from_ref(batch), batch.len())?;
        }
        accumulator.finish()
    }

    /// Both paths on the same batches, agreeing on the answer or agreeing on the complaint.
    fn agrees(name: &str, returns: &LogicalType, batches: &[Vector], note: &str) {
        let slow = row_at_a_time(name, returns, batches);
        let fast = a_vector_at_a_time(name, returns, batches);
        match (slow, fast) {
            (Ok(slow), Ok(fast)) => assert_eq!(slow, fast, "{note}"),
            (Err(slow), Err(fast)) => {
                assert_eq!(slow.message(), fast.message(), "{note}");
            }
            (slow, fast) => {
                panic!(
                    "{note}: one path answered and the other did not, {slow:?} against {fast:?}"
                );
            }
        }
    }

    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
        }
    }

    /// A number small enough to be legal in every type below, so that the property test is about
    /// the loops rather than about which types happen to hold which ranges.
    fn small(rng: &mut Rng) -> i64 {
        (rng.next() % 201) as i64 - 100
    }

    fn sample(ty: &LogicalType, rng: &mut Rng) -> Value {
        let number = small(rng);
        let positive = number.unsigned_abs();
        match *ty {
            LogicalType::TinyInt => Value::TinyInt(number as i8),
            LogicalType::SmallInt => Value::SmallInt(number as i16),
            LogicalType::Integer => Value::Integer(number as i32),
            LogicalType::BigInt => Value::BigInt(number),
            LogicalType::HugeInt => Value::HugeInt(i128::from(number)),
            LogicalType::UTinyInt => Value::UTinyInt(positive as u8),
            LogicalType::USmallInt => Value::USmallInt(positive as u16),
            LogicalType::UInteger => Value::UInteger(positive as u32),
            LogicalType::UBigInt => Value::UBigInt(positive),
            LogicalType::Float => Value::Float(number as f32 / 8.0),
            LogicalType::Double => Value::Double(number as f64 / 8.0),
            LogicalType::Decimal { width, scale } => {
                Value::Decimal { unscaled: i128::from(number) * 7, width, scale }
            }
            LogicalType::Varchar => Value::Varchar(format!("w{number}")),
            _ => panic!("no sample for {ty}"),
        }
    }

    fn flat(ty: &LogicalType, rows: usize, nulls: usize, rng: &mut Rng) -> Vector {
        let values: Vec<Value> = (0..rows)
            .map(
                |index| {
                    if nulls > 0 && index % nulls == 0 { Value::Null } else { sample(ty, rng) }
                },
            )
            .collect();
        Vector::from_values(ty.clone(), &values).expect("a vector of this type")
    }

    /// What the declared return type is for an aggregate over a column of this type.
    fn returns_of(name: &str, ty: &LogicalType) -> LogicalType {
        match name {
            "count" | "count_star" => LogicalType::BigInt,
            "avg" => LogicalType::Double,
            "min" | "max" => ty.clone(),
            _ => match *ty {
                LogicalType::Decimal { scale, .. } => {
                    LogicalType::decimal(rudb_common::MAX_DECIMAL_WIDTH, scale)
                        .expect("the widest decimal at this scale is legal")
                }
                LogicalType::Float | LogicalType::Double => LogicalType::Double,
                _ => LogicalType::HugeInt,
            },
        }
    }

    /// Every aggregate over every type this crate knows, in all three forms that have a loop and
    /// at three null densities, against the loop the loops replaced.
    #[test]
    fn every_aggregate_over_every_type_agrees_with_the_row_at_a_time_path() {
        let mut rng = Rng(0x5eed_ca11_ab1e_0003);
        let types = [
            LogicalType::TinyInt,
            LogicalType::SmallInt,
            LogicalType::Integer,
            LogicalType::BigInt,
            LogicalType::HugeInt,
            LogicalType::UTinyInt,
            LogicalType::USmallInt,
            LogicalType::UInteger,
            LogicalType::UBigInt,
            LogicalType::Float,
            LogicalType::Double,
            LogicalType::decimal(9, 2).expect("a legal decimal"),
            LogicalType::decimal(18, 4).expect("a legal decimal"),
            LogicalType::decimal(30, 6).expect("a legal decimal"),
            LogicalType::Varchar,
        ];
        for ty in &types {
            for name in ["count_star", "count", "sum", "avg", "min", "max"] {
                let returns = returns_of(name, ty);
                for nulls in [0_usize, 4, 1] {
                    // Two batches rather than one, because a running total that is restarted at
                    // every vector is right on one vector and wrong on the query.
                    let first = flat(ty, 97, nulls, &mut rng);
                    let second = flat(ty, 64, nulls, &mut rng);
                    let note = format!("{name} over {ty}, flat, one null in {nulls}");
                    agrees(name, &returns, &[first.clone(), second.clone()], &note);
                    let codes: Vec<u32> = (0..97).map(|index| (index % 13) as u32).collect();
                    let coded = Vector::dictionary(codes, first.clone()).expect("in range");
                    let note = format!("{name} over {ty}, dictionary, one null in {nulls}");
                    agrees(name, &returns, &[coded, second.clone()], &note);
                    // Runs of thirteen rows each, so a value the flat vector held once is read
                    // thirteen times and a run boundary lands inside a batch rather than on it.
                    let ends: Vec<u32> = (1..=8).map(|run| (run * 13).min(97)).collect();
                    let runs = Vector::runs(ends, first.slice(0, 8).expect("eight values"))
                        .expect("one value for each run");
                    let note = format!("{name} over {ty}, runs, one null in {nulls}");
                    agrees(name, &returns, &[runs, second], &note);
                }
            }
        }
    }

    /// How many aggregates a group holds in the test below, and which of them is the one measured.
    ///
    /// Not one and not the first one, because a stride of one and an offset of zero are the two
    /// values that make the index arithmetic right by accident.
    const STRIDE: usize = 3;
    const OFFSET: usize = 1;

    /// Which group each row belongs to, with some rows belonging to none.
    ///
    /// Round robin with a stride that is coprime with nothing in particular, so consecutive rows
    /// land in different groups, which is the case the scattered path exists for and the case a
    /// loop that quietly folded runs together would get wrong.
    fn deal(rows: usize, groups: usize) -> Vec<usize> {
        (0..rows).map(|row| if row % 11 == 5 { NOWHERE } else { (row * 7 + 3) % groups }).collect()
    }

    /// One accumulator per group fed a row at a time, which is the answer the scatter has to reach.
    fn group_at_a_time(
        name: &str,
        returns: &LogicalType,
        batches: &[(Vector, Vec<usize>)],
        groups: usize,
        reads: bool,
    ) -> Result<Vec<Value>> {
        let mut states = Vec::new();
        for _ in 0..groups {
            states.push(Accumulator::new(name, returns)?);
        }
        for (batch, slots) in batches {
            for (row, &slot) in slots.iter().enumerate() {
                if slot == NOWHERE {
                    continue;
                }
                if reads {
                    let value = batch.value_at(row);
                    states[slot].update(std::slice::from_ref(&value))?;
                } else {
                    states[slot].update(&[])?;
                }
            }
        }
        states.iter().map(Accumulator::finish).collect()
    }

    /// The same groups through one call per batch.
    fn group_at_once(
        name: &str,
        returns: &LogicalType,
        batches: &[(Vector, Vec<usize>)],
        groups: usize,
        reads: bool,
    ) -> Result<Vec<Value>> {
        let mut states = Vec::new();
        for _ in 0..groups * STRIDE {
            states.push(Accumulator::new(name, returns)?);
        }
        for (batch, slots) in batches {
            let input = reads.then_some(batch);
            update_scattered(&mut states, slots, STRIDE, OFFSET, input, slots.len())?;
        }
        (0..groups).map(|group| states[group * STRIDE + OFFSET].finish()).collect()
    }

    /// Every aggregate over every type, dealt out into five groups, against one accumulator each.
    ///
    /// This is the invariant the whole scattered path rests on and it is the same invariant the
    /// vector at a time test above asserts: a faster loop that reaches a different answer is not an
    /// answer. Two batches rather than one, because a state that is restarted at every vector is
    /// right on one vector and wrong on the query, and the slots change between them so no group
    /// sees the same rows twice.
    #[test]
    fn every_aggregate_scattered_into_groups_agrees_with_one_accumulator_per_group() {
        let mut rng = Rng(0x5eed_ca11_ab1e_0061);
        let groups = 5;
        let types = [
            LogicalType::TinyInt,
            LogicalType::SmallInt,
            LogicalType::Integer,
            LogicalType::BigInt,
            LogicalType::HugeInt,
            LogicalType::UTinyInt,
            LogicalType::USmallInt,
            LogicalType::UInteger,
            LogicalType::UBigInt,
            LogicalType::Float,
            LogicalType::Double,
            LogicalType::decimal(9, 2).expect("a legal decimal"),
            LogicalType::decimal(18, 4).expect("a legal decimal"),
            LogicalType::decimal(30, 6).expect("a legal decimal"),
            LogicalType::Varchar,
        ];
        for ty in &types {
            for name in ["count_star", "count", "sum", "avg", "min", "max"] {
                let returns = returns_of(name, ty);
                let reads = name != "count_star";
                for nulls in [0_usize, 4, 1] {
                    let first = flat(ty, 97, nulls, &mut rng);
                    let second = flat(ty, 64, nulls, &mut rng);
                    let codes: Vec<u32> = (0..97).map(|index| (index % 13) as u32).collect();
                    let coded = Vector::dictionary(codes, first.clone()).expect("codes in range");
                    for (shape, batches) in [
                        ("flat", vec![first.clone(), second.clone()]),
                        ("dictionary", vec![coded, second.clone()]),
                    ] {
                        let dealt: Vec<(Vector, Vec<usize>)> = batches
                            .into_iter()
                            .map(|batch| {
                                let slots = deal(batch.len(), groups);
                                (batch, slots)
                            })
                            .collect();
                        let note = format!("{name} over {ty}, {shape}, one null in {nulls}");
                        let slow = group_at_a_time(name, &returns, &dealt, groups, reads);
                        let fast = group_at_once(name, &returns, &dealt, groups, reads);
                        match (slow, fast) {
                            (Ok(slow), Ok(fast)) => assert_eq!(slow, fast, "{note}"),
                            (Err(slow), Err(fast)) => {
                                assert_eq!(slow.message(), fast.message(), "{note}");
                            }
                            (slow, fast) => panic!(
                                "{note}: one path answered and the other did not, \
                                 {slow:?} against {fast:?}"
                            ),
                        }
                    }
                }
            }
        }
    }

    /// A row pointing at [`NOWHERE`] contributes to nothing, which is how a `FILTER` and a spilled
    /// row are both said. Counting is the aggregate that would notice a row it should not have seen.
    #[test]
    fn a_row_that_belongs_to_no_group_is_counted_by_nobody() {
        let column = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(2), Value::Integer(3), Value::Integer(4)],
        )
        .expect("a vector of integers");
        let mut states = vec![Accumulator::new("sum", &LogicalType::HugeInt).expect("known"); 2];
        let slots = [0, NOWHERE, 1, NOWHERE];
        update_scattered(&mut states, &slots, 1, 0, Some(&column), 4).expect("folds them in");
        assert_eq!(states[0].finish().expect("finishes"), Value::HugeInt(1));
        assert_eq!(states[1].finish().expect("finishes"), Value::HugeInt(3));
    }

    /// The shapes a grouped ClickBench query is made of stay off the row at a time path, and a
    /// column the scatter has no typed loop for goes down it and says so.
    #[test]
    fn the_shapes_a_grouped_query_is_made_of_stay_off_the_row_at_a_time_path() {
        let numbers = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(2), Value::Integer(3)],
        )
        .expect("a vector of integers");
        let words = Vector::from_values(
            LogicalType::Varchar,
            &[Value::Varchar("a".into()), Value::Varchar("b".into()), Value::Varchar("c".into())],
        )
        .expect("a vector of strings");
        let slots = [0_usize, 1, 0];
        for (name, returns, column) in [
            ("count_star", LogicalType::BigInt, None),
            ("count", LogicalType::BigInt, Some(&numbers)),
            ("sum", LogicalType::HugeInt, Some(&numbers)),
            ("avg", LogicalType::Double, Some(&numbers)),
            ("min", LogicalType::Integer, Some(&numbers)),
            ("max", LogicalType::Integer, Some(&numbers)),
        ] {
            fallback::reset();
            let mut states = vec![Accumulator::new(name, &returns).expect("known"); 2];
            update_scattered(&mut states, &slots, 1, 0, column, 3).expect("folds them in");
            assert_eq!(
                fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat),
                0,
                "{name} over an integer column took the row at a time path"
            );
        }
        fallback::reset();
        let mut states = vec![Accumulator::new("min", &LogicalType::Varchar).expect("known"); 2];
        update_scattered(&mut states, &slots, 1, 0, Some(&words), 3).expect("folds them in");
        assert_eq!(states[0].finish().expect("finishes"), Value::Varchar("a".into()));
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);
    }

    /// The point of the shared accessor. A run length column goes down the same loop a dictionary
    /// does, so it does not reach the path that builds a `Value` a row, and the counter says so.
    #[test]
    fn a_sum_over_runs_takes_the_same_loop_a_dictionary_takes() {
        fallback::reset();
        let values = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(5), Value::Null, Value::Integer(7)],
        )
        .expect("a vector of integers");
        let runs = Vector::runs(vec![4, 6, 10], values).expect("one value for each run");
        assert_eq!(runs.form(), Form::Rle);
        let mut summing =
            Accumulator::new("sum", &LogicalType::HugeInt).expect("a known aggregate");
        summing.update_run(std::slice::from_ref(&runs), 10).expect("sums");
        // Four fives and four sevens, with the two nulls in the middle contributing nothing.
        assert_eq!(summing.finish().expect("finishes"), Value::HugeInt(48));
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Rle, Form::Rle), 0);
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);
        fallback::reset();
    }

    #[test]
    fn a_sum_of_numbers_stays_off_the_row_at_a_time_path_and_a_sum_of_strings_does_not() {
        fallback::reset();
        let numbers = Vector::from_values(
            LogicalType::Integer,
            &[Value::Integer(1), Value::Integer(2), Value::Integer(3)],
        )
        .expect("a vector of integers");
        let mut summing =
            Accumulator::new("sum", &LogicalType::HugeInt).expect("a known aggregate");
        summing.update_run(std::slice::from_ref(&numbers), 3).expect("sums");
        assert_eq!(summing.finish().expect("finishes"), Value::HugeInt(6));
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);

        let words = Vector::from_values(
            LogicalType::Varchar,
            &[Value::Varchar("a".into()), Value::Null, Value::Varchar("b".into())],
        )
        .expect("a vector of strings");
        let mut counting = Accumulator::new("count", &LogicalType::BigInt).expect("a known one");
        counting.update_run(std::slice::from_ref(&words), 3).expect("counts");
        assert_eq!(counting.finish().expect("finishes"), Value::BigInt(2));
        // A count reads the mask, so a type with no loop of its own is still not a fall through.
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);

        let mut wrong = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known aggregate");
        let error =
            wrong.update_run(std::slice::from_ref(&words), 3).expect_err("cannot sum those");
        assert!(error.message().contains("summing a"), "{error}");
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 1);
        fallback::reset();
    }

    /// The reason `update_run` carries the running total into the loop rather than totalling the
    /// vector on its own and adding the two at the end.
    #[test]
    fn a_floating_point_sum_carries_the_running_total_into_the_next_vector() {
        let first =
            Vector::from_values(LogicalType::Double, &[Value::Double(1.0e16)]).expect("a vector");
        let second = Vector::from_values(LogicalType::Double, &vec![Value::Double(1.0); 8])
            .expect("a vector");
        let batches = [first, second];
        let slow = row_at_a_time("sum", &LogicalType::Double, &batches).expect("sums");
        let fast = a_vector_at_a_time("sum", &LogicalType::Double, &batches).expect("sums");
        assert_eq!(slow, fast);
        // One at a time, every one of those eight disappears into the rounding. Eight at once does
        // not, which is what makes this a case worth having a test for.
        assert_eq!(slow, Value::Double(1.0e16));
        assert_ne!(1.0e16 + 8.0, 1.0e16);
    }

    /// The wrong answer that needs a dictionary, a null and one specific code to reproduce.
    #[test]
    fn a_null_behind_a_dictionary_code_is_skipped_by_every_aggregate() {
        let values = Vector::from_values(
            LogicalType::Integer,
            &[Value::Null, Value::Integer(5), Value::Integer(9)],
        )
        .expect("a vector of integers");
        let coded = Vector::dictionary(vec![0, 1, 0, 2, 0], values).expect("codes are in range");
        let batch = std::slice::from_ref(&coded);
        assert_eq!(
            a_vector_at_a_time("count", &LogicalType::BigInt, batch).expect("counts"),
            Value::BigInt(2)
        );
        assert_eq!(
            a_vector_at_a_time("sum", &LogicalType::HugeInt, batch).expect("sums"),
            Value::HugeInt(14)
        );
        assert_eq!(
            a_vector_at_a_time("min", &LogicalType::Integer, batch).expect("finds one"),
            Value::Integer(5)
        );
    }

    /// Stands in for the text a native file keeps, which is reachable one value at a time and is
    /// not a run of bytes anything can take a slice of.
    #[derive(Debug)]
    struct Filed(Vec<Vec<u8>>);

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

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

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

    /// A scan of a native text column hands over a dictionary whose payload is still in the file,
    /// so there is nothing for the gather to read and a minimum over one used to build a value per
    /// row. It is decided on the bytes, and the bytes can be asked for a code at a time.
    #[test]
    fn an_extreme_over_a_dictionary_that_keeps_its_bytes_in_a_file_is_decided_on_the_bytes() {
        let source = Arc::new(Filed(vec![b"pear".to_vec(), b"apple".to_vec(), b"plum".to_vec()]));
        let values = Vector::external_text(LogicalType::Varchar, source).expect("three values");
        let coded = Vector::dictionary(vec![0, 2, 1, 2, 0], values).expect("codes are in range");
        let batch = std::slice::from_ref(&coded);
        assert_eq!(
            a_vector_at_a_time("min", &LogicalType::Varchar, batch).expect("finds one"),
            Value::Varchar("apple".into())
        );
        assert_eq!(
            a_vector_at_a_time("max", &LogicalType::Varchar, batch).expect("finds one"),
            Value::Varchar("plum".into())
        );
        // The answers above are right either way, since falling back gets them too. These say the
        // vector path is the one that found them, and which row each of them won on.
        let smallest = gather(&coded, 5, &Validity::AllValid, Want::Extreme(true));
        assert!(matches!(smallest, Some(Contribution::Extreme(Some(2)))));
        let largest = gather(&coded, 5, &Validity::AllValid, Want::Extreme(false));
        assert!(matches!(largest, Some(Contribution::Extreme(Some(1)))));
    }

    /// The flat sum reads the first `rows` values as one slice rather than one index at a time, so
    /// a vector that is asked for fewer rows than it holds has to stop where it was told rather
    /// than where the data ends.
    #[test]
    fn a_run_shorter_than_the_vector_totals_only_the_rows_it_was_asked_for() {
        let rows: Vec<Value> = (1..=10).map(Value::Integer).collect();
        let vector = Vector::from_values(LogicalType::Integer, &rows).expect("a vector");
        let batch = std::slice::from_ref(&vector);
        // row at a time: the point is that each count gives a different answer, so there is
        // nothing to batch.
        for count in 0..=10usize {
            let mut accumulator =
                Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
            accumulator.update_run(batch, count).expect("totals");
            let wanted = (count * (count + 1) / 2) as i128;
            let got = accumulator.finish().expect("a total");
            if count == 0 {
                assert_eq!(got, Value::Null, "no rows is no total");
            } else {
                assert_eq!(got, Value::HugeInt(wanted), "the first {count} rows");
            }
        }
    }

    /// The padded copy and the plain gather are two ways to the same number, and which one runs
    /// depends on how wide the dictionary is. So the answer is held against the flat sum of the same
    /// rows at every width that matters: on each rung of the ladder, on the boundary between two of
    /// them, and past the last one where the copy is refused and the gather answers instead.
    #[test]
    fn a_dictionary_read_through_a_padded_copy_totals_what_the_same_rows_total_laid_out_flat() {
        // row at a time: each width is its own vector and its own expected answer.
        for distinct in [1usize, 2, 7, 255, TALLY_LIMIT, TALLY_LIMIT + 1, TALLY_LIMIT * 3] {
            let entries: Vec<Value> =
                (0..distinct).map(|slot| Value::Integer(slot as i32 * 7 - 11)).collect();
            let values = Vector::from_values(LogicalType::Integer, &entries).expect("a dictionary");
            // A stride that shares no factor with the rungs, so the codes walk the whole dictionary
            // rather than the first few entries of it.
            let codes: Vec<u32> = (0..1500u32).map(|row| row * 13 % distinct as u32).collect();
            let flat: Vec<Value> =
                codes.iter().map(|&code| entries[code as usize].clone()).collect();

            let coded = Vector::dictionary(codes, values).expect("codes are in range");
            let mut counted = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
            counted.update_run(std::slice::from_ref(&coded), 1500).expect("totals");

            let laid_out = Vector::from_values(LogicalType::Integer, &flat).expect("a vector");
            let mut gathered = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
            gathered.update_run(std::slice::from_ref(&laid_out), 1500).expect("totals");

            assert_eq!(
                counted.finish().expect("a total"),
                gathered.finish().expect("a total"),
                "a dictionary of {distinct} entries"
            );
        }
    }

    /// The copy is read for the rows it was asked for and not the ones past them, which the masking
    /// of the index would hide if the loop went over the whole code run.
    #[test]
    fn a_padded_dictionary_stops_at_the_rows_it_was_asked_for() {
        let entries = [Value::Integer(1), Value::Integer(100)];
        let values = Vector::from_values(LogicalType::Integer, &entries).expect("a dictionary");
        let coded = Vector::dictionary(vec![0, 0, 0, 1, 1], values).expect("codes are in range");
        let mut accumulator = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
        accumulator.update_run(std::slice::from_ref(&coded), 3).expect("totals");
        assert_eq!(
            accumulator.finish().expect("a total"),
            Value::HugeInt(3),
            "only the three ones"
        );
    }

    /// Why the whole sum stops at sixty four bits: at a hundred and twenty eight the total of one
    /// vector can overflow on its own, and the overflow is the answer rather than a detail.
    #[test]
    fn a_total_of_hugeints_goes_the_row_at_a_time_way_and_still_overflows() {
        fallback::reset();
        let rows = vec![Value::HugeInt(i128::MAX); 2];
        let vector = Vector::from_values(LogicalType::HugeInt, &rows).expect("a vector");
        let mut accumulator = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
        let error =
            accumulator.update_run(std::slice::from_ref(&vector), 2).expect_err("overflows");
        assert!(error.message().contains("Overflow in the running total"), "{error}");
        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 1);
        fallback::reset();
    }
}