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

#include <mutex>

#include "common/logging.h"

#include "common/status.h"

#include "storage/field.h"

#include "storage/index/index_file_reader.h"

#include "storage/index/index_file_writer.h"

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

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

#include "storage/olap_define.h"

#include "storage/rowset/beta_rowset.h"

#include "storage/rowset/rowset_writer_context.h"

#include "storage/segment/segment_loader.h"

#include "storage/storage_engine.h"

#include "storage/tablet/tablet_schema.h"

#include "util/debug_points.h"

#include "util/trace.h"

namespace doris {

#include "common/compile_check_begin.h"

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

                           const std::vector<TColumn>& columns,

                           const std::vector<doris::TOlapTableIndex>& alter_inverted_indexes,

                           bool is_drop_op)

        : _engine(engine),

          _tablet(std::move(tablet)),

          _columns(columns),

          _alter_inverted_indexes(alter_inverted_indexes),

          _is_drop_op(is_drop_op) {

    _olap_data_convertor = std::make_unique<OlapBlockDataConvertor>();

}

IndexBuilder::~IndexBuilder() {

    _olap_data_convertor.reset();

    _index_column_writers.clear();

}

Status IndexBuilder::init() {

    for (auto inverted_index : _alter_inverted_indexes) {

        _alter_index_ids.insert(inverted_index.index_id);

    }

    return Status::OK();

}

