// Copyright 2024 PingCAP, Inc.

use std::{
    cmp::{Ordering, min},
    collections::{BTreeMap, HashMap, HashSet},
    convert::TryFrom,
    mem,
    ops::Deref,
    sync::Arc,
    time::Duration,
};

use aligned_vec::{AVec, avec};
use api_version::ApiV2;
use arrow_buffer::i256;
use async_trait::async_trait;
use bytes::Buf;
use cloud_encryption::EncryptionKey;
use futures::future::try_join_all;
use tidb_query_datatype::{
    FieldTypeFlag, FieldTypeTp,
    codec::{
        Datum, datum,
        datum::{
            BYTES_FLAG, COMPACT_BYTES_FLAG, DECIMAL_FLAG, DURATION_FLAG, FLOAT_FLAG, INT_FLAG,
            JSON_FLAG, NIL_FLAG, UINT_FLAG, VAR_INT_FLAG, VAR_UINT_FLAG, VECTOR_FLOAT32_FLAG,
            decode,
        },
        mysql::{Decimal, DecimalDecoder, JsonEncoder, VectorFloat32Encoder},
        row::v2::{CODEC_VERSION, RowSlice, decode_v2_i64, decode_v2_u64},
        table::{
            PREFIX_LEN, decode_common_handle, decode_int_handle, encode_common_handle_row_key,
            encode_row_key,
        },
    },
};
use tikv_util::{
    codec::{
        BytesSlice,
        bytes::{decode_bytes, decode_compact_bytes},
        number::{decode_f64, decode_i64, decode_u64, decode_var_i64, decode_var_u64},
    },
    deadline::Deadline,
};
use tipb::ColumnInfo;

use super::filter::{FilterOpResult, FilterOpResults, FilterOperator};
use crate::{
    WRITE_CF,
    dfs::FileType,
    ia::{manager::IaManager, types::FileSegmentIdent},
    next_async, next_version_async,
    table::{
        self, InnerKey,
        blobtable::blobtable::BlobTable,
        columnar::{
            columnar::{
                Block, ColumnBuffer, ColumnMeta, ColumnarFile, TableMeta, decompress_pack,
                get_fixed_size,
            },
            get_primary_key,
        },
        file::File,
        schema_file::Schema,
        search,
    },
};

pub const GLOBAL_COMMON_HANDLE_END: &[u8] = &[255];
const PACK_CACHE_THRESHOLD: usize = 256 * 1024; // 256KB

#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
pub struct ColumnarRuntimeStats {
    pub mvcc_input_rows: u64,
    pub mvcc_input_bytes: u64,
    pub mvcc_output_rows: u64,
    pub read_block_ns: u64,
    pub serialize_ns: u64,
    pub init_reader_ns: u64,
    pub prefetch_ns: u64,
    pub rough_check_total_packs: u64,
    pub rough_check_selected_packs: u64,
    pub rough_check_skipped_packs: u64,
    pub rough_check_unknown_packs: u64,
    pub remote_segments: u64,
    pub total_segments: u64,
    /// Late packs actually loaded while gathering selected L2 rows.
    pub lm_late_packs_loaded: u64,
    /// Selected L2 rows gathered from late packs (excludes eager sidecar rows).
    pub lm_late_rows_gathered: u64,
}

impl ColumnarRuntimeStats {
    pub fn merge(&mut self, other: &Self) {
        self.mvcc_input_rows = self.mvcc_input_rows.saturating_add(other.mvcc_input_rows);
        self.mvcc_input_bytes = self.mvcc_input_bytes.saturating_add(other.mvcc_input_bytes);
        self.mvcc_output_rows = self.mvcc_output_rows.saturating_add(other.mvcc_output_rows);
        self.read_block_ns = self.read_block_ns.saturating_add(other.read_block_ns);
        self.serialize_ns = self.serialize_ns.saturating_add(other.serialize_ns);
        self.init_reader_ns = self.init_reader_ns.saturating_add(other.init_reader_ns);
        self.prefetch_ns = self.prefetch_ns.saturating_add(other.prefetch_ns);
        self.rough_check_total_packs = self
            .rough_check_total_packs
            .saturating_add(other.rough_check_total_packs);
        self.rough_check_selected_packs = self
            .rough_check_selected_packs
            .saturating_add(other.rough_check_selected_packs);
        self.rough_check_skipped_packs = self
            .rough_check_skipped_packs
            .saturating_add(other.rough_check_skipped_packs);
        self.rough_check_unknown_packs = self
            .rough_check_unknown_packs
            .saturating_add(other.rough_check_unknown_packs);
        self.remote_segments = self.remote_segments.saturating_add(other.remote_segments);
        self.total_segments = self.total_segments.saturating_add(other.total_segments);
        self.lm_late_packs_loaded = self
            .lm_late_packs_loaded
            .saturating_add(other.lm_late_packs_loaded);
        self.lm_late_rows_gathered = self
            .lm_late_rows_gathered
            .saturating_add(other.lm_late_rows_gathered);
    }

    fn clear_rough_check(&mut self) {
        self.rough_check_total_packs = 0;
        self.rough_check_selected_packs = 0;
        self.rough_check_skipped_packs = 0;
        self.rough_check_unknown_packs = 0;
    }
}

fn collect_rough_check_stats(results: &FilterOpResults) -> ColumnarRuntimeStats {
    let mut stats = ColumnarRuntimeStats {
        rough_check_total_packs: results.len() as u64,
        ..Default::default()
    };
    for result in results {
        match result {
            FilterOpResult::Some => {
                stats.rough_check_selected_packs =
                    stats.rough_check_selected_packs.saturating_add(1);
            }
            FilterOpResult::None => {
                stats.rough_check_skipped_packs = stats.rough_check_skipped_packs.saturating_add(1);
            }
            FilterOpResult::Unknown => {
                stats.rough_check_unknown_packs = stats.rough_check_unknown_packs.saturating_add(1);
            }
        }
    }
    stats
}

#[async_trait]
pub trait ColumnarReader: Send {
    fn schema(&self) -> &Schema;
    async fn seek(&mut self, handle: &[u8]) -> crate::table::Result<()>;
    async fn read(&mut self, block: &mut Block, limit: usize) -> crate::table::Result<usize>;
    fn reset(&mut self) -> crate::table::Result<()> {
        Ok(())
    }
    fn get_remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        Ok((vec![], 0))
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        ColumnarRuntimeStats::default()
    }

    fn take_runtime_stats(&mut self) -> ColumnarRuntimeStats {
        self.runtime_stats()
    }
}

pub(crate) struct ColumnarTableReader {
    table_meta: Arc<TableMeta>,
    schema: Schema,
    handle_reader: ColumnarColumnReader,
    version_reader: ColumnarColumnReader,
    columns_readers: Vec<ColumnarColumnReader>,
    runtime_stats: ColumnarRuntimeStats,
}

#[allow(dead_code)]
impl ColumnarTableReader {
    pub fn new(
        columnar_file: &ColumnarFile,
        schema: Schema,
        filter_op: Option<FilterOperator>,
        encryption_key: Option<EncryptionKey>,
    ) -> table::Result<ColumnarTableReader> {
        let table_meta = columnar_file.get_table(schema.table_id);
        debug_assert_eq!(table_meta.table_id, schema.table_id);
        let packs_filter = filter_op
            .as_ref()
            .map(|op| Arc::new(op.rough_check(&table_meta)));
        let runtime_stats = packs_filter
            .as_ref()
            .map(|results| collect_rough_check_stats(results.as_ref()))
            .unwrap_or_default();
        let file = columnar_file.get_file();
        let encryption_ver = columnar_file.get_encryption_ver();
        let encryption_fid = columnar_file.encryption_fid();
        let handle_reader = ColumnarColumnReader::new(
            file.clone(),
            table_meta.handle_column.clone(),
            false,
            encryption_key.clone(),
            encryption_ver,
            encryption_fid,
            packs_filter.clone(),
        )?;
        let version_reader = ColumnarColumnReader::new(
            file.clone(),
            table_meta.version_column.clone(),
            false,
            encryption_key.clone(),
            encryption_ver,
            encryption_fid,
            packs_filter.clone(),
        )?;
        let columns_readers = schema
            .columns
            .iter()
            .map(|col| {
                let col_id = col.get_column_id() as i32;
                let packs_filter = if get_fixed_size(col) == 0 {
                    None
                } else {
                    packs_filter.clone()
                };
                if let Some(col_meta) = table_meta.columns.get(&col_id) {
                    ColumnarColumnReader::new(
                        file.clone(),
                        col_meta.clone(),
                        false,
                        encryption_key.clone(),
                        encryption_ver,
                        encryption_fid,
                        packs_filter,
                    )
                } else {
                    let col_meta = ColumnMeta::new(col.clone(), false);
                    ColumnarColumnReader::new(
                        file.clone(),
                        Arc::new(col_meta),
                        true,
                        encryption_key.clone(),
                        encryption_ver,
                        encryption_fid,
                        packs_filter,
                    )
                }
            })
            .collect::<table::Result<Vec<_>>>()?;
        Ok(ColumnarTableReader {
            table_meta,
            schema,
            handle_reader,
            version_reader,
            columns_readers,
            runtime_stats,
        })
    }

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

    fn file_id(&self) -> u64 {
        self.handle_reader.pack_loader.file.id()
    }

    fn remaining_rows_in_current_pack(&self) -> usize {
        self.version_reader
            .pack_buffer
            .length()
            .saturating_sub(self.version_reader.row_idx_in_pack)
    }

    /// Reads normal early columns and returns a physical row reference for each
    /// appended row. The references are derived from the version reader, whose
    /// pack boundaries define the row namespace shared by every column.
    #[allow(dead_code)] // Consumed by the hybrid merge reader.
    pub(crate) async fn read_with_physical_rows(
        &mut self,
        block: &mut Block,
        physical_rows: &mut Vec<u32>,
        limit: usize,
    ) -> crate::table::Result<usize> {
        self.read_impl(block, Some(physical_rows), limit).await
    }

    async fn read_impl(
        &mut self,
        block: &mut Block,
        mut physical_rows: Option<&mut Vec<u32>>,
        limit: usize,
    ) -> crate::table::Result<usize> {
        // Read version column first to check if the pack is marked as skipped. If this
        // pack is marked as skipped, we should set_row_idx to other column readers,
        // including handle column if it is common handle.
        let mut total_read_row = 0;
        while total_read_row < limit {
            if !self.version_reader.valid() {
                break;
            }
            let physical_row_start =
                self.version_reader.pack_row_start + self.version_reader.row_idx_in_pack;
            // If the pack is skipped, the read will return ASAP, in this case we should
            // continue to read if total_read_row is less than limit.
            let (read_row, is_skipped, row_idx) = self
                .version_reader
                .read(&mut block.versions, limit - total_read_row)
                .await?;
            if is_skipped {
                let (handle_read_row, handle_is_skipped, ..) = self
                    .handle_reader
                    .read(&mut block.handles, read_row)
                    .await?;
                assert_eq!(
                    read_row,
                    handle_read_row,
                    "read_row: {}, handle_read_row: {}, version_pack_idx: {}, pack_idx: {}, version_num_packs: {}, num_packs: {}, handle_is_skipped: {}",
                    read_row,
                    handle_read_row,
                    self.version_reader.pack_idx(),
                    self.handle_reader.pack_idx(),
                    self.version_reader.num_packs(),
                    self.handle_reader.num_packs(),
                    handle_is_skipped
                );
                self.handle_reader.set_row_idx(row_idx).await?;
                for (i, col) in self.columns_readers.iter_mut().enumerate() {
                    let (col_read_row, col_is_skipped, ..) =
                        col.read(&mut block.columns[i], read_row).await?;
                    assert_eq!(
                        read_row,
                        col_read_row,
                        "read_row: {}, col_read_row: {}, col_pack_idx: {}, pack_idx: {}, col_num_packs: {}, num_packs: {}, col_is_skipped: {}",
                        read_row,
                        col_read_row,
                        col.pack_idx(),
                        self.version_reader.pack_idx(),
                        col.num_packs(),
                        self.version_reader.num_packs(),
                        col_is_skipped
                    );
                    col.set_row_idx(row_idx).await?;
                }
                if let Some(physical_rows) = physical_rows.as_deref_mut() {
                    Self::append_physical_rows(physical_rows, physical_row_start, read_row)?;
                }
                total_read_row += read_row;
            } else {
                let (handle_read_row, handle_is_skipped, _) = self
                    .handle_reader
                    .read(&mut block.handles, read_row)
                    .await?;
                assert_eq!(
                    read_row,
                    handle_read_row,
                    "read_row: {}, handle_read_row: {}, version_pack_idx: {}, pack_idx: {}, version_num_packs: {}, num_packs: {}, handle_is_skipped: {}",
                    read_row,
                    handle_read_row,
                    self.version_reader.pack_idx(),
                    self.handle_reader.pack_idx(),
                    self.version_reader.num_packs(),
                    self.handle_reader.num_packs(),
                    handle_is_skipped
                );
                assert!(!handle_is_skipped);
                for (i, col) in self.columns_readers.iter_mut().enumerate() {
                    let (col_read_row, col_is_skipped, _) =
                        col.read(&mut block.columns[i], read_row).await?;
                    assert_eq!(
                        read_row,
                        col_read_row,
                        "read_row: {}, col_read_row: {}, col_pack_idx: {}, pack_idx: {}, col_num_packs: {}, num_packs: {}, col_is_skipped: {}",
                        read_row,
                        col_read_row,
                        col.pack_idx(),
                        self.version_reader.pack_idx(),
                        col.num_packs(),
                        self.version_reader.num_packs(),
                        col_is_skipped
                    );
                    assert!(!col_is_skipped);
                }
                if let Some(physical_rows) = physical_rows.as_deref_mut() {
                    Self::append_physical_rows(physical_rows, physical_row_start, read_row)?;
                }
                total_read_row += read_row;
                return Ok(total_read_row);
            }
        }
        Ok(total_read_row)
    }

    fn append_physical_rows(
        physical_rows: &mut Vec<u32>,
        physical_row_start: usize,
        read_row: usize,
    ) -> crate::table::Result<()> {
        let start = u32::try_from(physical_row_start).map_err(|_| {
            table::Error::Other("columnar physical row position overflow".to_string())
        })?;
        let count = u32::try_from(read_row)
            .map_err(|_| table::Error::Other("columnar physical row count overflow".to_string()))?;
        let end = start.checked_add(count).ok_or_else(|| {
            table::Error::Other("columnar physical row position overflow".to_string())
        })?;
        physical_rows.extend(start..end);
        Ok(())
    }
}

#[async_trait]
impl ColumnarReader for ColumnarTableReader {
    fn schema(&self) -> &Schema {
        &self.schema
    }

    async fn seek(&mut self, handle: &[u8]) -> crate::table::Result<()> {
        let mut pack_idx = self.table_meta.handle_index.search_pack_idx(handle);
        while self.handle_reader.load_pack(pack_idx).await? {
            if pack_idx >= self.handle_reader.num_packs() {
                break;
            }
            // If the pack is skipped, try load next pack. Keep the row position align with
            // other column readers.
            pack_idx += 1;
        }
        let handle_buffer = &self.handle_reader.pack_buffer;
        let row_idx_in_pack = if handle.is_empty() {
            0
        } else if self.handle_reader.col_meta.fixed_size > 0 {
            let int_handle = (&handle[..]).get_i64_le();
            search(handle_buffer.length(), |i| {
                handle_buffer.get_int_handle_value(i) >= int_handle
            })
        } else {
            search(handle_buffer.length(), |i| {
                handle_buffer.get_not_null_value(i) >= handle
            })
        };
        self.handle_reader.row_idx_in_pack = row_idx_in_pack;
        let row_idx = self.handle_reader.pack_row_start + row_idx_in_pack;
        self.version_reader.set_row_idx(row_idx).await?;
        for col_reader in &mut self.columns_readers {
            col_reader.set_row_idx(row_idx).await?;
        }
        Ok(())
    }

    async fn read(&mut self, block: &mut Block, limit: usize) -> crate::table::Result<usize> {
        self.read_impl(block, None, limit).await
    }

    fn get_remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        let file = self.get_file();
        let mut ranges = vec![];
        // handle segment
        let (start_off, _) = self.handle_reader.col_meta.pack_offsets.get(0);
        ranges.push((start_off as u64, start_off as u64 + 1));
        // version segment
        let (start_off, _) = self.version_reader.col_meta.pack_offsets.get(0);
        ranges.push((start_off as u64, start_off as u64 + 1));
        // column segments
        for col in &self.columns_readers {
            let (start_off, _) = col.col_meta.pack_offsets.get(0);
            ranges.push((start_off as u64, start_off as u64 + 1));
        }
        file.get_remote_segments(&ranges)
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        self.runtime_stats
    }
}

pub(crate) struct ColumnarColumnReader {
    pack_loader: PackLoader,
    col_meta: Arc<ColumnMeta>,
    pack_buffer: ColumnBuffer,
    pack_idx: usize,
    pack_row_start: usize,
    pack_row_end: usize,
    row_idx_in_pack: usize,
    is_default_val: bool,
    default_val: Option<Vec<u8>>,
    decryption_buf: Vec<u8>,
    packs_filter: Option<Arc<FilterOpResults>>,
    tag: String,
    loaded_packs: u64,
}

impl ColumnarColumnReader {
    pub(crate) fn new(
        file: Arc<dyn File>,
        col_meta: Arc<ColumnMeta>,
        is_default_val: bool,
        encryption_key: Option<EncryptionKey>,
        encryption_ver: u32,
        encryption_fid: u64,
        packs_filter: Option<Arc<FilterOpResults>>,
    ) -> table::Result<ColumnarColumnReader> {
        let pack_loader = PackLoader::new(
            file,
            col_meta.clone(),
            encryption_key,
            encryption_ver,
            encryption_fid,
        );
        let pack_buffer = ColumnBuffer::new_from_col_info(&col_meta.col_info);
        let default_val = if is_default_val {
            parse_default_val(&col_meta.col_info)?
        } else {
            None
        };
        let tag = format!(
            "{}_{}",
            pack_loader.file.id(),
            col_meta.col_info.get_column_id(),
        );
        Ok(ColumnarColumnReader {
            pack_loader,
            col_meta,
            pack_buffer,
            pack_idx: 0,
            pack_row_start: 0,
            pack_row_end: 0,
            row_idx_in_pack: 0,
            is_default_val,
            default_val,
            decryption_buf: vec![],
            packs_filter,
            tag,
            loaded_packs: 0,
        })
    }

    pub(crate) async fn set_row_idx(&mut self, row_idx: usize) -> crate::table::Result<()> {
        if self.is_default_val {
            self.row_idx_in_pack = row_idx;
            return Ok(());
        }
        if self.pack_row_start <= row_idx && row_idx < self.pack_row_end {
            self.row_idx_in_pack = row_idx - self.pack_row_start;
            return Ok(());
        }
        let from_pack_idx = if self.pack_row_start < row_idx {
            self.pack_idx + 1
        } else {
            1
        };
        let pack_idx = self
            .col_meta
            .pack_offsets
            .search_pack_idx(from_pack_idx, row_idx as u32);
        self.load_pack(pack_idx).await?;
        if row_idx >= self.pack_row_start {
            self.row_idx_in_pack = row_idx - self.pack_row_start;
        }
        Ok(())
    }

    pub(crate) fn valid(&self) -> bool {
        self.pack_idx < self.col_meta.pack_offsets.num_packs()
    }

    pub(crate) fn pack_idx(&self) -> usize {
        self.pack_idx
    }

    pub(crate) fn num_packs(&self) -> usize {
        self.col_meta.pack_offsets.num_packs()
    }

    pub(crate) fn tag(&self) -> &str {
        &self.tag
    }

    pub(crate) async fn read(
        &mut self,
        to: &mut ColumnBuffer,
        limit: usize,
    ) -> crate::table::Result<(
        usize, // read_row
        bool,  // is_skipped
        usize, // row_idx
    )> {
        if self.is_default_val {
            let read_row = self.read_default(to, limit)?;
            return Ok((read_row, false, 0));
        }
        let mut read_row = 0;
        let mut first_row_idx = None;
        let num_packs = self.col_meta.pack_offsets.num_packs();
        while read_row < limit {
            let remain = self.pack_buffer.length() - self.row_idx_in_pack;
            if remain == 0 {
                let next_pack_idx = self.pack_idx + 1;
                if next_pack_idx >= num_packs {
                    break;
                }
                let is_skipped = self.load_pack(next_pack_idx).await?;
                self.row_idx_in_pack = 0;
                if is_skipped {
                    return Ok((read_row, true, self.pack_row_end));
                }
                continue;
            }
            let batch_size = min(remain, limit - read_row);
            if first_row_idx.is_none() {
                first_row_idx = Some(self.pack_row_start + self.row_idx_in_pack);
            }
            to.append(
                &self.pack_buffer,
                self.row_idx_in_pack,
                self.row_idx_in_pack + batch_size,
            );
            self.row_idx_in_pack += batch_size;
            read_row += batch_size;
        }
        Ok((
            read_row,
            false,
            first_row_idx.unwrap_or(self.pack_row_start + self.row_idx_in_pack),
        ))
    }

    fn read_default(&mut self, to: &mut ColumnBuffer, limit: usize) -> crate::table::Result<usize> {
        if let Some(val) = &self.default_val {
            for _ in 0..limit {
                to.push_value(val);
            }
        } else {
            for _ in 0..limit {
                to.push_null();
            }
        }
        Ok(limit)
    }

    async fn load_pack(&mut self, pack_idx: usize) -> crate::table::Result<bool /* is_skipped */> {
        if pack_idx == self.pack_idx && self.pack_buffer.length() > 0 {
            return Ok(false);
        }
        let num_packs = self.col_meta.pack_offsets.num_packs();
        if pack_idx >= num_packs {
            self.pack_idx = num_packs;
            self.pack_buffer.reset();
            let (_, row_end_off) = self.col_meta.pack_offsets.end_offset();
            self.pack_row_start = row_end_off as usize;
            self.pack_row_end = row_end_off as usize;
            self.row_idx_in_pack = 0;
            return Ok(false);
        }
        let ((pack_start, pack_row_start), (pack_end, pack_row_end)) =
            self.col_meta.get_pack_offset(pack_idx);
        if let Some(packs_filter) = &self.packs_filter {
            // If the pack is not support filter, result is Unknown, treat it as Some.
            if packs_filter[pack_idx] == FilterOpResult::None {
                debug!("{}: skip load_pack: {}", self.tag(), pack_idx);
                self.pack_idx = pack_idx;
                self.pack_buffer.reset();
                self.pack_row_start = pack_row_end as usize;
                self.pack_row_end = pack_row_end as usize;
                self.row_idx_in_pack = 0;
                return Ok(true);
            }
        }
        self.pack_loader
            .load_pack(
                &mut self.pack_buffer,
                pack_start,
                pack_end,
                &mut self.decryption_buf,
            )
            .await?;
        // Count only successful physical loads; cache hits and filtered packs
        // do not contribute to late-materialization read statistics.
        self.loaded_packs = self.loaded_packs.saturating_add(1);
        debug!("{}: load_pack: {}", self.tag(), pack_idx);
        self.pack_idx = pack_idx;
        self.pack_row_start = pack_row_start as usize;
        self.pack_row_end = pack_row_end as usize;
        Ok(false)
    }
}

/// One late value requested from a snapshot-pinned L2 file.
/// `selected_output_idx` is in merged/MVCC order, while `physical_row_idx` is
/// in the file's row space.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct LateReadSlot {
    pub physical_row_idx: u32,
    pub selected_output_idx: usize,
}

#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
pub struct LateColumnReadStats {
    pub packs_loaded: u64,
    pub rows_gathered: u64,
}

impl LateColumnReadStats {
    pub fn merge(&mut self, other: &Self) {
        self.packs_loaded = self.packs_loaded.saturating_add(other.packs_loaded);
        self.rows_gathered = self.rows_gathered.saturating_add(other.rows_gathered);
    }
}

/// Stable identity for an L2 row. `file_index` indexes the concat reader's
/// snapshot-pinned file list, not its mutable current cursor.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct L2PhysicalRowRef {
    pub file_index: u32,
    pub physical_row_idx: u32,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum MaterializationRef {
    Eager {
        eager_row_idx: u32,
    },
    L2Deferred {
        l2_reader_id: u32,
        file_index: u32,
        physical_row_idx: u32,
    },
}

/// The transient row space between global merge/MVCC and TiFlash's exact
/// predicate. `eager_late` contains only late columns, indexed by the stable
/// `Eager::eager_row_idx` values; source buffers are never referenced here.
pub struct HybridBatch {
    pub early: Block,
    pub refs: Vec<MaterializationRef>,
    pub eager_late: Vec<ColumnBuffer>,
}

/// Selection-dependent row mapping shared by every late output column in one
/// batch. Building it once avoids rescanning the full hybrid row space for
/// each late column.
#[derive(Clone, Debug)]
pub struct HybridMaterializationPlan {
    pub selected_rows: usize,
    pub eager_row_indexes: Vec<Option<u32>>,
    pub deferred: BTreeMap<(u32, u32), Vec<LateReadSlot>>,
}

impl HybridBatch {
    pub fn new(early: Block, eager_late: Vec<ColumnBuffer>) -> Self {
        Self {
            early,
            refs: vec![],
            eager_late,
        }
    }

    pub fn validate(&self) -> crate::table::Result<()> {
        if self.early.length() != self.refs.len() {
            return Err(table::Error::Other(format!(
                "hybrid early rows {} do not match materialization refs {}",
                self.early.length(),
                self.refs.len()
            )));
        }
        Ok(())
    }

    pub fn build_materialization_plan(
        &self,
        selection: &[u8],
    ) -> crate::table::Result<HybridMaterializationPlan> {
        self.validate()?;
        if selection.len() != self.refs.len() {
            return Err(table::Error::Other(
                "late materialization selection does not match hybrid row space".to_string(),
            ));
        }

        let selected_rows = selection.iter().filter(|&&selected| selected != 0).count();
        let mut eager_row_indexes = Vec::with_capacity(selected_rows);
        let mut deferred = BTreeMap::<(u32, u32), Vec<LateReadSlot>>::new();
        let mut selected_output_idx = 0;
        for (row_idx, materialization_ref) in self.refs.iter().enumerate() {
            let selected = selection[row_idx] != 0;
            match *materialization_ref {
                MaterializationRef::Eager { eager_row_idx } if selected => {
                    eager_row_indexes.push(Some(eager_row_idx));
                }
                MaterializationRef::L2Deferred {
                    l2_reader_id,
                    file_index,
                    physical_row_idx,
                } => {
                    if selected {
                        deferred
                            .entry((l2_reader_id, file_index))
                            .or_default()
                            .push(LateReadSlot {
                                physical_row_idx,
                                selected_output_idx,
                            });
                        eager_row_indexes.push(None);
                    }
                }
                MaterializationRef::Eager { .. } => {}
            }
            if selected {
                selected_output_idx += 1;
            }
        }
        debug_assert_eq!(selected_output_idx, selected_rows);
        Ok(HybridMaterializationPlan {
            selected_rows,
            eager_row_indexes,
            deferred,
        })
    }

    pub fn append_eager(
        &mut self,
        early: &Block,
        late: &[ColumnBuffer],
        start: usize,
        end: usize,
    ) -> crate::table::Result<()> {
        if start > end || end > early.length() || late.len() != self.eager_late.len() {
            return Err(table::Error::Other(
                "invalid eager hybrid append".to_string(),
            ));
        }
        if late.iter().any(|column| column.length() < end) {
            return Err(table::Error::Other(
                "eager late sidecar is shorter than its early block".to_string(),
            ));
        }
        let eager_row_start = self
            .eager_late
            .first()
            .map(ColumnBuffer::length)
            .unwrap_or_else(|| {
                self.refs
                    .iter()
                    .filter(|r| matches!(r, MaterializationRef::Eager { .. }))
                    .count()
            });
        self.early.append(early, start, end);
        for (target, source) in self.eager_late.iter_mut().zip(late) {
            target.append(source, start, end);
        }
        self.refs
            .extend((0..end - start).map(|offset| MaterializationRef::Eager {
                eager_row_idx: (eager_row_start + offset) as u32,
            }));
        self.validate()
    }

    pub fn append_deferred(
        &mut self,
        early: &Block,
        refs: &[MaterializationRef],
        start: usize,
        end: usize,
    ) -> crate::table::Result<()> {
        if start > end || end > early.length() || refs.len() < end {
            return Err(table::Error::Other(
                "invalid deferred hybrid append".to_string(),
            ));
        }
        if refs[start..end]
            .iter()
            .any(|reference| !matches!(reference, MaterializationRef::L2Deferred { .. }))
        {
            return Err(table::Error::Other(
                "deferred hybrid append contains an eager reference".to_string(),
            ));
        }
        self.early.append(early, start, end);
        self.refs.extend_from_slice(&refs[start..end]);
        self.validate()
    }

