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

use core::panic;
use std::{
    collections::{HashMap, HashSet, VecDeque},
    fmt::{Debug, Formatter},
    iter::Iterator as _,
    marker::PhantomData,
    ops::Deref,
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    time::Duration,
};

use api_version::{ApiV2, api_v2::KEYSPACE_PREFIX_LEN};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use cloud_encryption::{EncryptionKey, KEY_TYPE_AES_256_CTR_LEGACY, ToEncryptionState};
use futures::{StreamExt, TryStreamExt, stream};
use kvenginepb as pb;
use kvenginepb::TxnFileRefs;
use log_wrappers::Value as LogValue;
use protobuf::Message;
use schema::schema::StorageClassSpec;
use serde::{Deserialize, Serialize};
use tidb_query_datatype::{
    FieldTypeFlag,
    codec::{
        mysql::VectorFloat32Decoder,
        table::{decode_common_handle, decode_int_handle, decode_table_id},
    },
};
use tikv_util::{
    box_try,
    codec::number::{U32_SIZE, U64_SIZE},
    memory::{MemoryLimiter, MemoryLimiterGuard},
    mpsc,
    time::{Instant, duration_to_sec},
};
use tipb::{AnnQueryInfo, ColumnInfo, FtsQueryInfo};
use tokio::runtime::Handle;
use txn_types::LockOrSharedLocks;

use crate::{
    context::{IaCtx, SnapCtx},
    dfs::FileType,
    ia::types::FileSegmentIdent,
    limiter::RegionLimiter,
    metrics::*,
    table::{
        AsyncMergeIterator, BoundedDataSet, DataBound, EntryChecker, InnerKey, IterStats,
        Iterator as TableIterator, OwnedInnerKey, SkipOpTxnFileIterator, SnapVersion, TxnFile,
        TxnFileIterator, Value,
        blobtable::{
            BlobRef,
            blobtable::{BlobPrefetcher, BlobTable},
        },
        columnar::{
            Block, ColumnarConcatReader, ColumnarFile, ColumnarFilterReader, ColumnarMergeReader,
            ColumnarMvccReader, ColumnarPackCleanReader, ColumnarReader, ColumnarRowTableReader,
            ColumnarRuntimeStats, ColumnarTableReader, GLOBAL_COMMON_HANDLE_END, HANDLE_COL_ID,
            HybridBatch, HybridMaterializationPlan, HybridMergeReader, HybridMvccReader,
            HybridSourceReader, RowMvccReader, VERSION_COL_ID, decode_exclusive_end_int_handle,
            filter::TableScanCtx,
        },
        fts::{
            FtsBruteForceCondReader, FtsBruteForceReader, FtsDeltaCache, FtsJoinReader,
            validate_schema,
        },
        memtable::{CfTable, Hint, SkipList, WriteBatch, skl_ext::SkipListExt},
        schema_file::{Schema, SchemaBuf, SchemaFile},
        sstable::{L0Table, SsTable},
        table,
        txn_file::{OP_CHECK_NOT_EXIST, OP_INSERT},
        vector_index::{VectorDistanceProjector, VectorItemsReader},
    },
    table_id::encode_table_prefix_key,
    txn_chunk_manager::TxnChunkManager,
    *,
};

/// Mem data format V1: format_ver,skls_data
const MEM_DATA_FORMAT_V1: u32 = 1;
/// Mem data format V2:
/// format_ver,skls_length,skls_data,txn_file_refs_length,txn_file_refs_data
const MEM_DATA_FORMAT_V2: u32 = 2;
/// Mem data format V3:
/// format_ver,cf_count,(cf,skls_length,skls_data)*,txn_file_refs_length,
/// txn_file_refs_data.
///
/// Unlike V1/V2, V3 stores raw entries (including tombstones) for every CF.
pub const ONLINE_SNAPSHOT_MEM_DATA_FORMAT_VERSION: u32 = 3;

#[derive(Serialize, Deserialize)]
struct MemTableRowV3 {
    /// Inner key. The target shard supplies its own API-v2 keyspace prefix.
    key: Vec<u8>,
    meta: u8,
    user_meta: Vec<u8>,
    version: u64,
    value: Vec<u8>,
}

pub struct Item<'a> {
    val: table::Value,
    pub path: AccessPath,
    phantom: PhantomData<&'a i32>,

    // Uses to hold the value's memory when necessary, so that the life time of val can be at least
    // long as the life time of Item itself. This is necessary when the caller is not responsible
    // (or impossible) to manage the life time. e.g. During point get, caller does not hold the
    // memory as in scan (held by the iterator).
    owned_val: Option<Vec<u8>>,
    owned_blob: Option<Vec<u8>>,
}

impl std::ops::Deref for Item<'_> {
    type Target = table::Value;

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

impl Item<'_> {
    fn new() -> Self {
        Self {
            val: table::Value::new(),
            path: AccessPath::default(),
            phantom: Default::default(),
            owned_val: None,
            owned_blob: None,
        }
    }

    pub fn from_cached_value(cached: &ValueCacheValue) -> Self {
        let um = UserMeta::new(cached.start_ts, cached.version);
        let um_buf = um.to_array();
        let buf = Value::encode_buf(0, &um_buf, cached.version, cached.value.chunk());
        let val = Value::decode(&buf);
        Self {
            val,
            path: Default::default(),
            phantom: Default::default(),
            owned_val: Some(buf),
            owned_blob: None,
        }
    }

    /// Check if this item exists (has user metadata)
    #[inline]
    pub fn exists(&self) -> bool {
        self.user_meta_len() > 0
    }
}

impl Default for Item<'_> {
    fn default() -> Self {
        Self::new()
    }
}

#[derive(Default, Debug, Clone, Copy)]
pub struct AccessPath {
    pub mem_table: u8,
    pub l0: u8,
    pub ln: u8,
}

#[derive(Default, Debug, Clone, Copy)]
pub struct TsFilterStats {
    pub candidate_memtables: usize,
    pub candidate_l0s: usize,
    pub candidate_stables: usize,
    pub pruned_memtables_by_max_ts: usize,
    pub pruned_l0s_by_max_ts: usize,
    pub pruned_stables_by_max_ts: usize,
    pub surviving_memtables: usize,
    pub surviving_l0s: usize,
    pub surviving_stables: usize,
}

/// Timestamp-filtered iterators grouped by their I/O behavior.
///
/// Memtable and L0 iterators are always synchronous. Keeping them separate
/// from stable-level iterators allows async consumers to merge all synchronous
/// sources behind one always-sync child.
#[derive(Default)]
struct TsFilteredCfIters {
    sync_iters: Vec<Box<dyn table::Iterator>>,
    stable_iters: Vec<Box<dyn table::Iterator>>,
}

impl TsFilteredCfIters {
    fn into_iters(mut self) -> Vec<Box<dyn table::Iterator>> {
        self.sync_iters.append(&mut self.stable_iters);
        self.sync_iters
    }
}

#[derive(Clone)]
pub struct SnapAccess {
    pub core: Arc<SnapAccessCore>,
}

impl SnapAccess {
    pub fn new(shard: &Shard) -> Self {
        let core = Arc::new(SnapAccessCore::new(shard));
        Self { core }
    }

    /// Test-only constructor that injects an [`FtsDeltaCache`] into the
    /// snapshot.
    #[cfg(test)]
    pub fn new_with_fts_delta_cache(shard: &Shard, fts_delta_cache: FtsDeltaCache) -> Self {
        let mut core = SnapAccessCore::new(shard);
        core.fts_delta_cache = fts_delta_cache;
        Self {
            core: Arc::new(core),
        }
    }

    /// Constructs a snapshot from a prepared change set.
    ///
    /// This helper does not perform the key-readiness gate. Callers that will
    /// access encrypted data should use [`Self::construct_snapshot`] instead.
    pub async fn from_change_set(
        tag: &str,
        ctx: &SnapCtx,
        change_set: pb::ChangeSet,
        write_cf_only: bool,
    ) -> Result<Self> {
        let core = Arc::new(
            SnapAccessCore::from_change_set(tag, ctx, change_set, vec![], write_cf_only).await?,
        );
        Ok(Self { core })
    }

    async fn from_change_set_and_memtable_data(
        tag: &str,
        ctx: &SnapCtx,
        change_set: pb::ChangeSet,
        mem_tbls: Vec<CfTable>,
    ) -> Result<Self> {
        let core =
            Arc::new(SnapAccessCore::from_change_set(tag, ctx, change_set, mem_tbls, true).await?);
        Ok(Self { core })
    }

    pub async fn construct_snapshot(
        tag: &str,
        ctx: &SnapCtx,
        mem_table_data: &[u8],
        snapshot: &[u8],
        memory_limiter: MemoryLimiter,
    ) -> Result<(Self, MemoryLimiterGuard)> {
        let mut change_set = kvenginepb::ChangeSet::default();
        change_set.merge_from_bytes(snapshot).unwrap();
        if !change_set.has_snapshot() {
            error!("no snapshot in change set: {:?}", change_set);
            return Err(Error::RemoteRead("no snapshot in change set".to_string()));
        }

        let snap = change_set.get_snapshot();

        let mut tables_size = mem_table_data.len();
        if matches!(ctx.ia_ctx, IaCtx::Disabled(_)) {
            tables_size += estimate_tables_size_from_snapshot(snap, ctx.prepare_type);
        }
        let mem_limiter_guard = match memory_limiter.acquire(tables_size as u64) {
            Ok(guard) => guard,
            Err(exceeded_size) => {
                warn!("{} memory limit exceeded", tag;
                    "tables_size" => tables_size, "exceeded" => exceeded_size,
                    "limiter" => ?memory_limiter);
                return Err(Error::MemoryLimitExceeded(exceeded_size));
            }
        };

        let encryption_key = encryption_key_from_shard_properties(
            snap.get_properties(),
            ctx.encryption_key_manager.clone(),
            &ctx.master_key,
            ApiV2::get_u32_keyspace_id_by_key(snap.get_outer_start()),
        )
        .map_err(|_| {
            Error::EncryptionKeyUnavailable(format!(
                "[CMEK] KMS key is unavailable for remote snapshot {}; reject read before accessing encrypted data",
                tag
            ))
        })?;

        let shard_id = change_set.shard_id;
        let shard_ver = change_set.shard_ver;
        let mem_tbls = Self::construct_memtables(
            shard_id,
            shard_ver,
            mem_table_data,
            &ctx.txn_chunk_manager,
            encryption_key.clone(),
        )
        .await?;
        let snap = Self::from_change_set_and_memtable_data(tag, ctx, change_set, mem_tbls).await?;
        Ok((snap, mem_limiter_guard))
    }

    pub async fn construct_memtables(
        shard_id: u64,
        shard_ver: u64,
        mut mem_table_data: &[u8],
        txn_chunk_manager: &TxnChunkManager,
        encryption_key: Option<EncryptionKey>,
    ) -> Result<Vec<CfTable>> {
        if mem_table_data.is_empty() {
            return Ok(vec![]);
        }

        let format_version = mem_table_data.get_u32_le();
        if format_version == MEM_DATA_FORMAT_V1 {
            let mut mem_tbl = CfTable::new();
            Self::construct_skip_list(mem_table_data, &mut mem_tbl, WRITE_CF)?;
            Ok(vec![mem_tbl])
        } else if format_version == MEM_DATA_FORMAT_V2 {
            Self::construct_memtables_format_v2(
                shard_id,
                shard_ver,
                mem_table_data,
                txn_chunk_manager,
                encryption_key,
            )
            .await
        } else if format_version == ONLINE_SNAPSHOT_MEM_DATA_FORMAT_VERSION {
            Self::construct_memtables_format_v3(
                shard_id,
                shard_ver,
                mem_table_data,
                txn_chunk_manager,
                encryption_key,
            )
            .await
        } else {
            Err(Error::RemoteRead(format!(
                "unsupported mem data format {}",
                format_version
            )))
        }
    }

    fn construct_skip_list(skl_data: &[u8], mem_tbl: &mut CfTable, cf: usize) -> Result<()> {
        let mut wb = WriteBatch::new();
        let rows: Vec<table::Row> = bincode::deserialize(skl_data).map_err(|e| {
            Error::RemoteRead(format!("failed to deserialize mem table data: {}", e))
        })?;
        for row in rows {
            let key = InnerKey::from_outer_key(&row.key);
            wb.put(
                key,
                0,
                &row.user_meta.to_array(),
                row.user_meta.commit_ts,
                &row.value,
            );
        }
        mem_tbl.get_cf(cf).put_batch(&mut wb, None, cf);
        Ok(())
    }

    fn construct_write_batch_v3(skl_data: &[u8]) -> Result<WriteBatch> {
        let rows: Vec<MemTableRowV3> = bincode::deserialize(skl_data).map_err(|e| {
            Error::RemoteRead(format!("failed to deserialize V3 mem table data: {}", e))
        })?;
        let mut wb = WriteBatch::new();
        for row in rows {
            wb.put(
                InnerKey::from_inner_buf(&row.key),
                row.meta,
                &row.user_meta,
                row.version,
                &row.value,
            );
        }
        Ok(wb)
    }

    async fn load_txn_files(
        shard_id: u64,
        shard_ver: u64,
        data: &[u8],
        txn_chunk_manager: &TxnChunkManager,
        encryption_key: Option<EncryptionKey>,
    ) -> Result<Vec<TxnFile>> {
        let mut refs = TxnFileRefs::default();
        box_try!(refs.merge_from_bytes(data));
        if refs.get_txn_file_refs().is_empty() {
            return Ok(vec![]);
        }
        let worker_pool = txn_chunk_manager.worker_pool().clone();
        let txn_chunk_manager = txn_chunk_manager.clone();
        worker_pool
            .spawn_blocking(move || {
                txn_chunk_manager.load_txn_files_from_refs(
                    shard_id,
                    shard_ver,
                    refs.get_txn_file_refs(),
                    encryption_key.to_encryption_state(),
                    IoType::ForegroundRead,
                )
            })
            .await
            .unwrap()
    }

    async fn construct_memtables_format_v2(
        shard_id: u64,
        shard_ver: u64,
        mut mem_data: &[u8],
        txn_chunk_manager: &TxnChunkManager,
        encryption_key: Option<EncryptionKey>,
    ) -> Result<Vec<CfTable>> {
        let mut mem_tbl = CfTable::new();

        // Skiplists:
        let mem_size = mem_data.get_u64_le() as usize;
        let skl_data = &mem_data[..mem_size];
        mem_data.advance(mem_size);
        Self::construct_skip_list(skl_data, &mut mem_tbl, WRITE_CF)?;

        // Txn file refs:
        let msg_len = mem_data.get_u32_le() as usize;
        let msg_data = &mem_data[..msg_len];
        mem_data.advance(msg_len);

        let txn_files = Self::load_txn_files(
            shard_id,
            shard_ver,
            msg_data,
            txn_chunk_manager,
            encryption_key,
        )
        .await?;
        if !txn_files.is_empty() {
            mem_tbl = mem_tbl.add_write_cf_txn_files(&txn_files);
        }

        debug_assert!(
            mem_data.is_empty(),
            "mem_data: {:?}",
            LogValue::value(mem_data)
        );
        Ok(vec![mem_tbl])
    }

    async fn construct_memtables_format_v3(
        shard_id: u64,
        shard_ver: u64,
        mut mem_data: &[u8],
        txn_chunk_manager: &TxnChunkManager,
        encryption_key: Option<EncryptionKey>,
    ) -> Result<Vec<CfTable>> {
        if mem_data.remaining() < U32_SIZE {
            return Err(Error::RemoteRead(
                "truncated V3 mem table CF count".to_string(),
            ));
        }
        let cf_count = mem_data.get_u32_le() as usize;
        if cf_count != NUM_CFS {
            return Err(Error::RemoteRead(format!(
                "unexpected V3 mem table CF count {}, expected {}",
                cf_count, NUM_CFS
            )));
        }

        let mut seen_cfs = [false; NUM_CFS];
        let mut write_batches: [Option<WriteBatch>; NUM_CFS] = std::array::from_fn(|_| None);
        for _ in 0..cf_count {
            if mem_data.remaining() < U32_SIZE + U64_SIZE {
                return Err(Error::RemoteRead(
                    "truncated V3 mem table CF header".to_string(),
                ));
            }
            let cf = mem_data.get_u32_le() as usize;
            if cf >= NUM_CFS || seen_cfs[cf] {
                return Err(Error::RemoteRead(format!(
                    "invalid or duplicated V3 mem table CF {}",
                    cf
                )));
            }
            seen_cfs[cf] = true;
            let mem_size = mem_data.get_u64_le() as usize;
            if mem_data.remaining() < mem_size {
                return Err(Error::RemoteRead(format!(
                    "truncated V3 mem table CF {} payload",
                    cf
                )));
            }
            let skl_data = &mem_data[..mem_size];
            mem_data.advance(mem_size);
            write_batches[cf] = Some(Self::construct_write_batch_v3(skl_data)?);
        }

        let mut mem_tbl = CfTable::new();
        for (cf, batch) in IntoIterator::into_iter(write_batches).enumerate() {
            let mut batch = batch
                .ok_or_else(|| Error::RemoteRead(format!("missing V3 mem table CF {}", cf)))?;
            mem_tbl
                .get_cf(cf)
                .put_batch_preserve_tombstones(&mut batch, cf);
        }

        if mem_data.remaining() < U32_SIZE {
            return Err(Error::RemoteRead(
                "truncated V3 txn file refs length".to_string(),
            ));
        }
        let msg_len = mem_data.get_u32_le() as usize;
        if mem_data.remaining() < msg_len {
            return Err(Error::RemoteRead(
                "truncated V3 txn file refs payload".to_string(),
            ));
        }
        let msg_data = &mem_data[..msg_len];
        mem_data.advance(msg_len);

        let txn_files = Self::load_txn_files(
            shard_id,
            shard_ver,
            msg_data,
            txn_chunk_manager,
            encryption_key,
        )
        .await?;
        if !txn_files.is_empty() {
            mem_tbl = mem_tbl.add_write_cf_txn_files(&txn_files);
        }

        if !mem_data.is_empty() {
            return Err(Error::RemoteRead(format!(
                "unexpected {} trailing bytes in V3 mem table data",
                mem_data.len()
            )));
        }
        Ok(vec![mem_tbl])
    }
}

#[cfg(any(test, feature = "testexport"))]
impl SnapAccess {
    pub fn display(&self) -> pb::ChangeSet {
        let outer_range = (self.data.outer_start.clone(), self.data.outer_end.clone());
        self.to_change_set(&[outer_range], false)
    }

    pub fn write_cf_level_n_is_empty(&self) -> bool {
        self.data
            .get_cf(WRITE_CF)
            .levels
            .iter()
            .all(|lh| lh.tables.is_empty())
    }
}

impl Deref for SnapAccess {
    type Target = SnapAccessCore;

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

impl Debug for SnapAccess {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "snap access {}, seq: {}", self.tag, self.write_sequence,)
    }
}

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

pub struct SnapAccessCore {
    tag: ShardTag,
    managed_ts: u64,
    base_version: u64,
    meta_seq: u64,
    write_sequence: u64,
    data: ShardData,
    get_hint: Mutex<Hint>,
    blob_table_prefetch_size: usize,
    deleting_prefixes: Arc<DeletePrefixes>,
    encryption_key: Option<EncryptionKey>,
    encryption_meta_bytes: Option<Bytes>,
    storage_class_spec: StorageClassSpec,
    fts_delta_cache: FtsDeltaCache,
    is_sync: bool,
    read_columnar: bool,
}

impl SnapAccessCore {
    pub fn new(shard: &Shard) -> Self {
        let base_version = shard.get_base_version();
        let meta_seq = shard.get_meta_sequence();
        let write_sequence = shard.get_write_sequence();
        let data = shard.get_data();
        let is_sync = data.is_sync();
        let storage_class_spec = shard.get_storage_class_spec();
        Self {
            tag: shard.tag(),
            write_sequence,
            meta_seq,
            base_version,
            managed_ts: 0,
            data,
            get_hint: Mutex::new(Hint::new()),
            blob_table_prefetch_size: shard.opt.blob_prefetch_size,
            deleting_prefixes: shard.get_del_prefixes(),
            // Production callers must enter through `try_new_snap_access`, which
            // ensures the key is ready before constructing `SnapAccessCore`.
            encryption_key: shard.try_get_encryption_key().unwrap_or_else(|err| {
                panic!(
                    "[CMEK] {} failed to create snapshot because KMS key is unavailable: {}",
                    shard.tag(),
                    err
                )
            }),
            encryption_meta_bytes: shard.get_property(ENCRYPTION_META_KEY),
            storage_class_spec,
            fts_delta_cache: FtsDeltaCache::disabled(),
            is_sync,
            read_columnar: shard.opt.read_columnar,
        }
    }

    pub async fn from_change_set(
        tag: &str,
        ctx: &SnapCtx,
        change_set: pb::ChangeSet,
        mem_tbls: Vec<CfTable>,
        write_cf_only: bool,
    ) -> Result<Self> {
        let shard = Shard::from_change_set(tag, ctx, change_set, mem_tbls, write_cf_only).await?;
        let mut ret = Self::new(&shard);
        ret.fts_delta_cache = ctx.fts_delta_cache.clone();
        Ok(ret)
    }

    #[maybe_async::both]
    pub async fn new_iterator_skip_blob(
        &self,
        cf: usize,
        reversed: bool,
        all_versions: bool,
        read_ts: Option<u64>,
        fill_cache: bool,
    ) -> Iterator {
        let read_ts = if let Some(ts) = read_ts {
            ts
        } else if CF_MANAGED[cf] && self.managed_ts != 0 {
            self.managed_ts
        } else {
            u64::MAX
        };
        let data = self.data.clone();
        let mut key = BytesMut::new();
        key.extend_from_slice(data.keyspace_prefix());
        Iterator {
            all_versions,
            reversed,
            read_ts,
            met_newer_ts: false,
            key,
            val: table::Value::new(),
            skipped_tombstones: 0,
            inner: self
                .new_table_iterator(cf, reversed, fill_cache, None)
                .await,
            iter_stats: IterStats::default(),
            is_blob_value_record: false,
            blob_prefetcher: None,
            owned_blob: None,
            encryption_key: self.encryption_key.clone(),
            skip_blob: true,
            data,
            range: None,
        }
    }

    #[maybe_async::both]
    pub async fn new_iterator(
        &self,
        cf: usize,
        reversed: bool,
        all_versions: bool,
        read_ts: Option<u64>,
        fill_cache: bool,
    ) -> Iterator {
        let blob_prefetcher = if self.data.blob_tbl_map.is_empty()
            || self.data.blob_tbl_map.values().any(|tbl| !tbl.is_sync())
        {
            None
        } else {
            Some(BlobPrefetcher::new(
                self.data.blob_tbl_map.clone(),
                self.blob_table_prefetch_size,
                self.encryption_key.clone(),
            ))
        };
        let data = self.data.clone();
        let mut key = BytesMut::new();
        key.extend_from_slice(data.keyspace_prefix());
        Iterator {
            all_versions,
            reversed,
            read_ts: self.get_read_ts(cf, read_ts),
            met_newer_ts: false,
            key,
            val: table::Value::new(),
            skipped_tombstones: 0,
            inner: self
                .new_table_iterator(cf, reversed, fill_cache, None)
                .await,
            iter_stats: IterStats::default(),
            is_blob_value_record: false,
            blob_prefetcher,
            owned_blob: None,
            encryption_key: self.encryption_key.clone(),
            skip_blob: false,
            data,
            range: None,
        }
    }

    pub fn new_memtable_iterator(
        &self,
        cf: usize,
        reversed: bool,
        all_versions: bool,
        read_ts: Option<u64>,
    ) -> Iterator {
        let data = self.data.clone();
        let mut key = BytesMut::new();
        key.extend_from_slice(data.keyspace_prefix());
        Iterator {
            all_versions,
            reversed,
            read_ts: self.get_read_ts(cf, read_ts),
            met_newer_ts: false,
            key,
            val: table::Value::new(),
            skipped_tombstones: 0,
            inner: self.new_mem_table_iterator(cf, reversed),
            iter_stats: IterStats::default(),
            is_blob_value_record: false,
            blob_prefetcher: None,
            owned_blob: None,
            encryption_key: self.encryption_key.clone(),
            skip_blob: true,
            data,
            range: None,
        }
    }

    pub fn new_delta_iterator(&self, reversed: bool, since_ts: u64) -> Iterator {
        let blob_prefetcher = Some(BlobPrefetcher::new(
            self.data.blob_tbl_map.clone(),
            self.blob_table_prefetch_size,
            self.encryption_key.clone(),
        ));
        let data = self.data.clone();
        let mut key = BytesMut::new();
        key.extend_from_slice(data.keyspace_prefix());
        Iterator {
            all_versions: true,
            reversed,
            read_ts: self.get_read_ts(WRITE_CF, None),
            met_newer_ts: false,
            key,
            val: table::Value::new(),
            skipped_tombstones: 0,
            inner: self.new_delta_write_iterator(since_ts),
            iter_stats: IterStats::default(),
            is_blob_value_record: false,
            blob_prefetcher,
            owned_blob: None,
            encryption_key: self.encryption_key.clone(),
            skip_blob: false,
            data,
            range: None,
        }
    }

    fn get_read_ts(&self, cf: usize, read_ts: Option<u64>) -> u64 {
        if let Some(ts) = read_ts {
            ts
        } else if CF_MANAGED[cf] && self.managed_ts != 0 {
            self.managed_ts
        } else if !CF_MANAGED[cf] {
            self.get_mem_table_snap_version().into_inner()
        } else {
            u64::MAX
        }
    }

    #[maybe_async::both]
    async fn fetch_blob(&self, key: InnerKey<'_>, blob_ref: BlobRef) -> Vec<u8> {
        let blob_table = self
            .data
            .blob_tbl_map
            .get(&blob_ref.fid)
            .unwrap_or_else(|| {
                panic!(
                    "[{}] blob table not found {:?}, blob table id: {}",
                    self.tag, key, blob_ref.fid
                )
            });
        // TODO: avoid reallocation
        let mut decryption_buf = vec![];
        blob_table
            .get_by_ref(&blob_ref, &mut decryption_buf, self.encryption_key.clone())
            .await
            .unwrap_or_else(|e| panic!("[{}] blob table get failed {:?} {:?}", self.tag, key, e))
    }

    /// get an Item by key. Caller need to call is_some() before get_value.
    /// We don't return Option because we may need AccessPath even if the item
    /// is none.
    #[maybe_async::both]
    pub async fn get(&self, cf: usize, key: &[u8], version: u64) -> Item<'_> {
        let mut version = version;
        if version == 0 {
            version = u64::MAX;
        }

