diff --git a/be/src/storage/segment/row_binlog_segment_writer.cpp b/be/src/storage/segment/row_binlog_segment_writer.cpp index 807330b94960d2..555ca500d0a108 100644 --- a/be/src/storage/segment/row_binlog_segment_writer.cpp +++ b/be/src/storage/segment/row_binlog_segment_writer.cpp @@ -139,13 +139,13 @@ Status RowBinlogSegmentWriter::append_block(const Block* block, size_t row_pos, : std::vector(); std::vector row_binlog_partial_cids = partial_cids; if (is_partial_update) { - if (block->columns() <= source_schema->num_key_columns() || + if (block->columns() < source_schema->num_key_columns() || block->columns() >= 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(), _tablet_schema->num_key_columns(), - _tablet_schema->num_columns())); + block->columns(), source_schema->num_key_columns(), + source_schema->num_columns())); } // Partial update lists source cids. Row-binlog writes visible columns and hidden key diff --git a/be/test/olap/rowset/group_rowset_writer_test.cpp b/be/test/olap/rowset/group_rowset_writer_test.cpp index d1bc178bf8a575..caff31d01a1a1a 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 #include +#include #include #include #include #include #include +#include #include #include "common/config.h" @@ -90,6 +92,7 @@ class GroupRowsetWriterTest : public testing::Test { _request.tablet_schema.columns[2].__set_visible(false); _request.tablet_schema.columns[5].__set_visible(false); _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 +157,70 @@ class GroupRowsetWriterTest : public testing::Test { return block; } + std::shared_ptr create_mow_context() const { + return std::make_shared(1, 1, std::make_shared(), + std::vector {}, nullptr); + } + + Result> create_partial_update_row_binlog_writer( + const std::shared_ptr& partial_update_info, size_t num_rows, + const std::shared_ptr& 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>(); + 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> create_partial_update_group_writer( + const std::shared_ptr& 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(); + group_writer->set_data_writer( + std::shared_ptr(std::move(data_writer_result.value()))); + group_writer->set_row_binlog_writer( + std::shared_ptr(std::move(row_binlog_writer_result.value()))); + return group_writer; + } + Status create_group_rowset_writer(std::unique_ptr* group_writer, RowsetId* data_rowset_id, size_t num_rows) { RowsetWriterContext data_context; @@ -305,32 +372,8 @@ TEST_F(GroupRowsetWriterTest, partialUpdateSkipsHiddenNonKeyColumns) { EXPECT_EQ((std::vector {0, 1, 2, 3}), partial_update_info->update_cids); EXPECT_EQ((std::vector {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(1, 1, std::make_shared(), - std::vector {}, 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.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>(); - 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()); @@ -372,4 +415,106 @@ TEST_F(GroupRowsetWriterTest, partialUpdateSkipsHiddenNonKeyColumns) { EXPECT_TRUE(status.is()) << status; } +TEST_F(GroupRowsetWriterTest, keyOnlyFixedPartialUpdatePreservesNarrowBlock) { + auto partial_update_info = std::make_shared(); + 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 {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, 2> {{{1, 1001}, {4, 1004}}}) { + columns[0]->insert(Field::create_field(key)); + columns[1]->insert(Field::create_field(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 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 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()); + EXPECT_EQ(expected_hidden_key, + (*output_block.get_by_position(1).column)[row].get()); + EXPECT_EQ(7, (*output_block.get_by_position(2).column)[row].get()); + 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()); + EXPECT_EQ(ROW_BINLOG_APPEND, + (*output_block.get_by_position(6).column)[row].get()); + } + + Block eof_block = row_binlog_schema->create_block(); + status = rowset_reader->next_batch(&eof_block); + EXPECT_TRUE(status.is()) << status; +} + +TEST_F(GroupRowsetWriterTest, keyOnlyFixedPartialUpdateRejectsInvalidWidths) { + auto partial_update_info = std::make_shared(); + 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(1)); + auto status = too_narrow_writer->flush_single_block(&too_narrow); + EXPECT_TRUE(status.is()) << 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()) << 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/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 00000000000000..60a3e9e01ee7d6 --- /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()) +}