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

#include <butil/logging.h>

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/Exprs_types.h>

#include <gen_cpp/Partitions_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/descriptors.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <glog/logging.h>

#include <algorithm>

#include <cstddef>

#include <cstdint>

#include <memory>

#include <ostream>

#include <string>

#include <tuple>

#include "common/exception.h"

#include "common/logging.h"

#include "common/status.h"

#include "core/column/column.h"

#include "core/data_type/data_type.h"

#include "core/data_type/data_type_factory.hpp"

#include "core/data_type/define_primitive_type.h"

#include "core/data_type/primitive_type.h"

#include "core/value/large_int_value.h"

#include "runtime/descriptors.h"

#include "runtime/memory/mem_tracker.h"

#include "storage/tablet/tablet_schema.h"

#include "util/raw_value.h"

#include "util/string_parser.hpp"

#include "util/string_util.h"

// NOLINTNEXTLINE(unused-includes)

#include "core/value/vdatetime_value.h"

#include "exprs/function/cast/cast_to_date_or_datetime_impl.hpp"

#include "exprs/function/cast/cast_to_datetimev2_impl.hpp"

#include "exprs/function/cast/cast_to_datev2_impl.hpp"

#include "exprs/function/cast/cast_to_timestamptz.h"

#include "exprs/vexpr_context.h" // IWYU pragma: keep

#include "exprs/vliteral.h"

