fix(reader): zone-map pruning compares in the column's type domain (#159)

ScanIterator's comparator caught ClassCastException and returned 0, which
canPruneChunk reads as "cannot prune". Stats decode integers as Long and floats
as Float/Double, so a filter value boxed at a different width (Integer for I64,
Float for F32) threw internally and silently disabled pruning — a valid,
selective predicate degraded to a full scan with no signal.

The comparison now keys off the *column* type, not the boxed operand:
- floating column      -> Double.compare;
- unsigned int column  -> Long.compareUnsigned (U64 stats/values store raw bits,
  so a value >= 2^63 is a negative Long; signed compare keeps/drops the wrong
  chunks). U8/U16/U32 zero-extend to a positive Long, unaffected;
- signed int column    -> Long.compare.

Keying off the column also avoids routing an integer column through
double-compare, which would lose precision past 2^53 and mis-prune.

Eq/Neq previously had their own inline comparator with the same swallow; they now
route through the shared one. A genuinely incomparable filter value (e.g. a
String against a numeric column) now raises VortexException instead of a silent
no-prune — a behaviour change, noted in the changelog.

Adds DType.isUnsigned() (exhaustive over the sealed set) to classify the column.

Coverage — ZoneMapPruningTest (27): BoxedWidth (Integer == Long, all six
operators), Unsigned (U64 >= 2^63 keep/prune correctness), FloatWidths (F32 stat
vs Double/Float filters), IntegerColumnFloatFilter (Double filter on I64 compares
in the integer domain past 2^53), TypeMismatch (String throws). Plus
DTypeIsUnsignedTest.

Closes #159.

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

Author: dfa1

CHANGELOG.md

