summa_core/structures/fast_field/

codec.rs

//! Fast-field compression codecs with auto-selection.
//!
//! Four codecs are available, and the writer picks the smallest at build time:
//!
//! | Codec            | ID | Description                                      |
//! |------------------|----|--------------------------------------------------|
//! | Constant         |  0 | No data bytes — all values identical              |
//! | Bitpacked        |  1 | min-subtract + global bitpack                     |
//! | Linear           |  2 | Regression line, bitpack residuals                |
//! | BlockwiseLinear  |  3 | Per-512-block linear, bitpack residuals per block |

use std::io::{self, Write};

use byteorder::{LittleEndian, WriteBytesExt};

use super::{bitpack_read, bitpack_write, bits_needed_u64};

// ── Codec type tag ───────────────────────────────────────────────────────

/// Codec identifier stored in the column data region (first byte).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum CodecType {
    Constant = 0,
    Bitpacked = 1,
    Linear = 2,
    BlockwiseLinear = 3,
}

impl CodecType {
    pub fn from_u8(v: u8) -> Option<Self> {
        match v {
            0 => Some(Self::Constant),
            1 => Some(Self::Bitpacked),
            2 => Some(Self::Linear),
            3 => Some(Self::BlockwiseLinear),
            _ => None,
        }
    }
}

/// Block size for BlockwiseLinear codec (matching Tantivy).
pub const BLOCKWISE_LINEAR_BLOCK_SIZE: usize = 512;

/// Conservative raw-value interval derived only from an admitted codec header.
/// A wrapping interval or a codec without cheap bounds returns `None`.
pub(super) fn value_bounds(data: &[u8]) -> Option<(u64, u64)> {
    let (&tag, data) = data.split_first()?;
    let width_max = |bits: u8| u64::MAX.checked_shr(u32::from(64 - bits)).unwrap_or(0);
    match CodecType::from_u8(tag)? {
        CodecType::Constant => {
            let value = u64::from_le_bytes(data[..8].try_into().unwrap());
            Some((value, value))
        }
        CodecType::Bitpacked => {
            let min = u64::from_le_bytes(data[..8].try_into().unwrap());
            Some((min, min.checked_add(width_max(data[8]))?))
        }
        CodecType::Linear => {
            let first = u64::from_le_bytes(data[..8].try_into().unwrap());
            let last = u64::from_le_bytes(data[8..16].try_into().unwrap());
            let offset = i64::from_le_bytes(data[20..28].try_into().unwrap()) as i128;
            let min = i128::from(first.min(last)) + offset;
            let max = i128::from(first.max(last)) + offset + i128::from(width_max(data[28]));
            Some((u64::try_from(min).ok()?, u64::try_from(max).ok()?))
        }
        CodecType::BlockwiseLinear => None,
    }
}

/// Validate an auto-codec payload before exposing it through the infallible
/// hot-path readers below. This keeps every bounds check out of per-document
/// access while ensuring corrupt segment metadata cannot trigger slice panics.
pub fn validate_auto(data: &[u8], expected_values: usize) -> io::Result<()> {
    let (&codec_id, rest) = data.split_first().ok_or_else(|| {
        io::Error::new(io::ErrorKind::UnexpectedEof, "fast field codec is missing")
    })?;

    let packed_len = |count: usize, bpv: u8| -> io::Result<usize> {
        if bpv > 64 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "fast field bit width exceeds 64",
            ));
        }
        count
            .checked_mul(bpv as usize)
            .and_then(|bits| bits.checked_add(7))
            .map(|bits| bits / 8)
            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidData, "fast field size overflow"))
    };

    match CodecType::from_u8(codec_id) {
        Some(CodecType::Constant) => {
            if rest.len() != 8 {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "invalid constant fast field length",
                ));
            }
        }
        Some(CodecType::Bitpacked) => {
            if rest.len() < 9 {
                return Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "bitpacked fast field header is truncated",
                ));
            }
            let expected_len = 9usize
                .checked_add(packed_len(expected_values, rest[8])?)
                .ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidData, "fast field size overflow")
                })?;
            if rest.len() != expected_len {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "bitpacked fast field length is inconsistent",
                ));
            }
        }
        Some(CodecType::Linear) => {
            if rest.len() < 29 {
                return Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "linear fast field header is truncated",
                ));
            }
            let count = u32::from_le_bytes(rest[16..20].try_into().unwrap()) as usize;
            if count != expected_values || count < 2 {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "linear fast field value count is inconsistent",
                ));
            }
            let expected_len = 29usize
                .checked_add(packed_len(count, rest[28])?)
                .ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidData, "fast field size overflow")
                })?;
            if rest.len() != expected_len {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "linear fast field length is inconsistent",
                ));
            }
        }
        Some(CodecType::BlockwiseLinear) => {
            if rest.len() < 8 {
                return Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "blockwise fast field header is truncated",
                ));
            }
            let count = u32::from_le_bytes(rest[0..4].try_into().unwrap()) as usize;
            let num_blocks = u32::from_le_bytes(rest[4..8].try_into().unwrap()) as usize;
            if count != expected_values || num_blocks != count.div_ceil(BLOCKWISE_LINEAR_BLOCK_SIZE)
            {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "blockwise fast field counts are inconsistent",
                ));
            }

            let mut pos = 8usize;
            for block_idx in 0..num_blocks {
                let header_end = pos.checked_add(29).ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidData, "fast field offset overflow")
                })?;
                if header_end > rest.len() {
                    return Err(io::Error::new(
                        io::ErrorKind::UnexpectedEof,
                        "blockwise fast field block header is truncated",
                    ));
                }
                let bpv = rest[pos + 24];
                let declared =
                    u32::from_le_bytes(rest[pos + 25..header_end].try_into().unwrap()) as usize;
                let block_start = block_idx * BLOCKWISE_LINEAR_BLOCK_SIZE;
                let block_count = (count - block_start).min(BLOCKWISE_LINEAR_BLOCK_SIZE);
                let expected = packed_len(block_count, bpv)?;
                if declared != expected {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        "blockwise fast field packed length is inconsistent",
                    ));
                }
                pos = header_end.checked_add(declared).ok_or_else(|| {
                    io::Error::new(io::ErrorKind::InvalidData, "fast field offset overflow")
                })?;
                if pos > rest.len() {
                    return Err(io::Error::new(
                        io::ErrorKind::UnexpectedEof,
                        "blockwise fast field data is truncated",
                    ));
                }
            }
            if pos != rest.len() {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    "blockwise fast field contains trailing data",
                ));
            }
        }
        None => {
            return Err(io::Error::new(
                io::ErrorKind::InvalidData,
                "unknown fast field codec",
            ));
        }
    }
    Ok(())
}

// ── Estimator trait ──────────────────────────────────────────────────────

/// Estimates serialized size for a given codec.
///
/// Usage: call `collect(val)` for every value, then `finalize()`,
/// then `estimate()` returns the byte count.
pub trait CodecEstimator {
    fn collect(&mut self, value: u64);
    fn finalize(&mut self) {}
    fn estimate(&self) -> Option<u64>;
    fn serialize(&self, values: &[u64], writer: &mut dyn Write) -> io::Result<u64>;
}

// ── Constant codec ───────────────────────────────────────────────────────

/// All values are identical → zero data bytes. Value stored in the codec header.
#[derive(Default)]
pub struct ConstantEstimator {
    first: Option<u64>,
    all_same: bool,
}

impl CodecEstimator for ConstantEstimator {
    fn collect(&mut self, value: u64) {
        match self.first {
            None => {
                self.first = Some(value);
                self.all_same = true;
            }
            Some(f) => {
                if value != f {
                    self.all_same = false;
                }
            }
        }
    }

    fn estimate(&self) -> Option<u64> {
        if self.all_same {
            // codec_id(1) + value(8) = 9 bytes
            Some(9)
        } else {
            None
        }
    }

