perf(compute): dict code-scan lane in filteredAggregate

Extends the DictFilter lane to the multi-column kernel behind the
Calcite boundary aggregate push-down: a filter that is a single
comparison leaf over a dict-encoded column (bare or wrapped in a
one-element AND) scans raw u8 codes per chunk run instead of decoding
every row through the per-element RowPredicate. COUNT(*) folds as a
bare count of code matches; the full fold mirrors the kernel's own
lanes exactly (widenLong + unsigned-aware min/max in the long domain,
Double.compare order and Float extreme boxing in the double domain).
Multi-leaf filters, null tests, and non-long/double aggregates decline
to the unchanged RowPredicate path.

fusedFilteredAggregateDict (100M rows): 982.5 -> 178.3 ms/op (~5.5x).
The gap to the sum lane's 25.5 ms is the fuller per-match fold (sum,
min, max, two counters) keeping the loop body scalar — recorded in ADR
0013 as an optional follow-up. The filteredAggregate oracle test now
randomly dict-encodes its filter columns, sweeping the lane against
the unchanged brute-force reference, plus pinned corners: COUNT(*)
over codes, u64 unsigned aggregate order, f32 Float boxing, one-leaf
AND unwrap.

Also drafts ADR 0019 (Proposed): the columnar transducer facade —
declarative column-bound stages compiled to one fused pass, preserving
the lane-dispatch model this change extends.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: dfa1

TODO.md

@@ -96,10 +96,14 @@ Per-encoding gotchas:
   interfaces) were built then removed for the fused single-pass kernels — no intermediate bitmap.
   Encoded-domain value specialization was measured as a no-win (decode is dispatch-bound, see
   `forLoopDictEncoded`); the real dict lever — the monomorphic code-segment scan — shipped as the
-  `DictFilter` lane in `FusedFilterSum` (762 → 25.5 ms/op on the 100M-row `fusedFilteredSumDict`).
-  Next: (1) extend the `DictFilter` code-scan lane to `FusedFilterAggregate` (the Calcite boundary
-  push-down kernel still reads dict filters through the per-element accessor; its single-leaf case
-  can reuse the run-based scan directly); (2) the columnar transducer façade (ADR 0019).
+  `DictFilter` lane in both kernels (`fusedFilteredSumDict` 762 → 25.5 ms/op ≈ 30×;
+  `fusedFilteredAggregateDict` 983 → 178 ms/op ≈ 5.5×, single-leaf filters and `COUNT(*)`).
+  Next: (1) the columnar transducer façade — [ADR-0019](docs/adr/0019-columnar-transducer-facade.md)
+  drafted (Proposed): declarative column-bound stages compiled to one fused pass, multi-aggregate
+  folds for the Calcite boundary tier; review, then implement. (2) Optional: specialize the dict
+  aggregate lane's no-null long fold to close the 178 → 25 ms gap if the boundary tier needs it;
+  (3) extend the dict lane to multi-leaf `AND`s (dict leaf as the driving scan, residual leaves
+  tested per match).
 
 ## Encodings
 

docs/adr/0013-compute-primitives.md

@@ -85,10 +85,14 @@ backing segment — no per-element `findChunk` binary search, no broadcast-guard
 | `fusedFilteredSumDict` before (accessor chain per row) | 762.5 ± 7.4 ms/op |
 | `forLoopDictSegment` (raw segment scan ceiling) | 24.8 ms/op |
 | `fusedFilteredSumDict` after (`DictFilter` lane) | 25.5 ± 2.5 ms/op |
