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

#include <fmt/format.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <atomic>

#include <cstdint>

#include <cstdlib>

#include <list>

#include <map>

#include <memory>

#include <mutex>

#include <nlohmann/json.hpp>

#include <numeric>

#include <ostream>

#include <ranges>

#include <set>

#include <shared_mutex>

#include <utility>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/cloud_tablet.h"

#include "cloud/config.h"

#include "cloud/pb_convert.h"

#include "common/config.h"

#include "common/metrics/doris_metrics.h"

#include "common/status.h"

#include "cpp/sync_point.h"

#include "exec/common/variant_util.h"

#include "io/cache/block_file_cache_factory.h"

#include "io/fs/file_system.h"

#include "io/fs/file_writer.h"

#include "io/fs/remote_file_system.h"

#include "io/io_common.h"

#include "runtime/memory/mem_tracker_limiter.h"

#include "runtime/thread_context.h"

#include "storage/compaction/collection_statistics.h"

#include "storage/compaction/cumulative_compaction.h"

#include "storage/compaction/cumulative_compaction_binlog_policy.h"

#include "storage/compaction/cumulative_compaction_policy.h"

#include "storage/compaction/cumulative_compaction_time_series_policy.h"

#include "storage/compaction_task_tracker.h"

#include "storage/data_dir.h"

#include "storage/index/index_file_reader.h"

#include "storage/index/index_file_writer.h"

#include "storage/index/inverted/inverted_index_compaction.h"

#include "storage/index/inverted/inverted_index_desc.h"

#include "storage/index/inverted/inverted_index_fs_directory.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowset/beta_rowset.h"

#include "storage/rowset/beta_rowset_reader.h"

#include "storage/rowset/beta_rowset_writer.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_fwd.h"

#include "storage/rowset/rowset_meta.h"

#include "storage/rowset/rowset_writer.h"

#include "storage/rowset/rowset_writer_context.h"

#include "storage/storage_engine.h"

#include "storage/storage_policy.h"

#include "storage/tablet/tablet.h"

#include "storage/tablet/tablet_meta.h"

#include "storage/tablet/tablet_meta_manager.h"

#include "storage/task/engine_checksum_task.h"

#include "storage/txn/txn_manager.h"

#include "storage/utils.h"

#include "util/pretty_printer.h"

#include "util/time.h"

#include "util/trace.h"

using std::vector;

namespace doris {

using namespace ErrorCode;

// Determine whether to enable index-only file cache mode for compaction output.

// This function decides if only index files should be written to cache, based on:

// - write_file_cache: whether file cache is enabled

// - compaction_type: type of compaction (base or cumulative)

// - enable_base_index_only: config flag for base compaction

// - enable_cumu_index_only: config flag for cumulative compaction

// Returns true if index-only mode should be enabled, false otherwise.

bool should_enable_compaction_cache_index_only(bool write_file_cache, ReaderType compaction_type,

                                               bool enable_base_index_only,

                                               bool enable_cumu_index_only) {

    if (!write_file_cache) {

        return false;

    }

    if (compaction_type == ReaderType::READER_BASE_COMPACTION && enable_base_index_only) {

        return true;

    }

    if (compaction_type == ReaderType::READER_CUMULATIVE_COMPACTION && enable_cumu_index_only) {

        return true;

    }

    return false;

}

namespace {

bool is_rowset_tidy(std::string& pre_max_key, bool& pre_rs_key_bounds_truncated,

                    const RowsetSharedPtr& rhs) {

    size_t min_tidy_size = config::ordered_data_compaction_min_segment_size;

    if (rhs->num_segments() == 0) {

        return true;

    }

    if (rhs->is_segments_overlapping()) {

        return false;

    }

    // check segment size

    auto* beta_rowset = reinterpret_cast<BetaRowset*>(rhs.get());

    std::vector<size_t> segments_size;

    RETURN_FALSE_IF_ERROR(beta_rowset->get_segments_size(&segments_size));

    for (auto segment_size : segments_size) {

        // is segment is too small, need to do compaction

        if (segment_size < min_tidy_size) {

            return false;

        }

    }

    std::string min_key;

    auto ret = rhs->first_key(&min_key);

    if (!ret) {

        return false;

    }

    bool cur_rs_key_bounds_truncated {rhs->is_segments_key_bounds_truncated()};

    if (!Slice::lhs_is_strictly_less_than_rhs(Slice {pre_max_key}, pre_rs_key_bounds_truncated,

                                              Slice {min_key}, cur_rs_key_bounds_truncated)) {

        return false;

    }

    CHECK(rhs->last_key(&pre_max_key));

    pre_rs_key_bounds_truncated = cur_rs_key_bounds_truncated;

    return true;

}

TsoRange commit_tso_range(const std::vector<RowsetSharedPtr>& rowsets) {

    DCHECK(!rowsets.empty());

    auto range = rowsets.front()->commit_tso();

    for (const auto& rowset : rowsets) {

        const auto commit_tso = rowset->commit_tso();

        range.first = std::min(range.first, commit_tso.start_tso());

        range.second = std::max(range.second, commit_tso.end_tso());

    }

    return range;

}

} // namespace

Compaction::Compaction(BaseTabletSPtr tablet, const std::string& label)

        : _compaction_id(CompactionTaskTracker::instance()->next_compaction_id()),

          _mem_tracker(

                  MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::COMPACTION, label)),

          _tablet(std::move(tablet)),

          _is_vertical(config::enable_vertical_compaction),

          _allow_delete_in_cumu_compaction(config::enable_delete_when_cumu_compaction),

          _enable_vertical_compact_variant_subcolumns(

                  config::enable_vertical_compact_variant_subcolumns),

          _enable_inverted_index_compaction(config::inverted_index_compaction_enable) {

    init_profile(label);

    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);

    _rowid_conversion = std::make_unique<RowIdConversion>();

}

Compaction::~Compaction() {

    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker);

    _output_rs_writer.reset();

    _tablet.reset();

    _input_rowsets.clear();

    _output_rowset.reset();

    _cur_tablet_schema.reset();

    _rowid_conversion.reset();

}

std::string Compaction::input_version_range_str() const {

    if (_input_rowsets.empty()) return "";

    return fmt::format("[{}-{}]", _input_rowsets.front()->start_version(),

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

}

void Compaction::submit_profile_record(bool success, int64_t start_time_ms,

                                       const std::string& status_msg) {

    if (!profile_type().has_value()) {

        return;

    }

    auto* tracker = CompactionTaskTracker::instance();

    CompletionStats stats;

    // Input stats for backfill: local compaction fills these in build_basic_info()

    // which runs inside execute_compact_impl(), so they are available now.

    stats.input_version_range = input_version_range_str();

    stats.input_rowsets_count = static_cast<int64_t>(_input_rowsets.size());

    stats.input_row_num = _input_row_num;

    stats.input_data_size = _input_rowsets_data_size;

    stats.input_index_size = _input_rowsets_index_size;

    stats.input_total_size = _input_rowsets_total_size;

    stats.input_segments_num = input_segments_num_value();

    stats.end_time_ms = UnixMillis();

    stats.merged_rows = _stats.merged_rows;

    stats.filtered_rows = _stats.filtered_rows;

    stats.output_rows = _stats.output_rows;

    if (_output_rowset) {

        stats.output_row_num = _output_rowset->num_rows();

        stats.output_data_size = _output_rowset->data_disk_size();

        stats.output_index_size = _output_rowset->index_disk_size();

        stats.output_total_size = _output_rowset->total_disk_size();

        stats.output_segments_num = _output_rowset->num_segments();

    }

    stats.output_version = _output_version.to_string();

    stats.is_ordered_data_compaction = _is_ordered_data_compaction;

    if (_merge_rowsets_latency_timer) {

        stats.merge_latency_ms = _merge_rowsets_latency_timer->value() / 1000000;

    }

    stats.bytes_read_from_local = _stats.bytes_read_from_local;

    stats.bytes_read_from_remote = _stats.bytes_read_from_remote;

    if (_mem_tracker) {

        stats.peak_memory_bytes = _mem_tracker->peak_consumption();

    }

    if (success) {

        tracker->complete(_compaction_id, stats);

    } else {

        tracker->fail(_compaction_id, stats, status_msg);

    }

}

void Compaction::init_profile(const std::string& label) {

    _profile = std::make_unique<RuntimeProfile>(label);

    _input_rowsets_data_size_counter =

            ADD_COUNTER(_profile, "input_rowsets_data_size", TUnit::BYTES);

    _input_rowsets_counter = ADD_COUNTER(_profile, "input_rowsets_count", TUnit::UNIT);

    _input_row_num_counter = ADD_COUNTER(_profile, "input_row_num", TUnit::UNIT);

    _input_segments_num_counter = ADD_COUNTER(_profile, "input_segments_num", TUnit::UNIT);

    _merged_rows_counter = ADD_COUNTER(_profile, "merged_rows", TUnit::UNIT);

    _filtered_rows_counter = ADD_COUNTER(_profile, "filtered_rows", TUnit::UNIT);

    _output_rowset_data_size_counter =

            ADD_COUNTER(_profile, "output_rowset_data_size", TUnit::BYTES);

    _output_row_num_counter = ADD_COUNTER(_profile, "output_row_num", TUnit::UNIT);

    _output_segments_num_counter = ADD_COUNTER(_profile, "output_segments_num", TUnit::UNIT);

    _merge_rowsets_latency_timer = ADD_TIMER(_profile, "merge_rowsets_latency");

}

int64_t Compaction::merge_way_num() {

    int64_t way_num = 0;

    for (auto&& rowset : _input_rowsets) {

        way_num += rowset->rowset_meta()->get_merge_way_num();

    }

    return way_num;

}

Status Compaction::merge_input_rowsets() {

    std::vector<RowsetReaderSharedPtr> input_rs_readers;

    input_rs_readers.reserve(_input_rowsets.size());

    for (auto& rowset : _input_rowsets) {

        RowsetReaderSharedPtr rs_reader;

        RETURN_IF_ERROR(rowset->create_reader(&rs_reader));

        input_rs_readers.push_back(std::move(rs_reader));

    }

    RowsetWriterContext ctx;

    // Propagate input rowset readers into the rowset writer context before the writer is created.

    // Variant nested-group compaction uses this metadata to enable the streaming writer path.

    ctx.input_rs_readers = input_rs_readers;

    RETURN_IF_ERROR(construct_output_rowset_writer(ctx));

    // write merged rows to output rowset

    // The test results show that merger is low-memory-footprint, there is no need to tracker its mem pool

    // if ctx.columns_to_do_index_compaction.size() > 0, it means we need to do inverted index compaction.

    // the row ID conversion matrix needs to be used for inverted index compaction.

    if (!ctx.columns_to_do_index_compaction.empty() ||

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

         _tablet->enable_unique_key_merge_on_write())) {

        _stats.rowid_conversion = _rowid_conversion.get();

    }

    int64_t way_num = merge_way_num();

    Status res;

    {

        SCOPED_TIMER(_merge_rowsets_latency_timer);

        // 1. Merge segment files and write bkd inverted index

        // TODO implement vertical compaction for seq map

        if (_is_vertical && !_tablet->tablet_schema()->has_seq_map()) {

            if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {

                RETURN_IF_ERROR(update_delete_bitmap());

            }

            auto progress_cb = [compaction_id = this->_compaction_id](int64_t total,

                                                                      int64_t completed) {

                CompactionTaskTracker::instance()->update_progress(compaction_id, total, completed);

            };

            res = Merger::vertical_merge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,

                                                 input_rs_readers, _output_rs_writer.get(),

                                                 cast_set<uint32_t>(get_avg_segment_rows()),

                                                 way_num, &_stats, progress_cb);

        } else {

            if (!_tablet->tablet_schema()->cluster_key_uids().empty()) {

                return Status::InternalError(

                        "mow table with cluster keys does not support non vertical compaction");

            }

            res = Merger::vmerge_rowsets(_tablet, compaction_type(), *_cur_tablet_schema,

                                         input_rs_readers, _output_rs_writer.get(), &_stats);

        }

        _tablet->last_compaction_status = res;

        if (!res.ok()) {

            return res;

        }

        // 2. Merge the remaining inverted index files of the string type

        RETURN_IF_ERROR(do_inverted_index_compaction());

    }

    COUNTER_UPDATE(_merged_rows_counter, _stats.merged_rows);

    COUNTER_UPDATE(_filtered_rows_counter, _stats.filtered_rows);

    // 3. In the `build`, `_close_file_writers` is called to close the inverted index file writer and write the final compound index file.

    RETURN_NOT_OK_STATUS_WITH_WARN(_output_rs_writer->build(_output_rowset),

                                   fmt::format("rowset writer build failed. output_version: {}",

                                               _output_version.to_string()));

    _output_rowset->rowset_meta()->set_commit_tso(commit_tso_range(_input_rowsets));

    // When true, writers should remove variant extracted subcolumns from the

    // schema stored in RowsetMeta. This is used when compaction temporarily

    // extends schema to split variant subcolumns for vertical compaction but

    // the final rowset meta must not persist those extracted subcolumns.

    if (_enable_vertical_compact_variant_subcolumns &&

        (_cur_tablet_schema->num_variant_columns() > 0)) {

        _output_rowset->rowset_meta()->set_tablet_schema(

                _cur_tablet_schema->copy_without_variant_extracted_columns());

    }

    //RETURN_IF_ERROR(_engine.meta_mgr().commit_rowset(*_output_rowset->rowset_meta().get()));

    set_delete_predicate_for_output_rowset();

    _local_read_bytes_total = _stats.bytes_read_from_local;

    _remote_read_bytes_total = _stats.bytes_read_from_remote;

    DorisMetrics::instance()->local_compaction_read_bytes_total->increment(_local_read_bytes_total);

    DorisMetrics::instance()->remote_compaction_read_bytes_total->increment(

            _remote_read_bytes_total);

    DorisMetrics::instance()->local_compaction_write_bytes_total->increment(

            _stats.cached_bytes_total);

    COUNTER_UPDATE(_output_rowset_data_size_counter, _output_rowset->data_disk_size());

    COUNTER_UPDATE(_output_row_num_counter, _output_rowset->num_rows());

    COUNTER_UPDATE(_output_segments_num_counter, _output_rowset->num_segments());

    return check_correctness();

}

void Compaction::set_delete_predicate_for_output_rowset() {

    // Now we support delete in cumu compaction, to make all data in rowsets whose version

    // is below output_version to be delete in the future base compaction, we should carry

    // all delete predicate in the output rowset.

    // Output start version > 2 means we must set the delete predicate in the output rowset

    if (_output_rowset->version().first > 2 &&

        (_allow_delete_in_cumu_compaction || is_index_change_compaction())) {

        DeletePredicatePB delete_predicate;

        std::accumulate(_input_rowsets.begin(), _input_rowsets.end(), &delete_predicate,

                        [](DeletePredicatePB* delete_predicate, const RowsetSharedPtr& rs) {

                            if (rs->rowset_meta()->has_delete_predicate()) {

                                delete_predicate->MergeFrom(rs->rowset_meta()->delete_predicate());

                            }

                            return delete_predicate;

                        });

        // now version in delete_predicate is deprecated

        if (!delete_predicate.in_predicates().empty() ||

            !delete_predicate.sub_predicates_v2().empty() ||

            !delete_predicate.sub_predicates().empty()) {

            _output_rowset->rowset_meta()->set_delete_predicate(std::move(delete_predicate));

        }

    }

}

int64_t Compaction::get_avg_segment_rows() {

    // take care of empty rowset

    // input_rowsets_size is total disk_size of input_rowset, this size is the

    // final size after codec and compress, so expect dest segment file size

    // in disk is config::vertical_compaction_max_segment_size

    const auto& meta = _tablet->tablet_meta();

    if (meta->compaction_policy() == CUMULATIVE_TIME_SERIES_POLICY) {

        int64_t compaction_goal_size_mbytes = meta->time_series_compaction_goal_size_mbytes();

        // The output segment rows should be less than total input rows

        return std::min((compaction_goal_size_mbytes * 1024 * 1024 * 2) /

                                (_input_rowsets_data_size / (_input_row_num + 1) + 1),

                        _input_row_num + 1);

    }

    return std::min(config::vertical_compaction_max_segment_size /

                            (_input_rowsets_data_size / (_input_row_num + 1) + 1),

                    _input_row_num + 1);

}

CompactionMixin::CompactionMixin(StorageEngine& engine, TabletSharedPtr tablet,

                                 const std::string& label)

        : Compaction(tablet, label), _engine(engine) {}

CompactionMixin::~CompactionMixin() {

    if (_state != CompactionState::SUCCESS && _output_rowset != nullptr) {

        if (!_output_rowset->is_local()) {

            tablet()->record_unused_remote_rowset(_output_rowset->rowset_id(),

                                                  _output_rowset->rowset_meta()->resource_id(),

                                                  _output_rowset->num_segments());

            return;

        }

        _engine.add_unused_rowset(_output_rowset);

    }

}

Tablet* CompactionMixin::tablet() {

    return static_cast<Tablet*>(_tablet.get());

}

Status CompactionMixin::do_compact_ordered_rowsets() {

    RETURN_IF_ERROR(build_basic_info(true));

    RowsetWriterContext ctx;

    RETURN_IF_ERROR(construct_output_rowset_writer(ctx));

    const auto& output_rowset_dir = tablet()->tablet_path();

    LOG(INFO) << "start to do ordered data compaction, tablet=" << _tablet->tablet_id()

              << ", output_version=" << _output_version;

    // link data to new rowset

    auto seg_id = 0;

    bool segments_key_bounds_truncated {false};

    bool any_input_aggregated {false};

    std::vector<KeyBoundsPB> segment_key_bounds;

    std::vector<uint32_t> num_segment_rows;

    for (auto rowset : _input_rowsets) {

        RETURN_IF_ERROR(

                rowset->link_files_to(output_rowset_dir, _output_rs_writer->rowset_id(), seg_id));

        seg_id += rowset->num_segments();

        segments_key_bounds_truncated |= rowset->is_segments_key_bounds_truncated();

        any_input_aggregated |= rowset->rowset_meta()->is_segments_key_bounds_aggregated();

        std::vector<KeyBoundsPB> key_bounds;

        RETURN_IF_ERROR(rowset->get_segments_key_bounds(&key_bounds));

        segment_key_bounds.insert(segment_key_bounds.end(), key_bounds.begin(), key_bounds.end());

        std::vector<uint32_t> input_segment_rows;

        rowset->get_num_segment_rows(&input_segment_rows);

        num_segment_rows.insert(num_segment_rows.end(), input_segment_rows.begin(),

                                input_segment_rows.end());

    }

    // build output rowset

    RowsetMetaSharedPtr rowset_meta = std::make_shared<RowsetMeta>();

    rowset_meta->set_num_rows(_input_row_num);

    rowset_meta->set_total_disk_size(_input_rowsets_data_size + _input_rowsets_index_size);

    rowset_meta->set_data_disk_size(_input_rowsets_data_size);

    rowset_meta->set_index_disk_size(_input_rowsets_index_size);

    rowset_meta->set_empty(_input_row_num == 0);

    rowset_meta->set_num_segments(_input_num_segments);

    rowset_meta->set_segments_overlap(_trigger_quick_merge_by_binlog ? OVERLAPPING

                                                                     : NONOVERLAPPING);

    rowset_meta->set_rowset_state(VISIBLE);

    rowset_meta->set_segments_key_bounds_truncated(segments_key_bounds_truncated);

    // If any input was already aggregated we have no way to recover per-segment

    // bounds, so force aggregation on the output to keep the layout consistent

    // with `num_segments` / the aggregated flag, even if the config is off now.

    bool aggregate_key_bounds =

            any_input_aggregated ||

            (config::enable_aggregate_non_mow_key_bounds &&

             !_tablet->enable_unique_key_merge_on_write() && !tablet()->is_row_binlog_tablet());

    rowset_meta->set_segments_key_bounds(segment_key_bounds, aggregate_key_bounds);

    rowset_meta->set_num_segment_rows(num_segment_rows);

    rowset_meta->set_commit_tso(commit_tso_range(_input_rowsets));

    _output_rowset = _output_rs_writer->manual_build(rowset_meta);

    // 2. check variant column path stats

    RETURN_IF_ERROR(variant_util::VariantCompactionUtil::check_path_stats(_input_rowsets,

                                                                          _output_rowset, _tablet));

    return Status::OK();

}

Status CompactionMixin::build_basic_info(bool is_ordered_compaction) {

    for (auto& rowset : _input_rowsets) {

        const auto& rowset_meta = rowset->rowset_meta();

        auto index_size = rowset_meta->index_disk_size();

        auto total_size = rowset_meta->total_disk_size();

        auto data_size = rowset_meta->data_disk_size();

        // corrupted index size caused by bug before 2.1.5 or 3.0.0 version

        // try to get real index size from disk.

        if (index_size < 0 || index_size > total_size * 2) {

            LOG(ERROR) << "invalid index size:" << index_size << " total size:" << total_size

                       << " data size:" << data_size << " tablet:" << rowset_meta->tablet_id()

                       << " rowset:" << rowset_meta->rowset_id();

            index_size = 0;

            auto st = rowset->get_inverted_index_size(&index_size);

            if (!st.ok()) {

                LOG(ERROR) << "failed to get inverted index size. res=" << st;

            }

        }

        _input_rowsets_data_size += data_size;

        _input_rowsets_index_size += index_size;

        _input_rowsets_total_size += total_size;

        _input_row_num += rowset->num_rows();

        _input_num_segments += rowset->num_segments();

    }

    COUNTER_UPDATE(_input_rowsets_data_size_counter, _input_rowsets_data_size);

    COUNTER_UPDATE(_input_row_num_counter, _input_row_num);

    COUNTER_UPDATE(_input_segments_num_counter, _input_num_segments);

    TEST_SYNC_POINT_RETURN_WITH_VALUE("compaction::CompactionMixin::build_basic_info",

                                      Status::OK());

    _output_version =

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

    _newest_write_timestamp = _input_rowsets.back()->newest_write_timestamp();

    std::vector<RowsetMetaSharedPtr> rowset_metas(_input_rowsets.size());

    std::transform(_input_rowsets.begin(), _input_rowsets.end(), rowset_metas.begin(),

                   [](const RowsetSharedPtr& rowset) { return rowset->rowset_meta(); });

    _cur_tablet_schema = _tablet->tablet_schema_with_merged_max_schema_version(rowset_metas);

    // if enable_vertical_compact_variant_subcolumns is true, we need to compact the variant subcolumns in seperate column groups

    // so get_extended_compaction_schema will extended the schema for variant columns

    // for ordered compaction, we don't need to extend the schema for variant columns

    if (_enable_vertical_compact_variant_subcolumns && !is_ordered_compaction) {

        RETURN_IF_ERROR(variant_util::VariantCompactionUtil::get_extended_compaction_schema(

                _input_rowsets, _cur_tablet_schema));

    }

    return Status::OK();

}

bool CompactionMixin::handle_ordered_data_compaction() {

    if (config::is_cloud_mode()) {

        return false;

    }

    if (!config::enable_ordered_data_compaction) {

        return false;

    }

    // If some rowsets has idx files and some rowsets has not, we can not do link file compaction.

    // Since the output rowset will be broken.

    // Use schema version instead of schema hash to check if they are the same,

    // because light schema change will not change the schema hash on BE, but will increase the schema version

    // See fe/fe-core/src/main/java/org/apache/doris/alter/SchemaChangeHandler.java::2979

    std::vector<int32_t> schema_versions_of_rowsets;

    for (auto input_rowset : _input_rowsets) {

        schema_versions_of_rowsets.push_back(input_rowset->rowset_meta()->schema_version());

    }

    // If all rowsets has same schema version, then we can do link file compaction directly.

    bool all_same_schema_version =

            std::all_of(schema_versions_of_rowsets.begin(), schema_versions_of_rowsets.end(),

                        [&](int32_t v) { return v == schema_versions_of_rowsets.front(); });

    if (!all_same_schema_version) {

        return false;

    }

    if (compaction_type() == ReaderType::READER_COLD_DATA_COMPACTION ||

        compaction_type() == ReaderType::READER_FULL_COMPACTION) {

        // The remote file system and full compaction does not support to link files.

        return false;

    }

    bool is_binlog_compaction = _tablet->is_row_binlog_tablet();

    if (is_binlog_compaction && !_input_rowsets.empty()) {

        RowsetIdUnorderedSet input_rowset_ids;

        input_rowset_ids.reserve(_input_rowsets.size());

        for (const auto& rs : _input_rowsets) {

            input_rowset_ids.insert(rs->rowset_id());

        }

        if (_tablet->tablet_meta()->delete_bitmap().contain_rowsets(input_rowset_ids)) {

            return false;

        }

    }

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

        _tablet->enable_unique_key_merge_on_write() && !is_binlog_compaction) {

        return false;

    }

    if (_tablet->tablet_meta()->tablet_schema()->skip_write_index_on_load()) {

        // Expected to create index through normal compaction

        return false;

    }

    // check delete version: if compaction type is base compaction and

    // has a delete version, use original compaction

    if (compaction_type() == ReaderType::READER_BASE_COMPACTION ||

        (_allow_delete_in_cumu_compaction &&

         compaction_type() == ReaderType::READER_CUMULATIVE_COMPACTION)) {

        for (auto& rowset : _input_rowsets) {

            if (rowset->rowset_meta()->has_delete_predicate()) {

                return false;

            }

        }

    }

    if (is_binlog_compaction) {

        DCHECK(!_input_rowsets.empty()) << "tablet=" << _tablet->tablet_id();

        auto compaction_level = _input_rowsets.front()->rowset_meta()->compaction_level();

        bool can_quick_merge_binlog =

                compaction_level ==

                        BinlogCumulativeCompactionPolicy::kBinlogCompactionMaxLevel - 1 &&

                _input_rowsets.size() >= 2 && _input_rowsets[0]->start_version() == 0;

        if (!can_quick_merge_binlog) {

            return false;

        }

        _trigger_quick_merge_by_binlog = true;

        // Binlog quick merge at LMax is a special meta/link compaction path selected by

        // BinlogCumulativeCompactionPolicy. It does not require the whole input to satisfy the normal

        // ordered-data tidy check, but the output rowset built by do_compact_ordered_rowsets()

        // is still NONOVERLAPPING (excluding row-binlog).

        auto st = do_compact_ordered_rowsets();

        if (!st.ok()) {

            LOG(WARNING) << "failed to compact ordered rowsets: " << st;

            _pending_rs_guard.drop();

        }

        return st.ok();

    } else {

        // Check if rowsets are tidy so we can just modify meta and do link files to handle

        // ordered data compaction.

        auto input_size = _input_rowsets.size();

        std::string pre_max_key;

        bool pre_rs_key_bounds_truncated {false};

        for (auto i = 0; i < input_size; ++i) {

            if (!is_rowset_tidy(pre_max_key, pre_rs_key_bounds_truncated, _input_rowsets[i])) {

                if (i <= input_size / 2) {

                    return false;

                } else {

                    _input_rowsets.resize(i);

                    break;

                }

            }

        }

    }

    // most rowset of current compaction is nonoverlapping

    // just handle nonoverlappint rowsets

    auto st = do_compact_ordered_rowsets();

    if (!st.ok()) {

        LOG(WARNING) << "failed to compact ordered rowsets: " << st;

        _pending_rs_guard.drop();

    }

    return st.ok();

}

Status CompactionMixin::execute_compact() {

    int64_t profile_start_time_ms = UnixMillis();

    uint32_t checksum_before;

    uint32_t checksum_after;

    bool enable_compaction_checksum = config::enable_compaction_checksum;

    if (enable_compaction_checksum) {

        EngineChecksumTask checksum_task(_engine, _tablet->tablet_id(), _tablet->schema_hash(),

                                         _input_rowsets.back()->end_version(), &checksum_before);

        auto st = checksum_task.execute();

        if (!st.ok()) {

            submit_profile_record(false, profile_start_time_ms, st.to_string());

            return st;

        }

    }

    auto* data_dir = tablet()->data_dir();

    int64_t permits = get_compaction_permits();

    data_dir->disks_compaction_score_increment(permits);

    data_dir->disks_compaction_num_increment(1);

    auto record_compaction_stats = [&](const doris::Exception& ex) {

        _tablet->compaction_count.fetch_add(1, std::memory_order_relaxed);

        data_dir->disks_compaction_score_increment(-permits);

        data_dir->disks_compaction_num_increment(-1);

    };

    // Handler for execute_compact_impl failure (both Status error and C++ exception).

    // The macro calls this then returns, so submit_profile_record(false) must be here.

    auto on_compact_impl_failure = [&](const doris::Exception& ex) {

        record_compaction_stats(ex);

        submit_profile_record(false, profile_start_time_ms,

                              ex.what() ? std::string(ex.what()) : "");

    };

    HANDLE_EXCEPTION_IF_CATCH_EXCEPTION(execute_compact_impl(permits), on_compact_impl_failure);

    // Only reached on success (macro returns on failure).

    record_compaction_stats(doris::Exception());

    if (enable_compaction_checksum) {

        EngineChecksumTask checksum_task(_engine, _tablet->tablet_id(), _tablet->schema_hash(),

                                         _input_rowsets.back()->end_version(), &checksum_after);

        auto st = checksum_task.execute();

        if (!st.ok()) {

            submit_profile_record(false, profile_start_time_ms, st.to_string());

            return st;

        }

        if (checksum_before != checksum_after) {

            auto mismatch_st = Status::InternalError(

                    "compaction tablet checksum not consistent, before={}, after={}, tablet_id={}",

                    checksum_before, checksum_after, _tablet->tablet_id());

            submit_profile_record(false, profile_start_time_ms, mismatch_st.to_string());

            return mismatch_st;

        }

    }

    DorisMetrics::instance()->local_compaction_read_rows_total->increment(_input_row_num);

    DorisMetrics::instance()->local_compaction_read_bytes_total->increment(

            _input_rowsets_total_size);

    TEST_SYNC_POINT_RETURN_WITH_VALUE("compaction::CompactionMixin::execute_compact", Status::OK());

    DorisMetrics::instance()->local_compaction_write_rows_total->increment(

            _output_rowset->num_rows());

    DorisMetrics::instance()->local_compaction_write_bytes_total->increment(

            _output_rowset->total_disk_size());

    _load_segment_to_cache();

    submit_profile_record(true, profile_start_time_ms);

    return Status::OK();

}

Status CompactionMixin::execute_compact_impl(int64_t permits) {

    OlapStopWatch watch;

    if (handle_ordered_data_compaction()) {

        _is_ordered_data_compaction = true;

        RETURN_IF_ERROR(modify_rowsets());

        int64_t input_compaction_level = _input_rowsets.front()->rowset_meta()->compaction_level();

        LOG(INFO) << "succeed to do ordered data " << compaction_name()

                  << ". tablet=" << _tablet->tablet_id() << ", output_version=" << _output_version

                  << ", disk=" << tablet()->data_dir()->path()

                  << ", input_compaction_level=" << input_compaction_level

                  << ", segments=" << _input_num_segments << ", input_row_num=" << _input_row_num

                  << ", output_row_num=" << _output_rowset->num_rows()

                  << ", input_rowsets_data_size=" << _input_rowsets_data_size

                  << ", input_rowsets_index_size=" << _input_rowsets_index_size

                  << ", input_rowsets_total_size=" << _input_rowsets_total_size

                  << ", output_rowset_data_size=" << _output_rowset->data_disk_size()

                  << ", output_rowset_index_size=" << _output_rowset->index_disk_size()

                  << ", output_rowset_total_size=" << _output_rowset->total_disk_size()

                  << ". elapsed time=" << watch.get_elapse_second() << "s.";

        _state = CompactionState::SUCCESS;

        return Status::OK();

    }

    RETURN_IF_ERROR(build_basic_info());

    TEST_SYNC_POINT_RETURN_WITH_VALUE("compaction::CompactionMixin::execute_compact_impl",

                                      Status::OK());

    VLOG_DEBUG << "dump tablet schema: " << _cur_tablet_schema->dump_structure();

    LOG(INFO) << "start " << compaction_name() << ". tablet=" << _tablet->tablet_id()

              << ", output_version=" << _output_version << ", permits: " << permits;

    RETURN_IF_ERROR(merge_input_rowsets());

    // Currently, updates are only made in the time_series and binlog policies.

    update_compaction_level();

    RETURN_IF_ERROR(modify_rowsets());

    auto* cumu_policy = tablet()->cumulative_compaction_policy();

    DCHECK(cumu_policy);

    int64_t input_compaction_level = _input_rowsets.front()->rowset_meta()->compaction_level();

    LOG(INFO) << "succeed to do " << compaction_name() << " is_vertical=" << _is_vertical

              << ". tablet=" << _tablet->tablet_id() << ", output_version=" << _output_version

              << ", current_max_version=" << tablet()->max_version().second

              << ", disk=" << tablet()->data_dir()->path()

              << ", input_compaction_level=" << input_compaction_level

              << ", input_segments=" << _input_num_segments << ", input_rowsets_data_size="

              << PrettyPrinter::print_bytes(_input_rowsets_data_size)

              << ", input_rowsets_index_size="

              << PrettyPrinter::print_bytes(_input_rowsets_index_size)

              << ", input_rowsets_total_size="

              << PrettyPrinter::print_bytes(_input_rowsets_total_size)

              << ", output_rowset_data_size="

              << PrettyPrinter::print_bytes(_output_rowset->data_disk_size())

              << ", output_rowset_index_size="

              << PrettyPrinter::print_bytes(_output_rowset->index_disk_size())

              << ", output_rowset_total_size="

              << PrettyPrinter::print_bytes(_output_rowset->total_disk_size())

              << ", input_row_num=" << _input_row_num

              << ", output_row_num=" << _output_rowset->num_rows()

              << ", filtered_row_num=" << _stats.filtered_rows

              << ", merged_row_num=" << _stats.merged_rows

              << ". elapsed time=" << watch.get_elapse_second()

              << "s. cumulative_compaction_policy=" << cumu_policy->name()

              << ", compact_row_per_second="

              << cast_set<double>(_input_row_num) / watch.get_elapse_second();

    _state = CompactionState::SUCCESS;

    return Status::OK();

}

Status Compaction::do_inverted_index_compaction() {

    const auto& ctx = _output_rs_writer->context();

    if (!_enable_inverted_index_compaction || _input_row_num <= 0 ||

        ctx.columns_to_do_index_compaction.empty()) {

        return Status::OK();

    }

    auto error_handler = [this](int64_t index_id, int64_t column_uniq_id) {

        LOG(WARNING) << "failed to do index compaction"

                     << ". tablet=" << _tablet->tablet_id() << ". column uniq id=" << column_uniq_id

                     << ". index_id=" << index_id;

        for (auto& rowset : _input_rowsets) {

            rowset->set_skip_index_compaction(cast_set<int32_t>(column_uniq_id));

            LOG(INFO) << "mark skipping inverted index compaction next time"

                      << ". tablet=" << _tablet->tablet_id() << ", rowset=" << rowset->rowset_id()

                      << ", column uniq id=" << column_uniq_id << ", index_id=" << index_id;

        }

    };

    DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_rowid_conversion_null",

                    { _stats.rowid_conversion = nullptr; })

    if (!_stats.rowid_conversion) {

        LOG(WARNING) << "failed to do index compaction, rowid conversion is null"

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

                     << ", input row number=" << _input_row_num;

        mark_skip_index_compaction(ctx, error_handler);

        return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                "failed to do index compaction, rowid conversion is null. tablet={}",

                _tablet->tablet_id());

    }

    OlapStopWatch inverted_watch;

    // translation vec

    // <<dest_idx_num, dest_docId>>

    // the first level vector: index indicates src segment.

    // the second level vector: index indicates row id of source segment,

    // value indicates row id of destination segment.

    // <UINT32_MAX, UINT32_MAX> indicates current row not exist.

    const auto& trans_vec = _stats.rowid_conversion->get_rowid_conversion_map();

    // source rowset,segment -> index_id

    const auto& src_seg_to_id_map = _stats.rowid_conversion->get_src_segment_to_id_map();

    // dest rowset id

    RowsetId dest_rowset_id = _stats.rowid_conversion->get_dst_rowset_id();

    // dest segment id -> num rows

    std::vector<uint32_t> dest_segment_num_rows;

    RETURN_IF_ERROR(_output_rs_writer->get_segment_num_rows(&dest_segment_num_rows));

    auto src_segment_num = src_seg_to_id_map.size();

    auto dest_segment_num = dest_segment_num_rows.size();

    // when all the input rowsets are deleted, the output rowset will be empty and dest_segment_num will be 0.

    if (dest_segment_num <= 0) {

        LOG(INFO) << "skip doing index compaction due to no output segments"

                  << ". tablet=" << _tablet->tablet_id() << ", input row number=" << _input_row_num

                  << ". elapsed time=" << inverted_watch.get_elapse_second() << "s.";

        return Status::OK();

    }

    // Only write info files when debug index compaction is enabled.

    // The files are used to debug index compaction and works with index_tool.

    if (config::debug_inverted_index_compaction) {

        // src index files

        // format: rowsetId_segmentId

        std::vector<std::string> src_index_files(src_segment_num);

        for (const auto& m : src_seg_to_id_map) {

            std::pair<RowsetId, uint32_t> p = m.first;

            src_index_files[m.second] = p.first.to_string() + "_" + std::to_string(p.second);

        }

        // dest index files

        // format: rowsetId_segmentId

        std::vector<std::string> dest_index_files(dest_segment_num);

        for (int i = 0; i < dest_segment_num; ++i) {

            auto prefix = dest_rowset_id.to_string() + "_" + std::to_string(i);

            dest_index_files[i] = prefix;

        }

        auto write_json_to_file = [&](const nlohmann::json& json_obj,

                                      const std::string& file_name) {

            io::FileWriterPtr file_writer;

            std::string file_path =

                    fmt::format("{}/{}.json", std::string(getenv("LOG_DIR")), file_name);

            RETURN_IF_ERROR(io::global_local_filesystem()->create_file(file_path, &file_writer));

            RETURN_IF_ERROR(file_writer->append(json_obj.dump()));

            RETURN_IF_ERROR(file_writer->append("\n"));

            return file_writer->close();

        };

        // Convert trans_vec to JSON and print it

        nlohmann::json trans_vec_json = trans_vec;

        auto output_version =

                _output_version.to_string().substr(1, _output_version.to_string().size() - 2);

        RETURN_IF_ERROR(write_json_to_file(

                trans_vec_json,

                fmt::format("trans_vec_{}_{}", _tablet->tablet_id(), output_version)));

        nlohmann::json src_index_files_json = src_index_files;

        RETURN_IF_ERROR(write_json_to_file(

                src_index_files_json,

                fmt::format("src_idx_dirs_{}_{}", _tablet->tablet_id(), output_version)));

        nlohmann::json dest_index_files_json = dest_index_files;

        RETURN_IF_ERROR(write_json_to_file(

                dest_index_files_json,

                fmt::format("dest_idx_dirs_{}_{}", _tablet->tablet_id(), output_version)));

        nlohmann::json dest_segment_num_rows_json = dest_segment_num_rows;

        RETURN_IF_ERROR(write_json_to_file(

                dest_segment_num_rows_json,

                fmt::format("dest_seg_num_rows_{}_{}", _tablet->tablet_id(), output_version)));

    }

    // create index_writer to compaction indexes

    std::unordered_map<RowsetId, Rowset*> rs_id_to_rowset_map;

    for (auto&& rs : _input_rowsets) {

        rs_id_to_rowset_map.emplace(rs->rowset_id(), rs.get());

    }

    // src index dirs

    std::vector<std::unique_ptr<IndexFileReader>> index_file_readers(src_segment_num);

    for (const auto& m : src_seg_to_id_map) {

        const auto& [rowset_id, seg_id] = m.first;

        auto find_it = rs_id_to_rowset_map.find(rowset_id);

        DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_find_rowset_error",

                        { find_it = rs_id_to_rowset_map.end(); })

        if (find_it == rs_id_to_rowset_map.end()) [[unlikely]] {

            LOG(WARNING) << "failed to do index compaction, cannot find rowset. tablet_id="

                         << _tablet->tablet_id() << " rowset_id=" << rowset_id.to_string();

            mark_skip_index_compaction(ctx, error_handler);

            return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                    "failed to do index compaction, cannot find rowset. tablet_id={} rowset_id={}",

                    _tablet->tablet_id(), rowset_id.to_string());

        }

        auto* rowset = find_it->second;

        auto fs = rowset->rowset_meta()->fs();

        DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_get_fs_error", { fs = nullptr; })

        if (!fs) {

            LOG(WARNING) << "failed to do index compaction, get fs failed. resource_id="

                         << rowset->rowset_meta()->resource_id();

            mark_skip_index_compaction(ctx, error_handler);

            return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                    "get fs failed, resource_id={}", rowset->rowset_meta()->resource_id());

        }

        auto seg_path = rowset->segment_path(seg_id);

        DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_seg_path_nullptr", {

            seg_path = ResultError(Status::Error<ErrorCode::INTERNAL_ERROR>(

                    "do_inverted_index_compaction_seg_path_nullptr"));

        })

        if (!seg_path.has_value()) {

            LOG(WARNING) << "failed to do index compaction, get segment path failed. tablet_id="

                         << _tablet->tablet_id() << " rowset_id=" << rowset_id.to_string()

                         << " seg_id=" << seg_id;

            mark_skip_index_compaction(ctx, error_handler);

            return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                    "get segment path failed. tablet_id={} rowset_id={} seg_id={}",

                    _tablet->tablet_id(), rowset_id.to_string(), seg_id);

        }

        auto index_file_reader = std::make_unique<IndexFileReader>(

                fs,

                std::string {InvertedIndexDescriptor::get_index_file_path_prefix(seg_path.value())},

                _cur_tablet_schema->get_inverted_index_storage_format(),

                rowset->rowset_meta()->inverted_index_file_info(seg_id), _tablet->tablet_id());

        auto st = index_file_reader->init(config::inverted_index_read_buffer_size);

        DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_init_inverted_index_file_reader",

                        {

                            st = Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                                    "debug point: "

                                    "Compaction::do_inverted_index_compaction_init_inverted_index_"

                                    "file_reader error");

                        })

        if (!st.ok()) {

            LOG(WARNING) << "failed to do index compaction, init inverted index file reader "

                            "failed. tablet_id="

                         << _tablet->tablet_id() << " rowset_id=" << rowset_id.to_string()

                         << " seg_id=" << seg_id;

            mark_skip_index_compaction(ctx, error_handler);

            return Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                    "init inverted index file reader failed. tablet_id={} rowset_id={} seg_id={}",

                    _tablet->tablet_id(), rowset_id.to_string(), seg_id);

        }

        index_file_readers[m.second] = std::move(index_file_reader);

    }

    // dest index files

    // format: rowsetId_segmentId

    auto& inverted_index_file_writers =

            dynamic_cast<BaseBetaRowsetWriter*>(_output_rs_writer.get())->index_file_writers();