be/src/cloud/cloud_full_compaction.cpp - doris - Git at Google


            
            // 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_full_compaction.h"


            #include <gen_cpp/cloud.pb.h>


            #include <boost/container_hash/hash.hpp>


            #include "cloud/cloud_meta_mgr.h"

            
            #include "cloud/cloud_storage_engine.h"

            
            #include "cloud/config.h"

            
            #include "common/config.h"

            
            #include "common/status.h"

            
            #include "core/column/column.h"

            
            #include "cpp/sync_point.h"

            
            #include "service/backend_options.h"

            
            #include "storage/compaction/compaction.h"

            
            #include "storage/rowset/beta_rowset.h"

            
            #include "storage/tablet/tablet_meta.h"

            
            #include "util/debug_points.h"

            
            #include "util/thread.h"

            
            #include "util/uuid_generator.h"


            namespace doris {

            
            using namespace ErrorCode;


            bvar::Adder<uint64_t> full_output_size("full_compaction", "output_size");


            CloudFullCompaction::CloudFullCompaction(CloudStorageEngine& engine, CloudTabletSPtr tablet)

            
                    : CloudCompactionMixin(engine, tablet,

            
                                           "BaseCompaction:" + std::to_string(tablet->tablet_id())) {}


            CloudFullCompaction::~CloudFullCompaction() = default;


            Status CloudFullCompaction::prepare_compact() {

            
                Status st;

            
                Defer defer_set_st([&] {

            
                    if (!st.ok()) {

            
                        cloud_tablet()->set_last_full_compaction_status(st.to_string());

            
                        cloud_tablet()->set_last_full_compaction_failure_time(UnixMillis());

            
                    }

            
                });

            
                if (_tablet->tablet_state() != TABLET_RUNNING) {

            
                    st = Status::InternalError("invalid tablet state. tablet_id={}", _tablet->tablet_id());

            
                    return st;

            
                }


                // always sync latest rowset for full compaction

            
                st = cloud_tablet()->sync_rowsets();

            
                RETURN_IF_ERROR(st);


                st = pick_rowsets_to_compact();

            
                RETURN_IF_ERROR(st);


                for (auto& rs : _input_rowsets) {

            
                    _input_row_num += rs->num_rows();

            
                    _input_segments += rs->num_segments();

            
                    _input_rowsets_data_size += rs->data_disk_size();

            
                    _input_rowsets_index_size += rs->index_disk_size();

            
                    _input_rowsets_total_size += rs->total_disk_size();

            
                }

            
                LOG_INFO("start CloudFullCompaction, tablet_id={}, range=[{}-{}]", _tablet->tablet_id(),

            
                         _input_rowsets.front()->start_version(), _input_rowsets.back()->end_version())

            
                        .tag("job_id", _uuid)

            
                        .tag("input_rowsets", _input_rowsets.size())

            
                        .tag("input_rows", _input_row_num)

            
                        .tag("input_segments", _input_segments)

            
                        .tag("input_rowsets_data_size", _input_rowsets_data_size)

            
                        .tag("input_rowsets_index_size", _input_rowsets_index_size)

            
                        .tag("input_rowsets_total_size", _input_rowsets_total_size);

            
                return Status::OK();

            
            }

            
            Status CloudFullCompaction::request_global_lock() {

            
                // prepare compaction job

            
                cloud::TabletJobInfoPB job;

            
                auto idx = job.mutable_idx();

            
                idx->set_tablet_id(_tablet->tablet_id());

            
                idx->set_table_id(_tablet->table_id());

            
                idx->set_index_id(_tablet->index_id());

            
                idx->set_partition_id(_tablet->partition_id());

            
                auto compaction_job = job.add_compaction();

            
                compaction_job->set_id(_uuid);

            
                compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +

            
                                              std::to_string(config::heartbeat_service_port));

            
                compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);

            
                compaction_job->set_base_compaction_cnt(_base_compaction_cnt);

            
                compaction_job->set_cumulative_compaction_cnt(_cumulative_compaction_cnt);

            
                using namespace std::chrono;

            
                int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();

            
                _expiration = now + config::compaction_timeout_seconds;

            
                compaction_job->set_expiration(_expiration);

            
                compaction_job->set_lease(now + config::lease_compaction_interval_seconds * 4);

            
                // Set input version range to let meta-service judge version range conflict

            
                compaction_job->add_input_versions(_input_rowsets.front()->start_version());

            
                compaction_job->add_input_versions(_input_rowsets.back()->end_version());

            
                cloud::StartTabletJobResponse resp;

            
                auto st = _engine.meta_mgr().prepare_tablet_job(job, &resp);

            
                if (!st.ok()) {

            
                    if (resp.status().code() == cloud::STALE_TABLET_CACHE) {

            
                        // set last_sync_time to 0 to force sync tablet next time

            
                        cloud_tablet()->last_sync_rowsets_time_s = 0;

            
                    } else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {

            
                        // tablet not found

            
                        cloud_tablet()->clear_cache();

            
                    }

            
                }

            
                return st;

            
            }


            Status CloudFullCompaction::pick_rowsets_to_compact() {

            
                _input_rowsets.clear();

            
                {

            
                    std::shared_lock rlock(_tablet->get_header_lock());

            
                    _base_compaction_cnt = cloud_tablet()->base_compaction_cnt();

            
                    _cumulative_compaction_cnt = cloud_tablet()->cumulative_compaction_cnt();

            
                    _input_rowsets = cloud_tablet()->pick_candidate_rowsets_to_full_compaction_unlocked();

            
                }

            
                if (auto st = check_version_continuity(_input_rowsets); !st.ok()) {

            
                    DCHECK(false) << st;

            
                    return st;

            
                }

            
                if (_input_rowsets.size() <= 1) {

            
                    return Status::Error<BE_NO_SUITABLE_VERSION>(

            
                            "insufficent compaction input rowset, #rowsets={}", _input_rowsets.size());

            
                }


                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<BE_NO_SUITABLE_VERSION>("no suitable versions for compaction");

            
                }


                return Status::OK();

            
            }


            Status CloudFullCompaction::execute_compact() {

            
                DBUG_EXECUTE_IF("CloudFullCompaction::execute_compact.block", {

            
                    auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);

            
                    LOG_INFO(

            
                            "[verbose] CloudFullCompaction::execute_compact.block, target_tablet_id={}, "

            
                            "tablet_id={}",

            
                            target_tablet_id, cloud_tablet()->tablet_id());

            
                    if (target_tablet_id == cloud_tablet()->tablet_id()) {

            
                        DBUG_BLOCK;

            
                    }

            
                });

            
                TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudFullCompaction::execute_compact_impl", Status::OK(),

            
                                                  this);

            
            #ifndef __APPLE__

            
                if (config::enable_base_compaction_idle_sched) {

            
                    Thread::set_idle_sched();

            
                }

            
            #endif


                SCOPED_ATTACH_TASK(_mem_tracker);


                using namespace std::chrono;

            
                auto start = steady_clock::now();

            
                auto res = CloudCompactionMixin::execute_compact();

            
                cloud_tablet()->set_last_full_compaction_status(res.to_string());

            
                if (!res.ok()) {

            
                    cloud_tablet()->set_last_full_compaction_failure_time(UnixMillis());

            
                    LOG(WARNING) << "fail to do " << compaction_name() << ". res=" << res

            
                                 << ", tablet=" << _tablet->tablet_id()

            
                                 << ", output_version=" << _output_version;

            
                    return res;

            
                }

            
                LOG_INFO("finish CloudFullCompaction, tablet_id={}, cost={}ms", _tablet->tablet_id(),

            
                         duration_cast<milliseconds>(steady_clock::now() - start).count())

            
                        .tag("job_id", _uuid)

            
                        .tag("input_rowsets", _input_rowsets.size())

            
                        .tag("input_rows", _input_row_num)

            
                        .tag("input_segments", _input_segments)

            
                        .tag("input_rowsets_data_size", _input_rowsets_data_size)

            
                        .tag("input_rowsets_index_size", _input_rowsets_index_size)

            
                        .tag("input_rowsets_total_size", _input_rowsets_total_size)

            
                        .tag("output_rows", _output_rowset->num_rows())

            
                        .tag("output_segments", _output_rowset->num_segments())

            
                        .tag("output_rowset_data_size", _output_rowset->data_disk_size())

            
                        .tag("output_rowset_index_size", _output_rowset->index_disk_size())

            
                        .tag("output_rowset_total_size", _output_rowset->total_disk_size())

            
                        .tag("local_read_time_us", _stats.cloud_local_read_time)

            
                        .tag("remote_read_time_us", _stats.cloud_remote_read_time)

            
                        .tag("local_read_bytes", _local_read_bytes_total)

            
                        .tag("remote_read_bytes", _remote_read_bytes_total);


                _state = CompactionState::SUCCESS;


                DorisMetrics::instance()->full_compaction_deltas_total->increment(_input_rowsets.size());

            
                DorisMetrics::instance()->full_compaction_bytes_total->increment(_input_rowsets_total_size);

            
                full_output_size << _output_rowset->total_disk_size();


                cloud_tablet()->set_last_full_compaction_success_time(UnixMillis());

            
                cloud_tablet()->set_last_full_compaction_status(Status::OK().to_string());

            
                return Status::OK();

            
            }


            Status CloudFullCompaction::modify_rowsets() {

            
                // commit compaction job

            
                cloud::TabletJobInfoPB job;

            
                auto idx = job.mutable_idx();

            
                idx->set_tablet_id(_tablet->tablet_id());

            
                idx->set_table_id(_tablet->table_id());

            
                idx->set_index_id(_tablet->index_id());

            
                idx->set_partition_id(_tablet->partition_id());

            
                auto compaction_job = job.add_compaction();

            
                compaction_job->set_id(_uuid);

            
                compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +

            
                                              std::to_string(config::heartbeat_service_port));

            
                compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);

            
                compaction_job->set_input_cumulative_point(cloud_tablet()->cumulative_layer_point());

            
                compaction_job->set_output_cumulative_point(_output_rowset->end_version() + 1);

            
                compaction_job->set_num_input_rows(_input_row_num);

            
                compaction_job->set_num_output_rows(_output_rowset->num_rows());

            
                compaction_job->set_size_input_rowsets(_input_rowsets_total_size);

            
                compaction_job->set_size_output_rowsets(_output_rowset->total_disk_size());

            
                DBUG_EXECUTE_IF("CloudFullCompaction::modify_rowsets.wrong_compaction_data_size", {

            
                    compaction_job->set_size_input_rowsets(1);

            
                    compaction_job->set_size_output_rowsets(10000001);

            
                })

            
                compaction_job->set_num_input_segments(_input_segments);

            
                compaction_job->set_num_output_segments(_output_rowset->num_segments());

            
                compaction_job->set_num_input_rowsets(num_input_rowsets());

            
                compaction_job->set_num_output_rowsets(1);

            
                compaction_job->add_input_versions(_input_rowsets.front()->start_version());

            
                compaction_job->add_input_versions(_input_rowsets.back()->end_version());

            
                compaction_job->add_output_versions(_output_rowset->end_version());

            
                compaction_job->add_txn_id(_output_rowset->txn_id());

            
                compaction_job->add_output_rowset_ids(_output_rowset->rowset_id().to_string());

            
                compaction_job->set_index_size_input_rowsets(_input_rowsets_index_size);

            
                compaction_job->set_segment_size_input_rowsets(_input_rowsets_data_size);

            
                compaction_job->set_index_size_output_rowsets(_output_rowset->index_disk_size());

            
                compaction_job->set_segment_size_output_rowsets(_output_rowset->data_disk_size());


                DBUG_EXECUTE_IF("CloudFullCompaction::modify_rowsets.block", DBUG_BLOCK);


                if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&

            
                    _tablet->enable_unique_key_merge_on_write()) {

            
                    RETURN_IF_ERROR(_cloud_full_compaction_update_delete_bitmap(this->initiator()));

            
                    compaction_job->set_delete_bitmap_lock_initiator(this->initiator());

            
                }


                cloud::FinishTabletJobResponse resp;

            
                auto st = _engine.meta_mgr().commit_tablet_job(job, &resp);

            
                if (!st.ok()) {

            
                    if (resp.status().code() == cloud::TABLET_NOT_FOUND) {

            
                        cloud_tablet()->clear_cache();

            
                    }

            
                    return st;

            
                }

            
                auto& stats = resp.stats();

            
                LOG(INFO) << "tablet stats=" << stats.ShortDebugString();


                {

            
                    std::unique_lock wrlock(_tablet->get_header_lock());

            
                    // clang-format off

            
                    cloud_tablet()->set_last_base_compaction_success_time(std::max(cloud_tablet()->last_base_compaction_success_time(), stats.last_base_compaction_time_ms()));

            
                    cloud_tablet()->set_last_cumu_compaction_success_time(std::max(cloud_tablet()->last_cumu_compaction_success_time(), stats.last_cumu_compaction_time_ms()));

            
                    cloud_tablet()->set_last_full_compaction_success_time(std::max(cloud_tablet()->last_full_compaction_success_time(), stats.last_full_compaction_time_ms()));

            
                    // clang-format on

            
                    if (cloud_tablet()->base_compaction_cnt() >= stats.base_compaction_cnt()) {

            
                        // This could happen while calling `sync_tablet_rowsets` during `commit_tablet_job`

            
                        return Status::OK();

            
                    }

            
                    // Try to make output rowset visible immediately in tablet cache, instead of waiting for next synchronization from meta-service.

            
                    cloud_tablet()->delete_rowsets(_input_rowsets, wrlock);

            
                    cloud_tablet()->add_rowsets({_output_rowset}, false, wrlock);

            
                    cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());

            
                    cloud_tablet()->set_full_compaction_cnt(stats.full_compaction_cnt());

            
                    cloud_tablet()->set_cumulative_layer_point(stats.cumulative_point());

            
                    if (stats.cumulative_compaction_cnt() >= cloud_tablet()->cumulative_compaction_cnt()) {

            
                        cloud_tablet()->reset_approximate_stats(stats.num_rowsets(), stats.num_segments(),

            
                                                                stats.num_rows(), stats.data_size());

            
                    }

            
                }

            
                _tablet->prefill_dbm_agg_cache_after_compaction(_output_rowset);

            
                return Status::OK();

            
            }


            Status CloudFullCompaction::garbage_collection() {

            
                //file_cache_garbage_collection();

            
                cloud::TabletJobInfoPB job;

            
                auto idx = job.mutable_idx();

            
                idx->set_tablet_id(_tablet->tablet_id());

            
                idx->set_table_id(_tablet->table_id());

            
                idx->set_index_id(_tablet->index_id());

            
                idx->set_partition_id(_tablet->partition_id());

            
                auto compaction_job = job.add_compaction();

            
                compaction_job->set_id(_uuid);

            
                compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +

            
                                              std::to_string(config::heartbeat_service_port));

            
                compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);

            
                if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&

            
                    _tablet->enable_unique_key_merge_on_write()) {

            
                    compaction_job->set_delete_bitmap_lock_initiator(this->initiator());

            
                }

            
                auto st = _engine.meta_mgr().abort_tablet_job(job);

            
                if (!st.ok()) {

            
                    LOG_WARNING("failed to abort compaction job")

            
                            .tag("job_id", _uuid)

            
                            .tag("tablet_id", _tablet->tablet_id())

            
                            .error(st);

            
                }

            
                return st;

            
            }


            void CloudFullCompaction::do_lease() {

            
                cloud::TabletJobInfoPB job;

            
                auto idx = job.mutable_idx();

            
                idx->set_tablet_id(_tablet->tablet_id());

            
                idx->set_table_id(_tablet->table_id());

            
                idx->set_index_id(_tablet->index_id());

            
                idx->set_partition_id(_tablet->partition_id());

            
                auto compaction_job = job.add_compaction();

            
                compaction_job->set_id(_uuid);

            
                using namespace std::chrono;

            
                int64_t lease_time = duration_cast<seconds>(system_clock::now().time_since_epoch()).count() +

            
                                     config::lease_compaction_interval_seconds * 4;

            
                compaction_job->set_lease(lease_time);

            
                auto st = _engine.meta_mgr().lease_tablet_job(job);

            
                if (!st.ok()) {

            
                    LOG_WARNING("failed to lease compaction job")

            
                            .tag("job_id", _uuid)

            
                            .tag("tablet_id", _tablet->tablet_id())

            
                            .error(st);

            
                }

            
            }


            Status CloudFullCompaction::_cloud_full_compaction_update_delete_bitmap(int64_t initiator) {

            
                DCHECK(_output_rowset->start_version() <= 2) << _output_rowset->start_version();

            
                std::vector<RowsetSharedPtr> tmp_rowsets {};

            
                DeleteBitmapPtr delete_bitmap =

            
                        std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id());

            
                RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(cloud_tablet()));

            
                int64_t max_version = cloud_tablet()->max_version().second;

            
                DCHECK(max_version >= _output_rowset->version().second);

            
                if (max_version > _output_rowset->version().second) {

            
                    auto ret = DORIS_TRY(cloud_tablet()->capture_consistent_rowsets_unlocked(

            
                            {_output_rowset->version().second + 1, max_version}, CaptureRowsetOps {}));

            
                    tmp_rowsets = std::move(ret.rowsets);

            
                }

            
                for (const auto& it : tmp_rowsets) {

            
                    int64_t cur_version = it->rowset_meta()->start_version();

            
                    RETURN_IF_ERROR(_cloud_full_compaction_calc_delete_bitmap(it, cur_version, delete_bitmap));

            
                }


                RETURN_IF_ERROR(

            
                        _engine.meta_mgr().get_delete_bitmap_update_lock(*cloud_tablet(), -1, initiator));

            
                RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(cloud_tablet()));

            
                std::lock_guard header_lock(_tablet->get_header_lock());

            
                for (const auto& it : cloud_tablet()->rowset_map()) {

            
                    int64_t cur_version = it.first.first;

            
                    const RowsetSharedPtr& published_rowset = it.second;

            
                    if (cur_version > max_version) {

            
                        RETURN_IF_ERROR(_cloud_full_compaction_calc_delete_bitmap(published_rowset, cur_version,

            
                                                                                  delete_bitmap));

            
                    }

            
                }

            
                std::optional<StorageResource> storage_resource;

            
                auto storage_resource_result = _output_rowset->rowset_meta()->remote_storage_resource();

            
                if (storage_resource_result) {

            
                    storage_resource = *storage_resource_result.value();

            
                }

            
                RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(

            
                        *cloud_tablet(), -1, initiator, delete_bitmap.get(), delete_bitmap.get(),

            
                        _output_rowset->rowset_id().to_string(), storage_resource,

            
                        config::delete_bitmap_store_write_version, cloud_tablet()->table_id()));

            
                LOG_INFO("update delete bitmap in CloudFullCompaction, tablet_id={}, range=[{}-{}]",

            
                         _tablet->tablet_id(), _input_rowsets.front()->start_version(),

            
                         _input_rowsets.back()->end_version())

            
                        .tag("job_id", _uuid)

            
                        .tag("initiator", initiator)

            
                        .tag("input_rowsets", _input_rowsets.size())

            
                        .tag("input_rows", _input_row_num)

            
                        .tag("input_segments", _input_segments)

            
                        .tag("input_rowsets_total_size", _input_rowsets_total_size)

            
                        .tag("update_bitmap_size", delete_bitmap->delete_bitmap.size());

            
                _tablet->tablet_meta()->delete_bitmap().merge(*delete_bitmap);

            
                return Status::OK();

            
            }


            Status CloudFullCompaction::_cloud_full_compaction_calc_delete_bitmap(

            
                    const RowsetSharedPtr& published_rowset, int64_t cur_version,

            
                    DeleteBitmapPtr delete_bitmap) {

            
                std::vector<segment_v2::SegmentSharedPtr> segments;

            
                RETURN_IF_ERROR(

            
                        std::static_pointer_cast<BetaRowset>(published_rowset)->load_segments(&segments));

            
                std::vector<RowsetSharedPtr> specified_rowsets {_output_rowset};

            
                DeleteBitmapPtr tmp_delete_bitmap = std::make_shared<DeleteBitmap>(_tablet->tablet_id());


                OlapStopWatch watch;

            
                auto token = _engine.calc_delete_bitmap_executor()->create_token();

            
                RETURN_IF_ERROR(BaseTablet::calc_delete_bitmap(

            
                        _tablet, published_rowset, segments, specified_rowsets, tmp_delete_bitmap, cur_version,

            
                        token.get(), _output_rs_writer.get()));

            
                RETURN_IF_ERROR(token->wait());

            
                size_t total_rows = std::accumulate(

            
                        segments.begin(), segments.end(), 0,

            
                        [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });

            
                for (const auto& [k, v] : tmp_delete_bitmap->delete_bitmap) {

            
                    if (std::get<0>(k) == _output_rowset->rowset_id() &&

            
                        std::get<1>(k) != DeleteBitmap::INVALID_SEGMENT_ID) {

            
                        delete_bitmap->merge({std::get<0>(k), std::get<1>(k), cur_version}, v);

            
                    }

            
                }

            
                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: [" << _output_rowset->version().first << "-"

            
                           << _output_rowset->version().second << "]"

            
                           << ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows;

            
                return Status::OK();

            
            }


            } // namespace doris
	// 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_full_compaction.h"

	#include <gen_cpp/cloud.pb.h>

	#include <boost/container_hash/hash.hpp>

	#include "cloud/cloud_meta_mgr.h"
	#include "cloud/cloud_storage_engine.h"
	#include "cloud/config.h"
	#include "common/config.h"
	#include "common/status.h"
	#include "core/column/column.h"
	#include "cpp/sync_point.h"
	#include "service/backend_options.h"
	#include "storage/compaction/compaction.h"
	#include "storage/rowset/beta_rowset.h"
	#include "storage/tablet/tablet_meta.h"
	#include "util/debug_points.h"
	#include "util/thread.h"
	#include "util/uuid_generator.h"

	namespace doris {
	using namespace ErrorCode;

	bvar::Adder<uint64_t> full_output_size("full_compaction", "output_size");

	CloudFullCompaction::CloudFullCompaction(CloudStorageEngine& engine, CloudTabletSPtr tablet)
	: CloudCompactionMixin(engine, tablet,
	"BaseCompaction:" + std::to_string(tablet->tablet_id())) {}

	CloudFullCompaction::~CloudFullCompaction() = default;

	Status CloudFullCompaction::prepare_compact() {
	Status st;
	Defer defer_set_st([&] {
	if (!st.ok()) {
	cloud_tablet()->set_last_full_compaction_status(st.to_string());
	cloud_tablet()->set_last_full_compaction_failure_time(UnixMillis());
	}
	});
	if (_tablet->tablet_state() != TABLET_RUNNING) {
	st = Status::InternalError("invalid tablet state. tablet_id={}", _tablet->tablet_id());
	return st;
	}

	// always sync latest rowset for full compaction
	st = cloud_tablet()->sync_rowsets();
	RETURN_IF_ERROR(st);

	st = pick_rowsets_to_compact();
	RETURN_IF_ERROR(st);

	for (auto& rs : _input_rowsets) {
	_input_row_num += rs->num_rows();
	_input_segments += rs->num_segments();
	_input_rowsets_data_size += rs->data_disk_size();
	_input_rowsets_index_size += rs->index_disk_size();
	_input_rowsets_total_size += rs->total_disk_size();
	}
	LOG_INFO("start CloudFullCompaction, tablet_id={}, range=[{}-{}]", _tablet->tablet_id(),
	_input_rowsets.front()->start_version(), _input_rowsets.back()->end_version())
	.tag("job_id", _uuid)
	.tag("input_rowsets", _input_rowsets.size())
	.tag("input_rows", _input_row_num)
	.tag("input_segments", _input_segments)
	.tag("input_rowsets_data_size", _input_rowsets_data_size)
	.tag("input_rowsets_index_size", _input_rowsets_index_size)
	.tag("input_rowsets_total_size", _input_rowsets_total_size);
	return Status::OK();
	}
	Status CloudFullCompaction::request_global_lock() {
	// prepare compaction job
	cloud::TabletJobInfoPB job;
	auto idx = job.mutable_idx();
	idx->set_tablet_id(_tablet->tablet_id());
	idx->set_table_id(_tablet->table_id());
	idx->set_index_id(_tablet->index_id());
	idx->set_partition_id(_tablet->partition_id());
	auto compaction_job = job.add_compaction();
	compaction_job->set_id(_uuid);
	compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +
	std::to_string(config::heartbeat_service_port));
	compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);
	compaction_job->set_base_compaction_cnt(_base_compaction_cnt);
	compaction_job->set_cumulative_compaction_cnt(_cumulative_compaction_cnt);
	using namespace std::chrono;
	int64_t now = duration_cast<seconds>(system_clock::now().time_since_epoch()).count();
	_expiration = now + config::compaction_timeout_seconds;
	compaction_job->set_expiration(_expiration);
	compaction_job->set_lease(now + config::lease_compaction_interval_seconds * 4);
	// Set input version range to let meta-service judge version range conflict
	compaction_job->add_input_versions(_input_rowsets.front()->start_version());
	compaction_job->add_input_versions(_input_rowsets.back()->end_version());
	cloud::StartTabletJobResponse resp;
	auto st = _engine.meta_mgr().prepare_tablet_job(job, &resp);
	if (!st.ok()) {
	if (resp.status().code() == cloud::STALE_TABLET_CACHE) {
	// set last_sync_time to 0 to force sync tablet next time
	cloud_tablet()->last_sync_rowsets_time_s = 0;
	} else if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
	// tablet not found
	cloud_tablet()->clear_cache();
	}
	}
	return st;
	}

	Status CloudFullCompaction::pick_rowsets_to_compact() {
	_input_rowsets.clear();
	{
	std::shared_lock rlock(_tablet->get_header_lock());
	_base_compaction_cnt = cloud_tablet()->base_compaction_cnt();
	_cumulative_compaction_cnt = cloud_tablet()->cumulative_compaction_cnt();
	_input_rowsets = cloud_tablet()->pick_candidate_rowsets_to_full_compaction_unlocked();
	}
	if (auto st = check_version_continuity(_input_rowsets); !st.ok()) {
	DCHECK(false) << st;
	return st;
	}
	if (_input_rowsets.size() <= 1) {
	return Status::Error<BE_NO_SUITABLE_VERSION>(
	"insufficent compaction input rowset, #rowsets={}", _input_rowsets.size());
	}

	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<BE_NO_SUITABLE_VERSION>("no suitable versions for compaction");
	}

	return Status::OK();
	}

	Status CloudFullCompaction::execute_compact() {
	DBUG_EXECUTE_IF("CloudFullCompaction::execute_compact.block", {
	auto target_tablet_id = dp->param<int64_t>("tablet_id", -1);
	LOG_INFO(
	"[verbose] CloudFullCompaction::execute_compact.block, target_tablet_id={}, "
	"tablet_id={}",
	target_tablet_id, cloud_tablet()->tablet_id());
	if (target_tablet_id == cloud_tablet()->tablet_id()) {
	DBUG_BLOCK;
	}
	});
	TEST_SYNC_POINT_RETURN_WITH_VALUE("CloudFullCompaction::execute_compact_impl", Status::OK(),
	this);
	#ifndef __APPLE__
	if (config::enable_base_compaction_idle_sched) {
	Thread::set_idle_sched();
	}
	#endif

	SCOPED_ATTACH_TASK(_mem_tracker);

	using namespace std::chrono;
	auto start = steady_clock::now();
	auto res = CloudCompactionMixin::execute_compact();
	cloud_tablet()->set_last_full_compaction_status(res.to_string());
	if (!res.ok()) {
	cloud_tablet()->set_last_full_compaction_failure_time(UnixMillis());
	LOG(WARNING) << "fail to do " << compaction_name() << ". res=" << res
	<< ", tablet=" << _tablet->tablet_id()
	<< ", output_version=" << _output_version;
	return res;
	}
	LOG_INFO("finish CloudFullCompaction, tablet_id={}, cost={}ms", _tablet->tablet_id(),
	duration_cast<milliseconds>(steady_clock::now() - start).count())
	.tag("job_id", _uuid)
	.tag("input_rowsets", _input_rowsets.size())
	.tag("input_rows", _input_row_num)
	.tag("input_segments", _input_segments)
	.tag("input_rowsets_data_size", _input_rowsets_data_size)
	.tag("input_rowsets_index_size", _input_rowsets_index_size)
	.tag("input_rowsets_total_size", _input_rowsets_total_size)
	.tag("output_rows", _output_rowset->num_rows())
	.tag("output_segments", _output_rowset->num_segments())
	.tag("output_rowset_data_size", _output_rowset->data_disk_size())
	.tag("output_rowset_index_size", _output_rowset->index_disk_size())
	.tag("output_rowset_total_size", _output_rowset->total_disk_size())
	.tag("local_read_time_us", _stats.cloud_local_read_time)
	.tag("remote_read_time_us", _stats.cloud_remote_read_time)
	.tag("local_read_bytes", _local_read_bytes_total)
	.tag("remote_read_bytes", _remote_read_bytes_total);

	_state = CompactionState::SUCCESS;

	DorisMetrics::instance()->full_compaction_deltas_total->increment(_input_rowsets.size());
	DorisMetrics::instance()->full_compaction_bytes_total->increment(_input_rowsets_total_size);
	full_output_size << _output_rowset->total_disk_size();

	cloud_tablet()->set_last_full_compaction_success_time(UnixMillis());
	cloud_tablet()->set_last_full_compaction_status(Status::OK().to_string());
	return Status::OK();
	}

	Status CloudFullCompaction::modify_rowsets() {
	// commit compaction job
	cloud::TabletJobInfoPB job;
	auto idx = job.mutable_idx();
	idx->set_tablet_id(_tablet->tablet_id());
	idx->set_table_id(_tablet->table_id());
	idx->set_index_id(_tablet->index_id());
	idx->set_partition_id(_tablet->partition_id());
	auto compaction_job = job.add_compaction();
	compaction_job->set_id(_uuid);
	compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +
	std::to_string(config::heartbeat_service_port));
	compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);
	compaction_job->set_input_cumulative_point(cloud_tablet()->cumulative_layer_point());
	compaction_job->set_output_cumulative_point(_output_rowset->end_version() + 1);
	compaction_job->set_num_input_rows(_input_row_num);
	compaction_job->set_num_output_rows(_output_rowset->num_rows());
	compaction_job->set_size_input_rowsets(_input_rowsets_total_size);
	compaction_job->set_size_output_rowsets(_output_rowset->total_disk_size());
	DBUG_EXECUTE_IF("CloudFullCompaction::modify_rowsets.wrong_compaction_data_size", {
	compaction_job->set_size_input_rowsets(1);
	compaction_job->set_size_output_rowsets(10000001);
	})
	compaction_job->set_num_input_segments(_input_segments);
	compaction_job->set_num_output_segments(_output_rowset->num_segments());
	compaction_job->set_num_input_rowsets(num_input_rowsets());
	compaction_job->set_num_output_rowsets(1);
	compaction_job->add_input_versions(_input_rowsets.front()->start_version());
	compaction_job->add_input_versions(_input_rowsets.back()->end_version());
	compaction_job->add_output_versions(_output_rowset->end_version());
	compaction_job->add_txn_id(_output_rowset->txn_id());
	compaction_job->add_output_rowset_ids(_output_rowset->rowset_id().to_string());
	compaction_job->set_index_size_input_rowsets(_input_rowsets_index_size);
	compaction_job->set_segment_size_input_rowsets(_input_rowsets_data_size);
	compaction_job->set_index_size_output_rowsets(_output_rowset->index_disk_size());
	compaction_job->set_segment_size_output_rowsets(_output_rowset->data_disk_size());

	DBUG_EXECUTE_IF("CloudFullCompaction::modify_rowsets.block", DBUG_BLOCK);

	if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
	_tablet->enable_unique_key_merge_on_write()) {
	RETURN_IF_ERROR(_cloud_full_compaction_update_delete_bitmap(this->initiator()));
	compaction_job->set_delete_bitmap_lock_initiator(this->initiator());
	}

	cloud::FinishTabletJobResponse resp;
	auto st = _engine.meta_mgr().commit_tablet_job(job, &resp);
	if (!st.ok()) {
	if (resp.status().code() == cloud::TABLET_NOT_FOUND) {
	cloud_tablet()->clear_cache();
	}
	return st;
	}
	auto& stats = resp.stats();
	LOG(INFO) << "tablet stats=" << stats.ShortDebugString();

	{
	std::unique_lock wrlock(_tablet->get_header_lock());
	// clang-format off
	cloud_tablet()->set_last_base_compaction_success_time(std::max(cloud_tablet()->last_base_compaction_success_time(), stats.last_base_compaction_time_ms()));
	cloud_tablet()->set_last_cumu_compaction_success_time(std::max(cloud_tablet()->last_cumu_compaction_success_time(), stats.last_cumu_compaction_time_ms()));
	cloud_tablet()->set_last_full_compaction_success_time(std::max(cloud_tablet()->last_full_compaction_success_time(), stats.last_full_compaction_time_ms()));
	// clang-format on
	if (cloud_tablet()->base_compaction_cnt() >= stats.base_compaction_cnt()) {
	// This could happen while calling `sync_tablet_rowsets` during `commit_tablet_job`
	return Status::OK();
	}
	// Try to make output rowset visible immediately in tablet cache, instead of waiting for next synchronization from meta-service.
	cloud_tablet()->delete_rowsets(_input_rowsets, wrlock);
	cloud_tablet()->add_rowsets({_output_rowset}, false, wrlock);
	cloud_tablet()->set_base_compaction_cnt(stats.base_compaction_cnt());
	cloud_tablet()->set_full_compaction_cnt(stats.full_compaction_cnt());
	cloud_tablet()->set_cumulative_layer_point(stats.cumulative_point());
	if (stats.cumulative_compaction_cnt() >= cloud_tablet()->cumulative_compaction_cnt()) {
	cloud_tablet()->reset_approximate_stats(stats.num_rowsets(), stats.num_segments(),
	stats.num_rows(), stats.data_size());
	}
	}
	_tablet->prefill_dbm_agg_cache_after_compaction(_output_rowset);
	return Status::OK();
	}

	Status CloudFullCompaction::garbage_collection() {
	//file_cache_garbage_collection();
	cloud::TabletJobInfoPB job;
	auto idx = job.mutable_idx();
	idx->set_tablet_id(_tablet->tablet_id());
	idx->set_table_id(_tablet->table_id());
	idx->set_index_id(_tablet->index_id());
	idx->set_partition_id(_tablet->partition_id());
	auto compaction_job = job.add_compaction();
	compaction_job->set_id(_uuid);
	compaction_job->set_initiator(BackendOptions::get_localhost() + ':' +
	std::to_string(config::heartbeat_service_port));
	compaction_job->set_type(cloud::TabletCompactionJobPB::FULL);
	if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
	_tablet->enable_unique_key_merge_on_write()) {
	compaction_job->set_delete_bitmap_lock_initiator(this->initiator());
	}
	auto st = _engine.meta_mgr().abort_tablet_job(job);
	if (!st.ok()) {
	LOG_WARNING("failed to abort compaction job")
	.tag("job_id", _uuid)
	.tag("tablet_id", _tablet->tablet_id())
	.error(st);
	}
	return st;
	}

	void CloudFullCompaction::do_lease() {
	cloud::TabletJobInfoPB job;
	auto idx = job.mutable_idx();
	idx->set_tablet_id(_tablet->tablet_id());
	idx->set_table_id(_tablet->table_id());
	idx->set_index_id(_tablet->index_id());
	idx->set_partition_id(_tablet->partition_id());
	auto compaction_job = job.add_compaction();
	compaction_job->set_id(_uuid);
	using namespace std::chrono;
	int64_t lease_time = duration_cast<seconds>(system_clock::now().time_since_epoch()).count() +
	config::lease_compaction_interval_seconds * 4;
	compaction_job->set_lease(lease_time);
	auto st = _engine.meta_mgr().lease_tablet_job(job);
	if (!st.ok()) {
	LOG_WARNING("failed to lease compaction job")
	.tag("job_id", _uuid)
	.tag("tablet_id", _tablet->tablet_id())
	.error(st);
	}
	}

	Status CloudFullCompaction::_cloud_full_compaction_update_delete_bitmap(int64_t initiator) {
	DCHECK(_output_rowset->start_version() <= 2) << _output_rowset->start_version();
	std::vector<RowsetSharedPtr> tmp_rowsets {};
	DeleteBitmapPtr delete_bitmap =
	std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id());
	RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(cloud_tablet()));
	int64_t max_version = cloud_tablet()->max_version().second;
	DCHECK(max_version >= _output_rowset->version().second);
	if (max_version > _output_rowset->version().second) {
	auto ret = DORIS_TRY(cloud_tablet()->capture_consistent_rowsets_unlocked(
	{_output_rowset->version().second + 1, max_version}, CaptureRowsetOps {}));
	tmp_rowsets = std::move(ret.rowsets);
	}
	for (const auto& it : tmp_rowsets) {
	int64_t cur_version = it->rowset_meta()->start_version();
	RETURN_IF_ERROR(_cloud_full_compaction_calc_delete_bitmap(it, cur_version, delete_bitmap));
	}

	RETURN_IF_ERROR(
	_engine.meta_mgr().get_delete_bitmap_update_lock(*cloud_tablet(), -1, initiator));
	RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(cloud_tablet()));
	std::lock_guard header_lock(_tablet->get_header_lock());
	for (const auto& it : cloud_tablet()->rowset_map()) {
	int64_t cur_version = it.first.first;
	const RowsetSharedPtr& published_rowset = it.second;
	if (cur_version > max_version) {
	RETURN_IF_ERROR(_cloud_full_compaction_calc_delete_bitmap(published_rowset, cur_version,
	delete_bitmap));
	}
	}
	std::optional<StorageResource> storage_resource;
	auto storage_resource_result = _output_rowset->rowset_meta()->remote_storage_resource();
	if (storage_resource_result) {
	storage_resource = *storage_resource_result.value();
	}
	RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(
	*cloud_tablet(), -1, initiator, delete_bitmap.get(), delete_bitmap.get(),
	_output_rowset->rowset_id().to_string(), storage_resource,
	config::delete_bitmap_store_write_version, cloud_tablet()->table_id()));
	LOG_INFO("update delete bitmap in CloudFullCompaction, tablet_id={}, range=[{}-{}]",
	_tablet->tablet_id(), _input_rowsets.front()->start_version(),
	_input_rowsets.back()->end_version())
	.tag("job_id", _uuid)
	.tag("initiator", initiator)
	.tag("input_rowsets", _input_rowsets.size())
	.tag("input_rows", _input_row_num)
	.tag("input_segments", _input_segments)
	.tag("input_rowsets_total_size", _input_rowsets_total_size)
	.tag("update_bitmap_size", delete_bitmap->delete_bitmap.size());
	_tablet->tablet_meta()->delete_bitmap().merge(*delete_bitmap);
	return Status::OK();
	}

	Status CloudFullCompaction::_cloud_full_compaction_calc_delete_bitmap(
	const RowsetSharedPtr& published_rowset, int64_t cur_version,
	DeleteBitmapPtr delete_bitmap) {
	std::vector<segment_v2::SegmentSharedPtr> segments;
	RETURN_IF_ERROR(
	std::static_pointer_cast<BetaRowset>(published_rowset)->load_segments(&segments));
	std::vector<RowsetSharedPtr> specified_rowsets {_output_rowset};
	DeleteBitmapPtr tmp_delete_bitmap = std::make_shared<DeleteBitmap>(_tablet->tablet_id());

	OlapStopWatch watch;
	auto token = _engine.calc_delete_bitmap_executor()->create_token();
	RETURN_IF_ERROR(BaseTablet::calc_delete_bitmap(
	_tablet, published_rowset, segments, specified_rowsets, tmp_delete_bitmap, cur_version,
	token.get(), _output_rs_writer.get()));
	RETURN_IF_ERROR(token->wait());
	size_t total_rows = std::accumulate(
	segments.begin(), segments.end(), 0,
	[](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });
	for (const auto& [k, v] : tmp_delete_bitmap->delete_bitmap) {
	if (std::get<0>(k) == _output_rowset->rowset_id() &&
	std::get<1>(k) != DeleteBitmap::INVALID_SEGMENT_ID) {
	delete_bitmap->merge({std::get<0>(k), std::get<1>(k), cur_version}, v);
	}
	}
	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: [" << _output_rowset->version().first << "-"
	<< _output_rowset->version().second << "]"
	<< ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows;
	return Status::OK();
	}

	} // namespace doris