    /// Applies the MVCC-visible ranges without changing eager-sidecar indexes.
    /// The sidecar remains batch-owned until TiFlash provides exact selection.
    pub fn retain_ranges(&mut self, ranges: &[(usize, usize)]) -> crate::table::Result<()> {
        self.validate()?;
        let original_len = self.early.length();
        let mut keep = vec![false; original_len];
        for &(start, end) in ranges {
            if start > end || end > original_len {
                return Err(table::Error::Other("invalid hybrid MVCC range".to_string()));
            }
            keep[start..end].fill(true);
        }
        self.early.retain_rows(|row_idx| keep[row_idx]);
        self.refs = self
            .refs
            .iter()
            .enumerate()
            .filter_map(|(row_idx, materialization_ref)| {
                keep[row_idx].then_some(*materialization_ref)
            })
            .collect();
        self.validate()
    }

    /// Materializes one late output column after TiFlash has evaluated the
    /// exact predicate over `early`. `l2_files` is indexed by stable
    /// `l2_reader_id`, then by the concat reader's stable `file_index`.
    ///
    /// Returned stats cover L2 deferred gathers only (eager sidecar rows are
    /// not counted in `rows_gathered`).
    pub async fn materialize_selected_late_column(
        &self,
        plan: &HybridMaterializationPlan,
        late_column_idx: usize,
        late_column_id: i64,
        schema: &Schema,
        l2_files: &[Vec<ColumnarFile>],
        encryption_key: Option<EncryptionKey>,
    ) -> crate::table::Result<(ColumnBuffer, LateColumnReadStats)> {
        self.validate()?;
        let col_info = schema.find_column_by_id(late_column_id).ok_or_else(|| {
            table::Error::Other(format!(
                "late column {} is absent from the schema",
                late_column_id
            ))
        })?;
        let mut values = vec![None; plan.selected_rows];
        for (selected_output_idx, eager_row_idx) in plan.eager_row_indexes.iter().enumerate() {
            let Some(eager_row_idx) = eager_row_idx else {
                continue;
            };
            let eager_column = self
                .eager_late
                .get(late_column_idx)
                .ok_or_else(|| table::Error::Other("missing eager late column".to_string()))?;
            let eager_row_idx = *eager_row_idx as usize;
            if eager_row_idx >= eager_column.length() {
                return Err(table::Error::Other(
                    "eager late row is outside sidecar".to_string(),
                ));
            }
            let mut value = ColumnBuffer::new_from_col_info(col_info);
            value.append(eager_column, eager_row_idx, eager_row_idx + 1);
            values[selected_output_idx] = Some(value);
        }

        let mut late_stats = LateColumnReadStats::default();
        for ((l2_reader_id, file_index), selected_slots) in &plan.deferred {
            let files = l2_files.get(*l2_reader_id as usize).ok_or_else(|| {
                table::Error::Other("unknown L2 reader ID in materialization ref".to_string())
            })?;
            let file = files.get(*file_index as usize).ok_or_else(|| {
                table::Error::Other("unknown L2 file index in materialization ref".to_string())
            })?;
            let mut accessor =
                LateColumnAccessor::new(file, schema, late_column_id, encryption_key.clone())?;
            let file_stats = accessor
                .read_slots_into(selected_slots, &mut values)
                .await?;
            late_stats.merge(&file_stats);
        }

        let mut output = ColumnBuffer::new_from_col_info(col_info);
        for value in values {
            let value = value.ok_or_else(|| {
                table::Error::Other(
                    "late materialization did not produce every selected row".to_string(),
                )
            })?;
            output.append(&value, 0, 1);
        }
        Ok((output, late_stats))
    }
}

/// Reads selected rows from one column without using the cursor-oriented
/// `set_row_idx` API. A caller owns one accessor per snapshot-pinned
/// `(file, column)` and supplies physical row IDs from the L2 early reader.
pub struct LateColumnAccessor {
    reader: ColumnarColumnReader,
}

impl LateColumnAccessor {
    pub fn new(
        columnar_file: &ColumnarFile,
        schema: &Schema,
        col_id: i64,
        encryption_key: Option<EncryptionKey>,
    ) -> crate::table::Result<Self> {
        let table_meta = columnar_file.get_table(schema.table_id);
        let col_info = schema.find_column_by_id(col_id).ok_or_else(|| {
            table::Error::Other(format!("late column {} is absent from the schema", col_id))
        })?;
        let (col_meta, is_default_val) = if col_id == schema.handle_column.get_column_id() {
            (table_meta.handle_column.clone(), false)
        } else if col_id == schema.version_column.get_column_id() {
            (table_meta.version_column.clone(), false)
        } else if let Some(col_meta) = table_meta.columns.get(&(col_id as i32)) {
            (col_meta.clone(), false)
        } else {
            (Arc::new(ColumnMeta::new(col_info.clone(), false)), true)
        };
        let reader = ColumnarColumnReader::new(
            columnar_file.get_file(),
            col_meta,
            is_default_val,
            encryption_key,
            columnar_file.get_encryption_ver(),
            columnar_file.encryption_fid(),
            None,
        )?;
        Ok(Self { reader })
    }

    /// Materializes all requested rows in `selected_output_idx` order. Every
    /// complete touched pack is loaded once, and no untouched pack is loaded.
    pub async fn read_selected(
        &mut self,
        slots: &[LateReadSlot],
        to: &mut ColumnBuffer,
    ) -> crate::table::Result<LateColumnReadStats> {
        if slots.is_empty() {
            return Ok(LateColumnReadStats::default());
        }

        let mut values = vec![None; slots.len()];
        let stats = self.read_slots_into(slots, &mut values).await?;
        for value in values {
            let value = value.ok_or_else(|| {
                table::Error::Other("late read did not produce every selected row".to_string())
            })?;
            to.append(&value, 0, 1);
        }
        Ok(stats)
    }

    /// Fills globally indexed output fragments. This is used when one late
    /// column has selected rows from multiple snapshot-pinned L2 files.
    async fn read_slots_into(
        &mut self,
        slots: &[LateReadSlot],
        values: &mut [Option<ColumnBuffer>],
    ) -> crate::table::Result<LateColumnReadStats> {
        let mut seen_output_indexes = HashSet::<usize>::new();
        for slot in slots {
            if slot.selected_output_idx >= values.len() {
                return Err(table::Error::Other(
                    "late read output index is outside the selected row space".to_string(),
                ));
            }
            // Duplicate checks are local to one file. The caller owns global
            // duplicate detection across eager and deferred sources.
            if !seen_output_indexes.insert(slot.selected_output_idx) {
                return Err(table::Error::Other(
                    "late read slots must have unique contiguous output indexes".to_string(),
                ));
            }
        }

        let loaded_packs_before = self.reader.loaded_packs;
        let mut stats = LateColumnReadStats::default();
        if self.reader.is_default_val {
            for slot in slots {
                let mut value = ColumnBuffer::new_from_col_info(&self.reader.col_meta.col_info);
                self.reader.read_default(&mut value, 1)?;
                values[slot.selected_output_idx] = Some(value);
                stats.rows_gathered = stats.rows_gathered.saturating_add(1);
            }
        } else {
            let mut packs = BTreeMap::<usize, Vec<LateReadSlot>>::new();
            for &slot in slots {
                let pack_idx = self
                    .reader
                    .col_meta
                    .pack_offsets
                    .find_pack_idx(slot.physical_row_idx)
                    .ok_or_else(|| {
                        table::Error::Other(format!(
                            "physical row {} is outside late column {}",
                            slot.physical_row_idx,
                            self.reader.col_meta.col_info.get_column_id()
                        ))
                    })?;
                packs.entry(pack_idx).or_default().push(slot);
            }

            for (pack_idx, pack_slots) in packs {
                // The accessor has no rough-filter bitmap: each planned pack has
                // at least one selected row and must be read completely.
                if self.reader.load_pack(pack_idx).await? {
                    return Err(table::Error::Other(
                        "late column accessor unexpectedly skipped a requested pack".to_string(),
                    ));
                }
                for slot in pack_slots {
                    let row_idx = slot.physical_row_idx as usize;
                    let pack_row_idx =
                        row_idx
                            .checked_sub(self.reader.pack_row_start)
                            .ok_or_else(|| {
                                table::Error::Other("late read row precedes its pack".to_string())
                            })?;
                    if row_idx >= self.reader.pack_row_end
                        || pack_row_idx >= self.reader.pack_buffer.length()
                    {
                        return Err(table::Error::Other(
                            "late read row is outside its pack".to_string(),
                        ));
                    }
                    let mut value = ColumnBuffer::new_from_col_info(&self.reader.col_meta.col_info);
                    value.append(&self.reader.pack_buffer, pack_row_idx, pack_row_idx + 1);
                    if values[slot.selected_output_idx].is_some() {
                        return Err(table::Error::Other(
                            "duplicate late materialization output row".to_string(),
                        ));
                    }
                    values[slot.selected_output_idx] = Some(value);
                    stats.rows_gathered = stats.rows_gathered.saturating_add(1);
                }
            }
        }

        stats.packs_loaded = self.reader.loaded_packs.saturating_sub(loaded_packs_before);

        Ok(stats)
    }
}

struct PackLoader {
    file: Arc<dyn File>,
    col_meta: Arc<ColumnMeta>,
    segment_ident: FileSegmentIdent,
    segment_content: Vec<u8>,
    segment_content_offset: u32,
    compressed_buf: Vec<u8>,
    uncompressed_buf: AVec<u8>,
    encryption_key: Option<EncryptionKey>,
    encryption_ver: u32,
    encryption_fid: u64,
}

impl PackLoader {
    pub fn new(
        file: Arc<dyn File>,
        col_meta: Arc<ColumnMeta>,
        encryption_key: Option<EncryptionKey>,
        encryption_ver: u32,
        encryption_fid: u64,
    ) -> PackLoader {
        PackLoader {
            file,
            col_meta,
            segment_ident: FileSegmentIdent::new(0, 0, 0),
            segment_content: vec![],
            segment_content_offset: 0,
            encryption_key,
            encryption_ver,
            encryption_fid,
            compressed_buf: vec![],
            uncompressed_buf: avec![],
        }
    }

    async fn read_from_segment_cache(
        &mut self,
        buf: Option<&mut [u8]>, // None means read to compressed_buf
        pack_offset: u32,
    ) -> crate::table::Result<()> {
        let segment_ident = self.file.get_segment_ident(pack_offset as u64)?;
        let data_len = if let Some(buf) = &buf {
            buf.len()
        } else {
            self.compressed_buf.len()
        };

        // For large packs (>256KB), don't use cache, read directly
        if data_len > PACK_CACHE_THRESHOLD {
            // Read directly from file for large packs
            if let Some(buf) = buf {
                self.file.read_at_async(buf, pack_offset as u64).await?;
            } else {
                self.compressed_buf.resize(data_len, 0);
                self.file
                    .read_at_async(&mut self.compressed_buf, pack_offset as u64)
                    .await?;
            }
            return Ok(());
        }

        // For smaller packs, check if it's in our current cache
        let is_in_current_segment = segment_ident == self.segment_ident
            && pack_offset >= self.segment_content_offset
            && (pack_offset as usize + data_len)
                <= (self.segment_content_offset as usize + self.segment_content.len());

        if !is_in_current_segment {
            let (pack_end_offset, _) = self.col_meta.pack_offsets.end_offset();
            let read_size = std::cmp::min(
                PACK_CACHE_THRESHOLD,
                (pack_end_offset - pack_offset) as usize,
            );
            self.segment_content.resize(read_size, 0);
            self.file
                .read_at_async(&mut self.segment_content, pack_offset as u64)
                .await?;
            self.segment_content_offset = pack_offset;
            self.segment_ident = segment_ident;
        }

        let start_off = pack_offset - self.segment_content_offset;
        // Copy content from segment_content to buf
        if let Some(buf) = buf {
            buf.copy_from_slice(
                &self.segment_content[start_off as usize..start_off as usize + data_len],
            );
        } else {
            self.compressed_buf.copy_from_slice(
                &self.segment_content[start_off as usize..start_off as usize + data_len],
            );
        }

        Ok(())
    }

    pub async fn load_pack(
        &mut self,
        col_buf: &mut ColumnBuffer,
        pack_offset: u32,
        pack_end_offset: u32,
        decryption_buf: &mut Vec<u8>,
    ) -> crate::table::Result<()> {
        let length = (pack_end_offset - pack_offset) as usize;
        if let Some(encryption_key) = self.encryption_key.clone() {
            decryption_buf.resize(length, 0);
            if self.file.is_sync() {
                // InMemFile should always be sync.
                self.file.read_at(decryption_buf, pack_offset as u64)?;
            } else {
                // Try to read from segment cache, if not found, load it.
                self.read_from_segment_cache(Some(decryption_buf), pack_offset)
                    .await?;
            }
            self.compressed_buf.clear();
            let encryption_key = encryption_key
                .switch_if_header_mismatch(self.encryption_ver)
                .map_err(|_| {
                    crate::table::Error::EncryptionKeyUnavailable(
                        "encryption key is not ready".to_owned(),
                    )
                })?;
            encryption_key.decrypt(
                decryption_buf,
                self.encryption_fid,
                pack_offset,
                &mut self.compressed_buf,
            );
        } else {
            self.compressed_buf.resize(length, 0);
            if self.file.is_sync() {
                // InMemFile should always be sync.
                self.file
                    .read_at(&mut self.compressed_buf, pack_offset as u64)?;
            } else {
                // Try to read from segment cache, if not found, load it.
                self.read_from_segment_cache(None, pack_offset).await?;
            }
        }
        decompress_pack(&self.compressed_buf, &mut self.uncompressed_buf);
        col_buf.parse(&self.uncompressed_buf);
        Ok(())
    }
}

#[async_trait]
pub trait ColumnarFilterReader: Send {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()>;
    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()>;
    fn get_schema(&self) -> &Schema;

    fn reset(&mut self) {}

    fn get_remote_segment_stats(&self) -> crate::table::Result<(usize, usize)> {
        Ok((0, 0))
    }

    async fn prefetch_ia_remote_segments(
        &mut self,
        _tag: &str,
        _ia_mgr: &IaManager,
        _keyspace_id: u32,
        _timeout: Duration,
    ) -> crate::table::Result<Option<f64>> {
        Ok(None)
    }

    // Try read block, return the number of rows read and whether the reader is
    // drained.
    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize /* read_row */, bool /* drained */)>;

    // Return size may be less than limit, and it doesn't mean the reader is drained
    // if the return size above 0.
    async fn read_block(&mut self, block: &mut Block, limit: usize) -> crate::table::Result<usize> {
        loop {
            let (read_row, drained) = self.try_read_block(block, limit).await?;
            if read_row > 0 {
                return Ok(read_row);
            }
            if drained {
                return Ok(0);
            }
            // If all rows are filtered, we should try to read next block.
            continue;
        }
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        ColumnarRuntimeStats::default()
    }

    fn take_runtime_stats(&mut self) -> ColumnarRuntimeStats {
        self.runtime_stats()
    }

    #[cfg(any(test, feature = "testexport"))]
    async fn read_all(&mut self) -> Block {
        let schema = self.get_schema().clone();
        let mut out = Block::new(&schema);
        let mut buf = Block::new(&schema);
        loop {
            buf.reset();
            let read = self.read_block(&mut buf, 1024).await.unwrap();
            if read == 0 {
                break;
            }
            out.append(&buf, 0, read);
        }
        out
    }

    async fn set_unbounded_handle_range(&mut self) -> crate::table::Result<()> {
        if self.get_schema().is_common_handle() {
            self.set_handle_range(&[], GLOBAL_COMMON_HANDLE_END).await?;
        } else {
            self.set_int_handle_range(i64::MIN, None).await?;
        }
        Ok(())
    }
}

struct ColumnarFilter {
    ranges: Vec<(usize, usize)>,
    in_range: bool,
    range_start: usize,
    filter_block: Block,
}

impl ColumnarFilter {
    fn new(schema: &Schema) -> Self {
        Self {
            ranges: vec![],
            in_range: false,
            range_start: 0,
            filter_block: Block::new(schema),
        }
    }

    fn finish_range(&mut self, i: usize) {
        if self.in_range {
            self.ranges.push((self.range_start, i));
            self.in_range = false;
        }
    }

    fn start_range(&mut self, i: usize) {
        if !self.in_range {
            self.range_start = i;
            self.in_range = true;
        }
    }

    fn clear(&mut self) {
        self.ranges.clear();
        self.in_range = false;
    }

    fn do_filter(&mut self, read_row: usize, block: &mut Block) -> usize {
        if self.ranges.len() == 1 {
            let (start, end) = self.ranges[0];
            if start == 0 {
                return if end == read_row {
                    // All rows are valid, no need to filter.
                    read_row
                } else {
                    block.truncate(end);
                    end
                };
            }
        }
        self.filter_block.reset();
        let mut filtered_rows = 0;
        for &(start, end) in &self.ranges {
            self.filter_block.handles.append(&block.handles, start, end);
            self.filter_block
                .versions
                .append(&block.versions, start, end);
            for (i, col) in self.filter_block.columns.iter_mut().enumerate() {
                col.append(&block.columns[i], start, end);
            }
            filtered_rows += end - start;
        }
        mem::swap(block, &mut self.filter_block);
        filtered_rows
    }
}

struct ColumnarMvccState {
    read_ts: u64,
    filter: ColumnarFilter,
    end_handle: Vec<u8>,
    end_int_handle: Option<i64>,
    prev_int_handle: Option<i64>,
    prev_common_handle: Vec<u8>,
    range_exhausted: bool,
    runtime_stats: ColumnarRuntimeStats,
}

pub struct ColumnarMvccReader {
    src: Box<dyn ColumnarReader>,
    state: ColumnarMvccState,
}

async fn prefetch_ia_remote_segments_by_idents(
    tag: &str,
    ia_mgr: &IaManager,
    keyspace_id: u32,
    timeout: Duration,
    idents: Vec<FileSegmentIdent>,
    total_segments: usize,
) -> crate::table::Result<Option<f64>> {
    if idents.is_empty() || total_segments == 0 {
        // No need to prefetch, return 1.0 cache hit.
        return Ok(Some(1.0));
    }

    let _enter = ia_mgr.enter_runtime();
    let mut tasks = vec![];
    let deadline = Deadline::from_now(timeout);
    let prefetch_segments = idents.len();
    for ident in idents {
        let mgr = ia_mgr.clone();
        let task = async move {
            mgr.prefetch_segment(ident, FileType::Columnar, keyspace_id, deadline, None, true)
                .await
        };
        tasks.push(tokio::spawn(task));
    }
    drop(_enter);

    match tokio::time::timeout_at(deadline.to_tokio_instant(), try_join_all(tasks)).await {
        Ok(res) => {
            let res = res.map_err(|err| -> crate::table::Error {
                crate::table::Error::Other(format!("{} prefetch segment failed: {:?}", tag, err))
            })?;
            res.into_iter().collect::<crate::table::Result<()>>()?;
            let cache_hit =
                total_segments.saturating_sub(prefetch_segments) as f64 / total_segments as f64;
            info!("{} prefetch ia remote segments", tag;
                "prefetch_segments" => prefetch_segments, "cache_hit" => cache_hit);
            Ok(Some(cache_hit))
        }
        Err(_) => Err(crate::table::Error::DeadlineExceeded(format!(
            "{} prefetch ia remote segments timeout",
            tag
        ))),
    }
}

impl ColumnarMvccState {
    fn new(schema: &Schema, read_ts: u64) -> Self {
        Self {
            filter: ColumnarFilter::new(schema),
            read_ts,
            end_handle: vec![],
            end_int_handle: None,
            prev_int_handle: None,
            prev_common_handle: vec![],
            range_exhausted: false,
            runtime_stats: ColumnarRuntimeStats::default(),
        }
    }

    /// Computes the visible ranges in the merged row space. Keep this separate
    /// from `ColumnarFilter::do_filter`: hybrid batches must apply the exact
    /// same ranges to early rows, deferred references and eager sidecars.
    fn compute_visible_ranges(&mut self, block: &Block) {
        self.filter.clear();
        if block.handles.fixed_size > 0 {
            for i in 0..block.handles.length() {
                let handle = block.handles.get_int_handle_value(i);
                let version = block.versions.get_version(i);
                if version > self.read_ts
                    || (self.prev_int_handle.is_some() && handle == self.prev_int_handle.unwrap())
                {
                    self.filter.finish_range(i);
                    continue;
                }
                if block.versions.is_null(i) {
                    self.prev_int_handle = Some(handle);
                    self.filter.finish_range(i);
                    continue;
                }
                if self.end_int_handle.is_some() && handle >= self.end_int_handle.unwrap() {
                    self.range_exhausted = true;
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
                self.prev_int_handle = Some(handle);
            }
        } else {
            let length = block.handles.length();
            let last_handle = block.handles.get_not_null_value(length - 1);
            let check_handle = last_handle >= self.end_handle.as_slice();
            for i in 0..block.handles.length() {
                let handle = block.handles.get_not_null_value(i);
                let version = block.versions.get_version(i);
                if version > self.read_ts || handle == self.prev_common_handle {
                    self.filter.finish_range(i);
                    continue;
                }
                if block.versions.is_null(i) {
                    self.prev_common_handle = handle.to_vec();
                    self.filter.finish_range(i);
                    continue;
                }
                if check_handle && handle >= self.end_handle.as_slice() {
                    self.range_exhausted = true;
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
                self.prev_common_handle = handle.to_vec();
            }
        }
        self.filter.finish_range(block.handles.length());
    }

    /// Applies the same visibility decision used by the normal full-block path
    /// to an LM batch. Callers must supply rows already ordered by the existing
    /// global merge reader.
    fn apply_hybrid_visibility(&mut self, batch: &mut HybridBatch) -> crate::table::Result<()> {
        batch.validate()?;
        self.compute_visible_ranges(&batch.early);
        batch.retain_ranges(&self.filter.ranges)
    }
}

impl ColumnarMvccReader {
    pub fn new(src: Box<dyn ColumnarReader>, schema: &Schema, read_ts: u64) -> ColumnarMvccReader {
        ColumnarMvccReader {
            src,
            state: ColumnarMvccState::new(schema, read_ts),
        }
    }

    #[allow(dead_code)] // Consumed by the hybrid reader.
    pub(crate) fn apply_hybrid_visibility(
        &mut self,
        batch: &mut HybridBatch,
    ) -> crate::table::Result<()> {
        self.state.apply_hybrid_visibility(batch)
    }
}

#[async_trait]
impl ColumnarFilterReader for ColumnarMvccReader {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        self.state.end_handle = end_handle.to_vec();
        self.src.seek(start_handle).await?;
        self.state.filter.clear();
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
        Ok(())
    }

    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        self.state.end_int_handle = end_handle;
        self.src.seek(&start_handle.to_le_bytes()).await?;
        self.state.filter.clear();
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
        Ok(())
    }

    fn get_schema(&self) -> &Schema {
        self.src.schema()
    }

    fn reset(&mut self) {
        self.src.reset().unwrap();
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
    }

    fn get_remote_segment_stats(&self) -> crate::table::Result<(usize, usize)> {
        let (idents, total_segments) = self.src.get_remote_segments()?;
        Ok((idents.len(), total_segments))
    }

    async fn prefetch_ia_remote_segments(
        &mut self,
        tag: &str,
        ia_mgr: &IaManager,
        keyspace_id: u32,
        timeout: Duration,
    ) -> crate::table::Result<Option<f64>> {
        let (idents, total_segments) = self.src.get_remote_segments()?;
        prefetch_ia_remote_segments_by_idents(
            tag,
            ia_mgr,
            keyspace_id,
            timeout,
            idents,
            total_segments,
        )
        .await
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize /* read_row */, bool /* drained */)> {
        block.reset();
        if self.state.range_exhausted {
            return Ok((0, true));
        }
        let read_row = self.src.read(block, limit).await?;
        if read_row == 0 {
            // Return 0 rows means the reader is drained.
            return Ok((0, true));
        }
        self.state.runtime_stats.mvcc_input_rows = self
            .state
            .runtime_stats
            .mvcc_input_rows
            .saturating_add(read_row as u64);
        self.state.runtime_stats.mvcc_input_bytes = self
            .state
            .runtime_stats
            .mvcc_input_bytes
            .saturating_add(block.total_byte_size());
        self.state.compute_visible_ranges(block);
        let filtered_rows = self.state.filter.do_filter(read_row, block);
        self.state.runtime_stats.mvcc_output_rows = self
            .state
            .runtime_stats
            .mvcc_output_rows
            .saturating_add(filtered_rows as u64);
        Ok((filtered_rows, false))
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.state.runtime_stats;
        let inner_stats = self.src.runtime_stats();
        stats.merge(&inner_stats);
        stats
    }
}

pub struct ColumnarCompactReader {
    src: Box<dyn ColumnarReader>,
    level: u32,
    safe_ts: u64,
    filter: ColumnarFilter,
    end_handle: Vec<u8>,
    end_int_handle: Option<i64>,
    prev_int_handle: Option<i64>,
    prev_common_handle: Vec<u8>,
    reached_safe_ts_boundary: bool,
}

impl ColumnarCompactReader {
    pub fn new(
        src: Box<dyn ColumnarReader>,
        level: u32,
        schema: &Schema,
        safe_ts: u64,
    ) -> ColumnarCompactReader {
        ColumnarCompactReader {
            src,
            level,
            filter: ColumnarFilter::new(schema),
            safe_ts,
            end_handle: vec![],
            end_int_handle: None,
            prev_int_handle: None,
            prev_common_handle: vec![],
            reached_safe_ts_boundary: false,
        }
    }
}

#[async_trait]
impl ColumnarFilterReader for ColumnarCompactReader {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        self.end_handle = end_handle.to_vec();
        self.src.seek(start_handle).await?;
        self.filter.clear();
        Ok(())
    }

    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        self.end_int_handle = end_handle;
        self.src.seek(&start_handle.to_le_bytes()).await?;
        self.filter.clear();
        Ok(())
    }

    fn get_schema(&self) -> &Schema {
        self.src.schema()
    }

    fn reset(&mut self) {
        self.end_handle.clear();
        self.end_int_handle = None;
        self.prev_int_handle = None;
        self.prev_common_handle.clear();
        self.reached_safe_ts_boundary = false;
        self.filter.clear();
        self.src.reset().unwrap();
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize /* read_row */, bool /* drained */)> {
        self.filter.clear();
        block.reset();
        let read_row = self.src.read(block, limit).await?;
        if read_row == 0 {
            return Ok((0, true));
        }
        if block.handles.fixed_size > 0 {
            for i in 0..block.handles.length() {
                let handle = block.handles.get_int_handle_value(i);
                let version = block.versions.get_version(i);
                let is_same_handle =
                    self.prev_int_handle.is_some() && handle == self.prev_int_handle.unwrap();
                if !is_same_handle {
                    self.reached_safe_ts_boundary = false;
                } else if self.reached_safe_ts_boundary {
                    self.filter.finish_range(i);
                    continue;
                }
                if self.end_int_handle.is_some() && handle >= self.end_int_handle.unwrap() {
                    self.filter.finish_range(i);
                    break;
                }
                if version <= self.safe_ts {
                    self.reached_safe_ts_boundary = true;
                    if self.level == 2 && block.versions.is_null(i) {
                        // For bottommost level compaction, a tombstone <= safe_ts can be dropped.
                        self.prev_int_handle = Some(handle);
                        self.filter.finish_range(i);
                        continue;
                    }
                }
                self.filter.start_range(i);
                self.prev_int_handle = Some(handle);
            }
        } else {
            let length = block.handles.length();
            let last_handle = block.handles.get_not_null_value(length - 1);
            let check_handle = last_handle >= self.end_handle.as_slice();
            for i in 0..block.handles.length() {
                let handle = block.handles.get_not_null_value(i);
                let version = block.versions.get_version(i);
                let is_same_handle = handle == self.prev_common_handle;
                if !is_same_handle {
                    self.reached_safe_ts_boundary = false;
                } else if self.reached_safe_ts_boundary {
                    self.filter.finish_range(i);
                    continue;
                }
                if check_handle && handle >= self.end_handle.as_slice() {
                    self.filter.finish_range(i);
                    break;
                }
                if version <= self.safe_ts {
                    self.reached_safe_ts_boundary = true;
                    if self.level == 2 && block.versions.is_null(i) {
                        // For bottommost level compaction, a tombstone <= safe_ts can be dropped.
                        self.prev_common_handle = handle.to_vec();
                        self.filter.finish_range(i);
                        continue;
                    }
                }
                self.filter.start_range(i);
                self.prev_common_handle = handle.to_vec();
            }
        }
        self.filter.finish_range(read_row);
        Ok((self.filter.do_filter(read_row, block), false))
    }
}

pub struct ColumnarTruncateTsReader {
    src: Box<dyn ColumnarReader>,
    truncate_ts: u64,
    filter: ColumnarFilter,
    end_handle: Vec<u8>,
    end_int_handle: Option<i64>,
}

impl ColumnarTruncateTsReader {
    pub fn new(
        src: Box<dyn ColumnarReader>,
        schema: &Schema,
        truncate_ts: u64,
    ) -> ColumnarTruncateTsReader {
        ColumnarTruncateTsReader {
            src,
            filter: ColumnarFilter::new(schema),
            truncate_ts,
            end_handle: vec![],
            end_int_handle: None,
        }
    }
}

