This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 8fa27943a3c Fix delta forward index chunk caching (#19282)
8fa27943a3c is described below
commit 8fa27943a3c1be27cdd51690056160a20038cb3c
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Aug 18 16:18:13 2026 -0700
Fix delta forward index chunk caching (#19282)
---
.../forward/BaseChunkForwardIndexReader.java | 18 ++--
.../forward/FixedByteChunkSVForwardIndexTest.java | 117 +++++++++++++++++++++
2 files changed, 126 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..6d4cf22decc 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
@@ -19,6 +19,7 @@
package org.apache.pinot.segment.local.segment.index.forward;
import java.io.File;
+import java.io.RandomAccessFile;
import java.net.URL;
import java.util.Arrays;
import java.util.Random;
@@ -60,6 +61,122 @@ public class FixedByteChunkSVForwardIndexTest implements
PinotBuffersAfterMethod
.toArray(Object[][]::new);
}
+ @DataProvider(name = "deltaCombinations")
+ public static Object[][] deltaCombinations() {
+ return Arrays.stream(new
ChunkCompressionType[]{ChunkCompressionType.DELTA,
ChunkCompressionType.DELTADELTA})
+ .flatMap(chunkCompressionType -> IntStream.of(2, 3, 4)
+ .mapToObj(version -> new Object[]{chunkCompressionType, version}))
+ .toArray(Object[][]::new);
+ }
+
+ @Test(dataProvider = "deltaCombinations")
+ public void testDeltaIntChunkCaching(ChunkCompressionType compressionType,
int version)
+ throws Exception {
+ int[] expected = {101, 103, 107, 109, 211, 223, 227, 229, 307};
+ File outputFile = new File(TEST_FILE + "-int-" + compressionType + '-' +
version);
+ FileUtils.deleteQuietly(outputFile);
+
+ try {
+ try (FixedByteChunkForwardIndexWriter writer = new
FixedByteChunkForwardIndexWriter(outputFile,
+ compressionType, expected.length, 4, Integer.BYTES, version)) {
+ for (int value : expected) {
+ writer.putInt(value);
+ }
+ }
+
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(outputFile);
+ ForwardIndexReader<ChunkReaderContext> reader = version >= 4
+ ? new FixedBytePower2ChunkSVForwardIndexReader(buffer,
DataType.INT)
+ : new FixedByteChunkSVForwardIndexReader(buffer, DataType.INT);
+ ChunkReaderContext context = reader.createContext()) {
+ // Same-chunk reads, cross-chunk reads, a partial final chunk, and
revisits expose stale context buffers.
+ 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 = "deltaCombinations")
+ public void testDeltaLongChunkCaching(ChunkCompressionType compressionType,
int version)
+ 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 + '-' +
version);
+ FileUtils.deleteQuietly(outputFile);
+
+ try {
+ try (FixedByteChunkForwardIndexWriter writer = new
FixedByteChunkForwardIndexWriter(outputFile,
+ compressionType, expected.length, 4, Long.BYTES, version)) {
+ for (long value : expected) {
+ writer.putLong(value);
+ }
+ }
+
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(outputFile);
+ ForwardIndexReader<ChunkReaderContext> reader = version >= 4
+ ? new FixedBytePower2ChunkSVForwardIndexReader(buffer,
DataType.LONG)
+ : new FixedByteChunkSVForwardIndexReader(buffer, DataType.LONG);
+ ChunkReaderContext context = reader.createContext()) {
+ // Same-chunk reads, cross-chunk reads, a partial final chunk, and
revisits expose stale context buffers.
+ 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
+ public void testFailedDeltaDecodeInvalidatesCachedChunk()
+ throws Exception {
+ long[] expected = {101L, 103L, 107L, 109L, 211L, 223L, 227L, 229L};
+ File outputFile = new File(TEST_FILE + "-failed-delta-decode");
+ FileUtils.deleteQuietly(outputFile);
+
+ try {
+ try (FixedByteChunkForwardIndexWriter writer = new
FixedByteChunkForwardIndexWriter(outputFile,
+ ChunkCompressionType.DELTA, expected.length, 4, Long.BYTES,
+ FixedBytePower2ChunkSVForwardIndexReader.VERSION)) {
+ for (long value : expected) {
+ writer.putLong(value);
+ }
+ }
+
+ // Corrupt the second chunk's compressed-size field. DELTA writes the
first decoded value into the context
+ // before validating this field, so reading the malformed chunk
partially mutates the cached buffer and fails.
+ try (RandomAccessFile file = new RandomAccessFile(outputFile, "rw")) {
+ int dataHeaderStart = 7 * Integer.BYTES;
+ file.seek(dataHeaderStart + Long.BYTES);
+ long secondChunkOffset = file.readLong();
+ file.seek(secondChunkOffset + Byte.BYTES + Integer.BYTES + Long.BYTES);
+ file.writeInt(Integer.MAX_VALUE);
+ }
+
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(outputFile);
+ ForwardIndexReader<ChunkReaderContext> reader =
+ new FixedBytePower2ChunkSVForwardIndexReader(buffer,
DataType.LONG);
+ ChunkReaderContext context = reader.createContext()) {
+ Assert.assertEquals(reader.getLong(0, context), expected[0]);
+ Assert.assertEquals(context.getChunkId(), 0);
+
+ Assert.expectThrows(IllegalArgumentException.class, () ->
reader.getLong(4, context));
+ Assert.assertEquals(context.getChunkId(), -1);
+
+ // With a stale chunk id this would falsely return 211, which the
failed decode wrote at buffer offset zero.
+ Assert.assertEquals(reader.getLong(0, context), expected[0]);
+ Assert.assertEquals(reader.getLong(1, context), expected[1]);
+ }
+ } 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]
