Skip to content

[fix](fe) Use updated Routine Load CSV parser properties #67509

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
0AyanamiRei wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix](fe) Use updated Routine Load CSV parser properties #67509

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 4 additions & 8 deletions fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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() {
Expand Down
3 changes: 2 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
}
3 changes: 2 additions & 1 deletion ...e-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}

Expand Down
44 changes: 44 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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<String, String> jobProperties = Deencapsulation.getField(routineLoadJob, "jobProperties");
jobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE, "~");
jobProperties.put(CsvFileFormatProperties.PROP_ESCAPE, "!");

Map<String, String> 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<Env> 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<String, String> 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,
Expand Down
25 changes: 20 additions & 5 deletions regression-test/suites/load_p0/routine_load/test_routine_load_property.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand All @@ -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++
Expand Down Expand Up @@ -209,4 +224,4 @@ suite("test_routine_load_property","p0") {
sql "stop routine load for ${jobName}"
}
}
}
}