[fix](load) Pass CSV parser properties to multi-table tasks

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());

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;

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 = "";
 

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());

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<Env> 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<String, String> 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,

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 {