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/chart/src/resolve.rs

@@ -11,10 +11,6 @@ use crate::{ChartData, Rgba, SeriesData, TrendlineData};
 pub(crate) const MAX_POINTS: usize = 100_000;
 /// Cap on series per chart.
 pub(crate) const MAX_SERIES: usize = 255;
-/// Cap on the points of all series of one chart together (values, X values and bubble sizes):
-/// [`MAX_SERIES`] × [`MAX_POINTS`] would let a few short whole-column references retain and
-/// draw tens of millions of points. Series past it get only what is left, or no points.
-pub(crate) const MAX_CHART_POINTS: usize = 1_000_000;
 /// Cap on trendlines drawn per series.
 pub(crate) const MAX_TRENDLINES: usize = 16;
 
@@ -81,15 +77,10 @@ fn union_parts(f: &str) -> Option<Vec<&str>> {
 /// Evaluates a series formula to a flat list of values (row-major), capped. A union of
 /// references (`(A1:A3,A5)`, as PivotCharts use) is read part by part.
 fn eval_list(wb: &Workbook, sheet: usize, formula: &str) -> Vec<Value> {
-    eval_list_in(wb, sheet, formula, 0, MAX_POINTS)
+    eval_list_in(wb, sheet, formula, 0)
 }
 
-/// [`eval_list`] keeping at most `cap` values (and never more than [`MAX_POINTS`]).
-fn eval_list_capped(wb: &Workbook, sheet: usize, formula: &str, cap: usize) -> Vec<Value> {
-    eval_list_in(wb, sheet, formula, 0, cap.min(MAX_POINTS))
-}
-
-fn eval_list_in(wb: &Workbook, sheet: usize, formula: &str, depth: u32, cap: usize) -> Vec<Value> {
+fn eval_list_in(wb: &Workbook, sheet: usize, formula: &str, depth: u32) -> Vec<Value> {
     let f = formula.trim();
     if f.is_empty() {
         return vec![];
@@ -99,19 +90,16 @@ fn eval_list_in(wb: &Workbook, sheet: usize, formula: &str, depth: u32, cap: usi
     {
         let mut out = Vec::new();
         for p in parts {
-            if out.len() >= cap {
+            if out.len() >= MAX_POINTS {
                 break;
             }
-            let rest = cap - out.len();
-            out.extend(eval_list_in(wb, sheet, p, depth + 1, rest).into_iter().take(rest));
+            let rest = MAX_POINTS - out.len();
+            out.extend(eval_list_in(wb, sheet, p, depth + 1).into_iter().take(rest));
         }
         return out;
     }
-    if cap == 0 {
-        return vec![];
-    }
     match evaluate(wb, sheet, CellRef::default(), f) {
-        Value::Array(a) => a.iter().take(cap).cloned().collect(),
+        Value::Array(a) => a.iter().take(MAX_POINTS).cloned().collect(),
         v => vec![v],
     }
 }
@@ -123,11 +111,8 @@ fn to_num(v: &Value) -> Option<f64> {
     }
 }
 
-/// A series formula's numbers, at most `cap` of them; `cap` is reduced by what was taken.
-fn nums(wb: &Workbook, sheet: usize, formula: &str, cap: &mut usize) -> Vec<Option<f64>> {
-    let out: Vec<Option<f64>> = eval_list_capped(wb, sheet, formula, *cap).iter().map(to_num).collect();
-    *cap -= out.len().min(*cap);
-    out
+fn nums(wb: &Workbook, sheet: usize, formula: &str) -> Vec<Option<f64>> {
+    eval_list(wb, sheet, formula).iter().map(to_num).collect()
 }
 
 /// Number format code of the first cell of a reference ("General" when not a reference).
@@ -243,26 +228,15 @@ pub fn resolve(wb: &Workbook, sheet: usize, chart: &Chart) -> ChartData {
     let mut series = Vec::new();
     let mut categories: Option<Vec<String>> = None;
     let mut category_levels = Vec::new();
-    // One point budget for the whole chart, not one per series.
-    let mut budget = MAX_CHART_POINTS;
     for (i, s) in chart.series.iter().take(MAX_SERIES).enumerate() {
-        let values = nums(wb, sheet, &s.values, &mut budget);
+        let values = nums(wb, sheet, &s.values);
         let name = s.name.as_deref().map(|n| series_name(wb, sheet, n)).filter(|n| !n.is_empty()).unwrap_or_else(|| format!("Series{}", i + 1));
         let color =
             s.color.as_ref().and_then(|c| c.resolve(&wb.theme)).map(|[r, g, b]| [r, g, b, 0xFF]).unwrap_or_else(|| series_color(&wb.theme, i));
         let kind = effective_kind(chart.kind, s.kind, i);
         let x = match (&s.categories, numeric_x || matches!(kind, ChartKind::Scatter | ChartKind::ScatterLines | ChartKind::Bubble)) {
             // X values with text in them (labels) are ignored: the points are numbered 1..n.
-            (Some(f), true) => {
-                let xs = eval_list_capped(wb, sheet, f, budget);
-                // Only X values that are kept are charged to the budget.
-                if xs.iter().any(Value::is_text) {
-                    None
-                } else {
-                    budget -= xs.len().min(budget);
-                    Some(xs.iter().map(to_num).collect())
-                }
-            }
+            (Some(f), true) => Some(eval_list(wb, sheet, f)).filter(|xs| !xs.iter().any(Value::is_text)).map(|xs| xs.iter().map(to_num).collect()),
             _ => None,
         };
         if categories.is_none()
@@ -274,7 +248,7 @@ pub fn resolve(wb: &Workbook, sheet: usize, chart: &Chart) -> ChartData {
                 category_levels = outer;
             }
         }
-        let sizes = s.bubble_sizes.as_deref().map(|f| nums(wb, sheet, f, &mut budget));
+        let sizes = s.bubble_sizes.as_deref().map(|f| nums(wb, sheet, f));
         let trendlines = s
             .trendlines
             .iter()

crates/chart/src/tests.rs

@@ -1258,41 +1258,3 @@ fn box_whisker_draws_each_series_quartile_method() {
     // The boxes span 3.5 and 2.5 data units on the same value axis.
     assert!((heights[0] / heights[1] - 1.4).abs() < 0.001);
 }
-
-#[test]
-fn series_share_one_point_budget() {
-    // Many series over the same long column each get a share of one chart-wide budget, not
-    // `MAX_POINTS` apiece.
-    let mut wb = Workbook::new();
-    let s = wb.sheet_mut(0).unwrap();
-    s.set_value(CellRef::new(0, 0), Value::Number(1.0));
-    s.set_value(CellRef::new(59_999, 0), Value::Number(2.0));
-    let mut c = chart(ChartKind::ColumnClustered);
-    c.series = (0..40).map(|_| Series { values: "Sheet1!$A$1:$A$60000".into(), ..Default::default() }).collect();
-    let d = resolve(&wb, 0, &c);
-    assert_eq!(d.series.len(), 40);
-    assert_eq!(d.series[0].values.len(), 60_000);
-    let total: usize = d.series.iter().map(|s| s.values.len()).sum();
-    assert_eq!(total, crate::resolve::MAX_CHART_POINTS);
-    assert!(d.series.last().is_some_and(|s| s.values.is_empty()));
-}
-
-#[test]
-fn rejected_text_x_values_are_not_charged_to_the_point_budget() {
-    // A text X range is ignored (the points are numbered), so repeating it in many series must not
-    // use up the chart's budget and leave the later series without values.
-    let mut wb = Workbook::new();
-    let s = wb.sheet_mut(0).unwrap();
-    for r in 0..100 {
-        s.set_value(CellRef::new(r, 0), Value::Number(f64::from(r)));
-    }
-    s.set_value(CellRef::new(0, 1), Value::text("label"));
-    s.set_value(CellRef::new(59_999, 1), Value::Number(2.0));
-    let mut c = chart(ChartKind::Scatter);
-    c.series = (0..40)
-        .map(|_| Series { values: "Sheet1!$A$1:$A$100".into(), categories: Some("Sheet1!$B$1:$B$60000".into()), ..Default::default() })
-        .collect();
-    let d = resolve(&wb, 0, &c);
-    assert_eq!(d.series.len(), 40);
-    assert!(d.series.iter().all(|s| s.values.len() == 100 && s.x.is_none()));
-}

crates/engine/src/chart_svg.rs

@@ -10,10 +10,6 @@ use crate::io::esc;
 /// Widest or tallest chart written; larger anchors are clamped.
 const MAX_SIDE: f32 = 16_384.0;
 
-/// Most bytes of SVG written for the charts of one web page together. A chart past what is
-/// left is left out of the page rather than grown without bound.
-pub(crate) const MAX_PAGE_SVG_BYTES: usize = 64 * 1024 * 1024;
-
 struct HelveticaMeasure;
 
 impl gridcraft_chart::Measure for HelveticaMeasure {
@@ -22,34 +18,24 @@ impl gridcraft_chart::Measure for HelveticaMeasure {
     }
 }
 
-/// `chart` of sheet `sheet` as an `<svg>` element placed by `style`, or `None` for a degenerate
-/// size or an element longer than `budget` bytes.
-pub(crate) fn chart_svg(wb: &Workbook, sheet: usize, chart: &Chart, style: &str, budget: usize) -> Option<String> {
+/// `chart` of sheet `sheet` as an `<svg>` element placed by `style`, or `None` for a degenerate size.
+pub(crate) fn chart_svg(wb: &Workbook, sheet: usize, chart: &Chart, style: &str) -> Option<String> {
     let (w, h) = (chart.anchor.width.min(MAX_SIDE), chart.anchor.height.min(MAX_SIDE));
     if !(w >= 1.0 && h >= 1.0) {
         return None;
     }
     let data = gridcraft_chart::resolve(wb, sheet, chart);
     let prims = gridcraft_chart::render(chart, &data, w, h, &HelveticaMeasure);
-    prims_svg_within(&prims, w, h, style, budget)
+    Some(prims_svg(&prims, w, h, style))
 }
 
-#[cfg(test)]
 fn prims_svg(prims: &[Prim], w: f32, h: f32, style: &str) -> String {
-    prims_svg_within(prims, w, h, style, usize::MAX).unwrap_or_default()
-}
-
-/// The `<svg>` element, or `None` as soon as it grows past `budget` bytes.
-fn prims_svg_within(prims: &[Prim], w: f32, h: f32, style: &str, budget: usize) -> Option<String> {
     let mut out = String::new();
     let _ = write!(
         out,
         "<svg xmlns=\"http://www.w3.org/2000/svg\" width=\"{w}\" height=\"{h}\" viewBox=\"0 0 {w} {h}\" font-family=\"Helvetica,Arial,sans-serif\" style=\"{style}\">"
     );
     for p in prims {
-        if out.len() > budget {
-            return None;
-        }
         match p {
             Prim::Rect { x, y, w, h, fill, stroke, radius } => {
                 let _ = write!(out, "<rect x=\"{}\" y=\"{}\" width=\"{}\" height=\"{}\"", num(*x), num(*y), num(*w), num(*h));
@@ -106,7 +92,7 @@ fn prims_svg_within(prims: &[Prim], w: f32, h: f32, style: &str, budget: usize)
         }
     }
     out.push_str("</svg>");
-    (out.len() <= budget).then_some(out)
+    out
 }
 
 /// Fill and stroke attributes of an open shape element; the caller closes it.
@@ -163,17 +149,6 @@ fn num(v: f32) -> String {
 mod tests {
     use super::*;
 
-    #[test]
-    fn an_svg_past_its_byte_budget_is_left_out() {
-        let rect = Prim::Rect { x: 1.0, y: 2.0, w: 3.0, h: 4.0, fill: Some([255, 0, 0, 255]), stroke: None, radius: 0.0 };
-        let prims = vec![rect; 10_000];
-        let full = prims_svg(&prims, 100.0, 50.0, "");
-        assert!(full.len() > 100_000);
-        assert_eq!(prims_svg_within(&prims, 100.0, 50.0, "", full.len()), Some(full.clone()));
-        assert_eq!(prims_svg_within(&prims, 100.0, 50.0, "", full.len() - 1), None);
-        assert_eq!(prims_svg_within(&prims, 100.0, 50.0, "", 1_000), None);
-    }
-
     #[test]
     fn every_primitive_becomes_an_svg_element() {
         let red = [255, 0, 0, 255];

crates/engine/src/io.rs

@@ -736,10 +736,8 @@ pub fn to_html(wb: &Workbook, sheet: usize) -> Result<String> {
                 place(&im.anchor)
             );
         }
-        let mut svg_budget = crate::chart_svg::MAX_PAGE_SVG_BYTES;
         for ch in &sh.charts {
-            if let Some(svg) = crate::chart_svg::chart_svg(wb, sheet, ch, &place(&ch.anchor), svg_budget) {
-                svg_budget -= svg.len();
+            if let Some(svg) = crate::chart_svg::chart_svg(wb, sheet, ch, &place(&ch.anchor)) {
                 out.push_str(&svg);
             }
         }

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,