[config](load) enable new load scan node by default (#14808)

Set FE `enable_new_load_scan_node` to true by default. So that all load tasks(broker load, stream load, routine load, insert into) will use FileScanNode instead of BrokerScanNode to read data 1. Support loading parquet file in stream load with new load scan node. 2. Fix bug that new parquet reader can not read column without logical or converted type. 3. Change jsonb parser function to "jsonb_parse_error_to_null" So that if the input string is not a valid json string, it will return null for jsonb column in load task.
apache · Dec 16, 2022 · 0e1e5a8 · 0e1e5a8
1 parent e0d5289
commit 0e1e5a8
Show file tree

Hide file tree

Showing 36 changed files with 164 additions and 304 deletions.
diff --git a/be/src/http/action/stream_load.cpp b/be/src/http/action/stream_load.cpp
@@ -408,6 +408,7 @@ Status StreamLoadAction::_process_put(HttpRequest* http_req, StreamLoadContext*
         RETURN_IF_ERROR(file_sink->open());
         request.__isset.path = true;
         request.fileType = TFileType::FILE_LOCAL;
+        request.__set_file_size(ctx->body_bytes);
         ctx->body_sink = file_sink;
     }
     if (!http_req->header(HTTP_COLUMNS).empty()) {

diff --git a/be/src/util/jsonb_parser.h b/be/src/util/jsonb_parser.h
@@ -65,6 +65,7 @@
 #include "jsonb_document.h"
 #include "jsonb_error.h"
 #include "jsonb_writer.h"
+#include "string_parser.hpp"
 
 namespace doris {
 
@@ -894,8 +895,12 @@ class JsonbParserT {
         }
 
         *pbuf = 0; // set null-terminator
-        int64_t val = strtol(num_buf_, NULL, 10);
-        if (errno == ERANGE) {
+        StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+        int64_t val =
+                StringParser::string_to_int<int64_t>(num_buf_, pbuf - num_buf_, &parse_result);
+        if (parse_result != StringParser::PARSE_SUCCESS) {
+            VLOG_ROW << "debug string_to_int error for " << num_buf_ << " val=" << val
+                     << " parse_result=" << parse_result;
             err_ = JsonbErrType::E_DECIMAL_OVERFLOW;
             return false;
         }
@@ -950,7 +955,7 @@ class JsonbParserT {
         }
 
         *pbuf = 0; // set null-terminator
-        return internConvertBufferToDouble();
+        return internConvertBufferToDouble(num_buf_, pbuf - num_buf_);
     }
 
     // parse the exponent part of a double number
@@ -990,15 +995,17 @@ class JsonbParserT {
         }
 
         *pbuf = 0; // set null-terminator
-        return internConvertBufferToDouble();
+        return internConvertBufferToDouble(num_buf_, pbuf - num_buf_);
     }
 
     // call system function to parse double to string
-    bool internConvertBufferToDouble() {
-        double val = strtod(num_buf_, NULL);
-
-        if (errno == ERANGE) {
-            err_ = JsonbErrType::E_DOUBLE_OVERFLOW;
+    bool internConvertBufferToDouble(char* num_buf_, int len) {
+        StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;
+        double val = StringParser::string_to_float<double>(num_buf_, len, &parse_result);
+        if (parse_result != StringParser::PARSE_SUCCESS) {
+            VLOG_ROW << "debug string_to_float error for " << num_buf_ << " val=" << val
+                     << " parse_result=" << parse_result;
+            err_ = JsonbErrType::E_DECIMAL_OVERFLOW;
             return false;
         }
 

diff --git a/be/src/vec/exec/format/parquet/schema_desc.cpp b/be/src/vec/exec/format/parquet/schema_desc.cpp
@@ -168,20 +168,24 @@ TypeDescriptor FieldDescriptor::get_doris_type(const tparquet::SchemaElement& ph
         switch (physical_schema.type) {
         case tparquet::Type::BOOLEAN:
             type.type = TYPE_BOOLEAN;
-            return type;
+            break;
         case tparquet::Type::INT32:
             type.type = TYPE_INT;
-            return type;
+            break;
         case tparquet::Type::INT64:
         case tparquet::Type::INT96:
             type.type = TYPE_BIGINT;
-            return type;
+            break;
         case tparquet::Type::FLOAT:
             type.type = TYPE_FLOAT;
-            return type;
+            break;
         case tparquet::Type::DOUBLE:
             type.type = TYPE_DOUBLE;
-            return type;
+            break;
+        case tparquet::Type::BYTE_ARRAY:
+        case tparquet::Type::FIXED_LEN_BYTE_ARRAY:
+            type.type = TYPE_STRING;
+            break;
         default:
             break;
         }

diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -210,7 +210,11 @@ Status VFileScanner::_init_src_block(Block* block) {
             data_type = DataTypeFactory::instance().create_data_type(it->second, true);
         }
         if (data_type == nullptr) {
-            return Status::NotSupported(fmt::format("Not support arrow type:{}", slot->col_name()));
+            return Status::NotSupported(
+                    fmt::format("Not support data type:{} for column: {}",
+                                (it == _name_to_col_type.end() ? slot->type().debug_string()
+                                                               : it->second.debug_string()),
+                                slot->col_name()));
         }
         MutableColumnPtr data_column = data_type->create_column();
         _src_block.insert(

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
@@ -222,9 +222,14 @@ public DataDescription(String tableName,
     public DataDescription(String tableName, LoadTaskInfo taskInfo) {
         this.tableName = tableName;
         this.partitionNames = taskInfo.getPartitions();
-        // Add a dummy path to just make analyze() happy.
-        // Stream load does not need this field.
-        this.filePaths = Lists.newArrayList("dummy");
+
+        if (!Strings.isNullOrEmpty(taskInfo.getPath())) {
+            this.filePaths = Lists.newArrayList(taskInfo.getPath());
+        } else {
+            // Add a dummy path to just make analyze() happy.
+            this.filePaths = Lists.newArrayList("dummy");
+        }
+
         this.fileFieldNames = taskInfo.getColumnExprDescs().getFileColNames();
         this.columnSeparator = taskInfo.getColumnSeparator();
         this.lineDelimiter = taskInfo.getLineDelimiter();
@@ -259,7 +264,20 @@ private void getFileFormatAndCompressType(LoadTaskInfo taskInfo) {
                 // the compress type is saved in "compressType"
                 this.fileFormat = "csv";
             } else {
-                this.fileFormat = "json";
+                switch (type) {
+                    case FORMAT_ORC:
+                        this.fileFormat = "orc";
+                        break;
+                    case FORMAT_PARQUET:
+                        this.fileFormat = "parquet";
+                        break;
+                    case FORMAT_JSON:
+                        this.fileFormat = "json";
+                        break;
+                    default:
+                        this.fileFormat = "unknown";
+                        break;
+                }
             }
         }
         // get compress type
@@ -1019,3 +1037,4 @@ public String toString() {
         return toSql();
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/Config.java b/fe/fe-core/src/main/java/org/apache/doris/common/Config.java
@@ -1749,7 +1749,7 @@ public class Config extends ConfigBase {
      * Temp config, should be removed when new file scan node is ready.
      */
     @ConfField(mutable = true)
-    public static boolean enable_new_load_scan_node = false;
+    public static boolean enable_new_load_scan_node = true;
 
     /**
      * Max data version of backends serialize block.

diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java b/fe/fe-core/src/main/java/org/apache/doris/load/BrokerFileGroup.java
@@ -155,6 +155,7 @@ public BrokerFileGroup(DataDescription dataDescription) {
         this.deleteCondition = dataDescription.getDeleteCondition();
         this.mergeType = dataDescription.getMergeType();
         this.sequenceCol = dataDescription.getSequenceCol();
+        this.filePaths = dataDescription.getFilePaths();
     }
 
     // NOTE: DBLock will be held
@@ -598,3 +599,4 @@ public static BrokerFileGroup read(DataInput in) throws IOException {
         return fileGroup;
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/LoadScanNode.java
@@ -207,6 +207,8 @@ protected void finalizeParams(Map<String, SlotDescriptor> slotDescByName,
                 expr.analyze(analyzer);
             }
 
+            // for jsonb type, use jsonb_parse_xxx to parse src string to jsonb.
+            // and if input string is not a valid json string, return null.
             PrimitiveType dstType = destSlotDesc.getType().getPrimitiveType();
             PrimitiveType srcType = expr.getType().getPrimitiveType();
             if (dstType == PrimitiveType.JSONB
@@ -217,7 +219,7 @@ protected void finalizeParams(Map<String, SlotDescriptor> slotDescByName,
                 if (destSlotDesc.getIsNullable() || expr.isNullable()) {
                     nullable = "nullable";
                 }
-                String name = "jsonb_parse_" + nullable + "_error_to_invalid";
+                String name = "jsonb_parse_" + nullable + "_error_to_null";
                 expr = new FunctionCallExpr(name, args);
                 expr.analyze(analyzer);
             } else {
@@ -251,3 +253,4 @@ protected void toThrift(TPlanNode planNode) {
         planNode.setBrokerScanNode(brokerScanNode);
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/StreamLoadPlanner.java
@@ -52,6 +52,7 @@
 import org.apache.doris.thrift.PaloInternalServiceVersion;
 import org.apache.doris.thrift.TBrokerFileStatus;
 import org.apache.doris.thrift.TExecPlanFragmentParams;
+import org.apache.doris.thrift.TFileType;
 import org.apache.doris.thrift.TLoadErrorHubInfo;
 import org.apache.doris.thrift.TNetworkAddress;
 import org.apache.doris.thrift.TPlanFragmentExecParams;
@@ -180,9 +181,15 @@ public TExecPlanFragmentParams plan(TUniqueId loadId) throws UserException {
             fileGroup.parse(db, dataDescription);
             // 2. create dummy file status
             TBrokerFileStatus fileStatus = new TBrokerFileStatus();
-            fileStatus.setPath("");
-            fileStatus.setIsDir(false);
-            fileStatus.setSize(-1); // must set to -1, means stream.
+            if (taskInfo.getFileType() == TFileType.FILE_LOCAL) {
+                fileStatus.setPath(taskInfo.getPath());
+                fileStatus.setIsDir(false);
+                fileStatus.setSize(taskInfo.getFileSize()); // must set to -1, means stream.
+            } else {
+                fileStatus.setPath("");
+                fileStatus.setIsDir(false);
+                fileStatus.setSize(-1); // must set to -1, means stream.
+            }
             fileScanNode.setLoadInfo(loadId, taskInfo.getTxnId(), destTable, BrokerDesc.createForStreamLoad(),
                     fileGroup, fileStatus, taskInfo.isStrictMode(), taskInfo.getFileType());
             scanNode = fileScanNode;
@@ -324,3 +331,4 @@ private List<Long> getAllPartitionIds() throws DdlException, AnalysisException {
         return null;
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/ExternalFileScanNode.java
@@ -456,7 +456,25 @@ protected void finalizeParamsForLoad(ParamCreateContext context, Analyzer analyz
                 expr = new ArithmeticExpr(ArithmeticExpr.Operator.MULTIPLY, expr, new IntLiteral(-1));
                 expr.analyze(analyzer);
             }
-            expr = castToSlot(destSlotDesc, expr);
+
+            // for jsonb type, use jsonb_parse_xxx to parse src string to jsonb.
+            // and if input string is not a valid json string, return null.
+            PrimitiveType dstType = destSlotDesc.getType().getPrimitiveType();
+            PrimitiveType srcType = expr.getType().getPrimitiveType();
+            if (dstType == PrimitiveType.JSONB
+                    && (srcType == PrimitiveType.VARCHAR || srcType == PrimitiveType.STRING)) {
+                List<Expr> args = Lists.newArrayList();
+                args.add(expr);
+                String nullable = "notnull";
+                if (destSlotDesc.getIsNullable() || expr.isNullable()) {
+                    nullable = "nullable";
+                }
+                String name = "jsonb_parse_" + nullable + "_error_to_null";
+                expr = new FunctionCallExpr(name, args);
+                expr.analyze(analyzer);
+            } else {
+                expr = castToSlot(destSlotDesc, expr);
+            }
             params.putToExprOfDestSlot(destSlotDesc.getId().asInt(), expr.treeToThrift());
         }
         params.setDestSidToSrcSidWithoutTrans(destSidToSrcSidWithoutTrans);
@@ -550,3 +568,4 @@ public String getNodeExplainString(String prefix, TExplainLevel detailLevel) {
 }
 
 
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileGroupInfo.java
@@ -44,6 +44,7 @@
 import org.apache.doris.thrift.TScanRangeLocations;
 import org.apache.doris.thrift.TUniqueId;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -140,6 +141,10 @@ public int getLoadParallelism() {
         return loadParallelism;
     }
 
+    public TFileType getFileType() {
+        return fileType;
+    }
+
     public String getExplainString(String prefix) {
         StringBuilder sb = new StringBuilder();
         sb.append("file scan\n");
@@ -303,10 +308,20 @@ private TFileRangeDesc createFileRangeDesc(long curFileOffset, TBrokerFileStatus
             rangeDesc.setSize(rangeBytes);
             rangeDesc.setColumnsFromPath(columnsFromPath);
         } else {
+            // for stream load
+            if (getFileType() == TFileType.FILE_LOCAL) {
+                // when loading parquet via stream, there will be a local file saved on BE
+                // so to read it as a local file.
+                Preconditions.checkState(fileGroup.getFilePaths().size() == 1);
+                rangeDesc.setPath(fileGroup.getFilePaths().get(0));
+                rangeDesc.setStartOffset(0);
+                rangeDesc.setSize(fileStatus.size);
+            }
             rangeDesc.setLoadId(loadId);
             rangeDesc.setSize(fileStatus.size);
         }
         return rangeDesc;
     }
 }
 
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/LoadScanProvider.java
@@ -111,7 +111,7 @@ public ParamCreateContext createContext(Analyzer analyzer) throws UserException
         TFileAttributes fileAttributes = new TFileAttributes();
         setFileAttributes(ctx.fileGroup, fileAttributes);
         params.setFileAttributes(fileAttributes);
-        params.setFileType(fileGroupInfo.getBrokerDesc().getFileType());
+        params.setFileType(fileGroupInfo.getFileType());
         ctx.params = params;
 
         initColumns(ctx, analyzer);
@@ -252,3 +252,4 @@ public TableIf getTargetTable() {
         return fileGroupInfo.getTargetTable();
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -893,7 +893,7 @@ public TStreamLoadPutResult streamLoadPut(TStreamLoadPutRequest request) {
         } catch (Throwable e) {
             LOG.warn("catch unknown result.", e);
             status.setStatusCode(TStatusCode.INTERNAL_ERROR);
-            status.addToErrorMsgs(Strings.nullToEmpty(e.getMessage()));
+            status.addToErrorMsgs(e.getClass().getSimpleName() + ": " + Strings.nullToEmpty(e.getMessage()));
             return result;
         }
         return result;
@@ -1224,3 +1224,4 @@ private TInitExternalCtlMetaResult initDb(long catalogId, long dbId) throws TExc
         return result;
     }
 }
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/task/LoadTaskInfo.java
@@ -71,6 +71,10 @@ public interface LoadTaskInfo {
 
     String getPath();
 
+    default long getFileSize() {
+        return 0;
+    }
+
     double getMaxFilterRatio();
 
     ImportColumnDescs getColumnExprDescs();
@@ -118,3 +122,4 @@ public List<Expr> getColumnMappingList() {
         }
     }
 }
+
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
@@ -68,6 +68,7 @@ public class StreamLoadTask implements LoadTaskInfo {
     private Separator lineDelimiter;
     private PartitionNames partitions;
     private String path;
+    private long fileSize = 0;
     private boolean negative;
     private boolean strictMode = false; // default is false
     private String timezone = TimeUtils.DEFAULT_TIME_ZONE;
@@ -159,6 +160,11 @@ public String getPath() {
         return path;
     }
 
+    @Override
+    public long getFileSize() {
+        return fileSize;
+    }
+
     public boolean getNegative() {
         return negative;
     }
@@ -234,6 +240,7 @@ public boolean hasSequenceCol() {
         return !Strings.isNullOrEmpty(sequenceCol);
     }
 
+
     @Override
     public String getSequenceCol() {
         return sequenceCol;
@@ -249,6 +256,9 @@ public static StreamLoadTask fromTStreamLoadPutRequest(TStreamLoadPutRequest req
                 request.getFileType(), request.getFormatType(),
                 request.getCompressType());
         streamLoadTask.setOptionalFromTSLPutRequest(request);
+        if (request.isSetFileSize()) {
+            streamLoadTask.fileSize = request.getFileSize();
+        }
         return streamLoadTask;
     }
 
@@ -416,3 +426,4 @@ public double getMaxFilterRatio() {
         return maxFilterRatio;
     }
 }
+