This is an automated email from the ASF dual-hosted git repository.
gabriellee pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new cb5a0b7649a branch-3.0: [Improvement](shuffle) Use a knob to decide
whether a serial exchange… #44676 (#44731)
cb5a0b7649a is described below
commit cb5a0b7649afa21912bc6859b9a1b6efcdeee43f
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Mon Dec 2 11:46:43 2024 +0800
branch-3.0: [Improvement](shuffle) Use a knob to decide whether a serial
exchange… #44676 (#44731)
Cherry-picked from #44676
Co-authored-by: Gabriel <[email protected]>
---
.../main/java/org/apache/doris/planner/ExchangeNode.java | 13 ++++++++++++-
.../main/java/org/apache/doris/planner/PlanFragment.java | 14 ++++++--------
.../src/main/java/org/apache/doris/planner/PlanNode.java | 7 +++++++
.../src/main/java/org/apache/doris/planner/ScanNode.java | 5 +++++
.../src/main/java/org/apache/doris/qe/Coordinator.java | 10 ++--------
.../src/main/java/org/apache/doris/qe/SessionVariable.java | 11 +++++++++++
6 files changed, 43 insertions(+), 17 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
index 97d46b109b7..d904397a305 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ExchangeNode.java
@@ -25,6 +25,7 @@ import org.apache.doris.analysis.SortInfo;
import org.apache.doris.analysis.TupleDescriptor;
import org.apache.doris.analysis.TupleId;
import org.apache.doris.common.UserException;
+import org.apache.doris.qe.ConnectContext;
import org.apache.doris.statistics.StatisticalType;
import org.apache.doris.statistics.StatsRecursiveDerive;
import org.apache.doris.thrift.TExchangeNode;
@@ -165,6 +166,10 @@ public class ExchangeNode extends PlanNode {
@Override
protected void toThrift(TPlanNode msg) {
+ // If this fragment has another scan node, this exchange node is
serial or not should be decided by the scan
+ // node.
+ msg.setIsSerialOperator((isSerialOperator() ||
fragment.hasSerialScanNode())
+ && fragment.useSerialSource(ConnectContext.get()));
msg.node_type = TPlanNodeType.EXCHANGE_NODE;
msg.exchange_node = new TExchangeNode();
for (TupleId tid : tupleIds) {
@@ -224,11 +229,17 @@ public class ExchangeNode extends PlanNode {
*/
@Override
public boolean isSerialOperator() {
- return partitionType == TPartitionType.UNPARTITIONED && mergeInfo !=
null;
+ return (ConnectContext.get() != null &&
ConnectContext.get().getSessionVariable().isUseSerialExchange()
+ || partitionType == TPartitionType.UNPARTITIONED) && mergeInfo
!= null;
}
@Override
public boolean hasSerialChildren() {
return isSerialOperator();
}
+
+ @Override
+ public boolean hasSerialScanChildren() {
+ return false;
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
index 0ebd023ed41..fe386acdaf2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
@@ -511,15 +511,13 @@ public class PlanFragment extends TreeNode<PlanFragment> {
&& !hasNullAwareLeftAntiJoin()
// If planRoot is not a serial operator and has serial
children, we can use serial source and improve
// parallelism of non-serial operators.
- && sink instanceof DataStreamSink &&
!planRoot.isSerialOperator()
- && planRoot.hasSerialChildren();
+ // For bucket shuffle / colocate join fragment, always use
serial source if the bucket scan nodes are
+ // serial.
+ && (hasSerialScanNode() || (sink instanceof DataStreamSink &&
!planRoot.isSerialOperator()
+ && planRoot.hasSerialChildren()));
}
- public int getNumBackends() {
- return numBackends;
- }
-
- public void setNumBackends(int numBackends) {
- this.numBackends = numBackends;
+ public boolean hasSerialScanNode() {
+ return planRoot.hasSerialScanChildren();
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index 14bd34e93e1..73768435154 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -1388,4 +1388,11 @@ public abstract class PlanNode extends
TreeNode<PlanNode> implements PlanStats {
}
return children.stream().allMatch(PlanNode::hasSerialChildren);
}
+
+ public boolean hasSerialScanChildren() {
+ if (children.isEmpty()) {
+ return false;
+ }
+ return children.stream().anyMatch(PlanNode::hasSerialScanChildren);
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
index a2583868346..b4033a0535e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/ScanNode.java
@@ -861,4 +861,9 @@ public abstract class ScanNode extends PlanNode implements
SplitGenerator {
<
ConnectContext.get().getSessionVariable().getParallelExecInstanceNum() *
numScanBackends()
|| (ConnectContext.get() != null &&
ConnectContext.get().getSessionVariable().isForceToLocalShuffle());
}
+
+ @Override
+ public boolean hasSerialScanChildren() {
+ return isSerialOperator();
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 833fec1b5a0..78493a46ad1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -1890,17 +1890,11 @@ public class Coordinator implements CoordInterface {
return scanNode.getId().asInt() == planNodeId;
}).findFirst();
- /**
- * Ignore storage data distribution iff:
- * 1. `parallelExecInstanceNum * numBackends` is
larger than scan ranges.
- * 2. Use Nereids planner.
- */
boolean sharedScan = true;
int expectedInstanceNum =
Math.min(parallelExecInstanceNum,
leftMostNode.getNumInstances());
- boolean ignoreStorageDataDistribution =
node.isPresent()
- && fragment.useSerialSource(context);
- if (node.isPresent() && ignoreStorageDataDistribution)
{
+ boolean ignoreStorageDataDistribution =
fragment.useSerialSource(context);
+ if (ignoreStorageDataDistribution) {
expectedInstanceNum =
Math.max(expectedInstanceNum, 1);
// if have limit and no conjuncts, only need 1
instance to save cpu and
// mem resource
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 46f167c3446..265de4681f8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -264,6 +264,8 @@ public class SessionVariable implements Serializable,
Writable {
public static final String IGNORE_STORAGE_DATA_DISTRIBUTION =
"ignore_storage_data_distribution";
+ public static final String USE_SERIAL_EXCHANGE = "use_serial_exchange";
+
public static final String ENABLE_PARALLEL_SCAN = "enable_parallel_scan";
// Limit the max count of scanners to prevent generate too many scanners.
@@ -1107,6 +1109,10 @@ public class SessionVariable implements Serializable,
Writable {
varType = VariableAnnotation.EXPERIMENTAL, needForward = true)
private boolean ignoreStorageDataDistribution = true;
+ @VariableMgr.VarAttr(name = USE_SERIAL_EXCHANGE, fuzzy = true,
+ varType = VariableAnnotation.EXPERIMENTAL, needForward = true)
+ private boolean useSerialExchange = false;
+
@VariableMgr.VarAttr(
name = ENABLE_LOCAL_SHUFFLE, fuzzy = false, varType =
VariableAnnotation.EXPERIMENTAL,
description = {"是否在pipelineX引擎上开启local shuffle优化",
@@ -2327,6 +2333,7 @@ public class SessionVariable implements Serializable,
Writable {
this.parallelPrepareThreshold = random.nextInt(32) + 1;
this.enableCommonExprPushdown = random.nextBoolean();
this.enableLocalExchange = random.nextBoolean();
+ this.useSerialExchange = random.nextBoolean();
// This will cause be dead loop, disable it first
// this.disableJoinReorder = random.nextBoolean();
this.enableCommonExpPushDownForInvertedIndex = random.nextBoolean();
@@ -4535,6 +4542,10 @@ public class SessionVariable implements Serializable,
Writable {
return enableCooldownReplicaAffinity;
}
+ public boolean isUseSerialExchange() {
+ return useSerialExchange && getEnableLocalExchange();
+ }
+
public void setDisableInvertedIndexV1ForVaraint(boolean
disableInvertedIndexV1ForVaraint) {
this.disableInvertedIndexV1ForVaraint =
disableInvertedIndexV1ForVaraint;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