    fn serialize(&self, values: &[u64], writer: &mut dyn Write) -> io::Result<u64> {
        let val = if values.is_empty() { 0 } else { values[0] };
        writer.write_u8(CodecType::Constant as u8)?;
        writer.write_u64::<LittleEndian>(val)?;
        Ok(9)
    }
}

// ── Bitpacked codec ──────────────────────────────────────────────────────

/// Min-subtract + global bitpack. This is the existing codec, now behind a tag.
#[derive(Default)]
pub struct BitpackedEstimator {
    min: u64,
    max: u64,
    count: usize,
    initialized: bool,
}

impl CodecEstimator for BitpackedEstimator {
    fn collect(&mut self, value: u64) {
        if !self.initialized {
            self.min = value;
            self.max = value;
            self.initialized = true;
        } else {
            self.min = self.min.min(value);
            self.max = self.max.max(value);
        }
        self.count += 1;
    }

    fn estimate(&self) -> Option<u64> {
        if self.count == 0 {
            return Some(0);
        }
        let range = self.max - self.min;
        let bpv = bits_needed_u64(range) as u64;
        // codec_id(1) + min(8) + bpv(1) + packed data
        let data_bits = self.count as u64 * bpv;
        let data_bytes = data_bits.div_ceil(8);
        Some(1 + 8 + 1 + data_bytes)
    }

    fn serialize(&self, values: &[u64], writer: &mut dyn Write) -> io::Result<u64> {
        let (min_value, bpv) = if values.is_empty() {
            (0u64, 0u8)
        } else {
            let min_val = values.iter().copied().min().unwrap();
            let max_val = values.iter().copied().max().unwrap();
            (min_val, bits_needed_u64(max_val - min_val))
        };

        writer.write_u8(CodecType::Bitpacked as u8)?;
        writer.write_u64::<LittleEndian>(min_value)?;
        writer.write_u8(bpv)?;
        let mut bytes_written = 10u64; // 1 + 8 + 1

        if bpv > 0 && !values.is_empty() {
            let shifted: Vec<u64> = values.iter().map(|&v| v - min_value).collect();
            let mut packed = Vec::new();
            bitpack_write(&shifted, bpv, &mut packed);
            writer.write_all(&packed)?;
            bytes_written += packed.len() as u64;
        }
        Ok(bytes_written)
    }
}

/// Read a single value from a bitpacked-codec column.
///
/// `data` starts right after the codec_id byte (i.e. at min_value).
#[inline]
pub fn bitpacked_read(data: &[u8], index: usize) -> u64 {
    let min_value = u64::from_le_bytes(data[0..8].try_into().unwrap());
    let bpv = data[8];
    if bpv == 0 {
        return min_value;
    }
    let packed = &data[9..];
    bitpack_read(packed, bpv, index).wrapping_add(min_value)
}

// ── Linear codec ─────────────────────────────────────────────────────────

/// Fits y = slope * x + intercept across all values, stores residuals bitpacked.
///
/// Header: codec_id(1) + intercept(8) + slope_num(8) + slope_den(8) + bpv(1) + offset(8) = 34
///
/// Estimation uses O(1) memory by tracking value extremes during collection and
/// computing worst-case residual bounds in `finalize()`.
///
/// **Limitation**: the per-column offset is stored as i64 (8 bytes). When values
/// span nearly the full u64 range (e.g. `FAST_FIELD_MISSING` mixed with small
/// values), residuals can exceed i64 bounds.  The estimator returns `None` in
/// that case so the auto-selector falls back to bitpacked.
#[derive(Default)]
pub struct LinearEstimator {
    count: usize,
    first: u64,
    last: u64,
    min_val: u64,
    max_val: u64,
    min_residual: i64,
    max_residual: i64,
    values_collected: bool,
    /// Set by `finalize()` when residuals exceed i64 range.
    overflow: bool,
}

impl CodecEstimator for LinearEstimator {
    fn collect(&mut self, value: u64) {
        if !self.values_collected {
            self.first = value;
            self.min_val = value;
            self.max_val = value;
            self.values_collected = true;
        } else {
            self.min_val = self.min_val.min(value);
            self.max_val = self.max_val.max(value);
        }
        self.last = value;
        self.count += 1;
    }

    fn finalize(&mut self) {
        if self.count < 2 {
            return;
        }
        // Compute worst-case residual bounds from value extremes vs predicted line.
        // The predicted line spans [first, last]. The worst-case residuals occur when
        // the most extreme value is farthest from the nearest predicted value.
        // Predicted values range from min(first,last) to max(first,last), so:
        //   max_residual ≥ max_val - min(predicted) = max_val - min(first, last)
        //   min_residual ≤ min_val - max(predicted) = min_val - max(first, last)
        // This is a conservative bound (may slightly overestimate bpv vs exact).
        let pred_min = self.first.min(self.last) as i128;
        let pred_max = self.first.max(self.last) as i128;
        let min_res = self.min_val as i128 - pred_max;
        let max_res = self.max_val as i128 - pred_min;
        // The offset is stored as i64 on disk.  If residuals exceed i64 range,
        // this codec cannot represent the data — mark as overflow.
        if min_res < i64::MIN as i128 || max_res > i64::MAX as i128 {
            self.overflow = true;
            return;
        }
        self.min_residual = min_res as i64;
        self.max_residual = max_res as i64;
    }

    fn estimate(&self) -> Option<u64> {
        if self.count < 2 || self.overflow {
            return None;
        }
        // Check for overflow: if the range doesn't fit u64, this codec is not viable
        let range = (self.max_residual as i128 - self.min_residual as i128) as u64;
        let bpv = bits_needed_u64(range) as u64;
        let data_bits = self.count as u64 * bpv;
        let data_bytes = data_bits.div_ceil(8);
        // codec_id(1) + first(8) + last(8) + num_values(4) + offset(8) + bpv(1) + packed
        Some(1 + 8 + 8 + 4 + 8 + 1 + data_bytes)
    }

    fn serialize(&self, values: &[u64], writer: &mut dyn Write) -> io::Result<u64> {
        let n = values.len();
        if n < 2 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "linear needs ≥ 2 values",
            ));
        }
        let first = values[0];
        let last = values[n - 1];

        // Compute residuals using i128 to avoid overflow
        let mut min_residual = i128::MAX;
        for (i, &val) in values.iter().enumerate() {
            let predicted = interpolate(first, last, n, i);
            let residual = val as i128 - predicted as i128;
            min_residual = min_residual.min(residual);
        }

        // The offset field is i64 on disk — reject data that doesn't fit.
        if min_residual < i64::MIN as i128 || min_residual > i64::MAX as i128 {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "linear codec: residual offset exceeds i64 range",
            ));
        }
        let min_residual_i64 = min_residual as i64;

        // Shift residuals to non-negative
        let shifted: Vec<u64> = values
            .iter()
            .enumerate()
            .map(|(i, &val)| {
                let predicted = interpolate(first, last, n, i);
                let residual = val as i128 - predicted as i128;
                (residual - min_residual) as u64
            })
            .collect();
        let max_shifted = shifted.iter().copied().max().unwrap_or(0);
        let bpv = bits_needed_u64(max_shifted);
        writer.write_u8(CodecType::Linear as u8)?;
        writer.write_u64::<LittleEndian>(first)?;
        writer.write_u64::<LittleEndian>(last)?;
        writer.write_u32::<LittleEndian>(n as u32)?;
        writer.write_i64::<LittleEndian>(min_residual_i64)?;
        writer.write_u8(bpv)?;
        let mut bytes_written = 30u64; // 1+8+8+4+8+1

        if bpv > 0 {
            let mut packed = Vec::new();
            bitpack_write(&shifted, bpv, &mut packed);
            writer.write_all(&packed)?;
            bytes_written += packed.len() as u64;
        }

        Ok(bytes_written)
    }
}

