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/engine/src/cmd/edit/go_to.rs

@@ -17,13 +17,11 @@ pub(in super::super) fn go_to(s: &mut Session, p: &Json) -> Result<Json> {
     }
     let Some(reference) = str_param(p, "reference").or_else(|| str_param(p, "range")) else { return Err(bad("edit.goTo", "missing `reference`")) };
     let d = s.doc()?;
-    let mut names = Resolver::new(&d.wb);
     let mut target: Option<usize> = None;
     let mut areas: Vec<RangeRef> = Vec::new();
     let mut named = false;
     for part in reference_parts(reference) {
-        let mut found = Vec::new();
-        let Some((sheet, _, by_name)) = names.target(d.wb.active_sheet(), part, 0, Some(&mut found)) else {
+        let Some((sheet, found, by_name)) = go_to_target(&d.wb, d.wb.active_sheet(), part, 0) else {
             return Err(EngineError::Other("Reference isn't valid.".into()));
         };
         // One selection holds the areas of one sheet.
@@ -79,114 +77,57 @@ pub(super) fn reference_parts(text: &str) -> impl Iterator<Item = &str> {
     })
 }
 
-/// Resolves Go To references against one workbook. Defined names are found through an index
-/// built once, and whether a name refers to a reference is remembered: listing the names of a
-/// workbook full of names that alias other names would otherwise scan every name for every alias
-/// of every name.
-struct Resolver<'a> {
-    wb: &'a gridcraft_model::Workbook,
-    /// Indexes into `wb.names` by scope and folded name, in workbook order.
-    by_name: std::collections::HashMap<(Option<usize>, String), Vec<usize>>,
-    /// [`Resolver::name_areas`] without areas, by name index, sheet and depth: the target sheet
-    /// and the number of areas.
-    memo: std::collections::HashMap<(usize, usize, usize), Option<(usize, usize)>>,
+/// Where one part of a Go To reference points, seen from sheet `sheet`: the sheet, its areas,
+/// and whether a defined name or a table gave them. `Sheet2!Name` is the name scoped to Sheet2.
+/// A name counts when what it refers to is a reference; a constant or a formula is not one.
+fn go_to_target(wb: &gridcraft_model::Workbook, sheet: usize, part: &str, depth: usize) -> Option<(usize, Vec<RangeRef>, bool)> {
+    if part.is_empty() || depth > GO_TO_NAME_DEPTH {
+        return None;
+    }
+    let (prefix, body) = split_sheet(part);
+    let on = match &prefix {
+        Some(n) => wb.sheet_index(n)?,
+        None => sheet,
+    };
+    if let Some(r) = RangeRef::parse(&body.replace('$', "")) {
+        return Some((on, vec![r], false));
+    }
+    let name = match prefix {
+        Some(_) => wb.names.iter().find(|n| n.scope == Some(on) && gridcraft_formula::same_name(&n.name, body)),
+        None => wb.name(body, sheet),
+    };
+    if let Some(n) = name {
+        return name_areas(wb, n, sheet, depth).map(|(t, areas)| (t, areas, true));
+    }
+    if prefix.is_none()
+        && let Some((si, ti)) = wb.table(body)
+    {
+        // As in Excel, a table's name stands for its data, without the header and totals rows.
+        let t = wb.sheet(si)?.tables.get(ti)?;
+        return Some((si, vec![t.data_range().unwrap_or(t.range)], true));
+    }
+    None
 }
 
