perf(reader): decode and filter dictionary strings once per value, not per row

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - `Chunk.slice(offset, length)` gives a zero-copy view of a row range. ([#550](https://github.com/dfa1/vortex-java/pull/550))
 - The Calcite adapter tells the planner each table's row count, so join order and costing use the real table size. ([#550](https://github.com/dfa1/vortex-java/pull/550))
+- `Compute.matcher(Chunk, RowFilter)` lowers a filter once per chunk and returns a row test, for callers that test many rows. ([#552](https://github.com/dfa1/vortex-java/issues/552))
 - `vortex.json` and the `vortex.st.*` geometry columns open without a decoder and read as their storage, as vortex-jni reads them. ([#547](https://github.com/dfa1/vortex-java/pull/547))
 - The CLI takes `--allow-unknown` to open files with extension types it has no decoder for, and CSV `ExportOptions.withReadRegistry` picks the registry an export opens the file with. ([#547](https://github.com/dfa1/vortex-java/pull/547))
 - Custom extension types: register an `ExtensionDecoder` on `ReadRegistry` and an `ExtensionEncoder` on `WriteRegistry`, with a `validate` hook on both sides, and `Chunk.as` decodes any registered type. ([#530](https://github.com/dfa1/vortex-java/issues/530))
@@ -18,6 +19,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - The Calcite adapter pushes `IN` lists, comparisons with the literal on the left and `IS NULL` into the scan, so they prune chunks instead of being re-checked row by row. ([#550](https://github.com/dfa1/vortex-java/pull/550))
 - The Calcite adapter answers a filtered `COUNT`, `MIN` or `MAX` from Rust's `vortex.zoned` zone maps, the default, decoding only the zones the filter cuts. ([#550](https://github.com/dfa1/vortex-java/pull/550))
 - The Calcite adapter reads a table's schema and row count once instead of reopening the file for every planning call. ([#550](https://github.com/dfa1/vortex-java/pull/550))
+- Reading strings from a dictionary-encoded column is about 4× faster, and SQL over it about 30% faster: each distinct value is decoded once, not once per row. ([#551](https://github.com/dfa1/vortex-java/issues/551))
+- SQL filters on a dictionary-encoded string column are about twice as fast: the filter is evaluated once per dictionary value, as in Rust, not once per row. ([#552](https://github.com/dfa1/vortex-java/issues/552))
+- **Breaking:** `VarBinDictArray` and `VarBinConstantArray` are final classes instead of records, with the same accessors. ([#551](https://github.com/dfa1/vortex-java/issues/551))
 - **Breaking:** opening a file with an extension id that has no registered decoder fails unless `ReadRegistry.Builder#allowUnknown()` is set, as in Rust; the column then reads as its storage array. ([#530](https://github.com/dfa1/vortex-java/issues/530))
 - **Breaking:** `vortex.date`, `vortex.time`, `vortex.timestamp` and `vortex.uuid` dtypes are checked as Rust checks them (time unit, storage type, metadata) when writing and reading, and a file whose extension dtype has no metadata field is refused, as Rust refuses it. ([#547](https://github.com/dfa1/vortex-java/pull/547))
 - **Breaking:** `DateExtensionDecoder.decode` and `decodeAll` take the column's `DType.Extension`, which carries the time unit. ([#547](https://github.com/dfa1/vortex-java/pull/547))

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

@@ -34,6 +34,7 @@
 import java.io.UncheckedIOException;
 import java.nio.file.Path;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongPredicate;
 
 /// Streaming [Enumerator] over a Vortex scan: advances chunk by chunk, decoding each requested
 /// column once per chunk and materializing one `Object[]` row per [#moveNext()]. Rows are not
@@ -48,6 +49,7 @@ final class VortexEnumerator implements Enumerator<Object[]> {
     private final VortexReader reader;
     private final ScanIterator scan;
     private Chunk chunk;
+    private LongPredicate rowMatcher;
     private Object[] columns;
     private long rowInChunk;
     private long chunkRows;
@@ -61,7 +63,7 @@ final class VortexEnumerator implements Enumerator<Object[]> {
     /// @param types                  output column dtypes, parallel to `names`
     /// @param exactFilter            a predicate [VortexTable#scan(DataContext, java.util.List, int[])]
     ///                               removed from Calcite's own re-check list, enforced here row-by-row
-    ///                               via [Compute#matches(Chunk, RowFilter, long)] so the result stays
+    ///                               via [Compute#matcher(Chunk, RowFilter)] so the result stays
     ///                               exact — or `null` when every pushed predicate is still Calcite's to
     ///                               re-check
     VortexEnumerator(Path file, AtomicLong chunksScannedLastQuery, ScanOptions options,
@@ -101,7 +103,7 @@ public Object[] current() {
     public boolean moveNext() {
         while (true) {
             if (chunk != null && rowInChunk < chunkRows) {
-                if (exactFilter != null && !Compute.matches(chunk, exactFilter, rowInChunk)) {
+                if (exactFilter != null && !rowMatcher.test(rowInChunk)) {
                     rowInChunk++;
                     continue;
                 }
@@ -124,6 +126,7 @@ public boolean moveNext() {
             chunksScannedLastQuery.incrementAndGet();
             chunkRows = chunk.rowCount();
             rowInChunk = 0;
+            rowMatcher = exactFilter == null ? null : Compute.matcher(chunk, exactFilter);
             columns = new Object[names.length];
             for (int c = 0; c < names.length; c++) {
                 columns[c] = chunk.column(names[c]);

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

@@ -831,8 +831,8 @@ public Enumerable<Object[]> scan(DataContext root, List<RexNode> filters, int[]
         Optional<RowFilter> pushed = RexFilterTranslator.toRowFilter(filters, allNames, allTypes);
 
         // Exact push-down: a top-level predicate captured *in full* is enforced row-by-row below
-        // (via Compute#matches, which routes through the NaN-correct PredicateEvaluator /
-        // PrimitiveFilter) and removed from Calcite's own re-check list, so Calcite stops
+        // (via Compute#matcher, which lowers it per chunk through the NaN-correct PrimitiveFilter /
+        // PredicateEvaluator) and removed from Calcite's own re-check list, so Calcite stops
         // redundantly re-evaluating rows this scan has already excluded — Calcite's
         // TableScanNode#createProjectableFilterable wraps a `.where()` residual check around
         // whatever is still left in `filters` after this call returns.

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

@@ -312,7 +312,7 @@ void nanRowSortsAboveEveryValueInAFloatingWhereFilterEvenAfterAbandoningToScan()
         // Given a NaN price alongside ordinary values that would all otherwise satisfy the filter
         // below. A mix like this makes every zone at best a BOUNDARY under the floating column's
         // classify (never IN), so the fold abandons to a full scan — this test proves that scan
-        // path (VortexEnumerator -> Compute#matches -> PredicateEvaluator) orders NaN as the maximum
+        // path (VortexEnumerator -> Compute#matcher -> PrimitiveFilter) orders NaN as the maximum
         // (Rust's total order, #518) too, not just the fold's own classify.
         DType.Struct schema = new DType.Struct(
                 List.of(ColumnName.of("price"), ColumnName.of("vol")),

integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java

@@ -39,7 +39,9 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
 import java.io.IOException;
@@ -54,6 +56,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -311,6 +314,185 @@ void jniWriter_filteredUtf8Scan_comparesInUtf8ByteOrder(@TempDir Path tmp) throw
         assertThat(result).isEqualTo(1);
     }
 
+    @Test
+    void jniWriter_lowCardinalityUtf8_cachedDictStringsMatchRows(@TempDir Path tmp) throws IOException {
+        // Given — Rust dict-encodes a low-cardinality string column; Java caches one decoded String
+        // per dictionary entry (#551). Every row must still read its own value, through a full scan
+        // and through a filtered scan (whose predicate reads the same cached strings). The non-ASCII
+        // value checks the cached entry is the UTF-8 decode, not a byte-per-char shortcut.
+        String[] symbols = {"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "café"};
+        int n = 100_000;
+        String[] values = new String[n];
+        for (int i = 0; i < n; i++) {
+            values[i] = symbols[(i * 7 + i / 3) % symbols.length];
+        }
+        Path file = tmp.resolve("jni_utf8_lowcard.vtx");
+        writeJniUtf8(file, values);
+        long expectedMatches = Arrays.stream(values).filter("café"::equals).count();
+
+        // When
+        var all = new ArrayList<String>();
+        var filtered = new ArrayList<String>();
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) {
+            try (var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all())) {
+                iter.forEachRemaining(c -> {
+                    VarBinArray k = c.column(ColumnName.of("k"));
+                    assertThat(k).isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinDictArray.class);
+                    for (long i = 0; i < k.length(); i++) {
+                        all.add(k.getString(i));
+                    }
+                });
+            }
+            // The scan filter only prunes zones; rows are matched per row, as the Calcite adapter does
+            var filter = io.github.dfa1.vortex.reader.RowFilter.eq(ColumnName.of("k"), "café");
+            try (var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all().withFilter(filter))) {
+                iter.forEachRemaining(c -> {
+                    VarBinArray k = c.column(ColumnName.of("k"));
+                    for (long i = 0; i < k.length(); i++) {
+                        if (io.github.dfa1.vortex.reader.compute.Compute.matches(c, filter, i)) {
+                            filtered.add(k.getString(i));
+                        }
+                    }
+                });
+            }
+        }
+
+        // Then
+        assertThat(all).containsExactly(values);
+        assertThat(filtered).hasSize((int) expectedMatches).containsOnly("café");
+    }
+
+    @Test
+    void jniWriter_singleValueUtf8_cachedConstantStringMatchesRows(@TempDir Path tmp) throws IOException {
+        // Given — one distinct value: Rust writes it as a constant, which Java reads as a
+        // VarBinConstantArray that decodes the value once and shares it across rows (#551)
+        int n = 10_000;
+        String[] values = new String[n];
+        Arrays.fill(values, "café");
+        Path file = tmp.resolve("jni_utf8_single.vtx");
+        writeJniUtf8(file, values);
+
+        // When
+        var result = new ArrayList<String>();
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll());
+             var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all())) {
+            iter.forEachRemaining(c -> {
+                VarBinArray k = c.column(ColumnName.of("k"));
+                assertThat(k).isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinConstantArray.class);
+                for (long i = 0; i < k.length(); i++) {
+                    result.add(k.getString(i));
+                }
+            });
+        }
+
+        // Then
+        assertThat(result).containsExactly(values);
+    }
+
+    static Stream<Arguments> dictStringFilters() {
+        List<Arguments> cases = new ArrayList<>();
+        for (String op : List.of("EQ", "NOT_EQ", "GT", "GTE", "LT", "LTE")) {
+            for (String bound : new String[]{"café", "", "ZZZ"}) {
+                cases.add(Arguments.of(op, bound));
+            }
+        }
+        return cases.stream();
+    }
+
+    @ParameterizedTest(name = "{0} {1}")
+    @MethodSource("dictStringFilters")
+    void jniWriter_dictUtf8Filter_selectsTheRowsJniSelects(String op, String bound, @TempDir Path tmp)
+            throws IOException {
+        // Given — Rust dict-encodes a nullable low-cardinality string column. Java now evaluates a
+        // filter once per dictionary entry and tests rows by code (#552), as Rust's dict compare does.
+        // The emoji and U+E000 sort one way in UTF-8 (Rust's order) and the other in UTF-16, and
+        // null rows must never match, so Java must keep exactly the rows vortex-jni keeps.
+        String[] symbols = {"AAPL", "MSFT", "café", "😀", "", null};
+        String[] values = new String[20_000];
+        for (int i = 0; i < values.length; i++) {
+            values[i] = symbols[(i * 7 + i / 3) % symbols.length];
+        }
+        Path file = tmp.resolve("jni_dict_utf8_filter.vtx");
+        writeJniNullableUtf8(file, values);
+        ColumnName column = ColumnName.of("k");
+        io.github.dfa1.vortex.reader.RowFilter filter = switch (op) {
+            case "EQ" -> io.github.dfa1.vortex.reader.RowFilter.eq(column, bound);
+            case "NOT_EQ" -> io.github.dfa1.vortex.reader.RowFilter.neq(column, bound);
+            case "GT" -> io.github.dfa1.vortex.reader.RowFilter.gt(column, bound);
+            case "GTE" -> io.github.dfa1.vortex.reader.RowFilter.gte(column, bound);
+            case "LT" -> io.github.dfa1.vortex.reader.RowFilter.lt(column, bound);
+            default -> io.github.dfa1.vortex.reader.RowFilter.lte(column, bound);
+        };
+
+        // When
+        List<String> jni = readJniUtf8Filtered(file, Expression.binary(Expression.BinaryOp.valueOf(op),
+                Expression.column("k"), Expression.literal(bound)));
+        List<String> result = new ArrayList<>();
+        try (var vf = VortexReader.open(file, ReadRegistry.loadAll());
+             var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all().withFilter(filter))) {
+            iter.forEachRemaining(c -> {
+                Array k = c.column(column);
+                Array data = MaskedArray.unwrap(k).inner();
+                assertThat(data instanceof io.github.dfa1.vortex.reader.array.VarBinSlicedArray s ? s.inner() : data)
+                        .isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinDictArray.class);
+                var matcher = io.github.dfa1.vortex.reader.compute.Compute.matcher(c, filter);
+                for (long i = 0; i < k.length(); i++) {
+                    if (matcher.test(i)) {
+                        result.add(((VarBinArray) data).getString(i));
+                    }
+                }
+            });
+        }
+
+        // Then
+        assertThat(result).containsExactlyInAnyOrderElementsOf(jni);
+    }
+
+    private static void writeJniNullableUtf8(Path file, String[] values) throws IOException {
+        Schema schema = new Schema(List.of(Field.nullable("k", new ArrowType.Utf8())));
+        String uri = file.toAbsolutePath().toUri().toString();
+        try (VortexWriter writer = VortexWriter.builder(SESSION, uri, schema, ALLOCATOR).build();
+             VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) {
+            VarCharVector vec = (VarCharVector) root.getVector("k");
+            vec.allocateNew(values.length);
+            for (int i = 0; i < values.length; i++) {
+                if (values[i] == null) {
+                    vec.setNull(i);
+                } else {
+                    vec.setSafe(i, values[i].getBytes(java.nio.charset.StandardCharsets.UTF_8));
+                }
+            }
+            root.setRowCount(values.length);
+            try (ArrowArray arr = ArrowArray.allocateNew(ALLOCATOR);
+                 ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) {
+                Data.exportVectorSchemaRoot(ALLOCATOR, root, null, arr, arrowSchema);
+                writer.writeBatch(arr.memoryAddress(), arrowSchema.memoryAddress());
+            }
+        }
+    }
+
+    private static List<String> readJniUtf8Filtered(Path file, Expression filter) throws IOException {
+        ScanOptions opts = ScanOptions.builder()
+                .projection(Expression.select(new String[]{"k"}, Expression.root()))
+                .filter(filter)
+                .build();
+        var out = new ArrayList<String>();
+        Scan scan = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString()).scan(opts);
+        while (scan.hasNext()) {
+            Partition partition = scan.next();
+            try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) {
+                while (reader.loadNextBatch()) {
+                    VectorSchemaRoot root = reader.getVectorSchemaRoot();
+                    var vec = root.getVector("k");
+                    for (int i = 0; i < root.getRowCount(); i++) {
+                        out.add(vec.isNull(i) ? null : vec.getObject(i).toString());
+                    }
+                }
+            }
+        }
+        return out;
+    }
+
     private static long countChunks(Path file, io.github.dfa1.vortex.reader.ScanOptions opts) throws IOException {
         long[] count = {0};
         try (var vf = VortexReader.open(file, ReadRegistry.loadAll());