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

#include <gen_cpp/olap_file.pb.h>
#include <rapidjson/document.h>
#include <rapidjson/encodings.h>
#include <rapidjson/prettywriter.h>
#include <rapidjson/rapidjson.h>
#include <rapidjson/stringbuffer.h>

#include <atomic>
#include <memory>
#include <shared_mutex>
#include <unordered_map>
#include <vector>

#include "cloud/cloud_meta_mgr.h"
#include "cloud/cloud_storage_engine.h"
#include "cloud/cloud_tablet_mgr.h"
#include "common/config.h"
#include "common/logging.h"
#include "io/cache/block_file_cache_downloader.h"
#include "io/cache/block_file_cache_factory.h"
#include "olap/compaction.h"
#include "olap/cumulative_compaction_time_series_policy.h"
#include "olap/olap_define.h"
#include "olap/rowset/beta_rowset.h"
#include "olap/rowset/rowset.h"
#include "olap/rowset/rowset_factory.h"
#include "olap/rowset/rowset_fwd.h"
#include "olap/rowset/rowset_writer.h"
#include "olap/rowset/segment_v2/inverted_index_desc.h"
#include "olap/storage_policy.h"
#include "olap/tablet_schema.h"
#include "olap/txn_manager.h"
#include "util/debug_points.h"
#include "vec/common/schema_util.h"

namespace doris {
#include "common/compile_check_begin.h"
using namespace ErrorCode;

static constexpr int LOAD_INITIATOR_ID = -1;

CloudTablet::CloudTablet(CloudStorageEngine& engine, TabletMetaSharedPtr tablet_meta)
        : BaseTablet(std::move(tablet_meta)), _engine(engine) {}

CloudTablet::~CloudTablet() = default;

bool CloudTablet::exceed_version_limit(int32_t limit) {
    return _approximate_num_rowsets.load(std::memory_order_relaxed) > limit;
}

Status CloudTablet::capture_consistent_rowsets_unlocked(
        const Version& spec_version, std::vector<RowsetSharedPtr>* rowsets) const {
    Versions version_path;
    auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path);
    if (!st.ok()) {
        // Check no missed versions or req version is merged
        auto missed_versions = get_missed_versions(spec_version.second);
        if (missed_versions.empty()) {
            st.set_code(VERSION_ALREADY_MERGED); // Reset error code
        }
        st.append(" tablet_id=" + std::to_string(tablet_id()));
        return st;
    }
    VLOG_DEBUG << "capture consitent versions: " << version_path;
    return _capture_consistent_rowsets_unlocked(version_path, rowsets);
}

Status CloudTablet::capture_rs_readers(const Version& spec_version,
                                       std::vector<RowSetSplits>* rs_splits,
                                       bool skip_missing_version) {
    DBUG_EXECUTE_IF("CloudTablet.capture_rs_readers.return.e-230", {
        LOG_WARNING("CloudTablet.capture_rs_readers.return e-230").tag("tablet_id", tablet_id());
        return Status::Error<false>(-230, "injected error");
    });
    Versions version_path;
    std::shared_lock rlock(_meta_lock);
    auto st = _timestamped_version_tracker.capture_consistent_versions(spec_version, &version_path);
    if (!st.ok()) {
        rlock.unlock(); // avoid logging in lock range
        // Check no missed versions or req version is merged
        auto missed_versions = get_missed_versions(spec_version.second);
        if (missed_versions.empty()) {
            st.set_code(VERSION_ALREADY_MERGED); // Reset error code
            st.append(" versions are already compacted, ");
        }
        st.append(" tablet_id=" + std::to_string(tablet_id()));
        // clang-format off
        LOG(WARNING) << st << '\n' << [this]() { std::string json; get_compaction_status(&json); return json; }();
        // clang-format on
        return st;
    }
    VLOG_DEBUG << "capture consitent versions: " << version_path;
    return capture_rs_readers_unlocked(version_path, rs_splits);
}

Status CloudTablet::merge_rowsets_schema() {
    // Find the rowset with the max version
    auto max_version_rowset =
            std::max_element(
                    _rs_version_map.begin(), _rs_version_map.end(),
                    [](const auto& a, const auto& b) {
                        return !a.second->tablet_schema()
                                       ? true
                                       : (!b.second->tablet_schema()
                                                  ? false
                                                  : a.second->tablet_schema()->schema_version() <
                                                            b.second->tablet_schema()
                                                                    ->schema_version());
                    })
                    ->second;
    TabletSchemaSPtr max_version_schema = max_version_rowset->tablet_schema();
    // If the schema has variant columns, perform a merge to create a wide tablet schema
    if (max_version_schema->num_variant_columns() > 0) {
        std::vector<TabletSchemaSPtr> schemas;
        std::transform(_rs_version_map.begin(), _rs_version_map.end(), std::back_inserter(schemas),
                       [](const auto& rs_meta) { return rs_meta.second->tablet_schema(); });
        // Merge the collected schemas to obtain the least common schema
        RETURN_IF_ERROR(vectorized::schema_util::get_least_common_schema(schemas, nullptr,
                                                                         max_version_schema));
        VLOG_DEBUG << "dump schema: " << max_version_schema->dump_full_schema();
        _merged_tablet_schema = max_version_schema;
    }
    return Status::OK();
}