        let inner_key = InnerKey::from_outer_key(key);
        let mut item = Item::new();
        item.owned_val = Some(vec![]);
        item.val = self
            .get_value(
                cf,
                inner_key,
                version,
                &mut item.path,
                item.owned_val.as_mut().unwrap(),
                &self.data.lock_txn_files,
            )
            .await;
        if item.val.is_blob_ref() {
            item.owned_blob = Some(self.fetch_blob(inner_key, item.val.get_blob_ref()).await);
            item.val.fill_in_blob(item.owned_blob.as_ref().unwrap());
        }
        item
    }

    pub fn get_non_txn_file_lock(&self, key: &[u8]) -> Item<'_> {
        let inner_key = InnerKey::from_outer_key(key);
        let mut item = Item::new();
        item.owned_val = Some(vec![]);
        item.val = self.get_value(
            LOCK_CF,
            inner_key,
            u64::MAX,
            &mut item.path,
            item.owned_val.as_mut().unwrap(),
            &[],
        );
        item
    }

    #[maybe_async::both]
    async fn get_value(
        &self,
        cf: usize,
        inner_key: InnerKey<'_>,
        version: u64,
        path: &mut AccessPath,
        out_val_owner: &mut Vec<u8>,
        with_lock_txn_files: &[TxnFile],
    ) -> table::Value {
        self.check_sync(cf).await;
        if cf == LOCK_CF {
            for txn_file in with_lock_txn_files {
                if txn_file.version() > version {
                    continue;
                }
                let (_, val) = txn_file.get_value(inner_key, out_val_owner);
                if val.is_valid() {
                    return val;
                }
            }
        }
        for (index, mem_table) in self.data.mem_tbls.iter().enumerate() {
            let tbl = mem_table.get_cf(cf);
            let v = if index == 0 && cf == WRITE_CF && tbl.supports_get_hint() {
                let mut hint = self.get_hint.lock().unwrap();
                tbl.get_with_hint(inner_key, version, &mut hint, out_val_owner)
            } else {
                tbl.get(inner_key, version, out_val_owner)
            };
            path.mem_table = path.mem_table.saturating_add(1);
            if v.is_valid() {
                return v;
            }
        }
        let key_hash = farmhash::fingerprint64(inner_key.deref());
        for l0 in &self.data.l0_tbls {
            if let Some(tbl) = &l0.get_cf(cf) {
                if inner_key < tbl.smallest() || tbl.biggest() < inner_key {
                    continue;
                }
                let v = tbl.get(inner_key, version, key_hash, out_val_owner, 0);
                path.l0 = path.l0.saturating_add(1);
                if v.is_valid() {
                    return v;
                }
            }
        }
        let scf = self.data.get_cf(cf);
        for lh in &scf.levels {
            #[allow(clippy::if_same_then_else)]
            let v = if cf == WRITE_CF {
                lh.get(inner_key, version, key_hash, out_val_owner).await
            } else {
                lh.get(inner_key, version, key_hash, out_val_owner)
            };
            path.ln += 1;
            if v.is_valid() {
                return v;
            }
        }
        table::Value::new()
    }

    #[maybe_async::both]
    pub async fn multi_get(&self, cf: usize, keys: &[Vec<u8>], version: u64) -> Vec<Item<'_>> {
        let mut items = Vec::with_capacity(keys.len());
        for key in keys {
            let item = self.get(cf, key, version).await;
            items.push(item);
        }
        items
    }

    pub fn set_managed_ts(&mut self, managed_ts: u64) {
        self.managed_ts = managed_ts;
    }

    fn new_table_iterator(
        &self,
        cf: usize,
        reversed: bool,
        fill_cache: bool,
        skip_txn_file_with_start_ts: Option<u64>,
    ) -> Box<dyn table::Iterator> {
        self.check_sync(cf);
        let mut iters: Vec<Box<dyn table::Iterator>> = Vec::new();
        if cf == LOCK_CF && !self.data.lock_txn_files.is_empty() {
            for txn_file in &self.data.lock_txn_files {
                if skip_txn_file_with_start_ts
                    .is_some_and(|start_ts| txn_file.start_ts() == start_ts)
                {
                    continue;
                }
                let txn_file_iter = TxnFileIterator::new(txn_file.clone(), reversed);
                let skip_op_iter = SkipOpTxnFileIterator::new(txn_file_iter, false, true);
                iters.push(Box::new(skip_op_iter))
            }
        }
        for mem_tbl in &self.data.mem_tbls {
            iters.push(mem_tbl.get_cf(cf).new_iterator(reversed));
        }
        for l0 in &self.data.l0_tbls {
            if let Some(tbl) = &l0.get_cf(cf) {
                iters.push(tbl.new_iterator(reversed, fill_cache));
            }
        }
        let scf = self.data.get_cf(cf);
        for lh in scf.levels.as_slice() {
            if lh.tables.is_empty() {
                continue;
            }
            if lh.tables.len() == 1 {
                iters.push(lh.tables[0].new_iterator(reversed, fill_cache));
                continue;
            }
            iters.push(Box::new(ConcatIterator::new(
                lh.clone(),
                reversed,
                fill_cache,
            )));
        }
        table::new_merge_iterator(iters, reversed)
    }

    // This method is actually sync but mark as async to corporate with
    // `maybe_async`.
    async fn new_table_iterator_async(
        &self,
        cf: usize,
        reversed: bool,
        fill_cache: bool,
        skip_txn_file_with_start_ts: Option<u64>,
    ) -> Box<dyn table::Iterator> {
        if cf != WRITE_CF {
            return self.new_table_iterator(cf, reversed, fill_cache, skip_txn_file_with_start_ts);
        }

        let sync_merge_iter: Box<dyn table::Iterator> = {
            let mut sync_iters: Vec<Box<dyn table::Iterator>> =
                Vec::with_capacity(self.data.mem_tbls.len() + self.data.l0_tbls.len());
            for mem_tbl in &self.data.mem_tbls {
                sync_iters.push(mem_tbl.get_cf(cf).new_iterator(reversed));
            }
            for l0 in &self.data.l0_tbls {
                if let Some(tbl) = &l0.get_cf(cf) {
                    sync_iters.push(tbl.new_iterator(reversed, fill_cache));
                }
            }
            Box::new(AsyncMergeIterator::new(sync_iters, reversed, true)) as _
        };

        let mut iters = vec![sync_merge_iter];

        let scf = self.data.get_cf(cf);
        for lh in scf.levels.as_slice() {
            if lh.tables.is_empty() {
                continue;
            }
            if lh.tables.len() == 1 {
                iters.push(lh.tables[0].new_iterator(reversed, fill_cache));
                continue;
            }
            iters.push(Box::new(ConcatIterator::new(
                lh.clone(),
                reversed,
                fill_cache,
            )));
        }
        Box::new(AsyncMergeIterator::new(iters, reversed, false))
    }

    pub fn new_mem_table_iterator(&self, cf: usize, reversed: bool) -> Box<dyn table::Iterator> {
        let mut iters: Vec<Box<dyn table::Iterator>> = Vec::new();
        for mem_tbl in &self.data.mem_tbls {
            iters.push(mem_tbl.get_cf(cf).new_iterator(reversed));
        }
        table::new_merge_iterator(iters, reversed)
    }

    /// Create a delta iterator of WRITE_CF that iterates over all entries with
    /// ts > since_ts.
    ///
    /// NOTE: The iterator will still contain entries with ts <= since_ts.
    /// Callers need to filter them out.
    #[maybe_async::both]
    pub async fn new_delta_write_iterator(&self, since_ts: u64) -> Box<dyn table::Iterator> {
        self.check_sync(WRITE_CF).await;
        let mut stats = TsFilterStats::default();
        let TsFilteredCfIters {
            sync_iters,
            stable_iters,
        } = self.collect_cf_iters_above_ts(
            WRITE_CF,
            since_ts,
            None,
            true,
            |skl: &SkipListExt, ts| skl.new_delta_write_iterator(ts),
            &mut stats,
        );

        let sync_merge_iter: Box<dyn table::Iterator> =
            Box::new(AsyncMergeIterator::new(sync_iters, false, true));
        let mut iters = Vec::with_capacity(stable_iters.len() + 1);
        iters.push(sync_merge_iter);
        iters.extend(stable_iters);
        Box::new(AsyncMergeIterator::new(iters, false, false))
    }

    /// Build a key-bound EXTRA_CF iterator over records whose value version
    /// may be greater than `threshold`.
    ///
    /// `threshold` is an exclusive lower bound in EXTRA_CF value-version space:
    /// normal records use `commit_ts`; rollback records use `start_ts`.
    pub fn build_pruned_extra_cf_iter(
        &self,
        raw_key: &[u8],
        threshold: u64,
        fill_cache: bool,
        stats: &mut TsFilterStats,
    ) -> Box<dyn table::Iterator> {
        self.check_sync(EXTRA_CF);

        let prefix_bound = ExtraCfPrefixBound::new(raw_key);
        let data_bound = prefix_bound.as_data_bound();
        let iters = self
            .collect_cf_iters_above_ts(
                EXTRA_CF,
                threshold,
                Some(&data_bound),
                fill_cache,
                |skl: &SkipListExt, _| skl.new_iterator(false),
                stats,
            )
            .into_iters();
        table::new_merge_iterator(iters, false)
    }

    fn collect_cf_iters_above_ts(
        &self,
        cf: usize,
        ts_threshold: u64,
        key_bound: Option<&DataBound<'_>>,
        fill_cache: bool,
        make_mem_iter: impl Fn(&SkipListExt, u64) -> Box<dyn table::Iterator>,
        stats: &mut TsFilterStats,
    ) -> TsFilteredCfIters {
        let mut iters = TsFilteredCfIters::default();

        for (i, mem_tbl) in self.data.mem_tbls.iter().enumerate() {
            let cf_skl = mem_tbl.get_cf(cf);
            if cf_skl.is_empty() {
                continue;
            }
            stats.candidate_memtables += 1;
            // The writable memtable (index 0) is never pruned: `data_max_ts`
            // is updated after entries are inserted into the skiplist, so a
            // concurrent reader could observe a stale upper bound.  Frozen
            // memtables (index > 0) are immutable and safe to prune.
            if i > 0 && cf_skl.data_max_ts() <= ts_threshold {
                stats.pruned_memtables_by_max_ts += 1;
                continue;
            }
            stats.surviving_memtables += 1;
            iters.sync_iters.push(make_mem_iter(cf_skl, ts_threshold));
        }

        for l0 in &self.data.l0_tbls {
            let Some(tbl) = l0.get_cf(cf) else {
                continue;
            };
            if let Some(bound) = key_bound {
                let bound =
                    DataBound::new(bound.lower_bound, bound.upper_bound, bound.upper_inclusive);
                if !tbl.data_bound().overlap_bound(bound) {
                    continue;
                }
            }
            stats.candidate_l0s += 1;
            if tbl.max_ts <= ts_threshold {
                stats.pruned_l0s_by_max_ts += 1;
                continue;
            }
            stats.surviving_l0s += 1;
            iters.sync_iters.push(tbl.new_iterator(false, fill_cache));
        }

        let scf = self.data.get_cf(cf);
        for lh in scf.levels.as_slice() {
            if lh.tables.is_empty() {
                continue;
            }
            let (left, right) = if let Some(bound) = key_bound {
                bound.get_overlap_data_sets(&lh.tables)
            } else {
                (0, lh.tables.len())
            };
            if left == right {
                continue;
            }
            if lh.max_ts <= ts_threshold {
                let overlap_count = right - left;
                stats.candidate_stables += overlap_count;
                stats.pruned_stables_by_max_ts += overlap_count;
                continue;
            }
            let mut level_tables = Vec::with_capacity(right.saturating_sub(left));
            for tbl in &lh.tables[left..right] {
                stats.candidate_stables += 1;
                if tbl.max_ts <= ts_threshold {
                    stats.pruned_stables_by_max_ts += 1;
                    continue;
                }
                stats.surviving_stables += 1;
                level_tables.push(tbl.clone());
            }
            match level_tables.len() {
                0 => {}
                1 => iters
                    .stable_iters
                    .push(level_tables.pop().unwrap().new_iterator(false, fill_cache)),
                _ => iters
                    .stable_iters
                    .push(Box::new(ConcatIterator::new_with_tables(
                        level_tables,
                        false,
                        fill_cache,
                    ))),
            }
        }

        iters
    }

    pub fn iter_tables(&self, cf: usize, mut f: impl FnMut(TableRef<'_>)) {
        self.check_sync(cf);

        for mem_tbl in &self.data.mem_tbls {
            if !mem_tbl.get_cf(cf).is_empty() {
                f(TableRef::Memtable(mem_tbl));
            }
        }
        for l0 in &self.data.l0_tbls {
            if l0.get_cf(cf).is_some() {
                f(TableRef::L0(l0))
            }
        }
        let scf = self.data.get_cf(cf);
        for lh in scf.levels.as_slice() {
            if lh.tables.is_empty() {
                continue;
            }
            for tbl in &*lh.tables {
                f(TableRef::SsTable(tbl))
            }
        }
    }

    pub fn get_write_sequence(&self) -> u64 {
        self.write_sequence
    }

    pub fn get_mem_table_snap_version(&self) -> SnapVersion {
        SnapVersion::new(self.base_version, self.write_sequence)
    }

    pub fn get_start_key(&self) -> &[u8] {
        self.data.outer_start.chunk()
    }

    pub fn get_end_key(&self) -> &[u8] {
        self.data.outer_end.chunk()
    }

    pub fn key_is_in_range(&self, key: &[u8]) -> bool {
        self.get_start_key() <= key && key < self.get_end_key()
    }

    pub fn get_inner_start(&self) -> InnerKey<'_> {
        self.data.inner_start()
    }

    pub fn get_inner_end(&self) -> InnerKey<'_> {
        self.data.inner_end()
    }

    pub fn clone_start_key(&self) -> Bytes {
        self.data.outer_start.clone()
    }

    pub fn clone_end_key(&self) -> Bytes {
        self.data.outer_end.clone()
    }

    pub fn get_inner_key_offset(&self) -> usize {
        self.data.inner_key_off
    }

    pub fn prepend_keyspace_id(&self) -> Option<u32> {
        self.data.prepend_keyspace_id()
    }

    pub fn get_tag(&self) -> ShardTag {
        self.tag
    }

    pub fn get_id(&self) -> u64 {
        self.tag.id_ver.id
    }

    pub fn get_version(&self) -> u64 {
        self.tag.id_ver.ver
    }

    pub fn get_meta_seq(&self) -> u64 {
        self.meta_seq
    }

    pub fn get_columnar_levels(&self) -> Vec<(ColumnarFile, usize)> {
        self.data
            .col_levels
            .levels
            .iter()
            .flat_map(|l| {
                l.files
                    .iter()
                    .map(|f| (f.clone(), l.level))
                    .collect::<Vec<_>>()
            })
            .collect()
    }

    // Sync only.
    pub(crate) fn contains_in_older_table(&self, key: InnerKey<'_>, cf: usize) -> bool {
        self.check_sync(cf);
        let key_hash = farmhash::fingerprint64(key.deref());
        let mut outer_val_owner = vec![];
        for tbl in &self.data.mem_tbls[1..] {
            let val = tbl.get_cf(cf).get(key, u64::MAX, &mut outer_val_owner);
            if val.is_valid() {
                return !val.is_deleted();
            }
        }
        for l0 in &self.data.l0_tbls {
            let l0_cf = l0.get_cf(cf);
            if l0_cf.is_none() {
                continue;
            }
            let l0_cf = l0_cf.as_ref().unwrap();
            let mut owned_val = vec![];
            let val = l0_cf.get(key, u64::MAX, key_hash, &mut owned_val, 0);
            if val.is_valid() {
                return !val.is_deleted();
            }
        }
        for l in self.data.get_cf(cf).levels.as_slice() {
            if let Some(tbl) = l.get_table(key) {
                let mut owned_val = vec![];
                let val = tbl.get(key, u64::MAX, key_hash, &mut owned_val, l.level);
                if val.is_valid() {
                    return !val.is_deleted();
                }
            }
        }
        false
    }

    fn to_change_set(&self, outer_ranges: &[(Bytes, Bytes)], write_cf_only: bool) -> pb::ChangeSet {
        let mut cs = new_change_set(self.get_tag().id_ver.id, self.get_tag().id_ver.ver);
        cs.set_sequence(self.meta_seq);

        let snap = cs.mut_snapshot();
        let mut properties = pb::Properties::new();
        properties.shard_id = self.get_tag().id_ver.id;
        if let Some(encryption_key) = &self.encryption_key {
            if encryption_key.key_type == KEY_TYPE_AES_256_CTR_LEGACY {
                properties.mut_keys().push(ENCRYPTION_KEY.to_string());
                properties.mut_values().push(encryption_key.export());
            }
        }
        if let Some(encryption_meta_bytes) = &self.encryption_meta_bytes {
            properties.mut_keys().push(ENCRYPTION_META_KEY.to_string());
            properties.mut_values().push(encryption_meta_bytes.to_vec());
        }
        if self.storage_class_spec.is_specified() {
            set_shard_property(
                STORAGE_CLASS_KEY,
                &mut properties,
                self.storage_class_spec.marshal(),
            );
        }

        snap.set_outer_start(self.get_start_key().to_vec());
        snap.set_outer_end(self.get_end_key().to_vec());
        snap.set_inner_key_off(self.data.inner_key_off as u32);
        // NOTE: `ChangeSet.snapshot.data_sequence` is not set, as it's not able to
        // get the accurate data sequence of ShardMeta from Shard.
        //
        // snap.set_data_sequence(X);
        snap.set_properties(properties);
        let mut count = 0;
        let mut overlapped_count = 0;
        let mut max_ts = 0;
        let mut l0_ids = HashSet::new();
        let mut range_bounds = Vec::with_capacity(outer_ranges.len());
        for (outer_start, outer_end) in outer_ranges {
            let inner_start = InnerKey::from_outer_key(outer_start);
            let inner_end = InnerKey::from_outer_end_key(outer_end);
            range_bounds.push(DataBound::new(inner_start, inner_end, false));
        }
        for v in &self.data.l0_tbls {
            count += 1;
            if write_cf_only {
                if let Some(cf) = v.get_cf(WRITE_CF) {
                    if cf.size() == 0 {
                        continue;
                    }
                } else {
                    continue;
                }
            }
            if !range_bounds.iter().any(|bound| v.overlap_bound(*bound)) {
                continue;
            }
            overlapped_count += 1;
            max_ts = max_ts.max(v.max_ts());
            snap.mut_l0_creates().push(v.to_l0_create());
            l0_ids.insert(v.id());
        }
        for (k, v) in self.data.blob_tbl_map.iter() {
            assert_eq!(k, &v.id());
            count += 1;
            // FIXME: Overlap check
            snap.mut_blob_creates().push(v.to_blob_create());
        }
        self.data.for_each_level(|cf, lh| {
            if write_cf_only && cf != WRITE_CF {
                return false;
            }
            let mut overlap_ids = HashSet::new();
            for bound in &range_bounds {
                let (left, right) = bound.get_overlap_data_sets(&lh.tables);
                if left == right {
                    continue;
                }
                for tbl in &lh.tables[left..right] {
                    overlap_ids.insert(tbl.id());
                }
            }
            for tbl in lh.tables.iter() {
                count += 1;
                if overlap_ids.contains(&tbl.id()) {
                    overlapped_count += 1;
                    max_ts = max_ts.max(tbl.max_ts);
                    snap.mut_table_creates()
                        .push(tbl.to_table_create(cf, lh.level));
                }
            }
            false
        });
        self.data.for_each_columnar_level(|cl| {
            for col_file in cl.files.iter() {
                count += 1;
                // check overlap
                let mut overlap = false;
                for (outer_start, outer_end) in outer_ranges {
                    let inner_start = InnerKey::from_outer_key(outer_start);
                    let inner_end = InnerKey::from_outer_end_key(outer_end);
                    if col_file.has_data_in_range(inner_start, inner_end) {
                        overlap = true;
                        break;
                    }
                }
                if !overlap {
                    continue;
                }
                overlapped_count += 1;
                max_ts = max_ts.max(col_file.get_max_version());
                snap.mut_columnar_creates()
                    .push(col_file.to_columnar_create(cl.level));
            }
            false
        });
        self.data.col_levels.unconverted_l0s.iter().for_each(|l0| {
            // l0 may be skip due to non-overlap
            if l0_ids.contains(&l0.id()) {
                snap.mut_unconverted_l0s().push(l0.id());
            }
        });
        if let Some(schema_file) = &self.data.schema_file {
            snap.set_schema_meta(schema_file.to_schema_meta());
        }
        snap.set_columnar_table_ids(self.data.columnar_table_ids.clone());
        snap.set_columnar_l2_snap_version(self.data.col_levels.l2_snap_version.into_inner());
        let vector_indexes = snap.mut_vector_indexes();
        for index in self.data.vector_indexes.get_all() {
            vector_indexes.push(index.to_vector_index_pb());
        }
        let vector_indexes_count = vector_indexes.iter().map(|v| v.files.len()).sum::<usize>();

        let mut fts_tracked_indexes = Vec::new();
        let mut fts_index_metas = Vec::new();
        for (table_id, index_metas) in self.data.fts_levels.iter_tracked_indexes() {
            for (index_id, meta) in index_metas {
                let mut tbl_idx_id = pb::fts::TableIndexId::new();
                tbl_idx_id.set_table_id(*table_id);
                tbl_idx_id.set_index_id(*index_id);
                fts_tracked_indexes.push(tbl_idx_id);
                fts_index_metas.push(meta.clone());
            }
        }
        snap.set_fts_indexes(fts_tracked_indexes.into());
        snap.set_fts_index_metas(fts_index_metas.into());

        let fts_l0_files = snap.mut_fts_l0_files();
        for file in self.data.fts_levels.l0() {
            count += 1;
            let file_bound = file.data_bound();
            if !range_bounds
                .iter()
                .any(|bound| bound.overlap_bound(file_bound))
            {
                continue;
            }
            overlapped_count += 1;
            fts_l0_files.push(file.build_info());
        }

        let fts_l1_files = snap.mut_fts_l1_files();
        for file in self.data.fts_levels.l1() {
            count += 1;
            let file_bound = file.data_bound();
            if !range_bounds
                .iter()
                .any(|bound| bound.overlap_bound(file_bound))
            {
                continue;
            }
            overlapped_count += 1;
            fts_l1_files.push(file.build_info());
        }

        let fts_l2_files = snap.mut_fts_l2_files();
        for files in self.data.fts_levels.l2().values() {
            for file in files {
                count += 1;
                let file_bound = file.data_bound();
                if !range_bounds
                    .iter()
                    .any(|bound| bound.overlap_bound(file_bound))
                {
                    continue;
                }
                overlapped_count += 1;
                fts_l2_files.push(file.build_info());
            }
        }
        snap.set_fts_pending_l0_ids(self.data.fts_levels.pending_columnar_l0_ids().to_vec());
        snap.set_fts_l0_snap_version(self.data.fts_levels.l0_snap_version.into_inner());
        snap.set_max_ts(max_ts);

        info!(
            "{} convert snap access to change set, total {}, overlapped {}, unconverted_l0s {}, columnar {}, vector {}",
            self.get_tag(),
            count,
            overlapped_count,
            snap.get_unconverted_l0s().len(),
            snap.get_columnar_creates().len(),
            vector_indexes_count,
        );
        cs
    }

    pub fn marshal(
        &self,
        ranges: &[(Bytes, Bytes)],
        write_cf_only: bool,
        cache_key: bool,
    ) -> (String, Vec<u8>) {
        let cs = self.to_change_set(ranges, write_cf_only);
        let key = if cache_key {
            let mut ranges_bytes = Vec::new();
            for (start, end) in ranges {
                ranges_bytes.append(start.to_vec().as_mut());
                ranges_bytes.append(end.to_vec().as_mut());
            }
            let ranges_key = hex::encode(ranges_bytes);
            format!(
                "{}:{}:{}:{}",
                cs.shard_id, cs.shard_ver, cs.sequence, ranges_key,
            )
        } else {
            "".to_string()
        };
        (key, cs.write_to_bytes().unwrap())
    }

    /// Serializes full-shard online snapshot metadata with the exact write
    /// sequence used by the paired full-CF memtable payload and the source
    /// shard's Raft term. Unresolved txn-file locks are derived from the same
    /// captured shard data.
    pub fn marshal_online_snapshot(&self, term: u64) -> Vec<u8> {
        let ranges = [(self.clone_start_key(), self.clone_end_key())];
        let mut cs = self.to_change_set(&ranges, false);
        let snapshot = cs.mut_snapshot();
        snapshot.set_base_version(self.base_version);
        snapshot.set_data_sequence(self.write_sequence);
        set_shard_property(
            TERM_KEY,
            snapshot.mut_properties(),
            term.to_le_bytes().to_vec(),
        );

        let deleting_prefixes = self
            .deleting_prefixes
            .build_split(ranges[0].0.as_ref(), ranges[0].1.as_ref());
        if !deleting_prefixes.is_empty() {
            set_shard_property(
                DEL_PREFIXES_KEY,
                snapshot.mut_properties(),
                deleting_prefixes.marshal(),
            );
        }

        let mut lock_txn_file_refs = TxnFileRefs::default();
        for txn_file in &self.data.lock_txn_files {
            if let Some(txn_file_ref) = txn_file.to_txn_file_ref(None) {
                lock_txn_file_refs.mut_txn_file_refs().push(txn_file_ref);
            }
        }
        if !lock_txn_file_refs.get_txn_file_refs().is_empty() {
            set_shard_property(
                TXN_FILE_REF,
                snapshot.mut_properties(),
                lock_txn_file_refs.write_to_bytes().unwrap(),
            );
        }
        cs.write_to_bytes().unwrap()
    }

    /// Builds remote memtable data and reports whether serialization skipped a
    /// WRITE_CF version newer than `start_ts`.
    ///
    /// Remote cache admission needs this evidence because skipped versions are
    /// intentionally absent from the serialized memtable data.
    pub fn build_mem_data(
        &self,
        outer_ranges: &[(Bytes, Bytes)],
        start_ts: u64,
    ) -> (Vec<u8>, bool) {
        let data_bounds: Vec<DataBound<'_>> = outer_ranges
            .iter()
            .map(|range| {
                DataBound::new(
                    InnerKey::from_outer_key(&range.0),
                    InnerKey::from_outer_key(&range.1),
                    false,
                )
            })
            .collect();

        let mut mem_data = Vec::with_capacity(U32_SIZE /* format */);
        let (skls, txn_file_refs) = self.get_mem_tables_group_by_type(WRITE_CF, &data_bounds);
        mem_data.put_u32_le(MEM_DATA_FORMAT_V2);
        let met_newer_ts = self.build_skl_data(outer_ranges, start_ts, skls, &mut mem_data);
        self.build_txn_file_data(txn_file_refs, &mut mem_data);
        (mem_data, met_newer_ts)
    }

    /// Serializes a coherent full-shard, full-CF memtable snapshot for online
    /// branch capture. All versions and tombstones are preserved.
    pub fn build_full_mem_data(&self) -> Vec<u8> {
        let data_bounds = [DataBound::new(
            self.get_inner_start(),
            self.get_inner_end(),
            false,
        )];

        let mut mem_data = Vec::with_capacity(U32_SIZE * 2);
        mem_data.put_u32_le(ONLINE_SNAPSHOT_MEM_DATA_FORMAT_VERSION);
        mem_data.put_u32_le(NUM_CFS as u32);
        let mut write_txn_file_refs = TxnFileRefs::default();
        for cf in 0..NUM_CFS {
            let (skls, txn_file_refs) = self.get_mem_tables_group_by_type(cf, &data_bounds);
            mem_data.put_u32_le(cf as u32);
            self.build_raw_skl_data(cf, skls, &mut mem_data);
            if cf == WRITE_CF {
                write_txn_file_refs = txn_file_refs;
            } else {
                debug_assert!(txn_file_refs.get_txn_file_refs().is_empty());
            }
        }
        self.build_txn_file_data(write_txn_file_refs, &mut mem_data);
        mem_data
    }

    /// Returns the cumulative skiplist KV bytes used to approximate the peak
    /// memory of a full-CF online snapshot capture. Transaction-file contents
    /// are excluded because the payload serializes only their references.
    pub fn full_mem_table_kv_size(&self) -> usize {
        self.data.mem_tbls.iter().fold(0, |size, mem_tbl| {
            size.saturating_add(mem_tbl.skip_list_size())
        })
    }

    fn build_raw_skl_data(&self, cf: usize, skls: Vec<SkipList>, mem_data: &mut Vec<u8>) {
        let skl_iters = skls.iter().map(|skl| skl.new_iterator(false)).collect();
        let mut merge_iter = table::new_merge_iterator(skl_iters, false);
        let mut rows = Vec::new();
        // LOCK_CF uses the physical SnapVersion assigned by the Raft write
        // sequence. SnapAccess shares the writable memtable with apply, so a
        // later write can become visible before its sequence is published.
        // Exclude those later versions while retaining older versions of the
        // same key. Managed CF versions are timestamps and are not comparable
        // with this boundary.
        let max_version = (cf == LOCK_CF).then(|| self.get_mem_table_snap_version().into_inner());

        let inner_start = self.get_inner_start();
        let inner_end = self.get_inner_end();
        merge_iter.seek(inner_start);
        while merge_iter.valid() && merge_iter.key() < inner_end {
            let key = merge_iter.key().to_vec();
            let mut versions = Vec::new();
            loop {
                let value = merge_iter.value();
                if max_version.is_none_or(|max_version| value.version <= max_version) {
                    versions.push(MemTableRowV3 {
                        key: key.clone(),
                        meta: value.meta,
                        user_meta: value.user_meta().to_vec(),
                        version: value.version,
                        value: value.get_value().to_vec(),
                    });
                }
                if !merge_iter.next_version() {
                    break;
                }
            }
            versions.reverse();
            rows.extend(versions);
            merge_iter.next();
        }

        let mem_size = bincode::serialized_size(&rows)
            .map_err(|err| Error::Other(err))
            .unwrap();
        mem_data.reserve(U64_SIZE + mem_size as usize);
        mem_data.put_u64_le(mem_size);
        bincode::serialize_into(mem_data, &rows).unwrap();
    }

    fn build_skl_data(
        &self,
        ranges: &[(Bytes, Bytes)],
        start_ts: u64,
        skls: Vec<SkipList>,
        mem_data: &mut Vec<u8>,
    ) -> bool {
        let skl_iters = skls.iter().map(|skl| skl.new_iterator(false)).collect();
        let merge_iter = table::new_merge_iterator(skl_iters, false);

        let data = self.data.clone();
        let mut key = BytesMut::new();
        key.extend_from_slice(data.keyspace_prefix());
        // If start_ts is u64::MAX, it means all versions are needed.
        let all_versions = start_ts == u64::MAX;
        let mut mem_iterator = Iterator {
            all_versions,
            reversed: false,
            read_ts: start_ts,
            met_newer_ts: false,
            key,
            val: table::Value::new(),
            skipped_tombstones: 0,
            inner: merge_iter,
            iter_stats: IterStats::default(),
            is_blob_value_record: false,
            blob_prefetcher: None,
            owned_blob: None,
            encryption_key: self.encryption_key.clone(),
            skip_blob: true,
            data,
            range: None,
        };

        let mut rows = vec![];
        // Save mvcc versions of the same key temporarily, and append to rows with
        // reverse order.
        let mut mvcc_rows = vec![];
        let mut last_key = vec![];
        for (range_start, range_end) in ranges {
            mem_iterator.set_range(range_start.clone(), range_end.clone());
            while mem_iterator.valid() {
                let key = mem_iterator.key().to_vec();
                if key.as_slice() >= range_end.chunk() {
                    break;
                }
                if key != last_key {
                    for _ in 0..mvcc_rows.len() {
                        rows.push(mvcc_rows.pop().unwrap());
                    }
                }
                mvcc_rows.push(table::Row {
                    key: key.clone(),
                    user_meta: UserMeta::from_slice(mem_iterator.user_meta()),
                    value: mem_iterator.val().to_vec(),
                });
                mem_iterator.next();
                last_key = key;
            }
        }
        for _ in 0..mvcc_rows.len() {
            rows.push(mvcc_rows.pop().unwrap());
        }
        let mem_size = bincode::serialized_size(&rows)
            .map_err(|e| Error::Other(e))
            .unwrap();
        mem_data.reserve(U64_SIZE /* mem_size */ + mem_size as usize);
        mem_data.put_u64_le(mem_size);
        bincode::serialize_into(mem_data, &rows).unwrap();
        mem_iterator.met_newer_ts()
    }

    fn build_txn_file_data(&self, txn_file_refs: TxnFileRefs, mem_data: &mut Vec<u8>) {
        let msg_len = txn_file_refs.compute_size();
        mem_data.reserve(U32_SIZE /* msg_len */ + msg_len as usize);
        mem_data.put_u32_le(msg_len);
        txn_file_refs.write_to_vec(mem_data).unwrap();
    }

    fn get_mem_tables_group_by_type(
        &self,
        cf: usize,
        bounds: &[DataBound<'_>],
    ) -> (Vec<SkipList>, TxnFileRefs) {
        let mut skls = Vec::new();
        let mut txn_file_refs = TxnFileRefs::default();
        for mem_tbl in &self.data.mem_tbls {
            let skl_ext = mem_tbl.get_cf(cf);

            for txn_file_ref in skl_ext
                .get_txn_files()
                .into_iter()
                .filter_map(|txn_file| txn_file.to_txn_file_ref(Some(bounds)))
            {
                txn_file_refs.mut_txn_file_refs().push(txn_file_ref);
            }

            let skl = skl_ext.get_skl();
            if !skl.is_empty() {
                skls.push(skl);
            }
        }
        (skls, txn_file_refs)
    }

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

    /// Returns typed immutable file references from the exact state captured by
    /// this `SnapAccess`.
    pub fn get_all_files_with_type(&self) -> Vec<(FileType, u64)> {
        self.data.get_all_files_with_type()
    }

    pub fn has_schema_file(&self) -> bool {
        self.data.schema_file.is_some()
    }

    // Used by proxy kvengine.
    pub fn get_schema_file(&self) -> Option<SchemaFile> {
        self.data.schema_file.clone()
    }

    #[maybe_async::both]
    pub async fn get_newer(&self, cf: usize, key: &[u8], version: u64) -> Item<'_> {
        let inner_key = InnerKey::from_outer_key(key);
        let mut item = Item::new();
        item.owned_val = Some(vec![]);
        item.val = self
            .get_newer_val(cf, inner_key, version, item.owned_val.as_mut().unwrap())
            .await;
        if item.val.is_blob_ref() {
            item.owned_blob = Some(self.fetch_blob(inner_key, item.val.get_blob_ref()).await);
            item.val.fill_in_blob(item.owned_blob.as_ref().unwrap());
        }
        item
    }

    /// Returns whether `key` has a version newer than `version`.
    ///
    /// This metadata-only probe deliberately leaves blob references unresolved.
    /// Callers that only need version existence must not pay the object-store
    /// IO or failure cost of materializing an invisible blob value.
    #[maybe_async::both]
    pub async fn has_newer(&self, cf: usize, key: &[u8], version: u64) -> bool {
        let inner_key = InnerKey::from_outer_key(key);
        let mut value_owner = vec![];
        self.get_newer_val(cf, inner_key, version, &mut value_owner)
            .await
            .is_valid()
    }

    #[maybe_async::both]
    async fn get_newer_val(
        &self,
        cf: usize,
        inner_key: InnerKey<'_>,
        version: u64,
        out_val_owner: &mut Vec<u8>,
    ) -> table::Value {
        self.check_sync(cf).await;
        let key_hash = farmhash::fingerprint64(inner_key.deref());
        for i in 0..self.data.mem_tbls.len() {
            let tbl = self.data.mem_tbls.as_slice()[i].get_cf(cf);
            let v = tbl.get_newer(inner_key, version, out_val_owner);
            if v.is_valid() {
                return v;
            }
        }
        for l0 in &self.data.l0_tbls {
            if let Some(tbl) = &l0.get_cf(cf) {
                let v = tbl.get_newer(inner_key, version, key_hash, out_val_owner, 0);
                if v.is_valid() {
                    return v;
                }
            }
        }
        let scf = self.data.get_cf(cf);
        for lh in &scf.levels {
            let v = lh
                .get_newer(inner_key, version, key_hash, out_val_owner)
                .await;
            if v.is_valid() {
                return v;
            }
        }
        table::Value::new()
    }

    #[maybe_async::both]
    pub async fn has_data_in_prefix<'a>(&'a self, mut prefix: &'a [u8]) -> bool {
        self.check_sync(WRITE_CF).await;
        let keyspace_prefix = self.data.keyspace_prefix();

        let min_off = std::cmp::min(prefix.len(), keyspace_prefix.len());
        if min_off > 0 {
            if prefix[0..min_off] != keyspace_prefix[0..min_off] {
                return false;
            }
            if prefix.len() < keyspace_prefix.len() {
                // If prefix is less than keyspace_prefix_len, the data in shard always have
                // `keyspace_prefix`, just update prefix to keyspace_prefix.
                prefix = keyspace_prefix;
            }
        }

        let inner_prefix = InnerKey::from_outer_key(prefix);
        if self.deleting_prefixes.cover_prefix(inner_prefix) {
            return false;
        }
        let mut it = self
            .new_iterator(WRITE_CF, false, false, Some(u64::MAX), true)
            .await;
        it.seek(prefix).await;
        if !it.valid() {
            return false;
        }
        it.key().starts_with(prefix)
    }

    pub fn has_unloaded_tables(&self) -> bool {
        !self.data.unloaded_tbls.is_empty()
    }

    pub fn get_encryption_key(&self) -> Option<EncryptionKey> {
        self.encryption_key.clone()
    }

    pub fn estimated_range_blocks_size(&self, ranges: &[(Bytes, Bytes)]) -> usize {
        // ignore L0 tables, only estimate L1+ for simplicity and performance.
        let mut blocks_size = 0;
        let write_cf = &self.data.cfs[0];
        for lvl in &write_cf.levels {
            blocks_size += lvl.range_blocks_size(ranges);
        }
        blocks_size
    }

    pub fn get_keyspace_id(&self) -> u32 {
        self.data.keyspace_id
    }

    #[maybe_async::both]
    async fn seek_txn_file(&self, iter: &mut Box<dyn table::Iterator>, txn_file: &TxnFile) {
        if txn_file.smallest() < self.data.inner_start() {
            iter.seek(self.data.inner_start()).await;
        } else {
            iter.seek(txn_file.smallest()).await;
        }
    }

    fn get_upper_bound<'a>(&'a self, buf: &'a mut Vec<u8>, txn_file: &TxnFile) -> InnerKey<'a> {
        if txn_file.biggest() < self.data.inner_end() {
            buf.extend_from_slice(txn_file.biggest().deref());
            buf.push(0);
            InnerKey::from_inner_buf(buf)
        } else {
            self.data.inner_end()
        }
    }

    pub fn get_txn_file_conflict_lock(
        &self,
        txn_file: &TxnFile,
    ) -> Option<(Vec<u8>, LockOrSharedLocks)> {
        if txn_file.is_empty() {
            return None;
        }
        if txn_file.size() < self.data.get_cf(LOCK_CF).size() as usize {
            // It is more efficient to iterate the txn file when it is small.
            let mut other_lock_txn_files = self.data.lock_txn_files.clone();
            other_lock_txn_files.retain(|f| f.start_ts() != txn_file.start_ts());
            let mut txn_file_iter = TxnFileIterator::new(txn_file.clone(), false);
            txn_file_iter.rewind();
            while txn_file_iter.valid() {
                let mut val = vec![];
                let txn_file_key = txn_file_iter.key();
                let version = self.get_mem_table_snap_version().into_inner();
                let lock_val = self.get_value(
                    LOCK_CF,
                    txn_file_key,
                    version,
                    &mut AccessPath::default(),
                    &mut val,
                    &other_lock_txn_files,
                );
                if lock_val.is_valid() && !lock_val.is_deleted() {
                    let conflict_lock = txn_types::parse_lock(lock_val.get_value()).unwrap();
                    // Return outer key.
                    let mut key = self.data.keyspace_prefix().to_vec();
                    key.extend_from_slice(txn_file_key.deref());
                    return Some((key, conflict_lock));
                }
                txn_file_iter.next();
            }
            return None;
        }
        let mut lock_iter =
            self.new_table_iterator(LOCK_CF, false, true, Some(txn_file.start_ts()));
        self.seek_txn_file(&mut lock_iter, txn_file);
        let mut upper_bound_buf = vec![];
        let upper_bound = self.get_upper_bound(&mut upper_bound_buf, txn_file);
        let mut outer_val_buf = vec![];
        while lock_iter.valid() {
            if lock_iter.key() >= upper_bound {
                return None;
            }
            let lock_iter_val = lock_iter.value();
            if !lock_iter_val.is_deleted()
                && txn_file
                    .get_value(lock_iter.key(), &mut outer_val_buf)
                    .1
                    .is_valid()
            {
                let conflict_lock = txn_types::parse_lock(lock_iter_val.get_value()).unwrap();
                // Return outer key.
                let mut key = self.data.keyspace_prefix().to_vec();
                key.extend_from_slice(lock_iter.key().as_ref());
                return Some((key, conflict_lock));
            }
            lock_iter.next();
        }
        None
    }

    // Get writes with `commit_ts` LARGER than `txn_file.start_ts()`.
    // These writes are not seen by clients and should be considered as conflicts.
    // Ref: `SnapAccessCore::get_newer`.
    #[maybe_async::both]
    pub async fn get_txn_file_conflict_write(
        &self,
        txn_file: &TxnFile,
    ) -> Option<(Vec<u8>, UserMeta)> {
        self.check_sync(WRITE_CF).await;
        if txn_file.is_empty() {
            return None;
        }
        let mut write_iter = self.new_delta_write_iterator(txn_file.start_ts()).await;
        self.seek_txn_file(&mut write_iter, txn_file).await;
        let mut upper_bound_buf = vec![];
        let upper_bound = self.get_upper_bound(&mut upper_bound_buf, txn_file);
        let mut outer_val_buf = vec![];
        while write_iter.valid() {
            if write_iter.key() >= upper_bound {
                return None;
            }
            let write_iter_val = write_iter.value();
            if !write_iter_val.is_deleted()
                && txn_file
                    .get_value(write_iter.key(), &mut outer_val_buf)
                    .1
                    .is_valid()
            {
                debug_assert!(
                    !write_iter_val.user_meta().is_empty(),
                    "write_iter_val: {:?}",
                    write_iter_val,
                );
                let um = UserMeta::from_slice(write_iter_val.user_meta());
                if um.commit_ts > txn_file.start_ts() {
                    // Return outer key.
                    let mut key = self.data.keyspace_prefix().to_vec();
                    key.extend_from_slice(write_iter.key().as_ref());
                    return Some((key, um));
                }
            }
            next!(write_iter).await;
        }
        None
    }

    #[maybe_async::both]
    pub async fn check_txn_file_constraint(
        &self,
        txn_file: &TxnFile,
        read_ts: u64,
    ) -> Option<Vec<u8> /* already_exist_key */> {
        self.check_sync(WRITE_CF).await;
        let mut checker = TxnFileConstraintChecker::new(self, read_ts);
        txn_file
            .iter_check_entries(self.data.data_bound(), &mut checker)
            .await;
        checker.already_exist_key
    }

    pub fn get_lock_txn_file(&self, start_ts: u64) -> Option<TxnFile> {
        for txn_file in &self.data.lock_txn_files {
            if txn_file.start_ts() == start_ts {
                return Some(txn_file.clone());
            }
        }
        None
    }

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

    pub fn get_limiter(&self) -> &RegionLimiter {
        &self.data.limiter
    }

    pub fn new_schema_from_columns(&self, table_id: i64, columns: &[ColumnInfo]) -> Option<Schema> {
        if !self.read_columnar || !self.data.columnar_table_ids.contains(&table_id) {
            return None;
        }
        let schema_file = self.data.schema_file.as_ref()?;
        let table_schema = schema_file.get_table(table_id)?;
        if !table_schema.with_columnar() {
            return None;
        }
        let columns = columns
            .iter()
            .filter(|c| !c.get_pk_handle() && c.get_column_id() != HANDLE_COL_ID as i64)
            .cloned()
            .collect::<Vec<_>>();
        let schema_buf = SchemaBuf::new(
            table_id,
            table_schema.handle_column.clone(),
            table_schema.version_column.clone(),
            columns,
            table_schema.pk_col_ids.clone(),
            table_schema.max_col_id,
            table_schema.vector_indexes.clone(),
            table_schema.fulltext_indexes.clone(),
            table_schema.get_storage_class_spec().clone(),
            None,
        );
        Some(Schema::new(schema_buf))
    }

    pub fn new_columnar_mvcc_reader_from_row(
        &self,
        table_id: i64,
        columns: &[ColumnInfo],
        read_ts: u64,
    ) -> Result<Option<ColumnarMvccReader>> {
        let Some(schema) = self.new_schema_from_columns(table_id, columns) else {
            return Ok(None);
        };
        let mut readers: Vec<Box<dyn ColumnarReader>> = vec![];
        for mem in &self.data.mem_tbls {
            let skl = mem.get_cf(WRITE_CF);
            if !skl.is_empty() {
                let iter = skl.new_iterator(false);
                let row_reader = ColumnarRowTableReader::new(
                    schema.clone(),
                    iter,
                    None,
                    false,
                    self.encryption_key.clone(),
                )?;
                readers.push(Box::new(row_reader));
            }
        }
        for l0 in &self.data.l0_tbls {
            if let Some(l0_write) = l0.get_cf(WRITE_CF) {
                let iter = l0_write.new_iterator(false, false);
                let row_reader = ColumnarRowTableReader::new(
                    schema.clone(),
                    iter,
                    None,
                    false,
                    self.encryption_key.clone(),
                )?;
                readers.push(Box::new(row_reader));
            }
        }
        let mut row_reader_err: Option<table::Error> = None;
        self.data.for_each_level(|cf, lh| {
            if cf != WRITE_CF {
                return false;
            }
            for tbl in lh.tables.iter() {
                let iter = tbl.new_iterator(false, false);
                match ColumnarRowTableReader::new(
                    schema.clone(),
                    iter,
                    None,
                    false,
                    self.encryption_key.clone(),
                ) {
                    Ok(row_reader) => readers.push(Box::new(row_reader)),
                    Err(e) => {
                        row_reader_err = Some(e);
                        return true;
                    }
                }
            }
            false
        });
        if let Some(e) = row_reader_err {
            return Err(e.into());
        }

        let merge_reader = ColumnarMergeReader::new(schema.clone(), readers);
        Ok(Some(ColumnarMvccReader::new(
            Box::new(merge_reader),
            &schema,
            read_ts,
        )))
    }

    /// `new_columnar_mvcc_reader` will try to construct a reader in columnar
    /// mode. If `None` is returned, normal tikv row reading will be used.
    pub fn new_columnar_mvcc_reader(
        &self,
        table_id: i64,
        columns: &[ColumnInfo],
        scan_ctx: Option<&TableScanCtx>,
        read_ts: u64,
        ann_query: Option<Arc<AnnQueryInfo>>,
    ) -> Result<Option<Box<dyn ColumnarFilterReader>>> {
        self.new_columnar_mvcc_reader_impl(table_id, columns, scan_ctx, read_ts, ann_query, true)
    }

    /// Construct a columnar MVCC reader without pack clean read.
    ///
    /// Use this when the caller depends on the internal handle column in
    /// returned blocks, because pack clean read intentionally skips handle
    /// packs and fills internal placeholder handles.
    pub fn new_columnar_mvcc_reader_without_pack_clean_read(
        &self,
        table_id: i64,
        columns: &[ColumnInfo],
        scan_ctx: Option<&TableScanCtx>,
        read_ts: u64,
        ann_query: Option<Arc<AnnQueryInfo>>,
    ) -> Result<Option<Box<dyn ColumnarFilterReader>>> {
        self.new_columnar_mvcc_reader_impl(table_id, columns, scan_ctx, read_ts, ann_query, false)
    }

    /// Builds the L2-only late-materialization source graph. The normal reader
    /// remains the fallback for every ineligible scan; in particular this path
    /// never uses pack-clean because it needs real handle/version values.
    fn new_hybrid_columnar_mvcc_reader(
        &self,
        table_id: i64,
        columns: &[ColumnInfo],
        early_column_ids: &HashSet<i64>,
        scan_ctx: Option<&TableScanCtx>,
        read_ts: u64,
    ) -> Result<Option<HybridMvccReader>> {
        let Some(full_schema) = self.new_schema_from_columns(table_id, columns) else {
            info!(
                "{} late materialization unavailable: reason=schema_unavailable table_id={} early_columns={}",
                self.tag,
                table_id,
                early_column_ids.len()
            );
            return Ok(None);
        };
        if !early_column_ids.contains(&full_schema.handle_column.get_column_id())
            || !early_column_ids.contains(&full_schema.version_column.get_column_id())
        {
            info!(
                "{} late materialization unavailable: reason=missing_handle_or_version table_id={} handle_column_id={} version_column_id={} early_columns={}",
                self.tag,
                table_id,
                full_schema.handle_column.get_column_id(),
                full_schema.version_column.get_column_id(),
                early_column_ids.len()
            );
            return Ok(None);
        }

        let early_schema = Schema::new(
            full_schema
                .to_schema_buf()
                .retain_columns(|column| early_column_ids.contains(&column.get_column_id())),
        );
        let late_columns = full_schema
            .columns
            .iter()
            .filter(|column| !early_column_ids.contains(&column.get_column_id()))
            .cloned()
            .collect::<Vec<_>>();
        if late_columns.is_empty() {
            info!(
                "{} late materialization unavailable: reason=no_late_columns table_id={} schema_columns={} early_columns={}",
                self.tag,
                table_id,
                full_schema.columns.len(),
                early_column_ids.len()
            );
            return Ok(None);
        }

        let filter_op = scan_ctx
            .filter(|ctx| !ctx.is_empty_filter())
            .map(TableScanCtx::to_filter_operator);
        let mut sources = self
            .collect_column_row_readers(&full_schema, read_ts)?
            .into_iter()
            .map(HybridSourceReader::Eager)
            .collect::<Vec<_>>();

        let mut l2_files = vec![];
        let mut next_l2_reader_id = 0;
        for columnar_level in &self.data.col_levels.levels {
            if columnar_level.level == 2 {
                if !columnar_level.has_strictly_non_overlapping_files_for_table(table_id) {
                    info!(
                        "{} late materialization unavailable: reason=l2_files_overlap table_id={} level={}",
                        self.tag, table_id, columnar_level.level
                    );
                    return Ok(None);
                }
                let files = columnar_level
                    .files
                    .iter()
                    .filter(|file| file.has_table(table_id))
                    .cloned()
                    .collect::<Vec<_>>();
                if files.is_empty() {
                    continue;
                }
                let l2_reader_id = next_l2_reader_id;
                next_l2_reader_id += 1;
                sources.push(HybridSourceReader::L2 {
                    reader: ColumnarConcatReader::new(
                        &files,
                        early_schema.clone(),
                        filter_op.clone(),
                        self.encryption_key.clone(),
                    ),
                    l2_reader_id,
                });
                l2_files.push(files);
                continue;
            }

            for file in &columnar_level.files {
                if !file.has_table(table_id) {
                    continue;
                }
                sources.push(HybridSourceReader::Eager(Box::new(
                    ColumnarTableReader::new(
                        file,
                        full_schema.clone(),
                        filter_op.clone(),
                        self.encryption_key.clone(),
                    )?,
                )));
            }
        }
        if l2_files.is_empty() {
            info!(
                "{} late materialization unavailable: reason=no_l2_files table_id={}",
                self.tag, table_id
            );
            return Ok(None);
        }
        let merge = HybridMergeReader::new(early_schema, full_schema, late_columns, sources);
        Ok(Some(HybridMvccReader::new(merge, read_ts, l2_files)))
    }

    fn new_columnar_mvcc_reader_impl(
        &self,
        table_id: i64,
        columns: &[ColumnInfo],
        scan_ctx: Option<&TableScanCtx>,
        read_ts: u64,
        ann_query: Option<Arc<AnnQueryInfo>>,
        allow_pack_clean_read: bool,
    ) -> Result<Option<Box<dyn ColumnarFilterReader>>> {
        // If vector distance projection is enabled, a virtual distance column is
        // included in the schema (which is what TiDB expects the storage layer to
        // return) and there is no vector column in the schema. In this case, we
        // will first read using a new schema with the vector column included, then
        // return the data (what TiDB wants) with a distance column calculated from the
        // vector column.
        let has_vector_distance_proj = ann_query
            .as_ref()
            .map(|q| q.get_enable_distance_proj())
            .unwrap_or(false);

        let Some(schema) = self.new_schema_from_columns(table_id, columns) else {
            return Ok(None);
        };
        let has_filter = scan_ctx_has_filter(scan_ctx);
        let filter_op = if has_filter {
            scan_ctx.map(|ctx| ctx.to_filter_operator())
        } else {
            None
        };

        let schema_to_read = if has_vector_distance_proj {
            VectorDistanceProjector::generate_inner_schema(&schema, ann_query.as_ref().unwrap())?
        } else {
            schema.clone()
        };
        let mut readers = self.collect_column_row_readers(&schema_to_read, read_ts)?;
        let has_non_l2_columnar = self.data.col_levels.levels.iter().any(|columnar_level| {
            columnar_level.level != 2
                && columnar_level
                    .files
                    .iter()
                    .any(|file| file.has_table(schema_to_read.table_id))
        });
        // Pack clean read skips real handle/version packs and fills internal
        // placeholder buffers instead. Only enable it when the projection does
        // not expose handle, pk_handle or version.
        if readers.is_empty()
            && !has_non_l2_columnar
            && ann_query.is_none()
            && allow_pack_clean_read
            && columns_allow_pack_clean_read(columns)
        {
            let l2_files = self
                .data
                .col_levels
                .levels
                .iter()
                .find(|columnar_level| columnar_level.level == 2)
                .map(|columnar_level| columnar_level.files.as_slice())
                .unwrap_or(&[]);
            if l2_files_allow_pack_clean_read(l2_files, &schema_to_read, read_ts) {
                return Ok(Some(Box::new(ColumnarPackCleanReader::new(
                    l2_files,
                    schema_to_read,
                    read_ts,
                    self.encryption_key.clone(),
                    filter_op.clone(),
                ))));
            }
        }
        for columnar_level in &self.data.col_levels.levels {
            if columnar_level.level == 2 {
                let concat_reader = ColumnarConcatReader::new(
                    &columnar_level.files,
                    schema_to_read.clone(),
                    filter_op.clone(),
                    self.encryption_key.clone(),
                );
                readers.push(Box::new(concat_reader));
            } else {
                for col_file in &columnar_level.files {
                    if !col_file.has_table(schema_to_read.table_id) {
                        continue;
                    }
                    let col_reader = ColumnarTableReader::new(
                        col_file,
                        schema_to_read.clone(),
                        filter_op.clone(),
                        self.encryption_key.clone(),
                    )?;
                    readers.push(Box::new(col_reader));
                }
            }
        }
        let mut merged_reader: Box<dyn ColumnarReader> =
            Box::new(ColumnarMergeReader::new(schema_to_read.clone(), readers));
        // As commented above, if vector distance projection is enabled,
        // we will wrap a VectorDistanceProjector as the outside layer
        // to produce a distance column from the vector column.
        if has_vector_distance_proj {
            merged_reader = Box::new(VectorDistanceProjector::new(
                merged_reader,
                schema.clone(),
                ann_query.unwrap().clone(),
            )?);
        }

        let mvcc_reader = ColumnarMvccReader::new(merged_reader, &schema, read_ts);
        Ok(Some(Box::new(mvcc_reader)))
    }

    fn collect_column_row_readers(
        &self,
        schema: &Schema,
        read_ts: u64,
    ) -> Result<Vec<Box<dyn ColumnarReader>>> {
        let mut iters: Vec<Box<dyn TableIterator>> = vec![];

        for mem in &self.data.mem_tbls {
            let skl = mem.get_cf(WRITE_CF);
            if !skl.is_empty() {
                iters.push(skl.new_iterator(false));
            }
        }
        for l0 in &self.data.col_levels.unconverted_l0s {
            if let Some(l0_write) = l0.get_cf(WRITE_CF) {
                info!(
                    "{} add unconverted l0 columnar file {}",
                    self.tag,
                    l0_write.id()
                );
                iters.push(l0_write.new_iterator(false, true));
            }
        }
        if iters.is_empty() {
            return Ok(vec![]);
        }

        let blob_tbls =
            (!self.data.blob_tbl_map.is_empty()).then(|| self.data.blob_tbl_map.clone());
        let row_reader = RowMvccReader::new(
            schema.clone(),
            iters,
            blob_tbls,
            false, // match the row reader behavior in the normal columnar read path
            self.encryption_key.clone(),
            read_ts,
        )?;
        Ok(vec![Box::new(row_reader)])
    }

    fn new_fts_delta_index_schema(
        &self,
        table_id: i64,
        index_id: i64,
        tracked_index_layout: &crate::table::fts::TrackedIndexLayout,
    ) -> Result<Option<Schema>> {
        let Some(schema_file) = self.data.schema_file.as_ref() else {
            return Ok(None);
        };
        let Some(table_schema) = schema_file.get_table(table_id) else {
            return Ok(None);
        };
        let Some(index_def) = table_schema
            .fulltext_indexes
            .iter()
            .find(|index_def| index_def.get_index_id() == index_id)
        else {
            return Ok(None);
        };

        let tracked_index_fields =
            crate::table::fts::tracked_index_fields_from_layout_or_index_def(
                tracked_index_layout,
                index_def,
            )
            .map_err(|e| Error::Other(e.into()))?;
        if tracked_index_fields.is_empty()
            || tracked_index_fields.len() != tracked_index_layout.index_layout.fields().len()
        {
            return Ok(None);
        }

        let mut seen_col_ids = HashSet::with_capacity(tracked_index_fields.len());
        let mut columns = Vec::with_capacity(tracked_index_fields.len());
        for tracked_index_field in tracked_index_fields {
            let col_id = tracked_index_field.get_col_id();
            if !seen_col_ids.insert(col_id) {
                continue;
            }
            let Some(col) = table_schema.find_column_by_id(col_id) else {
                return Ok(None);
            };
            columns.push(col.clone());
        }

        Ok(Some(Schema::new(SchemaBuf::new(
            table_id,
            table_schema.handle_column.clone(),
            table_schema.version_column.clone(),
            columns,
            table_schema.pk_col_ids.clone(),
            table_schema.max_col_id,
            table_schema.vector_indexes.clone(),
            table_schema.fulltext_indexes.clone(),
            table_schema.get_storage_class_spec().clone(),
            table_schema.partitions.clone(),
        ))))
    }

    /// Construct a FTS reader for fallback purpose: It will always fulfill the
    /// FTS query request but the performance is not optimal. This function
    /// is used when:
    /// - Columnar storage is disabled.
    /// - No FTS index is found for the given query.
    ///
    /// IMPORTANT: schema and query must be first validated before calling this
    /// function.
    fn new_fts_reader_fallback(
        &self,
        query: &clara_fts::Query,
        output_schema: Schema,
        read_ts: u64,
    ) -> Result<FtsBruteForceReader<ColumnarMvccReader>> {
        let table_schema = self
            .data
            .schema_file
            .as_ref()
            .and_then(|schema_file| schema_file.get_table(output_schema.table_id))
            .ok_or_else(|| {
                Error::Other(
                    format!(
                        "missing table schema in schema_file for FTS fallback: table_id {}",
                        output_schema.table_id
                    )
                    .into(),
                )
            })?;
        let schema_row =
            FtsBruteForceReader::build_inner_schema(&output_schema, table_schema, query)?;
        let mut r_row = self.collect_column_row_readers(&schema_row, read_ts)?;
        for columnar_level in &self.data.col_levels.levels {
            if columnar_level.level == 2 {
                let concat_reader = ColumnarConcatReader::new(
                    &columnar_level.files,
                    schema_row.clone(),
                    None,
                    self.encryption_key.clone(),
                );
                r_row.push(Box::new(concat_reader));
            } else {
                for col_file in &columnar_level.files {
                    if !col_file.has_table(schema_row.table_id) {
                        continue;
                    }
                    let col_reader = ColumnarTableReader::new(
                        col_file,
                        schema_row.clone(),
                        None,
                        self.encryption_key.clone(),
                    )?;
                    r_row.push(Box::new(col_reader));
                }
            }
        }
        let r_merged = ColumnarMergeReader::new(schema_row.clone(), r_row);
        let r_mvcc = ColumnarMvccReader::new(Box::new(r_merged), &schema_row, read_ts);
        Ok(FtsBruteForceReader::new(r_mvcc, &output_schema, query)?)
    }

    fn new_fts_reader_inner(
        &self,
        runtime: &Handle,
        table_id: i64,
        query: clara_fts::Query,
        tracked_index_layout: crate::table::fts::TrackedIndexLayout,
        output_schema: Schema,
        read_ts: u64,
        // Optional handle bounds for pruning FTS index search.
        //
        // NOTE: int handles are encoded in TiDB row-key format (memcomparable i64,
        // i.e. `IntPk::encode`), NOT the columnar in-block format (i64 little-endian).
        // Common handles are raw bytes.
        start_handle: Option<Bytes>,
        end_handle: Option<Bytes>,
    ) -> Result<Option<Box<dyn ColumnarFilterReader>>> {
        if !self.read_columnar {
            return Ok(None);
        }
        let Some(delta_index_schema) = self.new_fts_delta_index_schema(
            table_id,
            query.info().get_index_id(),
            &tracked_index_layout,
        )?
        else {
            return Ok(None);
        };
        // FTS reader chain overview (handles are int/common depending on schema):
        //
        //   A) Indexed hits (source columns=ignored placeholders, Score=from index)
        //      FtsIndexReader -> FtsIndexColumnarReader
        //
        //   B) Unindexed data (source columns=from data, Score=NULL)
        //      memtables + unconverted row L0s + untracked columnar L0s
        //
        //   (A + B) -> ColumnarMergeReader
        //      -> ColumnarMvccReader
        //      -> FtsBruteForceCondReader  (filter and consume query source columns)
        //      -> FtsJoinReader            (load the final projection from row storage)
        //
        // Reader schema flow:
        //   merge/MVCC: (QuerySourceColumn..., NullableScore)
        //   conditional output: (NonNullableScore) or no data columns
        //   join output: `output_schema`
        //
        // Note: FTS index only covers columnar L0s up to `fts_levels.l0_snap_version`,
        // except `fts_levels.pending_l0_ids` which are explicitly not covered.
        //
        let read_row_schema = FtsBruteForceCondReader::build_inner_schema(
            &output_schema,
            &delta_index_schema,
            &query,
        )?;
        let readers = crate::table::fts::build_fts_delta_readers(
            &self.fts_delta_cache,
            &self.tag,
            table_id,
            &query,
            tracked_index_layout,
            delta_index_schema,
            read_row_schema.clone(),
            read_ts,
            Arc::clone(&self.data.fts_levels),
            &self.data.mem_tbls,
            &self.data.col_levels.unconverted_l0s,
            &self.data.col_levels.levels[0].files,
            self.data.blob_tbl_map.clone(),
            self.encryption_key.clone(),
            start_handle,
            end_handle,
        )?;

        // Merge + MVCC filtering.
        let merged = ColumnarMergeReader::new(read_row_schema.clone(), readers);
        let mvcc = ColumnarMvccReader::new(Box::new(merged), &read_row_schema, read_ts);

        // Do conditional brute force after MVCC. It consumes query source
        // columns and returns only handle/version plus an optional non-null
        // score. Join then loads all projected physical columns from row
        // storage.
        let cond = FtsBruteForceCondReader::new(mvcc, &query)?;

        let row_iter =
            runtime.block_on(self.new_table_iterator_async(WRITE_CF, false, false, None));
        let join = FtsJoinReader::new(
            cond,
            vec![row_iter],
            output_schema,
            self.data.blob_tbl_map.clone(),
            self.encryption_key.clone(),
        )?;

        Ok(Some(Box::new(join)))
    }

    /// Creates an FTS reader after building a fresh execution query against
    /// this shard snapshot's physical index layout for `table_id`.
    ///
    /// The request-level [`tipb::FtsQueryInfo`] may be shared by callers, but
    /// the resulting [`clara_fts::Query`] and its lazily prepared BM25 state
    /// are scoped to this reader construction.
    pub fn new_fts_reader(
        &self,
        runtime: &Handle,
        table_id: i64,
        query_info: &tipb::FtsQueryInfo,
        schema: Schema,
        read_ts: u64,
        // Optional handle bounds for pruning FTS index search.
        //
        // NOTE: int handles are encoded in TiDB row-key format (memcomparable i64,
        // i.e. `IntPk::encode`), NOT the columnar in-block format (i64 little-endian).
        // Common handles are raw bytes.
        start_handle: Option<Bytes>,
        end_handle: Option<Bytes>,
    ) -> Result<Box<dyn ColumnarFilterReader>> {
        let index_id = query_info.get_index_id();
        let tracked_index_layout = self
            .data
            .fts_levels
            .tracked_index_layout(table_id, index_id)
            .cloned()
            .ok_or_else(|| {
                Error::Other(
                    format!(
                        "fts query: layout resolution failed: unresolved tracked FTS layout for query: table_id {}, index_id {}",
                        table_id, index_id
                    )
                    .into(),
                )
            })?;
        let query = clara_fts::Query::new_with_index_layout(
            query_info.clone(),
            &tracked_index_layout.index_fields,
            &tracked_index_layout.index_layout,
        )
        .map_err(|e| Error::Other(format!("fts query: construction failed: {e}").into()))?;
        validate_schema(&schema, &query).map_err(|e| {
            Error::Other(format!("fts query: output schema validation failed: {e}").into())
        })?;
        if let Some(fts_reader) = self.new_fts_reader_inner(
            runtime,
            table_id,
            query.clone(),
            tracked_index_layout,
            schema.clone(),
            read_ts,
            start_handle,
            end_handle,
        )? {
            Ok(fts_reader)
        } else {
            warn!(
                "{} fallback to brute-force FTS reader for table_id {}",
                self.tag, table_id
            );
            Ok(Box::new(
                self.new_fts_reader_fallback(&query, schema, read_ts)?,
            ))
        }
    }

    pub fn new_vector_index_reader(
        &self,
        table_id: i64,
        index_id: i64,
        col_id: i64,
        target: &[f32],
        top_k: usize,
        schema: Schema,
        read_ts: u64,
        start_handle: Option<&[u8]>,
        end_handle: Option<&[u8]>,
        ann_query: Arc<AnnQueryInfo>,
    ) -> Result<Option<ColumnarMvccReader>> {
        if !self.read_columnar {
            return Ok(None);
        }
        if !self.data.col_levels.levels[2].files.is_empty() {
            // Make sure the l2_snap_version is set.
            debug_assert!(self.data.col_levels.l2_snap_version.is_not_zero());
        }

        let Some(vector_index) = self.data.vector_indexes.get(table_id, index_id, col_id) else {
            return Ok(None);
        };
        if vector_index.files.is_empty() {
            return Ok(None);
        }
        let enable_distance_proj = ann_query.get_enable_distance_proj();
        let metric = ann_query.get_distance_metric();

        let vector_items_reader = match VectorItemsReader::new(
            schema.clone(),
            vector_index,
            target,
            top_k,
            read_ts,
            start_handle,
            end_handle,
            &self.data.col_levels,
            self.encryption_key.clone(),
            metric,
            enable_distance_proj,
        ) {
            Ok(reader) => reader,
            Err(e) => {
                warn!("{} failed to search vector index {:?}", self.tag, e);
                return Ok(None);
            }
        };

        // When vector search distance proj is enabled, TableScan's schema
        // contains a VirtualDistance column at the end, and does not
        // contain the vector column. However, when we are reading from
        // data that vector index is not built yet, we need to read out
        // the vector column in order to calculate the distance column
        // that TableScan asks for.
        let schema_with_vec = if enable_distance_proj {
            VectorDistanceProjector::generate_inner_schema(&schema, &ann_query)?
        } else {
            schema.clone()
        };

        // For all readers except for the vector index reader, they will always read
        // with the vector column. We wrap a distance projection transformation
        // over these column readers so that they finally produce the expected output.
        // For vector index reader, it produces the vector distance column directly.
        let mut readers = self.collect_column_row_readers(&schema_with_vec, read_ts)?;
        let mem_l0_readers_count = readers.len();
        if enable_distance_proj {
            readers = readers
                .into_iter()
                .map(|r| {
                    VectorDistanceProjector::new(r, schema.clone(), ann_query.clone())
                        .map(|vr| Box::new(vr) as Box<dyn ColumnarReader>)
                        .map_err(|e| Error::Other(e.to_string().into()))
                })
                .collect::<Result<Vec<_>>>()?;
        }

        let mut columnar_readers_count = 0;
        for columnar_level in &self.data.col_levels.levels {
            if columnar_level.level == 2 {
                // level 2 columnar files are all included in the vector index.
                break;
            }
            for file in &columnar_level.files {
                // NOTE: the snap_version of columnar file maybe smaller than the vector index
                // and not generated the vector index after region merge. We should also need to
                // read the columnar file.
                if vector_index.contains_columnar_file(file) {
                    continue;
                }
                if !file.has_table(schema.table_id) {
                    continue;
                }
                let col_reader = ColumnarTableReader::new(
                    file,
                    schema_with_vec.clone(),
                    None,
                    self.encryption_key.clone(),
                )?;
                // The same as above, in distance projection, we wrap a distance projector over
                // the reader who reads the vector column.
                if enable_distance_proj {
                    let col_reader_distance = VectorDistanceProjector::new(
                        Box::new(col_reader),
                        schema.clone(),
                        ann_query.clone(),
                    )?;
                    readers.push(Box::new(col_reader_distance));
                } else {
                    readers.push(Box::new(col_reader));
                }
                columnar_readers_count += 1;
            }
        }
        readers.push(Box::new(vector_items_reader));
        info!(
            "{} use vector index reader, vector_index_file_count: {}, mem_l0_readers_count: {}, columnar_readers_count: {}",
            self.tag,
            vector_index.files.len(),
            mem_l0_readers_count,
            columnar_readers_count
        );
        let merged_reader = ColumnarMergeReader::new(schema.clone(), readers);
        let mvcc_reader = ColumnarMvccReader::new(Box::new(merged_reader), &schema, read_ts);
        Ok(Some(mvcc_reader))
    }

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

    /// Returns whether this snapshot is from an IA/Auto-IA region.
    #[inline]
    pub fn from_ia_region(&self) -> bool {
        self.storage_class_spec.can_be_ia()
    }

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

    /// Get IA segments which are in DFS. Used to prefetch in parallel.
    // TODO: Support columnar tables
    pub fn get_ia_remote_segments(
        &self,
        outer_range: (&[u8], &[u8]),
    ) -> Result<(
        Vec<(FileSegmentIdent, FileType)>,
        usize, // total_segments
    )> {
        let start_key = InnerKey::from_outer_key(outer_range.0);
        let end_key = InnerKey::from_outer_end_key(outer_range.1);
        let data_bound = DataBound::new(start_key, end_key, false);

        let mut segments = vec![];
        let mut total_segments = 0;
        let (sstables, blob_tables) = self.get_overlap_async_tables(data_bound);
        for t in sstables {
            let (segs, total) = t.get_remote_segments(data_bound)?;
            segments.extend(segs.into_iter().map(|ident| (ident, FileType::Sst)));
            total_segments += total;
        }
        for blob_table in blob_tables {
            let (segs, total) = blob_table.get_remote_segments(data_bound)?;
            segments.extend(segs.into_iter().map(|ident| (ident, FileType::Blob)));
            total_segments += total;
        }
        Ok((segments, total_segments))
    }

    fn get_overlap_async_tables(
        &self,
        data_bound: DataBound<'_>,
    ) -> (Vec<SsTable>, Vec<BlobTable>) {
        let mut tables = vec![];
        for lh in &self.data.cfs[WRITE_CF].levels {
            let (left, right) = data_bound.get_overlap_data_sets(&lh.tables);
            for t in &lh.tables[left..right] {
                if !t.is_sync() {
                    tables.push(t.clone());
                }
            }
        }
        let mut blob_tables = vec![];
        for blob_table in self.data.blob_tbl_map.values() {
            if !blob_table.is_sync() {
                blob_tables.push(blob_table.clone());
            }
        }
        (tables, blob_tables)
    }

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

    // Filter the columnar table ids that are not empty.
    // `table_ids` is the list of table ids to filter.
    // Returns the list of columnar table ids that are not empty. If the table_id is
    // not in the snapshot, it will be filtered out.
    // NOTE: The caller should guarantee that table_ids is sorted.
    pub fn filter_columnar_table_ids_not_empty(&self, table_ids: &[i64]) -> HashSet<i64> {
        if table_ids.is_empty() {
            return HashSet::new();
        }
        let columnar_table_ids = self
            .data
            .columnar_table_ids
            .iter()
            .cloned()
            .collect::<HashSet<_>>();
        let mut filtered_table_ids = HashSet::new();
        let mut iter = self.new_memtable_unconverted_l0_iterator();
        let mut delta_data_table_id = table_ids[0];
        for &table_id in table_ids {
            if !columnar_table_ids.contains(&table_id) {
                continue;
            }

            // Optimize the seek if there are lots of tables are empty.
            if table_id >= delta_data_table_id {
                // Check data exists in the snapshot delta data.
                let table_key_prefix = encode_table_prefix_key(table_id);
                iter.seek(table_key_prefix.as_ref());
                if iter.valid() {
                    let key = iter.key();
                    delta_data_table_id = decode_table_id(key.deref()).unwrap_or(i64::MAX);
                    if key.starts_with(table_key_prefix.as_ref().deref()) {
                        filtered_table_ids.insert(table_id);
                        continue;
                    }
                } else {
                    delta_data_table_id = i64::MAX;
                }
            }

            // Check data exists in the snapshots columnar files.
            if self.columnar_files_contains_table(table_id) {
                filtered_table_ids.insert(table_id);
            }
        }
        filtered_table_ids
    }

    fn new_memtable_unconverted_l0_iterator(&self) -> Box<dyn table::Iterator> {
        let mut iters: Vec<Box<dyn table::Iterator>> = Vec::new();
        for mem_tbl in &self.data.mem_tbls {
            iters.push(mem_tbl.get_cf(WRITE_CF).new_iterator(false));
        }
        for l0 in &self.data.col_levels.unconverted_l0s {
            let Some(l0_write_cf) = l0.get_cf(WRITE_CF) else {
                continue;
            };
            iters.push(l0_write_cf.new_iterator(false, false));
        }
        table::new_merge_iterator(iters, false)
    }

    fn columnar_files_contains_table(&self, table_id: i64) -> bool {
        for columnar_level in &self.data.col_levels.levels {
            for file in &columnar_level.files {
                if file.has_table(table_id) {
                    return true;
                }
            }
        }
        false
    }

    // validate_cached_value validate the cached value for the given inner_key and
    // snap_version.
    // By checking the mem tables with greater snap_version, we ensure the cached
    // value is still valid.
    // If there are newer values the newer values is returned.
    // If there maybe newer values in the persisted tables, None is returned.
    pub(crate) fn validate_cached_value(
        &self,
        inner_key: InnerKey<'_>,
        snap_version: SnapVersion,
        out_val_owner: &mut Vec<u8>,
    ) -> Option<table::Value> {
        for i in 0..self.data.mem_tbls.len() {
            let cf_tbl = &self.data.mem_tbls[i];
            let mem_tbl_snap_version = cf_tbl.get_snap_version();
            if mem_tbl_snap_version.is_not_zero() && mem_tbl_snap_version < snap_version {
                return Some(Value::new());
            }
            let tbl = cf_tbl.get_cf(WRITE_CF);
            let v = tbl.get(inner_key, u64::MAX, out_val_owner);
            if v.is_valid() {
                return Some(v);
            }
        }
        if self.data.persisted_version < snap_version {
            return Some(Value::new());
        }
        None
    }

    // To ensure that async `SnapAccess` must be called by async methods.
    // The checking in `IaFile` may be not reached when the block has been
    // loaded.
    #[inline]
    fn check_sync(&self, cf: usize) {
        debug_assert!(self.is_cf_sync(cf));
    }

    #[inline]
    async fn check_sync_async(&self, _: usize) {}
}

