From 8a6165186abaa45d709f58c76020f6c6f1ea605f Mon Sep 17 00:00:00 2001 From: Refrain Date: Fri, 4 Sep 2026 05:44:15 +0800 Subject: [PATCH] [fix](load) Preserve routine load CSV header formats ### What problem does this PR solve? Issue Number: None Related PR: None Problem Summary: Nereids routine load normalized csv_with_names and csv_with_names_and_types to plain CSV while creating the job, then omitted the header type during task planning. Header rows were consequently parsed as data and counted as errors. Preserve the validated CSV header format in job properties and forward it to the Nereids data description. ### Release note Routine Load now honors csv_with_names and csv_with_names_and_types without treating header rows as data. ### Check List (For Author) - Test: Unit Test - Extended KafkaRoutineLoadJobTest coverage for plain and header-aware CSV formats; not run per request. - Behavior changed: Yes. Routine Load skips CSV header rows according to the configured format. - Does this need documentation: No --- .../load/routineload/RoutineLoadJob.java | 4 +- .../load/NereidsRoutineLoadTaskInfo.java | 6 ++ .../routineload/KafkaRoutineLoadJobTest.java | 71 ++++++++++++------- 3 files changed, 54 insertions(+), 27 deletions(-) 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> partitionIdToOffset = Lists.newArrayList(); - List 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> partitionIdToOffset = Lists.newArrayList(); + List 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 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 kafkaPartitionResult = Deencapsulation.getField(kafkaRoutineLoadJob, "customKafkaPartitions"); - Assert.assertEquals(kafkaPartitionString, Joiner.on(",").join(kafkaPartitionResult)); - Assert.assertEquals(sequenceColumnName, kafkaRoutineLoadJob.getSequenceCol()); } }