branch-4.2: [refactor](local shuffle) Move local exchange planning from BE to FE #63366 #64793 #65348 #65129 #66295 #66570 #67177 #67941 - #68199
Conversation
|
run buildall |
|
/review |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Static review of exact head 837e903 across all 124 changed files. I found four blocking issues: the production BE cannot compile because AdaptivePassthroughExchanger initializes a non-base template specialization; the changed aggregation unit test still uses the removed ExchangeType enum; default-on FE local-shuffle planning can send the new LOCAL_EXCHANGE_NODE to an older BE during a rolling upgrade without a capability gate; and disabling both local-shuffle controls can leave a bucket-shuffle UNION/EXCEPT unmarked and silently drop rows for buckets pruned from its basic child. I also traced FE/BE task and dependency cardinalities, serial and PASS_TO_ONE boundaries, bucket/set-operation ownership and missing-bucket behavior, hash-key/property propagation, runtime-filter merge selection, and coordinator/load/cloud option propagation; no additional distinct correctness issue survived adjudication. Existing inline review context was empty, so these comments are not duplicates. No user-specific focus was provided. This was static-only review: builds and tests were prohibited and were not run.
| int free_block_limit) | ||
| : Exchanger<BlockWrapperSPtr>(running_sink_operators, num_partitions, | ||
| free_block_limit) { | ||
| : Exchanger<PartitionedBlock>(running_sink_operators, num_partitions, free_block_limit, |
There was a problem hiding this comment.
[P1] Initialize the base specialization this class actually inherits
AdaptivePassthroughExchanger still derives from Exchanger<BlockWrapperSPtr>, but this initializer now names the distinct Exchanger<PartitionedBlock> specialization. A constructor initializer cannot initialize a non-base class, so the production BE target fails to compile here. Please restore Exchanger<BlockWrapperSPtr>(...) (or consistently change the class and its block-wrapper implementation if that migration is intended).
| class MockDistributionOperator final : public OperatorX<MockLocalState> { | ||
| public: | ||
| MockDistributionOperator(ExchangeType exchange_type) : _exchange_type(exchange_type) {} | ||
| MockDistributionOperator(TLocalPartitionType::type exchange_type) |
There was a problem hiding this comment.
[P1] Finish migrating this test's callers off the removed enum
Changing this mock to accept TLocalPartitionType::type leaves toggle_hash_shuffle_for_safe_child below using the now-removed ExchangeType on lines 146, 151, and 156, so this test target no longer compiles. Please migrate those remaining NOOP references to TLocalPartitionType::NOOP and the hash expectation to TLocalPartitionType::GLOBAL_EXECUTION_HASH_SHUFFLE as well.
|
|
||
| private void addLocalExchangeAfterDistribute() { | ||
| SessionVariable sessionVariable = cascadesContext.getConnectContext().getSessionVariable(); | ||
| if (!sessionVariable.isEnableLocalShufflePlanner() || !sessionVariable.isEnableLocalShuffle()) { |
There was a problem hiding this comment.
[P1] Gate the new plan node for mixed-version backends
This is the only guard before the default-on planner emits the newly added LOCAL_EXCHANGE_NODE. During a rolling upgrade a new FE can therefore send enum value 38 to an older BE, which has neither the factory case nor this node contract; ignoring the new optional query flag cannot help because the request already contains an unsupported node. Please keep FE planning off unless every selected BE advertises support (or add an execution-version/capability gate and fall back to BE planning), and cover the new-FE/old-BE path.
| // basic child scans buckets directly), while BUCKET_SHUFFLE describes how the other | ||
| // children arrive (bucket-shuffle exchanges). Both routes converge to the same | ||
| // bucket-hash local exchange requirement in SetOperationNode.enforceAndDeriveLocalExchange. | ||
| if (context.getSessionVariable() != null |
There was a problem hiding this comment.
[P1] Preserve the bucket marker when local shuffle is disabled
setOperationBucketShuffleAllowed() still permits a storage-bucketed set operation when enable_local_shuffle=false, even if enable_local_shuffle_planner is also false, because one task per instance preserves bucket alignment. This extra guard leaves that UNION/EXCEPT marked PARTITIONED, so shouldFillUpInstances() skips receivers for buckets pruned from the basic child; rows from the shuffled child that hash to those buckets are then sent to dummy channels and silently dropped. Please use the same eligibility predicate here (or disallow the optimizer alternative), and cover the both-false missing-bucket case.
1 participant
Cherry-picked from #63366 #64793 #65348 #65129 #66295 #66570 #67177 #67941