This is an automated email from the ASF dual-hosted git repository.
yiguolei 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 a50bba39e4e Revert "[Improvement](shuffle) Use a knob to decide
whether a serial … (#44076)
a50bba39e4e is described below
commit a50bba39e4e8e334d2e40cdeb47ddb18d83dbc5f
Author: yiguolei <[email protected]>
AuthorDate: Sun Nov 17 00:24:29 2024 +0800
Revert "[Improvement](shuffle) Use a knob to decide whether a serial …
(#44076)
---
.../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, 17 insertions(+), 43 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 d904397a305..97d46b109b7 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,7 +25,6 @@ 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;
@@ -166,10 +165,6 @@ 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) {
@@ -229,17 +224,11 @@ public class ExchangeNode extends PlanNode {
*/
@Override
public boolean isSerialOperator() {
- return (ConnectContext.get() != null &&
ConnectContext.get().getSessionVariable().isUseSerialExchange()
- || partitionType == TPartitionType.UNPARTITIONED) && mergeInfo
!= null;
+ return 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 fe386acdaf2..0ebd023ed41 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,13 +511,15 @@ 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.
- // 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()));
+ && sink instanceof DataStreamSink &&
!planRoot.isSerialOperator()
+ && planRoot.hasSerialChildren();
}
- public boolean hasSerialScanNode() {
- return planRoot.hasSerialScanChildren();
+ public int getNumBackends() {
+ return numBackends;
+ }
+
+ public void setNumBackends(int numBackends) {
+ this.numBackends = numBackends;
}
}
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 73768435154..14bd34e93e1 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,11 +1388,4 @@ 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 85c8de68b8c..b392075be9b 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
@@ -855,9 +855,4 @@ 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 31699c0c245..9829f88cf52 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
@@ -1886,11 +1886,17 @@ 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 =
fragment.useSerialSource(context);
- if (ignoreStorageDataDistribution) {
+ boolean ignoreStorageDataDistribution =
node.isPresent()
+ && fragment.useSerialSource(context);
+ if (node.isPresent() && 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 8ac56acfbaf..4c9feab7dd4 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,8 +264,6 @@ 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.
@@ -1109,10 +1107,6 @@ 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优化",
@@ -2329,7 +2323,6 @@ 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();
@@ -4539,10 +4532,6 @@ public class SessionVariable implements Serializable,
Writable {
}
}
- public boolean isUseSerialExchange() {
- return useSerialExchange && getEnableLocalExchange();
- }
-
public boolean isEnableCooldownReplicaAffinity() {
return enableCooldownReplicaAffinity;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