/// Interpolate value at index `i` on the line from first to last over `n` values.
#[inline]
fn interpolate(first: u64, last: u64, n: usize, i: usize) -> u64 {
    if n <= 1 {
        return first;
    }
    // Use i128 to avoid overflow
    let first = first as i128;
    let last = last as i128;
    let n = n as i128;
    let i = i as i128;
    let result = first + (last - first) * i / (n - 1);
    result as u64
}

/// Read a single value from a linear-codec column.
///
/// `data` starts right after the codec_id byte.
#[inline]
pub fn linear_read(data: &[u8], index: usize) -> u64 {
    let first = u64::from_le_bytes(data[0..8].try_into().unwrap());
    let last = u64::from_le_bytes(data[8..16].try_into().unwrap());
    let n = u32::from_le_bytes(data[16..20].try_into().unwrap()) as usize;
    let offset = i64::from_le_bytes(data[20..28].try_into().unwrap());
    let bpv = data[28];
    let predicted = interpolate(first, last, n, index);
    let residual = if bpv == 0 {
        0u64
    } else {
        bitpack_read(&data[29..], bpv, index)
    };
    // Use i128 to avoid overflow with large values
    (predicted as i128 + offset as i128 + residual as i128) as u64
}

// ── BlockwiseLinear codec ────────────────────────────────────────────────

/// Per-512-element-block linear interpolation with per-block bitpacked residuals.
///
/// Header: codec_id(1) + num_values(4) + num_blocks(4)
/// Per block: first(8) + last(8) + offset(8) + bpv(1) + packed_len(4) + packed_data
#[derive(Clone, Copy)]
struct BlockwiseLinearBlockEstimate {
    min_residual: i64,
    bits_per_value: u8,
    serialized_size: u64,
}

#[derive(Default)]
pub struct BlockwiseLinearEstimator {
    count: usize,
    completed_size: u64,
    current_block: Vec<u64>,
    completed_blocks: Vec<BlockwiseLinearBlockEstimate>,
    tail_estimate: Option<BlockwiseLinearBlockEstimate>,
    overflow: bool,
}

impl BlockwiseLinearEstimator {
    /// Collect a complete column without retaining it.
    ///
    /// `serialize_auto` already owns the input slice, so full blocks can be
    /// estimated in place. Keeping this pass separate from the other
    /// per-value estimators also leaves their hot loop branch-free.
    fn collect_values(&mut self, values: &[u64]) {
        debug_assert_eq!(self.count, 0);
        debug_assert!(self.current_block.is_empty());

        self.count = values.len();
        let mut blocks = values.chunks_exact(BLOCKWISE_LINEAR_BLOCK_SIZE);
        for block in &mut blocks {
            match estimate_blockwise_linear_block(block) {
                Some(estimate) => {
                    self.completed_size += estimate.serialized_size;
                    self.completed_blocks.push(estimate);
                }
                None => {
                    self.overflow = true;
                    return;
                }
            }
        }
        self.current_block.extend_from_slice(blocks.remainder());
    }
}

/// Return the serialized size of one block, excluding the global header.
///
/// A block is buffered until its final value is known because that value is
/// part of the interpolation line. Keeping only this bounded scratch block
/// avoids retaining a second copy of the entire column during auto-selection.
fn estimate_blockwise_linear_block(block: &[u64]) -> Option<BlockwiseLinearBlockEstimate> {
    debug_assert!(!block.is_empty());
    let block_len = block.len();
    if block_len < 2 {
        return Some(BlockwiseLinearBlockEstimate {
            min_residual: 0,
            bits_per_value: 0,
            serialized_size: 29,
        });
    }

    let first = block[0];
    let last = block[block_len - 1];
    let mut min_res = i128::MAX;
    let mut max_res = i128::MIN;
    for (i, &val) in block.iter().enumerate() {
        let pred = interpolate(first, last, block_len, i);
        let res = val as i128 - pred as i128;
        min_res = min_res.min(res);
        max_res = max_res.max(res);
    }

    // Per-block offset is stored as i64. If either residual bound cannot be
    // represented, the codec cannot encode this column.
    if min_res < i64::MIN as i128 || max_res > i64::MAX as i128 {
        return None;
    }

    let bits_per_value = bits_needed_u64((max_res - min_res) as u64);
    let data_bytes = (block_len as u64 * u64::from(bits_per_value)).div_ceil(8);
    Some(BlockwiseLinearBlockEstimate {
        min_residual: min_res as i64,
        bits_per_value,
        serialized_size: 29 + data_bytes,
    })
}

impl CodecEstimator for BlockwiseLinearEstimator {
    fn collect(&mut self, value: u64) {
        self.count += 1;
        self.tail_estimate = None;
        if self.overflow {
            return;
        }

        self.current_block.push(value);
        if self.current_block.len() == BLOCKWISE_LINEAR_BLOCK_SIZE {
            match estimate_blockwise_linear_block(&self.current_block) {
                Some(estimate) => {
                    self.completed_size += estimate.serialized_size;
                    self.completed_blocks.push(estimate);
                }
                None => self.overflow = true,
            }
            self.current_block.clear();
        }
    }

    fn finalize(&mut self) {
        self.tail_estimate = if self.current_block.is_empty() || self.overflow {
            None
        } else {
            estimate_blockwise_linear_block(&self.current_block)
        };
        if !self.current_block.is_empty() && self.tail_estimate.is_none() {
            self.overflow = true;
        }
    }

    fn estimate(&self) -> Option<u64> {
        if self.count < 2 * BLOCKWISE_LINEAR_BLOCK_SIZE || self.overflow {
            // Only useful when there are enough values to amortize the per-block headers
            return None;
        }

        // codec_id(1) + num_values(4) + num_blocks(4)
        let mut total = 9 + self.completed_size;
        if !self.current_block.is_empty() {
            total += self
                .tail_estimate
                .or_else(|| estimate_blockwise_linear_block(&self.current_block))?
                .serialized_size;
        }
        Some(total)
    }

    fn serialize(&self, values: &[u64], writer: &mut dyn Write) -> io::Result<u64> {
        let n = values.len();
        let num_blocks = n.div_ceil(BLOCKWISE_LINEAR_BLOCK_SIZE);

        writer.write_u8(CodecType::BlockwiseLinear as u8)?;
        writer.write_u32::<LittleEndian>(n as u32)?;
        writer.write_u32::<LittleEndian>(num_blocks as u32)?;
        let mut bytes_written = 9u64;

        // Both scratch buffers are bounded by one block and reused across all
        // blocks, avoiding one allocation pair per block.
        let mut shifted = Vec::new();
        let mut packed = Vec::new();
        let estimates_match = self.count == n
            && self.completed_blocks.len() == n / BLOCKWISE_LINEAR_BLOCK_SIZE
            && (n.is_multiple_of(BLOCKWISE_LINEAR_BLOCK_SIZE) || self.tail_estimate.is_some());

        for b in 0..num_blocks {
            let start = b * BLOCKWISE_LINEAR_BLOCK_SIZE;
            let end = (start + BLOCKWISE_LINEAR_BLOCK_SIZE).min(n);
            let block = &values[start..end];
            let block_len = block.len();

            let first = block[0];
            let last = if block_len > 1 {
                block[block_len - 1]
            } else {
                first
            };

            let estimate = if estimates_match {
                self.completed_blocks.get(b).copied().or(self.tail_estimate)
            } else {
                None
            };
            let estimate = match estimate {
                Some(estimate) => estimate,
                None => estimate_blockwise_linear_block(block).ok_or_else(|| {
                    io::Error::new(
                        io::ErrorKind::InvalidInput,
                        "blockwise linear codec: per-block residual offset exceeds i64 range",
                    )
                })?,
            };
            let min_residual = i128::from(estimate.min_residual);

            shifted.clear();
            shifted.extend(block.iter().enumerate().map(|(i, &val)| {
                if block_len < 2 {
                    0
                } else {
                    let pred = interpolate(first, last, block_len, i);
                    let res = val as i128 - pred as i128;
                    (res - min_residual) as u64
                }
            }));
            writer.write_u64::<LittleEndian>(first)?;
            writer.write_u64::<LittleEndian>(last)?;
            writer.write_i64::<LittleEndian>(estimate.min_residual)?;
            writer.write_u8(estimate.bits_per_value)?;

            packed.clear();
            if estimate.bits_per_value > 0 {
                bitpack_write(&shifted, estimate.bits_per_value, &mut packed);
            }
            writer.write_u32::<LittleEndian>(packed.len() as u32)?;
            writer.write_all(&packed)?;
            bytes_written += 29 + packed.len() as u64;
        }

        Ok(bytes_written)
    }
}

