// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.

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

use api_version::{
    api_v2::{is_whole_keyspace_range, KEYSPACE_PREFIX_LEN},
    ApiV2,
};
use bytes::{Buf, Bytes};
use kvenginepb as pb;
use kvenginepb::{SchemaMeta, TxnFileRef, VectorIndex};
use protobuf::Message;
use slog_global::*;
use util::TxnFileRefExt as _;

use super::*;
use crate::{
    dfs::FileType,
    table::{BoundedDataSet, DataBound, InnerKey},
    util::{TxnFileLocks, TxnFileRefPropertyHelper},
};

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

    /// `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_file_id: u64,
    pub schema_file_ver: i64,
    pub schema_restore_ver: u64,
    pub columnar_snap_version: u64,
    pub unconverted_l0s: Vec<u64>,
    pub vector_indexes: Vec<kvenginepb::VectorIndex>,

    /// 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`.
    pub(crate) txn_file_locks: TxnFileLocks,
}

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 = if cs.has_snapshot() {
            cs.get_snapshot()
        } else if cs.has_initial_flush() {
            cs.get_initial_flush()
        } else if cs.has_restore_shard() {
            cs.get_restore_shard()
        } else {
            unreachable!();
        };

        let properties = Properties::new().apply_pb(snap.get_properties());
        let txn_file_locks = TxnFileRefPropertyHelper::from_property(properties.get(TXN_FILE_REF))
            .unwrap()
            .into_txn_file_locks(snap.data_sequence);

        let mut meta = Self {
            engine_id,
            id: cs.shard_id,
            ver: cs.shard_ver,
            range: ShardRange::from_snap(snap),
            seq: cs.sequence,
            properties,
            base_version: snap.base_version,
            data_sequence: snap.data_sequence,
            max_ts: snap.max_ts,
            columnar_snap_version: snap.columnar_snap_version,
            unconverted_l0s: snap.unconverted_l0s.clone(),
            txn_file_locks,
            ..Default::default()
        };
        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() {
            let sm = snap.get_schema_meta();
            meta.schema_file_id = sm.get_file_id();
            meta.schema_file_ver = sm.get_version();
            meta.schema_restore_ver = sm.get_restore_version();
        }
        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),
                );
            }
        }
        meta
    }

    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 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>,
        enable_inner_key_offset: bool,
    ) -> Self {
        let range = if enable_inner_key_offset && 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.range.inner_key_off
            );
            ShardRange::new(start, end, parent.range.inner_key_off)
        };
        Self {
            id,
            ver,
            range,
            parent: Some(parent),
            properties: Properties::new().apply_pb(props),
            ..Default::default()
        }
    }

    fn move_down_file(&mut self, id: u64, cf: i32, level: u32) {
        let fm = self.files.get_mut(&id).unwrap();
        assert_eq!(
            fm.get_level() + 1,
            level,
            "fm.level {} level {}",
            fm.get_level(),
            level
        );
        if fm.cf == -1 {
            fm.cf = cf as i8;
        } else {
            assert_eq!(fm.cf, cf as i8);
        }
        fm.level = level as u8;
    }

    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) {
        if self.has_file_at_level(id, level) {
            self.files.remove(&id);
        }
    }

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

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

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

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

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

    pub fn del_property(&mut self, key: &str) {
        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() {
            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.get_major_compaction());
            return;
        }
        if 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();
            assert_eq!(self.schema_restore_ver, sm.get_restore_version());
            if !cs.get_property_key().is_empty() {
                assert_eq!(cs.get_property_key(), STORAGE_CLASS_KEY);
                self.set_property(STORAGE_CLASS_KEY, cs.get_property_value());
            }
            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.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() == TRUNCATE_TS_KEY {
                let truncate_ts = TruncateTs::unmarshal(cs.get_property_value());
                match self.get_property(TRUNCATE_TS_KEY) {
                    Some(v) => {
                        let cur_truncate_ts = if v.is_empty() {
                            None
                        } else {
                            Some(TruncateTs::unmarshal(v.chunk()))
                        };
                        if need_update_truncate_ts(cur_truncate_ts, truncate_ts) {
                            self.properties
                                .set(cs.get_property_key(), cs.get_property_value());
                        }
                    }
                    None => self
                        .properties
                        .set(cs.get_property_key(), cs.get_property_value()),
                };
            } 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()));
        let blobs_already_deleted = comp
            .get_old_blob_tables()
            .iter()
            .any(|blob_tbl_delete| !self.has_file_at_level(*blob_tbl_delete, 0));
        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_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_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();
            if schema_version <= self.schema_file_ver {
                info!(
                    "{} skip duplicated update schema meta version {}",
                    self.tag(),
                    schema_version
                );
                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) {
            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()
    }

    pub fn check_overlap_for_load_data(
        &self,
        ingest_files: &pb::IngestFiles,
    ) -> Option<(
        u64, // existed file id
        u64, // ingest file id
    )> {
        let ingest_tables = ingest_files.get_table_creates();
        // Ingest tables should be sorted, but we still check here for safety.
        let is_sorted =
            ingest_tables.is_sorted_by(|x, y| x.lower_bound().partial_cmp(&y.lower_bound()));

        // Ingest files of load data will be in bottom level of WRITE_CF only.
        for (&existed_file_id, existed_file) in self.files.iter().filter(|(_, f)| {
            f.get_cf() == WRITE_CF as i32 && f.get_level() == WRITE_CF_BOTTOM_LEVEL
        }) {
            if let Some(i) = existed_file
                .data_bound()
                .find_overlap(ingest_tables, is_sorted)
            {
                return Some((existed_file_id, ingest_tables[i].id));
            }
        }
        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.schema_file_id > 0 && self.columnar_snap_version > 0 {
            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_snap_version" => new_meta.columnar_snap_version,
            "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);
            }
            return;
        }
        for id in comp.get_top_deletes() {
            self.delete_file(*id, comp.level);
        }
        for id in comp.get_bottom_deletes() {
            self.delete_file(*id, comp.level + 1);
        }
        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, comp: &pb::MajorCompaction) {
        for delete in comp.get_sstable_change().get_table_deletes() {
            self.delete_file(delete.get_id(), delete.get_level());
        }
        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);
        }
        for create in comp.get_new_blob_tables() {
            self.add_file(create.get_id(), FileMeta::from_blob_table(create));
        }
    }

    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()))
    }

    pub fn is_empty_table_change(tc: &pb::TableChange) -> bool {
        tc.get_table_deletes().is_empty() && tc.get_table_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());
        }
        for created in tc.get_table_creates() {
            self.add_file(created.id, FileMeta::from_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());
        // 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());

        // ChangeSet of TruncateTs contains the corresponding ts which should be cleaned
        // up.
        assert_eq!(cs.get_property_key(), TRUNCATE_TS_KEY);
        if self.get_property(TRUNCATE_TS_KEY).is_some() {
            let truncated_ts = TruncateTs::unmarshal(cs.get_property_value());
            let v = self.get_property(TRUNCATE_TS_KEY).unwrap();
            let cur_truncate_ts = if v.is_empty() {
                None
            } else {
                Some(TruncateTs::unmarshal(v.chunk()))
            };
            // if applied truncate_ts is smaller than truncate ts in Meta, remove it.
            if need_update_truncate_ts(cur_truncate_ts, truncated_ts) {
                self.set_property(TRUNCATE_TS_KEY, b"");
            }
            info!(
                "Request change truncate ts in meta from {:?} to {:?}",
                cur_truncate_ts, truncated_ts
            );
        }
    }

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

    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: current ver:{}, seq:{}, base_ver:{}, data_seq:{}",
            self.tag(),
            self.ver,
            self.seq,
            self.base_version,
            self.data_sequence
        );

        // 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: new ver:{}, seq:{}, base_ver:{}, data_seq:{}, properties:{:?}",
            self.tag(),
            self.ver,
            self.seq,
            self.base_version,
            self.data_sequence,
            self.properties.to_pb(self.id),
        );
    }

    pub fn apply_split(
        &self,
        split: &kvenginepb::Split,
        sequence: u64,
        initial_seq: u64,
        enable_inner_key_offset: bool,
    ) -> 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;
        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(old.clone()),
                enable_inner_key_offset,
            );
            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;
            }
            meta.columnar_snap_version = old.columnar_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_file_id = self.schema_file_id;
            meta.schema_file_ver = self.schema_file_ver;
            meta.schema_restore_ver = self.schema_restore_ver;
            if let Some(storage_class) = &old_storage_class {
                meta.set_property(STORAGE_CLASS_KEY, storage_class);
            }
            new_shards.push(meta);
        }
        for new_shard in &mut new_shards {
            let new_shard_bound = new_shard.range.data_bound();
            for (fid, fm) in &old.files {
                if new_shard_bound.overlap_bound(fm.data_bound()) {
                    if fm.get_level() == 0
                        && new_shard.schema_file_id > 0
                        && old.unconverted_l0s.contains(fid)
                    {
                        new_shard.unconverted_l0s.push(*fid);
                    }
                    new_shard.files.insert(*fid, fm.clone());
                }
            }
            for vec_idx in &old.vector_indexes {
                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());
                    }
                }
                if !files.is_empty() {
                    let mut new_vec_idx = vec_idx.clone();
                    new_vec_idx.set_files(files.into());
                    new_shard.vector_indexes.push(new_vec_idx);
                }
            }
        }
        new_shards
    }

    pub fn apply_columnar_compaction(&mut self, comp: &pb::ColumnarCompaction) {
        let col_change = comp.get_columnar_change();
        for col_create in col_change.get_table_creates() {
            self.files
                .insert(col_create.get_id(), FileMeta::from_table(col_create));
        }
        for col_delete in col_change.get_table_deletes() {
            self.files.remove(&col_delete.get_id());
        }
        self.unconverted_l0s
            .retain(|unconverted_l0| !comp.row_l0s.contains(unconverted_l0));
        if self.columnar_snap_version == 0 {
            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.schema_file_id = 0;
            self.columnar_snap_version = 0;
            self.unconverted_l0s.clear();
        } else {
            self.columnar_snap_version = self.columnar_snap_version.max(comp.snap_version);
        }
    }

    pub fn apply_update_vector_index(&mut self, update_vec_idx: &pb::UpdateVectorIndex) {
        if let Some(old_idx) = self.vector_indexes.iter_mut().find(|vec_idx| {
            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
        }) {
            let mut old_files = old_idx.take_files().into_vec();
            old_files.retain(|f| !update_vec_idx.removed.contains(&f.id));
            old_files.extend_from_slice(update_vec_idx.get_added());
            old_idx.set_files(old_files.into());
            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;
        for file in update_vec_idx.get_added() {
            vec_idx.mut_files().push(file.clone());
        }
        self.vector_indexes.push(vec_idx)
    }

    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 = 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.range.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_snap_version(self.columnar_snap_version);
        for (k, v) in self.files.iter() {
            match v.file_type {
                FileType::Sst => {
                    if v.get_level() == 0 {
                        let mut l0 = pb::L0Create::new();
                        l0.set_id(*k);
                        l0.set_smallest(v.smallest.to_vec());
                        l0.set_biggest(v.biggest.to_vec());
                        snap.mut_l0_creates().push(l0);
                    } else {
                        let mut tbl = pb::TableCreate::new();
                        tbl.set_id(*k);
                        tbl.set_cf(v.cf as i32);
                        tbl.set_level(v.get_level());
                        tbl.set_smallest(v.smallest.to_vec());
                        tbl.set_biggest(v.biggest.to_vec());
                        tbl.set_meta_offset(v.table_meta_off);
                        snap.mut_table_creates().push(tbl);
                    }
                }
                FileType::TxnChunk | FileType::Schema => unreachable!("not loaded to file meta"),
                FileType::Columnar => {
                    let mut col_file = pb::ColumnarCreate::new();
                    col_file.set_id(*k);
                    col_file.set_level(v.get_level());
                    col_file.set_meta_offset(v.table_meta_off);
                    snap.mut_columnar_creates().push(col_file);
                }
                FileType::Blob => {
                    let mut blob = pb::BlobCreate::new();
                    blob.set_id(*k);
                    blob.set_smallest(v.smallest.to_vec());
                    blob.set_biggest(v.biggest.to_vec());
                    snap.mut_blob_creates().push(blob);
                }
                FileType::VectorIndex => {} // already handled in vector_indexes.
            }
        }
        if self.schema_file_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);
        }
        snap.set_unconverted_l0s(self.unconverted_l0s.clone());
        snap.set_vector_indexes(self.vector_indexes.clone().into());
        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>)> {
        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()));
            }
        }
        blob_files
    }

    // the ingest level never skip to lower level if upper level file exists, this
    // may not be optimal but prevent compaction generate conflicting files.
    // It find the top most existing file's level as ingest level, if there is
    // overlap with existing files, it will use the one level upper.
    pub(crate) fn get_ingest_level(&self, table_bound: DataBound<'_>) -> u32 {
        // find the top most level as ingest level.
        let mut ingest_level = self
            .files
            .values()
            .filter(|f| f.file_type == FileType::Sst)
            .map(|f| f.level)
            .min()
            .unwrap_or(3);
        if ingest_level == 0 {
            return 0;
        }
        let overlap = self
            .files
            .values()
            .filter(|f| f.level == ingest_level && f.cf == 0)
            .any(|file| table_bound.overlap_bound(file.data_bound()));
        if overlap {
            ingest_level -= 1;
        }
        ingest_level as u32
    }

    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) {
        let source_storage_class = source.get_property(STORAGE_CLASS_KEY);
        let old_storage_class = self.get_property(STORAGE_CLASS_KEY);

        // 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);
        } else if source_storage_class.is_some() && old_storage_class.is_some() {
            let source_storage_class = source_storage_class.unwrap();
            let target_storage_class = old_storage_class.unwrap();
            if source_storage_class.eq(&target_storage_class) {
                self.set_property(STORAGE_CLASS_KEY, &target_storage_class);
            }
        }

        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();
            self.files.clear();
            self.vector_indexes.clear();
            // remove DEL_PREFIXES_KEY property if exists
            self.del_property(DEL_PREFIXES_KEY);
        }
        if !clear_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);
            }

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

            // 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);
            }
            if 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 {
            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;
        if self.columnar_snap_version == 0 && self.schema_file_id > 0 {
            // Target shard is waiting for columnar major compaction, do
            // nothing.
        } else {
            self.columnar_snap_version =
                max(self.columnar_snap_version, source.columnar_snap_version);
        }
        self.parent = Some(Box::new(parent));
        self.seq = sequence;
    }

    fn merge_vector_index(&mut self, vec_idx: &VectorIndex) {
        if let Some(old_idx) = self.vector_indexes.iter_mut().find(|v| {
            v.table_id == vec_idx.table_id
                && v.index_id == vec_idx.table_id
                && v.col_id == vec_idx.col_id
        }) {
            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());
                }
            }
        } else {
            self.vector_indexes.push(vec_idx.clone());
        }
    }

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

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 1+) & Columnar.
    pub table_meta_off: u32,
}

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

    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 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,
            _ => false,
        }
    }

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

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

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

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

    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,
        )
    }

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

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

