github-actions[bot] commented on code in PR #65973:
URL: https://github.com/apache/doris/pull/65973#discussion_r3654305576
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java:
##########
@@ -518,6 +519,72 @@ public void
testStreamScanWithSelectedPartitionIdsMarksPartitionPruned() {
Assertions.assertFalse(prunedScan.hasPartitionPredicate());
}
+ @Test
+ public void testIncrementalStreamScanForcesValueColumnsNullable() throws
Exception {
+ // s2 is an incremental (INCREMENTAL read mode) stream over a
unique-key table
+ // (k1 key, k2 value). computeOutput must force non-key value columns
to nullable so the
+ // scan output stays consistent with the row-binlog after/before value
columns, while key
+ // columns keep their original nullability (forceNullable is skipped
for keys).
+ Database db = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ OlapTable base = (OlapTable)
db.getTableOrMetaException("tbl_stream_base");
+ boolean baseK1Nullable = base.getBaseSchema(false).stream()
+ .filter(c -> c.getName().equals("k1"))
+ .findFirst()
+ .orElseThrow()
+ .isAllowNull();
+
+ Plan analyzedPlan = PlanChecker.from(connectContext)
+ .analyze("select * from test_stream.s2")
+ .getCascadesContext()
+ .getRewritePlan();
+
+ LogicalOlapTableStreamScan streamScan =
findFirstLogicalStreamScan(analyzedPlan);
+ Assertions.assertNotNull(streamScan);
+
+ Slot k1 = findSlot(streamScan, "k1");
+ Slot k2 = findSlot(streamScan, "k2");
+ Assertions.assertNotNull(k1, "key column k1 must be present in stream
scan output");
+ Assertions.assertNotNull(k2, "value column k2 must be present in
stream scan output");
+ Assertions.assertEquals(baseK1Nullable, k1.nullable(),
+ "key column k1 must keep its original nullability (not
force-nullable)");
+ Assertions.assertTrue(k2.nullable(), "non-key value column k2 must be
forced nullable");
Review Comment:
[P2] Exercise the nullability branch with a NOT NULL value
Both fixtures declare `k2` as plain `INT`, which Doris makes nullable by
default. This assertion therefore passes even without the new
`withNullable(true)`, and the RESET test likewise compares `true` with `true`.
Please use an explicitly `NOT NULL` value column (and assert that catalog
baseline is false) so INCREMENTAL proves widening to nullable while RESET
proves it stays non-nullable.
##########
be/src/exec/operator/olap_scan_operator.cpp:
##########
@@ -512,6 +512,13 @@ bool OlapScanLocalState::_is_key_column(const std::string&
key_name) {
return res != p._olap_scan_node.key_column_name.end();
}
+bool OlapScanLocalState::can_push_down_column_predicate(const SlotDescriptor*
slot) {
+ // The Operator-level method handles static column capabilities. The
LocalState-level
+ // condition additionally handles the current scan range's binlog merge
mode.
+ return Base::can_push_down_column_predicate(slot) &&
+ (!_is_binlog_merge_scan() || _is_key_column(slot->col_name()));
Review Comment:
[P1] Preserve legacy TSO pushdown on merge scans
During a rolling upgrade, an old FE sends this reduced plan:
```text
Filter(__DORIS_BINLOG_TSO__ > start [AND <= end])
OlapScan(MIN_DELTA, no start_tso/end_tso scan-range fields)
```
This gate rejects the TSO slot because it is non-key, so the filter stays in
`Scanner::_conjuncts` and runs only after `BlockReader` has grouped the raw
changes. For a key with `APPEND@90`, `DELETE@150`, and window `(100, 200]`,
MIN_DELTA sees APPEND+DELETE and returns SKIP instead of the required DELETE.
`_init_tso_pushdown()` cannot recover the bound because the old scan range has
no TSO fields. Please exempt the row-binlog TSO system column (or extract the
legacy conjunct into forced bounds) and cover an old-FE/new-BE plan.
##########
regression-test/suites/table_stream_p0/test_min_delta_op_filter_correctness.groovy:
##########
@@ -0,0 +1,196 @@
+// 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.
+
+suite("test_min_delta_op_filter_correctness", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+ sql "DROP DATABASE IF EXISTS test_min_delta_op_filter_correctness_db"
+ sql "CREATE DATABASE test_min_delta_op_filter_correctness_db"
+ sql "USE test_min_delta_op_filter_correctness_db"
+ sql "set enable_nereids_planner=true"
+ sql "set enable_fallback_to_original_planner=false"
+
+ try {
+ sql "DROP STREAM IF EXISTS s"
+ sql "DROP TABLE IF EXISTS src"
+ sql "DROP TABLE IF EXISTS audit"
+
+ // MoW UNIQUE KEY with a user-defined sequence column, consumed by a
+ // MIN_DELTA stream. Consumption is materialized via INSERT INTO audit
+ // SELECT ... FROM s, which commits and advances the stream offset.
+ // A bare SELECT over the stream rolls back (does NOT advance) the
+ // offset, so after a DELETE the two identical bare SELECTs must return
+ // the same DELETE rows.
+ sql """
+ CREATE TABLE src (
+ id BIGINT NOT NULL,
+ version_no BIGINT NOT NULL,
+ payload VARCHAR(32) NOT NULL
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "function_column.sequence_col" = "version_no",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """
+ CREATE TABLE audit (
+ id BIGINT,
+ version_no BIGINT,
+ payload VARCHAR(32),
+ change_type VARCHAR(32)
+ ) ENGINE=OLAP
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+
+ sql """
+ CREATE STREAM s ON TABLE src
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+
+ // Round 1: three fresh inserts -> APPEND, consumed into audit.
+ sql "INSERT INTO src VALUES (1, 1, 'a1'), (2, 1, 'a2'), (3, 1, 'a3')"
+ sql "sync"
+
+ order_qt_stream_append_filtered_1 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND')
+ ORDER BY id
+ """
+
+ order_qt_stream_append_filtered_2 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ ORDER BY id
+ """
+
+ order_qt_stream_append_filtered_3 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND',
'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE')
+ ORDER BY id
+ """
+
+ sql """
+ INSERT INTO audit
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND', 'INSERT',
'UPDATE_AFTER', 'DELETE')
+ """
+
+ order_qt_audit_round1 """
+ SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER
BY change_type
+ """
+
+ // Round 2: bump the sequence column for the same keys -> UPDATE, only
+ // the UPDATE_AFTER image survives the WHERE filter and is consumed.
+ sql "INSERT INTO src VALUES (1, 2, 'b1'), (2, 2, 'b2'), (3, 2, 'b3')"
+ sql "sync"
+
+ order_qt_stream_update_filtered_1 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('UPDATE_BEFORE')
+ ORDER BY id
+ """
+
+ order_qt_stream_update_filtered_2 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('UPDATE_AFTER')
+ ORDER BY id
+ """
+
+ order_qt_stream_update_filtered_3 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND',
'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE')
+ ORDER BY id
+ """
+
+ order_qt_stream_update_filtered_4 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ ORDER BY id
+ """
+
+ sql """
+ INSERT INTO audit
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND',
'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE')
+ """
+
+ order_qt_audit_round2 """
+ SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER
BY change_type
+ """
+
+ // Round 3: delete two keys, then read the stream twice with bare
+ // SELECTs. Because bare SELECT rolls back the offset, both reads must
+ // observe the same DELETE rows carrying the pre-delete snapshot.
+ sql "DELETE FROM src WHERE id <= 2"
+ sql "sync"
+
+ order_qt_stream_delete_filtered_1 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ ORDER BY id
+ """
+
+ order_qt_stream_delete_filtered_2 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('DELETE')
+ ORDER BY id
+ """
+
+ order_qt_stream_delete_filtered_3 """
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND',
'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE')
+ ORDER BY id
+ """
+ sql """
+ INSERT INTO audit
+ SELECT id, version_no, payload, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM s
+ WHERE __DORIS_STREAM_CHANGE_TYPE_COL__ IN ('APPEND',
'UPDATE_BEFORE', 'UPDATE_AFTER', 'DELETE')
+ """
+
+ order_qt_audit_round3 """
+ SELECT change_type, count(*) FROM audit GROUP BY change_type ORDER
BY change_type
+ """
+
+ } finally {
Review Comment:
[P3] Keep the failed regression state available
The repository test rules require dropping objects before use, not after, so
a failed suite leaves its state available for diagnosis. This `finally` block
removes the database even on failure, while line 22 already makes reruns
repeatable. Please remove the post-test drop.
--
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]