/// Read a single value from a blockwise-linear-codec column.
///
/// `data` starts right after the codec_id byte.
pub fn blockwise_linear_read(data: &[u8], index: usize) -> u64 {
    blockwise_linear_read_from(data, index, 0, 8)
}

/// Resume from a validated header checkpoint (offset excludes the codec byte).
pub(super) fn blockwise_linear_read_from(
    data: &[u8],
    index: usize,
    first_block: usize,
    mut pos: usize,
) -> u64 {
    let _num_values = u32::from_le_bytes(data[0..4].try_into().unwrap()) as usize;
    let num_blocks = u32::from_le_bytes(data[4..8].try_into().unwrap()) as usize;
    let target_block = index / BLOCKWISE_LINEAR_BLOCK_SIZE;
    let index_in_block = index % BLOCKWISE_LINEAR_BLOCK_SIZE;
    for b in first_block..num_blocks {
        let first = u64::from_le_bytes(data[pos..pos + 8].try_into().unwrap());
        let last = u64::from_le_bytes(data[pos + 8..pos + 16].try_into().unwrap());
        let offset = i64::from_le_bytes(data[pos + 16..pos + 24].try_into().unwrap());
        let bpv = data[pos + 24];
        let packed_len = u32::from_le_bytes(data[pos + 25..pos + 29].try_into().unwrap()) as usize;

        if b == target_block {
            let block_start = b * BLOCKWISE_LINEAR_BLOCK_SIZE;
            let block_end = ((b + 1) * BLOCKWISE_LINEAR_BLOCK_SIZE).min(_num_values);
            let block_len = block_end - block_start;

            let predicted = interpolate(first, last, block_len, index_in_block);
            let residual = if bpv == 0 {
                0u64
            } else {
                bitpack_read(&data[pos + 29..], bpv, index_in_block)
            };
            return (predicted as i128 + offset as i128 + residual as i128) as u64;
        }

        pos += 29 + packed_len;
    }

    0 // Should not reach here
}

/// Batch-read consecutive values from a blockwise-linear column.
///
/// Variable-length block records are scanned once to reach `start_index`, then
/// consumed in order. This avoids re-scanning every preceding block header for
/// each value in the batch.
pub fn blockwise_linear_read_batch(data: &[u8], start_index: usize, out: &mut [u64]) {
    blockwise_linear_read_batch_with_cursor(
        data,
        start_index,
        out,
        &mut BlockwiseLinearCursor::default(),
    );
}

/// Position in one admitted column payload. Reuse only for the same payload;
/// copied column blocks each start with a fresh cursor. No payload is retained.
pub(super) struct BlockwiseLinearCursor {
    block: usize,
    offset: usize,
}

impl Default for BlockwiseLinearCursor {
    fn default() -> Self {
        Self {
            block: 0,
            offset: 8,
        }
    }
}

fn blockwise_linear_read_batch_with_cursor(
    data: &[u8],
    start_index: usize,
    out: &mut [u64],
    cursor: &mut BlockwiseLinearCursor,
) {
    if out.is_empty() {
        return;
    }

    let num_values = u32::from_le_bytes(data[0..4].try_into().unwrap()) as usize;
    let num_blocks = u32::from_le_bytes(data[4..8].try_into().unwrap()) as usize;
    let valid_len = out.len().min(num_values.saturating_sub(start_index));
    out[valid_len..].fill(0);
    if valid_len == 0 {
        return;
    }

    let target_block = start_index / BLOCKWISE_LINEAR_BLOCK_SIZE;
    if target_block < cursor.block {
        *cursor = BlockwiseLinearCursor::default();
    }
    let mut pos = cursor.offset;
    let mut written = 0usize;

    for block_idx in cursor.block..num_blocks {
        let packed_len = u32::from_le_bytes(data[pos + 25..pos + 29].try_into().unwrap()) as usize;
        if block_idx < target_block {
            pos += 29 + packed_len;
            continue;
        }

        let first = u64::from_le_bytes(data[pos..pos + 8].try_into().unwrap());
        let last = u64::from_le_bytes(data[pos + 8..pos + 16].try_into().unwrap());
        let offset = i64::from_le_bytes(data[pos + 16..pos + 24].try_into().unwrap());
        let bpv = data[pos + 24];
        let packed = &data[pos + 29..pos + 29 + packed_len];

        let block_start = block_idx * BLOCKWISE_LINEAR_BLOCK_SIZE;
        let block_len = (num_values - block_start).min(BLOCKWISE_LINEAR_BLOCK_SIZE);
        let index_in_block = if block_idx == target_block {
            start_index - block_start
        } else {
            0
        };
        let take = (block_len - index_in_block).min(valid_len - written);

        for (i, value) in out[written..written + take].iter_mut().enumerate() {
            let block_index = index_in_block + i;
            let predicted = interpolate(first, last, block_len, block_index);
            let residual = if bpv == 0 {
                0
            } else {
                bitpack_read(packed, bpv, block_index)
            };
            *value = (predicted as i128 + offset as i128 + residual as i128) as u64;
        }

        written += take;
        if index_in_block + take == block_len {
            cursor.block = block_idx + 1;
            cursor.offset = pos + 29 + packed_len;
        } else {
            cursor.block = block_idx;
            cursor.offset = pos;
        }
        if written == valid_len {
            break;
        }
        pos += 29 + packed_len;
    }
}

// ── Auto-selection ───────────────────────────────────────────────────────

/// Serialize values using the codec that produces the smallest output.
///
/// Returns the number of bytes written.
pub fn serialize_auto(values: &[u64], writer: &mut dyn Write) -> io::Result<u64> {
    let mut constant = ConstantEstimator::default();
    let mut bitpacked = BitpackedEstimator::default();
    let mut linear = LinearEstimator::default();
    let mut blockwise = BlockwiseLinearEstimator::default();

    // Pass 1: collect
    for &v in values {
        constant.collect(v);
        bitpacked.collect(v);
        linear.collect(v);
    }
    blockwise.collect_values(values);

    // Finalize
    constant.finalize();
    bitpacked.finalize();
    linear.finalize();
    blockwise.finalize();

    // Pick smallest
    let candidates: Vec<(&dyn CodecEstimator, &str)> = vec![
        (&constant, "constant"),
        (&bitpacked, "bitpacked"),
        (&linear, "linear"),
        (&blockwise, "blockwise_linear"),
    ];

    let (best, _name) = candidates
        .into_iter()
        .filter_map(|(est, name)| est.estimate().map(|size| (est, name, size)))
        .min_by_key(|&(_, _, size)| size)
        .map(|(est, name, _)| (est, name))
        .unwrap_or((&bitpacked as &dyn CodecEstimator, "bitpacked"));

    best.serialize(values, writer)
}

