formualizer_workbook/backends/

calamine.rs

use crate::load_limits::enforce_sheet_dimension_limits;
use crate::traits::{
    AccessGranularity, AdapterLoadStats, BackendCaps, CalcSettings, CellData, DefinedName,
    DefinedNameDefinition, DefinedNameScope, MergedRange, SheetData, SpreadsheetReader,
};
use formualizer_common::{DateSystem, ExcelError, ExcelErrorKind, LiteralValue};
use parking_lot::RwLock;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::fs::File;
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::Path;
use std::sync::{Arc, Mutex, OnceLock};

use calamine::{Data, DataRef, Range, Reader, Xlsx, XlsxFormulaMetadata, open_workbook_from_rs};
use formualizer_common::RangeAddress;
use formualizer_eval::arrow_store::{IngestBuilder, OverlayValue, map_error_code};
use formualizer_eval::engine::ingest::EngineLoadStream;
use formualizer_eval::engine::{
    CancelToken, DeferredFormulaPackage, Engine as EvalEngine, ExplicitPartitionLegacyMembers,
    FormulaCompressedPreparation, FormulaCompressedSourceBatch, FormulaCompressedSourceReport,
    FormulaIngestBatch, FormulaIngestRecord, FormulaSpoolDiskPolicy, PartitionLegacyMember,
    PartitionLegacyMemberKind, PartitionReconciliation, PartitionedSourceFormulaFamily,
    SourceCoord, SourceFamilyId, SourceFormulaFamily, SourceFormulaOrder, SourceRect,
};
use formualizer_eval::traits::EvaluationContext;
use formualizer_parse::parser::{ASTNode, ReferenceType};
use quick_xml::Reader as XmlReader;
use quick_xml::events::{BytesRef, BytesStart, Event};
use quick_xml::name::QName;
use zip::ZipArchive;

mod compressed_evidence;
mod formula_replay;

use compressed_evidence::{EvidenceRecord, MonotonicFormulaEvidence};
use formula_replay::{
    CalamineDeferredFormulaReplay, FormulaReplaySpool, FormulaSpoolLimits,
    HybridFormulaReplaySpool, SpoolFormulaRecord, replay_spool_per_cell_filtered_with_family,
};

struct SharedFile {
    file: Mutex<File>,
    len: u64,
}

/// Per-cursor readahead window for the shared-file policy.
///
/// ZIP central-directory and part reads are small and mostly forward, so an
/// unbuffered cursor turns each into its own lock/seek/read syscall. The window
/// is keyed by absolute offset, so a seek that lands inside it is still served
/// from memory.
///
/// This widens the window in which a cursor can serve bytes a concurrent writer
/// has since changed. `SharedFile` was never a snapshot, so no guarantee moves:
/// a concurrently mutated file still yields ordinary EOF/I/O/ZIP errors or
/// ordinary stale bytes, never a mapped-memory fault.
const SHARED_FILE_READAHEAD: usize = 64 * 1024;

#[derive(Clone)]
struct SharedFileCursor {
    source: Arc<SharedFile>,
    position: u64,
    /// Bytes covering `[buffer_start, buffer_start + buffer.len())`.
    buffer: Vec<u8>,
    buffer_start: u64,
}

impl SharedFileCursor {
    fn new(source: Arc<SharedFile>) -> Self {
        Self {
            source,
            position: 0,
            buffer: Vec::new(),
            buffer_start: 0,
        }
    }

    /// Serves `buf` from the readahead window when it covers `self.position`.
    fn read_buffered(&mut self, buf: &mut [u8]) -> Option<usize> {
        let offset = usize::try_from(self.position.checked_sub(self.buffer_start)?).ok()?;
        let available = self.buffer.len().checked_sub(offset)?;
        if available == 0 {
            return None;
        }
        let read = buf.len().min(available);
        buf[..read].copy_from_slice(&self.buffer[offset..offset + read]);
        self.position += read as u64;
        Some(read)
    }

    /// One locked seek+read. Fills the readahead window unless `direct` is set,
    /// in which case the caller's larger buffer is filled straight through.
    fn read_physical(&mut self, buf: &mut [u8], limit: usize) -> std::io::Result<usize> {
        let mut file = self
            .source
            .file
            .lock()
            .map_err(|_| std::io::Error::other("shared XLSX file lock poisoned"))?;
        file.seek(SeekFrom::Start(self.position))?;
        if limit >= SHARED_FILE_READAHEAD {
            let read = file.read(&mut buf[..limit])?;
            drop(file);
            self.position += read as u64;
            self.buffer.clear();
            return Ok(read);
        }
        let window = limit.max(
            SHARED_FILE_READAHEAD
                .min(usize::try_from(self.source.len - self.position).unwrap_or(usize::MAX)),
        );
        self.buffer.resize(window, 0);
        let filled = file.read(&mut self.buffer[..window])?;
        drop(file);
        self.buffer.truncate(filled);
        self.buffer_start = self.position;
        Ok(self.read_buffered(buf).unwrap_or(0))
    }
}

#[cfg(any(unix, windows))]
struct MappedXlsxSource {
    // The handle pins the opened file rather than its path. It does not prevent
    // another process from truncating the file on Unix; see CalamineAdapter.
    _file: File,
    map: memmap2::Mmap,
}

#[derive(Clone)]
enum SharedXlsxReader {
    Bytes {
        bytes: Arc<[u8]>,
        position: u64,
    },
    File(SharedFileCursor),
    #[cfg(any(unix, windows))]
    Mapped {
        source: Arc<MappedXlsxSource>,
        position: u64,
    },
}

/// Selects the retained backing source used for an XLSX filesystem path.
///
/// This policy applies only to path-based Calamine XLSX loads. Byte and reader
/// inputs always retain one shared immutable byte allocation.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum XlsxPathSource {
    /// Retain one opened file and serialize physical seeks and reads while
    /// giving each XLSX consumer an independent logical cursor.
    ///
    /// This mutation-safe option is the default. Renaming or replacing the
    /// pathname is safe because the opened handle is retained.
    #[default]
    SharedFile,
    /// Retain the opened file and map it read-only.
    ///
    /// Mapping is attempted explicitly and errors are returned without falling
    /// back to [`SharedFile`](Self::SharedFile). The caller guarantees that the
    /// opened underlying file/inode is not destructively modified or truncated
    /// for the adapter's lifetime. Violating this contract can terminate the
    /// process; in particular, Unix may deliver `SIGBUS` when a mapped page past
    /// a new end-of-file is accessed. Renaming or replacing the pathname is safe
    /// because the original opened handle remains retained.
    DirectMmap,
}

fn seek_position(position: u64, len: u64, from: SeekFrom) -> std::io::Result<u64> {
    let next = match from {
        SeekFrom::Start(next) => Some(next),
        SeekFrom::Current(offset) => position.checked_add_signed(offset),
        SeekFrom::End(offset) => len.checked_add_signed(offset),
    };
    next.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "invalid seek"))
}

fn read_slice(source: &[u8], position: &mut u64, buf: &mut [u8]) -> std::io::Result<usize> {
    let start = usize::try_from(*position).unwrap_or(usize::MAX);
    if start >= source.len() {
        return Ok(0);
    }
    let read = buf.len().min(source.len() - start);
    buf[..read].copy_from_slice(&source[start..start + read]);
    *position = position
        .checked_add(read as u64)
        .ok_or_else(|| std::io::Error::other("reader position overflow"))?;
    Ok(read)
}

impl SharedXlsxReader {
    fn from_bytes(data: Vec<u8>) -> Self {
        Self::Bytes {
            bytes: Arc::from(data),
            position: 0,
        }
    }

    fn from_file(file: File, policy: XlsxPathSource) -> Result<Self, std::io::Error> {
        let len = file.metadata()?.len();
        match policy {
            XlsxPathSource::SharedFile => {
                Ok(Self::File(SharedFileCursor::new(Arc::new(SharedFile {
                    file: Mutex::new(file),
                    len,
                }))))
            }
            XlsxPathSource::DirectMmap => {
                #[cfg(any(unix, windows))]
                {
                    if len == 0 {
                        return Err(std::io::Error::new(
                            std::io::ErrorKind::InvalidInput,
                            "cannot memory-map an empty XLSX file",
                        ));
                    }
                    // SAFETY: the map is read-only and its source handle remains
                    // alive. The caller accepts the mutation/truncation contract
                    // documented on `XlsxPathSource::DirectMmap`.
                    let map = unsafe { memmap2::MmapOptions::new().map(&file) }?;
                    Ok(Self::Mapped {
                        source: Arc::new(MappedXlsxSource { _file: file, map }),
                        position: 0,
                    })
                }
                #[cfg(not(any(unix, windows)))]
                {
                    let _ = (file, len);
                    Err(std::io::Error::new(
                        std::io::ErrorKind::Unsupported,
                        "DirectMmap XLSX path loading is unsupported on this target",
                    ))
                }
            }
        }
    }

    fn reader(&self) -> Self {
        match self {
            Self::Bytes { bytes, .. } => Self::Bytes {
                bytes: Arc::clone(bytes),
                position: 0,
            },
            Self::File(cursor) => Self::File(SharedFileCursor::new(Arc::clone(&cursor.source))),
            #[cfg(any(unix, windows))]
            Self::Mapped { source, .. } => Self::Mapped {
                source: Arc::clone(source),
                position: 0,
            },
        }
    }

    #[cfg(test)]
    fn bytes_backing_ptr(&self) -> Option<*const u8> {
        match self {
            Self::Bytes { bytes, .. } => Some(bytes.as_ptr()),
            _ => None,
        }
    }

    #[cfg(test)]
    fn strong_count(&self) -> usize {
        match self {
            Self::Bytes { bytes, .. } => Arc::strong_count(bytes),
            Self::File(cursor) => Arc::strong_count(&cursor.source),
            #[cfg(any(unix, windows))]
            Self::Mapped { source, .. } => Arc::strong_count(source),
        }
    }
}

impl Read for SharedXlsxReader {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        match self {
            Self::Bytes { bytes, position } => read_slice(bytes, position, buf),
            Self::File(cursor) => {
                if buf.is_empty() || cursor.position >= cursor.source.len {
                    return Ok(0);
                }
                if let Some(read) = cursor.read_buffered(buf) {
                    return Ok(read);
                }
                let remaining = cursor.source.len - cursor.position;
                let limit = buf
                    .len()
                    .min(usize::try_from(remaining).unwrap_or(usize::MAX));
                cursor.read_physical(buf, limit)
            }
            #[cfg(any(unix, windows))]
            Self::Mapped { source, position } => read_slice(&source.map, position, buf),
        }
    }
}

impl Seek for SharedXlsxReader {
    fn seek(&mut self, from: SeekFrom) -> std::io::Result<u64> {
        let position = match self {
            Self::Bytes { bytes, position } => {
                *position = seek_position(*position, bytes.len() as u64, from)?;
                *position
            }
            Self::File(cursor) => {
                cursor.position = seek_position(cursor.position, cursor.source.len, from)?;
                cursor.position
            }
            #[cfg(any(unix, windows))]
            Self::Mapped { source, position } => {
                *position = seek_position(*position, source.map.len() as u64, from)?;
                *position
            }
        };
        Ok(position)
    }
}

/// A bounded, cooperative cancellation layer for Calamine's ZIP reads.
///
/// `ErrorKind::Interrupted` is deliberately not used: standard library helpers
/// such as `read_to_end` retry it automatically. Limiting each forwarded read
/// also bounds the time before a concurrently-signalled token is observed when
/// the backing source is an in-memory byte slice.
struct CancellableReader {
    inner: SharedXlsxReader,
    cancel: Option<CancelToken>,
}

impl CancellableReader {
    const MAX_READ: usize = 64 * 1024;

    fn new(inner: SharedXlsxReader, cancel: Option<CancelToken>) -> Self {
        Self { inner, cancel }
    }

    fn checkpoint(&self) -> std::io::Result<()> {
        if self.cancel.as_ref().is_some_and(CancelToken::is_cancelled) {
            return Err(std::io::Error::other("calamine load cancelled"));
        }
        Ok(())
    }
}

impl Read for CancellableReader {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        self.checkpoint()?;
        let limit = buf.len().min(Self::MAX_READ);
        self.inner.read(&mut buf[..limit])
    }
}