struct ExtraCfPrefixBound {
    lower: OwnedInnerKey,
    upper: OwnedInnerKey,
}

impl ExtraCfPrefixBound {
    // This helper accepts the raw user key, not the full EXTRA_CF key.
    // EXTRA_CF physical keys are encoded as `raw_key + !start_ts`,
    // so pruning by key prefix needs the timestamp-free raw key to cover
    // all status records for the same user key.
    // `OwnedInnerKey::new` normalizes the raw outer key into the table-layer inner
    // key before comparing it with table data bounds. This relies on TiDB
    // mem-comparable keys being prefix-free.
    fn new(raw_key: &[u8]) -> Self {
        let lower = OwnedInnerKey::new(Bytes::copy_from_slice(raw_key));
        let mut upper_buf = Vec::with_capacity(raw_key.len() + U64_SIZE);
        upper_buf.extend_from_slice(raw_key);
        upper_buf.resize(raw_key.len() + U64_SIZE, 0xFF);
        let upper = OwnedInnerKey::new(Bytes::from(upper_buf));
        Self { lower, upper }
    }

    fn as_data_bound(&self) -> DataBound<'_> {
        DataBound::new(self.lower.as_ref(), self.upper.as_ref(), true)
    }
}

pub struct Iterator {
    all_versions: bool,
    reversed: bool,
    read_ts: u64,
    // If true, the iterator has met a newer version than read_ts during iteration.
    met_newer_ts: bool,
    pub(crate) key: BytesMut,
    val: table::Value,
    skipped_tombstones: usize,
    pub(crate) inner: Box<dyn table::Iterator>,
    is_blob_value_record: bool,
    iter_stats: IterStats,
    blob_prefetcher: Option<BlobPrefetcher>,
    owned_blob: Option<Vec<u8>>,
    encryption_key: Option<EncryptionKey>,
    skip_blob: bool,
    data: ShardData,
    range: Option<(OwnedInnerKey, OwnedInnerKey)>, // [inner_lower_bound, inner_upper_bound)
}