#[async_trait]
impl ColumnarFilterReader for ColumnarTruncateTsReader {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        self.end_handle = end_handle.to_vec();
        self.src.seek(start_handle).await?;
        self.filter.clear();
        Ok(())
    }

    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        self.end_int_handle = end_handle;
        self.src.seek(&start_handle.to_le_bytes()).await?;
        self.filter.clear();
        Ok(())
    }

    fn get_schema(&self) -> &Schema {
        self.src.schema()
    }

    fn reset(&mut self) {
        self.end_handle.clear();
        self.end_int_handle = None;
        self.filter.clear();
        self.src.reset().unwrap();
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize /* read_row */, bool /* drained */)> {
        self.filter.clear();
        block.reset();
        let read_row = self.src.read(block, limit).await?;
        if read_row == 0 {
            return Ok((0, true));
        }
        if block.handles.fixed_size > 0 {
            for i in 0..block.handles.length() {
                let handle = block.handles.get_int_handle_value(i);
                let version = block.versions.get_version(i);
                if version > self.truncate_ts {
                    self.filter.finish_range(i);
                    continue;
                }
                if self.end_int_handle.is_some() && handle >= self.end_int_handle.unwrap() {
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
            }
        } else {
            let length = block.handles.length();
            let last_handle = block.handles.get_not_null_value(length - 1);
            let check_handle = last_handle >= self.end_handle.as_slice();
            for i in 0..block.handles.length() {
                let handle = block.handles.get_not_null_value(i);
                let version = block.versions.get_version(i);
                if version > self.truncate_ts {
                    self.filter.finish_range(i);
                    continue;
                }
                if check_handle && handle >= self.end_handle.as_slice() {
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
            }
        }
        self.filter.finish_range(read_row);
        Ok((self.filter.do_filter(read_row, block), false))
    }
}

pub(crate) struct ColumnarMergeReader {
    schema: Schema,
    #[allow(clippy::vec_box)]
    heap: Vec<Box<ColumnarReaderBuffer>>,
    // Save the invalid reader buffer to recover them when reset.
    #[allow(clippy::vec_box)]
    origin_heap: Vec<Box<ColumnarReaderBuffer>>,
    is_int_handle: bool,
    first_batch_end_row_idx: usize,
}

pub(crate) struct ColumnarReaderBuffer {
    reader: Box<dyn ColumnarReader>,
    block: Block,
    row_idx: usize,
}

impl ColumnarReaderBuffer {
    pub async fn seek(&mut self, handle: &[u8]) -> crate::table::Result<()> {
        self.reader.seek(handle).await?;
        self.block.reset();
        self.reader.read(&mut self.block, 1).await?;
        self.row_idx = 0;
        Ok(())
    }

    pub fn handle(&self) -> &[u8] {
        self.block.handles.get_not_null_value(self.row_idx)
    }

    pub fn int_handle(&self) -> i64 {
        self.block.handles.get_int_handle_value(self.row_idx)
    }

    pub fn inner_reader(&self) -> &dyn ColumnarReader {
        self.reader.as_ref()
    }

    pub fn reset(&mut self) {
        self.row_idx = 0;
        self.block.reset();
        self.reader.reset().unwrap();
    }

    pub fn version(&self) -> u64 {
        self.block.versions.get_version(self.row_idx)
    }

    pub fn valid(&self) -> bool {
        self.row_idx < self.block.handles.length()
    }

    pub async fn read_block(&mut self) -> crate::table::Result<()> {
        self.block.reset();
        self.reader.read(&mut self.block, 1024).await?;
        self.row_idx = 0;
        Ok(())
    }
}

#[allow(dead_code)]
impl ColumnarMergeReader {
    pub fn new(schema: Schema, readers: Vec<Box<dyn ColumnarReader>>) -> ColumnarMergeReader {
        let heap: Vec<Box<ColumnarReaderBuffer>> = readers
            .into_iter()
            .map(|reader| {
                Box::new(ColumnarReaderBuffer {
                    reader,
                    block: Block::new(&schema),
                    row_idx: 0,
                })
            })
            .collect();
        let is_int_handle = !schema.is_common_handle();
        ColumnarMergeReader {
            schema,
            heap,
            origin_heap: vec![],
            is_int_handle,
            first_batch_end_row_idx: 0,
        }
    }

    fn init_heap(&mut self) {
        for i in (0..self.heap.len() / 2).rev() {
            self.down(i);
        }
        if !self.heap.is_empty() {
            self.update_first_block_end_row_idx();
        }
    }

    fn down(&mut self, i0: usize) -> bool {
        let n = self.heap.len();
        let mut i = i0;
        loop {
            let left = 2 * i + 1;
            if left >= n {
                break;
            }
            let right = left + 1;
            let j = if right < n && self.less(right, left) {
                right
            } else {
                left
            };
            if !self.less(j, i) {
                break;
            }
            self.heap.swap(i, j);
            i = j;
        }
        i > i0
    }

    fn less(&mut self, a: usize, b: usize) -> bool {
        if self.is_int_handle {
            let a_handle = self.heap[a].int_handle();
            let b_handle = self.heap[b].int_handle();
            if a_handle != b_handle {
                a_handle < b_handle
            } else {
                self.heap[a].version() > self.heap[b].version()
            }
        } else {
            match self.heap[a].handle().cmp(self.heap[b].handle()) {
                Ordering::Less => true,
                Ordering::Equal => self.heap[a].version() > self.heap[b].version(),
                Ordering::Greater => false,
            }
        }
    }

    fn update_first_block_end_row_idx(&mut self) {
        let first = &self.heap[0];
        if self.heap.len() == 1 {
            self.first_batch_end_row_idx = first.block.handles.length();
            return;
        }
        let mut second_handle = if self.heap.len() >= 3 {
            if first.block.handles.fixed_size > 0 {
                if self.heap[1].int_handle() < self.heap[2].int_handle() {
                    self.heap[1].handle()
                } else {
                    self.heap[2].handle()
                }
            } else if self.heap[1].handle() < self.heap[2].handle() {
                self.heap[1].handle()
            } else {
                self.heap[2].handle()
            }
        } else {
            self.heap[1].handle()
        };
        if first.block.handles.fixed_size > 0 {
            let int_second_handle = second_handle.get_i64_le();
            for i in first.row_idx + 1..first.block.handles.length() {
                let handle = first.block.handles.get_int_handle_value(i);
                if handle >= int_second_handle {
                    self.first_batch_end_row_idx = i;
                    return;
                }
            }
        } else {
            for i in first.row_idx + 1..first.block.handles.length() {
                if first.block.handles.get_not_null_value(i) >= second_handle {
                    self.first_batch_end_row_idx = i;
                    return;
                }
            }
        }
        self.first_batch_end_row_idx = first.block.handles.length();
    }
}

#[async_trait]
impl ColumnarReader for ColumnarMergeReader {
    fn schema(&self) -> &Schema {
        &self.schema
    }

    fn reset(&mut self) -> crate::table::Result<()> {
        // Recover all readers from origin_heap and push them into heap.
        while let Some(reader) = self.origin_heap.pop() {
            self.heap.push(reader);
        }
        self.heap.iter_mut().for_each(|r| r.reset());
        Ok(())
    }

    async fn seek(&mut self, handle: &[u8]) -> crate::table::Result<()> {
        for reader in &mut self.heap {
            reader.seek(handle).await?;
        }
        // Move invalid readers to origin_heap, keep valid readers in heap.
        let (invalid, valid) = self.heap.drain(..).partition(|r| !r.valid());
        self.origin_heap = invalid;
        self.heap = valid;
        if !self.heap.is_empty() {
            self.init_heap();
        }
        Ok(())
    }

    async fn read(&mut self, block: &mut Block, limit: usize) -> crate::table::Result<usize> {
        let mut read_row = 0;
        while read_row < limit {
            if self.heap.is_empty() {
                return Ok(read_row);
            }
            let first = &mut self.heap[0];
            let remain = min(
                limit - read_row,
                self.first_batch_end_row_idx - first.row_idx,
            );

            block.append(&first.block, first.row_idx, first.row_idx + remain);
            first.row_idx += remain;
            read_row += remain;
            if first.row_idx == first.block.handles.length() {
                first.read_block().await?;
                if !first.valid() {
                    let removed = self.heap.swap_remove(0);
                    self.origin_heap.push(removed);
                    if self.heap.is_empty() {
                        return Ok(read_row);
                    }
                }
                self.down(0);
                self.update_first_block_end_row_idx();
            } else if first.row_idx == self.first_batch_end_row_idx {
                self.down(0);
                self.update_first_block_end_row_idx();
            }
        }
        Ok(read_row)
    }

    fn get_remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        let mut col_idents = vec![];
        let mut seg_count = 0;
        for reader in &self.heap {
            let (idents, count) = reader.inner_reader().get_remote_segments()?;
            col_idents.extend(idents);
            seg_count += count;
        }
        Ok((col_idents, seg_count))
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = ColumnarRuntimeStats::default();
        for reader in &self.heap {
            let reader_stats = reader.reader.runtime_stats();
            stats.merge(&reader_stats);
        }
        for reader in &self.origin_heap {
            let reader_stats = reader.reader.runtime_stats();
            stats.merge(&reader_stats);
        }
        stats
    }
}

/// A source used by the late-materialization merge. Eager sources still read
/// their full projection; L2 sources only read `early_schema` and retain a
/// stable physical row identity for the deferred projection.
pub(crate) enum HybridSourceReader {
    Eager(Box<dyn ColumnarReader>),
    L2 {
        reader: ColumnarConcatReader,
        l2_reader_id: u32,
    },
}

struct HybridReaderBuffer {
    source: HybridSourceReader,
    early: Block,
    eager_late: Vec<ColumnBuffer>,
    deferred_refs: Vec<MaterializationRef>,
    row_idx: usize,
}

impl HybridReaderBuffer {
    fn is_eager(&self) -> bool {
        matches!(self.source, HybridSourceReader::Eager(_))
    }

    fn handle(&self) -> &[u8] {
        self.early.handles.get_not_null_value(self.row_idx)
    }

    fn int_handle(&self) -> i64 {
        self.early.handles.get_int_handle_value(self.row_idx)
    }

    fn version(&self) -> u64 {
        self.early.versions.get_version(self.row_idx)
    }

    fn valid(&self) -> bool {
        self.row_idx < self.early.length()
    }

    fn reset(&mut self) -> crate::table::Result<()> {
        self.row_idx = 0;
        self.early.reset();
        self.eager_late.iter_mut().for_each(ColumnBuffer::reset);
        self.deferred_refs.clear();
        match &mut self.source {
            HybridSourceReader::Eager(reader) => reader.reset(),
            HybridSourceReader::L2 { reader, .. } => reader.reset(),
        }
    }

    async fn seek(&mut self, handle: &[u8], full_schema: &Schema) -> crate::table::Result<()> {
        self.row_idx = 0;
        self.early.reset();
        self.eager_late.iter_mut().for_each(ColumnBuffer::reset);
        self.deferred_refs.clear();
        match &mut self.source {
            HybridSourceReader::Eager(reader) => reader.seek(handle).await,
            HybridSourceReader::L2 { reader, .. } => reader.seek(handle).await,
        }?;
        self.read_block(full_schema).await
    }

    async fn read_block(&mut self, full_schema: &Schema) -> crate::table::Result<()> {
        self.row_idx = 0;
        self.early.reset();
        self.eager_late.iter_mut().for_each(ColumnBuffer::reset);
        self.deferred_refs.clear();
        match &mut self.source {
            HybridSourceReader::Eager(reader) => {
                let mut full = Block::new(full_schema);
                reader.read(&mut full, 1024).await?;
                self.early.handles.append(&full.handles, 0, full.length());
                self.early.versions.append(&full.versions, 0, full.length());
                for early_column in &mut self.early.columns {
                    early_column.append(
                        full.get_column(early_column.col_id() as i64),
                        0,
                        full.length(),
                    );
                }
                for late_column in &mut self.eager_late {
                    late_column.append(
                        full.get_column(late_column.col_id() as i64),
                        0,
                        full.length(),
                    );
                }
            }
            HybridSourceReader::L2 {
                reader,
                l2_reader_id,
            } => {
                let mut refs = vec![];
                reader
                    .read_with_physical_rows(&mut self.early, &mut refs, 1024)
                    .await?;
                self.deferred_refs.extend(refs.into_iter().map(|reference| {
                    MaterializationRef::L2Deferred {
                        l2_reader_id: *l2_reader_id,
                        file_index: reference.file_index,
                        physical_row_idx: reference.physical_row_idx,
                    }
                }));
                debug_assert_eq!(self.early.length(), self.deferred_refs.len());
            }
        }
        Ok(())
    }

    fn remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        match &self.source {
            HybridSourceReader::Eager(reader) => reader.get_remote_segments(),
            HybridSourceReader::L2 { reader, .. } => reader.get_remote_segments(),
        }
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        match &self.source {
            HybridSourceReader::Eager(reader) => reader.runtime_stats(),
            HybridSourceReader::L2 { reader, .. } => reader.runtime_stats(),
        }
    }
}

/// Mirrors `ColumnarMergeReader`'s heap ordering, but moves rows into a
/// `HybridBatch` rather than a full block. This keeps L2 filtering strictly
/// after global merge and MVCC.
pub(crate) struct HybridMergeReader {
    schema: Schema,
    full_schema: Schema,
    late_columns: Vec<ColumnInfo>,
    #[allow(clippy::vec_box)] // Mirrors ColumnarMergeReader's movable source buffers.
    heap: Vec<Box<HybridReaderBuffer>>,
    #[allow(clippy::vec_box)] // Keep exhausted buffers for reset without rebuilding readers.
    origin_heap: Vec<Box<HybridReaderBuffer>>,
    is_int_handle: bool,
    first_batch_end_row_idx: usize,
}

impl HybridMergeReader {
    pub(crate) fn new(
        schema: Schema,
        full_schema: Schema,
        late_columns: Vec<ColumnInfo>,
        sources: Vec<HybridSourceReader>,
    ) -> Self {
        let heap = sources
            .into_iter()
            .map(|source| {
                Box::new(HybridReaderBuffer {
                    source,
                    early: Block::new(&schema),
                    eager_late: late_columns
                        .iter()
                        .map(ColumnBuffer::new_from_col_info)
                        .collect(),
                    deferred_refs: vec![],
                    row_idx: 0,
                })
            })
            .collect();
        Self {
            is_int_handle: !schema.is_common_handle(),
            schema,
            full_schema,
            late_columns,
            heap,
            origin_heap: vec![],
            first_batch_end_row_idx: 0,
        }
    }

    fn less(&self, a: usize, b: usize) -> bool {
        if self.is_int_handle {
            let a_handle = self.heap[a].int_handle();
            let b_handle = self.heap[b].int_handle();
            a_handle < b_handle
                || (a_handle == b_handle && self.heap[a].version() > self.heap[b].version())
        } else {
            match self.heap[a].handle().cmp(self.heap[b].handle()) {
                Ordering::Less => true,
                Ordering::Equal => self.heap[a].version() > self.heap[b].version(),
                Ordering::Greater => false,
            }
        }
    }

    fn down(&mut self, i0: usize) {
        let mut i = i0;
        loop {
            let left = i * 2 + 1;
            if left >= self.heap.len() {
                return;
            }
            let right = left + 1;
            let child = if right < self.heap.len() && self.less(right, left) {
                right
            } else {
                left
            };
            if !self.less(child, i) {
                return;
            }
            self.heap.swap(i, child);
            i = child;
        }
    }

    fn update_first_block_end_row_idx(&mut self) {
        let first = &self.heap[0];
        if self.heap.len() == 1 {
            self.first_batch_end_row_idx = first.early.length();
            return;
        }
        let second_handle = if self.heap.len() >= 3 {
            if self.is_int_handle {
                if self.heap[1].int_handle() < self.heap[2].int_handle() {
                    self.heap[1].handle()
                } else {
                    self.heap[2].handle()
                }
            } else if self.heap[1].handle() < self.heap[2].handle() {
                self.heap[1].handle()
            } else {
                self.heap[2].handle()
            }
        } else {
            self.heap[1].handle()
        };
        if self.is_int_handle {
            let mut second_handle = second_handle;
            let second_handle = second_handle.get_i64_le();
            for row_idx in first.row_idx + 1..first.early.length() {
                if first.early.handles.get_int_handle_value(row_idx) >= second_handle {
                    self.first_batch_end_row_idx = row_idx;
                    return;
                }
            }
        } else {
            for row_idx in first.row_idx + 1..first.early.length() {
                if first.early.handles.get_not_null_value(row_idx) >= second_handle {
                    self.first_batch_end_row_idx = row_idx;
                    return;
                }
            }
        }
        self.first_batch_end_row_idx = first.early.length();
    }

    fn init_heap(&mut self) {
        for idx in (0..self.heap.len() / 2).rev() {
            self.down(idx);
        }
        if !self.heap.is_empty() {
            self.update_first_block_end_row_idx();
        }
    }

    pub(crate) async fn seek(&mut self, handle: &[u8]) -> crate::table::Result<()> {
        for buffer in &mut self.heap {
            buffer.seek(handle, &self.full_schema).await?;
        }
        let (invalid, valid) = self.heap.drain(..).partition(|buffer| !buffer.valid());
        self.origin_heap = invalid;
        self.heap = valid;
        self.init_heap();
        Ok(())
    }

    pub(crate) fn reset(&mut self) -> crate::table::Result<()> {
        while let Some(buffer) = self.origin_heap.pop() {
            self.heap.push(buffer);
        }
        for buffer in &mut self.heap {
            buffer.reset()?;
        }
        Ok(())
    }

    pub(crate) async fn read(
        &mut self,
        batch: &mut HybridBatch,
        limit: usize,
    ) -> crate::table::Result<usize> {
        batch.early.reset();
        batch.refs.clear();
        batch.eager_late.iter_mut().for_each(ColumnBuffer::reset);
        let mut rows = 0;
        while rows < limit && !self.heap.is_empty() {
            let first = &mut self.heap[0];
            let remain = min(limit - rows, self.first_batch_end_row_idx - first.row_idx);
            if first.is_eager() {
                batch.append_eager(
                    &first.early,
                    &first.eager_late,
                    first.row_idx,
                    first.row_idx + remain,
                )?;
            } else {
                batch.append_deferred(
                    &first.early,
                    &first.deferred_refs,
                    first.row_idx,
                    first.row_idx + remain,
                )?;
            }
            first.row_idx += remain;
            rows += remain;
            if first.row_idx == first.early.length() {
                first.read_block(&self.full_schema).await?;
                if !first.valid() {
                    self.origin_heap.push(self.heap.swap_remove(0));
                    if self.heap.is_empty() {
                        break;
                    }
                }
                self.down(0);
                self.update_first_block_end_row_idx();
            } else if first.row_idx == self.first_batch_end_row_idx {
                self.down(0);
                self.update_first_block_end_row_idx();
            }
        }
        batch.validate()?;
        Ok(rows)
    }

    pub(crate) fn remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        let mut idents = vec![];
        let mut count = 0;
        for buffer in self.heap.iter().chain(self.origin_heap.iter()) {
            let (source_idents, source_count) = buffer.remote_segments()?;
            idents.extend(source_idents);
            count += source_count;
        }
        Ok((idents, count))
    }

    pub(crate) fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = ColumnarRuntimeStats::default();
        for buffer in self.heap.iter().chain(self.origin_heap.iter()) {
            stats.merge(&buffer.runtime_stats());
        }
        stats
    }

    pub(crate) fn schema(&self) -> &Schema {
        &self.schema
    }

    pub(crate) fn late_columns(&self) -> &[ColumnInfo] {
        &self.late_columns
    }
}

/// MVCC wrapper for `HybridMergeReader`. It shares the same visibility state
/// implementation as the legacy reader, but preserves row references and
/// eager sidecars when rows are removed.
pub(crate) struct HybridMvccReader {
    src: HybridMergeReader,
    state: ColumnarMvccState,
    l2_files: Vec<Vec<ColumnarFile>>,
}

impl HybridMvccReader {
    pub(crate) fn new(
        src: HybridMergeReader,
        read_ts: u64,
        l2_files: Vec<Vec<ColumnarFile>>,
    ) -> Self {
        let state = ColumnarMvccState::new(src.schema(), read_ts);
        Self {
            src,
            state,
            l2_files,
        }
    }

    pub(crate) fn schema(&self) -> &Schema {
        self.src.schema()
    }

    pub(crate) fn late_columns(&self) -> &[ColumnInfo] {
        self.src.late_columns()
    }

    pub(crate) fn l2_files(&self) -> &[Vec<ColumnarFile>] {
        &self.l2_files
    }

    pub(crate) async fn set_handle_range(
        &mut self,
        start: &[u8],
        end: &[u8],
    ) -> crate::table::Result<()> {
        self.state.end_handle = end.to_vec();
        self.state.filter.clear();
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
        self.src.seek(start).await
    }

    pub(crate) async fn set_int_handle_range(
        &mut self,
        start: i64,
        end: Option<i64>,
    ) -> crate::table::Result<()> {
        self.state.end_int_handle = end;
        self.state.filter.clear();
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
        self.src.seek(&start.to_le_bytes()).await
    }

    pub(crate) fn reset(&mut self) -> crate::table::Result<()> {
        self.state.prev_int_handle = None;
        self.state.prev_common_handle.clear();
        self.state.range_exhausted = false;
        self.src.reset()
    }

    pub(crate) async fn read(
        &mut self,
        batch: &mut HybridBatch,
        limit: usize,
    ) -> crate::table::Result<usize> {
        if self.state.range_exhausted {
            batch.early.reset();
            batch.refs.clear();
            batch.eager_late.iter_mut().for_each(ColumnBuffer::reset);
            return Ok(0);
        }
        loop {
            let rows = self.src.read(batch, limit).await?;
            if rows == 0 {
                return Ok(0);
            }
            self.state.runtime_stats.mvcc_input_rows = self
                .state
                .runtime_stats
                .mvcc_input_rows
                .saturating_add(rows as u64);
            self.state.runtime_stats.mvcc_input_bytes = self
                .state
                .runtime_stats
                .mvcc_input_bytes
                .saturating_add(batch.early.total_byte_size());
            self.state.apply_hybrid_visibility(batch)?;
            let visible_rows = batch.early.length();
            self.state.runtime_stats.mvcc_output_rows = self
                .state
                .runtime_stats
                .mvcc_output_rows
                .saturating_add(visible_rows as u64);
            if visible_rows > 0 {
                return Ok(visible_rows);
            }
            if self.state.range_exhausted {
                return Ok(0);
            }
        }
    }

    pub(crate) fn remote_segments(&self) -> crate::table::Result<(usize, usize)> {
        let (idents, count) = self.src.remote_segments()?;
        Ok((idents.len(), count))
    }

    pub(crate) fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.state.runtime_stats;
        stats.merge(&self.src.runtime_stats());
        stats
    }
}

pub(crate) fn parse_default_val(col_info: &ColumnInfo) -> table::Result<Option<Vec<u8>>> {
    let mut default_val = col_info.get_default_val();
    if default_val.is_empty() {
        return Ok(None);
    }
    let default_len = default_val.len();
    let flag = default_val.get_u8();
    let col_id = col_info.get_column_id();
    let tp = col_info.get_tp();
    let decode_err = |err: &dyn std::fmt::Display| {
        table::Error::Other(format!(
            "failed to decode default value (col_id={}, tp={}, flag={}, len={}): {}",
            col_id, tp, flag, default_len, err
        ))
    };
    let val = match flag {
        INT_FLAG | DURATION_FLAG => decode_i64(&mut default_val)
            .map_err(|e| decode_err(&e))?
            .to_le_bytes()
            .to_vec(),
        UINT_FLAG => decode_u64(&mut default_val)
            .map_err(|e| decode_err(&e))?
            .to_le_bytes()
            .to_vec(),
        BYTES_FLAG => decode_bytes(&mut default_val, false).map_err(|e| decode_err(&e))?,
        COMPACT_BYTES_FLAG => decode_compact_bytes(&mut default_val).map_err(|e| decode_err(&e))?,
        NIL_FLAG => {
            return Ok(None);
        }
        FLOAT_FLAG => decode_f64(&mut default_val)
            .map_err(|e| decode_err(&e))?
            .to_le_bytes()
            .to_vec(),
        JSON_FLAG => default_val.to_vec(),
        VAR_INT_FLAG => decode_var_i64(&mut default_val)
            .map_err(|e| decode_err(&e))?
            .to_le_bytes()
            .to_vec(),
        VAR_UINT_FLAG => decode_var_u64(&mut default_val)
            .map_err(|e| decode_err(&e))?
            .to_le_bytes()
            .to_vec(),
        DECIMAL_FLAG => {
            let decimal = default_val.read_decimal().map_err(|e| decode_err(&e))?;
            decode_decimal_as_int(col_info, &decimal)
        }
        VECTOR_FLOAT32_FLAG => {
            warn!("unimplemented vector default");
            return Ok(None);
        }
        f => {
            warn!("parse default val get unknown flag {}", f);
            return Ok(None);
        }
    };
    Ok(Some(val))
}

pub struct ColumnarRowTableReader {
    schema: Schema,
    iter: Box<dyn table::Iterator>,
    blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
    prefix: Vec<u8>,
    default_vals: Vec<Option<Vec<u8>>>,
    is_int_handle: bool,
    check_schema: bool,
    max_col_id: i32,
    schema_version: i64,
    encryption_key: Option<EncryptionKey>,
    decryption_buf: Vec<u8>,
}

impl ColumnarRowTableReader {
    pub fn new(
        schema: Schema,
        iter: Box<dyn table::Iterator>,
        blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
        check_schema: bool,
        encryption_key: Option<EncryptionKey>,
    ) -> table::Result<ColumnarRowTableReader> {
        Self::new_with_schema_version(schema, iter, blob_tbls, check_schema, encryption_key, 0)
    }

    pub fn new_with_schema_version(
        schema: Schema,
        iter: Box<dyn table::Iterator>,
        blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
        check_schema: bool,
        encryption_key: Option<EncryptionKey>,
        schema_version: i64,
    ) -> table::Result<ColumnarRowTableReader> {
        let mut prefix = encode_row_key(schema.table_id, 0);
        prefix.truncate(PREFIX_LEN);
        let is_int_handle = get_fixed_size(&schema.handle_column) > 0;
        let default_vals = schema
            .columns
            .iter()
            .map(|col| parse_default_val(col))
            .collect::<table::Result<Vec<_>>>()?;
        let max_col_id = schema.max_col_id as i32;

        Ok(ColumnarRowTableReader {
            schema,
            iter,
            blob_tbls,
            prefix,
            default_vals,
            is_int_handle,
            check_schema,
            max_col_id,
            schema_version,
            encryption_key,
            decryption_buf: vec![],
        })
    }

    fn schema_out_of_date(&self, column_id: i64, detail: impl Into<String>) -> table::Error {
        table::Error::SchemaOutOfDate(table::SchemaOutOfDateInfo {
            schema_ver: self.schema_version,
            table_id: self.schema.table_id,
            column_id,
            detail: detail.into(),
        })
    }

    fn append_deleted_row_columns(block: &mut Block) {
        for col in &mut block.columns {
            if col.nullable {
                col.push_null();
            } else {
                col.push_zero();
            }
        }
    }

    fn push_handle_from_key(&self, block: &mut Block, key: InnerKey<'_>) {
        if self.is_int_handle {
            let int_handle = decode_int_handle(key.deref()).unwrap();
            block.handles.push_value(&int_handle.to_le_bytes());
        } else {
            let common_handle = decode_common_handle(key.deref()).unwrap();
            block.handles.push_value(common_handle);
        }
    }

    fn build_row_key_from_handle_bytes(&self, handle: &[u8]) -> Vec<u8> {
        if self.is_int_handle {
            let int_handle = if handle.is_empty() {
                i64::MIN
            } else {
                (&handle[..]).get_i64_le()
            };
            encode_row_key(self.schema.table_id, int_handle)
        } else {
            encode_common_handle_row_key(self.schema.table_id, handle)
        }
    }

    async fn push_row_value(
        &mut self,
        block: &mut Block,
        value: table::Value,
        ctx: &'static str,
    ) -> table::Result<()> {
        let version = value.version;
        let is_deleted = value.is_deleted() || value.is_value_empty();
        block.versions.push_version(version, is_deleted);
        if is_deleted {
            Self::append_deleted_row_columns(block);
            return Ok(());
        }

        if value.is_blob_ref() {
            let blob_ref = value.get_blob_ref();
            let blob_tbl = self
                .blob_tbls
                .as_ref()
                .and_then(|tbls| tbls.get(&blob_ref.fid))
                .ok_or_else(|| {
                    table::Error::Other(format!(
                        "{ctx} blob table not found (fid={})",
                        blob_ref.fid
                    ))
                })?;
            let value_buf = blob_tbl
                .get_by_ref_async(
                    &blob_ref,
                    &mut self.decryption_buf,
                    self.encryption_key.clone(),
                )
                .await
                .map_err(|e| table::Error::Other(e.to_string()))?;
            self.decode_row_columns(block, &value_buf)?;
            Ok(())
        } else {
            self.decode_row_columns(block, value.get_value())
        }
    }

