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

#include <bvar/bvar.h>

#include <gen_cpp/Descriptors_types.h>

#include <gen_cpp/FrontendService_types.h>

#include <gen_cpp/Types_types.h>

#include <gen_cpp/olap_common.pb.h>

#include <gen_cpp/olap_file.pb.h>

#include <gen_cpp/segment_v2.pb.h>

#include <gen_cpp/types.pb.h>

#include <json2pb/pb_to_json.h>

#include <time.h>

#include <cstdint>

#include <memory>

#include <random>

#include <set>

#include <utility>

#include "cloud/cloud_meta_mgr.h"

#include "cloud/cloud_storage_engine.h"

#include "cloud/config.h"

#include "common/config.h"

#include "io/fs/file_writer.h"

#include "io/fs/local_file_system.h"

#include "storage/data_dir.h"

#include "storage/file_header.h"

#include "storage/olap_common.h"

#include "storage/olap_define.h"

#include "storage/rowset/rowset.h"

#include "storage/rowset/rowset_meta_manager.h"

#include "storage/tablet/tablet_fwd.h"

#include "storage/tablet/tablet_meta_manager.h"

#include "storage/tablet/tablet_schema_cache.h"

#include "storage/utils.h"

#include "util/debug_points.h"

#include "util/lru_cache.h"

#include "util/mem_info.h"

#include "util/parse_util.h"

#include "util/string_util.h"

#include "util/time.h"

#include "util/uid_util.h"

using std::string;

using std::unordered_map;

using std::vector;

namespace doris {

using namespace ErrorCode;

bvar::Adder<uint64_t> g_contains_agg_with_cache_if_eligible_total(

        "g_contains_agg_with_cache_if_eligible_total");

bvar::Adder<uint64_t> g_contains_agg_with_cache_if_eligible_partial_hit(

        "g_contains_agg_with_cache_if_eligible_partial_hit");

bvar::Adder<uint64_t> g_contains_agg_with_cache_if_eligible_full_hit(

        "g_contains_agg_with_cache_if_eligible_full_hit");

bvar::Window<bvar::Adder<uint64_t>> g_contains_agg_with_cache_if_eligible_total_minute(

        "g_contains_agg_with_cache_if_eligible_total_1m",

        &g_contains_agg_with_cache_if_eligible_total, 60);

bvar::Window<bvar::Adder<uint64_t>> g_contains_agg_with_cache_if_eligible_partial_hit_minute(

        "g_contains_agg_with_cache_if_eligible_partial_hit_1m",

        &g_contains_agg_with_cache_if_eligible_partial_hit, 60);

bvar::Window<bvar::Adder<uint64_t>> g_contains_agg_with_cache_if_eligible_full_hit_minute(

        "g_contains_agg_with_cache_if_eligible_full_hit_1m",

        &g_contains_agg_with_cache_if_eligible_full_hit, 60);

TabletMetaSharedPtr TabletMeta::create(

        const TCreateTabletReq& request, const TabletUid& tablet_uid, uint64_t shard_id,

        uint32_t next_unique_id,

        const unordered_map<uint32_t, uint32_t>& col_ordinal_to_unique_id) {

    std::optional<TBinlogConfig> binlog_config;

    if (request.__isset.binlog_config) {

        binlog_config = request.binlog_config;

    }

    TInvertedIndexFileStorageFormat::type inverted_index_file_storage_format =

            request.inverted_index_file_storage_format;

    // We will discard this format. Don't make any further changes here.

    if (request.__isset.inverted_index_storage_format) {

        switch (request.inverted_index_storage_format) {

        case TInvertedIndexStorageFormat::V1:

            inverted_index_file_storage_format = TInvertedIndexFileStorageFormat::V1;

            break;

        case TInvertedIndexStorageFormat::V2:

            inverted_index_file_storage_format = TInvertedIndexFileStorageFormat::V2;

            break;

        default:

            break;

        }

    }

    // Decide storage format for this tablet. DEFAULT / not-set fall back to V2 on BE side.

    TStorageFormat::type storage_format =

            request.__isset.storage_format ? request.storage_format : TStorageFormat::V2;

    return std::make_shared<TabletMeta>(

            request.table_id, request.partition_id, request.tablet_id, request.replica_id,

            request.tablet_schema.schema_hash, shard_id, request.tablet_schema, next_unique_id,

            col_ordinal_to_unique_id, tablet_uid,

            request.__isset.tablet_type ? request.tablet_type : TTabletType::TABLET_TYPE_DISK,

            request.__isset.compression_type ? request.compression_type : TCompressionType::LZ4F,

            request.__isset.storage_policy_id ? request.storage_policy_id : -1,

            request.__isset.enable_unique_key_merge_on_write

                    ? request.enable_unique_key_merge_on_write

                    : false,

            std::move(binlog_config), request.compaction_policy,

            request.time_series_compaction_goal_size_mbytes,

            request.time_series_compaction_file_count_threshold,

            request.time_series_compaction_time_threshold_seconds,

            request.time_series_compaction_empty_rowsets_threshold,

            request.time_series_compaction_level_threshold, inverted_index_file_storage_format,

            request.tde_algorithm, storage_format,

            request.__isset.vertical_compaction_num_columns_per_group

                    ? request.vertical_compaction_num_columns_per_group

                    : 5,

            request.__isset.row_binlog_schema ? &request.row_binlog_schema : nullptr);

}

TabletMeta::~TabletMeta() {

    if (_handle) {

        TabletSchemaCache::instance()->release(_handle);

    }

}

TabletMeta::TabletMeta()

