You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by kx...@apache.org on 2023/11/13 09:11:13 UTC
(doris) branch branch-2.0 updated: [pipeline](exec) disable shared scan in default and disable shared scan in limit with where scan (#25952) (#26815)
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-2.0 by this push:
new d25f02cd25e [pipeline](exec) disable shared scan in default and disable shared scan in limit with where scan (#25952) (#26815)
d25f02cd25e is described below
commit d25f02cd25ead260879742107d518bb801c8d859
Author: HappenLee <ha...@hotmail.com>
AuthorDate: Mon Nov 13 17:11:04 2023 +0800
[pipeline](exec) disable shared scan in default and disable shared scan in limit with where scan (#25952) (#26815)
---
.../main/java/org/apache/doris/common/Config.java | 3 +++
.../org/apache/doris/planner/OlapScanNode.java | 10 ++++----
.../java/org/apache/doris/planner/ScanNode.java | 23 ++++++++++++++++--
.../main/java/org/apache/doris/qe/Coordinator.java | 27 ++++++++++++++--------
.../java/org/apache/doris/qe/SessionVariable.java | 10 ++++++++
5 files changed, 56 insertions(+), 17 deletions(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 35b8ab838df..a361cd7291a 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -1592,6 +1592,9 @@ public class Config extends ConfigBase {
@ConfField(mutable = true)
public static boolean disable_shared_scan = false;
+ @ConfField(mutable = true, expType = ExperimentalType.EXPERIMENTAL)
+ public static boolean enable_cpu_hard_limit = false;
+
@ConfField(mutable = false, masterOnly = true)
public static int backend_rpc_timeout_ms = 60000; // 1 min
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 5c85baca318..b77a3bb285f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -1255,11 +1255,9 @@ public class OlapScanNode extends ScanNode {
public int getNumInstances() {
// In pipeline exec engine, the instance num equals be_num * parallel instance.
// so here we need count distinct be_num to do the work. make sure get right instance
- if (ConnectContext.get().getSessionVariable().getEnablePipelineEngine()) {
- int parallelInstance = ConnectContext.get().getSessionVariable().getParallelExecInstanceNum();
- long numBackend = scanRangeLocations.stream().flatMap(rangeLoc -> rangeLoc.getLocations().stream())
- .map(loc -> loc.backend_id).distinct().count();
- return (int) (parallelInstance * numBackend);
+ if (ConnectContext.get().getSessionVariable().getEnablePipelineEngine()
+ && ConnectContext.get().getSessionVariable().getEnableSharedScan()) {
+ return ConnectContext.get().getSessionVariable().getParallelExecInstanceNum();
}
return scanRangeLocations.size();
}
@@ -1316,7 +1314,7 @@ public class OlapScanNode extends ScanNode {
// If scan is key search, should not enable the shared scan opt to prevent the performance problem
// 1. where contain the eq or in expr of key column slot
// 2. key column slot is distribution column and first column
- public boolean isKeySearch() {
+ protected boolean isKeySearch() {
List<SlotRef> whereSlot = Lists.newArrayList();
for (Expr conjunct : conjuncts) {
if (conjunct instanceof BinaryPredicate) {
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 8d0cb1152df..9398dabcd36 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
@@ -42,6 +42,7 @@ import org.apache.doris.common.NotImplementedException;
import org.apache.doris.common.UserException;
import org.apache.doris.nereids.glue.translator.PlanTranslatorContext;
import org.apache.doris.planner.external.FederationBackendPolicy;
+import org.apache.doris.qe.ConnectContext;
import org.apache.doris.spi.Split;
import org.apache.doris.statistics.StatisticalType;
import org.apache.doris.statistics.query.StatsDelta;
@@ -146,6 +147,13 @@ public abstract class ScanNode extends PlanNode {
*/
public abstract List<TScanRangeLocations> getScanRangeLocations(long maxScanRangeLength);
+ // If scan is key search, should not enable the shared scan opt to prevent the performance problem
+ // 1. where contain the eq or in expr of key column slot
+ // 2. key column slot is distribution column and first column
+ protected boolean isKeySearch() {
+ return false;
+ }
+
/**
* Update required_slots in scan node contexts. This is called after Nereids planner do the projection.
* In the projection process, some slots may be removed. So call this to update the slots info.
@@ -643,7 +651,18 @@ public abstract class ScanNode extends PlanNode {
return scanRangeLocation;
}
- public boolean isKeySearch() {
- return false;
+ // some scan should not enable the shared scan opt to prevent the performance problem
+ // 1. is key search
+ // 2. session variable not enable_shared_scan
+ public boolean shouldDisableSharedScan() {
+ boolean enableShardScan = false;
+ if (ConnectContext.get() != null && ConnectContext.get().getSessionVariable() != null) {
+ enableShardScan = ConnectContext.get().getSessionVariable().getEnableSharedScan();
+ }
+ return isKeySearch() || !enableShardScan;
+ }
+
+ public boolean haveLimitAndConjunts() {
+ return hasLimit() && !conjuncts.isEmpty();
}
}
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 4ad725ae663..af8eb518e9d 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
@@ -1724,21 +1724,24 @@ public class Coordinator implements CoordInterface {
}).findFirst();
// disable shared scan optimization if one of conditions below is met:
- // 1. Use non-pipeline engine
- // 2. Number of scan ranges is larger than instances
- // 3. This fragment has a colocated scan node
- // 4. This fragment has a FileScanNode
- // 5. Disable shared scan optimization by session variable
- if (!enablePipelineEngine || perNodeScanRanges.size() > parallelExecInstanceNum
- || (node.isPresent() && node.get().getShouldColoScan())
+ // 1. Use non-pipeline or pipelineX engine
+ // 2. This fragment has a colocated scan node
+ // 3. This fragment has a FileScanNode
+ // 4. Disable shared scan optimization by session variable
+ if (!enablePipelineEngine || (node.isPresent() && node.get().getShouldColoScan())
|| (node.isPresent() && node.get() instanceof FileScanNode)
- || (node.isPresent() && node.get().isKeySearch())
- || Config.disable_shared_scan) {
+ || (node.isPresent() && node.get().shouldDisableSharedScan())) {
int expectedInstanceNum = 1;
if (parallelExecInstanceNum > 1) {
//the scan instance num should not larger than the tablets num
expectedInstanceNum = Math.min(perNodeScanRanges.size(), parallelExecInstanceNum);
}
+ // if have limit and conjunts, only need 1 instance to save cpu and
+ // mem resource
+ if (node.isPresent() && node.get().haveLimitAndConjunts()) {
+ expectedInstanceNum = 1;
+ }
+
perInstanceScanRanges = ListUtil.splitBySize(perNodeScanRanges,
expectedInstanceNum);
sharedScanOpts = Collections.nCopies(perInstanceScanRanges.size(), false);
@@ -1746,6 +1749,12 @@ public class Coordinator implements CoordInterface {
int expectedInstanceNum = Math.min(parallelExecInstanceNum,
leftMostNode.getNumInstances());
expectedInstanceNum = Math.max(expectedInstanceNum, 1);
+ // if have limit and conjunts, only need 1 instance to save cpu and
+ // mem resource
+ if (node.isPresent() && node.get().haveLimitAndConjunts()) {
+ expectedInstanceNum = 1;
+ }
+
perInstanceScanRanges = Collections.nCopies(expectedInstanceNum, perNodeScanRanges);
sharedScanOpts = Collections.nCopies(perInstanceScanRanges.size(), true);
}
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 7425c97d7bb..fb02d9cc7cd 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
@@ -205,6 +205,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String ENABLE_PIPELINE_ENGINE = "enable_pipeline_engine";
+ public static final String ENABLE_SHARED_SCAN = "enable_shared_scan";
+
public static final String ENABLE_AGG_STATE = "enable_agg_state";
public static final String ENABLE_RPC_OPT_FOR_PIPELINE = "enable_rpc_opt_for_pipeline";
@@ -696,6 +698,10 @@ public class SessionVariable implements Serializable, Writable {
expType = ExperimentalType.EXPERIMENTAL)
private boolean enablePipelineEngine = true;
+ @VariableMgr.VarAttr(name = ENABLE_SHARED_SCAN, fuzzy = false, expType = ExperimentalType.EXPERIMENTAL,
+ needForward = true)
+ private boolean enableSharedScan = false;
+
@VariableMgr.VarAttr(name = ENABLE_AGG_STATE, fuzzy = false, expType = ExperimentalType.EXPERIMENTAL)
public boolean enableAggState = false;
@@ -2617,6 +2623,10 @@ public class SessionVariable implements Serializable, Writable {
return enablePipelineEngine;
}
+ public boolean getEnableSharedScan() {
+ return enableSharedScan;
+ }
+
public static boolean enablePipelineEngine() {
ConnectContext connectContext = ConnectContext.get();
if (connectContext == null) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org