From 71b7726253fa44a6be5d85ec31710f3a4fb1c8fe Mon Sep 17 00:00:00 2001 From: Refrain Date: Fri, 4 Sep 2026 05:01:47 +0800 Subject: [PATCH] [fix](load) Pass CSV parser properties to multi-table tasks ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Multi-table Routine Load creates per-table stream-load plans on the backend, but the task request did not carry CSV enclose, escape, or empty-field-as-null settings. As a result, dynamic-table CSV rows that depend on those settings were parsed with defaults. Add optional task fields and preserve them through the backend task context into each per-table planning request. ### Release note Multi-table Routine Load now applies enclose, escape, and empty_field_as_null when parsing CSV data. ### Check List (For Author) - Test: Unit Test (added FE task coverage; not run per request) - Behavior changed: Yes. Multi-table CSV Routine Load tasks pass their configured parser properties to per-table plans. - Does this need documentation: No --- be/src/io/fs/multi_table_pipe.cpp | 5 +++ .../routine_load_task_executor.cpp | 9 +++++ be/src/load/stream_load/stream_load_context.h | 3 ++ .../load/routineload/kafka/KafkaTaskInfo.java | 5 +++ .../routineload/KafkaRoutineLoadJobTest.java | 35 +++++++++++++++++++ gensrc/thrift/BackendService.thrift | 6 ++++ 6 files changed, 63 insertions(+) diff --git a/be/src/io/fs/multi_table_pipe.cpp b/be/src/io/fs/multi_table_pipe.cpp index 8a8d9fd3393e87..2f3a525ff6f332 100644 --- a/be/src/io/fs/multi_table_pipe.cpp +++ b/be/src/io/fs/multi_table_pipe.cpp @@ -186,6 +186,11 @@ Status MultiTablePipe::request_and_exec_plans() { request.__isset.table_names = true; request.txnId = _ctx->txn_id; request.formatType = _ctx->format; + if (_ctx->format == TFileFormatType::FORMAT_CSV_PLAIN) { + request.__set_enclose(_ctx->enclose); + request.__set_escape(_ctx->escape); + request.__set_empty_field_as_null(_ctx->empty_field_as_null); + } request.__set_compress_type(_ctx->compress_type); request.__set_header_type(_ctx->header_type); request.__set_loadId((pair.second->id).to_thrift()); diff --git a/be/src/load/routine_load/routine_load_task_executor.cpp b/be/src/load/routine_load/routine_load_task_executor.cpp index f1d126f8643bd6..4aff32324006f9 100644 --- a/be/src/load/routine_load/routine_load_task_executor.cpp +++ b/be/src/load/routine_load/routine_load_task_executor.cpp @@ -349,6 +349,15 @@ Status RoutineLoadTaskExecutor::submit_task(const TRoutineLoadTask& task) { if (task.__isset.format) { ctx->format = task.format; } + if (task.__isset.enclose) { + ctx->enclose = task.enclose; + } + if (task.__isset.escape) { + ctx->escape = task.escape; + } + if (task.__isset.empty_field_as_null) { + ctx->empty_field_as_null = task.empty_field_as_null; + } // the routine load task'txn has already began in FE. // so it need to rollback if encounter error. ctx->need_rollback = true; diff --git a/be/src/load/stream_load/stream_load_context.h b/be/src/load/stream_load/stream_load_context.h index 304cc0b9c622a6..11d829c9460988 100644 --- a/be/src/load/stream_load/stream_load_context.h +++ b/be/src/load/stream_load/stream_load_context.h @@ -232,6 +232,9 @@ class StreamLoadContext { bool use_streaming = false; TFileFormatType::type format = TFileFormatType::FORMAT_CSV_PLAIN; TFileCompressType::type compress_type = TFileCompressType::UNKNOWN; + int8_t enclose = 0; + int8_t escape = 0; + bool empty_field_as_null = false; bool group_commit = false; std::string group_commit_mode = ""; diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java index 98aabe63926483..7ea951f4c33c6e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaTaskInfo.java @@ -121,6 +121,11 @@ public TRoutineLoadTask createRoutineLoadTask() throws UserException { tRoutineLoadTask.setFormat(TFileFormatType.FORMAT_JSON); } else { tRoutineLoadTask.setFormat(TFileFormatType.FORMAT_CSV_PLAIN); + if (isMultiTable) { + tRoutineLoadTask.setEnclose(routineLoadJob.getEnclose()); + tRoutineLoadTask.setEscape(routineLoadJob.getEscape()); + tRoutineLoadTask.setEmptyFieldAsNull(routineLoadJob.getEmptyFieldAsNull()); + } } tRoutineLoadTask.setMemtableOnSinkNode(routineLoadJob.isMemtableOnSinkNode()); tRoutineLoadTask.setQualifiedUser(routineLoadJob.getUserIdentity().getQualifiedUser()); diff --git a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java index 7f0c8588372403..7c10a29f917e8e 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java @@ -272,6 +272,41 @@ public void testUpdateProgressWarnsWhenReadCommittedTaskHasZeroRowsAndLag() thro Assert.assertTrue(otherMsg.contains("some records may be in uncommitted transactions")); } + @Test + public void testMultiTableCsvTaskIncludesParserProperties() throws Exception { + RoutineLoadManager routineLoadManager = Mockito.mock(RoutineLoadManager.class); + Env env = Mockito.mock(Env.class); + InternalCatalog internalCatalog = Mockito.mock(InternalCatalog.class); + Database database = Mockito.mock(Database.class); + + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + envStatic.when(Env::getCurrentInternalCatalog).thenReturn(internalCatalog); + Mockito.when(env.getRoutineLoadManager()).thenReturn(routineLoadManager); + Mockito.when(internalCatalog.getDbOrMetaException(1L)).thenReturn(database); + Mockito.when(database.getFullName()).thenReturn("db1"); + + KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "multi_table_job", 1L, + "127.0.0.1:9020", "topic1", UserIdentity.ADMIN, true); + Deencapsulation.setField(routineLoadJob, "enclose", (byte) '^'); + Deencapsulation.setField(routineLoadJob, "escape", (byte) '?'); + Map jobProperties = Deencapsulation.getField(routineLoadJob, "jobProperties"); + jobProperties.put(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL, "true"); + Mockito.when(routineLoadManager.getJob(1L)).thenReturn(routineLoadJob); + + KafkaTaskInfo taskInfo = new KafkaTaskInfo(new UUID(1, 1), 1L, 20000, + Maps.newHashMap(), true, 1000, false); + TRoutineLoadTask task = Deencapsulation.invoke(taskInfo, "createRoutineLoadTask"); + + Assert.assertTrue(task.isSetEnclose()); + Assert.assertEquals((byte) '^', task.getEnclose()); + Assert.assertTrue(task.isSetEscape()); + Assert.assertEquals((byte) '?', task.getEscape()); + Assert.assertTrue(task.isSetEmptyFieldAsNull()); + Assert.assertTrue(task.isEmptyFieldAsNull()); + } + } + @Test public void testDisplayCustomPropertiesMasksKafkaSecrets() { KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L, diff --git a/gensrc/thrift/BackendService.thrift b/gensrc/thrift/BackendService.thrift index 19d80620702f5c..273af9a6facb2d 100644 --- a/gensrc/thrift/BackendService.thrift +++ b/gensrc/thrift/BackendService.thrift @@ -86,6 +86,12 @@ struct TRoutineLoadTask { 18: optional string qualified_user 19: optional string cloud_cluster 20: optional TKinesisLoadInfo kinesis_load_info + // only valid for multi-table CSV routine load + 21: optional i8 enclose + // only valid for multi-table CSV routine load + 22: optional i8 escape + // only valid for multi-table CSV routine load + 23: optional bool empty_field_as_null } struct TKafkaMetaProxyRequest {