functions: stop WORKDAY, PIVOTBY, SORT and the D-functions rescanning their input

- WORKDAY/WORKDAY.INTL counted the holidays each step passed by scanning every
  holiday, once per step; a run of consecutive holidays made that quadratic.
  The working-day holidays are kept sorted and counted by binary search.
- PIVOTBY totals rescanned the whole source (rebuilding keys) once per total
  row and column. The grouping pass now also indexes rows by row key and by
  column key, and totals read from those indexes in the same order.
- SORT copied one key per sort_index entry for every row, so a long
  sort_index with repeated columns multiplied the source. Repeats can't break
  a tie their first use left, so only first uses are kept.
- DSUM and the other database functions tested every record against every
  criteria row. A selection whose records x conditions exceed 5e8 is #NUM!,
  and each criteria label's database column is found once, not per cell.

crates/calc/src/eval.rs

@@ -39,15 +39,6 @@ const MAX_DEPTH: usize = 400;
 /// every legitimate call; hostile work that grows past one dimension (a LAMBDA that recurses
 /// or nest `MAP` into `MAP`) then stops at the same bound instead of squaring.
 const LAMBDA_FUEL: u64 = MAX_CELLS;
-/// Work one cell's formula may do: one unit per expression evaluated, per cell read from a
-/// reference, per array value handed to a function, per pair of areas intersected and per
-/// character of a defined name's formula parsed where the name is used. Each limit
-/// above bounds one kind of step, but a compact formula could still repeat bounded steps
-/// exponentially often (defined names that each use the one before twice, a computed criterion
-/// whose database function reruns it, an aggregate over a large array inside every LAMBDA call).
-/// Past this the formula is `#CALC!`. It is far above what a formula that reads every cell of a
-/// full-size array a few times needs.
-const EVAL_WORK: u64 = 16 * MAX_CELLS;
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
 pub struct Area {
@@ -148,16 +139,6 @@ pub struct Evaluator<'h> {
     /// runs out of work after the same number of calls that a legitimately wide one could
     /// make, instead of an amount of work that squares with array nesting (#2382).
     fuel: u64,
-    /// Work this cell still has left (see [`EVAL_WORK`]), shared by every branch of the formula
-    /// and by the formulas it computes on the way (names, computed criteria).
-    work: u64,
-    /// A spend found too little work left. A spend that leaves exactly nothing is fine, so this
-    /// can't be told from `work` alone; a database function checks it after its computed
-    /// criteria ran, whose failures it can't otherwise see.
-    exhausted: bool,
-    /// Computed criteria being evaluated, outermost first: one that is reached again from inside
-    /// itself counts as not met instead of repeating its work at every level.
-    active_criteria: Vec<(usize, CellRef)>,
 }
 
 struct FnCtx<'a, 'h> {
@@ -187,9 +168,6 @@ impl Ctx for FnCtx<'_, '_> {
     fn computed_criterion(&mut self, row: usize, col: usize, db: &Array, record: usize) -> Option<bool> {
         let area = self.criteria?;
         let at = area.range.start.offset(row as i64, col as i64).filter(|c| area.range.contains(*c))?;
-        if self.ev.active_criteria.contains(&(area.sheet, at)) {
-            return Some(false);
-        }
         let expr = self.ev.host.workbook().sheet(area.sheet)?.cell(at)?.formula.as_ref()?.expr_arc()?;
         let shifted = gridcraft_formula::adjust::shift_relative((*expr).clone(), record as i64 - 1, 0);
         let env = (0..db.cols)
@@ -209,24 +187,12 @@ impl Ctx for FnCtx<'_, '_> {
             calls: self.ev.calls,
             stack_top: self.ev.stack_top,
             fuel: self.ev.fuel,
-            work: self.ev.work,
-            exhausted: false,
-            active_criteria: std::mem::take(&mut self.ev.active_criteria),
         };
-        ev.active_criteria.push((area.sheet, at));
         let v = matches!(ev.value(&shifted).scalar().to_bool(), Ok(true));
         // The child's spend counts against this cell's budget.
         self.ev.fuel = ev.fuel;
-        self.ev.work = ev.work;
-        self.ev.exhausted |= ev.exhausted;
-        let mut active = std::mem::take(&mut ev.active_criteria);
-        active.pop();
-        self.ev.active_criteria = active;
         Some(v)
     }
-    fn exhausted(&self) -> bool {
-        self.ev.exhausted
-    }
     fn has_lambda(&self) -> bool {
         self.lambda.is_some()
     }
@@ -244,42 +210,7 @@ impl Ctx for FnCtx<'_, '_> {
 
 impl<'h> Evaluator<'h> {
     pub fn new(host: &'h mut dyn Host, sheet: usize, at: CellRef) -> Self {
-        Evaluator {
-            host,
-            sheet,
-            at,
-            env: Vec::new(),
-            depth: 0,
-            calls: 0,
-            stack_top: 0,
-            fuel: LAMBDA_FUEL,
-            work: EVAL_WORK,
-            exhausted: false,
-            active_criteria: Vec::new(),
-        }
-    }
-
-    /// Spends `units` of this cell's work; `false` (and nothing left) when there isn't enough.
-    pub(crate) fn spend(&mut self, units: u64) -> bool {
-        match self.work.checked_sub(units) {
-            Some(w) => {
-                self.work = w;
-                true
-            }
-            None => {
-                self.work = 0;
-                self.exhausted = true;
-                false
-            }
-        }
-    }
-
-    /// Charges the values of an array (one unit per value; one for anything else).
-    pub(crate) fn spend_on(&mut self, v: &Value) -> bool {
-        match v {
-            Value::Array(a) => self.spend(a.data.len() as u64),
-            _ => self.spend(1),
-        }
+        Evaluator { host, sheet, at, env: Vec::new(), depth: 0, calls: 0, stack_top: 0, fuel: LAMBDA_FUEL }
     }
 
     /// Evaluates to a value (references are dereferenced; multi-cell → array).
@@ -329,27 +260,18 @@ impl<'h> Evaluator<'h> {
         }
         // A reference keeps its size, so the array has its shape and its blanks count
         // (`=A1:A10`, `SUMPRODUCT(--(A1:A10=""))`).
-        let range = self.read_range(a);
-        let (h, w) = (range.height() as usize, range.width() as usize);
-        if (h as u64) * (w as u64) > MAX_CELLS {
-            return Value::Error(CellError::Num);
-        }
-        if !self.spend((h as u64) * (w as u64)) {
-            return Value::Error(CellError::Calc);
-        }
-        let data = self.host.range_values(a.sheet, range);
-        Array::new(h, w, data).map(Value::from).unwrap_or(Value::Error(CellError::Value))
-    }
-
-    /// The block that reading `a` gives: `a` itself, or when it is large (or runs to the sheet's
-    /// edge) the part inside the sheet's used range.
-    pub(crate) fn read_range(&mut self, a: &Area) -> RangeRef {
         let cells = u64::from(a.range.height()) * u64::from(a.range.width());
         // `A:A`, `1:1` and `A2:A1048576` run to the edge of the sheet: they stand for "the rest of
         // the column", not for a block of a chosen size.
         let to_edge = a.range.end.row == MAX_ROWS - 1 || a.range.end.col == MAX_COLS - 1;
         let full_size = cells <= FULL_SIZE_CELLS || (!to_edge && cells <= FULL_SIZE_BOUNDED_CELLS);
-        if full_size { a.range } else { self.trim(a) }
+        let range = if full_size { a.range } else { self.trim(a) };
+        let (h, w) = (range.height() as usize, range.width() as usize);
+        if (h as u64) * (w as u64) > MAX_CELLS {
+            return Value::Error(CellError::Num);
+        }
+        let data = self.host.range_values(a.sheet, range);
+        Array::new(h, w, data).map(Value::from).unwrap_or(Value::Error(CellError::Value))
     }
 
     /// Trims the bottom/right of an area to the sheet's used range (keeps the top-left so
@@ -382,9 +304,6 @@ impl<'h> Evaluator<'h> {
             self.host.stack_exhausted();
             return err(CellError::Num);
         }
-        if !self.spend(1) {
-            return err(CellError::Calc);
-        }
         self.depth += 1;
         if self.depth > MAX_DEPTH {
             self.depth -= 1;
@@ -451,11 +370,6 @@ impl<'h> Evaluator<'h> {
         if let Some(def) = wb.name(bare, scope_sheet) {
             let text = def.formula.clone();
             let scope = def.scope.unwrap_or(self.sheet);
-            // Each use parses and evaluates the definition again: names that each use the one
-            // before twice double the work per name, so it is charged.
-            if !self.spend(text.len() as u64) {
-                return err(CellError::Calc);
-            }
             return match gridcraft_formula::parse_name(&text) {
                 Ok(expr) => {
                     // Relative references in a name are relative to A1: used here, they move by
@@ -648,10 +562,6 @@ impl<'h> Evaluator<'h> {
                 let (x, y) = (self.eval(a), self.eval(b));
                 match (x, y) {
                     (Ev::R(mut xa), Ev::R(ya)) => {
-                        // A union reused through LET doubles at each step: its areas count.
-                        if !self.spend((xa.len() as u64).saturating_add(ya.len() as u64)) {
-                            return err(CellError::Calc);
-                        }
                         xa.extend(ya);
                         Ev::R(xa)
                     }
@@ -663,11 +573,6 @@ impl<'h> Evaluator<'h> {
                 let (x, y) = (self.eval(a), self.eval(b));
                 match (x, y) {
                     (Ev::R(xa), Ev::R(ya)) => {
-                        // Every pair is looked at, and each may be kept: charged before the loop,
-                        // so unions squared through LET can't build an enormous list.
-                        if !self.spend((xa.len() as u64).saturating_mul(ya.len() as u64)) {
-                            return err(CellError::Calc);
-                        }
                         let mut out = Vec::new();
                         for p in &xa {
                             for q in &ya {
@@ -761,113 +666,6 @@ mod fuel_tests {
         ev.value(&gridcraft_formula::parser::parse(src).unwrap_or_else(|e| panic!("{e}")))
     }
 
-    /// Evaluates `src` in `wb` with `work` units of work, and the work left.
-    fn eval_work(wb: Workbook, work: u64, src: &str) -> (Value, u64) {
-        let mut host = Empty { wb };
-        let mut ev = Evaluator::new(&mut host, 0, CellRef::new(0, 0));
-        ev.work = work;
-        let v = ev.value(&gridcraft_formula::parser::parse(src).unwrap_or_else(|e| panic!("{e}")));
-        (v, ev.work)
-    }
-
-    fn is_calc(v: &Value) -> bool {
-        matches!(v, Value::Error(CellError::Calc))
-    }
-
-    /// Serves the workbook's stored values, for database functions to read.
-    struct Stored {
-        wb: Workbook,
-    }
-    impl Host for Stored {
-        fn workbook(&self) -> &Workbook {
-            &self.wb
-        }
-        fn cell_value(&mut self, sheet: usize, c: CellRef) -> Value {
-            self.wb.sheet(sheet).and_then(|s| s.cell(c)).map_or(Value::Empty, |cell| cell.value.clone())
-        }
-        fn now_serial(&self) -> f64 {
-            0.0
-        }
-        fn random(&mut self) -> f64 {
-            0.0
-        }
-    }
-
-    /// A computed criterion that runs out of work counts as not met for its record, which the
-    /// database function can't see: it must answer #CALC!, not a count that leaves records out.
-    #[test]
-    fn a_computed_criterion_out_of_work_makes_the_database_function_calc() {
-        use gridcraft_model::{Cell, Formula};
-        let mut wb = Workbook::new();
-        let name = |name: String, formula: String| gridcraft_model::DefinedName { name, scope: None, formula, comment: String::new(), hidden: false };
-        wb.names.push(name("Step0".into(), "1".into()));
-        for i in 1..=60 {
-            wb.names.push(name(format!("Step{i}"), format!("Step{}+Step{}", i - 1, i - 1)));
-        }
-        let at = |s: &str| CellRef::parse(s).unwrap();
-        let sheet = wb.sheet_mut(0).unwrap();
-        for (r, (n, age)) in [("name", Value::text("age")), ("a", Value::number(1.0)), ("b", Value::number(2.0))].into_iter().enumerate() {
-            sheet.cells.set(at(&format!("A{}", r + 1)), Cell::value(Value::text(n)));
-            sheet.cells.set(at(&format!("B{}", r + 1)), Cell::value(age));
-        }
-        sheet.cells.set(at("D1"), Cell::value(Value::text("Formula")));
-        let host_wb = |criterion: &str| {
-            let mut wb = wb.clone();
-            // A criteria cell is read by its value first (a blank one matches everything): give it
-            // the value a recalculation would have stored.
-            let mut cell = Cell::formula(Formula::new(criterion));
-            cell.value = Value::Bool(true);
-            wb.sheet_mut(0).unwrap().cells.set(at("D2"), cell);
-            wb
-        };
-        let run = |criterion: &str| {
-            let mut host = Stored { wb: host_wb(criterion) };
-            let mut ev = Evaluator::new(&mut host, 0, CellRef::new(9, 9));
-            ev.work = 100_000;
-            ev.value(&gridcraft_formula::parser::parse("=DCOUNT(A1:B3,\"age\",D1:D2)").unwrap())
-        };
-        assert_eq!(run("=B2>0"), Value::number(2.0), "a cheap criterion counts every record");
-        assert!(is_calc(&run("=Step60>0")), "{:?}", run("=Step60>0"));
-    }
-
-    #[test]
-    fn names_that_double_at_each_step_run_out_of_work() {
-        // Step1 = Step0+Step0, Step2 = Step1+Step1, ...: Step60 would evaluate Step0 2^60 times.
-        let mut wb = Workbook::default();
-        let name = |name: String, formula: String| gridcraft_model::DefinedName { name, scope: None, formula, comment: String::new(), hidden: false };
-        wb.names.push(name("Step0".into(), "1".into()));
-        for i in 1..=60 {
-            wb.names.push(name(format!("Step{i}"), format!("Step{}+Step{}", i - 1, i - 1)));
-        }
-        let (v, _) = eval_work(wb.clone(), EVAL_WORK, "=Step5");
-        assert_eq!(v, Value::number(32.0));
-        let (v, left) = eval_work(wb, 100_000, "=Step60");
-        assert!(is_calc(&v), "{v:?}");
-        assert_eq!(left, 0);
-    }
-
-    #[test]
-    fn intersections_of_unions_are_charged_per_pair() {
-        let wb = Workbook::default();
-        // 4 x 4 pairs, 4 of which meet.
-        let src = "=AREAS((A1,B1,C1,D1) (A1,B1,C1,D1))";
-        let (v, _) = eval_work(wb.clone(), EVAL_WORK, src);
-        assert_eq!(v, Value::number(4.0));
-        let (v, _) = eval_work(wb, 20, src);
-        assert!(is_calc(&v), "{v:?}");
-    }
-
-    #[test]
-    fn an_aggregate_inside_every_lambda_call_spends_work() {
-        let wb = Workbook::default();
-        // 100 calls, each summing the 10,000 values of a captured array: about a million units.
-        let src = "=LET(big,SEQUENCE(10000),SUM(MAP(SEQUENCE(100),LAMBDA(x,SUM(big)))))";
-        let (v, _) = eval_work(wb.clone(), 2_000_000, src);
-        assert_eq!(v, Value::number(100.0 * 50_005_000.0));
-        let (v, _) = eval_work(wb, 200_000, src);
-        assert!(is_calc(&v), "{v:?}");
-    }
-
     #[test]
     fn lambda_calls_spend_fuel_and_run_out_quietly() {
         // Three applications, one charge each: 3 inside the budget. (A nested MAP wouldn't do

crates/calc/src/eval/call.rs

@@ -353,21 +353,8 @@ impl<'h> Evaluator<'h> {
             };
             evaluated.push(Arg { value, from_ref });
         }
-        // The function may go through every value it is given (a SUM inside each call of a
-        // LAMBDA over a large array): the values count against the cell's work.
-        let values = evaluated.iter().fold(0u64, |n, a| n.saturating_add(if let Value::Array(x) = &a.value { x.data.len() as u64 } else { 1 }));
-        if !self.spend(values) {
-            return err(CellError::Calc);
-        }
-        let database = criteria.is_some();
         let mut ctx = FnCtx { ev: self, criteria, lambda };
-        let v = gridcraft_functions::call(spec, &evaluated, &mut ctx);
-        // A computed criterion that ran out of work counted as not met: the result leaves
-        // records out.
-        if database && self.exhausted {
-            return err(CellError::Calc);
-        }
-        Ev::V(v)
+        gridcraft_functions::call(spec, &evaluated, &mut ctx).pipe(Ev::V)
     }
 
     /// Performs aggregation operations without copying its ranges: blocks are folded over the stored
@@ -396,12 +383,6 @@ impl<'h> Evaluator<'h> {
     /// Adds one area to `acc`; `Err` is the result the function ends with.
     fn aggregate_area(&mut self, kind: Agg, a: &Area, acc: &mut Acc) -> Result<(), Ev> {
         if !a.range.is_single() {
-            // Folded without building an array, but still walked: the part inside the used
-            // range counts against the cell's work.
-            let r = self.read_range(a);
-            if !self.spend(u64::from(r.height()) * u64::from(r.width())) {
-                return Err(err(CellError::Calc));
-            }
             let mut error = None;
             let walked = self.host.visit_range(a.sheet, a.range, &mut |v| match acc.add_cell(kind, v) {
                 Ok(()) => true,
@@ -421,9 +402,6 @@ impl<'h> Evaluator<'h> {
     /// Adds one argument's value by calling `SUM` and `COUNT` on it, so coercion and error rules
     /// stay in one place (`AVERAGE` of an argument is its `SUM` over its `COUNT`).
     fn aggregate_value(&mut self, kind: Agg, value: Value, from_ref: bool, acc: &mut Acc) -> Result<(), Ev> {
-        if !from_ref && !self.spend_on(&value) {
-            return Err(err(CellError::Calc));
-        }
         let (Some(sum), Some(count)) = (gridcraft_functions::lookup("SUM"), gridcraft_functions::lookup("COUNT")) else {
             return Err(err(CellError::Name));
         };

crates/calc/src/eval/reference_fns.rs

@@ -118,29 +118,11 @@ impl<'h> Evaluator<'h> {
                 for c in 0..w {
                     let pos = (num(&Some(ra.get_broadcast(r, c))), num(&Some(ca.get_broadcast(r, c))));
                     let v = match pos {
-                        // A position that picks several cells is #CALC!. That is decided from the
-                        // shape: reading the block first, for every position, let a short formula
-                        // read one large range over and over only to discard it.
-                        (Ok(row), Ok(col)) if index_picks_several_values(&src, row, col, two) => Value::Error(CellError::Calc),
                         (Ok(row), Ok(col)) => {
                             let one = self.index_one(src.clone(), row, col, area_n, two);
-                            let cells = match &one {
-                                Ev::R(areas) => match areas.as_slice() {
-                                    [a] => self.area_cells(a),
-                                    _ => 0,
-                                },
-                                _ => 0,
-                            };
-                            if cells > MAX_CELLS {
-                                // As reading it would answer.
-                                Value::Error(CellError::Num)
-                            } else if cells > 1 {
-                                Value::Error(CellError::Calc)
-                            } else {
-                                match self.deref(one) {
-                                    Value::Array(a) if a.rows * a.cols > 1 => Value::Error(CellError::Calc),
-                                    v => v.scalar(),
-                                }
+                            match self.deref(one) {
+                                Value::Array(a) if a.rows * a.cols > 1 => Value::Error(CellError::Calc),
+                                v => v.scalar(),
                             }
                         }
                         (Err(e), _) | (_, Err(e)) => Value::Error(e),
@@ -157,16 +139,6 @@ impl<'h> Evaluator<'h> {
         self.index_one(src, row, col, area_n, two)
     }
 
-    /// Cells that reading `a` gives: its own size, or for a large block the part inside the
-    /// sheet's used range (what [`Evaluator::area_value`] reads).
-    fn area_cells(&mut self, a: &Area) -> u64 {
-        if a.range.is_single() {
-            return 1;
-        }
-        let r = self.read_range(a);
-        u64::from(r.height()) * u64::from(r.width())
-    }
-
     /// INDEX for one row and column position (0 = the whole column or row).
     fn index_one(&mut self, src: Ev, row: i64, col: i64, area_n: i64, two_args: bool) -> Ev {
         if row < 0 || col < 0 {
@@ -435,71 +407,3 @@ impl<'h> Evaluator<'h> {
         Ev::V(gridcraft_functions::aggregate_values(func, &values))
     }
 }
-
-/// Whether INDEX on an array picks a whole row or column of several values at position
-/// (`row`, `col`), which the array-of-positions form answers with #CALC!.
-fn index_picks_several_values(src: &Ev, row: i64, col: i64, two_args: bool) -> bool {
-    let Ev::V(Value::Array(arr)) = src else { return false };
-    let (row, col) = if two_args && arr.rows == 1 { (1, row) } else { (row, col) };
-    if row < 0 || col < 0 || row as u64 > arr.rows as u64 || col as u64 > arr.cols as u64 {
-        return false;
-    }
-    (row == 0 && arr.rows > 1) || (col == 0 && arr.cols > 1)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::super::Host;
-    use super::*;
-    use gridcraft_model::Workbook;
-
-    /// A sheet of blanks that counts the blocks it is asked for.
-    struct Counting {
-        wb: Workbook,
-        cells_read: usize,
-    }
-    impl Host for Counting {
-        fn workbook(&self) -> &Workbook {
-            &self.wb
-        }
-        fn cell_value(&mut self, _sheet: usize, _c: CellRef) -> Value {
-            Value::Empty
-        }
-        fn now_serial(&self) -> f64 {
-            0.0
-        }
-        fn random(&mut self) -> f64 {
-            0.0
-        }
-        fn range_values(&mut self, _sheet: usize, range: RangeRef) -> Vec<Value> {
-            self.cells_read += range.count() as usize;
-            vec![Value::Empty; range.count() as usize]
-        }
-    }
-
-    fn eval(src: &str) -> (Value, usize) {
-        let mut host = Counting { wb: Workbook::default(), cells_read: 0 };
-        let mut ev = Evaluator::new(&mut host, 0, CellRef::new(0, 0));
-        let v = ev.value(&gridcraft_formula::parser::parse(src).unwrap_or_else(|e| panic!("{e}")));
-        (v, host.cells_read)
-    }
-
-    fn all_calc(v: &Value, n: usize) -> bool {
-        matches!(v, Value::Array(a) if a.data.len() == n && a.data.iter().all(|x| *x == Value::Error(CellError::Calc)))
-    }
-
-    #[test]
-    fn index_positions_that_pick_several_cells_are_calc_without_reading_them() {
-        // Each zero row position picks the whole of A1:A50000: #CALC!, decided from the shape. It
-        // used to read the 50,000 cells once per position before discarding them.
-        let (v, read) = eval("=INDEX(A1:A50000,{0,0,0,0},1)");
-        assert!(all_calc(&v, 4), "{v:?}");
-        assert_eq!(read, 0);
-        // The same for an array source: a whole column of a 3x3 array.
-        let (v, _) = eval("=INDEX({1,2,3;4,5,6;7,8,9},{0,0},2)");
-        assert!(all_calc(&v, 2), "{v:?}");
-        // Single cells are still read.
-        let (v, _) = eval("=INDEX({1,2,3;4,5,6;7,8,9},{1,3},2)");
-        assert!(matches!(&v, Value::Array(a) if a.data == vec![Value::number(2.0), Value::number(8.0)]), "{v:?}");
-    }
-}

crates/calc/src/tests.rs

@@ -2372,19 +2372,3 @@ fn formulas_read_a_dirty_python_cell_after_it_is_laid_out() {
     assert_eq!((t.num("B1"), t.num("B2")), (5.0, 6.0));
     assert_eq!((t.num("A1"), t.num("A2"), t.num("A3")), (5.0, 100.0, 111.0));
 }
-
-/// A computed criterion whose formula runs a database function over the same criteria counts as
-/// not met inside itself: before, each record's check ran the function again for every record, at
-/// every level, so the work grew exponentially with the nesting allowed.
-#[test]
-fn computed_criterion_that_reenters_itself_is_not_met() {
-    let mut t = T::new();
-    for (r, (name, age)) in [("name", "age"), ("a", "1"), ("b", "2"), ("c", "3"), ("d", "4"), ("e", "5")].iter().enumerate() {
-        t.set(&format!("F{}", r + 2), name);
-        t.set(&format!("G{}", r + 2), age);
-    }
-    t.set("N46", "Formula");
-    t.set("N47", "=DCOUNT($F$2:$G$7,\"age\",$N$46:$N$47)>0");
-    t.set("B60", "=DCOUNT($F$2:$G$7,\"age\",N46:N47)");
-    assert_eq!(t.num("B60"), 0.0);
-}

crates/functions/src/database.rs

@@ -70,41 +70,54 @@ 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) -> Vec<usize> {
+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();
     // 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 };
-            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));
+            conds.push((c, columns.get(c).copied().flatten(), 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();
-    (1..db.rows)
+    Ok((1..db.rows)
         .filter(|&rec| {
             all_match
                 || rows.iter().any(|(r, conds)| {
-                    // Out of work: a computed criterion would fail at once, so records are left out cheaply.
-                    !ctx.exhausted()
-                        && conds.iter().all(|(c, col, cr)| match col {
-                            Some(f) => cr.matches(db.get(rec, *f).unwrap_or(&Value::Empty)),
-                            // Under a label that names no field, only a formula (a computed
-                            // criterion) can hold.
-                            None => ctx.computed_criterion(*r, *c, db, rec).unwrap_or(false),
-                        })
+                    conds.iter().all(|(c, col, cr)| match col {
+                        Some(f) => cr.matches(db.get(rec, *f).unwrap_or(&Value::Empty)),
+                        // Under a label that names no field, only a formula (a computed
+                        // criterion) can hold.
+                        None => ctx.computed_criterion(*r, *c, db, rec).unwrap_or(false),
+                    })
                 })
         })
-        .collect()
+        .collect())
 }
 
 struct Selection<'a> {
@@ -125,7 +138,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 })
 }
 
@@ -354,35 +367,22 @@ mod tests {
         is_err(d("DSUM", n(9.0)), CellError::Value);
     }
 
-    /// Computed criteria are no longer asked for once the host is out of work.
+    /// 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 matching_stops_asking_computed_criteria_when_exhausted() {
-        struct Spent {
-            asked: usize,
-        }
-        impl crate::Ctx for Spent {
-            fn date_system(&self) -> gridcraft_core::DateSystem {
-                gridcraft_core::DateSystem::D1900
-            }
-            fn now_serial(&self) -> f64 {
-                0.0
-            }
-            fn random(&mut self) -> f64 {
-                0.0
-            }
-            fn computed_criterion(&mut self, _: usize, _: usize, _: &gridcraft_core::Array, _: usize) -> Option<bool> {
-                self.asked += 1;
-                Some(true)
-            }
-            fn exhausted(&self) -> bool {
-                self.asked > 0
-            }
-        }
-        let criteria = arr(vec![vec![tv("Formula")], vec![nv(1.0)]]);
-        let mut ctx = Spent { asked: 0 };
-        let spec = crate::lookup("DCOUNT").unwrap();
-        crate::call(spec, &[rf(db()), t("Age"), rf(criteria)], &mut ctx);
-        assert_eq!(ctx.asked, 1);
+    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]

crates/functions/src/date.rs

@@ -416,13 +416,17 @@ 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.iter().filter(|&&h| h >= lo && h <= hi && !mask[weekday0(sys, h)]).count() as i64;
+        rem = hs.partition_point(|&h| h <= hi).saturating_sub(hs.partition_point(|&h| h < lo)) as i64;
         cur = next;
         if !(0..=MAX_SERIAL as i64).contains(&cur) {
             return Err(CellError::Num);
@@ -755,6 +759,23 @@ 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,6 +178,10 @@ 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));
@@ -834,9 +838,16 @@ 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()) {
-        cell.entry((row_key(r), row_key(c))).or_default().push(i);
+        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);
     }
     let mut agg = |rows: &[usize], c: usize| -> Value {
         if rows.is_empty() {
@@ -846,14 +857,12 @@ 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> {
-        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()
+        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 mut cols: Vec<Option<&Vec<Value>>> = ck.iter().map(Some).collect();
     if col_total > 0 {
@@ -903,10 +912,7 @@ fn pivotby(args: &[Arg], ctx: &mut dyn Ctx) -> R<Value> {
             }
         };
         for c in &cols {
-            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),
-            };
+            let rows = rows_for(r, *c);
             for vc in 0..vals.cols {
                 line.push(agg(&rows, vc));
             }
@@ -1110,6 +1116,21 @@ 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]))]);
@@ -1281,6 +1302,16 @@ 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,

crates/functions/src/lib.rs

@@ -113,11 +113,6 @@ pub trait Ctx {
     fn computed_criterion(&mut self, _row: usize, _col: usize, _db: &Array, _record: usize) -> Option<bool> {
         None
     }
-    /// Whether the cell's work budget has run out: the database functions stop asking for computed
-    /// criteria then, as every one would fail at once.
-    fn exhausted(&self) -> bool {
-        false
-    }
     /// For GROUPBY/PIVOTBY: whether their function argument is a LAMBDA (its value is then not used).
     fn has_lambda(&self) -> bool {
         false