From d919c82c3d90a57cdf85c37e9499cc7e2f1a0f41 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 08:25:19 +0200 Subject: [PATCH 1/3] perf(performance): benchmark reads of a dictionary-encoded string column DictStringBenchmark reads a 1M-row, 8-value Utf8 column four ways: a bare reader scan through getString, and Calcite SQL scan, GROUP BY and filter. It is the before/after for #551 (string decode per row) and #552 (string comparison per row). Setup checks the column decodes to VarBinDictArray so a writer change cannot silently benchmark another representation. Co-Authored-By: Claude Sonnet 5 --- .../performance/DictStringBenchmark.java | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) create mode 100644 performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java diff --git a/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java b/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java new file mode 100644 index 000000000..1033633d7 --- /dev/null +++ b/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java @@ -0,0 +1,166 @@ +package io.github.dfa1.vortex.performance; + +import io.github.dfa1.vortex.calcite.VortexSchema; +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.reader.Chunk; +import io.github.dfa1.vortex.reader.ScanOptions; +import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.VarBinArray; +import io.github.dfa1.vortex.reader.array.VarBinDictArray; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; +import org.apache.calcite.jdbc.CalciteConnection; +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.TearDown; +import org.openjdk.jmh.annotations.Warmup; +import org.openjdk.jmh.infra.Blackhole; + +import java.io.IOException; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Map; +import java.util.Properties; +import java.util.Random; +import java.util.concurrent.TimeUnit; + +/// Reads of a low-cardinality, dictionary-encoded Utf8 column: the projection cost behind #551 +/// (one string decode per row instead of per dictionary entry) and the filter cost behind #552 +/// (a string comparison per row instead of per dictionary entry). +/// +/// Fixture: 1M rows, `sym` drawn from 8 tickers (the cardinality of the #550 profile that found +/// both issues) and an I64 `val`. `@Setup` checks `sym` decodes to a `VarBinDictArray`, so a writer +/// or decoder change cannot turn this into a benchmark of some other representation. +/// +/// - [#readStrings] isolates the array accessor: a bare reader scan calling `getString` per row. +/// - [#sqlScan], [#sqlGroupBy] and [#sqlFilter] run the same column through Calcite, where the +/// accessor cost competes with Calcite's own per-row overhead. +@State(Scope.Benchmark) +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.MILLISECONDS) +@Warmup(iterations = 3, time = 3) +@Measurement(iterations = 5, time = 3) +@Fork(value = 1, jvmArgsAppend = { + "--add-opens", "java.base/java.nio=ALL-UNNAMED", + "--enable-native-access=ALL-UNNAMED", + "--sun-misc-unsafe-memory-access=allow" +}) +public class DictStringBenchmark { + + private static final int TOTAL_ROWS = 1_000_000; + private static final int BATCH_SIZE = 50_000; + private static final ColumnName SYM = ColumnName.of("sym"); + private static final ColumnName VAL = ColumnName.of("val"); + private static final String[] TICKERS = {"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "AVGO"}; + + private Path file; + private Connection connection; + + @Setup(Level.Trial) + public void setup() throws IOException, SQLException { + file = Files.createTempFile("dict-string-bench", ".vortex"); + writeFixture(file); + try (VortexReader reader = VortexReader.open(file); + var iter = reader.scan(ScanOptions.columns(SYM)); + Chunk c = iter.next()) { + Array sym = c.column(SYM); + if (!(sym instanceof VarBinDictArray)) { + throw new VortexException("expected sym to decode to VarBinDictArray, got " + + sym.getClass().getSimpleName()); + } + } + Properties info = new Properties(); + info.setProperty("lex", "JAVA"); + connection = DriverManager.getConnection("jdbc:calcite:", info); + connection.unwrap(CalciteConnection.class).getRootSchema() + .add("vtx", new VortexSchema(Map.of("data", file))); + } + + @TearDown(Level.Trial) + public void cleanup() throws IOException, SQLException { + connection.close(); + Files.deleteIfExists(file); + } + + @Benchmark + public void readStrings(Blackhole bh) throws IOException { + try (VortexReader reader = VortexReader.open(file); + var iter = reader.scan(ScanOptions.columns(SYM))) { + while (iter.hasNext()) { + try (Chunk c = iter.next()) { + VarBinArray sym = c.column(SYM); + long n = sym.length(); + for (long i = 0; i < n; i++) { + bh.consume(sym.getString(i)); + } + } + } + } + } + + @Benchmark + public void sqlScan(Blackhole bh) throws SQLException { + query("select sym from vtx.data", bh); + } + + @Benchmark + public void sqlGroupBy(Blackhole bh) throws SQLException { + query("select sym, count(*) from vtx.data group by sym", bh); + } + + @Benchmark + public void sqlFilter(Blackhole bh) throws SQLException { + query("select val from vtx.data where sym = 'NVDA'", bh); + } + + private void query(String sql, Blackhole bh) throws SQLException { + try (Statement st = connection.createStatement(); + ResultSet rs = st.executeQuery(sql)) { + int columns = rs.getMetaData().getColumnCount(); + while (rs.next()) { + for (int c = 1; c <= columns; c++) { + bh.consume(rs.getObject(c)); + } + } + } + } + + private static void writeFixture(Path file) throws IOException { + DType.Struct schema = DType.structBuilder() + .field(SYM, DType.UTF8) + .field(VAL, DType.I64) + .build(); + Random rng = new Random(42L); + try (FileChannel ch = FileChannel.open(file, + StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING); + VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + for (int b = 0; b < TOTAL_ROWS / BATCH_SIZE; b++) { + String[] sym = new String[BATCH_SIZE]; + long[] val = new long[BATCH_SIZE]; + for (int i = 0; i < BATCH_SIZE; i++) { + sym[i] = TICKERS[rng.nextInt(TICKERS.length)]; + val[i] = rng.nextInt(1_000); + } + writer.writeChunk(Map.of(SYM, sym, VAL, val)); + } + } + } +} From db3a89b9379ec7b1854c74ce5164fda7c828efd3 Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 08:25:19 +0200 Subject: [PATCH 2/3] perf(reader): decode each dictionary string once, not once per row VarBinDictArray.getString copied and UTF-8 decoded the entry on every call, so an 8-value column over 1M rows decoded 1M strings. It now caches one String per dictionary entry and returns the same instance for every row with that code. VarBinConstantArray, which a single-value dictionary collapses to, caches its one decoded value the same way. The cache holds at most min(dictSize, rows, 65536) entries: the dictionary size comes from an untrusted offsets buffer, and a large shared dictionary over a small chunk must not allocate more than its rows. Codes past the bound decode per call; out-of-range codes still throw VortexException. Writes are unsynchronized: String is immutable, so a race only decodes an entry twice. getBytes still returns a fresh copy. Both classes become final classes (a record cannot hold the cache) with the same accessors. Rust has no such cache because its strings are views into the shared buffer; the cost is JVM-only, so this is no divergence. DictStringBenchmark, single fork, ms/op: readStrings 10.88 -> 2.63 (4.1x) sqlScan 26.91 -> 18.84 (-30%) sqlGroupBy 56.93 -> 36.07 (-37%) sqlFilter 35.03 -> 22.96 (-34%) Closes #551 Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 2 + .../RustWritesJavaReadsIntegrationTest.java | 75 +++++++++++ .../reader/array/VarBinConstantArray.java | 53 ++++++-- .../vortex/reader/array/VarBinDictArray.java | 117 ++++++++++++++++-- .../vortex/reader/array/VarBinArrayTest.java | 56 +++++++++ 5 files changed, 283 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b9f741296..f20f41eda 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,6 +18,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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)) +- Reading strings from a dictionary-encoded column is about 4× faster, and SQL over it about 30% faster: each distinct value is decoded once, not once per row. ([#551](https://github.com/dfa1/vortex-java/issues/551)) +- **Breaking:** `VarBinDictArray` and `VarBinConstantArray` are final classes instead of records, with the same accessors. ([#551](https://github.com/dfa1/vortex-java/issues/551)) - **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/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java index ed8647ee7..6e6b98e6d 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java @@ -311,6 +311,81 @@ void jniWriter_filteredUtf8Scan_comparesInUtf8ByteOrder(@TempDir Path tmp) throw assertThat(result).isEqualTo(1); } + @Test + void jniWriter_lowCardinalityUtf8_cachedDictStringsMatchRows(@TempDir Path tmp) throws IOException { + // Given — Rust dict-encodes a low-cardinality string column; Java caches one decoded String + // per dictionary entry (#551). Every row must still read its own value, through a full scan + // and through a filtered scan (whose predicate reads the same cached strings). The non-ASCII + // value checks the cached entry is the UTF-8 decode, not a byte-per-char shortcut. + String[] symbols = {"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "café"}; + int n = 100_000; + String[] values = new String[n]; + for (int i = 0; i < n; i++) { + values[i] = symbols[(i * 7 + i / 3) % symbols.length]; + } + Path file = tmp.resolve("jni_utf8_lowcard.vtx"); + writeJniUtf8(file, values); + long expectedMatches = Arrays.stream(values).filter("café"::equals).count(); + + // When + var all = new ArrayList(); + var filtered = new ArrayList(); + try (var vf = VortexReader.open(file, ReadRegistry.loadAll())) { + try (var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all())) { + iter.forEachRemaining(c -> { + VarBinArray k = c.column(ColumnName.of("k")); + assertThat(k).isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinDictArray.class); + for (long i = 0; i < k.length(); i++) { + all.add(k.getString(i)); + } + }); + } + // The scan filter only prunes zones; rows are matched per row, as the Calcite adapter does + var filter = io.github.dfa1.vortex.reader.RowFilter.eq(ColumnName.of("k"), "café"); + try (var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all().withFilter(filter))) { + iter.forEachRemaining(c -> { + VarBinArray k = c.column(ColumnName.of("k")); + for (long i = 0; i < k.length(); i++) { + if (io.github.dfa1.vortex.reader.compute.Compute.matches(c, filter, i)) { + filtered.add(k.getString(i)); + } + } + }); + } + } + + // Then + assertThat(all).containsExactly(values); + assertThat(filtered).hasSize((int) expectedMatches).containsOnly("café"); + } + + @Test + void jniWriter_singleValueUtf8_cachedConstantStringMatchesRows(@TempDir Path tmp) throws IOException { + // Given — one distinct value: Rust writes it as a constant, which Java reads as a + // VarBinConstantArray that decodes the value once and shares it across rows (#551) + int n = 10_000; + String[] values = new String[n]; + Arrays.fill(values, "café"); + Path file = tmp.resolve("jni_utf8_single.vtx"); + writeJniUtf8(file, values); + + // When + var result = new ArrayList(); + try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); + var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all())) { + iter.forEachRemaining(c -> { + VarBinArray k = c.column(ColumnName.of("k")); + assertThat(k).isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinConstantArray.class); + for (long i = 0; i < k.length(); i++) { + result.add(k.getString(i)); + } + }); + } + + // Then + assertThat(result).containsExactly(values); + } + private static long countChunks(Path file, io.github.dfa1.vortex.reader.ScanOptions opts) throws IOException { long[] count = {0}; try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinConstantArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinConstantArray.java index fdb1c9538..4169f9122 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinConstantArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinConstantArray.java @@ -13,12 +13,46 @@ /// materialization occurs — every accessor returns the same shared bytes, mirroring the /// `LazyConstantXxxArray` family the primitive array types use for the same encoding. /// -/// @param dtype logical element type (Utf8 or Binary) -/// @param length number of logical rows the constant is broadcast across -/// @param bytes the constant value's raw bytes, shared across every row; [#getBytes(long)] -/// clones it per that method's copy contract -@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. -public record VarBinConstantArray(DType dtype, long length, byte[] bytes) implements VarBinArray { +/// [#getString(long)] decodes the value once and returns that `String` for every row (#551): a +/// single-value dictionary collapses to this array, and decoding per row would pay the copy and +/// UTF-8 decode the dictionary cache avoids. +public final class VarBinConstantArray implements VarBinArray { + + private final DType dtype; + private final long length; + private final byte[] bytes; + /// Decoded on the first [#getString(long)]. Unsynchronized on purpose: `String` is immutable, + /// so a race only decodes the value twice. + private String string; + + /// Creates a constant array. + /// + /// @param dtype logical element type (Utf8 or Binary) + /// @param length number of logical rows the constant is broadcast across + /// @param bytes the constant value's raw bytes, shared across every row; [#getBytes(long)] + /// clones it per that method's copy contract + public VarBinConstantArray(DType dtype, long length, byte[] bytes) { + this.dtype = dtype; + this.length = length; + this.bytes = bytes; + } + + @Override + public DType dtype() { + return dtype; + } + + @Override + public long length() { + return length; + } + + /// Returns the constant value's raw bytes, shared rather than copied. + /// + /// @return the shared bytes; callers must not mutate them + public byte[] bytes() { + return bytes; + } @Override public MemorySegment bytesSegment() { @@ -42,7 +76,12 @@ public byte[] getBytes(long i) { @Override public String getString(long i) { Objects.checkIndex(i, length); - return new String(bytes, StandardCharsets.UTF_8); + String s = string; + if (s == null) { + s = new String(bytes, StandardCharsets.UTF_8); + string = s; + } + return s; } @Override diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java index 7b12a75bb..75ceb0ccd 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java @@ -15,18 +15,95 @@ /// Stores dictionary values and per-row codes; all accessors resolve through the /// dictionary without materializing strings at construction time. /// -/// @param dtype logical type (Utf8 or Binary) -/// @param length number of logical elements (rows) -/// @param bytesSegment concatenated raw bytes for all dictionary values -/// @param dictValOffsets offsets into `bytesSegment` for each dictionary entry (length = dictSize + 1) -/// @param dictValOffPType physical type of the dictionary value offsets -/// @param dictCodesSegs per-row dictionary code indices (length = `length`) -/// @param dictCodesPType physical type of the dictionary codes -@SuppressWarnings("java:S6218") // internal data carrier; record components are arrays of immutable primitives or refs that flow through pipelines without ever being compared. -public record VarBinDictArray(DType dtype, long length, MemorySegment bytesSegment, - MemorySegment dictValOffsets, PType dictValOffPType, - MemorySegment dictCodesSegs, PType dictCodesPType) - implements VarBinArray { +/// [#getString(long)] decodes each dictionary entry once and returns the same `String` for every +/// row with that code (#551): a low-cardinality column would otherwise copy and decode the same +/// few values once per row. The Rust reference has no such cache because its strings are views +/// into the shared buffer; building a `java.lang.String` always copies, so the cost is JVM-only. +public final class VarBinDictArray implements VarBinArray { + + /// Most entries the string cache holds. The dictionary size comes from an untrusted offsets + /// buffer, so codes past this bound decode per call rather than size a heap array from it. + private static final int MAX_CACHED_STRINGS = 1 << 16; + + private final DType dtype; + private final long length; + private final MemorySegment bytesSegment; + private final MemorySegment dictValOffsets; + private final PType dictValOffPType; + private final MemorySegment dictCodesSegs; + private final PType dictCodesPType; + /// Codes below this are cached; also bounded by `length` so a dictionary far larger than + /// the rows (a shared dictionary over a small chunk) never allocates more than the rows do. + private final int cacheSize; + /// Allocated on the first [#getString(long)]. Unsynchronized on purpose: `String` is + /// immutable, so a race only decodes an entry twice. + private String[] strings; + + /// Creates a dictionary-encoded array. + /// + /// @param dtype logical type (Utf8 or Binary) + /// @param length number of logical elements (rows) + /// @param bytesSegment concatenated raw bytes for all dictionary values + /// @param dictValOffsets offsets into `bytesSegment` for each dictionary entry (length = dictSize + 1) + /// @param dictValOffPType physical type of the dictionary value offsets + /// @param dictCodesSegs per-row dictionary code indices (length = `length`) + /// @param dictCodesPType physical type of the dictionary codes + public VarBinDictArray(DType dtype, long length, MemorySegment bytesSegment, + MemorySegment dictValOffsets, PType dictValOffPType, + MemorySegment dictCodesSegs, PType dictCodesPType) { + this.dtype = dtype; + this.length = length; + this.bytesSegment = bytesSegment; + this.dictValOffsets = dictValOffsets; + this.dictValOffPType = dictValOffPType; + this.dictCodesSegs = dictCodesSegs; + this.dictCodesPType = dictCodesPType; + long dictSize = dictValOffsets.byteSize() / dictValOffPType.byteSize() - 1; + this.cacheSize = (int) Math.max(0, Math.min(Math.min(dictSize, length), MAX_CACHED_STRINGS)); + } + + @Override + public DType dtype() { + return dtype; + } + + @Override + public long length() { + return length; + } + + @Override + public MemorySegment bytesSegment() { + return bytesSegment; + } + + /// Returns the offsets into [#bytesSegment()] for each dictionary entry. + /// + /// @return the value offsets segment (length = dictSize + 1) + public MemorySegment dictValOffsets() { + return dictValOffsets; + } + + /// Returns the physical type of the dictionary value offsets. + /// + /// @return the offsets ptype + public PType dictValOffPType() { + return dictValOffPType; + } + + /// Returns the per-row dictionary codes. + /// + /// @return the codes segment (length = [#length()]) + public MemorySegment dictCodesSegs() { + return dictCodesSegs; + } + + /// Returns the physical type of the dictionary codes. + /// + /// @return the codes ptype + public PType dictCodesPType() { + return dictCodesPType; + } @Override public byte[] getBytes(long i) { @@ -41,7 +118,21 @@ public byte[] getBytes(long i) { @Override public String getString(long i) { - return new String(getBytes(i), StandardCharsets.UTF_8); + long code = dictReadCode(i); + if (code < 0 || code >= cacheSize) { + return new String(getBytes(i), StandardCharsets.UTF_8); + } + String[] cache = strings; + if (cache == null) { + cache = new String[cacheSize]; + strings = cache; + } + String s = cache[(int) code]; + if (s == null) { + s = new String(getBytes(i), StandardCharsets.UTF_8); + cache[(int) code] = s; + } + return s; } @Override diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java index f01a7522e..2eedc05dd 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/array/VarBinArrayTest.java @@ -163,6 +163,48 @@ void getBytes_resolvesViaDict() { assertThat(sut.getString(2)).isEqualTo("bar"); } + /// #551: rows sharing a code share one decoded `String`, so a low-cardinality column + /// decodes once per dictionary entry instead of once per row. + @Test + void getString_sameCode_returnsSameInstance() { + // Given — rows 0 and 2 both carry code 1 + VarBinArray sut = ofDict(new String[]{"foo", "bar"}, new int[]{1, 0, 1}); + + // When + String result = sut.getString(2); + + // Then + assertThat(result).isSameAs(sut.getString(0)).isEqualTo("bar"); + } + + /// The cache is bounded by the row count too, so a dictionary larger than the rows (a + /// shared dictionary over a small chunk) leaves the high codes uncached. Those must still + /// decode correctly, as a fresh `String` per call. + @Test + void getString_codePastCacheBound_decodesPerCall() { + // Given — 3 entries but 2 rows, so only codes 0 and 1 are cacheable; both rows use code 2 + VarBinArray sut = ofDict(new String[]{"a", "bb", "ccc"}, new int[]{2, 2}); + + // When + String result = sut.getString(1); + + // Then + assertThat(result).isEqualTo("ccc").isNotSameAs(sut.getString(0)); + } + + /// A code past the dictionary must still fail as malformed input (ADR 0003): the cache + /// must never turn an out-of-range code into an `ArrayIndexOutOfBoundsException`. + @Test + void getString_codeOutsideDictionary_throwsVortexException() { + // Given — 2 entries, row 1 carries code 5 + VarBinArray sut = ofDict(new String[]{"a", "b"}, new int[]{0, 5, 1}); + + // When / Then + assertThatThrownBy(() -> sut.getString(1)) + .isInstanceOf(VortexException.class) + .hasMessageContaining("out of range"); + } + @Test void getByteLength_resolvesViaDict() { // Given — dict=["hi","there"], codes=[0,1,0] @@ -337,6 +379,20 @@ void getString_returnsSameValueForEveryRow() { assertThat(sut.getString(4)).isEqualTo("hi"); } + /// #551: a single-value dictionary collapses to this array, so every row must share one + /// decoded `String` rather than copy and decode the value per row. + @Test + void getString_everyRow_returnsSameInstance() { + // Given + VarBinArray sut = new VarBinConstantArray(UTF8, 3, "café".getBytes(StandardCharsets.UTF_8)); + + // When + String result = sut.getString(2); + + // Then + assertThat(result).isSameAs(sut.getString(0)).isEqualTo("café"); + } + @Test void getByteLength_returnsConstantLength() { // Given From 157a1c4bd736396f525e72f263318444962a0f0a Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Sun, 18 Oct 2026 08:33:40 +0200 Subject: [PATCH 3/3] perf(reader): filter dictionary strings once per value, not once per row A comparison filter on a dictionary-encoded string column read and compared every row's string. Rust's dict compare kernel evaluates the predicate against the dictionary values once and keeps the codes; Java now does the same. FusedFilterAggregate.leafTest lowers a leaf over a VarBinDictArray, or a slice of one, to a per-entry match table, so each row costs a code lookup. Null rows and codes outside the dictionary still go through the per-row evaluator, so null-aware predicates answer as before and bad codes still throw VortexException. Like Rust, the lowering declines when the dictionary has more entries than there are rows to test. The slice matters on real files: Rust's global dict layout decodes one dictionary for the whole column and the scan hands each chunk a VarBinSlicedArray over it (Raincloud sf-salaries, TPC-H c_mktsegment). The row path could not use it: Compute.matches tests one row at a time and has nowhere to keep per-chunk work. Compute.matcher(Chunk, RowFilter) lowers the filter once per chunk and returns a LongPredicate, and the Calcite enumerator now builds one per chunk instead of calling matches per row. The fused aggregate path already lowers through leafTest. VarBinDictArray gains values() and code(long), after Rust's DictArray. DictStringBenchmark now defaults to 4M rows and runs on a real file with -p file/column/value/agg. Single fork, ms/op, before #551 -> after #551 -> after this: 4M synthetic rows, 8 symbols sqlFilter 136.6 -> 85.7 -> 44.2 sqlFilterSum 132.5 -> 85.4 -> 37.2 sf-salaries (Rust-written, 1.18M rows, Organization Group) sqlFilter 55.5 -> 38.4 -> 16.0 sqlFilterSum 59.9 -> 41.1 -> 19.4 Part of the gain is the enumerator no longer resolving the filter column per row; the two were not measured apart. Closes #552 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01WtuVGQ5h5Lk6Zn4JnkxF1K --- CHANGELOG.md | 2 + .../dfa1/vortex/calcite/VortexEnumerator.java | 7 +- .../dfa1/vortex/calcite/VortexTable.java | 4 +- .../AggregateWhereCleanPartitionTest.java | 2 +- .../RustWritesJavaReadsIntegrationTest.java | 107 ++++++++++++++++ .../performance/DictStringBenchmark.java | 112 +++++++++++----- .../vortex/reader/array/VarBinDictArray.java | 17 +++ .../dfa1/vortex/reader/compute/Compute.java | 20 ++- .../reader/compute/FusedFilterAggregate.java | 66 +++++++++- .../vortex/reader/compute/ComputeArrays.java | 17 +++ .../vortex/reader/compute/ComputeTest.java | 121 ++++++++++++++++++ 11 files changed, 435 insertions(+), 40 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f20f41eda..f7999c84f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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)) +- `Compute.matcher(Chunk, RowFilter)` lowers a filter once per chunk and returns a row test, for callers that test many rows. ([#552](https://github.com/dfa1/vortex-java/issues/552)) - `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)) @@ -19,6 +20,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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)) - Reading strings from a dictionary-encoded column is about 4× faster, and SQL over it about 30% faster: each distinct value is decoded once, not once per row. ([#551](https://github.com/dfa1/vortex-java/issues/551)) +- SQL filters on a dictionary-encoded string column are about twice as fast: the filter is evaluated once per dictionary value, as in Rust, not once per row. ([#552](https://github.com/dfa1/vortex-java/issues/552)) - **Breaking:** `VarBinDictArray` and `VarBinConstantArray` are final classes instead of records, with the same accessors. ([#551](https://github.com/dfa1/vortex-java/issues/551)) - **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/VortexEnumerator.java b/calcite/src/main/java/io/github/dfa1/vortex/calcite/VortexEnumerator.java index cb42762bb..9800e9b9f 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 @@ -34,6 +34,7 @@ import java.io.UncheckedIOException; import java.nio.file.Path; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongPredicate; /// Streaming [Enumerator] over a Vortex scan: advances chunk by chunk, decoding each requested /// column once per chunk and materializing one `Object[]` row per [#moveNext()]. Rows are not @@ -48,6 +49,7 @@ final class VortexEnumerator implements Enumerator { private final VortexReader reader; private final ScanIterator scan; private Chunk chunk; + private LongPredicate rowMatcher; private Object[] columns; private long rowInChunk; private long chunkRows; @@ -61,7 +63,7 @@ final class VortexEnumerator implements Enumerator { /// @param types output column dtypes, parallel to `names` /// @param exactFilter a predicate [VortexTable#scan(DataContext, java.util.List, int[])] /// removed from Calcite's own re-check list, enforced here row-by-row - /// via [Compute#matches(Chunk, RowFilter, long)] so the result stays + /// via [Compute#matcher(Chunk, RowFilter)] so the result stays /// exact — or `null` when every pushed predicate is still Calcite's to /// re-check VortexEnumerator(Path file, AtomicLong chunksScannedLastQuery, ScanOptions options, @@ -101,7 +103,7 @@ public Object[] current() { public boolean moveNext() { while (true) { if (chunk != null && rowInChunk < chunkRows) { - if (exactFilter != null && !Compute.matches(chunk, exactFilter, rowInChunk)) { + if (exactFilter != null && !rowMatcher.test(rowInChunk)) { rowInChunk++; continue; } @@ -124,6 +126,7 @@ public boolean moveNext() { chunksScannedLastQuery.incrementAndGet(); chunkRows = chunk.rowCount(); rowInChunk = 0; + rowMatcher = exactFilter == null ? null : Compute.matcher(chunk, exactFilter); columns = new Object[names.length]; for (int c = 0; c < names.length; c++) { columns[c] = chunk.column(names[c]); 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 b761c8471..ad3385368 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 @@ -831,8 +831,8 @@ public Enumerable scan(DataContext root, List filters, int[] Optional pushed = RexFilterTranslator.toRowFilter(filters, allNames, allTypes); // Exact push-down: a top-level predicate captured *in full* is enforced row-by-row below - // (via Compute#matches, which routes through the NaN-correct PredicateEvaluator / - // PrimitiveFilter) and removed from Calcite's own re-check list, so Calcite stops + // (via Compute#matcher, which lowers it per chunk through the NaN-correct PrimitiveFilter / + // PredicateEvaluator) and removed from Calcite's own re-check list, so Calcite stops // redundantly re-evaluating rows this scan has already excluded — Calcite's // TableScanNode#createProjectableFilterable wraps a `.where()` residual check around // whatever is still left in `filters` after this call returns. diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java index fb081da4d..b742f05fb 100644 --- a/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/AggregateWhereCleanPartitionTest.java @@ -312,7 +312,7 @@ void nanRowSortsAboveEveryValueInAFloatingWhereFilterEvenAfterAbandoningToScan() // Given a NaN price alongside ordinary values that would all otherwise satisfy the filter // below. A mix like this makes every zone at best a BOUNDARY under the floating column's // classify (never IN), so the fold abandons to a full scan — this test proves that scan - // path (VortexEnumerator -> Compute#matches -> PredicateEvaluator) orders NaN as the maximum + // path (VortexEnumerator -> Compute#matcher -> PrimitiveFilter) orders NaN as the maximum // (Rust's total order, #518) too, not just the fold's own classify. DType.Struct schema = new DType.Struct( List.of(ColumnName.of("price"), ColumnName.of("vol")), diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java index 6e6b98e6d..4c200ad1e 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/RustWritesJavaReadsIntegrationTest.java @@ -39,7 +39,9 @@ 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.Arguments; import org.junit.jupiter.params.provider.CsvSource; +import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; @@ -54,6 +56,7 @@ import java.util.List; import java.util.Map; import java.util.function.Consumer; +import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; @@ -386,6 +389,110 @@ void jniWriter_singleValueUtf8_cachedConstantStringMatchesRows(@TempDir Path tmp assertThat(result).containsExactly(values); } + static Stream dictStringFilters() { + List cases = new ArrayList<>(); + for (String op : List.of("EQ", "NOT_EQ", "GT", "GTE", "LT", "LTE")) { + for (String bound : new String[]{"café", "", "ZZZ"}) { + cases.add(Arguments.of(op, bound)); + } + } + return cases.stream(); + } + + @ParameterizedTest(name = "{0} {1}") + @MethodSource("dictStringFilters") + void jniWriter_dictUtf8Filter_selectsTheRowsJniSelects(String op, String bound, @TempDir Path tmp) + throws IOException { + // Given — Rust dict-encodes a nullable low-cardinality string column. Java now evaluates a + // filter once per dictionary entry and tests rows by code (#552), as Rust's dict compare does. + // The emoji and U+E000 sort one way in UTF-8 (Rust's order) and the other in UTF-16, and + // null rows must never match, so Java must keep exactly the rows vortex-jni keeps. + String[] symbols = {"AAPL", "MSFT", "café", "😀", "", null}; + String[] values = new String[20_000]; + for (int i = 0; i < values.length; i++) { + values[i] = symbols[(i * 7 + i / 3) % symbols.length]; + } + Path file = tmp.resolve("jni_dict_utf8_filter.vtx"); + writeJniNullableUtf8(file, values); + ColumnName column = ColumnName.of("k"); + io.github.dfa1.vortex.reader.RowFilter filter = switch (op) { + case "EQ" -> io.github.dfa1.vortex.reader.RowFilter.eq(column, bound); + case "NOT_EQ" -> io.github.dfa1.vortex.reader.RowFilter.neq(column, bound); + case "GT" -> io.github.dfa1.vortex.reader.RowFilter.gt(column, bound); + case "GTE" -> io.github.dfa1.vortex.reader.RowFilter.gte(column, bound); + case "LT" -> io.github.dfa1.vortex.reader.RowFilter.lt(column, bound); + default -> io.github.dfa1.vortex.reader.RowFilter.lte(column, bound); + }; + + // When + List jni = readJniUtf8Filtered(file, Expression.binary(Expression.BinaryOp.valueOf(op), + Expression.column("k"), Expression.literal(bound))); + List result = new ArrayList<>(); + try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); + var iter = vf.scan(io.github.dfa1.vortex.reader.ScanOptions.all().withFilter(filter))) { + iter.forEachRemaining(c -> { + Array k = c.column(column); + Array data = MaskedArray.unwrap(k).inner(); + assertThat(data instanceof io.github.dfa1.vortex.reader.array.VarBinSlicedArray s ? s.inner() : data) + .isInstanceOf(io.github.dfa1.vortex.reader.array.VarBinDictArray.class); + var matcher = io.github.dfa1.vortex.reader.compute.Compute.matcher(c, filter); + for (long i = 0; i < k.length(); i++) { + if (matcher.test(i)) { + result.add(((VarBinArray) data).getString(i)); + } + } + }); + } + + // Then + assertThat(result).containsExactlyInAnyOrderElementsOf(jni); + } + + private static void writeJniNullableUtf8(Path file, String[] values) throws IOException { + Schema schema = new Schema(List.of(Field.nullable("k", new ArrowType.Utf8()))); + String uri = file.toAbsolutePath().toUri().toString(); + try (VortexWriter writer = VortexWriter.builder(SESSION, uri, schema, ALLOCATOR).build(); + VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + VarCharVector vec = (VarCharVector) root.getVector("k"); + vec.allocateNew(values.length); + for (int i = 0; i < values.length; i++) { + if (values[i] == null) { + vec.setNull(i); + } else { + vec.setSafe(i, values[i].getBytes(java.nio.charset.StandardCharsets.UTF_8)); + } + } + root.setRowCount(values.length); + try (ArrowArray arr = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, arr, arrowSchema); + writer.writeBatch(arr.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + + private static List readJniUtf8Filtered(Path file, Expression filter) throws IOException { + ScanOptions opts = ScanOptions.builder() + .projection(Expression.select(new String[]{"k"}, Expression.root())) + .filter(filter) + .build(); + var out = new ArrayList(); + Scan scan = DataSource.open(SESSION, file.toAbsolutePath().toUri().toString()).scan(opts); + while (scan.hasNext()) { + Partition partition = scan.next(); + try (ArrowReader reader = partition.scanArrow(ALLOCATOR)) { + while (reader.loadNextBatch()) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + var vec = root.getVector("k"); + for (int i = 0; i < root.getRowCount(); i++) { + out.add(vec.isNull(i) ? null : vec.getObject(i).toString()); + } + } + } + } + return out; + } + private static long countChunks(Path file, io.github.dfa1.vortex.reader.ScanOptions opts) throws IOException { long[] count = {0}; try (var vf = VortexReader.open(file, ReadRegistry.loadAll()); diff --git a/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java b/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java index 1033633d7..986ab360b 100644 --- a/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java +++ b/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java @@ -8,8 +8,10 @@ import io.github.dfa1.vortex.reader.ScanOptions; import io.github.dfa1.vortex.reader.VortexReader; import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.MaskedArray; import io.github.dfa1.vortex.reader.array.VarBinArray; import io.github.dfa1.vortex.reader.array.VarBinDictArray; +import io.github.dfa1.vortex.reader.array.VarBinSlicedArray; import io.github.dfa1.vortex.writer.VortexWriter; import io.github.dfa1.vortex.writer.WriteOptions; import org.apache.calcite.jdbc.CalciteConnection; @@ -20,6 +22,7 @@ import org.openjdk.jmh.annotations.Measurement; import org.openjdk.jmh.annotations.Mode; import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; import org.openjdk.jmh.annotations.Scope; import org.openjdk.jmh.annotations.Setup; import org.openjdk.jmh.annotations.State; @@ -46,13 +49,18 @@ /// (one string decode per row instead of per dictionary entry) and the filter cost behind #552 /// (a string comparison per row instead of per dictionary entry). /// -/// Fixture: 1M rows, `sym` drawn from 8 tickers (the cardinality of the #550 profile that found -/// both issues) and an I64 `val`. `@Setup` checks `sym` decodes to a `VarBinDictArray`, so a writer +/// Fixture: 4M rows, `sym` drawn from 8 tickers (the size and cardinality of the #550 profile that +/// found both issues) and an I64 `val`. A real file runs instead with `-p file=/path/to/data.vortex` +/// plus `-p column=`, `-p value=` and `-p agg=` naming its dictionary column, a value to filter on +/// and a numeric column to sum, e.g. the Raincloud `sf-salaries` table (`scripts/hydrate-raincloud-corpus.sh`): +/// `-p "column=Organization Group" -p "value=Public Protection" -p "agg=Total Compensation"`. `@Setup` checks the column decodes +/// to a `VarBinDictArray` (or a slice of one, as a dictionary spanning many chunks reads), so a writer /// or decoder change cannot turn this into a benchmark of some other representation. /// /// - [#readStrings] isolates the array accessor: a bare reader scan calling `getString` per row. -/// - [#sqlScan], [#sqlGroupBy] and [#sqlFilter] run the same column through Calcite, where the -/// accessor cost competes with Calcite's own per-row overhead. +/// - [#sqlScan], [#sqlGroupBy], [#sqlFilter] and [#sqlFilterSum] run the same column through +/// Calcite, where the accessor cost competes with Calcite's own per-row overhead. The two filters +/// take different paths: rows through the enumerator, the sum through the fused aggregate. @State(Scope.Benchmark) @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.MILLISECONDS) @@ -65,51 +73,76 @@ }) public class DictStringBenchmark { - private static final int TOTAL_ROWS = 1_000_000; + private static final int TOTAL_ROWS = 4_000_000; private static final int BATCH_SIZE = 50_000; - private static final ColumnName SYM = ColumnName.of("sym"); - private static final ColumnName VAL = ColumnName.of("val"); private static final String[] TICKERS = {"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "AVGO"}; - private Path file; + /// A `.vortex` file to read; empty generates the synthetic fixture. + @Param("") + public String file; + + /// The dictionary-encoded string column. + @Param("sym") + public String column; + + /// The value the filters select. + @Param("NVDA") + public String value; + + /// The numeric column the filtered sum adds up. + @Param("val") + public String agg; + + private Path path; + private boolean generated; + private ColumnName dictColumn; private Connection connection; @Setup(Level.Trial) public void setup() throws IOException, SQLException { - file = Files.createTempFile("dict-string-bench", ".vortex"); - writeFixture(file); - try (VortexReader reader = VortexReader.open(file); - var iter = reader.scan(ScanOptions.columns(SYM)); + generated = file.isEmpty(); + path = generated ? Files.createTempFile("dict-string-bench", ".vortex") : Path.of(file); + if (generated) { + writeFixture(path); + } + dictColumn = ColumnName.of(column); + try (VortexReader reader = VortexReader.open(path); + var iter = reader.scan(ScanOptions.columns(dictColumn)); Chunk c = iter.next()) { - Array sym = c.column(SYM); - if (!(sym instanceof VarBinDictArray)) { - throw new VortexException("expected sym to decode to VarBinDictArray, got " - + sym.getClass().getSimpleName()); + Array data = c.column(dictColumn); + data = data instanceof MaskedArray m ? m.inner() : data; + data = data instanceof VarBinSlicedArray s ? s.inner() : data; + if (!(data instanceof VarBinDictArray)) { + throw new VortexException("expected " + column + " to decode to VarBinDictArray, got " + + data.getClass().getSimpleName()); } } Properties info = new Properties(); info.setProperty("lex", "JAVA"); connection = DriverManager.getConnection("jdbc:calcite:", info); connection.unwrap(CalciteConnection.class).getRootSchema() - .add("vtx", new VortexSchema(Map.of("data", file))); + .add("vtx", new VortexSchema(Map.of("data", path))); } @TearDown(Level.Trial) public void cleanup() throws IOException, SQLException { connection.close(); - Files.deleteIfExists(file); + if (generated) { + Files.deleteIfExists(path); + } } @Benchmark public void readStrings(Blackhole bh) throws IOException { - try (VortexReader reader = VortexReader.open(file); - var iter = reader.scan(ScanOptions.columns(SYM))) { + try (VortexReader reader = VortexReader.open(path); + var iter = reader.scan(ScanOptions.columns(dictColumn))) { while (iter.hasNext()) { try (Chunk c = iter.next()) { - VarBinArray sym = c.column(SYM); - long n = sym.length(); + Array data = c.column(dictColumn); + VarBinArray strings = (VarBinArray) (data instanceof MaskedArray m ? m.inner() : data); + long n = strings.length(); for (long i = 0; i < n; i++) { - bh.consume(sym.getString(i)); + bh.consume(strings.getString(i)); } } } @@ -118,17 +151,27 @@ public void readStrings(Blackhole bh) throws IOException { @Benchmark public void sqlScan(Blackhole bh) throws SQLException { - query("select sym from vtx.data", bh); + query("select " + quote(column) + " from vtx.data", bh); } @Benchmark public void sqlGroupBy(Blackhole bh) throws SQLException { - query("select sym, count(*) from vtx.data group by sym", bh); + query("select " + quote(column) + ", count(*) from vtx.data group by " + quote(column), bh); } @Benchmark public void sqlFilter(Blackhole bh) throws SQLException { - query("select val from vtx.data where sym = 'NVDA'", bh); + query("select " + quote(agg) + " from vtx.data where " + quote(column) + " = '" + value + "'", bh); + } + + @Benchmark + public void sqlFilterSum(Blackhole bh) throws SQLException { + query("select sum(" + quote(agg) + ") from vtx.data where " + quote(column) + " = '" + value + "'", bh); + } + + /// Quotes an identifier for Calcite's JAVA lexer, so real files' column names may hold spaces. + private static String quote(String identifier) { + return "`" + identifier + "`"; } private void query(String sql, Blackhole bh) throws SQLException { @@ -143,23 +186,26 @@ private void query(String sql, Blackhole bh) throws SQLException { } } + /// Writes the synthetic fixture under the default `sym` / `val` column names. private static void writeFixture(Path file) throws IOException { + ColumnName sym = ColumnName.of("sym"); + ColumnName val = ColumnName.of("val"); DType.Struct schema = DType.structBuilder() - .field(SYM, DType.UTF8) - .field(VAL, DType.I64) + .field(sym, DType.UTF8) + .field(val, DType.I64) .build(); Random rng = new Random(42L); try (FileChannel ch = FileChannel.open(file, StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING); VortexWriter writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { for (int b = 0; b < TOTAL_ROWS / BATCH_SIZE; b++) { - String[] sym = new String[BATCH_SIZE]; - long[] val = new long[BATCH_SIZE]; + String[] syms = new String[BATCH_SIZE]; + long[] vals = new long[BATCH_SIZE]; for (int i = 0; i < BATCH_SIZE; i++) { - sym[i] = TICKERS[rng.nextInt(TICKERS.length)]; - val[i] = rng.nextInt(1_000); + syms[i] = TICKERS[rng.nextInt(TICKERS.length)]; + vals[i] = rng.nextInt(1_000); } - writer.writeChunk(Map.of(SYM, sym, VAL, val)); + writer.writeChunk(Map.of(sym, syms, val, vals)); } } } diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java index 75ceb0ccd..6f99be9d5 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinDictArray.java @@ -105,6 +105,23 @@ public PType dictCodesPType() { return dictCodesPType; } + /// Returns the dictionary values as an array, one element per entry, so a predicate can be + /// evaluated once per entry instead of once per row (Rust's `DictArray::values`). + /// + /// @return a zero-copy view over the dictionary entries + public VarBinOffsetArray values() { + long dictSize = Math.max(0, dictValOffsets.byteSize() / dictValOffPType.byteSize() - 1); + return new VarBinOffsetArray(dtype, dictSize, bytesSegment, dictValOffsets, dictValOffPType); + } + + /// Returns the dictionary code of row `i`: the index of its entry in [#values()]. + /// + /// @param i zero-based row index + /// @return the code, widened to a signed long; not checked against the dictionary size + public long code(long i) { + return dictReadCode(i); + } + @Override public byte[] getBytes(long i) { long code = dictReadCode(i); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/compute/Compute.java b/reader/src/main/java/io/github/dfa1/vortex/reader/compute/Compute.java index 89b60d7d7..ce9badda9 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/compute/Compute.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/compute/Compute.java @@ -6,6 +6,7 @@ import io.github.dfa1.vortex.reader.array.Array; import java.util.Objects; +import java.util.function.LongPredicate; /// The minimal, low-level entry point to the compute primitives of ADR 0013 — the fused, single-pass /// filter-and-aggregate kernels. @@ -21,7 +22,7 @@ /// aggregate column's `SUM` / `MIN` / `MAX` / non-null count over the rows it selects. /// [#matches(Chunk, RowFilter, long)] is the un-fused, per-row counterpart of the two: a plain /// boolean test with no aggregate at all, for a caller that needs to select rows rather than reduce -/// them. +/// them; [#matcher(Chunk, RowFilter)] lowers that test once per chunk for a caller testing many rows. /// /// Null handling follows the Rust reference: a filter excludes null positions (a null value makes a /// value predicate false), and the reduce skips them — `SUM` and `COUNT` over zero selected non-null @@ -96,6 +97,23 @@ public static boolean matches(Chunk chunk, RowFilter filter, long row) { return evaluate(chunk, filter, row); } + /// Lowers `filter` once for `chunk` and returns the row test, the form of + /// [#matches(Chunk, RowFilter, long)] for a caller that tests many rows of one chunk. + /// + /// The per-leaf work that does not depend on the row is done here rather than per row: a + /// numeric comparison resolves its bounds once, and a comparison over a dictionary-encoded + /// string column is evaluated once per dictionary entry, so each row costs a code lookup instead + /// of a string comparison. The selected rows are the same as [#matches(Chunk, RowFilter, long)]. + /// + /// @param chunk the decoded chunk holding every column `filter` references + /// @param filter the whole predicate: an n-ary `AND` of column-bound leaves, or a single leaf + /// @return a test of a zero-based row of `chunk`, valid while `chunk` is open + public static LongPredicate matcher(Chunk chunk, RowFilter filter) { + Objects.requireNonNull(chunk, "chunk"); + Objects.requireNonNull(filter, "filter"); + return FusedFilterAggregate.rowTest(chunk, filter); + } + private static boolean evaluate(Chunk chunk, RowFilter filter, long row) { return switch (filter) { case RowFilter.And(var parts) -> { diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/compute/FusedFilterAggregate.java b/reader/src/main/java/io/github/dfa1/vortex/reader/compute/FusedFilterAggregate.java index bcd7353e3..7459328b0 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/compute/FusedFilterAggregate.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/compute/FusedFilterAggregate.java @@ -8,6 +8,9 @@ import io.github.dfa1.vortex.reader.array.BoolArray; import io.github.dfa1.vortex.reader.array.DoubleArray; import io.github.dfa1.vortex.reader.array.FloatArray; +import io.github.dfa1.vortex.reader.array.VarBinDictArray; +import io.github.dfa1.vortex.reader.array.VarBinOffsetArray; +import io.github.dfa1.vortex.reader.array.VarBinSlicedArray; import java.util.ArrayList; import java.util.List; @@ -432,7 +435,68 @@ private static LongPredicate leafTest(Array column, Predicate predicate) { return longLeaf(data, validity, predicate); } } - return i -> PredicateEvaluator.evaluate(column, i, predicate); + LongPredicate generic = i -> PredicateEvaluator.evaluate(column, i, predicate); + if (data instanceof VarBinDictArray dict) { + return dictLeaf(dict, 0, dict.length(), validity, predicate, generic); + } + // A dictionary spanning many scan chunks (Rust's global dict layout) is decoded once and + // sliced per chunk, so real files mostly reach here as a slice. + if (data instanceof VarBinSlicedArray(var _, long rows, VarBinDictArray dict, long offset)) { + return dictLeaf(dict, offset, rows, validity, predicate, generic); + } + return generic; + } + + /// Builds a string dictionary leaf the way Rust's dict compare kernel does: the predicate is + /// evaluated once per dictionary entry, and a row tests its entry's result by code (#552). A + /// low-cardinality column then costs one code read per row instead of a string comparison. + /// + /// A null row still goes through `generic`, so a null-aware predicate (`IS NULL` under an `OR`) + /// answers exactly as before, and so does a code outside the dictionary, which `generic` reports + /// as malformed input. Like Rust, the lowering declines when the dictionary has more entries + /// than there are rows to test, where testing every entry costs more than testing the rows. + /// + /// @param dict the dictionary-encoded array the rows read from + /// @param offset the dictionary row of the first tested row (non-zero for a slice) + /// @param rows how many rows will be tested + /// @param validity the filter validity bitmap, or `null` when every row is valid + /// @param predicate the column-bound predicate + /// @param generic the per-row evaluator over the original (possibly masked) column + /// @return the lowered row predicate + private static LongPredicate dictLeaf(VarBinDictArray dict, long offset, long rows, BoolArray validity, + Predicate predicate, LongPredicate generic) { + VarBinOffsetArray values = dict.values(); + long size = values.length(); + if (size > rows || size > Integer.MAX_VALUE) { + return generic; + } + boolean[] entryMatches = new boolean[(int) size]; + for (int c = 0; c < size; c++) { + entryMatches[c] = PredicateEvaluator.evaluate(values, c, predicate); + } + return i -> { + if (validity != null && !validity.getBoolean(i)) { + return generic.test(i); + } + long code = dict.code(i + offset); + return code >= 0 && code < size ? entryMatches[(int) code] : generic.test(i); + }; + } + + /// Lowers a whole [RowFilter] over `chunk` into one row predicate: each leaf is lowered once by + /// [#leafTest(Array, Predicate)], and a row matches when every leaf accepts it. + /// + /// @param chunk the decoded chunk holding every column `filter` references + /// @param filter the whole predicate + /// @return a row predicate valid for `chunk`'s rows + static LongPredicate rowTest(Chunk chunk, RowFilter filter) { + List leafList = new ArrayList<>(); + collectLeaves(chunk, filter, leafList); + if (leafList.size() == 1) { + return leafList.getFirst(); + } + LongPredicate[] leaves = leafList.toArray(LongPredicate[]::new); + return i -> allMatch(leaves, i); } /// Builds a hoisted long-domain comparison leaf: the predicate lowers once to an inclusive range diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeArrays.java b/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeArrays.java index 28bcd2d6a..eff4c4fd9 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeArrays.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeArrays.java @@ -147,6 +147,23 @@ static VarBinArray utf8Array(Arena arena, String... values) { offsets, PType.I32); } + /// Builds a dictionary-encoded Utf8 [VarBinArray]: `pool` holds the entries and each row is an + /// I32 code into it, the shape the reader decodes a string dictionary to. + /// + /// @param arena the allocator for the off-heap buffers + /// @param pool the dictionary entries + /// @param codes the per-row codes into `pool` + /// @return a [VarBinArray] of [DType#UTF8] with `codes.length` rows + static VarBinArray utf8DictArray(Arena arena, String[] pool, int[] codes) { + VarBinOffsetArray values = (VarBinOffsetArray) utf8Array(arena, pool); + MemorySegment codeSegment = arena.allocate((long) Math.max(1, codes.length) * Integer.BYTES, Integer.BYTES); + for (int i = 0; i < codes.length; i++) { + codeSegment.setAtIndex(VortexFormat.LE_INT, i, codes[i]); + } + return VarBinArray.ofDict(DType.UTF8, codes.length, values.bytesSegment(), values.offsetsSegment(), + values.offsetsPtype(), codeSegment, PType.I32); + } + private static MemorySegment longSegment(Arena arena, long... values) { MemorySegment seg = arena.allocate(Math.max(1, values.length * 8L), 8); for (int i = 0; i < values.length; i++) { diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeTest.java index 2de1b6510..fcbbdf7d8 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/compute/ComputeTest.java @@ -1,15 +1,31 @@ package io.github.dfa1.vortex.reader.compute; +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.reader.Chunk; import io.github.dfa1.vortex.reader.RowFilter; import io.github.dfa1.vortex.reader.array.Array; +import io.github.dfa1.vortex.reader.array.MaskedArray; +import io.github.dfa1.vortex.reader.array.VarBinArray; +import io.github.dfa1.vortex.reader.array.VarBinSlicedArray; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; import java.lang.foreign.Arena; +import java.util.List; +import java.util.function.LongPredicate; +import java.util.stream.LongStream; +import java.util.stream.Stream; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatNullPointerException; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.junit.jupiter.api.Named.named; +import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; class ComputeTest { @@ -54,4 +70,109 @@ void filteredAggregateRejectsNullArguments() { assertThatNullPointerException() .isThrownBy(() -> Compute.filteredAggregate(chunk, null, ColumnName.of("v"))); } + + /// [Compute#matcher(Chunk, RowFilter)] lowers a filter once per chunk; over a string dictionary + /// it evaluates the predicate per dictionary entry instead of per row (#552). The per-row + /// [Compute#matches(Chunk, RowFilter, long)] is the oracle: both must select the same rows. + @Nested + class Matcher { + + private static final ColumnName SYM = ColumnName.of("sym"); + + /// An unsorted pool with a null row (row 3) whose code points at "café", so a lowering that + /// ignored validity would select row 3 for `eq` and miss it for `is null`. + private static Array nullableDict() { + Array dict = ComputeArrays.utf8DictArray(ARENA, new String[]{"café", "AAPL", "MSFT"}, + new int[]{0, 1, 2, 0, 1, 0}); + return new MaskedArray(dict, ComputeArrays.boolArray(ARENA, true, true, true, false, true, true)); + } + + static Stream predicates() { + return Stream.of( + Arguments.of(named("eq", new Predicate.Eq("café")), List.of(0L, 5L)), + Arguments.of(named("neq", new Predicate.Neq("café")), List.of(1L, 2L, 4L)), + Arguments.of(named("lt in UTF-8 order", new Predicate.Lt("café")), List.of(1L, 2L, 4L)), + Arguments.of(named("between", new Predicate.Between("AAPL", "MSFT")), List.of(1L, 2L, 4L)), + Arguments.of(named("no entry matches", new Predicate.Eq("TSLA")), List.of()), + Arguments.of(named("is null", new Predicate.IsNull()), List.of(3L)), + Arguments.of(named("is null or eq", new Predicate.Or(new Predicate.IsNull(), + new Predicate.Eq("AAPL"))), List.of(1L, 3L, 4L))); + } + + @ParameterizedTest + @MethodSource("predicates") + void selectsTheRowsMatchesSelects(Predicate predicate, List expected) { + // Given + Chunk chunk = mock(Chunk.class); + given(chunk.column(SYM)).willReturn(nullableDict()); + RowFilter filter = new RowFilter.Column(SYM, predicate); + + // When + LongPredicate sut = Compute.matcher(chunk, filter); + List result = LongStream.range(0, 6).filter(sut).boxed().toList(); + + // Then + assertThat(result).isEqualTo(expected) + .isEqualTo(LongStream.range(0, 6).filter(i -> Compute.matches(chunk, filter, i)).boxed().toList()); + } + + /// A dictionary spanning many scan chunks reaches the filter as a slice (Rust's global dict + /// layout, e.g. TPC-H `c_mktsegment`). The slice's rows must read the codes at its offset. + @Test + void slicedDictionary_testsCodesAtTheSliceOffset() { + // Given — rows 2..4 of the dictionary column: "b", "a", "b" + VarBinArray dict = ComputeArrays.utf8DictArray(ARENA, new String[]{"a", "b"}, new int[]{0, 0, 1, 0, 1}); + Chunk chunk = mock(Chunk.class); + given(chunk.column(SYM)).willReturn(new VarBinSlicedArray(DType.UTF8, 3, dict, 2)); + RowFilter filter = RowFilter.eq(SYM, "b"); + + // When + LongPredicate sut = Compute.matcher(chunk, filter); + List result = LongStream.range(0, 3).filter(sut).boxed().toList(); + + // Then + assertThat(result).containsExactly(0L, 2L) + .isEqualTo(LongStream.range(0, 3).filter(i -> Compute.matches(chunk, filter, i)).boxed().toList()); + } + + /// Rust's dict compare declines when the dictionary has more entries than there are rows; + /// the matcher then tests rows directly and must still select the right ones. + @Test + void dictionaryLargerThanRows_stillSelectsMatchingRows() { + // Given — 4 entries, 2 rows + Chunk chunk = mock(Chunk.class); + given(chunk.column(SYM)).willReturn(ComputeArrays.utf8DictArray(ARENA, + new String[]{"a", "b", "c", "d"}, new int[]{3, 1})); + + // When + LongPredicate sut = Compute.matcher(chunk, RowFilter.eq(SYM, "b")); + + // Then + assertThat(LongStream.range(0, 2).filter(sut).boxed().toList()).containsExactly(1L); + } + + /// A code past the dictionary is malformed input: it must fail as [VortexException], never as + /// an `ArrayIndexOutOfBoundsException` from the per-entry table (ADR 0003). + @Test + void codeOutsideDictionary_throwsVortexException() { + // Given — 2 entries, row 1 carries code 7 + Chunk chunk = mock(Chunk.class); + given(chunk.column(SYM)).willReturn(ComputeArrays.utf8DictArray(ARENA, + new String[]{"a", "b"}, new int[]{0, 7, 1})); + LongPredicate sut = Compute.matcher(chunk, RowFilter.eq(SYM, "b")); + + // When / Then + assertThatThrownBy(() -> sut.test(1)).isInstanceOf(VortexException.class); + } + + @Test + void rejectsNullArguments() { + // Given + Chunk chunk = mock(Chunk.class); + + // When / Then + assertThatNullPointerException().isThrownBy(() -> Compute.matcher(null, RowFilter.eq(SYM, "b"))); + assertThatNullPointerException().isThrownBy(() -> Compute.matcher(chunk, null)); + } + } }