// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
// Modifications Copyright 2022 PingCAP, Inc.

use std::{
    cmp,
    collections::{BTreeMap, HashMap, HashSet},
    fmt,
    iter::Iterator,
    ops::Deref,
    sync::{
        Arc, Mutex, RwLock,
        atomic::{AtomicBool, AtomicU64, Ordering::*, *},
    },
    time::Duration,
};

use api_version::{ApiV2, api_v2::KEYSPACE_PREFIX_LEN};
use bytes::{Buf, BufMut, Bytes};
use cloud_encryption::{
    EncryptionKey, EncryptionKeyManager, EncryptionMethod, EncryptionState, MasterKey,
    encryption_state_from_meta_bytes,
};
use kvenginepb::{self as pb, TxnFileRef};
use quick_cache::sync::Cache;
use rand::Rng;
use schema::schema::StorageClassSpec;
use slog_global::*;
#[cfg(any(test, feature = "testexport"))]
use tikv_util::HandyRwLock;
use tikv_util::{box_err, box_try, time::Instant};
use txn_types::{Key, TimeStamp};

use crate::{
    context::{IaCtx, MetaFileCacheWeighter, PrepareType, SnapCtx},
    dfs::FileType,
    ia::{
        ia_auto_file::TransitResult, ia_file::IaFile, manager::IaManager, types::FileSegmentIdent,
    },
    limiter::RegionLimiter,
    metrics::{ENGINE_COLUMNAR_TOO_MANY_UNCONVERTED_L0S, HALF_SPLIT_TABLES_HISTOGRAM},
    prepare::FilePrepareType,
    table::{
        self, BoundedDataSet, DataBound, InnerKey, OwnedInnerKey, SnapVersion, TxnFile,
        blobtable::blobtable::BlobTable,
        columnar::{ColumnarLevel, ColumnarLevels, VectorIndexDef},
        file::InMemFile,
        fts::FtsLevels,
        get_local_dir,
        memtable::{self, CfTable, MemTableImplementation},
        schema_file::SchemaFile,
        search,
        sstable::{L0Table, SsTable},
        tiny_meta::TypedTinyMeta,
        vector_index::{VectorIndex, VectorIndexes},
    },
    util::{TxnFileRefPropertyHelper, evenly_distribute},
    *,
};

#[derive(Clone, Copy, PartialEq)]
pub(crate) enum MajorCompactionType {
    Disable,
    Enable,
    EnableColumnar,
}

impl From<&[u8]> for MajorCompactionType {
    fn from(s: &[u8]) -> Self {
        match s {
            MANUAL_MAJOR_COMPACTION_ENABLE_COLUMNAR => MajorCompactionType::EnableColumnar,
            MANUAL_MAJOR_COMPACTION_ENABLE => MajorCompactionType::Enable,
            _ => MajorCompactionType::Disable,
        }
    }
}

#[derive(Clone)]
pub(crate) struct ShardPendingOperations {
    pub(crate) del_prefixes: Arc<DeletePrefixes>,
    pub(crate) truncate_ts: Option<u64>,
    pub(crate) trim_over_bound: bool,
    pub(crate) manual_major_compaction: MajorCompactionType,
    pub(crate) storage_class_spec: StorageClassSpec,
    pub(crate) gc_lock_files: HashMap<u64, SnapVersion>,
}

