diff --git a/CHANGELOG.md b/CHANGELOG.md index b9f741296..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)) @@ -18,6 +19,9 @@ 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)) +- 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)) - **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/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 ed8647ee7..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; @@ -311,6 +314,185 @@ 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); + } + + 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 new file mode 100644 index 000000000..986ab360b --- /dev/null +++ b/performance/src/main/java/io/github/dfa1/vortex/performance/DictStringBenchmark.java @@ -0,0 +1,212 @@ +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.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; +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.Param; +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: 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], [#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) +@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 = 4_000_000; + private static final int BATCH_SIZE = 50_000; + private static final String[] TICKERS = {"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA", "AVGO"}; + + /// 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 { + 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 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", path))); + } + + @TearDown(Level.Trial) + public void cleanup() throws IOException, SQLException { + connection.close(); + if (generated) { + Files.deleteIfExists(path); + } + } + + @Benchmark + public void readStrings(Blackhole bh) throws IOException { + try (VortexReader reader = VortexReader.open(path); + var iter = reader.scan(ScanOptions.columns(dictColumn))) { + while (iter.hasNext()) { + try (Chunk c = iter.next()) { + 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(strings.getString(i)); + } + } + } + } + } + + @Benchmark + public void sqlScan(Blackhole bh) throws SQLException { + query("select " + quote(column) + " from vtx.data", bh); + } + + @Benchmark + public void sqlGroupBy(Blackhole bh) throws SQLException { + query("select " + quote(column) + ", count(*) from vtx.data group by " + quote(column), bh); + } + + @Benchmark + public void sqlFilter(Blackhole bh) throws SQLException { + 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 { + 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)); + } + } + } + } + + /// 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) + .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[] syms = new String[BATCH_SIZE]; + long[] vals = new long[BATCH_SIZE]; + for (int i = 0; i < BATCH_SIZE; i++) { + syms[i] = TICKERS[rng.nextInt(TICKERS.length)]; + vals[i] = rng.nextInt(1_000); + } + writer.writeChunk(Map.of(sym, syms, val, vals)); + } + } + } +} 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..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 @@ -15,18 +15,112 @@ /// 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; + } + + /// 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) { @@ -41,7 +135,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/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/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 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)); + } + } }