From 435b1817080d72ab54f86906d4bf4a5c19ade98a Mon Sep 17 00:00:00 2001 From: Davide Angelocola Date: Mon, 19 Oct 2026 20:49:37 +0200 Subject: [PATCH] perf(calcite): rank string sort keys by bytes, walk reversed chunks backward MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ORDER BY ... LIMIT built a Java String for every row's string key just to compare it with the worst row kept, and almost every row loses. Now a row whose first key is a string is compared with that worst key's bytes first (VarBinArray#compareUnsigned): FSST decodes only up to the first differing byte (Decompressor#compareUnsigned), dictionary and offset arrays compare in place. A row that loses is skipped without a String; any other row takes the existing path, so results are unchanged by construction. Byte order equals String.compareTo order when the bound is ASCII: up to the first differing byte both strings decode the same, and a non-ASCII byte of the row, valid UTF-8 or not, decodes to a character above every ASCII one (a 5M-case fuzz found no disagreement). A non-ASCII bound falls back to building every row. A column stored in the opposite order to the query (eventIdentifier DESC on ascending ids) has every row beat the worst kept, so nothing was rejected. A chunk whose five evenly spaced sample rows rank in strictly reverse order is now walked backward: its best rows come first and the rest are rejected. A sample row the filter rejects is never read, as reading it can fail (a u64 past BIGINT), so such a chunk is walked forward. Walking backward admits a row equal to the worst, since equal keys rank by scan position. Walking forward goes through RowSelection#nextSelected, keeping #580's skipping of the rows a dictionary filter rejects. 100M-row cashflow file, jdbc:vortex: driver, warm JVM (median of the last 8 of 15 runs, 3 interleaved rounds), against main 9bc12092: order by eventIdentifier desc limit 10 5616 -> 1680 ms (3.3x) order by ISIN limit 10 1273 -> 962 ms (one round) Tests: calcite SortStringKeysTest covers ties across chunk edges walked both ways, a chunk under five rows, and a filtered-out sample row holding a u64 past BIGINT (it fails without the fix). The vortex-jni cross-check sorts Rust-written FSST, dictionary and plain strings over three Rust chunks, with "é" rows and ties at the bound, against Calcite's own sort. Co-Authored-By: Claude Sonnet 5 --- CHANGELOG.md | 1 + .../dfa1/vortex/calcite/VortexEnumerator.java | 31 +-- .../dfa1/vortex/calcite/VortexTable.java | 119 ++++++++++- .../vortex/calcite/SortStringKeysTest.java | 187 ++++++++++++++++++ docs/reference.md | 1 + .../github/dfa1/vortex/fsst/Decompressor.java | 41 ++++ .../dfa1/vortex/fsst/DecompressorTest.java | 40 ++++ .../CalciteInteropIntegrationTest.java | 101 +++++++++- .../reader/array/LazyFsstVarBinArray.java | 18 ++ .../dfa1/vortex/reader/array/VarBinArray.java | 13 ++ .../vortex/reader/array/VarBinArrays.java | 22 +++ .../vortex/reader/array/VarBinDictArray.java | 6 + .../reader/array/VarBinOffsetArray.java | 5 + .../reader/array/VarBinSlicedArray.java | 5 + .../vortex/reader/array/VarBinArrayTest.java | 47 +++++ .../decode/FsstEncodingDecoderTest.java | 59 ++++++ 16 files changed, 671 insertions(+), 25 deletions(-) create mode 100644 calcite/src/test/java/io/github/dfa1/vortex/calcite/SortStringKeysTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 90666030a..40e20fd52 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - **Breaking:** the `ScanIterator` constructor and the reader's Pco helpers (`PcoTansDecoder`, `PcoBin`, `LeBitReader`) are no longer public, and the built-in extension codecs are reached only through their `INSTANCE`. ([4ba6ea12](https://github.com/dfa1/vortex-java/commit/4ba6ea12)) +- SQL `ORDER BY ... LIMIT` on a string column is up to 3 times faster: rows that cannot make the top are ranked by their bytes without building a `String`, and a chunk stored in the opposite order is read backward. ([#575](https://github.com/dfa1/vortex-java/pull/575)) ## [0.18.0] — 2026-10-20 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 49850cc45..d006f553e 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 @@ -48,8 +48,7 @@ final class VortexEnumerator implements Enumerator { private final DType[] types; /// Per output column, its temporal conversion or `null`: resolved once, not per row. private final Temporal[] temporals; - private final RowFilter exactFilter; - /// The row test per chunk of `exactFilter`, or `null`: one per scan, so a shared dictionary is + /// The row test per chunk of the exact filter, or `null`: one per scan, so a shared dictionary is /// evaluated once. private final Function matcher; private final VortexReader reader; @@ -83,7 +82,6 @@ final class VortexEnumerator implements Enumerator { for (int c = 0; c < types.length; c++) { temporals[c] = Temporal.of(types[c]); } - this.exactFilter = exactFilter; this.matcher = exactFilter == null ? null : Compute.matcher(exactFilter); chunksScannedLastQuery.set(0); VortexReader openedReader = null; @@ -133,11 +131,10 @@ boolean advance() { return true; } - /// Moves to the next row the filter keeps in the current chunk, leaving the chunk open at its end - /// so a caller can still read rows of it by index ([#column(int, long)]). + /// Moves to the next row the filter keeps in the current chunk. /// /// @return `false` at the end of the chunk, or before the first [#nextChunk()] - boolean advanceInChunk() { + private boolean advanceInChunk() { if (chunk == null || rowInChunk >= chunkRows) { return false; } @@ -174,9 +171,15 @@ boolean nextChunk() { return true; } - /// The index in its chunk of the row [#advance()] moved to. - long rowInChunk() { - return currentRow; + /// The number of rows of the current chunk, before the filter. + long chunkRowCount() { + return chunkRows; + } + + /// The rows of the current chunk the filter keeps, or `null` when there is no filter and every + /// row is kept. + RowSelection selection() { + return rowMatcher; } /// Column `c` of the row [#advance()] moved to. @@ -184,13 +187,19 @@ Object column(int c) { return column(c, currentRow); } - /// Column `c` of row `r` of the current chunk. - Object column(int c, long r) { + /// Column `c` of the current chunk as decoded, read on first use. + Object array(int c) { Object array = columns[c]; if (array == null) { array = chunk.column(names[c]); columns[c] = array; } + return array; + } + + /// Column `c` of row `r` of the current chunk. + Object column(int c, long r) { + Object array = array(c); Temporal temporal = temporals[c]; return temporal == null ? value(array, types[c], r) : temporal.read(array, r); } 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 e0db8de46..f5073a4ea 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 @@ -52,6 +52,7 @@ import java.io.IOException; import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.util.ArrayList; import java.util.List; @@ -633,7 +634,9 @@ public record SortKey(int field, boolean descending, boolean nullsFirst) { /// read per row, and each value is compared as itself: the sort Calcite runs otherwise converts /// every number to a `BigDecimal` per comparison. A row is read in full only if it is still among /// the first `count` when its chunk ends, so a column sorted against the query, where every row - /// beats the last, costs `count` full rows per chunk instead of one per row. + /// beats the last, costs `count` full rows per chunk instead of one per row. Such a chunk is also + /// walked backward, best rows first, so the rest lose to the worst row kept; a string row that + /// loses on its first key is told by its bytes, without building its `String`. /// /// @param cols the table columns to emit, by position /// @param filter the rows to keep, or `null` for every row @@ -653,30 +656,52 @@ public List topN(int[] cols, RowFilter filter, List keys, int int capacity = count + Math.max(1024, Math.min(count, 1 << 20)); List kept = new ArrayList<>(); List fromChunk = new ArrayList<>(); - // the worst row kept once there are count of them: a row equal to it or worse cannot enter, as a - // later row never beats an equal key + SortKey first = keys.getFirst(); + boolean stringFirst = struct().fieldTypes().get(cols[first.field()]) instanceof DType.Utf8; + // the worst row kept once there are count of them, and its first key as bytes to reject a + // string row by without building it Object[] worst = null; + byte[] bound = null; Object[] probe = new Object[cols.length]; - long position = 0; + Object[] sample = new Object[cols.length]; + long chunkStart = 0; try (VortexEnumerator rows = enumerator(cols, filter)) { while (rows.nextChunk()) { - while (rows.advanceInChunk()) { + long n = rows.chunkRowCount(); + RowSelection selection = rows.selection(); + // a chunk sorted against the query is walked backward: its best rows come first, and + // every row after them loses to the bound; forward, the selection skips the rows the + // filter rejects a word at a time + boolean backward = sortedAgainst(rows, selection, keys, order, n, probe, sample); + for (long r = backward ? n - 1 : nextSelected(selection, 0, n); r >= 0; + r = backward ? r - 1 : nextSelected(selection, r + 1, n)) { + if (backward && selection != null && !selection.test(r) + || bound != null && firstKeyAfter(rows, first, r, bound)) { + continue; + } for (SortKey key : keys) { - probe[key.field()] = rows.column(key.field()); + probe[key.field()] = rows.column(key.field(), r); } - if (worst != null && order.compare(probe, worst) >= 0) { - position++; - continue; + // a row equal to the worst cannot enter when it comes later, but can when walking + // backward: equal keys rank by position + if (worst != null) { + int cmp = order.compare(probe, worst); + if (cmp > 0 || cmp == 0 && !backward) { + continue; + } } - Candidate candidate = new Candidate(probe.clone(), position++, rows.rowInChunk()); + Candidate candidate = new Candidate(probe.clone(), chunkStart + r, r); kept.add(candidate); fromChunk.add(candidate); if (kept.size() == capacity) { worst = cut(kept, ranking, count); + bound = stringFirst ? asciiBound(worst, first) : null; fromChunk.removeIf(c -> c.rowInChunk < 0); } } + chunkStart += n; worst = cut(kept, ranking, count); + bound = stringFirst ? asciiBound(worst, first) : null; // the survivors are read in full while their chunk is still open for (Candidate c : fromChunk) { if (c.rowInChunk >= 0) { @@ -692,6 +717,80 @@ public List topN(int[] cols, RowFilter filter, List keys, int return kept.stream().map(c -> c.row).toList(); } + /// The first row of `selection` at or after `from` among `n`, every row when it is `null`. + /// + /// @return the row, or `-1` past the last + private static long nextSelected(RowSelection selection, long from, long n) { + if (selection != null) { + return selection.nextSelected(from); + } + return from < n ? from : -1; + } + + /// Whether the current chunk looks sorted against `order`: five evenly spaced rows each rank + /// strictly before the one above them. Walking a chunk backward gains nothing on rows in no + /// order, so those must almost never pass (one chunk in 120). A sample row the filter rejects + /// is never read, as reading it may fail (a `u64` past `BIGINT`), so the chunk is walked forward. + /// + /// @param selection the rows the filter keeps, or `null` for every row + /// @param a scratch row for the sampled keys + /// @param b another scratch row + private static boolean sortedAgainst(VortexEnumerator rows, RowSelection selection, List keys, + java.util.Comparator order, long n, Object[] a, Object[] b) { + // ponytail: a sample, not a scan; a chunk sorted only in parts, or whose sample the filter + // cuts, is walked forward + if (n < 5) { + return false; + } + Object[] above = a; + Object[] below = b; + for (int s = 0; s <= 4; s++) { + long r = s * (n - 1) / 4; + if (selection != null && !selection.test(r)) { + return false; + } + for (SortKey key : keys) { + below[key.field()] = rows.column(key.field(), r); + } + if (s > 0 && order.compare(below, above) >= 0) { + return false; + } + Object[] swap = above; + above = below; + below = swap; + } + return true; + } + + /// The first key of `worst` as bytes, when it is an ASCII string. Comparing a row's UTF-8 bytes + /// with ASCII bytes gives [String#compareTo]'s answer: they agree up to the first differing byte, + /// where a non-ASCII byte of the row, valid or not, decodes to a character above every ASCII one. + /// + /// @return the bytes, or `null` when there is no worst row or its key is null or not ASCII + private static byte[] asciiBound(Object[] worst, SortKey first) { + // ponytail: a non-ASCII bound falls back to building every row; beyond ASCII, byte order and + // String order disagree on supplementary characters, which a fix-up at the first differing + // byte would handle + if (worst != null && worst[first.field()] instanceof String s && s.chars().allMatch(c -> c < 0x80)) { + return s.getBytes(StandardCharsets.US_ASCII); + } + return null; + } + + /// Whether row `r`'s first key, a string, ranks strictly after `bound`, read from its bytes. A null + /// key answers `false`, leaving the row to the full comparison. + private static boolean firstKeyAfter(VortexEnumerator rows, SortKey first, long r, byte[] bound) { + Object array = rows.array(first.field()); + if (array instanceof MaskedArray masked) { + if (!masked.isValid(r)) { + return false; + } + array = masked.inner(); + } + int cmp = ((VarBinArray) array).compareUnsigned(r, bound); + return first.descending() ? cmp < 0 : cmp > 0; + } + /// Sorts `kept` and drops all but the first `count`, marking the dropped ones so they are never read. /// /// @return the worst row kept when there are `count` of them, else `null` diff --git a/calcite/src/test/java/io/github/dfa1/vortex/calcite/SortStringKeysTest.java b/calcite/src/test/java/io/github/dfa1/vortex/calcite/SortStringKeysTest.java new file mode 100644 index 000000000..5b3676e73 --- /dev/null +++ b/calcite/src/test/java/io/github/dfa1/vortex/calcite/SortStringKeysTest.java @@ -0,0 +1,187 @@ +package io.github.dfa1.vortex.calcite; + +import io.github.dfa1.vortex.core.model.ColumnName; +import io.github.dfa1.vortex.core.model.DType; +import io.github.dfa1.vortex.core.model.PType; +import io.github.dfa1.vortex.reader.ScanOptions; +import io.github.dfa1.vortex.reader.VortexReader; +import io.github.dfa1.vortex.writer.VortexWriter; +import io.github.dfa1.vortex.writer.WriteOptions; + +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +/// The `ORDER BY ... LIMIT` top-N walks a chunk sorted against the query backward and skips string +/// rows by their bytes. Each query is checked against Calcite's own sort of the same rows (the key +/// wrapped in a `CASE` the push-down rule cannot take), or against the rows worked out by hand. +class SortStringKeysTest { + + // written with the legacy zone map, which keeps each batch a chunk of its own: default writes + // coalesce small batches into one chunk + private static final int BATCH_ROWS = 1_000; + + @TempDir + static Path tmp; + private static Path events; + /// The first row of the events file's second chunk. + private static long edge; + private static Path tiny; + private static Path unsigned; + + @BeforeAll + static void writeFiles() throws Exception { + // `t` ascends with the rows in runs of seven equal values; 1,000 rows a chunk is not a + // multiple of 7, so a run straddles every chunk edge + events = tmp.resolve("events.vortex"); + DType.Struct schema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("t"), DType.UTF8) + .build(); + try (var ch = FileChannel.open(events, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = VortexWriter.create(ch, schema, LegacyZoneMaps.OPTIONS)) { + for (int batch = 0; batch < 3; batch++) { + long[] id = new long[BATCH_ROWS]; + String[] t = new String[BATCH_ROWS]; + for (int i = 0; i < BATCH_ROWS; i++) { + id[i] = (long) batch * BATCH_ROWS + i; + t[i] = "K%05d".formatted(id[i] / 7); + } + writer.writeChunk(Map.of(ColumnName.of("id"), id, ColumnName.of("t"), t)); + } + } + try (VortexReader reader = VortexReader.open(events); + var scan = reader.scan(ScanOptions.all())) { + edge = scan.chunkRowCounts()[0]; + } + // fewer rows than the five a direction sample reads + tiny = tmp.resolve("tiny.vortex"); + try (var ch = FileChannel.open(tiny, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = VortexWriter.create(ch, schema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(ColumnName.of("id"), new long[]{0, 1, 2}, + ColumnName.of("t"), new String[]{"a", "b", "c"})); + } + // rows 0..4 hold u64 values past BIGINT, which fail when read; rows 5..9 hold 95..91 + unsigned = tmp.resolve("unsigned.vortex"); + DType.Struct unsignedSchema = DType.structBuilder() + .field(ColumnName.of("id"), DType.I64) + .field(ColumnName.of("u"), new DType.Primitive(PType.U64, false)) + .build(); + long[] u = new long[10]; + for (int i = 0; i < 10; i++) { + u[i] = i < 5 ? Long.MIN_VALUE + i : 100 - i; + } + try (var ch = FileChannel.open(unsigned, StandardOpenOption.CREATE, StandardOpenOption.WRITE); + var writer = VortexWriter.create(ch, unsignedSchema, WriteOptions.defaults())) { + writer.writeChunk(Map.of(ColumnName.of("id"), new long[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, + ColumnName.of("u"), u)); + } + } + + @Test + void eventsFile_spansSeveralChunks() throws Exception { + // Given + long[] result; + + // When + try (VortexReader reader = VortexReader.open(events); + var scan = reader.scan(ScanOptions.all())) { + result = scan.chunkRowCounts(); + } + + // Then — the queries below cross chunk edges only if the writer kept the chunks apart + assertThat(result).hasSizeGreaterThan(1); + } + + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + // sorted against the query: every chunk walked backward + "order by t desc limit 15 | order by case when id > -1 then t end desc limit 15", + // `EDGE` is the first row of the second chunk. The top rows end just past it, so the run of + // seven tied values around it is split between a chunk walked backward (its sample all + // kept) and one walked forward (its sample cut by the filter) + "where id < EDGE + 3 order by t desc limit 10 | where id < EDGE + 3 order by case when id > -1 then t end desc limit 10", + // sorted with the query: walked forward, from just before the edge, across it + "where id >= EDGE - 5 order by t limit 9 | where id >= EDGE - 5 order by case when id > -1 then t end limit 9", + "where id >= EDGE - 20 order by t desc, id limit 12 offset 3 | where id >= EDGE - 20 order by case when id > -1 then t end desc, id limit 12 offset 3"}) + void orderByStringLimit_returnsTheRowsCalciteSortReturns(String tail, String oracleTail) throws Exception { + // Given + assertThat(edge).as("the events file holds several chunks").isLessThan(3L * BATCH_ROWS); + String sql = "select id, t from vtx.t " + tail.replace("EDGE", Long.toString(edge)); + String oracle = "select id, t from vtx.t " + oracleTail.replace("EDGE", Long.toString(edge)); + + // When + List result = rows(events, sql); + + // Then + assertThat(plan(events, sql)).contains("Values").doesNotContain("TableScan"); + assertThat(result).isEqualTo(rows(events, oracle)).isNotEmpty(); + } + + @Test + void chunkSmallerThanTheSample_isWalkedForward() throws Exception { + // Given + String sql = "select id, t from vtx.t order by t desc limit 2"; + + // When + List result = rows(tiny, sql); + + // Then + assertThat(result).containsExactly("[2, c]", "[1, b]"); + } + + @Test + void sampleRowTheFilterRejects_isNeverRead() throws Exception { + // Given — sampling a filtered-out row read a u64 past BIGINT, which fails, although the + // filter keeps only rows 5..9 + String sql = "select id, u from vtx.t where id >= 5 order by u limit 3"; + + // When + List result = rows(unsigned, sql); + + // Then + assertThat(plan(unsigned, sql)).contains("Values"); + assertThat(result).containsExactly("[9, 91]", "[8, 92]", "[7, 93]"); + } + + private static String plan(Path file, String sql) throws Exception { + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery("explain plan for " + sql)) { + rs.next(); + return rs.getString(1); + } + } + + private static List rows(Path file, String sql) throws Exception { + List out = new ArrayList<>(); + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file)); + Statement st = conn.createStatement(); + ResultSet rs = st.executeQuery(sql)) { + int columns = rs.getMetaData().getColumnCount(); + while (rs.next()) { + Object[] row = new Object[columns]; + for (int c = 0; c < columns; c++) { + row[c] = rs.getObject(c + 1); + } + out.add(Arrays.toString(row)); + } + } + return out; + } +} diff --git a/docs/reference.md b/docs/reference.md index ae82dab10..1f5dc20ff 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -440,6 +440,7 @@ Decodes an FSST code stream against a trained table (parallel arrays indexed by | `static final int ESCAPE` | The escape code (`0xFF`) | | `decompress(byte[] in, int start, int end, byte[] out, long outPos)` | Decode; size `out` ≥ `8 * (end - start)` | | `decompress(MemorySegment in, long start, long end, MemorySegment out, long outPos)` | Unconditional-8-byte-store decode; caller MUST allocate `out` with ≥ 7 bytes of trailing slack | +| `compareUnsigned(MemorySegment in, long start, long end, byte[] other)` | Compare the decoded bytes with `other` (unsigned), decoding only up to the first difference | ### `Symbol` diff --git a/fsst/src/main/java/io/github/dfa1/vortex/fsst/Decompressor.java b/fsst/src/main/java/io/github/dfa1/vortex/fsst/Decompressor.java index 5d17d80c3..0fe9b2871 100644 --- a/fsst/src/main/java/io/github/dfa1/vortex/fsst/Decompressor.java +++ b/fsst/src/main/java/io/github/dfa1/vortex/fsst/Decompressor.java @@ -149,4 +149,45 @@ public long decompress(MemorySegment compressed, long start, long end, byte[] ou } return outIndex; } + + /// Compares the decoded bytes of the compressed range `[start, end)` with `other`, as + /// `Arrays.compareUnsigned` would compare the two arrays, decoding + /// only up to the first byte that differs: two strings usually differ within their first symbols, + /// so ranking a row against a bound costs a symbol or two instead of the whole row. + /// + /// Like [#decompress(MemorySegment, long, long, MemorySegment, long)] this trusts the code + /// stream: a trailing escape reads past `end`, and a code outside the table indexes past it. + /// + /// @param compressed the compressed code stream + /// @param start the index of the first code to decode, inclusive + /// @param end the index one past the last code to decode, exclusive + /// @param other the bytes to compare with + /// @return a negative, zero, or positive integer as the decoded bytes sort before, equal to, or + /// after `other` + public int compareUnsigned(MemorySegment compressed, long start, long end, byte[] other) { + long pos = start; + int at = 0; + while (pos < end) { + int code = Byte.toUnsignedInt(compressed.get(ValueLayout.JAVA_BYTE, pos++)); + long packed; + int length; + if (code == ESCAPE) { + packed = Byte.toUnsignedLong(compressed.get(ValueLayout.JAVA_BYTE, pos++)); + length = 1; + } else { + packed = packedSymbols[code]; + length = lengths[code]; + } + for (int k = 0; k < length; k++, at++) { + if (at == other.length) { + return 1; + } + int cmp = Integer.compare((int) (packed >>> (k * 8)) & 0xFF, Byte.toUnsignedInt(other[at])); + if (cmp != 0) { + return cmp; + } + } + } + return at == other.length ? 0 : -1; + } } diff --git a/fsst/src/test/java/io/github/dfa1/vortex/fsst/DecompressorTest.java b/fsst/src/test/java/io/github/dfa1/vortex/fsst/DecompressorTest.java index 8914e0e7b..fe76331e6 100644 --- a/fsst/src/test/java/io/github/dfa1/vortex/fsst/DecompressorTest.java +++ b/fsst/src/test/java/io/github/dfa1/vortex/fsst/DecompressorTest.java @@ -2,7 +2,10 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import java.lang.foreign.MemorySegment; import java.nio.charset.StandardCharsets; import java.util.Arrays; @@ -116,4 +119,41 @@ void decompress_sequentialRows_doNotCorruptEachOther() { assertThat(Arrays.copyOfRange(out, 2, 3)).isEqualTo("c".getBytes(StandardCharsets.UTF_8)); } } + + @Nested + class CompareUnsigned { + + // "ab" as one symbol, an escaped 0xC3, then "c": the escaped byte is negative as a Java + // byte, so a signed comparison would rank it below 'z' instead of above + private static final byte[] DECODED = {'a', 'b', (byte) 0xC3, 'c'}; + + @ParameterizedTest + @ValueSource(strings = {"abÃc", "ab", "abÃcd", "a", "ac", "abz", "abÄ", "", "b"}) + void compareUnsigned_agreesWithArraysCompareUnsignedOfTheDecodedBytes(String other) { + // Given — "Ã" is the single byte 0xC3 and "Ä" 0xC4 in ISO-8859-1, so each case is a + // prefix, an extension, or a first difference inside the symbol, at the escape, or after + Symbol ab = Symbol.of("ab".getBytes(StandardCharsets.UTF_8), 0, 2); + Decompressor sut = Decompressor.of(new long[]{ab.packedBytes(), 'c'}, new int[]{2, 1}); + MemorySegment compressed = MemorySegment.ofArray(new byte[]{0x00, (byte) Decompressor.ESCAPE, (byte) 0xC3, 0x01}); + byte[] otherBytes = other.getBytes(StandardCharsets.ISO_8859_1); + + // When + int result = sut.compareUnsigned(compressed, 0, compressed.byteSize(), otherBytes); + + // Then + assertThat(Integer.signum(result)).isEqualTo(Integer.signum(Arrays.compareUnsigned(DECODED, otherBytes))); + } + + @Test + void compareUnsigned_emptyRange_sortsBeforeAnyNonEmptyBytes() { + // Given — an empty string compresses to no codes at all + Decompressor sut = Decompressor.of(new long[0], new int[0]); + + // When + int result = sut.compareUnsigned(MemorySegment.ofArray(new byte[0]), 0, 0, new byte[]{0}); + + // Then + assertThat(result).isNegative(); + } + } } diff --git a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java index 1c791cc56..e493ece78 100644 --- a/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java +++ b/integration/src/test/java/io/github/dfa1/vortex/integration/CalciteInteropIntegrationTest.java @@ -6,6 +6,9 @@ import dev.vortex.jni.NativeLoader; import io.github.dfa1.vortex.calcite.VortexCalcite; import io.github.dfa1.vortex.calcite.VortexTable; +import io.github.dfa1.vortex.reader.ScanIterator; +import io.github.dfa1.vortex.reader.ScanOptions; +import io.github.dfa1.vortex.reader.VortexReader; import org.apache.arrow.c.ArrowArray; import org.apache.arrow.c.ArrowSchema; import org.apache.arrow.c.Data; @@ -17,6 +20,7 @@ import org.apache.arrow.vector.TimeStampMicroTZVector; import org.apache.arrow.vector.TimeStampMicroVector; import org.apache.arrow.vector.VarBinaryVector; +import org.apache.arrow.vector.VarCharVector; import org.apache.arrow.vector.VectorSchemaRoot; import org.apache.arrow.vector.types.DateUnit; import org.apache.arrow.vector.types.TimeUnit; @@ -29,6 +33,7 @@ import org.junit.jupiter.params.provider.CsvSource; import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.sql.Connection; import java.sql.ResultSet; @@ -52,6 +57,8 @@ class CalciteInteropIntegrationTest { // several batches, so the row count is a sum over chunks, not a single chunk's length private static final int BATCHES = 3; private static final int BATCH_ROWS = 1_000; + // over 1 MiB a batch, the least Rust writes as a chunk of its own + private static final int EVENT_BATCH_ROWS = 20_000; private static final LocalDate DATE = LocalDate.of(2024, 2, 29); private static final LocalTime TIME = LocalTime.of(13, 45, 30); // pre-epoch on purpose: millisecond truncation must floor, not round toward zero @@ -240,7 +247,7 @@ void jniWriter_minMaxOfStringsOutsideLatin1_answer(@TempDir Path tmp) throws Exc org.apache.arrow.vector.VarCharVector s = (org.apache.arrow.vector.VarCharVector) root.getVector("s"); s.allocateNew(values.length); for (int i = 0; i < values.length; i++) { - s.setSafe(i, values[i].getBytes(java.nio.charset.StandardCharsets.UTF_8)); + s.setSafe(i, values[i].getBytes(StandardCharsets.UTF_8)); } root.setRowCount(values.length); try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); @@ -318,6 +325,43 @@ void jniWriter_orderByLimit_returnsTheRowsCalciteSortReturns(String tail, String } } + @ParameterizedTest(name = "{0}") + @CsvSource(delimiter = '|', value = { + // e ascends with the rows, so a descending sort walks each chunk backward and rejects by bytes + "order by e desc limit 10 | order by case when id > -1 then e end desc limit 10", + "order by e limit 7 offset 3 | order by case when id > -1 then e end limit 7 offset 3", + "order by e desc nulls first limit 5 | order by case when id > -1 then e end desc nulls first limit 5", + // t repeats each value ten times, so the rows tied at the bound must keep their scan order + // when walked backward + "order by t desc limit 15 | order by case when id > -1 then t end desc limit 15", + "where id > 500 order by r desc, id limit 20 | where id > 500 order by case when id > -1 then r end desc, id limit 20", + "order by r limit 12 | order by case when id > -1 then r end limit 12"}) + void jniWriter_orderByStringLimit_returnsTheRowsCalciteSortReturns(String tail, String oracleTail, + @TempDir Path tmp) throws Exception { + // Given — Rust-written high-cardinality strings, some ending in "é" so a row's first differing + // byte can be above 0x7F; the top-N ranks rows by their bytes against an ASCII bound and must + // agree with the String order Calcite sorts the oracle by + Path file = tmp.resolve("jni_sort_strings.vortex"); + writeEventsWithJni(file); + try (VortexReader reader = VortexReader.open(file); + ScanIterator scan = reader.scan(ScanOptions.all())) { + assertThat(scan.chunkRowCounts()).as("Rust wrote several chunks, so ties cross chunk edges").hasSizeGreaterThan(1); + } + String select = "select id, e, t, r from vtx.t "; + + try (Connection conn = VortexCalcite.connect("vtx", Map.of("t", file))) { + // When + List result = sqlRows(conn, select + tail); + + // Then + try (Statement st = conn.createStatement(); ResultSet rs = st.executeQuery("explain plan for " + select + tail)) { + rs.next(); + assertThat(rs.getString(1)).containsIgnoringCase("Values").doesNotContain("TableScan"); + } + assertThat(result).isEqualTo(sqlRows(conn, select + oracleTail)).isNotEmpty(); + } + } + @ParameterizedTest(name = "{0}") @CsvSource(delimiter = '|', value = { "select distinct x from vtx.t order by x desc limit 5 " @@ -414,7 +458,7 @@ void jniWriter_aggregateWhereOnSmallint_isPushedDownAndEqualsTheScan(String wher s.allocateNew(BATCH_ROWS); i16.allocateNew(BATCH_ROWS); for (int i = 0; i < BATCH_ROWS; i++) { - s.setSafe(i, symbols[i % symbols.length].getBytes(java.nio.charset.StandardCharsets.UTF_8)); + s.setSafe(i, symbols[i % symbols.length].getBytes(StandardCharsets.UTF_8)); i16.set(i, (short) (i % 20 - 10)); } root.setRowCount(BATCH_ROWS); @@ -510,7 +554,7 @@ void jniWriter_decimalAggregates_arePushedDownAndEqualTheScan(String sql, String for (int i = 0; i < BATCH_ROWS; i++) { long id = (long) batch * BATCH_ROWS + i; ids.set(i, id); - s.setSafe(i, symbols[(int) (id % symbols.length)].getBytes(java.nio.charset.StandardCharsets.UTF_8)); + s.setSafe(i, symbols[(int) (id % symbols.length)].getBytes(StandardCharsets.UTF_8)); if (id % 13 == 0) { amount.setNull(i); } else { @@ -637,7 +681,7 @@ private static List writeGroupsWithJni(Path file, String[] symbols) th if (symbol == null) { s.setNull(i); } else { - s.setSafe(i, symbol.getBytes(java.nio.charset.StandardCharsets.UTF_8)); + s.setSafe(i, symbol.getBytes(StandardCharsets.UTF_8)); } ids.set(i, id); if (nValue == null) { @@ -714,6 +758,55 @@ void jniWriter_whereOnNarrowIntegers_selectsTheMatchingRows(String where, long e assertThat(result).isEqualTo(expected); } + /// Writes `id`, `e` (an ascending event id, "é"-suffixed every 13th row, null every 97th), `t` + /// (an ascending value repeated ten times) and `r` (a scrambled value with repeats) in [#BATCHES] + /// batches of [#EVENT_BATCH_ROWS], over 1 MiB each, so Rust writes several chunks: Rust coalesces + /// smaller batches into one. Its chunks hold a multiple of 8,192 rows, not of ten, so a run of + /// equal `t` straddles each chunk edge. + private static void writeEventsWithJni(Path file) throws Exception { + Schema schema = new Schema(List.of( + Field.notNullable("id", new ArrowType.Int(64, true)), + Field.nullable("e", new ArrowType.Utf8()), + Field.notNullable("t", new ArrowType.Utf8()), + Field.notNullable("r", new ArrowType.Utf8()))); + try (VortexWriter writer = VortexWriter.builder(SESSION, file.toAbsolutePath().toUri().toString(), schema, + ALLOCATOR).build()) { + for (int batch = 0; batch < BATCHES; batch++) { + try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, ALLOCATOR)) { + BigIntVector ids = (BigIntVector) root.getVector("id"); + VarCharVector e = (VarCharVector) root.getVector("e"); + VarCharVector t = (VarCharVector) root.getVector("t"); + VarCharVector r = (VarCharVector) root.getVector("r"); + ids.allocateNew(EVENT_BATCH_ROWS); + e.allocateNew(EVENT_BATCH_ROWS); + t.allocateNew(EVENT_BATCH_ROWS); + r.allocateNew(EVENT_BATCH_ROWS); + for (int i = 0; i < EVENT_BATCH_ROWS; i++) { + long id = (long) batch * EVENT_BATCH_ROWS + i; + ids.set(i, id); + if (id % 97 == 50) { + e.setNull(i); + } else { + e.setSafe(i, utf8("EVT%08d%s".formatted(id, id % 13 == 0 ? "é" : ""))); + } + t.setSafe(i, utf8("T%06d".formatted(id / 10))); + r.setSafe(i, utf8("K%05d".formatted(id * 7919 % 1009))); + } + root.setRowCount(EVENT_BATCH_ROWS); + try (ArrowArray array = ArrowArray.allocateNew(ALLOCATOR); + ArrowSchema arrowSchema = ArrowSchema.allocateNew(ALLOCATOR)) { + Data.exportVectorSchemaRoot(ALLOCATOR, root, null, array, arrowSchema); + writer.writeBatch(array.memoryAddress(), arrowSchema.memoryAddress()); + } + } + } + } + } + + private static byte[] utf8(String s) { + return s.getBytes(StandardCharsets.UTF_8); + } + private static void writeIdsWithJni(Path file, int batches, int batchRows) throws Exception { Schema schema = new Schema(List.of( Field.notNullable("id", new ArrowType.Int(64, true)), diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java index 898e899b6..ed98864e0 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/LazyFsstVarBinArray.java @@ -120,6 +120,24 @@ public String getString(long i) { return new String(getBytes(i), StandardCharsets.UTF_8); } + /// Decodes row `i` only up to its first byte that differs from `other`, so the claimed length + /// is not cross-checked as in [#getBytes(long)]: a row that never needs reading in full is + /// trusted the way Rust's compressed-domain compare trusts it. + @Override + public int compareUnsigned(long i, byte[] other) { + Objects.checkIndex(i, length); + CodeRange range = codeRange(i); + try { + return decompressor.compareUnsigned(compressedBytes, range.start(), range.end(), other); + } catch (IndexOutOfBoundsException e) { + // an escape at the very end of the code buffer, or a code past the trained table; an + // escape ending a row mid-buffer reads the next row's first code, as getBytes does + throw new VortexException(EncodingId.VORTEX_FSST, + "row " + i + " code range [" + range.start() + ", " + range.end() + + ") decodes past its bounds or references an unknown symbol code", e); + } + } + @Override public int getByteLength(long i) { Objects.checkIndex(i, length); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java index d79ebf090..3e79e34c1 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArray.java @@ -6,6 +6,7 @@ import java.lang.foreign.MemorySegment; import java.lang.foreign.SegmentAllocator; +import java.util.Arrays; import java.util.Optional; import java.util.function.IntConsumer; @@ -65,6 +66,18 @@ default Optional segmentIfPresent() { /// @return the UTF-8 string at position `i` String getString(long i); + /// Compares the raw bytes of element `i` with `other` as `Arrays.compareUnsigned` + /// does, without building a `String`: ranking many rows against one bound, as a top-N does, then + /// allocates nothing for the rows that lose. + /// + /// @param i zero-based logical index (must be in `[0, length)`) + /// @param other the bytes to compare with + /// @return a negative, zero, or positive integer as element `i` sorts before, equal to, or after + /// `other` + default int compareUnsigned(long i, byte[] other) { + return Arrays.compareUnsigned(getBytes(i), other); + } + /// Returns the byte length of element `i` without copying the data. /// /// @param i zero-based logical index (must be in `[0, length)`) diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArrays.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArrays.java index a9b8e6dc8..2d01d7b34 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArrays.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinArrays.java @@ -3,6 +3,7 @@ import io.github.dfa1.vortex.core.error.VortexException; import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; /// Package-private helpers shared by the `VarBin*Array` implementations of [VarBinArray]. final class VarBinArrays { @@ -32,4 +33,25 @@ static int checkedLength(MemorySegment bytes, long start, long end) { } return (int) len; } + + /// [VarBinArray#compareUnsigned(long, byte[])] of the element at `[start, end)` of `bytes`, read + /// in place. + /// + /// @param bytes data buffer the offsets index into + /// @param start start offset of the element + /// @param end end offset of the element, exclusive + /// @param other the bytes to compare with + /// @return a negative, zero, or positive integer as the element sorts before, equal to, or after + /// `other` + static int compareUnsigned(MemorySegment bytes, long start, long end, byte[] other) { + int len = checkedLength(bytes, start, end); + long at = MemorySegment.mismatch(bytes, start, start + len, MemorySegment.ofArray(other), 0, other.length); + if (at < 0) { + return 0; + } + if (at == len || at == other.length) { + return Integer.compare(len, other.length); + } + return Byte.compareUnsigned(bytes.get(ValueLayout.JAVA_BYTE, start + at), other[(int) at]); + } } 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 6f99be9d5..9451ae66e 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 @@ -133,6 +133,12 @@ public byte[] getBytes(long i) { return out; } + @Override + public int compareUnsigned(long i, byte[] other) { + long code = dictReadCode(i); + return VarBinArrays.compareUnsigned(bytesSegment, dictReadOff(code), dictReadOff(code + 1), other); + } + @Override public String getString(long i) { long code = dictReadCode(i); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinOffsetArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinOffsetArray.java index 6292b65b0..0f250a335 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinOffsetArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinOffsetArray.java @@ -33,6 +33,11 @@ public byte[] getBytes(long i) { return out; } + @Override + public int compareUnsigned(long i, byte[] other) { + return VarBinArrays.compareUnsigned(bytesSegment, readOffset(i), readOffset(i + 1), other); + } + @Override public String getString(long i) { return new String(getBytes(i), StandardCharsets.UTF_8); diff --git a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinSlicedArray.java b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinSlicedArray.java index f99e9c546..977d2a2d4 100644 --- a/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinSlicedArray.java +++ b/reader/src/main/java/io/github/dfa1/vortex/reader/array/VarBinSlicedArray.java @@ -43,6 +43,11 @@ public String getString(long i) { return inner.getString(i + offset); } + @Override + public int compareUnsigned(long i, byte[] other) { + return inner.compareUnsigned(i + offset, other); + } + @Override public int getByteLength(long i) { return inner.getByteLength(i + offset); 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 2eedc05dd..8e45a0175 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 @@ -7,6 +7,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.EnumSource; +import org.junit.jupiter.params.provider.MethodSource; import java.lang.foreign.Arena; import java.lang.foreign.MemorySegment; @@ -14,6 +15,7 @@ import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; @@ -366,6 +368,51 @@ private static MemorySegment packOffsets(int[] offs, PType ptype) { } + @Nested + class CompareUnsigned { + + // "é" is 0xC3 0xA9 in UTF-8: negative as a Java byte, so a signed comparison ranks it below 'z' + private static final String[] VALUES = {"abc", "ab", "abcd", "abé", "", "b"}; + + static List arrays() { + int[] codes = {0, 1, 2, 3, 4, 5}; + return List.of(of(VALUES), Dict.ofDict(VALUES, codes), + new VarBinSlicedArray(UTF8, VALUES.length, of("x", "abc", "ab", "abcd", "abé", "", "b"), 1)); + } + + @ParameterizedTest + @MethodSource("arrays") + void compareUnsigned_agreesWithArraysCompareUnsignedOfGetBytes(VarBinArray sut) { + // Given — every row against every value: equal, prefix, extension, and a first difference at + // a byte above 0x7F + List expected = new ArrayList<>(); + List result = new ArrayList<>(); + + // When + for (long i = 0; i < sut.length(); i++) { + for (String other : VALUES) { + byte[] otherBytes = other.getBytes(StandardCharsets.UTF_8); + result.add(Integer.signum(sut.compareUnsigned(i, otherBytes))); + expected.add(Integer.signum(Arrays.compareUnsigned(sut.getBytes(i), otherBytes))); + } + } + + // Then + assertThat(result).isEqualTo(expected); + } + + @Test + void compareUnsigned_offsetsPastTheData_throwVortexException() { + // Given — row 0 claims bytes [0, 99) of a 3-byte buffer + VarBinArray sut = new VarBinOffsetArray(UTF8, 1, MemorySegment.ofArray("abc".getBytes(StandardCharsets.UTF_8)), + MemorySegment.ofArray(new byte[]{0, 0, 0, 0, 99, 0, 0, 0}), PType.I32); + + // When / Then + assertThatThrownBy(() -> sut.compareUnsigned(0, new byte[]{'a'})) + .isInstanceOf(VortexException.class); + } + } + @Nested class Constant { diff --git a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java index c71f2c313..68c2d0568 100644 --- a/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java +++ b/reader/src/test/java/io/github/dfa1/vortex/reader/decode/FsstEncodingDecoderTest.java @@ -418,6 +418,65 @@ void getBytes_codeReferencesUnknownSymbol_throwsVortexExceptionNotRaw() { assertThatExceptionOfType(VortexException.class) .isThrownBy(() -> result.getBytes(0)); } + + @Test + void compareUnsigned_agreesWithTheDecodedRow() { + // Given — "ab" then "abX" (an escaped literal after the symbol), each compared with a value + // it equals, one it extends, and one it differs from inside the symbol + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x00, 0x00, (byte) ESCAPE, 'X'}; + long[] uncompLengths = {2, 3}; + long[] codeOffsets = {0, 1, 4}; + VarBinArray sut = decodeFsst(2, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When + int[] result = { + sut.compareUnsigned(0, "ab".getBytes(StandardCharsets.UTF_8)), + sut.compareUnsigned(1, "ab".getBytes(StandardCharsets.UTF_8)), + sut.compareUnsigned(1, "abXY".getBytes(StandardCharsets.UTF_8)), + sut.compareUnsigned(0, "aa".getBytes(StandardCharsets.UTF_8))}; + + // Then + assertThat(result[0]).isZero(); + assertThat(result[1]).isPositive(); + assertThat(result[2]).isNegative(); + assertThat(result[3]).isPositive(); + } + + @Test + void compareUnsigned_codeReferencesUnknownSymbol_throwsVortexExceptionNotRaw() { + // Given — the comparison decodes the row's first code, which names no trained symbol + long[] symbols = {packSymbol("ab")}; + byte[] symbolLengths = {2}; + byte[] compressed = {0x05}; + long[] uncompLengths = {2}; + long[] codeOffsets = {0, 1}; + VarBinArray sut = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> sut.compareUnsigned(0, new byte[]{'a'})); + } + + @Test + void compareUnsigned_truncatedEscapeAtBufferEnd_throwsVortexExceptionNotRaw() { + // Given — the row's only code is the escape marker with no literal after it, and the buffer + // ends there, so reading the literal falls off the segment + long[] symbols = {}; + byte[] symbolLengths = {}; + byte[] compressed = {(byte) 0xFF}; + long[] uncompLengths = {1}; + long[] codeOffsets = {0, 1}; + VarBinArray sut = decodeFsst(1, symbols, symbolLengths, compressed, + PType.U8, uncompLengths, PType.U8, codeOffsets); + + // When / Then + assertThatExceptionOfType(VortexException.class) + .isThrownBy(() -> sut.compareUnsigned(0, new byte[]{'a'})); + } } @Nested