namespace doris {

void OlapTableIndexSchema::to_protobuf(POlapTableIndexSchema* pindex) const {

    pindex->set_id(index_id);

    pindex->set_schema_hash(schema_hash);

    if (row_binlog_id > 0) {

        pindex->set_row_binlog_id(row_binlog_id);

    }

    for (auto* slot : slots) {

        pindex->add_columns(slot->col_name());

    }

    for (auto* column : columns) {

        column->to_schema_pb(pindex->add_columns_desc());

    }

    for (auto* index : indexes) {

        index->to_schema_pb(pindex->add_indexes_desc());

    }

}

bool VOlapTablePartKeyComparator::operator()(const BlockRowWithIndicator& lhs,

                                             const BlockRowWithIndicator& rhs) const {

    Block* l_block = std::get<0>(lhs);

    Block* r_block = std::get<0>(rhs);

    int32_t l_row = std::get<1>(lhs);

    int32_t r_row = std::get<1>(rhs);

    bool l_use_new = std::get<2>(lhs);

    bool r_use_new = std::get<2>(rhs);

    VLOG_TRACE << '\n' << l_block->dump_data() << '\n' << r_block->dump_data();

    if (l_row == -1) {

        return false;

    } else if (r_row == -1) {

        return true;

    }

    if (_param_locs.empty()) { // no transform, use origin column

        for (auto slot_loc : _slot_locs) {

            auto res = l_block->get_by_position(slot_loc).column->compare_at(

                    l_row, r_row, *r_block->get_by_position(slot_loc).column, -1);

            if (res != 0) {

                return res < 0;

            }

        }

    } else { // use transformed column to compare

        DCHECK(_slot_locs.size() == _param_locs.size())

                << _slot_locs.size() << ' ' << _param_locs.size();

        const std::vector<uint16_t>* l_index = l_use_new ? &_param_locs : &_slot_locs;

        const std::vector<uint16_t>* r_index = r_use_new ? &_param_locs : &_slot_locs;

        for (int i = 0; i < _slot_locs.size(); i++) {

            ColumnPtr l_col = l_block->get_by_position((*l_index)[i]).column;

            ColumnPtr r_col = r_block->get_by_position((*r_index)[i]).column;

            auto res = l_col->compare_at(l_row, r_row, *r_col, -1);

            if (res != 0) {

                return res < 0;

            }

        }

    }

    // equal, return false

    return false;

}

Status OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema) {

    _db_id = pschema.db_id();

    _table_id = pschema.table_id();

    _version = pschema.version();

    if (pschema.has_unique_key_update_mode()) {

        _unique_key_update_mode = pschema.unique_key_update_mode();

        if (pschema.has_sequence_map_col_unique_id()) {

            _sequence_map_col_uid = pschema.sequence_map_col_unique_id();

        }

    } else {

        // for backward compatibility

        if (pschema.has_partial_update() && pschema.partial_update()) {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;

        } else {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;

        }

    }

    _is_strict_mode = pschema.is_strict_mode();

    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {

        _auto_increment_column = pschema.auto_increment_column();

        if (!_auto_increment_column.empty() && pschema.auto_increment_column_unique_id() == -1) {

            return Status::InternalError(

                    "Auto increment column id is not set in FE. Maybe FE is an older version "

                    "different from BE.");

        }

        _auto_increment_column_unique_id = pschema.auto_increment_column_unique_id();

    }

    if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT) {

        if (pschema.has_partial_update_new_key_policy()) {

            _partial_update_new_row_policy = pschema.partial_update_new_key_policy();

        }

    }

    _timestamp_ms = pschema.timestamp_ms();

    if (pschema.has_nano_seconds()) {

        _nano_seconds = pschema.nano_seconds();

    }

    _timezone = pschema.timezone();

    for (const auto& col : pschema.partial_update_input_columns()) {

        _partial_update_input_columns.insert(col);

    }

    std::unordered_map<std::string, SlotDescriptor*> slots_map;

    _tuple_desc = _obj_pool.add(new TupleDescriptor(pschema.tuple_desc()));

    for (const auto& p_slot_desc : pschema.slot_descs()) {

        auto* slot_desc = _obj_pool.add(new SlotDescriptor(p_slot_desc));

        _tuple_desc->add_slot(slot_desc);

        std::string data_type;

        EnumToString(TPrimitiveType, to_thrift(slot_desc->col_type()), data_type);

        std::string is_null_str = slot_desc->is_nullable() ? "true" : "false";

        std::string data_type_str =

                std::to_string(int64_t(TabletColumn::get_field_type_by_string(data_type)));

        slots_map.emplace(to_lower(slot_desc->col_name()) + "+" + data_type_str + is_null_str,

                          slot_desc);

    }

    for (const auto& p_index : pschema.indexes()) {

        auto* index = _obj_pool.add(new OlapTableIndexSchema());

        index->index_id = p_index.id();

        index->schema_hash = p_index.schema_hash();

        if (p_index.has_row_binlog_id()) {

            index->row_binlog_id = p_index.row_binlog_id();

        }

        for (const auto& pcolumn_desc : p_index.columns_desc()) {

            if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS ||

                _partial_update_input_columns.contains(pcolumn_desc.name())) {

                std::string is_null_str = pcolumn_desc.is_nullable() ? "true" : "false";

                std::string data_type_str = std::to_string(

                        int64_t(TabletColumn::get_field_type_by_string(pcolumn_desc.type())));

                auto it = slots_map.find(to_lower(pcolumn_desc.name()) + "+" + data_type_str +

                                         is_null_str);

                if (it == std::end(slots_map)) {

                    std::string keys {};

                    for (const auto& [key, _] : slots_map) {

                        keys += fmt::format("{},", key);

                    }

                    LOG_EVERY_SECOND(WARNING) << fmt::format(

                            "[OlapTableSchemaParam::init(const POlapTableSchemaParam& pschema)]: "

                            "unknown index column, column={}, type={}, data_type_str={}, "

                            "is_null_str={}, slots_map.keys()=[{}], {}\npschema={}",

                            pcolumn_desc.name(), pcolumn_desc.type(), data_type_str, is_null_str,

                            keys, debug_string(), pschema.ShortDebugString());

                    return Status::InternalError("unknown index column, column={}, type={}",

                                                 pcolumn_desc.name(), pcolumn_desc.type());

                }

                index->slots.emplace_back(it->second);

            }

            TabletColumn* tc = _obj_pool.add(new TabletColumn());

            tc->init_from_pb(pcolumn_desc);

            index->columns.emplace_back(tc);

        }

        for (const auto& pindex_desc : p_index.indexes_desc()) {

            TabletIndex* ti = _obj_pool.add(new TabletIndex());

            ti->init_from_pb(pindex_desc);

            index->indexes.emplace_back(ti);

        }

        _indexes.emplace_back(index);

    }

    if (pschema.has_row_binlog_index_schema()) {

        const auto& p_index = pschema.row_binlog_index_schema();

        auto* index = _obj_pool.add(new OlapTableIndexSchema());

        index->index_id = p_index.id();

        index->schema_hash = p_index.schema_hash();

        if (p_index.has_row_binlog_id()) {

            index->row_binlog_id = p_index.row_binlog_id();

        }

        for (const auto& pcolumn_desc : p_index.columns_desc()) {

            TabletColumn* tc = _obj_pool.add(new TabletColumn());

            tc->init_from_pb(pcolumn_desc);

            index->columns.emplace_back(tc);

        }

        for (const auto& pindex_desc : p_index.indexes_desc()) {

            TabletIndex* ti = _obj_pool.add(new TabletIndex());

            ti->init_from_pb(pindex_desc);

            index->indexes.emplace_back(ti);

        }

        _row_binlog_index_schema = index;

    }

    std::sort(_indexes.begin(), _indexes.end(),

              [](const OlapTableIndexSchema* lhs, const OlapTableIndexSchema* rhs) {

                  return lhs->index_id < rhs->index_id;

              });

    return Status::OK();

}

