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 3cf8cc628fb [fix](fe) Make SHOW CREATE ROUTINE LOAD output recreate
the same job (#70118)
3cf8cc628fb is described below
commit 3cf8cc628fbe2c1427ddaa94fcf29e8f40433172
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Nov 5 07:18:24 2026 +0000
[fix](fe) Make SHOW CREATE ROUTINE LOAD output recreate the same job
(#70118)
### What problem does this PR solve?
Issue Number: None
Related PR: #69749
Problem Summary:
**In short.** `SHOW CREATE ROUTINE LOAD` prints the statement a job is
recreated from, but a job recreated from it was not the same job: eight
properties that CREATE accepts were never printed, a value with a quote
or a backslash made the statement unparseable, `TEMPORARY` partitions
turned into formal ones, and the comment was dropped. This PR makes the
statement carry all of them.
**Background.** Users recreate a routine load job from `SHOW CREATE
ROUTINE LOAD` to rename it, to move it to another database, or to start
it over after `STOP`; the statement prints the next Kafka offsets to
consume for exactly that purpose. #69749 made the statement carry the
`compute_group` a job declares, and found the rest of the list
incomplete: `RoutineLoadJob.getShowCreateInfo()` writes the properties
by hand and covered 17 of the 25 keys in
`CreateRoutineLoadInfo.PROPERTIES_SET`.
**The problem, and what it cost.** Measured on master `1aa505cb0be` with
a unit test that creates jobs from SQL, recreates each one from its
`SHOW CREATE` output and compares every property:
| Job | Recreated from its SHOW CREATE output |
|---|---|
| CSV job with `enclose` `"`, `escape` `\`, `empty_field_as_null`,
`send_batch_parallelism` = 3, `load_to_single_tablet`, `workload_group`
| all six back to their defaults; quoted fields get split at the
separators inside the quotes |
| `unique_key_update_mode` = `UPDATE_FLEXIBLE_COLUMNS` with
`partial_update_new_key_behavior` = `ERROR` | an `UPSERT` job with
`APPEND` |
| partial columns with `partial_update_new_key_behavior` = `ERROR` |
`APPEND`: rows with new keys get inserted instead of failing the load |
| JSON job with `jsonpaths` such as `["$.k1","$.v1"]` | the statement
does not parse (`mismatched input '$'`) |
| Kafka property whose value holds a `"`, such as a JAAS config | the
statement does not parse |
| `TEMPORARY PARTITION(tp1)` | printed as `PARTITION(tp1)`: `Partition
tp1 does not exist` |
| `COMMENT '...'` | gone |
The jobs themselves keep running as declared; only the statement is
wrong. The cost is a recreated job that silently loads differently (CSV
parsing, partial update semantics, single-tablet loading, workload
group, parallelism), or a statement that cannot be executed at all.
**How this PR fixes it.**
- `getShowCreateInfo()` prints the eight missing properties wherever
leaving one out would give the recreated job a different value:
`unique_key_update_mode` when it is not `UPSERT` (`partial_columns`
alone cannot express `UPDATE_FLEXIBLE_COLUMNS`),
`partial_update_new_key_behavior` when it is `ERROR` and a partial
update mode is on (CREATE rejects the policy otherwise), `enclose` and
`escape` when set, `empty_field_as_null` and `load_to_single_tablet`
when true, and `workload_group` when set. `send_batch_parallelism` is
always printed, as the timezone is, because CREATE otherwise takes it
from the session variable of whoever recreates the job. The values are
the ones the job runs with.
- Keys and values are quoted with `SqlLiteralUtils.quoteStringLiteral`,
which follows the session's `NO_BACKSLASH_ESCAPES` mode. A value without
a quote or a backslash prints exactly as before; the column separator is
left as it is, since it is kept as typed and already reads back.
- `TEMPORARY PARTITION(...)` keeps `TEMPORARY`, and the job's comment is
printed as `COMMENT "..."`.
- `CREATE ROUTINE LOAD` reads its `COMMENT` with
`SqlLiteralUtils.parseStringLiteral`, as table, view and column comments
are read, so that a printed comment with a quote reads back unchanged.
It used to strip the outer quotes only, so `COMMENT 'it''s'` was stored
as `it''s`.
For a job that uses none of these, the output gains a single line,
`"send_batch_parallelism" = "<the job's value>"`.
**Results.**
| Job | Before | After |
|---|---|---|
| CSV job above | 6 properties and the comment lost | recreated with the
same definition |
| flexible partial update with `ERROR` | `UPSERT` / `APPEND` | same
definition |
| JSON job with `jsonpaths` | statement does not parse | same definition
|
| Kafka property with `"` and `\` | statement does not parse | same
definition |
| temporary partition | `Partition tp1 does not exist` | same definition
|
| `NO_BACKSLASH_ESCAPES` session, `enclose` `"`, `escape` `\` | both
lost | same definition |
The same test also found that a job's `max_filter_ratio`,
`exec_mem_limit`, `send_batch_parallelism`, `load_to_single_tablet`,
`enclose` and `escape` fall back to their defaults when a follower
replays the job or the FE restarts, and that `ALTER ROUTINE LOAD` on
`enclose` / `escape` does not reach the running job. That is a separate
defect and will be fixed in its own PR; since this PR prints the values
the job runs with, the statement follows that fix. branch-4.1 and
branch-4.2 have the same gaps as master.
---
.../doris/load/routineload/RoutineLoadJob.java | 42 +++-
.../doris/nereids/parser/LogicalPlanBuilder.java | 5 +-
.../doris/load/routineload/RoutineLoadJobTest.java | 53 ++++
.../RoutineLoadShowCreateRoundTripTest.java | 268 +++++++++++++++++++++
4 files changed, 363 insertions(+), 5 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 6c644eb4792..68f1c320d58 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
@@ -58,6 +58,7 @@ import
org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand;
import org.apache.doris.nereids.trees.plans.commands.LoadCommand;
import
org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
import
org.apache.doris.nereids.trees.plans.commands.load.CreateRoutineLoadCommand;
+import org.apache.doris.nereids.util.SqlLiteralUtils;
import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
import org.apache.doris.persist.RoutineLoadOperation;
import org.apache.doris.persist.gson.GsonPostProcessable;
@@ -1824,7 +1825,8 @@ public abstract class RoutineLoadJob
}
// 4.4.partitions
if (partitionNamesInfo != null) {
-
sb.append("PARTITION(").append(Joiner.on(",").join(partitionNamesInfo.getPartitionNames())).append("),\n");
+ sb.append(partitionNamesInfo.isTemp() ? "TEMPORARY PARTITION(" :
"PARTITION(")
+
.append(Joiner.on(",").join(partitionNamesInfo.getPartitionNames())).append("),\n");
}
// 4.5.delete_on_predicates
if (deleteCondition != null) {
@@ -1844,7 +1846,7 @@ public abstract class RoutineLoadJob
if (sb.charAt(sb.length() - 2) == ',') {
sb.replace(sb.length() - 2, sb.length() - 1, "");
}
- // 5.job_properties. See PROPERTIES_SET of CreateRoutineLoadStmt
+ // 5.job_properties. See PROPERTIES_SET of CreateRoutineLoadInfo
sb.append("PROPERTIES\n(\n");
appendProperties(sb,
CreateRoutineLoadInfo.DESIRED_CONCURRENT_NUMBER_PROPERTY,
desireTaskConcurrentNum, false);
appendProperties(sb, CreateRoutineLoadInfo.MAX_ERROR_NUMBER_PROPERTY,
maxErrorNum, false);
@@ -1856,13 +1858,37 @@ public abstract class RoutineLoadJob
if (isPartialUpdate) {
appendProperties(sb, CreateRoutineLoadInfo.PARTIAL_COLUMNS,
isPartialUpdate, false);
}
+ // UPDATE_FLEXIBLE_COLUMNS leaves isPartialUpdate false, so without
the mode the job would be recreated
+ // as an UPSERT one. CREATE only accepts the new key policy together
with a partial update mode.
+ if (uniqueKeyUpdateMode != TUniqueKeyUpdateMode.UPSERT) {
+ appendProperties(sb, CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE,
uniqueKeyUpdateMode.name(), false);
+ if (partialUpdateNewKeyPolicy == TPartialUpdateNewRowPolicy.ERROR)
{
+ appendProperties(sb,
CreateRoutineLoadInfo.PARTIAL_UPDATE_NEW_KEY_POLICY, "ERROR", false);
+ }
+ }
appendProperties(sb, JsonFileFormatProperties.PROP_JSON_PATHS,
getJsonPaths(), false);
appendProperties(sb, JsonFileFormatProperties.PROP_STRIP_OUTER_ARRAY,
isStripOuterArray(), false);
appendProperties(sb, JsonFileFormatProperties.PROP_NUM_AS_STRING,
isNumAsString(), false);
appendProperties(sb, JsonFileFormatProperties.PROP_FUZZY_PARSE,
isFuzzyParse(), false);
appendProperties(sb, JsonFileFormatProperties.PROP_JSON_ROOT,
getJsonRoot(), false);
+ if (getEnclose() != 0) {
+ appendProperties(sb, CsvFileFormatProperties.PROP_ENCLOSE, new
String(new byte[]{getEnclose()}), false);
+ }
+ if (getEscape() != 0) {
+ appendProperties(sb, CsvFileFormatProperties.PROP_ESCAPE, new
String(new byte[]{getEscape()}), false);
+ }
+ if (getEmptyFieldAsNull()) {
+ appendProperties(sb,
CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL, true, false);
+ }
appendProperties(sb, LoadCommand.STRICT_MODE, isStrictMode(), false);
appendProperties(sb, LoadCommand.TIMEZONE, getTimezone(), false);
+ // Always written, as the timezone is: left out, CREATE takes it from
the session variable of whoever
+ // recreates the job.
+ appendProperties(sb, CreateRoutineLoadInfo.SEND_BATCH_PARALLELISM,
getSendBatchParallelism(), false);
+ if (isLoadToSingleTablet()) {
+ appendProperties(sb, CreateRoutineLoadInfo.LOAD_TO_SINGLE_TABLET,
true, false);
+ }
+ appendProperties(sb, CreateRoutineLoadInfo.WORKLOAD_GROUP,
getWorkloadGroup(), false);
// Only a declared group, which follows ALTER. Writing the effective
group instead would turn
// a group inherited from the creating session into a pinned one when
the statement is replayed.
appendProperties(sb, CreateRoutineLoadInfo.COMPUTE_GROUP,
getDeclaredComputeGroup(), false);
@@ -1885,7 +1911,12 @@ public abstract class RoutineLoadJob
}
// remove the last ","
sb.replace(sb.length() - 2, sb.length() - 1, "");
- sb.append(");");
+ sb.append(")");
+ // 8. comment
+ if (!Strings.isNullOrEmpty(comment)) {
+ sb.append("\nCOMMENT
").append(SqlLiteralUtils.quoteStringLiteral(comment));
+ }
+ sb.append(";");
return sb.toString();
}
@@ -1893,7 +1924,10 @@ public abstract class RoutineLoadJob
if (value == null || Strings.isNullOrEmpty(value.toString())) {
return;
}
- sb.append("\"").append(key).append("\"").append(" =
").append("\"").append(value).append("\"");
+ // Quoted the way the parser reads string literals back: a jsonpaths
value such as ["$.k1"] or an
+ // enclose of " would otherwise end the literal early, and an escape
of \ would escape the closing quote.
+ sb.append(SqlLiteralUtils.quoteStringLiteral(key)).append(" = ")
+ .append(SqlLiteralUtils.quoteStringLiteral(value.toString()));
if (!end) {
sb.append(",\n");
} else {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
index 1a3624a3527..2a11873e7cc 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java
@@ -2576,7 +2576,10 @@ public class LogicalPlanBuilder extends
DorisParserBaseVisitor<Object> {
mergeType = LoadTask.MergeType.MERGE;
}
}
- String comment = visitCommentSpec(ctx.commentSpec());
+ // Read as a string literal, as column comments are, so that the
COMMENT that SHOW CREATE ROUTINE LOAD
+ // prints reads back unchanged
+ String comment = ctx.commentSpec() == null ? ""
+ :
SqlLiteralUtils.parseStringLiteral(ctx.commentSpec().STRING_LITERAL().getText());
Map<String, LoadProperty> loadPropertyMap = new HashMap<>();
for (DorisParser.LoadPropertyContext oneLoadPropertyCOntext :
ctx.loadProperty()) {
LoadProperty loadProperty =
visitLoadProperty(oneLoadPropertyCOntext);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
index 87a9d33c27f..970b1b2e6a5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadJobTest.java
@@ -29,6 +29,7 @@ import org.apache.doris.common.UserException;
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.routineload.kafka.KafkaProgress;
import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
import org.apache.doris.load.routineload.kafka.KafkaTaskInfo;
@@ -369,6 +370,7 @@ public class RoutineLoadJobTest {
+ "\"fuzzy_parse\" = \"false\",\n"
+ "\"strict_mode\" = \"false\",\n"
+ "\"timezone\" = \"Asia/Shanghai\",\n"
+ + "\"send_batch_parallelism\" = \"1\",\n"
+ "\"exec_mem_limit\" = \"2147483648\"\n"
+ ")\n"
+ "FROM KAFKA\n"
@@ -413,6 +415,7 @@ public class RoutineLoadJobTest {
+ "\"fuzzy_parse\" = \"false\",\n"
+ "\"strict_mode\" = \"false\",\n"
+ "\"timezone\" = \"Asia/Shanghai\",\n"
+ + "\"send_batch_parallelism\" = \"1\",\n"
+ "\"compute_group\" = \"cg_b\",\n"
+ "\"exec_mem_limit\" = \"2147483648\"\n"
+ ")\n"
@@ -424,6 +427,56 @@ public class RoutineLoadJobTest {
Assertions.assertEquals(expect, routineLoadJob.getShowCreateInfo());
}
+ // The properties that change how rows are parsed and written, printed as
literals the statement parses back
+ // from. RoutineLoadShowCreateRoundTripTest recreates jobs from the
statement.
+ @Test
+ public void testGetShowCreateInfoCarriesWriteProperties() throws
UserException {
+ KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(111L,
"test_load", 1,
+ 11, "localhost:9092", "test_topic", UserIdentity.ADMIN);
+ Deencapsulation.setField(routineLoadJob, "maxErrorNum", 10);
+ Deencapsulation.setField(routineLoadJob, "maxBatchRows", 10);
+ Deencapsulation.setField(routineLoadJob, "sendBatchParallelism", 3);
+ Deencapsulation.setField(routineLoadJob, "loadToSingleTablet", true);
+ Deencapsulation.setField(routineLoadJob, "enclose", (byte) '"');
+ Deencapsulation.setField(routineLoadJob, "escape", (byte) '\\');
+ Map<String, String> jobProperties =
Deencapsulation.getField(routineLoadJob, "jobProperties");
+ jobProperties.put(CreateRoutineLoadInfo.WORKLOAD_GROUP, "wg1");
+ jobProperties.put(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL,
"true");
+ routineLoadJob.setComment("the \"test\" job");
+
+ String expect = "CREATE ROUTINE LOAD test_load ON 11\n"
+ + "WITH APPEND\n"
+ + "PROPERTIES\n"
+ + "(\n"
+ + "\"desired_concurrent_number\" = \"0\",\n"
+ + "\"max_error_number\" = \"10\",\n"
+ + "\"max_filter_ratio\" = \"1.0\",\n"
+ + "\"max_batch_interval\" = \"60\",\n"
+ + "\"max_batch_rows\" = \"10\",\n"
+ + "\"max_batch_size\" = \"1073741824\",\n"
+ + "\"format\" = \"csv\",\n"
+ + "\"strip_outer_array\" = \"false\",\n"
+ + "\"num_as_string\" = \"false\",\n"
+ + "\"fuzzy_parse\" = \"false\",\n"
+ + "\"enclose\" = \"\"\"\",\n"
+ + "\"escape\" = \"\\\\\",\n"
+ + "\"empty_field_as_null\" = \"true\",\n"
+ + "\"strict_mode\" = \"false\",\n"
+ + "\"timezone\" = \"Asia/Shanghai\",\n"
+ + "\"send_batch_parallelism\" = \"3\",\n"
+ + "\"load_to_single_tablet\" = \"true\",\n"
+ + "\"workload_group\" = \"wg1\",\n"
+ + "\"exec_mem_limit\" = \"2147483648\"\n"
+ + ")\n"
+ + "FROM KAFKA\n"
+ + "(\n"
+ + "\"kafka_broker_list\" = \"localhost:9092\",\n"
+ + "\"kafka_topic\" = \"test_topic\"\n"
+ + ")\n"
+ + "COMMENT \"the \"\"test\"\" job\";";
+ Assertions.assertEquals(expect, routineLoadJob.getShowCreateInfo());
+ }
+
@Test
public void testParseUniqueKeyUpdateMode() {
// Test valid mode strings
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadShowCreateRoundTripTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadShowCreateRoundTripTest.java
new file mode 100644
index 00000000000..da89161db68
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/routineload/RoutineLoadShowCreateRoundTripTest.java
@@ -0,0 +1,268 @@
+// 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.load.loadv2.LoadTask;
+import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
+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.qe.OriginStatement;
+import org.apache.doris.qe.SqlModeHelper;
+import org.apache.doris.thrift.TPartialUpdateNewRowPolicy;
+import org.apache.doris.thrift.TUniqueKeyUpdateMode;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * SHOW CREATE ROUTINE LOAD prints the statement a job is recreated from, to
rename it, to move it, or to start it
+ * over after STOP. Each case creates a job the way CREATE ROUTINE LOAD does,
recreates it from what SHOW CREATE
+ * prints for it, and checks that the recreated job is defined the same way.
+ */
+public class RoutineLoadShowCreateRoundTripTest extends TestWithFeService {
+
+ @Override
+ protected void runBeforeAll() throws Exception {
+ createDatabaseAndUse("rl_db");
+ // load_to_single_tablet needs a randomly distributed table
+ createTable("CREATE TABLE rl_db.dup_tbl (k1 INT, v1 VARCHAR(32))
DUPLICATE KEY(k1) "
+ + "DISTRIBUTED BY RANDOM BUCKETS 1 PROPERTIES
('replication_num' = '1')");
+ // flexible partial updates need the skip bitmap column
+ createTable("CREATE TABLE rl_db.mow_tbl (k1 INT, v1 VARCHAR(32))
UNIQUE KEY(k1) "
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1 PROPERTIES
('replication_num' = '1', "
+ + "'enable_unique_key_merge_on_write' = 'true', "
+ + "'enable_unique_key_skip_bitmap_column' = 'true')");
+ createTable("CREATE TABLE rl_db.range_tbl (k1 INT, v1 VARCHAR(32))
DUPLICATE KEY(k1) "
+ + "PARTITION BY RANGE(k1) (PARTITION p1 VALUES LESS THAN
('100')) "
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1 PROPERTIES
('replication_num' = '1')");
+ alterTableSync("ALTER TABLE rl_db.range_tbl ADD TEMPORARY PARTITION
tp1 VALUES LESS THAN ('100')");
+ }
+
+ // What RoutineLoadManager#createRoutineLoadJob does, except registering
the job, so that the statement
+ // SHOW CREATE prints can create it again under the same name.
+ 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));
+ job.setComment(info.getComment());
+ return job;
+ }
+
+ // Everything CREATE ROUTINE LOAD sets on a job, as the job uses it.
+ private static Map<String, Object> definition(KafkaRoutineLoadJob job) {
+ Map<String, Object> d = new LinkedHashMap<>();
+ d.put("desired_concurrent_number", Deencapsulation.getField(job,
"desireTaskConcurrentNum"));
+ d.put("max_error_number", Deencapsulation.getField(job,
"maxErrorNum"));
+ d.put("max_filter_ratio", job.getMaxFilterRatio());
+ d.put("max_batch_interval", job.getMaxBatchIntervalS());
+ d.put("max_batch_rows", job.getMaxBatchRows());
+ d.put("max_batch_size", job.getMaxBatchSizeBytes());
+ d.put("strict_mode", job.isStrictMode());
+ d.put("timezone", job.getTimezone());
+ d.put("exec_mem_limit", job.getMemLimit());
+ d.put("send_batch_parallelism", job.getSendBatchParallelism());
+ d.put("load_to_single_tablet", job.isLoadToSingleTablet());
+ d.put("partial_columns", job.isFixedPartialUpdate());
+ d.put("unique_key_update_mode", job.getUniqueKeyUpdateMode());
+ d.put("partial_update_new_key_behavior", Deencapsulation.getField(job,
"partialUpdateNewKeyPolicy"));
+ d.put("workload_group", job.getWorkloadGroup());
+ d.put("compute_group", job.getDeclaredComputeGroup());
+ d.put("format", job.getFormat());
+ d.put("jsonpaths", job.getJsonPaths());
+ d.put("strip_outer_array", job.isStripOuterArray());
+ d.put("num_as_string", job.isNumAsString());
+ d.put("fuzzy_parse", job.isFuzzyParse());
+ d.put("json_root", job.getJsonRoot());
+ d.put("enclose", job.getEnclose());
+ d.put("escape", job.getEscape());
+ d.put("empty_field_as_null", job.getEmptyFieldAsNull());
+ d.put("merge_type", job.getMergeType());
+ d.put("column_separator", job.getColumnSeparator() == null ? null :
job.getColumnSeparator().getSeparator());
+ d.put("columns", job.getColumnExprDescs().descs.toString());
+ d.put("where", String.valueOf(job.getWhereExpr()));
+ d.put("preceding_filter", String.valueOf(job.getPrecedingFilter()));
+ d.put("delete_on", String.valueOf(job.getDeleteCondition()));
+ d.put("partitions", job.getPartitionNamesInfo() == null ? null
+ : job.getPartitionNamesInfo().isTemp() + " " +
job.getPartitionNamesInfo().getPartitionNames());
+ d.put("comment", Deencapsulation.getField(job, "comment"));
+ d.put("data_source", job.getDataSourceProperties());
+ d.put("custom", job.getCustomProperties());
+ return d;
+ }
+
+ // Recreates the job from what SHOW CREATE prints for it, which prints the
same statement for the new job.
+ private KafkaRoutineLoadJob assertRoundTrip(String sql) throws Exception {
+ KafkaRoutineLoadJob job = createJob(sql);
+ String showCreate = job.getShowCreateInfo();
+ KafkaRoutineLoadJob recreated = createJob(showCreate);
+ Assertions.assertEquals(definition(job), definition(recreated),
showCreate);
+ Assertions.assertEquals(showCreate, recreated.getShowCreateInfo());
+ return job;
+ }
+
+ @Test
+ public void testCsvJob() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_csv ON dup_tbl\n"
+ + "COLUMNS TERMINATED BY ',',\n"
+ + "COLUMNS(k1, v1),\n"
+ + "WHERE k1 > 0\n"
+ + "PROPERTIES (\n"
+ + "'desired_concurrent_number' = '2',\n"
+ + "'max_error_number' = '5',\n"
+ + "'max_filter_ratio' = '0.5',\n"
+ + "'max_batch_interval' = '20',\n"
+ + "'max_batch_rows' = '300000',\n"
+ + "'max_batch_size' = '209715200',\n"
+ + "'strict_mode' = 'true',\n"
+ + "'timezone' = 'Asia/Tokyo',\n"
+ + "'exec_mem_limit' = '4294967296',\n"
+ + "'send_batch_parallelism' = '3',\n"
+ + "'load_to_single_tablet' = 'true',\n"
+ + "'workload_group' = 'normal',\n"
+ + "'format' = 'csv',\n"
+ + "'enclose' = '\"',\n"
+ + "'escape' = '\\\\',\n"
+ + "'empty_field_as_null' = 'true'\n"
+ + ")\n"
+ + "FROM KAFKA (\n"
+ + "'kafka_broker_list' = '127.0.0.1:9092',\n"
+ + "'kafka_topic' = 'topic_csv',\n"
+ + "'property.kafka_default_offsets' = 'OFFSET_BEGINNING'\n"
+ + ")\n"
+ + "COMMENT 'it''s the \"csv\" job'");
+ Assertions.assertEquals(3, job.getSendBatchParallelism());
+ Assertions.assertTrue(job.isLoadToSingleTablet());
+ Assertions.assertEquals("normal", job.getWorkloadGroup());
+ Assertions.assertEquals('"', job.getEnclose());
+ Assertions.assertEquals('\\', job.getEscape());
+ Assertions.assertTrue(job.getEmptyFieldAsNull());
+ Assertions.assertEquals("it's the \"csv\" job",
Deencapsulation.getField(job, "comment"));
+ }
+
+ // UPDATE_FLEXIBLE_COLUMNS leaves partial_columns false, the only thing
the statement used to carry.
+ @Test
+ public void testFlexiblePartialUpdateJob() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_flexible ON mow_tbl\n"
+ + "PROPERTIES (\n"
+ + "'format' = 'json',\n"
+ + "'unique_key_update_mode' = 'UPDATE_FLEXIBLE_COLUMNS',\n"
+ + "'partial_update_new_key_behavior' = 'ERROR',\n"
+ + "'strip_outer_array' = 'true',\n"
+ + "'num_as_string' = 'true',\n"
+ + "'json_root' = '$.data'\n"
+ + ")\n"
+ + "FROM KAFKA (\n"
+ + "'kafka_broker_list' = '127.0.0.1:9092',\n"
+ + "'kafka_topic' = 'topic_flexible'\n"
+ + ")");
+ Assertions.assertEquals(TUniqueKeyUpdateMode.UPDATE_FLEXIBLE_COLUMNS,
job.getUniqueKeyUpdateMode());
+ Assertions.assertEquals(TPartialUpdateNewRowPolicy.ERROR,
+ Deencapsulation.getField(job, "partialUpdateNewKeyPolicy"));
+ }
+
+ // The quotes in the json paths used to end the string literal, so the
statement did not parse.
+ @Test
+ public void testFixedPartialUpdateJobWithJsonPaths() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_fixed ON mow_tbl\n"
+ + "COLUMNS(k1, v1)\n"
+ + "PROPERTIES (\n"
+ + "'format' = 'json',\n"
+ + "'partial_columns' = 'true',\n"
+ + "'partial_update_new_key_behavior' = 'ERROR',\n"
+ + "'jsonpaths' = '[\"$.k1\",\"$.v1\"]',\n"
+ + "'strip_outer_array' = 'true',\n"
+ + "'fuzzy_parse' = 'true'\n"
+ + ")\n"
+ + "FROM KAFKA (\n"
+ + "'kafka_broker_list' = '127.0.0.1:9092',\n"
+ + "'kafka_topic' = 'topic_fixed'\n"
+ + ")");
+ Assertions.assertEquals(TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS,
job.getUniqueKeyUpdateMode());
+ Assertions.assertEquals("[\"$.k1\",\"$.v1\"]", job.getJsonPaths());
+ }
+
+ // The default new key policy and an UPSERT job print nothing new.
+ @Test
+ public void testFixedPartialUpdateJobWithDefaultPolicy() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_fixed_append ON mow_tbl\n"
+ + "COLUMNS(k1, v1)\n"
+ + "PROPERTIES ('format' = 'json', 'unique_key_update_mode' =
'UPDATE_FIXED_COLUMNS')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_fixed_append')");
+
Assertions.assertFalse(job.getShowCreateInfo().contains(CreateRoutineLoadInfo.PARTIAL_UPDATE_NEW_KEY_POLICY));
+ KafkaRoutineLoadJob upsert = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_upsert ON mow_tbl\n"
+ + "PROPERTIES ('format' = 'json')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_upsert')");
+
Assertions.assertFalse(upsert.getShowCreateInfo().contains(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE));
+ }
+
+ @Test
+ public void testMergeJob() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_merge ON mow_tbl\n"
+ + "WITH MERGE\n"
+ + "COLUMNS TERMINATED BY '\\t',\n"
+ + "PRECEDING FILTER k1 > 1,\n"
+ + "DELETE ON v1 = 'x'\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_merge')");
+ Assertions.assertEquals(LoadTask.MergeType.MERGE, job.getMergeType());
+ }
+
+ // TEMPORARY used to be dropped, so the statement named a formal partition
that does not exist.
+ @Test
+ public void testTemporaryPartitions() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_temp_partition ON range_tbl\n"
+ + "TEMPORARY PARTITION(tp1)\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_temp_partition')");
+ Assertions.assertTrue(job.getPartitionNamesInfo().isTemp());
+ }
+
+ @Test
+ public void testQuotedKafkaProperty() throws Exception {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_client_id ON dup_tbl\n"
+ + "FROM KAFKA (\n"
+ + "'kafka_broker_list' = '127.0.0.1:9092',\n"
+ + "'kafka_topic' = 'topic_client_id',\n"
+ + "'property.client.id' = 'a\"b\\\\c'\n"
+ + ")");
+ Assertions.assertEquals("a\"b\\c",
job.getCustomProperties().get("property.client.id"));
+ }
+
+ // Under NO_BACKSLASH_ESCAPES a backslash is an ordinary character, both
in what the session types and in
+ // what SHOW CREATE prints for it.
+ @Test
+ public void testNoBackslashEscapes() throws Exception {
+ long sqlMode = connectContext.getSessionVariable().getSqlMode();
+
connectContext.getSessionVariable().setSqlMode(SqlModeHelper.MODE_NO_BACKSLASH_ESCAPES);
+ try {
+ KafkaRoutineLoadJob job = assertRoundTrip("CREATE ROUTINE LOAD
rl_db.job_no_backslash_escapes ON dup_tbl\n"
+ + "PROPERTIES ('enclose' = '\"', 'escape' = '\\')\n"
+ + "FROM KAFKA ('kafka_broker_list' = '127.0.0.1:9092',
'kafka_topic' = 'topic_no_escapes')");
+ Assertions.assertEquals('"', job.getEnclose());
+ Assertions.assertEquals('\\', job.getEscape());
+ } finally {
+ connectContext.getSessionVariable().setSqlMode(sqlMode);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
