v2.9.0

docs/docs/architecture/pipeline.md

@@ -211,6 +211,20 @@ The join pipeline:
 2. **Build** — Build per-partition hash tables (each fits in L2). For inner joins, the executor selects the build side at runtime using actual materialized row counts: the smaller input becomes the build side, keeping hash tables as compact as possible. LEFT, FULL, and ANTI joins always build on the right to preserve left-row semantics. The small-input (chained) path also always builds on the right. During the per-partition open-addressing build, the executor tracks per-key duplicate counts; when a single key exceeds the duplication threshold (`RADIX_DUP_RUN_MAX = 512`), it abandons the radix attempt and re-runs the whole join through the chained hash table, which is O(n) regardless of duplication. No join (INNER, LEFT, or FULL) can degrade to quadratic build cost on a skewed key.
 3. **Probe** — Probe partitions in parallel across worker threads. Inner-join output order is partition- and thread-dependent; it is not guaranteed to be stable.
 
+### Group-By Aggregation
+
+Grouped aggregates run on the parallel aggregation engine, which picks a strategy from the key columns and the aggregates:
+
+- **Dense direct index** — integer, temporal, and symbol keys whose value ranges pack into a bounded slot space (a group id is the mixed-radix offset of the key tuple, no hashing). Composite keys whose raw ranges multiply out too far are first *compacted*: one parallel pass records the codes each key actually uses, and the plan maps each key through a code-to-component table. This matters for symbol keys, which share one domain and therefore interleave their codes with every other symbol column's.
+- **Task-local slabs** — each task accumulates into its own dense slab, merged once at the end. The number of replicated slabs is bounded by the last-level cache: every task updates random slots of its slab, so once the slabs together outgrow the cache each update misses to memory and more tasks make the query slower. The engine reads the cache size at startup (`ray_cache_llc_bytes`) and keeps replicated slabs within three quarters of it; when fewer than three slabs fit, it switches to partition ownership.
+- **Partition ownership** — rows are scattered by slot into partitions that each own a cache-sized slab; every group is reduced exactly once, with no merge. Used for large dense domains and for shared extrema that concurrent updates handle directly.
+- **Radix** — unbounded integer or symbol keys (many-million-group inputs) are hash-partitioned and reduced per partition.
+- **Shared directory** — float, string, GUID, and list keys use a shared parallel key directory.
+
+Arithmetic over aggregates (`(- (max v1) (min v2))`, `(pow (pearson_corr a b) 2)`) is decomposed at compile time: the aggregates run as hidden slots inside the same group pass, and the outer expression is evaluated once over the grouped result. This applies to any number of keys and to binary aggregates.
+
+Ordered top-N clauses (`desc: c take: 10`, `asc: m take: 5`) are selected inside the strategies: radix partitions and dense finishes finalize the ordering aggregate to one value per group in parallel, keep a bounded candidate set each, take the threshold from the union, and emit only the kept groups (ties included), so a 10M-group query never materializes 10M output rows. Routes that emit every group trim afterwards with the same decision. An unordered `take: N` emits the first N groups in first-seen order; on the dense task-local path that is a selection of the N smallest first rows. The radix full path restores first-seen order with a scatter, count and compaction that are all dispatched across the pool.
+
 ### Per-Thread Heaps
 
 Each worker thread has its own heap (`heap_id` in the pool header). Vectors allocated during parallel execution are tagged with their owning heap via the pool they reside in. Cross-heap frees are deferred to a lock-free LIFO and reclaimed when the owning heap flushes. See [Memory Model](memory.md) for details.

docs/docs/queries/select.md

