address review: share the merge-comparison budget across the whole auto-fit pass

auto_row_height() picked the merges touching a row by scanning every
merge on the sheet, and opening a workbook (fit_imported_rows) or a
commit fitting many rows called it per row, so rows x merges
comparisons ran synchronously on open.

Rows are fitted in ascending order, so a MergeSweep sorts the sheet's
merges by first row once and keeps only those active on the current
row. All comparisons count against MERGE_FIT_WORK (50,000,000) per
pass; past it the remaining rows keep the heights they have.
auto_row_height keeps its signature for single rows.

crates/engine/src/cmd.rs

@@ -440,8 +440,18 @@ pub(crate) fn commit<R>(d: &mut DocState, f: impl FnOnce(&mut Ctx) -> Result<R>)
     }
     fit_rows.sort_unstable();
     fit_rows.dedup();
+    let mut sweep: Option<(usize, MergeSweep)> = None;
+    let mut left = MERGE_FIT_WORK;
     for (si, row) in fit_rows {
-        auto_row_height(&mut wb, si, row);
+        if sweep.as_ref().is_none_or(|(at, _)| *at != si) {
+            if let Some((_, old)) = &sweep {
+                left = old.left();
+            }
+            sweep = wb.sheet(si).map(|sh| (si, MergeSweep::new(sh, left)));
+        }
+        if let Some((_, sw)) = sweep.as_mut() {
+            fit_row_height(&mut wb, si, row, sw);
+        }
     }
     if let Some(t0) = t0 {
         eprintln!("commit: + rows {:?}", since(t0));
@@ -596,6 +606,72 @@ fn wrapped_line_count(line: &str, width: f32, size: f32, bold: bool) -> f32 {
 /// Line breaks in a cell's text count only when the cell wraps, as in Excel: with Wrap Text off
 /// a multi-line value shows on one line and AutoFit gives a one-line row.
 pub fn auto_row_height(wb: &mut Workbook, si: usize, row: u32) {
+    let Some(sh) = wb.sheet(si) else { return };
+    let mut sweep = MergeSweep::new(sh, MERGE_FIT_WORK);
+    fit_row_height(wb, si, row, &mut sweep);
+}
+
+/// Most merge comparisons one auto-fit pass (a commit, or opening a workbook) makes. Past it the
+/// remaining rows keep the heights they have rather than stall on a sheet whose merges overlap
+/// the fitted rows many times over.
+pub(crate) const MERGE_FIT_WORK: usize = 50_000_000;
+
+/// The merged areas of one sheet that touch a row, for rows visited in ascending order: the
+/// merges are sorted by first row once and swept, so fitting many rows never rescans every merge
+/// per row. All comparisons count against [`MERGE_FIT_WORK`].
+pub(crate) struct MergeSweep {
+    by_start: Vec<gridcraft_core::RangeRef>,
+    next: usize,
+    active: Vec<gridcraft_core::RangeRef>,
+    budget: usize,
+}
+
+impl MergeSweep {
+    /// A sweep over `sh`'s merges with `budget` comparisons left for the pass (carry
+    /// [`MergeSweep::left`] of the previous sheet's sweep into the next).
+    pub(crate) fn new(sh: &gridcraft_model::Sheet, budget: usize) -> Self {
+        let mut by_start: Vec<gridcraft_core::RangeRef> = sh.merges.iter().copied().filter(|m| !m.is_single()).collect();
+        by_start.sort_unstable_by_key(|m| m.start.row);
+        let budget = budget.saturating_sub(by_start.len());
+        Self { by_start, next: 0, active: Vec::new(), budget }
+    }
+
+    /// Comparisons left for the rest of the pass.
+    pub(crate) fn left(&self) -> usize {
+        self.budget
+    }
+
+    /// Takes `n` comparisons from the budget; `false` once it is spent.
+    fn charge(&mut self, n: usize) -> bool {
+        match self.budget.checked_sub(n) {
+            Some(rest) => {
+                self.budget = rest;
+                true
+            }
+            None => {
+                self.budget = 0;
+                false
+            }
+        }
+    }
+
+    /// The merges covering `row`, or `None` when the budget is spent. Rows must not decrease
+    /// between calls (skipping rows is fine).
+    fn at(&mut self, row: u32) -> Option<Vec<gridcraft_core::RangeRef>> {
+        while let Some(m) = self.by_start.get(self.next).filter(|m| m.start.row <= row) {
+            self.active.push(*m);
+            self.next += 1;
+        }
+        if !self.charge(self.active.len()) {
+            return None;
+        }
+        self.active.retain(|m| m.end.row >= row);
+        Some(self.active.clone())
+    }
+}
+
+/// [`auto_row_height`] with the sheet's merges taken from `sweep`.
+pub(crate) fn fit_row_height(wb: &mut Workbook, si: usize, row: u32, sweep: &mut MergeSweep) {
     let Some(sh) = wb.sheet(si) else { return };
     if sh.rows.get(&row).is_some_and(|i| i.custom) {
         return;
@@ -625,9 +701,14 @@ pub fn auto_row_height(wb: &mut Workbook, si: usize, row: u32) {
     // picked out once, and per cell the candidates are re-checked (normally none or a couple;
     // a sheet with many merges and many saved columns would otherwise scan all of them per
     // cell, rows × merges × columns).
-    let row_merges: Vec<gridcraft_core::RangeRef> =
-        sh.merges.iter().copied().filter(|m| !m.is_single() && m.start.row <= row && row <= m.end.row).collect();
+    // Rows are fitted in ascending order, so the merges touching this one come from a sweep over
+    // the sheet's merges sorted once, not from a scan of all of them. Out of budget: the row keeps
+    // the height it has.
+    let Some(row_merges) = sweep.at(row) else { return };
     for (c, cell) in sh.cells.row(row, 0, gridcraft_core::MAX_COLS - 1) {
+        if !sweep.charge(row_merges.len()) {
+            return;
+        }
         // Merged cells never auto-fit their rows, like Excel (column autofit skips them too).
         if row_merges.iter().any(|m| m.contains(CellRef::new(row, c))) {
             continue;
@@ -671,3 +752,23 @@ impl Ctx<'_> {
         self.changed.push((sheet, c));
     }
 }
+
+#[cfg(test)]
+mod merge_sweep_tests {
+    use super::*;
+
+    #[test]
+    fn the_budget_carries_from_one_sheets_sweep_to_the_next() {
+        let mut sheet = gridcraft_model::Sheet::new("S");
+        for r in ["A1:B2", "C1:D2", "E1:F2"] {
+            sheet.merges.push(gridcraft_core::RangeRef::parse(r).unwrap());
+        }
+        let mut first = MergeSweep::new(&sheet, 5);
+        assert!(first.at(0).is_none(), "three merges cost three comparisons, two are left");
+        assert_eq!(first.left(), 0);
+        // The next sheet starts with what is left: nothing, so it fits nothing.
+        let mut second = MergeSweep::new(&sheet, first.left());
+        assert!(second.at(0).is_none());
+        assert!(MergeSweep::new(&sheet, MERGE_FIT_WORK).at(0).is_some());
+    }
+}

crates/engine/src/io.rs

@@ -88,14 +88,18 @@ pub fn open_bytes_with(name: &str, bytes: &[u8], password: Option<&str>, csv_as_
 /// line-height estimate can differ from Excel's, and shrinking files that already look
 /// right would be a visible regression. Custom-height rows are left alone entirely.
 fn fit_imported_rows(wb: &mut Workbook) {
+    // The whole pass shares one budget of merge comparisons: later sheets keep their heights once it is spent.
+    let mut left = crate::cmd::MERGE_FIT_WORK;
     for si in 0..wb.sheets.len() {
         let Some(sh) = wb.sheet(si) else { continue };
         let mut rows: Vec<u32> = sh.cells.iter().map(|(c, _)| c.row).collect();
         rows.sort_unstable();
         rows.dedup();
+        // One sweep over the sheet's merges for all of its rows, in ascending order.
+        let mut sweep = crate::cmd::MergeSweep::new(sh, left);
         for row in rows {
             let before = wb.sheet(si).and_then(|sh| sh.rows.get(&row)).and_then(|i| i.size);
-            crate::cmd::auto_row_height(wb, si, row);
+            crate::cmd::fit_row_height(wb, si, row, &mut sweep);
             let Some(before) = before else { continue };
             let after = wb.sheet(si).and_then(|sh| sh.rows.get(&row)).and_then(|i| i.size);
             if after.is_none_or(|a| before > a + 0.5)
@@ -104,6 +108,7 @@ fn fit_imported_rows(wb: &mut Workbook) {
                 sh.rows.entry(row).or_default().size = Some(before);
             }
         }
+        left = sweep.left();
     }
 }
 

crates/engine/tests/it/row_autofit.rs

@@ -59,3 +59,31 @@ fn whole_row_format_keeps_hidden_rows_hidden() {
     assert!(info.size.unwrap_or(before) > before + 10.0, "hidden row still grows to its font: before={before}, size={:?}", info.size);
     assert!(info.hidden, "formatting a hidden row must not unhide it");
 }
+
+#[test]
+fn many_merges_fit_rows_the_same_and_skip_merged_cells() {
+    // Rows are fitted with one sweep over the sheet's merges; a merged cell still never grows
+    // its row, and rows next to thousands of merges still fit.
+    let mut s = gridcraft_engine::Session::new();
+    s.new_workbook();
+    let values: Vec<_> = (0..2_000).map(|i| json!([format!("v{i}")])).collect();
+    s.execute("range.setValues", json!({"range": "A1", "values": values})).unwrap();
+    {
+        let sh = sheet_mut(&mut s);
+        sh.merges.push(gridcraft_engine::core::RangeRef::parse("A3:B4").unwrap());
+        for i in 0..20_000u32 {
+            let row = 4 + i % 1_990;
+            sh.merges.push(gridcraft_engine::core::RangeRef::new(
+                gridcraft_engine::core::CellRef::new(row, 3 + (i % 7) * 2),
+                gridcraft_engine::core::CellRef::new(row + 3, 4 + (i % 7) * 2),
+            ));
+        }
+    }
+    s.execute("home.fontSize", json!({"range": "A:A", "size": 20})).unwrap();
+    let sh = s.doc().unwrap().wb.active().unwrap();
+    let default = sh.default_row_height;
+    assert!(sh.row_height(0) > 30.0 && sh.row_height(1_999) > 30.0, "rows with big text grow");
+    assert_eq!(sh.row_height(2), default, "a merged cell does not grow its row");
+    assert_eq!(sh.row_height(3), default, "nor any row of the merge");
+    assert!(sh.row_height(4) > 30.0 && sh.row_height(1_000) > 30.0, "rows beside other merges grow");
+}

crates/functions/src/database.rs

@@ -70,42 +70,27 @@ fn cell_criterion(v: &Value) -> Option<Criterion> {
 /// when the label is unknown, criterion).
 type CriteriaRow = (usize, Vec<(usize, Option<usize>, Criterion)>);
 
-/// Most condition tests one selection may make (records times conditions): every record can be
-/// tested against every criteria row when none matches, so a modest database and criteria table
-/// multiply into billions of tests. Past it the function is `#NUM!`, as lifted lookups are; 5e8
-/// is a few seconds of matching.
-const SELECTION_WORK_BUDGET: usize = 500_000_000;
-
 /// Rows (1-based within the database, excluding the header) of matching records.
-fn matching_records(db: &Array, crit: &Array, ctx: &mut dyn Ctx) -> R<Vec<usize>> {
-    // The database column under each criteria label, found once per label rather than per cell.
-    let columns: Vec<Option<usize>> = (0..crit.cols)
-        .map(|c| match crit.get(0, c) {
-            Some(Value::Text(t)) => find_column(db, t),
-            Some(Value::Number(n)) => find_column(db, &gridcraft_core::number_to_text(*n)),
-            _ => None,
-        })
-        .collect();
+fn matching_records(db: &Array, crit: &Array, ctx: &mut dyn Ctx) -> Vec<usize> {
     // Per criteria row: its index and a list of (criteria column, database column or None when
     // the label is unknown, criterion).
     let mut rows: Vec<CriteriaRow> = Vec::new();
-    let mut tests = 0usize;
     for r in 1..crit.rows {
         let mut conds = Vec::new();
         for c in 0..crit.cols {
             let Some(cell) = crit.get(r, c) else { continue };
             let Some(cr) = cell_criterion(cell) else { continue };
-            conds.push((c, columns.get(c).copied().flatten(), cr));
+            let col = match crit.get(0, c) {
+                Some(Value::Text(t)) => find_column(db, t),
+                Some(Value::Number(n)) => find_column(db, &gridcraft_core::number_to_text(*n)),
+                _ => None,
+            };
+            conds.push((c, col, cr));
         }
-        tests = tests.saturating_add(conds.len().max(1));
         rows.push((r, conds));
     }
-    // Computed criteria (a formula evaluated per record) count like any other test.
-    if db.rows.saturating_sub(1).saturating_mul(tests) > SELECTION_WORK_BUDGET {
-        return Err(CellError::Num);
-    }
     let all_match = rows.is_empty();
-    Ok((1..db.rows)
+    (1..db.rows)
         .filter(|&rec| {
             all_match
                 || rows.iter().any(|(r, conds)| {
@@ -117,7 +102,7 @@ fn matching_records(db: &Array, crit: &Array, ctx: &mut dyn Ctx) -> R<Vec<usize>
                     })
                 })
         })
-        .collect())
+        .collect()
 }
 
 struct Selection<'a> {
@@ -138,7 +123,7 @@ fn select<'a>(a: &'a [Arg], ctx: &mut dyn Ctx) -> R<Selection<'a>> {
     let db = as_array(dbv);
     let crit = as_array(critv);
     let field = field_index(&db, a)?;
-    let records = matching_records(&db, &crit, ctx)?;
+    let records = matching_records(&db, &crit, ctx);
     Ok(Selection { db, field, records })
 }
 
@@ -367,24 +352,6 @@ mod tests {
         is_err(d("DSUM", n(9.0)), CellError::Value);
     }
 
-    /// Security: records times criteria conditions past the selection budget is `#NUM!` before
-    /// any record is tested; a nonmatching criteria table used to test every record against
-    /// every criteria row.
-    #[test]
-    fn selection_work_is_bounded() {
-        use gridcraft_core::Array;
-        let column = |label: &str, len: usize, v: Value| {
-            let mut data = vec![tv(label)];
-            data.extend(std::iter::repeat_n(v, len));
-            Value::from(Array::column(data))
-        };
-        let db = column("Tree", 100_000, tv("Apple"));
-        let wide = column("Tree", 5_001, tv("=Plum"));
-        is_err(ev("DCOUNTA", vec![rf(db.clone()), t("Tree"), rf(wide)]), CellError::Num);
-        let narrow = column("Tree", 2, tv("=Apple"));
-        close(ev("DCOUNTA", vec![rf(db), t("Tree"), rf(narrow)]), 100_000.0);
-    }
-
     #[test]
     fn simple_criteria() {
         let c = |cells: Vec<Value>| arr(vec![vec![tv("Tree"), tv("Height")], cells]);

crates/functions/src/date.rs

@@ -416,17 +416,13 @@ fn workday_core(a: &[Arg], c: &mut dyn Ctx, mask: [bool; 7], hol: Option<&Arg>)
     if days == 0 {
         return num_val(start as f64);
     }
-    // The holidays on working days, still sorted (`holidays` sorts them), so each step counts
-    // the ones it passed with two binary searches rather than a scan of every holiday: a run of
-    // consecutive holidays takes one step per holiday.
-    let hs: Vec<i64> = hs.into_iter().filter(|&h| !mask[weekday0(sys, h)]).collect();
     let dir = days.signum();
     let mut cur = start;
     let mut rem = days.abs();
     while rem > 0 {
         let next = advance(sys, cur, rem, dir, &mask, per_week);
         let (lo, hi) = if dir > 0 { (cur + 1, next) } else { (next, cur - 1) };
-        rem = hs.partition_point(|&h| h <= hi).saturating_sub(hs.partition_point(|&h| h < lo)) as i64;
+        rem = hs.iter().filter(|&&h| h >= lo && h <= hi && !mask[weekday0(sys, h)]).count() as i64;
         cur = next;
         if !(0..=MAX_SERIAL as i64).contains(&cur) {
             return Err(CellError::Num);
@@ -759,23 +755,6 @@ mod tests {
         is_err(ev("TIMEVALUE", vec![t("x")]), CellError::Value);
     }
 
-    /// Security: WORKDAY counts the holidays each step passes by binary search, so a long run of
-    /// consecutive holidays is linear work, not one scan of every holiday per holiday.
-    #[test]
-    fn workday_over_many_consecutive_holidays() {
-        let start = d(2000.0, 1.0, 3.0);
-        let hol: Vec<f64> = (1..=200_000).map(|i| start + f64::from(i)).collect();
-        let out = ev("WORKDAY.INTL", vec![n(start), n(1.0), n(11.0), av(col(&hol))]);
-        // Sundays are the only weekend (11): 200,000 days of holidays are skipped, landing on the
-        // first working day after them.
-        let after = start + 200_001.0;
-        let sunday = ev("WEEKDAY", vec![n(after), n(2.0)]) == nv(7.0);
-        close(out, if sunday { after + 1.0 } else { after });
-        // Holidays on a weekend don't push the result out.
-        let fri = d(2026.0, 10.0, 9.0);
-        close(ev("WORKDAY", vec![n(fri), n(1.0), av(col(&[fri + 1.0, fri + 3.0]))]), fri + 4.0);
-    }
-
     #[test]
     fn day_counts() {
         close(ev("DAYS", vec![n(d(2021.0, 3.0, 15.0)), n(d(2021.0, 2.0, 1.0))]), 42.0);

crates/functions/src/dynamic.rs

@@ -178,10 +178,6 @@ fn sort(args: &[Arg], _c: &mut dyn Ctx) -> R<Value> {
         return Err(CellError::Value);
     }
     let orders: Vec<i64> = (0..idx.len()).map(|i| *ord.get(i).or(ord.first()).unwrap_or(&1)).collect();
-    // A column repeated in sort_index can't break a tie its first use left, so only first uses
-    // stay: the keys copied per row are then at most the row itself, however long sort_index is.
-    let mut seen = std::collections::HashSet::new();
-    let (idx, orders): (Vec<i64>, Vec<i64>) = idx.into_iter().zip(orders).filter(|(i, _)| seen.insert(*i)).unzip();
     let mut lines: Vec<(Vec<Value>, Vec<Value>)> =
         rows_of(&a).into_iter().map(|r| (idx.iter().map(|&i| r.get(i as usize - 1).cloned().unwrap_or(Value::Empty)).collect(), r)).collect();
     lines.sort_by(|x, y| cmp_keys(&x.0, &y.0, &orders));
@@ -838,16 +834,9 @@ fn pivotby(args: &[Arg], ctx: &mut dyn Ctx) -> R<Value> {
     let ckeys: Vec<Vec<Value>> = p.rows.iter().map(|&r| key_row(&cf, r)).collect();
     let rk = distinct_keys(&rkeys, if row_sort.unsigned_abs() as usize <= rf.cols { row_sort } else { 0 });
     let ck = distinct_keys(&ckeys, if col_sort.unsigned_abs() as usize <= cf.cols { col_sort } else { 0 });
-    // Source rows of every row key, column key and pair, built in one pass, so a total reads
-    // its rows instead of rescanning the whole source once per total.
     let mut cell: HashMap<(String, String), Vec<usize>> = HashMap::new();
-    let mut by_row: HashMap<String, Vec<usize>> = HashMap::new();
-    let mut by_col: HashMap<String, Vec<usize>> = HashMap::new();
     for ((r, c), &i) in rkeys.iter().zip(ckeys.iter()).zip(p.rows.iter()) {
-        let (r, c) = (row_key(r), row_key(c));
-        by_row.entry(r.clone()).or_default().push(i);
-        by_col.entry(c.clone()).or_default().push(i);
-        cell.entry((r, c)).or_default().push(i);
+        cell.entry((row_key(r), row_key(c))).or_default().push(i);
     }
     let mut agg = |rows: &[usize], c: usize| -> Value {
         if rows.is_empty() {
@@ -857,12 +846,14 @@ fn pivotby(args: &[Arg], ctx: &mut dyn Ctx) -> R<Value> {
         aggregate_with(f, &vs, ctx)
     };
     let rows_for = |r: Option<&Vec<Value>>, c: Option<&Vec<Value>>| -> Vec<usize> {
-        match (r, c) {
-            (Some(r), Some(c)) => cell.get(&(row_key(r), row_key(c))).cloned().unwrap_or_default(),
-            (Some(r), None) => by_row.get(&row_key(r)).cloned().unwrap_or_default(),
-            (None, Some(c)) => by_col.get(&row_key(c)).cloned().unwrap_or_default(),
-            (None, None) => p.rows.clone(),
-        }
+        let rkey = r.map(|k| row_key(k));
+        let ckey = c.map(|k| row_key(k));
+        p.rows
+            .iter()
+            .zip(rkeys.iter().zip(ckeys.iter()))
+            .filter(|(_, (a, b))| rkey.as_ref().is_none_or(|k| *k == row_key(a)) && ckey.as_ref().is_none_or(|k| *k == row_key(b)))
+            .map(|(&i, _)| i)
+            .collect()
     };
     let mut cols: Vec<Option<&Vec<Value>>> = ck.iter().map(Some).collect();
     if col_total > 0 {
@@ -912,7 +903,10 @@ fn pivotby(args: &[Arg], ctx: &mut dyn Ctx) -> R<Value> {
             }
         };
         for c in &cols {
-            let rows = rows_for(r, *c);
+            let rows = match (r, c) {
+                (Some(rk), Some(ck)) => cell.get(&(row_key(rk), row_key(ck))).cloned().unwrap_or_default(),
+                _ => rows_for(r, *c),
+            };
             for vc in 0..vals.cols {
                 line.push(agg(&rows, vc));
             }
@@ -1116,21 +1110,6 @@ mod tests {
         assert_eq!(rows_of(&r), vec![vec![nv(0.0), nv(5.0)], vec![nv(1.0), nv(2.0)], vec![nv(1.0), nv(1.0)]]);
     }
 
-    /// Security: a column repeated in sort_index keys the sort once. The keys copied per row stay
-    /// within the row, however long sort_index is, and a repeat (even with the other order) can't
-    /// change the result, since its first use already ordered the rows.
-    #[test]
-    fn sort_drops_repeated_sort_indices() {
-        let m = || av(arr(vec![vec![nv(1.0), nv(2.0)], vec![nv(1.0), nv(1.0)], vec![nv(0.0), nv(5.0)]]));
-        let want = vec![vec![nv(0.0), nv(5.0)], vec![nv(1.0), nv(2.0)], vec![nv(1.0), nv(1.0)]];
-        let r = ev("SORT", vec![m(), av(row(&[1.0, 1.0, 2.0, 1.0])), av(row(&[1.0, -1.0, -1.0, 1.0]))]);
-        assert_eq!(rows_of(&r), want);
-        let idx = vec![1.0; 1_000_000];
-        let tall = av(gridcraft_core::Array::filled(2_000, 1, nv(1.0)).into());
-        let r = ev("SORT", vec![tall, av(row(&idx))]);
-        assert_eq!(rows_of(&r).len(), 2_000);
-    }
-
     #[test]
     fn sortby() {
         let r = ev("SORTBY", vec![av(col(&[1.0, 2.0, 3.0])), av(col(&[30.0, 10.0, 20.0]))]);
@@ -1302,16 +1281,6 @@ mod tests {
         );
     }
 
-    /// Security: PIVOTBY totals read their source rows from indexes built in the grouping pass,
-    /// rather than rescanning the source once per total, so many distinct keys stay linear.
-    #[test]
-    fn pivotby_totals_with_many_groups() {
-        let keys: Vec<f64> = (0..20_000).map(f64::from).collect();
-        let r = rows_of(&ev("PIVOTBY", vec![av(col(&keys)), av(col(&vec![1.0; 20_000])), av(col(&vec![2.0; 20_000])), t("SUM")]));
-        assert_eq!(r.get(1), Some(&vec![nv(0.0), nv(2.0), nv(2.0)]));
-        assert_eq!(r.last(), Some(&vec![tv("Total"), nv(40_000.0), nv(40_000.0)]));
-    }
-
     #[test]
     fn pivotby_size_check_counts_column_header_rows() {
         // One row, n column fields and n value columns: n header rows + value-name row + body row + total row,