This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new bff68004fda [fix](fe) Keep routine load job properties across replay
and apply altered enclose and escape (#70121)
bff68004fda is described below
commit bff68004fda41c0649efc9de4c5d14df29069440
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Nov 5 07:42:33 2026 +0000
[fix](fe) Keep routine load job properties across replay and apply altered
enclose and escape (#70121)
### What problem does this PR solve?
Issue Number: None
Related PR: #70118
Problem Summary:
**In short.** When an FE rebuilds a routine load job from metadata (a
follower replaying the edit log, a restart, a new master), the job loses
`max_filter_ratio`, `exec_mem_limit`, `send_batch_parallelism`,
`load_to_single_tablet`, `enclose`, `escape` and the
`enable_memtable_on_sink_node` it was created with, and runs with
defaults from then on. Separately, `ALTER ROUTINE LOAD` on `enclose` /
`escape` never reaches the running job, and a multi-table CSV job never
applies `enclose` / `escape` at all. This PR restores the values when
the job is rebuilt, applies the altered ones, and hands both to the
tasks of multi-table jobs.
**Background.** A routine load job is persisted with GSON, which writes
only the fields annotated with `@SerializedName`; these seven are plain
fields. `gsonPostProcess` re-parses the stored CREATE statement, but
only to rebuild the `RoutineLoadDesc` (columns, filters, separators,
partitions, merge type), and restores the partial update settings from
the persisted `jobProperties` map. CREATE records the other values in
`jobProperties` as well (all but `exec_mem_limit`), and SHOW ROUTINE
LOAD prints them from there, so the loss does not show. Found while
making `SHOW CREATE ROUTINE LOAD` print these properties (#70118).
**The problem, and what it cost.** Measured on master `856194ece0e` with
a unit test that creates jobs from SQL and rebuilds them with a GSON
round trip on a thread without a session, as a follower or a restarting
FE does:
| | Job as created | Job rebuilt from metadata |
|---|---|---|
| `max_filter_ratio` = 0.5 | 0.5 | 1.0: up to every row may be filtered
instead of the job pausing |
| `exec_mem_limit` = 4294967296 | 4 GB | 2 GB |
| `send_batch_parallelism` = 3, or 4 from the creating session | 3 / 4 |
1 |
| `load_to_single_tablet` = true | true | false |
| `enclose` `"`, `escape` `\` | `"` / `\` | none: quoted CSV fields are
split at the separators inside the quotes |
| `enable_memtable_on_sink_node` of the creating session (on by default)
| true | false |
| `ALTER ... ("enclose" = "'", "escape" = "\|")` | tasks keep `"` / `\`
| SHOW ROUTINE LOAD shows the new values, tasks never use them |
| `ALTER ... ("enclose" = "ab")` | accepted, without effect | CREATE
rejects it |
The CSV and filter ratio cases change what gets loaded, without any
error, after an event (a failover, a restart) that has nothing to do
with the job.
A multi-table CSV job (one created without a target table) does not use
its `enclose` and `escape` even before it is rebuilt. Each of its tasks
is planned from the backend's per-table request, which does not carry
them, with the job's settings laid over it by
`NereidsStreamLoadTask#setMultiTableBaseTaskInfo`, which copies the
separators and the JSON properties but not these two. So its quoted
fields are split at the separators inside the quotes, whatever CREATE or
ALTER set.
**How this PR fixes it.**
- `gsonPostProcess` restores `max_filter_ratio`,
`send_batch_parallelism`, `load_to_single_tablet`, `enclose` and
`escape` from `jobProperties`, in the loop that already restores the
partial update settings from there. CREATE has recorded these values
since each property was introduced, and ALTER records the new ones; a
job created before one of them existed has no key and keeps the default,
as today. `exec_mem_limit`, which no ALTER changes and `jobProperties`
does not hold, is taken from the CREATE statement that `gsonPostProcess`
already re-parses.
- CREATE records the creating session's `enable_memtable_on_sink_node`
in the job's persisted session variables, next to `sql_mode`, and
`gsonPostProcess` restores it. A job created before this has no record
and keeps `false`, which is what it has run with after every restart so
far.
- `ALTER ROUTINE LOAD` checks `enclose` and `escape` the way CREATE
does, and `modifyCommonJobProperties`, which Kafka and Kinesis jobs and
the replay of an ALTER share, installs them on the job. The two rules
move into `CsvFileFormatProperties.parseEnclose` / `parseEscape`, which
`analyzeFileFormatProperties` now calls without any change in behavior.
- `NereidsStreamLoadTask#setMultiTableBaseTaskInfo` copies `enclose` and
`escape` too, so the tasks of a multi-table job use the values CREATE or
ALTER set, as those of a single-table job do.
- CREATE and ALTER ROUTINE LOAD take only an ASCII character as
`enclose` or `escape` (`CreateRoutineLoadInfo#checkCsvControlValue`),
which is what the documentation of both properties asks for: a
single-byte character. No other value works for a routine load. Each
task is planned from the job's byte turned into text with `new
String(new byte[]{b})` and parsed again (`NereidsDataDescription`, since
#50882), which keeps every ASCII byte but turns every byte from 0x80 up
into 0xFD as enclose and 0xEF as escape. A job created with `"escape" =
"é"` (0xC3) has therefore run with escape 0xEF from its first task on,
which drops the first byte of full-width characters such as `，` in every
string cell. And routine load validates its data as UTF-8 by default,
where a single byte from 0x80 up is absent or part of a character
anyway: 0xC3 as escape would cut the first byte out of every `é`. CREATE
records the byte with `new String(new byte[]{b})` as it always has,
which keeps every value it now takes.
- Values that an older version accepted do not break the rebuild. The
ASCII check runs only when a statement is admitted, not when
`gsonPostProcess` re-parses a stored CREATE statement, as the compute
group and workload group checks beside it already do; failing there
would turn the job CANCELLED for good. A value from an older version
that CREATE and ALTER now reject, one that is not ASCII (an older CREATE
recorded every byte from 0x80 up as U+FFFD) or one that an older ALTER
stored without any check, is ignored with a warning instead of failing
the replay, in the same way when the job is rebuilt and when the ALTER
is replayed. The replay of such an ALTER leaves the job's value as it
was, and a rebuilt job keeps no enclose or escape, which is what it has
run with after every restart so far; ALTER ROUTINE LOAD can set an ASCII
value again.
No new persisted field: the only addition is one key in the job's
existing session variable map, which older FEs ignore.
**Results.**
| | Before | After |
|---|---|---|
| declared properties above, job rebuilt from metadata | defaults | as
declared, also in what tasks are built from |
| `send_batch_parallelism` and `enable_memtable_on_sink_node` from the
creating session | 1 / false | the session's values |
| ALTER of `enclose` / `escape` | shown, never used | used by the next
task, kept when the job is rebuilt |
| `enclose` / `escape` of a multi-table CSV job, from CREATE or ALTER |
never used | used by its tasks |
| ALTER with `"enclose" = "ab"` | accepted | `enclose should not be
longer than one byte.` |
| CREATE or ALTER with an `enclose` / `escape` that is not ASCII (`é`,
U+FFFD, `，`) | CREATE takes a byte of it, which tasks get as 0xFD / 0xEF
from 0x80 up; ALTER accepts it without effect | `enclose must be an
ASCII character` |
| such a value, or an invalid one, stored by an older version | no
effect | no effect, with a warning; the replay of the job and of the
ALTER goes on |
branch-4.1 and branch-4.2 have the same gaps.
---
.../fileformat/CsvFileFormatProperties.java | 33 ++-
.../doris/load/routineload/RoutineLoadJob.java | 57 ++++-
.../doris/nereids/load/NereidsStreamLoadTask.java | 2 +
.../plans/commands/AlterRoutineLoadCommand.java | 13 +-
.../plans/commands/info/CreateRoutineLoadInfo.java | 24 ++
.../load/routineload/RoutineLoadJobReplayTest.java | 247 +++++++++++++++++++++
.../nereids/load/NereidsStreamLoadTaskTest.java | 15 ++
.../commands/AlterRoutineLoadCommandTest.java | 44 ++++
.../routine_load/test_routine_load_property.groovy | 18 ++
9 files changed, 438 insertions(+), 15 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java
index 57ac88e6b2a..ba04368c3d0 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/fileformat/CsvFileFormatProperties.java
@@ -110,20 +110,13 @@ public class CsvFileFormatProperties extends
FileFormatProperties {
String enclosedString = getOrDefault(formatProperties,
PROP_ENCLOSE,
"", isRemoveOriginProperty);
if (!Strings.isNullOrEmpty(enclosedString)) {
- if (enclosedString.length() > 1) {
- throw new AnalysisException("enclose should not be longer
than one byte.");
- }
- enclose = (byte) enclosedString.charAt(0);
+ enclose = parseEnclose(enclosedString);
}
String escapeStr = getOrDefault(formatProperties, PROP_ESCAPE,
"", isRemoveOriginProperty);
if (!Strings.isNullOrEmpty(escapeStr)) {
- if (escapeStr.length() != 1) {
- throw new AnalysisException("escape must be single-char");
- } else {
- escape = escapeStr.getBytes()[0];
- }
+ escape = parseEscape(escapeStr);
}
trimDoubleQuotes = Boolean.valueOf(getOrDefault(formatProperties,
@@ -160,6 +153,28 @@ public class CsvFileFormatProperties extends
FileFormatProperties {
}
}
+ // The byte an enclose property stands for, 0 for an empty one.
+ public static byte parseEnclose(String enclose) {
+ if (Strings.isNullOrEmpty(enclose)) {
+ return 0;
+ }
+ if (enclose.length() > 1) {
+ throw new AnalysisException("enclose should not be longer than one
byte.");
+ }
+ return (byte) enclose.charAt(0);
+ }
+
+ // The byte an escape property stands for, 0 for an empty one.
+ public static byte parseEscape(String escape) {
+ if (Strings.isNullOrEmpty(escape)) {
+ return 0;
+ }
+ if (escape.length() != 1) {
+ throw new AnalysisException("escape must be single-char");
+ }
+ return escape.getBytes()[0];
+ }
+
public void checkSupportedCompressionType(boolean isWrite) {
// Currently, only check for write operation.
// Because we only support a subset of compression type for writing.
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 68f1c320d58..d861bd6a4c8 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
@@ -335,6 +335,8 @@ public abstract class RoutineLoadJob
} else {
sessionVariables.put(SessionVariable.SQL_MODE,
String.valueOf(SqlModeHelper.MODE_DEFAULT));
}
+ // a snapshot of the creating session, which gsonPostProcess has no
other way to restore
+ sessionVariables.put(SessionVariable.ENABLE_MEMTABLE_ON_SINK_NODE,
String.valueOf(memtableOnSinkNode));
}
/**
@@ -362,6 +364,8 @@ public abstract class RoutineLoadJob
} else {
sessionVariables.put(SessionVariable.SQL_MODE,
String.valueOf(SqlModeHelper.MODE_DEFAULT));
}
+ // a snapshot of the creating session, which gsonPostProcess has no
other way to restore
+ sessionVariables.put(SessionVariable.ENABLE_MEMTABLE_ON_SINK_NODE,
String.valueOf(memtableOnSinkNode));
}
protected void setOptional(CreateRoutineLoadInfo info) throws
UserException {
@@ -2064,7 +2068,10 @@ public abstract class RoutineLoadJob
uniqueKeyUpdateMode = TUniqueKeyUpdateMode.UPSERT;
}
}
- // Process remaining properties
+ // Process remaining properties. max_filter_ratio,
send_batch_parallelism, load_to_single_tablet, enclose and
+ // escape are fields that are not persisted: CREATE records the value
each one was given and ALTER the new
+ // ones, so without this a follower, a restarted FE or a new master
would run the job with defaults. A job
+ // created before one of them existed has no key and keeps the default.
jobProperties.forEach((k, v) -> {
if (k.equals(CreateRoutineLoadInfo.PARTIAL_COLUMNS)) {
// Backward compatibility: only use partial_columns if
unique_key_update_mode is not set
@@ -2081,8 +2088,24 @@ public abstract class RoutineLoadJob
} else {
partialUpdateNewKeyPolicy =
TPartialUpdateNewRowPolicy.APPEND;
}
+ } else if
(k.equals(CreateRoutineLoadInfo.MAX_FILTER_RATIO_PROPERTY)) {
+ maxFilterRatio = Double.parseDouble(v);
+ } else if (k.equals(CreateRoutineLoadInfo.SEND_BATCH_PARALLELISM))
{
+ sendBatchParallelism = Integer.parseInt(v);
+ } else if (k.equals(CreateRoutineLoadInfo.LOAD_TO_SINGLE_TABLET)) {
+ loadToSingleTablet = Boolean.parseBoolean(v);
+ } else if (k.equals(CsvFileFormatProperties.PROP_ENCLOSE)) {
+ enclose = csvControlByte(k, v, enclose);
+ } else if (k.equals(CsvFileFormatProperties.PROP_ESCAPE)) {
+ escape = csvControlByte(k, v, escape);
}
});
+ // Snapshotted from the creating session. A job created before the
snapshot was persisted keeps false, which
+ // is what it has run with after every restart.
+ String memtableOnSinkNodeValue =
sessionVariables.get(SessionVariable.ENABLE_MEMTABLE_ON_SINK_NODE);
+ if (memtableOnSinkNodeValue != null) {
+ memtableOnSinkNode = Boolean.parseBoolean(memtableOnSinkNodeValue);
+ }
try {
ConnectContext ctx = new ConnectContext();
ctx.setDatabase(Env.getCurrentEnv().getInternalCatalog().getDb(dbId)
@@ -2099,7 +2122,8 @@ public abstract class RoutineLoadJob
CreateRoutineLoadCommand command = (CreateRoutineLoadCommand)
nereidsParser.parseSingle(
origStmt.originStmt);
CreateRoutineLoadInfo createRoutineLoadInfo =
command.getCreateRoutineLoadInfo();
- // This re-parse only rebuilds the RoutineLoadDesc; it must
not re-check resources
+ // This re-parse only rebuilds the RoutineLoadDesc and
exec_mem_limit, which no ALTER
+ // changes and jobProperties does not hold; it must not
re-check resources
// that can legitimately have disappeared since the job was
created, because the
// catch below turns any failure into the final CANCELLED
state.
createRoutineLoadInfo.setReplay(true);
@@ -2118,6 +2142,7 @@ public abstract class RoutineLoadJob
}
createRoutineLoadInfo.validate(ctx);
setRoutineLoadDesc(createRoutineLoadInfo.getRoutineLoadDesc());
+ execMemLimit = createRoutineLoadInfo.getExecMemLimit();
} finally {
ctx.cleanup();
}
@@ -2205,6 +2230,34 @@ public abstract class RoutineLoadJob
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. The
metadata can hold a value that CREATE
+ // and ALTER reject now: ALTER of an older version did not check it, and
CREATE of an older version took any
+ // character and recorded its byte, which reads back as U+FFFD from 0x80
up. Such a value is ignored the same way
+ // when the job is rebuilt and when an ALTER is replayed: the job keeps
the byte it has, where an exception would
+ // fail the replay of the metadata. A job rebuilt with such a value keeps
none, as after every restart so far.
+ private byte csvControlByte(String key, String value, byte current) {
+ try {
+ CreateRoutineLoadInfo.checkCsvControlValue(key, value);
+ return key.equals(CsvFileFormatProperties.PROP_ENCLOSE)
+ ? CsvFileFormatProperties.parseEnclose(value) :
CsvFileFormatProperties.parseEscape(value);
+ } catch (AnalysisException |
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;
+ }
}
/**
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
index 95e31360ced..d3a7ba9ef75 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsStreamLoadTask.java
@@ -382,6 +382,8 @@ public class NereidsStreamLoadTask implements
NereidsLoadTaskInfo {
this.strictMode = task.isStrictMode();
this.timezone = task.getTimezone();
this.formatType = task.getFormatType();
+ this.enclose = task.getEnclose();
+ this.escape = task.getEscape();
this.stripOuterArray = task.isStripOuterArray();
this.numAsString = task.isNumAsString();
this.jsonPaths = task.getJsonPaths();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
index 3473a5e3fd3..4e52c730cab 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java
@@ -312,14 +312,19 @@ public class AlterRoutineLoadCommand extends AlterCommand
{
}
}
+ // Checked as CREATE checks them, since the job takes its byte from
the value the same way
if (jobProperties.containsKey(CsvFileFormatProperties.PROP_ENCLOSE)) {
- analyzedJobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE,
- jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE));
+ String enclose =
jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE);
+
CreateRoutineLoadInfo.checkCsvControlValue(CsvFileFormatProperties.PROP_ENCLOSE,
enclose);
+ CsvFileFormatProperties.parseEnclose(enclose);
+ analyzedJobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE,
enclose);
}
if (jobProperties.containsKey(CsvFileFormatProperties.PROP_ESCAPE)) {
- analyzedJobProperties.put(CsvFileFormatProperties.PROP_ESCAPE,
- jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE));
+ String escape =
jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE);
+
CreateRoutineLoadInfo.checkCsvControlValue(CsvFileFormatProperties.PROP_ESCAPE,
escape);
+ CsvFileFormatProperties.parseEscape(escape);
+ analyzedJobProperties.put(CsvFileFormatProperties.PROP_ESCAPE,
escape);
}
if
(jobProperties.containsKey(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL)) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
index 94ed9345356..b528912cc76 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateRoutineLoadInfo.java
@@ -57,6 +57,7 @@ import org.apache.doris.resource.workloadgroup.WorkloadGroup;
import org.apache.doris.thrift.TPartialUpdateNewRowPolicy;
import org.apache.doris.thrift.TUniqueKeyUpdateMode;
+import com.google.common.base.CharMatcher;
import com.google.common.base.Strings;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
@@ -284,6 +285,19 @@ public class CreateRoutineLoadInfo {
return mode;
}
+ /**
+ * Rejects an enclose or escape that is not an ASCII character. Tasks are
planned from the job's byte turned into
+ * text and parsed again (NereidsDataDescription), which keeps an ASCII
byte but turns every byte from 0x80 up into
+ * 0xFD as enclose and 0xEF as escape, and in UTF-8 data such a byte is
absent or part of a character anyway.
+ * CREATE and ALTER check with this; a job rebuilt from metadata ignores
such a value that an older version took
+ * (RoutineLoadJob#csvControlByte).
+ */
+ public static void checkCsvControlValue(String key, String value) throws
AnalysisException {
+ if (value != null && !CharMatcher.ascii().matchesAllOf(value)) {
+ throw new AnalysisException(key + " must be an ASCII character");
+ }
+ }
+
/**
* Checks a routine load job's workload group the way its tasks resolve it
({@code KafkaTaskInfo#rePlan}):
* as {@code user}, and in cloud mode in the namespace of {@code
computeGroup}, the group the job runs in. A
@@ -707,6 +721,16 @@ public class CreateRoutineLoadInfo {
}
}
+ // Like the groups above, checked only when a statement is admitted,
see isReplay: an older version took any
+ // character here, and failing on the metadata load path would cancel
the job permanently. The rebuilt job
+ // takes enclose and escape from jobProperties, ignoring a value there
that this rejects
+ // (RoutineLoadJob#csvControlByte).
+ if (!isReplay) {
+ checkCsvControlValue(CsvFileFormatProperties.PROP_ENCLOSE,
+ jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE));
+ checkCsvControlValue(CsvFileFormatProperties.PROP_ESCAPE,
+ jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE));
+ }
String format =
jobProperties.getOrDefault(FileFormatProperties.PROP_FORMAT, "csv");
fileFormatProperties =
FileFormatProperties.createFileFormatProperties(format);
fileFormatProperties.analyzeFileFormatProperties(jobProperties, false);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobReplayTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobReplayTest.java
new file mode 100644
index 00000000000..b70acbf4602
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobReplayTest.java
@@ -0,0 +1,247 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.load.routineload;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
+import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
+import org.apache.doris.nereids.load.NereidsRoutineLoadTaskInfo;
+import org.apache.doris.nereids.parser.NereidsParser;
+import
org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
+import
org.apache.doris.nereids.trees.plans.commands.load.CreateRoutineLoadCommand;
+import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
+import org.apache.doris.persist.gson.GsonUtils;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.OriginStatement;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.utframe.TestWithFeService;
+
+import com.google.common.collect.Maps;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+/**
+ * A follower replaying the edit log, a restarted FE and a new master rebuild
a routine load job from its metadata:
+ * a GSON round trip and gsonPostProcess, on a thread without a session. The
rebuilt job has to run as the job did
+ * before, including what the tasks are built from.
+ */
+public class RoutineLoadJobReplayTest extends TestWithFeService {
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabaseAndUse("rl_replay_db");
+ // load_to_single_tablet needs a randomly distributed table
+ createTable("CREATE TABLE rl_replay_db.dup_tbl (k1 INT, v1
VARCHAR(32)) DUPLICATE KEY(k1) "
+ + "DISTRIBUTED BY RANDOM BUCKETS 1 PROPERTIES
('replication_num' = '1')");
+ }
+
+ // What RoutineLoadManager#createRoutineLoadJob does, except registering
the job.
+ private KafkaRoutineLoadJob createJob(String sql) throws Exception {
+ CreateRoutineLoadCommand command = (CreateRoutineLoadCommand) new
NereidsParser().parseSingle(sql);
+ CreateRoutineLoadInfo info = command.getCreateRoutineLoadInfo();
+ info.validate(connectContext);
+ KafkaRoutineLoadJob job = KafkaRoutineLoadJob.fromCreateInfo(info,
connectContext);
+ job.setOrigStmt(new OriginStatement(sql, 0));
+ return job;
+ }
+
+ private KafkaRoutineLoadJob replay(RoutineLoadJob job) {
+ String json = GsonUtils.GSON.toJson(job);
+ ConnectContext.remove();
+ try {
+ return (KafkaRoutineLoadJob) GsonUtils.GSON.fromJson(json,
RoutineLoadJob.class);
+ } finally {
+ connectContext.setThreadLocalInfo();
+ }
+ }
+
+ @Test
+ public void testReplayKeepsTheDeclaredProperties() throws Exception {
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_declared ON dup_tbl\n"
+ + "PROPERTIES (\n"
+ + "'max_filter_ratio' = '0.5',\n"
+ + "'exec_mem_limit' = '4294967296',\n"
+ + "'send_batch_parallelism' = '3',\n"
+ + "'load_to_single_tablet' = 'true',\n"
+ + "'enclose' = '\"',\n"
+ + "'escape' = '\\\\'\n"
+ + ")\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_declared')");
+
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals(RoutineLoadJob.JobState.NEED_SCHEDULE,
replayed.getState());
+ Assertions.assertEquals(0.5, replayed.getMaxFilterRatio());
+ Assertions.assertEquals(4294967296L, replayed.getMemLimit());
+ Assertions.assertEquals(3, replayed.getSendBatchParallelism());
+ Assertions.assertTrue(replayed.isLoadToSingleTablet());
+ NereidsRoutineLoadTaskInfo taskInfo =
replayed.toNereidsRoutineLoadTaskInfo();
+ Assertions.assertEquals('"', taskInfo.getEnclose());
+ Assertions.assertEquals('\\', taskInfo.getEscape());
+ Assertions.assertEquals(0.5, taskInfo.getMaxFilterRatio());
+ Assertions.assertEquals(4294967296L, taskInfo.getMemLimit());
+ Assertions.assertEquals(3, taskInfo.getSendBatchParallelism());
+ Assertions.assertTrue(taskInfo.isLoadToSingleTablet());
+ }
+
+ // What CREATE takes from the creating session when the statement leaves
it out.
+ @Test
+ public void testReplayKeepsWhatTheCreatingSessionGave() throws Exception {
+ SessionVariable sessionVariable = connectContext.getSessionVariable();
+ int sendBatchParallelism = sessionVariable.sendBatchParallelism;
+ boolean memtableOnSinkNode = sessionVariable.enableMemtableOnSinkNode;
+ KafkaRoutineLoadJob withDefaults;
+ KafkaRoutineLoadJob withSessionValues;
+ try {
+ withDefaults = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_session_default ON dup_tbl\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_session')");
+ sessionVariable.sendBatchParallelism = 4;
+ sessionVariable.enableMemtableOnSinkNode = false;
+ withSessionValues = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_session_set ON dup_tbl\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_session')");
+ } finally {
+ sessionVariable.sendBatchParallelism = sendBatchParallelism;
+ sessionVariable.enableMemtableOnSinkNode = memtableOnSinkNode;
+ }
+
+ // enable_memtable_on_sink_node is on by default
+ Assertions.assertTrue(replay(withDefaults).isMemtableOnSinkNode());
+ KafkaRoutineLoadJob replayed = replay(withSessionValues);
+ Assertions.assertEquals(4, replayed.getSendBatchParallelism());
+ Assertions.assertFalse(replayed.isMemtableOnSinkNode());
+ }
+
+ // Without enclose and escape CREATE records the zero byte, and the replay
gives the job none.
+ @Test
+ public void testCreateWithoutEncloseAndEscapeRecordsTheZeroByte() throws
Exception {
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_no_enclose ON dup_tbl\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_no_enclose')");
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ Assertions.assertEquals("\u0000",
jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE));
+ Assertions.assertEquals("\u0000",
jobProperties.get(CsvFileFormatProperties.PROP_ESCAPE));
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals(0, replayed.getEnclose());
+ Assertions.assertEquals(0, replayed.getEscape());
+ }
+
+ // CREATE of an older version took any character as enclose or escape, and
recorded the byte it took, which reads
+ // back as U+FFFD from 0x80 up. The rebuild re-parses the stored
statement, which can hold such a character (U+FFFD
+ // where the client did not send UTF-8), and must not reject it as CREATE
does now, since that would cancel the
+ // job. The job keeps no enclose and no escape for the U+FFFD recorded,
which is what it has run with after every
+ // restart.
+ @Test
+ public void testOlderCreateStatementWithNonAsciiValueReplays() throws
Exception {
+ for (String value : new String[] {String.valueOf((char) 0xFFFD), "é"})
{
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_older_create ON dup_tbl\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_older_create')");
+ // What an older version admitted and recorded
+ job.setOrigStmt(new OriginStatement("CREATE ROUTINE LOAD
rl_replay_db.job_older_create ON dup_tbl\n"
+ + "PROPERTIES ('enclose' = '" + value + "', 'escape' = '"
+ value + "')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_older_create')", 0));
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ jobProperties.put(CsvFileFormatProperties.PROP_ENCLOSE,
String.valueOf((char) 0xFFFD));
+ jobProperties.put(CsvFileFormatProperties.PROP_ESCAPE,
String.valueOf((char) 0xFFFD));
+
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals(RoutineLoadJob.JobState.NEED_SCHEDULE,
replayed.getState(), value);
+ Assertions.assertEquals(0, replayed.getEnclose(), value);
+ Assertions.assertEquals(0, replayed.getEscape(), value);
+ }
+ }
+
+ // CREATE takes only an ASCII character as enclose or escape: the tasks
would get 0xFD or 0xEF for any byte from
+ // 0x80 up, which in UTF-8 data is absent or part of a character anyway.
U+FFFD is what a statement whose bytes
+ // were not UTF-8 decodes to, and the full-width comma was taken as its
low byte, 0x0C, as enclose.
+ @Test
+ public void testCreateRejectsNonAsciiEncloseAndEscape() {
+ for (String key : new String[] {CsvFileFormatProperties.PROP_ENCLOSE,
CsvFileFormatProperties.PROP_ESCAPE}) {
+ for (String value : new String[] {"é", String.valueOf((char)
0xFFFD), "，"}) {
+ String sql = "CREATE ROUTINE LOAD rl_replay_db.job_non_ascii
ON dup_tbl\n"
+ + "PROPERTIES ('" + key + "' = '" + value + "')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_non_ascii')";
+ Exception e = Assertions.assertThrows(Exception.class, () ->
createJob(sql));
+ Assertions.assertTrue(e.getMessage().contains(key + " must be
an ASCII character"), e.getMessage());
+ }
+ }
+ }
+
+ // ALTER ROUTINE LOAD installs the analyzed properties through
modifyPropertiesInternal, as the replay of its
+ // edit log does.
+ @Test
+ public void testAlteredEncloseAndEscapeReachTheTasks() throws Exception {
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_altered ON dup_tbl\n"
+ + "PROPERTIES ('enclose' = '\"', 'escape' = '\\\\')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_altered')");
+ Map<String, String> altered = Maps.newHashMap();
+ altered.put(CsvFileFormatProperties.PROP_ENCLOSE, "'");
+ altered.put(CsvFileFormatProperties.PROP_ESCAPE, "|");
+ job.replayModifyProperties(new
AlterRoutineLoadJobOperationLog(job.getId(), altered, null));
+
+ NereidsRoutineLoadTaskInfo taskInfo =
job.toNereidsRoutineLoadTaskInfo();
+ Assertions.assertEquals('\'', taskInfo.getEnclose());
+ Assertions.assertEquals('|', taskInfo.getEscape());
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals('\'', replayed.getEnclose());
+ Assertions.assertEquals('|', replayed.getEscape());
+ }
+
+ // ALTER of an older version stored enclose and escape without checking
them. Such a value never reached the
+ // job, and must neither now nor fail the replay of the job or of the
ALTER.
+ @Test
+ public void testInvalidValueAnOlderAlterStoredIsIgnored() throws Exception
{
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_invalid ON dup_tbl\n"
+ + "PROPERTIES ('enclose' = '\"', 'escape' = '\\\\')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_invalid')");
+ Map<String, String> altered = Maps.newHashMap();
+ altered.put(CsvFileFormatProperties.PROP_ENCLOSE, "ab");
+ job.replayModifyProperties(new
AlterRoutineLoadJobOperationLog(job.getId(), altered, null));
+ Assertions.assertEquals('"', job.getEnclose());
+
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ Assertions.assertEquals("ab",
jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE));
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals(RoutineLoadJob.JobState.NEED_SCHEDULE,
replayed.getState());
+ Assertions.assertEquals(0, replayed.getEnclose());
+ Assertions.assertEquals('\\', replayed.getEscape());
+ }
+
+ // ALTER of an older version also stored a value that is not ASCII.
Neither the replay of that ALTER nor a rebuild
+ // of the job, whose metadata then holds the value, applies it: the job
keeps the byte it has.
+ @Test
+ public void testNonAsciiValueAnOlderAlterStoredIsIgnored() throws
Exception {
+ for (String value : new String[] {"é", String.valueOf((char) 0xFFFD)})
{
+ KafkaRoutineLoadJob job = createJob("CREATE ROUTINE LOAD
rl_replay_db.job_older_alter ON dup_tbl\n"
+ + "PROPERTIES ('enclose' = '\"', 'escape' = '\\\\')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_older_alter')");
+ Map<String, String> altered = Maps.newHashMap();
+ altered.put(CsvFileFormatProperties.PROP_ENCLOSE, value);
+ altered.put(CsvFileFormatProperties.PROP_ESCAPE, value);
+ job.replayModifyProperties(new
AlterRoutineLoadJobOperationLog(job.getId(), altered, null));
+ Assertions.assertEquals('"', job.getEnclose(), value);
+ Assertions.assertEquals('\\', job.getEscape(), value);
+
+ Map<String, String> jobProperties = Deencapsulation.getField(job,
"jobProperties");
+ Assertions.assertEquals(value,
jobProperties.get(CsvFileFormatProperties.PROP_ENCLOSE));
+ KafkaRoutineLoadJob replayed = replay(job);
+ Assertions.assertEquals(RoutineLoadJob.JobState.NEED_SCHEDULE,
replayed.getState(), value);
+ Assertions.assertEquals(0, replayed.getEnclose(), value);
+ Assertions.assertEquals(0, replayed.getEscape(), value);
+ }
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
index 403766a6769..d9027f597a6 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/load/NereidsStreamLoadTaskTest.java
@@ -59,6 +59,21 @@ public class NereidsStreamLoadTaskTest {
Assertions.assertTrue(streamLoadTask.isNumAsString());
}
+ @Test
+ public void testMultiTableBaseTaskCopiesCsvProperties() throws Exception {
+ // The per-table request of a multi-table routine load carries no
enclose and no escape
+ NereidsStreamLoadTask streamLoadTask =
NereidsStreamLoadTask.fromTStreamLoadPutRequest(newRequest());
+ LoadTaskInfo routineLoadTask = Mockito.mock(LoadTaskInfo.class);
+
Mockito.when(routineLoadTask.getFormatType()).thenReturn(TFileFormatType.FORMAT_CSV_PLAIN);
+ Mockito.when(routineLoadTask.getEnclose()).thenReturn((byte) '"');
+ Mockito.when(routineLoadTask.getEscape()).thenReturn((byte) '\\');
+
+ streamLoadTask.setMultiTableBaseTaskInfo(routineLoadTask);
+
+ Assertions.assertEquals('"', streamLoadTask.getEnclose());
+ Assertions.assertEquals('\\', streamLoadTask.getEscape());
+ }
+
@Test
public void testDefaultSendBatchParallelism() throws UserException {
Assertions.assertEquals(1,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommandTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommandTest.java
index 0989bc127b0..d6bace72b99 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommandTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommandTest.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.Table;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
import org.apache.doris.load.loadv2.LoadTask;
import org.apache.doris.load.routineload.RoutineLoadJob;
import org.apache.doris.load.routineload.RoutineLoadManager;
@@ -115,4 +116,47 @@ public class AlterRoutineLoadCommandTest {
Assertions.assertTrue(command.getAnalyzedJobProperties().containsKey(CreateRoutineLoadInfo.STRICT_MODE));
Assertions.assertTrue(command.getAnalyzedJobProperties().containsKey(CreateRoutineLoadInfo.TIMEZONE));
}
+
+ // The job takes the byte from enclose and escape the way CREATE does, so
ALTER rejects what CREATE rejects.
+ @Test
+ public void testValidateChecksEncloseAndEscape() {
+ runBefore();
+ LabelNameInfo labelNameInfo = new LabelNameInfo("testDb", "label1");
+ Map<String, String> enclose = Maps.newHashMap();
+ enclose.put(CsvFileFormatProperties.PROP_ENCLOSE, "ab");
+ AlterRoutineLoadCommand badEnclose = new
AlterRoutineLoadCommand(labelNameInfo, enclose, Maps.newHashMap());
+ Exception e = Assertions.assertThrows(Exception.class, () ->
badEnclose.validate(connectContext));
+ Assertions.assertTrue(e.getMessage().contains("enclose should not be
longer than one byte"), e.getMessage());
+
+ Map<String, String> escape = Maps.newHashMap();
+ escape.put(CsvFileFormatProperties.PROP_ESCAPE, "ab");
+ AlterRoutineLoadCommand badEscape = new
AlterRoutineLoadCommand(labelNameInfo, escape, Maps.newHashMap());
+ e = Assertions.assertThrows(Exception.class, () ->
badEscape.validate(connectContext));
+ Assertions.assertTrue(e.getMessage().contains("escape must be
single-char"), e.getMessage());
+
+ Map<String, String> valid = Maps.newHashMap();
+ valid.put(CsvFileFormatProperties.PROP_ENCLOSE, "'");
+ valid.put(CsvFileFormatProperties.PROP_ESCAPE, "|");
+ AlterRoutineLoadCommand command = new
AlterRoutineLoadCommand(labelNameInfo, valid, Maps.newHashMap());
+ Assertions.assertDoesNotThrow(() -> command.validate(connectContext));
+ Assertions.assertEquals("'",
command.getAnalyzedJobProperties().get(CsvFileFormatProperties.PROP_ENCLOSE));
+ Assertions.assertEquals("|",
command.getAnalyzedJobProperties().get(CsvFileFormatProperties.PROP_ESCAPE));
+ }
+
+ // ALTER, like CREATE, takes only an ASCII character as enclose or escape.
+ @Test
+ public void testValidateRejectsNonAsciiEncloseAndEscape() {
+ runBefore();
+ LabelNameInfo labelNameInfo = new LabelNameInfo("testDb", "label1");
+ for (String key : new String[] {CsvFileFormatProperties.PROP_ENCLOSE,
CsvFileFormatProperties.PROP_ESCAPE}) {
+ for (String value : new String[] {"é", String.valueOf((char)
0xFFFD), "，"}) {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(key, value);
+ AlterRoutineLoadCommand command = new
AlterRoutineLoadCommand(labelNameInfo, properties,
+ Maps.newHashMap());
+ Exception e = Assertions.assertThrows(Exception.class, () ->
command.validate(connectContext));
+ Assertions.assertTrue(e.getMessage().contains(key + " must be
an ASCII character"), e.getMessage());
+ }
+ }
+ }
}
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 cd10af8dfe5..ea518c4a3d9 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
@@ -93,6 +93,24 @@ suite("test_routine_load_property","p0") {
PROPERTIES ("replication_allocation" = "tag.location.default: 1");
"""
+ // CREATE takes only an ASCII character as enclose or escape: the tasks
would get 0xFD or 0xEF for any byte from
+ // 0x80 up, which in UTF-8 data is absent or part of a character anyway.
U+FFFD, the Unicode replacement
+ // character, is what a statement whose bytes were not UTF-8 decodes to.
CREATE checks them before it connects to
+ // Kafka.
+ for (String key : ["enclose", "escape"]) {
+ for (String value : ["\uFFFD", "é"]) {
+ test {
+ sql """
+ CREATE ROUTINE LOAD test_${key}_non_ascii ON ${tableName}
+ COLUMNS TERMINATED BY ","
+ PROPERTIES ("${key}" = "${value}")
+ FROM KAFKA ("kafka_broker_list" = "127.0.0.1:9092",
"kafka_topic" = "test_${key}_non_ascii")
+ """
+ exception "${key} must be an ASCII character"
+ }
+ }
+ }
+
if (enabled != null && enabled.equalsIgnoreCase("true")) {
def jobName = "test_enclose_and_escape"
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
