formualizer-eval 0.10.0

use crate::engine::graph::DependencyGraph;
use formualizer_common::Coord as AbsCoord;
// use crate::engine::plan::RangeKey; // no longer needed directly here
use crate::engine::EvalConfig;
use crate::engine::arena::AstNodeId;
use crate::engine::ingest_pipeline::{DependencyPlanRow, FormulaAstInput};
use crate::engine::plan::{
    DependencyPlan, F_HAS_NAMES, F_HAS_RANGES, F_HAS_TABLES, F_VOLATILE, RangeKey,
};
use crate::{SheetId, engine::vertex::VertexId};
use formualizer_common::{CoordBuildHasher, ExcelError, PackedSheetCell};
use formualizer_parse::parser::ASTNode;
use rustc_hash::FxHashMap;
use std::collections::HashMap;
use std::sync::Arc;

/// Summary of bulk ingest
#[derive(Debug, Clone)]
pub struct BulkIngestSummary {
    pub sheets: usize,
    pub vertices: usize,
    pub formulas: usize,
    pub edges: usize,
    pub elapsed: std::time::Duration,
}

enum FormulaAstSource {
    Owned(ASTNode),
    Interned(AstNodeId),
    Planned {
        formula: crate::engine::graph::FormulaRef,
        plan: DependencyPlanRow,
    },
}

struct StagedFormula {
    row: u32,
    col: u32,
    ast: FormulaAstSource,
}

struct RegistryFunctionProvider;

impl crate::traits::FunctionProvider for RegistryFunctionProvider {
    fn planning_semantic_revision(&self) -> Option<u64> {
        Some(0)
    }

    fn get_function(&self, ns: &str, name: &str) -> Option<Arc<dyn crate::function::Function>> {
        crate::function_registry::get(ns, name)
    }

    fn get_function_for_planning(
        &self,
        ns: &str,
        name: &str,
    ) -> Option<std::sync::Arc<dyn crate::function::Function>> {
        crate::function_registry::get_for_planning(ns, name)
    }
}

struct SheetStage {
    name: String,
    id: SheetId,
    formulas: Vec<StagedFormula>,
    /// Unplanned formulas (1-based row, col, formula), planned by `finish`
    /// one chunk at a time.
    records: Vec<(u32, u32, crate::engine::graph::FormulaRef)>,
}

impl SheetStage {
    fn new(name: String, id: SheetId) -> Self {
        Self {
            name,
            id,
            formulas: Vec::new(),
            records: Vec::new(),
        }
    }
}

fn range_key_from_shared(
    range: &crate::reference::SharedRangeRef<'static>,
    sheet_reg: &crate::engine::sheet_registry::SheetRegistry,
    current_sheet: SheetId,
) -> RangeKey {
    // `Current` is the sheet the formula lives on. An unresolvable sheet name
    // falls back to it so the range still produces a key.
    let sheet = sheet_reg
        .resolve_locator(&range.sheet, current_sheet)
        .unwrap_or(current_sheet);
    match (
        range.start_row,
        range.start_col,
        range.end_row,
        range.end_col,
    ) {
        (Some(sr), Some(sc), Some(er), Some(ec)) => RangeKey::Rect {
            sheet,
            start: AbsCoord::new(sr.index, sc.index),
            end: AbsCoord::new(er.index, ec.index),
        },
        (None, Some(sc), None, Some(ec)) if sc.index == ec.index => RangeKey::WholeCol {
            sheet,
            col: sc.index + 1,
        },
        (Some(sr), None, Some(er), None) if sr.index == er.index => RangeKey::WholeRow {
            sheet,
            row: sr.index + 1,
        },
        _ => RangeKey::OpenRect {
            sheet,
            start_row: range.start_row.map(|bound| bound.index),
            start_col: range.start_col.map(|bound| bound.index),
            end_row: range.end_row.map(|bound| bound.index),
            end_col: range.end_col.map(|bound| bound.index),
        },
    }
}