Status OlapTableSchemaParam::init_unique_key_update_mode(const TOlapTableSchemaParam& tschema) {

    if (tschema.__isset.unique_key_update_mode) {

        switch (tschema.unique_key_update_mode) {

        case doris::TUniqueKeyUpdateMode::UPSERT: {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;

            break;

        }

        case doris::TUniqueKeyUpdateMode::UPDATE_FIXED_COLUMNS: {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;

            break;

        }

        case doris::TUniqueKeyUpdateMode::UPDATE_FLEXIBLE_COLUMNS: {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FLEXIBLE_COLUMNS;

            break;

        }

        default: {

            return Status::InternalError(

                    "Unknown unique_key_update_mode: {}, should be one of "

                    "UPSERT/UPDATE_FIXED_COLUMNS/UPDATE_FLEXIBLE_COLUMNS",

                    tschema.unique_key_update_mode);

        }

        }

        if (tschema.__isset.sequence_map_col_unique_id) {

            _sequence_map_col_uid = tschema.sequence_map_col_unique_id;

        }

    } else {

        // for backward compatibility

        if (tschema.__isset.is_partial_update && tschema.is_partial_update) {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS;

        } else {

            _unique_key_update_mode = UniqueKeyUpdateModePB::UPSERT;

        }

    }

    return Status::OK();

}

Status OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema) {

    _db_id = tschema.db_id;

    _table_id = tschema.table_id;

    _version = tschema.version;

    RETURN_IF_ERROR(init_unique_key_update_mode(tschema));

    if (tschema.__isset.is_strict_mode) {

        _is_strict_mode = tschema.is_strict_mode;

    }

    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {

        _auto_increment_column = tschema.auto_increment_column;

        if (!_auto_increment_column.empty() && tschema.auto_increment_column_unique_id == -1) {

            return Status::InternalError(

                    "Auto increment column id is not set in FE. Maybe FE is an older version "

                    "different from BE.");

        }

        _auto_increment_column_unique_id = tschema.auto_increment_column_unique_id;

    }

    if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPSERT) {

        if (tschema.__isset.partial_update_new_key_policy) {

            switch (tschema.partial_update_new_key_policy) {

            case doris::TPartialUpdateNewRowPolicy::APPEND: {

                _partial_update_new_row_policy = PartialUpdateNewRowPolicyPB::APPEND;

                break;

            }

            case doris::TPartialUpdateNewRowPolicy::ERROR: {

                _partial_update_new_row_policy = PartialUpdateNewRowPolicyPB::ERROR;

                break;

            }

            default: {

                return Status::InvalidArgument(

                        "Unknown partial_update_new_key_behavior: {}, should be one of "

                        "'APPEND' or 'ERROR'",

                        tschema.partial_update_new_key_policy);

            }

            }

        }

    }

    for (const auto& tcolumn : tschema.partial_update_input_columns) {

        _partial_update_input_columns.insert(tcolumn);

    }

    std::unordered_map<std::string, SlotDescriptor*> slots_map;

    _tuple_desc = _obj_pool.add(new TupleDescriptor(tschema.tuple_desc));

    for (const auto& t_slot_desc : tschema.slot_descs) {

        auto* slot_desc = _obj_pool.add(new SlotDescriptor(t_slot_desc));

        _tuple_desc->add_slot(slot_desc);

        std::string is_null_str = slot_desc->is_nullable() ? "true" : "false";

        std::string data_type_str = std::to_string(int64_t(slot_desc->col_type()));

        slots_map.emplace(to_lower(slot_desc->col_name()) + "+" + data_type_str + is_null_str,

                          slot_desc);

    }

    for (const auto& t_index : tschema.indexes) {

        std::unordered_map<std::string, int32_t> index_slots_map;

        auto* index = _obj_pool.add(new OlapTableIndexSchema());

        index->index_id = t_index.id;

        index->schema_hash = t_index.schema_hash;

        if (t_index.__isset.row_binlog_id) {

            index->row_binlog_id = t_index.row_binlog_id;

        }

        for (const auto& tcolumn_desc : t_index.columns_desc) {

            if (_unique_key_update_mode != UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS ||

                _partial_update_input_columns.contains(tcolumn_desc.column_name)) {

                std::string is_null_str = tcolumn_desc.is_allow_null ? "true" : "false";

                std::string data_type_str =

                        std::to_string(int64_t(thrift_to_type(tcolumn_desc.column_type.type)));

                auto it = slots_map.find(to_lower(tcolumn_desc.column_name) + "+" + data_type_str +

                                         is_null_str);

                if (it == slots_map.end()) {

                    std::stringstream ss;

                    ss << tschema;

                    std::string keys {};

                    for (const auto& [key, _] : slots_map) {

                        keys += fmt::format("{},", key);

                    }

                    LOG_EVERY_SECOND(WARNING) << fmt::format(

                            "[OlapTableSchemaParam::init(const TOlapTableSchemaParam& tschema)]: "

                            "unknown index column, column={}, type={}, data_type_str={}, "

                            "is_null_str={}, slots_map.keys()=[{}], {}\ntschema={}",

                            tcolumn_desc.column_name, tcolumn_desc.column_type.type, data_type_str,

                            is_null_str, keys, debug_string(), ss.str());

                    return Status::InternalError("unknown index column, column={}, type={}",

                                                 tcolumn_desc.column_name,

                                                 tcolumn_desc.column_type.type);

                }

                index->slots.emplace_back(it->second);

            }

            index_slots_map.emplace(to_lower(tcolumn_desc.column_name), tcolumn_desc.col_unique_id);

            TabletColumn* tc = _obj_pool.add(new TabletColumn());

            tc->init_from_thrift(tcolumn_desc);

            index->columns.emplace_back(tc);

        }

        if (t_index.__isset.indexes_desc) {

            for (const auto& tindex_desc : t_index.indexes_desc) {

                std::vector<int32_t> column_unique_ids(tindex_desc.columns.size());

                for (size_t i = 0; i < tindex_desc.columns.size(); i++) {

                    auto it = index_slots_map.find(to_lower(tindex_desc.columns[i]));

                    if (it != index_slots_map.end()) {

                        column_unique_ids[i] = it->second;

                    }

                }

                TabletIndex* ti = _obj_pool.add(new TabletIndex());

                ti->init_from_thrift(tindex_desc, column_unique_ids);

                index->indexes.emplace_back(ti);

            }

        }

        if (t_index.__isset.where_clause) {

            RETURN_IF_ERROR(VExpr::create_expr_tree(t_index.where_clause, index->where_clause));

        }

        _indexes.emplace_back(index);

    }

    if (tschema.__isset.row_binlog_index_schema) {

        const auto& t_index = tschema.row_binlog_index_schema;

        auto* index = _obj_pool.add(new OlapTableIndexSchema());

        index->index_id = t_index.id;

        index->schema_hash = t_index.schema_hash;

        if (t_index.__isset.row_binlog_id) {

            index->row_binlog_id = t_index.row_binlog_id;

        }

        for (const auto& tcolumn_desc : t_index.columns_desc) {

            TabletColumn* tc = _obj_pool.add(new TabletColumn());

            tc->init_from_thrift(tcolumn_desc);

            index->columns.emplace_back(tc);

        }

        _row_binlog_index_schema = index;

    }

    std::sort(_indexes.begin(), _indexes.end(),

              [](const OlapTableIndexSchema* lhs, const OlapTableIndexSchema* rhs) {

                  return lhs->index_id < rhs->index_id;

              });

    return Status::OK();

}