// There are only two tablet_states RUNNING and NOT_READY in cloud mode
// This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS.
Status CloudTablet::sync_rowsets(const SyncOptions& options, SyncRowsetStats* stats) {
    RETURN_IF_ERROR(sync_if_not_running(stats));

    if (options.query_version > 0) {
        std::shared_lock rlock(_meta_lock);
        if (_max_version >= options.query_version) {
            return Status::OK();
        }
    }

    // serially execute sync to reduce unnecessary network overhead
    std::unique_lock lock(_sync_meta_lock);
    if (options.query_version > 0) {
        std::shared_lock rlock(_meta_lock);
        if (_max_version >= options.query_version) {
            return Status::OK();
        }
    }

    auto st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, options, stats);
    if (st.is<ErrorCode::NOT_FOUND>()) {
        clear_cache();
    }

    return st;
}

// Sync tablet meta and all rowset meta if not running.
// This could happen when BE didn't finish schema change job and another BE committed this schema change job.
// It should be a quite rare situation.
Status CloudTablet::sync_if_not_running(SyncRowsetStats* stats) {
    if (tablet_state() == TABLET_RUNNING) {
        return Status::OK();
    }

    // Serially execute sync to reduce unnecessary network overhead
    std::unique_lock lock(_sync_meta_lock);

    {
        std::shared_lock rlock(_meta_lock);
        if (tablet_state() == TABLET_RUNNING) {
            return Status::OK();
        }
    }

    TabletMetaSharedPtr tablet_meta;
    auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
    if (!st.ok()) {
        if (st.is<ErrorCode::NOT_FOUND>()) {
            clear_cache();
        }
        return st;
    }

    if (tablet_meta->tablet_state() != TABLET_RUNNING) [[unlikely]] {
        // MoW may go to here when load while schema change
        return Status::OK();
    }

    TimestampedVersionTracker empty_tracker;
    {
        std::lock_guard wlock(_meta_lock);
        RETURN_IF_ERROR(set_tablet_state(TABLET_RUNNING));
        _rs_version_map.clear();
        _stale_rs_version_map.clear();
        std::swap(_timestamped_version_tracker, empty_tracker);
        _tablet_meta->clear_rowsets();
        _tablet_meta->clear_stale_rowset();
        _max_version = -1;
    }

    st = _engine.meta_mgr().sync_tablet_rowsets_unlocked(this, lock, {}, stats);
    if (st.is<ErrorCode::NOT_FOUND>()) {
        clear_cache();
    }
    return st;
}

TabletSchemaSPtr CloudTablet::merged_tablet_schema() const {
    std::shared_lock rlock(_meta_lock);
    return _merged_tablet_schema;
}

