be/src/storage/tablet/tablet_meta.cpp

Source
// 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 <ranges>
#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.tablet_role ? request.tablet_role : TTabletRole::TABLET_ROLE_DATA);
}

TabletMeta::~TabletMeta() {
    if (_handle) {
        TabletSchemaCache::instance()->release(_handle);
    }
}

TabletMeta::TabletMeta()
        : _tablet_uid(0, 0),
          _schema(new TabletSchema),
          _delete_bitmap(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,
                       TTabletRole::type tablet_role)
        : _tablet_uid(0, 0),
          _schema(new TabletSchema),
          _delete_bitmap(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);
    tablet_meta_pb.set_tablet_role(tablet_role == TTabletRole::TABLET_ROLE_ROW_BINLOG
                                           ? TabletRolePB::TABLET_ROLE_ROW_BINLOG
                                           : TabletRolePB::TABLET_ROLE_DATA);
    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);

    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);
        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_config(b._binlog_config),
          _tablet_role(b._tablet_role),
          _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);
    if (tablet_schema.__isset.binlog_tso_idx) {
        tablet_schema_pb->set_binlog_tso_col_idx(tablet_schema.binlog_tso_idx);
    }
    if (tablet_schema.__isset.binlog_lsn_idx) {
        tablet_schema_pb->set_binlog_lsn_col_idx(tablet_schema.binlog_lsn_idx);
    }
    if (tablet_schema.__isset.binlog_op_idx) {
        tablet_schema_pb->set_binlog_op_col_idx(tablet_schema.binlog_op_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);

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

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

    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();
        CHECK(rst_ids_size == seg_ids_size && seg_ids_size == seg_maps_size &&
              seg_maps_size == versions_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();
            delete_bitmap().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();
    }
    _tablet_role = tablet_meta_pb.tablet_role();
    _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()) {
        _encryption_algorithm = tablet_meta_pb.encryption_algorithm();
    }
}

void TabletMeta::to_meta_pb(TabletMetaPB* tablet_meta_pb, bool cloud_get_rowset_meta) {
    tablet_meta_pb->set_table_id(table_id());
    tablet_meta_pb->set_index_id(index_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());
    tablet_meta_pb->set_creation_time(creation_time());
    tablet_meta_pb->set_cumulative_layer_point(cumulative_layer_point());
    *(tablet_meta_pb->mutable_tablet_uid()) = tablet_uid().to_proto();
    tablet_meta_pb->set_tablet_type(_tablet_type);
    tablet_meta_pb->set_ttl_seconds(_ttl_seconds);
    switch (tablet_state()) {
    case TABLET_NOTREADY:
        tablet_meta_pb->set_tablet_state(PB_NOTREADY);
        break;
    case TABLET_RUNNING:
        tablet_meta_pb->set_tablet_state(PB_RUNNING);
        break;
    case TABLET_TOMBSTONED:
        tablet_meta_pb->set_tablet_state(PB_TOMBSTONED);
        break;
    case TABLET_STOPPED:
        tablet_meta_pb->set_tablet_state(PB_STOPPED);
        break;
    case TABLET_SHUTDOWN:
        tablet_meta_pb->set_tablet_state(PB_SHUTDOWN);
        break;
    }

    // RowsetMetaPB is separated from TabletMetaPB
    if (!config::is_cloud_mode() || cloud_get_rowset_meta) {
        for (const auto& [_, rs] : _rs_metas) {
            rs->to_rowset_pb(tablet_meta_pb->add_rs_metas());
        }
        for (const auto& [_, rs] : _stale_rs_metas) {
            rs->to_rowset_pb(tablet_meta_pb->add_stale_rs_metas());
        }
    }

    _schema->to_schema_pb(tablet_meta_pb->mutable_schema());

    tablet_meta_pb->set_in_restore_mode(in_restore_mode());

    // to avoid modify tablet meta to the greatest extend
    if (_preferred_rowset_type == BETA_ROWSET) {
        tablet_meta_pb->set_preferred_rowset_type(_preferred_rowset_type);
    }
    if (_storage_policy_id > 0) {
        tablet_meta_pb->set_storage_policy_id(_storage_policy_id);
    }
    if (_cooldown_meta_id.initialized()) {
        tablet_meta_pb->mutable_cooldown_meta_id()->CopyFrom(_cooldown_meta_id.to_proto());
    }

    tablet_meta_pb->set_enable_unique_key_merge_on_write(_enable_unique_key_merge_on_write);

    if (_enable_unique_key_merge_on_write) {
        std::set<RowsetId> stale_rs_ids;
        for (const auto& [_, rowset] : _stale_rs_metas) {
            stale_rs_ids.insert(rowset->rowset_id());
        }
        DeleteBitmapPB* delete_bitmap_pb = tablet_meta_pb->mutable_delete_bitmap();
        for (auto& [id, bitmap] : delete_bitmap().snapshot().delete_bitmap) {
            auto& [rowset_id, segment_id, ver] = id;
            if (stale_rs_ids.count(rowset_id) != 0) {
                continue;
            }
            delete_bitmap_pb->add_rowset_ids(rowset_id.to_string());
            delete_bitmap_pb->add_segment_ids(segment_id);
            delete_bitmap_pb->add_versions(ver);
            std::string bitmap_data(bitmap.getSizeInBytes(), '\0');
            bitmap.write(bitmap_data.data());
            *(delete_bitmap_pb->add_segment_delete_bitmaps()) = std::move(bitmap_data);
        }
    }
    _binlog_config.to_pb(tablet_meta_pb->mutable_binlog_config());
    tablet_meta_pb->set_tablet_role(_tablet_role);
    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());

    tablet_meta_pb->set_encryption_algorithm(_encryption_algorithm);
}

