-
Notifications
You must be signed in to change notification settings - Fork 468
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Break complex
Reduce operators into simpler ones (#17013)
This is an experimental PR where we break apart `Reduce` operators that would be rendered with the `Collation` plan into atomic `Reduce` operators that are joined together instead. This has the potential to be (much) better than the collation plan, or .. to be worse. If we put an `ArrangeBy` at the end it wouldn't be worse, but it could be much better. This PR is mostly to look at some plans and see what changes and how. Edit: More explanation here: https://materializeinc.slack.com/archives/C02PPB50ZHS/p1673028168703109 Edit2: Slack message copy/pasted for visibility: > Other Reduce thought: We have various flavors of reduce plans, roughly three (accumulable, hierarchical, and "generic"). We collate these together with another reduce though .. we should just use a delta join. All of the constituents present arrangements of their outputs, and a delta join would use no additional memory (unlike the collation operator). Moreover, we could then push down predicates and projection and mapping and such. > Downside: delta joins don't produce an output arrangement, so it wouldn't always be the case that Reduce has an arrangement of its output. We could always make one at not much additional cost (and I think strictly less than the collation operator). Edit 3: scope has changed to only breaking apart enough aggregates to prevent collation. Fixes MaterializeInc/database-issues#2273 ### Motivation <!-- Which of the following best describes the motivation behind this PR? * This PR fixes a recognized bug. [Ensure issue is linked somewhere.] * This PR adds a known-desirable feature. [Ensure issue is linked somewhere.] * This PR fixes a previously unreported bug. [Describe the bug in detail, as if you were filing a bug report.] * This PR adds a feature that has not yet been specified. [Write a brief specification for the feature, including justification for its inclusion in Materialize, as if you were writing the original feature specification.] * This PR refactors existing code. [Describe what was wrong with the existing code, if it is not obvious.] --> ### Tips for reviewer <!-- Leave some tips for your reviewer, like: * The diff is much smaller if viewed with whitespace hidden. * [Some function/module/file] deserves extra attention. * [Some function/module/file] is pure code movement and only needs a skim. Delete this section if no tips. --> ### Checklist - [ ] This PR has adequate test coverage / QA involvement has been duly considered. - [ ] This PR evolves [an existing `$T ⇔ Proto$T` mapping](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/command-and-response-binary-encoding.md) (possibly in a backwards-incompatible way) and therefore is tagged with a `T-proto` label. - [ ] This PR includes the following [user-facing behavior changes](https://github.com/MaterializeInc/materialize/blob/main/doc/developer/guide-changes.md#what-changes-require-a-release-note): - <!-- Add release notes here or explicitly state that there are no user-facing behavior changes. --> --------- Signed-off-by: Moritz Hoffmann <mh@materialize.com> Co-authored-by: Moritz Hoffmann <mh@materialize.com>
- Loading branch information
1 parent
94c514b
commit 38cce6b
Showing
20 changed files
with
8,259 additions
and
3,774 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| // Copyright Materialize, Inc. and contributors. All rights reserved. | ||
| // | ||
| // Use of this software is governed by the Business Source License | ||
| // included in the LICENSE file. | ||
| // | ||
| // As of the Change Date specified in that file, in accordance with | ||
| // the Business Source License, use of this software will be governed | ||
| // by the Apache License, Version 2.0. | ||
|
|
||
| //! Breaks complex `Reduce` variants into a join of simpler variants. | ||
| //! | ||
| //! Specifically, any `Reduce` that contains two different "types" of aggregation, | ||
| //! in the sense of `ReductionType`, will be broken in to one `Reduce` for each | ||
| //! type of aggregation, each containing the aggregations of that type, | ||
| //! and the results are then joined back together. | ||
|
|
||
| use crate::TransformCtx; | ||
| use mz_compute_types::plan::reduce::reduction_type; | ||
| use mz_expr::MirRelationExpr; | ||
|
|
||
| /// Breaks complex `Reduce` variants into a join of simpler variants. | ||
| #[derive(Debug)] | ||
| pub struct ReduceReduction; | ||
|
|
||
| impl crate::Transform for ReduceReduction { | ||
| /// Transforms an expression through accumulated knowledge. | ||
| #[mz_ore::instrument( | ||
| target = "optimizer", | ||
| level = "debug", | ||
| fields(path.segment = "reduce_reduction") | ||
| )] | ||
| fn transform( | ||
| &self, | ||
| relation: &mut MirRelationExpr, | ||
| ctx: &mut TransformCtx, | ||
| ) -> Result<(), crate::TransformError> { | ||
| if ctx.features.enable_reduce_reduction { | ||
| relation.visit_pre_mut(&mut Self::action); | ||
| mz_repr::explain::trace_plan(&*relation); | ||
| } | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| impl ReduceReduction { | ||
| /// Breaks complex `Reduce` variants into a join of simpler variants. | ||
| pub fn action(relation: &mut MirRelationExpr) { | ||
| if let MirRelationExpr::Reduce { | ||
| input, | ||
| group_key, | ||
| aggregates, | ||
| monotonic, | ||
| expected_group_size, | ||
| } = relation | ||
| { | ||
| // We start by segmenting the aggregates into those that should be rendered independently. | ||
| // Each element of this list is a pair of lists describing a bundle of aggregations that | ||
| // should be applied independently. Each pair of lists correspond to the aggregaties and | ||
| // the column positions in which they should appear in the output. | ||
| // Perhaps these should be lists of pairs, to ensure they align, but their subsequent use | ||
| // is as the shredded lists. | ||
| let mut segmented_aggregates: Vec<(Vec<mz_expr::AggregateExpr>, Vec<usize>)> = | ||
| Vec::new(); | ||
|
|
||
| // Our rendering currently produces independent dataflow paths for 1. all accumulable aggregations, | ||
| // 2. all hierarchical aggregations, and 3. *each* basic aggregation. | ||
| // We'll form groups for accumulable, hierarchical, and a list of basic aggregates. | ||
| let mut accumulable = (Vec::new(), Vec::new()); | ||
| let mut hierarchical = (Vec::new(), Vec::new()); | ||
|
|
||
| use mz_compute_types::plan::reduce::ReductionType; | ||
| for (index, aggr) in aggregates.iter().enumerate() { | ||
| match reduction_type(&aggr.func) { | ||
| ReductionType::Accumulable => { | ||
| accumulable.0.push(aggr.clone()); | ||
| accumulable.1.push(group_key.len() + index); | ||
| } | ||
| ReductionType::Hierarchical => { | ||
| hierarchical.0.push(aggr.clone()); | ||
| hierarchical.1.push(group_key.len() + index); | ||
| } | ||
| ReductionType::Basic => segmented_aggregates | ||
| .push((vec![aggr.clone()], vec![group_key.len() + index])), | ||
| } | ||
| } | ||
|
|
||
| // Fold in hierarchical and accumulable aggregates. | ||
| if !hierarchical.0.is_empty() { | ||
| segmented_aggregates.push(hierarchical); | ||
| } | ||
| if !accumulable.0.is_empty() { | ||
| segmented_aggregates.push(accumulable); | ||
| } | ||
| segmented_aggregates.sort(); | ||
|
|
||
| // Do nothing unless there are at least two distinct types of aggregations. | ||
| if segmented_aggregates.len() < 2 { | ||
| return; | ||
| } | ||
|
|
||
| // For each type of aggregation we'll plan the corresponding `Reduce`, | ||
| // and then join the at-least-two `Reduce` stages together. | ||
| // TODO: Perhaps we should introduce a `Let` stage rather than clone the input? | ||
| let mut reduces = Vec::with_capacity(segmented_aggregates.len()); | ||
| // Track the current and intended locations of each output column. | ||
| let mut columns = Vec::new(); | ||
|
|
||
| for (aggrs, indexes) in segmented_aggregates { | ||
| columns.extend(0..group_key.len()); | ||
| columns.extend(indexes); | ||
|
|
||
| reduces.push(MirRelationExpr::Reduce { | ||
| input: input.clone(), | ||
| group_key: group_key.clone(), | ||
| aggregates: aggrs, | ||
| monotonic: *monotonic, | ||
| expected_group_size: *expected_group_size, | ||
| }); | ||
| } | ||
|
|
||
| // Now build a `Join` of the reduces, on their keys, followed by a permutation of their aggregates. | ||
| // Equate all `group_key` columns in all inputs. | ||
| let mut equivalences = vec![Vec::with_capacity(reduces.len()); group_key.len()]; | ||
| for column in 0..group_key.len() { | ||
| for input in 0..reduces.len() { | ||
| equivalences[column].push((input, column)); | ||
| } | ||
| } | ||
|
|
||
| // Determine projection that puts aggregate columns in their intended locations, | ||
| // and projects away repeated key columns. | ||
| let max_column = columns.iter().max().expect("Non-empty aggregates expected"); | ||
| let mut projection = Vec::with_capacity(max_column + 1); | ||
| for column in 0..max_column + 1 { | ||
| projection.push(columns.iter().position(|c| *c == column).unwrap()) | ||
| } | ||
|
|
||
| // Now make the join. | ||
| *relation = MirRelationExpr::join(reduces, equivalences).project(projection); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.