void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
                              std::unique_lock<std::shared_mutex>& meta_lock,
                              bool warmup_delta_data) {
    if (to_add.empty()) {
        return;
    }

    auto add_rowsets_directly = [=, this](std::vector<RowsetSharedPtr>& rowsets) {
        for (auto& rs : rowsets) {
            if (version_overlap || warmup_delta_data) {
#ifndef BE_TEST
                // Warmup rowset data in background
                for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
                    const auto& rowset_meta = rs->rowset_meta();
                    constexpr int64_t interval = 600; // 10 mins
                    // When BE restart and receive the `load_sync` rpc, it will sync all historical rowsets first time.
                    // So we need to filter out the old rowsets avoid to download the whole table.
                    if (warmup_delta_data &&
                        ::time(nullptr) - rowset_meta->newest_write_timestamp() >= interval) {
                        continue;
                    }

                    auto storage_resource = rowset_meta->remote_storage_resource();
                    if (!storage_resource) {
                        LOG(WARNING) << storage_resource.error();
                        continue;
                    }

                    int64_t expiration_time =
                            _tablet_meta->ttl_seconds() == 0 ||
                                            rowset_meta->newest_write_timestamp() <= 0
                                    ? 0
                                    : rowset_meta->newest_write_timestamp() +
                                              _tablet_meta->ttl_seconds();
                    // clang-format off
                    _engine.file_cache_block_downloader().submit_download_task(io::DownloadFileMeta {
                            .path = storage_resource.value()->remote_segment_path(*rowset_meta, seg_id),
                            .file_size = rs->rowset_meta()->segment_file_size(seg_id),
                            .file_system = storage_resource.value()->fs,
                            .ctx =
                                    {
                                            .expiration_time = expiration_time,
                                            .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
                                    },
                            .download_done {},
                    });

                    auto download_idx_file = [&](const io::Path& idx_path) {
                        io::DownloadFileMeta meta {
                                .path = idx_path,
                                .file_size = -1,
                                .file_system = storage_resource.value()->fs,
                                .ctx =
                                        {
                                                .expiration_time = expiration_time,
                                                .is_dryrun = config::enable_reader_dryrun_when_download_file_cache,
                                        },
                                .download_done {},
                        };
                        _engine.file_cache_block_downloader().submit_download_task(std::move(meta));
                    };
                    // clang-format on
                    auto schema_ptr = rowset_meta->tablet_schema();
                    auto idx_version = schema_ptr->get_inverted_index_storage_format();
                    if (idx_version == InvertedIndexStorageFormatPB::V1) {
                        for (const auto& index : schema_ptr->inverted_indexes()) {
                            auto idx_path = storage_resource.value()->remote_idx_v1_path(
                                    *rowset_meta, seg_id, index->index_id(),
                                    index->get_index_suffix());
                            download_idx_file(idx_path);
                        }
                    } else {
                        if (schema_ptr->has_inverted_index()) {
                            auto idx_path = storage_resource.value()->remote_idx_v2_path(
                                    *rowset_meta, seg_id);
                            download_idx_file(idx_path);
                        }
                    }
                }
#endif
            }
            _rs_version_map.emplace(rs->version(), rs);
            _timestamped_version_tracker.add_version(rs->version());
            _max_version = std::max(rs->end_version(), _max_version);
            update_base_size(*rs);
        }
        _tablet_meta->add_rowsets_unchecked(rowsets);
    };

    if (!version_overlap) {
        add_rowsets_directly(to_add);
        return;
    }

    // Filter out existed rowsets
    auto remove_it =
            std::remove_if(to_add.begin(), to_add.end(), [this](const RowsetSharedPtr& rs) {
                if (auto find_it = _rs_version_map.find(rs->version());
                    find_it == _rs_version_map.end()) {
                    return false;
                } else if (find_it->second->rowset_id() == rs->rowset_id()) {
                    return true; // Same rowset
                }

                // If version of rowset in `to_add` is equal to rowset in tablet but rowset_id is not equal,
                // replace existed rowset with `to_add` rowset. This may occur when:
                //  1. schema change converts rowsets which have been double written to new tablet
                //  2. cumu compaction picks single overlapping input rowset to perform compaction
                _tablet_meta->delete_rs_meta_by_version(rs->version(), nullptr);
                _rs_version_map[rs->version()] = rs;
                _tablet_meta->add_rowsets_unchecked({rs});
                update_base_size(*rs);
                return true;
            });

    to_add.erase(remove_it, to_add.end());

    // delete rowsets with overlapped version
    std::vector<RowsetSharedPtr> to_add_directly;
    for (auto& to_add_rs : to_add) {
        // delete rowsets with overlapped version
        std::vector<RowsetSharedPtr> to_delete;
        Version to_add_v = to_add_rs->version();
        // if start_version  > max_version, we can skip checking overlap here.
        if (to_add_v.first > _max_version) {
            // if start_version  > max_version, we can skip checking overlap here.
            to_add_directly.push_back(to_add_rs);
        } else {
            to_add_directly.push_back(to_add_rs);
            for (auto& [v, rs] : _rs_version_map) {
                if (to_add_v.contains(v)) {
                    to_delete.push_back(rs);
                }
            }
            delete_rowsets(to_delete, meta_lock);
        }
    }

    add_rowsets_directly(to_add_directly);
}

void CloudTablet::delete_rowsets(const std::vector<RowsetSharedPtr>& to_delete,
                                 std::unique_lock<std::shared_mutex>&) {
    if (to_delete.empty()) {
        return;
    }
    std::vector<RowsetMetaSharedPtr> rs_metas;
    rs_metas.reserve(to_delete.size());
    for (auto&& rs : to_delete) {
        rs_metas.push_back(rs->rowset_meta());
        _stale_rs_version_map[rs->version()] = rs;
    }
    _timestamped_version_tracker.add_stale_path_version(rs_metas);
    for (auto&& rs : to_delete) {
        _rs_version_map.erase(rs->version());
    }

    _tablet_meta->modify_rs_metas({}, rs_metas, false);
}

uint64_t CloudTablet::delete_expired_stale_rowsets() {
    if (config::enable_mow_verbose_log) {
        LOG_INFO("begin delete_expired_stale_rowset for tablet={}", tablet_id());
    }
    std::vector<RowsetSharedPtr> expired_rowsets;
    // ATTN: trick, Use stale_rowsets to temporarily increase the reference count of the rowset shared pointer in _stale_rs_version_map so that in the recycle_cached_data function, it checks if the reference count is 2.
    std::vector<RowsetSharedPtr> stale_rowsets;
    int64_t expired_stale_sweep_endtime =
            ::time(nullptr) - config::tablet_rowset_stale_sweep_time_sec;
    std::vector<std::string> version_to_delete;
    {
        std::unique_lock wlock(_meta_lock);

        std::vector<int64_t> path_ids;
        // capture the path version to delete
        _timestamped_version_tracker.capture_expired_paths(expired_stale_sweep_endtime, &path_ids);

        if (path_ids.empty()) {
            return 0;
        }

        for (int64_t path_id : path_ids) {
            int64_t start_version = -1;
            int64_t end_version = -1;
            // delete stale versions in version graph
            auto version_path = _timestamped_version_tracker.fetch_and_delete_path_by_id(path_id);
            for (auto& v_ts : version_path->timestamped_versions()) {
                auto rs_it = _stale_rs_version_map.find(v_ts->version());
                if (rs_it != _stale_rs_version_map.end()) {
                    expired_rowsets.push_back(rs_it->second);
                    stale_rowsets.push_back(rs_it->second);
                    LOG(INFO) << "erase stale rowset, tablet_id=" << tablet_id()
                              << " rowset_id=" << rs_it->second->rowset_id().to_string()
                              << " version=" << rs_it->first.to_string();
                    _stale_rs_version_map.erase(rs_it);
                } else {
                    LOG(WARNING) << "cannot find stale rowset " << v_ts->version() << " in tablet "
                                 << tablet_id();
                    // clang-format off
                    DCHECK(false) << [this, &wlock]() { wlock.unlock(); std::string json; get_compaction_status(&json); return json; }();
                    // clang-format on
                }
                if (start_version < 0) {
                    start_version = v_ts->version().first;
                }
                end_version = v_ts->version().second;
                _tablet_meta->delete_stale_rs_meta_by_version(v_ts->version());
            }
            Version version(start_version, end_version);
            version_to_delete.emplace_back(version.to_string());
        }
        _reconstruct_version_tracker_if_necessary();
    }
    _tablet_meta->delete_bitmap().remove_stale_delete_bitmap_from_queue(version_to_delete);
    recycle_cached_data(expired_rowsets);
    if (config::enable_mow_verbose_log) {
        LOG_INFO("finish delete_expired_stale_rowset for tablet={}", tablet_id());
    }
    return expired_rowsets.size();
}