void TabletMeta::to_json(string* json_string, json2pb::Pb2JsonOptions& options) {
    TabletMetaPB tablet_meta_pb;
    to_meta_pb(&tablet_meta_pb, false);
    json2pb::ProtoMessageToJson(tablet_meta_pb, json_string, options);
}

Version TabletMeta::max_version() const {
    Version max_version = {-1, 0};
    for (const auto& [_, rs_meta] : _rs_metas) {
        if (rs_meta->end_version() > max_version.second) {
            max_version = rs_meta->version();
        }
    }
    return max_version;
}

size_t TabletMeta::version_count_cross_with_range(const Version& range) const {
    size_t count = 0;
    for (const auto& [_, rs_meta] : _rs_metas) {
        if (!(range.first > rs_meta->version().second || range.second < rs_meta->version().first)) {
            count++;
        }
    }
    return count;
}

Status TabletMeta::add_rs_meta(const RowsetMetaSharedPtr& rs_meta) {
    // check RowsetMeta is valid
    for (const auto& [_, rs] : _rs_metas) {
        if (rs->version() == rs_meta->version()) {
            if (rs->rowset_id() != rs_meta->rowset_id()) {
                return Status::Error<PUSH_VERSION_ALREADY_EXIST>(
                        "version already exist. rowset_id={}, version={}, tablet={}",
                        rs->rowset_id().to_string(), rs->version().to_string(), tablet_id());
            } else {
                // rowsetid,version is equal, it is a duplicate req, skip it
                return Status::OK();
            }
        }
    }
    _rs_metas.emplace(rs_meta->version(), rs_meta);
    return Status::OK();
}

void TabletMeta::add_rowsets_unchecked(const std::vector<RowsetSharedPtr>& to_add) {
    for (const auto& rs : to_add) {
        _rs_metas.emplace(rs->rowset_meta()->version(), rs->rowset_meta());
    }
}

void TabletMeta::delete_rs_meta_by_version(const Version& version,
                                           std::vector<RowsetMetaSharedPtr>* deleted_rs_metas) {
    size_t rowset_cache_version_size = 0;
    if (auto it = _rs_metas.find(version); it != _rs_metas.end()) {
        if (deleted_rs_metas != nullptr) {
            deleted_rs_metas->push_back(it->second);
        }
        auto rowset_id = it->second->rowset_id();
        _rs_metas.erase(it);
        if (_enable_unique_key_merge_on_write) {
            rowset_cache_version_size = _delete_bitmap->remove_rowset_cache_version(rowset_id);
        }
        return;
    }
    _check_mow_rowset_cache_version_size(rowset_cache_version_size);
}

void TabletMeta::modify_rs_metas(const std::vector<RowsetMetaSharedPtr>& to_add,
                                 const std::vector<RowsetMetaSharedPtr>& to_delete,
                                 bool same_version) {
    size_t rowset_cache_version_size = 0;
    // Remove to_delete rowsets from _rs_metas
    for (auto rs_to_del : to_delete) {
        if (auto it = _rs_metas.find(rs_to_del->version()); it != _rs_metas.end()) {
            auto rowset_id = it->second->rowset_id();
            _rs_metas.erase(it);
            if (_enable_unique_key_merge_on_write) {
                rowset_cache_version_size = _delete_bitmap->remove_rowset_cache_version(rowset_id);
            }
        }
    }
    if (!same_version) {
        // put to_delete rowsets in _stale_rs_metas.
        for (auto rs_to_del : to_delete) {
            _stale_rs_metas.emplace(rs_to_del->version(), rs_to_del);
        }
    }

    // put to_add rowsets in _rs_metas.
    for (auto rs_to_add : to_add) {
        _rs_metas.emplace(rs_to_add->version(), rs_to_add);
    }
    _check_mow_rowset_cache_version_size(rowset_cache_version_size);
}

// Use the passing "rs_metas" to replace the rs meta in this tablet meta
// Also clear the _stale_rs_metas because this tablet meta maybe copyied from
// an existing tablet before. Add after revise, only the passing "rs_metas"
// is needed.
void TabletMeta::revise_rs_metas(std::vector<RowsetMetaSharedPtr>&& rs_metas) {
    {
        std::lock_guard<std::shared_mutex> wrlock(_meta_lock);
        _rs_metas.clear();
        for (auto& rs_meta : rs_metas) {
            _rs_metas.emplace(rs_meta->version(), rs_meta);
        }
        _stale_rs_metas.clear();
    }
    if (_enable_unique_key_merge_on_write) {
        _delete_bitmap->clear_rowset_cache_version();
    }
}

