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

use std::{
    cmp::Ordering,
    convert::TryFrom,
    fmt,
    future::Future,
    ops::Deref,
    path::{Path, PathBuf},
    sync::Arc,
};

use api_version::ApiV2;
use byteorder::{ByteOrder, LittleEndian};
use bytes::{Buf, Bytes, BytesMut};
use cloud_encryption::{EncryptionKey, EncryptionState};
use kvenginepb::TableCreate;
use schema::schema::StorageClass;
use tikv_util::sys::SysQuota;
use xorf::{BinaryFuse8, Filter};

use super::{builder::*, iterator::TableIterator};
use crate::{
    IoContext, WRITE_CF,
    ia::{ia_auto_file::IaAutoFile, ia_file::IaFile, types::FileSegmentIdent},
    metrics::ENGINE_SST_CRYPT_DURATION_STATIC,
    next_version, next_version_async,
    table::{
        file::File,
        table::{Iterator, Result},
        tiny_meta::SstTinyMeta,
        *,
    },
};

// higher level ttl is longer than lower level.
const SMALL_VALUE_SIZE: u64 = 96;

/// The SSTable format:
/// +-------------------+
/// | Current Data Block|
/// |-------------------|
/// | key3@v3 -> value3 |  // Latest version handled by block_builder
/// | key4@v2 -> value2 |
/// | key5@v1 -> value1 |
/// +-------------------+
/// | Old Data Block    |
/// |-------------------|
/// | key3@v2 -> value2 |  // Older version handled by old_builder
/// | key3@v1 -> value1 |
/// | key4@v1 -> value1 |
/// +-------------------+
/// | Index Block       |
/// |-------------------|
/// | key3 -> Current Data Block |
/// | key3 -> Old Data Block     |
/// | key4 -> Current Data Block |
/// | key4 -> Old Data Block     |
/// | key5 -> Current Data Block |
/// +-------------------+
/// | Bloom Filter      |
/// +-------------------+
/// | Footer            |
/// +-------------------+
#[derive(Clone)]
pub struct SsTable {
    core: Arc<SsTableCore>,
}

impl Deref for SsTable {
    type Target = SsTableCore;
    fn deref(&self) -> &Self::Target {
        &self.core
    }
}

impl fmt::Debug for SsTable {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SsTable")
            .field("id", &self.id())
            .field("size", &self.size())
            .field("entries", &self.entries)
            .field("old_entries", &self.old_entries)
            .field("tombs", &self.tombs)
            .field("max_ts", &self.max_ts)
            .field("kv_size", &self.kv_size)
            .field("snap_version", &self.snap_version)
            .field("is_sync", &self.is_sync())
            .finish()
    }
}

impl SsTable {
    pub fn new(
        file: Arc<dyn File>,
        cache: BlockCache,
        encryption_state: EncryptionState,
    ) -> Result<Self> {
        let mut ctx = NewSsTableCtx::default();
        Self::new_with_ctx(file, cache, encryption_state, &mut ctx)
    }

    pub fn new_with_ctx(
        file: Arc<dyn File>,
        cache: BlockCache,
        encryption_state: EncryptionState,
        ctx: &mut NewSsTableCtx,
    ) -> Result<Self> {
        let size = file.size();
        let core = SsTableCore::new(file, 0, size, cache, encryption_state, ctx)?;
        Ok(Self {
            core: Arc::new(core),
        })
    }

    // L0 does not use IA files.
    pub fn new_l0_cf(
        file: Arc<dyn File>,
        start: u64,
        end: u64,
        cache: BlockCache,
        encryption_state: EncryptionState,
    ) -> Result<Self> {
        let mut ctx = NewSsTableCtx::default();
        let core = SsTableCore::new(file, start, end, cache, encryption_state, &mut ctx)?;
        Ok(Self {
            core: Arc::new(core),
        })
    }

    pub fn new_iterator(&self, reversed: bool, fill_cache: bool) -> Box<TableIterator> {
        let it = TableIterator::new(self.clone(), reversed, fill_cache);
        Box::new(it)
    }

    // Get the value with the given key and version.
    // It is caller's responsibility to maintain the lifetime of the returned value.
    // Value will be filled in out_val_owner , while the returned value is a parsed
    // slice of it.
    #[maybe_async::both]
    pub async fn get(
        &self,
        key: InnerKey<'_>,
        version: u64,
        mut key_hash: u64,
        out_val_owner: &mut Vec<u8>,
        level: usize,
    ) -> table::Value {
        // For small value on level 3, load the filter is not cost-effective.
        // TODO: avoid build filter on level 3 small value table.
        let small_value = self.kv_size < self.entries as u64 * SMALL_VALUE_SIZE;
        let skip_filter = small_value && level == 3;
        self.maybe_update_key_hash(key, &mut key_hash);
        if self.filter_size() > 0 && !skip_filter {
            let filter = self.get_filter();
            if !filter.contains(&key_hash) {
                return table::Value::new();
            }
        }
        let mut it = self.new_iterator(false, true);
        it.seek(key).await;
        if !it.valid() || key != it.key() {
            return table::Value::new();
        }
        while it.value().version > version {
            if !next_version!(it).await {
                return table::Value::new();
            }
        }
        // Reconstruct the value to avoid the lifetime issue.
        let val = it.value();
        out_val_owner.resize(val.encoded_size(), 0);
        val.encode(out_val_owner.as_mut_slice());
        Value::decode(out_val_owner.as_slice())
    }

    fn maybe_update_key_hash(&self, key: InnerKey<'_>, key_hash: &mut u64) {
        if let Some(keyspace_id) = self.keyspace_id {
            let mut outer_key = ApiV2::get_txn_keyspace_prefix(keyspace_id);
            outer_key.extend_from_slice(key.deref());
            *key_hash = farmhash::fingerprint64(&outer_key);
        }
    }

    #[maybe_async::both]
    async fn check_overlap_by_seek(&self, data_bound: DataBound<'_>) -> bool {
        let mut it = self.new_iterator(false, true);
        it.seek(data_bound.lower_bound).await;
        if !it.valid() {
            return it.error().is_some();
        }
        !data_bound.less_than_key(it.key())
    }

    #[maybe_async::both]
    pub async fn has_overlap(&self, data_bound: DataBound<'_>) -> bool {
        if !self.data_bound().overlap_bound(data_bound) {
            return false;
        }
        self.check_overlap_by_seek(data_bound).await
    }

    /// Used when seek on async tables is expensive and strictly overlapped is
    /// not necessary.
    pub fn has_overlap_loose(&self, data_bound: DataBound<'_>) -> bool {
        if !self.data_bound().overlap_bound(data_bound) {
            return false;
        }
        !self.is_sync() || self.check_overlap_by_seek(data_bound)
    }

    #[maybe_async::both]
    pub async fn has_any_overlap<'a>(
        &self,
        bounds: impl std::iter::Iterator<Item = DataBound<'a>>,
    ) -> bool {
        for bound in bounds {
            if self.has_overlap(bound).await {
                return true;
            }
        }
        false
    }

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

    pub(crate) fn clone_smallest(&self) -> Bytes {
        self.smallest_buf.clone()
    }

    pub(crate) fn clone_biggest(&self) -> Bytes {
        self.biggest_buf.clone()
    }

    pub const fn footer_size() -> usize {
        FOOTER_SIZE
    }

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

    #[inline]
    pub fn is_cf_sync(&self, cf: usize) -> bool {
        cf != WRITE_CF || self.is_sync()
    }

    pub fn get_remote_segments(
        &self,
        data_bound: DataBound<'_>,
    ) -> Result<(Vec<FileSegmentIdent>, usize /* total_segments */)> {
        if self.file.is_sync() {
            return Ok((vec![], 0));
        }

        let idx = self.load_index(true);
        let (first_block, exclusive_last_block) = idx.seek_overlap_blocks(data_bound);
        let start_off = idx.get_block_addr(first_block).curr_off as u64;
        let end_off = if exclusive_last_block < idx.num_blocks() {
            idx.get_block_addr(exclusive_last_block).curr_off as u64
        } else {
            self.data_end_offset() as u64
        };

        self.file.get_remote_segments(&[(start_off, end_off)])
    }

    pub fn get_meta_data(file_data: &[u8]) -> Result<Bytes> {
        let data_len = file_data.len();
        let footer_size = SsTable::footer_size();
        if data_len < footer_size {
            error!("get_meta_data: invalid fie size"; "data_len" => data_len);
            return Err(Error::InvalidFileSize);
        }
        let footer_data = &file_data[data_len - footer_size..];

        let mut footer = Footer::default();
        footer.unmarshal(footer_data);
        footer.validate_offsets(data_len)?;

        let meta_off = footer.meta_offset() as usize;
        if data_len < meta_off {
            error!("get_meta_data: invalid meta offset"; "data_len" => data_len, "footer" => ?footer);
            return Err(Error::InvalidFileSize);
        }
        Ok(Bytes::copy_from_slice(&file_data[meta_off..]))
    }
}

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

pub struct SsTableCore {
    file: Arc<dyn File>,
    cache: BlockCache,
    start_off: u64,
    end_off: u64,
    footer: Footer,
    smallest_buf: Bytes,
    biggest_buf: Bytes,
    pub max_ts: u64,
    pub max_lock_ts: u64,
    pub entries: u32,
    pub old_entries: u32,
    pub tombs: u32,
    pub kv_size: u64,
    pub in_use_total_blob_size: u64,
    encryption_state: EncryptionState,
    encryption_ver: u32,
    pub snap_version: SnapVersion,
    // Legacy table has key with keyspace_id prefix,
    // We need to prepend it to perform bloom filter get.
    pub keyspace_id: Option<u32>,
}

pub enum SsTableProperty {
    MaxTs(u64),
}

impl SsTableCore {
    pub fn new(
        file: Arc<dyn File>,
        start_off: u64,
        end_off: u64,
        cache: BlockCache,
        encryption_state: EncryptionState,
        ctx: &mut NewSsTableCtx,
    ) -> Result<Self> {
        if end_off > file.size() {
            return Err(table::Error::InvalidFileSize);
        }
        let size = end_off
            .checked_sub(start_off)
            .ok_or(table::Error::InvalidFileSize)?;

        let (footer, props_data) = match ctx.take_footer_and_properties() {
            Some(x) => x,
            None => {
                let (footer, props_data) = Self::read_footer_and_properties_from_file(
                    file.as_ref(),
                    start_off,
                    end_off,
                    &encryption_state,
                )?;
                if end_off == file.size() {
                    // Handle normal SST (not L0) only by now.
                    ctx.put_footer_and_properties(footer, props_data.clone());
                }
                (footer, props_data)
            }
        };
        footer.validate_offsets(size as usize)?;

        let mut prop_slice = props_data
            .chunk()
            .get(4..)
            .ok_or(table::Error::InvalidFileSize)?; // Trim checksum.
        let mut smallest_buf = Bytes::new();
        let mut biggest_buf = Bytes::new();
        let mut max_ts = 0;
        let mut max_lock_ts = 0;
        let mut entries = 0;
        let mut old_entries = 0;
        let mut tombs = 0;
        let mut kv_size = None;
        let mut in_use_total_blob_size = 0u64;
        let mut encryption_ver = 0;
        let mut snap_version = SnapVersion::default();
        while !prop_slice.is_empty() {
            let (key, val, remain) = parse_prop_data(prop_slice)?;
            prop_slice = remain;
            if key == PROP_KEY_SMALLEST {
                smallest_buf = Bytes::copy_from_slice(val);
            } else if key == PROP_KEY_BIGGEST {
                biggest_buf = Bytes::copy_from_slice(val);
            } else if key == PROP_KEY_MAX_TS {
                max_ts = parse_u64_property(val)?;
            } else if key == PROP_KEY_MAX_LOCK_TS {
                max_lock_ts = parse_u64_property(val)?;
            } else if key == PROP_KEY_ENTRIES {
                entries = parse_u32_property(val)?;
            } else if key == PROP_KEY_OLD_ENTRIES {
                old_entries = parse_u32_property(val)?;
            } else if key == PROP_KEY_TOMBS {
                tombs = parse_u32_property(val)?;
            } else if key == PROP_KEY_KV_SIZE {
                kv_size = Some(parse_u64_property(val)?);
            } else if key == PROP_KEY_IN_USE_TOTAL_BLOB_SIZE {
                in_use_total_blob_size = parse_u64_property(val)?;
            } else if key == PROP_KEY_ENCRYPTION_VER {
                encryption_ver = parse_u32_property(val)?;
            } else if key == PROP_KEY_SNAP_VERSION {
                snap_version = parse_u64_property(val)?.into();
            }
        }
        let keyspace_id = ApiV2::get_u32_keyspace_id_by_key(smallest_buf.chunk());
        let core = Self {
            file,
            cache,
            start_off,
            end_off,
            footer,
            smallest_buf,
            biggest_buf,
            max_ts,
            max_lock_ts,
            entries,
            old_entries,
            tombs,
            kv_size: kv_size.unwrap_or(size),
            in_use_total_blob_size,
            encryption_ver,
            encryption_state,
            snap_version,
            keyspace_id,
        };
        Ok(core)
    }

