morningman commented on code in PR #70121:
URL: https://github.com/apache/doris/pull/70121#discussion_r4463177785
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -2171,6 +2196,51 @@ protected void modifyCommonJobProperties(Map<String,
String> jobProperties) thro
this.jobProperties.put(CreateRoutineLoadInfo.PARTIAL_COLUMNS,
String.valueOf(isPartialUpdate));
this.jobProperties.put(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE,
uniqueKeyUpdateMode.name());
}
+
+ // Tasks take enclose and escape from these fields, not from
jobProperties, where the caller puts the
+ // new values
+ if (jobProperties.containsKey(CsvFileFormatProperties.PROP_ENCLOSE)) {
+ this.enclose = csvControlByte(CsvFileFormatProperties.PROP_ENCLOSE,
+ jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE),
this.enclose);
+ }
+ if (jobProperties.containsKey(CsvFileFormatProperties.PROP_ESCAPE)) {
+ this.escape = csvControlByte(CsvFileFormatProperties.PROP_ESCAPE,
+ jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE),
this.escape);
+ }
+ }
+
+ // The byte an enclose or escape value stands for, as CREATE takes it.
ALTER of an older version did not check
+ // the value, so the metadata can hold one that CREATE rejects. Such a
value never reached the job, and it does
+ // not now: the job keeps the byte it has, where an exception would fail
the replay of the metadata.
+ private byte csvControlByte(String key, String value, byte current) {
+ try {
+ return key.equals(CsvFileFormatProperties.PROP_ENCLOSE)
+ ? CsvFileFormatProperties.parseEnclose(value) :
CsvFileFormatProperties.parseEscape(value);
+ } catch (org.apache.doris.nereids.exceptions.AnalysisException e) {
+ LOG.warn("routine load job {} keeps its {} instead of the invalid
value '{}': {}",
+ id, key, value, e.getMessage());
+ return current;
+ }
+ }
+
+ // What CREATE records for enclose or escape: the value the statement
gave, which the replay parses as CREATE did.
+ // The byte itself does not survive a String: new String(new byte[]{b})
decodes every byte from 0x80 up to U+FFFD.
+ // Without a value the byte is 0, recorded as it always was.
+ private static String csvControlValue(CreateRoutineLoadInfo info, String
key, byte value) {
+ String given = info.getJobProperties().get(key);
+ return Strings.isNullOrEmpty(given) ? new String(new byte[]{value}) :
given;
+ }
+
+ // csvControlByte for a value CREATE or ALTER recorded. CREATE of an older
version recorded the byte, so a byte
+ // from 0x80 up came back as U+FFFD, which would now parse to 0xFD for
enclose and 0xEF for escape. That byte is
+ // lost: the job keeps the default, which is what it has run with after
every restart.
+ private byte recordedCsvControlByte(String key, String value, byte
current) {
+ if (value.equals("\uFFFD")) {
Review Comment:
Fixed in `73dbcc1d452`.
Confirmed. The fallback could not tell a U+FFFD that an older CREATE
recorded for a lost byte from one the user wrote, so a job created with
`"escape" = "�"` ran with escape 0xEF on the master and with no escape on a
follower as soon as the follower replayed the CREATE.
CREATE and ALTER ROUTINE LOAD now reject U+FFFD as `enclose` or `escape`
(`CreateRoutineLoadInfo#checkCsvControlValue`), so the metadata holds it only
from an older version: a byte its CREATE lost, or a value its ALTER stored
unchecked, which never reached the job either. The fallback in the replay stays
as it was. U+FFFD never stood for the byte the statement meant: it is what a
statement decodes to when its bytes are not UTF-8 (a Latin-1 client's `é`, for
example), and the parsers turn it into 0xFD and 0xEF, not into that byte. A
user who wants 0xFD as `enclose` can write `ý`, and 0xEF as `escape` any
character from U+F000 to U+FFFC.
`RoutineLoadJobReplayTest#testCreateRejectsTheReplacementCharacter` and
`AlterRoutineLoadCommandTest#testValidateRejectsTheReplacementCharacter` cover
both keys, and `test_routine_load_property` gets a `test { sql; exception }`
case for each key outside its Kafka guard, since CREATE checks the properties
before it connects to Kafka. Each new UT case fails when its check is removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
