perf(calcite): stream scan rows lazily instead of materialising a List<Object[]>

VortexTable.scan() built the entire result as a List<Object[]> (one fresh array +
boxed cell per row) before returning. For a 1M-row full scan an async-profiler run
showed ~72% of CPU in G1 GC: every row was promoted into the old gen and the whole
result stayed live at once. Column decode itself was ~0.5%.

Replace it with a streaming Enumerator that advances chunk by chunk, decoding each
requested column once per chunk and yielding one row per moveNext(). Rows are no
longer retained, so the working set is one chunk and rows die in the young gen.
Fresh array per row is kept (correct for ORDER BY / joins that retain rows).

Measured (CalciteDemo, 1M rows, MIN/MAX/COUNT full scan): GC 71% -> 3%,
~52 ms/query -> ~28 ms/query.

CalciteDemo is a profiling harness, disabled unless -Ddemo.profile=true; run under
async-profiler by attaching to the forked test JVM (argLine is owned by the
byte-buddy agent goal, so attach by PID rather than -agentpath).

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

Author: dfa1

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

@@ -15,8 +15,9 @@
 
 import org.apache.calcite.DataContext;
 import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.linq4j.AbstractEnumerable;
 import org.apache.calcite.linq4j.Enumerable;
-import org.apache.calcite.linq4j.Linq4j;
+import org.apache.calcite.linq4j.Enumerator;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.rex.RexBuilder;
@@ -139,21 +140,108 @@ public Enumerable<Object[]> scan(DataContext root, List<RexNode> filters, int[]
             options = ScanOptions.columns(outNames);
         }
 