-impl<'a> Resolver<'a> {
-    fn new(wb: &'a gridcraft_model::Workbook) -> Self {
-        let mut by_name: std::collections::HashMap<(Option<usize>, String), Vec<usize>> = std::collections::HashMap::new();
-        for (i, n) in wb.names.iter().enumerate() {
-            by_name.entry((n.scope, gridcraft_formula::fold_name(&n.name))).or_default().push(i);
-        }
-        Self { wb, by_name, memo: std::collections::HashMap::new() }
-    }
-
-    /// The first name of scope `scope` that `same` accepts, as a scan of `wb.names` finds it.
-    /// Names `same` accepts fold alike, so only that bucket is checked.
-    fn find(&self, scope: Option<usize>, body: &str, same: impl Fn(&str, &str) -> bool) -> Option<usize> {
-        self.by_name
-            .get(&(scope, gridcraft_formula::fold_name(body)))?
-            .iter()
-            .copied()
-            .find(|&i| self.wb.names.get(i).is_some_and(|n| same(&n.name, body)))
-    }
-
-    /// Where one part of a Go To reference points, seen from sheet `sheet`: the sheet, how many
-    /// areas (appended to `areas` when given), and whether a defined name or a table gave them.
-    /// `Sheet2!Name` is the name scoped to Sheet2. A name counts when what it refers to is a
-    /// reference; a constant or a formula is not one.
-    fn target(&mut self, sheet: usize, part: &str, depth: usize, areas: Option<&mut Vec<RangeRef>>) -> Option<(usize, usize, bool)> {
-        let wb = self.wb;
-        if part.is_empty() || depth > GO_TO_NAME_DEPTH {
+/// The sheet and areas defined name `n` refers to, seen from sheet `sheet`, when that is a
+/// reference.
+fn name_areas(wb: &gridcraft_model::Workbook, n: &gridcraft_model::DefinedName, sheet: usize, depth: usize) -> Option<(usize, Vec<RangeRef>)> {
+    let from = n.scope.unwrap_or(sheet);
+    let formula = n.formula.trim().trim_start_matches('=');
+    let mut target: Option<usize> = None;
+    let mut areas = Vec::new();
+    for p in reference_parts(formula) {
+        let (s, found, _) = go_to_target(wb, from, p, depth + 1)?;
+        if target.is_some_and(|t| t != s) {
             return None;
         }
-        let (prefix, body) = split_sheet(part);
-        let on = match &prefix {
-            Some(n) => wb.sheet_index(n)?,
-            None => sheet,
-        };
-        if let Some(r) = RangeRef::parse(&body.replace('$', "")) {
-            if let Some(a) = areas {
-                a.push(r);
-            }
-            return Some((on, 1, false));
+        target = Some(s);
+        areas.extend(found);
+        if areas.len() > GO_TO_MAX_AREAS {
+            return None;
         }
-        // As `Workbook::name`: the sheet's own name, else the workbook's.
-        let name = match prefix {
-            Some(_) => self.find(Some(on), body, gridcraft_formula::same_name),
-            None => self.find(Some(sheet), body, str::eq_ignore_ascii_case).or_else(|| self.find(None, body, str::eq_ignore_ascii_case)),
-        };
-        if let Some(i) = name {
-            return self.name_areas(i, sheet, depth, areas).map(|(t, n)| (t, n, true));
-        }
-        if prefix.is_none()
-            && let Some((si, ti)) = wb.table(body)
-        {
-            // As in Excel, a table's name stands for its data, without the header and totals rows.
-            let t = wb.sheet(si)?.tables.get(ti)?;
-            if let Some(a) = areas {
-                a.push(t.data_range().unwrap_or(t.range));
-            }
-            return Some((si, 1, true));
-        }
-        None
-    }
-
-    /// The sheet and number of areas defined name `wb.names[i]` refers to, seen from sheet
-    /// `sheet`, when that is a reference; the areas are appended to `areas` when given.
-    fn name_areas(&mut self, i: usize, sheet: usize, depth: usize, mut areas: Option<&mut Vec<RangeRef>>) -> Option<(usize, usize)> {
-        let key = (i, sheet, depth);
-        if areas.is_none()
-            && let Some(known) = self.memo.get(&key)
-        {
-            return *known;
-        }
-        let out = self.name_areas_uncached(i, sheet, depth, areas.as_deref_mut());
-        if areas.is_none() {
-            self.memo.insert(key, out);
-        }
-        out
-    }
-
-    fn name_areas_uncached(&mut self, i: usize, sheet: usize, depth: usize, mut areas: Option<&mut Vec<RangeRef>>) -> Option<(usize, usize)> {
-        let n = self.wb.names.get(i)?;
-        let from = n.scope.unwrap_or(sheet);
-        let formula = n.formula.trim().trim_start_matches('=');
-        let mut target: Option<usize> = None;
-        let mut count = 0usize;
-        for p in reference_parts(formula) {
-            let (s, found, _) = self.target(from, p, depth + 1, areas.as_deref_mut())?;
-            if target.is_some_and(|t| t != s) {
-                return None;
-            }
-            target = Some(s);
-            count += found;
-            if count > GO_TO_MAX_AREAS {
-                return None;
-            }
-        }
-        target.map(|t| (t, count))
     }
+    target.map(|t| (t, areas))
 }
 
 /// The names the Go To dialog lists, seen from sheet `sheet`: the named ranges of the workbook and
@@ -196,13 +137,11 @@ impl<'a> Resolver<'a> {
 pub fn go_to_names(wb: &gridcraft_model::Workbook, sheet: usize) -> Vec<String> {
     let own: std::collections::HashSet<String> =
         wb.names.iter().filter(|n| n.scope == Some(sheet)).map(|n| gridcraft_formula::fold_name(&n.name)).collect();
-    let mut names = Resolver::new(wb);
     let mut out: Vec<String> = wb
         .names
         .iter()
-        .enumerate()
-        .filter(|(i, n)| !n.hidden && names.name_areas(*i, sheet, 0, None).is_some())
-        .filter_map(|(_, n)| match n.scope {
+        .filter(|n| !n.hidden && name_areas(wb, n, sheet, 0).is_some())
+        .filter_map(|n| match n.scope {
             None if own.contains(&gridcraft_formula::fold_name(&n.name)) => None,
             None => Some(n.name.clone()),
             Some(s) if s == sheet => Some(n.name.clone()),

crates/engine/src/cmd/edit/tests.rs

@@ -42,28 +42,6 @@ fn go_to_lists_the_named_ranges_of_every_sheet() {
     assert_eq!(list["names"], json!(["Areas", "Local", "Price", "Sheet1!Own", "type"]), "{list}");
 }
 
-/// Thousands of names that alias other names are listed with one lookup per alias, not a scan of
-/// every name for each; aliases resolve through the chain like a single name does.
-#[test]
-fn go_to_lists_many_aliases_and_follows_them() {
-    let mut s = go_to_workbook();
-    let wb = std::sync::Arc::make_mut(&mut s.doc_mut().unwrap().wb);
-    let alias = |name: String, formula: String| gridcraft_model::DefinedName { name, scope: None, formula, comment: String::new(), hidden: false };
-    wb.names.push(alias("Chain0".into(), "Areas".into()));
-    for i in 1..20_000 {
-        wb.names.push(alias(format!("Chain{i}"), format!("Chain{},price", i - 1)));
-    }
-    let list = s.execute("edit.goTo", json!({"list": true})).unwrap();
-    let names: Vec<&str> = list["names"].as_array().unwrap().iter().filter_map(|n| n.as_str()).collect();
-    // Chain0..Chain6 are within the name depth Go To follows; longer chains are not ranges.
-    for i in 0..7 {
-        assert!(names.contains(&format!("Chain{i}").as_str()), "Chain{i} in {names:?}");
-    }
-    assert!(!names.contains(&"Chain7"), "{names:?}");
-    let out = s.execute("edit.goTo", json!({"reference": "chain1"})).unwrap();
-    assert_eq!(out["selection"], "A1,C3:C4,B2:B5", "{out}");
-}
-
 #[test]
 fn go_to_a_name_scoped_to_another_sheet_goes_to_that_sheet() {
     let mut s = go_to_workbook();

crates/functions/src/database.rs

@@ -70,27 +70,42 @@ 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)| {
@@ -102,7 +117,7 @@ fn matching_records(db: &Array, crit: &Array, ctx: &mut dyn Ctx) -> Vec<usize> {
                     })
                 })
         })
-        .collect()
+        .collect())
 }
 
 struct Selection<'a> {
@@ -123,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 })
 }
 
@@ -352,6 +367,24 @@ 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,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,