github-actions[bot] commented on code in PR #64071:
URL: https://github.com/apache/doris/pull/64071#discussion_r3713606275
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -313,6 +315,69 @@ private void checkJobProperties() throws UserException {
analyzedJobProperties.put(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL,
String.valueOf(emptyFieldAsNull));
}
+
+ if
(jobProperties.containsKey(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS))
{
+ String val =
jobProperties.get(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS);
+ if (!"true".equalsIgnoreCase(val) &&
!"false".equalsIgnoreCase(val)) {
+ throw new
AnalysisException(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS
+ + " must be 'true' or 'false', but found: " + val);
+ }
+ // fill_missing_columns is a JSON-only property. This alter cannot
change the job format,
+ // so reject it for non-JSON jobs; otherwise the value would be
persisted but silently
+ // ignored by the CSV scan path.
+ RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
+ .getJob(getDbName(), getJobName());
+ if (!"json".equalsIgnoreCase(job.getFormat())) {
+ throw new
AnalysisException(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS
+ + " is only supported for JSON format, but found
format: " + job.getFormat());
+ }
+
analyzedJobProperties.put(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS,
val);
+ }
+
+ // fill_missing_columns performs a full-row upsert, which is mutually
exclusive with fixed
+ // partial columns update (see
CreateRoutineLoadInfo#checkJobProperties). Resolve both the
+ // fill_missing_columns flag and the update mode this alter will
result in (either changed by
+ // this alter or kept from the current job) and reject the combination.
+ if
(jobProperties.containsKey(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS)
+ ||
jobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE)
+ ||
jobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)) {
+ RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
+ .getJob(getDbName(), getJobName());
+ if (effectiveFillMissingColumns(job)
+ && effectiveUniqueKeyUpdateMode(job) ==
TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS) {
+ throw new
AnalysisException(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS
+ + " is not supported with fixed partial columns
update");
+ }
+ }
+ }
+
+ /**
+ * Resolve whether fill_missing_columns will be enabled after this alter:
prefer the value being
+ * changed by this alter, otherwise fall back to the current job's value.
+ */
+ private boolean effectiveFillMissingColumns(RoutineLoadJob job) {
+ if
(analyzedJobProperties.containsKey(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS))
{
+ return Boolean.parseBoolean(
+
analyzedJobProperties.get(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS));
+ }
+ return job.isFillMissingColumns();
+ }
+
+ /**
+ * Resolve the unique-key update mode this alter will result in: prefer
the mode being changed by
+ * this alter (unique_key_update_mode or the legacy partial_columns flag),
otherwise fall back to
+ * the current job's mode.
+ */
+ private TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode(RoutineLoadJob
job) {
+ if
(analyzedJobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE))
{
Review Comment:
The prospective mode here does not match the state transition that
`modifyCommonJobProperties()` actually applies. For a paused JSON job already
in `UPDATE_FIXED_COLUMNS`, an ALTER containing `fill_missing_columns=true` and
legacy `partial_columns=false` is treated here as UPSERT and passes, but the
mutation only changes `uniqueKeyUpdateMode` when the legacy flag is true and
the old mode is UPSERT; it therefore leaves the job fixed and then persists
fill=true. Likewise, on an UPSERT job, sending both
`unique_key_update_mode=UPSERT` and `partial_columns=true` is predicted as
UPSERT here but is applied sequentially as fixed mode. Both Kafka and Kinesis
journal and replay that forbidden state, re-exposing the fixed-partial
row-filtering behavior already discussed. Please derive validation and mutation
from one transition function (or reject conflicting mode keys) and cover actual
mutation/replay, not only command validation.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java:
##########
@@ -193,9 +195,41 @@ private void
fillContextExprMap(List<NereidsImportColumnDesc> columnDescList, Ne
// If user does not specify the file field names, generate it by using
base schema of table.
// So that the following process can be unified
boolean specifyFileFieldNames = copiedColumnExprs.stream().anyMatch(p
-> p.isColumn());
- if (!specifyFileFieldNames) {
+ boolean fillMissing = isFillMissingColumns(fileGroup);
+ if (!specifyFileFieldNames || fillMissing) {
Review Comment:
These synthesized file slots are not paired with entries in a non-empty
positional `jsonpaths` array. For example, with `COLUMNS(id, score,
score_x2=score*2)`, paths `["$.payload.id","$.payload.score"]`, and an omitted
non-null `name`, this loop appends `name` as a third source slot while
`toTFileAttributes()` still sends only the two user paths.
`NewJsonReader::_simdjson_write_columns_by_jsonpath()` treats every slot beyond
the path-array length as missing and calls `_fill_missing_column()`, so it
never reads a present `payload.name` and can reject the row. Please either
reject `fill_missing_columns=true` with non-empty `jsonpaths`, or rebuild the
paths in exactly the synthesized slot order, and add an end-to-end case for an
omitted non-null column.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -313,6 +315,69 @@ private void checkJobProperties() throws UserException {
analyzedJobProperties.put(CsvFileFormatProperties.PROP_EMPTY_FIELD_AS_NULL,
String.valueOf(emptyFieldAsNull));
}
+
+ if
(jobProperties.containsKey(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS))
{
+ String val =
jobProperties.get(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS);
+ if (!"true".equalsIgnoreCase(val) &&
!"false".equalsIgnoreCase(val)) {
+ throw new
AnalysisException(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS
+ + " must be 'true' or 'false', but found: " + val);
+ }
+ // fill_missing_columns is a JSON-only property. This alter cannot
change the job format,
+ // so reject it for non-JSON jobs; otherwise the value would be
persisted but silently
+ // ignored by the CSV scan path.
+ RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
+ .getJob(getDbName(), getJobName());
+ if (!"json".equalsIgnoreCase(job.getFormat())) {
+ throw new
AnalysisException(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS
+ + " is only supported for JSON format, but found
format: " + job.getFormat());
+ }
+
analyzedJobProperties.put(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS,
val);
+ }
+
+ // fill_missing_columns performs a full-row upsert, which is mutually
exclusive with fixed
+ // partial columns update (see
CreateRoutineLoadInfo#checkJobProperties). Resolve both the
+ // fill_missing_columns flag and the update mode this alter will
result in (either changed by
+ // this alter or kept from the current job) and reject the combination.
+ if
(jobProperties.containsKey(JsonFileFormatProperties.PROP_FILL_MISSING_COLUMNS)
+ ||
jobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE)
+ ||
jobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)) {
+ RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
+ .getJob(getDbName(), getJobName());
+ if (effectiveFillMissingColumns(job)
Review Comment:
This invariant is evaluated before either routine-load subclass acquires the
per-job write lock. Starting from a paused JSON job at fill=false/UPSERT, two
connection threads can validate `fill_missing_columns=true` and
`unique_key_update_mode=UPDATE_FIXED_COLUMNS` separately; each sees the old
compatible counterpart and passes. Their later Kafka/Kinesis mutations are
serialized, but the disjoint writes still leave fill=true plus fixed mode in
either order, and both edit-log records reproduce it after replay. Please
validate the combined current-plus-incoming state inside the same job
write-lock critical section that mutates and journals it, and add a
barrier-based concurrency test proving one conflicting ALTER is rejected.
##########
regression-test/suites/load_p0/routine_load/test_routine_load_fill_missing_columns.groovy:
##########
@@ -0,0 +1,455 @@
+// 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.
+
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerRecord
+
+// End-to-end coverage for the json `fill_missing_columns` routine load option.
+// It verifies that a job which declares only a derived column in COLUMNS can
still load
+// into a table that has a sequence column (the sequence column and other
base-schema
+// columns are auto-filled), and that the same job with `fill_missing_columns`
= false
+// keeps the original behavior and fails with "need to specify the sequence
column".
+suite("test_routine_load_fill_missing_columns", "p0") {
+ String enabled = context.config.otherConfigs.get("enableKafkaTest")
+ String kafka_port = context.config.otherConfigs.get("kafka_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ return
+ }
+
+ def dataFile = "test_routine_load_fill_missing_columns.json"
+ // Use a per-run topic suffix so OFFSET_BEGINNING jobs only see the
records produced in this run.
+ // Otherwise a topic retained from a previous run would let the exact
row-count assertions read
+ // stale messages and fail even when fill_missing_columns works correctly.
+ def topicSuffix = System.currentTimeMillis()
+ def topic = "test_routine_load_fill_missing_columns_${topicSuffix}"
+
+ // produce one json object per kafka message
+ def props = new Properties()
+ props.put("bootstrap.servers", "${externalEnvIp}:${kafka_port}".toString())
+ props.put("key.serializer",
"org.apache.kafka.common.serialization.StringSerializer")
+ props.put("value.serializer",
"org.apache.kafka.common.serialization.StringSerializer")
+ def producer = new KafkaProducer<>(props)
+ try {
+ def lines = new
File("""${context.file.parent}/data/${dataFile}""").text.readLines()
+ lines.each { line ->
+ if (line.trim().isEmpty()) {
+ return
+ }
+ logger.info("=====${line}========")
+ producer.send(new ProducerRecord<>(topic, null, line))
+ }
+ } finally {
+ producer.close()
+ }
+
+ // Build a unique-key table whose sequence column maps to a value column
that is NOT
+ // listed in COLUMNS. Without fill_missing_columns the sequence column
cannot be
+ // resolved and the job must fail.
+ def createTable = { tableName ->
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """
+ CREATE TABLE ${tableName} (
+ id INT NOT NULL,
+ name VARCHAR(50) NULL,
+ score INT NULL,
+ score_x2 INT NULL,
+ update_time BIGINT NULL
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "function_column.sequence_col" = "update_time"
+ );
+ """
+ }
+
+ //
---------------------------------------------------------------------------------
+ // Positive case: fill_missing_columns = true.
+ // COLUMNS only declares the derived column `score_x2`;
id/name/score/update_time and
+ // the sequence column are auto-filled from the base schema.
+ //
---------------------------------------------------------------------------------
+ def posTable = "test_routine_load_fill_missing_columns_pos"
+ def posJob = "test_routine_load_fill_missing_columns_pos_job"
+ try {
+ createTable(posTable)
+ sql "sync"
+
+ sql """
+ CREATE ROUTINE LOAD ${posJob} ON ${posTable}
+ COLUMNS(score_x2 = score * 2)
+ PROPERTIES
+ (
+ "format" = "json",
+ "fill_missing_columns" = "true",
+ "max_batch_interval" = "5",
+ "max_batch_rows" = "300000",
+ "max_batch_size" = "209715200",
+ "strict_mode" = "false"
+ )
+ FROM KAFKA
+ (
+ "kafka_broker_list" = "${externalEnvIp}:${kafka_port}",
+ "kafka_topic" = "${topic}",
+ "property.kafka_default_offsets" = "OFFSET_BEGINNING"
+ );
+ """
+ sql "sync"
+
+ // (a) the job must reach RUNNING and must NOT pause with the
sequence-column error
+ def count = 0
+ while (true) {
+ sleep(1000)
+ def res = sql "show routine load for ${posJob}"
+ def state = res[0][8].toString()
+ def reason = res[0][17].toString()
+ log.info("positive job state: ${state}, reason:
${reason}".toString())
+ assertFalse(reason.contains("need to specify the sequence column"),
+ "fill_missing_columns=true must not fail with sequence
column error, reason: ${reason}")
+ if (state == "RUNNING") {
+ break
+ }
+ count++
+ if (count >= 60) {
+ assertEquals("RUNNING", state)
+ break
+ }
+ }
+
+ // (b) the unspecified columns are auto-filled from the base schema
+ count = 0
+ while (true) {
+ def res = sql "select count(*) from ${posTable}"
+ def state = sql "show routine load for ${posJob}"
+ log.info("positive routine load state:
${state[0][8].toString()}".toString())
+ log.info("positive routine load statistic:
${state[0][14].toString()}".toString())
+ if (res[0][0] >= 3) {
+ break
+ }
+ if (count >= 60) {
+ log.error("positive routine load can not load data for long
time")
+ assertEquals(3, res[0][0])
+ break
+ }
+ sleep(5000)
+ count++
+ }
+ sql "sync"
+
+ def rows = sql "select id, name, score, score_x2, update_time from
${posTable} order by id"
Review Comment:
This and the other final ordered result blocks are deterministic after the
readiness loop, but the suite checks every cell with Groovy assertions and adds
no generated `.out` file. The regression-test contract requires determined
results to use named `order_qt_*`/`qt_*` cases so expected changes are
runner-generated and reviewable. Please keep assertions only for asynchronous
readiness/state checks, convert the final row/count queries at lines 155, 329,
and 434 to query-test cases, and commit the generated output.
--
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]