        : _tablet_uid(0, 0),

          _schema(new TabletSchema),

          _delete_bitmap(new DeleteBitmap(_tablet_id)),

          _binlog_delvec(new DeleteBitmap(_tablet_id)) {}

TabletMeta::TabletMeta(int64_t table_id, int64_t partition_id, int64_t tablet_id,

                       int64_t replica_id, int32_t schema_hash, int32_t shard_id,

                       const TTabletSchema& tablet_schema, uint32_t next_unique_id,

                       const std::unordered_map<uint32_t, uint32_t>& col_ordinal_to_unique_id,

                       TabletUid tablet_uid, TTabletType::type tabletType,

                       TCompressionType::type compression_type, int64_t storage_policy_id,

                       bool enable_unique_key_merge_on_write,

                       std::optional<TBinlogConfig> binlog_config, std::string compaction_policy,

                       int64_t time_series_compaction_goal_size_mbytes,

                       int64_t time_series_compaction_file_count_threshold,

                       int64_t time_series_compaction_time_threshold_seconds,

                       int64_t time_series_compaction_empty_rowsets_threshold,

                       int64_t time_series_compaction_level_threshold,

                       TInvertedIndexFileStorageFormat::type inverted_index_file_storage_format,

                       TEncryptionAlgorithm::type tde_algorithm,

                       TStorageFormat::type storage_format,

                       int32_t vertical_compaction_num_columns_per_group,

                       const TTabletSchema* row_binlog_schema)

        : _tablet_uid(0, 0),

          _schema(new TabletSchema),

          _delete_bitmap(new DeleteBitmap(tablet_id)),

          _binlog_delvec(new DeleteBitmap(tablet_id)),

          _storage_format(storage_format) {

    TabletMetaPB tablet_meta_pb;

    tablet_meta_pb.set_table_id(table_id);

    tablet_meta_pb.set_partition_id(partition_id);

    tablet_meta_pb.set_tablet_id(tablet_id);

    tablet_meta_pb.set_replica_id(replica_id);

    tablet_meta_pb.set_schema_hash(schema_hash);

    tablet_meta_pb.set_shard_id(shard_id);

    // Persist the creation time, but it is not used

    tablet_meta_pb.set_creation_time(time(nullptr));

    tablet_meta_pb.set_cumulative_layer_point(-1);

    tablet_meta_pb.set_tablet_state(PB_RUNNING);

    *(tablet_meta_pb.mutable_tablet_uid()) = tablet_uid.to_proto();

    tablet_meta_pb.set_tablet_type(tabletType == TTabletType::TABLET_TYPE_DISK

                                           ? TabletTypePB::TABLET_TYPE_DISK

                                           : TabletTypePB::TABLET_TYPE_MEMORY);

    tablet_meta_pb.set_enable_unique_key_merge_on_write(enable_unique_key_merge_on_write);

    tablet_meta_pb.set_storage_policy_id(storage_policy_id);

    tablet_meta_pb.set_compaction_policy(compaction_policy);

    tablet_meta_pb.set_time_series_compaction_goal_size_mbytes(

            time_series_compaction_goal_size_mbytes);

    tablet_meta_pb.set_time_series_compaction_file_count_threshold(

            time_series_compaction_file_count_threshold);

    tablet_meta_pb.set_time_series_compaction_time_threshold_seconds(

            time_series_compaction_time_threshold_seconds);

    tablet_meta_pb.set_time_series_compaction_empty_rowsets_threshold(

            time_series_compaction_empty_rowsets_threshold);

    tablet_meta_pb.set_time_series_compaction_level_threshold(

            time_series_compaction_level_threshold);

    tablet_meta_pb.set_vertical_compaction_num_columns_per_group(

            vertical_compaction_num_columns_per_group);

    SchemaCreateOptions schema_create_options_for_data = {

            .col_ordinal_to_unique_id = col_ordinal_to_unique_id,

            .compression_type = compression_type,

            .inverted_index_file_storage_format = inverted_index_file_storage_format,

            .next_unique_id = next_unique_id};

    TabletSchemaPB* schema_pb_for_data = tablet_meta_pb.mutable_schema();

    init_schema_from_thrift(tablet_schema, schema_create_options_for_data, schema_pb_for_data);

    tablet_meta_pb.set_in_restore_mode(false);

    TabletSchemaPB* schema_pb_for_row_binlog = nullptr;

    if (row_binlog_schema != nullptr) {

        tablet_meta_pb.set_row_binlog_schema_hash(row_binlog_schema->schema_hash);

        DCHECK(binlog_config.has_value());

        DCHECK(binlog_config->enable && binlog_config->binlog_format == TBinlogFormat::ROW);

        std::unordered_map<uint32_t, uint32_t> row_binlog_col_ordinal_to_unique_id;

        uint32_t row_binlog_next_unique_id = 0;

        for (uint32_t col_ordinal = 0; col_ordinal < row_binlog_schema->columns.size();

             ++col_ordinal) {

            const auto& tcolumn = row_binlog_schema->columns[col_ordinal];

            uint32_t unique_id = 0;

            if (tcolumn.col_unique_id >= 0) {

                unique_id = tcolumn.col_unique_id;

            } else {

                unique_id = col_ordinal;

            }

            row_binlog_col_ordinal_to_unique_id[col_ordinal] = unique_id;

            if (row_binlog_next_unique_id <= unique_id) {

                row_binlog_next_unique_id = unique_id + 1;

            }

        }

        SchemaCreateOptions schema_create_options_for_row_binlog = {

                .col_ordinal_to_unique_id = row_binlog_col_ordinal_to_unique_id,

                .compression_type = compression_type,

                .inverted_index_file_storage_format = inverted_index_file_storage_format,

                .next_unique_id = row_binlog_next_unique_id};

        schema_pb_for_row_binlog = tablet_meta_pb.mutable_row_binlog_schema();

        init_schema_from_thrift(*row_binlog_schema, schema_create_options_for_row_binlog,

                                schema_pb_for_row_binlog);

    }

    if (binlog_config.has_value()) {

        BinlogConfig tmp_binlog_config;

        tmp_binlog_config = binlog_config.value();

        tmp_binlog_config.to_pb(tablet_meta_pb.mutable_binlog_config());

    }

    switch (tde_algorithm) {

    case doris::TEncryptionAlgorithm::AES256:

        tablet_meta_pb.set_encryption_algorithm(EncryptionAlgorithmPB::AES_256_CTR);

        break;

    case doris::TEncryptionAlgorithm::SM4:

        tablet_meta_pb.set_encryption_algorithm(EncryptionAlgorithmPB::SM4_128_CTR);

        break;

    default:

        tablet_meta_pb.set_encryption_algorithm(EncryptionAlgorithmPB::PLAINTEXT);

    }

    // Initialize default external ColumnMeta usage according to storage format.

    // V2: legacy behavior, inline ColumnMetaPB only.

    // V3: V2 + external ColumnMetaPB (CMO) enabled by default.

    switch (_storage_format) {

    case TStorageFormat::V2:

    case TStorageFormat::DEFAULT:

    case TStorageFormat::V1:

        break;

    case TStorageFormat::V3:

        schema_pb_for_data->set_storage_format(TabletStorageFormatPB::TABLET_STORAGE_FORMAT_V3);

        _schema->set_storage_format(TabletStorageFormatPB::TABLET_STORAGE_FORMAT_V3);

        if (schema_pb_for_row_binlog != nullptr) {

            schema_pb_for_row_binlog->set_storage_format(

                    TabletStorageFormatPB::TABLET_STORAGE_FORMAT_V3);

        }

        break;

    default:

        break;

    }

    init_from_pb(tablet_meta_pb);

}

TabletMeta::TabletMeta(const TabletMeta& b)

