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 369f887b254 [fix](table stream) Fix multi-segment row-binlog reads for
MoW tables (#66338)
369f887b254 is described below
commit 369f887b254d9b2e7ecd7fc4a077e19588414ef2
Author: TsukiokaKogane <[email protected]>
AuthorDate: Tue Aug 25 12:04:33 2026 +0800
[fix](table stream) Fix multi-segment row-binlog reads for MoW tables
(#66338)
### What problem does this PR solve?
Issue Number: close #65901
Related PR: #65418
Problem Summary:
Table Stream queries over ROW binlog could crash or return incorrect
results when a row-binlog rowset contained multiple segments. This
affected both multi-segment rowsets generated by load and non-singleton
overlapping rowsets produced by row-binlog LMax quick merge. Although
the source table may use UNIQUE_KEYS with Merge-on-Write, the scan reads
from its dedicated DUP_KEYS row-binlog tablet; the problem was caused by
inconsistent iterator selection and incorrect ordering and grouping
assumptions.
For a non-overlapping multi-segment rowset, `force_key_ordered_read`
caused `BetaRowsetReader` to build a merge iterator solely because
ordered output was required. The outer `VCollectIterator` could still
select its block-based union path because the rowset was
non-overlapping. This mismatch made the child iterator enter
by-reference mode while its parent called the block interface,
triggering the `!_get_data_by_ref` assertion and aborting the BE.
Non-overlapping segments do not require a merge iterator because an
ordered union already preserves key order.
Row-binlog LMax quick merge introduces another multi-segment layout: it
links segments from multiple rowsets into a non-singleton rowset marked
`OVERLAPPING`. The previous `is_segments_overlapping()` implementation
only recognized singleton delta rowsets, so these quick-merge rowsets
could incorrectly use the union path instead of merging their segments
by key and TSO. The TSO column could also be skipped after zone-map
pruning even though it is required as the merge sequence column.
In addition, MIN_DELTA processing used `IteratorRowRef::is_same` to
identify consecutive events belonging to the same user key. This flag is
a merge/dedup marker and does not reliably mark same-key events that
reside in one segment after quick merge. As a result, one key's change
chain could be split into multiple groups and folded incorrectly.
This change separates ordered-output requirements from iterator
selection: non-overlapping segments use the ordered union path, while
segments explicitly marked as overlapping use the merge path. It
recognizes non-singleton overlapping rowsets created by quick merge,
always materializes the row-binlog TSO column when it is needed for
ordering, and groups MIN_DELTA events by comparing their user-key
columns directly.
Before this change, affected queries could abort the BE, read
quick-merge rowsets out of order, or produce incorrect MIN_DELTA
results. After this change, multi-segment row-binlog rowsets are read
through a consistent iterator interface and processed in key/TSO order
with correct per-key change folding.
---
be/src/storage/iterator/block_reader.cpp | 28 +-
be/src/storage/merger.cpp | 6 +
be/src/storage/rowset/beta_rowset_reader.h | 3 +-
be/src/storage/rowset/beta_rowset_writer.cpp | 11 +
be/src/storage/rowset/rowset_meta.h | 5 +-
be/src/storage/rowset/rowset_reader_context.h | 1 -
be/src/storage/segment/segment_iterator.cpp | 9 +
be/src/storage/tablet/tablet_reader.cpp | 3 +-
be/src/storage/tablet/tablet_reader.h | 7 +-
be/test/exec/scan/vgeneric_iterators_test.cpp | 165 ++++++-
.../block_reader_binlog_vcollect_merge_test.cpp | 402 +++++++++++++++
.../iterator/block_reader_min_delta_test.cpp | 100 ++++
.../storage/row_binlog_vmerge_compaction_test.cpp | 400 +++++++++++++++
be/test/storage/rowset/rowset_meta_test.cpp | 54 +++
be/test/storage/tablet_reader_test.cpp | 6 +-
.../test_table_stream_multi_segment_mow.out | 79 +++
.../test_table_stream_multi_segment_mow.groovy | 538 +++++++++++++++++++++
17 files changed, 1800 insertions(+), 17 deletions(-)
diff --git a/be/src/storage/iterator/block_reader.cpp
b/be/src/storage/iterator/block_reader.cpp
index 29aa476bd0e..e91b16cd021 100644
--- a/be/src/storage/iterator/block_reader.cpp
+++ b/be/src/storage/iterator/block_reader.cpp
@@ -216,6 +216,11 @@ Status BlockReader::_min_delta_next_block(Block* block,
bool* eof) {
const int32_t tso_ordinal = _read_schema->tso_ordinal();
const int32_t lsn_ordinal = _read_schema->lsn_ordinal();
const int32_t op_ordinal = _read_schema->op_ordinal();
+ // A group is a run of consecutive rows sharing the same user key.
Row-binlog reads are
+ // globally key-ordered (ReaderParams::force_key_ordered_read), and the
key columns are the
+ // leading num_key_columns() columns of every read block, so a group
boundary is exactly
+ // where the user key changes. _stored_data_columns keeps the group's
first row at index 0.
+ const size_t num_key_columns = _tablet_schema->num_key_columns();
while (output_row_count < batch_max_rows()) {
if (_emit_pending_row(target_columns, output_row_count)) {
continue;
@@ -238,8 +243,27 @@ Status BlockReader::_min_delta_next_block(Block* block,
bool* eof) {
return res;
}
- if (!_eof && _next_row.is_same) {
- continue;
+ // Extend the current group while the next row shares the same user
key. is_same cannot
+ // be used here: it marks cross-segment key matches for dedup, so
consecutive same-key
+ // rows that a compaction/quick-merge folded into one segment are left
unmarked, which
+ // would split one key's change chain into several groups. Compare the
leading key
+ // columns directly against the group's first row (index 0 of
_stored_data_columns).
+ if (!_eof) {
+ if (_next_row.is_same) {
+ continue;
+ }
+ bool same_key = true;
+ for (size_t k = 0; k < num_key_columns; ++k) {
+ if (_stored_data_columns[k]->compare_at(0, _next_row.row_pos,
+
*_next_row.block->get_by_position(k).column,
+ -1) != 0) {
+ same_key = false;
+ break;
+ }
+ }
+ if (same_key) {
+ continue;
+ }
}
size_t group_size = _stored_data_columns[0]->size();
auto first_op = _read_binlog_op(*_stored_data_columns[op_ordinal], 0);
diff --git a/be/src/storage/merger.cpp b/be/src/storage/merger.cpp
index 45cf07f045b..dfcd333cad5 100644
--- a/be/src/storage/merger.cpp
+++ b/be/src/storage/merger.cpp
@@ -77,6 +77,12 @@ Status Merger::vmerge_rowsets(BaseTabletSPtr tablet,
ReaderType reader_type,
reader_params.tablet = tablet;
reader_params.reader_type = reader_type;
reader_params.read_row_binlog = tablet->is_row_binlog_tablet();
+ if (reader_params.read_row_binlog) {
+ // Row-binlog horizontal (non-vertical) compaction must produce a
globally
+ // (key, TSO)-ordered output.
+ reader_params.read_orderby_key = true;
+ reader_params.force_key_ordered_read = true;
+ }
TabletReadSource read_source;
read_source.rs_splits.reserve(src_rowset_readers.size());
diff --git a/be/src/storage/rowset/beta_rowset_reader.h
b/be/src/storage/rowset/beta_rowset_reader.h
index c1e2d96224b..c223a62089d 100644
--- a/be/src/storage/rowset/beta_rowset_reader.h
+++ b/be/src/storage/rowset/beta_rowset_reader.h
@@ -61,8 +61,7 @@ public:
bool is_merge_iterator() const override {
return _read_context->need_ordered_result && _get_segment_num() > 1 &&
- (_rowset->rowset_meta()->is_segments_overlapping() ||
- _read_context->force_key_ordered_read);
+ _rowset->rowset_meta()->is_segments_overlapping();
}
bool delete_flag() override { return _rowset->delete_flag(); }
diff --git a/be/src/storage/rowset/beta_rowset_writer.cpp
b/be/src/storage/rowset/beta_rowset_writer.cpp
index f967b8f49c0..013a49c1890 100644
--- a/be/src/storage/rowset/beta_rowset_writer.cpp
+++ b/be/src/storage/rowset/beta_rowset_writer.cpp
@@ -1074,6 +1074,17 @@ Status
BaseBetaRowsetWriter::_build_rowset_meta(RowsetMeta* rowset_meta, bool ch
!is_segment_overlapping(segments_encoded_key_bounds) &&
_context.tablet_schema->cluster_key_uids().empty()) {
rowset_meta->set_segments_overlap(NONOVERLAPPING);
+ } else if (!segments_encoded_key_bounds.empty() &&
+ is_segment_overlapping(segments_encoded_key_bounds) &&
+ _rowset_meta->is_row_binlog()) {
+ // A row-binlog horizontal merge output can hold the same user key in
more than
+ // one output segment (a forced segment boundary splits a key range),
so the
+ // segments genuinely overlap. The compaction ctx default is
NONOVERLAPPING for
+ // the non-quick-merge path, and the branch above only ever
downgrades, so upgrade
+ // here. Otherwise RowsetMeta::is_segments_overlapping() and
+ // BetaRowsetReader::is_merge_iterator() would pick a segment union
instead of a
+ // merge and break downstream MIN_DELTA reads.
+ rowset_meta->set_segments_overlap(OVERLAPPING);
}
auto segment_num = _num_seg();
diff --git a/be/src/storage/rowset/rowset_meta.h
b/be/src/storage/rowset/rowset_meta.h
index fde75545091..e73842368cf 100644
--- a/be/src/storage/rowset/rowset_meta.h
+++ b/be/src/storage/rowset/rowset_meta.h
@@ -306,8 +306,11 @@ public:
// 2. the rowset's start version == end version (non-singleton rowset was
generated by compaction process
// which always produces non-overlapped segments)
// 3. segments_overlap() flag is not NONOVERLAPPING (OVERLAP_UNKNOWN and
OVERLAPPING are OK)
+ // 4. relaxing is_singleton_delta() for is_row_binlog as Row-binlog LMax
quick merge does produce
+ // non-singleton overlapping segments, but it writes segments_overlap =
OVERLAPPING explicitly
bool is_segments_overlapping() const {
- return num_segments() > 1 && is_singleton_delta() &&
segments_overlap() != NONOVERLAPPING;
+ return num_segments() > 1 && segments_overlap() != NONOVERLAPPING &&
+ (is_singleton_delta() || (is_row_binlog() && segments_overlap()
== OVERLAPPING));
}
bool produced_by_compaction() const {
diff --git a/be/src/storage/rowset/rowset_reader_context.h
b/be/src/storage/rowset/rowset_reader_context.h
index 90f522c99ad..5f8e536a655 100644
--- a/be/src/storage/rowset/rowset_reader_context.h
+++ b/be/src/storage/rowset/rowset_reader_context.h
@@ -52,7 +52,6 @@ struct RowsetReaderContext {
// For rows with the same key, use ascending order (small-to-large) for
tie-breakers.
// For example, use lower rowset version / segment id first.
bool use_insert_order_when_same = false;
- bool force_key_ordered_read = false;
// ordinals (positions in read_schema) of the orderby key columns
std::vector<uint32_t>* read_orderby_key_columns = nullptr;
// limit of rows for read_orderby_key
diff --git a/be/src/storage/segment/segment_iterator.cpp
b/be/src/storage/segment/segment_iterator.cpp
index d1464cc43c9..08ec84d700f 100644
--- a/be/src/storage/segment/segment_iterator.cpp
+++ b/be/src/storage/segment/segment_iterator.cpp
@@ -1561,6 +1561,15 @@ bool SegmentIterator::_need_read_data(ColumnId cid) {
// occurring, return true here that column data needs to be read
return true;
}
+ // Row-binlog incremental reads force-push the TSO range predicate (see
+ // OlapScanner::_init_tso_predicate), and the merge iterator uses the TSO
column as its
+ // sequence sort key (BetaRowsetReader sets binlog_tso_idx). On a
cross-version rowset
+ // (e.g. produced by binlog LMax quick-merge) the TSO zonemap can be
always-true, so the
+ // pruning below would skip reading it and fill placeholder zeros,
breaking the merge
+ // ordering. The TSO column carries real values on disk, so force it to be
read.
+ if (_opts.read_row_binlog && cid == _schema->tso_ordinal()) {
+ return true;
+ }
const auto& column = *_schema->column(cid);
// Different subcolumns may share the same parent_unique_id, so we choose
to abandon this optimization.
if (column.is_extracted_column() &&
diff --git a/be/src/storage/tablet/tablet_reader.cpp
b/be/src/storage/tablet/tablet_reader.cpp
index 8986bf0b136..38eae53632f 100644
--- a/be/src/storage/tablet/tablet_reader.cpp
+++ b/be/src/storage/tablet/tablet_reader.cpp
@@ -159,12 +159,11 @@ Status TabletReader::_capture_rs_readers(const
ReaderParams& read_params) {
_reader_context.read_row_binlog = read_params.read_row_binlog;
_reader_context.version = read_params.version;
_reader_context.tablet_schema = _tablet_schema;
- _reader_context.need_ordered_result = need_ordered_result;
+ _reader_context.need_ordered_result = need_ordered_result ||
read_params.force_key_ordered_read;
_reader_context.topn_filter_source_node_ids =
read_params.topn_filter_source_node_ids;
_reader_context.read_orderby_key_reverse =
read_params.read_orderby_key_reverse;
_reader_context.use_insert_order_when_same =
read_params.use_insert_order_when_same ||
read_params.read_row_binlog;
- _reader_context.force_key_ordered_read =
read_params.force_key_ordered_read;
_reader_context.read_orderby_key_limit =
read_params.read_orderby_key_limit;
_reader_context.read_schema = _read_schema;
_reader_context.read_orderby_key_columns =
diff --git a/be/src/storage/tablet/tablet_reader.h
b/be/src/storage/tablet/tablet_reader.h
index b8a9f6366c4..03a15982d54 100644
--- a/be/src/storage/tablet/tablet_reader.h
+++ b/be/src/storage/tablet/tablet_reader.h
@@ -183,10 +183,9 @@ public:
// For rows with the same key, use ascending order (small-to-large)
for tie-breakers.
// For example, use lower rowset version / segment id first.
bool use_insert_order_when_same = false;
- // Force a key-ordered merge across all segments even when their key
ranges do not
- // overlap. By default a rowset reader can skip the merge heap if its
segments are
- // mono-ascending and disjoint, but row-binlog scans require strict
global key order
- // (e.g. so MIN_DELTA can group consecutive same-key changes), so this
flag is set.
+ // Force globally key-ordered reading for row-binlog scans (e.g. so
MIN_DELTA can
+ // group consecutive same-key changes across segments). Overlapping
segments use the
+ // merge iterator; segments proven non-overlapping use an ordered
union.
// See BetaRowsetReader::is_merge_iterator() in
beta_rowset_reader.h:62.
bool force_key_ordered_read = false;
// num of columns for orderby key
diff --git a/be/test/exec/scan/vgeneric_iterators_test.cpp
b/be/test/exec/scan/vgeneric_iterators_test.cpp
index d55b96b9ce8..1299057cb50 100644
--- a/be/test/exec/scan/vgeneric_iterators_test.cpp
+++ b/be/test/exec/scan/vgeneric_iterators_test.cpp
@@ -50,7 +50,7 @@ public:
virtual ~VGenericIteratorsTest() {}
};
-static ReadSchema create_schema() {
+static std::vector<TabletColumnPtr> create_col_schemas() {
std::vector<TabletColumnPtr> col_schemas;
auto c1 =
std::make_shared<TabletColumn>(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
FieldType::OLAP_FIELD_TYPE_SMALLINT, true);
@@ -65,7 +65,11 @@ static ReadSchema create_schema() {
col_schemas.emplace_back(
std::make_shared<TabletColumn>(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM,
FieldType::OLAP_FIELD_TYPE_BIGINT,
true));
- return ReadSchema(std::move(col_schemas));
+ return col_schemas;
+}
+
+static ReadSchema create_schema() {
+ return ReadSchema(create_col_schemas());
}
static void create_block(ReadSchema& schema, Block& block) {
@@ -474,4 +478,161 @@ TEST(VGenericIteratorsTest,
MergeWithSeqColumnSmallSeqFirst) {
EXPECT_EQ(0, actual_value);
}
+// Models the row-binlog quick-merge read path: the merge output rowset holds
the
+// same user key across multiple segments, each carrying a distinct binlog
event
+// (distinct TSO). A row-binlog scan reads it with is_unique=false and the TSO
as
+// the sequence column with small_seq_first=true. Unlike the is_unique=true
cases
+// above (which collapse same-key rows to one), the merge iterator must keep
EVERY
+// event and emit them ordered by ascending TSO, so a table-stream query over a
+// quick-merge rowset returns the complete, ordered change chain without
dropping
+// any event.
+TEST(VGenericIteratorsTest, MergeKeepsAllBinlogEventsOrderedByTso) {
+ auto schema = create_schema();
+ auto output_schema = std::make_shared<ReadSchema>(schema);
+ std::vector<RowwiseIteratorUPtr> inputs;
+
+ int seq_column_id = 2; // BIGINT column stands in for __DORIS_BINLOG_TSO__.
+ int num_rows = 1;
+ int rows_begin = 0;
+ // Same key in every segment, each carrying a distinct TSO. The children
are fed in a
+ // deterministic SHUFFLED TSO order (not ascending), so the test only
passes if the
+ // merge actually orders by the TSO sequence column; a comparator that
ignored the
+ // sequence and fell back to child/data-id arrival order would emit this
shuffled
+ // sequence unchanged and fail the ascending-order assertion below.
+ const std::vector<int> tso_feed_order = {7, 1, 9, 0, 5, 8, 2, 6, 4, 3};
+ int seg_iter_num = static_cast<int>(tso_feed_order.size());
+ for (int tso : tso_feed_order) {
+ inputs.push_back(std::make_unique<SeqColumnUtIterator>(schema,
num_rows, rows_begin,
+ seq_column_id,
tso));
+ }
+
+ // is_unique=false keeps every same-key event; small_seq_first=true orders
by
+ // ascending TSO. This mirrors beta_rowset_reader's row-binlog merge setup.
+ auto iter = new_merge_iterator(std::move(inputs), seq_column_id,
/*is_unique=*/false,
+ /*is_reverse=*/false,
/*merged_rows=*/nullptr, output_schema,
+ /*small_seq_first=*/true);
+ StorageReadOptions opts;
+ auto st = iter->init(opts);
+ EXPECT_TRUE(st.ok());
+
+ Block block;
+ std::vector<bool> row_is_same;
+ BlockWithSameBit block_with_same_bit {.block = &block, .same_bit =
row_is_same};
+ create_block(schema, block);
+
+ do {
+ st = iter->next_batch(&block_with_same_bit);
+ } while (st.ok());
+
+ EXPECT_TRUE(st.is<END_OF_FILE>());
+ // Every event survives: no per-key dedup.
+ EXPECT_EQ(block.rows(), seg_iter_num);
+
+ // Despite the shuffled input order, events are emitted in ascending TSO
order.
+ auto seq_col = block.get_by_position(seq_column_id).column;
+ for (int i = 0; i < seg_iter_num; ++i) {
+ EXPECT_EQ(i, (*seq_col)[i].get<TYPE_BIGINT>());
+ }
+}
+
+// Emits num_rows rows for a schema whose projection (col_ids) may be narrower
than the
+// full tablet schema. The filled block layout follows the projection,
matching how
+// VMergeIteratorContext::block_reset builds its block from the output schema.
+class ProjectedColumnsUtIterator : public RowwiseIterator {
+public:
+ ProjectedColumnsUtIterator(ReadSchema schema, size_t num_rows)
+ : _schema(std::move(schema)), _num_rows(num_rows) {}
+ ~ProjectedColumnsUtIterator() override = default;
+
+ Status init(const StorageReadOptions& opts) override { return
Status::OK(); }
+
+ Status next_batch(Block* block) override {
+ if (_rows_returned >= _num_rows) {
+ return Status::EndOfFile("End of ProjectedColumnsUtIterator");
+ }
+ while (_rows_returned < _num_rows) {
+ for (size_t j = 0; j < _schema.num_block_columns(); ++j) {
+ ColumnWithTypeAndName& vc = block->get_by_position(j);
+ IColumn& vi = (IColumn&)(*vc.column);
+
+ char data[16] = {};
+ size_t data_len = 0;
+ const auto* col_schema = _schema.column(j);
+ switch (col_schema->type()) {
+ case FieldType::OLAP_FIELD_TYPE_SMALLINT:
+ *(int16_t*)data = static_cast<int16_t>(_rows_returned);
+ data_len = sizeof(int16_t);
+ break;
+ case FieldType::OLAP_FIELD_TYPE_INT:
+ *(int32_t*)data = static_cast<int32_t>(_rows_returned);
+ data_len = sizeof(int32_t);
+ break;
+ case FieldType::OLAP_FIELD_TYPE_BIGINT:
+ *(int64_t*)data = static_cast<int64_t>(_rows_returned);
+ data_len = sizeof(int64_t);
+ break;
+ default:
+ break;
+ }
+
+ vi.insert_data(data, data_len);
+ }
+ ++_rows_returned;
+ }
+ return Status::OK();
+ }
+
+ const ReadSchema& schema() const override { return _schema; }
+
+private:
+ ReadSchema _schema;
+ size_t _num_rows;
+ size_t _rows_returned = 0;
+};
+
+// The merge-heap comparator compares the leading num_key_columns() block
positions when no
+// explicit compare columns are given. The read schema must therefore place
its key columns as
+// the leading prefix; if a non-key column precedes a key column, init() must
reject the merge
+// with a "compare contract violated" error instead of silently comparing a
value column as a key
+// (issue #66390: a ROW-binlog APPEND_ONLY scan projected value columns ahead
of a key).
+TEST(VGenericIteratorsTest, MergeRejectsProjectionMissingKeyPrefix) {
+ // Full tablet schema: k0(smallint), k1(int), v2(bigint). Project {v2,
k0}: the schema has
+ // one key column (k0) but it is not the leading column, so ordinal 0 is a
non-key column.
+ ReadSchema projected(
+ project_columns_by_ordinal(create_col_schemas(),
std::vector<ColumnId> {2, 0}));
+ auto output_schema = std::make_shared<ReadSchema>(projected);
+
+ std::vector<RowwiseIteratorUPtr> inputs;
+ inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected,
10));
+ inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected,
10));
+
+ auto iter = new_merge_iterator(std::move(inputs), -1, false, false,
nullptr, output_schema);
+ StorageReadOptions opts;
+ auto st = iter->init(opts);
+ EXPECT_FALSE(st.ok());
+ EXPECT_TRUE(st.to_string().find("compare contract violated") !=
std::string::npos)
+ << st.to_string();
+}
+
+// Same contract, a different non-leading-key layout: project {v2, k1}.
num_key_columns() is 1
+// (k1), but position 0 is the value column v2, so the merge comparator would
compare v2 as if it
+// were the leading key. init() must reject it.
+TEST(VGenericIteratorsTest, MergeRejectsProjectionWithoutLeadingKey) {
+ // Full tablet schema: k0(smallint), k1(int), v2(bigint); project {v2, k1}.
+ ReadSchema projected(
+ project_columns_by_ordinal(create_col_schemas(),
std::vector<ColumnId> {2, 1}));
+ auto output_schema = std::make_shared<ReadSchema>(projected);
+
+ std::vector<RowwiseIteratorUPtr> inputs;
+ inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected,
10));
+ inputs.push_back(std::make_unique<ProjectedColumnsUtIterator>(projected,
10));
+
+ auto iter = new_merge_iterator(std::move(inputs), -1, false, false,
nullptr, output_schema);
+ StorageReadOptions opts;
+ auto st = iter->init(opts);
+ EXPECT_FALSE(st.ok());
+ EXPECT_TRUE(st.to_string().find("compare contract violated") !=
std::string::npos)
+ << st.to_string();
+}
+
} // namespace doris
diff --git
a/be/test/storage/iterator/block_reader_binlog_vcollect_merge_test.cpp
b/be/test/storage/iterator/block_reader_binlog_vcollect_merge_test.cpp
new file mode 100644
index 00000000000..62cef9f8e34
--- /dev/null
+++ b/be/test/storage/iterator/block_reader_binlog_vcollect_merge_test.cpp
@@ -0,0 +1,402 @@
+// 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.
+
+// End-to-end coverage of the full BlockReader -> VCollectIterator path for a
+// row-binlog scan that reads TWO rowsets holding the SAME user key with
DIFFERENT
+// TSOs. Unlike block_reader_change_next_block_test.cpp (which injects a single
+// LevelIterator) and unlike a hand-built Level1Iterator, this test drives the
REAL
+// VCollectIterator::add_child + build_heap path:
+// * add_child() wraps each mock rowset reader in a real Level0Iterator, and
+// * build_heap() assembles the merge Level1Iterator that orders same-key
rows by
+// ascending TSO (binlog_tso_col_idx -> sequence column, small_seq_first =
true).
+//
+// The read schema is a real row-binlog schema: DUP_KEYS, which is what the FE
actually
+// generates for the hidden row-binlog table. With DUP_KEYS, build_heap
computes
+// _skip_same = false, so the merge keeps EVERY change event instead of
deduplicating by
+// user key. The test verifies that the two rowsets are merged into one
globally
+// (key, TSO)-ordered stream and that BlockReader folds it correctly:
+// * MIN_DELTA collapses each key's consecutive same-key changes into its
net change,
+// * DETAIL emits every change event verbatim in ascending-TSO order,
+// covering same-key/different-TSO events spread across two rowsets without
dropping any.
+
+#if defined(__clang__)
+#pragma clang diagnostic push
+#pragma clang diagnostic ignored "-Wkeyword-macro"
+#endif
+#include "storage/iterator/block_reader.h"
+#include "storage/iterator/vcollect_iterator.h"
+#if defined(__clang__)
+#pragma clang diagnostic pop
+#endif
+
+#include <gtest/gtest.h>
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/config.h"
+#include "common/status.h"
+#include "core/assert_cast.h"
+#include "core/block/block.h"
+#include "core/column/column_vector.h"
+#include "core/data_type/data_type_number.h"
+#include "io/fs/local_file_system.h"
+#include "storage/binlog.h"
+#include "storage/iterator/binlog_block_reader_utils.h"
+#include "storage/rowset/rowset.h"
+#include "storage/rowset/rowset_meta.h"
+#include "storage/rowset/rowset_reader.h"
+#include "storage/schema.h"
+#include "storage/tablet/tablet_schema.h"
+#include "storage/utils.h"
+
+namespace doris {
+
+using namespace ErrorCode;
+
+namespace {
+
+// Merged binlog block layout. The leading column is the primary key used both
for the
+// merge-heap key comparison and for MIN_DELTA group boundaries:
+// 0: key (Int64, the only key column)
+// 1: val (Int64, "after" value)
+// 2: __BEFORE__val__ (Int64, "before" value mirror)
+// 3: __DORIS_BINLOG_TSO__ (Int64, the merge sequence / order column)
+// 4: __DORIS_BINLOG_LSN__ (Int64)
+// 5: __DORIS_BINLOG_OP__ (Int64, ROW_BINLOG_APPEND/UPDATE/DELETE)
+constexpr int KEY_IDX = 0;
+constexpr int VAL_IDX = 1;
+constexpr int BEFORE_VAL_IDX = 2;
+constexpr int TSO_IDX = 3;
+constexpr int LSN_IDX = 4;
+constexpr int OP_IDX = 5;
+constexpr int NUM_COLS = 6;
+
+struct Row {
+ int64_t key;
+ int64_t val;
+ int64_t before_val;
+ int64_t tso;
+ int64_t lsn;
+ int64_t op;
+};
+
+// Row-binlog read schema: single leading key column, a value column plus its
__BEFORE__
+// mirror, and the three binlog meta columns. Marking BINLOG_TSO_COL by name
makes
+// TabletSchema::binlog_tso_col_idx() return its position, which
Level1Iterator::init()
+// uses to pick the TSO as the merge sequence column.
+std::shared_ptr<TabletSchema> make_binlog_schema() {
+ auto schema = std::make_shared<TabletSchema>();
+ const std::string col_names[] = {
+ "key", "val",
binlog::build_before_column_name("val"),
+ BINLOG_TSO_COL, BINLOG_LSN_COL, BINLOG_OP_COL};
+ for (int i = 0; i < NUM_COLS; ++i) {
+ TabletColumn col;
+ col.set_name(col_names[i]);
+ col.set_type(FieldType::OLAP_FIELD_TYPE_BIGINT);
+ col.set_unique_id(i);
+ col.set_is_key(i == KEY_IDX); // only the leading key column is a key
+ schema->append_column(std::move(col));
+ }
+ // DUP_KEYS matches the schema the FE generates for the hidden row-binlog
table; it makes
+ // build_heap compute _skip_same = false so the merge keeps every change
event.
+ schema->_keys_type = KeysType::DUP_KEYS;
+ return schema;
+}
+
+// Minimal RowsetMeta carrying only num_rows (used by build_heap to pick the
base rowset).
+class FakeRowsetMeta : public RowsetMeta {
+public:
+ FakeRowsetMeta() : RowsetMeta() { _fs = io::global_local_filesystem(); }
+ io::FileSystemSPtr fs() override { return _fs; }
+
+private:
+ io::FileSystemSPtr _fs;
+};
+
+// Minimal Rowset shell: no IO, just holds the meta so
rowset()->rowset_meta()->num_rows()
+// works and type()==BETA_ROWSET satisfies Level0Iterator's DCHECK.
+class FakeRowset : public Rowset {
+public:
+ FakeRowset(TabletSchemaSPtr schema, RowsetMetaSharedPtr meta)
+ : Rowset(schema, meta, "/fake/tablet/path") {}
+
+ Status create_reader(std::shared_ptr<RowsetReader>* result) override {
+ return Status::NotSupported("");
+ }
+ Status remove() override { return Status::OK(); }
+ Status link_files_to(const std::string&, RowsetId, size_t,
std::set<int64_t>*) override {
+ return Status::OK();
+ }
+ Status copy_files_to(const std::string&, const RowsetId&) override {
return Status::OK(); }
+ Status remove_old_files(std::vector<std::string>*) override { return
Status::OK(); }
+ Status check_file_exist() override { return Status::OK(); }
+ Status upload_to(const StorageResource&, const RowsetId&) override {
return Status::OK(); }
+ Status get_inverted_index_size(int64_t* index_size) override {
+ *index_size = 0;
+ return Status::OK();
+ }
+ void clear_inverted_index_cache() override {}
+ Status init() override { return Status::OK(); }
+ void do_close() override {}
+ Status check_current_rowset_segment() override { return Status::OK(); }
+ int64_t num_segments() const override { return 0; }
+ Result<std::string> segment_path(int64_t) override {
+ return ResultError(Status::InternalError(""));
+ }
+};
+
+// Fake per-rowset reader. next_batch(Block*) fills the caller-provided block
(built by
+// Level0Iterator from the read schema) with the preset rows on the first
call, then
+// reports END_OF_FILE with an empty block. This is the single-rowset stream a
real
+// BetaRowsetReader would hand a Level0Iterator, so add_child/build_heap run
for real.
+class FakeRowsetReader : public RowsetReader {
+public:
+ FakeRowsetReader(std::vector<Row> rows, int64_t version)
+ : _rows(std::move(rows)), _version(version) {
+ auto meta = std::make_shared<FakeRowsetMeta>();
+ meta->set_num_rows(static_cast<int64_t>(_rows.size()));
+ meta->set_rowset_type(BETA_ROWSET);
+ _rowset = std::make_shared<FakeRowset>(nullptr, meta);
+ _read_schema =
std::make_shared<ReadSchema>(make_binlog_schema()->columns());
+ }
+
+ Status init(RowsetReaderContext*, const RowSetSplits&) override { return
Status::OK(); }
+ Status get_segment_iterators(RowsetReaderContext*,
std::vector<RowwiseIteratorUPtr>*,
+ bool) override {
+ return Status::OK();
+ }
+ void reset_read_options() override {}
+
+ Status next_batch(Block* block) override {
+ if (_emitted) {
+ return Status::Error<END_OF_FILE>("");
+ }
+ _emitted = true;
+ // The block is created from the read schema (all BIGINT,
non-nullable) and
+ // cleared before each refresh, so append straight into its Int64
columns. The
+ // scoped guard writes the filled columns back into `block` on
destruction, so
+ // block->rows() reflects the appended rows (Level0Iterator relies on
that).
+ auto columns_guard = block->mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ auto col_of = [&](int idx) -> ColumnInt64& {
+ return assert_cast<ColumnInt64&>(*columns[idx]);
+ };
+ for (const auto& r : _rows) {
+ col_of(KEY_IDX).insert_value(r.key);
+ col_of(VAL_IDX).insert_value(r.val);
+ col_of(BEFORE_VAL_IDX).insert_value(r.before_val);
+ col_of(TSO_IDX).insert_value(r.tso);
+ col_of(LSN_IDX).insert_value(r.lsn);
+ col_of(OP_IDX).insert_value(r.op);
+ }
+ return Status::OK();
+ }
+ Status next_batch(BlockView*) override { return Status::NotSupported(""); }
+ Status next_batch(BlockWithSameBit*) override { return
Status::NotSupported(""); }
+
+ bool delete_flag() override { return false; }
+ Version version() override { return Version(_version, _version); }
+ RowsetSharedPtr rowset() override { return _rowset; }
+ const ReadSchema& read_schema() const override { return *_read_schema; }
+ int64_t filtered_rows() override { return 0; }
+ uint64_t merged_rows() override { return 0; }
+ RowsetTypePB type() const override { return BETA_ROWSET; }
+ int64_t newest_write_timestamp() override { return 0; }
+ void update_profile(RuntimeProfile*) override {}
+ RowsetReaderSharedPtr clone() override {
+ return std::make_shared<FakeRowsetReader>(_rows, _version);
+ }
+ void set_topn_limit(size_t) override {}
+
+private:
+ std::vector<Row> _rows;
+ int64_t _version;
+ bool _emitted = false;
+ RowsetSharedPtr _rowset;
+ ReadSchemaSPtr _read_schema;
+};
+
+// Wire a BlockReader as if init() had completed for a row-binlog
MIN_DELTA/DETAIL scan,
+// then drive the REAL VCollectIterator init + add_child + build_heap path
with one
+// child per rowset. Nothing here hardcodes _skip_same; build_heap computes it
from the
+// (DUP_KEYS) row-binlog read schema, which yields _skip_same = false so every
event is
+// kept.
+void configure_two_rowset_merge(BlockReader& reader, std::vector<Row>
rowset0_rows,
+ std::vector<Row> rowset1_rows, size_t
batch_size) {
+ config::enable_adaptive_batch_size = false;
+ reader._reader_context.batch_size = batch_size;
+ reader._reader_context.read_row_binlog = true;
+ reader._reader_context.read_orderby_key_columns = nullptr;
+ // Non-QUERY reader type keeps VCollectIterator::init() from dereferencing
the (unset)
+ // _tablet while deciding _merge; the row-binlog branch then forces the
merge anyway.
+ reader._reader_type = ReaderType::READER_BASE_COMPACTION;
+
+ reader._tablet_schema = make_binlog_schema();
+ // Identity read schema over the full row-binlog layout. ReadSchema
derives num_key_columns()
+ // from is_key() and tso_ordinal() from the BINLOG_TSO_COL name, which
Level1Iterator::init()
+ // uses to pick the TSO as the merge sequence column, and which
+ // _validate_merge_compare_contract() checks for the leading key prefix.
+ reader._read_schema =
std::make_shared<ReadSchema>(reader._tablet_schema->columns());
+
+ // Mirror BlockReader::_init_collect_iter: init the collect iterator
(force_merge=true,
+ // as a MIN_DELTA stream does), add one child per rowset, then build the
heap.
+ reader._vcollect_iter.init(&reader, /*ori_data_overlapping=*/true,
/*force_merge=*/true,
+ /*is_reverse=*/false);
+
+ std::vector<RowsetReaderSharedPtr> rs_readers;
+ // Give rowset0 more rows so build_heap picks a deterministic base rowset;
behavior is
+ // symmetric, this just fixes the base/cumulative split for the two-child
path.
+ auto reader0 = std::make_shared<FakeRowsetReader>(std::move(rowset0_rows),
/*version=*/2);
+ auto reader1 = std::make_shared<FakeRowsetReader>(std::move(rowset1_rows),
/*version=*/3);
+ for (auto& rs_reader : {reader0, reader1}) {
+ RowSetSplits split(rs_reader);
+ ASSERT_TRUE(reader._vcollect_iter.add_child(split).ok());
+ rs_readers.push_back(rs_reader);
+ }
+ ASSERT_TRUE(reader._vcollect_iter.build_heap(rs_readers).ok());
+
+ auto st = reader._vcollect_iter.current_row(&reader._next_row);
+ reader._eof = st.is<END_OF_FILE>();
+}
+
+Block make_output_block() {
+ Block block;
+ auto type = std::make_shared<DataTypeInt64>();
+ block.insert({ColumnInt64::create(), type, "key"});
+ block.insert({ColumnInt64::create(), type, "val"});
+ block.insert({ColumnInt64::create(), type,
binlog::build_before_column_name("val")});
+ block.insert({ColumnInt64::create(), type, BINLOG_TSO_COL});
+ block.insert({ColumnInt64::create(), type, BINLOG_LSN_COL});
+ block.insert({ColumnInt64::create(), type, BINLOG_OP_COL});
+ return block;
+}
+
+int64_t out_i64(const Block& block, int col, int row) {
+ return assert_cast<const
ColumnInt64&>(*block.get_by_position(col).column).get_element(row);
+}
+
+struct OutRow {
+ int64_t key;
+ int64_t val;
+ int64_t op;
+};
+
+std::vector<OutRow> drain(BlockReader& reader, Status
(BlockReader::*fn)(Block*, bool*)) {
+ std::vector<OutRow> result;
+ bool eof = false;
+ int guard = 0;
+ while (!eof) {
+ Block block = make_output_block();
+ Status st = (reader.*fn)(&block, &eof);
+ EXPECT_TRUE(st.ok()) << st;
+ for (size_t r = 0; r < block.rows(); ++r) {
+ result.push_back({out_i64(block, KEY_IDX, r), out_i64(block,
VAL_IDX, r),
+ out_i64(block, OP_IDX, r)});
+ }
+ if (++guard >= 1000) {
+ ADD_FAILURE() << "drain did not terminate";
+ break;
+ }
+ }
+ return result;
+}
+
+} // namespace
+
+class BlockReaderBinlogVCollectMergeTest : public testing::Test {
+protected:
+ void SetUp() override { _saved_adaptive =
config::enable_adaptive_batch_size; }
+ void TearDown() override { config::enable_adaptive_batch_size =
_saved_adaptive; }
+ bool _saved_adaptive = false;
+};
+
+// Two rowsets, same key, different TSO. rowset0 holds the APPEND (tso=1),
rowset1 holds
+// a later UPDATE (tso=2). With the DUP_KEYS row-binlog schema, build_heap
keeps every
+// same-key row: the merge orders them by ascending TSO into one key group,
and MIN_DELTA
+// folds APPEND+UPDATE into a single INSERT carrying the most recent value.
+TEST_F(BlockReaderBinlogVCollectMergeTest,
MinDeltaSameKeyAcrossRowsetsFoldsToInsert) {
+ BlockReader reader;
+ configure_two_rowset_merge(
+ reader,
+ {{/*key=*/1, /*val=*/10, /*before=*/0, /*tso=*/1, /*lsn=*/1,
ROW_BINLOG_APPEND}},
+ {{/*key=*/1, /*val=*/20, /*before=*/10, /*tso=*/2, /*lsn=*/2,
ROW_BINLOG_UPDATE}},
+ /*batch_size=*/16);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ ASSERT_EQ(out.size(), 1);
+ EXPECT_EQ(out[0].op, binlog::STREAM_CHANGE_INSERT);
+ EXPECT_EQ(out[0].key, 1);
+ EXPECT_EQ(out[0].val, 20); // most recent value, i.e. rowset1's later TSO
+}
+
+// Same key across two rowsets where the later rowset deletes it: APPEND
(tso=1) then
+// DELETE (tso=3) folds to SKIP, so the merged group produces no output. This
only holds
+// if both rows land in one key group; if the merge dropped one, the surviving
lone event
+// would fold to a stray INSERT or DELETE instead of cancelling out.
+TEST_F(BlockReaderBinlogVCollectMergeTest,
MinDeltaSameKeyAcrossRowsetsFoldsToSkip) {
+ BlockReader reader;
+ configure_two_rowset_merge(reader, {{5, 100, 0, 1, 1, ROW_BINLOG_APPEND}},
+ {{5, 100, 100, 3, 3, ROW_BINLOG_DELETE}}, 16);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ EXPECT_TRUE(out.empty());
+}
+
+// Interleaved keys across two rowsets. Each rowset is independently
key-ordered, and each
+// key's change chain is spread over both rowsets under different TSOs:
+// key 1: APPEND(tso=1, rowset0) + UPDATE(tso=3, rowset1) -> INSERT(val=15)
+// key 2: UPDATE(tso=2, rowset0) + DELETE(tso=4, rowset1) ->
DELETE(before=18)
+// The merge must produce the globally (key, TSO)-ordered stream and keep
every event so
+// MIN_DELTA groups each key across the rowset boundary.
+TEST_F(BlockReaderBinlogVCollectMergeTest,
MinDeltaInterleavedKeysAcrossRowsets) {
+ BlockReader reader;
+ configure_two_rowset_merge(
+ reader, {{1, 10, 0, 1, 1, ROW_BINLOG_APPEND}, {2, 20, 18, 2, 2,
ROW_BINLOG_UPDATE}},
+ {{1, 15, 10, 3, 3, ROW_BINLOG_UPDATE}, {2, 25, 20, 4, 4,
ROW_BINLOG_DELETE}}, 16);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ ASSERT_EQ(out.size(), 2);
+ EXPECT_EQ(out[0].key, 1);
+ EXPECT_EQ(out[0].op, binlog::STREAM_CHANGE_INSERT);
+ EXPECT_EQ(out[0].val, 15); // most recent value for key 1
+ EXPECT_EQ(out[1].key, 2);
+ EXPECT_EQ(out[1].op, binlog::STREAM_CHANGE_DELETE);
+ EXPECT_EQ(out[1].val, 18); // delete uses the first op's __BEFORE__ value
+}
+
+// DETAIL scan over the same two-rowset / same-key / different-TSO input emits
every event
+// verbatim in ascending-TSO order: the rowset0 APPEND becomes an INSERT and
the rowset1
+// UPDATE becomes a BEFORE + AFTER pair. A dedup in build_heap would drop the
APPEND (or
+// the UPDATE), leaving history incomplete.
+TEST_F(BlockReaderBinlogVCollectMergeTest,
DetailSameKeyAcrossRowsetsOrderedByTso) {
+ BlockReader reader;
+ configure_two_rowset_merge(reader, {{1, 10, 0, 1, 1, ROW_BINLOG_APPEND}},
+ {{1, 20, 10, 2, 2, ROW_BINLOG_UPDATE}}, 16);
+
+ auto out = drain(reader, &BlockReader::_detail_change_next_block);
+ ASSERT_EQ(out.size(), 3);
+ EXPECT_EQ(out[0].op, binlog::STREAM_CHANGE_INSERT);
+ EXPECT_EQ(out[0].val, 10); // rowset0 APPEND (tso=1) first
+ EXPECT_EQ(out[1].op, binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out[1].val, 10); // rowset1 UPDATE (tso=2) before value
+ EXPECT_EQ(out[2].op, binlog::STREAM_CHANGE_UPDATE_AFTER);
+ EXPECT_EQ(out[2].val, 20); // rowset1 UPDATE after value
+}
+
+} // namespace doris
diff --git a/be/test/storage/iterator/block_reader_min_delta_test.cpp
b/be/test/storage/iterator/block_reader_min_delta_test.cpp
index 54129b4190e..0c65b31cbb8 100644
--- a/be/test/storage/iterator/block_reader_min_delta_test.cpp
+++ b/be/test/storage/iterator/block_reader_min_delta_test.cpp
@@ -136,4 +136,104 @@ TEST_F(BlockReaderMinDeltaTest,
RowBinlogOperationCodeLayoutGuard) {
EXPECT_EQ(2, ROW_BINLOG_DELETE);
}
+// Contract test for the group-boundary rule in
BlockReader::_min_delta_next_block. A MIN_DELTA
+// group is a run of consecutive rows sharing the same user key; the reader
folds each group into
+// one net change via calculate_result(first_op, last_op). Exercising
_min_delta_next_block end to
+// end needs a fully constructed BlockReader + VCollectIterator + rowset
readers, which is far too
+// heavy for a unit test, so we pin the boundary rule itself.
+//
+// The subtle bug this guards against: the reader used to split groups on
IteratorRowRef::is_same,
+// which the segment merge sets only for CROSS-segment same-key matches (it
drives dedup). When a
+// compaction / row-binlog LMax quick-merge folds a key's whole change chain
into ONE segment, those
+// consecutive same-key rows carry is_same = false, so an is_same-based
grouping shatters one key
+// into many single-row groups. The fix groups by comparing the leading key
columns directly.
+namespace {
+struct QmRow {
+ int64_t key;
+ int64_t op;
+ bool is_same; // as produced by the segment merge (cross-segment dedup
marker)
+};
+
+// Old behavior: start a new group whenever is_same is false.
+std::vector<ResultType> fold_by_is_same(const std::vector<QmRow>& rows) {
+ std::vector<ResultType> out;
+ size_t i = 0;
+ while (i < rows.size()) {
+ int64_t first_op = rows[i].op;
+ int64_t last_op = rows[i].op;
+ size_t j = i + 1;
+ while (j < rows.size() && rows[j].is_same) {
+ last_op = rows[j].op;
+ ++j;
+ }
+
out.push_back(binlog::AggregateFunctionMinDelta::calculate_result(first_op,
last_op));
+ i = j;
+ }
+ return out;
+}
+
+// New behavior: start a new group when the user key changes (input is
globally key-ordered).
+std::vector<ResultType> fold_by_key(const std::vector<QmRow>& rows) {
+ std::vector<ResultType> out;
+ size_t i = 0;
+ while (i < rows.size()) {
+ int64_t first_op = rows[i].op;
+ int64_t last_op = rows[i].op;
+ size_t j = i + 1;
+ while (j < rows.size() && rows[j].key == rows[i].key) {
+ last_op = rows[j].op;
+ ++j;
+ }
+
out.push_back(binlog::AggregateFunctionMinDelta::calculate_result(first_op,
last_op));
+ i = j;
+ }
+ return out;
+}
+} // namespace
+
+TEST_F(BlockReaderMinDeltaTest, GroupBoundaryUsesKeyNotIsSame) {
+ // Three keys, each with a whole change chain folded into a single
quick-merge segment, so every
+ // row's is_same is false (no cross-segment match). Rows are globally
key-ordered by TSO.
+ // key 1: APPEND, UPDATE, UPDATE -> folds to INSERT
+ // key 2: UPDATE, UPDATE -> folds to UPDATE_BEFORE_AFTER
+ // key 3: APPEND, DELETE -> folds to SKIP
+ const std::vector<QmRow> rows = {
+ {1, ROW_BINLOG_APPEND, false}, {1, ROW_BINLOG_UPDATE, false},
+ {1, ROW_BINLOG_UPDATE, false}, {2, ROW_BINLOG_UPDATE, false},
+ {2, ROW_BINLOG_UPDATE, false}, {3, ROW_BINLOG_APPEND, false},
+ {3, ROW_BINLOG_DELETE, false},
+ };
+
+ // Grouping by key yields exactly one folded change per key.
+ const std::vector<ResultType> by_key = fold_by_key(rows);
+ ASSERT_EQ(3u, by_key.size());
+ EXPECT_EQ(ResultType::INSERT, by_key[0]);
+ EXPECT_EQ(ResultType::UPDATE_BEFORE_AFTER, by_key[1]);
+ EXPECT_EQ(ResultType::SKIP, by_key[2]);
+
+ // The old is_same-based grouping shatters each key into single-row
groups: 7 rows -> 7 groups,
+ // none of which reflect the true per-key net change (each APPEND/UPDATE
alone folds to INSERT/
+ // UPDATE, and the key-3 APPEND+DELETE that should cancel to SKIP is
instead two separate rows).
+ const std::vector<ResultType> by_is_same = fold_by_is_same(rows);
+ EXPECT_EQ(7u, by_is_same.size());
+ EXPECT_NE(by_key.size(), by_is_same.size());
+}
+
+TEST_F(BlockReaderMinDeltaTest, GroupBoundaryMixedIsSameStillGroupsByKey) {
+ // Realistic mix: some same-key rows are cross-segment (is_same=true),
others were folded into one
+ // segment (is_same=false). Grouping by key must be insensitive to how
is_same happened to be set.
+ // key 1: APPEND(false), UPDATE(true), UPDATE(false) -> INSERT
+ // key 2: UPDATE(false), DELETE(false) -> DELETE
+ const std::vector<QmRow> rows = {
+ {1, ROW_BINLOG_APPEND, false}, {1, ROW_BINLOG_UPDATE, true},
+ {1, ROW_BINLOG_UPDATE, false}, {2, ROW_BINLOG_UPDATE, false},
+ {2, ROW_BINLOG_DELETE, false},
+ };
+
+ const std::vector<ResultType> by_key = fold_by_key(rows);
+ ASSERT_EQ(2u, by_key.size());
+ EXPECT_EQ(ResultType::INSERT, by_key[0]);
+ EXPECT_EQ(ResultType::DELETE, by_key[1]);
+}
+
} // namespace doris
diff --git a/be/test/storage/row_binlog_vmerge_compaction_test.cpp
b/be/test/storage/row_binlog_vmerge_compaction_test.cpp
new file mode 100644
index 00000000000..62bdd43963d
--- /dev/null
+++ b/be/test/storage/row_binlog_vmerge_compaction_test.cpp
@@ -0,0 +1,400 @@
+// 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.
+
+// Producer-side coverage for horizontal (non-vertical) row-binlog compaction.
+//
+// With enable_vertical_compaction=false a row-binlog cumulative compaction
that takes the
+// real merge path runs Merger::vmerge_rowsets. Two things must hold for
downstream
+// MIN_DELTA reads to be correct:
+// 1. The merged output must be globally (key, TSO)-ordered, not a UNION of
the inputs.
+// A UNION would write [email protected]@TSO1 then [email protected]@TSO2, so
a key's
+// change chain is no longer consecutive.
+// 2. When a forced segment boundary splits a key range across output
segments (the same
+// user key lands in more than one segment), the output rowset meta must
be OVERLAPPING
+// so BetaRowsetReader::is_merge_iterator() picks a merge iterator.
+//
+// This test drives the real Merger::vmerge_rowsets on a row-binlog tablet
with two input
+// rowsets that share every key under different TSOs, forces a multi-segment
output via
+// RowsetWriterContext::max_rows_per_segment, and asserts both properties. A
negative case
+// on a plain (non-binlog) DUP_KEYS tablet proves the fix is gated on the
row-binlog path.
+
+#include <gen_cpp/olap_common.pb.h>
+#include <gen_cpp/olap_file.pb.h>
+#include <gtest/gtest.h>
+#include <unistd.h>
+
+#include <memory>
+#include <string>
+#include <tuple>
+#include <unordered_map>
+#include <vector>
+
+#include "common/status.h"
+#include "core/block/block.h"
+#include "io/fs/local_file_system.h"
+#include "runtime/exec_env.h"
+#include "storage/merger.h"
+#include "storage/olap_common.h"
+#include "storage/rowset/beta_rowset.h"
+#include "storage/rowset/rowset.h"
+#include "storage/rowset/rowset_factory.h"
+#include "storage/rowset/rowset_meta.h"
+#include "storage/rowset/rowset_reader.h"
+#include "storage/rowset/rowset_reader_context.h"
+#include "storage/rowset/rowset_writer.h"
+#include "storage/rowset/rowset_writer_context.h"
+#include "storage/schema.h"
+#include "storage/storage_engine.h"
+#include "storage/tablet/tablet.h"
+#include "storage/tablet/tablet_meta.h"
+#include "storage/tablet/tablet_schema.h"
+#include "storage/utils.h"
+#include "util/uid_util.h"
+
+namespace doris {
+using namespace ErrorCode;
+
+namespace {
+constexpr uint32_t kMaxPathLen = 1024;
+constexpr char kTestDir[] = "/row_binlog_vmerge_test";
+} // namespace
+
+class RowBinlogVmergeCompactionTest : public testing::Test {
+protected:
+ void SetUp() override {
+ char buffer[kMaxPathLen];
+ EXPECT_NE(getcwd(buffer, kMaxPathLen), nullptr);
+ _absolute_dir = std::string(buffer) + kTestDir;
+ auto st =
io::global_local_filesystem()->delete_directory(_absolute_dir);
+ ASSERT_TRUE(st.ok()) << st;
+ st = io::global_local_filesystem()->create_directory(_absolute_dir);
+ ASSERT_TRUE(st.ok()) << st;
+ EXPECT_TRUE(io::global_local_filesystem()
+ ->create_directory(_absolute_dir + "/tablet_path")
+ .ok());
+ doris::EngineOptions options;
+ auto engine = std::make_unique<StorageEngine>(options);
+ _engine = engine.get();
+ ExecEnv::GetInstance()->set_storage_engine(std::move(engine));
+ }
+
+ void TearDown() override {
+
EXPECT_TRUE(io::global_local_filesystem()->delete_directory(_absolute_dir).ok());
+ _engine = nullptr;
+ ExecEnv::GetInstance()->set_storage_engine(nullptr);
+ }
+
+ // Row-binlog read schema: single leading key column, a value column, and
the binlog TSO
+ // column (marked via binlog_tso_col_idx so Level1Iterator uses it as the
merge sequence
+ // column). DUP_KEYS matches the FE-generated hidden row-binlog table,
which keeps every
+ // change event instead of deduplicating by user key.
+ TabletSchemaSPtr create_row_binlog_schema() {
+ TabletSchemaPB pb;
+ pb.set_keys_type(DUP_KEYS);
+ pb.set_num_short_key_columns(1);
+ pb.set_num_rows_per_row_block(1024);
+ pb.set_compress_kind(COMPRESS_NONE);
+ pb.set_next_column_unique_id(4);
+
+ ColumnPB* key = pb.add_column();
+ key->set_unique_id(0);
+ key->set_name("k");
+ key->set_type("INT");
+ key->set_is_key(true);
+ key->set_length(4);
+ key->set_index_length(4);
+ key->set_is_nullable(false);
+
+ ColumnPB* val = pb.add_column();
+ val->set_unique_id(1);
+ val->set_name("v");
+ val->set_type("INT");
+ val->set_is_key(false);
+ val->set_length(4);
+ val->set_is_nullable(false);
+
+ // The binlog TSO column must be BIGINT;
SegmentIterator::_update_tso_col_if_needed
+ // asserts OLAP_FIELD_TYPE_BIGINT for it.
+ ColumnPB* tso = pb.add_column();
+ tso->set_unique_id(2);
+ tso->set_name(BINLOG_TSO_COL);
+ tso->set_type("BIGINT");
+ tso->set_is_key(false);
+ tso->set_length(8);
+ tso->set_is_nullable(false);
+ tso->set_visible(false);
+ pb.set_binlog_tso_col_idx(2);
+
+ auto schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(pb);
+ return schema;
+ }
+
+ TabletSchemaSPtr create_plain_dup_schema() {
+ TabletSchemaPB pb;
+ pb.set_keys_type(DUP_KEYS);
+ pb.set_num_short_key_columns(1);
+ pb.set_num_rows_per_row_block(1024);
+ pb.set_compress_kind(COMPRESS_NONE);
+ pb.set_next_column_unique_id(3);
+
+ ColumnPB* key = pb.add_column();
+ key->set_unique_id(0);
+ key->set_name("k");
+ key->set_type("INT");
+ key->set_is_key(true);
+ key->set_length(4);
+ key->set_index_length(4);
+ key->set_is_nullable(false);
+
+ ColumnPB* val = pb.add_column();
+ val->set_unique_id(1);
+ val->set_name("v");
+ val->set_type("INT");
+ val->set_is_key(false);
+ val->set_length(4);
+ val->set_is_nullable(false);
+
+ auto schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(pb);
+ return schema;
+ }
+
+ RowsetWriterContext create_rowset_writer_context(const TabletSchemaSPtr&
tablet_schema,
+ const SegmentsOverlapPB&
overlap,
+ uint32_t
max_rows_per_segment, Version version,
+ bool enable_binlog) {
+ static int64_t inc_id = 0;
+ RowsetWriterContext context;
+ RowsetId rowset_id;
+ rowset_id.init(++inc_id);
+ context.rowset_id = rowset_id;
+ context.rowset_type = BETA_ROWSET;
+ context.rowset_state = VISIBLE;
+ context.tablet_schema = tablet_schema;
+ context.tablet_path = _absolute_dir + "/tablet_path";
+ context.version = version;
+ context.segments_overlap = overlap;
+ context.max_rows_per_segment = max_rows_per_segment;
+ if (enable_binlog) {
+ context.write_binlog_opt().enable = true;
+ }
+ return context;
+ }
+
+ // Build a single-segment input rowset. Each (key, val, tso) tuple becomes
one row; every
+ // input rowset shares the same keys but a distinct tso so the two rowsets
overlap on keys.
+ RowsetSharedPtr create_input_rowset(const TabletSchemaSPtr& tablet_schema,
bool with_tso,
+ const std::vector<std::tuple<int, int,
int>>& rows,
+ int64_t version) {
+ auto context = create_rowset_writer_context(tablet_schema,
NONOVERLAPPING, UINT32_MAX,
+ {version, version},
/*enable_binlog=*/false);
+ auto res = RowsetFactory::create_rowset_writer(*_engine, context,
false);
+ EXPECT_TRUE(res.has_value()) << res.error();
+ auto writer = std::move(res).value();
+
+ Block block = tablet_schema->create_storage_block();
+ auto columns = std::move(block).mutate_columns();
+ for (const auto& [k, v, tso] : rows) {
+ columns[0]->insert_data((const char*)&k, sizeof(k));
+ columns[1]->insert_data((const char*)&v, sizeof(v));
+ if (with_tso) {
+ int64_t tso64 = tso;
+ columns[2]->insert_data((const char*)&tso64, sizeof(tso64));
+ }
+ }
+ block.set_columns(std::move(columns));
+ EXPECT_TRUE(writer->add_block(&block).ok());
+ EXPECT_TRUE(writer->flush().ok());
+
+ RowsetSharedPtr rowset;
+ EXPECT_EQ(Status::OK(), writer->build(rowset));
+ return rowset;
+ }
+
+ TabletSharedPtr create_tablet(const TabletSchema& tablet_schema, bool
row_binlog_role) {
+ std::vector<TColumn> cols;
+ std::unordered_map<uint32_t, uint32_t> col_ordinal_to_unique_id;
+ for (auto i = 0; i < tablet_schema.num_columns(); i++) {
+ const TabletColumn& column = tablet_schema.column(i);
+ TColumn col;
+ col.column_type.type = TPrimitiveType::INT;
+ col.__set_column_name(column.name());
+ col.__set_is_key(column.is_key());
+ cols.push_back(col);
+ col_ordinal_to_unique_id[i] = column.unique_id();
+ }
+
+ TTabletSchema t_tablet_schema;
+
t_tablet_schema.__set_short_key_column_count(tablet_schema.num_short_key_columns());
+ t_tablet_schema.__set_schema_hash(3333);
+ t_tablet_schema.__set_keys_type(TKeysType::DUP_KEYS);
+ t_tablet_schema.__set_storage_type(TStorageType::COLUMN);
+ t_tablet_schema.__set_columns(cols);
+ TabletMetaSharedPtr tablet_meta(new TabletMeta(
+ 1, 1, 1, 1, 1, 1, t_tablet_schema, 1,
col_ordinal_to_unique_id, UniqueId(1, 2),
+ TTabletType::TABLET_TYPE_DISK, TCompressionType::LZ4F, 0,
false));
+ if (row_binlog_role) {
+ tablet_meta->set_tablet_role(TabletRolePB::TABLET_ROLE_ROW_BINLOG);
+ }
+
+ TabletSharedPtr tablet(new Tablet(*_engine, tablet_meta, nullptr));
+ static_cast<void>(tablet->init());
+ return tablet;
+ }
+
+ // Read the output rowset in physical (segment) order.
need_ordered_result=false makes the
+ // reader concatenate segments as written, so the returned sequence
reflects the on-disk
+ // layout produced by the merge.
+ std::vector<std::tuple<int, int, int>> read_all(const RowsetSharedPtr&
rowset,
+ const TabletSchemaSPtr&
tablet_schema,
+ bool with_tso) {
+ std::vector<ColumnId> ordinals;
+ for (uint32_t i = 0; i < tablet_schema->num_columns(); ++i) {
+ ordinals.push_back(i);
+ }
+ auto read_schema = std::make_shared<ReadSchema>(
+ project_columns_by_ordinal(tablet_schema->columns(),
ordinals));
+
+ RowsetReaderContext reader_context;
+ reader_context.tablet_schema = tablet_schema;
+ reader_context.need_ordered_result = false;
+ reader_context.read_schema = read_schema;
+
+ RowsetReaderSharedPtr reader;
+ EXPECT_TRUE(rowset->create_reader(&reader).ok());
+ EXPECT_TRUE(reader->init(&reader_context).ok());
+
+ std::vector<std::tuple<int, int, int>> out;
+ Status s;
+ do {
+ Block block = read_schema->create_read_block();
+ s = reader->next_batch(&block);
+ auto columns = block.get_columns_with_type_and_name();
+ for (auto i = 0; i < block.rows(); i++) {
+ int tso = with_tso ?
static_cast<int>(columns[2].column->get_int(i)) : 0;
+
out.emplace_back(static_cast<int>(columns[0].column->get_int(i)),
+
static_cast<int>(columns[1].column->get_int(i)), tso);
+ }
+ } while (s.ok());
+ EXPECT_TRUE(s.is<END_OF_FILE>()) << s;
+ return out;
+ }
+
+ std::string _absolute_dir;
+ StorageEngine* _engine = nullptr;
+};
+
+// A real (key, TSO) merge of two row-binlog rowsets that share every key must
interleave the
+// events by key then TSO, and a forced segment boundary that splits a key
range must mark the
+// output OVERLAPPING.
+TEST_F(RowBinlogVmergeCompactionTest,
HorizontalMergeIsKeyTsoOrderedAndOverlapping) {
+ TabletSchemaSPtr schema = create_row_binlog_schema();
+ TabletSharedPtr tablet = create_tablet(*schema, /*row_binlog_role=*/true);
+ ASSERT_TRUE(tablet->is_row_binlog_tablet());
+
+ // rs0: keys 1..3 at tso=10; rs1: keys 1..3 at tso=20. The two rowsets
overlap on keys.
+ auto rs0 = create_input_rowset(schema, /*with_tso=*/true,
+ {{1, 100, 10}, {2, 200, 10}, {3, 300, 10}},
/*version=*/1);
+ auto rs1 = create_input_rowset(schema, /*with_tso=*/true,
+ {{1, 110, 20}, {2, 210, 20}, {3, 310, 20}},
/*version=*/2);
+
+ // Output default is NONOVERLAPPING (matches compaction.cpp for the
non-quick-merge path).
+ // With the (key, TSO) merge the physical order is
+ // (1,100),(1,110),(2,200),(2,210),(3,300),(3,310)
+ // so max_rows_per_segment=3 splits key 2 across two segments (seg0
max_key=2, seg1
+ // min_key=2), making the segments genuinely overlap.
+ auto ctx = create_rowset_writer_context(schema, NONOVERLAPPING,
/*max_rows_per_segment=*/3,
+ {0, rs1->end_version()},
/*enable_binlog=*/true);
+ auto res = RowsetFactory::create_rowset_writer(*_engine, ctx,
/*is_vertical=*/false);
+ ASSERT_TRUE(res.has_value()) << res.error();
+ auto writer = std::move(res).value();
+
+ std::vector<RowsetReaderSharedPtr> input_rs_readers;
+ for (auto& rowset : {rs0, rs1}) {
+ RowsetReaderSharedPtr rs_reader;
+ ASSERT_TRUE(rowset->create_reader(&rs_reader).ok());
+ input_rs_readers.push_back(std::move(rs_reader));
+ }
+
+ Merger::Statistics stats;
+ ASSERT_TRUE(Merger::vmerge_rowsets(tablet,
ReaderType::READER_CUMULATIVE_COMPACTION, *schema,
+ input_rs_readers, writer.get(), &stats)
+ .ok());
+ RowsetSharedPtr out_rowset;
+ ASSERT_EQ(Status::OK(), writer->build(out_rowset));
+
+ // Fix (B): a multi-segment output whose segments share a boundary key is
OVERLAPPING.
+ EXPECT_GT(out_rowset->rowset_meta()->num_segments(), 1);
+ EXPECT_EQ(OVERLAPPING, out_rowset->rowset_meta()->segments_overlap());
+ EXPECT_TRUE(out_rowset->rowset_meta()->is_segments_overlapping());
+
+ // Fix (A): the merged stream is globally (key, TSO)-ordered, so each
key's two events are
+ // consecutive with the earlier-TSO (lower version) row first. A buggy
UNION would instead
+ // produce (1,100),(2,200),(3,300),(1,110),(2,210),(3,310). The binlog TSO
column itself is
+ // rewritten to the rowset commit TSO on read (_update_tso_col_if_needed),
so we assert on
+ // (key, val) which carries the per-event identity.
+ auto rows = read_all(out_rowset, schema, /*with_tso=*/false);
+ std::vector<std::pair<int, int>> got;
+ for (const auto& [k, v, tso] : rows) {
+ got.emplace_back(k, v);
+ }
+ std::vector<std::pair<int, int>> expected = {{1, 100}, {1, 110}, {2, 200},
+ {2, 210}, {3, 300}, {3, 310}};
+ EXPECT_EQ(expected, got);
+}
+
+// Guard: the same overlapping-input + forced-boundary layout on a plain (non
row-binlog)
+// DUP_KEYS tablet must NOT be marked OVERLAPPING, proving both fixes are
gated on the
+// row-binlog path.
+TEST_F(RowBinlogVmergeCompactionTest, PlainDupMergeStaysNonOverlapping) {
+ TabletSchemaSPtr schema = create_plain_dup_schema();
+ TabletSharedPtr tablet = create_tablet(*schema, /*row_binlog_role=*/false);
+ ASSERT_FALSE(tablet->is_row_binlog_tablet());
+
+ auto rs0 =
+ create_input_rowset(schema, /*with_tso=*/false, {{1, 100, 0}, {2,
200, 0}, {3, 300, 0}},
+ /*version=*/1);
+ auto rs1 =
+ create_input_rowset(schema, /*with_tso=*/false, {{1, 110, 0}, {2,
210, 0}, {3, 310, 0}},
+ /*version=*/2);
+
+ auto ctx = create_rowset_writer_context(schema, NONOVERLAPPING,
/*max_rows_per_segment=*/2,
+ {0, rs1->end_version()},
/*enable_binlog=*/false);
+ auto res = RowsetFactory::create_rowset_writer(*_engine, ctx,
/*is_vertical=*/false);
+ ASSERT_TRUE(res.has_value()) << res.error();
+ auto writer = std::move(res).value();
+
+ std::vector<RowsetReaderSharedPtr> input_rs_readers;
+ for (auto& rowset : {rs0, rs1}) {
+ RowsetReaderSharedPtr rs_reader;
+ ASSERT_TRUE(rowset->create_reader(&rs_reader).ok());
+ input_rs_readers.push_back(std::move(rs_reader));
+ }
+
+ Merger::Statistics stats;
+ ASSERT_TRUE(Merger::vmerge_rowsets(tablet,
ReaderType::READER_CUMULATIVE_COMPACTION, *schema,
+ input_rs_readers, writer.get(), &stats)
+ .ok());
+ RowsetSharedPtr out_rowset;
+ ASSERT_EQ(Status::OK(), writer->build(out_rowset));
+
+ EXPECT_NE(OVERLAPPING, out_rowset->rowset_meta()->segments_overlap());
+ EXPECT_FALSE(out_rowset->rowset_meta()->is_segments_overlapping());
+}
+
+} // namespace doris
diff --git a/be/test/storage/rowset/rowset_meta_test.cpp
b/be/test/storage/rowset/rowset_meta_test.cpp
index 954b8e9ed22..f73ce1b9c3f 100644
--- a/be/test/storage/rowset/rowset_meta_test.cpp
+++ b/be/test/storage/rowset/rowset_meta_test.cpp
@@ -455,4 +455,58 @@ TEST_F(RowsetMetaTest,
TestSegmentsKeyBoundsAggregationTruncation) {
EXPECT_TRUE(rs_meta.is_segments_key_bounds_truncated());
}
+// is_segments_overlapping() is decided by segment count, the segments_overlap
flag,
+// and whether the rowset is a singleton delta (start_version == end_version)
or a
+// row-binlog rowset. For a singleton delta the ambiguous OVERLAP_UNKNOWN flag
is still
+// treated as overlapping (a freshly ingested delta may have overlapping
segments only
+// tagged OVERLAP_UNKNOWN). Row-binlog LMax quick merge produces non-singleton
rowsets
+// whose segments overlap, but it writes segments_overlap = OVERLAPPING
explicitly, so
+// it is recognized only via the explicit-OVERLAPPING branch; a row-binlog
rowset left
+// as OVERLAP_UNKNOWN is NOT treated as overlapping. A plain non-singleton
rowset (e.g. a
+// compaction output) keeps the original semantics and stays non-overlapping
regardless
+// of the flag, so that old metadata written before the flag was always set
does not
+// inflate compaction score / merge ways or degrade ordered reads after an
upgrade.
+TEST_F(RowsetMetaTest, TestIsSegmentsOverlapping) {
+ auto check = [](int64_t num_segments, SegmentsOverlapPB overlap, int64_t
start_version,
+ int64_t end_version, bool is_row_binlog, bool expected) {
+ RowsetMeta rs_meta;
+ rs_meta.set_num_segments(num_segments);
+ rs_meta.set_segments_overlap(overlap);
+ rs_meta.set_version({start_version, end_version});
+ if (is_row_binlog) {
+ rs_meta.mark_row_binlog();
+ }
+ EXPECT_EQ(rs_meta.is_segments_overlapping(), expected);
+ };
+
+ // Single segment is never overlapping, regardless of the flag.
+ check(1, OVERLAPPING, 2, 2, false, false);
+ check(1, OVERLAP_UNKNOWN, 2, 2, false, false);
+
+ // Multiple segments explicitly marked NONOVERLAPPING are not overlapping.
+ check(3, NONOVERLAPPING, 2, 2, false, false);
+ check(3, NONOVERLAPPING, 2, 5, false, false);
+
+ // Singleton delta (start == end) with overlapping / unknown segments is
overlapping:
+ // a freshly ingested delta may have overlapping segments only tagged
OVERLAP_UNKNOWN.
+ check(3, OVERLAPPING, 2, 2, false, true);
+ check(3, OVERLAP_UNKNOWN, 2, 2, false, true);
+
+ // Plain (non row-binlog) non-singleton + OVERLAPPING keeps the original
semantics: a
+ // compaction output (start < end) is treated as non-overlapping.
+ check(3, OVERLAPPING, 2, 5, false, false);
+
+ // Row-binlog non-singleton explicitly marked OVERLAPPING is overlapping:
this is the
+ // row-binlog LMax quick-merge output, which sets the flag explicitly.
+ check(3, OVERLAPPING, 2, 5, true, true);
+
+ // Row-binlog non-singleton left as OVERLAP_UNKNOWN must NOT be treated as
overlapping.
+ check(3, OVERLAP_UNKNOWN, 2, 5, true, false);
+
+ // Old-metadata compatibility: a plain compaction output (start < end)
left as
+ // OVERLAP_UNKNOWN must stay non-overlapping to avoid inflating
+ // get_compaction_score()/get_merge_way_num() and degrading ordered reads
after upgrade.
+ check(3, OVERLAP_UNKNOWN, 2, 5, false, false);
+}
+
} // namespace doris
diff --git a/be/test/storage/tablet_reader_test.cpp
b/be/test/storage/tablet_reader_test.cpp
index 1c8f1e7712b..0978308c458 100644
--- a/be/test/storage/tablet_reader_test.cpp
+++ b/be/test/storage/tablet_reader_test.cpp
@@ -36,9 +36,10 @@ namespace doris {
class TabletReaderTest : public testing::Test {
protected:
static TabletSchemaSPtr create_schema(
- const std::vector<std::pair<std::string, int32_t>>& name_and_uid) {
+ const std::vector<std::pair<std::string, int32_t>>& name_and_uid,
+ KeysType keys_type = KeysType::DUP_KEYS) {
TabletSchemaPB schema_pb;
- schema_pb.set_keys_type(KeysType::DUP_KEYS);
+ schema_pb.set_keys_type(keys_type);
bool first = true;
for (const auto& [name, uid] : name_and_uid) {
auto* col = schema_pb.add_column();
@@ -134,5 +135,4 @@ TEST_F(TabletReaderTest,
remove_delete_columns_keeps_unrelated_paths) {
EXPECT_EQ(size_t(2), access_paths.size());
}
-
} // namespace doris
diff --git
a/regression-test/data/table_stream_p0/test_table_stream_multi_segment_mow.out
b/regression-test/data/table_stream_p0/test_table_stream_multi_segment_mow.out
new file mode 100644
index 00000000000..7c7c051fd91
--- /dev/null
+++
b/regression-test/data/table_stream_p0/test_table_stream_multi_segment_mow.out
@@ -0,0 +1,79 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !mow_count --
+10000
+
+-- !mow_change_types --
+APPEND 10000
+
+-- !mow_sample --
+1 1 APPEND
+10000 10000 APPEND
+5000 5000 APPEND
+5001 5001 APPEND
+
+-- !mow_ov_count --
+7500
+
+-- !mow_ov_change_types --
+APPEND 7500
+
+-- !mow_ov_sample --
+1 1 APPEND
+2500 2500 APPEND
+2501 2501 APPEND
+5000 5000 APPEND
+5001 105001 APPEND
+7500 107500 APPEND
+
+-- !mow_md_count --
+10000
+
+-- !mow_md_change_types --
+APPEND 10000
+
+-- !mow_md_sample --
+1 1 APPEND
+10000 10000 APPEND
+5000 5000 APPEND
+5001 5001 APPEND
+
+-- !mow_md_ov_count --
+7500
+
+-- !mow_md_ov_change_types --
+APPEND 7500
+
+-- !mow_md_ov_sample --
+1 1 APPEND
+2500 2500 APPEND
+2501 102501 APPEND
+5000 105000 APPEND
+5001 105001 APPEND
+7500 107500 APPEND
+
+-- !mow_qm_binlog --
+0 1 10 \N
+0 2 20 \N
+0 3 30 \N
+1 1 11 10
+1 1 12 11
+1 1 13 12
+1 1 14 13
+1 2 21 20
+1 2 22 21
+1 2 23 22
+1 2 24 23
+1 3 31 30
+1 3 32 31
+
+-- !mow_qm_count --
+3
+
+-- !mow_qm_change_types --
+APPEND 3
+
+-- !mow_qm_sample --
+1 14 APPEND
+2 24 APPEND
+3 32 APPEND
+
diff --git
a/regression-test/suites/table_stream_p0/test_table_stream_multi_segment_mow.groovy
b/regression-test/suites/table_stream_p0/test_table_stream_multi_segment_mow.groovy
new file mode 100644
index 00000000000..ad749e8aaba
--- /dev/null
+++
b/regression-test/suites/table_stream_p0/test_table_stream_multi_segment_mow.groovy
@@ -0,0 +1,538 @@
+// 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_table_stream_multi_segment_mow", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+
+ sql "DROP DATABASE IF EXISTS test_table_stream_multi_segment_mow_db"
+ sql "CREATE DATABASE test_table_stream_multi_segment_mow_db"
+ sql "USE test_table_stream_multi_segment_mow_db"
+
+ def customBeConfig = [
+ doris_scanner_row_bytes: 1,
+ // Section E drives row-binlog LMax quick-merge. Tiny thresholds let
every small
+ // rowset count as "compaction enough" so the cumulative point
advances past two
+ // LMax rowsets, and wait_timesec=0 stops freshly-visible singleton
deltas from
+ // being filtered out of the compaction candidate set. time_threshold
is set huge
+ // so the time-based trigger never preempts the quick-merge branch.
+ binlog_compaction_goal_size_mbytes: 0,
+ binlog_compaction_file_count_threshold: 2,
+ binlog_compaction_wait_timesec_after_visible: 0,
+ binlog_compaction_time_threshold_seconds: 86400
+ ]
+
+ setBeConfigTemporary(customBeConfig) {
+ try {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ // shrink doris_scanner_row_bytes and enable MemTable.need_flush
to make
+ // each 5000-row batch produce multiple segments stably.
+ GetDebugPoint().enableDebugPointForAllBEs("MemTable.need_flush")
+
+ String batch1 = ""
+ String batch2 = ""
+ (1..5000).each { batch1 += "${it},${it}\n" }
+ (5001..10000).each { batch2 += "${it},${it}\n" }
+
+
+ // ========================================================
+ // Section A. MoW + append_only.
+ // ========================================================
+ sql "DROP STREAM IF EXISTS ts_ms_mow_stream"
+ sql "DROP TABLE IF EXISTS ts_ms_mow_base FORCE"
+ sql """
+ CREATE TABLE ts_ms_mow_base (
+ id BIGINT,
+ v INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ts_ms_mow_stream
+ ON TABLE ts_ms_mow_base
+ PROPERTIES (
+ "type" = "append_only",
+ "show_initial_rows" = "false"
+ )
+ """
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(batch1.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(batch2.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ sleep(1200)
+
+ order_qt_mow_count "SELECT COUNT(*) FROM ts_ms_mow_stream"
+ order_qt_mow_change_types """
+ SELECT __DORIS_STREAM_CHANGE_TYPE_COL__, COUNT(*)
+ FROM ts_ms_mow_stream
+ GROUP BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ ORDER BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ order_qt_mow_sample """
+ SELECT id, v, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ts_ms_mow_stream
+ WHERE id IN (1, 5000, 5001, 10000)
+ ORDER BY id
+ """
+
+ // ========================================================
+ // Section B. MoW + append_only, overlapping batches.
+ // updates on existing keys are filtered, only brand-new keys
append.
+ // batch1: id [1, 5000] -> 5000 new keys, all APPEND
+ // batch2: id [2501, 7500] -> [2501,5000] updates filtered,
+ // [5001,7500] 2500 new keys
APPEND
+ // expect: 7500 APPEND rows.
+ // ========================================================
+ String overlapBatch1 = ""
+ String overlapBatch2 = ""
+ (1..5000).each { overlapBatch1 += "${it},${it}\n" }
+ (2501..7500).each { overlapBatch2 += "${it},${it + 100000}\n" }
+
+ sql "DROP STREAM IF EXISTS ts_ms_mow_ov_stream"
+ sql "DROP TABLE IF EXISTS ts_ms_mow_ov_base FORCE"
+ sql """
+ CREATE TABLE ts_ms_mow_ov_base (
+ id BIGINT,
+ v INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ts_ms_mow_ov_stream
+ ON TABLE ts_ms_mow_ov_base
+ PROPERTIES (
+ "type" = "append_only",
+ "show_initial_rows" = "false"
+ )
+ """
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_ov_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(overlapBatch1.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_ov_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(overlapBatch2.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ sleep(1200)
+
+ qt_mow_ov_count "SELECT COUNT(*) FROM ts_ms_mow_ov_stream"
+ order_qt_mow_ov_change_types """
+ SELECT __DORIS_STREAM_CHANGE_TYPE_COL__, COUNT(*)
+ FROM ts_ms_mow_ov_stream
+ GROUP BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ ORDER BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ order_qt_mow_ov_sample """
+ SELECT id, v, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ts_ms_mow_ov_stream
+ WHERE id IN (1, 2500, 2501, 5000, 5001, 7500)
+ ORDER BY id
+ """
+
+ // ========================================================
+ // Section C. MoW + min_delta, non-overlapping batches.
+ // batch1: id [1, 5000] -> 5000 brand-new keys
+ // batch2: id [5001, 10000] -> 5000 brand-new keys
+ // each key inserted once, nothing to fold -> 10000 APPEND
rows.
+ // ========================================================
+ sql "DROP STREAM IF EXISTS ts_ms_mow_md_stream"
+ sql "DROP TABLE IF EXISTS ts_ms_mow_md_base FORCE"
+ sql """
+ CREATE TABLE ts_ms_mow_md_base (
+ id BIGINT,
+ v INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ts_ms_mow_md_stream
+ ON TABLE ts_ms_mow_md_base
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_md_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(batch1.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_md_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(batch2.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ sleep(1200)
+
+ qt_mow_md_count "SELECT COUNT(*) FROM ts_ms_mow_md_stream"
+ order_qt_mow_md_change_types """
+ SELECT __DORIS_STREAM_CHANGE_TYPE_COL__, COUNT(*)
+ FROM ts_ms_mow_md_stream
+ GROUP BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ ORDER BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ order_qt_mow_md_sample """
+ SELECT id, v, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ts_ms_mow_md_stream
+ WHERE id IN (1, 5000, 5001, 10000)
+ ORDER BY id
+ """
+
+ // ========================================================
+ // Section D. MoW + min_delta, overlapping batches.
+ // Combines Section B's overlapping layout with min_delta
folding.
+ // The stream is created on an empty table
(show_initial_rows=false),
+ // so there is no historical baseline to act as UPDATE_BEFORE.
Every
+ // key's net change relative to the stream start is therefore a
single
+ // APPEND carrying the latest value; an insert followed by an
update
+ // folds to one APPEND(new value), NOT an
UPDATE_BEFORE/UPDATE_AFTER
+ // pair. The point is that min_delta must keep every net-changed
key
+ // without dropping events across segments.
+ // batch1: id [1, 5000] v = id
+ // batch2: id [2501, 7500] v = id + 100000
+ // expect: 7500 APPEND rows, overlapping keys carrying the
latest value.
+ // [1, 2500] inserted once -> APPEND(id)
+ // [2501, 5000] insert then update -> APPEND(id +
100000)
+ // [5001, 7500] inserted once -> APPEND(id + 100000)
+ // ========================================================
+ sql "DROP STREAM IF EXISTS ts_ms_mow_md_ov_stream"
+ sql "DROP TABLE IF EXISTS ts_ms_mow_md_ov_base FORCE"
+ sql """
+ CREATE TABLE ts_ms_mow_md_ov_base (
+ id BIGINT,
+ v INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ts_ms_mow_md_ov_stream
+ ON TABLE ts_ms_mow_md_ov_base
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_md_ov_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(overlapBatch1.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ streamLoad {
+ db "test_table_stream_multi_segment_mow_db"
+ table "ts_ms_mow_md_ov_base"
+ set 'column_separator', ','
+ set 'columns', 'id,v'
+ inputStream new ByteArrayInputStream(overlapBatch2.getBytes())
+ time 60000
+ check { result, exception, startTime, endTime ->
+ if (exception != null) { throw exception }
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ sql "sync"
+ sleep(1200)
+
+ qt_mow_md_ov_count "SELECT COUNT(*) FROM ts_ms_mow_md_ov_stream"
+ order_qt_mow_md_ov_change_types """
+ SELECT __DORIS_STREAM_CHANGE_TYPE_COL__, COUNT(*)
+ FROM ts_ms_mow_md_ov_stream
+ GROUP BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ ORDER BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ order_qt_mow_md_ov_sample """
+ SELECT id, v, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ts_ms_mow_md_ov_stream
+ WHERE id IN (1, 2500, 2501, 5000, 5001, 7500)
+ ORDER BY id
+ """
+
+ // ========================================================
+ // Section E. MoW + row-binlog LMax quick-merge.
+ // Sections A-D produce a single multi-segment rowset per load.
This section
+ // targets the OTHER multi-segment source: binlog cumulative
compaction's LMax
+ // quick-merge, which LINKS segments from several row-binlog
rowsets into one
+ // non-singleton OVERLAPPING output rowset (compaction.cpp sets
segments_overlap
+ // = OVERLAPPING for the quick-merge path). Because the same
user key recurs in
+ // both merged LMax rowsets, after the link it lands in
different segments of the
+ // one output rowset, each carrying a distinct binlog TSO.
Reading it must NOT
+ // deduplicate by user key (is_unique=false on the row-binlog
path); a UNIQUE
+ // dedup would drop same-key/different-TSO events. This
reproduces that layout and
+ // asserts every event survives.
+ //
+ // The BE binlog thresholds are lowered in customBeConfig so
every small rowset is
+ // "compaction enough" and the cumulative point advances past
two LMax rowsets.
+ // trigger_and_wait_compaction on the base table also drives the
co-located hidden
+ // row-binlog tablet (SHOW TABLETS lists it; its
compaction_policy="binlog"), so the
+ // cumulative trigger runs BinlogCumulativeCompactionPolicy on
it. The load/trigger
+ // cadence mirrors test_binlog_compaction.groovy case2's
Round1..Round7.
+ // ========================================================
+ sql "DROP STREAM IF EXISTS ts_ms_mow_qm_stream"
+ sql "DROP TABLE IF EXISTS ts_ms_mow_qm_base FORCE"
+ sql """
+ CREATE TABLE ts_ms_mow_qm_base (
+ id BIGINT,
+ v INT
+ ) ENGINE=OLAP
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "disable_auto_compaction" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """
+ CREATE STREAM ts_ms_mow_qm_stream
+ ON TABLE ts_ms_mow_qm_base
+ PROPERTIES (
+ "type" = "min_delta",
+ "show_initial_rows" = "false"
+ )
+ """
+
+ // Build the FIRST LMax rowset [0-6]. Each INSERT is one binlog
version.
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (1, 10), (2, 20)" // v1
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (1, 11)" // v2
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (2, 21)" // v3
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (3, 30)" // v4
+ // Round 1: Level0 [0-1],[2-2],[3-3],[4-4] -> Level1 [0-4].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (1, 12)" // v5
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (2, 22)" // v6
+ // Round 2: Level0 [5-5],[6-6] -> Level1 [5-6].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+ // Round 3: Level1 [0-4],[5-6] -> Level2 (LMax) [0-6].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+
+ // Build the SECOND LMax rowset [7-12], reusing keys 1/2/3 so they
recur across
+ // both LMax rowsets (distinct TSOs), producing same-key rows in
different segments
+ // after the quick-merge link.
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (1, 13)" // v7
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (2, 23)" // v8
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (3, 31)" // v9
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (1, 14)" // v10
+ // Round 4: Level0 [7-7],[8-8],[9-9],[10-10] -> Level1 [7-10].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (2, 24)" // v11
+ sql "INSERT INTO ts_ms_mow_qm_base VALUES (3, 32)" // v12
+ // Round 5: Level0 [11-11],[12-12] -> Level1 [11-12].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+ // Round 6: Level1 [7-10],[11-12] -> Level2 (LMax) [7-12].
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+
+ // Round 7: Level2 {[0-6],[7-12]} are both before the cumulative
point and both
+ // compaction-enough (compact_enough_size=2>1) -> LMax quick-merge
-> [0-12] with
+ // segments_overlap=OVERLAPPING. This is the non-singleton
overlapping rowset.
+ trigger_and_wait_compaction("ts_ms_mow_qm_base", "cumulative")
+ sql "sync"
+ sleep(1200)
+
+ // Guard: assert the quick-merge actually produced the target
rowset before querying.
+ // trigger_and_wait_compaction drives every tablet of the table
via compact_type=
+ // cumulative, including the co-located hidden row-binlog tablet
(IsRowBinlog=true,
+ // compaction_policy="binlog"), which BE routes to
BinlogCumulativeCompactionPolicy +
+ // LMax quick-merge. /api/compaction/show returns each rowset as
+ // "[start-end] <num_segments> DATA <OVERLAP> <rowset_id> <size>
level=<n>"
+ // (Rowset::get_rowset_info_str). The quick-merge signature
(compaction.cpp: output
+ // version = [front.start, back.end] with start_version==0 the
trigger precondition,
+ // and segments_overlap forced to OVERLAPPING) is: on the
row-binlog tablet, a rowset
+ // that starts at version 0, is OVERLAPPING, and links more than
one segment. We match
+ // that rather than a hardcoded end version, since the row-binlog
tablet's version
+ // range depends on the initial [0-1] rowset (here the merged
rowset is [0-13]).
+ // Without this guard, a compaction that never hit the quick-merge
branch would
+ // silently fall back to reading the original rowsets and the
section could pass
+ // without exercising the intended path.
+ def qmBackendIdToIp = [:]
+ def qmBackendIdToHttpPort = [:]
+ getBackendIpHttpPort(qmBackendIdToIp, qmBackendIdToHttpPort)
+ def qmTablets = sql_return_maparray "show tablets from
ts_ms_mow_qm_base"
+ def foundQuickMergeRowset = false
+ for (qmTablet in qmTablets) {
+ // The quick-merge OVERLAPPING rowset only lives on the hidden
row-binlog tablet.
+ if (qmTablet.IsRowBinlog != "true") {
+ continue
+ }
+ def qmBeHost = qmBackendIdToIp["${qmTablet.BackendId}"]
+ def qmBePort = qmBackendIdToHttpPort["${qmTablet.BackendId}"]
+ def (qmExitCode, qmStdout, qmStderr) =
+ be_show_tablet_status(qmBeHost, qmBePort,
qmTablet.TabletId)
+ assert qmExitCode == 0 : "show tablet status failed:
${qmStderr}"
+ def qmStatus = parseJson(qmStdout.trim())
+ logger.info("row-binlog tablet ${qmTablet.TabletId} policy=" +
+ "${qmStatus['cumulative policy type']}
rowsets=${qmStatus.rowsets}")
+ for (rowsetStr in qmStatus.rowsets) {
+ // e.g. "[0-13] 2 DATA OVERLAPPING <id> 2.80 KB level=0"
+ def fields = rowsetStr.trim().split(/\s+/)
+ if (fields.length >= 4 && fields[0].startsWith("[0-") &&
+ fields[1].isInteger() && (fields[1] as int) > 1 &&
+ fields[3] == "OVERLAPPING") {
+ foundQuickMergeRowset = true
+ logger.info("found LMax quick-merge rowset on
row-binlog tablet " +
+ "${qmTablet.TabletId}: ${rowsetStr}")
+ }
+ }
+ }
+ assert foundQuickMergeRowset :
+ "expected a non-singleton multi-segment OVERLAPPING
row-binlog rowset starting " +
+ "at version 0 from LMax quick-merge, but none was found;
row-binlog quick-merge " +
+ "did not run and Section E would read the original rowsets
instead"
+
+ // Ground truth for "no dropped events": the raw row-binlog op
stream. This reads
+ // the quick-merge OVERLAPPING rowset through the same row-binlog
TabletReader path
+ // (is_unique=false); a UNIQUE dedup would silently drop
same-key/different-TSO rows,
+ // shrinking this result.
+ order_qt_mow_qm_binlog """
+ SELECT __DORIS_BINLOG_OP__ AS op,
+ id,
+ v,
+ __BEFORE__v__
+ FROM binlog("table" = "ts_ms_mow_qm_base")
+ ORDER BY id, v, op
+ """
+ order_qt_mow_qm_count "SELECT COUNT(*) FROM ts_ms_mow_qm_stream"
+ order_qt_mow_qm_change_types """
+ SELECT __DORIS_STREAM_CHANGE_TYPE_COL__, COUNT(*)
+ FROM ts_ms_mow_qm_stream
+ GROUP BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ ORDER BY __DORIS_STREAM_CHANGE_TYPE_COL__
+ """
+ order_qt_mow_qm_sample """
+ SELECT id, v, __DORIS_STREAM_CHANGE_TYPE_COL__
+ FROM ts_ms_mow_qm_stream
+ ORDER BY id, v
+ """
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ }
+ }
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