impl Iterator {
    pub fn valid(&self) -> bool {
        self.val.is_valid()
    }

    pub fn key(&self) -> &[u8] {
        self.key.chunk()
    }

    pub fn val(&mut self) -> &[u8] {
        if !self.skip_blob && self.val.is_blob_ref() {
            if let Some(prefetcher) = &mut self.blob_prefetcher {
                let blob_ref = self.val.get_blob_ref();
                if !self.is_blob_value_record {
                    self.is_blob_value_record = true;
                    self.iter_stats.total_size += blob_ref.original_len as usize;
                }
                return prefetcher.get(&blob_ref).unwrap_or_else(|e| {
                    panic!("failed to get blob, blob_ref: {:?}, err: {:?}", blob_ref, e)
                });
            }
        }
        self.val.get_value()
    }

    pub fn version(&self) -> u64 {
        self.val.version
    }

    pub fn meta(&self) -> u8 {
        self.val.meta
    }

    pub fn user_meta(&self) -> &[u8] {
        self.val.user_meta()
    }

    pub fn valid_for_prefix(&self, prefix: &[u8]) -> bool {
        self.key.starts_with(prefix)
    }

    /// Returns and clears the number of internal tombstones skipped since the
    /// previous call.
    ///
    /// Call this right after seek/range-positioning or next-style operations so
    /// the caller can attribute the skipped tombstones to the matching scan
    /// statistic bucket.
    pub fn take_skipped_tombstones(&mut self) -> usize {
        std::mem::take(&mut self.skipped_tombstones)
    }

    #[maybe_async::both]
    pub async fn next(&mut self) {
        let timer = Instant::now();

        if self.all_versions
            && self.valid()
            && next_version!(self.inner, &mut self.iter_stats).await
            && !self.inner.value().is_deleted()
        {
            self.update_item();
            self.maybe_resolve_blob_ref().await;
            return;
        }
        next!(self.inner, &mut self.iter_stats).await;
        self.parse_item().await;

        ENGINE_SEEK_DURATION_STATIC
            .next
            .observe(duration_to_sec(timer.saturating_elapsed()));
    }

    fn update_item(&mut self) {
        self.is_blob_value_record = false;
        self.key.truncate(self.data.keyspace_prefix_len());
        self.key.extend_from_slice(self.inner.key().deref());
        self.val = self.inner.value();
    }

    fn record_inner_kv(&mut self) {
        if self.inner.valid() {
            self.iter_stats
                .record_kv(self.inner.key().len(), self.inner.value().value_len());
        }
    }

    #[maybe_async::both]
    async fn maybe_resolve_blob_ref(&mut self) {
        if self.skip_blob || self.blob_prefetcher.is_some() || !self.val.is_blob_ref() {
            self.owned_blob = None;
            return;
        }
        let blob_ref = self.val.get_blob_ref();
        let blob_table = self
            .data
            .blob_tbl_map
            .get(&blob_ref.fid)
            .unwrap_or_else(|| {
                panic!("blob table not found, blob_ref: {:?}", blob_ref);
            });
        let mut decryption_buf = vec![];
        let blob = blob_table
            .get_by_ref(&blob_ref, &mut decryption_buf, self.encryption_key.clone())
            .await
            .unwrap_or_else(|e| {
                panic!("failed to get blob, blob_ref: {:?}, err: {:?}", blob_ref, e)
            });
        // record extra blob value length.
        self.iter_stats.total_size += blob.len();
        self.owned_blob = Some(blob);
        self.val.fill_in_blob(self.owned_blob.as_ref().unwrap());
    }

    #[maybe_async::both]
    async fn parse_item(&mut self) {
        while self.inner.valid() {
            if self.is_inner_key_over_bound() {
                break;
            }
            let val = self.inner.value();
            if val.version > self.read_ts {
                self.met_newer_ts = true;
                if !self
                    .inner
                    .seek_to_version(self.read_ts, &mut self.iter_stats)
                    .await
                {
                    next!(self.inner, &mut self.iter_stats).await;
                    continue;
                }
            }
            if self.inner.value().is_deleted() {
                self.skipped_tombstones += 1;
                next!(self.inner, &mut self.iter_stats).await;
                continue;
            }
            self.update_item();
            self.maybe_resolve_blob_ref().await;
            return;
        }
        self.owned_blob = None;
        self.val = table::Value::new();
    }

    // seek would seek to the provided key if present. If absent, it would seek to
    // the next smallest key greater than provided if iterating in the forward
    // direction. Behavior would be reversed is iterating backwards.
    #[maybe_async::both]
    pub async fn seek(&mut self, key: &[u8]) {
        let timer = Instant::now();

        let (inner_lower_bound, inner_upper_bound) = self.current_inner_range_owned();
        if key.len() <= self.data.inner_key_off {
            self.inner.rewind().await;
        } else {
            self.inner.seek(InnerKey::from_outer_key(key)).await;
        }
        self.record_inner_kv();
        self.clamp_cursor_to_bounds(inner_lower_bound.as_ref(), inner_upper_bound.as_ref())
            .await;
        self.parse_item().await;

        ENGINE_SEEK_DURATION_STATIC
            .seek
            .observe(duration_to_sec(timer.saturating_elapsed()));
    }

    fn current_inner_range(&self) -> (InnerKey<'_>, InnerKey<'_>) {
        if let Some((inner_lower, inner_upper)) = &self.range {
            (inner_lower.as_ref(), inner_upper.as_ref())
        } else {
            (self.data.inner_start(), self.data.inner_end())
        }
    }

    fn current_inner_range_owned(&self) -> (OwnedInnerKey, OwnedInnerKey) {
        if let Some((inner_lower, inner_upper)) = &self.range {
            (inner_lower.clone(), inner_upper.clone())
        } else {
            (self.data.inner_start().into(), self.data.inner_end().into())
        }
    }

    #[maybe_async::both]
    async fn clamp_cursor_to_bounds(
        &mut self,
        inner_lower_bound: InnerKey<'_>,
        inner_upper_bound: InnerKey<'_>,
    ) {
        // The upper bound is exclusive.
        if self.inner.valid() && self.inner.key() == inner_upper_bound {
            next!(self.inner, &mut self.iter_stats).await;
        }
        if !self.inner.valid() {
            return;
        }
        if self.reversed {
            if self.inner.key() >= inner_upper_bound {
                self.inner.seek(inner_upper_bound).await;
                self.record_inner_kv();
                if self.inner.valid() && self.inner.key() == inner_upper_bound {
                    next!(self.inner, &mut self.iter_stats).await;
                }
            }
        } else if self.inner.key() < inner_lower_bound {
            self.inner.seek(inner_lower_bound).await;
            self.record_inner_kv();
        }
    }

    // rewind would rewind the iterator cursor all the way to zero-th position,
    // which would be the smallest key if iterating forward, and largest if
    // iterating backward. It does not keep track of whether the cursor started
    // with a seek().
    #[maybe_async::both]
    pub async fn rewind(&mut self) {
        let timer = Instant::now();

        let (inner_lower_bound, inner_upper_bound) = self.current_inner_range_owned();
        self.inner.rewind().await;
        self.record_inner_kv();
        self.clamp_cursor_to_bounds(inner_lower_bound.as_ref(), inner_upper_bound.as_ref())
            .await;
        self.parse_item().await;

        ENGINE_SEEK_DURATION_STATIC
            .seek
            .observe(duration_to_sec(timer.saturating_elapsed()));
    }

    pub fn set_all_versions(&mut self, all_versions: bool) {
        self.all_versions = all_versions;
    }

    pub fn is_reverse(&self) -> bool {
        self.reversed
    }

    // set the new range of the iterator, it the range is monotonic, we can avoid
    // seek. return true if seek is performed.
    #[maybe_async::both]
    #[allow(clippy::collapsible_else_if)]
    pub async fn set_range(
        &mut self,
        outer_lower_bound_include: Bytes,
        outer_upper_bound_exclude: Bytes,
    ) -> bool {
        // Restrict the iterator range to the snapshot bounds, it's important to
        // ensure the iterator is within the snapshot range.
        let snapshot_lower_bound = self.data.outer_start.clone();
        let snapshot_upper_bound = self.data.outer_end.clone();
        let inner_lower_bound = std::cmp::max(
            InnerKey::from_outer_key(&snapshot_lower_bound),
            InnerKey::from_outer_key(&outer_lower_bound_include),
        );
        let inner_upper_bound = std::cmp::min(
            InnerKey::from_outer_end_key(&snapshot_upper_bound),
            InnerKey::from_outer_end_key(&outer_upper_bound_exclude),
        );
        let mut seeked = false;
        // reset monotonic range can be optimized to avoid seek.
        if self.is_reset_monotonic_range(inner_lower_bound, inner_upper_bound) {
            // If inner is not valid, the iterator has reached the end, there is no more
            // data to return, we can avoid the seek.
            if self.inner.valid() {
                if self.reversed {
                    // If the new inner_upper_bound is greater than the current key, we can
                    // continue to use the current key to iterate backward, avoid the seek.
                    if self.inner.key() > inner_upper_bound {
                        self.inner.seek(inner_upper_bound).await;
                        self.record_inner_kv();
                        seeked = true;
                    }
                    // the upper bound is exclusive, so we need to skip the current key.
                    if self.inner.key() == inner_upper_bound {
                        next!(self.inner, &mut self.iter_stats).await;
                    }
                } else {
                    // If the new inner_lower_bound is greater than or equal to the current key,
                    // we can continue to use the current key to iterate forward, avoid the seek.
                    if self.inner.key() < inner_lower_bound {
                        self.inner.seek(inner_lower_bound).await;
                        self.record_inner_kv();
                        seeked = true;
                    }
                }
            }
        } else {
            // always seek if not reset monotonic range.
            if self.reversed {
                self.inner.seek(inner_upper_bound).await;
                self.record_inner_kv();
                if self.inner.valid() && self.inner.key() == inner_upper_bound {
                    next!(self.inner, &mut self.iter_stats).await;
                }
            } else {
                self.inner.seek(inner_lower_bound).await;
                self.record_inner_kv();
            }
            seeked = true;
        }
        self.range = Some((inner_lower_bound.into(), inner_upper_bound.into()));
        self.parse_item().await;
        seeked
    }

    fn is_reset_monotonic_range(
        &self,
        inner_lower_bound: InnerKey<'_>,
        inner_upper_bound: InnerKey<'_>,
    ) -> bool {
        if let Some((inner_old_lower, inner_old_upper)) = &self.range {
            if self.reversed {
                inner_upper_bound <= inner_old_lower.as_ref()
            } else {
                inner_old_upper.as_ref() <= inner_lower_bound
            }
        } else {
            false
        }
    }

    pub(crate) fn is_inner_key_over_bound(&self) -> bool {
        if !self.inner.valid() {
            return true;
        }
        let (inner_lower_bound, inner_upper_bound) = self.current_inner_range();
        self.inner.key() < inner_lower_bound || self.inner.key() >= inner_upper_bound
    }

    pub fn take_iter_stats(&mut self) -> IterStats {
        std::mem::take(&mut self.iter_stats)
    }

