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 80dbb7d02df Read MAP keys selectively from sealed forward indexes
(#19273)
80dbb7d02df is described below
commit 80dbb7d02dfd92ccc20ae2606ac712ea4f73d3cf
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Aug 30 12:05:32 2026 -0700
Read MAP keys selectively from sealed forward indexes (#19273)
Override selective MAP-key reads for legacy V2/V3 and V4-V6 chunk readers
so projections avoid materializing the full map. Add sealed benchmark pairs and
coverage across reader versions and compression types.
---
.../apache/pinot/perf/BenchmarkMapKeyAccess.java | 61 ++++++++++++++++-
.../forward/VarByteChunkForwardIndexReaderV4.java | 14 ++++
.../forward/VarByteChunkSVForwardIndexReader.java | 13 ++++
.../segment/index/creator/VarByteChunkV4Test.java | 61 +++++++++++++++++
.../forward/VarByteChunkSVForwardIndexTest.java | 78 ++++++++++++++++++++++
5 files changed, 225 insertions(+), 2 deletions(-)
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapKeyAccess.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapKeyAccess.java
index cf26f711085..96fd4026363 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapKeyAccess.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapKeyAccess.java
@@ -18,12 +18,19 @@
*/
package org.apache.pinot.perf;
+import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager;
+import
org.apache.pinot.segment.local.io.writer.impl.VarByteChunkForwardIndexWriterV4;
import
org.apache.pinot.segment.local.realtime.impl.forward.VarByteSVMutableForwardIndex;
+import
org.apache.pinot.segment.local.segment.index.readers.forward.VarByteChunkForwardIndexReaderV4;
+import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
+import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.utils.MapUtils;
@@ -87,9 +94,14 @@ public class BenchmarkMapKeyAccess {
private PreparedMapKey _targetMapKey;
private PinotDataBufferMemoryManager _memoryManager;
private VarByteSVMutableForwardIndex _forwardIndex;
+ private File _sealedIndexDir;
+ private PinotDataBuffer _sealedDataBuffer;
+ private VarByteChunkForwardIndexReaderV4 _sealedForwardIndex;
+ private VarByteChunkForwardIndexReaderV4.ReaderContext _sealedContext;
@Setup(Level.Trial)
- public void setUp() {
+ public void setUp()
+ throws IOException {
Map<String, Object> map = new LinkedHashMap<>();
for (int i = 0; i < _numEntries; i++) {
String value = "value-with-enough-bytes-to-exercise-json-parsing-" + i;
@@ -116,15 +128,34 @@ public class BenchmarkMapKeyAccess {
_forwardIndex =
new VarByteSVMutableForwardIndex(DataType.MAP, _memoryManager,
"mapColumn", 1, serialized.length);
_forwardIndex.setBytes(0, serialized);
+
+ // The sealed counterpart of the same frame. A completed segment stores
the MAP column in a chunked raw forward
+ // index, so key access there goes through a different reader than the
consuming path above.
+ _sealedIndexDir =
Files.createTempDirectory(BenchmarkMapKeyAccess.class.getSimpleName()).toFile();
+ File indexFile = new File(_sealedIndexDir, "map.fwd");
+ try (VarByteChunkForwardIndexWriterV4 writer = new
VarByteChunkForwardIndexWriterV4(indexFile,
+ ChunkCompressionType.LZ4, Math.max(1024, serialized.length * 2))) {
+ writer.putBytes(serialized);
+ }
+ _sealedDataBuffer = PinotDataBuffer.mapReadOnlyBigEndianFile(indexFile);
+ _sealedForwardIndex = new
VarByteChunkForwardIndexReaderV4(_sealedDataBuffer, DataType.MAP, true);
+ _sealedContext = _sealedForwardIndex.createContext();
}
@TearDown(Level.Trial)
public void tearDown()
throws IOException {
try {
+ _sealedContext.close();
+ _sealedForwardIndex.close();
+ _sealedDataBuffer.close();
_forwardIndex.close();
} finally {
- _memoryManager.close();
+ try {
+ _memoryManager.close();
+ } finally {
+ FileUtils.deleteQuietly(_sealedIndexDir);
+ }
}
}
@@ -160,4 +191,30 @@ public class BenchmarkMapKeyAccess {
public Object selectiveMapValueAsString() {
return _forwardIndex.getMapEntryValueAsString(0, null, _targetMapKey);
}
+
+ /// The selective object lookup against a completed segment's chunked
forward index.
+ @Benchmark
+ public Object sealedSelectiveMapValue() {
+ return _sealedForwardIndex.getMapEntryValue(0, _sealedContext,
_targetMapKey);
+ }
+
+ /// The selective string lookup against a completed segment's chunked
forward index.
+ @Benchmark
+ public Object sealedSelectiveMapValueAsString() {
+ return _sealedForwardIndex.getMapEntryValueAsString(0, _sealedContext,
_targetMapKey);
+ }
+
+ /// The sealed object baseline: deserialize every entry of the chunk value,
then select the requested key.
+ @Benchmark
+ public Object sealedFullMapValue() {
+ return _sealedForwardIndex.getMap(0, _sealedContext).get(_targetKey);
+ }
+
+ /// The sealed string baseline: deserialize every entry, select the key,
then apply the default accessor's
+ /// null-safe `toString()` conversion.
+ @Benchmark
+ public Object sealedFullMapValueAsString() {
+ Object value = _sealedForwardIndex.getMap(0,
_sealedContext).get(_targetKey);
+ return value == null ? null : value.toString();
+ }
}
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 0d65fba1b26..a7b6bb5b4d7 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
@@ -27,6 +27,7 @@ import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import javax.annotation.Nullable;
import org.apache.pinot.segment.local.io.compression.ChunkCompressorFactory;
import
org.apache.pinot.segment.local.io.writer.impl.VarByteChunkForwardIndexWriterV4;
import org.apache.pinot.segment.local.utils.ArraySerDeUtils;
@@ -39,6 +40,7 @@ import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.MapUtils;
+import org.apache.pinot.spi.utils.MapUtils.PreparedMapKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -143,6 +145,18 @@ public class VarByteChunkForwardIndexReaderV4
return MapUtils.frameToJsonString(context.getValue(docId));
}
+ @Nullable
+ @Override
+ public Object getMapEntryValue(int docId, ReaderContext context,
PreparedMapKey key) {
+ return MapUtils.deserializeMapEntryValue(context.getValue(docId), key);
+ }
+
+ @Nullable
+ @Override
+ public String getMapEntryValueAsString(int docId, ReaderContext context,
PreparedMapKey key) {
+ return
MapUtils.deserializeMapEntryValueAsString(ByteBuffer.wrap(context.getValue(docId)),
key);
+ }
+
@Override
public int getIntMV(int docId, int[] valueBuffer,
VarByteChunkForwardIndexReaderV4.ReaderContext context) {
return
ArraySerDeUtils.deserializeIntArrayWithLength(context.getValue(docId),
valueBuffer);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkSVForwardIndexReader.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkSVForwardIndexReader.java
index b732e483243..4a881b7b723 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkSVForwardIndexReader.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/readers/forward/VarByteChunkSVForwardIndexReader.java
@@ -28,6 +28,7 @@ import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.MapUtils;
+import org.apache.pinot.spi.utils.MapUtils.PreparedMapKey;
import static java.nio.charset.StandardCharsets.UTF_8;
@@ -135,6 +136,18 @@ public final class VarByteChunkSVForwardIndexReader
extends BaseChunkForwardInde
return MapUtils.frameToJsonString(getBytes(docId, context));
}
+ @Nullable
+ @Override
+ public Object getMapEntryValue(int docId, ChunkReaderContext context,
PreparedMapKey key) {
+ return MapUtils.deserializeMapEntryValue(getBytes(docId, context), key);
+ }
+
+ @Nullable
+ @Override
+ public String getMapEntryValueAsString(int docId, ChunkReaderContext
context, PreparedMapKey key) {
+ return
MapUtils.deserializeMapEntryValueAsString(ByteBuffer.wrap(getBytes(docId,
context)), key);
+ }
+
/// Helper method to read BYTES value from the compressed index.
private byte[] getBytesCompressed(int docId, ChunkReaderContext context) {
int chunkRowId = docId % _numDocsPerChunk;
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/VarByteChunkV4Test.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/VarByteChunkV4Test.java
index 359bf41fb19..6b6ed9336d2 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/VarByteChunkV4Test.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/creator/VarByteChunkV4Test.java
@@ -23,7 +23,9 @@ import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.BiConsumer;
@@ -39,12 +41,16 @@ import
org.apache.pinot.segment.local.segment.index.readers.forward.VarByteChunk
import org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.utils.MapUtils;
+import org.apache.pinot.spi.utils.MapUtils.PreparedMapKey;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
public class VarByteChunkV4Test implements PinotBuffersAfterClassCheckRule {
@@ -132,6 +138,61 @@ public class VarByteChunkV4Test implements
PinotBuffersAfterClassCheckRule {
FileUtils.deleteQuietly(bytesMVFile);
}
+ /// A sealed MAP column is a chunked raw index over serialized frames, and a
projected `attributes['key']` resolves
+ /// to a selective read against it. Pins that the V4 implementation, also
inherited by V5 and V6, agrees with
+ /// deserializing the whole frame for every compression type.
+ @Test(dataProvider = "params")
+ public void testMapSV(File file, ChunkCompressionType compressionType, int
longestEntry, int chunkSize)
+ throws Exception {
+ File mapSVFile = new File(file, "testMapSV");
+ int numDocs = 1000;
+ List<Map<String, Object>> maps = new ArrayList<>(numDocs);
+ try (VarByteChunkWriter writer = createWriter(mapSVFile, compressionType,
chunkSize)) {
+ for (int i = 0; i < numDocs; i++) {
+ // Dotted OpenTelemetry-style keys, the first two of equal length so
the scan has to compare their bytes
+ // rather than skip on a length mismatch.
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("k8s.workload.name", "workload-" + i);
+ map.put("k8s.workload.kind", "Deployment");
+ map.put("k8s.namespace.name", "namespace-" + i % 7);
+ map.put("host_logical_cpus", i);
+ maps.add(map);
+ writer.putBytes(MapUtils.serializeMap(map));
+ }
+ }
+
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(mapSVFile);
+ VarByteChunkForwardIndexReaderV4 reader = createReader(buffer,
FieldSpec.DataType.MAP, true);
+ VarByteChunkForwardIndexReaderV4.ReaderContext context =
reader.createContext()) {
+ // A value-only assertion would also pass through ForwardIndexReader's
full-map fallback. Pin the actual
+ // dispatch target so removing either optimized override makes this
regression test fail.
+ assertSame(reader.getClass()
+ .getMethod("getMapEntryValue", int.class,
VarByteChunkForwardIndexReaderV4.ReaderContext.class,
+ PreparedMapKey.class)
+ .getDeclaringClass(),
+ VarByteChunkForwardIndexReaderV4.class);
+ assertSame(reader.getClass()
+ .getMethod("getMapEntryValueAsString", int.class,
VarByteChunkForwardIndexReaderV4.ReaderContext.class,
+ PreparedMapKey.class)
+ .getDeclaringClass(),
+ VarByteChunkForwardIndexReaderV4.class);
+ PreparedMapKey workloadName = new PreparedMapKey("k8s.workload.name");
+ PreparedMapKey cpus = new PreparedMapKey("host_logical_cpus");
+ PreparedMapKey absent = new PreparedMapKey("k8s.workload.namf");
+ for (int i = 0; i < numDocs; i++) {
+ Map<String, Object> expected = maps.get(i);
+ assertEquals(reader.getMap(i, context), expected);
+ assertEquals(reader.getMapEntryValue(i, context, workloadName),
expected.get("k8s.workload.name"));
+ assertEquals(reader.getMapEntryValueAsString(i, context,
workloadName), expected.get("k8s.workload.name"));
+ assertEquals(reader.getMapEntryValue(i, context, cpus),
expected.get("host_logical_cpus"));
+ assertEquals(reader.getMapEntryValueAsString(i, context, cpus),
String.valueOf(i));
+ assertNull(reader.getMapEntryValue(i, context, absent));
+ assertNull(reader.getMapEntryValueAsString(i, context, absent));
+ }
+ }
+ FileUtils.deleteQuietly(mapSVFile);
+ }
+
static class StringSplitterMV implements Function<String, String[]> {
@Override
public String[] apply(String input) {
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/VarByteChunkSVForwardIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/VarByteChunkSVForwardIndexTest.java
index 920fd3789bc..70ddf5f7ae4 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/VarByteChunkSVForwardIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/forward/VarByteChunkSVForwardIndexTest.java
@@ -23,6 +23,10 @@ import java.io.IOException;
import java.net.URL;
import java.nio.ByteOrder;
import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
import java.util.Random;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.RandomStringUtils;
@@ -36,10 +40,13 @@ import
org.apache.pinot.segment.spi.compression.ChunkCompressionType;
import org.apache.pinot.segment.spi.memory.PinotByteBuffer;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.MapUtils;
+import org.apache.pinot.spi.utils.MapUtils.PreparedMapKey;
import org.testng.Assert;
import org.testng.annotations.Test;
import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.fail;
@@ -80,6 +87,77 @@ public class VarByteChunkSVForwardIndexTest implements
PinotBuffersAfterMethodCh
test(ChunkCompressionType.GZIP);
}
+ /// Covers selective MAP reads through the legacy V2/V3 reader for both
compressed and pass-through chunks. These
+ /// formats are still selected for existing segments and do not share the
V4/V5/V6 reader implementation.
+ @Test
+ public void testMapSelectiveRead()
+ throws Exception {
+ for (ChunkCompressionType compressionType
+ : new ChunkCompressionType[]{ChunkCompressionType.SNAPPY,
ChunkCompressionType.PASS_THROUGH}) {
+ for (int writerVersion : new int[]{2, 3}) {
+ testMapSelectiveRead(compressionType, writerVersion);
+ }
+ }
+ }
+
+ private void testMapSelectiveRead(ChunkCompressionType compressionType, int
writerVersion)
+ throws Exception {
+ int numDocs = 5;
+ List<Map<String, Object>> maps = new ArrayList<>(numDocs);
+ byte[][] frames = new byte[numDocs][];
+ int longestEntry = 0;
+ for (int i = 0; i < numDocs; i++) {
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("k8s.workload.name", "workload-" + i);
+ map.put("k8s.workload.kind", "Deployment");
+ map.put("host_logical_cpus", i);
+ maps.add(map);
+ frames[i] = MapUtils.serializeMap(map);
+ longestEntry = Math.max(longestEntry, frames[i].length);
+ }
+
+ File outFile = Files.createTempFile(getClass().getSimpleName() + "-map-v"
+ writerVersion, ".fwd").toFile();
+ try {
+ try (VarByteChunkForwardIndexWriter writer = new
VarByteChunkForwardIndexWriter(outFile, compressionType,
+ numDocs, 2, longestEntry, writerVersion)) {
+ for (byte[] frame : frames) {
+ writer.putBytes(frame);
+ }
+ }
+
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(outFile);
+ VarByteChunkSVForwardIndexReader reader = new
VarByteChunkSVForwardIndexReader(buffer, DataType.MAP);
+ ChunkReaderContext context = reader.createContext()) {
+ // The inherited fallback returns the same values after materializing
the whole map, so also pin the methods
+ // that must own the selective implementation.
+ assertSame(reader.getClass()
+ .getMethod("getMapEntryValue", int.class,
ChunkReaderContext.class, PreparedMapKey.class)
+ .getDeclaringClass(),
+ VarByteChunkSVForwardIndexReader.class);
+ assertSame(reader.getClass()
+ .getMethod("getMapEntryValueAsString", int.class,
ChunkReaderContext.class, PreparedMapKey.class)
+ .getDeclaringClass(),
+ VarByteChunkSVForwardIndexReader.class);
+ PreparedMapKey workloadName = new PreparedMapKey("k8s.workload.name");
+ PreparedMapKey cpus = new PreparedMapKey("host_logical_cpus");
+ PreparedMapKey absent = new PreparedMapKey("k8s.workload.namf");
+ for (int i = 0; i < numDocs; i++) {
+ Map<String, Object> expected = maps.get(i);
+ Assert.assertEquals(reader.getMap(i, context), expected);
+ Assert.assertEquals(reader.getMapEntryValue(i, context,
workloadName), expected.get("k8s.workload.name"));
+ Assert.assertEquals(reader.getMapEntryValueAsString(i, context,
workloadName),
+ expected.get("k8s.workload.name"));
+ Assert.assertEquals(reader.getMapEntryValue(i, context, cpus),
expected.get("host_logical_cpus"));
+ Assert.assertEquals(reader.getMapEntryValueAsString(i, context,
cpus), String.valueOf(i));
+ Assert.assertNull(reader.getMapEntryValue(i, context, absent));
+ Assert.assertNull(reader.getMapEntryValueAsString(i, context,
absent));
+ }
+ }
+ } finally {
+ FileUtils.deleteQuietly(outFile);
+ }
+ }
+
/// This test writes [#NUM_ENTRIES] using [VarByteChunkForwardIndexWriter].
It then reads
/// the strings & bytes using [VarByteChunkSVForwardIndexReader], and
asserts that what was written is the
/// same as
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
