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

[fix](sequence column) fix update fail on nereids planner (#28031)

Browse Source 
1. if we set enable_fallback_to_original_planner to false, the UPDATE sql in regression case test_unique_table_sequence will fail due to：“Table test_uniq_sequence has sequence column, need to specify the sequence column”，The bug is introduced by [fix](sequence column) insert into should require sequence column in all scenario #27780
2. fix insert fail in transaction mode, which is introduced by [refactor](Nereids) let insert into compatible with legacy planner #27947
3. add cases with session variable enable_fallback_to_original_planner false
This commit is contained in:
zhannngchen
2023-12-13 19:41:41 +08:00
committed by GitHub
parent dbc2b081d7
commit 0868526f6a
17 changed files with 563 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
fe/fe-core/src/main/java/org/apache/doris/nereids/analyzer/UnboundTableSink.java
View File

@ -27,6 +27,7 @@ import org.apache.doris.nereids.trees.plans.BlockFuncDepsPropagation;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.algebra.Sink;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalSink;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.util.Utils;
@ -50,36 +51,36 @@ public class UnboundTableSink<CHILD_TYPE extends Plan> extends LogicalSink<CHILD
private final boolean temporaryPartition;
private final List<String> partitions;
private final boolean isPartialUpdate;
private final boolean isFromNativeInsertStmt;
private final DMLCommandType dmlCommandType;
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
List<String> partitions, CHILD_TYPE child) {
this(nameParts, colNames, hints, false, partitions,
false, false, Optional.empty(), Optional.empty(), child);
false, DMLCommandType.NONE, Optional.empty(), Optional.empty(), child);
}
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
boolean temporaryPartition, List<String> partitions, CHILD_TYPE child) {
this(nameParts, colNames, hints, temporaryPartition, partitions,
false, false, Optional.empty(), Optional.empty(), child);
false, DMLCommandType.NONE, Optional.empty(), Optional.empty(), child);
}
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
List<String> partitions, boolean isPartialUpdate, CHILD_TYPE child) {
this(nameParts, colNames, hints, false, partitions, isPartialUpdate, false,
this(nameParts, colNames, hints, false, partitions, isPartialUpdate, DMLCommandType.NONE,
Optional.empty(), Optional.empty(), child);
}
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
boolean temporaryPartition, List<String> partitions, boolean isPartialUpdate, CHILD_TYPE child) {
this(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate, false,
this(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate, DMLCommandType.NONE,
Optional.empty(), Optional.empty(), child);
}
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
boolean temporaryPartition, List<String> partitions,
boolean isPartialUpdate, boolean isFromNativeInsertStmt, CHILD_TYPE child) {
this(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate, isFromNativeInsertStmt,
boolean isPartialUpdate, DMLCommandType dmlCommandType, CHILD_TYPE child) {
this(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate, dmlCommandType,
Optional.empty(), Optional.empty(), child);
}
@ -88,7 +89,7 @@ public class UnboundTableSink<CHILD_TYPE extends Plan> extends LogicalSink<CHILD
*/
public UnboundTableSink(List<String> nameParts, List<String> colNames, List<String> hints,
boolean temporaryPartition, List<String> partitions,
boolean isPartialUpdate, boolean isFromNativeInsertStmt,
boolean isPartialUpdate, DMLCommandType dmlCommandType,
Optional<GroupExpression> groupExpression, Optional<LogicalProperties> logicalProperties,
CHILD_TYPE child) {
super(PlanType.LOGICAL_UNBOUND_OLAP_TABLE_SINK, ImmutableList.of(), groupExpression, logicalProperties, child);
@ -98,7 +99,7 @@ public class UnboundTableSink<CHILD_TYPE extends Plan> extends LogicalSink<CHILD
this.temporaryPartition = temporaryPartition;
this.partitions = Utils.copyRequiredList(partitions);
this.isPartialUpdate = isPartialUpdate;
this.isFromNativeInsertStmt = isFromNativeInsertStmt;
this.dmlCommandType = dmlCommandType;
}
public List<String> getColNames() {
@ -125,15 +126,15 @@ public class UnboundTableSink<CHILD_TYPE extends Plan> extends LogicalSink<CHILD
return isPartialUpdate;
}
public boolean isFromNativeInsertStmt() {
return isFromNativeInsertStmt;
public DMLCommandType getDMLCommandType() {
return dmlCommandType;
}
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1, "UnboundOlapTableSink only accepts one child");
return new UnboundTableSink<>(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate,
isFromNativeInsertStmt, groupExpression, Optional.empty(), children.get(0));
dmlCommandType, groupExpression, Optional.empty(), children.get(0));
}
@Override
@ -169,14 +170,14 @@ public class UnboundTableSink<CHILD_TYPE extends Plan> extends LogicalSink<CHILD
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
return new UnboundTableSink<>(nameParts, colNames, hints, temporaryPartition, partitions, isPartialUpdate,
isFromNativeInsertStmt, groupExpression, Optional.of(getLogicalProperties()), child());
dmlCommandType, groupExpression, Optional.of(getLogicalProperties()), child());
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
return new UnboundTableSink<>(nameParts, colNames, hints, temporaryPartition, partitions,
isPartialUpdate, isFromNativeInsertStmt, groupExpression, logicalProperties, children.get(0));
isPartialUpdate, dmlCommandType, groupExpression, logicalProperties, children.get(0));
}
@Override

