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

use std::{
    cmp::max,
    collections::{HashMap, HashSet},
    fmt,
    iter::Iterator,
};

use api_version::{
    ApiV2,
    api_v2::{KEYSPACE_PREFIX_LEN, is_whole_keyspace_range},
};
use bytes::{Buf, Bytes};
use fail::fail_point;
use kvenginepb as pb;
use kvenginepb::{
    SchemaMeta, TxnFileRef, VectorIndex,
    fts::{FtsFieldLayout, FtsTrackedIndexMeta, TableIndexId},
    get_any_snap_from_changeset,
};
use protobuf::Message;
use schema::schema::StorageClassSpec;
use slog_global::*;
use util::TxnFileRefExt as _;

use super::*;
use crate::{
    dfs::FileType,
    table::{
        BoundedDataSet, DataBound, InnerKey, OwnedInnerKey, SnapVersion,
        fts::{TrackedFtsIndexMap, new_legacy_tracked_index_meta},
    },
    table_id::{
        get_table_id_from_data_bound, get_table_id_from_ingest_files, merge_columnar_table_ids,
    },
    util::{TxnFileLocks, TxnFileRefPropertyHelper, is_matched_vector_index, is_same_vector_index},
};

/// An owned LSM layout used for legacy ingest level selection.
///
/// The normal path copies WRITE_CF ranges from one applied [`ShardData`]
/// snapshot so L0 and L1+ cannot describe different compaction states. The
/// timeout fallback copies one [`ShardMeta`] snapshot and conservatively uses
/// all-CF physical ranges for L0 because its metadata does not retain CF
/// ranges.
pub(crate) struct LegacyIngestLevelSnapshot {
    l0_bounds: Vec<(OwnedInnerKey, OwnedInnerKey)>,
    topmost_level: u8,
    topmost_level_bounds: Vec<(OwnedInnerKey, OwnedInnerKey)>,
}

impl LegacyIngestLevelSnapshot {
    pub(crate) fn from_shard_data(shard_data: &ShardData) -> Self {
        let l0_bounds = shard_data
            .l0_tbls
            .iter()
            .filter_map(|l0| l0.get_cf(WRITE_CF).as_ref())
            .map(|write_cf| (write_cf.smallest().into(), write_cf.biggest().into()))
            .collect();
        let (topmost_level, topmost_level_bounds) = shard_data
            .get_cf(WRITE_CF)
            .levels
            .iter()
            .find(|level| !level.tables.is_empty())
            .map(|level| {
                (
                    level.level as u8,
                    level
                        .tables
                        .iter()
                        .map(|table| (table.smallest().into(), table.biggest().into()))
                        .collect(),
                )
            })
            .unwrap_or((WRITE_CF_BOTTOM_LEVEL as u8, Vec::new()));
        Self {
            l0_bounds,
            topmost_level,
            topmost_level_bounds,
        }
    }

    pub(crate) fn from_shard_meta(meta: &ShardMeta) -> Self {
        let l0_bounds = meta
            .files
            .values()
            .filter(|file| file.file_type == FileType::Sst && file.level == 0)
            .map(|file| {
                let bound = file.data_bound();
                (bound.lower_bound.into(), bound.upper_bound.into())
            })
            .collect();
        let topmost_level = meta
            .files
            .values()
            .filter(|file| {
                file.file_type == FileType::Sst && file.level > 0 && file.cf == WRITE_CF as i8
            })
            .map(|file| file.level)
            .min()
            .unwrap_or(WRITE_CF_BOTTOM_LEVEL as u8);
        let topmost_level_bounds = meta
            .files
            .values()
            .filter(|file| {
                file.file_type == FileType::Sst
                    && file.level == topmost_level
                    && file.cf == WRITE_CF as i8
            })
            .map(|file| {
                let bound = file.data_bound();
                (bound.lower_bound.into(), bound.upper_bound.into())
            })
            .collect();
        Self {
            l0_bounds,
            topmost_level,
            topmost_level_bounds,
        }
    }

    /// Finds the WRITE_CF ingest level represented by this single LSM
    /// snapshot.
    pub(crate) fn get_ingest_level(&self, table_bound: DataBound<'_>) -> u32 {
        if Self::overlaps(&self.l0_bounds, table_bound) {
            return 0;
        }
        if Self::overlaps(&self.topmost_level_bounds, table_bound) {
            return self.topmost_level.saturating_sub(1) as u32;
        }
        self.topmost_level as u32
    }

    fn overlaps(bounds: &[(OwnedInnerKey, OwnedInnerKey)], table_bound: DataBound<'_>) -> bool {
        bounds.iter().any(|(smallest, biggest)| {
            table_bound.overlap_bound(DataBound::new(smallest.as_ref(), biggest.as_ref(), true))
        })
    }
}

#[derive(Default, Clone, Debug)]
pub struct ShardMeta {
    pub engine_id: u64,
    pub id: u64,
    pub ver: u64,
    pub range: ShardRange,
    pub inner_key_off: usize,

    /// `seq` is the raft log index of the applied change set.
    ///
    /// Should be updated and ONLY be updated when LSM is changed.
    pub seq: u64,
    pub(crate) files: HashMap<u64, FileMeta>,

    pub(crate) properties: Properties,
    pub base_version: u64,

    /// `data_sequence` is the raft log index of data included in the latest L0
    /// file.
    ///
    /// Note: if meta is not initial flushed, the data persisted log index
    /// should be got from parent. Use `ShardMeta::data_persisted_log_index`.
    pub data_sequence: u64,

    /// `max_ts` is the max ts in all sst files.
    pub max_ts: u64,
    pub parent: Option<Box<ShardMeta>>,
    pub schema: SchemaFileMeta,
    pub columnar_table_ids: Vec<i64>,
    pub unconverted_l0s: Vec<u64>,
    pub vector_indexes: Vec<kvenginepb::VectorIndex>,
    pub columnar_l2_snap_version: SnapVersion,

    // The followings are FTS related fields ============
    pub fts_l0_snap_version: SnapVersion,
    /// Currently tracked FTS indexes. We use this list to determine whether
    /// it is incremental update or index add.
    pub fts_indexes: HashSet<(i64 /* table_id */, i64 /* index_id */)>,
    /// Explicit tracked FTS metadata keyed by `(table_id, index_id)`.
    pub fts_index_metas: TrackedFtsIndexMap,
    /// FTS Level 0 files (order is not guaranteed).
    pub fts_l0_files: Vec<kvenginepb::FtsPackedFileInfo>,
    /// FTS Level 1 packed partitions (order is not guaranteed).
    pub fts_l1_files: Vec<kvenginepb::FtsPackedFileInfo>,
    /// FTS Level 2 dedicated partitions grouped by logical partition.
    pub fts_l2_files: Vec<kvenginepb::FtsDedFileInfo>,
    /// Columnar L0 ids that still need incremental FTS processing after merge.
    pub fts_pending_l0_ids: Vec<u64>,

    /// The following are memory-based field(s).
    ///
    /// Recover process:
    ///
    /// * On restart: recover from shard after kvengine is recovered. See
    ///   `ShardMeta::recover_txn_file_locks_from_kv`.
    ///
    /// * On restore from snapshot: replay raft logs from snapshot index
    ///   (instead of meta.seq). See `PeerStorage::restore_snapshot`.
    ///
    /// * On restore from backup: replay raft logs from applied index (instead
    ///   of preprocessed index). See `BackupCluster::preprocess_shard`.
    ///
    /// Invariant: for every live txn-file lock keyed by `start_ts`, this map
    /// stores the latest applied `lock_val_prefix` so split checks and active
    /// resolve reconstruct current lock state.
    pub(crate) txn_file_locks: TxnFileLocks,
}

impl fmt::Display for ShardMeta {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ShardMeta")
            .field("tag", &self.tag())
            .field("range", &self.range)
            .field("inner_key_off", &self.inner_key_off)
            .field("seq", &self.seq)
            .field("files", &self.files.len())
            .field("properties", &self.properties)
            .field("base_ver", &self.base_version)
            .field("data_seq", &self.data_sequence)
            .field("max_ts", &self.max_ts)
            .field("parent", &self.parent.is_some())
            .field("schema", &self.schema)
            .field("columnar_table_ids", &self.columnar_table_ids.len())
            .field("unconverted_l0s", &self.unconverted_l0s.len())
            .field("vector_indexes", &self.vector_indexes.len())
            .field("columnar_l2_snap_version", &self.columnar_l2_snap_version)
            .field("txn_file_locks", &self.txn_file_locks.len())
            .finish()
    }
}

impl ShardMeta {
    pub fn new(engine_id: u64, cs: &pb::ChangeSet) -> Self {
        assert!(cs.has_snapshot() || cs.has_initial_flush() || cs.has_restore_shard());
        let snap = get_any_snap_from_changeset(cs).unwrap();
        let properties = Properties::new().apply_pb(snap.get_properties());

        let mut meta = Self {
            engine_id,
            id: cs.shard_id,
            ver: cs.shard_ver,
            range: ShardRange::from_snap(snap),
            inner_key_off: snap.inner_key_off as usize,
            seq: cs.sequence,
            properties,
            base_version: snap.base_version,
            data_sequence: snap.data_sequence,
            max_ts: snap.max_ts,
            columnar_table_ids: snap.columnar_table_ids.clone(),
            unconverted_l0s: snap.unconverted_l0s.clone(),
            columnar_l2_snap_version: snap.columnar_l2_snap_version.into(),
            ..Default::default()
        };
        meta.rebuild_txn_file_locks_from_properties();
        for l0 in snap.get_l0_creates() {
            meta.add_file(l0.id, FileMeta::from_l0_table(l0));
        }
        for blob in snap.get_blob_creates() {
            meta.add_file(blob.id, FileMeta::from_blob_table(blob));
        }
        for tbl in snap.get_table_creates() {
            meta.add_file(tbl.id, FileMeta::from_table(tbl));
        }
        for col in snap.get_columnar_creates() {
            meta.add_file(col.get_id(), FileMeta::from_columnar_table(col));
        }
        if cs.has_parent() {
            let parent_meta = Box::new(Self::new(engine_id, cs.get_parent()));
            meta.parent = Some(parent_meta);
        }
        if snap.has_schema_meta() {
            meta.schema = SchemaFileMeta::from_snapshot(snap);
        }
        for vec_idx in snap.get_vector_indexes() {
            meta.vector_indexes.push(vec_idx.clone());
            for vec_idx_file in vec_idx.get_files() {
                meta.add_file(
                    vec_idx_file.id,
                    FileMeta::from_vector_index_file(vec_idx_file),
                );
            }
        }
        for tracked_meta in snap.get_fts_index_metas() {
            meta.insert_fts_index_meta(tracked_meta.clone());
        }
        for tbl_idx_id in snap.get_fts_indexes() {
            if meta
                .fts_indexes
                .contains(&(tbl_idx_id.table_id, tbl_idx_id.index_id))
            {
                continue;
            }
            meta.insert_fts_index_meta(new_legacy_tracked_index_meta(
                tbl_idx_id.table_id,
                tbl_idx_id.index_id,
            ));
        }
        for fts_l0_file in snap.get_fts_l0_files() {
            meta.fts_l0_files.push(fts_l0_file.clone());
        }
        for fts_l1_file in snap.get_fts_l1_files() {
            meta.fts_l1_files.push(fts_l1_file.clone());
        }
        for fts_l2_file in snap.get_fts_l2_files() {
            meta.fts_l2_files.push(fts_l2_file.clone());
        }
        meta.fts_pending_l0_ids = snap.get_fts_pending_l0_ids().to_vec();
        meta.fts_l0_snap_version = SnapVersion::from(snap.get_fts_l0_snap_version());
        meta
    }

    fn insert_fts_index_meta(&mut self, meta: FtsTrackedIndexMeta) {
        let table_id = meta.get_table_id();
        let index_id = meta.get_index_id();
        self.fts_indexes.insert((table_id, index_id));

        let indexes = self.fts_index_metas.entry(table_id).or_default();
        match indexes.get_mut(&index_id) {
            Some(existing) => {
                if existing.get_field_layout() == FtsFieldLayout::FtsFieldLayoutUnspecified
                    || (existing.get_index_fields().is_empty()
                        && !meta.get_index_fields().is_empty())
                {
                    existing.set_field_layout(meta.get_field_layout());
                }
                if existing.get_index_fields().is_empty() && !meta.get_index_fields().is_empty() {
                    existing.set_index_fields(meta.get_index_fields().to_vec().into());
                } else if existing != &meta {
                    debug_assert!(
                        false,
                        "tracked FTS metadata changed in place for ({}, {}): {:?} -> {:?}",
                        table_id, index_id, existing, meta,
                    );
                    *existing = meta;
                }
            }
            None => {
                indexes.insert(index_id, meta);
            }
        }
    }

    fn remove_fts_index(&mut self, table_id: i64, index_id: i64) {
        self.fts_indexes.remove(&(table_id, index_id));
        if let Some(indexes) = self.fts_index_metas.get_mut(&table_id) {
            indexes.remove(&index_id);
            if indexes.is_empty() {
                self.fts_index_metas.remove(&table_id);
            }
        }
    }

    pub fn rebuild_txn_file_locks_from_properties(&mut self) {
        let txn_file_locks =
            TxnFileRefPropertyHelper::from_property(self.properties.get(TXN_FILE_REF))
                .unwrap()
                .into_txn_file_locks(self.data_sequence);
        self.txn_file_locks = txn_file_locks;

        debug!(
            "{} rebuild_txn_file_locks: {}",
            self.tag(),
            self.txn_file_locks
        );
    }

    pub fn recover_txn_file_locks_from_kv(&mut self, kv: &Engine) {
        match kv.get_shard_with_ver(self.id, self.ver) {
            Ok(shard) => {
                self.recover_txn_file_locks_from_shard(&shard);
            }
            Err(err) => {
                warn!("{} recover_txn_file_locks_from_kv: failed to get shard", self.tag(); "err" => ?err);
            }
        }
    }

    pub fn fix_table_meta_offset(&mut self, kv: &Engine) {
        let tag = self.tag();
        let mut to_fix_files: HashMap<&u64, &mut FileMeta> = self
            .files
            .iter_mut()
            .filter(|(_, fm)| {
                // Columnar files must have `table_meta_off` > 0.
                fm.is_sst_file() && fm.can_use_ia() && fm.table_meta_off == 0
            })
            .collect();
        if to_fix_files.is_empty() {
            return;
        }

        info!("{} fix_table_meta_offset", tag; "files" => ?to_fix_files);
        match kv.get_shard_with_ver(self.id, self.ver) {
            Ok(shard) => {
                let data = shard.get_data();
                data.for_each_level(|_cf, lh| {
                    for t in lh.tables.iter() {
                        if let Some(fm) = to_fix_files.remove(&t.id()) {
                            fm.table_meta_off = t.meta_offset();
                        }
                    }
                    to_fix_files.is_empty()
                });
                if !to_fix_files.is_empty() {
                    warn!("{} fix_table_meta_offset: failed to fix table meta offset", tag;
                        "files" => ?to_fix_files);
                }
            }
            Err(err) => {
                warn!("{} fix_table_meta_offset: failed to get shard", tag;
                    "err" => ?err, "files" => ?to_fix_files);
            }
        }
    }

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

    #[inline]
    pub fn initial_flushed(&self) -> bool {
        self.parent.is_none()
    }

    pub fn new_split(
        id: u64,
        ver: u64,
        start: &[u8],
        end: &[u8],
        props: &pb::Properties,
        parent: Box<ShardMeta>,
    ) -> Self {
        let (range, inner_key_off) = if is_whole_keyspace_range(start, end) {
            (ShardRange::new(start, end), KEYSPACE_PREFIX_LEN)
        } else {
            debug!(
                "engine new_split";
                "shard_id" => id,
                "start_key" => format!("{:?}", start),
                "end_key" => format!("{:?}", end),
                "inner_key_off" => parent.inner_key_off
            );
            (ShardRange::new(start, end), parent.inner_key_off)
        };
        Self {
            id,
            ver,
            range,
            inner_key_off,
            parent: Some(parent),
            properties: Properties::new().apply_pb(props),
            ..Default::default()
        }
    }

    fn move_down_file(&mut self, id: u64, cf: i32, level: u32, meta_offset: u32) {
        let tag = self.tag();
        let fm = self
            .files
            .get_mut(&id)
            .unwrap_or_else(|| panic!("{} move_down_file, file {} not found", tag, id));
        assert_eq!(
            fm.get_level() + 1,
            level,
            "{} fm.level {} level {} id: {}",
            tag,
            fm.get_level(),
            level,
            id
        );
        if fm.cf == -1 {
            fm.cf = cf as i8;
        } else {
            assert_eq!(fm.cf, cf as i8);
        }
        fm.level = level as u8;
        fm.l0_size = 0;
        fm.table_meta_off = meta_offset;
    }

    pub fn add_file(&mut self, id: u64, file_meta: FileMeta) {
        self.files.insert(id, file_meta);
    }

    fn delete_file(&mut self, id: u64, level: u32, ft: FileType) {
        if self.has_file_at_level(id, level, ft) {
            self.files.remove(&id);
        }
    }

    fn has_file_at_level(&self, id: u64, level: u32, ft: FileType) -> bool {
        self.file_level(id, ft) == Some(level)
    }

    fn file_level(&self, id: u64, ft: FileType) -> Option<u32> {
        self.files
            .get(&id)
            .filter(|fm| fm.file_type == ft)
            .map(|fm| fm.get_level())
    }

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

    pub fn get_properties_pb(&self) -> kvenginepb::Properties {
        self.properties.clone().to_pb(self.id)
    }

    pub fn set_property(&mut self, key: &str, value: &[u8]) {
        self.properties.set(key, value);
    }

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

    pub fn set_property_bytes(&mut self, key: &str, value: Bytes) {
        self.properties.set_bytes(key, value);
    }

    fn set_property_from_other(&mut self, key: &str, other: &ShardMeta) {
        if let Some(value) = other.get_property(key) {
            self.set_property_bytes(key, value);
        } else {
            self.del_property(key);
        }
    }

    pub fn del_property(&mut self, key: &str) -> Option<Bytes> {
        self.properties.remove(key)
    }

