refactor(compute): unify RowFilter over Predicate (ADR 0013 §5)

RowFilter and the compute Predicate were two parallel vocabularies bridged
by a translation seam in the Calcite adapter (toPredicate, with >=/<=/<>
lowered to Or compositions). They are now one: Predicate is the shared,
public, column-agnostic value-test vocabulary (gaining first-class Neq/Gte/
Lte), and RowFilter is a column bound to a Predicate — RowFilter.Column(
column, predicate) plus the n-ary And. The public factories (eq/neq/gt/gte/
lt/lte/isNull/isNotNull/and) keep their signatures.

So push-down is now literally "the same predicate, compiled two ways":
ScanIterator.canPruneChunk evaluates the Predicate against a column's
zone-map stats (a recursive, strictly-conservative tester — it skips a
chunk only when provably no row can match, declining to prune on any
unknown stat or uncertain case), and the boundary fold runs the very same
Predicate through Compute.filter on the decoded chunk. The Calcite
toPredicate seam and the Or-lowering are deleted; the kernels evaluate
Neq/Gte/Lte natively (the primitive fast lane folds them via a hoisted,
loop-invariant negate/range, no hot-loop regression).

Neq pruning now additionally requires a provably null-free chunk before
skipping a min==max==v zone — stricter than before, conservative-safe.

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

Author: dfa1

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

