Method build_basic_aggregate

src/compute/src/render/reduce.rs:481–773

Build the dataflow to compute a single basic aggregation. This method also applies distinctness if required.

(
        &self,
        input: VecCollection<'s, T, (Row, Row), Diff>,
        index: usize,
        aggr: &AggregateExpr,
        validating: bool,
        key_arity: usize,
        mfp_after: Optio

    ///
    /// This method also applies distinctness if required.
    fn build_basic_aggregate<'s>(
        &self,
        input: VecCollection<'s, T, (Row, Row), Diff>,
        index: usize,
        aggr: &AggregateExpr,
        validating: bool,
        key_arity: usize,
        mfp_after: Option<SafeMfpPlan>,
        fused_unnest_list: bool,
    ) -> (
        RowRowArrangement<'s, T>,
        Option<VecCollection<'s, T, DataflowErrorSer, Diff>>,
    ) {
        let AggregateExpr {
            func,
            expr: _,
            distinct,
        } = aggr.clone();

        // Extract the value we were asked to aggregate over.
        let mut partial = input.map(move |(key, row)| {
            let mut row_builder = SharedRow::get();
            let value = row.iter().nth(index).unwrap();
            row_builder.packer().push(value);
            (key, row_builder.clone())
        });

        let mut err_output = None;

        // If `distinct` is set, we restrict ourselves to the distinct `(key, val)`.
        if distinct {
            // We map `(Row, Row)` to `Row` to take advantage of `Row*Spine` types.
            let pairer = Pairer::new(key_arity);
            let keyed = partial.map(move |(key, val)| pairer.merge(&key, &val));
            if validating {
                let (oks, errs) = self
                    .build_reduce_inaccumulable_distinct::<
                        RowValBuilder<Result<(), String>, _, _>,
                        RowValSpine<Result<(), String>, _, _>,
                    >(keyed, None)
                    .as_collection(|k, v| {
                        (
                            k.to_row(),
                            v.as_ref()
                                .map(|&()| ())
                                .map_err(|m| m.as_str().into()),
                        )
                    })
                    .map_fallible::<
                        CapacityContainerBuilder<_>,
                        CapacityContainerBuilder<_>,
                        _,
                        _,
                        _,
                    >(
                        "Demux Errors",
                        move |(key_val, result)| match result {
                            Ok(()) => Ok(pairer.split(&key_val)),