Method eval

src/expr/src/relation/func.rs:2080–2109

Evaluates the aggregate over an iterator of `(datum, diff)` pairs. Each aggregate consumes the multiplicity (`diff`) in whatever way is most efficient: `count` sums the diffs, multiplicity-insensitive aggregates (see `AggregateFunc::ignores_multiplicity`) ignore them, and everything else expands each datum into `diff` copies (see `expand_counts`).

(&self, datums: I, temp_storage: &'a RowArena)

    /// (see `AggregateFunc::ignores_multiplicity`) ignore them, and everything
    /// else expands each datum into `diff` copies (see `expand_counts`).
    pub fn eval<'a, I>(&self, datums: I, temp_storage: &'a RowArena) -> Datum<'a>
    where
        I: IntoIterator<Item = (Datum<'a>, Diff)>,
    {
        // Accumulable aggregates consume multiplicity directly rather than
        // expanding each `(datum, diff)` into `diff` copies. The cases handled
        // here mirror the dataflow's accumulable reduction (`build_accumulable`
        // in `mz_compute::render::reduce`) so that constant folding produces the
        // same result the dataflow would. Signed integer sums are folded here;
        // unsigned sums are not, because their negative-accumulation case is a
        // query error in the dataflow that this `Datum`-returning path cannot
        // signal. Floats and numerics use bespoke fixed-point/wide-decimal
        // accumulators in the dataflow that `expand_counts` does not reproduce.
        match self {
            AggregateFunc::Count => count(datums),
            AggregateFunc::SumInt16 | AggregateFunc::SumInt32 => {
                // `finalize_accum` narrows these to `i64` with wrapping.
                sum_signed_int_counted(datums, |accum| {
                    #[allow(clippy::as_conversions)]
                    let narrowed = accum as i64;
                    Datum::Int64(narrowed)
                })
            }
            AggregateFunc::SumInt64 => sum_signed_int_counted(datums, Datum::from),
            _ if self.ignores_multiplicity() => {
                self.eval_datums(datums.into_iter().map(|(datum, _diff)| datum), temp_storage)
            }
            _ => self.eval_datums(expand_counts(datums), temp_storage),
        }
    }

    /// Evaluates the aggregate over a flat iterator of datums, ignoring multiplicity.
    fn eval_datums<'a, I>(&self, datums: I, temp_storage: &'a RowArena) -> Datum<'a>