diff --git a/be/src/load/memtable/memtable.cpp b/be/src/load/memtable/memtable.cpp index 5530cee63aa61c..32baa6f6fdceb3 100644 --- a/be/src/load/memtable/memtable.cpp +++ b/be/src/load/memtable/memtable.cpp @@ -23,6 +23,7 @@ #include #include +#include #include #include @@ -82,7 +83,7 @@ MemTable::MemTable(int64_t tablet_id, std::shared_ptr tablet_schem } _init_columns_offset_by_slot_descs(slot_descs, tuple_desc); // TODO: Support ZOrderComparator in the future - _row_in_blocks = std::make_unique>>(); + _row_in_blocks = std::make_unique>(); _load_mem_limit = MemInfo::mem_limit() * config::load_process_max_memory_limit_percent / 100; } @@ -165,8 +166,8 @@ MemTable::~MemTable() { SCOPED_CONSUME_MEM_TRACKER(_mem_tracker); g_memtable_cnt << -1; if (_keys_type != KeysType::DUP_KEYS) { - for (auto it = _row_in_blocks->begin(); it != _row_in_blocks->end(); it++) { - if (!(*it)->has_init_agg()) { + for (const auto& row : *_row_in_blocks) { + if (!_has_agg(row)) { continue; } // We should release agg_places here, because they are not released when a @@ -174,7 +175,7 @@ MemTable::~MemTable() { for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) { auto function = _agg_functions[i]; DCHECK(function != nullptr); - function->destroy((*it)->agg_places(i)); + function->destroy(_agg_place(row, i)); } } } @@ -266,28 +267,29 @@ Status MemTable::insert(const Block* input_block, const TabletAddRowsPayload& ro RETURN_IF_ERROR(_input_mutable_block.add_rows(input_block, row_idxs.data(), row_idxs.data() + num_rows, &_column_offset)); for (int i = 0; i < num_rows; i++) { - _row_in_blocks->emplace_back(std::make_shared( - cursor_in_mutableblock + i, _need_row_binlog_lsn ? row_binlog_lsns[i] : 0)); + _row_in_blocks->emplace_back(cursor_in_mutableblock + i, + _need_row_binlog_lsn ? row_binlog_lsns[i] : 0); } _stat.raw_rows += num_rows; return Status::OK(); } -void MemTable::_merge_row_binlog_lsn(RowInBlock* src_row, RowInBlock* dst_row) { +void MemTable::_merge_row_binlog_lsn(const RowInBlock& src_row, RowInBlock& dst_row) { if (_need_row_binlog_lsn) { - dst_row->_row_binlog_lsn = std::max(dst_row->_row_binlog_lsn, src_row->_row_binlog_lsn); + dst_row._row_binlog_lsn = std::max(dst_row._row_binlog_lsn, src_row._row_binlog_lsn); } } -void MemTable::_append_output_row_binlog_lsn(RowInBlock* row) { +void MemTable::_append_output_row_binlog_lsn(const RowInBlock& row) { if (_need_row_binlog_lsn) { - _output_row_binlog_lsns.emplace_back(row->_row_binlog_lsn); + _output_row_binlog_lsns.emplace_back(row._row_binlog_lsn); } } void MemTable::_aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, - RowInBlock* src_row, RowInBlock* dst_row) { + const RowInBlock& src_row, + RowInBlock& dst_row) { _merge_row_binlog_lsn(src_row, dst_row); // for each mapping replace value columns according to the sequence column compare result // for example: a b c d s1 s2 (key:a , s1=>[b,c], s2=>[d]) @@ -299,7 +301,7 @@ void MemTable::_aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, for (const auto& it : seq_map) { auto sequence = it.first; auto* sequence_col_ptr = mutable_block.mutable_columns()[sequence].get(); - auto res = sequence_col_ptr->compare_at(dst_row->_row_pos, src_row->_row_pos, + auto res = sequence_col_ptr->compare_at(dst_row._row_pos, src_row._row_pos, *sequence_col_ptr, -1); if (res > 0) { continue; @@ -307,27 +309,27 @@ void MemTable::_aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, for (auto cid : it.second) { if (cid < _num_columns) { auto* col_ptr = mutable_block.mutable_columns()[cid].get(); - _agg_functions[cid]->add(dst_row->agg_places(cid), + _agg_functions[cid]->add(_agg_place(dst_row, cid), const_cast(&col_ptr), - src_row->_row_pos, _arena); + src_row._row_pos, _arena); } } if (sequence < _num_columns) { - _agg_functions[sequence]->add(dst_row->agg_places(sequence), + _agg_functions[sequence]->add(_agg_place(dst_row, sequence), const_cast(&sequence_col_ptr), - src_row->_row_pos, _arena); + src_row._row_pos, _arena); // must use replace column instead of update row_pos // because one row may have multi sequence column // and agg function add method won't change the real column value - sequence_col_ptr->replace_column_data(*sequence_col_ptr, src_row->_row_pos, - dst_row->_row_pos); + sequence_col_ptr->replace_column_data(*sequence_col_ptr, src_row._row_pos, + dst_row._row_pos); } } } template -void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlock* src_row, - RowInBlock* dst_row) { +void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block, const RowInBlock& src_row, + RowInBlock& dst_row) { _merge_row_binlog_lsn(src_row, dst_row); // for flexible partial update, the caller must guarantees that either src_row and dst_row // both specify the sequence column, or src_row and dst_row both don't specify the @@ -335,23 +337,23 @@ void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlo if (_tablet_schema->has_sequence_col() && _seq_col_idx_in_block >= 0) { DCHECK_LT(_seq_col_idx_in_block, mutable_block.columns()); auto col_ptr = mutable_block.mutable_columns()[_seq_col_idx_in_block].get(); - auto res = col_ptr->compare_at(dst_row->_row_pos, src_row->_row_pos, *col_ptr, -1); + auto res = col_ptr->compare_at(dst_row._row_pos, src_row._row_pos, *col_ptr, -1); // dst sequence column larger than src, don't need to update if (res > 0) { return; } // need to update the row pos in dst row to the src row pos when has // sequence column - dst_row->_row_pos = src_row->_row_pos; + dst_row._row_pos = src_row._row_pos; } // dst is non-sequence row, or dst sequence is smaller if constexpr (!has_skip_bitmap_col) { DCHECK(_skip_bitmap_col_idx == -1); for (size_t cid = _tablet_schema->num_key_columns(); cid < _num_columns; ++cid) { auto* col_ptr = mutable_block.mutable_columns()[cid].get(); - _agg_functions[cid]->add(dst_row->agg_places(cid), - const_cast(&col_ptr), - src_row->_row_pos, _arena); + _agg_functions[cid]->add(_agg_place(dst_row, cid), + const_cast(&col_ptr), src_row._row_pos, + _arena); } } else { DCHECK(_skip_bitmap_col_idx != -1); @@ -359,16 +361,16 @@ void MemTable::_aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlo const BitmapValue& skip_bitmap = assert_cast( mutable_block.mutable_columns()[_skip_bitmap_col_idx].get()) - ->get_data()[src_row->_row_pos]; + ->get_data()[src_row._row_pos]; for (size_t cid = _tablet_schema->num_key_columns(); cid < _num_columns; ++cid) { const auto& col = _tablet_schema->column(cid); if (cid != _skip_bitmap_col_idx && skip_bitmap.contains(col.unique_id())) { continue; } auto* col_ptr = mutable_block.mutable_columns()[cid].get(); - _agg_functions[cid]->add(dst_row->agg_places(cid), - const_cast(&col_ptr), - src_row->_row_pos, _arena); + _agg_functions[cid]->add(_agg_place(dst_row, cid), + const_cast(&col_ptr), src_row._row_pos, + _arena); } } } @@ -380,14 +382,30 @@ Status MemTable::_put_into_output(Block& in_block) { if (_need_row_binlog_lsn) { _output_row_binlog_lsns.reserve(_output_row_binlog_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_row_binlog_lsn((*_row_in_blocks)[i].get()); + for (const auto& row : *_row_in_blocks) { + row_pos_vec.emplace_back(row._row_pos); + _append_output_row_binlog_lsn(row); } return _output_mutable_block.add_rows(&in_block, row_pos_vec.data(), row_pos_vec.data() + in_block.rows()); } +void MemTable::_sort_one_column(DorisVector& row_in_blocks, Tie& tie, + std::function 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](const RowInBlock& lhs, const RowInBlock& 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], row_in_blocks[i]) == 0); + } + } +} + size_t MemTable::_sort() { SCOPED_RAW_TIMER(&_stat.sort_ns); _stat.sort_times++; @@ -395,8 +413,8 @@ size_t MemTable::_sort() { // sort new rows Tie tie = Tie(_last_sorted_pos, _row_in_blocks->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 RowInBlock& lhs, const RowInBlock& rhs) -> int { + return _input_mutable_block.compare_one_column(lhs._row_pos, rhs._row_pos, i, -1); }; _sort_one_column(*_row_in_blocks, tie, cmp); } @@ -406,20 +424,19 @@ size_t MemTable::_sort() { 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; + [&is_dup](const RowInBlock& lhs, const RowInBlock& rhs) -> bool { + return is_dup ? lhs._row_pos > rhs._row_pos : lhs._row_pos < rhs._row_pos; }); 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_func = [this, is_dup, &same_keys_num](const RowInBlock& l, + const RowInBlock& r) -> bool { + auto value = (*(this->_vec_row_comparator))(&l, &r); if (value == 0) { same_keys_num++; - return is_dup ? l->_row_pos > r->_row_pos : l->_row_pos < r->_row_pos; + return is_dup ? l._row_pos > r._row_pos : l._row_pos < r._row_pos; } else { return value < 0; } @@ -439,15 +456,13 @@ Status MemTable::_sort_by_cluster_keys() { MutableBlock mutable_block = MutableBlock::build_mutable_block(std::move(in_block)); _output_mutable_block = MutableBlock::build_mutable_block(std::move(clone_block)); - DorisVector> row_in_blocks; + DorisVector row_in_blocks; row_in_blocks.reserve(mutable_block.rows()); if (_need_row_binlog_lsn) { DCHECK_EQ(_output_row_binlog_lsns.size(), mutable_block.rows()); } for (size_t i = 0; i < mutable_block.rows(); i++) { - row_in_blocks.emplace_back( - _need_row_binlog_lsn ? std::make_shared(i, _output_row_binlog_lsns[i]) - : std::make_shared(i)); + row_in_blocks.emplace_back(i, _need_row_binlog_lsn ? _output_row_binlog_lsns[i] : 0); } if (_need_row_binlog_lsn) { _output_row_binlog_lsns.clear(); @@ -461,8 +476,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 { - return mutable_block.compare_one_column(lhs->_row_pos, rhs->_row_pos, index, -1); + auto cmp = [&](const RowInBlock& lhs, const RowInBlock& rhs) -> int { + return mutable_block.compare_one_column(lhs._row_pos, rhs._row_pos, index, -1); }; _sort_one_column(row_in_blocks, tie, cmp); } @@ -472,8 +487,9 @@ Status MemTable::_sort_by_cluster_keys() { while (iter.next()) { pdqsort(std::next(row_in_blocks.begin(), iter.left()), std::next(row_in_blocks.begin(), iter.right()), - [](const std::shared_ptr& lhs, const std::shared_ptr& rhs) - -> bool { return lhs->_row_pos < rhs->_row_pos; }); + [](const RowInBlock& lhs, const RowInBlock& rhs) -> bool { + return lhs._row_pos < rhs._row_pos; + }); } in_block = mutable_block.to_block(); @@ -481,9 +497,9 @@ Status MemTable::_sort_by_cluster_keys() { DorisVector row_pos_vec; DCHECK(in_block.rows() <= std::numeric_limits::max()); row_pos_vec.reserve(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_row_binlog_lsn(row_in_blocks[i].get()); + for (const auto& row : row_in_blocks) { + row_pos_vec.emplace_back(row._row_pos); + _append_output_row_binlog_lsn(row); } std::vector column_offset; for (int i = 0; i < _column_offset.size(); ++i) { @@ -493,32 +509,18 @@ 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) { - 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; }); - 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); - } - } -} - template -void MemTable::_finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, int row_pos) { +void MemTable::_finalize_one_row(RowInBlock& row, MutableBlock& mutable_block, int row_pos) { // move key columns for (size_t i = 0; i < _tablet_schema->num_key_columns(); ++i) { _output_mutable_block.get_column_by_position(i)->insert_from( - *mutable_block.get_column_by_position(i), row->_row_pos); + *mutable_block.get_column_by_position(i), row._row_pos); } - if (row->has_init_agg()) { + if (_has_agg(row)) { // get value columns from agg_places for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) { auto function = _agg_functions[i]; - auto* agg_place = row->agg_places(i); + auto* agg_place = _agg_place(row, i); auto* col_ptr = _output_mutable_block.get_column_by_position(i).get(); function->insert_result_into(agg_place, *col_ptr); @@ -530,11 +532,11 @@ void MemTable::_finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, i } if constexpr (is_final) { - row->remove_init_agg(); + row._agg_mem = nullptr; } else { for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) { auto function = _agg_functions[i]; - auto* agg_place = row->agg_places(i); + auto* agg_place = _agg_place(row, i); auto* col_ptr = _output_mutable_block.get_column_by_position(i).get(); function->add(agg_place, const_cast(&col_ptr), row_pos, _arena); @@ -544,34 +546,33 @@ void MemTable::_finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, i // move columns for rows do not need agg for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) { _output_mutable_block.get_column_by_position(i)->insert_from( - *mutable_block.get_column_by_position(i), row->_row_pos); + *mutable_block.get_column_by_position(i), row._row_pos); } } _append_output_row_binlog_lsn(row); if constexpr (!is_final) { - row->_row_pos = row_pos; + row._row_pos = row_pos; } } -void MemTable::_init_row_for_agg(RowInBlock* row, MutableBlock& mutable_block) { - row->init_agg_places(_arena.aligned_alloc(_total_size_of_aggregate_states, 16), - _offsets_of_aggregate_states.data()); +void MemTable::_init_row_for_agg(RowInBlock& row, MutableBlock& mutable_block) { + row._agg_mem = _arena.aligned_alloc(_total_size_of_aggregate_states, 16); for (auto cid = _tablet_schema->num_key_columns(); cid < _num_columns; cid++) { auto* col_ptr = mutable_block.mutable_columns()[cid].get(); - auto* data = row->agg_places(cid); + auto* data = _agg_place(row, cid); _agg_functions[cid]->create(data); - _agg_functions[cid]->add(data, const_cast(&col_ptr), row->_row_pos, + _agg_functions[cid]->add(data, const_cast(&col_ptr), row._row_pos, _arena); } } -void MemTable::_clear_row_agg(RowInBlock* row) { - if (row->has_init_agg()) { +void MemTable::_clear_row_agg(RowInBlock& row) { + if (_has_agg(row)) { for (size_t i = _tablet_schema->num_key_columns(); i < _num_columns; ++i) { auto function = _agg_functions[i]; - auto* agg_place = row->agg_places(i); + auto* agg_place = _agg_place(row, i); function->destroy(agg_place); } - row->remove_init_agg(); + row._agg_mem = nullptr; } } // only in `to_block` the `is_final` flag will be true, in other cases, it will be false @@ -583,46 +584,48 @@ void MemTable::_aggregate() { std::unique_ptr empty_input_block = in_block.create_same_struct_block(0); MutableBlock mutable_block = MutableBlock::build_mutable_block(std::move(in_block)); _vec_row_comparator->set_block(&mutable_block); - DorisVector> temp_row_in_blocks; - temp_row_in_blocks.reserve(_last_sorted_pos); + DorisVector temp_row_in_blocks; + // Rows are held by value, so prev_row below points into temp_row_in_blocks. + // Reserving the upper bound up front keeps that pointer valid for the whole + // loop. + temp_row_in_blocks.reserve(_row_in_blocks->size()); //only init agg if needed if constexpr (!has_skip_bitmap_col) { RowInBlock* prev_row = nullptr; int row_pos = -1; - for (const auto& cur_row_ptr : *_row_in_blocks) { - RowInBlock* cur_row = cur_row_ptr.get(); - if (!temp_row_in_blocks.empty() && (*_vec_row_comparator)(prev_row, cur_row) == 0) { - if (!prev_row->has_init_agg()) { - _init_row_for_agg(prev_row, mutable_block); + for (RowInBlock& cur_row : *_row_in_blocks) { + if (!temp_row_in_blocks.empty() && (*_vec_row_comparator)(prev_row, &cur_row) == 0) { + if (!_has_agg(*prev_row)) { + _init_row_for_agg(*prev_row, mutable_block); } _stat.merged_rows++; if (_tablet_schema->has_seq_map()) { - _aggregate_two_row_with_sequence_map(mutable_block, cur_row, prev_row); + _aggregate_two_row_with_sequence_map(mutable_block, cur_row, *prev_row); } else { _aggregate_two_row_in_block(mutable_block, cur_row, - prev_row); + *prev_row); } // Clean up aggregation state of the merged row to avoid memory leak - if (cur_row) { - _clear_row_agg(cur_row); - } + _clear_row_agg(cur_row); } else { - prev_row = cur_row; if (!temp_row_in_blocks.empty()) { // The rows from the previous batch of _row_in_blocks have been merged into temp_row_in_blocks, // now call finalize to write the aggregation results into _output_mutable_block. - _finalize_one_row(temp_row_in_blocks.back().get(), mutable_block, - row_pos); + _finalize_one_row(temp_row_in_blocks.back(), mutable_block, row_pos); } - temp_row_in_blocks.push_back(cur_row_ptr); + // Aggregation mutates the group representative, and the copy that + // _finalize_one_row will read is the one in temp_row_in_blocks, so + // prev_row has to point there rather than into _row_in_blocks. + temp_row_in_blocks.push_back(cur_row); + prev_row = &temp_row_in_blocks.back(); row_pos++; } } if (!temp_row_in_blocks.empty()) { // finalize the last low - _finalize_one_row(temp_row_in_blocks.back().get(), mutable_block, row_pos); + _finalize_one_row(temp_row_in_blocks.back(), mutable_block, row_pos); } } else { DCHECK(_delete_sign_col_idx != -1); @@ -641,15 +644,18 @@ void MemTable::_aggregate() { _output_mutable_block = MutableBlock::build_mutable_block(std::move(*empty_input_block)); _output_mutable_block.clear_column_data(); _output_row_binlog_lsns.clear(); - *_row_in_blocks = temp_row_in_blocks; - _last_sorted_pos = _row_in_blocks->size(); } + // Rows are held by value, so the entries left in _row_in_blocks are stale + // copies of the ones _finalize_one_row just worked on -- including their + // _agg_mem, whose state it may have released. Adopting the finalized rows + // unconditionally keeps ~MemTable from destroying a state a second time. + *_row_in_blocks = std::move(temp_row_in_blocks); + _last_sorted_pos = _row_in_blocks->size(); } template void MemTable::_aggregate_for_flexible_partial_update_without_seq_col( - MutableBlock& mutable_block, DorisVector>& temp_row_in_blocks) { - std::shared_ptr prev_row {nullptr}; + MutableBlock& mutable_block, DorisVector& temp_row_in_blocks) { int row_pos = -1; auto& skip_bitmaps = assert_cast(mutable_block.mutable_columns()[_skip_bitmap_col_idx].get()) @@ -657,66 +663,73 @@ void MemTable::_aggregate_for_flexible_partial_update_without_seq_col( auto& delete_signs = assert_cast(mutable_block.mutable_columns()[_delete_sign_col_idx].get()) ->get_data(); - std::shared_ptr row_with_delete_sign {nullptr}; - std::shared_ptr row_without_delete_sign {nullptr}; + // Rows are held by value here: a held row is only appended to + // temp_row_in_blocks once its whole key group has been consumed, so it is + // aggregated into while it lives in one of these two slots. + std::optional row_with_delete_sign; + std::optional row_without_delete_sign; auto finalize_rows = [&]() { - if (row_with_delete_sign != nullptr) { - temp_row_in_blocks.push_back(row_with_delete_sign); - _finalize_one_row(row_with_delete_sign.get(), mutable_block, ++row_pos); - row_with_delete_sign = nullptr; + if (row_with_delete_sign.has_value()) { + temp_row_in_blocks.push_back(*row_with_delete_sign); + _finalize_one_row(temp_row_in_blocks.back(), mutable_block, ++row_pos); + row_with_delete_sign.reset(); } - if (row_without_delete_sign != nullptr) { - temp_row_in_blocks.push_back(row_without_delete_sign); - _finalize_one_row(row_without_delete_sign.get(), mutable_block, ++row_pos); - row_without_delete_sign = nullptr; + if (row_without_delete_sign.has_value()) { + temp_row_in_blocks.push_back(*row_without_delete_sign); + _finalize_one_row(temp_row_in_blocks.back(), mutable_block, ++row_pos); + row_without_delete_sign.reset(); } // _arena.clear(); }; - auto add_row = [&](std::shared_ptr row, bool with_delete_sign) { + auto add_row = [&](const RowInBlock& row, bool with_delete_sign) { if (with_delete_sign) { - row_with_delete_sign = std::move(row); + row_with_delete_sign = row; } else { - row_without_delete_sign = std::move(row); + row_without_delete_sign = row; } }; - for (const auto& cur_row_ptr : *_row_in_blocks) { - RowInBlock* cur_row = cur_row_ptr.get(); - const BitmapValue& skip_bitmap = skip_bitmaps[cur_row->_row_pos]; + for (RowInBlock& cur_row : *_row_in_blocks) { + const BitmapValue& skip_bitmap = skip_bitmaps[cur_row._row_pos]; bool cur_row_has_delete_sign = (!skip_bitmap.contains(_delete_sign_col_unique_id) && - delete_signs[cur_row->_row_pos] != 0); - prev_row = - (row_with_delete_sign == nullptr) ? row_without_delete_sign : row_with_delete_sign; + delete_signs[cur_row._row_pos] != 0); // compare keys, the keys of row_with_delete_sign and row_without_delete_sign is the same, // choose any of them if it's valid - if (prev_row != nullptr && (*_vec_row_comparator)(prev_row.get(), cur_row) == 0) { + RowInBlock* prev_row = + row_with_delete_sign.has_value() + ? &row_with_delete_sign.value() + : (row_without_delete_sign.has_value() ? &row_without_delete_sign.value() + : nullptr); + if (prev_row != nullptr && (*_vec_row_comparator)(prev_row, &cur_row) == 0) { if (cur_row_has_delete_sign) { - if (row_without_delete_sign != nullptr) { + if (row_without_delete_sign.has_value()) { // if there exits row without delete sign, remove it first - _merge_row_binlog_lsn(row_without_delete_sign.get(), cur_row); - _clear_row_agg(row_without_delete_sign.get()); + _merge_row_binlog_lsn(*row_without_delete_sign, cur_row); + _clear_row_agg(*row_without_delete_sign); _stat.merged_rows++; - row_without_delete_sign = nullptr; + row_without_delete_sign.reset(); } // and then unconditionally replace the previous row - prev_row = row_with_delete_sign; + prev_row = + row_with_delete_sign.has_value() ? &row_with_delete_sign.value() : nullptr; } else { - prev_row = row_without_delete_sign; + prev_row = row_without_delete_sign.has_value() ? &row_without_delete_sign.value() + : nullptr; } if (prev_row == nullptr) { - add_row(cur_row_ptr, cur_row_has_delete_sign); + add_row(cur_row, cur_row_has_delete_sign); } else { - if (!prev_row->has_init_agg()) { - _init_row_for_agg(prev_row.get(), mutable_block); + if (!_has_agg(*prev_row)) { + _init_row_for_agg(*prev_row, mutable_block); } _stat.merged_rows++; - _aggregate_two_row_in_block(mutable_block, cur_row, prev_row.get()); + _aggregate_two_row_in_block(mutable_block, cur_row, *prev_row); } } else { finalize_rows(); - add_row(cur_row_ptr, cur_row_has_delete_sign); + add_row(cur_row, cur_row_has_delete_sign); } } // finalize the last lows @@ -725,14 +738,13 @@ void MemTable::_aggregate_for_flexible_partial_update_without_seq_col( template void MemTable::_aggregate_for_flexible_partial_update_with_seq_col( - MutableBlock& mutable_block, DorisVector>& temp_row_in_blocks) { + MutableBlock& mutable_block, DorisVector& temp_row_in_blocks) { // For flexible partial update, when table has sequence column, we don't do any aggregation // in memtable. These duplicate rows will be aggregated in VerticalSegmentWriter int row_pos = -1; - for (const auto& row_ptr : *_row_in_blocks) { - RowInBlock* row = row_ptr.get(); - temp_row_in_blocks.push_back(row_ptr); - _finalize_one_row(row, mutable_block, ++row_pos); + for (const RowInBlock& row : *_row_in_blocks) { + temp_row_in_blocks.push_back(row); + _finalize_one_row(temp_row_in_blocks.back(), mutable_block, ++row_pos); } } @@ -813,7 +825,7 @@ Status MemTable::_to_block(std::unique_ptr* res) { if (_need_row_binlog_lsn) { _output_row_binlog_lsns.reserve(_row_in_blocks->size()); for (const auto& row : *_row_in_blocks) { - _append_output_row_binlog_lsn(row.get()); + _append_output_row_binlog_lsn(row); } } } else { diff --git a/be/src/load/memtable/memtable.h b/be/src/load/memtable/memtable.h index 431608966599d4..1525359f07761a 100644 --- a/be/src/load/memtable/memtable.h +++ b/be/src/load/memtable/memtable.h @@ -50,29 +50,21 @@ enum KeysType : int; // FLUSH: the memtable is under flushing, write segment to disk. enum MemType { ACTIVE = 0, WRITE_FINISHED = 1, FLUSH = 2 }; -// row pos in _input_mutable_block +// A row of _input_mutable_block, kept by value in MemTable::_row_in_blocks. +// Small and trivially copyable on purpose: there is one of these per loaded row, +// so anything stored here is multiplied by the memtable's row count. struct RowInBlock { size_t _row_pos; int64_t _row_binlog_lsn = 0; + // Aggregate state of this row, allocated from MemTable::_arena; null means + // the row has not been aggregated into yet. The offsets of the individual + // states are the same for every row, so they live on the MemTable rather + // than being repeated here. char* _agg_mem = nullptr; - size_t* _agg_state_offset = nullptr; - bool _has_init_agg; - RowInBlock(size_t row) : _row_pos(row), _has_init_agg(false) {} + RowInBlock(size_t row) : _row_pos(row) {} RowInBlock(size_t row, int64_t row_binlog_lsn) - : _row_pos(row), _row_binlog_lsn(row_binlog_lsn), _has_init_agg(false) {} - - void init_agg_places(char* agg_mem, size_t* agg_state_offset) { - _has_init_agg = true; - _agg_mem = agg_mem; - _agg_state_offset = agg_state_offset; - } - - char* agg_places(size_t offset) const { return _agg_mem + _agg_state_offset[offset]; } - - inline bool has_init_agg() const { return _has_init_agg; } - - inline void remove_init_agg() { _has_init_agg = false; } + : _row_pos(row), _row_binlog_lsn(row_binlog_lsn) {} }; class Tie { @@ -217,18 +209,19 @@ class MemTable { private: // for vectorized template - void _aggregate_two_row_in_block(MutableBlock& mutable_block, RowInBlock* new_row, - RowInBlock* row_in_skiplist); + void _aggregate_two_row_in_block(MutableBlock& mutable_block, const RowInBlock& new_row, + RowInBlock& row_in_skiplist); // Merge row-binlog LSN sidecar only when MemTable merges two RowInBlock objects. // Table models that require complex merge semantics, such as AGG tables and unique key // merge-on-read tables, do not support row-binlog LSN now and are rejected in insert(). - void _merge_row_binlog_lsn(RowInBlock* src_row, RowInBlock* dst_row); + void _merge_row_binlog_lsn(const RowInBlock& src_row, RowInBlock& dst_row); - void _append_output_row_binlog_lsn(RowInBlock* row); + void _append_output_row_binlog_lsn(const RowInBlock& row); - void _aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, RowInBlock* new_row, - RowInBlock* row_in_skiplist); + void _aggregate_two_row_with_sequence_map(MutableBlock& mutable_block, + const RowInBlock& new_row, + RowInBlock& row_in_skiplist); // Used to wrapped by to_block to do exception handle logic Status _to_block(std::unique_ptr* res); @@ -275,25 +268,28 @@ 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); + void _sort_one_column(DorisVector& row_in_blocks, Tie& tie, + std::function cmp); template - void _finalize_one_row(RowInBlock* row, MutableBlock& mutable_block, int row_pos); - void _init_row_for_agg(RowInBlock* row, MutableBlock& mutable_block); - void _clear_row_agg(RowInBlock* row); + void _finalize_one_row(RowInBlock& row, MutableBlock& mutable_block, int row_pos); + void _init_row_for_agg(RowInBlock& row, MutableBlock& mutable_block); + void _clear_row_agg(RowInBlock& row); + + static bool _has_agg(const RowInBlock& row) { return row._agg_mem != nullptr; } + char* _agg_place(const RowInBlock& row, size_t cid) const { + return row._agg_mem + _offsets_of_aggregate_states[cid]; + } template void _aggregate(); template void _aggregate_for_flexible_partial_update_without_seq_col( - MutableBlock& mutable_block, - DorisVector>& temp_row_in_blocks); + MutableBlock& mutable_block, DorisVector& temp_row_in_blocks); template void _aggregate_for_flexible_partial_update_with_seq_col( - MutableBlock& mutable_block, - DorisVector>& temp_row_in_blocks); + MutableBlock& mutable_block, DorisVector& temp_row_in_blocks); Status _put_into_output(Block& in_block); bool _is_first_insertion; @@ -302,7 +298,7 @@ class MemTable { std::vector _agg_functions; std::vector _offsets_of_aggregate_states; size_t _total_size_of_aggregate_states; - std::unique_ptr>> _row_in_blocks; + std::unique_ptr> _row_in_blocks; size_t _num_columns; int32_t _seq_col_idx_in_block {-1}; diff --git a/be/test/load/memtable/memtable_sort_test.cpp b/be/test/load/memtable/memtable_sort_test.cpp index 53e92e3c4bb2be..13cb77102ad1a6 100644 --- a/be/test/load/memtable/memtable_sort_test.cpp +++ b/be/test/load/memtable/memtable_sort_test.cpp @@ -17,11 +17,200 @@ #include +#include +#include +#include +#include + +#include "core/block/block.h" +#include "core/column/column_complex.h" +#include "core/column/column_nullable.h" +#include "core/column/column_string.h" +#include "core/column/column_vector.h" +#include "load/delta_writer/delta_writer_context.h" #include "load/memtable/memtable.h" +#include "runtime/descriptor_helper.h" +#include "runtime/descriptors.h" +#include "runtime/memory/mem_tracker_limiter.h" +#include "runtime/workload_management/resource_context.h" +#include "storage/tablet/tablet_schema.h" namespace doris { -class MemTableSortTest : public ::testing::Test {}; +namespace { + +// Schema used by every case: k1 INT (key), k2 VARCHAR (key, nullable), v INT. +// Two key columns are needed so the equal-range refinement between key columns +// is exercised, and k2 is nullable so the ColumnNullable sort path is covered. +// AGG_KEYS additionally gets bm BITMAP BITMAP_UNION. Its aggregate state owns +// heap memory, unlike SUM over an int, so releasing a state twice is an actual +// double free there and the shrink-round cases below can catch it. +bool has_bitmap_col(KeysType keys_type) { + return keys_type == KeysType::AGG_KEYS; +} + +TabletSchemaSPtr create_schema(KeysType keys_type) { + TabletSchemaPB pb; + pb.set_keys_type(keys_type); + + auto add = [&](const std::string& name, const std::string& type, bool is_key, bool nullable, + int32_t length, const std::string& agg) { + ColumnPB* c = pb.add_column(); + c->set_unique_id(pb.column_size()); + c->set_name(name); + c->set_type(type); + c->set_is_key(is_key); + c->set_is_nullable(nullable); + c->set_length(length); + c->set_aggregation(agg); + c->set_is_bf_column(false); + }; + add("k1", "INT", true, false, 4, "NONE"); + add("k2", "VARCHAR", true, true, 20, "NONE"); + // value aggregation only matters for AGG_KEYS; UNIQUE_KEYS always replaces + add("v", "INT", false, false, 4, keys_type == KeysType::AGG_KEYS ? "SUM" : "REPLACE"); + if (has_bitmap_col(keys_type)) { + add("bm", "BITMAP", false, false, 16, "BITMAP_UNION"); + } + + auto schema = std::make_shared(); + schema->init_from_pb(pb); + return schema; +} + +TDescriptorTable create_descriptor_table(KeysType keys_type) { + TDescriptorTableBuilder dtb; + TTupleDescriptorBuilder tuple_builder; + tuple_builder.add_slot(TSlotDescriptorBuilder() + .type(TYPE_INT) + .nullable(false) + .column_name("k1") + .column_pos(0) + .build()); + tuple_builder.add_slot(TSlotDescriptorBuilder() + .string_type(20) + .nullable(true) + .column_name("k2") + .column_pos(1) + .build()); + tuple_builder.add_slot(TSlotDescriptorBuilder() + .type(TYPE_INT) + .nullable(false) + .column_name("v") + .column_pos(2) + .build()); + if (has_bitmap_col(keys_type)) { + tuple_builder.add_slot(TSlotDescriptorBuilder() + .type(TYPE_BITMAP) + .nullable(false) + .column_name("bm") + .column_pos(3) + .build()); + } + tuple_builder.build(&dtb); + return dtb.desc_tbl(); +} + +struct Row { + int32_t k1; + const char* k2; // nullptr means SQL NULL + int32_t v; +}; + +} // namespace + +class MemTableSortTest : public testing::Test { +protected: + // Feeds `rows` through a MemTable in `batches` insert() calls and hands back + // the flushed block in `out`. Going through to_block() means the real _sort() + // runs. Fatal assertions abort this helper, so callers wrap it in + // ASSERT_NO_FATAL_FAILURE rather than reading a half-built block. + void run(KeysType keys_type, const std::vector& rows, size_t batches, + std::unique_ptr* out, bool shrink_between_batches = false) { + TabletSchemaSPtr schema = create_schema(keys_type); + TDescriptorTable tdesc = create_descriptor_table(keys_type); + ObjectPool pool; + DescriptorTbl* desc_tbl = nullptr; + ASSERT_TRUE(DescriptorTbl::create(&pool, tdesc, &desc_tbl).ok()); + TupleDescriptor* tuple_desc = desc_tbl->get_tuple_descriptor(0); + ASSERT_NE(nullptr, tuple_desc); + auto resource_ctx = ResourceContext::create_shared(); + // MemTable dereferences this tracker in its constructor, and a freshly + // created context has none; production installs one the same way. + resource_ctx->memory_context()->set_mem_tracker(MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::LOAD, "MemTableSortTest")); + + MemTable mem_table(10000, schema, &tuple_desc->slots(), tuple_desc, + false /*enable_unique_key_mow*/, nullptr /*partial_update_info*/, + resource_ctx, false /*need_row_binlog_lsn*/); + + const size_t per_batch = (rows.size() + batches - 1) / batches; + for (size_t begin = 0; begin < rows.size(); begin += per_batch) { + const size_t end = std::min(begin + per_batch, rows.size()); + Block block; + for (const auto* slot : tuple_desc->slots()) { + block.insert(ColumnWithTypeAndName(slot->get_empty_mutable_column(), slot->type(), + slot->col_name())); + } + auto columns = std::move(block).mutate_columns(); + for (size_t i = begin; i < end; ++i) { + columns[0]->insert_data(reinterpret_cast(&rows[i].k1), + sizeof(rows[i].k1)); + if (rows[i].k2 == nullptr) { + columns[1]->insert_default(); + } else { + columns[1]->insert_data(rows[i].k2, strlen(rows[i].k2)); + } + columns[2]->insert_data(reinterpret_cast(&rows[i].v), + sizeof(rows[i].v)); + if (has_bitmap_col(keys_type)) { + // Enough values to push BitmapValue past its inline SINGLE + // representation into a heap-backed roaring bitmap, so that + // releasing the aggregate state twice is a real double free. + BitmapValue bitmap; + for (uint64_t b = 0; b < 128; ++b) { + bitmap.add(static_cast(rows[i].v) * 100000 + b * 977); + } + assert_cast(columns[3].get())->insert_value(std::move(bitmap)); + } + } + block.set_columns(std::move(columns)); + + TabletAddRowsPayload payload; + for (uint32_t i = 0; i < end - begin; ++i) { + payload.row_idxs.push_back(i); + } + Status st = mem_table.insert(&block, payload); + ASSERT_TRUE(st.ok()) << st; + if (shrink_between_batches && end < rows.size()) { + // Runs a non-final aggregate: rows that survive keep their + // aggregate state and get aggregated into again next round. + // Deliberately skipped for the last batch, so the duplicates it + // introduces are still there for the final aggregate in + // to_block() to fold into those surviving rows. + mem_table.shrink_memtable_by_agg(); + } + } + + Status st = mem_table.to_block(out); + ASSERT_TRUE(st.ok()) << st; + ASSERT_NE(nullptr, *out); + } + + static std::string k2_of(const Block& b, size_t row) { + StringRef ref = b.get_by_position(1).column->get_data_at(row); + return ref.data == nullptr ? std::string("") : ref.to_string(); + } + static int32_t int_of(const Block& b, size_t pos, size_t row) { + ColumnPtr col = b.get_by_position(pos).column; + if (const auto* nullable = check_and_get_column(col.get())) { + col = nullable->get_nested_column_ptr(); + } + return static_cast(col->get_int(row)); + } + static int32_t k1_of(const Block& b, size_t row) { return int_of(b, 0, row); } + static int32_t v_of(const Block& b, size_t row) { return int_of(b, 2, row); } +}; TEST_F(MemTableSortTest, Tie) { auto t0 = Tie {0, 0}; @@ -80,4 +269,126 @@ TEST_F(MemTableSortTest, Tie) { EXPECT_FALSE(it3.next()); } +// Keys are ordered by (k1, k2); the second key column must only be used to +// refine rows that tie on the first one. +TEST_F(MemTableSortTest, DupKeysOrdersByAllKeyColumns) { + std::vector rows = {{2, "b", 20}, {1, "b", 11}, {2, "a", 21}, {1, "a", 10}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::DUP_KEYS, rows, 1, &out)); + ASSERT_EQ(4, out->rows()); + EXPECT_EQ(1, k1_of(*out, 0)); + EXPECT_EQ("a", k2_of(*out, 0)); + EXPECT_EQ(1, k1_of(*out, 1)); + EXPECT_EQ("b", k2_of(*out, 1)); + EXPECT_EQ(2, k1_of(*out, 2)); + EXPECT_EQ("a", k2_of(*out, 2)); + EXPECT_EQ(2, k1_of(*out, 3)); + EXPECT_EQ("b", k2_of(*out, 3)); +} + +// Rows sharing the whole key are stabilised on descending row position for +// DUP_KEYS, i.e. reverse insertion order. Nothing depends on that direction in +// principle, but a number of regression cases record it, so pin it down. +TEST_F(MemTableSortTest, DupKeysReversesEqualKeys) { + std::vector rows = {{1, "a", 100}, {1, "a", 101}, {1, "a", 102}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::DUP_KEYS, rows, 1, &out)); + ASSERT_EQ(3, out->rows()); + EXPECT_EQ(102, v_of(*out, 0)); + EXPECT_EQ(101, v_of(*out, 1)); + EXPECT_EQ(100, v_of(*out, 2)); +} + +// NULL sorts before any value, matching the nan_direction_hint = -1 the previous +// comparator used. +TEST_F(MemTableSortTest, NullKeySortsFirst) { + std::vector rows = {{1, "b", 2}, {1, nullptr, 1}, {1, "a", 3}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::DUP_KEYS, rows, 1, &out)); + ASSERT_EQ(3, out->rows()); + EXPECT_EQ("", k2_of(*out, 0)); + EXPECT_EQ("a", k2_of(*out, 1)); + EXPECT_EQ("b", k2_of(*out, 2)); +} + +// Splitting the same rows across several insert() calls must not change the +// result: _sort() maps a sorted row position back to its RowInBlock through the +// base of the appended range, which only holds if row positions stay contiguous +// across insert() calls. The equal keys make that mapping observable. +TEST_F(MemTableSortTest, ResultIsIndependentOfBatching) { + std::vector rows = {{3, "c", 1}, {1, "a", 2}, {2, "b", 3}, {1, "b", 4}, + {3, "a", 5}, {2, "a", 6}, {1, "a", 7}, {2, "a", 8}}; + std::unique_ptr one; + std::unique_ptr many; + ASSERT_NO_FATAL_FAILURE(run(KeysType::DUP_KEYS, rows, 1, &one)); + ASSERT_NO_FATAL_FAILURE(run(KeysType::DUP_KEYS, rows, 3, &many)); + ASSERT_EQ(one->rows(), many->rows()); + for (size_t i = 0; i < one->rows(); ++i) { + EXPECT_EQ(k1_of(*one, i), k1_of(*many, i)) << "row " << i; + EXPECT_EQ(k2_of(*one, i), k2_of(*many, i)) << "row " << i; + EXPECT_EQ(v_of(*one, i), v_of(*many, i)) << "row " << i; + } +} + +// For UNIQUE_KEYS the last inserted row must win, which relies on equal keys +// being ordered ascending by row position before aggregation runs. +TEST_F(MemTableSortTest, UniqueKeysLastWriterWins) { + std::vector rows = {{1, "a", 10}, {2, "b", 20}, {1, "a", 11}, {1, "a", 12}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::UNIQUE_KEYS, rows, 1, &out)); + ASSERT_EQ(2, out->rows()); + EXPECT_EQ(1, k1_of(*out, 0)); + EXPECT_EQ(12, v_of(*out, 0)) << "the last inserted value must survive"; + EXPECT_EQ(2, k1_of(*out, 1)); + EXPECT_EQ(20, v_of(*out, 1)); +} + +// shrink_memtable_by_agg() aggregates without finalising, so a surviving row +// carries its aggregate state into the next round and to_block() finalises it. +// Interleaving that with more inserts must not change the result -- this covers +// the state handoff between rounds, and the release of those states, which is +// where holding rows by value differs most from holding them behind a pointer. +TEST_F(MemTableSortTest, AggKeysSurviveShrinkRounds) { + std::vector rows = {{1, "a", 1}, {2, "b", 10}, {1, "a", 2}, {3, "c", 100}, + {2, "b", 20}, {1, "a", 4}, {2, "b", 30}, {1, "a", 8}}; + std::unique_ptr plain; + std::unique_ptr shrunk; + ASSERT_NO_FATAL_FAILURE(run(KeysType::AGG_KEYS, rows, 4, &plain)); + ASSERT_NO_FATAL_FAILURE(run(KeysType::AGG_KEYS, rows, 4, &shrunk, true)); + ASSERT_EQ(3, shrunk->rows()); + ASSERT_EQ(plain->rows(), shrunk->rows()); + for (size_t i = 0; i < plain->rows(); ++i) { + EXPECT_EQ(k1_of(*plain, i), k1_of(*shrunk, i)) << "row " << i; + EXPECT_EQ(v_of(*plain, i), v_of(*shrunk, i)) << "row " << i; + } + EXPECT_EQ(15, v_of(*shrunk, 0)); // 1 + 2 + 4 + 8 + EXPECT_EQ(60, v_of(*shrunk, 1)); // 10 + 20 + 30 + EXPECT_EQ(100, v_of(*shrunk, 2)); +} + +// Same handoff for UNIQUE_KEYS, where a round must keep the newest row rather +// than accumulate. +TEST_F(MemTableSortTest, UniqueKeysSurviveShrinkRounds) { + std::vector rows = {{1, "a", 10}, {2, "b", 20}, {1, "a", 11}, + {2, "b", 21}, {1, "a", 12}, {3, "c", 30}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::UNIQUE_KEYS, rows, 3, &out, true)); + ASSERT_EQ(3, out->rows()); + EXPECT_EQ(12, v_of(*out, 0)); + EXPECT_EQ(21, v_of(*out, 1)); + EXPECT_EQ(30, v_of(*out, 2)); +} + +// AGG_KEYS with SUM: every duplicate must be folded into the group exactly once. +TEST_F(MemTableSortTest, AggKeysSumsDuplicates) { + std::vector rows = {{1, "a", 1}, {2, "b", 100}, {1, "a", 2}, {1, "a", 4}, {2, "b", 200}}; + std::unique_ptr out; + ASSERT_NO_FATAL_FAILURE(run(KeysType::AGG_KEYS, rows, 1, &out)); + ASSERT_EQ(2, out->rows()); + EXPECT_EQ(1, k1_of(*out, 0)); + EXPECT_EQ(7, v_of(*out, 0)); + EXPECT_EQ(2, k1_of(*out, 1)); + EXPECT_EQ(300, v_of(*out, 1)); +} + } // namespace doris