3
fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
View File

@ -346,6 +346,7 @@ import org.apache.doris.nereids.trees.plans.commands.info.BulkStorageDesc;
import org.apache.doris.nereids.trees.plans.commands.info.ColumnDefinition;
import org.apache.doris.nereids.trees.plans.commands.info.CreateMTMVInfo;
import org.apache.doris.nereids.trees.plans.commands.info.CreateTableInfo;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.commands.info.DefaultValue;
import org.apache.doris.nereids.trees.plans.commands.info.DistributionDescriptor;
import org.apache.doris.nereids.trees.plans.commands.info.DropMTMVInfo;
@ -495,7 +496,7 @@ public class LogicalPlanBuilder extends DorisParserBaseVisitor<Object> {
temporaryPartition,
partitions,
ConnectContext.get().getSessionVariable().isEnableUniqueKeyPartialUpdate(),
true,
DMLCommandType.INSERT,
plan);
LogicalPlan command;
if (isOverwrite) {

13
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindSink.java
View File

@ -46,6 +46,7 @@ import org.apache.doris.nereids.trees.expressions.literal.Literal;
import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapTableSink;
import org.apache.doris.nereids.trees.plans.logical.LogicalOneRowRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
@ -101,7 +102,7 @@ public class BindSink implements AnalysisRuleFactory {
.map(NamedExpression.class::cast)
.collect(ImmutableList.toImmutableList()),
isPartialUpdate,
sink.isFromNativeInsertStmt(),
sink.getDMLCommandType(),
child);
if (isPartialUpdate) {
@ -110,7 +111,7 @@ public class BindSink implements AnalysisRuleFactory {
throw new AnalysisException("Partial update is only allowed on "
+ "unique table with merge-on-write enabled.");
}
if (sink.getColNames().isEmpty() && sink.isFromNativeInsertStmt()) {
if (sink.getColNames().isEmpty() && sink.getDMLCommandType() == DMLCommandType.INSERT) {
throw new AnalysisException("You must explicitly specify the columns to be updated when "
+ "updating partial columns using the INSERT statement.");
}
@ -168,7 +169,13 @@ public class BindSink implements AnalysisRuleFactory {
}
}
if (!haveInputSeqCol && !isPartialUpdate) {
// Don't require user to provide sequence column for partial updates,
// including the following cases:
// 1. it's a load job with `partial_columns=true`
// 2. UPDATE and DELETE, planner will automatically add these hidden columns
if (!haveInputSeqCol && !isPartialUpdate && (
boundSink.getDmlCommandType() != DMLCommandType.UPDATE
&& boundSink.getDmlCommandType() != DMLCommandType.DELETE)) {
if (!seqColInTable.isPresent() || seqColInTable.get().getDefaultValue() == null
|| !seqColInTable.get().getDefaultValue()
.equals(DefaultValue.CURRENT_TIMESTAMP)) {

2
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/implementation/LogicalOlapTableSinkToPhysicalOlapTableSink.java
View File

@ -41,7 +41,7 @@ public class LogicalOlapTableSinkToPhysicalOlapTableSink extends OneImplementati
sink.getOutputExprs(),
ctx.connectContext.getSessionVariable().isEnableSingleReplicaInsert(),
sink.isPartialUpdate(),
sink.isFromNativeInsertStmt(),
sink.getDmlCommandType(),
Optional.empty(),
sink.getLogicalProperties(),
sink.child());

3
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/DeleteCommand.java
View File

@ -30,6 +30,7 @@ import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
import org.apache.doris.nereids.trees.plans.Explainable;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@ -121,7 +122,7 @@ public class DeleteCommand extends Command implements ForwardWithSync, Explainab
// make UnboundTableSink
return new UnboundTableSink<>(nameParts, cols, ImmutableList.of(),
partitions, isPartialUpdate, logicalQuery);
false, partitions, isPartialUpdate, DMLCommandType.DELETE, logicalQuery);
}
public LogicalPlan getLogicalQuery() {

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

@ -474,7 +474,12 @@ public class InsertExecutor {
.setTimeout((int) timeoutSecond)
.setTimezone(timeZone)
.setSendBatchParallelism(sendBatchParallelism)
.setTrimDoubleQuotes(true);
.setTrimDoubleQuotes(true)
.setSequenceCol(columns.stream()
.filter(c -> Column.SEQUENCE_COL.equalsIgnoreCase(c.getName()))
.map(Column::getName)
.findFirst()
.orElse(null));
// execute begin txn
InsertStreamTxnExecutor executor = new InsertStreamTxnExecutor(txnEntry);

3
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertIntoTableCommand.java
View File

@ -36,6 +36,7 @@ import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Explainable;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapTableSink;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@ -156,7 +157,7 @@ public class InsertIntoTableCommand extends Command implements ForwardWithSync,
physicalOlapTableSink.getTargetTable(), label, planner);
insertExecutor.beginTransaction();
insertExecutor.finalizeSink(sink, physicalOlapTableSink.isPartialUpdate(),
physicalOlapTableSink.isFromNativeInsertStmt());
physicalOlapTableSink.getDmlCommandType() == DMLCommandType.INSERT);
} finally {
targetTableIf.readUnlock();
}

2
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertOverwriteTableCommand.java
View File

@ -199,7 +199,7 @@ public class InsertOverwriteTableCommand extends Command implements ForwardWithS
true,
tempPartitionNames,
sink.isPartialUpdate(),
sink.isFromNativeInsertStmt(),
sink.getDMLCommandType(),
(LogicalPlan) (sink.child(0)));
new InsertIntoTableCommand(copySink, labelName).run(ctx, executor);
if (ctx.getState().getStateType() == MysqlStateType.ERR) {

3
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/LoadCommand.java
View File

@ -44,6 +44,7 @@ import org.apache.doris.nereids.trees.expressions.literal.TinyIntLiteral;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.BulkLoadDataDesc;
import org.apache.doris.nereids.trees.plans.commands.info.BulkStorageDesc;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalCheckPolicy;
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
@ -226,7 +227,7 @@ public class LoadCommand extends Command implements ForwardWithSync {
boolean isPartialUpdate = olapTable.getEnableUniqueKeyMergeOnWrite()
&& sinkCols.size() < olapTable.getColumns().size();
return new UnboundTableSink<>(dataDesc.getNameParts(), sinkCols, ImmutableList.of(),
dataDesc.getPartitionNames(), isPartialUpdate, tvfLogicalPlan);
false, dataDesc.getPartitionNames(), isPartialUpdate, DMLCommandType.LOAD, tvfLogicalPlan);
}
private static void fillDeleteOnColumn(BulkLoadDataDesc dataDesc, OlapTable olapTable,

3
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateCommand.java
View File

@ -33,6 +33,7 @@ import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.plans.Explainable;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
@ -137,7 +138,7 @@ public class UpdateCommand extends Command implements ForwardWithSync, Explainab
// make UnboundTableSink
return new UnboundTableSink<>(nameParts, ImmutableList.of(), ImmutableList.of(),
ImmutableList.of(), isPartialUpdate, logicalQuery);
false, ImmutableList.of(), isPartialUpdate, DMLCommandType.UPDATE, logicalQuery);
}
private void checkTable(ConnectContext ctx) throws AnalysisException {

37
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/DMLCommandType.java Normal file
View File

@ -0,0 +1,37 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
package org.apache.doris.nereids.trees.plans.commands.info;
/**
* Type of DML Command.
* For Unique Key table, we need to process some
* hidden columns in different way for different DML type.
*/
public enum DMLCommandType {
// Not a DML command
NONE,
// for INSERT INTO or INSERT INTO SELECT
INSERT,
// for UPDATE
UPDATE,
// for DELETE
DELETE,
// for all other load jobs, including Stream Load, Broker Load, S3 Load
// Routine Load etc.
LOAD
}

29
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableSink.java
View File

@ -27,6 +27,7 @@ import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.PropagateFuncDeps;
import org.apache.doris.nereids.trees.plans.algebra.Sink;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.util.Utils;
@ -48,12 +49,12 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
private final List<Column> cols;
private final List<Long> partitionIds;
private final boolean isPartialUpdate;
private final boolean isFromNativeInsertStmt;
private final DMLCommandType dmlCommandType;
public LogicalOlapTableSink(Database database, OlapTable targetTable, List<Column> cols, List<Long> partitionIds,
List<NamedExpression> outputExprs, boolean isPartialUpdate, boolean isFromNativeInsertStmt,
List<NamedExpression> outputExprs, boolean isPartialUpdate, DMLCommandType dmlCommandType,
CHILD_TYPE child) {
this(database, targetTable, cols, partitionIds, outputExprs, isPartialUpdate, isFromNativeInsertStmt,
this(database, targetTable, cols, partitionIds, outputExprs, isPartialUpdate, dmlCommandType,
Optional.empty(), Optional.empty(), child);
}
@ -62,14 +63,14 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
*/
public LogicalOlapTableSink(Database database, OlapTable targetTable, List<Column> cols,
List<Long> partitionIds, List<NamedExpression> outputExprs, boolean isPartialUpdate,
boolean isFromNativeInsertStmt, Optional<GroupExpression> groupExpression,
DMLCommandType dmlCommandType, Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, CHILD_TYPE child) {
super(PlanType.LOGICAL_OLAP_TABLE_SINK, outputExprs, groupExpression, logicalProperties, child);
this.database = Objects.requireNonNull(database, "database != null in LogicalOlapTableSink");
this.targetTable = Objects.requireNonNull(targetTable, "targetTable != null in LogicalOlapTableSink");
this.cols = Utils.copyRequiredList(cols);
this.isPartialUpdate = isPartialUpdate;
this.isFromNativeInsertStmt = isFromNativeInsertStmt;
this.dmlCommandType = dmlCommandType;
this.partitionIds = Utils.copyRequiredList(partitionIds);
}
@ -78,14 +79,14 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
.map(NamedExpression.class::cast)
.collect(ImmutableList.toImmutableList());
return new LogicalOlapTableSink<>(database, targetTable, cols, partitionIds, output, isPartialUpdate,
isFromNativeInsertStmt, Optional.empty(), Optional.empty(), child);
dmlCommandType, Optional.empty(), Optional.empty(), child);
}
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1, "LogicalOlapTableSink only accepts one child");
return new LogicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, isPartialUpdate,
isFromNativeInsertStmt, Optional.empty(), Optional.empty(), children.get(0));
dmlCommandType, Optional.empty(), Optional.empty(), children.get(0));
}
public Database getDatabase() {
@ -108,8 +109,8 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
return isPartialUpdate;
}
public boolean isFromNativeInsertStmt() {
return isFromNativeInsertStmt;
public DMLCommandType getDmlCommandType() {
return dmlCommandType;
}
@Override
@ -124,7 +125,7 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
return false;
}
LogicalOlapTableSink<?> that = (LogicalOlapTableSink<?>) o;
return isPartialUpdate == that.isPartialUpdate && isFromNativeInsertStmt == that.isFromNativeInsertStmt
return isPartialUpdate == that.isPartialUpdate && dmlCommandType == that.dmlCommandType
&& Objects.equals(database, that.database)
&& Objects.equals(targetTable, that.targetTable) && Objects.equals(cols, that.cols)
&& Objects.equals(partitionIds, that.partitionIds);
@ -133,7 +134,7 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
@Override
public int hashCode() {
return Objects.hash(super.hashCode(), database, targetTable, cols, partitionIds,
isPartialUpdate, isFromNativeInsertStmt);
isPartialUpdate, dmlCommandType);
}
@Override
@ -145,7 +146,7 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
"cols", cols,
"partitionIds", partitionIds,
"isPartialUpdate", isPartialUpdate,
"isFromNativeInsertStmt", isFromNativeInsertStmt
"dmlCommandType", dmlCommandType
);
}
@ -157,13 +158,13 @@ public class LogicalOlapTableSink<CHILD_TYPE extends Plan> extends LogicalSink<C
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
return new LogicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, isPartialUpdate,
isFromNativeInsertStmt, groupExpression, Optional.of(getLogicalProperties()), child());
dmlCommandType, groupExpression, Optional.of(getLogicalProperties()), child());
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
return new LogicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, isPartialUpdate,
isFromNativeInsertStmt, groupExpression, logicalProperties, children.get(0));
dmlCommandType, groupExpression, logicalProperties, children.get(0));
}
}

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