impl ShardPendingOperations {
    fn new(keyspace_id: u32) -> Self {
        Self {
            del_prefixes: Arc::new(DeletePrefixes::new_with_keyspace_id(keyspace_id)),
            truncate_ts: None,
            trim_over_bound: false,
            manual_major_compaction: MajorCompactionType::Disable,
            storage_class_spec: StorageClassSpec::default(),
            gc_lock_files: HashMap::new(),
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PendingFileType {
    ChangeSet,
    TxnFile,
}

#[derive(Debug, Clone)]
pub struct PendingFileInfo {
    pub file_type: PendingFileType,
    pub file_ids: Vec<u64>,
}

/// A half-split key selected from loaded WRITE SST metadata.
pub enum ApproxSplitKey {
    /// The key is the first key of an SST.
    Sst(Key),
    /// The key is the first key of a data block in one SST.
    Block(Key),
}

pub struct Shard {
    pub engine_id: u64,
    pub id: u64,
    pub ver: u64,
    pub range: ShardRange,
    pub parent_id: u64,
    pub(crate) data: RwLock<ShardData>,
    pub(crate) opt: Arc<Options>,

    // If the shard is not active, flush mem table and do compaction will ignore this shard.
    pub(crate) active: AtomicBool,

    pub(crate) properties: Properties,
    pub(crate) pending_ops: RwLock<ShardPendingOperations>,
    pub(crate) compacting: AtomicBool, // for statistics only
    pub(crate) initial_flushed: AtomicBool,

    pub(crate) base_version: AtomicU64,
    // Size of all SSTables + all blobs referred (note, not the blob table size).
    pub(crate) estimated_size: AtomicU64,
    pub(crate) estimated_entries: AtomicU64,
    pub(crate) max_ts: AtomicU64,
    // Logical KV payload size of all row-format SSTables (L0 + CF levels).
    pub(crate) estimated_kv_size: AtomicU64,
    pub(crate) estimated_ia_kv_size: AtomicU64,
    // Physical file size of all columnar files (L0/L1/L2).
    pub(crate) estimated_columnar_size: AtomicU64,
    // Uncompressed logical data size of all columnar files (analogous to SST kv_size).
    pub(crate) estimated_columnar_kv_size: AtomicU64,
    pub(crate) store_usage_stats: RwLock<ShardStoreUsageStats>,

    pub(crate) sst_max_ts: AtomicU64, // the max_ts of sst files (mem-tables excluded)

    // Used for GC major compaction. Stats are collected from WRITE_CF & level 2+ only.
    pub(crate) lv2plus_max_ts: AtomicU64, // the max_ts of sst files
    pub(crate) lv2plus_tombs: AtomicU64,  // number of tombstone entries
    pub(crate) lv2plus_entries_write_cf: AtomicU64, // number of entries in WRITE_CF
    pub(crate) lv2plus_data_size: AtomicU64,
    pub(crate) lv2plus_old_data_size: AtomicU64,
    pub(crate) lv2plus_old_entries: AtomicU64, // number of old-version entries

    // File-level GC wakeup hint. Min candidate `max_ts` among files passing the
    // cheap static gates (ratio >= threshold and lower-level overlap within
    // limit; L0 must be converted).
    // `u64::MAX` = no candidate, or parked by a scan that found every matured
    // candidate overlap-gated; either way the high-frequency leader tick skips
    // the WRITE_CF metadata walk with an O(1) compare. A refresh recomputes it
    // from the live layout, so a newly reclaimable file reopens the scan.
    pub(crate) file_gc_candidate_min_max_ts: AtomicU64,

    // Used for gc lock cf files.
    pub(crate) lv1plus_min_lock_file_max_ts: AtomicU64, // the min max_ts of all lock cf sst files.

    // Used for gc extra cf files.
    pub(crate) lv1_min_max_extra_ts: AtomicU64, // the min max_ts of all extra cf sst files.

    // meta_seq is the raft log index of the applied change set.
    // Because change set are applied in the worker thread, the value is usually smaller
    // than write_sequence.
    pub(crate) meta_seq: AtomicU64,

    // write_sequence is the raft log index of the applied write batch.
    pub(crate) write_sequence: AtomicU64,

    pub(crate) compaction_priority: RwLock<Option<CompactionPriority>>,

    pub(crate) encryption_state: RwLock<EncryptionState>,

    // outdated_schema_ver is used to record the last schema outdated error in convert L0 to
    // columnar if the schema is not updated, we skip retrying the convert.
    pub(crate) outdated_schema_ver: AtomicI64,

    pub(crate) checked_schema_ver: AtomicI64,
    pub(crate) last_transit_storage_class_instant_sec: AtomicI64,

    /// Tracks files that are being processed but not yet applied. The key is
    /// the raft log sequence number, and the value contains file type
    /// (ChangeSet/TxnFile) and file IDs.
    ///
    /// Insert timing:
    /// - During prepare of ChangeSet/TxnFile, before downloading files
    ///
    /// Cleanup timing:
    /// - After successful apply of ChangeSet/TxnFile
    /// - During apply of subsequent operations (cleans up files from aborted
    ///   ChangeSet/TxnFile operations), files with sequence numbers <= current
    ///   operation's sequence are cleaned
    ///
    /// This tracking is critical for:
    /// 1. Protecting files from GC during the download -> apply lifecycle
    /// 2. Performance impact is minimal as the map is usually empty in normal
    ///    cases
    pub(crate) unapplied_pending_files: Mutex<BTreeMap<u64, PendingFileInfo>>,

    #[cfg(any(test, feature = "testexport"))]
    pub(crate) max_used_gc_safe_point: RwLock<TimeStamp>,
}

// Note: when add new property, consider whether to add it to following process:
// * `PropertiesHelper`
// * `is_property_need_initial_flush`
// * `is_property_need_flush`
// * `is_property_change_set`
// * `PROPERTIES_COPY_FROM_PARENT_IN_RECOVERY`

// Following properties are maintained by Shard, and should be flushed to
// `ShardMeta` for persistence.
pub const TERM_KEY: &str = "term";
pub const INGEST_ID_KEY: &str = "_ingest_id";
pub const TRIM_OVER_BOUND: &str = "_trim_over_bound";
pub const MANUAL_MAJOR_COMPACTION: &str = "_manual_major_compaction";
pub const TXN_FILE_REF: &str = "_txn_file_ref";

// Following properties are maintained by ShardMeta, and should be skipped
// during flush.
pub const DEL_PREFIXES_KEY: &str = "_del_prefixes";
pub const ENCRYPTION_KEY: &str = "_encryption";
pub const ENCRYPTION_META_KEY: &str = "_encryption_meta";
pub const STORAGE_CLASS_KEY: &str = "_storage_class";

// Note: TERM_KEY should not be flushed during initial flush, to keep
// `ShardMeta.data_sequence` consistent with `TERM_KEY`.
#[inline]
pub fn is_property_need_initial_flush(key: &str) -> bool {
    matches!(
        key,
        INGEST_ID_KEY | TRIM_OVER_BOUND | MANUAL_MAJOR_COMPACTION | TXN_FILE_REF
    )
}

#[inline]
pub fn is_property_need_flush(key: &str) -> bool {
    matches!(key, TERM_KEY) || is_property_need_initial_flush(key)
}

// The properties must persist before advance data sequence.
// See `ShardData::all_persisted` and `PeerMsgHandler::on_raft_log_gc_tick`.
// Some properties in `is_property_need_flush` are not considered as "must
// persist" for performance:
// - TERM_KEY: Specially handle in `PeerMsgHandler::on_raft_log_gc_tick`.
// - INGEST_ID_KEY: deprecated for legacy import.
// - TRIM_OVER_BOUND: safe to ignore.
// - MANUAL_MAJOR_COMPACTION: safe to ignore.
#[inline]
pub fn is_property_must_persist(key: &str) -> bool {
    matches!(key, TXN_FILE_REF)
}

// Properties will change in parent shard during recovery.
// See `Shard::add_parent_data`.
const PROPERTIES_COPY_FROM_PARENT_IN_RECOVERY: &[&str] = &[DEL_PREFIXES_KEY, TXN_FILE_REF];

pub const TRIM_OVER_BOUND_ENABLE: &[u8] = &[1];
pub const TRIM_OVER_BOUND_DISABLE: &[u8] = b"";

pub const MANUAL_MAJOR_COMPACTION_ENABLE: &[u8] = &[1];
pub const MANUAL_MAJOR_COMPACTION_ENABLE_COLUMNAR: &[u8] = &[2];
pub const MANUAL_MAJOR_COMPACTION_DISABLE: &[u8] = b"";

pub(crate) const INITIAL_UPDATE_COUNTER: u64 = 0;

const MAX_COL_L0_FILE_COUNTS: usize = 32;
const MAX_COL_L1_FILE_COUNTS: usize = 64;
const MAX_UNCONVERTED_L0_FILE_COUNTS: usize = 32;

fn is_same_pending_encryption_state(
    current: &EncryptionState,
    resolved_from: &EncryptionState,
) -> bool {
    match (current, resolved_from) {
        (
            EncryptionState::Pending {
                keyspace_id: current_keyspace_id,
                data_key_id: current_data_key_id,
                encryption_meta: current_meta,
                ..
            },
            EncryptionState::Pending {
                keyspace_id,
                data_key_id,
                encryption_meta,
                ..
            },
        ) => {
            current_keyspace_id == keyspace_id
                && current_data_key_id == data_key_id
                && Arc::ptr_eq(current_meta, encryption_meta)
        }
        _ => false,
    }
}

impl Deref for Shard {
    type Target = ShardRange;

    fn deref(&self) -> &Self::Target {
        &self.range
    }
}

impl Shard {
    pub fn new(
        engine_id: u64,
        props: &pb::Properties,
        ver: u64,
        range: ShardRange,
        inner_key_off: usize,
        opt: Arc<Options>,
        master_key: &MasterKey,
        encryption_key_manager: Arc<EncryptionKeyManager>,
    ) -> Self {
        let encryption_state = encryption_state_from_shard_properties(
            props,
            encryption_key_manager.clone(),
            master_key,
            Some(range.keyspace_id),
        );
        let _ = encryption_key_manager
            .register_current_encryption_state_for_shard(props.shard_id, &encryption_state);
        let limiter = RegionLimiter::new(
            opt.flow_control.region_memtable_limiter_options(),
            opt.flow_control.region_l0table_limiter_options(),
        );
        let memtable_implementation =
            MemTableImplementation::from_crossbeam_enabled(opt.enable_crossbeam_memtable);
        let now = Instant::now_coarse();
        let shard = Self {
            engine_id,
            id: props.shard_id,
            ver,
            range: range.clone(),
            parent_id: 0,
            pending_ops: RwLock::new(ShardPendingOperations::new(range.keyspace_id)),
            data: RwLock::new(ShardData::new_empty(
                range,
                inner_key_off,
                limiter,
                memtable_implementation,
            )),
            opt,
            active: Default::default(),
            properties: Properties::new().apply_pb(props),
            compacting: Default::default(),
            initial_flushed: Default::default(),
            base_version: Default::default(),
            estimated_size: Default::default(),
            estimated_entries: Default::default(),
            max_ts: Default::default(),
            estimated_kv_size: Default::default(),
            estimated_ia_kv_size: Default::default(),
            estimated_columnar_size: Default::default(),
            estimated_columnar_kv_size: Default::default(),
            store_usage_stats: RwLock::new(ShardStoreUsageStats::default()),
            sst_max_ts: Default::default(),
            lv2plus_max_ts: Default::default(),
            lv2plus_tombs: Default::default(),
            lv2plus_entries_write_cf: Default::default(),
            lv2plus_data_size: Default::default(),
            lv2plus_old_data_size: Default::default(),
            lv2plus_old_entries: Default::default(),
            file_gc_candidate_min_max_ts: AtomicU64::new(u64::MAX),
            lv1plus_min_lock_file_max_ts: Default::default(),
            lv1_min_max_extra_ts: Default::default(),
            meta_seq: Default::default(),
            write_sequence: Default::default(),
            compaction_priority: RwLock::new(None),
            encryption_state: RwLock::new(encryption_state),
            outdated_schema_ver: Default::default(),
            checked_schema_ver: Default::default(),
            last_transit_storage_class_instant_sec: AtomicI64::new(now.second()),
            unapplied_pending_files: Mutex::new(BTreeMap::new()),
            #[cfg(any(test, feature = "testexport"))]
            max_used_gc_safe_point: RwLock::new(0.into()),
        };
        {
            let mut pending_ops = shard.pending_ops.write().unwrap();
            if let Some(val) = get_shard_property(DEL_PREFIXES_KEY, props) {
                let mut del_prefixes = DeletePrefixes::unmarshal(&val, shard.keyspace_id);
                del_prefixes.schedule_at = shard.gen_rand_schedule_del_range_time();
                pending_ops.del_prefixes = Arc::new(del_prefixes);
            }
            if let Some(val) = get_shard_property(TRIM_OVER_BOUND, props) {
                if !val.is_empty() {
                    pending_ops.trim_over_bound = true;
                }
            }
            if let Some(val) = get_shard_property(MANUAL_MAJOR_COMPACTION, props) {
                if !val.is_empty() {
                    pending_ops.manual_major_compaction = MajorCompactionType::from(val.as_slice());
                }
            }
            if let Some(val) = get_shard_property(STORAGE_CLASS_KEY, props) {
                pending_ops.storage_class_spec = StorageClassSpec::unmarshal(Some(&val));
            }
        }
        shard
    }

    pub fn new_for_ingest(
        engine_id: u64,
        cs: &pb::ChangeSet,
        opt: Arc<Options>,
        master_key: &MasterKey,
        encryption_key_manager: Arc<EncryptionKeyManager>,
    ) -> Self {
        let snap = cs.get_snapshot();
        let range = ShardRange::from_snap(snap);
        let mut shard = Self::new(
            engine_id,
            snap.get_properties(),
            cs.shard_ver,
            range,
            snap.inner_key_off as usize,
            opt,
            master_key,
            encryption_key_manager,
        );
        if !cs.has_parent() {
            store_bool(&shard.initial_flushed, true);
        } else {
            shard.parent_id = cs.get_parent().shard_id;
        }
        shard.base_version.store(snap.base_version, Release);
        shard.meta_seq.store(cs.sequence, Release);
        shard.write_sequence.store(snap.data_sequence, Release);
        shard
    }

    fn collect_ids_from_snapshot(
        snap: &pb::Snapshot,
        prepare_type: PrepareType,
    ) -> HashMap<u64, FileMeta> {
        let prepare_sst = matches!(prepare_type, PrepareType::SstOnly | PrepareType::All);
        let prepare_columnar = matches!(prepare_type, PrepareType::ColumnarOnly | PrepareType::All);

        let mut ids = HashMap::new();
        let unconverted_l0s: HashSet<u64> = snap.get_unconverted_l0s().iter().copied().collect();
        for l0 in snap.get_l0_creates() {
            // If type is prepare columnar only, unconverted l0s should also be prepared.
            if prepare_columnar && !prepare_sst && !unconverted_l0s.contains(&l0.id) {
                continue;
            }
            ids.insert(l0.id, FileMeta::from_l0_table(l0));
        }
        if prepare_sst {
            for ln in snap.get_table_creates() {
                ids.insert(ln.id, FileMeta::from_table(ln));
            }
            for blob in snap.get_blob_creates() {
                ids.insert(blob.id, FileMeta::from_blob_table(blob));
            }
        }
        if prepare_columnar {
            for columnar in snap.get_columnar_creates() {
                ids.insert(columnar.id, FileMeta::from_columnar_table(columnar));
            }
            // `file_id` in shard meta will set to `0` when the columnar replica was
            // removed. We also need to check if the schema file is valid.
            if snap.has_schema_meta() && snap.get_schema_meta().get_file_id() > 0 {
                ids.insert(
                    snap.get_schema_meta().get_file_id(),
                    FileMeta::from_schema_meta(),
                );
            }
            for vec_index in snap.get_vector_indexes() {
                for f in vec_index.files.iter() {
                    ids.insert(f.id, FileMeta::from_vector_index_file(f));
                }
            }
            for l0_file in snap.get_fts_l0_files() {
                ids.insert(l0_file.id, FileMeta::from_fts_l0_file(l0_file));
            }
            for l1_file in snap.get_fts_l1_files() {
                ids.insert(l1_file.id, FileMeta::from_fts_l1_file(l1_file));
            }
            for l2_file in snap.get_fts_l2_files() {
                ids.insert(l2_file.id, FileMeta::from_fts_l2_file(l2_file));
            }
        }
        ids
    }

    /// Constructs a shard from a prepared change set without resolving pending
    /// encryption keys. Use a key-ready snapshot entry point before accessing
    /// encrypted data.
    pub async fn from_change_set(
        tag: &str,
        ctx: &SnapCtx,
        change_set: pb::ChangeSet,
        mut mem_tbls: Vec<CfTable>,
        write_cf_only: bool,
    ) -> Result<Self> {
        let mut cs = ChangeSet::new(change_set);
        let mut ids = HashMap::new();
        let mut lock_txn_file_refs: Vec<TxnFileRef> = vec![];
        if mem_tbls.is_empty() {
            mem_tbls.push(CfTable::new());
        }
        let mut shard_prepare_type = FilePrepareType::default();
        let encryption_state = if cs.has_snapshot() {
            let snap = cs.get_snapshot();
            shard_prepare_type = FilePrepareType::from_snapshot(snap);
            ids = Self::collect_ids_from_snapshot(snap, ctx.prepare_type);
            if !write_cf_only {
                lock_txn_file_refs = collect_snap_lock_txn_file_refs(snap);
            }
            encryption_state_from_shard_properties(
                snap.get_properties(),
                ctx.encryption_key_manager.clone(),
                &ctx.master_key,
                ApiV2::get_u32_keyspace_id_by_key(snap.get_outer_start()),
            )
        } else {
            EncryptionState::Disabled
        };
        // Non-IA regions may still use IA prepare when worker IA is enabled,
        // but only snapshots from IA/Auto-IA regions consume IA remote-read throttle.
        let charge_ia_remote_read = shard_prepare_type.not_local();
        if let IaCtx::Enabled(ia_mgr, _) = &ctx.ia_ctx {
            let mut added_files = HashSet::new();
            // For cached files, we prepare with the meta data directly to reduce latency.
            for (id, fm) in &ids {
                let file = if fm.can_use_ia() && charge_ia_remote_read {
                    ctx.meta_file_cache.get(id)
                } else if fm.is_l0_sst_with_size() {
                    // Cache the whole file as a segment.
                    let ident = FileSegmentIdent::new(*id, 0, fm.l0_size as u64);
                    if ia_mgr.is_segment_cached(ident) {
                        let handle = ia_mgr
                            .get_segment_handle(
                                ident,
                                fm.file_type,
                                Some(ctx.dfs.deref()),
                                charge_ia_remote_read,
                            )
                            .await?;
                        Some(handle.into_inner())
                    } else {
                        None
                    }
                } else {
                    None
                };
                if let Some(file) = file {
                    cs.add_file(
                        *id,
                        file,
                        fm,
                        ctx.block_cache.clone(),
                        ctx.vector_index_cache.clone(),
                        ctx.columnar_file_cache.clone(),
                        ctx.fts_cache.clone(),
                        encryption_state.clone(),
                        ctx.columnar_meta_cache.clone(),
                        TypedTinyMeta::None,
                        None,
                    )?;
                    added_files.insert(*id);
                }
            }
            ids.retain(|id, _| !added_files.contains(id));
        }
        let (result_tx, mut result_rx) = tokio::sync::mpsc::unbounded_channel();
        let runtime = ctx.dfs.get_runtime();
        let opts = dfs::Options::default().with_shard(cs.shard_id, cs.shard_ver);
        let mut msg_count = 0;
        for (&id, fm) in &ids {
            if fm.is_schema_file() {
                if let Some(schema_files) = ctx.schema_files.as_ref() {
                    if let Some(schema_file) = schema_files.get(&id) {
                        cs.set_schema_file(Some(schema_file.clone()));
                        continue;
                    }
                }
            }
            if write_cf_only && fm.get_cf() != WRITE_CF as i32 && fm.get_level() != 0 {
                continue;
            }
            let fs = ctx.dfs.clone();
            let tx = result_tx.clone();
            let fm = fm.clone();
            let tag = tag.to_string();
            let ia_ctx = if fm.can_use_ia() || fm.is_l0_sst_with_size() {
                ctx.ia_ctx.clone()
            } else {
                IaCtx::Disabled(ctx.ia_ctx.transition_limiter())
            };
            let meta_file_cache = ctx.meta_file_cache.clone();
            let strict_file_memory_quota = ctx.strict_file_memory_quota.clone();
            let strict_file_memory_quota_wait_timeout = ctx.strict_file_memory_quota_wait_timeout;
            runtime.spawn(async move {
                let res = Self::prepare_file(
                    id,
                    &fm,
                    fs.as_ref(),
                    &opts,
                    &ia_ctx,
                    meta_file_cache,
                    strict_file_memory_quota,
                    strict_file_memory_quota_wait_timeout,
                    charge_ia_remote_read,
                )
                .await;
                if let Err(err) = tx.send(res.map(|file| (id, fm, file))) {
                    error!("failed to send result"; "tag" => tag, "file_id" => id, "err" => %err);
                }
            });
            msg_count += 1;
        }
        let mut errors: Vec<Error> = vec![];
        for _ in 0..msg_count {
            match result_rx.recv().await.unwrap() {
                Ok((id, fm, file)) => {
                    cs.add_file(
                        id,
                        file,
                        &fm,
                        ctx.block_cache.clone(),
                        ctx.vector_index_cache.clone(),
                        ctx.columnar_file_cache.clone(),
                        ctx.fts_cache.clone(),
                        encryption_state.clone(),
                        ctx.columnar_meta_cache.clone(),
                        TypedTinyMeta::None,
                        None,
                    )?;
                    if fm.is_schema_file() {
                        if let Some(schema_files) = ctx.schema_files.as_ref() {
                            schema_files.insert(id, cs.get_schema_file().unwrap());
                        }
                    }
                }
                Err(err) => {
                    error!("prefetch failed {:?}", &err);
                    errors.push(err);
                }
            }
        }
        if !errors.is_empty() {
            // Preserve MemoryLimitExceeded only when EVERY per-file prefetch
            // failed with that variant. Mixed failures fall back to the
            // generic `box_err!` wrapper — surfacing MemoryLimitExceeded in
            // the mixed case would hide unrelated non-retryable errors, and
            // tikv-server would retry indefinitely until the deadline.
            //
            // cop-worker server.rs maps `Error::MemoryLimitExceeded(_)` to
            // HTTP 503 (retryable) and everything else to 500 (terminal).
            // Collapsing the typed variant via `box_err!` turns transient
            // strict-file-memory-quota pressure into a hard client failure.
            if errors
                .iter()
                .all(|e| matches!(e, Error::MemoryLimitExceeded(_)))
            {
                if let Error::MemoryLimitExceeded(size) = errors[0] {
                    return Err(Error::MemoryLimitExceeded(size));
                }
            }
            return Err(box_err!("errors is not empty: {:?}", errors));
        }

        // Load lock_txn_files:
        if !lock_txn_file_refs.is_empty() {
            let worker_pool = ctx.txn_chunk_manager.worker_pool().clone();
            let shard_id = cs.shard_id;
            let shard_ver = cs.shard_ver;
            let txn_chunk_manager = ctx.txn_chunk_manager.clone();
            let encryption_state = encryption_state.clone();
            cs.lock_txn_files = box_try!(
                worker_pool
                    .spawn_blocking(move || {
                        txn_chunk_manager.load_txn_files_from_refs(
                            shard_id,
                            shard_ver,
                            &lock_txn_file_refs,
                            encryption_state,
                            IoType::ForegroundRead,
                        )
                    })
                    .await
                    .unwrap()
            );
        }

        let mut opts = Options::default();
        opts.read_columnar = ctx.read_columnar;

        let mut shard = Shard::new_for_ingest(
            0,
            &cs,
            Arc::new(opts),
            &ctx.master_key,
            ctx.encryption_key_manager.clone(),
        );
        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_mem_tbls(mem_tbls);
        create_snapshot_tables(
            &mut builder,
            cs.get_snapshot(),
            &cs,
            write_cf_only,
            ctx.prepare_type,
        );
        builder.set_schema(
            cs.get_schema_version(),
            cs.get_restore_version(),
            cs.get_schema_file(),
        );
        shard.id = cs.shard_id;
        shard.set_data(builder.build());
        Ok(shard)
    }

    async fn prepare_file(
        id: u64,
        fm: &FileMeta,
        fs: &dyn dfs::Dfs,
        opts: &dfs::Options,
        ia_ctx: &IaCtx,
        meta_file_cache: Arc<Cache<u64, Arc<dyn table::file::File>, MetaFileCacheWeighter>>,
        strict_file_memory_quota: Arc<tikv_util::memory::MemoryQuota>,
        strict_file_memory_quota_wait_timeout: std::time::Duration,
        charge_ia_remote_read: bool,
    ) -> Result<Arc<dyn table::file::File>> {
        fail::fail_point!("prepare_file_memory_limit_exceeded", |_| {
            Err(Error::MemoryLimitExceeded(0))
        });
        match ia_ctx {
            IaCtx::Disabled(_) => {
                let opts = opts.with_type(fm.file_type);
                let data = match fm.size_hint_for_download() {
                    Some(size) => {
                        let guard = dfs::reserve_quota_async(
                            &strict_file_memory_quota,
                            size as usize,
                            strict_file_memory_quota_wait_timeout,
                        )
                        .await?;
                        let bytes = fs.read_file_with_size_hint(id, size, opts).await?;
                        Bytes::from_owner(dfs::FileWithQuota {
                            bytes,
                            _quota: guard,
                        })
                    }
                    None => {
                        dfs::STRICT_FILE_MEMORY_QUOTA_NOHINT_FALLBACK.inc();
                        fs.read_file(id, opts).await?
                    }
                };
                Ok(Arc::new(InMemFile::new(id, data)) as _)
            }
            IaCtx::Enabled(ia_mgr, data_dirs) => {
                if fm.can_use_ia() {
                    if charge_ia_remote_read {
                        let file = meta_file_cache
                            .get_or_insert_async(&id, async move {
                                let data_dir = get_local_dir(data_dirs, id);
                                let data = IaFile::prepare_table_meta(
                                    id,
                                    fm.file_type,
                                    fm.table_meta_off as u64,
                                    data_dir.as_path(),
                                    opts,
                                    ia_mgr,
                                    Some(fs),
                                )
                                .await?;
                                Self::prepare_ia_file_with_meta_data(
                                    id,
                                    fm,
                                    data,
                                    ia_mgr,
                                    charge_ia_remote_read,
                                )
                            })
                            .await?;
                        Ok(file)
                    } else {
                        let data_dir = get_local_dir(data_dirs, id);
                        let data = IaFile::prepare_table_meta(
                            id,
                            fm.file_type,
                            fm.table_meta_off as u64,
                            data_dir.as_path(),
                            opts,
                            ia_mgr,
                            Some(fs),
                        )
                        .await?;
                        Self::prepare_ia_file_with_meta_data(
                            id,
                            fm,
                            data,
                            ia_mgr,
                            charge_ia_remote_read,
                        )
                    }
                } else if fm.is_l0_sst_with_size() {
                    // Cache the whole file as a segment.
                    let ident = FileSegmentIdent::new(id, 0, fm.l0_size as u64);
                    let handle = ia_mgr
                        .get_segment_handle(ident, fm.file_type, Some(fs), charge_ia_remote_read)
                        .await;
                    handle
                        .map(|handle| handle.into_inner())
                        .map_err(|err| err.into())
                } else {
                    unreachable!()
                }
            }
        }
    }

    fn prepare_ia_file_with_meta_data(
        id: u64,
        fm: &FileMeta,
        data: Bytes,
        ia_mgr: &IaManager,
        charge_ia_remote_read: bool,
    ) -> Result<Arc<dyn table::file::File>> {
        debug_assert!(fm.can_use_ia(), "{}: {:?}", id, fm);
        let table_meta_file = Arc::new(InMemFile::new(id, data));
        let file = IaFile::open(
            id,
            fm,
            table_meta_file,
            ia_mgr.clone(),
            charge_ia_remote_read,
        )?;
        Ok(Arc::new(file))
    }

    pub fn get_cf_total_size(&self, cf: usize) -> u64 {
        let mut total_size = 0;
        let data = self.get_data();
        for cf_tbl in &data.mem_tbls {
            let tbl = cf_tbl.get_cf(cf);
            total_size += tbl.size() as u64;
        }
        let shard_cf = data.get_cf(cf);
        for lh in shard_cf.levels.iter() {
            for tbl in lh.tables.iter() {
                total_size += tbl.kv_size;
            }
        }
        total_size
    }

    pub fn id_ver(&self) -> IdVer {
        IdVer::new(self.id, self.ver)
    }

    pub(crate) fn set_active(&self, active: bool) {
        self.active.store(active, Release);
    }

    pub fn is_active(&self) -> bool {
        self.active.load(Acquire)
    }

    pub(crate) fn refresh_states(&self) {
        self.refresh_estimated_size_and_entries();
        self.refresh_compaction_priority();
    }

    pub fn refresh_estimated_size_and_entries(&self) {
        let data = self.get_data();
        let sc_spec = self.get_storage_class_spec();

        let l0_stats = data.get_l0_stats();
        let mut lv_stats = l0_stats.clone();
        let mut store_usage_stats = ShardStoreUsageStats {
            l0_table_size: l0_stats.data_size,
            ..Default::default()
        };
        data.for_each_level(|cf, l| {
            let level_stats = data.get_level_stats(cf, l, &sc_spec);
            lv_stats.add(&level_stats, cf);
            store_usage_stats.cf_level_sizes[cf][l.level - 1] = level_stats.data_size;
            false
        });

        data.for_each_columnar_level(|cl| {
            lv_stats.add(&data.get_columnar_level_stats(cl.level), WRITE_CF);
            false
        });

        let mut vector_index_size = 0;
        for vi in data.vector_indexes.get_all() {
            for file in &vi.files {
                vector_index_size += file.file_size();
            }
        }

        let mut fts_size = 0;
        for file in data.fts_levels.l0() {
            fts_size += file.file().size();
        }
        for file in data.fts_levels.l1() {
            fts_size += file.file().size();
        }
        for files in data.fts_levels.l2().values() {
            for file in files {
                fts_size += file.file().size();
            }
        }

        store_u64(
            &self.estimated_size,
            lv_stats.data_size + lv_stats.blob_size,
        );
        store_u64(&self.estimated_entries, lv_stats.entries);
        store_u64(&self.sst_max_ts, lv_stats.max_ts);
        store_u64(
            &self.max_ts,
            std::cmp::max(lv_stats.max_ts, data.get_mem_table_max_ts()),
        );
        store_u64(&self.estimated_kv_size, lv_stats.kv_size);
        store_u64(&self.estimated_ia_kv_size, lv_stats.ia_kv_size);
        store_u64(&self.estimated_columnar_size, lv_stats.columnar_size);
        store_u64(&self.estimated_columnar_kv_size, lv_stats.columnar_kv_size);

        store_u64(&self.lv2plus_max_ts, lv_stats.lv2plus_max_ts);
        store_u64(&self.lv2plus_tombs, lv_stats.lv2plus_tombs);
        store_u64(
            &self.lv2plus_entries_write_cf,
            lv_stats.lv2plus_entries_write_cf,
        );
        store_u64(&self.lv2plus_data_size, lv_stats.lv2plus_data_size);
        store_u64(&self.lv2plus_old_data_size, lv_stats.lv2plus_old_data_size);
        store_u64(&self.lv2plus_old_entries, lv_stats.lv2plus_old_entries);
        store_u64(
            &self.lv1plus_min_lock_file_max_ts,
            lv_stats.lv1plus_min_max_lock_ts,
        );
        store_u64(&self.lv1_min_max_extra_ts, lv_stats.lv1_min_max_extra_ts);

        // Row-format catalog only (L0 + CF SST). Columnar / FTS / vector are
        // excluded from store used_size and tracked in separate fields.
        store_usage_stats.row_size = l0_stats.data_size;
        for cf in 0..NUM_CFS {
            for level in 0..CF_LEVELS[cf] {
                store_usage_stats.row_size += store_usage_stats.cf_level_sizes[cf][level];
            }
        }
        store_usage_stats.columnar_size = lv_stats.columnar_size;
        store_usage_stats.fts_size = fts_size;
        store_usage_stats.vector_size = vector_index_size;
        store_usage_stats.ia.data_size = lv_stats.ia_data_size;
        store_usage_stats.ia.kv_size = lv_stats.ia_kv_size;
        store_usage_stats.ia.num_shards = usize::from(sc_spec.is_specified());
        self.classify_write_cf_obsolete(&mut store_usage_stats.obsolete_file_buckets);
        *self.store_usage_stats.write().unwrap() = store_usage_stats;

        data.refresh_for_limiter(&self.tag());

        // Recompute the file-level GC wakeup hint from the just-stored stats and
        // the current file layout, so the leader tick can skip its metadata walk
        // until a candidate matures. Un-parks a hint parked at u64::MAX by an
        // overlap-gated scan, so a newly visible candidate is picked up at once.
        self.refresh_file_gc_candidate_hint(&data);
    }

    pub(crate) fn gen_rand_schedule_del_range_time(&self) -> u64 {
        let now = tikv_util::time::precise_time_ns();
        let delay = rand::thread_rng().gen_range(0..self.opt.max_del_range_delay.as_nanos() as u64);
        now + delay
    }

    pub(crate) fn merge_del_prefix(&self, val: &[u8]) {
        info!("{} Shard::merge_del_prefix: {:?}", self.tag(), val);
        let mut pending_ops = self.pending_ops.write().unwrap();
        let mut del_prefixes = (*pending_ops.del_prefixes).merge_prefix(val);
        del_prefixes.schedule_at = self.gen_rand_schedule_del_range_time();
        pending_ops.del_prefixes = Arc::new(del_prefixes);
    }

    pub(crate) fn set_del_prefixes(&self, val: &[u8]) {
        info!("{} Shard::set_del_prefixes: {:?}", self.tag(), val);
        let mut pending_ops = self.pending_ops.write().unwrap();
        if val.is_empty() {
            pending_ops.del_prefixes =
                Arc::new(DeletePrefixes::new_with_keyspace_id(self.keyspace_id));
            return;
        }
        let mut del_prefixes = DeletePrefixes::unmarshal(val, self.keyspace_id);
        del_prefixes.schedule_at = self.gen_rand_schedule_del_range_time();
        pending_ops.del_prefixes = Arc::new(del_prefixes);
    }

    pub(crate) fn set_trim_over_bound(&self, val: &[u8]) -> bool {
        let mut pending_ops = self.pending_ops.write().unwrap();
        if !val.is_empty() {
            pending_ops.trim_over_bound = true;
            return true;
        }
        false
    }

    pub(crate) fn set_manual_major_compaction(&self, val: &[u8]) -> bool {
        let mut pending_ops = self.pending_ops.write().unwrap();
        pending_ops.manual_major_compaction = MajorCompactionType::from(val);
        !val.is_empty()
    }

    /// Get suggest key for region split.
    pub fn get_suggest_split_key(&self, bucket_size: u64) -> Option<Bytes> {
        let candidate_keys = self.get_candidate_inner_keys(bucket_size);
        if candidate_keys.is_empty() {
            return None;
        }
        let split_idx = candidate_keys.len() * 2 / 3;
        Some(self.to_outer_key(candidate_keys[split_idx].as_ref()))
    }

    fn to_outer_key(&self, inner_key: InnerKey<'_>) -> Bytes {
        [self.keyspace_prefix(), inner_key.deref()].concat().into()
    }

    fn get_candidate_inner_keys(&self, bucket_size: u64) -> Vec<OwnedInnerKey> {
        let data = self.get_data();
        let mut ln_tables = 0;
        data.for_each_level(|_, lvl| {
            ln_tables += lvl.tables.len();
            false
        });
        let num_tables = data.l0_tbls.len() + ln_tables;
        let mut candidate_inner_keys = Vec::with_capacity(num_tables);
        for cf in WRITE_CF..=LOCK_CF {
            let shard_cf = data.get_cf(cf);
            for lvl in &shard_cf.levels {
                for tbl in lvl.tables.iter() {
                    if tbl.smallest() > data.inner_start() {
                        // table smallest must be less than inner_end or it will not be included.
                        candidate_inner_keys.push(OwnedInnerKey::new(tbl.clone_smallest()));
                    }
                }
            }
        }
        let l0_size: u64 = data.l0_tbls.iter().map(|l0| l0.size()).sum();
        if l0_size > bucket_size {
            let max_table_size = self.opt.table_builder_options.max_table_size;
            let block_size = self.opt.table_builder_options.block_size;
            let step = max_table_size / block_size;
            for l0 in data.l0_tbls.iter() {
                for cf in WRITE_CF..=LOCK_CF {
                    if let Some(tbl) = l0.get_cf(cf) {
                        if tbl.size() < max_table_size as u64 {
                            // ignore small l0 table.
                            continue;
                        }
                        let idx = tbl.load_index(true);
                        if idx.num_blocks() <= step {
                            continue;
                        }
                        for i in (step..idx.num_blocks()).step_by(step) {
                            let sample_key = idx.clone_block_key(i);
                            if sample_key.as_ref() > data.inner_start()
                                && sample_key.as_ref() < data.inner_end()
                            {
                                candidate_inner_keys.push(sample_key);
                            }
                        }
                    }
                }
            }
        }
        candidate_inner_keys.sort_by(|a, b| a.as_ref().cmp(&b.as_ref()));
        candidate_inner_keys
    }

    pub fn get_evenly_split_keys(&self, count: usize, bucket_size: u64) -> Option<Vec<Bytes>> {
        if count <= 1 {
            return None;
        }
        let candidate_inner_keys = self.get_candidate_inner_keys(bucket_size);
        if candidate_inner_keys.len() < 2 {
            return None;
        }
        // Split at table boundaries.
        let steps = evenly_distribute(candidate_inner_keys.len(), count);
        let mut split_keys = Vec::with_capacity(count - 1);
        let mut key_idx = steps[0];
        for step in steps.into_iter().skip(1) {
            let split_key = self.to_outer_key(candidate_inner_keys[key_idx].as_ref());
            split_keys.push(split_key);
            key_idx += step;
        }
        split_keys.dedup();
        Some(split_keys)
    }

    /// Finds an approximate half-split key in the raw outer-key range
    /// `[start_key, end_key)` without scanning rows.
    ///
    /// SST start keys are preferred because splitting there avoids sharing that
    /// SST between both child shards. If no SST starts strictly inside the
    /// effective range, this reads the current index of the largest overlapping
    /// loaded WRITE SST and returns its middle block start key. A related
    /// unloaded WRITE SST makes the result incomplete, so this returns no key.
    /// A range whose cached keyspace ID disagrees with its outer start, or that
    /// spans more than one keyspace, also returns no key because an inner key
    /// cannot be converted back to an unambiguous outer key.
    pub fn approx_split(&self, start_key: &[u8], end_key: &[u8]) -> Result<Option<ApproxSplitKey>> {
        // `keyspace_id` is cached when the range is created, but range bounds
        // can be replaced by merge/recovery code. Re-derive the ID before
        // using `to_outer_key`, and reject ranges whose exclusive end belongs
        // to another keyspace.
        let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&self.outer_start).unwrap_or_default();
        if keyspace_id != self.keyspace_id || keyspace_id != self.range.get_end_keyspace_id() {
            return Ok(None);
        }

        let data = self.get_data();

        let range_start = if start_key.is_empty() {
            data.outer_start.as_ref()
        } else if data.outer_start.is_empty() {
            start_key
        } else {
            cmp::max(start_key, data.outer_start.as_ref())
        };
        let range_end = if end_key.is_empty() {
            data.outer_end.as_ref()
        } else if data.outer_end.is_empty() {
            end_key
        } else {
            cmp::min(end_key, data.outer_end.as_ref())
        };
        if !range_end.is_empty() && range_start >= range_end {
            return Ok(None);
        }

        let range = DataBound::new(
            InnerKey::from_outer_key(range_start),
            InnerKey::from_outer_end_key(range_end),
            false,
        );
        if range.lower_bound >= range.upper_bound {
            return Ok(None);
        }

        let mut tables = Vec::new();
        let mut add_table = |table: &SsTable| {
            if table.entries > 0 && table.overlap_bound(range) {
                tables.push(table.clone());
            }
        };
        for l0 in &data.l0_tbls {
            if let Some(table) = l0.get_cf(WRITE_CF) {
                add_table(table);
            }
        }
        for level in &data.get_cf(WRITE_CF).levels {
            for table in level.tables.iter() {
                add_table(table);
            }
        }
        HALF_SPLIT_TABLES_HISTOGRAM.observe(tables.len() as f64);

        if data.unloaded_tbls.values().any(|table| {
            table.is_sst_file()
                && (table.level == 0 || table.cf == WRITE_CF as i8)
                && table.overlap_bound(range)
        }) {
            return Ok(None);
        }

        let mut sst_start_keys: Vec<_> = tables
            .iter()
            .filter(|table| range.exclusive_overlap_key(table.smallest()))
            .map(|table| OwnedInnerKey::new(table.clone_smallest()))
            .collect();
        sst_start_keys.sort_unstable_by(|a, b| a.as_ref().cmp(&b.as_ref()));
        sst_start_keys.dedup_by(|a, b| a.as_ref() == b.as_ref());
        if !sst_start_keys.is_empty() {
            let split_key = sst_start_keys[sst_start_keys.len() / 2].as_ref();
            let outer_key = self.to_outer_key(split_key);
            return Ok(Some(ApproxSplitKey::Sst(Key::from_raw(&outer_key))));
        }

        let Some(mut representative) = tables.first() else {
            return Ok(None);
        };
        for table in tables.iter().skip(1) {
            if table.kv_size > representative.kv_size
                || table.kv_size == representative.kv_size && table.id() < representative.id()
            {
                representative = table;
            }
        }

        let index = representative.try_load_index(false)?;
        let first = index.seek_block(range.lower_bound);
        let last = index.seek_block_bigger_or_equal(range.upper_bound);
        if first >= last {
            return Ok(None);
        }
        let block_key = index.clone_block_key(first + (last - first) / 2);
        let outer_key = self.to_outer_key(block_key.as_ref());
        Ok(Some(ApproxSplitKey::Block(Key::from_raw(&outer_key))))
    }

    pub fn get_property(&self, key: &str) -> Option<Bytes> {
        self.properties.get(key)
    }

    pub fn set_property(&self, key: &str, val: &[u8]) {
        // TODO: remove property if available.
        self.properties.set(key, val);
        // sync shard pending ops when set shard property.
        match key {
            DEL_PREFIXES_KEY => {
                self.set_del_prefixes(val);
            }
            TRIM_OVER_BOUND => {
                let mut pending_ops = self.pending_ops.write().unwrap();
                pending_ops.trim_over_bound = !val.is_empty();
            }
            MANUAL_MAJOR_COMPACTION => {
                if val.is_empty() {
                    self.properties.remove(MANUAL_MAJOR_COMPACTION);
                }
                let mut pending_ops = self.pending_ops.write().unwrap();
                pending_ops.manual_major_compaction = MajorCompactionType::from(val);
            }
            STORAGE_CLASS_KEY => {
                let mut pending_ops = self.pending_ops.write().unwrap();
                pending_ops.storage_class_spec = StorageClassSpec::unmarshal(Some(val));
            }
            _ => {}
        }
    }

    pub fn set_property_opt(&self, key: &str, val: Option<&[u8]>) {
        if let Some(val) = val {
            self.set_property(key, val);
        }
    }

    pub fn del_property(&self, key: &str) {
        self.properties.remove(key);
    }

    pub(crate) fn load_mem_table_snap_version(&self) -> SnapVersion {
        SnapVersion::new(self.get_base_version(), self.get_write_sequence())
    }

    pub fn get_all_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_files_with_type()
            .into_iter()
            .map(|(_, file_id)| file_id)
            .collect()
    }

    pub fn get_all_sst_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_sst_files()
    }