// This method should call after revise_rs_metas, since new rs_metas might be a subset
// of original tablet, we should revise the delete_bitmap according to current rowset.
//
// Delete bitmap is protected by Tablet::_meta_lock, we don't need to acquire the
// TabletMeta's _meta_lock
void TabletMeta::revise_delete_bitmap_unlocked(const DeleteBitmap& delete_bitmap) {
    _delete_bitmap = std::make_unique<DeleteBitmap>(tablet_id());
    for (const auto& [_, rs] : _rs_metas) {
        DeleteBitmap rs_bm(tablet_id());
        delete_bitmap.subset({rs->rowset_id(), 0, 0}, {rs->rowset_id(), UINT32_MAX, INT64_MAX},
                             &rs_bm);
        _delete_bitmap->merge(rs_bm);
    }
    for (const auto& [_, rs] : _stale_rs_metas) {
        DeleteBitmap rs_bm(tablet_id());
        delete_bitmap.subset({rs->rowset_id(), 0, 0}, {rs->rowset_id(), UINT32_MAX, INT64_MAX},
                             &rs_bm);
        _delete_bitmap->merge(rs_bm);
    }
}

void TabletMeta::delete_stale_rs_meta_by_version(const Version& version) {
    _stale_rs_metas.erase(version);
}

RowsetMetaSharedPtr TabletMeta::acquire_rs_meta_by_version(const Version& version) const {
    if (auto it = _rs_metas.find(version); it != _rs_metas.end()) {
        return it->second;
    }
    return nullptr;
}

RowsetMetaSharedPtr TabletMeta::acquire_stale_rs_meta_by_version(const Version& version) const {
    if (auto it = _stale_rs_metas.find(version); it != _stale_rs_metas.end()) {
        return it->second;
    }
    return nullptr;
}

Status TabletMeta::set_partition_id(int64_t partition_id) {
    if ((_partition_id > 0 && _partition_id != partition_id) || partition_id < 1) {
        LOG(WARNING) << "cur partition id=" << _partition_id << " new partition id=" << partition_id
                     << " not equal";
    }
    _partition_id = partition_id;
    return Status::OK();
}

void TabletMeta::clear_stale_rowset() {
    _stale_rs_metas.clear();
    if (_enable_unique_key_merge_on_write) {
        _delete_bitmap->clear_rowset_cache_version();
    }
}

void TabletMeta::clear_rowsets() {
    _rs_metas.clear();
    if (_enable_unique_key_merge_on_write) {
        _delete_bitmap->clear_rowset_cache_version();
    }
}

void TabletMeta::_check_mow_rowset_cache_version_size(size_t rowset_cache_version_size) {
    if (_enable_unique_key_merge_on_write && config::enable_mow_verbose_log &&
        rowset_cache_version_size > _rs_metas.size() + _stale_rs_metas.size()) {
        std::stringstream ss;
        auto rowset_ids = _delete_bitmap->get_rowset_cache_version();
        std::set<std::string> tablet_rowset_ids;
        {
            std::shared_lock rlock(_meta_lock);
            for (const auto& [_, rs_meta] : _rs_metas) {
                tablet_rowset_ids.emplace(rs_meta->rowset_id().to_string());
            }
            for (const auto& [_, rs_meta] : _stale_rs_metas) {
                tablet_rowset_ids.emplace(rs_meta->rowset_id().to_string());
            }
        }
        for (const auto& rowset_id : rowset_ids) {
            if (tablet_rowset_ids.find(rowset_id) == tablet_rowset_ids.end()) {
                ss << rowset_id << ", ";
            }
        }
        // size(rowset_cache_version) <= size(_rs_metas) + size(_stale_rs_metas) + size(_unused_rs)
        std::string msg = fmt::format(
                "tablet: {}, rowset_cache_version size: {}, "
                "_rs_metas size: {}, _stale_rs_metas size: {}, delta: {}. rowset only in cache: {}",
                _tablet_id, rowset_cache_version_size, _rs_metas.size(), _stale_rs_metas.size(),
                rowset_cache_version_size - _rs_metas.size() - _stale_rs_metas.size(), ss.str());
        LOG(INFO) << msg;
    }
}

bool operator==(const TabletMeta& a, const TabletMeta& b) {
    if (a._table_id != b._table_id) return false;
    if (a._index_id != b._index_id) return false;
    if (a._partition_id != b._partition_id) return false;
    if (a._tablet_id != b._tablet_id) return false;
    if (a._replica_id != b._replica_id) return false;
    if (a._schema_hash != b._schema_hash) return false;
    if (a._shard_id != b._shard_id) return false;
    if (a._creation_time != b._creation_time) return false;
    if (a._cumulative_layer_point != b._cumulative_layer_point) return false;
    if (a._tablet_uid != b._tablet_uid) return false;
    if (a._tablet_type != b._tablet_type) return false;
    if (a._tablet_state != b._tablet_state) return false;
    if (*a._schema != *b._schema) return false;
    if (a._rs_metas != b._rs_metas) return false;
    if (a._in_restore_mode != b._in_restore_mode) return false;
    if (a._preferred_rowset_type != b._preferred_rowset_type) return false;
    if (a._storage_policy_id != b._storage_policy_id) return false;
    if (a._compaction_policy != b._compaction_policy) return false;
    if (a._time_series_compaction_goal_size_mbytes != b._time_series_compaction_goal_size_mbytes)
        return false;
    if (a._time_series_compaction_file_count_threshold !=
        b._time_series_compaction_file_count_threshold)
        return false;
    if (a._time_series_compaction_time_threshold_seconds !=
        b._time_series_compaction_time_threshold_seconds)
        return false;
    if (a._time_series_compaction_empty_rowsets_threshold !=
        b._time_series_compaction_empty_rowsets_threshold)
        return false;
    if (a._time_series_compaction_level_threshold != b._time_series_compaction_level_threshold)
        return false;
    return true;
}

