This is an automated email from the ASF dual-hosted git repository.
sollhui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 31e3606b265 [fix](load) Fix Arrow stream load with uppercase columns
(#65127)
31e3606b265 is described below
commit 31e3606b2655f2f42d864b3e3263350cbfac8deb
Author: hui lai <[email protected]>
AuthorDate: Mon Jul 13 10:16:50 2026 +0800
[fix](load) Fix Arrow stream load with uppercase columns (#65127)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Arrow Stream Load through the Nereids load planner
lowercased source column names even though Arrow schema field names are
case-sensitive. For tables with uppercase target columns, the Arrow
scan-slot path could resolve the source name through a case-sensitive
type map and fail planning with a null type or construct a scan slot
whose name no longer matched the Arrow field. This change preserves
source column names for Arrow load format, including non-lowercase Arrow
format spellings, and resolves Arrow scan slot types through the target
table's case-insensitive column lookup.
---
.../org/apache/doris/analysis/DataDescription.java | 14 +-
.../java/org/apache/doris/common/util/Util.java | 10 ++
.../doris/nereids/load/NereidsDataDescription.java | 19 +-
.../nereids/load/NereidsLoadScanProvider.java | 13 +-
.../doris/nereids/load/NereidsLoadUtils.java | 10 ++
.../nereids/load/NereidsLoadingTaskPlanner.java | 18 +-
.../nereids/load/NereidsStreamLoadPlanner.java | 26 ++-
.../nereids/load/NereidsLoadScanProviderTest.java | 192 +++++++++++++++++++++
.../stream_load/arrow_uppercase_column.arrow | Bin 0 -> 456 bytes
.../test_arrow_stream_load_uppercase_column.out | 8 +
.../test_arrow_stream_load_uppercase_column.groovy | 89 ++++++++++
11 files changed, 358 insertions(+), 41 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 be306098077..b59288d4442 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
@@ -29,6 +29,7 @@ import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.Pair;
import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.Util;
import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
import org.apache.doris.datasource.property.fileformat.FileFormatProperties;
import
org.apache.doris.datasource.property.fileformat.JsonFileFormatProperties;
@@ -231,7 +232,7 @@ public class DataDescription {
putAnalysisMapIfNonNull(FileFormatProperties.PROP_FORMAT, fileFormat);
putAnalysisMapIfNonNull(FileFormatProperties.PROP_COMPRESS_TYPE,
compressType);
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
columnsNameToLowerCase(columnsFromPath);
this.isHadoopLoad = isHadoopLoad;
@@ -297,7 +298,7 @@ public class DataDescription {
putAnalysisMapIfNonNull(FileFormatProperties.PROP_FORMAT, fileFormat);
putAnalysisMapIfNonNull(FileFormatProperties.PROP_COMPRESS_TYPE,
compressType);
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
columnsNameToLowerCase(columnsFromPath);
}
@@ -652,9 +653,16 @@ public class DataDescription {
}
}
+ private void fileFieldNamesToLowerCase(List<String> columns) {
+ if
(Util.isCasePreservingFormat(analysisMap.get(FileFormatProperties.PROP_FORMAT)))
{
+ return;
+ }
+ columnsNameToLowerCase(columns);
+ }
+
// Change all the columns name to lower case, because Doris column is
case-insensitive.
private void columnsNameToLowerCase(List<String> columns) {
- if (columns == null || columns.isEmpty() ||
"json".equals(analysisMap.get(FileFormatProperties.PROP_FORMAT))) {
+ if (columns == null || columns.isEmpty()) {
return;
}
for (int i = 0; i < columns.size(); i++) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
index 25c0d81295b..0a4e8f967e3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/Util.java
@@ -505,6 +505,16 @@ public class Util {
|| fileFormatType == TFileFormatType.FORMAT_CSV_PLAIN;
}
+ public static boolean isCasePreservingFormat(String formatName) {
+ return !Strings.isNullOrEmpty(formatName)
+ &&
isCasePreservingFormat(getFileFormatTypeFromName(formatName));
+ }
+
+ public static boolean isCasePreservingFormat(TFileFormatType
fileFormatType) {
+ return fileFormatType == TFileFormatType.FORMAT_JSON
+ || fileFormatType == TFileFormatType.FORMAT_ARROW;
+ }
+
public static void logAndThrowRuntimeException(Logger logger, String msg,
Throwable e) {
logger.warn(msg, e);
throw new RuntimeException(msg, e);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsDataDescription.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsDataDescription.java
index 8e7b45f48c2..7c0c4b2fd97 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsDataDescription.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsDataDescription.java
@@ -241,7 +241,7 @@ public class NereidsDataDescription {
}
putAnalysisMapIfNonNull(FileFormatProperties.PROP_FORMAT, fileFormat);
putAnalysisMapIfNonNull(FileFormatProperties.PROP_COMPRESS_TYPE,
compressType);
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
columnsNameToLowerCase(columnsFromPath);
}
@@ -294,7 +294,7 @@ public class NereidsDataDescription {
}
putAnalysisMapIfNonNull(FileFormatProperties.PROP_FORMAT, fileFormat);
putAnalysisMapIfNonNull(FileFormatProperties.PROP_COMPRESS_TYPE,
compressType);
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
columnsNameToLowerCase(columnsFromPath);
}
@@ -375,7 +375,7 @@ public class NereidsDataDescription {
}
putAnalysisMapIfNonNull(CsvFileFormatProperties.PROP_SKIP_LINES,
String.valueOf(skipLines));
this.isMysqlLoad = true;
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
}
/**
@@ -437,7 +437,7 @@ public class NereidsDataDescription {
String.valueOf(taskInfo.isReadJsonByLine()));
putAnalysisMapIfNonNull(JsonFileFormatProperties.PROP_NUM_AS_STRING,
String.valueOf(taskInfo.isNumAsString()));
this.uniquekeyUpdateMode = taskInfo.getUniqueKeyUpdateMode();
- columnsNameToLowerCase(fileFieldNames);
+ fileFieldNamesToLowerCase(fileFieldNames);
}
private String getFileFormat(NereidsLoadTaskInfo taskInfo) {
@@ -990,9 +990,16 @@ public class NereidsDataDescription {
}
}
- // Change all the columns name to lower case, because Doris column is
case-insensitive.
+ private void fileFieldNamesToLowerCase(List<String> columns) {
+ if
(Util.isCasePreservingFormat(analysisMap.get(FileFormatProperties.PROP_FORMAT)))
{
+ return;
+ }
+ columnsNameToLowerCase(columns);
+ }
+
+ // Change column names to lower case, because Doris column is
case-insensitive.
private void columnsNameToLowerCase(List<String> columns) {
- if (columns == null || columns.isEmpty() ||
"json".equals(analysisMap.get(FileFormatProperties.PROP_FORMAT))) {
+ if (columns == null || columns.isEmpty()) {
return;
}
for (int i = 0; i < columns.size(); i++) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java
index b9949a825b1..42bc039f694 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java
@@ -23,7 +23,6 @@ import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf;
-import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.IdGenerator;
@@ -59,7 +58,6 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -204,7 +202,8 @@ public class NereidsLoadScanProvider {
continue;
}
NereidsImportColumnDesc columnDesc;
- if (fileGroup.getFileFormatProperties().getFileFormatType() ==
TFileFormatType.FORMAT_JSON) {
+ TFileFormatType fileFormatType =
fileGroup.getFileFormatProperties().getFileFormatType();
+ if (Util.isCasePreservingFormat(fileFormatType)) {
columnDesc = new NereidsImportColumnDesc(column.getName());
} else {
columnDesc = new
NereidsImportColumnDesc(column.getName().toLowerCase());
@@ -281,7 +280,6 @@ public class NereidsLoadScanProvider {
}
}
- HashMap<String, Type> colToType = new HashMap<>();
// check default value and auto-increment column
for (Column column : tbl.getBaseSchema()) {
if (fileGroupInfo.getUniqueKeyUpdateMode() ==
TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS
@@ -289,7 +287,6 @@ public class NereidsLoadScanProvider {
continue;
}
String columnName = column.getName();
- colToType.put(columnName, column.getType());
Expression expression = null;
if (column.getGeneratedColumnInfo() != null) {
// the generated column will be handled by bindSink
@@ -372,7 +369,11 @@ public class NereidsLoadScanProvider {
// Use real column type for arrow/native format, other formats
read as varchar first
if (fileFormatType == TFileFormatType.FORMAT_ARROW
|| fileFormatType == TFileFormatType.FORMAT_NATIVE) {
- slotColumn = new Column(realColName,
colToType.get(realColName), true);
+ if (tblColumn == null) {
+ throw new AnalysisException("Unknown column " +
realColName + " in table " + tbl.getName()
+ + " for " + fileFormatType + " load");
+ }
+ slotColumn = new Column(realColName, tblColumn.getType(),
true);
} else {
if (fileGroupInfo.getUniqueKeyUpdateMode() ==
TUniqueKeyUpdateMode.UPDATE_FLEXIBLE_COLUMNS
&& hasSkipBitmapColumn) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadUtils.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadUtils.java
index 10d48cee510..2b1148a253c 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadUtils.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadUtils.java
@@ -81,6 +81,16 @@ import java.util.TreeSet;
* NereidsLoadUtils
*/
public class NereidsLoadUtils {
+ static boolean hasImportColumn(List<NereidsImportColumnDesc>
importColumnDescs, Column tableColumn) {
+ for (NereidsImportColumnDesc importColumnDesc : importColumnDescs) {
+ if (importColumnDesc.getColumnName() != null
+ &&
importColumnDesc.getColumnName().equalsIgnoreCase(tableColumn.getName())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/**
* parse a expression list as 'select expr1, expr2,... exprn' into a
nereids Expression List
*/
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
index 001ede1fbcc..924cc43c1d8 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadingTaskPlanner.java
@@ -124,19 +124,15 @@ public class NereidsLoadingTaskPlanner {
HashSet<String> partialUpdateInputColumns = new HashSet<>();
if (isPartialUpdate) {
+ List<NereidsImportColumnDesc> importColumnDescs =
fileGroups.get(0).getColumnExprList();
for (Column col : table.getFullSchema()) {
- boolean existInExpr = false;
- for (NereidsImportColumnDesc importColumnDesc :
fileGroups.get(0).getColumnExprList()) {
- if (importColumnDesc.getColumnName() != null
- &&
importColumnDesc.getColumnName().equals(col.getName())) {
- if (!col.isVisible() &&
!Column.DELETE_SIGN.equals(col.getName())) {
- throw new UserException("Partial update should not
include invisible column except"
- + " delete sign column: " + col.getName());
- }
- partialUpdateInputColumns.add(col.getName());
- existInExpr = true;
- break;
+ boolean existInExpr =
NereidsLoadUtils.hasImportColumn(importColumnDescs, col);
+ if (existInExpr) {
+ if (!col.isVisible() &&
!Column.DELETE_SIGN.equals(col.getName())) {
+ throw new UserException("Partial update should not
include invisible column except"
+ + " delete sign column: " + col.getName());
}
+ partialUpdateInputColumns.add(col.getName());
}
if (col.isKey() && !existInExpr) {
throw new UserException("Partial update should include all
key columns, missing: " + col.getName());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
index 69d68365189..6bc4a4974e9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadPlanner.java
@@ -159,21 +159,17 @@ public class NereidsStreamLoadPlanner {
if (col.hasOnUpdateDefaultValue()) {
partialUpdateInputColumns.add(col.getName());
}
- for (NereidsImportColumnDesc importColumnDesc :
taskInfo.getColumnExprDescs().descs) {
- if (importColumnDesc.getColumnName() != null
- &&
importColumnDesc.getColumnName().equals(col.getName())) {
- if (!col.isVisible() &&
!Column.DELETE_SIGN.equals(col.getName())) {
- throw new UserException("Partial update should not
include invisible column except"
- + " delete sign column: " + col.getName());
- }
- partialUpdateInputColumns.add(col.getName());
- if (destTable.hasSequenceCol()
- && (taskInfo.hasSequenceCol() ||
(destTable.getSequenceMapCol() != null
- &&
destTable.getSequenceMapCol().equalsIgnoreCase(col.getName())))) {
- partialUpdateInputColumns.add(Column.SEQUENCE_COL);
- }
- existInExpr = true;
- break;
+ existInExpr =
NereidsLoadUtils.hasImportColumn(taskInfo.getColumnExprDescs().descs, col);
+ if (existInExpr) {
+ if (!col.isVisible() &&
!Column.DELETE_SIGN.equals(col.getName())) {
+ throw new UserException("Partial update should not
include invisible column except"
+ + " delete sign column: " + col.getName());
+ }
+ partialUpdateInputColumns.add(col.getName());
+ if (destTable.hasSequenceCol()
+ && (taskInfo.hasSequenceCol() ||
(destTable.getSequenceMapCol() != null
+ &&
destTable.getSequenceMapCol().equalsIgnoreCase(col.getName())))) {
+ partialUpdateInputColumns.add(Column.SEQUENCE_COL);
}
}
if (!existInExpr) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
new file mode 100644
index 00000000000..95d469a4a93
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsLoadScanProviderTest.java
@@ -0,0 +1,192 @@
+// 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.load;
+
+import org.apache.doris.analysis.BrokerDesc;
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.KeysType;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.common.UserException;
+import
org.apache.doris.datasource.property.fileformat.ArrowFileFormatProperties;
+import org.apache.doris.datasource.property.fileformat.FileFormatProperties;
+import
org.apache.doris.datasource.property.fileformat.NativeFileFormatProperties;
+import org.apache.doris.load.loadv2.LoadTask;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.thrift.TFileCompressType;
+import org.apache.doris.thrift.TFileFormatType;
+import org.apache.doris.thrift.TFileType;
+import org.apache.doris.thrift.TStreamLoadPutRequest;
+import org.apache.doris.thrift.TUniqueId;
+import org.apache.doris.thrift.TUniqueKeyUpdateMode;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class NereidsLoadScanProviderTest {
+
+ @Test
+ public void testArrowStreamLoadKeepsExplicitColumnCase() throws Exception {
+ TStreamLoadPutRequest request = new TStreamLoadPutRequest();
+ request.setLoadId(new TUniqueId(1, 2));
+ request.setTxnId(3);
+ request.setFileType(TFileType.FILE_STREAM);
+ request.setFormatType(TFileFormatType.FORMAT_ARROW);
+ request.setCompressType(TFileCompressType.UNKNOWN);
+ request.setColumns("time,securityid,EV");
+
+ NereidsStreamLoadTask task =
NereidsStreamLoadTask.fromTStreamLoadPutRequest(request);
+ NereidsDataDescription dataDescription = new
NereidsDataDescription("t_upper", task);
+
+ Assertions.assertEquals(Lists.newArrayList("time", "securityid", "EV"),
+ dataDescription.getFileFieldNames());
+ }
+
+ @Test
+ public void
testArrowBrokerLoadKeepsExplicitColumnCaseForNonLowercaseFormat() {
+ NereidsDataDescription dataDescription = new
NereidsDataDescription("t_upper", null,
+ Lists.newArrayList("dummy"), Lists.newArrayList("time",
"securityid", "EV"),
+ null, null, "ARROW", null, null, false,
Collections.emptyList(), null, null,
+ LoadTask.MergeType.APPEND, null, null, null);
+
+ Assertions.assertEquals(Lists.newArrayList("time", "securityid", "EV"),
+ dataDescription.getFileFieldNames());
+ }
+
+ @Test
+ public void testArrowBrokerLoadLowersColumnsFromPath() {
+ NereidsDataDescription dataDescription = new
NereidsDataDescription("t_upper", null,
+ Lists.newArrayList("dummy"), Lists.newArrayList("time",
"securityid", "EV"),
+ null, null, "ARROW", null, Lists.newArrayList("DT"), false,
Collections.emptyList(), null, null,
+ LoadTask.MergeType.APPEND, null, null, null);
+
+ Assertions.assertEquals(Lists.newArrayList("time", "securityid", "EV"),
+ dataDescription.getFileFieldNames());
+ Assertions.assertEquals(Lists.newArrayList("dt"),
dataDescription.getColumnsFromPath());
+ }
+
+ @Test
+ public void testArrowAutoGeneratedSlotsKeepUppercaseTableColumn() throws
Exception {
+ OlapTable table = mockTable();
+ NereidsParamCreateContext context = createArrowLoadContext(table,
ImmutableList.of());
+
+ assertSlot(context, "time", PrimitiveType.DATETIME);
+ assertSlot(context, "securityid", PrimitiveType.INT);
+ assertSlot(context, "EV", PrimitiveType.DOUBLE);
+ Assertions.assertFalse(context.scanSlots.stream().anyMatch(slot ->
slot.getName().equals("ev")));
+ }
+
+ @Test
+ public void testArrowSourceColumnUsesCaseInsensitiveTableType() throws
Exception {
+ OlapTable table = mockTable();
+ NereidsParamCreateContext context = createArrowLoadContext(table,
+ ImmutableList.of(new NereidsImportColumnDesc("time"),
+ new NereidsImportColumnDesc("securityid"),
+ new NereidsImportColumnDesc("ev")));
+
+ assertSlot(context, "ev", PrimitiveType.DOUBLE);
+ }
+
+ @Test
+ public void testNativeSourceColumnUsesCaseInsensitiveTableType() throws
Exception {
+ OlapTable table = mockTable();
+ NereidsParamCreateContext context = createLoadContext(table,
+ ImmutableList.of(new NereidsImportColumnDesc("time"),
+ new NereidsImportColumnDesc("securityid"),
+ new NereidsImportColumnDesc("ev")),
+ new NativeFileFormatProperties());
+
+ assertSlot(context, "ev", PrimitiveType.DOUBLE);
+ }
+
+ @Test
+ public void testImportColumnMatchesTableColumnCaseInsensitive() {
+ Column tableColumn = new Column("EV", PrimitiveType.DOUBLE, true);
+
+ Assertions.assertTrue(NereidsLoadUtils.hasImportColumn(
+ ImmutableList.of(new NereidsImportColumnDesc("ev")),
tableColumn));
+ Assertions.assertTrue(NereidsLoadUtils.hasImportColumn(
+ ImmutableList.of(new NereidsImportColumnDesc("EV")),
tableColumn));
+ Assertions.assertFalse(NereidsLoadUtils.hasImportColumn(
+ ImmutableList.of(new NereidsImportColumnDesc("other")),
tableColumn));
+ }
+
+ private NereidsParamCreateContext createArrowLoadContext(
+ OlapTable table, List<NereidsImportColumnDesc> columnExprList)
throws UserException {
+ return createLoadContext(table, columnExprList, new
ArrowFileFormatProperties());
+ }
+
+ private NereidsParamCreateContext createLoadContext(OlapTable table,
List<NereidsImportColumnDesc> columnExprList,
+ FileFormatProperties fileFormatProperties)
+ throws UserException {
+ NereidsBrokerFileGroup fileGroup = new NereidsBrokerFileGroup(1L,
false, null,
+ Lists.newArrayList("dummy"), null, null, null, columnExprList,
null, null, null, null,
+ LoadTask.MergeType.APPEND, null, -1L, false, false,
fileFormatProperties);
+ TBrokerFileStatus fileStatus = new TBrokerFileStatus();
+ fileStatus.setPath("");
+ fileStatus.setIsDir(false);
+ fileStatus.setSize(-1);
+ NereidsFileGroupInfo fileGroupInfo = new NereidsFileGroupInfo(new
TUniqueId(1, 2), 3L, table,
+ BrokerDesc.createForStreamLoad(), fileGroup, fileStatus,
false, TFileType.FILE_STREAM, null,
+ TUniqueKeyUpdateMode.UPSERT, null);
+ return new NereidsLoadScanProvider(fileGroupInfo,
Collections.emptySet()).createLoadContext();
+ }
+
+ private OlapTable mockTable() {
+ List<Column> schema = Arrays.asList(
+ new Column("time", PrimitiveType.DATETIME, true),
+ new Column("securityid", PrimitiveType.INT, true),
+ new Column("EV", PrimitiveType.DOUBLE, true));
+ Map<String, Column> nameToColumn = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ for (Column column : schema) {
+ nameToColumn.put(column.getName(), column);
+ }
+
+ OlapTable table = Mockito.mock(OlapTable.class);
+ Mockito.when(table.getName()).thenReturn("t_upper");
+ Mockito.when(table.getBaseSchema()).thenReturn(schema);
+ Mockito.when(table.getBaseSchema(false)).thenReturn(schema);
+ Mockito.when(table.getBaseSchema(true)).thenReturn(schema);
+ Mockito.when(table.getColumn(Mockito.anyString()))
+ .thenAnswer(invocation ->
nameToColumn.get(invocation.getArgument(0)));
+ Mockito.when(table.hasSequenceCol()).thenReturn(false);
+ Mockito.when(table.getSequenceMapCol()).thenReturn(null);
+ Mockito.when(table.hasSkipBitmapColumn()).thenReturn(false);
+ Mockito.when(table.getKeysType()).thenReturn(KeysType.UNIQUE_KEYS);
+
Mockito.when(table.getFullQualifiers()).thenReturn(ImmutableList.of("internal",
"db", "t_upper"));
+ return table;
+ }
+
+ private void assertSlot(NereidsParamCreateContext context, String name,
PrimitiveType type) {
+ SlotReference slot = context.scanSlots.stream()
+ .filter(scanSlot -> scanSlot.getName().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("missing scan slot " +
name));
+ Assertions.assertEquals(type,
slot.getOriginalColumn().get().getDataType());
+ }
+}
diff --git
a/regression-test/data/load_p0/stream_load/arrow_uppercase_column.arrow
b/regression-test/data/load_p0/stream_load/arrow_uppercase_column.arrow
new file mode 100644
index 00000000000..b8bebcf3d3e
Binary files /dev/null and
b/regression-test/data/load_p0/stream_load/arrow_uppercase_column.arrow differ
diff --git
a/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
b/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
new file mode 100644
index 00000000000..31344c45f9d
--- /dev/null
+++
b/regression-test/data/load_p0/stream_load/test_arrow_stream_load_uppercase_column.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !explicit --
+1:2.5
+2:3.5
+
+-- !auto --
+1:2.5
+2:3.5
diff --git
a/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
b/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
new file mode 100644
index 00000000000..16e3fe4c861
--- /dev/null
+++
b/regression-test/suites/load_p0/stream_load/test_arrow_stream_load_uppercase_column.groovy
@@ -0,0 +1,89 @@
+// 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.
+
+suite("test_arrow_stream_load_uppercase_column", "p0") {
+ sql """DROP TABLE IF EXISTS
test_arrow_stream_load_uppercase_column_explicit"""
+ sql """
+ CREATE TABLE test_arrow_stream_load_uppercase_column_explicit (
+ `id` INT,
+ `EV` DOUBLE
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+
+ streamLoad {
+ table "test_arrow_stream_load_uppercase_column_explicit"
+ set 'format', 'arrow'
+ set 'columns', 'id,EV'
+ file 'arrow_uppercase_column.arrow'
+ time 10000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${result}".toString())
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(2, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+
+ sql "sync"
+ order_qt_explicit """
+ SELECT concat(cast(id AS STRING), ':', cast(EV AS STRING))
+ FROM test_arrow_stream_load_uppercase_column_explicit
+ ORDER BY id
+ """
+
+ sql """DROP TABLE IF EXISTS test_arrow_stream_load_uppercase_column_auto"""
+ sql """
+ CREATE TABLE test_arrow_stream_load_uppercase_column_auto (
+ `id` INT,
+ `EV` DOUBLE
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`)
+ DISTRIBUTED BY HASH(`id`) BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+ """
+
+ streamLoad {
+ table "test_arrow_stream_load_uppercase_column_auto"
+ set 'format', 'arrow'
+ file 'arrow_uppercase_column.arrow'
+ time 10000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${result}".toString())
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(2, json.NumberTotalRows)
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+
+ sql "sync"
+ order_qt_auto """
+ SELECT concat(cast(id AS STRING), ':', cast(EV AS STRING))
+ FROM test_arrow_stream_load_uppercase_column_auto
+ ORDER BY id
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
