perf(reader): a filtered scan copies the matching rows into flat arrays (#602)

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Performance
+- A filtered scan copies the matching rows into flat arrays, as Rust's `filter` does, instead of reading them through a per-row index lookup, and allocates up to 4.7× less. ([#602](https://github.com/dfa1/vortex-java/issues/602))
 - Scanning an integer column with a dict layout folds its codes against the dictionary instead of expanding every row first, about 16% faster on a two-column taxi scan. ([#603](https://github.com/dfa1/vortex-java/pull/603))
 - Folding dict-encoded float columns and scan windows reads their buffers directly, about 3% faster on a two-column taxi scan. ([#600](https://github.com/dfa1/vortex-java/pull/600))
 

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

@@ -113,12 +113,12 @@ public Chunk slice(long offset, long length) {
     /// It takes over this chunk's arena, so this chunk must not be used again; closing the result
     /// releases both.
     ///
-    /// @param rows      the rows to keep, ascending, each in `[0, rowCount())`
+    /// @param rows      the rows to keep, strictly ascending, each in `[0, rowCount())`
     /// @param names     the columns to keep, a subset of this chunk's
     /// @param positions each of `names` to its index
     /// @param onClose   called when the result closes
     /// @return the selected rows
-    Chunk select(long[] rows, List<ColumnName> names, Map<ColumnName, Integer> positions, Consumer<Chunk> onClose) {
+    Chunk select(int[] rows, List<ColumnName> names, Map<ColumnName, Integer> positions, Consumer<Chunk> onClose) {
         IntFunction<Column> selected = i -> {
             Column column = column(position(names.get(i)));
             return new Column(SelectedArrays.select(column.array(), rows, arena), column.dtype());

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

@@ -108,7 +108,7 @@ public final class ScanIterator implements Iterator<Chunk>, AutoCloseable {
     // A chunk hasNext() decoded to test the filter on, and its matching rows, until next() returns
     // them: a chunk with no match is skipped, as Rust's scan returns no empty batch.
     private Chunk pending;
-    private long[] pendingRows;
+    private int[] pendingRows;
     private List<ColumnName> projectedNames;
     private List<DType> projectedDtypes;
     private Map<ColumnName, Integer> projectedPositions;
@@ -364,11 +364,16 @@ private boolean selectMatchingRows(ChunkSpec spec) {
         Chunk full = chunkFor(spec, spec.rowCount());
         try {
             RowSelection selection = Compute.matcher(full, filter);
-            long[] rows = new long[(int) Math.min(spec.rowCount(), limit - rowsReturned)];
+            // the matching rows, grown as they come: memory follows the matches, not the chunk
+            int allowed = (int) Math.min(spec.rowCount(), limit - rowsReturned);
+            int[] rows = new int[Math.min(allowed, 1024)];
             int count = 0;
-            for (long row = selection.nextSelected(0); row >= 0 && count < rows.length;
+            for (long row = selection.nextSelected(0); row >= 0 && count < allowed;
                     row = selection.nextSelected(row + 1)) {
-                rows[count++] = row;
+                if (count == rows.length) {
+                    rows = Arrays.copyOf(rows, (int) Math.min(allowed, 2L * count));
+                }
+                rows[count++] = (int) row;
             }
             if (count == 0) {
                 full.close();
@@ -421,7 +426,7 @@ public Chunk next() {
         Chunk chunk;
         if (pending != null) {
             chunk = pending;
-            long[] rows = pendingRows;
+            int[] rows = pendingRows;
             pending = null;
             pendingRows = null;
             chunk = selectedChunk(chunk, rows);
@@ -437,7 +442,7 @@ public Chunk next() {
 
     /// The `rows` of `full` over the output columns, or `full` itself when that is every row and
     /// every column.
-    private Chunk selectedChunk(Chunk full, long[] rows) {
+    private Chunk selectedChunk(Chunk full, int[] rows) {
         List<ColumnName> names = outputNames;
         Map<ColumnName, Integer> positions = outputPositions;
         if (singleColumnIsSyntheticWrapper) {