    pub fn iter_stats(&self) -> IterStats {
        self.iter_stats
    }

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

#[derive(Debug, Clone)]
pub struct TableCtx {
    pub table_id: i64,
    pub ranges: Vec<tipb::KeyRange>,
}

fn columns_allow_pack_clean_read(columns: &[ColumnInfo]) -> bool {
    columns.iter().all(|col| {
        !col.get_pk_handle()
            && col.get_column_id() != HANDLE_COL_ID as i64
            && col.get_column_id() != VERSION_COL_ID as i64
    })
}

fn scan_ctx_has_filter(scan_ctx: Option<&TableScanCtx>) -> bool {
    scan_ctx.map(|ctx| !ctx.is_empty_filter()).unwrap_or(false)
}

fn l2_files_allow_pack_clean_read(files: &[ColumnarFile], schema: &Schema, read_ts: u64) -> bool {
    let mut bounds = vec![];
    for file in files {
        if !file.has_table(schema.table_id) {
            continue;
        }
        let Ok(table_meta) = file.try_get_table(schema.table_id) else {
            return false;
        };
        if table_meta.handle_index.buf.length() <= table_meta.handle_column.pack_offsets.num_packs()
        {
            return false;
        }
        if !table_meta.has_pack_mvcc_stats()
            || table_meta
                .max_version
                .unwrap_or_else(|| file.get_max_version())
                > read_ts
        {
            return false;
        }
        let start = table_meta.handle_index.buf.get_not_null_value(0).to_vec();
        let end = table_meta
            .handle_index
            .buf
            .get_not_null_value(table_meta.handle_index.buf.length() - 1)
            .to_vec();
        bounds.push((start, end));
    }
    if bounds.is_empty() {
        return false;
    }
    if schema.is_common_handle() {
        bounds.sort_by(|(lhs, _), (rhs, _)| lhs.cmp(rhs));
        return bounds.windows(2).all(|pair| pair[0].1 <= pair[1].0);
    }

    bounds.sort_by_key(|(start, _)| {
        let mut start = start.as_slice();
        start.get_i64_le()
    });
    bounds.windows(2).all(|pair| {
        let mut prev_end = pair[0].1.as_slice();
        let mut next_start = pair[1].0.as_slice();
        prev_end.get_i64_le() <= next_start.get_i64_le()
    })
}

/// CloudColumnarReaders coordinates reading columnar data from multiple tables,
/// dispatching per-table work onto worker threads to leverage CPU parallelism.
pub struct CloudColumnarReaders {
    runtime: Handle,
    snap_access: SnapAccess,
    tables: Vec<TableCtx>,
    columns: Arc<Vec<ColumnInfo>>,
    scan_ctx: TableScanCtx,
    ann_query: Option<Arc<tipb::AnnQueryInfo>>,
    fts_query_info: Option<Arc<tipb::FtsQueryInfo>>,
    ia_ctx: IaCtx,
    start_ts: u64,
    concurrency: usize,
    reader: Option<CloudColumnarReader>,
    result_rx: Option<mpsc::Receiver<ReadResult>>,
    current_block: Option<BlockResult>,
    read_limit: Option<usize>,
    finished: bool,
    runtime_stats: ColumnarRuntimeStats,
    concurrent_runtime_stats: Option<Arc<Mutex<ColumnarRuntimeStats>>>,
}

impl CloudColumnarReaders {
    pub fn new(
        runtime: Handle,
        snap_access: SnapAccess,
        tables: Vec<TableCtx>,
        columns: Vec<ColumnInfo>,
        scan_ctx: TableScanCtx,
        ann_query_info: Option<tipb::AnnQueryInfo>,
        fts_query_info: Option<FtsQueryInfo>,
        ia_ctx: IaCtx,
        start_ts: u64,
        concurrency: usize,
    ) -> Result<Self> {
        let mut table_ids = tables
            .iter()
            .map(|table| table.table_id)
            .collect::<Vec<_>>();
        table_ids.sort_unstable();
        let filtered_table_ids = snap_access.filter_columnar_table_ids_not_empty(&table_ids);
        let tables = tables
            .into_iter()
            .filter(|table| filtered_table_ids.contains(&table.table_id))
            .collect::<Vec<_>>();
        let concurrency = if tables.len() > 1 && concurrency > 1 {
            concurrency
        } else {
            0
        };

        if let Some(info) = fts_query_info.as_ref() {
            clara_fts::Query::validate_info(info).map_err(|e| {
                crate::Error::Other(format!("fts query: validation failed: {e}").into())
            })?;
        }
        let ann_query = ann_query_info.map(Arc::new);
        let fts_query_info = fts_query_info.map(Arc::new);

        Ok(Self {
            runtime,
            snap_access,
            tables,
            columns: Arc::new(columns),
            scan_ctx,
            ann_query,
            fts_query_info,
            ia_ctx,
            start_ts,
            concurrency,
            reader: None,
            result_rx: None,
            current_block: None,
            read_limit: None,
            finished: false,
            runtime_stats: ColumnarRuntimeStats::default(),
            concurrent_runtime_stats: None,
        })
    }

    fn log_error_with_ctx(
        tag: ShardTag,
        table_ctx: &TableCtx,
        scan_ctx: &TableScanCtx,
        columns: &Arc<Vec<ColumnInfo>>,
        ann_query: &Option<Arc<tipb::AnnQueryInfo>>,
        fts_query_info: &Option<Arc<tipb::FtsQueryInfo>>,
        start_ts: u64,
        err: &crate::Error,
    ) {
        struct FtsQueryInfoRedacted<'a>(&'a tipb::FtsQueryInfo);

        impl std::fmt::Debug for FtsQueryInfoRedacted<'_> {
            fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
                let info = self.0;
                f.debug_struct("FtsQueryInfoRedacted")
                    .field("query_type", &info.get_query_type())
                    .field("index_id", &info.get_index_id())
                    .field("top_k", &info.get_top_k())
                    .field("tokenizer", &info.get_query_tokenizer())
                    .field("column_names", &info.get_column_names())
                    .field("query_text_len", &info.get_query_text().len())
                    .field("query_func", &info.get_query_func())
                    .field("match_expr", &info.get_match_expr())
                    .finish()
            }
        }

        let columns_summary = columns
            .iter()
            .map(|c| (c.get_column_id(), c.get_tp(), c.get_pk_handle()))
            .collect::<Vec<_>>();
        let ann_summary = ann_query.as_ref().map(|q| {
            let vec_bytes_len = q.get_ref_vec_f32().len();
            let vec_f32_len = q
                .get_ref_vec_f32()
                .read_vector_float32()
                .ok()
                .map(|v| v.as_ref().len());
            let vec_col_id = get_ann_vec_col_id(q.as_ref()).ok();
            (
                q.get_index_id(),
                q.get_top_k(),
                vec_col_id,
                vec_bytes_len,
                vec_f32_len,
            )
        });
        let fts_query_info = fts_query_info
            .as_ref()
            .map(|info| FtsQueryInfoRedacted(info));
        error!(
            "CloudColumnarReader::new failed, tag={} table_id={} ranges_len={} start_ts={} scan_ctx={:?} columns_len={} columns={:?} ann_query={:?} fts_query={:?} err={}",
            tag,
            table_ctx.table_id,
            table_ctx.ranges.len(),
            start_ts,
            scan_ctx,
            columns.len(),
            columns_summary,
            ann_summary,
            fts_query_info,
            err
        );
    }

    fn need_concurrency(&self) -> bool {
        self.concurrency > 0
    }

    /// Number of tables still owned by this coordinator: the active sequential
    /// reader (if any) plus queued table contexts.
    fn lm_table_count(&self) -> usize {
        self.tables.len() + usize::from(self.reader.is_some())
    }

    /// Hard gate for all LM entry points. Sequential multi-table iteration and
    /// worker aggregation both break the single pending-batch LM protocol.
    fn ensure_lm_single_table(&self) -> Result<()> {
        if self.need_concurrency() {
            return Err(crate::Error::Other(
                "late materialization does not support multi-table worker aggregation"
                    .to_string()
                    .into(),
            ));
        }
        if self.lm_table_count() > 1 {
            return Err(crate::Error::Other(
                format!(
                    "late materialization does not support sequential multi-table readers, table_count={}",
                    self.lm_table_count()
                )
                .into(),
            ));
        }
        Ok(())
    }

    pub fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.runtime_stats;
        if let Some(reader) = self.reader.as_ref() {
            let reader_stats = reader.runtime_stats();
            stats.merge(&reader_stats);
        }
        if let Some(shared_stats) = self.concurrent_runtime_stats.as_ref() {
            let shared = shared_stats.lock().unwrap();
            stats.merge(&shared);
        }
        stats
    }

    pub fn ffi_read_block(&mut self, read_limit: usize) -> Result<usize> {
        // If there is only one table, we use sequential read.
        if !self.need_concurrency() {
            return self.sequential_read_block(read_limit);
        }

        if self.finished {
            self.current_block = None;
            return Ok(0);
        }
        if read_limit == 0 {
            self.finished = true;
            self.current_block = None;
            return Ok(0);
        }
        if self.result_rx.is_none() {
            self.start_workers(read_limit)?;
        }
        if let Some(expected) = self.read_limit {
            debug_assert_eq!(expected, read_limit);
        }
        let Some(rx) = self.result_rx.as_ref() else {
            return Ok(0);
        };
        match rx.recv() {
            Ok(ReadResult::Block(block)) => {
                let read_count = block.read_count;
                self.current_block = Some(block);
                Ok(read_count)
            }
            Ok(ReadResult::Finished) => {
                self.finished = true;
                self.current_block = None;
                Ok(0)
            }
            Ok(ReadResult::Error(err)) => {
                self.finished = true;
                self.current_block = None;
                Err(err)
            }
            Err(_) => {
                self.finished = true;
                self.current_block = None;
                Ok(0)
            }
        }
    }

    pub fn ffi_read_early_block(
        &mut self,
        read_limit: usize,
        early_column_ids: &[i64],
    ) -> Result<Option<(u64, usize, i64)>> {
        self.ensure_lm_single_table()?;
        if read_limit == 0 {
            return Ok(None);
        }

        loop {
            if self.reader.is_none() {
                let Some(table_ctx) = self.tables.first().cloned() else {
                    return Ok(None);
                };
                self.tables.remove(0);
                self.reader = Some(self.make_sequential_reader(table_ctx)?);
            }

            let result = self
                .reader
                .as_mut()
                .unwrap()
                .ffi_read_early_block(read_limit, early_column_ids)?;
            if result.is_some() {
                return Ok(result);
            }

            let drained_reader = self.reader.take().unwrap();
            self.runtime_stats.merge(&drained_reader.runtime_stats());
        }
    }

    pub fn ffi_is_late_materialization_supported(
        &mut self,
        early_column_ids: &[i64],
    ) -> Result<bool> {
        // The pending LM batch is owned by one sequential reader. Keep the
        // protocol disabled for both worker aggregation and sequential
        // multi-table iteration; the legacy path can handle both safely.
        if self.need_concurrency() {
            info!(
                "{} late materialization unavailable: reason=reader_concurrency table_count={} concurrency={}",
                self.snap_access.core.tag,
                self.lm_table_count(),
                self.concurrency
            );
            return Ok(false);
        }
        if self.lm_table_count() > 1 {
            info!(
                "{} late materialization unavailable: reason=multiple_tables table_count={}",
                self.snap_access.core.tag,
                self.lm_table_count()
            );
            return Ok(false);
        }
        if self.reader.is_none() {
            let Some(table_ctx) = self.tables.first().cloned() else {
                info!(
                    "{} late materialization unavailable: reason=no_tables table_count=0",
                    self.snap_access.core.tag
                );
                return Ok(false);
            };
            self.tables.remove(0);
            self.reader = Some(self.make_sequential_reader(table_ctx)?);
        }
        self.reader
            .as_mut()
            .expect("sequential reader is initialized")
            .ffi_is_late_materialization_supported(early_column_ids)
    }

    pub fn ffi_read_early_column(&mut self, batch_id: u64, column_id: i64) -> Result<Vec<u8>> {
        self.lm_reader_mut()?
            .ffi_read_early_column(batch_id, column_id)
    }

    pub fn ffi_materialize_selected(
        &mut self,
        batch_id: u64,
        selection: LateMaterializationSelection<'_>,
    ) -> Result<usize> {
        self.lm_reader_mut()?
            .ffi_materialize_selected(batch_id, selection)
    }

    pub fn ffi_read_late_column(&mut self, batch_id: u64, column_id: i64) -> Result<Vec<u8>> {
        self.lm_reader_mut()?
            .ffi_read_late_column(batch_id, column_id)
    }

    pub fn ffi_finish_materialized_block(&mut self, batch_id: u64) -> Result<()> {
        self.lm_reader_mut()?
            .ffi_finish_materialized_block(batch_id)
    }

    pub fn ffi_discard_late_materialization_batch(&mut self, batch_id: u64) -> Result<()> {
        self.lm_reader_mut()?
            .ffi_discard_late_materialization_batch(batch_id)
    }

    pub fn ffi_read_handle(&mut self) -> Vec<u8> {
        if !self.need_concurrency() {
            if let Some(reader) = self.reader.as_mut() {
                return reader.ffi_read_handle();
            }
            return vec![];
        }
        self.current_block
            .as_mut()
            .map(|block| std::mem::take(&mut block.handle_data))
            .unwrap_or_default()
    }

    pub fn ffi_read_version(&mut self) -> Vec<u8> {
        if !self.need_concurrency() {
            if let Some(reader) = self.reader.as_mut() {
                return reader.ffi_read_version();
            }
            return vec![];
        }
        self.current_block
            .as_mut()
            .map(|block| std::mem::take(&mut block.version_data))
            .unwrap_or_default()
    }

    pub fn ffi_read_column(&mut self, col_id: i64) -> Vec<u8> {
        if !self.need_concurrency() {
            if let Some(reader) = self.reader.as_mut() {
                return reader.ffi_read_column(col_id);
            }
            return vec![];
        }
        self.current_block
            .as_mut()
            .map(|block| block.take_column_data(col_id))
            .unwrap_or_default()
    }

    pub fn ffi_physical_table_id(&self) -> i64 {
        if !self.need_concurrency() {
            return self
                .reader
                .as_ref()
                .map(|reader| reader.physical_table_id())
                .unwrap_or(-1);
        }
        self.current_block
            .as_ref()
            .map(|block| block.table_id)
            .unwrap_or(-1)
    }

    fn start_workers(&mut self, read_limit: usize) -> Result<()> {
        debug_assert!(self.concurrency > 0);

        let (tx, rx) = mpsc::bounded(self.concurrency * 2);
        self.read_limit = Some(read_limit);
        let shared_stats = Arc::new(Mutex::new(ColumnarRuntimeStats::default()));
        self.concurrent_runtime_stats = Some(Arc::clone(&shared_stats));

        let tables = std::mem::take(&mut self.tables);
        if tables.is_empty() {
            self.finished = true;
            self.result_rx = Some(rx);
            let _ = tx.send(ReadResult::Finished);
            return Ok(());
        }

        let table_count = tables.len();
        let queue = Arc::new(Mutex::new(VecDeque::from(tables)));
        let worker_count = std::cmp::min(self.concurrency, table_count);
        let active_workers = Arc::new(AtomicUsize::new(worker_count));
        let ann_query = self.ann_query.clone();
        let fts_query_info = self.fts_query_info.clone();
        let columns = Arc::clone(&self.columns);
        let start_ts = self.start_ts;

        for _ in 0..worker_count {
            let tx_clone = tx.clone();
            let queue_clone = Arc::clone(&queue);
            let ann_query = ann_query.clone();
            let fts_query_info = fts_query_info.clone();
            let columns = Arc::clone(&columns);
            let snap_access = self.snap_access.clone();
            let ia_ctx = self.ia_ctx.clone();
            let scan_ctx = self.scan_ctx.clone();
            let active_workers = Arc::clone(&active_workers);
            let runtime = self.runtime.clone();
            let shared_stats = Arc::clone(&shared_stats);

            self.runtime.spawn_blocking(move || {
                let send_finished = Self::worker_loop(
                    &runtime,
                    &queue_clone,
                    &tx_clone,
                    &snap_access,
                    &ia_ctx,
                    &columns,
                    &scan_ctx,
                    &ann_query,
                    &fts_query_info,
                    start_ts,
                    read_limit,
                    &shared_stats,
                );

                let prev = active_workers.fetch_sub(1, Ordering::AcqRel);
                if send_finished && prev == 1 {
                    let _ = tx_clone.send(ReadResult::Finished);
                }
            });
        }
        drop(tx);
        self.result_rx = Some(rx);
        Ok(())
    }

    fn worker_loop(
        runtime: &Handle,
        queue: &Arc<Mutex<VecDeque<TableCtx>>>,
        tx: &mpsc::Sender<ReadResult>,
        snap_access: &SnapAccess,
        ia_ctx: &IaCtx,
        columns: &Arc<Vec<ColumnInfo>>,
        scan_ctx: &TableScanCtx,
        ann_query: &Option<Arc<tipb::AnnQueryInfo>>,
        fts_query_info: &Option<Arc<tipb::FtsQueryInfo>>,
        start_ts: u64,
        read_limit: usize,
        shared_stats: &Arc<Mutex<ColumnarRuntimeStats>>,
    ) -> bool {
        while let Some(table_ctx) = Self::pop_table(queue) {
            if !Self::process_table(
                runtime,
                table_ctx,
                read_limit,
                tx,
                snap_access,
                ia_ctx,
                columns,
                scan_ctx,
                ann_query,
                fts_query_info,
                start_ts,
                shared_stats,
            ) {
                return false;
            }
        }
        true
    }

    fn pop_table(queue: &Arc<Mutex<VecDeque<TableCtx>>>) -> Option<TableCtx> {
        let mut guard = queue.lock().unwrap();
        guard.pop_front()
    }

    fn process_table(
        runtime: &Handle,
        table_ctx: TableCtx,
        read_limit: usize,
        tx: &mpsc::Sender<ReadResult>,
        snap_access: &SnapAccess,
        ia_ctx: &IaCtx,
        columns: &Arc<Vec<ColumnInfo>>,
        scan_ctx: &TableScanCtx,
        ann_query: &Option<Arc<tipb::AnnQueryInfo>>,
        fts_query_info: &Option<Arc<tipb::FtsQueryInfo>>,
        start_ts: u64,
        shared_stats: &Arc<Mutex<ColumnarRuntimeStats>>,
    ) -> bool {
        let mut reader = match CloudColumnarReader::new(
            runtime.clone(),
            snap_access.clone(),
            ia_ctx.clone(),
            table_ctx.table_id,
            table_ctx.ranges.clone(),
            columns,
            scan_ctx,
            ann_query.clone(),
            fts_query_info.clone(),
            start_ts,
        ) {
            Ok(reader) => reader,
            Err(err) => {
                Self::log_error_with_ctx(
                    snap_access.get_tag(),
                    &table_ctx,
                    scan_ctx,
                    columns,
                    ann_query,
                    fts_query_info,
                    start_ts,
                    &err,
                );
                let err = crate::Error::from(crate::table::Error::Other(err.to_string()));
                let _ = tx.send(ReadResult::Error(err));
                return false;
            }
        };

        let keep_running = Self::stream_table_blocks(&mut reader, read_limit, tx);
        let reader_stats = reader.runtime_stats();
        shared_stats.lock().unwrap().merge(&reader_stats);
        keep_running
    }

    fn stream_table_blocks(
        reader: &mut CloudColumnarReader,
        read_limit: usize,
        tx: &mpsc::Sender<ReadResult>,
    ) -> bool {
        loop {
            match reader.ffi_read_block(read_limit) {
                Ok(0) => return true,
                Ok(read_count) => {
                    let block = BlockResult::new(read_count, reader);
                    if tx.send(ReadResult::Block(block)).is_err() {
                        return false;
                    }
                }
                Err(e) => {
                    let _ = tx.send(ReadResult::Error(e));
                    return false;
                }
            }
        }
    }

    fn sequential_read_block(&mut self, read_limit: usize) -> Result<usize> {
        if read_limit == 0 {
            return Ok(0);
        }

        if self.reader.is_none() {
            if self.tables.is_empty() {
                return Ok(0);
            }
            let table_ctx = self.tables.remove(0);
            self.reader = Some(self.make_sequential_reader(table_ctx)?);
        }

        loop {
            let read_count = self.reader.as_mut().unwrap().ffi_read_block(read_limit)?;
            if read_count == 0 {
                if let Some(drained_reader) = self.reader.take() {
                    let reader_stats = drained_reader.runtime_stats();
                    self.runtime_stats.merge(&reader_stats);
                }
                if self.tables.is_empty() {
                    return Ok(0);
                }
                let table_ctx = self.tables.remove(0);
                self.reader = Some(self.make_sequential_reader(table_ctx)?);
                continue;
            }
            return Ok(read_count);
        }
    }

    fn lm_reader_mut(&mut self) -> Result<&mut CloudColumnarReader> {
        self.ensure_lm_single_table()?;
        self.reader.as_mut().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })
    }

    fn make_sequential_reader(&self, table_ctx: TableCtx) -> Result<CloudColumnarReader> {
        CloudColumnarReader::new(
            self.runtime.clone(),
            self.snap_access.clone(),
            self.ia_ctx.clone(),
            table_ctx.table_id,
            table_ctx.ranges.clone(),
            &self.columns,
            &self.scan_ctx,
            self.ann_query.clone(),
            self.fts_query_info.clone(),
            self.start_ts,
        )
        .map_err(|err| {
            Self::log_error_with_ctx(
                self.snap_access.get_tag(),
                &table_ctx,
                &self.scan_ctx,
                &self.columns,
                &self.ann_query,
                &self.fts_query_info,
                self.start_ts,
                &err,
            );
            crate::Error::from(crate::table::Error::Other(err.to_string()))
        })
    }
}

struct BlockResult {
    table_id: i64,
    read_count: usize,
    schema: Schema,
    handle_column_id: i64,
    version_column_id: i64,
    handle_data: Vec<u8>,
    version_data: Vec<u8>,
    columns: HashMap<i64, Vec<u8>>,
}

impl BlockResult {
    fn new(read_count: usize, reader: &mut CloudColumnarReader) -> Self {
        let schema = reader.schema.clone();
        // The concurrent cloud-columnar path keeps these buffers in the block
        // result for existing plumbing. When pack clean read is selected they
        // may contain internal placeholders, which is safe because that path is
        // only enabled when the projection excludes handle/pk_handle/version.
        let handle_data = reader.ffi_read_handle();
        let version_data = reader.ffi_read_version();
        let mut columns = HashMap::with_capacity(schema.columns.len());
        for col_info in &schema.columns {
            let col_id = col_info.get_column_id();
            if col_info.get_pk_handle() {
                continue;
            }
            columns.insert(col_id, reader.ffi_read_column(col_id));
        }
        let handle_column_id = schema.handle_column.get_column_id();
        let version_column_id = schema.version_column.get_column_id();
        let table_id = schema.table_id;
        Self {
            table_id,
            read_count,
            schema,
            handle_column_id,
            version_column_id,
            handle_data,
            version_data,
            columns,
        }
    }

    // Take the column data and remove it from the block result to avoid cloning.
    // NOTE: This method can only be called once for each column.
    fn take_column_data(&mut self, col_id: i64) -> Vec<u8> {
        if col_id == self.handle_column_id {
            return std::mem::take(&mut self.handle_data);
        }
        if col_id == self.version_column_id {
            return std::mem::take(&mut self.version_data);
        }
        if let Some(col_info) = self.schema.find_column_by_id(col_id) {
            if col_info.get_pk_handle() {
                return std::mem::take(&mut self.handle_data);
            }
        }
        self.columns.remove(&col_id).unwrap_or_default()
    }
}

enum ReadResult {
    Block(BlockResult),
    Error(crate::Error),
    Finished,
}

pub struct CloudColumnarReader {
    runtime: Handle,
    tag: String,
    reader: Box<dyn ColumnarFilterReader>,
    ranges: Vec<tipb::KeyRange>,
    hybrid_ranges: Vec<tipb::KeyRange>,
    block: Block,
    schema: Schema,
    keyspace_id: u32,
    ia_ctx: IaCtx,
    inited: bool,
    runtime_stats: ColumnarRuntimeStats,
    next_lm_batch_id: u64,
    pending_lm_batch: Option<PendingLateMaterializationBatch>,
    hybrid_reader: Option<HybridMvccReader>,
    lm_snap_access: SnapAccess,
    lm_columns: Vec<ColumnInfo>,
    lm_scan_ctx: TableScanCtx,
    lm_start_ts: u64,
    lm_eligible: bool,
    /// Early projection used to construct `hybrid_reader`. Set when the hybrid
    /// graph is first built so later LM calls can reject a mismatched
    /// projection.
    hybrid_early_column_ids: Option<HashSet<i64>>,
    /// True after the hybrid reader has seeked the first key range and segment
    /// stats have been recorded. Distinct from `hybrid_reader.is_some()`:
    /// capability checks may construct the reader without initializing the
    /// read cursor.
    hybrid_inited: bool,
}

