feat(compute): fused multi-column filteredAggregate; migrate boundary fold off Mask

Compute.filteredAggregate(Chunk, RowFilter, aggColumn) evaluates a whole
multi-column RowFilter (n-ary AND of column-bound predicates, three-valued)
and folds the aggregate column in ONE pass — returning selectedRows,
aggNonNullCount, sum, min and max with no intermediate selection bitmap.
Leaves reuse PrimitiveFilter's predicate lowering and fall back to the
generic kernel for composite/non-primitive predicates, so results are
identical to the deprecated filter->Mask->reduce path (proven by an
independent brute-force oracle test plus the unchanged boundary-fold suite).

VortexTable.foldBoundaryZone now calls it instead of buildMask + Mask.and +
per-leaf Compute.filter + masked reduce; buildMask/collectLeafMasks are
deleted and the boundary path no longer touches the deprecated Mask (now
production-dead, retained only behind its deprecation for now).

Also fixes a latent bug surfaced en route: Values.valueAt had no
Float16Array case, so SUM/MIN/MAX over an f16 column threw on BOTH the old
and fused paths; f16 now boxes via getFloat and sums as a Double, matching
the documented reduction semantics.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dfa1

calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java

@@ -1,6 +1,5 @@
 package io.github.dfa1.vortex.calcite;
 
-import io.github.dfa1.vortex.core.error.VortexException;
 import io.github.dfa1.vortex.core.model.DType;
 import io.github.dfa1.vortex.reader.ArrayStats;
 import io.github.dfa1.vortex.reader.Chunk;
@@ -10,10 +9,9 @@
 import io.github.dfa1.vortex.reader.VortexReader;
 import io.github.dfa1.vortex.reader.compute.Compare;
 import io.github.dfa1.vortex.reader.compute.Compute;
-import io.github.dfa1.vortex.reader.compute.Mask;
+import io.github.dfa1.vortex.reader.compute.FilteredAggregate;
 import io.github.dfa1.vortex.reader.compute.Predicate;
 import io.github.dfa1.vortex.reader.compute.ZoneReducer;
-import io.github.dfa1.vortex.reader.array.Array;
 import io.github.dfa1.vortex.reader.array.BoolArray;
 import io.github.dfa1.vortex.reader.array.ByteArray;
 import io.github.dfa1.vortex.reader.array.DoubleArray;
@@ -47,7 +45,6 @@
 
 import java.io.IOException;
 import java.io.UncheckedIOException;