void OlapTableSchemaParam::to_protobuf(POlapTableSchemaParam* pschema) const {

    pschema->set_db_id(_db_id);

    pschema->set_table_id(_table_id);

    pschema->set_version(_version);

    pschema->set_unique_key_update_mode(_unique_key_update_mode);

    if (_unique_key_update_mode == UniqueKeyUpdateModePB::UPDATE_FIXED_COLUMNS) {

        // for backward compatibility

        pschema->set_partial_update(true);

    }

    pschema->set_partial_update_new_key_policy(_partial_update_new_row_policy);

    pschema->set_is_strict_mode(_is_strict_mode);

    pschema->set_auto_increment_column(_auto_increment_column);

    pschema->set_auto_increment_column_unique_id(_auto_increment_column_unique_id);

    pschema->set_timestamp_ms(_timestamp_ms);

    pschema->set_timezone(_timezone);

    pschema->set_nano_seconds(_nano_seconds);

    pschema->set_sequence_map_col_unique_id(_sequence_map_col_uid);

    for (auto col : _partial_update_input_columns) {

        *pschema->add_partial_update_input_columns() = col;

    }

    _tuple_desc->to_protobuf(pschema->mutable_tuple_desc());

    for (auto* slot : _tuple_desc->slots()) {

        slot->to_protobuf(pschema->add_slot_descs());

    }

    for (auto* index : _indexes) {

        index->to_protobuf(pschema->add_indexes());

    }

    if (_row_binlog_index_schema != nullptr) {

        _row_binlog_index_schema->to_protobuf(pschema->mutable_row_binlog_index_schema());

    }

}

std::string OlapTableSchemaParam::debug_string() const {

    std::stringstream ss;

    ss << "tuple_desc=" << _tuple_desc->debug_string();

    return ss.str();

}