@@ -5,6 +5,20 @@ All notable changes to **vortex-java** are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- `DType.isUnsigned()` — `true` for the unsigned integer primitives (`U8`–`U64`), `false` otherwise. ([#159](https://github.com/dfa1/vortex-java/issues/159))
+
+### Fixed
+
+- Zone-map pruning now compares filter values in the *column's* type domain rather than by the boxed value's type. A predicate whose value is boxed at a different width (e.g. `Integer` on an `I64` column) — or any value on a `U64` column — previously pruned nothing and silently degraded to a full scan; it now prunes correctly (unsigned columns by unsigned order). ([#159](https://github.com/dfa1/vortex-java/issues/159))
+
+### Changed
+
+- **Behaviour change:** a filter value that is genuinely incomparable to its column (e.g. a `String` against a numeric column) now raises `VortexException` during the scan instead of silently disabling pruning. Callers that relied on the previous silent full scan will see an exception. ([#159](https://github.com/dfa1/vortex-java/issues/159))
+
 ## [0.9.0] — 2026-06-24
 
 Two import-only breaking changes — the `vortex-core` types moved under `io.github.dfa1.vortex.core.*`, and the no-arg `DType` factories became constants. In return, Vortex now ships with **no FlatBuffers or Protobuf runtime dependency**: the `.fbs`/`.proto` schemas compile in-house to `MemorySegment`-native Java, dropping `com.google.flatbuffers:flatbuffers-java` — the last automatic-module dependency — so a named JPMS `module-info` is viable, and the generated wire classes are prefixed so they no longer collide on your classpath (ADR 0017).

core/src/main/java/io/github/dfa1/vortex/core/model/DType.java

@@ -26,6 +26,20 @@ public sealed interface DType
     /// @return `true` if null values are permitted
     boolean nullable();
 
+    /// Returns whether this is an unsigned integer type (`U8`–`U64`). `false` for every other
+    /// type, including signed integers, floats, and the composite/extension types. Useful where
+    /// unsigned values are stored in a signed `long` (e.g. zone-map comparisons), so the caller
+    /// knows to use unsigned ordering.
+    ///
+    /// @return `true` if this is an unsigned-integer [Primitive]
+    default boolean isUnsigned() {
+        return switch (this) {
+            case Primitive(var pt, _) -> pt.isUnsigned();
+            case Null _, Bool _, Decimal _, Utf8 _, Binary _, Struct _, List _,
+                 FixedSizeList _, Extension _, Variant _ -> false;
+        };
+    }
+
     /// Returns a copy of this type marked nullable. Sugar over
     /// [#withNullable(boolean)] so call sites read as a fluent adjective:
     /// `DType.I64.asNullable()`.

core/src/test/java/io/github/dfa1/vortex/core/model/DTypeIsUnsignedTest.java

@@ -0,0 +1,38 @@
+package io.github.dfa1.vortex.core.model;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class DTypeIsUnsignedTest {
+
+    @ParameterizedTest
+    @EnumSource(value = PType.class, names = {"U8", "U16", "U32", "U64"})
+    void unsignedPrimitives_areUnsigned(PType pt) {
+        // Given / When / Then
+        assertThat(new DType.Primitive(pt, false).isUnsigned()).isTrue();
+    }
+
+    @ParameterizedTest
+    @EnumSource(value = PType.class, names = {"I8", "I16", "I32", "I64", "F16", "F32", "F64"})
+    void signedAndFloatPrimitives_areNotUnsigned(PType pt) {
+        // Given / When / Then
+        assertThat(new DType.Primitive(pt, false).isUnsigned()).isFalse();
+    }
+
+    @Test
+    void nonPrimitiveTypes_areNotUnsigned() {
+        // Given — composite/extension types are never "unsigned", even one that wraps a U64 column
+        List<DType> types = List.of(
+                DType.BOOL, DType.UTF8, DType.BINARY, DType.NULL, DType.VARIANT,
+                new DType.Decimal((byte) 10, (byte) 2, false),
+                new DType.Struct(List.of("u"), List.of(DType.U64), false));
+
+        // When / Then
+        assertThat(types).allSatisfy(t -> assertThat(t.isUnsigned()).isFalse());
+    }
+}

reader/src/main/java/io/github/dfa1/vortex/reader/ScanIterator.java

@@ -43,6 +43,7 @@
 import java.lang.foreign.SegmentAllocator;
 import java.lang.foreign.ValueLayout;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -85,6 +86,7 @@ public final class ScanIterator implements Iterator<Chunk>, AutoCloseable {
     private List<ChunkSpec> chunks;
     private List<String> projectedNames;
     private List<DType> projectedDtypes;
+    private Map<String, DType> columnDtypes;
     private int chunkIndex;
     private int peekedChunkIdx = -1;
     private long rowsReturned;
@@ -182,14 +184,49 @@ private static ChunkSpec buildChunkSpec(String[] colNames, Map<String, List<Layo
     // ── Layout tree traversal ─────────────────────────────────────────────────
 
     @SuppressWarnings("unchecked")
-    private static int compareValues(Object a, Object b) {
+    private static int compareValues(Object a, Object b, DType column) {
+        // Key the compare mode off the *column* type, not the boxed operand type. Stats decode
+        // integers as Long and floats as Float/Double, and a caller may box a filter value at the
+        // column's natural width (Integer for I32) or in a different width entirely. Letting the
+        // column decide keeps pruning width-agnostic (issue #159) without ever routing an integer
+        // column through double-compare (which would lose precision past 2^53 and mis-prune).
+        if (a instanceof Number na && b instanceof Number nb && column instanceof DType.Primitive prim) {
+            if (prim.ptype().isFloating()) {
+                return Double.compare(na.doubleValue(), nb.doubleValue());
+            }
+            // U64 stats/values store the raw 64 bits, so a value >= 2^63 is a negative Long; an
+            // unsigned column must compare unsigned. U8/U16/U32 are zero-extended to a positive
+            // Long where signed == unsigned, so this stays correct for them too.
+            return column.isUnsigned()
+                    ? Long.compareUnsigned(na.longValue(), nb.longValue())
+                    : Long.compare(na.longValue(), nb.longValue());
+        }
         try {
             return ((Comparable<Object>) a).compareTo(b);
-        } catch (ClassCastException _) {
-            return 0;
+        } catch (ClassCastException e) {
+            // A genuinely incomparable filter value (e.g. a String against a numeric column) is a
+            // caller error — surface it instead of swallowing it into a silent no-prune.
+            throw new VortexException("filter value of type " + b.getClass().getSimpleName()
+                    + " is not comparable to the column's zone-map statistic of type "
+                    + a.getClass().getSimpleName(), e);
         }
     }
 
+    /// Returns the declared [DType] of column `col`, or `null` if the file is not a struct or has
+    /// no such column. Resolved once from the file's struct schema and cached; used to drive
+    /// zone-map comparisons by the column's true type rather than the filter value's boxing.
+    private DType columnDType(String col) {
+        if (columnDtypes == null) {
+            columnDtypes = new HashMap<>();
+            if (file.dtype() instanceof DType.Struct struct) {
+                for (int i = 0; i < struct.fieldNames().size(); i++) {
+                    columnDtypes.put(struct.fieldNames().get(i), struct.fieldTypes().get(i));
+                }
+            }
+        }
+        return columnDtypes.get(col);
+    }
+
     private static Map<String, Array> expandStruct(StructArray sa) {
         DType.Struct sd = (DType.Struct) sa.dtype();
         List<String> names = sd.fieldNames();
@@ -708,31 +745,31 @@ private boolean canPruneChunk(ChunkSpec chunk, RowFilter filter) {
                     yield false;
                 }
                 Object max = readFlatStats(flat).max();
-                yield max != null && compareValues(max, val) <= 0;
+                yield max != null && compareValues(max, val, columnDType(col)) <= 0;
             }
             case RowFilter.Gte(var col, var val) -> {
                 Layout flat = chunk.layoutFor(col);
                 if (flat == null) {
                     yield false;
                 }
                 Object max = readFlatStats(flat).max();
-                yield max != null && compareValues(max, val) < 0;
+                yield max != null && compareValues(max, val, columnDType(col)) < 0;
             }
             case RowFilter.Lt(var col, var val) -> {
                 Layout flat = chunk.layoutFor(col);
                 if (flat == null) {
                     yield false;
                 }
                 Object min = readFlatStats(flat).min();
-                yield min != null && compareValues(min, val) >= 0;
+                yield min != null && compareValues(min, val, columnDType(col)) >= 0;
             }
             case RowFilter.Lte(var col, var val) -> {
                 Layout flat = chunk.layoutFor(col);
                 if (flat == null) {
                     yield false;
                 }
                 Object min = readFlatStats(flat).min();
-                yield min != null && compareValues(min, val) > 0;
+                yield min != null && compareValues(min, val, columnDType(col)) > 0;
             }
             case RowFilter.Eq(var col, var val) -> {
                 Layout flat = chunk.layoutFor(col);
@@ -745,13 +782,10 @@ private boolean canPruneChunk(ChunkSpec chunk, RowFilter filter) {
                 if (min == null || max == null) {
                     yield false;
                 }
-                try {
-                    @SuppressWarnings("unchecked")
-                    Comparable<Object> cv = (Comparable<Object>) val;
-                    yield cv.compareTo(min) < 0 || cv.compareTo(max) > 0;
-                } catch (ClassCastException _) {
-                    yield false;
-                }
+                // val < min || val > max → no row in this chunk can equal val. Route through the
+                // shared comparator so this path is width-agnostic and unsigned-aware too (#159).
+                DType ct = columnDType(col);
+                yield compareValues(val, min, ct) < 0 || compareValues(val, max, ct) > 0;
             }
             case RowFilter.Neq(var col, var val) -> {
                 Layout flat = chunk.layoutFor(col);
@@ -764,13 +798,9 @@ private boolean canPruneChunk(ChunkSpec chunk, RowFilter filter) {
                 if (min == null || max == null) {
                     yield false;
                 }
-                try {
-                    @SuppressWarnings("unchecked")
-                    Comparable<Object> cv = (Comparable<Object>) val;
-                    yield cv.compareTo(min) == 0 && cv.compareTo(max) == 0;
-                } catch (ClassCastException _) {
-                    yield false;
-                }
+                // Every row equals val (min == max == val) → no row is != val.
+                DType ct = columnDType(col);
+                yield compareValues(val, min, ct) == 0 && compareValues(val, max, ct) == 0;
             }
             case RowFilter.IsNull(var col) -> {
                 Layout flat = chunk.layoutFor(col);