    pub fn apply_change_set(&mut self, cs: &pb::ChangeSet) {
        // Keep this assert. As we will invoke this method directly in some scene.
        debug_assert!(cs.sequence > 0, "{}: invalid cs: {:?}", self.tag(), cs);
        self.seq = cs.sequence;
        if cs.has_initial_flush() {
            self.apply_initial_flush(cs);
            return;
        }
        if cs.has_flush() {
            fail_point!("skip_apply_flush_shard_meta", |_| {});
            self.apply_flush(cs);
            return;
        }
        if cs.has_compaction() {
            self.apply_compaction(cs.get_compaction());
            return;
        }
        if cs.has_destroy_range() {
            self.apply_destroy_range(cs);
            return;
        }
        if cs.has_truncate_ts() {
            self.apply_truncate_ts(cs);
            return;
        }
        if cs.has_trim_over_bound() {
            self.apply_trim_over_bound(cs);
            return;
        }
        if cs.has_ingest_files() {
            self.apply_ingest_files(cs.get_ingest_files());
            return;
        }
        if cs.has_restore_shard() {
            self.apply_restore_shard(cs);
            return;
        }
        if cs.has_major_compaction() {
            self.apply_major_compaction(cs);
            return;
        }
        if cs.has_update_schema_meta() {
            self.schema.update_from_schema_meta(cs);
            return;
        }
        if cs.has_columnar_compaction() {
            self.apply_columnar_compaction(cs.get_columnar_compaction());
            return;
        }
        if cs.has_update_vector_index() {
            self.apply_update_vector_index(cs.get_update_vector_index());
            return;
        }
        if cs.has_fts_update() {
            self.apply_fts_update(cs.get_fts_update());
            return;
        }
        if cs.get_clear_columnar() || cs.has_clear_columnar_with_restore_version() {
            self.clear_columnar(cs);
            return;
        }
        // NOTE: Used for shard meta persist by diff.
        if cs.has_snapshot_diff() {
            self.apply_snapshot_diff(cs);
            return;
        }
        if !cs.get_property_key().is_empty() {
            if cs.get_property_merge() {
                // Now only DEL_PREFIXES_KEY is mergeable.
                assert_eq!(cs.get_property_key(), DEL_PREFIXES_KEY);
                let prefix = cs.get_property_value();
                self.properties.set(
                    DEL_PREFIXES_KEY,
                    &self
                        .properties
                        .get(DEL_PREFIXES_KEY)
                        .map(|b| DeletePrefixes::unmarshal(b.chunk(), self.range.keyspace_id))
                        .unwrap_or_else(|| {
                            DeletePrefixes::new_with_keyspace_id(self.range.keyspace_id)
                        })
                        .merge_prefix(prefix)
                        .marshal(),
                );
            } else if cs.get_property_key() == STORAGE_CLASS_KEY {
                self.apply_update_storage_class(cs);
            } else {
                self.properties
                    .set(cs.get_property_key(), cs.get_property_value());
            }
            return;
        }
        panic!("unexpected change set {:?}", cs)
    }

    fn is_duplicated_compaction(&self, comp: &mut pb::Compaction) -> bool {
        for i in 0..comp.get_top_deletes().len() {
            let id = comp.get_top_deletes()[i];
            if self.is_compaction_file_deleted(id, comp.level, comp) {
                return true;
            }
        }
        for i in 0..comp.get_bottom_deletes().len() {
            let id = comp.get_bottom_deletes()[i];
            if self.is_compaction_file_deleted(id, comp.level + 1, comp) {
                return true;
            }
        }
        false
    }

    pub fn is_duplicated_major_compaction(&self, comp: &mut pb::MajorCompaction) -> bool {
        let ssts_already_deleted =
            comp.get_sstable_change()
                .get_table_deletes()
                .iter()
                .any(|sst_delete| {
                    !self.has_file_at_level(
                        sst_delete.get_id(),
                        sst_delete.get_level(),
                        FileType::Sst,
                    )
                });
        let blobs_already_deleted = comp
            .get_old_blob_tables()
            .iter()
            .any(|blob_tbl_delete| !self.has_file_at_level(*blob_tbl_delete, 0, FileType::Blob));
        if ssts_already_deleted || blobs_already_deleted {
            info!("{} skip duplicated major compaction {:?}", self.tag(), comp);
            // set compaction to be conflicted, so that newly created files will be GCed.
            comp.conflicted = true;
            return true;
        }
        false
    }

    pub fn is_duplicated_columnar_compaction(&self, comp: &pb::ColumnarCompaction) -> bool {
        match comp.get_target_level() {
            0 => {
                // L0 to columnar compaction.
                let unconverted_l0s = self.unconverted_l0s.iter().collect::<HashSet<_>>();
                if comp
                    .get_row_l0s()
                    .iter()
                    .any(|l0| !unconverted_l0s.contains(l0))
                {
                    return true;
                }
            }
            1 | 2 => {
                // L0/L1 or major columnar compaction.
                let col_already_deleted = comp
                    .get_columnar_change()
                    .get_columnar_deletes()
                    .iter()
                    .any(|delete| {
                        !self.has_file_at_level(
                            delete.get_id(),
                            delete.get_level(),
                            FileType::Columnar,
                        )
                    });
                if col_already_deleted {
                    return true;
                }
            }
            _ => {}
        }
        false
    }

    pub fn is_duplicated_change_set(&self, cs: &mut pb::ChangeSet) -> bool {
        // Keep this assert. As we will invoke this method directly in some scene.
        debug_assert!(cs.sequence > 0, "{}: invalid cs: {:?}", self.tag(), cs);
        if self.seq >= cs.sequence {
            info!(
                "{} skip duplicated change {:?}, meta_seq {}",
                self.tag(),
                cs,
                self.seq
            );
            return true;
        }
        if cs.has_initial_flush() && self.initial_flushed() {
            info!(
                "{} skip duplicated initial flush, already flushed",
                self.tag()
            );
            return true;
        }
        if cs.has_flush() {
            let flush = cs.get_flush();
            if flush.get_version() <= self.data_version() {
                info!(
                    "{} skip duplicated flush {:?} for old version",
                    self.tag(),
                    cs
                );
                return true;
            }
        }
        if cs.has_compaction() {
            let comp = cs.mut_compaction();
            if self.is_duplicated_compaction(comp) {
                return true;
            }
        }
        if cs.has_major_compaction() {
            let comp = cs.mut_major_compaction();
            if self.is_duplicated_major_compaction(comp) {
                return true;
            }
        }
        if cs.has_columnar_compaction()
            && self.is_duplicated_columnar_compaction(cs.get_columnar_compaction())
        {
            info!(
                "{} skip duplicated columnar compaction {:?}",
                self.tag(),
                cs
            );
            return true;
        }
        if cs.has_destroy_range() && self.is_duplicated_table_change(cs.get_destroy_range()) {
            info!("{} skip duplicated destroy range {:?}", self.tag(), cs);
            return true;
        }
        if cs.has_truncate_ts() && self.is_duplicated_table_change(cs.get_truncate_ts()) {
            info!("{} skip duplicated truncate ts {:?}", self.tag(), cs);
            return true;
        }
        if cs.has_trim_over_bound() && self.is_duplicated_table_change(cs.get_trim_over_bound()) {
            info!("{} skip duplicated trim_over_bound {:?}", self.tag(), cs);
            return true;
        }
        if cs.has_ingest_files() {
            let ingest_files = cs.mut_ingest_files();
            if let Some(ingest_id) = is_legacy_ingest(ingest_files) {
                // For legacy BR.
                if let Some(old_ingest_id) = self.get_property(INGEST_ID_KEY) {
                    if ingest_id.eq(&old_ingest_id) {
                        info!(
                            "{} skip duplicated ingest files, ingest_id:{:?}",
                            self.tag(),
                            ingest_id,
                        );
                        return true;
                    }
                }
            } else {
                // For load data.
                let is_empty = self.dedup_ingest_files_of_load_data(ingest_files);
                if is_empty {
                    info!("{} skip duplicated ingest files", self.tag(),);
                    return true;
                }
            }
        }
        if cs.has_restore_shard() {
            // TODO: skip duplicated
        }
        if cs.has_update_schema_meta() {
            let schema_version = cs.get_update_schema_meta().get_version();
            let cur_ver = self.schema.file_ver();
            let valid = self.schema.is_valid();
            let is_duplicated = if valid {
                cur_ver >= schema_version
            } else {
                cur_ver > schema_version
            };
            if is_duplicated {
                info!(
                    "{} skip duplicated update schema meta version {}, cur_ver: {}, valid: {}",
                    self.tag(),
                    schema_version,
                    cur_ver,
                    valid
                );
                return true;
            }
        }
        if cs.has_fts_update() {
            let update = cs.get_fts_update();
            if self.is_duplicated_fts_update(update) {
                info!(
                    "{} skip duplicated fts update {:?}, fts_l0_snap_version: {:?}",
                    self.tag(),
                    update,
                    self.fts_l0_snap_version
                );
                return true;
            }
        }
        if cs.get_property_key() == STORAGE_CLASS_KEY {
            let sc_spec = StorageClassSpec::unmarshal(Some(&cs.property_value));
            return self.get_storage_class_spec() == sc_spec;
        }
        false
    }

    fn is_duplicated_fts_update(&self, update: &pb::FtsUpdate) -> bool {
        if update.get_set_snap_version() > 0
            && update.get_set_snap_version() < self.fts_l0_snap_version.into_inner()
        {
            return true;
        }
        if update
            .get_l0_remove_files()
            .iter()
            .any(|id| !self.fts_l0_files.iter().any(|file| file.get_id() == *id))
            || update
                .get_l1_remove_files()
                .iter()
                .any(|id| !self.fts_l1_files.iter().any(|file| file.get_id() == *id))
            || update
                .get_l2_remove_files()
                .iter()
                .any(|id| !self.fts_l2_files.iter().any(|file| file.get_id() == *id))
        {
            return true;
        }
        false
    }

    fn is_compaction_file_deleted(&self, id: u64, level: u32, comp: &mut pb::Compaction) -> bool {
        if !self.has_file_at_level(id, level, FileType::Sst) {
            info!(
                "{} skip duplicated compaction file {} at level {}, already deleted.",
                self.tag(),
                id,
                level,
            );
            comp.conflicted = true;
            return true;
        }
        false
    }

    // We assume that the shard is empty before load data.
    // Duplication happens when client side retry, or the shard is merged from two
    // shards and one of which has not been ingested.
    fn dedup_ingest_files_of_load_data(&self, ingest_files: &mut pb::IngestFiles) -> bool /* is_empty */
    {
        let l0_files = ingest_files
            .take_l0_creates()
            .into_iter()
            .filter(|file| !self.all_files().contains_key(&file.id))
            .collect::<Vec<_>>();
        ingest_files.set_l0_creates(l0_files.into());

        let ln_files = ingest_files
            .take_table_creates()
            .into_iter()
            .filter(|file| !self.all_files().contains_key(&file.id))
            .collect::<Vec<_>>()
            .into();
        ingest_files.set_table_creates(ln_files);

        let blob_files = ingest_files
            .take_blob_creates()
            .into_iter()
            .filter(|file| !self.all_files().contains_key(&file.id))
            .collect::<Vec<_>>()
            .into();
        ingest_files.set_blob_creates(blob_files);

        ingest_files.get_l0_creates().is_empty()
            && ingest_files.get_table_creates().is_empty()
            && ingest_files.get_blob_creates().is_empty()
    }

    /// Returns an existing and incoming file ID when an ingest would create
    /// overlapping WRITE_CF files at the same L1+ level.
    pub fn check_overlap_for_ingest(
        &self,
        ingest_files: &pb::IngestFiles,
    ) -> Option<(
        u64, // existed file id
        u64, // ingest file id
    )> {
        let ingest_tables = ingest_files.get_table_creates();
        for ingest_file in ingest_tables {
            for (existed_file_id, existed_file) in self.files.iter().filter(|(_, file)| {
                file.get_cf() == WRITE_CF as i32
                    && file.get_level() == ingest_file.get_level()
                    && file.is_sst_file()
            }) {
                if existed_file
                    .data_bound()
                    .overlap_bound(ingest_file.data_bound())
                {
                    return Some((*existed_file_id, ingest_file.id));
                }
            }
        }
        None
    }

    // Check if any compaction output files would overlap with existing files.
    // Handles both major compaction and non-major compaction (L0/L1Plus).
    //
    // Returns Some((existed_file_id, output_file_id)) if overlap detected,
    // None otherwise.
    pub fn check_overlap_for_compaction(
        &self,
        cs: &pb::ChangeSet,
    ) -> Option<(
        u64, // existed file id
        u64, // output file id
    )> {
        // Helper closure to check if output files overlap with existing files
        let check_overlap = |output_files: &[pb::TableCreate],
                             files_to_delete: &std::collections::HashSet<u64>|
         -> Option<(u64, u64)> {
            for output_file in output_files {
                for (existed_file_id, existed_file) in self.files.iter().filter(|(id, file)| {
                    // Skip files being deleted by this compaction
                    !files_to_delete.contains(id)
                        && file.get_cf() == output_file.get_cf()
                        && file.get_level() == output_file.get_level()
                        && file.is_sst_file()
                }) {
                    if existed_file
                        .data_bound()
                        .overlap_bound(output_file.data_bound())
                    {
                        return Some((*existed_file_id, output_file.id));
                    }
                }
            }
            None
        };

        if cs.has_major_compaction() {
            let major_compaction = cs.get_major_compaction();

            // Empty MajorCompaction (no sstable_change) can happen when manual compaction
            // is triggered but there are no tables to compact. In this case, the compaction
            // is still sent through raft to remove the MANUAL_MAJOR_COMPACTION property.
            // No overlap to check.
            let sstable_change = major_compaction.sstable_change.as_ref()?;

            // Filter out output files that replace input files (normal overlap).
            // Only check for overlap with OTHER existing files (e.g., newly ingested).
            let files_to_delete: std::collections::HashSet<u64> = sstable_change
                .get_table_deletes()
                .iter()
                .map(|del| del.get_id())
                .collect();

            check_overlap(sstable_change.get_table_creates(), &files_to_delete)
        } else if cs.has_compaction() {
            let compaction = cs.get_compaction();

            // Only need to filter bottom_deletes (files at output level).
            // top_deletes are at upper level and won't overlap with output files.
            let files_to_delete: std::collections::HashSet<u64> =
                compaction.get_bottom_deletes().iter().copied().collect();

            check_overlap(compaction.get_table_creates(), &files_to_delete)
        } else {
            None
        }
    }

    fn apply_flush(&mut self, cs: &pb::ChangeSet) {
        let flush = cs.get_flush();
        self.apply_properties(flush.get_properties());
        let mut new_l0s = vec![];
        if flush.has_l0_create() {
            let l0 = flush.get_l0_create();
            self.add_file(l0.id, FileMeta::from_l0_table(l0));
            new_l0s.push(l0.id);
        }
        for l0 in flush.get_l0_creates() {
            self.add_file(l0.id, FileMeta::from_l0_table(l0));
            new_l0s.push(l0.id);
        }
        if !self.columnar_table_ids.is_empty() {
            // Persist the same invariant as live ShardData: a temporary schema-file gap
            // must not drop pending row L0s while columnar state is still active.
            self.unconverted_l0s.extend_from_slice(&new_l0s);
        }
        let new_data_seq = flush.get_version() - self.base_version;
        if self.data_sequence < new_data_seq {
            debug!(
                "{} apply flush update data sequence from {} to {}, flush ver {} base {}",
                self.tag(),
                self.data_sequence,
                new_data_seq,
                flush.get_version(),
                self.base_version
            );
            self.data_sequence = new_data_seq;
        }
        self.max_ts = std::cmp::max(self.max_ts, flush.max_ts);
    }

    pub fn apply_initial_flush(&mut self, cs: &pb::ChangeSet) {
        let old_props = self.properties.clone();
        let mut new_meta = Self::new(self.engine_id, cs);
        new_meta.range = self.range.clone();
        new_meta.properties.complement_merge(old_props);
        new_meta.txn_file_locks = self.txn_file_locks.clone();
        // self.data_sequence may be advanced on raft log gc tick.
        new_meta.data_sequence = std::cmp::max(new_meta.data_sequence, self.data_sequence);
        new_meta.max_ts = std::cmp::max(new_meta.max_ts, self.max_ts);

        info!("{} apply_initial_flush", self.tag();
            "prop" => ?new_meta.properties,
            "data_seq" => new_meta.data_sequence,
            "columnar_table_ids" => ?new_meta.columnar_table_ids,
            "max_ts" => new_meta.max_ts);
        *self = new_meta;
    }

    fn apply_compaction(&mut self, comp: &pb::Compaction) {
        if is_move_down(comp) {
            for tbl in comp.get_table_creates() {
                self.move_down_file(tbl.id, tbl.cf, tbl.level, tbl.meta_offset);
            }
            return;
        }
        for id in comp.get_top_deletes() {
            self.delete_file(*id, comp.level, FileType::Sst);
        }
        for id in comp.get_bottom_deletes() {
            self.delete_file(*id, comp.level + 1, FileType::Sst);
        }
        for tbl in comp.get_table_creates() {
            self.add_file(tbl.get_id(), FileMeta::from_table(tbl));
        }
        for blob_table in comp.get_blob_tables() {
            self.add_file(blob_table.get_id(), FileMeta::from_blob_table(blob_table));
        }
    }

    fn apply_major_compaction(&mut self, cs: &pb::ChangeSet) {
        let comp = cs.get_major_compaction();
        for delete in comp.get_sstable_change().get_table_deletes() {
            self.delete_file(delete.get_id(), delete.get_level(), FileType::Sst);
        }
        for create in comp.get_sstable_change().get_table_creates() {
            self.add_file(create.id, FileMeta::from_table(create));
        }
        for delete in comp.get_old_blob_tables() {
            self.delete_file(*delete, 0, FileType::Blob);
        }
        for create in comp.get_new_blob_tables() {
            self.add_file(create.get_id(), FileMeta::from_blob_table(create));
        }
        self.del_property(MANUAL_MAJOR_COMPACTION);
    }

    fn is_duplicated_table_change(&self, tc: &pb::TableChange) -> bool {
        tc.get_table_deletes().iter().any(|deleted| {
            !self.has_file_at_level(deleted.get_id(), deleted.get_level(), FileType::Sst)
        }) || tc.get_columnar_deletes().iter().any(|deleted| {
            !self.has_file_at_level(deleted.get_id(), deleted.get_level(), FileType::Columnar)
        })
    }

    pub fn is_empty_table_change(tc: &pb::TableChange) -> bool {
        tc.get_table_deletes().is_empty()
            && tc.get_table_creates().is_empty()
            && tc.get_columnar_deletes().is_empty()
            && tc.get_columnar_creates().is_empty()
    }

    fn apply_table_change(&mut self, tc: &pb::TableChange) {
        for deleted in tc.get_table_deletes() {
            self.delete_file(deleted.get_id(), deleted.get_level(), FileType::Sst);
        }
        for created in tc.get_table_creates() {
            self.add_file(created.get_id(), FileMeta::from_table(created));
        }
        for deleted in tc.get_columnar_deletes() {
            self.delete_file(deleted.get_id(), deleted.get_level(), FileType::Columnar);
        }
        for created in tc.get_columnar_creates() {
            self.add_file(created.get_id(), FileMeta::from_columnar_table(created));
        }
    }

    fn apply_table_change_to_unconverted_l0s(&mut self, tc: &pb::TableChange) {
        if self.unconverted_l0s.is_empty() {
            return;
        }
        let mut unconverted_l0s: HashSet<u64> = self.unconverted_l0s.iter().cloned().collect();
        // `file_ids_map` is sst table file ids that each deleted file id followed by a
        // created one. If there is no created file for a deleted file, it will not be
        // contained in the `file_ids_map`.
        let mut iter = tc.get_file_ids_map().iter();
        while let (Some(delete_id), Some(create_id)) = (iter.next(), iter.next()) {
            // Only add the corresponding created file if the deleted file is in the
            // `unconverted_l0s`.
            if unconverted_l0s.contains(delete_id) {
                unconverted_l0s.insert(*create_id);
            }
        }
        for deleted in tc.get_table_deletes() {
            unconverted_l0s.remove(&deleted.get_id());
        }

        self.unconverted_l0s = unconverted_l0s.into_iter().collect();
    }

