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..9adbd13eea0e85 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 @@ -282,10 +282,6 @@ public boolean isFinalState() { @SerializedName("ccid") private String cloudClusterId; - protected byte enclose = 0; - - protected byte escape = 0; - protected boolean emptyFieldAsNull = false; // use for cloud cluster mode @@ -415,8 +411,6 @@ protected void setOptional(CreateRoutineLoadInfo info) throws UserException { new String(new byte[]{csvFileFormatProperties.getEscape()})); jobProperties.put(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL, String.valueOf(csvFileFormatProperties.getEmptyFieldAsNull())); - this.enclose = csvFileFormatProperties.getEnclose(); - this.escape = csvFileFormatProperties.getEscape(); this.emptyFieldAsNull = csvFileFormatProperties.getEmptyFieldAsNull(); } else if (fileFormatProperties instanceof JsonFileFormatProperties) { JsonFileFormatProperties jsonFileFormatProperties = (JsonFileFormatProperties) fileFormatProperties; @@ -619,11 +613,13 @@ public Separator getLineDelimiter() { } public byte getEnclose() { - return enclose; + String value = jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE); + return Strings.isNullOrEmpty(value) ? 0 : (byte) value.charAt(0); } public byte getEscape() { - return escape; + String value = jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE); + return Strings.isNullOrEmpty(value) ? 0 : value.getBytes()[0]; } public boolean getEmptyFieldAsNull() { diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java index 885021440351d7..bda01118206cbb 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java @@ -1119,7 +1119,8 @@ public NereidsRoutineLoadTaskInfo toNereidsRoutineLoadTaskInfo() throws UserExce } return new NereidsRoutineLoadTaskInfo(execMemLimit, new HashMap<>(jobProperties), maxBatchIntervalS, partitionNamesInfo, mergeType, deleteCondition, sequenceCol, maxFilterRatio, importColumnDescs, - precedingFilter, whereExpr, columnSeparator, lineDelimiter, enclose, escape, sendBatchParallelism, + precedingFilter, whereExpr, columnSeparator, lineDelimiter, getEnclose(), getEscape(), + sendBatchParallelism, loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java index 7cebc3f5165b49..a5615ba883529d 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java +++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java @@ -875,7 +875,8 @@ public NereidsRoutineLoadTaskInfo toNereidsRoutineLoadTaskInfo() throws UserExce } return new NereidsRoutineLoadTaskInfo(execMemLimit, new HashMap<>(jobProperties), maxBatchIntervalS, partitionNamesInfo, mergeType, deleteCondition, sequenceCol, maxFilterRatio, importColumnDescs, - precedingFilter, whereExpr, columnSeparator, lineDelimiter, enclose, escape, sendBatchParallelism, + precedingFilter, whereExpr, columnSeparator, lineDelimiter, getEnclose(), getEscape(), + sendBatchParallelism, loadToSingleTablet, uniqueKeyUpdateMode, partialUpdateNewKeyPolicy, memtableOnSinkNode); } 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..8db457efb51100 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.CsvFileFormatProperties; import org.apache.doris.load.RoutineLoadDesc; import org.apache.doris.load.loadv2.LoadTask; import org.apache.doris.load.routineload.kafka.KafkaConfiguration; @@ -40,10 +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.NereidsRoutineLoadTaskInfo; +import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand; 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.persist.AlterRoutineLoadJobOperationLog; +import org.apache.doris.persist.EditLog; import org.apache.doris.qe.ConnectContext; import org.apache.doris.thrift.TResourceInfo; import org.apache.doris.thrift.TRoutineLoadTask; @@ -272,6 +277,45 @@ public void testUpdateProgressWarnsWhenReadCommittedTaskHasZeroRowsAndLag() thro Assert.assertTrue(otherMsg.contains("some records may be in uncommitted transactions")); } + @Test + public void testAlterCsvParserPropertiesUpdateNewTasksAndReplay() throws Exception { + KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L, + 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN); + Map jobProperties = Deencapsulation.getField(routineLoadJob, "jobProperties"); + jobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE, "~"); + jobProperties.put(CsvFileFormatProperties.PROP_ESCAPE, "!"); + + Map alteredProperties = Maps.newHashMap(); + alteredProperties.put(CsvFileFormatProperties.PROP_ENCLOSE, "^"); + alteredProperties.put(CsvFileFormatProperties.PROP_ESCAPE, "?"); + Deencapsulation.setField(routineLoadJob, "state", RoutineLoadJob.JobState.PAUSED); + AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class); + Mockito.when(command.getAnalyzedJobProperties()).thenReturn(alteredProperties); + Mockito.when(command.getDataSourceProperties()).thenReturn(null); + Env env = Mockito.mock(Env.class); + Mockito.when(env.getEditLog()).thenReturn(Mockito.mock(EditLog.class)); + try (MockedStatic envStatic = Mockito.mockStatic(Env.class)) { + envStatic.when(Env::getCurrentEnv).thenReturn(env); + routineLoadJob.modifyProperties(command); + } + + Assert.assertEquals((byte) '^', routineLoadJob.getEnclose()); + Assert.assertEquals((byte) '?', routineLoadJob.getEscape()); + NereidsRoutineLoadTaskInfo taskInfo = routineLoadJob.toNereidsRoutineLoadTaskInfo(); + Assert.assertEquals((byte) '^', taskInfo.getEnclose()); + Assert.assertEquals((byte) '?', taskInfo.getEscape()); + + KafkaRoutineLoadJob replayJob = new KafkaRoutineLoadJob(2L, "replay_job", 1L, + 1L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN); + Map replayJobProperties = Deencapsulation.getField(replayJob, "jobProperties"); + replayJobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE, "~"); + replayJobProperties.put(CsvFileFormatProperties.PROP_ESCAPE, "!"); + replayJob.replayModifyProperties(new AlterRoutineLoadJobOperationLog( + replayJob.getId(), alteredProperties, null)); + Assert.assertEquals((byte) '^', replayJob.getEnclose()); + Assert.assertEquals((byte) '?', replayJob.getEscape()); + } + @Test public void testDisplayCustomPropertiesMasksKafkaSecrets() { KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L, diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load_property.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load_property.groovy index cd10af8dfe5fd1..c24eb1ea33dace 100644 --- a/regression-test/suites/load_p0/routine_load/test_routine_load_property.groovy +++ b/regression-test/suites/load_p0/routine_load/test_routine_load_property.groovy @@ -139,7 +139,24 @@ suite("test_routine_load_property","p0") { sql "pause routine load for ${jobName}" def res = sql "show routine load for ${jobName}" log.info("routine load job properties: ${res[0][11].toString()}".toString()) - sql "ALTER ROUTINE LOAD FOR ${jobName} PROPERTIES(\"enclose\" = \"g\");" + sql "ALTER ROUTINE LOAD FOR ${jobName} PROPERTIES(\"enclose\" = \"^\", \"escape\" = \"?\");" + sql "truncate table ${tableName}" + + def alteredProps = new Properties() + alteredProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "${kafka_broker}".toString()) + alteredProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringSerializer") + alteredProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, + "org.apache.kafka.common.serialization.StringSerializer") + alteredProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "10000") + alteredProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, "10000") + def alteredProducer = new KafkaProducer<>(alteredProps) + try { + alteredProducer.send(new ProducerRecord<>(kafkaCsvTpoics[0], null, + "1,^ab,ced^,2023-07-15,d?e,2023-07-20 05:48:31,\"ghi\"")).get() + } finally { + alteredProducer.close() + } sql "resume routine load for ${jobName}" count = 0 while (true) { @@ -152,9 +169,7 @@ suite("test_routine_load_property","p0") { break } if (count >= 120) { - log.error("routine load can not visible for long time") - assertEquals(20, res[0][0]) - break + throw new IllegalStateException("altered routine load data can not be visible for long time") } sleep(1000) count++ @@ -209,4 +224,4 @@ suite("test_routine_load_property","p0") { sql "stop routine load for ${jobName}" } } -} \ No newline at end of file +}