-
-The lane reproduces the raw-scan ceiling (≈ 30×) with no measurable kernel overhead. It covers
-`Compute.filteredSum`; extending it to `FusedFilterAggregate`'s per-leaf path is the remaining
-follow-up (tracked in TODO).
+| `fusedFilteredAggregateDict` before (per-leaf `RowPredicate`) | 982.5 ± 30.2 ms/op |
+| `fusedFilteredAggregateDict` after (`DictFilter` lane) | 178.3 ± 4.1 ms/op |
+
+The `filteredSum` lane reproduces the raw-scan ceiling (≈ 30×) with no measurable kernel overhead.
+The `filteredAggregate` lane (a single comparison leaf driving the same code scan, extended to the
+full `SUM`/`MIN`/`MAX`/count fold and `COUNT(*)`) lands at ≈ 5.5× — off the pure-scan ceiling
+because the per-match fold (sum, min, max, two counters) keeps its loop body scalar; specializing
+the common no-null long-aggregate fold is a possible follow-up if the boundary tier ever needs it.
 
 ## Context
 

docs/adr/0019-columnar-transducer-facade.md

@@ -0,0 +1,136 @@
+# ADR 0019: Columnar transducer façade for compute
+
+- **Status:** Proposed
+- **Date:** 2026-07-02
+- **Deciders:** project maintainer
+- **Supersedes:** —
+- **Superseded by:** —
+
+## Context
+
+ADR 0013 shipped the compute primitives as two fused kernels behind `reader.compute.Compute`:
+`filteredSum(filterColumn, predicate, aggColumn)` and `filteredAggregate(chunk, rowFilter,
+aggColumn)`. They are deliberately minimal — static entry points, package-private kernels — so
+early callers would not couple to a kernel shape before the ergonomics were settled. That bet paid
+off twice:
+
+1. The original §1–§3 design (a materialized `Mask` between filter and reduce) was built and then
+   **removed**: a selection bitmap between stages is strictly slower than fusing the stages into one
+   pass. Whatever the public API looks like, execution must stay single-pass.
+2. The dict code-scan lane (`DictFilter`) showed that the wins now come from **encoding-aware
+   structural dispatch**: the kernel inspects the filter column's shape (an offset slice over a
+   dict view with `u8` codes) and swaps the whole loop, not the per-row callback. Measured on 100M
+   rows: 762 → 25.5 ms/op for `filteredSum`, 983 → 178 ms/op for `filteredAggregate` (the fuller
+   per-match fold — sum, min, max, two counters — keeps the aggregate lane off the pure-scan
+   ceiling). A per-row lambda (`RowPredicate`) cannot express that — the lane needs to see the
+   *description* of the filter, not a compiled `test(i)`.
+
+Meanwhile the caller side is growing. The Calcite adapter (ADR 0018) invokes `filteredAggregate`
+from the boundary-chunk tier of the aggregate push-down; each new capability (a second aggregate
+column, `GROUP BY` push-down Phase 3, a projection) would today mean another bespoke static method
+on `Compute` and another hand-wired call site in the push-down rule. There is no way to *compose* a
+chunk-level computation — filter by two columns, fold three aggregates — without either a new
+kernel entry point per combination or falling back to per-row accessor loops, which the dict-lane
+numbers show cost 30× on encoded columns.
+
+The goal is an ergonomic public compute API that composes, while preserving both hard-won execution
+properties: single-pass fusion and encoding-aware lane dispatch. Explicitly **not** a goal: a query
+engine. Join, sort, expression trees, and query planning belong to Calcite; this façade only makes
+the per-chunk leaf computation composable and fast (see `project_compute_facade`: the aim is faster
+Calcite push-down).
+
+## Decision
+
+Introduce a **columnar transducer**: an immutable, declarative description of a per-chunk pipeline
+— column-bound filter stages and aggregate folds — that a terminal operation compiles into one
+fused pass. "Transducer" in the original sense: the transformation is described independently of
+the source and reused across chunks; it is *not* the Clojure-style composition of per-element
+functions, precisely because per-element functions are what the kernels must never see.
+
+Sketch (names to be settled during implementation):
+
+```java
+ChunkFold fold = ChunkFold.builder()
+        .filter("category", Predicate.eq(7L))         // stages are DATA: column + Predicate
+        .filter("price", Predicate.gt(500.0))
+        .aggregate("measure", Fold.SUM, Fold.MIN)      // one or many folds, one or many columns
+        .aggregate("volume", Fold.SUM)
+        .build();                                      // lowered/validated once
+
+FilteredAggregate measure = fold.apply(chunk).column("measure");
+```
+
+Design rules, in force from the first commit:
+
+- **Stages are data, never lambdas.** A filter stage is a `(column, Predicate)` pair — the same
+  `Predicate` vocabulary as `RowFilter` and the zone maps. An aggregate stage is a `(column, fold
+  kind)` pair. Because stages are inspectable, the terminal compile step can route each filter to
+  its best lane (dict code-scan, primitive typed-accessor, generic boxing) exactly as
+  `FusedFilterSum` does today — and future lanes (a `RunEnd` run-scan, a `Constant` short-circuit)
+  drop in without touching the API.
+- **Terminal compilation, single pass.** `apply(chunk)` lowers every stage once (reusing
+  `PrimitiveFilter` / `DictFilter` lowering — the single source of truth), chooses the driving lane,
+  and folds everything in one scan with no intermediate bitmap or materialized selection. The
+  n-ary `AND` of filter stages keeps the existing three-valued-logic semantics bit-identical to
+  `filteredAggregate`.
+- **The existing `Compute` statics become thin wrappers** over one-stage pipelines and are kept:
+  they are shipped API and the right call for the simple cases. The kernels stay package-private.
+- **Scope: fold-shaped terminals only.** Filters + aggregate folds (`SUM` / `MIN` / `MAX` /
+  `COUNT`), matching what the Calcite boundary tier consumes. A row-materializing terminal
+  (project/collect) is explicitly out of scope until a concrete consumer exists — it is where
+  "façade" would start sliding toward "query engine".
+
+The Calcite `VortexAggregatePushDownRule` boundary tier is the first consumer: it currently loops
+aggregate columns one `filteredAggregate` call per column; a multi-aggregate pipeline folds them in
+one scan of the filter column.
+
+## Consequences
+
+### Positive
+
+- Composition without new entry points: k filters × m aggregates is one pipeline, one scan, one
+  API — not `k×m` static methods.
+- Encoding-aware lanes keep working and keep winning: the declarative stages preserve exactly the
+  information (`Predicate` + column shape) that made the 30× dict lane possible.
+- The Calcite boundary tier stops re-scanning the filter column once per aggregate column.
+- One lowering path shared with `RowFilter` / zone maps — no semantic drift between pruning and
+  row-level evaluation.
+
+### Negative
+
+- A builder API is more surface than two statics; it must be documented and versioned as public
+  API (the "small public APIs" rule says: start minimal, grow only on demonstrated need).
+- Multi-aggregate folds widen the kernel's inner loop (several accumulators per selected row);
+  the single-aggregate fast shape must not regress — benchmark before accepting.
+
+### Risks to manage
+
+- **Lambda creep.** The moment a stage accepts a user lambda, lane dispatch dies (megamorphic
+  per-row calls, the exact failure ADR 0013 measured). The API must not offer one, even as a
+  convenience overload.
+- **Query-engine scope creep.** `GROUP BY`, joins, expressions stay in Calcite. The pipeline's
+  vocabulary is bounded by what the kernels can fuse in one pass.
+- **Premature generality.** Ship with the Calcite boundary tier as the proving consumer; grow the
+  vocabulary only when a second concrete consumer needs it.
+
+## Alternatives considered
+
+- **`java.util.stream.Stream` over rows.** Idiomatic, but value-level: per-row boxing and
+  megamorphic per-element calls — the measured 30× loss on encoded columns — and no way to see the
+  predicate's description for lane dispatch or zone pruning.
+- **Clojure-style function transducers** (composed `(acc, row) -> acc` reducers). Same per-element
+  dispatch problem; composition happens at the wrong level (values, not columns).
+- **Keep growing `Compute` statics.** No composition; every filter/aggregate combination is a new
+  method and a new hand-fused kernel; the Calcite rule accumulates call-site switches. This is the
+  status quo the façade replaces.
+- **Expose the kernels directly.** Couples callers to loop shapes that ADR 0013 already replaced
+  once (Mask → fused); the package-private kernel rule exists to keep that freedom.
+
+## References
+
+- [ADR 0013](0013-compute-primitives.md) — the fused kernels, the removed `Mask` design, and the
+  measured dict code-scan results this façade must preserve.
+- [ADR 0018](0018-calcite-sql-adapter.md) — the aggregate push-down boundary tier, the first
+  consumer.
+- `reader.compute.DictFilter` — the encoding-aware lane whose dispatch model the stage-as-data rule
+  protects.