fn dependency_plan_from_rows(
    sheet_reg: &crate::engine::sheet_registry::SheetRegistry,
    sheet_id: SheetId,
    rows: &[(u32, u32, DependencyPlanRow)],
) -> DependencyPlan {
    let mut plan = DependencyPlan::default();
    let mut cell_index: HashMap<PackedSheetCell, u32, CoordBuildHasher> =
        HashMap::with_hasher(CoordBuildHasher);
    let mut vertex_pool_index: HashMap<PackedSheetCell, u32, CoordBuildHasher> =
        HashMap::with_hasher(CoordBuildHasher);

    fn ensure_vertex_pool_index(
        plan: &mut DependencyPlan,
        index: &mut HashMap<PackedSheetCell, u32, CoordBuildHasher>,
        key: (SheetId, AbsCoord),
    ) -> u32 {
        let packed = PackedSheetCell::try_new(key.0, key.1.row(), key.1.col())
            .expect("plan vertex pool coordinate must fit PackedSheetCell");
        if let Some(&idx) = index.get(&packed) {
            idx
        } else {
            let idx = plan.vertex_pool.len() as u32;
            plan.vertex_pool.push(key);
            plan.vertex_pool_packed.push(packed);
            index.insert(packed, idx);
            idx
        }
    }

    // Targets enter the vertex pool first, column by column, so new
    // formula vertices of a chunk get consecutive ids down each column
    // (a filled-down family is one id run per column: Program 2 keeps such
    // runs out of the per-cell maps). Referenced cells follow.
    let mut targets: Vec<(u32, u32)> = rows.iter().map(|&(row, col, _)| (col, row)).collect();
    targets.sort_unstable();
    for (col, row) in targets {
        ensure_vertex_pool_index(
            &mut plan,
            &mut vertex_pool_index,
            (sheet_id, AbsCoord::from_excel(row, col)),
        );
    }

    for (row, col, row_plan) in rows {
        let target = (sheet_id, AbsCoord::from_excel(*row, *col));
        plan.formula_targets.push(target);
        let target_pool_idx = ensure_vertex_pool_index(&mut plan, &mut vertex_pool_index, target);
        plan.formula_target_pool_indices.push(target_pool_idx);

        let mut per_cells = Vec::new();
        for dep in &row_plan.direct_cell_deps {
            let key = (
                dep.sheet_id,
                AbsCoord::new(dep.coord.row(), dep.coord.col()),
            );
            let packed = PackedSheetCell::try_new(key.0, key.1.row(), key.1.col())
                .expect("plan dependency coordinate must fit PackedSheetCell");
            let idx = if let Some(&idx) = cell_index.get(&packed) {
                idx
            } else {
                let idx = plan.global_cells.len() as u32;
                plan.global_cells.push(key);
                cell_index.insert(packed, idx);
                let pool_idx = ensure_vertex_pool_index(&mut plan, &mut vertex_pool_index, key);
                plan.global_cell_pool_indices.push(pool_idx);
                idx
            };
            per_cells.push(idx);
        }

        let flags = (if row_plan.volatile { F_VOLATILE } else { 0 })
            | (if row_plan.range_deps.is_empty() {
                0
            } else {
                F_HAS_RANGES
            })
            | (if row_plan.named_refs.is_empty() && row_plan.resolved_named_refs.is_empty() {
                0
            } else {
                F_HAS_NAMES
            })
            | (if row_plan.table_refs.is_empty() {
                0
            } else {
                F_HAS_TABLES
            });
        plan.per_formula_cells.push(per_cells);
        plan.per_formula_ranges.push(
            row_plan
                .range_deps
                .iter()
                .map(|range| range_key_from_shared(range, sheet_reg, sheet_id))
                .collect(),
        );
        let mut names = row_plan.resolved_named_refs.clone();
        names.extend(row_plan.named_refs.clone());
        names.extend(row_plan.source_refs.clone());
        plan.per_formula_names.push(names);
        let mut tables = row_plan.table_refs.clone();
        tables.extend(row_plan.source_refs.clone());
        plan.per_formula_tables.push(tables);
        plan.per_formula_flags.push(flags);
    }
    plan
}

pub struct BulkIngestBuilder<'g> {
    g: &'g mut DependencyGraph,
    sheets: FxHashMap<SheetId, SheetStage>,
    cfg_saved: EvalConfig,
    admission_preflighted: bool,
}

impl<'g> BulkIngestBuilder<'g> {
    pub fn new(g: &'g mut DependencyGraph) -> Self {
        let cfg_saved = g.get_config().clone();
        // Respect current sheet index mode (loader may set Lazy to skip index work during ingest)
        Self {
            g,
            sheets: FxHashMap::default(),
            cfg_saved,
            admission_preflighted: false,
        }
    }

    pub(crate) fn mark_admission_preflighted(&mut self) {
        self.admission_preflighted = true;
    }

    pub fn add_sheet(&mut self, name: &str) -> SheetId {
        let id = self.g.sheet_id(name).unwrap_or_else(|| {
            panic!(
                "BulkIngestBuilder::add_sheet requires pre-existing sheet; call Engine::add_sheet first: {name}"
            )
        });
        self.sheets
            .entry(id)
            .or_insert_with(|| SheetStage::new(name.to_string(), id));
        id
    }

    /// [`Self::add_sheet`] returning `None` for an unknown sheet.
    pub(crate) fn add_sheet_checked(&mut self, name: &str) -> Option<SheetId> {
        let id = self.g.sheet_id(name)?;
        self.sheets
            .entry(id)
            .or_insert_with(|| SheetStage::new(name.to_string(), id));
        Some(id)
    }

