diff --git a/fsst/src/main/java/io/github/dfa1/vortex/fsst/Compressor.java b/fsst/src/main/java/io/github/dfa1/vortex/fsst/Compressor.java index a39e9432..192020af 100644 --- a/fsst/src/main/java/io/github/dfa1/vortex/fsst/Compressor.java +++ b/fsst/src/main/java/io/github/dfa1/vortex/fsst/Compressor.java @@ -1,5 +1,7 @@ package io.github.dfa1.vortex.fsst; +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; import java.util.ArrayList; import java.util.List; @@ -17,6 +19,12 @@ public final class Compressor { /// Escape code: emitted as `0xFF` followed by one literal byte. Real symbol codes are `0..254`. static final int ESCAPE = Decompressor.ESCAPE; + /// Little-endian `long` layout for `MemorySegment` access. Defined locally rather than reusing + /// `core`'s `VortexFormat.LE_LONG` so this module keeps its zero dependency on `core`; the FSST + /// wire format packs symbol bytes LSB-first, so all `MemorySegment` reads/writes are little-endian. + static final ValueLayout.OfLong LE_LONG = + ValueLayout.JAVA_LONG_UNALIGNED.withOrder(java.nio.ByteOrder.LITTLE_ENDIAN); + private final List symbolsByGainDescending; private final Matcher matcher; @@ -155,6 +163,47 @@ public long compress(byte[] input, int start, int end, byte[] out, long outPos) return outIndex; } + /// Compresses the byte range `[start, end)` of the `MemorySegment` `input` into `out` using + /// longest-match-first greedy parsing, writing the code stream starting at `outPos`. + /// + /// This is the `MemorySegment`-native hot path, matching the FFM-everywhere convention at this + /// module's boundary; it mirrors [#compress(byte[], int, int, byte[], long)] exactly, including + /// the `pos + length <= end` guard that rejects an over-long spurious match against zero-padded + /// bytes past the input end. Dropping that guard would let a trained symbol whose trailing bytes + /// are zero satisfy the branch-free [Matcher]'s masked compare against the padding and decode to + /// more bytes than were compressed, corrupting the tail of any NUL-containing or binary row. + /// + /// The caller must size `out` so every code fits: a range of `end - start` input bytes expands to + /// at most `2 * (end - start)` output bytes (a full run of escapes). This is an internal + /// algorithm boundary, not an untrusted-input boundary, so no defensive bounds checks are + /// performed here. + /// + /// @param input the raw bytes to compress + /// @param start the index of the first byte to compress, inclusive + /// @param end the index one past the last byte to compress, exclusive + /// @param out the destination segment for the code stream + /// @param outPos the index in `out` at which to start writing + /// @return the index in `out` one past the last byte written + public long compress(MemorySegment input, long start, long end, MemorySegment out, long outPos) { + long pos = start; + long outIndex = outPos; + while (pos < end) { + int packedMatch = matcher.longestMatch(loadWord(input, pos, end)); + int length = Matcher.lengthOf(packedMatch); + // Same boundary guard as the byte[] overload: reject a match that reaches past the real + // input end, since loadWord zero-pads and the branch-free Matcher has no notion of end. + if (length > 0 && pos + length <= end) { + out.set(ValueLayout.JAVA_BYTE, outIndex++, (byte) Matcher.codeOf(packedMatch)); + pos += length; + } else { + out.set(ValueLayout.JAVA_BYTE, outIndex++, (byte) ESCAPE); + out.set(ValueLayout.JAVA_BYTE, outIndex++, input.get(ValueLayout.JAVA_BYTE, pos)); + pos++; + } + } + return outIndex; + } + /// Loads an 8-byte little-endian word from `input` starting at `pos`, zero-padding any bytes at /// or past `end` so the [Matcher]'s masked compare never reads across the range boundary. /// @@ -170,4 +219,27 @@ static long loadWord(byte[] input, int pos, int end) { } return word; } + + /// Loads an 8-byte little-endian word from the `MemorySegment` `input` starting at `pos`, + /// zero-padding any bytes at or past `end` so the [Matcher]'s masked compare never reads across + /// the range boundary. + /// + /// Unlike the `byte[]` overload — which may safely over-read within a same-array bounds check — a + /// wide unconditional 8-byte read is not safe here: `input` may represent exactly one row's range + /// with no guaranteed trailing slack past `end`, so reading past the segment's own allocated + /// bounds would fault. The word is therefore assembled byte by byte, bounded by whichever of `end` + /// and the segment's own size comes first. + /// + /// @param input the source segment + /// @param pos the first byte to load + /// @param end the exclusive range end; bytes at or after it read as zero + /// @return the 8-byte little-endian word at `pos` + static long loadWord(MemorySegment input, long pos, long end) { + long word = 0; + long limit = Math.min(pos + 8, Math.min(end, input.byteSize())); + for (long k = pos; k < limit; k++) { + word |= Byte.toUnsignedLong(input.get(ValueLayout.JAVA_BYTE, k)) << ((k - pos) * 8); + } + return word; + } } 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 6ca758e5..9c1d176c 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 @@ -1,5 +1,8 @@ package io.github.dfa1.vortex.fsst; +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; + /// Decodes an FSST-compressed byte stream against a trained symbol table (the FSST paper's /// Algorithm 1). /// @@ -64,4 +67,46 @@ public long decompress(byte[] compressed, int start, int end, byte[] out, long o } return outIndex; } + + /// Decodes the compressed byte range `[start, end)` of the `MemorySegment` `compressed` into + /// `out`, writing the decoded bytes starting at `outPos`, using the FSST paper's Algorithm 1 + /// "unconditional 8-byte store" trick. + /// + /// For each non-escape code this writes the symbol's packed bytes as a single unconditional + /// 8-byte little-endian store and then advances the output cursor by only the symbol's true + /// length (1-8). The extra bytes written past the symbol's length are slack: the next code's + /// store overwrites them, so no per-byte copy loop and no per-length branch are needed in the hot + /// path — the store width is a compile-time constant and the loop body stays uniform, satisfying + /// the hot-loop rule. + /// + /// Precondition — the caller MUST allocate `out` with at least 7 bytes of slack past the true + /// logical end of all decoded output (the sum of every row's decompressed length). The final + /// symbol of the final row still writes a full 8 bytes even though it may contribute as few as 1 + /// real byte; without the trailing 7-byte slack that store would run past the segment's bounds and + /// fault. The escape branch writes exactly one literal byte and needs no slack of its own. + /// + /// @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 out the destination segment for decoded bytes, allocated with at least 7 bytes of slack + /// past the total decoded length + /// @param outPos the index in `out` at which to start writing + /// @return the index in `out` one past the last real decoded byte written + public long decompress(MemorySegment compressed, long start, long end, MemorySegment out, long outPos) { + long pos = start; + long outIndex = outPos; + while (pos < end) { + int code = Byte.toUnsignedInt(compressed.get(ValueLayout.JAVA_BYTE, pos++)); + if (code == ESCAPE) { + out.set(ValueLayout.JAVA_BYTE, outIndex++, compressed.get(ValueLayout.JAVA_BYTE, pos++)); + } else { + // One unconditional 8-byte store, then advance by the true length only; the slack + // bytes past the symbol length are overwritten by the next store (or covered by the + // caller's required +7 trailing slack for the very last symbol). + out.set(Compressor.LE_LONG, outIndex, packedSymbols[code]); + outIndex += lengths[code]; + } + } + return outIndex; + } } diff --git a/fsst/src/test/java/io/github/dfa1/vortex/fsst/MemorySegmentParityTest.java b/fsst/src/test/java/io/github/dfa1/vortex/fsst/MemorySegmentParityTest.java new file mode 100644 index 00000000..9a40ddde --- /dev/null +++ b/fsst/src/test/java/io/github/dfa1/vortex/fsst/MemorySegmentParityTest.java @@ -0,0 +1,198 @@ +package io.github.dfa1.vortex.fsst; + +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; + +import java.lang.foreign.Arena; +import java.lang.foreign.MemorySegment; +import java.lang.foreign.ValueLayout; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; + +/// Verifies the `MemorySegment`-native `compress`/`decompress` hot paths against their `byte[]` +/// counterparts and covers the paper's unconditional-8-byte-store decode trick. +class MemorySegmentParityTest { + + @Nested + class Parity { + + @Test + void compressAndDecompress_emptyInput_matchByteArrayPath() { + // Given — nothing to compress; both paths must agree on the zero-length result. + Compressor sut = new CompressorBuilder().train(logLikeRows(500)); + + // When / Then + assertRoundTripParity(sut, new byte[0]); + } + + @Test + void compressAndDecompress_singleByte_matchByteArrayPath() { + // Given — a lone byte, the smallest non-trivial input. + Compressor sut = new CompressorBuilder().train(logLikeRows(500)); + + // When / Then + assertRoundTripParity(sut, new byte[]{'x'}); + } + + @Test + void compressAndDecompress_repetitiveText_matchByteArrayPath() { + // Given — highly repetitive text that trains multi-byte symbols, so most positions take + // the match branch rather than the escape branch on both paths. + Compressor sut = new CompressorBuilder().train(logLikeRows(2_000)); + + // When / Then + assertRoundTripParity(sut, logLikeRows(2_000)[7]); + } + + @Test + void compressAndDecompress_incompressibleData_matchByteArrayPath() { + // Given — pseudo-random bytes trained on unrelated text, so almost every position escapes; + // this exercises the escape branch of both compress and decompress on both paths. + Compressor sut = new CompressorBuilder().train(logLikeRows(500)); + byte[] noise = new byte[4_096]; + new Random(99).nextBytes(noise); + + // When / Then + assertRoundTripParity(sut, noise); + } + + @Test + void compressAndDecompress_boundaryOverrunScenario_matchByteArrayPath() { + // Given — PR 3's regression case: a length-8 symbol "AB\0\0\0\0\0\0" and the 3-byte input + // "AB\0", where a dropped end guard would spuriously match the padded symbol. Both paths + // must produce identical output, proving the MemorySegment overload carries the guard. + long packed = ('A' & 0xFFL) | (('B' & 0xFFL) << 8); + Compressor sut = Compressor.of(List.of(new Symbol(packed, 8))); + + // When / Then + assertRoundTripParity(sut, new byte[]{'A', 'B', 0}); + } + } + + @Nested + class UnconditionalStoreTrick { + + @Test + void decompress_multipleRowsIntoSharedSegment_noCrossRowCorruption() { + // Given — three consecutive rows decoded into one shared output segment. The 8-byte-store + // trick over-writes up to 7 slack bytes past each symbol; if a row's slack bled into the + // next row's real data it would corrupt it, so we verify each row's region independently. + Symbol ab = Symbol.of("ab".getBytes(StandardCharsets.UTF_8), 0, 2); + Symbol wxyz = Symbol.of("wxyz".getBytes(StandardCharsets.UTF_8), 0, 4); + Decompressor sut = Decompressor.of(new long[]{ab.packedBytes(), wxyz.packedBytes()}, new int[]{2, 4}); + + byte[] row0 = {0x00}; // "ab" + byte[] row1 = {(byte) Decompressor.ESCAPE, 'Q'}; // "Q" + byte[] row2 = {0x01, 0x00}; // "wxyzab" + byte[] expected0 = "ab".getBytes(StandardCharsets.UTF_8); + byte[] expected1 = "Q".getBytes(StandardCharsets.UTF_8); + byte[] expected2 = "wxyzab".getBytes(StandardCharsets.UTF_8); + long total = expected0.length + expected1.length + expected2.length; + + try (Arena arena = Arena.ofConfined()) { + MemorySegment c0 = segmentOf(arena, row0); + MemorySegment c1 = segmentOf(arena, row1); + MemorySegment c2 = segmentOf(arena, row2); + // +7 slack past the total decoded length is the documented precondition of the trick. + MemorySegment out = arena.allocate(total + 7); + + // When — decode all three rows sequentially into the shared segment. + long after0 = sut.decompress(c0, 0, c0.byteSize(), out, 0); + long after1 = sut.decompress(c1, 0, c1.byteSize(), out, after0); + long after2 = sut.decompress(c2, 0, c2.byteSize(), out, after1); + + // Then — each row's region is exactly correct; no row bled into the next. + assertThat(after0).isEqualTo(expected0.length); + assertThat(after1).isEqualTo((long) expected0.length + expected1.length); + assertThat(after2).isEqualTo(total); + assertThat(bytesOf(out, 0, expected0.length)).isEqualTo(expected0); + assertThat(bytesOf(out, after0, expected1.length)).isEqualTo(expected1); + assertThat(bytesOf(out, after1, expected2.length)).isEqualTo(expected2); + } + } + } + + @Nested + class EndOfInput { + + @Test + void compress_symbolWithTrailingZeros_doesNotMatchPastEndOfInput_memorySegment() { + // Given — the exact PR 3 regression ported to the MemorySegment overload: a length-8 + // symbol "AB\0\0\0\0\0\0" and the 3-byte input "AB\0". The MemorySegment compress must + // carry the pos+length<=end guard and escape rather than emit an over-long match that + // would decode to 8 bytes and corrupt the tail of this NUL-tailed row. + long packed = ('A' & 0xFFL) | (('B' & 0xFFL) << 8); + Compressor sut = Compressor.of(List.of(new Symbol(packed, 8))); + byte[] input = {'A', 'B', 0}; + + try (Arena arena = Arena.ofConfined()) { + MemorySegment in = segmentOf(arena, input); + MemorySegment compressed = arena.allocate(input.length * 2L); + MemorySegment decompressed = arena.allocate(input.length + 7L); + + // When — compress then decompress the shorter input. + long compressedLength = sut.compress(in, 0, in.byteSize(), compressed, 0); + long decompressedLength = + sut.toDecompressor().decompress(compressed, 0, compressedLength, decompressed, 0); + + // Then — the round-trip preserves the exact original length and bytes. + assertThat(decompressedLength).isEqualTo(input.length); + assertThat(bytesOf(decompressed, 0, input.length)).isEqualTo(input); + } + } + } + + private static void assertRoundTripParity(Compressor sut, byte[] input) { + // byte[] path. + byte[] heapCompressed = new byte[input.length * 2]; + long heapCompressedLength = sut.compress(input, 0, input.length, heapCompressed, 0); + byte[] heapDecompressed = new byte[input.length + 7]; + long heapDecompressedLength = + sut.toDecompressor().decompress(heapCompressed, 0, (int) heapCompressedLength, heapDecompressed, 0); + + try (Arena arena = Arena.ofConfined()) { + MemorySegment in = segmentOf(arena, input); + MemorySegment segCompressed = arena.allocate(Math.max(1, input.length * 2L)); + long segCompressedLength = sut.compress(in, 0, in.byteSize(), segCompressed, 0); + + MemorySegment segDecompressed = arena.allocate(input.length + 7L); + long segDecompressedLength = + sut.toDecompressor().decompress(segCompressed, 0, segCompressedLength, segDecompressed, 0); + + // Then — both paths compress to the identical code stream and decompress to the identical + // bytes, so the two implementations can never silently diverge. + assertThat(segCompressedLength).isEqualTo(heapCompressedLength); + assertThat(bytesOf(segCompressed, 0, segCompressedLength)) + .isEqualTo(java.util.Arrays.copyOf(heapCompressed, (int) heapCompressedLength)); + assertThat(segDecompressedLength).isEqualTo(heapDecompressedLength); + assertThat(bytesOf(segDecompressed, 0, segDecompressedLength)) + .isEqualTo(java.util.Arrays.copyOf(heapDecompressed, (int) heapDecompressedLength)); + // And the decoded bytes equal the original input. + assertThat(bytesOf(segDecompressed, 0, input.length)).isEqualTo(input); + } + } + + private static MemorySegment segmentOf(Arena arena, byte[] data) { + MemorySegment segment = arena.allocate(Math.max(1, data.length)); + MemorySegment.copy(data, 0, segment, ValueLayout.JAVA_BYTE, 0, data.length); + return segment.asSlice(0, data.length); + } + + private static byte[] bytesOf(MemorySegment segment, long offset, long length) { + byte[] result = new byte[(int) length]; + MemorySegment.copy(segment, ValueLayout.JAVA_BYTE, offset, result, 0, (int) length); + return result; + } + + private static byte[][] logLikeRows(int count) { + byte[][] rows = new byte[count][]; + String prefix = "2026-07-20T12:34:56Z ERROR request failed host=example.com id="; + for (int i = 0; i < count; i++) { + rows[i] = (prefix + i).getBytes(StandardCharsets.UTF_8); + } + return rows; + } +}