-import java.lang.foreign.Arena;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;
@@ -191,10 +188,11 @@ public record FilteredFold(Number sum, Object min, Object max, Long nullCount, l
     /// Folds an aggregate over the rows a pushed `WHERE` `filter` selects, in two tiers (ADR 0013 §6).
     /// A zone the filter selects entirely (or excludes entirely) is folded from — or skipped via — its
     /// footer statistics with no decode; a boundary zone the filter only partially selects is decoded
-    /// for the aggregate and filter columns, a row-level selection [Mask] is built by evaluating the
-    /// filter through the compute kernels, and the aggregate is reduced over the masked rows. The two
-    /// tiers combine into one fold, so the caller answers the filtered aggregate while decoding only
-    /// the boundary zones.
+    /// for the aggregate and filter columns and reduced in one fused pass — the whole filter and the
+    /// aggregate fold evaluated together via
+    /// [Compute#filteredAggregate(Chunk, RowFilter, String)], with no intermediate selection bitmap.
+    /// The two tiers combine into one fold, so the caller answers the filtered aggregate while decoding
+    /// only the boundary zones.
     ///
     /// `aggColumn` is the column whose values are reduced over the selected rows (`null` for
     /// `COUNT(*)`, which needs only the selected row count). The result is empty — abandoning to a
@@ -317,79 +315,30 @@ public Optional<FilteredFold> filteredFold(RowFilter filter, String aggColumn) {
     }
 
     /// Decodes one boundary chunk and folds the aggregate over the rows the filter selects in it,
-    /// contributing to the running [Fold]. The chunk and a fresh confined [Arena] for the mask
-    /// bitmaps are released per zone via try-with-resources, so nothing outlives the reduction.
+    /// contributing to the running [Fold]. The chunk is released per zone via try-with-resources, so
+    /// nothing outlives the reduction.
     ///
-    /// `SUM` is guarded: the compute kernel throws on a non-numeric aggregate column, so a thrown
-    /// [VortexException] disables the sum (mirroring a zone that records no usable sum) rather than
-    /// failing the whole fold — a `MIN`/`MAX` over the same column still folds. The aggregate
-    /// column's null count among the selected rows is the selected count minus the non-null count.
-    @SuppressWarnings("removal") // transitional — uses the deprecated Mask until the fused multi-column filteredReduce lands
+    /// The whole filter — an n-ary `AND` of column-bound [Predicate] leaves — and the aggregate fold
+    /// run in one fused pass over the chunk's rows via
+    /// [Compute#filteredAggregate(Chunk, RowFilter, String)], with no intermediate selection bitmap.
+    /// `SUM` is guarded: the fused kernel reports a `null` sum for a non-numeric aggregate column,
+    /// which disables the sum (mirroring a zone that records no usable sum) rather than failing the
+    /// whole fold — a `MIN`/`MAX` over the same column still folds. The aggregate column's null count
+    /// among the selected rows is the selected count minus the non-null count.
     private static void foldBoundaryZone(VortexReader reader, int zone, List<String> columns,
                                          RowFilter filter, String aggColumn, Fold fold) {
-        try (Chunk chunk = reader.decodeChunk(zone, columns);
-             Arena maskArena = Arena.ofConfined()) {
-            Mask mask = buildMask(chunk, filter, maskArena);
-            long selected = mask.trueCount();
-            fold.keptRows += selected;
+        try (Chunk chunk = reader.decodeChunk(zone, columns)) {
+            FilteredAggregate aggregate = Compute.filteredAggregate(chunk, filter, aggColumn);
+            fold.keptRows += aggregate.selectedRows();
             if (aggColumn == null) {
                 return;
             }
-            Array aggArray = chunk.column(aggColumn);
-            if (fold.sumUsable) {
-                try {
-                    fold.addSum(Compute.sum(aggArray, mask));
-                } catch (VortexException e) {
-                    fold.sumUsable = false; // non-numeric agg column — sum unanswerable, like a missing zone sum
-                }
-            }
-            fold.addMin(Compute.min(aggArray, mask));
-            fold.addMax(Compute.max(aggArray, mask));
-            fold.addNulls(selected - Compute.count(aggArray, mask));
-        }
-    }
-
-    /// Builds the row-level selection [Mask] for a boundary chunk by evaluating `filter` through the
-    /// compute kernels: each comparison or null-test leaf becomes a [Predicate] filtered against its
-    /// column's decoded array, and an n-ary `AND` intersects the per-leaf masks (a row is selected
-    /// only when every leaf accepts it). The kernels apply SQL three-valued logic, so a null row
-    /// never satisfies a value predicate — matching the fold's null semantics.
-    @SuppressWarnings("removal") // transitional — uses the deprecated Mask until the fused multi-column filteredReduce lands
-    private static Mask buildMask(Chunk chunk, RowFilter filter, Arena arena) {
-        long n = chunk.rowCount();
-        List<Mask> masks = new ArrayList<>();
-        collectLeafMasks(chunk, filter, arena, masks);
-        if (masks.isEmpty()) {
-            return Mask.allTrue(n); // an AND with no leaves constrains nothing (never reached for a boundary)
-        }
-        if (masks.size() == 1) {
-            return masks.getFirst();
-        }
-        // Fold the per-leaf masks with the reader's Mask.and combinator (logical intersect): a row
-        // survives only when every leaf accepts it (the n-ary AND). The combinator allocates the
-        // result bitmap off-heap from this boundary zone's arena.
-        Mask result = masks.getFirst();
-        for (int k = 1; k < masks.size(); k++) {
-            result = result.and(masks.get(k), arena);
-        }
-        return result;
-    }
-
-    /// Walks the filter tree, evaluating each leaf into a per-column [Mask] and collecting them into
-    /// `out`; the conjuncts of an n-ary `AND` are flattened so [#buildMask] intersects them together.
-    /// A [RowFilter.Column] already holds the exact compute [Predicate], so the kernel runs it
-    /// directly against the bound column — no translation seam, since `>=`, `<=` and `<>` are now
-    /// first-class predicate variants.
-    @SuppressWarnings("removal") // transitional — uses the deprecated Mask and Compute.filter until the fused multi-column filteredReduce lands
-    private static void collectLeafMasks(Chunk chunk, RowFilter filter, Arena arena, List<Mask> out) {
-        switch (filter) {
-            case RowFilter.And(var parts) -> {
-                for (RowFilter part : parts) {
-                    collectLeafMasks(chunk, part, arena, out);
-                }
-            }
-            case RowFilter.Column(var col, var predicate) ->
-                    out.add(Compute.filter(chunk.column(col), predicate, arena));
+            // A null sum marks a non-numeric aggregate column — unanswerable, like a missing zone sum;
+            // addSum(null) disables only the sum, leaving MIN/MAX to fold.
+            fold.addSum(aggregate.sum());
+            fold.addMin(aggregate.min());
+            fold.addMax(aggregate.max());
+            fold.addNulls(aggregate.selectedRows() - aggregate.aggNonNullCount());
         }
     }
 
@@ -622,9 +571,9 @@ private static boolean isUnsigned(DType.Struct struct, String column) {
         return idx >= 0 && struct.fieldTypes().get(idx) instanceof DType.Primitive p && p.ptype().isUnsigned();
     }
 
-    /// Whether `column` is a floating-point primitive, whose compute-kernel compare ([#buildMask])
-    /// sorts NaN as the maximum and so cannot soundly build a boundary row mask (the fold abandons
-    /// such a filter column to the NaN-correct scan).
+    /// Whether `column` is a floating-point primitive, whose compute-kernel compare
+    /// ([Compare#values(Object, Object, DType)]) sorts NaN as the maximum and so cannot soundly fold a
+    /// boundary partition (the fold abandons such a filter column to the NaN-correct scan).
     private static boolean isFloating(DType.Struct struct, String column) {
         int idx = struct.fieldNames().indexOf(column);
         return idx >= 0 && struct.fieldTypes().get(idx) instanceof DType.Primitive p && p.ptype().isFloating();

reader/src/main/java/io/github/dfa1/vortex/reader/compute/Compute.java

@@ -1,5 +1,7 @@
 package io.github.dfa1.vortex.reader.compute;
 
+import io.github.dfa1.vortex.reader.Chunk;
+import io.github.dfa1.vortex.reader.RowFilter;
 import io.github.dfa1.vortex.reader.array.Array;
 
 import java.lang.foreign.Arena;
@@ -87,6 +89,32 @@ public static Number filteredSum(Array filterColumn, Predicate predicate, Array
         return FusedFilterSum.filteredSum(filterColumn, predicate, aggColumn);
     }
 
+    /// Evaluates `filter` over `chunk` and folds `aggColumn` over the rows it selects, in one fused
+    /// pass — the multi-column counterpart of [#filteredSum(Array, Predicate, Array)] and the
+    /// replacement for building per-leaf [Mask]s, intersecting them, and reducing under the result.
+    ///
+    /// The `filter` is the whole chunk predicate: an n-ary `AND` of column-bound [Predicate] leaves
+    /// (or a single leaf). A row is selected only when every leaf accepts it under SQL three-valued
+    /// logic — a null in any filter column rejects that leaf. Over the selected rows whose aggregate
+    /// value is non-null, the kernel folds the aggregate column's `SUM`, `MIN`, `MAX` and non-null
+    /// count in the same scan, with no intermediate bitmap.
+    ///
+    /// `aggColumn` may be `null` for a `COUNT(*)`-style fold that only counts selected rows; the
+    /// returned [FilteredAggregate]'s aggregate fields are then empty. `SUM` is `null` for a
+    /// non-numeric aggregate column (an unanswerable sum); `MIN` / `MAX` are `null` when no selected
+    /// row is non-null. The aggregate's null count among the selected rows is
+    /// `selectedRows − aggNonNullCount`.
+    ///
+    /// @param chunk     the decoded chunk holding the filter and aggregate columns
+    /// @param filter    the whole chunk predicate to evaluate
+    /// @param aggColumn the column to reduce over the selected rows, or `null` to count rows only
+    /// @return the fold over the rows `filter` selects
+    public static FilteredAggregate filteredAggregate(Chunk chunk, RowFilter filter, String aggColumn) {
+        Objects.requireNonNull(chunk, "chunk");
+        Objects.requireNonNull(filter, "filter");
+        return FusedFilterAggregate.aggregate(chunk, filter, aggColumn);
+    }
+
     /// Counts the selected non-null values of `array`.
     ///
     /// Pairs with the (deprecated) [Mask]; transitional pending the fused reductions such as

reader/src/main/java/io/github/dfa1/vortex/reader/compute/FilteredAggregate.java

@@ -0,0 +1,25 @@
+package io.github.dfa1.vortex.reader.compute;
+
+/// The result of a fused multi-column filtered aggregation over one chunk, returned by
+/// [Compute#filteredAggregate(io.github.dfa1.vortex.reader.Chunk, io.github.dfa1.vortex.reader.RowFilter, String)].
+///
+/// The fold covers exactly the rows the whole [io.github.dfa1.vortex.reader.RowFilter] selects. The
+/// aggregate fields are populated only when an aggregate column was given; with no aggregate column
+/// (a `COUNT(*)`) only [#selectedRows()] is meaningful and the rest are zero / `null`.
+///
+/// Null semantics mirror the reductions: `SUM` over zero selected non-null rows is the additive
+/// identity (`0`), while `sum` is `null` only when the aggregate column is non-numeric (an
+/// unanswerable sum); `MIN` / `MAX` are `null` when no selected row is non-null. The aggregate
+/// column's null count among the selected rows is `selectedRows − aggNonNullCount`.
+///
+/// @param selectedRows    the number of rows the whole filter selected
+/// @param aggNonNullCount the number of selected rows whose aggregate value is non-null (`0` when
+///                        there is no aggregate column)
+/// @param sum             the `SUM` over the selected non-null rows ([Long] for an integer aggregate,
+///                        [Double] for a floating one), `null` when there is no aggregate column or it
+///                        is non-numeric
+/// @param min             the `MIN` over the selected non-null rows under the column's dtype-aware
+///                        order, or `null` when none contributed
+/// @param max             the `MAX` over the selected non-null rows, or `null` when none contributed
+public record FilteredAggregate(long selectedRows, long aggNonNullCount, Number sum, Object min, Object max) {
+}