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 3175cc01d11 [fix](be) Suppress no-op updates in min delta scans
(#66849)
3175cc01d11 is described below
commit 3175cc01d113013e76d30290f22fb9ee6dcc65bc
Author: Gavin Chou <[email protected]>
AuthorDate: Mon Aug 24 16:44:11 2026 +0800
[fix](be) Suppress no-op updates in min delta scans (#66849)
Problem Summary:
MIN_DELTA row-binlog scans previously classified each primary-key window
only from its first and last operation. As a result, an update emitted
`UPDATE_BEFORE` and `UPDATE_AFTER` rows even when the complete row image
was unchanged, including multi-update chains that eventually returned to
their original values.
This change reads all AFTER/BEFORE value columns needed by MIN_DELTA,
compares the first BEFORE image with the final AFTER image, and converts
a net-zero update into `SKIP`. Comparison-only columns remain hidden
from the SQL projection. Missing, incompatible, or unsupported BEFORE
images preserve the existing UPDATE output conservatively.
The tests cover direct no-op updates, changes in unprojected columns,
multiple keys and columns, pending output across batch boundaries, and
complex insert/delete/update chains that either return to the original
image or finish changed.
### Release note
MIN_DELTA row-binlog queries no longer return update rows when the
complete row image is unchanged across the query window.
---
.../storage/iterator/binlog_block_reader_utils.h | 1 -
be/src/storage/iterator/block_reader.cpp | 50 ++
be/src/storage/iterator/block_reader.h | 5 +
be/src/storage/schema.cpp | 85 +++
be/src/storage/schema.h | 16 +
.../iterator/binlog_block_reader_utils_test.cpp | 29 -
.../block_reader_change_next_block_test.cpp | 583 ++++++++++++++++++++-
be/test/storage/read_schema_test.cpp | 52 ++
.../glue/translator/PhysicalPlanTranslator.java | 14 +-
.../translator/PhysicalPlanTranslatorTest.java | 5 +-
.../row_binlog_p0/test_binlog_changes_syntax.out | 3 +
.../test_binlog_changes_syntax.groovy | 19 +
12 files changed, 807 insertions(+), 55 deletions(-)
diff --git a/be/src/storage/iterator/binlog_block_reader_utils.h
b/be/src/storage/iterator/binlog_block_reader_utils.h
index 1dfc3520482..5b3bf53ce02 100644
--- a/be/src/storage/iterator/binlog_block_reader_utils.h
+++ b/be/src/storage/iterator/binlog_block_reader_utils.h
@@ -31,7 +31,6 @@ constexpr int64_t STREAM_CHANGE_INSERT = 0;
constexpr int64_t STREAM_CHANGE_DELETE = 1;
constexpr int64_t STREAM_CHANGE_UPDATE_BEFORE = 2;
constexpr int64_t STREAM_CHANGE_UPDATE_AFTER = 3;
-
enum class MinDeltaResultType { SKIP, INSERT, DELETE, UPDATE_BEFORE_AFTER };
// MIN_DELTA uses row binlog op codes as indices into a 2D lookup table, so we
guard the op layout here.
diff --git a/be/src/storage/iterator/block_reader.cpp
b/be/src/storage/iterator/block_reader.cpp
index 729682d038f..29aa476bd0e 100644
--- a/be/src/storage/iterator/block_reader.cpp
+++ b/be/src/storage/iterator/block_reader.cpp
@@ -31,6 +31,7 @@
#include "cloud/config.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
+#include "common/exception.h"
#include "common/status.h"
#include "core/block/column_with_type_and_name.h"
#include "core/column/column_nullable.h"
@@ -118,6 +119,42 @@ uint32_t
BlockReader::_resolve_source_column_ordinal(uint32_t ordinal, bool use_
return use_before ? _read_schema->before_column_ordinal(ordinal) : ordinal;
}
+bool BlockReader::_min_delta_values_equal(size_t last_row) {
+ const auto& value_column_pairs =
_read_schema->row_binlog_value_column_pairs();
+ if (!_read_schema->row_binlog_value_pairs_complete() ||
value_column_pairs.empty() ||
+ _min_delta_value_compare_unsupported) {
+ return false;
+ }
+ bool before_image_available = false;
+ for (const auto& [after_idx, before_idx] : value_column_pairs) {
+ const auto* before_column = _stored_data_columns[before_idx].get();
+ if (const auto* nullable =
check_and_get_column<ColumnNullable>(*before_column)) {
+ before_image_available |= !nullable->is_null_at(0);
+ } else {
+ before_image_available = true;
+ }
+ try {
+ if (_stored_data_columns[before_idx]->compare_at(
+ 0, last_row, *_stored_data_columns[after_idx], -1) !=
0) {
+ return false;
+ }
+ } catch (const Exception& e) {
+ if (e.code() != ErrorCode::NOT_IMPLEMENTED_ERROR) {
+ throw;
+ }
+ // Column types are stable for the lifetime of this reader. Once
one value column
+ // reports that compare_at is unsupported, no row can be proven to
be a no-op. Cache
+ // that capability result so BITMAP-like columns pay the exception
cost at most once.
+ _min_delta_value_compare_unsupported = true;
+ return false;
+ }
+ }
+ // Historical lookup and the compatibility writer both represent an
unavailable BEFORE image
+ // as an all-NULL row. Without a persisted validity bit, retaining the
UPDATE is the only safe
+ // choice; otherwise a missing-key DELETE followed by an all-NULL INSERT
would disappear.
+ return before_image_available;
+}
+
void BlockReader::_init_pending_row_columns(const Block& block) {
if (!_pending_row_columns.empty()) {
return;
@@ -243,6 +280,11 @@ Status BlockReader::_min_delta_next_block(Block* block,
bool* eof) {
output_row_count++;
break;
case
binlog::AggregateFunctionMinDelta::ResultType::UPDATE_BEFORE_AFTER:
+ if (binlog::is_valid_row_binlog_op(first_op) &&
+ binlog::is_valid_row_binlog_op(last_op) &&
+ _min_delta_values_equal(group_size - 1)) {
+ break;
+ }
for (size_t ordinal = 0; ordinal < target_columns.size();
++ordinal) {
if (static_cast<int32_t>(ordinal) == op_ordinal) {
RETURN_IF_ERROR(_write_binlog_op(*target_columns[ordinal],
@@ -498,6 +540,14 @@ Status BlockReader::init(const ReaderParams& read_params) {
_read_schema = std::move(read_schema);
}
+ if (read_params.binlog_scan_type == TBinlogScanType::MIN_DELTA ||
+ read_params.binlog_scan_type == TBinlogScanType::DETAIL) {
+ auto read_schema = std::make_shared<ReadSchema>(*_read_schema);
+ read_schema->init_row_binlog_column_mappings(*_tablet_schema);
+ _read_schema = std::move(read_schema);
+ _min_delta_value_compare_unsupported = false;
+ }
+
// Every merge and caller Block has the exact read-schema layout. Cache
only
// the non-key columns of AGG tables that go through the aggregate
machinery.
const auto num_block_columns =
static_cast<uint32_t>(_read_schema->num_block_columns());
diff --git a/be/src/storage/iterator/block_reader.h
b/be/src/storage/iterator/block_reader.h
index a93f96ee870..de1ad68fdcd 100644
--- a/be/src/storage/iterator/block_reader.h
+++ b/be/src/storage/iterator/block_reader.h
@@ -81,6 +81,8 @@ private:
uint32_t _resolve_source_column_ordinal(uint32_t ordinal, bool use_before)
const;
+ bool _min_delta_values_equal(size_t last_row);
+
void _init_pending_row_columns(const Block& block);
bool _emit_pending_row(MutableColumns& target_columns, size_t&
output_row_count);
@@ -153,6 +155,9 @@ private:
bool _is_rowsets_overlapping = true;
+ // Unsupported compare_at means equality cannot be proven, so MIN_DELTA
conservatively keeps
+ // UPDATE rows. Column types are stable within a reader; cache this after
the first exception.
+ bool _min_delta_value_compare_unsupported = false;
Arena _arena;
};
diff --git a/be/src/storage/schema.cpp b/be/src/storage/schema.cpp
index e1b7a7c9228..c3fd1af402b 100644
--- a/be/src/storage/schema.cpp
+++ b/be/src/storage/schema.cpp
@@ -29,6 +29,25 @@
namespace doris {
+namespace {
+
+bool row_binlog_value_columns_have_same_type(const TabletColumn& lhs, const
TabletColumn& rhs) {
+ if (lhs.type() != rhs.type() || lhs.is_nullable() != rhs.is_nullable() ||
+ lhs.length() != rhs.length() || lhs.precision() != rhs.precision() ||
+ lhs.frac() != rhs.frac() || lhs.get_subtype_count() !=
rhs.get_subtype_count()) {
+ return false;
+ }
+ for (uint32_t i = 0; i < lhs.get_subtype_count(); ++i) {
+ if (!row_binlog_value_columns_have_same_type(lhs.get_sub_column(i),
+ rhs.get_sub_column(i))) {
+ return false;
+ }
+ }
+ return true;
+}
+
+} // namespace
+
std::vector<TabletColumnPtr> project_columns_by_ordinal(
const std::vector<TabletColumnPtr>& columns,
const std::vector<ColumnId>& source_column_ordinals) {
@@ -99,6 +118,72 @@ void ReadSchema::_init_before_column_ordinals() {
}
}
+void ReadSchema::init_row_binlog_column_mappings(const TabletSchema&
tablet_schema) {
+ DORIS_CHECK_GE(_op_ordinal, 0);
+ DORIS_CHECK_EQ(_before_column_ordinals.size(), _num_block_columns);
+
+ _row_binlog_value_column_pairs.clear();
+ _row_binlog_value_pairs_complete = false;
+
+ std::vector<ColumnId> value_column_ids;
+ value_column_ids.reserve(tablet_schema.num_columns() -
tablet_schema.num_key_columns());
+ for (ColumnId cid = 0; cid < tablet_schema.num_columns(); ++cid) {
+ if (static_cast<int32_t>(cid) == tablet_schema.binlog_tso_col_idx() ||
+ static_cast<int32_t>(cid) == tablet_schema.binlog_lsn_col_idx() ||
+ static_cast<int32_t>(cid) == tablet_schema.binlog_op_col_idx() ||
+ tablet_schema.column(cid).is_key()) {
+ continue;
+ }
+ value_column_ids.push_back(cid);
+ }
+
+ if (value_column_ids.empty() || value_column_ids.size() % 2 != 0) {
+ return;
+ }
+
+ const size_t value_column_count = value_column_ids.size() / 2;
+ for (size_t i = 0; i < value_column_count; ++i) {
+ const auto& after = tablet_schema.column(value_column_ids[i]);
+ const auto& before = tablet_schema.column(value_column_ids[i +
value_column_count]);
+ if (before.name() != binlog::build_before_column_name(after.name()) ||
+ !row_binlog_value_columns_have_same_type(after, before)) {
+ return;
+ }
+ }
+
+ // A valid physical row-binlog layout starts from an identity mapping.
Only AFTER value
+ // columns map to their BEFORE companions; keys, metadata and BEFORE
columns map to themselves.
+ for (ColumnId ordinal = 0; ordinal < _num_block_columns; ++ordinal) {
+ _before_column_ordinals[ordinal] = ordinal;
+ }
+
+ bool complete = true;
+ _row_binlog_value_column_pairs.reserve(value_column_count);
+ for (size_t i = 0; i < value_column_count; ++i) {
+ const auto after_cid = value_column_ids[i];
+ const auto before_cid = value_column_ids[i + value_column_count];
+ const int32_t after_ordinal =
ordinal_by_uid(tablet_schema.column(after_cid).unique_id());
+ const int32_t before_ordinal =
ordinal_by_uid(tablet_schema.column(before_cid).unique_id());
+ if (after_ordinal < 0 || before_ordinal < 0 ||
+ static_cast<size_t>(after_ordinal) >= _num_block_columns ||
+ static_cast<size_t>(before_ordinal) >= _num_block_columns) {
+ complete = false;
+ continue;
+ }
+
+ const auto after = cast_set<ColumnId>(after_ordinal);
+ const auto before = cast_set<ColumnId>(before_ordinal);
+ _before_column_ordinals[after] = before;
+ if (!_read_types[after]->equals(*_read_types[before])) {
+ complete = false;
+ continue;
+ }
+ _row_binlog_value_column_pairs.emplace_back(after, before);
+ }
+ _row_binlog_value_pairs_complete =
+ complete && _row_binlog_value_column_pairs.size() ==
value_column_count;
+}
+
Block ReadSchema::create_read_block() const {
Block block;
for (size_t ordinal = 0; ordinal < _num_block_columns; ++ordinal) {
diff --git a/be/src/storage/schema.h b/be/src/storage/schema.h
index f26732cc67e..473e603a05a 100644
--- a/be/src/storage/schema.h
+++ b/be/src/storage/schema.h
@@ -25,6 +25,7 @@
#include <memory>
#include <string>
#include <unordered_map>
+#include <utility>
#include <vector>
#include "common/consts.h"
@@ -58,6 +59,7 @@ std::vector<TabletColumnPtr> project_columns_by_ordinal(
class ReadSchema {
public:
using SequenceMap = std::unordered_map<ColumnId, std::vector<ColumnId>>;
+ using RowBinlogValueColumnPairs = std::vector<std::pair<ColumnId,
ColumnId>>;
explicit ReadSchema(std::vector<TabletColumnPtr> columns);
@@ -87,6 +89,12 @@ public:
const SequenceMap& sequence_map() const { return _sequence_map; }
+ // Initialize all row-binlog column relationships from the physical tablet
schema and map
+ // them to this ReadSchema's dense ordinals. Physical pairing avoids
ambiguous column-name
+ // lookup, while schemas without a complete physical layout retain the
name-based BEFORE
+ // mapping initialized by the constructor.
+ void init_row_binlog_column_mappings(const TabletSchema& tablet_schema);
+
// Return the matching before-image ordinal for a Row Binlog value column.
For example, in
// [v1, v2, __BEFORE__v1__, __BEFORE__v2__], 0 maps to 2 and 1 maps to 3.
Columns without a
// before image, including TSO/LSN/OP, map to themselves.
@@ -95,6 +103,12 @@ public:
return _before_column_ordinals[ordinal];
}
+ const RowBinlogValueColumnPairs& row_binlog_value_column_pairs() const {
+ return _row_binlog_value_column_pairs;
+ }
+
+ bool row_binlog_value_pairs_complete() const { return
_row_binlog_value_pairs_complete; }
+
const TabletColumn* column(size_t ordinal) const { return
_read_columns[ordinal].get(); }
// Total columns used inside storage, including appended storage-only
columns.
@@ -224,6 +238,8 @@ private:
std::unordered_map<int32_t, int32_t> _uid_to_ordinal;
SequenceMap _sequence_map;
std::vector<ColumnId> _before_column_ordinals;
+ RowBinlogValueColumnPairs _row_binlog_value_column_pairs;
+ bool _row_binlog_value_pairs_complete = false;
};
} // namespace doris
diff --git a/be/test/storage/iterator/binlog_block_reader_utils_test.cpp
b/be/test/storage/iterator/binlog_block_reader_utils_test.cpp
deleted file mode 100644
index 9d099fe1c92..00000000000
--- a/be/test/storage/iterator/binlog_block_reader_utils_test.cpp
+++ /dev/null
@@ -1,29 +0,0 @@
-// Licensed to the Apache Software Foundation (ASF) under one
-// or more contributor license agreements. See the NOTICE file
-// distributed with this work for additional information
-// regarding copyright ownership. The ASF licenses this file
-// to you under the Apache License, Version 2.0 (the
-// "License"); you may not use this file except in compliance
-// with the License. You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing,
-// software distributed under the License is distributed on an
-// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-// KIND, either express or implied. See the License for the
-// specific language governing permissions and limitations
-// under the License.
-
-#include "storage/iterator/binlog_block_reader_utils.h"
-
-#include <gtest/gtest.h>
-
-namespace doris {
-
-class BinlogBlockReaderUtilsTest : public testing::Test {};
-TEST_F(BinlogBlockReaderUtilsTest, BuildBeforeColumnName) {
- EXPECT_EQ(binlog::build_before_column_name("v1"), "__BEFORE__v1__");
-}
-
-} // namespace doris
diff --git a/be/test/storage/iterator/block_reader_change_next_block_test.cpp
b/be/test/storage/iterator/block_reader_change_next_block_test.cpp
index d7fcd447152..a75bd4a6cfe 100644
--- a/be/test/storage/iterator/block_reader_change_next_block_test.cpp
+++ b/be/test/storage/iterator/block_reader_change_next_block_test.cpp
@@ -36,13 +36,20 @@
#include <memory>
#include <string>
+#include <string_view>
+#include <utility>
#include <vector>
#include "common/config.h"
+#include "common/exception.h"
#include "common/status.h"
#include "core/assert_cast.h"
#include "core/block/block.h"
+#include "core/column/column_dummy.h"
+#include "core/column/column_nullable.h"
#include "core/column/column_vector.h"
+#include "core/data_type/data_type_nothing.h"
+#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "storage/binlog.h"
#include "storage/iterator/binlog_block_reader_utils.h"
@@ -77,6 +84,70 @@ struct Row {
int64_t op;
};
+class ThrowOnCompareColumn final : public COWHelper<IColumnDummy,
ThrowOnCompareColumn> {
+private:
+ friend class COWHelper<IColumnDummy, ThrowOnCompareColumn>;
+
+ ThrowOnCompareColumn(size_t size, int error_code, std::shared_ptr<size_t>
compare_calls)
+ : _error_code(error_code),
_compare_calls(std::move(compare_calls)) {
+ s = size;
+ }
+ ThrowOnCompareColumn(const ThrowOnCompareColumn&) = default;
+
+public:
+ std::string get_name() const override { return "ThrowOnCompare"; }
+
+ MutableColumnPtr clone_dummy(size_t size) const override {
+ return ThrowOnCompareColumn::create(size, _error_code, _compare_calls);
+ }
+
+ bool structure_equals(const IColumn& rhs) const override {
+ return typeid(rhs) == typeid(ThrowOnCompareColumn);
+ }
+
+ int compare_at(size_t, size_t, const IColumn&, int) const override {
+ ++*_compare_calls;
+ throw Exception(_error_code, "injected compare_at failure");
+ }
+
+private:
+ int _error_code;
+ std::shared_ptr<size_t> _compare_calls;
+};
+
+TabletColumn make_test_column(std::string name, FieldType type, bool is_key,
bool is_nullable,
+ int32_t unique_id) {
+ TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
type, is_nullable,
+ unique_id, sizeof(int64_t));
+ column.set_name(std::move(name));
+ column.set_is_key(is_key);
+ column.set_index_length(sizeof(int64_t));
+ return column;
+}
+
+TabletSchemaSPtr make_test_tablet_schema(
+ const std::vector<std::pair<std::string, FieldType>>& value_columns = {
+ {"val", FieldType::OLAP_FIELD_TYPE_BIGINT}}) {
+ auto schema = std::make_shared<TabletSchema>();
+ int32_t unique_id = 0;
+ schema->append_column(
+ make_test_column("key", FieldType::OLAP_FIELD_TYPE_BIGINT, true,
false, unique_id++));
+ for (const auto& [name, type] : value_columns) {
+ schema->append_column(make_test_column(name, type, false, true,
unique_id++));
+ }
+ for (const auto& [name, type] : value_columns) {
+
schema->append_column(make_test_column(binlog::build_before_column_name(name),
type, false,
+ true, unique_id++));
+ }
+ schema->append_column(make_test_column(BINLOG_TSO_COL,
FieldType::OLAP_FIELD_TYPE_BIGINT, false,
+ true, unique_id++));
+ schema->append_column(make_test_column(BINLOG_LSN_COL,
FieldType::OLAP_FIELD_TYPE_BIGINT, false,
+ false, unique_id++));
+ schema->append_column(make_test_column(BINLOG_OP_COL,
FieldType::OLAP_FIELD_TYPE_BIGINT, false,
+ false, unique_id));
+ return schema;
+}
+
std::shared_ptr<Block> make_source_block(const std::vector<Row>& rows) {
auto block = std::make_shared<Block>();
auto type = std::make_shared<DataTypeInt64>();
@@ -104,6 +175,118 @@ std::shared_ptr<Block> make_source_block(const
std::vector<Row>& rows) {
return block;
}
+std::shared_ptr<Block> make_all_null_source_block() {
+ auto block = std::make_shared<Block>();
+ auto int_type = std::make_shared<DataTypeInt64>();
+ auto nullable_int_type = make_nullable(int_type);
+ auto key_col = ColumnInt64::create();
+ auto val_col = nullable_int_type->create_column();
+ auto before_col = nullable_int_type->create_column();
+ auto tso_col = ColumnInt64::create();
+ auto lsn_col = ColumnInt64::create();
+ auto op_col = ColumnInt64::create();
+
+ key_col->insert_many_vals(1, 2);
+ val_col->insert_many_defaults(2);
+ before_col->insert_many_defaults(2);
+ tso_col->insert_value(1);
+ tso_col->insert_value(2);
+ lsn_col->insert_value(1);
+ lsn_col->insert_value(2);
+ op_col->insert_value(ROW_BINLOG_DELETE);
+ op_col->insert_value(ROW_BINLOG_APPEND);
+
+ block->insert({std::move(key_col), int_type, "key"});
+ block->insert({std::move(val_col), nullable_int_type, "val"});
+ block->insert(
+ {std::move(before_col), nullable_int_type,
binlog::build_before_column_name("val")});
+ block->insert({std::move(tso_col), int_type, BINLOG_TSO_COL});
+ block->insert({std::move(lsn_col), int_type, BINLOG_LSN_COL});
+ block->insert({std::move(op_col), int_type, BINLOG_OP_COL});
+ return block;
+}
+
+std::shared_ptr<Block> make_colliding_name_source_block(int64_t after_v,
int64_t after_collision,
+ int64_t before_v,
+ int64_t
before_collision) {
+ auto block = std::make_shared<Block>();
+ auto type = std::make_shared<DataTypeInt64>();
+ const std::vector<std::pair<std::string, int64_t>> values = {
+ {"key", 1},
+ {"v", after_v},
+ {"__BEFORE__v__", after_collision},
+ {binlog::build_before_column_name("v"), before_v},
+ {binlog::build_before_column_name("__BEFORE__v__"),
before_collision},
+ {BINLOG_TSO_COL, 1},
+ {BINLOG_LSN_COL, 1},
+ {BINLOG_OP_COL, ROW_BINLOG_UPDATE},
+ };
+ for (const auto& [name, value] : values) {
+ auto column = ColumnInt64::create();
+ column->insert_value(value);
+ block->insert({std::move(column), type, name});
+ }
+ return block;
+}
+
+std::shared_ptr<Block> make_signed_zero_source_block() {
+ auto block = std::make_shared<Block>();
+ auto int_type = std::make_shared<DataTypeInt64>();
+ auto float_type = std::make_shared<DataTypeFloat32>();
+ auto key_col = ColumnInt64::create();
+ auto val_col = ColumnFloat32::create();
+ auto before_col = ColumnFloat32::create();
+ auto tso_col = ColumnInt64::create();
+ auto lsn_col = ColumnInt64::create();
+ auto op_col = ColumnInt64::create();
+
+ key_col->insert_value(1);
+ val_col->insert_value(-0.0F);
+ before_col->insert_value(+0.0F);
+ tso_col->insert_value(1);
+ lsn_col->insert_value(1);
+ op_col->insert_value(ROW_BINLOG_UPDATE);
+
+ block->insert({std::move(key_col), int_type, "key"});
+ block->insert({std::move(val_col), float_type, "val"});
+ block->insert({std::move(before_col), float_type,
binlog::build_before_column_name("val")});
+ block->insert({std::move(tso_col), int_type, BINLOG_TSO_COL});
+ block->insert({std::move(lsn_col), int_type, BINLOG_LSN_COL});
+ block->insert({std::move(op_col), int_type, BINLOG_OP_COL});
+ return block;
+}
+
+std::shared_ptr<Block> make_unsupported_compare_source_block(
+ const std::shared_ptr<size_t>& compare_calls) {
+ auto block = std::make_shared<Block>();
+ auto int_type = std::make_shared<DataTypeInt64>();
+ auto nothing_type = std::make_shared<DataTypeNothing>();
+ auto key_col = ColumnInt64::create();
+ auto val_col = ThrowOnCompareColumn::create(0,
ErrorCode::NOT_IMPLEMENTED_ERROR, compare_calls);
+ auto before_col =
+ ThrowOnCompareColumn::create(0, ErrorCode::NOT_IMPLEMENTED_ERROR,
compare_calls);
+ auto tso_col = ColumnInt64::create();
+ auto lsn_col = ColumnInt64::create();
+ auto op_col = ColumnInt64::create();
+
+ for (int64_t key : {1, 2}) {
+ key_col->insert_value(key);
+ val_col->insert_default();
+ before_col->insert_default();
+ tso_col->insert_value(key);
+ lsn_col->insert_value(key);
+ op_col->insert_value(ROW_BINLOG_UPDATE);
+ }
+
+ block->insert({std::move(key_col), int_type, "key"});
+ block->insert({std::move(val_col), nothing_type, "val"});
+ block->insert({std::move(before_col), nothing_type,
binlog::build_before_column_name("val")});
+ block->insert({std::move(tso_col), int_type, BINLOG_TSO_COL});
+ block->insert({std::move(lsn_col), int_type, BINLOG_LSN_COL});
+ block->insert({std::move(op_col), int_type, BINLOG_OP_COL});
+ return block;
+}
+
// Fake merge iterator: hands out rows from `source` one at a time. `is_same`
is
// derived from primary-key equality with the previous emitted row, matching
the
// real merge iterator's contract that consecutive same-key rows are flagged.
@@ -137,7 +320,7 @@ public:
Status next(Block* /*block*/) override { return
Status::Error<END_OF_FILE>(""); }
- RowLocation current_row_location() override { return RowLocation(); }
+ RowLocation current_row_location() override { return {}; }
Status current_block_row_locations(std::vector<RowLocation>* /*loc*/)
override {
return Status::OK();
}
@@ -170,24 +353,26 @@ ReadSchemaSPtr make_read_schema(const
std::vector<std::string>& names = {
return std::make_shared<ReadSchema>(std::move(cols));
}
-TabletSchemaSPtr make_tablet_schema(const ReadSchemaSPtr& schema) {
- auto tablet_schema = std::make_shared<TabletSchema>();
- for (const auto& column : schema->columns()) {
- tablet_schema->append_column(*column);
- }
- return tablet_schema;
-}
-
// Wire a BlockReader as if init() had already completed for a row-binlog
change
// scan over the fixed 6-column schema, then plug in the fake merge iterator.
-void configure_reader(BlockReader& reader, std::shared_ptr<Block> source,
size_t batch_size) {
+void configure_reader(BlockReader& reader, std::shared_ptr<Block> source,
size_t batch_size,
+ TabletSchemaSPtr schema = make_test_tablet_schema()) {
config::enable_adaptive_batch_size = false;
reader._reader_context.batch_size = batch_size;
- // Must be set before constructing the fake LevelIterator: its base ctor
- // snapshots reader->_read_schema.
- reader._read_schema = make_read_schema();
- reader._tablet_schema = make_tablet_schema(reader._read_schema);
+ // The physical tablet schema supplies stable unique ids for AFTER/BEFORE
pairing. The fake
+ // source block supplies the materialized types used by this read schema.
+ reader._tablet_schema = std::move(schema);
+ ASSERT_EQ(reader._tablet_schema->num_columns(), source->columns());
+ std::vector<DataTypePtr> read_types;
+ read_types.reserve(source->columns());
+ for (size_t ordinal = 0; ordinal < source->columns(); ++ordinal) {
+ read_types.emplace_back(source->get_by_position(ordinal).type);
+ }
+ auto read_schema =
+ std::make_shared<ReadSchema>(reader._tablet_schema->columns(),
std::move(read_types));
+ read_schema->init_row_binlog_column_mappings(*reader._tablet_schema);
+ reader._read_schema = std::move(read_schema);
reader._next_row.block = source;
reader._next_row.row_pos = 0;
@@ -228,12 +413,17 @@ std::vector<OutRow> drain(BlockReader& reader, Status
(BlockReader::*fn)(Block*,
bool eof = false;
int guard = 0;
while (!eof) {
- Block block = make_output_block();
+ Block block = reader._read_schema->create_read_block();
Status st = (reader.*fn)(&block, &eof);
EXPECT_TRUE(st.ok()) << st;
+ const int32_t op_ordinal = reader._read_schema->op_ordinal();
+ if (op_ordinal < 0) {
+ ADD_FAILURE() << "row-binlog read schema has no op column";
+ return result;
+ }
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)});
+ out_i64(block, op_ordinal, r)});
}
if (++guard >= 1000) {
ADD_FAILURE() << "drain did not terminate";
@@ -243,6 +433,16 @@ std::vector<OutRow> drain(BlockReader& reader, Status
(BlockReader::*fn)(Block*,
return result;
}
+void expect_out_rows(const std::vector<OutRow>& actual, const
std::vector<OutRow>& expected) {
+ ASSERT_EQ(actual.size(), expected.size());
+ for (size_t i = 0; i < expected.size(); ++i) {
+ SCOPED_TRACE(i);
+ EXPECT_EQ(actual[i].key, expected[i].key);
+ EXPECT_EQ(actual[i].val, expected[i].val);
+ EXPECT_EQ(actual[i].op, expected[i].op);
+ }
+}
+
} // namespace
class BlockReaderChangeNextBlockTest : public testing::Test {
@@ -333,6 +533,338 @@ TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaUpdateBeforeAfter) {
EXPECT_EQ(out[1].val, 30); // after value from the last op
}
+// A physical UPDATE whose complete BEFORE and AFTER row values are equal has
no net delta.
+TEST_F(BlockReaderChangeNextBlockTest, MinDeltaNoOpUpdateIsSkipped) {
+ auto source = make_source_block({
+ {1, 20, 20, 1, 1, ROW_BINLOG_UPDATE},
+ });
+ BlockReader reader;
+ configure_reader(reader, source, 16);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ EXPECT_TRUE(out.empty());
+}
+
+// An unavailable historical row is encoded as an all-NULL BEFORE image. It
must not cancel an
+// all-NULL row inserted later, because the net state changes from absent to
present.
+TEST_F(BlockReaderChangeNextBlockTest, MinDeltaAllNullBeforeImageIsRetained) {
+ auto source = make_all_null_source_block();
+ BlockReader reader;
+ configure_reader(reader, source, 16);
+
+ bool eof = false;
+ Block output = source->clone_empty();
+ ASSERT_TRUE(reader._min_delta_next_block(&output, &eof).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(out_i64(output, OP_IDX, 0), binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out_i64(output, OP_IDX, 1), binlog::STREAM_CHANGE_UPDATE_AFTER);
+ const auto& value_column =
+ assert_cast<const
ColumnNullable&>(*output.get_by_position(VAL_IDX).column);
+ EXPECT_TRUE(value_column.is_null_at(0));
+ EXPECT_TRUE(value_column.is_null_at(1));
+ EXPECT_TRUE(eof);
+}
+
+// A user column may have the same name as another column's generated BEFORE
mirror. Pairing by
+// schema ordinals must still recognize an unchanged complete row and suppress
the UPDATE.
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaNoOpWithCollidingBeforeNameIsSkipped) {
+ auto source = make_colliding_name_source_block(/*after_v=*/10,
/*after_collision=*/20,
+ /*before_v=*/10,
/*before_collision=*/20);
+ auto schema = make_test_tablet_schema({{"v",
FieldType::OLAP_FIELD_TYPE_BIGINT},
+ {"__BEFORE__v__",
FieldType::OLAP_FIELD_TYPE_BIGINT}});
+ BlockReader reader;
+ configure_reader(reader, source, 16, std::move(schema));
+
+ bool eof = false;
+ Block output = source->clone_empty();
+ ASSERT_TRUE(reader._min_delta_next_block(&output, &eof).ok());
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_TRUE(eof);
+}
+
+// compare_at considers +0 and -0 equal, so they represent no net MIN_DELTA
change.
+TEST_F(BlockReaderChangeNextBlockTest, MinDeltaSignedZeroUpdateIsSkipped) {
+ auto source = make_signed_zero_source_block();
+ BlockReader reader;
+ configure_reader(reader, source, 16,
+ make_test_tablet_schema({{"val",
FieldType::OLAP_FIELD_TYPE_FLOAT}}));
+
+ bool eof = false;
+ Block output = source->clone_empty();
+ ASSERT_TRUE(reader._min_delta_next_block(&output, &eof).ok());
+ EXPECT_EQ(output.rows(), 0);
+ EXPECT_TRUE(eof);
+}
+
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaUnsupportedCompareIsRetainedAndCached) {
+ auto compare_calls = std::make_shared<size_t>(0);
+ auto source = make_unsupported_compare_source_block(compare_calls);
+ BlockReader reader;
+ configure_reader(reader, source, 16);
+
+ bool eof = false;
+ Block output = source->clone_empty();
+ ASSERT_TRUE(reader._min_delta_next_block(&output, &eof).ok());
+
+ // Both equal-value updates must be retained. The second key uses the
cached unsupported
+ // capability, so the exception-based capability probe runs exactly once
per reader.
+ ASSERT_EQ(output.rows(), 4);
+ EXPECT_EQ(out_i64(output, OP_IDX, 0), binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out_i64(output, OP_IDX, 1), binlog::STREAM_CHANGE_UPDATE_AFTER);
+ EXPECT_EQ(out_i64(output, OP_IDX, 2), binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out_i64(output, OP_IDX, 3), binlog::STREAM_CHANGE_UPDATE_AFTER);
+ EXPECT_EQ(*compare_calls, 1);
+ EXPECT_TRUE(reader._min_delta_value_compare_unsupported);
+ EXPECT_TRUE(eof);
+}
+
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaUnexpectedCompareExceptionIsNotSuppressed) {
+ auto source = make_source_block({{1, 20, 20, 1, 1, ROW_BINLOG_UPDATE}});
+ auto throwing_type = std::make_shared<DataTypeNothing>();
+ auto compare_calls = std::make_shared<size_t>(0);
+ source->get_by_position(VAL_IDX).column =
+ ThrowOnCompareColumn::create(1, ErrorCode::INTERNAL_ERROR,
compare_calls);
+ source->get_by_position(VAL_IDX).type = throwing_type;
+ source->get_by_position(VAL_IDX + 1).column =
+ ThrowOnCompareColumn::create(1, ErrorCode::INTERNAL_ERROR,
compare_calls);
+ source->get_by_position(VAL_IDX + 1).type = throwing_type;
+ BlockReader reader;
+ configure_reader(reader, source, 16);
+
+ reader._stored_data_columns = source->clone_empty_columns();
+ for (size_t i = 0; i < source->columns(); ++i) {
+
reader._stored_data_columns[i]->insert_from(*source->get_by_position(i).column,
0);
+ }
+
+ try {
+ static_cast<void>(reader._min_delta_values_equal(0));
+ FAIL() << "expected a non-NOT_IMPLEMENTED comparison exception";
+ } catch (const Exception& e) {
+ EXPECT_EQ(e.code(), ErrorCode::INTERNAL_ERROR);
+ }
+ EXPECT_EQ(*compare_calls, 1);
+ EXPECT_FALSE(reader._min_delta_value_compare_unsupported);
+}
+
+// The comparison is between the first BEFORE and last AFTER values, so A -> B
-> A also has no
+// net delta even though neither individual row-binlog UPDATE is a no-op.
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaUpdatesReturningToOriginalAreSkipped) {
+ auto source = make_source_block({
+ {1, 20, 10, 1, 1, ROW_BINLOG_UPDATE},
+ {1, 10, 20, 2, 2, ROW_BINLOG_UPDATE},
+ });
+ BlockReader reader;
+ configure_reader(reader, source, 16);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ EXPECT_TRUE(out.empty());
+}
+
+// Exercise long operation chains where intermediate rows repeatedly change
existence and value.
+// MIN_DELTA must preserve only the net state transition across the whole key
window.
+TEST_F(BlockReaderChangeNextBlockTest, MinDeltaComplexOperationChains) {
+ struct TestCase {
+ std::string_view name;
+ std::vector<Row> rows;
+ std::vector<OutRow> expected;
+ size_t batch_size;
+ };
+ const std::vector<TestCase> test_cases = {
+ {
+ "insert_delete_reinsert_update_back_delete",
+ {
+ {1, 1, 0, 1, 1, ROW_BINLOG_APPEND},
+ {1, 1, 1, 2, 2, ROW_BINLOG_DELETE},
+ {1, 1, 0, 3, 3, ROW_BINLOG_APPEND},
+ {1, 2, 1, 4, 4, ROW_BINLOG_UPDATE},
+ {1, 3, 2, 5, 5, ROW_BINLOG_UPDATE},
+ {1, 1, 3, 6, 6, ROW_BINLOG_UPDATE},
+ {1, 1, 1, 7, 7, ROW_BINLOG_DELETE},
+ },
+ {},
+ 1,
+ },
+ {
+ "existing_row_delete_reinsert_and_return_to_original",
+ {
+ {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE},
+ {1, 2, 2, 2, 2, ROW_BINLOG_DELETE},
+ {1, 2, 0, 3, 3, ROW_BINLOG_APPEND},
+ {1, 3, 2, 4, 4, ROW_BINLOG_UPDATE},
+ {1, 1, 3, 5, 5, ROW_BINLOG_UPDATE},
+ },
+ {},
+ 2,
+ },
+ {
+ "existing_row_delete_reinsert_and_finish_changed",
+ {
+ {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE},
+ {1, 2, 2, 2, 2, ROW_BINLOG_DELETE},
+ {1, 2, 0, 3, 3, ROW_BINLOG_APPEND},
+ {1, 3, 2, 4, 4, ROW_BINLOG_UPDATE},
+ {1, 4, 3, 5, 5, ROW_BINLOG_UPDATE},
+ },
+ {
+ {1, 1, binlog::STREAM_CHANGE_UPDATE_BEFORE},
+ {1, 4, binlog::STREAM_CHANGE_UPDATE_AFTER},
+ },
+ 1,
+ },
+ {
+ "new_row_temporarily_deleted_but_finishes_present",
+ {
+ {1, 1, 0, 1, 1, ROW_BINLOG_APPEND},
+ {1, 2, 1, 2, 2, ROW_BINLOG_UPDATE},
+ {1, 2, 2, 3, 3, ROW_BINLOG_DELETE},
+ {1, 5, 0, 4, 4, ROW_BINLOG_APPEND},
+ {1, 6, 5, 5, 5, ROW_BINLOG_UPDATE},
+ },
+ {
+ {1, 6, binlog::STREAM_CHANGE_INSERT},
+ },
+ 1,
+ },
+ {
+ "existing_row_temporarily_reinserted_but_finishes_deleted",
+ {
+ {1, 2, 1, 1, 1, ROW_BINLOG_UPDATE},
+ {1, 2, 2, 2, 2, ROW_BINLOG_DELETE},
+ {1, 3, 0, 3, 3, ROW_BINLOG_APPEND},
+ {1, 4, 3, 4, 4, ROW_BINLOG_UPDATE},
+ {1, 4, 4, 5, 5, ROW_BINLOG_DELETE},
+ },
+ {
+ {1, 1, binlog::STREAM_CHANGE_DELETE},
+ },
+ 1,
+ },
+ {
+ "delete_reinsert_update_and_return_to_original",
+ {
+ {1, 1, 1, 1, 1, ROW_BINLOG_DELETE},
+ {1, 1, 0, 2, 2, ROW_BINLOG_APPEND},
+ {1, 2, 1, 3, 3, ROW_BINLOG_UPDATE},
+ {1, 1, 2, 4, 4, ROW_BINLOG_UPDATE},
+ },
+ {},
+ 1,
+ },
+ };
+
+ for (const auto& test_case : test_cases) {
+ SCOPED_TRACE(test_case.name);
+ auto source = make_source_block(test_case.rows);
+ BlockReader reader;
+ configure_reader(reader, source, test_case.batch_size);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ expect_out_rows(out, test_case.expected);
+ }
+}
+
+// Build a source block with a second value column that is present in the
physical MIN_DELTA
+// projection but absent from the SQL output projection.
+struct TwoValueRow {
+ int64_t key;
+ int64_t val1;
+ int64_t val2;
+ int64_t before_val1;
+ int64_t before_val2;
+ int64_t tso;
+ int64_t lsn;
+ int64_t op;
+};
+
+std::shared_ptr<Block> make_two_value_source_block(const
std::vector<TwoValueRow>& rows) {
+ auto block = std::make_shared<Block>();
+ auto type = std::make_shared<DataTypeInt64>();
+ auto key_col = ColumnInt64::create();
+ auto val1_col = ColumnInt64::create();
+ auto val2_col = ColumnInt64::create();
+ auto before_val1_col = ColumnInt64::create();
+ auto before_val2_col = ColumnInt64::create();
+ auto tso_col = ColumnInt64::create();
+ auto lsn_col = ColumnInt64::create();
+ auto op_col = ColumnInt64::create();
+ for (const auto& row : rows) {
+ key_col->insert_value(row.key);
+ val1_col->insert_value(row.val1);
+ val2_col->insert_value(row.val2);
+ before_val1_col->insert_value(row.before_val1);
+ before_val2_col->insert_value(row.before_val2);
+ tso_col->insert_value(row.tso);
+ lsn_col->insert_value(row.lsn);
+ op_col->insert_value(row.op);
+ }
+ block->insert({std::move(key_col), type, "key"});
+ block->insert({std::move(val1_col), type, "val"});
+ block->insert({std::move(val2_col), type, "val2"});
+ block->insert({std::move(before_val1_col), type,
binlog::build_before_column_name("val")});
+ block->insert({std::move(before_val2_col), type,
binlog::build_before_column_name("val2")});
+ block->insert({std::move(tso_col), type, BINLOG_TSO_COL});
+ block->insert({std::move(lsn_col), type, BINLOG_LSN_COL});
+ block->insert({std::move(op_col), type, BINLOG_OP_COL});
+ return block;
+}
+
+void configure_two_value_reader(BlockReader& reader, std::shared_ptr<Block>
source,
+ size_t batch_size = 16) {
+ configure_reader(reader, source, batch_size,
+ make_test_tablet_schema({{"val",
FieldType::OLAP_FIELD_TYPE_BIGINT},
+ {"val2",
FieldType::OLAP_FIELD_TYPE_BIGINT}}));
+}
+
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaNoOpUpdateComparesAllValueColumns) {
+ auto source = make_two_value_source_block({
+ {/*key=*/1, /*val1=*/20, /*val2=*/30, /*before_val1=*/20,
/*before_val2=*/30,
+ /*tso=*/1, /*lsn=*/1, ROW_BINLOG_UPDATE},
+ });
+ BlockReader reader;
+ configure_two_value_reader(reader, source);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ EXPECT_TRUE(out.empty());
+}
+
+TEST_F(BlockReaderChangeNextBlockTest,
MinDeltaRetainsChangeInUnprojectedValueColumn) {
+ auto source = make_two_value_source_block({
+ {/*key=*/1, /*val1=*/20, /*val2=*/31, /*before_val1=*/20,
/*before_val2=*/30,
+ /*tso=*/1, /*lsn=*/1, ROW_BINLOG_UPDATE},
+ });
+ BlockReader reader;
+ configure_two_value_reader(reader, source);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ ASSERT_EQ(out.size(), 2);
+ EXPECT_EQ(out[0].op, binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out[1].op, binlog::STREAM_CHANGE_UPDATE_AFTER);
+ EXPECT_EQ(out[0].val, 20);
+ EXPECT_EQ(out[1].val, 20);
+}
+
+// key 1 changes both columns several times and returns to its complete
original row image, so it
+// disappears. key 2 returns only the projected value column to its original
value while the hidden
+// value column remains changed, so its UPDATE pair must survive. batch_size=1
also forces the pair
+// through the pending-row path after the skipped key.
+TEST_F(BlockReaderChangeNextBlockTest, MinDeltaComplexMultiColumnChains) {
+ auto source = make_two_value_source_block({
+ {1, 11, 100, 10, 100, 1, 1, ROW_BINLOG_UPDATE},
+ {1, 11, 101, 11, 100, 2, 2, ROW_BINLOG_UPDATE},
+ {1, 10, 100, 11, 101, 3, 3, ROW_BINLOG_UPDATE},
+ {2, 21, 200, 20, 200, 4, 4, ROW_BINLOG_UPDATE},
+ {2, 20, 201, 21, 200, 5, 5, ROW_BINLOG_UPDATE},
+ });
+ BlockReader reader;
+ configure_two_value_reader(reader, source, /*batch_size=*/1);
+
+ auto out = drain(reader, &BlockReader::_min_delta_next_block);
+ expect_out_rows(out, {
+ {2, 20, binlog::STREAM_CHANGE_UPDATE_BEFORE},
+ {2, 20, binlog::STREAM_CHANGE_UPDATE_AFTER},
+ });
+}
+
// Multiple distinct keys, each in its own group, are folded independently.
TEST_F(BlockReaderChangeNextBlockTest, MinDeltaMultipleKeys) {
auto source = make_source_block({
@@ -440,6 +972,25 @@ TEST_F(BlockReaderChangeNextBlockTest, DetailUpdatePair) {
EXPECT_EQ(out[1].val, 20); // after
}
+TEST_F(BlockReaderChangeNextBlockTest,
DetailUsesOrdinalBeforePairWhenNamesCollide) {
+ auto source = make_colliding_name_source_block(/*after_v=*/11,
/*after_collision=*/20,
+ /*before_v=*/10,
/*before_collision=*/20);
+ auto schema = make_test_tablet_schema({{"v",
FieldType::OLAP_FIELD_TYPE_BIGINT},
+ {"__BEFORE__v__",
FieldType::OLAP_FIELD_TYPE_BIGINT}});
+ BlockReader reader;
+ configure_reader(reader, source, 16, std::move(schema));
+
+ bool eof = false;
+ Block output = source->clone_empty();
+ ASSERT_TRUE(reader._detail_change_next_block(&output, &eof).ok());
+ ASSERT_EQ(output.rows(), 2);
+ EXPECT_EQ(out_i64(output, /*v=*/1, 0), 10);
+ EXPECT_EQ(out_i64(output, /*v=*/1, 1), 11);
+ EXPECT_EQ(out_i64(output, /*op=*/7, 0),
binlog::STREAM_CHANGE_UPDATE_BEFORE);
+ EXPECT_EQ(out_i64(output, /*op=*/7, 1),
binlog::STREAM_CHANGE_UPDATE_AFTER);
+ EXPECT_TRUE(eof);
+}
+
// Mixed ops emitted verbatim in order.
TEST_F(BlockReaderChangeNextBlockTest, DetailMixedOps) {
auto source = make_source_block({
diff --git a/be/test/storage/read_schema_test.cpp
b/be/test/storage/read_schema_test.cpp
index 6c41a522dba..5ee2abd5f0a 100644
--- a/be/test/storage/read_schema_test.cpp
+++ b/be/test/storage/read_schema_test.cpp
@@ -26,6 +26,7 @@
#include "core/block/block.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_struct.h"
+#include "storage/binlog.h"
#include "storage/schema.h"
namespace doris {
@@ -54,6 +55,20 @@ TabletColumnPtr create_struct_column(int32_t unique_id) {
return column;
}
+TabletSchemaSPtr create_row_binlog_schema_with_colliding_names() {
+ auto schema = std::make_shared<TabletSchema>();
+ schema->append_column(*create_int_column(10, "key", true));
+ schema->append_column(*create_int_column(11, "v"));
+ schema->append_column(*create_int_column(12, "__BEFORE__v__"));
+ schema->append_column(*create_int_column(13,
binlog::build_before_column_name("v")));
+ schema->append_column(
+ *create_int_column(14,
binlog::build_before_column_name("__BEFORE__v__")));
+ schema->append_column(*create_int_column(15, BINLOG_TSO_COL));
+ schema->append_column(*create_int_column(16, BINLOG_LSN_COL));
+ schema->append_column(*create_int_column(17, BINLOG_OP_COL));
+ return schema;
+}
+
TEST(ReadSchemaTest, DefaultColumnsAreVisible) {
std::vector<TabletColumnPtr> storage_columns {create_int_column(10, "k",
true),
create_int_column(11,
"dropped"),
@@ -125,5 +140,42 @@ TEST(ReadSchemaTest,
AppendedDroppedColumnDoesNotExtendReadBlock) {
EXPECT_EQ(2, read_schema.create_read_block().columns());
}
+TEST(ReadSchemaTest, RowBinlogMappingsUsePhysicalSchemaOrdinals) {
+ auto tablet_schema = create_row_binlog_schema_with_colliding_names();
+ // Reorder the two AFTER values and their BEFORE companions to verify that
ReadSchema stores
+ // dense read ordinals resolved by unique id, rather than physical tablet
column ids or names.
+ ReadSchema read_schema(project_columns_by_ordinal(
+ tablet_schema->columns(), std::vector<ColumnId> {0, 2, 1, 4, 3, 5,
6, 7}));
+
+ read_schema.init_row_binlog_column_mappings(*tablet_schema);
+
+ EXPECT_TRUE(read_schema.row_binlog_value_pairs_complete());
+ EXPECT_EQ(read_schema.row_binlog_value_column_pairs(),
+ (ReadSchema::RowBinlogValueColumnPairs {{2, 4}, {1, 3}}));
+ EXPECT_EQ(4, read_schema.before_column_ordinal(2));
+ EXPECT_EQ(3, read_schema.before_column_ordinal(1));
+ for (ColumnId ordinal : {0, 3, 4, 5, 6, 7}) {
+ EXPECT_EQ(ordinal, read_schema.before_column_ordinal(ordinal));
+ }
+}
+
+TEST(ReadSchemaTest, MalformedRowBinlogLayoutKeepsConservativeNameMapping) {
+ TabletSchema tablet_schema;
+ tablet_schema.append_column(*create_int_column(10, "key", true));
+ tablet_schema.append_column(*create_int_column(11, "v"));
+ tablet_schema.append_column(*create_int_column(12,
binlog::build_before_column_name("v")));
+ tablet_schema.append_column(*create_int_column(13, "orphan"));
+ tablet_schema.append_column(*create_int_column(14, BINLOG_TSO_COL));
+ tablet_schema.append_column(*create_int_column(15, BINLOG_LSN_COL));
+ tablet_schema.append_column(*create_int_column(16, BINLOG_OP_COL));
+ ReadSchema read_schema(tablet_schema.columns());
+
+ read_schema.init_row_binlog_column_mappings(tablet_schema);
+
+ EXPECT_FALSE(read_schema.row_binlog_value_pairs_complete());
+ EXPECT_TRUE(read_schema.row_binlog_value_column_pairs().empty());
+ EXPECT_EQ(2, read_schema.before_column_ordinal(1));
+}
+
} // namespace
} // namespace doris
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index 8c1395c9923..cf5ac2f5e18 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -3084,10 +3084,10 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
/**
* An {@code @incr} read folds all changes to a row into one record: TSO
is the tie-break column
* ordering them, and the folded result is written back into OP as the
change kind the user
- * sees, so every scan type needs both slots present. {@code SELECT k, v
FROM
- * t@incr("incrementType" = "DETAIL")} also needs the keys to group by
and, when the table keeps
- * historical values, {@code __BEFORE__v__} to report what v was before
the change; APPEND_ONLY
- * never groups and needs neither.
+ * sees, so every scan type needs both slots present. MIN_DELTA compares
the complete first
+ * BEFORE and last AFTER row images, including value columns omitted by
the SQL projection.
+ * DETAIL needs BEFORE images only for projected values. APPEND_ONLY never
groups and needs
+ * neither keys nor BEFORE images.
*/
private void preserveRowBinlogSemanticSlots(OlapScanNode scanNode,
Set<SlotId> requiredSlotIds) {
RowBinlogTableWrapper wrapper = (RowBinlogTableWrapper)
scanNode.getOlapTable();
@@ -3121,12 +3121,14 @@ public class PhysicalPlanTranslator extends
DefaultPlanVisitor<PlanFragment, Pla
// are no before-image storage dependencies to preserve.
return;
}
+ boolean preserveCompleteRow = scanType == TBinlogScanType.MIN_DELTA;
for (SlotDescriptor slot : scanSlots) {
Column column = slot.getColumn();
- if (column == null || column.isKey() ||
!requiredSlotIds.contains(slot.getId())
- || isRowBinlogInternalColumn(column)) {
+ if (column == null || column.isKey() ||
isRowBinlogInternalColumn(column)
+ || (!preserveCompleteRow &&
!requiredSlotIds.contains(slot.getId()))) {
continue;
}
+ preserveStorageSlot(slot, requiredSlotIds);
preserveStorageSlot(slotByName.get(Column.generateBeforeColName(column.getName())),
requiredSlotIds);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
index eb885887ded..e6f1fd92bd0 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslatorTest.java
@@ -237,10 +237,9 @@ public class PhysicalPlanTranslatorTest extends
TestWithFeService {
.collect(Collectors.toList());
Assertions.assertTrue(scanColumns.containsAll(ImmutableList.of(
- "k1", "k2", "v1", Column.generateBeforeColName("v1"),
+ "k1", "k2", "v1", "v2", Column.generateBeforeColName("v1"),
+ Column.generateBeforeColName("v2"),
Column.BINLOG_OPERATION_COL, Column.BINLOG_TSO_COL)));
- Assertions.assertFalse(scanColumns.contains("v2"));
-
Assertions.assertFalse(scanColumns.contains(Column.generateBeforeColName("v2")));
Assertions.assertFalse(scanColumns.contains(Column.BINLOG_LSN_COL));
Assertions.assertTrue(scanNode.getExtraKeyColumnSlotIds().isEmpty());
diff --git a/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
index 801a324eceb..7550bbab228 100644
--- a/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
+++ b/regression-test/data/row_binlog_p0/test_binlog_changes_syntax.out
@@ -241,3 +241,6 @@
1 1,2 2
1 3,4 3
+-- !bitmap_equal_min_delta --
+1 3,4 2
+1 3,4 3
diff --git
a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
index 2ce56948450..c7de72472d5 100644
--- a/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
+++ b/regression-test/suites/row_binlog_p0/test_binlog_changes_syntax.groovy
@@ -584,6 +584,25 @@ suite("test_binlog_changes_syntax", "nonConcurrent") {
"incrementType" = "MIN_DELTA")
ORDER BY __DORIS_BINLOG_OP__
"""
+
+ // BITMAP does not implement compare_at in production. MIN_DELTA must
conservatively
+ // retain an equal-value UPDATE instead of failing the query or
suppressing an unproven
+ // no-op.
+ sleep(1200)
+ def bitmapEqualT0 = incrTimeFormat.format(new Date())
+ sleep(1200)
+ sql "INSERT INTO ${mowBitmapTable} VALUES (1,
BITMAP_FROM_STRING('3,4'))"
+ sql "sync"
+ sleep(1200)
+ def bitmapEqualT1 = incrTimeFormat.format(new Date())
+
+ order_qt_bitmap_equal_min_delta """
+ SELECT id, BITMAP_TO_STRING(b), __DORIS_BINLOG_OP__
+ FROM ${mowBitmapTable}@incr('startTimestamp' = '${bitmapEqualT0}',
+ "endTimestamp" = "${bitmapEqualT1}",
+ "incrementType" = "MIN_DELTA")
+ ORDER BY __DORIS_BINLOG_OP__
+ """
} finally {
sql "DROP DATABASE IF EXISTS test_binlog_changes_syntax_db"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