    fn apply_destroy_range(&mut self, cs: &pb::ChangeSet) {
        assert!(cs.has_destroy_range());
        self.apply_table_change(cs.get_destroy_range());
        self.apply_table_change_to_unconverted_l0s(cs.get_destroy_range());
        if cs.has_fts_update() {
            self.apply_fts_update(cs.get_fts_update());
        }
        // ChangeSet of DestroyRange contains the corresponding delete-prefixes which
        // should be cleaned up.
        assert_eq!(cs.get_property_key(), DEL_PREFIXES_KEY);
        if let Some(data) = self.properties.get(DEL_PREFIXES_KEY) {
            let old = DeletePrefixes::unmarshal(data.chunk(), self.range.keyspace_id);
            let done = DeletePrefixes::unmarshal(cs.get_property_value(), self.range.keyspace_id);
            let new = old.split(&done);
            self.properties.set(DEL_PREFIXES_KEY, &new.marshal());
            info!(
                "Destroy range in meta changed from {:?} to {:?} (split done: {:?})",
                old, new, done
            );
        }
    }

    fn apply_truncate_ts(&mut self, cs: &pb::ChangeSet) {
        debug!("apply truncate ts in meta {:?}", self.id);
        assert!(cs.has_truncate_ts());
        self.apply_table_change(cs.get_truncate_ts());
        // During keyspace restoration, truncate_ts will be triggered. We also need
        // manipulate the unconverted_l0s.
        self.apply_table_change_to_unconverted_l0s(cs.get_truncate_ts());
        if cs.has_fts_update() {
            self.apply_fts_update(cs.get_fts_update());
        }
    }

    fn apply_trim_over_bound(&mut self, cs: &pb::ChangeSet) {
        debug!("apply changeset.trim_over_bound in meta {:?}", self.id);
        assert!(cs.has_trim_over_bound());
        self.apply_table_change(cs.get_trim_over_bound());
        self.apply_table_change_to_unconverted_l0s(cs.get_trim_over_bound());
        if cs.has_fts_update() {
            self.apply_fts_update(cs.get_fts_update());
        }
        self.set_property(TRIM_OVER_BOUND, TRIM_OVER_BOUND_DISABLE);
    }

    fn apply_ingest_files(&mut self, ingest_files: &pb::IngestFiles) {
        self.max_ts = std::cmp::max(self.max_ts, ingest_files.max_ts);
        self.apply_properties(ingest_files.get_properties());
        for tbl in ingest_files.get_table_creates() {
            self.add_file(tbl.id, FileMeta::from_table(tbl));
        }
        for l0_tbl in ingest_files.get_l0_creates() {
            self.add_file(l0_tbl.id, FileMeta::from_l0_table(l0_tbl));
        }
        for blob_tbl in ingest_files.get_blob_creates() {
            self.add_file(blob_tbl.id, FileMeta::from_blob_table(blob_tbl));
        }
        if let Some(table_id) = get_table_id_from_ingest_files(ingest_files) {
            self.columnar_table_ids.retain(|id| *id != table_id);
        }
    }

    fn apply_properties(&mut self, props: &pb::Properties) {
        for i in 0..props.get_keys().len() {
            let key = &props.get_keys()[i];
            let val = &props.get_values()[i];
            self.properties.set(key, val.as_slice());
        }
    }

    fn apply_restore_shard(&mut self, cs: &pb::ChangeSet) {
        assert!(cs.has_restore_shard());
        assert_eq!(self.range.outer_start, cs.get_restore_shard().outer_start);
        assert_eq!(self.range.outer_end, cs.get_restore_shard().outer_end);
        info!(
            "{} apply_restore_shard in meta", self.tag();
            "current ver" => self.ver,
            "seq" => self.seq,
            "base_ver" => self.base_version,
            "data_seq" => self.data_sequence,
            "inner_key_off" => self.inner_key_off,
        );

        // Increase shard version to make change sets generated before restore shard
        // stale.
        let mut new_meta = Self::new(self.engine_id, cs);
        new_meta.data_sequence = cs.sequence;
        new_meta.ver = cs.shard_ver + 1;
        *self = new_meta;
        info!(
            "{} apply_restore_shard in meta", self.tag();
            "new ver" => self.ver,
            "seq" => self.seq,
            "base_ver" => self.base_version,
            "data_seq" => self.data_sequence,
            "properties" => ?self.properties.to_pb(self.id),
            "inner_key_off" => self.inner_key_off,
        );
    }

    pub fn apply_split(
        &self,
        split: &kvenginepb::Split,
        sequence: u64,
        initial_seq: u64,
    ) -> Vec<ShardMeta> {
        let old = self;
        let old_storage_class = old.get_property(STORAGE_CLASS_KEY);
        let new_shards_len = split.get_new_shards().len();
        let mut new_shards = Vec::with_capacity(new_shards_len);
        let new_ver = old.ver + new_shards_len as u64 - 1;

        let mut parent = old.clone();
        // We only need lock CF files to recover the mem-table.
        // remove other files to reduce memory usage and rfengine manifest file size.
        parent.files.retain(|_, fm| {
            (fm.file_type == FileType::Sst) && (fm.level == 0 || fm.cf == LOCK_CF as i8)
        });
        parent.vector_indexes.clear();

        for i in 0..new_shards_len {
            let (start_key, end_key) = get_splitting_start_end(
                &old.range.outer_start,
                &old.range.outer_end,
                split.get_keys(),
                i,
            );
            let new_shard = &split.get_new_shards()[i];
            let id = new_shard.get_shard_id();
            let mut meta = ShardMeta::new_split(
                id,
                new_ver,
                start_key,
                end_key,
                new_shard,
                Box::new(parent.clone()),
            );
            meta.engine_id = self.engine_id;
            if id == old.id {
                meta.base_version = old.base_version;
                meta.data_sequence = sequence;
                meta.seq = sequence;
            } else {
                debug!(
                    "new base for {}, {} {}",
                    meta.tag(),
                    old.base_version,
                    sequence
                );
                meta.base_version = old.base_version + sequence;
                meta.data_sequence = initial_seq;
                meta.seq = initial_seq;
            }
            if is_whole_keyspace_range(&meta.range.outer_start, &meta.range.outer_end) {
                new_shards.push(meta);
                continue;
            }
            let (min_table_id, max_table_id) = get_table_id_from_data_bound(meta.data_bound());
            let columnar_table_ids: Vec<_> = old
                .columnar_table_ids
                .iter()
                .filter(|&&table_id| table_id >= min_table_id && table_id <= max_table_id)
                .copied()
                .collect();
            meta.columnar_table_ids = columnar_table_ids;
            meta.columnar_l2_snap_version = old.columnar_l2_snap_version;
            // Although `max_ts` will be updated in initial flush again, still set here to
            // avoid issue in unexpected corner case.
            meta.max_ts = self.max_ts;
            meta.schema = self.schema.clone();
            meta.set_property_opt(STORAGE_CLASS_KEY, old_storage_class.as_deref());
            new_shards.push(meta);
        }
        for new_shard in &mut new_shards {
            if is_whole_keyspace_range(&new_shard.range.outer_start, &new_shard.range.outer_end) {
                continue;
            }
            let has_columnar_tables = !new_shard.columnar_table_ids.is_empty();
            let new_shard_bound = new_shard.range.data_bound();
            for (fid, fm) in &old.files {
                if !new_shard_bound.overlap_bound(fm.data_bound()) {
                    continue;
                }

                if has_columnar_tables && fm.get_level() == 0 && old.unconverted_l0s.contains(fid) {
                    new_shard.unconverted_l0s.push(*fid);
                }

                if fm.is_columnar_file() && !has_columnar_tables {
                    continue;
                }

                new_shard.files.insert(*fid, fm.clone());
            }
            for vec_idx in &old.vector_indexes {
                if !new_shard.columnar_table_ids.contains(&vec_idx.table_id) {
                    continue;
                }
                let mut files = vec![];
                for vec_file in vec_idx.get_files() {
                    if new_shard_bound.overlap_bound(vec_file.data_bound()) {
                        files.push(vec_file.clone());
                    }
                }
                let mut new_vec_idx = vec_idx.clone();
                new_vec_idx.set_files(files.into());
                new_shard.vector_indexes.push(new_vec_idx);
            }
            Self::apply_fts_split_to_new_shard(new_shard, old);
        }
        new_shards
    }

    fn apply_fts_split_to_new_shard(new_shard: &mut ShardMeta, old_shard: &ShardMeta) {
        if new_shard.columnar_table_ids.is_empty() {
            return;
        }
        new_shard.fts_l0_snap_version = old_shard.fts_l0_snap_version;
        for &(table_id, index_id) in &old_shard.fts_indexes {
            if new_shard.columnar_table_ids.contains(&table_id) {
                let meta = old_shard
                    .fts_index_metas
                    .get(&table_id)
                    .and_then(|indexes| indexes.get(&index_id))
                    .cloned()
                    .unwrap_or_else(|| new_legacy_tracked_index_meta(table_id, index_id));
                new_shard.insert_fts_index_meta(meta);
            }
        }
        let new_shard_bound = new_shard.range.data_bound();
        new_shard.fts_l0_files.extend(
            old_shard
                .fts_l0_files
                .iter()
                .filter(|file| {
                    let file_bound = DataBound::new(
                        InnerKey::from_inner_buf(&file.smallest),
                        InnerKey::from_inner_buf(&file.biggest),
                        true,
                    );
                    new_shard_bound.overlap_bound(file_bound)
                })
                .cloned(),
        );
        new_shard.fts_l1_files.extend(
            old_shard
                .fts_l1_files
                .iter()
                .filter(|file| {
                    let file_bound = DataBound::new(
                        InnerKey::from_inner_buf(&file.smallest),
                        InnerKey::from_inner_buf(&file.biggest),
                        true,
                    );
                    new_shard_bound.overlap_bound(file_bound)
                })
                .cloned(),
        );
        new_shard.fts_l2_files.extend(
            old_shard
                .fts_l2_files
                .iter()
                .filter(|file| {
                    let file_bound = DataBound::new(
                        InnerKey::from_inner_buf(&file.smallest),
                        InnerKey::from_inner_buf(&file.biggest),
                        true,
                    );
                    new_shard_bound.overlap_bound(file_bound)
                })
                .cloned(),
        );
        let pending_l0_ids: Vec<u64> = old_shard
            .fts_pending_l0_ids
            .iter()
            .filter(|id| {
                old_shard
                    .files
                    .get(id)
                    .map(|fm| new_shard.data_bound().overlap_bound(fm.data_bound()))
                    .unwrap_or(false)
            })
            .copied()
            .collect();
        new_shard.fts_pending_l0_ids.extend(pending_l0_ids);
        new_shard.fts_pending_l0_ids.sort_unstable();
        new_shard.fts_pending_l0_ids.dedup();
    }

    pub fn apply_columnar_compaction(&mut self, comp: &pb::ColumnarCompaction) {
        if comp.target_level == 2 {
            self.columnar_l2_snap_version = comp.snap_version.into();
        }

        let col_change = comp.get_columnar_change();
        for col_create in col_change.get_columnar_creates() {
            self.files.insert(
                col_create.get_id(),
                FileMeta::from_columnar_table(col_create),
            );
        }
        for col_delete in col_change.get_columnar_deletes() {
            self.files.remove(&col_delete.get_id());
        }
        self.unconverted_l0s
            .retain(|unconverted_l0| !comp.row_l0s.contains(unconverted_l0));
        if self.columnar_table_ids.is_empty() {
            let new_flushed_l0s: Vec<u64> = self
                .files
                .iter()
                .filter(|(id, fm)| {
                    fm.get_level() == 0 && fm.is_sst_file() && !comp.row_l0s.contains(id)
                })
                .map(|(id, _)| *id)
                .collect();
            self.unconverted_l0s.extend(new_flushed_l0s);
        }
        if comp.snap_version == 0 {
            self.columnar_table_ids.clear();
            self.unconverted_l0s.clear();
            self.vector_indexes.clear();
            self.clear_fts();
            self.schema.clear(self.schema.restore_ver());
        } else {
            self.columnar_table_ids
                .extend_from_slice(comp.get_columnar_table_ids());
            self.columnar_table_ids.sort_unstable();
            self.columnar_table_ids.dedup();
            self.columnar_table_ids
                .retain(|id| !comp.get_columnar_table_ids_to_clear().contains(id));
            let columnar_table_ids = self.columnar_table_ids.clone();
            self.vector_indexes
                .retain(|vec_idx| columnar_table_ids.contains(&vec_idx.table_id));
            // If columnar_table_ids is empty, this must be a major compaction to clear the
            // last table. Make sure all unconverted l0s are cleared.
            if columnar_table_ids.is_empty() {
                self.unconverted_l0s.clear();
            }

            self.retain_fts(&columnar_table_ids);
        }
        if comp.get_is_manual_major_compaction() {
            self.del_property(MANUAL_MAJOR_COMPACTION);
        }
    }

    pub fn apply_update_vector_index(&mut self, update_vec_idx: &pb::UpdateVectorIndex) {
        for &id in &update_vec_idx.removed {
            self.files.remove(&id);
        }
        for file in update_vec_idx.get_added() {
            self.add_file(file.id, FileMeta::from_vector_index_file(file));
        }

        let is_delete_marker = update_vec_idx.snap_version == 0;
        if let Some(pos) = self
            .vector_indexes
            .iter()
            .position(|vec_idx| is_matched_vector_index(vec_idx, update_vec_idx))
        {
            let mut old_idx = self.vector_indexes.remove(pos);
            if is_delete_marker {
                return;
            }
            let mut files = old_idx.take_files().into_vec();
            files.retain(|f| !update_vec_idx.removed.contains(&f.id));
            files.extend_from_slice(update_vec_idx.get_added());
            old_idx.set_files(files.into());
            old_idx.set_snap_version(update_vec_idx.snap_version);
            self.vector_indexes.insert(pos, old_idx);
            return;
        }

        if is_delete_marker {
            return;
        }

        let mut vec_idx = kvenginepb::VectorIndex::new();
        vec_idx.table_id = update_vec_idx.table_id;
        vec_idx.index_id = update_vec_idx.index_id;
        vec_idx.col_id = update_vec_idx.col_id;
        vec_idx.set_snap_version(update_vec_idx.snap_version);
        for file in update_vec_idx.get_added() {
            vec_idx.mut_files().push(file.clone());
        }
        self.vector_indexes.push(vec_idx)
    }

    /// Applies a combined FTS update to the current ShardMeta.
    pub fn apply_fts_update(&mut self, update: &pb::FtsUpdate) {
        if update.get_clear_all() {
            self.fts_indexes.clear();
            self.fts_index_metas.clear();
            self.fts_l0_files.clear();
            self.fts_l1_files.clear();
            self.fts_l2_files.clear();
            self.fts_l0_snap_version = SnapVersion::zero();
            self.fts_pending_l0_ids.clear();
            return;
        }
        for tbl_idx_id in update.get_remove_tracked_indexes() {
            self.remove_fts_index(tbl_idx_id.table_id, tbl_idx_id.index_id);
        }
        for meta in update.get_add_tracked_index_metas() {
            self.insert_fts_index_meta(meta.clone());
        }
        for tbl_idx_id in update.get_add_tracked_indexes() {
            if !self
                .fts_indexes
                .contains(&(tbl_idx_id.table_id, tbl_idx_id.index_id))
            {
                self.insert_fts_index_meta(new_legacy_tracked_index_meta(
                    tbl_idx_id.table_id,
                    tbl_idx_id.index_id,
                ));
            }
        }
        if self.fts_indexes.is_empty() {
            self.fts_pending_l0_ids.clear();
        }
        for fts_l0_file in update.get_l0_add_files() {
            self.fts_l0_files.push(fts_l0_file.clone());
        }
        for file in update.get_l1_add_files() {
            self.fts_l1_files.push(file.clone());
        }
        for file in update.get_l2_add_files() {
            self.fts_l2_files.push(file.clone());
        }
        if !update.get_l0_remove_files().is_empty() {
            let remove: HashSet<u64> = update.get_l0_remove_files().iter().copied().collect();
            self.fts_l0_files
                .retain(|file| !remove.contains(&file.get_id()));
        }
        if !update.get_l1_remove_files().is_empty() {
            let remove: HashSet<u64> = update.get_l1_remove_files().iter().copied().collect();
            self.fts_l1_files
                .retain(|file| !remove.contains(&file.get_id()));
        }
        if !update.get_l2_remove_files().is_empty() {
            let remove: HashSet<u64> = update.get_l2_remove_files().iter().copied().collect();
            self.fts_l2_files
                .retain(|file| !remove.contains(&file.get_id()));
        }
        if update.get_set_snap_version() > 0 {
            self.fts_l0_snap_version = SnapVersion::from(update.get_set_snap_version());
        }
        if !update.get_pending_columnar_l0_remove_files().is_empty()
            && !self.fts_pending_l0_ids.is_empty()
        {
            self.fts_pending_l0_ids
                .retain(|id| !update.get_pending_columnar_l0_remove_files().contains(id));
        }
    }

    pub fn apply_snapshot_diff(&mut self, cs: &pb::ChangeSet) {
        self.ver = cs.get_shard_ver();
        let snap = cs.get_snapshot_diff();
        if snap.has_properties() {
            let properties = std::mem::take(&mut self.properties);
            self.properties = properties.apply_pb(snap.get_properties());
        }
        if snap.get_data_sequence() > 0 {
            self.data_sequence = snap.get_data_sequence();
        }
        if snap.has_schema_meta() {
            let schema_meta = snap.get_schema_meta();
            self.schema.schema_file_id = schema_meta.get_file_id();
            self.schema.schema_file_ver = schema_meta.get_version();
            self.schema.schema_restore_ver = schema_meta.get_restore_version();
        }
    }

    pub fn clear_columnar(&mut self, cs: &pb::ChangeSet) {
        let restore_version = if cs.has_clear_columnar_with_restore_version() {
            cs.get_clear_columnar_with_restore_version()
                .get_restore_version()
        } else {
            self.schema.restore_ver()
        };
        info!(
            "{} shard_meta apply clear_columnar, restore_version: {:?}",
            self.tag(),
            restore_version
        );
        self.schema.clear(restore_version);
        self.columnar_table_ids.clear();
        self.columnar_l2_snap_version = SnapVersion::zero();
        self.unconverted_l0s.clear();
        self.vector_indexes.clear();
        self.clear_fts();
        self.files.retain(|_, fm| {
            fm.file_type != FileType::Columnar
                && fm.file_type != FileType::Schema
                && fm.file_type != FileType::VectorIndex
                && fm.file_type != FileType::FtsPackedFile
                && fm.file_type != FileType::FtsDedicatedFile
        });
    }