void CloudTablet::update_base_size(const Rowset& rs) {
    // Define base rowset as the rowset of version [2-x]
    if (rs.start_version() == 2) {
        _base_size = rs.total_disk_size();
    }
}

void CloudTablet::clear_cache() {
    CloudTablet::recycle_cached_data(get_snapshot_rowset(true));
    _engine.tablet_mgr().erase_tablet(tablet_id());
}

void CloudTablet::recycle_cached_data(const std::vector<RowsetSharedPtr>& rowsets) {
    for (const auto& rs : rowsets) {
        // rowsets and tablet._rs_version_map each hold a rowset shared_ptr, so at this point, the reference count of the shared_ptr is at least 2.
        if (rs.use_count() > 2) {
            LOG(WARNING) << "Rowset " << rs->rowset_id().to_string() << " has " << rs.use_count()
                         << " references. File Cache won't be recycled when query is using it.";
            return;
        }
        rs->clear_cache();
    }
}

void CloudTablet::reset_approximate_stats(int64_t num_rowsets, int64_t num_segments,
                                          int64_t num_rows, int64_t data_size) {
    _approximate_num_rowsets.store(num_rowsets, std::memory_order_relaxed);
    _approximate_num_segments.store(num_segments, std::memory_order_relaxed);
    _approximate_num_rows.store(num_rows, std::memory_order_relaxed);
    _approximate_data_size.store(data_size, std::memory_order_relaxed);
    int64_t cumu_num_deltas = 0;
    int64_t cumu_num_rowsets = 0;
    auto cp = _cumulative_point.load(std::memory_order_relaxed);
    for (auto& [v, r] : _rs_version_map) {
        if (v.second < cp) {
            continue;
        }

        cumu_num_deltas += r->is_segments_overlapping() ? r->num_segments() : 1;
        ++cumu_num_rowsets;
    }
    _approximate_cumu_num_rowsets.store(cumu_num_rowsets, std::memory_order_relaxed);
    _approximate_cumu_num_deltas.store(cumu_num_deltas, std::memory_order_relaxed);
}

Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_rowset_writer(
        RowsetWriterContext& context, bool vertical) {
    context.rowset_id = _engine.next_rowset_id();
    // FIXME(plat1ko): Seems `tablet_id` and `index_id` has been set repeatedly
    context.tablet_id = tablet_id();
    context.index_id = index_id();
    context.partition_id = partition_id();
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
    return RowsetFactory::create_rowset_writer(_engine, context, vertical);
}

// create a rowset writer with rowset_id and seg_id
// after writer, merge this transient rowset with original rowset
Result<std::unique_ptr<RowsetWriter>> CloudTablet::create_transient_rowset_writer(
        const Rowset& rowset, std::shared_ptr<PartialUpdateInfo> partial_update_info,
        int64_t txn_expiration) {
    if (rowset.rowset_meta_state() != RowsetStatePB::BEGIN_PARTIAL_UPDATE &&
        rowset.rowset_meta_state() != RowsetStatePB::COMMITTED) [[unlikely]] {
        auto msg = fmt::format(
                "wrong rowset state when create_transient_rowset_writer, rowset state should be "
                "BEGIN_PARTIAL_UPDATE or COMMITTED, but found {}, rowset_id={}, tablet_id={}",
                RowsetStatePB_Name(rowset.rowset_meta_state()), rowset.rowset_id().to_string(),
                tablet_id());
        // see `CloudRowsetWriter::build` for detail.
        // if this is in a retry task, the rowset state may have been changed to RowsetStatePB::COMMITTED
        // in `RowsetMeta::merge_rowset_meta()` in previous trials.
        LOG(WARNING) << msg;
        DCHECK(false) << msg;
    }
    RowsetWriterContext context;
    context.rowset_state = PREPARED;
    context.segments_overlap = OVERLAPPING;
    // During a partial update, the extracted columns of a variant should not be included in the tablet schema.
    // This is because the partial update for a variant needs to ignore the extracted columns.
    // Otherwise, the schema types in different rowsets might be inconsistent. When performing a partial update,
    // the complete variant is constructed by reading all the sub-columns of the variant.
    context.tablet_schema = rowset.tablet_schema()->copy_without_variant_extracted_columns();
    context.newest_write_timestamp = UnixSeconds();
    context.tablet_id = table_id();
    context.enable_segcompaction = false;
    context.write_type = DataWriteType::TYPE_DIRECT;
    context.partial_update_info = std::move(partial_update_info);
    context.is_transient_rowset_writer = true;
    context.rowset_id = rowset.rowset_id();
    context.tablet_id = tablet_id();
    context.index_id = index_id();
    context.partition_id = partition_id();
    context.enable_unique_key_merge_on_write = enable_unique_key_merge_on_write();
    context.txn_expiration = txn_expiration;

    auto storage_resource = rowset.rowset_meta()->remote_storage_resource();
    if (!storage_resource) {
        return ResultError(std::move(storage_resource.error()));
    }

    context.storage_resource = *storage_resource.value();

    return RowsetFactory::create_rowset_writer(_engine, context, false)
            .transform([&](auto&& writer) {
                writer->set_segment_start_id(cast_set<int32_t>(rowset.num_segments()));
                return writer;
            });
}