bool operator!=(const TabletMeta& a, const TabletMeta& b) {
    return !(a == b);
}

// We cannot just copy the underlying memory to construct a string
// due to equivalent objects may have different padding bytes.
// Reading padding bytes is undefined behavior, neither copy nor
// placement new will help simplify the code.
// Refer to C11 standards §6.2.6.1/6 and §6.7.9/21 for more info.
static std::string agg_cache_key(int64_t tablet_id, const DeleteBitmap::BitmapKey& bmk) {
    std::string ret(sizeof(tablet_id) + sizeof(bmk), '\0');
    *reinterpret_cast<int64_t*>(ret.data()) = tablet_id;
    auto t = reinterpret_cast<DeleteBitmap::BitmapKey*>(ret.data() + sizeof(tablet_id));
    std::get<RowsetId>(*t).version = std::get<RowsetId>(bmk).version;
    std::get<RowsetId>(*t).hi = std::get<RowsetId>(bmk).hi;
    std::get<RowsetId>(*t).mi = std::get<RowsetId>(bmk).mi;
    std::get<RowsetId>(*t).lo = std::get<RowsetId>(bmk).lo;
    std::get<1>(*t) = std::get<1>(bmk);
    std::get<2>(*t) = std::get<2>(bmk);
    return ret;
}

// decode cache key info from a agg_cache_key
static void decode_agg_cache_key(const std::string& key_str, int64_t& tablet_id,
                                 DeleteBitmap::BitmapKey& bmk) {
    const char* ptr = key_str.data();
    tablet_id = *reinterpret_cast<const int64_t*>(ptr);
    ptr += sizeof(tablet_id);
    const auto* t = reinterpret_cast<const DeleteBitmap::BitmapKey*>(ptr);
    std::get<RowsetId>(bmk).version = std::get<RowsetId>(*t).version;
    std::get<RowsetId>(bmk).hi = std::get<RowsetId>(*t).hi;
    std::get<RowsetId>(bmk).mi = std::get<RowsetId>(*t).mi;
    std::get<RowsetId>(bmk).lo = std::get<RowsetId>(*t).lo;
    std::get<1>(bmk) = std::get<1>(*t);
    std::get<2>(bmk) = std::get<2>(*t);
}

DeleteBitmapAggCache::DeleteBitmapAggCache(size_t capacity)
        : LRUCachePolicy(CachePolicy::CacheType::DELETE_BITMAP_AGG_CACHE, capacity,
                         LRUCacheType::SIZE, config::delete_bitmap_agg_cache_stale_sweep_time_sec,
                         /*num_shards*/ 256,
                         /*element_count_capacity*/ 0, /*enable_prune*/ true,
                         /*is_lru_k*/ false) {}

DeleteBitmapAggCache* DeleteBitmapAggCache::instance() {
    return ExecEnv::GetInstance()->delete_bitmap_agg_cache();
}

DeleteBitmapAggCache* DeleteBitmapAggCache::create_instance(size_t capacity) {
    return new DeleteBitmapAggCache(capacity);
}

DeleteBitmap DeleteBitmapAggCache::snapshot(int64_t tablet_id) {
    DeleteBitmap ret(tablet_id);
    auto collector = [&](const LRUHandle* handle) {
        auto key = handle->key().to_string();
        int64_t key_tablet_id;
        DeleteBitmap::BitmapKey bmk;
        decode_agg_cache_key(key, key_tablet_id, bmk);
        if (key_tablet_id == tablet_id) {
            const auto& dbm = reinterpret_cast<DeleteBitmapAggCache::Value*>(handle->value)->bitmap;
            ret.set(bmk, dbm);
        }
    };
    DeleteBitmapAggCache::instance()->for_each_entry(collector);
    return ret;
}

DeleteBitmap::DeleteBitmap(int64_t tablet_id) : _tablet_id(tablet_id) {}

DeleteBitmap::DeleteBitmap(const DeleteBitmap& o) {
    std::shared_lock l1(o.lock);
    delete_bitmap = o.delete_bitmap;
    _tablet_id = o._tablet_id;
}

DeleteBitmap& DeleteBitmap::operator=(const DeleteBitmap& o) {
    if (this == &o) return *this;
    if (this < &o) {
        std::unique_lock l1(lock);
        std::shared_lock l2(o.lock);
        delete_bitmap = o.delete_bitmap;
        _tablet_id = o._tablet_id;
    } else {
        std::shared_lock l2(o.lock);
        std::unique_lock l1(lock);
        delete_bitmap = o.delete_bitmap;
        _tablet_id = o._tablet_id;
    }
    return *this;
}

DeleteBitmap::DeleteBitmap(DeleteBitmap&& o) noexcept {
    std::scoped_lock l(o.lock, o._rowset_cache_version_lock);
    delete_bitmap = std::move(o.delete_bitmap);
    _tablet_id = std::move(o._tablet_id);
    o._rowset_cache_version.clear();
}

DeleteBitmap& DeleteBitmap::operator=(DeleteBitmap&& o) noexcept {
    if (this == &o) return *this;
    std::scoped_lock l(lock, o.lock, o._rowset_cache_version_lock);
    delete_bitmap = std::move(o.delete_bitmap);
    _tablet_id = std::move(o._tablet_id);
    o._rowset_cache_version.clear();
    return *this;
}