docs/adr/ADR.md

@@ -33,3 +33,4 @@ the decision shipped in (blank = not yet shipped).
 | 0016 | vortex-arrow bridge module for Arrow interop  | Proposed  |          |
 | 0017 | In-house FlatBuffers codegen + MemorySegment runtime | Accepted  |          |
 | 0018 | Apache Calcite SQL adapter — push-down source | Accepted  |          |
+| 0019 | Columnar transducer façade for compute        | Proposed  |          |

performance/src/main/java/io/github/dfa1/vortex/performance/ComputeKernelBenchmark.java

@@ -3,6 +3,7 @@
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.reader.Chunk;
 import io.github.dfa1.vortex.reader.ReadRegistry;
+import io.github.dfa1.vortex.reader.RowFilter;
 import io.github.dfa1.vortex.reader.VortexReader;
 import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.ByteArray;
@@ -16,6 +17,7 @@
 import io.github.dfa1.vortex.reader.array.OffsetDoubleArray;
 import io.github.dfa1.vortex.reader.array.OffsetLongArray;
 import io.github.dfa1.vortex.reader.compute.Compute;
+import io.github.dfa1.vortex.reader.compute.FilteredAggregate;
 import io.github.dfa1.vortex.reader.compute.Predicate;
 import io.github.dfa1.vortex.writer.VortexWriter;
 import io.github.dfa1.vortex.writer.WriteOptions;