@@ -121,6 +121,13 @@ Group by multiple keys:
 ; GOOG Buy    125
 ```
 
+Expressions over aggregates work with any number of keys and with binary aggregates such as `pearson_corr`, `cov`, or `wsum`; the aggregates are computed in the grouping pass and the outer expression once per group:
+
+```lisp
+(select {from: trades spread: (- (max price) (min price)) by: {sym: sym venue: venue}})
+(select {from: trades r2: (pow (pearson_corr price size) 2) by: {sym: sym}})
+```
+
 ### Filter + Group-by
 
 Filter first, then group by. Chain two `select` calls for correct results:

docs/grouping-engine-scaling-plan.md

@@ -253,3 +253,79 @@ buffer for binary and mixed streaming reductions. The existing native-width
 readers, source selection and partition ownership remain shared. Chunk sizing
 bounds the active payload footprint; scratch consists of partition cursors,
 without another row-sized buffer.
+
+## Follow-up: core scaling on cache-bounded and legacy-routed shapes
+
+A scaling sweep over 1/8/16/28 threads on a 10M-row grouping input found five
+reasons queries stopped speeding up, or slowed down, as cores were added. None
+was in the pool; all were routing or footprint decisions.
+
+- **Replicated slab footprint.** The task-local dense strategy gave every worker a
+  full slab, and the strategy choice was deliberately independent of cache size.
+  Once the slabs together outgrew the last-level cache, every update missed to
+  memory: a 100k-group sum measured 6 ms with 8 slabs and 21 ms with 28 slabs on
+  the same pool. The runtime now reads the last-level cache size (summed over
+  every cache instance, so multi-die parts count each die) and bounds replicated
+  slabs to three quarters of it; below three slabs it prefers partition
+  ownership, whose per-partition slabs are cache-sized by construction. The
+  memory budgets stay data-derived; the cache bound only limits replication.
+  Task assignment stays static so floating-point sums remain deterministic for a
+  given pool size.
+- **Multi-key arithmetic over aggregates.** Any non-aggregate output on a
+  two-or-more-key `by:` forced eval-level grouping (no parallelism, ~850 ms),
+  because that decision ran before the arithmetic-over-aggregates decomposition.
+  Routing now probes decomposability first, and the hidden-slot decomposition
+  also admits binary aggregates and literal-probability quantiles.
+- **Top-N emit filter.** `desc: AGG take: N` armed an emit filter the parallel
+  engine does not implement, so every such query ran on the legacy ladder
+  (0.1x parallel at 28 threads for a three-aggregate query). Shapes the
+  parallel engine admits with a bounded dense plan now run there and are
+  trimmed to the top-N superset; unbounded key domains stay on the ladder
+  because the radix route's first-seen ordering and emission tail is still
+  serial-heavy there.
+- **Filtered prescan.** The selected-row dense plan walked the selection
+  serially (~13 ms per 10M rows at any core count); it is now split across the
+  pool.
+- **Composite symbol keys.** Symbol columns share one domain, so their codes
+  interleave and a two-key raw range product overflowed the dense limit,
+  sending 10k-group queries to radix (505 ms vs 52 ms on one core). Plans now
+  compact keys to the codes they use when the raw product overflows or exceeds
+  65,536 slots. The hot loops select the compacted or raw slot computation once
+  per run: a per-row check made the two-key loop 12x slower.
+
+Still open: the radix route's post-reduce stages (an input-sized order map to
+restore first-seen order, then emission) scale poorly on many-million-group
+keys; the legacy ladder remains faster for those top-N shapes. An unordered
+`take:` on a grouped select still uses radix's bounded emit rather than the
+dense plan.
+
+## Follow-up results: top-N, bounded emit and radix ordering
+
+Five more changes closed the shapes the previous follow-up left open. The
+parallel engine now owns the top-N emit filter: radix partitions and dense
+finishes finalize the ordering aggregate per group in parallel, keep bounded
+candidate heaps, take one threshold from their union and emit only the kept
+superset; every other route trims its full result with the same decision,
+and the legacy ladder's carve-outs are gone. The radix full path's first-seen
+ordering was dispatched by element grain over partitions and chunks, which
+produced a single task; it is dispatched by task count and compacts out of
+place. An unordered `take: N` on a bounded key selects the N smallest first
+rows on the dense task-local path instead of the radix bounded emit. A test
+driver flag records which `.rfl` lines still reach the legacy ladder
+(`docs/grouping-legacy-census.md`).
+
+Measured on 10M rows (min of five warm runs, ms; before = the branch point,
+after = this follow-up):
+
+| Query | 1 core | 8 cores | 28 cores |
+|---|---|---|---|
+| three-key count, desc take 10 (10M groups) | 962 → 285 | 46 → 54 | 46 → 44 |
+| three aggregates by 100k key, desc take 10 | 77 → 41 | 12.8 → 11.0 | 75 → 10.4 |
+| where + count by 15-value key, desc take 10 | 451 → 51 | 59 → 9.6 | 23.5 → 6.3 |
+| count by 100k key, unordered take 10 | 107 → 17 | 21 → 4.0 | 17.8 → 5.4 |
+| count by 100k key, desc take 10 | 28 → 17 | 4.9 → 4.0 | 4.8 → 5.5 |
+| sum + count by six keys, no take (10M groups) | 1543 → 1592 | 245 → 201 | 187 → 137 |
+| pow(pearson) by two keys | 1840 → 68 | 993 → 12 | 936 → 9.2 |
+
+The remaining legacy routes are enumerated in the census; none of them is a
+top-N, bounded-emit or compound-expression shape.

docs/grouping-legacy-census.md

@@ -0,0 +1,56 @@
+# Grouping: census of shapes still served by the legacy ladder
+
+Status: recorded 2026-09-19 after the core-scaling follow-up. The parallel
+grouping engine ("v2") now owns every top-N emit filter, unordered bounded
+emit, multi-key arithmetic over aggregates, and composite symbol keys. This
+census lists what still reaches the legacy grouping ladder in
+`exec_group_run`, as the factual starting point for retiring it.
+
+## How it was measured
+
+The test driver records, for every evaluated `.rfl` line, whether the
+grouping executor took the legacy route and the admission reason it was
+given:
+
+```
+make test                     # builds ./rayforce.test
+RAYFORCE_CORES=2 ./rayforce.test --census census.tsv -f rfl/
+cut -f1 census.tsv | sort | uniq -c
+```
+
+Each census line is `reason<TAB>file:line<TAB>source`. Run it with the
+suite's worker count: one admission reason (`parallel_wide`) depends on the
+pool size, and several ordering assertions in the corpus assume two workers.
+
+## Results (514 `.rfl` files, 340 legacy hits)
+
+| Reason | Lines | With `by:` | What it means | What v2 needs |
+|---|---:|---:|---|---|
+| `shape` | 200 | 5 | 195 are scalar aggregations with no keys (`select {s: (sum v) from: T}`), mostly over partitioned stores; the 5 grouped ones are nested selects whose outer scalar select carries the reason, plus one `by:` on a missing column | A keyless (single-group) path in v2, or routing scalar aggregation to the vector aggregators; the grouped 5 are already served |
+| `agg_expression` | 63 | 63 | Aggregate over an expression: `(sum (strlen s))` ×35, `(sum (at ...))` ×15, `(first (at ...))` ×14, `(avg (strlen s))` ×11, `(count (select ...))` ×7 | The expression-input path (`exec_group_v2_exprs`) materializes inputs it can evaluate as a full column; string-length and indexed/nested inputs are declined today |
+| `key_expression` | 31 | 27 | Computed keys: `(+ k 1)` ×10, `(xbar ts N)` ×5, `(substr ref 0 3)` ×2, `(minute EventTime)`, dotted temporal accessors `ts.date` | Materialize computed keys as scan columns before admission, the way expression aggregate inputs are; temporal accessors need the same |
+| `parallel_wide` | 30 | 30 | Float keys with 2 to 8 workers and no indexed aggregate (`by: {f: f}` on F64) — a measured choice to prefer the ladder's directory over v2's replicated float hash | Make the v2 float route competitive at small pools (or drop the pool-size gate once measured) |
+| `admitted` | 16 | 15 | Admitted by v2 but executed on the ladder: partitioned-store sources (`from: Pmc by: date`) where the per-partition executor merges partial groups on the legacy path, and one derived-key symbol-domain case | A partition-aware merge in v2 (streaming states merge; buffered ones need the ladder's row slices) |
+
+No line reached the ladder because of an emit filter, a bounded emit, or a
+multi-key compound expression; those classes are closed.
+
+## Reading the table
+
+- Two classes are not grouping at all: keyless scalar aggregation (195 lines)
+  and the outer scalar select of a nested query. They keep the ladder alive
+  for reasons unrelated to group-by scaling and can be moved independently.
+- The grouped remainder is 140 lines in four classes. Three of them
+  (`agg_expression`, `key_expression`, `admitted` on partitioned sources) are
+  admission gaps that a materialize-then-scan step closes without a new
+  strategy; `parallel_wide` is a performance gate that needs a measurement,
+  not a feature.
+- The benchmark query files add two shapes the corpus lacks: a computed key
+  through several nested string functions, and a `where:` filter on the
+  grouped result (`(> c 100000)`), which the emit filter already expresses.
+
+## Next step
+
+Retiring the ladder is its own plan: close the four grouped classes above,
+route keyless aggregation to the vector aggregators, then delete
+`exec_group_run` and the consumers it alone uses.