/// The exact selection is owned by TiFlash. CSE only validates and applies it
/// after merge and MVCC have produced the pending row space.
pub enum LateMaterializationSelection<'a> {
    All,
    None,
    Bytes(&'a [u8]),
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum LateMaterializationBatchState {
    PendingEarly,
    Materialized,
}

struct PendingLateMaterializationBatch {
    id: u64,
    row_count: usize,
    early_column_ids: HashSet<i64>,
    late_column_ids: HashSet<i64>,
    late_columns_taken: HashSet<i64>,
    state: LateMaterializationBatchState,
    hybrid: Option<HybridBatch>,
    selection: Option<Vec<u8>>,
    materialization_plan: Option<HybridMaterializationPlan>,
    l2_files: Vec<Vec<ColumnarFile>>,
    late_column_positions: HashMap<i64, usize>,
    materialized_late_columns: HashMap<i64, Vec<u8>>,
}

const LATE_COLUMN_READ_CONCURRENCY: usize = 4;

impl CloudColumnarReader {
    pub fn new(
        runtime: Handle,
        snap_access: SnapAccess,
        ia_ctx: IaCtx,
        table_id: i64,
        ranges: Vec<tipb::KeyRange>,
        columns: &[ColumnInfo],
        scan_ctx: &TableScanCtx,
        ann_query: Option<Arc<tipb::AnnQueryInfo>>,
        fts_query_info: Option<Arc<tipb::FtsQueryInfo>>,
        start_ts: u64,
    ) -> Result<Self> {
        let lm_eligible = ann_query.is_none() && fts_query_info.is_none();
        let reader: Box<dyn ColumnarFilterReader> = if let Some(query_info) = fts_query_info {
            let schema = snap_access
                .new_schema_from_columns(table_id, columns)
                .ok_or(crate::Error::Other(
                    "construct schema from columns failed".to_string().into(),
                ))?;

            // Optional bounds for pruning the FTS index search.
            //
            // NOTE: These are extracted from the *row key* bytes (memcomparable int-handle
            // or raw common-handle), not the columnar in-block handle format.
            let pk_off = KEYSPACE_PREFIX_LEN + tidb_query_datatype::codec::table::PREFIX_LEN;
            let fts_start_handle = ranges
                .iter()
                .try_fold(None::<&[u8]>, |best, range| -> std::result::Result<_, ()> {
                    let pk = range.get_low().get(pk_off..).ok_or(())?;
                    if pk.is_empty() {
                        return Err(());
                    }
                    Ok(Some(match best {
                        Some(best) if best <= pk => best,
                        _ => pk,
                    }))
                })
                .ok()
                .flatten()
                .map(Bytes::copy_from_slice);
            let fts_end_handle = ranges
                .iter()
                .try_fold(None::<&[u8]>, |best, range| -> std::result::Result<_, ()> {
                    let pk = range.get_high().get(pk_off..).ok_or(())?;
                    if pk.is_empty() {
                        return Err(());
                    }
                    Ok(Some(match best {
                        Some(best) if best >= pk => best,
                        _ => pk,
                    }))
                })
                .ok()
                .flatten()
                .map(Bytes::copy_from_slice);

            snap_access.new_fts_reader(
                &runtime,
                table_id,
                query_info.as_ref(),
                schema,
                start_ts,
                fts_start_handle,
                fts_end_handle,
            )?
        } else if let Some(ann_query) = ann_query {
            let index_id = ann_query.get_index_id();
            let target = ann_query
                .get_ref_vec_f32()
                .read_vector_float32()
                .ok()
                .ok_or(crate::Error::Other(
                    "read vector float32 failed".to_string().into(),
                ))?;
            let col_id = get_ann_vec_col_id(&ann_query)?;
            let top_k = ann_query.get_top_k();
            let schema = snap_access
                .new_schema_from_columns(table_id, columns)
                .ok_or(crate::Error::Other(
                    "construct schema from columns failed".to_string().into(),
                ))?;

            let vector_reader_result = snap_access.new_vector_index_reader(
                table_id,
                index_id,
                col_id,
                target.as_ref().data(),
                top_k as usize,
                schema,
                start_ts,
                None, // Pass None to disable filter by handle range.
                None,
                Arc::clone(&ann_query),
            )?;
            if let Some(reader) = vector_reader_result {
                Box::new(reader)
            } else {
                warn!("no vector index reader available, use columnar reader instead");
                snap_access
                    .new_columnar_mvcc_reader(
                        table_id,
                        columns,
                        Some(scan_ctx),
                        start_ts,
                        Some(Arc::clone(&ann_query)),
                    )?
                    .ok_or(crate::Error::Other(
                        "no vector index reader available after fallback"
                            .to_string()
                            .into(),
                    ))?
            }
        } else {
            snap_access
                .new_columnar_mvcc_reader(table_id, columns, Some(scan_ctx), start_ts, None)?
                .ok_or(crate::Error::Other(
                    "no columnar available".to_string().into(),
                ))?
        };
        let schema_file = snap_access.get_schema_file().unwrap();
        let table_schema = schema_file.get_table(table_id).unwrap();
        let lm_columns = columns.to_vec();
        let mut columns = columns.to_vec();
        columns.retain(|c| !c.get_pk_handle() && c.get_column_id() != HANDLE_COL_ID as i64);
        let schema_buf = SchemaBuf::new(
            table_id,
            table_schema.handle_column.clone(),
            table_schema.version_column.clone(),
            columns,
            table_schema.pk_col_ids.clone(),
            table_schema.max_col_id,
            table_schema.vector_indexes.clone(),
            table_schema.fulltext_indexes.clone(),
            table_schema.get_storage_class_spec().clone(),
            table_schema.partitions.clone(),
        );
        let schema = Schema::new(schema_buf);
        let block = Block::new(&schema);

        let reader = Self {
            runtime,
            tag: snap_access.get_tag().to_string(),
            reader,
            ranges,
            hybrid_ranges: vec![],
            block,
            schema: schema.clone(),
            keyspace_id: snap_access.get_keyspace_id(),
            runtime_stats: ColumnarRuntimeStats::default(),
            ia_ctx,
            inited: false,
            next_lm_batch_id: 1,
            pending_lm_batch: None,
            hybrid_reader: None,
            lm_snap_access: snap_access.clone(),
            lm_columns,
            lm_scan_ctx: scan_ctx.clone(),
            lm_start_ts: start_ts,
            lm_eligible,
            hybrid_early_column_ids: None,
            hybrid_inited: false,
        };
        // Do not call init here to avoid blocking the dispatch thread.
        Ok(reader)
    }

    fn init(&mut self) -> Result<()> {
        let init_start = Instant::now_coarse();
        let (remote_segments, total_segments) = self.reader.get_remote_segment_stats()?;
        self.runtime_stats.remote_segments = self
            .runtime_stats
            .remote_segments
            .saturating_add(remote_segments as u64);
        self.runtime_stats.total_segments = self
            .runtime_stats
            .total_segments
            .saturating_add(total_segments as u64);
        if let IaCtx::Enabled(ia_mgr, _) = &self.ia_ctx {
            let ia_mgr = ia_mgr.clone();
            let start = Instant::now_coarse();
            let prefetch = self
                .runtime
                .block_on(self.reader.prefetch_ia_remote_segments(
                    &self.tag,
                    &ia_mgr,
                    self.keyspace_id,
                    Duration::from_secs(600), // 10 minutes is enough for prefetching one region.
                ))
                .map_err(|e| {
                    crate::Error::Other(
                        format!("prefetch_ia_remote_segments failed: {:?}", e).into(),
                    )
                })?;
            let prefetch_elapsed = start.saturating_elapsed();
            self.runtime_stats.prefetch_ns = self
                .runtime_stats
                .prefetch_ns
                .saturating_add(prefetch_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
            COLUMNAR_PREFETCH_HISTOGRAM.observe(start.saturating_elapsed_secs());
            COLUMNAR_PREFETCH_CACHE_HIT_HISTOGRAM.observe(prefetch.unwrap_or(1.0) * 100.0);
        }
        let rt = self.runtime.clone();
        rt.block_on(self.update_range_handle())?;
        self.inited = true;
        let init_elapsed = init_start.saturating_elapsed();
        self.runtime_stats.init_reader_ns = self
            .runtime_stats
            .init_reader_ns
            .saturating_add(init_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
        Ok(())
    }

    /// Side-effect-free LM eligibility check for capability probes. Does not
    /// construct or pin a hybrid reader; the first [`ensure_hybrid_reader`] on
    /// the read path owns graph construction and early projection pinning.
    fn check_hybrid_lm_eligible(&self, early_column_ids: &HashSet<i64>) -> Result<bool> {
        if !self.lm_eligible {
            return Ok(false);
        }
        if self.inited {
            return Ok(false);
        }
        Ok(self
            .lm_snap_access
            .new_hybrid_columnar_mvcc_reader(
                self.schema.table_id,
                &self.lm_columns,
                early_column_ids,
                Some(&self.lm_scan_ctx),
                self.lm_start_ts,
            )?
            .is_some())
    }

    /// Builds and pins the hybrid LM reader graph for the read path. Does not
    /// seek key ranges or record segment stats; call [`init_hybrid_for_read`]
    /// after this returns true. Must not be used for capability probes.
    fn ensure_hybrid_reader(&mut self, early_column_ids: &HashSet<i64>) -> Result<bool> {
        if !self.lm_eligible {
            return Ok(false);
        }
        if self.inited {
            return Err(crate::Error::Other(
                "cannot switch an initialized legacy reader to late materialization"
                    .to_string()
                    .into(),
            ));
        }
        if let Some(existing) = self.hybrid_early_column_ids.as_ref() {
            if existing != early_column_ids {
                return Err(crate::Error::Other(
                    "late-materialization early projection changed after hybrid reader construction"
                        .to_string()
                        .into(),
                ));
            }
            return Ok(self.hybrid_reader.is_some());
        }
        self.hybrid_reader = self.lm_snap_access.new_hybrid_columnar_mvcc_reader(
            self.schema.table_id,
            &self.lm_columns,
            early_column_ids,
            Some(&self.lm_scan_ctx),
            self.lm_start_ts,
        )?;
        if self.hybrid_reader.is_none() {
            info!(
                "{} late materialization unavailable: reason=hybrid_reader_unavailable table_id={}",
                self.tag, self.schema.table_id
            );
            return Ok(false);
        }
        self.hybrid_early_column_ids = Some(early_column_ids.clone());
        Ok(true)
    }

    /// Seeks the first hybrid key range and records segment stats. Call only
    /// from the LM read path after [`ensure_hybrid_reader`] returned true.
    fn init_hybrid_for_read(&mut self) -> Result<bool> {
        if self.hybrid_inited {
            return Ok(true);
        }
        if self.hybrid_reader.is_none() {
            return Ok(false);
        }
        if self.inited {
            return Err(crate::Error::Other(
                "cannot switch an initialized legacy reader to late materialization"
                    .to_string()
                    .into(),
            ));
        }
        if self.hybrid_ranges.is_empty() {
            self.hybrid_ranges = self.ranges.clone();
        }
        if self.hybrid_ranges.is_empty() {
            info!(
                "{} late materialization unavailable: reason=no_hybrid_ranges table_id={}",
                self.tag, self.schema.table_id
            );
            return Ok(false);
        }
        let (remote_segments, total_segments) = self
            .hybrid_reader
            .as_ref()
            .expect("hybrid reader is initialized")
            .remote_segments()?;
        self.runtime_stats.remote_segments = self
            .runtime_stats
            .remote_segments
            .saturating_add(remote_segments as u64);
        self.runtime_stats.total_segments = self
            .runtime_stats
            .total_segments
            .saturating_add(total_segments as u64);
        self.runtime
            .clone()
            .block_on(self.update_hybrid_range_handle())?;
        self.hybrid_inited = true;
        Ok(true)
    }

    async fn update_hybrid_range_handle(&mut self) -> Result<bool> {
        if self.hybrid_ranges.is_empty() {
            return Ok(false);
        }
        let range = self.hybrid_ranges.remove(0);
        let reader = self
            .hybrid_reader
            .as_mut()
            .expect("hybrid reader is initialized");
        if reader.schema().is_common_handle() {
            let start =
                decode_common_handle(&range.get_low()[KEYSPACE_PREFIX_LEN..]).unwrap_or(&[]);
            let end = decode_common_handle(&range.get_high()[KEYSPACE_PREFIX_LEN..])
                .unwrap_or(GLOBAL_COMMON_HANDLE_END);
            reader.set_handle_range(start, end).await?;
        } else {
            let start =
                decode_int_handle(&range.get_low()[KEYSPACE_PREFIX_LEN..]).unwrap_or(i64::MIN);
            let end = decode_exclusive_end_int_handle(&range.get_high()[KEYSPACE_PREFIX_LEN..]);
            reader.set_int_handle_range(start, end).await?;
        }
        Ok(true)
    }

    async fn read_hybrid_batch_with_range(
        &mut self,
        batch: &mut HybridBatch,
        read_limit: usize,
    ) -> Result<usize> {
        loop {
            let reader = self
                .hybrid_reader
                .as_mut()
                .expect("hybrid reader is initialized");
            let rows = reader.read(batch, read_limit).await?;
            if rows > 0 {
                return Ok(rows);
            }
            reader.reset()?;
            if !self.update_hybrid_range_handle().await? {
                return Ok(0);
            }
        }
    }

    // Set next range handle.
    // Return Ok(true) if next range handle is set, Ok(false) if no more range.
    async fn update_range_handle(&mut self) -> Result<bool> {
        if self.ranges.is_empty() {
            return Ok(false);
        }
        let range = self.ranges.remove(0);
        info!(
            "{} update range handle, table_id: {}, range: {:?}",
            self.tag, self.schema.table_id, range
        );
        let is_common_handle = self.reader.get_schema().is_common_handle();
        if is_common_handle {
            let start_handle =
                decode_common_handle(&range.get_low()[KEYSPACE_PREFIX_LEN..]).unwrap_or(&[]);
            let end_handle = decode_common_handle(&range.get_high()[KEYSPACE_PREFIX_LEN..])
                .unwrap_or(GLOBAL_COMMON_HANDLE_END);
            self.reader
                .set_handle_range(start_handle, end_handle)
                .await?;
        } else {
            let start_handle =
                decode_int_handle(&range.get_low()[KEYSPACE_PREFIX_LEN..]).unwrap_or(i64::MIN);
            let end_handle =
                decode_exclusive_end_int_handle(&range.get_high()[KEYSPACE_PREFIX_LEN..]);
            self.reader
                .set_int_handle_range(start_handle, end_handle)
                .await?;
        }
        Ok(true)
    }

    pub fn ffi_read_block(&mut self, read_limit: usize) -> Result<usize> {
        if self.pending_lm_batch.is_some() {
            return Err(crate::Error::Other(
                "cannot read a legacy block while a late-materialization batch is pending"
                    .to_string()
                    .into(),
            ));
        }
        if !self.inited {
            // Postpone the initialization to here to use concurrent threads.
            self.init()
                .map_err(|e| crate::table::Error::Other(e.to_string()))?;
        }
        self.block.reset();
        let start = Instant::now_coarse();
        let rt = self.runtime.clone();
        let ret = rt.block_on(self.read_block_with_range(read_limit));
        let read_elapsed = start.saturating_elapsed();
        self.runtime_stats.read_block_ns = self
            .runtime_stats
            .read_block_ns
            .saturating_add(read_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
        ret
    }

    /// Reads one merge/MVCC-visible batch for the late-materialization ABI.
    /// Eligible readers construct an L2-only hybrid batch; ineligible readers
    /// retain the legacy full block as a compatibility fallback.
    pub fn ffi_read_early_block(
        &mut self,
        read_limit: usize,
        early_column_ids: &[i64],
    ) -> Result<Option<(u64, usize, i64)>> {
        if self.pending_lm_batch.is_some() {
            return Err(crate::Error::Other(
                "cannot read an early block while a late-materialization batch is pending"
                    .to_string()
                    .into(),
            ));
        }

        let mut early_column_ids = early_column_ids.iter().copied().collect::<HashSet<_>>();
        if early_column_ids.remove(&(HANDLE_COL_ID as i64)) {
            early_column_ids.insert(self.schema.handle_column.get_column_id());
        }
        self.validate_lm_projection(&early_column_ids)?;
        if self.ensure_hybrid_reader(&early_column_ids)? && self.init_hybrid_for_read()? {
            let hybrid = self
                .hybrid_reader
                .as_ref()
                .expect("hybrid reader is initialized");
            let late_columns = hybrid.late_columns().to_vec();
            let l2_files = hybrid.l2_files().to_vec();
            let mut batch = HybridBatch::new(
                Block::new(hybrid.schema()),
                late_columns
                    .iter()
                    .map(crate::table::columnar::ColumnBuffer::new_from_col_info)
                    .collect(),
            );
            let start = Instant::now_coarse();
            let row_count = self
                .runtime
                .clone()
                .block_on(self.read_hybrid_batch_with_range(&mut batch, read_limit))?;
            self.runtime_stats.read_block_ns = self.runtime_stats.read_block_ns.saturating_add(
                start
                    .saturating_elapsed()
                    .as_nanos()
                    .min(u128::from(u64::MAX)) as u64,
            );
            if row_count == 0 {
                return Ok(None);
            }
            let batch_id = self.next_lm_batch_id;
            self.next_lm_batch_id = self.next_lm_batch_id.checked_add(1).ok_or_else(|| {
                crate::Error::Other("late-materialization batch ID overflow".to_string().into())
            })?;
            let late_column_ids = late_columns
                .iter()
                .map(|column| column.get_column_id())
                .collect::<HashSet<_>>();
            let late_column_positions = late_columns
                .iter()
                .enumerate()
                .map(|(idx, column)| (column.get_column_id(), idx))
                .collect();
            self.pending_lm_batch = Some(PendingLateMaterializationBatch {
                id: batch_id,
                row_count,
                early_column_ids,
                late_column_ids,
                late_columns_taken: HashSet::default(),
                state: LateMaterializationBatchState::PendingEarly,
                hybrid: Some(batch),
                selection: None,
                materialization_plan: None,
                l2_files,
                late_column_positions,
                materialized_late_columns: HashMap::new(),
            });
            return Ok(Some((batch_id, row_count, self.physical_table_id())));
        }
        let row_count = self.ffi_read_block(read_limit)?;
        if row_count == 0 {
            return Ok(None);
        }

        let late_column_ids = self
            .schema
            .columns
            .iter()
            .filter_map(|column| {
                let column_id = column.get_column_id();
                (!column.get_pk_handle() && !early_column_ids.contains(&column_id))
                    .then_some(column_id)
            })
            .collect();
        let batch_id = self.next_lm_batch_id;
        self.next_lm_batch_id = self.next_lm_batch_id.checked_add(1).ok_or_else(|| {
            crate::Error::Other("late-materialization batch ID overflow".to_string().into())
        })?;
        self.pending_lm_batch = Some(PendingLateMaterializationBatch {
            id: batch_id,
            row_count,
            early_column_ids,
            late_column_ids,
            late_columns_taken: HashSet::default(),
            state: LateMaterializationBatchState::PendingEarly,
            hybrid: None,
            selection: None,
            materialization_plan: None,
            l2_files: vec![],
            late_column_positions: HashMap::default(),
            materialized_late_columns: HashMap::new(),
        });
        Ok(Some((batch_id, row_count, self.physical_table_id())))
    }

    /// Reports whether this reader can use L2 late materialization for the
    /// given early projection. This is a side-effect-free capability probe: it
    /// must not seek key ranges, record segment stats, or pin an early
    /// projection. The first [`ffi_read_early_block`] owns hybrid construction.
    pub fn ffi_is_late_materialization_supported(
        &mut self,
        early_column_ids: &[i64],
    ) -> Result<bool> {
        if !self.lm_eligible {
            info!(
                "{} late materialization unavailable: reason=ann_or_fts_query table_id={}",
                self.tag, self.schema.table_id
            );
            return Ok(false);
        }
        let mut early_column_ids = early_column_ids.iter().copied().collect::<HashSet<_>>();
        if early_column_ids.remove(&(HANDLE_COL_ID as i64)) {
            early_column_ids.insert(self.schema.handle_column.get_column_id());
        }
        self.validate_lm_projection(&early_column_ids)?;
        self.check_hybrid_lm_eligible(&early_column_ids)
    }

    #[cfg(test)]
    pub(crate) fn hybrid_inited_for_test(&self) -> bool {
        self.hybrid_inited
    }

    #[cfg(test)]
    pub(crate) fn hybrid_reader_is_some_for_test(&self) -> bool {
        self.hybrid_reader.is_some()
    }

    #[cfg(test)]
    pub(crate) fn hybrid_early_column_ids_is_some_for_test(&self) -> bool {
        self.hybrid_early_column_ids.is_some()
    }

    #[cfg(test)]
    pub(crate) fn hybrid_ranges_len_for_test(&self) -> usize {
        self.hybrid_ranges.len()
    }

    #[cfg(test)]
    pub(crate) fn ranges_len_for_test(&self) -> usize {
        self.ranges.len()
    }

    pub fn ffi_read_early_column(&mut self, batch_id: u64, column_id: i64) -> Result<Vec<u8>> {
        // TiFlash identifies an integer handle as HANDLE_COL_ID, while CSE's
        // schema keeps the actual primary-key column ID in the pending projection.
        let column_id = if column_id == HANDLE_COL_ID as i64 {
            self.schema.handle_column.get_column_id()
        } else {
            column_id
        };
        let pending = self.pending_lm_batch.as_ref().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })?;
        if pending.id != batch_id || pending.state != LateMaterializationBatchState::PendingEarly {
            return Err(crate::Error::Other(
                "invalid early-column batch state".to_string().into(),
            ));
        }
        if !pending.early_column_ids.contains(&column_id) {
            return Err(crate::Error::Other(
                format!("column {} is not in the early projection", column_id).into(),
            ));
        }
        if let Some(hybrid) = pending.hybrid.as_ref() {
            let (column, column_info) = if column_id == self.schema.handle_column.get_column_id() {
                (hybrid.early.get_handle_buf(), &self.schema.handle_column)
            } else if column_id == self.schema.version_column.get_column_id() {
                (hybrid.early.get_version_buf(), &self.schema.version_column)
            } else {
                (
                    hybrid.early.get_column(column_id),
                    self.schema.find_column_by_id(column_id).unwrap(),
                )
            };
            return Ok(Self::serialize_column_for_tiflash(column, column_info));
        }
        if column_id == self.schema.handle_column.get_column_id() {
            Ok(self.ffi_read_handle())
        } else if column_id == self.schema.version_column.get_column_id() {
            Ok(self.ffi_read_version())
        } else {
            Ok(self.ffi_read_column(column_id))
        }
    }

    pub fn ffi_materialize_selected(
        &mut self,
        batch_id: u64,
        selection: LateMaterializationSelection<'_>,
    ) -> Result<usize> {
        let pending = self.pending_lm_batch.as_ref().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })?;
        if pending.id != batch_id || pending.state != LateMaterializationBatchState::PendingEarly {
            return Err(crate::Error::Other(
                "invalid materialization batch state".to_string().into(),
            ));
        }

        let selection = match selection {
            LateMaterializationSelection::All => None,
            LateMaterializationSelection::None => Some(vec![0; pending.row_count]),
            LateMaterializationSelection::Bytes(bytes) => {
                if bytes.len() != pending.row_count {
                    return Err(crate::Error::Other(
                        format!(
                            "late-materialization selection length {} does not match batch row count {}",
                            bytes.len(),
                            pending.row_count
                        )
                        .into(),
                    ));
                }
                Some(bytes.to_vec())
            }
        };

        let selected_rows = selection
            .as_ref()
            .map(|bytes| bytes.iter().filter(|&&value| value != 0).count())
            .unwrap_or(pending.row_count);
        if let Some(hybrid) = pending.hybrid.as_ref() {
            let selection = selection.unwrap_or_else(|| vec![1; pending.row_count]);
            let materialization_plan = hybrid.build_materialization_plan(&selection)?;
            let pending = self.pending_lm_batch.as_mut().unwrap();
            pending.selection = Some(selection);
            pending.materialization_plan = Some(materialization_plan);
        } else if let Some(selection) = selection {
            self.block.retain_rows(|row_idx| selection[row_idx] != 0);
        }
        self.pending_lm_batch.as_mut().unwrap().state = LateMaterializationBatchState::Materialized;
        Ok(selected_rows)
    }

    fn materialize_late_columns_concurrently(&mut self) -> Result<()> {
        let (plan, specs, files, hybrid) = {
            let pending = self.pending_lm_batch.as_ref().ok_or_else(|| {
                crate::Error::Other(
                    "no late-materialization batch is pending"
                        .to_string()
                        .into(),
                )
            })?;
            let Some(hybrid) = pending.hybrid.as_ref() else {
                return Ok(());
            };
            if !pending.materialized_late_columns.is_empty() {
                return Ok(());
            }
            (
                pending.materialization_plan.clone().ok_or_else(|| {
                    crate::Error::Other(
                        "late-materialization materialization plan is missing"
                            .to_string()
                            .into(),
                    )
                })?,
                pending
                    .late_column_positions
                    .iter()
                    .map(|(&id, &idx)| (id, idx))
                    .collect::<Vec<_>>(),
                pending.l2_files.clone(),
                hybrid,
            )
        };
        let schema = self.schema.clone();
        let key = self.lm_snap_access.encryption_key.clone();
        let result = self.runtime.block_on(async {
            stream::iter(specs.into_iter().map(|(id, idx)| {
                let plan = &plan;
                let schema = &schema;
                let files = &files;
                let key = key.clone();
                async move {
                    let (column, late_stats) = hybrid
                        .materialize_selected_late_column(plan, idx, id, schema, files, key)
                        .await?;
                    Ok::<_, crate::Error>((id, column, late_stats))
                }
            }))
            // Limit fan-out so a wide projection does not exhaust the runtime or remote IO slots.
            .buffered(LATE_COLUMN_READ_CONCURRENCY)
            .try_collect::<Vec<_>>()
            .await
        })?;
        let mut output = HashMap::new();
        for (id, column, late_stats) in result {
            self.runtime_stats.lm_late_packs_loaded = self
                .runtime_stats
                .lm_late_packs_loaded
                .saturating_add(late_stats.packs_loaded);
            self.runtime_stats.lm_late_rows_gathered = self
                .runtime_stats
                .lm_late_rows_gathered
                .saturating_add(late_stats.rows_gathered);
            output.insert(
                id,
                Self::serialize_column_for_tiflash(
                    &column,
                    self.schema.find_column_by_id(id).unwrap(),
                ),
            );
        }
        self.pending_lm_batch
            .as_mut()
            .unwrap()
            .materialized_late_columns = output;
        Ok(())
    }

    pub fn ffi_read_late_column(&mut self, batch_id: u64, column_id: i64) -> Result<Vec<u8>> {
        let runtime = self.runtime.clone();
        let schema = self.schema.clone();
        let encryption_key = self.lm_snap_access.encryption_key.clone();
        let pending = self.pending_lm_batch.as_mut().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })?;
        if pending.id != batch_id || pending.state != LateMaterializationBatchState::Materialized {
            return Err(crate::Error::Other(
                "invalid late-column batch state".to_string().into(),
            ));
        }
        if !pending.late_column_ids.contains(&column_id) {
            return Err(crate::Error::Other(
                format!("invalid or duplicate late column {}", column_id).into(),
            ));
        }
        if pending.late_columns_taken.contains(&column_id) {
            return Err(crate::Error::Other(
                format!("invalid or duplicate late column {}", column_id).into(),
            ));
        }
        self.materialize_late_columns_concurrently()?;
        let pending = self.pending_lm_batch.as_mut().unwrap();
        if let Some(data) = pending.materialized_late_columns.remove(&column_id) {
            pending.late_columns_taken.insert(column_id);
            return Ok(data);
        }
        if let Some(hybrid) = pending.hybrid.as_ref() {
            let materialization_plan = pending.materialization_plan.as_ref().ok_or_else(|| {
                crate::Error::Other(
                    "late-materialization materialization plan is missing"
                        .to_string()
                        .into(),
                )
            })?;
            let late_column_idx =
                *pending
                    .late_column_positions
                    .get(&column_id)
                    .ok_or_else(|| {
                        crate::Error::Other(
                            "late-materialization column position is missing"
                                .to_string()
                                .into(),
                        )
                    })?;
            let column_info = schema.find_column_by_id(column_id).unwrap();
            let (column, late_stats) =
                runtime.block_on(hybrid.materialize_selected_late_column(
                    materialization_plan,
                    late_column_idx,
                    column_id,
                    &schema,
                    &pending.l2_files,
                    encryption_key,
                ))?;
            self.runtime_stats.lm_late_packs_loaded = self
                .runtime_stats
                .lm_late_packs_loaded
                .saturating_add(late_stats.packs_loaded);
            self.runtime_stats.lm_late_rows_gathered = self
                .runtime_stats
                .lm_late_rows_gathered
                .saturating_add(late_stats.rows_gathered);
            pending.late_columns_taken.insert(column_id);
            return Ok(Self::serialize_column_for_tiflash(&column, column_info));
        }
        let data = self.ffi_read_column(column_id);
        self.pending_lm_batch
            .as_mut()
            .expect("late-materialization batch is still pending")
            .late_columns_taken
            .insert(column_id);
        Ok(data)
    }

    pub fn ffi_finish_materialized_block(&mut self, batch_id: u64) -> Result<()> {
        let pending = self.pending_lm_batch.as_ref().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })?;
        if pending.id != batch_id || pending.state != LateMaterializationBatchState::Materialized {
            return Err(crate::Error::Other(
                "invalid finish batch state".to_string().into(),
            ));
        }
        if pending.late_columns_taken != pending.late_column_ids {
            return Err(crate::Error::Other(
                "not every late column was consumed".to_string().into(),
            ));
        }
        self.pending_lm_batch = None;
        self.block.reset();
        Ok(())
    }

    pub fn ffi_discard_late_materialization_batch(&mut self, batch_id: u64) -> Result<()> {
        let pending = self.pending_lm_batch.as_ref().ok_or_else(|| {
            crate::Error::Other(
                "no late-materialization batch is pending"
                    .to_string()
                    .into(),
            )
        })?;
        if pending.id != batch_id {
            return Err(crate::Error::Other(
                "invalid discard batch ID".to_string().into(),
            ));
        }
        self.pending_lm_batch = None;
        self.block.reset();
        Ok(())
    }

    fn validate_lm_projection(&self, early_column_ids: &HashSet<i64>) -> Result<()> {
        for column_id in early_column_ids {
            if *column_id != self.schema.handle_column.get_column_id()
                && *column_id != self.schema.version_column.get_column_id()
                && self.schema.find_column_by_id(*column_id).is_none()
            {
                return Err(crate::Error::Other(
                    format!("unknown early projection column {}", column_id).into(),
                ));
            }
        }
        if !early_column_ids.contains(&self.schema.handle_column.get_column_id())
            || !early_column_ids.contains(&self.schema.version_column.get_column_id())
        {
            return Err(crate::Error::Other(
                "the early projection must include handle and version columns"
                    .to_string()
                    .into(),
            ));
        }
        Ok(())
    }

    pub async fn read_block_with_range(&mut self, read_limit: usize) -> Result<usize> {
        loop {
            let read_count = self.reader.read_block(&mut self.block, read_limit).await?;
            if read_count > 0 {
                return Ok(read_count);
            }
            // Data may be read out and filtered out in previous range, we need to reset the
            // reader here to continue reading.
            self.reader.reset();
            if !self.update_range_handle().await? {
                return Ok(0);
            }
        }
    }

    pub fn ffi_read_handle(&mut self) -> Vec<u8> {
        let start = Instant::now_coarse();
        let mut data = vec![];
        let col_info = &self.schema.handle_column;
        self.block.get_handle_buf().serialize_for_tiflash(
            &mut data,
            col_info.get_tp(),
            Self::is_unsigned(col_info),
            col_info.get_column_len() as usize,
        );
        let serialize_elapsed = start.saturating_elapsed();
        self.runtime_stats.serialize_ns = self
            .runtime_stats
            .serialize_ns
            .saturating_add(serialize_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
        data
    }

    pub fn ffi_read_version(&mut self) -> Vec<u8> {
        let start = Instant::now_coarse();
        let mut data = vec![];
        let col_info = &self.schema.version_column;
        self.block.get_version_buf().serialize_for_tiflash(
            &mut data,
            col_info.get_tp(),
            Self::is_unsigned(col_info),
            col_info.get_column_len() as usize,
        );
        let serialize_elapsed = start.saturating_elapsed();
        self.runtime_stats.serialize_ns = self
            .runtime_stats
            .serialize_ns
            .saturating_add(serialize_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
        data
    }

    fn is_unsigned(col_info: &ColumnInfo) -> bool {
        match FieldTypeFlag::from_bits(col_info.get_flag() as u32) {
            Some(flag) => flag.contains(FieldTypeFlag::UNSIGNED),
            None => false,
        }
    }

    fn serialize_column_for_tiflash(
        column: &crate::table::columnar::ColumnBuffer,
        col_info: &ColumnInfo,
    ) -> Vec<u8> {
        let mut data = vec![];
        column.serialize_for_tiflash(
            &mut data,
            col_info.get_tp(),
            Self::is_unsigned(col_info),
            col_info.get_column_len() as usize,
        );
        data
    }

    pub fn ffi_read_column(&mut self, col_id: i64) -> Vec<u8> {
        let start = Instant::now_coarse();
        let mut data = vec![];
        let col_info = self.schema.find_column_by_id(col_id).unwrap();
        if col_info.get_pk_handle() {
            self.block.get_handle_buf().serialize_for_tiflash(
                &mut data,
                col_info.get_tp(),
                Self::is_unsigned(col_info),
                col_info.get_column_len() as usize,
            );
        } else {
            self.block.get_column(col_id).serialize_for_tiflash(
                &mut data,
                col_info.get_tp(),
                Self::is_unsigned(col_info),
                col_info.get_column_len() as usize,
            );
        }
        let serialize_elapsed = start.saturating_elapsed();
        self.runtime_stats.serialize_ns = self
            .runtime_stats
            .serialize_ns
            .saturating_add(serialize_elapsed.as_nanos().min(u128::from(u64::MAX)) as u64);
        data
    }

    pub fn physical_table_id(&self) -> i64 {
        self.schema.table_id
    }

    pub fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.runtime_stats;
        let reader_stats = self.reader.runtime_stats();
        stats.merge(&reader_stats);
        if let Some(reader) = self.hybrid_reader.as_ref() {
            stats.merge(&reader.runtime_stats());
        }
        stats
    }

    #[cfg(any(test, feature = "testexport"))]
    pub fn read_all_for_test(&mut self) -> Result<Block> {
        if !self.inited {
            self.init()?;
        }
        let schema = self.schema.clone();
        let mut out = Block::new(&schema);
        let rt = self.runtime.clone();
        loop {
            self.block.reset();
            let read = rt.block_on(self.read_block_with_range(1024))?;
            if read == 0 {
                break;
            }
            out.append(&self.block, 0, read);
        }
        Ok(out)
    }
}