pub fn is_move_down(comp: &pb::Compaction) -> bool {
    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];

trait MetaReader {
    fn iterate_meta<F>(&self, f: F) -> Result<()>
    where
        F: Fn(&pb::ChangeSet) -> Result<()>;
}

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

    use super::*;

    #[test]
    fn test_ingest_level() {
        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", level);
            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", 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)),
            // L1:
            (101, FileMeta::new(0, 1, FileType::Sst, b"", b"", 0)),
            (102, FileMeta::new(0, 1, FileType::Sst, b"", b"", 0)),
            // L2:
            (201, FileMeta::new(0, 2, FileType::Sst, b"", b"", 0)),
            (202, FileMeta::new(0, 2, FileType::Sst, b"", b"", 0)),
        ];

        // 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.set_property_key(TRUNCATE_TS_KEY.to_string());
                        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() {
        test_table_overlap_helper(false);
        test_table_overlap_helper(true);
    }

    fn test_table_overlap_helper(enable_inner_key_off: bool) {
        let range = if enable_inner_key_off {
            ShardRange::new(&[b'x', 2, 3, 4, 10], &[b'x', 2, 3, 4, 20], 4)
        } else {
            ShardRange::new(&[b'x', 2, 3, 4, 10], &[b'x', 2, 3, 4, 20], 0)
        };
        let meta = ShardMeta {
            range,
            ..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_load_data() {
        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,
                        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),
                );
            }
            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_load_data(&ingest_files);
            assert_eq!(overlap, expected, "case {}", idx);
        }
    }
}