    fn read_footer_and_properties_from_file(
        file: &dyn File,
        start_off: u64,
        end_off: u64,
        encryption_state: &EncryptionState,
    ) -> Result<(Footer, Bytes)> {
        let size = end_off - start_off;

        let mut footer = Footer::default();
        let footer_data = if end_off == file.size() {
            file.read_footer(FOOTER_SIZE)?
        } else {
            if size < FOOTER_SIZE as u64 {
                return Err(table::Error::InvalidFileSize);
            }
            // It's a L0 sst, and must be sync.
            file.read(end_off - FOOTER_SIZE as u64, FOOTER_SIZE)?
        };
        footer.unmarshal(footer_data.chunk());
        if footer.magic != MAGIC_NUMBER && footer.magic != MAGIC_NUMBER_SPLIT_L0 {
            return Err(table::Error::InvalidMagicNumber);
        }
        footer.validate_offsets(size as usize)?;
        let props_data = file.read_table_meta(
            start_off + footer.properties_offset as u64,
            footer.properties_len(size as usize),
        )?;
        let prop_slice = props_data.chunk();
        validate_checksum_with_fix(prop_slice, &footer, file, encryption_state)?;

        Ok((footer, props_data))
    }

    // Support max_ts only.
    // TODO: Support other properties.
    pub fn extract_property_from_table_meta_file(
        table_meta_file: &dyn File,
        prop_key: &[u8],
    ) -> Result<SsTableProperty> {
        let mut footer = Footer::default();
        let footer_data = table_meta_file.read_footer(FOOTER_SIZE)?;
        footer.unmarshal(footer_data.chunk());
        if !footer.is_match() {
            return Err(table::Error::InvalidMagicNumber);
        }

        let meta_size = table_meta_file.size();
        let table_meta_off = footer.meta_offset() as u64;
        let table_size = table_meta_off
            .checked_add(meta_size)
            .and_then(|size| usize::try_from(size).ok())
            .ok_or(table::Error::InvalidFileSize)?;
        footer.validate_offsets(table_size)?;
        let properties_off = u64::from(footer.properties_offset)
            .checked_sub(table_meta_off)
            .ok_or(table::Error::InvalidFileSize)?;

        let props_data =
            table_meta_file.read_table_meta(properties_off, footer.properties_len(table_size))?;
        let mut prop_slice = props_data.chunk();
        validate_checksum_with_fix(
            prop_slice,
            &footer,
            table_meta_file,
            &EncryptionState::Disabled,
        )?;
        prop_slice = prop_slice.get(4..).ok_or(table::Error::InvalidFileSize)?;
        while !prop_slice.is_empty() {
            let (key, val, remain) = parse_prop_data(prop_slice)?;
            prop_slice = remain;
            if key == prop_key {
                let v = match key {
                    PROP_KEY_MAX_TS => SsTableProperty::MaxTs(parse_u64_property(val)?),
                    _ => {
                        return Err(table::Error::Other(format!(
                            "property not supported: {:?}",
                            key
                        )));
                    }
                };
                return Ok(v);
            }
        }
        Err(table::Error::Other(format!(
            "property not found: {:?}",
            prop_key
        )))
    }

    pub fn footer(&self) -> Footer {
        self.footer
    }

    fn index_cache_key(&self) -> BlockCacheKey {
        // Index sections do not have BlockAddress entries. Use the physical
        // file id plus the absolute index-section offset so normal SSTs, L0
        // per-CF SST segments, and current/old indexes never collide.
        BlockCacheKey::new(self.id(), self.index_offset())
    }

    fn old_index_cache_key(&self) -> BlockCacheKey {
        // Keep the old-version index under its own section offset. It can be
        // loaded independently from the current-version index by MVCC reads.
        BlockCacheKey::new(self.id(), self.old_index_offset())
    }

    pub fn init_index(&self, offset: u32, length: usize) -> Result<Index> {
        let idx_data = self
            .file
            .read_table_meta(self.start_off + offset as u64, length)?;
        self.validate_checksum_with_fix(idx_data.chunk())?;
        // Copy only when idx_data slices into a larger shared buffer (e.g. an
        // InMemFile whole-file load) so the cached Index never pins that file.
        // A standalone allocation (e.g. a LocalFile pread) is already unique
        // and pins nothing, so keep it as-is. A shared slice is never unique,
        // so the pinning case always copies.
        let idx_data = if idx_data.is_unique() {
            idx_data
        } else {
            Bytes::copy_from_slice(idx_data.chunk())
        };
        Index::new(idx_data)
    }

    /// Loads the current-version block index.
    ///
    /// When `fill_cache` is false, an existing cached index is reused, but a
    /// cache miss is loaded without inserting the result into the shared block
    /// cache.
    pub fn try_load_index(&self, fill_cache: bool) -> Result<Arc<Index>> {
        let offset = self.footer.index_offset;
        let length = self
            .footer
            .old_index_offset
            .checked_sub(offset)
            .ok_or(table::Error::InvalidFileSize)? as usize;
        let key = self.index_cache_key();
        if fill_cache {
            self.cache
                .try_get_index_with(key, || self.init_index(offset, length))
        } else if let Some(index) = self.cache.get_cached_index(&key) {
            Ok(index)
        } else {
            self.cache.report_cache_miss();
            self.init_index(offset, length).map(Arc::new)
        }
    }

    pub fn load_index(&self, fill_cache: bool) -> Arc<Index> {
        self.try_load_index(fill_cache).expect("load index")
    }

    pub fn load_old_index(&self, fill_cache: bool) -> Arc<Index> {
        let key = self.old_index_cache_key();
        let (offset, length) = (self.footer.old_index_offset, self.footer.old_index_len());
        if fill_cache {
            self.cache
                .try_get_index_with(key, || self.init_index(offset, length))
                .expect("load old index")
        } else if let Some(index) = self.cache.get_cached_index(&key) {
            index
        } else {
            self.cache.report_cache_miss();
            Arc::new(self.init_index(offset, length).expect("load old index"))
        }
    }

    pub fn expire_cache(&self, file_ttl: u64) {
        self.file.expire_open_file(file_ttl);
        // Filters and indexes live in the shared block cache. They are evicted
        // by weighted capacity together with data blocks, not by the previous
        // per-table TTL policy.
    }

    #[maybe_async::both]
    pub async fn load_block(
        &self,
        idx: &Index,
        pos: usize,
        buf: &mut Vec<u8>,
        decryption_buf: &mut Vec<u8>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        let addr = idx.get_block_addr(pos);
        let length = if pos + 1 < idx.num_blocks() {
            let next_addr = idx.get_block_addr(pos + 1);
            (next_addr.curr_off - addr.curr_off) as usize
        } else {
            self.start_off as usize + self.footer.data_len() - addr.curr_off as usize
        };
        self.load_block_by_addr_len(addr, length, buf, decryption_buf, fill_cache)
            .await
    }

    fn load_block_by_addr_len(
        &self,
        addr: BlockAddress,
        length: usize,
        buf: &mut Vec<u8>,
        decryption_buf: &mut Vec<u8>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        let cache_key = BlockCacheKey::new(addr.origin_fid, addr.origin_off);
        self.cache.try_get_with_ext(
            cache_key,
            || self.read_block_from_file(addr, length, buf, decryption_buf),
            fill_cache,
        )
    }

    async fn load_block_by_addr_len_async(
        &self,
        addr: BlockAddress,
        length: usize,
        buf: &mut Vec<u8>,
        decryption_buf: &mut Vec<u8>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        let cache_key = BlockCacheKey::new(addr.origin_fid, addr.origin_off);
        self.cache
            .try_get_with_ext_async(
                cache_key,
                self.read_block_from_file_async(addr, length, buf, decryption_buf),
                fill_cache,
            )
            .await
    }

    #[maybe_async::both]
    async fn read_block_from_file(
        &self,
        addr: BlockAddress,
        length: usize,
        buf: &mut Vec<u8>,
        decryption_buf: &mut Vec<u8>,
    ) -> Result<Bytes> {
        let compression_type = self.footer.compression_type;
        if compression_type == NO_COMPRESSION {
            let mut raw_block = self.file.read(addr.curr_off as u64, length).await?;
            if let Some(encryption_key) = self.ready_encryption_key()? {
                let timer = tikv_util::time::Instant::now_coarse();
                let mut block = Vec::with_capacity(length + encryption_key.encryption_block_size());
                encryption_key.decrypt(
                    raw_block.chunk(),
                    addr.origin_fid,
                    addr.curr_off,
                    &mut block,
                );
                if !encryption_key.is_noop() {
                    ENGINE_SST_CRYPT_DURATION_STATIC
                        .decrypt_data_block
                        .observe(timer.saturating_elapsed().as_secs_f64());
                }
                raw_block = Bytes::from(block)
            }
            self.validate_checksum_with_fix(raw_block.chunk())?;
            return Ok(raw_block.slice(4..));
        }
        buf.resize(length, 0);
        if let Some(encryption_key) = self.ready_encryption_key()? {
            decryption_buf.resize(length, 0);
            self.file
                .read_at(decryption_buf, addr.curr_off as u64)
                .await?;
            let timer = tikv_util::time::Instant::now_coarse();
            buf.truncate(0);
            encryption_key.decrypt(decryption_buf, addr.origin_fid, addr.curr_off, buf);
            if !encryption_key.is_noop() {
                ENGINE_SST_CRYPT_DURATION_STATIC
                    .decrypt_data_block
                    .observe(timer.saturating_elapsed().as_secs_f64());
            }
        } else {
            self.file.read_at(buf, addr.curr_off as u64).await?;
        }
        self.validate_checksum_with_fix(buf)?;
        let content = &buf[4..];
        match compression_type {
            LZ4_COMPRESSION => {
                let block = lz4::block::decompress(content, None)
                    .table_ctx(self.file.id(), "sst.read_block.lz4_decompress")?;
                Ok(Bytes::from(block))
            }
            ZSTD_COMPRESSION => {
                let capacity = unsafe {
                    zstd_sys::ZSTD_getFrameContentSize(
                        content.as_ptr() as *const libc::c_void,
                        content.len(),
                    ) as usize
                };
                let mut block = Vec::<u8>::with_capacity(capacity);
                unsafe {
                    let result = zstd_sys::ZSTD_decompress(
                        block.as_mut_ptr() as *mut libc::c_void,
                        capacity,
                        content.as_ptr() as *const libc::c_void,
                        content.len(),
                    );
                    assert_eq!(zstd_sys::ZSTD_isError(result), 0u32);
                    block.set_len(capacity);
                }
                Ok(Bytes::from(block))
            }
            _ => panic!("unknown compression type {}", compression_type),
        }
    }

    fn ready_encryption_key(&self) -> Result<Option<EncryptionKey>> {
        if matches!(&self.encryption_state, EncryptionState::Disabled) {
            return Ok(None);
        }
        self.encryption_state
            .switch_if_header_mismatch(self.encryption_ver)
            .map(Some)
            .map_err(|_| {
                table::Error::EncryptionKeyUnavailable(
                    "failed to resolve encryption key".to_owned(),
                )
            })
    }