DeleteBitmap DeleteBitmap::from_pb(const DeleteBitmapPB& pb, int64_t tablet_id) {
    size_t len = pb.rowset_ids().size();
    DCHECK_EQ(len, pb.segment_ids().size());
    DCHECK_EQ(len, pb.versions().size());
    DeleteBitmap delete_bitmap(tablet_id);
    for (int32_t i = 0; i < len; ++i) {
        RowsetId rs_id;
        rs_id.init(pb.rowset_ids(i));
        BitmapKey key = {rs_id, pb.segment_ids(i), pb.versions(i)};
        delete_bitmap.delete_bitmap[key] =
                roaring::Roaring::read(pb.segment_delete_bitmaps(i).data());
    }
    return delete_bitmap;
}

DeleteBitmapPB DeleteBitmap::to_pb() {
    std::shared_lock l(lock);
    DeleteBitmapPB ret;
    for (const auto& [k, v] : delete_bitmap) {
        ret.mutable_rowset_ids()->Add(std::get<0>(k).to_string());
        ret.mutable_segment_ids()->Add(std::get<1>(k));
        ret.mutable_versions()->Add(std::get<2>(k));
        std::string bitmap_data(v.getSizeInBytes(), '\0');
        v.write(bitmap_data.data());
        ret.mutable_segment_delete_bitmaps()->Add(std::move(bitmap_data));
    }
    return ret;
}

DeleteBitmap DeleteBitmap::snapshot() const {
    std::shared_lock l(lock);
    return DeleteBitmap(*this);
}

DeleteBitmap DeleteBitmap::snapshot(Version version) const {
    // Take snapshot first, then remove keys greater than given version.
    DeleteBitmap snapshot = this->snapshot();
    auto it = snapshot.delete_bitmap.begin();
    while (it != snapshot.delete_bitmap.end()) {
        if (std::get<2>(it->first) > version) {
            it = snapshot.delete_bitmap.erase(it);
        } else {
            it++;
        }
    }
    return snapshot;
}

void DeleteBitmap::add(const BitmapKey& bmk, uint32_t row_id) {
    std::lock_guard l(lock);
    delete_bitmap[bmk].add(row_id);
}

int DeleteBitmap::remove(const BitmapKey& bmk, uint32_t row_id) {
    std::lock_guard l(lock);
    auto it = delete_bitmap.find(bmk);
    if (it == delete_bitmap.end()) return -1;
    it->second.remove(row_id);
    return 0;
}

void DeleteBitmap::remove(const BitmapKey& start, const BitmapKey& end) {
    std::lock_guard l(lock);
    for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end();) {
        auto& [k, _] = *it;
        if (k >= end) {
            break;
        }
        it = delete_bitmap.erase(it);
    }
}

void DeleteBitmap::remove(const std::vector<std::tuple<BitmapKey, BitmapKey>>& key_ranges) {
    std::lock_guard l(lock);
    for (auto& [start, end] : key_ranges) {
        for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end();) {
            auto& [k, _] = *it;
            if (k >= end) {
                break;
            }
            it = delete_bitmap.erase(it);
        }
    }
}

bool DeleteBitmap::contains(const BitmapKey& bmk, uint32_t row_id) const {
    std::shared_lock l(lock);
    auto it = delete_bitmap.find(bmk);
    return it != delete_bitmap.end() && it->second.contains(row_id);
}

bool DeleteBitmap::contain_rowsets(const RowsetIdUnorderedSet& rowset_ids) const {
    std::shared_lock l(lock);
    return std::any_of(delete_bitmap.begin(), delete_bitmap.end(), [&](const auto& entry) {
        return rowset_ids.contains(std::get<0>(entry.first));
    });
}

bool DeleteBitmap::contains_agg(const BitmapKey& bmk, uint32_t row_id) const {
    return get_agg(bmk)->contains(row_id);
}

bool DeleteBitmap::empty() const {
    std::shared_lock l(lock);
    return delete_bitmap.empty();
}

uint64_t DeleteBitmap::cardinality() const {
    std::shared_lock l(lock);
    uint64_t res = 0;
    for (auto entry : delete_bitmap) {
        if (std::get<1>(entry.first) != DeleteBitmap::INVALID_SEGMENT_ID) {
            res += entry.second.cardinality();
        }
    }
    return res;
}

uint64_t DeleteBitmap::get_size() const {
    std::shared_lock l(lock);
    uint64_t charge = 0;
    for (auto& [k, v] : delete_bitmap) {
        if (std::get<1>(k) != DeleteBitmap::INVALID_SEGMENT_ID) {
            charge += v.getSizeInBytes();
        }
    }
    return charge;
}

