This is an automated email from the ASF dual-hosted git repository.
morningman 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 86c78b92b1c [fix](binlog) Fix BE crash when APPEND_ONLY row-binlog
scan omits key columns (#66432)
86c78b92b1c is described below
commit 86c78b92b1c691b0720878059208ad5d9e678f36
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Aug 6 12:47:38 2026 +0800
[fix](binlog) Fix BE crash when APPEND_ONLY row-binlog scan omits key
columns (#66432)
### What problem does this PR solve?
Issue Number: close #66390
Related PR: #63850
Related Issue: #65418
Problem Summary:
A BE received SIGSEGV in `VMergeIteratorContext::compare()` while
`VMergeIterator::init()` was building its merge heap during a ROW-binlog
regression workload.
Root cause chain:
1. Any row-binlog scan (`binlog_scan_type != NONE`, including
`APPEND_ONLY`) forced a key-ordered merged read with
`read_orderby_key_num_prefix_columns == 0`.
2. With a zero prefix, `read_orderby_key_columns` stays null, so both
merge comparators (`VMergeIteratorContext::compare()` at the rowset
level and `VCollectIterator::LevelIteratorComparator` at the rowset-tree
level) fall back to comparing the first `num_key_columns` block
positions. `Schema::num_key_columns()` counts the key columns of the
**whole tablet schema**, independent of the projection, so the fallback
silently assumes the projection starts with the full ordered key prefix.
3. `MIN_DELTA` / `DETAIL` satisfy that contract because they widen the
storage projection with every key column. `APPEND_ONLY` takes the
`direct_mode` branch (`return_columns = SQL projection`), which may omit
some or all key columns (e.g. `SELECT v1 FROM t@incr(... "incrementType"
= "APPEND_ONLY")` on a table with two key columns).
4. `Block::compare_at()` has only DCHECK bounds checks, so in a release
build the first heap comparison reads past the block's column array and
kills the BE. This is reachable whenever a rowset has more than one
segment (`force_key_ordered_read` makes `is_merge_iterator()` true even
for non-overlapping segments) or rowsets overlap. A projection that has
enough columns but not the leading keys would instead be compared on the
wrong columns — a silent misordering.
Fix (three layers):
1. **Root cause (`olap_scanner.cpp`)**: only `MIN_DELTA` / `DETAIL` —
the modes that actually group by key and reconstruct BEFORE/AFTER rows,
and whose projection is widened with the full key prefix — force the
key-ordered merged read. `APPEND_ONLY` does no key grouping (it is a
plain op-filter + TSO-range stream), so it now reads unordered like a
plain scan, which also removes an unnecessary merge-heap cost from the
highest-throughput mode. ORDER BY / TopN pushdown params stay disabled
for all binlog scan types, as before.
2. **Defense in `VMergeIteratorContext::init()`**: validate the compare
contract once the first block is loaded — explicit compare columns must
point inside the block; the default key-prefix comparison requires the
projection to start with exactly the schema's key column ids in order;
the sequence tie-break position must be in range. Violations return an
`InternalError` carrying tablet/rowset/version/projection details
instead of an out-of-bounds read.
3. **Same defense in `VCollectIterator::Level1Iterator::init()`**:
validate every child's first block against the same contract before
anything is pushed into the merge heap.
---
be/src/exec/scan/olap_scanner.cpp | 33 ++++--
be/src/storage/iterator/vcollect_iterator.cpp | 57 +++++++++
be/src/storage/iterator/vcollect_iterator.h | 7 ++
be/src/storage/iterator/vgeneric_iterators.cpp | 53 +++++++++
be/src/storage/iterator/vgeneric_iterators.h | 6 +
be/test/exec/scan/vgeneric_iterators_test.cpp | 103 +++++++++++++++-
.../test_binlog_append_only_projection.groovy | 132 +++++++++++++++++++++
7 files changed, 379 insertions(+), 12 deletions(-)
diff --git a/be/src/exec/scan/olap_scanner.cpp
b/be/src/exec/scan/olap_scanner.cpp
index fee0f1b6199..14c5c02b9b5 100644
--- a/be/src/exec/scan/olap_scanner.cpp
+++ b/be/src/exec/scan/olap_scanner.cpp
@@ -476,13 +476,14 @@ Status OlapScanner::_init_tablet_reader_params(
}
};
- // For row-binlog scans that emit BEFORE/AFTER pairs (MIN_DELTA / DETAIL),
we must read
- // every key column, every requested value column, the binlog meta columns
(tso / op)
- // and their __BEFORE__ mirrors, so the BlockReader can reconstruct change
rows.
- const bool need_before_columns =
+ // MIN_DELTA / DETAIL row-binlog scans reconstruct change rows in
BlockReader through a
+ // key-ordered merge. They must read every key column, every requested
value column, the
+ // binlog meta columns (tso / op) and their __BEFORE__ mirrors.
APPEND_ONLY streams rows
+ // as-is and stays on the plain projection paths below.
+ const bool is_binlog_merge_scan =
_tablet_reader_params.binlog_scan_type ==
TBinlogScanType::MIN_DELTA ||
_tablet_reader_params.binlog_scan_type == TBinlogScanType::DETAIL;
- if (need_before_columns) {
+ if (is_binlog_merge_scan) {
for (size_t i = 0; i < tablet_schema->num_key_columns(); ++i) {
add_return_column_if_absent(static_cast<uint32_t>(i));
}
@@ -565,16 +566,26 @@ Status OlapScanner::_init_tablet_reader_params(
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
- // BEFORE/AFTER pairs in deterministic order (DETAIL). Disable ORDER BY /
TopN pushdowns
- // and reset their related params, since they would otherwise re-order the
stream.
+ // Row-binlog scans must not be re-ordered or truncated by ORDER BY / TopN
pushdowns,
+ // so reset every reorder-related param for all binlog scan types.
+ //
+ // Only MIN_DELTA / DETAIL additionally 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 BEFORE/AFTER pairs in deterministic order (DETAIL).
Their storage
+ // projection is widened above with the full key prefix, which the
key-ordered merge
+ // comparator relies on: with read_orderby_key_num_prefix_columns == 0 the
comparator
+ // falls back to comparing the first num_key_columns block positions.
+ //
+ // APPEND_ONLY does no key grouping and keeps the raw SQL projection,
which may omit
+ // some or even all key columns. Forcing a key-ordered merge would make
the fallback
+ // comparator read key positions that do not exist in the projected blocks
and crash
+ // the BE (issue #66390), so it reads unordered like a plain scan.
if (_tablet_reader_params.binlog_scan_type != TBinlogScanType::NONE) {
- _tablet_reader_params.read_orderby_key = true;
+ _tablet_reader_params.read_orderby_key = is_binlog_merge_scan;
+ _tablet_reader_params.force_key_ordered_read = is_binlog_merge_scan;
_tablet_reader_params.read_orderby_key_reverse = false;
_tablet_reader_params.read_orderby_key_num_prefix_columns = 0;
_tablet_reader_params.read_orderby_key_limit = 0;
- _tablet_reader_params.force_key_ordered_read = true;
_tablet_reader_params.topn_filter_source_node_ids.clear();
}
diff --git a/be/src/storage/iterator/vcollect_iterator.cpp
b/be/src/storage/iterator/vcollect_iterator.cpp
index 8b2a786e1a6..51371b34e66 100644
--- a/be/src/storage/iterator/vcollect_iterator.cpp
+++ b/be/src/storage/iterator/vcollect_iterator.cpp
@@ -709,6 +709,8 @@ Status VCollectIterator::Level1Iterator::init(bool
get_data_by_ref) {
}
}
+ RETURN_IF_ERROR(_validate_merge_compare_contract(sequence_loc));
+
_heap = std::make_unique<MergeHeap>(LevelIteratorComparator(
sequence_loc, _is_reverse,
_reader->_reader_context.use_insert_order_when_same,
tso_col_id >= 0));
@@ -766,6 +768,61 @@ void
VCollectIterator::Level1Iterator::init_level0_iterators_for_union() {
}
}
+// LevelIteratorComparator reads block positions that carry no bounds checks
in release
+// builds: either the explicit compare-column positions, or positions
+// [0, tablet_schema().num_key_columns()) plus the sequence tie-break
position. A read
+// projection that omits or reorders the leading key columns would turn the
first heap
+// comparison into an out-of-bounds or semantically wrong positional access
(issue #66390).
+// Validate the contract against every child's first block before anything
enters _heap.
+Status VCollectIterator::Level1Iterator::_validate_merge_compare_contract(int
sequence_loc) const {
+ const auto& return_columns = _reader->_return_columns;
+ const size_t num_key_columns = _schema.num_key_columns();
+ for (const auto& child : _children) {
+ const IteratorRowRef* ref = child->current_row_ref();
+ if (ref->block == nullptr) {
+ continue;
+ }
+ const size_t block_columns = ref->block->columns();
+ auto contract_error = [&](const std::string& detail) {
+ std::string projected_ids;
+ for (auto cid : return_columns) {
+ if (!projected_ids.empty()) {
+ projected_ids += ',';
+ }
+ projected_ids += std::to_string(cid);
+ }
+ return Status::InternalError(
+ "merge heap compare contract violated: {}, tablet_id={},
block_columns={}, "
+ "num_key_columns={}, sequence_loc={}, return_columns=[{}]",
+ detail, _reader->_tablet->tablet_id(), block_columns,
num_key_columns,
+ sequence_loc, projected_ids);
+ };
+ if (_compare_columns != nullptr) {
+ for (uint32_t pos : *_compare_columns) {
+ if (pos >= block_columns) {
+ return contract_error(
+ fmt::format("compare column position {} out of
range", pos));
+ }
+ }
+ } else {
+ if (num_key_columns > return_columns.size() || num_key_columns >
block_columns) {
+ return contract_error("projection has fewer columns than the
key prefix");
+ }
+ for (size_t i = 0; i < num_key_columns; ++i) {
+ if (return_columns[i] != i) {
+ return contract_error(
+ fmt::format("position {} holds column id {}
instead of key column {}",
+ i, return_columns[i], i));
+ }
+ }
+ }
+ if (sequence_loc != -1 && static_cast<size_t>(sequence_loc) >=
block_columns) {
+ return contract_error("sequence column position out of range");
+ }
+ }
+ return Status::OK();
+}
+
Status VCollectIterator::Level1Iterator::_merge_next(IteratorRowRef* ref) {
auto res = _cur_child->next(ref);
if (LIKELY(res.ok())) {
diff --git a/be/src/storage/iterator/vcollect_iterator.h
b/be/src/storage/iterator/vcollect_iterator.h
index bf99a85fcac..9af05e0ced1 100644
--- a/be/src/storage/iterator/vcollect_iterator.h
+++ b/be/src/storage/iterator/vcollect_iterator.h
@@ -324,6 +324,13 @@ private:
bool collected_enough_rows(const MutableColumns& columns, int
rows_to_merge) const;
private:
+ // Validate that every block position LevelIteratorComparator may
touch exists in
+ // each child's current block and, for the default key-prefix
comparison, that the
+ // read projection really starts with the full ordered key prefix.
Called before
+ // any child is pushed into _heap, so a broken projection surfaces as
an error
+ // instead of an out-of-bounds positional access (issue #66390).
+ Status _validate_merge_compare_contract(int sequence_loc) const;
+
Status _merge_next(IteratorRowRef* ref);
Status _normal_next(IteratorRowRef* ref);
diff --git a/be/src/storage/iterator/vgeneric_iterators.cpp
b/be/src/storage/iterator/vgeneric_iterators.cpp
index b854776c6f8..13e6040b799 100644
--- a/be/src/storage/iterator/vgeneric_iterators.cpp
+++ b/be/src/storage/iterator/vgeneric_iterators.cpp
@@ -296,6 +296,7 @@ Status VMergeIteratorContext::init(const
StorageReadOptions& opts) {
_record_rowids = opts.record_rowids;
RETURN_IF_ERROR(_load_next_block());
if (valid()) {
+ RETURN_IF_ERROR(_validate_compare_contract(opts));
RETURN_IF_ERROR(advance());
}
_pre_ctx_same_bit.reserve(_block_row_max);
@@ -303,6 +304,58 @@ Status VMergeIteratorContext::init(const
StorageReadOptions& opts) {
return Status::OK();
}
+// compare() reads block positions that are only DCHECK-bounds-checked in
Block::compare_at(),
+// so in a release build a projection violating the merge contract turns into
an out-of-bounds
+// read inside std::push_heap and kills the BE (issue #66390). Verify the
contract once the
+// first block is loaded and surface a diagnosable error instead:
+// - explicit compare columns (_compare_columns) must all point inside the
block;
+// - otherwise the default comparison touches positions [0,
_num_key_columns), where
+// _num_key_columns counts the key columns of the WHOLE tablet schema. Key
columns always
+// occupy column ids [0, num_key_columns) of the tablet schema, so the
projection must
+// start with exactly those ids, in order, for the positional comparison
to be key order;
+// - the sequence tie-break column, when present, must point inside the
block as well.
+Status VMergeIteratorContext::_validate_compare_contract(const
StorageReadOptions& opts) const {
+ const size_t block_columns = _block->columns();
+ auto contract_error = [&](const std::string& detail) {
+ std::string projected_ids;
+ for (auto cid : _output_schema->column_ids()) {
+ if (!projected_ids.empty()) {
+ projected_ids += ',';
+ }
+ projected_ids += std::to_string(cid);
+ }
+ return Status::InternalError(
+ "merge iterator compare contract violated: {}, tablet_id={},
rowset_id={}, "
+ "version={}, block_columns={}, num_key_columns={},
sequence_id_idx={}, "
+ "projected_column_ids=[{}]",
+ detail, opts.tablet_id, opts.rowset_id.to_string(),
opts.version.to_string(),
+ block_columns, _num_key_columns, _sequence_id_idx,
projected_ids);
+ };
+ if (_compare_columns != nullptr) {
+ for (uint32_t pos : *_compare_columns) {
+ if (pos >= block_columns) {
+ return contract_error(fmt::format("compare column position {}
out of range", pos));
+ }
+ }
+ } else {
+ const auto num_key_columns = static_cast<size_t>(_num_key_columns);
+ if (num_key_columns > _output_schema->num_column_ids() ||
num_key_columns > block_columns) {
+ return contract_error("projection has fewer columns than the key
prefix");
+ }
+ for (size_t i = 0; i < num_key_columns; ++i) {
+ if (_output_schema->column_ids()[i] != static_cast<ColumnId>(i)) {
+ return contract_error(
+ fmt::format("position {} holds column id {} instead of
key column {}", i,
+ _output_schema->column_ids()[i], i));
+ }
+ }
+ }
+ if (_sequence_id_idx != -1 && static_cast<size_t>(_sequence_id_idx) >=
block_columns) {
+ return contract_error("sequence column position out of range");
+ }
+ return Status::OK();
+}
+
Status VMergeIteratorContext::advance() {
_skip = false;
_same = false;
diff --git a/be/src/storage/iterator/vgeneric_iterators.h
b/be/src/storage/iterator/vgeneric_iterators.h
index aacf8175457..38eee70cfae 100644
--- a/be/src/storage/iterator/vgeneric_iterators.h
+++ b/be/src/storage/iterator/vgeneric_iterators.h
@@ -185,6 +185,12 @@ private:
// Load next block into _block
Status _load_next_block();
+ // Validate that every block position compare() may touch actually exists
in _block
+ // and, for the default key-prefix comparison, that the projection really
starts with
+ // the full ordered key prefix. Returns an error instead of letting
compare() perform
+ // an out-of-bounds or semantically wrong positional access (issue #66390).
+ Status _validate_compare_contract(const StorageReadOptions& opts) const;
+
RowwiseIteratorUPtr _iter;
int _sequence_id_idx = -1;
diff --git a/be/test/exec/scan/vgeneric_iterators_test.cpp
b/be/test/exec/scan/vgeneric_iterators_test.cpp
index f8e3e09401a..d461512c4e0 100644
--- a/be/test/exec/scan/vgeneric_iterators_test.cpp
+++ b/be/test/exec/scan/vgeneric_iterators_test.cpp
@@ -42,7 +42,7 @@ public:
virtual ~VGenericIteratorsTest() {}
};
-static Schema 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);
@@ -57,6 +57,11 @@ static Schema create_schema() {
col_schemas.emplace_back(
std::make_shared<TabletColumn>(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM,
FieldType::OLAP_FIELD_TYPE_BIGINT,
true));
+ return col_schemas;
+}
+
+static Schema create_schema() {
+ std::vector<TabletColumnPtr> col_schemas = create_col_schemas();
std::vector<ColumnId> column_ids(col_schemas.size());
for (uint32_t cid = 0; cid < column_ids.size(); ++cid) {
@@ -402,4 +407,100 @@ TEST(VGenericIteratorsTest,
MergeWithSeqColumnSmallSeqFirst) {
EXPECT_EQ(0, actual_value);
}
+// 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(Schema 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_column_ids(); ++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(_schema.column_id(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 Schema& schema() const override { return _schema; }
+
+private:
+ Schema _schema;
+ size_t _num_rows;
+ size_t _rows_returned = 0;
+};
+
+// The merge-heap comparator compares the first num_key_columns block
positions when no
+// explicit compare columns are given, and num_key_columns counts the key
columns of the
+// WHOLE tablet schema. A projection narrower than the key prefix must be
rejected at
+// init time with an error instead of crashing inside std::push_heap (issue
#66390: a
+// ROW-binlog APPEND_ONLY scan projected value columns only).
+TEST(VGenericIteratorsTest, MergeRejectsProjectionMissingKeyPrefix) {
+ // Full tablet schema: k0(smallint), k1(int), v2(bigint); project only v2.
+ Schema projected(create_col_schemas(), std::vector<ColumnId> {2});
+ auto output_schema = std::make_shared<Schema>(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 as above, but the projection has enough columns while not starting
with the full
+// ordered key prefix (k0 is missing): position 0 would be compared as if it
were k0.
+TEST(VGenericIteratorsTest, MergeRejectsProjectionWithoutLeadingKey) {
+ // Full tablet schema: k0(smallint), k1(int), v2(bigint); project {k1, v2}.
+ Schema projected(create_col_schemas(), std::vector<ColumnId> {1, 2});
+ auto output_schema = std::make_shared<Schema>(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/regression-test/suites/row_binlog_p0/test_binlog_append_only_projection.groovy
b/regression-test/suites/row_binlog_p0/test_binlog_append_only_projection.groovy
new file mode 100644
index 00000000000..23369879bb9
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_binlog_append_only_projection.groovy
@@ -0,0 +1,132 @@
+// 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.
+
+// Regression test for issue #66390: a ROW-binlog APPEND_ONLY scan whose SQL
projection
+// contains fewer columns than the tablet key count (or omits the leading key
columns)
+// used to force a key-ordered merge in the storage layer. The merge
comparator then read
+// key positions that do not exist in the projected blocks and crashed the BE
with SIGSEGV
+// inside VMergeIterator::init / std::push_heap.
+//
+// The table has TWO key columns and the queries below project one value
column only
+// (0 key columns read), or a non-leading key subset, over overlapping
multi-rowset data,
+// which is exactly the shape that used to crash.
+suite("test_binlog_append_only_projection", "nonConcurrent") {
+ if (isCloudMode()) {
+ return
+ }
+ sql "DROP DATABASE IF EXISTS test_binlog_append_only_projection_db"
+ sql "CREATE DATABASE test_binlog_append_only_projection_db"
+ sql "USE test_binlog_append_only_projection_db"
+ sql "set enable_nereids_planner=true"
+ sql "set enable_fallback_to_original_planner=false"
+
+ def dupTable = "append_only_proj_dup"
+ def incrTimeFormat = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
+
+ try {
+ sql "DROP TABLE IF EXISTS ${dupTable}"
+
+ // Two key columns so that any single-column projection is narrower
than the
+ // key prefix the merge comparator would use.
+ sql """
+ CREATE TABLE ${dupTable} (
+ k1 BIGINT,
+ k2 INT,
+ v1 INT,
+ v2 VARCHAR(16) NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k1, k2)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+
+ sql "INSERT INTO ${dupTable} VALUES (1, 1, 10, 'seed1')"
+ sql "INSERT INTO ${dupTable} VALUES (2, 2, 20, 'seed2')"
+ sql "sync"
+ sleep(1200)
+ def t0 = incrTimeFormat.format(new Date())
+ sleep(1200)
+
+ // Each INSERT produces its own rowset; interleaved keys make the
rowsets'
+ // key ranges overlap, so a key-ordered read would need a real merge.
+ sql "INSERT INTO ${dupTable} VALUES (1, 3, 30, 'w1'), (9, 1, 31, 'w1')"
+ sql "INSERT INTO ${dupTable} VALUES (2, 4, 40, NULL), (8, 2, 41, 'w2')"
+ sql "INSERT INTO ${dupTable} VALUES (1, 5, 50, 'w3'), (9, 3, 51, NULL)"
+ sql "sync"
+ sleep(1200)
+ def t1 = incrTimeFormat.format(new Date())
+ sleep(1200)
+ sql "INSERT INTO ${dupTable} VALUES (7, 7, 70, 'late')"
+ sql "sync"
+
+ // 1. Project a single value column: 1 projected column < 2 key
columns.
+ // This is the exact shape that used to crash the BE.
+ assertEquals([[30], [31], [40], [41], [50], [51]],
+ sql("""SELECT v1
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "APPEND_ONLY")
+ ORDER BY v1"""))
+
+ // 2. Project only the second key column: enough columns to compare,
but the
+ // leading key k1 is absent, so a positional key comparison would
have
+ // silently compared the wrong columns.
+ assertEquals([[1], [2], [3], [3], [4], [5]],
+ sql("""SELECT k2
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "APPEND_ONLY")
+ ORDER BY k2"""))
+
+ // 3. Projection in reversed column order relative to the schema.
+ assertEquals([[30, 3L], [31, 1L], [40, 4L], [41, 2L], [50, 5L], [51,
3L]],
+ sql("""SELECT v1, CAST(k2 AS BIGINT)
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "APPEND_ONLY")
+ ORDER BY v1"""))
+
+ // 4. Aggregate over a narrow projection.
+ assertEquals([[6L, 243L]],
+ sql("""SELECT count(*), sum(v1)
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "APPEND_ONLY")"""))
+
+ // 5. DETAIL / MIN_DELTA keep the forced key-ordered merge but widen
the
+ // storage projection with the full key prefix internally; a narrow
SQL
+ // projection must still work and return the same rows for a dup
table.
+ assertEquals([[30], [31], [40], [41], [50], [51]],
+ sql("""SELECT v1
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "DETAIL")
+ ORDER BY v1"""))
+ assertEquals([[30], [31], [40], [41], [50], [51]],
+ sql("""SELECT v1
+ FROM ${dupTable}@incr('startTimestamp' = '${t0}',
+ "endTimestamp" = "${t1}",
+ "incrementType" = "MIN_DELTA")
+ ORDER BY v1"""))
+ } finally {
+ sql "DROP DATABASE IF EXISTS test_binlog_append_only_projection_db"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
