address review: base the in-history forecast cap on the slots the fit range spans

MAX_WORK bounded history rows x forecast rows, but a Forecast Start
inside the history adds a hindcast row for every history point after
it, each holding the same ETS formulas that refit the model over the
history before the start. Those rows were not counted, so a start in
the middle of 10,000 rows of history built millions of uncharged
model fits per recalculation.

The forecast row cap is now computed from the rows the model is fit
to, after subtracting the hindcast rows; a start so early that the
hindcast alone passes the limit is refused, and the default number of
periods stays within the cap.

crates/engine/src/cmd/forecast.rs

@@ -36,6 +36,21 @@ fn max_periods(history: usize) -> u32 {
     u32::try_from(MAX_WORK / h).unwrap_or(u32::MAX).clamp(1, MAX_PERIODS)
 }
 
+/// Most forecast rows when the model is fit to the first `fit` of `history` rows, which span
+/// `slots` timeline slots once their gaps are filled. With a Forecast Start inside the history,
+/// every row after it is forecast too (a hindcast row holds the same ETS formulas, each refitting
+/// the model over the `slots` slots), so those rows (counted from the raw rows) count against
+/// [`MAX_WORK`] before any forecast row does. `None` when the hindcast alone exceeds it.
+fn max_periods_from(history: usize, fit: usize, slots: usize) -> Option<u32> {
+    if fit >= history {
+        return Some(max_periods(history));
+    }
+    let f = u64::try_from(slots.max(fit)).unwrap_or(u64::MAX).max(1);
+    let hindcast = u64::try_from(history - fit).unwrap_or(u64::MAX);
+    let rest = (MAX_WORK / f).checked_sub(hindcast).filter(|r| *r >= 1)?;
+    Some(u32::try_from(rest).unwrap_or(u32::MAX).min(MAX_PERIODS))
+}
+
 pub fn specs() -> Vec<CommandSpec> {
     vec![cmd!(
         "data.forecastSheet",
@@ -257,13 +272,23 @@ fn forecast_sheet(s: &mut Session, p: &Json) -> Result<Json> {
             }
         }
     };
-    let cap = max_periods(rows.len());
+    // The slots the model is fit over: the fit rows' span of the timeline, gaps included.
+    let slots = match (fit < rows.len(), times.get(fit.wrapping_sub(1)), times.first()) {
+        (true, Some(last_fit), Some(first)) => match (timeline.steps_after_last(*last_fit), timeline.steps_after_last(*first)) {
+            (Some(b), Some(a)) => ((b - a).round() + 1.0).clamp(0.0, 1e9) as usize,
+            _ => fit,
+        },
+        _ => fit,
+    };
+    let cap = max_periods_from(rows.len(), fit, slots).ok_or_else(|| {
+        EngineError::Other("The forecast start is too early for this much history: choose a later forecast start or less history.".into())
+    })?;
     let periods = match end {
         Some(end) => steps_to(&timeline, end, cap)?,
         None => match number(p, "periods")? {
             Some(n) if n >= 1.0 && n.ceil() <= f64::from(cap) => n.ceil() as u32,
             Some(_) => return Err(bad(CMD, format!("`periods` is from 1 to {cap} for {} rows of history", rows.len()))),
-            None => default_periods(rows.len()),
+            None => default_periods(rows.len()).min(cap),
         },
     };
     let future: Vec<f64> =
@@ -530,6 +555,51 @@ mod tests {
         assert_eq!(s.doc().unwrap().wb.sheets.len(), sheets - 1);
     }
 
+    #[test]
+    fn hindcast_rows_count_against_the_work_limit() {
+        use super::{MAX_WORK, max_periods, max_periods_from};
+        assert_eq!(max_periods_from(1_300, 1_300, 1_300), Some(max_periods(1_300)));
+        // 2,000 rows fit to the first 1,899: 101 hindcast rows, then 162 forecast rows.
+        assert_eq!(max_periods_from(2_000, 1_899, 1_899), Some(162));
+        // Fit to the first 1,000: the 1,000 hindcast rows alone are twice the limit.
+        assert_eq!(max_periods_from(2_000, 1_000, 1_000), None);
+        // 800 rows fit to the first 400, which span 571 slots (gaps filled): each formula refits
+        // 571 slots, so 475 forecast rows after the 400 hindcast rows, not 850.
+        assert_eq!(max_periods_from(800, 400, 571), Some(475));
+        assert!((475 + 400) * 571 <= MAX_WORK as usize);
+        for (h, f) in [(2_000usize, 1_899usize), (10_000, 9_990), (100, 2), (60, 50)] {
+            if let Some(p) = max_periods_from(h, f, f) {
+                assert!((u64::from(p) + (h - f) as u64) * f as u64 <= MAX_WORK, "{h} {f}");
+            }
+        }
+        let mut s = Session::new();
+        s.new_workbook();
+        let mut rows = vec![json!(["t", "y"])];
+        rows.extend((1..=2_000).map(|t| json!([t, (t % 7) as f64 + t as f64 * 0.5])));
+        s.execute("range.setValues", json!({"range": "A1", "values": rows})).unwrap();
+        let sheets = s.doc().unwrap().wb.sheets.len();
+        let err = s.execute("data.forecastSheet", json!({"range": "A1:B2001", "forecastStart": 1_001})).expect_err("hindcast too long");
+        assert!(err.to_string().contains("forecast start is too early"), "{err}");
+        let err = s.execute("data.forecastSheet", json!({"range": "A1:B2001", "forecastStart": 1_900, "periods": 163})).expect_err("past the cap");
+        assert!(err.to_string().contains("from 1 to 162"), "{err}");
+        assert_eq!(s.doc().unwrap().wb.sheets.len(), sheets);
+    }
+
+    #[test]
+    fn the_hindcast_cap_follows_the_slots_the_fit_spans_not_its_rows() {
+        // 800 rows, 20% of the slots missing: the first 400 rows span 500 slots, so each ETS
+        // formula refits 500 slots and the cap is 1,000 - 400 = 600 (it was 850 by rows).
+        let mut s = Session::new();
+        s.new_workbook();
+        let times: Vec<u32> = (1u32..).filter(|t| !matches!(t % 10, 3 | 6)).take(800).collect();
+        let mut rows = vec![json!(["t", "y"])];
+        rows.extend(times.iter().map(|t| json!([t, (t % 7) as f64 + f64::from(*t) * 0.5])));
+        s.execute("range.setValues", json!({"range": "A1", "values": rows})).unwrap();
+        let start = times[400];
+        let err = s.execute("data.forecastSheet", json!({"range": "A1:B801", "forecastStart": start, "periods": 601})).expect_err("past the cap");
+        assert!(err.to_string().contains("from 1 to 600"), "{err}");
+    }
+
     #[test]
     fn history_and_periods_are_capped() {
         use super::{MAX_HISTORY, MAX_PERIODS, MAX_WORK, default_periods, max_periods};

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,