This is an automated email from the ASF dual-hosted git repository. yiguolei pushed a commit to branch branch-4.2 in repository https://gitbox.apache.org/repos/asf/doris.git
commit bc2f21d30cd69ee0f2facb2c11b5f67b9fbcd3a5 Author: Jamie <[email protected]> AuthorDate: Thu Sep 10 09:24:14 2026 +0800 branch-4.1: [fix](compaction) Release exhausted segment contexts during vertical compaction #65963 (#67599) ### What problem does this PR solve? Issue Number: None Related PR: #65963 Problem Summary: Backport #65963 to `branch-4.1`. Vertical compaction kept exhausted segment iterator contexts and dictionary/block memory alive until all value-column groups finished. This change tracks final row-source consumption and releases exhausted contexts immediately, while adding active/peak context diagnostics and regression coverage. The backport adapts the storage/iterator versus olap/vec source layout and preserves the target branch APIs. Branches without `use_insert_order_when_same` retain their existing ordering behavior. ### Release note Release exhausted segment contexts during vertical compaction to bound retained memory. ### Check List (For Author) - Test: Static validation - Clang Format 16 `--dry-run --Werror`: passed for all changed C++ files - `git diff --check`: passed - Changed-file scope, latest base, and cherry-pick provenance verified - Focused local compilation was attempted on branch-4.1, but the available generated Thrift/Protobuf headers did not match this branch; hosted CI is pending - Behavior changed: Yes - Releases a segment context after its final row-source reference is consumed - Adds active and peak segment-context diagnostics - Does this need documentation: No --- be/src/storage/iterator/vertical_block_reader.cpp | 9 +- be/src/storage/iterator/vertical_block_reader.h | 7 +- .../storage/iterator/vertical_merge_iterator.cpp | 109 ++++++-- be/src/storage/iterator/vertical_merge_iterator.h | 53 +++- be/src/storage/merger.cpp | 16 +- be/src/storage/merger.h | 3 +- .../compaction/vertical_compaction_test.cpp | 285 +++++++++++++++++++++ 7 files changed, 444 insertions(+), 38 deletions(-) diff --git a/be/src/storage/iterator/vertical_block_reader.cpp b/be/src/storage/iterator/vertical_block_reader.cpp index dbc4d7f55ba..1fe1274dac5 100644 --- a/be/src/storage/iterator/vertical_block_reader.cpp +++ b/be/src/storage/iterator/vertical_block_reader.cpp @@ -150,16 +150,17 @@ Status VerticalBlockReader::_init_collect_iter(const ReaderParams& read_params, _vcollect_iter = new_vertical_fifo_merge_iterator( std::move(*segment_iters_ptr), iterator_init_flag, rowset_ids, ori_return_col_size, read_params.tablet->keys_type(), seq_col_idx, - _row_sources_buffer); + _row_sources_buffer, _context_stats); } else { _vcollect_iter = new_vertical_heap_merge_iterator( std::move(*segment_iters_ptr), iterator_init_flag, rowset_ids, ori_return_col_size, read_params.tablet->keys_type(), seq_col_idx, - _row_sources_buffer, read_params.key_group_cluster_key_idxes); + _row_sources_buffer, _context_stats, read_params.key_group_cluster_key_idxes); } } else { - _vcollect_iter = new_vertical_mask_merge_iterator(std::move(*segment_iters_ptr), - ori_return_col_size, _row_sources_buffer); + _vcollect_iter = + new_vertical_mask_merge_iterator(std::move(*segment_iters_ptr), ori_return_col_size, + _row_sources_buffer, _context_stats); } // init collect iterator StorageReadOptions opts; diff --git a/be/src/storage/iterator/vertical_block_reader.h b/be/src/storage/iterator/vertical_block_reader.h index 80e36434f08..baba2a215f8 100644 --- a/be/src/storage/iterator/vertical_block_reader.h +++ b/be/src/storage/iterator/vertical_block_reader.h @@ -44,11 +44,13 @@ struct RowsetId; class RowSourcesBuffer; struct RowBatch; +struct VerticalCompactionContextStats; class VerticalBlockReader final : public TabletReader { public: - VerticalBlockReader(RowSourcesBuffer* row_sources_buffer) - : _row_sources_buffer(row_sources_buffer) { + VerticalBlockReader(RowSourcesBuffer* row_sources_buffer, + VerticalCompactionContextStats* context_stats = nullptr) + : _row_sources_buffer(row_sources_buffer), _context_stats(context_stats) { _id = nextId++; } @@ -113,6 +115,7 @@ private: Status (VerticalBlockReader::*_next_block_func)(Block* block, bool* eof) = nullptr; RowSourcesBuffer* _row_sources_buffer; + VerticalCompactionContextStats* _context_stats; ColumnPtr _delete_filter_column; // for agg mode diff --git a/be/src/storage/iterator/vertical_merge_iterator.cpp b/be/src/storage/iterator/vertical_merge_iterator.cpp index 1df2b6c4bc4..856bd78226e 100644 --- a/be/src/storage/iterator/vertical_merge_iterator.cpp +++ b/be/src/storage/iterator/vertical_merge_iterator.cpp @@ -17,6 +17,7 @@ #include "storage/iterator/vertical_merge_iterator.h" +#include <bvar/bvar.h> #include <fcntl.h> #include <gen_cpp/olap_file.pb.h> #include <stdlib.h> @@ -41,6 +42,13 @@ namespace doris { #include "common/compile_check_begin.h" using namespace ErrorCode; +namespace { + +bvar::Adder<int64_t> g_vertical_compaction_active_segment_contexts( + "vertical_compaction_active_segment_contexts"); + +} // namespace + // -------------- row source ---------------// RowSource::RowSource(uint16_t source_num, bool agg_flag) { _data = (source_num & SOURCE_FLAG) | (source_num & AGG_FLAG); @@ -85,8 +93,14 @@ Status RowSourcesBuffer::append(const std::vector<RowSource>& row_sources) { _reset_buffer(); } } + uint64_t source_position = _total_size; for (const auto& source : row_sources) { _buffer.push_back(source.data()); + auto source_num = source.get_source_num(); + if (source_num >= _last_source_positions.size()) { + _last_source_positions.resize(source_num + 1); + } + _last_source_positions[source_num] = source_position++; } _total_size += row_sources.size(); return Status::OK(); @@ -94,6 +108,7 @@ Status RowSourcesBuffer::append(const std::vector<RowSource>& row_sources) { Status RowSourcesBuffer::seek_to_begin() { _buf_idx = 0; + _read_index = 0; if (_fd > 0) { auto offset = lseek(_fd, 0, SEEK_SET); if (offset != 0) { @@ -274,6 +289,45 @@ Status RowSourcesBuffer::_deserialize() { } // ---------- vertical merge iterator context ----------// +VerticalMergeIteratorContext::~VerticalMergeIteratorContext() { + release_resources(); +} + +void VerticalMergeIteratorContext::release_resources() { + _valid = false; + _iter.reset(); + _block.reset(); + // Unlike the physical EOF fallback, source exhaustion proves that this context will not + // access these blocks again. External IteratorRowRef/RowBatch owners keep them alive through + // shared_ptr, so dropping the context-owned references here is intentional. + _block_list.clear(); + _mark_inactive(); +} + +void VerticalMergeIteratorContext::_mark_active() { + DCHECK(!_is_active_context_counted); + _is_active_context_counted = true; + g_vertical_compaction_active_segment_contexts << 1; + if (_context_stats != nullptr) { + ++_context_stats->active_segment_contexts; + _context_stats->active_segment_contexts_peak = + std::max(_context_stats->active_segment_contexts_peak, + _context_stats->active_segment_contexts); + } +} + +void VerticalMergeIteratorContext::_mark_inactive() { + if (!_is_active_context_counted) { + return; + } + _is_active_context_counted = false; + g_vertical_compaction_active_segment_contexts << -1; + if (_context_stats != nullptr) { + DCHECK_GT(_context_stats->active_segment_contexts, 0); + --_context_stats->active_segment_contexts; + } +} + Status VerticalMergeIteratorContext::block_reset(const std::shared_ptr<Block>& block) { if (!block->columns()) { const Schema& schema = _iter->schema(); @@ -376,6 +430,7 @@ Status VerticalMergeIteratorContext::init(const StorageReadOptions& opts, sample_info->rows += rows(); } if (valid()) { + _mark_active(); RETURN_IF_ERROR(advance()); } _inited = true; @@ -460,6 +515,9 @@ Status VerticalMergeIteratorContext::_load_next_block() { // the column iterator in the segment iterator will hold the dictionary. // Release the segment iterator to free up the dictionary. _iter.reset(); + if (_block_list.empty()) { + _mark_inactive(); + } return Status::OK(); } else { return st; @@ -582,7 +640,7 @@ Status VerticalHeapMergeIterator::init(const StorageReadOptions& opts, auto& iter = _origin_iters[seg_order]; auto ctx = std::make_unique<VerticalMergeIteratorContext>( std::move(iter), _rowset_ids[seg_order], _ori_return_cols, seg_order, _seq_col_idx, - _key_group_cluster_key_idxes); + _context_stats, _key_group_cluster_key_idxes); _ori_iter_ctx.push_back(std::move(ctx)); } _origin_iters.clear(); @@ -648,9 +706,9 @@ Status VerticalFifoMergeIterator::next_batch(Block* block) { cur_order++) { auto& next_iter = _origin_iters[cur_order]; std::unique_ptr<VerticalMergeIteratorContext> next_ctx( - new VerticalMergeIteratorContext(std::move(next_iter), - _rowset_ids[cur_order], _ori_return_cols, - cur_order, _seq_col_idx)); + new VerticalMergeIteratorContext( + std::move(next_iter), _rowset_ids[cur_order], _ori_return_cols, + cur_order, _seq_col_idx, _context_stats, std::vector<uint32_t> {})); RETURN_IF_ERROR(next_ctx->init(_opts)); if (next_ctx->valid()) { _cur_iter_ctx.swap(next_ctx); @@ -688,9 +746,9 @@ Status VerticalFifoMergeIterator::init(const StorageReadOptions& opts, // will not be pushed into heap, we should init next one util we find a valid iter // so this rowset can work in heap for (auto& iter : _origin_iters) { - std::unique_ptr<VerticalMergeIteratorContext> ctx( - new VerticalMergeIteratorContext(std::move(iter), _rowset_ids[seg_order], - _ori_return_cols, seg_order, _seq_col_idx)); + std::unique_ptr<VerticalMergeIteratorContext> ctx(new VerticalMergeIteratorContext( + std::move(iter), _rowset_ids[seg_order], _ori_return_cols, seg_order, _seq_col_idx, + _context_stats)); RETURN_IF_ERROR(ctx->init(opts, sample_info)); if (!ctx->valid()) { ++seg_order; @@ -718,6 +776,14 @@ Status VerticalMaskMergeIterator::check_all_iter_finished() { } return Status::OK(); } + +void VerticalMaskMergeIterator::consume_row_sources(uint16_t order, size_t count) { + _row_sources_buf->advance(count); + if (_row_sources_buf->is_source_exhausted(order)) { + _origin_iter_ctx[order]->release_resources(); + } +} + Status VerticalMaskMergeIterator::next_row(IteratorRowRef* ref) { DCHECK(_row_sources_buf); auto st = _row_sources_buf->has_remaining(); @@ -745,7 +811,7 @@ Status VerticalMaskMergeIterator::next_row(IteratorRowRef* ref) { } ctx->set_is_first_row(false); - _row_sources_buf->advance(); + consume_row_sources(order); return Status::OK(); } RETURN_IF_ERROR(ctx->advance()); @@ -755,7 +821,7 @@ Status VerticalMaskMergeIterator::next_row(IteratorRowRef* ref) { _filtered_rows++; } - _row_sources_buf->advance(); + consume_row_sources(order); return Status::OK(); } @@ -778,15 +844,16 @@ Status VerticalMaskMergeIterator::unique_key_next_row(IteratorRowRef* ref) { // Except first row, we call advance first and than get cur row ctx->set_cur_row_ref(ref); ctx->set_is_first_row(false); - _row_sources_buf->advance(); + consume_row_sources(order); return Status::OK(); } RETURN_IF_ERROR(ctx->advance()); - _row_sources_buf->advance(); if (!row_source.agg_flag()) { ctx->set_cur_row_ref(ref); + consume_row_sources(order); return Status::OK(); } + consume_row_sources(order); _filtered_rows++; st = _row_sources_buf->has_remaining(); } @@ -843,7 +910,7 @@ Status VerticalMaskMergeIterator::unique_key_next_batch(std::vector<RowBatch>* b // If current row has agg_flag=true, skip it (single row) if (row_source.agg_flag()) { - _row_sources_buf->advance(); + consume_row_sources(order); _filtered_rows++; continue; } @@ -864,7 +931,7 @@ Status VerticalMaskMergeIterator::unique_key_next_batch(std::vector<RowBatch>* b RETURN_IF_ERROR(ctx->advance_by(run_count - 1)); } - _row_sources_buf->advance(run_count); + consume_row_sources(order, run_count); // Try to merge into current batch or create new batch if (current_block == block && start_row == batch_start + batch_count) { @@ -941,8 +1008,8 @@ Status VerticalMaskMergeIterator::init(const StorageReadOptions& opts, RowsetId rs_id; for (auto& iter : _origin_iters) { - auto ctx = std::make_unique<VerticalMergeIteratorContext>(std::move(iter), rs_id, - _ori_return_cols, -1, -1); + auto ctx = std::make_unique<VerticalMergeIteratorContext>( + std::move(iter), rs_id, _ori_return_cols, -1, -1, _context_stats); _origin_iter_ctx.push_back(std::move(ctx)); } _origin_iters.clear(); @@ -957,26 +1024,28 @@ std::shared_ptr<RowwiseIterator> new_vertical_heap_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>& iterator_init_flag, const std::vector<RowsetId>& rowset_ids, size_t ori_return_cols, KeysType keys_type, uint32_t seq_col_idx, RowSourcesBuffer* row_sources, + VerticalCompactionContextStats* context_stats, std::vector<uint32_t> key_group_cluster_key_idxes) { return std::make_shared<VerticalHeapMergeIterator>( std::move(inputs), iterator_init_flag, rowset_ids, ori_return_cols, keys_type, - seq_col_idx, row_sources, key_group_cluster_key_idxes); + seq_col_idx, row_sources, context_stats, key_group_cluster_key_idxes); } std::shared_ptr<RowwiseIterator> new_vertical_fifo_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>& iterator_init_flag, const std::vector<RowsetId>& rowset_ids, size_t ori_return_cols, KeysType keys_type, - uint32_t seq_col_idx, RowSourcesBuffer* row_sources) { + uint32_t seq_col_idx, RowSourcesBuffer* row_sources, + VerticalCompactionContextStats* context_stats) { return std::make_shared<VerticalFifoMergeIterator>(std::move(inputs), iterator_init_flag, rowset_ids, ori_return_cols, keys_type, - seq_col_idx, row_sources); + seq_col_idx, row_sources, context_stats); } std::shared_ptr<RowwiseIterator> new_vertical_mask_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, size_t ori_return_cols, - RowSourcesBuffer* row_sources) { + RowSourcesBuffer* row_sources, VerticalCompactionContextStats* context_stats) { return std::make_shared<VerticalMaskMergeIterator>(std::move(inputs), ori_return_cols, - row_sources); + row_sources, context_stats); } #include "common/compile_check_end.h" diff --git a/be/src/storage/iterator/vertical_merge_iterator.h b/be/src/storage/iterator/vertical_merge_iterator.h index 865399c4747..dbb01c63678 100644 --- a/be/src/storage/iterator/vertical_merge_iterator.h +++ b/be/src/storage/iterator/vertical_merge_iterator.h @@ -42,6 +42,12 @@ namespace doris { #include "common/compile_check_begin.h" enum KeysType : int; +// Owned by one vertical_merge_rowsets invocation and updated synchronously by its iterators. +struct VerticalCompactionContextStats { + int64_t active_segment_contexts = 0; + int64_t active_segment_contexts_peak = 0; +}; + // Row source represent row location in multi-segments // use a uint16_t to store info // the lower 15 bits means segment_id in segment pool, and the higher 1 bits means agg flag. @@ -102,11 +108,17 @@ public: void advance(int64_t step = 1) { DCHECK(_buf_idx + step <= _buffer.size()); _buf_idx += step; + _read_index += step; } uint64_t buf_idx() const { return _buf_idx; } uint64_t total_size() const { return _total_size; } uint64_t buffered_size() { return _buffer.size(); } + bool is_source_exhausted(uint16_t source) const { + DCHECK(source < _last_source_positions.size()); + return source < _last_source_positions.size() && + _read_index > _last_source_positions[source]; + } void set_agg_flag(uint64_t index, bool agg); bool get_agg_flag(uint64_t index); @@ -142,6 +154,8 @@ private: int _fd = -1; PaddedPODArray<UInt16> _buffer; uint64_t _total_size = 0; + uint64_t _read_index = 0; + std::vector<uint64_t> _last_source_positions; }; // --------------- VerticalMergeIteratorContext ------------- // @@ -150,6 +164,7 @@ class VerticalMergeIteratorContext { public: VerticalMergeIteratorContext(RowwiseIteratorUPtr&& iter, RowsetId rowset_id, size_t ori_return_cols, uint32_t order, uint32_t seq_col_idx, + VerticalCompactionContextStats* context_stats, std::vector<uint32_t> key_group_cluster_key_idxes = {}) : _iter(std::move(iter)), _rowset_id(rowset_id), @@ -157,14 +172,15 @@ public: _order(order), _seq_col_idx(seq_col_idx), _num_key_columns(_iter->schema().num_key_columns()), - _key_group_cluster_key_idxes(std::move(key_group_cluster_key_idxes)) {} + _key_group_cluster_key_idxes(std::move(key_group_cluster_key_idxes)), + _context_stats(context_stats) {} VerticalMergeIteratorContext(const VerticalMergeIteratorContext&) = delete; VerticalMergeIteratorContext(VerticalMergeIteratorContext&&) = delete; VerticalMergeIteratorContext& operator=(const VerticalMergeIteratorContext&) = delete; VerticalMergeIteratorContext& operator=(VerticalMergeIteratorContext&&) = delete; - ~VerticalMergeIteratorContext() = default; + ~VerticalMergeIteratorContext(); Status block_reset(const std::shared_ptr<Block>& block); Status init(const StorageReadOptions& opts, CompactionSampleInfo* sample_info = nullptr); bool compare(const VerticalMergeIteratorContext& rhs) const; @@ -230,9 +246,15 @@ public: const std::shared_ptr<Block>& block_ptr() const { return _block; } + // No later row source references this context. The returned IteratorRowRef/RowBatch keeps + // its own shared_ptr<Block>, so the segment reader and context-owned blocks can be released. + void release_resources(); + private: // Load next block into _block Status _load_next_block(); + void _mark_active(); + void _mark_inactive(); RowwiseIteratorUPtr _iter; RowsetId _rowset_id; @@ -250,6 +272,7 @@ private: size_t _block_row_max = 0; int64_t _num_key_columns; const std::vector<uint32_t> _key_group_cluster_key_idxes; + VerticalCompactionContextStats* _context_stats; size_t _cur_batch_num = 0; // used to store data load from iterator->next_batch(Block*) @@ -258,6 +281,7 @@ private: std::list<std::shared_ptr<Block>> _block_list; // use to identify whether it's first block load from RowwiseIterator bool _is_first_row = true; + bool _is_active_context_counted = false; bool _record_rowids = false; std::vector<RowLocation> _block_row_locations; }; @@ -271,6 +295,7 @@ public: std::vector<RowsetId> rowset_ids, size_t ori_return_cols, KeysType keys_type, int32_t seq_col_idx, RowSourcesBuffer* row_sources_buf, + VerticalCompactionContextStats* context_stats, std::vector<uint32_t> key_group_cluster_key_idxes) : _origin_iters(std::move(iters)), _iterator_init_flags(std::move(iterator_init_flags)), @@ -279,6 +304,7 @@ public: _keys_type(keys_type), _seq_col_idx(seq_col_idx), _row_sources_buf(row_sources_buf), + _context_stats(context_stats), _key_group_cluster_key_idxes(std::move(key_group_cluster_key_idxes)) {} ~VerticalHeapMergeIterator() override = default; @@ -323,6 +349,7 @@ private: KeysType _keys_type; int32_t _seq_col_idx = -1; RowSourcesBuffer* _row_sources_buf; + VerticalCompactionContextStats* _context_stats; uint32_t _merged_rows = 0; StorageReadOptions _opts; bool _record_rowids = false; @@ -338,14 +365,16 @@ public: std::vector<bool> iterator_init_flags, std::vector<RowsetId> rowset_ids, size_t ori_return_cols, KeysType keys_type, int32_t seq_col_idx, - RowSourcesBuffer* row_sources_buf) + RowSourcesBuffer* row_sources_buf, + VerticalCompactionContextStats* context_stats) : _origin_iters(std::move(iters)), _iterator_init_flags(std::move(iterator_init_flags)), _rowset_ids(std::move(rowset_ids)), _ori_return_cols(ori_return_cols), _keys_type(keys_type), _seq_col_idx(seq_col_idx), - _row_sources_buf(row_sources_buf) {} + _row_sources_buf(row_sources_buf), + _context_stats(context_stats) {} ~VerticalFifoMergeIterator() override = default; VerticalFifoMergeIterator(const VerticalFifoMergeIterator&) = delete; @@ -377,6 +406,7 @@ private: KeysType _keys_type; int32_t _seq_col_idx = -1; RowSourcesBuffer* _row_sources_buf; + VerticalCompactionContextStats* _context_stats; uint32_t _merged_rows = 0; StorageReadOptions _opts; bool _record_rowids = false; @@ -399,10 +429,12 @@ class VerticalMaskMergeIterator : public RowwiseIterator { public: // VerticalMaskMergeIterator takes the ownership of input iterators VerticalMaskMergeIterator(std::vector<RowwiseIteratorUPtr>&& iters, size_t ori_return_cols, - RowSourcesBuffer* row_sources_buf) + RowSourcesBuffer* row_sources_buf, + VerticalCompactionContextStats* context_stats) : _origin_iters(std::move(iters)), _ori_return_cols(ori_return_cols), - _row_sources_buf(row_sources_buf) {} + _row_sources_buf(row_sources_buf), + _context_stats(context_stats) {} ~VerticalMaskMergeIterator() override = default; VerticalMaskMergeIterator(const VerticalMaskMergeIterator&) = delete; @@ -429,6 +461,8 @@ private: int64_t _get_size(Block* block) { return block->rows(); } Status check_all_iter_finished(); + // Advance the row-source cursor and release its context after the final reference. + void consume_row_sources(uint16_t order, size_t count = 1); // released after build ctx std::vector<RowwiseIteratorUPtr> _origin_iters; @@ -441,6 +475,7 @@ private: int _block_row_max = 0; size_t _filtered_rows = 0; RowSourcesBuffer* _row_sources_buf; + VerticalCompactionContextStats* _context_stats; StorageReadOptions _opts; CompactionSampleInfo* _sample_info = nullptr; }; @@ -450,16 +485,18 @@ std::shared_ptr<RowwiseIterator> new_vertical_heap_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>& iterator_init_flag, const std::vector<RowsetId>& rowset_ids, size_t _ori_return_cols, KeysType key_type, uint32_t seq_col_idx, RowSourcesBuffer* row_sources_buf, + VerticalCompactionContextStats* context_stats, std::vector<uint32_t> key_group_cluster_key_idxes); std::shared_ptr<RowwiseIterator> new_vertical_fifo_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, const std::vector<bool>& iterator_init_flag, const std::vector<RowsetId>& rowset_ids, size_t _ori_return_cols, KeysType key_type, - uint32_t seq_col_idx, RowSourcesBuffer* row_sources_buf); + uint32_t seq_col_idx, RowSourcesBuffer* row_sources_buf, + VerticalCompactionContextStats* context_stats); std::shared_ptr<RowwiseIterator> new_vertical_mask_merge_iterator( std::vector<RowwiseIteratorUPtr>&& inputs, size_t ori_return_cols, - RowSourcesBuffer* row_sources_buf); + RowSourcesBuffer* row_sources_buf, VerticalCompactionContextStats* context_stats); #include "common/compile_check_end.h" } // namespace doris diff --git a/be/src/storage/merger.cpp b/be/src/storage/merger.cpp index f574f63f056..f5364d48787 100644 --- a/be/src/storage/merger.cpp +++ b/be/src/storage/merger.cpp @@ -60,6 +60,7 @@ #include "storage/tablet/tablet_reader.h" #include "storage/types.h" #include "storage/utils.h" +#include "util/defer_op.h" #include "util/slice.h" namespace doris { @@ -250,10 +251,11 @@ Status Merger::vertical_compact_one_group( const std::vector<RowsetReaderSharedPtr>& src_rowset_readers, RowsetWriter* dst_rowset_writer, uint32_t max_rows_per_segment, Statistics* stats_output, std::vector<uint32_t> key_group_cluster_key_idxes, int64_t batch_size, - CompactionSampleInfo* sample_info, bool enable_sparse_optimization) { + CompactionSampleInfo* sample_info, VerticalCompactionContextStats* context_stats, + bool enable_sparse_optimization) { // build tablet reader VLOG_NOTICE << "vertical compact one group, max_rows_per_segment=" << max_rows_per_segment; - VerticalBlockReader reader(row_source_buf); + VerticalBlockReader reader(row_source_buf, context_stats); TabletReader::ReaderParams reader_params; reader_params.is_key_column_group = is_key; reader_params.key_group_cluster_key_idxes = key_group_cluster_key_idxes; @@ -494,6 +496,13 @@ Status Merger::vertical_merge_rowsets(BaseTabletSPtr tablet, ReaderType reader_t Statistics* stats_output, VerticalCompactionProgressCallback progress_cb) { LOG(INFO) << "Start to do vertical compaction, tablet_id: " << tablet->tablet_id(); + VerticalCompactionContextStats context_stats; + Defer log_context_stats {[&] { + DCHECK_EQ(context_stats.active_segment_contexts, 0); + LOG(INFO) << "Vertical compaction segment context statistics, tablet_id: " + << tablet->tablet_id() << ", vertical_compaction_active_segment_contexts_peak: " + << context_stats.active_segment_contexts_peak; + }}; std::vector<std::vector<uint32_t>> column_groups; std::vector<uint32_t> key_group_cluster_key_idxes; // If BE config vertical_compaction_num_columns_per_group has been modified from @@ -698,7 +707,8 @@ Status Merger::vertical_merge_rowsets(BaseTabletSPtr tablet, ReaderType reader_t Status st = vertical_compact_one_group( tablet, reader_type, tablet_schema, is_key, column_groups[i], &row_sources_buf, src_rowset_readers, dst_rowset_writer, max_rows_per_segment, group_stats_ptr, - key_group_cluster_key_idxes, batch_size, &sample_info, enable_sparse_optimization); + key_group_cluster_key_idxes, batch_size, &sample_info, &context_stats, + enable_sparse_optimization); { std::unique_lock<std::mutex> lock(sample_info_lock); sample_infos[i] = sample_info; diff --git a/be/src/storage/merger.h b/be/src/storage/merger.h index 4dcf4787c20..5185c97e6b0 100644 --- a/be/src/storage/merger.h +++ b/be/src/storage/merger.h @@ -38,6 +38,7 @@ class SegmentWriter; class RowSourcesBuffer; class VerticalBlockReader; +struct VerticalCompactionContextStats; using VerticalCompactionProgressCallback = std::function<void(int64_t total_groups, int64_t completed_groups)>; @@ -85,7 +86,7 @@ public: RowsetWriter* dst_rowset_writer, uint32_t max_rows_per_segment, Statistics* stats_output, std::vector<uint32_t> key_group_cluster_key_idxes, int64_t batch_size, CompactionSampleInfo* sample_info, - bool enable_sparse_optimization = false); + VerticalCompactionContextStats* context_stats, bool enable_sparse_optimization = false); // for segcompaction static Status vertical_compact_one_group( diff --git a/be/test/storage/compaction/vertical_compaction_test.cpp b/be/test/storage/compaction/vertical_compaction_test.cpp index 079bbb3f9af..1d9f8abe14c 100644 --- a/be/test/storage/compaction/vertical_compaction_test.cpp +++ b/be/test/storage/compaction/vertical_compaction_test.cpp @@ -16,6 +16,7 @@ // specific language governing permissions and limitations // under the License. +#include <bvar/variable.h> #include <gen_cpp/AgentService_types.h> #include <gen_cpp/Descriptors_types.h> #include <gen_cpp/PaloInternalService_types.h> @@ -50,6 +51,7 @@ #include "io/io_common.h" #include "json2pb/json_to_pb.h" #include "runtime/exec_env.h" +#include "runtime/thread_context.h" #include "storage/compaction/compaction.h" #include "storage/delete/delete_handler.h" #include "storage/iterator/vertical_merge_iterator.h" @@ -73,6 +75,7 @@ #include "storage/tablet/tablet_schema.h" #include "storage/txn/txn_manager.h" #include "storage/utils.h" +#include "util/defer_op.h" #include "util/uid_util.h" namespace doris { @@ -584,6 +587,9 @@ TEST_F(VerticalCompactionTest, TestRowSourcesBuffer) { size_t limit = 10; static_cast<void>(buffer.flush()); static_cast<void>(buffer.seek_to_begin()); + EXPECT_FALSE(buffer.is_source_exhausted(0)); + EXPECT_FALSE(buffer.is_source_exhausted(1)); + EXPECT_FALSE(buffer.is_source_exhausted(2)); int idx = -1; while (buffer.has_remaining().ok()) { @@ -594,7 +600,12 @@ TEST_F(VerticalCompactionTest, TestRowSourcesBuffer) { auto same = buffer.same_source_count(cur, limit); EXPECT_EQ(same, 2); buffer.advance(same); + EXPECT_TRUE(buffer.is_source_exhausted(cur)); } + static_cast<void>(buffer.seek_to_begin()); + EXPECT_FALSE(buffer.is_source_exhausted(0)); + EXPECT_FALSE(buffer.is_source_exhausted(1)); + EXPECT_FALSE(buffer.is_source_exhausted(2)); RowSourcesBuffer buffer1(101, absolute_dir, ReaderType::READER_CUMULATIVE_COMPACTION); EXPECT_TRUE(buffer1.append(tmp_row_source).ok()); @@ -695,6 +706,9 @@ TEST_F(VerticalCompactionTest, TestRowSourcesBufferSpillThreshold) { ++read_back; } EXPECT_EQ(read_back, expected_total); + for (uint16_t source = 0; source < 8; ++source) { + EXPECT_TRUE(buffer.is_source_exhausted(source)); + } } TEST_F(VerticalCompactionTest, TestDupKeyVerticalMerge) { @@ -1169,6 +1183,275 @@ TEST_F(VerticalCompactionTest, << "the higher sequence row should stay visible after compaction"; } +TEST_F(VerticalCompactionTest, TestUniqueKeyNonOverlappingSegmentContextRetention) { + constexpr uint32_t num_input_rowsets = 3; + constexpr uint32_t num_segments_per_rowset = 4; + constexpr uint32_t rows_per_segment = 64; + constexpr int64_t total_segments = num_input_rowsets * num_segments_per_rowset; + constexpr int64_t total_rows = total_segments * rows_per_segment; + + auto old_compaction_batch_size = config::compaction_batch_size; + auto old_sparse_threshold = config::sparse_column_compaction_threshold_percent; + Defer restore_config {[&] { + config::compaction_batch_size = old_compaction_batch_size; + config::sparse_column_compaction_threshold_percent = old_sparse_threshold; + }}; + config::compaction_batch_size = 32; + config::sparse_column_compaction_threshold_percent = 0; + + std::vector<std::vector<std::vector<std::tuple<int64_t, int64_t>>>> input_data; + for (uint32_t rowset_id = 0; rowset_id < num_input_rowsets; ++rowset_id) { + std::vector<std::vector<std::tuple<int64_t, int64_t>>> rowset_data; + for (uint32_t segment_id = 0; segment_id < num_segments_per_rowset; ++segment_id) { + std::vector<std::tuple<int64_t, int64_t>> segment_data; + for (uint32_t row_id = 0; row_id < rows_per_segment; ++row_id) { + int64_t logical_row = segment_id * rows_per_segment + row_id; + int64_t key = logical_row * num_input_rowsets + rowset_id; + segment_data.emplace_back(key, key + 1); + } + rowset_data.emplace_back(std::move(segment_data)); + } + input_data.emplace_back(std::move(rowset_data)); + } + + TabletSchemaSPtr tablet_schema = create_schema(UNIQUE_KEYS); + std::vector<RowsetSharedPtr> input_rowsets; + for (uint32_t rowset_id = 0; rowset_id < num_input_rowsets; ++rowset_id) { + auto rowset = + create_rowset(tablet_schema, NONOVERLAPPING, input_data[rowset_id], rowset_id); + ASSERT_FALSE(rowset->rowset_meta()->is_segments_overlapping()); + ASSERT_EQ(num_segments_per_rowset, rowset->num_segments()); + input_rowsets.push_back(rowset); + } + + TabletSharedPtr tablet = create_tablet(*tablet_schema, false); + auto run_case = [&](double sparse_threshold) { + config::sparse_column_compaction_threshold_percent = sparse_threshold; + tablet->compaction_density.store(1.0); + + std::vector<RowsetReaderSharedPtr> input_rs_readers; + for (const auto& rowset : input_rowsets) { + RowsetReaderSharedPtr rs_reader; + ASSERT_TRUE(rowset->create_reader(&rs_reader).ok()); + input_rs_readers.push_back(std::move(rs_reader)); + } + + auto writer_context = + create_rowset_writer_context(tablet_schema, NONOVERLAPPING, UINT32_MAX, + {0, input_rowsets.back()->end_version()}); + auto res = RowsetFactory::create_rowset_writer(*engine_ref, writer_context, true); + ASSERT_TRUE(res.has_value()) << res.error(); + auto output_rs_writer = std::move(res).value(); + + ASSERT_EQ("0", + bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); + + Merger::Statistics stats; + auto st = Merger::vertical_merge_rowsets( + tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema, input_rs_readers, + output_rs_writer.get(), UINT32_MAX, total_segments, &stats); + ASSERT_TRUE(st.ok()) << st; + + EXPECT_EQ("0", + bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); + EXPECT_EQ(total_rows, stats.output_rows); + EXPECT_EQ(0, stats.merged_rows); + EXPECT_EQ(0, stats.filtered_rows); + + RowsetSharedPtr output_rowset; + ASSERT_EQ(Status::OK(), output_rs_writer->build(output_rowset)); + ASSERT_TRUE(output_rowset); + EXPECT_EQ(total_rows, output_rowset->num_rows()); + }; + + run_case(0); + run_case(1.0); +} + +TEST_F(VerticalCompactionTest, TestUniqueKeySegmentContextMemoryAmplification) { + constexpr uint32_t num_input_rowsets = 10; + constexpr uint32_t batch_size = 32; + constexpr uint32_t payload_size = 8 * 1024; + + auto old_compaction_batch_size = config::compaction_batch_size; + auto old_sparse_threshold = config::sparse_column_compaction_threshold_percent; + Defer restore_config {[&] { + config::compaction_batch_size = old_compaction_batch_size; + config::sparse_column_compaction_threshold_percent = old_sparse_threshold; + }}; + config::compaction_batch_size = batch_size; + config::sparse_column_compaction_threshold_percent = 0; + + TabletSchemaSPtr tablet_schema = std::make_shared<TabletSchema>(); + TabletSchemaPB tablet_schema_pb; + tablet_schema_pb.set_keys_type(UNIQUE_KEYS); + tablet_schema_pb.set_num_short_key_columns(1); + tablet_schema_pb.set_num_rows_per_row_block(1024); + tablet_schema_pb.set_compress_kind(COMPRESS_NONE); + tablet_schema_pb.set_next_column_unique_id(4); + + ColumnPB* key_column = tablet_schema_pb.add_column(); + key_column->set_unique_id(1); + key_column->set_name("key"); + key_column->set_type("INT"); + key_column->set_is_key(true); + key_column->set_length(4); + key_column->set_index_length(4); + key_column->set_is_nullable(false); + key_column->set_is_bf_column(false); + + ColumnPB* value_column = tablet_schema_pb.add_column(); + value_column->set_unique_id(2); + value_column->set_name("value"); + value_column->set_type("VARCHAR"); + value_column->set_is_key(false); + value_column->set_length(payload_size); + value_column->set_index_length(20); + value_column->set_is_nullable(false); + value_column->set_is_bf_column(false); + + ColumnPB* delete_sign_column = tablet_schema_pb.add_column(); + delete_sign_column->set_unique_id(3); + delete_sign_column->set_name(DELETE_SIGN); + delete_sign_column->set_type("TINYINT"); + delete_sign_column->set_is_key(false); + delete_sign_column->set_length(1); + delete_sign_column->set_index_length(1); + delete_sign_column->set_is_nullable(false); + delete_sign_column->set_is_bf_column(false); + + tablet_schema->init_from_pb(tablet_schema_pb); + TabletSharedPtr tablet = create_tablet(*tablet_schema, false); + + struct Observation { + int64_t memory_peak; + }; + std::vector<Observation> observations; + + auto run_case = [&](uint32_t num_segments_per_rowset, uint32_t rows_per_segment, + int64_t base_version) { + int64_t total_segments = num_input_rowsets * num_segments_per_rowset; + int64_t total_rows = total_segments * rows_per_segment; + std::string payload(payload_size, 'x'); + std::vector<RowsetSharedPtr> input_rowsets; + std::vector<RowsetReaderSharedPtr> input_rs_readers; + + for (uint32_t rowset_id = 0; rowset_id < num_input_rowsets; ++rowset_id) { + auto writer_context = create_rowset_writer_context( + tablet_schema, NONOVERLAPPING, UINT32_MAX, + {base_version + rowset_id, base_version + rowset_id}); + auto res = RowsetFactory::create_rowset_writer(*engine_ref, writer_context, true); + ASSERT_TRUE(res.has_value()) << res.error(); + auto rowset_writer = std::move(res).value(); + + for (uint32_t segment_id = 0; segment_id < num_segments_per_rowset; ++segment_id) { + Block block = tablet_schema->create_block(); + auto columns = std::move(block).mutate_columns(); + for (uint32_t row_id = 0; row_id < rows_per_segment; ++row_id) { + int32_t logical_row = segment_id * rows_per_segment + row_id; + int32_t key = logical_row * num_input_rowsets + rowset_id; + uint8_t delete_sign = 0; + columns[0]->insert_data(reinterpret_cast<const char*>(&key), sizeof(key)); + columns[1]->insert_data(payload.data(), payload.size()); + columns[2]->insert_data(reinterpret_cast<const char*>(&delete_sign), + sizeof(delete_sign)); + } + ASSERT_TRUE(add_block_with_columns(rowset_writer.get(), &block, &columns).ok()); + ASSERT_TRUE(rowset_writer->flush().ok()); + } + + RowsetSharedPtr rowset; + ASSERT_EQ(Status::OK(), rowset_writer->build(rowset)); + ASSERT_FALSE(rowset->rowset_meta()->is_segments_overlapping()); + ASSERT_EQ(num_segments_per_rowset, rowset->num_segments()); + ASSERT_EQ(num_segments_per_rowset * rows_per_segment, rowset->num_rows()); + input_rowsets.push_back(rowset); + + RowsetReaderSharedPtr rs_reader; + ASSERT_TRUE(rowset->create_reader(&rs_reader).ok()); + input_rs_readers.push_back(std::move(rs_reader)); + } + + auto output_writer_context = + create_rowset_writer_context(tablet_schema, NONOVERLAPPING, UINT32_MAX, + {base_version, base_version + num_input_rowsets - 1}); + auto output_res = + RowsetFactory::create_rowset_writer(*engine_ref, output_writer_context, true); + ASSERT_TRUE(output_res.has_value()) << output_res.error(); + auto output_rs_writer = std::move(output_res).value(); + + ASSERT_EQ("0", + bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); + + Merger::Statistics stats; + int64_t memory_peak = 0; + { + SCOPED_PEAK_MEM(&memory_peak); + auto st = Merger::vertical_merge_rowsets( + tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema, input_rs_readers, + output_rs_writer.get(), UINT32_MAX, total_segments, &stats); + ASSERT_TRUE(st.ok()) << st; + } + + ASSERT_EQ("0", + bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); + ASSERT_EQ(total_rows, stats.output_rows); + ASSERT_EQ(0, stats.merged_rows); + ASSERT_EQ(0, stats.filtered_rows); + + RowsetSharedPtr output_rowset; + ASSERT_EQ(Status::OK(), output_rs_writer->build(output_rowset)); + ASSERT_TRUE(output_rowset); + ASSERT_EQ(total_rows, output_rowset->num_rows()); + + RowsetReaderContext reader_context; + reader_context.tablet_schema = tablet_schema; + reader_context.need_ordered_result = false; + std::vector<uint32_t> return_columns = {0, 1, 2}; + reader_context.return_columns = &return_columns; + RowsetReaderSharedPtr output_rs_reader; + create_and_init_rowset_reader(output_rowset.get(), reader_context, &output_rs_reader); + + int64_t expected_key = 0; + Status read_status; + do { + Block output_block = tablet_schema->create_block(); + read_status = output_rs_reader->next_batch(&output_block); + const auto& output_columns = output_block.get_columns_with_type_and_name(); + ASSERT_EQ(3, output_columns.size()); + for (size_t row = 0; row < output_block.rows(); ++row) { + ASSERT_EQ(expected_key, output_columns[0].column->get_int(row)); + ASSERT_EQ(payload, output_columns[1].column->get_data_at(row).to_string()); + ASSERT_EQ(0, output_columns[2].column->get_int(row)); + ++expected_key; + } + } while (read_status.ok()); + ASSERT_TRUE(read_status.is<END_OF_FILE>()) << read_status; + ASSERT_EQ(total_rows, expected_key); + + observations.push_back({memory_peak}); + }; + + // Both cases contain exactly 32000 rows and the same 8 KiB value payload per row. + // Only the segment distribution differs. + run_case(10, 320, 1000); + ASSERT_EQ(1, observations.size()); + run_case(50, 64, 2000); + ASSERT_EQ(2, observations.size()); + + const auto& low_segment_case = observations[0]; + const auto& high_segment_case = observations[1]; + LOG(INFO) << "equal-data vertical compaction observation: low_segments_memory_peak=" + << low_segment_case.memory_peak + << ", high_segments_memory_peak=" << high_segment_case.memory_peak; + EXPECT_GT(low_segment_case.memory_peak, 0); + EXPECT_GT(high_segment_case.memory_peak, 0); + auto memory_peak_delta = low_segment_case.memory_peak > high_segment_case.memory_peak + ? low_segment_case.memory_peak - high_segment_case.memory_peak + : high_segment_case.memory_peak - low_segment_case.memory_peak; + EXPECT_LT(memory_peak_delta, 2 * 1024 * 1024); +} + TEST_F(VerticalCompactionTest, TestDupKeyVerticalMergeWithDelete) { auto num_input_rowset = 2; auto num_segments = 2; @@ -1422,10 +1705,12 @@ TEST_F(VerticalCompactionTest, TestAggKeyVerticalMerge) { Merger::Statistics stats; RowIdConversion rowid_conversion; stats.rowid_conversion = &rowid_conversion; + ASSERT_EQ("0", bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); auto s = Merger::vertical_merge_rowsets(tablet, ReaderType::READER_BASE_COMPACTION, *tablet_schema, input_rs_readers, output_rs_writer.get(), 100, num_segments, &stats); EXPECT_TRUE(s.ok()); + EXPECT_EQ("0", bvar::Variable::describe_exposed("vertical_compaction_active_segment_contexts")); RowsetSharedPtr out_rowset; EXPECT_EQ(Status::OK(), output_rs_writer->build(out_rowset)); --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