    /// Returns all live blob table file IDs referenced by this shard.
    pub fn get_all_blob_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_blob_files()
    }

    /// Returns all live blob table file IDs referenced by this shard, split by
    /// storage location.
    ///
    /// Note: `local_files` and `ia_files` may contain duplicated file IDs.
    pub fn get_local_blob_files(
        &self,
    ) -> (
        Vec<u64>, // local_files
        Vec<u64>, // ia_files
    ) {
        let data = self.get_data();
        data.get_local_blob_files(true)
    }

    /// Returns all live SST file IDs referenced by this shard, split by storage
    /// location.
    ///
    /// Note: `local_files` and `ia_files` may contain duplicated file IDs.
    ///
    /// Blob tables are excluded. Use [`Shard::get_all_blob_files`] to list
    /// `.blob` files.
    pub fn get_local_sst_files(
        &self,
    ) -> (
        Vec<u64>, // local_files
        Vec<u64>, // ia_files
    ) {
        let data = self.get_data();
        data.get_local_sst_files(true)
    }

    pub fn get_txn_chunks(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_txn_chunks()
    }

    pub fn get_all_col_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_col_files()
    }

    pub fn get_all_vec_idx_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_vec_idx_files()
    }

    pub fn get_all_fts_files(&self) -> Vec<u64> {
        let data = self.get_data();
        data.get_all_fts_files()
    }

    pub fn get_tracked_fts_field_layout(
        &self,
        table_id: i64,
        index_id: i64,
    ) -> Option<kvenginepb::fts::FtsFieldLayout> {
        self.get_data()
            .fts_levels
            .tracked_indexes_ref()
            .get(&table_id)
            .and_then(|indexes| indexes.get(&index_id))
            .map(|meta| meta.get_field_layout())
    }

    pub fn get_unapplied_pending_files(&self) -> HashSet<u64> {
        let pending_files = self.unapplied_pending_files.lock().unwrap();
        let mut all_files = HashSet::new();

        for info in pending_files.values() {
            all_files.extend(&info.file_ids);
        }

        all_files
    }

    #[inline]
    pub fn has_txn_file_locks(&self) -> bool {
        self.get_data().has_txn_file_locks()
    }

    pub(crate) fn split_mem_tables(&self, parent_mem_tbls: &[CfTable]) -> Vec<CfTable> {
        let mut new_mem_tbls = vec![self.get_writable_mem_table().new_empty_like()];
        for old_mem_tbl in parent_mem_tbls {
            if old_mem_tbl.is_force_switch() || old_mem_tbl.has_data_in_bound(self.data_bound()) {
                new_mem_tbls.push(old_mem_tbl.new_split());
            }
        }
        new_mem_tbls
    }

    pub fn get_base_version(&self) -> u64 {
        self.base_version.load(Ordering::Acquire)
    }

    // For test only.
    #[cfg(test)]
    pub fn set_base_version(&self, ver: u64) {
        self.base_version.store(ver, Ordering::Release);
    }

    pub fn get_write_sequence(&self) -> u64 {
        self.write_sequence.load(Ordering::Acquire)
    }

    pub fn update_write_sequence_on_recover(&self, seq: u64) {
        self.write_sequence.store(seq, Ordering::Release);
    }

    pub fn get_persisted_snap_version(&self) -> SnapVersion {
        self.get_data().persisted_version
    }

    pub fn get_columnar_snap_version(&self) -> SnapVersion {
        self.get_data().get_columnar_snap_version()
    }

    pub fn get_columnar_l2_snap_version(&self) -> SnapVersion {
        self.get_data().col_levels.l2_snap_version
    }

    pub fn get_columnar_table_ids(&self) -> Vec<i64> {
        self.get_data().columnar_table_ids.clone()
    }

    pub fn get_meta_sequence(&self) -> u64 {
        self.meta_seq.load(Ordering::Acquire)
    }

    pub fn get_estimated_size(&self) -> u64 {
        self.estimated_size.load(Ordering::Relaxed)
    }

    pub fn get_estimated_entries(&self) -> u64 {
        self.estimated_entries.load(Ordering::Relaxed)
    }

    // Note: `get_max_ts` would be not up-to-date.
    // Invoke `Shard::refresh_states` to refresh it.
    pub fn get_max_ts(&self) -> u64 {
        cmp::max(
            self.max_ts.load(Ordering::Relaxed),
            self.get_data().get_mem_table_max_ts(),
        )
    }

    pub fn get_sst_max_ts(&self) -> u64 {
        self.sst_max_ts.load(Ordering::Relaxed)
    }

    pub fn get_estimated_kv_size(&self) -> u64 {
        self.estimated_kv_size.load(Ordering::Relaxed)
    }

    pub fn get_estimated_ia_kv_size(&self) -> u64 {
        self.estimated_ia_kv_size.load(Ordering::Relaxed)
    }

    pub fn get_estimated_columnar_size(&self) -> u64 {
        self.estimated_columnar_size.load(Ordering::Relaxed)
    }

    pub fn get_estimated_columnar_kv_size(&self) -> u64 {
        self.estimated_columnar_kv_size.load(Ordering::Relaxed)
    }

    pub fn get_initial_flushed(&self) -> bool {
        self.initial_flushed.load(Acquire)
    }

    pub(crate) fn ready_to_destroy_range(
        del_prefixes: &DeletePrefixes,
        shard_data: &ShardData,
    ) -> bool {
        !del_prefixes.is_empty() && del_prefixes.after_scheduled_time()
        // No memtable contains data covered by deleted prefixes.
        && !shard_data.mem_tbls.iter().any(|mem_tbl| {
            del_prefixes
                .inner_delete_bounds()
                .any(|bound| mem_tbl.has_data_in_bound(bound))
        })
    }

    fn ready_to_truncate_ts(truncate_ts: &Option<u64>, shard_data: &ShardData) -> bool {
        truncate_ts.is_some()
        // No memtable contains data with version > truncate_ts.
        && !shard_data.mem_tbls.iter().any(|mem_tbl| {
            mem_tbl.data_max_ts() > truncate_ts.unwrap()
        })
    }

    fn ready_to_trim_over_bound(trim_over_bound: bool, shard_data: &ShardData) -> bool {
        trim_over_bound && !shard_data.has_mem_over_bound_data()
    }

    fn ready_to_manual_major_compaction(comp: MajorCompactionType, shard_data: &ShardData) -> bool {
        match comp {
            MajorCompactionType::Disable => false,
            MajorCompactionType::Enable => !shard_data.has_unconverted_l0s(),
            MajorCompactionType::EnableColumnar => {
                shard_data
                    .schema_file
                    .as_ref()
                    .map(|f| !f.tables_with_columnar().is_empty())
                    .unwrap_or(false)
                    && !shard_data.columnar_table_ids.is_empty()
                    && !shard_data.get_all_columnar_files().is_empty()
            }
        }
    }

    fn refresh_compaction_priority(&self) {
        let data = self.get_data();

        // unconverted L0 to columnar has the highest priority. In merging state, only
        // L0 to columnar compaction is allowed.
        if let Some(l0_to_columnar) = self.check_l0_to_columnar_priority(&data) {
            *self.compaction_priority.write().unwrap() = Some(l0_to_columnar);
            return;
        }

        let pending_ops = self.pending_ops.read().unwrap();
        // Set manual major compact the higher priority than inplace compactions for
        // shard.
        if Self::ready_to_manual_major_compaction(pending_ops.manual_major_compaction, &data) {
            let is_columnar =
                pending_ops.manual_major_compaction == MajorCompactionType::EnableColumnar;
            if is_columnar {
                // Manual columnar major compaction can delete columnar L0 files before FTS
                // indexes them (e.g. during/after region merge when there are explicitly
                // pending L0 IDs). Once the columnar L0 is gone, the pending list will be
                // pruned by the columnar compaction apply path and the FTS read path may
                // incorrectly assume full coverage by index and return stale results.
                //
                // Prefer completing FTS incremental indexing first, then run major compaction.
                if let Some(priority) = self.get_fts_incremental_update_priority(&data) {
                    *self.compaction_priority.write().unwrap() = Some(priority);
                    return;
                }
            }
            *self.compaction_priority.write().unwrap() = if !is_columnar {
                Some(CompactionPriority::Major {
                    score: 2.0,
                    is_manual: true,
                })
            } else {
                Some(CompactionPriority::ColumnarMajor {
                    table_ids_to_add: vec![],
                    table_ids_to_clear: vec![],
                    is_manual: true,
                    schema_version: data.schema_file.as_ref().unwrap().get_version(),
                })
            };
            return;
        } else if Self::ready_to_destroy_range(&pending_ops.del_prefixes, &data) {
            *self.compaction_priority.write().unwrap() = Some(CompactionPriority::DestroyRange);
            return;
        } else if Self::ready_to_truncate_ts(&pending_ops.truncate_ts, &data) {
            *self.compaction_priority.write().unwrap() = Some(CompactionPriority::TruncateTs);
            return;
        } else if Self::ready_to_trim_over_bound(pending_ops.trim_over_bound, &data) {
            *self.compaction_priority.write().unwrap() = Some(CompactionPriority::TrimOverBound);
            return;
        }
        if !data.blob_tbl_map.is_empty() {
            let blob_table_utilization = {
                let mut in_use_blob_size = 0;
                let mut total_blob_size = 0;
                for l0 in &data.l0_tbls {
                    in_use_blob_size += l0.total_blob_size();
                }
                for cf in &data.cfs {
                    for lh in &cf.levels {
                        for sst in &(*lh.tables) {
                            in_use_blob_size += sst.total_blob_size();
                        }
                    }
                }

                for blob_table in data.blob_tbl_map.values() {
                    total_blob_size += blob_table.total_blob_size();
                }
                if total_blob_size == 0 {
                    1.0
                } else {
                    in_use_blob_size as f64 / total_blob_size as f64
                }
            };
            if blob_table_utilization < self.opt.blob_table_gc_ratio && !data.has_unconverted_l0s()
            {
                let mut lock = self.compaction_priority.write().unwrap();
                *lock = Some(CompactionPriority::Major {
                    score: f64::MAX,
                    is_manual: false,
                });
                return;
            }
        }

        if let Some(priority) = self.get_col_or_vec_idx_compaction_high_priority(&data) {
            let mut lock = self.compaction_priority.write().unwrap();
            *lock = Some(priority);
            return;
        }

        let mut score = 0.0;
        let mut cf_with_highest_score = -1;
        let mut level_with_highest_score = 0;
        if data.l0_tbls.len() > 2 {
            let size_score = data.get_l0_total_size() as f64 / self.opt.base_size as f64;
            let num_tbl_score = data.l0_tbls.len() as f64 / 4.0;
            score = size_score * 0.6 + num_tbl_score * 0.4;
        }
        let shard_bound = data.data_bound();
        for l0 in &data.l0_tbls {
            if !shard_bound.contains_bound(l0.data_bound()) {
                // set highest priority for newly split L0.
                score += 2.0;
                let mut lock = self.compaction_priority.write().unwrap();
                *lock = Some(CompactionPriority::L0 {
                    score,
                    file_candidate: None,
                });
                return;
            }
        }
        for cf in 0..NUM_CFS {
            let scf = data.get_cf(cf);
            for lh in &scf.levels[..scf.levels.len() - 1] {
                // Use `StorageClassSpec::default()`.
                // The `sc_spec` just affect `ia_kv_size` which is not used here.
                let lvl_stats = data.get_level_stats(cf, lh, &StorageClassSpec::default());
                let lvl_score = lvl_stats.data_size as f64
                    / ((self.opt.base_size as f64) * 10f64.powf((lh.level - 1) as f64));
                if score < lvl_score {
                    score = lvl_score;
                    level_with_highest_score = lh.level;
                    cf_with_highest_score = cf as isize;
                }
            }
        }
        let priority = if score > 1.0 {
            let max_pri = if level_with_highest_score == 0 {
                CompactionPriority::L0 {
                    score,
                    file_candidate: None,
                }
            } else {
                CompactionPriority::L1Plus {
                    cf: cf_with_highest_score,
                    level: level_with_highest_score,
                    score,
                    file_gc: false,
                }
            };
            Some(max_pri)
        } else {
            // NOTE: the columnar compaction priority should be lower than sst compaction
            // to avoid sst compaction being blocked by columnar major compaction.
            self.get_columnar_compaction_priority(&data)
                .or_else(|| self.get_vector_index_priority(&data))
                .or_else(|| self.get_fts_compaction_priority(&data))
        };

        // Trigger L0 compaction for test purpose.
        let priority = priority.or_else(|| {
            let handle_priority_none = || -> Option<CompactionPriority> {
                if !data.l0_tbls.is_empty() {
                    fail::fail_point!("refresh_compaction_priority_for_l0", |_| {
                        Some(CompactionPriority::L0 {
                            score: 2.0,
                            file_candidate: None,
                        })
                    });
                }
                None
            };
            handle_priority_none()
        });

        let mut lock = self.compaction_priority.write().unwrap();
        *lock = priority;
    }

    fn check_l0_to_columnar_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        if self.opt.ignore_columnar_table_load {
            return None;
        }

        // ColumnarMajor compaction must be done before converting L0 to columnar.
        if !data.col_levels.unconverted_l0s.is_empty() {
            debug_assert!(!data.columnar_table_ids.is_empty());
            let Some(schema_file) = data.schema_file.as_ref() else {
                return Some(CompactionPriority::ColumnarClear);
            };
            // Schema is outdated, wait for update or clear columnar if there are too many
            // unconverted L0.
            if self.get_outdated_schema_ver() == schema_file.get_version() {
                if data.has_too_many_unconverted_l0s() {
                    ENGINE_COLUMNAR_TOO_MANY_UNCONVERTED_L0S.inc();
                    warn!(
                        "{} schema is outdated and has {} unconverted L0 files, clear schema file and columnar files.",
                        self.tag(),
                        data.col_levels.unconverted_l0s.len()
                    );
                    return Some(CompactionPriority::ColumnarClear);
                } else {
                    warn!(
                        "{} schema is outdated and has {} unconverted L0 files, wait for schema update.",
                        self.tag(),
                        data.col_levels.unconverted_l0s.len()
                    );
                    // Do nothing, give it a chance to update schema file.
                    return None;
                }
            }
            fail::fail_point!("kvengine_l0_to_columnar", |_| None);
            return Some(CompactionPriority::L0ToColumnar);
        }

        None
    }

    fn get_col_or_vec_idx_compaction_high_priority(
        &self,
        data: &ShardData,
    ) -> Option<CompactionPriority> {
        if self.opt.ignore_columnar_table_load {
            return None;
        }

        if data.schema_file.is_none() && !data.columnar_table_ids.is_empty() {
            return Some(CompactionPriority::ColumnarClear);
        }

        // The vector index is removed from schema file, remove it.
        let schema_file = data.schema_file.as_ref()?;
        for vec_idx in data.vector_indexes.get_all() {
            if let Some(schema) = schema_file.get_table(vec_idx.table_id) {
                if !schema
                    .vector_indexes
                    .iter()
                    .any(|idx| idx.index_id == vec_idx.index_id && idx.col_id == vec_idx.col_id)
                {
                    return Some(CompactionPriority::UpdateVectorIndex {
                        score: 2.0,
                        table_id: vec_idx.table_id,
                        index_id: vec_idx.index_id,
                        col_id: vec_idx.col_id,
                        rebuild: true,
                    });
                }
            }
        }

        if !data.fts_levels.tracked_indexes_ref().is_empty() {
            let active_fts_indexes: HashSet<(i64, i64)> = schema_file
                .iter_tables()
                .flat_map(|(&table_id, schema)| {
                    schema
                        .fulltext_indexes
                        .iter()
                        .map(move |idx| (table_id, idx.index_id))
                })
                .collect();
            let mut dropped_indexes: Vec<(i64, i64)> = data
                .fts_levels
                .iter_tracked_indexes()
                .flat_map(|(table_id, indexes)| {
                    indexes.keys().map(move |index_id| (*table_id, *index_id))
                })
                .filter(|pair| !active_fts_indexes.contains(pair))
                .collect();
            dropped_indexes.sort_unstable();
            dropped_indexes.dedup();
            if !dropped_indexes.is_empty() {
                return Some(CompactionPriority::FtsDropIndex { dropped_indexes });
            }
        }

        None
    }

    fn get_table_ids_need_columnar_major_compaction(
        &self,
        data: &ShardData,
    ) -> (
        Vec<i64>, // table ids that need to add columnar
        Vec<i64>, // table ids that need to clear columnar
    ) {
        if data.schema_file.is_none() {
            // Return empty table ids means no table need to add or clear columnar.
            return (vec![], vec![]);
        }
        // Collect table ids that exists only in shard or schema file. If table id only
        // exists in schema file, it means the table need to add columnar; If table id
        // only exists in shard, it means the table need to clear columnar.
        let tables_in_shard = data
            .columnar_table_ids
            .iter()
            .cloned()
            .collect::<HashSet<_>>();
        let tables_in_schema = data
            .schema_file
            .as_ref()
            .map(|schema| {
                schema
                    .overlap_columnar_tables(data.range.inner_start(), data.range.inner_end())
                    .into_iter()
                    .collect::<HashSet<_>>()
            })
            .unwrap_or_default();
        let mut tables_to_add: Vec<i64> = tables_in_schema
            .difference(&tables_in_shard)
            .cloned()
            .collect();
        let mut tables_to_clear: Vec<i64> = tables_in_shard
            .difference(&tables_in_schema)
            .cloned()
            .collect();
        tables_to_add.sort();
        tables_to_clear.sort();
        (tables_to_add, tables_to_clear)
    }

    fn get_columnar_compaction_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        if self.opt.ignore_columnar_table_load {
            return None;
        }

        let (table_ids_to_add, table_ids_to_clear) =
            self.get_table_ids_need_columnar_major_compaction(data);
        if self.opt.build_columnar()
            && (!table_ids_to_add.is_empty() || !table_ids_to_clear.is_empty())
        {
            // Columnar major compaction may rewrite/drop columnar L0 files. If FTS still
            // has pending incremental indexing work (especially after region
            // merge), compacting columnar first can make some unindexed data
            // invisible to the FTS read path.
            //
            // Prefer completing the FTS incremental update first, then rewrite columnar.
            if let Some(priority) = self.get_fts_incremental_update_priority(data) {
                return Some(priority);
            }
            return Some(CompactionPriority::ColumnarMajor {
                table_ids_to_add,
                table_ids_to_clear,
                is_manual: false,
                schema_version: data.schema_file.as_ref().unwrap().get_version(),
            });
        }

        // TODO: use dependent base_size for columnar
        let col_l0_score =
            data.get_columnar_level_stats(0).columnar_size as f64 / self.opt.base_size as f64;
        let col_l1_score = data.get_columnar_level_stats(1).columnar_size as f64
            / (10 * self.opt.base_size) as f64;
        if data.schema_file.is_some()
            && (data.get_col_table_counts(0) > MAX_COL_L0_FILE_COUNTS
                || (col_l0_score >= col_l1_score && col_l0_score > 1.0))
        {
            if let Some(priority) = self.maybe_override_by_fts_index(data) {
                Some(priority)
            } else {
                Some(CompactionPriority::ColumnarL0 {
                    score: col_l0_score,
                })
            }
        } else if data.schema_file.is_some()
            && (data.get_col_table_counts(1) > MAX_COL_L1_FILE_COUNTS
                || (col_l0_score < col_l1_score && col_l1_score > 1.0))
        {
            if let Some(priority) = Self::maybe_override_by_vector_index(data) {
                Some(priority)
            } else {
                Some(CompactionPriority::ColumnarL1 {
                    score: col_l1_score,
                })
            }
        } else {
            None
        }
    }

    // To prevent L1 columnar compaction invalidate existing vector index, we
    // replace the L1 columnar compaction with update vector index.
    fn maybe_override_by_vector_index(data: &ShardData) -> Option<CompactionPriority> {
        if !data.vector_indexes.is_empty() {
            let first_l1_columnar = data.col_levels.levels[1].files.first().unwrap();
            let l2_snap_after_compaction = first_l1_columnar.get_snap_version().unwrap();
            for vec_idx in data.vector_indexes.get_all() {
                if vec_idx.need_rebuild() {
                    return Some(CompactionPriority::UpdateVectorIndex {
                        score: 2.0,
                        table_id: vec_idx.table_id,
                        index_id: vec_idx.index_id,
                        col_id: vec_idx.col_id,
                        rebuild: true,
                    });
                }
                if vec_idx.snap_version() < l2_snap_after_compaction {
                    return Some(CompactionPriority::UpdateVectorIndex {
                        score: 2.0,
                        table_id: vec_idx.table_id,
                        index_id: vec_idx.index_id,
                        col_id: vec_idx.col_id,
                        rebuild: false,
                    });
                }
            }
        }
        None
    }

    fn maybe_override_by_fts_index(&self, data: &ShardData) -> Option<CompactionPriority> {
        self.get_fts_incremental_update_priority(data)
    }

    fn get_vector_index_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        if data.columnar_table_ids.is_empty() {
            return None;
        }
        let schema_file = data.schema_file.as_ref()?;
        let schemas = schema_file.get_vector_index_schemas();
        let mut priority = None;
        let mut max_score = 1.0;
        for schema in &schemas {
            for vec_idx in &schema.vector_indexes {
                let (score, rebuild) = self.vector_index_score(data, schema.table_id, vec_idx);
                if score > max_score {
                    max_score = score;
                    priority = Some(CompactionPriority::UpdateVectorIndex {
                        score,
                        table_id: schema.table_id,
                        index_id: vec_idx.index_id,
                        col_id: vec_idx.col_id,
                        rebuild,
                    });
                }
            }
        }
        priority
    }

    fn vector_index_score(
        &self,
        data: &ShardData,
        table_id: i64,
        vec_idx: &VectorIndexDef,
    ) -> (f64, bool) {
        // Trigger rebuild vector index file to v2 format.
        // TODO: remove this after v2 format is transitioned.
        if let Some(vec_idx) = data
            .vector_indexes
            .get(table_id, vec_idx.index_id, vec_idx.col_id)
        {
            if vec_idx.files.iter().any(|file| file.is_legacy_format()) {
                return (2.0, true);
            }
        }

        let snap_version = if let Some(vec_idx) =
            data.vector_indexes
                .get(table_id, vec_idx.index_id, vec_idx.col_id)
        {
            if vec_idx.files_count() >= self.opt.vector_index_build_options.rebuild_file_count {
                return (1.1, true);
            }
            vec_idx.snap_version()
        } else {
            SnapVersion::zero()
        };
        if !self.need_build_vector_index(data, table_id, vec_idx, snap_version) {
            return (0.0, false);
        }
        (1.1, snap_version.is_zero())
    }

    fn need_build_vector_index(
        &self,
        data: &ShardData,
        table_id: i64,
        vec_idx: &VectorIndexDef,
        snap_version: SnapVersion,
    ) -> bool {
        let mut total_row_count = 0usize;
        for col_level in &data.col_levels.levels {
            if col_level.level == 2 && snap_version.is_not_zero() {
                // The level 2 columnar files are already included in vector index.
                continue;
            }
            for file in &col_level.files {
                if !file.has_table(table_id) {
                    continue;
                }
                if snap_version.is_not_zero()
                    && file.get_snap_version().unwrap_or_default() <= snap_version
                {
                    continue;
                }
                let tbl_meta = file.get_table(table_id);
                let (_, row_count) = tbl_meta.handle_column.pack_offsets.end_offset();
                total_row_count += row_count as usize;
            }
        }

        // Note: Remember to update `collect_columnar_index_stats` when strategy here
        // is changed, because we provide a field to notify user how many rows
        // WILL build vector index.

        let total_size = total_row_count * vec_idx.dimension * 4;
        // We don't need to build vector index for small number of vectors.
        total_size >= self.opt.vector_index_build_options.delta_size
    }

    fn get_fts_compaction_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        if !self.opt.build_fts_index() {
            return None;
        }
        self.get_fts_add_index_priority(data)
            .or_else(|| self.get_fts_incremental_update_priority(data))
            .or_else(|| self.get_fts_l0_compact_priority(data))
            .or_else(|| self.get_fts_l2_compact_priority(data))
            .or_else(|| self.get_fts_cleanup_priority(data))
            .or_else(|| self.get_fts_l2_full_compact_priority(data))
    }

    fn get_fts_add_index_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let schema_file = data.schema_file.as_ref()?;
        let mut new_indexes = Vec::new();
        for (&table_id, table_schema) in schema_file.iter_tables() {
            if table_schema.fulltext_indexes.is_empty() {
                continue;
            }
            for fts_index in &table_schema.fulltext_indexes {
                let index_id = fts_index.index_id;
                if !data.fts_levels.has_tracked_index(table_id, index_id) {
                    new_indexes.push((table_id, index_id));
                }
            }
        }

        // Handle new index addition (highest priority)
        // FtsAddIndex doesn't need columnar files - it just tracks new indexes

        if new_indexes.is_empty() {
            None
        } else {
            Some(CompactionPriority::FtsAddIndex { new_indexes })
        }
    }

    fn get_fts_incremental_update_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let schema_file = data.schema_file.as_ref()?;

        if data.columnar_table_ids.is_empty() || !self.opt.build_fts_index() {
            return None;
        }

        // Handle incremental updates for existing tracked indexes
        // Check if there are actual new columnar L0 files to process

        let max_columnar_l0_version = data.col_levels.levels[0]
            .files
            .iter()
            .fold(data.fts_levels.l0_snap_version, |v, f| {
                std::cmp::max(v, f.get_snap_version().unwrap_or_default())
            });

        if !data
            .fts_levels
            .needs_delta_indexing(max_columnar_l0_version)
        {
            return None;
        }

        // It could be possible that tracked indexes are dropped. In this case we will
        // not consider them for incremental updates.
        let new_active_indexes = {
            let mut new_active_indexes = vec![];
            for (table_id, index_metas) in data.fts_levels.iter_tracked_indexes() {
                let Some(full_schema) = schema_file.get_table(*table_id) else {
                    continue;
                };
                for fts_index in &full_schema.fulltext_indexes {
                    if index_metas.contains_key(&fts_index.index_id) {
                        new_active_indexes.push((*table_id, fts_index.index_id));
                    }
                }
            }
            new_active_indexes
        };

        if new_active_indexes.is_empty() {
            return None;
        }

        // Collect columnar L0 files newer than the FTS watermark for incremental
        // indexing
        let l0_watermark = data.fts_levels.l0_snap_version;
        let mut col_file_ids: HashSet<u64> = HashSet::new();

        // Filter columnar L0 files >= current snap_version or explicitly pending.
        // "Explicitly pending" files usually comes from shard merge. In that case,
        // the merged shard's L0 watermark has already been advanced beyond some of the
        // existing L0 files, but those files still need to be indexed.

        // Columnar l0 files are sorted by version descendingly, so we can stop once
        // we see a file <= l0_watermark and there is no remaining pending item.
        for col_file in &data.col_levels.levels[0].files {
            let l0_version = col_file.get_snap_version().unwrap_or_default();
            if l0_version > l0_watermark {
                col_file_ids.insert(col_file.id());
            } else if l0_version <= l0_watermark {
                break;
            }
        }
        if !data.fts_levels.pending_columnar_l0_ids().is_empty() {
            // Filter out stale pending IDs (e.g. removed by columnar compaction) to
            // avoid repeatedly scheduling tasks for missing files.
            let valid_l0_ids: HashSet<u64> = data.col_levels.levels[0]
                .files
                .iter()
                .map(|f| f.id())
                .collect();
            for pending_id in data.fts_levels.pending_columnar_l0_ids() {
                if valid_l0_ids.contains(pending_id) {
                    col_file_ids.insert(*pending_id);
                }
            }
        }

        if col_file_ids.is_empty() {
            // No new columnar L0 files beyond the L0 snap_version watermark
            return None;
        }

        Some(CompactionPriority::FtsCreateL0 {
            col_file_ids: col_file_ids.into_iter().collect(),
            active_tracked_indexes: new_active_indexes,
            // Use the max snap_version of existing columnar L0 files as the version for the
            // generated FTS L0. This keeps file ordering consistent when we need to "fill a hole"
            // by indexing explicitly pending L0s whose snap_version can be <= current
            // `fts_levels.l0_snap_version` (common after region merge).
            snap_version: max_columnar_l0_version,
        })
    }

    fn get_fts_l0_compact_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let levels = &data.fts_levels;

        // Check if there has any l1 files with same lp_key overlap due to region merge.
        // If there are, we need to compact L0 and L1 files together to make sure there
        // are no overlapping lp_key in l1 files.
        let has_overlapping_l1_files = levels
            .l1()
            .windows(2)
            .any(|w| w[0].props().get_largest_lp_key() >= w[1].props().get_smallest_lp_key());
        if has_overlapping_l1_files {
            return Some(CompactionPriority::FtsCompactL0);
        }

        if levels.l0().is_empty() {
            return None;
        }

        if levels.tracked_indexes_ref().is_empty() {
            // No tracked indexes (e.g. all fulltext indexes are dropped). Compact anyway
            // to make sure stale L0 data gets dropped.
            return Some(CompactionPriority::FtsCompactL0);
        }

        let total_size: u64 = levels.l0().iter().map(|file| file.file().size()).sum();
        let build_opts = &self.opt.fts_build_options;
        if levels.l0().len() < build_opts.max_l0_files && total_size < build_opts.max_l0_sizes.0 {
            return None;
        }
        Some(CompactionPriority::FtsCompactL0)
    }

    /// Expose for test only.
    #[cfg(test)]
    pub(crate) fn test_get_fts_l2_compact_priority(&self) -> Option<CompactionPriority> {
        let data = self.get_data();
        self.get_fts_l2_compact_priority(&data)
    }

    #[cfg(test)]
    pub(crate) fn test_get_fts_incremental_update_priority(&self) -> Option<CompactionPriority> {
        let data = self.get_data();
        self.get_fts_incremental_update_priority(&data)
    }

    #[cfg(test)]
    pub(crate) fn test_get_fts_add_index_priority(&self) -> Option<CompactionPriority> {
        let data = self.get_data();
        self.get_fts_add_index_priority(&data)
    }

    #[cfg(test)]
    pub(crate) fn test_get_fts_cleanup_priority(&self) -> Option<CompactionPriority> {
        let data = self.get_data();
        self.get_fts_cleanup_priority(&data)
    }

    fn get_fts_l2_compact_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let l2 = data.fts_levels.l2();
        if l2.is_empty() {
            return None;
        }

        let opts = &self.opt.fts_build_options;
        if opts.l2_max_files_per_lp == 0 {
            return None;
        }

        let tracked_indexes: HashSet<(i64, i64)> = data
            .fts_levels
            .iter_tracked_indexes()
            .flat_map(|(table_id, indexes)| {
                indexes.keys().map(move |index_id| (*table_id, *index_id))
            })
            .collect();
        if tracked_indexes.is_empty() {
            return None;
        }

        let mut best: Option<(Vec<u8>, Vec<u64>)> = None;

        for (lp_key, files) in l2 {
            let candidates: Vec<crate::table::fts::LogMergeCandidate> = files
                .iter()
                .filter(|f| {
                    tracked_indexes.contains(&(f.props().get_table_id(), f.props().get_index_id()))
                })
                .map(|f| crate::table::fts::LogMergeCandidate {
                    id: f.id(),
                    pk_total: f.props().get_pk_total(),
                    size: f.file().size(),
                    snap_version: f.props().get_snap_version(),
                })
                .collect();

            if candidates.len() < opts.l2_min_merge_files {
                continue;
            }

            if let Some(file_ids) = crate::table::fts::pick_l2_merge_files(&candidates, opts) {
                let replace = match &best {
                    None => true,
                    Some((_, prev_ids)) => file_ids.len() > prev_ids.len(),
                };
                if replace {
                    best = Some((lp_key.clone(), file_ids));
                }
            }
        }

        best.map(|(lp_key, file_ids)| CompactionPriority::FtsCompactL2 { lp_key, file_ids })
    }

    fn get_fts_l2_full_compact_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let l2 = data.fts_levels.l2();
        if l2.is_empty() {
            return None;
        }

        let opts = &self.opt.fts_build_options;
        if opts.l2_full_compact_dead_ratio <= 0.0 || opts.l2_full_compact_dead_bytes.0 == 0 {
            return None;
        }

        let tracked_indexes: HashSet<(i64, i64)> = data
            .fts_levels
            .iter_tracked_indexes()
            .flat_map(|(table_id, indexes)| {
                indexes.keys().map(move |index_id| (*table_id, *index_id))
            })
            .collect();
        if tracked_indexes.is_empty() {
            return None;
        }

        let mut best: Option<(f64, Vec<u8>)> = None;

        for (lp_key, files) in l2 {
            let mut total_size = 0u64;
            let mut est_dead_bytes = 0u64;
            let mut has_tracked = false;

            for file in files {
                let table_index = (file.props().get_table_id(), file.props().get_index_id());
                if !tracked_indexes.contains(&table_index) {
                    continue;
                }
                has_tracked = true;

                let file_size = file.file().size();
                total_size = total_size.saturating_add(file_size);

                let pk_total = file.props().get_pk_total();
                if pk_total == 0 {
                    continue;
                }
                let pk_live_unique = file.props().get_pk_live_unique();
                let dead_versions = pk_total.saturating_sub(pk_live_unique);
                let dead_ratio = dead_versions as f64 / pk_total as f64;
                est_dead_bytes =
                    est_dead_bytes.saturating_add((file_size as f64 * dead_ratio) as u64);
            }

            if !has_tracked || total_size == 0 {
                continue;
            }
            if est_dead_bytes < opts.l2_full_compact_dead_bytes.0 {
                continue;
            }

            let dead_ratio = est_dead_bytes as f64 / total_size as f64;
            if dead_ratio < opts.l2_full_compact_dead_ratio {
                continue;
            }

            // Keep the score close to intra-L2 compaction, but slightly lower
            // than FtsCompactL0. Clamp to keep the score stable.
            let score = 1.75 + dead_ratio.min(1.0) * 0.1;
            let replace = match &best {
                None => true,
                Some((best_score, best_lp_key)) => {
                    score > *best_score || (score == *best_score && lp_key < best_lp_key)
                }
            };
            if replace {
                best = Some((score, lp_key.clone()));
            }
        }

        best.map(|(score, lp_key)| CompactionPriority::FtsFullCompactL2 { score, lp_key })
    }

    fn get_fts_cleanup_priority(&self, data: &ShardData) -> Option<CompactionPriority> {
        let (l1_file_ids, l2_file_ids) = data.fts_levels.collect_cleanup_candidates();
        if l1_file_ids.is_empty() && l2_file_ids.is_empty() {
            None
        } else {
            Some(CompactionPriority::FtsCleanup {
                l1_file_ids,
                l2_file_ids,
            })
        }
    }

    pub(crate) fn get_compaction_priority(&self) -> Option<CompactionPriority> {
        self.compaction_priority.read().unwrap().clone()
    }

    /// Returns true if any compaction priority is currently set.
    pub fn has_compaction_priority(&self) -> bool {
        self.compaction_priority.read().unwrap().is_some()
    }

    pub fn is_compacting(&self) -> bool {
        load_bool(&self.compacting)
    }

    /// Check whether to trigger GC-related compaction.
    ///
    /// Due to high costs of major compaction, the current strategies are
    /// relatively conservative.
    ///
    /// Row-store (SST) obsolete data -- tombstones and old versions -- is
    /// normally reclaimed per file by file-level GC compaction
    /// (`find_file_gc_candidate_level` / `find_file_gc_candidate_l0`). So for
    /// row-store this shard-aggregate major chiefly acts as the fallback for
    /// what file-level GC leaves behind: a file whose lower-level overlap
    /// exceeds `file_gc_compaction_max_overlap_ratio` (skipped to bound
    /// write amplification), and bottommost-level data while
    /// `file_gc_compaction_allow_l3` is off (usually only during a rolling
    /// upgrade). Blob bytes, which file-level GC never reclaims, are also
    /// freed by any major, but their trigger lives in
    /// `refresh_compaction_priority`, not here.
    pub fn check_need_gc_compaction(&self, gc_safe_point: u64, gc_lock_extra: bool) -> bool {
        if self.is_compacting()
            || self.get_compaction_priority().is_some()
            || self.pending_ops.read().unwrap().manual_major_compaction
                != MajorCompactionType::Disable
        {
            return false;
        }
        if gc_lock_extra && self.check_need_gc_extra_file(gc_safe_point) {
            return true;
        }
        if gc_lock_extra && self.check_need_gc_lock_file(gc_safe_point) {
            return true;
        }
        let lv2plus_max_ts = self.lv2plus_max_ts.load(Ordering::Relaxed);
        if lv2plus_max_ts > gc_safe_point {
            return false;
        }

        if self.get_data().has_unconverted_l0s() {
            return false;
        }

        let tombs = self.lv2plus_tombs.load(Ordering::Relaxed);
        let write_entries = self.lv2plus_entries_write_cf.load(Ordering::Relaxed);
        if tombs >= self.opt.compaction_tombs_count && write_entries >= tombs && write_entries > 0 {
            let tombs_ratio = tombs as f64 / write_entries as f64;
            if tombs_ratio >= self.opt.compaction_tombs_ratio {
                let mut priority = self.compaction_priority.write().unwrap();
                if priority.is_some() {
                    return false;
                }

                info!("{} trigger major compaction for tombstones", self.tag();
                    "tombs" => tombs, "write_entries" => write_entries, "ratio" => tombs_ratio,
                    "max_ts" => lv2plus_max_ts, "gc_safe_point" => gc_safe_point);
                *priority = Some(CompactionPriority::Major {
                    score: tombs_ratio,
                    is_manual: false,
                });
                return true;
            }
        }

        let old_data_size_threshold = self.opt.compaction_old_data_size();
        if old_data_size_threshold == 0 {
            return false;
        }
        let old_data_size = self.lv2plus_old_data_size.load(Ordering::Relaxed);
        if old_data_size < old_data_size_threshold {
            return false;
        }
        let data_size = self.lv2plus_data_size.load(Ordering::Relaxed);
        if data_size == 0 || old_data_size > data_size {
            return false;
        }
        let old_data_ratio = old_data_size as f64 / data_size as f64;
        let old_data_ratio_threshold = self.opt.compaction_old_data_ratio();
        if old_data_ratio >= old_data_ratio_threshold {
            let mut priority = self.compaction_priority.write().unwrap();
            if priority.is_some() {
                return false;
            }

            info!("{} trigger major compaction for old-version data", self.tag();
                "old_data_size" => old_data_size, "data_size" => data_size,
                "ratio" => old_data_ratio, "ratio_threshold" => old_data_ratio_threshold,
                "size_threshold" => old_data_size_threshold,
                "max_ts" => lv2plus_max_ts, "gc_safe_point" => gc_safe_point);
            *priority = Some(CompactionPriority::Major {
                score: old_data_ratio,
                is_manual: false,
            });
            return true;
        }

        false
    }

    /// Whether a file-GC candidate's overlap with `lower_tables` stays within
    /// `file_gc_compaction_max_overlap_ratio`. Shared by
    /// `find_file_gc_candidate_level`, `find_file_gc_candidate_l0`, and the
    /// wakeup hint so all three apply the identical overlap gate; a mismatch
    /// would wake a tick the candidate scan rejects, or hide a real candidate.
    fn file_gc_overlap_within_limit(
        &self,
        bound: DataBound<'_>,
        candidate_size: u64,
        lower_tables: &[SsTable],
    ) -> bool {
        file_gc_overlap_ratio(bound, candidate_size, lower_tables)
            <= self.opt.file_gc_compaction_max_overlap_ratio()
    }

    /// The reclaim ratio of an L1+ WRITE_CF file at `level` if it is a static
    /// file-GC candidate: ratio passes and, above the bottommost level, its
    /// lower-level overlap stays within the limit. Maturity
    /// (`max_ts <= gc_safe_point`) and the bottommost-level allow_l3 gate are
    /// the caller's. `find_file_gc_candidate_level` and the wakeup hint
    /// share this so their per-file gates cannot drift.
    fn evaluate_file_gc_l1plus_ratio(
        &self,
        scf: &ShardCf,
        tbl: &SsTable,
        level: usize,
        bottommost_level: usize,
    ) -> Option<f64> {
        let ratio = file_gc_ratio(tbl)?;
        if ratio < self.opt.file_gc_compaction_ratio() {
            return None;
        }
        // The bottommost level rewrites in place with no lower-level input, so the
        // overlap-bloat gate only applies above it.
        if level < bottommost_level {
            let lower = scf.get_level(level + 1);
            if !self.file_gc_overlap_within_limit(tbl.data_bound(), tbl.size(), &lower.tables) {
                return None;
            }
        }
        Some(ratio)
    }

    /// The reclaim ratio of an L0 WRITE_CF file if it is a static file-GC
    /// candidate: converted, ratio passes, and its L1 overlap stays within the
    /// limit. Maturity is the caller's. Shared by `find_file_gc_candidate_l0`
    /// and the wakeup hint.
    fn evaluate_file_gc_l0_ratio(&self, data: &ShardData, l0: &L0Table) -> Option<f64> {
        if data
            .col_levels
            .unconverted_l0s
            .iter()
            .any(|u| u.id() == l0.id())
        {
            return None;
        }
        let ratio = l0_file_gc_ratio(l0)?;
        if ratio < self.opt.file_gc_compaction_ratio() {
            return None;
        }
        let l1_tables = &data.get_cf(WRITE_CF).get_level(1).tables;
        if !self.file_gc_overlap_within_limit(l0.data_bound(), l0.size(), l1_tables) {
            return None;
        }
        Some(ratio)
    }

    /// Find a file-level GC candidate SST in WRITE_CF levels
    /// `1..=bottommost_level`, deepest level first. Returns `(level,
    /// max_ratio)`, or `None` on no hit. File-level gates: `max_ts <=
    /// gc_safe_point`, ratio `>= file_gc_compaction_ratio`, and above the
    /// bottommost level overlap `<= file_gc_compaction_max_overlap_ratio`
    /// of its size.
    pub fn find_file_gc_candidate_level(&self, gc_safe_point: u64) -> Option<(usize, f64)> {
        let data = self.get_data();
        let scf = data.get_cf(WRITE_CF);
        let bottommost_level = scf.levels.last()?.level;
        // The lv2+ counters do not cover L1, so a zero combined counter only
        // narrows the scan to L1 instead of skipping the tick.
        let lv2plus_counter = self.lv2plus_tombs.load(Ordering::Relaxed)
            + self.lv2plus_old_entries.load(Ordering::Relaxed);
        let deepest_scan_level = if lv2plus_counter == 0 {
            1
        } else {
            bottommost_level
        };
        // Deeper levels rarely trigger size-driven compaction, so reclaimable
        // data piles up there; scan deepest first.
        for target_level in (1..=deepest_scan_level).rev() {
            // A bottommost-level full drop emits an empty change set; peers whose
            // is_move_down lacks the empty-top_deletes guard panic on it. Gate
            // the whole L3 path until the cluster is fully upgraded.
            if target_level == bottommost_level && !self.opt.file_gc_compaction_allow_l3() {
                continue;
            }
            let lh = scf.get_level(target_level);
            let mut max_ratio = 0.0_f64;
            for tbl in lh.tables.iter() {
                // Maturity is the only gate beyond the shared static gate.
                if tbl.max_ts > gc_safe_point {
                    continue;
                }
                let Some(ratio) =
                    self.evaluate_file_gc_l1plus_ratio(scf, tbl, target_level, bottommost_level)
                else {
                    continue;
                };
                if ratio > max_ratio {
                    max_ratio = ratio;
                }
            }
            if max_ratio > 0.0 {
                return Some((target_level, max_ratio));
            }
        }
        None
    }

    /// Set a file-level GC L1Plus priority if none is set. Returns true if set;
    /// callers should `trigger_compact` only then.
    pub fn try_set_file_gc_triggered_l1_plus_priority(&self, level: usize, ratio: f64) -> bool {
        {
            let mut priority = self.compaction_priority.write().unwrap();
            if priority.is_some() {
                return false;
            }
            *priority = Some(CompactionPriority::L1Plus {
                cf: WRITE_CF as isize,
                level,
                score: file_gc_priority_score(ratio),
                file_gc: true,
            });
        }
        // Log outside the write lock.
        info!(
            "{} trigger file-level GC L1Plus",
            self.tag();
            "level" => level,
            "ratio" => ratio,
        );
        true
    }

    /// L0 counterpart of `find_file_gc_candidate_level`. Returns `(file_id,
    /// ratio)` of the highest-ratio qualifying L0Table, or `None`. The
    /// L1-overlap gate skips files whose reclaimable data spreads thin over
    /// lots of L1 data: little scan benefit for a large compaction. Unconverted
    /// L0s never qualify; `pick_candidate_l0_input_indices` would bail on them
    /// every tick until conversion finishes.
    pub fn find_file_gc_candidate_l0(&self, gc_safe_point: u64) -> Option<(u64, f64)> {
        let data = self.get_data();
        let mut best: Option<(u64, f64)> = None;
        for l0 in &data.l0_tbls {
            // Maturity is the only gate beyond the shared static gate.
            if l0.max_ts() > gc_safe_point {
                continue;
            }
            let Some(ratio) = self.evaluate_file_gc_l0_ratio(&data, l0) else {
                continue;
            };
            if best.is_none_or(|(_, r)| ratio > r) {
                best = Some((l0.id(), ratio));
            }
        }
        best
    }

    /// L0 counterpart of `try_set_file_gc_triggered_l1_plus_priority`,
    /// carrying the candidate file's ID for `pick_candidate_l0_input_indices`.
    pub fn try_set_file_gc_triggered_l0_priority(&self, candidate_id: u64, ratio: f64) -> bool {
        {
            let mut priority = self.compaction_priority.write().unwrap();
            if priority.is_some() {
                return false;
            }
            *priority = Some(CompactionPriority::L0 {
                score: file_gc_priority_score(ratio),
                file_candidate: Some(candidate_id),
            });
        }
        info!(
            "{} trigger file-level GC L0 compaction",
            self.tag();
            "candidate_id" => candidate_id,
            "ratio" => ratio,
        );
        true
    }

    /// Count this shard's WRITE_CF obsolete files (delete tombstones + old
    /// versions) into the size / ratio / overlap distributions, mirroring
    /// `find_file_gc_candidate_level` / `find_file_gc_candidate_l0`. There is
    /// no maturity gate: it counts all obsolete files, not just GC-eligible
    /// ones, because this runs in a per-shard refresh with no GC safe
    /// point. Overlap is counted only for ratio-passing files.
    pub(crate) fn classify_write_cf_obsolete(&self, buckets: &mut ObsoleteFileBuckets) {
        let ratio_threshold = self.opt.file_gc_compaction_ratio();
        let data = self.get_data();
        let scf = data.get_cf(WRITE_CF);
        let Some(bottommost_level) = scf.levels.last().map(|lh| lh.level) else {
            // WRITE_CF always has CF_LEVELS levels; this only guards a
            // level-less shard, where the L0 path's get_level(1) would panic.
            return;
        };
        // L1..=bottommost_level. The bottommost level rewrites in place (no overlap
        // gate); the diagnostic walks it even when allow_l3 is off, since deep
        // levels are where obsolete data piles up.
        for level in 1..=bottommost_level {
            let lh = scf.get_level(level);
            for tbl in lh.tables.iter() {
                if tbl.tombs == 0 && tbl.old_entries == 0 {
                    continue; // clean file, cheap early-out
                }
                let Some(ratio) = file_gc_ratio(tbl) else {
                    continue;
                };
                buckets.size[size_bucket(tbl.old_data_size())] += 1;
                buckets.ratio[ratio_bucket(ratio)] += 1;
                if ratio >= ratio_threshold {
                    let overlap = if level < bottommost_level {
                        let lower_level = scf.get_level(level + 1);
                        file_gc_overlap_ratio(tbl.data_bound(), tbl.size(), &lower_level.tables)
                    } else {
                        0.0
                    };
                    buckets.overlap[overlap_bucket(overlap)] += 1;
                }
            }
        }
        // L0: lower level is L1; skip unconverted L0s like
        // `pick_candidate_l0_input_indices`.
        let l1_tables = &scf.get_level(1).tables;
        for l0 in &data.l0_tbls {
            if l0.tombs() == 0 && l0.old_entries_write_cf() == 0 {
                continue;
            }
            if data
                .col_levels
                .unconverted_l0s
                .iter()
                .any(|u| u.id() == l0.id())
            {
                continue;
            }
            let Some(ratio) = l0_file_gc_ratio(l0) else {
                continue;
            };
            buckets.size[size_bucket(l0.old_data_size_write_cf())] += 1;
            buckets.ratio[ratio_bucket(ratio)] += 1;
            if ratio >= ratio_threshold {
                let overlap = file_gc_overlap_ratio(l0.data_bound(), l0.size(), l1_tables);
                buckets.overlap[overlap_bucket(overlap)] += 1;
            }
        }
    }

    /// The file-level GC wakeup hint: the min candidate `max_ts` among files
    /// passing the cheap static gates, or `u64::MAX` when none qualify or the
    /// hint was parked by an overlap-gated scan.
    pub fn file_gc_candidate_min_max_ts(&self) -> u64 {
        self.file_gc_candidate_min_max_ts.load(Ordering::Relaxed)
    }

    /// Unconditionally set the wakeup hint. Production parks via
    /// `park_file_gc_candidate_hint`; this raw setter is for forcing the hint
    /// in tests.
    pub fn set_file_gc_candidate_min_max_ts(&self, max_ts: u64) {
        self.file_gc_candidate_min_max_ts
            .store(max_ts, Ordering::Relaxed);
    }

    /// Park the wakeup hint at `u64::MAX` after an empty scan, but only if it
    /// still holds `observed` (read before the scan). A concurrent refresh may
    /// have recomputed a fresh candidate `max_ts` in between; the
    /// compare-exchange leaves that fresh value instead of clobbering it, so a
    /// race cannot strand a real candidate until the next refresh.
    pub fn park_file_gc_candidate_hint(&self, observed: u64) {
        let _ = self.file_gc_candidate_min_max_ts.compare_exchange(
            observed,
            u64::MAX,
            Ordering::Relaxed,
            Ordering::Relaxed,
        );
    }

    /// Recompute the candidate hint under the current config. Pub entry for the
    /// leader tick after an online config change; the walk and `data` are
    /// crate-private, so it cannot be inlined at the rfstore call site.
    pub fn refresh_candidate_hint(&self) {
        let guard = self.data.read().unwrap();
        self.refresh_file_gc_candidate_hint(&guard);
    }

    /// Recompute the file-level GC wakeup hint: the min `max_ts` over WRITE_CF
    /// files passing every static file-GC gate (ratio, overlap, allow_l3, and
    /// for L0 already converted) via the same `evaluate_file_gc_*_ratio`
    /// helpers `find_file_gc_candidate_level` and
    /// `find_file_gc_candidate_l0` use, leaving out only maturity (`max_ts
    /// <= gc_safe_point`, which the tick applies against the advancing safe
    /// point). So the hint is the smallest safe point at which a scan will
    /// find a trigger; `u64::MAX` means no file can ever trigger under the
    /// current layout. Sharing the helpers keeps the hint and
    /// `find_file_gc_candidate_level` / `find_file_gc_candidate_l0` from
    /// drifting -- a stricter hint would drop a real trigger, a looser one
    /// would park live candidates.
    fn refresh_file_gc_candidate_hint(&self, data: &ShardData) {
        let mut min_max_ts = u64::MAX;

        let scf = data.get_cf(WRITE_CF);
        if let Some(bottommost_level) = scf.levels.last().map(|lh| lh.level) {
            // Mirror find_file_gc_candidate_level's narrowing: scan L2+ only
            // when the lv2+ counters show reclaimable data there.
            let lv2plus_counter = self.lv2plus_tombs.load(Ordering::Relaxed)
                + self.lv2plus_old_entries.load(Ordering::Relaxed);
            let deepest_scan_level = if lv2plus_counter == 0 {
                1
            } else {
                bottommost_level
            };
            for target_level in 1..=deepest_scan_level {
                // Mirror `find_file_gc_candidate_level`'s bottommost-level allow_l3
                // gate so a shard with only bottommost-level data is not woken for an
                // always-empty scan.
                if target_level == bottommost_level && !self.opt.file_gc_compaction_allow_l3() {
                    continue;
                }
                for tbl in scf.get_level(target_level).tables.iter() {
                    if self
                        .evaluate_file_gc_l1plus_ratio(scf, tbl, target_level, bottommost_level)
                        .is_some()
                    {
                        min_max_ts = min_max_ts.min(tbl.max_ts);
                    }
                }
            }
        }

        for l0 in &data.l0_tbls {
            if self.evaluate_file_gc_l0_ratio(data, l0).is_some() {
                min_max_ts = min_max_ts.min(l0.max_ts());
            }
        }

        self.file_gc_candidate_min_max_ts
            .store(min_max_ts, Ordering::Relaxed);
    }

    fn check_need_gc_lock_file(&self, safe_ts: u64) -> bool {
        let lv1plus_max_lock_ts = self.lv1plus_min_lock_file_max_ts.load(Ordering::Relaxed);
        if lv1plus_max_lock_ts == 0 || lv1plus_max_lock_ts > safe_ts {
            return false;
        }
        let mut priority = self.compaction_priority.write().unwrap();
        if priority.is_some() {
            return false;
        }
        *priority = Some(CompactionPriority::GcLockFile);
        info!("{} trigger gc lock file", self.tag(); "safe_ts" => safe_ts);
        true
    }

    fn check_need_gc_extra_file(&self, safe_ts: u64) -> bool {
        let lv1_max_extra_ts = self.lv1_min_max_extra_ts.load(Ordering::Relaxed);
        if lv1_max_extra_ts == 0 || lv1_max_extra_ts > safe_ts {
            return false;
        }
        let mut priority = self.compaction_priority.write().unwrap();
        if priority.is_some() {
            return false;
        }
        *priority = Some(CompactionPriority::GcExtraFile);
        info!("{} trigger gc extra file", self.tag(); "safe_ts" => safe_ts);
        true
    }

    pub fn has_failed_compaction(&self) -> bool {
        if self.is_compacting() {
            return false;
        }
        let priority = self.compaction_priority.read().unwrap();
        priority.is_some()
    }

    pub(crate) fn get_data(&self) -> ShardData {
        self.data.read().unwrap().clone()
    }

    pub(crate) fn set_data(&self, data: ShardData) {
        self.set_data_opt(data, true);
    }

    pub(crate) fn set_data_opt(&self, data: ShardData, check_update_counter: bool) {
        let mut guard = self.data.write().unwrap();
        if check_update_counter {
            debug_assert_eq!(
                guard.update_counter + 1,
                data.update_counter,
                "{} set_data: update counter mismatch: old: {:?}, new: {:?}",
                self.tag(),
                guard,
                data
            );
            if guard.update_counter + 1 != data.update_counter {
                warn!("{} set_data: update counter mismatch", self.tag(); "old" => ?guard, "new" => ?data);
            }
        }
        *guard = data
    }

    /// Returns a snapshot after requiring its encryption key to be ready.
    ///
    /// This unchecked legacy API panics if the key cannot be made ready. New
    /// callers should prefer [`Self::try_new_snap_access`] so they can
    /// return an error instead; existing callers will be migrated gradually.
    pub fn new_snap_access(&self) -> SnapAccess {
        self.ensure_encryption_key_ready().unwrap_or_else(|err| {
            panic!(
                "[CMEK] {} failed to create snapshot because KMS key is unavailable: {}",
                self.tag(),
                err
            )
        });
        SnapAccess::new(self)
    }

    /// Returns a snapshot, reporting encryption-key readiness failures.
    ///
    /// This is the fallible counterpart of [`Self::new_snap_access`].
    pub fn try_new_snap_access(&self) -> Result<SnapAccess> {
        self.ensure_encryption_key_ready()?;
        Ok(SnapAccess::new(self))
    }

    pub fn get_writable_mem_table_state(
        &self,
    ) -> (
        u64,   // mem_table_size
        usize, // unpersisted_props_size
    ) {
        let guard = self.data.read().unwrap();
        let mem_tbl = &guard.mem_tbls[0];
        let mut size = mem_tbl.size();
        if guard.prepend_keyspace_id().is_some() {
            size += (mem_tbl.skip_list_entries() * KEYSPACE_PREFIX_LEN) as u64;
        }
        (size, mem_tbl.unpersisted_props_size())
    }

    pub fn get_writable_mem_table(&self) -> memtable::CfTable {
        let guard = self.data.read().unwrap();
        guard.get_writable_mem_table().clone()
    }

    /// Replaces memtables while constructing a restore-only engine from a
    /// coherent remote snapshot.
    ///
    /// Callers must install memtables captured from the same `SnapAccess` as
    /// this shard's snapshot metadata before making the engine observable.
    pub fn install_memtables_for_restore(&self, mem_tbls: Vec<CfTable>) -> Result<()> {
        if mem_tbls.is_empty() {
            return Err(Error::Other(box_err!(
                "{} cannot install an empty memtable list for restore",
                self.tag()
            )));
        }
        if !self.opt.for_restore {
            return Err(Error::Other(box_err!(
                "{} can install remote memtables only in a restore engine",
                self.tag()
            )));
        }
        let mut builder = ShardDataBuilder::new(self.get_data());
        builder.set_mem_tbls(mem_tbls);
        self.set_data(builder.build());
        Ok(())
    }

    pub fn get_mem_table_max_version(&self) -> SnapVersion {
        let guard = self.data.read().unwrap();
        guard
            .mem_tbls
            .iter()
            .map(|t| t.get_snap_version())
            .max()
            .unwrap_or_default()
    }

    pub fn has_over_bound_data(&self) -> bool {
        self.get_data().has_over_bound_data()
    }

    pub fn get_trim_over_bound(&self) -> bool {
        self.pending_ops.read().unwrap().trim_over_bound
    }

    pub fn has_pending_manual_major_compaction(&self) -> bool {
        self.pending_ops.read().unwrap().manual_major_compaction != MajorCompactionType::Disable
    }

    pub fn get_del_prefixes(&self) -> Arc<DeletePrefixes> {
        self.pending_ops.read().unwrap().del_prefixes.clone()
    }

    pub fn get_truncate_ts(&self) -> Option<u64> {
        self.pending_ops.read().unwrap().truncate_ts
    }

    pub fn get_storage_class_spec(&self) -> StorageClassSpec {
        self.pending_ops.read().unwrap().storage_class_spec.clone()
    }

    #[cfg(feature = "testexport")]
    pub fn storage_class_spec_equals(&self, spec: &StorageClassSpec) -> bool {
        &self.pending_ops.read().unwrap().storage_class_spec == spec
    }

    pub fn data_all_persisted(&self) -> bool {
        self.data.read().unwrap().all_persisted()
    }

    pub fn tag(&self) -> ShardTag {
        ShardTag::new(self.engine_id, IdVer::new(self.id, self.ver))
    }

    pub fn get_checked_schema_ver(&self) -> i64 {
        self.checked_schema_ver.load(Acquire)
    }

    pub fn set_checked_schema_ver(&self, schema_ver: i64) {
        self.checked_schema_ver.store(schema_ver, Ordering::Release);
    }

    pub fn get_schema_file(&self) -> Option<SchemaFile> {
        self.data.read().unwrap().schema_file.clone()
    }

    pub fn has_columnar_table(&self, table_id: i64) -> bool {
        let data = self.get_data();
        data.schema_file.is_some() && data.columnar_table_ids.contains(&table_id)
    }

    pub fn vector_index_ready(&self, table_id: i64, index_id: i64) -> bool {
        let data = self.get_data();
        let Some(schema_file) = data.schema_file.as_ref() else {
            return false;
        };
        let Some(schema) = schema_file.get_table(table_id) else {
            return false;
        };
        let vec_idx_def = schema
            .vector_indexes
            .iter()
            .find(|idx| idx.index_id == index_id);
        let Some(vec_idx_def) = vec_idx_def else {
            return false;
        };
        let vec_idx = data
            .vector_indexes
            .get(table_id, index_id, vec_idx_def.col_id);
        if vec_idx.is_some() {
            return true;
        }
        // If the table already applied the columnar major compaction, but no
        // columnar files contains the table or the columnar data is too small to
        // build vector index, treat it as ready.
        if data.columnar_table_ids.contains(&table_id)
            && !self.need_build_vector_index(&data, table_id, vec_idx_def, SnapVersion::zero())
        {
            return true;
        }
        false
    }

    pub fn fts_index_ready(&self, table_id: i64, index_id: i64) -> bool {
        let data = self.get_data();
        // The fts index may be deleted, but fts_levels have not yet untracked this
        // change. Therefore, this situation needs to be checked.
        let Some(schema_file) = data.schema_file.as_ref() else {
            return false;
        };
        let Some(schema) = schema_file.get_table(table_id) else {
            return false;
        };
        if !schema
            .fulltext_indexes
            .iter()
            .any(|idx| idx.index_id == index_id)
        {
            return false;
        }
        // We don't check all data being ready because it may never become true when
        // there are continuous write workload. As a trade off, we currently only check
        // historical data being built (i.e. has_tracked_index)
        data.fts_levels.has_tracked_index(table_id, index_id)
    }

    pub fn has_unconverted_l0s(&self) -> bool {
        self.get_data().has_unconverted_l0s()
    }

    pub fn has_vector_index(&self) -> bool {
        !self.get_data().vector_indexes.is_empty()
    }

    pub fn get_vector_index(
        &self,
        table_id: i64,
        index_id: i64,
        col_id: i64,
    ) -> Option<VectorIndex> {
        let data = self.get_data();
        data.vector_indexes.get(table_id, index_id, col_id).cloned()
    }

    pub fn try_get_encryption_key(&self) -> Result<Option<EncryptionKey>> {
        let key_state = self.encryption_state.read().unwrap().clone();
        let encryption_key = key_state.resolve_ready_key().map_err(|_| {
            Error::EncryptionKeyUnavailable(format!(
                "[CMEK] KMS key is unavailable for keyspace {}, shard {}; reject read before accessing encrypted data",
                self.range.keyspace_id,
                self.tag()
            ))
        })?;
        if matches!(key_state, EncryptionState::Pending { .. }) {
            let Some(encryption_key) = &encryption_key else {
                return Ok(None);
            };
            let mut guard = self.encryption_state.write().unwrap();
            if is_same_pending_encryption_state(&guard, &key_state) {
                *guard = EncryptionState::Ready(encryption_key.clone());
            }
        }
        Ok(encryption_key)
    }

    /// Returns the shard's ready key or unresolved key state.
    ///
    /// This is for replay paths that may need to parse encrypted raft entries
    /// but should not require KMS until decryption is actually needed.
    pub fn get_encryption_state(&self) -> EncryptionState {
        self.encryption_state.read().unwrap().clone()
    }

    pub fn set_encryption_state(&self, key_state: EncryptionState) {
        *self.encryption_state.write().unwrap() = key_state;
    }

    /// Ensures the shard-held encryption key can decrypt data.
    pub(crate) fn ensure_encryption_key_ready(&self) -> Result<()> {
        self.try_get_encryption_key().map(|_| ())
    }

    pub fn is_encrypted(&self) -> bool {
        self.encryption_state.read().unwrap().is_encrypted()
    }

    // When recover is called multiple times, the write_sequence may be smaller than
    // the meta.data_sequence, so we need to update it before recover to avoid
    // entries unavailable error.
    pub fn sync_data_sequence(&self, meta: &ShardMeta) {
        if self.get_meta_sequence() == meta.seq && self.get_write_sequence() < meta.data_sequence {
            self.write_sequence
                .store(meta.data_sequence, Ordering::Release);
        }
    }

    pub fn ready_to_compact(&self) -> bool {
        self.is_active() && self.get_initial_flushed()
    }

    pub fn add_parent_data(&self, parent: Arc<Shard>) {
        let shard_data = self.get_data();
        let parent_data = parent.get_data();

        let mem_tbls = self.split_mem_tables(&parent_data.mem_tbls);
        let mem_tbl_vers: Vec<SnapVersion> =
            mem_tbls.iter().map(|tbl| tbl.get_snap_version()).collect();

        for prop_key in PROPERTIES_COPY_FROM_PARENT_IN_RECOVERY {
            if let Some(val) = parent.get_property(prop_key) {
                self.set_property(prop_key, &val);
            }
        }

        let lock_txn_files = parent_data.lock_txn_files.clone();

        info!("{} add parent data", self.tag();
            "mem-tables" => ?mem_tbl_vers,
            "lock_txn_files" => ?lock_txn_files.iter().map(|x| x.start_ts()).collect::<Vec<_>>(),
            "properties" => ?self.properties.multi_get(PROPERTIES_COPY_FROM_PARENT_IN_RECOVERY));
        let mut builder = ShardDataBuilder::new(shard_data);
        builder.set_mem_tbls(mem_tbls);
        builder.set_lock_txn_files(lock_txn_files);
        self.set_data(builder.build());
    }

    pub(crate) fn keyspace_prefix(&self) -> &[u8] {
        if self.keyspace_id > 0 {
            return &self.outer_start[0..KEYSPACE_PREFIX_LEN];
        }
        &[]
    }

    /// Whether the shard is empty.
    ///
    /// NOTE: use with caution.
    /// The empty status of shard is not reliable, as the applying to kvengine
    /// would be late.
    pub(crate) fn is_empty(&self) -> bool {
        let data = self.get_data();
        if data.mem_tbls.iter().any(|mem_tbl| !mem_tbl.is_empty()) {
            return false;
        }
        self.get_estimated_size() == 0
    }

    pub(crate) fn clear_finished_txn_file_refs(&self, version: SnapVersion) {
        if let Some(val) = self.get_property(TXN_FILE_REF) {
            let mut prop = TxnFileRefPropertyHelper::from_property(Some(val)).unwrap();
            prop.clear_finished_txn_file_lock(version);
            self.set_property(TXN_FILE_REF, &prop.marshall());
            debug!("{} clear finished txn file ref", self.tag(); "prop" => ?prop, "version" => version);
        }
    }

    pub(crate) fn get_outdated_schema_ver(&self) -> i64 {
        self.outdated_schema_ver.load(Ordering::Acquire)
    }

    pub(crate) fn set_outdated_schema_ver(&self, ver: i64) {
        self.outdated_schema_ver.store(ver, Ordering::Release);
    }

    pub fn set_last_transit_storage_class_instant_to_now(&self) {
        let now = Instant::now_coarse();
        self.last_transit_storage_class_instant_sec
            .fetch_max(now.second(), Ordering::Relaxed);
    }

    fn last_transit_storage_class_elapsed(&self) -> Duration {
        Instant::from_timespec_second_coarse(
            self.last_transit_storage_class_instant_sec
                .load(Ordering::Relaxed),
        )
        .saturating_elapsed()
    }

    pub fn should_try_storage_class_transition(&self) -> bool {
        self.get_storage_class_spec().can_transit_to_other_sc()
            && self.last_transit_storage_class_elapsed() > self.opt.ia.auto_ia_check_interval.0
    }

    pub fn try_transit_storage_class(&self) -> usize /* tables_transited_to_ia */ {
        let spec = self.get_storage_class_spec();
        if !spec.can_transit_to_other_sc() {
            return 0;
        }

        let tag = self.tag();
        let data = self.get_data();
        let current_ts = TimeStamp::now().into_inner();

        let mut transited_to_ia = 0;
        for level in &data.cfs[WRITE_CF].levels {
            for t in level.tables.as_ref() {
                let Some(auto_file) = t.try_get_auto_ia_file() else {
                    // Must be updating storage class spec.
                    info!("{} try_transit_sc: skip, not auto IA file", tag);
                    return transited_to_ia;
                };

                match auto_file.try_transit(current_ts) {
                    TransitResult::NoChange => {}
                    TransitResult::TransitedToIa => transited_to_ia += 1,
                }
            }
        }
        transited_to_ia
    }

    #[inline]
    #[allow(unused_variables)]
    pub(crate) fn record_max_used_gc_safe_point(&self, gc_safe_point: TimeStamp) {
        #[cfg(any(test, feature = "testexport"))]
        {
            let mut current = self.max_used_gc_safe_point.wl();
            if *current < gc_safe_point {
                *current = gc_safe_point;
            }
        }
    }
}

