zhannngchen commented on code in PR #21177: URL: https://github.com/apache/doris/pull/21177#discussion_r1251629005
########## be/src/olap/full_compaction.cpp: ########## @@ -0,0 +1,143 @@ +// 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 <time.h> + +#include <mutex> +#include <ostream> + +#include "common/config.h" +#include "olap/olap_define.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>(); + } + + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); + if (!lock.owns_lock()) { + LOG(WARNING) << "another full compaction is running. tablet=" << _tablet->full_name(); + return Status::Error<TRY_LOCK_FAILED>(); + } + + // 1. pick rowsets to compact + RETURN_IF_ERROR(pick_rowsets_to_compact()); + TRACE_COUNTER_INCREMENT("input_rowsets_count", _input_rowsets.size()); + _tablet->set_clone_occurred(false); + + return Status::OK(); +} + +Status FullCompaction::execute_compact_impl() { + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); + if (!lock.owns_lock()) { + LOG(WARNING) << "another full compaction is running. tablet=" << _tablet->full_name(); + return Status::Error<TRY_LOCK_FAILED>(); + } + + // 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>(); + } + + SCOPED_ATTACH_TASK(_mem_tracker); + + // 2. do base compaction, merge rowsets + int64_t permits = get_compaction_permits(); + RETURN_IF_ERROR(do_compaction(permits)); + + // 3. set state to success + _state = CompactionState::SUCCESS; + + 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_rowset_overlapping(_input_rowsets)); Review Comment: You don't need to check overlapping ########## be/src/http/action/compaction_action.cpp: ########## @@ -229,6 +231,12 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet, << ", table=" << tablet->full_name(); } } + } else if (compaction_type == PARAM_COMPACTION_FULL) { + FullCompaction full_compaction(tablet); + res = full_compaction.compact(); + if (!res) { + // todo Review Comment: update here ########## be/src/olap/full_compaction.cpp: ########## @@ -0,0 +1,143 @@ +// 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 <time.h> + +#include <mutex> +#include <ostream> + +#include "common/config.h" +#include "olap/olap_define.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>(); + } + + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); + if (!lock.owns_lock()) { + LOG(WARNING) << "another full compaction is running. tablet=" << _tablet->full_name(); + return Status::Error<TRY_LOCK_FAILED>(); + } + + // 1. pick rowsets to compact + RETURN_IF_ERROR(pick_rowsets_to_compact()); + TRACE_COUNTER_INCREMENT("input_rowsets_count", _input_rowsets.size()); + _tablet->set_clone_occurred(false); + + return Status::OK(); +} + +Status FullCompaction::execute_compact_impl() { + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); Review Comment: same as above ########## be/src/olap/full_compaction.cpp: ########## @@ -0,0 +1,143 @@ +// 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 <time.h> + +#include <mutex> +#include <ostream> + +#include "common/config.h" +#include "olap/olap_define.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>(); + } + + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); + if (!lock.owns_lock()) { + LOG(WARNING) << "another full compaction is running. tablet=" << _tablet->full_name(); + return Status::Error<TRY_LOCK_FAILED>(); + } + + // 1. pick rowsets to compact + RETURN_IF_ERROR(pick_rowsets_to_compact()); + TRACE_COUNTER_INCREMENT("input_rowsets_count", _input_rowsets.size()); + _tablet->set_clone_occurred(false); + + return Status::OK(); +} + +Status FullCompaction::execute_compact_impl() { + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); + if (!lock.owns_lock()) { + LOG(WARNING) << "another full compaction is running. tablet=" << _tablet->full_name(); + return Status::Error<TRY_LOCK_FAILED>(); + } + + // 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>(); + } + + SCOPED_ATTACH_TASK(_mem_tracker); + + // 2. do base compaction, merge rowsets + int64_t permits = get_compaction_permits(); + RETURN_IF_ERROR(do_compaction(permits)); + + // 3. set state to success + _state = CompactionState::SUCCESS; + + 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_rowset_overlapping(_input_rowsets)); + if (_input_rowsets.size() <= 1) { + return Status::Error<BE_NO_SUITABLE_VERSION>(); + } + + // If there are delete predicate rowsets in tablet, start_version > 0 implies some rowsets before + // delete version cannot apply these delete predicates, which can cause incorrect query result. + // So we must abort this base compaction. + // A typical scenario is that some rowsets before cumulative point are on remote storage. + if (_input_rowsets.front()->start_version() > 0) { Review Comment: 不需要检查这个，这里只需要检查是否包含了所有的rowset就行 这个检查是宇轩加的，为了防止有一些冷数据存到了S3上，无法参与base compaction而做的处理 但是full compaction的目的就是为了compact所有的rowset ########## be/src/olap/full_compaction.cpp: ########## @@ -0,0 +1,143 @@ +// 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 <time.h> + +#include <mutex> +#include <ostream> + +#include "common/config.h" +#include "olap/olap_define.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>(); + } + + std::unique_lock<std::mutex> lock(_tablet->get_full_compaction_lock(), std::try_to_lock); Review Comment: You need to acquire the base compaction lock and cu compaction lock as well ########## be/src/http/action/compaction_action.h: ########## @@ -39,6 +39,7 @@ enum class CompactionActionType { const std::string PARAM_COMPACTION_TYPE = "compact_type"; const std::string PARAM_COMPACTION_BASE = "base"; const std::string PARAM_COMPACTION_CUMULATIVE = "cumulative"; +const std::string PARAM_COMPACTION_FULL = "full"; Review Comment: You need to process this new compaction type in `TabletReader::_init_delete_condition` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
