forked from apache/doris
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Feature](Compaction)Support full compaction (apache#21177)
- Loading branch information
1 parent
a7eb186
commit c409fa0
Showing
16 changed files
with
568 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,215 @@ | ||
| // 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/full_compaction.h" | ||
|
|
||
| #include <glog/logging.h> | ||
| #include <time.h> | ||
|
|
||
| #include <memory> | ||
| #include <mutex> | ||
| #include <ostream> | ||
| #include <shared_mutex> | ||
|
|
||
| #include "common/config.h" | ||
| #include "common/status.h" | ||
| #include "olap/cumulative_compaction_policy.h" | ||
| #include "olap/olap_common.h" | ||
| #include "olap/olap_define.h" | ||
| #include "olap/rowset/beta_rowset.h" | ||
| #include "olap/rowset/rowset.h" | ||
| #include "olap/schema_change.h" | ||
| #include "olap/tablet_meta.h" | ||
| #include "runtime/thread_context.h" | ||
| #include "util/thread.h" | ||
| #include "util/trace.h" | ||
|
|
||
| namespace doris { | ||
| using namespace ErrorCode; | ||
|
|
||
| FullCompaction::FullCompaction(const TabletSharedPtr& tablet) | ||
| : Compaction(tablet, "FullCompaction:" + std::to_string(tablet->tablet_id())) {} | ||
|
|
||
| FullCompaction::~FullCompaction() {} | ||
|
|
||
| Status FullCompaction::prepare_compact() { | ||
| if (!_tablet->init_succeeded()) { | ||
| return Status::Error<INVALID_ARGUMENT>("Full compaction init failed"); | ||
| } | ||
|
|
||
| std::unique_lock full_lock(_tablet->get_full_compaction_lock()); | ||
| std::unique_lock base_lock(_tablet->get_base_compaction_lock()); | ||
| std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock()); | ||
|
|
||
| // 1. pick rowsets to compact | ||
| RETURN_IF_ERROR(pick_rowsets_to_compact()); | ||
| _tablet->set_clone_occurred(false); | ||
|
|
||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::execute_compact_impl() { | ||
| std::unique_lock full_lock(_tablet->get_full_compaction_lock()); | ||
| std::unique_lock base_lock(_tablet->get_base_compaction_lock()); | ||
| std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock()); | ||
|
|
||
| // Clone task may happen after compaction task is submitted to thread pool, and rowsets picked | ||
| // for compaction may change. In this case, current compaction task should not be executed. | ||
| if (_tablet->get_clone_occurred()) { | ||
| _tablet->set_clone_occurred(false); | ||
| return Status::Error<BE_CLONE_OCCURRED>("get_clone_occurred failed"); | ||
| } | ||
|
|
||
| SCOPED_ATTACH_TASK(_mem_tracker); | ||
|
|
||
| // 2. do full compaction, merge rowsets | ||
| int64_t permits = get_compaction_permits(); | ||
| RETURN_IF_ERROR(do_compaction(permits)); | ||
|
|
||
| // 3. set state to success | ||
| _state = CompactionState::SUCCESS; | ||
|
|
||
| // 4. set cumulative point | ||
| Version last_version = _input_rowsets.back()->version(); | ||
| _tablet->cumulative_compaction_policy()->update_cumulative_point(_tablet.get(), _input_rowsets, | ||
| _output_rowset, last_version); | ||
| VLOG_CRITICAL << "after cumulative compaction, current cumulative point is " | ||
| << _tablet->cumulative_layer_point() << ", tablet=" << _tablet->full_name(); | ||
|
|
||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::pick_rowsets_to_compact() { | ||
| _input_rowsets = _tablet->pick_candidate_rowsets_to_full_compaction(); | ||
| RETURN_IF_ERROR(check_version_continuity(_input_rowsets)); | ||
| RETURN_IF_ERROR(_check_all_version(_input_rowsets)); | ||
| if (_input_rowsets.size() <= 1) { | ||
| return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version"); | ||
| } | ||
|
|
||
| if (_input_rowsets.size() == 2 && _input_rowsets[0]->end_version() == 1) { | ||
| // the tablet is with rowset: [0-1], [2-y] | ||
| // and [0-1] has no data. in this situation, no need to do full compaction. | ||
| return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version"); | ||
| } | ||
|
|
||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::modify_rowsets(const Merger::Statistics* stats) { | ||
| if (_tablet->keys_type() == KeysType::UNIQUE_KEYS && | ||
| _tablet->enable_unique_key_merge_on_write()) { | ||
| RETURN_IF_ERROR( | ||
| _full_compaction_update_delete_bitmap(_output_rowset, _output_rs_writer.get())); | ||
| } | ||
| std::vector<RowsetSharedPtr> output_rowsets(1, _output_rowset); | ||
| RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true)); | ||
| _tablet->save_meta(); | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::_check_all_version(const std::vector<RowsetSharedPtr>& rowsets) { | ||
| if (rowsets.empty()) { | ||
| return Status::Error<FULL_MISS_VERSION>("There is no input rowset when do full compaction"); | ||
| } | ||
| const RowsetSharedPtr& last_rowset = rowsets.back(); | ||
| const RowsetSharedPtr& first_rowset = rowsets.front(); | ||
| if (last_rowset->version() != _tablet->max_version() || first_rowset->version().first != 0) { | ||
| return Status::Error<FULL_MISS_VERSION>( | ||
| "Full compaction rowsets' versions not equal to all exist rowsets' versions. " | ||
| "full compaction rowsets max version={}-{}" | ||
| ", current rowsets max version={}-{}" | ||
| "full compaction rowsets min version={}-{}, current rowsets min version=0-1", | ||
| last_rowset->start_version(), last_rowset->end_version(), | ||
| _tablet->max_version().first, _tablet->max_version().second, | ||
| first_rowset->start_version(), first_rowset->end_version()); | ||
| } | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::_full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset, | ||
| RowsetWriter* rowset_writer) { | ||
| std::vector<RowsetSharedPtr> tmp_rowsets {}; | ||
|
|
||
| // tablet is under alter process. The delete bitmap will be calculated after conversion. | ||
| if (_tablet->tablet_state() == TABLET_NOTREADY && | ||
| SchemaChangeHandler::tablet_in_converting(_tablet->tablet_id())) { | ||
| LOG(INFO) << "tablet is under alter process, update delete bitmap later, tablet_id=" | ||
| << _tablet->tablet_id(); | ||
| return Status::OK(); | ||
| } | ||
|
|
||
| int64_t max_version = _tablet->max_version().second; | ||
| DCHECK(max_version >= rowset->version().second); | ||
| if (max_version > rowset->version().second) { | ||
| _tablet->capture_consistent_rowsets({rowset->version().second + 1, max_version}, | ||
| &tmp_rowsets); | ||
| } | ||
|
|
||
| for (const auto& it : tmp_rowsets) { | ||
| const int64_t& cur_version = it->rowset_meta()->start_version(); | ||
| RETURN_IF_ERROR( | ||
| _full_compaction_calc_delete_bitmap(it, rowset, cur_version, rowset_writer)); | ||
| } | ||
|
|
||
| std::lock_guard rowset_update_lock(_tablet->get_rowset_update_lock()); | ||
| std::lock_guard header_lock(_tablet->get_header_lock()); | ||
| for (const auto& it : _tablet->rowset_map()) { | ||
| const int64_t& cur_version = it.first.first; | ||
| const RowsetSharedPtr& published_rowset = it.second; | ||
| if (cur_version > max_version) { | ||
| RETURN_IF_ERROR(_full_compaction_calc_delete_bitmap(published_rowset, rowset, | ||
| cur_version, rowset_writer)); | ||
| } | ||
| } | ||
|
|
||
| return Status::OK(); | ||
| } | ||
|
|
||
| Status FullCompaction::_full_compaction_calc_delete_bitmap(const RowsetSharedPtr& published_rowset, | ||
| const RowsetSharedPtr& rowset, | ||
| const int64_t& cur_version, | ||
| RowsetWriter* rowset_writer) { | ||
| std::vector<segment_v2::SegmentSharedPtr> segments; | ||
| auto beta_rowset = reinterpret_cast<BetaRowset*>(published_rowset.get()); | ||
| RETURN_IF_ERROR(beta_rowset->load_segments(&segments)); | ||
| DeleteBitmapPtr delete_bitmap = | ||
| std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id()); | ||
| std::vector<RowsetSharedPtr> specified_rowsets(1, rowset); | ||
|
|
||
| OlapStopWatch watch; | ||
| RETURN_IF_ERROR(_tablet->calc_delete_bitmap(published_rowset, segments, specified_rowsets, | ||
| delete_bitmap, cur_version, rowset_writer)); | ||
| size_t total_rows = std::accumulate( | ||
| segments.begin(), segments.end(), 0, | ||
| [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); }); | ||
| VLOG_DEBUG << "[Full compaction] construct delete bitmap tablet: " << _tablet->tablet_id() | ||
| << ", published rowset version: [" << published_rowset->version().first << "-" | ||
| << published_rowset->version().second << "]" | ||
| << ", full compaction rowset version: [" << rowset->version().first << "-" | ||
| << rowset->version().second << "]" | ||
| << ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows; | ||
|
|
||
| for (const auto& [k, v] : delete_bitmap->delete_bitmap) { | ||
| _tablet->tablet_meta()->delete_bitmap().merge({std::get<0>(k), std::get<1>(k), cur_version}, | ||
| v); | ||
| } | ||
|
|
||
| return Status::OK(); | ||
| } | ||
|
|
||
| } // namespace doris |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| // Licensed to the Apache Software Foundation (ASF) under one | ||
| // or more contributor license agreements. See the NOTICE file | ||
| // distributed with this work for additional information | ||
| // regarding copyright ownership. The ASF licenses this file | ||
| // to you under the Apache License, Version 2.0 (the | ||
| // "License"); you may not use this file except in compliance | ||
| // with the License. You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, | ||
| // software distributed under the License is distributed on an | ||
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| // KIND, either express or implied. See the License for the | ||
| // specific language governing permissions and limitations | ||
| // under the License. | ||
|
|
||
| #pragma once | ||
|
|
||
| #include <butil/macros.h> | ||
|
|
||
| #include <string> | ||
| #include <vector> | ||
|
|
||
| #include "common/status.h" | ||
| #include "io/io_common.h" | ||
| #include "olap/compaction.h" | ||
| #include "olap/rowset/rowset.h" | ||
| #include "olap/tablet.h" | ||
|
|
||
| namespace doris { | ||
|
|
||
| class FullCompaction : public Compaction { | ||
| public: | ||
| FullCompaction(const TabletSharedPtr& tablet); | ||
| ~FullCompaction() override; | ||
|
|
||
| Status prepare_compact() override; | ||
| Status execute_compact_impl() override; | ||
| Status modify_rowsets(const Merger::Statistics* stats = nullptr) override; | ||
|
|
||
| std::vector<RowsetSharedPtr> get_input_rowsets() { return _input_rowsets; } | ||
|
|
||
| protected: | ||
| Status pick_rowsets_to_compact() override; | ||
| std::string compaction_name() const override { return "full compaction"; } | ||
|
|
||
| ReaderType compaction_type() const override { return ReaderType::READER_FULL_COMPACTION; } | ||
|
|
||
| private: | ||
| Status _check_all_version(const std::vector<RowsetSharedPtr>& rowsets); | ||
| Status _full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset, | ||
| RowsetWriter* rowset_writer); | ||
| Status _full_compaction_calc_delete_bitmap(const RowsetSharedPtr& published_rowset, | ||
| const RowsetSharedPtr& rowset, | ||
| const int64_t& cur_version, | ||
| RowsetWriter* rowset_writer); | ||
|
|
||
| DISALLOW_COPY_AND_ASSIGN(FullCompaction); | ||
| }; | ||
|
|
||
| } // namespace doris |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.