        : MetadataAdder(b),

          _table_id(b._table_id),

          _index_id(b._index_id),

          _partition_id(b._partition_id),

          _tablet_id(b._tablet_id),

          _replica_id(b._replica_id),

          _schema_hash(b._schema_hash),

          _shard_id(b._shard_id),

          _creation_time(b._creation_time),

          _cumulative_layer_point(b._cumulative_layer_point),

          _tablet_uid(b._tablet_uid),

          _tablet_type(b._tablet_type),

          _tablet_state(b._tablet_state),

          _schema(b._schema),

          _rs_metas(b._rs_metas),

          _stale_rs_metas(b._stale_rs_metas),

          _in_restore_mode(b._in_restore_mode),

          _preferred_rowset_type(b._preferred_rowset_type),

          _storage_policy_id(b._storage_policy_id),

          _cooldown_meta_id(b._cooldown_meta_id),

          _enable_unique_key_merge_on_write(b._enable_unique_key_merge_on_write),

          _delete_bitmap(b._delete_bitmap),

          _binlog_delvec(b._binlog_delvec),

          _row_binlog_schema_hash(b._row_binlog_schema_hash),

          _row_binlog_schema(b._row_binlog_schema),

          _row_binlog_rs_metas(b._row_binlog_rs_metas),

          _binlog_config(b._binlog_config),

          _compaction_policy(b._compaction_policy),

          _time_series_compaction_goal_size_mbytes(b._time_series_compaction_goal_size_mbytes),

          _time_series_compaction_file_count_threshold(

                  b._time_series_compaction_file_count_threshold),

          _time_series_compaction_time_threshold_seconds(

                  b._time_series_compaction_time_threshold_seconds),

          _time_series_compaction_empty_rowsets_threshold(

                  b._time_series_compaction_empty_rowsets_threshold),

          _time_series_compaction_level_threshold(b._time_series_compaction_level_threshold),