bool DeleteBitmap::contains_agg_with_cache_if_eligible(const BitmapKey& bmk,
                                                       uint32_t row_id) const {
    g_contains_agg_with_cache_if_eligible_total << 1;
    int64_t start_version {0};
    if (config::enable_mow_get_agg_by_cache) {
        auto deleter = [&](Cache::Handle* handle) {
            DeleteBitmapAggCache::instance()->release(handle);
        };
        std::unique_ptr<Cache::Handle, decltype(deleter)> dbm_handle(nullptr, deleter);
        int64_t cached_version = 0;
        // 1. try to lookup the desired key directly
        dbm_handle.reset(DeleteBitmapAggCache::instance()->lookup(agg_cache_key(_tablet_id, bmk)));
        if (dbm_handle != nullptr) {
            cached_version = std::get<2>(bmk);
        } else {
            // 2. if not found, try to lookup with cached version
            cached_version = _get_rowset_cache_version(bmk);
            if (cached_version > 0) {
                if (cached_version > std::get<2>(bmk)) {
                    cached_version = 0;
                } else {
                    dbm_handle.reset(DeleteBitmapAggCache::instance()->lookup(agg_cache_key(
                            _tablet_id, {std::get<0>(bmk), std::get<1>(bmk), cached_version})));
                }
            }
        }
        if (dbm_handle != nullptr) {
            const auto& cached_dbm =
                    reinterpret_cast<DeleteBitmapAggCache::Value*>(
                            DeleteBitmapAggCache::instance()->value(dbm_handle.get()))
                            ->bitmap;
            if (cached_version == std::get<2>(bmk)) {
                g_contains_agg_with_cache_if_eligible_full_hit << 1;
            } else {
                g_contains_agg_with_cache_if_eligible_partial_hit << 1;
            }
            if (cached_dbm.contains(row_id)) {
                return true;
            }
            if (cached_version == std::get<2>(bmk)) {
                return false;
            }
            start_version = cached_version + 1;
        }
    }
    DeleteBitmap::BitmapKey start {std::get<0>(bmk), std::get<1>(bmk), start_version};
    std::shared_lock l(lock);
    for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
        auto& [k, bm] = *it;
        if (std::get<0>(k) != std::get<0>(bmk) || std::get<1>(k) != std::get<1>(bmk) ||
            std::get<2>(k) > std::get<2>(bmk)) {
            break;
        }
        if (bm.contains(row_id)) {
            return true;
        }
    }
    return false;
}

void DeleteBitmap::remove_sentinel_marks() {
    std::lock_guard l(lock);
    for (auto it = delete_bitmap.begin(), end = delete_bitmap.end(); it != end;) {
        if (std::get<1>(it->first) == DeleteBitmap::INVALID_SEGMENT_ID) {
            it = delete_bitmap.erase(it);
        } else {
            ++it;
        }
    }
}

int DeleteBitmap::set(const BitmapKey& bmk, const roaring::Roaring& segment_delete_bitmap) {
    std::lock_guard l(lock);
    auto [_, inserted] = delete_bitmap.insert_or_assign(bmk, segment_delete_bitmap);
    return inserted;
}

int DeleteBitmap::get(const BitmapKey& bmk, roaring::Roaring* segment_delete_bitmap) const {
    std::shared_lock l(lock);
    auto it = delete_bitmap.find(bmk);
    if (it == delete_bitmap.end()) return -1;
    *segment_delete_bitmap = it->second; // copy
    return 0;
}

const roaring::Roaring* DeleteBitmap::get(const BitmapKey& bmk) const {
    std::shared_lock l(lock);
    auto it = delete_bitmap.find(bmk);
    if (it == delete_bitmap.end()) return nullptr;
    return &(it->second); // get address
}

void DeleteBitmap::subset(const BitmapKey& start, const BitmapKey& end,
                          DeleteBitmap* subset_rowset_map) const {
    DCHECK(start < end);
    std::shared_lock l(lock);
    for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
        auto& [k, bm] = *it;
        if (k >= end) {
            break;
        }
        subset_rowset_map->set(k, bm);
    }
}

void DeleteBitmap::subset(std::vector<std::pair<RowsetId, int64_t>>& rowset_ids,
                          int64_t start_version, int64_t end_version,
                          DeleteBitmap* subset_delete_map) const {
    DCHECK(start_version <= end_version);
    for (auto& [rowset_id, _] : rowset_ids) {
        BitmapKey start {rowset_id, 0, 0};
        BitmapKey end {rowset_id, UINT32_MAX, end_version + 1};
        std::shared_lock l(lock);
        for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
            auto& [k, bm] = *it;
            if (k >= end) {
                break;
            }
            auto version = std::get<2>(k);
            if (version >= start_version && version <= end_version) {
                subset_delete_map->merge(k, bm);
                VLOG_DEBUG << "subset delete bitmap, tablet=" << _tablet_id << ", version=["
                           << start_version << ", " << end_version
                           << "]. rowset=" << std::get<0>(k).to_string()
                           << ", segment=" << std::get<1>(k) << ", version=" << version
                           << ", cardinality=" << bm.cardinality();
            }
        }
    }
}

void DeleteBitmap::subset_and_agg(std::vector<std::pair<RowsetId, int64_t>>& rowset_ids,
                                  int64_t start_version, int64_t end_version,
                                  DeleteBitmap* subset_delete_map) const {
    DCHECK(start_version <= end_version);
    for (auto& [rowset_id, segment_num] : rowset_ids) {
        for (int64_t seg_id = 0; seg_id < segment_num; ++seg_id) {
            BitmapKey end {rowset_id, seg_id, end_version};
            auto bm = get_agg_without_cache(end, start_version);
            VLOG_DEBUG << "subset delete bitmap, tablet=" << _tablet_id << ", rowset=" << rowset_id
                       << ", segment=" << seg_id << ", version=[" << start_version << "-"
                       << end_version << "], cardinality=" << bm->cardinality();
            if (bm->isEmpty()) {
                continue;
            }
            subset_delete_map->merge(end, *bm);
        }
    }
}

