Build the dataflow to compute and arrange multiple non-accumulable, non-hierarchical aggregations on `input`. This function assumes that we are explicitly rendering multiple basic aggregations. For each aggregate, we render a different reduce operator, and then fuse results together into a final arrangement that presents all the results in the order specified by `aggrs`.
(
&self,
input: VecCollection<'s, T, (Row, Row), Diff>,
aggrs: Vec<AggregateExpr>,
key_arity: usize,
mfp_after: Option<SafeMfpPlan>,
)
| 371 | /// results together into a final arrangement that presents all the results |
| 372 | /// in the order specified by `aggrs`. |
| 373 | fn build_basic_aggregates<'s>( |
| 374 | &self, |
| 375 | input: VecCollection<'s, T, (Row, Row), Diff>, |
| 376 | aggrs: Vec<AggregateExpr>, |
| 377 | key_arity: usize, |
| 378 | mfp_after: Option<SafeMfpPlan>, |
| 379 | ) -> ( |
| 380 | RowRowArrangement<'s, T>, |
| 381 | VecCollection<'s, T, DataflowErrorSer, Diff>, |
| 382 | ) { |
| 383 | // We are only using this function to render multiple basic aggregates and |
| 384 | // stitch them together. If that's not true we should complain. |
| 385 | if aggrs.len() <= 1 { |
| 386 | self.error_logger().soft_panic_or_log( |
| 387 | "Too few aggregations when building basic aggregates", |
| 388 | &format!("len={}", aggrs.len()), |
| 389 | ) |
| 390 | } |
| 391 | let mut err_output = None; |
| 392 | let mut to_collect = Vec::new(); |
| 393 | for (index, aggr) in aggrs.into_iter().enumerate() { |
| 394 | let (result, errs) = self.build_basic_aggregate( |
| 395 | input.clone(), |
| 396 | index, |
| 397 | &aggr, |
| 398 | err_output.is_none(), |
| 399 | key_arity, |
| 400 | None, |
| 401 | false, |
| 402 | ); |
| 403 | if errs.is_some() { |
| 404 | err_output = errs |
| 405 | } |
| 406 | to_collect |
| 407 | .push(result.as_collection(move |key, val| (key.to_row(), (index, val.to_row())))); |
| 408 | } |
| 409 | |
| 410 | // Allocations for the two closures. |
| 411 | let mut datums1 = DatumVec::new(); |
| 412 | let mut datums2 = DatumVec::new(); |
| 413 | let mfp_after1 = mfp_after.clone(); |
| 414 | let mfp_after2 = mfp_after.filter(|mfp| mfp.could_error()); |
| 415 | |
| 416 | let arranged = differential_dataflow::collection::concatenate(input.scope(), to_collect) |
| 417 | .mz_arrange::< |
| 418 | ColumnationChunker<_>, |
| 419 | RowValBatcher<_, _, _>, |
| 420 | RowValBuilder<_, _, _>, |
| 421 | RowValSpine<_, _, _>, |
| 422 | >( |
| 423 | "Arranged ReduceFuseBasic input", |
| 424 | ); |
| 425 | |
| 426 | let output = arranged |
| 427 | .clone() |
| 428 | .mz_reduce_abelian::<_, RowRowBuilder<_, _>, RowRowSpine<_, _>>("ReduceFuseBasic", { |
| 429 | move |key, input, output| { |
| 430 | let temp_storage = RowArena::new(); |
no test coverage detected