    #[maybe_async::both]
    pub async fn load_old_block(
        &self,
        old_idx: &Index,
        pos: usize,
        buf: &mut Vec<u8>,
        decryption_buf: &mut Vec<u8>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        let addr = old_idx.get_block_addr(pos);
        let length = if pos + 1 < old_idx.num_blocks() {
            let next_addr = old_idx.get_block_addr(pos + 1);
            (next_addr.curr_off - addr.curr_off) as usize
        } else {
            self.start_off as usize + self.footer.index_offset as usize - addr.curr_off as usize
        };
        self.load_block_by_addr_len(addr, length, buf, decryption_buf, fill_cache)
            .await
    }

    fn get_filter(&self) -> Arc<BinaryFuse8> {
        self.cache
            .try_get_filter_with(self.filter_cache_key(), || {
                let filter_data = self.read_filter_data_from_file()?;
                // decode_filter (try_from_bytes) shares filter_data zero-copy,
                // so copy first when it slices into a larger shared buffer (an
                // InMemFile whole-file load) to keep the cached filter from
                // pinning that file. A standalone allocation (a LocalFile pread)
                // is already unique and pins nothing, so keep it as-is.
                let filter_data = if filter_data.is_unique() {
                    filter_data
                } else {
                    Bytes::copy_from_slice(filter_data.chunk())
                };
                self.decode_filter(&filter_data)
            })
            .expect("load filter")
    }

    fn filter_cache_key(&self) -> BlockCacheKey {
        // The filter section does not have a BlockAddress in the SST index.
        // The physical file id plus absolute filter-section offset is stable
        // for both normal SSTs and L0 files that concatenate per-CF SSTs.
        BlockCacheKey::new(self.id(), self.filter_offset())
    }

    fn read_filter_data_from_file(&self) -> Result<Bytes> {
        let mut data = self
            .file
            .read_table_meta(self.filter_offset() as u64, self.filter_size() as usize)?;
        self.validate_checksum_with_fix(data.chunk())?;
        data.get_u32_le();
        assert_eq!(data.get_u32_le(), AUX_INDEX_BINARY_FUSE8);
        let len = data.get_u32_le();
        assert_eq!(len as usize, data.len());
        Ok(data)
    }

    fn decode_filter(&self, data: &Bytes) -> Result<BinaryFuse8> {
        BinaryFuse8::try_from_bytes(data).map_err(|e| Error::Other(e.to_string()))
    }

    pub fn id(&self) -> u64 {
        self.file.id()
    }

    pub fn file(&self) -> &Arc<dyn File> {
        &self.file
    }

    pub fn try_get_ia_file(&self) -> Option<Arc<IaFile>> {
        self.file.clone().as_any().downcast::<IaFile>().ok()
    }

    pub fn try_get_auto_ia_file(&self) -> Option<Arc<IaAutoFile>> {
        self.file.clone().as_any().downcast::<IaAutoFile>().ok()
    }

    #[inline]
    pub(crate) fn is_storage_class_ia(&self) -> bool {
        self.file.storage_class() == StorageClass::Ia
    }

    /// The size on disk in bytes.
    /// Should be equal to `self.file.size()` for L1+, and less than or equal to
    /// for L0.
    pub fn size(&self) -> u64 {
        self.end_off - self.start_off
    }

    /// The compressed size of old-version data blocks in bytes.
    pub fn old_data_size(&self) -> u64 {
        self.footer.old_data_len() as u64
    }

    pub fn index_size(&self) -> u64 {
        (self.footer.index_len() + self.footer.old_index_len()) as u64
    }

    fn data_end_offset(&self) -> u32 {
        self.start_off as u32 + self.footer.data_len() as u32
    }

    /// The offset of first table meta (index).
    #[inline]
    pub fn meta_offset(&self) -> u32 {
        self.start_off as u32 + self.footer.meta_offset()
    }

    pub fn index_offset(&self) -> u32 {
        self.start_off as u32 + self.footer.index_offset
    }

    fn old_index_offset(&self) -> u32 {
        self.start_off as u32 + self.footer.old_index_offset
    }

    fn filter_offset(&self) -> u32 {
        self.start_off as u32 + self.footer.aux_index_offset
    }

    pub fn filter_size(&self) -> u64 {
        self.footer.aux_index_len() as u64
    }

    pub fn smallest(&self) -> InnerKey<'_> {
        InnerKey::from_inner_buf(self.smallest_buf.chunk())
    }

    pub fn biggest(&self) -> InnerKey<'_> {
        InnerKey::from_inner_buf(self.biggest_buf.chunk())
    }

    pub fn compression_type(&self) -> u8 {
        self.footer.compression_type
    }

    pub fn total_blob_size(&self) -> u64 {
        self.in_use_total_blob_size
    }

    pub fn encryption_ver(&self) -> u32 {
        self.encryption_ver
    }

    pub fn get_max_lock_ts(&self) -> u64 {
        // For legacy file, the max_lock_ts is zero, we use file id instead.
        // TODO: use tbl.max_lock_ts directly after all old lock files are GCed.
        if self.max_lock_ts > 0 {
            self.max_lock_ts
        } else {
            self.file.id()
        }
    }

    #[inline]
    fn validate_checksum_with_fix(&self, data: &[u8]) -> Result<()> {
        validate_checksum_with_fix(
            data,
            &self.footer,
            self.file.as_ref(),
            &self.encryption_state,
        )
    }

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

// Used in sstable and blobtable.
#[derive(Clone)]
pub struct Index {
    common_prefix: Bytes,
    block_key_offs: Bytes,
    block_addrs: Bytes,
    block_keys: Bytes,
    encoded_len: usize,
}

impl Index {
    pub(crate) fn new(data: Bytes) -> Result<Self> {
        let index = Self::new_with_opts(data, BLOCK_ADDR_SIZE)?;
        index.validate_block_key_order()?;
        Ok(index)
    }

    pub(crate) fn new_for_blob(data: Bytes) -> Result<Self> {
        Self::new_with_opts(data, 4)
    }

    fn new_with_opts(mut data: Bytes, block_addr_size: usize) -> Result<Self> {
        let encoded_len = data.len();
        if data.len() < 12 {
            return Err(table::Error::InvalidFileSize);
        }
        data.advance(4); // The checksum is validated before parsing the index.
        let format = data.get_u32_le();
        if format != INDEX_FORMAT_V1 {
            return Err(table::Error::Other(format!(
                "unsupported index format {}",
                format
            )));
        }
        let num_blocks = data.get_u32_le() as usize;
        let block_key_offs_len = num_blocks
            .checked_mul(4)
            .ok_or(table::Error::InvalidFileSize)?;
        let block_addrs_len = num_blocks
            .checked_mul(block_addr_size)
            .ok_or(table::Error::InvalidFileSize)?;
        let fixed_fields_len = block_key_offs_len
            .checked_add(block_addrs_len)
            .and_then(|len| len.checked_add(2))
            .ok_or(table::Error::InvalidFileSize)?;
        if data.len() < fixed_fields_len {
            return Err(table::Error::InvalidFileSize);
        }

        let block_key_offs = data.slice(..block_key_offs_len);
        data.advance(block_key_offs.len());
        let block_addrs = data.slice(..block_addrs_len);
        data.advance(block_addrs.len());
        let common_prefix_len = data.get_u16_le() as usize;
        if data.len() < common_prefix_len + 4 {
            return Err(table::Error::InvalidFileSize);
        }
        let common_prefix = data.slice(..common_prefix_len);
        data.advance(common_prefix.len());
        let block_key_len = data.get_u32_le() as usize;
        if data.len() != block_key_len {
            return Err(table::Error::InvalidFileSize);
        }
        let block_keys = data.slice(..block_key_len);

        let mut previous_off = 0;
        for (index, offset) in block_key_offs.chunks_exact(4).enumerate() {
            let offset = LittleEndian::read_u32(offset) as usize;
            if (index == 0 && offset != 0) || offset < previous_off || offset > block_key_len {
                return Err(table::Error::InvalidFileSize);
            }
            previous_off = offset;
        }
        Ok(Self {
            common_prefix,
            block_key_offs,
            block_addrs,
            block_keys,
            encoded_len,
        })
    }

    fn validate_block_key_order(&self) -> Result<()> {
        for i in 1..self.num_blocks() {
            if self.block_diff_key(i - 1) >= self.block_diff_key(i) {
                return Err(table::Error::InvalidFileSize);
            }
        }
        Ok(())
    }

    fn get_common_prefix(&self) -> InnerKey<'_> {
        InnerKey::from_inner_buf(self.common_prefix.chunk())
    }

    pub(crate) fn get_block_addr(&self, pos: usize) -> BlockAddress {
        let off = pos * BLOCK_ADDR_SIZE;
        BlockAddress::from_slice(&self.block_addrs[off..off + BLOCK_ADDR_SIZE])
    }

    pub(crate) fn get_block_addr_for_blob(&self, pos: usize) -> u32 {
        let off = pos * 4;
        LittleEndian::read_u32(&self.block_addrs[off..off + 4])
    }

    pub fn num_blocks(&self) -> usize {
        self.block_key_offs.len() / 4
    }

    /// Returns the block index of the first block whose key > `key`.
    pub fn seek_block(&self, key: InnerKey<'_>) -> usize {
        let common_prefix = self.get_common_prefix();
        if key.len() <= common_prefix.len() {
            if key <= common_prefix {
                return 0;
            }
            return self.num_blocks();
        }
        let cmp = key.slice(0, common_prefix.len()).cmp(&common_prefix);
        match cmp {
            Ordering::Less => 0,
            Ordering::Equal => {
                let diff_key = &key[common_prefix.len()..];
                search(self.num_blocks(), |i| self.block_diff_key(i) > diff_key)
            }
            Ordering::Greater => self.num_blocks(),
        }
    }

    /// Returns the block index of the first block whose key >= `key`.
    pub fn seek_block_bigger_or_equal(&self, key: InnerKey<'_>) -> usize {
        let common_prefix = self.get_common_prefix();
        if key.len() <= common_prefix.len() {
            if key <= common_prefix {
                return 0;
            }
            return self.num_blocks();
        }
        let cmp = key[..common_prefix.len()].cmp(&common_prefix);
        match cmp {
            Ordering::Less => 0,
            Ordering::Equal => {
                let diff_key = &key[common_prefix.len()..];
                search(self.num_blocks(), |i| self.block_diff_key(i) >= diff_key)
            }
            Ordering::Greater => self.num_blocks(),
        }
    }

    pub fn seek_overlap_blocks(
        &self,
        data_bound: DataBound<'_>,
    ) -> (
        usize, // first_block_idx
        usize, // exclusive_last_block_idx
    ) {
        let first_idx = self.seek_block(data_bound.lower_bound).saturating_sub(1);
        let last_exclusive_idx = if data_bound.upper_inclusive {
            self.seek_block(data_bound.upper_bound)
        } else {
            self.seek_block_bigger_or_equal(data_bound.upper_bound)
        };
        (first_idx, last_exclusive_idx)
    }

    fn block_diff_key(&self, i: usize) -> &[u8] {
        let off = self.get_block_key_off(i);
        let end_off = if i + 1 < self.num_blocks() {
            self.get_block_key_off(i + 1)
        } else {
            self.block_keys.len()
        };
        &self.block_keys[off..end_off]
    }

    fn get_block_key_off(&self, i: usize) -> usize {
        (&self.block_key_offs[i * 4..]).get_u32_le() as usize
    }

    pub(crate) fn clone_block_key(&self, i: usize) -> OwnedInnerKey {
        let diff_key = self.block_diff_key(i);
        let mut buf = BytesMut::new();
        buf.extend_from_slice(self.common_prefix.chunk());
        buf.extend_from_slice(diff_key);
        OwnedInnerKey::new(buf.freeze())
    }

    fn size(&self) -> usize {
        // The four `Bytes` fields are slices of one encoded index block and
        // keep that whole allocation alive. Count the encoded block once, and
        // let the caller add `size_of::<Index>()` for the slice handles.
        self.encoded_len
    }
}

#[derive(Clone, Copy, Hash, PartialEq, Eq, Debug)]
pub struct BlockCacheKey {
    origin_id: u64,
    origin_off: u32,
}