Status IndexBuilder::update_inverted_index_info() {

    // just do link files

    LOG(INFO) << "begin to update_inverted_index_info, tablet=" << _tablet->tablet_id()

              << ", is_drop_op=" << _is_drop_op;

    // index ids that will not be linked

    std::set<int64_t> without_index_uids;

    _output_rowsets.reserve(_input_rowsets.size());

    _pending_rs_guards.reserve(_input_rowsets.size());

    for (auto&& input_rowset : _input_rowsets) {

        bool is_local_rowset = input_rowset->is_local();

        DBUG_EXECUTE_IF("IndexBuilder::update_inverted_index_info_is_local_rowset",

                        { is_local_rowset = false; })

        if (!is_local_rowset) [[unlikely]] {

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

            return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}",

                                         _tablet->tablet_id(),

                                         input_rowset->rowset_id().to_string());

        }

        TabletSchemaSPtr output_rs_tablet_schema = std::make_shared<TabletSchema>();

        const auto& input_rs_tablet_schema = input_rowset->tablet_schema();

        output_rs_tablet_schema->copy_from(*input_rs_tablet_schema);

        int64_t total_index_size = 0;

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

        auto size_st = beta_rowset->get_inverted_index_size(&total_index_size);

        DBUG_EXECUTE_IF("IndexBuilder::update_inverted_index_info_size_st_not_ok", {

            size_st = Status::Error<ErrorCode::INIT_FAILED>("debug point: get fs failed");

        })

        if (!size_st.ok() && !size_st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>() &&

            !size_st.is<ErrorCode::NOT_FOUND>()) {

            return size_st;

        }

        auto num_segments = input_rowset->num_segments();

        size_t drop_index_size = 0;

        if (_is_drop_op) {

            for (const auto& t_inverted_index : _alter_inverted_indexes) {

                DCHECK_EQ(t_inverted_index.columns.size(), 1);

                auto column_name = t_inverted_index.columns[0];

                auto column_idx = output_rs_tablet_schema->field_index(column_name);

                if (column_idx < 0) {

                    if (!t_inverted_index.column_unique_ids.empty()) {

                        auto column_unique_id = t_inverted_index.column_unique_ids[0];

                        column_idx = output_rs_tablet_schema->field_index(column_unique_id);

                    }

                    if (column_idx < 0) {

                        LOG(WARNING) << "referenced column was missing. "

                                     << "[column=" << column_name

                                     << " referenced_column=" << column_idx << "]";

                        continue;

                    }

                }

                auto column = output_rs_tablet_schema->column(column_idx);

                // inverted index

                auto index_metas = output_rs_tablet_schema->inverted_indexs(column);

                for (const auto& index_meta : index_metas) {

                    // Only drop the index that matches the requested index_id,

                    // not all indexes on this column

                    if (index_meta->index_id() != t_inverted_index.index_id) {

                        continue;

                    }

                    if (output_rs_tablet_schema->get_inverted_index_storage_format() ==

                        InvertedIndexStorageFormatPB::V1) {

                        const auto& fs = io::global_local_filesystem();

                        for (int seg_id = 0; seg_id < num_segments; seg_id++) {

                            auto seg_path = local_segment_path(

                                    _tablet->tablet_path(), input_rowset->rowset_id().to_string(),

                                    seg_id);

                            auto index_path = InvertedIndexDescriptor::get_index_file_path_v1(

                                    InvertedIndexDescriptor::get_index_file_path_prefix(seg_path),

                                    index_meta->index_id(), index_meta->get_index_suffix());

                            int64_t index_size = 0;

                            RETURN_IF_ERROR(fs->file_size(index_path, &index_size));

                            VLOG_DEBUG << "inverted index file:" << index_path

                                       << " size:" << index_size;

                            drop_index_size += index_size;

                        }

                    }

                    _dropped_inverted_indexes.push_back(*index_meta);

                    // ATTN: DO NOT REMOVE INDEX AFTER OUTPUT_ROWSET_WRITER CREATED.

                    // remove dropped index_meta from output rowset tablet schema

                    output_rs_tablet_schema->remove_index(index_meta->index_id());

                }

                // ann index

                const auto* ann_index = output_rs_tablet_schema->ann_index(column);

                if (!ann_index) {

                    continue;

                }

                // Only drop the ann index that matches the requested index_id

                if (ann_index->index_id() != t_inverted_index.index_id) {

                    continue;

                }

                DCHECK(output_rs_tablet_schema->get_inverted_index_storage_format() !=

                       InvertedIndexStorageFormatPB::V1);

                _dropped_inverted_indexes.push_back(*ann_index);

                // ATTN: DO NOT REMOVE INDEX AFTER OUTPUT_ROWSET_WRITER CREATED.

                // remove dropped index_meta from output rowset tablet schema

                output_rs_tablet_schema->remove_index(ann_index->index_id());

            }

            DBUG_EXECUTE_IF("index_builder.update_inverted_index_info.drop_index", {

                auto indexes_count = DebugPoints::instance()->get_debug_param_or_default<int32_t>(

                        "index_builder.update_inverted_index_info.drop_index", "indexes_count", 0);

                if (indexes_count < 0) {

                    return Status::Error<ErrorCode::INTERNAL_ERROR>(

                            "indexes count cannot be negative");

                }

                auto indexes_size = output_rs_tablet_schema->inverted_indexes().size();

                if (indexes_count != indexes_size) {

                    return Status::Error<ErrorCode::INTERNAL_ERROR>(

                            "indexes count not equal to expected");

                }

            })

        } else {

            // base on input rowset's tablet_schema to build

            // output rowset's tablet_schema which only add

            // the indexes specified in this build index request

            for (auto t_inverted_index : _alter_inverted_indexes) {

                TabletIndex index;

                index.init_from_thrift(t_inverted_index, *input_rs_tablet_schema);

                auto column_uid = index.col_unique_ids()[0];

                if (column_uid < 0) {

                    LOG(WARNING) << "referenced column was missing. "

                                 << "[column=" << t_inverted_index.columns[0]

                                 << " referenced_column=" << column_uid << "]";

                    continue;

                }

                const TabletColumn& col = output_rs_tablet_schema->column_by_uid(column_uid);

                // inverted index

                auto exist_indexs = output_rs_tablet_schema->inverted_indexs(col);

                for (const auto& exist_index : exist_indexs) {

                    if (exist_index->index_id() != index.index_id()) {

                        if (exist_index->is_same_except_id(&index)) {

                            LOG(WARNING) << fmt::format(

                                    "column: {} has a exist inverted index, but the index id not "

                                    "equal "

                                    "request's index id, , exist index id: {}, request's index id: "

                                    "{}, "

                                    "remove exist index in new output_rs_tablet_schema",

                                    column_uid, exist_index->index_id(), index.index_id());

                            without_index_uids.insert(exist_index->index_id());

                            output_rs_tablet_schema->remove_index(exist_index->index_id());

                        }

                    }

                }

                // ann index

                const auto* exist_index = output_rs_tablet_schema->ann_index(col);

                if (exist_index && exist_index->index_id() != index.index_id()) {

                    if (exist_index->is_same_except_id(&index)) {

                        LOG(WARNING) << fmt::format(

                                "column: {} has a exist ann index, but the index id not "

                                "equal request's index id, , exist index id: {}, request's index "

                                "id: {}, remove exist index in new output_rs_tablet_schema",

                                column_uid, exist_index->index_id(), index.index_id());

                        without_index_uids.insert(exist_index->index_id());

                        output_rs_tablet_schema->remove_index(exist_index->index_id());

                    }

                }

                output_rs_tablet_schema->append_index(std::move(index));

            }

        }

        // construct input rowset reader

        RowsetReaderSharedPtr input_rs_reader;

        RETURN_IF_ERROR(input_rowset->create_reader(&input_rs_reader));

        // construct output rowset writer

        RowsetWriterContext context;

        context.version = input_rs_reader->version();

        context.rowset_state = VISIBLE;

        context.segments_overlap = input_rowset->rowset_meta()->segments_overlap();

        context.tablet_schema = output_rs_tablet_schema;

        context.newest_write_timestamp = input_rs_reader->newest_write_timestamp();

        auto output_rs_writer = DORIS_TRY(_tablet->create_rowset_writer(context, false));

        _pending_rs_guards.push_back(_engine.add_pending_rowset(context));

        // if without_index_uids is not empty, copy _alter_index_ids to it

        // else just use _alter_index_ids to avoid copy

        if (!without_index_uids.empty()) {

            without_index_uids.insert(_alter_index_ids.begin(), _alter_index_ids.end());

        }

        // build output rowset

        RETURN_IF_ERROR(input_rowset->link_files_to(

                _tablet->tablet_path(), output_rs_writer->rowset_id(), 0,

                without_index_uids.empty() ? &_alter_index_ids : &without_index_uids));

        auto input_rowset_meta = input_rowset->rowset_meta();

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

        rowset_meta->set_num_rows(input_rowset_meta->num_rows());

        if (output_rs_tablet_schema->get_inverted_index_storage_format() ==

            InvertedIndexStorageFormatPB::V1) {

            if (_is_drop_op) {

                VLOG_DEBUG << "data_disk_size:" << input_rowset_meta->data_disk_size()

                           << " total_disk_size:" << input_rowset_meta->total_disk_size()

                           << " index_disk_size:" << input_rowset_meta->index_disk_size()

                           << " drop_index_size:" << drop_index_size;

                rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size() -

                                                 drop_index_size);

                rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());

                rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size() -

                                                 drop_index_size);

            } else {

                rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size());

                rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());

                rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size());

            }

        } else {

            for (int seg_id = 0; seg_id < num_segments; seg_id++) {

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

                auto idx_file_reader = std::make_unique<IndexFileReader>(

                        context.fs(),

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

                        output_rs_tablet_schema->get_inverted_index_storage_format());

                auto st = idx_file_reader->init();

                DBUG_EXECUTE_IF(

                        "IndexBuilder::update_inverted_index_info_index_file_reader_init_not_ok", {

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

                                    "debug point: reader init error");

                        })

                if (!st.ok() && !st.is<ErrorCode::INVERTED_INDEX_FILE_NOT_FOUND>()) {

                    return st;

                }

                _index_file_readers.emplace(

                        std::make_pair(output_rs_writer->rowset_id().to_string(), seg_id),

                        std::move(idx_file_reader));

            }

            rowset_meta->set_total_disk_size(input_rowset_meta->total_disk_size() -

                                             total_index_size);

            rowset_meta->set_data_disk_size(input_rowset_meta->data_disk_size());

            rowset_meta->set_index_disk_size(input_rowset_meta->index_disk_size() -

                                             total_index_size);

        }

        rowset_meta->set_empty(input_rowset_meta->empty());

        rowset_meta->set_num_segments(input_rowset_meta->num_segments());

        rowset_meta->set_segments_overlap(input_rowset_meta->segments_overlap());

        rowset_meta->set_rowset_state(input_rowset_meta->rowset_state());

        std::vector<KeyBoundsPB> key_bounds;

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

        rowset_meta->set_segments_key_bounds_truncated(

                input_rowset_meta->is_segments_key_bounds_truncated());

        rowset_meta->set_segments_key_bounds(key_bounds);

        std::vector<uint32_t> num_segment_rows;

        input_rowset_meta->get_num_segment_rows(&num_segment_rows);

        rowset_meta->set_num_segment_rows(num_segment_rows);

        auto output_rowset = output_rs_writer->manual_build(rowset_meta);

        if (input_rowset_meta->has_delete_predicate()) {

            output_rowset->rowset_meta()->set_delete_predicate(

                    input_rowset_meta->delete_predicate());

        }

        _output_rowsets.push_back(output_rowset);

    }

    return Status::OK();

}

