PDF export: index merged areas when drawing a block

draw_block checked every cell against every merge meeting the block,
and the fill, border and text passes searched the merge list again
for each merge start, so an imported sheet with many merge ranges in
a print area cost cells x merges (and merges x merges) comparisons.

The merges meeting a block are now gathered once and swept down the
rows: whether a cell is covered is a BTreeMap lookup among the merges
active on its row, and merge starts are found through a HashMap. A
merge overlapping one kept before it (only possible in a damaged file)
is left out, as Excel does when it repairs the file. Hyperlink
rectangles use the same index instead of scanning all sheet merges.

crates/engine/src/cmd/print/draw.rs

@@ -228,6 +228,82 @@ fn draw_border(pg: &mut Page, b: &BorderLine, wb: &Workbook, start: Point, end:
     pg.stroke_line_dashed(Point::new(x0, y0), Point::new(x1, y1), lw, color, &border_dash(b.style, k));
 }
 
+/// Merged areas active at one row of a downward sweep, kept disjoint: by first column, their last
+/// column and last row. Adding, dropping and looking up a merge costs log(active merges).
+#[derive(Default)]
+struct MergeSweep {
+    cols: std::collections::BTreeMap<u32, (u32, u32)>,
+    /// (last row, first column) of the active merges, soonest ending first.
+    ends: std::collections::BinaryHeap<std::cmp::Reverse<(u32, u32)>>,
+}
+
+impl MergeSweep {
+    /// Moves down to `row` (rows never decrease), dropping the merges that end above it.
+    fn advance(&mut self, row: u32) {
+        while let Some(std::cmp::Reverse((end, col))) = self.ends.peek().copied() {
+            if end >= row {
+                break;
+            }
+            self.ends.pop();
+            self.cols.remove(&col);
+        }
+    }
+
+    /// Does an active merge meet columns `c0..=c1`? The intervals are disjoint, so only the last
+    /// one starting at or before `c1` can.
+    fn meets(&self, c0: u32, c1: u32) -> bool {
+        self.cols.range(..=c1).next_back().is_some_and(|(_, (end, _))| *end >= c0)
+    }
+
+    /// Adds merge `m`, at row `row`, unless it overlaps an active one.
+    fn add(&mut self, m: RangeRef, row: u32) -> bool {
+        self.advance(row);
+        if self.meets(m.start.col, m.end.col) {
+            return false;
+        }
+        self.cols.insert(m.start.col, (m.end.col, m.end.row));
+        self.ends.push(std::cmp::Reverse((m.end.row, m.start.col)));
+        true
+    }
+}
+
+/// The merged areas that meet `block`, ordered by the row where they enter it, then by column,
+/// leaving out any that overlaps one kept before it (only a damaged file has overlapping merges;
+/// Excel drops them when it repairs one). One sweep down the rows instead of a scan of every
+/// merge per cell.
+fn block_merges(sh: &Sheet, block: RangeRef) -> Vec<RangeRef> {
+    let top = |m: &RangeRef| m.start.row.max(block.start.row);
+    let mut merges: Vec<RangeRef> = sh.merges.iter().filter(|m| m.intersects(&block)).copied().collect();
+    merges.sort_by_key(|m| (top(m), m.start.col));
+    let mut sweep = MergeSweep::default();
+    merges.retain(|m| sweep.add(*m, top(m)));
+    merges
+}
+
+/// Answers "is this cell inside a merge?" for cells asked in row-major order, from the merges of
+/// [`block_merges`].
+struct Covered<'a> {
+    merges: &'a [RangeRef],
+    top: u32,
+    next: usize,
+    sweep: MergeSweep,
+}
+
+impl<'a> Covered<'a> {
+    fn new(merges: &'a [RangeRef], block: RangeRef) -> Self {
+        Self { merges, top: block.start.row, next: 0, sweep: MergeSweep::default() }
+    }
+
+    fn covers(&mut self, c: CellRef) -> bool {
+        while let Some(m) = self.merges.get(self.next).filter(|m| m.start.row.max(self.top) <= c.row) {
+            self.sweep.add(*m, m.start.row.max(self.top));
+            self.next += 1;
+        }
+        self.sweep.advance(c.row);
+        self.sweep.meets(c.col, c.col)
+    }
+}
+
 /// Draws the cells of a block: fills, gridlines, data bars, borders and text.
 pub(super) fn draw_block(pg: &mut Page, b: &Block, gridlines: bool, cf: &mut Option<crate::cf::CfCache>) {
     let (bx, by) = (b.xs.first().copied().unwrap_or(0.0), b.ys.first().copied().unwrap_or(0.0));
@@ -237,17 +313,18 @@ pub(super) fn draw_block(pg: &mut Page, b: &Block, gridlines: bool, cf: &mut Opt
     }
     let block = b.range();
     let sh = b.sh;
-    let merges: Vec<RangeRef> = sh.merges.iter().filter(|m| m.intersects(&block)).copied().collect();
-    let covered = |c: CellRef| merges.iter().any(|m| m.contains(c));
-    // Cells worth visiting.
+    let merges = block_merges(sh, block);
+    let by_start: std::collections::HashMap<CellRef, RangeRef> = merges.iter().map(|m| (m.start, *m)).collect();
+    // Cells worth visiting, in row-major order (which `Covered` needs).
+    let mut covered = Covered::new(&merges, block);
     let cells: Vec<CellRef> = if block.count() <= DENSE_BLOCK {
-        block.iter().filter(|c| !covered(*c)).collect()
+        block.iter().filter(|c| !covered.covers(*c)).collect()
     } else {
         let mut set: Vec<CellRef> = sh.cells.iter_range(block).map(|(c, _)| c).collect();
         set.extend(sh.spill.range(block.start..=block.end).map(|(c, _)| *c).filter(|c| block.contains(*c)));
         set.sort_unstable();
         set.dedup();
-        set.retain(|c| !covered(*c));
+        set.retain(|c| !covered.covers(*c));
         set
     };
     pg.save();
@@ -277,14 +354,14 @@ pub(super) fn draw_block(pg: &mut Page, b: &Block, gridlines: bool, cf: &mut Opt
         .filter(|(c, l)| {
             l.fill.is_some()
                 || l.cf.is_some()
-                || merges.iter().any(|m| m.start == *c)
+                || by_start.contains_key(c)
                 || !sh.value(*c).is_empty()
                 || l.style.borders != Borders::default()
                 || l.style.num_fmt.as_str() == "checkbox"
         })
         .collect();
     let rect_of = |c: CellRef| -> (f32, f32, f32, f32, bool) {
-        match merges.iter().find(|m| m.start == c) {
+        match by_start.get(&c) {
             Some(m) => {
                 let (x, y, w, h) = b.rect(*m);
                 (x, y, w, h, true)
@@ -322,7 +399,7 @@ pub(super) fn draw_block(pg: &mut Page, b: &Block, gridlines: bool, cf: &mut Opt
             continue;
         }
         let bd = &l.style.borders;
-        let (right, bottom) = match merges.iter().find(|m| m.start == *c).filter(|_| merged) {
+        let (right, bottom) = match by_start.get(c).filter(|_| merged) {
             Some(m) => (
                 b.wb.styles.get(sh.style_id(CellRef::new(m.start.row, m.end.col))).borders.right,
                 b.wb.styles.get(sh.style_id(CellRef::new(m.end.row, m.start.col))).borders.bottom,
@@ -407,11 +484,12 @@ pub(super) struct PrintedBlock {
 pub(super) fn draw_block_links(pg: &mut Page, b: &Block, page: usize, pending: &mut Vec<PendingPdfLink>) {
     let range = b.range();
     let (Some(&bx), Some(&by), Some(&right), Some(&bottom)) = (b.xs.first(), b.ys.first(), b.xs.last(), b.ys.last()) else { return };
+    let by_start: std::collections::HashMap<CellRef, RangeRef> = block_merges(b.sh, range).into_iter().map(|m| (m.start, m)).collect();
     for (cell, link) in b.sh.hyperlinks.range(range.start..=range.end) {
         if !range.contains(*cell) {
             continue;
         }
-        let cell_range = b.sh.merges.iter().find(|m| m.start == *cell).copied().unwrap_or(RangeRef::cell(*cell));
+        let cell_range = by_start.get(cell).copied().unwrap_or(RangeRef::cell(*cell));
         let (x, y, w, h) = b.rect(cell_range);
         let (x0, y0, x1, y1) = (x.max(bx), y.max(by), (x + w).min(right), (y + h).min(bottom));
         if x1 <= x0 || y1 <= y0 {
@@ -503,3 +581,44 @@ pub(super) fn make_block<'a>(
     }
     Block { wb, si, sh, rows, cols, xs, ys, k, scale: lay.scale }
 }
+
+#[cfg(test)]
+mod merge_tests {
+    use super::{CellRef, Covered, RangeRef, Sheet, block_merges};
+
+    fn r(a: &str) -> RangeRef {
+        RangeRef::parse(a).unwrap()
+    }
+
+    #[test]
+    fn block_merges_drop_overlaps_and_cover_like_a_scan() {
+        let mut sh = Sheet::new("S");
+        sh.merges = vec![r("B2:C3"), r("E1:F8"), r("B2:B5"), r("A10:D10"), r("H3:H4"), r("Z1:Z2"), r("C3:D4"), r("A1:A1048576")];
+        let block = r("B1:H12");
+        let merges = block_merges(&sh, block);
+        // B2:B5 and C3:D4 overlap B2:C3, which comes first; Z1:Z2 is outside the block.
+        assert_eq!(merges, vec![r("E1:F8"), r("B2:C3"), r("H3:H4"), r("A10:D10")]);
+        let mut covered = Covered::new(&merges, block);
+        for c in block.iter() {
+            assert_eq!(covered.covers(c), merges.iter().any(|m| m.contains(c)), "{}", c.a1());
+        }
+        // Cells asked sparsely, still in order.
+        let mut covered = Covered::new(&merges, block);
+        for (a, want) in [("C1", false), ("F5", true), ("G5", false), ("D10", true), ("H12", false)] {
+            assert_eq!(covered.covers(CellRef::parse(a).unwrap()), want, "{a}");
+        }
+    }
+
+    #[test]
+    fn many_stacked_merges_are_swept_once() {
+        // 50,000 merges over the same cells: one is kept, and the block is checked against it
+        // without comparing every cell with every merge.
+        let mut sh = Sheet::new("S");
+        sh.merges = (0..50_000u32).map(|i| RangeRef::new(CellRef::new(0, 0), CellRef::new(199, 1 + i % 50))).collect();
+        let block = r("A1:AZ200");
+        let merges = block_merges(&sh, block);
+        assert_eq!(merges, vec![r("A1:B200")]);
+        let mut covered = Covered::new(&merges, block);
+        assert_eq!(block.iter().filter(|c| covered.covers(*c)).count(), 400);
+    }
+}

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,