/// Returns a ready encryption key from shard properties.
/// - `ENCRYPTION_META_KEY` is the additive CMEK format.
/// - `ENCRYPTION_KEY` is the legacy exported-key format.
///
/// Unlike `encryption_state_from_shard_properties`, this contacts KMS
/// for CMEK metadata and returns an error if the key cannot be resolved.
/// If both properties exist unexpectedly, prefer `ENCRYPTION_META_KEY` and
/// warn so the shard can keep running while operators investigate.
pub fn encryption_key_from_shard_properties(
    props: &kvenginepb::Properties,
    encryption_key_manager: Arc<EncryptionKeyManager>,
    legacy_master_key: &MasterKey,
    keyspace_id: Option<u32>,
) -> cloud_encryption::Result<Option<EncryptionKey>> {
    encryption_state_from_shard_properties(
        props,
        encryption_key_manager,
        legacy_master_key,
        keyspace_id,
    )
    .resolve_ready_key()
}

/// Loads shard encryption properties without requiring KMS availability.
///
/// Legacy exported keys are returned as ready keys. CMEK metadata is only
/// registered and returned as a pending `EncryptionState`.
pub fn encryption_state_from_shard_properties(
    props: &kvenginepb::Properties,
    encryption_key_manager: Arc<EncryptionKeyManager>,
    legacy_master_key: &MasterKey,
    keyspace_id: Option<u32>,
) -> EncryptionState {
    let encryption_meta = get_shard_property(ENCRYPTION_META_KEY, props);
    let legacy_key = get_shard_property(ENCRYPTION_KEY, props);

    if encryption_meta.is_some() && legacy_key.is_some() {
        warn!(
            "shard carries both encryption property formats, prefer ENCRYPTION_META_KEY";
            "shard_id" => props.shard_id,
            "preferred" => ENCRYPTION_META_KEY,
            "ignored" => ENCRYPTION_KEY,
        );
    }

    let Some((mode, value)) = encryption_meta
        .map(|value| (EncryptionMethod::Metadata, value))
        .or_else(|| legacy_key.map(|value| (EncryptionMethod::Legacy, value)))
    else {
        return EncryptionState::Disabled;
    };
    encryption_key_manager.set_mode_if_unset(mode);

    match mode {
        EncryptionMethod::Metadata => {
            encryption_state_from_meta_bytes(encryption_key_manager, &value, keyspace_id)
        }
        EncryptionMethod::Legacy => legacy_master_key
            .decrypt_encryption_key(&value)
            .map(EncryptionState::Ready)
            .unwrap_or_else(|err| EncryptionState::from_error(keyspace_id, err)),
    }
}

