fn build_collation<G>(
    arrangements: Vec<(ReductionType, Arrangement<G, Row>)>,
    aggregate_types: Vec<ReductionType>,
    scope: &mut G,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let mut to_concat = vec![];

    // First, lets collect all results into a single collection.
    for (reduction_type, arrangement) in arrangements.into_iter() {
        let collection =
            arrangement.as_collection(move |key, val| (key.clone(), (reduction_type, val.clone())));
        to_concat.push(collection);
    }

    use differential_dataflow::collection::concatenate;
    concatenate(scope, to_concat)
        .reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceCollation", {
            let mut row_packer = Row::default();
            move |_key, input, output| {
                // The inputs are pairs of a reduction type, and a row consisting of densely packed fused
                // aggregate values.
                // We need to reconstitute the final value by:
                // 1. Extracting out the fused rows by type
                // 2. For each aggregate, figure out what type it is, and grab the relevant value
                //    from the corresponding fused row.
                // 3. Stitch all the values together into one row.

                let mut accumulable = DatumList::empty().iter();
                let mut hierarchical = DatumList::empty().iter();
                let mut basic = DatumList::empty().iter();

                // We expect not to have any negative multiplicities, but are not 100% sure it will
                // never happen so for now just log an error if it does.
                for (val, cnt) in input.iter() {
                    soft_assert_or_log!(
                        *cnt >= 0,
                        "[customer-data] Negative accumulation in ReduceCollation: {:?} with count {:?}",
                        val, cnt,
                    );
                }

                for ((reduction_type, row), _) in input.iter() {
                    match reduction_type {
                        ReductionType::Accumulable => {
                            accumulable = row.iter();
                        }
                        ReductionType::Hierarchical => {
                            hierarchical = row.iter();
                        }
                        ReductionType::Basic => {
                            basic = row.iter();
                        }
                    }
                }

                // Merge results into the order they were asked for.
                for typ in aggregate_types.iter() {
                    match typ {
                        ReductionType::Accumulable => {
                            row_packer.push(accumulable.next().unwrap())
                        }
                        ReductionType::Hierarchical => {
                            row_packer.push(hierarchical.next().unwrap())
                        }
                        ReductionType::Basic => {
                            row_packer.push(basic.next().unwrap())
                        }
                    }
                }
                output.push((row_packer.finish_and_reuse(), 1));
            }
        })
}
/// Build the dataflow to combine arrangements containing results of different
/// aggregation types into a single arrangement.
///
/// This computes the same thing as a join on the group key followed by shuffling
/// the values into the correct order. This implementation assumes that all input
/// arrangements present values in a way that respects the desired output order,
/// so we can do a linear merge to form the output.
Build the dataflow to combine arrangements containing results of different
aggregation types into a single arrangement.