VOlapTablePartitionParam::VOlapTablePartitionParam(std::shared_ptr<OlapTableSchemaParam>& schema,

                                                   const TOlapTablePartitionParam& t_param)

        : _schema(schema),

          _t_param(t_param),

          _slots(_schema->tuple_desc()->slots()),

          _mem_tracker(std::make_unique<MemTracker>("OlapTablePartitionParam")),

          _part_type(t_param.partition_type) {

    if (t_param.__isset.enable_automatic_partition && t_param.enable_automatic_partition) {

        _is_auto_partition = true;

        auto size = t_param.partition_function_exprs.size();

        _part_func_ctx.resize(size);

        _partition_function.resize(size);

        DCHECK((t_param.partition_type == TPartitionType::RANGE_PARTITIONED && size == 1) ||

               (t_param.partition_type == TPartitionType::LIST_PARTITIONED && size >= 1))

                << "now support only 1 partition column for auto range partitions. "

                << t_param.partition_type << " " << size;

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

            Status st =

                    VExpr::create_expr_tree(t_param.partition_function_exprs[i], _part_func_ctx[i]);

            if (!st.ok()) {

                throw Exception(Status::InternalError("Partition function expr is not valid"),

                                "Partition function expr is not valid");

            }

            _partition_function[i] = _part_func_ctx[i]->root();

        }

    }

    if (t_param.__isset.enable_auto_detect_overwrite && t_param.enable_auto_detect_overwrite) {

        _is_auto_detect_overwrite = true;

        DCHECK(t_param.__isset.overwrite_group_id);

        _overwrite_group_id = t_param.overwrite_group_id;

    }

    if (t_param.__isset.master_address) {

        _master_address = std::make_shared<TNetworkAddress>(t_param.master_address);

    }

    if (_is_auto_partition) {

        // the nullable mode depends on partition_exprs. not column slots. so use them.

        DCHECK(_partition_function.size() <= _slots.size())

                << _partition_function.size() << ", " << _slots.size();

        // suppose (k0, [k1], [k2]), so get [k1, 0], [k2, 1]

        std::map<std::string, int> partition_slots_map; // name to idx in part_exprs

        for (size_t i = 0; i < t_param.partition_columns.size(); i++) {

            partition_slots_map.emplace(t_param.partition_columns[i], i);

        }

        // here we rely on the same order and number of the _part_funcs and _slots in the prefix

        // _part_block contains all slots of table.

        for (auto* slot : _slots) {

            // try to replace with partition expr.

            if (auto it = partition_slots_map.find(slot->col_name());

                it != partition_slots_map.end()) { // it's a partition column slot

                auto& expr_type = _partition_function[it->second]->data_type();

                _partition_block.insert({expr_type->create_column(), expr_type, slot->col_name()});

            } else {

                _partition_block.insert({slot->get_empty_mutable_column(),

                                         slot->get_data_type_ptr(), slot->col_name()});

            }

        }

        VLOG_TRACE << _partition_block.dump_structure();

    } else {

        // we insert all. but not all will be used. it will controlled by _partition_slot_locs

        for (auto* slot : _slots) {

            _partition_block.insert({slot->get_empty_mutable_column(), slot->get_data_type_ptr(),

                                     slot->col_name()});

        }

    }

}

VOlapTablePartitionParam::~VOlapTablePartitionParam() {

    _mem_tracker->release(_mem_usage);

}

Status VOlapTablePartitionParam::init() {

    std::vector<std::string> slot_column_names;

    for (auto* slot_desc : _schema->tuple_desc()->slots()) {

        slot_column_names.emplace_back(slot_desc->col_name());

    }

    auto find_slot_locs = [&slot_column_names](const std::string& slot_name,

                                               std::vector<uint16_t>& locs,

                                               const std::string& column_type) {

        auto it = std::find(slot_column_names.begin(), slot_column_names.end(), slot_name);

        if (it == slot_column_names.end()) {

            return Status::InternalError("{} column not found, column ={}", column_type, slot_name);

        }

        locs.emplace_back(it - slot_column_names.begin());

        return Status::OK();

    };

    // here we find the partition columns. others maybe non-partition columns/special columns.

    if (_t_param.__isset.partition_columns) {

        for (auto& part_col : _t_param.partition_columns) {

            RETURN_IF_ERROR(find_slot_locs(part_col, _partition_slot_locs, "partition"));

        }

    }

    _partitions_map = std::make_unique<

            std::map<BlockRowWithIndicator, VOlapTablePartition*, VOlapTablePartKeyComparator>>(

            VOlapTablePartKeyComparator(_partition_slot_locs, _transformed_slot_locs));

    if (_t_param.__isset.distributed_columns) {

        for (auto& col : _t_param.distributed_columns) {

            RETURN_IF_ERROR(find_slot_locs(col, _distributed_slot_locs, "distributed"));

        }

    }

    // for both auto/non-auto partition table.

    _is_in_partition = _part_type == TPartitionType::type::LIST_PARTITIONED;

    // initial partitions. if meet dummy partitions only for open BE nodes, not generate key of them for finding

    for (const auto& t_part : _t_param.partitions) {

        VOlapTablePartition* part = nullptr;

        RETURN_IF_ERROR(generate_partition_from(t_part, part));

        _partitions.emplace_back(part);

        if (!_t_param.partitions_is_fake) {

            if (_is_in_partition) {

                for (auto& in_key : part->in_keys) {

                    _partitions_map->emplace(std::tuple {in_key.first, in_key.second, false}, part);

                }

            } else {

                _partitions_map->emplace(

                        std::tuple {part->end_key.first, part->end_key.second, false}, part);

            }

        }

    }

    _mem_usage = _partition_block.allocated_bytes();

    _mem_tracker->consume(_mem_usage);

    return Status::OK();

}