Status IndexBuilder::handle_single_rowset(RowsetMetaSharedPtr output_rowset_meta,

                                          std::vector<segment_v2::SegmentSharedPtr>& segments) {

    bool is_local_rowset = output_rowset_meta->is_local();

    DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_is_local_rowset",

                    { is_local_rowset = false; })

    if (!is_local_rowset) [[unlikely]] {

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

        return Status::InternalError("should be local rowset. tablet_id={} rowset_id={}",

                                     _tablet->tablet_id(),

                                     output_rowset_meta->rowset_id().to_string());

    }

    if (_is_drop_op) {

        const auto& output_rs_tablet_schema = output_rowset_meta->tablet_schema();

        if (output_rs_tablet_schema->get_inverted_index_storage_format() !=

            InvertedIndexStorageFormatPB::V1) {

            const auto& fs = output_rowset_meta->fs();

            const auto& output_rowset_schema = output_rowset_meta->tablet_schema();

            size_t inverted_index_size = 0;

            for (auto& seg_ptr : segments) {

                auto idx_file_reader_iter = _index_file_readers.find(

                        std::make_pair(output_rowset_meta->rowset_id().to_string(), seg_ptr->id()));

                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_can_not_find_reader_drop_op",

                                { idx_file_reader_iter = _index_file_readers.end(); })

                if (idx_file_reader_iter == _index_file_readers.end()) {

                    LOG(ERROR) << "idx_file_reader_iter" << output_rowset_meta->rowset_id() << ":"

                               << seg_ptr->id() << " cannot be found";

                    continue;

                }

                auto dirs = DORIS_TRY(idx_file_reader_iter->second->get_all_directories());

                std::string index_path_prefix {

                        InvertedIndexDescriptor::get_index_file_path_prefix(local_segment_path(

                                _tablet->tablet_path(), output_rowset_meta->rowset_id().to_string(),

                                seg_ptr->id()))};

                std::string index_path =

                        InvertedIndexDescriptor::get_index_file_path_v2(index_path_prefix);

                io::FileWriterPtr file_writer;

                Status st = fs->create_file(index_path, &file_writer);

                if (!st.ok()) {

                    LOG(WARNING) << "failed to create writable file. path=" << index_path

                                 << ", err: " << st;

                    return st;

                }

                auto index_file_writer = std::make_unique<IndexFileWriter>(

                        fs, std::move(index_path_prefix),

                        output_rowset_meta->rowset_id().to_string(), seg_ptr->id(),

                        output_rowset_schema->get_inverted_index_storage_format(),

                        std::move(file_writer));

                RETURN_IF_ERROR(index_file_writer->initialize(dirs));

                // create inverted index writer

                for (auto& index_meta : _dropped_inverted_indexes) {

                    RETURN_IF_ERROR(index_file_writer->delete_index(&index_meta));

                }

                _index_file_writers.emplace(seg_ptr->id(), std::move(index_file_writer));

            }

            for (auto&& [seg_id, index_file_writer] : _index_file_writers) {

                auto st = index_file_writer->begin_close();

                if (!st.ok()) {

                    LOG(ERROR) << "close index_file_writer error:" << st;

                    return st;

                }

                inverted_index_size += index_file_writer->get_index_file_total_size();

            }

            for (auto&& [seg_id, index_file_writer] : _index_file_writers) {

                auto st = index_file_writer->finish_close();

                if (!st.ok()) {

                    LOG(ERROR) << "wait close index_file_writer error:" << st;

                    return st;

                }

            }

            _index_file_writers.clear();

            output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());

            output_rowset_meta->set_total_disk_size(output_rowset_meta->total_disk_size() +

                                                    inverted_index_size);

            output_rowset_meta->set_index_disk_size(output_rowset_meta->index_disk_size() +

                                                    inverted_index_size);

        }

        LOG(INFO) << "all row nums. source_rows=" << output_rowset_meta->num_rows();

        return Status::OK();

    } else {

        // create inverted or ann index writer

        const auto& fs = output_rowset_meta->fs();

        auto output_rowset_schema = output_rowset_meta->tablet_schema();

        size_t inverted_index_size = 0;

        for (auto& seg_ptr : segments) {

            std::string index_path_prefix {

                    InvertedIndexDescriptor::get_index_file_path_prefix(local_segment_path(

                            _tablet->tablet_path(), output_rowset_meta->rowset_id().to_string(),

                            seg_ptr->id()))};

            std::vector<ColumnId> return_columns;

            std::vector<std::pair<int64_t, int64_t>> inverted_index_writer_signs;

            _olap_data_convertor->reserve(_alter_inverted_indexes.size());

            std::unique_ptr<IndexFileWriter> index_file_writer = nullptr;

            if (output_rowset_schema->get_inverted_index_storage_format() >=

                InvertedIndexStorageFormatPB::V2) {

                auto idx_file_reader_iter = _index_file_readers.find(

                        std::make_pair(output_rowset_meta->rowset_id().to_string(), seg_ptr->id()));

                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_can_not_find_reader",

                                { idx_file_reader_iter = _index_file_readers.end(); })

                if (idx_file_reader_iter == _index_file_readers.end()) {

                    LOG(ERROR) << "idx_file_reader_iter" << output_rowset_meta->rowset_id() << ":"

                               << seg_ptr->id() << " cannot be found";

                    continue;

                }

                std::string index_path =

                        InvertedIndexDescriptor::get_index_file_path_v2(index_path_prefix);

                io::FileWriterPtr file_writer;

                Status st = fs->create_file(index_path, &file_writer);

                if (!st.ok()) {

                    LOG(WARNING) << "failed to create writable file. path=" << index_path

                                 << ", err: " << st;

                    return st;

                }

                auto dirs = DORIS_TRY(idx_file_reader_iter->second->get_all_directories());

                index_file_writer = std::make_unique<IndexFileWriter>(

                        fs, index_path_prefix, output_rowset_meta->rowset_id().to_string(),

                        seg_ptr->id(), output_rowset_schema->get_inverted_index_storage_format(),

                        std::move(file_writer));

                RETURN_IF_ERROR(index_file_writer->initialize(dirs));

            } else {

                index_file_writer = std::make_unique<IndexFileWriter>(

                        fs, index_path_prefix, output_rowset_meta->rowset_id().to_string(),

                        seg_ptr->id(), output_rowset_schema->get_inverted_index_storage_format());

            }

            // create inverted index writer, or ann index writer

            for (auto inverted_index : _alter_inverted_indexes) {

                DCHECK(inverted_index.index_type == TIndexType::INVERTED ||

                       inverted_index.index_type == TIndexType::ANN);

                DCHECK_EQ(inverted_index.columns.size(), 1);

                auto index_id = inverted_index.index_id;

                auto column_name = inverted_index.columns[0];

                auto column_idx = output_rowset_schema->field_index(column_name);

                if (column_idx < 0) {

                    if (inverted_index.__isset.column_unique_ids &&

                        !inverted_index.column_unique_ids.empty()) {

                        column_idx = output_rowset_schema->field_index(

                                inverted_index.column_unique_ids[0]);

                    }

                    if (column_idx < 0) {

                        LOG(WARNING) << "referenced column was missing. "

                                     << "[column=" << column_name

                                     << " referenced_column=" << column_idx << "]";

                        continue;

                    }

                }

                auto column = output_rowset_schema->column(column_idx);

                // variant column is not support for building index

                auto is_support_inverted_index =

                        IndexColumnWriter::check_support_inverted_index(column);

                auto is_support_ann_index = IndexColumnWriter::check_support_ann_index(column);

                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_support_inverted_index",

                                { is_support_inverted_index = false; })

                if (!is_support_inverted_index && !is_support_ann_index) {

                    continue;

                }

                DCHECK(output_rowset_schema->has_inverted_index_with_index_id(index_id));

                _olap_data_convertor->add_column_data_convertor(column);

                return_columns.emplace_back(column_idx);

                std::unique_ptr<StorageField> field(StorageFieldFactory::create(column));

                if (inverted_index.index_type == TIndexType::INVERTED) {

                    // inverted index

                    auto index_metas = output_rowset_schema->inverted_indexs(column);

                    for (const auto& index_meta : index_metas) {

                        if (index_meta->index_id() != index_id) {

                            continue;

                        }

                        std::unique_ptr<segment_v2::IndexColumnWriter> inverted_index_builder;

                        try {

                            RETURN_IF_ERROR(segment_v2::IndexColumnWriter::create(

                                    field.get(), &inverted_index_builder, index_file_writer.get(),

                                    index_meta));

                            DBUG_EXECUTE_IF(

                                    "IndexBuilder::handle_single_rowset_index_column_writer_create_"

                                    "error",

                                    {

                                        _CLTHROWA(CL_ERR_IO,

                                                  "debug point: "

                                                  "handle_single_rowset_index_column_writer_create_"

                                                  "error");

                                    })

                        } catch (const std::exception& e) {

                            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                                    "CLuceneError occured: {}", e.what());

                        }

                        if (inverted_index_builder) {

                            auto writer_sign = std::make_pair(seg_ptr->id(), index_id);

                            _index_column_writers.insert(

                                    std::make_pair(writer_sign, std::move(inverted_index_builder)));

                            inverted_index_writer_signs.emplace_back(writer_sign);

                        }

                    }

                } else if (inverted_index.index_type == TIndexType::ANN) {

                    // ann index

                    const auto* index_meta = output_rowset_schema->ann_index(column);

                    if (index_meta && index_meta->index_id() == index_id) {

                        std::unique_ptr<segment_v2::IndexColumnWriter> index_writer;

                        try {

                            RETURN_IF_ERROR(segment_v2::IndexColumnWriter::create(

                                    field.get(), &index_writer, index_file_writer.get(),

                                    index_meta));

                            DBUG_EXECUTE_IF(

                                    "IndexBuilder::handle_single_rowset_index_column_writer_create_"

                                    "error",

                                    {

                                        _CLTHROWA(CL_ERR_IO,

                                                  "debug point: "

                                                  "handle_single_rowset_index_column_writer_create_"

                                                  "error");

                                    })

                        } catch (const std::exception& e) {

                            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                                    "CLuceneError occured: {}", e.what());

                        }

                        if (index_writer) {

                            auto writer_sign = std::make_pair(seg_ptr->id(), index_id);

                            _index_column_writers.insert(

                                    std::make_pair(writer_sign, std::move(index_writer)));

                            inverted_index_writer_signs.emplace_back(writer_sign);

                        }

                    }

                }

            }

            // DO NOT forget index_file_writer for the segment, otherwise, original inverted index will be deleted.

            _index_file_writers.emplace(seg_ptr->id(), std::move(index_file_writer));

            if (return_columns.empty()) {

                // no columns to read

                continue;

            }

            // create iterator for each segment

            StorageReadOptions read_options;

            OlapReaderStatistics stats;

            read_options.stats = &stats;

            read_options.tablet_schema = output_rowset_schema;

            std::shared_ptr<Schema> schema =

                    std::make_shared<Schema>(output_rowset_schema->columns(), return_columns);

            std::unique_ptr<RowwiseIterator> iter;

            auto res = seg_ptr->new_iterator(schema, read_options, &iter);

            DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_create_iterator_error", {

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

                        "debug point: handle_single_rowset_create_iterator_error");

            })

            if (!res.ok()) {

                LOG(WARNING) << "failed to create iterator[" << seg_ptr->id()

                             << "]: " << res.to_string();

                return Status::Error<ErrorCode::ROWSET_READER_INIT>(res.to_string());

            }

            auto block = Block::create_unique(output_rowset_schema->create_block(return_columns));

            while (true) {

                auto status = iter->next_batch(block.get());

                DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_iterator_next_batch_error", {

                    status = Status::Error<ErrorCode::SCHEMA_CHANGE_INFO_INVALID>(

                            "next_batch fault injection");

                });

                if (!status.ok()) {

                    if (status.is<ErrorCode::END_OF_FILE>()) {

                        break;

                    }

                    LOG(WARNING)

                            << "failed to read next block when schema change for inverted index."

                            << ", err=" << status.to_string();

                    return status;

                }

                // write inverted index data, or ann index data

                status = _write_inverted_index_data(output_rowset_schema, iter->data_id(),

                                                    block.get());

                DBUG_EXECUTE_IF(

                        "IndexBuilder::handle_single_rowset_write_inverted_index_data_error", {

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

                                    "debug point: "

                                    "handle_single_rowset_write_inverted_index_data_error");

                        })

                if (!status.ok()) {

                    return Status::Error<ErrorCode::SCHEMA_CHANGE_INFO_INVALID>(

                            "failed to write block.");

                }

                block->clear_column_data();

            }

            // finish write inverted index, flush data to compound file

            for (auto& writer_sign : inverted_index_writer_signs) {

                try {

                    if (_index_column_writers[writer_sign]) {

                        RETURN_IF_ERROR(_index_column_writers[writer_sign]->finish());

                    }

                    DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_index_build_finish_error", {

                        _CLTHROWA(CL_ERR_IO,

                                  "debug point: handle_single_rowset_index_build_finish_error");

                    })

                } catch (const std::exception& e) {

                    return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                            "CLuceneError occured: {}", e.what());

                }

            }

            _olap_data_convertor->reset();

        }

        for (auto&& [seg_id, index_file_writer] : _index_file_writers) {

            auto st = index_file_writer->begin_close();

            DBUG_EXECUTE_IF("IndexBuilder::handle_single_rowset_file_writer_close_error", {

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

                        "debug point: handle_single_rowset_file_writer_close_error");

            })

            if (!st.ok()) {

                LOG(ERROR) << "close index_file_writer error:" << st;

                return st;

            }

            inverted_index_size += index_file_writer->get_index_file_total_size();

        }

        for (auto&& [seg_id, index_file_writer] : _index_file_writers) {

            auto st = index_file_writer->finish_close();

            if (!st.ok()) {

                LOG(ERROR) << "wait close index_file_writer error:" << st;

                return st;

            }

        }

        _index_column_writers.clear();

        _index_file_writers.clear();

        output_rowset_meta->set_data_disk_size(output_rowset_meta->data_disk_size());

        output_rowset_meta->set_total_disk_size(output_rowset_meta->total_disk_size() +

                                                inverted_index_size);

        output_rowset_meta->set_index_disk_size(output_rowset_meta->index_disk_size() +

                                                inverted_index_size);

        LOG(INFO) << "all row nums. source_rows=" << output_rowset_meta->num_rows();

    }

    return Status::OK();

}

