luwei16 commented on code in PR #65973: URL: https://github.com/apache/doris/pull/65973#discussion_r3665365549
########## 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') Review Comment: [P2] Exercise #65705 in the consuming INSERT The separate filtered SELECTs catch the current generic early-pushdown bug, but this consuming assertion does not reproduce the reported `INSERT SELECT` path. By accepting `UPDATE_BEFORE` here (and at line 150), the resolved stream set includes code `2`; raw `ROW_BINLOG_DELETE` is also `2`, so an incorrectly pushed raw-op filter still lets DELETE through and this audit assertion can pass. Use the exact issue set `APPEND, INSERT, UPDATE_AFTER, DELETE` in both consuming INSERTs and regenerate the output. That also makes round 2 match its comment that only UPDATE_AFTER is consumed. ########## 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 """ Review Comment: [P2] Cover the mutable before-image and filtered DETAIL paths Every predicate in this suite targets `__DORIS_STREAM_CHANGE_TYPE_COL__`. It therefore still passes if a mutable value predicate is evaluated against the raw after image before `BlockReader`: the exact #65909 case is an update such as `payload: a1 -> b1`, where filtering for `a1` must retain `UPDATE_BEFORE(a1)` and filtering for `b1` must retain `UPDATE_AFTER(b1)`. Add those assertions. The same gate changes DETAIL, but existing DETAIL end-to-end cases are unfiltered; add one filtered DETAIL query as well. The eligibility unit test alone does not prove that the residual survives and executes after reconstruction. ########## be/test/storage/tablet_reader_test.cpp: ########## @@ -128,4 +128,57 @@ TEST_F(TabletReaderTest, remove_delete_columns_keeps_unrelated_paths) { EXPECT_EQ(size_t(2), access_paths.size()); } +// Contract test for the binlog/snapshot incremental-read TSO range forwarding added to +// TabletReader::_capture_rs_readers (tablet_reader.cpp:184-186): +// _reader_context.start_tso = read_params.start_tso; +// _reader_context.end_tso = read_params.end_tso; +// Exercising _capture_rs_readers end to end would require a fully constructed Tablet + rowset +// readers (it unconditionally dereferences _tablet), which is far too heavy and brittle for a +// unit test. Instead we pin down the field contract on both sides: both must be +// std::optional<int64_t> defaulting to nullopt, and the forwarding must preserve the optional +// state (both set / only one / none). This guards against the fields being dropped or their +// type changed, which would silently break the forwarding. +TEST_F(TabletReaderTest, forward_tso_range_field_contract) { + // Both sides default to nullopt. + TabletReader::ReaderParams params; + EXPECT_FALSE(params.start_tso.has_value()); + EXPECT_FALSE(params.end_tso.has_value()); + + RowsetReaderContext default_ctx; + EXPECT_FALSE(default_ctx.start_tso.has_value()); + EXPECT_FALSE(default_ctx.end_tso.has_value()); + + // Equivalent of the forwarding assignments; the optional state must be preserved verbatim. + auto forward = [](const TabletReader::ReaderParams& p) { Review Comment: [P3] Test the production forwarding rather than copying it This lambda reimplements the expected assignments in test code, so deleting either production assignment at `tablet_reader.cpp:185-186`, swapping start/end, or forwarding only one bound leaves this test green. It therefore cannot guard the behavior described in lines 131-140. Exercise `_capture_rs_readers` through a minimal fixture, extract the context-population step into a production helper used by this test, or remove this false-assurance unit and rely on the real end-to-end coverage. -- 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]