impl BlockCacheKey {
    pub fn new(origin_id: u64, origin_off: u32) -> Self {
        Self {
            origin_id,
            origin_off,
        }
    }
}

const BLOCK_CACHE_KEY_SIZE: usize = std::mem::size_of::<BlockCacheKey>();
const BLOCK_CACHE_SHARDS_PER_CORE: usize = 64;
// Account for the `CacheItem` struct kept in the cache entry in addition to its
// payload.
const BLOCK_CACHE_VALUE_SIZE: usize = std::mem::size_of::<CacheItem>();

#[inline]
fn block_weight(_k: &BlockCacheKey, v: &CacheItem) -> usize {
    BLOCK_CACHE_KEY_SIZE + BLOCK_CACHE_VALUE_SIZE + v.size()
}

#[repr(u8)]
#[derive(Clone, Serialize, Deserialize, PartialEq, Debug, Copy, Default)]
#[serde(rename_all = "kebab-case")]
pub enum BlockCacheType {
    #[default]
    #[serde(alias = "moka")] // TODO: remove after compatibility period.
    Quick = 1,
    None = 2,
}

#[derive(Clone)]
enum CacheItem {
    Block(Bytes),
    Filter(Arc<BinaryFuse8>),
    Index(Arc<Index>),
}

impl CacheItem {
    fn size(&self) -> usize {
        match self {
            CacheItem::Block(bytes) => bytes.len(),
            CacheItem::Filter(filter) => {
                std::mem::size_of::<BinaryFuse8>() + filter.fingerprints.len()
            }
            CacheItem::Index(index) => std::mem::size_of::<Index>() + index.size(),
        }
    }
}

type QuickBlockCacheInner = quick_cache::sync::Cache<BlockCacheKey, CacheItem, BlockWeighter>;

/// Opaque handle for the quick-cache backed block cache.
#[derive(Clone)]
pub struct QuickBlockCache(Arc<QuickBlockCacheInner>);

#[derive(Clone)]
pub enum BlockCache {
    /// Quick cache implementation, which uses a synchronous cache for storing
    /// blocks, filters, and indexes.
    Quick(QuickBlockCache),
    /// No caching; all data will be read directly from the source.
    None,
}

impl BlockCache {
    pub fn new(tp: BlockCacheType, max_capacity: u64, block_size: usize) -> Self {
        if max_capacity == 0 {
            return BlockCache::None;
        }
        let block_cache_shards =
            (SysQuota::cpu_cores_quota() as usize).max(1) * BLOCK_CACHE_SHARDS_PER_CORE;
        match tp {
            BlockCacheType::Quick => {
                let opts = quick_cache::OptionsBuilder::new()
                    .shards(block_cache_shards)
                    .weight_capacity(max_capacity)
                    .estimated_items_capacity(
                        max_capacity as usize
                            / (BLOCK_CACHE_KEY_SIZE + BLOCK_CACHE_VALUE_SIZE + block_size),
                    )
                    .build()
                    .unwrap();
                let cache = quick_cache::sync::Cache::with_options(
                    opts,
                    BlockWeighter,
                    quick_cache::DefaultHashBuilder::default(),
                    quick_cache::sync::DefaultLifecycle::default(),
                );
                Self::Quick(QuickBlockCache(Arc::new(cache)))
            }
            BlockCacheType::None => Self::None,
        }
    }

    pub fn get(&self, key: &BlockCacheKey) -> Option<Bytes> {
        self.report_cache_lookup();
        match self {
            Self::Quick(cache) => match cache.0.get(key) {
                Some(CacheItem::Block(bytes)) => Some(bytes),
                _ => None,
            },
            Self::None => None,
        }
    }

    /// Returns a cached index without inserting on miss. Used by cache-neutral
    /// (fill_cache=false) reads so they reuse a hot index but never pin one.
    fn get_cached_index(&self, key: &BlockCacheKey) -> Option<Arc<Index>> {
        self.report_cache_lookup();
        match self {
            Self::Quick(cache) => match cache.0.get(key) {
                Some(CacheItem::Index(index)) => Some(index),
                _ => None,
            },
            Self::None => None,
        }
    }

    #[cfg(test)]
    fn contains_filter(&self, key: &BlockCacheKey) -> bool {
        match self {
            Self::Quick(cache) => matches!(cache.0.peek(key), Some(CacheItem::Filter(_))),
            Self::None => false,
        }
    }

    fn try_get_filter_with(
        &self,
        key: BlockCacheKey,
        init: impl FnOnce() -> Result<BinaryFuse8>,
    ) -> Result<Arc<BinaryFuse8>> {
        let init = || {
            self.report_cache_miss();
            init().map(|filter| CacheItem::Filter(Arc::new(filter)))
        };
        self.report_cache_lookup();
        let res = match self {
            Self::Quick(cache) => cache.0.get_or_insert_with(&key, init),
            Self::None => init(),
        };
        match res? {
            CacheItem::Filter(filter) => Ok(filter),
            _ => unreachable!(),
        }
    }

    #[cfg(test)]
    fn contains_index(&self, key: &BlockCacheKey) -> bool {
        match self {
            Self::Quick(cache) => matches!(cache.0.peek(key), Some(CacheItem::Index(_))),
            Self::None => false,
        }
    }

    fn try_get_index_with(
        &self,
        key: BlockCacheKey,
        init: impl FnOnce() -> Result<Index>,
    ) -> Result<Arc<Index>> {
        let init = || {
            self.report_cache_miss();
            init().map(|index| CacheItem::Index(Arc::new(index)))
        };
        self.report_cache_lookup();
        let res = match self {
            Self::Quick(cache) => cache.0.get_or_insert_with(&key, init),
            Self::None => init(),
        };
        match res? {
            CacheItem::Index(index) => Ok(index),
            _ => unreachable!(),
        }
    }

    pub fn try_get_with(
        &self,
        key: BlockCacheKey,
        init: impl FnOnce() -> Result<Bytes>,
    ) -> Result<Bytes> {
        let init = || {
            self.report_cache_miss();
            init().map(CacheItem::Block)
        };
        self.report_cache_lookup();
        let res = match self {
            Self::Quick(cache) => cache.0.get_or_insert_with(&key, init),
            Self::None => init(),
        };
        match res? {
            CacheItem::Block(bytes) => Ok(bytes),
            _ => unreachable!(),
        }
    }

    pub fn try_get_with_ext(
        &self,
        key: BlockCacheKey,
        init: impl FnOnce() -> Result<Bytes>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        if fill_cache {
            self.try_get_with(key, init)
        } else if let Some(block) = self.get(&key) {
            Ok(block)
        } else {
            self.report_cache_miss();
            init()
        }
    }

    pub async fn try_get_with_async(
        &self,
        key: BlockCacheKey,
        init: impl Future<Output = Result<Bytes>>,
    ) -> Result<Bytes> {
        let init = async {
            self.report_cache_miss();
            init.await.map(CacheItem::Block)
        };
        self.report_cache_lookup();
        let res = match self {
            Self::None => init.await,
            Self::Quick(cache) => cache.0.get_or_insert_async(&key, init).await,
        };
        match res? {
            CacheItem::Block(bytes) => Ok(bytes),
            _ => unreachable!(),
        }
    }

    pub async fn try_get_with_ext_async(
        &self,
        key: BlockCacheKey,
        init: impl Future<Output = Result<Bytes>>,
        fill_cache: bool,
    ) -> Result<Bytes> {
        if fill_cache {
            self.try_get_with_async(key, init).await
        } else if let Some(block) = self.get(&key) {
            Ok(block)
        } else {
            self.report_cache_miss();
            init.await
        }
    }

    pub fn weighted_size(&self) -> u64 {
        match self {
            Self::Quick(cache) => cache.0.weight(),
            Self::None => 0,
        }
    }

    pub fn capacity(&self) -> Option<u64> {
        match self {
            Self::Quick(cache) => Some(cache.0.capacity()),
            Self::None => None,
        }
    }

    pub fn set_capacity(&self, max_capacity: u64) -> std::result::Result<(), &'static str> {
        match self {
            Self::Quick(cache) => {
                cache.0.set_capacity(max_capacity);
                Ok(())
            }
            Self::None => Err("block cache is disabled"),
        }
    }

    #[inline]
    fn report_cache_lookup(&self) {
        if matches!(self, Self::Quick(_)) {
            crate::metrics::ENGINE_CACHE_LOOKUP.inc();
        }
    }

    fn report_cache_miss(&self) {
        match self {
            Self::Quick(_) => {
                crate::metrics::ENGINE_CACHE_MISS.inc_by(1);
                tracker::with_tls_tracker(|t| {
                    t.metrics.kv_engine_cache_miss += 1;
                });
            }
            Self::None => {}
        }
    }
}

#[derive(Clone)]
pub struct BlockWeighter;

impl quick_cache::Weighter<BlockCacheKey, CacheItem> for BlockWeighter {
    fn weight(&self, key: &BlockCacheKey, val: &CacheItem) -> u64 {
        block_weight(key, val) as u64
    }
}

pub(crate) fn validate_checksum(file_id: u64, data: &[u8], footer: &Footer) -> Result<()> {
    if data.len() < 4 {
        return Err(table::Error::InvalidChecksum(String::from(
            "data is too short",
        )));
    }
    let checksum_type: ChecksumType = footer.checksum_type.into();
    let checksum = LittleEndian::read_u32(data);
    let content = &data[4..];
    let got_checksum = checksum_type.checksum(content);
    if checksum != got_checksum {
        error!("checksum mismatch";
            "file" => file_id,
            "expect" => checksum,
            "got" => got_checksum,
            "footer" => ?footer,
        );
        return Err(table::Error::InvalidChecksum(format!(
            "{:?} checksum mismatch expect {} got {}",
            checksum_type, checksum, got_checksum
        )));
    }
    Ok(())
}

#[inline]
pub(crate) fn validate_checksum_with_fix(
    data: &[u8],
    footer: &Footer,
    file: &dyn File,
    encryption_state: &EncryptionState,
) -> Result<()> {
    match validate_checksum(file.id(), data, footer) {
        Ok(()) => Ok(()),
        Err(err) => {
            if let Some(file_path) = file.path() {
                warn!(
                    "file {} path {:?} failed to validate checksum, encryption_state {:?}, try to remove it from local disk",
                    file.id(),
                    file.path(),
                    encryption_state
                );
                // TODO: add `remove` to `File` to remove `IaFile`.
                // Just remove the sst file in local and download from dfs during next restart.
                std::fs::remove_file(file_path)
                    .table_ctx(file.id(), "sst.validate_checksum.remove_file")?;
            }
            Err(err)
        }
    }
}

const FILE_SUFFIX: &str = ".sst";

pub fn parse_file_id(path: &Path) -> Result<u64> {
    let name = path.file_name().unwrap().to_str().unwrap();
    if !name.ends_with(FILE_SUFFIX) {
        return Err(table::Error::InvalidFileName);
    }
    let digit_part = &name[..name.len() - FILE_SUFFIX.len()];
    if let Ok(id) = u64::from_str_radix(digit_part, 16) {
        return Ok(id);
    }
    Err(table::Error::InvalidFileName)
}

fn parse_prop_data(prop_data: &[u8]) -> Result<(&[u8], &[u8], &[u8])> {
    let key_len = prop_data
        .get(..2)
        .map(LittleEndian::read_u16)
        .ok_or(table::Error::InvalidFileSize)? as usize;
    let prop_data = prop_data.get(2..).ok_or(table::Error::InvalidFileSize)?;
    let key = prop_data
        .get(..key_len)
        .ok_or(table::Error::InvalidFileSize)?;
    let prop_data = prop_data
        .get(key_len..)
        .ok_or(table::Error::InvalidFileSize)?;
    let val_len = prop_data
        .get(..4)
        .map(LittleEndian::read_u32)
        .ok_or(table::Error::InvalidFileSize)? as usize;
    let prop_data = prop_data.get(4..).ok_or(table::Error::InvalidFileSize)?;
    let val = prop_data
        .get(..val_len)
        .ok_or(table::Error::InvalidFileSize)?;
    let remained = prop_data
        .get(val_len..)
        .ok_or(table::Error::InvalidFileSize)?;
    Ok((key, val, remained))
}