int64_t CloudTablet::get_cloud_base_compaction_score() const {
    if (_tablet_meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {
        bool has_delete = false;
        int64_t point = cumulative_layer_point();
        std::shared_lock<std::shared_mutex> rlock(_meta_lock);
        for (const auto& rs_meta : _tablet_meta->all_rs_metas()) {
            if (rs_meta->start_version() >= point) {
                continue;
            }
            if (rs_meta->has_delete_predicate()) {
                has_delete = true;
                break;
            }
        }
        if (!has_delete) {
            return 0;
        }
    }

    return _approximate_num_rowsets.load(std::memory_order_relaxed) -
           _approximate_cumu_num_rowsets.load(std::memory_order_relaxed);
}

int64_t CloudTablet::get_cloud_cumu_compaction_score() const {
    // TODO(plat1ko): Propose an algorithm that considers tablet's key type, number of delete rowsets,
    //  number of tablet versions simultaneously.
    return _approximate_cumu_num_deltas.load(std::memory_order_relaxed);
}

// return a json string to show the compaction status of this tablet
void CloudTablet::get_compaction_status(std::string* json_result) {
    rapidjson::Document root;
    root.SetObject();

    rapidjson::Document path_arr;
    path_arr.SetArray();

    std::vector<RowsetSharedPtr> rowsets;
    std::vector<RowsetSharedPtr> stale_rowsets;
    {
        std::shared_lock rdlock(_meta_lock);
        rowsets.reserve(_rs_version_map.size());
        for (auto& it : _rs_version_map) {
            rowsets.push_back(it.second);
        }
        stale_rowsets.reserve(_stale_rs_version_map.size());
        for (auto& it : _stale_rs_version_map) {
            stale_rowsets.push_back(it.second);
        }
    }
    std::sort(rowsets.begin(), rowsets.end(), Rowset::comparator);
    std::sort(stale_rowsets.begin(), stale_rowsets.end(), Rowset::comparator);

    // get snapshot version path json_doc
    _timestamped_version_tracker.get_stale_version_path_json_doc(path_arr);
    root.AddMember("cumulative point", _cumulative_point.load(), root.GetAllocator());

    // print all rowsets' version as an array
    rapidjson::Document versions_arr;
    rapidjson::Document missing_versions_arr;
    versions_arr.SetArray();
    missing_versions_arr.SetArray();
    int64_t last_version = -1;
    for (auto& rowset : rowsets) {
        const Version& ver = rowset->version();
        if (ver.first != last_version + 1) {
            rapidjson::Value miss_value;
            miss_value.SetString(fmt::format("[{}-{}]", last_version + 1, ver.first - 1).c_str(),
                                 missing_versions_arr.GetAllocator());
            missing_versions_arr.PushBack(miss_value, missing_versions_arr.GetAllocator());
        }
        rapidjson::Value value;
        std::string version_str = rowset->get_rowset_info_str();
        value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()),
                        versions_arr.GetAllocator());
        versions_arr.PushBack(value, versions_arr.GetAllocator());
        last_version = ver.second;
    }
    root.AddMember("rowsets", versions_arr, root.GetAllocator());
    root.AddMember("missing_rowsets", missing_versions_arr, root.GetAllocator());

    // print all stale rowsets' version as an array
    rapidjson::Document stale_versions_arr;
    stale_versions_arr.SetArray();
    for (auto& rowset : stale_rowsets) {
        rapidjson::Value value;
        std::string version_str = rowset->get_rowset_info_str();
        value.SetString(version_str.c_str(), cast_set<uint32_t>(version_str.length()),
                        stale_versions_arr.GetAllocator());
        stale_versions_arr.PushBack(value, stale_versions_arr.GetAllocator());
    }
    root.AddMember("stale_rowsets", stale_versions_arr, root.GetAllocator());

    // add stale version rowsets
    root.AddMember("stale version path", path_arr, root.GetAllocator());

    // to json string
    rapidjson::StringBuffer strbuf;
    rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(strbuf);
    root.Accept(writer);
    *json_result = std::string(strbuf.GetString());
}