impl Seek for CancellableReader {
    fn seek(&mut self, from: SeekFrom) -> std::io::Result<u64> {
        self.checkpoint()?;
        self.inner.seek(from)
    }
}

struct CalamineWorkbook(Xlsx<CancellableReader>);

impl CalamineWorkbook {
    fn worksheet_range(&mut self, sheet: &str) -> Result<Range<Data>, calamine::Error> {
        self.0.worksheet_range(sheet).map_err(Into::into)
    }

    fn worksheet_formula(&mut self, sheet: &str) -> Result<Range<String>, calamine::Error> {
        self.0.worksheet_formula(sheet).map_err(Into::into)
    }
}

struct DebugTimer {
    #[cfg(not(target_arch = "wasm32"))]
    started: std::time::Instant,
}

struct DenseState {
    aib: IngestBuilder,
    row_vals: Vec<LiteralValue>,
    current_row0: usize,
    rows_appended: usize,
    row_started: bool,
}

#[derive(Clone, Copy)]
struct WorkbookSpoolUsage {
    bytes: u64,
    files: u32,
}

struct StreamWorksheetOptions {
    chunk_rows: usize,
    debug: bool,
    workbook_spool_usage: WorkbookSpoolUsage,
    shadow_relocation_comparator: Option<ShadowRelocationComparator>,
}

struct FormulaStaging {
    parse_cache: rustc_hash::FxHashMap<String, Option<formualizer_eval::engine::AstNodeId>>,
    formulas: Vec<FormulaIngestRecord>,
    observed: usize,
    handed_to_engine: usize,
}

impl FormulaStaging {
    fn new() -> Self {
        let mut parse_cache = rustc_hash::FxHashMap::default();
        parse_cache.reserve(4096);
        Self {
            parse_cache,
            formulas: Vec::new(),
            observed: 0,
            handed_to_engine: 0,
        }
    }
}

struct StreamedSheet {
    arrow_sheet: formualizer_eval::arrow_store::ArrowSheet,
    dimensions: (usize, usize),
    max_col_seen: usize,
    used_sparse_fallback: bool,
    value_cells_observed: usize,
    values_handed_to_engine: usize,
    formulas_observed: usize,
    formulas_handed_to_engine: usize,
    formulas: Vec<FormulaIngestRecord>,
    formula_source_report: FormulaCompressedSourceReport,
    compressed_families: Vec<SourceFormulaFamily>,
    partitioned_families: Vec<PartitionedSourceFormulaFamily>,
    direct_preparation: Option<FormulaCompressedPreparation>,
    deferred_package: Option<DeferredFormulaPackage>,
    shared_formula_tags: usize,
    formula_spool_bytes: u64,
    formula_spool_spilled: bool,
    stream_millis: u128,
}

#[inline]
fn data_ref_to_literal(value: &DataRef<'_>, date_system: DateSystem) -> Option<LiteralValue> {
    match value {
        DataRef::Empty => None,
        DataRef::String(s) if s.is_empty() => None,
        DataRef::SharedString("") => None,
        DataRef::String(s) => Some(LiteralValue::Text(s.clone())),
        DataRef::SharedString(s) => Some(LiteralValue::Text((*s).to_string())),
        DataRef::Float(f) => Some(LiteralValue::Number(*f)),
        DataRef::Int(i) => Some(LiteralValue::Number(*i as f64)),
        DataRef::Bool(b) => Some(LiteralValue::Boolean(*b)),
        DataRef::Error(e) => Some(LiteralValue::Error(ExcelError::new(
            match CalamineAdapter::calamine_error_code(e) {
                1 => ExcelErrorKind::Null,
                2 => ExcelErrorKind::Ref,
                3 => ExcelErrorKind::Name,
                4 => ExcelErrorKind::Value,
                5 => ExcelErrorKind::Div,
                6 => ExcelErrorKind::Na,
                7 => ExcelErrorKind::Num,
                _ => ExcelErrorKind::Error,
            },
        ))),
        DataRef::DateTime(dt) => Some(
            LiteralValue::try_from_serial_number_for(date_system, dt.as_f64())
                .unwrap_or_else(LiteralValue::Error),
        ),
        DataRef::DateTimeIso(s) => Some(LiteralValue::Text(s.clone())),
        DataRef::DurationIso(s) => Some(LiteralValue::Text(s.clone())),
    }
}

#[inline]
fn data_ref_to_overlay(value: &DataRef<'_>) -> Option<OverlayValue> {
    match value {
        DataRef::Empty => None,
        DataRef::String(s) if s.is_empty() => None,
        DataRef::SharedString("") => None,
        DataRef::String(s) => Some(OverlayValue::Text(Arc::from(s.as_str()))),
        DataRef::SharedString(s) => Some(OverlayValue::Text(Arc::from(*s))),
        DataRef::Float(f) => Some(OverlayValue::Number(*f)),
        DataRef::Int(i) => Some(OverlayValue::Number(*i as f64)),
        DataRef::Bool(b) => Some(OverlayValue::Boolean(*b)),
        DataRef::Error(e) => Some(OverlayValue::Error(CalamineAdapter::calamine_error_code(e))),
        DataRef::DateTime(dt) => Some(OverlayValue::Number(dt.as_f64())),
        DataRef::DateTimeIso(s) => Some(OverlayValue::Text(Arc::from(s.as_str()))),
        DataRef::DurationIso(s) => Some(OverlayValue::Text(Arc::from(s.as_str()))),
    }
}

fn data_ref_format(value: &DataRef<'_>) -> Option<formualizer_eval::format::FormatId> {
    match value {
        DataRef::DateTime(dt) if dt.is_duration() => {
            Some(formualizer_eval::format::FormatId::DURATION)
        }
        DataRef::DateTime(dt) if (0.0..1.0).contains(&dt.as_f64()) => {
            Some(formualizer_eval::format::FormatId::TIME)
        }
        DataRef::DateTime(dt) if dt.as_f64().fract().abs() > f64::EPSILON => {
            Some(formualizer_eval::format::FormatId::DATETIME)
        }
        DataRef::DateTime(_) => Some(formualizer_eval::format::FormatId::DATE),
        _ => None,
    }
}

impl DebugTimer {
    fn start() -> Self {
        Self {
            #[cfg(not(target_arch = "wasm32"))]
            started: std::time::Instant::now(),
        }
    }

    fn elapsed_millis(&self) -> u128 {
        #[cfg(not(target_arch = "wasm32"))]
        {
            self.started.elapsed().as_millis()
        }
        #[cfg(target_arch = "wasm32")]
        {
            0
        }
    }
}

type ShadowRelocationComparator = Arc<dyn Fn(&ASTNode, &ASTNode) -> bool + Send + Sync>;

/// Read-only XLSX adapter backed by one shared source acquisition.
///
/// [`SpreadsheetReader::open_path`] retains one opened file and uses serialized
/// file I/O with independent logical cursors. [`Self::open_path_with_source`]
/// can explicitly request a read-only memory map instead. Byte and reader inputs
/// use shared immutable owned bytes. See [`XlsxPathSource::DirectMmap`] for the
/// mapped-file safety contract.
pub struct CalamineAdapter {
    workbook: RwLock<CalamineWorkbook>,
    source: SharedXlsxReader,
    /// Present only for adapters opened through [`Self::open_bytes_cancellable`].
    cancel: Option<CancelToken>,
    loaded_sheets: HashSet<String>,
    cached_names: Option<Vec<String>>,
    /// Calamine's already-parsed `(name, formula)` pairs, captured at open time.
    ///
    /// Held outside `workbook` on purpose: [`Self::lazy_defined_names`] must not
    /// acquire the `workbook` lock, because `parking_lot::RwLock` is not
    /// reentrant and any caller holding the write lock would self-deadlock.
    calamine_defined_names: Vec<(String, String)>,
    defined_names: OnceLock<Vec<DefinedName>>,
    external_link_targets: OnceLock<BTreeMap<u32, String>>,
    calc_settings: OnceLock<Option<CalcSettings>>,
    load_stats: AdapterLoadStats,
    shadow_relocation_comparator: Option<ShadowRelocationComparator>,
    #[cfg(test)]
    lazy_scan_counts: LazyScanCounts,
    #[cfg(test)]
    stream_row_checkpoint_hook: Option<Arc<dyn Fn() + Send + Sync>>,
}

#[cfg(test)]
#[derive(Default)]
struct LazyScanCounts {
    external_links: std::sync::atomic::AtomicUsize,
    calc_settings: std::sync::atomic::AtomicUsize,
    defined_names: std::sync::atomic::AtomicUsize,
}

impl CalamineAdapter {
    const EXCEL_MAX_ROWS: u32 = 1_048_576;

    /// Opens one XLSX path using the requested retained backing source.
    ///
    /// [`XlsxPathSource::SharedFile`] matches [`SpreadsheetReader::open_path`].
    /// [`XlsxPathSource::DirectMmap`] performs an actual read-only mapping on
    /// Unix and Windows and returns the mapping error without fallback. Other
    /// targets return a `calamine::Error::Io` whose I/O kind is `Unsupported`.
    pub fn open_path_with_source<P: AsRef<Path>>(
        path: P,
        source: XlsxPathSource,
    ) -> Result<Self, calamine::Error> {
        let file = File::open(path).map_err(calamine::Error::Io)?;
        let source = SharedXlsxReader::from_file(file, source).map_err(calamine::Error::Io)?;
        Self::from_shared_source(source, None)
    }

    /// Opens XLSX bytes with cooperative cancellation for parsing and later
    /// [`EngineLoadStream::stream_into_engine`] work.
    ///
    /// Cancellation is reported as a non-`Interrupted` I/O error wrapped in
    /// [`calamine::Error`], so callers can inspect the supplied token and map
    /// it to their own typed cancellation result.
    pub fn open_bytes_cancellable(
        bytes: Vec<u8>,
        cancel: CancelToken,
    ) -> Result<Self, calamine::Error> {
        Self::from_shared_source(SharedXlsxReader::from_bytes(bytes), Some(cancel))
    }

    #[doc(hidden)]
    pub fn set_shadow_relocation_comparator_for_test(
        &mut self,
        comparator: impl Fn(&ASTNode, &ASTNode) -> bool + Send + Sync + 'static,
    ) {
        self.shadow_relocation_comparator = Some(Arc::new(comparator));
    }

    fn shadow_relocation_matches(
        comparator: &ShadowRelocationComparator,
        family: &SourceFormulaFamily,
        coord0: SourceCoord,
        expanded_formula: &str,
    ) -> bool {
        let expanded_formula = format!("={}", expanded_formula.trim_start_matches('='));
        let anchor_formula = format!("={}", family.anchor_text.trim_start_matches('='));
        let Ok(expanded) = formualizer_parse::parser::parse(&expanded_formula) else {
            return false;
        };
        let Ok(anchor) = formualizer_parse::parser::parse(&anchor_formula) else {
            return false;
        };
        let Ok(relocated) =
            formualizer_eval::formula_plane::structural::relocate_ast_for_template_placement(
                &anchor,
                i64::from(coord0.row) - i64::from(family.anchor_coord0.row),
                i64::from(coord0.col) - i64::from(family.anchor_coord0.col),
            )
        else {
            return false;
        };
        comparator(&expanded, &relocated)
    }
    const EXCEL_MAX_COLS: u32 = 16_384;

