github-actions[bot] commented on code in PR #68199:
URL: https://github.com/apache/doris/pull/68199#discussion_r4046215530
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java:
##########
@@ -2605,6 +2633,35 @@ public PlanFragment visitPhysicalSetOperation(
setOperationNode.setColocate(true);
}
+ // Storage-bucketed children only appear when the FE local shuffle
planner is active:
+ // ChildrenPropertiesRegulator and RequestPropertyDeriver both gate
the bucket-shuffle
+ // alternative on enableLocalShufflePlanner. Gate the marker on the
same flag so the
+ // dependency is explicit and a future planner change that produced a
STORAGE_BUCKETED
+ // distribution outside the local-shuffle planner cannot silently mark
BUCKET_SHUFFLE here.
+ //
+ // Within that gate a storage-bucketed child means the regulator chose
the bucket shuffle
+ // alternative (it enforces the other children onto the basic child's
buckets), so the marker
+ // simply follows that decision. It must not re-check the table id
independently: the basic
+ // child selection in the regulator is the single place that vets the
layout, and re-checking
+ // here could suppress a bucket shuffle the property model already
committed to and desync a
+ // parent that aligned to the set operation output.
+ //
+ // Unlike hash join, BUCKET_SHUFFLE is not exclusive with isColocate
above: for a set
+ // operation isColocate describes the bucket-aligned scheduling of the
fragment (the
+ // 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
Review 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.
##########
be/test/exec/operator/agg_operator_test.cpp:
##########
@@ -96,7 +96,8 @@ struct MockAggSourceOperator : public AggSourceOperatorX {
class MockDistributionOperator final : public OperatorX<MockLocalState> {
public:
- MockDistributionOperator(ExchangeType exchange_type) :
_exchange_type(exchange_type) {}
+ MockDistributionOperator(TLocalPartitionType::type exchange_type)
Review 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.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java:
##########
@@ -718,6 +720,19 @@ protected void distribute(PhysicalPlan physicalPlan,
ExplainLevel explainLevel)
splitFragments(physicalPlan);
doDistribute(canUseNereidsDistributePlanner, explainLevel);
+
+ addLocalExchangeAfterDistribute();
+ }
+
+ private void addLocalExchangeAfterDistribute() {
+ SessionVariable sessionVariable =
cascadesContext.getConnectContext().getSessionVariable();
+ if (!sessionVariable.isEnableLocalShufflePlanner() ||
!sessionVariable.isEnableLocalShuffle()) {
Review 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.
##########
be/src/exec/exchange/local_exchanger.h:
##########
@@ -357,16 +358,15 @@ class AdaptivePassthroughExchanger : public
Exchanger<BlockWrapperSPtr> {
ENABLE_FACTORY_CREATOR(AdaptivePassthroughExchanger);
AdaptivePassthroughExchanger(int running_sink_operators, int
num_partitions,
int free_block_limit)
- : Exchanger<BlockWrapperSPtr>(running_sink_operators,
num_partitions,
- free_block_limit) {
+ : Exchanger<PartitionedBlock>(running_sink_operators,
num_partitions, free_block_limit,
Review 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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
