database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[feature](shuffle) enable strict consistency dml by default (#32958) (#34641)

Browse Source
This commit is contained in:
Kaijie Chen
2024-05-10 14:31:50 +08:00
committed by GitHub
parent 5a3107442a
commit 07207b7b51
5 changed files with 45 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
fe/fe-common/src/main/java/org/apache/doris/common/Config.java
View File

@ -636,6 +636,12 @@ public class Config extends ConfigBase {
varType = VariableAnnotation.EXPERIMENTAL)
public static boolean enable_single_replica_load = false;
@ConfField(mutable = true, masterOnly = true, description = {
"对于 tablet 数量小于该数目的 DUPLICATE KEY 表，将不会启用 shuffle",
"Shuffle won't be enabled for DUPLICATE KEY tables if its tablet num is lower than this number"},
varType = VariableAnnotation.EXPERIMENTAL)
public static int min_tablets_for_dup_table_shuffle = 64;
@ConfField(mutable = true, masterOnly = true, description = {
"单个数据库最大并发运行的事务数，包括 prepare 和 commit 事务。",
"Maximum concurrent running txn num including prepare, commit txns under a single db.",

7
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapTableSink.java
View File

@ -21,8 +21,10 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.DistributionInfo;
import org.apache.doris.catalog.HashDistributionInfo;
import org.apache.doris.catalog.KeysType;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.RandomDistributionInfo;
import org.apache.doris.common.Config;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.memo.GroupExpression;
import org.apache.doris.nereids.properties.LogicalProperties;
@ -199,6 +201,11 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalTabl
if (targetTable.isPartitionDistributed()) {
DistributionInfo distributionInfo = targetTable.getDefaultDistributionInfo();
if (distributionInfo instanceof HashDistributionInfo) {
// Do not enable shuffle for duplicate key tables when its tablet num is less than threshold.
if (targetTable.getKeysType() == KeysType.DUP_KEYS
&& distributionInfo.getBucketNum() < Config.min_tablets_for_dup_table_shuffle) {
return PhysicalProperties.ANY;
}
return PhysicalProperties.TABLET_ID_SHUFFLE;
} else if (distributionInfo instanceof RandomDistributionInfo) {
return PhysicalProperties.ANY;

10
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
View File

@ -893,7 +893,7 @@ public class SessionVariable implements Serializable, Writable {
public boolean enableNereidsDmlWithPipeline = true;
@VariableMgr.VarAttr(name = ENABLE_STRICT_CONSISTENCY_DML, needForward = true)
public boolean enableStrictConsistencyDml = false;
public boolean enableStrictConsistencyDml = true;
@VariableMgr.VarAttr(name = ENABLE_VECTORIZED_ENGINE, varType = VariableAnnotation.EXPERIMENTAL_ONLINE)
public boolean enableVectorizedEngine = true;
@ -3457,6 +3457,14 @@ public class SessionVariable implements Serializable, Writable {
this.dumpNereidsMemo = dumpNereidsMemo;
}
public boolean isEnableStrictConsistencyDml() {
return this.enableStrictConsistencyDml;
}
public void setEnableStrictConsistencyDml(boolean value) {
this.enableStrictConsistencyDml = value;
}
public void disableStrictConsistencyDmlOnce() throws DdlException {
if (!enableStrictConsistencyDml) {
return;

2
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
View File

@ -3127,6 +3127,8 @@ public class StmtExecutor {
try {
if (sessionVariable.isEnableNereidsPlanner()) {
try {
// disable shuffle for http stream (only 1 sink)
sessionVariable.disableStrictConsistencyDmlOnce();
httpStreamParams = generateHttpStreamNereidsPlan(queryId);
} catch (NereidsException | ParseException e) {
if (context.getMinidump() != null && context.getMinidump().toString(4) != null) {

22
fe/fe-core/src/test/java/org/apache/doris/planner/PlannerTest.java
View File

@ -701,10 +701,12 @@ public class PlannerTest extends TestWithFeService {
connectContext.getSessionVariable().setEnableNereidsPlanner(v);
}
// 1. should not contains exchange node in new planner
// 2. should not contains exchange node in new planner
v = connectContext.getSessionVariable().isEnableNereidsPlanner();
boolean v2 = connectContext.getSessionVariable().isEnableStrictConsistencyDml();
try {
connectContext.getSessionVariable().setEnableNereidsPlanner(true);
connectContext.getSessionVariable().setEnableStrictConsistencyDml(false);
String sql1 = "explain insert into db1.tbl1 select * from db1.tbl1";
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
@ -713,6 +715,24 @@ public class PlannerTest extends TestWithFeService {
Assertions.assertFalse(plan1.contains("VEXCHANGE"));
} finally {
connectContext.getSessionVariable().setEnableNereidsPlanner(v);
connectContext.getSessionVariable().setEnableStrictConsistencyDml(v2);
}
// 3. should contain exchange node in new planner if enable strict consistency dml
v = connectContext.getSessionVariable().isEnableNereidsPlanner();
v2 = connectContext.getSessionVariable().isEnableStrictConsistencyDml();
try {
connectContext.getSessionVariable().setEnableNereidsPlanner(true);
connectContext.getSessionVariable().setEnableStrictConsistencyDml(true);
String sql1 = "explain insert into db1.tbl1 select * from db1.tbl1";
StmtExecutor stmtExecutor1 = new StmtExecutor(connectContext, sql1);
stmtExecutor1.execute();
Planner planner1 = stmtExecutor1.planner();
String plan1 = planner1.getExplainString(new ExplainOptions(false, false, false));
Assertions.assertTrue(plan1.contains("VEXCHANGE"));
} finally {
connectContext.getSessionVariable().setEnableNereidsPlanner(v);
connectContext.getSessionVariable().setEnableStrictConsistencyDml(v2);
}
}
}