impl BoundedDataSet for Shard {
    fn data_bound(&self) -> DataBound<'_> {
        self.range.data_bound()
    }
}

pub(crate) struct ShardDataBuilder {
    old: ShardData,
    range: Option<ShardRange>,
    inner_key_off: Option<usize>,
    mem_tbls: Option<Vec<CfTable>>,
    l0_tbls: Option<Vec<L0Table>>,
    cfs: Option<[ShardCf; 3]>,
    unloaded_tbls: Option<HashMap<u64, FileMeta>>,
    limiter: Option<RegionLimiter>,
    blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
    lock_txn_files: Option<Vec<TxnFile>>,
    schema: Option<(
        i64, // schema_version
        u64, // restore_version
        Option<SchemaFile>,
    )>,
    columnar_levels: Option<ColumnarLevels>,
    vector_indexes: Option<VectorIndexes>,
    columnar_table_ids: Option<Vec<i64>>,
    fts_levels: Option<FtsLevels>,
    persisted_version: Option<SnapVersion>,
}

impl ShardDataBuilder {
    pub(crate) fn new(old: ShardData) -> Self {
        Self {
            old,
            range: None,
            inner_key_off: None,
            mem_tbls: None,
            l0_tbls: None,
            cfs: None,
            unloaded_tbls: None,
            limiter: None,
            blob_tbls: None,
            lock_txn_files: None,
            schema: None,
            columnar_levels: None,
            vector_indexes: None,
            columnar_table_ids: None,
            fts_levels: None,
            persisted_version: None,
        }
    }