impl Drop for CloudColumnarReader {
    fn drop(&mut self) {
        let stats = self.runtime_stats();
        info!(
            "{} CloudColumnarReader dropped, table_id: {} read_block_cost: {:.3}s, serialize_cost: {:.3}s",
            self.tag,
            self.schema.table_id,
            stats.read_block_ns as f64 / 1_000_000_000.0,
            stats.serialize_ns as f64 / 1_000_000_000.0
        );
    }
}

// Due to the discarding of column_id in the vector index, the value assigned to
// ann_query_info may be different in different versions of tidb. For
// compatibility, use this function to get column_id.
// For field changes, see: https://github.com/pingcap/tipb/pull/358
fn get_ann_vec_col_id(ann_query: &AnnQueryInfo) -> Result<i64> {
    // Check `has_deprecated_column_id` first, because `column` is marked as
    // not null in TiDB side so that it will be always populated with some value.
    if ann_query.has_deprecated_column_id() {
        Ok(ann_query.get_deprecated_column_id())
    } else if ann_query.has_column() {
        Ok(ann_query.get_column().get_column_id())
    } else {
        Err(crate::Error::Other(
            "unexpected empty vector column id in ANNQueryInfo"
                .to_string()
                .into(),
        ))
    }
}

pub enum TableRef<'a> {
    Memtable(&'a CfTable),
    L0(&'a L0Table),
    SsTable(&'a SsTable),
}

/// To check that the key already exists or not.
struct TxnFileConstraintChecker<'a> {
    already_exist_key: Option<Vec<u8>>,
    item: Item<'a>,
    snap: &'a SnapAccessCore,
    read_ts: u64,
}

impl<'a> TxnFileConstraintChecker<'a> {
    fn new(snap: &'a SnapAccessCore, read_ts: u64) -> Self {
        let mut item = Item::new();
        item.owned_val = Some(vec![]);
        Self {
            already_exist_key: None,
            item,
            snap,
            read_ts,
        }
    }
}

#[maybe_async::async_trait]
impl EntryChecker for TxnFileConstraintChecker<'_> {
    fn should_check(&self, op: u8, _action: u8) -> bool {
        op == OP_INSERT || op == OP_CHECK_NOT_EXIST
    }

    #[maybe_async]
    async fn check(&mut self, _op: u8, _action: u8, key: InnerKey<'_>) -> bool {
        self.item.path = AccessPath::default();
        self.item.val = self
            .snap
            .get_value(
                WRITE_CF,
                key,
                self.read_ts,
                &mut self.item.path,
                self.item.owned_val.as_mut().unwrap(),
                &[],
            )
            .await;
        // Blob value is not needed here.

        if self.item.value_len() > 0 {
            let outer_key = self.snap.data.to_outer_key(key);
            self.already_exist_key = Some(outer_key);
            return false; // Return false to stop iteration.
        }
        true
    }
}

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

    use api_version::{ApiV2, api_v2::KEYSPACE_PREFIX_LEN};
    use bytes::{Buf, Bytes};
    use cloud_encryption::{
        EncryptionKey, EncryptionKeyManager, EncryptionState, MasterKey, MasterKeyConfig,
        ToEncryptionState,
    };
    use futures::executor::block_on;
    use kvenginepb::{TableCreate, TxnFileRefs};
    use proptest::prelude::*;
    use protobuf::Message;
    use schema::schema::{StorageClass, StorageClassSpec, StorageClassTransitRule};
    use tikv_util::memory::MemoryLimiter;

    use crate::{
        ChangeSet, DEL_PREFIXES_KEY, DeletePrefixes, ENCRYPTION_META_KEY, EXTRA_CF,
        GLOBAL_SHARD_END_KEY, IoType, LOCK_CF, NUM_CFS, STORAGE_CLASS_KEY, Shard, ShardRange,
        SnapAccess, TERM_KEY, TXN_FILE_REF, UserMeta, WRITE_CF,
        apply::create_snapshot_tables,
        context::{IaCtx, PrepareType, SnapCtx, new_meta_file_cache},
        dfs::{self, Dfs, FileType, InMemFs},
        read::{MEM_DATA_FORMAT_V2, ONLINE_SNAPSHOT_MEM_DATA_FORMAT_VERSION},
        shard::{
            ShardCf, ShardCfBuilder, ShardDataBuilder, get_shard_property, set_shard_property,
        },
        table::{
            self, ChecksumType, InnerKey, OP_PUT, OwnedInnerKey, PessimisticAction,
            TXN_FILE_FORMAT_V2, TxnChunk, TxnChunkBuilder, TxnCtx, TxnFile, TxnFileId,
            blobtable::BlobRef,
            columnar::{ColumnarMetaCache, new_int_handle_column_info, new_version_column_info},
            file::InMemFile,
            fts::{FtsCache, FtsDeltaCache},
            memtable::{CfTable, MemTableImplementation},
            schema_file::{Schema, SchemaBuf},
            sstable::{
                BlockCache, L0Builder, L0Table, SsTable,
                test_util::{
                    TEST_ID_ALLOC, build_test_table_with_kvs, new_table_builder_for_test,
                    new_test_cache,
                },
            },
            vector_index::VectorIndexes,
        },
        txn_chunk_manager::{TxnChunkManager, TxnChunkManagerConfig, with_pool_size},
        util::test_util::KeyBuilder,
    };

    const KEYSPACE_ID: u32 = 42;
    const TERM: u64 = 11;

    #[test]
    fn test_reconstruct_large_remote_memtable_with_crossbeam() {
        let value = vec![b'x'; 16 * 1024 * 1024 + 1];
        let rows = vec![table::Row {
            key: b"large".to_vec(),
            user_meta: UserMeta::new(1, 2),
            value: value.clone(),
        }];
        let data = bincode::serialize(&rows).unwrap();
        let mut memtable = CfTable::new();

        SnapAccess::construct_skip_list(&data, &mut memtable, WRITE_CF).unwrap();

        assert_eq!(memtable.implementation(), MemTableImplementation::Crossbeam);
        let mut owner = Vec::new();
        let stored =
            memtable
                .get_cf(WRITE_CF)
                .get(InnerKey::from_outer_key(b"large"), u64::MAX, &mut owner);
        assert_eq!(stored.get_value(), value);
    }

    #[test]
    fn test_full_mem_data_v3_preserves_all_cfs_and_tombstones() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let dfs: Arc<dyn Dfs> = Arc::new(InMemFs::new());
        let txn_chunk_manager = TxnChunkManager::new(
            vec![],
            dfs,
            BlockCache::None,
            None,
            with_pool_size(1),
            TxnChunkManagerConfig::default(),
        );
        let keyspace_prefix = ApiV2::get_txn_keyspace_prefix(KEYSPACE_ID);
        let outer_start = [keyspace_prefix.as_slice(), b"a"].concat();
        let outer_end = [keyspace_prefix.as_slice(), b"z"].concat();
        let range = ShardRange::new(&outer_start, &outer_end);
        let master_key = MasterKey::new(&[1u8; 32]);
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            7,
            range,
            KEYSPACE_PREFIX_LEN,
            Arc::new(crate::options::Options::default()),
            &master_key,
            Arc::new(EncryptionKeyManager::new()),
        );
        let delete_prefix = [keyspace_prefix.as_slice(), b"deleted"].concat();
        let outside_delete_prefix = [keyspace_prefix.as_slice(), b"zz-deleted"].concat();
        let deleting_prefixes = DeletePrefixes::new_with_keyspace_id(KEYSPACE_ID)
            .merge_prefix(&delete_prefix)
            .merge_prefix(&outside_delete_prefix);
        shard.set_property(DEL_PREFIXES_KEY, &deleting_prefixes.marshal());
        let sst_max_ts = 50;
        let mut l0_builder =
            L0Builder::new(123, 1024, sst_max_ts.into(), ChecksumType::Crc32c, None);
        let l0_key = [outer_start.as_slice(), b"sst"].concat();
        l0_builder
            .add(
                WRITE_CF,
                InnerKey::from_outer_key(&l0_key),
                &table::Value::new_with_meta_version(0, sst_max_ts, 0, b"sst-value"),
                None,
            )
            .unwrap();
        let (_, l0_data) = l0_builder.finish().unwrap();
        let l0 = L0Table::new(
            Arc::new(InMemFile::new(123, l0_data)),
            BlockCache::None,
            false,
            EncryptionState::Disabled,
        )
        .unwrap()
        .unwrap();

        let mem_tbl = CfTable::with_implementation(MemTableImplementation::Crossbeam);
        let keys = [b"write".as_slice(), b"lock".as_slice(), b"extra".as_slice()];
        let large_lock_value = vec![b'l'; 16 * 1024 * 1024 + 1];

        let mut write_batch = table::memtable::WriteBatch::new();
        write_batch.put(
            InnerKey::from_outer_key(&[outer_start.as_slice(), keys[0]].concat()),
            0,
            &UserMeta::new(1, 10).to_array(),
            10,
            b"value",
        );
        write_batch.put(
            InnerKey::from_outer_key(&[outer_start.as_slice(), keys[0]].concat()),
            table::BIT_DELETE,
            &[],
            20,
            &[],
        );
        write_batch.put(
            InnerKey::from_outer_key(&[outer_end.as_slice(), b"out-of-range"].concat()),
            0,
            &[],
            30,
            b"out-of-range-value",
        );
        mem_tbl
            .get_cf(WRITE_CF)
            .put_batch_preserve_tombstones(&mut write_batch, WRITE_CF);

        for (cf, key, version, value) in [
            (LOCK_CF, keys[1], 30, large_lock_value.as_slice()),
            (EXTRA_CF, keys[2], 40, b"extra-value".as_slice()),
        ] {
            let mut batch = table::memtable::WriteBatch::new();
            batch.put(
                InnerKey::from_outer_key(&[outer_start.as_slice(), key].concat()),
                0,
                &[],
                version,
                value,
            );
            mem_tbl.get_cf(cf).put_batch(&mut batch, None, cf);
        }

        let expected_mem_table_kv_size = mem_tbl.skip_list_size();
        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_mem_tbls(vec![mem_tbl]);
        builder.set_l0_tbls(vec![l0]);
        shard.set_data(builder.build());
        shard.update_write_sequence_on_recover(30);
        let snap = shard.new_snap_access();
        assert_eq!(snap.full_mem_table_kv_size(), expected_mem_table_kv_size);
        let mem_data = snap.build_full_mem_data();
        let mut online_snapshot = kvenginepb::ChangeSet::default();
        online_snapshot
            .merge_from_bytes(&snap.marshal_online_snapshot(TERM))
            .unwrap();
        assert_eq!(
            online_snapshot.get_snapshot().get_data_sequence(),
            snap.get_write_sequence()
        );
        assert_eq!(online_snapshot.get_snapshot().get_max_ts(), sst_max_ts);
        assert_eq!(
            get_shard_property(
                DEL_PREFIXES_KEY,
                online_snapshot.get_snapshot().get_properties()
            ),
            Some(
                DeletePrefixes::new_with_keyspace_id(KEYSPACE_ID)
                    .merge_prefix(&delete_prefix)
                    .marshal()
            )
        );
        assert_eq!(
            mem_data.as_slice().get_u32_le(),
            ONLINE_SNAPSHOT_MEM_DATA_FORMAT_VERSION
        );

        let restored = runtime
            .block_on(SnapAccess::construct_memtables(
                shard.id,
                shard.ver,
                &mem_data,
                &txn_chunk_manager,
                None,
            ))
            .unwrap();
        assert_eq!(restored.len(), 1);
        assert_eq!(
            restored[0].implementation(),
            MemTableImplementation::Crossbeam
        );
        for cf in 0..NUM_CFS {
            assert!(!restored[0].get_cf(cf).is_empty(), "CF {} is empty", cf);
        }

        let mut write_iter = restored[0].get_cf(WRITE_CF).new_iterator(false);
        write_iter.rewind();
        assert!(write_iter.valid());
        assert_eq!(write_iter.value().version, 20);
        assert!(write_iter.value().is_deleted());
        assert!(write_iter.next_version());
        assert_eq!(write_iter.value().version, 10);
        assert_eq!(write_iter.value().get_value(), b"value");
        write_iter.next();
        assert!(!write_iter.valid(), "out-of-range row was captured");

        for (cf, version, value) in [
            (LOCK_CF, 30, large_lock_value.as_slice()),
            (EXTRA_CF, 40, b"extra-value".as_slice()),
        ] {
            let mut iter = restored[0].get_cf(cf).new_iterator(false);
            iter.rewind();
            assert!(iter.valid());
            assert_eq!(iter.value().version, version);
            assert_eq!(iter.value().get_value(), value);
        }
    }

    #[test]
    fn test_full_mem_data_v3_filters_lock_versions_after_captured_sequence() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let dfs: Arc<dyn Dfs> = Arc::new(InMemFs::new());
        let txn_chunk_manager = TxnChunkManager::new(
            vec![],
            dfs,
            BlockCache::None,
            None,
            with_pool_size(1),
            TxnChunkManagerConfig::default(),
        );
        let keyspace_prefix = ApiV2::get_txn_keyspace_prefix(KEYSPACE_ID);
        let outer_start = [keyspace_prefix.as_slice(), b"a"].concat();
        let outer_end = [keyspace_prefix.as_slice(), b"z"].concat();
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            7,
            ShardRange::new(&outer_start, &outer_end),
            KEYSPACE_PREFIX_LEN,
            Arc::new(crate::options::Options::default()),
            &MasterKey::new(&[1u8; 32]),
            Arc::new(EncryptionKeyManager::new()),
        );
        shard.set_base_version(1_000);
        shard.update_write_sequence_on_recover(100);

        let retained_key = [outer_start.as_slice(), b"-retained"].concat();
        let future_only_key = [outer_start.as_slice(), b"-future-only"].concat();
        let mem_tbl = CfTable::with_implementation(MemTableImplementation::Crossbeam);
        let mut lock_batch = table::memtable::WriteBatch::new();
        lock_batch.put(
            InnerKey::from_outer_key(&retained_key),
            0,
            &[],
            1_090,
            b"lock-before-capture",
        );
        lock_batch.put(
            InnerKey::from_outer_key(&retained_key),
            0,
            &[],
            1_101,
            b"lock-after-capture",
        );
        lock_batch.put(
            InnerKey::from_outer_key(&future_only_key),
            0,
            &[],
            1_102,
            b"future-only-lock",
        );
        mem_tbl
            .get_cf(LOCK_CF)
            .put_batch(&mut lock_batch, None, LOCK_CF);

        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_mem_tbls(vec![mem_tbl]);
        shard.set_data(builder.build());
        let snap = shard.new_snap_access();
        assert_eq!(snap.get_mem_table_snap_version().into_inner(), 1_100);
        let mut online_snapshot = kvenginepb::ChangeSet::default();
        online_snapshot
            .merge_from_bytes(&snap.marshal_online_snapshot(TERM))
            .unwrap();
        assert_eq!(online_snapshot.get_snapshot().get_base_version(), 1_000);

        let restored = runtime
            .block_on(SnapAccess::construct_memtables(
                shard.id,
                shard.ver,
                &snap.build_full_mem_data(),
                &txn_chunk_manager,
                None,
            ))
            .unwrap();
        assert_eq!(restored.len(), 1);

        let mut value_owner = Vec::new();
        let retained = restored[0].get_cf(LOCK_CF).get(
            InnerKey::from_outer_key(&retained_key),
            u64::MAX,
            &mut value_owner,
        );
        assert!(retained.is_valid());
        assert_eq!(retained.version, 1_090);
        assert_eq!(retained.get_value(), b"lock-before-capture");

        value_owner.clear();
        let future_only = restored[0].get_cf(LOCK_CF).get(
            InnerKey::from_outer_key(&future_only_key),
            u64::MAX,
            &mut value_owner,
        );
        assert!(!future_only.is_valid());
    }

    #[test]
    fn test_online_snapshot_preserves_unresolved_txn_file_locks() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let _enter = runtime.enter();
        let dfs: Arc<dyn Dfs> = Arc::new(InMemFs::new());
        let txn_chunk_manager = TxnChunkManager::new(
            vec![],
            dfs.clone(),
            BlockCache::None,
            None,
            with_pool_size(1),
            TxnChunkManagerConfig::default(),
        );

        let keyspace_prefix = ApiV2::get_txn_keyspace_prefix(KEYSPACE_ID);
        let outer_start = [keyspace_prefix.as_slice(), b"a"].concat();
        let outer_end = [keyspace_prefix.as_slice(), b"z"].concat();
        let lock_key = [keyspace_prefix.as_slice(), b"lock-key"].concat();
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            7,
            ShardRange::new(&outer_start, &outer_end),
            KEYSPACE_PREFIX_LEN,
            Arc::new(crate::options::Options::default()),
            &MasterKey::new(&[1u8; 32]),
            Arc::new(EncryptionKeyManager::new()),
        );
        let chunk_id = 77;
        let mut chunk_builder = TxnChunkBuilder::new(chunk_id, 64, None, TXN_FILE_FORMAT_V2);
        chunk_builder.add_entry(
            InnerKey::from_outer_key(&lock_key),
            OP_PUT,
            PessimisticAction::SkipCheck,
            b"value",
        );
        let mut chunk_data = Vec::new();
        chunk_builder.finish(&mut chunk_data);
        runtime
            .block_on(dfs.create(
                chunk_id,
                chunk_data.into(),
                dfs::Options::default().with_type(FileType::TxnChunk),
            ))
            .unwrap();
        txn_chunk_manager
            .prepare(chunk_id, EncryptionState::Disabled, IoType::Other)
            .unwrap();
        let txn_chunk = txn_chunk_manager.get(chunk_id).unwrap();

        let start_ts = 100;
        let mut lock = txn_types::Lock::new(
            txn_types::LockType::Put,
            lock_key.clone(),
            start_ts.into(),
            3000,
            None,
            0.into(),
            0,
            (start_ts + 1).into(),
        );
        lock.is_txn_file = true;
        let lock_value = lock.to_bytes();
        let shard_data = shard.get_data();
        let txn_ctx = TxnCtx::new(
            Bytes::new(),
            lock_value.clone().into(),
            start_ts,
            shard_data.inner_start(),
            shard_data.inner_end(),
        );
        let txn_file = TxnFile::new(
            TxnFileId::new(shard.id, shard.ver, start_ts),
            vec![txn_chunk],
            txn_ctx,
        )
        .unwrap();
        let mut shard_data_builder = ShardDataBuilder::new(shard_data);
        shard_data_builder.set_lock_txn_files(vec![txn_file]);
        shard.set_data(shard_data_builder.build());

        // Loaded TxnFile objects in ShardData are the source of truth. Neither
        // SnapAccess construction nor file enumeration should parse this
        // property again.
        shard.set_property(TXN_FILE_REF, &[0x0a]);
        let source_snapshot = shard.new_snap_access();
        assert_eq!(
            source_snapshot.get_all_files_with_type(),
            vec![(FileType::TxnChunk, chunk_id)]
        );
        let snapshot_data = source_snapshot.marshal_online_snapshot(TERM);
        let mut change_set = kvenginepb::ChangeSet::default();
        change_set.merge_from_bytes(&snapshot_data).unwrap();
        assert_eq!(
            get_shard_property(TERM_KEY, change_set.get_snapshot().get_properties()),
            Some(TERM.to_le_bytes().to_vec())
        );
        let txn_file_ref_data =
            get_shard_property(TXN_FILE_REF, change_set.get_snapshot().get_properties())
                .expect("online snapshot must carry unresolved txn-file refs");
        let mut txn_file_refs = TxnFileRefs::default();
        txn_file_refs.merge_from_bytes(&txn_file_ref_data).unwrap();
        assert_eq!(txn_file_refs.get_txn_file_refs().len(), 1);
        let txn_file_ref = &txn_file_refs.get_txn_file_refs()[0];
        assert_eq!(txn_file_ref.get_start_ts(), start_ts);
        assert_eq!(txn_file_ref.get_chunk_ids(), &[chunk_id]);
        assert_eq!(txn_file_ref.get_lock_val_prefix(), lock_value);

        let snap_ctx = SnapCtx {
            dfs,
            master_key: MasterKey::new(&[1u8; 32]),
            block_cache: BlockCache::None,
            vector_index_cache: None,
            columnar_file_cache: None,
            fts_cache: FtsCache::disabled(),
            fts_delta_cache: FtsDeltaCache::disabled(),
            schema_files: None,
            txn_chunk_manager,
            ia_ctx: IaCtx::disabled(&Default::default()),
            prepare_type: PrepareType::All,
            read_columnar: true,
            meta_file_cache: new_meta_file_cache(1024),
            columnar_meta_cache: ColumnarMetaCache::default(),
            encryption_key_manager: Arc::new(EncryptionKeyManager::new()),
            strict_file_memory_quota: Arc::new(tikv_util::memory::MemoryQuota::new(usize::MAX)),
            strict_file_memory_quota_wait_timeout: std::time::Duration::from_secs(2),
        };
        let restored_snapshot = block_on(SnapAccess::from_change_set(
            "test-online-snapshot-txn-file-lock",
            &snap_ctx,
            change_set,
            false,
        ))
        .unwrap();
        assert_eq!(restored_snapshot.get_lock_txn_files().len(), 1);
        let restored_txn_file = &restored_snapshot.get_lock_txn_files()[0];
        assert_eq!(restored_txn_file.start_ts(), start_ts);
        assert_eq!(restored_txn_file.chunk_ids(), vec![chunk_id]);
        assert_eq!(restored_txn_file.get_lock_val_prefix(), lock_value);
    }

    fn build_encryption_meta_for_test(
        runtime: &tokio::runtime::Runtime,
        keyspace_id: u32,
    ) -> kvenginepb::EncryptionMeta {
        let mut config = MasterKeyConfig::default();
        config.vendor = "test".to_owned();
        let (master_key, master_key_ciphertext) =
            runtime.block_on(config.generate_new_master_key()).unwrap();

        let data_key_id = cloud_encryption::generate_data_key_id();
        let data_key = master_key
            .generate_encryption_key_v2()
            .with_key_id(data_key_id)
            .unwrap();

        let mut meta = kvenginepb::EncryptionMeta::default();
        meta.set_keyspace_id(keyspace_id);
        let mut mk = kvenginepb::MasterKey::default();
        mk.vendor = config.vendor;
        mk.ciphertext = master_key_ciphertext.to_vec();
        cloud_encryption::set_master_key_encryption_context(&mut mk, &config.encryption_context);
        meta.set_master_key(mk);
        meta.mut_current().set_data_key_id(data_key_id);

        let mut dk = kvenginepb::DataKey::new();
        dk.set_ciphertext(data_key.export());
        meta.mut_data_keys().insert(data_key_id, dk);
        meta
    }

    #[test]
    fn test_scan_ctx_has_filter_ignores_empty_scan_ctx() {
        assert!(!super::scan_ctx_has_filter(None));

        let empty_scan_ctx = crate::table::columnar::filter::TableScanCtx::default();
        assert!(!super::scan_ctx_has_filter(Some(&empty_scan_ctx)));

        let non_empty_scan_ctx = crate::table::columnar::filter::TableScanCtx::new(
            tipb::Executor::default(),
            vec![tipb::Expr::default()],
        );
        assert!(super::scan_ctx_has_filter(Some(&non_empty_scan_ctx)));
    }

    #[test]
    fn test_remote_snapshot_preserves_storage_class_spec() {
        let runtime = tokio::runtime::Runtime::new().unwrap();
        let _enter = runtime.enter();
        let dfs: Arc<dyn Dfs> = Arc::new(InMemFs::new());
        let txn_chunk_manager = TxnChunkManager::new(
            vec![],
            dfs.clone(),
            BlockCache::None,
            None,
            with_pool_size(2),
            TxnChunkManagerConfig::default(),
        );
        let snap_ctx = SnapCtx {
            dfs,
            master_key: MasterKey::new(&[1u8; 32]),
            block_cache: BlockCache::None,
            vector_index_cache: None,
            columnar_file_cache: None,
            fts_cache: FtsCache::disabled(),
            fts_delta_cache: FtsDeltaCache::disabled(),
            schema_files: None,
            txn_chunk_manager,
            ia_ctx: IaCtx::disabled(&Default::default()),
            prepare_type: PrepareType::All,
            read_columnar: true,
            meta_file_cache: new_meta_file_cache(1024),
            columnar_meta_cache: ColumnarMetaCache::default(),
            encryption_key_manager: Arc::new(EncryptionKeyManager::new()),
            strict_file_memory_quota: Arc::new(tikv_util::memory::MemoryQuota::new(usize::MAX)),
            strict_file_memory_quota_wait_timeout: std::time::Duration::from_secs(2),
        };
        let storage_class_specs = [
            StorageClassSpec::default(),
            StorageClass::Standard.into(),
            StorageClass::Ia.into(),
            StorageClassSpec {
                default_tier: StorageClass::Unspecified,
                transit_rules: vec![StorageClassTransitRule {
                    tier: StorageClass::Ia,
                    transit_after: std::time::Duration::from_secs(10),
                }],
            },
        ];

        for (shard_id, storage_class_spec) in storage_class_specs.iter().cloned().enumerate() {
            let shard_id = shard_id as u64 + 1;
            let mut properties = kvenginepb::Properties {
                shard_id,
                ..Default::default()
            };
            if storage_class_spec.is_specified() {
                set_shard_property(
                    STORAGE_CLASS_KEY,
                    &mut properties,
                    storage_class_spec.marshal(),
                );
            }
            let master_key = MasterKey::new(&[1u8; 32]);
            let shard = Shard::new(
                1,
                &properties,
                1,
                ShardRange::new(&[], GLOBAL_SHARD_END_KEY),
                0,
                Arc::new(crate::options::Options::default()),
                &master_key,
                Arc::new(EncryptionKeyManager::new()),
            );
            let snap_access = shard.new_snap_access();
            let (_, snapshot_bytes) = snap_access.marshal(
                &[(Bytes::new(), Bytes::copy_from_slice(GLOBAL_SHARD_END_KEY))],
                false,
                false,
            );

            let mut change_set = kvenginepb::ChangeSet::default();
            change_set.merge_from_bytes(&snapshot_bytes).unwrap();
            let serialized_spec = get_shard_property(
                STORAGE_CLASS_KEY,
                change_set.get_snapshot().get_properties(),
            );
            assert_eq!(serialized_spec.is_some(), storage_class_spec.is_specified());
            assert_eq!(
                StorageClassSpec::unmarshal(serialized_spec.as_deref()),
                storage_class_spec
            );

            let remote_snap = block_on(SnapAccess::from_change_set(
                "test", &snap_ctx, change_set, false,
            ))
            .unwrap();
            assert_eq!(remote_snap.storage_class_spec, storage_class_spec);
            assert_eq!(remote_snap.from_ia_region(), storage_class_spec.can_be_ia());
        }
    }

    #[test]
    fn test_vector_index_reader_empty_index_fallback() {
        let mut opts = crate::options::Options::default();
        opts.read_columnar = true;
        let opt = Arc::new(opts);
        let master_key = MasterKey::new(&[1u8; 32]);
        let range = ShardRange::new(&[], GLOBAL_SHARD_END_KEY);
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            1,
            range,
            0,
            opt,
            &master_key,
            Arc::new(cloud_encryption::EncryptionKeyManager::new()),
        );

        let mut vector_indexes = VectorIndexes::default();
        vector_indexes.update_snap_version(130, 2, 8, 904705.into());

        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_vector_indexes(vector_indexes);
        shard.set_data(builder.build());

        let snap_access = SnapAccess::new(&shard);
        assert!(snap_access.read_columnar);

        let mut vec_col = tipb::ColumnInfo::new();
        vec_col.set_column_id(8);
        let schema = Schema::new(SchemaBuf::new(
            130,
            new_int_handle_column_info(),
            new_version_column_info(),
            vec![vec_col],
            vec![],
            8,
            vec![],
            vec![],
            StorageClassSpec::default(),
            None,
        ));

        let ann_query = Arc::new(tipb::AnnQueryInfo::default());
        let reader = snap_access
            .new_vector_index_reader(
                130,
                2,
                8,
                &[0.0, 0.0, 0.0],
                10,
                schema,
                u64::MAX,
                None,
                None,
                ann_query,
            )
            .unwrap();
        assert!(reader.is_none());
    }

    #[test]
    fn test_estimated_range_blocks_size() {
        let mut cs_pb = kvenginepb::ChangeSet::default();
        let mut cs = ChangeSet::new(cs_pb.clone());
        let snap = cs_pb.mut_snapshot();
        let mut build_table_fn =
            |start: i32, end: i32, level: u32, tbl_entries: usize, step: usize| {
                let mut kvs = vec![];
                for i in (start..end).step_by(step) {
                    let key = format!("key{:05x}", i);
                    let val = key.repeat(10);
                    kvs.push((key, val));
                    if kvs.len() == tbl_entries {
                        let tbl = build_test_table_with_kvs(&kvs);
                        let mut tbl_create = TableCreate::default();
                        tbl_create.id = tbl.id();
                        tbl_create.level = level;
                        tbl_create.smallest = tbl.smallest().to_vec();
                        tbl_create.biggest = tbl.biggest().to_vec();
                        tbl_create.meta_offset = tbl.meta_offset();
                        kvs.truncate(0);
                        cs.ln_tables.insert(tbl.id(), tbl.clone());
                        snap.mut_table_creates().push(tbl_create);
                    }
                }
            };
        build_table_fn(0, 10000, 3, 1000, 1);
        build_table_fn(0, 10000, 2, 500, 5);
        build_table_fn(0, 10000, 1, 200, 20);
        cs.change_set = cs_pb;
        let range = ShardRange::new(&[], GLOBAL_SHARD_END_KEY);
        let opt = Arc::new(crate::options::Options::default());
        let master_key = MasterKey::new(&[1u8; 32]);
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            1,
            range,
            0,
            opt,
            &master_key,
            Arc::new(cloud_encryption::EncryptionKeyManager::new()),
        );
        let mut builder = ShardDataBuilder::new(shard.get_data());
        create_snapshot_tables(
            &mut builder,
            cs.get_snapshot(),
            &cs,
            false,
            PrepareType::All,
        );
        builder.set_schema(
            cs.get_schema_version(),
            cs.get_restore_version(),
            cs.get_schema_file(),
        );
        shard.set_data(builder.build());
        let snap = shard.new_snap_access();

        let build_range_fn = |start: i32, end: i32| {
            (
                Bytes::from(format!("key{:05x}", start)),
                Bytes::from(format!("key{:05x}", end)),
            )
        };
        let total_blocks_size = snap.estimated_range_blocks_size(&[build_range_fn(0, 10000)]);
        assert_eq!(total_blocks_size, 1103566);

        // verify that many small ranges are properly deduplicated, num blocks never
        // exceed total.
        let mut many_small_ranges = vec![];
        for i in (0..10000).step_by(10) {
            let small_range = build_range_fn(i, i + 5);
            many_small_ranges.push(small_range);
        }
        let many_small_ranges_blocks_size = snap.estimated_range_blocks_size(&many_small_ranges);
        assert_eq!(many_small_ranges_blocks_size + 112, total_blocks_size);

        let half_num_blocks = snap.estimated_range_blocks_size(&[build_range_fn(5000, 10000)]);
        assert_eq!(half_num_blocks, 548206);

        for i in (100..10000).step_by(100) {
            let blocks_size = snap.estimated_range_blocks_size(&[build_range_fn(i, i + 1)]);
            // some range on level 1 doesn't overlap any table, so blocks_size may vary.
            assert!(blocks_size == 11137 || blocks_size == 7577);
        }

        let blocks_size = snap.estimated_range_blocks_size(&[
            build_range_fn(1, 2),
            build_range_fn(2, 3),
            build_range_fn(3, 4),
            build_range_fn(4, 5),
        ]);
        // each level only access one block.
        assert_eq!(blocks_size, 11137);

        let blocks_size = snap.estimated_range_blocks_size(&[
            build_range_fn(1, 2),
            build_range_fn(2000, 2001),
            build_range_fn(3000, 3001),
            build_range_fn(4000, 4001),
        ]);
        // each range on each level access one block.
        assert_eq!(blocks_size, 44548);
    }

    #[test]
    fn test_marshal_omits_snapshot_versions() {
        let mut properties = kvenginepb::Properties::new();
        properties.shard_id = 1;
        let range = ShardRange::new(&[], GLOBAL_SHARD_END_KEY);
        let opt = Arc::new(crate::options::Options::default());
        let master_key = MasterKey::new(&[1u8; 32]);
        let shard = Shard::new(
            1,
            &properties,
            1,
            range,
            0,
            opt,
            &master_key,
            Arc::new(cloud_encryption::EncryptionKeyManager::new()),
        );
        shard.set_base_version(13);
        shard
            .meta_seq
            .store(57, std::sync::atomic::Ordering::Release);
        shard
            .write_sequence
            .store(89, std::sync::atomic::Ordering::Release);

        let snap_access = shard.new_snap_access();
        shard
            .write_sequence
            .store(144, std::sync::atomic::Ordering::Release);

        let ranges = [(Bytes::new(), Bytes::from_static(GLOBAL_SHARD_END_KEY))];
        let (_, snapshot_bytes) = snap_access.marshal(&ranges, true, false);
        let mut change_set = kvenginepb::ChangeSet::default();
        change_set.merge_from_bytes(&snapshot_bytes).unwrap();
        let snapshot = change_set.get_snapshot();

        assert_eq!(change_set.get_sequence(), 57);
        assert_eq!(snapshot.get_base_version(), 0);
        assert_eq!(snapshot.get_data_sequence(), 0);
    }

    fn new_snap_access_with_sst(sst: SsTable) -> SnapAccess {
        let range = ShardRange::new(&[], GLOBAL_SHARD_END_KEY);
        let shard = Shard::new(
            1,
            &kvenginepb::Properties::new(),
            1,
            range,
            0,
            Arc::new(crate::options::Options::default()),
            &MasterKey::new(&[1u8; 32]),
            Arc::new(EncryptionKeyManager::new()),
        );
        let mut write_cf_builder = ShardCfBuilder::new(WRITE_CF);
        write_cf_builder.add_table(sst, 1);
        let mut builder = ShardDataBuilder::new(shard.get_data());
        // ShardData requires a writable memtable. Keep it empty so all records
        // observed by the iterator still come from the SST.
        builder.set_mem_tbls(vec![CfTable::new()]);
        builder.set_cfs([
            write_cf_builder.build(),
            ShardCf::new(LOCK_CF),
            ShardCf::new(EXTRA_CF),
        ]);
        shard.set_data(builder.build());

        shard.new_snap_access()
    }

    fn new_sst_read_iterator(kvs: &[(&[u8], u64, &[u8])], read_ts: u64) -> super::Iterator {
        let sst_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_id);
        for &(key, version, value) in kvs {
            let value = table::Value::new_with_meta_version(OP_PUT, version, 0, value);
            sst_builder
                .add(InnerKey::from_inner_buf(key), &value, 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(InMemFile::new(sst_id, sst_buf.into()));
        let sst = SsTable::new(sst_file, new_test_cache(), EncryptionState::Disabled).unwrap();

        // Every fixture contains both single-version keys in the normal data
        // section and a multi-version key whose history is in old data.
        assert!(sst.footer().data_len() > 0);
        assert!(sst.footer().old_data_len() > 0);

        new_snap_access_with_sst(sst).new_iterator(WRITE_CF, false, false, Some(read_ts), false)
    }

    #[test]
    fn test_has_newer_does_not_resolve_hidden_blob_value() {
        let sst_id = TEST_ID_ALLOC.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
        let mut sst_builder = new_table_builder_for_test(sst_id);
        let key = InnerKey::from_inner_buf(b"blob-key");

        // The latest version references a blob table that is deliberately not
        // installed in the snapshot. Resolving this value would panic.
        let mut newer = table::Value::new_with_meta_version(OP_PUT, 30, 0, b"hidden");
        newer.set_blob_ref();
        let missing_blob_ref = BlobRef::new(u64::MAX, 0, 1, 1);
        sst_builder
            .add(key, &newer, Some(missing_blob_ref))
            .unwrap();

        let older = table::Value::new_with_meta_version(OP_PUT, 20, 0, b"visible");
        sst_builder.add(key, &older, 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(InMemFile::new(sst_id, sst_buf.into()));
        let sst = SsTable::new(sst_file, new_test_cache(), EncryptionState::Disabled).unwrap();
        assert!(sst.footer().old_data_len() > 0);

        let snap = new_snap_access_with_sst(sst);
        let visible = snap.get(WRITE_CF, b"blob-key", 20);
        assert_eq!(visible.version, 20);
        assert_eq!(visible.get_value(), b"visible");

        assert!(snap.has_newer(WRITE_CF, b"blob-key", 20));
        assert!(!snap.has_newer(WRITE_CF, b"blob-key", 30));
    }

    #[test]
    fn test_sst_iterator_met_newer_ts_stays_false_without_newer_versions() {
        let mut iter = new_sst_read_iterator(
            &[
                (b"key1", 10, b"v10"),
                (b"key2", 20, b"v20"),
                (b"key2", 10, b"v10"),
                (b"key3", 15, b"v15"),
            ],
            20,
        );

        assert!(!iter.met_newer_ts());
        iter.rewind();
        assert_eq!(iter.key(), b"key1");
        assert_eq!(iter.version(), 10);
        assert!(!iter.met_newer_ts());

        iter.next();
        assert_eq!(iter.key(), b"key2");
        assert_eq!(iter.version(), 20);
        assert!(!iter.met_newer_ts());

        iter.next();
        assert_eq!(iter.key(), b"key3");
        assert_eq!(iter.version(), 15);
        assert!(!iter.met_newer_ts());
    }

    #[test]
    fn test_sst_iterator_met_newer_ts_tracks_newer_version_with_visible_history() {
        let mut iter = new_sst_read_iterator(
            &[
                (b"key1", 10, b"v10"),
                (b"key2", 30, b"v30"),
                (b"key2", 20, b"v20"),
                (b"key2", 10, b"v10"),
                (b"key3", 20, b"v20"),
            ],
            20,
        );

        iter.rewind();
        assert_eq!(iter.key(), b"key1");
        assert_eq!(iter.version(), 10);
        assert!(!iter.met_newer_ts());

        // The latest key2 version is in the normal data section. Selecting its
        // visible version crosses into the SST old-data section.
        iter.next();
        assert_eq!(iter.key(), b"key2");
        assert_eq!(iter.version(), 20);
        assert!(iter.met_newer_ts());

        // The observation is iterator-wide and remains set after later reads.
        iter.next();
        assert_eq!(iter.key(), b"key3");
        assert_eq!(iter.version(), 20);
        assert!(iter.met_newer_ts());
    }

    #[test]
    fn test_sst_iterator_met_newer_ts_tracks_skipped_newer_only_key() {
        let mut iter = new_sst_read_iterator(
            &[
                (b"key1", 30, b"v30"),
                (b"key2", 20, b"v20"),
                (b"key3", 40, b"v40"),
                (b"key3", 10, b"v10"),
            ],
            20,
        );

        // key1 has no visible version, so rewind skips it but records that a
        // version newer than read_ts was observed.
        iter.rewind();
        assert_eq!(iter.key(), b"key2");
        assert_eq!(iter.version(), 20);
        assert!(iter.met_newer_ts());

        // The observation is iterator-wide and remains set after later reads.
        iter.next();
        assert_eq!(iter.key(), b"key3");
        assert_eq!(iter.version(), 10);
        assert!(iter.met_newer_ts());
    }

    const MAX_I: usize = 100;

    fn mem_table_op_strategy() -> impl Strategy<Value = MemTableOp> {
        prop_oneof![
            (
                prop::collection::vec(1..MAX_I, 0..=20usize),
                any::<bool>(),
                0..5usize,
            )
                .prop_map(|(batch, switch, mvcc_count)| MemTableOp::WriteBatch(
                    batch, switch, mvcc_count
                )),
            (prop::collection::vec(1..MAX_I, 0..=20usize), any::<bool>())
                .prop_map(|(batch, switch)| MemTableOp::WriteTxnFile(batch, switch)),
        ]
    }

    prop_compose! {
        fn arb_mem_table_ops(size: usize)
            (ops in prop::collection::vec(mem_table_op_strategy(), 0..=size))
            -> Vec<MemTableOp> {
            ops
        }
    }

    // [10, 20, 30, 40, 50] -> [(10, 20), (30, 40)]
    fn query_ranges() -> impl Strategy<Value = Vec<(usize, usize)>> {
        prop::collection::vec(0..MAX_I + 1, 0..=10usize).prop_map(|mut is| -> Vec<(usize, usize)> {
            is.sort();
            let ranges = is
                .chunks_exact(2)
                .filter_map(|chunk| (chunk[0] != chunk[1]).then_some((chunk[0], chunk[1])))
                .collect::<Vec<_>>();
            if ranges.is_empty() {
                vec![(0, MAX_I)]
            } else {
                ranges
            }
        })
    }

    proptest! {
        #[test]
        fn test_serde_mem_tables(
            ops in arb_mem_table_ops(10),
            ranges in query_ranges(),
            enable_enc in any::<bool>(),
        ) {
            let runtime = tokio::runtime::Runtime::new().unwrap();
            let _enter = runtime.enter();

            let engine_id = 1;
            let shard_id = 1;
            let shard_ver = 10;

            let kb = KeyBuilder::new(KEYSPACE_ID, "t_");
            let dfs: Arc<dyn crate::dfs::Dfs> = Arc::new(InMemFs::new());
            let txn_chunk_manager = TxnChunkManager::new(vec![], dfs.clone(), BlockCache::None, None, with_pool_size(2), TxnChunkManagerConfig::default());

            let master_key = MasterKey::new(&[1u8; 32]);
            let enc_key_mgr = Arc::new(EncryptionKeyManager::new());
            let enc_meta = enable_enc.then(|| build_encryption_meta_for_test(&runtime, KEYSPACE_ID));
            let enc_key = enc_meta.clone().and_then(|m| {
                cloud_encryption::current_data_key_from_encryption_meta_bytes(
                    enc_key_mgr.clone(),
                    &m.write_to_bytes().unwrap(),
                    Some(KEYSPACE_ID),
                )
                .ok()
            });

            let (mem_tbls, ref_store) = make_mem_tables(ops, &kb, dfs.clone(), enc_key.as_ref());
            verify_mem_tables(&mem_tbls, &ref_store, &[(kb.i_to_inner_key(0).as_ref(), kb.i_to_inner_key(MAX_I).as_ref())])?;

            let (outer_start, outer_end) = ApiV2::get_txn_keyspace_range(KEYSPACE_ID);
            let range = ShardRange::new(&outer_start, &outer_end);
            let inner_key_off = KEYSPACE_PREFIX_LEN;
            let opt = Arc::new(crate::options::Options::default());

            let inner_ranges = ranges.iter().map(|&(start, end)| {
                (kb.i_to_inner_key(start), kb.i_to_inner_key(end))
            }).collect::<Vec<_>>();
            let outer_ranges = ranges.into_iter().map(|(start, end)| {
                (Bytes::from(kb.i_to_outer_key(start)), Bytes::from(kb.i_to_outer_key(end)))
            }).collect::<Vec<_>>();

            // Serialize
            let mem_bin = {
                let mut props = kvenginepb::Properties {
                    shard_id,
                    ..Default::default()
                };
                if let Some(enc_meta) = &enc_meta {
                    props.mut_keys().push(ENCRYPTION_META_KEY.to_string());
                    props.mut_values().push(enc_meta.write_to_bytes().unwrap());
                }
                let shard = Shard::new(
                    engine_id,
                    &props,
                    shard_ver,
                    range,
                    inner_key_off,
                    opt,
                    &master_key,
                    Arc::new(cloud_encryption::EncryptionKeyManager::new()),
                );
                let mut builder = ShardDataBuilder::new(shard.get_data());
                builder.set_mem_tbls(mem_tbls);
                shard.set_data(builder.build());
                let snap_access = shard.new_snap_access();

                snap_access.build_mem_data(&outer_ranges, u64::MAX).0
            };
            let format_ver = mem_bin.as_slice().get_u32_le();
            prop_assert_eq!(format_ver, MEM_DATA_FORMAT_V2);

            // Deserialize
            let snap_ctx = SnapCtx {
                dfs,
                master_key,
                block_cache: BlockCache::None,
                vector_index_cache: None,
                columnar_file_cache: None,
                fts_cache: FtsCache::disabled(),
                fts_delta_cache: FtsDeltaCache::disabled(),
                schema_files: None,
                txn_chunk_manager,
                ia_ctx: IaCtx::disabled(&Default::default()),
                prepare_type: PrepareType::All,
                read_columnar: true,
                meta_file_cache: new_meta_file_cache(1024),
                columnar_meta_cache: ColumnarMetaCache::default(),
                encryption_key_manager: Arc::new(cloud_encryption::EncryptionKeyManager::new()),
                strict_file_memory_quota: Arc::new(tikv_util::memory::MemoryQuota::new(usize::MAX)),
                strict_file_memory_quota_wait_timeout: std::time::Duration::from_secs(2),
            };
            let mut snap_pb = kvenginepb::Snapshot::default();
            snap_pb.set_inner_key_off(KEYSPACE_PREFIX_LEN as u32);
            let props = snap_pb.mut_properties();
            if let Some(enc_meta) = &enc_meta {
                props.mut_keys().push(ENCRYPTION_META_KEY.to_string());
                props.mut_values().push(enc_meta.write_to_bytes().unwrap());
            }
            let mut cs = kvenginepb::ChangeSet::default();
            cs.set_shard_id(shard_id);
            cs.set_shard_ver(shard_ver);
            cs.set_snapshot(snap_pb);
            let snap_bin = cs.write_to_bytes().unwrap();
            let mem_limiter = MemoryLimiter::new(u64::MAX, None);
            let remote_snap = block_on(SnapAccess::construct_snapshot("test", &snap_ctx, &mem_bin, &snap_bin, mem_limiter)).unwrap().0;

            let inner_ranges = inner_ranges.iter().map(|(start, end)| (start.as_ref(), end.as_ref())).collect::<Vec<_>>();
            let ref_store_in_ranges = ref_store.new_in_ranges(&inner_ranges);
            verify_mem_tables(remote_snap.data.mem_tbls.as_slice(), &ref_store_in_ranges, &inner_ranges)?;
        }
    }

    #[derive(Debug, Clone)]
    enum MemTableOp {
        WriteBatch(
            Vec<usize>,
            bool,  // switch
            usize, // mvcc version count
        ),
        WriteTxnFile(Vec<usize>, bool /* switch */),
    }

    #[derive(Default, Debug)]
    struct RefStore {
        inner: BTreeMap<(Bytes, u64), String>,
    }

    impl RefStore {
        fn put(&mut self, key: OwnedInnerKey, ver: u64, val: String) {
            self.inner.insert((key.into_inner(), ver), val);
        }

        fn get_value(&self, key: InnerKey<'_>, ver: u64) -> Option<&String> {
            self.inner.get(&(Bytes::copy_from_slice(key.deref()), ver))
        }

        fn new_in_ranges(&self, ranges: &[(InnerKey<'_>, InnerKey<'_>)]) -> Self {
            let mut new_store = RefStore::default();
            for &(start, end) in ranges {
                let start = Bytes::copy_from_slice(start.deref());
                let end = Bytes::copy_from_slice(end.deref());
                for ((k, ver), v) in self
                    .inner
                    .range::<(Bytes, u64), _>((start.clone(), 0)..(end, 0))
                {
                    new_store.inner.insert((k.clone(), *ver), v.clone());
                }
            }
            new_store
        }

        fn len(&self) -> usize {
            self.inner.len()
        }
    }

    fn make_mem_tables(
        ops: Vec<MemTableOp>,
        kb: &KeyBuilder,
        dfs: Arc<dyn Dfs>,
        enc_key: Option<&EncryptionKey>,
    ) -> (Vec<CfTable>, RefStore) {
        let mut mem_tbls = vec![CfTable::new()];
        let mut ref_store = RefStore::default();
        let mut wb = crate::table::memtable::WriteBatch::new();
        let mut next_txn_chunk_id = 1000;

        let switch_mem_table = |mem_tbls: &mut Vec<CfTable>| {
            let mut new_mem_tbls = Vec::with_capacity(mem_tbls.len() + 1);
            new_mem_tbls.push(CfTable::new());
            new_mem_tbls.append(mem_tbls);
            *mem_tbls = new_mem_tbls;
        };

        for (idx, op) in ops.into_iter().enumerate() {
            let start_ts = idx as u64 * 100;
            let commit_ts = (idx as u64 + 1) * 100;
            let user_meta = UserMeta::new(start_ts, commit_ts);

            match op {
                MemTableOp::WriteBatch(is, switch, mvcc_count) => {
                    for i in is {
                        let key = kb.i_to_inner_key(i);
                        let val = kb.i_to_val(start_ts as usize + i);
                        wb.put(
                            key.as_ref(),
                            0,
                            &user_meta.to_array(),
                            commit_ts,
                            val.as_bytes(),
                        );
                        ref_store.put(key.clone(), commit_ts, val.clone());
                        let mut start_ts = start_ts;
                        let mut commit_ts = commit_ts;
                        for _ in 0..mvcc_count {
                            start_ts += 1;
                            commit_ts += 1;
                            let user_meta = UserMeta::new(start_ts, commit_ts);
                            let val = kb.i_to_val(start_ts as usize + i);
                            wb.put(
                                key.as_ref(),
                                0,
                                &user_meta.to_array(),
                                commit_ts,
                                val.as_bytes(),
                            );
                            ref_store.put(key.clone(), commit_ts, val.clone());
                        }
                    }
                    let mem_tbl = mem_tbls[0].get_cf(WRITE_CF);
                    mem_tbl.put_batch(&mut wb, None, WRITE_CF);
                    wb.reset();

                    if switch {
                        switch_mem_table(&mut mem_tbls);
                    }
                }
                MemTableOp::WriteTxnFile(mut is, switch) => {
                    is.sort();
                    is.dedup();
                    let mut txn_chunks = vec![];
                    for chunk in is.chunks(3) {
                        next_txn_chunk_id += 1;
                        let mut builder = TxnChunkBuilder::new(
                            next_txn_chunk_id,
                            64,
                            enc_key.cloned(),
                            TXN_FILE_FORMAT_V2,
                        );
                        for &i in chunk {
                            let key = kb.i_to_inner_key(i);
                            let val = kb.i_to_val(start_ts as usize + i);
                            builder.add_entry(
                                InnerKey::from_outer_key(&kb.i_to_key(i)),
                                OP_PUT,
                                PessimisticAction::SkipCheck,
                                val.as_bytes(),
                            );
                            ref_store.put(key, commit_ts, val.clone());
                        }
                        let mut chunk_data = vec![];
                        builder.finish(&mut chunk_data);

                        let opts = dfs::Options::default().with_type(dfs::FileType::TxnChunk);
                        block_on(dfs.create(next_txn_chunk_id, chunk_data.clone().into(), opts))
                            .unwrap();

                        let chunk_file =
                            Arc::new(InMemFile::new(next_txn_chunk_id, chunk_data.into()));
                        let encryption_state = enc_key.to_encryption_state();
                        let txn_chunk =
                            TxnChunk::new(chunk_file, BlockCache::None, encryption_state).unwrap();
                        txn_chunks.push(txn_chunk);
                    }

                    if !txn_chunks.is_empty() {
                        let id = TxnFileId::new(1, 1, start_ts);
                        let lower_bound = InnerKey::from_inner_buf(b"");
                        let upper_bound = InnerKey::from_inner_buf(GLOBAL_SHARD_END_KEY);
                        let txn_ctx = TxnCtx::new(
                            user_meta.to_array().to_vec().into(),
                            Default::default(),
                            commit_ts,
                            lower_bound,
                            upper_bound,
                        );
                        let txn_file = TxnFile::new(id, txn_chunks, txn_ctx).unwrap();
                        mem_tbls[0] = mem_tbls[0].add_write_cf_txn_files(&[txn_file]);
                        if switch {
                            switch_mem_table(&mut mem_tbls);
                        }
                    }
                }
            }
        }
        (mem_tbls, ref_store)
    }

    #[test]
    fn test_build_mem_data_reports_filtered_newer_write() {
        let kb = KeyBuilder::new(KEYSPACE_ID, "t_");
        let dfs: Arc<dyn Dfs> = Arc::new(InMemFs::new());
        let (mem_tbls, _) = make_mem_tables(
            vec![MemTableOp::WriteBatch(vec![0], false, 0)],
            &kb,
            dfs,
            None,
        );

        let (outer_start, outer_end) = ApiV2::get_txn_keyspace_range(KEYSPACE_ID);
        let range = ShardRange::new(&outer_start, &outer_end);
        let props = kvenginepb::Properties {
            shard_id: 1,
            ..Default::default()
        };
        let shard = Shard::new(
            1,
            &props,
            1,
            range,
            KEYSPACE_PREFIX_LEN,
            Arc::new(crate::options::Options::default()),
            &MasterKey::new(&[1u8; 32]),
            Arc::new(EncryptionKeyManager::new()),
        );
        let mut builder = ShardDataBuilder::new(shard.get_data());
        builder.set_mem_tbls(mem_tbls);
        shard.set_data(builder.build());
        let snap = shard.new_snap_access();
        let ranges = [(
            Bytes::from(kb.i_to_outer_key(0)),
            Bytes::from(kb.i_to_outer_key(1)),
        )];

        let (_, met_newer_ts) = snap.build_mem_data(&ranges, 50);
        assert!(met_newer_ts);

        let (_, met_newer_ts) = snap.build_mem_data(&ranges, 100);
        assert!(!met_newer_ts);
    }

    fn verify_mem_tables(
        mem_tbls: &[CfTable],
        ref_store: &RefStore,
        ranges: &[(InnerKey<'_>, InnerKey<'_>)],
    ) -> std::result::Result<(), TestCaseError> {
        let mem_iters = mem_tbls
            .iter()
            .map(|m| m.get_cf(WRITE_CF).new_iterator(false))
            .collect();
        let mut iter = table::new_merge_iterator(mem_iters, false);
        let mut count = 0;
        for &(start, end) in ranges {
            iter.seek(start);
            while iter.valid() && iter.key() < end {
                count += 1;
                let expect = ref_store.get_value(iter.key(), iter.value().version);
                prop_assert!(expect.is_some(), "key: {:?}", iter.key());
                let val = iter.value();
                prop_assert_eq!(val.get_value(), expect.unwrap().as_bytes());

                iter.next_all_version();
            }
        }
        prop_assert_eq!(count, ref_store.len());
        Ok(())
    }

    // Regression for shard.rs::from_change_set: when every per-file prefetch
    // fails with MemoryLimitExceeded, the typed variant must survive so
    // cop-worker returns HTTP 503 (retryable) instead of 500 (terminal).
    #[cfg(feature = "failpoints")]
    #[test]
    fn test_construct_snapshot_preserves_memory_limit_exceeded() {
        use tikv_util::defer;

        let runtime = tokio::runtime::Runtime::new().unwrap();
        let _enter = runtime.enter();

        let dfs: Arc<dyn crate::dfs::Dfs> = Arc::new(InMemFs::new());
        let txn_chunk_manager = TxnChunkManager::new(
            vec![],
            dfs.clone(),
            BlockCache::None,
            None,
            with_pool_size(2),
            TxnChunkManagerConfig::default(),
        );
        let master_key = MasterKey::new(&[1u8; 32]);
        let snap_ctx = SnapCtx {
            dfs,
            master_key,
            block_cache: BlockCache::None,
            vector_index_cache: None,
            columnar_file_cache: None,
            fts_cache: FtsCache::disabled(),
            fts_delta_cache: FtsDeltaCache::disabled(),
            schema_files: None,
            txn_chunk_manager,
            ia_ctx: IaCtx::disabled(&Default::default()),
            prepare_type: PrepareType::All,
            read_columnar: true,
            meta_file_cache: new_meta_file_cache(1024),
            columnar_meta_cache: ColumnarMetaCache::default(),
            encryption_key_manager: Arc::new(cloud_encryption::EncryptionKeyManager::new()),
            strict_file_memory_quota: Arc::new(tikv_util::memory::MemoryQuota::new(usize::MAX)),
            strict_file_memory_quota_wait_timeout: std::time::Duration::from_secs(2),
        };

        let mut snap_pb = kvenginepb::Snapshot::default();
        let mut l0 = kvenginepb::L0Create::default();
        l0.set_id(1);
        snap_pb.mut_l0_creates().push(l0);
        let mut cs = kvenginepb::ChangeSet::default();
        cs.set_shard_id(1);
        cs.set_shard_ver(1);
        cs.set_snapshot(snap_pb);
        let snap_bin = cs.write_to_bytes().unwrap();

        fail::cfg("prepare_file_memory_limit_exceeded", "return").unwrap();
        defer! { fail::remove("prepare_file_memory_limit_exceeded") }
        let result = block_on(SnapAccess::construct_snapshot(
            "test",
            &snap_ctx,
            &[],
            &snap_bin,
            MemoryLimiter::new(u64::MAX, None),
        ));

        match result {
            Ok(_) => panic!("construct_snapshot must fail when prepare_file fails"),
            Err(crate::Error::MemoryLimitExceeded(_)) => {}
            Err(err) => panic!(
                "expected Error::MemoryLimitExceeded so cop-worker returns 503; got {:?}",
                err,
            ),
        }
    }
}