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 bff70e12b82 Improve JSON decoder selected-field performance (#19127)
bff70e12b82 is described below
commit bff70e12b82caa40a1a9f9308904e51b96ef656a
Author: Xiang Fu <[email protected]>
AuthorDate: Sun Aug 2 15:36:36 2026 -0700
Improve JSON decoder selected-field performance (#19127)
* Improve JSON decoder selected-field performance
* Address JSON decoder review feedback
* Fix benchmark Javadoc style
* Read top-level scalars directly off the parser in the direct decode path
Scalar values in the streaming decode loop previously went through
ObjectReader.readValue, which allocates a fresh DeserializationContext
per value. Read them straight off the JsonParser instead, mirroring
databind's untyped materialization: getNumberValue() preserves the
binary formats' Float and widens oversized ints through the shared
JSONRecordExtractor.convert contract; embedded objects (byte[]) pass
through unchanged. Containers still materialize through databind.
Measured with BenchmarkJsonParsing (-prof gc, JDK 25): ~72 B/op saved
per top-level scalar field on all-field decoding (medium payload:
3177 -> 2529 B/op, -20%); selected-field workloads are unchanged.
Also adds decoder-level coverage for binary scalar shapes (Float never
upcast to Double, byte[] pass-through) and for boolean/null values
through the direct path, including explicit JSON null overwriting a
stale value in a reused row.
* Clarify direct JSON parser contract
Document destination mutation semantics for direct parsing and add decoder
coverage proving sliced payload offsets and lengths are honored.
---
pinot-perf/pom.xml | 4 +
.../apache/pinot/perf/BenchmarkJsonParsing.java | 78 +++++++++++++++++-
.../inputformat/json/JSONMessageDecoder.java | 11 ++-
.../inputformat/json/JSONRecordExtractor.java | 2 +-
.../json/format/AutoDetectPayloadParser.java | 11 +++
.../json/format/JacksonPayloadParser.java | 95 +++++++++++++++++++++-
.../inputformat/json/format/JsonPayloadParser.java | 19 +++++
.../json/format/PostgresJsonbPayloadParser.java | 27 +++++-
.../json/format/TextJsonPayloadParser.java | 15 ++--
.../json/JSONMessageDecoderBinaryTest.java | 18 ++++
.../inputformat/json/JSONMessageDecoderTest.java | 91 +++++++++++++++++++++
11 files changed, 354 insertions(+), 17 deletions(-)
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index b0e8ed9efcb..68fd9816503 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -58,6 +58,10 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-protobuf</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-json</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonParsing.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonParsing.java
index 4c8714e9fad..43f5a67043d 100644
--- a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonParsing.java
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonParsing.java
@@ -25,7 +25,11 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
+import org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder;
+import org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.JsonUtils;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
@@ -80,9 +84,18 @@ public class BenchmarkJsonParsing {
private List<byte[]> _jsonPayloads;
private int _currentIndex = 0;
+ private JSONMessageDecoder _allFieldsDecoder;
+ private JSONMessageDecoder _selectedFieldsDecoder;
+ private JSONRecordExtractor _allFieldsExtractor;
+ private JSONRecordExtractor _selectedFieldsExtractor;
+ private GenericRow _mapAllFieldsRow;
+ private GenericRow _directAllFieldsRow;
+ private GenericRow _mapSelectedFieldsRow;
+ private GenericRow _directSelectedFieldsRow;
@Setup(Level.Trial)
- public void setUp() {
+ public void setUp()
+ throws Exception {
_jsonPayloads = new ArrayList<>(NUM_MESSAGES);
Random random = new Random(42);
@@ -90,6 +103,20 @@ public class BenchmarkJsonParsing {
String json = generateJsonPayload(_payloadType, random, i);
_jsonPayloads.add(json.getBytes(StandardCharsets.UTF_8));
}
+
+ Set<String> selectedFields = selectedFields(_payloadType);
+ _allFieldsDecoder = new JSONMessageDecoder();
+ _allFieldsDecoder.init(Map.of(), null, "benchmark");
+ _selectedFieldsDecoder = new JSONMessageDecoder();
+ _selectedFieldsDecoder.init(Map.of(), selectedFields, "benchmark");
+ _allFieldsExtractor = new JSONRecordExtractor();
+ _allFieldsExtractor.init(null, null);
+ _selectedFieldsExtractor = new JSONRecordExtractor();
+ _selectedFieldsExtractor.init(selectedFields, null);
+ _mapAllFieldsRow = new GenericRow();
+ _directAllFieldsRow = new GenericRow();
+ _mapSelectedFieldsRow = new GenericRow();
+ _directSelectedFieldsRow = new GenericRow();
}
/// Generates different types of JSON payloads to simulate real-world
streaming data.
@@ -229,6 +256,55 @@ public class BenchmarkJsonParsing {
return result;
}
+ /// BASELINE DECODER: materialize the top-level map, then copy all fields
into GenericRow.
+ @Benchmark
+ public GenericRow mapThenExtractAllFields()
+ throws IOException {
+ _mapAllFieldsRow.clear();
+ byte[] payload = getNextPayload();
+ return _allFieldsExtractor.extract(JsonUtils.bytesToMap(payload, 0,
payload.length), _mapAllFieldsRow);
+ }
+
+ /// OPTIMIZED DECODER: stream all top-level values directly into GenericRow.
+ @Benchmark
+ public GenericRow directToGenericRowAllFields() {
+ _directAllFieldsRow.clear();
+ byte[] payload = getNextPayload();
+ return _allFieldsDecoder.decode(payload, _directAllFieldsRow);
+ }
+
+ /// BASELINE DECODER: materialize the full top-level map, then copy selected
fields into GenericRow.
+ @Benchmark
+ public GenericRow mapThenExtractSelectedFields()
+ throws IOException {
+ _mapSelectedFieldsRow.clear();
+ byte[] payload = getNextPayload();
+ return _selectedFieldsExtractor.extract(JsonUtils.bytesToMap(payload, 0,
payload.length), _mapSelectedFieldsRow);
+ }
+
+ /// OPTIMIZED DECODER: materialize selected values only and skip unselected
containers.
+ @Benchmark
+ public GenericRow directToGenericRowSelectedFields() {
+ _directSelectedFieldsRow.clear();
+ byte[] payload = getNextPayload();
+ return _selectedFieldsDecoder.decode(payload, _directSelectedFieldsRow);
+ }
+
+ private static Set<String> selectedFields(String payloadType) {
+ switch (payloadType) {
+ case "small":
+ return Set.of("id", "status");
+ case "medium":
+ return Set.of("eventId", "userId", "timestamp", "country");
+ case "large":
+ return Set.of("eventId", "timestamp");
+ case "nested":
+ return Set.of("order");
+ default:
+ return Set.of("id");
+ }
+ }
+
// Helper methods for generating realistic test data
private static String randomEventType(Random random) {
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java
index 5a23281b6ea..75ce66d50cb 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java
@@ -21,6 +21,7 @@ package org.apache.pinot.plugin.inputformat.json;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadFormat;
import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadParser;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -53,6 +54,8 @@ public class JSONMessageDecoder implements
StreamMessageDecoder<byte[]> {
"org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor";
private RecordExtractor<Map<String, Object>> _jsonRecordExtractor;
+ private Set<String> _fieldsToRead;
+ private boolean _usesDefaultRecordExtractor;
// For AUTO this resolves the concrete format per message; otherwise it is
the pinned format's parser.
private JsonPayloadParser _parser;
@@ -70,6 +73,10 @@ public class JSONMessageDecoder implements
StreamMessageDecoder<byte[]> {
}
_jsonRecordExtractor =
PluginManager.get().createInstance(recordExtractorClass);
_jsonRecordExtractor.init(fieldsToRead, null);
+ _fieldsToRead = CollectionUtils.isNotEmpty(fieldsToRead) ?
Set.copyOf(fieldsToRead) : null;
+ // Direct parsing implements JSONRecordExtractor's conversion contract and
bypasses extract(). Require the
+ // exact default class so a configured extractor or subclass cannot lose
custom extraction behavior.
+ _usesDefaultRecordExtractor = _jsonRecordExtractor.getClass() ==
JSONRecordExtractor.class;
_parser = JsonPayloadFormat.fromConfig(jsonFormat).getParser();
}
@@ -81,7 +88,9 @@ public class JSONMessageDecoder implements
StreamMessageDecoder<byte[]> {
@Override
public GenericRow decode(byte[] payload, int offset, int length, GenericRow
destination) {
try {
- // Parse directly to Map, avoiding an intermediate JsonNode
representation for better performance.
+ if (_usesDefaultRecordExtractor && _parser.parse(payload, offset,
length, destination, _fieldsToRead)) {
+ return destination;
+ }
Map<String, Object> jsonMap = _parser.parse(payload, offset, length);
return _jsonRecordExtractor.extract(jsonMap, destination);
} catch (Exception e) {
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java
index d2fbe09da64..8ada5adbf99 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java
@@ -61,7 +61,7 @@ public class JSONRecordExtractor extends
BaseRecordExtractor<Map<String, Object>
/// Walks a non-null Jackson-parsed value and produces the contract shape:
`BigDecimal` for `BigInteger`
/// (oversized ints), `Object[]` for JSON arrays, `Map<String, Object>` for
JSON objects, pass-through for
/// the other Jackson scalar types (`Boolean`, `Integer`, `Long`, `Double`,
`String`).
- private static Object convert(Object value) {
+ public static Object convert(Object value) {
// BigInteger widens (Pinot has no BigInteger type)
if (value instanceof BigInteger) {
return new BigDecimal((BigInteger) value);
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java
index af1f6783c20..c73d84774e2 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java
@@ -19,6 +19,9 @@
package org.apache.pinot.plugin.inputformat.json.format;
import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.GenericRow;
/// Resolves the concrete parser per message from the payload's leading magic
/ version bytes, then delegates.
@@ -37,4 +40,12 @@ class AutoDetectPayloadParser implements JsonPayloadParser {
throws Exception {
return JsonPayloadFormat.detectParser(payload, offset,
length).parse(payload, offset, length);
}
+
+ @Override
+ public boolean parse(byte[] payload, int offset, int length, GenericRow
destination,
+ @Nullable Set<String> fields)
+ throws Exception {
+ return JsonPayloadFormat.detectParser(payload, offset, length)
+ .parse(payload, offset, length, destination, fields);
+ }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java
index 711c55e2bec..e3dbc4d2d07 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java
@@ -19,9 +19,16 @@
package org.apache.pinot.plugin.inputformat.json.format;
import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
+import java.io.IOException;
import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor;
+import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -31,15 +38,101 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// The [ObjectReader] is immutable and thread-safe, so a single instance is
shared across all decode calls.
abstract class JacksonPayloadParser implements JsonPayloadParser {
+ private final JsonFactory _factory;
private final ObjectReader _mapReader;
+ private final ObjectReader _valueReader;
JacksonPayloadParser(JsonFactory factory) {
- _mapReader = new
ObjectMapper(factory).readerFor(JsonUtils.MAP_TYPE_REFERENCE);
+ this(new ObjectMapper(factory).reader());
+ }
+
+ /// The reader must use default deserialization features: [#readValue]'s
scalar fast path bypasses databind,
+ /// so features like `USE_BIG_DECIMAL_FOR_FLOATS` would apply to containers
but silently not to top-level
+ /// scalars. Every current caller passes a default-configured reader.
+ JacksonPayloadParser(ObjectReader reader) {
+ _factory = reader.getFactory();
+ _mapReader = reader.forType(JsonUtils.MAP_TYPE_REFERENCE);
+ _valueReader = reader.forType(Object.class);
}
@Override
public Map<String, Object> parse(byte[] payload, int offset, int length)
throws Exception {
+ return parseMap(payload, offset, length);
+ }
+
+ protected final Map<String, Object> parseMap(byte[] payload, int offset, int
length)
+ throws Exception {
return _mapReader.readValue(payload, offset, length);
}
+
+ /// Streams the top-level object fields into the row. Nested selected values
still use Jackson's normal
+ /// materialization and are converted by [JSONRecordExtractor], but the
top-level map is never allocated.
+ /// Unselected containers are skipped without materializing their contents.
+ @Override
+ public boolean parse(byte[] payload, int offset, int length, GenericRow
destination,
+ @Nullable Set<String> fields)
+ throws Exception {
+ if (fields != null) {
+ // Match JSONRecordExtractor's missing-field behavior and overwrite
values left in a reused row.
+ for (String field : fields) {
+ destination.putValue(field, null);
+ }
+ }
+ try (JsonParser parser = _factory.createParser(payload, offset, length)) {
+ if (parser.nextToken() != JsonToken.START_OBJECT) {
+ throw new IllegalArgumentException("Top-level JSON value must be an
object");
+ }
+ JsonToken token;
+ while ((token = parser.nextToken()) != JsonToken.END_OBJECT) {
+ if (token == null) {
+ throw new IllegalArgumentException("Unexpected end of JSON object");
+ }
+ if (token != JsonToken.FIELD_NAME) {
+ throw new IllegalArgumentException("Expected a JSON object field,
found: " + token);
+ }
+ String fieldName = parser.currentName();
+ JsonToken valueToken = parser.nextToken();
+ if (valueToken == null) {
+ throw new IllegalArgumentException("Unexpected end of JSON value for
field: " + fieldName);
+ }
+ if (fields == null || fields.contains(fieldName)) {
+ destination.putValue(fieldName, readValue(parser, valueToken));
+ } else if (valueToken.isStructStart()) {
+ parser.skipChildren();
+ }
+ }
+ }
+ return true;
+ }
+
+ /// Materializes the value at the parser's current token in
[JSONRecordExtractor]'s converted shape. Scalars
+ /// are read straight off the parser rather than through databind, which
allocates a fresh
+ /// `DeserializationContext` per `readValue` call. `getNumberValue()` keeps
the binary formats' `Float`
+ /// (never upcast to `Double`) and text JSON's `Double`, matching Jackson's
untyped materialization.
+ @Nullable
+ private Object readValue(JsonParser parser, JsonToken valueToken)
+ throws IOException {
+ switch (valueToken) {
+ case VALUE_STRING:
+ return parser.getText();
+ case VALUE_NUMBER_INT:
+ case VALUE_NUMBER_FLOAT:
+ // Oversized ints widen to BigDecimal via the shared contract (Pinot
has no BigInteger type)
+ return JSONRecordExtractor.convert(parser.getNumberValue());
+ case VALUE_TRUE:
+ return Boolean.TRUE;
+ case VALUE_FALSE:
+ return Boolean.FALSE;
+ case VALUE_NULL:
+ return null;
+ case VALUE_EMBEDDED_OBJECT:
+ // Binary formats' native scalars (e.g. byte[]) pass through
unchanged, as in the map path.
+ return parser.getEmbeddedObject();
+ default:
+ // START_OBJECT / START_ARRAY: containers materialize through databind
and convert recursively.
+ Object value = _valueReader.readValue(parser);
+ return value != null ? JSONRecordExtractor.convert(value) : null;
+ }
+ }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java
index 49b6292f05b..64e6b8e34c5 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java
@@ -19,6 +19,9 @@
package org.apache.pinot.plugin.inputformat.json.format;
import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.GenericRow;
/// Parses a stream payload encoded in a particular (text or binary) JSON
representation into a Jackson-style
@@ -51,4 +54,20 @@ public interface JsonPayloadParser {
/// @throws Exception if the region is not valid for this format
Map<String, Object> parse(byte[] payload, int offset, int length)
throws Exception;
+
+ /// Parses the payload directly into `destination`, avoiding a top-level
per-record map when the
+ /// implementation supports it. When this method returns `false`, it must
leave `destination` unchanged so
+ /// the caller can safely fall back to [#parse]. When it returns `true`, it
must overwrite every requested
+ /// field (using `null` for missing fields) and leave unrequested fields
unchanged. When `fields` is `null`,
+ /// it must populate every top-level field present in the payload. If
parsing throws, `destination` may be
+ /// partially modified and the caller must discard or clear it before reuse.
+ ///
+ /// @param fields fields to populate, or `null` to populate every top-level
field
+ /// @return `true` when the payload was decoded into `destination`; `false`
when the caller
+ /// should fall back to [#parse]
+ default boolean parse(byte[] payload, int offset, int length, GenericRow
destination,
+ @Nullable Set<String> fields)
+ throws Exception {
+ return false;
+ }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java
index 2a6a5b4c9af..e7318f021ae 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java
@@ -19,6 +19,9 @@
package org.apache.pinot.plugin.inputformat.json.format;
import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.utils.JsonUtils;
@@ -32,13 +35,17 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// logical replication via `pgoutput` — routes through that same send
function, so the version-byte + text
/// framing is what a stream actually carries.
///
-/// Because the body is ordinary text JSON, values decode through
[JsonUtils#bytesToMap] and therefore follow
-/// exactly the same type contract as [TextJsonPayloadParser].
-class PostgresJsonbPayloadParser implements JsonPayloadParser {
+/// Because the body is ordinary text JSON, values follow exactly the same
type contract as
+/// [TextJsonPayloadParser].
+class PostgresJsonbPayloadParser extends JacksonPayloadParser {
/// The only version `jsonb_recv` accepts.
private static final byte JSONB_VERSION = 1;
+ PostgresJsonbPayloadParser() {
+ super(JsonUtils.DEFAULT_READER);
+ }
+
@Override
public boolean matches(byte[] payload, int offset, int length) {
// Version byte followed by a text JSON document. Requiring the JSON start
character keeps this from
@@ -50,9 +57,21 @@ class PostgresJsonbPayloadParser implements
JsonPayloadParser {
@Override
public Map<String, Object> parse(byte[] payload, int offset, int length)
throws Exception {
+ validate(payload, offset, length);
+ return parseMap(payload, offset + 1, length - 1);
+ }
+
+ @Override
+ public boolean parse(byte[] payload, int offset, int length, GenericRow
destination,
+ @Nullable Set<String> fields)
+ throws Exception {
+ validate(payload, offset, length);
+ return super.parse(payload, offset + 1, length - 1, destination, fields);
+ }
+
+ private static void validate(byte[] payload, int offset, int length) {
if (length < 2 || payload[offset] != JSONB_VERSION) {
throw new IllegalArgumentException("Payload is not a version-1
PostgreSQL jsonb value");
}
- return JsonUtils.bytesToMap(payload, offset + 1, length - 1);
}
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java
index 397fea32094..e4bb9b2d986 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java
@@ -18,13 +18,16 @@
*/
package org.apache.pinot.plugin.inputformat.json.format;
-import java.util.Map;
import org.apache.pinot.spi.utils.JsonUtils;
/// Parses UTF-8 text JSON, the historical
[org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder]
-/// behavior. Delegates to [JsonUtils#bytesToMap] so the produced value types
exactly match the rest of Pinot.
-class TextJsonPayloadParser implements JsonPayloadParser {
+/// behavior. Uses the same Jackson map/value materialization contract as the
binary JSON parsers.
+class TextJsonPayloadParser extends JacksonPayloadParser {
+
+ TextJsonPayloadParser() {
+ super(JsonUtils.DEFAULT_READER);
+ }
@Override
public boolean matches(byte[] payload, int offset, int length) {
@@ -46,10 +49,4 @@ class TextJsonPayloadParser implements JsonPayloadParser {
}
return false;
}
-
- @Override
- public Map<String, Object> parse(byte[] payload, int offset, int length)
- throws Exception {
- return JsonUtils.bytesToMap(payload, offset, length);
- }
}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java
index 9e21ada953b..64d622a9d3b 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.dataformat.cbor.CBORGenerator;
import com.fasterxml.jackson.dataformat.smile.SmileFactory;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
+import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.pinot.spi.data.readers.GenericRow;
@@ -99,6 +100,23 @@ public class JSONMessageDecoderBinaryTest {
assertRich(decode(Map.of(), RICH_FIELDS, TEXT_DOC));
}
+ /// The direct streaming path materializes top-level scalars straight off
the parser; guard the binary-only
+ /// scalar shapes (Float must not upcast to Double, byte[] must pass
through) at the decoder level, with and
+ /// without a field selection.
+ @Test
+ public void testDirectDecodePreservesBinaryScalars()
+ throws Exception {
+ Map<String, Object> doc = Map.of("f", 1.5f, "bin", new byte[]{1, 2, 3});
+ for (byte[] payload : List.of(smile(doc), cbor(doc))) {
+ for (Set<String> fields : Arrays.asList(null, Set.of("f", "bin"))) {
+ GenericRow row = decode(AUTO, fields, payload);
+ assertTrue(row.getValue("f") instanceof Float, "expected Float, got "
+ row.getValue("f").getClass());
+ assertEquals(row.getValue("f"), 1.5f);
+ assertEquals((byte[]) row.getValue("bin"), new byte[]{1, 2, 3});
+ }
+ }
+ }
+
/// An unset jsonFormat must not silently auto-detect binary payloads: those
streams keep failing as they did
/// before this feature existed, rather than being decoded (or worse,
ingested as a partial row).
@Test
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java
index 5c2bda51b87..2a81b23b533 100644
---
a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java
@@ -22,23 +22,114 @@ import com.fasterxml.jackson.databind.JsonNode;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
+import java.math.BigDecimal;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.stream.StreamMessageDecoder;
import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.fail;
public class JSONMessageDecoderTest {
+ @Test
+ public void testDirectDecodePreservesValueConversion()
+ throws Exception {
+ byte[] payload = ("{\"id\":1,\"huge\":99999999999999999999999999,"
+ + "\"nested\":{\"values\":[1,null,3]},\"tags\":[\"a\",\"b\"],"
+ +
"\"flag\":true,\"off\":false,\"gone\":null}").getBytes(StandardCharsets.UTF_8);
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ decoder.init(Map.of(), null, "testTopic");
+ GenericRow row = new GenericRow();
+ // Without a field selection there is no pre-nulling pass, so an explicit
JSON null must still overwrite
+ // the previous message's value in a reused row.
+ row.putValue("gone", "stale");
+
+ decoder.decode(payload, row);
+
+ assertEquals(row.getValue("id"), 1);
+ assertEquals(row.getValue("huge"), new
BigDecimal("99999999999999999999999999"));
+ assertEquals((Object[]) ((Map<?, ?>)
row.getValue("nested")).get("values"), new Object[]{1, null, 3});
+ assertEquals((Object[]) row.getValue("tags"), new Object[]{"a", "b"});
+ assertEquals(row.getValue("flag"), Boolean.TRUE);
+ assertEquals(row.getValue("off"), Boolean.FALSE);
+ assertEquals(row.getValue("gone"), null);
+ }
+
+ @Test
+ public void testDirectDecodeSelectedFieldsOverwritesMissingValues()
+ throws Exception {
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ decoder.init(Map.of(), Set.of("id", "missing"), "testTopic");
+ GenericRow row = new GenericRow();
+ row.putValue("missing", "stale");
+ row.putValue("unselected", "preserved");
+
+
decoder.decode("{\"id\":1,\"ignored\":[1,2,3]}".getBytes(StandardCharsets.UTF_8),
row);
+
+ assertEquals(row.getValue("id"), 1);
+ assertEquals(row.getValue("missing"), null);
+ assertEquals(row.getValue("unselected"), "preserved");
+ assertEquals(row.getFieldToValueMap().keySet(), Set.of("id", "missing",
"unselected"));
+ }
+
+ @Test
+ public void testDirectDecodeHonorsOffsetAndLength()
+ throws Exception {
+ String prefix = "invalid-prefix";
+ String record = "{\"id\":1,\"name\":\"alice\"}";
+ byte[] payload = (prefix + record +
"invalid-suffix").getBytes(StandardCharsets.UTF_8);
+ int offset = prefix.getBytes(StandardCharsets.UTF_8).length;
+ int length = record.getBytes(StandardCharsets.UTF_8).length;
+
+ JSONMessageDecoder allFieldsDecoder = new JSONMessageDecoder();
+ allFieldsDecoder.init(Map.of(), null, "testTopic");
+ GenericRow allFieldsRow = allFieldsDecoder.decode(payload, offset, length,
new GenericRow());
+ assertEquals(allFieldsRow.getFieldToValueMap(), Map.of("id", 1, "name",
"alice"));
+
+ JSONMessageDecoder selectedFieldsDecoder = new JSONMessageDecoder();
+ selectedFieldsDecoder.init(Map.of(), Set.of("id"), "testTopic");
+ GenericRow selectedFieldsRow = selectedFieldsDecoder.decode(payload,
offset, length, new GenericRow());
+ assertEquals(selectedFieldsRow.getFieldToValueMap(), Map.of("id", 1));
+
+ // Exclude the closing brace to prove the parser honors length instead of
reading the remaining array.
+ assertThrows(RuntimeException.class,
+ () -> allFieldsDecoder.decode(payload, offset, length - 1, new
GenericRow()));
+ }
+
+ @Test
+ public void testCustomExtractorUsesMapFallback()
+ throws Exception {
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ decoder.init(Map.of(StreamMessageDecoder.RECORD_EXTRACTOR_CONFIG_KEY,
CustomJSONRecordExtractor.class.getName()),
+ Set.of("id"), "testTopic");
+
+ GenericRow row =
decoder.decode("{\"id\":1}".getBytes(StandardCharsets.UTF_8), new GenericRow());
+
+ assertEquals(row.getValue("id"), 1);
+ assertEquals(row.getValue("custom"), true);
+ }
+
+ public static class CustomJSONRecordExtractor extends JSONRecordExtractor {
+ @Override
+ public GenericRow extract(Map<String, Object> from, GenericRow to) {
+ to.putValue("custom", true);
+ return super.extract(from, to);
+ }
+ }
+
@Test
public void testJsonDecoderWithoutOutgoingTimeSpec()
throws Exception {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