    pub fn add_formulas<I>(&mut self, sheet: SheetId, formulas: I)
    where
        I: IntoIterator<Item = (u32, u32, ASTNode)>,
    {
        let stage = self
            .sheets
            .entry(sheet)
            .or_insert_with(|| SheetStage::new(self.g.sheet_name(sheet).to_string(), sheet));
        for (r, c, ast) in formulas {
            stage.formulas.push(StagedFormula {
                row: r,
                col: c,
                ast: FormulaAstSource::Owned(ast),
            });
        }
    }

    pub fn add_formula_ids<I>(&mut self, sheet: SheetId, formulas: I)
    where
        I: IntoIterator<Item = (u32, u32, AstNodeId)>,
    {
        let stage = self
            .sheets
            .entry(sheet)
            .or_insert_with(|| SheetStage::new(self.g.sheet_name(sheet).to_string(), sheet));
        for (r, c, ast_id) in formulas {
            stage.formulas.push(StagedFormula {
                row: r,
                col: c,
                ast: FormulaAstSource::Interned(ast_id),
            });
        }
    }

    pub(crate) fn add_formula_plans<I>(&mut self, sheet: SheetId, formulas: I)
    where
        I: IntoIterator<
            Item = (
                u32,
                u32,
                crate::engine::graph::FormulaRef,
                DependencyPlanRow,
            ),
        >,
    {
        let stage = self
            .sheets
            .entry(sheet)
            .or_insert_with(|| SheetStage::new(self.g.sheet_name(sheet).to_string(), sheet));
        for (r, c, formula, plan) in formulas {
            stage.formulas.push(StagedFormula {
                row: r,
                col: c,
                ast: FormulaAstSource::Planned { formula, plan },
            });
        }
    }

    /// Stage formulas (own ASTs or load-time family members) for planning
    /// in `finish`, a chunk at a time: the dependency plans of the whole
    /// batch never coexist. A planning error leaves the chunks before it
    /// applied, so this is for loads (where an error fails the load).
    pub(crate) fn add_formula_refs<I>(&mut self, sheet: SheetId, formulas: I)
    where
        I: IntoIterator<Item = (u32, u32, crate::engine::graph::FormulaRef)>,
    {
        let stage = self
            .sheets
            .entry(sheet)
            .or_insert_with(|| SheetStage::new(self.g.sheet_name(sheet).to_string(), sheet));
        stage.records.extend(formulas);
    }

    pub fn finish(self) -> Result<BulkIngestSummary, ExcelError> {
        self.finish_with_provider(&RegistryFunctionProvider)
    }

