This is an automated email from the ASF dual-hosted git repository.
morrysnow pushed a commit to branch branch-3.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-3.1 by this push:
new 0da40bc746e branch-3.1: [fix](broker load)fix broker load fail when
<column from path> column already exists in file #58579 (#58904)
0da40bc746e is described below
commit 0da40bc746ef2a385e0dcdd4cd318bd27e9f712d
Author: daidai <[email protected]>
AuthorDate: Mon Dec 15 11:17:23 2025 +0800
branch-3.1: [fix](broker load)fix broker load fail when <column from path>
column already exists in file #58579 (#58904)
bp #58579
---
be/src/vec/exec/scan/vfile_scanner.cpp | 16 +-
.../broker_load/test_load_data_from_path.groovy | 215 +++++++++++++++++++++
2 files changed, 230 insertions(+), 1 deletion(-)
diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp
b/be/src/vec/exec/scan/vfile_scanner.cpp
index 3cfdc3b159e..bb5f0f257d2 100644
--- a/be/src/vec/exec/scan/vfile_scanner.cpp
+++ b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -1314,7 +1314,21 @@ Status VFileScanner::_get_next_reader() {
std::unordered_map<std::string, TypeDescriptor> name_to_col_type;
RETURN_IF_ERROR(_cur_reader->get_columns(&name_to_col_type,
&_missing_cols));
for (const auto& [col_name, col_type] : name_to_col_type) {
- _slot_lower_name_to_col_type.emplace(to_lower(col_name), col_type);
+ auto col_name_lower = to_lower(col_name);
+ if (_partition_col_descs.contains(col_name_lower)) {
+ /*
+ * `_slot_lower_name_to_col_type` is used by `_init_src_block`
and `_cast_to_input_block` during LOAD to
+ * generate columns of the corresponding type, which records
the columns existing in the file.
+ *
+ * When a column in `COLUMNS FROM PATH` exists in a file
column, the column type in the block will
+ * not match the slot type in `_output_tuple_desc`, causing an
error when
+ * Serde `deserialize_one_cell_from_json` fills the partition
values.
+ *
+ * So for partition column not need fill
_slot_lower_name_to_col_type.
+ */
+ continue;
+ }
+ _slot_lower_name_to_col_type.emplace(col_name_lower, col_type);
}
if (!_fill_partition_from_path &&
config::enable_iceberg_partition_column_fallback) {
diff --git
a/regression-test/suites/load_p0/broker_load/test_load_data_from_path.groovy
b/regression-test/suites/load_p0/broker_load/test_load_data_from_path.groovy
index e4c6a5d2eed..46fc29b2948 100644
--- a/regression-test/suites/load_p0/broker_load/test_load_data_from_path.groovy
+++ b/regression-test/suites/load_p0/broker_load/test_load_data_from_path.groovy
@@ -202,4 +202,219 @@ suite("test_load_columns_from_path", "load_p0") {
} finally {
sql """ DROP TABLE ${tableName} """
}
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+
+ sql """
+ CREATE TABLE ${tableName} (
+ k1 INT,
+ k2 INT,
+ k3 INT
+ )
+ DUPLICATE KEY(`k1`)
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1"
+ );
+ """
+
+ label = UUID.randomUUID().toString().replace("-", "2")
+ try {
+ sql """
+ LOAD LABEL ${label}
+ (
+ DATA
INFILE("s3://${s3BucketName}/load/k1=10/k2=20/test.parquet")
+ INTO TABLE ${tableName}
+ FORMAT AS "parquet"
+ (k1, k2, k3)
+ )
+ WITH S3
+ (
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "s3.endpoint" = "${s3Endpoint}",
+ "s3.region" = "${s3Region}"
+ )
+ """
+
+ // Wait for load job to finish
+ def maxRetry = 60
+ def result = ""
+ for (int i = 0; i < maxRetry; i++) {
+ result = sql_return_maparray "SHOW LOAD WHERE LABEL = '${label}'"
+ if (result[0].State == "FINISHED" || result[0]. State ==
"CANCELLED") {
+ break
+ }
+ sleep(1000)
+ }
+
+ // Check load job state
+ assertEquals("FINISHED", result[0].State)
+
+ // Verify the loaded data
+ def rowCount = sql "SELECT COUNT(*) FROM ${tableName}"
+ assertTrue(rowCount[0][0] > 0, "No data was loaded")
+
+ def pathData = sql "SELECT * FROM ${tableName} LIMIT 1"
+ logger.info("path data 1: " + pathData)
+ // 1 2 3
+ assertEquals(1, pathData[0][0])
+ assertEquals(2, pathData[0][1])
+ assertEquals(3, pathData[0][2])
+
+ } finally {
+ sql """ TRUNCATE TABLE ${tableName} """
+ }
+
+ label = UUID.randomUUID().toString().replace("-", "2")
+
+
+ try {
+ sql """
+ LOAD LABEL ${label}
+ (
+ DATA INFILE("s3://${s3BucketName}/load/k1=10/k2=20/test.orc")
+ INTO TABLE ${tableName}
+ FORMAT AS "orc"
+ (k1, k3)
+ COLUMNS FROM PATH AS (k2)
+ )
+ WITH S3
+ (
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "s3.endpoint" = "${s3Endpoint}",
+ "s3.region" = "${s3Region}"
+ )
+ """
+
+ // Wait for load job to finish
+ def maxRetry = 60
+ def result = ""
+ for (int i = 0; i < maxRetry; i++) {
+ result = sql_return_maparray "SHOW LOAD WHERE LABEL = '${label}'"
+ if (result[0].State == "FINISHED" || result[0]. State ==
"CANCELLED") {
+ break
+ }
+ sleep(1000)
+ }
+
+ // Check load job state
+ assertEquals("FINISHED", result[0].State)
+
+ // Verify the loaded data
+ def rowCount = sql "SELECT COUNT(*) FROM ${tableName}"
+ assertTrue(rowCount[0][0] > 0, "No data was loaded")
+
+ def pathData = sql "SELECT * FROM ${tableName} LIMIT 1"
+ logger.info("path data 2: " + pathData)
+ // 1 20 3
+ assertEquals(1, pathData[0][0])
+ assertEquals(20, pathData[0][1])
+ assertEquals(3, pathData[0][2])
+
+ } finally {
+ sql """ TRUNCATE TABLE ${tableName} """
+ }
+
+ label = UUID.randomUUID().toString().replace("-", "2")
+ try {
+ sql """
+ LOAD LABEL ${label}
+ (
+ DATA
INFILE("s3://${s3BucketName}/load/k1=10/k3=30/test.parquet")
+ INTO TABLE ${tableName}
+ FORMAT AS "parquet"
+ (k2)
+ COLUMNS FROM PATH AS (k1,k3)
+ )
+ WITH S3
+ (
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "s3.endpoint" = "${s3Endpoint}",
+ "s3.region" = "${s3Region}"
+ )
+ """
+
+ // Wait for load job to finish
+ def maxRetry = 60
+ def result = ""
+ for (int i = 0; i < maxRetry; i++) {
+ result = sql_return_maparray "SHOW LOAD WHERE LABEL = '${label}'"
+ if (result[0].State == "FINISHED" || result[0]. State ==
"CANCELLED") {
+ break
+ }
+ sleep(1000)
+ }
+
+ // Check load job state
+ assertEquals("FINISHED", result[0].State)
+
+ // Verify the loaded data
+ def rowCount = sql "SELECT COUNT(*) FROM ${tableName}"
+ assertTrue(rowCount[0][0] > 0, "No data was loaded")
+
+ def pathData = sql "SELECT * FROM ${tableName} LIMIT 1"
+ logger.info("path data 2: " + pathData)
+ // 10 2 30
+ assertEquals(10, pathData[0][0])
+ assertEquals(2, pathData[0][1])
+ assertEquals(30, pathData[0][2])
+
+ } finally {
+ sql """ TRUNCATE TABLE ${tableName} """
+ }
+
+
+ label = UUID.randomUUID().toString().replace("-", "2")
+ try {
+ sql """
+ LOAD LABEL ${label}
+ (
+ DATA INFILE("s3://${s3BucketName}/load/k1=10/k3=30/test.orc")
+ INTO TABLE ${tableName}
+ FORMAT AS "orc"
+ (k1,k2)
+ COLUMNS FROM PATH AS (k3)
+ )
+ WITH S3
+ (
+ "s3.access_key" = "${ak}",
+ "s3.secret_key" = "${sk}",
+ "s3.endpoint" = "${s3Endpoint}",
+ "s3.region" = "${s3Region}"
+ )
+ """
+
+ // Wait for load job to finish
+ def maxRetry = 60
+ def result = ""
+ for (int i = 0; i < maxRetry; i++) {
+ result = sql_return_maparray "SHOW LOAD WHERE LABEL = '${label}'"
+ if (result[0].State == "FINISHED" || result[0]. State ==
"CANCELLED") {
+ break
+ }
+ sleep(1000)
+ }
+
+ // Check load job state
+ assertEquals("FINISHED", result[0].State)
+
+ // Verify the loaded data
+ def rowCount = sql "SELECT COUNT(*) FROM ${tableName}"
+ assertTrue(rowCount[0][0] > 0, "No data was loaded")
+
+ def pathData = sql "SELECT * FROM ${tableName} LIMIT 1"
+ logger.info("path data 2: " + pathData)
+ assertEquals(1, pathData[0][0])
+ assertEquals(2, pathData[0][1])
+ assertEquals(30, pathData[0][2])
+
+
+ // [[1, 2, 30]]
+ } finally {
+ sql """ TRUNCATE TABLE ${tableName} """
+ }
+
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