    pub fn new_snapshot_diff_pb(&self) -> pb::ChangeSet {
        let mut cs = new_change_set(self.id, self.ver);
        cs.set_sequence(self.seq);
        let snap = cs.mut_snapshot_diff();
        let props = snap.mut_properties();
        props.shard_id = self.id;
        if let Some(term) = self.get_property(TERM_KEY) {
            props.keys.push(TERM_KEY.to_string());
            props.values.push(term.to_vec());
        }
        snap.set_data_sequence(self.data_sequence);
        if self.schema.has_value() {
            let schema_meta = snap.mut_schema_meta();
            schema_meta.set_file_id(self.schema.file_id());
            schema_meta.set_version(self.schema.file_ver());
            schema_meta.set_restore_version(self.schema.restore_ver());
            schema_meta.set_keyspace_id(self.range.keyspace_id);
        }
        cs
    }

    fn new_snapshop_pb(&self) -> pb::Snapshot {
        let mut snap = pb::Snapshot::new();
        snap.set_outer_start(self.range.outer_start.to_vec());
        snap.set_outer_end(self.range.outer_end.to_vec());
        snap.set_inner_key_off(self.inner_key_off as u32);
        snap.set_properties(self.properties.to_pb(self.id));
        snap.set_base_version(self.base_version);
        snap.set_data_sequence(self.data_sequence);
        snap.set_max_ts(self.max_ts);
        snap.set_columnar_table_ids(self.columnar_table_ids.clone());
        snap.set_columnar_l2_snap_version(self.columnar_l2_snap_version.into_inner());
        self.schema.to_snapshot(&mut snap);
        snap.set_unconverted_l0s(self.unconverted_l0s.clone());
        snap.set_vector_indexes(self.vector_indexes.clone().into());
        snap.set_fts_indexes(
            self.fts_indexes
                .iter()
                .map(|&(t, i)| {
                    let mut tbl_idx_id = TableIndexId::new();
                    tbl_idx_id.set_table_id(t);
                    tbl_idx_id.set_index_id(i);
                    tbl_idx_id
                })
                .collect(),
        );
        snap.set_fts_index_metas(
            self.fts_index_metas
                .values()
                .flat_map(|indexes| indexes.values().cloned())
                .collect::<Vec<_>>()
                .into(),
        );
        snap.set_fts_l0_snap_version(self.fts_l0_snap_version.into_inner());
        snap.set_fts_l0_files(self.fts_l0_files.clone().into());
        snap.set_fts_l1_files(self.fts_l1_files.clone().into());
        snap.set_fts_l2_files(self.fts_l2_files.clone().into());
        if !self.fts_indexes.is_empty() {
            snap.set_fts_pending_l0_ids(self.collect_fts_pending_l0_ids(&[]));
        }
        snap
    }

    fn collect_fts_pending_l0_ids(&self, source_pending_l0_ids: &[u64]) -> Vec<u64> {
        let mut fts_pending_l0_ids: HashSet<u64> = self
            .fts_pending_l0_ids
            .iter()
            .filter(|id| self.files.contains_key(id))
            .copied()
            .collect();
        fts_pending_l0_ids.extend(
            self.files
                .iter()
                .filter(|(_, fm)| {
                    fm.file_type == FileType::Columnar
                        && fm.get_level() == 0
                        && fm.get_snap_version() > self.fts_l0_snap_version
                })
                .map(|(id, _)| *id),
        );
        fts_pending_l0_ids.extend(source_pending_l0_ids.iter().copied());
        fts_pending_l0_ids.into_iter().collect()
    }

    pub fn to_change_set(&self) -> pb::ChangeSet {
        let mut cs = new_change_set(self.id, self.ver);
        cs.set_sequence(self.seq);
        let mut snap = self.new_snapshop_pb();
        for (&k, v) in self.files.iter() {
            match v.file_type {
                FileType::Sst => {
                    if v.get_level() == 0 {
                        snap.mut_l0_creates().push(v.to_l0_create(k));
                    } else {
                        snap.mut_table_creates().push(v.to_table_create(k));
                    }
                }
                FileType::TxnChunk | FileType::Schema => unreachable!("not loaded to file meta"),
                FileType::Columnar => {
                    snap.mut_columnar_creates().push(v.to_columnar_create(k));
                }
                FileType::Blob => {
                    snap.mut_blob_creates().push(v.to_blob_create(k));
                }
                FileType::VectorIndex => {} // already handled in vector_indexes.
                FileType::FtsPackedFile => {} // handled separately.
                FileType::FtsDedicatedFile => {} // handled separately.
                FileType::EncryptionDict => {}
            }
        }
        // Put schema file to the changeset to trigger load schema file.
        if self.schema.file_id() > 0 {
            let mut schema_meta = SchemaMeta::new();
            schema_meta.file_id = self.schema.file_id();
            schema_meta.version = self.schema.file_ver();
            schema_meta.restore_version = self.schema.restore_ver();
            snap.set_schema_meta(schema_meta);
        }
        cs.set_snapshot(snap);
        if let Some(parent) = &self.parent {
            cs.set_parent(parent.to_change_set());
        }
        cs
    }

    pub fn marshal(&self) -> Vec<u8> {
        let cs = self.to_change_set();
        cs.write_to_bytes().unwrap()
    }

    pub fn all_file_keys(&self) -> Vec<u64> {
        let mut ids = Vec::with_capacity(self.files.len());
        for k in self.files.keys() {
            ids.push(*k);
        }
        ids
    }

    pub fn all_files(&self) -> &HashMap<u64, FileMeta> {
        &self.files
    }

    pub fn all_txn_chunk_ids(&self) -> Vec<u64> {
        TxnFileRefPropertyHelper::from_property(self.properties.get(TXN_FILE_REF))
            .unwrap()
            .get_all_chunk_ids()
    }

    pub(crate) fn get_blob_files(&self) -> Vec<(u64, Vec<u8>, Vec<u8>, u32)> {
        let mut blob_files = vec![];
        for (id, file) in &self.files {
            if file.file_type == FileType::Blob {
                blob_files.push((
                    *id,
                    file.smallest.to_vec(),
                    file.biggest.to_vec(),
                    file.table_meta_off,
                ));
            }
        }
        blob_files
    }

    /// Finds the appropriate WRITE_CF ingest level by checking from top level
    /// (0) down to bottom level. L1+ files in other CFs are ignored. L0 is
    /// always considered because its FileMeta uses cf = -1 and does not retain
    /// which CFs are present in the physical file.
    pub fn get_ingest_level(&self, table_bound: DataBound<'_>) -> u32 {
        // Check each level from 0 to bottom level to find the first level with overlap
        for level in 0..=WRITE_CF_BOTTOM_LEVEL as u8 {
            let has_overlap = self
                .files
                .values()
                .filter(|f| {
                    f.file_type == FileType::Sst
                        && f.level == level
                        && (level == 0 || f.cf == WRITE_CF as i8)
                })
                .any(|file| table_bound.overlap_bound(file.data_bound()));

            if has_overlap {
                // Found first level with overlap, ingest at the previous level
                return level.saturating_sub(1) as u32;
            }
        }

        // No overlap found at any level, ingest at the bottom level
        WRITE_CF_BOTTOM_LEVEL
    }

    pub(crate) fn data_version(&self) -> u64 {
        self.base_version + self.data_sequence
    }

    pub fn data_persisted_log_index(&self) -> u64 {
        match &self.parent {
            Some(parent) if parent.id == self.id => parent.data_sequence,
            _ => self.data_sequence,
        }
    }

    pub fn prepare_merge(&mut self, sequence: u64) {
        let parent = self.clone();
        self.ver += 1;
        self.seq = sequence;
        self.data_sequence = sequence;
        self.parent = Some(Box::new(parent));
    }

    pub fn rollback_merge(&mut self, sequence: u64) {
        self.ver += 1;
        self.seq = sequence;
    }

    pub fn commit_merge(&mut self, source: &ShardMeta, sequence: u64) {
        // If the regions are not belong to the same keyspace, reset the encryption_key
        // property.
        let belongs_to_same_keyspace =
            ApiV2::is_belongs_to_same_keyspace(&source.range.outer_start, &self.range.outer_start);
        if !belongs_to_same_keyspace {
            self.del_property(ENCRYPTION_KEY);
            self.del_property(ENCRYPTION_META_KEY);
        }

        let (clear_source, clear_target) =
            need_clear_region_data_on_merge(&source.range.outer_start, &self.range.outer_start);

        // Include the max_ts and source files in the parent for future initial flush.
        self.max_ts = std::cmp::max(self.max_ts, source.max_ts);
        let mut parent = self.clone();

        if clear_target {
            info!(
                "{} clear data of target region on merge, target: {:?}",
                self.tag(),
                self.range,
            );

            // clear all files if exists
            parent.files.clear();
            parent.vector_indexes.clear();
            parent.clear_fts();
            self.files.clear();
            self.vector_indexes.clear();
            self.clear_fts();
            // remove DEL_PREFIXES_KEY property if exists
            self.del_property(DEL_PREFIXES_KEY);
            self.del_property(STORAGE_CLASS_KEY);
        }
        if !clear_source {
            // Merge FTS state before merging source columnar file metas into `self.files`.
            // Otherwise, pending calculation based on target watermark may mistakenly
            // include source columnar L0s that are already indexed in the source shard.
            self.merge_fts(source);

            for (&id, source_file) in &source.files {
                parent.files.insert(id, source_file.clone());
            }
            for (&id, source_file) in &source.files {
                self.files.insert(id, source_file.clone());
            }
            for vec_idx in &source.vector_indexes {
                self.merge_vector_index(vec_idx);
            }

            if clear_target {
                // `inner_key_off` will be different when merge regions of different keyspaces.
                parent.inner_key_off = source.inner_key_off;
                self.inner_key_off = source.inner_key_off;

                parent.set_property_from_other(STORAGE_CLASS_KEY, source);
                self.set_property_from_other(STORAGE_CLASS_KEY, source);
            }

            // merge DEL_PREFIXES_KEY from source if needed
            let source_del_prefixes = source.get_property(DEL_PREFIXES_KEY);
            let parent_del_prefixes = self.get_property(DEL_PREFIXES_KEY);
            if let Some(new_del_prefixes) = merge_del_prefixes_if_needed(
                source_del_prefixes,
                parent_del_prefixes,
                self.range.keyspace_id,
            ) {
                self.set_property(DEL_PREFIXES_KEY, &new_del_prefixes);
            }
            self.schema.merge_from(&self.tag(), &source.schema);
        } else {
            info!(
                "{} clear data of source region on merge, source: {:?}",
                self.tag(),
                source.range,
            );
        }

        if self.range.outer_end == source.range.outer_start {
            self.range.outer_end = source.range.outer_end.clone();
        } else {
            self.range.outer_start = source.range.outer_start.clone();
        }
        self.ver = max(self.ver, source.ver) + 1;
        let source_mem_tbl_version = source.base_version + source.seq;
        let target_mem_tbl_version = self.base_version + sequence;
        self.base_version = max(source_mem_tbl_version, target_mem_tbl_version) - sequence;
        self.data_sequence = sequence;
        self.columnar_table_ids = merge_columnar_table_ids(
            &source.columnar_table_ids,
            &self.columnar_table_ids,
            source.data_bound(),
            self.data_bound(),
        );
        self.columnar_l2_snap_version = max(
            self.columnar_l2_snap_version,
            source.columnar_l2_snap_version,
        );
        // Remove columnar and vector if related table is not in the merged
        // columnar_table_ids.
        self.retain_columnar_and_indexes();
        self.parent = Some(Box::new(parent));
        self.seq = sequence;
    }

    fn retain_columnar_and_indexes(&mut self) {
        let columnar_table_ids = self.columnar_table_ids.clone();
        self.files.retain(|_, file| {
            if !file.is_columnar_file() && !file.is_vector_index_file() && !file.is_fts_file() {
                return true;
            }
            let data_bound = file.data_bound();
            let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
            columnar_table_ids
                .iter()
                .any(|&id| id >= min_table_id && id <= max_table_id)
        });
        // Source or target shard may has pending columnar major compaction, after merge
        // the columnar_table_ids will not contains the table. If columnar_table_ids is
        // empty, we should also clear the unconverted_l0s.
        if columnar_table_ids.is_empty() {
            self.unconverted_l0s.clear();
        }
        self.vector_indexes
            .retain(|vec| columnar_table_ids.contains(&vec.table_id));
        self.retain_fts(&columnar_table_ids);
    }

    fn merge_vector_index(&mut self, vec_idx: &VectorIndex) {
        if let Some(old_idx) = self
            .vector_indexes
            .iter_mut()
            .find(|v| is_same_vector_index(v, vec_idx))
        {
            let old_file_ids: HashSet<u64> = old_idx.files.iter().map(|f| f.id).collect();
            for file in vec_idx.files.iter() {
                if !old_file_ids.contains(&file.id) {
                    old_idx.files.push(file.clone());
                }
            }
            let snap_version =
                std::cmp::max(old_idx.get_snap_version(), vec_idx.get_snap_version());
            old_idx.set_snap_version(snap_version);
        } else {
            self.vector_indexes.push(vec_idx.clone());
        }
    }

    fn merge_fts(&mut self, source: &ShardMeta) {
        for &(table_id, index_id) in &source.fts_indexes {
            let meta = source
                .fts_index_metas
                .get(&table_id)
                .and_then(|indexes| indexes.get(&index_id))
                .cloned()
                .unwrap_or_else(|| new_legacy_tracked_index_meta(table_id, index_id));
            self.insert_fts_index_meta(meta);
        }
        Self::extend_unique_fts_files(&mut self.fts_l0_files, &source.fts_l0_files, |f| f.get_id());
        Self::extend_unique_fts_files(&mut self.fts_l1_files, &source.fts_l1_files, |f| f.get_id());
        Self::extend_unique_fts_files(&mut self.fts_l2_files, &source.fts_l2_files, |f| f.get_id());
        // See `kvengine::split::merge_fts` for the reasoning behind adding
        // "source-unindexed" columnar L0 ids to the pending list.
        let mut source_pending_l0_ids = source.fts_pending_l0_ids.clone();
        source_pending_l0_ids.extend(
            source
                .files
                .iter()
                .filter(|(_, fm)| {
                    fm.file_type == FileType::Columnar
                        && fm.get_level() == 0
                        && fm.get_snap_version() > source.fts_l0_snap_version
                })
                .map(|(id, _)| *id),
        );
        self.fts_pending_l0_ids = self.collect_fts_pending_l0_ids(&source_pending_l0_ids);
        self.fts_l0_snap_version = max(self.fts_l0_snap_version, source.fts_l0_snap_version);
    }

    fn extend_unique_fts_files<T, F>(target: &mut Vec<T>, source: &[T], id_fn: F)
    where
        T: Clone,
        F: Fn(&T) -> u64,
    {
        let mut existing = HashSet::with_capacity(target.len() + source.len());
        for item in target.iter() {
            existing.insert(id_fn(item));
        }
        for item in source {
            let id = id_fn(item);
            if existing.insert(id) {
                target.push(item.clone());
            }
        }
    }

    fn clear_fts(&mut self) {
        self.fts_indexes.clear();
        self.fts_index_metas.clear();
        self.fts_l0_files.clear();
        self.fts_l1_files.clear();
        self.fts_l2_files.clear();
        self.fts_l0_snap_version = SnapVersion::zero();
        self.fts_pending_l0_ids.clear();
    }

    fn retain_fts(&mut self, columnar_table_ids: &[i64]) {
        self.fts_indexes
            .retain(|(table_id, _)| columnar_table_ids.contains(table_id));
        self.fts_index_metas
            .retain(|table_id, _| columnar_table_ids.contains(table_id));
        self.fts_l0_files.retain(|fts_l0_file| {
            let data_bound = DataBound::new(
                InnerKey::from_inner_buf(&fts_l0_file.smallest),
                InnerKey::from_inner_buf(&fts_l0_file.biggest),
                true,
            );
            let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
            columnar_table_ids
                .iter()
                .any(|&id| id >= min_table_id && id <= max_table_id)
        });
        self.fts_l1_files.retain(|fts_l1_file| {
            let data_bound = DataBound::new(
                InnerKey::from_inner_buf(&fts_l1_file.smallest),
                InnerKey::from_inner_buf(&fts_l1_file.biggest),
                true,
            );
            let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
            columnar_table_ids
                .iter()
                .any(|&id| id >= min_table_id && id <= max_table_id)
        });
        self.fts_l2_files.retain(|fts_l2_file| {
            let data_bound = DataBound::new(
                InnerKey::from_inner_buf(&fts_l2_file.smallest),
                InnerKey::from_inner_buf(&fts_l2_file.biggest),
                true,
            );
            let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
            columnar_table_ids
                .iter()
                .any(|&id| id >= min_table_id && id <= max_table_id)
        });
        let files = &self.files;
        self.fts_pending_l0_ids.retain(|pending_id| {
            files.get(pending_id).is_some_and(|fm| {
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(fm.data_bound());
                columnar_table_ids
                    .iter()
                    .any(|&id| id >= min_table_id && id <= max_table_id)
            })
        });
    }

    pub fn merge_txn_file_ref(&mut self, wb_ref: &TxnFileRef, log_index: u64) {
        let tag = self.tag();
        let current_seq = self.txn_file_locks.seq();

        debug_assert!(
            current_seq < log_index,
            "{} duplicated merge txn file ref, seq {}, log_index {}, wb_ref {:?}",
            tag,
            current_seq,
            log_index,
            wb_ref
        );

        if wb_ref.is_locked() {
            self.txn_file_locks
                .insert(log_index, wb_ref.start_ts, wb_ref.get_lock_val_prefix());
        } else {
            let existed = self.txn_file_locks.remove(log_index, wb_ref.start_ts);
            if !existed {
                warn!(
                    "{} unexpected txn not existed, wb_ref {:?}, log_index {}, last_seq {}, current locks {:?}",
                    tag, wb_ref, log_index, current_seq, self.txn_file_locks
                );
            }
        }
        debug!("{} ShardMeta merge txn file ref", tag;
            "wb_ref" => ?wb_ref,
            "log_index" => log_index,
            "locks" => %self.txn_file_locks);
    }

    pub fn has_txn_file_locks(&self) -> bool {
        !self.txn_file_locks.is_empty()
    }

    pub fn txn_file_locks(&self) -> &TxnFileLocks {
        &self.txn_file_locks
    }

    pub fn recover_txn_file_locks_from_shard(&mut self, shard: &Shard) {
        self.txn_file_locks = TxnFileLocks::from_lock_txn_files(
            shard.get_write_sequence(),
            shard.get_data().get_lock_txn_files(),
        );
        debug!("{} recover txn file locks from shard", self.tag(); "locks" => ?self.txn_file_locks);
    }

    pub fn collect_files_for_storage_class(
        &mut self,
        cs: &kvenginepb::ChangeSet,
    ) -> Option<pb::Snapshot> {
        debug_assert_eq!(cs.get_property_key(), STORAGE_CLASS_KEY);
        let mut snap = pb::Snapshot::new();
        for (&k, v) in self.files.iter() {
            if v.file_type == FileType::Sst && v.get_level() > 0 && v.get_cf() as usize == WRITE_CF
            {
                snap.mut_table_creates().push(v.to_table_create(k));
            }
        }
        if snap.table_creates.is_empty() {
            return None;
        }
        Some(snap)
    }

    pub fn get_storage_class_spec(&self) -> StorageClassSpec {
        StorageClassSpec::unmarshal(self.get_property(STORAGE_CLASS_KEY).as_deref())
    }