-        List<Object[]> rows = new ArrayList<>();
-        long scanned = 0;
-        try (VortexReader reader = VortexReader.open(file);
-             ScanIterator scan = reader.scan(options)) {
-            while (scan.hasNext()) {
-                try (Chunk chunk = scan.next()) {
-                    scanned++;
-                    appendChunk(chunk, outNames, outTypes, rows);
+        // Stream rows lazily: decode one chunk at a time and yield a fresh row, so rows die young
+        // (in G1's young gen) instead of piling a whole-result List<Object[]> into the old gen — the
+        // dominant cost an async-profiler run showed for the full-scan path (~72% in GC).
+        ScanOptions scanOptions = options;
+        return new AbstractEnumerable<>() {
+            @Override
+            public Enumerator<Object[]> enumerator() {
+                return new VortexEnumerator(scanOptions, outNames, outTypes);
+            }
+        };
+    }
+
+    /// Streaming [Enumerator] over a Vortex scan: advances chunk by chunk, decoding each requested
+    /// column once per chunk and materialising one `Object[]` row per [#moveNext()]. Rows are not
+    /// retained, so the working set stays at one chunk rather than the whole result.
+    private final class VortexEnumerator implements Enumerator<Object[]> {
+
+        private final String[] names;
+        private final DType[] types;
+        private final VortexReader reader;
+        private final ScanIterator scan;
+        private Chunk chunk;
+        private Object[] columns;
+        private long rowInChunk;
+        private long chunkRows;
+        private Object[] current;
+
+        private VortexEnumerator(ScanOptions options, String[] names, DType[] types) {
+            this.names = names;
+            this.types = types;
+            chunksScannedLastQuery.set(0);
+            VortexReader openedReader = null;
+            try {
+                openedReader = VortexReader.open(file);
+                this.reader = openedReader;
+                this.scan = openedReader.scan(options);
+            } catch (IOException e) {
+                closeQuietly(openedReader);
+                throw new UncheckedIOException("cannot scan " + file, e);
+            } catch (RuntimeException e) {
+                closeQuietly(openedReader);
+                throw e;
+            }
+        }
+
+        private void closeQuietly(VortexReader r) {
+            if (r != null) {
+                r.close();
+            }
+        }
+
+        @Override
+        public Object[] current() {
+            return current;
+        }
+
+        @Override
+        public boolean moveNext() {
+            while (true) {
+                if (chunk != null && rowInChunk < chunkRows) {
+                    Object[] row = new Object[names.length];
+                    for (int c = 0; c < names.length; c++) {
+                        row[c] = value(columns[c], types[c], rowInChunk);
+                    }
+                    rowInChunk++;
+                    current = row;
+                    return true;
+                }
+                if (chunk != null) {
+                    chunk.close();
+                    chunk = null;
+                }
+                if (!scan.hasNext()) {
+                    return false;
+                }
+                chunk = scan.next();
+                chunksScannedLastQuery.incrementAndGet();
+                chunkRows = chunk.rowCount();
+                rowInChunk = 0;
+                columns = new Object[names.length];
+                for (int c = 0; c < names.length; c++) {
+                    columns[c] = chunk.column(names[c]);
                 }
             }
-        } catch (IOException e) {
-            throw new UncheckedIOException("cannot scan " + file, e);
         }
-        chunksScannedLastQuery.set(scanned);
-        return Linq4j.asEnumerable(rows);
+
+        @Override
+        public void reset() {
+            throw new UnsupportedOperationException("VortexEnumerator does not support reset");
+        }
+
+        @Override
+        public void close() {
+            try {
+                if (chunk != null) {
+                    chunk.close();
+                }
+            } finally {
+                scan.close();
+                reader.close();
+            }
+        }
     }
 
     private DType.Struct struct() {
@@ -175,21 +263,6 @@ private static int[] allColumns(int n) {
         return all;
     }
 
-    private static void appendChunk(Chunk chunk, String[] names, DType[] types, List<Object[]> rows) {
-        long n = chunk.rowCount();
-        Object[] arrays = new Object[names.length];
-        for (int c = 0; c < names.length; c++) {
-            arrays[c] = chunk.column(names[c]);
-        }
-        for (long r = 0; r < n; r++) {
-            Object[] row = new Object[names.length];
-            for (int c = 0; c < names.length; c++) {
-                row[c] = value(arrays[c], types[c], r);
-            }
-            rows.add(row);
-        }
-    }
-
     private static Object value(Object array, DType type, long r) {
         return switch (type) {
             case DType.Primitive p -> switch (p.ptype()) {

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

@@ -0,0 +1,61 @@
+package io.github.dfa1.vortex.calcite;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.Map;
+import java.util.Properties;
+
+/// Profiling harness (disabled in normal runs). Writes a 1M-row OHLC Vortex file once, then runs
+/// the full-table `MIN/MAX/COUNT` SQL many times through Calcite so a CPU profile has enough
+/// samples to show where the scan path spends its time.
+///
+/// Run under async-profiler:
+/// ```
+/// ./mvnw test -pl calcite -am -Dtest=CalciteDemo -Ddemo.profile=true \
+///   -DargLine="-agentpath:/opt/homebrew/lib/libasyncProfiler.dylib=start,event=cpu,file=/tmp/calcite.collapsed,collapsed"
+/// ```
+class CalciteDemo {
+
+    @Test
+    @EnabledIfSystemProperty(named = "demo.profile", matches = "true")
+    void profileFullScan() throws Exception {
+        int rows = Integer.getInteger("demo.rows", 1_000_000);
+        int iterations = Integer.getInteger("demo.iterations", 300);
+
+        Path file = Files.createTempFile("ohlc-demo", ".vortex");
+        try {
+            OhlcGenerator.write(file, rows, 10_000);
+            System.out.printf("wrote %,d rows -> %.1f MB%n", rows, Files.size(file) / 1048576.0);
+
+            Properties info = new Properties();
+            info.setProperty("lex", "JAVA");
+            String sql = "select min(low) lo, max(high) hi, count(*) c from vtx.ohlc";
+
+            try (Connection conn = DriverManager.getConnection("jdbc:calcite:", info)) {
+                conn.unwrap(org.apache.calcite.jdbc.CalciteConnection.class).getRootSchema()
+                        .add("vtx", new VortexSchema(Map.of("ohlc", file)));
+
+                long t0 = System.nanoTime();
+                long count = 0;
+                for (int i = 0; i < iterations; i++) {
+                    try (Statement st = conn.createStatement();
+                         ResultSet rs = st.executeQuery(sql)) {
+                        rs.next();
+                        count = rs.getLong("c");
+                    }
+                }
+                double ms = (System.nanoTime() - t0) / 1e6 / iterations;
+                System.out.printf("full scan x%d: %.2f ms/query | count=%,d%n", iterations, ms, count);
+            }
+        } finally {
+            Files.deleteIfExists(file);
+        }
+    }
+}