          _vertical_compaction_num_columns_per_group(

                  b._vertical_compaction_num_columns_per_group) {};

void TabletMeta::init_column_from_tcolumn(uint32_t unique_id, const TColumn& tcolumn,

                                          ColumnPB* column) {

    column->set_unique_id(unique_id);

    column->set_name(tcolumn.column_name);

    column->set_is_auto_increment(tcolumn.is_auto_increment);

    if (tcolumn.__isset.is_on_update_current_timestamp) {

        column->set_is_on_update_current_timestamp(tcolumn.is_on_update_current_timestamp);

    }

    string data_type;

    EnumToString(TPrimitiveType, tcolumn.column_type.type, data_type);

    column->set_type(data_type);

    uint32_t length = TabletColumn::get_field_length_by_type(tcolumn.column_type.type,

                                                             tcolumn.column_type.len);

    column->set_length(length);

    column->set_index_length(length);

    column->set_precision(tcolumn.column_type.precision);

    column->set_frac(tcolumn.column_type.scale);

    if (tcolumn.__isset.result_is_nullable) {

        column->set_result_is_nullable(tcolumn.result_is_nullable);

    }

    if (tcolumn.__isset.be_exec_version) {

        column->set_be_exec_version(tcolumn.be_exec_version);

    }

    if (tcolumn.column_type.type == TPrimitiveType::VARCHAR ||

        tcolumn.column_type.type == TPrimitiveType::STRING) {

        if (!tcolumn.column_type.__isset.index_len) {

            column->set_index_length(10);

        } else {

            column->set_index_length(tcolumn.column_type.index_len);

        }

    }

    if (!tcolumn.is_key) {

        column->set_is_key(false);

        if (tcolumn.__isset.aggregation) {

            column->set_aggregation(tcolumn.aggregation);

        } else {

            string aggregation_type;

            EnumToString(TAggregationType, tcolumn.aggregation_type, aggregation_type);

            column->set_aggregation(aggregation_type);

        }

    } else {

        column->set_is_key(true);

        column->set_aggregation("NONE");

    }

    column->set_is_nullable(tcolumn.is_allow_null);

    if (tcolumn.__isset.default_value) {

        column->set_default_value(tcolumn.default_value);

    }

    if (tcolumn.__isset.is_bloom_filter_column) {

        column->set_is_bf_column(tcolumn.is_bloom_filter_column);

    }

    if (tcolumn.__isset.visible) {

        column->set_visible(tcolumn.visible);

    }

    for (size_t i = 0; i < tcolumn.children_column.size(); i++) {

        ColumnPB* children_column = column->add_children_columns();

        init_column_from_tcolumn(tcolumn.children_column[i].col_unique_id,

                                 tcolumn.children_column[i], children_column);

    }

    if (tcolumn.column_type.__isset.variant_max_subcolumns_count) {

        column->set_variant_max_subcolumns_count(tcolumn.column_type.variant_max_subcolumns_count);

    }

    if (tcolumn.__isset.pattern_type) {

        switch (tcolumn.pattern_type) {

        case TPatternType::MATCH_NAME:

            column->set_pattern_type(PatternTypePB::MATCH_NAME);

            break;

        case TPatternType::MATCH_NAME_GLOB:

            column->set_pattern_type(PatternTypePB::MATCH_NAME_GLOB);

        }

    }

    if (tcolumn.__isset.variant_enable_typed_paths_to_sparse) {

        column->set_variant_enable_typed_paths_to_sparse(

                tcolumn.variant_enable_typed_paths_to_sparse);

    }

    if (tcolumn.__isset.variant_max_sparse_column_statistics_size) {

        column->set_variant_max_sparse_column_statistics_size(

                tcolumn.variant_max_sparse_column_statistics_size);

    }

    if (tcolumn.__isset.variant_sparse_hash_shard_count) {

        column->set_variant_sparse_hash_shard_count(tcolumn.variant_sparse_hash_shard_count);

    }

    if (tcolumn.column_type.__isset.variant_enable_doc_mode) {

        column->set_variant_enable_doc_mode(tcolumn.column_type.variant_enable_doc_mode);

    }

    if (tcolumn.__isset.variant_doc_materialization_min_rows) {

        column->set_variant_doc_materialization_min_rows(

                tcolumn.variant_doc_materialization_min_rows);

    }

    if (tcolumn.__isset.variant_doc_hash_shard_count) {

        column->set_variant_doc_hash_shard_count(tcolumn.variant_doc_hash_shard_count);

    }

    if (tcolumn.__isset.variant_enable_nested_group) {

        column->set_variant_enable_nested_group(tcolumn.variant_enable_nested_group);

    }

}

void TabletMeta::init_schema_from_thrift(const TTabletSchema& tablet_schema,

                                         const SchemaCreateOptions& schema_create_options,

                                         TabletSchemaPB* tablet_schema_pb) {

    const std::unordered_map<uint32_t, uint32_t>& col_ordinal_to_unique_id =

            schema_create_options.col_ordinal_to_unique_id;

    TCompressionType::type compression_type = schema_create_options.compression_type;

    TInvertedIndexFileStorageFormat::type inverted_index_file_storage_format =

            schema_create_options.inverted_index_file_storage_format;

    uint32_t next_unique_id = schema_create_options.next_unique_id;

    tablet_schema_pb->set_num_short_key_columns(tablet_schema.short_key_column_count);

    tablet_schema_pb->set_num_rows_per_row_block(config::default_num_rows_per_column_file_block);

    tablet_schema_pb->set_sequence_col_idx(tablet_schema.sequence_col_idx);

    auto p_seq_map = tablet_schema_pb->mutable_seq_map(); // ColumnGroupsPB

    for (auto& it : tablet_schema.seq_map) {              // std::vector< ::doris::TColumnGroup>

        uint32_t key = it.sequence_column;

        ColumnGroupPB* cg_pb = p_seq_map->add_cg(); // ColumnGroupPB {key: {v1, v2, v3}}

        cg_pb->set_sequence_column(key);

        for (auto v : it.columns_in_group) {

            cg_pb->add_columns_in_group(v);

        }

    }

    switch (tablet_schema.keys_type) {

    case TKeysType::DUP_KEYS:

        tablet_schema_pb->set_keys_type(KeysType::DUP_KEYS);

        break;

    case TKeysType::UNIQUE_KEYS:

        tablet_schema_pb->set_keys_type(KeysType::UNIQUE_KEYS);

        break;

    case TKeysType::AGG_KEYS:

        tablet_schema_pb->set_keys_type(KeysType::AGG_KEYS);

        break;

    default:

        LOG(WARNING) << "unknown tablet keys type";

        break;

    }

    // compress_kind used to compress segment files

    tablet_schema_pb->set_compress_kind(COMPRESS_LZ4);

    // compression_type used to compress segment page

    switch (compression_type) {

    case TCompressionType::NO_COMPRESSION:

        tablet_schema_pb->set_compression_type(segment_v2::NO_COMPRESSION);

        break;

    case TCompressionType::SNAPPY:

        tablet_schema_pb->set_compression_type(segment_v2::SNAPPY);

        break;

    case TCompressionType::LZ4:

        tablet_schema_pb->set_compression_type(segment_v2::LZ4);

        break;

    case TCompressionType::LZ4F:

        tablet_schema_pb->set_compression_type(segment_v2::LZ4F);

        break;

    case TCompressionType::ZLIB:

        tablet_schema_pb->set_compression_type(segment_v2::ZLIB);

        break;

    case TCompressionType::ZSTD:

        tablet_schema_pb->set_compression_type(segment_v2::ZSTD);

        break;

    default:

        tablet_schema_pb->set_compression_type(segment_v2::LZ4F);

        break;

    }

    switch (inverted_index_file_storage_format) {

    case TInvertedIndexFileStorageFormat::V1:

        tablet_schema_pb->set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V1);

        break;

    case TInvertedIndexFileStorageFormat::V2:

        tablet_schema_pb->set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V2);

