[Analytics-Engine][MPP] Move MPP exchange placement into top-down CBO - #22833
[Analytics-Engine][MPP] Move MPP exchange placement into top-down CBO#22833LantaoJin wants to merge 74 commits into
Conversation
…e for top-down MPP planning Signed-off-by: Lantao Jin <ltjin@amazon.com>
…nd Filter Signed-off-by: Lantao Jin <ltjin@amazon.com>
…oin behind setTopDownOpt Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ator split rule Signed-off-by: Lantao Jin <ltjin@amazon.com>
…sm in the cost model Signed-off-by: Lantao Jin <ltjin@amazon.com>
…lelism-collapse charge Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ash partitioning Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ion provides Signed-off-by: Lantao Jin <ltjin@amazon.com>
… models parallelism Signed-off-by: Lantao Jin <ltjin@amazon.com>
…orcement pass Signed-off-by: Lantao Jin <ltjin@amazon.com>
… measured trait staleness Signed-off-by: Lantao Jin <ltjin@amazon.com>
…lized shuffles Signed-off-by: Lantao Jin <ltjin@amazon.com>
… satisfies-gated enforcer Signed-off-by: Lantao Jin <ltjin@amazon.com>
… trait staleness Signed-off-by: Lantao Jin <ltjin@amazon.com>
…and distributed-join inputs Signed-off-by: Lantao Jin <ltjin@amazon.com>
…o the enforcement pass Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ggregation toggle Signed-off-by: Lantao Jin <ltjin@amazon.com>
…annot move yet Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ize floor Signed-off-by: Lantao Jin <ltjin@amazon.com>
… aggregates Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ecord the narrowing Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ed buffers Signed-off-by: Lantao Jin <ltjin@amazon.com>
…fuse co-partitioned join tiers Signed-off-by: Lantao Jin <ltjin@amazon.com>
…d child and null out unbuildable enforcers Signed-off-by: Lantao Jin <ltjin@amazon.com>
…re LIMIT Signed-off-by: Lantao Jin <ltjin@amazon.com>
…deleted derive rule Signed-off-by: Lantao Jin <ltjin@amazon.com>
… where it is optimal Signed-off-by: Lantao Jin <ltjin@amazon.com>
…sidual failures were regressions Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ts branch Signed-off-by: Lantao Jin <ltjin@amazon.com>
Signed-off-by: Lantao Jin <ltjin@amazon.com>
…e deadlock" Signed-off-by: Lantao Jin <ltjin@amazon.com>
…fails in a minute" Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ts producers" Signed-off-by: Lantao Jin <ltjin@amazon.com>
Signed-off-by: Lantao Jin <ltjin@amazon.com>
…ition" Signed-off-by: Lantao Jin <ltjin@amazon.com>
|
❌ Gradle check result for 0e6070d: null Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
…n-traits Signed-off-by: Lantao Jin <ltjin@amazon.com>
0e6070d to
d3c439b
Compare
|
Persistent review updated to latest commit d3c439b |
|
Persistent review updated to latest commit 6c7ac24 |
|
❌ Gradle check result for 6c7ac24: null Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
| * needed (unlike {@link OpenSearchProject}), which makes this the simplest pass-through in the tree. | ||
| */ | ||
| @Override | ||
| public Pair<RelTraitSet, List<RelTraitSet>> passThroughTraits(RelTraitSet required) { |
There was a problem hiding this comment.
I don't see computeSelfCost hacks getting removed with this PR which removes the gains from having properly implemented Trait Propagation.
There was a problem hiding this comment.
Agreed as a criterion, with one correction on what's removable.
The method itself has to stay, the parallelism terms in it are new here and load-bearing. Before this PR every legal join shape returned makeTinyCost(), which is exactly why top-down always picked the coordinator: an N-way distributed join priced identically to a serial one. Traits alone don't fix that.
The makeInfiniteCost() legality branches are the hacks, and they survive for one reason: rules still construct nodes with hand-built trait sets, bypassing passThroughTraits/deriveTraits. The hooks are tight — Join accepts only SINGLETON on pass-through, derives only coord-singleton or key-matched WORKER+HASH, and Aggregate prohibits derivation for SINGLE/FINAL, so search can't derive an illegal shape. But the physical conversion rules build OpenSearchAggregate(SINGLE) and OpenSearchJoin at whatever traits the child has (including a partitioned scan, i.e. the under-count), and the split rules hand-build the broadcast shape that deriveTraits deliberately declines. Infinite cost is what currently prunes those.
So the fix is to stop hand-building traits in rules, have them convert() inputs to the distribution the operator requires and let deriveTraits produce the output, after which the Join shape check becomes an assertion rather than a cost. I'll take that here.
The one branch that can't go that way is AggregateSplitRule's PARTIAL: it builds at child.getTraitSet(), and at match time the child's distribution can still be Type.ANY, so the rule cannot check legality itself and cost has to prune it when the trait resolves. That's the missing "partitioned, specification irrelevant" value in OpenSearchDistribution (Type.ANY means unresolved, not anything acceptable), the same gap behind your other two comments, and adding it also deletes OpenSearchPartialAggregatePushdownRewriter.
Will attempt it again.
expani
left a comment
expani
left a comment
There was a problem hiding this comment.
Thanks for getting this started @LantaoJin If you are planning some quick bug fixes I would suggest keeping them separate from Trait Propagation
Have some high level comments on the general gating criteria for ensuring Trait Propagation is implemented properly
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public final class OpenSearchLargeJoinDistributionRewriter { |
There was a problem hiding this comment.
Why do we need a rewrite for large Join distribution ?
We should be striving to remove rewrites as we go on.
There was a problem hiding this comment.
Because there are two named CBO gaps and an exit condition: isPureShardScanShape rejects any input subtree containing a Join (so an upper join never gets a WORKER+HASH alternative at all), and exchange cost has no memory/spill term (so gathering always prices cheaper). Measured worth: without it, sf=10 drops 18/22 → 14/22 with q3/q5/q7/q11/q21 hitting ReduceSizeExceededException.
| * | ||
| * @opensearch.internal | ||
| */ | ||
| public final class OpenSearchPartialAggregatePushdownRewriter { |
There was a problem hiding this comment.
There are multiple places that handle pushing down partial aggs. Another rewriter for doing it with Joins seems incorrect.
Also, it looks like we are introducing Join awareness in Aggregate split rules can we try to see how to keep it constrainted to individual relational operators instead ?
That should make the code simpler and remove a lot of these rules
I did. see #22800 #22787 #22878 #22818 #22798 #22915 #22916 (I would appreciate it if you could review them as well). These are all bugfix PRs which are not related to MPP but found in developing this optimization and benchmarking. This PR now includes some of them since I need to make the TPC-H pass. With these PRs being merged, I will rebase here. |
A PARTIAL aggregate rides its input's partitioning, so it cannot deliver a gather. Passing a SINGLETON demand down to its input built PARTIAL-over-gathered — the shape computeSelfCost then has to price at infinity — so the hook manufactured an alternative only cost could reject. Decline it instead; the gather belongs above a PARTIAL (the ExchangeReducer that FINAL reads), never below it. Signed-off-by: Lantao Jin <ltjin@amazon.com>
DistributionAware was the bottom-up distribution contract the post-CBO DistributionEnforcementPass consulted, as its own javadoc said. With that pass gone, requiredInputDistribution has no production caller at all and deriveOutputDistribution has one intra-class use in OpenSearchProject, so the interface left the tree carrying two parallel trait APIs where only Calcite's PhysicalNode hooks are live. Removes the interface and its eight implementations, keeps Project's output remapping as a private helper, and rewrites the javadoc that still narrated the deleted pass as if it ran — including the two TODO(trait-propagation) blocks that promised the migration this branch performs. OpenSearchJoin's and OpenSearchAggregate's infinite-cost gates are documented as what they actually are: the requirement declaration for a node that is both the HEP marking output and the physical operator. Seeding those nodes UNRESOLVED to retire the gates was measured and does not work — a demand for Type.ANY is satisfied by anything, so an ANY input subset makes the parent's check skip and a per-partition SINGLE aggregate over a hash-partitioned join wins on tiny cost. DistributionAwareTests only exercised the deleted methods. JoinTraitDerivationTests replaces it against deriveTraits/passThroughTraits and adds the first unit coverage of the RIGHT/FULL invariant: those joins derive no hash alternative at all, because their null-extended rows carry NULL left keys and no 'partitioned, specification unknown' value exists to describe the output with. Signed-off-by: Lantao Jin <ltjin@amazon.com>
OpenSearchAggregateSplitRule and OpenSearchPartialAggregatePushdownRewriter each built FINAL(gather(PARTIAL(input))) with their own copy of the same assembly, the rewriter reaching across for three of the rule's public statics. The only real difference is who owns the exchange: convert() during CBO so Volcano can dedup it, buildReducer() post-CBO because the satisfies-gated enforcer would insert nothing over an input whose trait set still carries CBO's coordSingleton. AggregatePartialFinalSplit now owns the assembly and takes that difference as a gather function; the three helpers become private to it. Both call sites keep their own answer to WHEN to split, which is the part that genuinely differs. Also records why childForcesGather cannot become a trait check: a RelSubset can advertise a partitioned trait while every operator inside it is only implementable via a gather, so trusting the trait costs 7 CannotPlanException failures. Registering both alternatives instead was measured too — no gain, 4 regressions. Signed-off-by: Lantao Jin <ltjin@amazon.com>
LantaoJin
changed the title
…n-traits Signed-off-by: Lantao Jin <ltjin@amazon.com>
6c7ac24 to
8bbf7dd
Compare
|
Persistent review updated to latest commit 8bbf7dd |
… cost Placement legality was a shape table per operator: OpenSearchJoin listed three legal distributions with a per-input match check, OpenSearchAggregate priced SINGLE-over-partitioned and PARTIAL-over-singleton at infinity. Each of those tables also had to SKIP an input whose distribution was still Type.ANY, because ANY is Volcano's UNRESOLVED placeholder and judging an unresolved trait as illegal prices out alternatives the memo has not finished expanding. Every one of those skips was a hole. A node that cannot demand traits of its inputs (the HEP marking output — convert() is a no-op in HEP) can only be seeded UNRESOLVED, and an unresolved subset is then satisfied-by-anything: it reaches a parent, the parent's check skips, the alternative wins on tiny cost, and the subset resolves to the cheapest concrete member. Measured directly: a Union that skipped the check let a per-partition SINGLE aggregate through, and the gather above concatenated three partial results without merging them. hasUnresolvedInput() replaces the skips with one operator-independent invariant — an unresolved input cannot be consumed, because nothing above it has a defined cost or a defined correctness — applied in all eight operators that consume inputs. buildEnforcer declines it too: you cannot gather data whose location is undecided, and allowing it produced the 'exchange on top of the unresolved subtree' shape, which is always cheaper (it moves the operator's output instead of its input) and would therefore always win. Also passes a SINGLETON demand through verbatim in Join's co-located case rather than narrowing it to COORDINATOR, so a locality-agnostic demand over two 1-shard inputs of the same table is satisfied where the data already sits. Signed-off-by: Lantao Jin <ltjin@amazon.com>
…infinite cost A SINGLE aggregate is only correct over already-gathered input, and until now that requirement lived in computeSelfCost as makeInfiniteCost() — legality expressed through the cost channel, which is meant for ranking. It had to live there because the requirement was never stated as a trait: passThroughTraits declined a SINGLETON demand for any splittable aggregate, so nothing ever asked the aggregate to deliver a distribution, and the HEP marking rule stamped it with whatever its child happened to carry. The aggregate now DECLARES it. The marking rule seeds the node UNRESOLVED (it runs in HEP and cannot demand anything of its input, so it must not claim a placement either), passThroughTraits claims any SINGLETON demand and passes the required locality through verbatim, and the SINGLE-over-partitioned infinite-cost branch is deleted: the illegal pair is no longer constructible, so there is nothing left for cost to reject. The sibling invariant that keeps this safe is hasUnresolvedInput() from the previous commit. Sort and the window/pinned Project stop narrowing a SINGLETON demand to coordSingleton for the same reason. The root asks for anySingleton (locality null), which a 1-shard SINGLETON(SHARD) subtree already satisfies; narrowing it inserted a gather to move data that was already on one node. What a window frame needs is that all rows are on ONE node, not that the node is the coordinator. The PARTIAL-over-singleton branch stays. OpenSearchAggregateSplitRule builds its PARTIAL at child.getTraitSet(), which at match time can still be ANY, so that rule cannot check its own legality and registers optimistically; retiring that branch needs a 'partitioned, specification irrelevant' value in OpenSearchDistribution. Signed-off-by: Lantao Jin <ltjin@amazon.com>
Eleven assertions moved. Grouped by why: Fused worker tiers (CascadeShuffleProbeTests x4): co-partitioned join levels fuse into ONE worker tier instead of lowering to two binary tiers, so the upper join consumes the lower join directly and only the bottom join sits over two shuffles. assertFusedTier in the same file already asserted this shape for the join-only cases; these four catch up for the agg-over-join cases. Rule counts (RuleProfilingListenerTests x2): ExpandConversionRule drops to 0 — with the gather placed by Convention.enforce there is no AbstractConverter left to expand. OpenSearchAggregateSplitRule rises to 8 and 3 because an UNRESOLVED seed makes it match once per (aggregate, input-subset) pair; that is search space, not extra plan nodes. One-row estimates (AggregateSplitCostTests x2): with the estimate at the 1.0 floor, single-stage over a gather is genuinely cheaper than two-phase aggregation, and CBO now picks it because the shape is no longer forced by an infinite-cost branch. Renamed off 'partialStaysBelowExchange', which contradicted the new shape. The invariant worth guarding still holds and is stated in the javadoc: a PARTIAL above the gather remains unconstructible. Fewer exchanges (DAGShapeTests x1): a single-shard self-join with a scalar aggregate collapses from two stages to one — the co-located join delivers SINGLETON(SHARD) and nothing forces a gather. Known single-shard regression (PlanShapeTests x2): an aggregate under a single-shard join or union now sits over a gather it does not need, because the only concrete singleton alternative offered to it is coordSingleton. Both tests carry the cause and the boundary in a comment. Four attempts to remove it each moved the extra gather elsewhere instead. Multi-shard plans are unaffected: the sf=10 TPC-H sweep scores 17/22 before and after, with identical exchange shapes in 21 of 22 queries and P50 total 29.5s vs 29.7s. Signed-off-by: Lantao Jin <ltjin@amazon.com>
|
Persistent review updated to latest commit 5fc726a |
|
❌ Gradle check result for 5fc726a: QUEUE_TIMEOUT Please examine the workflow log, locate, and copy-paste the failure(s) below, then iterate to green. Is the failure a flaky test unrelated to your change? |
3 participants
Description
This is a mentioned followup of #21844:
Exchange placement for MPP joins and distributed aggregates was decided by
DistributionEnforcementPassthat ran after CBO , peeled the exchanges CBO had placed, and re-decided placement from scratch.
That existed because bottom-up Volcano optimises a subtree before its parent's requirement is known, so
distribution could not be a search dimension: every join was gathered to the coordinator on sight, and only a
later pass had the global view to fix it.
Calcite's top-down mode removes that constraint. This PR makes distribution a real trait so placement falls out
of search instead of being repaired afterwards.
What changed
Top-down trait propagation.
setTopDownOpt(true)plusPhysicalNodehooks (passThroughTraits/deriveTraits/getDeriveMode) on Project, Filter, Sort, Aggregate, Join and the exchanges, withOpenSearchConvention.enforcesupplying enforcers.Cost model learns parallelism. Every legal join shape previously returned
makeTinyCost(), so acoordinator join looked as cheap as an N-way distributed one and top-down always picked the coordinator. Join
cost is now
inputRows / parallelism, taken from the shuffle partition count or the broadcast probe estimate.DistributionEnforcementPassdeleted. Two single-purpose rewrites survive, run fromPlannerImplalongside the existing
OpenSearchTopKRewriter/OpenSearchSortPushdownRewriter.Verification
SF=10 Benchmark (3-workers EC2 cluster, 32g heap)
PASS: baseline 17/22 -> feature 20/22. Zero PASS->FAIL.
Aggregate over the 18 queries timed on BOTH sides: 21.5s -> 20.9s = −3% (1.03× faster).
q19/q21 are excluded because the baseline has no time for them; including them would flatter the branch by
counting queries main cannot complete.
SF=100 Benchmark (8-workers EC2 cluster, 16g heap):
PASS: baseline 6/22 -> feature 10/22. Five gains (q4, q7, q12, q13, q14) against one loss (q8).
Why q17 still failed in sf=10 with top-down CBO:
Both build an enormous grouping over the full 60M-row lineitem, and in both cases the grouping key is high-cardinality:
analytics.mpp.shuffle.partitions=24. sum(l_quantity) group by l_orderkey over all 60M rows gives ~15M groups, and that semi-join subquery sits underneath a 3-way customer ⋈ orders ⋈ lineitem. lineitem is consumed twice in one query.Related Issues
Resolves #[Issue number to be closed when this PR is merged]
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.