    fn apply_update_storage_class(&self, cs: &pb::ChangeSet) {
        debug_assert_eq!(cs.get_property_key(), STORAGE_CLASS_KEY);
        let sc_spec = StorageClassSpec::unmarshal(Some(cs.get_property_value()));
        if sc_spec.is_specified() {
            debug!("{} set storage class spec", self.tag(); "spec" => ?sc_spec);
            self.properties
                .set(cs.get_property_key(), cs.get_property_value());
        } else {
            debug!("{} clear storage class spec", self.tag());
            self.properties.remove(cs.get_property_key());
        }
    }

    // Used in native_br.
    pub fn clear_columnar_related_meta(&mut self) {
        self.files.retain(|_, file| {
            file.file_type != FileType::Columnar && file.file_type != FileType::VectorIndex
        });
        self.columnar_table_ids.clear();
        self.columnar_l2_snap_version = SnapVersion::zero();
        self.unconverted_l0s.clear();
        self.vector_indexes.clear();
        self.clear_fts();
        if let Some(parent) = &mut self.parent {
            parent.clear_columnar_related_meta();
        }
    }
}

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

#[derive(Clone, Debug)]
pub struct FileMeta {
    pub cf: i8,
    pub level: u8,
    pub file_type: FileType,
    pub smallest: Bytes,
    pub biggest: Bytes,

    // Available ONLY for SST level 0.
    pub l0_size: u32,

    // Available ONLY for SST (level 1+) & Columnar & VectorIndex & Blob.
    pub table_meta_off: u32,

    // Available ONLY for Columnar.
    pub snap_version: Option<SnapVersion>,
}

impl FileMeta {
    fn new(
        cf: i32,
        level: u32,
        file_type: FileType,
        smallest: &[u8],
        biggest: &[u8],
        l0_size: u32,
        table_meta_off: u32,
        snap_version: Option<SnapVersion>,
    ) -> Self {
        Self {
            cf: cf as i8,
            level: level as u8,
            file_type,
            smallest: Bytes::copy_from_slice(smallest),
            biggest: Bytes::copy_from_slice(biggest),
            l0_size,
            table_meta_off,
            snap_version,
        }
    }

    pub fn get_level(&self) -> u32 {
        self.level as u32
    }

    pub fn get_cf(&self) -> i32 {
        self.cf as i32
    }

    pub fn has_locks(&self) -> bool {
        self.level == 0 || self.cf == LOCK_CF as i8
    }

    pub fn is_schema_file(&self) -> bool {
        self.file_type == FileType::Schema
    }

    pub fn is_sst_file(&self) -> bool {
        self.file_type == FileType::Sst
    }

    pub fn is_columnar_file(&self) -> bool {
        self.file_type == FileType::Columnar
    }

    pub fn is_vector_index_file(&self) -> bool {
        self.file_type == FileType::VectorIndex
    }

    pub fn is_fts_file(&self) -> bool {
        self.file_type == FileType::FtsPackedFile || self.file_type == FileType::FtsDedicatedFile
    }

    pub fn can_use_ia(&self) -> bool {
        match self.file_type {
            FileType::Sst if (self.cf as usize == WRITE_CF && self.level > 0) => true,
            FileType::Columnar => true,
            FileType::VectorIndex => true,
            FileType::FtsPackedFile => true,
            FileType::FtsDedicatedFile => true,
            FileType::Blob => true,
            _ => false,
        }
    }

    pub fn get_snap_version(&self) -> SnapVersion {
        self.snap_version.unwrap_or(SnapVersion::zero())
    }

    pub fn is_l0_sst_with_size(&self) -> bool {
        self.file_type == FileType::Sst && self.level == 0 && self.l0_size > 0
    }

    /// Size usable as a `Dfs::read_file_with_size_hint` argument. Returns
    /// `None` for file types whose size is not carried in meta (Schema /
    /// TxnChunk / Fts); callers should fall back to `read_file` in that
    /// case. L1+ SST / Columnar / Blob: `table_meta_off` is the body
    /// length, which undercounts the meta section + footer by a few KB —
    /// acceptable against a GB-scale memory limit budget.
    pub fn size_hint_for_download(&self) -> Option<u64> {
        if self.is_l0_sst_with_size() {
            Some(self.l0_size as u64)
        } else if self.table_meta_off > 0 {
            Some(self.table_meta_off as u64)
        } else {
            None
        }
    }

    pub fn from_l0_table(table: &kvenginepb::L0Create) -> Self {
        Self::new(
            -1,
            0,
            FileType::Sst,
            table.get_smallest(),
            table.get_biggest(),
            table.size,
            0,
            None,
        )
    }

    pub fn from_table(table: &kvenginepb::TableCreate) -> Self {
        Self::new(
            table.cf,
            table.level,
            FileType::Sst,
            table.get_smallest(),
            table.get_biggest(),
            0,
            table.meta_offset,
            None,
        )
    }

    pub fn from_columnar_table(table: &kvenginepb::ColumnarCreate) -> Self {
        Self::new(
            0,
            table.level,
            FileType::Columnar,
            table.get_smallest(),
            table.get_biggest(),
            0,
            table.meta_offset,
            Some(SnapVersion::from(table.snap_version)),
        )
    }

    pub fn from_blob_table(table: &kvenginepb::BlobCreate) -> Self {
        Self::new(
            -1,
            0,
            FileType::Blob,
            table.get_smallest(),
            table.get_biggest(),
            0,
            table.get_meta_offset(),
            None,
        )
    }

    pub fn from_vector_index_file(vec_idx_file: &kvenginepb::VectorIndexFile) -> Self {
        Self::new(
            0,
            0,
            FileType::VectorIndex,
            vec_idx_file.get_smallest(),
            vec_idx_file.get_biggest(),
            0,
            vec_idx_file.get_meta_offset(),
            None,
        )
    }

    pub fn from_fts_l0_file(fts_file: &kvenginepb::FtsPackedFileInfo) -> Self {
        Self::new(
            0,
            0,
            FileType::FtsPackedFile,
            fts_file.get_smallest(),
            fts_file.get_biggest(),
            0,
            fts_file.get_meta_offset(),
            None,
        )
    }

    pub fn from_fts_l1_file(fts_file: &kvenginepb::FtsPackedFileInfo) -> Self {
        Self::new(
            0,
            1,
            FileType::FtsPackedFile,
            fts_file.get_smallest(),
            fts_file.get_biggest(),
            0,
            fts_file.get_meta_offset(),
            None,
        )
    }

    pub fn from_fts_l2_file(fts_file: &kvenginepb::FtsDedFileInfo) -> Self {
        Self::new(
            0,
            2,
            FileType::FtsDedicatedFile,
            fts_file.get_smallest(),
            fts_file.get_biggest(),
            0,
            fts_file.get_meta_offset(),
            None,
        )
    }

    pub fn from_schema_meta() -> Self {
        Self::new(0, 0, FileType::Schema, &[], &[], 0, 0, None)
    }

    pub fn to_l0_create(&self, id: u64) -> kvenginepb::L0Create {
        let mut l0_create = kvenginepb::L0Create::new();
        l0_create.set_id(id);
        l0_create.set_smallest(self.smallest.to_vec());
        l0_create.set_biggest(self.biggest.to_vec());
        l0_create.set_size(self.l0_size);
        l0_create
    }

    pub fn to_table_create(&self, id: u64) -> kvenginepb::TableCreate {
        let mut table_create = kvenginepb::TableCreate::new();
        table_create.set_id(id);
        table_create.set_cf(self.cf as i32);
        table_create.set_level(self.get_level());
        table_create.set_smallest(self.smallest.to_vec());
        table_create.set_biggest(self.biggest.to_vec());
        table_create.set_meta_offset(self.table_meta_off);
        table_create
    }

    pub fn to_columnar_create(&self, id: u64) -> kvenginepb::ColumnarCreate {
        let mut columnar_create = kvenginepb::ColumnarCreate::new();
        columnar_create.set_id(id);
        columnar_create.set_level(self.get_level());
        columnar_create.set_smallest(self.smallest.to_vec());
        columnar_create.set_biggest(self.biggest.to_vec());
        columnar_create.set_meta_offset(self.table_meta_off);
        columnar_create.set_snap_version(self.snap_version.map(|v| v.into_inner()).unwrap_or(0));
        columnar_create
    }

    pub fn to_blob_create(&self, id: u64) -> kvenginepb::BlobCreate {
        let mut blob_create = kvenginepb::BlobCreate::new();
        blob_create.set_id(id);
        blob_create.set_smallest(self.smallest.to_vec());
        blob_create.set_biggest(self.biggest.to_vec());
        blob_create
    }
}

impl BoundedDataSet for FileMeta {
    fn data_bound(&self) -> DataBound<'_> {
        DataBound::new(
            InnerKey::from_inner_buf(&self.smallest),
            InnerKey::from_inner_buf(&self.biggest),
            true,
        )
    }
}

impl Default for FileMeta {
    fn default() -> Self {
        Self::new(0, 0, FileType::Sst, b"", b"", 0, 0, None)
    }
}

pub fn is_move_down(comp: &pb::Compaction) -> bool {
    !comp.top_deletes.is_empty()
        && comp.top_deletes.len() == comp.table_creates.len()
        && comp.top_deletes[0] == comp.table_creates[0].id
}

pub fn new_change_set(shard_id: u64, shard_ver: u64) -> pb::ChangeSet {
    let mut cs = pb::ChangeSet::new();
    cs.set_shard_id(shard_id);
    cs.set_shard_ver(shard_ver);
    cs
}

pub const GLOBAL_SHARD_END_KEY: &[u8] = &[255, 255, 255, 255, 255, 255, 255, 255];

#[derive(Default, Clone, Debug)]
pub struct SchemaFileMeta {
    schema_file_id: u64,
    schema_file_ver: i64,
    schema_restore_ver: u64,
}

impl SchemaFileMeta {
    pub fn is_valid(&self) -> bool {
        self.schema_file_id > 0
    }

    #[inline]
    pub fn file_id(&self) -> u64 {
        self.schema_file_id
    }

    #[inline]
    pub fn file_ver(&self) -> i64 {
        self.schema_file_ver
    }

    #[inline]
    pub fn restore_ver(&self) -> u64 {
        self.schema_restore_ver
    }

    #[inline]
    pub fn has_value(&self) -> bool {
        self.schema_file_id > 0 || self.schema_file_ver > 0 || self.schema_restore_ver > 0
    }

    pub fn from_snapshot(snap: &pb::Snapshot) -> Self {
        debug_assert!(snap.has_schema_meta());
        let sm = snap.get_schema_meta();
        Self {
            schema_file_id: sm.get_file_id(),
            schema_file_ver: sm.get_version(),
            schema_restore_ver: sm.get_restore_version(),
        }
    }

    pub fn update_from_schema_meta(&mut self, cs: &pb::ChangeSet) {
        debug_assert!(cs.has_update_schema_meta());
        let sm = cs.get_update_schema_meta();
        self.schema_file_id = sm.get_file_id();
        self.schema_file_ver = sm.get_version();
        debug_assert_eq!(self.schema_restore_ver, sm.get_restore_version());
    }

    pub fn update_by_restore(
        &mut self,
        new_file_id: u64,
        schema_file_ver: i64,
        schema_restore_ver: u64,
    ) {
        self.schema_file_id = new_file_id;
        self.schema_file_ver = schema_file_ver;
        self.schema_restore_ver = schema_restore_ver;
    }

    pub fn clear(&mut self, reset_restore_version: u64) {
        self.schema_file_id = 0;
        self.schema_restore_ver = reset_restore_version;
    }

    pub fn to_snapshot(&self, snap: &mut pb::Snapshot) {
        if self.is_valid() || self.schema_restore_ver > 0 {
            let mut sm = SchemaMeta::new();
            sm.set_file_id(self.schema_file_id);
            sm.set_version(self.schema_file_ver);
            sm.set_restore_version(self.schema_restore_ver);
            snap.set_schema_meta(sm);
        }
    }

    pub fn merge_from(&mut self, tag: &ShardTag, source: &SchemaFileMeta) {
        if source.schema_restore_ver == self.schema_restore_ver
            && source.schema_file_ver >= self.schema_file_ver
        {
            self.schema_file_id = source.schema_file_id;
            self.schema_file_ver = source.schema_file_ver;
        } else if source.schema_restore_ver != self.schema_restore_ver
            && source.schema_file_id > self.schema_file_id
        {
            // Happens during restore and only one of the source/target has been restored.
            // The restored schema should have larger schema file id.
            info!("{} merge schema file meta with different restore ver", tag;
                "current" => ?self, "source" => ?source);
            *self = source.clone();
        }
    }
}

#[cfg(test)]
mod tests {
    use std::{
        iter::{FromIterator, Iterator},
        sync::Arc,
    };

    use kvenginepb::fts::FullTextIndexFieldDef;
    use tidb_query_datatype::codec::table::encode_row_key;

    use super::*;
    use crate::{
        EncryptionState,
        limiter::{LimiterOptions, RegionLimiter},
        shard::ShardDataBuilder,
        table::{
            ChecksumType, Value,
            file::InMemFile,
            fts::new_tracked_index_meta,
            memtable::MemTableImplementation,
            sstable::{BlockCache, L0Builder, L0Table},
        },
    };

    fn tracked_field(col_id: i64) -> FullTextIndexFieldDef {
        let mut field = FullTextIndexFieldDef::new();
        field.set_col_id(col_id);
        field
    }

    // Helper to create a ShardMeta with files for testing.
    // files: (id, cf, level, smallest, biggest)
    #[cfg(any(test, feature = "testexport"))]
    fn make_meta_with_files(files: &[(u64, i32, u32, &[u8], &[u8])]) -> ShardMeta {
        let mut meta = ShardMeta::default();
        for &(id, cf, level, smallest, biggest) in files {
            let file_meta = FileMeta {
                cf: cf as i8,
                level: level as u8,
                smallest: smallest.to_vec().into(),
                biggest: biggest.to_vec().into(),
                file_type: FileType::Sst,
                ..Default::default()
            };
            meta.add_file(id, file_meta);
        }
        meta
    }

    fn empty_shard_data() -> ShardData {
        ShardData::new_empty(
            ShardRange::default(),
            0,
            RegionLimiter::new(LimiterOptions::default(), LimiterOptions::default()),
            MemTableImplementation::Crossbeam,
        )
    }

    fn shard_data_with_l0(l0: L0Table) -> ShardData {
        let mut builder = ShardDataBuilder::new(empty_shard_data());
        builder.set_l0_tbls(vec![l0]);
        builder.build()
    }

    fn build_l0(id: u64, entries: &[(usize, &[u8])]) -> L0Table {
        let mut builder = L0Builder::new(id, 1024, 1.into(), ChecksumType::Crc32c, None);
        for (cf, key) in entries {
            builder
                .add(
                    *cf,
                    InnerKey::from_inner_buf(key),
                    &Value::new_with_version(b"value", 1),
                    None,
                )
                .unwrap();
        }
        let (_, data) = builder.finish().unwrap();
        L0Table::new(
            Arc::new(InMemFile::new(id, data)),
            BlockCache::None,
            false,
            EncryptionState::Disabled,
        )
        .unwrap()
        .unwrap()
    }

    #[test]
    fn test_fts_update_remove_tracked_indexes() {
        let mut meta = ShardMeta::default();
        meta.fts_indexes.insert((10, 1));
        meta.fts_indexes.insert((10, 2));
        meta.fts_pending_l0_ids = vec![100, 200];

        let mut update = pb::FtsUpdate::new();
        let mut tbl_idx_id = TableIndexId::new();
        tbl_idx_id.set_table_id(10);
        tbl_idx_id.set_index_id(1);
        update.mut_remove_tracked_indexes().push(tbl_idx_id);

        meta.apply_fts_update(&update);

        assert!(!meta.fts_indexes.contains(&(10, 1)));
        assert!(meta.fts_indexes.contains(&(10, 2)));
        assert_eq!(
            meta.fts_pending_l0_ids,
            vec![100, 200],
            "pending ids should be kept if there are remaining tracked indexes"
        );

        let mut update = pb::FtsUpdate::new();
        let mut tbl_idx_id = TableIndexId::new();
        tbl_idx_id.set_table_id(10);
        tbl_idx_id.set_index_id(2);
        update.mut_remove_tracked_indexes().push(tbl_idx_id);

        meta.apply_fts_update(&update);
        assert!(meta.fts_indexes.is_empty());
        assert!(
            meta.fts_pending_l0_ids.is_empty(),
            "pending ids should be cleared when no indexes are tracked"
        );
    }

    #[test]
    fn test_fts_update_add_tracked_index_meta() {
        let mut meta = ShardMeta::default();

        let mut update = pb::FtsUpdate::new();
        update
            .mut_add_tracked_index_metas()
            .push(new_tracked_index_meta(
                10,
                3,
                vec![tracked_field(1001)],
                pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody,
            ));

        meta.apply_fts_update(&update);

        assert!(meta.fts_indexes.contains(&(10, 3)));
        let tracked_meta = meta
            .fts_index_metas
            .get(&10)
            .and_then(|indexes| indexes.get(&3))
            .unwrap();
        assert_eq!(tracked_meta.get_index_fields()[0].get_col_id(), 1001);
        assert_eq!(
            tracked_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody
        );
    }

    #[test]
    fn test_fts_update_applies_meta_before_legacy_ids() {
        let mut meta = ShardMeta::default();

        let mut update = pb::FtsUpdate::new();
        let mut duplicated_legacy_idx = TableIndexId::new();
        duplicated_legacy_idx.set_table_id(10);
        duplicated_legacy_idx.set_index_id(3);
        update.mut_add_tracked_indexes().push(duplicated_legacy_idx);
        let mut legacy_idx = TableIndexId::new();
        legacy_idx.set_table_id(10);
        legacy_idx.set_index_id(4);
        update.mut_add_tracked_indexes().push(legacy_idx);
        update
            .mut_add_tracked_index_metas()
            .push(new_tracked_index_meta(
                10,
                3,
                vec![tracked_field(1001), tracked_field(1002)],
                pb::fts::FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
            ));

        meta.apply_fts_update(&update);

        assert!(meta.fts_indexes.contains(&(10, 3)));
        assert!(meta.fts_indexes.contains(&(10, 4)));
        let tracked_meta = meta
            .fts_index_metas
            .get(&10)
            .and_then(|indexes| indexes.get(&3))
            .unwrap();
        assert_eq!(tracked_meta.get_index_fields()[0].get_col_id(), 1001);
        assert_eq!(tracked_meta.get_index_fields()[1].get_col_id(), 1002);
        assert_eq!(
            tracked_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString
        );
        let legacy_meta = meta
            .fts_index_metas
            .get(&10)
            .and_then(|indexes| indexes.get(&4))
            .unwrap();
        assert!(legacy_meta.get_index_fields().is_empty());
        assert_eq!(
            legacy_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody
        );
    }

