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 cd95aef7f68 Render MAP projections straight from the serialized frame
(#19169)
cd95aef7f68 is described below
commit cd95aef7f687a2d59dca90d482d4452f39c0a4e5
Author: Xiang Fu <[email protected]>
AuthorDate: Thu Aug 13 13:02:29 2026 -0700
Render MAP projections straight from the serialized frame (#19169)
Projecting a whole MAP column as a string parsed every entry into a
HashMap and then serialized that map back to JSON, once per row. Both
steps are avoidable: the frame already stores each value as the JSON
bytes serializeMap produced, so the values can be copied through
verbatim and only the keys need quoting.
Add MapUtils#frameToJsonString and a ForwardIndexReader#getMapAsJsonString
hook, overridden by the three readers that hold the map as a frame. The
default still materializes the map, so a reader that keeps the map
columnar-decomposed is unaffected.
Output is unchanged. Both forward-index write paths - ForwardIndexCreator
at segment build and MutableSegmentImpl while consuming - frame maps
through the key-sorting serializeMap, and nested values are sorted by the
same writer, so emitting in frame order reproduces what
toString(deserializeMap(frame)) produced. A test pins that equivalence
over scalars, nesting, unicode, and keys needing escapes.
Isolated JMH, JDK 25:
entries shape before after before B/op after B/op
4 flat 0.866 0.258 4768 768
4 nested 1.312 0.274 7320 848
16 flat 3.261 1.018 16936 2688
16 nested 5.352 1.049 26968 3008
64 flat 14.111 3.885 65656 10464
64 nested 24.071 4.227 104312 11744
3.2-5.7x faster, 6-9x less garbage. Nested values cost the old path 71%
more than flat ones at 64 entries because Jackson materializes a
container per value; the new path is within 9% of flat since it never
looks inside a value.
Co-authored-by: Claude Opus 5 <[email protected]>
---
.../apache/pinot/perf/BenchmarkMapProjection.java | 92 +++++++++++++++++
.../impl/forward/VarByteSVMutableForwardIndex.java | 5 +
.../forward/VarByteChunkForwardIndexReaderV4.java | 5 +
.../forward/VarByteChunkSVForwardIndexReader.java | 5 +
.../spi/index/reader/ForwardIndexReader.java | 15 ++-
.../java/org/apache/pinot/spi/utils/MapUtils.java | 110 +++++++++++++++++++++
.../org/apache/pinot/spi/utils/MapUtilsTest.java | 58 +++++++++++
7 files changed, 289 insertions(+), 1 deletion(-)
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapProjection.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapProjection.java
new file mode 100644
index 00000000000..a3e7d41e18b
--- /dev/null
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkMapProjection.java
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.perf;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.spi.utils.MapUtils;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.ChainedOptionsBuilder;
+import org.openjdk.jmh.runner.options.CommandLineOptions;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+
+/// Measures projecting a whole MAP column as a JSON string - the `SELECT
attributes` / `LASTWITHTIME(attributes)`
+/// shape - rendering straight from the serialized frame versus deserializing
into a map and serializing it again.
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@Fork(value = 2)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@State(Scope.Thread)
+public class BenchmarkMapProjection {
+
+ public static void main(String[] args)
+ throws Exception {
+ ChainedOptionsBuilder opt = new OptionsBuilder().parent(new
CommandLineOptions(args))
+ .include(BenchmarkMapProjection.class.getSimpleName());
+ new Runner(opt.build()).run();
+ }
+
+ @Param({"4", "16", "64"})
+ private int _numEntries;
+
+ /// `flat` models scalar attribute maps; `nested` models object-valued
entries, where the deserializing path pays
+ /// to materialize a container per value.
+ @Param({"flat", "nested"})
+ private String _valueShape;
+
+ private byte[] _serialized;
+
+ @Setup(Level.Trial)
+ public void setUp() {
+ boolean nested = "nested".equals(_valueShape);
+ Map<String, Object> map = new LinkedHashMap<>();
+ for (int i = 0; i < _numEntries; i++) {
+ String value = "value-with-enough-bytes-to-exercise-json-parsing-" + i;
+ map.put(String.format("k8s.attribute.%03d.name", i), nested ?
Map.of("value", value) : value);
+ }
+ _serialized = MapUtils.serializeMap(map);
+ }
+
+ /// The existing path: parse every entry into a map, then serialize the map
back to JSON.
+ @Benchmark
+ public String deserializeThenSerialize() {
+ return MapUtils.toString(MapUtils.deserializeMap(_serialized));
+ }
+
+ /// The optimized path: copy the already-JSON value bytes through and quote
only the keys.
+ @Benchmark
+ public String renderFromFrame() {
+ return MapUtils.frameToJsonString(_serialized);
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/VarByteSVMutableForwardIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/VarByteSVMutableForwardIndex.java
index 6672e392e0f..cb2bc35f9da 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/VarByteSVMutableForwardIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/forward/VarByteSVMutableForwardIndex.java
@@ -108,6 +108,11 @@ public class VarByteSVMutableForwardIndex implements
MutableForwardIndex {
return
MapUtils.deserializeMapEntryValue(_byteArrayStore.getByteBuffer(docId), key);
}
+ @Override
+ public String getMapAsJsonString(int docId, ForwardIndexReaderContext
context) {
+ return MapUtils.frameToJsonString(getBytes(docId));
+ }
+
@Override
public void setBigDecimal(int docId, BigDecimal value) {
setBytes(docId, BigDecimalUtils.serialize(value));
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 2e60c6cd916..0d65fba1b26 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
@@ -138,6 +138,11 @@ public class VarByteChunkForwardIndexReaderV4
return MapUtils.deserializeMap(context.getValue(docId));
}
+ @Override
+ public String getMapAsJsonString(int docId, ReaderContext context) {
+ return MapUtils.frameToJsonString(context.getValue(docId));
+ }
+
@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 28b79581a88..b732e483243 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
@@ -130,6 +130,11 @@ public final class VarByteChunkSVForwardIndexReader
extends BaseChunkForwardInde
return MapUtils.deserializeMap(getBytes(docId, context));
}
+ @Override
+ public String getMapAsJsonString(int docId, ChunkReaderContext context) {
+ return MapUtils.frameToJsonString(getBytes(docId, context));
+ }
+
/// 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-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/ForwardIndexReader.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/ForwardIndexReader.java
index 66749669d94..1a9c37b0f7c 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/ForwardIndexReader.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/reader/ForwardIndexReader.java
@@ -385,7 +385,7 @@ public interface ForwardIndexReader<T extends
ForwardIndexReaderContext> extends
break;
case MAP:
for (int i = 0; i < length; i++) {
- values[i] = MapUtils.toString(getMap(docIds[i], context));
+ values[i] = getMapAsJsonString(docIds[i], context);
}
break;
default:
@@ -486,6 +486,19 @@ public interface ForwardIndexReader<T extends
ForwardIndexReaderContext> extends
return getMap(docId, context).get(key.getKey());
}
+ /// Reads the MAP type single-value at the given document id, rendered as a
JSON object string.
+ ///
+ /// Readers that store the map as a serialized frame should override this to
render straight from those bytes
+ /// (see [MapUtils#frameToJsonString(byte\[\])]); the default here
materializes the map first and serializes it
+ /// again, which is what a reader holding the map columnar-decomposed has to
do anyway.
+ ///
+ /// @param docId Document id
+ /// @param context Reader context
+ /// @return MAP type single-value at the given document id, as JSON
+ default String getMapAsJsonString(int docId, T context) {
+ return MapUtils.toString(getMap(docId, context));
+ }
+
default int get32BitsMurmur3Hash(int docId, T context) {
return MurmurHashFunctions.murmurHash3X64Bit32(getBytes(docId, context),
0);
}
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/MapUtils.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/MapUtils.java
index a6e5011e384..26362b9f22e 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/MapUtils.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/MapUtils.java
@@ -19,6 +19,7 @@
package org.apache.pinot.spi.utils;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.io.JsonStringEncoder;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.google.common.base.Preconditions;
@@ -28,6 +29,7 @@ import java.io.OutputStream;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map;
@@ -299,6 +301,114 @@ public class MapUtils {
}
}
+ /// Renders a serialized MAP frame as a JSON object without materializing
the map.
+ ///
+ /// The frame already stores each value as the JSON bytes that
[#serializeMap] produced, so the values are copied
+ /// through verbatim and only the keys are quoted. That skips the
parse-into-`HashMap`-then-serialize-again round
+ /// trip [#toString(Map)] performs, and it skips Jackson entirely.
+ ///
+ /// Entries are emitted in frame order. Both forward-index write paths
(`ForwardIndexCreator#putValue` at segment
+ /// build and `MutableSegmentImpl` while consuming) frame maps through the
key-sorting [#serializeMap(Map)], so for
+ /// those frames this is byte-identical to
`toString(deserializeMap(frame))`. A frame written through
+ /// [#serializeMap(Map, boolean)] with `sortByKey = false` renders in its
own insertion order instead.
+ public static String frameToJsonString(byte[] bytes) {
+ return frameToJsonString(ByteBuffer.wrap(bytes));
+ }
+
+ /// Variant of [#frameToJsonString(byte\[\])] reading from the buffer's
current position.
+ public static String frameToJsonString(ByteBuffer byteBuffer) {
+ byteBuffer.order(ByteOrder.BIG_ENDIAN);
+ int size = byteBuffer.getInt();
+ if (size == 0) {
+ return "{}";
+ }
+ // Quoting a key adds 2 bytes and the separators add 2, while the two
length prefixes it replaces free up 8, so
+ // an unescaped rendering never exceeds the remaining frame bytes.
Escaping is the only path that can grow.
+ JsonBuilder builder = new JsonBuilder(byteBuffer.remaining() + 2);
+ builder.append((byte) '{');
+ for (int i = 0; i < size; i++) {
+ if (i > 0) {
+ builder.append((byte) ',');
+ }
+ int keyLength = byteBuffer.getInt();
+ checkLength(byteBuffer, keyLength);
+ builder.appendQuotedKey(byteBuffer, keyLength);
+ byteBuffer.position(byteBuffer.position() + keyLength);
+ builder.append((byte) ':');
+ int valueLength = byteBuffer.getInt();
+ checkLength(byteBuffer, valueLength);
+ builder.appendRaw(byteBuffer, valueLength);
+ byteBuffer.position(byteBuffer.position() + valueLength);
+ }
+ builder.append((byte) '}');
+ return builder.toUtf8String();
+ }
+
+ /// Growable byte sink for [#frameToJsonString]. Assembling UTF-8 bytes and
decoding once at the end avoids
+ /// decoding every key individually, which is what makes the common
all-ASCII frame allocation-light.
+ private static final class JsonBuilder {
+ private byte[] _bytes;
+ private int _length;
+
+ private JsonBuilder(int capacity) {
+ _bytes = new byte[capacity];
+ }
+
+ private void ensure(int extra) {
+ if (_length + extra > _bytes.length) {
+ _bytes = Arrays.copyOf(_bytes, Math.max(_length + extra, _bytes.length
* 2));
+ }
+ }
+
+ private void append(byte b) {
+ ensure(1);
+ _bytes[_length++] = b;
+ }
+
+ /// Copies `length` bytes from the buffer's current position without
advancing it.
+ private void appendRaw(ByteBuffer byteBuffer, int length) {
+ ensure(length);
+ int offset = byteBuffer.position();
+ for (int i = 0; i < length; i++) {
+ _bytes[_length++] = byteBuffer.get(offset + i);
+ }
+ }
+
+ /// Writes the key as a JSON string. Keys needing no escaping - the
overwhelming majority - are copied as raw
+ /// UTF-8; anything else falls back to decoding the key and letting
Jackson escape it.
+ private void appendQuotedKey(ByteBuffer byteBuffer, int length) {
+ int offset = byteBuffer.position();
+ boolean clean = true;
+ for (int i = 0; i < length; i++) {
+ byte b = byteBuffer.get(offset + i);
+ // Signed bytes below 0x20 are control characters; continuation bytes
of multi-byte UTF-8 are negative and
+ // need no escaping, so only the ASCII range is inspected.
+ if ((b >= 0 && b < 0x20) || b == '"' || b == '\\') {
+ clean = false;
+ break;
+ }
+ }
+ append((byte) '"');
+ if (clean) {
+ appendRaw(byteBuffer, length);
+ } else {
+ byte[] keyBytes = new byte[length];
+ for (int i = 0; i < length; i++) {
+ keyBytes[i] = byteBuffer.get(offset + i);
+ }
+ byte[] escaped =
JsonStringEncoder.getInstance().quoteAsUTF8(Utf8Utils.decode(keyBytes));
+ ensure(escaped.length);
+ System.arraycopy(escaped, 0, _bytes, _length, escaped.length);
+ _length += escaped.length;
+ }
+ append((byte) '"');
+ }
+
+ private String toUtf8String() {
+ return new String(_bytes, 0, _length, StandardCharsets.UTF_8);
+ }
+ }
+
public static String toString(Map<String, Object> map) {
return toString(map, true);
}
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/MapUtilsTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/MapUtilsTest.java
index 4b90ba8ee97..b67edb879b3 100644
--- a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/MapUtilsTest.java
+++ b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/MapUtilsTest.java
@@ -381,4 +381,62 @@ public class MapUtilsTest {
map.put("nested", nested);
assertEquals(MapUtils.toString(map),
"{\"nested\":{\"d\":\"2022-02-08\"}}");
}
+
+ /// The contract that lets the forward-index read path swap
`toString(deserializeMap(frame))` for
+ /// `frameToJsonString(frame)`: for any frame written through the
key-sorting [MapUtils#serializeMap(Map)] - which
+ /// is what both forward-index write paths use - the two must produce
identical output.
+ @Test
+ void testFrameToJsonStringMatchesToString() {
+ Map<String, Object> nested = new LinkedHashMap<>();
+ nested.put("z", 1);
+ nested.put("a", List.of(1, 2, 3));
+
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("k8s.workload.name", "pinot-server");
+ map.put("int", 42);
+ map.put("long", 9999999999L);
+ map.put("double", 1.5);
+ map.put("bool", true);
+ map.put("nullValue", null);
+ map.put("nested", nested);
+ map.put("list", List.of("a", "b"));
+ map.put("emptyString", "");
+ map.put("unicodeValue", "çöğüşÇÖĞÜŞéÉ");
+ map.put("命名空间", "namespace");
+ map.put("date", LocalDate.of(2022, 2, 8));
+ map.put("quote\"key", "quoted");
+ map.put("back\\slash", "escaped");
+ map.put("tab\tkey", "control");
+ map.put("value with \"quotes\" and \\ backslash", "in value");
+
+ assertEquals(MapUtils.frameToJsonString(MapUtils.serializeMap(map)),
MapUtils.toString(map));
+ }
+
+ @Test
+ void testFrameToJsonStringHandlesEmptyMap() {
+ assertEquals(MapUtils.frameToJsonString(MapUtils.serializeMap(Map.of())),
"{}");
+ assertEquals(MapUtils.frameToJsonString(MapUtils.serializeMap(Map.of())),
MapUtils.toString(Map.of()));
+ }
+
+ /// Rendering must round-trip back through the JSON reader to the same map,
independent of the string comparison
+ /// above - that guards against two implementations agreeing on malformed
output.
+ @Test
+ void testFrameToJsonStringRoundTrips() {
+ Map<String, Object> map = new LinkedHashMap<>();
+ map.put("a", "value");
+ map.put("b", List.of(1, 2));
+ map.put("çö", Map.of("inner", true));
+
assertEquals(MapUtils.fromString(MapUtils.frameToJsonString(MapUtils.serializeMap(map))),
+ MapUtils.deserializeMap(MapUtils.serializeMap(map)));
+ }
+
+ /// An off-heap forward-index view arrives in the platform's native byte
order while the frame is written
+ /// big-endian, so the renderer has to force the order rather than trust the
buffer.
+ @Test
+ void testFrameToJsonStringForcesBigEndian() {
+ byte[] serialized = MapUtils.serializeMap(Map.of("k8s.workload.name",
"pinot-server"));
+ ByteBuffer littleEndian =
ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);
+
+ assertEquals(MapUtils.frameToJsonString(littleEndian),
"{\"k8s.workload.name\":\"pinot-server\"}");
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
