diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 9873368f405114..28a72df1817881 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -408,7 +408,9 @@ protected void setOptional(CreateRoutineLoadInfo info) throws UserException {
         FileFormatProperties fileFormatProperties = info.getFileFormatProperties();
         if (fileFormatProperties instanceof CsvFileFormatProperties) {
             CsvFileFormatProperties csvFileFormatProperties = (CsvFileFormatProperties) fileFormatProperties;
-            jobProperties.put(FileFormatProperties.PROP_FORMAT, "csv");
+            String headerType = csvFileFormatProperties.getHeaderType();
+            jobProperties.put(FileFormatProperties.PROP_FORMAT, headerType.isEmpty()
+                    ? FileFormatProperties.FORMAT_CSV : headerType);
             jobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE,
                     new String(new byte[]{csvFileFormatProperties.getEnclose()}));
             jobProperties.put(CsvFileFormatProperties.PROP_ESCAPE,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
index ef159cfb6f494a..1a324e8f785a78 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java
@@ -22,6 +22,7 @@
 import org.apache.doris.common.Config;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
+import org.apache.doris.datasource.property.fileformat.FileFormatProperties;
 import org.apache.doris.load.loadv2.LoadTask;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.plans.commands.LoadCommand;
@@ -305,6 +306,11 @@ public boolean isLoadToSingleTablet() {
 
     @Override
     public String getHeaderType() {
+        String format = getFormat();
+        if (format.equals(FileFormatProperties.FORMAT_CSV_WITH_NAMES)
+                || format.equals(FileFormatProperties.FORMAT_CSV_WITH_NAMES_AND_TYPES)) {
+            return format;
+        }
         return "";
     }
 
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..9bd66ef2087753 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
@@ -32,6 +32,7 @@
 import org.apache.doris.common.jmockit.Deencapsulation;
 import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.datasource.kafka.KafkaUtil;
+import org.apache.doris.datasource.property.fileformat.FileFormatProperties;
 import org.apache.doris.load.RoutineLoadDesc;
 import org.apache.doris.load.loadv2.LoadTask;
 import org.apache.doris.load.routineload.kafka.KafkaConfiguration;
@@ -40,11 +41,14 @@
 import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
 import org.apache.doris.load.routineload.kafka.KafkaTaskInfo;
 import org.apache.doris.mysql.privilege.MockedAuth;
+import org.apache.doris.nereids.load.NereidsDataDescription;
+import org.apache.doris.nereids.load.NereidsRoutineLoadTaskInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.LabelNameInfo;
 import org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
 import org.apache.doris.nereids.trees.plans.commands.load.LoadSeparator;
 import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.thrift.TFileFormatType;
 import org.apache.doris.thrift.TResourceInfo;
 import org.apache.doris.thrift.TRoutineLoadTask;
 
@@ -637,33 +641,48 @@ public void testFromCreateStmt() throws UserException {
                         return pairList;
                     });
 
-            CreateRoutineLoadInfo createRoutineLoadInfo = initCreateRoutineLoadInfo();
-            createRoutineLoadInfo.validate(connectContext);
-            RoutineLoadDesc routineLoadDesc = new RoutineLoadDesc(columnSeparator, null, null, null, null, partitionNames, null,
-                    LoadTask.MergeType.APPEND, sequenceColumnName);
-            Deencapsulation.setField(createRoutineLoadInfo, "routineLoadDesc", routineLoadDesc);
-            List<Pair<Integer, Long>> partitionIdToOffset = Lists.newArrayList();
-            List<PartitionInfo> kafkaPartitionInfoList = Lists.newArrayList();
-            for (String s : kafkaPartitionString.split(",")) {
-                partitionIdToOffset.add(Pair.of(Integer.valueOf(s), 0L));
-                PartitionInfo partitionInfo = new PartitionInfo(topicName, Integer.valueOf(s), null, null, null);
-                kafkaPartitionInfoList.add(partitionInfo);
+            for (String csvFormat : Arrays.asList(FileFormatProperties.FORMAT_CSV,
+                    FileFormatProperties.FORMAT_CSV_WITH_NAMES,
+                    FileFormatProperties.FORMAT_CSV_WITH_NAMES_AND_TYPES)) {
+                CreateRoutineLoadInfo createRoutineLoadInfo = initCreateRoutineLoadInfo();
+                createRoutineLoadInfo.getJobProperties().put(FileFormatProperties.PROP_FORMAT, csvFormat);
+                createRoutineLoadInfo.validate(connectContext);
+                RoutineLoadDesc routineLoadDesc = new RoutineLoadDesc(columnSeparator, null, null, null, null,
+                        partitionNames, null, LoadTask.MergeType.APPEND, sequenceColumnName);
+                Deencapsulation.setField(createRoutineLoadInfo, "routineLoadDesc", routineLoadDesc);
+                List<Pair<Integer, Long>> partitionIdToOffset = Lists.newArrayList();
+                List<PartitionInfo> kafkaPartitionInfoList = Lists.newArrayList();
+                for (String s : kafkaPartitionString.split(",")) {
+                    partitionIdToOffset.add(Pair.of(Integer.valueOf(s), 0L));
+                    PartitionInfo partitionInfo = new PartitionInfo(topicName, Integer.valueOf(s), null, null, null);
+                    kafkaPartitionInfoList.add(partitionInfo);
+                }
+                KafkaDataSourceProperties dsProperties = new KafkaDataSourceProperties(null);
+                dsProperties.setKafkaPartitionOffsets(partitionIdToOffset);
+                Deencapsulation.setField(dsProperties, "brokerList", serverAddress);
+                Deencapsulation.setField(dsProperties, "topic", topicName);
+                Deencapsulation.setField(createRoutineLoadInfo, "dataSourceProperties", dsProperties);
+
+                KafkaRoutineLoadJob kafkaRoutineLoadJob =
+                        KafkaRoutineLoadJob.fromCreateInfo(createRoutineLoadInfo, connectContext);
+                Assert.assertEquals(jobName, kafkaRoutineLoadJob.getName());
+                Assert.assertEquals(dbId, kafkaRoutineLoadJob.getDbId());
+                Assert.assertEquals(tableId, kafkaRoutineLoadJob.getTableId());
+                Assert.assertEquals(serverAddress, Deencapsulation.getField(kafkaRoutineLoadJob, "brokerList"));
+                Assert.assertEquals(topicName, Deencapsulation.getField(kafkaRoutineLoadJob, "topic"));
+                List<Integer> kafkaPartitionResult =
+                        Deencapsulation.getField(kafkaRoutineLoadJob, "customKafkaPartitions");
+                Assert.assertEquals(kafkaPartitionString, Joiner.on(",").join(kafkaPartitionResult));
+                Assert.assertEquals(sequenceColumnName, kafkaRoutineLoadJob.getSequenceCol());
+                Assert.assertEquals(csvFormat, kafkaRoutineLoadJob.getFormat());
+
+                NereidsRoutineLoadTaskInfo taskInfo = kafkaRoutineLoadJob.toNereidsRoutineLoadTaskInfo();
+                String expectedHeaderType = csvFormat.equals(FileFormatProperties.FORMAT_CSV) ? "" : csvFormat;
+                Assert.assertEquals(TFileFormatType.FORMAT_CSV_PLAIN, taskInfo.getFormatType());
+                Assert.assertEquals(expectedHeaderType, taskInfo.getHeaderType());
+                Assert.assertTrue(new NereidsDataDescription(tableNameString, taskInfo).toSql()
+                        .contains("FORMAT AS '" + csvFormat + "'"));
             }
-            KafkaDataSourceProperties dsProperties = new KafkaDataSourceProperties(null);
-            dsProperties.setKafkaPartitionOffsets(partitionIdToOffset);
-            Deencapsulation.setField(dsProperties, "brokerList", serverAddress);
-            Deencapsulation.setField(dsProperties, "topic", topicName);
-            Deencapsulation.setField(createRoutineLoadInfo, "dataSourceProperties", dsProperties);
-
-            KafkaRoutineLoadJob kafkaRoutineLoadJob = KafkaRoutineLoadJob.fromCreateInfo(createRoutineLoadInfo, connectContext);
-            Assert.assertEquals(jobName, kafkaRoutineLoadJob.getName());
-            Assert.assertEquals(dbId, kafkaRoutineLoadJob.getDbId());
-            Assert.assertEquals(tableId, kafkaRoutineLoadJob.getTableId());
-            Assert.assertEquals(serverAddress, Deencapsulation.getField(kafkaRoutineLoadJob, "brokerList"));
-            Assert.assertEquals(topicName, Deencapsulation.getField(kafkaRoutineLoadJob, "topic"));
-            List<Integer> kafkaPartitionResult = Deencapsulation.getField(kafkaRoutineLoadJob, "customKafkaPartitions");
-            Assert.assertEquals(kafkaPartitionString, Joiner.on(",").join(kafkaPartitionResult));
-            Assert.assertEquals(sequenceColumnName, kafkaRoutineLoadJob.getSequenceCol());
         }
     }