    fn stage_formula<C: EvaluationContext>(
        engine: &mut EvalEngine<C>,
        sheet: &str,
        position: (u32, u32),
        formula: &str,
        debug: bool,
        staging: &mut FormulaStaging,
    ) -> Result<(), calamine::Error> {
        let excel_row = position.0 + 1;
        let excel_col = position.1 + 1;
        let normalized = if formula.starts_with('=') {
            formula.to_string()
        } else {
            format!("={formula}")
        };
        if debug && staging.observed < 16 {
            eprintln!("[fz][load] formula observed at R{excel_row}C{excel_col}");
        }
        if engine.config.defer_graph_building {
            engine.stage_formula_text(sheet, excel_row, excel_col, normalized);
            staging.handed_to_engine += 1;
        } else {
            let ast_id = if let Some(cached) = staging.parse_cache.get(&normalized) {
                *cached
            } else {
                let parsed = match formualizer_parse::parser::parse(&normalized) {
                    Ok(parsed) => Some(parsed),
                    Err(error) => engine
                        .handle_formula_parse_error(
                            sheet,
                            excel_row,
                            excel_col,
                            &normalized,
                            error.to_string(),
                        )
                        .map_err(|error| {
                            calamine::Error::Io(std::io::Error::other(error.to_string()))
                        })?,
                };
                let ast_id = parsed.as_ref().map(|ast| engine.intern_formula_ast(ast));
                staging.parse_cache.insert(normalized.clone(), ast_id);
                ast_id
            };
            if let Some(ast_id) = ast_id {
                staging.formulas.push(FormulaIngestRecord::new(
                    excel_row,
                    excel_col,
                    ast_id,
                    Some(Arc::<str>::from(normalized)),
                ));
                staging.handed_to_engine += 1;
            }
        }
        staging.observed += 1;
        Ok(())
    }

    fn stream_worksheet<RS, C>(
        workbook: &mut Xlsx<RS>,
        sheet: &str,
        engine: &mut EvalEngine<C>,
        sheet_instance: u32,
        options: StreamWorksheetOptions,
        cancel: Option<&CancelToken>,
        #[cfg(test)] row_checkpoint_hook: Option<&(dyn Fn() + Send + Sync)>,
    ) -> Result<StreamedSheet, calamine::Error>
    where
        RS: Read + Seek,
        C: EvaluationContext,
    {
        Self::cancellation_checkpoint(cancel)?;
        let timer = DebugTimer::start();
        let StreamWorksheetOptions {
            chunk_rows,
            debug,
            workbook_spool_usage,
            shadow_relocation_comparator,
        } = options;
        let mut reader = workbook
            .worksheet_cells_reader(sheet)
            .map_err(calamine::Error::Xlsx)?;
        let declared = reader.dimensions();
        let mut dims_rows = (declared.end.0 as usize + 1).max(1);
        let mut dims_cols = (declared.end.1 as usize + 1).max(1);
        enforce_sheet_dimension_limits(
            "calamine",
            sheet,
            dims_rows as u32,
            dims_cols as u32,
            engine.workbook_load_limits(),
        )
        .map_err(|error| calamine::Error::Io(std::io::Error::other(error.to_string())))?;

        let force_sparse_from_start = (dims_rows as u64).saturating_mul(dims_cols as u64)
            > engine.workbook_load_limits().max_sheet_logical_cells;
        let mut dense = (!force_sparse_from_start).then(|| DenseState {
            aib: IngestBuilder::new(sheet, dims_cols, chunk_rows, engine.config.date_system),
            row_vals: vec![LiteralValue::Empty; dims_cols],
            current_row0: 0,
            rows_appended: 0,
            row_started: false,
        });
        let mut sparse = force_sparse_from_start.then(|| {
            formualizer_eval::arrow_store::ArrowSheet::new_sparse_with_date_system(
                sheet,
                dims_cols,
                dims_rows,
                chunk_rows,
                engine.config.date_system,
            )
        });
        let mut used_sparse_fallback = force_sparse_from_start;
        let mut max_row_seen = 0usize;
        let mut max_col_seen = 0usize;
        let mut value_cells_observed = 0usize;
        let mut values_handed_to_engine = 0usize;
        let mut formula_staging = FormulaStaging::new();
        let mut formula_count = 0usize;
        let mut deferred_source_coordinates = engine.config.defer_graph_building.then(Vec::new);
        let mut formula_evidence = MonotonicFormulaEvidence::new();
        let spool_limits = engine.workbook_load_limits();
        let workbook_bytes_remaining = spool_limits
            .max_formula_spool_bytes_per_workbook
            .saturating_sub(workbook_spool_usage.bytes);
        let spill_files_remaining = spool_limits
            .max_formula_spool_files_per_workbook
            .saturating_sub(workbook_spool_usage.files);
        let mut formula_spool = HybridFormulaReplaySpool::new(FormulaSpoolLimits {
            sheet_bytes: spool_limits.max_formula_spool_bytes_per_sheet,
            workbook_bytes_remaining,
            workbook_bytes_used: workbook_spool_usage.bytes,
            memory_prefix_bytes: spool_limits.formula_spool_memory_prefix_bytes,
            memory_only_bytes: spool_limits.max_formula_spool_memory_bytes,
            allow_disk: spool_limits.formula_spool_disk_policy
                == FormulaSpoolDiskPolicy::NativeSpill,
            spill_files_remaining,
            spill_files_limit: spool_limits.max_formula_spool_files_per_workbook,
        });
        let mut last_formula_coord = None;
        let mut shared_formula_tags = 0usize;
        let mut last_cancel_row = None;

        while let Some(record) = reader
            .next_cell_with_formula_metadata()
            .map_err(calamine::Error::Xlsx)?
        {
            let (row0, col0) = record.pos;
            let row = row0 as usize;
            let col = col0 as usize;
            if last_cancel_row != Some(row) {
                #[cfg(test)]
                if let Some(hook) = row_checkpoint_hook {
                    hook();
                }
                Self::cancellation_checkpoint(cancel)?;
                last_cancel_row = Some(row);
            }
            if row >= dims_rows || col >= dims_cols {
                dims_rows = dims_rows.max(row + 1);
                dims_cols = dims_cols.max(col + 1);
                enforce_sheet_dimension_limits(
                    "calamine",
                    sheet,
                    dims_rows as u32,
                    dims_cols as u32,
                    engine.workbook_load_limits(),
                )
                .map_err(|error| calamine::Error::Io(std::io::Error::other(error.to_string())))?;
            }
            max_row_seen = max_row_seen.max(row);
            max_col_seen = max_col_seen.max(col);

            let has_formula = record.formula.is_some();
            if let Some(metadata) = record.formula {
                if u64::try_from(value_cells_observed)
                    .unwrap_or(u64::MAX)
                    .saturating_add(u64::try_from(formula_count).unwrap_or(u64::MAX))
                    .saturating_add(1)
                    > engine.workbook_load_limits().max_sheet_logical_cells
                {
                    return Err(calamine::Error::Io(std::io::Error::other(format!(
                        "Workbook load budget exceeded in calamine for sheet {sheet}: observed populated cell count exceeds configured logical-cell budget of {}",
                        engine.workbook_load_limits().max_sheet_logical_cells
                    ))));
                }
                let coord0 = SourceCoord {
                    row: row0,
                    col: col0,
                };
                let source_sequence = formula_count as u64;
                match metadata {
                    XlsxFormulaMetadata::Normal { formula } => {
                        if let Some(coordinates) = deferred_source_coordinates.as_mut() {
                            coordinates.push((coord0, None));
                        }
                        formula_evidence.observe_ordered(
                            coord0,
                            SourceFormulaOrder::new(source_sequence),
                            EvidenceRecord::Ordinary,
                        );
                        formula_spool.append(SpoolFormulaRecord::Ordinary {
                            sequence: source_sequence,
                            coord0,
                            text: &formula,
                        })
                    }
                    XlsxFormulaMetadata::Shared {
                        shared_index,
                        range,
                        formula,
                    } => {
                        shared_formula_tags += 1;
                        let declared_range = range.map(|range| SourceRect {
                            start: SourceCoord {
                                row: range.start.0,
                                col: range.start.1,
                            },
                            end: SourceCoord {
                                row: range.end.0,
                                col: range.end.1,
                            },
                        });
                        let family = SourceFamilyId {
                            sheet_instance,
                            source_index: shared_index,
                        };
                        if let Some(coordinates) = deferred_source_coordinates.as_mut() {
                            coordinates.push((coord0, Some(family)));
                        }
                        formula_evidence.observe_ordered(
                            coord0,
                            SourceFormulaOrder::new(source_sequence),
                            EvidenceRecord::Anchor {
                                family,
                                range: declared_range,
                                text: &formula,
                            },
                        );
                        formula_spool.append(SpoolFormulaRecord::SharedAnchor {
                            sequence: source_sequence,
                            coord0,
                            shared_index,
                            declared_range,
                            text: &formula,
                        })
                    }
                    XlsxFormulaMetadata::SharedDerived { shared_index } => {
                        shared_formula_tags += 1;
                        let family = SourceFamilyId {
                            sheet_instance,
                            source_index: shared_index,
                        };
                        if let Some(coordinates) = deferred_source_coordinates.as_mut() {
                            coordinates.push((coord0, Some(family)));
                        }
                        formula_evidence.observe_ordered(
                            coord0,
                            SourceFormulaOrder::new(source_sequence),
                            EvidenceRecord::Descendant { family },
                        );
                        formula_spool.append(SpoolFormulaRecord::SharedDescendant {
                            sequence: source_sequence,
                            coord0,
                            shared_index,
                        })
                    }
                    _ => {
                        if let Some(coordinates) = deferred_source_coordinates.as_mut() {
                            coordinates.push((coord0, None));
                        }
                        formula_evidence.observe_ordered(
                            coord0,
                            SourceFormulaOrder::new(source_sequence),
                            EvidenceRecord::Unsupported,
                        );
                        formula_spool.append(SpoolFormulaRecord::Unsupported {
                            sequence: source_sequence,
                            coord0,
                        })
                    }
                }
                .map_err(|error| calamine::Error::Io(std::io::Error::other(error.to_string())))?;
                last_formula_coord = Some((row, col));
                if let Some(state) = dense.as_mut()
                    && state.row_started
                    && state.current_row0 == row
                    && col < state.row_vals.len()
                {
                    state.row_vals[col] = LiteralValue::Empty;
                }
                formula_count += 1;
            }

            // Preserve existing KeepCachedValue behavior: a formula's cached
            // value is not handed to the value plane.
            if has_formula {
                continue;
            }
            let Some(literal) = data_ref_to_literal(&record.value, engine.config.date_system)
            else {
                continue;
            };
            value_cells_observed += 1;
            if u64::try_from(value_cells_observed)
                .unwrap_or(u64::MAX)
                .saturating_add(u64::try_from(formula_count).unwrap_or(u64::MAX))
                > engine.workbook_load_limits().max_sheet_logical_cells
            {
                return Err(calamine::Error::Io(std::io::Error::other(format!(
                    "Workbook load budget exceeded in calamine for sheet {sheet}: observed populated cell count exceeds configured logical-cell budget of {}",
                    engine.workbook_load_limits().max_sheet_logical_cells
                ))));
            }
            if last_formula_coord == Some((row, col)) {
                continue;
            }

            if let Some(arrow_sheet) = sparse.as_mut() {
                if let Some(value) = data_ref_to_overlay(&record.value) {
                    arrow_sheet.set_sparse_overlay_value(row, col, value);
                    arrow_sheet.set_sparse_overlay_format(row, col, data_ref_format(&record.value));
                    values_handed_to_engine += 1;
                }
                continue;
            }

            let state = dense.as_mut().expect("dense or sparse ingest mode");
            let non_monotonic = state.row_started && row < state.current_row0;
            let col_overflow = col >= state.row_vals.len();
            let gap_rows = if state.row_started {
                row.saturating_sub(state.current_row0)
            } else {
                row
            };
            let large_gap = gap_rows > 128;
            let would_exceed_dense_budget =
                u64::try_from(state.rows_appended.saturating_mul(state.row_vals.len()))
                    .unwrap_or(u64::MAX)
                    > engine.workbook_load_limits().max_sheet_logical_cells;
            if non_monotonic || col_overflow || large_gap || would_exceed_dense_budget {
                let mut state = dense.take().expect("dense state present");
                if state.row_started && state.current_row0 == state.rows_appended {
                    state.aib.append_row(&state.row_vals).map_err(|error| {
                        calamine::Error::Io(std::io::Error::other(error.to_string()))
                    })?;
                    state.rows_appended += 1;
                }
                let mut arrow_sheet = state.aib.finish();
                arrow_sheet.ensure_row_capacity(dims_rows.max(row + 1));
                if col >= arrow_sheet.columns.len() {
                    arrow_sheet.insert_columns(
                        arrow_sheet.columns.len(),
                        col + 1 - arrow_sheet.columns.len(),
                    );
                }
                if let Some(value) = data_ref_to_overlay(&record.value) {
                    arrow_sheet.set_sparse_overlay_value(row, col, value);
                    arrow_sheet.set_sparse_overlay_format(row, col, data_ref_format(&record.value));
                    values_handed_to_engine += 1;
                }
                sparse = Some(arrow_sheet);
                used_sparse_fallback = true;
                continue;
            }

            if !state.row_started {
                while state.rows_appended < row {
                    state
                        .aib
                        .append_row(&vec![LiteralValue::Empty; state.row_vals.len()])
                        .map_err(|error| {
                            calamine::Error::Io(std::io::Error::other(error.to_string()))
                        })?;
                    state.rows_appended += 1;
                }
                state.current_row0 = row;
                state.row_started = true;
            } else if row > state.current_row0 {
                state.aib.append_row(&state.row_vals).map_err(|error| {
                    calamine::Error::Io(std::io::Error::other(error.to_string()))
                })?;
                state.rows_appended += 1;
                state.row_vals.fill(LiteralValue::Empty);
                while state.rows_appended < row {
                    state
                        .aib
                        .append_row(&vec![LiteralValue::Empty; state.row_vals.len()])
                        .map_err(|error| {
                            calamine::Error::Io(std::io::Error::other(error.to_string()))
                        })?;
                    state.rows_appended += 1;
                }
                state.current_row0 = row;
            }
            state.row_vals[col] = literal;
            values_handed_to_engine += 1;
        }

        // Replay and validate the complete sheet-local source stream before any
        // formula staging, parsing, or graph mutation. The Arrow result is also
        // still local and is installed by the caller only after this succeeds.
        let compressed_evidence = formula_evidence.finish();
        let mut formula_source_report = compressed_evidence.report;
        let mut compressed_families = compressed_evidence.families;
        let partitioned_families = compressed_evidence
            .fragmented
            .into_iter()
            .map(|proposal| {
                let mut legacy_members: Vec<_> = proposal
                    .fallback_members
                    .into_iter()
                    .map(|coord| PartitionLegacyMember {
                        coord,
                        kind: PartitionLegacyMemberKind::SharedFamilyMember,
                    })
                    .collect();
                let mut ordinary_exceptions = 0u64;
                let mut holes = 0u64;
                for exclusion in proposal.exclusions {
                    match exclusion {
                        compressed_evidence::SourceExclusion::Hole(_) => {
                            holes = holes.saturating_add(1);
                        }
                        compressed_evidence::SourceExclusion::OrdinaryFormula(coord) => {
                            ordinary_exceptions = ordinary_exceptions.saturating_add(1);
                            legacy_members.push(PartitionLegacyMember {
                                coord,
                                kind: PartitionLegacyMemberKind::OrdinaryException,
                            });
                        }
                    }
                }
                let legacy_members = ExplicitPartitionLegacyMembers::try_new(legacy_members)
                    .map_err(|reason| calamine::Error::Io(std::io::Error::other(reason)))?;
                Ok(PartitionedSourceFormulaFamily {
                    source_id: proposal.source_id,
                    source_order: proposal.source_order,
                    template_origin0: proposal.anchor_coord0,
                    template_text: proposal.anchor_text,
                    declared: proposal.declared,
                    surviving_member_count: proposal.member_count,
                    fragments: proposal.fragments,
                    legacy_members,
                    reconciliation: PartitionReconciliation {
                        shared_members: proposal.member_count,
                        ordinary_exceptions,
                        holes,
                    },
                })
            })
            .collect::<Result<Vec<_>, calamine::Error>>()?;
        let deferred_source_coordinates = deferred_source_coordinates
            .unwrap_or_default()
            .into_iter()
            .map(|(coord, _)| coord)
            .collect::<Vec<_>>();
        let formula_spool_bytes = if formula_count == 0 {
            0
        } else {
            formula_spool.encoded_bytes()
        };
        let formula_spool_spilled = formula_spool.spilled();
        formula_source_report.source_formula_records_spooled = formula_count as u64;
        formula_source_report.source_spool_encoded_bytes = formula_spool_bytes;
        formula_source_report.source_spool_peak_memory_bytes = formula_spool.peak_memory_bytes();
        formula_source_report.source_spool_spilled_bytes = if formula_spool_spilled {
            formula_spool_bytes
        } else {
            0
        };
        formula_source_report.source_spool_spill_files = u64::from(formula_spool_spilled);
        let _formula_spool_storage = formula_spool.storage_kind();
        debug_assert!(formula_count == 0 || formula_spool_bytes >= 5);
        let mut formula_spool = Some(formula_spool);
        let direct_preparation = if engine.config.formula_plane_mode
            == formualizer_eval::engine::FormulaPlaneMode::AuthoritativeExperimental
            && !engine.config.defer_graph_building
        {
            Self::cancellation_checkpoint(cancel)?;
            let replay: Box<dyn formualizer_eval::engine::DeferredFormulaReplay> =
                Box::new(CalamineDeferredFormulaReplay::new(
                    formula_spool.take().expect("eager formula spool available"),
                    sheet.to_string(),
                    sheet_instance,
                ));
            Some(
                engine
                    .source_formula_ingress()
                    .prepare_eager_proposals(
                        sheet,
                        &compressed_families,
                        &partitioned_families,
                        formula_count as u64,
                        replay,
                    )
                    .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?,
            )
        } else {
            None
        };
        Self::cancellation_checkpoint(cancel)?;
        if direct_preparation.is_none() && !engine.config.defer_graph_building {
            formula_source_report.source_spool_replays = 1;
            let compare_shadow = engine.config.formula_plane_mode
                == formualizer_eval::engine::FormulaPlaneMode::Shadow;
            let mut relocation_mismatches = BTreeSet::new();
            replay_spool_per_cell_filtered_with_family(
                formula_spool
                    .as_mut()
                    .expect("replay formula spool available"),
                sheet,
                |_| false,
                |coord0, formula, shared_index| {
                    Self::cancellation_checkpoint(cancel)?;
                    if compare_shadow
                        && let (Some(comparator), Some(shared_index)) =
                            (shadow_relocation_comparator.as_ref(), shared_index)
                        && let Some(family) = compressed_families
                            .iter()
                            .find(|family| family.source_id.source_index == shared_index)
                        && !Self::shadow_relocation_matches(comparator, family, coord0, formula)
                    {
                        relocation_mismatches.insert(shared_index);
                    }
                    Self::stage_formula(
                        engine,
                        sheet,
                        (coord0.row, coord0.col),
                        formula,
                        debug,
                        &mut formula_staging,
                    )
                },
            )?;
            if !relocation_mismatches.is_empty() {
                compressed_families.retain(|family| {
                    !relocation_mismatches.contains(&family.source_id.source_index)
                });
            }
        }

        if u64::try_from(value_cells_observed)
            .unwrap_or(u64::MAX)
            .saturating_add(u64::try_from(formula_count).unwrap_or(u64::MAX))
            > engine.workbook_load_limits().max_sheet_logical_cells
        {
            return Err(calamine::Error::Io(std::io::Error::other(format!(
                "Workbook load budget exceeded in calamine for sheet {sheet}: observed populated cell count exceeds configured logical-cell budget of {}",
                engine.workbook_load_limits().max_sheet_logical_cells
            ))));
        }
        enforce_sheet_dimension_limits(
            "calamine",
            sheet,
            dims_rows as u32,
            dims_cols as u32,
            engine.workbook_load_limits(),
        )
        .map_err(|error| calamine::Error::Io(std::io::Error::other(error.to_string())))?;

        let mut arrow_sheet = if let Some(mut arrow_sheet) = sparse {
            arrow_sheet.ensure_row_capacity(dims_rows.max(max_row_seen + 1));
            arrow_sheet
        } else {
            let mut state = dense.take().expect("dense state present");
            if state.row_started {
                state.aib.append_row(&state.row_vals).map_err(|error| {
                    calamine::Error::Io(std::io::Error::other(error.to_string()))
                })?;
            }
            let mut arrow_sheet = state.aib.finish();
            arrow_sheet.ensure_row_capacity(dims_rows.max(max_row_seen + 1));
            arrow_sheet
        };
        if dims_cols > arrow_sheet.columns.len() {
            arrow_sheet.insert_columns(
                arrow_sheet.columns.len(),
                dims_cols - arrow_sheet.columns.len(),
            );
        }
        let deferred_package = engine.config.defer_graph_building.then(|| {
            DeferredFormulaPackage::new_with_source_coordinates(
                sheet.to_string(),
                formula_source_report.clone(),
                compressed_families.clone(),
                partitioned_families.clone(),
                deferred_source_coordinates,
                Box::new(CalamineDeferredFormulaReplay::new(
                    formula_spool
                        .take()
                        .expect("deferred formula spool available"),
                    sheet.to_string(),
                    sheet_instance,
                )),
            )
            .with_complete_coordinate_coverage()
        });
        Ok(StreamedSheet {
            arrow_sheet,
            dimensions: (dims_rows, dims_cols),
            max_col_seen,
            used_sparse_fallback,
            value_cells_observed,
            values_handed_to_engine,
            formulas_observed: formula_count,
            formulas_handed_to_engine: formula_count,
            formulas: formula_staging.formulas,
            formula_source_report,
            compressed_families,
            partitioned_families,
            direct_preparation,
            deferred_package,
            shared_formula_tags,
            formula_spool_bytes,
            formula_spool_spilled,
            stream_millis: timer.elapsed_millis(),
        })
    }

