fn create_inner(
        typ: ReductionType,
        aggregates_list: Vec<(usize, AggregateExpr)>,
        monotonic: bool,
        expected_group_size: Option<usize>,
    ) -> Self {
        assert!(
            aggregates_list.len() > 0,
            "error: tried to render a reduce dataflow with no aggregates"
        );
        match typ {
            ReductionType::Accumulable => {
                let mut simple_aggrs = vec![];
                let mut distinct_aggrs = vec![];
                let full_aggrs: Vec<_> = aggregates_list
                    .iter()
                    .cloned()
                    .map(|(_, aggr)| aggr)
                    .collect();
                for (accumulable_index, (datum_index, aggr)) in
                    aggregates_list.into_iter().enumerate()
                {
                    // Accumulable aggregations need to do extra per-aggregate work
                    // for aggregations with the distinct bit set, so we'll separate
                    // those out now.
                    if aggr.distinct {
                        distinct_aggrs.push((accumulable_index, datum_index, aggr));
                    } else {
                        simple_aggrs.push((accumulable_index, datum_index, aggr));
                    };
                }
                ReducePlan::Accumulable(AccumulablePlan {
                    full_aggrs,
                    simple_aggrs,
                    distinct_aggrs,
                })
            }
            ReductionType::Hierarchical => {
                let aggr_funcs: Vec<_> = aggregates_list
                    .iter()
                    .cloned()
                    .map(|(_, aggr)| aggr.func)
                    .collect();
                let indexes: Vec<_> = aggregates_list
                    .into_iter()
                    .map(|(index, _)| index)
                    .collect();

                // We don't have random access over Rows so we can simplify the
                // task of grabbing the inputs we are aggregating over by
                // generating a list of "skips" an iterator over the Row needs
                // to do to get the desired indexes.
                let skips = convert_indexes_to_skips(indexes);
                if monotonic {
                    let monotonic = MonotonicPlan { aggr_funcs, skips };
                    ReducePlan::Hierarchical(HierarchicalPlan::Monotonic(monotonic))
                } else {
                    let mut buckets = vec![];
                    let mut current = 16;

                    // Plan for 4B records in the expected case if the user
                    // didn't specify a group size.
                    let limit = expected_group_size.unwrap_or(4_000_000_000);

                    // Distribute buckets in powers of 16, so that we can strike
                    // a balance between how many inputs each layer gets from
                    // the preceding layer, while also limiting the number of
                    // layers.
                    while current < limit {
                        buckets.push(current as u64);
                        current *= 16;
                    }

                    // We need to store the bucket numbers in decreasing order.
                    buckets.reverse();

                    let bucketed = BucketedPlan {
                        aggr_funcs,
                        skips,
                        buckets,
                    };

                    ReducePlan::Hierarchical(HierarchicalPlan::Bucketed(bucketed))
                }
            }
            ReductionType::Basic => {
                if aggregates_list.len() == 1 {
                    ReducePlan::Basic(BasicPlan::Single(
                        aggregates_list[0].0,
                        aggregates_list[0].1.clone(),
                    ))
                } else {
                    ReducePlan::Basic(BasicPlan::Multiple(aggregates_list))
                }
            }
        }
    }
/// Generate a plan for computing the specified type of aggregations.
///
/// This function assumes that all of the supplied aggregates are
/// actually of the correct reduction type, and are a subsequence
/// of the total list of requested aggregations.
Generate a plan for computing the specified type of aggregations.
This function assumes that all of the supplied aggregates are
actually of the correct reduction type, and are a subsequence
of the total list of requested aggregations.
Generate a plan for computing the specified type of aggregations . This function assumes that all of the supplied aggregates are actually of the correct reduction type and are a subsequence of the total list of requested aggregations .
fn create_inner(
        typ: ReductionType,
        aggregates_list: Vec<(usize, AggregateExpr)>,
        monotonic: bool,
        expected_group_size: Option<usize>,
    ) -> Self {
        assert!(
            aggregates_list.len() > 0,
            "error: tried to render a reduce dataflow with no aggregates"
        );
        match typ {
            ReductionType::Accumulable => {
                let mut simple_aggrs = vec![];
                let mut distinct_aggrs = vec![];
                let full_aggrs: Vec<_> = aggregates_list
                    .iter()
                    .cloned()
                    .map(|(_, aggr)| aggr)
                    .collect();
                for (accumulable_index, (datum_index, aggr)) in
                    aggregates_list.into_iter().enumerate()
                {
                    // Accumulable aggregations need to do extra per-aggregate work
                    // for aggregations with the distinct bit set, so we'll separate
                    // those out now.
                    if aggr.distinct {
                        distinct_aggrs.push((accumulable_index, datum_index, aggr));
                    } else {
                        simple_aggrs.push((accumulable_index, datum_index, aggr));
                    };
                }
                ReducePlan::Accumulable(AccumulablePlan {
                    full_aggrs,
                    simple_aggrs,
                    distinct_aggrs,
                })
            }
            ReductionType::Hierarchical => {
                let aggr_funcs: Vec<_> = aggregates_list
                    .iter()
                    .cloned()
                    .map(|(_, aggr)| aggr.func)
                    .collect();
                let indexes: Vec<_> = aggregates_list
                    .into_iter()
                    .map(|(index, _)| index)
                    .collect();
                // We don't have random access over Rows so we can simplify the
                // task of grabbing the inputs we are aggregating over by
                // generating a list of "skips" an iterator over the Row needs
                // to do to get the desired indexes.
                let skips = convert_indexes_to_skips(indexes);
                if monotonic {
                    let monotonic = MonotonicPlan { aggr_funcs, skips };
                    ReducePlan::Hierarchical(HierarchicalPlan::Monotonic(monotonic))
                } else {
                    let mut buckets = vec![];
                    let mut current = 16;
                    // Plan for 4B records in the expected case if the user
                    // didn't specify a group size.
                    let limit = expected_group_size.unwrap_or(4_000_000_000);
                    // Distribute buckets in powers of 16, so that we can strike
                    // a balance between how many inputs each layer gets from
                    // the preceding layer, while also limiting the number of
                    // layers.
                    while current < limit {
                        buckets.push(current as u64);
                        current *= 16;
                    }
                    // We need to store the bucket numbers in decreasing order.
                    buckets.reverse();
                    let bucketed = BucketedPlan {
                        aggr_funcs,
                        skips,
                        buckets,
                    };
                    ReducePlan::Hierarchical(HierarchicalPlan::Bucketed(bucketed))
                }
            }
            ReductionType::Basic => {
                if aggregates_list.len() == 1 {
                    ReducePlan::Basic(BasicPlan::Single(
                        aggregates_list[0].0,
                        aggregates_list[0].1.clone(),
                    ))
                } else {
                    ReducePlan::Basic(BasicPlan::Multiple(aggregates_list))
                }
            }
        }
    }
