| // 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. |
| |
| #include "olap/merger.h" |
| |
| #include <gen_cpp/olap_file.pb.h> |
| #include <gen_cpp/types.pb.h> |
| #include <stddef.h> |
| #include <unistd.h> |
| |
| #include <algorithm> |
| #include <iterator> |
| #include <memory> |
| #include <mutex> |
| #include <numeric> |
| #include <ostream> |
| #include <shared_mutex> |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include "cloud/config.h" |
| #include "common/config.h" |
| #include "common/logging.h" |
| #include "common/status.h" |
| #include "olap/base_tablet.h" |
| #include "olap/iterators.h" |
| #include "olap/olap_common.h" |
| #include "olap/olap_define.h" |
| #include "olap/rowid_conversion.h" |
| #include "olap/rowset/rowset.h" |
| #include "olap/rowset/rowset_meta.h" |
| #include "olap/rowset/rowset_writer.h" |
| #include "olap/rowset/segment_v2/segment_writer.h" |
| #include "olap/storage_engine.h" |
| #include "olap/tablet.h" |
| #include "olap/tablet_fwd.h" |
| #include "olap/tablet_meta.h" |
| #include "olap/tablet_reader.h" |
| #include "olap/utils.h" |
| #include "util/slice.h" |
| #include "vec/core/block.h" |
| #include "vec/olap/block_reader.h" |
| #include "vec/olap/vertical_block_reader.h" |
| #include "vec/olap/vertical_merge_iterator.h" |
| |
| namespace doris { |
| #include "common/compile_check_begin.h" |
| Status Merger::vmerge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type, |
| const TabletSchema& cur_tablet_schema, |
| const std::vector<RowsetReaderSharedPtr>& src_rowset_readers, |
| RowsetWriter* dst_rowset_writer, Statistics* stats_output) { |
| if (!cur_tablet_schema.cluster_key_uids().empty()) { |
| return Status::InternalError( |
| "mow table with cluster keys does not support non vertical compaction"); |
| } |
| vectorized::BlockReader reader; |
| TabletReader::ReaderParams reader_params; |
| reader_params.tablet = tablet; |
| reader_params.reader_type = reader_type; |
| |
| TabletReader::ReadSource read_source; |
| read_source.rs_splits.reserve(src_rowset_readers.size()); |
| for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) { |
| read_source.rs_splits.emplace_back(rs_reader); |
| } |
| read_source.fill_delete_predicates(); |
| reader_params.set_read_source(std::move(read_source)); |
| |
| reader_params.version = dst_rowset_writer->version(); |
| |
| TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>(); |
| merge_tablet_schema->copy_from(cur_tablet_schema); |
| |
| // Merge the columns in delete predicate that not in latest schema in to current tablet schema |
| for (auto& del_pred_rs : reader_params.delete_predicates) { |
| merge_tablet_schema->merge_dropped_columns(*del_pred_rs->tablet_schema()); |
| } |
| reader_params.tablet_schema = merge_tablet_schema; |
| if (!tablet->tablet_schema()->cluster_key_uids().empty()) { |
| reader_params.delete_bitmap = &tablet->tablet_meta()->delete_bitmap(); |
| } |
| |
| if (stats_output && stats_output->rowid_conversion) { |
| reader_params.record_rowids = true; |
| reader_params.rowid_conversion = stats_output->rowid_conversion; |
| stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id()); |
| } |
| |
| reader_params.return_columns.resize(cur_tablet_schema.num_columns()); |
| std::iota(reader_params.return_columns.begin(), reader_params.return_columns.end(), 0); |
| reader_params.origin_return_columns = &reader_params.return_columns; |
| RETURN_IF_ERROR(reader.init(reader_params)); |
| |
| vectorized::Block block = cur_tablet_schema.create_block(reader_params.return_columns); |
| size_t output_rows = 0; |
| bool eof = false; |
| while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) { |
| auto tablet_state = tablet->tablet_state(); |
| if (tablet_state != TABLET_RUNNING && tablet_state != TABLET_NOTREADY) { |
| tablet->clear_cache(); |
| return Status::Error<INTERNAL_ERROR>("tablet {} is not used any more", |
| tablet->tablet_id()); |
| } |
| |
| // Read one block from block reader |
| RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof), |
| "failed to read next block when merging rowsets of tablet " + |
| std::to_string(tablet->tablet_id())); |
| RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->add_block(&block), |
| "failed to write block when merging rowsets of tablet " + |
| std::to_string(tablet->tablet_id())); |
| |
| if (reader_params.record_rowids && block.rows() > 0) { |
| std::vector<uint32_t> segment_num_rows; |
| RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows)); |
| stats_output->rowid_conversion->add(reader.current_block_row_locations(), |
| segment_num_rows); |
| } |
| |
| output_rows += block.rows(); |
| block.clear_column_data(); |
| } |
| if (ExecEnv::GetInstance()->storage_engine().stopped()) { |
| return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped", |
| tablet->tablet_id()); |
| } |
| |
| if (stats_output != nullptr) { |
| stats_output->output_rows = output_rows; |
| stats_output->merged_rows = reader.merged_rows(); |
| stats_output->filtered_rows = reader.filtered_rows(); |
| stats_output->bytes_read_from_local = reader.stats().file_cache_stats.bytes_read_from_local; |
| stats_output->bytes_read_from_remote = |
| reader.stats().file_cache_stats.bytes_read_from_remote; |
| stats_output->cached_bytes_total = reader.stats().file_cache_stats.bytes_write_into_cache; |
| if (config::is_cloud_mode()) { |
| stats_output->cloud_local_read_time = |
| reader.stats().file_cache_stats.local_io_timer / 1000; |
| stats_output->cloud_remote_read_time = |
| reader.stats().file_cache_stats.remote_io_timer / 1000; |
| } |
| } |
| |
| RETURN_NOT_OK_STATUS_WITH_WARN(dst_rowset_writer->flush(), |
| "failed to flush rowset when merging rowsets of tablet " + |
| std::to_string(tablet->tablet_id())); |
| |
| return Status::OK(); |
| } |
| |
| // split columns into several groups, make sure all keys in one group |
| // unique_key should consider sequence&delete column |
| void Merger::vertical_split_columns(const TabletSchema& tablet_schema, |
| std::vector<std::vector<uint32_t>>* column_groups, |
| std::vector<uint32_t>* key_group_cluster_key_idxes) { |
| size_t num_key_cols = tablet_schema.num_key_columns(); |
| size_t total_cols = tablet_schema.num_columns(); |
| std::vector<uint32_t> key_columns; |
| for (auto i = 0; i < num_key_cols; ++i) { |
| key_columns.emplace_back(i); |
| } |
| // in unique key, sequence & delete sign column should merge with key columns |
| int32_t sequence_col_idx = -1; |
| int32_t delete_sign_idx = -1; |
| // in key column compaction, seq_col real index is _num_key_columns |
| // and delete_sign column is _block->columns() - 1 |
| if (tablet_schema.keys_type() == KeysType::UNIQUE_KEYS) { |
| if (tablet_schema.has_sequence_col()) { |
| sequence_col_idx = tablet_schema.sequence_col_idx(); |
| key_columns.emplace_back(sequence_col_idx); |
| } |
| delete_sign_idx = tablet_schema.field_index(DELETE_SIGN); |
| if (delete_sign_idx != -1) { |
| key_columns.emplace_back(delete_sign_idx); |
| } |
| if (!tablet_schema.cluster_key_uids().empty()) { |
| for (const auto& cid : tablet_schema.cluster_key_uids()) { |
| auto idx = tablet_schema.field_index(cid); |
| DCHECK(idx >= 0) << "could not find cluster key column with unique_id=" << cid |
| << " in tablet schema, table_id=" << tablet_schema.table_id(); |
| if (idx >= num_key_cols) { |
| key_columns.emplace_back(idx); |
| } |
| } |
| // tablet schema unique ids: [1, 2, 5, 3, 6, 4], [1 2] is key columns |
| // cluster key unique ids: [3, 1, 4] |
| // the key_columns should be [0, 1, 3, 5] |
| // the key_group_cluster_key_idxes should be [2, 1, 3] |
| for (const auto& cid : tablet_schema.cluster_key_uids()) { |
| auto idx = tablet_schema.field_index(cid); |
| for (auto i = 0; i < key_columns.size(); ++i) { |
| if (idx == key_columns[i]) { |
| key_group_cluster_key_idxes->emplace_back(i); |
| break; |
| } |
| } |
| } |
| } |
| } |
| VLOG_NOTICE << "sequence_col_idx=" << sequence_col_idx |
| << ", delete_sign_idx=" << delete_sign_idx; |
| // for duplicate no keys |
| if (!key_columns.empty()) { |
| column_groups->emplace_back(key_columns); |
| } |
| |
| std::vector<uint32_t> value_columns; |
| |
| for (size_t i = num_key_cols; i < total_cols; ++i) { |
| if (i == sequence_col_idx || i == delete_sign_idx || |
| key_columns.end() != std::find(key_columns.begin(), key_columns.end(), i)) { |
| continue; |
| } |
| |
| if (!value_columns.empty() && |
| value_columns.size() % config::vertical_compaction_num_columns_per_group == 0) { |
| column_groups->push_back(value_columns); |
| value_columns.clear(); |
| } |
| value_columns.push_back(cast_set<uint32_t>(i)); |
| } |
| |
| if (!value_columns.empty()) { |
| column_groups->push_back(value_columns); |
| } |
| } |
| |
| Status Merger::vertical_compact_one_group( |
| BaseTabletSPtr tablet, ReaderType reader_type, const TabletSchema& tablet_schema, |
| bool is_key, const std::vector<uint32_t>& column_group, |
| vectorized::RowSourcesBuffer* row_source_buf, |
| 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) { |
| // build tablet reader |
| VLOG_NOTICE << "vertical compact one group, max_rows_per_segment=" << max_rows_per_segment; |
| vectorized::VerticalBlockReader reader(row_source_buf); |
| TabletReader::ReaderParams reader_params; |
| reader_params.is_key_column_group = is_key; |
| reader_params.key_group_cluster_key_idxes = key_group_cluster_key_idxes; |
| reader_params.tablet = tablet; |
| reader_params.reader_type = reader_type; |
| |
| TabletReader::ReadSource read_source; |
| read_source.rs_splits.reserve(src_rowset_readers.size()); |
| for (const RowsetReaderSharedPtr& rs_reader : src_rowset_readers) { |
| read_source.rs_splits.emplace_back(rs_reader); |
| } |
| read_source.fill_delete_predicates(); |
| reader_params.set_read_source(std::move(read_source)); |
| |
| reader_params.version = dst_rowset_writer->version(); |
| |
| TabletSchemaSPtr merge_tablet_schema = std::make_shared<TabletSchema>(); |
| merge_tablet_schema->copy_from(tablet_schema); |
| |
| for (auto& del_pred_rs : reader_params.delete_predicates) { |
| merge_tablet_schema->merge_dropped_columns(*del_pred_rs->tablet_schema()); |
| } |
| |
| reader_params.tablet_schema = merge_tablet_schema; |
| bool has_cluster_key = false; |
| if (!tablet->tablet_schema()->cluster_key_uids().empty()) { |
| reader_params.delete_bitmap = &tablet->tablet_meta()->delete_bitmap(); |
| has_cluster_key = true; |
| } |
| |
| if (is_key && stats_output && stats_output->rowid_conversion) { |
| reader_params.record_rowids = true; |
| reader_params.rowid_conversion = stats_output->rowid_conversion; |
| stats_output->rowid_conversion->set_dst_rowset_id(dst_rowset_writer->rowset_id()); |
| } |
| |
| reader_params.return_columns = column_group; |
| reader_params.origin_return_columns = &reader_params.return_columns; |
| reader_params.batch_size = batch_size; |
| RETURN_IF_ERROR(reader.init(reader_params, sample_info)); |
| |
| vectorized::Block block = tablet_schema.create_block(reader_params.return_columns); |
| size_t output_rows = 0; |
| bool eof = false; |
| while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) { |
| auto tablet_state = tablet->tablet_state(); |
| if (tablet_state != TABLET_RUNNING && tablet_state != TABLET_NOTREADY) { |
| tablet->clear_cache(); |
| return Status::Error<INTERNAL_ERROR>("tablet {} is not used any more", |
| tablet->tablet_id()); |
| } |
| // Read one block from block reader |
| RETURN_NOT_OK_STATUS_WITH_WARN(reader.next_block_with_aggregation(&block, &eof), |
| "failed to read next block when merging rowsets of tablet " + |
| std::to_string(tablet->tablet_id())); |
| RETURN_NOT_OK_STATUS_WITH_WARN( |
| dst_rowset_writer->add_columns(&block, column_group, is_key, max_rows_per_segment, |
| has_cluster_key), |
| "failed to write block when merging rowsets of tablet " + |
| std::to_string(tablet->tablet_id())); |
| |
| if (is_key && reader_params.record_rowids && block.rows() > 0) { |
| std::vector<uint32_t> segment_num_rows; |
| RETURN_IF_ERROR(dst_rowset_writer->get_segment_num_rows(&segment_num_rows)); |
| stats_output->rowid_conversion->add(reader.current_block_row_locations(), |
| segment_num_rows); |
| } |
| output_rows += block.rows(); |
| block.clear_column_data(); |
| } |
| if (ExecEnv::GetInstance()->storage_engine().stopped()) { |
| return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped", |
| tablet->tablet_id()); |
| } |
| |
| if (is_key && stats_output != nullptr) { |
| stats_output->output_rows = output_rows; |
| stats_output->merged_rows = reader.merged_rows(); |
| stats_output->filtered_rows = reader.filtered_rows(); |
| stats_output->bytes_read_from_local = reader.stats().file_cache_stats.bytes_read_from_local; |
| stats_output->bytes_read_from_remote = |
| reader.stats().file_cache_stats.bytes_read_from_remote; |
| stats_output->cached_bytes_total = reader.stats().file_cache_stats.bytes_write_into_cache; |
| if (config::is_cloud_mode()) { |
| stats_output->cloud_local_read_time = |
| reader.stats().file_cache_stats.local_io_timer / 1000; |
| stats_output->cloud_remote_read_time = |
| reader.stats().file_cache_stats.remote_io_timer / 1000; |
| } |
| } |
| RETURN_IF_ERROR(dst_rowset_writer->flush_columns(is_key)); |
| |
| return Status::OK(); |
| } |
| |
| // for segcompaction |
| Status Merger::vertical_compact_one_group( |
| int64_t tablet_id, ReaderType reader_type, const TabletSchema& tablet_schema, bool is_key, |
| const std::vector<uint32_t>& column_group, vectorized::RowSourcesBuffer* row_source_buf, |
| vectorized::VerticalBlockReader& src_block_reader, |
| segment_v2::SegmentWriter& dst_segment_writer, Statistics* stats_output, |
| uint64_t* index_size, KeyBoundsPB& key_bounds, SimpleRowIdConversion* rowid_conversion) { |
| // TODO: record_rowids |
| vectorized::Block block = tablet_schema.create_block(column_group); |
| size_t output_rows = 0; |
| bool eof = false; |
| while (!eof && !ExecEnv::GetInstance()->storage_engine().stopped()) { |
| // Read one block from block reader |
| RETURN_NOT_OK_STATUS_WITH_WARN(src_block_reader.next_block_with_aggregation(&block, &eof), |
| "failed to read next block when merging rowsets of tablet " + |
| std::to_string(tablet_id)); |
| if (!block.rows()) { |
| break; |
| } |
| RETURN_NOT_OK_STATUS_WITH_WARN(dst_segment_writer.append_block(&block, 0, block.rows()), |
| "failed to write block when merging rowsets of tablet " + |
| std::to_string(tablet_id)); |
| |
| if (is_key && rowid_conversion != nullptr) { |
| rowid_conversion->add(src_block_reader.current_block_row_locations()); |
| } |
| output_rows += block.rows(); |
| block.clear_column_data(); |
| } |
| if (ExecEnv::GetInstance()->storage_engine().stopped()) { |
| return Status::Error<INTERNAL_ERROR>("tablet {} failed to do compaction, engine stopped", |
| tablet_id); |
| } |
| |
| if (is_key && stats_output != nullptr) { |
| stats_output->output_rows = output_rows; |
| stats_output->merged_rows = src_block_reader.merged_rows(); |
| stats_output->filtered_rows = src_block_reader.filtered_rows(); |
| stats_output->bytes_read_from_local = |
| src_block_reader.stats().file_cache_stats.bytes_read_from_local; |
| stats_output->bytes_read_from_remote = |
| src_block_reader.stats().file_cache_stats.bytes_read_from_remote; |
| stats_output->cached_bytes_total = |
| src_block_reader.stats().file_cache_stats.bytes_write_into_cache; |
| } |
| |
| // segcompaction produce only one segment at once |
| RETURN_IF_ERROR(dst_segment_writer.finalize_columns_data()); |
| RETURN_IF_ERROR(dst_segment_writer.finalize_columns_index(index_size)); |
| |
| if (is_key) { |
| Slice min_key = dst_segment_writer.min_encoded_key(); |
| Slice max_key = dst_segment_writer.max_encoded_key(); |
| DCHECK_LE(min_key.compare(max_key), 0); |
| key_bounds.set_min_key(min_key.to_string()); |
| key_bounds.set_max_key(max_key.to_string()); |
| } |
| |
| return Status::OK(); |
| } |
| |
| int64_t estimate_batch_size(int group_index, BaseTabletSPtr tablet, int64_t way_cnt) { |
| std::unique_lock<std::mutex> lock(tablet->sample_info_lock); |
| CompactionSampleInfo info = tablet->sample_infos[group_index]; |
| if (way_cnt <= 0) { |
| LOG(INFO) << "estimate batch size for vertical compaction, tablet id: " |
| << tablet->tablet_id() << " way cnt: " << way_cnt; |
| return 4096 - 32; |
| } |
| int64_t block_mem_limit = config::compaction_memory_bytes_limit / way_cnt; |
| if (tablet->last_compaction_status.is<ErrorCode::MEM_LIMIT_EXCEEDED>()) { |
| block_mem_limit /= 4; |
| } |
| |
| int64_t group_data_size = 0; |
| if (info.group_data_size > 0 && info.bytes > 0 && info.rows > 0) { |
| double smoothing_factor = 0.5; |
| group_data_size = |
| int64_t((cast_set<double>(info.group_data_size) * (1 - smoothing_factor)) + |
| (cast_set<double>(info.bytes / info.rows) * smoothing_factor)); |
| tablet->sample_infos[group_index].group_data_size = group_data_size; |
| } else if (info.group_data_size > 0 && (info.bytes <= 0 || info.rows <= 0)) { |
| group_data_size = info.group_data_size; |
| } else if (info.group_data_size <= 0 && info.bytes > 0 && info.rows > 0) { |
| group_data_size = info.bytes / info.rows; |
| tablet->sample_infos[group_index].group_data_size = group_data_size; |
| } else { |
| LOG(INFO) << "estimate batch size for vertical compaction, tablet id: " |
| << tablet->tablet_id() << " group data size: " << info.group_data_size |
| << " row num: " << info.rows << " consume bytes: " << info.bytes; |
| return 1024 - 32; |
| } |
| |
| if (group_data_size <= 0) { |
| LOG(WARNING) << "estimate batch size for vertical compaction, tablet id: " |
| << tablet->tablet_id() << " unexpected group data size: " << group_data_size; |
| return 4096 - 32; |
| } |
| |
| tablet->sample_infos[group_index].bytes = 0; |
| tablet->sample_infos[group_index].rows = 0; |
| |
| int64_t batch_size = block_mem_limit / group_data_size; |
| int64_t res = std::max(std::min(batch_size, int64_t(4096 - 32)), int64_t(32L)); |
| LOG(INFO) << "estimate batch size for vertical compaction, tablet id: " << tablet->tablet_id() |
| << " group data size: " << info.group_data_size << " row num: " << info.rows |
| << " consume bytes: " << info.bytes << " way cnt: " << way_cnt |
| << " batch size: " << res; |
| return res; |
| } |
| |
| // steps to do vertical merge: |
| // 1. split columns into column groups |
| // 2. compact groups one by one, generate a row_source_buf when compact key group |
| // and use this row_source_buf to compact value column groups |
| // 3. build output rowset |
| Status Merger::vertical_merge_rowsets(BaseTabletSPtr tablet, ReaderType reader_type, |
| const TabletSchema& tablet_schema, |
| const std::vector<RowsetReaderSharedPtr>& src_rowset_readers, |
| RowsetWriter* dst_rowset_writer, |
| uint32_t max_rows_per_segment, int64_t merge_way_num, |
| Statistics* stats_output) { |
| LOG(INFO) << "Start to do vertical compaction, tablet_id: " << tablet->tablet_id(); |
| std::vector<std::vector<uint32_t>> column_groups; |
| std::vector<uint32_t> key_group_cluster_key_idxes; |
| vertical_split_columns(tablet_schema, &column_groups, &key_group_cluster_key_idxes); |
| |
| vectorized::RowSourcesBuffer row_sources_buf( |
| tablet->tablet_id(), dst_rowset_writer->context().tablet_path, reader_type); |
| { |
| std::unique_lock<std::mutex> lock(tablet->sample_info_lock); |
| tablet->sample_infos.resize(column_groups.size(), {0, 0, 0}); |
| } |
| // compact group one by one |
| for (auto i = 0; i < column_groups.size(); ++i) { |
| VLOG_NOTICE << "row source size: " << row_sources_buf.total_size(); |
| bool is_key = (i == 0); |
| int64_t batch_size = config::compaction_batch_size != -1 |
| ? config::compaction_batch_size |
| : estimate_batch_size(i, tablet, merge_way_num); |
| CompactionSampleInfo sample_info; |
| 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, stats_output, |
| key_group_cluster_key_idxes, batch_size, &sample_info); |
| { |
| std::unique_lock<std::mutex> lock(tablet->sample_info_lock); |
| tablet->sample_infos[i] = sample_info; |
| } |
| RETURN_IF_ERROR(st); |
| if (is_key) { |
| RETURN_IF_ERROR(row_sources_buf.flush()); |
| } |
| RETURN_IF_ERROR(row_sources_buf.seek_to_begin()); |
| } |
| |
| // finish compact, build output rowset |
| VLOG_NOTICE << "finish compact groups"; |
| RETURN_IF_ERROR(dst_rowset_writer->final_flush()); |
| |
| return Status::OK(); |
| } |
| #include "common/compile_check_end.h" |
| } // namespace doris |