    fn from_shared_source(
        source: SharedXlsxReader,
        cancel: Option<CancelToken>,
    ) -> Result<Self, calamine::Error> {
        let workbook: Xlsx<CancellableReader> =
            open_workbook_from_rs(CancellableReader::new(source.reader(), cancel.clone()))?;
        let sheet_names = workbook.sheet_names().to_vec();
        let calamine_defined_names = workbook.defined_names().to_vec();

        Ok(Self {
            workbook: RwLock::new(CalamineWorkbook(workbook)),
            source,
            cancel,
            loaded_sheets: HashSet::new(),
            cached_names: Some(sheet_names),
            calamine_defined_names,
            defined_names: OnceLock::new(),
            external_link_targets: OnceLock::new(),
            calc_settings: OnceLock::new(),
            load_stats: AdapterLoadStats::default(),
            shadow_relocation_comparator: None,
            #[cfg(test)]
            lazy_scan_counts: LazyScanCounts::default(),
            #[cfg(test)]
            stream_row_checkpoint_hook: None,
        })
    }

    fn cancellable_reader(&self) -> CancellableReader {
        CancellableReader::new(self.source.reader(), self.cancel.clone())
    }

    fn checkpoint_cancel(&self) -> Result<(), calamine::Error> {
        if self.cancel.as_ref().is_some_and(CancelToken::is_cancelled) {
            return Err(calamine::Error::Io(std::io::Error::other(
                "calamine load cancelled",
            )));
        }
        Ok(())
    }

    fn cancellation_checkpoint(cancel: Option<&CancelToken>) -> Result<(), calamine::Error> {
        if cancel.is_some_and(CancelToken::is_cancelled) {
            return Err(calamine::Error::Io(std::io::Error::other(
                "calamine load cancelled",
            )));
        }
        Ok(())
    }

    fn lazy_external_link_targets(&self) -> &BTreeMap<u32, String> {
        self.external_link_targets.get_or_init(|| {
            #[cfg(test)]
            self.lazy_scan_counts
                .external_links
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            Self::scan_external_link_targets_from_reader(self.cancellable_reader())
        })
    }

    fn lazy_calc_settings(&self) -> &Option<CalcSettings> {
        self.calc_settings.get_or_init(|| {
            #[cfg(test)]
            self.lazy_scan_counts
                .calc_settings
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            Self::scan_calc_settings_from_reader(self.cancellable_reader())
        })
    }

    /// Resolves workbook/sheet-scoped defined names on first request.
    ///
    /// Deliberately lock-free with respect to [`Self::workbook`]: it reads only
    /// the shared source and the `calamine_defined_names` snapshot taken at open
    /// time, so it is safe to call while the `workbook` write lock is held.
    fn lazy_defined_names(&self) -> &Vec<DefinedName> {
        self.defined_names.get_or_init(|| {
            #[cfg(test)]
            self.lazy_scan_counts
                .defined_names
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            if self.calamine_defined_names.is_empty() {
                return Vec::new();
            }

            let sheet_names = self.cached_names.as_deref().unwrap_or_default();
            let parsed =
                Self::scan_defined_names_from_reader(self.cancellable_reader(), sheet_names);
            if parsed.is_empty() {
                Self::fallback_defined_names(&self.calamine_defined_names, sheet_names)
            } else {
                parsed
            }
        })
    }