fn parse_u32_property(val: &[u8]) -> Result<u32> {
    if val.len() != std::mem::size_of::<u32>() {
        return Err(table::Error::InvalidFileSize);
    }
    Ok(LittleEndian::read_u32(val))
}

fn parse_u64_property(val: &[u8]) -> Result<u64> {
    if val.len() != std::mem::size_of::<u64>() {
        return Err(table::Error::InvalidFileSize);
    }
    Ok(LittleEndian::read_u64(val))
}

pub fn id_to_filename(id: u64) -> String {
    format!("{:016x}.sst", id)
}

pub fn new_filename(id: u64, dir: &Path) -> PathBuf {
    dir.join(id_to_filename(id))
}

#[derive(Default)]
pub struct NewSsTableCtx {
    // Input
    pub(crate) tiny_meta: Option<SstTinyMeta>,

    // Output
    pub(crate) footer_and_properties_data: Option<(Footer, Bytes)>,
}

impl NewSsTableCtx {
    pub fn take_footer_and_properties(&mut self) -> Option<(Footer, Bytes)> {
        self.tiny_meta
            .take()
            .and_then(|x| x.get_footer_and_properties().ok())
    }

    pub fn put_footer_and_properties(&mut self, footer: Footer, props_data: Bytes) {
        self.footer_and_properties_data = Some((footer, props_data));
    }
}

#[cfg(test)]
pub(crate) mod test_util {
    use std::sync::Arc;

    use cloud_encryption::EncryptionState;
    use rand::Rng;

    use crate::table::{
        ChecksumType, InnerKey, NO_COMPRESSION, Value, file,
        sstable::{BlockCache, BlockCacheType, Builder, SsTable},
    };

    pub(crate) static TEST_ID_ALLOC: std::sync::atomic::AtomicU64 =
        std::sync::atomic::AtomicU64::new(1);

    pub(crate) fn get_test_value(n: usize) -> String {
        format!("{}", n)
    }

    pub(crate) fn generate_key_values(prefix: &str, n: usize) -> Vec<(String, String)> {
        assert!(n <= 10000);
        let mut results = Vec::with_capacity(n);
        for i in 0..n {
            let k = get_test_key(prefix, i);
            let v = get_test_value(i);
            results.push((k, v));
        }
        results
    }

    #[maybe_async::both]
    pub(crate) async fn build_test_table_with_kvs(kvs: &Vec<(String, String)>) -> SsTable {
        let sst_fid = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_fid);
        let meta = 0u8;

        for (k, v) in kvs {
            let value_buf = Value::encode_buf(meta, &[0], 0, v.as_bytes());
            let value = &mut Value::decode(value_buf.as_slice());
            sst_builder
                .add(InnerKey::from_inner_buf(k.as_bytes()), value, None)
                .unwrap();
        }

        let mut buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut buf).unwrap();
        let sst_file = file::InMemFile::new(sst_fid, buf.into()).await;

        SsTable::new(
            Arc::new(sst_file),
            new_test_cache(),
            EncryptionState::Disabled,
        )
        .unwrap()
    }

    pub(crate) fn new_table_builder_for_test(sst_fid: u64) -> Builder {
        Builder::new(
            sst_fid,
            4096,
            NO_COMPRESSION,
            0,
            ChecksumType::default(),
            None,
        )
    }

    #[maybe_async::both]
    pub(crate) async fn build_test_table_with_prefix(
        prefix: &str,
        n: usize,
    ) -> (SsTable, Vec<(String, String)>) {
        let kvs = generate_key_values(prefix, n);
        (build_test_table_with_kvs(&kvs).await, kvs)
    }

    pub(crate) fn new_test_cache() -> BlockCache {
        BlockCache::new(BlockCacheType::Quick, 1024, 16)
    }

    pub(crate) fn get_test_key(prefix: &str, i: usize) -> String {
        format!("{}{:04}", prefix, i)
    }

    #[maybe_async::both]
    pub(crate) async fn create_sst_table(
        prefix: &str,
        n: usize,
    ) -> (SsTable, Vec<(String, String)>) {
        let kvs = generate_key_values(prefix, n);
        (build_test_table_with_kvs(&kvs).await, kvs)
    }

    // `kvs` must be sorted by key.
    #[maybe_async::both]
    pub(crate) async fn create_multi_version_sst(kvs: &[(String, String)]) -> (SsTable, usize) {
        let sst_fid = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_fid);
        let mut all_cnt = kvs.len();
        let meta = 0u8;
        for (k, v) in kvs {
            let val_buf = Value::encode_buf(meta, &[0], 9, v.as_bytes());
            sst_builder
                .add(
                    InnerKey::from_inner_buf(k.as_bytes()),
                    &Value::decode(val_buf.as_slice()),
                    None,
                )
                .unwrap();
            let mut r = rand::thread_rng();
            for i in (1..=8).rev() {
                // A lower probability than 1/8 to generate more entries with no old version.
                if r.gen_ratio(1, 10) {
                    let val_str = format!("{}_{}", v, i);
                    let val_buf = Value::encode_buf(meta, &[0], i, val_str.as_bytes());
                    sst_builder
                        .add(
                            InnerKey::from_inner_buf(k.as_bytes()),
                            &Value::decode(val_buf.as_slice()),
                            None,
                        )
                        .unwrap();
                    all_cnt += 1;
                }
            }
        }
        let mut sst_buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut sst_buf).unwrap();
        let sst_file = Arc::new(file::InMemFile::new(sst_fid, sst_buf.into()).await);
        (
            SsTable::new(sst_file, new_test_cache(), EncryptionState::Disabled).unwrap(),
            all_cnt,
        )
    }
}

#[cfg(test)]
mod tests {
    use std::{convert::TryFrom, iter::Iterator as StdIterator};

    use bytes::BytesMut;
    use futures::executor::block_on;
    use rand::Rng;
    use rstest::rstest;

    use super::{test_util::*, *};
    use crate::{Iterator, next, next_async};

    fn new_test_filter() -> BinaryFuse8 {
        let hashes: Vec<u64> = (0..100).collect();
        BinaryFuse8::try_from(&hashes).unwrap()
    }

    fn new_test_index(encoded_len: usize) -> Index {
        Index {
            common_prefix: Bytes::from_static(b"k"),
            block_key_offs: Bytes::from(vec![0; 4]),
            block_addrs: Bytes::from(vec![0; BLOCK_ADDR_SIZE]),
            block_keys: Bytes::from_static(b"k"),
            encoded_len,
        }
    }

    fn encode_test_property(key: &[u8], val: &[u8]) -> Vec<u8> {
        let mut encoded = Vec::with_capacity(2 + key.len() + 4 + val.len());
        encoded.extend_from_slice(&(key.len() as u16).to_le_bytes());
        encoded.extend_from_slice(key);
        encoded.extend_from_slice(&(val.len() as u32).to_le_bytes());
        encoded.extend_from_slice(val);
        encoded
    }

    fn rewrite_test_checksum(data: &mut [u8], start: usize, end: usize, checksum_type: u8) {
        let checksum = ChecksumType::from(checksum_type).checksum(&data[start + 4..end]);
        LittleEndian::write_u32(&mut data[start..start + 4], checksum);
    }

    #[test]
    fn test_index_rejects_malformed_encoding() {
        assert!(matches!(
            Index::new(Bytes::new()),
            Err(crate::table::Error::InvalidFileSize)
        ));

        let mut data = vec![0; 12];
        LittleEndian::write_u32(&mut data[4..8], INDEX_FORMAT_V1);
        LittleEndian::write_u32(&mut data[8..12], 1);
        assert!(matches!(
            Index::new(data.into()),
            Err(crate::table::Error::InvalidFileSize)
        ));

        let mut data = vec![0; 12];
        LittleEndian::write_u32(&mut data[4..8], INDEX_FORMAT_V1 + 1);
        assert!(matches!(
            Index::new(data.into()),
            Err(crate::table::Error::Other(_))
        ));
    }

    #[maybe_async::test]
    async fn test_sstable_rejects_malformed_properties_with_valid_checksum() {
        let file_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut builder = new_table_builder_for_test(file_id);
        let value_buf = Value::encode_buf(0, &[0], 1, b"value");
        builder
            .add(
                InnerKey::from_inner_buf(b"key"),
                &Value::decode(&value_buf),
                None,
            )
            .unwrap();
        let mut valid_data = Vec::with_capacity(builder.estimated_size());
        builder.finish(0, &mut valid_data).unwrap();

        let valid_footer_offset = valid_data.len() - FOOTER_SIZE;
        let mut footer = Footer::default();
        footer.unmarshal(&valid_data[valid_footer_offset..]);
        let properties_offset = footer.properties_offset as usize;

        let build_sst_with_properties = |properties: &[u8]| {
            let mut data = valid_data[..properties_offset].to_vec();
            data.extend_from_slice(&[0; 4]);
            data.extend_from_slice(properties);
            footer.marshal(&mut data);
            let footer_offset = data.len() - FOOTER_SIZE;
            rewrite_test_checksum(
                &mut data,
                properties_offset,
                footer_offset,
                footer.checksum_type,
            );
            validate_checksum(file_id, &data[properties_offset..footer_offset], &footer).unwrap();
            data
        };

        let malformed_for_both_readers = [
            vec![0],
            vec![1, 0],
            vec![1, 0, b'k'],
            vec![1, 0, b'k', 1, 0, 0],
            vec![1, 0, b'k', 2, 0, 0, 0, b'v'],
            encode_test_property(PROP_KEY_MAX_TS, &[0; 7]),
        ];

        for malformed in malformed_for_both_readers {
            let data = build_sst_with_properties(&malformed);
            let file = Arc::new(file::InMemFile::new(file_id, data.clone().into()).await);
            assert!(matches!(
                SsTable::new(file, BlockCache::None, EncryptionState::Disabled),
                Err(crate::table::Error::InvalidFileSize)
            ));

            let table_meta = SsTable::get_meta_data(&data).unwrap();
            let meta_file = file::InMemFile::new(file_id, table_meta).await;
            assert!(matches!(
                SsTableCore::extract_property_from_table_meta_file(&meta_file, PROP_KEY_MAX_TS),
                Err(crate::table::Error::InvalidFileSize)
            ));
        }

        // The MAX_TS extractor validates property framing and MAX_TS itself.
        // Fixed-width values of unrelated properties are owned by the full
        // SST parser.
        let malformed_entries = encode_test_property(PROP_KEY_ENTRIES, &[0; 3]);
        let data = build_sst_with_properties(&malformed_entries);
        let file = Arc::new(file::InMemFile::new(file_id, data.into()).await);
        assert!(matches!(
            SsTable::new(file, BlockCache::None, EncryptionState::Disabled),
            Err(crate::table::Error::InvalidFileSize)
        ));
    }