@ -34,6 +34,7 @@ import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
import org.apache.doris.nereids.trees.plans.algebra.Sink;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.nereids.util.Utils;
import org.apache.doris.statistics.Statistics;
@ -60,17 +61,17 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
private final List<Long> partitionIds;
private final boolean singleReplicaLoad;
private final boolean isPartialUpdate;
private final boolean isFromNativeInsertStmt;
private final DMLCommandType dmlCommandType;
/**
* Constructor
*/
public PhysicalOlapTableSink(Database database, OlapTable targetTable, List<Column> cols, List<Long> partitionIds,
List<NamedExpression> outputExprs, boolean singleReplicaLoad,
boolean isPartialUpdate, boolean isFromNativeInsertStmt,
boolean isPartialUpdate, DMLCommandType dmlCommandType,
Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties, CHILD_TYPE child) {
this(database, targetTable, cols, partitionIds, outputExprs,
singleReplicaLoad, isPartialUpdate, isFromNativeInsertStmt,
singleReplicaLoad, isPartialUpdate, dmlCommandType,
groupExpression, logicalProperties, PhysicalProperties.GATHER, null, child);
}
@ -79,7 +80,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
*/
public PhysicalOlapTableSink(Database database, OlapTable targetTable, List<Column> cols, List<Long> partitionIds,
List<NamedExpression> outputExprs, boolean singleReplicaLoad,
boolean isPartialUpdate, boolean isFromNativeInsertStmt,
boolean isPartialUpdate, DMLCommandType dmlCommandType,
Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties,
PhysicalProperties physicalProperties, Statistics statistics, CHILD_TYPE child) {
super(PlanType.PHYSICAL_OLAP_TABLE_SINK, outputExprs, groupExpression,
@ -90,7 +91,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
this.partitionIds = Utils.copyRequiredList(partitionIds);
this.singleReplicaLoad = singleReplicaLoad;
this.isPartialUpdate = isPartialUpdate;
this.isFromNativeInsertStmt = isFromNativeInsertStmt;
this.dmlCommandType = dmlCommandType;
}
public Database getDatabase() {
@ -117,15 +118,15 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
return isPartialUpdate;
}
public boolean isFromNativeInsertStmt() {
return isFromNativeInsertStmt;
public DMLCommandType getDmlCommandType() {
return dmlCommandType;
}
@Override
public Plan withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1, "PhysicalOlapTableSink only accepts one child");
return new PhysicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs,
singleReplicaLoad, isPartialUpdate, isFromNativeInsertStmt, groupExpression,
singleReplicaLoad, isPartialUpdate, dmlCommandType, groupExpression,
getLogicalProperties(), physicalProperties, statistics, children.get(0));
}
@ -140,7 +141,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
PhysicalOlapTableSink<?> that = (PhysicalOlapTableSink<?>) o;
return singleReplicaLoad == that.singleReplicaLoad
&& isPartialUpdate == that.isPartialUpdate
&& isFromNativeInsertStmt == that.isFromNativeInsertStmt
&& dmlCommandType == that.dmlCommandType
&& Objects.equals(database, that.database)
&& Objects.equals(targetTable, that.targetTable)
&& Objects.equals(cols, that.cols)
@ -150,7 +151,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
@Override
public int hashCode() {
return Objects.hash(database, targetTable, cols, partitionIds, singleReplicaLoad,
isPartialUpdate, isFromNativeInsertStmt);
isPartialUpdate, dmlCommandType);
}
@Override
@ -163,7 +164,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
"partitionIds", partitionIds,
"singleReplicaLoad", singleReplicaLoad,
"isPartialUpdate", isPartialUpdate,
"isFromNativeInsertStmt", isFromNativeInsertStmt
"dmlCommandType", dmlCommandType
);
}
@ -193,20 +194,20 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
@Override
public Plan withGroupExpression(Optional<GroupExpression> groupExpression) {
return new PhysicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, singleReplicaLoad,
isPartialUpdate, isFromNativeInsertStmt, groupExpression, getLogicalProperties(), child());
isPartialUpdate, dmlCommandType, groupExpression, getLogicalProperties(), child());
}
@Override
public Plan withGroupExprLogicalPropChildren(Optional<GroupExpression> groupExpression,
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
return new PhysicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, singleReplicaLoad,
isPartialUpdate, isFromNativeInsertStmt, groupExpression, logicalProperties.get(), children.get(0));
isPartialUpdate, dmlCommandType, groupExpression, logicalProperties.get(), children.get(0));
}
@Override
public PhysicalPlan withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties, Statistics statistics) {
return new PhysicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, singleReplicaLoad,
isPartialUpdate, isFromNativeInsertStmt, groupExpression, getLogicalProperties(),
isPartialUpdate, dmlCommandType, groupExpression, getLogicalProperties(),
physicalProperties, statistics, child());
}
@ -247,7 +248,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
@Override
public PhysicalOlapTableSink<Plan> resetLogicalProperties() {
return new PhysicalOlapTableSink<>(database, targetTable, cols, partitionIds, outputExprs, singleReplicaLoad,
isPartialUpdate, isFromNativeInsertStmt, groupExpression,
isPartialUpdate, dmlCommandType, groupExpression,
null, physicalProperties, statistics, child());
}
}