Conversation
|
run buildall |
|
/review |
|
Superseded by #68199, which backports the whole local shuffle series as one PR. |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Requesting changes for one production performance regression and three test defects on exact head 15e5b8589eddb404f72b1d93ed9bd5d71ffea462. The PR is now closed and marked superseded by #68199; these findings apply to this reviewed diff and should be carried forward if the same changes are present there.
Critical checkpoints:
- Data correctness: the core placement fix is otherwise sound. Nereids key/layout regulation, serial flattening, exact-requirement satisfaction, fragment boundaries, and constant-Union redistribution preserve result correctness across the reviewed paths.
- Performance: the new bucket-aligned Union arm overrides an upgraded join's exact LOCAL requirement and adds a redundant heavy BUCKET hash immediately before the mandatory LOCAL hash.
- Parallel paths and tests: the new audit misses
NOOPand nested multi-input placements, its nominal upgrade-disabled arm can still upgrade, and the focused planner test leaks a session setting across methods. - Configuration, compatibility, lifecycle, concurrency, persistence, writes, observability, and FE/BE boundaries: no additional production defect was found.
- Scope and user focus: all eight changed files and relevant upstream/downstream call chains were covered; no additional user-provided focus was present.
- Convergence: the third and final normal/risk-focused round returned
NO_NEW_VALUABLE_FINDINGS; all candidates were independently adjudicated.
Validation was static only, as required by the review runner; no builds or tests were run.
| // by a different hash function and break build/probe alignment. | ||
| if (!canPropagateHash) { | ||
| requireChild = LocalExchangeTypeRequire.noRequire(); | ||
| outputType = LocalExchangeType.NOOP; |
There was a problem hiding this comment.
[P2] Preserve an exact LOCAL parent requirement here
When a bucket join takes its bucket-to-local parallelism upgrade, it deliberately passes RequireSpecific(LOCAL_EXECUTION_HASH_SHUFFLE) to this Union. This branch now replaces that exact request with BUCKET: the pooled basic side gets PASSTHROUGH -> BUCKET_HASH, the Union reports BUCKET, and the join immediately inserts LOCAL_HASH above the whole Union because BUCKET cannot satisfy its exact requirement. Union only concatenates, so the lower BUCKET redistribution adds CPU, buffering, and backpressure—usually on the largest basic side—without affecting correctness before every row is rehashed again. Please use the BUCKET override only for a generic hash request and preserve exact LOCAL/GLOBAL requirements; add an upgrade-path shape assertion that rejects this double-hash chain.
| Map<LocalExchangeType, List<Integer>> byPlacement = new LinkedHashMap<>(); | ||
| for (int i = 0; i < node.getChildren().size(); i++) { | ||
| LocalExchangeType placement = effectivePlacement(node.getChild(i), ctx); | ||
| if (placement.isHashShuffle()) { |
There was a problem hiding this comment.
[P2] Make missing and nested placements visible to this audit
effectivePlacement() returns NOOP for every multi-input child, and this filter then discards every NOOP/non-hash branch. A bucket/partitioned HashJoinNode, IntersectNode, or ExceptNode with one pooling branch left unshuffled and one BUCKET/GLOBAL sibling therefore leaves a one-entry map and passes; an outer operator likewise cannot see a nested set operation's output placement. Those are precisely missing-realignment regressions this matrix is meant to catch. Please make the check distribution-mode aware and require every correctness-sensitive branch to resolve to a compatible placement, with focused failing cases for NOOP + BUCKET and a nested multi-input child.
| // ratio <= 1 disables the upgrade entirely; 1.01 makes it fire whenever instances | ||
| // slightly exceed buckets-with-data, which is what a bucket join above a mis-claiming | ||
| // child needs in order to be fooled by that claim. | ||
| sv.setLocalShuffleBucketUpgradeRatio(bucketUpgrade ? 1.01 : 1.5); |
There was a problem hiding this comment.
[P2] Actually disable the upgrade in the false matrix arm
shouldUpgradeBucketParallelism() treats every ratio above 1 as enabled; it disables the optimization only for ratio <= 1. Here the six/seven buckets are spread over three backends while the test requests eight instances per worker, so 1.5 can still satisfy instances > bucketsPerWorker * ratio and both boolean arms can exercise the upgraded branch. Please use 0 or 1 for the false arm (or assert the computed eligibility) so this matrix really covers both production paths.
| // ← Exchange (bucket shuffle) ← scan(t2) | ||
| setupLocalShuffleSession(sv -> { | ||
| sv.setForceToLocalShuffle(true); | ||
| sv.setBucketShuffleDowngradeRatio(0); |
There was a problem hiding this comment.
[P2] Restore the bucket-shuffle downgrade setting after this test
TestWithFeService is PER_CLASS, so every method here shares this ConnectContext. setupLocalShuffleSession() resets the other local-shuffle knobs but not bucketShuffleDowngradeRatio; this test changes the default from 0.8 to 0 and leaves it there, so whichever planner tests run afterward can choose different bucket-shuffle plans than they do in isolation. Please reset this variable in the common setup before applying tweaks, or snapshot and restore it in this test.
1 participant
Cherry-picked from #66295