void CloudTablet::set_cumulative_layer_point(int64_t new_point) {
    if (new_point == Tablet::K_INVALID_CUMULATIVE_POINT || new_point >= _cumulative_point) {
        _cumulative_point = new_point;
        return;
    }
    // cumulative point should only be reset to -1, or be increased
    // FIXME: could happen in currently unresolved race conditions
    LOG(WARNING) << "Unexpected cumulative point: " << new_point
                 << ", origin: " << _cumulative_point.load();
}

std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_base_compaction() {
    std::vector<RowsetSharedPtr> candidate_rowsets;
    {
        std::shared_lock rlock(_meta_lock);
        for (const auto& [version, rs] : _rs_version_map) {
            if (version.first != 0 && version.first < _cumulative_point &&
                (_alter_version == -1 || version.second <= _alter_version)) {
                candidate_rowsets.push_back(rs);
            }
        }
    }
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
    return candidate_rowsets;
}

std::vector<RowsetSharedPtr> CloudTablet::pick_candidate_rowsets_to_full_compaction() {
    std::vector<RowsetSharedPtr> candidate_rowsets;
    {
        std::shared_lock rlock(_meta_lock);
        for (auto& [v, rs] : _rs_version_map) {
            // MUST NOT compact rowset [0-1] for some historical reasons (see cloud_schema_change)
            if (v.first != 0) {
                candidate_rowsets.push_back(rs);
            }
        }
    }
    std::sort(candidate_rowsets.begin(), candidate_rowsets.end(), Rowset::comparator);
    return candidate_rowsets;
}

CalcDeleteBitmapExecutor* CloudTablet::calc_delete_bitmap_executor() {
    return _engine.calc_delete_bitmap_executor();
}

Status CloudTablet::save_delete_bitmap(const TabletTxnInfo* txn_info, int64_t txn_id,
                                       DeleteBitmapPtr delete_bitmap, RowsetWriter* rowset_writer,
                                       const RowsetIdUnorderedSet& cur_rowset_ids, int64_t lock_id,
                                       int64_t next_visible_version) {
    RowsetSharedPtr rowset = txn_info->rowset;
    int64_t cur_version = rowset->start_version();
    // update delete bitmap info, in order to avoid recalculation when trying again
    RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info(
            txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::PREPARE));

    if (txn_info->partial_update_info && txn_info->partial_update_info->is_partial_update() &&
        rowset_writer->num_rows() > 0) {
        DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.update_tmp_rowset.error", {
            return Status::InternalError<false>("injected update_tmp_rowset error.");
        });
        const auto& rowset_meta = rowset->rowset_meta();
        RETURN_IF_ERROR(_engine.meta_mgr().update_tmp_rowset(*rowset_meta));
    }

    RETURN_IF_ERROR(save_delete_bitmap_to_ms(cur_version, txn_id, delete_bitmap, lock_id,
                                             next_visible_version));

    // store the delete bitmap with sentinel marks in txn_delete_bitmap_cache because if the txn is retried for some reason,
    // it will use the delete bitmap from txn_delete_bitmap_cache when re-calculating the delete bitmap, during which it will do
    // delete bitmap correctness check. If we store the new_delete_bitmap, the delete bitmap correctness check will fail
    RETURN_IF_ERROR(_engine.txn_delete_bitmap_cache().update_tablet_txn_info(
            txn_id, tablet_id(), delete_bitmap, cur_rowset_ids, PublishStatus::SUCCEED,
            txn_info->publish_info));

    DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.enable_sleep", {
        auto sleep_sec = dp->param<int>("sleep", 5);
        std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
    });

    DBUG_EXECUTE_IF("CloudTablet::save_delete_bitmap.injected_error", {
        auto retry = dp->param<bool>("retry", false);
        auto sleep_sec = dp->param<int>("sleep", 0);
        std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
        if (retry) { // return DELETE_BITMAP_LOCK_ERROR to let it retry
            return Status::Error<ErrorCode::DELETE_BITMAP_LOCK_ERROR>(
                    "injected DELETE_BITMAP_LOCK_ERROR");
        } else {
            return Status::InternalError<false>("injected non-retryable error");
        }
    });

    return Status::OK();
}

Status CloudTablet::save_delete_bitmap_to_ms(int64_t cur_version, int64_t txn_id,
                                             DeleteBitmapPtr delete_bitmap, int64_t lock_id,
                                             int64_t next_visible_version) {
    DeleteBitmapPtr new_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id());
    for (auto iter = delete_bitmap->delete_bitmap.begin();
         iter != delete_bitmap->delete_bitmap.end(); ++iter) {
        // skip sentinel mark, which is used for delete bitmap correctness check
        if (std::get<1>(iter->first) != DeleteBitmap::INVALID_SEGMENT_ID) {
            new_delete_bitmap->merge(
                    {std::get<0>(iter->first), std::get<1>(iter->first), cur_version},
                    iter->second);
        }
    }
    // lock_id != -1 means this is in an explict txn
    bool is_explicit_txn = (lock_id != -1);
    auto ms_lock_id = !is_explicit_txn ? txn_id : lock_id;

    RETURN_IF_ERROR(_engine.meta_mgr().update_delete_bitmap(*this, ms_lock_id, LOAD_INITIATOR_ID,
                                                            new_delete_bitmap.get(), txn_id,
                                                            is_explicit_txn, next_visible_version));
    return Status::OK();
}

