From 8b41e3bb8994a35b19606d51ac0bd34d175b5307 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:42:05 +0200 Subject: [PATCH 1/7] perf(calcite): read a table's schema and row count once VortexTable reopened and memory-mapped the file for every getRowType, scan, totalRows and aggregate call, so one query opened it several times. The struct dtype and chunk row count are now read on first use and cached (a Vortex file is write-once). Column statistics stay uncached: folding them walks every column's zones, which only the aggregate rule needs. Also drops "demo" from the module description. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + calcite/pom.xml | 2 +- .../dfa1/vortex/calcite/VortexTable.java | 51 +++++++++++++------ .../calcite/VortexAdapterCoverageTest.java | 16 ++++++ 4 files changed, 54 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ae9b68d05..6bda8976f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,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 +- 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)) - **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)) diff --git a/calcite/pom.xml b/calcite/pom.xml index 2f0eff1e5..c41131f9b 100644 --- a/calcite/pom.xml +++ b/calcite/pom.xml @@ -10,7 +10,7 @@ vortex-calcite vortex-calcite - Apache Calcite SQL adapter over the Vortex columnar file format (demo: filter/project/aggregate push-down). + Apache Calcite SQL adapter over the Vortex columnar file format with projection, filter and aggregate push-down. 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 99b3afc41..ed86a92ed 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 @@ -50,10 +50,22 @@ /// only partially captures is left in `filters` for Calcite's own row check. The win is decoding /// far fewer chunks when the filter is selective on a clustered column, and — for a fully captured /// filter — not re-evaluating it a second time over the chunks that are decoded. +/// +/// The schema and row count are read from the footer on first use and cached: +/// a Vortex file is written once, so they cannot change under the table. public final class VortexTable extends AbstractTable implements ProjectableFilterableTable, TranslatableTable { private final Path file; private final AtomicLong chunksScannedLastQuery = new AtomicLong(); + // Read lazily so a missing file fails at query time, as before; a racing first read only + // reads the footer twice, so no lock. + private volatile Footer footer; + + /// The footer facts every planning call needs, read once instead of reopening the file for each + /// `getRowType`, `scan` and aggregate. Column statistics stay out: folding them walks every + /// column's zones, which only the aggregate rule needs. + private record Footer(DType.Struct struct, long totalRows) { + } /// Creates a table backed by the Vortex file at `file`. /// @@ -88,9 +100,9 @@ public record ColumnStats(ArrayStats stats, long totalRows) { /// @param column the column name /// @return the column's aggregated statistics paired with the file's total row count public ColumnStats statsAndRows(ColumnName column) { + long totalRows = footer().totalRows(); try (VortexReader reader = VortexReader.open(file)) { - ArrayStats stats = reader.columnStats().getOrDefault(column, ArrayStats.empty()); - return new ColumnStats(stats, countRows(reader)); + return new ColumnStats(reader.columnStats().getOrDefault(column, ArrayStats.empty()), totalRows); } catch (IOException e) { throw new UncheckedIOException("cannot read stats of " + file, e); } @@ -120,11 +132,12 @@ public record ZoneSum(Number sum, Long nullCount, long totalRows) { /// @param column the numeric column name /// @return the folded sum with the null and row counts needed to interpret a zero public ZoneSum zoneSum(ColumnName column) { + long totalRows = footer().totalRows(); try (VortexReader reader = VortexReader.open(file)) { Number sum = new ZoneReducer(reader).sum(column); Long nullCount = reader.columnStats() .getOrDefault(column, ArrayStats.empty()).nullCount(); - return new ZoneSum(sum, nullCount, countRows(reader)); + return new ZoneSum(sum, nullCount, totalRows); } catch (IOException e) { throw new UncheckedIOException("cannot read zone sum of " + file, e); } @@ -134,15 +147,30 @@ public ZoneSum zoneSum(ColumnName column) { /// /// @return the number of rows in the file public long totalRows() { + return footer().totalRows(); + } + + private Footer footer() { + Footer f = footer; + if (f == null) { + f = readFooter(); + footer = f; + } + return f; + } + + private Footer readFooter() { try (VortexReader reader = VortexReader.open(file)) { - return countRows(reader); + if (!(reader.dtype() instanceof DType.Struct s)) { + throw new IllegalStateException("top-level Vortex dtype is not a struct: " + reader.dtype()); + } + return new Footer(s, countRows(reader)); } catch (IOException e) { - throw new UncheckedIOException("cannot count rows of " + file, e); + throw new UncheckedIOException("cannot read the footer of " + file, e); } } - /// Sums the chunk row counts of `reader` from chunk metadata, without decoding data — shared by - /// [#totalRows()] and [#zoneSum(ColumnName)] so both read the count from a single open reader. + /// Sums the chunk row counts of `reader` from chunk metadata, without decoding data. private static long countRows(VortexReader reader) throws IOException { long total = 0; try (ScanIterator scan = reader.scan(ScanOptions.all())) { @@ -699,14 +727,7 @@ public Enumerator enumerator() { } private DType.Struct struct() { - try (VortexReader reader = VortexReader.open(file)) { - if (!(reader.dtype() instanceof DType.Struct s)) { - throw new IllegalStateException("top-level Vortex dtype is not a struct: " + reader.dtype()); - } - return s; - } catch (IOException e) { - throw new UncheckedIOException("cannot read schema of " + file, e); - } + return footer().struct(); } private static int[] allColumns(int n) { diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java index b8da92fbf..94f55adf4 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java @@ -137,6 +137,22 @@ void table_totalRowsAndStats() { assertThat(columnStats.totalRows()).isEqualTo(3); } + @Test + void schemaAndRowCount_areReadFromTheFooterOnce(@TempDir Path dir) throws Exception { + // Given a table whose footer was read, then the file goes away + Path copy = java.nio.file.Files.copy(file, dir.resolve("copy.vortex")); + VortexTable table = new VortexTable(copy); + table.getRowType(new JavaTypeFactoryImpl()); + java.nio.file.Files.delete(copy); + + // When — planning calls that used to reopen the file each time + long result = table.totalRows(); + + // Then + assertThat(result).isEqualTo(3); + assertThat(table.getRowType(new JavaTypeFactoryImpl()).getFieldCount()).isEqualTo(12); + } + @Test void enumerator_resetUnsupportedAndCloseReleasesOpenChunk() { // Given a scan positioned inside the first chunk From 73ef04c1cd64cfcf4897363b7d6d0e2d909e5c73 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:42:19 +0200 Subject: [PATCH 2/7] feat(calcite): report each table's row count to the planner VortexTable did not override getStatistic, so Calcite saw an unknown row count and guessed table sizes when ordering joins. It now reports the row count read from chunk metadata (no decode). Cross-checked against a vortex-jni-written multi-batch file: the statistic and SELECT COUNT(*) both match what Rust wrote. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../dfa1/vortex/calcite/VortexTable.java | 13 ++- .../calcite/VortexAdapterCoverageTest.java | 13 +++ .../CalciteInteropIntegrationTest.java | 85 +++++++++++++++++++ 4 files changed, 111 insertions(+), 1 deletion(-) create mode 100644 integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bda8976f..b355cf7f9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### Added +- 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)) - `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)) 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 ed86a92ed..d350a2066 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 @@ -25,6 +25,8 @@ import org.apache.calcite.rel.type.RelDataTypeFactory; import org.apache.calcite.rex.RexNode; import org.apache.calcite.schema.ProjectableFilterableTable; +import org.apache.calcite.schema.Statistic; +import org.apache.calcite.schema.Statistics; import org.apache.calcite.schema.TranslatableTable; import org.apache.calcite.schema.impl.AbstractTable; import org.apache.calcite.sql.type.SqlTypeName; @@ -62,7 +64,7 @@ public final class VortexTable extends AbstractTable implements ProjectableFilte private volatile Footer footer; /// The footer facts every planning call needs, read once instead of reopening the file for each - /// `getRowType`, `scan` and aggregate. Column statistics stay out: folding them walks every + /// `getRowType`, `scan` and statistic. Column statistics stay out: folding them walks every /// column's zones, which only the aggregate rule needs. private record Footer(DType.Struct struct, long totalRows) { } @@ -150,6 +152,15 @@ public long totalRows() { return footer().totalRows(); } + /// Gives the planner the file's row count, so join ordering and costing see the real table + /// size instead of Calcite's default guess. + /// + /// @return a statistic carrying the row count, with no declared keys + @Override + public Statistic getStatistic() { + return Statistics.of(footer().totalRows(), List.of()); + } + private Footer footer() { Footer f = footer; if (f == null) { diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java index 94f55adf4..dc431fed5 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/VortexAdapterCoverageTest.java @@ -137,6 +137,19 @@ void table_totalRowsAndStats() { assertThat(columnStats.totalRows()).isEqualTo(3); } + @Test + void getStatistic_reportsTheRowCount() { + // Given + VortexTable table = new VortexTable(file); + + // When + Double result = table.getStatistic().getRowCount(); + + // Then — Calcite's default (no statistic) is a null row count, which makes it guess the + // table size when ordering joins + assertThat(result).isEqualTo(3.0); + } + @Test void schemaAndRowCount_areReadFromTheFooterOnce(@TempDir Path dir) throws Exception { // Given a table whose footer was read, then the file goes away diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java new file mode 100644 index 000000000..0bf6a8029 --- /dev/null +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -0,0 +1,85 @@ +package io.github.dfa1.vortex.integration; + +import dev.vortex.api.Session; +import dev.vortex.api.VortexWriter; +import dev.vortex.arrow.ArrowAllocation; +import dev.vortex.jni.NativeLoader; +import io.github.dfa1.vortex.calcite.VortexCalcite; +import io.github.dfa1.vortex.calcite.VortexTable; +import org.apache.arrow.c.ArrowArray; +import org.apache.arrow.c.ArrowSchema; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// SQL over a vortex-jni-written file through the Calcite adapter: what the planner is told about +/// the table must match what Rust wrote. +class CalciteInteropIntegrationTest { + + private static final Session SESSION = Session.create(); + private static final BufferAllocator ALLOCATOR = ArrowAllocation.rootAllocator(); + // several batches, so the row count is a sum over chunks, not a single chunk's length + private static final int BATCHES = 3; + private static final int BATCH_ROWS = 1_000; + + static { + NativeLoader.loadJni(); + } + + @Test + void jniWriter_tableStatistic_isTheRowCountRustWrote(@TempDir Path tmp) throws Exception { + // Given + Path file = tmp.resolve("jni.vortex"); + writeWithJni(file); + VortexTable sut = new VortexTable(file); + + // When + Double result = sut.getStatistic().getRowCount(); + + // Then — and SQL agrees with the statistic + assertThat(result).isEqualTo((double) BATCHES * BATCH_ROWS); + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select count(*) from vtx.t")) { + assertThat(rs.next()).isTrue(); + assertThat(rs.getLong(1)).isEqualTo((long) BATCHES * BATCH_ROWS); + } + } + + private static void writeWithJni(Path file) throws Exception { + Schema schema = new Schema(List.of(Field.notNullable("id", new ArrowType.Int(64, true)))); + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build()) { + for (int batch = 0; batch < BATCHES; batch++) { + try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + BigIntVector ids = (BigIntVector) root.getVector("id"); + ids.allocateNew(BATCH_ROWS); + for (int i = 0; i < BATCH_ROWS; i++) { + ids.set(i, (long) batch * BATCH_ROWS + i); + } + root.setRowCount(BATCH_ROWS); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + } + } +} From 6b912bbfc535b13f6cbb741dcac5e6ae76a56f9a Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:46:42 +0200 Subject: [PATCH 3/7] fix(core): extension dtype metadata outlives the reader DType.Extension kept its metadata as a slice of the reader's file mapping, so a dtype held past VortexReader.close() threw "Already closed" on first use, e.g. reading a timestamp's unit or zone from a cached SQL schema. The constructor now copies off-heap metadata to the heap; the existing 64 KiB cap is checked first, so a crafted file cannot force a large copy. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../github/dfa1/vortex/core/model/DType.java | 7 +++++++ .../vortex/core/model/DTypeExtensionTest.java | 16 ++++++++++++++++ .../DateInteropIntegrationTest.java | 18 ++++++++++++++++++ 4 files changed, 42 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index b355cf7f9..74a280b68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** `ExtensionId` is a sealed `WellKnown`/`Custom` type like `EncodingId`, so third-party extension ids can be named, and `ExtensionId.parse` returns the id instead of an `Optional`. ([#530](https://github.com/dfa1/vortex-java/issues/530)) ### Fixed +- An extension dtype read from a file stays usable after its `VortexReader` is closed, instead of failing with "Already closed". ([#550](https://github.com/dfa1/vortex-java/pull/550)) - `vortex count`, `schema`, `stats` and `export` report a malformed file as an error instead of crashing with a stack trace. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - `vortex.date` columns in milliseconds, as Rust writes an Arrow `Date64`, decode to the right dates instead of failing or reading the count as days. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - An extension column with no metadata no longer writes a file Rust refuses to read. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java b/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java index f807b8d1d..fe0e1d588 100644 --- a/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java +++ b/core/src/main/java/io/github/dfa1/vortex/core/model/DType.java @@ -365,6 +365,13 @@ record Extension( throw new VortexException("extension metadata too large: " + metadata.byteSize() + " > " + MAX_METADATA_SIZE.bytes()); } + // A dtype outlives the reader that parsed it (callers keep reader.dtype()), but a slice + // of the file mapping fails once the reader closes: copy it onto the heap. Small, as + // the cap above bounds it. + if (metadata != null && metadata.isNative()) { + metadata = MemorySegment.ofArray(metadata.toArray(java.lang.foreign.ValueLayout.JAVA_BYTE)) + .asReadOnly(); + } } /// Value equality with content-based metadata comparison. diff --git a/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java b/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java index cd2ad1f7f..dcf51c9e5 100644 --- a/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java +++ b/core/src/test/java/io/github/dfa1/vortex/core/model/DTypeExtensionTest.java @@ -36,6 +36,22 @@ void metadataOverCap_rejected() { .hasMessageContaining("metadata too large"); } + @Test + void nativeMetadata_outlivesItsArena() { + // Given metadata sliced from off-heap memory, as a reader parses it out of the file mapping + DType.Extension result; + try (java.lang.foreign.Arena arena = java.lang.foreign.Arena.ofConfined()) { + MemorySegment meta = arena.allocate(2); + meta.set(java.lang.foreign.ValueLayout.JAVA_BYTE, 0, (byte) 3); + + // When + result = new DType.Extension("vortex.timestamp", DType.I64, meta, false); + } + + // Then — the dtype is still readable after the arena (the reader) is closed + assertThat(result.metadata().get(java.lang.foreign.ValueLayout.JAVA_BYTE, 0)).isEqualTo((byte) 3); + } + @Test void nullMetadata_accepted() { // Given / When / Then — date/uuid extensions carry no metadata; null must remain valid diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java index 5d22af281..8e471ea24 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/DateInteropIntegrationTest.java @@ -110,6 +110,24 @@ void rustWrites_dtypeRecordsTheUnit(DateUnit unit, @TempDir Path tmp) throws IOE } } + @ParameterizedTest + @EnumSource(DateUnit.class) + void rustWrites_dtypeOutlivesTheReader(DateUnit unit, @TempDir Path tmp) throws IOException { + // Given + Path file = tmp.resolve("dates.vortex"); + writeWithJni(file, unit); + DType.Extension result; + + // When — the dtype is kept past close, as a SQL schema caches it + try (VortexReader reader = VortexReader.open(file, ReadRegistry.loadAll())) { + result = (DType.Extension) ((DType.Struct) reader.dtype()).fieldTypes().get(1); + } + + // Then — the metadata was a slice of the closed file mapping, so this threw "Already closed" + TimeUnit expectedUnit = unit == DateUnit.DAY ? TimeUnit.Days : TimeUnit.Milliseconds; + assertThat(TimeUnit.fromTag(result.metadata().get(ValueLayout.JAVA_BYTE, 0))).isEqualTo(expectedUnit); + } + /// Every ninth row is null; the rest step 37 days from 1915, so a quarter of them precede the /// epoch, where flooring a millisecond count must not round toward zero. private static LocalDate expected(long row) { From 5c08e0a9033cc93e48e087c0b4f8479662580638 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:47:23 +0200 Subject: [PATCH 4/7] fix(calcite): map decimal, binary, temporal, uuid and nested columns toSqlType threw on any column that was not a primitive, string or bool, so a file with a single timestamp could not be queried at all, not even its other columns. Decimal maps to DECIMAL(p, s), binary to VARBINARY, vortex.date/time/uuid to DATE/TIME/UUID, vortex.timestamp to TIMESTAMP (the instant as a UTC wall clock, zoned or not: Avatica's JDBC getters shift TIMESTAMP WITH LOCAL TIME ZONE by the JVM zone), any other extension to its storage type. Nested types map to ANY so the rest of the table stays queryable; selecting one fails loudly. Cross-checked against a vortex-jni-written file with date, time, timestamp (with and without zone), decimal and binary columns. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../dfa1/vortex/calcite/VortexEnumerator.java | 26 +++- .../dfa1/vortex/calcite/VortexTable.java | 20 ++- .../vortex/calcite/ColumnTypesSqlTest.java | 140 ++++++++++++++++++ docs/reference.md | 15 ++ .../CalciteInteropIntegrationTest.java | 72 +++++++++ 6 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 74a280b68..c97a8334e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** `ExtensionId` is a sealed `WellKnown`/`Custom` type like `EncodingId`, so third-party extension ids can be named, and `ExtensionId.parse` returns the id instead of an `Optional`. ([#530](https://github.com/dfa1/vortex-java/issues/530)) ### Fixed +- The Calcite adapter reads decimal, binary, date, time, timestamp and uuid columns, and a list, struct, map, union or variant column no longer stops the other columns of its table from being queried. ([#550](https://github.com/dfa1/vortex-java/pull/550)) - An extension dtype read from a file stays usable after its `VortexReader` is closed, instead of failing with "Already closed". ([#550](https://github.com/dfa1/vortex-java/pull/550)) - `vortex count`, `schema`, `stats` and `export` report a malformed file as an error instead of crashing with a stack trace. ([#547](https://github.com/dfa1/vortex-java/pull/547)) - `vortex.date` columns in milliseconds, as Rust writes an Arrow `Date64`, decode to the right dates instead of failing or reading the count as days. ([#547](https://github.com/dfa1/vortex-java/pull/547)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java index f8436228f..cb42762bb 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java @@ -3,14 +3,17 @@ import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.ExtensionId; import io.github.dfa1.vortex.reader.Chunk; import io.github.dfa1.vortex.reader.RowFilter; import io.github.dfa1.vortex.reader.ScanIterator; import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; import io.github.dfa1.vortex.reader.compute.Compute; +import io.github.dfa1.vortex.reader.array.Array; import io.github.dfa1.vortex.reader.array.BoolArray; import io.github.dfa1.vortex.reader.array.ByteArray; +import io.github.dfa1.vortex.reader.array.DecimalArray; import io.github.dfa1.vortex.reader.array.DoubleArray; import io.github.dfa1.vortex.reader.array.Float16Array; import io.github.dfa1.vortex.reader.array.FloatArray; @@ -19,7 +22,12 @@ import io.github.dfa1.vortex.reader.array.MaskedArray; import io.github.dfa1.vortex.reader.array.ShortArray; import io.github.dfa1.vortex.reader.array.VarBinArray; +import io.github.dfa1.vortex.reader.extension.DateExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.TimeExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.TimestampExtensionDecoder; +import io.github.dfa1.vortex.reader.extension.UuidExtensionDecoder; +import org.apache.calcite.avatica.util.ByteString; import org.apache.calcite.linq4j.Enumerator; import java.io.IOException; @@ -168,8 +176,24 @@ private static Object value(Object array, DType type, long r) { case I8, U8 -> ((ByteArray) array).getInt(r); }; case DType.Utf8 _ -> ((VarBinArray) array).getString(r); + case DType.Binary _ -> new ByteString(((VarBinArray) array).getBytes(r)); case DType.Bool _ -> ((BoolArray) array).getBoolean(r); - default -> throw new IllegalStateException("unsupported column dtype: " + type); + case DType.Null _ -> null; + case DType.Decimal _ -> ((DecimalArray) array).getDecimal(r); + // Calcite's internal forms: DATE is days since the epoch (int), TIME milliseconds of the + // day (int), TIMESTAMP milliseconds since the epoch (long) + case DType.Extension ext -> switch (ExtensionId.parse(ext.extensionId())) { + case ExtensionId.WellKnown.VORTEX_DATE -> + Math.toIntExact(DateExtensionDecoder.INSTANCE.decode(ext, (Array) array, r).toEpochDay()); + case ExtensionId.WellKnown.VORTEX_TIME -> + (int) (TimeExtensionDecoder.INSTANCE.decode(ext, (Array) array, r).toNanoOfDay() / 1_000_000); + case ExtensionId.WellKnown.VORTEX_TIMESTAMP -> + TimestampExtensionDecoder.INSTANCE.instant(ext, (Array) array, r).toEpochMilli(); + case ExtensionId.WellKnown.VORTEX_UUID -> UuidExtensionDecoder.INSTANCE.decode((Array) array, r); + default -> value(array, ext.storageDType(), r); + }; + case DType.List _, DType.FixedSizeList _, DType.Struct _, DType.Map _, DType.Union _, + DType.Variant _ -> throw new VortexException("SQL cannot read a " + type + " column yet"); }; } 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 d350a2066..c5cf329e4 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 @@ -2,6 +2,7 @@ import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.ExtensionId; import io.github.dfa1.vortex.core.model.PType; import io.github.dfa1.vortex.reader.ArrayStats; import io.github.dfa1.vortex.reader.Chunk; @@ -754,6 +755,12 @@ private static int[] allColumns(int n) { /// `U32`->`BIGINT`); `U64` is the exception — no wider SQL integer exists, so it maps to /// `BIGINT` and [VortexEnumerator] guards the high-half values that cannot fit. /// + /// `vortex.timestamp` is a `TIMESTAMP` holding the stored value as a UTC wall clock, zone or + /// not: Avatica shifts `TIMESTAMP WITH LOCAL TIME ZONE` by the JVM's zone in its JDBC getters, + /// so a zoned instant would read back wrong. An extension with no SQL counterpart reads as its storage + /// type. Nested types (list, struct, map, union, variant) map to `ANY` so the rest of the table + /// stays queryable; selecting one fails in [VortexEnumerator]. + /// /// @param factory the Calcite type factory /// @param type the Vortex column dtype /// @return the nullable-adjusted SQL type @@ -769,8 +776,19 @@ private static RelDataType toSqlType(RelDataTypeFactory factory, DType type) { case I8 -> factory.createSqlType(SqlTypeName.TINYINT); }; case DType.Utf8 _ -> factory.createSqlType(SqlTypeName.VARCHAR); + case DType.Binary _ -> factory.createSqlType(SqlTypeName.VARBINARY); case DType.Bool _ -> factory.createSqlType(SqlTypeName.BOOLEAN); - default -> throw new IllegalStateException("unsupported column dtype: " + type); + case DType.Null _ -> factory.createSqlType(SqlTypeName.NULL); + case DType.Decimal d -> factory.createSqlType(SqlTypeName.DECIMAL, d.precision(), d.scale()); + case DType.Extension ext -> switch (ExtensionId.parse(ext.extensionId())) { + case ExtensionId.WellKnown.VORTEX_DATE -> factory.createSqlType(SqlTypeName.DATE); + case ExtensionId.WellKnown.VORTEX_TIME -> factory.createSqlType(SqlTypeName.TIME); + case ExtensionId.WellKnown.VORTEX_TIMESTAMP -> factory.createSqlType(SqlTypeName.TIMESTAMP); + case ExtensionId.WellKnown.VORTEX_UUID -> factory.createSqlType(SqlTypeName.UUID); + default -> toSqlType(factory, ext.storageDType()); + }; + case DType.List _, DType.FixedSizeList _, DType.Struct _, DType.Map _, DType.Union _, + DType.Variant _ -> factory.createSqlType(SqlTypeName.ANY); }; return factory.createTypeWithNullability(sql, type.nullable()); } diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java new file mode 100644 index 000000000..2c21b1dc6 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ColumnTypesSqlTest.java @@ -0,0 +1,140 @@ +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.core.model.TimeDtype; +import io.github.dfa1.vortex.core.model.TimeUnit; +import io.github.dfa1.vortex.core.model.TimestampDtype; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import io.github.dfa1.vortex.writer.encode.DateExtensionEncoder; +import io.github.dfa1.vortex.writer.encode.ListData; +import io.github.dfa1.vortex.writer.encode.UuidExtensionEncoder; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.math.BigDecimal; +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.SQLException; +import java.sql.Statement; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/// Column types beyond numbers, strings and booleans. Before, `getRowType` threw on the first +/// decimal, binary, temporal, uuid or nested column, so a file with a single timestamp could not be +/// queried at all, not even its other columns. +class ColumnTypesSqlTest { + + private static final LocalDate DATE = LocalDate.of(2024, 2, 29); + private static final LocalTime TIME = LocalTime.of(13, 45, 30, 123_000_000); + // pre-epoch on purpose: millisecond truncation must floor, not round toward zero + private static final Instant INSTANT = Instant.parse("1969-12-31T23:59:59.250Z"); + private static final UUID ID = UUID.fromString("123e4567-e89b-12d3-a456-426614174000"); + + @TempDir + static Path tmp; + private static Path file; + + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("types.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I32) + .field(ColumnName.of("amount"), DType.decimal(10, 2)) + .field(ColumnName.of("blob"), DType.BINARY) + .field(ColumnName.of("day"), DateExtensionEncoder.INSTANCE.dtype(false)) + .field(ColumnName.of("at"), TimeDtype.of(TimeUnit.Milliseconds, false)) + .field(ColumnName.of("wall"), TimestampDtype.of(TimeUnit.Microseconds, null, false)) + .field(ColumnName.of("instant"), TimestampDtype.of(TimeUnit.Microseconds, ZoneId.of("UTC"), false)) + .field(ColumnName.of("uuid"), UuidExtensionEncoder.INSTANCE.dtype(false)) + .field(ColumnName.of("tags"), new DType.List(DType.I64, false)) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of( + ColumnName.of("id"), new int[]{1}, + ColumnName.of("amount"), new BigDecimal[]{new BigDecimal("12.34")}, + ColumnName.of("blob"), new byte[][]{{1, 2, 3}}, + ColumnName.of("day"), List.of(DATE), + ColumnName.of("at"), List.of(TIME), + ColumnName.of("wall"), List.of(INSTANT), + ColumnName.of("instant"), List.of(INSTANT), + ColumnName.of("uuid"), List.of(ID), + ColumnName.of("tags"), new ListData(new long[]{7, 8}, new long[]{0, 2}, 1))); + } + } + + @Test + void everyScalarType_readsThroughSql() throws Exception { + // Given + String sql = "select amount, blob, `day`, at, wall, `instant`, `uuid` from vtx.t"; + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery(sql)) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getBigDecimal("amount")).isEqualByComparingTo("12.34"); + assertThat(result.getBytes("blob")).containsExactly(1, 2, 3); + assertThat(result.getDate("day").toLocalDate()).isEqualTo(DATE); + assertThat(result.getTime("at").toLocalTime()).isEqualTo(TIME.withNano(0)); + assertThat(result.getString("at")).isEqualTo("13:45:30"); + // TIMESTAMP is a wall clock: the stored value read as UTC, with no zone shift + assertThat(result.getTimestamp("wall").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + // a zoned timestamp too: the instant as a UTC wall clock, the same for every JVM zone + assertThat(result.getTimestamp("instant").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getObject("uuid")).isEqualTo(ID); + assertThat(result.next()).isFalse(); + } + } + + @Test + void nestedColumn_doesNotBlockTheOtherColumns() throws Exception { + // Given a table with a list column the adapter cannot read yet + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery("select id from vtx.t")) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getInt("id")).isEqualTo(1); + } + } + + @Test + void nestedColumn_failsLoudWhenSelected() throws Exception { + // Given + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement()) { + + // When / Then + assertThatThrownBy(() -> { + try (ResultSet rs = st.executeQuery("select tags from vtx.t")) { + rs.next(); + } + }).isInstanceOf(SQLException.class).hasStackTraceContaining("cannot read a List"); + } + } +} diff --git a/docs/reference.md b/docs/reference.md index 849bac9b0..f1f5d4836 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -611,6 +611,21 @@ preserved) and the **Babel** SQL parser, so a column named after a reserved word keywords that open a typed literal (`date`, `time`, `timestamp`, `interval`), which stay reserved even under Babel and need back-ticks: `` select `date` from vtx.ohlc ``. +Column types map as follows. Unsigned integers widen to the next signed type (`U64` stays `BIGINT` +and fails on values past 2^63); `F16` reads as `REAL`. + +| Vortex | SQL | +|---------------------------------------------|---------------------------------------------------| +| `bool`, `utf8`, `binary` | `BOOLEAN`, `VARCHAR`, `VARBINARY` | +| `decimal(p, s)` | `DECIMAL(p, s)` | +| `vortex.date`, `vortex.time`, `vortex.uuid` | `DATE`, `TIME`, `UUID` | +| `vortex.timestamp`, with or without a zone | `TIMESTAMP`, the stored instant as a UTC wall clock | +| other extensions | their storage type | +| list, struct, map, union, variant | `ANY`; selecting the column fails | + +`DATE`, `TIME` and `TIMESTAMP` values carry millisecond precision. A zoned timestamp is not +`TIMESTAMP WITH LOCAL TIME ZONE` because Avatica's JDBC getters shift that type by the JVM's zone. + Whole-table aggregates (`min`/`max`/`sum`) are answered from zone-map statistics via `VortexAggregatePushDownRule` where possible — no data segment is decoded. The fold needs one zone per chunk and a per-zone sum, i.e. the legacy `vortex.stats` zone map (writes targeting an edition diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index 0bf6a8029..ee7f9ef34 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -11,17 +11,31 @@ import org.apache.arrow.c.Data; import org.apache.arrow.memory.BufferAllocator; import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.DateDayVector; +import org.apache.arrow.vector.DecimalVector; +import org.apache.arrow.vector.TimeMilliVector; +import org.apache.arrow.vector.TimeStampMicroTZVector; +import org.apache.arrow.vector.TimeStampMicroVector; +import org.apache.arrow.vector.VarBinaryVector; import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.types.DateUnit; +import org.apache.arrow.vector.types.TimeUnit; import org.apache.arrow.vector.types.pojo.ArrowType; import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.Schema; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; +import java.math.BigDecimal; import java.nio.file.Path; import java.sql.Connection; import java.sql.ResultSet; import java.sql.Statement; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.time.ZoneOffset; import java.util.List; import java.util.Map; @@ -36,6 +50,10 @@ class CalciteInteropIntegrationTest { // several batches, so the row count is a sum over chunks, not a single chunk's length private static final int BATCHES = 3; private static final int BATCH_ROWS = 1_000; + private static final LocalDate DATE = LocalDate.of(2024, 2, 29); + private static final LocalTime TIME = LocalTime.of(13, 45, 30); + // pre-epoch on purpose: millisecond truncation must floor, not round toward zero + private static final Instant INSTANT = Instant.parse("1969-12-31T23:59:59.250Z"); static { NativeLoader.loadJni(); @@ -61,6 +79,60 @@ void jniWriter_tableStatistic_isTheRowCountRustWrote(@TempDir Path tmp) throws E } } + @Test + void jniWriter_temporalDecimalAndBinaryColumns_readThroughSql(@TempDir Path tmp) throws Exception { + // Given — a file with a timestamp could not be queried at all: the row type threw + Path file = tmp.resolve("jni_types.vortex"); + writeTypesWithJni(file); + String sql = "select `day`, at, wall, `instant`, amount, blob from vtx.t"; + + // When + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet result = st.executeQuery(sql)) { + + // Then + assertThat(result.next()).isTrue(); + assertThat(result.getDate("day").toLocalDate()).isEqualTo(DATE); + assertThat(result.getTime("at").toLocalTime()).isEqualTo(TIME); + assertThat(result.getTimestamp("wall").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getTimestamp("instant").toLocalDateTime()) + .isEqualTo(LocalDateTime.ofInstant(INSTANT, ZoneOffset.UTC)); + assertThat(result.getBigDecimal("amount")).isEqualByComparingTo("12.34"); + assertThat(result.getBytes("blob")).containsExactly(1, 2, 3); + assertThat(result.next()).isFalse(); + } + } + + private static void writeTypesWithJni(Path file) throws Exception { + Schema schema = new Schema(List.of( + Field.notNullable("day", new ArrowType.Date(DateUnit.DAY)), + Field.notNullable("at", new ArrowType.Time(TimeUnit.MILLISECOND, 32)), + Field.notNullable("wall", new ArrowType.Timestamp(TimeUnit.MICROSECOND, null)), + Field.notNullable("instant", new ArrowType.Timestamp(TimeUnit.MICROSECOND, "UTC")), + Field.notNullable("amount", new ArrowType.Decimal(10, 2, 128)), + Field.notNullable("blob", new ArrowType.Binary()))); + long micros = INSTANT.getEpochSecond() * 1_000_000 + INSTANT.getNano() / 1_000; + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + root.allocateNew(); + ((DateDayVector) root.getVector("day")).setSafe(0, Math.toIntExact(DATE.toEpochDay())); + ((TimeMilliVector) root.getVector("at")).setSafe(0, (int) (TIME.toNanoOfDay() / 1_000_000)); + ((TimeStampMicroVector) root.getVector("wall")).setSafe(0, micros); + ((TimeStampMicroTZVector) root.getVector("instant")).setSafe(0, micros); + ((DecimalVector) root.getVector("amount")).setSafe(0, new BigDecimal("12.34")); + ((VarBinaryVector) root.getVector("blob")).setSafe(0, new byte[]{1, 2, 3}); + root.setRowCount(1); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + private static void writeWithJni(Path file) throws Exception { Schema schema = new Schema(List.of(Field.notNullable("id", new ArrowType.Int(64, true)))); try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, From f8faa725567cde27c22ce0b61620838ab8f0f6fe Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:55:43 +0200 Subject: [PATCH 5/7] feat(calcite): fold filtered aggregates over Rust's vortex.zoned zones The filtered fold required one zone per chunk, which only the legacy vortex.stats zone map has. Rust's vortex.zoned (our default since #447, and every Rust-written file) uses fixed 8192-row zones while a chunk spans many of them, so every filtered COUNT/MIN/MAX on such a file fell back to a full scan. The fold now works per zone: IN zones fold from statistics, OUT zones are skipped, and a boundary zone folds only its own rows through the new zero-copy Chunk.slice(offset, length), even when it straddles a chunk edge. Files without a placeable zone map fall back to per-chunk stats as before. SUM still scans on zoned files: Rust records no zone sum. Also fixes a latent wrong MIN/MAX: an IN zone that has values but no min/max stat (zoned string columns keep only truncated bounds) was silently skipped by the fold; it now makes the extreme unknown, so the rule falls back to the scan. Cross-checked on a vortex-jni-written file: the plan has no scan and COUNT/MIN/MAX match the data. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 2 + .../dfa1/vortex/calcite/VortexTable.java | 257 +++++++++++++----- .../dfa1/vortex/calcite/LegacyZoneMaps.java | 4 +- .../calcite/ZonedAggregateFoldTest.java | 132 +++++++++ docs/reference.md | 13 +- .../CalciteInteropIntegrationTest.java | 43 ++- .../io/github/dfa1/vortex/reader/Chunk.java | 29 ++ .../dfa1/vortex/reader/ChunkSliceTest.java | 75 +++++ 8 files changed, 475 insertions(+), 80 deletions(-) create mode 100644 calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java create mode 100644 reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index c97a8334e..6cfbde46f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,12 +8,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] ### 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)) - `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)) ### Changed +- 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)) - **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)) 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 c5cf329e4..052ad7c56 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 @@ -1,5 +1,6 @@ package io.github.dfa1.vortex.calcite; +import io.github.dfa1.vortex.core.error.VortexException; import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.core.model.ExtensionId; @@ -10,6 +11,7 @@ import io.github.dfa1.vortex.reader.ScanIterator; import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.reader.Zone; import io.github.dfa1.vortex.reader.compute.Compare; import io.github.dfa1.vortex.reader.compute.Compute; import io.github.dfa1.vortex.reader.compute.FilteredAggregate; @@ -223,14 +225,18 @@ public record FilteredFold(Number sum, Object min, Object max, Long nullCount, l /// `aggColumn` is the column whose values are reduced over the selected rows (`null` for /// `COUNT(*)`, which needs only the selected row count). The result is empty — abandoning to a /// full scan, which is always correctness-safe — when the fold cannot be trusted: a `U64` - /// column involved (whose stat order [#isU64] cannot classify — see the guard below), a - /// zone-map zone count that does not align 1:1 with the chunks, or a boundary zone whose mask - /// cannot be built. + /// column involved (whose stat order [#isU64] cannot classify — see the guard below), columns + /// whose zones lie on different rows, zones that do not tile the chunks' rows, or a boundary + /// zone whose columns cannot be sliced. + /// + /// Zones need not line up with chunks: Rust's `vortex.zoned` writes fixed 8192-row zones while a + /// chunk often spans many of them, so a boundary zone is folded from just its rows of the chunk + /// (or two chunks) holding it. /// /// The fold also abandons for performance, not correctness, when there is at least one boundary /// zone but no fully-selected (IN) zone: the only win the fold has over a scan is folding IN zones /// from statistics without decoding them, so with zero IN zones it would decode exactly the chunks - /// a scan decodes and only add the stats-table classify and mask-building overhead. The + /// a scan decodes and only add the stats-table classify overhead. The /// zone-map-pruned scan is the cheaper answer. An all-OUT selection (no IN, no boundary) is not /// this case — it folds to a correct empty aggregate with no decode and is kept. /// @@ -264,30 +270,35 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum return Optional.empty(); } } - long[] rowCounts = scan.chunkRowCounts(); - int zones = rowCounts.length; + long[] chunkRows = scan.chunkRowCounts(); + // Every column must place its zones on the same rows, so one zone index classifies + // the whole filter. Rust's vortex.zoned uses fixed zones independent of the chunks; + // the legacy vortex.stats and the no-zone-map fallback have one zone per chunk. + List geometry = null; java.util.Map> zoneStats = new java.util.HashMap<>(); for (ColumnName column : columns) { - List perZone = scan.columnZoneStats(column); - if (perZone.size() != zones) { - return Optional.empty(); // zone-map zones don't align 1:1 with chunks — cannot classify + List zones = zonesOf(scan, column, chunkRows); + if (zones == null || geometry != null && !sameRows(geometry, zones)) { + return Optional.empty(); } - zoneStats.put(column, perZone); + geometry = zones; + zoneStats.put(column, zones.stream().map(Zone::stats).toList()); + } + if (!tiles(geometry, chunkRows)) { + return Optional.empty(); // zones and chunks disagree on the rows } // Classify every zone first, so the IN count is known before any boundary decode. The - // fold's only win over a scan is folding IN zones from statistics with no decode; a - // BOUNDARY zone is decoded exactly as a scan would decode it. So when there is at least - // one BOUNDARY zone but no IN zone, the fold decodes the same chunks a scan does and adds - // the stats-table classify + mask-building overhead on top — a net loss. Abandon to the - // (zone-map-pruned) scan in that case. The all-OUT case (no IN, no BOUNDARY) is NOT this - // case: it folds to a correct empty/zero aggregate with no decode at all, so it is kept. - Match[] matches = new Match[zones]; + // fold's only win over a scan is folding IN zones from statistics with no decode, so + // when there is at least one BOUNDARY zone but no IN zone, abandon to the + // (zone-map-pruned) scan. The all-OUT case (no IN, no BOUNDARY) folds to a correct + // empty/zero aggregate with no decode at all, so it is kept. + Match[] matches = new Match[geometry.size()]; int inZones = 0; int boundaryZones = 0; - for (int zone = 0; zone < zones; zone++) { - Match match = classify(filter, zone, zoneStats, rowCounts[zone], struct); - matches[zone] = match; + for (int z = 0; z < geometry.size(); z++) { + Match match = classify(filter, z, zoneStats, geometry.get(z).rowCount(), struct); + matches[z] = match; if (match == Match.IN) { inZones++; } else if (match == Match.BOUNDARY) { @@ -300,27 +311,32 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum List decodeColumns = new ArrayList<>(columns); Fold fold = new Fold(); - for (int zone = 0; zone < zones; zone++) { - Match match = matches[zone]; - if (match == Match.OUT) { - continue; - } - if (match == Match.BOUNDARY) { - // Tier 2: a partially-selected zone. Decode this chunk (and the filter columns), - // build a row mask from the filter, and reduce the masked rows. The chunk index - // equals the zone index — the perZone.size() == zones guard above proved the - // zone-map rows align 1:1 with the chunks. - foldBoundaryZone(reader, zone, decodeColumns, filter, aggColumn, fold); - continue; - } - // Tier 1 (Match.IN): every row matches — fold the whole zone from its statistics. - fold.keptRows += rowCounts[zone]; - if (aggColumn != null) { - ArrayStats stats = zoneStats.get(aggColumn).get(zone); - fold.addSum(stats.sum()); - fold.addMin(stats.min()); - fold.addMax(stats.max()); - fold.addNulls(stats.nullCount()); + try (BoundaryChunks chunks = new BoundaryChunks(reader, decodeColumns, chunkRows)) { + for (int z = 0; z < geometry.size(); z++) { + Zone zone = geometry.get(z); + if (matches[z] == Match.OUT) { + continue; + } + if (matches[z] == Match.BOUNDARY) { + // Tier 2: fold the rows the filter selects, from the zone's rows of the + // chunks holding it — a zone may straddle a chunk edge. + if (!chunks.fold(zone, filter, aggColumn, fold)) { + return Optional.empty(); + } + continue; + } + // Tier 1 (Match.IN): every row matches — fold the whole zone from its statistics. + fold.keptRows += zone.rowCount(); + if (aggColumn != null) { + ArrayStats stats = zoneStats.get(aggColumn).get(z); + // a zone with values but no min/max stat (a string zone holds only bounds) + // leaves the extreme unknown, rather than silently ignored + boolean hasValues = stats.nullCount() == null || stats.nullCount() < zone.rowCount(); + fold.addSum(stats.sum()); + fold.addMin(stats.min(), hasValues); + fold.addMax(stats.max(), hasValues); + fold.addNulls(stats.nullCount()); + } } } return Optional.of(fold.result()); @@ -329,30 +345,79 @@ public Optional filteredFold(RowFilter filter, ColumnName aggColum } } - /// Decodes one boundary chunk and folds the aggregate over the rows the filter selects in it, - /// contributing to the running [Fold]. The chunk is released per zone via try-with-resources, so - /// nothing outlives the reduction. - /// - /// The whole filter — an n-ary `AND` of column-bound [Predicate] leaves — and the aggregate fold - /// run in one fused pass over the chunk's rows via - /// [Compute#filteredAggregate(Chunk, RowFilter, ColumnName)], with no intermediate selection bitmap. - /// `SUM` is guarded: the fused kernel reports a `null` sum for a non-numeric aggregate column, - /// which disables the sum (mirroring a zone that records no usable sum) rather than failing the - /// whole fold — a `MIN`/`MAX` over the same column still folds. The aggregate column's null count - /// among the selected rows is the selected count minus the non-null count. - private static void foldBoundaryZone(VortexReader reader, int zone, List columns, RowFilter filter, ColumnName aggColumn, Fold fold) { - try (Chunk chunk = reader.decodeChunk(zone, columns)) { - FilteredAggregate aggregate = Compute.filteredAggregate(chunk, filter, aggColumn); - fold.keptRows += aggregate.selectedRows(); - if (aggColumn == null) { + /// Decodes the chunks that hold boundary zones, each once, in row order, and folds a zone's rows + /// through a [Chunk#slice(long, long)] view. Zones arrive in row order, so only the chunk under + /// the current zone is kept open. + private static final class BoundaryChunks implements AutoCloseable { + private final VortexReader reader; + private final List columns; + private final long[] chunkRows; + private int index = -1; + private long chunkStart; + private Chunk chunk; + + BoundaryChunks(VortexReader reader, List columns, long[] chunkRows) { + this.reader = reader; + this.columns = columns; + this.chunkRows = chunkRows; + } + + /// Folds the rows `filter` selects in `zone`, in one fused pass per chunk part via + /// [Compute#filteredAggregate(Chunk, RowFilter, ColumnName)]. `SUM` is guarded: the kernel + /// reports a `null` sum for a non-numeric aggregate column, which disables only the sum. + /// + /// @return `false` when a column cannot be sliced, so the caller abandons to the scan + boolean fold(Zone zone, RowFilter filter, ColumnName aggColumn, Fold fold) { + long row = zone.firstRow(); + long end = row + zone.rowCount(); + while (row < end) { + moveTo(row); + long offset = row - chunkStart; + long length = Math.min(end, chunkStart + chunkRows[index]) - row; + FilteredAggregate aggregate; + try (Chunk part = chunk.slice(offset, length)) { + aggregate = Compute.filteredAggregate(part, filter, aggColumn); + } catch (VortexException e) { + // ponytail: some array types (F16, variant) have no slice yet; the scan answers + return false; + } + fold.keptRows += aggregate.selectedRows(); + if (aggColumn != null) { + // the kernel's min/max are null only when no selected row has a value + fold.addSum(aggregate.sum()); + fold.addMin(aggregate.min(), false); + fold.addMax(aggregate.max(), false); + fold.addNulls(aggregate.selectedRows() - aggregate.aggNonNullCount()); + } + row += length; + } + return true; + } + + /// Opens the chunk holding `row`; `tiles` proved every row lies in some chunk. + private void moveTo(long row) { + if (chunk != null && row < chunkStart + chunkRows[index]) { return; } - // A null sum marks a non-numeric aggregate column — unanswerable, like a missing zone sum; - // addSum(null) disables only the sum, leaving MIN/MAX to fold. - fold.addSum(aggregate.sum()); - fold.addMin(aggregate.min()); - fold.addMax(aggregate.max()); - fold.addNulls(aggregate.selectedRows() - aggregate.aggNonNullCount()); + if (chunk != null) { + chunk.close(); + chunk = null; + } + if (index < 0) { + index = 0; + } + while (row >= chunkStart + chunkRows[index]) { + chunkStart += chunkRows[index]; + index++; + } + chunk = reader.decodeChunk(index, columns); + } + + @Override + public void close() { + if (chunk != null) { + chunk.close(); + } } } @@ -367,6 +432,7 @@ private static final class Fold { private boolean sumUsable = true; private Object min; private Object max; + private boolean extremesKnown = true; private long nullCount; private boolean nullCountKnown = true; private long keptRows; @@ -388,11 +454,18 @@ private void addSum(Object contribution) { } } - private void addMin(Object candidate) { + /// Adds a min candidate; a `null` one from rows that do hold values makes the min unknown. + private void addMin(Object candidate, boolean hasValues) { + if (candidate == null && hasValues) { + extremesKnown = false; + } min = pickExtreme(min, candidate, true); } - private void addMax(Object candidate) { + private void addMax(Object candidate, boolean hasValues) { + if (candidate == null && hasValues) { + extremesKnown = false; + } max = pickExtreme(max, candidate, false); } @@ -418,7 +491,10 @@ private FilteredFold result() { foldedSum = longSum; } Long foldedNulls = nullCountKnown ? nullCount : null; - return new FilteredFold(foldedSum, min, max, foldedNulls, keptRows); + // an unknown extreme is reported as absent, which the rule cannot mistake for an empty + // selection while rows were kept + return new FilteredFold(foldedSum, extremesKnown ? min : null, extremesKnown ? max : null, + foldedNulls, keptRows); } } @@ -439,6 +515,55 @@ public Optional translatePushedFilters(List filters) { return RexFilterTranslator.translateStrict(filters, names, types); } + /// A column's zones: its zone map when the zones can be placed on rows, else one zone per chunk + /// from the chunks' own statistics (the legacy `vortex.stats` table with batches of mixed + /// length, or no zone map at all). `null` when neither lines up with the chunks. + private static List zonesOf(ScanIterator scan, ColumnName column, long[] chunkRows) { + List zones = scan.columnZones(column); + if (!zones.isEmpty()) { + return zones; + } + List perChunk = scan.columnZoneStats(column); + if (perChunk.size() != chunkRows.length) { + return null; + } + List out = new ArrayList<>(chunkRows.length); + long first = 0; + for (int c = 0; c < chunkRows.length; c++) { + out.add(new Zone(first, chunkRows[c], perChunk.get(c))); + first += chunkRows[c]; + } + return out; + } + + private static boolean sameRows(List a, List b) { + if (a.size() != b.size()) { + return false; + } + for (int i = 0; i < a.size(); i++) { + if (a.get(i).firstRow() != b.get(i).firstRow() || a.get(i).rowCount() != b.get(i).rowCount()) { + return false; + } + } + return true; + } + + /// Whether `zones` cover the chunks' rows exactly, in order and with no gap or overlap. + private static boolean tiles(List zones, long[] chunkRows) { + long next = 0; + for (Zone zone : zones) { + if (zone.firstRow() != next) { + return false; + } + next += zone.rowCount(); + } + long total = 0; + for (long rows : chunkRows) { + total += rows; + } + return next == total; + } + /// How a zone relates to a filter: every row matches ([Match#IN]), no row matches /// ([Match#OUT]), or the zone straddles the predicate ([Match#BOUNDARY]) and can be folded only /// by decoding it. diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java index d28d9a320..be998f7f9 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/LegacyZoneMaps.java @@ -5,8 +5,8 @@ /// Editions targeting `core2025.10.0`, the last core edition before Rust's `vortex.zoned`: the /// writer then emits the legacy `vortex.stats` zone map, one zone per `writeChunk` batch with a -/// per-zone sum — the shape the zone fold folds. Default writes use 8192-row `vortex.zoned` zones -/// without sums, which the fold cannot use yet, so those aggregates fall back to a scan. +/// per-zone sum. Default writes use 8192-row `vortex.zoned` zones without sums, so a `SUM` over +/// them falls back to a scan. final class LegacyZoneMaps { static final WriteOptions OPTIONS = WriteOptions.defaults().withEdition(Editions.CORE_2025_10_0); diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java new file mode 100644 index 000000000..3e4bfd047 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/ZonedAggregateFoldTest.java @@ -0,0 +1,132 @@ +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 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.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// Filtered aggregates over Rust's `vortex.zoned` zone map, the default since #447: fixed 8192-row +/// zones that do not line up with chunks. The fold required one zone per chunk, so on these files +/// every filtered aggregate fell back to a scan. +class ZonedAggregateFoldTest { + + private static final int ZONE = 8_192; + // batches that straddle zone edges, so zones and chunks disagree + private static final int BATCH = 5_000; + private static final int ROWS = 12 * ZONE; + + @TempDir + static Path tmp; + private static Path file; + + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("zoned.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("val"), DType.I64) + .field(ColumnName.of("s"), DType.UTF8) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + for (int first = 0; first < ROWS; first += BATCH) { + int n = Math.min(BATCH, ROWS - first); + long[] id = new long[n]; + long[] val = new long[n]; + String[] s = new String[n]; + for (int i = 0; i < n; i++) { + id[i] = first + i; + val[i] = 3L * (first + i); + s[i] = String.format("k%06d", first + i); + } + writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("val"), val, ColumnName.of("s"), s)); + } + } + } + + @Test + void countMinMax_foldFromZonesAndBoundaryChunks() throws Exception { + // Given a range that covers several whole zones and cuts two of them + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select count(*) c, count(val) cv, min(val) mn, max(val) mx from vtx.t where id between " + + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String result = explain(conn, sql); + + // Then — answered without a scan, and the values match the data + assertThat(result).containsIgnoringCase("Values").doesNotContain("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong("c")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("cv")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("mn")).isEqualTo(3 * lo); + assertThat(rs.getLong("mx")).isEqualTo(3 * hi); + } + } + } + + @Test + void sum_hasNoZoneStatistic_soItScans() throws Exception { + // Given — Rust's zoned aggregate set records no sum, so a whole zone cannot be folded + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select sum(val) from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String result = explain(conn, sql); + + // Then + assertThat(result).contains("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong(1)).isEqualTo(3 * (hi * (hi + 1) / 2 - (lo - 1) * lo / 2)); + } + } + } + + @Test + void minOfAString_withoutAnExactZoneStatistic_isNotFoldedFromTheBoundaryAlone() throws Exception { + // Given — a zoned string column records only truncated bounds, so whole zones carry no min; + // folding only the boundary chunks would answer the min of the cut zone, not of the range + long lo = ZONE + 100; + long hi = 9L * ZONE + 300; + String sql = "select min(s), max(s) from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + // When + ResultSet result = st.executeQuery(sql)) { + + // Then + result.next(); + assertThat(result.getString(1)).isEqualTo(String.format("k%06d", lo)); + assertThat(result.getString(2)).isEqualTo(String.format("k%06d", hi)); + } + } + + private static String explain(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 f1f5d4836..efacf3fe5 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -269,6 +269,7 @@ never desync. | `columns()` | `SequencedMap`, schema order, unmodifiable | | ` column(ColumnName name)` | Typed column lookup; throws `VortexException` if absent | | `as(ColumnName name, Class domainType)` | Extension column → typed `List`, via the registered `ExtensionDecoder` | +| `slice(long offset, long length)` | Zero-copy view of a row range; valid while this chunk is open, closing it does nothing | | `isClosed()` | Whether `close()` has run | | `close()` | Releases the chunk's arena. Idempotent. | @@ -626,11 +627,13 @@ and fails on values past 2^63); `F16` reads as `REAL`. `DATE`, `TIME` and `TIMESTAMP` values carry millisecond precision. A zoned timestamp is not `TIMESTAMP WITH LOCAL TIME ZONE` because Avatica's JDBC getters shift that type by the JVM's zone. -Whole-table aggregates (`min`/`max`/`sum`) are answered from zone-map statistics via -`VortexAggregatePushDownRule` where possible — no data segment is decoded. The fold needs one zone -per chunk and a per-zone sum, i.e. the legacy `vortex.stats` zone map (writes targeting an edition -before `core2026.08.0`); files with Rust's `vortex.zoned` — the default since #447, and every -Rust-written file — answer the same aggregates through a scan. `connect` is the +`count`/`min`/`max`/`sum` (and `avg`, reduced to `sum`/`count`) are answered from zone-map +statistics via `VortexAggregatePushDownRule` where possible, with or without a `WHERE` on +column-vs-literal comparisons: zones the filter selects entirely fold from their statistics, and only +the rows of zones it cuts are decoded. `sum` needs a per-zone sum, which only the legacy +`vortex.stats` zone map records (writes targeting an edition before `core2026.08.0`); with Rust's +`vortex.zoned` — the default since #447, and every Rust-written file — `sum` is answered by a scan. +`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 by hand instead of going through `connect`. diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index ee7f9ef34..8994e2155 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -63,7 +63,7 @@ class CalciteInteropIntegrationTest { void jniWriter_tableStatistic_isTheRowCountRustWrote(@TempDir Path tmp) throws Exception { // Given Path file = tmp.resolve("jni.vortex"); - writeWithJni(file); + writeIdsWithJni(file, BATCHES, BATCH_ROWS); VortexTable sut = new VortexTable(file); // When @@ -79,6 +79,35 @@ void jniWriter_tableStatistic_isTheRowCountRustWrote(@TempDir Path tmp) throws E } } + @Test + void jniWriter_filteredCountMinMax_foldFromRustZones(@TempDir Path tmp) throws Exception { + // Given — Rust writes fixed 8192-row vortex.zoned zones over chunks of its own sizing; the + // fold needed one zone per chunk, so every filtered aggregate on such a file scanned + Path file = tmp.resolve("jni_zoned.vortex"); + writeIdsWithJni(file, 8, 25_000); + long lo = 10_000; + long hi = 170_000; + String sql = "select count(*) c, min(id) mn, max(id) mx from vtx.t where id between " + lo + " and " + hi; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + String plan; + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery("explain plan for " + sql)) { + rs.next(); + plan = rs.getString(1); + } + + // Then — answered from zones and boundary rows, and equal to the data + assertThat(plan).containsIgnoringCase("Values").doesNotContain("TableScan"); + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery(sql)) { + rs.next(); + assertThat(rs.getLong("c")).isEqualTo(hi - lo + 1); + assertThat(rs.getLong("mn")).isEqualTo(lo); + assertThat(rs.getLong("mx")).isEqualTo(hi); + } + } + } + @Test void jniWriter_temporalDecimalAndBinaryColumns_readThroughSql(@TempDir Path tmp) throws Exception { // Given — a file with a timestamp could not be queried at all: the row type threw @@ -133,18 +162,18 @@ private static void writeTypesWithJni(Path file) throws Exception { } } - private static void writeWithJni(Path file) throws Exception { + private static void writeIdsWithJni(Path file, int batches, int batchRows) throws Exception { Schema schema = new Schema(List.of(Field.notNullable("id", new ArrowType.Int(64, true)))); try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, ALLOCATOR).build()) { - for (int batch = 0; batch < BATCHES; batch++) { + for (int batch = 0; batch < batches; batch++) { try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { BigIntVector ids = (BigIntVector) root.getVector("id"); - ids.allocateNew(BATCH_ROWS); - for (int i = 0; i < BATCH_ROWS; i++) { - ids.set(i, (long) batch * BATCH_ROWS + i); + ids.allocateNew(batchRows); + for (int i = 0; i < batchRows; i++) { + ids.set(i, (long) batch * batchRows + i); } - root.setRowCount(BATCH_ROWS); + root.setRowCount(batchRows); try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java index 0470446ba..253c0bd6c 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/Chunk.java @@ -8,6 +8,8 @@ import io.github.dfa1.vortex.reader.extension.ExtensionDecoder; import java.lang.foreign.Arena; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; import java.util.Objects; import java.util.SequencedMap; @@ -44,15 +46,39 @@ public final class Chunk implements AutoCloseable { private final Arena arena; private final ReadRegistry registry; private final Consumer onClose; + // a slice shares its parent's arena, which only the parent releases + private final boolean ownsArena; private boolean closed; Chunk(long rowCount, SequencedMap columns, Arena arena, ReadRegistry registry, Consumer onClose) { + this(rowCount, columns, arena, registry, onClose, true); + } + + private Chunk(long rowCount, SequencedMap columns, + Arena arena, ReadRegistry registry, Consumer onClose, boolean ownsArena) { this.rowCount = rowCount; this.columns = Objects.requireNonNull(columns); this.arena = Objects.requireNonNull(arena); this.registry = Objects.requireNonNull(registry); this.onClose = Objects.requireNonNull(onClose); + this.ownsArena = ownsArena; + } + + /// A view of rows `[offset, offset + length)` of every column, sharing this chunk's memory + /// with no copy. The view is valid while this chunk is open; closing the view does nothing. + /// + /// @param offset first row of the view + /// @param length number of rows in the view + /// @return the row-range view + /// @throws IndexOutOfBoundsException if the range is not within `[0, rowCount())` + /// @throws VortexException if a column's array type cannot be sliced + public Chunk slice(long offset, long length) { + Objects.checkFromIndexSize(offset, length, rowCount); + SequencedMap sliced = new LinkedHashMap<>(); + columns.forEach((name, column) -> sliced.put(name, new Column( + ScanIterator.sliceArray(column.array(), offset, length, column.dtype()), column.dtype()))); + return new Chunk(length, Collections.unmodifiableSequencedMap(sliced), arena, registry, _ -> { }, false); } /// One decoded column: its zero-copy [Array] view and the [DType] it was declared with in the @@ -155,6 +181,9 @@ public void close() { return; } closed = true; + if (!ownsArena) { + return; + } try { onClose.accept(this); } finally { diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java new file mode 100644 index 000000000..eef5ccde0 --- /dev/null +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/ChunkSliceTest.java @@ -0,0 +1,75 @@ +package io.github.dfa1.vortex.reader; + +import io.github.dfa1.vortex.core.io.VortexFormat; +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.reader.array.LongArray; +import io.github.dfa1.vortex.reader.array.MaterializedLongArray; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.lang.foreign.Arena; +import java.lang.foreign.MemorySegment; +import java.util.LinkedHashMap; +import java.util.SequencedMap; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ChunkSliceTest { + + private static final ColumnName ID = ColumnName.of("id"); + private static final Arena ARENA = Arena.ofAuto(); + + @Test + void slice_viewsTheRowRange() { + // Given + Chunk sut = chunk(10); + + // When + Chunk result = sut.slice(3, 4); + + // Then + assertThat(result.rowCount()).isEqualTo(4); + LongArray ids = result.column(ID); + assertThat(ids.length()).isEqualTo(4); + assertThat(ids.getLong(0)).isEqualTo(3); + assertThat(ids.getLong(3)).isEqualTo(6); + } + + @Test + void closingTheSlice_leavesTheParentOpen() { + // Given + Chunk sut = chunk(10); + Chunk slice = sut.slice(0, 5); + + // When + slice.close(); + + // Then — the shared arena belongs to the parent + assertThat(sut.isClosed()).isFalse(); + assertThat(((LongArray) sut.column(ID)).getLong(9)).isEqualTo(9); + } + + @ParameterizedTest + @CsvSource({"-1, 1", "0, 11", "10, 1", "5, -1"}) + void slice_outsideTheChunk_throws(long offset, long length) { + // Given + Chunk sut = chunk(10); + + // When / Then + assertThatThrownBy(() -> sut.slice(offset, length)).isInstanceOf(IndexOutOfBoundsException.class); + } + + private static Chunk chunk(int rows) { + MemorySegment segment = ARENA.allocate(rows * 8L, 8); + for (int i = 0; i < rows; i++) { + segment.setAtIndex(VortexFormat.LE_LONG, i, i); + } + SequencedMap columns = new LinkedHashMap<>(); + columns.put(ID, new Chunk.Column(new MaterializedLongArray(DType.I64, rows, segment), DType.I64)); + return new Chunk(rows, columns, ARENA, ReadRegistry.loadAll(), _ -> { }); + } +} From 783d7a660ec3a68184d0dec31e2b0db356c44ce1 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sat, 17 Oct 2026 22:58:53 +0200 Subject: [PATCH 6/7] feat(calcite): push IN lists, flipped comparisons and IS NULL into the scan Three predicate shapes never reached the scan, so they pruned no chunk and Calcite re-checked every row: - an IN list: Calcite expands SEARCH to an OR of equalities, and OR was not translated. A single-column OR now becomes one column predicate (Predicate.Or); an OR over several columns, or with any untranslatable disjunct, stays with Calcite in both lenient and strict walks, since dropping a disjunct would lose rows. - a literal on the left (5 < x): the comparison is reversed. - IS NULL / IS NOT NULL: translated on the lenient pruning path too, so chunks with a zero null count are skipped. The aggregate fold now classifies single-column AND/OR predicates from zone statistics instead of treating them as boundary zones. Cross-checked on a vortex-jni-written file with a nullable column. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../vortex/calcite/RexFilterTranslator.java | 88 +++++++++--- .../dfa1/vortex/calcite/VortexTable.java | 25 +++- .../vortex/calcite/FilterTranslationTest.java | 132 ++++++++++++++++++ docs/reference.md | 7 +- .../CalciteInteropIntegrationTest.java | 46 +++++- 6 files changed, 271 insertions(+), 28 deletions(-) create mode 100644 calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cfbde46f..b9f741296 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,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 +- 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)) - **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)) diff --git a/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java index 454f52f2f..ff83817f5 100644 --- a/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java +++ b/calcite/src/main/java/io/github/dfa1/vortex/calcite/RexFilterTranslator.java @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.core.model.DType; import io.github.dfa1.vortex.reader.RowFilter; +import io.github.dfa1.vortex.reader.compute.Predicate; import org.apache.calcite.jdbc.JavaTypeFactoryImpl; import org.apache.calcite.rex.RexBuilder; @@ -32,9 +33,10 @@ final class RexFilterTranslator { private RexFilterTranslator() { } - /// Translates the Calcite predicates into a zone-map [RowFilter], keeping only the comparisons - /// we can prune on (`=`, `<>`, `<`, `<=`, `>`, `>=` between a column and a literal, plus `AND`). - /// Predicates we don't understand are simply not pushed — Calcite still applies them. + /// Translates the Calcite predicates into a zone-map [RowFilter], keeping only what we can prune + /// on: `=`, `<>`, `<`, `<=`, `>`, `>=` between a column and a literal, `IS [NOT] NULL`, a + /// single-column `OR` (such as an `IN` list), and `AND`. Predicates we don't understand are + /// simply not pushed — Calcite still applies them. /// /// @param filters the predicates Calcite pushed onto the scan /// @param names the table's column names, by field index @@ -108,9 +110,8 @@ private static Optional toComparison(RexNode node, List n return comparison(node, names, types, false); } - /// Strict counterpart of [#toComparison]: the same column-vs-literal / `AND` grammar, but a - /// single unrecognized node (or one `AND` conjunct) collapses the whole result to empty rather - /// than being dropped, and bare `IS NULL` / `IS NOT NULL` are also translatable. Used by + /// Strict counterpart of [#toComparison]: the same grammar, but a single unrecognized node (or + /// one `AND` conjunct) collapses the whole result to empty rather than being dropped. Used by /// [#translateStrict] so aggregate push-down answers from stats only when the [RowFilter] /// captures the predicate in full. Delegates to the shared [#comparison] dispatch with /// `strict = true`. @@ -119,14 +120,14 @@ private static Optional strictComparison(RexNode node, List`, `<`, `<=`, `>`, `>=`) translate identically - /// in both walks through [#binary]. + /// (strict). The two differ only in how an untranslatable `AND` conjunct is handled: a `strict` + /// walk abandons the whole predicate on the first one, while a lenient walk drops it and keeps + /// the rest. Comparisons (`=`, `<>`, `<`, `<=`, `>`, `>=`, either operand order) and the null + /// tests translate identically in both walks. + /// + /// An `OR` translates only when every disjunct does and all of them test one column, into one + /// column predicate (an `IN` list arrives this way once `SEARCH` is expanded). Dropping a + /// disjunct would narrow the filter and lose rows, so an `OR` is all-or-nothing in both walks. /// /// @param node the (already `SEARCH`-expanded) predicate node /// @param names the table's column names, by field index @@ -153,13 +154,50 @@ private static Optional comparison(RexNode node, List nam yield parts.isEmpty() ? Optional.empty() : Optional.of(RowFilter.and(parts.toArray(RowFilter[]::new))); } + case OR -> { + ColumnName column = null; + Predicate predicate = null; + for (RexNode operand : call.getOperands()) { + Optional part = comparison(operand, names, types, true) + .flatMap(RexFilterTranslator::singleColumn); + if (part.isEmpty() || column != null && !column.equals(part.get().column())) { + yield Optional.empty(); + } + column = part.get().column(); + predicate = predicate == null ? part.get().predicate() + : new Predicate.Or(predicate, part.get().predicate()); + } + yield Optional.of(new RowFilter.Column(column, predicate)); + } case EQUALS, NOT_EQUALS, LESS_THAN, LESS_THAN_OR_EQUAL, GREATER_THAN, GREATER_THAN_OR_EQUAL -> binary(call, names, types); - case IS_NULL, IS_NOT_NULL -> strict ? nullCheck(call, names) : Optional.empty(); + case IS_NULL, IS_NOT_NULL -> nullCheck(call, names); default -> Optional.empty(); }; } + /// The filter as one column's predicate: itself for a column filter, the conjunction of an `AND` + /// whose parts all test the same column, else empty. + private static Optional singleColumn(RowFilter filter) { + return switch (filter) { + case RowFilter.Column column -> Optional.of(column); + case RowFilter.And(var parts) -> { + ColumnName column = null; + Predicate predicate = null; + for (RowFilter part : parts) { + Optional one = singleColumn(part); + if (one.isEmpty() || column != null && !column.equals(one.get().column())) { + yield Optional.empty(); + } + column = one.get().column(); + predicate = predicate == null ? one.get().predicate() + : new Predicate.And(predicate, one.get().predicate()); + } + yield Optional.of(new RowFilter.Column(column, predicate)); + } + }; + } + /// Translates `col IS NULL` / `col IS NOT NULL` over a bare column reference into the matching /// [RowFilter], which the zone-map fold answers from each zone's null count. Anything other than /// a direct [RexInputRef] operand (e.g. `IS NULL` over an expression) abandons the translation — @@ -177,7 +215,21 @@ private static Optional nullCheck(RexCall call, List name private static Optional binary(RexCall call, List names, List types) { List ops = call.getOperands(); - if (ops.size() != 2 || !(ops.get(0) instanceof RexInputRef ref) || !(ops.get(1) instanceof RexLiteral lit)) { + if (ops.size() != 2) { + return Optional.empty(); + } + SqlKind kind = call.getKind(); + RexInputRef ref; + RexLiteral lit; + if (ops.get(0) instanceof RexInputRef r && ops.get(1) instanceof RexLiteral l) { + ref = r; + lit = l; + } else if (ops.get(0) instanceof RexLiteral l && ops.get(1) instanceof RexInputRef r) { + // `5 < x` is `x > 5` + ref = r; + lit = l; + kind = kind.reverse(); + } else { return Optional.empty(); } Object val = literalValue(lit, types.get(ref.getIndex())); @@ -186,14 +238,14 @@ private static Optional binary(RexCall call, List names, } ColumnName col = names.get(ref.getIndex()); Comparable cmp = (Comparable) val; - return Optional.of(switch (call.getKind()) { + return Optional.of(switch (kind) { case EQUALS -> RowFilter.eq(col, val); case NOT_EQUALS -> RowFilter.neq(col, val); case LESS_THAN -> RowFilter.lt(col, cmp); case LESS_THAN_OR_EQUAL -> RowFilter.lte(col, cmp); case GREATER_THAN -> RowFilter.gt(col, cmp); case GREATER_THAN_OR_EQUAL -> RowFilter.gte(col, cmp); - default -> throw new IllegalStateException("unreachable kind: " + call.getKind()); + default -> throw new IllegalStateException("unreachable kind: " + kind); }); } 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 052ad7c56..b761c8471 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 @@ -620,10 +620,9 @@ private static Match classifyColumn(Predicate predicate, ArrayStats s, long rowC /// [Predicate] unification: an unrecognized stat shape or a partially-overlapping zone is /// [Match#BOUNDARY], a zone provably outside the predicate is [Match#OUT], and a zone every row /// of which matches (which, for a value comparison, also requires the zone to carry no nulls) is - /// [Match#IN]. The composite and range predicates ([Predicate.Between] / [Predicate.And] / - /// [Predicate.Or]) cannot arise from the Calcite translation — `SEARCH` / `BETWEEN` expand to an - /// `AND` of single-leaf columns — so they fall through to [Match#BOUNDARY], which decodes and - /// applies the kernel mask and is therefore always correct. + /// [Match#IN]. A single-column [Predicate.And] / [Predicate.Or] (an `IN` list, a range union) + /// combines its sides' verdicts; [Predicate.Between] cannot arise from the Calcite translation + /// and stays [Match#BOUNDARY], which decodes and is therefore always correct. /// /// @param predicate the column-bound value-test /// @param s the zone's statistics for that column @@ -684,8 +683,22 @@ private static Match classifyColumnRaw(Predicate predicate, ArrayStats s, long r yield nulls == rowCount ? Match.OUT : Match.BOUNDARY; } case Predicate.Between _ -> Match.BOUNDARY; - case Predicate.And _ -> Match.BOUNDARY; - case Predicate.Or _ -> Match.BOUNDARY; + case Predicate.And(var left, var right) -> { + Match l = classifyColumnRaw(left, s, rowCount); + Match r = classifyColumnRaw(right, s, rowCount); + if (l == Match.OUT || r == Match.OUT) { + yield Match.OUT; + } + yield l == Match.IN && r == Match.IN ? Match.IN : Match.BOUNDARY; + } + case Predicate.Or(var left, var right) -> { + Match l = classifyColumnRaw(left, s, rowCount); + Match r = classifyColumnRaw(right, s, rowCount); + if (l == Match.IN || r == Match.IN) { + yield Match.IN; + } + yield l == Match.OUT && r == Match.OUT ? Match.OUT : Match.BOUNDARY; + } }; } diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java new file mode 100644 index 000000000..e367fab7e --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/FilterTranslationTest.java @@ -0,0 +1,132 @@ +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.encode.NullableData; + +import org.apache.calcite.jdbc.JavaTypeFactoryImpl; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rex.RexBuilder; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.type.SqlTypeName; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.math.BigDecimal; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// Predicate shapes the translator used to leave to Calcite: an `IN` list (an `OR` of equalities +/// once Calcite's `SEARCH` is expanded), a literal on the left, and `IS NULL`. Left untranslated +/// they neither prune chunks nor are enforced by the scan, so Calcite re-checks every row. +class FilterTranslationTest { + + private static final RexBuilder REX = new RexBuilder(new JavaTypeFactoryImpl()); + private static final RelDataType BIGINT = REX.getTypeFactory().createSqlType(SqlTypeName.BIGINT); + private static final RelDataType NULLABLE_BIGINT = REX.getTypeFactory() + .createTypeWithNullability(BIGINT, true); + + @TempDir + static Path tmp; + private static Path file; + + // two chunks: id 1..3 with no nulls, id 4..6 where n is null on id 5 + @BeforeAll + static void write() throws Exception { + file = tmp.resolve("translate.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("n"), DType.I64.asNullable()) + .build(); + try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + VortexWriter w = VortexWriter.create(ch, schema, LegacyZoneMaps.OPTIONS)) { + w.writeChunk(Map.of(ColumnName.of("id"), new long[]{1, 2, 3}, + ColumnName.of("n"), new NullableData(new long[]{10, 20, 30}, new boolean[]{true, true, true}))); + w.writeChunk(Map.of(ColumnName.of("id"), new long[]{4, 5, 6}, + ColumnName.of("n"), new NullableData(new long[]{40, 0, 60}, new boolean[]{true, false, true}))); + } + } + + @Test + void inList_isEnforcedAndPrunes() throws Exception { + // Given id IN (1, 3) + RexNode in = REX.makeIn(REX.makeInputRef(BIGINT, 0), List.of(literal(1), literal(3))); + + // When / Then — only the first chunk is decoded, and the scan enforces the list itself + assertScan(in, List.of(1L, 3L), 1); + } + + @Test + void inListOverTwoColumns_isLeftToCalcite() throws Exception { + // Given id = 1 OR n = 60 — no single-column predicate captures it + RexNode or = REX.makeCall(SqlStdOperatorTable.OR, + REX.makeCall(SqlStdOperatorTable.EQUALS, REX.makeInputRef(BIGINT, 0), literal(1)), + REX.makeCall(SqlStdOperatorTable.EQUALS, REX.makeInputRef(NULLABLE_BIGINT, 1), literal(60))); + List filters = new ArrayList<>(List.of(or)); + + // When + List result = ids(filters); + + // Then — nothing pushed, so Calcite keeps the predicate and the scan returns every row + assertThat(filters).hasSize(1); + assertThat(result).hasSize(6); + } + + @Test + void literalOnTheLeft_isEnforcedAndPrunes() throws Exception { + // Given 3 >= id + RexNode cmp = REX.makeCall(SqlStdOperatorTable.GREATER_THAN_OR_EQUAL, literal(3), REX.makeInputRef(BIGINT, 0)); + + // When / Then + assertScan(cmp, List.of(1L, 2L, 3L), 1); + } + + @Test + void isNull_prunesChunksWithoutNulls() throws Exception { + // Given n IS NULL — the first chunk has a zero null count + RexNode isNull = REX.makeCall(SqlStdOperatorTable.IS_NULL, REX.makeInputRef(NULLABLE_BIGINT, 1)); + + // When / Then + assertScan(isNull, List.of(5L), 1); + } + + private static void assertScan(RexNode predicate, List expectedIds, long expectedChunks) { + // When + List filters = new ArrayList<>(List.of(predicate)); + VortexTable table = new VortexTable(file); + List result = ids(table, filters); + + // Then + assertThat(filters).as("captured in full, so Calcite does not re-check").isEmpty(); + assertThat(result).containsExactlyElementsOf(expectedIds); + assertThat(table.chunksScannedLastQuery()).as("chunks decoded").isEqualTo(expectedChunks); + } + + private static List ids(List filters) { + return ids(new VortexTable(file), filters); + } + + private static List ids(VortexTable table, List filters) { + List ids = new ArrayList<>(); + try (Enumerator en = table.scan(null, filters, new int[]{0}).enumerator()) { + while (en.moveNext()) { + ids.add(en.current()[0]); + } + } + return ids; + } + + private static RexNode literal(long value) { + return REX.makeExactLiteral(BigDecimal.valueOf(value), BIGINT); + } +} diff --git a/docs/reference.md b/docs/reference.md index efacf3fe5..b1833afbc 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -612,6 +612,9 @@ preserved) and the **Babel** SQL parser, so a column named after a reserved word keywords that open a typed literal (`date`, `time`, `timestamp`, `interval`), which stay reserved even under Babel and need back-ticks: `` select `date` from vtx.ohlc ``. +A `WHERE` of those predicates (an `OR` only within one column) prunes chunks from zone maps and is +enforced by the scan, so Calcite does not re-check it; anything else is left to Calcite. + Column types map as follows. Unsigned integers widen to the next signed type (`U64` stays `BIGINT` and fails on values past 2^63); `F16` reads as `REAL`. @@ -628,8 +631,8 @@ and fails on values past 2^63); `F16` reads as `REAL`. `TIMESTAMP WITH LOCAL TIME ZONE` because Avatica's JDBC getters shift that type by the JVM's zone. `count`/`min`/`max`/`sum` (and `avg`, reduced to `sum`/`count`) are answered from zone-map -statistics via `VortexAggregatePushDownRule` where possible, with or without a `WHERE` on -column-vs-literal comparisons: zones the filter selects entirely fold from their statistics, and only +statistics via `VortexAggregatePushDownRule` where possible, with or without a `WHERE` of +column-vs-literal comparisons, `IN` lists and `IS [NOT] NULL` joined by `AND`: zones the filter selects entirely fold from their statistics, and only the rows of zones it cuts are decoded. `sum` needs a per-zone sum, which only the legacy `vortex.stats` zone map records (writes targeting an edition before `core2026.08.0`); with Rust's `vortex.zoned` — the default since #447, and every Rust-written file — `sum` is answered by a scan. diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index 8994e2155..5f5f07326 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -25,6 +25,8 @@ import org.apache.arrow.vector.types.pojo.Schema; 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.CsvSource; import java.math.BigDecimal; import java.nio.file.Path; @@ -108,6 +110,36 @@ void jniWriter_filteredCountMinMax_foldFromRustZones(@TempDir Path tmp) throws E } } + @ParameterizedTest(name = "where {0}") + @CsvSource(delimiter = '|', value = { + // an IN list arrives as an OR of equalities on one column + "id in (3, 1500, 2999) | 3, 1500, 2999", + // a literal on the left + "2996 < id | 2997, 2998, 2999", + // n is null on every 1000th row + "n is null | 0, 1000, 2000", + "n is null and id in (0, 1, 1000) | 0, 1000"}) + void jniWriter_translatedFilters_returnExactlyTheMatchingRows(String where, String expected, @TempDir Path tmp) + throws Exception { + // Given + Path file = tmp.resolve("jni_filters.vortex"); + writeIdsWithJni(file, BATCHES, BATCH_ROWS); + + // When + List result = new java.util.ArrayList<>(); + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("select id from vtx.t where " + where + " order by id")) { + while (rs.next()) { + result.add(rs.getLong(1)); + } + } + + // Then + assertThat(result).containsExactlyElementsOf( + java.util.Arrays.stream(expected.split(",")).map(String::trim).map(Long::valueOf).toList()); + } + @Test void jniWriter_temporalDecimalAndBinaryColumns_readThroughSql(@TempDir Path tmp) throws Exception { // Given — a file with a timestamp could not be queried at all: the row type threw @@ -163,15 +195,25 @@ private static void writeTypesWithJni(Path file) throws Exception { } private static void writeIdsWithJni(Path file, int batches, int batchRows) throws Exception { - Schema schema = new Schema(List.of(Field.notNullable("id", new ArrowType.Int(64, true)))); + Schema schema = new Schema(List.of( + Field.notNullable("id", new ArrowType.Int(64, true)), + Field.nullable("n", new ArrowType.Int(64, true)))); try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, ALLOCATOR).build()) { for (int batch = 0; batch < batches; batch++) { try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { BigIntVector ids = (BigIntVector) root.getVector("id"); + BigIntVector n = (BigIntVector) root.getVector("n"); ids.allocateNew(batchRows); + n.allocateNew(batchRows); for (int i = 0; i < batchRows; i++) { - ids.set(i, (long) batch * batchRows + i); + long id = (long) batch * batchRows + i; + ids.set(i, id); + if (id % 1000 == 0) { + n.setNull(i); + } else { + n.set(i, id); + } } root.setRowCount(batchRows); try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); From f7377d947fcb43d6e02515ac1cdc4a4b3006ce4f Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 08:07:33 +0200 Subject: [PATCH 7/7] fix(performance): Calcite boundary benchmark times the fold again Since #447 the fixture was written with the default vortex.zoned zone map: 8192-row zones with no sum, and the writer repartitions the 100 batches into 8 chunks. The fold could not answer SUM on that layout, so boundaryFold fell back to the scan and the A/B timed the baseline twice. With the zoned fold on this branch the fold succeeds but reports a null sum, and the benchmark threw a NullPointerException. - Write the fixture for core2025.10.0, the last edition before vortex.zoned: one vortex.stats zone per batch, with sums. - Fail setup when the fold cannot answer SUM where interior zones exist, so the benchmark cannot silently go stale again. - Run the honesty check only in the boundaryFold trial. Decoding through the fold in the baseline's JVM polluted the reader's type profiles: fullScanBaseline medium ran 2.99-3.04 ms with the check, 1.08-1.28 ms without. Single fork, ms/op: range fold main fold branch scan main scan branch wide 0.354 0.361 4.888 4.945 medium 0.308 0.347 1.127 1.119 narrow 0.743 0.774 0.705 0.738 The fold is about 14x faster than the scan on wide and at parity on narrow, where it has no interior zone and abandons to the scan. Co-Authored-By: Claude Sonnet 5 --- .../CalciteBoundaryAggregateBenchmark.java | 26 +++++++++++++++---- 1 file changed, 21 insertions(+), 5 deletions(-) diff --git a/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java b/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java index f578177dd..e917fb1fe 100644 --- a/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java +++ b/performance/src/main/java/io/github/dfa1/vortex/performance/CalciteBoundaryAggregateBenchmark.java @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.core.model.ColumnName; import io.github.dfa1.vortex.calcite.VortexTable; import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.Editions; import io.github.dfa1.vortex.reader.Chunk; import io.github.dfa1.vortex.reader.RowFilter; import io.github.dfa1.vortex.reader.ScanOptions; @@ -23,6 +24,7 @@ import org.openjdk.jmh.annotations.State; import org.openjdk.jmh.annotations.TearDown; import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.BenchmarkParams; import java.io.IOException; import java.nio.channels.FileChannel; @@ -70,8 +72,8 @@ /// [#fullScanBaseline] decodes one chunk per surviving zone, so the fold wins in proportion to the /// interior-chunk count — large on `wide`, smaller on `medium`, parity on `narrow`. The mode is /// [Mode#AverageTime] in milliseconds because the metric of interest is per-query latency. Both -/// methods return the sum so JMH cannot dead-code-eliminate the work; [#setup] asserts the two paths -/// agree once before timing so the A/B is honest. +/// methods return the sum so JMH cannot dead-code-eliminate the work; the fold's trial asserts in +/// [#setup] that the two paths agree before timing, so the A/B is honest. @State(Scope.Benchmark) @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.MILLISECONDS) @@ -103,7 +105,7 @@ public class CalciteBoundaryAggregateBenchmark { private long hi; @Setup(Level.Trial) - public void setup() throws IOException { + public void setup(BenchmarkParams params) throws IOException { long[] bounds = bounds(range); lo = bounds[0]; hi = bounds[1]; @@ -116,6 +118,17 @@ public void setup() throws IOException { filter = RowFilter.and(RowFilter.gte(ColumnName.of("id"), lo), RowFilter.lte(ColumnName.of("id"), hi)); baselineOptions = ScanOptions.columns(ColumnName.of("id"), ColumnName.of(AGG_COLUMN)).withFilter(filter); + // Only the fold's trial runs the checks: decoding through the fold in the baseline's JVM + // pollutes the reader's type profiles and slowed the scan 2-3x, timing the setup, not the scan. + if (!params.getBenchmark().endsWith(".boundaryFold")) { + return; + } + // The fold must not abandon where interior zones exist: an abandoned fold falls back to the + // scan, and the A/B would silently time the baseline twice. + if (!"narrow".equals(range) && table.filteredFold(filter, ColumnName.of(AGG_COLUMN)) + .map(VortexTable.FilteredFold::sum).isEmpty()) { + throw new IllegalStateException("boundary fold cannot answer SUM for range=" + range); + } // Honesty check: the optimized fold and the full-scan baseline must agree before we time them. long folded = boundaryFold(); long scanned = fullScanBaseline(); @@ -193,8 +206,11 @@ private static void writeFixture(Path file) throws IOException { .field(ColumnName.of("id"), DType.I64) .field(ColumnName.of("val"), DType.I64) .build(); - // enableZoneMaps=true emits the per-chunk min/max/sum/null-count the interior-zone fold reads. - WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(true).withoutEditions(); + // core2025.10.0 predates Rust's vortex.zoned, so the writer emits the legacy vortex.stats zone + // map: one zone per batch with the min/max/sum/null-count the interior-zone fold reads. The + // default vortex.zoned has 8192-row zones and no sum, so a filtered SUM always scans. + WriteOptions opts = WriteOptions.cascading(0).withZoneMaps(true).withGlobalDict(true) + .withEdition(Editions.CORE_2025_10_0); java.util.Random rng = new java.util.Random(SEED); try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING);