// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#include "olap/compaction.h"

#include <fmt/format.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <atomic>

#include <cstdlib>

#include <list>

#include <map>

#include <memory>

#include <mutex>

#include <nlohmann/json.hpp>

#include <numeric>

#include <ostream>

#include <set>

#include <shared_mutex>

#include <utility>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "common/config.h"

#include "common/status.h"

#include "cpp/sync_point.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 "olap/cumulative_compaction_policy.h"

#include "olap/cumulative_compaction_time_series_policy.h"

#include "olap/data_dir.h"

#include "olap/olap_common.h"

#include "olap/olap_define.h"

#include "olap/rowset/beta_rowset.h"

#include "olap/rowset/beta_rowset_writer.h"

#include "olap/rowset/rowset.h"

#include "olap/rowset/rowset_fwd.h"

#include "olap/rowset/rowset_meta.h"

#include "olap/rowset/rowset_writer.h"

#include "olap/rowset/rowset_writer_context.h"

#include "olap/rowset/segment_v2/inverted_index_compaction.h"

#include "olap/rowset/segment_v2/inverted_index_desc.h"

#include "olap/rowset/segment_v2/inverted_index_file_reader.h"

#include "olap/rowset/segment_v2/inverted_index_file_writer.h"

#include "olap/rowset/segment_v2/inverted_index_fs_directory.h"

#include "olap/storage_engine.h"

#include "olap/storage_policy.h"

#include "olap/tablet.h"

#include "olap/tablet_meta.h"

#include "olap/tablet_meta_manager.h"

#include "olap/task/engine_checksum_task.h"

#include "olap/txn_manager.h"

#include "olap/utils.h"

#include "runtime/memory/mem_tracker_limiter.h"

#include "runtime/thread_context.h"

#include "util/time.h"

#include "util/trace.h"

using std::vector;

namespace doris {

using namespace ErrorCode;

namespace {

bool is_rowset_tidy(std::string& pre_max_key, 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;

    }

    if (min_key <= pre_max_key) {

        return false;

    }

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

    return true;

}

} // namespace

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

        : _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) {

    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();

}

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;

    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

        if (_is_vertical) {

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

                                                 input_rs_readers, _output_rs_writer.get(),

                                                 get_avg_segment_rows(), way_num, &_stats);

        } else {

            if (!_tablet->tablet_schema()->cluster_key_idxes().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()));

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

    // 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 (_allow_delete_in_cumu_compaction && _output_rowset->version().first > 2) {

        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));

        }

    }

    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();

}

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();

        return (compaction_goal_size_mbytes * 1024 * 1024 * 2) /

               (_input_rowsets_data_size / (_input_row_num + 1) + 1);

    }

    return config::vertical_compaction_max_segment_size /

           (_input_rowsets_data_size / (_input_row_num + 1) + 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() {

    build_basic_info();

    RowsetWriterContext ctx;

    RETURN_IF_ERROR(construct_output_rowset_writer(ctx));

    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;

    std::vector<KeyBoundsPB> segment_key_bounds;

    for (auto rowset : _input_rowsets) {

        RETURN_IF_ERROR(rowset->link_files_to(tablet()->tablet_path(),

                                              _output_rs_writer->rowset_id(), seg_id));

        seg_id += rowset->num_segments();

        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());

    }

    // 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(NONOVERLAPPING);

    rowset_meta->set_rowset_state(VISIBLE);

    rowset_meta->set_segments_key_bounds(segment_key_bounds);

    _output_rowset = _output_rs_writer->manual_build(rowset_meta);

    return Status::OK();

}

void CompactionMixin::build_basic_info() {

    for (auto& rowset : _input_rowsets) {

        _input_rowsets_data_size += rowset->data_disk_size();

        _input_rowsets_index_size += rowset->index_disk_size();

        _input_rowsets_total_size += rowset->total_disk_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);

    _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);

}

bool CompactionMixin::handle_ordered_data_compaction() {

    if (!config::enable_ordered_data_compaction) {

        return false;

    }

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

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

        return false;

    }

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

        _tablet->enable_unique_key_merge_on_write()) {

        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;

            }

        }

    }

    // check if rowsets are tidy so we can just modify meta and do link

    // files to handle compaction

    auto input_size = _input_rowsets.size();

    std::string pre_max_key;

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

        if (!is_rowset_tidy(pre_max_key, _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() {

    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);

        RETURN_IF_ERROR(checksum_task.execute());

    }

    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);

    };

    HANDLE_EXCEPTION_IF_CATCH_EXCEPTION(execute_compact_impl(permits), record_compaction_stats);

    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);

        RETURN_IF_ERROR(checksum_task.execute());

        if (checksum_before != checksum_after) {

            return Status::InternalError(

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

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

        }

    }

    _load_segment_to_cache();

    return Status::OK();

}

Status CompactionMixin::execute_compact_impl(int64_t permits) {

    OlapStopWatch watch;

    if (handle_ordered_data_compaction()) {

        RETURN_IF_ERROR(modify_rowsets());

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

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

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

                  << ", 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();

    }

    build_basic_info();

    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());

    RETURN_IF_ERROR(modify_rowsets());

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

    DCHECK(cumu_policy);

    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() << ", segments=" << _input_num_segments

              << ", input_data_size=" << _input_rowsets_data_size

              << ", output_rowset_size=" << _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=" << int(_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 (!config::inverted_index_compaction_enable || _input_row_num <= 0 ||

        !_stats.rowid_conversion || ctx.columns_to_do_index_compaction.empty()) {

        return Status::OK();

    }

    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();

    DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_dest_segment_num_is_zero",

                    { dest_segment_num = 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<InvertedIndexFileReader>> inverted_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]] {

            // DCHECK(false) << _tablet->tablet_id() << ' ' << rowset_id;

            return Status::InternalError("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) {

            return Status::InternalError("get fs failed, resource_id={}",

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

        }

        auto seg_path = DORIS_TRY(rowset->segment_path(seg_id));

        auto inverted_index_file_reader = std::make_unique<InvertedIndexFileReader>(

                fs, std::string {InvertedIndexDescriptor::get_index_file_path_prefix(seg_path)},

                _cur_tablet_schema->get_inverted_index_storage_format(),

                rowset->rowset_meta()->inverted_index_file_info(seg_id));

        RETURN_NOT_OK_STATUS_WITH_WARN(

                inverted_index_file_reader->init(config::inverted_index_read_buffer_size),

                "inverted_index_file_reader init faiqled");

        inverted_index_file_readers[m.second] = std::move(inverted_index_file_reader);

    }

    // dest index files

    // format: rowsetId_segmentId

    auto& inverted_index_file_writers = dynamic_cast<BaseBetaRowsetWriter*>(_output_rs_writer.get())

                                                ->inverted_index_file_writers();

    DCHECK_EQ(inverted_index_file_writers.size(), dest_segment_num);

    // use tmp file dir to store index files

    auto tmp_file_dir = ExecEnv::GetInstance()->get_tmp_file_dirs()->get_tmp_file_dir();

    auto index_tmp_path = tmp_file_dir / dest_rowset_id.to_string();

    LOG(INFO) << "start index compaction"

              << ". tablet=" << _tablet->tablet_id() << ", source index size=" << src_segment_num

              << ", destination index size=" << dest_segment_num << ".";

    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(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;

        }

    };

    Status status = Status::OK();

    for (auto&& column_uniq_id : ctx.columns_to_do_index_compaction) {

        auto col = _cur_tablet_schema->column_by_uid(column_uniq_id);

        const auto* index_meta = _cur_tablet_schema->inverted_index(col);

        DBUG_EXECUTE_IF("Compaction::do_inverted_index_compaction_can_not_find_index_meta",

                        { index_meta = nullptr; })

        if (index_meta == nullptr) {

            status = Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(

                    fmt::format("Can not find index_meta for col {}", col.name()));

            break;

        }

        std::vector<lucene::store::Directory*> dest_index_dirs(dest_segment_num);

        try {

            std::vector<std::unique_ptr<DorisCompoundReader>> src_idx_dirs(src_segment_num);

            for (int src_segment_id = 0; src_segment_id < src_segment_num; src_segment_id++) {

                src_idx_dirs[src_segment_id] =

                        DORIS_TRY(inverted_index_file_readers[src_segment_id]->open(index_meta));

            }

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

                auto dest_dir =

                        DORIS_TRY(inverted_index_file_writers[dest_segment_id]->open(index_meta));

                // Destination directories in dest_index_dirs do not need to be deconstructed,

                // but their lifecycle must be managed by inverted_index_file_writers.

                dest_index_dirs[dest_segment_id] = dest_dir.get();

            }

            auto st = compact_column(index_meta->index_id(), src_idx_dirs, dest_index_dirs,

                                     index_tmp_path.native(), trans_vec, dest_segment_num_rows);

            if (!st.ok()) {

                error_handler(index_meta->index_id(), column_uniq_id);

                status = Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(st.msg());

            }

        } catch (CLuceneError& e) {

            error_handler(index_meta->index_id(), column_uniq_id);

            status = Status::Error<INVERTED_INDEX_COMPACTION_ERROR>(e.what());

        }

    }

    // check index compaction status. If status is not ok, we should return error and end this compaction round.

    if (!status.ok()) {

        return status;

    }

    LOG(INFO) << "succeed to do index compaction"

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

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

    return Status::OK();

}

void Compaction::construct_index_compaction_columns(RowsetWriterContext& ctx) {

    for (const auto& index : _cur_tablet_schema->inverted_indexes()) {

        auto col_unique_ids = index->col_unique_ids();

        // check if column unique ids is empty to avoid crash

        if (col_unique_ids.empty()) {

            LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] index[" << index->index_id()

                         << "] has no column unique id, will skip index compaction."

                         << " tablet_schema=" << _cur_tablet_schema->dump_full_schema();

            continue;

        }

        auto col_unique_id = col_unique_ids[0];

        // Avoid doing inverted index compaction on non-slice type columns

        if (!field_is_slice_type(_cur_tablet_schema->column_by_uid(col_unique_id).type())) {

            continue;

        }

        // if index properties are different, index compaction maybe needs to be skipped.

        bool is_continue = false;

        std::optional<std::map<std::string, std::string>> first_properties;

        for (const auto& rowset : _input_rowsets) {

            const auto* tablet_index = rowset->tablet_schema()->inverted_index(col_unique_id);

            // no inverted index or index id is different from current index id

            if (tablet_index == nullptr || tablet_index->index_id() != index->index_id()) {

                is_continue = true;

                break;

            }

            auto properties = tablet_index->properties();

            if (!first_properties.has_value()) {

                first_properties = properties;

            } else {

                DBUG_EXECUTE_IF(

                        "Compaction::do_inverted_index_compaction_index_properties_different",

                        { properties.emplace("dummy_key", "dummy_value"); })

                if (properties != first_properties.value()) {

                    is_continue = true;

                    break;

                }

            }

        }

        if (is_continue) {

            continue;

        }

        auto has_inverted_index = [&](const RowsetSharedPtr& src_rs) {

            auto* rowset = static_cast<BetaRowset*>(src_rs.get());

            DBUG_EXECUTE_IF("Compaction::construct_skip_inverted_index_is_skip_index_compaction",

                            { rowset->set_skip_index_compaction(col_unique_id); })

            if (rowset->is_skip_index_compaction(col_unique_id)) {

                LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] rowset["

                             << rowset->rowset_id() << "] column_unique_id[" << col_unique_id

                             << "] skip inverted index compaction due to last failure";

                return false;

            }

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

            DBUG_EXECUTE_IF("Compaction::construct_skip_inverted_index_get_fs_error",

                            { fs = nullptr; })

            if (!fs) {

                LOG(WARNING) << "get fs failed, resource_id="

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

                return false;

            }

            const auto* index_meta = rowset->tablet_schema()->inverted_index(col_unique_id);

            DBUG_EXECUTE_IF("Compaction::construct_skip_inverted_index_index_meta_nullptr",

                            { index_meta = nullptr; })

            if (index_meta == nullptr) {

                LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["

                             << col_unique_id << "] index meta is null, will skip index compaction";

                return false;

            }

            for (auto i = 0; i < rowset->num_segments(); i++) {

                // TODO: inverted_index_path

                auto seg_path = rowset->segment_path(i);

                DBUG_EXECUTE_IF("Compaction::construct_skip_inverted_index_seg_path_nullptr", {

                    seg_path = ResultError(Status::Error<ErrorCode::INTERNAL_ERROR>("error"));

                })

                if (!seg_path) {

                    LOG(WARNING) << seg_path.error();

                    return false;

                }

                std::string index_file_path;

                try {

                    auto inverted_index_file_reader = std::make_unique<InvertedIndexFileReader>(

                            fs,

                            std::string {

                                    InvertedIndexDescriptor::get_index_file_path_prefix(*seg_path)},

                            _cur_tablet_schema->get_inverted_index_storage_format(),

                            rowset->rowset_meta()->inverted_index_file_info(i));

                    auto st = inverted_index_file_reader->init(

                            config::inverted_index_read_buffer_size);

                    index_file_path = inverted_index_file_reader->get_index_file_path(index_meta);

                    DBUG_EXECUTE_IF(

                            "Compaction::construct_skip_inverted_index_index_file_reader_init_"

                            "status_not_ok",

                            {

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

                                        "debug point: "

                                        "construct_skip_inverted_index_index_file_reader_init_"

                                        "status_"

                                        "not_ok");

                            })

                    if (!st.ok()) {

                        LOG(WARNING) << "init index " << index_file_path << " error:" << st;

                        return false;

                    }

                    // check index meta

                    auto result = inverted_index_file_reader->open(index_meta);

                    DBUG_EXECUTE_IF(

                            "Compaction::construct_skip_inverted_index_index_file_reader_open_"

                            "error",

                            {

                                result = ResultError(

                                        Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                                                "CLuceneError occur when open idx file"));

                            })

                    if (!result.has_value()) {

                        LOG(WARNING)

                                << "open index " << index_file_path << " error:" << result.error();

                        return false;

                    }

                    auto reader = std::move(result.value());

                    std::vector<std::string> files;

                    reader->list(&files);

                    reader->close();

                    DBUG_EXECUTE_IF(

                            "Compaction::construct_skip_inverted_index_index_reader_close_error",

                            { _CLTHROWA(CL_ERR_IO, "debug point: reader close error"); })

                    DBUG_EXECUTE_IF("Compaction::construct_skip_inverted_index_index_files_count",

                                    { files.clear(); })

                    // why is 3?

                    // slice type index file at least has 3 files: null_bitmap, segments_N, segments.gen

                    if (files.size() < 3) {

                        LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["

                                     << col_unique_id << "]," << index_file_path

                                     << " is corrupted, will skip index compaction";

                        return false;

                    }

                } catch (CLuceneError& err) {

                    LOG(WARNING) << "tablet[" << _tablet->tablet_id() << "] column_unique_id["

                                 << col_unique_id << "] open index[" << index_file_path

                                 << "], will skip index compaction, error:" << err.what();

                    return false;

                }

            }

            return true;

        };

        bool all_have_inverted_index = std::all_of(_input_rowsets.begin(), _input_rowsets.end(),

                                                   std::move(has_inverted_index));

        if (all_have_inverted_index) {

            ctx.columns_to_do_index_compaction.insert(col_unique_id);

        }

    }

}

Status CompactionMixin::construct_output_rowset_writer(RowsetWriterContext& ctx) {

    // only do index compaction for dup_keys and unique_keys with mow enabled

    if (config::inverted_index_compaction_enable &&

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

           _tablet->enable_unique_key_merge_on_write()) ||

          _tablet->keys_type() == KeysType::DUP_KEYS))) {

        construct_index_compaction_columns(ctx);

    }

    ctx.version = _output_version;

    ctx.rowset_state = VISIBLE;

    ctx.segments_overlap = NONOVERLAPPING;

    ctx.tablet_schema = _cur_tablet_schema;

    ctx.newest_write_timestamp = _newest_write_timestamp;

    ctx.write_type = DataWriteType::TYPE_COMPACTION;

    _output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(ctx, _is_vertical));

    _pending_rs_guard = _engine.add_pending_rowset(ctx);

    return Status::OK();

}

Status CompactionMixin::modify_rowsets() {

    std::vector<RowsetSharedPtr> output_rowsets;

    output_rowsets.push_back(_output_rowset);

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

        _tablet->enable_unique_key_merge_on_write()) {

        Version version = tablet()->max_version();

        DeleteBitmap output_rowset_delete_bitmap(_tablet->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->tablet_id();

        }

        std::unique_ptr<std::map<RowsetSharedPtr, RowLocationPairList>> location_map;

        if (config::enable_rowid_conversion_correctness_check) {

            location_map = std::make_unique<std::map<RowsetSharedPtr, RowLocationPairList>>();

            LOG(INFO) << "Location Map inited succ for tablet:" << _tablet->tablet_id();

        }

        // Convert the delete bitmap of the input rowsets to output rowset.

        // New loads are not blocked, so some keys of input rowsets might

        // be deleted during the time. We need to deal with delete bitmap

        // of incremental data later.

        // TODO(LiaoXin): check if there are duplicate keys

        std::size_t missed_rows_size = 0;

        tablet()->calc_compaction_output_rowset_delete_bitmap(

                _input_rowsets, *_rowid_conversion, 0, version.second + 1, missed_rows.get(),

                location_map.get(), _tablet->tablet_meta()->delete_bitmap(),

                &output_rowset_delete_bitmap);

        if (missed_rows) {

            missed_rows_size = missed_rows->size();

            std::size_t merged_missed_rows_size = _stats.merged_rows;

            if (!_tablet->tablet_meta()->tablet_schema()->cluster_key_idxes().empty()) {

                merged_missed_rows_size += _stats.filtered_rows;

            }

            if (_tablet->tablet_state() == TABLET_RUNNING &&

                merged_missed_rows_size != missed_rows_size) {

                std::stringstream ss;

                ss << "cumulative compaction: the merged rows(" << _stats.merged_rows

                   << "), filtered rows(" << _stats.filtered_rows

                   << ") is not equal to missed rows(" << missed_rows_size

                   << ") in rowid conversion, tablet_id: " << _tablet->tablet_id()

                   << ", table_id:" << _tablet->table_id();

                if (missed_rows_size == 0) {

                    ss << ", debug info: ";

                    DeleteBitmap subset_map(_tablet->tablet_id());

                    for (auto rs : _input_rowsets) {

                        _tablet->tablet_meta()->delete_bitmap().subset(

                                {rs->rowset_id(), 0, 0},

                                {rs->rowset_id(), rs->num_segments(), version.second + 1},

                                &subset_map);

                        ss << "(rowset id: " << rs->rowset_id()

                           << ", delete bitmap cardinality: " << subset_map.cardinality() << ")";

                    }

                    ss << ", version[0-" << version.second + 1 << "]";

                }

                std::string err_msg = fmt::format(

                        "cumulative compaction: the merged rows({}), filtered rows({})"

                        " is not equal to missed rows({}) in rowid conversion,"

                        " tablet_id: {}, table_id:{}",

                        _stats.merged_rows, _stats.filtered_rows, missed_rows_size,

                        _tablet->tablet_id(), _tablet->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(tablet()->check_rowid_conversion(_output_rowset, *location_map));

            location_map->clear();