/// Batch-read `out.len()` consecutive values starting at `start_index` from bitpacked data.
///
/// `data` starts right after the codec_id byte (at min_value).
/// Byte-aligned bpv (8, 16, 32, 64) use fixed-width chunks with upfront range
/// checks to enable auto-vectorization; inspect the target/profile's codegen.
/// For arbitrary bpv, uses a tight scalar loop with the u64 fast-path.
pub fn bitpacked_read_batch(data: &[u8], start_index: usize, out: &mut [u64]) {
    let min_value = u64::from_le_bytes(data[0..8].try_into().unwrap());
    let bpv = data[8];

    if bpv == 0 {
        out.iter_mut().for_each(|v| *v = min_value);
        return;
    }

    let packed = &data[9..];

    match bpv {
        // Prove the whole byte range once so the inner loop has no per-value
        // input bounds checks. Generic decode closures inline into each width.
        8 => {
            decode_byte_aligned_batch::<1>(packed, start_index, out, min_value, |v| u64::from(v[0]))
        }
        16 => decode_byte_aligned_batch::<2>(packed, start_index, out, min_value, |v| {
            u64::from(u16::from_le_bytes(v))
        }),
        32 => decode_byte_aligned_batch::<4>(packed, start_index, out, min_value, |v| {
            u64::from(u32::from_le_bytes(v))
        }),
        64 => {
            decode_byte_aligned_batch::<8>(packed, start_index, out, min_value, u64::from_le_bytes)
        }
        // Arbitrary bpv — tight scalar loop using u64 fast-path read
        _ => {
            for (i, v) in out.iter_mut().enumerate() {
                *v = super::bitpack_read(packed, bpv, start_index + i).wrapping_add(min_value);
            }
        }
    }
}

/// Select exactly one input chunk per output before entering the decode loop.
/// The upfront slice checks also prevent zip from silently accepting short data.
#[inline]
fn decode_byte_aligned_batch<const WIDTH: usize>(
    packed: &[u8],
    start: usize,
    out: &mut [u64],
    min: u64,
    decode: impl Fn([u8; WIDTH]) -> u64,
) {
    if out.is_empty() {
        return;
    }
    let byte_start = start.checked_mul(WIDTH).expect("bitpacked start overflow");
    let byte_len = out
        .len()
        .checked_mul(WIDTH)
        .expect("bitpacked length overflow");
    let bytes = &packed[byte_start..][..byte_len];
    let (chunks, _) = bytes.as_chunks::<WIDTH>();
    for (value, &chunk) in out.iter_mut().zip(chunks) {
        *value = decode(chunk).wrapping_add(min);
    }
}

/// Batch-read `out.len()` consecutive values starting at `start_index` from auto-codec data.
///
/// Dispatches codec type once (vs. per-value in `auto_read`), enabling tight inner
/// loops that the compiler auto-vectorizes for byte-aligned bitpacked columns.
pub fn auto_read_batch(data: &[u8], start_index: usize, out: &mut [u64]) {
    auto_read_batch_with_cursor(
        data,
        start_index,
        out,
        &mut BlockwiseLinearCursor::default(),
    );
}

/// Batch decode with an optional sequential advantage for BlockwiseLinear.
/// The caller must reset the cursor when switching column payloads.
pub(super) fn auto_read_batch_with_cursor(
    data: &[u8],
    start_index: usize,
    out: &mut [u64],
    cursor: &mut BlockwiseLinearCursor,
) {
    if data.is_empty() || out.is_empty() {
        out.iter_mut().for_each(|v| *v = 0);
        return;
    }
    let codec_id = data[0];
    let rest = &data[1..];
    match CodecType::from_u8(codec_id) {
        Some(CodecType::Constant) => {
            let val = u64::from_le_bytes(rest[0..8].try_into().unwrap());
            out.iter_mut().for_each(|v| *v = val);
        }
        Some(CodecType::Bitpacked) => bitpacked_read_batch(rest, start_index, out),
        Some(CodecType::Linear) => {
            for (i, v) in out.iter_mut().enumerate() {
                *v = linear_read(rest, start_index + i);
            }
        }
        Some(CodecType::BlockwiseLinear) => {
            blockwise_linear_read_batch_with_cursor(rest, start_index, out, cursor)
        }
        None => out.iter_mut().for_each(|v| *v = 0),
    }
}

/// Read a single value from auto-codec encoded data.
///
/// The first byte identifies the codec.
#[inline]
pub fn auto_read(data: &[u8], index: usize) -> u64 {
    if data.is_empty() {
        return 0;
    }
    let codec_id = data[0];
    let rest = &data[1..];
    match CodecType::from_u8(codec_id) {
        Some(CodecType::Constant) => {
            // rest = value(8)
            u64::from_le_bytes(rest[0..8].try_into().unwrap())
        }
        Some(CodecType::Bitpacked) => bitpacked_read(rest, index),
        Some(CodecType::Linear) => linear_read(rest, index),
        Some(CodecType::BlockwiseLinear) => blockwise_linear_read(rest, index),
        None => 0,
    }
}