    pub(crate) fn set_range(&mut self, range: ShardRange) {
        self.range = Some(range);
    }

    pub(crate) fn set_inner_key_off(&mut self, inner_key_off: usize) {
        self.inner_key_off = Some(inner_key_off);
    }

    pub(crate) fn set_mem_tbls(&mut self, mem_tbls: Vec<CfTable>) {
        self.mem_tbls = Some(mem_tbls);
    }

    pub(crate) fn set_l0_tbls(&mut self, l0_tbls: Vec<L0Table>) {
        self.l0_tbls = Some(l0_tbls);
    }

    pub(crate) fn set_cfs(&mut self, cfs: [ShardCf; 3]) {
        self.cfs = Some(cfs);
    }

    pub(crate) fn set_unloaded_tbls(&mut self, unloaded_tbls: HashMap<u64, FileMeta>) {
        self.unloaded_tbls = Some(unloaded_tbls);
    }

    pub(crate) fn with_new_limiter(&mut self) {
        self.limiter = Some(RegionLimiter::new_from(&self.old.limiter))
    }

    pub(crate) fn set_blob_tbls(&mut self, blob_tbls: HashMap<u64, BlobTable>) {
        self.blob_tbls = Some(Arc::new(blob_tbls));
    }

    pub(crate) fn set_lock_txn_files(&mut self, lock_txn_files: Vec<TxnFile>) {
        self.lock_txn_files = Some(lock_txn_files);
    }

    pub(crate) fn set_schema(
        &mut self,
        schema_version: i64,
        restore_version: u64,
        schema_file: Option<SchemaFile>,
    ) {
        debug_assert!(
            schema_file
                .as_ref()
                .is_none_or(|x| x.get_version() == schema_version)
        );
        self.schema = Some((schema_version, restore_version, schema_file));
    }

    #[inline]
    pub(crate) fn clear_schema(&mut self, restore_version: u64) {
        self.schema = Some((0, restore_version, None));
    }

    pub(crate) fn set_columnar_levels(&mut self, columnar_levels: ColumnarLevels) {
        self.columnar_levels = Some(columnar_levels);
    }

    pub(crate) fn set_vector_indexes(&mut self, vector_indexes: VectorIndexes) {
        self.vector_indexes = Some(vector_indexes);
    }

    pub(crate) fn set_columnar_table_ids(&mut self, columnar_table_ids: Vec<i64>) {
        self.columnar_table_ids = Some(columnar_table_ids);
    }

    pub(crate) fn set_fts_levels(&mut self, fts_levels: FtsLevels) {
        self.fts_levels = Some(fts_levels);
    }

    pub(crate) fn set_persisted_version(&mut self, persisted_version: SnapVersion) {
        self.persisted_version = Some(persisted_version);
    }

    pub(crate) fn build(mut self) -> ShardData {
        let should_sync_fts_layouts = self.schema.is_some() || self.fts_levels.is_some();
        let (schema_version, restore_version, schema_file) =
            self.schema.take().unwrap_or_else(|| {
                (
                    self.old.schema_version,
                    self.old.restore_version,
                    self.old.schema_file.clone(),
                )
            });
        let col_levels = self
            .columnar_levels
            .take()
            .unwrap_or_else(|| self.old.col_levels.clone());
        let mut vector_indexes = self
            .vector_indexes
            .take()
            .unwrap_or_else(|| self.old.vector_indexes.clone());
        // Collect the extra columnar files to read for vector index.
        // Note: there may have columnar files not transformed to vector index with
        // smaller version than l2_snap_version after region merge.
        for vec_idx in vector_indexes.get_mut_all() {
            vec_idx.update_extra_columnar_files(&col_levels);
        }
        let mut fts_levels = self
            .fts_levels
            .take()
            .unwrap_or_else(|| (*self.old.fts_levels).clone());
        if should_sync_fts_layouts {
            fts_levels.sync_tracked_index_layouts(schema_file.as_ref());
        }
        ShardData::new(
            self.range.take().unwrap_or_else(|| self.old.range.clone()),
            self.inner_key_off.take().unwrap_or(self.old.inner_key_off),
            self.mem_tbls
                .take()
                .unwrap_or_else(|| self.old.mem_tbls.clone()),
            self.l0_tbls
                .take()
                .unwrap_or_else(|| self.old.l0_tbls.clone()),
            self.blob_tbls
                .take()
                .unwrap_or_else(|| self.old.blob_tbl_map.clone()),
            self.cfs.take().unwrap_or_else(|| self.old.cfs.clone()),
            self.unloaded_tbls
                .take()
                .unwrap_or_else(|| self.old.unloaded_tbls.clone()),
            self.lock_txn_files
                .take()
                .unwrap_or_else(|| self.old.lock_txn_files.clone()),
            self.limiter
                .take()
                .unwrap_or_else(|| self.old.limiter.clone()),
            self.old.update_counter + 1,
            schema_version,
            restore_version,
            schema_file,
            col_levels,
            vector_indexes,
            self.columnar_table_ids
                .take()
                .unwrap_or_else(|| self.old.columnar_table_ids.clone()),
            fts_levels,
            self.persisted_version
                .take()
                .unwrap_or(self.old.persisted_version),
        )
    }
}

#[derive(Clone)]
pub(crate) struct ShardData {
    pub(crate) core: Arc<ShardDataCore>,
}

impl fmt::Debug for ShardData {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ShardData")
            .field("range", &self.range)
            .field("inner_key_off", &self.inner_key_off)
            .field("sst_files", &self.get_all_sst_files())
            .field("txn_chunks", &self.get_txn_chunks())
            .field("columnar_files", &self.get_all_columnar_files())
            .field("columnar_table_ids", &self.columnar_table_ids)
            .field("mem_table_max_ts", &self.get_mem_table_max_ts())
            .field("mem_table_size", &self.get_mem_table_size())
            .field("l0_total_size", &self.get_l0_total_size())
            .field("unloaded_tbls", &self.unloaded_tbls)
            .field("update_counter", &self.update_counter)
            .field("schema_version", &self.schema_version)
            .field("restore_version", &self.restore_version)
            .field("schema_file", &self.schema_file_id())
            .field("columnar_table_ids", &self.columnar_table_ids)
            .finish()
    }
}

impl Deref for ShardData {
    type Target = ShardDataCore;

    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

impl BoundedDataSet for ShardData {
    fn data_bound(&self) -> DataBound<'_> {
        self.range.data_bound()
    }
}

impl ShardData {
    pub(crate) fn new_empty(
        shard_range: ShardRange,
        inner_key_offset: usize,
        limiter: RegionLimiter,
        memtable_implementation: MemTableImplementation,
    ) -> Self {
        Self::new(
            shard_range,
            inner_key_offset,
            vec![CfTable::with_implementation(memtable_implementation)],
            vec![],
            Arc::new(HashMap::new()),
            [ShardCf::new(0), ShardCf::new(1), ShardCf::new(2)],
            HashMap::new(),
            vec![],
            limiter,
            INITIAL_UPDATE_COUNTER,
            0,
            0,
            None,
            ColumnarLevels::new(),
            VectorIndexes::default(),
            vec![],
            FtsLevels::default(),
            SnapVersion::zero(),
        )
    }

    pub(crate) fn new(
        range: ShardRange,
        inner_key_off: usize,
        mem_tbls: Vec<memtable::CfTable>,
        l0_tbls: Vec<L0Table>,
        blob_tbl_map: Arc<HashMap<u64, BlobTable>>,
        cfs: [ShardCf; 3],
        unloaded_tbls: HashMap<u64, FileMeta>,
        lock_txn_files: Vec<TxnFile>,
        limiter: RegionLimiter,
        update_counter: u64,
        schema_version: i64,
        restore_version: u64,
        schema_file: Option<SchemaFile>,
        col_levels: ColumnarLevels,
        vector_indexes: VectorIndexes,
        columnar_table_ids: Vec<i64>,
        fts_levels: FtsLevels,
        persisted_version: SnapVersion,
    ) -> Self {
        assert!(!mem_tbls.is_empty());

        Self {
            core: Arc::new(ShardDataCore {
                range,
                inner_key_off,
                mem_tbls,
                lock_txn_files,
                l0_tbls,
                blob_tbl_map,
                cfs,
                unloaded_tbls,
                limiter,
                update_counter,
                schema_version,
                restore_version,
                schema_file,
                col_levels,
                vector_indexes,
                columnar_table_ids,
                fts_levels: Arc::new(fts_levels),
                persisted_version,
            }),
        }
    }

    #[inline]
    pub fn can_use_ia(cf: usize, level: usize) -> bool {
        cf == WRITE_CF && level > 0
    }
}

pub(crate) struct ShardDataCore {
    pub(crate) range: ShardRange,
    pub(crate) inner_key_off: usize,
    pub(crate) mem_tbls: Vec<memtable::CfTable>,
    pub(crate) lock_txn_files: Vec<TxnFile>,
    pub(crate) l0_tbls: Vec<L0Table>,
    pub(crate) blob_tbl_map: Arc<HashMap<u64, BlobTable>>,
    pub(crate) cfs: [ShardCf; 3],
    /// Tables that are not loaded from DFS yet.
    pub(crate) unloaded_tbls: HashMap<u64, FileMeta>,
    pub limiter: RegionLimiter,
    pub update_counter: u64,
    /// `schema_version` is used to indicate the schema version even if the
    /// `schema_file` is None.
    pub(crate) schema_version: i64,
    /// `restore_version` is used to indicate the restore version of the shard
    /// even if the `schema_file` is None.
    pub(crate) restore_version: u64,
    pub(crate) schema_file: Option<SchemaFile>,
    pub(crate) col_levels: ColumnarLevels,
    pub(crate) vector_indexes: VectorIndexes,
    // columnar_table_ids is the list of columnar table ids that have been columnar major
    // compacted.
    pub(crate) columnar_table_ids: Vec<i64>,
    pub(crate) fts_levels: Arc<FtsLevels>,
    pub(crate) persisted_version: SnapVersion,
}

impl Deref for ShardDataCore {
    type Target = ShardRange;

    fn deref(&self) -> &Self::Target {
        &self.range
    }
}

impl ShardDataCore {
    pub fn get_writable_mem_table(&self) -> &memtable::CfTable {
        self.mem_tbls.first().unwrap()
    }

    pub(crate) fn get_cf(&self, cf: usize) -> &ShardCf {
        &self.cfs[cf]
    }

    pub(crate) fn get_all_sst_files(&self) -> Vec<u64> {
        let (files, ia_files) = self.get_local_sst_files(false);
        debug_assert!(ia_files.is_empty());
        files
    }

    /// Returns the authoritative set of immutable DFS files referenced by this
    /// coherent shard state.
    ///
    /// Keep GC reporting and metadata-only snapshot export on this single
    /// enumeration path so a newly introduced file type cannot be protected by
    /// one path but omitted by the other.
    pub(crate) fn get_all_files_with_type(&self) -> Vec<(FileType, u64)> {
        let mut files = Vec::new();
        files.extend(
            self.get_all_sst_files()
                .into_iter()
                .map(|id| (FileType::Sst, id)),
        );
        files.extend(
            self.get_all_blob_files()
                .into_iter()
                .map(|id| (FileType::Blob, id)),
        );
        files.extend(
            self.get_txn_chunks()
                .into_iter()
                .map(|id| (FileType::TxnChunk, id)),
        );
        files.extend(
            self.get_all_col_files()
                .into_iter()
                .map(|id| (FileType::Columnar, id)),
        );
        files.extend(
            self.get_all_vec_idx_files()
                .into_iter()
                .map(|id| (FileType::VectorIndex, id)),
        );
        files.extend(
            self.fts_levels
                .l0()
                .iter()
                .chain(self.fts_levels.l1())
                .map(|file| (FileType::FtsPackedFile, file.id())),
        );
        files.extend(
            self.fts_levels
                .l2()
                .values()
                .flatten()
                .map(|file| (FileType::FtsDedicatedFile, file.id())),
        );
        files.extend(
            self.schema_file
                .as_ref()
                .map(|file| (FileType::Schema, file.get_file_id())),
        );
        files.sort_unstable();
        files.dedup();
        files
    }

    // Set `distinguish_ia` to true when need to distinguish local & ia files.
    // Note: `local_files` & `ia_files` may have duplicated ids.
    pub(crate) fn get_local_sst_files(
        &self,
        distinguish_ia: bool,
    ) -> (
        Vec<u64>, // local_files
        Vec<u64>, // ia_files
    ) {
        let mut local_files = Vec::new();
        let mut ia_files = Vec::new();

        for l0 in &self.l0_tbls {
            local_files.push(l0.id());
        }

        let shard_is_sync = self.is_sync();
        self.for_each_level(|cf, lh| {
            let cf_is_sync = shard_is_sync || !ShardData::can_use_ia(cf, lh.level);
            for tbl in lh.tables.iter() {
                debug_assert_eq!(tbl.is_sync(), cf_is_sync, "{}", tbl.id());
                if !distinguish_ia || tbl.is_sync() {
                    local_files.push(tbl.id())
                } else {
                    ia_files.push(tbl.id());
                    if !tbl.is_storage_class_ia() {
                        // Async tables with storage class not IA, must have local file.
                        local_files.push(tbl.id());
                    }
                }
            }
            false
        });
        local_files.sort_unstable();
        ia_files.sort_unstable();
        (local_files, ia_files)
    }

    pub(crate) fn get_local_blob_files(
        &self,
        distinguish_ia: bool,
    ) -> (
        Vec<u64>, // local_files
        Vec<u64>, // ia_files
    ) {
        let mut local_files = Vec::new();
        let mut ia_files = Vec::new();

        for (&file_id, tbl) in self.blob_tbl_map.iter() {
            if !distinguish_ia || tbl.is_sync() {
                local_files.push(file_id);
                continue;
            }

            ia_files.push(file_id);
            if !tbl.is_storage_class_ia() {
                // Async tables with storage class not IA, must have local file.
                local_files.push(file_id);
            }
        }

        local_files.sort_unstable();
        ia_files.sort_unstable();
        (local_files, ia_files)
    }

    pub(crate) fn get_all_blob_files(&self) -> Vec<u64> {
        let mut blob_file_ids = self.blob_tbl_map.keys().copied().collect::<Vec<_>>();
        blob_file_ids.sort_unstable();
        blob_file_ids
    }

    pub(crate) fn get_txn_chunks(&self) -> Vec<u64> {
        let mut files = Vec::new();
        for txn_file in &self.lock_txn_files {
            files.extend_from_slice(&txn_file.chunk_ids());
        }
        for mem_tbl in &self.mem_tbls {
            let skl = mem_tbl.get_cf(WRITE_CF);
            for txn_file in skl.get_txn_files() {
                files.extend_from_slice(&txn_file.chunk_ids());
            }
        }
        files.sort_unstable();
        files
    }

    pub(crate) fn get_all_col_files(&self) -> Vec<u64> {
        let mut col_ids = vec![];
        self.for_each_columnar_level(|cl| {
            for tbl in cl.files.iter() {
                col_ids.push(tbl.id());
            }
            false
        });
        col_ids
    }

    pub(crate) fn get_all_vec_idx_files(&self) -> Vec<u64> {
        let mut vec_idx_file_ids = vec![];
        for vec_idx in self.vector_indexes.get_all() {
            for file in &vec_idx.files {
                vec_idx_file_ids.push(file.file_id())
            }
        }
        vec_idx_file_ids
    }

    pub(crate) fn get_all_fts_files(&self) -> Vec<u64> {
        let mut fts_file_ids = vec![];
        for fts_l0_file in self.fts_levels.l0() {
            fts_file_ids.push(fts_l0_file.id());
        }
        for fts_l1_file in self.fts_levels.l1() {
            fts_file_ids.push(fts_l1_file.id());
        }
        for fts_l2_file in self.fts_levels.l2().values() {
            fts_file_ids.extend(fts_l2_file.iter().map(|file| file.id()));
        }
        fts_file_ids
    }

    #[inline]
    pub(crate) fn has_txn_file_locks(&self) -> bool {
        !self.lock_txn_files.is_empty()
    }

    #[inline]
    pub(crate) fn get_lock_txn_files(&self) -> &[TxnFile] {
        &self.lock_txn_files
    }

    pub(crate) fn for_each_level<F>(&self, mut f: F)
    where
        F: FnMut(usize /* cf */, &LevelHandler) -> bool, // stopped
    {
        for cf in 0..NUM_CFS {
            let scf = self.get_cf(cf);
            for lh in scf.levels.as_slice() {
                if f(cf, lh) {
                    return;
                }
            }
        }
    }

    pub(crate) fn for_each_columnar_level<F>(&self, mut f: F)
    where
        F: FnMut(&ColumnarLevel) -> bool, // stopped
    {
        for cl in self.col_levels.levels.iter() {
            if f(cl) {
                return;
            }
        }
    }

    pub(crate) fn get_col_table_counts(&self, level: usize) -> usize {
        assert!(level < 3);
        self.col_levels.levels[level].files.len()
    }

    pub(crate) fn get_all_columnar_files(&self) -> Vec<u64> {
        let mut files = Vec::new();
        for cl in self.col_levels.levels.iter() {
            for f in cl.files.iter() {
                files.push(f.id());
            }
        }
        files.sort_unstable();
        files
    }

    pub(crate) fn get_columnar_snap_version(&self) -> SnapVersion {
        let mut max_version = SnapVersion::zero();
        for cl in self.col_levels.levels.iter() {
            if cl.level < 2 {
                for f in cl.files.iter() {
                    max_version = cmp::max(
                        max_version,
                        f.get_snap_version().unwrap_or(SnapVersion::zero()),
                    );
                }
            } else {
                max_version = cmp::max(max_version, self.col_levels.l2_snap_version);
            }
        }
        max_version
    }

    // Return (max_ts).
    pub(crate) fn get_mem_table_max_ts(&self) -> u64 {
        let mut max_ts = 0;
        self.mem_tbls.iter().for_each(|t| {
            max_ts = cmp::max(max_ts, t.data_max_ts());
        });
        max_ts
    }

    fn get_mem_table_size(&self) -> u64 {
        self.mem_tbls.iter().map(|t| t.size()).sum()
    }

    fn get_mem_table_memory_usage(&self) -> u64 {
        self.mem_tbls.iter().map(|t| t.memory_usage()).sum()
    }

    pub(crate) fn get_l0_total_size(&self) -> u64 {
        self.get_l0_stats().data_size
    }

    pub(crate) fn get_l0_stats(&self) -> LevelStatsLite {
        let mut stats = LevelStatsLite::default();
        self.l0_tbls.iter().for_each(|l0| {
            if self.data_bound().contains_bound(l0.data_bound()) {
                stats.data_size += l0.size();
                stats.entries += l0.entries();
                stats.kv_size += l0.kv_size();
            } else {
                // If the L0 table is not fully contained in the shard data bound, count half
                // of its size, entries, and kv_size as a split-boundary estimate.
                stats.data_size += l0.size() / 2;
                stats.entries += l0.entries() / 2;
                stats.kv_size += l0.kv_size() / 2;
            }
            stats.max_ts = cmp::max(stats.max_ts, l0.max_ts());
        });
        stats
    }

    pub(crate) fn get_level_stats(
        &self,
        cf: usize,
        level: &LevelHandler,
        sc_spec: &StorageClassSpec,
    ) -> LevelStatsLite {
        let mut stats = LevelStatsLite::default();
        level.tables.iter().enumerate().for_each(|(i, tbl)| {
            if self.is_over_bound_table(level, i, tbl) {
                stats.data_size += tbl.size() / 2;
                stats.blob_size += tbl.in_use_total_blob_size / 2;
                stats.entries += tbl.entries as u64 / 2;
                if cf == WRITE_CF {
                    stats.kv_size += tbl.kv_size / 2;
                    if level.level >= 2 {
                        stats.lv2plus_max_ts = cmp::max(stats.lv2plus_max_ts, tbl.max_ts);
                        stats.lv2plus_tombs += tbl.tombs as u64 / 2;
                        stats.lv2plus_entries_write_cf += tbl.entries as u64 / 2;
                        stats.lv2plus_data_size += tbl.size() / 2;
                        stats.lv2plus_old_data_size += tbl.old_data_size() / 2;
                        stats.lv2plus_old_entries += tbl.old_entries as u64 / 2;
                    }
                }
            } else {
                stats.data_size += tbl.size();
                stats.blob_size += tbl.in_use_total_blob_size;
                stats.entries += tbl.entries as u64;
                if cf == WRITE_CF {
                    stats.kv_size += tbl.kv_size;
                    if level.level >= 2 {
                        stats.lv2plus_max_ts = cmp::max(stats.lv2plus_max_ts, tbl.max_ts);
                        stats.lv2plus_tombs += tbl.tombs as u64;
                        stats.lv2plus_entries_write_cf += tbl.entries as u64;
                        stats.lv2plus_data_size += tbl.size();
                        stats.lv2plus_old_data_size += tbl.old_data_size();
                        stats.lv2plus_old_entries += tbl.old_entries as u64;
                    }
                } else if cf == LOCK_CF {
                    let max_lock_ts = tbl.get_max_lock_ts();
                    if stats.lv1plus_min_max_lock_ts == 0
                        || stats.lv1plus_min_max_lock_ts > max_lock_ts
                    {
                        stats.lv1plus_min_max_lock_ts = max_lock_ts;
                    }
                } else if cf == EXTRA_CF {
                    #[allow(clippy::collapsible_if)]
                    if stats.lv1_min_max_extra_ts == 0 || stats.lv1_min_max_extra_ts > tbl.max_ts {
                        stats.lv1_min_max_extra_ts = tbl.max_ts;
                    }
                }
            }
            stats.max_ts = cmp::max(stats.max_ts, tbl.max_ts);
        });
        if cf == WRITE_CF {
            let (ia_data_size, ia_kv_size) =
                self.calc_storage_class_stats_for_level(level, sc_spec, &stats);
            stats.ia_data_size = ia_data_size;
            stats.ia_kv_size = ia_kv_size;
        }
        stats
    }