    pub fn external_link_target(&self, index: u32) -> Option<&str> {
        self.lazy_external_link_targets()
            .get(&index)
            .map(String::as_str)
    }

    fn normalize_open_ended_bounds(
        start_row: Option<u32>,
        start_col: Option<u32>,
        end_row: Option<u32>,
        end_col: Option<u32>,
    ) -> Option<(u32, u32, u32, u32)> {
        let mut sr = start_row;
        let mut sc = start_col;
        let mut er = end_row;
        let mut ec = end_col;

        if sr.is_none() && er.is_none() {
            sr = Some(1);
            er = Some(Self::EXCEL_MAX_ROWS);
        }
        if sc.is_none() && ec.is_none() {
            sc = Some(1);
            ec = Some(Self::EXCEL_MAX_COLS);
        }

        if sr.is_some() && er.is_none() {
            er = Some(Self::EXCEL_MAX_ROWS);
        }
        if er.is_some() && sr.is_none() {
            sr = Some(1);
        }

        if sc.is_some() && ec.is_none() {
            ec = Some(Self::EXCEL_MAX_COLS);
        }
        if ec.is_some() && sc.is_none() {
            sc = Some(1);
        }

        let sr = sr?;
        let sc = sc?;
        let er = er?;
        let ec = ec?;

        if er < sr || ec < sc {
            return None;
        }

        Some((sr, sc, er, ec))
    }

    fn convert_defined_name(
        name: &str,
        raw_formula: &str,
        local_sheet_id: Option<usize>,
        sheet_names: &[String],
    ) -> Option<DefinedName> {
        let mut trimmed = raw_formula.trim();
        if let Some(rest) = trimmed.strip_prefix('=') {
            trimmed = rest.trim();
        }
        if trimmed.is_empty() || trimmed.contains(',') {
            return None;
        }

        let reference = ReferenceType::from_string(trimmed).ok()?;
        let scope_sheet = local_sheet_id.and_then(|idx| sheet_names.get(idx).cloned());
        let scope = if scope_sheet.is_some() {
            DefinedNameScope::Sheet
        } else {
            DefinedNameScope::Workbook
        };
        let base_sheet = scope_sheet.as_deref();

        let (sheet_name, start_row, start_col, end_row, end_col) = match reference {
            ReferenceType::Cell {
                sheet, row, col, ..
            } => {
                let sheet = sheet.or_else(|| base_sheet.map(|s| s.to_string()))?;
                (sheet, row, col, row, col)
            }
            ReferenceType::Range {
                sheet,
                start_row,
                start_col,
                end_row,
                end_col,
                ..
            } => {
                let (sr, sc, er, ec) =
                    Self::normalize_open_ended_bounds(start_row, start_col, end_row, end_col)?;
                let sheet = sheet.or_else(|| base_sheet.map(|s| s.to_string()))?;
                (sheet, sr, sc, er, ec)
            }
            _ => return None,
        };

        let address = RangeAddress::new(sheet_name, start_row, start_col, end_row, end_col).ok()?;

        Some(DefinedName {
            name: name.to_string(),
            scope,
            scope_sheet,
            definition: DefinedNameDefinition::Range { address },
        })
    }

    fn decode_attr<R: BufRead>(
        reader: &XmlReader<R>,
        start: &BytesStart<'_>,
        key: &[u8],
    ) -> Option<String> {
        start
            .attributes()
            .filter_map(Result::ok)
            .find(|attr| attr.key == QName(key))
            .and_then(|attr| {
                attr.decode_and_unescape_value(reader.decoder())
                    .ok()
                    .map(|v| v.into_owned())
            })
    }

    fn append_xml_entity(
        entity: &BytesRef<'_>,
        buffer: &mut String,
    ) -> Result<(), quick_xml::Error> {
        let decoded = entity.decode()?;
        match decoded.as_ref() {
            "lt" => buffer.push('<'),
            "gt" => buffer.push('>'),
            "amp" => buffer.push('&'),
            "apos" => buffer.push('\''),
            "quot" => buffer.push('"'),
            _ => {
                if let Some(ch) = entity.resolve_char_ref()? {
                    buffer.push(ch);
                } else {
                    return Err(quick_xml::Error::Escape(
                        quick_xml::escape::EscapeError::UnrecognizedEntity(
                            0..0,
                            format!("&{decoded};"),
                        ),
                    ));
                }
            }
        }
        Ok(())
    }

    fn fallback_defined_names(
        calamine_defined_names: &[(String, String)],
        sheet_names: &[String],
    ) -> Vec<DefinedName> {
        let mut out = Vec::new();
        let mut seen: HashSet<(DefinedNameScope, Option<String>, String)> = HashSet::new();

        for (name, formula) in calamine_defined_names {
            if let Some(converted) = Self::convert_defined_name(name, formula, None, sheet_names) {
                let key = (
                    converted.scope.clone(),
                    converted.scope_sheet.clone(),
                    converted.name.clone(),
                );
                if seen.insert(key) {
                    out.push(converted);
                }
            }
        }

        out
    }

    fn scan_defined_names_from_reader<R>(reader: R, sheet_names: &[String]) -> Vec<DefinedName>
    where
        R: Read + Seek,
    {
        let mut archive = match ZipArchive::new(reader) {
            Ok(a) => a,
            Err(_) => return Vec::new(),
        };
        let entry = match archive.by_name("xl/workbook.xml") {
            Ok(e) => e,
            Err(_) => return Vec::new(),
        };

        // Calamine's public defined_names() surface flattens OOXML defined names to
        // (name, formula_text) and drops localSheetId. We recover only the scoped
        // defined-name metadata we need here with a targeted streaming pass over
        // workbook.xml, avoiding a full file String allocation or any sheet XML reparse.
        let mut xml = XmlReader::from_reader(BufReader::new(entry));
        xml.config_mut().trim_text(true);

        let mut out = Vec::new();
        let mut seen: HashSet<(DefinedNameScope, Option<String>, String)> = HashSet::new();
        let mut buf = Vec::new();
        let mut inner_buf = Vec::new();
        let mut in_defined_names = false;

        loop {
            buf.clear();
            match xml.read_event_into(&mut buf) {
                Ok(Event::Start(ref e)) if e.local_name().as_ref() == b"definedNames" => {
                    in_defined_names = true;
                }
                Ok(Event::End(ref e)) if e.local_name().as_ref() == b"definedNames" => {
                    break;
                }
                Ok(Event::Start(ref e))
                    if in_defined_names && e.local_name().as_ref() == b"definedName" =>
                {
                    let name = Self::decode_attr(&xml, e, b"name");
                    let local_sheet_id = Self::decode_attr(&xml, e, b"localSheetId")
                        .and_then(|v| v.parse::<usize>().ok());
                    let mut value = String::new();

                    loop {
                        inner_buf.clear();
                        match xml.read_event_into(&mut inner_buf) {
                            Ok(Event::Text(t)) => match t.xml10_content() {
                                Ok(text) => value.push_str(&text),
                                Err(_) => return Vec::new(),
                            },
                            Ok(Event::GeneralRef(entity)) => {
                                if Self::append_xml_entity(&entity, &mut value).is_err() {
                                    return Vec::new();
                                }
                            }
                            Ok(Event::End(end)) if end.name() == e.name() => break,
                            Ok(Event::Eof) => return Vec::new(),
                            Err(_) => return Vec::new(),
                            _ => {}
                        }
                    }

                    if let Some(name) = name
                        && let Some(converted) =
                            Self::convert_defined_name(&name, &value, local_sheet_id, sheet_names)
                    {
                        let key = (
                            converted.scope.clone(),
                            converted.scope_sheet.clone(),
                            converted.name.clone(),
                        );
                        if seen.insert(key) {
                            out.push(converted);
                        }
                    }
                }
                Ok(Event::Eof) => break,
                Err(_) => return Vec::new(),
                _ => {}
            }
        }

        out
    }

    fn scan_external_link_targets_from_reader<R>(reader: R) -> BTreeMap<u32, String>
    where
        R: Read + Seek,
    {
        let mut archive = match ZipArchive::new(reader) {
            Ok(a) => a,
            Err(_) => return BTreeMap::new(),
        };

        fn extract_target(xml: &str) -> Option<String> {
            let key = "Target=\"";
            let start = xml.find(key)? + key.len();
            let end = xml[start..].find('"')? + start;
            Some(xml[start..end].to_string())
        }

        let mut out = BTreeMap::new();
        for i in 0..archive.len() {
            let mut entry = match archive.by_index(i) {
                Ok(e) => e,
                Err(_) => continue,
            };
            let name = entry.name().to_string();
            let Some(rest) = name.strip_prefix("xl/externalLinks/_rels/externalLink") else {
                continue;
            };
            let Some(num_str) = rest.strip_suffix(".xml.rels") else {
                continue;
            };
            let Ok(idx) = num_str.parse::<u32>() else {
                continue;
            };

            let mut xml = String::new();
            if entry.read_to_string(&mut xml).is_ok()
                && let Some(target) = extract_target(&xml)
            {
                out.insert(idx, target);
            }
        }
        out
    }

    /// Parse the workbook-level `<calcPr>` settings (spec §9) straight from the
    /// `.xlsx` zip — calamine does not surface these. Reuses the shared
    /// `calc_pr` parser; returns `None` when `xl/workbook.xml` is missing or has
    /// no `<calcPr>` element.
    fn scan_calc_settings_from_reader<R>(reader: R) -> Option<CalcSettings>
    where
        R: Read + Seek,
    {
        let mut archive = ZipArchive::new(reader).ok()?;
        let mut entry = archive.by_name("xl/workbook.xml").ok()?;
        let mut xml = Vec::new();
        entry.read_to_end(&mut xml).ok()?;
        crate::calc_pr::parse_calc_pr(&xml)
    }

    fn calamine_error_code(e: &calamine::CellErrorType) -> u8 {
        let kind = match e {
            calamine::CellErrorType::Div0 => ExcelErrorKind::Div,
            calamine::CellErrorType::NA => ExcelErrorKind::Na,
            calamine::CellErrorType::Name => ExcelErrorKind::Name,
            calamine::CellErrorType::Null => ExcelErrorKind::Null,
            calamine::CellErrorType::Num => ExcelErrorKind::Num,
            calamine::CellErrorType::Ref => ExcelErrorKind::Ref,
            calamine::CellErrorType::Value => ExcelErrorKind::Value,
            _ => ExcelErrorKind::Error,
        };
        map_error_code(kind)
    }

