Skip to main content

rudb_kernels/
aggregate.rs

1//! The aggregate accumulators.
2//!
3//! One of these per group per aggregate, so a `GROUP BY` over a million distinct keys with three
4//! aggregates in it holds three million of them. That is the reason the state is an enum of small
5//! fixed cases rather than a boxed trait object: the hash table is going to hold these inline and a
6//! pointer chase per row per aggregate is a cost that shows up on every grouped query there is.
7//!
8//! Null is skipped by every aggregate. `sum` over a column that is entirely null is null and not
9//! zero, `count(x)` counts the rows where `x` is not null, and `count(*)` counts rows without
10//! looking at anything. Those three are not variations on a theme, they are three different
11//! questions, and the reason `count(*)` is a separate function rather than `count` with a star
12//! argument is so the executor never has to work out which one it was handed.
13//!
14//! # How the vectorized path is put together
15//!
16//! The other four kernel files take a vector and give one back. This one has state, so the batch
17//! interface is [`Accumulator::update_run`], which folds a whole vector into the running state in
18//! one pass. What it can do in one pass depends on the aggregate, and the four shapes are worth
19//! naming because they are not the same problem.
20//!
21//! `count` and `count(*)` do not read the data at all. A count of rows is the row count and a count
22//! of values is the number of bits set in the validity mask, so those two are answered from the
23//! mask whatever the form and whatever the type, which is why they are the only ones that come back
24//! before the form dispatch.
25//!
26//! A whole sum reads the data and ignores the order. Rows that are null are masked out with a
27//! conditional move rather than a branch, and the accumulator is an `i128` so that nothing narrower
28//! than a `HUGEINT` can overflow inside one vector and the check only has to happen once, where the
29//! vector's total meets the running total.
30//!
31//! A floating point sum reads the data and does not ignore the order, because floating point
32//! addition is not associative and the answer this has to reach is the one the row at a time loop
33//! reaches. So that loop stays sequential and gives up the vectorization the whole sum gets. It is
34//! still about fifty times faster than building a `Value` per row, and an answer that is fast and
35//! different from the reference is not an answer.
36//!
37//! `min` and `max` read the data to find which row won and then ask the vector for that one row.
38//! One `Value` per vector instead of one per row, and one call into the comparison kernel instead
39//! of one per row.
40
41use rudb_common::{Error, LogicalType, Result, Value};
42use rudb_vector::{Data, Form, Validity, Vector};
43
44use crate::compare::order;
45use crate::fallback::{self, Kernel};
46use crate::number::{fit, integral, pow10, rescale};
47use crate::shape::{identity, nulls_of};
48
49/// A running aggregate.
50#[derive(Debug, Clone)]
51pub struct Accumulator {
52    kind: Kind,
53    returns: LogicalType,
54    state: State,
55}
56
57/// Which aggregate.
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59enum Kind {
60    CountStar,
61    Count,
62    Sum,
63    Avg,
64    Min,
65    Max,
66}
67
68/// What the aggregate has seen so far.
69#[derive(Debug, Clone)]
70enum State {
71    /// A row count, for `count` and `count(*)`.
72    Counted(i64),
73    /// A whole running total and whether anything landed in it.
74    Whole { total: i128, seen: bool },
75    /// A running total in floating point, and the count `avg` divides by.
76    Real { total: f64, seen: i64 },
77    /// A running total for `avg`, exact while every value folded in is a whole number.
78    ///
79    /// `avg` over an integer column has to add the column up exactly and divide once at the end.
80    /// Adding into a double as it goes gives a different number: each addition past `2^53` rounds,
81    /// and the roundings do not cancel. `AVG(UserID)` over ten thousand rows of the benchmark file
82    /// came out `435091026172918.3` that way where duckdb says `435091026172920.25`, which is the
83    /// sum divided once. So the total is an `i128` and the division is the only rounding.
84    ///
85    /// `exact` goes false the first time a value is not a whole number, or the first time the total
86    /// would overflow, and from then on `real` carries it. A column of doubles therefore lands on
87    /// the same additions in the same order as before, which is what the float path has to keep.
88    Mean { whole: i128, real: f64, seen: i64, exact: bool },
89    /// A running total at a fixed decimal scale.
90    Scaled { total: i128, scale: u8, seen: bool },
91    /// The smallest or largest value so far.
92    Extreme(Option<Value>),
93}
94
95impl Accumulator {
96    /// A fresh accumulator for a named aggregate returning `returns`.
97    ///
98    /// # Errors
99    ///
100    /// If the name is not an aggregate this crate implements.
101    pub fn new(name: &str, returns: &LogicalType) -> Result<Self> {
102        let kind = match name {
103            "count_star" => Kind::CountStar,
104            "count" => Kind::Count,
105            "sum" => Kind::Sum,
106            "avg" => Kind::Avg,
107            "min" => Kind::Min,
108            "max" => Kind::Max,
109            other => {
110                return Err(Error::not_implemented(format!("the {other} aggregate")));
111            }
112        };
113        let state = match kind {
114            Kind::CountStar | Kind::Count => State::Counted(0),
115            Kind::Avg => State::Mean { whole: 0, real: 0.0, seen: 0, exact: true },
116            Kind::Min | Kind::Max => State::Extreme(None),
117            Kind::Sum => match returns {
118                LogicalType::Decimal { scale, .. } => {
119                    State::Scaled { total: 0, scale: *scale, seen: false }
120                }
121                LogicalType::Float | LogicalType::Double => State::Real { total: 0.0, seen: 0 },
122                _ => State::Whole { total: 0, seen: false },
123            },
124        };
125        Ok(Self { kind, returns: returns.clone(), state })
126    }
127
128    /// Folds one row in.
129    ///
130    /// # Errors
131    ///
132    /// If the argument count is wrong for the aggregate, if the value is not one the aggregate can
133    /// accumulate, or if a whole running total overflows.
134    pub fn update(&mut self, args: &[Value]) -> Result<()> {
135        if self.kind == Kind::CountStar {
136            if let State::Counted(count) = &mut self.state {
137                *count += 1;
138            }
139            return Ok(());
140        }
141        let value = match args {
142            [only] => only,
143            _ => {
144                return Err(Error::internal(format!("an aggregate over {} arguments", args.len())));
145            }
146        };
147        if value.is_null() {
148            return Ok(());
149        }
150        match &mut self.state {
151            State::Counted(count) => *count += 1,
152            State::Whole { total, seen } => {
153                let whole = integral(value).ok_or_else(|| not_narrow(value))?;
154                *total = total.checked_add(whole).ok_or_else(overflowed)?;
155                *seen = true;
156            }
157            State::Real { total, seen } => {
158                *total += approximate_or_error(value)?;
159                *seen += 1;
160            }
161            State::Mean { whole, real, seen, exact } => {
162                match integral(value)
163                    .filter(|_| *exact)
164                    .and_then(|number| whole.checked_add(number))
165                {
166                    Some(total) => *whole = total,
167                    None => {
168                        // The first value that is not whole, or the first one that would overflow.
169                        // What was counted exactly so far comes across as one conversion, and the
170                        // rest of the column is added the way it always was.
171                        if *exact {
172                            *real = exactly(*whole);
173                            *exact = false;
174                        }
175                        *real += approximate_or_error(value)?;
176                    }
177                }
178                *seen += 1;
179            }
180            State::Scaled { total, scale, seen } => {
181                let unscaled = at_scale(value, *scale).ok_or_else(|| not_narrow(value))?;
182                *total = total.checked_add(unscaled).ok_or_else(overflowed)?;
183                *seen = true;
184            }
185            State::Extreme(held) => {
186                let replace = match held {
187                    None => true,
188                    Some(current) => {
189                        let ordering = order(value, current)?;
190                        match self.kind {
191                            Kind::Min => ordering.is_lt(),
192                            _ => ordering.is_gt(),
193                        }
194                    }
195                };
196                if replace {
197                    *held = Some(value.clone());
198                }
199            }
200        }
201        Ok(())
202    }
203
204    /// Folds a whole vector in, in one pass over the data and without building a [`Value`] per row.
205    ///
206    /// `rows` is how many rows to fold, which is the row count of the chunk rather than the capacity
207    /// of the vectors in it. `count(*)` takes no argument and reads nothing but that number, and
208    /// every other aggregate here takes exactly one vector.
209    ///
210    /// A shape the one pass form does not cover falls through to [`Accumulator::update`] per row and
211    /// records itself in [`crate::fallback`], so this always reaches the answer the row at a time
212    /// loop reaches and never a different one. That is not a slogan about floating point: the
213    /// running total below is carried into the vector loop rather than restarted at zero, precisely
214    /// so that the additions happen in the same order and round the same way.
215    ///
216    /// # Errors
217    ///
218    /// The same errors [`Accumulator::update`] raises, for the same reasons.
219    pub fn update_run(&mut self, args: &[Vector], rows: usize) -> Result<()> {
220        if self.kind == Kind::CountStar {
221            if let State::Counted(count) = &mut self.state {
222                *count += i64::try_from(rows).map_err(|_| overlong())?;
223            }
224            return Ok(());
225        }
226        let input = match args {
227            [only] => only,
228            _ => {
229                return Err(Error::internal(format!("an aggregate over {} arguments", args.len())));
230            }
231        };
232        if input.len() < rows {
233            return Err(Error::internal(format!(
234                "an aggregate handed {rows} rows and a vector of {}",
235                input.len()
236            )));
237        }
238        if self.folded(input, rows)? {
239            return Ok(());
240        }
241        // An aggregate reads one vector, so its form goes in both halves of the report rather than
242        // leaving a column of zeros next to every row of it.
243        fallback::record(Kernel::Aggregate, input.form(), input.form());
244        // row at a time: the path recorded on the line above, which exists to be correct for an
245        // aggregate `folded` does not cover and counts itself so that aggregate shows up.
246        for row in 0..rows {
247            let value = input.value_at(row);
248            self.update(std::slice::from_ref(&value))?;
249        }
250        Ok(())
251    }
252
253    /// Folds a vector in in one pass, or says this is a shape the one pass form does not cover.
254    ///
255    /// # Errors
256    ///
257    /// If a whole running total overflows where the row at a time loop would also have overflowed.
258    fn folded(&mut self, input: &Vector, rows: usize) -> Result<bool> {
259        let nulls = nulls_of(input);
260        // Both counts are answered by the mask on its own, whatever the form is and whatever the
261        // type is, so they come back before there is any question of which loop to run.
262        if let State::Counted(count) = &mut self.state {
263            *count += i64::try_from(nulls.count_valid(rows)).map_err(|_| overlong())?;
264            return Ok(true);
265        }
266        let least = self.kind == Kind::Min;
267        let want = match (&self.state, input.logical_type()) {
268            (State::Whole { .. }, _) => Want::Whole,
269            (State::Real { total, .. }, ty) => {
270                Want::Real { scale: decimal_scale(ty), from: *total }
271            }
272            // An exact mean over an integer column is read the way a sum is and divided at the end.
273            // Anything else is the float path, carried on from wherever the total is now, which for
274            // a mean that was exact until this vector is the exact total converted once.
275            (State::Mean { exact: true, .. }, ty) if ty.is_integer() => Want::Whole,
276            (State::Mean { whole, real, exact, .. }, ty) => {
277                let from = if *exact { exactly(*whole) } else { *real };
278                Want::Real { scale: decimal_scale(ty), from }
279            }
280            // A total at the scale the column is already held at is a sum of the raw unscaled
281            // integers and nothing else, which is the case every real query is in, because the sum
282            // of a `DECIMAL(15, 2)` column is declared at scale two. An integer summed into a
283            // decimal total, or a decimal at some other scale, needs a rescale per row that the row
284            // at a time path already does correctly, so those go that way and the counter says
285            // whether that was the wrong call.
286            (State::Scaled { scale, .. }, LogicalType::Decimal { scale: held, .. })
287                if held == scale =>
288            {
289                Want::Whole
290            }
291            (State::Scaled { .. }, _) => return Ok(false),
292            (State::Extreme(_), _) => Want::Extreme(least),
293            (State::Counted(_), _) => return Ok(false),
294        };
295        let Some(contribution) = gather(input, rows, &nulls, want) else {
296            return Ok(false);
297        };
298        let live = nulls.count_valid(rows);
299        match (&mut self.state, contribution) {
300            (
301                State::Whole { total, seen } | State::Scaled { total, seen, .. },
302                Contribution::Whole(sum),
303            ) => {
304                *total = total.checked_add(sum).ok_or_else(overflowed)?;
305                *seen |= live > 0;
306            }
307            (State::Real { total, seen }, Contribution::Real { total: carried, seen: added }) => {
308                *total = carried;
309                *seen += added;
310            }
311            (State::Mean { whole, seen, .. }, Contribution::Whole(sum)) => {
312                // An overflow here is not an error the way it is for a sum, because the row at a
313                // time loop answers an overflowing mean in floating point rather than raising. So
314                // this hands the vector back and that loop folds it in, state untouched.
315                let Some(total) = whole.checked_add(sum) else { return Ok(false) };
316                *whole = total;
317                *seen += i64::try_from(live).map_err(|_| overlong())?;
318            }
319            (
320                State::Mean { real, seen, exact, .. },
321                Contribution::Real { total: carried, seen: added },
322            ) => {
323                *real = carried;
324                *exact = false;
325                *seen += added;
326            }
327            (State::Extreme(held), Contribution::Extreme(Some(index))) => {
328                // One `Value` for the whole vector and one call into the comparison kernel, rather
329                // than one of each per row. The row that won is found on the numbers.
330                let candidate = input.value_at(index);
331                let replace = match held {
332                    None => true,
333                    Some(current) => {
334                        let ordering = order(&candidate, current)?;
335                        if least { ordering.is_lt() } else { ordering.is_gt() }
336                    }
337                };
338                if replace {
339                    *held = Some(candidate);
340                }
341            }
342            (State::Extreme(_), Contribution::Extreme(None)) => {}
343            // The `want` above picks the contribution, so the pairs left over are ones that cannot
344            // be built. Falling through costs a slow loop and a wrong answer costs a lot more.
345            _ => return Ok(false),
346        }
347        Ok(true)
348    }
349
350    /// The aggregate's answer.
351    ///
352    /// # Errors
353    ///
354    /// If the running total does not fit the declared return type.
355    pub fn finish(&self) -> Result<Value> {
356        match &self.state {
357            State::Counted(count) => Ok(Value::BigInt(*count)),
358            State::Whole { total, seen } => {
359                if !seen {
360                    return Ok(Value::Null);
361                }
362                fit(*total, &self.returns).ok_or_else(|| {
363                    Error::out_of_range(format!(
364                        "a sum of {total} does not fit in {}",
365                        self.returns
366                    ))
367                })
368            }
369            State::Real { total, seen } => {
370                if *seen == 0 {
371                    return Ok(Value::Null);
372                }
373                #[expect(
374                    clippy::cast_precision_loss,
375                    reason = "the count of rows in one group is well inside the exact range"
376                )]
377                let answer = if self.kind == Kind::Avg { total / *seen as f64 } else { *total };
378                if matches!(self.returns, LogicalType::Float) {
379                    #[expect(
380                        clippy::cast_possible_truncation,
381                        reason = "a declared FLOAT result is a FLOAT"
382                    )]
383                    return Ok(Value::Float(answer as f32));
384                }
385                Ok(Value::Double(answer))
386            }
387            State::Mean { whole, real, seen, exact } => {
388                if *seen == 0 {
389                    return Ok(Value::Null);
390                }
391                let total = if *exact { exactly(*whole) } else { *real };
392                #[expect(
393                    clippy::cast_precision_loss,
394                    reason = "the count of rows in one group is well inside the exact range"
395                )]
396                let answer = total / *seen as f64;
397                if matches!(self.returns, LogicalType::Float) {
398                    #[expect(
399                        clippy::cast_possible_truncation,
400                        reason = "a declared FLOAT result is a FLOAT"
401                    )]
402                    return Ok(Value::Float(answer as f32));
403                }
404                Ok(Value::Double(answer))
405            }
406            State::Scaled { total, scale, seen } => {
407                if !seen {
408                    return Ok(Value::Null);
409                }
410                let width = match self.returns {
411                    LogicalType::Decimal { width, .. } => width,
412                    _ => rudb_common::MAX_DECIMAL_WIDTH,
413                };
414                Ok(Value::Decimal { unscaled: *total, width, scale: *scale })
415            }
416            State::Extreme(held) => Ok(held.clone().unwrap_or(Value::Null)),
417        }
418    }
419}
420
421fn not_narrow(value: &Value) -> Error {
422    Error::not_implemented(format!("summing a {}", value.logical_type()))
423}
424
425/// An exact total as the double a mean divides, which is the one rounding `avg` over whole numbers
426/// is allowed to do and is where duckdb does it too.
427#[expect(
428    clippy::cast_precision_loss,
429    reason = "a total past 2^53 rounding once here is the definition of a double result"
430)]
431fn exactly(total: i128) -> f64 {
432    total as f64
433}
434
435fn approximate_or_error(value: &Value) -> Result<f64> {
436    crate::number::approximate(value).ok_or_else(|| not_narrow(value))
437}
438
439/// A value as an unscaled integer at a fixed scale.
440fn at_scale(value: &Value, scale: u8) -> Option<i128> {
441    match *value {
442        Value::Decimal { unscaled, scale: held, .. } => rescale(unscaled, held, scale),
443        _ => integral(value).and_then(|whole| whole.checked_mul(pow10(scale))),
444    }
445}
446
447fn overflowed() -> Error {
448    Error::out_of_range("Overflow in the running total of a sum".to_string())
449}
450
451fn overlong() -> Error {
452    Error::out_of_range("more rows in one vector than a count can hold".to_string())
453}
454
455/// The scale a type holds its numbers at, which is zero for everything that is not a decimal.
456fn decimal_scale(ty: &LogicalType) -> u8 {
457    match *ty {
458        LogicalType::Decimal { scale, .. } => scale,
459        _ => 0,
460    }
461}
462
463/// What one vector has to be read for.
464#[derive(Clone, Copy)]
465enum Want {
466    /// A total of exact numbers, read at whatever scale they are already held at.
467    Whole,
468    /// A total in floating point, carried on from what is already there rather than restarted.
469    Real { scale: u8, from: f64 },
470    /// The row that wins, the smallest one if true and the largest one if false.
471    Extreme(bool),
472}
473
474/// What one vector contributes.
475enum Contribution {
476    Whole(i128),
477    Real { total: f64, seen: i64 },
478    Extreme(Option<usize>),
479}
480
481/// Reads a vector once, whichever form it is in.
482fn gather(input: &Vector, rows: usize, nulls: &Validity, want: Want) -> Option<Contribution> {
483    match input.form() {
484        Form::Flat => {
485            let data = input.data()?;
486            if data.len() < rows {
487                return None;
488            }
489            collect(data, identity, rows, nulls, want)
490        }
491        Form::Dictionary => {
492            let (codes, values) = input.dictionary_parts()?;
493            if codes.len() < rows {
494                return None;
495            }
496            // Every code is inside the dictionary because `Vector::dictionary` checks that on the
497            // way in, so the gather below indexes without a bound of its own.
498            collect(values.data()?, |index| codes[index] as usize, rows, nulls, want)
499        }
500        // A constant folds in as one value repeated and a sequence as an arithmetic series, and
501        // both have a closed form that is better than any loop. Neither is what a scan of a column
502        // produces, so both wait for the counter to ask for them.
503        _ => None,
504    }
505}
506
507fn collect<M: Fn(usize) -> usize>(
508    data: &Data,
509    at: M,
510    rows: usize,
511    nulls: &Validity,
512    want: Want,
513) -> Option<Contribution> {
514    match want {
515        Want::Whole => whole_sum(data, at, rows, nulls).map(Contribution::Whole),
516        Want::Real { scale, from } => real_sum(data, at, rows, nulls, scale, from),
517        Want::Extreme(least) => extreme(data, at, rows, nulls, least).map(Contribution::Extreme),
518    }
519}
520
521/// The total of the rows that are not null, as an exact number.
522///
523/// The accumulator is an `i128` and the widest thing read into it is sixty four bits, so a vector
524/// would have to be about 2^63 rows long before its own total could overflow. That is what lets the
525/// only overflow check be the one where this total meets the running total, which in turn is what
526/// lets the loop vectorize at all.
527fn whole_sum<M: Fn(usize) -> usize>(
528    data: &Data,
529    at: M,
530    rows: usize,
531    nulls: &Validity,
532) -> Option<i128> {
533    macro_rules! summed {
534        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
535            match data {
536                $(Data::$variant(values) => summed!(@run values),)+
537                // A total of hugeints can overflow inside one vector, and then the overflow is the
538                // answer rather than a detail. The `narrow` group is exactly the widths where it
539                // cannot, so both hugeints are out of it and both go the row at a time way, which is
540                // the way that raises.
541                _ => return None,
542            }
543        };
544        (@run $values:expr) => {{
545            let values = $values;
546            let mut total: i128 = 0;
547            match nulls {
548                Validity::AllValid => {
549                    for index in 0..rows {
550                        total += i128::from(values[at(index)]);
551                    }
552                }
553                Validity::AllInvalid => {}
554                Validity::Mask(mask) => {
555                    // A word of the mask at a time, and a conditional move rather than a branch
556                    // inside it, because the rows a filter leaves behind are in no pattern a branch
557                    // predictor is going to learn.
558                    for start in (0..rows).step_by(64) {
559                        let word = mask.word(start / 64);
560                        for index in start..(start + 64).min(rows) {
561                            let number = i128::from(values[at(index)]);
562                            total += if word >> (index - start) & 1 == 1 { number } else { 0 };
563                        }
564                    }
565                }
566            }
567            total
568        }};
569    }
570    Some(rudb_vector::for_each_layout!(narrow, summed))
571}
572
573/// The running total carried through the rows that are not null, in floating point.
574///
575/// Sequential on purpose. Floating point addition is not associative, so four accumulators or a
576/// reassociation would give an answer that is close to the one the row at a time loop gives rather
577/// than the same one, and this file's whole job is to be the thing the fast paths are checked
578/// against. What it does buy is the `Value` per row, the enum match per row and the null check per
579/// row, and that is most of the cost.
580#[expect(
581    clippy::cast_precision_loss,
582    reason = "a wide integer past 2^53 losing digits is what a double is, and this is the float path"
583)]
584fn real_sum<M: Fn(usize) -> usize>(
585    data: &Data,
586    at: M,
587    rows: usize,
588    nulls: &Validity,
589    scale: u8,
590    from: f64,
591) -> Option<Contribution> {
592    let factor = pow10(scale) as f64;
593    let scaled = scale != 0;
594    let all = i64::try_from(rows).ok()?;
595    // Every integer arm converts with `as`, which for the widths below `2^53` is the same value
596    // `f64::from` gives and for the ones above it is the rounding this whole function is about.
597    // Splitting the list in two so that the narrow half could say `from` would be two lists that
598    // produce the same code, which is two chances to put a width in the wrong one.
599    macro_rules! added {
600        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
601            match data {
602                $(Data::$variant(values) => added!(@run values, |number| number as f64),)+
603                Data::Float32(values) => added!(@run values, f64::from),
604                Data::Float64(values) => added!(@run values, |number: f64| number),
605                _ => return None,
606            }
607        };
608        (@run $values:expr, $convert:expr) => {{
609            let values = $values;
610            let convert = $convert;
611            let mut total = from;
612            let mut seen: i64 = 0;
613            // Which rows count is decided once for the vector rather than once per row. A match on
614            // the validity enum inside the loop is three quarters of a nanosecond a row on top of
615            // an addition that takes one, which is a thing worth finding out by measuring.
616            match nulls {
617                Validity::AllValid => {
618                    for index in 0..rows {
619                        let number = convert(values[at(index)]);
620                        total += if scaled { number / factor } else { number };
621                    }
622                    seen = all;
623                }
624                Validity::AllInvalid => {}
625                Validity::Mask(mask) => {
626                    for start in (0..rows).step_by(64) {
627                        let word = mask.word(start / 64);
628                        for index in start..(start + 64).min(rows) {
629                            if word >> (index - start) & 1 == 0 {
630                                continue;
631                            }
632                            let number = convert(values[at(index)]);
633                            total += if scaled { number / factor } else { number };
634                            seen += 1;
635                        }
636                    }
637                }
638            }
639            (total, seen)
640        }};
641    }
642    let (total, seen) = rudb_vector::for_each_layout!(integer, added);
643    Some(Contribution::Real { total, seen })
644}
645
646/// Which row holds the smallest or largest number, or none if every row is null.
647fn extreme<M: Fn(usize) -> usize>(
648    data: &Data,
649    at: M,
650    rows: usize,
651    nulls: &Validity,
652    least: bool,
653) -> Option<Option<usize>> {
654    macro_rules! best {
655        ($(($variant:ident, $native:ty, $zero:expr)),+ $(,)?) => {
656            match data {
657                $(Data::$variant(values) => best!(@run values),)+
658                // A float orders NaN the way the comparison kernel says rather than the way the
659                // hardware does, and a string extreme is a comparison of bytes rather than of
660                // numbers. Both are worth a loop of their own and neither gets a wrong one here.
661                // The two hugeints are out because the seed and the running best are both `i128`.
662                _ => return None,
663            }
664        };
665        (@run $values:expr) => {{
666            let values = $values;
667            // The winner is a row number and a number, not an `Option` of a pair. Carrying the
668            // option into the loop puts a discriminant test on every row, and the first row is the
669            // only row that needs one, so the seed is the first row that is not null and the loop
670            // starts after it.
671            let mut held = usize::MAX;
672            let mut mark: i128 = 0;
673            match nulls {
674                Validity::AllValid => {
675                    if rows > 0 {
676                        mark = i128::from(values[at(0)]);
677                        held = 0;
678                        for index in 1..rows {
679                            let number = i128::from(values[at(index)]);
680                            let win = if least { number < mark } else { number > mark };
681                            if win {
682                                mark = number;
683                                held = index;
684                            }
685                        }
686                    }
687                }
688                Validity::AllInvalid => {}
689                Validity::Mask(mask) => {
690                    for start in (0..rows).step_by(64) {
691                        let word = mask.word(start / 64);
692                        for index in start..(start + 64).min(rows) {
693                            if word >> (index - start) & 1 == 0 {
694                                continue;
695                            }
696                            let number = i128::from(values[at(index)]);
697                            let win = if least { number < mark } else { number > mark };
698                            if held == usize::MAX || win {
699                                mark = number;
700                                held = index;
701                            }
702                        }
703                    }
704                }
705            }
706            (held != usize::MAX).then_some(held)
707        }};
708    }
709    Some(rudb_vector::for_each_layout!(narrow, best))
710}
711
712#[cfg(test)]
713mod tests {
714    use super::*;
715
716    fn run(name: &str, returns: &LogicalType, rows: &[Value]) -> Value {
717        let mut accumulator = Accumulator::new(name, returns).expect("a known aggregate");
718        for row in rows {
719            accumulator.update(std::slice::from_ref(row)).expect("accumulates");
720        }
721        accumulator.finish().expect("finishes")
722    }
723
724    #[test]
725    fn count_star_counts_rows_and_count_counts_values() {
726        let mut stars = Accumulator::new("count_star", &LogicalType::BigInt).expect("known");
727        for _ in 0..3 {
728            stars.update(&[]).expect("no arguments");
729        }
730        assert_eq!(stars.finish().expect("finishes"), Value::BigInt(3));
731        let counted = run(
732            "count",
733            &LogicalType::BigInt,
734            &[Value::Integer(1), Value::Null, Value::Integer(3)],
735        );
736        assert_eq!(counted, Value::BigInt(2));
737    }
738
739    /// The distinction that makes `sum` over an empty group different from `count` over one.
740    #[test]
741    fn a_sum_of_nothing_is_null_and_a_count_of_nothing_is_zero() {
742        assert_eq!(run("sum", &LogicalType::HugeInt, &[]), Value::Null);
743        assert_eq!(run("sum", &LogicalType::HugeInt, &[Value::Null]), Value::Null);
744        assert_eq!(run("count", &LogicalType::BigInt, &[]), Value::BigInt(0));
745        assert_eq!(run("count_star", &LogicalType::BigInt, &[]), Value::BigInt(0));
746    }
747
748    #[test]
749    fn a_sum_of_integers_accumulates_wider_than_it_reads() {
750        let rows = vec![Value::Integer(i32::MAX); 4];
751        let total = run("sum", &LogicalType::HugeInt, &rows);
752        assert_eq!(total, Value::HugeInt(i128::from(i32::MAX) * 4));
753    }
754
755    #[test]
756    fn an_average_divides_by_the_rows_it_saw_rather_than_the_rows_there_were() {
757        let average =
758            run("avg", &LogicalType::Double, &[Value::Integer(1), Value::Null, Value::Integer(3)]);
759        assert_eq!(average, Value::Double(2.0));
760    }
761
762    /// Four whole numbers that are all past 2^53, so the two ways of averaging them differ.
763    ///
764    /// The last one is what the benchmark's `UserID` column is made of and is the reason this test
765    /// exists: `AVG(UserID)` came out `435091026172918.3` here where duckdb said
766    /// `435091026172920.25`.
767    const WIDE: [i64; 4] = [435090932899640449, 435090932899640450, 1000003, 999999999999999999];
768
769    /// The mean of [`WIDE`] the way duckdb computes it, which is the sum and then one division.
770    fn wide_mean() -> f64 {
771        exactly(WIDE.iter().map(|&number| i128::from(number)).sum()) / 4.0
772    }
773
774    fn wide_values() -> Vec<Value> {
775        WIDE.iter().map(|&number| Value::BigInt(number)).collect()
776    }
777
778    #[test]
779    fn an_average_of_whole_numbers_adds_them_up_exactly_and_divides_once() {
780        // Adding these into a double as they arrive rounds at every step and the roundings do not
781        // cancel, so the running answer is off in the last digit. The assertion that the two ways
782        // disagree is there because without it this test would pass on a build that never fixed
783        // anything.
784        let mut running = 0.0_f64;
785        for value in wide_values() {
786            running += crate::number::approximate(&value).expect("a number");
787        }
788        assert_ne!(running / 4.0, wide_mean(), "the two ways of averaging have to differ here");
789        assert_eq!(run("avg", &LogicalType::Double, &wide_values()), Value::Double(wide_mean()));
790    }
791
792    #[test]
793    fn the_vector_path_averages_whole_numbers_exactly_as_well() {
794        let values = wide_values();
795        let vector = Vector::from_values(LogicalType::BigInt, &values).expect("a vector of these");
796        let mut accumulator = Accumulator::new("avg", &LogicalType::Double).expect("a known one");
797        accumulator.update_run(std::slice::from_ref(&vector), values.len()).expect("folds them in");
798        assert_eq!(accumulator.finish().expect("finishes"), Value::Double(wide_mean()));
799    }
800
801    /// A column that is not whole numbers is added the way it always was, in order, in a double.
802    #[test]
803    fn an_average_of_doubles_is_the_running_total_the_float_path_produces() {
804        let rows = [Value::Double(1e17), Value::Double(1.0), Value::Double(3.0)];
805        let mut running = 0.0_f64;
806        for value in &rows {
807            running += crate::number::approximate(value).expect("a number");
808        }
809        assert_eq!(run("avg", &LogicalType::Double, &rows), Value::Double(running / 3.0));
810    }
811
812    #[test]
813    fn min_and_max_skip_nulls_and_keep_the_value_rather_than_a_number() {
814        let smallest = run(
815            "min",
816            &LogicalType::Varchar,
817            &[Value::Varchar("b".into()), Value::Null, Value::Varchar("a".into())],
818        );
819        assert_eq!(smallest, Value::Varchar("a".into()));
820        let largest = run(
821            "max",
822            &LogicalType::Integer,
823            &[Value::Integer(1), Value::Integer(7), Value::Integer(3)],
824        );
825        assert_eq!(largest, Value::Integer(7));
826    }
827
828    #[test]
829    fn a_decimal_sums_at_its_own_scale() {
830        let ty = LogicalType::decimal(10, 2).expect("a legal decimal");
831        let total = run(
832            "sum",
833            &ty,
834            &[
835                Value::Decimal { unscaled: 250, width: 10, scale: 2 },
836                Value::Decimal { unscaled: 125, width: 10, scale: 2 },
837            ],
838        );
839        assert_eq!(total, Value::Decimal { unscaled: 375, width: 10, scale: 2 });
840    }
841
842    #[test]
843    fn an_aggregate_nobody_has_written_says_which_one() {
844        let error = Accumulator::new("median", &LogicalType::Double)
845            .expect_err("median is not written yet");
846        assert!(error.message().contains("the median aggregate"), "{error}");
847    }
848
849    /// The row at a time path, which is the answer the one pass path has to reach.
850    fn row_at_a_time(name: &str, returns: &LogicalType, batches: &[Vector]) -> Result<Value> {
851        let mut accumulator = Accumulator::new(name, returns)?;
852        for batch in batches {
853            for row in 0..batch.len() {
854                let value = batch.value_at(row);
855                accumulator.update(std::slice::from_ref(&value))?;
856            }
857        }
858        accumulator.finish()
859    }
860
861    fn a_vector_at_a_time(name: &str, returns: &LogicalType, batches: &[Vector]) -> Result<Value> {
862        let mut accumulator = Accumulator::new(name, returns)?;
863        for batch in batches {
864            accumulator.update_run(std::slice::from_ref(batch), batch.len())?;
865        }
866        accumulator.finish()
867    }
868
869    /// Both paths on the same batches, agreeing on the answer or agreeing on the complaint.
870    fn agrees(name: &str, returns: &LogicalType, batches: &[Vector], note: &str) {
871        let slow = row_at_a_time(name, returns, batches);
872        let fast = a_vector_at_a_time(name, returns, batches);
873        match (slow, fast) {
874            (Ok(slow), Ok(fast)) => assert_eq!(slow, fast, "{note}"),
875            (Err(slow), Err(fast)) => {
876                assert_eq!(slow.message(), fast.message(), "{note}");
877            }
878            (slow, fast) => {
879                panic!(
880                    "{note}: one path answered and the other did not, {slow:?} against {fast:?}"
881                );
882            }
883        }
884    }
885
886    struct Rng(u64);
887
888    impl Rng {
889        fn next(&mut self) -> u64 {
890            self.0 ^= self.0 << 13;
891            self.0 ^= self.0 >> 7;
892            self.0 ^= self.0 << 17;
893            self.0
894        }
895    }
896
897    /// A number small enough to be legal in every type below, so that the property test is about
898    /// the loops rather than about which types happen to hold which ranges.
899    fn small(rng: &mut Rng) -> i64 {
900        (rng.next() % 201) as i64 - 100
901    }
902
903    fn sample(ty: &LogicalType, rng: &mut Rng) -> Value {
904        let number = small(rng);
905        let positive = number.unsigned_abs();
906        match *ty {
907            LogicalType::TinyInt => Value::TinyInt(number as i8),
908            LogicalType::SmallInt => Value::SmallInt(number as i16),
909            LogicalType::Integer => Value::Integer(number as i32),
910            LogicalType::BigInt => Value::BigInt(number),
911            LogicalType::HugeInt => Value::HugeInt(i128::from(number)),
912            LogicalType::UTinyInt => Value::UTinyInt(positive as u8),
913            LogicalType::USmallInt => Value::USmallInt(positive as u16),
914            LogicalType::UInteger => Value::UInteger(positive as u32),
915            LogicalType::UBigInt => Value::UBigInt(positive),
916            LogicalType::Float => Value::Float(number as f32 / 8.0),
917            LogicalType::Double => Value::Double(number as f64 / 8.0),
918            LogicalType::Decimal { width, scale } => {
919                Value::Decimal { unscaled: i128::from(number) * 7, width, scale }
920            }
921            LogicalType::Varchar => Value::Varchar(format!("w{number}")),
922            _ => panic!("no sample for {ty}"),
923        }
924    }
925
926    fn flat(ty: &LogicalType, rows: usize, nulls: usize, rng: &mut Rng) -> Vector {
927        let values: Vec<Value> = (0..rows)
928            .map(
929                |index| {
930                    if nulls > 0 && index % nulls == 0 { Value::Null } else { sample(ty, rng) }
931                },
932            )
933            .collect();
934        Vector::from_values(ty.clone(), &values).expect("a vector of this type")
935    }
936
937    /// What the declared return type is for an aggregate over a column of this type.
938    fn returns_of(name: &str, ty: &LogicalType) -> LogicalType {
939        match name {
940            "count" | "count_star" => LogicalType::BigInt,
941            "avg" => LogicalType::Double,
942            "min" | "max" => ty.clone(),
943            _ => match *ty {
944                LogicalType::Decimal { scale, .. } => {
945                    LogicalType::decimal(rudb_common::MAX_DECIMAL_WIDTH, scale)
946                        .expect("the widest decimal at this scale is legal")
947                }
948                LogicalType::Float | LogicalType::Double => LogicalType::Double,
949                _ => LogicalType::HugeInt,
950            },
951        }
952    }
953
954    /// Every aggregate over every type this crate knows, in both forms that have a loop and at
955    /// three null densities, against the loop the loops replaced.
956    #[test]
957    fn every_aggregate_over_every_type_agrees_with_the_row_at_a_time_path() {
958        let mut rng = Rng(0x5eed_ca11_ab1e_0003);
959        let types = [
960            LogicalType::TinyInt,
961            LogicalType::SmallInt,
962            LogicalType::Integer,
963            LogicalType::BigInt,
964            LogicalType::HugeInt,
965            LogicalType::UTinyInt,
966            LogicalType::USmallInt,
967            LogicalType::UInteger,
968            LogicalType::UBigInt,
969            LogicalType::Float,
970            LogicalType::Double,
971            LogicalType::decimal(9, 2).expect("a legal decimal"),
972            LogicalType::decimal(18, 4).expect("a legal decimal"),
973            LogicalType::decimal(30, 6).expect("a legal decimal"),
974            LogicalType::Varchar,
975        ];
976        for ty in &types {
977            for name in ["count_star", "count", "sum", "avg", "min", "max"] {
978                let returns = returns_of(name, ty);
979                for nulls in [0_usize, 4, 1] {
980                    // Two batches rather than one, because a running total that is restarted at
981                    // every vector is right on one vector and wrong on the query.
982                    let first = flat(ty, 97, nulls, &mut rng);
983                    let second = flat(ty, 64, nulls, &mut rng);
984                    let note = format!("{name} over {ty}, flat, one null in {nulls}");
985                    agrees(name, &returns, &[first.clone(), second.clone()], &note);
986                    let codes: Vec<u32> = (0..97).map(|index| (index % 13) as u32).collect();
987                    let coded = Vector::dictionary(codes, first).expect("codes are in range");
988                    let note = format!("{name} over {ty}, dictionary, one null in {nulls}");
989                    agrees(name, &returns, &[coded, second], &note);
990                }
991            }
992        }
993    }
994
995    #[test]
996    fn a_sum_of_numbers_stays_off_the_row_at_a_time_path_and_a_sum_of_strings_does_not() {
997        fallback::reset();
998        let numbers = Vector::from_values(
999            LogicalType::Integer,
1000            &[Value::Integer(1), Value::Integer(2), Value::Integer(3)],
1001        )
1002        .expect("a vector of integers");
1003        let mut summing =
1004            Accumulator::new("sum", &LogicalType::HugeInt).expect("a known aggregate");
1005        summing.update_run(std::slice::from_ref(&numbers), 3).expect("sums");
1006        assert_eq!(summing.finish().expect("finishes"), Value::HugeInt(6));
1007        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);
1008
1009        let words = Vector::from_values(
1010            LogicalType::Varchar,
1011            &[Value::Varchar("a".into()), Value::Null, Value::Varchar("b".into())],
1012        )
1013        .expect("a vector of strings");
1014        let mut counting = Accumulator::new("count", &LogicalType::BigInt).expect("a known one");
1015        counting.update_run(std::slice::from_ref(&words), 3).expect("counts");
1016        assert_eq!(counting.finish().expect("finishes"), Value::BigInt(2));
1017        // A count reads the mask, so a type with no loop of its own is still not a fall through.
1018        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 0);
1019
1020        let mut wrong = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known aggregate");
1021        let error =
1022            wrong.update_run(std::slice::from_ref(&words), 3).expect_err("cannot sum those");
1023        assert!(error.message().contains("summing a"), "{error}");
1024        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 1);
1025        fallback::reset();
1026    }
1027
1028    /// The reason `update_run` carries the running total into the loop rather than totalling the
1029    /// vector on its own and adding the two at the end.
1030    #[test]
1031    fn a_floating_point_sum_carries_the_running_total_into_the_next_vector() {
1032        let first =
1033            Vector::from_values(LogicalType::Double, &[Value::Double(1.0e16)]).expect("a vector");
1034        let second = Vector::from_values(LogicalType::Double, &vec![Value::Double(1.0); 8])
1035            .expect("a vector");
1036        let batches = [first, second];
1037        let slow = row_at_a_time("sum", &LogicalType::Double, &batches).expect("sums");
1038        let fast = a_vector_at_a_time("sum", &LogicalType::Double, &batches).expect("sums");
1039        assert_eq!(slow, fast);
1040        // One at a time, every one of those eight disappears into the rounding. Eight at once does
1041        // not, which is what makes this a case worth having a test for.
1042        assert_eq!(slow, Value::Double(1.0e16));
1043        assert_ne!(1.0e16 + 8.0, 1.0e16);
1044    }
1045
1046    /// The wrong answer that needs a dictionary, a null and one specific code to reproduce.
1047    #[test]
1048    fn a_null_behind_a_dictionary_code_is_skipped_by_every_aggregate() {
1049        let values = Vector::from_values(
1050            LogicalType::Integer,
1051            &[Value::Null, Value::Integer(5), Value::Integer(9)],
1052        )
1053        .expect("a vector of integers");
1054        let coded = Vector::dictionary(vec![0, 1, 0, 2, 0], values).expect("codes are in range");
1055        let batch = std::slice::from_ref(&coded);
1056        assert_eq!(
1057            a_vector_at_a_time("count", &LogicalType::BigInt, batch).expect("counts"),
1058            Value::BigInt(2)
1059        );
1060        assert_eq!(
1061            a_vector_at_a_time("sum", &LogicalType::HugeInt, batch).expect("sums"),
1062            Value::HugeInt(14)
1063        );
1064        assert_eq!(
1065            a_vector_at_a_time("min", &LogicalType::Integer, batch).expect("finds one"),
1066            Value::Integer(5)
1067        );
1068    }
1069
1070    /// Why the whole sum stops at sixty four bits: at a hundred and twenty eight the total of one
1071    /// vector can overflow on its own, and the overflow is the answer rather than a detail.
1072    #[test]
1073    fn a_total_of_hugeints_goes_the_row_at_a_time_way_and_still_overflows() {
1074        fallback::reset();
1075        let rows = vec![Value::HugeInt(i128::MAX); 2];
1076        let vector = Vector::from_values(LogicalType::HugeInt, &rows).expect("a vector");
1077        let mut accumulator = Accumulator::new("sum", &LogicalType::HugeInt).expect("a known one");
1078        let error =
1079            accumulator.update_run(std::slice::from_ref(&vector), 2).expect_err("overflows");
1080        assert!(error.message().contains("Overflow in the running total"), "{error}");
1081        assert_eq!(fallback::count(Kernel::Aggregate, Form::Flat, Form::Flat), 1);
1082        fallback::reset();
1083    }
1084}