    #[test]
    fn test_insert_fts_index_meta_upgrades_placeholder_to_ordinal_layout() {
        let mut meta = ShardMeta::default();

        meta.insert_fts_index_meta(new_legacy_tracked_index_meta(10, 3));
        meta.insert_fts_index_meta(new_tracked_index_meta(
            10,
            3,
            vec![tracked_field(1001), tracked_field(1002)],
            pb::fts::FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString,
        ));

        let tracked_meta = meta
            .fts_index_metas
            .get(&10)
            .and_then(|indexes| indexes.get(&3))
            .unwrap();
        assert_eq!(tracked_meta.get_index_fields()[0].get_col_id(), 1001);
        assert_eq!(tracked_meta.get_index_fields()[1].get_col_id(), 1002);
        assert_eq!(
            tracked_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutIndexFieldOrdinalString
        );
    }

    #[test]
    fn test_legacy_snapshot_fts_indexes_materialize_as_legacy_meta() {
        let mut cs = new_change_set(11, 1);
        let snap = cs.mut_snapshot();
        snap.set_outer_end(GLOBAL_SHARD_END_KEY.to_vec());
        let mut tbl_idx_id = TableIndexId::new();
        tbl_idx_id.set_table_id(10);
        tbl_idx_id.set_index_id(7);
        snap.mut_fts_indexes().push(tbl_idx_id);

        let meta = ShardMeta::new(1, &cs);
        let tracked_meta = meta
            .fts_index_metas
            .get(&10)
            .and_then(|indexes| indexes.get(&7))
            .unwrap();
        assert!(tracked_meta.get_index_fields().is_empty());
        assert_eq!(
            tracked_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody
        );
    }

    #[test]
    fn test_snapshot_serializes_fts_index_metas() {
        let mut meta = new_shard_meta_for_range(1, 70, 0, 20, 5);
        meta.insert_fts_index_meta(new_tracked_index_meta(
            70,
            9,
            vec![tracked_field(42)],
            pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody,
        ));

        let snap = meta.new_snapshop_pb();

        assert_eq!(snap.get_fts_indexes().len(), 1);
        assert_eq!(snap.get_fts_index_metas().len(), 1);
        let tracked_meta = &snap.get_fts_index_metas()[0];
        assert_eq!(tracked_meta.get_table_id(), 70);
        assert_eq!(tracked_meta.get_index_id(), 9);
        assert_eq!(tracked_meta.get_index_fields()[0].get_col_id(), 42);
        assert_eq!(
            tracked_meta.get_field_layout(),
            pb::fts::FtsFieldLayout::FtsFieldLayoutLegacyBody
        );
    }

    #[test]
    fn test_ingest_level() {
        // Test basic single-level scenarios
        for level in 0..=3 {
            let mut cs = new_change_set(1, 1);
            let snap = cs.mut_snapshot();
            snap.set_outer_end(GLOBAL_SHARD_END_KEY.to_vec());
            snap.set_max_ts(100);
            let mut id = 0;
            let mut make_table = |level: u32, smallest: &str, biggest: &str| {
                id += 1;
                let mut tbl = pb::TableCreate::new();
                tbl.id = id;
                tbl.level = level;
                tbl.smallest = smallest.as_bytes().to_vec();
                tbl.biggest = biggest.as_bytes().to_vec();
                tbl
            };
            let tables = snap.mut_table_creates();
            tables.push(make_table(level, "1000", "2000"));
            let meta = ShardMeta::new(1, &cs);
            assert_eq!(meta.max_ts, 100);
            let assert_get_ingest_level = |smallest: &str, biggest: &str, level| {
                assert_eq!(
                    meta.get_ingest_level(DataBound::new(
                        InnerKey::from_inner_buf(smallest.as_bytes()),
                        InnerKey::from_inner_buf(biggest.as_bytes()),
                        true
                    )),
                    level
                );
            };
            let overlap_level = level.saturating_sub(1);
            assert_get_ingest_level("0000", "0999", 3);
            assert_get_ingest_level("1000", "1500", overlap_level);
            assert_get_ingest_level("1000", "2000", overlap_level);
            assert_get_ingest_level("1500", "2000", overlap_level);
            assert_get_ingest_level("1500", "2999", overlap_level);
            assert_get_ingest_level("2000", "2999", overlap_level);
            assert_get_ingest_level("2500", "2999", 3);
        }

        // Test multiple-level overlap scenarios
        {
            let mut cs = new_change_set(1, 1);
            let snap = cs.mut_snapshot();
            snap.set_outer_end(GLOBAL_SHARD_END_KEY.to_vec());
            snap.set_max_ts(100);

            let mut id = 0;
            let mut make_table = |level: u32, smallest: &str, biggest: &str| {
                id += 1;
                let mut tbl = pb::TableCreate::new();
                tbl.id = id;
                tbl.level = level;
                tbl.smallest = smallest.as_bytes().to_vec();
                tbl.biggest = biggest.as_bytes().to_vec();
                tbl
            };

            let tables = snap.mut_table_creates();
            // Add files at different levels that will create overlaps
            tables.push(make_table(3, "1000", "2000")); // Level 3: 1000-2000
            tables.push(make_table(2, "1500", "2500")); // Level 2: 1500-2500 (overlaps with level 3)
            tables.push(make_table(1, "3000", "4000")); // Level 1: 3000-4000 (no overlap)

            let meta = ShardMeta::new(1, &cs);

            let assert_get_ingest_level = |smallest: &str, biggest: &str, expected_level| {
                let actual_level = meta.get_ingest_level(DataBound::new(
                    InnerKey::from_inner_buf(smallest.as_bytes()),
                    InnerKey::from_inner_buf(biggest.as_bytes()),
                    true,
                ));
                assert_eq!(
                    actual_level, expected_level,
                    "Failed for range {}-{}: expected level {}, got level {}",
                    smallest, biggest, expected_level, actual_level
                );
            };

            // Test case 1: New file overlaps with level 2
            // Range 1200-1800 overlaps with level 2 (1500-2500)
            // Algorithm finds overlap at level 2, returns level 1
            assert_get_ingest_level("1200", "1800", 1);

            // Test case 2: New file overlaps with level 2
            // Range 1600-2400 overlaps with level 2 (1500-2500)
            // Algorithm finds overlap at level 2, returns level 1
            assert_get_ingest_level("1600", "2400", 1);

            // Test case 3: New file overlaps with level 3 but not level 2
            // Range 0900-1200 overlaps with level 3 (1000-2000) but not level 2 (1500-2500)
            // Algorithm finds overlap at level 3, returns level 2
            assert_get_ingest_level("0900", "1200", 2);

            // Test case 4: No overlap with any existing files
            // Range 5000-6000 doesn't overlap with any existing files
            // Should go to bottom level (3)
            assert_get_ingest_level("5000", "6000", 3);

            // Test case 5: File that overlaps with level 1 (the topmost level)
            // Range 2800-3200 overlaps with level 1 (3000-4000)
            // Algorithm finds overlap at level 1, returns level 0
            assert_get_ingest_level("2800", "3200", 0);
        }
    }

    #[test]
    fn test_ingest_level_ignores_other_cfs() {
        let table_bound = || {
            DataBound::new(
                InnerKey::from_inner_buf(b"1500"),
                InnerKey::from_inner_buf(b"1600"),
                true,
            )
        };

        // L1+ overlap is only meaningful within WRITE_CF for the regular
        // selector.
        let meta = make_meta_with_files(&[(1, EXTRA_CF as i32, 1, b"1000", b"2000")]);
        assert_eq!(meta.get_ingest_level(table_bound()), WRITE_CF_BOTTOM_LEVEL);

        // Ignore the EXTRA_CF L1 file and choose the level above the actual
        // WRITE_CF overlap at L2.
        let meta = make_meta_with_files(&[
            (1, EXTRA_CF as i32, 1, b"1000", b"2000"),
            (2, WRITE_CF as i32, 2, b"1000", b"2000"),
        ]);
        assert_eq!(meta.get_ingest_level(table_bound()), 1);

        // L0 FileMeta uses cf = -1 and cannot identify which CFs are present,
        // so its overlap must still be considered.
        let meta = make_meta_with_files(&[(1, -1, 0, b"1000", b"2000")]);
        assert_eq!(meta.get_ingest_level(table_bound()), 0);
    }

    #[test]
    fn test_legacy_ingest_level_ignores_other_cfs_and_preserves_topmost_level() {
        let table_bound = || {
            DataBound::new(
                InnerKey::from_inner_buf(b"1500"),
                InnerKey::from_inner_buf(b"1600"),
                true,
            )
        };
        let ingest_level = |meta: &ShardMeta| {
            LegacyIngestLevelSnapshot::from_shard_meta(meta).get_ingest_level(table_bound())
        };

        // A non-overlapping WRITE_CF L1 still keeps legacy ingest at L1 to
        // avoid racing with a concurrent compaction.
        let meta = make_meta_with_files(&[(1, WRITE_CF as i32, 1, b"3000", b"4000")]);
        assert_eq!(ingest_level(&meta), 1);

        // An L1 file from another CF does not impose the legacy guard.
        let meta = make_meta_with_files(&[(1, EXTRA_CF as i32, 1, b"1000", b"2000")]);
        assert_eq!(ingest_level(&meta), WRITE_CF_BOTTOM_LEVEL);

        // EXTRA_CF files do not affect the topmost WRITE_CF level.
        let meta = make_meta_with_files(&[
            (1, EXTRA_CF as i32, 1, b"1000", b"2000"),
            (2, WRITE_CF as i32, 2, b"3000", b"4000"),
        ]);
        assert_eq!(ingest_level(&meta), 2);

        // An overlapping WRITE_CF L2 file chooses L1.
        let meta = make_meta_with_files(&[(1, WRITE_CF as i32, 2, b"1000", b"2000")]);
        assert_eq!(ingest_level(&meta), 1);
    }

    #[test]
    fn test_legacy_ingest_level_uses_write_cf_l0_range() {
        let table_bound = || {
            DataBound::new(
                InnerKey::from_inner_buf(b"1500"),
                InnerKey::from_inner_buf(b"1600"),
                true,
            )
        };

        // The physical L0 range overlaps, but only LOCK_CF covers the ingest
        // range. The WRITE_CF range is elsewhere and must not force L0.
        let shard_data = shard_data_with_l0(build_l0(
            1,
            &[
                (LOCK_CF, b"1000"),
                (LOCK_CF, b"2000"),
                (WRITE_CF, b"3000"),
                (WRITE_CF, b"4000"),
            ],
        ));
        let level_snapshot = LegacyIngestLevelSnapshot::from_shard_data(&shard_data);
        assert_eq!(
            level_snapshot.get_ingest_level(table_bound()),
            WRITE_CF_BOTTOM_LEVEL
        );

        // A WRITE_CF L0 range overlap still requires an L0 ingest.
        let shard_data =
            shard_data_with_l0(build_l0(1, &[(WRITE_CF, b"1000"), (WRITE_CF, b"2000")]));
        let level_snapshot = LegacyIngestLevelSnapshot::from_shard_data(&shard_data);
        assert_eq!(level_snapshot.get_ingest_level(table_bound()), 0);
    }

    #[test]
    fn test_legacy_ingest_level_ignores_columnar_overlap() {
        let table_bound = || {
            DataBound::new(
                InnerKey::from_inner_buf(b"1500"),
                InnerKey::from_inner_buf(b"1600"),
                true,
            )
        };
        for level in [1, 2] {
            let mut meta = make_meta_with_files(&[(1, WRITE_CF as i32, level, b"3000", b"4000")]);
            let mut columnar = pb::ColumnarCreate::new();
            columnar.set_id(2);
            columnar.set_level(level);
            columnar.set_smallest(b"1000".to_vec());
            columnar.set_biggest(b"2000".to_vec());
            meta.add_file(2, FileMeta::from_columnar_table(&columnar));

            let level_snapshot = LegacyIngestLevelSnapshot::from_shard_meta(&meta);
            assert_eq!(level_snapshot.get_ingest_level(table_bound()), level);
        }
    }