size_t DeleteBitmap::get_count_with_range(const BitmapKey& start, const BitmapKey& end) const {
    DCHECK(start < end);
    size_t count = 0;
    std::shared_lock l(lock);
    for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
        auto& [k, bm] = *it;
        if (k >= end) {
            break;
        }
        count++;
    }
    return count;
}

void DeleteBitmap::merge(const BitmapKey& bmk, const roaring::Roaring& segment_delete_bitmap) {
    std::lock_guard l(lock);
    auto [iter, succ] = delete_bitmap.emplace(bmk, segment_delete_bitmap);
    if (!succ) {
        iter->second |= segment_delete_bitmap;
    }
}

void DeleteBitmap::merge(const DeleteBitmap& other) {
    std::lock_guard l(lock);
    for (auto& i : other.delete_bitmap) {
        auto [j, succ] = this->delete_bitmap.insert(i);
        if (!succ) j->second |= i.second;
    }
}

uint64_t DeleteBitmap::get_delete_bitmap_count() {
    std::shared_lock l(lock);
    uint64_t count = 0;
    for (auto it = delete_bitmap.begin(); it != delete_bitmap.end(); it++) {
        if (std::get<1>(it->first) != DeleteBitmap::INVALID_SEGMENT_ID) {
            count++;
        }
    }
    return count;
}

void DeleteBitmap::traverse_rowset_and_version(
        const std::function<int(const RowsetId& rowsetId, int64_t version)>& func) const {
    std::shared_lock l(lock);
    auto it = delete_bitmap.cbegin();
    while (it != delete_bitmap.cend()) {
        RowsetId rowset_id = std::get<0>(it->first);
        int64_t version = std::get<2>(it->first);
        int result = func(rowset_id, version);
        if (result == -2) {
            // find next <rowset, version>
            it++;
        } else {
            // find next <rowset>
            it = delete_bitmap.upper_bound({rowset_id, std::numeric_limits<SegmentId>::max(),
                                            std::numeric_limits<Version>::max()});
        }
    }
}

bool DeleteBitmap::has_calculated_for_multi_segments(const RowsetId& rowset_id) const {
    return contains({rowset_id, INVALID_SEGMENT_ID, TEMP_VERSION_COMMON}, ROWSET_SENTINEL_MARK);
}

size_t DeleteBitmap::remove_rowset_cache_version(const RowsetId& rowset_id) {
    std::lock_guard l(_rowset_cache_version_lock);
    _rowset_cache_version.erase(rowset_id);
    VLOG_DEBUG << "remove agg cache version for tablet=" << _tablet_id
               << ", rowset=" << rowset_id.to_string();
    return _rowset_cache_version.size();
}

void DeleteBitmap::clear_rowset_cache_version() {
    std::lock_guard l(_rowset_cache_version_lock);
    _rowset_cache_version.clear();
    VLOG_DEBUG << "clear agg cache version for tablet=" << _tablet_id;
}

std::set<std::string> DeleteBitmap::get_rowset_cache_version() {
    std::set<std::string> set;
    std::shared_lock l(_rowset_cache_version_lock);
    for (auto& [k, _] : _rowset_cache_version) {
        set.insert(k.to_string());
    }
    return set;
}

DeleteBitmap::Version DeleteBitmap::_get_rowset_cache_version(const BitmapKey& bmk) const {
    std::shared_lock l(_rowset_cache_version_lock);
    if (auto it = _rowset_cache_version.find(std::get<0>(bmk)); it != _rowset_cache_version.end()) {
        auto& segment_cache_version = it->second;
        if (auto it1 = segment_cache_version.find(std::get<1>(bmk));
            it1 != segment_cache_version.end()) {
            return it1->second;
        }
    }
    return 0;
}

DeleteBitmap DeleteBitmap::agg_cache_snapshot() {
    return DeleteBitmapAggCache::instance()->snapshot(_tablet_id);
}

void DeleteBitmap::set_tablet_id(int64_t tablet_id) {
    _tablet_id = tablet_id;
}

