You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2020/12/11 04:01:50 UTC
[incubator-doris] branch master updated: [Bug][MultiLoad] Fix
multiload missing userinfo and rebase error (#5058)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-doris.git
The following commit(s) were added to refs/heads/master by this push:
new 1267d6b [Bug][MultiLoad] Fix multiload missing userinfo and rebase error (#5058)
1267d6b is described below
commit 1267d6bf6618d238a4c224a4b999f0aed4c3698a
Author: Zhengguo Yang <ya...@gmail.com>
AuthorDate: Fri Dec 11 12:01:32 2020 +0800
[Bug][MultiLoad] Fix multiload missing userinfo and rebase error (#5058)
---
.../org/apache/doris/analysis/DataDescription.java | 412 ++++++++++-----------
.../java/org/apache/doris/analysis/LoadStmt.java | 2 +-
.../org/apache/doris/planner/BrokerScanNode.java | 31 +-
.../apache/doris/planner/MultiLoadScanNode.java | 39 --
.../java/org/apache/doris/qe/MultiLoadMgr.java | 10 +-
.../java/org/apache/doris/task/StreamLoadTask.java | 2 +-
gensrc/script/gen_builtins_functions.py | 2 +-
7 files changed, 226 insertions(+), 272 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
index e1dabcd..64cfd76 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DataDescription.java
@@ -34,15 +34,6 @@ import org.apache.doris.load.loadv2.LoadTask;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TNetworkAddress;
-
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Strings;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -53,6 +44,14 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
// used to describe data info which is needed to import.
//
// data_desc:
@@ -99,20 +98,16 @@ public class DataDescription {
private final ColumnSeparator columnSeparator;
private final String fileFormat;
private final boolean isNegative;
-
- // this only used in multi load, all filePaths is file not dir
- private List<Long> fileSize;
-
- // column names of source files
- private List<String> fileFieldNames;
// column names in the path
private final List<String> columnsFromPath;
// save column mapping in SET(xxx = xxx) clause
private final List<Expr> columnMappingList;
private final Expr whereExpr;
-
private final String srcTableName;
-
+ // this only used in multi load, all filePaths is file not dir
+ private List<Long> fileSize;
+ // column names of source files
+ private List<String> fileFieldNames;
// Used for mini load
private TNetworkAddress beAddr;
private String lineDelimiter;
@@ -132,12 +127,12 @@ public class DataDescription {
* For hadoop load, this param is also used to persistence.
* The function in this param is copied from 'parsedColumnExprList'
*/
- private Map<String, Pair<String, List<String>>> columnToHadoopFunction = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
+ private final Map<String, Pair<String, List<String>>> columnToHadoopFunction = Maps.newTreeMap(String.CASE_INSENSITIVE_ORDER);
private boolean isHadoopLoad = false;
private LoadTask.MergeType mergeType = LoadTask.MergeType.APPEND;
- private Expr deleteCondition;
+ private final Expr deleteCondition;
public DataDescription(String tableName,
PartitionNames partitionNames,
@@ -204,6 +199,179 @@ public class DataDescription {
this.deleteCondition = deleteCondition;
}
+ public static void validateMappingFunction(String functionName, List<String> args,
+ Map<String, String> columnNameMap,
+ Column mappingColumn, boolean isHadoopLoad) throws AnalysisException {
+ if (functionName.equalsIgnoreCase("alignment_timestamp")) {
+ validateAlignmentTimestamp(args, columnNameMap);
+ } else if (functionName.equalsIgnoreCase("strftime")) {
+ validateStrftime(args, columnNameMap);
+ } else if (functionName.equalsIgnoreCase("time_format")) {
+ validateTimeFormat(args, columnNameMap);
+ } else if (functionName.equalsIgnoreCase("default_value")) {
+ validateDefaultValue(args, mappingColumn);
+ } else if (functionName.equalsIgnoreCase("md5sum")) {
+ validateMd5sum(args, columnNameMap);
+ } else if (functionName.equalsIgnoreCase("replace_value")) {
+ validateReplaceValue(args, mappingColumn);
+ } else if (functionName.equalsIgnoreCase(FunctionSet.HLL_HASH)) {
+ validateHllHash(args, columnNameMap);
+ } else if (functionName.equalsIgnoreCase("now")) {
+ validateNowFunction(mappingColumn);
+ } else if (functionName.equalsIgnoreCase("substitute")) {
+ validateSubstituteFunction(args, columnNameMap);
+ } else {
+ if (isHadoopLoad) {
+ throw new AnalysisException("Unknown function: " + functionName);
+ }
+ }
+ }
+
+ // eg: k2 = substitute(k1)
+ // this is used for creating derivative column from existing column
+ private static void validateSubstituteFunction(List<String> args, Map<String, String> columnNameMap)
+ throws AnalysisException {
+ if (args.size() != 1) {
+ throw new AnalysisException("Should has only one argument: " + args);
+ }
+
+ String argColumn = args.get(0);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+
+ args.set(0, columnNameMap.get(argColumn));
+ }
+
+ private static void validateAlignmentTimestamp(List<String> args, Map<String, String> columnNameMap)
+ throws AnalysisException {
+ if (args.size() != 2) {
+ throw new AnalysisException("Function alignment_timestamp args size is not 2");
+ }
+
+ String precision = args.get(0).toLowerCase();
+ String regex = "^year|month|day|hour$";
+ if (!precision.matches(regex)) {
+ throw new AnalysisException("Alignment precision error. regex: " + regex + ", arg: " + precision);
+ }
+
+ String argColumn = args.get(1);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+
+ args.set(1, columnNameMap.get(argColumn));
+ }
+
+ private static void validateStrftime(List<String> args, Map<String, String> columnNameMap) throws
+ AnalysisException {
+ if (args.size() != 2) {
+ throw new AnalysisException("Function strftime needs 2 args");
+ }
+
+ String format = args.get(0);
+ String regex = "^(%[YMmdHhiSs][ -:]?){0,5}%[YMmdHhiSs]$";
+ if (!format.matches(regex)) {
+ throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + format);
+ }
+
+ String argColumn = args.get(1);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+
+ args.set(1, columnNameMap.get(argColumn));
+ }
+
+ private static void validateTimeFormat(List<String> args, Map<String, String> columnNameMap) throws
+ AnalysisException {
+ if (args.size() != 3) {
+ throw new AnalysisException("Function time_format needs 3 args");
+ }
+
+ String outputFormat = args.get(0);
+ String inputFormat = args.get(1);
+ String regex = "^(%[YMmdHhiSs][ -:]?){0,5}%[YMmdHhiSs]$";
+ if (!outputFormat.matches(regex)) {
+ throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + outputFormat);
+ }
+ if (!inputFormat.matches(regex)) {
+ throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + inputFormat);
+ }
+
+ String argColumn = args.get(2);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+
+ args.set(2, columnNameMap.get(argColumn));
+ }
+
+ private static void validateDefaultValue(List<String> args, Column column) throws AnalysisException {
+ if (args.size() != 1) {
+ throw new AnalysisException("Function default_value needs 1 arg");
+ }
+
+ if (!column.isAllowNull() && args.get(0) == null) {
+ throw new AnalysisException("Column is not null, column: " + column.getName());
+ }
+
+ if (args.get(0) != null) {
+ ColumnDef.validateDefaultValue(column.getOriginType(), args.get(0));
+ }
+ }
+
+ private static void validateMd5sum(List<String> args, Map<String, String> columnNameMap) throws AnalysisException {
+ for (int i = 0; i < args.size(); ++i) {
+ String argColumn = args.get(i);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+
+ args.set(i, columnNameMap.get(argColumn));
+ }
+ }
+
+ private static void validateReplaceValue(List<String> args, Column column) throws AnalysisException {
+ String replaceValue = null;
+ if (args.size() == 1) {
+ replaceValue = column.getDefaultValue();
+ if (replaceValue == null) {
+ throw new AnalysisException("Column " + column.getName() + " has no default value");
+ }
+
+ args.add(replaceValue);
+ } else if (args.size() == 2) {
+ replaceValue = args.get(1);
+ } else {
+ throw new AnalysisException("Function replace_value need 1 or 2 args");
+ }
+
+ if (!column.isAllowNull() && replaceValue == null) {
+ throw new AnalysisException("Column is not null, column: " + column.getName());
+ }
+
+ if (replaceValue != null) {
+ ColumnDef.validateDefaultValue(column.getOriginType(), replaceValue);
+ }
+ }
+
+ private static void validateHllHash(List<String> args, Map<String, String> columnNameMap) throws AnalysisException {
+ for (int i = 0; i < args.size(); ++i) {
+ String argColumn = args.get(i);
+ if (!columnNameMap.containsKey(argColumn)) {
+ throw new AnalysisException("Column is not in sources, column: " + argColumn);
+ }
+ args.set(i, columnNameMap.get(argColumn));
+ }
+ }
+
+ private static void validateNowFunction(Column mappingColumn) throws AnalysisException {
+ if (!mappingColumn.getOriginType().equals(Type.DATE) && !mappingColumn.getOriginType().equals(Type.DATETIME)) {
+ throw new AnalysisException("Now() function is only support for DATE/DATETIME column");
+ }
+ }
+
public String getTableName() {
return tableName;
}
@@ -212,7 +380,7 @@ public class DataDescription {
return partitionNames;
}
- public Expr getWhereExpr(){
+ public Expr getWhereExpr() {
return whereExpr;
}
@@ -359,15 +527,15 @@ public class DataDescription {
/*
* Analyze parsedExprMap and columnToHadoopFunction from columns, columns from path and columnMappingList
- * Example:
- * columns (col1, tmp_col2, tmp_col3)
+ * Example:
+ * columns (col1, tmp_col2, tmp_col3)
* columns from path as (col4, col5)
* set (col2=tmp_col2+1, col3=strftime("%Y-%m-%d %H:%M:%S", tmp_col3))
- *
+ *
* Result:
* parsedExprMap = {"col1": null, "tmp_col2": null, "tmp_col3": null, "col4": null, "col5": null,
* "col2": "tmp_col2+1", "col3": "strftime("%Y-%m-%d %H:%M:%S", tmp_col3)"}
- * columnToHadoopFunction = {"col3": "strftime("%Y-%m-%d %H:%M:%S", tmp_col3)"}
+ * columnToHadoopFunction = {"col3": "strftime("%Y-%m-%d %H:%M:%S", tmp_col3)"}
*/
private void analyzeColumns() throws AnalysisException {
if ((fileFieldNames == null || fileFieldNames.isEmpty()) && (columnsFromPath != null && !columnsFromPath.isEmpty())) {
@@ -445,11 +613,12 @@ public class DataDescription {
}
}
}
+
private void analyzeMultiLoadColumns() throws AnalysisException {
if (columnDef == null || columnDef.isEmpty()) {
return;
}
- String columnsSQL = new String("COLUMNS (" + columnDef + ")");
+ String columnsSQL = "COLUMNS (" + columnDef + ")";
SqlParser parser = new SqlParser(new SqlScanner(new StringReader(columnsSQL)));
ImportColumnsStmt columnsStmt;
try {
@@ -477,7 +646,7 @@ public class DataDescription {
}
private void analyzeColumnToHadoopFunction(String columnName, Expr child1) throws AnalysisException {
- Preconditions.checkState(child1 instanceof FunctionCallExpr);
+ Preconditions.checkState(child1 instanceof FunctionCallExpr);
FunctionCallExpr functionCallExpr = (FunctionCallExpr) child1;
String functionName = functionCallExpr.getFnName().getFunction();
if (!HADOOP_SUPPORT_FUNCTION_NAMES.contains(functionName.toLowerCase())) {
@@ -552,180 +721,7 @@ public class DataDescription {
}
if (!hasSourceSequenceCol) {
throw new AnalysisException("There is no sequence column " + sequenceCol + " in the " + table.getName()
- + " or the COLUMNS and SET clause");
- }
- }
-
- public static void validateMappingFunction(String functionName, List<String> args,
- Map<String, String> columnNameMap,
- Column mappingColumn, boolean isHadoopLoad) throws AnalysisException {
- if (functionName.equalsIgnoreCase("alignment_timestamp")) {
- validateAlignmentTimestamp(args, columnNameMap);
- } else if (functionName.equalsIgnoreCase("strftime")) {
- validateStrftime(args, columnNameMap);
- } else if (functionName.equalsIgnoreCase("time_format")) {
- validateTimeFormat(args, columnNameMap);
- } else if (functionName.equalsIgnoreCase("default_value")) {
- validateDefaultValue(args, mappingColumn);
- } else if (functionName.equalsIgnoreCase("md5sum")) {
- validateMd5sum(args, columnNameMap);
- } else if (functionName.equalsIgnoreCase("replace_value")) {
- validateReplaceValue(args, mappingColumn);
- } else if (functionName.equalsIgnoreCase(FunctionSet.HLL_HASH)) {
- validateHllHash(args, columnNameMap);
- } else if (functionName.equalsIgnoreCase("now")) {
- validateNowFunction(mappingColumn);
- } else if (functionName.equalsIgnoreCase("substitute")) {
- validateSubstituteFunction(args, columnNameMap);
- } else {
- if (isHadoopLoad) {
- throw new AnalysisException("Unknown function: " + functionName);
- }
- }
- }
-
- // eg: k2 = substitute(k1)
- // this is used for creating derivative column from existing column
- private static void validateSubstituteFunction(List<String> args, Map<String, String> columnNameMap)
- throws AnalysisException {
- if (args.size() != 1) {
- throw new AnalysisException("Should has only one argument: " + args);
- }
-
- String argColumn = args.get(0);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
-
- args.set(0, columnNameMap.get(argColumn));
- }
-
- private static void validateAlignmentTimestamp(List<String> args, Map<String, String> columnNameMap)
- throws AnalysisException {
- if (args.size() != 2) {
- throw new AnalysisException("Function alignment_timestamp args size is not 2");
- }
-
- String precision = args.get(0).toLowerCase();
- String regex = "^year|month|day|hour$";
- if (!precision.matches(regex)) {
- throw new AnalysisException("Alignment precision error. regex: " + regex + ", arg: " + precision);
- }
-
- String argColumn = args.get(1);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
-
- args.set(1, columnNameMap.get(argColumn));
- }
-
- private static void validateStrftime(List<String> args, Map<String, String> columnNameMap) throws
- AnalysisException {
- if (args.size() != 2) {
- throw new AnalysisException("Function strftime needs 2 args");
- }
-
- String format = args.get(0);
- String regex = "^(%[YMmdHhiSs][ -:]?){0,5}%[YMmdHhiSs]$";
- if (!format.matches(regex)) {
- throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + format);
- }
-
- String argColumn = args.get(1);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
-
- args.set(1, columnNameMap.get(argColumn));
- }
-
- private static void validateTimeFormat(List<String> args, Map<String, String> columnNameMap) throws
- AnalysisException {
- if (args.size() != 3) {
- throw new AnalysisException("Function time_format needs 3 args");
- }
-
- String outputFormat = args.get(0);
- String inputFormat = args.get(1);
- String regex = "^(%[YMmdHhiSs][ -:]?){0,5}%[YMmdHhiSs]$";
- if (!outputFormat.matches(regex)) {
- throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + outputFormat);
- }
- if (!inputFormat.matches(regex)) {
- throw new AnalysisException("Date format error. regex: " + regex + ", arg: " + inputFormat);
- }
-
- String argColumn = args.get(2);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
-
- args.set(2, columnNameMap.get(argColumn));
- }
-
- private static void validateDefaultValue(List<String> args, Column column) throws AnalysisException {
- if (args.size() != 1) {
- throw new AnalysisException("Function default_value needs 1 arg");
- }
-
- if (!column.isAllowNull() && args.get(0) == null) {
- throw new AnalysisException("Column is not null, column: " + column.getName());
- }
-
- if (args.get(0) != null) {
- ColumnDef.validateDefaultValue(column.getOriginType(), args.get(0));
- }
- }
-
- private static void validateMd5sum(List<String> args, Map<String, String> columnNameMap) throws AnalysisException {
- for (int i = 0; i < args.size(); ++i) {
- String argColumn = args.get(i);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
-
- args.set(i, columnNameMap.get(argColumn));
- }
- }
-
- private static void validateReplaceValue(List<String> args, Column column) throws AnalysisException {
- String replaceValue = null;
- if (args.size() == 1) {
- replaceValue = column.getDefaultValue();
- if (replaceValue == null) {
- throw new AnalysisException("Column " + column.getName() + " has no default value");
- }
-
- args.add(replaceValue);
- } else if (args.size() == 2) {
- replaceValue = args.get(1);
- } else {
- throw new AnalysisException("Function replace_value need 1 or 2 args");
- }
-
- if (!column.isAllowNull() && replaceValue == null) {
- throw new AnalysisException("Column is not null, column: " + column.getName());
- }
-
- if (replaceValue != null) {
- ColumnDef.validateDefaultValue(column.getOriginType(), replaceValue);
- }
- }
-
- private static void validateHllHash(List<String> args, Map<String, String> columnNameMap) throws AnalysisException {
- for (int i = 0; i < args.size(); ++i) {
- String argColumn = args.get(i);
- if (!columnNameMap.containsKey(argColumn)) {
- throw new AnalysisException("Column is not in sources, column: " + argColumn);
- }
- args.set(i, columnNameMap.get(argColumn));
- }
- }
-
- private static void validateNowFunction(Column mappingColumn) throws AnalysisException {
- if (!mappingColumn.getOriginType().equals(Type.DATE) && !mappingColumn.getOriginType().equals(Type.DATETIME)) {
- throw new AnalysisException("Now() function is only support for DATE/DATETIME column");
+ + " or the COLUMNS and SET clause");
}
}
@@ -745,10 +741,10 @@ public class DataDescription {
// check hive table auth
if (isLoadFromTable()) {
if (!Catalog.getCurrentCatalog().getAuth().checkTblPriv(ConnectContext.get(), fullDbName, srcTableName,
- PrivPredicate.SELECT)) {
+ PrivPredicate.SELECT)) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "SELECT",
- ConnectContext.get().getQualifiedUser(),
- ConnectContext.get().getRemoteIP(), srcTableName);
+ ConnectContext.get().getQualifiedUser(),
+ ConnectContext.get().getRemoteIP(), srcTableName);
}
}
}
@@ -804,25 +800,25 @@ public class DataDescription {
}
/*
- * If user does not specify COLUMNS in load stmt, we fill it here.
+ * If user does not specify COLUMNS in load stmt, we fill it here.
* eg1:
* both COLUMNS and SET clause is empty. after fill:
* (k1,k2,k3)
- *
+ *
* eg2:
* COLUMNS is empty, SET is not empty
* SET ( k2 = default_value("2") )
* after fill:
* (k1, k2, k3)
* SET ( k2 = default_value("2") )
- *
+ *
* eg3:
* COLUMNS is empty, SET is not empty
* SET (k2 = strftime("%Y-%m-%d %H:%M:%S", k2)
* after fill:
* (k1,k2,k3)
* SET (k2 = strftime("%Y-%m-%d %H:%M:%S", k2)
- *
+ *
*/
public void fillColumnInfoIfNotSpecified(List<Column> baseSchema) {
if (fileFieldNames != null && !fileFieldNames.isEmpty()) {
@@ -835,7 +831,7 @@ public class DataDescription {
for (ImportColumnDesc importColumnDesc : parsedColumnExprList) {
mappingColNames.add(importColumnDesc.getColumnName());
}
-
+
for (Column column : baseSchema) {
if (!mappingColNames.contains(column.getName())) {
parsedColumnExprList.add(new ImportColumnDesc(column.getName(), null));
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/LoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/LoadStmt.java
index 97f3745..8cd6013 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/LoadStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/LoadStmt.java
@@ -336,7 +336,7 @@ public class LoadStmt extends DdlStmt {
}
} else if (brokerDesc != null) {
etlJobType = EtlJobType.BROKER;
- } else if (etlJobType != EtlJobType.UNKNOWN) {
+ } else {
// if cluster is null, use default hadoop cluster
// if cluster is not null, use this hadoop cluster
etlJobType = EtlJobType.HADOOP;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
index 86c5bbf..e7dd2fe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/BrokerScanNode.java
@@ -21,10 +21,13 @@ import org.apache.doris.analysis.Analyzer;
import org.apache.doris.analysis.BrokerDesc;
import org.apache.doris.analysis.Expr;
import org.apache.doris.analysis.ImportColumnDesc;
+import org.apache.doris.analysis.IntLiteral;
import org.apache.doris.analysis.SlotDescriptor;
+import org.apache.doris.analysis.SlotRef;
import org.apache.doris.analysis.TupleDescriptor;
import org.apache.doris.catalog.BrokerTable;
import org.apache.doris.catalog.Catalog;
+import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.FsBroker;
import org.apache.doris.catalog.Table;
import org.apache.doris.common.AnalysisException;
@@ -34,6 +37,7 @@ import org.apache.doris.common.UserException;
import org.apache.doris.common.util.BrokerUtil;
import org.apache.doris.load.BrokerFileGroup;
import org.apache.doris.load.Load;
+import org.apache.doris.load.loadv2.LoadTask;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TBrokerFileStatus;
import org.apache.doris.thrift.TBrokerRangeDesc;
@@ -219,6 +223,16 @@ public class BrokerScanNode extends LoadScanNode {
List<ImportColumnDesc> columnExprs = Lists.newArrayList();
if (isLoad()) {
columnExprs = context.fileGroup.getColumnExprList();
+ if (mergeType == LoadTask.MergeType.MERGE) {
+ columnExprs.add(ImportColumnDesc.newDeleteSignImportColumnDesc(deleteCondition));
+ } else if (mergeType == LoadTask.MergeType.DELETE) {
+ columnExprs.add(ImportColumnDesc.newDeleteSignImportColumnDesc(new IntLiteral(1)));
+ }
+ // add columnExpr for sequence column
+ if (context.fileGroup.hasSequenceCol()) {
+ columnExprs.add(new ImportColumnDesc(Column.SEQUENCE_COL,
+ new SlotRef(null, context.fileGroup.getSequenceCol())));
+ }
}
Load.initColumns(targetTable, columnExprs,
@@ -271,23 +285,6 @@ public class BrokerScanNode extends LoadScanNode {
return locations;
}
- private TScanRangeLocations newLocations(TBrokerScanRangeParams params)
- throws UserException {
- // Generate on broker scan range
- TBrokerScanRange brokerScanRange = new TBrokerScanRange();
- brokerScanRange.setParams(params);
-
- // Scan range
- TScanRange scanRange = new TScanRange();
- scanRange.setBrokerScanRange(brokerScanRange);
-
- // Locations
- TScanRangeLocations locations = new TScanRangeLocations();
- locations.setScanRange(scanRange);
-
- return locations;
- }
-
private TBrokerScanRange brokerScanRange(TScanRangeLocations locations) {
return locations.scan_range.broker_scan_range;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/MultiLoadScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/MultiLoadScanNode.java
deleted file mode 100644
index a8f0495..0000000
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/MultiLoadScanNode.java
+++ /dev/null
@@ -1,39 +0,0 @@
-// 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.planner;
-
-import org.apache.doris.analysis.TupleDescriptor;
-import org.apache.doris.thrift.TScanRangeLocations;
-
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
-import java.util.List;
-
-public class MultiLoadScanNode extends LoadScanNode {
- private static final Logger LOG = LogManager.getLogger(MultiLoadScanNode.class);
-
- public MultiLoadScanNode(PlanNodeId id, TupleDescriptor desc, String planNodeName) {
- super(id, desc, planNodeName);
- }
-
- @Override
- public List<TScanRangeLocations> getScanRangeLocations(long maxScanRangeLength) {
- return null;
- }
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/MultiLoadMgr.java b/fe/fe-core/src/main/java/org/apache/doris/qe/MultiLoadMgr.java
index b3e746e..9d2d853 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/MultiLoadMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/MultiLoadMgr.java
@@ -376,6 +376,7 @@ public class MultiLoadMgr {
LoadStmt loadStmt = new LoadStmt(commitLabel, dataDescriptions, brokerDesc, null, properties);
loadStmt.setEtlJobType(EtlJobType.BROKER);
loadStmt.setOrigStmt(new OriginStatement("", 0));
+ loadStmt.setUserInfo(ConnectContext.get().getCurrentUserIdentity());
Analyzer analyzer = new Analyzer(ConnectContext.get().getCatalog(), ConnectContext.get());
try {
loadStmt.analyze(analyzer);
@@ -496,10 +497,6 @@ public class MultiLoadMgr {
if (properties.get(LoadStmt.KEY_IN_PARAM_DELETE_CONDITION) != null) {
deleteCondition = parseWhereExpr(properties.get(LoadStmt.KEY_IN_PARAM_DELETE_CONDITION));
}
- backend = Catalog.getCurrentSystemInfo().getBackend(backendId);
- if (backend == null) {
- throw new DdlException("Backend [" + backendId + "] not found. ");
- }
if (fileFormat != null && fileFormat.equalsIgnoreCase("json")) {
stripOuterArray = Boolean.valueOf(
properties.getOrDefault(LoadStmt.KEY_IN_PARAM_STRIP_OUTER_ARRAY, "false"));
@@ -507,11 +504,14 @@ public class MultiLoadMgr {
jsonRoot = properties.getOrDefault(LoadStmt.KEY_IN_PARAM_JSONROOT, "");
}
}
-
DataDescription dataDescription = new DataDescription(tbl, partitionNames, files, null, columnSeparator,
fileFormat, null, isNegative, null, whereExpr, mergeType, deleteCondition,
sequenceColName);
dataDescription.setColumnDef(colString);
+ backend = Catalog.getCurrentSystemInfo().getBackend(backendId);
+ if (backend == null) {
+ throw new DdlException("Backend [" + backendId + "] not found. ");
+ }
dataDescription.setBeAddr(new TNetworkAddress(backend.getHost(), backend.getHeartbeatPort()));
dataDescription.setFileSize(fileSizes);
dataDescription.setBackendId(backendId);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
index 512b21e..2baf1b1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/task/StreamLoadTask.java
@@ -251,7 +251,7 @@ public class StreamLoadTask implements LoadTaskInfo {
deleteCondition = parseWhereExpr(request.getDeleteCondition());
}
if (negative && mergeType != LoadTask.MergeType.APPEND) {
- throw new AnalysisException("Negative is only used when merge type is append.");
+ throw new AnalysisException("Negative is only used when merge type is APPEND.");
}
if (mergeType == LoadTask.MergeType.MERGE) {
columnExprDescs.add(ImportColumnDesc.newDeleteSignImportColumnDesc(deleteCondition));
diff --git a/gensrc/script/gen_builtins_functions.py b/gensrc/script/gen_builtins_functions.py
index a30364c..e4808a3 100755
--- a/gensrc/script/gen_builtins_functions.py
+++ b/gensrc/script/gen_builtins_functions.py
@@ -48,7 +48,7 @@ java_registry_epilogue = '\
}\n'
FE_PATH = "../../../fe/fe-core/target/generated-sources/build/org/apache/doris/builtins/"
-print FE_PATH
+print(FE_PATH)
# This contains all the metadata to describe all the builtins.
# Each meta data entry is itself a map to store all the meta data
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org