fn build_monotonic<G>(
    collection: Collection<G, (Row, Row)>,
    MonotonicPlan { aggr_funcs, skips }: MonotonicPlan,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    // Gather the relevant values into a vec of rows ordered by aggregation_index
    let mut packer = Row::default();
    let collection = collection.map(move |(key, row)| {
        let mut values = Vec::with_capacity(skips.len());
        let mut row_iter = row.iter();
        for skip in skips.iter() {
            packer.push((&mut row_iter).nth(*skip).unwrap());
            values.push(packer.finish_and_reuse());
        }

        (key, values)
    });

    // We can place our rows directly into the diff field, and only keep the
    // relevant one corresponding to evaluating our aggregate, instead of having
    // to do a hierarchical reduction.
    use timely::dataflow::operators::Map;

    // We arrange the inputs ourself to force it into a leaner structure because we know we
    // won't care about values.
    let partial = collection
        .consolidate()
        .inner
        .map(move |((key, values), time, diff)| {
            assert!(diff > 0);
            let mut output = Vec::new();
            for (row, func) in values.into_iter().zip(aggr_funcs.iter()) {
                output.push(monoids::get_monoid(row, func).expect(
                    "hierarchical aggregations are expected to have monoid implementations",
                ));
            }

            (key, time, output)
        })
        .as_collection();
    partial
        .arrange_by_self()
        .reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceMonotonic", {
            let mut row_packer = Row::default();
            move |_key, input, output| {
                let accum = &input[0].1;
                for monoid in accum.iter() {
                    match monoid {
                        monoids::ReductionMonoid::Min(row) => row_packer.extend(row.iter()),
                        monoids::ReductionMonoid::Max(row) => row_packer.extend(row.iter()),
                    }
                }
                output.push((row_packer.finish_and_reuse(), 1));
            }
        })
}
/// Build the dataflow to compute and arrange multiple hierarchical aggregations
/// on monotonic inputs.
Build the dataflow to compute and arrange multiple hierarchical aggregations
on monotonic inputs.
Build the dataflow to compute and arrange multiple hierarchical aggregations on monotonic inputs .
fn build_monotonic<G>(
    collection: Collection<G, (Row, Row)>,
    MonotonicPlan { aggr_funcs, skips }: MonotonicPlan,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    // Gather the relevant values into a vec of rows ordered by aggregation_index
    let mut packer = Row::default();
    let collection = collection.map(move |(key, row)| {
        let mut values = Vec::with_capacity(skips.len());
        let mut row_iter = row.iter();
        for skip in skips.iter() {
            packer.push((&mut row_iter).nth(*skip).unwrap());
            values.push(packer.finish_and_reuse());
        }
        (key, values)
    });
    // We can place our rows directly into the diff field, and only keep the
    // relevant one corresponding to evaluating our aggregate, instead of having
    // to do a hierarchical reduction.
    use timely::dataflow::operators::Map;
    // We arrange the inputs ourself to force it into a leaner structure because we know we
    // won't care about values.
    let partial = collection
        .consolidate()
        .inner
        .map(move |((key, values), time, diff)| {
            assert!(diff > 0);
            let mut output = Vec::new();
            for (row, func) in values.into_iter().zip(aggr_funcs.iter()) {
                output.push(monoids::get_monoid(row, func).expect(
                    "hierarchical aggregations are expected to have monoid implementations",
                ));
            }
            (key, time, output)
        })
        .as_collection();
    partial
        .arrange_by_self()
        .reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceMonotonic", {
            let mut row_packer = Row::default();
            move |_key, input, output| {
                let accum = &input[0].1;
                for monoid in accum.iter() {
                    match monoid {
                        monoids::ReductionMonoid::Min(row) => row_packer.extend(row.iter()),
                        monoids::ReductionMonoid::Max(row) => row_packer.extend(row.iter()),
                    }
                }
                output.push((row_packer.finish_and_reuse(), 1));
            }
        })
}
fn build_monotonic < G > ( collection : Collection < G , ( Row , Row ) > , MonotonicPlan { aggr_funcs , skips } : MonotonicPlan , ) -> Arrangement < G , Row > where G : Scope , G :: Timestamp : Lattice , { // Gather the relevant values into a vec of rows ordered by aggregation_index let mut packer = Row :: default ( ) ; let collection = collection . map ( move | ( key , row ) | { let mut values = Vec :: with_capacity ( skips . len ( ) ) ; let mut row_iter = row . iter ( ) ; for skip in skips . iter ( ) { packer . push ( ( & mut row_iter ) . nth ( * skip ) . unwrap ( ) ) ; values . push ( packer . finish_and_reuse ( ) ) ; } ( key , values ) } ) ; // We can place our rows directly into the diff field, and only keep the // relevant one corresponding to evaluating our aggregate, instead of having // to do a hierarchical reduction. use timely :: dataflow :: operators :: Map ; // We arrange the inputs ourself to force it into a leaner structure because we know we // won't care about values. let partial = collection . consolidate ( ) . inner . map ( move | ( ( key , values ) , time , diff ) | { assert ! ( diff 0 ) ; let mut output = Vec :: new ( ) ; for ( row , func ) in values . into_iter ( ) . zip ( aggr_funcs . iter ( ) ) { output . push ( monoids :: get_monoid ( row , func ) . expect ( " " , ) ) ; } ( key , time , output ) } ) . as_collection ( ) ; partial . arrange_by_self ( ) . reduce_abelian :: < _ , RowSpine < _ , _ , _ , _ > > ( " " , { let mut row_packer = Row :: default ( ) ; move | _key , input , output | { let accum = & input [ 0 ] . 1 ; for monoid in accum . iter ( ) { match monoid { monoids :: ReductionMonoid :: Min ( row ) => row_packer . extend ( row . iter ( ) ) , monoids :: ReductionMonoid :: Max ( row ) => row_packer . extend ( row . iter ( ) ) , } } output . push ( ( row_packer . finish_and_reuse ( ) , 1 ) ) ; } } ) }
Build the dataflow to compute and arrange multiple hierarchical aggregations
on monotonic inputs.
Build the dataflow to compute and arrange multiple hierarchical aggregations on monotonic inputs .
collection
MonotonicPlan
collection
MonotonicPlan