This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/codex/codec-delta-chunk-cache-fix in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 19912b44879b3e828c52a1e639c7c14f9fdf1eab Author: Xiang Fu <[email protected]> AuthorDate: Mon Aug 17 16:11:52 2026 -0700 Fix delta forward index chunk caching --- .../forward/BaseChunkForwardIndexReader.java | 18 +++--- .../forward/FixedByteChunkSVForwardIndexTest.java | 67 ++++++++++++++++++++++ 2 files changed, 76 insertions(+), 9 deletions(-) diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/BaseChunkForwardIndexReader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/BaseChunkForwardIndexReader.java index ef28c4bba0f..0e837cd48eb 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/BaseChunkForwardIndexReader.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/BaseChunkForwardIndexReader.java @@ -211,17 +211,17 @@ public abstract class BaseChunkForwardIndexReader implements ForwardIndexReader< ByteBuffer decompressedBuffer = context.getChunkBuffer(); decompressedBuffer.clear(); + // Invalidate the cached chunk before decoding. If decompression fails, a subsequent read must + // retry instead of returning a partially-mutated buffer as a cache hit. + context.setChunkId(-1); try { - if (_compressionType == ChunkCompressionType.DELTA || _compressionType == ChunkCompressionType.DELTADELTA) { - // For delta-based compression, pre-size the output using decompressor's length calculation. - ByteBuffer compressedBuffer = _dataBuffer.toDirectByteBuffer(chunkPosition, chunkSize); - int decompressedSize = _chunkDecompressor.decompressedLength(compressedBuffer); - decompressedBuffer = ByteBuffer.allocateDirect(decompressedSize); - _chunkDecompressor.decompress(compressedBuffer, decompressedBuffer); - } else { - _chunkDecompressor.decompress(_dataBuffer.toDirectByteBuffer(chunkPosition, chunkSize), decompressedBuffer); - } + // ChunkReaderContext is sized for a full decoded chunk. Decode every compression type into + // that owned buffer so the buffer returned on the first read is also the one cached for + // subsequent reads in the same chunk. The old DELTA/DELTADELTA branch allocated a separate + // buffer without installing it in the context, causing all same-chunk reads after the first + // one to observe the untouched context buffer. + _chunkDecompressor.decompress(_dataBuffer.toDirectByteBuffer(chunkPosition, chunkSize), decompressedBuffer); } catch (IOException e) { LOGGER.error("Exception caught while decompressing data chunk", e); throw new RuntimeException(e); diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/FixedByteChunkSVForwardIndexTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/FixedByteChunkSVForwardIndexTest.java index ea811198199..33099e98246 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/FixedByteChunkSVForwardIndexTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/FixedByteChunkSVForwardIndexTest.java @@ -60,6 +60,73 @@ public class FixedByteChunkSVForwardIndexTest implements PinotBuffersAfterMethod .toArray(Object[][]::new); } + @DataProvider(name = "deltaCompressions") + public static Object[][] deltaCompressions() { + return new Object[][]{ + {ChunkCompressionType.DELTA}, + {ChunkCompressionType.DELTADELTA} + }; + } + + @Test(dataProvider = "deltaCompressions") + public void testDeltaIntChunkCaching(ChunkCompressionType compressionType) + throws Exception { + int[] expected = {101, 103, 107, 109, 211, 223, 227, 229, 307}; + File outputFile = new File(TEST_FILE + "-int-" + compressionType); + FileUtils.deleteQuietly(outputFile); + + try { + try (FixedByteChunkForwardIndexWriter writer = new FixedByteChunkForwardIndexWriter(outputFile, + compressionType, expected.length, 4, Integer.BYTES, FixedBytePower2ChunkSVForwardIndexReader.VERSION)) { + for (int value : expected) { + writer.putInt(value); + } + } + + try (PinotDataBuffer buffer = PinotDataBuffer.mapReadOnlyBigEndianFile(outputFile); + ForwardIndexReader<ChunkReaderContext> reader = + new FixedBytePower2ChunkSVForwardIndexReader(buffer, DataType.INT); + ChunkReaderContext context = reader.createContext()) { + int[] docIds = {0, 1, 4, 5, 8, 2, 3, 6, 7}; + for (int docId : docIds) { + Assert.assertEquals(reader.getInt(docId, context), expected[docId]); + } + } + } finally { + FileUtils.deleteQuietly(outputFile); + } + } + + @Test(dataProvider = "deltaCompressions") + public void testDeltaLongChunkCaching(ChunkCompressionType compressionType) + throws Exception { + long[] expected = {10_000_000_001L, 10_000_000_003L, 10_000_000_007L, 10_000_000_009L, 20_000_000_011L, + 20_000_000_033L, 20_000_000_039L, 20_000_000_051L, 30_000_000_077L}; + File outputFile = new File(TEST_FILE + "-long-" + compressionType); + FileUtils.deleteQuietly(outputFile); + + try { + try (FixedByteChunkForwardIndexWriter writer = new FixedByteChunkForwardIndexWriter(outputFile, + compressionType, expected.length, 4, Long.BYTES, FixedBytePower2ChunkSVForwardIndexReader.VERSION)) { + for (long value : expected) { + writer.putLong(value); + } + } + + try (PinotDataBuffer buffer = PinotDataBuffer.mapReadOnlyBigEndianFile(outputFile); + ForwardIndexReader<ChunkReaderContext> reader = + new FixedBytePower2ChunkSVForwardIndexReader(buffer, DataType.LONG); + ChunkReaderContext context = reader.createContext()) { + int[] docIds = {0, 1, 4, 5, 8, 2, 3, 6, 7}; + for (int docId : docIds) { + Assert.assertEquals(reader.getLong(docId, context), expected[docId]); + } + } + } finally { + FileUtils.deleteQuietly(outputFile); + } + } + @Test(dataProvider = "combinations") public void testInt(ChunkCompressionType compressionType, int version) throws Exception { --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