    fn calc_storage_class_stats_for_level(
        &self,
        level: &LevelHandler,
        sc_spec: &StorageClassSpec,
        stats: &LevelStatsLite,
    ) -> (u64, u64) {
        if sc_spec.must_be_ia() {
            (stats.data_size, stats.kv_size)
        } else if sc_spec.can_be_ia() {
            let mut ia_data_size = 0;
            let mut ia_kv_size = 0;
            level.tables.iter().enumerate().for_each(|(i, tbl)| {
                if tbl.is_storage_class_ia() {
                    if self.is_over_bound_table(level, i, tbl) {
                        ia_data_size += tbl.size() / 2;
                        ia_kv_size += tbl.kv_size / 2;
                    } else {
                        ia_data_size += tbl.size();
                        ia_kv_size += tbl.kv_size;
                    }
                }
            });
            (ia_data_size, ia_kv_size)
        } else {
            (0, 0)
        }
    }

    pub(crate) fn get_columnar_level_stats(&self, level: usize) -> LevelStatsLite {
        let level = &self.col_levels.levels[level];
        let mut stats = LevelStatsLite::default();
        level.files.iter().for_each(|tbl| {
            stats.columnar_size += tbl.size();
            stats.columnar_kv_size += tbl.get_estimated_kv_size() as u64;
        });
        stats
    }

    fn is_over_bound_table(&self, level: &LevelHandler, i: usize, tbl: &SsTable) -> bool {
        let is_bound = i == 0 || i == level.tables.len() - 1;
        is_bound && !self.data_bound().contains_bound(tbl.data_bound())
    }

    pub fn all_persisted(&self) -> bool {
        self.mem_tbls.len() == 1
            && self.mem_tbls[0].size() == 0
            && self.mem_tbls[0].unpersisted_props_size() == 0
    }

    pub(crate) fn has_mem_over_bound_data(&self) -> bool {
        for mem_tbl in &self.mem_tbls {
            for cf in 0..NUM_CFS {
                let skl = mem_tbl.get_cf(cf);
                if skl.has_over_bound_data(self.inner_start(), self.inner_end()) {
                    return true;
                }
            }
        }
        false
    }

    pub(crate) fn has_file_over_bound_data(&self) -> bool {
        let shard_bound = self.data_bound();
        for l0 in &self.l0_tbls {
            if !shard_bound.contains_bound(l0.data_bound()) {
                return true;
            }
        }
        for cf in 0..NUM_CFS {
            let scf = &self.cfs[cf];
            for level in &scf.levels {
                if let Some(level_bound) = level.get_data_bound() {
                    if !shard_bound.contains_bound(level_bound) {
                        return true;
                    }
                }
            }
        }
        for cl in &self.col_levels.levels {
            for col_file in &cl.files {
                if !shard_bound.contains_bound(col_file.data_bound()) {
                    return true;
                }
            }
        }
        if self.has_over_bound_fts_data() {
            return true;
        }

        false
    }

    fn has_over_bound_fts_data(&self) -> bool {
        let shard_bound = self.data_bound();
        for fts_l0_file in self.fts_levels.l0() {
            if !shard_bound.contains_bound(fts_l0_file.data_bound()) {
                return true;
            }
        }
        for fts_l1_file in self.fts_levels.l1() {
            if !shard_bound.contains_bound(fts_l1_file.data_bound()) {
                return true;
            }
        }
        for fts_l2_file in self.fts_levels.l2().values() {
            if fts_l2_file
                .iter()
                .any(|file| !shard_bound.contains_bound(file.data_bound()))
            {
                return true;
            }
        }
        false
    }

    pub fn has_over_bound_data(&self) -> bool {
        self.has_mem_over_bound_data() || self.has_file_over_bound_data()
    }

    pub fn refresh_for_limiter(&self, tag: &ShardTag) {
        let mem_table_memory_usage = self.get_mem_table_memory_usage();
        let l0_table_size = self.get_l0_total_size();
        self.limiter.update_usage(
            tag,
            mem_table_memory_usage,
            l0_table_size,
            tikv_util::sys::memory_usage_over_throttling_level,
        );
    }

    pub fn get_unconverted_l0s(&self) -> Vec<u64> {
        self.col_levels
            .unconverted_l0s
            .iter()
            .map(|l0| l0.id())
            .collect()
    }

    pub fn has_unconverted_l0s(&self) -> bool {
        !self.col_levels.unconverted_l0s.is_empty()
    }

    pub(crate) fn has_too_many_unconverted_l0s(&self) -> bool {
        self.col_levels.unconverted_l0s.len() > MAX_UNCONVERTED_L0_FILE_COUNTS
    }

    pub(crate) fn is_sync(&self) -> bool {
        self.cfs[WRITE_CF].is_sync()
    }

    pub fn prepend_keyspace_id(&self) -> Option<u32> {
        (self.range.keyspace_id > 0 && self.inner_key_off == 0).then_some(self.range.keyspace_id)
    }

    pub fn schema_file_id(&self) -> Option<u64> {
        self.schema_file.as_ref().map(|x| x.get_file_id())
    }

    pub fn get_columnar_table_ids_in_schema(&self) -> Vec<i64> {
        if self.schema_file.is_none() {
            return vec![];
        }
        let mut table_ids: Vec<i64> = self
            .columnar_table_ids
            .iter()
            .filter(|&&id| {
                self.schema_file
                    .as_ref()
                    .unwrap()
                    .contains_columnar_table(id)
            })
            .copied()
            .collect();
        table_ids.sort();
        table_ids
    }
}

pub fn store_u64(ptr: &AtomicU64, val: u64) {
    ptr.store(val, Release);
}

pub fn load_u64(ptr: &AtomicU64) -> u64 {
    ptr.load(Acquire)
}

pub fn store_bool(ptr: &AtomicBool, val: bool) {
    ptr.store(val, Release)
}

pub fn load_bool(ptr: &AtomicBool) -> bool {
    ptr.load(Acquire)
}

pub(crate) struct ShardCfBuilder {
    levels: Vec<LevelHandlerBuilder>,
}

impl ShardCfBuilder {
    pub(crate) fn new(cf: usize) -> Self {
        Self {
            levels: vec![LevelHandlerBuilder::new(); CF_LEVELS[cf]],
        }
    }

    pub(crate) fn build(&mut self) -> ShardCf {
        let mut levels = Vec::with_capacity(self.levels.len());
        for i in 0..self.levels.len() {
            levels.push(self.levels[i].build(i + 1))
        }
        ShardCf { levels }
    }

    pub(crate) fn add_table(&mut self, tbl: SsTable, level: usize) {
        self.levels[level - 1].add_table(tbl)
    }
}

#[derive(Clone)]
struct LevelHandlerBuilder {
    tables: Option<Vec<SsTable>>,
}

impl LevelHandlerBuilder {
    fn new() -> Self {
        Self {
            tables: Some(vec![]),
        }
    }

    fn build(&mut self, level: usize) -> LevelHandler {
        let mut tables = self.tables.take().unwrap();
        tables.sort_by(|a, b| a.smallest().cmp(&b.smallest()));
        LevelHandler::new(level, tables)
    }

    fn add_table(&mut self, tbl: SsTable) {
        if self.tables.is_none() {
            self.tables = Some(vec![])
        }
        self.tables.as_mut().unwrap().push(tbl)
    }
}

#[derive(Clone)]
pub(crate) struct ShardCf {
    pub(crate) levels: Vec<LevelHandler>,
}

impl ShardCf {
    pub(crate) fn new(cf: usize) -> Self {
        let mut levels = vec![];
        for j in 1..=CF_LEVELS[cf] {
            levels.push(LevelHandler::new(j, vec![]));
        }
        Self { levels }
    }

    pub(crate) fn get_level(&self, level: usize) -> &LevelHandler {
        &self.levels[level - 1]
    }

    pub(crate) fn set_level(&mut self, level_handler: LevelHandler) {
        let level_idx = level_handler.level - 1;
        self.levels[level_idx] = level_handler
    }

    pub(crate) fn size(&self) -> u64 {
        self.levels
            .iter()
            .map(|lh| lh.tables.iter().map(|tbl| tbl.size()).sum::<u64>())
            .sum()
    }

    pub(crate) fn is_sync(&self) -> bool {
        for l in &self.levels {
            if !l.tables.is_empty() {
                return l.tables[0].is_sync();
            }
        }
        true
    }
}

#[derive(Default, Clone)]
pub struct LevelHandler {
    pub(crate) tables: Arc<Vec<SsTable>>,
    pub(crate) level: usize,
    pub(crate) max_ts: u64,
}

impl LevelHandler {
    pub fn new(level: usize, tables: Vec<SsTable>) -> Self {
        let mut max_ts = 0;
        for tbl in &tables {
            if max_ts < tbl.max_ts {
                max_ts = tbl.max_ts;
            }
        }
        Self {
            tables: Arc::new(tables),
            level,
            max_ts,
        }
    }

    #[maybe_async::both]
    pub async fn get(
        &self,
        key: InnerKey<'_>,
        version: u64,
        key_hash: u64,
        out_val_owner: &mut Vec<u8>,
    ) -> table::Value {
        match self.get_table(key) {
            Some(t) => {
                self.get_in_table(key, version, key_hash, t, out_val_owner)
                    .await
            }
            None => table::Value::new(),
        }
    }

    #[maybe_async::both]
    async fn get_in_table(
        &self,
        key: InnerKey<'_>,
        version: u64,
        key_hash: u64,
        tbl: &SsTable,
        out_val_owner: &mut Vec<u8>,
    ) -> table::Value {
        tbl.get(key, version, key_hash, out_val_owner, self.level)
            .await
    }

    // Note: the `key` may be on the left outside the returned sstable.
    pub(crate) fn get_table(&self, key: InnerKey<'_>) -> Option<&SsTable> {
        let idx = search(self.tables.len(), |i| self.tables[i].biggest() >= key);
        if idx >= self.tables.len() {
            return None;
        }
        Some(&self.tables[idx])
    }

    pub(crate) fn get_table_by_id(&self, id: u64) -> Option<SsTable> {
        for tbl in self.tables.as_slice() {
            if tbl.id() == id {
                return Some(tbl.clone());
            }
        }
        None
    }

    pub(crate) fn check_order(&self, cf: usize, tag: ShardTag) {
        let tables = &self.tables;
        if tables.len() <= 1 {
            return;
        }
        for i in 0..(tables.len() - 1) {
            let ti = &tables[i];
            let tj = &tables[i + 1];
            if ti.smallest() > ti.biggest()
                || ti.smallest() >= tj.smallest()
                || ti.biggest() >= tj.biggest()
            {
                panic!(
                    "[{}] check table fail table ti {} smallest {:?}, biggest {:?}, tj {} smallest {:?}, biggest {:?}, cf: {}, level: {}",
                    tag,
                    ti.id(),
                    ti.smallest(),
                    ti.biggest(),
                    tj.id(),
                    tj.smallest(),
                    tj.biggest(),
                    cf,
                    self.level,
                );
            }
        }
    }

    #[maybe_async::both]
    pub(crate) async fn get_newer(
        &self,
        key: InnerKey<'_>,
        version: u64,
        key_hash: u64,
        out_val_owner: &mut Vec<u8>,
    ) -> table::Value {
        if self.max_ts <= version {
            return table::Value::new();
        }
        if let Some(tbl) = self.get_table(key) {
            return tbl
                .get_newer(key, version, key_hash, out_val_owner, self.level)
                .await;
        }
        table::Value::new()
    }

    pub(crate) fn get_data_bound(&self) -> Option<DataBound<'_>> {
        if self.tables.is_empty() {
            return None;
        }
        let first = self.tables.first().unwrap();
        let last = self.tables.last().unwrap();
        Some(DataBound::new(first.smallest(), last.biggest(), true))
    }

    pub(crate) fn range_blocks_size(&self, ranges: &[(Bytes, Bytes)]) -> usize {
        let mut blocks_size = 0;
        let mut prev_table_id = 0;
        let mut prev_block_right = 0;
        for ran in ranges {
            let start_key = InnerKey::from_outer_key(&ran.0);
            let end_key = InnerKey::from_outer_end_key(&ran.1);
            let bound = DataBound::new(start_key, end_key, false);
            let (left, right) = bound.get_overlap_data_sets(&self.tables);
            if left == right {
                continue;
            }
            let overlap_tables = &self.tables[left..right];
            let first_table = overlap_tables.first().unwrap();
            let first_idx = first_table.load_index(true);
            let first_avg_block_size = first_table.kv_size as usize / first_idx.num_blocks();
            let first_block_left = first_idx.seek_block(start_key).saturating_sub(1);
            let first_block_right = if overlap_tables.len() == 1 {
                first_idx.seek_block_bigger_or_equal(end_key)
            } else {
                first_idx.num_blocks()
            };
            blocks_size += (first_block_right - first_block_left) * first_avg_block_size;
            if first_table.id() == prev_table_id && first_block_left < prev_block_right {
                // do not count duplicated block.
                blocks_size -= first_avg_block_size;
            }
            if overlap_tables.len() == 1 {
                prev_table_id = first_table.id();
                prev_block_right = first_block_right;
                continue;
            }
            if overlap_tables.len() > 2 {
                let middle_tables = &overlap_tables[1..overlap_tables.len() - 1];
                for middle_table in middle_tables {
                    blocks_size += middle_table.kv_size as usize;
                }
            }
            let last_table = overlap_tables.last().unwrap();
            let last_idx = last_table.load_index(true);
            let last_block_right = last_idx.seek_block_bigger_or_equal(end_key);
            let last_avg_block_size = last_table.kv_size as usize / last_idx.num_blocks();
            blocks_size += last_block_right * last_avg_block_size;
            prev_table_id = last_table.id();
            prev_block_right = last_block_right;
        }
        blocks_size
    }
}

#[derive(Default, Clone, Debug)]
pub struct Properties {
    m: Arc<RwLock<HashMap<String, Bytes>>>,
}

impl Properties {
    pub fn new() -> Self {
        Self {
            m: Arc::new(RwLock::new(HashMap::new())),
        }
    }

    pub fn set(&self, key: &str, val: &[u8]) {
        let mut m = self.m.write().unwrap();
        m.insert(key.to_string(), Bytes::copy_from_slice(val));
    }

    pub fn set_bytes(&self, key: &str, val: Bytes) {
        let mut m = self.m.write().unwrap();
        m.insert(key.to_string(), val);
    }

    pub fn get(&self, key: &str) -> Option<Bytes> {
        let m = self.m.read().unwrap();
        let bin = m.get(key)?;
        Some(bin.clone())
    }

    pub fn multi_get(&self, keys: &[&str]) -> Vec<(String, Bytes)> {
        let m = self.m.read().unwrap();
        m.iter()
            .filter(|(k, _)| keys.contains(&k.as_str()))
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect()
    }

    pub fn remove(&self, key: &str) -> Option<Bytes> {
        let mut m = self.m.write().unwrap();
        m.remove(key)
    }

    pub fn to_pb(&self, shard_id: u64) -> kvenginepb::Properties {
        let mut props = kvenginepb::Properties::new();
        props.shard_id = shard_id;
        let m = self.m.read().unwrap();
        m.iter().for_each(|(k, v)| {
            props.keys.push(k.clone());
            props.values.push(v.to_vec());
        });
        props
    }

    pub fn apply_pb(self, props: &kvenginepb::Properties) -> Self {
        let keys = props.get_keys();
        let vals = props.get_values();
        for i in 0..keys.len() {
            let key = &keys[i];
            let val = &vals[i];
            self.set(key, val.as_slice());
        }
        self
    }

    // Complement properties from `props` if it's currently not set in self.
    pub fn complement_merge(&mut self, props: Self) {
        let other_m = props.m.read().unwrap();
        // copy to vector to avoid potential deadlock.
        let entries: Vec<(String, Bytes)> = other_m
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        drop(other_m);
        let mut self_m = self.m.write().unwrap();
        for (k, v) in entries {
            self_m.entry(k).or_insert(v);
        }
    }
}

impl PartialEq for Properties {
    fn eq(&self, other: &Self) -> bool {
        let self_m = self.m.read().unwrap();
        let other_m = other.m.read().unwrap();
        if self_m.len() != other_m.len() {
            return false;
        }
        self_m
            .iter()
            .all(|(k, v)| other_m.get(k).is_some_and(|other_v| v == other_v))
    }
}

pub fn get_shard_property(key: &str, props: &kvenginepb::Properties) -> Option<Vec<u8>> {
    debug_assert_eq!(props.get_keys().len(), props.get_values().len());
    let keys = props.get_keys();
    for i in 0..keys.len() {
        if key == keys[i] {
            return Some(props.get_values()[i].clone());
        }
    }
    None
}

pub fn set_shard_property(key: &str, props: &mut kvenginepb::Properties, val: Vec<u8>) {
    debug_assert_eq!(props.get_keys().len(), props.get_values().len());
    let keys = props.get_keys();
    for i in 0..keys.len() {
        if key == keys[i] {
            props.values[i] = val;
            return;
        }
    }
    props.keys.push(key.to_string());
    props.values.push(val);
}

/// Returns the ingest id if it is a legacy ingest request from the TiDB BR
/// tool.
#[inline]
pub fn is_legacy_ingest(ingest_files: &kvenginepb::IngestFiles) -> Option<Vec<u8>> {
    get_shard_property(INGEST_ID_KEY, ingest_files.get_properties())
}

pub fn get_splitting_start_end<'a: 'b, 'b>(
    start: &'a [u8],
    end: &'a [u8],
    split_keys: &'b [Vec<u8>],
    i: usize,
) -> (&'b [u8], &'b [u8]) {
    let start_key = if i != 0 {
        split_keys[i - 1].as_slice()
    } else {
        start as &'b [u8]
    };
    let end_key = if i == split_keys.len() {
        end
    } else {
        split_keys[i].as_slice()
    };
    (start_key, end_key)
}

#[derive(Clone)]
pub struct DeletePrefixes {
    pub(crate) prefixes: Vec<Vec<u8>>,
    // prefixes_nexts are prefix-next keys of prefixes, i.e., [prefixes[i], prefixes_nexts[i]) is
    // the range that should be destroyed.
    prefixes_nexts: Vec<Vec<u8>>,
    pub(crate) schedule_at: u64,
    pub(crate) keyspace_prefix_len: usize,
}

impl std::fmt::Debug for DeletePrefixes {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("DeletePrefixes")
            .field(
                "prefixes",
                &self
                    .prefixes
                    .iter()
                    .map(|p| log_wrappers::Value::key(p))
                    .collect::<Vec<_>>(),
            )
            .field(
                "prefixes_nexts",
                &self
                    .prefixes_nexts
                    .iter()
                    .map(|p| log_wrappers::Value::key(p))
                    .collect::<Vec<_>>(),
            )
            .field("schedule_at", &self.schedule_at)
            .field("keyspace_prefix_len", &self.keyspace_prefix_len)
            .finish()
    }
}

impl DeletePrefixes {
    pub fn new_with_keyspace_id(keyspace_id: u32) -> Self {
        Self {
            prefixes: vec![],
            prefixes_nexts: vec![],
            schedule_at: 0,
            keyspace_prefix_len: Self::keyspace_id_to_keyspace_prefix_len(keyspace_id),
        }
    }

    fn keyspace_id_to_keyspace_prefix_len(keyspace_id: u32) -> usize {
        if keyspace_id > 0 {
            KEYSPACE_PREFIX_LEN
        } else {
            0
        }
    }

    fn gen_prefixes_nexts(prefixes: &[Vec<u8>]) -> Vec<Vec<u8>> {
        prefixes
            .iter()
            .map(|p| {
                let mut p_n = p.to_vec();
                tidb_query_common::util::convert_to_prefix_next(&mut p_n);
                p_n
            })
            .collect()
    }

    pub fn is_empty(&self) -> bool {
        self.prefixes.is_empty()
    }

    pub fn after_scheduled_time(&self) -> bool {
        tikv_util::time::precise_time_ns() > self.schedule_at
    }

    #[must_use]
    pub fn merge_prefix(&self, prefix: &[u8]) -> Self {
        assert!(prefix.len() >= self.keyspace_prefix_len);

        let mut new_prefixes = vec![];
        for old_prefix in &self.prefixes {
            if prefix.starts_with(old_prefix) {
                // old prefix covers new prefix, do not need to change.
                return self.clone();
            }
            if old_prefix.starts_with(prefix) {
                // new prefix covers old prefix. old prefix can be dropped.
                continue;
            }
            new_prefixes.push(old_prefix.clone());
        }
        new_prefixes.push(prefix.to_vec());
        new_prefixes.sort_unstable();
        let new_prefixes_nexts = DeletePrefixes::gen_prefixes_nexts(&new_prefixes);
        Self {
            prefixes: new_prefixes,
            prefixes_nexts: new_prefixes_nexts,
            schedule_at: 0,
            keyspace_prefix_len: self.keyspace_prefix_len,
        }
    }

    pub fn merge_prefix_in_place(&mut self, prefix: &[u8]) {
        let merged = self.merge_prefix(prefix);
        *self = merged;
    }

    /// Merges multiple prefixes & self into a new one.
    ///
    /// `sorted_prefixes` must be sorted.
    ///
    /// If `sorted_prefixes` contains a prefix that is covered by another prefix
    /// in it, the covered one will not be dropped.
    ///
    /// `schedule_at` & `inner_key_off` are copied from self.
    #[must_use]
    fn merge_prefixes<'a, S: AsRef<[u8]> + 'a>(
        &self,
        sorted_prefixes: impl Iterator<Item = S>,
    ) -> Self {
        let mut new_prefixes = Vec::with_capacity(self.prefixes.len());

        let mut first_iter = self.prefixes.iter();
        let mut first_opt = first_iter.next();
        let mut second_iter = sorted_prefixes;
        let mut second_opt = second_iter.next();

        while let (Some(first), Some(second)) = (first_opt, second_opt.as_ref()) {
            let second = second.as_ref();
            if first.as_slice().starts_with(second) {
                // `second` covers `first`
                first_opt = first_iter.next();
            } else if second.starts_with(first) {
                // `first` covers `second`
                second_opt = second_iter.next();
            } else {
                // `first` and `second` are not covered by each other
                if first.as_slice() < second {
                    new_prefixes.push(first.clone());
                    first_opt = first_iter.next();
                } else {
                    new_prefixes.push(second.to_vec());
                    second_opt = second_iter.next();
                }
            }
        }

        while let Some(first) = first_opt {
            new_prefixes.push(first.clone());
            first_opt = first_iter.next();
        }
        while let Some(second) = second_opt {
            new_prefixes.push(second.as_ref().to_vec());
            second_opt = second_iter.next();
        }

        let new_prefixes_nexts = DeletePrefixes::gen_prefixes_nexts(&new_prefixes);
        Self {
            prefixes: new_prefixes,
            prefixes_nexts: new_prefixes_nexts,
            schedule_at: self.schedule_at,
            keyspace_prefix_len: self.keyspace_prefix_len,
        }
    }

    /// Merge another `DeletePrefixes` into self.
    ///
    /// `schedule_at` & `inner_key_off` are unchanged.
    pub fn merge(&mut self, other: &Self) {
        let merged = self.merge_prefixes(other.prefixes.iter().map(|p| p.as_slice()));
        *self = merged;
    }

    /// Removes all prefixes in the `other` one.
    pub fn split(&self, other: &DeletePrefixes) -> Self {
        let mut new = self.clone();
        new.prefixes.retain(|p| !other.prefixes.contains(p));
        new.prefixes_nexts = DeletePrefixes::gen_prefixes_nexts(&new.prefixes);
        new
    }

