diff --git a/CHANGELOG.md b/CHANGELOG.md index 820a1989d..ab6a88806 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 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 new file mode 100644 index 000000000..ff82e2a69 --- /dev/null +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexOrderedTable.java @@ -0,0 +1,137 @@ +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 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; + 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()) { + returned = index + 1L; + return true; + } + if (exhausted) { + return false; + } + // 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 : (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; counted when advanced over, not + // when read, since a client may step through rows without reading a column + index = (int) returned; + boolean more = index < batch.size(); + if (more) { + returned = index + 1L; + } + return more; + } + + @Override + public Object[] current() { + 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/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 new file mode 100644 index 000000000..09282fd05 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LazySortScanTest.java @@ -0,0 +1,249 @@ +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.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; +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 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 + 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..cda5fe8a5 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -650,7 +650,14 @@ 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 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