Status IndexBuilder::_write_inverted_index_data(TabletSchemaSPtr tablet_schema, int64_t segment_idx,

                                                Block* block) {

    VLOG_DEBUG << "begin to write inverted/ann index";

    // converter block data

    _olap_data_convertor->set_source_content(block, 0, block->rows());

    for (auto i = 0; i < _alter_inverted_indexes.size(); ++i) {

        auto inverted_index = _alter_inverted_indexes[i];

        auto index_id = inverted_index.index_id;

        auto column_name = inverted_index.columns[0];

        auto column_idx = tablet_schema->field_index(column_name);

        DBUG_EXECUTE_IF("IndexBuilder::_write_inverted_index_data_column_idx_is_negative",

                        { column_idx = -1; })

        if (column_idx < 0) {

            if (!inverted_index.column_unique_ids.empty()) {

                auto column_unique_id = inverted_index.column_unique_ids[0];

                column_idx = tablet_schema->field_index(column_unique_id);

            }

            if (column_idx < 0) {

                LOG(WARNING) << "referenced column was missing. "

                             << "[column=" << column_name << " referenced_column=" << column_idx

                             << "]";

                continue;

            }

        }

        auto column = tablet_schema->column(column_idx);

        auto writer_sign = std::make_pair(segment_idx, index_id);

        std::unique_ptr<StorageField> field(StorageFieldFactory::create(column));

        auto converted_result = _olap_data_convertor->convert_column_data(i);

        DBUG_EXECUTE_IF("IndexBuilder::_write_inverted_index_data_convert_column_data_error", {

            converted_result.first = Status::Error<ErrorCode::INTERNAL_ERROR>(

                    "debug point: _write_inverted_index_data_convert_column_data_error");

        })

        if (converted_result.first != Status::OK()) {

            LOG(WARNING) << "failed to convert block, errcode: " << converted_result.first;

            return converted_result.first;

        }

        const auto* ptr = (const uint8_t*)converted_result.second->get_data();

        const auto* null_map = converted_result.second->get_nullmap();

        if (null_map) {

            RETURN_IF_ERROR(_add_nullable(column_name, writer_sign, field.get(), null_map, &ptr,

                                          block->rows()));

        } else {

            RETURN_IF_ERROR(_add_data(column_name, writer_sign, field.get(), &ptr, block->rows()));

        }

    }

    _olap_data_convertor->clear_source_content();

    return Status::OK();

}