// ── Tests ────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn cursor_batches_preserve_all_codecs_boundaries_and_backward_reads() {
        let values = blockwise_values(2053);
        let mut constant = Vec::new();
        ConstantEstimator::default()
            .serialize(&[42; 2053], &mut constant)
            .unwrap();
        let mut bitpacked = Vec::new();
        let mut estimator = BitpackedEstimator::default();
        for &value in &values {
            estimator.collect(value);
        }
        estimator.finalize();
        estimator.serialize(&values, &mut bitpacked).unwrap();
        let descending: Vec<_> = (0..2053).map(|i| u64::MAX - i * 7).collect();
        let mut linear = Vec::new();
        let mut estimator = LinearEstimator::default();
        for &value in &descending {
            estimator.collect(value);
        }
        estimator.finalize();
        estimator.serialize(&descending, &mut linear).unwrap();
        let mut blockwise = Vec::new();
        BlockwiseLinearEstimator::default()
            .serialize(&values, &mut blockwise)
            .unwrap();
        assert_eq!(blockwise, serialize_blockwise_reference(&values));
        for (tag, encoded) in [constant, bitpacked, linear, blockwise].iter().enumerate() {
            assert_eq!(usize::from(encoded[0]), tag);
            validate_auto(encoded, 2053).unwrap();
            let expected: Vec<_> = (0..2053).map(|i| auto_read(encoded, i)).collect();
            for size in [1, 255, 256, 257, 511, 512, 513, 1000, 2053] {
                let mut cursor = BlockwiseLinearCursor::default();
                let mut actual = vec![0; 2053];
                for (batch, out) in actual.chunks_mut(size).enumerate() {
                    auto_read_batch_with_cursor(encoded, batch * size, &mut [], &mut cursor);
                    auto_read_batch_with_cursor(encoded, batch * size, out, &mut cursor);
                }
                assert_eq!(actual, expected, "codec {tag}, batch {size}");
                for start in [1023, 511, 0, 2048] {
                    let mut out = [0; 5];
                    auto_read_batch_with_cursor(encoded, start, &mut out, &mut cursor);
                    assert_eq!(out, expected[start..start + 5]);
                }
                if tag == CodecType::BlockwiseLinear as usize {
                    let mut out = [u64::MAX; 17];
                    auto_read_batch_with_cursor(encoded, 2050, &mut out, &mut cursor);
                    assert_eq!(&out[..3], &expected[2050..]);
                    assert_eq!(&out[3..], &[0; 14]);
                    auto_read_batch_with_cursor(encoded, usize::MAX, &mut out, &mut cursor);
                    assert_eq!(out, [0; 17]);
                }
            }
        }
    }

    #[test]
    fn header_bounds_contain_every_bitpacked_value_including_wrapping_payloads() {
        for bits in 0..=64 {
            let max = u64::MAX.checked_shr(64 - bits).unwrap_or(0);
            for min in [0u64, 17, u64::MAX - 17, u64::MAX] {
                let raw = [0, max / 2, max];
                let mut encoded = vec![CodecType::Bitpacked as u8];
                encoded.extend_from_slice(&min.to_le_bytes());
                encoded.push(bits as u8);
                bitpack_write(&raw, bits as u8, &mut encoded);
                validate_auto(&encoded, raw.len()).unwrap();
                let bounds = value_bounds(&encoded);
                if min.checked_add(max).is_none() {
                    assert_eq!(bounds, None, "wrapping bounds must not prune");
                } else {
                    assert_eq!(bounds, Some((min, min + max)));
                }
                for i in 0..raw.len() {
                    let value = auto_read(&encoded, i);
                    assert!(bounds.is_none_or(|(lo, hi)| value >= lo && value <= hi));
                }
            }
        }
    }

    #[test]
    fn linear_header_bounds_preserve_descending_extreme_and_wrapping_values() {
        for first in [0u64, 100, u64::MAX - 100] {
            for last in [0u64, 100, u64::MAX - 100] {
                for offset in [i64::MIN, -17, 0, 17, i64::MAX] {
                    for bits in [0u8, 1, 4, 64] {
                        let max = u64::MAX.checked_shr(u32::from(64 - bits)).unwrap_or(0);
                        let raw: Vec<_> =
                            (0..17).map(|i| if i % 2 == 0 { max } else { 0 }).collect();
                        let mut encoded = vec![CodecType::Linear as u8];
                        encoded.extend_from_slice(&first.to_le_bytes());
                        encoded.extend_from_slice(&last.to_le_bytes());
                        encoded.extend_from_slice(&17u32.to_le_bytes());
                        encoded.extend_from_slice(&offset.to_le_bytes());
                        encoded.push(bits);
                        bitpack_write(&raw, bits, &mut encoded);
                        validate_auto(&encoded, raw.len()).unwrap();
                        let bounds = value_bounds(&encoded);
                        for i in 0..raw.len() {
                            let value = auto_read(&encoded, i);
                            assert!(bounds.is_none_or(|(lo, hi)| value >= lo && value <= hi));
                        }
                    }
                }
            }
        }
        let encoded = [CodecType::Constant as u8]
            .into_iter()
            .chain(u64::MAX.to_le_bytes())
            .collect::<Vec<_>>();
        assert_eq!(value_bounds(&encoded), Some((u64::MAX, u64::MAX)));
    }

    fn roundtrip(values: &[u64]) -> Vec<u64> {
        let mut buf = Vec::new();
        serialize_auto(values, &mut buf).unwrap();
        (0..values.len()).map(|i| auto_read(&buf, i)).collect()
    }

    fn blockwise_values(len: usize) -> Vec<u64> {
        (0..len)
            .map(|i| {
                let block = i / BLOCKWISE_LINEAR_BLOCK_SIZE;
                let index = i % BLOCKWISE_LINEAR_BLOCK_SIZE;
                block as u64 * 1_000_000
                    + index as u64 * (block as u64 + 3)
                    + ((i * 17 + block * 11) % 23) as u64
            })
            .collect()
    }

    /// Reference implementation matching the original allocation-per-block
    /// serializer. This guards the on-disk representation while the production
    /// implementation reuses bounded scratch buffers.
    fn serialize_blockwise_reference(values: &[u64]) -> Vec<u8> {
        let n = values.len();
        let num_blocks = n.div_ceil(BLOCKWISE_LINEAR_BLOCK_SIZE);
        let mut encoded = Vec::new();
        encoded.push(CodecType::BlockwiseLinear as u8);
        encoded.extend_from_slice(&(n as u32).to_le_bytes());
        encoded.extend_from_slice(&(num_blocks as u32).to_le_bytes());

        for block in values.chunks(BLOCKWISE_LINEAR_BLOCK_SIZE) {
            let block_len = block.len();
            let first = block[0];
            let last = if block_len > 1 {
                block[block_len - 1]
            } else {
                first
            };
            let min_residual = if block_len < 2 {
                0
            } else {
                block
                    .iter()
                    .enumerate()
                    .map(|(i, &value)| {
                        value as i128 - interpolate(first, last, block_len, i) as i128
                    })
                    .min()
                    .unwrap()
            };
            let shifted: Vec<u64> = block
                .iter()
                .enumerate()
                .map(|(i, &value)| {
                    if block_len < 2 {
                        0
                    } else {
                        let predicted = interpolate(first, last, block_len, i);
                        (value as i128 - predicted as i128 - min_residual) as u64
                    }
                })
                .collect();
            let bpv = bits_needed_u64(shifted.iter().copied().max().unwrap_or(0));
            let mut packed = Vec::new();
            bitpack_write(&shifted, bpv, &mut packed);

            encoded.extend_from_slice(&first.to_le_bytes());
            encoded.extend_from_slice(&last.to_le_bytes());
            encoded.extend_from_slice(&(min_residual as i64).to_le_bytes());
            encoded.push(bpv);
            encoded.extend_from_slice(&(packed.len() as u32).to_le_bytes());
            encoded.extend_from_slice(&packed);
        }

        encoded
    }

    #[test]
    fn test_constant_codec() {
        let values: Vec<u64> = vec![42; 100];
        let mut buf = Vec::new();
        serialize_auto(&values, &mut buf).unwrap();
        assert_eq!(buf[0], CodecType::Constant as u8);
        assert_eq!(buf.len(), 9);
        assert_eq!(roundtrip(&values), values);
    }

    #[test]
    fn test_bitpacked_codec() {
        let values: Vec<u64> = (0..50).map(|i| 1000 + (i % 7) * 13).collect();
        let result = roundtrip(&values);
        assert_eq!(result, values);
    }

    #[test]
    fn test_linear_codec_sequential() {
        // Perfectly linear → 0 bpv residuals
        let values: Vec<u64> = (0..1000).map(|i| 100 + i * 3).collect();
        let mut buf = Vec::new();
        serialize_auto(&values, &mut buf).unwrap();
        // Should pick linear (smaller than bitpacked for sequential data)
        assert_eq!(roundtrip(&values), values);
    }

    #[test]
    fn test_blockwise_linear_codec() {
        // Two distinct linear segments
        let mut values: Vec<u64> = Vec::new();
        for i in 0..1500 {
            if i < 750 {
                values.push(100 + i * 2);
            } else {
                values.push(5000 + (i - 750) * 5);
            }
        }
        let result = roundtrip(&values);
        assert_eq!(result, values);
    }

    #[test]
    fn test_blockwise_estimate_is_bounded_and_serialization_is_byte_compatible() {
        for len in [1024, 1025, 1536, 1537, 4097] {
            let values = blockwise_values(len);
            let reference = serialize_blockwise_reference(&values);
            let mut estimator = BlockwiseLinearEstimator::default();
            for &value in &values {
                estimator.collect(value);
            }
            estimator.finalize();

            assert_eq!(estimator.estimate(), Some(reference.len() as u64));
            assert!(
                estimator.current_block.len() < BLOCKWISE_LINEAR_BLOCK_SIZE,
                "estimator retained more than one partial block"
            );
            assert!(
                estimator.current_block.capacity() <= BLOCKWISE_LINEAR_BLOCK_SIZE,
                "estimator scratch grew beyond one block"
            );

            let mut encoded = Vec::new();
            estimator.serialize(&values, &mut encoded).unwrap();
            assert_eq!(
                encoded, reference,
                "serialized bytes changed for {len} values"
            );
        }
    }

    #[test]
    fn test_empty() {
        let values: Vec<u64> = vec![];
        let mut buf = Vec::new();
        serialize_auto(&values, &mut buf).unwrap();
        assert!(buf.len() <= 10);
    }

    #[test]
    fn test_validate_rejects_truncated_and_inconsistent_payloads() {
        assert!(validate_auto(&[], 1).is_err());
        assert!(validate_auto(&[CodecType::Constant as u8], 1).is_err());

        let mut bitpacked = vec![CodecType::Bitpacked as u8];
        bitpacked.extend_from_slice(&0u64.to_le_bytes());
        bitpacked.push(65);
        assert!(validate_auto(&bitpacked, 1).is_err());

        let mut valid = Vec::new();
        serialize_auto(&[1, 2, 3, 4], &mut valid).unwrap();
        assert!(validate_auto(&valid, 4).is_ok());
        valid.pop();
        assert!(validate_auto(&valid, 4).is_err());
    }

    #[test]
    fn test_single_value() {
        let values = vec![999u64];
        assert_eq!(roundtrip(&values), values);
    }

    #[test]
    fn test_two_values() {
        let values = vec![10u64, 20];
        assert_eq!(roundtrip(&values), values);
    }

    #[test]
    fn test_large_range() {
        let values = vec![0u64, u64::MAX / 2, u64::MAX];
        assert_eq!(roundtrip(&values), values);
    }

    #[test]
    fn test_timestamps_pick_linear_or_blockwise() {
        // Simulate timestamps (monotonically increasing with small jitter)
        let mut values: Vec<u64> = Vec::new();
        let mut ts = 1_700_000_000u64;
        for _ in 0..2000 {
            values.push(ts);
            ts += 1000 + (ts % 7); // ~1000 with jitter
        }
        let result = roundtrip(&values);
        assert_eq!(result, values);
    }

    /// Helper: roundtrip via auto_read_batch and compare with per-element auto_read.
    fn roundtrip_batch(values: &[u64]) {
        let mut buf = Vec::new();
        serialize_auto(values, &mut buf).unwrap();

        // Batch read all values
        let mut batch_out = vec![0u64; values.len()];
        auto_read_batch(&buf, 0, &mut batch_out);
        assert_eq!(batch_out, values, "batch read mismatch");

        // Batch read a sub-range
        if values.len() >= 10 {
            let start = 3;
            let count = values.len() - 6;
            let mut sub = vec![0u64; count];
            auto_read_batch(&buf, start, &mut sub);
            assert_eq!(
                sub,
                &values[start..start + count],
                "sub-range batch mismatch"
            );
        }
    }

    #[test]
    fn test_batch_read_constant() {
        roundtrip_batch(&vec![42u64; 100]);
    }

    #[test]
    fn test_batch_read_bitpacked_8bit() {
        // Values with range < 256 → 8-bit bpv
        let values: Vec<u64> = (0..200).map(|i| 1000 + (i % 200)).collect();
        roundtrip_batch(&values);
    }

    #[test]
    fn test_batch_read_bitpacked_16bit() {
        // Values with range fitting 16 bits
        let values: Vec<u64> = (0..200).map(|i| 50000 + i * 100).collect();
        roundtrip_batch(&values);
    }

    #[test]
    fn test_batch_read_bitpacked_arbitrary() {
        // Arbitrary bpv (e.g. 13 bits)
        let values: Vec<u64> = (0..100).map(|i| 999 + (i * 37) % 8000).collect();
        roundtrip_batch(&values);
    }

    #[test]
    fn byte_aligned_batches_preserve_offsets_tails_and_wrapping_values() {
        for bpv in [8u8, 16, 32, 64] {
            let min = u64::MAX - 11;
            let mut data = min.to_le_bytes().to_vec();
            data.push(bpv);
            let mut expected = Vec::new();
            for i in 0..521u64 {
                let raw = i.wrapping_mul(0x9e37_79b9_7f4a_7c15) >> (64 - bpv);
                data.extend_from_slice(&raw.to_le_bytes()[..usize::from(bpv / 8)]);
                expected.push(raw.wrapping_add(min));
            }
            for start in [0, 1, 3, 255, 256, 519, 521] {
                for len in [0, 1, 2, 7, 16, 255, 256, 257] {
                    if start + len > expected.len() {
                        continue;
                    }
                    let mut out = vec![0; len];
                    bitpacked_read_batch(&data, start, &mut out);
                    assert_eq!(
                        out,
                        expected[start..start + len],
                        "bpv={bpv}, start={start}, len={len}"
                    );
                }
            }
            // Empty output performs no payload access, even with a large start.
            bitpacked_read_batch(&data, usize::MAX, &mut []);
        }
    }

    #[test]
    fn byte_aligned_batches_reject_truncated_input_instead_of_short_decoding() {
        for bpv in [8u8, 16, 32, 64] {
            let mut data = 0u64.to_le_bytes().to_vec();
            data.push(bpv);
            data.resize(9 + usize::from(bpv / 8) * 3 - 1, 0);
            assert!(
                std::panic::catch_unwind(|| {
                    bitpacked_read_batch(&data, 0, &mut [0u64; 3]);
                })
                .is_err()
            );
        }
    }

    #[test]
    fn test_batch_read_linear() {
        let values: Vec<u64> = (0..500).map(|i| 100 + i * 3).collect();
        roundtrip_batch(&values);
    }

    #[test]
    fn test_batch_read_blockwise() {
        let mut values = Vec::new();
        for i in 0..1500u64 {
            values.push(if i < 750 {
                100 + i * 2
            } else {
                5000 + (i - 750) * 5
            });
        }
        roundtrip_batch(&values);
    }

    #[test]
    fn test_blockwise_batch_read_across_block_boundaries() {
        let values = blockwise_values(2053);
        let estimator = BlockwiseLinearEstimator::default();
        let mut encoded = Vec::new();
        estimator.serialize(&values, &mut encoded).unwrap();
        assert_eq!(encoded[0], CodecType::BlockwiseLinear as u8);

        for (start, len) in [
            (0, values.len()),
            (510, 5),
            (511, 3),
            (512, 513),
            (777, 900),
            (1023, 514),
            (1535, 518),
            (2048, 5),
        ] {
            let expected = &values[start..start + len];

            let mut direct = vec![u64::MAX; len];
            blockwise_linear_read_batch(&encoded[1..], start, &mut direct);
            assert_eq!(direct, expected, "direct batch mismatch at {start}");

            let mut auto = vec![u64::MAX; len];
            auto_read_batch(&encoded, start, &mut auto);
            assert_eq!(auto, expected, "auto batch mismatch at {start}");
        }
    }

    /// Regression: zigzag-encoded i64 timestamps mixed with FAST_FIELD_MISSING (u64::MAX).
    /// The linear codec's min_residual clamping to i64 corrupts data when values
    /// span nearly the full u64 range.
    #[test]
    fn test_zigzag_timestamps_with_missing() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        // Simulate issued_at column: most docs have timestamps, some are missing
        let timestamps: Vec<i64> = vec![
            1724630400, // 2024-08-26
            1724716800, // 2024-08-27
            1724803200, // 2024-08-28
            1700000000, // 2023-11-14
            1680000000, // 2023-03-28
            1724630400, // duplicate
        ];

        // Build values array: zigzag-encoded timestamps + some FAST_FIELD_MISSING gaps
        let mut values = Vec::new();
        for (i, &ts) in timestamps.iter().enumerate() {
            values.push(zigzag_encode(ts));
            // Insert a missing value after every 2nd doc
            if i % 2 == 1 {
                values.push(FAST_FIELD_MISSING);
            }
        }

        let result = roundtrip(&values);
        assert_eq!(
            result, values,
            "zigzag timestamps + missing roundtrip failed"
        );
    }

    /// Test each codec individually with zigzag-encoded values + FAST_FIELD_MISSING
    #[test]
    fn test_codecs_individually_with_zigzag_and_missing() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        let values: Vec<u64> = vec![
            zigzag_encode(1724630400), // 3449260800
            zigzag_encode(1700000000), // 3400000000
            FAST_FIELD_MISSING,
            zigzag_encode(1680000000), // 3360000000
            zigzag_encode(1724716800), // 3449433600
            FAST_FIELD_MISSING,
            zigzag_encode(1724630400), // 3449260800
            zigzag_encode(0),          // 0
        ];

        // Test bitpacked directly
        {
            let mut est = BitpackedEstimator::default();
            for &v in &values {
                est.collect(v);
            }
            est.finalize();
            if est.estimate().is_some() {
                let mut buf = Vec::new();
                est.serialize(&values, &mut buf).unwrap();
                for (i, &expected) in values.iter().enumerate() {
                    let got = auto_read(&buf, i);
                    assert_eq!(
                        got, expected,
                        "bitpacked: index {} expected {} got {}",
                        i, expected, got
                    );
                }
            }
        }

        // Test linear directly (needs ≥ 2 values)
        {
            let mut est = LinearEstimator::default();
            for &v in &values {
                est.collect(v);
            }
            est.finalize();
            if est.estimate().is_some() {
                let mut buf = Vec::new();
                est.serialize(&values, &mut buf).unwrap();
                for (i, &expected) in values.iter().enumerate() {
                    let got = auto_read(&buf, i);
                    assert_eq!(
                        got, expected,
                        "linear: index {} expected {} got {}",
                        i, expected, got
                    );
                }
            }
        }

        // Test auto (whichever is selected)
        let result = roundtrip(&values);
        assert_eq!(result, values, "auto codec roundtrip failed");
    }

    /// Regression: value that the user observed corrupted in production
    #[test]
    fn test_specific_issued_at_roundtrip() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        // Reproduce exact scenario: 100 docs, mix of timestamps and missing
        let mut values = Vec::new();
        let base_ts = 1724630400i64; // 2024-08-26 epoch
        for i in 0..100u64 {
            if i % 5 == 0 {
                // Every 5th doc has no issued_at
                values.push(FAST_FIELD_MISSING);
            } else {
                // Varying timestamps
                let ts = base_ts - (i as i64 * 86400); // one day apart
                values.push(zigzag_encode(ts));
            }
        }

        let result = roundtrip(&values);
        for (i, (&expected, &got)) in values.iter().zip(result.iter()).enumerate() {
            assert_eq!(
                got,
                expected,
                "doc {}: expected {} (zigzag of {}), got {}",
                i,
                expected,
                if expected == FAST_FIELD_MISSING {
                    -1 // placeholder
                } else {
                    super::super::zigzag_decode(expected)
                },
                got
            );
        }
    }

    /// Large-scale test: exercise blockwise linear codec with realistic timestamp data.
    /// Tests 10K, 50K, 100K docs to catch codec edge cases.
    #[test]
    fn test_large_scale_timestamp_roundtrip() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        for num_docs in [10_000, 50_000, 100_000] {
            let mut values = Vec::with_capacity(num_docs);
            let base_ts = 1724630400i64;

            for i in 0..num_docs {
                if i % 7 == 0 {
                    values.push(FAST_FIELD_MISSING);
                } else {
                    // Timestamps spanning ~5 years, with some jitter
                    let ts = base_ts - (i as i64 * 3600) + ((i as i64 * 37) % 1000);
                    values.push(zigzag_encode(ts));
                }
            }

            // Check which codec is selected
            let mut buf = Vec::new();
            serialize_auto(&values, &mut buf).unwrap();
            let codec_id = buf[0];
            let codec_name = match CodecType::from_u8(codec_id) {
                Some(CodecType::Constant) => "constant",
                Some(CodecType::Bitpacked) => "bitpacked",
                Some(CodecType::Linear) => "linear",
                Some(CodecType::BlockwiseLinear) => "blockwise_linear",
                None => "unknown",
            };

            // Verify roundtrip
            let mut failures = Vec::new();
            for (i, &expected) in values.iter().enumerate() {
                let got = auto_read(&buf, i);
                if got != expected {
                    failures.push((i, expected, got));
                    if failures.len() >= 5 {
                        break;
                    }
                }
            }

            assert!(
                failures.is_empty(),
                "num_docs={}, codec={}: {} failures. First 5: {:?}",
                num_docs,
                codec_name,
                failures.len(),
                failures
            );
        }
    }

    /// Regression: blockwise linear codec selected for column where most blocks
    /// are efficient (sorted timestamps only) but a few blocks contain
    /// FAST_FIELD_MISSING, causing min_residual clamping corruption.
    #[test]
    fn test_blockwise_linear_with_clustered_missing() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        // 3000 values: first 512 are all FAST_FIELD_MISSING,
        // remaining 2488 are sorted timestamps (efficient linear blocks).
        // This should trigger blockwise linear selection overall,
        // but the first block has a mix that triggers the bug.
        let mut values = Vec::new();

        // Block 0 (indices 0-511): mix of MISSING and timestamps
        // — first 100 are MISSING, rest are timestamps
        for i in 0..512 {
            if i < 100 {
                values.push(FAST_FIELD_MISSING);
            } else {
                let ts = 1724630400i64 + (i as i64 * 100);
                values.push(zigzag_encode(ts));
            }
        }

        // Blocks 1-5 (indices 512-3071): sorted timestamps only
        for i in 512..3072 {
            let ts = 1724630400i64 + (i as i64 * 100);
            values.push(zigzag_encode(ts));
        }

        let result = roundtrip(&values);
        let mut failures = Vec::new();
        for (i, (&expected, &got)) in values.iter().zip(result.iter()).enumerate() {
            if got != expected {
                failures.push((i, expected, got));
            }
        }
        assert!(
            failures.is_empty(),
            "blockwise linear with clustered missing: {} failures. First 5: {:?}",
            failures.len(),
            &failures[..failures.len().min(5)]
        );
    }

    /// Test each codec FORCED with zigzag timestamps + FAST_FIELD_MISSING.
    /// This catches bugs that only manifest when a specific codec is forced.
    #[test]
    fn test_forced_codecs_with_timestamps_and_missing() {
        use super::super::{FAST_FIELD_MISSING, zigzag_encode};

        let mut values = Vec::new();
        let base_ts = 1724630400i64;
        for i in 0..200 {
            if i % 5 == 0 {
                values.push(FAST_FIELD_MISSING);
            } else {
                let ts = base_ts - (i as i64 * 86400);
                values.push(zigzag_encode(ts));
            }
        }

        // Force bitpacked
        {
            let est = BitpackedEstimator::default();
            let mut buf = Vec::new();
            est.serialize(&values, &mut buf).unwrap();
            for (i, &expected) in values.iter().enumerate() {
                let got = bitpacked_read(&buf[1..], i); // skip codec_id byte
                assert_eq!(got, expected, "forced bitpacked: index {} failed", i);
            }
        }

        // Force linear — should error because FAST_FIELD_MISSING + timestamps
        // produce residuals exceeding i64 range
        {
            let est = LinearEstimator::default();
            let mut buf = Vec::new();
            let result = est.serialize(&values, &mut buf);
            assert!(
                result.is_err(),
                "linear codec should reject data with residuals exceeding i64"
            );
        }

        // Force linear with values that DO fit in i64 (no FAST_FIELD_MISSING)
        {
            let safe_values: Vec<u64> = values
                .iter()
                .filter(|&&v| v != FAST_FIELD_MISSING)
                .copied()
                .collect();
            let est = LinearEstimator::default();
            let mut buf = Vec::new();
            est.serialize(&safe_values, &mut buf).unwrap();
            for (i, &expected) in safe_values.iter().enumerate() {
                let got = linear_read(&buf[1..], i);
                assert_eq!(got, expected, "forced linear (safe): index {} failed", i);
            }
        }

        // Blockwise linear estimator should return None for data with FAST_FIELD_MISSING
        {
            let mut large_values = Vec::new();
            for i in 0..2000 {
                if i % 5 == 0 {
                    large_values.push(FAST_FIELD_MISSING);
                } else {
                    let ts = base_ts - (i as i64 * 86400);
                    large_values.push(zigzag_encode(ts));
                }
            }
            let mut est = BlockwiseLinearEstimator::default();
            for &v in &large_values {
                est.collect(v);
            }
            assert!(
                est.estimate().is_none(),
                "blockwise linear should reject data with per-block residuals exceeding i64"
            );
        }
    }
}