This computes the same thing as a join on the group key followed by shuffling
the values into the correct order. This implementation assumes that all input
arrangements present values in a way that respects the desired output order,
so we can do a linear merge to form the output.
Build the dataflow to combine arrangements containing results of different aggregation types into a single arrangement . This computes the same thing as a join on the group key followed by shuffling the values into the correct order . This implementation assumes that all input arrangements present values in a way that respects the desired output order so we can do a linear merge to form the output .
fn build_collation<G>(
    arrangements: Vec<(ReductionType, Arrangement<G, Row>)>,
    aggregate_types: Vec<ReductionType>,
    scope: &mut G,
) -> Arrangement<G, Row>
where
    G: Scope,
    G::Timestamp: Lattice,
{
    let mut to_concat = vec![];
    // First, lets collect all results into a single collection.
    for (reduction_type, arrangement) in arrangements.into_iter() {
        let collection =
            arrangement.as_collection(move |key, val| (key.clone(), (reduction_type, val.clone())));
        to_concat.push(collection);
    }
    use differential_dataflow::collection::concatenate;
    concatenate(scope, to_concat)
        .reduce_abelian::<_, RowSpine<_, _, _, _>>("ReduceCollation", {
            let mut row_packer = Row::default();
            move |_key, input, output| {
                // The inputs are pairs of a reduction type, and a row consisting of densely packed fused
                // aggregate values.
                // We need to reconstitute the final value by:
                // 1. Extracting out the fused rows by type
                // 2. For each aggregate, figure out what type it is, and grab the relevant value
                //    from the corresponding fused row.
                // 3. Stitch all the values together into one row.
                let mut accumulable = DatumList::empty().iter();
                let mut hierarchical = DatumList::empty().iter();
                let mut basic = DatumList::empty().iter();
                // We expect not to have any negative multiplicities, but are not 100% sure it will
                // never happen so for now just log an error if it does.
                for (val, cnt) in input.iter() {
                    soft_assert_or_log!(
                        *cnt >= 0,
                        "[customer-data] Negative accumulation in ReduceCollation: {:?} with count {:?}",
                        val, cnt,
                    );
                }
                for ((reduction_type, row), _) in input.iter() {
                    match reduction_type {
                        ReductionType::Accumulable => {
                            accumulable = row.iter();
                        }
                        ReductionType::Hierarchical => {
                            hierarchical = row.iter();
                        }
                        ReductionType::Basic => {
                            basic = row.iter();
                        }
                    }
                }
                // Merge results into the order they were asked for.
                for typ in aggregate_types.iter() {
                    match typ {
                        ReductionType::Accumulable => {
                            row_packer.push(accumulable.next().unwrap())
                        }
                        ReductionType::Hierarchical => {
                            row_packer.push(hierarchical.next().unwrap())
                        }
                        ReductionType::Basic => {
                            row_packer.push(basic.next().unwrap())
                        }
                    }
                }
                output.push((row_packer.finish_and_reuse(), 1));
            }
        })
}
fn build_collation < G > ( arrangements : Vec < ( ReductionType , Arrangement < G , Row > ) > , aggregate_types : Vec < ReductionType > , scope : & mut G , ) -> Arrangement < G , Row > where G : Scope , G :: Timestamp : Lattice , { let mut to_concat = vec ! [ ] ; // First, lets collect all results into a single collection. for ( reduction_type , arrangement ) in arrangements . into_iter ( ) { let collection = arrangement . as_collection ( move | key , val | ( key . clone ( ) , ( reduction_type , val . clone ( ) ) ) ) ; to_concat . push ( collection ) ; } use differential_dataflow :: collection :: concatenate ; concatenate ( scope , to_concat ) . reduce_abelian :: < _ , RowSpine < _ , _ , _ , _ > > ( " " , { let mut row_packer = Row :: default ( ) ; move | _key , input , output | { // The inputs are pairs of a reduction type, and a row consisting of densely packed fused // aggregate values. // We need to reconstitute the final value by: // 1. Extracting out the fused rows by type // 2. For each aggregate, figure out what type it is, and grab the relevant value //    from the corresponding fused row. // 3. Stitch all the values together into one row. let mut accumulable = DatumList :: empty ( ) . iter ( ) ; let mut hierarchical = DatumList :: empty ( ) . iter ( ) ; let mut basic = DatumList :: empty ( ) . iter ( ) ; // We expect not to have any negative multiplicities, but are not 100% sure it will // never happen so for now just log an error if it does. for ( val , cnt ) in input . iter ( ) { soft_assert_or_log ! ( cnt 0 " " val cnt ) ; } for ( ( reduction_type , row ) , _ ) in input . iter ( ) { match reduction_type { ReductionType :: Accumulable => { accumulable = row . iter ( ) ; } ReductionType :: Hierarchical => { hierarchical = row . iter ( ) ; } ReductionType :: Basic => { basic = row . iter ( ) ; } } } // Merge results into the order they were asked for. for typ in aggregate_types . iter ( ) { match typ { ReductionType :: Accumulable => { row_packer . push ( accumulable . next ( ) . unwrap ( ) ) } ReductionType :: Hierarchical => { row_packer . push ( hierarchical . next ( ) . unwrap ( ) ) } ReductionType :: Basic => { row_packer . push ( basic . next ( ) . unwrap ( ) ) } } } output . push ( ( row_packer . finish_and_reuse ( ) , 1 ) ) ; } } ) }
Build the dataflow to combine arrangements containing results of different
aggregation types into a single arrangement.
Build the dataflow to combine arrangements containing results of different aggregation types into a single arrangement .
arrangements
Vec<(ReductionType, Arrangement<G, Row>)>
aggregate_types
Vec<ReductionType>
scope
&mut G
arrangements
Vec<(ReductionType, Arrangement<G, Row>)>
aggregate_types
Vec<ReductionType>
scope
&mut G