bool VOlapTablePartitionParam::_part_contains(VOlapTablePartition* part,

                                              BlockRowWithIndicator key) const {

    VOlapTablePartKeyComparator comparator(_partition_slot_locs, _transformed_slot_locs);

    // we have used upper_bound to find to ensure key < part.right and this part is closest(right - key is min)

    // now we only have to check (key >= part.left). the comparator(a,b) means a < b, so we use anti

    return part->start_key.second == -1 /* spj: start_key.second == -1 means only single partition*/

           || !comparator(key, std::tuple {part->start_key.first, part->start_key.second, false});

}

// insert value into _partition_block's column

// NOLINTBEGIN(readability-function-size)

static Status _create_partition_key(const TExprNode& t_expr, BlockRow* part_key, uint16_t pos) {

    auto column = std::move(*part_key->first->get_by_position(pos).column).mutate();

    switch (t_expr.node_type) {

    case TExprNodeType::DATE_LITERAL: {

        auto primitive_type =

                DataTypeFactory::instance().create_data_type(t_expr.type)->get_primitive_type();

        if (primitive_type == TYPE_DATEV2) {

            DateV2Value<DateV2ValueType> dt;

            CastParameters params;

            if (!CastToDateV2::from_string_strict_mode<DatelikeParseMode::STRICT>(

                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,

                        nullptr, params)) {

                std::stringstream ss;

                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;

                return Status::InternalError(ss.str());

            }

            column->insert_data(reinterpret_cast<const char*>(&dt), 0);

        } else if (primitive_type == TYPE_DATETIMEV2) {

            DateV2Value<DateTimeV2ValueType> dt;

            const int32_t scale =

                    t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;

            CastParameters params;

            if (!CastToDatetimeV2::from_string_strict_mode<DatelikeParseMode::STRICT>(

                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,

                        nullptr, scale, params)) {

                std::stringstream ss;

                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;

                return Status::InternalError(ss.str());

            }

            column->insert_data(reinterpret_cast<const char*>(&dt), 0);

        } else if (primitive_type == TYPE_TIMESTAMPTZ) {

            TimestampTzValue res;

            CastParameters params {.status = Status::OK(), .is_strict = true};

            const int32_t scale =

                    t_expr.type.types.empty() ? -1 : t_expr.type.types.front().scalar_type.scale;

            if (!CastToTimestampTz::from_string(

                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, res,

                        params, nullptr, scale)) [[unlikely]] {

                std::stringstream ss;

                ss << "invalid timestamptz literal in partition column, value="

                   << t_expr.date_literal;

                return Status::InternalError(ss.str());

            }

            column->insert_data(reinterpret_cast<const char*>(&res), 0);

        } else {

            VecDateTimeValue dt;

            CastParameters params;

            if (!CastToDateOrDatetime::from_string_strict_mode<DatelikeParseMode::STRICT,

                                                               DatelikeTargetType::DATE_TIME>(

                        {t_expr.date_literal.value.c_str(), t_expr.date_literal.value.size()}, dt,

                        nullptr, params)) {

                std::stringstream ss;

                ss << "invalid date literal in partition column, date=" << t_expr.date_literal;

                return Status::InternalError(ss.str());

            }

            if (primitive_type == TYPE_DATE) {

                dt.cast_to_date();

            }

            column->insert_data(reinterpret_cast<const char*>(&dt), 0);

        }

        break;

    }

    case TExprNodeType::INT_LITERAL: {

        switch (t_expr.type.types[0].scalar_type.type) {

        case TPrimitiveType::TINYINT: {

            auto value = cast_set<int8_t>(t_expr.int_literal.value);

            column->insert_data(reinterpret_cast<const char*>(&value), 0);

            break;

        }

        case TPrimitiveType::SMALLINT: {

            auto value = cast_set<int16_t>(t_expr.int_literal.value);

            column->insert_data(reinterpret_cast<const char*>(&value), 0);

            break;

        }

        case TPrimitiveType::INT: {

            auto value = cast_set<int32_t>(t_expr.int_literal.value);

            column->insert_data(reinterpret_cast<const char*>(&value), 0);

            break;

        }

        default:

            int64_t value = t_expr.int_literal.value;

            column->insert_data(reinterpret_cast<const char*>(&value), 0);

        }

        break;

    }

    case TExprNodeType::LARGE_INT_LITERAL: {

        StringParser::ParseResult parse_result = StringParser::PARSE_SUCCESS;

        auto value = StringParser::string_to_int<__int128>(t_expr.large_int_literal.value.c_str(),

                                                           t_expr.large_int_literal.value.size(),

                                                           &parse_result);

        if (parse_result != StringParser::PARSE_SUCCESS) {

            value = MAX_INT128;

        }

        column->insert_data(reinterpret_cast<const char*>(&value), 0);

        break;

    }

    case TExprNodeType::STRING_LITERAL: {

        size_t len = t_expr.string_literal.value.size();

        const char* str_val = t_expr.string_literal.value.c_str();

        column->insert_data(str_val, len);

        break;

    }

    case TExprNodeType::BOOL_LITERAL: {

        column->insert_data(reinterpret_cast<const char*>(&t_expr.bool_literal.value), 0);

        break;

    }

    case TExprNodeType::NULL_LITERAL: {

        // insert a null literal

        if (!column->is_nullable()) {

            // https://github.com/apache/doris/pull/39449 have forbid this cause. always add this check as protective measures

            return Status::InternalError("The column {} is not null, can't insert into NULL value.",

                                         part_key->first->get_by_position(pos).name);

        }

        column->insert_data(nullptr, 0);

        break;

    }

    default: {

        return Status::InternalError("unsupported partition column node type, type={}",

                                     t_expr.node_type);

    }

    }

    part_key->second = cast_set<int32_t>(column->size() - 1);

    return Status::OK();

}