@@ -375,63 +375,19 @@ private static Mask buildMask(Chunk chunk, RowFilter filter, Arena arena) {
 
     /// 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.
     private static void collectLeafMasks(Chunk chunk, RowFilter filter, Arena arena, List<Mask> out) {
-        if (filter instanceof RowFilter.And(var parts)) {
-            for (RowFilter part : parts) {
-                collectLeafMasks(chunk, part, arena, out);
+        switch (filter) {
+            case RowFilter.And(var parts) -> {
+                for (RowFilter part : parts) {
+                    collectLeafMasks(chunk, part, arena, out);
+                }
             }
-            return;
+            case RowFilter.Column(var col, var predicate) ->
+                    out.add(Compute.filter(chunk.column(col), predicate, arena));
         }
-        out.add(Compute.filter(chunk.column(leafColumn(filter)), toPredicate(filter), arena));
-    }
-
-    /// The column a leaf [RowFilter] references (never called for an `AND`, which has no single
-    /// column).
-    private static String leafColumn(RowFilter leaf) {
-        return switch (leaf) {
-            case RowFilter.Eq(var col, var ignored) -> col;
-            case RowFilter.Neq(var col, var ignored) -> col;
-            case RowFilter.Gt(var col, var ignored) -> col;
-            case RowFilter.Gte(var col, var ignored) -> col;
-            case RowFilter.Lt(var col, var ignored) -> col;
-            case RowFilter.Lte(var col, var ignored) -> col;
-            case RowFilter.IsNull(var col) -> col;
-            case RowFilter.IsNotNull(var col) -> col;
-            case RowFilter.And ignored ->
-                    throw new IllegalArgumentException("leafColumn called on a composite AND filter");
-        };
-    }
-
-    /// Translates a leaf [RowFilter] into the equivalent compute [Predicate]. The compute vocabulary
-    /// has no direct `>=`, `<=`, or `<>`, so each is composed from the ops it does have — all
-    /// null-correct because every component excludes nulls:
-    /// - `>= v` is `= v OR > v`;
-    /// - `<= v` is `= v OR < v`;
-    /// - `<> v` is `< v OR > v`.
-    /// This keeps a `BETWEEN` (which Calcite lowers to `>= a AND <= b`) foldable on a boundary zone
-    /// rather than abandoning it.
-    private static Predicate toPredicate(RowFilter leaf) {
-        return switch (leaf) {
-            case RowFilter.Eq(var ignored, var value) -> new Predicate.Eq(value);
-            case RowFilter.Neq(var ignored, var value) ->
-                    new Predicate.Or(new Predicate.Lt(asComparable(value)), new Predicate.Gt(asComparable(value)));
-            case RowFilter.Gt(var ignored, var value) -> new Predicate.Gt(value);
-            case RowFilter.Gte(var ignored, var value) ->
-                    new Predicate.Or(new Predicate.Eq(value), new Predicate.Gt(value));
-            case RowFilter.Lt(var ignored, var value) -> new Predicate.Lt(value);
-            case RowFilter.Lte(var ignored, var value) ->
-                    new Predicate.Or(new Predicate.Eq(value), new Predicate.Lt(value));
-            case RowFilter.IsNull(var ignored) -> new Predicate.IsNull();
-            case RowFilter.IsNotNull(var ignored) -> new Predicate.IsNotNull();
-            case RowFilter.And ignored ->
-                    throw new IllegalArgumentException("toPredicate called on a composite AND filter");
-        };
-    }
-
-    /// Casts a leaf's value to [Comparable] for the ordering predicates (`< v` / `> v`). The values
-    /// the filter translation produces are always [Long] / [Double] / [String], all comparable.
-    private static Comparable<?> asComparable(Object value) {
-        return (Comparable<?>) value;
     }
 
     /// Mutable accumulator for a filtered fold, shared by the whole-zone (tier 1) and boundary-zone
@@ -558,8 +514,28 @@ private static Match classify(RowFilter filter, int zone,
                 }
                 yield allIn ? Match.IN : Match.BOUNDARY;
             }
-            case RowFilter.Eq(var col, var value) -> {
-                ArrayStats s = zoneStats.get(col).get(zone);
+            case RowFilter.Column(var col, var predicate) ->
+                    classifyColumn(predicate, zoneStats.get(col).get(zone), rowCount);
+        };
+    }
+
+    /// Classifies one zone against a column-bound [Predicate] from the zone's statistics `s`. The
+    /// comparison ops carry the same three-valued-logic semantics as before the [RowFilter] /
+    /// [Predicate] unification: an unrecognised stat shape or a partially-overlapping zone is
+    /// [Match#BOUNDARY], a zone provably outside the predicate is [Match#OUT], and a zone every row
+    /// of which matches (which, for a value comparison, also requires the zone to carry no nulls) is
+    /// [Match#IN]. The composite and range predicates ([Predicate.Between] / [Predicate.And] /
+    /// [Predicate.Or]) cannot arise from the Calcite translation — `SEARCH` / `BETWEEN` expand to an
+    /// `AND` of single-leaf columns — so they fall through to [Match#BOUNDARY], which decodes and
+    /// applies the kernel mask and is therefore always correct.
+    ///
+    /// @param predicate the column-bound value-test
+    /// @param s         the zone's statistics for that column
+    /// @param rowCount  the zone's row count
+    /// @return how the zone relates to the predicate
+    private static Match classifyColumn(Predicate predicate, ArrayStats s, long rowCount) {
+        return switch (predicate) {
+            case Predicate.Eq(var value) -> {
                 if (uncomparable(s, value)) {
                     yield Match.BOUNDARY;
                 }
@@ -571,8 +547,7 @@ private static Match classify(RowFilter filter, int zone,
                 }
                 yield Match.BOUNDARY;
             }
-            case RowFilter.Neq(var col, var value) -> {
-                ArrayStats s = zoneStats.get(col).get(zone);
+            case Predicate.Neq(var value) -> {
                 if (uncomparable(s, value)) {
                     yield Match.BOUNDARY;
                 }
@@ -584,16 +559,16 @@ private static Match classify(RowFilter filter, int zone,
                 }
                 yield Match.BOUNDARY;
             }
-            case RowFilter.Gt(var col, var value) -> compare(zoneStats.get(col).get(zone), value,
+            case Predicate.Gt(var value) -> compare(s, value,
                     (st, v) -> compareStat(st.max(), v) <= 0, (st, v) -> compareStat(st.min(), v) > 0);
-            case RowFilter.Gte(var col, var value) -> compare(zoneStats.get(col).get(zone), value,
+            case Predicate.Gte(var value) -> compare(s, value,
                     (st, v) -> compareStat(st.max(), v) < 0, (st, v) -> compareStat(st.min(), v) >= 0);
-            case RowFilter.Lt(var col, var value) -> compare(zoneStats.get(col).get(zone), value,
+            case Predicate.Lt(var value) -> compare(s, value,
                     (st, v) -> compareStat(st.min(), v) >= 0, (st, v) -> compareStat(st.max(), v) < 0);
-            case RowFilter.Lte(var col, var value) -> compare(zoneStats.get(col).get(zone), value,
+            case Predicate.Lte(var value) -> compare(s, value,
                     (st, v) -> compareStat(st.min(), v) > 0, (st, v) -> compareStat(st.max(), v) <= 0);
-            case RowFilter.IsNull(var col) -> {
-                Long nulls = zoneStats.get(col).get(zone).nullCount();
+            case Predicate.IsNull ignored -> {
+                Long nulls = s.nullCount();
                 if (nulls == null) {
                     yield Match.BOUNDARY;
                 }
@@ -602,8 +577,8 @@ private static Match classify(RowFilter filter, int zone,
                 }
                 yield nulls == 0 ? Match.OUT : Match.BOUNDARY;
             }
-            case RowFilter.IsNotNull(var col) -> {
-                Long nulls = zoneStats.get(col).get(zone).nullCount();
+            case Predicate.IsNotNull ignored -> {
+                Long nulls = s.nullCount();
                 if (nulls == null) {
                     yield Match.BOUNDARY;
                 }
@@ -612,6 +587,9 @@ private static Match classify(RowFilter filter, int zone,
                 }
                 yield nulls == rowCount ? Match.OUT : Match.BOUNDARY;
             }
+            case Predicate.Between ignored -> Match.BOUNDARY;
+            case Predicate.And ignored -> Match.BOUNDARY;
+            case Predicate.Or ignored -> Match.BOUNDARY;
         };
     }
 
@@ -917,14 +895,7 @@ private static Optional<RowFilter> toRowFilter(List<RexNode> filters, List<Strin
     private static void collectColumns(RowFilter filter, java.util.Set<String> out) {
         switch (filter) {
             case RowFilter.And(var parts) -> parts.forEach(f -> collectColumns(f, out));
-            case RowFilter.Eq(var col, var ignored) -> out.add(col);
-            case RowFilter.Neq(var col, var ignored) -> out.add(col);
-            case RowFilter.Gt(var col, var ignored) -> out.add(col);
-            case RowFilter.Gte(var col, var ignored) -> out.add(col);
-            case RowFilter.Lt(var col, var ignored) -> out.add(col);
-            case RowFilter.Lte(var col, var ignored) -> out.add(col);
-            case RowFilter.IsNull(var col) -> out.add(col);
-            case RowFilter.IsNotNull(var col) -> out.add(col);
+            case RowFilter.Column(var col, var ignored) -> out.add(col);
         }
     }
 

calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterPushDownTest.java

@@ -64,12 +64,12 @@ static void write() throws Exception {
     @ParameterizedTest(name = "[{index}] WHERE {0} -> {1} rows")
     @CsvSource({
             // every comparison kind the RowFilter translator supports, over a Long column
-            "i64 = 1000,            1",   // EQUALS        -> RowFilter.Eq
-            "i64 <> 1000,           5",   // NOT_EQUALS    -> RowFilter.Neq
-            "i64 < 3000,            2",   // LESS_THAN     -> RowFilter.Lt
-            "i64 <= 3000,           3",   // LESS_THAN_EQ  -> RowFilter.Lte
-            "i64 > 4000,            2",   // GREATER_THAN  -> RowFilter.Gt
-            "i64 >= 4000,           3",   // GREATER_EQ    -> RowFilter.Gte
+            "i64 = 1000,            1",   // EQUALS        -> Column(i64, Predicate.Eq)
+            "i64 <> 1000,           5",   // NOT_EQUALS    -> Column(i64, Predicate.Neq)
+            "i64 < 3000,            2",   // LESS_THAN     -> Column(i64, Predicate.Lt)
+            "i64 <= 3000,           3",   // LESS_THAN_EQ  -> Column(i64, Predicate.Lte)
+            "i64 > 4000,            2",   // GREATER_THAN  -> Column(i64, Predicate.Gt)
+            "i64 >= 4000,           3",   // GREATER_EQ    -> Column(i64, Predicate.Gte)
             "s = 'a',               1",   // Utf8 literal coercion
             "f64 > 3.0,             3",   // floating literal coercion
             // multiple conjuncts arrive as separate filters -> RowFilter.and over the list

cli/src/main/java/io/github/dfa1/vortex/cli/FilterCommand.java

@@ -15,6 +15,7 @@
 import io.github.dfa1.vortex.csv.RowPredicate;
 import io.github.dfa1.vortex.reader.RowFilter;
 import io.github.dfa1.vortex.reader.ScanOptions;
+import io.github.dfa1.vortex.reader.compute.Predicate;
 
 import java.io.IOException;
 import java.io.OutputStreamWriter;
@@ -150,18 +151,7 @@ private static Comparable<?> parseValue(String raw) {
 
     private static RowPredicate toRowPredicate(RowFilter filter) {
         return switch (filter) {
-            case RowFilter.Gt(var col, var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) > 0;
-            case RowFilter.Gte(var col, var val) ->
-                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) >= 0;
-            case RowFilter.Lt(var col, var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) < 0;
-            case RowFilter.Lte(var col, var val) ->
-                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) <= 0;
-            case RowFilter.Eq(var col, var val) ->
-                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, (Comparable<?>) val) == 0;
-            case RowFilter.Neq(var col, var val) ->
-                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, (Comparable<?>) val) != 0;
-            case RowFilter.IsNull(var col) -> (chunk, rowIdx) -> isRowNull(chunk.column(col), rowIdx);
-            case RowFilter.IsNotNull(var col) -> (chunk, rowIdx) -> !isRowNull(chunk.column(col), rowIdx);
+            case RowFilter.Column(var col, var predicate) -> columnPredicate(col, predicate);
             case RowFilter.And(var filters) -> {
                 RowPredicate[] preds = filters.stream().map(FilterCommand::toRowPredicate).toArray(RowPredicate[]::new);
                 yield (chunk, rowIdx) -> {
@@ -176,6 +166,30 @@ private static RowPredicate toRowPredicate(RowFilter filter) {
         };
     }
 
+    /// Compiles a column-bound [Predicate] into a per-row test over the decoded chunk. Only the
+    /// comparison and null-test leaves the filter grammar ([#parseFilter]) produces are supported;
+    /// the composite and range predicate variants never arise from a parsed CLI expression.
+    private static RowPredicate columnPredicate(String col, Predicate predicate) {
+        return switch (predicate) {
+            case Predicate.Gt(var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) > 0;
+            case Predicate.Gte(var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) >= 0;
+            case Predicate.Lt(var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) < 0;
+            case Predicate.Lte(var val) -> (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, val) <= 0;
+            case Predicate.Eq(var val) ->
+                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, (Comparable<?>) val) == 0;
+            case Predicate.Neq(var val) ->
+                    (chunk, rowIdx) -> compareValue(chunk.column(col), rowIdx, (Comparable<?>) val) != 0;
+            case Predicate.IsNull ignored -> (chunk, rowIdx) -> isRowNull(chunk.column(col), rowIdx);
+            case Predicate.IsNotNull ignored -> (chunk, rowIdx) -> !isRowNull(chunk.column(col), rowIdx);
+            case Predicate.Between ignored ->
+                    throw new IllegalArgumentException("BETWEEN filters are not supported on the command line");
+            case Predicate.And ignored ->
+                    throw new IllegalArgumentException("nested predicate AND is not supported on the command line");
+            case Predicate.Or ignored ->
+                    throw new IllegalArgumentException("nested predicate OR is not supported on the command line");
+        };
+    }
+
     // Only a masked column carries nulls; an unmasked array is null-free, so every row is non-null.
     private static boolean isRowNull(Array arr, long rowIdx) {
         return arr instanceof MaskedArray masked && !masked.isValid(rowIdx);