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 203923ca9ac [fix](binlog) Allow key-only partial updates with row
binlog (#66840)
203923ca9ac is described below
commit 203923ca9acbcaa8dbdbbf4b25d048a07c7c9f8c
Author: Luwei <[email protected]>
AuthorDate: Thu Aug 20 23:05:13 2026 +0800
[fix](binlog) Allow key-only partial updates with row binlog (#66840)
Related PR: #63110
Problem Summary: ROW binlog rejected fixed partial-update blocks when
the input width exactly matched the source schema key count, even though
key-only updates are valid. Align the lower-bound check with the regular
partial-update validation, report source-schema dimensions, and classify
malformed widths as invalid arguments. Add BE coverage for the original
narrow block and invalid boundaries, plus regression coverage for APPEND
and ERROR new-key policies, historical values, and failed-transaction
atomicity.
### Release note
Allow key-only fixed partial updates on UNIQUE KEY merge-on-write tables
with ROW binlog enabled.
---
be/src/storage/transform/row_binlog_derive.cpp | 4 +-
be/test/olap/rowset/group_rowset_writer_test.cpp | 198 ++++++++++++++++++---
.../storage/transform/row_binlog_derive_test.cpp | 7 +-
...test_row_binlog_partial_update_only_keys.groovy | 114 ++++++++++++
4 files changed, 291 insertions(+), 32 deletions(-)
diff --git a/be/src/storage/transform/row_binlog_derive.cpp
b/be/src/storage/transform/row_binlog_derive.cpp
index a12763e2add..fbb5f176e09 100644
--- a/be/src/storage/transform/row_binlog_derive.cpp
+++ b/be/src/storage/transform/row_binlog_derive.cpp
@@ -346,9 +346,9 @@ Status
MowRowBinlogDeriveStage::derive(TransformExecContext& ctx, Block* block,
cfg.source.partial_update_info->is_fixed_partial_update() &&
is_source_direct_write;
if (is_partial_update) {
- if (block->columns() <= c.source_schema->num_key_columns() ||
+ if (block->columns() < c.source_schema->num_key_columns() ||
block->columns() >= c.source_schema->num_columns()) {
- return Status::InternalError(fmt::format(
+ return Status::InvalidArgument(fmt::format(
"illegal partial update block columns: {}, num key
columns: {}, total "
"schema columns: {}",
block->columns(), c.source_schema->num_key_columns(),
diff --git a/be/test/olap/rowset/group_rowset_writer_test.cpp
b/be/test/olap/rowset/group_rowset_writer_test.cpp
index 96054020b66..e9d8f85a93e 100644
--- a/be/test/olap/rowset/group_rowset_writer_test.cpp
+++ b/be/test/olap/rowset/group_rowset_writer_test.cpp
@@ -21,11 +21,13 @@
#include <gtest/gtest.h>
#include <unistd.h>
+#include <array>
#include <chrono>
#include <memory>
#include <set>
#include <string>
#include <thread>
+#include <utility>
#include <vector>
#include "common/config.h"
@@ -89,7 +91,9 @@ protected:
_request.tablet_schema.columns[1].__set_visible(false);
_request.tablet_schema.columns[2].__set_visible(false);
_request.tablet_schema.columns[5].__set_visible(false);
+ _request.tablet_schema.columns[5].__set_default_value("0");
_request.tablet_schema.columns[2].__set_is_allow_null(true);
+ _request.tablet_schema.columns[3].__set_default_value("7");
_request.tablet_schema.columns[4].__set_is_allow_null(true);
_request.__set_enable_unique_key_merge_on_write(true);
testutil::enable_row_binlog(&_request);
@@ -154,6 +158,70 @@ protected:
return block;
}
+ std::shared_ptr<MowContext> create_mow_context() const {
+ return std::make_shared<MowContext>(1, 1,
std::make_shared<RowsetIdUnorderedSet>(),
+ std::vector<RowsetSharedPtr> {},
nullptr);
+ }
+
+ Result<std::unique_ptr<RowsetWriter>>
create_partial_update_row_binlog_writer(
+ const std::shared_ptr<PartialUpdateInfo>& partial_update_info,
size_t num_rows,
+ const std::shared_ptr<MowContext>& mow_context) {
+ RowsetWriterContext row_binlog_context;
+ row_binlog_context.tablet = _row_binlog_tablet;
+ row_binlog_context.tablet_schema = _row_binlog_tablet->tablet_schema();
+ row_binlog_context.rowset_state = PREPARED;
+ row_binlog_context.segments_overlap = NONOVERLAPPING;
+ row_binlog_context.max_rows_per_segment = 1024;
+ row_binlog_context.write_type = DataWriteType::TYPE_DIRECT;
+ row_binlog_context.partial_update_info = partial_update_info;
+ row_binlog_context.mow_context = mow_context;
+ row_binlog_context.write_binlog_opt().enable = true;
+ auto& binlog_options =
row_binlog_context.write_binlog_opt().write_binlog_config();
+ binlog_options.source.base_tablet = _tablet;
+ binlog_options.source.tablet_schema = _tablet->tablet_schema();
+ binlog_options.source.partial_update_info = partial_update_info;
+ binlog_options.source.mow_context = mow_context;
+ binlog_options.source.source_write_type = DataWriteType::TYPE_DIRECT;
+
+ auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
+ lsn_buffer->append_range_for_test(1000, num_rows);
+ auto lsn_ids = std::make_shared<std::vector<int64_t>>();
+ RETURN_IF_ERROR_RESULT(allocate_binlog_lsn(lsn_buffer, num_rows,
*lsn_ids));
+ binlog_options.insert_seg_lsn(0, lsn_ids);
+ return _row_binlog_tablet->create_rowset_writer(row_binlog_context,
false);
+ }
+
+ Result<std::unique_ptr<GroupRowsetWriter>>
create_partial_update_group_writer(
+ const std::shared_ptr<PartialUpdateInfo>& partial_update_info,
size_t num_rows) {
+ auto mow_context = create_mow_context();
+ RowsetWriterContext data_context;
+ data_context.tablet = _tablet;
+ data_context.tablet_schema = _tablet->tablet_schema();
+ data_context.rowset_state = PREPARED;
+ data_context.segments_overlap = OVERLAPPING;
+ data_context.max_rows_per_segment = 1024;
+ data_context.write_type = DataWriteType::TYPE_DIRECT;
+ data_context.partial_update_info = partial_update_info;
+ data_context.mow_context = mow_context;
+ auto data_writer_result = _tablet->create_rowset_writer(data_context,
false);
+ if (!data_writer_result.has_value()) {
+ return unexpected(data_writer_result.error());
+ }
+
+ auto row_binlog_writer_result =
+ create_partial_update_row_binlog_writer(partial_update_info,
num_rows, mow_context);
+ if (!row_binlog_writer_result.has_value()) {
+ return unexpected(row_binlog_writer_result.error());
+ }
+
+ auto group_writer = std::make_unique<GroupRowsetWriter>();
+ group_writer->set_data_writer(
+
std::shared_ptr<RowsetWriter>(std::move(data_writer_result.value())));
+ group_writer->set_row_binlog_writer(
+
std::shared_ptr<RowsetWriter>(std::move(row_binlog_writer_result.value())));
+ return group_writer;
+ }
+
Status create_group_rowset_writer(std::unique_ptr<GroupRowsetWriter>*
group_writer,
RowsetId* data_rowset_id, size_t
num_rows) {
RowsetWriterContext data_context;
@@ -306,33 +374,8 @@ TEST_F(GroupRowsetWriterTest,
partialUpdateSkipsHiddenNonKeyColumns) {
EXPECT_EQ((std::vector<uint32_t> {0, 1, 2, 3}),
partial_update_info->update_cids);
EXPECT_EQ((std::vector<uint32_t> {4, 5}),
partial_update_info->missing_cids);
- RowsetWriterContext row_binlog_context;
- row_binlog_context.tablet = _row_binlog_tablet;
- row_binlog_context.tablet_schema = _row_binlog_tablet->tablet_schema();
- row_binlog_context.rowset_state = PREPARED;
- row_binlog_context.segments_overlap = NONOVERLAPPING;
- row_binlog_context.max_rows_per_segment = 1024;
- row_binlog_context.write_type = DataWriteType::TYPE_DIRECT;
- row_binlog_context.partial_update_info = partial_update_info;
- row_binlog_context.mow_context =
- std::make_shared<MowContext>(1, 1,
std::make_shared<RowsetIdUnorderedSet>(),
- std::vector<RowsetSharedPtr> {},
nullptr);
- row_binlog_context.write_binlog_opt().enable = true;
- auto& binlog_options =
row_binlog_context.write_binlog_opt().write_binlog_config();
- binlog_options.source.tablet_schema = _tablet->tablet_schema();
- binlog_options.source.base_tablet = _tablet;
- binlog_options.source.partial_update_info = partial_update_info;
- binlog_options.source.mow_context = row_binlog_context.mow_context;
- binlog_options.source.source_write_type = DataWriteType::TYPE_DIRECT;
-
- auto lsn_buffer = AutoIncIDBuffer::create_shared(1, 1,
kBinlogLsnAutoIncId);
- lsn_buffer->append_range_for_test(1000, 1);
- auto lsn_ids = std::make_shared<std::vector<int64_t>>();
- ASSERT_TRUE(allocate_binlog_lsn(lsn_buffer, 1, *lsn_ids).ok());
- binlog_options.insert_seg_lsn(0, lsn_ids);
-
auto row_binlog_writer_res =
- _row_binlog_tablet->create_rowset_writer(row_binlog_context,
false);
+ create_partial_update_row_binlog_writer(partial_update_info, 1,
create_mow_context());
ASSERT_TRUE(row_binlog_writer_res.has_value());
auto row_binlog_writer = std::move(row_binlog_writer_res.value());
@@ -374,4 +417,107 @@ TEST_F(GroupRowsetWriterTest,
partialUpdateSkipsHiddenNonKeyColumns) {
EXPECT_TRUE(status.is<ErrorCode::END_OF_FILE>()) << status;
}
+// NOLINTNEXTLINE(readability-function-cognitive-complexity) -- GTest
assertions inflate it.
+TEST_F(GroupRowsetWriterTest, keyOnlyFixedPartialUpdatePreservesNarrowBlock) {
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ ASSERT_TRUE(partial_update_info
+ ->init(_tablet->tablet_id(), 1,
*_tablet->tablet_schema(),
+ UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS,
+ PartialUpdateNewRowPolicyPB::APPEND,
+ {"k1", "__DORIS_TEST_HIDDEN_KEY__"}, false, 0,
0, "", "")
+ .ok());
+ EXPECT_EQ((std::vector<uint32_t> {0, 1}),
partial_update_info->update_cids);
+ ASSERT_EQ(_tablet->tablet_schema()->num_key_columns(),
partial_update_info->update_cids.size());
+
+ auto group_writer_result =
create_partial_update_group_writer(partial_update_info, 2);
+ ASSERT_TRUE(group_writer_result.has_value()) <<
group_writer_result.error();
+ auto group_writer = std::move(group_writer_result.value());
+
+ Block block =
_tablet->tablet_schema()->create_block_by_cids(partial_update_info->update_cids);
+ {
+ auto columns_guard = block.mutate_columns_scoped();
+ auto& columns = columns_guard.mutable_columns();
+ for (const auto& [key, hidden_key] :
+ std::array<std::pair<int32_t, int64_t>, 2> {{{1, 1001}, {4,
1004}}}) {
+
columns[0]->insert(Field::create_field<PrimitiveType::TYPE_INT>(key));
+
columns[1]->insert(Field::create_field<PrimitiveType::TYPE_BIGINT>(hidden_key));
+ }
+ }
+ ASSERT_EQ(block.columns(), _tablet->tablet_schema()->num_key_columns());
+ auto status = group_writer->flush_single_block(&block);
+ ASSERT_TRUE(status.ok()) << status;
+ EXPECT_EQ(block.columns(), _tablet->tablet_schema()->num_key_columns());
+
+ std::vector<RowsetSharedPtr> rowsets;
+ status = group_writer->build_rowsets(rowsets);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(2, rowsets.size());
+
+ const auto& row_binlog_schema = _row_binlog_tablet->tablet_schema();
+ std::vector<uint32_t> return_columns {0, 1, 2, 3, 4, 5, 6};
+ RowsetReaderContext reader_context;
+ reader_context.tablet_schema = row_binlog_schema;
+ reader_context.need_ordered_result = false;
+ reader_context.return_columns = &return_columns;
+
+ RowsetReaderSharedPtr rowset_reader;
+ ASSERT_TRUE(rowsets[1]->create_reader(&rowset_reader).ok());
+ ASSERT_TRUE(rowset_reader->init(&reader_context).ok());
+
+ Block output_block = row_binlog_schema->create_block();
+ status = rowset_reader->next_batch(&output_block);
+ ASSERT_TRUE(status.ok()) << status;
+ ASSERT_EQ(2, output_block.rows());
+ for (size_t row = 0; row < output_block.rows(); ++row) {
+ const int32_t expected_key = row == 0 ? 1 : 4;
+ const int64_t expected_hidden_key = row == 0 ? 1001 : 1004;
+ EXPECT_EQ(expected_key,
(*output_block.get_by_position(0).column)[row].get<TYPE_INT>());
+ EXPECT_EQ(expected_hidden_key,
+
(*output_block.get_by_position(1).column)[row].get<TYPE_BIGINT>());
+ EXPECT_EQ(7,
(*output_block.get_by_position(2).column)[row].get<TYPE_INT>());
+ EXPECT_TRUE(output_block.get_by_position(3).column->is_null_at(row));
+ EXPECT_EQ(1000 + row,
(*output_block.get_by_position(5).column)[row].get<TYPE_BIGINT>());
+ EXPECT_EQ(ROW_BINLOG_APPEND,
+
(*output_block.get_by_position(6).column)[row].get<TYPE_BIGINT>());
+ }
+
+ Block eof_block = row_binlog_schema->create_block();
+ status = rowset_reader->next_batch(&eof_block);
+ EXPECT_TRUE(status.is<ErrorCode::END_OF_FILE>()) << status;
+}
+
+TEST_F(GroupRowsetWriterTest, keyOnlyFixedPartialUpdateRejectsInvalidWidths) {
+ auto partial_update_info = std::make_shared<PartialUpdateInfo>();
+ ASSERT_TRUE(partial_update_info
+ ->init(_tablet->tablet_id(), 1,
*_tablet->tablet_schema(),
+ UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS,
+ PartialUpdateNewRowPolicyPB::APPEND,
+ {"k1", "__DORIS_TEST_HIDDEN_KEY__"}, false, 0,
0, "", "")
+ .ok());
+
+ auto too_narrow_writer_result =
+ create_partial_update_row_binlog_writer(partial_update_info, 1,
create_mow_context());
+ ASSERT_TRUE(too_narrow_writer_result.has_value()) <<
too_narrow_writer_result.error();
+ auto too_narrow_writer = std::move(too_narrow_writer_result.value());
+ Block too_narrow = _tablet->tablet_schema()->create_block_by_cids({0});
+ too_narrow.get_by_position(0).column->assert_mutable()->insert(
+ Field::create_field<PrimitiveType::TYPE_INT>(1));
+ auto status = too_narrow_writer->flush_single_block(&too_narrow);
+ EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
+ EXPECT_NE(std::string::npos,
+ status.to_string().find("illegal partial update block columns:
1"));
+ EXPECT_NE(std::string::npos, status.to_string().find("total schema
columns: 6"));
+
+ auto full_width_writer_result =
+ create_partial_update_row_binlog_writer(partial_update_info, 1,
create_mow_context());
+ ASSERT_TRUE(full_width_writer_result.has_value()) <<
full_width_writer_result.error();
+ auto full_width_writer = std::move(full_width_writer_result.value());
+ Block full_width = create_block(10, 1);
+ status = full_width_writer->flush_single_block(&full_width);
+ EXPECT_TRUE(status.is<ErrorCode::INVALID_ARGUMENT>()) << status;
+ EXPECT_NE(std::string::npos,
+ status.to_string().find("illegal partial update block columns:
6"));
+ EXPECT_NE(std::string::npos, status.to_string().find("total schema
columns: 6"));
+}
+
} // namespace doris
diff --git a/be/test/storage/transform/row_binlog_derive_test.cpp
b/be/test/storage/transform/row_binlog_derive_test.cpp
index d740d7d464d..43e7ea079ed 100644
--- a/be/test/storage/transform/row_binlog_derive_test.cpp
+++ b/be/test/storage/transform/row_binlog_derive_test.cpp
@@ -1446,7 +1446,7 @@ TEST_F(RowBinlogDeriveTest,
MowRejectsFlexiblePartialUpdate) {
}
// N5/N6 (B10): the fixed-PU MoW derive rejects a block that is too narrow
-// (<= num_key_columns) or too wide (>= num_columns).
+// (< num_key_columns) or too wide (>= num_columns).
TEST_F(RowBinlogDeriveTest, MowFixedPartialUpdateRejectsBadWidth) {
auto binlog_tablet =
create_binlog_tablets(/*tablet_id=*/8601, TKeysType::DUP_KEYS,
/*mow=*/false).binlog;
@@ -1488,12 +1488,11 @@ TEST_F(RowBinlogDeriveTest,
MowFixedPartialUpdateRejectsBadWidth) {
return ctx;
};
- // too narrow: only the key column (1 <= num_key_columns 1)
+ // too narrow: no columns (0 < num_key_columns 1)
{
cfg.insert_seg_lsn(0, make_seg_lsn(1));
TransformExecContext ctx = make_ctx();
- Block block = source_schema->create_block_by_cids({0});
- block.get_by_position(0).column->assert_mutable()->insert_default();
+ Block block;
auto st = chain.apply(ctx, &block);
EXPECT_FALSE(st.ok());
EXPECT_NE(st.to_string().find("illegal partial update block columns"),
std::string::npos)
diff --git
a/regression-test/suites/row_binlog_p0/test_row_binlog_partial_update_only_keys.groovy
b/regression-test/suites/row_binlog_p0/test_row_binlog_partial_update_only_keys.groovy
new file mode 100644
index 00000000000..60a3e9e01ee
--- /dev/null
+++
b/regression-test/suites/row_binlog_p0/test_row_binlog_partial_update_only_keys.groovy
@@ -0,0 +1,114 @@
+// 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_row_binlog_partial_update_only_keys", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS test_row_binlog_partial_update_only_keys FORCE"
+
+ sql """
+ CREATE TABLE test_row_binlog_partial_update_only_keys (
+ k1 INT,
+ k2 INT,
+ v_default INT NOT NULL DEFAULT "7",
+ v_nullable STRING NULL
+ )
+ UNIQUE KEY(k1, k2)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """
+ INSERT INTO test_row_binlog_partial_update_only_keys VALUES
+ (1, 10, 100, 'old-1'),
+ (2, 20, 200, 'old-2')
+ """
+
+ sql "SET enable_unique_key_partial_update = true"
+ sql "SET partial_update_new_key_behavior = 'APPEND'"
+ sql """
+ INSERT INTO test_row_binlog_partial_update_only_keys(k1, k2) VALUES
+ (1, 10),
+ (3, 30)
+ """
+
+ def queryTable = {
+ sql """
+ SELECT k1, k2, v_default, v_nullable
+ FROM test_row_binlog_partial_update_only_keys
+ ORDER BY k1, k2
+ """
+ }
+ def queryBinlog = {
+ sql """
+ SELECT __DORIS_BINLOG_OP__ AS op,
+ k1,
+ k2,
+ v_default,
+ v_nullable,
+ __BEFORE__v_default__,
+ __BEFORE__v_nullable__
+ FROM binlog("table" = "test_row_binlog_partial_update_only_keys")
+ ORDER BY __DORIS_BINLOG_TSO__, __DORIS_BINLOG_LSN__
+ """
+ }
+
+ def expectedTable = [
+ [1, 10, 100, "old-1"],
+ [2, 20, 200, "old-2"],
+ [3, 30, 7, null]
+ ]
+ assertEquals(expectedTable, queryTable())
+
+ def expectedAppendBinlog = [
+ [0L, 1, 10, 100, "old-1", null, null],
+ [0L, 2, 20, 200, "old-2", null, null],
+ [1L, 1, 10, 100, "old-1", 100, "old-1"],
+ [0L, 3, 30, 7, null, null, null]
+ ]
+ assertEquals(expectedAppendBinlog, queryBinlog())
+
+ sql "SET partial_update_new_key_behavior = 'ERROR'"
+ sql """
+ INSERT INTO test_row_binlog_partial_update_only_keys(k1, k2) VALUES
+ (1, 10),
+ (2, 20)
+ """
+
+ assertEquals(expectedTable, queryTable())
+ def expectedErrorExistingBinlog = expectedAppendBinlog + [
+ [1L, 1, 10, 100, "old-1", 100, "old-1"],
+ [1L, 2, 20, 200, "old-2", 200, "old-2"]
+ ]
+ assertEquals(expectedErrorExistingBinlog, queryBinlog())
+
+ test {
+ sql """
+ INSERT INTO test_row_binlog_partial_update_only_keys(k1, k2) VALUES
+ (1, 10),
+ (4, 40)
+ """
+ exception "[E-7003]Can't append new rows in partial update when
partial_update_new_key_behavior is ERROR"
+ }
+
+ assertEquals(expectedTable, queryTable())
+ assertEquals(expectedErrorExistingBinlog, queryBinlog())
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
