github-actions[bot] commented on code in PR #64071:
URL: https://github.com/apache/doris/pull/64071#discussion_r3642581672
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java:
##########
@@ -421,15 +445,48 @@ private void
fillContextExprMap(List<NereidsImportColumnDesc> columnDescList, Ne
}
/**
- * if not set sequence column and column size is null or only have deleted
sign ,return true
+ * Returns true when the sequence column should be auto-added, i.e.,
+ * if not set sequence column and column size is null or only have deleted
sign,
+ * or fill_missing_columns is enabled, meaning schema will be auto-filled.
*/
- private boolean shouldAddSequenceColumn(List<NereidsImportColumnDesc>
columnDescList) {
+ private boolean shouldAddSequenceColumn(List<NereidsImportColumnDesc>
columnDescList,
+ NereidsBrokerFileGroup fileGroup) {
+ if (isFillMissingColumns(fileGroup)) {
+ return true;
+ }
if (columnDescList.isEmpty()) {
return true;
}
return columnDescList.size() == 1 &&
columnDescList.get(0).getColumnName().equalsIgnoreCase(Column.DELETE_SIGN);
}
+ /**
+ * Returns true if the file format is JSON and fill_missing_columns is
enabled. Only meaningful for JSON.
+ */
+ private boolean isFillMissingColumns(NereidsBrokerFileGroup fileGroup) {
+ return fileGroup.getFileFormatProperties() instanceof
JsonFileFormatProperties
+ && ((JsonFileFormatProperties)
fileGroup.getFileFormatProperties()).isFillMissingColumns();
+ }
+
+ /**
+ * Returns true when a mapping descriptor (expr != null) references a
source column whose name
+ * matches its own target column, e.g. COLUMNS(k1 = k1) or COLUMNS(k1 = k1
+ 1). Such a mapping
+ * still consumes the same-named source slot, so the base scan descriptor
for that column must be
+ * preserved when filling missing columns. Returns false for true file
fields and for mappings
+ * that do not reference their own column (constants or other-column
derivations).
+ */
+ private boolean mappingReferencesOwnColumn(NereidsImportColumnDesc desc) {
+ if (desc.isColumn()) {
+ return false;
+ }
+ for (Slot slot : desc.getExpr().getInputSlots()) {
+ if (slot.getName().equalsIgnoreCase(desc.getColumnName())) {
Review Comment:
The self-reference check is case-insensitive, but the raw descriptor added
by the base-schema loop uses the table column's spelling. For a table column
`Score`, JSON `{"score":10}`, and `COLUMNS(Score = score + 1)`, Nereids binds
`score` case-insensitively while `NewJsonReader` matches non-Hive JSON keys to
slot names exactly; it therefore scans `Score`, ignores `score`, and the
mapping receives NULL. Please preserve the matching source-slot spelling for
case-preserving formats and add a mixed-case JSON regression.
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/scalar/JsonExtractNoQuotesTest.java:
##########
@@ -0,0 +1,141 @@
+// 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.nereids.trees.expressions.functions.scalar;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.AlwaysNullable;
+import
org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.literal.StringLiteral;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.JsonType;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+public class JsonExtractNoQuotesTest {
+
+ @Test
+ public void testConstructorPreservesNameAndChildren() {
+ StringLiteral json = new StringLiteral("{\"a\":1}");
+ StringLiteral path = new StringLiteral("$.a");
+ JsonExtractNoQuotes fn = new JsonExtractNoQuotes(json, path);
+
+ Assertions.assertEquals("json_extract_no_quotes", fn.getName());
+ Assertions.assertEquals(2, fn.arity());
+ Assertions.assertSame(json, fn.child(0));
+ Assertions.assertSame(path, fn.child(1));
+ }
+
+ @Test
+ public void testConstructorAcceptsExtraVarArgs() {
+ StringLiteral json = new StringLiteral("{\"a\":{\"b\":1}}");
+ StringLiteral path0 = new StringLiteral("$.a");
+ StringLiteral path1 = new StringLiteral("$.a.b");
+ StringLiteral path2 = new StringLiteral("$.a.b.c");
+
+ JsonExtractNoQuotes fn = new JsonExtractNoQuotes(json, path0, path1,
path2);
+ Assertions.assertEquals(4, fn.arity());
+ Assertions.assertSame(path2, fn.child(3));
+ }
+
+ @Test
+ public void testGetSignaturesMatchesReturnAndArgs() {
+ JsonExtractNoQuotes fn = new JsonExtractNoQuotes(
+ new StringLiteral("{}"), new StringLiteral("$.a"));
+
+ List<FunctionSignature> signatures = fn.getSignatures();
+ Assertions.assertEquals(1, signatures.size());
+ FunctionSignature signature = signatures.get(0);
+ Assertions.assertEquals(JsonType.INSTANCE, signature.returnType);
+ Assertions.assertTrue(signature.hasVarArgs);
+ // First arg is JSON, remaining var args are VARCHAR paths.
+ Assertions.assertEquals(JsonType.INSTANCE,
signature.argumentsTypes.get(0));
+ Assertions.assertEquals(VarcharType.SYSTEM_DEFAULT,
signature.argumentsTypes.get(1));
+ }
+
+ @Test
+ public void testAlwaysNullableTrait() {
+ JsonExtractNoQuotes fn = new JsonExtractNoQuotes(
+ new StringLiteral("{}"), new StringLiteral("$.a"));
+
+ Assertions.assertTrue(fn instanceof AlwaysNullable);
+ Assertions.assertTrue(fn instanceof ExplicitlyCastableSignature);
+ Assertions.assertTrue(fn.nullable());
+ }
+
+ @Test
+ public void testWithChildrenKeepsNameAndReplacesChildren() {
+ JsonExtractNoQuotes original = new JsonExtractNoQuotes(
+ new StringLiteral("{\"a\":1}"), new StringLiteral("$.a"));
+
+ StringLiteral newJson = new StringLiteral("{\"b\":2}");
+ StringLiteral newPath = new StringLiteral("$.b");
+ JsonExtractNoQuotes rebuilt =
original.withChildren(ImmutableList.of(newJson, newPath));
+
+ Assertions.assertNotSame(original, rebuilt);
+ Assertions.assertEquals("json_extract_no_quotes", rebuilt.getName());
+ Assertions.assertEquals(2, rebuilt.arity());
+ Assertions.assertSame(newJson, rebuilt.child(0));
+ Assertions.assertSame(newPath, rebuilt.child(1));
+ }
+
+ @Test
+ public void testWithChildrenRejectsLessThanTwoArgs() {
+ JsonExtractNoQuotes original = new JsonExtractNoQuotes(
+ new StringLiteral("{\"a\":1}"), new StringLiteral("$.a"));
+
+ // Preconditions.checkArgument(children.size() >= 2) — dropping a
child violates the
+ // varargs signature (which requires at least the JSON value + one
path), so we must
+ // fail fast rather than silently return a malformed expression.
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> original.withChildren(ImmutableList.of(new
StringLiteral("{}"))));
+ Assertions.assertThrows(IllegalArgumentException.class,
+ () -> original.withChildren(ImmutableList.of()));
+ }
+
+ @Test
+ public void testAcceptDispatchesToJsonExtractNoQuotesVisitor() {
+ JsonExtractNoQuotes fn = new JsonExtractNoQuotes(
+ new StringLiteral("{}"), new StringLiteral("$.a"));
+
+ RecordingVisitor visitor = new RecordingVisitor();
+ String result = fn.accept(visitor, "ctx");
+ Assertions.assertEquals("visited:ctx", result);
+ Assertions.assertSame(fn, visitor.captured);
+ }
+
+ private static final class RecordingVisitor extends
ExpressionVisitor<String, String> {
+ private JsonExtractNoQuotes captured;
Review Comment:
This test is now a merge blocker: the live COMPILE and FE-UT jobs both fail
`testCompile` at this field and the override below because current master
removed `JsonExtractNoQuotes` (it became a `json_extract_string` alias in
#65380). Since this test is unrelated to `fill_missing_columns`, please remove
it from this PR (or rework it against the current API in the owning change); as
written the PR cannot compile after merging master.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsLoadScanProvider.java:
##########
@@ -193,9 +195,31 @@ 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);
Review Comment:
This also runs for `UPDATE_FIXED_COLUMNS`, although the partial-update set
was computed from the original `COLUMNS` list. The added columns enter the
scan/output tuple and `FileScanner` null-checks them, but BE later excludes
them from the writer's index slots. Consequently a fixed partial row can be
filtered because an omitted non-null column is NULL even though that column is
not being updated. Please reject this option with fixed partial mode, or keep
columns outside `partialUpdateInputColumns` out of the scan/output validation
path, and cover the combination.
##########
regression-test/suites/load_p0/routine_load/test_routine_load_fill_missing_columns.groovy:
##########
@@ -0,0 +1,353 @@
+// 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"
+ assertEquals(3, rows.size())
+ // id=1: name/score/update_time auto-filled from json, score_x2
derived as score*2
+ assertEquals(1, rows[0][0])
+ assertEquals("alice", rows[0][1].toString())
+ assertEquals(10, rows[0][2])
+ assertEquals(20, rows[0][3])
+ assertEquals(100L, rows[0][4])
+ // id=2
+ assertEquals(2, rows[1][0])
+ assertEquals("bob", rows[1][1].toString())
+ assertEquals(20, rows[1][2])
+ assertEquals(40, rows[1][3])
+ assertEquals(200L, rows[1][4])
+ // id=3
+ assertEquals(3, rows[2][0])
+ assertEquals("carol", rows[2][1].toString())
+ assertEquals(30, rows[2][2])
+ assertEquals(60, rows[2][3])
+ assertEquals(300L, rows[2][4])
+ } finally {
+ try {
+ sql "stop routine load for ${posJob}"
+ } catch (Exception e) {
+ log.info("stop positive routine load failed:
${e.getMessage()}".toString())
+ }
+ sql "DROP TABLE IF EXISTS ${posTable}"
Review Comment:
Please remove this table drop and the matching drops in the other two
`finally` blocks. These asynchronous tests already drop before creation;
dropping in `finally` also deletes the rows/state when polling or assertions
fail, contrary to the regression-test rule to preserve failed-test state for
debugging. Keeping the best-effort routine-load stop is fine.
--
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]