        break;

    case TInvertedIndexFileStorageFormat::V3:

        tablet_schema_pb->set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V3);

        break;

    default:

        tablet_schema_pb->set_inverted_index_storage_format(InvertedIndexStorageFormatPB::V3);

        break;

    }

    switch (tablet_schema.sort_type) {

    case TSortType::type::ZORDER:

        tablet_schema_pb->set_sort_type(SortType::ZORDER);

        break;

    default:

        tablet_schema_pb->set_sort_type(SortType::LEXICAL);

    }

    tablet_schema_pb->set_sort_col_num(tablet_schema.sort_col_num);

    for (const auto& i : tablet_schema.cluster_key_uids) {

        tablet_schema_pb->add_cluster_key_uids(i);

    }

    // set column information

    uint32_t col_ordinal = 0;

    bool has_bf_columns = false;

    for (TColumn tcolumn : tablet_schema.columns) {

        ColumnPB* column = tablet_schema_pb->add_column();

        uint32_t unique_id = -1;

        if (tcolumn.col_unique_id >= 0) {

            unique_id = tcolumn.col_unique_id;

        } else {

            unique_id = col_ordinal_to_unique_id.at(col_ordinal);

        }

        init_column_from_tcolumn(unique_id, tcolumn, column);

        if (column->name() == BINLOG_LSN_COL) {

            tablet_schema_pb->set_binlog_lsn_col_idx(col_ordinal);

        } else if (column->name() == BINLOG_TIMESTAMP_COL) {

            tablet_schema_pb->set_binlog_timestamp_col_idx(col_ordinal);

        }

        col_ordinal++;

        if (column->is_bf_column()) {

            has_bf_columns = true;

        }

        if (tablet_schema.__isset.indexes) {

            for (auto& index : tablet_schema.indexes) {

                if (index.index_type == TIndexType::type::BLOOMFILTER ||

                    index.index_type == TIndexType::type::NGRAM_BF) {

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

                    if (iequal(tcolumn.column_name, index.columns[0])) {

                        column->set_is_bf_column(true);

                        break;

                    }

                }

            }

        }

    }

    // copy index meta

    if (tablet_schema.__isset.indexes) {

        for (auto& index : tablet_schema.indexes) {

            TabletIndexPB* index_pb = tablet_schema_pb->add_index();

            index_pb->set_index_id(index.index_id);

            index_pb->set_index_name(index.index_name);

            // init col_unique_id in index at be side, since col_unique_id may be -1 at fe side

            // get column unique id by name

            for (auto column_name : index.columns) {

                for (auto column : tablet_schema_pb->column()) {

                    if (iequal(column.name(), column_name)) {

                        index_pb->add_col_unique_id(column.unique_id());

                    }

                }

            }

            switch (index.index_type) {

            case TIndexType::BITMAP:

                index_pb->set_index_type(IndexType::BITMAP);

                break;

            case TIndexType::INVERTED:

                index_pb->set_index_type(IndexType::INVERTED);

                break;

            case TIndexType::ANN:

                index_pb->set_index_type(IndexType::ANN);

                break;

            case TIndexType::BLOOMFILTER:

                index_pb->set_index_type(IndexType::BLOOMFILTER);

                break;

            case TIndexType::NGRAM_BF:

                index_pb->set_index_type(IndexType::NGRAM_BF);

                break;

            }

            if (index.__isset.properties) {

                auto properties = index_pb->mutable_properties();

                for (auto kv : index.properties) {

                    (*properties)[kv.first] = kv.second;

                }

            }

        }

    }

    tablet_schema_pb->set_next_column_unique_id(next_unique_id);

    if (has_bf_columns && tablet_schema.__isset.bloom_filter_fpp) {

        tablet_schema_pb->set_bf_fpp(tablet_schema.bloom_filter_fpp);

    }

    if (tablet_schema.__isset.is_in_memory) {

        tablet_schema_pb->set_is_in_memory(tablet_schema.is_in_memory);

    }

    if (tablet_schema.__isset.disable_auto_compaction) {

        tablet_schema_pb->set_disable_auto_compaction(tablet_schema.disable_auto_compaction);

    }

    // Deprecated legacy flatten-nested switch. Distinct from variant_enable_nested_group.

    if (tablet_schema.__isset.variant_enable_flatten_nested) {

        tablet_schema_pb->set_enable_variant_flatten_nested(

                tablet_schema.variant_enable_flatten_nested);

    }

    if (tablet_schema.__isset.delete_sign_idx) {

        tablet_schema_pb->set_delete_sign_idx(tablet_schema.delete_sign_idx);

    }

    if (tablet_schema.__isset.commit_tso_col_idx) {

        tablet_schema_pb->set_commit_tso_col_idx(tablet_schema.commit_tso_col_idx);

    }

    if (tablet_schema.__isset.store_row_column) {

        tablet_schema_pb->set_store_row_column(tablet_schema.store_row_column);

    }

    if (tablet_schema.__isset.row_store_page_size) {

        tablet_schema_pb->set_row_store_page_size(tablet_schema.row_store_page_size);

    }

    if (tablet_schema.__isset.storage_page_size) {

        tablet_schema_pb->set_storage_page_size(tablet_schema.storage_page_size);

    }

    if (tablet_schema.__isset.storage_dict_page_size) {

        tablet_schema_pb->set_storage_dict_page_size(tablet_schema.storage_dict_page_size);

    }

    if (tablet_schema.__isset.skip_write_index_on_load) {

        tablet_schema_pb->set_skip_write_index_on_load(tablet_schema.skip_write_index_on_load);

    }

    if (tablet_schema.__isset.row_store_col_cids) {

        tablet_schema_pb->mutable_row_store_column_unique_ids()->Add(

                tablet_schema.row_store_col_cids.begin(), tablet_schema.row_store_col_cids.end());

    }

}