fn create_inner ( typ : ReductionType , aggregates_list : Vec < ( usize , AggregateExpr ) > , monotonic : bool , expected_group_size : Option < usize > , ) -> Self { assert ! ( aggregates_list len ( ) 0 " " ) ; match typ { ReductionType :: Accumulable => { let mut simple_aggrs = vec ! [ ] ; let mut distinct_aggrs = vec ! [ ] ; let full_aggrs : Vec < _ > = aggregates_list . iter ( ) . cloned ( ) . map ( | ( _ , aggr ) | aggr ) . collect ( ) ; for ( accumulable_index , ( datum_index , aggr ) ) in aggregates_list . into_iter ( ) . enumerate ( ) { // Accumulable aggregations need to do extra per-aggregate work // for aggregations with the distinct bit set, so we'll separate // those out now. if aggr . distinct { distinct_aggrs . push ( ( accumulable_index , datum_index , aggr ) ) ; } else { simple_aggrs . push ( ( accumulable_index , datum_index , aggr ) ) ; } ; } ReducePlan :: Accumulable ( AccumulablePlan { full_aggrs , simple_aggrs , distinct_aggrs , } ) } ReductionType :: Hierarchical => { let aggr_funcs : Vec < _ > = aggregates_list . iter ( ) . cloned ( ) . map ( | ( _ , aggr ) | aggr . func ) . collect ( ) ; let indexes : Vec < _ > = aggregates_list . into_iter ( ) . map ( | ( index , _ ) | index ) . collect ( ) ; // We don't have random access over Rows so we can simplify the // task of grabbing the inputs we are aggregating over by // generating a list of "skips" an iterator over the Row needs // to do to get the desired indexes. let skips = convert_indexes_to_skips ( indexes ) ; if monotonic { let monotonic = MonotonicPlan { aggr_funcs , skips } ; ReducePlan :: Hierarchical ( HierarchicalPlan :: Monotonic ( monotonic ) ) } else { let mut buckets = vec ! [ ] ; let mut current = 16 ; // Plan for 4B records in the expected case if the user // didn't specify a group size. let limit = expected_group_size . unwrap_or ( 4_000_000_000 ) ; // Distribute buckets in powers of 16, so that we can strike // a balance between how many inputs each layer gets from // the preceding layer, while also limiting the number of // layers. while current < limit { buckets . push ( current as u64 ) ; current *= 16 ; } // We need to store the bucket numbers in decreasing order. buckets . reverse ( ) ; let bucketed = BucketedPlan { aggr_funcs , skips , buckets , } ; ReducePlan :: Hierarchical ( HierarchicalPlan :: Bucketed ( bucketed ) ) } } ReductionType :: Basic => { if aggregates_list . len ( ) == 1 { ReducePlan :: Basic ( BasicPlan :: Single ( aggregates_list [ 0 ] . 0 , aggregates_list [ 0 ] . 1 . clone ( ) , ) ) } else { ReducePlan :: Basic ( BasicPlan :: Multiple ( aggregates_list ) ) } } } }
Generate a plan for computing the specified type of aggregations.
Generate a plan for computing the specified type of aggregations .
typ
ReductionType
aggregates_list
Vec<(usize, AggregateExpr)>
monotonic
bool
expected_group_size
Option<usize>
typ
ReductionType
aggregates_list
Vec<(usize, AggregateExpr)>
monotonic
bool
expected_group_size
Option<usize>