Status IndexBuilder::_add_nullable(const std::string& column_name,

                                   const std::pair<int64_t, int64_t>& index_writer_sign,

                                   StorageField* field, const uint8_t* null_map,

                                   const uint8_t** ptr, size_t num_rows) {

    // TODO: need to process null data for inverted index

    if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {

        DCHECK(field->get_sub_field_count() == 1);

        // [size, offset_ptr, item_data_ptr, item_nullmap_ptr]

        const auto* data_ptr = reinterpret_cast<const uint64_t*>(*ptr);

        // total number length

        auto offset_data = *(data_ptr + 1);

        const auto* offsets_ptr = (const uint8_t*)offset_data;

        try {

            auto data = *(data_ptr + 2);

            auto nested_null_map = *(data_ptr + 3);

            RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_array_values(

                    field->get_sub_field(0)->size(), reinterpret_cast<const void*>(data),

                    reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows));

            DBUG_EXECUTE_IF("IndexBuilder::_add_nullable_add_array_values_error", {

                _CLTHROWA(CL_ERR_IO, "debug point: _add_nullable_add_array_values_error");

            })

            RETURN_IF_ERROR(

                    _index_column_writers[index_writer_sign]->add_array_nulls(null_map, num_rows));

        } catch (const std::exception& e) {

            return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>(

                    "CLuceneError occured: {}", e.what());

        }

        return Status::OK();

    }

    size_t offset = 0;

    auto next_run_step = [&]() {

        size_t step = 1;

        for (auto i = offset + 1; i < num_rows; ++i) {

            if (null_map[offset] == null_map[i]) {

                step++;

            } else {

                break;

            }

        }

        return step;

    };

    try {

        do {

            auto step = next_run_step();

            if (null_map[offset]) {

                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_nulls(

                        static_cast<uint32_t>(step)));

            } else {

                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_values(column_name,

                                                                                     *ptr, step));

            }

            *ptr += field->size() * step;

            offset += step;

            DBUG_EXECUTE_IF("IndexBuilder::_add_nullable_throw_exception",

                            { _CLTHROWA(CL_ERR_IO, "debug point: _add_nullable_throw_exception"); })

        } while (offset < num_rows);

    } catch (const std::exception& e) {

        return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>("CLuceneError occured: {}",

                                                                      e.what());

    }

    return Status::OK();

}