    /// [`Self::finish`], planning staged records with `provider`.
    pub(crate) fn finish_with_provider(
        mut self,
        records_provider: &dyn crate::traits::FunctionProvider,
    ) -> Result<BulkIngestSummary, ExcelError> {
        use crate::instant::FzInstant as Instant;
        let t0 = Instant::now();
        let dbg = std::env::var("FZ_DEBUG_INGEST")
            .ok()
            .is_some_and(|v| v != "0")
            || std::env::var("FZ_DEBUG_LOAD")
                .ok()
                .is_some_and(|v| v != "0");
        let provider = RegistryFunctionProvider;
        for stage in self.sheets.values_mut() {
            let mut inputs = Vec::new();
            let mut indices = Vec::new();
            for (index, formula) in stage.formulas.iter().enumerate() {
                let placement = crate::reference::CellRef::new(
                    stage.id,
                    crate::reference::Coord::from_excel(formula.row, formula.col, true, true),
                );
                match &formula.ast {
                    FormulaAstSource::Owned(ast) => {
                        indices.push(index);
                        inputs.push((FormulaAstInput::Tree(ast.clone()), placement, None));
                    }
                    FormulaAstSource::Interned(ast_id) => {
                        indices.push(index);
                        inputs.push((FormulaAstInput::RawArena(*ast_id), placement, None));
                    }
                    FormulaAstSource::Planned { .. } => {}
                }
            }
            if !inputs.is_empty() {
                let ingested = self.g.ingest_pipeline(&provider).ingest_batch(inputs)?;
                for (index, formula) in indices.into_iter().zip(ingested) {
                    stage.formulas[index].ast = FormulaAstSource::Planned {
                        formula: crate::engine::graph::FormulaRef::of_ingested(
                            formula.ast_id,
                            formula.member_anchor,
                        ),
                        plan: formula.dep_plan,
                    };
                }
            }
        }
        let budgets = self.cfg_saved.resolved_evaluation_budgets();
        if !self.admission_preflighted
            && crate::engine::resource_ledger::graph_admission_enabled(&budgets)
        {
            let mut admission_plans = Vec::new();
            for stage in self.sheets.values() {
                for formula in &stage.formulas {
                    let FormulaAstSource::Planned { plan, .. } = &formula.ast else {
                        unreachable!("bulk formulas are planned before admission")
                    };
                    admission_plans.push((stage.id, formula.row, formula.col, plan.clone()));
                }
            }
            let admission = self.g.preview_formula_mutations(&admission_plans)?;
            crate::engine::resource_ledger::preflight_graph_admission(&budgets, admission, None)
                .map_err(crate::engine::ResourceLedgerError::into_excel_error)?;
        }

        let had_vertices = self.g.vertex_count() != 0;
        let mut dirty_roots = Vec::new();
        let mut total_vertices = 0usize;
        let mut total_formulas = 0usize;
        let mut total_edges = 0usize;

        if dbg {
            eprintln!(
                "[fz][ingest] starting bulk ingest with {} sheets",
                self.sheets.len()
            );
        }

        // Materialize per-sheet to keep caches warm and reduce cross-sheet churn
        // Accumulate a flat adjacency for a single-shot CSR build
        #[cfg_attr(not(any(test, feature = "legacy_oracle")), allow(unused_mut))]
        let mut edges_adj: Vec<(u32, Vec<u32>)> = Vec::new();
        #[cfg_attr(not(any(test, feature = "legacy_oracle")), allow(unused_mut))]
        let mut coord_accum: Vec<crate::engine::addr::VertexAddr> = Vec::new();
        #[cfg_attr(not(any(test, feature = "legacy_oracle")), allow(unused_mut))]
        let mut id_accum: Vec<u32> = Vec::new();
        for (_sid, mut stage) in self.sheets.drain() {
            let t_sheet0 = Instant::now();
            let mut t_plan_ms = 0u128;
            let mut t_ensure_ms = 0u128;
            let mut t_assign_ms = 0u128;
            let mut t_edges_ms = 0u128;
            let mut t_ranges_ms = 0u128;
            let mut n_targets = 0usize;
            let mut n_globals = 0usize;
            let mut n_cell_deps = 0usize;
            let mut n_range_deps = 0usize;
            if dbg {
                eprintln!("[fz][ingest] sheet '{}' begin", stage.name);
            }
            // 1) Build plans for formulas on this sheet in chunks: staged
            // formulas, then records (planned here, one chunk at a time; the
            // shape memo carries across chunks).
            if !stage.formulas.is_empty() || !stage.records.is_empty() {
                let formula_batch_size: usize = std::env::var("FZ_INGEST_FORMULA_BATCH")
                    .ok()
                    .and_then(|s| s.parse().ok())
                    .filter(|&n| n > 0)
                    .unwrap_or(10_000);
                let mut batch_count = 0usize;
                let mut next_formula = 0usize;
                let mut next_record = 0usize;
                let mut memo_state = None;
                let records = std::mem::take(&mut stage.records);
                // First load: every formula target of the sheet gets its
                // vertex up front, column by column, so each column's
                // formulas are one id run (identity runs are the planner's
                // slices: chunk-sized runs fragment its plans). Load-time
                // family members become virtual runs right away.
                if self.g.first_load_assume_new() {
                    let targets: Vec<(u32, u32, Option<crate::engine::graph::FormulaRef>)> = stage
                        .formulas
                        .iter()
                        .map(|f| (f.row, f.col, None))
                        .chain(records.iter().map(|&(r, c, f)| (r, c, Some(f))))
                        .collect();
                    total_vertices += self.g.preallocate_load_targets(stage.id, targets);
                }

                loop {
                    let tp0 = Instant::now();
                    let (ast_ids, row_plans): (
                        Vec<crate::engine::graph::FormulaRef>,
                        Vec<(u32, u32, DependencyPlanRow)>,
                    ) = if next_formula < stage.formulas.len() {
                        let end = (next_formula + formula_batch_size).min(stage.formulas.len());
                        let chunk = &mut stage.formulas[next_formula..end];
                        next_formula = end;
                        let phase_span = crate::engine::trace::fz_span!(
                            tracing::Level::INFO,
                            "builder",
                            "builder.phase",
                            phase = "plan"
                        );
                        let mut prepared: Vec<
                            Option<(crate::engine::graph::FormulaRef, DependencyPlanRow)>,
                        > = (0..chunk.len()).map(|_| None).collect();
                        let mut pipeline_inputs = Vec::new();
                        for (idx, formula) in chunk.iter_mut().enumerate() {
                            match &mut formula.ast {
                                FormulaAstSource::Planned { formula, plan } => {
                                    prepared[idx] = Some((*formula, std::mem::take(plan)));
                                }
                                FormulaAstSource::Owned(ast) => {
                                    let placement = crate::reference::CellRef::new(
                                        stage.id,
                                        crate::reference::Coord::from_excel(
                                            formula.row,
                                            formula.col,
                                            true,
                                            true,
                                        ),
                                    );
                                    pipeline_inputs.push((
                                        idx,
                                        FormulaAstInput::Tree(ast.clone()),
                                        placement,
                                    ));
                                }
                                FormulaAstSource::Interned(ast_id) => {
                                    let placement = crate::reference::CellRef::new(
                                        stage.id,
                                        crate::reference::Coord::from_excel(
                                            formula.row,
                                            formula.col,
                                            true,
                                            true,
                                        ),
                                    );
                                    pipeline_inputs.push((
                                        idx,
                                        FormulaAstInput::RawArena(*ast_id),
                                        placement,
                                    ));
                                }
                            }
                        }
                        if !pipeline_inputs.is_empty() {
                            let indices: Vec<usize> =
                                pipeline_inputs.iter().map(|(idx, _, _)| *idx).collect();
                            let provider = RegistryFunctionProvider;
                            let ingested = {
                                let mut pipeline = self.g.ingest_pipeline(&provider);
                                let inputs = pipeline_inputs
                                    .into_iter()
                                    .map(|(_, input, placement)| (input, placement, None));
                                pipeline.ingest_batch(inputs)?
                            };
                            for (idx, ingested) in indices.into_iter().zip(ingested) {
                                prepared[idx] = Some((
                                    crate::engine::graph::FormulaRef::of_ingested(
                                        ingested.ast_id,
                                        ingested.member_anchor,
                                    ),
                                    ingested.dep_plan,
                                ));
                            }
                        }
                        let prepared: Vec<(crate::engine::graph::FormulaRef, DependencyPlanRow)> =
                            prepared
                                .into_iter()
                                .map(|entry| entry.expect("formula must be planned"))
                                .collect();
                        let ast_ids: Vec<crate::engine::graph::FormulaRef> =
                            prepared.iter().map(|(formula, _)| *formula).collect();
                        let row_plans: Vec<(u32, u32, DependencyPlanRow)> = chunk
                            .iter()
                            .zip(prepared.into_iter())
                            .map(|(formula, (_, plan))| (formula.row, formula.col, plan))
                            .collect();
                        (ast_ids, row_plans)
                    } else if next_record < records.len() {
                        let end = (next_record + formula_batch_size).min(records.len());
                        let chunk = &records[next_record..end];
                        next_record = end;
                        let mut pipeline = self
                            .g
                            .ingest_pipeline(records_provider)
                            .with_memo_state(memo_state.take());
                        let mut ids = Vec::with_capacity(chunk.len());
                        let mut row_plans = Vec::with_capacity(chunk.len());
                        for &(row, col, formula) in chunk {
                            let placement = crate::reference::CellRef::new(
                                stage.id,
                                crate::reference::Coord::from_excel(row, col, true, true),
                            );
                            let input = match formula {
                                crate::engine::graph::FormulaRef::Own(id) => {
                                    FormulaAstInput::RawArena(id)
                                }
                                crate::engine::graph::FormulaRef::Member { template, anchor } => {
                                    FormulaAstInput::Member { template, anchor }
                                }
                            };
                            let ingested = pipeline.ingest_formula(input, placement, None)?;
                            ids.push(crate::engine::graph::FormulaRef::of_ingested(
                                ingested.ast_id,
                                ingested.member_anchor,
                            ));
                            row_plans.push((row, col, ingested.dep_plan));
                        }
                        memo_state = Some(pipeline.take_memo_state());
                        (ids, row_plans)
                    } else {
                        break;
                    };
                    batch_count += 1;
                    let phase_span = crate::engine::trace::fz_span!(
                        tracing::Level::INFO,
                        "builder",
                        "builder.phase",
                        phase = "plan"
                    );
                    let plan = dependency_plan_from_rows(self.g.sheet_reg(), stage.id, &row_plans);
                    #[cfg(any(test, feature = "legacy_oracle"))]
                    edges_adj.reserve(plan.formula_targets.len());
                    t_plan_ms += tp0.elapsed().as_millis();
                    drop(phase_span);
                    n_targets += plan.formula_targets.len();
                    n_globals += plan.global_cells.len();

                    // Reserve capacity hints before large ensure / hash-map growth.
                    self.g.reserve_cells(plan.vertex_pool.len());

                    // Ensure targets and referenced cells exist using batch allocation when missing.
                    let te0 = Instant::now();
                    let phase_span = crate::engine::trace::fz_span!(
                        tracing::Level::INFO,
                        "builder",
                        "builder.phase",
                        phase = "ensure"
                    );
                    // First load with compression: family members' new
                    // vertices skip the per-cell maps; the chunk's members
                    // are installed as virtual runs after assignment.
                    let mut unmapped = vec![false; plan.vertex_pool_packed.len()];
                    if self.g.first_load_assume_new() && self.g.formula_compression_enabled() {
                        for (i, &pos) in plan.formula_target_pool_indices.iter().enumerate() {
                            let rp = &row_plans[i].2;
                            if matches!(ast_ids[i], crate::engine::graph::FormulaRef::Member { .. })
                                && !rp.volatile
                                && !rp.dynamic
                            {
                                unmapped[pos as usize] = true;
                            }
                        }
                    }
                    // Targets fill the front of the pool; referenced cells
                    // after them get no vertex unless they have one
                    // (decision 27: references create none).
                    let n_targets = plan
                        .formula_target_pool_indices
                        .iter()
                        .max()
                        .map_or(0, |&m| m as usize + 1);
                    // A target whose formula was replaced by a value takes
                    // its retired id back (decision 27).
                    if self.g.retired_id_count() != 0 {
                        for &(sid, pc) in &plan.vertex_pool[..n_targets] {
                            self.g.revive_retired_id(&crate::reference::CellRef::new(
                                sid,
                                crate::reference::Coord::new(pc.row(), pc.col(), true, true),
                            ));
                        }
                    }
                    let (target_pool_vids, add_batch) =
                        self.g.ensure_vertices_batch_packed_ordered_unmapped(
                            &plan.vertex_pool_packed[..n_targets],
                            &mut unmapped[..n_targets],
                        );
                    let all_vids: Vec<Option<VertexId>> = target_pool_vids
                        .into_iter()
                        .map(Some)
                        .chain(plan.vertex_pool[n_targets..].iter().map(|&(sid, pc)| {
                            let v = self.g.dep_vertex(&crate::reference::CellRef::new(
                                sid,
                                crate::reference::Coord::new(pc.row(), pc.col(), true, true),
                            ));
                            if v.is_none() {
                                // Legacy's placeholder: part of the used extent.
                                self.g.note_extent_cell(sid, pc.row(), pc.col());
                            }
                            v
                        }))
                        .collect();
                    total_vertices += add_batch.len();
                    #[cfg(any(test, feature = "legacy_oracle"))]
                    if !add_batch.is_empty() {
                        for (pc, id) in &add_batch {
                            coord_accum.push(*pc);
                            id_accum.push(*id);
                        }
                    }
                    t_ensure_ms += te0.elapsed().as_millis();
                    drop(phase_span);

                    // Assign formula vertices using the canonical AST ids and flags produced by the pipeline.
                    let ta0 = Instant::now();
                    let phase_span = crate::engine::trace::fz_span!(
                        tracing::Level::INFO,
                        "builder",
                        "builder.phase",
                        phase = "assign"
                    );
                    self.g.reserve_formula_metadata(plan.formula_targets.len());

                    let mut dep_vids: Vec<Option<VertexId>> =
                        Vec::with_capacity(plan.global_cells.len());
                    for &pos in &plan.global_cell_pool_indices {
                        dep_vids.push(all_vids[pos as usize]);
                    }

                    let mut target_vids: Vec<VertexId> =
                        Vec::with_capacity(plan.formula_targets.len());
                    let load_fast = self.g.first_load_assume_new();
                    let mut members = Vec::new();
                    for (i, &pos) in plan.formula_target_pool_indices.iter().enumerate() {
                        let vid = all_vids[pos as usize].expect("formula targets have vertices");
                        target_vids.push(vid);
                        let row_plan = &row_plans[i].2;
                        if self.g.is_virtual_member(vid) {
                            // Pre-allocated as a virtual member: keep it
                            // unless it needs per-vertex state.
                            self.g.assign_preallocated_member(
                                vid,
                                ast_ids[i],
                                row_plan.volatile,
                                row_plan.dynamic,
                            );
                        } else if unmapped[pos as usize] {
                            self.g.assign_unmapped_member_load_fast(vid);
                            let (sheet, pc) = plan.formula_targets[i];
                            members.push((vid, sheet, pc.row(), pc.col(), ast_ids[i]));
                        } else if load_fast {
                            self.g.assign_formula_ref_load_fast(
                                vid,
                                ast_ids[i],
                                row_plan.volatile,
                                row_plan.dynamic,
                            );
                        } else {
                            self.g.assign_formula_ref(
                                vid,
                                ast_ids[i],
                                row_plan.volatile,
                                row_plan.dynamic,
                            );
                        }
                    }
                    self.g.install_load_members(members);
                    self.g.mark_vertices_dirty_batch(&target_vids);
                    if had_vertices {
                        dirty_roots.extend_from_slice(&target_vids);
                    }
                    total_formulas += target_vids.len();
                    t_assign_ms += ta0.elapsed().as_millis();
                    drop(phase_span);

                    // Collect edges into adjacency rows for a later one-shot CSR build.
                    let ted0 = Instant::now();
                    let phase_span = crate::engine::trace::fz_span!(
                        tracing::Level::INFO,
                        "builder",
                        "builder.phase",
                        phase = "edges"
                    );
                    let ranges_phase_span = crate::engine::trace::fz_span!(
                        tracing::Level::INFO,
                        "builder",
                        "builder.phase",
                        phase = "ranges"
                    );
                    for (fi, &tvid) in target_vids.iter().enumerate() {
                        let mut row: smallvec::SmallVec<[u32; 8]> = smallvec::SmallVec::new();
                        if let Some(indices) = plan.per_formula_cells.get(fi) {
                            let mut dep_count = 0usize;
                            row.reserve(indices.len());
                            let mut vertexless = Vec::new();
                            for &idx in indices {
                                match dep_vids[idx as usize] {
                                    Some(dep_vid) => row.push(dep_vid.0),
                                    None => {
                                        let (sid, pc) = plan.global_cells[idx as usize];
                                        vertexless.push((sid, pc.row(), pc.col()));
                                    }
                                }
                                dep_count += 1;
                            }
                            self.g.note_vertexless_deps(tvid, vertexless);
                            total_edges += dep_count;
                            n_cell_deps += dep_count;
                        }

                        let tr0 = Instant::now();
                        if let Some(rks) = plan.per_formula_ranges.get(fi) {
                            n_range_deps += rks.len();
                            self.g.add_range_deps_from_keys(tvid, rks, stage.id);
                        }
                        t_ranges_ms += tr0.elapsed().as_millis();
                        if let Some(names) = plan.per_formula_names.get(fi)
                            && !names.is_empty()
                        {
                            let mut name_vertices = Vec::new();
                            let (formula_sheet, _) = plan
                                .formula_targets
                                .get(fi)
                                .copied()
                                .unwrap_or((stage.id, AbsCoord::new(1, 1)));
                            for name in names {
                                if let Some(named) = self.g.resolve_name_entry(name, formula_sheet)
                                {
                                    row.push(named.vertex.0);
                                    name_vertices.push(named.vertex);
                                } else if let Some(source) =
                                    self.g.resolve_source_scalar_entry(name)
                                {
                                    row.push(source.vertex.0);
                                } else {
                                    self.g
                                        .record_pending_name_reference(formula_sheet, name, tvid);
                                }
                            }
                            if !name_vertices.is_empty() {
                                self.g.attach_vertex_to_names(tvid, &name_vertices);
                            }
                        }

                        if let Some(tables) = plan.per_formula_tables.get(fi)
                            && !tables.is_empty()
                        {
                            for table_name in tables {
                                if let Some(table) = self.g.resolve_table_entry(table_name) {
                                    row.push(table.vertex.0);
                                } else if let Some(source) =
                                    self.g.resolve_source_table_entry(table_name)
                                {
                                    row.push(source.vertex.0);
                                }
                            }
                        }
                        #[cfg(any(test, feature = "legacy_oracle"))]
                        edges_adj.push((tvid.0, row.into_vec()));
                        // Only oracle builds keep the CSR; others need just
                        // the per-vertex direct-edge count.
                        #[cfg(not(any(test, feature = "legacy_oracle")))]
                        self.g.note_dep_edges(tvid, row.len());
                    }
                    drop(ranges_phase_span);
                    drop(phase_span);
                    t_edges_ms += ted0.elapsed().as_millis();
                }

                if dbg && batch_count > 1 {
                    eprintln!(
                        "[fz][ingest] sheet '{}' processed in {} formula batches (batch_size={})",
                        stage.name, batch_count, formula_batch_size
                    );
                }
            }
            if dbg {
                eprintln!(
                    "[fz][ingest] sheet '{}' done: plan={}ms ensure={}ms assign={}ms edges={}ms ranges={}ms targets={} globals={} cell_deps={} range_groups={} total={}ms",
                    stage.name,
                    t_plan_ms,
                    t_ensure_ms,
                    t_assign_ms,
                    t_edges_ms,
                    t_ranges_ms,
                    n_targets,
                    n_globals,
                    n_cell_deps,
                    n_range_deps,
                    t_sheet0.elapsed().as_millis()
                );
            }
        }
        if dbg {
            eprintln!("[fz][ingest] beginning finalize");
        }

        // Admission's direct-edge count (legacy's CSR edges) is kept in
        // every build; the CSR itself only in oracle builds.
        #[cfg(not(any(test, feature = "legacy_oracle")))]
        let _ = (&edges_adj, &coord_accum, &id_accum);
        // Finalize: pick strategy based on graph size and number of edge rows
        #[cfg(any(test, feature = "legacy_oracle"))]
        if !edges_adj.is_empty() {
            let _csr_phase_span = crate::engine::trace::fz_span!(
                tracing::Level::INFO,
                "builder",
                "builder.phase",
                phase = "csr"
            );
            let rows = edges_adj.len();
            let total_vertices_now = self.g.vertex_count();
            let t_fin0 = Instant::now();
            if dbg {
                eprintln!(
                    "[fz][ingest] finalize: start rows={rows}, vertices={total_vertices_now}"
                );
            }
            // Heuristic: avoid one-shot CSR when vertices are huge and rows are sparse
            let sparse_vs_huge =
                total_vertices_now > 800_000 && (rows as f64) / (total_vertices_now as f64) < 0.05;
            if sparse_vs_huge {
                let t_delta0 = Instant::now();
                if dbg {
                    eprintln!("[fz][ingest] finalize: using delta path (begin)");
                }
                self.g.begin_batch();
                for (tvid_raw, row) in &edges_adj {
                    let tvid = crate::engine::vertex::VertexId(*tvid_raw);
                    if !row.is_empty() {
                        let deps: Vec<crate::engine::vertex::VertexId> = row
                            .iter()
                            .map(|d| crate::engine::vertex::VertexId(*d))
                            .collect();
                        self.g.add_edges_nobatch(tvid, &deps);
                    }
                }
                self.g.end_batch();
                if dbg {
                    eprintln!(
                        "[fz][ingest] finalize: delta done in {} ms (total {} ms)",
                        t_delta0.elapsed().as_millis(),
                        t_fin0.elapsed().as_millis()
                    );
                }
            } else {
                // One-shot CSR build from accumulated adjacency and coords/ids
                let mut t_coords_ms = 0u128;
                // Allocation batches contain only vertices created by this ingest,
                // not necessarily the complete graph (including symbol vertices).
                // Keep the complete initial-load fast path, but never install partial
                // membership: later rebuilds use it to carry untouched edges forward.
                // vertex_count includes deleted slots, which can only cause a safe
                // extra collection here; each accumulated id is newly allocated once.
                if id_accum.len() != total_vertices_now {
                    coord_accum.clear();
                    id_accum.clear();
                    if dbg {
                        eprintln!("[fz][ingest] finalize: gathering coords/ids");
                    }
                    let t_coords0 = Instant::now();
                    for vid in self.g.iter_vertex_ids() {
                        coord_accum.push(self.g.vertex_addr(vid));
                        id_accum.push(vid.0);
                    }
                    t_coords_ms = t_coords0.elapsed().as_millis();
                }
                if dbg {
                    eprintln!("[fz][ingest] finalize: building CSR");
                }
                let t_csr0 = Instant::now();
                for (tvid_raw, row) in &edges_adj {
                    self.g
                        .note_dep_edges(crate::engine::vertex::VertexId(*tvid_raw), row.len());
                }
                self.g
                    .build_edges_from_adjacency(edges_adj, coord_accum, id_accum);
                if dbg {
                    eprintln!(
                        "[fz][ingest] finalize: rows={}, gather_coords={} ms, csr_build={} ms, total={} ms",
                        rows,
                        t_coords_ms,
                        t_csr0.elapsed().as_millis(),
                        t_fin0.elapsed().as_millis()
                    );
                }
            }
        }

        // Replacing a formula also invalidates its existing consumers. Do this
        // once, after all new edges are installed, rather than one BFS per row.
        // On a complete first load every formula is already dirty.
        if !dirty_roots.is_empty() {
            self.g.mark_dirty_many(&dirty_roots);
        } else if !had_vertices && !self.g.first_load_assume_new() {
            // Nothing to propagate (no earlier formulas; value cells have
            // no vertex), but read-only planning needs the authority in
            // step with the new formulas, as the propagation's sync left it.
            self.g.authority_sync();
        }

        // Restore config
        self.g.set_sheet_index_mode(self.cfg_saved.sheet_index_mode);
        // Load-time members are complete (formulas, flags, edge counts):
        // their full vertex pages keep no rows.
        self.g.virtualize_member_pages();
        crate::engine::trace::fz_event!(
            tracing::Level::INFO,
            "builder",
            "builder.summary",
            vertices = total_vertices,
            edges = total_edges,
            formulas = total_formulas
        );
        Ok(BulkIngestSummary {
            sheets: 0, // could populate later
            vertices: total_vertices,
            formulas: total_formulas,
            edges: total_edges,
            elapsed: t0.elapsed(),
        })
    }
}