@@ -245,6 +247,31 @@ public long fusedFilteredSumDict() {
         return acc;
     }
 
+    /// Fused dict-filtered multi-column aggregate: per chunk,
+    /// [Compute#filteredAggregate(Chunk, RowFilter, String)] evaluates `category == 7` as a whole
+    /// [RowFilter] and folds `measure`'s `SUM` / `MIN` / `MAX` / non-null count over the selected
+    /// rows — the kernel behind the Calcite boundary-chunk aggregate push-down, on the same
+    /// dict-filtered workload as [#fusedFilteredSumDict()]. Folds the selected count and sum into
+    /// the return value so JMH cannot eliminate the loop.
+    ///
+    /// Result (2026-07-02, 100M rows): 982.5 ± 30.2 ms/op through the per-leaf accessor path;
+    /// 178.3 ± 4.1 ms/op with the `DictFilter` lane — ≈ 5.5×. The gap to [#fusedFilteredSumDict()]'s
+    /// 25.5 ms is the fuller per-match fold (sum, min, max, two counters) whose heavier loop body
+    /// stays scalar; closing it is a possible follow-up specialization.
+    ///
+    /// @return the selected row count plus the sum of `measure` where `category == 7`, folded
+    ///         across the whole dataset
+    @Benchmark
+    public long fusedFilteredAggregateDict() {
+        long acc = 0;
+        for (Chunk chunk : chunks) {
+            FilteredAggregate aggregate = Compute.filteredAggregate(
+                    chunk, RowFilter.eq("category", CATEGORY_VALUE), "measure");
+            acc += aggregate.selectedRows() + aggregate.sum().longValue();
+        }
+        return acc;
+    }
+
     /// Hand-fused control for [#fusedFilteredSumAlp()]: the obvious developer loop a fused kernel must
     /// match — `for i: if (price.getDouble(i) > 500) acc += measure.getLong(i)` per chunk, with no
     /// off-heap bitmap. Decodes each price and (when selected) each measure through the accessor.