    fn range_to_cells(
        range: &Range<Data>,
        formulas: Option<&Range<String>>,
        date_system: DateSystem,
    ) -> BTreeMap<(u32, u32), CellData> {
        let mut cells = BTreeMap::new();

        // We use the cells() iterator which gives us actual positions

        // Process values using actual positions

        let start_row = range.start().unwrap_or_default().0 as usize;
        let start_col = range.start().unwrap_or_default().1 as usize;

        for (row, col, val) in range.used_cells() {
            // Calamine uses 0-based indexing, convert to 1-based for Excel
            let excel_row = (row + start_row + 1) as u32;
            let excel_col = (col + start_col + 1) as u32;

            // Convert value (skip empty cells and empty strings)
            let value = match val {
                Data::Empty => None,
                Data::String(s) if s.is_empty() => None, // Treat empty strings as no value
                Data::String(s) => Some(LiteralValue::Text(s.clone())),
                Data::Float(f) => Some(LiteralValue::Number(*f)),
                Data::Int(i) => Some(LiteralValue::Int(*i)),
                Data::Bool(b) => Some(LiteralValue::Boolean(*b)),
                Data::Error(e) => {
                    let kind = match e {
                        calamine::CellErrorType::Div0 => ExcelErrorKind::Div,
                        calamine::CellErrorType::NA => ExcelErrorKind::Na,
                        calamine::CellErrorType::Name => ExcelErrorKind::Name,
                        calamine::CellErrorType::Null => ExcelErrorKind::Null,
                        calamine::CellErrorType::Num => ExcelErrorKind::Num,
                        calamine::CellErrorType::Ref => ExcelErrorKind::Ref,
                        calamine::CellErrorType::Value => ExcelErrorKind::Value,
                        _ => ExcelErrorKind::Value,
                    };
                    Some(LiteralValue::Error(ExcelError::new(kind)))
                }
                Data::DateTime(dt) => Some(
                    LiteralValue::try_from_serial_number_for(date_system, dt.as_f64())
                        .unwrap_or_else(LiteralValue::Error),
                ),
                Data::DateTimeIso(s) => Some(LiteralValue::Text(s.clone())),
                Data::DurationIso(s) => Some(LiteralValue::Text(s.clone())),
            };

            if value.is_some() {
                cells.insert(
                    (excel_row, excel_col),
                    CellData {
                        value,
                        formula: None,
                        style: None,
                    },
                );
            }
        }

        // Process formulas using their actual positions
        if let Some(frm_range) = formulas {
            let start_row = frm_range.start().unwrap_or_default().0 as usize;
            let start_col = frm_range.start().unwrap_or_default().1 as usize;

            for (row, col, formula) in frm_range.used_cells() {
                if !formula.is_empty() {
                    // Convert to 1-based Excel coordinates
                    let excel_row = (row + start_row + 1) as u32;
                    let excel_col = (col + start_col + 1) as u32;

                    // Ensure formula starts with '=' for proper parsing
                    let formula_with_eq = if formula.starts_with('=') {
                        formula.clone()
                    } else {
                        format!("={formula}")
                    };

                    // Update existing cell or create new one with formula
                    cells
                        .entry((excel_row, excel_col))
                        .and_modify(|cell| cell.formula = Some(formula_with_eq.clone()))
                        .or_insert_with(|| CellData {
                            value: None,
                            formula: Some(formula_with_eq),
                            style: None,
                        });
                }
            }
        }

        cells
    }
}

impl SpreadsheetReader for CalamineAdapter {
    type Error = calamine::Error;

    fn access_granularity(&self) -> AccessGranularity {
        AccessGranularity::Sheet
    }

    fn capabilities(&self) -> BackendCaps {
        BackendCaps {
            read: true,
            formulas: true,
            named_ranges: true,
            lazy_loading: false,
            random_access: false,
            styles: false,
            bytes_input: true,
            // conservative defaults
            date_system_1904: false,
            merged_cells: false,
            rich_text: false,
            hyperlinks: false,
            data_validations: false,
            shared_formulas: false,
            ..Default::default()
        }
    }

    fn sheet_names(&self) -> Result<Vec<String>, Self::Error> {
        Ok(self.cached_names.clone().unwrap_or_default())
    }

    fn load_stats(&self) -> Option<AdapterLoadStats> {
        Some(self.load_stats.clone())
    }

    fn defined_names(&mut self) -> Result<Vec<DefinedName>, Self::Error> {
        Ok(self.lazy_defined_names().clone())
    }

    fn calc_settings(&self) -> Option<CalcSettings> {
        self.lazy_calc_settings().clone()
    }

    fn open_path<P: AsRef<Path>>(path: P) -> Result<Self, Self::Error>
    where
        Self: Sized,
    {
        Self::open_path_with_source(path, XlsxPathSource::SharedFile)
    }

    fn open_reader(mut reader: Box<dyn Read + Send + Sync>) -> Result<Self, Self::Error>
    where
        Self: Sized,
    {
        let mut data = Vec::new();
        reader.read_to_end(&mut data).map_err(calamine::Error::Io)?;
        Self::from_shared_source(SharedXlsxReader::from_bytes(data), None)
    }

    fn open_bytes(data: Vec<u8>) -> Result<Self, Self::Error>
    where
        Self: Sized,
    {
        Self::from_shared_source(SharedXlsxReader::from_bytes(data), None)
    }

    fn read_range(
        &mut self,
        sheet: &str,
        start: (u32, u32),
        end: (u32, u32),
    ) -> Result<BTreeMap<(u32, u32), CellData>, Self::Error> {
        // Calamine loads entire sheet; filter after read_sheet
        let data = self.read_sheet(sheet)?;
        Ok(data
            .cells
            .into_iter()
            .filter(|((r, c), _)| *r >= start.0 && *r <= end.0 && *c >= start.1 && *c <= end.1)
            .collect())
    }

    fn read_sheet(&mut self, sheet: &str) -> Result<SheetData, Self::Error> {
        // Values
        let mut wb = self.workbook.write();
        let range = wb.worksheet_range(sheet)?;
        // Formulas (same dims as range, may be empty strings)
        let formulas = wb.worksheet_formula(sheet).ok();

        let dims = (range.height() as u32, range.width() as u32);
        // Calamine's ordinary reader API does not currently expose the
        // workbook date system at this boundary. Keep that policy explicit so
        // future metadata support only changes this selection point.
        let cells = Self::range_to_cells(&range, formulas.as_ref(), DateSystem::Excel1900);

        self.loaded_sheets.insert(sheet.to_string());

        Ok(SheetData {
            cells,
            dimensions: Some(dims),
            tables: vec![],
            named_ranges: vec![],
            date_system_1904: false, // calamine XLSX currently doesn’t expose this
            merged_cells: Vec::<MergedRange>::new(),
            hidden: false,
            // Explicit fallback: calamine does not expose row visibility metadata.
            row_hidden_manual: vec![],
            // Explicit fallback: filter-hidden row state is unavailable via calamine.
            row_hidden_filter: vec![],
        })
    }

    fn sheet_bounds(&self, sheet: &str) -> Option<(u32, u32)> {
        let mut wb = self.workbook.write();
        wb.worksheet_range(sheet)
            .ok()
            .map(|r| (r.height() as u32, r.width() as u32))
    }

    fn is_loaded(&self, sheet: &str, _row: Option<u32>, _col: Option<u32>) -> bool {
        self.loaded_sheets.contains(sheet)
    }
}