// NOLINTEND(readability-function-size)

Status VOlapTablePartitionParam::_create_partition_keys(const std::vector<TExprNode>& t_exprs,

                                                        BlockRow* part_key) {

    for (int i = 0; i < t_exprs.size(); i++) {

        RETURN_IF_ERROR(_create_partition_key(t_exprs[i], part_key, _partition_slot_locs[i]));

    }

    return Status::OK();

}

Status VOlapTablePartitionParam::generate_partition_from(const TOlapTablePartition& t_part,

                                                         VOlapTablePartition*& part_result) {

    DCHECK(part_result == nullptr);

    // here we set the default value of partition bounds first! if it doesn't have some key, it will be -1.

    part_result = _obj_pool.add(new VOlapTablePartition(&_partition_block));

    part_result->id = t_part.id;

    part_result->is_mutable = t_part.is_mutable;

    // only load_to_single_tablet = true will set load_tablet_idx

    if (t_part.__isset.load_tablet_idx) {

        part_result->load_tablet_idx = t_part.load_tablet_idx;

    }

    if (_is_in_partition) {

        for (const auto& keys : t_part.in_keys) {

            RETURN_IF_ERROR(_create_partition_keys(

                    keys, &part_result->in_keys.emplace_back(&_partition_block, -1)));

        }

        if (t_part.__isset.is_default_partition && t_part.is_default_partition &&

            _default_partition == nullptr) {

            _default_partition = part_result;

        }

    } else { // range

        if (t_part.__isset.start_keys) {

            RETURN_IF_ERROR(_create_partition_keys(t_part.start_keys, &part_result->start_key));

        }

        // we generate the right bound but not insert into partition map

        if (t_part.__isset.end_keys) {

            RETURN_IF_ERROR(_create_partition_keys(t_part.end_keys, &part_result->end_key));

        }

    }

    part_result->num_buckets = t_part.num_buckets;

    auto num_indexes = _schema->indexes().size();

    if (t_part.indexes.size() != num_indexes) {

        return Status::InternalError(

                "number of partition's index is not equal with schema's"

                ", num_part_indexes={}, num_schema_indexes={}",

                t_part.indexes.size(), num_indexes);

    }

    part_result->indexes = t_part.indexes;

    std::sort(part_result->indexes.begin(), part_result->indexes.end(),

              [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {

                  return lhs.index_id < rhs.index_id;

              });

    // check index

    for (int j = 0; j < num_indexes; ++j) {

        if (part_result->indexes[j].index_id != _schema->indexes()[j]->index_id) {

            return Status::InternalError(

                    "partition's index is not equal with schema's"

                    ", part_index={}, schema_index={}",

                    part_result->indexes[j].index_id, _schema->indexes()[j]->index_id);

        }

    }

    if (t_part.__isset.total_replica_num) {

        part_result->total_replica_num = t_part.total_replica_num;

    }

    if (t_part.__isset.load_required_replica_num) {

        part_result->load_required_replica_num = t_part.load_required_replica_num;

    }

    if (t_part.__isset.tablet_version_gap_backends) {

        for (const auto& [tablet_id, backend_ids] : t_part.tablet_version_gap_backends) {

            auto& gap_set = part_result->tablet_version_gap_backends[tablet_id];

            for (auto backend_id : backend_ids) {

                gap_set.insert(backend_id);

            }

        }

    }

    return Status::OK();

}

Status VOlapTablePartitionParam::add_partitions(

        const std::vector<TOlapTablePartition>& partitions) {

    for (const auto& t_part : partitions) {

        auto* part = _obj_pool.add(new VOlapTablePartition(&_partition_block));

        part->id = t_part.id;

        part->is_mutable = t_part.is_mutable;

        // we dont pass right keys when it's MAX_VALUE. so there's possibility we only have start_key but not end_key

        // range partition

        if (t_part.__isset.start_keys) {

            RETURN_IF_ERROR(_create_partition_keys(t_part.start_keys, &part->start_key));

        }

        if (t_part.__isset.end_keys) {

            RETURN_IF_ERROR(_create_partition_keys(t_part.end_keys, &part->end_key));

        }

        // list partition - we only set 1 value in 1 partition for new created ones

        if (t_part.__isset.in_keys) {

            for (const auto& keys : t_part.in_keys) {

                RETURN_IF_ERROR(_create_partition_keys(

                        keys, &part->in_keys.emplace_back(&_partition_block, -1)));

            }

            if (t_part.__isset.is_default_partition && t_part.is_default_partition) {

                _default_partition = part;

            }

        }

        part->num_buckets = t_part.num_buckets;

        auto num_indexes = _schema->indexes().size();

        if (t_part.indexes.size() != num_indexes) {

            return Status::InternalError(

                    "number of partition's index is not equal with schema's"

                    ", num_part_indexes={}, num_schema_indexes={}",

                    t_part.indexes.size(), num_indexes);

        }

        part->indexes = t_part.indexes;

        std::sort(part->indexes.begin(), part->indexes.end(),

                  [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {

                      return lhs.index_id < rhs.index_id;

                  });

        // check index

        for (int j = 0; j < num_indexes; ++j) {

            if (part->indexes[j].index_id != _schema->indexes()[j]->index_id) {

                return Status::InternalError(

                        "partition's index is not equal with schema's"

                        ", part_index={}, schema_index={}",

                        part->indexes[j].index_id, _schema->indexes()[j]->index_id);

            }

        }

        _partitions.emplace_back(part);

        // after _creating_partiton_keys

        if (_is_in_partition) {

            for (auto& in_key : part->in_keys) {

                _partitions_map->emplace(std::tuple {in_key.first, in_key.second, false}, part);

            }

        } else {

            _partitions_map->emplace(std::tuple {part->end_key.first, part->end_key.second, false},

                                     part);

        }

    }

    return Status::OK();

}

Status VOlapTablePartitionParam::replace_partitions(

        std::vector<int64_t>& old_partition_ids,

        const std::vector<TOlapTablePartition>& new_partitions) {

    // remove old replaced partitions

    DCHECK(old_partition_ids.size() == new_partitions.size());

    // init and add new partitions. insert into _partitions

    for (int i = 0; i < new_partitions.size(); i++) {

        const auto& t_part = new_partitions[i];

        // pair old_partition_ids and new_partitions one by one. TODO: sort to opt performance

        VOlapTablePartition* old_part = nullptr;

        auto old_part_id = old_partition_ids[i];

        if (auto it = std::find_if(

                    _partitions.begin(), _partitions.end(),

                    [=](const VOlapTablePartition* lhs) { return lhs->id == old_part_id; });

            it != _partitions.end()) {

            old_part = *it;

        } else {

            return Status::InternalError("Cannot find old tablet {} in replacing", old_part_id);

        }

        auto* part = _obj_pool.add(new VOlapTablePartition(&_partition_block));

        part->id = t_part.id;

        part->is_mutable = t_part.is_mutable;

        /// just substitute directly. no need to remove and reinsert keys.

        // range partition

        part->start_key = std::move(old_part->start_key);

        part->end_key = std::move(old_part->end_key);

        // list partition

        part->in_keys = std::move(old_part->in_keys);

        if (t_part.__isset.is_default_partition && t_part.is_default_partition) {

            _default_partition = part;

        }

        part->num_buckets = t_part.num_buckets;

        auto num_indexes = _schema->indexes().size();

        if (t_part.indexes.size() != num_indexes) {

            return Status::InternalError(

                    "number of partition's index is not equal with schema's"

                    ", num_part_indexes={}, num_schema_indexes={}",

                    t_part.indexes.size(), num_indexes);

        }

        part->indexes = t_part.indexes;

        std::sort(part->indexes.begin(), part->indexes.end(),

                  [](const OlapTableIndexTablets& lhs, const OlapTableIndexTablets& rhs) {

                      return lhs.index_id < rhs.index_id;

                  });

        // check index

        for (int j = 0; j < num_indexes; ++j) {

            if (part->indexes[j].index_id != _schema->indexes()[j]->index_id) {

                return Status::InternalError(

                        "partition's index is not equal with schema's"

                        ", part_index={}, schema_index={}",

                        part->indexes[j].index_id, _schema->indexes()[j]->index_id);

            }

        }

        // add new partitions with new id.

        _partitions.emplace_back(part);

        VLOG_NOTICE << "params add new partition " << part->id;

        // replace items in _partition_maps

        if (_is_in_partition) {

            for (auto& in_key : part->in_keys) {

                (*_partitions_map)[std::tuple {in_key.first, in_key.second, false}] = part;

            }

        } else {

            (*_partitions_map)[std::tuple {part->end_key.first, part->end_key.second, false}] =

                    part;

        }

    }

    // remove old partitions by id

    std::ranges::sort(old_partition_ids);

    for (auto it = _partitions.begin(); it != _partitions.end();) {

        if (std::ranges::binary_search(old_partition_ids, (*it)->id)) {

            it = _partitions.erase(it);

        } else {

            it++;

        }

    }

    return Status::OK();

}

} // namespace doris