    pub fn marshal(&self) -> Vec<u8> {
        let mut buf = Vec::new();
        for prefix in &self.prefixes {
            buf.put_u16_le(prefix.len() as u16);
            buf.extend_from_slice(prefix);
        }
        buf
    }

    pub fn unmarshal(mut data: &[u8], keyspace_id: u32) -> Self {
        let mut prefixes = vec![];
        while !data.is_empty() {
            let len = data.get_u16_le() as usize;
            prefixes.push(data[..len].to_vec());
            data.advance(len);
        }
        let prefixes_nexts = DeletePrefixes::gen_prefixes_nexts(&prefixes);
        Self {
            prefixes,
            prefixes_nexts,
            schedule_at: 0,
            keyspace_prefix_len: Self::keyspace_id_to_keyspace_prefix_len(keyspace_id),
        }
    }

    pub fn build_split(&self, start: &[u8], end: &[u8]) -> Self {
        let prefixes: Vec<_> = self
            .prefixes
            .iter()
            .filter(|prefix| {
                (start <= prefix.as_slice() && prefix.as_slice() < end) || start.starts_with(prefix)
            })
            .cloned()
            .collect();
        let prefixes_nexts = DeletePrefixes::gen_prefixes_nexts(&prefixes);
        let scheduled_at = self.schedule_at;
        Self {
            prefixes,
            prefixes_nexts,
            schedule_at: scheduled_at,
            keyspace_prefix_len: self.keyspace_prefix_len,
        }
    }

    pub(crate) fn cover_full_keyspace(&self, keyspace_prefix: &[u8]) -> bool {
        let keyspace_prefix_len = self.keyspace_prefix_len;
        self.prefixes.iter().any(|p| {
            if keyspace_prefix_len >= p.len() {
                return true;
            }
            keyspace_prefix.starts_with(p.as_slice())
        })
    }

    pub(crate) fn cover_prefix(&self, inner_prefix: InnerKey<'_>) -> bool {
        let keyspace_prefix_len = self.keyspace_prefix_len;
        self.prefixes.iter().any(|p| {
            if keyspace_prefix_len >= p.len() {
                return true;
            }
            let inner_p = InnerKey::from_outer_key(p);
            inner_prefix.starts_with(inner_p.deref())
        })
    }

    pub(crate) fn cover_range(&self, inner_start: InnerKey<'_>, inner_end: InnerKey<'_>) -> bool {
        let keyspace_prefix_len = self.keyspace_prefix_len;
        self.prefixes.iter().any(|p| {
            if keyspace_prefix_len >= p.len() {
                return true;
            }
            let inner_p = InnerKey::from_outer_key(p);
            inner_start.starts_with(inner_p.deref()) && inner_end.starts_with(inner_p.deref())
        })
    }

    pub(crate) fn inner_delete_bounds(&self) -> impl Iterator<Item = DataBound<'_>> {
        let keyspace_prefix_len = self.keyspace_prefix_len;
        self.prefixes
            .iter()
            .zip(self.prefixes_nexts.iter())
            .map(move |(p, p_n)| {
                if keyspace_prefix_len >= p.len() {
                    DataBound::new(
                        InnerKey::from_inner_buf(&[]),
                        InnerKey::from_inner_buf(GLOBAL_SHARD_END_KEY),
                        false,
                    )
                } else {
                    DataBound::new(
                        InnerKey::from_outer_key(p),
                        InnerKey::from_outer_end_key(p_n),
                        false,
                    )
                }
            })
    }

    /// Get ranges that are not covered by the delete prefixes.
    pub fn get_complementary_ranges(
        &self,
        outer_start: &[u8],
        outer_end: &[u8],
    ) -> Vec<(Vec<u8>, Vec<u8>)> {
        let mut ranges = Vec::with_capacity(self.prefixes.len() + 1);
        let mut start = outer_start.to_vec();
        for (prefix, prefix_next) in self.prefixes.iter().zip(self.prefixes_nexts.iter()) {
            if prefix_next.as_slice() <= outer_start {
                continue;
            }
            if prefix.as_slice() >= outer_end {
                break;
            }
            // `start` is possible to be equal to `prefix` here.
            // e.q. prefixes: [00001, 00002] will generate iterations (00001, 00002),
            // (00002, 00003).
            if &start < prefix {
                ranges.push((start.clone(), prefix.clone()));
            }
            start = prefix_next.clone();
        }
        if start.as_slice() < outer_end {
            ranges.push((start, outer_end.to_vec()));
        }
        ranges
    }

    pub fn rewrite_range_prefix(&mut self, prefix: &[u8]) {
        assert_eq!(
            prefix.len(),
            self.keyspace_prefix_len,
            "unexpected prefix.len()"
        );

        // If the prefixes is empty or the prefixes no need to update, return.
        if self.prefixes.is_empty() || self.prefixes.first().unwrap().starts_with(prefix) {
            return;
        }
        let offset = self.keyspace_prefix_len;
        self.prefixes.iter_mut().for_each(|p| {
            p[0..offset].copy_from_slice(prefix);
        });
        self.prefixes_nexts.iter_mut().for_each(|p| {
            p[0..offset].copy_from_slice(prefix);
        });
    }
}

pub fn merge_del_prefixes_if_needed(
    first: Option<Bytes>,
    second: Option<Bytes>,
    keyspace_id: u32,
) -> Option<Vec<u8>> {
    if first.is_none() && second.is_none() {
        return None;
    }

    let mut first_prefixes = first
        .map(|b| DeletePrefixes::unmarshal(b.chunk(), keyspace_id))
        .unwrap_or_else(|| DeletePrefixes::new_with_keyspace_id(keyspace_id));
    let second_prefixes = second
        .map(|b| DeletePrefixes::unmarshal(b.chunk(), keyspace_id))
        .unwrap_or_else(|| DeletePrefixes::new_with_keyspace_id(keyspace_id));
    for prefix in &second_prefixes.prefixes {
        first_prefixes = first_prefixes.merge_prefix(prefix);
    }
    Some(first_prefixes.marshal())
}

#[derive(Clone, Default, PartialEq)]
pub struct ShardRange {
    pub outer_start: Bytes,
    pub outer_end: Bytes,
    pub keyspace_id: u32,
}

impl BoundedDataSet for ShardRange {
    fn data_bound(&self) -> DataBound<'_> {
        DataBound::new(self.inner_start(), self.inner_end(), false)
    }
}

impl ShardRange {
    pub fn new(outer_start: &[u8], outer_end: &[u8]) -> Self {
        let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(outer_start).unwrap_or_default();
        Self {
            outer_start: Bytes::from(outer_start.to_vec()),
            outer_end: Bytes::from(outer_end.to_vec()),
            keyspace_id,
        }
    }

    /// Returns the keyspace ID at the exclusive end, attributing an exact
    /// non-zero keyspace boundary to the preceding keyspace.
    pub fn get_end_keyspace_id(&self) -> u32 {
        let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(&self.outer_end).unwrap_or_default();
        if self.outer_end.len() == KEYSPACE_PREFIX_LEN {
            keyspace_id.saturating_sub(1)
        } else {
            keyspace_id
        }
    }

    pub(crate) fn from_snap(snap: &kvenginepb::Snapshot) -> Self {
        Self::new(snap.get_outer_start(), snap.get_outer_end())
    }

    pub fn inner_start(&self) -> InnerKey<'_> {
        InnerKey::from_outer_key(&self.outer_start)
    }

    pub fn inner_end(&self) -> InnerKey<'_> {
        InnerKey::from_outer_end_key(&self.outer_end)
    }

    pub fn keyspace_prefix(&self) -> &[u8] {
        &self.outer_start[..self.keyspace_prefix_len()]
    }

    pub(crate) fn keyspace_prefix_len(&self) -> usize {
        if self.keyspace_id > 0 {
            KEYSPACE_PREFIX_LEN
        } else {
            0
        }
    }

    pub fn to_outer_key(&self, inner_key: InnerKey<'_>) -> Vec<u8> {
        [self.keyspace_prefix(), inner_key.deref()].concat()
    }
}

impl fmt::Debug for ShardRange {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ShardRange")
            .field(
                "outer_start",
                &log_wrappers::hex_encode_upper(&self.outer_start),
            )
            .field(
                "outer_end",
                &log_wrappers::hex_encode_upper(&self.outer_end),
            )
            .field("keyspace_id", &self.keyspace_id)
            .finish()
    }
}

#[cfg(test)]
mod tests {
    use protobuf::Message;

    use super::*;

    #[test]
    fn test_del_prefixes() {
        for keyspace_id in [0, 1] {
            let assert_prefix_invariant = |del_prefix: &DeletePrefixes| {
                assert_eq!(del_prefix.prefixes.len(), del_prefix.prefixes_nexts.len());
                assert_eq!(
                    del_prefix.prefixes.len(),
                    del_prefix.inner_delete_bounds().count()
                );
                assert!(del_prefix.inner_delete_bounds().all(|bound| {
                    let mut p_c = bound.lower_bound.to_vec();
                    tidb_query_common::util::convert_to_prefix_next(&mut p_c);
                    p_c == bound.upper_bound.deref()
                }));
            };

            let mut del_prefix = DeletePrefixes::new_with_keyspace_id(keyspace_id);
            assert_prefix_invariant(&del_prefix);
            assert!(del_prefix.is_empty());
            del_prefix = del_prefix.merge_prefix("0000101".as_bytes());
            assert!(!del_prefix.is_empty());
            assert_eq!(del_prefix.prefixes.len(), 1);
            assert_prefix_invariant(&del_prefix);
            for prefix in ["00001010", "0000101"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(del_prefix.cover_prefix(inner_prefix));
            }
            for prefix in ["000010", "0000103"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(!del_prefix.cover_prefix(inner_prefix));
            }

            del_prefix = del_prefix.merge_prefix("0000105".as_bytes());
            assert_prefix_invariant(&del_prefix);
            let bin = del_prefix.marshal();
            del_prefix = DeletePrefixes::unmarshal(&bin, keyspace_id);
            assert_prefix_invariant(&del_prefix);
            assert_eq!(del_prefix.prefixes.len(), 2);
            for prefix in ["00001010", "0000101", "00001050", "0000105"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(del_prefix.cover_prefix(inner_prefix));
            }
            for prefix in ["000010", "0000103"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(!del_prefix.cover_prefix(inner_prefix));
            }

            del_prefix = del_prefix.merge_prefix("000010".as_bytes());
            assert_prefix_invariant(&del_prefix);
            assert_eq!(del_prefix.prefixes.len(), 1);
            for prefix in ["000010", "0000101", "0000103", "0000104"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(del_prefix.cover_prefix(inner_prefix));
            }
            for prefix in ["00001", "000011"] {
                let inner_prefix = InnerKey::from_outer_key(prefix.as_bytes());
                assert!(!del_prefix.cover_prefix(inner_prefix));
            }

            del_prefix = DeletePrefixes::new_with_keyspace_id(keyspace_id);
            del_prefix = del_prefix.merge_prefix("0000101".as_bytes());
            del_prefix = del_prefix.merge_prefix("0000102".as_bytes());
            assert_prefix_invariant(&del_prefix);
            for (start, end) in [("0000101", "00001011"), ("0000102", "00001022")] {
                let inner_start = InnerKey::from_outer_key(start.as_bytes());
                let inner_end = InnerKey::from_outer_key(end.as_bytes());
                assert!(del_prefix.cover_range(inner_start, inner_end));
            }
            for (start, end) in [
                ("000099", "0000100"),
                ("0000101", "0000102"),
                ("0000102", "0000103"),
            ] {
                let inner_start = InnerKey::from_outer_key(start.as_bytes());
                let inner_end = InnerKey::from_outer_key(end.as_bytes());
                assert!(!del_prefix.cover_range(inner_start, inner_end));
            }

            del_prefix = DeletePrefixes::new_with_keyspace_id(keyspace_id);
            del_prefix = del_prefix.merge_prefix("0000101".as_bytes());
            del_prefix = del_prefix.merge_prefix("00001033".as_bytes());
            del_prefix = del_prefix.merge_prefix("00001055".as_bytes());
            del_prefix = del_prefix.merge_prefix("0000107".as_bytes());
            assert_prefix_invariant(&del_prefix);

            let split_del_range =
                del_prefix.build_split("00001033".as_bytes(), "00001055".as_bytes());
            assert_prefix_invariant(&split_del_range);
            assert_eq!(split_del_range.prefixes.len(), 1);
            assert_eq!(&split_del_range.prefixes[0], "00001033".as_bytes());

            let split_del_range =
                del_prefix.build_split("00001034".as_bytes(), "00001055".as_bytes());
            assert_prefix_invariant(&split_del_range);
            assert_eq!(split_del_range.prefixes.len(), 0);

            let split_del_range =
                del_prefix.build_split("000010334".as_bytes(), "000010555".as_bytes());
            assert_prefix_invariant(&split_del_range);
            assert_eq!(split_del_range.prefixes.len(), 2);
            assert_eq!(&split_del_range.prefixes[0], "00001033".as_bytes());
            assert_eq!(&split_del_range.prefixes[1], "00001055".as_bytes());

            del_prefix = DeletePrefixes::new_with_keyspace_id(keyspace_id)
                .merge_prefix("0000100".as_bytes())
                .merge_prefix("0000200".as_bytes())
                .merge_prefix("0000300".as_bytes());
            assert_prefix_invariant(&del_prefix);
            del_prefix = del_prefix.split(
                &DeletePrefixes::new_with_keyspace_id(keyspace_id)
                    .merge_prefix("0000100".as_bytes()),
            );
            assert_prefix_invariant(&del_prefix);
            assert_eq!(del_prefix.prefixes.len(), 2);
            assert!(!del_prefix.cover_prefix(InnerKey::from_outer_key("0000100".as_bytes(),)));
            assert!(del_prefix.cover_prefix(InnerKey::from_outer_key("0000200".as_bytes(),)));
            assert!(del_prefix.cover_prefix(InnerKey::from_outer_key("0000300".as_bytes(),)));
            del_prefix = del_prefix.split(
                &DeletePrefixes::new_with_keyspace_id(keyspace_id)
                    .merge_prefix("0000200".as_bytes())
                    .merge_prefix("0000300".as_bytes()),
            );
            assert_prefix_invariant(&del_prefix);
            assert_eq!(del_prefix.prefixes.len(), 0);
            assert!(!del_prefix.cover_prefix(InnerKey::from_outer_key("0000100".as_bytes(),)));
            assert!(!del_prefix.cover_prefix(InnerKey::from_outer_key("0000200".as_bytes(),)));
            assert!(!del_prefix.cover_prefix(InnerKey::from_outer_key("0000300".as_bytes(),)));
        }
    }

    #[test]
    fn test_encryption_key_from_shard_properties_prefers_meta_when_mixed() {
        let master_key = MasterKey::new(&[7; 32]);
        let legacy_key = master_key.generate_encryption_key();
        let data_key = master_key
            .generate_encryption_key_v2()
            .with_key_id(1)
            .unwrap();

        let mut meta = kvenginepb::EncryptionMeta::default();
        meta.set_keyspace_id(11);
        meta.mut_current().set_data_key_id(1);
        let mut master_key_pb = kvenginepb::MasterKey::default();
        master_key_pb.set_vendor("test".into());
        master_key_pb.set_cmek_id("cmek-11".into());
        master_key_pb.set_ciphertext(master_key.export().to_vec());
        meta.set_master_key(master_key_pb);
        let mut data_key_pb = kvenginepb::DataKey::default();
        data_key_pb.set_ciphertext(data_key.export());
        meta.mut_data_keys().insert(1, data_key_pb);

        let mut props = kvenginepb::Properties::default();
        props.shard_id = 42;
        props.mut_keys().push(ENCRYPTION_KEY.to_string());
        props.mut_values().push(legacy_key.export());
        props.mut_keys().push(ENCRYPTION_META_KEY.to_string());
        props.mut_values().push(meta.write_to_bytes().unwrap());

        let encryption_key = encryption_key_from_shard_properties(
            &props,
            Arc::new(EncryptionKeyManager::new()),
            &master_key,
            None,
        )
        .unwrap()
        .expect("metadata encryption property should resolve to a ready key");
        assert_eq!(
            encryption_key.get_type(),
            cloud_encryption::KEY_TYPE_AES_256_CTR
        );
    }

    #[test]
    fn test_del_prefixes_merge_prefix() {
        for keyspace_id in [0, 1] {
            let mut first = DeletePrefixes::new_with_keyspace_id(keyspace_id);
            first = first.merge_prefix("0000101".as_bytes());
            first = first.merge_prefix("00001033".as_bytes());
            first = first.merge_prefix("00001055".as_bytes());
            first = first.merge_prefix("0000107".as_bytes());

            let mut second = DeletePrefixes::new_with_keyspace_id(keyspace_id);
            second = second.merge_prefix("0000101".as_bytes());
            second = second.merge_prefix("00001022".as_bytes());
            second = second.merge_prefix("00001044".as_bytes());
            second = second.merge_prefix("00001066".as_bytes());
            second = second.merge_prefix("0000107".as_bytes());

            let merged = merge_del_prefixes_if_needed(
                Some(Bytes::from(first.marshal())),
                Some(Bytes::from(second.marshal())),
                keyspace_id,
            )
            .unwrap();
            let merged_del_prefixes = DeletePrefixes::unmarshal(&merged, keyspace_id);
            assert_eq!(
                merged_del_prefixes.prefixes,
                vec![
                    "0000101".as_bytes(),
                    "00001022".as_bytes(),
                    "00001033".as_bytes(),
                    "00001044".as_bytes(),
                    "00001055".as_bytes(),
                    "00001066".as_bytes(),
                    "0000107".as_bytes()
                ]
            );

            let rev_merged = merge_del_prefixes_if_needed(
                Some(Bytes::from(second.marshal())),
                Some(Bytes::from(first.marshal())),
                keyspace_id,
            );
            let rev_merged_del_prefixes =
                DeletePrefixes::unmarshal(&rev_merged.unwrap(), keyspace_id);
            assert_eq!(
                merged_del_prefixes.prefixes,
                rev_merged_del_prefixes.prefixes,
            );
        }
    }

    #[test]
    fn test_del_prefixes_merge_prefixes() {
        let cases = vec![
            (
                vec![], // sorted_prefixes
                vec![], // expected prefixes
            ),
            (vec!["00005000"], vec!["00005000"]),
            (vec![], vec!["00005000"]),
            (
                vec!["00005000", "00005001", "00006"],
                vec!["00005000", "00005001", "00006"],
            ),
            (
                vec!["00004000", "0000500", "00007"],
                vec!["00004000", "0000500", "00006", "00007"],
            ),
            (vec!["0000", "0001"], vec!["0000", "0001"]),
            (vec!["00001"], vec!["0000", "0001"]),
        ];

        let mut del_prefix = DeletePrefixes::new_with_keyspace_id(1);
        for (idx, (sorted_prefixes, expected_prefixes)) in cases.into_iter().enumerate() {
            del_prefix.schedule_at = idx as u64;
            del_prefix = del_prefix.merge_prefixes(sorted_prefixes.iter().map(|p| p.as_bytes()));
            let expected_prefixes = expected_prefixes
                .iter()
                .map(|p| p.as_bytes().to_vec())
                .collect::<Vec<_>>();
            assert_eq!(del_prefix.prefixes, expected_prefixes, "case {}", idx);
            assert_eq!(del_prefix.schedule_at, idx as u64);
            assert_eq!(del_prefix.keyspace_prefix_len, 4);
        }
    }

    #[test]
    fn test_del_prefixes_complementary_ranges() {
        let mut del_prefix = DeletePrefixes::new_with_keyspace_id(1);
        del_prefix = del_prefix.merge_prefix("0000-1".as_bytes());
        del_prefix = del_prefix.merge_prefix("0000-2".as_bytes());
        del_prefix = del_prefix.merge_prefix("0000-33".as_bytes());
        del_prefix = del_prefix.merge_prefix("0000-5".as_bytes());

        let cases: Vec<(
            (&str, &str),      // outer_range
            Vec<(&str, &str)>, // expected
        )> = vec![
            (("0000-0", "0000-1"), vec![("0000-0", "0000-1")]),
            (("0000-0", "0000-2"), vec![("0000-0", "0000-1")]),
            (("0000-0", "0000-3"), vec![("0000-0", "0000-1")]),
            (
                ("0000-0", "0000-33"),
                vec![("0000-0", "0000-1"), ("0000-3", "0000-33")],
            ),
            (
                ("0000-0", "0000-333"),
                vec![("0000-0", "0000-1"), ("0000-3", "0000-33")],
            ),
            (
                ("0000-0", "0000-4"),
                vec![
                    ("0000-0", "0000-1"),
                    ("0000-3", "0000-33"),
                    ("0000-34", "0000-4"),
                ],
            ),
            (
                ("0000-0", "0000-9"),
                vec![
                    ("0000-0", "0000-1"),
                    ("0000-3", "0000-33"),
                    ("0000-34", "0000-5"),
                    ("0000-6", "0000-9"),
                ],
            ),
        ];

        for (i, (outer_range, expected)) in cases.into_iter().enumerate() {
            let ranges = del_prefix
                .get_complementary_ranges(outer_range.0.as_bytes(), outer_range.1.as_bytes());
            let expected = expected
                .into_iter()
                .map(|(start, end)| (start.as_bytes().to_vec(), end.as_bytes().to_vec()))
                .collect::<Vec<_>>();
            assert_eq!(ranges, expected, "case {}: {:?}", i, outer_range);
        }
    }

    #[test]
    fn test_range_keyspace_id() {
        let mut range = ShardRange::new(&[b'x', 0, 0, 1], &[b'x', 0, 0, 2]);
        assert_eq!(range.keyspace_id, 1);
        assert_eq!(range.get_end_keyspace_id(), 1);
        range = ShardRange::new(&[b'x', 0, 0, 1], &[b'x', 0, 0, 2, b'a']);
        assert_eq!(range.get_end_keyspace_id(), 2);
        range = ShardRange::new(&[], GLOBAL_SHARD_END_KEY);
        assert_eq!(range.keyspace_id, 0);
        assert_eq!(range.get_end_keyspace_id(), 0);
        range = ShardRange::new(&[b't', 0, 0, 0], &[]);
        assert_eq!(range.keyspace_id, 0);
    }

    #[test]
    fn test_rewrite_range_prefix() {
        let mut del_prefix = DeletePrefixes::new_with_keyspace_id(1);
        del_prefix = del_prefix.merge_prefix("0000-01".as_bytes());
        del_prefix = del_prefix.merge_prefix("0000-10".as_bytes());
        del_prefix = del_prefix.merge_prefix("0000-21".as_bytes());
        assert!(del_prefix.prefixes.contains(&"0000-01".as_bytes().to_vec()));
        assert!(del_prefix.prefixes.contains(&"0000-10".as_bytes().to_vec()));
        assert!(del_prefix.prefixes.contains(&"0000-21".as_bytes().to_vec()));

        del_prefix.rewrite_range_prefix("1111".as_bytes());
        assert!(del_prefix.prefixes.contains(&"1111-01".as_bytes().to_vec()));
        assert!(del_prefix.prefixes.contains(&"1111-10".as_bytes().to_vec()));
        assert!(del_prefix.prefixes.contains(&"1111-21".as_bytes().to_vec()));
    }

    #[test]
    fn test_properties() {
        let cases: Vec<(
            &'static str, // property_key
            bool,         // need_initial_flush
            bool,         // need_flush
        )> = vec![(TERM_KEY, false, true), (TXN_FILE_REF, true, true)];

        for (property_key, need_initial_flush, need_flush) in cases {
            assert_eq!(
                is_property_need_initial_flush(property_key),
                need_initial_flush
            );
            assert_eq!(is_property_need_flush(property_key), need_flush);
        }
    }
}