    // See issue #572
    // https://github.com/tidbcloud/cloud-storage-engine/issues/572
    #[test]
    fn test_delete_file_with_level() {
        let files = vec![
            // L0:
            (1, FileMeta::new(0, 0, FileType::Sst, b"", b"", 0, 0, None)),
            // L1:
            (
                101,
                FileMeta::new(0, 1, FileType::Sst, b"", b"", 0, 0, None),
            ),
            (
                102,
                FileMeta::new(0, 1, FileType::Sst, b"", b"", 0, 0, None),
            ),
            // L2:
            (
                201,
                FileMeta::new(0, 2, FileType::Sst, b"", b"", 0, 0, None),
            ),
            (
                202,
                FileMeta::new(0, 2, FileType::Sst, b"", b"", 0, 0, None),
            ),
        ];

        // comp_level, top_deletes, bottom_deletes, is_duplicated, result_files
        struct Case(u32, &'static [u64], &'static [u64], bool, &'static [u64]);
        let cases: Vec<Case> = vec![
            // delete top
            Case(0, &[1], &[], false, &[101, 102, 201, 202]),
            Case(1, &[101], &[], false, &[1, 102, 201, 202]),
            // delete bottom
            Case(0, &[], &[102], false, &[1, 101, 201, 202]),
            Case(1, &[], &[201, 202], false, &[1, 101, 102]),
            // delete both top & bottom
            Case(1, &[102], &[202], false, &[1, 101, 201]),
            // top deleted
            Case(0, &[888], &[], true, &[1, 101, 102, 201, 202]),
            Case(1, &[888], &[], true, &[1, 101, 102, 201, 202]),
            // bottom deleted
            Case(1, &[], &[888], true, &[1, 101, 102, 201, 202]),
            // bottom level not match
            Case(1, &[], &[101], true, &[1, 101, 102, 201, 202]),
            // top level not match
            Case(1, &[201], &[202], true, &[1, 101, 102, 201]),
        ];

        for (idx, Case(comp_level, top_deletes, bottom_deletes, is_duplicated, result_files)) in
            cases.into_iter().enumerate()
        {
            let meta = ShardMeta {
                files: HashMap::from_iter(files.clone().into_iter()),
                ..Default::default()
            };

            // Test compaction.
            {
                let mut meta = meta.clone();
                let mut comp = pb::Compaction::default();
                comp.set_level(comp_level);
                comp.mut_top_deletes().extend_from_slice(top_deletes);
                comp.mut_bottom_deletes().extend_from_slice(bottom_deletes);
                comp.mut_table_creates().push(pb::TableCreate {
                    id: 100000,
                    level: comp_level + 1,
                    ..Default::default()
                });

                assert_eq!(
                    meta.is_duplicated_compaction(&mut comp),
                    is_duplicated,
                    "case {}",
                    idx
                );

                meta.apply_compaction(&comp);
                assert_eq!(result_files.len() + 1, meta.files.len());
                for id in result_files {
                    assert!(meta.files.contains_key(id), "case {} file id {}", idx, id);
                }
            }

            // Test other table changes.
            {
                let mut table_change = pb::TableChange::default();
                let table_deletes = table_change.mut_table_deletes();
                for &id in top_deletes {
                    let deleted = pb::TableDelete {
                        id,
                        level: comp_level,
                        ..Default::default()
                    };
                    table_deletes.push(deleted);
                }
                for &id in bottom_deletes {
                    let deleted = pb::TableDelete {
                        id,
                        level: comp_level + 1,
                        ..Default::default()
                    };
                    table_deletes.push(deleted);
                }

                let changesets = vec![
                    {
                        // destroy_range
                        let mut cs = pb::ChangeSet::default();
                        cs.set_destroy_range(table_change.clone());
                        cs.set_property_key(DEL_PREFIXES_KEY.to_string());
                        cs
                    },
                    {
                        // truncate_ts
                        let mut cs = pb::ChangeSet::default();
                        cs.set_truncate_ts(table_change.clone());
                        cs
                    },
                    {
                        // trim_over_bound
                        let mut cs = pb::ChangeSet::default();
                        cs.set_trim_over_bound(table_change.clone());
                        cs
                    },
                ];

                let mut seq = 5; // RAFT_INIT_LOG_INDEX
                for mut cs in changesets {
                    cs.set_sequence(seq);
                    seq += 1;

                    let mut meta = meta.clone();
                    assert_eq!(
                        meta.is_duplicated_change_set(&mut cs),
                        is_duplicated,
                        "case {} cs {:?}",
                        idx,
                        cs,
                    );

                    meta.apply_change_set(&cs);
                    assert_eq!(result_files.len(), meta.files.len());
                    for id in result_files {
                        assert!(
                            meta.files.contains_key(id),
                            "case {} cs {:?} file id {}",
                            idx,
                            cs,
                            id
                        );
                    }
                }
            }
        }
    }

    #[test]
    fn test_table_overlap() {
        let (range, inner_key_off) = (
            ShardRange::new(&[b'x', 2, 3, 4, 10], &[b'x', 2, 3, 4, 20]),
            4,
        );
        let meta = ShardMeta {
            range,
            inner_key_off,
            ..Default::default()
        };

        let cases: Vec<(u8, u8, bool, bool)> = vec![
            // smallest, biggest, overlap, contains
            (3, 5, false, false),
            (3, 10, true, false),
            (3, 15, true, false),
            (3, 20, true, false),
            (3, 25, true, false),
            (10, 15, true, true),
            (10, 20, true, false),
            (10, 25, true, false),
            (13, 15, true, true),
            (13, 20, true, false),
            (13, 25, true, false),
            (20, 25, false, false),
            (23, 25, false, false),
        ];

        for (idx, (smallest, biggest, overlap, contains_bound)) in cases.into_iter().enumerate() {
            let smallest_buf = [smallest];
            let smallest = InnerKey::from_inner_buf(&smallest_buf);
            let biggest_buf = [biggest];
            let biggest = InnerKey::from_inner_buf(&biggest_buf);
            let table_bound = DataBound::new(smallest, biggest, true);
            let shard_bound = meta.range.data_bound();
            assert_eq!(
                shard_bound.overlap_bound(table_bound),
                overlap,
                "case {}",
                idx
            );
            assert_eq!(
                shard_bound.contains_bound(table_bound),
                contains_bound,
                "case {}",
                idx
            );
        }
    }

    #[test]
    fn test_dedup_ingest_files() {
        let make_ingest_files = |l0: &[u64], ln: &[u64], blob: &[u64]| -> pb::IngestFiles {
            let mut ingest_files = pb::IngestFiles::default();
            for &id in l0 {
                ingest_files.mut_l0_creates().push(kvenginepb::L0Create {
                    id,
                    ..Default::default()
                });
            }
            for &id in ln {
                ingest_files
                    .mut_table_creates()
                    .push(kvenginepb::TableCreate {
                        id,
                        ..Default::default()
                    });
            }
            for &id in blob {
                ingest_files
                    .mut_blob_creates()
                    .push(kvenginepb::BlobCreate {
                        id,
                        ..Default::default()
                    });
            }
            ingest_files
        };

        {
            let meta = ShardMeta::default();
            let mut ingest_files = make_ingest_files(&[1], &[2], &[3]);
            let expected = ingest_files.clone();

            let is_empty = meta.dedup_ingest_files_of_load_data(&mut ingest_files);
            assert!(!is_empty);
            assert_eq!(ingest_files, expected);
        }

        {
            let files = (1..=7).map(|id| (id, FileMeta::default()));
            let meta = ShardMeta {
                files: HashMap::from_iter(files),
                ..Default::default()
            };
            let mut ingest_files = make_ingest_files(&[1, 2], &[3, 4], &[5, 6, 7]);
            let is_empty = meta.dedup_ingest_files_of_load_data(&mut ingest_files);
            assert!(is_empty);
            assert_eq!(ingest_files, pb::IngestFiles::default());
        }

        {
            let files = (1..=7).map(|id| (id, FileMeta::default()));
            let meta = ShardMeta {
                files: HashMap::from_iter(files),
                ..Default::default()
            };
            let mut ingest_files =
                make_ingest_files(&[1, 2, 10, 11], &[3, 4, 12], &[5, 6, 7, 13, 14]);
            let is_empty = meta.dedup_ingest_files_of_load_data(&mut ingest_files);
            assert!(!is_empty);
            let expected = make_ingest_files(&[10, 11], &[12], &[13, 14]);
            assert_eq!(ingest_files, expected);
        }
    }

    #[test]
    fn test_check_overlap_for_ingest() {
        let make_smallest_biggest = |t: &(usize, usize)| {
            (
                format!("{:04}", t.0).into_bytes(),
                format!("{:04}", t.1).into_bytes(),
            )
        };
        let make_ingest_files = |ln: &[(usize, usize)]| -> pb::IngestFiles {
            let mut ingest_files = pb::IngestFiles::default();
            for (id, t) in ln.iter().enumerate() {
                let (smallest, biggest) = make_smallest_biggest(t);
                ingest_files
                    .mut_table_creates()
                    .push(kvenginepb::TableCreate {
                        id: id as u64,
                        level: WRITE_CF_BOTTOM_LEVEL,
                        smallest,
                        biggest,
                        ..Default::default()
                    });
            }
            ingest_files
        };
        let make_meta = |files: &[(usize, usize)]| -> ShardMeta {
            let mut meta = ShardMeta::default();
            for (id, t) in files.iter().enumerate() {
                let (smallest, biggest) = make_smallest_biggest(t);
                meta.files.insert(
                    id as u64,
                    FileMeta::new(
                        WRITE_CF as i32,
                        3,
                        FileType::Sst,
                        &smallest,
                        &biggest,
                        0,
                        0,
                        None,
                    ),
                );
            }
            meta
        };

        let ingest_files = make_ingest_files(&[(1, 1), (2, 5), (10, 20)]);
        let cases = vec![
            (
                vec![(0, 0)], // existed files
                None,         // expected
            ),
            (vec![(0, 0), (1, 2)], Some((1, 0))),
            (vec![(0, 0), (6, 6), (8, 11)], Some((2, 2))),
            (vec![(0, 0), (6, 6), (7, 7), (30, 40)], None),
        ];

        for (idx, (existed_files, expected)) in cases.into_iter().enumerate() {
            let meta = make_meta(&existed_files);
            let overlap = meta.check_overlap_for_ingest(&ingest_files);
            assert_eq!(overlap, expected, "case {}", idx);
        }

        // Columnar file may share same cf/level metadata fields in `files`, but it
        // should not participate in SST load-data overlap checks.
        let mut meta = ShardMeta::default();
        let mut col = kvenginepb::ColumnarCreate::new();
        col.set_id(100);
        col.set_level(WRITE_CF_BOTTOM_LEVEL);
        let (smallest, biggest) = make_smallest_biggest(&(1, 1));
        col.set_smallest(smallest);
        col.set_biggest(biggest);
        meta.add_file(100, FileMeta::from_columnar_table(&col));
        assert_eq!(meta.check_overlap_for_ingest(&ingest_files), None);
    }

    #[test]
    fn test_merge_vector_index() {
        let make_vector_index_file =
            |id: u64, smallest: &[u8], biggest: &[u8]| -> kvenginepb::VectorIndexFile {
                let mut file = kvenginepb::VectorIndexFile::new();
                file.set_id(id);
                file.set_smallest(smallest.to_vec());
                file.set_biggest(biggest.to_vec());
                file.set_meta_offset(100);
                file
            };

        let make_vector_index = |table_id: i64,
                                 index_id: i64,
                                 col_id: i64,
                                 files: Vec<kvenginepb::VectorIndexFile>|
         -> VectorIndex {
            let mut vec_idx = VectorIndex::new();
            vec_idx.set_table_id(table_id);
            vec_idx.set_index_id(index_id);
            vec_idx.set_col_id(col_id);
            vec_idx.set_files(files.into());
            vec_idx
        };

        // Test case 1: Merge files into existing vector index
        {
            let mut meta = ShardMeta::default();

            // Add initial vector index with files [1, 2]
            let initial_files = vec![
                make_vector_index_file(1, b"key1", b"key2"),
                make_vector_index_file(2, b"key3", b"key4"),
            ];
            let initial_vec_idx = make_vector_index(100, 1, 10, initial_files);
            meta.vector_indexes.push(initial_vec_idx);

            // Merge new files [2, 3] (file 2 already exists, file 3 is new)
            let new_files = vec![
                make_vector_index_file(2, b"key3", b"key4"), // duplicate
                make_vector_index_file(3, b"key5", b"key6"), // new
            ];
            let new_vec_idx = make_vector_index(100, 1, 10, new_files);

            meta.merge_vector_index(&new_vec_idx);

            // Verify: should have 1 vector index with files [1, 2, 3]
            assert_eq!(meta.vector_indexes.len(), 1);
            let merged_idx = &meta.vector_indexes[0];
            assert_eq!(merged_idx.table_id, 100);
            assert_eq!(merged_idx.index_id, 1);
            assert_eq!(merged_idx.col_id, 10);
            assert_eq!(merged_idx.files.len(), 3);

            let file_ids: std::collections::HashSet<u64> =
                merged_idx.files.iter().map(|f| f.id).collect();
            assert!(file_ids.contains(&1));
            assert!(file_ids.contains(&2));
            assert!(file_ids.contains(&3));
        }

        // Test case 2: Add new vector index (no matching existing index)
        {
            let mut meta = ShardMeta::default();

            // Add initial vector index
            let initial_files = vec![make_vector_index_file(1, b"key1", b"key2")];
            let initial_vec_idx = make_vector_index(100, 1, 10, initial_files);
            meta.vector_indexes.push(initial_vec_idx);

            // Add new vector index with different table_id
            let new_files = vec![make_vector_index_file(4, b"key7", b"key8")];
            let new_vec_idx = make_vector_index(200, 1, 10, new_files);

            meta.merge_vector_index(&new_vec_idx);

            // Verify: should have 2 vector indexes
            assert_eq!(meta.vector_indexes.len(), 2);

            // Find the new index
            let new_idx = meta
                .vector_indexes
                .iter()
                .find(|idx| idx.table_id == 200)
                .unwrap();
            assert_eq!(new_idx.index_id, 1);
            assert_eq!(new_idx.col_id, 10);
            assert_eq!(new_idx.files.len(), 1);
            assert_eq!(new_idx.files[0].id, 4);
        }

        // Test case 3: Add new vector index with different index_id
        {
            let mut meta = ShardMeta::default();

            // Add initial vector index
            let initial_files = vec![make_vector_index_file(1, b"key1", b"key2")];
            let initial_vec_idx = make_vector_index(100, 1, 10, initial_files);
            meta.vector_indexes.push(initial_vec_idx);

            // Add new vector index with different index_id
            let new_files = vec![make_vector_index_file(5, b"key9", b"key10")];
            let new_vec_idx = make_vector_index(100, 2, 10, new_files);

            meta.merge_vector_index(&new_vec_idx);

            // Verify: should have 2 vector indexes
            assert_eq!(meta.vector_indexes.len(), 2);

            // Find the new index
            let new_idx = meta
                .vector_indexes
                .iter()
                .find(|idx| idx.index_id == 2)
                .unwrap();
            assert_eq!(new_idx.table_id, 100);
            assert_eq!(new_idx.col_id, 10);
            assert_eq!(new_idx.files.len(), 1);
            assert_eq!(new_idx.files[0].id, 5);
        }

        // Test case 4: Add new vector index with different col_id
        {
            let mut meta = ShardMeta::default();

            // Add initial vector index
            let initial_files = vec![make_vector_index_file(1, b"key1", b"key2")];
            let initial_vec_idx = make_vector_index(100, 1, 10, initial_files);
            meta.vector_indexes.push(initial_vec_idx);

            // Add new vector index with different col_id
            let new_files = vec![make_vector_index_file(6, b"key11", b"key12")];
            let new_vec_idx = make_vector_index(100, 1, 20, new_files);

            meta.merge_vector_index(&new_vec_idx);

            // Verify: should have 2 vector indexes
            assert_eq!(meta.vector_indexes.len(), 2);

            // Find the new index
            let new_idx = meta
                .vector_indexes
                .iter()
                .find(|idx| idx.col_id == 20)
                .unwrap();
            assert_eq!(new_idx.table_id, 100);
            assert_eq!(new_idx.index_id, 1);
            assert_eq!(new_idx.files.len(), 1);
            assert_eq!(new_idx.files[0].id, 6);
        }
    }

    #[test]
    fn test_merge_vector_index_snap_version() {
        let make_vector_index = |table_id: i64,
                                 index_id: i64,
                                 col_id: i64,
                                 snap_version: u64,
                                 files: Vec<kvenginepb::VectorIndexFile>|
         -> VectorIndex {
            let mut vec_idx = VectorIndex::new();
            vec_idx.set_table_id(table_id);
            vec_idx.set_index_id(index_id);
            vec_idx.set_col_id(col_id);
            if snap_version != 0 {
                vec_idx.set_snap_version(snap_version);
            }
            vec_idx.set_files(files.into());
            vec_idx
        };

        let mut meta = ShardMeta::default();
        meta.merge_vector_index(&make_vector_index(1, 2, 3, 10, vec![]));
        meta.merge_vector_index(&make_vector_index(1, 2, 3, 42, vec![]));
        assert_eq!(meta.vector_indexes.len(), 1);
        assert_eq!(meta.vector_indexes[0].get_snap_version(), 42);
    }

    #[test]
    fn test_apply_update_vector_index_delete_marker() {
        let make_vec_idx_file = |id: u64| -> kvenginepb::VectorIndexFile {
            let mut file = kvenginepb::VectorIndexFile::new();
            file.set_id(id);
            file.set_smallest(b"key1".to_vec());
            file.set_biggest(b"key2".to_vec());
            file.set_meta_offset(100);
            file
        };

        let mut meta = ShardMeta::default();
        let mut vec_idx = kvenginepb::VectorIndex::new();
        vec_idx.set_table_id(1);
        vec_idx.set_index_id(2);
        vec_idx.set_col_id(3);
        vec_idx.set_snap_version(42);
        vec_idx.mut_files().push(make_vec_idx_file(10));
        vec_idx.mut_files().push(make_vec_idx_file(11));
        meta.vector_indexes.push(vec_idx);
        meta.add_file(10, FileMeta::from_vector_index_file(&make_vec_idx_file(10)));
        meta.add_file(11, FileMeta::from_vector_index_file(&make_vec_idx_file(11)));

        let mut update = pb::UpdateVectorIndex::new();
        update.set_table_id(1);
        update.set_index_id(2);
        update.set_col_id(3);
        update.set_removed(vec![10, 11]);
        update.set_snap_version(0);
        meta.apply_update_vector_index(&update);

        assert!(meta.vector_indexes.is_empty());
        assert!(!meta.files.contains_key(&10));
        assert!(!meta.files.contains_key(&11));
    }

    #[test]
    fn test_apply_update_vector_index_keep_empty_watermark() {
        let mut meta = ShardMeta::default();
        let mut update = pb::UpdateVectorIndex::new();
        update.set_table_id(1);
        update.set_index_id(2);
        update.set_col_id(3);
        update.set_snap_version(42);
        meta.apply_update_vector_index(&update);

        assert_eq!(meta.vector_indexes.len(), 1);
        assert_eq!(meta.vector_indexes[0].get_snap_version(), 42);
        assert!(meta.vector_indexes[0].files.is_empty());

        // Delete marker should remove the empty watermark.
        let mut delete = pb::UpdateVectorIndex::new();
        delete.set_table_id(1);
        delete.set_index_id(2);
        delete.set_col_id(3);
        delete.set_snap_version(0);
        meta.apply_update_vector_index(&delete);
        assert!(meta.vector_indexes.is_empty());
    }

    #[test]
    fn test_fts_meta_merge() {
        ::test_util::init_log_for_test();

        let keyspace_id = 1;
        let table_id = 70;
        let mut target = new_shard_meta_for_range(keyspace_id, table_id, 0, 50, 20);
        target.fts_l0_snap_version = SnapVersion::from(5);
        target.fts_indexes.insert((table_id, 1));
        target
            .fts_l0_files
            .push(new_fts_packed_info(8000, table_id, 0, 40, Some(5)));
        target
            .fts_l1_files
            .push(new_fts_packed_info(8100, table_id, 0, 40, None));
        target
            .fts_l2_files
            .push(new_fts_ded_info(8200, table_id, 0, 40));

        let mut source = new_shard_meta_for_range(keyspace_id, table_id, 50, 100, 30);
        source.fts_l0_snap_version = SnapVersion::from(9);
        source.fts_indexes.insert((table_id, 2));
        source.fts_indexes.insert((table_id + 1, 1));
        let incoming_l0_id = 8300;
        source.fts_l0_files.push(new_fts_packed_info(
            incoming_l0_id,
            table_id,
            60,
            90,
            Some(9),
        ));
        source
            .fts_l1_files
            .push(new_fts_packed_info(8400, table_id, 60, 95, None));
        source
            .fts_l2_files
            .push(new_fts_ded_info(8500, table_id, 60, 95));
        source
            .fts_l2_files
            .push(new_fts_ded_info(8600, table_id + 1, 60, 95));

        target.commit_merge(&source, 500);

        assert!(target.fts_indexes.contains(&(table_id, 1)));
        assert!(target.fts_indexes.contains(&(table_id, 2)));
        assert!(
            !target
                .fts_indexes
                .iter()
                .any(|(tid, _)| *tid == table_id + 1),
            "indexes from unrelated tables should be dropped"
        );

        assert!(
            target
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == incoming_l0_id),
            "merged shard should include new L0 file"
        );
        assert!(
            target.fts_l1_files.iter().any(|info| info.get_id() == 8400),
            "merged shard should include new L1 file"
        );
        assert!(
            target.fts_l2_files.iter().any(|info| info.get_id() == 8500),
            "merged shard should include new L2 file"
        );
        assert!(
            target.fts_l2_files.iter().all(|info| {
                let data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&info.smallest),
                    InnerKey::from_inner_buf(&info.biggest),
                    true,
                );
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
                table_id >= min_table_id && table_id <= max_table_id
            }),
            "L2 files belonging to other tables should be removed"
        );
        assert_eq!(target.fts_l0_snap_version, source.fts_l0_snap_version);
    }

    #[test]
    fn test_fts_meta_split() {
        ::test_util::init_log_for_test();

        let keyspace_id = 1;
        let table_id = 70;
        let mut parent = new_shard_meta_for_range(keyspace_id, table_id, 0, 100, 20);
        parent.columnar_table_ids.push(table_id + 1);
        parent.fts_l0_snap_version = SnapVersion::from(11);
        parent.fts_indexes.insert((table_id, 1));
        parent.fts_indexes.insert((table_id + 1, 2));

        let left_l0_id = 9000;
        let right_l0_id = 9001;
        let other_l0_id = 9002;
        parent
            .fts_l0_files
            .push(new_fts_packed_info(left_l0_id, table_id, 0, 40, Some(11)));
        parent
            .fts_l0_files
            .push(new_fts_packed_info(right_l0_id, table_id, 60, 95, Some(11)));
        parent.fts_l0_files.push(new_fts_packed_info(
            other_l0_id,
            table_id + 1,
            0,
            40,
            Some(11),
        ));

        let left_l1_id = 9100;
        let right_l1_id = 9101;
        let other_l1_id = 9102;
        parent
            .fts_l1_files
            .push(new_fts_packed_info(left_l1_id, table_id, 0, 40, None));
        parent
            .fts_l1_files
            .push(new_fts_packed_info(right_l1_id, table_id, 60, 95, None));
        parent
            .fts_l1_files
            .push(new_fts_packed_info(other_l1_id, table_id + 1, 0, 40, None));

        let left_l2_id = 9200;
        let right_l2_id = 9201;
        let other_l2_id = 9202;
        parent
            .fts_l2_files
            .push(new_fts_ded_info(left_l2_id, table_id, 0, 40));
        parent
            .fts_l2_files
            .push(new_fts_ded_info(right_l2_id, table_id, 60, 95));
        parent
            .fts_l2_files
            .push(new_fts_ded_info(other_l2_id, table_id + 1, 0, 40));

        let split_key = outer_row_key(keyspace_id, table_id, 50);
        let mut split = kvenginepb::Split::new();
        split.mut_keys().push(split_key.clone());
        let mut left_props = kvenginepb::Properties::new();
        left_props.set_shard_id(parent.id + 1);
        split.mut_new_shards().push(left_props);
        let mut right_props = kvenginepb::Properties::new();
        right_props.set_shard_id(parent.id);
        split.mut_new_shards().push(right_props);

        let new_shards = parent.apply_split(&split, 500, 400);
        assert_eq!(new_shards.len(), 2);

        let mut left = None;
        let mut right = None;
        let parent_start = parent.range.outer_start.clone();
        for shard in &new_shards {
            if shard.range.outer_start.as_ref() == parent_start.as_ref() {
                left = Some(shard);
            } else if shard.range.outer_start.as_ref() == split_key.as_slice() {
                right = Some(shard);
            }
        }
        let left = left.expect("left shard should exist");
        let right = right.expect("right shard should exist");

        assert_eq!(left.fts_l0_snap_version, parent.fts_l0_snap_version);
        assert_eq!(right.fts_l0_snap_version, parent.fts_l0_snap_version);

        assert!(left.fts_indexes.contains(&(table_id, 1)));
        assert!(right.fts_indexes.contains(&(table_id, 1)));
        assert!(left.fts_indexes.iter().all(|(tid, _)| *tid == table_id));
        assert!(right.fts_indexes.iter().all(|(tid, _)| *tid == table_id));

        assert!(
            left.fts_l0_files
                .iter()
                .any(|info| info.get_id() == left_l0_id),
            "left shard should keep overlapping L0 file"
        );
        assert!(
            !left
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == right_l0_id),
            "left shard should drop non-overlapping L0 file"
        );
        assert!(
            !left
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == other_l0_id),
            "left shard should drop files from other tables"
        );
        assert!(
            right
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == right_l0_id),
            "right shard should keep overlapping L0 file"
        );
        assert!(
            !right
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == left_l0_id),
            "right shard should drop non-overlapping L0 file"
        );
        assert!(
            !right
                .fts_l0_files
                .iter()
                .any(|info| info.get_id() == other_l0_id),
            "right shard should drop files from other tables"
        );

        assert!(
            left.fts_l1_files
                .iter()
                .any(|info| info.get_id() == left_l1_id),
            "left shard should keep overlapping L1 file"
        );
        assert!(
            !left
                .fts_l1_files
                .iter()
                .any(|info| info.get_id() == right_l1_id),
            "left shard should drop non-overlapping L1 file"
        );
        assert!(
            !left
                .fts_l1_files
                .iter()
                .any(|info| info.get_id() == other_l1_id),
            "left shard should drop files from other tables"
        );
        assert!(
            right
                .fts_l1_files
                .iter()
                .any(|info| info.get_id() == right_l1_id),
            "right shard should keep overlapping L1 file"
        );
        assert!(
            !right
                .fts_l1_files
                .iter()
                .any(|info| info.get_id() == left_l1_id),
            "right shard should drop non-overlapping L1 file"
        );
        assert!(
            !right
                .fts_l1_files
                .iter()
                .any(|info| info.get_id() == other_l1_id),
            "right shard should drop files from other tables"
        );

        assert!(
            left.fts_l2_files
                .iter()
                .any(|info| info.get_id() == left_l2_id),
            "left shard should keep overlapping L2 file"
        );
        assert!(
            !left
                .fts_l2_files
                .iter()
                .any(|info| info.get_id() == right_l2_id),
            "left shard should drop non-overlapping L2 file"
        );
        assert!(
            !left
                .fts_l2_files
                .iter()
                .any(|info| info.get_id() == other_l2_id),
            "left shard should drop files from other tables"
        );
        assert!(
            right
                .fts_l2_files
                .iter()
                .any(|info| info.get_id() == right_l2_id),
            "right shard should keep overlapping L2 file"
        );
        assert!(
            !right
                .fts_l2_files
                .iter()
                .any(|info| info.get_id() == left_l2_id),
            "right shard should drop non-overlapping L2 file"
        );
        assert!(
            !right
                .fts_l2_files
                .iter()
                .any(|info| info.get_id() == other_l2_id),
            "right shard should drop files from other tables"
        );

        assert!(
            left.fts_l0_files.iter().all(|info| {
                let data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&info.smallest),
                    InnerKey::from_inner_buf(&info.biggest),
                    true,
                );
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
                table_id >= min_table_id && table_id <= max_table_id
            }),
            "left shard should only keep files from target table"
        );
        assert!(
            right.fts_l0_files.iter().all(|info| {
                let data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&info.smallest),
                    InnerKey::from_inner_buf(&info.biggest),
                    true,
                );
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
                table_id >= min_table_id && table_id <= max_table_id
            }),
            "right shard should only keep files from target table"
        );
        assert!(
            left.fts_l2_files.iter().all(|info| {
                let data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&info.smallest),
                    InnerKey::from_inner_buf(&info.biggest),
                    true,
                );
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
                table_id >= min_table_id && table_id <= max_table_id
            }),
            "left shard should only keep files from target table"
        );
        assert!(
            right.fts_l2_files.iter().all(|info| {
                let data_bound = DataBound::new(
                    InnerKey::from_inner_buf(&info.smallest),
                    InnerKey::from_inner_buf(&info.biggest),
                    true,
                );
                let (min_table_id, max_table_id) = get_table_id_from_data_bound(data_bound);
                table_id >= min_table_id && table_id <= max_table_id
            }),
            "right shard should only keep files from target table"
        );
    }

    #[test]
    fn test_meta_split_preserves_unconverted_l0s_when_schema_invalid_but_columnar_active() {
        ::test_util::init_log_for_test();

        let keyspace_id = 1;
        let table_id = 70;
        let mut parent = new_shard_meta_for_range(keyspace_id, table_id, 0, 100, 20);
        parent.schema.clear(0);
        assert!(
            !parent.schema.is_valid(),
            "test requires invalid schema meta to emulate schema-file gap"
        );
        parent.columnar_table_ids.push(table_id + 1);

        let left_l0_id = 9000;
        let right_l0_id = 9001;
        let untracked_l0_id = 9002;
        parent.unconverted_l0s = vec![left_l0_id, right_l0_id];

        let left_smallest = outer_row_key(keyspace_id, table_id, 0);
        let left_biggest = outer_row_key(keyspace_id, table_id, 40);
        let mut left_l0 = kvenginepb::L0Create::new();
        left_l0.set_id(left_l0_id);
        left_l0.set_smallest(left_smallest.clone());
        left_l0.set_biggest(left_biggest.clone());
        left_l0.set_size(1024);
        parent.add_file(left_l0_id, FileMeta::from_l0_table(&left_l0));

        let right_smallest = outer_row_key(keyspace_id, table_id, 60);
        let right_biggest = outer_row_key(keyspace_id, table_id, 95);
        let mut right_l0 = kvenginepb::L0Create::new();
        right_l0.set_id(right_l0_id);
        right_l0.set_smallest(right_smallest.clone());
        right_l0.set_biggest(right_biggest.clone());
        right_l0.set_size(1024);
        parent.add_file(right_l0_id, FileMeta::from_l0_table(&right_l0));

        let untracked_smallest = outer_row_key(keyspace_id, table_id, 10);
        let untracked_biggest = outer_row_key(keyspace_id, table_id, 20);
        let mut untracked_l0 = kvenginepb::L0Create::new();
        untracked_l0.set_id(untracked_l0_id);
        untracked_l0.set_smallest(untracked_smallest.clone());
        untracked_l0.set_biggest(untracked_biggest.clone());
        untracked_l0.set_size(1024);
        parent.add_file(untracked_l0_id, FileMeta::from_l0_table(&untracked_l0));

        let split_key = outer_row_key(keyspace_id, table_id, 50);
        let mut split = kvenginepb::Split::new();
        split.mut_keys().push(split_key.clone());
        let mut left_props = kvenginepb::Properties::new();
        left_props.set_shard_id(parent.id + 1);
        split.mut_new_shards().push(left_props);
        let mut right_props = kvenginepb::Properties::new();
        right_props.set_shard_id(parent.id);
        split.mut_new_shards().push(right_props);

        let new_shards = parent.apply_split(&split, 500, 400);
        assert_eq!(new_shards.len(), 2);

        let mut left = None;
        let mut right = None;
        let parent_start = parent.range.outer_start.clone();
        for shard in &new_shards {
            if shard.range.outer_start.as_ref() == parent_start.as_ref() {
                left = Some(shard);
            } else if shard.range.outer_start.as_ref() == split_key.as_slice() {
                right = Some(shard);
            }
        }
        let left = left.expect("left shard should exist");
        let right = right.expect("right shard should exist");

        assert!(!left.schema.is_valid());
        assert!(!right.schema.is_valid());
        assert!(!left.columnar_table_ids.is_empty());
        assert!(!right.columnar_table_ids.is_empty());
        assert_eq!(left.unconverted_l0s, vec![left_l0_id]);
        assert_eq!(right.unconverted_l0s, vec![right_l0_id]);
    }

    fn new_shard_meta_for_range(
        keyspace_id: u32,
        table_id: i64,
        start_handle: i64,
        end_handle: i64,
        shard_id: u64,
    ) -> ShardMeta {
        let outer_start = outer_row_key(keyspace_id, table_id, start_handle);
        let outer_end = outer_row_key(keyspace_id, table_id, end_handle);
        let mut meta = ShardMeta::default();
        meta.engine_id = 1;
        meta.id = shard_id;
        meta.ver = 1;
        meta.seq = 1;
        meta.range = ShardRange::new(&outer_start, &outer_end);
        meta.columnar_table_ids = vec![table_id];
        meta
    }

    fn new_fts_packed_info(
        id: u64,
        table_id: i64,
        smallest_handle: i64,
        biggest_handle: i64,
        snap_version: Option<u64>,
    ) -> kvenginepb::FtsPackedFileInfo {
        let mut info = kvenginepb::FtsPackedFileInfo::new();
        info.set_id(id);
        info.set_smallest(encode_row_key(table_id, smallest_handle));
        info.set_biggest(encode_row_key(table_id, biggest_handle));
        if let Some(snap_version) = snap_version {
            info.set_snap_version(snap_version);
        }
        info
    }

    fn new_fts_ded_info(
        id: u64,
        table_id: i64,
        smallest_handle: i64,
        biggest_handle: i64,
    ) -> kvenginepb::FtsDedFileInfo {
        let mut info = kvenginepb::FtsDedFileInfo::new();
        info.set_id(id);
        info.set_smallest(encode_row_key(table_id, smallest_handle));
        info.set_biggest(encode_row_key(table_id, biggest_handle));
        info
    }

    fn outer_row_key(keyspace_id: u32, table_id: i64, handle: i64) -> Vec<u8> {
        let mut key = ApiV2::get_txn_keyspace_prefix(keyspace_id);
        key.extend_from_slice(&encode_row_key(table_id, handle));
        key
    }

    #[test]
    fn test_check_overlap_for_compaction() {
        use kvenginepb::{
            ChangeSet, ColumnarCreate, Compaction, MajorCompaction, TableChange, TableCreate,
            TableDelete,
        };

        // Helper to create a ChangeSet with MajorCompaction
        fn make_major_compaction_cs(
            deletes: &[(u64, u8, u32)],
            creates: &[(u64, u8, u32, &[u8], &[u8])],
        ) -> ChangeSet {
            let mut cs = ChangeSet::new();
            let mut comp = MajorCompaction::new();
            let mut table_change = TableChange::new();

            for &(id, cf, level) in deletes {
                let mut del = TableDelete::new();
                del.set_id(id);
                del.set_cf(cf as i32);
                del.set_level(level);
                table_change.mut_table_deletes().push(del);
            }

            for &(id, cf, level, smallest, biggest) in creates {
                let mut create = TableCreate::new();
                create.set_id(id);
                create.set_cf(cf as i32);
                create.set_level(level);
                create.set_smallest(smallest.to_vec());
                create.set_biggest(biggest.to_vec());
                table_change.mut_table_creates().push(create);
            }

            comp.set_sstable_change(table_change);
            cs.set_major_compaction(comp);
            cs
        }

        // Helper to create a ChangeSet with non-major Compaction (L0/L1Plus)
        fn make_non_major_compaction_cs(
            cf: i32,
            level: u32,
            top_deletes: &[u64],
            bottom_deletes: &[u64],
            creates: &[(u64, &[u8], &[u8])],
        ) -> ChangeSet {
            let mut cs = ChangeSet::new();
            let mut comp = Compaction::new();
            comp.set_cf(cf);
            comp.set_level(level);
            comp.mut_top_deletes().extend_from_slice(top_deletes);
            comp.mut_bottom_deletes().extend_from_slice(bottom_deletes);

            for &(id, smallest, biggest) in creates {
                let mut create = TableCreate::new();
                create.set_id(id);
                create.set_cf(cf);
                create.set_level(level);
                create.set_smallest(smallest.to_vec());
                create.set_biggest(biggest.to_vec());
                comp.mut_table_creates().push(create);
            }

            cs.set_compaction(comp);
            cs
        }

        // ===== Major Compaction Tests =====

        // Major Case 0: No sstable_change - should return None immediately
        let meta = make_meta_with_files(&[(1, 0, 3, b"key000", b"key049")]);
        let mut cs_no_change = ChangeSet::new();
        cs_no_change.set_major_compaction(MajorCompaction::new());
        assert_eq!(
            meta.check_overlap_for_compaction(&cs_no_change),
            None,
            "Major Case 0: Should return None when has_sstable_change is false"
        );

        // Major Case 1: No overlap when output only overlaps with files being deleted
        // L3 has: File 1 [0,50), File 2 [100,150)
        // Compaction: delete [1,2], create File 10 [0,150) on L3
        // Expected: No overlap (File 10 replaces File 1 and 2, which is normal)
        let meta = make_meta_with_files(&[
            (1, 0, 3, b"key000", b"key049"), // File 1: [0,50) on L3
            (2, 0, 3, b"key100", b"key149"), // File 2: [100,150) on L3
        ]);
        let cs = make_major_compaction_cs(
            &[(1, 0, 3), (2, 0, 3)],             // delete File 1, 2
            &[(10, 0, 3, b"key000", b"key149")], // create File 10 [0,150)
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Major Case 1: Should not report overlap when output only overlaps with deleted files"
        );

        // Major Case 2: Detect overlap when output overlaps with a file NOT being
        // deleted L3 has: File 1 [0,50), File 2 [100,150), File 3 [50,100)
        // (newly ingested) Compaction: delete [1,2], create File 10 [0,150) on
        // L3 Expected: Overlap detected between File 10 and File 3
        let meta = make_meta_with_files(&[
            (1, 0, 3, b"key000", b"key049"), // File 1: [0,50) on L3
            (2, 0, 3, b"key100", b"key149"), // File 2: [100,150) on L3
            (3, 0, 3, b"key050", b"key099"), // File 3: [50,100) on L3 (not deleted)
        ]);
        let cs = make_major_compaction_cs(
            &[(1, 0, 3), (2, 0, 3)],             // delete File 1, 2 only
            &[(10, 0, 3, b"key000", b"key149")], // create File 10 [0,150)
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            Some((3, 10)), // File 3 overlaps with File 10
            "Major Case 2: Should detect overlap with files not being deleted"
        );

        // Major Case 3: No overlap when files are on different levels
        // L2 has: File 1 [0,100)
        // L3 has: File 2 [0,50)
        // Compaction: delete File 2, create File 10 [0,100) on L3
        // Expected: No overlap (File 1 is on L2, different from output level L3)
        let meta = make_meta_with_files(&[
            (1, 0, 2, b"key000", b"key099"), // File 1 on L2
            (2, 0, 3, b"key000", b"key049"), // File 2 on L3
        ]);
        let cs = make_major_compaction_cs(
            &[(2, 0, 3)],                        // delete File 2
            &[(10, 0, 3, b"key000", b"key099")], // create File 10 on L3
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Major Case 3: Should not report overlap with files on different levels"
        );

        // Major Case 4: No overlap when files are on different CFs
        // CF 0, L3: File 1 [0,50)
        // CF 1, L3: File 2 [0,100)
        // Compaction: delete File 1, create File 10 [0,100) on CF 0, L3
        // Expected: No overlap (File 2 is on different CF)
        let meta = make_meta_with_files(&[
            (1, 0, 3, b"key000", b"key049"), // CF 0
            (2, 1, 3, b"key000", b"key099"), // CF 1
        ]);
        let cs = make_major_compaction_cs(
            &[(1, 0, 3)],                        // delete File 1 on CF 0
            &[(10, 0, 3, b"key000", b"key099")], // create File 10 on CF 0
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Major Case 4: Should not report overlap with files on different CFs"
        );

        // Major Case 5: Multiple output files, one has overlap
        // L3 has: File 1 [0,50), File 2 [100,150), File 3 [200,250)
        // Compaction: delete [1,2], create File 10 [0,50), File 11 [100,250) on L3
        // Expected: Overlap between File 11 and File 3
        let meta = make_meta_with_files(&[
            (1, 0, 3, b"key000", b"key049"),
            (2, 0, 3, b"key100", b"key149"),
            (3, 0, 3, b"key200", b"key249"), // Not deleted
        ]);
        let cs = make_major_compaction_cs(
            &[(1, 0, 3), (2, 0, 3)],
            &[
                (10, 0, 3, b"key000", b"key049"),
                (11, 0, 3, b"key100", b"key249"), // Overlaps with File 3
            ],
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            Some((3, 11)),
            "Major Case 5: Should detect overlap in one of multiple output files"
        );

        // Major Case 6: Ignore overlap with columnar files at the same cf/level.
        let mut meta = make_meta_with_files(&[(1, 0, 1, b"key000", b"key049")]);
        let mut col = ColumnarCreate::new();
        col.set_id(100);
        col.set_level(1);
        col.set_smallest(b"key050".to_vec());
        col.set_biggest(b"key099".to_vec());
        meta.add_file(100, FileMeta::from_columnar_table(&col));
        let cs = make_major_compaction_cs(
            &[(1, 0, 1)],
            &[(10, 0, 1, b"key000", b"key099")], // Overlaps with columnar file 100.
        );
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Major Case 6: Should ignore overlap with columnar files"
        );

        // ===== Non-Major Compaction (L0/L1Plus) Tests =====

        // Non-Major Case 1: No overlap when output only overlaps with files being
        // deleted L1: File 1 [0,50), L2: File 2 [0,100), File 3 [100,150)
        // L1Plus (L1→L2): delete [1,2], create File 10 [0,100) on L2
        // Expected: No overlap (File 10 replaces File 1+2)
        let meta = make_meta_with_files(&[
            (1, 0, 1, b"key000", b"key049"),
            (2, 0, 2, b"key000", b"key099"),
            (3, 0, 2, b"key100", b"key149"),
        ]);
        let cs = make_non_major_compaction_cs(0, 2, &[1], &[2], &[(10, b"key000", b"key099")]);
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Non-Major Case 1: No overlap when output only overlaps with deleted files"
        );

        // Non-Major Case 2: Detect overlap with newly ingested file
        // L1: File 1 [0,50), L2: File 2 [0,50), File 3 [50,100) (ingested)
        // L1Plus (L1→L2): delete [1,2], create File 10 [0,100) on L2
        // Expected: Overlap File 10 ∩ File 3
        let meta = make_meta_with_files(&[
            (1, 0, 1, b"key000", b"key049"),
            (2, 0, 2, b"key000", b"key049"),
            (3, 0, 2, b"key050", b"key099"),
        ]);
        let cs = make_non_major_compaction_cs(0, 2, &[1], &[2], &[(10, b"key000", b"key099")]);
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            Some((3, 10)),
            "Non-Major Case 2: Should detect overlap with ingested file"
        );

        // Non-Major Case 3: No overlap when files are on different levels
        let meta = make_meta_with_files(&[
            (1, 0, 1, b"key000", b"key099"),
            (2, 0, 2, b"key000", b"key049"),
            (3, 0, 3, b"key000", b"key099"),
        ]);
        let cs = make_non_major_compaction_cs(0, 2, &[1], &[2], &[(10, b"key000", b"key099")]);
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Non-Major Case 3: No overlap when existing file is on different level"
        );

        // Non-Major Case 4: L0 compaction with overlap
        // L0: File 1 [0,100), L1: File 2 [0,50), File 3 [50,100) (ingested)
        // L0 (L0→L1): delete L0 [1] and L1 [2], create File 10 [0,100) on L1
        // Expected: Overlap with File 3
        let meta = make_meta_with_files(&[
            (1, 0, 0, b"key000", b"key099"), // L0
            (2, 0, 1, b"key000", b"key049"), // L1
            (3, 0, 1, b"key050", b"key099"), // L1 (ingested)
        ]);
        let cs = make_non_major_compaction_cs(0, 1, &[1], &[2], &[(10, b"key000", b"key099")]);
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            Some((3, 10)),
            "Non-Major Case 4: Should detect overlap for L0 compaction"
        );

        // Non-Major Case 5: Ignore overlap with columnar files at output level.
        let mut meta = make_meta_with_files(&[
            (1, 0, 0, b"key000", b"key099"), // L0 top
            (2, 0, 1, b"key000", b"key049"), // L1 bottom to be deleted
        ]);
        let mut col = ColumnarCreate::new();
        col.set_id(101);
        col.set_level(1);
        col.set_smallest(b"key050".to_vec());
        col.set_biggest(b"key099".to_vec());
        meta.add_file(101, FileMeta::from_columnar_table(&col));
        let cs = make_non_major_compaction_cs(0, 1, &[1], &[2], &[(10, b"key000", b"key099")]);
        assert_eq!(
            meta.check_overlap_for_compaction(&cs),
            None,
            "Non-Major Case 5: Should ignore overlap with columnar files"
        );
    }
}