    fn decode_row_columns(&self, block: &mut Block, row_value: &[u8]) -> table::Result<()> {
        if row_value[0] == CODEC_VERSION {
            self.decode_row_columns_v2(block, row_value)
        } else {
            self.decode_row_columns_v1(block, row_value)
        }
    }

    fn decode_row_columns_v1(&self, block: &mut Block, row_value: &[u8]) -> table::Result<()> {
        let mut data: BytesSlice<'_> = row_value;
        let datums = decode(&mut data).map_err(|e| table::Error::Other(e.to_string()))?;
        let mut datums_map = HashMap::with_capacity(datums.len() / 2);
        let mut datums_iter = datums.into_iter();
        while let (Some(col_id), Some(data)) = (datums_iter.next(), datums_iter.next()) {
            if let Ok(Some(col_id)) = col_id.as_int() {
                datums_map.insert(col_id, data);
            } else {
                return Err(table::Error::Other("invalid col id".to_string()));
            }
        }

        for (offset, col_info) in self.schema.columns.iter().enumerate() {
            let col_id = col_info.get_column_id();
            let col_buf = &mut block.columns[offset];

            if datums_map.contains_key(&col_id) {
                if self.check_schema
                    && matches!(datums_map.get(&col_id).unwrap(), Datum::Null)
                    && !col_buf.nullable
                {
                    return Err(self.schema_out_of_date(col_id, "null_in_not_null"));
                }
                Self::push_col_buf_with_row_v1_datum(
                    col_buf,
                    col_info,
                    datums_map.get(&col_id).unwrap(),
                );
            } else if !self.is_int_handle && get_primary_key(col_info) {
                // get value from common handle
                let mut common_handle =
                    block.handles.get_not_null_value(block.handles.length() - 1);
                for &pk_col_id in &self.schema.pk_col_ids {
                    let (datum, remain) = datum::split_datum(common_handle, false).unwrap();
                    if pk_col_id == col_id {
                        if let Err(e) =
                            Self::push_col_buf_with_common_handle_datum(col_buf, col_info, datum)
                        {
                            error!(
                                "decode_row_columns_v1 pk: {:?}, col_info col_id: {}, col_buf col_id: {}, tp: {}, offset: {}, val_len: {}, schema: {:?}",
                                e,
                                col_id,
                                col_buf.col_id,
                                col_info.get_tp(),
                                offset,
                                datum.len(),
                                self.schema
                            );
                            return Err(table::Error::Other(e.to_string()));
                        }
                        break;
                    }
                    common_handle = remain;
                }
            } else if let Some(default_val) = &self.default_vals[offset] {
                col_buf.push_value(default_val);
            } else if col_buf.is_nullable() {
                col_buf.push_null();
            } else {
                // The column is skipped in row data and should be read as NULL value, but the
                // ColumnBuffer is not nullable.
                return Err(self.schema_out_of_date(col_id, "read null for not null column"));
            }
        }

        Ok(())
    }

    fn decode_row_columns_v2(&self, block: &mut Block, row_value: &[u8]) -> table::Result<()> {
        let row_slice = RowSlice::from_bytes(row_value).unwrap();
        if self.check_schema {
            let row_max_col_id = row_slice.max_col_id();
            if self.max_col_id < row_max_col_id {
                return Err(self.schema_out_of_date(
                    row_max_col_id as i64,
                    "row contains newer column".to_owned(),
                ));
            }
        }

        let values = row_slice.values();
        for (offset, col_info) in self.schema.columns.iter().enumerate() {
            let col_id = col_info.get_column_id();
            let col_buf = &mut block.columns[offset];
            assert_eq!(col_buf.col_id as i64, col_id);
            if row_slice.search_in_null_ids(col_id) {
                if self.check_schema && !col_buf.nullable {
                    return Err(self.schema_out_of_date(col_id, "null_in_not_null"));
                }
                col_buf.push_null();
                continue;
            }
            if let Some((start, end)) = row_slice.search_in_non_null_ids(col_id).unwrap() {
                let col_val = &values[start..end];
                if let Err(e) = Self::push_col_buf_with_field_type(col_buf, col_info, col_val) {
                    error!(
                        "decode_row_columns_v2: {:?}, col_info col_id: {}, col_buf col_id: {}, tp: {}, offset: {}, val_len: {}, schema: {:?}",
                        e,
                        col_id,
                        col_buf.col_id,
                        col_info.get_tp(),
                        offset,
                        col_val.len(),
                        self.schema
                    );
                    return Err(table::Error::Other(e.to_string()));
                }
            } else if !self.is_int_handle && get_primary_key(col_info) {
                // get value from common handle
                let mut common_handle =
                    block.handles.get_not_null_value(block.handles.length() - 1);
                for &pk_col_id in &self.schema.pk_col_ids {
                    let (datum, remain) = datum::split_datum(common_handle, false).unwrap();
                    if pk_col_id == col_id {
                        if let Err(e) =
                            Self::push_col_buf_with_common_handle_datum(col_buf, col_info, datum)
                        {
                            error!(
                                "decode_row_columns_v2 pk: {:?}, col_info col_id: {}, col_buf col_id: {}, tp: {}, offset: {}, val_len: {}, schema: {:?}",
                                e,
                                col_id,
                                col_buf.col_id,
                                col_info.get_tp(),
                                offset,
                                datum.len(),
                                self.schema
                            );
                            return Err(table::Error::Other(e.to_string()));
                        }
                        break;
                    }
                    common_handle = remain;
                }
            } else if let Some(default_val) = &self.default_vals[offset] {
                col_buf.push_value(default_val);
            } else if col_buf.is_nullable() {
                col_buf.push_null();
            } else {
                // The column is skipped in row data and should be read as NULL value, but the
                // ColumnBuffer is not nullable.
                return Err(self.schema_out_of_date(col_id, "read null for not null column"));
            }
        }
        Ok(())
    }

