From 8bc06472e8e748ebcbb5537f0913cd2e316bb9d1 Mon Sep 17 00:00:00 2001 From: Zhigao Hong Date: Fri, 4 Sep 2026 00:24:17 +0800 Subject: [PATCH] improvement(be): reduce duplicate key memtable memory --- be/benchmark/benchmark_main.cpp | 1 + be/benchmark/benchmark_memtable.hpp | 70 ++++++ be/src/load/memtable/memtable.cpp | 179 +++++++++----- be/src/load/memtable/memtable.h | 15 +- be/test/load/memtable/memtable_sort_test.cpp | 220 ++++++++++++++++++ .../test_dup_key_compact_row_position.out | 8 + .../test_dup_key_compact_row_position.groovy | 72 ++++++ 7 files changed, 508 insertions(+), 57 deletions(-) create mode 100644 be/benchmark/benchmark_memtable.hpp create mode 100644 regression-test/data/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.out create mode 100644 regression-test/suites/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.groovy diff --git a/be/benchmark/benchmark_main.cpp b/be/benchmark/benchmark_main.cpp index cf3780a4e7e3e6..b5466ac403b1ff 100644 --- a/be/benchmark/benchmark_main.cpp +++ b/be/benchmark/benchmark_main.cpp @@ -32,6 +32,7 @@ #include "benchmark_hll_merge.hpp" #include "benchmark_hybrid_set.hpp" #include "benchmark_json_extract.hpp" +#include "benchmark_memtable.hpp" #include "benchmark_variant_segment.hpp" #include "benchmark_zone_map_index.hpp" #include "binary_cast_benchmark.hpp" diff --git a/be/benchmark/benchmark_memtable.hpp b/be/benchmark/benchmark_memtable.hpp new file mode 100644 index 00000000000000..6ac0bae5b09530 --- /dev/null +++ b/be/benchmark/benchmark_memtable.hpp @@ -0,0 +1,70 @@ +// 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. + +#pragma once + +#include + +#include +#include +#include + +#include "load/memtable/memtable.h" + +namespace doris { + +static void BM_DuplicateKeyRowIndexSharedPtr(benchmark::State& state) { + const auto rows = static_cast(state.range(0)); + for (auto _ : state) { + DorisVector> row_index; + row_index.reserve(rows); + for (size_t i = 0; i < rows; ++i) { + row_index.emplace_back(std::make_shared(i)); + } + benchmark::DoNotOptimize(row_index); + benchmark::ClobberMemory(); + } + state.SetItemsProcessed(state.iterations() * static_cast(rows)); + state.counters["estimated_bytes_per_row"] = + benchmark::Counter(sizeof(std::shared_ptr) + sizeof(RowInBlock)); +} + +static void BM_DuplicateKeyRowIndexCompact(benchmark::State& state) { + const auto rows = static_cast(state.range(0)); + for (auto _ : state) { + DorisVector row_index; + row_index.resize(rows); + std::iota(row_index.begin(), row_index.end(), uint32_t {0}); + benchmark::DoNotOptimize(row_index); + benchmark::ClobberMemory(); + } + state.SetItemsProcessed(state.iterations() * static_cast(rows)); + state.counters["estimated_bytes_per_row"] = benchmark::Counter(sizeof(uint32_t)); +} + +BENCHMARK(BM_DuplicateKeyRowIndexSharedPtr) + ->Arg(4096) + ->Arg(65536) + ->Arg(1048576) + ->Unit(benchmark::kMicrosecond); +BENCHMARK(BM_DuplicateKeyRowIndexCompact) + ->Arg(4096) + ->Arg(65536) + ->Arg(1048576) + ->Unit(benchmark::kMicrosecond); + +} // namespace doris diff --git a/be/src/load/memtable/memtable.cpp b/be/src/load/memtable/memtable.cpp index 2a9ac792f3b0a0..7547369c1a8c5c 100644 --- a/be/src/load/memtable/memtable.cpp +++ b/be/src/load/memtable/memtable.cpp @@ -23,6 +23,7 @@ #include #include +#include #include #include @@ -84,6 +85,10 @@ MemTable::MemTable(int64_t tablet_id, std::shared_ptr tablet_schem _init_columns_offset_by_slot_descs(slot_descs, tuple_desc); _row_lsn_col_pos = tablet_schema->row_lsn_col_idx(); // TODO: Support ZOrderComparator in the future + _duplicate_key_row_positions = std::make_unique>(); + if (_need_lsn) { + _duplicate_key_allocated_lsns = std::make_unique>(); + } _row_in_blocks = std::make_unique>>(); _load_mem_limit = MemInfo::mem_limit() * config::load_process_max_memory_limit_percent / 100; } @@ -184,6 +189,8 @@ MemTable::~MemTable() { _arena.clear(true); _vec_row_comparator.reset(); + _duplicate_key_row_positions.reset(); + _duplicate_key_allocated_lsns.reset(); _row_in_blocks.reset(); _agg_functions.clear(); _input_mutable_block.clear(); @@ -266,6 +273,20 @@ Status MemTable::insert(const Block* input_block, const TabletAddRowsPayload& ro auto num_rows = row_idxs.size(); size_t cursor_in_mutableblock = _input_mutable_block.rows(); + constexpr auto max_rows = static_cast(std::numeric_limits::max()); + DORIS_CHECK(num_rows <= max_rows && cursor_in_mutableblock <= max_rows - num_rows); + // Reserve all row metadata before appending columns. After add_rows() succeeds, filling the + // already-reserved vectors cannot leave the block and its permutation out of sync. + if (_keys_type == KeysType::DUP_KEYS) { + RETURN_IF_CATCH_EXCEPTION( + _duplicate_key_row_positions->reserve(cursor_in_mutableblock + num_rows)); + if (_need_lsn) { + RETURN_IF_CATCH_EXCEPTION( + _duplicate_key_allocated_lsns->reserve(cursor_in_mutableblock + num_rows)); + } + } else { + RETURN_IF_CATCH_EXCEPTION(_row_in_blocks->reserve(_row_in_blocks->size() + num_rows)); + } RETURN_IF_ERROR(_input_mutable_block.add_rows(input_block, row_idxs.data(), row_idxs.data() + num_rows, &_column_offset)); if (_need_lsn && _row_lsn_col_pos >= 0) { @@ -274,9 +295,27 @@ Status MemTable::insert(const Block* input_block, const TabletAddRowsPayload& ro _input_mutable_block.get_column_by_position(_row_lsn_col_pos) ->replace_column_data_range(*lsn_column, 0, num_rows, cursor_in_mutableblock); } - for (int i = 0; i < num_rows; i++) { - _row_in_blocks->emplace_back(std::make_shared( - cursor_in_mutableblock + i, _need_lsn ? allocated_lsns[i] : 0)); + if (_keys_type == KeysType::DUP_KEYS) { + // The mutable block appends rows contiguously, so the initial permutation for this batch + // is exactly [cursor, cursor + num_rows). Sorting later mutates only this compact vector. + auto& positions = *_duplicate_key_row_positions; + DORIS_CHECK(positions.size() == cursor_in_mutableblock); + DORIS_CHECK(cursor_in_mutableblock + num_rows <= positions.capacity()); + positions.resize(cursor_in_mutableblock + num_rows); + std::iota(positions.begin() + cursor_in_mutableblock, positions.end(), + static_cast(cursor_in_mutableblock)); + if (_need_lsn) { + DORIS_CHECK(_duplicate_key_allocated_lsns->size() == cursor_in_mutableblock); + DORIS_CHECK(cursor_in_mutableblock + num_rows <= + _duplicate_key_allocated_lsns->capacity()); + _duplicate_key_allocated_lsns->insert(_duplicate_key_allocated_lsns->end(), + allocated_lsns.begin(), allocated_lsns.end()); + } + } else { + for (size_t i = 0; i < num_rows; i++) { + _row_in_blocks->emplace_back(std::make_shared( + cursor_in_mutableblock + i, _need_lsn ? allocated_lsns[i] : 0)); + } } _stat.raw_rows += num_rows; @@ -383,62 +422,89 @@ void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlo } Status MemTable::_put_into_output(Block& in_block) { SCOPED_RAW_TIMER(&_stat.put_into_output_ns); - DorisVector row_pos_vec; DCHECK(in_block.rows() <= std::numeric_limits::max()); - row_pos_vec.reserve(in_block.rows()); if (_need_lsn) { _output_allocated_lsns->reserve(_output_allocated_lsns->size() + in_block.rows()); } - for (int i = 0; i < _row_in_blocks->size(); i++) { - row_pos_vec.emplace_back((*_row_in_blocks)[i]->_row_pos); - _append_output_allocated_lsn((*_row_in_blocks)[i].get()); + if (_keys_type == KeysType::DUP_KEYS) { + DCHECK_EQ(_duplicate_key_row_positions->size(), in_block.rows()); + if (_need_lsn) { + // LSNs are stored in physical input order. Apply the row permutation here so each + // sorted output row keeps the LSN assigned to its original position. + DCHECK_EQ(_duplicate_key_allocated_lsns->size(), in_block.rows()); + for (const auto row_pos : *_duplicate_key_row_positions) { + _output_allocated_lsns->emplace_back((*_duplicate_key_allocated_lsns)[row_pos]); + } + } + return _output_mutable_block.add_rows( + &in_block, _duplicate_key_row_positions->data(), + _duplicate_key_row_positions->data() + _duplicate_key_row_positions->size()); + } + + DorisVector row_pos_vec; + row_pos_vec.reserve(in_block.rows()); + for (const auto& row : *_row_in_blocks) { + row_pos_vec.emplace_back(row->_row_pos); + _append_output_allocated_lsn(row.get()); } return _output_mutable_block.add_rows(&in_block, row_pos_vec.data(), - row_pos_vec.data() + in_block.rows()); + row_pos_vec.data() + row_pos_vec.size()); } -size_t MemTable::_sort() { - SCOPED_RAW_TIMER(&_stat.sort_ns); - _stat.sort_times++; +template +size_t MemTable::_sort_rows(DorisVector& rows, RowPosGetter&& get_row_pos) { + // Both representations sort by the same block rows. RowPosGetter keeps DUP_KEYS compact while + // allowing UNIQUE_KEYS and AGG_KEYS to retain their aggregation-bearing RowInBlock objects. size_t same_keys_num = 0; - // sort new rows - Tie tie = Tie(_last_sorted_pos, _row_in_blocks->size()); + Tie tie(_last_sorted_pos, rows.size()); for (size_t i = 0; i < _tablet_schema->num_key_columns(); i++) { - auto cmp = [&](RowInBlock* lhs, RowInBlock* rhs) -> int { - return _input_mutable_block.compare_one_column(lhs->_row_pos, rhs->_row_pos, i, -1); + auto cmp = [&](const RowRef& lhs, const RowRef& rhs) -> int { + return _input_mutable_block.compare_one_column(get_row_pos(lhs), get_row_pos(rhs), i, + -1); }; - _sort_one_column(*_row_in_blocks, tie, cmp); + _sort_one_column(rows, tie, cmp); } - bool is_dup = (_keys_type == KeysType::DUP_KEYS); - // sort extra round by _row_pos to make the sort stable + + const bool is_dup = (_keys_type == KeysType::DUP_KEYS); + // Preserve the existing deterministic tie order: newer DUP rows precede older rows, whereas + // aggregation models retain increasing row position before equal rows are merged. auto iter = tie.iter(); while (iter.next()) { - pdqsort(std::next(_row_in_blocks->begin(), iter.left()), - std::next(_row_in_blocks->begin(), iter.right()), - [&is_dup](const std::shared_ptr& lhs, - const std::shared_ptr& rhs) -> bool { - return is_dup ? lhs->_row_pos > rhs->_row_pos : lhs->_row_pos < rhs->_row_pos; + pdqsort(std::next(rows.begin(), iter.left()), std::next(rows.begin(), iter.right()), + [&](const RowRef& lhs, const RowRef& rhs) -> bool { + return is_dup ? get_row_pos(lhs) > get_row_pos(rhs) + : get_row_pos(lhs) < get_row_pos(rhs); }); same_keys_num += iter.right() - iter.left(); } - // merge new rows and old rows - _vec_row_comparator->set_block(&_input_mutable_block); - auto cmp_func = [this, is_dup, &same_keys_num](const std::shared_ptr& l, - const std::shared_ptr& r) -> bool { - auto value = (*(this->_vec_row_comparator))(l.get(), r.get()); + + auto cmp = [&](const RowRef& lhs, const RowRef& rhs) -> bool { + const auto lhs_pos = get_row_pos(lhs); + const auto rhs_pos = get_row_pos(rhs); + const auto value = _input_mutable_block.compare_at( + lhs_pos, rhs_pos, _tablet_schema->num_key_columns(), _input_mutable_block, -1); if (value == 0) { same_keys_num++; - return is_dup ? l->_row_pos > r->_row_pos : l->_row_pos < r->_row_pos; - } else { - return value < 0; + return is_dup ? lhs_pos > rhs_pos : lhs_pos < rhs_pos; } + return value < 0; }; - auto new_row_it = std::next(_row_in_blocks->begin(), _last_sorted_pos); - std::inplace_merge(_row_in_blocks->begin(), new_row_it, _row_in_blocks->end(), cmp_func); - _last_sorted_pos = _row_in_blocks->size(); + auto new_row_it = std::next(rows.begin(), _last_sorted_pos); + std::inplace_merge(rows.begin(), new_row_it, rows.end(), cmp); + _last_sorted_pos = rows.size(); return same_keys_num; } +size_t MemTable::_sort() { + SCOPED_RAW_TIMER(&_stat.sort_ns); + _stat.sort_times++; + if (_keys_type == KeysType::DUP_KEYS) { + return _sort_rows(*_duplicate_key_row_positions, [](uint32_t row_pos) { return row_pos; }); + } + return _sort_rows(*_row_in_blocks, + [](const std::shared_ptr& row) { return row->_row_pos; }); +} + Status MemTable::_sort_by_cluster_keys() { SCOPED_RAW_TIMER(&_stat.sort_ns); _stat.sort_times++; @@ -470,7 +536,8 @@ Status MemTable::_sort_by_cluster_keys() { return Status::InternalError("could not find cluster key column with unique_id=" + std::to_string(cid) + " in tablet schema"); } - auto cmp = [&](const RowInBlock* lhs, const RowInBlock* rhs) -> int { + auto cmp = [&](const std::shared_ptr& lhs, + const std::shared_ptr& rhs) -> int { return mutable_block.compare_one_column(lhs->_row_pos, rhs->_row_pos, index, -1); }; _sort_one_column(row_in_blocks, tie, cmp); @@ -502,16 +569,16 @@ Status MemTable::_sort_by_cluster_keys() { row_pos_vec.data() + in_block.rows(), &column_offset); } -void MemTable::_sort_one_column(DorisVector>& row_in_blocks, Tie& tie, - std::function cmp) { +template +void MemTable::_sort_one_column(DorisVector& rows, Tie& tie, Comparator&& cmp) { auto iter = tie.iter(); while (iter.next()) { - pdqsort(std::next(row_in_blocks.begin(), static_cast(iter.left())), - std::next(row_in_blocks.begin(), static_cast(iter.right())), - [&cmp](auto lhs, auto rhs) -> bool { return cmp(lhs.get(), rhs.get()) < 0; }); + pdqsort(std::next(rows.begin(), static_cast(iter.left())), + std::next(rows.begin(), static_cast(iter.right())), + [&](const RowRef& lhs, const RowRef& rhs) -> bool { return cmp(lhs, rhs) < 0; }); tie[iter.left()] = 0; for (auto i = iter.left() + 1; i < iter.right(); i++) { - tie[i] = (cmp(row_in_blocks[i - 1].get(), row_in_blocks[i].get()) == 0); + tie[i] = (cmp(rows[i - 1], rows[i]) == 0); } } } @@ -815,20 +882,24 @@ size_t MemTable::get_flush_reserve_memory_size() const { Status MemTable::_to_block(std::unique_ptr* res) { _output_allocated_lsns = std::make_shared>(); + if (_keys_type == KeysType::DUP_KEYS && _tablet_schema->num_key_columns() == 0) { + // There is no key to sort. Keep the physical block and LSN sidecar in insertion order and + // avoid building Tie state or sorting a permutation that would be discarded immediately. + _output_mutable_block.swap(_input_mutable_block); + if (_need_lsn) { + DCHECK_EQ(_duplicate_key_allocated_lsns->size(), _output_mutable_block.rows()); + _output_allocated_lsns->assign(_duplicate_key_allocated_lsns->begin(), + _duplicate_key_allocated_lsns->end()); + } + _input_mutable_block.clear(); + *res = Block::create_unique(_output_mutable_block.to_block()); + return Status::OK(); + } + size_t same_keys_num = _sort(); if (_keys_type == KeysType::DUP_KEYS || same_keys_num == 0) { - if (_keys_type == KeysType::DUP_KEYS && _tablet_schema->num_key_columns() == 0) { - _output_mutable_block.swap(_input_mutable_block); - if (_need_lsn) { - _output_allocated_lsns->reserve(_row_in_blocks->size()); - for (const auto& row : *_row_in_blocks) { - _append_output_allocated_lsn(row.get()); - } - } - } else { - Block in_block = _input_mutable_block.to_block(); - RETURN_IF_ERROR(_put_into_output(in_block)); - } + Block in_block = _input_mutable_block.to_block(); + RETURN_IF_ERROR(_put_into_output(in_block)); } else { (_skip_bitmap_col_idx == -1) ? _aggregate() : _aggregate(); } diff --git a/be/src/load/memtable/memtable.h b/be/src/load/memtable/memtable.h index 4228da6b928663..8565fbc4a97a30 100644 --- a/be/src/load/memtable/memtable.h +++ b/be/src/load/memtable/memtable.h @@ -22,7 +22,6 @@ #include #include -#include #include #include @@ -275,8 +274,10 @@ class MemTable { //return number of same keys size_t _sort(); Status _sort_by_cluster_keys(); - void _sort_one_column(DorisVector>& row_in_blocks, Tie& tie, - std::function cmp); + template + size_t _sort_rows(DorisVector& rows, RowPosGetter&& get_row_pos); + template + void _sort_one_column(DorisVector& rows, Tie& tie, Comparator&& cmp); template void _finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, int row_pos); void _init_row_for_agg(RowInBlock* row, MutableBlock& mutable_block); @@ -302,6 +303,14 @@ class MemTable { std::vector _agg_functions; std::vector _offsets_of_aggregate_states; size_t _total_size_of_aggregate_states; + // DUP_KEYS only needs a permutation of source row positions for sorting. A uint32_t vector + // avoids one shared_ptr, one RowInBlock, and one allocation/control block per input. + std::unique_ptr> _duplicate_key_row_positions; + // Optional row-binlog LSNs remain indexed by the original DUP_KEYS row position. Output uses + // the sorted position vector to apply the identical permutation to rows and their LSNs. + std::unique_ptr> _duplicate_key_allocated_lsns; + // Only UNIQUE_KEYS and AGG_KEYS use RowInBlock. Those models still need per-row aggregation + // state and, when enabled, the allocated LSN stored in RowInBlock. std::unique_ptr>> _row_in_blocks; size_t _num_columns; diff --git a/be/test/load/memtable/memtable_sort_test.cpp b/be/test/load/memtable/memtable_sort_test.cpp index 53e92e3c4bb2be..041e5b8a0ea8f2 100644 --- a/be/test/load/memtable/memtable_sort_test.cpp +++ b/be/test/load/memtable/memtable_sort_test.cpp @@ -17,7 +17,21 @@ #include +#include +#include +#include +#include + +#include "common/object_pool.h" +#include "core/data_type/data_type_array.h" +#include "core/data_type/data_type_number.h" +#include "core/data_type/data_type_string.h" +#include "core/field.h" +#include "load/delta_writer/delta_writer_context.h" #include "load/memtable/memtable.h" +#include "runtime/descriptors.h" +#include "runtime/thread_context.h" +#include "testutil/desc_tbl_builder.h" namespace doris { @@ -80,4 +94,210 @@ TEST_F(MemTableSortTest, Tie) { EXPECT_FALSE(it3.next()); } +class DuplicateKeyMemTableTest : public testing::Test { +protected: + using TestRow = std::tuple; + + void SetUp() override { + _smallint_type = std::make_shared(); + _string_type = std::make_shared(); + _array_type = std::make_shared(_string_type); + + DescriptorTblBuilder builder(&_object_pool); + builder.declare_tuple() << std::make_tuple(_smallint_type, "shard_num") + << std::make_tuple(_string_type, "label_name") + << std::make_tuple(_array_type, "label_value"); + _desc_tbl = builder.build(); + _tuple_desc = _desc_tbl->get_tuple_descriptor(0); + + _tablet_schema = create_tablet_schema(true); + _resource_ctx = thread_context()->resource_ctx(); + _memtable = create_memtable(false); + } + + std::unique_ptr create_memtable(bool need_lsn) const { + return create_memtable(_tablet_schema, need_lsn); + } + + std::unique_ptr create_memtable(const TabletSchemaSPtr& schema, bool need_lsn) const { + return std::make_unique(1, schema, &_tuple_desc->slots(), _tuple_desc, false, + nullptr, _resource_ctx, need_lsn); + } + + TabletSchemaSPtr create_tablet_schema(bool with_keys) const { + TabletSchemaPB schema_pb; + schema_pb.set_keys_type(KeysType::DUP_KEYS); + schema_pb.set_num_short_key_columns(with_keys ? 2 : 0); + schema_pb.set_num_rows_per_row_block(1024); + schema_pb.set_next_column_unique_id(4); + auto schema = std::make_shared(); + schema->init_from_pb(schema_pb); + schema->append_column( + create_column(1, "shard_num", FieldType::OLAP_FIELD_TYPE_SMALLINT, with_keys, 2)); + schema->append_column( + create_column(2, "label_name", FieldType::OLAP_FIELD_TYPE_VARCHAR, with_keys, 64)); + auto array_column = + create_column(3, "label_value", FieldType::OLAP_FIELD_TYPE_ARRAY, false, 0); + auto array_item_column = + create_column(-1, "item", FieldType::OLAP_FIELD_TYPE_VARCHAR, false, 128); + array_column.add_sub_column(array_item_column); + schema->append_column(std::move(array_column)); + return schema; + } + + static TabletColumn create_column(int32_t unique_id, std::string name, FieldType type, + bool is_key, int32_t length) { + TabletColumn column; + column.set_unique_id(unique_id); + column.set_name(std::move(name)); + column.set_type(type); + column.set_is_key(is_key); + column.set_is_nullable(false); + column.set_length(length); + column.set_index_length(length); + return column; + } + + Block create_block(const std::vector& rows) const { + Block block; + for (const auto* slot : _tuple_desc->slots()) { + block.insert({slot->get_empty_mutable_column(), slot->type(), slot->col_name()}); + } + auto columns_guard = block.mutate_columns_scoped(); + auto& columns = columns_guard.mutable_columns(); + for (const auto& [shard_num, label_name, label_value] : rows) { + columns[0]->insert_data(reinterpret_cast(&shard_num), sizeof(shard_num)); + columns[1]->insert_data(label_name.data(), label_name.size()); + columns[2]->insert(Field::create_field(label_value)); + } + return block; + } + + static Array make_array(std::initializer_list values) { + Array array; + array.reserve(values.size()); + for (const auto& value : values) { + array.emplace_back(Field::create_field(value)); + } + return array; + } + + void expect_row(const Block& block, size_t row, int16_t shard_num, std::string_view label_name, + const Array& label_value) const { + Field actual_shard_num; + Field actual_label_name; + Field actual_label_value; + block.get_by_position(0).column->get(row, actual_shard_num); + block.get_by_position(1).column->get(row, actual_label_name); + block.get_by_position(2).column->get(row, actual_label_value); + EXPECT_EQ(actual_shard_num, Field::create_field(shard_num)); + EXPECT_EQ(actual_label_name, Field::create_field(std::string(label_name))); + EXPECT_EQ(actual_label_value, Field::create_field(label_value)); + } + + ObjectPool _object_pool; + DescriptorTbl* _desc_tbl = nullptr; + TupleDescriptor* _tuple_desc = nullptr; + DataTypePtr _smallint_type; + DataTypePtr _string_type; + DataTypePtr _array_type; + TabletSchemaSPtr _tablet_schema; + std::shared_ptr _resource_ctx; + std::unique_ptr _memtable; +}; + +// Verify incremental sorting across insert batches, including the newest-first tie order for equal +// duplicate keys and correct movement of a non-POD ARRAY value column. +TEST_F(DuplicateKeyMemTableTest, SortMultipleBatchesWithDuplicateKeysAndArrays) { + auto first = create_block({{2, "b", make_array({"first"})}, + {1, "a", make_array({"old"})}, + {1, "z", make_array({})}}); + ASSERT_TRUE(_memtable->insert(&first, TabletAddRowsPayload {.row_idxs = {0, 1, 2}}).ok()); + + auto second = create_block({{1, "a", make_array({"new", "value"})}, + {0, "z", make_array({"lowest"})}, + {2, "b", make_array({"latest"})}}); + ASSERT_TRUE(_memtable->insert(&second, TabletAddRowsPayload {.row_idxs = {0, 1, 2}}).ok()); + + std::unique_ptr output; + ASSERT_TRUE(_memtable->to_block(&output).ok()); + ASSERT_EQ(output->rows(), 6); + expect_row(*output, 0, 0, "z", make_array({"lowest"})); + expect_row(*output, 1, 1, "a", make_array({"new", "value"})); + expect_row(*output, 2, 1, "a", make_array({"old"})); + expect_row(*output, 3, 1, "z", make_array({})); + expect_row(*output, 4, 2, "b", make_array({"latest"})); + expect_row(*output, 5, 2, "b", make_array({"first"})); +} + +// An empty insert must not create row metadata or disturb the next non-empty batch. +TEST_F(DuplicateKeyMemTableTest, EmptyBatchDoesNotAffectSort) { + auto empty = create_block({}); + ASSERT_TRUE(_memtable->insert(&empty, TabletAddRowsPayload {}).ok()); + EXPECT_TRUE(_memtable->empty()); + + auto data = create_block({{2, "b", make_array({"second"})}, {1, "a", make_array({"first"})}}); + ASSERT_TRUE(_memtable->insert(&data, TabletAddRowsPayload {.row_idxs = {0, 1}}).ok()); + + std::unique_ptr output; + ASSERT_TRUE(_memtable->to_block(&output).ok()); + ASSERT_EQ(output->rows(), 2); + expect_row(*output, 0, 1, "a", make_array({"first"})); + expect_row(*output, 1, 2, "b", make_array({"second"})); +} + +// Exercise the offset-sensitive append path with LSNs split across two batches, then verify that +// the row permutation and LSN permutation remain identical after sorting. +TEST_F(DuplicateKeyMemTableTest, PreserveAllocatedLsnsAfterSort) { + auto memtable = create_memtable(true); + auto first = create_block({{2, "b", make_array({"second"})}, {1, "a", make_array({"first"})}}); + ASSERT_TRUE(memtable->insert(&first, TabletAddRowsPayload {.row_idxs = {0, 1}, + .allocated_lsns = {20, 10}}) + .ok()); + + auto second = create_block({{2, "b", make_array({"latest"})}}); + ASSERT_TRUE(memtable->insert(&second, + TabletAddRowsPayload {.row_idxs = {0}, .allocated_lsns = {21}}) + .ok()); + + std::unique_ptr output; + ASSERT_TRUE(memtable->to_block(&output).ok()); + ASSERT_EQ(output->rows(), 3); + EXPECT_EQ(*memtable->allocated_lsns(), (std::vector {10, 21, 20})); +} + +// A zero-key duplicate table bypasses sorting and swaps the input block directly. Its LSN sidecar +// must therefore remain in physical insertion order, and the sort counter must stay unchanged. +TEST_F(DuplicateKeyMemTableTest, PreserveAllocatedLsnsWithoutKeyColumns) { + auto memtable = create_memtable(create_tablet_schema(false), true); + auto data = create_block({{2, "b", make_array({"second"})}, + {1, "a", make_array({"first"})}, + {2, "b", make_array({"latest"})}}); + ASSERT_TRUE(memtable->insert(&data, TabletAddRowsPayload {.row_idxs = {0, 1, 2}, + .allocated_lsns = {20, 10, 21}}) + .ok()); + + std::unique_ptr output; + ASSERT_TRUE(memtable->to_block(&output).ok()); + ASSERT_EQ(output->rows(), 3); + expect_row(*output, 0, 2, "b", make_array({"second"})); + expect_row(*output, 1, 1, "a", make_array({"first"})); + expect_row(*output, 2, 2, "b", make_array({"latest"})); + EXPECT_EQ(*memtable->allocated_lsns(), (std::vector {20, 10, 21})); + EXPECT_EQ(memtable->stat().sort_times.load(), 0); +} + +// Check the production insert path, rather than a test-only helper: DUP_KEYS populates the compact +// contiguous permutation and never allocates RowInBlock objects. +TEST_F(DuplicateKeyMemTableTest, InsertUsesCompactContiguousRowPositions) { + auto data = create_block({{2, "b", make_array({"second"})}, + {1, "a", make_array({"first"})}, + {2, "b", make_array({"latest"})}}); + ASSERT_TRUE(_memtable->insert(&data, TabletAddRowsPayload {.row_idxs = {0, 1, 2}}).ok()); + + EXPECT_EQ(*_memtable->_duplicate_key_row_positions, (DorisVector {0, 1, 2})); + EXPECT_TRUE(_memtable->_row_in_blocks->empty()); + EXPECT_LT(sizeof(uint32_t) * 8, sizeof(std::shared_ptr) + sizeof(RowInBlock)); +} + } // namespace doris diff --git a/regression-test/data/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.out b/regression-test/data/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.out new file mode 100644 index 00000000000000..78ffc86f9a1c13 --- /dev/null +++ b/regression-test/data/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.out @@ -0,0 +1,8 @@ +-- This file is automatically generated. You should know what you did if you want to edit this +-- !dup_key_array -- +0 z ["lowest"] 1 2026-08-15 +1 a ["new", "value"] 11 2026-08-15 +1 a ["old"] 10 2026-08-15 +1 z [] 30 2026-08-15 +2 b ["first"] 20 2026-08-15 +2 b ["latest"] 21 2026-08-15 diff --git a/regression-test/suites/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.groovy b/regression-test/suites/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.groovy new file mode 100644 index 00000000000000..4e05aca866c15d --- /dev/null +++ b/regression-test/suites/data_model_p0/duplicate/storage/test_dup_key_compact_row_position.groovy @@ -0,0 +1,72 @@ +// 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_dup_key_compact_row_position") { + sql "DROP TABLE IF EXISTS test_dup_key_compact_row_position" + sql "DROP TABLE IF EXISTS test_dup_key_compact_row_position_src" + + sql """ + CREATE TABLE test_dup_key_compact_row_position_src ( + shard_num SMALLINT NOT NULL, + label_name VARCHAR(64) NOT NULL, + label_value ARRAY NOT NULL, + uid INT NOT NULL, + dt DATE NOT NULL + ) + DUPLICATE KEY(shard_num, label_name) + DISTRIBUTED BY HASH(shard_num) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + + sql """ + CREATE TABLE test_dup_key_compact_row_position ( + shard_num SMALLINT NOT NULL, + label_name VARCHAR(64) NOT NULL, + label_value ARRAY NOT NULL, + uid INT NOT NULL, + dt DATE NOT NULL + ) + DUPLICATE KEY(shard_num, label_name) + DISTRIBUTED BY HASH(shard_num) BUCKETS 1 + PROPERTIES ("replication_num" = "1") + """ + + sql """ + INSERT INTO test_dup_key_compact_row_position_src VALUES + (2, 'b', ['first'], 20, '2026-08-15'), + (1, 'a', ['old'], 10, '2026-08-15'), + (1, 'z', [], 30, '2026-08-15') + """ + sql """ + INSERT INTO test_dup_key_compact_row_position_src VALUES + (1, 'a', ['new', 'value'], 11, '2026-08-15'), + (0, 'z', ['lowest'], 1, '2026-08-15'), + (2, 'b', ['latest'], 21, '2026-08-15') + """ + + sql """ + INSERT INTO test_dup_key_compact_row_position + SELECT shard_num, label_name, label_value, uid, dt + FROM test_dup_key_compact_row_position_src + """ + + order_qt_dup_key_array """ + SELECT shard_num, label_name, label_value, uid, dt + FROM test_dup_key_compact_row_position + ORDER BY shard_num, label_name, uid + """ +}