void TabletMeta::remove_rowset_delete_bitmap(const RowsetId& rowset_id, const Version& version) {

    if (_enable_unique_key_merge_on_write) {

        delete_bitmap().remove({rowset_id, 0, 0}, {rowset_id, UINT32_MAX, 0});

        if (config::enable_mow_verbose_log) {

            LOG_INFO("delete rowset delete bitmap. tablet={}, rowset={}, version={}", tablet_id(),

                     rowset_id.to_string(), version.to_string());

        }

        size_t rowset_cache_version_size = delete_bitmap().remove_rowset_cache_version(rowset_id);

        _check_mow_rowset_cache_version_size(rowset_cache_version_size);

    }

}

Status TabletMeta::create_from_file(const string& file_path) {

    TabletMetaPB tablet_meta_pb;

    RETURN_IF_ERROR(load_from_file(file_path, &tablet_meta_pb));

    init_from_pb(tablet_meta_pb);

    return Status::OK();

}

Status TabletMeta::load_from_file(const string& file_path, TabletMetaPB* tablet_meta_pb) {

    FileHeader<TabletMetaPB> file_header(file_path);

    // In file_header.deserialize(), it validates file length, signature, checksum of protobuf.

    RETURN_IF_ERROR(file_header.deserialize());

    try {

        tablet_meta_pb->CopyFrom(file_header.message());

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

        LOG(WARNING) << "Failed to copy protocol buffer object: " << e.what()

                     << ", file=" << file_path;

        return Status::Error<PARSE_PROTOBUF_ERROR>(

                "fail to copy protocol buffer object. file={}, error={}", file_path, e.what());

    }

    return Status::OK();

}

Status TabletMeta::create_from_buffer(const uint8_t* buffer, size_t buffer_size) {

    FileHeader<TabletMetaPB> file_header(""); // empty file path

    RETURN_IF_ERROR(file_header.deserialize_from_memory(buffer, buffer_size));

    TabletMetaPB tablet_meta_pb;

    try {

        tablet_meta_pb.CopyFrom(file_header.message());

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

        LOG(WARNING) << "Failed to copy protocol buffer object from buffer: " << e.what();

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

                "fail to copy protocol buffer object from buffer. error={}", e.what());

    }

    init_from_pb(tablet_meta_pb);

    return Status::OK();

}

std::string TabletMeta::construct_header_file_path(const string& schema_hash_path,

                                                   int64_t tablet_id) {

    std::stringstream header_name_stream;

    header_name_stream << schema_hash_path << "/" << tablet_id << ".hdr";

    return header_name_stream.str();

}

Status TabletMeta::save_as_json(const string& file_path) {

    std::string json_meta;

    json2pb::Pb2JsonOptions json_options;

    json_options.pretty_json = true;

    json_options.bytes_to_base64 = true;

    to_json(&json_meta, json_options);

    // save to file

    io::FileWriterPtr file_writer;

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

    RETURN_IF_ERROR(file_writer->append(json_meta));

    RETURN_IF_ERROR(file_writer->close());

    return Status::OK();

}

Status TabletMeta::save(const string& file_path) {

    TabletMetaPB tablet_meta_pb;

    to_meta_pb(&tablet_meta_pb, false);

    return TabletMeta::save(file_path, tablet_meta_pb);

}

Status TabletMeta::save(const string& file_path, const TabletMetaPB& tablet_meta_pb) {

    DCHECK(!file_path.empty());

    FileHeader<TabletMetaPB> file_header(file_path);

    try {

        file_header.mutable_message()->CopyFrom(tablet_meta_pb);

    } catch (...) {

        LOG(WARNING) << "fail to copy protocol buffer object. file='" << file_path;

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

                "fail to copy protocol buffer object. file={}", file_path);

    }

    RETURN_IF_ERROR(file_header.prepare());

    RETURN_IF_ERROR(file_header.serialize());

    return Status::OK();

}

