From 93475da7c619cc2da826eebd9c74ead6efdc02d6 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 22:46:07 +0200 Subject: [PATCH 1/2] feat(calcite): read ORDER BY without LIMIT lazily on big tables Calcite's own sort buffers every row of the table before it returns the first. Clicking a column header in an IDE grid sends exactly that (SELECT t.* ... ORDER BY col), and on a 100M-row table it ran the JDBC process out of memory. Rewrite a limitless sort over a table of more than 100,000 rows to a scan of a derived, already-ordered table (VortexOrderedTable). It returns the first 1,024 rows from the existing bounded top-N pass and, when the client reads past them, repeats the pass for twice as many and skips what was returned. A grid page costs one pass and holds one page; a client that reads everything holds what Calcite's sort would. The table publishes its collation so the planner drops the sort the query asked for, and the planner's row estimate for the sort input so a filter does not make the plain sort look cheaper. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../vortex/calcite/VortexOrderedTable.java | 126 +++++++++++ .../calcite/VortexSortPushDownRule.java | 49 ++++- .../dfa1/vortex/calcite/LazySortScanTest.java | 200 ++++++++++++++++++ docs/reference.md | 8 +- 5 files changed, 376 insertions(+), 8 deletions(-) create mode 100644 calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java create mode 100644 calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 820a1989d..7ad76bada 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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)) ### Changed +- SQL `ORDER BY` with no `LIMIT` over a table of more than 100,000 rows returns its first page without sorting the whole table, so an IDE grid sorted by clicking a column header no longer runs out of memory. ([#570](https://github.com/dfa1/vortex-java/pull/570)) - **Breaking:** `VortexWriter.create` refuses a nullable root struct: Rust reads a nullable struct only with a validity child the writer never wrote, so such files were unreadable outside vortex-java. ([#493](https://github.com/dfa1/vortex-java/issues/493)) - **Breaking:** a file that does not embed its dtype is refused when it opens, as Rust refuses it unless the caller supplies the dtype, instead of failing the scan with `NullPointerException`. ([#493](https://github.com/dfa1/vortex-java/issues/493)) - **Breaking:** every FlatBuffer in a file is verified before it is read, as Rust verifies it, so corrupt file metadata fails with `VortexException`, and like Rust the reader now refuses tables nested more than 64 deep. ([#493](https://github.com/dfa1/vortex-java/issues/493)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java new file mode 100644 index 000000000..d10583e3c --- /dev/null +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java @@ -0,0 +1,126 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.reader.RowFilter; + +import org.apache.calcite.DataContext; +import org.apache.calcite.linq4j.AbstractEnumerable; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.RelCollation; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.schema.ScannableTable; +import org.apache.calcite.schema.Statistic; +import org.apache.calcite.schema.Statistics; +import org.apache.calcite.schema.impl.AbstractTable; + +import java.util.List; + +/// A [VortexTable] already sorted: what `ORDER BY` with no `LIMIT` over a big table is rewritten to +/// scan, so the rows come back lazily instead of through Calcite's sort, which buffers every row of +/// the table before it returns the first (a 100M-row table ran an IDE's JDBC process out of memory). +/// +/// The first rows come from a bounded top-N pass, [VortexTable#topN]. When the client reads past +/// them, the pass runs again for twice as many rows and the rows already returned are skipped: a +/// client that stops after a grid page costs one pass and holds only that page, one that reads +/// everything costs about two full passes and holds what Calcite's own sort would. Ties rank by scan +/// position, so a longer pass always starts with the rows of the shorter one. +final class VortexOrderedTable extends AbstractTable implements ScannableTable { + + /// Rows the first pass keeps: a few IDE grid pages. + private static final int FIRST_BATCH = 1024; + + private final VortexTable table; + private final int[] cols; + private final RowFilter filter; + private final List keys; + private final RelDataType rowType; + private final double estimatedRows; + private final RelCollation collation; + + VortexOrderedTable(VortexTable table, int[] cols, RowFilter filter, List keys, + RelDataType rowType, double estimatedRows, RelCollation collation) { + this.table = table; + this.cols = cols; + this.filter = filter; + this.keys = keys; + this.rowType = rowType; + this.estimatedRows = estimatedRows; + this.collation = collation; + } + + @Override + public RelDataType getRowType(RelDataTypeFactory typeFactory) { + return rowType; + } + + @Override + public Statistic getStatistic() { + // the planner's estimate for the rows the sort would have received (a filter shrinks it), so this + // scan is costed against the plain sort over the same filtered scan on equal terms; the + // enumerator below never uses it as a bound + // declaring the order is what lets the planner drop the sort the query asked for: without it + // it adds one back on top and the cheaper plain scan wins + return Statistics.of(estimatedRows, List.of(), List.of(collation)); + } + + @Override + public Enumerable scan(DataContext root) { + return new AbstractEnumerable<>() { + @Override + public Enumerator enumerator() { + return new Batches(); + } + }; + } + + /// Returns the sorted rows, running a longer top-N pass whenever the current one is used up. + private final class Batches implements Enumerator { + + private List batch = List.of(); + private int requested; + private long returned; + private int index = -1; + private boolean exhausted; + + @Override + public boolean moveNext() { + index++; + if (index < batch.size()) { + return true; + } + if (exhausted) { + return false; + } + // the first pass keeps FIRST_BATCH rows, every later one twice the last; a pass cannot + // keep more rows than the table has + long tableRows = (long) table.getStatistic().getRowCount().doubleValue(); + long want = requested == 0 ? FIRST_BATCH : 2L * requested; + requested = (int) Math.min(want, Math.min(Integer.MAX_VALUE - 8L, Math.max(1, tableRows))); + batch = table.topN(cols, filter, keys, requested); + exhausted = batch.size() < requested || requested >= tableRows; + // skip the rows an earlier, shorter pass already returned + index = (int) returned; + if (index >= batch.size()) { + return false; + } + return true; + } + + @Override + public Object[] current() { + returned = index + 1L; + return batch.get(index); + } + + @Override + public void reset() { + throw new UnsupportedOperationException(); + } + + @Override + public void close() { + batch = List.of(); + } + } +} diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexSortPushDownRule.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexSortPushDownRule.java index 61d027af4..c6d443e54 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexSortPushDownRule.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexSortPushDownRule.java @@ -7,12 +7,15 @@ import org.apache.calcite.plan.RelOptRule; import org.apache.calcite.plan.RelOptRuleCall; import org.apache.calcite.plan.RelOptRuleOperand; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.prepare.RelOptTableImpl; import org.apache.calcite.rel.RelFieldCollation; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.core.Filter; import org.apache.calcite.rel.core.Project; import org.apache.calcite.rel.core.Sort; import org.apache.calcite.rel.core.TableScan; +import org.apache.calcite.rel.logical.LogicalTableScan; import org.apache.calcite.rel.logical.LogicalValues; import org.apache.calcite.rel.type.RelDataType; import org.apache.calcite.rex.RexBuilder; @@ -49,6 +52,10 @@ public final class VortexSortPushDownRule extends RelOptRule { /// Most rows a rewrite emits; each becomes a literal row of the plan. private static final int MAX_ROWS = 1000; + /// Smallest table, in rows, whose `ORDER BY` with no `LIMIT` is rewritten to a lazy scan + /// ([VortexOrderedTable]); a smaller one fits in memory and Calcite sorts it faster. + static final int LAZY_SORT_MIN_ROWS = 100_000; + /// Most literal cells (rows times columns) a push-down may put in one `VALUES`. Calcite compiles /// the whole `VALUES` into a single Java method, which the JVM caps at 64 KB of bytecode: measured /// on a mixed-type table, 4,500 cells compile and 6,000 do not ("Code grows beyond 64 KB"). A bigger @@ -73,14 +80,24 @@ private VortexSortPushDownRule(RelOptRuleOperand operand, String description) { @Override public void onMatch(RelOptRuleCall call) { Sort sort = call.rel(0); - if (sort.fetch == null || sort.getCollation().getFieldCollations().isEmpty()) { - return; // a full ORDER BY must emit every row - } - int fetch = intLiteral(sort.fetch); - int offset = sort.offset == null ? 0 : intLiteral(sort.offset); - if (fetch < 0 || offset < 0 || fetch + (long) offset > MAX_ROWS) { + if (sort.getCollation().getFieldCollations().isEmpty()) { return; } + // no LIMIT: every row must come back sorted, which only a big table's lazy scan can do cheaply + boolean lazy = sort.fetch == null; + int fetch = 0; + int offset = 0; + if (lazy) { + if (sort.offset != null) { + return; + } + } else { + fetch = intLiteral(sort.fetch); + offset = sort.offset == null ? 0 : intLiteral(sort.offset); + if (fetch < 0 || offset < 0 || fetch + (long) offset > MAX_ROWS) { + return; + } + } Project project = null; Filter filter = null; RelNode current = call.rel(1); @@ -98,6 +115,10 @@ public void onMatch(RelOptRuleCall call) { if (table == null) { return; } + Double rowCount = table.getStatistic().getRowCount(); + if (lazy && (rowCount == null || rowCount <= LAZY_SORT_MIN_ROWS)) { + return; // small enough for Calcite to sort in memory + } List predicates = new ArrayList<>(); if (filter != null) { predicates.add(filter.getCondition()); @@ -115,7 +136,7 @@ public void onMatch(RelOptRuleCall call) { } int[] cols = columns(scan, project); // only the rows after the offset become literals; the offset rows stay in the heap (MAX_ROWS) - if (cols == null || (long) fetch * cols.length > MAX_VALUES_CELLS) { + if (cols == null || (!lazy && (long) fetch * cols.length > MAX_VALUES_CELLS)) { return; } List keys = new ArrayList<>(); @@ -126,6 +147,11 @@ public void onMatch(RelOptRuleCall call) { keys.add(new VortexTable.SortKey(collation.getFieldIndex(), descending, nulls == RelFieldCollation.NullDirection.FIRST)); } + if (lazy) { + call.transformTo(orderedScan(sort, scan, new VortexOrderedTable(table, cols, rowFilter, keys, + sort.getRowType(), call.getMetadataQuery().getRowCount(sort.getInput()), sort.getCollation()))); + return; + } int count = offset + fetch; RowFilter finalFilter = rowFilter; List foldKey = List.of("topN", table, java.util.Arrays.toString(cols), @@ -150,6 +176,15 @@ public void onMatch(RelOptRuleCall call) { call.transformTo(LogicalValues.create(sort.getCluster(), sort.getRowType(), ImmutableList.copyOf(tuples))); } + /// A scan of `ordered` that Calcite plans like any table, in place of the sort and everything it + /// sorted. + private static RelNode orderedScan(Sort sort, TableScan scan, VortexOrderedTable ordered) { + RelOptTable base = scan.getTable(); + RelOptTable derived = RelOptTableImpl.create(base.getRelOptSchema(), sort.getRowType(), ordered, + ImmutableList.builder().addAll(base.getQualifiedName()).add("ordered").build()); + return LogicalTableScan.create(sort.getCluster(), derived, scan.getHints()); + } + /// The table columns the sort's input emits, in order, or `null` when a projected expression is /// not a plain column. private static int[] columns(TableScan scan, Project project) { diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java new file mode 100644 index 000000000..2fbfed8ab --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java @@ -0,0 +1,200 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; + +import org.junit.jupiter.api.BeforeAll; +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.ValueSource; + +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.stream.IntStream; + +import static org.assertj.core.api.Assertions.assertThat; + +/// An `ORDER BY` with no `LIMIT` over a big table is what an IDE grid sends when a column header is +/// clicked. Calcite's own sort buffers every row of the table before it returns the first, which +/// ran a 100M-row table out of memory; the rewrite must instead hand rows back lazily, in the order +/// Calcite's stable sort would, doing work in proportion to how many rows the client reads. +class LazySortScanTest { + + private static final int ROWS = VortexSortPushDownRule.LAZY_SORT_MIN_ROWS + 20_000; + + @TempDir + static Path tmp; + private static Path big; + private static Path small; + + /// `k` repeats (ties must come out in scan order), `n` has nulls, `s` is a string with ties. + @BeforeAll + static void writeFiles() throws Exception { + big = write("big.vortex", ROWS); + small = write("small.vortex", 500); + } + + @Test + void orderByWithoutLimit_overABigTableIsNotSortedByCalcite() throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", big))) { + + // When + String plan = plan(conn, "select * from vtx.t order by k"); + + // Then no EnumerableSort buffering the whole table + assertThat(plan).doesNotContain("Sort"); + } + } + + @Test + void orderByWithoutLimit_overASmallTableStaysWithCalcite() throws Exception { + // Given a table that fits in memory anyway + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", small))) { + + // When + String plan = plan(conn, "select * from vtx.t order by k"); + + // Then + assertThat(plan).contains("Sort"); + } + } + + @ParameterizedTest + @ValueSource(strings = {"k", "k desc", "s desc, k", "n", "n desc", "n nulls first", "n desc nulls last"}) + void readingEveryRow_matchesAStableSort(String orderBy) throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", big))) { + + // When every row is read, past several batch doublings + String sql = "select id from vtx.t order by " + orderBy; + List result = ids(conn, sql); + + // Then the same order a stable sort over the scan gives: ties in scan order, and it was + // the lazy scan that produced it, not Calcite's sort + assertThat(plan(conn, sql)).doesNotContain("Sort"); + assertThat(result).isEqualTo(expected(orderBy)); + } + } + + @Test + void readingOnlyTheFirstPage_returnsTheTopRows() throws Exception { + // Given a client that stops after a grid page, as an IDE does + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", big)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select id from vtx.t order by k desc")) { + + // When + List result = new ArrayList<>(); + while (result.size() < 500 && rs.next()) { + result.add(rs.getLong(1)); + } + + // Then + assertThat(result).isEqualTo(expected("k desc").subList(0, 500)); + } + } + + @Test + void filterAndProjection_stillSortLazilyAndCorrectly() throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", big))) { + + // When + String sql = "select id from vtx.t where k > 900 order by s desc, k"; + List result = ids(conn, sql); + + // Then + List expected = IntStream.range(0, ROWS).boxed() + .filter(i -> key(i) > 900) + .sorted(Comparator.comparing(LazySortScanTest::text).reversed() + .thenComparingLong(LazySortScanTest::key)) + .map(Integer::longValue).toList(); + assertThat(plan(conn, sql)).doesNotContain("Sort"); + assertThat(result).isEqualTo(expected); + } + } + + private static List expected(String orderBy) { + Comparator byK = Comparator.comparingLong(LazySortScanTest::key); + Comparator order = switch (orderBy) { + case "k" -> byK; + case "k desc" -> byK.reversed(); + case "s desc, k" -> Comparator.comparing(LazySortScanTest::text).reversed().thenComparing(byK); + // Calcite's default null collation is HIGH: a null is the largest value, so it ends an + // ascending sort and starts a descending one + case "n" -> Comparator.comparing(LazySortScanTest::nullable, Comparator.nullsLast(Comparator.naturalOrder())); + case "n desc" -> Comparator.comparing(LazySortScanTest::nullable, Comparator.nullsFirst(Comparator.reverseOrder())); + case "n nulls first" -> Comparator.comparing(LazySortScanTest::nullable, Comparator.nullsFirst(Comparator.naturalOrder())); + case "n desc nulls last" -> Comparator.comparing(LazySortScanTest::nullable, Comparator.nullsLast(Comparator.reverseOrder())); + default -> throw new IllegalArgumentException(orderBy); + }; + return IntStream.range(0, ROWS).boxed().sorted(order).map(Integer::longValue).toList(); + } + + private static long key(int i) { + return (i * 7919L) % 1000; + } + + private static Long nullable(int i) { + return i % 7 == 0 ? null : (i * 31L) % 500; + } + + private static String text(int i) { + return "s" + (i % 13); + } + + private static Path write(String name, int rows) throws Exception { + Path file = tmp.resolve(name); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("k"), DType.I64) + .field(ColumnName.of("n"), DType.I64.asNullable()) + .field(ColumnName.of("s"), new DType.Utf8(false)) + .build(); + long[] id = new long[rows]; + long[] k = new long[rows]; + Long[] n = new Long[rows]; + String[] s = new String[rows]; + for (int i = 0; i < rows; i++) { + id[i] = i; + k[i] = key(i); + n[i] = nullable(i); + s[i] = text(i); + } + try (var channel = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = VortexWriter.create(channel, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("k"), k, + ColumnName.of("n"), n, ColumnName.of("s"), s)); + } + return file; + } + + private static List ids(Connection conn, String sql) throws Exception { + List ids = new ArrayList<>(); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + while (rs.next()) { + ids.add(rs.getLong(1)); + } + } + return ids; + } + + private static String plan(Connection conn, String sql) throws Exception { + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery("explain plan for " + sql)) { + rs.next(); + return rs.getString(1); + } + } +} diff --git a/docs/reference.md b/docs/reference.md index 8cf361aff..aa8d8b9da 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -650,7 +650,13 @@ by the scan. `ORDER BY ... LIMIT n` (any `OFFSET`, `n + offset` up to 1,000) over plain columns, with the same kind of `WHERE`, keeps the first rows in a bounded heap during the scan instead of sorting every row; it falls back to Calcite's sort when a returned value has no SQL literal of its type (`NaN`, -an infinity, a string outside Latin-1, binary, `UUID`, nested). +an infinity, a string outside Latin-1, binary, `UUID`, nested), and is also capped by the 4,000 +literal cells (rows times columns) one compiled `VALUES` can hold, past which Calcite sorts. An +`ORDER BY` with no `LIMIT` over a table of more than 100,000 rows is read lazily: the first 1,024 rows +come from a bounded heap, and a client that reads further triggers a pass for twice as many, skipping +the rows already returned. A grid that shows one page costs one pass and holds one page (Calcite's own +sort buffers every row before it returns the first), and a client that reads everything holds what +Calcite's sort would. Ties come out in scan order, as in Calcite's stable sort. `connect` is the supported entry point; `VortexSchema`/`VortexTable` (`io.github.dfa1.vortex.calcite`) are the underlying schema/table plumbing it assembles, exposed for callers building a Calcite schema tree From d1045e6d532504a5c3904102c88d7e4f28bae8ac Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Mon, 19 Oct 2026 06:57:02 +0200 Subject: [PATCH 2/2] fix(calcite): address review of the lazy sort scan - count the rows handed out when the cursor advances, not when a column is read: a client that only steps through rows got earlier rows again on the next pass; - grow each pass eightfold and say what a pass costs: every pass reads the whole table, so a client reading to the end of 100M rows pays seven, not two; - drop the previous batch before the next pass runs; - make topN report through the table's chunksScannedLastQuery counter; - trim the CHANGELOG entry to one sentence. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 2 +- .../vortex/calcite/VortexOrderedTable.java | 35 ++++++++----- .../dfa1/vortex/calcite/VortexTable.java | 2 +- .../dfa1/vortex/calcite/LazySortScanTest.java | 49 +++++++++++++++++++ docs/reference.md | 9 ++-- 5 files changed, 79 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ad76bada..ab6a88806 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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)) ### Changed -- SQL `ORDER BY` with no `LIMIT` over a table of more than 100,000 rows returns its first page without sorting the whole table, so an IDE grid sorted by clicking a column header no longer runs out of memory. ([#570](https://github.com/dfa1/vortex-java/pull/570)) +- SQL `ORDER BY` with no `LIMIT` over a large table returns its first rows without sorting the whole table first. ([#570](https://github.com/dfa1/vortex-java/pull/570)) - **Breaking:** `VortexWriter.create` refuses a nullable root struct: Rust reads a nullable struct only with a validity child the writer never wrote, so such files were unreadable outside vortex-java. ([#493](https://github.com/dfa1/vortex-java/issues/493)) - **Breaking:** a file that does not embed its dtype is refused when it opens, as Rust refuses it unless the caller supplies the dtype, instead of failing the scan with `NullPointerException`. ([#493](https://github.com/dfa1/vortex-java/issues/493)) - **Breaking:** every FlatBuffer in a file is verified before it is read, as Rust verifies it, so corrupt file metadata fails with `VortexException`, and like Rust the reader now refuses tables nested more than 64 deep. ([#493](https://github.com/dfa1/vortex-java/issues/493)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java index d10583e3c..ff82e2a69 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java @@ -21,15 +21,22 @@ /// the table before it returns the first (a 100M-row table ran an IDE's JDBC process out of memory). /// /// The first rows come from a bounded top-N pass, [VortexTable#topN]. When the client reads past -/// them, the pass runs again for twice as many rows and the rows already returned are skipped: a -/// client that stops after a grid page costs one pass and holds only that page, one that reads -/// everything costs about two full passes and holds what Calcite's own sort would. Ties rank by scan -/// position, so a longer pass always starts with the rows of the shorter one. +/// them, the pass runs again for eight times as many rows and the rows already returned are skipped. +/// Ties rank by scan position, so a longer pass always starts with the rows of the shorter one. +/// +/// Each pass reads the whole table: the heap bounds memory, not I/O. A client that stops after a grid +/// page costs one pass and holds only that page, which is the case this exists for. One that reads on +/// pays another full pass each time it outgrows the last (a 100M-row table read to the end is seven), and holds what Calcite's own sort would at the end. A file replaced while a result is being +/// read is not detected: each pass reopens it by path. final class VortexOrderedTable extends AbstractTable implements ScannableTable { /// Rows the first pass keeps: a few IDE grid pages. private static final int FIRST_BATCH = 1024; + /// How much longer each later pass is. Every pass is a full scan, so growing fast bounds how many a + /// client that reads far pays for. + private static final int GROWTH = 8; + private final VortexTable table; private final int[] cols; private final RowFilter filter; @@ -87,29 +94,33 @@ private final class Batches implements Enumerator { public boolean moveNext() { index++; if (index < batch.size()) { + returned = index + 1L; return true; } if (exhausted) { return false; } - // the first pass keeps FIRST_BATCH rows, every later one twice the last; a pass cannot - // keep more rows than the table has + // the old batch is dead weight while the next pass runs + batch = List.of(); + // the first pass keeps FIRST_BATCH rows, every later one GROWTH times the last; a pass + // cannot keep more rows than the table has long tableRows = (long) table.getStatistic().getRowCount().doubleValue(); - long want = requested == 0 ? FIRST_BATCH : 2L * requested; + long want = requested == 0 ? FIRST_BATCH : (long) GROWTH * requested; requested = (int) Math.min(want, Math.min(Integer.MAX_VALUE - 8L, Math.max(1, tableRows))); batch = table.topN(cols, filter, keys, requested); exhausted = batch.size() < requested || requested >= tableRows; - // skip the rows an earlier, shorter pass already returned + // skip the rows an earlier, shorter pass already returned; counted when advanced over, not + // when read, since a client may step through rows without reading a column index = (int) returned; - if (index >= batch.size()) { - return false; + boolean more = index < batch.size(); + if (more) { + returned = index + 1L; } - return true; + return more; } @Override public Object[] current() { - returned = index + 1L; return batch.get(index); } diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java index 5ef917170..bd6b8c399 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexTable.java @@ -645,7 +645,7 @@ public List topN(int[] cols, RowFilter filter, List keys, int .thenComparingLong(Ranked::position); // the worst kept row at the head, so a better row evicts it java.util.PriorityQueue heap = new java.util.PriorityQueue<>(count + 1, ranking.reversed()); - VortexEnumerator rows = new VortexEnumerator(file, new AtomicLong(), options, outNames, outTypes, filter); + VortexEnumerator rows = new VortexEnumerator(file, chunksScannedLastQuery, options, outNames, outTypes, filter); // only the key columns are read per row; a row is built once it beats the worst one kept. A // later row never beats an equal key, so the first rows of a tie are the ones kept Object[] probe = new Object[cols.length]; diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java index 2fbfed8ab..09282fd05 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java @@ -5,6 +5,11 @@ import io.github.dfa1.vortex.writer.VortexWriter; import io.github.dfa1.vortex.writer.WriteOptions; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.RelCollations; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeSystem; +import org.apache.calcite.sql.type.SqlTypeFactoryImpl; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; @@ -106,6 +111,50 @@ void readingOnlyTheFirstPage_returnsTheTopRows() throws Exception { } } + @Test + void advancingWithoutReadingARow_neverRepeatsOrDropsOne() { + // Given the enumerator driven by moveNext() alone, as a cursor that skips rows or only counts + // them does; JDBC cannot show this, Calcite's cursor reads every row it passes + VortexTable table = new VortexTable(big); + RelDataType rowType = table.getRowType(new SqlTypeFactoryImpl(RelDataTypeSystem.DEFAULT)); + int[] allColumns = IntStream.range(0, rowType.getFieldCount()).toArray(); + VortexOrderedTable sut = new VortexOrderedTable(table, allColumns, null, + List.of(new VortexTable.SortKey(1, false, false)), rowType, ROWS, RelCollations.of(1)); + + // When it is stepped through every batch, reading a row only now and then + List sampled = new ArrayList<>(); + long count = 0; + try (Enumerator rows = sut.scan(null).enumerator()) { + while (rows.moveNext()) { + if (count % 5_000 == 0) { + sampled.add((Long) rows.current()[0]); + } + count++; + } + } + + // Then each pass resumed exactly after the last row it handed out + List expected = expected("k"); + assertThat(count).isEqualTo(ROWS); + assertThat(sampled).isEqualTo(IntStream.range(0, (ROWS + 4_999) / 5_000) + .mapToObj(i -> expected.get(i * 5_000)).toList()); + } + + @Test + void orderedScan_reportsItsChunksThroughTheTablesInstrumentation() throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", big))) { + VortexTable table = conn.unwrap(org.apache.calcite.jdbc.CalciteConnection.class).getRootSchema() + .subSchemas().get("vtx").unwrap(VortexSchema.class).table("t"); + + // When + ids(conn, "select id from vtx.t order by k"); + + // Then the counter is the sorted scan's, not whatever an earlier query left + assertThat(table.chunksScannedLastQuery()).isPositive(); + } + } + @Test void filterAndProjection_stillSortLazilyAndCorrectly() throws Exception { // Given diff --git a/docs/reference.md b/docs/reference.md index aa8d8b9da..cda5fe8a5 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -653,10 +653,11 @@ row; it falls back to Calcite's sort when a returned value has no SQL literal of an infinity, a string outside Latin-1, binary, `UUID`, nested), and is also capped by the 4,000 literal cells (rows times columns) one compiled `VALUES` can hold, past which Calcite sorts. An `ORDER BY` with no `LIMIT` over a table of more than 100,000 rows is read lazily: the first 1,024 rows -come from a bounded heap, and a client that reads further triggers a pass for twice as many, skipping -the rows already returned. A grid that shows one page costs one pass and holds one page (Calcite's own -sort buffers every row before it returns the first), and a client that reads everything holds what -Calcite's sort would. Ties come out in scan order, as in Calcite's stable sort. +come from a bounded heap, and a client that reads further triggers another pass for eight times as +many, skipping the rows already returned. Every pass reads the whole table (the heap bounds memory, +not I/O), so a grid that shows one page costs one pass and holds one page, while a client that reads +a 100M-row table to the end pays seven passes and holds what Calcite's own sort would (which buffers +every row before it returns the first). Ties come out in scan order, as in Calcite's stable sort. `connect` is the supported entry point; `VortexSchema`/`VortexTable` (`io.github.dfa1.vortex.calcite`) are the underlying schema/table plumbing it assembles, exposed for callers building a Calcite schema tree