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 52d9c8027b8 Add scoped buffer reads for compressed forward indexes
(#20069)
52d9c8027b8 is described below
commit 52d9c8027b829cddaf590b34abea3ed2bdbae5db
Author: Xiang Fu <[email protected]>
AuthorDate: Wed Oct 28 16:54:12 2026 -0700
Add scoped buffer reads for compressed forward indexes (#20069)
* Add scoped buffer reads for compressed forward indexes
* Use the existing IOFunction type for scoped reads
* Use Commons IO transitively from pinot-spi
* Clarify scoped buffer access and name it withByteBuffer
* Name the scoped buffer API readWithByteBuffer
---
.../readers/forward/BaseChunkForwardIndexReader.java | 10 +++++++---
.../readers/forward/VarByteChunkForwardIndexReaderV4.java | 15 ++++++++-------
.../apache/pinot/segment/spi/memory/PinotDataBuffer.java | 12 ++++++++++++
3 files changed, 27 insertions(+), 10 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 0e837cd48eb..571c9ee60f1 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
@@ -209,19 +209,23 @@ public abstract class BaseChunkForwardIndexReader
implements ForwardIndexReader<
chunkSize = (int) (nextChunkOffset - chunkPosition);
}
- ByteBuffer decompressedBuffer = context.getChunkBuffer();
- decompressedBuffer.clear();
+ ByteBuffer chunkBuffer = context.getChunkBuffer();
+ chunkBuffer.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);
+ ByteBuffer decompressedBuffer;
try {
// 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);
+ decompressedBuffer = _dataBuffer.readWithByteBuffer(chunkPosition,
chunkSize, compressedBuffer -> {
+ _chunkDecompressor.decompress(compressedBuffer, chunkBuffer);
+ return chunkBuffer;
+ });
} catch (IOException e) {
LOGGER.error("Exception caught while decompressing data chunk", e);
throw new RuntimeException(e);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkForwardIndexReaderV4.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkForwardIndexReaderV4.java
index b3a74be0378..57deb8e8878 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkForwardIndexReaderV4.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkForwardIndexReaderV4.java
@@ -427,13 +427,14 @@ public class VarByteChunkForwardIndexReaderV4
protected byte[] processChunkAndReadFirstValue(int docId, long offset,
long limit)
throws IOException {
_decompressedBuffer.clear();
- ByteBuffer compressed = _chunks.toDirectByteBuffer(offset, (int) (limit
- offset));
- if (_regularChunk) {
- decompressChunk(compressed);
- return readSmallUncompressedValue(docId);
- }
- // huge value, no benefit from buffering, return the whole thing
- return readHugeCompressedValue(compressed,
_chunkDecompressor.decompressedLength(compressed));
+ return _chunks.readWithByteBuffer(offset, (int) (limit - offset),
compressed -> {
+ if (_regularChunk) {
+ decompressChunk(compressed);
+ return readSmallUncompressedValue(docId);
+ }
+ // huge value, no benefit from buffering, return the whole thing
+ return readHugeCompressedValue(compressed,
_chunkDecompressor.decompressedLength(compressed));
+ });
}
/// Decompresses a regular chunk and reads the number of docs. Subclasses
(e.g. V6) can override
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PinotDataBuffer.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PinotDataBuffer.java
index ed60deb1e4d..b929082b965 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PinotDataBuffer.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PinotDataBuffer.java
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
+import org.apache.commons.io.function.IOFunction;
import org.apache.pinot.segment.spi.memory.unsafe.UnsafePinotBufferFactory;
import org.apache.pinot.segment.spi.utils.JavaVersion;
import org.apache.pinot.spi.env.PinotConfiguration;
@@ -649,6 +650,17 @@ public abstract class PinotDataBuffer implements
DataBuffer {
return toDirectByteBuffer(offset, size, order());
}
+ /// Reads a range synchronously through a borrowed ByteBuffer with this
receiver's byte order.
+ /// The buffer may be heap-backed or direct, depending on the receiver.
+ /// The callback receives a buffer positioned at zero with its limit set to
`size`. It may change the position
+ /// and limit, but must not modify the contents, release the buffer, or
retain or return the buffer or any alias.
+ /// The callback must not recursively access or release this receiver or its
views. The borrowed buffer is only
+ /// valid during the callback; the result must be independent of it.
+ public <T> T readWithByteBuffer(long offset, int size,
IOFunction<ByteBuffer, T> reader)
+ throws IOException {
+ return reader.apply(toDirectByteBuffer(offset, size));
+ }
+
@Override
public ByteBuffer copyOrView(long offset, int size, ByteOrder byteOrder) {
return toDirectByteBuffer(offset, size, byteOrder);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