Versions CloudTablet::calc_missed_versions(int64_t spec_version, Versions existing_versions) const {
    DCHECK(spec_version > 0) << "invalid spec_version: " << spec_version;

    // sort the existing versions in ascending order
    std::sort(existing_versions.begin(), existing_versions.end(),
              [](const Version& a, const Version& b) {
                  // simple because 2 versions are certainly not overlapping
                  return a.first < b.first;
              });

    // From the first version(=0), find the missing version until spec_version
    int64_t last_version = -1;
    Versions missed_versions;
    for (const Version& version : existing_versions) {
        if (version.first > last_version + 1) {
            // there is a hole between versions
            missed_versions.emplace_back(last_version + 1, std::min(version.first, spec_version));
        }
        last_version = version.second;
        if (last_version >= spec_version) {
            break;
        }
    }
    if (last_version < spec_version) {
        // there is a hole between the last version and the specificed version.
        missed_versions.emplace_back(last_version + 1, spec_version);
    }
    return missed_versions;
}

Status CloudTablet::calc_delete_bitmap_for_compaction(
        const std::vector<RowsetSharedPtr>& input_rowsets, const RowsetSharedPtr& output_rowset,
        const RowIdConversion& rowid_conversion, ReaderType compaction_type, int64_t merged_rows,
        int64_t filtered_rows, int64_t initiator, DeleteBitmapPtr& output_rowset_delete_bitmap,
        bool allow_delete_in_cumu_compaction) {
    output_rowset_delete_bitmap = std::make_shared<DeleteBitmap>(tablet_id());
    std::unique_ptr<RowLocationSet> missed_rows;
    if ((config::enable_missing_rows_correctness_check ||
         config::enable_mow_compaction_correctness_check_core) &&
        !allow_delete_in_cumu_compaction &&
        compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION) {
        missed_rows = std::make_unique<RowLocationSet>();
        LOG(INFO) << "RowLocation Set inited succ for tablet:" << tablet_id();
    }

    std::unique_ptr<std::map<RowsetSharedPtr, RowLocationPairList>> location_map;
    if (config::enable_rowid_conversion_correctness_check &&
        tablet_schema()->cluster_key_uids().empty()) {
        location_map = std::make_unique<std::map<RowsetSharedPtr, RowLocationPairList>>();
        LOG(INFO) << "Location Map inited succ for tablet:" << tablet_id();
    }

    // 1. calc delete bitmap for historical data
    RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this));
    Version version = max_version();
    std::size_t missed_rows_size = 0;
    calc_compaction_output_rowset_delete_bitmap(
            input_rowsets, rowid_conversion, 0, version.second + 1, missed_rows.get(),
            location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get());
    if (missed_rows) {
        missed_rows_size = missed_rows->size();
        if (!allow_delete_in_cumu_compaction) {
            if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION &&
                tablet_state() == TABLET_RUNNING) {
                if (merged_rows + filtered_rows >= 0 &&
                    merged_rows + filtered_rows != missed_rows_size) {
                    std::string err_msg = fmt::format(
                            "cumulative compaction: the merged rows({}), the filtered rows({}) is "
                            "not equal to missed rows({}) in rowid conversion, tablet_id: {}, "
                            "table_id:{}",
                            merged_rows, filtered_rows, missed_rows_size, tablet_id(), table_id());
                    if (config::enable_mow_compaction_correctness_check_core) {
                        CHECK(false) << err_msg;
                    } else {
                        DCHECK(false) << err_msg;
                    }
                    LOG(WARNING) << err_msg;
                }
            }
        }
    }
    if (location_map) {
        RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map));
        location_map->clear();
    }

    // 2. calc delete bitmap for incremental data
    int64_t t1 = MonotonicMicros();
    RETURN_IF_ERROR(_engine.meta_mgr().get_delete_bitmap_update_lock(
            *this, COMPACTION_DELETE_BITMAP_LOCK_ID, initiator));
    int64_t t2 = MonotonicMicros();
    RETURN_IF_ERROR(_engine.meta_mgr().sync_tablet_rowsets(this));
    int64_t t3 = MonotonicMicros();

    calc_compaction_output_rowset_delete_bitmap(
            input_rowsets, rowid_conversion, version.second, UINT64_MAX, missed_rows.get(),
            location_map.get(), tablet_meta()->delete_bitmap(), output_rowset_delete_bitmap.get());
    int64_t t4 = MonotonicMicros();
    if (location_map) {
        RETURN_IF_ERROR(check_rowid_conversion(output_rowset, *location_map));
    }
    int64_t t5 = MonotonicMicros();
    if (missed_rows) {
        DCHECK_EQ(missed_rows->size(), missed_rows_size);
        if (missed_rows->size() != missed_rows_size) {
            LOG(WARNING) << "missed rows don't match, before: " << missed_rows_size
                         << " after: " << missed_rows->size();
        }
    }

    // 3. store delete bitmap
    auto st = _engine.meta_mgr().update_delete_bitmap(*this, -1, initiator,
                                                      output_rowset_delete_bitmap.get());
    int64_t t6 = MonotonicMicros();
    LOG(INFO) << "calc_delete_bitmap_for_compaction, tablet_id=" << tablet_id()
              << ", get lock cost " << (t2 - t1) << " us, sync rowsets cost " << (t3 - t2)
              << " us, calc delete bitmap cost " << (t4 - t3) << " us, check rowid conversion cost "
              << (t5 - t4) << " us, store delete bitmap cost " << (t6 - t5)
              << " us, st=" << st.to_string();
    return st;
}

