spg_engine/aggregate.rs
1//! Aggregate executor.
2//!
3//! Handles `SELECT … <aggs> … [GROUP BY …]` queries. The planning strategy
4//! is straightforward:
5//!
6//! 1. Walk the SELECT (and ORDER BY) expressions to find every aggregate
7//! function call. Dedupe by AST equality and assign each `__agg_<i>`.
8//! 2. Same for every `GROUP BY` expression: assign `__grp_<j>`.
9//! 3. Stream the WHERE-filtered rows, group by the tuple of GROUP BY
10//! values, and update per-group aggregate state.
11//! 4. Materialise a synthetic per-group row containing
12//! `[__grp_0..__grp_K, __agg_0..__agg_N]` and rewrite the user's
13//! SELECT / ORDER BY expressions to reference those synthetic columns
14//! instead of the originals.
15//! 5. Evaluate the rewritten expressions against the synthetic schema and
16//! emit results.
17//!
18//! v1.8 implements `count(*)`, `count(expr)`, `sum`, `min`, `max`, `avg`.
19//! NULL semantics follow PG: aggregates skip NULL inputs (except
20//! `count(*)`, which counts rows). `sum(int)` widens to `BigInt`;
21//! `avg(int|bigint)` returns `Float`.
22
23use alloc::borrow::Cow;
24use alloc::boxed::Box;
25use alloc::collections::BTreeSet;
26use alloc::format;
27use alloc::string::{String, ToString};
28use alloc::vec::Vec;
29
30use spg_sql::ast::{Expr, SelectItem, SelectStatement};
31use spg_storage::{ColumnSchema, DataType, Row, Value};
32
33use crate::eval::{self, EvalContext, EvalError};
34use crate::join::RowRef;
35
36/// True if this statement should go through the aggregate path.
37pub fn uses_aggregate(stmt: &SelectStatement) -> bool {
38 if stmt.group_by.is_some() || stmt.having.is_some() {
39 return true;
40 }
41 for item in &stmt.items {
42 if let SelectItem::Expr { expr, .. } = item
43 && contains_aggregate(expr)
44 {
45 return true;
46 }
47 }
48 for o in &stmt.order_by {
49 if contains_aggregate(&o.expr) {
50 return true;
51 }
52 }
53 if let Some(h) = &stmt.having
54 && contains_aggregate(h)
55 {
56 return true;
57 }
58 false
59}
60
61pub fn contains_aggregate(e: &Expr) -> bool {
62 match e {
63 Expr::FunctionCall { name, args } => {
64 is_aggregate_name(name) || args.iter().any(contains_aggregate)
65 }
66 Expr::AggregateOrdered { .. } => true,
67 Expr::Binary { lhs, rhs, .. } => contains_aggregate(lhs) || contains_aggregate(rhs),
68 Expr::Unary { expr, .. } | Expr::Cast { expr, .. } | Expr::IsNull { expr, .. } => {
69 contains_aggregate(expr)
70 }
71 Expr::Like { expr, pattern, .. } => contains_aggregate(expr) || contains_aggregate(pattern),
72 Expr::Extract { source, .. } => contains_aggregate(source),
73 // v4.10 subqueries + v4.12 window functions / Literal /
74 // Column — all non-aggregate leaves from the regular
75 // aggregate planner's POV. Window-bearing projections are
76 // routed to exec_select_with_window before this runs.
77 Expr::ScalarSubquery(_)
78 | Expr::Exists { .. }
79 | Expr::InSubquery { .. }
80 | Expr::WindowFunction { .. }
81 | Expr::Literal(_)
82 | Expr::Placeholder(_)
83 | Expr::Column(_) => false,
84 // v7.10.10 — recurse into array constructor / subscript /
85 // ANY/ALL children. Aggregates inside `ARRAY[SUM(x)]` are
86 // valid PG and must be detected here.
87 Expr::Array(items) => items.iter().any(contains_aggregate),
88 Expr::ArraySubscript { target, index } => {
89 contains_aggregate(target) || contains_aggregate(index)
90 }
91 Expr::AnyAll { expr, array, .. } => contains_aggregate(expr) || contains_aggregate(array),
92 Expr::InList { expr, list, .. } => {
93 contains_aggregate(expr) || list.iter().any(contains_aggregate)
94 }
95 // v7.13.0 — CASE WHEN … END. Recurse into operand,
96 // every (WHEN, THEN) pair, and the ELSE branch.
97 Expr::Case {
98 operand,
99 branches,
100 else_branch,
101 } => {
102 operand.as_deref().is_some_and(contains_aggregate)
103 || branches
104 .iter()
105 .any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
106 || else_branch.as_deref().is_some_and(contains_aggregate)
107 }
108 }
109}
110
111pub fn is_aggregate_name(name: &str) -> bool {
112 matches!(
113 name.to_ascii_lowercase().as_str(),
114 "count"
115 | "count_star"
116 | "sum"
117 | "min"
118 | "max"
119 | "avg"
120 // v7.17.0 — variadic / collection aggregates. ORM
121 // reports (Hibernate / Rails / Django) emit these in
122 // GROUP BY rollups; pre-7.17 SPG hit "unknown
123 // aggregate".
124 | "string_agg"
125 | "array_agg"
126 // v7.17.0 — boolean aggregates. `every` is SQL-standard
127 // alias for `bool_and`.
128 | "bool_and"
129 | "bool_or"
130 | "every"
131 // v7.32 (round-29) — statistical aggregates (every BI /
132 // dashboard emits these in rollups).
133 | "stddev" | "stddev_samp" | "stddev_pop"
134 | "variance" | "var_samp" | "var_pop"
135 // v7.32 (round-29) — bitwise aggregates.
136 | "bit_and" | "bit_or" | "bit_xor"
137 // v7.32 (round-29) — ordered-set aggregates (used with
138 // `WITHIN GROUP (ORDER BY …)`).
139 | "percentile_cont" | "percentile_disc" | "mode"
140 // v7.32 (round-29) — hypothetical-set aggregates (also
141 // `WITHIN GROUP`): the rank the direct args WOULD have.
142 | "rank" | "dense_rank" | "percent_rank" | "cume_dist"
143 // v7.32 (round-29) — two-argument regression family.
144 | "covar_pop" | "covar_samp" | "corr"
145 | "regr_count" | "regr_avgx" | "regr_avgy" | "regr_slope"
146 | "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy"
147 // v7.32 (round-29) — JSON aggregates.
148 | "json_agg" | "jsonb_agg" | "json_object_agg" | "jsonb_object_agg"
149 )
150}
151
152/// v7.32 (round-29) — two-argument regression aggregates `f(Y, X)`.
153fn is_regression_name(name: &str) -> bool {
154 matches!(
155 name,
156 "covar_pop"
157 | "covar_samp"
158 | "corr"
159 | "regr_count"
160 | "regr_avgx"
161 | "regr_avgy"
162 | "regr_slope"
163 | "regr_intercept"
164 | "regr_r2"
165 | "regr_sxx"
166 | "regr_syy"
167 | "regr_sxy"
168 )
169}
170
171/// v7.32 (round-29) — aggregates that consume a second positional
172/// argument: `string_agg(v, sep)`, the regression family `f(Y, X)`, and
173/// `json_object_agg(key, value)`.
174fn agg_uses_second_arg(name: &str) -> bool {
175 name == "string_agg"
176 || name == "json_object_agg"
177 || name == "jsonb_object_agg"
178 || is_regression_name(name)
179}
180
181/// v7.32 (round-29) — ordered-set aggregates: the value to aggregate
182/// comes from the `WITHIN GROUP (ORDER BY …)` sort spec, and any
183/// in-parens arguments are *direct* arguments (the percentile fraction).
184/// `mode()` takes no direct argument.
185pub fn is_ordered_set_name(name: &str) -> bool {
186 // v7.32 — `eq_ignore_ascii_case` instead of `to_ascii_lowercase()`:
187 // these classifiers run in the aggregate row/group loop, where the
188 // old per-call `String` allocation showed up as ~16% of the inbox's
189 // aggregate path in a sampled profile (the names are constant).
190 ["percentile_cont", "percentile_disc", "mode"]
191 .iter()
192 .any(|k| name.eq_ignore_ascii_case(k))
193}
194
195/// v7.32 (round-29) — hypothetical-set aggregates: `rank(args) WITHIN
196/// GROUP (ORDER BY …)` and friends compute the rank the hypothetical
197/// row would have. Like ordered-set, the value stream comes from the
198/// sort spec and the in-parens args are direct (the hypothetical row).
199pub fn is_hypothetical_set_name(name: &str) -> bool {
200 ["rank", "dense_rank", "percent_rank", "cume_dist"]
201 .iter()
202 .any(|k| name.eq_ignore_ascii_case(k))
203}
204
205/// v7.32 (round-29) — every aggregate that takes its value stream from
206/// a `WITHIN GROUP (ORDER BY …)` clause (ordered-set + hypothetical-set).
207pub fn is_within_group_name(name: &str) -> bool {
208 is_ordered_set_name(name) || is_hypothetical_set_name(name)
209}
210
211/// v7.37.4 (R34) — pre-computed aggregate kind. Replaces per-row
212/// string matches in `update_state` with a single `match` on a
213/// `Copy` enum (compiles to a jump table). For the mailrs prod
214/// `/api/conversations` shape (14 aggregates × 100 k rows = 1.4 M
215/// inner-loop iterations) this is the dominant per-row cost.
216///
217/// Lowered from `AggSpec::name` at spec build time via
218/// [`classify_agg_name`]; populated by the three `AggSpec`
219/// construction sites (window+ORDER, plain, `first_ordered`
220/// `array_agg`).
221#[derive(Copy, Clone, Debug, PartialEq, Eq)]
222enum AggKind {
223 CountStar,
224 Count,
225 Sum,
226 Avg,
227 Min,
228 Max,
229 StringAgg,
230 ArrayAgg,
231 BoolAnd,
232 BoolOr,
233 /// stddev / stddev_samp / stddev_pop / variance / var_samp / var_pop.
234 StddevFamily,
235 BitAnd,
236 BitOr,
237 BitXor,
238 /// ordered-set (`percentile_cont/disc`, `mode`) +
239 /// hypothetical-set (`rank`/`dense_rank`/etc.) aggregates that
240 /// share the WITHIN-GROUP collection path.
241 WithinGroup,
242 /// covar_samp / covar_pop / corr / regr_*.
243 Regression,
244 JsonAgg,
245 JsonObjectAgg,
246}
247
248/// v7.37.4 (R34) — name → kind, called once per spec at build time.
249/// Hot path (`update_state_kind`) only sees the enum; the canonical
250/// string still travels with the spec so `finalize` and errors can
251/// quote it.
252fn classify_agg_name(name: &str) -> AggKind {
253 match name {
254 "count_star" => AggKind::CountStar,
255 "count" => AggKind::Count,
256 "sum" => AggKind::Sum,
257 "avg" => AggKind::Avg,
258 "min" => AggKind::Min,
259 "max" => AggKind::Max,
260 "string_agg" => AggKind::StringAgg,
261 "array_agg" => AggKind::ArrayAgg,
262 "bool_and" => AggKind::BoolAnd,
263 "bool_or" => AggKind::BoolOr,
264 "stddev" | "stddev_samp" | "stddev_pop" | "variance" | "var_samp" | "var_pop" => {
265 AggKind::StddevFamily
266 }
267 "bit_and" => AggKind::BitAnd,
268 "bit_or" => AggKind::BitOr,
269 "bit_xor" => AggKind::BitXor,
270 "json_agg" | "jsonb_agg" => AggKind::JsonAgg,
271 "json_object_agg" | "jsonb_object_agg" => AggKind::JsonObjectAgg,
272 n if is_within_group_name(n) => AggKind::WithinGroup,
273 n if is_regression_name(n) => AggKind::Regression,
274 other => panic!("classify_agg_name: unknown aggregate {other}"),
275 }
276}
277
278/// Per-aggregate running state.
279#[derive(Debug, Default, Clone)]
280struct AggState {
281 count: i64,
282 sum_int: i64,
283 sum_float: f64,
284 extreme: Option<Value<'static>>,
285 use_float: bool,
286 /// v7.17.0 — running collection for string_agg / array_agg.
287 /// Each entry is one row's contribution (NULL preserved as
288 /// `Value::Null`; string_agg's finalize step drops them, but
289 /// array_agg keeps them). Pushing in insertion order matches
290 /// PG behaviour when no `ORDER BY` is given inside the
291 /// aggregate call.
292 items: Vec<Value<'static>>,
293 /// v7.25 (round-17) — per-group dedupe set for DISTINCT
294 /// aggregates (encoded values; NULLs never reach it because
295 /// the caller's skip runs after the per-aggregate NULL rules).
296 /// v7.37.4 measured `hashbrown::HashSet` as worse at this
297 /// shape — the per-(group × distinct-spec) hash table alloc
298 /// overhead beats the lookup-speed gain when each set is
299 /// small. Sticking with `BTreeSet`; the dispatch-side enum
300 /// fix in `update_state` is the R34 win.
301 seen: BTreeSet<String>,
302 /// v7.37.x (docker-fair DISTA attack) — fast-path BigInt seen
303 /// set. The hot DISTINCT path used `encode_key_refs_into` to
304 /// turn `Value::BigInt(n)` into a string key like `"I<n>|"` then
305 /// inserted that into the String BTreeSet — ~100 ns of pure alloc
306 /// + format churn per row × 25 k rows × 1 BigInt DISTINCT spec
307 /// (the DISTA `COUNT(DISTINCT m.id)` shape) ≈ 2.5 ms of waste.
308 /// Direct `BTreeSet<i64>` skips encode entirely; lookups stay
309 /// O(log small) on the per-group set. Lazy-allocated — only the
310 /// BigInt-DISTINCT path constructs it.
311 seen_int: Option<BTreeSet<i64>>,
312 /// v7.24 (round-16 A) — per-item ORDER BY key tuples, parallel
313 /// to `items` (pushed under the same skip/keep conditions).
314 /// Empty when the aggregate carries no internal ordering.
315 item_keys: Vec<Vec<Value<'static>>>,
316 /// v7.17.0 — captured separator for string_agg. PG accepts a
317 /// non-constant separator expression but in practice every
318 /// caller passes a literal; the engine snapshots the last
319 /// non-NULL text it sees, which matches PG's "use the latest
320 /// row's value" behaviour.
321 separator: Option<String>,
322 /// v7.17.0 — running boolean accumulator for bool_and /
323 /// bool_or / every. `None` until the first non-NULL input;
324 /// at finalize None → SQL NULL.
325 bool_acc: Option<bool>,
326 /// v7.32 (round-29) — sum of squares for the variance / stddev
327 /// family (`sum_float` carries the running sum; `count` the n).
328 sum_sq: f64,
329 /// v7.32 (round-29) — running accumulator for bit_and / bit_or /
330 /// bit_xor. `None` until the first non-NULL input → SQL NULL.
331 bit_acc: Option<i64>,
332 /// v7.32 (round-29) — two-argument regression family
333 /// (`covar_*` / `corr` / `regr_*`), PG arg order `f(Y, X)`. Only
334 /// rows where BOTH inputs are non-NULL contribute (`count` is the
335 /// paired n, independent of the single-arg `sum_*`).
336 reg_n: i64,
337 reg_sx: f64,
338 reg_sy: f64,
339 reg_sxx: f64,
340 reg_syy: f64,
341 reg_sxy: f64,
342 /// v7.32 (round-29) — second value stream for `json_object_agg`
343 /// (`items` holds the keys, `aux_items` the values).
344 aux_items: Vec<Value<'static>>,
345 /// v7.33 (array_agg argmax) — for a `first_ordered` spec
346 /// (`(array_agg(x ORDER BY y))[1]`), the running first-by-order
347 /// (sort-key tuple, value). Replaced only when a new row's key sorts
348 /// strictly before the current best (ties keep the earliest row, =
349 /// the stable-sort `[1]`). No items/item_keys array is built.
350 first_best: Option<(Vec<Value<'static>>, Value<'static>)>,
351}
352
353#[derive(Debug, Clone)]
354struct AggSpec {
355 name: String, // lowercased
356 /// First argument (value expression) for every aggregate
357 /// except `count(*)`. `None` for `count_star`.
358 arg: Option<Expr>,
359 /// v7.17.0 — second argument. Only `string_agg(value, sep)`
360 /// uses it today. `None` for every other aggregate (or for
361 /// `array_agg`, which is single-arg). Carried in the spec so
362 /// per-row evaluation can re-use the same separator
363 /// expression across calls.
364 arg2: Option<Expr>,
365 /// v7.25 (round-17) — `COUNT(DISTINCT x)` & friends: dedupe
366 /// the input stream per group before accumulation.
367 distinct: bool,
368 /// v7.24 (round-16 A) — aggregate-internal ORDER BY keys
369 /// (`array_agg(x ORDER BY y DESC NULLS LAST)`). Empty for the
370 /// plain form. Only the collection aggregates honour it;
371 /// other aggregates are order-insensitive and ignore it (PG
372 /// accepts the syntax everywhere too).
373 order_by: Vec<spg_sql::ast::OrderBy>,
374 /// v7.32 (round-29) — `FILTER (WHERE cond)`: a per-row predicate
375 /// evaluated against the source row before accumulation. A row
376 /// whose `cond` is not TRUE (false or NULL) is excluded from this
377 /// aggregate only. `None` for the unfiltered form.
378 filter: Option<Expr>,
379 /// v7.32 (round-29) — ordered-set aggregates only: the *direct*
380 /// argument (the percentile fraction for `percentile_cont/disc`).
381 /// PG requires it constant, so it is evaluated once. `None` for
382 /// `mode()` and for every non-ordered-set aggregate.
383 direct_arg: Option<Expr>,
384 /// v7.33 (array_agg argmax) — set when this spec came from
385 /// `(array_agg(x ORDER BY y))[1]`: accumulate only the first-by-order
386 /// element (a running argmax/argmin) and finalise to that scalar
387 /// value, instead of collecting + sorting + materialising the whole
388 /// per-group array just to take element 1. Returns the element type,
389 /// not the array type.
390 first_ordered: bool,
391 /// v7.37.4 (R34) — derived from `name` at spec build time so the
392 /// per-row inner loop dispatches via a `match` on `Copy` enum
393 /// instead of a string compare for every (row × aggregate)
394 /// iteration.
395 kind: AggKind,
396}
397
398/// Output of running the aggregate path. Schema describes one row per
399/// group; rows are not yet ORDER BY-sorted (caller does it).
400#[derive(Debug)]
401pub struct AggResult {
402 pub columns: Vec<ColumnSchema>,
403 pub rows: Vec<Row<'static>>,
404 /// v7.31 (perf — PG lesson #1, post-LIMIT subquery projection):
405 /// select-list items whose rewritten expr carries a subquery and
406 /// is referenced by neither ORDER BY nor HAVING. Their output
407 /// cells hold NULL placeholders; the caller truncates to
408 /// LIMIT+OFFSET first and only then evaluates these for the
409 /// surviving rows (PG runs the same shape with SubPlan loops=50
410 /// instead of loops=24000). `(output_col, rewritten_expr)`.
411 pub deferred: Vec<(usize, Expr)>,
412 /// Synthetic group rows aligned 1:1 with `rows`; populated only
413 /// when `deferred` is non-empty.
414 pub synth_rows: Vec<Row<'static>>,
415 /// Schema the deferred exprs evaluate against.
416 pub synth_schema: Vec<ColumnSchema>,
417}
418
419/// Execute aggregate logic against an already-WHERE-filtered iterator of
420/// rows. `table_alias` is the alias accepted by column resolution.
421#[allow(clippy::too_many_lines)]
422/// v7.25.2 (round-19 A) — caller-injected evaluator for synth-row
423/// expressions that still carry subquery nodes after the rewrite
424/// (correlated subqueries in the select list / HAVING / aggregate
425/// ORDER BY of a GROUP BY query). The engine passes its
426/// correlated-aware evaluator; pure-library callers pass None and
427/// surviving subqueries keep erroring loudly.
428pub type CorrelatedEval<'a> =
429 &'a dyn Fn(&Expr, &Row<'static>, &EvalContext<'_>) -> Result<Value<'static>, EvalError>;
430
431/// Output of the per-group projection stage (`project_groups`): the
432/// output schema, the projected rows, the synth rows kept alongside
433/// them for post-LIMIT deferred evaluation, the deferred subquery
434/// items, and the rewritten ORDER BY exprs (shared with the sort).
435struct Projection {
436 columns: Vec<ColumnSchema>,
437 out_rows: Vec<Row<'static>>,
438 kept_synth: Vec<Row<'static>>,
439 deferred: Vec<(usize, Expr)>,
440 order_rewritten: Vec<Expr>,
441 /// v7.37.x — when `defer_projection` is requested, `out_rows`
442 /// carries empty placeholders and the caller runs the per-item
443 /// eval pass after sort+truncate over the surviving ≤ keep_n
444 /// rows. `None` when projection was performed inline.
445 deferred_project: Option<DeferredProject>,
446}
447
448struct DeferredProject {
449 items_rewritten: Vec<Option<Expr>>,
450 items_compiled: Vec<Option<eval::CompiledExpr>>,
451}
452
453/// v7.35.0 — detect the `SELECT COUNT(*) FROM … [WHERE …]` shape
454/// (single item, no GROUP BY / HAVING / ORDER BY / DISTINCT /
455/// LIMIT WITH TIES / FILTER / window). For this shape the answer
456/// is exactly `rows.len()` as `BigInt`, no group state needed.
457/// Returns `None` for any deviation so the caller's full pipeline
458/// runs verbatim.
459///
460/// v7.35.2 — also short-circuit `COUNT(<literal>)` (e.g.
461/// `COUNT(1)`) and `COUNT(<column>)` when the column is declared
462/// NOT NULL on the input schema. PG handles both cases as
463/// `COUNT(*)` (the non-null filter is a no-op), so doing the same
464/// here keeps every `count this thing` shape on the same fast path
465/// instead of routing the literal / non-null-col variants through
466/// the four-stage aggregate pipeline.
467fn try_pure_count_star_short_circuit(
468 stmt: &SelectStatement,
469 rows: &[RowRef<'_>],
470 schema_cols: &[ColumnSchema],
471 table_alias: Option<&str>,
472) -> Option<AggResult> {
473 if stmt.distinct
474 || stmt.limit_with_ties
475 || stmt.group_by.is_some()
476 || stmt.having.is_some()
477 || !stmt.order_by.is_empty()
478 {
479 return None;
480 }
481 if stmt.items.len() != 1 {
482 return None;
483 }
484 let SelectItem::Expr { expr, alias } = &stmt.items[0] else {
485 return None;
486 };
487 let Expr::FunctionCall { name, args } = expr else {
488 return None;
489 };
490 if !name.eq_ignore_ascii_case("count") && !name.eq_ignore_ascii_case("count_star") {
491 return None;
492 }
493 let count_star_shape = match args.as_slice() {
494 // `COUNT(*)` parses to `count_star` with no args.
495 [] if name.eq_ignore_ascii_case("count_star") => true,
496 // `COUNT(<literal>)` — the per-row test is "is this literal
497 // non-null?" which is constant, so it's COUNT(*) when the
498 // literal is non-null.
499 [Expr::Literal(lit)] => !matches!(lit, spg_sql::ast::Literal::Null),
500 // `COUNT(<column>)` — same answer as COUNT(*) when the
501 // column is statically declared NOT NULL on the input
502 // schema. Resolve through the alias if one is set.
503 [Expr::Column(c)] => {
504 if let Some(q) = c.qualifier.as_deref()
505 && let Some(alias) = table_alias
506 && !q.eq_ignore_ascii_case(alias)
507 {
508 return None;
509 }
510 schema_cols
511 .iter()
512 .find(|s| s.name.eq_ignore_ascii_case(&c.name))
513 .is_some_and(|s| !s.nullable)
514 }
515 _ => return None,
516 };
517 if !count_star_shape {
518 return None;
519 }
520 let col_name = alias.clone().unwrap_or_else(|| "count".to_string());
521 let count = i64::try_from(rows.len()).unwrap_or(i64::MAX);
522 Some(AggResult {
523 columns: alloc::vec![ColumnSchema::new(col_name, DataType::BigInt, false)],
524 rows: alloc::vec![Row::new(alloc::vec![Value::BigInt(count)])],
525 deferred: Vec::new(),
526 synth_rows: Vec::new(),
527 synth_schema: Vec::new(),
528 })
529}
530
531pub(crate) fn run(
532 stmt: &SelectStatement,
533 rows: &[RowRef<'_>],
534 schema_cols: &[ColumnSchema],
535 table_alias: Option<&str>,
536 correlated_eval: Option<CorrelatedEval<'_>>,
537) -> Result<AggResult, EvalError> {
538 // v7.38 P0 元机制 A — fires at the top of the aggregate
539 // executor with the number of input rows. Tests use this to
540 // block before a hypothetical spill decision; in release it
541 // expands to `let _ = (...);`.
542 let __spg_row_count = rows.len();
543 crate::injection_point!("aggregate_spill_trigger", &__spg_row_count);
544 // v7.35.0 — pure `SELECT COUNT(*) FROM … WHERE …` short-circuit.
545 // The caller already filtered rows by WHERE (we run on the
546 // post-WHERE survivor set), so for the canonical pure-COUNT(*)
547 // shape (no GROUP BY / HAVING / ORDER BY / DISTINCT / FILTER /
548 // window) the answer is simply `rows.len()`. The four-stage
549 // aggregate pipeline below (accumulate_groups → build_synth_schema
550 // → finalize_synth_rows → project_groups) collapses to a single
551 // BigInt cell when there's a single group, but each stage still
552 // pays its own allocation tax — group state map, synth schema
553 // vec, finalize loop. `exists_in_60` (mailrs prod #4 baseline)
554 // is exactly this shape on a 25 k-row JOIN.
555 if let Some(short) = try_pure_count_star_short_circuit(stmt, rows, schema_cols, table_alias) {
556 return Ok(short);
557 }
558 let group_exprs: Vec<Expr> = stmt.group_by.clone().unwrap_or_default();
559
560 // Collect aggregate sub-expressions across items + order_by.
561 let mut agg_specs: Vec<AggSpec> = Vec::new();
562 for item in &stmt.items {
563 if let SelectItem::Expr { expr, .. } = item {
564 collect_aggregates(expr, &mut agg_specs);
565 }
566 }
567 for o in &stmt.order_by {
568 collect_aggregates(&o.expr, &mut agg_specs);
569 }
570 if let Some(h) = &stmt.having {
571 collect_aggregates(h, &mut agg_specs);
572 }
573 // v7.17.0 — arity validation. The collector tolerates an
574 // arbitrary positional-arg count; here we enforce the
575 // per-aggregate contract so a malformed call (e.g.
576 // `array_agg()` or `string_agg(x)`) surfaces as a SQL error
577 // rather than silently coercing to a degenerate aggregate.
578 validate_agg_arities(stmt, &agg_specs)?;
579 validate_within_group(&agg_specs)?;
580
581 // (1) Stream the WHERE-filtered rows into insertion-ordered group state.
582 let order = accumulate_groups(
583 rows,
584 &group_exprs,
585 &agg_specs,
586 schema_cols,
587 table_alias,
588 correlated_eval,
589 )?;
590
591 // (2) Build the synthetic per-group schema and finalise each group's row.
592 let synth_schema =
593 build_synth_schema(rows, &group_exprs, &agg_specs, schema_cols, table_alias)?;
594 let synth_rows = finalize_synth_rows(
595 &order,
596 &agg_specs,
597 &synth_schema,
598 rows,
599 schema_cols,
600 table_alias,
601 )?;
602
603 // v7.37.x (mailrs Track A 100k attack) — defer the bound
604 // per-item SELECT projection on the synth rows until AFTER
605 // sort + LIMIT truncation. On a `GROUP BY t ORDER BY agg DESC
606 // LIMIT 50` with 20 000 groups (the mailrs minimal 100k shape)
607 // pre-defer ran 20 000 × N_items compiled-VM evals + Row
608 // allocations before discarding 99.75 % at the sort truncation
609 // step. HAVING still runs inline on every group because it
610 // filters BEFORE the LIMIT; we only skip the SELECT-list eval.
611 let defer_projection = !stmt.order_by.is_empty()
612 && !stmt.distinct
613 && !stmt.limit_with_ties
614 && stmt.having.is_none()
615 && stmt.limit_literal().is_some_and(|l| {
616 let off = stmt.offset_literal().unwrap_or(0) as usize;
617 let k = (l as usize).saturating_add(off);
618 k > 0 && k < synth_rows.len()
619 });
620
621 // (3) Rewrite the user's expressions, filter groups by HAVING and project.
622 let Projection {
623 columns,
624 mut out_rows,
625 mut kept_synth,
626 deferred,
627 order_rewritten,
628 deferred_project,
629 } = project_groups(
630 synth_rows,
631 stmt,
632 &group_exprs,
633 &agg_specs,
634 &synth_schema,
635 correlated_eval,
636 defer_projection,
637 )?;
638
639 // (4) ORDER BY on the aggregated output (the caller applies LIMIT).
640 //
641 // v7.37.3 (mailrs prod /api/contacts 3.21× regression — and the
642 // general inbox-listing-shape SPG-vs-PG gap) — top-K sink for
643 // `ORDER BY <agg> [DESC] LIMIT k`. Pre-7.37.3 this stage ran a
644 // full O(N log N) sort over every surviving group, then the
645 // caller truncated to `k`. With high-cardinality GROUP BY (a
646 // sender column with hundreds-thousands of distinct values) the
647 // truncated set is a tiny fraction of `N` — keep an O(k) top-K
648 // sink and never sort the discarded majority. Matches PG /
649 // MySQL / MariaDB's standard "LIMIT k under ORDER BY agg"
650 // optimisation; SPG previously implemented it only on the
651 // streamed inner-join path (`try_streamed_inner_join_topn`)
652 // and not on the aggregate output.
653 //
654 // Gate: needs a literal LIMIT (placeholder LIMIT we can't bound
655 // statically here), no DISTINCT (would need post-dedup, can't
656 // truncate during sort), no LIMIT WITH TIES (which extends past
657 // the literal k by run-time tie-key comparison).
658 let keep_n: Option<usize> =
659 if !stmt.order_by.is_empty() && !stmt.distinct && !stmt.limit_with_ties {
660 stmt.limit_literal().map(|l| {
661 let off = stmt.offset_literal().unwrap_or(0) as usize;
662 (l as usize).saturating_add(off)
663 })
664 } else {
665 None
666 };
667 if !stmt.order_by.is_empty() {
668 let (sorted_synth, sorted_out) = sort_synth_by_order_by(
669 &synth_schema,
670 &stmt.order_by,
671 &order_rewritten,
672 kept_synth,
673 out_rows,
674 correlated_eval,
675 keep_n,
676 )?;
677 kept_synth = sorted_synth;
678 out_rows = sorted_out;
679 }
680
681 // v7.37.x — run deferred SELECT-list projection on the truncated
682 // top-K survivors. For `GROUP BY thread_id ORDER BY MAX(date) DESC
683 // LIMIT 50` against 20 000 groups, this turns ~40 000 compiled-VM
684 // evals + Row allocations into 100, saving ~2-3 ms on the mailrs
685 // minimal 100k shape.
686 if let Some(DeferredProject {
687 items_rewritten,
688 items_compiled,
689 }) = deferred_project
690 {
691 let synth_ctx = EvalContext::new(&synth_schema, None);
692 let mut stack: Vec<Value<'static>> = Vec::new();
693 for (idx, srow) in kept_synth.iter().enumerate() {
694 let mut values: Vec<Value<'static>> = Vec::with_capacity(columns.len());
695 for (i, rewritten) in items_rewritten.iter().enumerate() {
696 let Some(rewritten) = rewritten else { continue };
697 if deferred.iter().any(|(c, _)| *c == i) {
698 values.push(Value::Null);
699 continue;
700 }
701 values.push(if let Some(cc) = &items_compiled[i] {
702 eval::eval_compiled(cc, srow, &synth_ctx, &mut stack)?
703 } else {
704 match correlated_eval {
705 Some(f) if crate::expr_has_subquery(rewritten) => {
706 f(rewritten, srow, &synth_ctx)?
707 }
708 _ => eval::eval_expr(rewritten, srow, &synth_ctx)?,
709 }
710 });
711 }
712 out_rows[idx] = Row::new(values);
713 }
714 }
715
716 let (synth_rows_out, synth_schema_out) = if deferred.is_empty() {
717 (Vec::new(), Vec::new())
718 } else {
719 (kept_synth, synth_schema.clone())
720 };
721 Ok(AggResult {
722 columns,
723 rows: out_rows,
724 deferred,
725 synth_rows: synth_rows_out,
726 synth_schema: synth_schema_out,
727 })
728}
729
730/// v7.32 (round-29) — validate the structural requirements of WITHIN
731/// GROUP (ordered-set / hypothetical-set) aggregates up front, so a
732/// malformed call surfaces as a SQL error rather than a silently
733/// degenerate aggregate.
734fn validate_within_group(agg_specs: &[AggSpec]) -> Result<(), EvalError> {
735 // v7.32 (round-29) — WITHIN GROUP aggregates require the clause (PG
736 // raises a hard error otherwise rather than silently degrading), and
737 // SPG supports the single-sort-key form only.
738 for spec in agg_specs {
739 if is_within_group_name(&spec.name) {
740 if spec.order_by.is_empty() {
741 return Err(EvalError::TypeMismatch {
742 detail: format!("{}() requires WITHIN GROUP (ORDER BY …)", spec.name),
743 });
744 }
745 // mode() is the only WITHIN GROUP aggregate with no direct
746 // argument; the rest carry one (percentile fraction /
747 // hypothetical value).
748 if spec.name != "mode" && spec.direct_arg.is_none() {
749 return Err(EvalError::TypeMismatch {
750 detail: format!("{}() requires a direct argument", spec.name),
751 });
752 }
753 // Multi-key WITHIN GROUP (multiple sort keys / hypothetical
754 // args) is not supported yet — error loudly instead of
755 // silently using only the first key.
756 if spec.order_by.len() > 1 {
757 return Err(EvalError::TypeMismatch {
758 detail: format!(
759 "{}() with multiple WITHIN GROUP sort keys is not supported yet",
760 spec.name
761 ),
762 });
763 }
764 }
765 }
766 Ok(())
767}
768
769/// (1) Stream the WHERE-filtered rows, group by the GROUP BY value
770/// tuple, and update per-group aggregate state. Returns the groups in
771/// insertion order. See `run` for the bind-once fast path rationale.
772#[allow(clippy::too_many_lines, clippy::type_complexity)]
773fn accumulate_groups(
774 rows: &[RowRef<'_>],
775 group_exprs: &[Expr],
776 agg_specs: &[AggSpec],
777 schema_cols: &[ColumnSchema],
778 table_alias: Option<&str>,
779 correlated_eval: Option<CorrelatedEval<'_>>,
780) -> Result<Vec<(Vec<Value<'static>>, Vec<AggState>)>, EvalError> {
781 let ctx = EvalContext::new(schema_cols, table_alias);
782 // Map group key (vec of values, encoded as canonical string) -> group state.
783 // v7.32 (architecture v2, P2b) — insertion-ordered group state in
784 // a Vec; the hash map only maps key → index. Removes the parallel
785 // `key_order: Vec<String>` (a second per-group key clone) and the
786 // per-group re-probe `groups[k]` at finalize (24k hash lookups for
787 // the inbox shape). The map owns its key once on vacant insert.
788 let mut order: Vec<(Vec<Value<'static>>, Vec<AggState>)> = Vec::new();
789 let mut groups: hashbrown::HashMap<String, usize> = hashbrown::HashMap::new();
790 // v7.37.x (mailrs Track A perf — SPGE ≫ PG18) — single-Text GROUP
791 // BY column fast path. The canonical-string encode (`S<text>|`)
792 // + `encode_key_refs_into` reuse-buffer churn dominated the 30 k-
793 // row mailrs minimal probe (~3-4 ms / 30 k). For `GROUP BY t` on
794 // a TEXT column (the inbox-listing / conversation-grouping shape)
795 // the column text IS the canonical key — no encoder, no prefix
796 // byte, no `refs` Vec rebuild per row. The fallback `groups` map
797 // above is retained for multi-col / non-Text / collation paths;
798 // this map only fires when the schema and value structurally
799 // permit it. `null_group_idx` collects NULL group rows (SQL groups
800 // all NULLs into one bucket).
801 let mut groups_text: hashbrown::HashMap<String, usize> = hashbrown::HashMap::new();
802 let mut null_group_idx: Option<usize> = None;
803 // When there are no GROUP BY exprs *and* there is at least one aggregate,
804 // every row collapses into a single anonymous group keyed by "".
805 if rows.is_empty() && group_exprs.is_empty() {
806 // Single empty-aggregate group: count=0, sum=0, max=NULL, etc.
807 // No rows follow, so the map is never probed — seed `order` only.
808 let init: Vec<AggState> = (0..agg_specs.len()).map(|_| AggState::default()).collect();
809 order.push((Vec::new(), init));
810 }
811
812 // v7.30 (perf campaign) - hoist the per-row work that doesn't
813 // depend on the row: which group exprs need collation folding
814 // (none, for most queries - the old code cloned the whole
815 // group_vals vec per row just in case).
816 // v7.30 (perf campaign) - the no-tax row loop. When a group
817 // expr or an aggregate argument is a bare column reference
818 // (the overwhelmingly common shape), bind its position ONCE
819 // and read row cells by offset in the loop - no per-row tree
820 // walk, no owned-Value clone out of resolve_column. Anything
821 // more complex keeps the eval path.
822 let col_pos = |e: &Expr| -> Option<usize> {
823 // Qualified references only: the bare-name resolver carries
824 // alias/ambiguity logic the bind-once path must not fork.
825 if let Expr::Column(c) = e
826 && c.qualifier.is_some()
827 {
828 eval::find_column_pos(c, &ctx)
829 } else {
830 None
831 }
832 };
833 let group_pos: Vec<Option<usize>> = group_exprs.iter().map(col_pos).collect();
834 let all_groups_bound = group_pos.iter().all(Option::is_some);
835 // v7.37.x — single-col GROUP BY on a TEXT-typed column lets the
836 // hot loop key the hash map by the column text directly. Resolved
837 // once from the bound position against `schema_cols`.
838 let single_text_group_col: bool = group_pos.len() == 1
839 && group_pos[0].is_some_and(|p| {
840 schema_cols
841 .get(p)
842 .is_some_and(|c| matches!(c.ty, spg_storage::DataType::Text))
843 });
844 let arg_pos: Vec<Option<usize>> = agg_specs
845 .iter()
846 .map(|spec| spec.arg.as_ref().and_then(|e| col_pos(e)))
847 .collect();
848 // v7.37.x (mailrs Track A 100k attack) — dedicated tight loop
849 // for the "single-Text GROUP BY + single MAX(bound numeric arg)"
850 // shape. This is the mailrs `/api/conversations` minimal shape
851 // (`GROUP BY thread_id, MAX(internal_date)`) and an inbox-listing
852 // staple across the SPG customer set. Skipping the per-row spec
853 // loop, FILTER / arg2 / order_keys checks, and the union-typed
854 // `update_state` enum jump saves ~80-100 ns/row at 100 k input
855 // — the gap closing the SPGE vs PG18 ratio at this scale.
856 let dedicated_max_loop: bool = single_text_group_col
857 && agg_specs.len() == 1
858 && matches!(agg_specs[0].kind, AggKind::Max)
859 && agg_specs[0].filter.is_none()
860 && agg_specs[0].arg2.is_none()
861 && agg_specs[0].order_by.is_empty()
862 && !agg_specs[0].distinct
863 && !agg_specs[0].first_ordered
864 && arg_pos[0].is_some();
865 // v7.36 (perf — mailrs Ask 1 SUM(LENGTH(text_body)) 18ms → ?) —
866 // pre-compile every aggregate arg that's a `fully_compilable`
867 // PURE expression over bound columns. Without this, `LENGTH(col)`
868 // / `COALESCE(col, '')` / `CAST(col AS BIGINT)` etc. ALL fell
869 // through to the `(None, Some(e)) => eval_arg(e, mat, ...)` slow
870 // path that materialises a Cow<Row> per input row — for a 25k-row
871 // JOIN that's 25k full-row clones for one column read. The Step
872 // VM (`eval_compiled_ref`) reads columns by RowRef::get and runs
873 // the same `apply_function` dispatcher with zero materialisation.
874 let arg_compiled: Vec<Option<eval::CompiledExpr>> = agg_specs
875 .iter()
876 .enumerate()
877 .map(|(i, spec)| match (&arg_pos[i], &spec.arg) {
878 (Some(_), _) => None,
879 (None, Some(e)) if eval::fully_compilable(e) => Some(eval::compile_expr(e, &ctx)),
880 _ => None,
881 })
882 .collect();
883 // v7.37.4 (L1 — executor-time CSE / mailrs P0) — dedupe
884 // compiled aggregate-arg expressions across specs. mailrs's
885 // `/api/conversations` SQL has 14 aggregates whose compiled
886 // CASE/CAST arg expressions overlap heavily (`m.message_id != ''`
887 // re-appears 4×, the inner `CASE WHEN m.message_id != '' THEN
888 // m.message_id ELSE CAST(m.id AS TEXT) END` re-appears 3×). Each
889 // dup currently costs one Step-VM walk per row — 100k rows ×
890 // ~3-4 redundant evals = ~300-400k wasted Step-VM runs.
891 //
892 // Dedupe key = source `Expr` (PartialEq). `CompiledExpr` itself
893 // is not `Hash` / `Eq`, but n_specs is small (≤ ~20 in practice);
894 // O(n²) PartialEq probe cost = ~196 cmp per query, vs millions
895 // of saved per-row evals. `fully_compilable` requires PURE
896 // scalars (no NOW / RANDOM / sequence accessors), so an earlier
897 // eval has identical observable semantics to the original.
898 //
899 // `arg_slot[i] = Some(s)` means spec `i`'s compiled arg lives in
900 // slot `s` of `arg_unique_idx` (which points back into
901 // `arg_compiled` for the canonical owner). Per-row cache fills
902 // LAZILY — preserves the current FILTER semantics where an arg
903 // whose spec is filtered out is never evaluated (and never
904 // surfaces a type error). Reset to `None` at the top of each row.
905 let mut arg_unique_idx: Vec<usize> = Vec::new();
906 let mut arg_slot: Vec<Option<usize>> = Vec::with_capacity(agg_specs.len());
907 arg_slot.resize(agg_specs.len(), None);
908 for (i, spec) in agg_specs.iter().enumerate() {
909 if arg_pos[i].is_some() || arg_compiled[i].is_none() {
910 continue;
911 }
912 let src = spec.arg.as_ref().expect("arg_compiled => spec.arg is Some");
913 let pos = arg_unique_idx
914 .iter()
915 .position(|&j| agg_specs[j].arg.as_ref().is_some_and(|other| other == src));
916 arg_slot[i] = Some(match pos {
917 Some(p) => p,
918 None => {
919 arg_unique_idx.push(i);
920 arg_unique_idx.len() - 1
921 }
922 });
923 }
924 let mut row_eval_cache: Vec<Option<Value>> = Vec::with_capacity(arg_unique_idx.len());
925 row_eval_cache.resize(arg_unique_idx.len(), None);
926 // v7.33 (array_agg perf) — bound positions for each spec's internal
927 // ORDER BY keys, so an ordered aggregate (`array_agg(x ORDER BY y)`)
928 // reads the sort key by reference (RowRef::get) instead of
929 // materialising the whole combined join row per input row just to
930 // eval one bound column. Mirrors arg_pos. On the inbox shape this
931 // turned 24k full-row (~1 KB each) clones into 24k single-cell reads.
932 let order_pos: Vec<Vec<Option<usize>>> = agg_specs
933 .iter()
934 .map(|spec| spec.order_by.iter().map(|o| col_pos(&o.expr)).collect())
935 .collect();
936 // v7.37.43 (DISTA A-3) — precompute the per-spec arg2 when it is a
937 // bare literal. `string_agg(DISTINCT col, ',')` and every other
938 // call with a constant separator goes through this path; PG evaluates
939 // arg2 as a Const once at plan time. SPG was paying a Cow row
940 // materialisation per input row purely so `eval_arg(literal, &row)`
941 // could run — but a literal doesn't read the row at all. Hoist the
942 // literal value into a per-query table; per-row arg2 just clones it.
943 //
944 // Sentinel: when arg2 is present but NOT a literal, the entry stays
945 // `None` and the per-row path still falls into the eval branch
946 // (which forces `needs_mat`).
947 let arg2_literal_val: Vec<Option<Value<'static>>> = agg_specs
948 .iter()
949 .map(|s| match &s.arg2 {
950 Some(Expr::Literal(l)) => Some(eval::literal_to_value(l)),
951 _ => None,
952 })
953 .collect();
954 // Does any spec need the fully-materialised row in the bound fast
955 // path — a FILTER, a non-bound value arg, a NON-LITERAL second arg,
956 // or a non-bound ORDER key? When false (every aggregate arg/key is a
957 // bound column — the inbox shape, and the DISTA shape after A-3)
958 // the bound fast path never materialises a row.
959 let needs_mat = agg_specs.iter().enumerate().any(|(i, s)| {
960 s.filter.is_some()
961 || (s.arg.is_some() && arg_pos[i].is_none() && arg_compiled[i].is_none())
962 || (s.arg2.is_some() && arg2_literal_val[i].is_none())
963 || order_pos[i].iter().any(Option::is_none)
964 });
965 let ci_positions: Vec<usize> = group_exprs
966 .iter()
967 .enumerate()
968 .filter(|(_, g)| {
969 matches!(
970 eval::column_collation(g, &ctx),
971 Some(spg_storage::Collation::CaseInsensitive)
972 )
973 })
974 .map(|(i, _)| i)
975 .collect();
976 // v7.31 (perf 3e) — per-row scratch buffers. The fast path used
977 // to allocate a key String (and a refs Vec) for EVERY row just
978 // to probe the group map; hits — the overwhelming case — now
979 // touch the allocator zero times.
980 let mut keybuf_s = String::new();
981 // v7.36 — reused Step VM eval stack for compiled aggregate args.
982 // v7.37.9 T3 S2 — elided lifetime so the Vec's `'val` binds to the
983 // row-borrow lifetime per call (`eval_compiled_ref<'row, 'val>` now
984 // requires `'row: 'val`). Caller-side Vec<Value<'_>> lets compiler
985 // infer the shortest lifetime that covers all calls.
986 let mut eval_stack: Vec<Value<'_>> = Vec::new();
987 let mut dkeybuf = String::new();
988 let mut refs: Vec<&Value> = Vec::with_capacity(group_pos.len());
989 // v7.32 (round-31) — an aggregate's argument / FILTER / second arg /
990 // ORDER key may itself be a *correlated* subquery, e.g.
991 // `MAX((SELECT i.v FROM inner i WHERE i.fk = o.id))`. A non-correlated
992 // subquery is pre-resolved to a literal before this loop, but a
993 // correlated one survives as a subquery node and must be evaluated per
994 // outer row through the correlated evaluator — the same hook the
995 // select-list / HAVING / ORDER finalisers already use below. Plain
996 // `eval_expr` would hit "subquery reached row eval".
997 //
998 // The `any_agg_subquery` gate is computed once here so the common case
999 // (no subquery anywhere in the aggregate args — including every hot
1000 // scan/group aggregate) short-circuits before the per-row
1001 // `expr_has_subquery` walk: `eval_arg` is then exactly `eval_expr`.
1002 let any_agg_subquery = correlated_eval.is_some()
1003 && agg_specs.iter().any(|s| {
1004 s.filter
1005 .as_ref()
1006 .is_some_and(|e| crate::expr_has_subquery(e))
1007 || s.arg.as_ref().is_some_and(|e| crate::expr_has_subquery(e))
1008 || s.arg2.as_ref().is_some_and(|e| crate::expr_has_subquery(e))
1009 || s.order_by.iter().any(|o| crate::expr_has_subquery(&o.expr))
1010 });
1011 let eval_arg =
1012 |e: &Expr, r: &Row<'static>, c: &EvalContext<'_>| -> Result<Value<'static>, EvalError> {
1013 match correlated_eval {
1014 Some(f) if any_agg_subquery && crate::expr_has_subquery(e) => f(e, r, c),
1015 _ => eval::eval_expr(e, r, c),
1016 }
1017 };
1018 // v7.36 (perf — mailrs Phase 1, post u64-hash) — single
1019 // anonymous group fast path. When the query has no GROUP BY
1020 // (`SELECT SUM(LENGTH(col)) FROM ...`, COUNT, AVG, etc.) the
1021 // whole input collapses into one group. The fast path below
1022 // still pays one `groups.get("")` hash probe per row plus
1023 // `entry = &mut order[0]` reindex even when the empty-key
1024 // path encodes nothing — measured ~50 ns/row across 25 k rows
1025 // = ~1.25 ms of pure bookkeeping on the user_storage_usage
1026 // baseline.
1027 //
1028 // Bypass: lift `entry` outside the loop and feed every row
1029 // straight into it. Same `update_state` machinery, zero
1030 // per-row hash work, zero per-row index lookup.
1031 let single_anon_group = group_exprs.is_empty() && !rows.is_empty();
1032 if single_anon_group {
1033 // Seed the single group at idx 0 once.
1034 let init: Vec<AggState> = (0..agg_specs.len()).map(|_| AggState::default()).collect();
1035 order.clear();
1036 order.push((Vec::new(), init));
1037 }
1038 // v7.36 (perf — mailrs Phase 1, count_messages 2.58 → ?) —
1039 // `COUNT(*)` short-circuit. For a single-anon-group `COUNT(*)`
1040 // with no FILTER / DISTINCT, every survivor counts once — the
1041 // answer IS `rows.len()`. Skips the 25 k iterations of
1042 // `update_state("count_star", …)` on the mailrs count_messages
1043 // shape; the JOIN already produced exactly the set of rows
1044 // that must be counted.
1045 if single_anon_group
1046 && agg_specs.len() == 1
1047 && agg_specs[0].name == "count_star"
1048 && agg_specs[0].filter.is_none()
1049 && agg_specs[0].arg.is_none()
1050 && agg_specs[0].arg2.is_none()
1051 && agg_specs[0].order_by.is_empty()
1052 && !agg_specs[0].distinct
1053 {
1054 let state = &mut order[0].1[0];
1055 state.count = rows.len() as i64;
1056 return Ok(order);
1057 }
1058 // v7.36 (perf — mailrs Phase 1) — `COUNT(<bound col>)` (non-`*`)
1059 // collapses to: read the cell, increment when not NULL. Skips
1060 // the per-row spec dispatch + `update_state("count", …)`.
1061 if single_anon_group
1062 && agg_specs.len() == 1
1063 && agg_specs[0].name == "count"
1064 && agg_specs[0].filter.is_none()
1065 && agg_specs[0].arg2.is_none()
1066 && agg_specs[0].order_by.is_empty()
1067 && !agg_specs[0].distinct
1068 && arg_pos[0].is_some()
1069 {
1070 let p = arg_pos[0].unwrap();
1071 let mut count: i64 = 0;
1072 for row in rows {
1073 if !matches!(row.get(p), Some(Value::Null) | None) {
1074 count += 1;
1075 }
1076 }
1077 let state = &mut order[0].1[0];
1078 state.count = count;
1079 return Ok(order);
1080 }
1081 // v7.36 (perf — mailrs Phase 1, user_storage_usage 7.5 → ?) —
1082 // single-aggregate streaming accumulator. For
1083 // `SUM(<compiled-expr>)` / `SUM(<bound col>)` with no GROUP BY,
1084 // no FILTER, no arg2, no ORDER BY, no DISTINCT, the whole
1085 // per-row work collapses to: eval the arg, match the Value
1086 // variant, accumulate. Skips the spec-dispatch loop +
1087 // `update_state` per-row name match. On a 25 k-row JOIN
1088 // (user_storage_usage `SUM(LENGTH(text_body))`) that's
1089 // ~50-100 ns/row of pure spec-dispatch overhead removed.
1090 if single_anon_group
1091 && agg_specs.len() == 1
1092 && agg_specs[0].filter.is_none()
1093 && agg_specs[0].arg2.is_none()
1094 && agg_specs[0].order_by.is_empty()
1095 && !agg_specs[0].distinct
1096 && (agg_specs[0].name == "sum" || agg_specs[0].name == "avg")
1097 && (arg_pos[0].is_some() || arg_compiled[0].is_some())
1098 {
1099 let arg_pos0 = arg_pos[0];
1100 let arg_c0 = &arg_compiled[0];
1101 let mut sum_int: i64 = 0;
1102 let mut sum_float: f64 = 0.0;
1103 let mut use_float = false;
1104 let mut count: i64 = 0;
1105 // Borrow-aware fast inner: avoid the per-row clone when arg
1106 // is a bound column position.
1107 if let Some(p) = arg_pos0 {
1108 for row in rows {
1109 let v_ref = row.get(p).unwrap_or(&Value::Null);
1110 match v_ref {
1111 Value::Null => continue,
1112 Value::SmallInt(n) => {
1113 sum_int += i64::from(*n);
1114 count += 1;
1115 }
1116 Value::Int(n) => {
1117 sum_int += i64::from(*n);
1118 count += 1;
1119 }
1120 Value::BigInt(n) => {
1121 sum_int += *n;
1122 count += 1;
1123 }
1124 Value::Float(x) => {
1125 sum_float += *x;
1126 use_float = true;
1127 count += 1;
1128 }
1129 other => {
1130 return Err(EvalError::TypeMismatch {
1131 detail: format!("sum/avg need numeric, got {:?}", other.data_type()),
1132 });
1133 }
1134 }
1135 }
1136 } else if let Some(p) = arg_c0.as_ref().and_then(|c| c.as_single_column_length()) {
1137 // v7.36 (perf — mailrs Phase 1, user_storage_usage hot
1138 // inner) — `SUM(LENGTH(<text col>))` collapses to a
1139 // straight scan: read the cell by ref, branch on the
1140 // variant, do an ASCII probe + `len()` (or
1141 // `chars().count()` on non-ASCII), accumulate. No Step
1142 // VM, no stack push/pop, no `BigInt` boxing on the way
1143 // out — pure i64 sum. The original Step VM path keeps
1144 // running for everything outside this shape (`SUM(col)`,
1145 // `SUM(expr)`, multi-step compiled args).
1146 for row in rows {
1147 let Some(v_ref) = row.get(p) else {
1148 continue;
1149 };
1150 let n = match v_ref {
1151 Value::Null => continue,
1152 Value::Text(s) => {
1153 if s.is_ascii() {
1154 s.len() as i64
1155 } else {
1156 s.chars().count() as i64
1157 }
1158 }
1159 other => {
1160 return Err(EvalError::TypeMismatch {
1161 detail: format!("length() needs text, got {:?}", other.data_type()),
1162 });
1163 }
1164 };
1165 sum_int += n;
1166 count += 1;
1167 }
1168 } else {
1169 let c = arg_c0.as_ref().unwrap();
1170 for row in rows {
1171 let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
1172 match v {
1173 Value::Null => continue,
1174 Value::SmallInt(n) => {
1175 sum_int += i64::from(n);
1176 count += 1;
1177 }
1178 Value::Int(n) => {
1179 sum_int += i64::from(n);
1180 count += 1;
1181 }
1182 Value::BigInt(n) => {
1183 sum_int += n;
1184 count += 1;
1185 }
1186 Value::Float(x) => {
1187 sum_float += x;
1188 use_float = true;
1189 count += 1;
1190 }
1191 other => {
1192 return Err(EvalError::TypeMismatch {
1193 detail: format!("sum/avg need numeric, got {:?}", other.data_type()),
1194 });
1195 }
1196 }
1197 }
1198 }
1199 let state = &mut order[0].1[0];
1200 state.count = count;
1201 state.sum_int = sum_int;
1202 state.sum_float = sum_float;
1203 state.use_float = use_float;
1204 return Ok(order);
1205 }
1206 // v7.37.x (mailrs Track A 100k attack) — tight inlined loop for
1207 // the "single-Text GROUP BY + single MAX(bound numeric arg)"
1208 // shape. See `dedicated_max_loop` above for the gate. Returns
1209 // straight to the caller; the rest of the function (single-anon,
1210 // bound-fast, eval-slow paths) is skipped.
1211 if dedicated_max_loop && !single_anon_group {
1212 let gpos = group_pos[0].expect("dedicated_max_loop gates on Some");
1213 let apos = arg_pos[0].expect("dedicated_max_loop gates on Some");
1214 for row in rows {
1215 let kv = row.get(gpos).unwrap_or(&Value::Null);
1216 let idx = match kv {
1217 Value::Text(s) => match groups_text.get(s.as_ref()) {
1218 Some(&i) => i,
1219 None => {
1220 let i = order.len();
1221 order.push((
1222 alloc::vec![Value::text(s.clone())],
1223 alloc::vec![AggState::default()],
1224 ));
1225 groups_text.insert(s.to_string(), i);
1226 i
1227 }
1228 },
1229 Value::Null => match null_group_idx {
1230 Some(i) => i,
1231 None => {
1232 let i = order.len();
1233 order.push((alloc::vec![Value::Null], alloc::vec![AggState::default()]));
1234 null_group_idx = Some(i);
1235 i
1236 }
1237 },
1238 _ => {
1239 // Schema said Text but value isn't — fall back to
1240 // the generic encoded path for correctness.
1241 refs.clear();
1242 refs.push(kv);
1243 encode_key_refs_into(&refs, &mut keybuf_s);
1244 match groups.get(keybuf_s.as_str()) {
1245 Some(&i) => i,
1246 None => {
1247 let i = order.len();
1248 order.push((
1249 alloc::vec![kv.clone().into_owned()],
1250 alloc::vec![AggState::default()],
1251 ));
1252 groups.insert(keybuf_s.clone(), i);
1253 i
1254 }
1255 }
1256 }
1257 };
1258 // Inline MAX accumulator — skip the union-typed
1259 // `update_state` enum jump and per-spec arg dispatch.
1260 let av = row.get(apos).unwrap_or(&Value::Null);
1261 if !matches!(av, Value::Null) {
1262 let st = &mut order[idx].1[0];
1263 let upd = match &st.extreme {
1264 None => true,
1265 Some(prev) => value_cmp(av, prev) == core::cmp::Ordering::Greater,
1266 };
1267 if upd {
1268 st.extreme = Some(av.clone().into_owned());
1269 }
1270 }
1271 }
1272 return Ok(order);
1273 }
1274
1275 for row in rows {
1276 // v7.37.4 (L1 CSE) — reset per-row cache for shared compiled
1277 // aggregate-arg evals. No-op when no dedupe (empty vec).
1278 for slot in row_eval_cache.iter_mut() {
1279 *slot = None;
1280 }
1281 if single_anon_group {
1282 let entry = &mut order[0];
1283 let mat: Option<Cow<'_, Row>> = if needs_mat { Some(row.as_row()) } else { None };
1284 for (i, spec) in agg_specs.iter().enumerate() {
1285 if let Some(f) = &spec.filter
1286 && !matches!(
1287 eval_arg(f, mat.as_deref().expect("needs_mat for FILTER"), &ctx)?,
1288 Value::Bool(true)
1289 )
1290 {
1291 continue;
1292 }
1293 let arg_owned: Value;
1294 let arg_ref: &Value = match (&arg_pos[i], arg_slot[i], &spec.arg) {
1295 (Some(p), _, _) => {
1296 // v7.37.9 Phase 1A-ext counter — fast position-bound arg.
1297 crate::bump_counter!(AGG_PER_ROW_FAST_POS);
1298 row.get(*p).unwrap_or(&Value::Null)
1299 }
1300 (None, None, None) => {
1301 // COUNT(*) sentinel
1302 crate::bump_counter!(AGG_PER_ROW_COUNT_STAR_SENTINEL);
1303 arg_owned = Value::Bool(true);
1304 &arg_owned
1305 }
1306 (None, Some(s), _) => {
1307 if row_eval_cache[s].is_none() {
1308 // v7.37.9 Phase 1A-ext counter — Step-VM ran (cache miss).
1309 crate::bump_counter!(AGG_PER_ROW_COMPILED_MISS);
1310 let c = arg_compiled[arg_unique_idx[s]]
1311 .as_ref()
1312 .expect("arg_unique_idx points at a compiled spec");
1313 let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
1314 row_eval_cache[s] = Some(v);
1315 } else {
1316 // v7.37.9 Phase 1A-ext counter — CSE cache hit
1317 // (compiled arg deduped across specs in same row).
1318 crate::bump_counter!(AGG_PER_ROW_COMPILED_HIT);
1319 }
1320 row_eval_cache[s].as_ref().expect("just filled above")
1321 }
1322 (None, None, Some(e)) => {
1323 // v7.37.9 Phase 1A-ext counter — eval_expr fallback
1324 // (uncompilable spec — Cow row materialise per row).
1325 crate::bump_counter!(AGG_PER_ROW_EVAL_FALLBACK);
1326 arg_owned = eval_arg(
1327 e,
1328 mat.as_deref().expect("needs_mat for non-bound arg"),
1329 &ctx,
1330 )?;
1331 &arg_owned
1332 }
1333 };
1334 let arg2_val = match (&spec.arg2, &arg2_literal_val[i]) {
1335 (None, _) => None,
1336 // v7.37.43 (DISTA A-3) — literal arg2: clone the
1337 // precomputed value, skip per-row eval & row mat.
1338 (Some(_), Some(lit)) => {
1339 // v7.37.9 Phase 0 diagnostic — count per-row
1340 // hits of the DISTA A-3 fast path.
1341 crate::bump_counter!(DISTA_LITERAL_ARG2_CACHE_FIRE);
1342 Some(lit.clone())
1343 }
1344 (Some(e), None) => Some(eval_arg(
1345 e,
1346 mat.as_deref().expect("needs_mat for arg2"),
1347 &ctx,
1348 )?),
1349 };
1350 let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
1351 None
1352 } else {
1353 crate::bump_counter!(AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE);
1354 let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
1355 for (k, o) in spec.order_by.iter().enumerate() {
1356 let v: Value<'static> = if let Some(p) = order_pos[i][k] {
1357 row.get(p)
1358 .cloned()
1359 .map(Value::into_owned)
1360 .unwrap_or(Value::Null)
1361 } else {
1362 eval_arg(
1363 &o.expr,
1364 mat.as_deref().expect("needs_mat for ORDER key"),
1365 &ctx,
1366 )?
1367 };
1368 keys.push(v);
1369 }
1370 Some(keys)
1371 };
1372 // v7.36 (perf — bugfix v7.36.1 candidate) — first_ordered
1373 // was missing from the single_anon_group fast path,
1374 // sending `(array_agg(x ORDER BY y))[1]` values into
1375 // `update_state(array_agg, …)` whose finalize ignored
1376 // the absent `first_best` and returned `[]`. The slow
1377 // path below has the same branch — keep them aligned.
1378 if spec.first_ordered {
1379 if let Some(keys) = order_keys {
1380 let st = &mut entry.1[i];
1381 let better = match &st.first_best {
1382 None => true,
1383 Some((bk, _)) => {
1384 cmp_order_keys(&spec.order_by, &keys, bk)
1385 == core::cmp::Ordering::Less
1386 }
1387 };
1388 if better {
1389 st.first_best = Some((keys, arg_ref.clone().into_owned()));
1390 }
1391 }
1392 continue;
1393 }
1394 if spec.distinct {
1395 // v7.37.x (mailrs Track A 100k distinct_aggs attack)
1396 // — single-Text DISTINCT fast path. Within a single
1397 // distinct spec all input values come from one
1398 // expression and share one type, so the encode-
1399 // prefix (`S<text>|`) is redundant: the column
1400 // text alone is collision-free within this spec's
1401 // `seen` set. Skips encode_one + 2-walk
1402 // contains+insert; only Text arms apply, others
1403 // ride the encoded path unchanged.
1404 //
1405 // v7.37.x (docker-fair DISTA attack) — extend the
1406 // single-family fast path to BigInt via a parallel
1407 // `seen_int: Option<BTreeSet<i64>>`. The DISTA
1408 // `COUNT(DISTINCT m.id)` shape pumps 25 k BigInt
1409 // probes; skipping `encode_key_refs_into` saves
1410 // ~100 ns of alloc + format churn per row.
1411 if let Value::Text(s) = arg_ref {
1412 if entry.1[i].seen.contains(s.as_ref()) {
1413 continue;
1414 }
1415 entry.1[i].seen.insert(s.to_string());
1416 } else if let Value::BigInt(n) = arg_ref {
1417 let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
1418 if !set.insert(*n) {
1419 continue;
1420 }
1421 } else if let Value::Int(n) = arg_ref {
1422 let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
1423 if !set.insert(i64::from(*n)) {
1424 continue;
1425 }
1426 } else {
1427 encode_key_refs_into(core::slice::from_ref(&arg_ref), &mut dkeybuf);
1428 if entry.1[i].seen.contains(dkeybuf.as_str()) {
1429 continue;
1430 }
1431 entry.1[i].seen.insert(dkeybuf.clone());
1432 }
1433 }
1434 // v7.37.x (mailrs Track A 100k attack) — inline the
1435 // common aggregate kinds (MAX / MIN / Count / CountStar
1436 // / BoolOr / BoolAnd) here instead of dispatching
1437 // through `update_state`'s enum jump + per-kind branch.
1438 // Skipping the function-call overhead saves ~20-30 ns
1439 // per spec per row at 100 k; the slow kinds keep the
1440 // dispatched call.
1441 match spec.kind {
1442 AggKind::Max => {
1443 if !matches!(arg_ref, Value::Null) {
1444 let st = &mut entry.1[i];
1445 let upd = match &st.extreme {
1446 None => true,
1447 Some(prev) => {
1448 value_cmp(arg_ref, prev) == core::cmp::Ordering::Greater
1449 }
1450 };
1451 if upd {
1452 st.extreme = Some(arg_ref.clone().into_owned());
1453 }
1454 }
1455 }
1456 AggKind::Min => {
1457 if !matches!(arg_ref, Value::Null) {
1458 let st = &mut entry.1[i];
1459 let upd = match &st.extreme {
1460 None => true,
1461 Some(prev) => value_cmp(arg_ref, prev) == core::cmp::Ordering::Less,
1462 };
1463 if upd {
1464 st.extreme = Some(arg_ref.clone().into_owned());
1465 }
1466 }
1467 }
1468 AggKind::CountStar => {
1469 entry.1[i].count += 1;
1470 }
1471 AggKind::Count => {
1472 if !matches!(arg_ref, Value::Null) {
1473 entry.1[i].count += 1;
1474 }
1475 }
1476 AggKind::BoolOr => match arg_ref {
1477 Value::Bool(b) => {
1478 let st = &mut entry.1[i];
1479 st.bool_acc = Some(st.bool_acc.unwrap_or(false) || *b);
1480 }
1481 Value::Null => {}
1482 _ => update_state(
1483 &mut entry.1[i],
1484 spec.kind,
1485 &spec.name,
1486 arg_ref,
1487 arg2_val.as_ref(),
1488 order_keys,
1489 )?,
1490 },
1491 AggKind::BoolAnd => match arg_ref {
1492 Value::Bool(b) => {
1493 let st = &mut entry.1[i];
1494 st.bool_acc = Some(st.bool_acc.unwrap_or(true) && *b);
1495 }
1496 Value::Null => {}
1497 _ => update_state(
1498 &mut entry.1[i],
1499 spec.kind,
1500 &spec.name,
1501 arg_ref,
1502 arg2_val.as_ref(),
1503 order_keys,
1504 )?,
1505 },
1506 _ => {
1507 update_state(
1508 &mut entry.1[i],
1509 spec.kind,
1510 &spec.name,
1511 arg_ref,
1512 arg2_val.as_ref(),
1513 order_keys,
1514 )?;
1515 }
1516 }
1517 }
1518 continue;
1519 }
1520 // Fast key: bound positions + no ci folding -> encode
1521 // straight from borrowed cells; group_vals materialise
1522 // only when the group is NEW.
1523 if all_groups_bound && ci_positions.is_empty() {
1524 // v7.37.x — single-Text fast path uses the raw text as the
1525 // map key (no encode_one's `S<text>|` prefix/suffix push,
1526 // no refs Vec rebuild). NULL values land in a dedicated
1527 // slot so SQL's "all NULLs share one group" semantics hold.
1528 let idx = if single_text_group_col {
1529 let v = row.get(group_pos[0].unwrap()).unwrap_or(&Value::Null);
1530 match v {
1531 Value::Text(s) => match groups_text.get(s.as_ref()) {
1532 Some(&i) => i,
1533 None => {
1534 let i = order.len();
1535 let init: Vec<AggState> =
1536 (0..agg_specs.len()).map(|_| AggState::default()).collect();
1537 order.push((alloc::vec![Value::text(s.clone())], init));
1538 groups_text.insert(s.to_string(), i);
1539 i
1540 }
1541 },
1542 Value::Null => match null_group_idx {
1543 Some(i) => i,
1544 None => {
1545 let i = order.len();
1546 let init: Vec<AggState> =
1547 (0..agg_specs.len()).map(|_| AggState::default()).collect();
1548 order.push((alloc::vec![Value::Null], init));
1549 null_group_idx = Some(i);
1550 i
1551 }
1552 },
1553 _ => {
1554 // Schema says Text but value is something else
1555 // (coercion edge case). Fall back to the encoded
1556 // path for correctness — same logic as the
1557 // non-single-Text branch below.
1558 refs.clear();
1559 refs.push(v);
1560 encode_key_refs_into(&refs, &mut keybuf_s);
1561 match groups.get(keybuf_s.as_str()) {
1562 Some(&i) => i,
1563 None => {
1564 let i = order.len();
1565 let init: Vec<AggState> =
1566 (0..agg_specs.len()).map(|_| AggState::default()).collect();
1567 order.push((alloc::vec![v.clone().into_owned()], init));
1568 groups.insert(keybuf_s.clone(), i);
1569 i
1570 }
1571 }
1572 }
1573 }
1574 } else {
1575 refs.clear();
1576 refs.extend(
1577 group_pos
1578 .iter()
1579 .map(|p| row.get(p.unwrap()).unwrap_or(&Value::Null)),
1580 );
1581 encode_key_refs_into(&refs, &mut keybuf_s);
1582 match groups.get(keybuf_s.as_str()) {
1583 Some(&i) => i,
1584 None => {
1585 let i = order.len();
1586 let init: Vec<AggState> =
1587 (0..agg_specs.len()).map(|_| AggState::default()).collect();
1588 let owned: Vec<Value<'static>> =
1589 refs.iter().map(|v| (*v).clone().into_owned()).collect();
1590 order.push((owned, init));
1591 groups.insert(keybuf_s.clone(), i);
1592 i
1593 }
1594 }
1595 };
1596 let entry = &mut order[idx];
1597 // v7.33 (array_agg perf) — materialise the combined row AT
1598 // MOST once per input row, and only when a spec actually
1599 // needs the eval path (FILTER / non-bound arg / arg2 / non-
1600 // bound ORDER key). Bound args and bound ORDER keys read
1601 // cells by reference below, so the inbox shape (all bound)
1602 // never materialises — killing the per-row ~1 KB clone that
1603 // dominated the ordered-aggregate cost.
1604 let mat: Option<Cow<'_, Row>> = if needs_mat { Some(row.as_row()) } else { None };
1605 for (i, spec) in agg_specs.iter().enumerate() {
1606 // v7.32 (round-29) — FILTER (WHERE cond): exclude rows
1607 // where cond is not TRUE before they reach this
1608 // aggregate's accumulator (and before DISTINCT dedup).
1609 if let Some(f) = &spec.filter
1610 && !matches!(
1611 eval_arg(f, mat.as_deref().expect("needs_mat for FILTER"), &ctx)?,
1612 Value::Bool(true)
1613 )
1614 {
1615 continue;
1616 }
1617 let arg_owned: Value;
1618 let arg_ref: &Value = match (&arg_pos[i], arg_slot[i], &spec.arg) {
1619 (Some(p), _, _) => {
1620 crate::bump_counter!(AGG_PER_ROW_FAST_POS);
1621 row.get(*p).unwrap_or(&Value::Null)
1622 }
1623 (None, None, None) => {
1624 crate::bump_counter!(AGG_PER_ROW_COUNT_STAR_SENTINEL);
1625 arg_owned = Value::Bool(true);
1626 &arg_owned
1627 }
1628 (None, Some(s), _) => {
1629 // v7.37.4 (L1 CSE) — shared compiled-arg slot.
1630 // First spec that needs slot `s` this row pays
1631 // the Step-VM eval; siblings reading the same
1632 // slot get the cached Value for free. Preserves
1633 // FILTER semantics: a spec filtered out above
1634 // never reaches here, so its arg stays unevaled.
1635 if row_eval_cache[s].is_none() {
1636 crate::bump_counter!(AGG_PER_ROW_COMPILED_MISS);
1637 let c = arg_compiled[arg_unique_idx[s]]
1638 .as_ref()
1639 .expect("arg_unique_idx points at a compiled spec");
1640 let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
1641 row_eval_cache[s] = Some(v);
1642 } else {
1643 crate::bump_counter!(AGG_PER_ROW_COMPILED_HIT);
1644 }
1645 row_eval_cache[s].as_ref().expect("just filled above")
1646 }
1647 (None, None, Some(e)) => {
1648 crate::bump_counter!(AGG_PER_ROW_EVAL_FALLBACK);
1649 arg_owned = eval_arg(
1650 e,
1651 mat.as_deref().expect("needs_mat for non-bound arg"),
1652 &ctx,
1653 )?;
1654 &arg_owned
1655 }
1656 };
1657 let arg2_val = match (&spec.arg2, &arg2_literal_val[i]) {
1658 (None, _) => None,
1659 // v7.37.43 (DISTA A-3) — literal arg2: clone the
1660 // precomputed value, skip per-row eval & row mat.
1661 (Some(_), Some(lit)) => {
1662 // v7.37.9 Phase 0 diagnostic — count per-row
1663 // hits of the DISTA A-3 fast path.
1664 crate::bump_counter!(DISTA_LITERAL_ARG2_CACHE_FIRE);
1665 Some(lit.clone())
1666 }
1667 (Some(e), None) => Some(eval_arg(
1668 e,
1669 mat.as_deref().expect("needs_mat for arg2"),
1670 &ctx,
1671 )?),
1672 };
1673 let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
1674 None
1675 } else {
1676 crate::bump_counter!(AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE);
1677 let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
1678 for (k, o) in spec.order_by.iter().enumerate() {
1679 // Bound ORDER key → read the cell by reference; only
1680 // a non-bound key falls to the materialised eval path.
1681 keys.push(match order_pos[i][k] {
1682 Some(p) => row
1683 .get(p)
1684 .cloned()
1685 .map(Value::into_owned)
1686 .unwrap_or(Value::Null),
1687 None => eval_arg(
1688 &o.expr,
1689 mat.as_deref().expect("needs_mat for non-bound ORDER key"),
1690 &ctx,
1691 )?,
1692 });
1693 }
1694 Some(keys)
1695 };
1696 // v7.33 (array_agg argmax) — first_ordered: keep only the
1697 // running first-by-order element (strict-less replacement
1698 // = ties keep the earliest row, matching the stable-sort
1699 // `[1]`), no array build.
1700 if spec.first_ordered {
1701 if let Some(keys) = order_keys {
1702 let st = &mut entry.1[i];
1703 let better = match &st.first_best {
1704 None => true,
1705 Some((bk, _)) => {
1706 cmp_order_keys(&spec.order_by, &keys, bk)
1707 == core::cmp::Ordering::Less
1708 }
1709 };
1710 if better {
1711 st.first_best = Some((keys, arg_ref.clone().into_owned()));
1712 }
1713 }
1714 continue;
1715 }
1716 if spec.distinct {
1717 // v7.37.x — single-Text DISTINCT fast path (see
1718 // bound fast path counterpart above). Per-spec
1719 // type invariance lets us use the column text as
1720 // the `seen` key directly, no `S<text>|` prefix.
1721 // v7.37.x (docker-fair DISTA) — BigInt parallel
1722 // path skips encode_key_refs_into entirely.
1723 if let Value::Text(s) = arg_ref {
1724 if entry.1[i].seen.contains(s.as_ref()) {
1725 continue;
1726 }
1727 entry.1[i].seen.insert(s.to_string());
1728 } else if let Value::BigInt(n) = arg_ref {
1729 let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
1730 if !set.insert(*n) {
1731 continue;
1732 }
1733 } else if let Value::Int(n) = arg_ref {
1734 let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
1735 if !set.insert(i64::from(*n)) {
1736 continue;
1737 }
1738 } else {
1739 encode_key_refs_into(core::slice::from_ref(&arg_ref), &mut dkeybuf);
1740 if entry.1[i].seen.contains(dkeybuf.as_str()) {
1741 continue;
1742 }
1743 entry.1[i].seen.insert(dkeybuf.clone());
1744 }
1745 }
1746 // v7.37.x (mailrs Track A 100k attack) — inline the
1747 // common aggregate kinds (MAX / MIN / Count / CountStar
1748 // / BoolOr / BoolAnd) here instead of dispatching
1749 // through `update_state`'s enum jump + per-kind branch.
1750 // Skipping the function-call overhead saves ~20-30 ns
1751 // per spec per row at 100 k; the slow kinds keep the
1752 // dispatched call.
1753 match spec.kind {
1754 AggKind::Max => {
1755 if !matches!(arg_ref, Value::Null) {
1756 let st = &mut entry.1[i];
1757 let upd = match &st.extreme {
1758 None => true,
1759 Some(prev) => {
1760 value_cmp(arg_ref, prev) == core::cmp::Ordering::Greater
1761 }
1762 };
1763 if upd {
1764 st.extreme = Some(arg_ref.clone().into_owned());
1765 }
1766 }
1767 }
1768 AggKind::Min => {
1769 if !matches!(arg_ref, Value::Null) {
1770 let st = &mut entry.1[i];
1771 let upd = match &st.extreme {
1772 None => true,
1773 Some(prev) => value_cmp(arg_ref, prev) == core::cmp::Ordering::Less,
1774 };
1775 if upd {
1776 st.extreme = Some(arg_ref.clone().into_owned());
1777 }
1778 }
1779 }
1780 AggKind::CountStar => {
1781 entry.1[i].count += 1;
1782 }
1783 AggKind::Count => {
1784 if !matches!(arg_ref, Value::Null) {
1785 entry.1[i].count += 1;
1786 }
1787 }
1788 AggKind::BoolOr => match arg_ref {
1789 Value::Bool(b) => {
1790 let st = &mut entry.1[i];
1791 st.bool_acc = Some(st.bool_acc.unwrap_or(false) || *b);
1792 }
1793 Value::Null => {}
1794 _ => update_state(
1795 &mut entry.1[i],
1796 spec.kind,
1797 &spec.name,
1798 arg_ref,
1799 arg2_val.as_ref(),
1800 order_keys,
1801 )?,
1802 },
1803 AggKind::BoolAnd => match arg_ref {
1804 Value::Bool(b) => {
1805 let st = &mut entry.1[i];
1806 st.bool_acc = Some(st.bool_acc.unwrap_or(true) && *b);
1807 }
1808 Value::Null => {}
1809 _ => update_state(
1810 &mut entry.1[i],
1811 spec.kind,
1812 &spec.name,
1813 arg_ref,
1814 arg2_val.as_ref(),
1815 order_keys,
1816 )?,
1817 },
1818 _ => {
1819 update_state(
1820 &mut entry.1[i],
1821 spec.kind,
1822 &spec.name,
1823 arg_ref,
1824 arg2_val.as_ref(),
1825 order_keys,
1826 )?;
1827 }
1828 }
1829 }
1830 continue;
1831 }
1832 // v7.32 (P4 increment 2) — eval (non-bound) path: present the
1833 // row as a borrowed Row once (Owned → zero-cost borrow; a join
1834 // tuple materialises here exactly once, never on the bound fast
1835 // path above), then the original eval loop runs unchanged.
1836 let row_materialised = row.as_row();
1837 let row: &Row<'static> = &row_materialised;
1838 let group_vals: Vec<Value<'static>> = group_exprs
1839 .iter()
1840 .map(|g| eval::eval_expr(g, row, &ctx))
1841 .collect::<Result<_, _>>()?;
1842 // v7.17.0 Phase 2.5b — case-insensitive group keying: fold
1843 // only the ci columns, and only when any exist. Display
1844 // value (`group_vals`) stays original — only the key folds.
1845 let key = if ci_positions.is_empty() {
1846 encode_key(&group_vals)
1847 } else {
1848 let mut key_vals = group_vals.clone();
1849 for &i in &ci_positions {
1850 if let Value::Text(s) = &key_vals[i] {
1851 key_vals[i] = Value::text(s.to_ascii_lowercase());
1852 }
1853 }
1854 encode_key(&key_vals)
1855 };
1856 // Probe by index; the map owns the key once on vacant insert.
1857 let idx = match groups.get(key.as_str()) {
1858 Some(&i) => i,
1859 None => {
1860 let i = order.len();
1861 let init: Vec<AggState> =
1862 (0..agg_specs.len()).map(|_| AggState::default()).collect();
1863 order.push((group_vals.clone(), init));
1864 groups.insert(key, i);
1865 i
1866 }
1867 };
1868 let entry = &mut order[idx];
1869 for (i, spec) in agg_specs.iter().enumerate() {
1870 // v7.32 (round-29) — FILTER (WHERE cond): exclude rows where
1871 // cond is not TRUE before accumulation (and before DISTINCT).
1872 if let Some(f) = &spec.filter
1873 && !matches!(eval_arg(f, row, &ctx)?, Value::Bool(true))
1874 {
1875 continue;
1876 }
1877 let arg_val = match &spec.arg {
1878 None => Value::Bool(true), // count_star: sentinel non-null
1879 Some(e) => eval_arg(e, row, &ctx)?,
1880 };
1881 // v7.17.0 — `string_agg(value, separator)` evaluates the
1882 // separator per row but PG treats it as constant; we
1883 // pass the per-row value into update_state so a future
1884 // varying-separator caller still sees correct output,
1885 // even though SPG (like PG) only uses the most recent.
1886 let arg2_val = match &spec.arg2 {
1887 None => None,
1888 Some(e) => Some(eval_arg(e, row, &ctx)?),
1889 };
1890 // v7.24 (round-16 A) — aggregate-internal ORDER BY:
1891 // evaluate the key tuple against the source row.
1892 let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
1893 None
1894 } else {
1895 let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
1896 for o in &spec.order_by {
1897 keys.push(eval_arg(&o.expr, row, &ctx)?);
1898 }
1899 Some(keys)
1900 };
1901 // v7.33 (array_agg argmax) — first_ordered: keep the running
1902 // first-by-order element only (mirrors the bound fast path).
1903 if spec.first_ordered {
1904 if let Some(keys) = order_keys {
1905 let st = &mut entry.1[i];
1906 let better = match &st.first_best {
1907 None => true,
1908 Some((bk, _)) => {
1909 cmp_order_keys(&spec.order_by, &keys, bk) == core::cmp::Ordering::Less
1910 }
1911 };
1912 if better {
1913 st.first_best = Some((keys, arg_val.clone().into_owned()));
1914 }
1915 }
1916 continue;
1917 }
1918 // v7.25 (round-17) — DISTINCT: drop repeated inputs
1919 // before they reach the accumulator. NULLs flow through
1920 // (each aggregate's own NULL rule applies; PG also
1921 // treats NULL as a single distinct value for array_agg).
1922 // v7.37.x — single-Text fast path same shape as the
1923 // bound/slow paths above.
1924 if spec.distinct {
1925 // v7.37.x (docker-fair DISTA) — single-family fast
1926 // paths skip encode_key for Text/BigInt/Int.
1927 let inserted = match &arg_val {
1928 Value::Text(s) => entry.1[i].seen.insert(s.to_string()),
1929 Value::BigInt(n) => entry.1[i]
1930 .seen_int
1931 .get_or_insert_with(BTreeSet::new)
1932 .insert(*n),
1933 Value::Int(n) => entry.1[i]
1934 .seen_int
1935 .get_or_insert_with(BTreeSet::new)
1936 .insert(i64::from(*n)),
1937 _ => {
1938 let key = encode_key(core::slice::from_ref(&arg_val));
1939 entry.1[i].seen.insert(key)
1940 }
1941 };
1942 if !inserted {
1943 continue;
1944 }
1945 }
1946 update_state(
1947 &mut entry.1[i],
1948 spec.kind,
1949 &spec.name,
1950 &arg_val,
1951 arg2_val.as_ref(),
1952 order_keys,
1953 )?;
1954 }
1955 }
1956 Ok(order)
1957}
1958
1959/// (2a) Build the synthetic per-group schema: `__grp_0..K` then
1960/// `__agg_0..N`. Group types are probed from the first row; aggregate
1961/// types from each spec.
1962fn build_synth_schema(
1963 rows: &[RowRef<'_>],
1964 group_exprs: &[Expr],
1965 agg_specs: &[AggSpec],
1966 schema_cols: &[ColumnSchema],
1967 table_alias: Option<&str>,
1968) -> Result<Vec<ColumnSchema>, EvalError> {
1969 let ctx = EvalContext::new(schema_cols, table_alias);
1970 // Build synthetic schema: __grp_0..K then __agg_0..N.
1971 let group_types: Vec<DataType> = if rows.is_empty() {
1972 // Use Text as a safe stand-in — empty result means schema isn't
1973 // observable. Avoids needing to evaluate group exprs on no row.
1974 group_exprs.iter().map(|_| DataType::Text).collect()
1975 } else {
1976 let probe_row = rows[0].as_row();
1977 let probe: &Row<'static> = &probe_row;
1978 group_exprs
1979 .iter()
1980 .map(|g| {
1981 eval::eval_expr(g, probe, &ctx).map(|v| v.data_type().unwrap_or(DataType::Text))
1982 })
1983 .collect::<Result<_, _>>()?
1984 };
1985 let agg_types: Vec<DataType> = agg_specs
1986 .iter()
1987 .map(|spec| infer_agg_type(spec, schema_cols))
1988 .collect();
1989 let mut synth_schema: Vec<ColumnSchema> = Vec::new();
1990 for (i, ty) in group_types.iter().enumerate() {
1991 synth_schema.push(ColumnSchema::new(format!("__grp_{i}"), *ty, true));
1992 }
1993 for (i, ty) in agg_types.iter().enumerate() {
1994 synth_schema.push(ColumnSchema::new(format!("__agg_{i}"), *ty, true));
1995 }
1996 Ok(synth_schema)
1997}
1998
1999/// (2b) Materialise one synthetic row per group (insertion order):
2000/// apply each aggregate's internal ORDER BY, then finalise the running
2001/// state into the group + aggregate cells.
2002/// v7.33 — compare two aggregate-internal ORDER BY key tuples under the
2003/// per-key DESC / NULLS directives. This is the exact comparator the
2004/// finalize sort uses, factored out so the `first_ordered` argmax
2005/// accumulator's "keep first" decision is provably identical to taking
2006/// element `[1]` of the fully-sorted array.
2007fn cmp_order_keys(
2008 order_by: &[spg_sql::ast::OrderBy],
2009 a: &[Value<'static>],
2010 b: &[Value<'static>],
2011) -> core::cmp::Ordering {
2012 for (k, o) in order_by.iter().enumerate() {
2013 let cmp = crate::order_by_value_cmp(o.desc, o.nulls_first, &a[k], &b[k]);
2014 if cmp != core::cmp::Ordering::Equal {
2015 return cmp;
2016 }
2017 }
2018 core::cmp::Ordering::Equal
2019}
2020
2021fn finalize_synth_rows(
2022 order: &[(Vec<Value<'static>>, Vec<AggState>)],
2023 agg_specs: &[AggSpec],
2024 synth_schema: &[ColumnSchema],
2025 rows: &[RowRef<'_>],
2026 schema_cols: &[ColumnSchema],
2027 table_alias: Option<&str>,
2028) -> Result<Vec<Row<'static>>, EvalError> {
2029 let ctx = EvalContext::new(schema_cols, table_alias);
2030 // v7.32 (round-29) — ordered-set direct arguments (the percentile
2031 // fraction) are constant per PG, so evaluate each once up front.
2032 let direct_arg_vals: Vec<Option<Value>> = agg_specs
2033 .iter()
2034 .map(|spec| match (&spec.direct_arg, rows.first()) {
2035 (Some(e), Some(r)) => eval::eval_expr(e, &r.as_row(), &ctx).map(Some),
2036 _ => Ok(None),
2037 })
2038 .collect::<Result<_, _>>()?;
2039
2040 // Materialise synthetic rows (insertion order = `order`).
2041 let mut synth_rows: Vec<Row<'static>> = Vec::new();
2042 for (gvals, states) in order {
2043 let mut values: Vec<Value<'static>> = Vec::with_capacity(synth_schema.len());
2044 values.extend(gvals.iter().cloned());
2045 for (i, st) in states.iter().enumerate() {
2046 // v7.33 (array_agg argmax) — first_ordered: the running
2047 // first-by-order value IS the result; no array build/sort.
2048 if agg_specs[i].first_ordered {
2049 values.push(
2050 st.first_best
2051 .as_ref()
2052 .map_or(Value::Null, |(_, v)| v.clone()),
2053 );
2054 continue;
2055 }
2056 // v7.24 (round-16 A) — order the collected items per the
2057 // aggregate-internal ORDER BY before finalize consumes
2058 // them.
2059 let st_sorted;
2060 let st_final: &AggState =
2061 if !agg_specs[i].order_by.is_empty() && st.item_keys.len() == st.items.len() {
2062 let mut idx: Vec<usize> = (0..st.items.len()).collect();
2063 let ob = &agg_specs[i].order_by;
2064 idx.sort_by(|&x, &y| cmp_order_keys(ob, &st.item_keys[x], &st.item_keys[y]));
2065 let mut sorted = st.clone();
2066 sorted.items = idx.iter().map(|&j| st.items[j].clone()).collect();
2067 st_sorted = sorted;
2068 &st_sorted
2069 } else {
2070 st
2071 };
2072 // Ordered-set aggregates compute from the sorted items + the
2073 // direct fraction; everything else uses the running state.
2074 let v = if is_within_group_name(&agg_specs[i].name) {
2075 finalize_ordered_set(
2076 &agg_specs[i].name,
2077 st_final,
2078 direct_arg_vals[i].as_ref(),
2079 agg_specs[i].order_by.first(),
2080 )
2081 } else {
2082 finalize(&agg_specs[i].name, st_final)
2083 };
2084 values.push(v);
2085 }
2086 synth_rows.push(Row::new(values));
2087 }
2088 Ok(synth_rows)
2089}
2090
2091/// (3) Rewrite the user's SELECT items + HAVING to reference the
2092/// synthetic columns, filter groups by HAVING, and project each
2093/// surviving group into an output row. The synth rows ride alongside
2094/// (`kept_synth`) so post-LIMIT deferred subqueries can evaluate later.
2095#[allow(clippy::too_many_lines)]
2096fn project_groups(
2097 synth_rows: Vec<Row<'static>>,
2098 stmt: &SelectStatement,
2099 group_exprs: &[Expr],
2100 agg_specs: &[AggSpec],
2101 synth_schema: &[ColumnSchema],
2102 correlated_eval: Option<CorrelatedEval<'_>>,
2103 defer_projection: bool,
2104) -> Result<Projection, EvalError> {
2105 // Rewrite the user's SELECT items + ORDER BY to reference synthetic
2106 // columns. After rewriting, every remaining `Expr::Column` must
2107 // resolve against the synthetic schema (i.e. must have been a GROUP
2108 // BY expression).
2109 let columns: Vec<ColumnSchema> = stmt
2110 .items
2111 .iter()
2112 .map(|item| match item {
2113 SelectItem::Wildcard => Err(EvalError::TypeMismatch {
2114 detail: "SELECT * with aggregates is not supported".into(),
2115 }),
2116 SelectItem::Expr { expr, alias } => {
2117 let rewritten = rewrite_expr(expr, group_exprs, agg_specs);
2118 let name = alias.clone().unwrap_or_else(|| expr.to_string());
2119 Ok(ColumnSchema::new(
2120 name,
2121 agg_or_group_type(&rewritten, synth_schema),
2122 true,
2123 ))
2124 }
2125 })
2126 .collect::<Result<_, _>>()?;
2127
2128 // Project per synthetic row. HAVING filters out groups *before*
2129 // we keep the projected row — same semantics as PG: HAVING runs
2130 // against the aggregated row (so `HAVING count(*) > 1` works) and
2131 // sees only group-by'd columns plus aggregate values.
2132 let synth_ctx = EvalContext::new(synth_schema, None);
2133 let having_rewritten = stmt
2134 .having
2135 .as_ref()
2136 .map(|h| rewrite_expr(h, group_exprs, agg_specs));
2137 // v7.30 (phase 3e-1) - rewrite SELECT items ONCE. This ran per
2138 // GROUP (23.5k x 9 items of AST cloning = ~48% of the inbox
2139 // query in sampled stacks); the rewrite is group-independent.
2140 // Stable addresses also let the per-expression subquery plans
2141 // (v7.29 3c) hit across groups instead of rebuilding.
2142 let items_rewritten: alloc::vec::Vec<Option<Expr>> = stmt
2143 .items
2144 .iter()
2145 .map(|item| match item {
2146 SelectItem::Expr { expr, .. } => Some(rewrite_expr(expr, group_exprs, agg_specs)),
2147 SelectItem::Wildcard => None,
2148 })
2149 .collect();
2150 // v7.31 (perf — PG lesson #1): subquery-bearing select items
2151 // deferred to post-LIMIT, when no sort/filter key can observe
2152 // them. ORDER BY rewrites are hoisted here so the safety check
2153 // and the sort below share one rewrite pass.
2154 let order_rewritten: Vec<Expr> = stmt
2155 .order_by
2156 .iter()
2157 .map(|o| rewrite_expr(&o.expr, group_exprs, agg_specs))
2158 .collect();
2159 let defer_enabled = correlated_eval.is_some()
2160 && !stmt.distinct
2161 && !having_rewritten
2162 .as_ref()
2163 .is_some_and(crate::expr_has_subquery)
2164 && !order_rewritten.iter().any(crate::expr_has_subquery);
2165 let deferred: Vec<(usize, Expr)> = if defer_enabled {
2166 items_rewritten
2167 .iter()
2168 .enumerate()
2169 .filter_map(|(i, r)| {
2170 r.as_ref()
2171 .filter(|e| crate::expr_has_subquery(e))
2172 .map(|e| (i, e.clone()))
2173 })
2174 .collect()
2175 } else {
2176 Vec::new()
2177 };
2178 // v7.32 (architecture v2, P2) — compile the per-group synth-row
2179 // expressions ONCE. The projection / HAVING here run per GROUP
2180 // (24k for the inbox shape) × per item; the rewritten exprs are
2181 // mostly `Column(__agg_N)` / `Column(__grp_K)` against the synth
2182 // schema — flat step programs, no tree walk per group.
2183 let having_compiled = having_rewritten
2184 .as_ref()
2185 .filter(|h| eval::fully_compilable(h))
2186 .map(|h| eval::compile_expr(h, &synth_ctx));
2187 let items_compiled: Vec<Option<eval::CompiledExpr>> = items_rewritten
2188 .iter()
2189 .enumerate()
2190 .map(|(i, r)| {
2191 r.as_ref()
2192 .filter(|e| !deferred.iter().any(|(c, _)| *c == i) && eval::fully_compilable(e))
2193 .map(|e| eval::compile_expr(e, &synth_ctx))
2194 })
2195 .collect();
2196 let mut kept_synth: Vec<Row<'static>> = Vec::new();
2197 let mut out_rows: Vec<Row<'static>> = Vec::new();
2198 let mut stack: Vec<Value<'static>> = Vec::new();
2199 for srow in synth_rows {
2200 if let Some(hc) = &having_compiled {
2201 let cond = eval::eval_compiled(hc, &srow, &synth_ctx, &mut stack)?;
2202 if !matches!(cond, Value::Bool(true)) {
2203 continue;
2204 }
2205 } else if let Some(h) = &having_rewritten {
2206 let cond = match correlated_eval {
2207 Some(f) if crate::expr_has_subquery(h) => f(h, &srow, &synth_ctx)?,
2208 _ => eval::eval_expr(h, &srow, &synth_ctx)?,
2209 };
2210 if !matches!(cond, Value::Bool(true)) {
2211 continue;
2212 }
2213 }
2214 // v7.37.x — when caller pre-truncates via ORDER BY+LIMIT, skip
2215 // per-item projection here; the caller fills the placeholder
2216 // out_rows from the top-K survivors below.
2217 if defer_projection {
2218 kept_synth.push(srow);
2219 out_rows.push(Row::new(Vec::new()));
2220 continue;
2221 }
2222 let mut values: Vec<Value<'static>> = Vec::with_capacity(columns.len());
2223 for (i, rewritten) in items_rewritten.iter().enumerate() {
2224 let Some(rewritten) = rewritten else { continue };
2225 if deferred.iter().any(|(c, _)| *c == i) {
2226 values.push(Value::Null);
2227 continue;
2228 }
2229 values.push(if let Some(cc) = &items_compiled[i] {
2230 eval::eval_compiled(cc, &srow, &synth_ctx, &mut stack)?
2231 } else {
2232 match correlated_eval {
2233 Some(f) if crate::expr_has_subquery(rewritten) => {
2234 f(rewritten, &srow, &synth_ctx)?
2235 }
2236 _ => eval::eval_expr(rewritten, &srow, &synth_ctx)?,
2237 }
2238 });
2239 }
2240 kept_synth.push(srow);
2241 out_rows.push(Row::new(values));
2242 }
2243 let deferred_project_state = if defer_projection {
2244 Some(DeferredProject {
2245 items_rewritten,
2246 items_compiled,
2247 })
2248 } else {
2249 None
2250 };
2251 Ok(Projection {
2252 columns,
2253 out_rows,
2254 kept_synth,
2255 deferred,
2256 order_rewritten,
2257 deferred_project: deferred_project_state,
2258 })
2259}
2260
2261/// (4) Sort the projected output by the rewritten ORDER BY keys. The
2262/// synth rows ride through the sort so deferred subqueries evaluate
2263/// against the surviving groups after the caller's LIMIT truncation.
2264fn sort_synth_by_order_by(
2265 synth_schema: &[ColumnSchema],
2266 order_by: &[spg_sql::ast::OrderBy],
2267 order_rewritten: &[Expr],
2268 mut kept_synth: Vec<Row<'static>>,
2269 mut out_rows: Vec<Row<'static>>,
2270 correlated_eval: Option<CorrelatedEval<'_>>,
2271 keep_n: Option<usize>,
2272) -> Result<(Vec<Row<'static>>, Vec<Row<'static>>), EvalError> {
2273 let synth_ctx = EvalContext::new(synth_schema, None);
2274 // v6.4.0 — multi-key ORDER BY on aggregate output. Each key
2275 // gets its own rewrite + per-key DESC flag. (Rewrites hoisted
2276 // above as `order_rewritten` — shared with the deferral
2277 // safety check.)
2278 let keys_meta: Vec<(bool, Option<bool>)> =
2279 order_by.iter().map(|o| (o.desc, o.nulls_first)).collect();
2280 // P2: compile order-by keys once (per-group sort keys are
2281 // the same `__agg_N` / `__grp_K` shape as the projection).
2282 let order_compiled: Vec<Option<eval::CompiledExpr>> = order_rewritten
2283 .iter()
2284 .map(|e| {
2285 Some(e)
2286 .filter(|e| eval::fully_compilable(e))
2287 .map(|e| eval::compile_expr(e, &synth_ctx))
2288 })
2289 .collect();
2290 // The synth row rides through the sort so deferred exprs can
2291 // evaluate against the surviving groups after the caller's
2292 // LIMIT truncation.
2293 let mut keystack: Vec<Value<'static>> = Vec::new();
2294 let mut tagged: Vec<(Vec<Value<'static>>, Row, Row)> = Vec::with_capacity(kept_synth.len());
2295 for (s, o) in kept_synth.into_iter().zip(out_rows) {
2296 let mut keys = Vec::with_capacity(order_rewritten.len());
2297 for (e, oc) in order_rewritten.iter().zip(&order_compiled) {
2298 keys.push(if let Some(oc) = oc {
2299 eval::eval_compiled(oc, &s, &synth_ctx, &mut keystack)?
2300 } else {
2301 match correlated_eval {
2302 Some(f) if crate::expr_has_subquery(e) => f(e, &s, &synth_ctx)?,
2303 _ => eval::eval_expr(e, &s, &synth_ctx)?,
2304 }
2305 });
2306 }
2307 tagged.push((keys, s, o));
2308 }
2309 let cmp = |a: &(Vec<Value<'static>>, Row, Row), b: &(Vec<Value<'static>>, Row, Row)| {
2310 use core::cmp::Ordering;
2311 for (i, (ka, kb)) in a.0.iter().zip(b.0.iter()).enumerate() {
2312 let (desc, nf) = keys_meta[i];
2313 let c = crate::order_by_value_cmp(desc, nf, ka, kb);
2314 if c != Ordering::Equal {
2315 return c;
2316 }
2317 }
2318 Ordering::Equal
2319 };
2320 // v7.37.3 — top-K partial sort when `keep_n` is small enough to
2321 // matter (`Some(k)` with `k < tagged.len()` and `k > 0`).
2322 // `select_nth_unstable_by` partitions in O(N), then we sort the
2323 // surviving prefix in O(K log K). Total = O(N + K log K) vs
2324 // O(N log N) the full sort would pay — matches the inbox-listing
2325 // shape PG uses.
2326 //
2327 match keep_n {
2328 Some(k) if k < tagged.len() && k > 0 => {
2329 let pivot = k - 1;
2330 tagged.select_nth_unstable_by(pivot, cmp);
2331 tagged[..k].sort_by(cmp);
2332 tagged.truncate(k);
2333 }
2334 _ => {
2335 tagged.sort_by(cmp);
2336 }
2337 }
2338 kept_synth = Vec::with_capacity(tagged.len());
2339 out_rows = Vec::with_capacity(tagged.len());
2340 for (_, s, o) in tagged {
2341 kept_synth.push(s);
2342 out_rows.push(o);
2343 }
2344 Ok((kept_synth, out_rows))
2345}
2346
2347/// v7.17.0 — walk the statement again to validate the positional
2348/// arity of every aggregate call site. Done after AST collection
2349/// rather than inside `collect_aggregates` so the collector stays
2350/// infallible; callers in `run()` can do a single early-error
2351/// exit before any per-row work.
2352fn validate_agg_arities(stmt: &SelectStatement, _specs: &[AggSpec]) -> Result<(), EvalError> {
2353 fn walk(e: &Expr) -> Result<(), EvalError> {
2354 if let Expr::FunctionCall { name, args } = e {
2355 let lower = name.to_ascii_lowercase();
2356 let expected: Option<usize> = match lower.as_str() {
2357 "count_star" => Some(0),
2358 "count" | "sum" | "avg" | "min" | "max" | "array_agg"
2359 // v7.17.0 — boolean aggregates also take exactly
2360 // one arg. `every` is an alias normalised inside
2361 // collect_aggregates / rewrite_expr.
2362 | "bool_and" | "bool_or" | "every"
2363 // v7.32 (round-29) — statistical + bitwise aggregates
2364 // + single-arg JSON aggregate.
2365 | "stddev" | "stddev_samp" | "stddev_pop"
2366 | "variance" | "var_samp" | "var_pop"
2367 | "bit_and" | "bit_or" | "bit_xor"
2368 | "json_agg" | "jsonb_agg" => Some(1),
2369 // v7.32 (round-29) — two-argument aggregates: string_agg,
2370 // the regression family f(Y, X), and json_object_agg.
2371 "string_agg"
2372 | "covar_pop" | "covar_samp" | "corr"
2373 | "regr_count" | "regr_avgx" | "regr_avgy" | "regr_slope"
2374 | "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy"
2375 | "json_object_agg" | "jsonb_object_agg" => Some(2),
2376 _ => None,
2377 };
2378 if let Some(want) = expected
2379 && args.len() != want
2380 {
2381 return Err(EvalError::TypeMismatch {
2382 detail: alloc::format!("{lower}() takes {want} arg(s), got {}", args.len()),
2383 });
2384 }
2385 for a in args {
2386 walk(a)?;
2387 }
2388 } else if let Expr::Binary { lhs, rhs, .. } = e {
2389 walk(lhs)?;
2390 walk(rhs)?;
2391 } else if let Expr::Unary { expr, .. }
2392 | Expr::Cast { expr, .. }
2393 | Expr::IsNull { expr, .. } = e
2394 {
2395 walk(expr)?;
2396 }
2397 Ok(())
2398 }
2399 for item in &stmt.items {
2400 if let SelectItem::Expr { expr, .. } = item {
2401 walk(expr)?;
2402 }
2403 }
2404 for o in &stmt.order_by {
2405 walk(&o.expr)?;
2406 }
2407 if let Some(h) = &stmt.having {
2408 walk(h)?;
2409 }
2410 Ok(())
2411}
2412
2413/// v7.33 (array_agg argmax) — recognise `(array_agg(x ORDER BY y))[1]`,
2414/// the argmax/argmin idiom: a non-DISTINCT ordered `array_agg`
2415/// subscripted by the constant 1. Returns `(value_arg, order_by,
2416/// filter)` on a match. When matched, the whole per-group array build +
2417/// sort + materialise is replaced by a running first-by-order scalar
2418/// accumulator and the subscript node is consumed (replaced by the
2419/// synthetic column). collect_aggregates and rewrite_expr share this one
2420/// matcher so their `__agg_<i>` assignment stays in lockstep.
2421fn first_ordered_array_agg(e: &Expr) -> Option<(&Expr, &[spg_sql::ast::OrderBy], Option<&Expr>)> {
2422 let Expr::ArraySubscript { target, index } = e else {
2423 return None;
2424 };
2425 if !matches!(
2426 index.as_ref(),
2427 Expr::Literal(spg_sql::ast::Literal::Integer(1))
2428 ) {
2429 return None;
2430 }
2431 let Expr::AggregateOrdered {
2432 call,
2433 order_by,
2434 distinct,
2435 filter,
2436 } = target.as_ref()
2437 else {
2438 return None;
2439 };
2440 if *distinct || order_by.is_empty() {
2441 return None;
2442 }
2443 let Expr::FunctionCall { name, args } = call.as_ref() else {
2444 return None;
2445 };
2446 if !name.eq_ignore_ascii_case("array_agg") || args.len() != 1 {
2447 return None;
2448 }
2449 Some((&args[0], order_by, filter.as_deref()))
2450}
2451
2452fn collect_aggregates(e: &Expr, out: &mut Vec<AggSpec>) {
2453 match e {
2454 // v7.24 (round-16 A) — ordered aggregate: register the inner
2455 // call's spec with the ordering attached.
2456 Expr::AggregateOrdered {
2457 call,
2458 order_by,
2459 distinct,
2460 filter,
2461 } => {
2462 if let Expr::FunctionCall { name, args } = call.as_ref() {
2463 let lower = name.to_ascii_lowercase();
2464 if is_aggregate_name(&lower) {
2465 let canonical = if lower == "every" {
2466 "bool_and".to_string()
2467 } else {
2468 lower
2469 };
2470 // Ordered-set aggregates (`percentile_cont(f)
2471 // WITHIN GROUP (ORDER BY x)`) take the value to
2472 // aggregate from the sort spec and the in-parens
2473 // arg as the direct (fraction) argument.
2474 let ordered_set = is_within_group_name(&canonical);
2475 let (arg, direct_arg) = if ordered_set {
2476 (
2477 order_by.first().map(|o| o.expr.clone()),
2478 args.first().cloned(),
2479 )
2480 } else {
2481 (args.first().cloned(), None)
2482 };
2483 let spec = AggSpec {
2484 kind: classify_agg_name(&canonical),
2485 name: canonical.clone(),
2486 arg,
2487 arg2: if agg_uses_second_arg(&canonical) {
2488 args.get(1).cloned()
2489 } else {
2490 None
2491 },
2492 distinct: *distinct,
2493 order_by: order_by.clone(),
2494 filter: filter.as_deref().cloned(),
2495 direct_arg,
2496 first_ordered: false,
2497 };
2498 if !out.iter().any(|s| {
2499 s.name == spec.name
2500 && s.arg == spec.arg
2501 && s.arg2 == spec.arg2
2502 && s.distinct == spec.distinct
2503 && s.order_by == spec.order_by
2504 && s.filter == spec.filter
2505 && s.direct_arg == spec.direct_arg
2506 && s.first_ordered == spec.first_ordered
2507 }) {
2508 out.push(spec);
2509 }
2510 return;
2511 }
2512 }
2513 collect_aggregates(call, out);
2514 for o in order_by {
2515 collect_aggregates(&o.expr, out);
2516 }
2517 }
2518 Expr::FunctionCall { name, args } => {
2519 let lower = name.to_ascii_lowercase();
2520 if is_aggregate_name(&lower) {
2521 let arg = if lower == "count_star" {
2522 None
2523 } else {
2524 args.first().cloned()
2525 };
2526 // v7.17.0 — second positional arg for
2527 // `string_agg(value, separator)`; v7.32 — also the
2528 // regression family `f(Y, X)` and `json_object_agg`.
2529 let arg2 = if agg_uses_second_arg(&lower) {
2530 args.get(1).cloned()
2531 } else {
2532 None
2533 };
2534 // v7.17.0 — `every` is the SQL-standard alias for
2535 // `bool_and`; collapse at collection time so
2536 // update_state / finalize need only one arm.
2537 let canonical = if lower == "every" {
2538 "bool_and".to_string()
2539 } else {
2540 lower
2541 };
2542 let spec = AggSpec {
2543 kind: classify_agg_name(&canonical),
2544 name: canonical,
2545 arg: arg.clone(),
2546 arg2: arg2.clone(),
2547 distinct: false,
2548 order_by: Vec::new(),
2549 filter: None,
2550 direct_arg: None,
2551 first_ordered: false,
2552 };
2553 if !out.iter().any(|s| {
2554 s.name == spec.name
2555 && s.arg == spec.arg
2556 && s.arg2 == spec.arg2
2557 && !s.distinct
2558 && s.order_by == spec.order_by
2559 && s.filter.is_none()
2560 && !s.first_ordered
2561 }) {
2562 out.push(spec);
2563 }
2564 // Don't recurse into the arg — nested aggregates are
2565 // illegal in standard SQL.
2566 } else {
2567 for a in args {
2568 collect_aggregates(a, out);
2569 }
2570 }
2571 }
2572 Expr::Binary { lhs, rhs, .. } => {
2573 collect_aggregates(lhs, out);
2574 collect_aggregates(rhs, out);
2575 }
2576 Expr::Unary { expr, .. } | Expr::Cast { expr, .. } | Expr::IsNull { expr, .. } => {
2577 collect_aggregates(expr, out);
2578 }
2579 Expr::Like { expr, pattern, .. } => {
2580 collect_aggregates(expr, out);
2581 collect_aggregates(pattern, out);
2582 }
2583 Expr::InList { expr, list, .. } => {
2584 collect_aggregates(expr, out);
2585 for item in list {
2586 collect_aggregates(item, out);
2587 }
2588 }
2589 Expr::Extract { source, .. } => collect_aggregates(source, out),
2590 // v4.10 subquery + v4.12 window / Literal / Column —
2591 // non-recursing leaves for the aggregate collector.
2592 Expr::ScalarSubquery(_)
2593 | Expr::Exists { .. }
2594 | Expr::InSubquery { .. }
2595 | Expr::WindowFunction { .. }
2596 | Expr::Literal(_)
2597 | Expr::Placeholder(_)
2598 | Expr::Column(_) => {}
2599 // v7.10.10 — recurse into array constructor children +
2600 // subscript / ANY/ALL operands.
2601 Expr::Array(items) => {
2602 for elem in items {
2603 collect_aggregates(elem, out);
2604 }
2605 }
2606 Expr::ArraySubscript { target, index } => {
2607 // v7.33 (array_agg argmax) — `(array_agg(x ORDER BY y))[1]`
2608 // collects as a first_ordered spec; the subscript is consumed
2609 // here (do NOT recurse into the array_agg, or it would also
2610 // register a plain full-array spec).
2611 if let Some((arg, order_by, filter)) = first_ordered_array_agg(e) {
2612 let spec = AggSpec {
2613 kind: AggKind::ArrayAgg,
2614 name: "array_agg".to_string(),
2615 arg: Some(arg.clone()),
2616 arg2: None,
2617 distinct: false,
2618 order_by: order_by.to_vec(),
2619 filter: filter.cloned(),
2620 direct_arg: None,
2621 first_ordered: true,
2622 };
2623 if !out.iter().any(|s| {
2624 s.name == spec.name
2625 && s.arg == spec.arg
2626 && s.order_by == spec.order_by
2627 && s.filter == spec.filter
2628 && s.first_ordered
2629 }) {
2630 out.push(spec);
2631 }
2632 return;
2633 }
2634 collect_aggregates(target, out);
2635 collect_aggregates(index, out);
2636 }
2637 Expr::AnyAll { expr, array, .. } => {
2638 collect_aggregates(expr, out);
2639 collect_aggregates(array, out);
2640 }
2641 Expr::Case {
2642 operand,
2643 branches,
2644 else_branch,
2645 } => {
2646 if let Some(o) = operand {
2647 collect_aggregates(o, out);
2648 }
2649 for (w, t) in branches {
2650 collect_aggregates(w, out);
2651 collect_aggregates(t, out);
2652 }
2653 if let Some(e) = else_branch {
2654 collect_aggregates(e, out);
2655 }
2656 }
2657 }
2658}
2659
2660fn update_state(
2661 st: &mut AggState,
2662 kind: AggKind,
2663 name: &str,
2664 v: &Value<'_>,
2665 arg2: Option<&Value<'_>>,
2666 order_keys: Option<Vec<Value<'static>>>,
2667) -> Result<(), EvalError> {
2668 let is_null = matches!(v, Value::Null);
2669 // v7.37.4 (R34) — dispatch by pre-classified `kind` (`Copy`
2670 // enum), not by per-row string match. Hot inner loop on
2671 // multi-aggregate queries (mailrs `/api/conversations`: 14
2672 // aggregates × 100 k rows = 1.4 M dispatches) sees an enum
2673 // jump table instead of a sequence of `eq_str` checks. `name`
2674 // is still threaded through for error messages so the user-
2675 // facing wording is unchanged.
2676 match kind {
2677 AggKind::CountStar => st.count += 1,
2678 AggKind::Count => {
2679 if !is_null {
2680 st.count += 1;
2681 }
2682 }
2683 AggKind::Sum | AggKind::Avg => {
2684 if is_null {
2685 return Ok(());
2686 }
2687 st.count += 1;
2688 match v {
2689 Value::Int(n) => st.sum_int += i64::from(*n),
2690 Value::BigInt(n) => st.sum_int += *n,
2691 Value::Float(x) => {
2692 st.use_float = true;
2693 st.sum_float += *x;
2694 }
2695 other => {
2696 return Err(EvalError::TypeMismatch {
2697 detail: format!("sum/avg need numeric, got {:?}", other.data_type()),
2698 });
2699 }
2700 }
2701 }
2702 AggKind::Min => {
2703 if is_null {
2704 return Ok(());
2705 }
2706 match &st.extreme {
2707 None => st.extreme = Some(v.clone().into_owned()),
2708 Some(cur) => {
2709 if value_cmp(v, cur) == core::cmp::Ordering::Less {
2710 st.extreme = Some(v.clone().into_owned());
2711 }
2712 }
2713 }
2714 }
2715 AggKind::Max => {
2716 if is_null {
2717 return Ok(());
2718 }
2719 match &st.extreme {
2720 None => st.extreme = Some(v.clone().into_owned()),
2721 Some(cur) => {
2722 if value_cmp(v, cur) == core::cmp::Ordering::Greater {
2723 st.extreme = Some(v.clone().into_owned());
2724 }
2725 }
2726 }
2727 }
2728 // v7.17.0 — string_agg(value, separator). NULL value is
2729 // skipped (PG aggregate-skip-null). Separator captured
2730 // from the latest row that flows through; matches PG's
2731 // semantics of evaluating the separator per row but using
2732 // the last value at finalize time (in practice it's
2733 // constant). count is bumped so we can distinguish "empty
2734 // group → NULL" from "all-NULL group → NULL".
2735 AggKind::StringAgg => {
2736 if let Some(sep) = arg2
2737 && let Value::Text(s) = sep
2738 {
2739 st.separator = Some(s.to_string());
2740 }
2741 if is_null {
2742 return Ok(());
2743 }
2744 if let Value::Text(s) = v {
2745 st.items.push(Value::text(s.clone()));
2746 if let Some(k) = order_keys {
2747 st.item_keys.push(k);
2748 }
2749 st.count += 1;
2750 } else {
2751 return Err(EvalError::TypeMismatch {
2752 detail: format!("string_agg requires text value, got {:?}", v.data_type()),
2753 });
2754 }
2755 }
2756 // v7.17.0 — array_agg(value). Unlike string_agg, NULL
2757 // elements are KEPT in the array (PG behaviour); the
2758 // result is NULL only when ZERO rows fed in. Element type
2759 // is locked from the first row's value type; subsequent
2760 // rows must match (PG also rejects mixed-type array_agg).
2761 AggKind::ArrayAgg => {
2762 st.items.push(v.clone().into_owned());
2763 if let Some(k) = order_keys {
2764 st.item_keys.push(k);
2765 }
2766 st.count += 1;
2767 }
2768 // v7.17.0 — bool_and(p): TRUE iff every non-NULL input is
2769 // TRUE. NULL skipped; running accumulator stays at TRUE
2770 // until the first non-NULL FALSE.
2771 AggKind::BoolAnd => {
2772 if is_null {
2773 return Ok(());
2774 }
2775 let b = match v {
2776 Value::Bool(b) => *b,
2777 other => {
2778 return Err(EvalError::TypeMismatch {
2779 detail: format!("bool_and requires bool, got {:?}", other.data_type()),
2780 });
2781 }
2782 };
2783 st.bool_acc = Some(st.bool_acc.map_or(b, |acc| acc && b));
2784 }
2785 // v7.17.0 — bool_or(p): TRUE iff any non-NULL input is
2786 // TRUE. NULL skipped.
2787 AggKind::BoolOr => {
2788 if is_null {
2789 return Ok(());
2790 }
2791 let b = match v {
2792 Value::Bool(b) => *b,
2793 other => {
2794 return Err(EvalError::TypeMismatch {
2795 detail: format!("bool_or requires bool, got {:?}", other.data_type()),
2796 });
2797 }
2798 };
2799 st.bool_acc = Some(st.bool_acc.map_or(b, |acc| acc || b));
2800 }
2801 // v7.32 (round-29) — variance / stddev family. Accumulate the
2802 // running sum (sum_float) and sum of squares (sum_sq) over the
2803 // non-NULL numeric inputs; finalize divides by n or n-1.
2804 AggKind::StddevFamily => {
2805 if is_null {
2806 return Ok(());
2807 }
2808 let x = match v {
2809 Value::Int(n) => f64::from(*n),
2810 Value::SmallInt(n) => f64::from(*n),
2811 Value::BigInt(n) => *n as f64,
2812 Value::Float(x) => *x,
2813 other => {
2814 return Err(EvalError::TypeMismatch {
2815 detail: format!("{name} needs numeric, got {:?}", other.data_type()),
2816 });
2817 }
2818 };
2819 st.count += 1;
2820 st.sum_float += x;
2821 st.sum_sq += x * x;
2822 }
2823 // v7.32 (round-29) — bitwise aggregates over integer inputs.
2824 AggKind::BitAnd | AggKind::BitOr | AggKind::BitXor => {
2825 if is_null {
2826 return Ok(());
2827 }
2828 let n = match v {
2829 Value::Int(n) => i64::from(*n),
2830 Value::SmallInt(n) => i64::from(*n),
2831 Value::BigInt(n) => *n,
2832 other => {
2833 return Err(EvalError::TypeMismatch {
2834 detail: format!("{name} needs integer, got {:?}", other.data_type()),
2835 });
2836 }
2837 };
2838 st.bit_acc = Some(match (st.bit_acc, kind) {
2839 (None, _) => n,
2840 (Some(acc), AggKind::BitAnd) => acc & n,
2841 (Some(acc), AggKind::BitOr) => acc | n,
2842 (Some(acc), _) => acc ^ n, // BitXor
2843 });
2844 }
2845 // v7.32 (round-29) — WITHIN GROUP aggregates (ordered-set +
2846 // hypothetical-set) collect the sort value (NULLs ignored, per
2847 // PG) into `items`, sorted at finalize by the parallel
2848 // `item_keys`.
2849 AggKind::WithinGroup => {
2850 if is_null {
2851 return Ok(());
2852 }
2853 st.items.push(v.clone().into_owned());
2854 if let Some(k) = order_keys {
2855 st.item_keys.push(k);
2856 }
2857 st.count += 1;
2858 }
2859 // v7.32 (round-29) — regression family f(Y, X). Only rows with
2860 // BOTH inputs non-NULL contribute (PG semantics). `v` is Y,
2861 // `arg2` is X.
2862 AggKind::Regression => {
2863 let (Some(y), Some(x)) = (agg_value_to_f64(v), arg2.and_then(agg_value_to_f64)) else {
2864 return Ok(()); // NULL (or non-numeric) in either input
2865 };
2866 st.reg_n += 1;
2867 st.reg_sx += x;
2868 st.reg_sy += y;
2869 st.reg_sxx += x * x;
2870 st.reg_syy += y * y;
2871 st.reg_sxy += x * y;
2872 }
2873 // v7.32 (round-29) — json_agg / jsonb_agg collect every input
2874 // (NULL becomes JSON null, per PG) in row order.
2875 AggKind::JsonAgg => {
2876 st.items.push(v.clone().into_owned());
2877 st.count += 1;
2878 }
2879 // v7.32 (round-29) — json_object_agg(key, value): keys in
2880 // `items`, values in `aux_items`. A NULL key is skipped (PG
2881 // raises; we drop it rather than abort the whole query).
2882 AggKind::JsonObjectAgg => {
2883 if is_null {
2884 return Ok(());
2885 }
2886 st.items.push(v.clone().into_owned());
2887 st.aux_items
2888 .push(arg2.cloned().map(Value::into_owned).unwrap_or(Value::Null));
2889 st.count += 1;
2890 }
2891 }
2892 Ok(())
2893}
2894
2895#[allow(clippy::cast_precision_loss)]
2896fn finalize(name: &str, st: &AggState) -> Value<'static> {
2897 match name {
2898 "count" | "count_star" => Value::BigInt(st.count),
2899 "sum" => {
2900 if st.count == 0 {
2901 Value::Null
2902 } else if st.use_float {
2903 Value::Float(st.sum_float + (st.sum_int as f64))
2904 } else {
2905 Value::BigInt(st.sum_int)
2906 }
2907 }
2908 "avg" => {
2909 if st.count == 0 {
2910 Value::Null
2911 } else {
2912 let total = if st.use_float {
2913 st.sum_float + (st.sum_int as f64)
2914 } else {
2915 st.sum_int as f64
2916 };
2917 Value::Float(total / (st.count as f64))
2918 }
2919 }
2920 "min" | "max" => st.extreme.clone().unwrap_or(Value::Null),
2921 // v7.17.0 — string_agg: join all collected text items with
2922 // the captured separator. Empty / all-NULL group → NULL
2923 // (PG semantics).
2924 "string_agg" => {
2925 if st.items.is_empty() {
2926 return Value::Null;
2927 }
2928 let sep = st.separator.clone().unwrap_or_default();
2929 let mut out = String::new();
2930 for (i, item) in st.items.iter().enumerate() {
2931 if i > 0 {
2932 out.push_str(&sep);
2933 }
2934 if let Value::Text(s) = item {
2935 out.push_str(s);
2936 }
2937 }
2938 Value::text(out)
2939 }
2940 // v7.17.0 — array_agg: collect into a typed array. NULL
2941 // elements are preserved per PG. Result type is decided
2942 // by the first non-NULL element seen (or Text fallback
2943 // when the whole group is NULL — PG would surface the
2944 // declared input type, but SPG hasn't yet wired the
2945 // aggregate's static input-type from `describe`).
2946 "array_agg" => {
2947 if st.items.is_empty() {
2948 return Value::Null;
2949 }
2950 let probe = st.items.iter().find(|v| !v.is_null());
2951 match probe.and_then(spg_storage::Value::data_type) {
2952 Some(DataType::Int) | Some(DataType::SmallInt) => {
2953 let items: Vec<Option<i32>> = st
2954 .items
2955 .iter()
2956 .map(|v| match v {
2957 Value::Int(n) => Some(*n),
2958 Value::SmallInt(n) => Some(i32::from(*n)),
2959 _ => None,
2960 })
2961 .collect();
2962 Value::IntArray(items)
2963 }
2964 Some(DataType::BigInt) => {
2965 let items: Vec<Option<i64>> = st
2966 .items
2967 .iter()
2968 .map(|v| match v {
2969 Value::BigInt(n) => Some(*n),
2970 _ => None,
2971 })
2972 .collect();
2973 Value::BigIntArray(items)
2974 }
2975 _ => {
2976 let items: Vec<Option<String>> = st
2977 .items
2978 .iter()
2979 .map(|v| match v {
2980 Value::Text(s) => Some(s.to_string()),
2981 Value::Null => None,
2982 other => Some(format!("{other:?}")),
2983 })
2984 .collect();
2985 Value::TextArray(items)
2986 }
2987 }
2988 }
2989 // v7.17.0 — bool_and / bool_or finalize: lazy-init pattern
2990 // means `None` is exactly "empty group or all-NULL", which
2991 // PG surfaces as SQL NULL.
2992 "bool_and" | "bool_or" => st.bool_acc.map_or(Value::Null, Value::Bool),
2993 // v7.32 (round-29) — variance / stddev. PG: `variance` ==
2994 // `var_samp`, `stddev` == `stddev_samp`. samp needs n >= 2
2995 // (n < 2 → NULL); pop needs n >= 1 (n == 1 → 0).
2996 "variance" | "var_samp" | "var_pop" | "stddev" | "stddev_samp" | "stddev_pop" => {
2997 let n = st.count;
2998 if n == 0 {
2999 return Value::Null;
3000 }
3001 let nf = n as f64;
3002 // Sum of squared deviations from the mean.
3003 let ss = st.sum_sq - (st.sum_float * st.sum_float) / nf;
3004 let pop = name.ends_with("_pop");
3005 let denom = if pop { nf } else { nf - 1.0 };
3006 if denom <= 0.0 {
3007 // var_samp / stddev (samp) with n == 1 → NULL.
3008 return Value::Null;
3009 }
3010 let var = (ss / denom).max(0.0); // clamp fp noise below 0
3011 if name.starts_with("stddev") {
3012 Value::Float(crate::eval::f64_sqrt(var))
3013 } else {
3014 Value::Float(var)
3015 }
3016 }
3017 // v7.32 (round-29) — bitwise aggregates: None (empty / all-NULL)
3018 // → SQL NULL.
3019 "bit_and" | "bit_or" | "bit_xor" => st.bit_acc.map_or(Value::Null, Value::BigInt),
3020 // v7.32 (round-29) — regression family. `regr_count` is the
3021 // paired n; everything else is NULL over an empty set. Terms
3022 // are the mean-centred sums of squares / cross-products.
3023 "regr_count" => Value::BigInt(st.reg_n),
3024 "covar_pop" | "covar_samp" | "corr" | "regr_avgx" | "regr_avgy" | "regr_slope"
3025 | "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy" => {
3026 let n = st.reg_n;
3027 if n == 0 {
3028 return Value::Null;
3029 }
3030 let nf = n as f64;
3031 let sxx = st.reg_sxx - st.reg_sx * st.reg_sx / nf;
3032 let syy = st.reg_syy - st.reg_sy * st.reg_sy / nf;
3033 let sxy = st.reg_sxy - st.reg_sx * st.reg_sy / nf;
3034 let avgx = st.reg_sx / nf;
3035 let avgy = st.reg_sy / nf;
3036 let out = match name {
3037 "regr_avgx" => Some(avgx),
3038 "regr_avgy" => Some(avgy),
3039 "regr_sxx" => Some(sxx),
3040 "regr_syy" => Some(syy),
3041 "regr_sxy" => Some(sxy),
3042 "covar_pop" => Some(sxy / nf),
3043 "covar_samp" => (n >= 2).then(|| sxy / (nf - 1.0)),
3044 "regr_slope" => (sxx != 0.0).then(|| sxy / sxx),
3045 "regr_intercept" => (sxx != 0.0).then(|| avgy - (sxy / sxx) * avgx),
3046 "corr" => {
3047 let d = sxx * syy;
3048 (d > 0.0).then(|| sxy / crate::eval::f64_sqrt(d))
3049 }
3050 // PG: NULL when sxx==0; 1 when syy==0 (and sxx>0).
3051 "regr_r2" => {
3052 if sxx == 0.0 {
3053 None
3054 } else if syy == 0.0 {
3055 Some(1.0)
3056 } else {
3057 Some((sxy * sxy) / (sxx * syy))
3058 }
3059 }
3060 _ => None,
3061 };
3062 out.map_or(Value::Null, Value::Float)
3063 }
3064 // v7.32 (round-29) — json_agg / jsonb_agg: a JSON array of every
3065 // collected element in row order; empty set → SQL NULL.
3066 "json_agg" | "jsonb_agg" => {
3067 if st.items.is_empty() {
3068 return Value::Null;
3069 }
3070 let mut out = String::from("[");
3071 for (i, item) in st.items.iter().enumerate() {
3072 if i > 0 {
3073 out.push_str(", ");
3074 }
3075 out.push_str(&crate::json::value_to_json_text(item));
3076 }
3077 out.push(']');
3078 Value::json(out)
3079 }
3080 // v7.32 (round-29) — json_object_agg: a JSON object built from
3081 // the parallel key (`items`) / value (`aux_items`) streams.
3082 "json_object_agg" | "jsonb_object_agg" => {
3083 if st.items.is_empty() {
3084 return Value::Null;
3085 }
3086 let mut out = String::from("{");
3087 for (i, key) in st.items.iter().enumerate() {
3088 if i > 0 {
3089 out.push_str(", ");
3090 }
3091 // Object keys are always JSON strings (PG coerces).
3092 let key_text = match key {
3093 Value::Text(s) | Value::Json(s) => s.to_string(),
3094 other => crate::json::value_to_json_text(other),
3095 };
3096 out.push_str(&crate::json::value_to_json_text(&Value::text(key_text)));
3097 out.push_str(": ");
3098 let val = st.aux_items.get(i).unwrap_or(&Value::Null);
3099 out.push_str(&crate::json::value_to_json_text(val));
3100 }
3101 out.push('}');
3102 Value::json(out)
3103 }
3104 // Ordered-set aggregates are finalized in `run` (they need the
3105 // sorted items + the direct fraction argument), never here.
3106 _ => unreachable!(),
3107 }
3108}
3109
3110/// v7.32 (round-29) — numeric coercion for the percentile interpolation.
3111fn agg_value_to_f64(v: &Value) -> Option<f64> {
3112 match v {
3113 Value::Int(n) => Some(f64::from(*n)),
3114 Value::SmallInt(n) => Some(f64::from(*n)),
3115 Value::BigInt(n) => Some(*n as f64),
3116 Value::Float(x) => Some(*x),
3117 _ => None,
3118 }
3119}
3120
3121/// v7.32 (round-29) — finalize a WITHIN GROUP aggregate. `st.items` is
3122/// already sorted by the `WITHIN GROUP (ORDER BY …)` spec. `direct` is
3123/// the evaluated direct argument: the fraction for `percentile_*`, the
3124/// hypothetical value for the hypothetical-set family (`rank` etc.),
3125/// and unused by `mode`. `order` is the (single) sort key, needed by
3126/// the hypothetical-set family to compare in the sort direction.
3127#[allow(
3128 clippy::cast_precision_loss,
3129 clippy::cast_possible_truncation,
3130 clippy::cast_sign_loss
3131)]
3132fn finalize_ordered_set(
3133 name: &str,
3134 st: &AggState,
3135 direct: Option<&Value>,
3136 order: Option<&spg_sql::ast::OrderBy>,
3137) -> Value<'static> {
3138 let fraction = direct;
3139 let items = &st.items;
3140 if items.is_empty() {
3141 // A hypothetical row ranks first over an empty group; the
3142 // distribution functions are 0 / divide-by-(n+1).
3143 return match name {
3144 "rank" | "dense_rank" => Value::BigInt(1),
3145 "percent_rank" => Value::Float(0.0),
3146 "cume_dist" => Value::Float(1.0),
3147 _ => Value::Null,
3148 };
3149 }
3150 let n = items.len();
3151 match name {
3152 // v7.32 (round-29) — hypothetical-set: the rank the direct value
3153 // would have if inserted into the group, in the sort direction.
3154 "rank" | "dense_rank" | "percent_rank" | "cume_dist" => {
3155 let Some(h) = fraction else {
3156 return Value::Null;
3157 };
3158 let (desc, nulls_first) = order.map_or((false, None), |o| (o.desc, o.nulls_first));
3159 let mut before = 0usize; // sort strictly before h
3160 let mut before_or_eq = 0usize; // sort before-or-peer with h
3161 let mut distinct_before = 0usize;
3162 let mut last_before: Option<&Value> = None;
3163 for it in items {
3164 match crate::order_by_value_cmp(desc, nulls_first, it, h) {
3165 core::cmp::Ordering::Less => {
3166 before += 1;
3167 before_or_eq += 1;
3168 if last_before
3169 .is_none_or(|p| value_cmp(p, it) != core::cmp::Ordering::Equal)
3170 {
3171 distinct_before += 1;
3172 last_before = Some(it);
3173 }
3174 }
3175 core::cmp::Ordering::Equal => before_or_eq += 1,
3176 core::cmp::Ordering::Greater => {}
3177 }
3178 }
3179 let nn = n as f64;
3180 match name {
3181 "rank" => Value::BigInt((before + 1) as i64),
3182 "dense_rank" => Value::BigInt((distinct_before + 1) as i64),
3183 "percent_rank" => Value::Float(before as f64 / nn),
3184 "cume_dist" => Value::Float((before_or_eq as f64 + 1.0) / (nn + 1.0)),
3185 _ => unreachable!(),
3186 }
3187 }
3188 // Most frequent value; equal values are adjacent in the sorted
3189 // run, and a frequency tie resolves to the earliest run (the
3190 // smallest value under an ascending sort), matching PG.
3191 "mode" => {
3192 let (mut best_i, mut best_cnt) = (0usize, 1usize);
3193 let (mut run_i, mut run_cnt) = (0usize, 1usize);
3194 for i in 1..n {
3195 if value_cmp(&items[i], &items[run_i]) == core::cmp::Ordering::Equal {
3196 run_cnt += 1;
3197 } else {
3198 run_i = i;
3199 run_cnt = 1;
3200 }
3201 if run_cnt > best_cnt {
3202 best_cnt = run_cnt;
3203 best_i = run_i;
3204 }
3205 }
3206 items[best_i].clone()
3207 }
3208 // The first value whose cumulative fraction reaches `f`.
3209 "percentile_disc" => {
3210 let f = fraction
3211 .and_then(agg_value_to_f64)
3212 .unwrap_or(0.0)
3213 .clamp(0.0, 1.0);
3214 let idx = if f <= 0.0 {
3215 0
3216 } else {
3217 (crate::eval::f64_ceil(f * n as f64) as usize)
3218 .saturating_sub(1)
3219 .min(n - 1)
3220 };
3221 items[idx].clone()
3222 }
3223 // Linear interpolation between the two bracketing values.
3224 "percentile_cont" => {
3225 let f = fraction
3226 .and_then(agg_value_to_f64)
3227 .unwrap_or(0.0)
3228 .clamp(0.0, 1.0);
3229 let Some(nums) = items
3230 .iter()
3231 .map(agg_value_to_f64)
3232 .collect::<Option<Vec<f64>>>()
3233 else {
3234 return Value::Null; // non-numeric ordered set
3235 };
3236 if n == 1 {
3237 return Value::Float(nums[0]);
3238 }
3239 let rank = f * (n as f64 - 1.0);
3240 let lo = crate::eval::f64_floor(rank) as usize;
3241 let hi = crate::eval::f64_ceil(rank) as usize;
3242 let frac = rank - lo as f64;
3243 Value::Float(nums[lo] + (nums[hi] - nums[lo]) * frac)
3244 }
3245 _ => unreachable!(),
3246 }
3247}
3248
3249fn infer_agg_type(spec: &AggSpec, schema_cols: &[ColumnSchema]) -> DataType {
3250 // v7.26 (round-20 C) — the argument's statically-derived shape
3251 // types MIN/MAX/SUM/array_agg properly; RowDescription used to
3252 // report TEXT for these, breaking every sqlx typed decode.
3253 let arg_ty = spec
3254 .arg
3255 .as_ref()
3256 .and_then(|a| crate::describe::describe_expr(a, schema_cols))
3257 .map(|shape| shape.ty);
3258 // v7.33 (array_agg argmax) — `(array_agg(x ORDER BY y))[1]` yields the
3259 // ELEMENT type (x), not the array type.
3260 if spec.first_ordered {
3261 return arg_ty.unwrap_or(DataType::Text);
3262 }
3263 match spec.name.as_str() {
3264 "count" | "count_star" => DataType::BigInt,
3265 "sum" => match arg_ty {
3266 Some(DataType::Float) => DataType::Float,
3267 _ => DataType::BigInt,
3268 },
3269 "avg" => DataType::Float,
3270 // v7.17.0 — string_agg always returns TEXT.
3271 "string_agg" => DataType::Text,
3272 "array_agg" => match arg_ty {
3273 Some(DataType::Int | DataType::SmallInt) => DataType::IntArray,
3274 Some(DataType::BigInt) => DataType::BigIntArray,
3275 _ => DataType::TextArray,
3276 },
3277 // v7.17.0 — boolean aggregates always return BOOL (nullable
3278 // — empty / all-NULL group → NULL).
3279 "bool_and" | "bool_or" => DataType::Bool,
3280 // v7.32 (round-29) — variance / stddev are floating point;
3281 // percentile_cont interpolates to float; the regression family
3282 // (except regr_count) is floating point.
3283 "stddev" | "stddev_samp" | "stddev_pop" | "variance" | "var_samp" | "var_pop"
3284 | "percentile_cont" | "covar_pop" | "covar_samp" | "corr" | "regr_avgx" | "regr_avgy"
3285 | "regr_slope" | "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy" => {
3286 DataType::Float
3287 }
3288 // v7.32 (round-29) — bitwise aggregates, regr_count, and the
3289 // integer hypothetical-set ranks return an integer.
3290 "bit_and" | "bit_or" | "bit_xor" | "regr_count" | "rank" | "dense_rank" => DataType::BigInt,
3291 // v7.32 (round-29) — hypothetical-set distribution functions.
3292 "percent_rank" | "cume_dist" => DataType::Float,
3293 // v7.32 (round-29) — JSON aggregates return JSON.
3294 "json_agg" | "jsonb_agg" | "json_object_agg" | "jsonb_object_agg" => DataType::Json,
3295 // min/max, percentile_disc, mode, and anything pass-through:
3296 // the argument's shape (for ordered-set aggs `spec.arg` is the
3297 // WITHIN GROUP value expression).
3298 _ => arg_ty.unwrap_or(DataType::Text),
3299 }
3300}
3301
3302fn agg_or_group_type(e: &Expr, synth: &[ColumnSchema]) -> DataType {
3303 if let Expr::Column(c) = e
3304 && let Some(s) = synth.iter().find(|s| s.name == c.name)
3305 {
3306 return s.ty;
3307 }
3308 // v7.26 (round-20 C) — compound expressions over aggregates
3309 // (COALESCE(BOOL_OR(…), false), (array_agg(…))[1], CASE …)
3310 // derive their shape statically against the synth schema; the
3311 // old Text fallback broke sqlx typed decodes of exactly these
3312 // columns.
3313 crate::describe::describe_expr(e, synth)
3314 .map(|shape| shape.ty)
3315 .unwrap_or(DataType::Text)
3316}
3317
3318fn rewrite_expr(e: &Expr, group_exprs: &[Expr], aggs: &[AggSpec]) -> Expr {
3319 // v7.33 (array_agg argmax) — `(array_agg(x ORDER BY y))[1]` rewrites
3320 // to its first_ordered synth column, consuming the subscript. Checked
3321 // before the AggregateOrdered/recursion arms (which would otherwise
3322 // rewrite the inner array_agg and leave the subscript). Same matcher
3323 // as collect_aggregates, so the spec it finds is the one collected.
3324 if let Some((arg, order_by, filter)) = first_ordered_array_agg(e) {
3325 let arg_owned = Some(arg.clone());
3326 let filter_owned = filter.cloned();
3327 for (i, spec) in aggs.iter().enumerate() {
3328 if spec.first_ordered
3329 && spec.name == "array_agg"
3330 && spec.arg == arg_owned
3331 && spec.order_by == *order_by
3332 && spec.filter == filter_owned
3333 {
3334 return Expr::Column(spg_sql::ast::ColumnName {
3335 qualifier: None,
3336 name: format!("__agg_{i}"),
3337 });
3338 }
3339 }
3340 }
3341 // v7.24 (round-16 A) — ordered aggregate: match on the inner
3342 // call PLUS the ordering keys.
3343 if let Expr::AggregateOrdered {
3344 call,
3345 order_by,
3346 distinct,
3347 filter,
3348 } = e
3349 && let Expr::FunctionCall { name, args } = call.as_ref()
3350 {
3351 let lower = name.to_ascii_lowercase();
3352 if is_aggregate_name(&lower) {
3353 let canonical: &str = if lower == "every" { "bool_and" } else { &lower };
3354 // Mirror collect_aggregates: ordered-set aggregates take the
3355 // value from the sort spec and the in-parens arg as direct.
3356 let (arg, direct_arg) = if is_within_group_name(canonical) {
3357 (
3358 order_by.first().map(|o| o.expr.clone()),
3359 args.first().cloned(),
3360 )
3361 } else {
3362 (args.first().cloned(), None)
3363 };
3364 let arg2 = if agg_uses_second_arg(canonical) {
3365 args.get(1).cloned()
3366 } else {
3367 None
3368 };
3369 let filter_owned = filter.as_deref().cloned();
3370 for (i, spec) in aggs.iter().enumerate() {
3371 if spec.name == canonical
3372 && spec.arg == arg
3373 && spec.arg2 == arg2
3374 && spec.distinct == *distinct
3375 && spec.order_by == *order_by
3376 && spec.filter == filter_owned
3377 && spec.direct_arg == direct_arg
3378 {
3379 return Expr::Column(spg_sql::ast::ColumnName {
3380 qualifier: None,
3381 name: format!("__agg_{i}"),
3382 });
3383 }
3384 }
3385 }
3386 }
3387 // Match aggregate FunctionCalls first — they sit outside group_by.
3388 if let Expr::FunctionCall { name, args } = e {
3389 let lower = name.to_ascii_lowercase();
3390 if is_aggregate_name(&lower) {
3391 let arg = if lower == "count_star" {
3392 None
3393 } else {
3394 args.first().cloned()
3395 };
3396 // v7.17.0 — match the spec we registered for
3397 // string_agg(value, separator) on the full pair; v7.32 also
3398 // the regression family and json_object_agg.
3399 let arg2 = if agg_uses_second_arg(&lower) {
3400 args.get(1).cloned()
3401 } else {
3402 None
3403 };
3404 // v7.17.0 — `every` collapses into `bool_and` at
3405 // collection; mirror that here so the rewrite finds
3406 // the matching synth column.
3407 let canonical: &str = if lower == "every" {
3408 "bool_and"
3409 } else {
3410 lower.as_str()
3411 };
3412 for (i, spec) in aggs.iter().enumerate() {
3413 if spec.name == canonical
3414 && spec.arg == arg
3415 && spec.arg2 == arg2
3416 && !spec.distinct
3417 && spec.order_by.is_empty()
3418 {
3419 return Expr::Column(spg_sql::ast::ColumnName {
3420 qualifier: None,
3421 name: format!("__agg_{i}"),
3422 });
3423 }
3424 }
3425 }
3426 }
3427 // Match a group_by expression by AST equality.
3428 for (i, g) in group_exprs.iter().enumerate() {
3429 if g == e {
3430 return Expr::Column(spg_sql::ast::ColumnName {
3431 qualifier: None,
3432 name: format!("__grp_{i}"),
3433 });
3434 }
3435 }
3436 // Recurse into children.
3437 match e {
3438 Expr::AggregateOrdered {
3439 call,
3440 order_by,
3441 distinct,
3442 filter,
3443 } => Expr::AggregateOrdered {
3444 call: Box::new(rewrite_expr(call, group_exprs, aggs)),
3445 distinct: *distinct,
3446 order_by: order_by
3447 .iter()
3448 .map(|o| spg_sql::ast::OrderBy {
3449 expr: rewrite_expr(&o.expr, group_exprs, aggs),
3450 desc: o.desc,
3451 nulls_first: o.nulls_first,
3452 })
3453 .collect(),
3454 // The filter is evaluated against SOURCE rows during
3455 // accumulation, never against synth rows — keep it as-is.
3456 filter: filter.clone(),
3457 },
3458 Expr::Binary { lhs, op, rhs } => Expr::Binary {
3459 lhs: Box::new(rewrite_expr(lhs, group_exprs, aggs)),
3460 op: *op,
3461 rhs: Box::new(rewrite_expr(rhs, group_exprs, aggs)),
3462 },
3463 Expr::Unary { op, expr } => Expr::Unary {
3464 op: *op,
3465 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3466 },
3467 Expr::Cast { expr, target } => Expr::Cast {
3468 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3469 target: target.clone(),
3470 },
3471 Expr::IsNull { expr, negated } => Expr::IsNull {
3472 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3473 negated: *negated,
3474 },
3475 Expr::FunctionCall { name, args } => Expr::FunctionCall {
3476 name: name.clone(),
3477 args: args
3478 .iter()
3479 .map(|a| rewrite_expr(a, group_exprs, aggs))
3480 .collect(),
3481 },
3482 Expr::Like {
3483 expr,
3484 pattern,
3485 negated,
3486 case_insensitive,
3487 } => Expr::Like {
3488 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3489 pattern: Box::new(rewrite_expr(pattern, group_exprs, aggs)),
3490 negated: *negated,
3491 case_insensitive: *case_insensitive,
3492 },
3493 Expr::Extract { field, source } => Expr::Extract {
3494 field: *field,
3495 source: Box::new(rewrite_expr(source, group_exprs, aggs)),
3496 },
3497 // v7.25.2 (round-19 A) — subquery nodes: rewrite group-key
3498 // references INSIDE the body to `__grp_N` so the correlated
3499 // resolver can substitute them against the synthesised group
3500 // row (aggs are NOT matched inside the body — a COUNT in the
3501 // subquery is the subquery's own aggregate).
3502 Expr::ScalarSubquery(s) => {
3503 Expr::ScalarSubquery(Box::new(rewrite_group_keys_in_select(s, group_exprs)))
3504 }
3505 Expr::Exists { subquery, negated } => Expr::Exists {
3506 subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
3507 negated: *negated,
3508 },
3509 Expr::InSubquery {
3510 expr,
3511 subquery,
3512 negated,
3513 } => Expr::InSubquery {
3514 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3515 subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
3516 negated: *negated,
3517 },
3518 // v4.12 window / Literal / Column — clone-pass (these don't
3519 // participate in aggregate rewrite).
3520 Expr::WindowFunction { .. } | Expr::Literal(_) | Expr::Placeholder(_) | Expr::Column(_) => {
3521 e.clone()
3522 }
3523 // v7.10.10 — recurse children for array nodes.
3524 Expr::Array(items) => Expr::Array(
3525 items
3526 .iter()
3527 .map(|elem| rewrite_expr(elem, group_exprs, aggs))
3528 .collect(),
3529 ),
3530 Expr::ArraySubscript { target, index } => Expr::ArraySubscript {
3531 target: Box::new(rewrite_expr(target, group_exprs, aggs)),
3532 index: Box::new(rewrite_expr(index, group_exprs, aggs)),
3533 },
3534 Expr::AnyAll {
3535 expr,
3536 op,
3537 array,
3538 is_any,
3539 } => Expr::AnyAll {
3540 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3541 op: *op,
3542 array: Box::new(rewrite_expr(array, group_exprs, aggs)),
3543 is_any: *is_any,
3544 },
3545 Expr::InList {
3546 expr,
3547 list,
3548 negated,
3549 } => Expr::InList {
3550 expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
3551 list: list
3552 .iter()
3553 .map(|item| rewrite_expr(item, group_exprs, aggs))
3554 .collect(),
3555 negated: *negated,
3556 },
3557 Expr::Case {
3558 operand,
3559 branches,
3560 else_branch,
3561 } => Expr::Case {
3562 operand: operand
3563 .as_deref()
3564 .map(|o| Box::new(rewrite_expr(o, group_exprs, aggs))),
3565 branches: branches
3566 .iter()
3567 .map(|(w, t)| {
3568 (
3569 rewrite_expr(w, group_exprs, aggs),
3570 rewrite_expr(t, group_exprs, aggs),
3571 )
3572 })
3573 .collect(),
3574 else_branch: else_branch
3575 .as_deref()
3576 .map(|e| Box::new(rewrite_expr(e, group_exprs, aggs))),
3577 },
3578 }
3579}
3580
3581/// v7.25.2 (round-19 A) — rewrite group-key references inside a
3582/// subquery body to `__grp_N` synthetic columns (aggregates are
3583/// not touched: empty spec list). Runs through the canonical
3584/// Select walker so every expression slot is covered.
3585fn rewrite_group_keys_in_select(
3586 s: &spg_sql::ast::SelectStatement,
3587 group_exprs: &[Expr],
3588) -> spg_sql::ast::SelectStatement {
3589 let mut out = s.clone();
3590 let _ = crate::walk_select_exprs_mut(&mut out, &mut |e| {
3591 *e = rewrite_expr(e, group_exprs, &[]);
3592 Ok(())
3593 });
3594 out
3595}
3596
3597/// Canonical string key for a tuple of group values. Used as map key.
3598/// Per-value group-key encoding (shared by owned and borrowed paths).
3599fn encode_one(out: &mut String, v: &Value) {
3600 use core::fmt::Write;
3601 match v {
3602 Value::Null => out.push_str("N|"),
3603 // v7.36 (perf — mailrs Phase 1) — switch the integer / float
3604 // encoders to `write!`. `n.to_string()` allocates a fresh
3605 // `String` per cell just to push its bytes into the
3606 // (already-cleared) reuse buffer — for the 25 k-row JOIN
3607 // probe in `count_messages` that's 25 k heap allocs per
3608 // query. `write!(&mut String, ...)` formats straight into
3609 // the buffer; no intermediate alloc.
3610 Value::SmallInt(n) => {
3611 let _ = write!(out, "s{n}|");
3612 }
3613 Value::Int(n) => {
3614 let _ = write!(out, "I{n}|");
3615 }
3616 Value::BigInt(n) => {
3617 let _ = write!(out, "B{n}|");
3618 }
3619 Value::Float(x) => {
3620 let _ = write!(out, "F{x}|");
3621 }
3622 Value::Bool(b) => {
3623 out.push(if *b { 'T' } else { 'f' });
3624 out.push('|');
3625 }
3626 Value::Text(s) => {
3627 out.push('S');
3628 out.push_str(s);
3629 out.push('|');
3630 }
3631 Value::Vector(v) => {
3632 out.push('V');
3633 for x in v.iter() {
3634 out.push_str(&x.to_string());
3635 out.push(',');
3636 }
3637 out.push('|');
3638 }
3639 // v6.0.1: GROUP BY on a `VECTOR(N) USING SQ8` column.
3640 // Two cells with byte-identical `(min, max, bytes)`
3641 // share the same group; equivalence is byte-equality
3642 // (same as f32 grouping today — neither path tries to
3643 // normalise nan/-0).
3644 Value::Sq8Vector(q) => {
3645 out.push('Q');
3646 out.push_str(&q.min.to_string());
3647 out.push('@');
3648 out.push_str(&q.max.to_string());
3649 out.push(':');
3650 for b in &q.bytes {
3651 out.push_str(&b.to_string());
3652 out.push(',');
3653 }
3654 out.push('|');
3655 }
3656 // v6.0.3: GROUP BY on a `VECTOR(N) USING HALF` column.
3657 // Byte-equality over the raw u16 bits; matches the SQ8
3658 // path's byte-key model.
3659 Value::HalfVector(h) => {
3660 out.push('H');
3661 for b in &h.bytes {
3662 out.push_str(&b.to_string());
3663 out.push(',');
3664 }
3665 out.push('|');
3666 }
3667 Value::Numeric { scaled, scale } => {
3668 out.push('D');
3669 out.push_str(&scaled.to_string());
3670 out.push('@');
3671 out.push_str(&scale.to_string());
3672 out.push('|');
3673 }
3674 Value::Date(d) => {
3675 out.push('d');
3676 out.push_str(&d.to_string());
3677 out.push('|');
3678 }
3679 Value::Timestamp(t) => {
3680 out.push('t');
3681 out.push_str(&t.to_string());
3682 out.push('|');
3683 }
3684 Value::Interval {
3685 months,
3686 days,
3687 micros,
3688 } => {
3689 out.push('i');
3690 out.push_str(&months.to_string());
3691 out.push('m');
3692 out.push_str(&days.to_string());
3693 out.push('d');
3694 out.push_str(µs.to_string());
3695 out.push('|');
3696 }
3697 Value::Json(s) => {
3698 out.push('j');
3699 out.push_str(s);
3700 out.push('|');
3701 }
3702 // v7.5.0 — Value is #[non_exhaustive] for downstream
3703 // forward-compat. Any future variant lacking explicit
3704 // handling here will share a debug-derived group key,
3705 // which is observably wrong but won't crash.
3706 _ => {
3707 out.push('?');
3708 out.push_str(&format!("{v:?}"));
3709 out.push('|');
3710 }
3711 }
3712}
3713
3714/// v7.30 (perf campaign) - encode from borrowed cells without
3715/// materialising an owned Vec<Value<'static>> first.
3716pub(crate) fn encode_key_refs(vals: &[&Value]) -> String {
3717 let mut out = String::new();
3718 for v in vals {
3719 encode_one(&mut out, v);
3720 }
3721 out
3722}
3723
3724/// v7.31 (perf 3e) — encode into a caller-owned scratch buffer.
3725/// The per-row key paths (group hash, DISTINCT set, join build/
3726/// probe) ran 24k+ String allocations per query through the
3727/// allocator just to LOOK UP a map; the scratch form allocates
3728/// only when a map actually has to take ownership (vacant insert).
3729pub(crate) fn encode_key_refs_into(vals: &[&Value], out: &mut String) {
3730 out.clear();
3731 for v in vals {
3732 encode_one(out, v);
3733 }
3734}
3735
3736pub(crate) fn encode_key(vals: &[Value<'static>]) -> String {
3737 let mut out = String::new();
3738 for v in vals {
3739 encode_one(&mut out, v);
3740 }
3741 out
3742}
3743
3744#[allow(clippy::cast_precision_loss)]
3745fn value_cmp(a: &Value, b: &Value) -> core::cmp::Ordering {
3746 use core::cmp::Ordering::Equal;
3747 match (a, b) {
3748 (Value::Null, Value::Null) => Equal,
3749 (Value::Null, _) => core::cmp::Ordering::Greater, // NULLs last
3750 (_, Value::Null) => core::cmp::Ordering::Less,
3751 (Value::Int(x), Value::Int(y)) => x.cmp(y),
3752 (Value::BigInt(x), Value::BigInt(y)) => x.cmp(y),
3753 (Value::Int(x), Value::BigInt(y)) => i64::from(*x).cmp(y),
3754 (Value::BigInt(x), Value::Int(y)) => x.cmp(&i64::from(*y)),
3755 (Value::Float(x), Value::Float(y)) => x.partial_cmp(y).unwrap_or(Equal),
3756 (Value::Int(x), Value::Float(y)) => f64::from(*x).partial_cmp(y).unwrap_or(Equal),
3757 (Value::Float(x), Value::Int(y)) => x.partial_cmp(&f64::from(*y)).unwrap_or(Equal),
3758 (Value::BigInt(x), Value::Float(y)) => (*x as f64).partial_cmp(y).unwrap_or(Equal),
3759 (Value::Float(x), Value::BigInt(y)) => x.partial_cmp(&(*y as f64)).unwrap_or(Equal),
3760 (Value::Text(x), Value::Text(y)) => x.cmp(y),
3761 (Value::Bool(x), Value::Bool(y)) => x.cmp(y),
3762 _ => Equal,
3763 }
3764}
3765
3766/// v7.37.9 Phase 0 diagnostic counters — see
3767/// `.claude/notes/v7.37.9-class-a-c-cascade-closure-plan.md`. These
3768/// are read-only telemetry, do not gate any code path. Used by
3769/// `xtests/dogfood_replay/src/bin/counter_dump.rs` to verify
3770/// whether the DISTA A-3 + array_agg-ordered fast paths actually
3771/// fire on the mailrs Class A SQL shape.
3772pub static DISTA_LITERAL_ARG2_CACHE_FIRE: core::sync::atomic::AtomicU64 =
3773 core::sync::atomic::AtomicU64::new(0);
3774pub static AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE: core::sync::atomic::AtomicU64 =
3775 core::sync::atomic::AtomicU64::new(0);
3776
3777/// v7.37.9 Phase 1A-ext — per-row spec dispatch branches in
3778/// `accumulate_groups`'s hot loop. Verifies the Phase 1A
3779/// decomposition agent's S06 assumption ("14 specs × eval_expr per
3780/// row"). Sum should equal `n_specs × n_input_rows`. Branch
3781/// distribution tells which attack target ROI is highest:
3782/// FAST_POS many = baseline OK; COMPILED_MISS many = Step-VM is
3783/// hot path; EVAL_FALLBACK > 0 = uncompilable specs walking the
3784/// eval_expr tree per row × Cow row materialise.
3785pub static AGG_PER_ROW_FAST_POS: core::sync::atomic::AtomicU64 =
3786 core::sync::atomic::AtomicU64::new(0);
3787pub static AGG_PER_ROW_COMPILED_HIT: core::sync::atomic::AtomicU64 =
3788 core::sync::atomic::AtomicU64::new(0);
3789pub static AGG_PER_ROW_COMPILED_MISS: core::sync::atomic::AtomicU64 =
3790 core::sync::atomic::AtomicU64::new(0);
3791pub static AGG_PER_ROW_EVAL_FALLBACK: core::sync::atomic::AtomicU64 =
3792 core::sync::atomic::AtomicU64::new(0);
3793pub static AGG_PER_ROW_COUNT_STAR_SENTINEL: core::sync::atomic::AtomicU64 =
3794 core::sync::atomic::AtomicU64::new(0);