Status TabletMeta::save_meta(DataDir* data_dir) {

    std::lock_guard<std::shared_mutex> wrlock(_meta_lock);

    return _save_meta(data_dir);

}

Status TabletMeta::_save_meta(DataDir* data_dir) {

    // check if tablet uid is valid

    if (_tablet_uid.hi == 0 && _tablet_uid.lo == 0) {

        LOG(FATAL) << "tablet_uid is invalid"

                   << " tablet=" << tablet_id() << " _tablet_uid=" << _tablet_uid.to_string();

    }

    string meta_binary;

    auto t1 = MonotonicMicros();

    serialize(&meta_binary);

    auto t2 = MonotonicMicros();

    Status status = TabletMetaManager::save(data_dir, tablet_id(), schema_hash(), meta_binary);

    if (!status.ok()) {

        LOG(FATAL) << "fail to save tablet_meta. status=" << status << ", tablet_id=" << tablet_id()

                   << ", schema_hash=" << schema_hash();

    }

    auto t3 = MonotonicMicros();

    auto cost = t3 - t1;

    if (cost > 1 * 1000 * 1000) {

        LOG(INFO) << "save tablet(" << tablet_id() << ") meta too slow. serialize cost " << t2 - t1

                  << "(us), serialized binary size: " << meta_binary.length()

                  << "(bytes), write rocksdb cost " << t3 - t2 << "(us)";

    }

    return status;

}

void TabletMeta::serialize(string* meta_binary) {

    TabletMetaPB tablet_meta_pb;

    to_meta_pb(&tablet_meta_pb, false);

    if (tablet_meta_pb.partition_id() <= 0) {

        LOG(WARNING) << "invalid partition id " << tablet_meta_pb.partition_id() << " tablet "

                     << tablet_meta_pb.tablet_id();

    }

    DBUG_EXECUTE_IF("TabletMeta::serialize::zero_partition_id", {

        long partition_id = tablet_meta_pb.partition_id();

        tablet_meta_pb.set_partition_id(0);

        LOG(WARNING) << "set debug point TabletMeta::serialize::zero_partition_id old="

                     << partition_id << " new=" << tablet_meta_pb.DebugString();

    });

    bool serialize_success = tablet_meta_pb.SerializeToString(meta_binary);

    if (!_rs_metas.empty() || !_stale_rs_metas.empty()) {

        _avg_rs_meta_serialize_size =

                meta_binary->length() / (_rs_metas.size() + _stale_rs_metas.size());

        if (meta_binary->length() > config::tablet_meta_serialize_size_limit ||

            !serialize_success) {

            int64_t origin_meta_size = meta_binary->length();

            int64_t stale_rowsets_num = tablet_meta_pb.stale_rs_metas().size();

            tablet_meta_pb.clear_stale_rs_metas();

            meta_binary->clear();

            serialize_success = tablet_meta_pb.SerializeToString(meta_binary);

            LOG(WARNING) << "tablet meta serialization size exceeds limit: "

                         << config::tablet_meta_serialize_size_limit

                         << " clean up stale rowsets, tablet id: " << tablet_id()

                         << " stale rowset num: " << stale_rowsets_num

                         << " serialization size before clean " << origin_meta_size

                         << " serialization size after clean " << meta_binary->length();

        }

    }

    if (!serialize_success) {

        LOG(FATAL) << "failed to serialize meta " << tablet_id();

    }

}

Status TabletMeta::deserialize(std::string_view meta_binary) {

    TabletMetaPB tablet_meta_pb;

    bool parsed = tablet_meta_pb.ParseFromArray(meta_binary.data(),

                                                static_cast<int32_t>(meta_binary.size()));

    if (!parsed) {

        return Status::Error<INIT_FAILED>("parse tablet meta failed");

    }

    init_from_pb(tablet_meta_pb);

    return Status::OK();

}