Status IndexBuilder::_add_data(const std::string& column_name,

                               const std::pair<int64_t, int64_t>& index_writer_sign,

                               StorageField* field, const uint8_t** ptr, size_t num_rows) {

    try {

        if (field->type() == FieldType::OLAP_FIELD_TYPE_ARRAY) {

            DCHECK(field->get_sub_field_count() == 1);

            // [size, offset_ptr, item_data_ptr, item_nullmap_ptr]

            const auto* data_ptr = reinterpret_cast<const uint64_t*>(*ptr);

            // total number length

            auto element_cnt = size_t((unsigned long)(*data_ptr));

            auto offset_data = *(data_ptr + 1);

            const auto* offsets_ptr = (const uint8_t*)offset_data;

            if (element_cnt > 0) {

                auto data = *(data_ptr + 2);

                auto nested_null_map = *(data_ptr + 3);

                RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_array_values(

                        field->get_sub_field(0)->size(), reinterpret_cast<const void*>(data),

                        reinterpret_cast<const uint8_t*>(nested_null_map), offsets_ptr, num_rows));

            }

        } else {

            RETURN_IF_ERROR(_index_column_writers[index_writer_sign]->add_values(column_name, *ptr,

                                                                                 num_rows));

        }

        DBUG_EXECUTE_IF("IndexBuilder::_add_data_throw_exception",

                        { _CLTHROWA(CL_ERR_IO, "debug point: _add_data_throw_exception"); })

    } catch (const std::exception& e) {

        return Status::Error<ErrorCode::INVERTED_INDEX_CLUCENE_ERROR>("CLuceneError occured: {}",

                                                                      e.what());

    }

    return Status::OK();

}