    #[maybe_async::test]
    async fn test_sstable_index_requires_strictly_increasing_block_keys() {
        let file_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut builder =
            Builder::new(file_id, 1, NO_COMPRESSION, 0, ChecksumType::default(), None);
        for key in [b"a".as_slice(), b"ab", b"ac"] {
            let value_buf = Value::encode_buf(0, &[0], 1, b"value");
            builder
                .add(
                    InnerKey::from_inner_buf(key),
                    &Value::decode(&value_buf),
                    None,
                )
                .unwrap();
        }
        let mut valid_data = Vec::with_capacity(builder.estimated_size());
        builder.finish(0, &mut valid_data).unwrap();

        let footer_offset = valid_data.len() - FOOTER_SIZE;
        let mut footer = Footer::default();
        footer.unmarshal(&valid_data[footer_offset..]);
        let index_start = footer.index_offset as usize;
        let index_end = footer.old_index_offset as usize;
        let num_blocks = LittleEndian::read_u32(&valid_data[index_start + 8..]) as usize;
        assert_eq!(num_blocks, 3);

        let key_offsets_start = index_start + 12;
        let block_addrs_start = key_offsets_start + num_blocks * 4;
        let common_prefix_len_offset = block_addrs_start + num_blocks * BLOCK_ADDR_SIZE;
        let common_prefix_len =
            LittleEndian::read_u16(&valid_data[common_prefix_len_offset..]) as usize;
        let block_keys_len_offset = common_prefix_len_offset + 2 + common_prefix_len;
        let block_keys_len = LittleEndian::read_u32(&valid_data[block_keys_len_offset..]) as usize;
        let block_keys_start = block_keys_len_offset + 4;
        assert_eq!(common_prefix_len, 1);
        assert_eq!(block_keys_len, 2);
        assert_eq!(
            [
                LittleEndian::read_u32(&valid_data[key_offsets_start..]),
                LittleEndian::read_u32(&valid_data[key_offsets_start + 4..]),
                LittleEndian::read_u32(&valid_data[key_offsets_start + 8..]),
            ],
            [0, 0, 1]
        );

        let valid_file = Arc::new(file::InMemFile::new(file_id, valid_data.clone().into()).await);
        let valid_table =
            SsTable::new(valid_file, BlockCache::None, EncryptionState::Disabled).unwrap();
        assert_eq!(valid_table.try_load_index(false).unwrap().num_blocks(), 3);

        let mut duplicate = valid_data.clone();
        LittleEndian::write_u32(&mut duplicate[key_offsets_start + 8..], 0);

        let mut reversed = valid_data.clone();
        reversed.swap(block_keys_start, block_keys_start + 1);

        for mut data in [duplicate, reversed] {
            rewrite_test_checksum(&mut data, index_start, index_end, footer.checksum_type);
            validate_checksum(file_id, &data[index_start..index_end], &footer).unwrap();

            let file = Arc::new(file::InMemFile::new(file_id, data.into()).await);
            let table = SsTable::new(file, BlockCache::None, EncryptionState::Disabled).unwrap();
            assert!(matches!(
                table.try_load_index(false),
                Err(crate::table::Error::InvalidFileSize)
            ));
        }
    }

    #[maybe_async::both]
    async fn build_test_table_with_cache(
        kvs: &Vec<(String, String)>,
        cache: BlockCache,
    ) -> SsTable {
        SsTable::new(
            build_test_table_file(kvs).await,
            cache,
            EncryptionState::Disabled,
        )
        .unwrap()
    }

    #[maybe_async::both]
    async fn build_test_table_file(kvs: &Vec<(String, String)>) -> Arc<file::InMemFile> {
        let sst_fid = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_fid);
        for (k, v) in kvs {
            let value_buf = Value::encode_buf(0, &[0], 0, v.as_bytes());
            let value = &mut Value::decode(value_buf.as_slice());
            sst_builder
                .add(InnerKey::from_inner_buf(k.as_bytes()), value, None)
                .unwrap();
        }