void TabletMeta::init_from_pb(const TabletMetaPB& tablet_meta_pb) {

    _table_id = tablet_meta_pb.table_id();

    _index_id = tablet_meta_pb.index_id();

    _partition_id = tablet_meta_pb.partition_id();

    _tablet_id = tablet_meta_pb.tablet_id();

    _replica_id = tablet_meta_pb.replica_id();

    _schema_hash = tablet_meta_pb.schema_hash();

    _shard_id = tablet_meta_pb.shard_id();

    _creation_time = tablet_meta_pb.creation_time();

    _cumulative_layer_point = tablet_meta_pb.cumulative_layer_point();

    _tablet_uid = TabletUid(tablet_meta_pb.tablet_uid());

    _ttl_seconds = tablet_meta_pb.ttl_seconds();

    if (tablet_meta_pb.has_tablet_type()) {

        _tablet_type = tablet_meta_pb.tablet_type();

    } else {

        _tablet_type = TabletTypePB::TABLET_TYPE_DISK;

    }

    // init _tablet_state

    switch (tablet_meta_pb.tablet_state()) {

    case PB_NOTREADY:

        _tablet_state = TabletState::TABLET_NOTREADY;

        break;

    case PB_RUNNING:

        _tablet_state = TabletState::TABLET_RUNNING;

        break;

    case PB_TOMBSTONED:

        _tablet_state = TabletState::TABLET_TOMBSTONED;

        break;

    case PB_STOPPED:

        _tablet_state = TabletState::TABLET_STOPPED;

        break;

    case PB_SHUTDOWN:

        _tablet_state = TabletState::TABLET_SHUTDOWN;

        break;

    default:

        LOG(WARNING) << "tablet has no state. tablet=" << tablet_id()

                     << ", schema_hash=" << schema_hash();

    }

    // init _schema

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

    schema->init_from_pb(tablet_meta_pb.schema());

    if (_handle) {

        TabletSchemaCache::instance()->release(_handle);

    }

    auto pair = TabletSchemaCache::instance()->insert(schema->to_key());

    _handle = pair.first;

    _schema = pair.second;

    if (tablet_meta_pb.has_row_binlog_schema()) {

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

        row_binlog_schema->init_from_pb(tablet_meta_pb.row_binlog_schema());

        _row_binlog_schema = std::move(row_binlog_schema);

        _row_binlog_schema_hash = tablet_meta_pb.row_binlog_schema_hash();

    }

    if (tablet_meta_pb.has_enable_unique_key_merge_on_write()) {

        _enable_unique_key_merge_on_write = tablet_meta_pb.enable_unique_key_merge_on_write();

        _delete_bitmap->set_tablet_id(_tablet_id);

        _binlog_delvec->set_tablet_id(_tablet_id);

    }

    // init _rs_metas

    for (auto& it : tablet_meta_pb.rs_metas()) {

        RowsetMetaSharedPtr rs_meta(new RowsetMeta());

        rs_meta->init_from_pb(it);

        _rs_metas.emplace(rs_meta->version(), rs_meta);

    }

    // For mow table, delete bitmap of stale rowsets has not been persisted.

    // When be restart, query should not read the stale rowset, otherwise duplicate keys

    // will be read out. Therefore, we don't add them to _stale_rs_meta for mow table.

    if (!config::skip_loading_stale_rowset_meta && !_enable_unique_key_merge_on_write) {

        for (auto& it : tablet_meta_pb.stale_rs_metas()) {

            RowsetMetaSharedPtr rs_meta(new RowsetMeta());

            rs_meta->init_from_pb(it);

            _stale_rs_metas.emplace(rs_meta->version(), rs_meta);

        }

    }

    for (auto& it : tablet_meta_pb.row_binlog_rs_metas()) {

        RowsetMetaSharedPtr rs_meta(new RowsetMeta());

        rs_meta->init_from_pb(it);

        _row_binlog_rs_metas.emplace(rs_meta->version(), rs_meta);

    }

    if (tablet_meta_pb.has_in_restore_mode()) {

        _in_restore_mode = tablet_meta_pb.in_restore_mode();

    }

    if (tablet_meta_pb.has_preferred_rowset_type()) {

        _preferred_rowset_type = tablet_meta_pb.preferred_rowset_type();

    }

    _storage_policy_id = tablet_meta_pb.storage_policy_id();

    if (tablet_meta_pb.has_cooldown_meta_id()) {

        _cooldown_meta_id = tablet_meta_pb.cooldown_meta_id();

    }

    if (tablet_meta_pb.has_delete_bitmap()) {

        int rst_ids_size = tablet_meta_pb.delete_bitmap().rowset_ids_size();

        int seg_ids_size = tablet_meta_pb.delete_bitmap().segment_ids_size();

        int versions_size = tablet_meta_pb.delete_bitmap().versions_size();

        int seg_maps_size = tablet_meta_pb.delete_bitmap().segment_delete_bitmaps_size();

        int binlog_mark_size = tablet_meta_pb.delete_bitmap().is_binlog_delvec_size();

        CHECK(rst_ids_size == seg_ids_size && seg_ids_size == seg_maps_size &&

              seg_maps_size == versions_size);

        CHECK(binlog_mark_size == 0 || binlog_mark_size == rst_ids_size);

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

            RowsetId rst_id;

            rst_id.init(tablet_meta_pb.delete_bitmap().rowset_ids(i));

            auto seg_id = tablet_meta_pb.delete_bitmap().segment_ids(i);

            auto ver = tablet_meta_pb.delete_bitmap().versions(i);

            auto bitmap = tablet_meta_pb.delete_bitmap().segment_delete_bitmaps(i).data();

            bool from_binlog = tablet_meta_pb.delete_bitmap().is_binlog_delvec_size() > 0

                                       ? tablet_meta_pb.delete_bitmap().is_binlog_delvec(i)

                                       : false;

            if (!from_binlog) {

                delete_bitmap().delete_bitmap[{rst_id, seg_id, ver}] =

                        roaring::Roaring::read(bitmap);

            } else {

                binlog_delvec().delete_bitmap[{rst_id, seg_id, ver}] =

                        roaring::Roaring::read(bitmap);

            }

        }

    }

    if (tablet_meta_pb.has_binlog_config()) {

        _binlog_config = tablet_meta_pb.binlog_config();

    }

    _compaction_policy = tablet_meta_pb.compaction_policy();

    _time_series_compaction_goal_size_mbytes =

            tablet_meta_pb.time_series_compaction_goal_size_mbytes();

    _time_series_compaction_file_count_threshold =

            tablet_meta_pb.time_series_compaction_file_count_threshold();

    _time_series_compaction_time_threshold_seconds =

            tablet_meta_pb.time_series_compaction_time_threshold_seconds();

    _time_series_compaction_empty_rowsets_threshold =

            tablet_meta_pb.time_series_compaction_empty_rowsets_threshold();

    _time_series_compaction_level_threshold =

            tablet_meta_pb.time_series_compaction_level_threshold();

    _vertical_compaction_num_columns_per_group =

            tablet_meta_pb.vertical_compaction_num_columns_per_group();

    if (tablet_meta_pb.has_encryption_algorithm()) {