fn build_basic_aggregate<G>(
    input: Collection<G, (Row, Row)>,
    index: usize,
    aggr: &AggregateExpr,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let AggregateExpr {
        func,
        expr: _,
        distinct,
    } = aggr.clone();

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

    // If `distinct` is set, we restrict ourselves to the distinct `(key, val)`.
    if distinct {
        partial = partial.distinct();
    }

    partial.reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceInaccumulable", {
        let mut row_packer = Row::default();
        move |_key, source, target| {
            // Negative counts would be surprising, but until we are 100% certain we wont
            // see them, we should report when we do. We may want to bake even more info
            // in here in the future.
            if source.iter().any(|(_val, cnt)| cnt < &0) {
                // XXX: This reports user data, which we perhaps should not do!
                for (val, cnt) in source.iter() {
                    soft_assert_or_log!(
                        *cnt >= 0,
                        "[customer-data] Negative accumulation in ReduceInaccumulable: {:?} with count {:?}",
                        val, cnt,
                    );
                }
            } else {
                // We respect the multiplicity here (unlike in hierarchical aggregation)
                // because we don't know that the aggregation method is not sensitive
                // to the number of records.
                let iter = source.iter().flat_map(|(v, w)| {
                    std::iter::repeat(v.iter().next().unwrap()).take(*w as usize)
                });
                row_packer.push(func.eval(iter, &RowArena::new()));
                target.push((row_packer.finish_and_reuse(), 1));
            }
        }
    })
}
/// Build the dataflow to compute a single basic aggregation.
///
/// This method also applies distinctness if required.
Build the dataflow to compute a single basic aggregation.
This method also applies distinctness if required.
Build the dataflow to compute a single basic aggregation . This method also applies distinctness if required .
fn build_basic_aggregate<G>(
    input: Collection<G, (Row, Row)>,
    index: usize,
    aggr: &AggregateExpr,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let AggregateExpr {
        func,
        expr: _,
        distinct,
    } = aggr.clone();
    // Extract the value we were asked to aggregate over.
    let mut packer = Row::default();
    let mut partial = input.map(move |(key, row)| {
        let value = row.iter().nth(index).unwrap();
        packer.push(value);
        (key, packer.finish_and_reuse())
    });
    // If `distinct` is set, we restrict ourselves to the distinct `(key, val)`.
    if distinct {
        partial = partial.distinct();
    }
    partial.reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceInaccumulable", {
        let mut row_packer = Row::default();
        move |_key, source, target| {
            // Negative counts would be surprising, but until we are 100% certain we wont
            // see them, we should report when we do. We may want to bake even more info
            // in here in the future.
            if source.iter().any(|(_val, cnt)| cnt < &0) {
                // XXX: This reports user data, which we perhaps should not do!
                for (val, cnt) in source.iter() {
                    soft_assert_or_log!(
                        *cnt >= 0,
                        "[customer-data] Negative accumulation in ReduceInaccumulable: {:?} with count {:?}",
                        val, cnt,
                    );
                }
            } else {
                // We respect the multiplicity here (unlike in hierarchical aggregation)
                // because we don't know that the aggregation method is not sensitive
                // to the number of records.
                let iter = source.iter().flat_map(|(v, w)| {
                    std::iter::repeat(v.iter().next().unwrap()).take(*w as usize)
                });
                row_packer.push(func.eval(iter, &RowArena::new()));
                target.push((row_packer.finish_and_reuse(), 1));
            }
        }
    })
}
fn build_basic_aggregate < G > ( input : Collection < G , ( Row , Row ) > , index : usize , aggr : & AggregateExpr , ) -> Arrangement < G , Row > where G : Scope , G :: Timestamp : Lattice , { let AggregateExpr { func , expr : _ , distinct , } = aggr . clone ( ) ; // Extract the value we were asked to aggregate over. let mut packer = Row :: default ( ) ; let mut partial = input . map ( move | ( key , row ) | { let value = row . iter ( ) . nth ( index ) . unwrap ( ) ; packer . push ( value ) ; ( key , packer . finish_and_reuse ( ) ) } ) ; // If `distinct` is set, we restrict ourselves to the distinct `(key, val)`. if distinct { partial = partial . distinct ( ) ; } partial . reduce_abelian :: < _ , RowSpine < _ , _ , _ , _ > > ( " " , { let mut row_packer = Row :: default ( ) ; move | _key , source , target | { // Negative counts would be surprising, but until we are 100% certain we wont // see them, we should report when we do. We may want to bake even more info // in here in the future. if source . iter ( ) . any ( | ( _val , cnt ) | cnt < & 0 ) { // XXX: This reports user data, which we perhaps should not do! for ( val , cnt ) in source . iter ( ) { soft_assert_or_log ! ( cnt 0 " " val cnt ) ; } } else { // We respect the multiplicity here (unlike in hierarchical aggregation) // because we don't know that the aggregation method is not sensitive // to the number of records. let iter = source . iter ( ) . flat_map ( | ( v , w ) | { std :: iter :: repeat ( v . iter ( ) . next ( ) . unwrap ( ) ) . take ( * w as usize ) } ) ; row_packer . push ( func . eval ( iter , & RowArena :: new ( ) ) ) ; target . push ( ( row_packer . finish_and_reuse ( ) , 1 ) ) ; } } } ) }
Build the dataflow to compute a single basic aggregation.
Build the dataflow to compute a single basic aggregation .
input
Collection<G, (Row, Row)>
index
usize
aggr
&AggregateExpr
input
Collection<G, (Row, Row)>
index
usize
aggr
&AggregateExpr