std::shared_ptr<roaring::Roaring> DeleteBitmap::get_agg(const BitmapKey& bmk) const {
    std::string key_str = agg_cache_key(_tablet_id, bmk); // Cache key container
    CacheKey key(key_str);
    Cache::Handle* handle = DeleteBitmapAggCache::instance()->lookup(key);

    DeleteBitmapAggCache::Value* val =
            handle == nullptr ? nullptr
                              : reinterpret_cast<DeleteBitmapAggCache::Value*>(
                                        DeleteBitmapAggCache::instance()->value(handle));
    // FIXME: do we need a mutex here to get rid of duplicated initializations
    //        of cache entries in some cases?
    if (val == nullptr) { // Renew if needed, put a new Value to cache
        val = new DeleteBitmapAggCache::Value();
        Version start_version =
                config::enable_mow_get_agg_by_cache ? _get_rowset_cache_version(bmk) : 0;
        if (start_version > 0) {
            Cache::Handle* handle2 = DeleteBitmapAggCache::instance()->lookup(
                    agg_cache_key(_tablet_id, {std::get<0>(bmk), std::get<1>(bmk), start_version}));

            DBUG_EXECUTE_IF("DeleteBitmap::get_agg.cache_miss", {
                if (handle2 != nullptr) {
                    auto p = dp->param("percent", 0.3);
                    std::mt19937 gen {std::random_device {}()};
                    std::bernoulli_distribution inject_fault {p};
                    if (inject_fault(gen)) {
                        LOG_INFO("injection DeleteBitmap::get_agg.cache_miss, tablet_id={}",
                                 _tablet_id);
                        handle2 = nullptr;
                    }
                }
            });
            if (handle2 == nullptr || start_version > std::get<2>(bmk)) {
                start_version = 0;
            } else {
                val->bitmap |= reinterpret_cast<DeleteBitmapAggCache::Value*>(
                                       DeleteBitmapAggCache::instance()->value(handle2))
                                       ->bitmap;
                VLOG_DEBUG << "get agg cache version=" << start_version
                           << " for tablet=" << _tablet_id
                           << ", rowset=" << std::get<0>(bmk).to_string()
                           << ", segment=" << std::get<1>(bmk);
                start_version += 1;
            }
            if (handle2 != nullptr) {
                DeleteBitmapAggCache::instance()->release(handle2);
            }
        }
        {
            std::shared_lock l(lock);
            DeleteBitmap::BitmapKey start {std::get<0>(bmk), std::get<1>(bmk), start_version};
            for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
                auto& [k, bm] = *it;
                if (std::get<0>(k) != std::get<0>(bmk) || std::get<1>(k) != std::get<1>(bmk) ||
                    std::get<2>(k) > std::get<2>(bmk)) {
                    break;
                }
                val->bitmap |= bm;
            }
        }
        size_t charge = val->bitmap.getSizeInBytes() + sizeof(DeleteBitmapAggCache::Value);
        handle = DeleteBitmapAggCache::instance()->insert(key, val, charge, charge,
                                                          CachePriority::NORMAL);
        if (config::enable_mow_get_agg_by_cache && !val->bitmap.isEmpty()) {
            std::lock_guard l(_rowset_cache_version_lock);
            // this version is already agg
            _rowset_cache_version[std::get<0>(bmk)][std::get<1>(bmk)] = std::get<2>(bmk);
            VLOG_DEBUG << "set agg cache version=" << std::get<2>(bmk)
                       << " for tablet=" << _tablet_id
                       << ", rowset=" << std::get<0>(bmk).to_string()
                       << ", segment=" << std::get<1>(bmk);
        }
        if (start_version > 0 && config::enable_mow_get_agg_correctness_check_core) {
            std::shared_ptr<roaring::Roaring> bitmap = get_agg_without_cache(bmk);
            if (val->bitmap != *bitmap) {
                CHECK(false) << ". get agg correctness check failed for tablet=" << _tablet_id
                             << ", rowset=" << std::get<0>(bmk).to_string()
                             << ", segment=" << std::get<1>(bmk) << ", version=" << std::get<2>(bmk)
                             << ". start_version from cache=" << start_version
                             << ", delete_bitmap cardinality with cache="
                             << val->bitmap.cardinality()
                             << ", delete_bitmap cardinality without cache="
                             << bitmap->cardinality();
            }
        }
    }

    // It is natural for the cache to reclaim the underlying memory
    return std::shared_ptr<roaring::Roaring>(
            &val->bitmap, [handle](...) { DeleteBitmapAggCache::instance()->release(handle); });
}

std::shared_ptr<roaring::Roaring> DeleteBitmap::get_agg_without_cache(
        const BitmapKey& bmk, const int64_t start_version) const {
    std::shared_ptr<roaring::Roaring> bitmap = std::make_shared<roaring::Roaring>();
    std::shared_lock l(lock);
    DeleteBitmap::BitmapKey start {std::get<0>(bmk), std::get<1>(bmk), start_version};
    for (auto it = delete_bitmap.lower_bound(start); it != delete_bitmap.end(); ++it) {
        auto& [k, bm] = *it;
        if (std::get<0>(k) != std::get<0>(bmk) || std::get<1>(k) != std::get<1>(bmk) ||
            std::get<2>(k) > std::get<2>(bmk)) {
            break;
        }
        *bitmap |= bm;
    }
    return bitmap;
}

DeleteBitmap DeleteBitmap::diffset(const std::set<BitmapKey>& key_set) const {
    std::shared_lock l(lock);
    auto diff_key_set_view =
            delete_bitmap | std::ranges::views::transform([](const auto& kv) { return kv.first; }) |
            std::ranges::views::filter(
                    [&key_set](const auto& key) { return !key_set.contains(key); });

    DeleteBitmap dbm(_tablet_id);
    for (const auto& key : diff_key_set_view) {
        const auto* bitmap = get(key);
        DCHECK_NE(bitmap, nullptr);
        dbm.delete_bitmap[key] = *bitmap;
    }
    return dbm;
}

std::string tablet_state_name(TabletState state) {
    switch (state) {
    case TABLET_NOTREADY:
        return "TABLET_NOTREADY";

    case TABLET_RUNNING:
        return "TABLET_RUNNING";

    case TABLET_TOMBSTONED:
        return "TABLET_TOMBSTONED";

    case TABLET_STOPPED:
        return "TABLET_STOPPED";

    case TABLET_SHUTDOWN:
        return "TABLET_SHUTDOWN";

    default:
        return "TabletState(" + std::to_string(state) + ")";
    }
}

} // namespace doris

Coverage Report

Created: 2026-08-14 23:14