This is an automated email from the ASF dual-hosted git repository.
gavinchou 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 efb70b32966 [fix](table stream) force binlog tso pred pushdown and
forbid other pred pushdown (#65973)
efb70b32966 is described below
commit efb70b3296692fec8594b32df12ab070e810b661
Author: TsukiokaKogane <[email protected]>
AuthorDate: Wed Jul 29 19:59:10 2026 +0800
[fix](table stream) force binlog tso pred pushdown and forbid other pred
pushdown (#65973)
### What problem does this PR solve?
Issue Number: close #65705 #65909
Problem Summary:
Table Stream queries may return incorrect results when predicates are
pushed down during `MIN_DELTA` or `DETAIL` row-binlog scans.
The root cause is that a row-binlog scan does not have the same row
semantics as its final Table Stream output:
1. A binlog record may contain both current values and before-image
values.
2. The raw binlog operation is different from the final stream
operation. For example, one raw UPDATE operation may produce
`UPDATE_BEFORE` and `UPDATE_AFTER` stream rows.
3. In `MIN_DELTA` and `DETAIL` modes, `BlockReader` may merge multiple
records or split one stored binlog row into multiple output rows.
Therefore, predicates on value columns or the binlog operation column
must not be evaluated by `SegmentIterator` before the binlog merge/split
step. At that point, the values and operation do not yet represent the
final stream rows. Early predicate evaluation can remove the raw row
needed to generate a valid `DELETE`, `UPDATE_BEFORE`, or `UPDATE_AFTER`
result.
This PR keeps non-key predicates as residual predicates above
`BlockReader` for `MIN_DELTA` and `DETAIL` scans. Key-column predicates
remain safe to push down because the key is unchanged by the binlog
transformation.
The TSO range predicate is a special case and must still be applied
before `BlockReader` performs MIN_DELTA grouping and aggregation. If
rows outside `(start_tso, end_tso]` reach the same-key group, they can
change the merged result and produce incorrect incremental output.
However, the normal TabletReader key/value predicate split does not
guarantee that the TSO predicate reaches `SegmentIterator`.
To guarantee correctness, this PR passes the TSO range through:
`OlapScanner -> TabletReader::ReaderParams -> RowsetReaderContext ->
BetaRowsetReader`
`BetaRowsetReader` then creates the TSO comparison predicates and
injects them directly into `StorageReadOptions`. This guarantees that
the TSO range is applied at the storage layer while all predicates that
depend on the final stream-row semantics remain above the binlog
merge/split operation.
The resulting behavior is:
- TSO predicates are always pushed down to restrict the raw binlog
input.
- Key-column predicates may still be pushed down.
- Value-column and operation-dependent predicates are evaluated only
after `MIN_DELTA` or `DETAIL` processing has produced the final stream
rows.
---
be/src/exec/operator/olap_scan_operator.cpp | 7 +
be/src/exec/operator/olap_scan_operator.h | 4 +
be/src/exec/operator/scan_operator.cpp | 2 +-
be/src/exec/operator/scan_operator.h | 4 +
be/src/exec/scan/olap_scanner.cpp | 41 ++---
be/src/exec/scan/olap_scanner.h | 2 +-
be/src/storage/rowset/beta_rowset_reader.cpp | 43 ++++-
be/src/storage/rowset/rowset_reader_context.h | 4 +
be/src/storage/tablet/tablet_reader.cpp | 2 +
be/src/storage/tablet/tablet_reader.h | 5 +
be/test/exec/operator/olap_scan_operator_test.cpp | 96 ++++++++++
be/test/storage/rowset/beta_rowset_reader_test.cpp | 192 ++++++++++++++++++++
be/test/storage/tablet_reader_test.cpp | 53 ++++++
.../doris/nereids/rules/analysis/BindRelation.java | 50 +++---
.../plans/logical/LogicalOlapTableStreamScan.java | 13 +-
.../trees/plans/ExplainTableStreamPlanTest.java | 125 ++++++++++++-
.../test_min_delta_op_filter_correctness.out | 68 +++++++
.../test_min_delta_op_filter_correctness.groovy | 196 +++++++++++++++++++++
18 files changed, 849 insertions(+), 58 deletions(-)
diff --git a/be/src/exec/operator/olap_scan_operator.cpp
b/be/src/exec/operator/olap_scan_operator.cpp
index 50d41d3dcc9..7a346f12152 100644
--- a/be/src/exec/operator/olap_scan_operator.cpp
+++ b/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()));
+}
+
Status OlapScanLocalState::_should_push_down_function_filter(VectorizedFnCall*
fn_call,
VExprContext*
expr_ctx,
StringRef*
constant_str,
diff --git a/be/src/exec/operator/olap_scan_operator.h
b/be/src/exec/operator/olap_scan_operator.h
index 1d2d2015039..9f7ddca0527 100644
--- a/be/src/exec/operator/olap_scan_operator.h
+++ b/be/src/exec/operator/olap_scan_operator.h
@@ -79,6 +79,8 @@ private:
Status _process_conjuncts(RuntimeState* state) override;
bool _is_key_column(const std::string& col_name) override;
+ bool can_push_down_column_predicate(const SlotDescriptor* slot) override;
+
Status _should_push_down_function_filter(VectorizedFnCall* fn_call,
VExprContext* expr_ctx,
StringRef* constant_str,
doris::FunctionContext** fn_ctx,
@@ -87,6 +89,7 @@ private:
PushDownType _should_push_down_bloom_filter() const override {
return PushDownType::ACCEPTABLE;
}
+
PushDownType _should_push_down_topn_filter() const override { return
PushDownType::ACCEPTABLE; }
PushDownType _should_push_down_is_null_predicate(VectorizedFnCall*
fn_call) const override {
@@ -95,6 +98,7 @@ private:
? PushDownType::ACCEPTABLE
: PushDownType::UNACCEPTABLE;
}
+
PushDownType _should_push_down_in_predicate() const override {
return PushDownType::ACCEPTABLE;
}
diff --git a/be/src/exec/operator/scan_operator.cpp
b/be/src/exec/operator/scan_operator.cpp
index 3bb234d16b1..e1aff774e7a 100644
--- a/be/src/exec/operator/scan_operator.cpp
+++ b/be/src/exec/operator/scan_operator.cpp
@@ -619,7 +619,7 @@ bool
ScanLocalState<Derived>::_is_predicate_acting_on_slot(const VExprSPtrs& chi
if (_slot_id_to_value_range.end() == sid_to_range) {
return false;
}
- if (!_parent->cast<typename
Derived::Parent>().can_push_down_column_predicate(*slot_desc)) {
+ if (!can_push_down_column_predicate(*slot_desc)) {
return false;
}
*range = &(sid_to_range->second);
diff --git a/be/src/exec/operator/scan_operator.h
b/be/src/exec/operator/scan_operator.h
index ae93e155852..e11eb506a77 100644
--- a/be/src/exec/operator/scan_operator.h
+++ b/be/src/exec/operator/scan_operator.h
@@ -295,6 +295,10 @@ protected:
}
virtual bool _should_push_down_common_expr(const VExprSPtr&) { return
false; }
+ virtual bool can_push_down_column_predicate(const SlotDescriptor* slot) {
+ return _parent->cast<typename
Derived::Parent>().can_push_down_column_predicate(slot);
+ }
+
virtual bool _storage_no_merge() { return false; }
virtual bool _is_key_column(const std::string& col_name) { return false; }
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index 834ee012737..fee0f1b6199 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -60,7 +60,6 @@
#include "storage/olap_common.h"
#include "storage/olap_tuple.h"
#include "storage/olap_utils.h"
-#include "storage/predicate/predicate_creator.h"
#include "storage/storage_engine.h"
#include "storage/tablet/tablet_schema.h"
#ifndef NDEBUG
@@ -103,7 +102,9 @@ OlapScanner::OlapScanner(ScanLocalStateBase* parent,
OlapScanner::Params&& param
.collection_statistics {},
.ann_topn_runtime {},
.condition_cache_digest =
parent->get_condition_cache_digest(),
- .binlog_scan_type = params.binlog_scan_type}),
+ .binlog_scan_type = params.binlog_scan_type,
+ .start_tso = std::nullopt,
+ .end_tso = std::nullopt}),
_start_tso(params.start_tso),
_end_tso(params.end_tso),
_initial_file_cache_stats(std::move(params.initial_file_cache_stats)) {
@@ -328,10 +329,11 @@ Status OlapScanner::_open_impl(RuntimeState* state) {
return Status::OK();
}
-// For binlog partition-based incremental read. Pushes down [start_tso,
end_tso] range
-// predicates onto the TSO column. If the TSO column is not already returned,
pass it as
-// a storage-only predicate column instead of widening scan output.
-Status OlapScanner::_init_tso_predicates() {
+// For binlog/snapshot incremental read. Forwards the (start_tso, end_tso]
range and the TSO
+// column id down to BetaRowsetReader, which builds the comparison predicates
directly on read
+// options. This bypasses the value/key predicate split in
TabletReader::_init_conditions_param,
+// guaranteeing the range filter always reaches storage (a correctness
requirement for MIN_DELTA).
+Status OlapScanner::_init_tso_pushdown() {
if (!_start_tso.has_value() && !_end_tso.has_value()) {
return Status::OK();
}
@@ -348,17 +350,11 @@ Status OlapScanner::_init_tso_predicates() {
column_name);
}
- auto data_type = std::make_shared<DataTypeInt64>();
- if (_start_tso.has_value()) {
- Field start_value = Field::create_field<TYPE_BIGINT>(*_start_tso);
-
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::GT>(
- tso_index, column_name, data_type, start_value, false));
- }
- if (_end_tso.has_value()) {
- Field end_value = Field::create_field<TYPE_BIGINT>(*_end_tso);
-
_tablet_reader_params.predicates.push_back(create_comparison_predicate<PredicateType::LE>(
- tso_index, column_name, data_type, end_value, false));
- }
+ // Push the TSO range down as-is; BetaRowsetReader builds the comparison
predicates and
+ // injects them straight into read options, so they cannot be dropped by
the value/key
+ // predicate split in TabletReader::_init_conditions_param.
+ _tablet_reader_params.start_tso = _start_tso;
+ _tablet_reader_params.end_tso = _end_tso;
// The storage-layer statistics fast path (VStatisticsIterator, picked when
// push_down_agg_type is COUNT/MINMAX) bypasses SegmentIterator and
returns raw
@@ -369,11 +365,10 @@ Status OlapScanner::_init_tso_predicates() {
// the binlog DETAIL/MIN_DELTA handling.
_tablet_reader_params.push_down_agg_type_opt = TPushAggOp::NONE;
- if (std::find(_tablet_reader_params.return_columns.begin(),
- _tablet_reader_params.return_columns.end(),
- tso_index) == _tablet_reader_params.return_columns.end()) {
- _tablet_reader_params.tso_predicate_column_id =
static_cast<ColumnId>(tso_index);
- }
+ // Always carry the tso column id so BetaRowsetReader can build predicates
on it.
+ // Whether the column must be appended to read_columns (because it is not
in
+ // return_columns) is decided downstream in BetaRowsetReader.
+ _tablet_reader_params.tso_predicate_column_id =
static_cast<ColumnId>(tso_index);
return Status::OK();
}
@@ -568,7 +563,7 @@ Status OlapScanner::_init_tablet_reader_params(
}
}
- RETURN_IF_ERROR(_init_tso_predicates());
+ RETURN_IF_ERROR(_init_tso_pushdown());
// For any row-binlog scan, force the storage layer to deliver rows
strictly in primary-key
// order so the BlockReader can group consecutive same-key changes
(MIN_DELTA) or emit
diff --git a/be/src/exec/scan/olap_scanner.h b/be/src/exec/scan/olap_scanner.h
index 308a6194756..0802140960e 100644
--- a/be/src/exec/scan/olap_scanner.h
+++ b/be/src/exec/scan/olap_scanner.h
@@ -111,7 +111,7 @@ private:
predicates,
const std::vector<FunctionFilter>& function_filters);
- [[nodiscard]] Status _init_tso_predicates();
+ [[nodiscard]] Status _init_tso_pushdown();
[[nodiscard]] Status _init_return_columns();
[[nodiscard]] Status _init_variant_columns();
#ifndef NDEBUG
diff --git a/be/src/storage/rowset/beta_rowset_reader.cpp
b/be/src/storage/rowset/beta_rowset_reader.cpp
index 708346e8414..a70708e765e 100644
--- a/be/src/storage/rowset/beta_rowset_reader.cpp
+++ b/be/src/storage/rowset/beta_rowset_reader.cpp
@@ -31,6 +31,7 @@
#include "common/logging.h"
#include "common/status.h"
#include "core/block/block.h"
+#include "core/data_type/data_type_number.h"
#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "runtime/query_context.h"
@@ -41,6 +42,7 @@
#include "storage/olap_define.h"
#include "storage/predicate/block_column_predicate.h"
#include "storage/predicate/column_predicate.h"
+#include "storage/predicate/predicate_creator.h"
#include "storage/row_cursor.h"
#include "storage/rowset/rowset_meta.h"
#include "storage/rowset/rowset_reader_context.h"
@@ -145,12 +147,15 @@ Status
BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
read_columns.push_back(cid);
}
}
- if (_read_context->tso_predicate_column_id.has_value()) {
+ if (_read_context->tso_predicate_column_id.has_value() &&
+ read_columns_set.find(*_read_context->tso_predicate_column_id) ==
read_columns_set.end()) {
read_columns.push_back(*_read_context->tso_predicate_column_id);
}
- // disable condition cache if you have delete condition
+ // disable condition cache if you have delete condition or forced pushed
tso predicate
_read_context->condition_cache_digest =
- delete_columns_set.empty() ? _read_context->condition_cache_digest
: 0;
+ (delete_columns_set.empty() &&
!_read_context->tso_predicate_column_id.has_value())
+ ? _read_context->condition_cache_digest
+ : 0;
// create segment iterators
VLOG_NOTICE << "read columns size: " << read_columns.size();
_input_schema =
std::make_shared<Schema>(_read_context->tablet_schema->columns(), read_columns);
@@ -174,6 +179,38 @@ Status
BetaRowsetReader::get_segment_iterators(RowsetReaderContext* read_context
}
}
+ // Forced TSO range pushdown for binlog/snapshot incremental read. Build
the (start_tso,
+ // end_tso] comparison predicates here and inject them directly, so they
always reach the
+ // SegmentIterator for row-level filtering instead of going through the
value/key predicate
+ // split in TabletReader::_init_conditions_param (where they may be
dropped). This is a
+ // correctness requirement for MIN_DELTA, which groups consecutive
same-key rows and would
+ // produce wrong results if out-of-range rows leaked into a group.
+ if (_read_context->tso_predicate_column_id.has_value() &&
+ (_read_context->start_tso.has_value() ||
_read_context->end_tso.has_value())) {
+ ColumnId tso_cid = *_read_context->tso_predicate_column_id;
+ auto tso_data_type = std::make_shared<DataTypeInt64>();
+ const std::string& tso_col_name =
_read_context->tablet_schema->column(tso_cid).name();
+ auto add_tso_predicate = [&](std::shared_ptr<ColumnPredicate> pred) {
+ if (_read_options.col_id_to_predicates.count(pred->column_id()) <
1) {
+ _read_options.col_id_to_predicates.insert(
+ {pred->column_id(),
AndBlockColumnPredicate::create_shared()});
+ }
+
_read_options.col_id_to_predicates[pred->column_id()]->add_column_predicate(
+ SingleColumnBlockPredicate::create_unique(pred));
+ _read_options.column_predicates.push_back(std::move(pred));
+ };
+ if (_read_context->start_tso.has_value()) {
+ add_tso_predicate(create_comparison_predicate<PredicateType::GT>(
+ tso_cid, tso_col_name, tso_data_type,
+
Field::create_field<TYPE_BIGINT>(*_read_context->start_tso), false));
+ }
+ if (_read_context->end_tso.has_value()) {
+ add_tso_predicate(create_comparison_predicate<PredicateType::LE>(
+ tso_cid, tso_col_name, tso_data_type,
+ Field::create_field<TYPE_BIGINT>(*_read_context->end_tso),
false));
+ }
+ }
+
// Take a delete-bitmap for each segment, the bitmap contains all deletes
// until the max read version, which is read_context->version.second
if (_read_context->delete_bitmap != nullptr) {
diff --git a/be/src/storage/rowset/rowset_reader_context.h
b/be/src/storage/rowset/rowset_reader_context.h
index 7d37b03dc2a..18eb442eccc 100644
--- a/be/src/storage/rowset/rowset_reader_context.h
+++ b/be/src/storage/rowset/rowset_reader_context.h
@@ -61,6 +61,10 @@ struct RowsetReaderContext {
const std::vector<uint32_t>* return_columns = nullptr;
// TSO predicate column that is absent from return_columns but must be
read by storage.
std::optional<ColumnId> tso_predicate_column_id;
+ // Binlog/snapshot incremental read TSO range (start_tso, end_tso].
BetaRowsetReader builds
+ // the tso comparison predicates from this range and forces them onto read
options.
+ std::optional<int64_t> start_tso;
+ std::optional<int64_t> end_tso;
TPushAggOp::type push_down_agg_type_opt = TPushAggOp::NONE;
// column name -> column predicate
// adding column_name for predicate to make use of column selectivity
diff --git a/be/src/storage/tablet/tablet_reader.cpp
b/be/src/storage/tablet/tablet_reader.cpp
index 3feb0d9009b..f6eadc46ca6 100644
--- a/be/src/storage/tablet/tablet_reader.cpp
+++ b/be/src/storage/tablet/tablet_reader.cpp
@@ -182,6 +182,8 @@ Status TabletReader::_capture_rs_readers(const
ReaderParams& read_params) {
_reader_context.read_orderby_key_limit =
read_params.read_orderby_key_limit;
_reader_context.return_columns = &_return_columns;
_reader_context.tso_predicate_column_id =
read_params.tso_predicate_column_id;
+ _reader_context.start_tso = read_params.start_tso;
+ _reader_context.end_tso = read_params.end_tso;
_reader_context.read_orderby_key_columns =
!_orderby_key_columns.empty() ? &_orderby_key_columns : nullptr;
_reader_context.predicates = &_col_predicates;
diff --git a/be/src/storage/tablet/tablet_reader.h
b/be/src/storage/tablet/tablet_reader.h
index b916721588a..771588e9c93 100644
--- a/be/src/storage/tablet/tablet_reader.h
+++ b/be/src/storage/tablet/tablet_reader.h
@@ -228,6 +228,11 @@ public:
// General LIMIT budget forwarded to SegmentIterator. -1 means no
limit.
int64_t general_read_limit = -1;
TBinlogScanType::type binlog_scan_type = TBinlogScanType::NONE;
+ // Binlog/snapshot incremental read TSO range (start_tso, end_tso].
Forced down to
+ // BetaRowsetReader, which builds the tso predicates directly on read
options,
+ // bypassing the value/key predicate split in _init_conditions_param.
+ std::optional<int64_t> start_tso;
+ std::optional<int64_t> end_tso;
};
TabletReader() = default;
diff --git a/be/test/exec/operator/olap_scan_operator_test.cpp
b/be/test/exec/operator/olap_scan_operator_test.cpp
new file mode 100644
index 00000000000..bc66290106d
--- /dev/null
+++ b/be/test/exec/operator/olap_scan_operator_test.cpp
@@ -0,0 +1,96 @@
+// 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.
+
+#include "exec/operator/olap_scan_operator.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "common/object_pool.h"
+#include "core/data_type/data_type_number.h"
+#include "gen_cpp/PlanNodes_types.h"
+#include "gen_cpp/QueryCache_types.h"
+#include "testutil/desc_tbl_builder.h"
+#include "testutil/mock/mock_runtime_state.h"
+
+namespace doris {
+
+class OlapScanOperatorBinlogPushDownTest : public testing::Test {
+protected:
+ void SetUp() override {
+ _state = std::make_shared<MockRuntimeState>();
+
+ DescriptorTblBuilder desc_builder(&_pool);
+ desc_builder.declare_tuple()
+ << TupleDescBuilder::SlotType
{std::make_shared<DataTypeInt32>(), "k1"}
+ << TupleDescBuilder::SlotType
{std::make_shared<DataTypeInt32>(), "v1"};
+ _descs = desc_builder.build();
+ ASSERT_NE(_descs, nullptr);
+
+ auto* tuple_desc = _descs->get_tuple_descriptor(0);
+ ASSERT_NE(tuple_desc, nullptr);
+ ASSERT_EQ(tuple_desc->slots().size(), 2);
+ _key_slot = tuple_desc->slots()[0];
+ _value_slot = tuple_desc->slots()[1];
+
+ TOlapScanNode olap_scan_node;
+ olap_scan_node.__set_tuple_id(0);
+ olap_scan_node.__set_keyType(TKeysType::UNIQUE_KEYS);
+ olap_scan_node.__set_key_column_name({"k1"});
+ olap_scan_node.__set_key_column_type({TPrimitiveType::INT});
+
+ TPlanNode plan_node;
+ plan_node.__set_node_id(0);
+ plan_node.__set_node_type(TPlanNodeType::OLAP_SCAN_NODE);
+ plan_node.__set_num_children(0);
+ plan_node.__set_row_tuples({0});
+ plan_node.__set_olap_scan_node(olap_scan_node);
+
+ _parent = std::make_shared<OlapScanOperatorX>(&_pool, plan_node, 0,
*_descs, 1,
+ TQueryCacheParam {});
+ _local_state = OlapScanLocalState::create_shared(_state.get(),
_parent.get());
+
+ auto range = std::make_unique<TPaloScanRange>();
+ range->__set_binlog_scan_type(TBinlogScanType::MIN_DELTA);
+ _local_state->_scan_ranges.push_back(std::move(range));
+ ASSERT_TRUE(_local_state->_is_binlog_merge_scan());
+ }
+
+ ObjectPool _pool;
+ DescriptorTbl* _descs = nullptr;
+ SlotDescriptor* _key_slot = nullptr;
+ SlotDescriptor* _value_slot = nullptr;
+ std::shared_ptr<MockRuntimeState> _state;
+ std::shared_ptr<OlapScanOperatorX> _parent;
+ std::shared_ptr<OlapScanLocalState> _local_state;
+};
+
+TEST_F(OlapScanOperatorBinlogPushDownTest, MergeScanPushesDownOnlyKeyColumns) {
+ EXPECT_TRUE(_local_state->can_push_down_column_predicate(_key_slot));
+ EXPECT_FALSE(_local_state->can_push_down_column_predicate(_value_slot));
+}
+
+TEST_F(OlapScanOperatorBinlogPushDownTest,
AppendOnlyKeepsValuePredicatePushDown) {
+
_local_state->_scan_ranges[0]->__set_binlog_scan_type(TBinlogScanType::APPEND_ONLY);
+ ASSERT_FALSE(_local_state->_is_binlog_merge_scan());
+
+ EXPECT_TRUE(_local_state->can_push_down_column_predicate(_key_slot));
+ EXPECT_TRUE(_local_state->can_push_down_column_predicate(_value_slot));
+}
+
+} // namespace doris
diff --git a/be/test/storage/rowset/beta_rowset_reader_test.cpp
b/be/test/storage/rowset/beta_rowset_reader_test.cpp
new file mode 100644
index 00000000000..a85c8cadfa3
--- /dev/null
+++ b/be/test/storage/rowset/beta_rowset_reader_test.cpp
@@ -0,0 +1,192 @@
+// 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.
+
+#include "storage/rowset/beta_rowset_reader.h"
+
+#include <gen_cpp/olap_file.pb.h>
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <vector>
+
+#include "storage/predicate/column_predicate.h"
+#include "storage/rowset/beta_rowset.h"
+#include "storage/rowset/rowset_meta.h"
+#include "storage/rowset/rowset_reader_context.h"
+#include "storage/tablet/tablet_schema.h"
+
+namespace doris {
+
+// Verifies the forced TSO range pushdown in
BetaRowsetReader::get_segment_iterators (introduced
+// for binlog/snapshot incremental read). The (start_tso, end_tso] comparison
predicates and the
+// tso predicate column must be injected directly onto read options, and the
tso column must be
+// appended to read_columns only when it is not already in return_columns.
+//
+// The rowset here has zero segments, so get_segment_iterators skips segment
iterator creation
+// and returns after the injection logic, which is exactly the code path under
test.
+class BetaRowsetReaderTsoInjectionTest : public testing::Test {
+protected:
+ // Schema: (k1 int key, __binlog_tso__ bigint). Column index 1 is used as
the tso column.
+ static constexpr ColumnId kTsoColumnId = 1;
+
+ void SetUp() override {
+ _tablet_schema = std::make_shared<TabletSchema>();
+ TabletSchemaPB schema_pb;
+ schema_pb.set_keys_type(DUP_KEYS);
+ schema_pb.set_num_short_key_columns(1);
+ schema_pb.set_num_rows_per_row_block(1024);
+ schema_pb.set_next_column_unique_id(3);
+
+ ColumnPB* key = schema_pb.add_column();
+ key->set_unique_id(1);
+ key->set_name("k1");
+ key->set_type("INT");
+ key->set_is_key(true);
+ key->set_length(4);
+ key->set_index_length(4);
+ key->set_is_nullable(false);
+
+ ColumnPB* tso = schema_pb.add_column();
+ tso->set_unique_id(2);
+ tso->set_name("__binlog_tso__");
+ tso->set_type("BIGINT");
+ tso->set_is_key(false);
+ tso->set_length(8);
+ tso->set_is_nullable(false);
+
+ _tablet_schema->init_from_pb(schema_pb);
+
+ _rowset_meta = std::make_shared<RowsetMeta>();
+ RowsetMetaPB rowset_meta_pb;
+ rowset_meta_pb.set_rowset_id(10001);
+ rowset_meta_pb.set_tablet_id(1);
+ rowset_meta_pb.set_rowset_type(BETA_ROWSET);
+ rowset_meta_pb.set_rowset_state(VISIBLE);
+ rowset_meta_pb.set_start_version(2);
+ rowset_meta_pb.set_end_version(2);
+ rowset_meta_pb.set_num_segments(0);
+ rowset_meta_pb.set_num_rows(0);
+ rowset_meta_pb.set_empty(true);
+ _rowset_meta->init_from_pb(rowset_meta_pb);
+
+ _rowset = std::make_shared<BetaRowset>(_tablet_schema, _rowset_meta,
"");
+ _reader = std::make_shared<BetaRowsetReader>(_rowset);
+ }
+
+ RowsetReaderContext make_context() {
+ RowsetReaderContext ctx;
+ ctx.tablet_schema = _tablet_schema;
+ ctx.return_columns = &_return_columns;
+ ctx.need_ordered_result = false;
+ return ctx;
+ }
+
+ // Counts predicates that act on the tso column in the injected read
options.
+ static int count_tso_predicates(const BetaRowsetReader& reader) {
+ int count = 0;
+ for (const auto& pred : reader._read_options.column_predicates) {
+ if (pred->column_id() == kTsoColumnId) {
+ ++count;
+ }
+ }
+ return count;
+ }
+
+ TabletSchemaSPtr _tablet_schema;
+ RowsetMetaSharedPtr _rowset_meta;
+ BetaRowsetSharedPtr _rowset;
+ std::shared_ptr<BetaRowsetReader> _reader;
+ std::vector<uint32_t> _return_columns = {0};
+};
+
+TEST_F(BetaRowsetReaderTsoInjectionTest, InjectBothStartAndEndTso) {
+ RowsetReaderContext ctx = make_context();
+ ctx.tso_predicate_column_id = kTsoColumnId;
+ ctx.start_tso = 100;
+ ctx.end_tso = 200;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ // Both GT(start) and LE(end) predicates are injected on the tso column.
+ EXPECT_EQ(count_tso_predicates(*_reader), 2);
+ EXPECT_EQ(_reader->_read_options.col_id_to_predicates.count(kTsoColumnId),
1);
+}
+
+TEST_F(BetaRowsetReaderTsoInjectionTest, InjectOnlyStartTso) {
+ RowsetReaderContext ctx = make_context();
+ ctx.tso_predicate_column_id = kTsoColumnId;
+ ctx.start_tso = 100;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ EXPECT_EQ(count_tso_predicates(*_reader), 1);
+ EXPECT_EQ(_reader->_read_options.col_id_to_predicates.count(kTsoColumnId),
1);
+}
+
+TEST_F(BetaRowsetReaderTsoInjectionTest, InjectOnlyEndTso) {
+ RowsetReaderContext ctx = make_context();
+ ctx.tso_predicate_column_id = kTsoColumnId;
+ ctx.end_tso = 200;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ EXPECT_EQ(count_tso_predicates(*_reader), 1);
+ EXPECT_EQ(_reader->_read_options.col_id_to_predicates.count(kTsoColumnId),
1);
+}
+
+TEST_F(BetaRowsetReaderTsoInjectionTest, NoInjectionWhenTsoRangeAbsent) {
+ RowsetReaderContext ctx = make_context();
+ // Column id present but neither start nor end tso set: nothing should be
injected.
+ ctx.tso_predicate_column_id = kTsoColumnId;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ EXPECT_EQ(count_tso_predicates(*_reader), 0);
+}
+
+TEST_F(BetaRowsetReaderTsoInjectionTest,
AppendTsoColumnWhenNotInReturnColumns) {
+ RowsetReaderContext ctx = make_context();
+ // return_columns only has k1 (id 0), tso column (id 1) must be appended
to read columns.
+ ctx.tso_predicate_column_id = kTsoColumnId;
+ ctx.start_tso = 100;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ // input schema (read columns) = return_columns + appended tso column = 2
columns.
+ EXPECT_EQ(_reader->_input_schema->num_column_ids(), 2);
+}
+
+TEST_F(BetaRowsetReaderTsoInjectionTest,
NotAppendTsoColumnWhenAlreadyInReturnColumns) {
+ RowsetReaderContext ctx = make_context();
+ // tso column already selected in return_columns: it must not be appended
twice.
+ std::vector<uint32_t> return_columns = {0, kTsoColumnId};
+ ctx.return_columns = &return_columns;
+ ctx.tso_predicate_column_id = kTsoColumnId;
+ ctx.start_tso = 100;
+
+ std::vector<RowwiseIteratorUPtr> iters;
+ ASSERT_TRUE(_reader->get_segment_iterators(&ctx, &iters).ok());
+
+ EXPECT_EQ(_reader->_input_schema->num_column_ids(), 2);
+}
+
+} // namespace doris
diff --git a/be/test/storage/tablet_reader_test.cpp
b/be/test/storage/tablet_reader_test.cpp
index 6c7db4fcb0c..ef8e9454122 100644
--- a/be/test/storage/tablet_reader_test.cpp
+++ b/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) {
+ RowsetReaderContext ctx;
+ ctx.start_tso = p.start_tso;
+ ctx.end_tso = p.end_tso;
+ return ctx;
+ };
+
+ // both set
+ params.start_tso = 100;
+ params.end_tso = 200;
+ RowsetReaderContext ctx = forward(params);
+ ASSERT_TRUE(ctx.start_tso.has_value());
+ ASSERT_TRUE(ctx.end_tso.has_value());
+ EXPECT_EQ(*ctx.start_tso, 100);
+ EXPECT_EQ(*ctx.end_tso, 200);
+
+ // only start set
+ params.start_tso = 100;
+ params.end_tso = std::nullopt;
+ ctx = forward(params);
+ ASSERT_TRUE(ctx.start_tso.has_value());
+ EXPECT_EQ(*ctx.start_tso, 100);
+ EXPECT_FALSE(ctx.end_tso.has_value());
+
+ // none set
+ params.start_tso = std::nullopt;
+ params.end_tso = std::nullopt;
+ ctx = forward(params);
+ EXPECT_FALSE(ctx.start_tso.has_value());
+ EXPECT_FALSE(ctx.end_tso.has_value());
+}
+
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
index 2348453c52d..73c0bf3c5fe 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java
@@ -76,7 +76,6 @@ import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.EqualTo;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.Expression;
-import org.apache.doris.nereids.trees.expressions.GreaterThan;
import org.apache.doris.nereids.trees.expressions.InPredicate;
import org.apache.doris.nereids.trees.expressions.LessThanEqual;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
@@ -256,7 +255,12 @@ public class BindRelation extends OneAnalysisRuleFactory {
List<Long> tabletIds = unboundRelation.getTabletIds();
StreamScanType changeScanType = checkChangeScanCondition((OlapTable)
table, unboundRelation.getScanParams());
if (changeScanType != null) {
- table = new RowBinlogTableWrapper((OlapTable) table);
+ table = new RowBinlogTableWrapper((OlapTable) table,
+ CollectionUtils.isEmpty(partIds)
+ ? makeUniformedTimestampRangeMap(((OlapTable)
table).getPartitionIds(),
+
parseTimestampRange(unboundRelation.getScanParams())) :
+ makeUniformedTimestampRangeMap(partIds,
+
parseTimestampRange(unboundRelation.getScanParams())));
} else if (unboundRelation.getScanParams() != null) {
unboundRelation.getScanParams().validateOlapTable();
}
@@ -307,8 +311,7 @@ public class BindRelation extends OneAnalysisRuleFactory {
throw new AnalysisException(
"PREAGGOPEN hint is not supported on @incr
(change-read) scans.");
}
- return checkAndAddChangeScanFilter(scan, changeScanType,
- parseTimestampRange(unboundRelation.getScanParams()),
false);
+ return checkAndAddChangeScanFilter(scan, changeScanType, false);
}
// Time-travel (FOR VERSION/TIME AS OF): wrap the scan with a
__DORIS_COMMIT_TSO_COL__
// predicate (dup) or a base/binlog union (mow).
@@ -597,7 +600,9 @@ public class BindRelation extends OneAnalysisRuleFactory {
// right: binlog MIN_DELTA over tso>t1, keep UPDATE_BEFORE/DELETE rows
(before image),
// projected to the same visible schema. BE splits each change so
UPDATE_BEFORE/DELETE rows
// already carry the pre-change value in the (same-named) value
columns.
- RowBinlogTableWrapper binlogTable = new
RowBinlogTableWrapper(olapTable);
+ RowBinlogTableWrapper binlogTable = new
RowBinlogTableWrapper(olapTable, CollectionUtils.isEmpty(partIds)
+ ? makeUniformedTimestampRangeMap(olapTable.getPartitionIds(),
Pair.of(targetTso, null)) :
+ makeUniformedTimestampRangeMap(partIds, Pair.of(targetTso,
null)));
RelationId binlogRelationId =
cascadesContext.getStatementContext().getNextRelationId();
LogicalOlapScan binlogScan = CollectionUtils.isEmpty(partIds)
? new LogicalOlapScan(binlogRelationId, binlogTable,
qualifier, tabletIds,
@@ -610,8 +615,7 @@ public class BindRelation extends OneAnalysisRuleFactory {
binlogScan = binlogScan.withTableScanParams(
new TableScanParams(TableScanParams.INCREMENTAL_READ,
incrParams, Lists.newArrayList()));
- LogicalPlan right = checkAndAddChangeScanFilter(binlogScan,
StreamScanType.MIN_DELTA, Pair.of(targetTso, null),
- true);
+ LogicalPlan right = checkAndAddChangeScanFilter(binlogScan,
StreamScanType.MIN_DELTA, true);
right = projectFromOriginSlots(right, visibleOutput);
// both children are bound; BindExpression aligns by position and
fills the union output.
@@ -1056,38 +1060,25 @@ public class BindRelation extends
OneAnalysisRuleFactory {
* Add append only filter on olap scan with changes if need.
*/
public static LogicalPlan checkAndAddChangeScanFilter(LogicalOlapScan scan,
- StreamScanType
scanType,
- Pair<Long, Long>
timestampRange, boolean beforeImageOnly) {
+ StreamScanType
scanType, boolean beforeImageOnly) {
LogicalPlan plan = scan;
- Slot timestampSlot = null;
Slot opSlot = null;
for (Slot slot : scan.getOutput()) {
- if (slot.getName().equals(Column.BINLOG_TSO_COL)) {
- timestampSlot = slot;
- }
if (slot.getName().equals(Column.BINLOG_OPERATION_COL)) {
opSlot = slot;
- }
- if (opSlot != null && timestampSlot != null) {
break;
}
}
- List<Expression> conjuncts = Lists.newArrayList();
- Preconditions.checkArgument(timestampSlot != null, "timestampSlot is
null");
- conjuncts.add(new GreaterThan(timestampSlot, new
BigIntLiteral(timestampRange.first)));
- if (timestampRange.second != null) {
- conjuncts.add(new LessThanEqual(timestampSlot, new
BigIntLiteral(timestampRange.second)));
- }
if (scanType.equals(StreamScanType.APPEND_ONLY)) {
Preconditions.checkArgument(opSlot != null, "opSlot is null");
- conjuncts.add(new EqualTo(opSlot, new
BigIntLiteral(BinlogUtils.ROW_BINLOG_APPEND)));
- }
- if (beforeImageOnly) {
- conjuncts.add(new InPredicate(opSlot, ImmutableList.of(
+ return new LogicalFilter<>(ImmutableSet.of(new EqualTo(opSlot,
+ new BigIntLiteral(BinlogUtils.ROW_BINLOG_APPEND))), plan);
+ } else if (beforeImageOnly) {
+ return new LogicalFilter<>(ImmutableSet.of(new InPredicate(opSlot,
ImmutableList.of(
new BigIntLiteral(BinlogUtils.ROW_BINLOG_DELETE),
- new BigIntLiteral(BinlogUtils.ROW_BINLOG_UPDATE_BEFORE))));
+ new
BigIntLiteral(BinlogUtils.ROW_BINLOG_UPDATE_BEFORE)))), plan);
}
- return new LogicalFilter<>(ImmutableSet.copyOf(conjuncts), plan);
+ return plan;
}
private LogicalPlan projectFromOriginSlots(LogicalPlan plan, List<Slot>
wantedSlots) {
@@ -1104,4 +1095,9 @@ public class BindRelation extends OneAnalysisRuleFactory {
}
return new LogicalProject<>(project, plan);
}
+
+ private Map<Long, Pair<Long, Long>>
makeUniformedTimestampRangeMap(List<Long> partIds,
+
Pair<Long, Long> timestampRange) {
+ return partIds.stream().collect(Collectors.toMap(partId -> partId,
partId -> timestampRange));
+ }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
index 47ba78d7a61..725bf059c55 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalOlapTableStreamScan.java
@@ -144,7 +144,18 @@ public class LogicalOlapTableStreamScan extends
LogicalOlapScan {
continue;
}
Pair<Long, String> key = Pair.of(selectedIndexId, col.getName());
- Slot slot = cacheSlotWithSlotName.computeIfAbsent(key, k ->
slotFromColumn.get(index));
+ // For INCREMENTAL / SNAPSHOT reads, non-key value columns are
materialized from the
+ // base table row-binlog whose after/before value columns are
always nullable (see
+ // Column.generateAfterValueColumn / generateBeforeValueColumn).
Declare these value
+ // columns as nullable here so the stream scan output stays
consistent with the plan
+ // expanded in NormalizeOlapTableStreamScan, otherwise
AdjustNullable reports a
+ // not-nullable -> nullable conflict. RESET does a full base-table
scan, so keep its
+ // original nullability.
+ Slot slot = cacheSlotWithSlotName.computeIfAbsent(key, k -> {
+ SlotReference slotRef = slotFromColumn.get(index);
+ boolean forceNullable = readMode != StreamReadMode.RESET &&
!baseSchema.get(index).isKey();
+ return forceNullable ? slotRef.withNullable(true) : slotRef;
+ });
slots.add(slot);
if (colToSubPathsMap.containsKey(key.getValue())) {
for (List<String> subPath :
colToSubPathsMap.get(key.getValue())) {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
index edc47a7dce5..52c419556b5 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
@@ -44,6 +44,7 @@ import org.apache.doris.nereids.parser.NereidsParser;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator;
import org.apache.doris.nereids.trees.expressions.literal.VarcharLiteral;
@@ -61,6 +62,7 @@ import org.apache.doris.qe.ConnectContext;
import org.apache.doris.thrift.TBinlogScanType;
import org.apache.doris.thrift.TPaloScanRange;
import org.apache.doris.thrift.TScanRangeLocations;
+import org.apache.doris.tso.TSOTimestamp;
import org.apache.doris.utframe.TestWithFeService;
import org.junit.jupiter.api.Assertions;
@@ -90,7 +92,7 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
String createBaseTable = "create table test_stream.tbl_stream_base (\n"
+ " k1 int,\n"
- + " k2 int\n"
+ + " k2 int not null\n"
+ ")\n"
+ "unique key(k1)\n"
+ "partition by range(k1)\n"
@@ -123,7 +125,7 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
String createDuplicateBaseTable = "create table
test_stream.tbl_dup_stream_base (\n"
+ " k1 int,\n"
- + " k2 int\n"
+ + " k2 int not null\n"
+ ")\n"
+ "duplicate key(k1)\n"
+ "distributed by hash(k1) buckets 1\n"
@@ -518,6 +520,125 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
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();
+ boolean baseK2Nullable = base.getBaseSchema(false).stream()
+ .filter(c -> c.getName().equals("k2"))
+ .findFirst()
+ .orElseThrow()
+ .isAllowNull();
+ Assertions.assertFalse(baseK2Nullable,
+ "value column k2 must be NOT NULL in the catalog baseline so
widening to nullable is provable");
+
+ 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");
+ }
+
+ @Test
+ public void testResetStreamScanKeepsOriginalNullability() throws Exception
{
+ // s_dup@reset does a full base-table scan (RESET read mode).
computeOutput must NOT force
+ // value columns to nullable; the value column keeps the base table's
original nullability.
+ Database db = (Database)
Env.getCurrentInternalCatalog().getDbOrMetaException("test_stream");
+ OlapTable dupBase = (OlapTable)
db.getTableOrMetaException("tbl_dup_stream_base");
+ boolean baseK2Nullable = dupBase.getBaseSchema(false).stream()
+ .filter(c -> c.getName().equals("k2"))
+ .findFirst()
+ .orElseThrow()
+ .isAllowNull();
+ Assertions.assertFalse(baseK2Nullable,
+ "value column k2 must be NOT NULL in the catalog baseline so
RESET can prove it stays non-nullable");
+
+ Plan analyzedPlan = PlanChecker.from(connectContext)
+ .analyze("select * from test_stream.s_dup@reset()")
+ .getCascadesContext()
+ .getRewritePlan();
+
+ LogicalOlapTableStreamScan streamScan =
findFirstLogicalStreamScan(analyzedPlan);
+ Assertions.assertNotNull(streamScan);
+
+ Slot k2 = findSlot(streamScan, "k2");
+ Assertions.assertNotNull(k2);
+ Assertions.assertEquals(baseK2Nullable, k2.nullable(),
+ "RESET scan must keep the base table's original nullability
for value columns");
+ }
+
+ private static Slot findSlot(LogicalOlapTableStreamScan scan, String name)
{
+ for (Slot slot : scan.getOutput()) {
+ if (slot.getName().equals(name)) {
+ return slot;
+ }
+ }
+ return null;
+ }
+
+ @Test
+ public void testIncrTimestampRangePropagatesToScanRangeTso() throws
Exception {
+ // An @incr('startTimestamp'=..., 'endTimestamp'=...) query on the MOW
base table goes
+ // through makeUniformedTimestampRangeMap in BindRelation, which seeds
a per-partition
+ // (start, end) offset on the RowBinlogTableWrapper.
OlapScanNode.getScanRangeLocations then
+ // stamps those offsets onto every scan range as startTso/endTso.
Verify the whole chain by
+ // asserting every incremental scan range carries the composed
start/end TSO for its partition.
+ String startTs = "2026-05-25 20:51:28";
+ String endTs = "2026-05-25 21:51:28";
+ long expectedStartTso =
TSOTimestamp.composeFullTimestamp(OlapScanNode.parseChangeTimestamp(startTs));
+ long expectedEndTso =
TSOTimestamp.composeFullTimestamp(OlapScanNode.parseChangeTimestamp(endTs));
+
+ ConnectContext ctx = createDefaultCtx();
+ ctx.setDatabase("test_stream");
+ ctx.getSessionVariable().showHiddenColumns = true;
+ StatementScopeIdGenerator.clear();
+ String sql = "explain select * from test_stream.tbl_stream_base"
+ + "@incr('startTimestamp' = '" + startTs + "', 'endTimestamp'
= '" + endTs + "')";
+ PlanFragment fragment = getFragment(ctx, sql);
+
+ List<OlapScanNode> scanNodes = new ArrayList<>();
+ collectOlapScanNodes(fragment.getPlanRoot(), scanNodes);
+ Assertions.assertFalse(scanNodes.isEmpty());
+
+ boolean assertedAtLeastOne = false;
+ for (OlapScanNode scanNode : scanNodes) {
+ if (!(scanNode.getOlapTable() instanceof RowBinlogTableWrapper)) {
+ continue;
+ }
+ List<TScanRangeLocations> locations =
scanNode.getScanRangeLocations(Long.MAX_VALUE);
+ Assertions.assertFalse(locations.isEmpty());
+ for (TScanRangeLocations loc : locations) {
+ TPaloScanRange range = loc.getScanRange().getPaloScanRange();
+ Assertions.assertEquals(expectedStartTso, range.getStartTso(),
+ "startTSO should equal the composed @incr
startTimestamp for every partition");
+ Assertions.assertEquals(expectedEndTso, range.getEndTso(),
+ "endTSO should equal the composed @incr endTimestamp
for every partition");
+ assertedAtLeastOne = true;
+ }
+ }
+ Assertions.assertTrue(assertedAtLeastOne,
+ "expected at least one incremental scan range to assert TSO
bounds against");
+ }
+
@Test
public void testRecordPlanForMvPreRewriteNormalizesStreamScanInsideCte()
throws Exception {
ConnectContext ctx = createDefaultCtx();
diff --git
a/regression-test/data/table_stream_p0/test_min_delta_op_filter_correctness.out
b/regression-test/data/table_stream_p0/test_min_delta_op_filter_correctness.out
new file mode 100755
index 00000000000..b563cf8c592
--- /dev/null
+++
b/regression-test/data/table_stream_p0/test_min_delta_op_filter_correctness.out
@@ -0,0 +1,68 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !stream_append_filtered_1 --
+1 1 a1 APPEND
+2 1 a2 APPEND
+3 1 a3 APPEND
+
+-- !stream_append_filtered_2 --
+1 1 a1 APPEND
+2 1 a2 APPEND
+3 1 a3 APPEND
+
+-- !stream_append_filtered_3 --
+1 1 a1 APPEND
+2 1 a2 APPEND
+3 1 a3 APPEND
+
+-- !audit_round1 --
+APPEND 3
+
+-- !stream_update_filtered_1 --
+1 1 a1 UPDATE_BEFORE
+2 1 a2 UPDATE_BEFORE
+3 1 a3 UPDATE_BEFORE
+
+-- !stream_update_filtered_2 --
+1 2 b1 UPDATE_AFTER
+2 2 b2 UPDATE_AFTER
+3 2 b3 UPDATE_AFTER
+
+-- !stream_update_filtered_3 --
+1 1 a1 UPDATE_BEFORE
+1 2 b1 UPDATE_AFTER
+2 1 a2 UPDATE_BEFORE
+2 2 b2 UPDATE_AFTER
+3 1 a3 UPDATE_BEFORE
+3 2 b3 UPDATE_AFTER
+
+-- !stream_update_filtered_4 --
+1 1 a1 UPDATE_BEFORE
+1 2 b1 UPDATE_AFTER
+2 1 a2 UPDATE_BEFORE
+2 2 b2 UPDATE_AFTER
+3 1 a3 UPDATE_BEFORE
+3 2 b3 UPDATE_AFTER
+
+-- !audit_round2 --
+APPEND 3
+UPDATE_AFTER 3
+UPDATE_BEFORE 3
+
+-- !stream_delete_filtered_1 --
+1 2 b1 DELETE
+2 2 b2 DELETE
+
+-- !stream_delete_filtered_2 --
+1 2 b1 DELETE
+2 2 b2 DELETE
+
+-- !stream_delete_filtered_3 --
+1 2 b1 DELETE
+2 2 b2 DELETE
+
+-- !audit_round3 --
+APPEND 3
+DELETE 2
+UPDATE_AFTER 3
+UPDATE_BEFORE 3
+
diff --git
a/regression-test/suites/table_stream_p0/test_min_delta_op_filter_correctness.groovy
b/regression-test/suites/table_stream_p0/test_min_delta_op_filter_correctness.groovy
new file mode 100644
index 00000000000..455d279e69e
--- /dev/null
+++
b/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 {
+ sql "DROP DATABASE IF EXISTS test_min_delta_op_filter_correctness_db"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