Status IndexBuilder::handle_inverted_index_data() {

    LOG(INFO) << "begin to handle_inverted_index_data";

    DCHECK(_input_rowsets.size() == _output_rowsets.size());

    for (auto& _output_rowset : _output_rowsets) {

        SegmentCacheHandle segment_cache_handle;

        RETURN_IF_ERROR(SegmentLoader::instance()->load_segments(

                std::static_pointer_cast<BetaRowset>(_output_rowset), &segment_cache_handle));

        auto output_rowset_meta = _output_rowset->rowset_meta();

        auto& segments = segment_cache_handle.get_segments();

        RETURN_IF_ERROR(handle_single_rowset(output_rowset_meta, segments));

    }

    return Status::OK();

}

Status IndexBuilder::do_build_inverted_index() {

    LOG(INFO) << "begin to do_build_inverted_index, tablet=" << _tablet->tablet_id()

              << ", is_drop_op=" << _is_drop_op;

    DBUG_EXECUTE_IF("IndexBuilder::do_build_inverted_index_alter_inverted_indexes_empty",

                    { _alter_inverted_indexes.clear(); })

    if (_alter_inverted_indexes.empty()) {

        return Status::OK();

    }

    static constexpr long TRY_LOCK_TIMEOUT = 30;

    std::unique_lock schema_change_lock(_tablet->get_schema_change_lock(), std::defer_lock);

    bool owns_lock = schema_change_lock.try_lock_for(std::chrono::seconds(TRY_LOCK_TIMEOUT));

    if (!owns_lock) {

        return Status::ObtainLockFailed(

                "try schema_change_lock failed. There might be schema change or cooldown running "

                "on "

                "tablet={} ",

                _tablet->tablet_id());

    }

    // Check executing serially with compaction task.

    std::unique_lock<std::mutex> base_compaction_lock(_tablet->get_base_compaction_lock(),

                                                      std::try_to_lock);

    if (!base_compaction_lock.owns_lock()) {

        return Status::ObtainLockFailed("try base_compaction_lock failed. tablet={} ",

                                        _tablet->tablet_id());

    }

    std::unique_lock<std::mutex> cumu_compaction_lock(_tablet->get_cumulative_compaction_lock(),

                                                      std::try_to_lock);

    if (!cumu_compaction_lock.owns_lock()) {

        return Status::ObtainLockFailed("try cumu_compaction_lock failed. tablet={}",

                                        _tablet->tablet_id());

    }

    std::unique_lock<std::mutex> cold_compaction_lock(_tablet->get_cold_compaction_lock(),

                                                      std::try_to_lock);

    if (!cold_compaction_lock.owns_lock()) {

        return Status::ObtainLockFailed("try cold_compaction_lock failed. tablet={}",

                                        _tablet->tablet_id());

    }

    std::unique_lock<std::mutex> build_inverted_index_lock(_tablet->get_build_inverted_index_lock(),

                                                           std::try_to_lock);

    if (!build_inverted_index_lock.owns_lock()) {

        return Status::ObtainLockFailed("failed to obtain build inverted index lock. tablet={}",

                                        _tablet->tablet_id());

    }

    std::shared_lock migration_rlock(_tablet->get_migration_lock(), std::try_to_lock);

    if (!migration_rlock.owns_lock()) {

        return Status::ObtainLockFailed("got migration_rlock failed. tablet={}",

                                        _tablet->tablet_id());

    }

    DCHECK(!_alter_index_ids.empty());

    _input_rowsets =

            _tablet->pick_candidate_rowsets_to_build_inverted_index(_alter_index_ids, _is_drop_op);

    if (_input_rowsets.empty()) {

        LOG(INFO) << "_input_rowsets is empty";

        return Status::OK();

    }

    auto st = update_inverted_index_info();

    if (!st.ok()) {

        LOG(WARNING) << "failed to update_inverted_index_info. "

                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;

        gc_output_rowset();

        return st;

    }

    // create inverted index file for output rowset

    st = handle_inverted_index_data();

    if (!st.ok()) {

        LOG(WARNING) << "failed to handle_inverted_index_data. "

                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;

        gc_output_rowset();

        return st;

    }

    // modify rowsets in memory

    st = modify_rowsets();

    DBUG_EXECUTE_IF("IndexBuilder::do_build_inverted_index_modify_rowsets_status_error", {

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

                "debug point: do_build_inverted_index_modify_rowsets_status_error");

    })

    if (!st.ok()) {

        LOG(WARNING) << "failed to modify rowsets in memory. "

                     << "tablet=" << _tablet->tablet_id() << ", error=" << st;

        gc_output_rowset();

        return st;

    }

    return Status::OK();

}

