github-actions[bot] commented on code in PR #68185:
URL: https://github.com/apache/doris/pull/68185#discussion_r4045751498
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/properties/ChildrenPropertiesRegulator.java:
##########
@@ -58,6 +58,8 @@
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
+import org.apache.logging.log4j.LogManager;
Review Comment:
[P1] Remove the unused imports that fail CheckStyle
Neither `LogManager` nor `Logger` is referenced in this file. The exact-head
CheckStyle job already rejects these two added lines (and reports no other
Checkstyle errors), so this PR cannot pass the required gate until both imports
are removed or used.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java:
##########
@@ -187,6 +187,13 @@ public boolean isSerialNode() {
@Override
public boolean isSerialOperatorOnBe(ConnectContext context) {
+ if (context != null &&
context.getSessionVariable().isEnableLocalShufflePlanner()) {
Review Comment:
[P1] Preserve serial handling for funneled non-bucket exchanges
Removing the fragment-level serial-scan condition here makes a RANDOM
exchange in a pooled scan fragment non-serial. Mixed partitioned/unpartitioned
set operations create that shape: the partitioned scan child is absorbed, while
the unpartitioned child remains behind a RANDOM exchange. `DistributePlanner`
still sends every non-bucket input only to `getFirstInstancePerWorker()`, and
the new zero-sender completion only recognizes bucket shuffle. The other local
tasks therefore construct receivers with the full sender count but get no
channel or EOS, so the query can wait forever. Please keep these funneled
exchanges serial, or spread and terminate their receivers as well.
##########
fe/fe-core/src/main/java/org/apache/doris/planner/AddLocalExchange.java:
##########
@@ -81,11 +90,107 @@ public void
addLocalExchange(FragmentIdMapping<DistributedPlan> distributedPlans
if (maxPerBeInstances <= 1) {
continue;
}
+ context.setCurrentFragmentBucketUpgradeEligible(
+ isBucketUpgradeEligible(pipePlan, maxPerBeInstances,
context));
PlanFragment fragment = pipePlan.getFragmentJob().getFragment();
addLocalExchangeForFragment(fragment, context);
}
}
+ /**
+ * Bucket → local-hash parallelism upgrade eligibility .
+ *
+ * A pooled bucket-join fragment runs its bucket joins at bucket-count
parallelism:
+ * each LocalShuffleBucketJoinAssignedJob owns a disjoint set of join
buckets and only
+ * instances with buckets do join work (e.g. 8 buckets/BE but 16
instances/BE → 8 idle).
+ * When nothing above the join needs bucket alignment, HashJoinNode can
re-distribute
+ * both sides with LOCAL_EXECUTION_HASH_SHUFFLE to use all instances — see
+ * {@link HashJoinNode#enforceAndDeriveLocalExchange}.
+ *
+ * This method computes the per-fragment numeric condition from the actual
instance
+ * assignment: maxPerBeInstances > maxBucketsWithDataPerWorker × ratio.
The ratio comes
+ * from session variable {@code local_shuffle_bucket_upgrade_ratio};
values <= 1 disable
+ * the upgrade entirely (a required parallelism gain of at most 1x means
no gain).
+ */
+ private boolean isBucketUpgradeEligible(PipelineDistributedPlan pipePlan,
+ long maxPerBeInstances, PlanTranslatorContext context) {
+ ConnectContext connectContext = context.getConnectContext();
+ if (connectContext == null || connectContext.getSessionVariable() ==
null) {
+ return false;
+ }
+ double ratio =
connectContext.getSessionVariable().getLocalShuffleBucketUpgradeRatio();
+ List<AssignedJob> instanceJobs = pipePlan.getInstanceJobs();
+ if (instanceJobs.isEmpty()
+ ||
!instanceJobs.stream().allMatch(LocalShuffleBucketJoinAssignedJob.class::isInstance))
{
+ // Only pooled bucket-join fragments have the bucket-count
parallelism cap.
+ return false;
+ }
+ Map<Long, Set<Integer>> bucketsPerWorker = new HashMap<>();
+ Map<Long, Integer> instancesPerWorker = new HashMap<>();
+ Map<Long, Integer> coresPerWorker = new HashMap<>();
+ for (AssignedJob job : instanceJobs) {
+ long workerId = job.getAssignedWorker().id();
+ bucketsPerWorker.computeIfAbsent(workerId, k -> new HashSet<>())
+ .addAll(((LocalShuffleBucketJoinAssignedJob)
job).getAssignedJoinBucketIndexes());
+ instancesPerWorker.merge(workerId, 1, Integer::sum);
+ coresPerWorker.computeIfAbsent(workerId, k ->
resolveWorkerCores(job.getAssignedWorker()));
+ }
+ // Conservative: every worker that owns buckets must clear the gain
bar. The gain is
+ // computed on EFFECTIVE parallelism (capped by the BE's executor
threads): when the
+ // bucket count already saturates the cores, adding instances cannot
speed the join
+ // up and the extra local exchange is a pure cost.
+ boolean anyBuckets = false;
+ for (Map.Entry<Long, Set<Integer>> entry :
bucketsPerWorker.entrySet()) {
+ int buckets = entry.getValue().size();
+ if (buckets == 0) {
+ continue;
+ }
+ anyBuckets = true;
+ int instances = instancesPerWorker.getOrDefault(entry.getKey(), 0);
+ int cores = coresPerWorker.getOrDefault(entry.getKey(),
Integer.MAX_VALUE);
+ if (!shouldUpgradeBucketParallelism(ratio,
+ Math.min(instances, cores), Math.min(buckets, cores))) {
+ return false;
+ }
+ }
+ return anyBuckets;
+ }
+
+ /**
+ * Effective execution threads of the worker's backend
(pipelineExecutorSize, falling
+ * back to cpuCores). Values <= 1 mean the heartbeat has not reported yet
— treat the
+ * capacity as unknown/uncapped rather than blocking the upgrade.
+ */
+ private static int resolveWorkerCores(
+
org.apache.doris.nereids.trees.plans.distribute.worker.DistributedPlanWorker
worker) {
+ if (worker instanceof
org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker) {
+ org.apache.doris.system.Backend backend =
+
((org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker)
worker).getBackend();
+ int size = backend.getPipelineExecutorSize();
+ if (size <= 1) {
Review Comment:
[P2] Honor a reported single pipeline executor
`pipeline_executor_size = 1` is a valid configured capacity: the BE reports
every positive value verbatim, but this branch treats one as ‘not reported’ and
replaces it with the machine CPU count. On a many-core BE intentionally
configured with one pipeline executor, the ratio can therefore enable two local
hash exchanges even though effective concurrency remains one, making this
default-on optimization pure overhead. Please distinguish the unreported state,
or conservatively preserve any positive reported executor size.
--
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]