        let mut buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut buf).unwrap();
        Arc::new(file::InMemFile::new(sst_fid, buf.into()).await)
    }

    fn new_test_v2_encryption_state() -> EncryptionState {
        let key = EncryptionKey::new(
            vec![1; 32],
            vec![2; 32],
            cloud_encryption::KEY_TYPE_AES_256_CTR,
        )
        .with_key_id(1)
        .unwrap();
        assert_ne!(key.encryption_header(), 0);
        EncryptionState::Ready(
            key.with_keyspace_id(7)
                .with_key_manager(Arc::new(cloud_encryption::EncryptionKeyManager::new())),
        )
    }

    #[maybe_async::both]
    async fn build_test_multi_version_table_file(prefix: &str, n: usize) -> Arc<file::InMemFile> {
        let sst_fid = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_fid);
        for i in 0..n {
            let key = get_test_key(prefix, i);
            for version in (1..=2).rev() {
                let value = format!("{}_{}", get_test_value(i), version);
                let value_buf = Value::encode_buf(0, &[0], version, value.as_bytes());
                sst_builder
                    .add(
                        InnerKey::from_inner_buf(key.as_bytes()),
                        &Value::decode(value_buf.as_slice()),
                        None,
                    )
                    .unwrap();
            }
        }

        let mut buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut buf).unwrap();
        Arc::new(file::InMemFile::new(sst_fid, buf.into()).await)
    }

    #[test]
    fn test_quick_block_cache_set_capacity() {
        let cache = BlockCache::new(BlockCacheType::Quick, 1024, 16);
        assert_eq!(cache.capacity(), Some(1024));

        cache.set_capacity(2048).unwrap();
        assert_eq!(cache.capacity(), Some(2048));

        cache.set_capacity(512).unwrap();
        assert_eq!(cache.capacity(), Some(512));
    }

    #[test]
    fn test_block_cache_set_capacity_rejects_unsupported_modes() {
        let cache = BlockCache::new(BlockCacheType::None, 1024, 16);
        assert!(cache.set_capacity(2048).unwrap_err().contains("disabled"));
    }

    #[test]
    fn test_block_weight_counts_bytes_handle() {
        let key = BlockCacheKey::new(1, 2);
        let value = CacheItem::Block(Bytes::from_static(b"cached-block"));
        let expected = BLOCK_CACHE_KEY_SIZE + BLOCK_CACHE_VALUE_SIZE + value.size();

        assert_eq!(block_weight(&key, &value), expected);
        assert_eq!(
            quick_cache::Weighter::weight(&BlockWeighter, &key, &value),
            expected as u64
        );
    }

    #[test]
    fn test_block_weight_counts_filter_payload() {
        let key = BlockCacheKey::new(1, 2);
        let value = CacheItem::Filter(Arc::new(new_test_filter()));
        let expected = BLOCK_CACHE_KEY_SIZE + BLOCK_CACHE_VALUE_SIZE + value.size();

        assert_eq!(block_weight(&key, &value), expected);
        assert_eq!(
            quick_cache::Weighter::weight(&BlockWeighter, &key, &value),
            expected as u64
        );
    }

    #[test]
    fn test_block_weight_counts_index_payload() {
        let encoded_len = 128;
        let key = BlockCacheKey::new(1, 2);
        let value = CacheItem::Index(Arc::new(new_test_index(encoded_len)));
        let expected = BLOCK_CACHE_KEY_SIZE + BLOCK_CACHE_VALUE_SIZE + value.size();

        assert_eq!(value.size(), std::mem::size_of::<Index>() + encoded_len);
        assert_eq!(block_weight(&key, &value), expected);
        assert_eq!(
            quick_cache::Weighter::weight(&BlockWeighter, &key, &value),
            expected as u64
        );
    }

    #[test]
    fn test_block_cache_filter_item_is_reused() {
        let cache = BlockCache::new(BlockCacheType::Quick, 1024 * 1024, 16);
        let key = BlockCacheKey::new(1, 2);

        let first_filter = cache
            .try_get_filter_with(key, || Ok(new_test_filter()))
            .unwrap();

        let second_filter = cache
            .try_get_filter_with(key, || panic!("filter should be cached"))
            .unwrap();
        assert!(Arc::ptr_eq(&first_filter, &second_filter));
        assert!(cache.get(&key).is_none());
    }

    #[test]
    fn test_block_cache_none_does_not_retain_filter_item() {
        let cache = BlockCache::new(BlockCacheType::None, 1024 * 1024, 16);
        let key = BlockCacheKey::new(1, 2);

        let first_filter = cache
            .try_get_filter_with(key, || Ok(new_test_filter()))
            .unwrap();
        let second_filter = cache
            .try_get_filter_with(key, || Ok(new_test_filter()))
            .unwrap();

        assert!(!cache.contains_filter(&key));
        assert!(!Arc::ptr_eq(&first_filter, &second_filter));
    }

    #[test]
    fn test_block_cache_index_item_is_reused() {
        let cache = BlockCache::new(BlockCacheType::Quick, 1024 * 1024, 16);
        let key = BlockCacheKey::new(1, 2);

        let first_index = cache
            .try_get_index_with(key, || Ok(new_test_index(128)))
            .unwrap();

        let second_index = cache
            .try_get_index_with(key, || panic!("index should be cached"))
            .unwrap();
        assert!(Arc::ptr_eq(&first_index, &second_index));
        assert!(cache.get(&key).is_none());
    }

    #[test]
    fn test_block_cache_none_does_not_retain_index_item() {
        let cache = BlockCache::new(BlockCacheType::None, 1024 * 1024, 16);
        let key = BlockCacheKey::new(1, 2);

        let first_index = cache
            .try_get_index_with(key, || Ok(new_test_index(128)))
            .unwrap();
        let second_index = cache
            .try_get_index_with(key, || Ok(new_test_index(128)))
            .unwrap();

        assert!(!cache.contains_index(&key));
        assert!(!Arc::ptr_eq(&first_index, &second_index));
    }

    #[maybe_async::test]
    async fn test_sstable_filter_is_cached_in_block_cache() {
        let cache = BlockCache::new(BlockCacheType::Quick, 1024 * 1024, 16);
        let sst_file = build_test_table_file(&generate_key_values("key", 200)).await;
        let table =
            SsTable::new(sst_file.clone(), cache.clone(), EncryptionState::Disabled).unwrap();

        let filter = table.get_filter();

        let reopened = SsTable::new(sst_file, cache, EncryptionState::Disabled).unwrap();
        let reopened_filter = reopened.get_filter();
        assert!(Arc::ptr_eq(&filter, &reopened_filter));
    }

    #[maybe_async::test]
    async fn test_sstable_filter_is_not_retained_without_block_cache() {
        let table = build_test_table_with_cache(
            &generate_key_values("key", 200),
            BlockCache::new(BlockCacheType::None, 1024 * 1024, 16),
        )
        .await;

        let first_filter = table.get_filter();
        let second_filter = table.get_filter();

        assert!(!Arc::ptr_eq(&first_filter, &second_filter));
        assert!(first_filter.contains(&farmhash::fingerprint64(b"key0000")));
    }

    #[maybe_async::test]
    async fn test_sstable_indexes_are_cached_in_block_cache() {
        let cache = BlockCache::new(BlockCacheType::Quick, 1024 * 1024, 16);
        let sst_file = build_test_multi_version_table_file("key", 200).await;
        let table =
            SsTable::new(sst_file.clone(), cache.clone(), EncryptionState::Disabled).unwrap();

        let index = table.load_index(true);
        let old_index = table.load_old_index(true);
        assert!(old_index.num_blocks() > 0);

        let reopened = SsTable::new(sst_file, cache, EncryptionState::Disabled).unwrap();
        let reopened_index = reopened.load_index(true);
        let reopened_old_index = reopened.load_old_index(true);
        assert!(Arc::ptr_eq(&index, &reopened_index));
        assert!(Arc::ptr_eq(&old_index, &reopened_old_index));
    }

    #[maybe_async::test]
    async fn test_sstable_indexes_are_not_retained_without_block_cache() {
        let table = SsTable::new(
            build_test_multi_version_table_file("key", 200).await,
            BlockCache::new(BlockCacheType::None, 1024 * 1024, 16),
            EncryptionState::Disabled,
        )
        .unwrap();

        let first_index = table.load_index(true);
        let second_index = table.load_index(true);
        assert!(!Arc::ptr_eq(&first_index, &second_index));

        let first_old_index = table.load_old_index(true);
        let second_old_index = table.load_old_index(true);
        assert!(!Arc::ptr_eq(&first_old_index, &second_old_index));
    }

    #[maybe_async::test]
    async fn test_sstable_index_not_cached_when_fill_cache_false() {
        let table = SsTable::new(
            build_test_multi_version_table_file("key", 200).await,
            BlockCache::new(BlockCacheType::Quick, 1024 * 1024, 16),
            EncryptionState::Disabled,
        )
        .unwrap();

        // A cache-neutral load (fill_cache=false) must not retain the index in
        // the shared block cache: each call returns a fresh transient index.
        let first = table.try_load_index(false).unwrap();
        let second = table.try_load_index(false).unwrap();
        assert!(first.num_blocks() > 0);
        assert!(!Arc::ptr_eq(&first, &second));

        // fill_cache=true caches it; a later fill_cache=false reuses the cached
        // entry instead of inserting a new one.
        let cached = table.load_index(true);
        let reused = table.try_load_index(false).unwrap();
        assert!(Arc::ptr_eq(&cached, &reused));
    }

    #[maybe_async::test]
    async fn test_try_load_index_returns_checksum_error() {
        let file_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut builder = new_table_builder_for_test(file_id);
        for (key, value) in generate_key_values("key", 200) {
            let value_buf = Value::encode_buf(0, &[0], 0, value.as_bytes());
            builder
                .add(
                    InnerKey::from_inner_buf(key.as_bytes()),
                    &Value::decode(&value_buf),
                    None,
                )
                .unwrap();
        }
        let mut file_data = Vec::with_capacity(builder.estimated_size());
        builder.finish(0, &mut file_data).unwrap();

        let clean_file = Arc::new(file::InMemFile::new(file_id, file_data.clone().into()).await);
        let clean_table =
            SsTable::new(clean_file, BlockCache::None, EncryptionState::Disabled).unwrap();
        file_data[clean_table.index_offset() as usize] ^= 1;

        let corrupted_file = Arc::new(file::InMemFile::new(file_id, file_data.into()).await);
        let corrupted_table =
            SsTable::new(corrupted_file, BlockCache::None, EncryptionState::Disabled).unwrap();
        assert!(matches!(
            corrupted_table.try_load_index(false),
            Err(crate::table::Error::InvalidChecksum(_))
        ));
    }

    #[maybe_async::test]
    async fn test_sstable_rejects_reversed_index_offsets() {
        let file_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut builder = new_table_builder_for_test(file_id);
        for (key, value) in generate_key_values("key", 200) {
            let value_buf = Value::encode_buf(0, &[0], 0, value.as_bytes());
            builder
                .add(
                    InnerKey::from_inner_buf(key.as_bytes()),
                    &Value::decode(&value_buf),
                    None,
                )
                .unwrap();
        }
        let mut file_data = Vec::with_capacity(builder.estimated_size());
        builder.finish(0, &mut file_data).unwrap();

        let footer_offset = file_data.len() - FOOTER_SIZE;
        let mut footer = Footer::default();
        footer.unmarshal(&file_data[footer_offset..]);
        assert!(footer.index_offset > 0);
        footer.old_index_offset = footer.index_offset - 1;
        let mut footer_data = Vec::with_capacity(FOOTER_SIZE);
        footer.marshal(&mut footer_data);
        file_data[footer_offset..].copy_from_slice(&footer_data);

        let file = Arc::new(file::InMemFile::new(file_id, file_data.into()).await);
        assert!(matches!(
            SsTable::new(file, BlockCache::None, EncryptionState::Disabled),
            Err(crate::table::Error::InvalidFileSize)
        ));
    }

    #[maybe_async::test]
    async fn test_extract_property_rejects_invalid_footer_offsets() {
        let file_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut builder = new_table_builder_for_test(file_id);
        for (key, value) in generate_key_values("key", 200) {
            let value_buf = Value::encode_buf(0, &[0], 0, value.as_bytes());
            builder
                .add(
                    InnerKey::from_inner_buf(key.as_bytes()),
                    &Value::decode(&value_buf),
                    None,
                )
                .unwrap();
        }
        let mut file_data = Vec::with_capacity(builder.estimated_size());
        builder.finish(0, &mut file_data).unwrap();

        let table_meta = SsTable::get_meta_data(&file_data).unwrap();
        let footer_offset = table_meta.len() - FOOTER_SIZE;
        let mut footer = Footer::default();
        footer.unmarshal(&table_meta[footer_offset..]);

        let mut reversed = footer;
        assert!(reversed.index_offset > 0);
        reversed.properties_offset = reversed.index_offset - 1;

        let table_size = footer.index_offset as usize + table_meta.len();
        let max_offset = table_size - FOOTER_SIZE;
        let mut out_of_bounds = footer;
        out_of_bounds.properties_offset = u32::try_from(max_offset + 1).unwrap();
        assert!(out_of_bounds.aux_index_offset <= out_of_bounds.properties_offset);

        for invalid_footer in [reversed, out_of_bounds] {
            let mut corrupted_meta = table_meta.to_vec();
            let mut footer_data = Vec::with_capacity(FOOTER_SIZE);
            invalid_footer.marshal(&mut footer_data);
            corrupted_meta[footer_offset..].copy_from_slice(&footer_data);
            let file = file::InMemFile::new(file_id, corrupted_meta.into()).await;

            assert!(matches!(
                SsTableCore::extract_property_from_table_meta_file(&file, PROP_KEY_MAX_TS),
                Err(crate::table::Error::InvalidFileSize)
            ));
        }
    }

    #[maybe_async::test]
    async fn test_compaction_old_data_sstable_old_data_size_counts_old_version_blocks() {
        let sst_fid = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_fid);
        let key = InnerKey::from_inner_buf(b"same-key");
        for version in (1..=3).rev() {
            let value_buf = Value::encode_buf(0, &[0], version, b"value");
            sst_builder
                .add(key, &Value::decode(value_buf.as_slice()), None)
                .unwrap();
        }
        let mut sst_buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut sst_buf).unwrap();
        let sst_file = Arc::new(file::InMemFile::new(sst_fid, sst_buf.into()).await);

        let table = SsTable::new(sst_file, new_test_cache(), EncryptionState::Disabled).unwrap();

        assert_eq!(table.old_entries, 2);
        assert!(table.old_data_size() > 0);
        assert_eq!(table.old_data_size(), table.footer().old_data_len() as u64);
    }

    #[maybe_async::test]
    async fn test_table_iterator() {
        for n in 99..=101 {
            let (t, _) = create_sst_table("key", n).await;
            let mut it = t.new_iterator(false, true);
            let mut count = 0;
            it.rewind().await;
            while it.valid() {
                let k = it.key();
                assert_eq!(k.deref(), get_test_key("key", count).as_bytes());
                let v = it.value();
                assert_eq!(v.get_value(), get_test_value(count).as_bytes());
                count += 1;
                next!(it).await;
            }
        }
    }

    #[maybe_async::test]
    async fn test_point_get() {
        let (t, _) = create_sst_table("key", 8000).await;
        for i in 0..8000 {
            let k = get_test_key("key", i);
            let k_h = farmhash::fingerprint64(k.as_bytes());
            let mut owned_v = vec![];
            let v = t
                .get(
                    InnerKey::from_inner_buf(k.as_bytes()),
                    u64::MAX,
                    k_h,
                    &mut owned_v,
                    1,
                )
                .await;
            assert!(!v.is_empty())
        }
        for i in 8000..10000 {
            let k = get_test_key("key", i);
            let k_h = farmhash::fingerprint64(k.as_bytes());
            let mut owned_v = vec![];
            let v = t
                .get(
                    InnerKey::from_inner_buf(k.as_bytes()),
                    u64::MAX,
                    k_h,
                    &mut owned_v,
                    1,
                )
                .await;
            assert!(v.is_empty())
        }
    }

    #[maybe_async::test]
    async fn test_v2_key_reads_legacy_unencrypted_sst_header_zero_as_noop() {
        // Keep this compatibility test: enabling CMEK for API v2 must not make
        // older unencrypted SSTs unreadable. A zero encryption header still
        // means plaintext/noop even when the shard has a ready encryption key.
        let kvs = generate_key_values("key", 8);
        let table = SsTable::new(
            build_test_table_file(&kvs).await,
            BlockCache::None,
            new_test_v2_encryption_state(),
        )
        .unwrap();
        assert_eq!(table.encryption_ver(), 0);

        for (key, value) in kvs {
            let key_hash = farmhash::fingerprint64(key.as_bytes());
            let mut owned_value = vec![];
            let got = table
                .get(
                    InnerKey::from_inner_buf(key.as_bytes()),
                    u64::MAX,
                    key_hash,
                    &mut owned_value,
                    1,
                )
                .await;
            assert_eq!(got.get_value(), value.as_bytes());
        }
    }

    #[maybe_async::test]
    async fn test_seek_to_first() {
        let nums = &[99, 100, 101, 199, 200, 250, 9999, 10000];
        for n in nums {
            let (t, _) = create_sst_table("key", *n).await;
            let mut it = t.new_iterator(false, true);
            // Seek to the first element, if the iterator is not in reversed mode.
            it.rewind().await;
            assert!(it.valid());
            let v = it.value();
            assert_eq!(v.get_value(), get_test_value(0).as_bytes());
            assert_eq!(v.user_meta(), &[0u8]);
        }
    }

    struct TestData {
        input: &'static str,
        valid: bool,
        output: &'static str,
    }
    impl TestData {
        fn new(input: &'static str, valid: bool, output: &'static str) -> Self {
            Self {
                input,
                valid,
                output,
            }
        }
    }

    #[maybe_async::test]
    async fn test_seek_to_last() {
        let nums = vec![99, 100, 101, 199, 200, 250, 9999, 10000];
        for n in nums {
            let (t, _) = create_sst_table("key", n).await;
            let mut it = t.new_iterator(true, true);
            it.rewind().await;
            assert!(it.valid());
            let v = it.value();
            assert_eq!(v.get_value(), get_test_value(n - 1).as_bytes());
            assert!(!v.is_blob_ref());
            assert_eq!(v.user_meta(), &[0u8]);
            next!(it).await;
            assert!(it.valid());
            let v = it.value();
            assert_eq!(v.get_value(), get_test_value(n - 2).as_bytes());
            assert!(!v.is_blob_ref());
            assert_eq!(v.user_meta(), &[0u8]);
        }
    }

    #[maybe_async::test]
    async fn test_seek_basic() {
        let test_datas: Vec<TestData> = vec![
            TestData::new("abc", true, "k0000"),
            TestData::new("k0100", true, "k0100"),
            TestData::new("k0100b", true, "k0101"),
            TestData::new("k1234", true, "k1234"),
            TestData::new("k1234b", true, "k1235"),
            TestData::new("k9999", true, "k9999"),
            TestData::new("z", false, ""),
        ];
        let (t, _) = create_sst_table("k", 10000).await;
        let mut it = t.new_iterator(false, true);
        for td in test_datas {
            it.seek(InnerKey::from_inner_buf(td.input.as_bytes())).await;
            if !td.valid {
                assert!(!it.valid());
                continue;
            }
            assert!(it.valid());
            assert_eq!(it.key().deref(), td.output.as_bytes());
        }
    }

    #[maybe_async::test]
    async fn test_seek_for_prev() {
        let test_datas: Vec<TestData> = vec![
            TestData::new("abc", false, ""),
            TestData::new("k0100", true, "k0100"),
            TestData::new("k0100b", true, "k0100"),
            TestData::new("k1234", true, "k1234"),
            TestData::new("k1234b", true, "k1234"),
            TestData::new("k9999", true, "k9999"),
            TestData::new("z", true, "k9999"),
        ];
        let (t, _) = create_sst_table("k", 10000).await;
        let mut it = t.new_iterator(true, true);
        for td in test_datas {
            it.seek(InnerKey::from_inner_buf(td.input.as_bytes())).await;
            if !td.valid {
                assert!(!it.valid());
                continue;
            }
            assert!(it.valid());
            assert_eq!(it.key().deref(), td.output.as_bytes());
        }
    }

    #[maybe_async::test]
    async fn test_iterate_from_start() {
        let nums = vec![99, 100, 101, 199, 200, 250, 9999, 10000];
        for n in nums {
            let (t, _) = create_sst_table("key", n).await;
            let mut it = t.new_iterator(false, true);
            let mut count = 0;
            it.rewind().await;
            assert!(it.valid());
            while it.valid() {
                let k = it.key();
                assert_eq!(k.deref(), get_test_key("key", count).as_bytes());
                let v = it.value();
                assert_eq!(v.get_value(), get_test_value(count).as_bytes());
                assert!(!v.is_blob_ref());
                count += 1;
                next!(it).await
            }
        }
    }

    #[maybe_async::test]
    async fn test_iterate_from_end() {
        let nums = vec![99, 100, 101, 199, 200, 250, 9999, 10000];
        for n in nums {
            let (t, _) = create_sst_table("key", n).await;
            let mut it = t.new_iterator(true, true);
            it.seek(InnerKey::from_inner_buf("zzzzzz".as_bytes())).await; // Seek to end, an invalid element.
            assert!(it.valid());
            it.rewind().await;
            for i in (0..n).rev() {
                assert!(it.valid());
                let v = it.value();
                assert_eq!(v.get_value(), get_test_value(i).as_bytes());
                assert!(!v.is_blob_ref());
                next!(it).await;
            }
            next!(it).await;
            assert!(!it.valid());
        }
    }

    #[maybe_async::test]
    async fn test_table() {
        let (t, _) = create_sst_table("key", 10000).await;
        let mut it = t.new_iterator(false, true);
        let mut kid = 1010_usize;
        let seek = get_test_key("key", kid);
        it.seek(InnerKey::from_inner_buf(seek.as_bytes())).await;
        while it.valid() {
            assert_eq!(it.key().deref(), get_test_key("key", kid).as_bytes());
            kid += 1;
            next!(it).await;
        }
        assert_eq!(kid, 10000);

        it.seek(InnerKey::from_inner_buf(
            get_test_key("key", 99999).as_bytes(),
        ))
        .await;
        assert!(!it.valid());

        it.seek(InnerKey::from_inner_buf(get_test_key("kex", 0).as_bytes()))
            .await;
        assert!(it.valid());
        assert_eq!(it.key().deref(), get_test_key("key", 0).as_bytes());
    }

    #[maybe_async::test]
    async fn test_iterate_back_and_forth() {
        let (t, _) = create_sst_table("key", 10000).await;
        let seek = get_test_key("key", 1010);
        let mut it = t.new_iterator(false, true);
        it.seek(InnerKey::from_inner_buf(seek.as_bytes())).await;
        assert!(it.valid());
        assert_eq!(it.key().deref(), seek.as_bytes());

        it.set_reversed(true);
        next!(it).await;
        next!(it).await;
        assert!(it.valid());
        assert_eq!(it.key().deref(), get_test_key("key", 1008).as_bytes());

        it.set_reversed(false);
        next!(it).await;
        next!(it).await;
        assert_eq!(it.valid(), true);
        assert_eq!(it.key().deref(), get_test_key("key", 1010).as_bytes());

        it.seek(InnerKey::from_inner_buf(
            get_test_key("key", 2000).as_bytes(),
        ))
        .await;
        assert_eq!(it.valid(), true);
        assert_eq!(it.key().deref(), get_test_key("key", 2000).as_bytes());

        it.set_reversed(true);
        next!(it).await;
        assert_eq!(it.valid(), true);
        assert_eq!(it.key().deref(), get_test_key("key", 1999).as_bytes());

        it.set_reversed(false);
        it.rewind().await;
        assert_eq!(it.key().deref(), get_test_key("key", 0).as_bytes());
    }

    #[maybe_async::test]
    async fn test_iterate_multi_version() {
        let num = 4000;
        let kvs = generate_key_values("key", num);
        let (t, all_cnt) = create_multi_version_sst(&kvs).await;
        let mut it = t.new_iterator(false, true);
        let mut it_cnt = 0;
        let mut last_key = BytesMut::new();
        it.rewind().await;
        while it.valid() {
            if !last_key.is_empty() {
                assert!(last_key < it.key().deref());
            }
            last_key.truncate(0);
            last_key.extend_from_slice(it.key().deref());
            it_cnt += 1;
            while next_version!(it).await {
                it_cnt += 1;
            }
            next!(it).await;
        }
        assert_eq!(it_cnt, all_cnt);
        let mut r = rand::thread_rng();
        for _ in 0..1000 {
            let k = get_test_key("key", r.gen_range(0..num));
            let ver = 5 + r.gen_range(0..5) as u64;
            let k_h = farmhash::fingerprint64(k.as_bytes());
            let mut owned_v = vec![];
            let val = t
                .get(
                    InnerKey::from_inner_buf(k.as_bytes()),
                    ver,
                    k_h,
                    &mut owned_v,
                    1,
                )
                .await;
            if !val.is_empty() {
                assert!(val.version <= ver);
            }
        }
        let mut rev_it = t.new_iterator(true, true);
        last_key.truncate(0);
        rev_it.rewind().await;
        while rev_it.valid() {
            if !last_key.is_empty() {
                assert!(last_key > rev_it.key().deref());
            }
            last_key.truncate(0);
            last_key.extend_from_slice(rev_it.key().deref());
            next!(rev_it).await;
        }
        for _ in 0..1000 {
            let k = get_test_key("key", r.gen_range(0..num));
            // reverse iterator never seek to the same key with smaller version.
            rev_it.seek(InnerKey::from_inner_buf(k.as_bytes())).await;
            if !rev_it.valid() {
                continue;
            }
            assert_eq!(rev_it.value().version, 9);
            assert!(rev_it.key().deref() <= k.as_bytes());
        }
    }

    #[maybe_async::test]
    async fn test_uni_iterator() {
        let (t, _) = create_sst_table("key", 10000).await;
        {
            let mut it = t.new_iterator(false, true);
            let mut cnt = 0;
            it.rewind().await;
            while it.valid() {
                let v = it.value();
                assert_eq!(v.get_value(), get_test_value(cnt).as_bytes());
                assert!(!v.is_blob_ref());
                cnt += 1;
                next!(it).await;
            }
            assert_eq!(cnt, 10000);
        }
        {
            let mut it = t.new_iterator(true, true);
            let mut cnt = 0;
            it.rewind().await;
            while it.valid() {
                let v = it.value();
                assert_eq!(v.get_value(), get_test_value(10000 - 1 - cnt).as_bytes());
                assert!(!v.is_blob_ref());
                cnt += 1;
                next!(it).await;
            }
        }
    }

    // For:
    // https://github.com/tidbcloud/cloud-storage-engine/issues/1957
    // https://github.com/tidbcloud/cloud-storage-engine/issues/2062
    #[rstest]
    #[case::n1(1)]
    #[case::n10(10)]
    fn test_reset_old_block_iter(#[case] n: usize) {
        test_reset_old_block_iter_impl(n);
        block_on(test_reset_old_block_iter_impl_async(n));
    }

    #[maybe_async::both]
    async fn test_reset_old_block_iter_impl(n: usize) {
        let kvs = generate_key_values("key", n);
        let (t, _) = create_multi_version_sst(&kvs).await;
        let mut it = t.new_iterator(false, true);

        for (k, _) in kvs {
            it.seek(InnerKey::from_inner_buf(k.as_bytes())).await;
            assert!(it.valid());
            next_version!(it).await;
            next_version!(it).await; // Reach the second version in old block.

            it.seek(InnerKey::from_inner_buf(k.as_bytes())).await;
            assert!(it.valid());
            // If old block iterator is not reset, it will point to the second version and
            // panic for version not match.
            next_version!(it).await;
        }
    }

    #[tokio::test]
    async fn test_is_next_version_sync_on_not_valid() {
        // No old version.
        {
            let (t, _) = create_sst_table_async("key", 10).await;
            let mut it = t.new_iterator(false, true);
            it.rewind_async().await;
            assert!(it.is_next_version_sync());
            assert!(!next_version_async!(it));
        }

        // No more version.
        {
            let kvs = generate_key_values("key", 10);
            let (t, _) = create_multi_version_sst_async(&kvs).await;
            let mut it = t.new_iterator(false, true);
            it.rewind_async().await;
            while next_version_async!(it) {}
            assert!(it.is_next_version_sync());
            assert!(!next_version_async!(it));
        }
    }

    #[bench]
    fn bench_decode_filter(b: &mut test::Bencher) {
        let (t, _) = create_sst_table("key", 10000);
        let data = t.read_filter_data_from_file().unwrap();
        b.iter(|| {
            test::black_box(t.decode_filter(&data).unwrap());
        });
    }

    #[maybe_async::test]
    async fn test_has_overlap() {
        // k0000 ~ k0009
        let (t, _) = create_sst_table("k", 10).await;

        // lower_bound, upper_bound, upper_inclusive, expected
        let cases: Vec<(&'static str, &'static str, bool, bool)> = vec![
            ("k0", "k00", false, false),
            ("k000", "k0000", false, false),
            ("k000", "k0000", true, true),
            ("k0000", "k00000", false, true),
            ("k00000", "k00001", false, false), // inner no overlap.
            ("k0001", "k00010", false, true),
            ("k00010", "k0002", false, false), // inner no overlap.
            ("k00010", "k0002", true, true),
            ("k0000", "k0009", false, true),
            ("k00090", "k00091", true, false),
        ];

        for (lower, upper, upper_inclusive, expected) in cases {
            let lower_bound = InnerKey::from_inner_buf(lower.as_bytes());
            let upper_bound = InnerKey::from_inner_buf(upper.as_bytes());
            let data_bound = DataBound {
                lower_bound,
                upper_bound,
                upper_inclusive,
            };
            let has_overlap = t.has_overlap(data_bound).await;
            assert_eq!(
                has_overlap, expected,
                "case: {} {} {} {}",
                lower, upper, upper_inclusive, expected
            );
        }
    }

    #[tokio::test]
    async fn test_has_overlap_loose() {
        // k0000 ~ k0009
        let (t_sync, _) = create_sst_table("k", 10);
        let (t_async, _) = create_sst_table_async("k", 10).await;

        // lower_bound, upper_bound, upper_inclusive, sync_expected, async_expected
        let cases: Vec<(&'static str, &'static str, bool, bool, bool)> = vec![
            ("k0", "k00", false, false, false),
            ("k000", "k0000", false, false, false),
            ("k000", "k0000", true, true, true),
            ("k0000", "k00000", false, true, true),
            ("k00000", "k00001", false, false, true), // inner no overlap.
            ("k0001", "k00010", false, true, true),
            ("k00010", "k0002", false, false, true), // inner no overlap.
            ("k00010", "k0002", true, true, true),
            ("k0000", "k0009", false, true, true),
            ("k00090", "k00091", true, false, false),
        ];

        for (lower, upper, upper_inclusive, sync_expected, async_expected) in cases {
            let lower_bound = InnerKey::from_inner_buf(lower.as_bytes());
            let upper_bound = InnerKey::from_inner_buf(upper.as_bytes());
            let data_bound = DataBound {
                lower_bound,
                upper_bound,
                upper_inclusive,
            };

            let has_overlap = t_sync.has_overlap_loose(data_bound);
            assert_eq!(
                has_overlap, sync_expected,
                "sync case: {} {} {} {}",
                lower, upper, upper_inclusive, sync_expected
            );

            let has_overlap = t_async.has_overlap_loose(data_bound);
            assert_eq!(
                has_overlap, async_expected,
                "async case: {} {} {} {}",
                lower, upper, upper_inclusive, async_expected
            );
        }
    }

    #[test]
    fn test_get_meta_data() {
        let mut sst_builder = new_table_builder_for_test(100);
        for (k, v) in generate_key_values("k", 10) {
            let value_buf = Value::encode_buf(0u8, &[0], 0, v.as_bytes());
            let value = &mut Value::decode(value_buf.as_slice());
            sst_builder
                .add(InnerKey::from_inner_buf(k.as_bytes()), value, None)
                .unwrap();
        }

        let mut buf = Vec::with_capacity(sst_builder.estimated_size());
        sst_builder.finish(0, &mut buf).unwrap();
        let buf = Bytes::from(buf);
        let sst_file = file::InMemFile::new(100, buf.clone());
        let sst = SsTable::new(
            Arc::new(sst_file),
            new_test_cache(),
            EncryptionState::Disabled,
        )
        .unwrap();

        let meta_data = SsTable::get_meta_data(&buf).unwrap();
        assert_eq!(meta_data.as_ref(), &buf[sst.meta_offset() as usize..]);

        SsTable::get_meta_data(&[]).unwrap_err();
        SsTable::get_meta_data(&[0xff; SsTable::footer_size()]).unwrap_err();
    }

    #[test]
    fn test_deprecated_moka_block_cache_type_still_uses_cache() {
        let cfg: crate::config::Config = serde_json::from_str(
            r#"
            {"block-cache-type":"moka"}
            "#,
        )
        .unwrap();
        assert_eq!(cfg.block_cache_type, BlockCacheType::Quick);
        assert!(matches!(
            BlockCache::new(cfg.block_cache_type, 1024, 16),
            BlockCache::Quick(_)
        ));
    }
}