Status CloudTablet::sync_meta() {
    if (!config::enable_file_cache) {
        return Status::OK();
    }

    TabletMetaSharedPtr tablet_meta;
    auto st = _engine.meta_mgr().get_tablet_meta(tablet_id(), &tablet_meta);
    if (!st.ok()) {
        if (st.is<ErrorCode::NOT_FOUND>()) {
            clear_cache();
        }
        return st;
    }

    auto new_ttl_seconds = tablet_meta->ttl_seconds();
    if (_tablet_meta->ttl_seconds() != new_ttl_seconds) {
        _tablet_meta->set_ttl_seconds(new_ttl_seconds);
        int64_t cur_time = UnixSeconds();
        std::shared_lock rlock(_meta_lock);
        for (auto& [_, rs] : _rs_version_map) {
            for (int seg_id = 0; seg_id < rs->num_segments(); ++seg_id) {
                int64_t new_expiration_time =
                        new_ttl_seconds + rs->rowset_meta()->newest_write_timestamp();
                new_expiration_time = new_expiration_time > cur_time ? new_expiration_time : 0;
                auto file_key = Segment::file_cache_key(rs->rowset_id().to_string(), seg_id);
                auto* file_cache = io::FileCacheFactory::instance()->get_by_path(file_key);
                file_cache->modify_expiration_time(file_key, new_expiration_time);
            }
        }
    }

    auto new_compaction_policy = tablet_meta->compaction_policy();
    if (_tablet_meta->compaction_policy() != new_compaction_policy) {
        _tablet_meta->set_compaction_policy(new_compaction_policy);
    }
    auto new_time_series_compaction_goal_size_mbytes =
            tablet_meta->time_series_compaction_goal_size_mbytes();
    if (_tablet_meta->time_series_compaction_goal_size_mbytes() !=
        new_time_series_compaction_goal_size_mbytes) {
        _tablet_meta->set_time_series_compaction_goal_size_mbytes(
                new_time_series_compaction_goal_size_mbytes);
    }
    auto new_time_series_compaction_file_count_threshold =
            tablet_meta->time_series_compaction_file_count_threshold();
    if (_tablet_meta->time_series_compaction_file_count_threshold() !=
        new_time_series_compaction_file_count_threshold) {
        _tablet_meta->set_time_series_compaction_file_count_threshold(
                new_time_series_compaction_file_count_threshold);
    }
    auto new_time_series_compaction_time_threshold_seconds =
            tablet_meta->time_series_compaction_time_threshold_seconds();
    if (_tablet_meta->time_series_compaction_time_threshold_seconds() !=
        new_time_series_compaction_time_threshold_seconds) {
        _tablet_meta->set_time_series_compaction_time_threshold_seconds(
                new_time_series_compaction_time_threshold_seconds);
    }
    auto new_time_series_compaction_empty_rowsets_threshold =
            tablet_meta->time_series_compaction_empty_rowsets_threshold();
    if (_tablet_meta->time_series_compaction_empty_rowsets_threshold() !=
        new_time_series_compaction_empty_rowsets_threshold) {
        _tablet_meta->set_time_series_compaction_empty_rowsets_threshold(
                new_time_series_compaction_empty_rowsets_threshold);
    }
    auto new_time_series_compaction_level_threshold =
            tablet_meta->time_series_compaction_level_threshold();
    if (_tablet_meta->time_series_compaction_level_threshold() !=
        new_time_series_compaction_level_threshold) {
        _tablet_meta->set_time_series_compaction_level_threshold(
                new_time_series_compaction_level_threshold);
    }

    return Status::OK();
}

void CloudTablet::build_tablet_report_info(TTabletInfo* tablet_info) {
    std::shared_lock rdlock(_meta_lock);
    tablet_info->__set_total_version_count(_tablet_meta->version_count());
    tablet_info->__set_tablet_id(_tablet_meta->tablet_id());
    // Currently, this information will not be used by the cloud report,
    // but it may be used in the future.
}

Status CloudTablet::check_delete_bitmap_cache(int64_t txn_id,
                                              DeleteBitmap* expected_delete_bitmap) {
    DeleteBitmapPtr cached_delete_bitmap;
    CloudStorageEngine& engine = ExecEnv::GetInstance()->storage_engine().to_cloud();
    Status st = engine.txn_delete_bitmap_cache().get_delete_bitmap(
            txn_id, tablet_id(), &cached_delete_bitmap, nullptr, nullptr);
    if (st.ok()) {
        bool res = (expected_delete_bitmap->cardinality() == cached_delete_bitmap->cardinality());
        auto msg = fmt::format(
                "delete bitmap cache check failed, cur_cardinality={}, cached_cardinality={}"
                "txn_id={}, tablet_id={}",
                expected_delete_bitmap->cardinality(), cached_delete_bitmap->cardinality(), txn_id,
                tablet_id());
        if (!res) {
            DCHECK(res) << msg;
            return Status::InternalError<false>(msg);
        }
    }
    return Status::OK();
}

#include "common/compile_check_end.h"
} // namespace doris