impl<R> EngineLoadStream<R> for CalamineAdapter
where
    R: EvaluationContext,
{
    type Error = calamine::Error;

    fn stream_into_engine(&mut self, engine: &mut EvalEngine<R>) -> Result<(), Self::Error> {
        use formualizer_eval::engine::named_range::{NameScope, NamedDefinition};
        use formualizer_eval::reference::{CellRef, Coord};

        #[cfg(feature = "tracing")]
        let _span_load = tracing::info_span!(
            "io_stream_into_engine",
            backend = "calamine",
            formula_records = true,
        )
        .entered();

        // Calamine 0.36 streams cached values and formula metadata from each XLSX
        // cell record in one pass. FormulaPlane staging and authoritative family
        // grouping remain unchanged downstream.
        self.checkpoint_cancel()?;
        let cancel = self.cancel.clone();
        let debug = std::env::var("FZ_DEBUG_LOAD")
            .ok()
            .is_some_and(|v| v != "0");
        let t0 = DebugTimer::start();
        let names = self.sheet_names()?;
        if debug {
            eprintln!("[fz][load] calamine: {} sheets", names.len());
        }
        // Single seam for sheet registration across every backend: folds the
        // engine's seeded default sheet into the file's first sheet on a fresh
        // engine and rejects duplicate names (#332). Registration is one bulk
        // call now, so the former per-sheet `io_load_sheet` span becomes one
        // `io_load_sheets` span over the whole registration.
        #[cfg(feature = "tracing")]
        let _span_sheets =
            tracing::info_span!("io_load_sheets", sheet_count = names.len()).entered();
        engine
            .adopt_file_sheets(names.iter().map(|n| n.as_str()))
            .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
        #[cfg(feature = "tracing")]
        drop(_span_sheets);

        let prev_index_mode = engine.config.sheet_index_mode;
        engine.set_sheet_index_mode(formualizer_eval::engine::SheetIndexMode::Lazy);
        let prev_range_limit = engine.config.range_expansion_limit;
        engine.config.range_expansion_limit = 0;
        let prev_first_load = engine.first_load_assume_new();
        engine.set_first_load_assume_new(true);
        engine.reset_ensure_touched();

        let load_result = (|| -> Result<(), calamine::Error> {
            let chunk_rows: usize = 32 * 1024;
            let mut total_values = 0usize;
            let mut total_value_cells_observed = 0usize;
            let mut total_formulas = 0usize;
            let mut total_formula_handed_to_engine = 0usize;
            let mut total_shared_formula_tags = 0usize;
            let mut workbook_spool_bytes_used = 0u64;
            let mut workbook_spill_files_used = 0u32;
            let mut eager_formula_batches: Vec<(FormulaIngestBatch, FormulaCompressedSourceBatch)> =
                Vec::new();
            let mut eager_direct_batches: Vec<(
                FormulaIngestBatch,
                FormulaCompressedSourceReport,
                FormulaCompressedPreparation,
            )> = Vec::new();

            for (sheet_instance, n) in names.iter().enumerate() {
                Self::cancellation_checkpoint(cancel.as_ref())?;
                let t_sheet = DebugTimer::start();
                if debug {
                    eprintln!("[fz][load] >> sheet '{n}'");
                }
                #[cfg(feature = "tracing")]
                let _span_sheet =
                    tracing::info_span!("io_populate_sheet", sheet = n.as_str()).entered();

                let shadow_relocation_comparator =
                    self.shadow_relocation_comparator.as_ref().map(Arc::clone);
                #[cfg(test)]
                let row_checkpoint_hook = self.stream_row_checkpoint_hook.as_deref();
                let streamed = {
                    let mut workbook = self.workbook.write();
                    Self::stream_worksheet(
                        &mut workbook.0,
                        n,
                        engine,
                        sheet_instance as u32,
                        StreamWorksheetOptions {
                            chunk_rows,
                            debug,
                            workbook_spool_usage: WorkbookSpoolUsage {
                                bytes: workbook_spool_bytes_used,
                                files: workbook_spill_files_used,
                            },
                            shadow_relocation_comparator,
                        },
                        cancel.as_ref(),
                        #[cfg(test)]
                        row_checkpoint_hook,
                    )?
                };
                let StreamedSheet {
                    arrow_sheet: asheet,
                    dimensions: (dims_rows, dims_cols),
                    max_col_seen,
                    used_sparse_fallback,
                    value_cells_observed: sheet_value_cells_observed,
                    values_handed_to_engine,
                    formulas_observed: parsed_n,
                    formulas_handed_to_engine: formula_handed_to_engine,
                    formulas,
                    formula_source_report,
                    compressed_families,
                    partitioned_families,
                    direct_preparation,
                    deferred_package,
                    shared_formula_tags,
                    formula_spool_bytes,
                    formula_spool_spilled,
                    stream_millis,
                } = streamed;
                workbook_spool_bytes_used = workbook_spool_bytes_used
                    .checked_add(formula_spool_bytes)
                    .expect("spool workbook accounting was preflighted");
                if formula_spool_spilled {
                    workbook_spill_files_used = workbook_spill_files_used
                        .checked_add(1)
                        .expect("spool file accounting was preflighted");
                }
                total_values += values_handed_to_engine;
                total_value_cells_observed += sheet_value_cells_observed;
                total_shared_formula_tags += shared_formula_tags;

                let store = engine.sheet_store_mut();
                if let Some(pos) = store.sheets.iter().position(|s| s.name.as_ref() == n) {
                    store.sheets[pos] = asheet;
                } else {
                    store.sheets.push(asheet);
                }

                if engine.config.defer_graph_building {
                    if let Some(package) = deferred_package {
                        engine.source_formula_ingress().stage_deferred(package);
                    }
                } else if !formulas.is_empty() || formula_source_report.source_formula_events != 0 {
                    let batch = FormulaIngestBatch::new(n.clone(), formulas);
                    if let Some(preparation) = direct_preparation {
                        eager_direct_batches.push((batch, formula_source_report, preparation));
                    } else {
                        eager_formula_batches.push((
                            batch,
                            FormulaCompressedSourceBatch::with_proposals(
                                n.clone(),
                                formula_source_report,
                                compressed_families,
                                partitioned_families,
                            ),
                        ));
                    }
                }

                total_formulas += parsed_n;
                total_formula_handed_to_engine += formula_handed_to_engine;
                if debug {
                    eprintln!(
                        "[fz][load]    streamed rows={} cols={} max_record_col={} sparse_fallback={} values={} formulas={} in {} ms",
                        dims_rows,
                        dims_cols,
                        max_col_seen + 1,
                        used_sparse_fallback,
                        sheet_value_cells_observed,
                        parsed_n,
                        stream_millis,
                    );
                    eprintln!(
                        "[fz][load] << sheet '{}' staged in {} ms",
                        n,
                        t_sheet.elapsed_millis()
                    );
                }
                self.loaded_sheets.insert(n.to_string());

                let row_hidden_manual: &[u32] = &[];
                let row_hidden_filter: &[u32] = &[];
                for row in row_hidden_manual {
                    engine
                        .set_row_hidden(
                            n,
                            *row,
                            true,
                            formualizer_eval::engine::RowVisibilitySource::Manual,
                        )
                        .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
                }
                for row in row_hidden_filter {
                    engine
                        .set_row_hidden(
                            n,
                            *row,
                            true,
                            formualizer_eval::engine::RowVisibilitySource::Filter,
                        )
                        .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
                }
            }

            if !engine.config.defer_graph_building && !eager_formula_batches.is_empty() {
                Self::cancellation_checkpoint(cancel.as_ref())?;
                engine
                    .source_formula_ingress()
                    .ingest_replay_batches(eager_formula_batches)
                    .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
                Self::cancellation_checkpoint(cancel.as_ref())?;
            }
            if !eager_direct_batches.is_empty() {
                Self::cancellation_checkpoint(cancel.as_ref())?;
                engine
                    .source_formula_ingress()
                    .finish_prepared(eager_direct_batches)
                    .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
                Self::cancellation_checkpoint(cancel.as_ref())?;
            }

            {
                use rustc_hash::FxHashSet;

                Self::cancellation_checkpoint(cancel.as_ref())?;
                let defined = self.defined_names()?;
                let mut seen: FxHashSet<(DefinedNameScope, Option<String>, String)> =
                    FxHashSet::default();

                for dn in defined {
                    Self::cancellation_checkpoint(cancel.as_ref())?;
                    let key = (dn.scope.clone(), dn.scope_sheet.clone(), dn.name.clone());
                    if !seen.insert(key) {
                        continue;
                    }

                    let scope = match dn.scope {
                        DefinedNameScope::Workbook => NameScope::Workbook,
                        DefinedNameScope::Sheet => {
                            let sheet_name = dn.scope_sheet.as_deref().ok_or_else(|| {
                                calamine::Error::Io(std::io::Error::other(format!(
                                    "sheet-scoped defined name `{}` missing scope_sheet",
                                    dn.name
                                )))
                            })?;
                            let sid = engine.sheet_id(sheet_name).ok_or_else(|| {
                                calamine::Error::Io(std::io::Error::other(format!(
                                    "scope sheet not found: {sheet_name}"
                                )))
                            })?;
                            NameScope::Sheet(sid)
                        }
                    };

                    let definition = match dn.definition {
                        DefinedNameDefinition::Range { address } => {
                            let sheet_id = engine
                                .sheet_id(&address.sheet)
                                .or_else(|| engine.add_sheet(&address.sheet).ok())
                                .ok_or_else(|| {
                                    calamine::Error::Io(std::io::Error::other(format!(
                                        "sheet not found: {}",
                                        address.sheet
                                    )))
                                })?;

                            let sr0 = address.start_row.saturating_sub(1);
                            let sc0 = address.start_col.saturating_sub(1);
                            let er0 = address.end_row.saturating_sub(1);
                            let ec0 = address.end_col.saturating_sub(1);

                            let start_ref =
                                CellRef::new(sheet_id, Coord::new(sr0, sc0, true, true));
                            if sr0 == er0 && sc0 == ec0 {
                                NamedDefinition::Cell(start_ref)
                            } else {
                                let end_ref =
                                    CellRef::new(sheet_id, Coord::new(er0, ec0, true, true));
                                let range_ref =
                                    formualizer_eval::reference::RangeRef::new(start_ref, end_ref);
                                NamedDefinition::Range(range_ref)
                            }
                        }
                        DefinedNameDefinition::Literal { value } => NamedDefinition::Literal(value),
                    };

                    engine
                        .define_name(&dn.name, definition, scope)
                        .map_err(|e| calamine::Error::Io(std::io::Error::other(e.to_string())))?;
                }
            }

            if debug {
                eprintln!(
                    "[fz][load] done: values={}, formulas={}, total={} ms",
                    total_values,
                    total_formulas,
                    t0.elapsed_millis(),
                );
            }
            for n in &names {
                engine.finalize_sheet_index(n);
            }

            self.load_stats = AdapterLoadStats {
                formula_cells_observed: Some(total_formulas as u64),
                value_cells_observed: Some(total_value_cells_observed as u64),
                value_slots_handed_to_engine: Some(total_values as u64),
                formula_cells_handed_to_engine: Some(total_formula_handed_to_engine as u64),
                shared_formula_tags_observed: Some(total_shared_formula_tags as u64),
            };
            Ok(())
        })();

        // Restore every temporary engine setting even when parsing, limits, or
        // graph ingest exits early.
        engine.set_first_load_assume_new(prev_first_load);
        engine.reset_ensure_touched();
        engine.set_sheet_index_mode(prev_index_mode);
        engine.config.range_expansion_limit = prev_range_limit;
        load_result
    }
}

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

    fn metadata_fixture() -> Vec<u8> {
        metadata_fixture_with_padding(0)
    }

    /// `padding` inserts a stored filler part immediately before the worksheet,
    /// pushing the worksheet's bytes past any readahead window that earlier
    /// parts warmed. Used to test behaviour at offsets that cannot be cached.
    fn metadata_fixture_with_padding(padding: usize) -> Vec<u8> {
        use std::io::Write;
        use zip::write::SimpleFileOptions;
        use zip::{CompressionMethod, ZipWriter};

        let mut writer = ZipWriter::new(Cursor::new(Vec::new()));
        let options = SimpleFileOptions::default().compression_method(CompressionMethod::Deflated);
        let entries = [
            (
                "[Content_Types].xml",
                r#"<?xml version="1.0"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"><Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/><Default Extension="xml" ContentType="application/xml"/><Override PartName="/xl/workbook.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet.main+xml"/><Override PartName="/xl/worksheets/sheet1.xml" ContentType="application/vnd.openxmlformats-officedocument.spreadsheetml.worksheet+xml"/></Types>"#,
            ),
            (
                "_rels/.rels",
                r#"<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="xl/workbook.xml"/></Relationships>"#,
            ),
            (
                "xl/workbook.xml",
                r#"<?xml version="1.0"?><workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main" xmlns:r="http://schemas.openxmlformats.org/officeDocument/2006/relationships"><sheets><sheet name="Data" sheetId="1" r:id="rId1"/></sheets><definedNames><definedName name="GlobalData">Data!$A$1</definedName><definedName name="LocalData" localSheetId="0">$A$1</definedName></definedNames><calcPr iterate="1" iterateCount="17" iterateDelta="0.01" calcMode="auto"/></workbook>"#,
            ),
            (
                "xl/_rels/workbook.xml.rels",
                r#"<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/worksheet" Target="worksheets/sheet1.xml"/></Relationships>"#,
            ),
            (
                "xl/worksheets/sheet1.xml",
                r#"<?xml version="1.0"?><worksheet xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"><dimension ref="A1:B1"/><sheetData><row r="1"><c r="A1"><v>42</v></c><c r="B1"><f>A1*2</f><v>84</v></c></row></sheetData></worksheet>"#,
            ),
            (
                "xl/externalLinks/_rels/externalLink7.xml.rels",
                r#"<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"><Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/externalLinkPath" Target="file:///tmp/source.xlsx" TargetMode="External"/></Relationships>"#,
            ),
        ];
        for (name, xml) in entries {
            if padding > 0 && name == "xl/worksheets/sheet1.xml" {
                writer
                    .start_file(
                        "xl/media/pad.bin",
                        SimpleFileOptions::default().compression_method(CompressionMethod::Stored),
                    )
                    .unwrap();
                writer
                    .write_all(&(0..=255_u8).cycle().take(padding).collect::<Vec<_>>())
                    .unwrap();
            }
            writer.start_file(name, options).unwrap();
            writer.write_all(xml.as_bytes()).unwrap();
        }
        writer.finish().unwrap().into_inner()
    }

    #[test]
    fn shared_bytes_have_one_backing_allocation() {
        let adapter = CalamineAdapter::open_bytes(metadata_fixture()).unwrap();
        let pointer = adapter.source.bytes_backing_ptr().unwrap();

        // One Arc is retained by the adapter for scanners and one by calamine's
        // cursor. Metadata cursors are cheap temporary Arc clones.
        assert_eq!(adapter.source.strong_count(), 2);
        assert_eq!(
            adapter.external_link_target(7),
            Some("file:///tmp/source.xlsx")
        );
        assert_eq!(adapter.source.bytes_backing_ptr(), Some(pointer));
        assert_eq!(adapter.source.strong_count(), 2);
    }

    #[test]
    fn shared_file_cursors_keep_independent_positions_under_concurrency() {
        use std::io::Write;

        let mut file = tempfile::tempfile().unwrap();
        let data: Vec<u8> = (0..=255).cycle().take(16 * 1024).collect();
        file.write_all(&data).unwrap();
        let source = SharedXlsxReader::File(SharedFileCursor::new(Arc::new(SharedFile {
            file: Mutex::new(file),
            len: data.len() as u64,
        })));

        std::thread::scope(|scope| {
            for thread_index in 0..8_u64 {
                let mut reader = source.reader();
                let data = &data;
                scope.spawn(move || {
                    for round in 0..200_u64 {
                        let offset = (thread_index * 997 + round * 43) % 16_000;
                        reader.seek(SeekFrom::Start(offset)).unwrap();
                        let mut actual = [0_u8; 31];
                        reader.read_exact(&mut actual).unwrap();
                        assert_eq!(&actual, &data[offset as usize..offset as usize + 31]);
                        assert_eq!(reader.stream_position().unwrap(), offset + 31);
                    }
                });
            }
        });
    }

    /// Truncation under the safe default must degrade to ordinary I/O results,
    /// never to a mapped-memory fault. The fixture is padded past the readahead
    /// window so the worksheet lives at an offset no cursor can have cached;
    /// bytes already inside a window may still be served after truncation,
    /// which is a staleness widening, not a memory-safety hazard.
    #[cfg(unix)]
    #[test]
    fn truncation_after_safe_open_returns_normal_fallbacks_and_errors() {
        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(
            file.path(),
            metadata_fixture_with_padding(4 * SHARED_FILE_READAHEAD),
        )
        .unwrap();
        let mut adapter = CalamineAdapter::open_path(file.path()).unwrap();
        assert!(matches!(adapter.source, SharedXlsxReader::File(_)));

        std::fs::OpenOptions::new()
            .write(true)
            .open(file.path())
            .unwrap()
            .set_len(0)
            .unwrap();

        assert_eq!(adapter.external_link_target(7), None);
        assert_eq!(adapter.calc_settings(), None);
        assert!(adapter.read_sheet("Data").is_err());
    }

    #[cfg(any(unix, windows))]
    #[test]
    fn runtime_direct_mmap_uses_mapping_for_nonempty_path() {
        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(file.path(), metadata_fixture()).unwrap();
        let adapter =
            CalamineAdapter::open_path_with_source(file.path(), XlsxPathSource::DirectMmap)
                .unwrap();
        assert!(matches!(adapter.source, SharedXlsxReader::Mapped { .. }));
    }

    #[test]
    fn default_open_path_uses_shared_file() {
        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(file.path(), metadata_fixture()).unwrap();
        let adapter = CalamineAdapter::open_path(file.path()).unwrap();
        assert_eq!(XlsxPathSource::default(), XlsxPathSource::SharedFile);
        assert!(matches!(adapter.source, SharedXlsxReader::File(_)));
    }

    /// The legacy `mmap` Cargo feature is a compatibility alias and must not
    /// select load behaviour. This test only compiles under
    /// `--features calamine,mmap`, so the no-op is pinned rather than assumed.
    #[cfg(feature = "mmap")]
    #[test]
    fn legacy_mmap_feature_does_not_select_path_behaviour() {
        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(file.path(), metadata_fixture()).unwrap();

        assert_eq!(XlsxPathSource::default(), XlsxPathSource::SharedFile);
        let default_adapter = CalamineAdapter::open_path(file.path()).unwrap();
        assert!(matches!(default_adapter.source, SharedXlsxReader::File(_)));
        let requested_default =
            CalamineAdapter::open_path_with_source(file.path(), XlsxPathSource::default()).unwrap();
        assert!(matches!(
            requested_default.source,
            SharedXlsxReader::File(_)
        ));

        // Mapping remains reachable only through the explicit opt-in.
        let mapped =
            CalamineAdapter::open_path_with_source(file.path(), XlsxPathSource::DirectMmap)
                .unwrap();
        assert!(matches!(mapped.source, SharedXlsxReader::Mapped { .. }));
    }

    /// The readahead window is keyed by absolute offset, so backward and
    /// forward seeks inside it must not re-read the file or desynchronise.
    #[test]
    fn shared_file_readahead_serves_seeks_within_the_window() {
        use std::io::Write;

        let mut file = tempfile::tempfile().unwrap();
        let data: Vec<u8> = (0..=255).cycle().take(4 * SHARED_FILE_READAHEAD).collect();
        file.write_all(&data).unwrap();
        let source = SharedXlsxReader::File(SharedFileCursor::new(Arc::new(SharedFile {
            file: Mutex::new(file),
            len: data.len() as u64,
        })));

        let mut reader = source.reader();
        for offset in [0_u64, 4096, 1, 64, 4095, 8192, 200_000, 199_999] {
            reader.seek(SeekFrom::Start(offset)).unwrap();
            let mut actual = [0_u8; 17];
            reader.read_exact(&mut actual).unwrap();
            assert_eq!(&actual, &data[offset as usize..offset as usize + 17]);
            assert_eq!(reader.stream_position().unwrap(), offset + 17);
        }

        // A request larger than the window bypasses it and still reads exactly.
        reader.seek(SeekFrom::Start(3)).unwrap();
        let mut large = vec![0_u8; 2 * SHARED_FILE_READAHEAD];
        reader.read_exact(&mut large).unwrap();
        assert_eq!(&large[..], &data[3..3 + 2 * SHARED_FILE_READAHEAD]);
    }

    #[cfg(any(unix, windows))]
    #[test]
    fn direct_mmap_mapping_failure_is_returned_without_fallback() {
        let file = tempfile::NamedTempFile::new().unwrap();
        let error = SharedXlsxReader::from_file(file.reopen().unwrap(), XlsxPathSource::DirectMmap)
            .err()
            .expect("empty files cannot be mapped");
        assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput);

        let error = CalamineAdapter::open_path_with_source(file.path(), XlsxPathSource::DirectMmap)
            .err()
            .expect("the public API must return the mapping error");
        assert!(
            matches!(error, calamine::Error::Io(error) if error.kind() == std::io::ErrorKind::InvalidInput)
        );
    }

    #[test]
    fn metadata_scans_are_lazy_and_run_at_most_once() {
        use std::sync::atomic::Ordering;

        let mut adapter = CalamineAdapter::open_bytes(metadata_fixture()).unwrap();
        assert_eq!(
            adapter
                .lazy_scan_counts
                .external_links
                .load(Ordering::Relaxed),
            0
        );
        assert_eq!(
            adapter
                .lazy_scan_counts
                .calc_settings
                .load(Ordering::Relaxed),
            0
        );
        assert_eq!(
            adapter
                .lazy_scan_counts
                .defined_names
                .load(Ordering::Relaxed),
            0
        );

        std::thread::scope(|scope| {
            let adapter = &adapter;
            for _ in 0..4 {
                scope.spawn(move || {
                    assert_eq!(
                        adapter.external_link_target(7),
                        Some("file:///tmp/source.xlsx")
                    );
                    assert_eq!(adapter.calc_settings().unwrap().iterate_count, Some(17));
                });
            }
        });
        assert_eq!(adapter.defined_names().unwrap().len(), 2);
        assert_eq!(adapter.defined_names().unwrap().len(), 2);

        assert_eq!(
            adapter
                .lazy_scan_counts
                .external_links
                .load(Ordering::Relaxed),
            1
        );
        assert_eq!(
            adapter
                .lazy_scan_counts
                .calc_settings
                .load(Ordering::Relaxed),
            1
        );
        assert_eq!(
            adapter
                .lazy_scan_counts
                .defined_names
                .load(Ordering::Relaxed),
            1
        );
    }

    #[test]
    fn path_and_bytes_metadata_are_identical() {
        let bytes = metadata_fixture();
        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(file.path(), &bytes).unwrap();
        let mut from_path = CalamineAdapter::open_path(file.path()).unwrap();
        let mut from_bytes = CalamineAdapter::open_bytes(bytes).unwrap();

        assert_eq!(
            from_path.sheet_names().unwrap(),
            from_bytes.sheet_names().unwrap()
        );
        assert_eq!(
            from_path.defined_names().unwrap(),
            from_bytes.defined_names().unwrap()
        );
        assert_eq!(from_path.calc_settings(), from_bytes.calc_settings());
        assert_eq!(
            from_path.external_link_target(7),
            from_bytes.external_link_target(7)
        );
    }

    #[cfg(any(unix, windows))]
    #[test]
    fn shared_file_and_direct_mmap_have_full_semantic_parity() {
        use crate::{LoadStrategy, Workbook, WorkbookConfig};

        #[derive(Debug, PartialEq)]
        struct Observed {
            sheet_names: Vec<String>,
            defined_names: Vec<DefinedName>,
            calc_settings: Option<CalcSettings>,
            external_target: Option<String>,
            value: Option<LiteralValue>,
            formula: Option<String>,
            evaluated: LiteralValue,
        }

        fn observe(path: &Path, policy: XlsxPathSource) -> Observed {
            let mut adapter = CalamineAdapter::open_path_with_source(path, policy).unwrap();
            let sheet_names = adapter.sheet_names().unwrap();
            let defined_names = adapter.defined_names().unwrap();
            let calc_settings = adapter.calc_settings();
            let external_target = adapter.external_link_target(7).map(str::to_string);
            let mut workbook =
                Workbook::from_reader(adapter, LoadStrategy::EagerAll, WorkbookConfig::ephemeral())
                    .unwrap();
            Observed {
                sheet_names,
                defined_names,
                calc_settings,
                external_target,
                value: workbook.get_value("Data", 1, 1),
                formula: workbook.get_formula("Data", 1, 2),
                evaluated: workbook.evaluate_cell("Data", 1, 2).unwrap(),
            }
        }

        let file = tempfile::NamedTempFile::new().unwrap();
        std::fs::write(file.path(), metadata_fixture()).unwrap();
        let shared = observe(file.path(), XlsxPathSource::SharedFile);
        let mapped = observe(file.path(), XlsxPathSource::DirectMmap);

        assert_eq!(shared, mapped);
        assert_eq!(shared.sheet_names, ["Data"]);
        assert_eq!(shared.defined_names.len(), 2);
        assert!(
            shared
                .defined_names
                .iter()
                .any(|name| name.name == "GlobalData" && name.scope_sheet.is_none())
        );
        assert!(
            shared
                .defined_names
                .iter()
                .any(|name| name.name == "LocalData" && name.scope_sheet.as_deref() == Some("Data"))
        );
        assert_eq!(shared.calc_settings.unwrap().iterate_count, Some(17));
        assert_eq!(
            shared.external_target.as_deref(),
            Some("file:///tmp/source.xlsx")
        );
        assert_eq!(shared.value, Some(LiteralValue::Number(42.0)));
        assert_eq!(shared.formula.as_deref(), Some("=A1 * 2"));
        assert_eq!(shared.evaluated, LiteralValue::Number(84.0));
    }

    #[cfg(unix)]
    #[test]
    fn path_is_not_reopened_for_lazy_metadata_or_sheet_reads() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("source.xlsx");
        std::fs::write(&path, metadata_fixture()).unwrap();
        let mut adapter = CalamineAdapter::open_path(&path).unwrap();
        std::fs::remove_file(&path).unwrap();

        assert_eq!(
            adapter.external_link_target(7),
            Some("file:///tmp/source.xlsx")
        );
        assert_eq!(adapter.calc_settings().unwrap().iterate_count, Some(17));
        assert_eq!(adapter.defined_names().unwrap().len(), 2);
        assert_eq!(
            adapter.read_sheet("Data").unwrap().cells[&(1, 1)].value,
            Some(LiteralValue::Number(42.0))
        );
    }

    #[test]
    fn new_calamine_error_variants_preserve_generic_error_semantics() {
        let error = calamine::CellErrorType::GettingData;
        assert!(matches!(
            data_ref_to_literal(&DataRef::Error(error.clone()), DateSystem::Excel1900),
            Some(LiteralValue::Error(ref value)) if value.kind == ExcelErrorKind::Error
        ));
        assert!(matches!(
            data_ref_to_overlay(&DataRef::Error(error)),
            Some(OverlayValue::Error(8))
        ));
    }

    #[test]
    fn cancellable_open_rejects_pre_cancelled_token_without_interrupted_error() {
        let token = CancelToken::new();
        token.cancel();
        let mut reader = CancellableReader::new(
            SharedXlsxReader::from_bytes(metadata_fixture()),
            Some(token.clone()),
        );
        let read_error = reader.read(&mut [0_u8; 1]).expect_err("read must stop");
        assert_ne!(read_error.kind(), std::io::ErrorKind::Interrupted);

        let error = match CalamineAdapter::open_bytes_cancellable(metadata_fixture(), token) {
            Ok(_) => panic!("pre-cancelled open must stop before parsing"),
            Err(error) => error,
        };
        assert!(
            !error.to_string().is_empty(),
            "cancellation must remain a reportable Calamine error"
        );
    }

    #[test]
    fn cancellable_stream_stops_at_a_row_checkpoint() {
        let token = CancelToken::new();
        let mut adapter =
            CalamineAdapter::open_bytes_cancellable(metadata_fixture(), token.clone())
                .expect("open uncancelled workbook");
        adapter.stream_row_checkpoint_hook = Some(Arc::new(move || token.cancel()));
        let context = formualizer_eval::test_workbook::TestWorkbook::new();
        let mut engine = EvalEngine::new(context, Default::default());

        let error = adapter
            .stream_into_engine(&mut engine)
            .expect_err("row checkpoint must observe cancellation");
        assert!(matches!(
            error,
            calamine::Error::Io(ref error) if error.kind() != std::io::ErrorKind::Interrupted
        ));
    }

    #[test]
    fn uncancelled_cancellable_open_matches_open_bytes() {
        let bytes = metadata_fixture();
        let plain = CalamineAdapter::open_bytes(bytes.clone()).expect("plain open");
        let cancellable = CalamineAdapter::open_bytes_cancellable(bytes, CancelToken::new())
            .expect("uncancelled cancellable open");
        assert_eq!(
            plain.sheet_names().unwrap(),
            cancellable.sheet_names().unwrap()
        );
        assert_eq!(
            plain.calamine_defined_names,
            cancellable.calamine_defined_names
        );
    }

    #[test]
    fn dateish_backend_signal_maps_date_time_datetime_and_duration() {
        use calamine::{ExcelDateTime, ExcelDateTimeType};
        use formualizer_eval::format::FormatId;

        let date = DataRef::DateTime(ExcelDateTime::new(
            45_583.0,
            ExcelDateTimeType::DateTime,
            false,
        ));
        let time = DataRef::DateTime(ExcelDateTime::new(0.5, ExcelDateTimeType::DateTime, false));
        let datetime = DataRef::DateTime(ExcelDateTime::new(
            45_583.5,
            ExcelDateTimeType::DateTime,
            false,
        ));
        let duration =
            DataRef::DateTime(ExcelDateTime::new(1.5, ExcelDateTimeType::TimeDelta, false));

        assert_eq!(data_ref_format(&date), Some(FormatId::DATE));
        assert_eq!(data_ref_format(&time), Some(FormatId::TIME));
        assert_eq!(data_ref_format(&datetime), Some(FormatId::DATETIME));
        assert_eq!(data_ref_format(&duration), Some(FormatId::DURATION));
    }
}