fn build_bucketed_stage<G>(
    input: Collection<G, ((Row, u64), Vec<Row>)>,
    aggrs: Vec<AggregateFunc>,
    buckets: u64,
) -> Collection<G, ((Row, u64), Vec<Row>)>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let input = input.map(move |((key, hash), values)| ((key, hash % buckets), values));

    let negated_output = input
        .reduce_named("MinsMaxesHierarchical", {
            move |key, source, target| {
                // Should negative accumulations reach us, we should loudly complain.
                if source.iter().any(|(_val, cnt)| cnt <= &0) {
                    for (val, cnt) in source.iter() {
                        // XXX: This reports user data, which we perhaps should not do!
                        soft_assert_or_log!(
                            *cnt > 0,
                            "[customer-data] Non-positive accumulation in MinsMaxesHierarchical: key: {:?}\tvalue: {:?}\tcount: {:?}",
                            key, val, cnt,
                        );
                    }
                } else {
                    let mut output = Vec::with_capacity(aggrs.len());
                    for (aggr_index, func) in aggrs.iter().enumerate() {
                        let iter = source.iter().map(|(values, _cnt)| values[aggr_index].iter().next().unwrap());
                        output.push(Row::pack_slice(&[func.eval(iter, &RowArena::new())]));
                    }
                    // We only want to arrange the parts of the input that are not part of the output.
                    // More specifically, we want to arrange it so that `input.concat(&output.negate())`
                    // gives us the intended value of this aggregate function. Also we assume that regardless
                    // of the multiplicity of the final result in the input, we only want to have one copy
                    // in the output.

                    target.push((output, -1));
                    target.extend(source.iter().map(|(values, cnt)| ((*values).clone(), *cnt)));
                }
            }
        });

    negated_output.negate().concat(&input).consolidate()
}
/// Build the dataflow for one stage of a reduction tree for multiple hierarchical
/// aggregates.
///
/// `buckets` indicates the number of buckets in this stage. We do some non
/// obvious trickery here to limit the memory usage per layer by internally
/// holding only the elements that were rejected by this stage. However, the
/// output collection maintains the `((key, bucket), (passing value)` for this
/// stage.
Build the dataflow for one stage of a reduction tree for multiple hierarchical
aggregates.

`buckets` indicates the number of buckets in this stage. We do some non
obvious trickery here to limit the memory usage per layer by internally
holding only the elements that were rejected by this stage. However, the
output collection maintains the `((key, bucket), (passing value)` for this
stage.
Build the dataflow for one stage of a reduction tree for multiple hierarchical aggregates . ` buckets ` indicates the number of buckets in this stage . We do some non obvious trickery here to limit the memory usage per layer by internally holding only the elements that were rejected by this stage . However the output collection maintains the ` (( key bucket ) ( passing value ) ` for this stage .
fn build_bucketed_stage<G>(
    input: Collection<G, ((Row, u64), Vec<Row>)>,
    aggrs: Vec<AggregateFunc>,
    buckets: u64,
) -> Collection<G, ((Row, u64), Vec<Row>)>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let input = input.map(move |((key, hash), values)| ((key, hash % buckets), values));
    let negated_output = input
        .reduce_named("MinsMaxesHierarchical", {
            move |key, source, target| {
                // Should negative accumulations reach us, we should loudly complain.
                if source.iter().any(|(_val, cnt)| cnt <= &0) {
                    for (val, cnt) in source.iter() {
                        // XXX: This reports user data, which we perhaps should not do!
                        soft_assert_or_log!(
                            *cnt > 0,
                            "[customer-data] Non-positive accumulation in MinsMaxesHierarchical: key: {:?}\tvalue: {:?}\tcount: {:?}",
                            key, val, cnt,
                        );
                    }
                } else {
                    let mut output = Vec::with_capacity(aggrs.len());
                    for (aggr_index, func) in aggrs.iter().enumerate() {
                        let iter = source.iter().map(|(values, _cnt)| values[aggr_index].iter().next().unwrap());
                        output.push(Row::pack_slice(&[func.eval(iter, &RowArena::new())]));
                    }
                    // We only want to arrange the parts of the input that are not part of the output.
                    // More specifically, we want to arrange it so that `input.concat(&output.negate())`
                    // gives us the intended value of this aggregate function. Also we assume that regardless
                    // of the multiplicity of the final result in the input, we only want to have one copy
                    // in the output.
                    target.push((output, -1));
                    target.extend(source.iter().map(|(values, cnt)| ((*values).clone(), *cnt)));
                }
            }
        });
    negated_output.negate().concat(&input).consolidate()
}
fn build_bucketed_stage < G > ( input : Collection < G , ( ( Row , u64 ) , Vec < Row > ) > , aggrs : Vec < AggregateFunc > , buckets : u64 , ) -> Collection < G , ( ( Row , u64 ) , Vec < Row > ) > where G : Scope , G :: Timestamp : Lattice , { let input = input . map ( move | ( ( key , hash ) , values ) | ( ( key , hash % buckets ) , values ) ) ; let negated_output = input . reduce_named ( " " , { move | key , source , target | { // Should negative accumulations reach us, we should loudly complain. if source . iter ( ) . any ( | ( _val , cnt ) | cnt <= & 0 ) { for ( val , cnt ) in source . iter ( ) { // XXX: This reports user data, which we perhaps should not do! soft_assert_or_log ! ( cnt 0 " \t \t " key val cnt ) ; } } else { let mut output = Vec :: with_capacity ( aggrs . len ( ) ) ; for ( aggr_index , func ) in aggrs . iter ( ) . enumerate ( ) { let iter = source . iter ( ) . map ( | ( values , _cnt ) | values [ aggr_index ] . iter ( ) . next ( ) . unwrap ( ) ) ; output . push ( Row :: pack_slice ( & [ func . eval ( iter , & RowArena :: new ( ) ) ] ) ) ; } // We only want to arrange the parts of the input that are not part of the output. // More specifically, we want to arrange it so that `input.concat(&output.negate())` // gives us the intended value of this aggregate function. Also we assume that regardless // of the multiplicity of the final result in the input, we only want to have one copy // in the output. target . push ( ( output , - 1 ) ) ; target . extend ( source . iter ( ) . map ( | ( values , cnt ) | ( ( * values ) . clone ( ) , * cnt ) ) ) ; } } } ) ; negated_output . negate ( ) . concat ( & input ) . consolidate ( ) }
Build the dataflow for one stage of a reduction tree for multiple hierarchical
aggregates.
Build the dataflow for one stage of a reduction tree for multiple hierarchical aggregates .
input
Collection<G, ((Row, u64), Vec<Row>)>
aggrs
Vec<AggregateFunc>
buckets
u64
input
Collection<G, ((Row, u64), Vec<Row>)>
aggrs
Vec<AggregateFunc>
buckets
u64