wangbo commented on code in PR #54032: URL: https://github.com/apache/doris/pull/54032#discussion_r2278378331
########## be/src/cloud/cloud_index_change_compaction.cpp: ########## @@ -0,0 +1,558 @@ +// 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 "cloud/cloud_index_change_compaction.h" + +#include "cloud/cloud_meta_mgr.h" +#include "cloud/config.h" +#include "common/status.h" +#include "cpp/sync_point.h" + +namespace doris { + +CloudIndexChangeCompaction::~CloudIndexChangeCompaction() = default; + +CloudIndexChangeCompaction::CloudIndexChangeCompaction(CloudStorageEngine& engine, + CloudTabletSPtr tablet, bool is_drop, + std::vector<TOlapTableIndex>& alter_indexes) + : CloudCompactionMixin(engine, tablet, + "CloudIndexChangeCompaction:" + std::to_string(tablet->tablet_id())), + _is_drop(is_drop), + _alter_indexes(alter_indexes) {} + +Status CloudIndexChangeCompaction::prepare_compact() { + TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudIndexChangeCompaction::prepare_compact", Status::OK()); + + std::set<int64_t> alter_index_ids; + for (auto index : _alter_indexes) { + alter_index_ids.insert(index.index_id); + } + + if (alter_index_ids.size() == 0) { + LOG(WARNING) << "[index_change] no index is specified."; + return Status::InternalError("no specified index."); + } else { + VLOG_DEBUG << "[index_change] alter_index_ids size=" << alter_index_ids.size(); + } + + if (_tablet->tablet_state() != TABLET_RUNNING) { + LOG(WARNING) << "[index_change] tablet state is not running. tablet_id=" + << _tablet->tablet_id(); + return Status::InternalError("invalid tablet state. tablet_id={}", _tablet->tablet_id()); + } + + RETURN_IF_ERROR(cloud_tablet()->sync_rowsets()); + + { + std::shared_lock rlock(_tablet->get_header_lock()); + _base_compaction_cnt = cloud_tablet()->base_compaction_cnt(); + _cumulative_compaction_cnt = cloud_tablet()->cumulative_compaction_cnt(); + } + + bool is_base_rowset = false; + auto input_rowset = DORIS_TRY(cloud_tablet()->pick_a_rowset_for_index_change( + alter_index_ids, _is_drop, is_base_rowset)); + if (input_rowset == nullptr) { + return Status::OK(); Review Comment: 对于CloudIndexChangeCompaction来说，inputrowset为空意味着可以结束了。目前是通过```is_finish_index_change() { return _input_rowsets.size() == 0; }``` 方法判断的，如果prepare结束没有可用的rowset，整个任务就直接结束了。之前没加一个新的status的原因是改status感觉比较重，另外就是is_finish_index_change相当于这个Compaction有一个比较持久的状态标识目前的Compaction已经结束， -- 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]