Status IndexBuilder::modify_rowsets(const Merger::Statistics* stats) {

    DCHECK(std::ranges::all_of(

            _output_rowsets.begin(), _output_rowsets.end(), [&engine = _engine](auto&& rs) {

                if (engine.check_rowset_id_in_unused_rowsets(rs->rowset_id())) {

                    LOG(ERROR) << "output rowset: " << rs->rowset_id() << " in unused rowsets";

                    return false;

                }

                return true;

            }));

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

        _tablet->enable_unique_key_merge_on_write()) {

        std::lock_guard<std::mutex> rowset_update_wlock(_tablet->get_rowset_update_lock());

        std::lock_guard<std::shared_mutex> meta_wlock(_tablet->get_header_lock());

        SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);

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

        for (auto i = 0; i < _input_rowsets.size(); ++i) {

            RowsetId input_rowset_id = _input_rowsets[i]->rowset_id();

            RowsetId output_rowset_id = _output_rowsets[i]->rowset_id();

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

                RowsetId rs_id = std::get<0>(k);

                if (rs_id == input_rowset_id) {

                    DeleteBitmap::BitmapKey output_rs_key = {output_rowset_id, std::get<1>(k),

                                                             std::get<2>(k)};

                    auto res = delete_bitmap->set(output_rs_key, v);

                    DCHECK(res > 0) << "delete_bitmap set failed, res=" << res;

                }

            }

        }

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

        // modify_rowsets will remove the delete_bitmap for input rowsets,

        // should call it after merge delete_bitmap

        RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets, _input_rowsets, true));

    } else {

        std::lock_guard<std::shared_mutex> wrlock(_tablet->get_header_lock());

        RETURN_IF_ERROR(_tablet->modify_rowsets(_output_rowsets, _input_rowsets, true));

    }

#ifndef BE_TEST

    {

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

        _tablet->save_meta();

    }

#endif

    return Status::OK();

}

void IndexBuilder::gc_output_rowset() {

    for (auto&& output_rowset : _output_rowsets) {

        auto is_local_rowset = output_rowset->is_local();

        DBUG_EXECUTE_IF("IndexBuilder::gc_output_rowset_is_local_rowset",

                        { is_local_rowset = false; })