    fn push_col_buf_with_field_type(
        col_buf: &mut ColumnBuffer,
        col_info: &ColumnInfo,
        col_val: &[u8],
    ) -> tidb_query_datatype::codec::Result<()> {
        let ft = FieldTypeTp::from_i32(col_info.get_tp()).ok_or(
            tidb_query_datatype::codec::Error::InvalidDataType("invalid field type".to_string()),
        )?;
        match ft {
            FieldTypeTp::Tiny
            | FieldTypeTp::Short
            | FieldTypeTp::Int24
            | FieldTypeTp::Long
            | FieldTypeTp::LongLong => {
                if is_unsigned(col_info) {
                    let v = decode_v2_u64(col_val).unwrap();
                    col_buf.push_value(&v.to_le_bytes());
                } else {
                    let v = decode_v2_i64(col_val)?;
                    col_buf.push_value(&v.to_le_bytes());
                }
            }
            FieldTypeTp::Date
            | FieldTypeTp::DateTime
            | FieldTypeTp::Timestamp
            | FieldTypeTp::NewDate
            | FieldTypeTp::Enum
            | FieldTypeTp::Bit
            | FieldTypeTp::Set => {
                let v = decode_v2_u64(col_val)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            FieldTypeTp::Year | FieldTypeTp::Duration => {
                let v = decode_v2_i64(col_val)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            FieldTypeTp::Float | FieldTypeTp::Double => {
                let mut val = col_val;
                let v = decode_f64(&mut val)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            FieldTypeTp::Null => {
                col_buf.push_null();
            }
            FieldTypeTp::NewDecimal => {
                // Marshal decimal in TiFlash compatible format.
                let mut val = col_val;
                let decimal = val.read_decimal().unwrap();
                col_buf.push_value(&decode_decimal_as_int(col_info, &decimal));
            }
            FieldTypeTp::Unspecified
            | FieldTypeTp::VarChar
            | FieldTypeTp::Json
            | FieldTypeTp::TinyBlob
            | FieldTypeTp::MediumBlob
            | FieldTypeTp::LongBlob
            | FieldTypeTp::Blob
            | FieldTypeTp::VarString
            | FieldTypeTp::String
            | FieldTypeTp::Geometry
            | FieldTypeTp::TiDbVectorFloat32 => {
                col_buf.push_value(col_val);
            }
        }
        Ok(())
    }

    fn push_col_buf_with_row_v1_datum(
        col_buf: &mut ColumnBuffer,
        col_info: &ColumnInfo,
        datum: &Datum,
    ) {
        match datum {
            Datum::I64(v) => {
                col_buf.push_value(&v.to_le_bytes());
            }
            Datum::Dur(ref d) => {
                col_buf.push_value(&d.to_nanos().to_le_bytes());
            }
            Datum::U64(v) => {
                col_buf.push_value(&v.to_le_bytes());
            }
            Datum::Bytes(ref bs) => {
                col_buf.push_value(bs);
            }
            Datum::Null => {
                col_buf.push_null();
            }
            Datum::F64(v) => {
                col_buf.push_value(&v.to_le_bytes());
            }
            Datum::Dec(ref d) => {
                col_buf.push_value(&decode_decimal_as_int(col_info, d));
            }
            Datum::VectorFloat32(ref v) => {
                let mut buf = vec![];
                buf.write_vector_float32(v.as_ref()).unwrap();
                col_buf.push_value(&buf);
            }
            Datum::Json(ref j) => {
                let mut buf = vec![];
                buf.write_json(j.as_ref()).unwrap();
                col_buf.push_value(&buf);
            }
            _ => {
                panic!("unsupported datum type: {:?}", datum);
            }
        }
    }

    fn push_col_buf_with_common_handle_datum(
        col_buf: &mut ColumnBuffer,
        col_info: &ColumnInfo,
        mut datum: &[u8],
    ) -> tidb_query_datatype::codec::Result<()> {
        let flag = datum.get_u8();
        match flag {
            INT_FLAG | DURATION_FLAG => {
                let v = decode_i64(&mut datum)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            UINT_FLAG => {
                let v = decode_u64(&mut datum)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            BYTES_FLAG => {
                let v = decode_bytes(&mut datum, false)?;
                col_buf.push_value(&v);
            }
            NIL_FLAG => {
                col_buf.push_null();
            }
            FLOAT_FLAG => {
                let v = decode_f64(&mut datum)?;
                col_buf.push_value(&v.to_le_bytes());
            }
            DECIMAL_FLAG => {
                // let v = datum.read_decimal().unwrap();
                // let mut buf = vec![];
                // let prec = col_info.get_column_len() as u8;
                // let frac = col_info.get_decimal() as u8;
                // buf.write_decimal(&v, prec, frac).unwrap();
                // col_buf.push_value(&buf);
                //
                // Marshal decimal to TiFlash compatible format.
                let decimal = datum.read_decimal().unwrap();
                col_buf.push_value(&decode_decimal_as_int(col_info, &decimal));
            }
            JSON_FLAG | VECTOR_FLOAT32_FLAG | VAR_UINT_FLAG | VAR_INT_FLAG | COMPACT_BYTES_FLAG => {
                return Err(tidb_query_datatype::codec::Error::InvalidDataType(format!(
                    "invalid flag {} in common handle",
                    flag
                )));
            }
            _ => {
                return Err(tidb_query_datatype::codec::Error::InvalidDataType(format!(
                    "unknown flag {} in common handle",
                    flag
                )));
            }
        }
        Ok(())
    }
}

pub fn is_unsigned(col_info: &ColumnInfo) -> bool {
    FieldTypeFlag::from_bits_truncate(col_info.get_flag() as u32).contains(FieldTypeFlag::UNSIGNED)
}

#[async_trait]
impl ColumnarReader for ColumnarRowTableReader {
    fn schema(&self) -> &Schema {
        &self.schema
    }

    async fn seek(&mut self, handle: &[u8]) -> table::Result<()> {
        let row_key = if self.is_int_handle && !handle.is_empty() {
            encode_row_key(self.schema.table_id, (&handle[..]).get_i64_le())
        } else {
            encode_common_handle_row_key(self.schema.table_id, handle)
        };
        if self.iter.is_cf_sync(WRITE_CF) {
            self.iter.seek(InnerKey::from_inner_buf(&row_key));
        } else {
            self.iter
                .seek_async(InnerKey::from_inner_buf(&row_key))
                .await;
        }
        Ok(())
    }

    async fn read(&mut self, block: &mut Block, limit: usize) -> table::Result<usize> {
        let mut read_rows = 0;
        while self.iter.valid() && read_rows < limit {
            let key = self.iter.key();
            if !key.starts_with(&self.prefix) {
                break;
            }
            let value = self.iter.value();
            self.push_handle_from_key(block, key);
            self.push_row_value(block, value, "ColumnarRowTableReader")
                .await?;
            read_rows += 1;
            if self.iter.is_cf_sync(WRITE_CF) {
                self.iter.next_all_version();
            } else {
                self.iter.next_all_version_async().await;
            }
        }
        Ok(read_rows)
    }
}

struct PrefixedRowIterator {
    prefix: Vec<u8>,
    iter: Box<dyn table::Iterator>,
}

impl PrefixedRowIterator {
    fn new(prefix: Vec<u8>, iter: Box<dyn table::Iterator>) -> Self {
        Self { prefix, iter }
    }
}

#[maybe_async::async_trait]
impl table::Iterator for PrefixedRowIterator {
    #[maybe_async]
    async fn next(&mut self) {
        self.iter.next().await;
    }

    fn is_next_sync(&self) -> bool {
        self.iter.is_next_sync()
    }

    #[maybe_async]
    async fn next_version(&mut self) -> bool {
        self.iter.next_version().await
    }

    fn is_next_version_sync(&self) -> bool {
        self.iter.is_next_version_sync()
    }

    #[maybe_async]
    async fn rewind(&mut self) {
        self.iter.rewind().await;
    }

    #[maybe_async]
    async fn seek(&mut self, key: InnerKey<'_>) {
        self.iter.seek(key).await;
    }

    fn is_cf_sync(&self, cf: usize) -> bool {
        self.iter.is_cf_sync(cf)
    }

    fn key(&self) -> InnerKey<'_> {
        self.iter.key()
    }

    fn value(&self) -> table::Value {
        self.iter.value()
    }

    fn valid(&self) -> bool {
        if !self.iter.valid() {
            return false;
        }
        self.iter.key().starts_with(&self.prefix)
    }
}

/// Reader/struct relationships introduced for merged MVCC decode.
///
/// Ownership graph (has-a) + data flow (left-to-right):
///
/// ```text
/// RowMvccReader(read_ts)      RowToColumnarReader(safe_ts, level)
///          |                               |
///          +---------------+---------------+
///                          v
///               ColumnarRowTableReader
///               (iter = AsyncMergeIterator)
///                    |
///                    +--> AsyncMergeIterator<Vec<Box<dyn table::Iterator>>>
///                             ^
///                             |
///                    PrefixedRowIterator(table_prefix)
///                             ^
///                             |
///                  per-source row iterator (memtable / SST)
///
/// Output:
///   (handle, version, row_value) --decode--> Block
/// ```
///
/// Notes:
/// - Prefix filtering is required because the merged iterator contains mixed
///   keys; `PrefixedRowIterator` guarantees a bounded table-only stream.
/// - `ColumnarRowTableReader` provides the row-value decode logic (including
///   BlobRef resolution + decryption buffer reuse). `RowMvccReader` and
///   `RowToColumnarReader` drive the merged iterator because they need custom
///   MVCC visibility / `safe_ts` pruning on the merged stream.
fn build_prefixed_row_merge_iter(
    table_id: i64,
    iters: Vec<Box<dyn table::Iterator>>,
) -> Box<dyn table::Iterator> {
    let is_always_sync = iters.iter().all(|iter| iter.is_cf_sync(WRITE_CF));
    let mut prefix = encode_row_key(table_id, 0);
    prefix.truncate(PREFIX_LEN);
    let iters: Vec<Box<dyn table::Iterator>> = iters
        .into_iter()
        .map(|iter| {
            Box::new(PrefixedRowIterator::new(prefix.clone(), iter)) as Box<dyn table::Iterator>
        })
        .collect();
    Box::new(crate::table::AsyncMergeIterator::new(
        iters,
        false,
        is_always_sync,
    ))
}

/// Merge multiple row iterators (typically from memtables / row SSTs) into a
/// single globally ordered MVCC stream for one table, pick the latest visible
/// version for each handle (by `read_ts`) on the merged stream, and decode only
/// that version into a columnar block.
///
/// Why this exists:
/// - `ColumnarRowTableReader` decodes *every* MVCC version it visits. During
///   reads, older versions are typically filtered out later by
///   `ColumnarMvccReader`, but decoding them eagerly can fail with
///   `SchemaOutOfDate` (e.g. `read null for not null column`) even though the
///   visible version is schema-compatible.
/// - MVCC visibility must be decided on the *merged* MVCC stream across all row
///   inputs; filtering per input is insufficient when a newer version in one
///   source overwrites an older (and potentially schema-incompatible) version
///   in another source.
pub(crate) struct RowMvccReader {
    row_reader: ColumnarRowTableReader,
    read_ts: u64,
}

impl RowMvccReader {
    pub fn new(
        schema: Schema,
        iters: Vec<Box<dyn table::Iterator>>,
        blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
        check_schema: bool,
        encryption_key: Option<EncryptionKey>,
        read_ts: u64,
    ) -> table::Result<Self> {
        let merge_iter = build_prefixed_row_merge_iter(schema.table_id, iters);
        Ok(Self {
            row_reader: ColumnarRowTableReader::new_with_schema_version(
                schema,
                merge_iter,
                blob_tbls,
                check_schema,
                encryption_key,
                0,
            )?,
            read_ts,
        })
    }
}

#[async_trait]
impl ColumnarReader for RowMvccReader {
    fn schema(&self) -> &Schema {
        self.row_reader.schema()
    }

    async fn seek(&mut self, handle: &[u8]) -> table::Result<()> {
        let row_key = self.row_reader.build_row_key_from_handle_bytes(handle);
        self.row_reader
            .iter
            .seek_async(InnerKey::from_inner_buf(&row_key))
            .await;
        Ok(())
    }

    async fn read(&mut self, block: &mut Block, limit: usize) -> table::Result<usize> {
        let mut read_rows = 0;
        while read_rows < limit {
            if !self.row_reader.iter.valid() {
                break;
            }

            // For the current handle (row key), walk versions in descending order
            // and pick the first version <= read_ts.
            while self.row_reader.iter.valid()
                && self.row_reader.iter.value().version > self.read_ts
            {
                if next_version_async!(self.row_reader.iter) {
                    continue;
                }
                // No more versions for this handle across all sources.
                next_async!(self.row_reader.iter);
            }
            if !self.row_reader.iter.valid() {
                break;
            }
            if self.row_reader.iter.value().version > self.read_ts {
                // A safety net: in case the iterator can't advance versions but
                // also can't advance keys, avoid an infinite loop.
                next_async!(self.row_reader.iter);
                continue;
            }

            let key = self.row_reader.iter.key();
            let value = self.row_reader.iter.value();

            self.row_reader.push_handle_from_key(block, key);
            self.row_reader
                .push_row_value(block, value, "RowMvccReader")
                .await?;
            read_rows += 1;

            // Skip remaining versions of the same handle.
            next_async!(self.row_reader.iter);
        }
        Ok(read_rows)
    }
}

/// Merge multiple row iterators (typically from multiple row SSTs / memtables)
/// into a single globally ordered MVCC stream for one table, apply `safe_ts`
/// pruning on the merged stream, and decode only the retained versions into a
/// columnar block.
///
/// Why this exists:
/// - `safe_ts` pruning semantics are defined on the merged MVCC stream (across
///   all inputs).
/// - Decoding older versions that will be pruned can trigger `SchemaOutOfDate`
///   (e.g. `null_in_not_null`) and abort a columnar rebuild unnecessarily.
///
/// The pruning rule matches `ColumnarCompactReader`:
/// - Keep all versions with `version > safe_ts`.
/// - Keep the first `version <= safe_ts` per handle (the boundary).
/// - Drop older versions after the boundary.
/// - Additionally, for bottommost level (`level == 2`), drop a tombstone
///   boundary row (`is_deleted` and `version <= safe_ts`).
pub(crate) struct RowToColumnarReader {
    row_reader: ColumnarRowTableReader,
    safe_ts: u64,
    level: u32,
    end_row_key: Option<Vec<u8>>,
    drained: bool,
}

impl RowToColumnarReader {
    pub fn new(
        schema: Schema,
        iters: Vec<Box<dyn table::Iterator>>,
        blob_tbls: Option<Arc<HashMap<u64, BlobTable>>>,
        check_schema: bool,
        encryption_key: Option<EncryptionKey>,
        schema_version: i64,
        safe_ts: u64,
        level: u32,
    ) -> table::Result<Self> {
        let merge_iter = build_prefixed_row_merge_iter(schema.table_id, iters);
        Ok(Self {
            row_reader: ColumnarRowTableReader::new_with_schema_version(
                schema,
                merge_iter,
                blob_tbls,
                check_schema,
                encryption_key,
                schema_version,
            )?,
            safe_ts,
            level,
            end_row_key: None,
            drained: true,
        })
    }

    fn reached_end_by_key(&self, key: InnerKey<'_>) -> bool {
        if let Some(end_row_key) = self.end_row_key.as_ref() {
            key.deref() >= end_row_key.as_slice()
        } else {
            false
        }
    }
}

#[async_trait]
impl ColumnarFilterReader for RowToColumnarReader {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        self.drained = false;
        self.end_row_key = Some(self.row_reader.build_row_key_from_handle_bytes(end_handle));
        let start_row_key = self
            .row_reader
            .build_row_key_from_handle_bytes(start_handle);
        self.row_reader
            .iter
            .seek_async(InnerKey::from_inner_buf(&start_row_key))
            .await;
        Ok(())
    }

    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        self.drained = false;
        let table_id = self.row_reader.schema().table_id;
        self.end_row_key = end_handle.map(|end_handle| encode_row_key(table_id, end_handle));
        let start_row_key = encode_row_key(table_id, start_handle);
        self.row_reader
            .iter
            .seek_async(InnerKey::from_inner_buf(&start_row_key))
            .await;
        Ok(())
    }

    fn get_schema(&self) -> &Schema {
        self.row_reader.schema()
    }

    fn reset(&mut self) {
        // The underlying `table::Iterator` does not have a cheap reset that is both
        // sync and async-friendly, and current callers only use this reader for
        // a single forward scan. Keep the default no-op behavior but clear
        // local state to avoid surprises.
        self.end_row_key = None;
        self.drained = true;
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize, bool)> {
        block.reset();
        if limit == 0 {
            // Avoid triggering the `read_block()` retry loop.
            return Ok((0, true));
        }
        if self.drained {
            return Ok((0, true));
        }
        let mut read_row = 0;
        while read_row < limit {
            if !self.row_reader.iter.valid() {
                break;
            }
            let key = self.row_reader.iter.key();
            if self.reached_end_by_key(key) {
                self.drained = true;
                break;
            }

            let value = self.row_reader.iter.value();
            let version = value.version;

            let is_deleted = value.is_deleted() || value.is_value_empty();
            if version <= self.safe_ts {
                // `version <= safe_ts` is the per-handle boundary; after retaining
                // (or dropping) it, we can drop all older versions by moving to
                // the next handle directly.
                if !(self.level == 2 && is_deleted) {
                    self.row_reader.push_handle_from_key(block, key);
                    self.row_reader
                        .push_row_value(block, value, "RowToColumnarReader")
                        .await?;
                    read_row += 1;
                }
                next_async!(self.row_reader.iter);
                continue;
            }

            self.row_reader.push_handle_from_key(block, key);
            self.row_reader
                .push_row_value(block, value, "RowToColumnarReader")
                .await?;

            read_row += 1;
            if !next_version_async!(self.row_reader.iter) {
                next_async!(self.row_reader.iter);
            }
        }

        let drained = self.drained || !self.row_reader.iter.valid();
        if read_row == 0 && drained {
            return Ok((0, true));
        }
        Ok((read_row, drained))
    }
}

pub struct ColumnarConcatReader {
    schema: Schema,
    filter_op: Option<FilterOperator>,
    files: Vec<ColumnarFile>,
    reader: Option<ColumnarTableReader>,
    idx: usize,
    encryption_key: Option<EncryptionKey>,
    runtime_stats: ColumnarRuntimeStats,
}

impl ColumnarConcatReader {
    pub fn new(
        files: &[ColumnarFile],
        schema: Schema,
        filter_op: Option<FilterOperator>,
        encryption_key: Option<EncryptionKey>,
    ) -> ColumnarConcatReader {
        let mut reader_files = vec![];
        for file in files {
            if file.has_table(schema.table_id) {
                reader_files.push(file.clone());
            }
        }
        ColumnarConcatReader {
            schema,
            filter_op,
            files: reader_files,
            reader: None,
            idx: 0,
            encryption_key,
            runtime_stats: ColumnarRuntimeStats::default(),
        }
    }

    fn merge_reader_runtime_stats(&mut self, reader: &ColumnarTableReader) {
        let reader_stats = reader.runtime_stats();
        self.runtime_stats.merge(&reader_stats);
    }

    fn take_current_reader_runtime_stats(&mut self) {
        if let Some(reader) = self.reader.take() {
            self.merge_reader_runtime_stats(&reader);
        }
    }

    #[allow(dead_code)] // Consumed by the hybrid merge reader.
    pub(crate) async fn read_with_physical_rows(
        &mut self,
        block: &mut Block,
        refs: &mut Vec<L2PhysicalRowRef>,
        limit: usize,
    ) -> table::Result<usize> {
        let mut read_rows = 0;
        while self.idx < self.files.len() && read_rows < limit {
            let mut physical_rows = vec![];
            let cnt = {
                let reader = self.reader.as_mut().ok_or_else(|| {
                    table::Error::Other("L2 concat reader has no current file reader".to_string())
                })?;
                reader
                    .read_with_physical_rows(block, &mut physical_rows, limit - read_rows)
                    .await?
            };
            if cnt == 0 {
                self.take_current_reader_runtime_stats();
                self.idx += 1;
                if self.idx < self.files.len() {
                    let file = &self.files[self.idx];
                    let mut reader = ColumnarTableReader::new(
                        file,
                        self.schema.clone(),
                        self.filter_op.clone(),
                        self.encryption_key.clone(),
                    )?;
                    reader.seek(&[]).await?;
                    self.reader = Some(reader);
                }
                continue;
            }
            debug_assert_eq!(cnt, physical_rows.len());
            refs.extend(
                physical_rows
                    .into_iter()
                    .map(|physical_row_idx| L2PhysicalRowRef {
                        file_index: self.idx as u32,
                        physical_row_idx,
                    }),
            );
            read_rows += cnt;
        }
        Ok(read_rows)
    }
}

#[async_trait]
impl ColumnarReader for ColumnarConcatReader {
    fn schema(&self) -> &Schema {
        &self.schema
    }

    async fn seek(&mut self, handle: &[u8]) -> table::Result<()> {
        if self.files.is_empty() {
            return Ok(());
        }
        self.take_current_reader_runtime_stats();
        let mut row_key = if get_fixed_size(&self.schema.handle_column) > 0 {
            encode_row_key(self.schema.table_id, (&handle[..]).get_i64_le())
        } else {
            encode_common_handle_row_key(self.schema.table_id, handle)
        };
        if let Some(prefix) = ApiV2::get_keyspace_prefix(self.files[0].get_smallest().deref()) {
            let mut buf = Vec::with_capacity(prefix.len() + row_key.len());
            buf.extend_from_slice(prefix);
            buf.extend_from_slice(&row_key);
            row_key = buf;
        }
        for (i, file) in self.files.iter().enumerate() {
            if file.get_biggest().deref() < row_key.as_slice() {
                continue;
            }
            let mut reader = ColumnarTableReader::new(
                file,
                self.schema.clone(),
                self.filter_op.clone(),
                self.encryption_key.clone(),
            )?;
            reader.seek(handle).await?;
            self.reader = Some(reader);
            self.idx = i;
            return Ok(());
        }
        self.idx = self.files.len();
        Ok(())
    }

    async fn read(&mut self, block: &mut Block, limit: usize) -> table::Result<usize> {
        let mut read_rows = 0;
        while self.idx < self.files.len() && read_rows < limit {
            let cnt = {
                let reader = self.reader.as_mut().unwrap();
                reader.read(block, limit - read_rows).await?
            };
            if cnt == 0 {
                self.take_current_reader_runtime_stats();
                self.idx += 1;
                if self.idx < self.files.len() {
                    let file = &self.files[self.idx];
                    let mut reader = ColumnarTableReader::new(
                        file,
                        self.schema.clone(),
                        self.filter_op.clone(),
                        self.encryption_key.clone(),
                    )?;
                    reader.seek(&[]).await?;
                    self.reader = Some(reader);
                }
                continue;
            }
            read_rows += cnt;
        }
        Ok(read_rows)
    }

    fn get_remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        let mut seg_idents = vec![];
        let mut seg_count = 0;
        for file in &self.files {
            let reader = ColumnarTableReader::new(
                file,
                self.schema.clone(),
                self.filter_op.clone(),
                self.encryption_key.clone(),
            )?;
            let (idents, count) = reader.get_remote_segments()?;
            seg_idents.extend(idents);
            seg_count += count;
        }
        Ok((seg_idents, seg_count))
    }

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

pub(crate) struct ColumnarPackCleanReader {
    schema: Schema,
    files: Vec<ColumnarFile>,
    read_ts: u64,
    encryption_key: Option<EncryptionKey>,
    filter_op: Option<FilterOperator>,
    range_state: Option<PackReadRangeState>,
    current_segment: Option<ColumnarPackReadSegmentKind>,
    clean_reader: Option<ColumnarCleanTableReader>,
    normal_reader: Option<ColumnarPackBoundedMvccReader>,
    runtime_stats: ColumnarRuntimeStats,
}

#[derive(Clone, Copy)]
enum ColumnarPackReadSegmentKind {
    Clean,
    Normal,
}

enum PackReadRange {
    Int {
        start_handle: i64,
        end_handle: Option<i64>,
    },
    Common {
        start_handle: Vec<u8>,
        end_handle: Vec<u8>,
    },
}

impl Clone for PackReadRange {
    fn clone(&self) -> Self {
        match self {
            Self::Int {
                start_handle,
                end_handle,
            } => Self::Int {
                start_handle: *start_handle,
                end_handle: *end_handle,
            },
            Self::Common {
                start_handle,
                end_handle,
            } => Self::Common {
                start_handle: start_handle.clone(),
                end_handle: end_handle.clone(),
            },
        }
    }
}

struct PackReadCursor {
    file_idx: usize,
    pack_idx: usize,
}

struct PackReadRangeState {
    range: PackReadRange,
    cursor: PackReadCursor,
}

impl PackReadRangeState {
    fn reset_cursor(&mut self) {
        self.cursor = PackReadCursor {
            file_idx: 0,
            pack_idx: 0,
        };
    }
}

fn pack_filter_result(
    filter_op: Option<&FilterOperator>,
    table_meta: &TableMeta,
    pack_idx: usize,
) -> FilterOpResult {
    filter_op
        .map(|op| op.rough_check_pack(pack_idx, table_meta))
        .unwrap_or(FilterOpResult::Some)
}

impl ColumnarPackCleanReader {
    pub(crate) fn new(
        files: &[ColumnarFile],
        schema: Schema,
        read_ts: u64,
        encryption_key: Option<EncryptionKey>,
        filter_op: Option<FilterOperator>,
    ) -> Self {
        let mut runtime_stats = ColumnarRuntimeStats::default();
        let files: Vec<ColumnarFile> = files
            .iter()
            .filter(|file| file.has_table(schema.table_id))
            .cloned()
            .collect();
        if let Some(op) = filter_op.as_ref() {
            for file in &files {
                let table_meta = file.get_table(schema.table_id);
                let stats = collect_rough_check_stats(op.rough_check(&table_meta).as_ref());
                runtime_stats.merge(&stats);
            }
        }
        Self {
            schema,
            files,
            read_ts,
            encryption_key,
            filter_op,
            range_state: None,
            current_segment: None,
            clean_reader: None,
            normal_reader: None,
            runtime_stats,
        }
    }

    fn get_remote_segments(&self) -> crate::table::Result<(Vec<FileSegmentIdent>, usize)> {
        let mut seg_idents = vec![];
        let mut seg_count = 0;
        for file in &self.files {
            let table_meta = file.try_get_table(self.schema.table_id)?;
            let file_reader = file.get_file();
            let mut ranges = vec![];
            // Prefetching happens before the caller binds an exact handle range,
            // so we do not yet know whether a later pack will read through the
            // clean path or fall back to the bounded MVCC reader. Always include
            // handle/version segments to avoid regressing mixed clean/normal IA
            // scans, especially for empty projections.
            let (start_off, _) = table_meta.handle_column.pack_offsets.get(0);
            ranges.push((start_off as u64, start_off as u64 + 1));
            let (start_off, _) = table_meta.version_column.pack_offsets.get(0);
            ranges.push((start_off as u64, start_off as u64 + 1));
            for col in &self.schema.columns {
                let col_id = col.get_column_id() as i32;
                if let Some(col_meta) = table_meta.columns.get(&col_id) {
                    let (start_off, _) = col_meta.pack_offsets.get(0);
                    ranges.push((start_off as u64, start_off as u64 + 1));
                }
            }
            let (idents, count) = file_reader.get_remote_segments(&ranges)?;
            seg_idents.extend(idents);
            seg_count += count;
        }
        Ok((seg_idents, seg_count))
    }

    fn pack_end_handle(table_meta: &TableMeta, pack_idx: usize) -> Option<&[u8]> {
        (pack_idx + 1 < table_meta.handle_index.buf.length())
            .then(|| table_meta.handle_index.buf.get_not_null_value(pack_idx + 1))
    }

    fn prepare_clean_segment(
        &mut self,
        file: &ColumnarFile,
        table_meta: Arc<TableMeta>,
        pack_idx: usize,
    ) -> table::Result<ColumnarPackReadSegmentKind> {
        let (row_start, row_end) = table_meta.pack_row_range(pack_idx);
        match self.clean_reader.as_mut() {
            Some(reader) if reader.file_id() == file.id() => {
                reader.reset_range(row_start, row_end);
            }
            Some(reader) => {
                reader.rebind(
                    file,
                    table_meta,
                    row_start,
                    row_end,
                    self.encryption_key.clone(),
                )?;
            }
            None => {
                self.clean_reader = Some(ColumnarCleanTableReader::new(
                    file,
                    table_meta,
                    self.schema.clone(),
                    row_start,
                    row_end,
                    self.encryption_key.clone(),
                )?);
            }
        }
        Ok(ColumnarPackReadSegmentKind::Clean)
    }

    async fn prepare_normal_int_segment(
        &mut self,
        file: &ColumnarFile,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> table::Result<ColumnarPackReadSegmentKind> {
        match self.normal_reader.as_mut() {
            Some(reader) => {
                reader
                    .rebind_int(
                        file,
                        self.read_ts,
                        self.encryption_key.clone(),
                        self.filter_op.clone(),
                        start_handle,
                        end_handle,
                    )
                    .await?;
            }
            None => {
                let mut table_reader = ColumnarTableReader::new(
                    file,
                    self.schema.clone(),
                    self.filter_op.clone(),
                    self.encryption_key.clone(),
                )?;
                table_reader.seek(&start_handle.to_le_bytes()).await?;
                self.normal_reader = Some(ColumnarPackBoundedMvccReader::new_int(
                    table_reader,
                    &self.schema,
                    self.read_ts,
                    end_handle,
                ));
            }
        }
        Ok(ColumnarPackReadSegmentKind::Normal)
    }

    async fn prepare_normal_common_segment(
        &mut self,
        file: &ColumnarFile,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> table::Result<ColumnarPackReadSegmentKind> {
        match self.normal_reader.as_mut() {
            Some(reader) => {
                reader
                    .rebind_common(
                        file,
                        self.read_ts,
                        self.encryption_key.clone(),
                        self.filter_op.clone(),
                        start_handle,
                        end_handle,
                    )
                    .await?;
            }
            None => {
                let mut table_reader = ColumnarTableReader::new(
                    file,
                    self.schema.clone(),
                    self.filter_op.clone(),
                    self.encryption_key.clone(),
                )?;
                table_reader.seek(start_handle).await?;
                self.normal_reader = Some(ColumnarPackBoundedMvccReader::new_common(
                    table_reader,
                    &self.schema,
                    self.read_ts,
                    end_handle,
                ));
            }
        }
        Ok(ColumnarPackReadSegmentKind::Normal)
    }

    fn init_common_range_state(&mut self, start_handle: &[u8], end_handle: &[u8]) {
        self.range_state = Some(PackReadRangeState {
            range: PackReadRange::Common {
                start_handle: start_handle.to_vec(),
                end_handle: end_handle.to_vec(),
            },
            cursor: PackReadCursor {
                file_idx: 0,
                pack_idx: 0,
            },
        });
        self.current_segment = None;
    }

    fn init_int_range_state(&mut self, start_handle: i64, end_handle: Option<i64>) {
        self.range_state = Some(PackReadRangeState {
            range: PackReadRange::Int {
                start_handle,
                end_handle,
            },
            cursor: PackReadCursor {
                file_idx: 0,
                pack_idx: 0,
            },
        });
        self.current_segment = None;
    }

    fn advance_cursor(state: &mut PackReadRangeState, table_meta: &TableMeta) {
        state.cursor.pack_idx += 1;
        if state.cursor.pack_idx >= table_meta.handle_column.pack_offsets.num_packs() {
            state.cursor.file_idx += 1;
            state.cursor.pack_idx = 0;
        }
    }

    async fn build_next_segment(&mut self) -> table::Result<bool> {
        loop {
            let Some(state) = self.range_state.as_ref() else {
                return Ok(false);
            };
            if state.cursor.file_idx >= self.files.len() {
                self.range_state = None;
                return Ok(false);
            }

            let file_idx = state.cursor.file_idx;
            let pack_idx = state.cursor.pack_idx;
            let range = state.range.clone();

            let file = self.files[file_idx].clone();
            let table_meta = file.try_get_table(self.schema.table_id)?;
            if pack_idx >= table_meta.handle_column.pack_offsets.num_packs() {
                let state = self.range_state.as_mut().unwrap();
                state.cursor.file_idx += 1;
                state.cursor.pack_idx = 0;
                continue;
            }

            let maybe_segment = match range {
                PackReadRange::Int {
                    start_handle,
                    end_handle,
                } => {
                    let mut pack_start = table_meta.handle_index.buf.get_not_null_value(pack_idx);
                    let pack_start = pack_start.get_i64_le();
                    let pack_end = Self::pack_end_handle(&table_meta, pack_idx)
                        .map(|mut handle| handle.get_i64_le());
                    if !int_ranges_overlap(start_handle, end_handle, pack_start, pack_end)
                        || pack_filter_result(self.filter_op.as_ref(), &table_meta, pack_idx)
                            == FilterOpResult::None
                    {
                        None
                    } else {
                        let full_pack =
                            int_range_covers_pack(start_handle, end_handle, pack_start, pack_end);
                        if full_pack
                            && table_meta
                                .max_version
                                .unwrap_or_else(|| file.get_max_version())
                                <= self.read_ts
                            && table_meta.is_pack_clean_for_read(pack_idx)
                        {
                            Some(self.prepare_clean_segment(&file, table_meta.clone(), pack_idx)?)
                        } else {
                            let segment_start = start_handle.max(pack_start);
                            let segment_end = min_int_handle_end(end_handle, pack_end);
                            if int_handle_range_is_empty(segment_start, segment_end) {
                                None
                            } else {
                                Some(
                                    self.prepare_normal_int_segment(
                                        &file,
                                        segment_start,
                                        segment_end,
                                    )
                                    .await?,
                                )
                            }
                        }
                    }
                }
                PackReadRange::Common {
                    start_handle,
                    end_handle,
                } => {
                    let pack_start = table_meta.handle_index.buf.get_not_null_value(pack_idx);
                    let pack_end = Self::pack_end_handle(&table_meta, pack_idx)
                        .unwrap_or(GLOBAL_COMMON_HANDLE_END);
                    if start_handle.as_slice() >= pack_end
                        || end_handle.as_slice() <= pack_start
                        || pack_filter_result(self.filter_op.as_ref(), &table_meta, pack_idx)
                            == FilterOpResult::None
                    {
                        None
                    } else {
                        let full_pack = start_handle.as_slice() <= pack_start
                            && pack_end <= end_handle.as_slice();
                        if full_pack
                            && table_meta
                                .max_version
                                .unwrap_or_else(|| file.get_max_version())
                                <= self.read_ts
                            && table_meta.is_pack_clean_for_read(pack_idx)
                        {
                            Some(self.prepare_clean_segment(&file, table_meta.clone(), pack_idx)?)
                        } else {
                            let segment_start = std::cmp::max(start_handle.as_slice(), pack_start);
                            let segment_end = std::cmp::min(end_handle.as_slice(), pack_end);
                            if segment_start >= segment_end {
                                None
                            } else {
                                Some(
                                    self.prepare_normal_common_segment(
                                        &file,
                                        segment_start,
                                        segment_end,
                                    )
                                    .await?,
                                )
                            }
                        }
                    }
                }
            };

            let state = self.range_state.as_mut().unwrap();
            Self::advance_cursor(state, &table_meta);
            if let Some(segment) = maybe_segment {
                self.current_segment = Some(segment);
                return Ok(true);
            }
        }
    }
}

#[async_trait]
impl ColumnarFilterReader for ColumnarPackCleanReader {
    async fn set_handle_range(
        &mut self,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        self.init_common_range_state(start_handle, end_handle);
        Ok(())
    }

    async fn set_int_handle_range(
        &mut self,
        start_handle: i64,
        end_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        self.init_int_range_state(start_handle, end_handle);
        Ok(())
    }

    fn get_schema(&self) -> &Schema {
        &self.schema
    }

    fn reset(&mut self) {
        self.current_segment = None;
        if let Some(state) = self.range_state.as_mut() {
            state.reset_cursor();
        }
    }

    fn get_remote_segment_stats(&self) -> crate::table::Result<(usize, usize)> {
        let (idents, total_segments) = self.get_remote_segments()?;
        Ok((idents.len(), total_segments))
    }

    async fn prefetch_ia_remote_segments(
        &mut self,
        tag: &str,
        ia_mgr: &IaManager,
        keyspace_id: u32,
        timeout: Duration,
    ) -> crate::table::Result<Option<f64>> {
        let (idents, total_segments) = self.get_remote_segments()?;
        prefetch_ia_remote_segments_by_idents(
            tag,
            ia_mgr,
            keyspace_id,
            timeout,
            idents,
            total_segments,
        )
        .await
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize, bool)> {
        block.reset();
        if limit == 0 {
            return Ok((0, true));
        }
        let mut total_read_rows = 0;
        let mut segment_block = Block::new(&self.schema);
        loop {
            if self.current_segment.is_none() && !self.build_next_segment().await? {
                return Ok((total_read_rows, total_read_rows == 0));
            }
            let current_segment = self.current_segment.unwrap();
            let remaining_limit = limit - total_read_rows;
            let (read_rows, drained) = match current_segment {
                ColumnarPackReadSegmentKind::Clean => {
                    self.clean_reader
                        .as_mut()
                        .unwrap()
                        .read(block, remaining_limit)
                        .await?
                }
                ColumnarPackReadSegmentKind::Normal => {
                    segment_block.reset();
                    let (read_rows, drained) = self
                        .normal_reader
                        .as_mut()
                        .unwrap()
                        .try_read_block(&mut segment_block, remaining_limit)
                        .await?;
                    if read_rows > 0 {
                        block.append(&segment_block, 0, read_rows);
                    }
                    (read_rows, drained)
                }
            };
            if read_rows > 0 {
                total_read_rows += read_rows;
                if drained {
                    self.current_segment = None;
                }
                if total_read_rows == limit {
                    return Ok((total_read_rows, false));
                }
                continue;
            }
            if drained {
                self.current_segment = None;
                continue;
            }
            return Ok((total_read_rows, false));
        }
    }

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

struct ColumnarCleanTableReader {
    file_id: u64,
    schema: Schema,
    row_start: usize,
    row_end: usize,
    row_idx: usize,
    columns_readers: Vec<ColumnarColumnReader>,
    inited: bool,
}

struct ColumnarPackBoundedMvccReader {
    src: ColumnarTableReader,
    read_ts: u64,
    filter: ColumnarFilter,
    end_handle: Vec<u8>,
    end_int_handle: Option<i64>,
    prev_int_handle: Option<i64>,
    prev_common_handle: Vec<u8>,
    remaining_src_rows: usize,
    exhausted: bool,
    runtime_stats: ColumnarRuntimeStats,
}

impl ColumnarPackBoundedMvccReader {
    fn new_int(
        src: ColumnarTableReader,
        schema: &Schema,
        read_ts: u64,
        end_int_handle: Option<i64>,
    ) -> Self {
        let remaining_src_rows = src.remaining_rows_in_current_pack();
        Self {
            src,
            read_ts,
            filter: ColumnarFilter::new(schema),
            end_handle: vec![],
            end_int_handle,
            prev_int_handle: None,
            prev_common_handle: vec![],
            remaining_src_rows,
            exhausted: remaining_src_rows == 0,
            runtime_stats: ColumnarRuntimeStats::default(),
        }
    }

    fn new_common(
        src: ColumnarTableReader,
        schema: &Schema,
        read_ts: u64,
        end_handle: &[u8],
    ) -> Self {
        let remaining_src_rows = src.remaining_rows_in_current_pack();
        Self {
            src,
            read_ts,
            filter: ColumnarFilter::new(schema),
            end_handle: end_handle.to_vec(),
            end_int_handle: None,
            prev_int_handle: None,
            prev_common_handle: vec![],
            remaining_src_rows,
            exhausted: remaining_src_rows == 0,
            runtime_stats: ColumnarRuntimeStats::default(),
        }
    }

    fn file_id(&self) -> u64 {
        self.src.file_id()
    }

    fn src_runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.src.runtime_stats();
        stats.clear_rough_check();
        stats
    }

    fn reset_int_state(&mut self, read_ts: u64, end_int_handle: Option<i64>) {
        self.read_ts = read_ts;
        self.filter.clear();
        self.end_handle.clear();
        self.end_int_handle = end_int_handle;
        self.prev_int_handle = None;
        self.prev_common_handle.clear();
        self.remaining_src_rows = self.src.remaining_rows_in_current_pack();
        self.exhausted = self.remaining_src_rows == 0;
    }

    fn reset_common_state(&mut self, read_ts: u64, end_handle: &[u8]) {
        self.read_ts = read_ts;
        self.filter.clear();
        self.end_handle.clear();
        self.end_handle.extend_from_slice(end_handle);
        self.end_int_handle = None;
        self.prev_int_handle = None;
        self.prev_common_handle.clear();
        self.remaining_src_rows = self.src.remaining_rows_in_current_pack();
        self.exhausted = self.remaining_src_rows == 0;
    }

    async fn rebind_int(
        &mut self,
        file: &ColumnarFile,
        read_ts: u64,
        encryption_key: Option<EncryptionKey>,
        filter_op: Option<FilterOperator>,
        start_handle: i64,
        end_int_handle: Option<i64>,
    ) -> crate::table::Result<()> {
        if self.file_id() != file.id() {
            let src_stats = self.src_runtime_stats();
            self.runtime_stats.merge(&src_stats);
            self.src = ColumnarTableReader::new(
                file,
                self.src.schema().clone(),
                filter_op,
                encryption_key,
            )?;
        }
        self.src.seek(&start_handle.to_le_bytes()).await?;
        self.reset_int_state(read_ts, end_int_handle);
        Ok(())
    }

    async fn rebind_common(
        &mut self,
        file: &ColumnarFile,
        read_ts: u64,
        encryption_key: Option<EncryptionKey>,
        filter_op: Option<FilterOperator>,
        start_handle: &[u8],
        end_handle: &[u8],
    ) -> crate::table::Result<()> {
        if self.file_id() != file.id() {
            let src_stats = self.src_runtime_stats();
            self.runtime_stats.merge(&src_stats);
            self.src = ColumnarTableReader::new(
                file,
                self.src.schema().clone(),
                filter_op,
                encryption_key,
            )?;
        }
        self.src.seek(start_handle).await?;
        self.reset_common_state(read_ts, end_handle);
        Ok(())
    }

    async fn try_read_block(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> crate::table::Result<(usize /* read_row */, bool /* drained */)> {
        self.filter.clear();
        block.reset();
        if self.exhausted || limit == 0 || self.remaining_src_rows == 0 {
            self.exhausted = true;
            return Ok((0, true));
        }

        let read_limit = limit.min(self.remaining_src_rows);
        let read_row = self.src.read(block, read_limit).await?;
        self.remaining_src_rows = self.remaining_src_rows.saturating_sub(read_row);
        if read_row == 0 {
            self.exhausted = true;
            return Ok((0, true));
        }
        self.runtime_stats.mvcc_input_rows = self
            .runtime_stats
            .mvcc_input_rows
            .saturating_add(read_row as u64);
        self.runtime_stats.mvcc_input_bytes = self
            .runtime_stats
            .mvcc_input_bytes
            .saturating_add(block.total_byte_size());

        if block.handles.fixed_size > 0 {
            for i in 0..block.handles.length() {
                let handle = block.handles.get_int_handle_value(i);
                let version = block.versions.get_version(i);
                if version > self.read_ts
                    || (self.prev_int_handle.is_some() && handle == self.prev_int_handle.unwrap())
                {
                    self.filter.finish_range(i);
                    continue;
                }
                if block.versions.is_null(i) {
                    self.prev_int_handle = Some(handle);
                    self.filter.finish_range(i);
                    continue;
                }
                if self.end_int_handle.is_some() && handle >= self.end_int_handle.unwrap() {
                    self.exhausted = true;
                    self.remaining_src_rows = 0;
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
                self.prev_int_handle = Some(handle);
            }
        } else {
            let length = block.handles.length();
            let last_handle = block.handles.get_not_null_value(length - 1);
            let check_handle = last_handle >= self.end_handle.as_slice();
            for i in 0..block.handles.length() {
                let handle = block.handles.get_not_null_value(i);
                let version = block.versions.get_version(i);
                if version > self.read_ts || handle == self.prev_common_handle {
                    self.filter.finish_range(i);
                    continue;
                }
                if block.versions.is_null(i) {
                    self.prev_common_handle = handle.to_vec();
                    self.filter.finish_range(i);
                    continue;
                }
                if check_handle && handle >= self.end_handle.as_slice() {
                    self.exhausted = true;
                    self.remaining_src_rows = 0;
                    self.filter.finish_range(i);
                    break;
                }
                self.filter.start_range(i);
                self.prev_common_handle = handle.to_vec();
            }
        }
        if self.remaining_src_rows == 0 {
            self.exhausted = true;
        }
        self.filter.finish_range(read_row);
        let filtered_rows = self.filter.do_filter(read_row, block);
        self.runtime_stats.mvcc_output_rows = self
            .runtime_stats
            .mvcc_output_rows
            .saturating_add(filtered_rows as u64);
        Ok((filtered_rows, self.exhausted))
    }

    fn runtime_stats(&self) -> ColumnarRuntimeStats {
        let mut stats = self.runtime_stats;
        let src_stats = self.src_runtime_stats();
        stats.merge(&src_stats);
        stats
    }
}

impl ColumnarCleanTableReader {
    fn new(
        columnar_file: &ColumnarFile,
        table_meta: Arc<TableMeta>,
        schema: Schema,
        row_start: usize,
        row_end: usize,
        encryption_key: Option<EncryptionKey>,
    ) -> table::Result<Self> {
        let file = columnar_file.get_file();
        let encryption_ver = columnar_file.get_encryption_ver();
        let encryption_fid = columnar_file.encryption_fid();
        let columns_readers = schema
            .columns
            .iter()
            .map(|col| {
                let col_id = col.get_column_id() as i32;
                if let Some(col_meta) = table_meta.columns.get(&col_id) {
                    ColumnarColumnReader::new(
                        file.clone(),
                        col_meta.clone(),
                        false,
                        encryption_key.clone(),
                        encryption_ver,
                        encryption_fid,
                        None,
                    )
                } else {
                    let col_meta = ColumnMeta::new(col.clone(), false);
                    ColumnarColumnReader::new(
                        file.clone(),
                        Arc::new(col_meta),
                        true,
                        encryption_key.clone(),
                        encryption_ver,
                        encryption_fid,
                        None,
                    )
                }
            })
            .collect::<table::Result<Vec<_>>>()?;
        Ok(Self {
            file_id: columnar_file.id(),
            schema,
            row_start,
            row_end,
            row_idx: row_start,
            columns_readers,
            inited: false,
        })
    }

    fn file_id(&self) -> u64 {
        self.file_id
    }

    fn reset_range(&mut self, row_start: usize, row_end: usize) {
        self.row_start = row_start;
        self.row_end = row_end;
        self.row_idx = row_start;
        self.inited = false;
    }

    fn rebind(
        &mut self,
        columnar_file: &ColumnarFile,
        table_meta: Arc<TableMeta>,
        row_start: usize,
        row_end: usize,
        encryption_key: Option<EncryptionKey>,
    ) -> table::Result<()> {
        *self = Self::new(
            columnar_file,
            table_meta,
            self.schema.clone(),
            row_start,
            row_end,
            encryption_key,
        )?;
        Ok(())
    }

    async fn read(
        &mut self,
        block: &mut Block,
        limit: usize,
    ) -> table::Result<(usize /* read_rows */, bool /* drained */)> {
        if self.row_idx >= self.row_end || limit == 0 {
            return Ok((0, true));
        }
        if !self.inited {
            for col_reader in &mut self.columns_readers {
                col_reader.set_row_idx(self.row_start).await?;
            }
            self.inited = true;
        }
        let read_rows = min(limit, self.row_end - self.row_idx);
        append_dummy_handle_versions(&self.schema, block, read_rows);
        for (idx, col_reader) in self.columns_readers.iter_mut().enumerate() {
            let (col_read_rows, col_is_skipped, _) =
                col_reader.read(&mut block.columns[idx], read_rows).await?;
            if col_read_rows != read_rows || col_is_skipped {
                return Err(table::Error::Other(format!(
                    "clean columnar read got unexpected column rows: col_id={}, read={}, expected={}, skipped={}",
                    block.columns[idx].col_id(),
                    col_read_rows,
                    read_rows,
                    col_is_skipped,
                )));
            }
        }
        self.row_idx += read_rows;
        Ok((read_rows, self.row_idx >= self.row_end))
    }
}

fn append_dummy_handle_versions(schema: &Schema, block: &mut Block, rows: usize) {
    if rows == 0 {
        return;
    }
    // These buffers only preserve row-count/alignment for clean packs. They are
    // not real MVCC data and must never be used as observable handle/version
    // output; pack clean read is gated so projected output excludes them.
    let int_zero = 0_i64.to_le_bytes();
    if schema.is_common_handle() {
        block.handles.append_const_value(&[], false, rows);
    } else {
        block.handles.append_const_value(&int_zero, false, rows);
    }
    block
        .versions
        .append_const_value(&0_u64.to_le_bytes(), false, rows);
}

fn int_ranges_overlap(
    start_handle: i64,
    end_handle: Option<i64>,
    pack_start: i64,
    pack_end: Option<i64>,
) -> bool {
    let start_before_pack_end = pack_end.map(|end| start_handle < end).unwrap_or(true);
    let end_after_pack_start = end_handle.map(|end| end > pack_start).unwrap_or(true);
    start_before_pack_end && end_after_pack_start
}

fn int_range_covers_pack(
    start_handle: i64,
    end_handle: Option<i64>,
    pack_start: i64,
    pack_end: Option<i64>,
) -> bool {
    start_handle <= pack_start
        && match (pack_end, end_handle) {
            (_, None) => true,
            (Some(pack_end), Some(end_handle)) => pack_end <= end_handle,
            (None, Some(_)) => false,
        }
}

fn min_int_handle_end(lhs: Option<i64>, rhs: Option<i64>) -> Option<i64> {
    match (lhs, rhs) {
        (Some(lhs), Some(rhs)) => Some(lhs.min(rhs)),
        (Some(lhs), None) => Some(lhs),
        (None, Some(rhs)) => Some(rhs),
        (None, None) => None,
    }
}

fn int_handle_range_is_empty(start_handle: i64, end_handle: Option<i64>) -> bool {
    end_handle.map(|end| start_handle >= end).unwrap_or(false)
}

pub fn decode_decimal_as_int(col_info: &ColumnInfo, decimal: &Decimal) -> Vec<u8> {
    let mut buf = vec![];
    let mut decimal_str = decimal.to_string();
    decimal_str.retain(|c| c != '.');
    let prec = col_info.get_column_len();
    if prec <= 9 {
        let val = decimal_str.parse::<i32>().unwrap_or_default();
        buf.extend_from_slice(&val.to_le_bytes());
    } else if prec <= 18 {
        let val = decimal_str.parse::<i64>().unwrap_or_default();
        buf.extend_from_slice(&val.to_le_bytes());
    } else if prec <= 38 {
        let val = decimal_str.parse::<i128>().unwrap_or_default();
        buf.extend_from_slice(&val.to_le_bytes());
    } else if prec <= 65 {
        let val = decimal_str.parse::<i256>().unwrap_or_default();
        buf.extend_from_slice(&val.to_le_bytes());
    } else {
        panic!("unsupported precision: {}", prec);
    }
    buf
}

#[cfg(test)]
pub use tests::MockColumnarFilterReader;

#[cfg(test)]
pub mod tests {
    use std::{
        collections::VecDeque,
        sync::{Arc, Mutex},
    };

    use async_trait::async_trait;
    use bytes::Bytes;
    use codec::number::NumberEncoder;
    use futures::executor::block_on;
    use proptest::{arbitrary::any, proptest};
    use rand::Rng;
    use rstest::rstest;
    use schema::schema::{StorageClass, StorageClassSpec};
    use test_util::init_log_for_test;
    use tidb_query_datatype::{
        Collation::Utf8Mb4GeneralCi,
        FieldTypeAccessor, FieldTypeTp,
        codec::{
            mysql::DecimalEncoder,
            row::v2::encoder_for_test::{Column, RowEncoder},
        },
        expr::EvalContext,
    };

    use super::*;
    use crate::{
        UserMeta, WRITE_CF,
        table::{
            columnar::{
                ColumnarMetaCache, PACK_HAS_DELETE, PACK_HAS_MULTI_VERSION,
                builder::{
                    ColumnarFileBuilder, ColumnarTableBuildOptions, ColumnarTableBuilder,
                    new_common_handle_column_info, new_int_handle_column_info,
                    new_version_column_info,
                },
                columnar::ColumnarFile,
                filter::TableScanCtx,
                reader::{ColumnarMvccReader, ColumnarReader, ColumnarTableReader},
            },
            file::{File, FileMmapGuard, InMemFile, MmapData},
            memtable::{CfTable, WriteBatch},
            schema_file::{SchemaBuf, SchemaBufBuilder},
        },
    };

    /// A simple `ColumnarFilterReader` implementation for tests that returns a
    /// prebuilt block once.
    pub struct MockColumnarFilterReader {
        schema: Schema,
        block: Option<Block>,
    }

    impl MockColumnarFilterReader {
        pub fn new(schema: Schema, block: Block) -> Self {
            Self {
                schema,
                block: Some(block),
            }
        }
    }

    #[async_trait]
    impl ColumnarFilterReader for MockColumnarFilterReader {
        async fn set_handle_range(
            &mut self,
            _start_handle: &[u8],
            _end_handle: &[u8],
        ) -> crate::table::Result<()> {
            Ok(())
        }

        async fn set_int_handle_range(
            &mut self,
            _start_handle: i64,
            _end_handle: Option<i64>,
        ) -> crate::table::Result<()> {
            Ok(())
        }

        fn get_schema(&self) -> &Schema {
            &self.schema
        }

        async fn try_read_block(
            &mut self,
            block: &mut Block,
            _limit: usize,
        ) -> crate::table::Result<(usize, bool)> {
            block.reset();
            if let Some(data) = self.block.take() {
                let rows = data.length();
                block.append(&data, 0, rows);
                Ok((rows, true))
            } else {
                Ok((0, true))
            }
        }
    }

    struct MockColumnarReader {
        schema: Schema,
        blocks: VecDeque<Block>,
        read_calls: Arc<Mutex<usize>>,
    }

    impl MockColumnarReader {
        fn new(schema: Schema, blocks: Vec<Block>, read_calls: Arc<Mutex<usize>>) -> Self {
            Self {
                schema,
                blocks: blocks.into(),
                read_calls,
            }
        }
    }

    #[async_trait]
    impl ColumnarReader for MockColumnarReader {
        fn schema(&self) -> &Schema {
            &self.schema
        }

        async fn seek(&mut self, _handle: &[u8]) -> crate::table::Result<()> {
            Ok(())
        }

        async fn read(&mut self, block: &mut Block, _limit: usize) -> crate::table::Result<usize> {
            *self.read_calls.lock().unwrap() += 1;
            block.reset();
            if let Some(data) = self.blocks.pop_front() {
                let rows = data.length();
                block.append(&data, 0, rows);
                Ok(rows)
            } else {
                Ok(0)
            }
        }
    }

    #[derive(Default, Clone)]
    pub struct RefRow {
        handle: Vec<u8>,
        is_common_handle: bool,
        version: u64,
        txn_id: u64,
        is_deleted: bool,
        c0: Option<i64>,
        c1: Option<Vec<u8>>,
    }

    impl RefRow {
        pub fn in_bound(&self, mut start: &[u8], mut end: &[u8]) -> bool {
            if self.is_common_handle {
                self.handle.as_slice() >= start && self.handle.as_slice() < end
            } else {
                let handle = self.handle.as_slice().get_i64_le();
                handle >= start.get_i64_le() && handle < end.get_i64_le()
            }
        }

        pub fn cmp(&self, other: &Self) -> Ordering {
            let order = if self.is_common_handle {
                self.handle.cmp(&other.handle)
            } else {
                self.handle
                    .as_slice()
                    .get_i64_le()
                    .cmp(&other.handle.as_slice().get_i64_le())
            };
            if order.is_eq() {
                other.version.cmp(&self.version)
            } else {
                order
            }
        }
    }

    fn new_column_info(col_id: i64) -> ColumnInfo {
        let mut col_info = ColumnInfo::new();
        col_info.set_column_id(col_id);
        col_info
    }

    pub fn new_schema(table_id: i64, common_handle: bool) -> Schema {
        new_schema_with_nullable(table_id, common_handle, true)
    }

    pub fn new_schema_with_nullable(table_id: i64, common_handle: bool, nullable: bool) -> Schema {
        let handle_column = if common_handle {
            new_common_handle_column_info()
        } else {
            new_int_handle_column_info()
        };
        let version_column = new_version_column_info();
        let mut col_1 = new_column_info(1);
        col_1.set_tp(FieldTypeTp::LongLong as i32);
        let mut col_2 = new_column_info(2);
        col_2.set_tp(FieldTypeTp::VarChar as i32);
        col_2.set_collation(Utf8Mb4GeneralCi as i32);
        if !nullable {
            col_2.set_flag(FieldTypeFlag::NOT_NULL.bits() as i32);
        }
        let columns = vec![col_1, col_2];
        Schema::new(SchemaBuf::new(
            table_id,
            handle_column,
            version_column,
            columns,
            vec![],
            0,
            vec![],
            vec![],
            StorageClassSpec::default(),
            None,
        ))
    }

    fn new_schema_two_columns_not_null(table_id: i64, common_handle: bool) -> Schema {
        let handle_column = if common_handle {
            new_common_handle_column_info()
        } else {
            new_int_handle_column_info()
        };
        let version_column = new_version_column_info();

        let mut col_1 = new_column_info(1);
        col_1.set_tp(FieldTypeTp::LongLong as i32);
        let mut col_2 = new_column_info(2);
        col_2.set_tp(FieldTypeTp::VarChar as i32);
        col_2.set_collation(Utf8Mb4GeneralCi as i32);
        col_2.set_flag(FieldTypeFlag::NOT_NULL.bits() as i32);

        let mut schema_buf_builder = SchemaBufBuilder::new(table_id);
        schema_buf_builder.columns(
            handle_column,
            version_column,
            vec![col_1, col_2],
            vec![],
            // max_col_id
            2,
            vec![],
            vec![],
        );
        Schema::new(schema_buf_builder.build())
    }

    #[derive(Clone)]
    struct CountingFile {
        inner: Arc<dyn File>,
        reads: Arc<Mutex<Vec<(u64, usize)>>>,
        remote_segments: Arc<Mutex<Vec<FileSegmentIdent>>>,
        total_segments: usize,
    }

    impl CountingFile {
        fn new(inner: Arc<dyn File>) -> Self {
            Self {
                inner,
                reads: Arc::new(Mutex::new(vec![])),
                remote_segments: Arc::new(Mutex::new(vec![])),
                total_segments: 0,
            }
        }

        fn with_remote_segments(
            inner: Arc<dyn File>,
            remote_segments: Vec<FileSegmentIdent>,
            total_segments: usize,
        ) -> Self {
            Self {
                inner,
                reads: Arc::new(Mutex::new(vec![])),
                remote_segments: Arc::new(Mutex::new(remote_segments)),
                total_segments,
            }
        }

        fn clear_reads(&self) {
            self.reads.lock().unwrap().clear();
        }

        fn reads(&self) -> Vec<(u64, usize)> {
            self.reads.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl File for CountingFile {
        fn id(&self) -> u64 {
            self.inner.id()
        }

        fn size(&self) -> u64 {
            self.inner.size()
        }

        fn is_sync(&self) -> bool {
            self.inner.is_sync()
        }

        fn read(&self, off: u64, length: usize) -> table::Result<Bytes> {
            self.inner.read(off, length)
        }

        fn read_at(&self, buf: &mut [u8], offset: u64) -> table::Result<()> {
            self.reads.lock().unwrap().push((offset, buf.len()));
            self.inner.read_at(buf, offset)
        }

        fn read_table_meta(&self, off: u64, length: usize) -> table::Result<Bytes> {
            self.inner.read_table_meta(off, length)
        }

        fn mmap(&self) -> table::Result<MmapData> {
            self.inner.mmap()
        }

        fn mmap2(&self) -> table::Result<Bytes> {
            self.inner.mmap2()
        }

        async fn mmap_range(
            &self,
            offset: u64,
            length: usize,
        ) -> table::Result<(Bytes, FileMmapGuard)> {
            self.inner.mmap_range(offset, length).await
        }

        fn get_remote_segments(
            &self,
            _ranges: &[(u64 /* start_off */, u64 /* end_off */)],
        ) -> table::Result<(Vec<FileSegmentIdent>, usize)> {
            Ok((
                self.remote_segments.lock().unwrap().clone(),
                self.total_segments,
            ))
        }

        fn storage_class(&self) -> StorageClass {
            self.inner.storage_class()
        }

        fn as_any(self: Arc<Self>) -> Arc<dyn std::any::Any + Send + Sync> {
            self
        }
    }

    fn encode_test_row(columns: Vec<Column>) -> Vec<u8> {
        let mut buf = vec![];
        buf.write_row(&mut EvalContext::default(), columns).unwrap();
        buf
    }

    fn put_mvcc_row(wb: &mut WriteBatch, row_key: &[u8], txn_id: u64, version: u64, value: &[u8]) {
        let meta = UserMeta::new(txn_id, version).to_array();
        wb.put(InnerKey::from_inner_buf(row_key), 0, &meta, version, value);
    }

    pub fn build_table(
        file_id: u64,
        schema: &Schema,
        start: i32,
        end: i32,
        version: u64,
    ) -> (Arc<dyn File>, Vec<RefRow>) {
        let (tbl, mut refs) = build_table_with_encryption(
            file_id,
            std::slice::from_ref(schema),
            start,
            end,
            version,
            false,
            None,
        );
        (tbl, refs.pop().unwrap())
    }

    pub fn build_table_with_all_rows_deleted(
        file_id: u64,
        schema: &Schema,
        start: i32,
        end: i32,
        version: u64,
    ) -> (Arc<dyn File>, Vec<RefRow>) {
        let (tbl, mut refs) = build_table_with_encryption(
            file_id,
            std::slice::from_ref(schema),
            start,
            end,
            version,
            true,
            None,
        );
        (tbl, refs.pop().unwrap())
    }

    pub fn build_mixed_table(
        file_id: u64,
        schemas: &[Schema],
        start: i32,
        end: i32,
        version: u64,
    ) -> (Arc<dyn File>, Vec<Vec<RefRow>>) {
        build_table_with_encryption(file_id, schemas, start, end, version, false, None)
    }

    fn new_test_encryption_key() -> EncryptionKey {
        EncryptionKey::new(
            b"cipher".to_vec(),
            b"plain".to_vec(),
            cloud_encryption::KEY_TYPE_AES_256_CTR_LEGACY,
        )
    }

    pub fn build_table_with_encryption(
        file_id: u64,
        schemas: &[Schema],
        start: i32,
        end: i32,
        version: u64,
        all_deleted: bool, // generate all deleted rows
        encryption_key: Option<EncryptionKey>,
    ) -> (Arc<dyn File>, Vec<Vec<RefRow>>) {
        let mut rng = rand::thread_rng();
        let mut file_builder = ColumnarFileBuilder::new(
            file_id,
            Some(version.into()), // use version as snap_version for test
            encryption_key.clone(),
        );
        let cf_tbl = CfTable::new();
        let mut tables_ref_rows = vec![];
        for schema in schemas {
            let mut ref_rows = vec![];
            let is_common_handle = get_fixed_size(&schema.handle_column) == 0;
            for i in start..end {
                let mut ref_row = new_ref_row(i, version, is_common_handle);
                if all_deleted {
                    ref_row.is_deleted = true;
                }
                ref_rows.push(ref_row);
                if rng.gen_ratio(1, 8) {
                    let mut ref_row = new_ref_row(i, version - 1, is_common_handle);
                    if all_deleted {
                        ref_row.is_deleted = true;
                    }
                    ref_rows.push(ref_row);
                }
                if rng.gen_ratio(1, 16) {
                    let mut ref_row = new_ref_row(i, version - 2, is_common_handle);
                    if all_deleted {
                        ref_row.is_deleted = true;
                    }
                    ref_rows.push(ref_row);
                }
            }
            let mut eval_ctx = EvalContext::default();
            let mut wb = WriteBatch::new();
            for ref_row in ref_rows.iter().rev() {
                let row_key = if is_common_handle {
                    encode_common_handle_row_key(schema.table_id, ref_row.handle.as_slice())
                } else {
                    encode_row_key(schema.table_id, ref_row.handle.as_slice().get_i64_le())
                };
                let inner_key = InnerKey::from_inner_buf(&row_key);
                let mut row_val = vec![];
                let cols = vec![
                    Column::new(1, ref_row.c0),
                    Column::new(2, ref_row.c1.clone()),
                ];
                if !ref_row.is_deleted {
                    row_val.write_row(&mut eval_ctx, cols).unwrap();
                }
                let user_meta = UserMeta::new(ref_row.txn_id, ref_row.version).to_array();
                wb.put(inner_key, 0, &user_meta, ref_row.version, &row_val);
            }
            cf_tbl.get_cf(WRITE_CF).put_batch(&mut wb, None, WRITE_CF);
            tables_ref_rows.push(ref_rows);
        }
        for (i, schema) in schemas.iter().enumerate() {
            let iter = cf_tbl.get_cf(WRITE_CF).new_iterator(false);
            let mut row_tbl_reader = ColumnarRowTableReader::new(
                schema.clone(),
                iter,
                None,
                false,
                encryption_key.clone(),
            )
            .unwrap();
            let mut block = Block::new(schema);
            let mut opts = ColumnarTableBuildOptions::default();
            opts.pack_max_row_count = 8;
            opts.pack_max_size = 256;
            let mut table_builder =
                ColumnarTableBuilder::new(schema.clone(), opts, encryption_key.clone(), file_id, 0);
            block_on(row_tbl_reader.seek(&tables_ref_rows[i][0].handle)).unwrap();
            let mut append_rows = 0;
            let mut block_off = 0;
            while append_rows < tables_ref_rows[i].len() {
                if block_off == block.length() {
                    block_off = 0;
                    block.reset();
                    let limit = rng.gen_range(2..10);
                    let read = block_on(row_tbl_reader.read(&mut block, limit)).unwrap();
                    if read == 0 {
                        break;
                    }
                }
                let block_off_before = block_off;
                block_off = table_builder.append_block(&block, block_off);
                append_rows += block_off - block_off_before;
            }
            file_builder.add_table(table_builder);
        }

        let (file_data, _) = file_builder.build();
        (
            Arc::new(InMemFile::new(file_id, file_data.into())),
            tables_ref_rows,
        )
    }

    fn new_ref_row(i: i32, version: u64, is_common_handle: bool) -> RefRow {
        let mut ref_row = RefRow::default();
        ref_row.is_common_handle = is_common_handle;
        if is_common_handle {
            ref_row.handle = i_to_common_handle(i);
        } else {
            ref_row.handle = (i as i64).to_le_bytes().to_vec();
        }
        ref_row.version = version;
        ref_row.txn_id = version - 1;
        let mut rng = rand::thread_rng();
        let is_delete = rng.gen_ratio(1, 10);
        ref_row.is_deleted = is_delete;
        let is_null = is_delete || rng.gen_ratio(1, 5);
        if !is_null {
            ref_row.c0 = Some(i as i64);
            ref_row.c1 = Some(i_to_string_col(i, version));
        }
        ref_row
    }

    fn i_to_string_col(i: i32, version: u64) -> Vec<u8> {
        let repeat = 1 + i % 16;
        format!("abc_{}_{}", i, version)
            .repeat(repeat as usize)
            .into_bytes()
    }

    pub fn i_to_common_handle(i: i32) -> Vec<u8> {
        format!("{:08x}", i).into_bytes()
    }

    fn build_columnar_file_from_block(
        file_id: u64,
        schema: &Schema,
        block: &Block,
        pack_max_row_count: usize,
        level: u32,
    ) -> Arc<dyn File> {
        let mut file_builder = ColumnarFileBuilder::new(file_id, Some(100.into()), None);
        let mut opts = ColumnarTableBuildOptions::default();
        opts.pack_max_row_count = pack_max_row_count;
        opts.pack_max_size = 1024 * 1024;
        let mut table_builder =
            ColumnarTableBuilder::new(schema.clone(), opts, None, file_id, level);
        let mut offset = 0;
        while offset < block.length() {
            let next_offset = table_builder.append_block(block, offset);
            assert!(next_offset > offset);
            offset = next_offset;
        }
        file_builder.add_table(table_builder);
        let (file_data, _) = file_builder.build();
        Arc::new(InMemFile::new(file_id, file_data.into()))
    }

    fn build_columnar_file_from_block_with_options(
        file_id: u64,
        schema: &Schema,
        block: &Block,
        opts: ColumnarTableBuildOptions,
        level: u32,
    ) -> Arc<dyn File> {
        let mut file_builder = ColumnarFileBuilder::new(file_id, Some(100.into()), None);
        let mut table_builder =
            ColumnarTableBuilder::new(schema.clone(), opts, None, file_id, level);
        let mut offset = 0;
        while offset < block.length() {
            let next_offset = table_builder.append_block(block, offset);
            assert!(next_offset > offset);
            offset = next_offset;
        }
        file_builder.add_table(table_builder);
        let (file_data, _) = file_builder.build();
        Arc::new(InMemFile::new(file_id, file_data.into()))
    }

    fn build_clean_block(schema: &Schema, start: i32, end: i32, version: u64) -> Block {
        let mut block = Block::new(schema);
        for i in start..end {
            if schema.is_common_handle() {
                block.handles.push_value(&i_to_common_handle(i));
            } else {
                block.handles.push_value(&(i as i64).to_le_bytes());
            }
            block.versions.push_version(version, false);
            block.columns[0].push_value(&(i as i64).to_le_bytes());
            block.columns[1].push_value(format!("v{i}").as_bytes());
        }
        block
    }

    fn column_pack_ranges(col_meta: &ColumnMeta) -> Vec<(u64, u64)> {
        (0..col_meta.pack_offsets.num_packs())
            .map(|pack_idx| {
                let ((start, _), (end, _)) = col_meta.get_pack_offset(pack_idx);
                (start as u64, end as u64)
            })
            .collect()
    }

    fn read_overlaps(read: (u64, usize), ranges: &[(u64, u64)]) -> bool {
        let (read_start, len) = read;
        let read_end = read_start + len as u64;
        ranges
            .iter()
            .any(|&(start, end)| read_start < end && start < read_end)
    }

    fn assert_handle_version_read(
        reads: &[(u64, usize)],
        handle_ranges: &[(u64, u64)],
        version_ranges: &[(u64, u64)],
        expected: bool,
    ) {
        let got = reads
            .iter()
            .any(|&read| read_overlaps(read, handle_ranges) || read_overlaps(read, version_ranges));
        assert_eq!(got, expected, "reads: {reads:?}");
    }

    fn build_less_than_filter(schema: &Schema, col_id: i64, val: i64) -> FilterOperator {
        let col_offset = schema
            .columns
            .iter()
            .position(|col| col.get_column_id() == col_id)
            .unwrap() as i64;

        let mut col = tipb::Expr::default();
        col.set_tp(tipb::ExprType::ColumnRef);
        col.mut_val().write_i64(col_offset).unwrap();
        col.mut_field_type()
            .as_mut_accessor()
            .set_tp(FieldTypeTp::LongLong);

        let mut value = tipb::Expr::default();
        value.set_tp(tipb::ExprType::Int64);
        let mut buf = Vec::with_capacity(8);
        buf.write_i64(val).unwrap();
        value.set_val(buf);
        value
            .mut_field_type()
            .as_mut_accessor()
            .set_tp(FieldTypeTp::LongLong);

        let mut cond = tipb::Expr::default();
        cond.set_tp(tipb::ExprType::ScalarFunc);
        cond.set_sig(tipb::ScalarFuncSig::LtInt);
        cond.mut_field_type()
            .as_mut_accessor()
            .set_tp(FieldTypeTp::LongLong);
        cond.mut_children().push(col);
        cond.mut_children().push(value);

        let mut table_scan = tipb::Executor::default();
        table_scan
            .mut_tbl_scan()
            .set_columns(schema.columns.clone().into());
        TableScanCtx::new(table_scan, vec![cond]).to_filter_operator()
    }

    #[test]
    fn test_columnar_builder() {
        init_log_for_test();
        for common_handle in [true, false] {
            let columnar_meta_cache = ColumnarMetaCache::default();
            let schema = new_schema(1, common_handle);
            let (file, ref_rows) = build_table(1, &schema, 100, 150, 100);
            let columnar_file =
                ColumnarFile::open(file, None, columnar_meta_cache.clone()).unwrap();
            let mut reader =
                ColumnarTableReader::new(&columnar_file, schema.clone(), None, None).unwrap();
            block_on(reader.seek(&0u64.to_le_bytes())).unwrap();
            let mut block = Block::new(&schema);
            block_on(reader.read(&mut block, 100)).unwrap();
            verify_with_ref_rows(&block, &ref_rows);
        }
    }

    #[test]
    fn test_encrypted_columnar_uses_origin_fid_after_file_id_remap() {
        const ORIGIN_FILE_ID: u64 = 10;
        const REMAPPED_FILE_ID: u64 = 20;

        init_log_for_test();
        let schema = new_schema(1, false);
        let encryption_key = new_test_encryption_key();
        let (file, mut ref_rows) = build_table_with_encryption(
            ORIGIN_FILE_ID,
            std::slice::from_ref(&schema),
            100,
            150,
            100,
            false,
            Some(encryption_key.clone()),
        );
        let file_data = file.read_all().unwrap();
        let remapped_file: Arc<dyn File> = Arc::new(InMemFile::new(REMAPPED_FILE_ID, file_data));
        let columnar_file =
            ColumnarFile::open(remapped_file, None, ColumnarMetaCache::default()).unwrap();

        assert_eq!(columnar_file.id(), REMAPPED_FILE_ID);
        assert_eq!(columnar_file.get_origin_fid(), Some(ORIGIN_FILE_ID));

        let mut reader =
            ColumnarTableReader::new(&columnar_file, schema.clone(), None, Some(encryption_key))
                .unwrap();
        block_on(reader.seek(&0u64.to_le_bytes())).unwrap();
        let mut block = Block::new(&schema);
        block_on(reader.read(&mut block, 100)).unwrap();
        verify_with_ref_rows(&block, &ref_rows.pop().unwrap());
    }

    #[test]
    fn test_pack_mvcc_stats_written_and_parsed() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 10, 100);
        let file = build_columnar_file_from_block(10, &schema, &block, 4, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let stats = table_meta.pack_mvcc_stats.as_ref().unwrap();
        assert_eq!(stats.flags(), &[0, 0, 0]);
        assert_eq!(table_meta.max_version, Some(100));
    }

    #[test]
    fn test_pack_mvcc_stats_marks_delete_and_multi_version() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let mut block = Block::new(&schema);
        for (handle, version, is_delete) in
            [(1_i64, 100_u64, false), (1, 90, false), (2, 100, true)]
        {
            block.handles.push_value(&handle.to_le_bytes());
            block.versions.push_version(version, is_delete);
            block.columns[0].push_value(&handle.to_le_bytes());
            if is_delete {
                block.columns[1].push_null();
            } else {
                block.columns[1].push_value(format!("v{handle}_{version}").as_bytes());
            }
        }
        let file = build_columnar_file_from_block(11, &schema, &block, 8, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let stats = table_meta.pack_mvcc_stats.as_ref().unwrap();
        assert_eq!(stats.flags(), &[PACK_HAS_DELETE | PACK_HAS_MULTI_VERSION]);
    }

    #[rstest]
    #[case::int_handle(false)]
    #[case::common_handle(true)]
    fn test_pack_clean_reader_skips_handle_and_version_packs(#[case] common_handle: bool) {
        init_log_for_test();
        let schema = new_schema(1, common_handle);
        let block = build_clean_block(&schema, 0, 12, 100);
        let file = build_columnar_file_from_block(12, &schema, &block, 4, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let handle_ranges = column_pack_ranges(&table_meta.handle_column);
        let version_ranges = column_pack_ranges(&table_meta.version_column);
        counting_file.clear_reads();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_unbounded_handle_range()).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 12);
        for i in 0..12 {
            assert_eq!(out.columns[0].get_value(i).unwrap().get_i64_le(), i as i64);
            assert_eq!(
                out.columns[1].get_value(i).unwrap(),
                format!("v{i}").as_bytes()
            );
        }
        assert_handle_version_read(
            &counting_file.reads(),
            &handle_ranges,
            &version_ranges,
            false,
        );
    }

    #[test]
    fn test_pack_clean_reader_applies_filter_pack_pruning() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 12, 100);
        let file = build_columnar_file_from_block(16, &schema, &block, 4, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let handle_ranges = column_pack_ranges(&table_meta.handle_column);
        let version_ranges = column_pack_ranges(&table_meta.version_column);
        let filter_op = build_less_than_filter(&schema, 1, 4);
        counting_file.clear_reads();

        let mut reader = ColumnarPackCleanReader::new(
            &[columnar_file],
            schema.clone(),
            200,
            None,
            Some(filter_op),
        );
        block_on(reader.set_unbounded_handle_range()).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 4);
        for i in 0..4 {
            assert_eq!(out.columns[0].get_value(i).unwrap().get_i64_le(), i as i64);
            assert_eq!(
                out.columns[1].get_value(i).unwrap(),
                format!("v{i}").as_bytes()
            );
        }
        assert_handle_version_read(
            &counting_file.reads(),
            &handle_ranges,
            &version_ranges,
            false,
        );
    }

    #[test]
    fn test_pack_clean_reader_collects_runtime_stats() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 12, 100);
        let file = build_columnar_file_from_block(21, &schema, &block, 4, 2);
        let remote_segments = vec![
            FileSegmentIdent::new(21, 0, 64),
            FileSegmentIdent::new(21, 64, 128),
        ];
        let counting_file = CountingFile::with_remote_segments(file, remote_segments, 5);
        let columnar_file =
            ColumnarFile::open(Arc::new(counting_file), None, ColumnarMetaCache::default())
                .unwrap();
        let filter_op = build_less_than_filter(&schema, 1, 4);
        let mut reader = ColumnarPackCleanReader::new(
            &[columnar_file],
            schema.clone(),
            200,
            None,
            Some(filter_op),
        );
        block_on(reader.set_unbounded_handle_range()).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 4);

        let remote_stats = reader.get_remote_segment_stats().unwrap();
        assert_eq!(remote_stats, (2, 5));

        let stats = reader.runtime_stats();
        assert_eq!(stats.rough_check_total_packs, 3);
        assert_eq!(stats.rough_check_selected_packs, 1);
        assert_eq!(stats.rough_check_skipped_packs, 2);
        assert_eq!(stats.rough_check_unknown_packs, 0);

        let stats_again = reader.runtime_stats();
        assert_eq!(stats_again, stats);
    }

    #[test]
    fn test_pack_clean_reader_falls_back_when_read_ts_is_too_old() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 4, 100);
        let file = build_columnar_file_from_block(13, &schema, &block, 4, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let handle_ranges = column_pack_ranges(&table_meta.handle_column);
        let version_ranges = column_pack_ranges(&table_meta.version_column);
        counting_file.clear_reads();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 50, None, None);
        block_on(reader.set_unbounded_handle_range()).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 0);
        assert_handle_version_read(
            &counting_file.reads(),
            &handle_ranges,
            &version_ranges,
            true,
        );
    }

    #[test]
    fn test_pack_clean_reader_falls_back_for_dirty_pack() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let mut block = Block::new(&schema);
        for (handle, version, is_delete, value) in [
            (1_i64, 100_u64, false, Some(b"new".as_slice())),
            (1, 90, false, Some(b"old".as_slice())),
            (2, 100, true, None),
            (3, 100, false, Some(b"live".as_slice())),
        ] {
            block.handles.push_value(&handle.to_le_bytes());
            block.versions.push_version(version, is_delete);
            block.columns[0].push_value(&handle.to_le_bytes());
            if let Some(value) = value {
                block.columns[1].push_value(value);
            } else {
                block.columns[1].push_null();
            }
        }
        let file = build_columnar_file_from_block(14, &schema, &block, 8, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let handle_ranges = column_pack_ranges(&table_meta.handle_column);
        let version_ranges = column_pack_ranges(&table_meta.version_column);
        counting_file.clear_reads();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_unbounded_handle_range()).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 2);
        assert_eq!(out.handles.get_int_handle_value(0), 1);
        assert_eq!(out.columns[1].get_value(0).unwrap(), b"new");
        assert_eq!(out.handles.get_int_handle_value(1), 3);
        assert_eq!(out.columns[1].get_value(1).unwrap(), b"live");
        assert_handle_version_read(
            &counting_file.reads(),
            &handle_ranges,
            &version_ranges,
            true,
        );

        let stats = reader.runtime_stats();
        assert_eq!(stats.mvcc_input_rows, 4);
        assert!(stats.mvcc_input_bytes > 0);
        assert_eq!(stats.mvcc_output_rows, 2);
        assert_eq!(stats.rough_check_total_packs, 0);
        assert_eq!(stats.rough_check_selected_packs, 0);
        assert_eq!(stats.rough_check_skipped_packs, 0);
        assert_eq!(stats.rough_check_unknown_packs, 0);
    }

    #[test]
    fn test_concat_reader_accumulates_runtime_stats_across_files() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let filter_op = build_less_than_filter(&schema, 1, 8);

        let file_1 = build_columnar_file_from_block(
            31,
            &schema,
            &build_clean_block(&schema, 0, 4, 100),
            2,
            2,
        );
        let file_2 = build_columnar_file_from_block(
            32,
            &schema,
            &build_clean_block(&schema, 4, 8, 100),
            2,
            2,
        );
        let columnar_1 = ColumnarFile::open(file_1, None, ColumnarMetaCache::default()).unwrap();
        let columnar_2 = ColumnarFile::open(file_2, None, ColumnarMetaCache::default()).unwrap();

        let mut reader = ColumnarConcatReader::new(
            &[columnar_1, columnar_2],
            schema.clone(),
            Some(filter_op),
            None,
        );
        block_on(reader.seek(&0_i64.to_le_bytes())).unwrap();

        let mut block = Block::new(&schema);
        let read_rows = block_on(reader.read(&mut block, 16)).unwrap();
        assert_eq!(read_rows, 8);
        assert_eq!(block.length(), 8);

        let stats = reader.runtime_stats();
        assert_eq!(stats.rough_check_total_packs, 4);
        assert_eq!(stats.rough_check_selected_packs, 4);
        assert_eq!(stats.rough_check_skipped_packs, 0);
        assert_eq!(stats.rough_check_unknown_packs, 0);
    }

    #[test]
    fn test_pack_clean_reader_falls_back_for_boundary_pack() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 4, 100);
        let file = build_columnar_file_from_block(15, &schema, &block, 4, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        let handle_ranges = column_pack_ranges(&table_meta.handle_column);
        let version_ranges = column_pack_ranges(&table_meta.version_column);
        counting_file.clear_reads();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_int_handle_range(1, Some(3))).unwrap();
        let out = block_on(reader.read_all());

        assert_eq!(out.length(), 2);
        assert_eq!(out.handles.get_int_handle_value(0), 1);
        assert_eq!(out.handles.get_int_handle_value(1), 2);
        assert_handle_version_read(
            &counting_file.reads(),
            &handle_ranges,
            &version_ranges,
            true,
        );
    }

    #[test]
    fn test_pack_clean_reader_builds_segments_lazily_and_resets_range() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 8, 100);
        let file = build_columnar_file_from_block(18, &schema, &block, 4, 2);
        let counting_file = CountingFile::new(file);
        let columnar_file = ColumnarFile::open(
            Arc::new(counting_file.clone()),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        counting_file.clear_reads();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_int_handle_range(1, None)).unwrap();
        assert!(
            counting_file.reads().is_empty(),
            "set range should not eagerly build normal segments"
        );

        let mut first_block = Block::new(&schema);
        let read_rows = block_on(reader.read_block(&mut first_block, 3)).unwrap();
        assert_eq!(read_rows, 3);
        assert_eq!(first_block.handles.get_int_handle_value(0), 1);
        assert_eq!(first_block.handles.get_int_handle_value(1), 2);
        assert_eq!(first_block.handles.get_int_handle_value(2), 3);
        assert!(
            reader.current_segment.is_none(),
            "drained segment should be released immediately"
        );
        let state = reader.range_state.as_ref().unwrap();
        assert_eq!(state.cursor.file_idx, 0);
        assert_eq!(state.cursor.pack_idx, 1);

        reader.reset();
        let out = block_on(reader.read_all());
        assert_eq!(out.length(), 7);
        for i in 0..7 {
            assert_eq!(
                out.columns[0].get_value(i).unwrap().get_i64_le(),
                i as i64 + 1
            );
            assert_eq!(
                out.columns[1].get_value(i).unwrap(),
                format!("v{}", i + 1).as_bytes()
            );
        }
    }

    #[test]
    fn test_pack_clean_reader_reuses_clean_reader_across_packs() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 8, 100);
        let file = build_columnar_file_from_block(19, &schema, &block, 4, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_unbounded_handle_range()).unwrap();

        let mut first_block = Block::new(&schema);
        let read_rows = block_on(reader.read_block(&mut first_block, 4)).unwrap();
        assert_eq!(read_rows, 4);
        let first_reader_ptr = reader.clean_reader.as_ref().unwrap() as *const _;

        let mut second_block = Block::new(&schema);
        let read_rows = block_on(reader.read_block(&mut second_block, 4)).unwrap();
        assert_eq!(read_rows, 4);
        let second_reader_ptr = reader.clean_reader.as_ref().unwrap() as *const _;

        assert_eq!(first_reader_ptr, second_reader_ptr);
    }

    #[test]
    fn test_pack_clean_reader_reuses_normal_reader_across_boundary_segments() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let block = build_clean_block(&schema, 0, 8, 100);
        let file = build_columnar_file_from_block(20, &schema, &block, 4, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();

        let mut reader =
            ColumnarPackCleanReader::new(&[columnar_file], schema.clone(), 200, None, None);
        block_on(reader.set_int_handle_range(1, Some(7))).unwrap();

        let mut first_block = Block::new(&schema);
        let read_rows = block_on(reader.read_block(&mut first_block, 3)).unwrap();
        assert_eq!(read_rows, 3);
        let first_reader_ptr = reader.normal_reader.as_ref().unwrap() as *const _;

        let mut second_block = Block::new(&schema);
        let read_rows = block_on(reader.read_block(&mut second_block, 3)).unwrap();
        assert_eq!(read_rows, 3);
        let second_reader_ptr = reader.normal_reader.as_ref().unwrap() as *const _;

        assert_eq!(first_reader_ptr, second_reader_ptr);
    }

    fn new_mvcc_reader(
        schema: &Schema,
        files: &[Arc<dyn File>],
        read_ts: u64,
        encryption_key: Option<EncryptionKey>,
    ) -> ColumnarMvccReader {
        let mut readers: Vec<Box<dyn ColumnarReader>> = vec![];
        let columnar_meta_cache = ColumnarMetaCache::default();
        for file in files {
            let columnar_file =
                ColumnarFile::open(file.clone(), None, columnar_meta_cache.clone()).unwrap();
            let reader = ColumnarTableReader::new(
                &columnar_file,
                schema.clone(),
                None,
                encryption_key.clone(),
            )
            .unwrap();
            readers.push(Box::new(reader));
        }
        let merge_reader = ColumnarMergeReader::new(schema.clone(), readers);
        ColumnarMvccReader::new(Box::new(merge_reader), schema, read_ts)
    }

    fn new_compact_reader(
        schema: &Schema,
        level: u32,
        files: &[Arc<dyn File>],
        safe_ts: u64,
    ) -> ColumnarCompactReader {
        let columnar_meta_cache = ColumnarMetaCache::default();
        let mut readers: Vec<Box<dyn ColumnarReader>> = vec![];
        for file in files {
            let columnar_file =
                ColumnarFile::open(file.clone(), None, columnar_meta_cache.clone()).unwrap();
            if !columnar_file.has_table(schema.table_id) {
                continue;
            }
            let reader =
                ColumnarTableReader::new(&columnar_file, schema.clone(), None, None).unwrap();
            readers.push(Box::new(reader));
        }
        let merge_reader = ColumnarMergeReader::new(schema.clone(), readers);
        ColumnarCompactReader::new(Box::new(merge_reader), level, schema, safe_ts)
    }

    pub fn verify_with_ref_rows(block: &Block, ref_rows: &[RefRow]) {
        for i in 0..ref_rows.len() {
            let ref_row = &ref_rows[i];
            assert_eq!(ref_row.handle, block.handles.get_not_null_value(i), "{}", i);
            assert_eq!(ref_row.version, block.versions.get_version(i));
            assert_eq!(ref_row.is_deleted, block.versions.is_null(i));
            assert_eq!(ref_row.c0.is_none(), block.columns[0].is_null(i));
            assert_eq!(ref_row.c1.is_none(), block.columns[1].is_null(i));
            if let Some(c0) = ref_row.c0 {
                assert_eq!(c0, block.columns[0].get_value(i).unwrap().get_i64_le());
            }
            if let Some(c1) = &ref_row.c1 {
                assert_eq!(c1.as_slice(), block.columns[1].get_value(i).unwrap());
            }
        }
    }

    pub fn merge_refs(
        refs: Vec<Vec<RefRow>>,
        level: u32,
        read_ts: Option<u64>,
        safe_ts: Option<u64>,
        bound: Option<(Vec<u8>, Vec<u8>)>,
    ) -> Vec<RefRow> {
        let mut merged = vec![];
        for ref_rows in refs {
            merged.extend_from_slice(&ref_rows);
        }
        if let Some(read_ts) = read_ts {
            merged.retain(|r| r.version <= read_ts);
        }
        if let Some((start, end)) = bound {
            merged.retain(|r| r.in_bound(&start, &end));
        }
        merged.sort_by(|a, b| a.cmp(b));
        if read_ts.is_some() {
            merged.dedup_by(|a, b| a.handle == b.handle);
            merged.retain(|r| !r.is_deleted);
        }
        if let Some(safe_ts) = safe_ts {
            let mut new_merged = Vec::with_capacity(merged.len());
            let mut prev_handle: Vec<u8> = vec![];
            let mut reached_safe_ts_boundary = false;
            for r in &merged {
                if prev_handle != r.handle {
                    prev_handle = r.handle.clone();
                    reached_safe_ts_boundary = false;
                } else if reached_safe_ts_boundary {
                    continue;
                }

                if r.version <= safe_ts {
                    reached_safe_ts_boundary = true;
                    // For bottommost level compaction, a tombstone <= safe_ts can be dropped.
                    if level == 2 && r.is_deleted {
                        continue;
                    }
                }
                new_merged.push(r.clone());
            }
            merged = new_merged;
        }
        merged
    }

    #[rstest]
    #[case::enable_encryption(true)]
    #[case::disable_encryption(false)]
    fn test_reader(#[case] enable_encryption: bool) {
        init_log_for_test();
        let encryption_key = if enable_encryption {
            Some(new_test_encryption_key())
        } else {
            None
        };
        let mut options = vec![];
        let mut rng = rand::thread_rng();
        for _ in 0..100 {
            options.push(rng.gen_bool(0.5))
        }
        for common_handle in options {
            let schema = new_schema(1, common_handle);
            let (file_1, mut ref_1) = build_table_with_encryption(
                1,
                std::slice::from_ref(&schema),
                100,
                150,
                100,
                false,
                encryption_key.clone(),
            );
            let (file_2, mut ref_2) = build_table_with_encryption(
                2,
                std::slice::from_ref(&schema),
                140,
                190,
                110,
                false,
                encryption_key.clone(),
            );
            let (file_3, mut ref_3) = build_table_with_encryption(
                3,
                std::slice::from_ref(&schema),
                185,
                240,
                120,
                false,
                encryption_key.clone(),
            );
            let files = [file_1, file_2, file_3];
            let ref_rows = vec![
                ref_1.pop().unwrap(),
                ref_2.pop().unwrap(),
                ref_3.pop().unwrap(),
            ];
            let mut block = Block::new(&schema);
            let mut rng = rand::thread_rng();
            for read_ts in [90, 100, 110, 120] {
                let start_handle = rng.gen_range(90i64..170i64);
                let end_handle = start_handle + rng.gen_range(1i64..200i64);
                let mut mvcc_reader =
                    new_mvcc_reader(&schema, &files, read_ts, encryption_key.clone());
                if common_handle {
                    let common_start_handle = i_to_common_handle(start_handle as i32);
                    let common_end_handle = i_to_common_handle(end_handle as i32);
                    block_on(
                        mvcc_reader.set_handle_range(&common_start_handle, &common_end_handle),
                    )
                    .unwrap();
                } else {
                    block_on(mvcc_reader.set_int_handle_range(start_handle, Some(end_handle)))
                        .unwrap();
                }
                block_on(mvcc_reader.read_block(&mut block, 500)).unwrap();
                let range_bound = if common_handle {
                    (
                        i_to_common_handle(start_handle as i32),
                        i_to_common_handle(end_handle as i32),
                    )
                } else {
                    (
                        start_handle.to_le_bytes().to_vec(),
                        end_handle.to_le_bytes().to_vec(),
                    )
                };
                let merged_refs =
                    merge_refs(ref_rows.clone(), 0, Some(read_ts), None, Some(range_bound));
                verify_with_ref_rows(&block, &merged_refs);
            }
        }
    }

    proptest! {
        #[test]
        fn test_concat_reader(
            common_handle in any::<bool>(),
        ) {
            init_log_for_test();
            let schema = new_schema(1, common_handle);
            let columnar_meta_cache = ColumnarMetaCache::default();
            let (file_1, ref_1) = build_table(1, &schema, 100, 150, 100);
            let (file_2, ref_2) = build_table(2, &schema, 160, 190, 100);
            let (file_3, ref_3) = build_table(3, &schema, 191, 240, 100);
            let files = [file_1, file_2, file_3];
            let ref_rows = vec![ref_1, ref_2, ref_3];
            let col_files: Vec<ColumnarFile> = files.iter().map(|f| ColumnarFile::open(f.clone(), None, columnar_meta_cache.clone()).unwrap()).collect();
            for _ in 0..50 {
                let mut rng = rand::thread_rng();
                let start_handle = rng.gen_range(90i64..170i64);
                let end_handle = start_handle + rng.gen_range(1i64..200i64);
                let mut block = Block::new(&schema);
                let concat_reader = ColumnarConcatReader::new(&col_files, schema.clone(), None, None);
                let mut mvcc_reader = ColumnarMvccReader::new(Box::new(concat_reader), &schema, 100);
                if common_handle {
                    let common_start_handle = i_to_common_handle(start_handle as i32);
                    let common_end_handle = i_to_common_handle(end_handle as i32);
                    block_on(mvcc_reader
                        .set_handle_range(&common_start_handle, &common_end_handle))
                        .unwrap();
                } else {
                    block_on(mvcc_reader
                        .set_int_handle_range(start_handle, Some(end_handle)))
                        .unwrap();
                }
                block_on(mvcc_reader.read_block(&mut block, 500)).unwrap();
                let range_bound = if common_handle {
                    (
                        i_to_common_handle(start_handle as i32),
                        i_to_common_handle(end_handle as i32),
                    )
                } else {
                    (
                        start_handle.to_le_bytes().to_vec(),
                        end_handle.to_le_bytes().to_vec(),
                    )
                };
                let merged_refs = merge_refs(ref_rows.clone(), 0, Some(100), None, Some(range_bound));
                verify_with_ref_rows(&block, &merged_refs);
            }
        }
    }

    proptest! {
        #[test]
        fn test_compact_reader(
            common_handle in any::<bool>(),
        ) {
            init_log_for_test();
            let schema = new_schema(1, common_handle);
            let (file_1, ref_1) = build_table(1, &schema, 100, 200, 100);
            let (file_2, ref_2) = build_table(2, &schema, 150, 250, 200);
            let (file_3, ref_3) = build_table(3, &schema, 200, 300, 300);
            let files = vec![file_1, file_2, file_3];
            let ref_rows = vec![ref_1, ref_2, ref_3];
            for _ in 0..50 {
                let mut rng = rand::thread_rng();
                let start_handle = rng.gen_range(90i64..170i64);
                let end_handle = start_handle + rng.gen_range(1i64..200i64);
                let mut block = Block::new(&schema);
                let mut compact_reader = new_compact_reader(&schema, 2, &files, 250);
                if common_handle {
                    let common_start_handle = i_to_common_handle(start_handle as i32);
                    let common_end_handle = i_to_common_handle(end_handle as i32);
                    block_on(compact_reader
                        .set_handle_range(&common_start_handle, &common_end_handle))
                        .unwrap();
                } else {
                    block_on(compact_reader
                        .set_int_handle_range(start_handle, Some(end_handle)))
                        .unwrap();
                }
                block_on(compact_reader.read_block(&mut block, 1000)).unwrap();
                let range_bound = if common_handle {
                    (
                        i_to_common_handle(start_handle as i32),
                        i_to_common_handle(end_handle as i32),
                    )
                } else {
                    (
                        start_handle.to_le_bytes().to_vec(),
                        end_handle.to_le_bytes().to_vec(),
                    )
                };
                let merged_refs = merge_refs(ref_rows.clone(), 2, None, Some(250), Some(range_bound));
                verify_with_ref_rows(&block, &merged_refs);
            }
        }
    }

    #[test]
    fn test_compact_reader_keeps_safe_ts_boundary_for_newer_delete() {
        init_log_for_test();
        for common_handle in [false, true] {
            let schema = new_schema(1, common_handle);
            let handle = if common_handle {
                i_to_common_handle(1)
            } else {
                1_i64.to_le_bytes().to_vec()
            };
            let delete_row = RefRow {
                handle: handle.clone(),
                is_common_handle: common_handle,
                version: 300,
                txn_id: 299,
                is_deleted: true,
                c0: None,
                c1: None,
            };
            let old_row = RefRow {
                handle,
                is_common_handle: common_handle,
                version: 100,
                txn_id: 99,
                is_deleted: false,
                c0: Some(1),
                c1: Some(b"old".to_vec()),
            };
            let ref_rows = [delete_row, old_row];

            let is_common_handle = get_fixed_size(&schema.handle_column) == 0;
            let cf_tbl = CfTable::new();
            let mut eval_ctx = EvalContext::default();
            let mut wb = WriteBatch::new();
            for ref_row in ref_rows.iter().rev() {
                let row_key = if is_common_handle {
                    encode_common_handle_row_key(schema.table_id, ref_row.handle.as_slice())
                } else {
                    encode_row_key(schema.table_id, ref_row.handle.as_slice().get_i64_le())
                };
                let inner_key = InnerKey::from_inner_buf(&row_key);
                let mut row_val = vec![];
                if !ref_row.is_deleted {
                    let cols = vec![
                        Column::new(1, ref_row.c0),
                        Column::new(2, ref_row.c1.clone()),
                    ];
                    row_val.write_row(&mut eval_ctx, cols).unwrap();
                }
                let user_meta = UserMeta::new(ref_row.txn_id, ref_row.version).to_array();
                wb.put(inner_key, 0, &user_meta, ref_row.version, &row_val);
            }
            cf_tbl.get_cf(WRITE_CF).put_batch(&mut wb, None, WRITE_CF);

            let iter = cf_tbl.get_cf(WRITE_CF).new_iterator(false);
            let row_reader =
                ColumnarRowTableReader::new(schema.clone(), iter, None, false, None).unwrap();
            let mut compact_reader =
                ColumnarCompactReader::new(Box::new(row_reader), 2, &schema, 200);
            block_on(compact_reader.set_unbounded_handle_range()).unwrap();
            let out = block_on(compact_reader.read_all());

            assert_eq!(
                out.length(),
                2,
                "expect both the newer delete and the safe_ts boundary version to remain, common_handle={}",
                common_handle,
            );
            assert!(out.versions.is_null(0), "common_handle={}", common_handle);
            assert_eq!(
                out.versions.get_version(0),
                300,
                "common_handle={}",
                common_handle
            );
            assert!(!out.versions.is_null(1), "common_handle={}", common_handle);
            assert_eq!(
                out.versions.get_version(1),
                100,
                "common_handle={}",
                common_handle
            );
        }
    }

    #[rstest]
    #[case::int_handle(false)]
    #[case::common_handle(true)]
    fn test_mvcc_reader_stops_reading_after_range_end(#[case] common_handle: bool) {
        init_log_for_test();
        let schema = new_schema(1, common_handle);
        let read_calls = Arc::new(Mutex::new(0));

        let mut first = Block::new(&schema);
        for i in 1..=4 {
            let handle = if common_handle {
                i_to_common_handle(i)
            } else {
                (i as i64).to_le_bytes().to_vec()
            };
            first.handles.push_value(&handle);
            first.versions.push_value(&100_u64.to_le_bytes());
            first.columns[0].push_value(&(i as i64).to_le_bytes());
            first.columns[1].push_value(format!("v{i}").as_bytes());
        }

        let mut second = Block::new(&schema);
        for i in 5..=6 {
            let handle = if common_handle {
                i_to_common_handle(i)
            } else {
                (i as i64).to_le_bytes().to_vec()
            };
            second.handles.push_value(&handle);
            second.versions.push_value(&100_u64.to_le_bytes());
            second.columns[0].push_value(&(i as i64).to_le_bytes());
            second.columns[1].push_value(format!("v{i}").as_bytes());
        }

        let src = MockColumnarReader::new(schema.clone(), vec![first, second], read_calls.clone());
        let mut reader = ColumnarMvccReader::new(Box::new(src), &schema, 100);
        if common_handle {
            block_on(reader.set_handle_range(&i_to_common_handle(1), &i_to_common_handle(3)))
                .unwrap();
        } else {
            block_on(reader.set_int_handle_range(1, Some(3))).unwrap();
        }

        let mut out = Block::new(&schema);
        let rows = block_on(reader.read_block(&mut out, 1024)).unwrap();
        assert_eq!(rows, 2);
        if common_handle {
            assert_eq!(out.handles.get_not_null_value(0), i_to_common_handle(1));
            assert_eq!(out.handles.get_not_null_value(1), i_to_common_handle(2));
        } else {
            assert_eq!(out.handles.get_int_handle_value(0), 1);
            assert_eq!(out.handles.get_int_handle_value(1), 2);
        }

        let rows = block_on(reader.read_block(&mut out, 1024)).unwrap();
        assert_eq!(rows, 0);
        assert_eq!(*read_calls.lock().unwrap(), 1);
    }

    #[test]
    fn test_reader_with_multi_tables() {
        init_log_for_test();
        let schema_1 = new_schema(1, false);
        let schema_2 = new_schema(2, false);
        let schemas = vec![schema_1.clone(), schema_2.clone()];
        let (file_1, ref_rows) = build_mixed_table(1, &schemas, 0, 100, 100);

        for (i, schema) in schemas.iter().enumerate() {
            let mut block = Block::new(schema);
            let ref_row = ref_rows[i].clone();
            let mut mvcc_reader = new_mvcc_reader(schema, std::slice::from_ref(&file_1), 110, None);
            block_on(mvcc_reader.set_unbounded_handle_range()).unwrap();
            block_on(mvcc_reader.read_block(&mut block, 500)).unwrap();
            let merged_refs = merge_refs(vec![ref_row], 0, Some(110), None, None);
            verify_with_ref_rows(&block, &merged_refs);
        }
    }

    #[test]
    fn test_read_column_default_value() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let (file, ref_row) = build_table(1, &schema, 0, 100, 100);

        let mut schema_buf_builder = SchemaBufBuilder::new(1);
        let mut columns = schema.columns.clone();
        let mut column_info = ColumnInfo::new();
        column_info.set_tp(FieldTypeTp::Int24 as i32);
        let default_val = [vec![UINT_FLAG], 666_u64.to_be_bytes().to_vec()].concat();
        column_info.set_default_val(default_val);
        column_info.set_column_id(100);
        columns.push(column_info);
        schema_buf_builder.columns(
            schema.handle_column.clone(),
            schema.version_column.clone(),
            columns,
            schema.pk_col_ids.clone(),
            100,
            schema.vector_indexes.clone(),
            schema.fulltext_indexes.clone(),
        );
        let new_schema = Schema::new(schema_buf_builder.build());
        // Add a new column with default value to read.
        let mut mvcc_reader = new_mvcc_reader(&new_schema, &[file], 110, None);
        block_on(mvcc_reader.set_unbounded_handle_range()).unwrap();
        let mut block = Block::new(&new_schema);
        block_on(mvcc_reader.read_block(&mut block, 500)).unwrap();
        let merged_refs = merge_refs(vec![ref_row], 0, Some(110), None, None);
        verify_with_ref_rows(&block, &merged_refs);
        let column = block.get_column(100);
        assert_eq!(column.length(), block.handles.length());
        for i in 0..block.handles.length() {
            assert_eq!(column.get_value(i).unwrap(), 666_u64.to_le_bytes());
        }
    }

    #[rstest]
    #[case::int_handle(false)]
    #[case::common_handle(true)]
    fn test_late_column_accessor_uses_variable_width_offsets(#[case] common_handle: bool) {
        init_log_for_test();
        let schema = new_schema(1, common_handle);
        let mut block = Block::new(&schema);
        let mut expected_values = Vec::new();
        for i in 0..24 {
            let handle = if common_handle {
                i_to_common_handle(i)
            } else {
                (i as i64).to_le_bytes().to_vec()
            };
            block.handles.push_value(&handle);
            block.versions.push_version(100, false);
            block.columns[0].push_value(&(i as i64).to_le_bytes());
            let value = vec![b'x'; 1 + (i as usize % 6) * 17];
            block.columns[1].push_value(&value);
            expected_values.push(value);
        }

        let mut opts = ColumnarTableBuildOptions::default();
        opts.pack_max_row_count = 8;
        opts.pack_max_size = 32;
        let file = build_columnar_file_from_block_with_options(45, &schema, &block, opts, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let table_meta = columnar_file.get_table(schema.table_id);
        assert!(table_meta.columns[&2].pack_offsets.num_packs() > 3);

        let mut accessor = LateColumnAccessor::new(&columnar_file, &schema, 2, None).unwrap();
        let slots = [
            LateReadSlot {
                physical_row_idx: 17,
                selected_output_idx: 0,
            },
            LateReadSlot {
                physical_row_idx: 1,
                selected_output_idx: 1,
            },
            LateReadSlot {
                physical_row_idx: 22,
                selected_output_idx: 2,
            },
        ];
        let mut out = ColumnBuffer::new_from_col_info(schema.find_column_by_id(2).unwrap());
        block_on(accessor.read_selected(&slots, &mut out)).unwrap();

        assert_eq!(out.length(), slots.len());
        for (output_idx, slot) in slots.iter().enumerate() {
            assert_eq!(
                out.get_value(output_idx),
                Some(expected_values[slot.physical_row_idx as usize].as_slice())
            );
        }

        let duplicate_slots = [
            LateReadSlot {
                physical_row_idx: 3,
                selected_output_idx: 0,
            },
            LateReadSlot {
                physical_row_idx: 4,
                selected_output_idx: 0,
            },
        ];
        let mut rejected = ColumnBuffer::new_from_col_info(schema.find_column_by_id(2).unwrap());
        block_on(accessor.read_selected(&duplicate_slots, &mut rejected)).unwrap_err();
    }

    #[test]
    fn test_hybrid_common_handle_merge_materializes_late_values_and_ranges() {
        init_log_for_test();
        let full_schema = new_schema(1, true);
        let handle_id = full_schema.handle_column.get_column_id();
        let version_id = full_schema.version_column.get_column_id();
        let early_schema = Schema::new(full_schema.to_schema_buf().retain_columns(|column| {
            let column_id = column.get_column_id();
            column_id == handle_id || column_id == version_id || column_id == 1
        }));
        let late_column = full_schema.find_column_by_id(2).unwrap().clone();

        let first_block = build_clean_block(&full_schema, 0, 4, 100);
        let second_block = build_clean_block(&full_schema, 4, 8, 100);
        let first_file = ColumnarFile::open(
            build_columnar_file_from_block(46, &full_schema, &first_block, 2, 2),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();
        let second_file = ColumnarFile::open(
            build_columnar_file_from_block(47, &full_schema, &second_block, 2, 2),
            None,
            ColumnarMetaCache::default(),
        )
        .unwrap();

        let make_reader = || {
            let sources = vec![
                HybridSourceReader::L2 {
                    reader: ColumnarConcatReader::new(
                        std::slice::from_ref(&second_file),
                        early_schema.clone(),
                        None,
                        None,
                    ),
                    l2_reader_id: 1,
                },
                HybridSourceReader::L2 {
                    reader: ColumnarConcatReader::new(
                        std::slice::from_ref(&first_file),
                        early_schema.clone(),
                        None,
                        None,
                    ),
                    l2_reader_id: 0,
                },
            ];
            HybridMvccReader::new(
                HybridMergeReader::new(
                    early_schema.clone(),
                    full_schema.clone(),
                    vec![late_column.clone()],
                    sources,
                ),
                100,
                vec![vec![first_file.clone()], vec![second_file.clone()]],
            )
        };

        let mut reader = make_reader();
        block_on(reader.set_handle_range(&[], GLOBAL_COMMON_HANDLE_END)).unwrap();
        let mut batch = HybridBatch::new(
            Block::new(&early_schema),
            vec![ColumnBuffer::new_from_col_info(&late_column)],
        );
        assert_eq!(block_on(reader.read(&mut batch, 64)).unwrap(), 8);
        assert_eq!(
            (0..8).map(i_to_common_handle).collect::<Vec<_>>(),
            (0..batch.early.length())
                .map(|row_idx| batch.early.handles.get_not_null_value(row_idx).to_vec())
                .collect::<Vec<_>>()
        );

        let plan = batch.build_materialization_plan(&[1; 8]).unwrap();
        let (late, late_stats) = block_on(batch.materialize_selected_late_column(
            &plan,
            0,
            2,
            &full_schema,
            reader.l2_files(),
            None,
        ))
        .unwrap();
        assert!(late_stats.packs_loaded > 0);
        assert_eq!(late_stats.rows_gathered, 8);
        for i in 0..8 {
            assert_eq!(late.get_value(i), Some(format!("v{i}").as_bytes()));
        }

        let mut reader = make_reader();
        block_on(reader.set_handle_range(&i_to_common_handle(2), &i_to_common_handle(6))).unwrap();
        let mut bounded = HybridBatch::new(
            Block::new(&early_schema),
            vec![ColumnBuffer::new_from_col_info(&late_column)],
        );
        assert_eq!(block_on(reader.read(&mut bounded, 64)).unwrap(), 4);
        for (row_idx, expected_handle) in (2..6).enumerate() {
            assert_eq!(
                bounded.early.handles.get_not_null_value(row_idx),
                i_to_common_handle(expected_handle)
            );
        }

        let mut reader = make_reader();
        let same_handle = i_to_common_handle(4);
        block_on(reader.set_handle_range(&same_handle, &same_handle)).unwrap();
        let mut empty = HybridBatch::new(
            Block::new(&early_schema),
            vec![ColumnBuffer::new_from_col_info(&late_column)],
        );
        assert_eq!(block_on(reader.read(&mut empty, 64)).unwrap(), 0);
    }

    #[test]
    fn test_table_reader_reports_version_column_physical_rows() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let expected = build_clean_block(&schema, 0, 20, 100);
        let file = build_columnar_file_from_block(43, &schema, &expected, 8, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let mut reader =
            ColumnarTableReader::new(&columnar_file, schema.clone(), None, None).unwrap();
        block_on(reader.seek(&[])).unwrap();

        let mut block = Block::new(&schema);
        let mut physical_rows = vec![];
        assert_eq!(
            block_on(reader.read_with_physical_rows(&mut block, &mut physical_rows, 15)).unwrap(),
            15
        );
        assert_eq!(physical_rows, (0..15).collect::<Vec<_>>());
        assert_eq!(block.handles.length(), physical_rows.len());
    }

    #[test]
    fn test_table_reader_keeps_physical_rows_across_rough_filter_gap() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let mut expected = build_clean_block(&schema, 0, 12, 100);
        expected.columns[0].reset();
        for value in [0_i64, 1, 2, 3, 10, 11, 12, 13, 0, 1, 2, 3] {
            expected.columns[0].push_value(&value.to_le_bytes());
        }

        let file = build_columnar_file_from_block(44, &schema, &expected, 4, 2);
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let filter = build_less_than_filter(&schema, 1, 4);
        let mut reader =
            ColumnarTableReader::new(&columnar_file, schema.clone(), Some(filter), None).unwrap();
        block_on(reader.seek(&[])).unwrap();

        let mut early = Block::new(&schema);
        let mut physical_rows = vec![];
        assert_eq!(
            block_on(reader.read_with_physical_rows(&mut early, &mut physical_rows, 12)).unwrap(),
            8
        );
        assert_eq!(
            block_on(reader.read_with_physical_rows(&mut early, &mut physical_rows, 12)).unwrap(),
            0
        );

        assert_eq!(physical_rows, vec![0, 1, 2, 3, 8, 9, 10, 11]);
        assert_eq!(early.length(), physical_rows.len());
        for (output_idx, physical_row_idx) in physical_rows.iter().enumerate() {
            assert_eq!(
                early.handles.get_int_handle_value(output_idx),
                *physical_row_idx as i64
            );
        }

        let mut late_reader = LateColumnAccessor::new(&columnar_file, &schema, 2, None).unwrap();
        let slots = physical_rows
            .iter()
            .enumerate()
            .map(|(selected_output_idx, &physical_row_idx)| LateReadSlot {
                physical_row_idx,
                selected_output_idx,
            })
            .collect::<Vec<_>>();
        let mut late = ColumnBuffer::new_from_col_info(schema.find_column_by_id(2).unwrap());
        block_on(late_reader.read_selected(&slots, &mut late)).unwrap();
        for (output_idx, physical_row_idx) in physical_rows.iter().enumerate() {
            assert_eq!(
                late.get_value(output_idx).unwrap(),
                format!("v{physical_row_idx}").as_bytes()
            );
        }
    }

    #[test]
    fn test_hybrid_batch_retains_early_rows_and_refs_together() {
        let schema = new_schema(1, false);
        let mut early = build_clean_block(&schema, 0, 4, 100);
        early.columns.truncate(1);
        let mut batch = HybridBatch::new(
            early,
            vec![ColumnBuffer::new_from_col_info(
                schema.find_column_by_id(2).unwrap(),
            )],
        );
        batch.refs = vec![
            MaterializationRef::Eager { eager_row_idx: 0 },
            MaterializationRef::L2Deferred {
                l2_reader_id: 0,
                file_index: 3,
                physical_row_idx: 11,
            },
            MaterializationRef::Eager { eager_row_idx: 1 },
            MaterializationRef::L2Deferred {
                l2_reader_id: 0,
                file_index: 4,
                physical_row_idx: 12,
            },
        ];

        batch.retain_ranges(&[(1, 2), (3, 4)]).unwrap();

        assert_eq!(batch.early.length(), 2);
        assert_eq!(batch.refs.len(), 2);
        assert_eq!(batch.early.handles.get_int_handle_value(0), 1);
        assert_eq!(batch.early.handles.get_int_handle_value(1), 3);
        assert!(matches!(
            batch.refs[0],
            MaterializationRef::L2Deferred { file_index: 3, .. }
        ));
        assert!(matches!(
            batch.refs[1],
            MaterializationRef::L2Deferred { file_index: 4, .. }
        ));
    }

    #[test]
    fn test_hybrid_batch_append_keeps_eager_sidecar_indexes_stable() {
        let schema = new_schema(1, false);
        let source = build_clean_block(&schema, 0, 4, 100);
        let mut batch = HybridBatch::new(
            Block::new(&schema),
            vec![ColumnBuffer::new_from_col_info(
                schema.find_column_by_id(2).unwrap(),
            )],
        );
        batch
            .append_eager(&source, &[source.columns[1].clone()], 1, 3)
            .unwrap();
        assert_eq!(batch.refs.len(), 2);
        assert!(matches!(
            batch.refs[0],
            MaterializationRef::Eager { eager_row_idx: 0 }
        ));
        assert!(matches!(
            batch.refs[1],
            MaterializationRef::Eager { eager_row_idx: 1 }
        ));
        assert_eq!(
            batch.eager_late[0].get_value(0),
            source.columns[1].get_value(1)
        );
        assert_eq!(
            batch.eager_late[0].get_value(1),
            source.columns[1].get_value(2)
        );
    }

    #[test]
    fn test_hybrid_batch_materializes_selected_l2_late_values() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let full = build_clean_block(&schema, 0, 16, 100);
        let file = build_columnar_file_from_block(44, &schema, &full, 8, 2);
        let file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();
        let early = build_clean_block(&schema, 0, 2, 100);
        let mut batch = HybridBatch::new(
            early,
            vec![ColumnBuffer::new_from_col_info(
                schema.find_column_by_id(2).unwrap(),
            )],
        );
        batch.refs = vec![
            MaterializationRef::L2Deferred {
                l2_reader_id: 0,
                file_index: 0,
                physical_row_idx: 1,
            },
            MaterializationRef::L2Deferred {
                l2_reader_id: 0,
                file_index: 0,
                physical_row_idx: 9,
            },
        ];

        let plan = batch.build_materialization_plan(&[1, 1]).unwrap();
        let (output, late_stats) = block_on(batch.materialize_selected_late_column(
            &plan,
            0,
            2,
            &schema,
            &[vec![file.clone()]],
            None,
        ))
        .unwrap();
        assert!(late_stats.packs_loaded > 0);
        assert_eq!(late_stats.rows_gathered, 2);
        assert_eq!(output.get_value(0), full.columns[1].get_value(1));
        assert_eq!(output.get_value(1), full.columns[1].get_value(9));

        let plan = batch.build_materialization_plan(&[1, 0]).unwrap();
        let (output, late_stats) = block_on(batch.materialize_selected_late_column(
            &plan,
            0,
            2,
            &schema,
            &[vec![file]],
            None,
        ))
        .unwrap();
        assert!(late_stats.packs_loaded > 0);
        assert_eq!(late_stats.rows_gathered, 1);
        assert_eq!(output.get_value(0), full.columns[1].get_value(1));
    }

    #[test]
    fn test_l2_late_materialization_requires_strict_file_non_overlap() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let first = build_columnar_file_from_block(
            50,
            &schema,
            &build_clean_block(&schema, 0, 8, 100),
            8,
            2,
        );
        let second = build_columnar_file_from_block(
            51,
            &schema,
            &build_clean_block(&schema, 8, 16, 100),
            8,
            2,
        );
        let overlapping = build_columnar_file_from_block(
            52,
            &schema,
            &build_clean_block(&schema, 7, 15, 100),
            8,
            2,
        );
        let open = |file| ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();

        let mut level = crate::table::columnar::columnar::ColumnarLevel::new(2);
        level.files = vec![open(first), open(second)];
        assert!(level.has_strictly_non_overlapping_files_for_table(schema.table_id));

        level.files = vec![
            open(build_columnar_file_from_block(
                53,
                &schema,
                &build_clean_block(&schema, 0, 8, 100),
                8,
                2,
            )),
            open(overlapping),
        ];
        assert!(!level.has_strictly_non_overlapping_files_for_table(schema.table_id));
    }

    #[test]
    fn test_read_decimal_column_default_value() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let (file, ref_row) = build_table(1, &schema, 0, 100, 100);

        let mut schema_buf_builder = SchemaBufBuilder::new(1);
        let mut columns = schema.columns.clone();
        let mut column_info = ColumnInfo::new();
        column_info.set_tp(FieldTypeTp::NewDecimal as i32);
        column_info.set_flag(FieldTypeFlag::NOT_NULL.bits() as i32);
        column_info.set_column_len(10);
        column_info.set_decimal(2);
        let decimal = "12345678.90".parse::<Decimal>().unwrap();
        let expected = decode_decimal_as_int(&column_info, &decimal);
        let mut default_val = vec![DECIMAL_FLAG];
        let res = default_val.write_decimal(&decimal, 10, 2).unwrap();
        assert!(res.is_ok(), "unexpected decimal encode result: {:?}", res);
        column_info.set_default_val(default_val);
        column_info.set_column_id(100);
        columns.push(column_info);
        schema_buf_builder.columns(
            schema.handle_column.clone(),
            schema.version_column.clone(),
            columns,
            schema.pk_col_ids.clone(),
            100,
            schema.vector_indexes.clone(),
            schema.fulltext_indexes.clone(),
        );
        let new_schema = Schema::new(schema_buf_builder.build());

        // Add a new decimal column with default value to read.
        let mut mvcc_reader = new_mvcc_reader(&new_schema, &[file], 110, None);
        block_on(mvcc_reader.set_unbounded_handle_range()).unwrap();
        let mut block = Block::new(&new_schema);
        block_on(mvcc_reader.read_block(&mut block, 500)).unwrap();
        let merged_refs = merge_refs(vec![ref_row], 0, Some(110), None, None);
        verify_with_ref_rows(&block, &merged_refs);
        let column = block.get_column(100);
        assert_eq!(column.length(), block.handles.length());
        for i in 0..block.handles.length() {
            assert_eq!(column.get_value(i).unwrap(), expected.as_slice());
        }
    }

    #[test]
    fn test_filter_reader_drained() {
        init_log_for_test();
        let schema = new_schema(1, false);
        let (file_1, ref_1) = build_table(1, &schema, 100, 200, 100);
        let (file_2, ref_2) = build_table(2, &schema, 150, 250, 200);
        let (file_3, ref_3) = build_table(3, &schema, 200, 300, 300);
        let (file_4, ref_4) = build_table_with_all_rows_deleted(4, &schema, 50, 150, 90);
        let files = vec![file_1, file_2, file_3, file_4];
        let ref_rows = vec![ref_1, ref_2, ref_3, ref_4];
        let mut block = Block::new(&schema);
        let mut compact_reader = new_compact_reader(&schema, 2, &files, 250);
        block_on(compact_reader.set_unbounded_handle_range()).unwrap();
        loop {
            let mut read_block = Block::new(&schema);
            let rows = block_on(compact_reader.read_block(&mut read_block, 10)).unwrap();
            if rows == 0 {
                break;
            }
            block.append(&read_block, 0, rows);
        }
        let merged_refs = merge_refs(ref_rows.clone(), 2, None, Some(250), None);
        verify_with_ref_rows(&block, &merged_refs);
    }

    #[test]
    fn test_decode_row_v2_null_in_not_null_column() {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, false);

        let iter = Box::new(crate::table::EmptyIterator);
        let row_tbl_reader =
            ColumnarRowTableReader::new(schema.clone(), iter, None, true, None).unwrap();
        let mut block = Block::new(&schema);
        block.handles.push_value(&1_i64.to_le_bytes());

        let row_val = encode_test_row(vec![
            Column::new(1, Some(1_i64)),
            Column::new(2, Option::<Vec<u8>>::None),
        ]);

        let err = row_tbl_reader
            .decode_row_columns_v2(&mut block, &row_val)
            .unwrap_err();
        match err {
            table::Error::SchemaOutOfDate(info) => assert_eq!(info.detail, "null_in_not_null"),
            other => panic!("unexpected error: {:?}", other),
        }
    }

    #[test]
    fn test_decode_row_v1_null_in_not_null_column() {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, false);

        let iter = Box::new(crate::table::EmptyIterator);
        let row_tbl_reader =
            ColumnarRowTableReader::new(schema.clone(), iter, None, true, None).unwrap();
        let mut block = Block::new(&schema);
        block.handles.push_value(&1_i64.to_le_bytes());

        let datums = vec![Datum::I64(1), Datum::I64(1), Datum::I64(2), Datum::Null];
        let row_val = datum::encode_value(&mut EvalContext::default(), &datums).unwrap();

        let err = row_tbl_reader
            .decode_row_columns_v1(&mut block, &row_val)
            .unwrap_err();
        match err {
            table::Error::SchemaOutOfDate(info) => assert_eq!(info.detail, "null_in_not_null"),
            other => panic!("unexpected error: {:?}", other),
        }
    }

    #[test]
    fn test_row_mvcc_reader_skips_older_versions_before_decode() {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, false);

        let row_key = encode_row_key(schema.table_id, 1);
        let latest_row = encode_test_row(vec![
            Column::new(1, Some(1_i64)),
            Column::new(2, Some(b"ok".to_vec())),
        ]);

        // An older schema-incompatible version (missing a NOT NULL column).
        let old_row_missing_not_null_col = encode_test_row(vec![Column::new(1, Some(1_i64))]);

        let mut wb = WriteBatch::new();
        put_mvcc_row(&mut wb, &row_key, 99, 150, &old_row_missing_not_null_col);
        put_mvcc_row(&mut wb, &row_key, 299, 300, &latest_row);
        let cf_tbl = CfTable::new();
        cf_tbl.get_cf(WRITE_CF).put_batch(&mut wb, None, WRITE_CF);

        // Baseline: decoding all versions fails on the old schema-incompatible version.
        let iter_all = cf_tbl.get_cf(WRITE_CF).new_iterator(false);
        let mut row_tbl_reader =
            ColumnarRowTableReader::new(schema.clone(), iter_all, None, false, None).unwrap();
        block_on(row_tbl_reader.seek(&1_i64.to_le_bytes())).unwrap();
        let mut block = Block::new(&schema);
        let err = block_on(row_tbl_reader.read(&mut block, 16)).unwrap_err();
        match err {
            table::Error::SchemaOutOfDate(info) => {
                assert_eq!(info.detail, "read null for not null column")
            }
            other => panic!("unexpected error: {:?}", other),
        }

        // New behavior: pick the latest visible version per handle first, then decode.
        // The old version is skipped and should not trigger SchemaOutOfDate.
        let iter_visible = cf_tbl.get_cf(WRITE_CF).new_iterator(false);
        let mut mvcc_reader = RowMvccReader::new(
            schema.clone(),
            vec![iter_visible],
            None,
            false,
            None,
            500, // read_ts, latest version is visible
        )
        .unwrap();
        block_on(mvcc_reader.seek(&1_i64.to_le_bytes())).unwrap();
        let mut out = Block::new(&schema);
        let read = block_on(mvcc_reader.read(&mut out, 16)).unwrap();
        assert_eq!(read, 1);
        assert_eq!(out.length(), 1);
        assert_eq!(out.handles.get_int_handle_value(0), 1);
        assert_eq!(out.versions.get_version(0), 300);
        assert!(!out.versions.is_null(0));
        assert_eq!(out.columns[1].get_value(0).unwrap(), b"ok");
    }

    #[rstest]
    #[case::int_handle(false, 1_i64.to_le_bytes().to_vec())]
    #[case::common_handle(true, b"h1".to_vec())]
    fn test_row_to_columnar_safe_ts_prunes_globally_before_decode(
        #[case] common_handle: bool,
        #[case] handle: Vec<u8>,
    ) {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, common_handle);
        let row_key = if common_handle {
            encode_common_handle_row_key(schema.table_id, handle.as_slice())
        } else {
            encode_row_key(schema.table_id, handle.as_slice().get_i64_le())
        };
        let safe_ts = 300_u64;

        let latest_row = encode_test_row(vec![
            Column::new(1, Some(1_i64)),
            Column::new(2, Some(b"ok".to_vec())),
        ]);
        let old_row_missing_not_null_col = encode_test_row(vec![Column::new(1, Some(1_i64))]);

        let mut wb_a = WriteBatch::new();
        put_mvcc_row(&mut wb_a, &row_key, 199, 200, &old_row_missing_not_null_col);
        let cf_tbl_a = CfTable::new();
        cf_tbl_a
            .get_cf(WRITE_CF)
            .put_batch(&mut wb_a, None, WRITE_CF);

        let mut wb_b = WriteBatch::new();
        put_mvcc_row(&mut wb_b, &row_key, 299, 300, &latest_row);
        let cf_tbl_b = CfTable::new();
        cf_tbl_b
            .get_cf(WRITE_CF)
            .put_batch(&mut wb_b, None, WRITE_CF);

        if !common_handle {
            // Baseline: merge row readers + safe_ts prune after decode fails because source
            // A attempts to decode a globally prunable old version first.
            let iter_a = cf_tbl_a.get_cf(WRITE_CF).new_iterator(false);
            let iter_b = cf_tbl_b.get_cf(WRITE_CF).new_iterator(false);
            let reader_a =
                ColumnarRowTableReader::new(schema.clone(), iter_a, None, true, None).unwrap();
            let reader_b =
                ColumnarRowTableReader::new(schema.clone(), iter_b, None, true, None).unwrap();
            let merge_reader = ColumnarMergeReader::new(
                schema.clone(),
                vec![Box::new(reader_a), Box::new(reader_b)],
            );
            let mut compact_reader =
                ColumnarCompactReader::new(Box::new(merge_reader), 0, &schema, safe_ts);
            let err = block_on(compact_reader.set_unbounded_handle_range()).unwrap_err();
            assert!(matches!(err, table::Error::SchemaOutOfDate(_)));
        }

        // Fixed: global safe_ts pruning before decode skips source A's old version.
        let iter_a = cf_tbl_a.get_cf(WRITE_CF).new_iterator(false);
        let iter_b = cf_tbl_b.get_cf(WRITE_CF).new_iterator(false);
        let mut row_reader = RowToColumnarReader::new(
            schema.clone(),
            vec![iter_a, iter_b],
            None,
            true,
            None,
            0,
            safe_ts,
            0,
        )
        .unwrap();
        block_on(row_reader.set_unbounded_handle_range()).unwrap();
        let mut block = Block::new(&schema);
        let read_rows = block_on(row_reader.read_block(&mut block, 16)).unwrap();
        assert_eq!(read_rows, 1);
        assert_eq!(block.length(), 1);
        assert_eq!(block.handles.get_value(0).unwrap(), handle.as_slice());
        assert_eq!(block.versions.get_version(0), 300);
        assert!(!block.versions.is_null(0));
        assert_eq!(block.columns[1].get_value(0).unwrap(), b"ok");
    }

    #[test]
    fn test_row_to_columnar_safe_ts_drops_tombstone_boundary_at_level_2() {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, false);

        let row_key_1 = encode_row_key(schema.table_id, 1);
        let row_key_2 = encode_row_key(schema.table_id, 2);

        let row_2 = encode_test_row(vec![
            Column::new(1, Some(2_i64)),
            Column::new(2, Some(b"ok".to_vec())),
        ]);

        // An older live version that is schema-incompatible. It must be pruned by the
        // tombstone boundary (safe_ts <= tombstone version) at level 2,
        // otherwise the reader would fail with `SchemaOutOfDate`.
        let old_row_missing_not_null_col = encode_test_row(vec![Column::new(1, Some(1_i64))]);

        let mut wb = WriteBatch::new();
        put_mvcc_row(&mut wb, &row_key_1, 99, 150, &old_row_missing_not_null_col);
        put_mvcc_row(&mut wb, &row_key_1, 199, 200, &[]);
        put_mvcc_row(&mut wb, &row_key_2, 299, 300, &row_2);
        let cf_tbl = CfTable::new();
        cf_tbl.get_cf(WRITE_CF).put_batch(&mut wb, None, WRITE_CF);

        let iter = cf_tbl.get_cf(WRITE_CF).new_iterator(false);
        let mut row_reader =
            RowToColumnarReader::new(schema.clone(), vec![iter], None, true, None, 0, 200, 2)
                .unwrap();
        block_on(row_reader.set_unbounded_handle_range()).unwrap();

        let mut block = Block::new(&schema);
        let read_rows = block_on(row_reader.read_block(&mut block, 16)).unwrap();
        assert_eq!(read_rows, 1);
        assert_eq!(block.length(), 1);
        assert_eq!(block.handles.get_int_handle_value(0), 2);
        assert_eq!(block.versions.get_version(0), 300);
        assert!(!block.versions.is_null(0));
        assert_eq!(block.columns[1].get_value(0).unwrap(), b"ok");
    }

    #[test]
    fn test_row_to_columnar_safe_ts_multi_handle_keeps_boundary_and_resets_state() {
        init_log_for_test();
        let table_id = 1;
        let schema = new_schema_two_columns_not_null(table_id, false);

        let row_key_1 = encode_row_key(schema.table_id, 1);
        let row_key_2 = encode_row_key(schema.table_id, 2);

        let h1_v400 = encode_test_row(vec![
            Column::new(1, Some(1_i64)),
            Column::new(2, Some(b"h1v400".to_vec())),
        ]);
        let h1_v250 = encode_test_row(vec![
            Column::new(1, Some(1_i64)),
            Column::new(2, Some(b"h1v250".to_vec())),
        ]);
        let h2_v500 = encode_test_row(vec![
            Column::new(1, Some(2_i64)),
            Column::new(2, Some(b"h2v500".to_vec())),
        ]);
        let h2_v100 = encode_test_row(vec![
            Column::new(1, Some(2_i64)),
            Column::new(2, Some(b"h2v100".to_vec())),
        ]);

        // Schema-incompatible old versions (missing NOT NULL col2). They are globally
        // prunable because a more recent boundary `version <= safe_ts` exists
        // in another source.
        let h1_v200_missing_not_null_col = encode_test_row(vec![Column::new(1, Some(1_i64))]);
        let h2_v50_missing_not_null_col = encode_test_row(vec![Column::new(1, Some(2_i64))]);

        let safe_ts = 300;

        let mut wb_a = WriteBatch::new();
        put_mvcc_row(
            &mut wb_a,
            &row_key_1,
            199,
            200,
            &h1_v200_missing_not_null_col,
        );
        put_mvcc_row(&mut wb_a, &row_key_2, 49, 50, &h2_v50_missing_not_null_col);
        let cf_tbl_a = CfTable::new();
        cf_tbl_a
            .get_cf(WRITE_CF)
            .put_batch(&mut wb_a, None, WRITE_CF);

        let mut wb_b = WriteBatch::new();
        // Insert versions in increasing order for each key.
        put_mvcc_row(&mut wb_b, &row_key_1, 249, 250, &h1_v250);
        put_mvcc_row(&mut wb_b, &row_key_1, 399, 400, &h1_v400);
        put_mvcc_row(&mut wb_b, &row_key_2, 99, 100, &h2_v100);
        put_mvcc_row(&mut wb_b, &row_key_2, 499, 500, &h2_v500);
        let cf_tbl_b = CfTable::new();
        cf_tbl_b
            .get_cf(WRITE_CF)
            .put_batch(&mut wb_b, None, WRITE_CF);

        let iter_a = cf_tbl_a.get_cf(WRITE_CF).new_iterator(false);
        let iter_b = cf_tbl_b.get_cf(WRITE_CF).new_iterator(false);
        let mut row_reader = RowToColumnarReader::new(
            schema.clone(),
            vec![iter_a, iter_b],
            None,
            true,
            None,
            0,
            safe_ts,
            0,
        )
        .unwrap();
        block_on(row_reader.set_unbounded_handle_range()).unwrap();
        let mut block = Block::new(&schema);
        let read_rows = block_on(row_reader.read_block(&mut block, 16)).unwrap();
        assert_eq!(read_rows, 4);
        assert_eq!(block.length(), 4);

        assert_eq!(block.handles.get_int_handle_value(0), 1);
        assert_eq!(block.versions.get_version(0), 400);
        assert_eq!(block.columns[1].get_value(0).unwrap(), b"h1v400");

        assert_eq!(block.handles.get_int_handle_value(1), 1);
        assert_eq!(block.versions.get_version(1), 250);
        assert_eq!(block.columns[1].get_value(1).unwrap(), b"h1v250");

        assert_eq!(block.handles.get_int_handle_value(2), 2);
        assert_eq!(block.versions.get_version(2), 500);
        assert_eq!(block.columns[1].get_value(2).unwrap(), b"h2v500");

        assert_eq!(block.handles.get_int_handle_value(3), 2);
        assert_eq!(block.versions.get_version(3), 100);
        assert_eq!(block.columns[1].get_value(3).unwrap(), b"h2v100");
    }

    #[test]
    fn test_build_columnar_skips_virtual_generated_columns() {
        init_log_for_test();
        let table_id = 1;
        let file_id = 1;

        let mut col_1 = new_column_info(1);
        col_1.set_tp(FieldTypeTp::LongLong as i32);

        let mut col_2 = new_column_info(2);
        col_2.set_tp(FieldTypeTp::VarChar as i32);
        col_2.set_collation(Utf8Mb4GeneralCi as i32);

        // Simulate TiDB virtual generated column behavior: it may be NOT NULL, while
        // the storage layer returns NULL placeholders and lets TiDB compute the real
        // values later.
        let mut col_3 = new_column_info(3);
        col_3.set_tp(FieldTypeTp::LongLong as i32);
        col_3.set_flag((FieldTypeFlag::NOT_NULL | FieldTypeFlag::VIRTUAL_GENERATED).bits() as i32);

        let mut schema_buf_builder = SchemaBufBuilder::new(table_id);
        schema_buf_builder.columns(
            new_int_handle_column_info(),
            new_version_column_info(),
            vec![col_1, col_2, col_3],
            vec![],
            // max_col_id
            3,
            vec![],
            vec![],
        );
        let full_schema = Schema::new(schema_buf_builder.build());
        let storage_schema: Schema = full_schema
            .retain_columns(|col| !crate::table::columnar::is_virtual_generated_column(col))
            .into();

        let mut block = Block::new(&storage_schema);
        for i in 0..10_i64 {
            block.handles.push_value(&i.to_le_bytes());
            block.versions.push_version(100, false);
            block.columns[0].push_value(&i.to_le_bytes());
            block.columns[1].push_value(format!("v{i}").as_bytes());
        }

        let mut file_builder = ColumnarFileBuilder::new(file_id, Some(100.into()), None);
        let mut opts = ColumnarTableBuildOptions::default();
        opts.pack_max_row_count = 8;
        opts.pack_max_size = 256;
        let mut table_builder =
            ColumnarTableBuilder::new(storage_schema.clone(), opts, None, file_id, 0);
        let mut offset = 0;
        while offset < block.length() {
            offset = table_builder.append_block(&block, offset);
        }
        file_builder.add_table(table_builder);
        let (file_data, _) = file_builder.build();
        let file: Arc<dyn File> = Arc::new(InMemFile::new(file_id, file_data.into()));
        let columnar_file = ColumnarFile::open(file, None, ColumnarMetaCache::default()).unwrap();

        // The virtual column (col_id=3) should not be materialized in the columnar
        // file.
        let table_meta = columnar_file.get_table(table_id);
        assert!(!table_meta.columns.contains_key(&3));

        // But it should still be readable as NULL placeholder via the missing-column
        // path.
        let mut reader =
            ColumnarTableReader::new(&columnar_file, full_schema.clone(), None, None).unwrap();
        block_on(reader.seek(&0_i64.to_le_bytes())).unwrap();
        let mut out = Block::new(&full_schema);
        let read = block_on(reader.read(&mut out, 20)).unwrap();
        assert_eq!(read, 10);
        let virtual_col = out.get_column(3);
        assert!(virtual_col.is_nullable());
        for i in 0..read {
            assert!(virtual_col.get_value(i).is_none());
        }
    }
}