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 d59403e7441 Support binary JSON payload formats (PostgreSQL jsonb,
SQLite JSONB, Smile, CBOR) in the JSON stream decoder (#18953)
d59403e7441 is described below
commit d59403e74410ced40350a735b6568cc56734ac49
Author: Xiang Fu <[email protected]>
AuthorDate: Tue Jul 14 03:34:31 2026 +0200
Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile,
CBOR) in the JSON stream decoder (#18953)
* Support binary JSON payload formats in the JSON stream decoder
JSONMessageDecoder only understood UTF-8 text JSON. Add a pluggable
payload-format layer so a stream can also carry PostgreSQL jsonb, SQLite
JSONB, Smile, or CBOR, selected with the new `jsonFormat` decoder
property.
When `jsonFormat` is unset (equivalently AUTO) the encoding is detected
per message from its leading magic/version bytes, falling back to text
JSON. Detection is allocation-free and cannot mis-route a well-formed
text JSON document: a top-level `{`/`[`, optionally after whitespace,
collides with none of the binary signatures. Pin `TEXT` to skip
detection entirely.
Every format decodes to the same `Map<String, Object>` contract Jackson
produces for text JSON, so the existing JSONRecordExtractor handles all
of them unchanged.
Notes on the two hand-rolled decoders:
- PostgreSQL jsonb's wire format is *not* its on-disk JsonbContainer
layout. `jsonb_send` renders the value with JsonbToCString and emits a
version byte followed by text JSON; `jsonb_recv` reverses it. Every
standard binary producer (v3 protocol, COPY ... WITH (FORMAT binary),
logical replication via pgoutput) routes through that send function,
so this parser strips the version byte and parses the text body.
- SQLite JSONB (3.45+) is a genuine binary format and is decoded per
spec, including the JSON5 element types. Nested elements are bounded
by their enclosing container, not merely by the payload, so a crafted
element cannot overrun its parent and silently swallow later siblings.
Adds jackson-dataformat-smile and jackson-dataformat-cbor (versions
managed by jackson-bom) and records both in LICENSE-binary.
* Address review: reject SQLite JSONB payloads not exactly filled by their
top-level element
SQLite's JSONB validity rule requires the outer element to exactly fill
the BLOB. Without that check a top-level element declaring a short size
(e.g. a bare 0x0C empty object followed by real data) decoded to a
partial -- possibly empty -- row and silently discarded the trailing
bytes, which AUTO made reachable for any payload whose first byte
carries the OBJECT nibble. Reject when the cursor does not land on the
payload limit, and cover it with parser- and decoder-level regression
tests.
Also from review:
- Fix a malformed nested {@code {} construct in the JSONMessageDecoder
Javadoc that terminated the inline tag early.
- Keep the original exception as the cause when fromConfig() rejects an
unsupported jsonFormat.
- Rename testDefaultIsText -> testUnsetFormatAutoDetectsText; the
default is AUTO, not a pinned TEXT format.
- Pin test fixtures to UTF-8 instead of the platform default charset.
Adds coverage for the previously untested SQLite branches: size
descriptors 14/15 (including rejection of an oversized uint64 size),
signed hex INT5, FLOAT5 with a leading '+', the \u escape, the TEXT5
escape extensions, and the missing-value / non-text-label errors.
* Bound the decode-failure diagnostic and pin it to an explicit charset
The decode() failure message echoed the whole payload via
new String(payload, offset, length), which uses the platform default
charset. That was tolerable when the payload was always text JSON, but
the binary encodings make it a liability: an arbitrarily large Smile /
CBOR / SQLite JSONB message was decoded as mojibake and dumped into the
log in full.
Render at most the first 128 bytes, as UTF-8 when they contain no
control characters other than whitespace and as hex otherwise, and
always report the true payload length. Text payloads keep their readable
rendering; binary payloads become a short hex prefix.
* Use separate text and hex windows for the decode-failure diagnostic
A single 128-byte cap kept binary payloads out of the log but truncated
malformed text records before the offending field, losing the operator
context the old full-payload message provided.
Render up to 512 bytes of text but only 128 bytes as hex; since hex costs
two characters per byte, both produce a message of comparable length.
* Assert the two decode-diagnostic behaviors that were claimed but untested
Both changes were described in the review thread but nothing verified
them, so a regression would have gone unnoticed:
- fromConfig() now has a test that captures the IllegalArgumentException
and asserts both the message contents and that the underlying valueOf
failure survives as the cause. The previous assertThrows passed even
before the cause was chained.
- describePayload()'s "bytes >= 0x80 are not control characters, so
non-ASCII UTF-8 still renders as text" branch had no fixture: every
case was pure ASCII or contained a low control byte forcing the hex
path. Added a multibyte UTF-8 payload.
* Default unset jsonFormat to TEXT and gate SQLite auto-detection on exact
fill
Defaulting an unset jsonFormat to AUTO turned per-message detection on
for every existing JSON stream, which is a backward-compatibility and
data-correctness change: SqliteJsonbPayloadParser.matches() gated only on
the OBJECT low nibble, so roughly one arbitrary binary byte in sixteen
claimed the payload, and a lone 0x0c is a validly-exact empty object. A
corrupt message that previously failed text decoding could therefore be
ingested as an empty row.
Two changes:
- An unset jsonFormat now resolves to TEXT, preserving the decoder's
historical behavior byte for byte. AUTO must be requested explicitly.
- SQLite detection additionally requires the top-level object's declared
size to exactly fill the payload -- the same validity rule parse()
enforces -- so AUTO can no longer hand a corrupt message to that parser
on the strength of a single nibble. The check is allocation-free and
covers all four size descriptors, including an 8-byte size whose sign
bit is set.
Tests assert that an unset format decodes text and rejects Smile, CBOR,
Postgres, SQLite and a bare 0x0c; that explicit AUTO still detects each
format; and that detection accepts only exactly-filling SQLite objects.
* Share one header decoder between SQLite detection and parsing
matches() and readElement() each carried their own copy of the element
header layout (size-descriptor -> header width and declared size). They
agreed, but nothing held them to it: every existing size-descriptor test
drives parse(), so a byte-order slip in the matches() copy -- reading a
2-byte size as 0x0400 rather than 0x0004, say -- would have shipped
silently, either rejecting a valid SQLite object so AUTO fell back to
text, or claiming a payload that does not fill.
Extract headerLength() and declaredSize() and call them from both, so
detection and parsing can no longer disagree about the layout, and the
parse() tests now exercise the same code detection uses.
Assert matches() over every size descriptor rather than only 14: an
exactly-filling and an off-by-one case for each of 12/13/14/15, a
byte-order guard for the 2-byte size, truncated headers for each width,
and that anything matches() claims parse() also accepts.
* Add pinot-json README documenting the jsonFormat decoder property
A user manual for the JSON input format plugin: the jsonFormat values,
the TEXT default and opt-in AUTO detection (with the per-message
detection order and its signatures), realtime table config examples for
text, a pinned binary format, and AUTO, plus per-format notes and the
decode-failure diagnostic behavior.
* Reject non-canonical SQLite JSONB numbers and invalid UTF-8
The SQLite JSONB decoder was more permissive than text JSON on two paths,
letting a corrupt stream message ingest a value Jackson would reject:
- Canonical numbers (TYPE_INT / TYPE_FLOAT) shared the permissive JSON5
parsers. Double.parseDouble accepts NaN, Infinity, a leading '+', Java
hex floats and type suffixes; Long.parseLong / BigInteger accept a
leading '+' and leading zeros. So a malformed canonical number decoded
to a non-finite or non-canonical value instead of failing the record.
Validate TYPE_INT / TYPE_FLOAT against the RFC 8259 number grammar
before parsing, keeping the permissive handling only for the JSON5
types (TYPE_INT5 / TYPE_FLOAT5).
- Text was decoded with new String(..., UTF_8), which substitutes U+FFFD
for malformed bytes. A corrupt record could therefore ingest mutated
field names or values. Decode with a CharsetDecoder configured for
CodingErrorAction.REPORT so invalid UTF-8 fails the record.
Adds regressions: canonical NaN / Infinity / +1.5 / hex-float and +5 /
007 are rejected while the JSON5 variants and canonical happy paths still
decode; valid multibyte UTF-8 decodes while a lone invalid byte in a
value or field name is rejected.
* Cap SQLite JSONB nesting depth and numeric-token length
Two hardening fixes on the SQLite JSONB decode path (per review):
- Blocker: readElement -> readArray/readObject recursed one frame per
nesting level with no depth cap. Each level is only ~1-3 wire bytes, so
a small (sub-MB) but deeply nested message could throw
StackOverflowError. Because that is an Error, not an Exception, it
escaped both JSONMessageDecoder.decode and StreamDataDecoderImpl.decode
(which catch Exception) and propagated to the consumer thread's
catch (Throwable), moving the whole consuming segment to ERROR -- a
poison pill from one bad record. Cap nesting at 1000 (Jackson's
StreamReadConstraints default) and throw IllegalArgumentException past
it, so deep nesting is handled as a bad record like every other
malformed input.
- Numeric tokens fell back to new BigInteger(text), which is ~O(n^2). An
unbounded digit run (up to the payload size) was a per-message CPU sink
with no analogue in text JSON, which caps number length via Jackson.
Cap numeric tokens at 1000 characters to match.
Adds deep-nesting and oversized-numeric regression tests (the existing
tests only exercised shallow nesting), and documents the bare-0x0C
empty-object AUTO caveat in the README.
---
LICENSE-binary | 2 +
.../pinot-input-format/pinot-json/README.md | 170 ++++++
.../pinot-input-format/pinot-json/pom.xml | 12 +
.../inputformat/json/JSONMessageDecoder.java | 73 ++-
.../json/format/AutoDetectPayloadParser.java | 40 ++
.../json/format/CborJsonPayloadParser.java | 49 ++
.../json/format/JacksonPayloadParser.java | 45 ++
.../inputformat/json/format/JsonPayloadFormat.java | 89 +++
.../inputformat/json/format/JsonPayloadParser.java | 54 ++
.../json/format/PostgresJsonbPayloadParser.java | 58 ++
.../json/format/SmileJsonPayloadParser.java | 44 ++
.../json/format/SqliteJsonbPayloadParser.java | 508 ++++++++++++++++
.../json/format/TextJsonPayloadParser.java | 55 ++
.../json/JSONMessageDecoderBinaryTest.java | 263 ++++++++
.../json/format/JsonPayloadFormatTest.java | 670 +++++++++++++++++++++
15 files changed, 2128 insertions(+), 4 deletions(-)
diff --git a/LICENSE-binary b/LICENSE-binary
index 8d688bb0936..9430db9a15f 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -213,7 +213,9 @@ com.dynatrace.hash4j:hash4j:0.30.0
com.fasterxml.jackson.core:jackson-annotations:2.21
com.fasterxml.jackson.core:jackson-core:2.21.1
com.fasterxml.jackson.core:jackson-databind:2.21.1
+com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.21.1
com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.21.1
+com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.21.1
com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.21.1
com.fasterxml.jackson.datatype:jackson-datatype-guava:2.21.1
com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.21.1
diff --git a/pinot-plugins/pinot-input-format/pinot-json/README.md
b/pinot-plugins/pinot-input-format/pinot-json/README.md
new file mode 100644
index 00000000000..0893ef81fde
--- /dev/null
+++ b/pinot-plugins/pinot-input-format/pinot-json/README.md
@@ -0,0 +1,170 @@
+<!--
+
+ 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.
+
+-->
+
+# Pinot JSON input format
+
+This plugin reads JSON records into Pinot. It provides:
+
+- `JSONMessageDecoder` — a `StreamMessageDecoder` for real-time (streaming)
ingestion.
+- `JSONRecordReader` — a `RecordReader` for batch ingestion of
newline-delimited JSON files.
+- `JSONRecordExtractor` — the shared extractor that turns a parsed JSON object
into a Pinot `GenericRow`.
+
+By default the stream decoder reads **UTF-8 text JSON**, exactly as it always
has. It can additionally decode
+several **binary** JSON encodings, selected with the `jsonFormat` decoder
property described below.
+
+> The `jsonFormat` property applies to the **stream decoder**
(`JSONMessageDecoder`) only. Batch ingestion via
+> `JSONRecordReader` always reads text JSON.
+
+## `jsonFormat` decoder property
+
+Set the decoder property `jsonFormat` (full stream-config key
`stream.<type>.decoder.prop.jsonFormat`) to one
+of:
+
+| `jsonFormat` | Payload encoding
|
+|------------------|-------------------------------------------------------------------------|
+| *(unset)* / `TEXT` | UTF-8 text JSON. The historical behavior; unchanged.
|
+| `POSTGRES_JSONB` | PostgreSQL `jsonb` binary wire format (version byte +
text JSON). |
+| `SQLITE_JSONB` | SQLite 3.45+ JSONB binary format.
|
+| `SMILE` | [Jackson
Smile](https://github.com/FasterXML/smile-format-specification) binary JSON. |
+| `CBOR` | [CBOR](https://www.rfc-editor.org/rfc/rfc8949.html) (RFC
8949). |
+| `AUTO` | Detect the encoding per message from its leading bytes
(opt-in). |
+
+The value is case-insensitive. An unrecognized value fails table creation with
a message listing the supported
+values.
+
+Every format decodes to the same value contract, so the same table config,
schema, and transform functions
+work regardless of the wire encoding. Numbers narrow to `INT` / `LONG` /
`BIG_DECIMAL`, decimals to `DOUBLE`,
+and objects/arrays to nested JSON — just as for text JSON. (The binary formats
can additionally carry native
+`float` and `byte[]` scalars, which Pinot converts to the target column type.)
+
+## Default behavior and `AUTO`
+
+An unset `jsonFormat` means `TEXT`, so **existing tables are completely
unaffected** by this feature.
+
+`AUTO` is opt-in rather than the default. Detection is a heuristic over a few
leading bytes; it never
+mis-routes a well-formed text JSON document (a top-level `{` or `[`,
optionally after whitespace, matches none
+of the binary signatures), but it can claim a *corrupt* message that text
decoding would otherwise have
+rejected. Turning that on implicitly for every existing stream would be a
silent behavior change, so you ask
+for it explicitly.
+
+When `AUTO` is set, each message is classified in this order, falling back to
text JSON:
+
+| Order | Format | Signature
|
+|-------|------------------|------------------------------------------------------------------------------|
+| 1 | `SMILE` | 3-byte header `3A 29 0A` (`:)\n`).
|
+| 2 | `CBOR` | Self-describe tag `D9 D9 F7` (RFC 8949 §3.4.6).
CBOR **without** the tag is not detected. |
+| 3 | `POSTGRES_JSONB` | Version byte `01` followed by a JSON document
start. |
+| 4 | `SQLITE_JSONB` | A top-level object whose declared size exactly
fills the payload. |
+| 5 | `TEXT` | First non-whitespace byte is `{` or `[`.
|
+
+Pin an explicit format instead of `AUTO` when you know the encoding: it skips
detection and, for formats
+without a strong magic number (notably tag-less CBOR), is the only way to
decode them.
+
+The SQLite signature is intentionally narrow, but not zero-risk: a message
consisting of the single byte
+`0x0C` is a valid, exactly-filling empty SQLite object, so under `AUTO` a
stray `0x0C` decodes to an empty row
+rather than being rejected. This is the heuristic nature of `AUTO`; pin `TEXT`
(or your actual format) to avoid
+it entirely.
+
+## Examples
+
+All examples use a Kafka real-time table; substitute your stream type
(`stream.<type>.…`) as needed.
+
+### 1. Text JSON (default — no configuration needed)
+
+```json
+{
+ "streamConfigMaps": [
+ {
+ "streamType": "kafka",
+ "stream.kafka.topic.name": "events",
+ "stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
+ "stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
+ "stream.kafka.broker.list": "localhost:9092"
+ }
+ ]
+}
+```
+
+### 2. A pinned binary format (SQLite JSONB)
+
+Add a single decoder property:
+
+```json
+"stream.kafka.decoder.prop.jsonFormat": "SQLITE_JSONB"
+```
+
+Full block:
+
+```json
+{
+ "streamConfigMaps": [
+ {
+ "streamType": "kafka",
+ "stream.kafka.topic.name": "events",
+ "stream.kafka.decoder.class.name":
"org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder",
+ "stream.kafka.decoder.prop.jsonFormat": "SQLITE_JSONB",
+ "stream.kafka.consumer.factory.class.name":
"org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory",
+ "stream.kafka.broker.list": "localhost:9092"
+ }
+ ]
+}
+```
+
+Swap the value for `POSTGRES_JSONB`, `SMILE`, or `CBOR` to decode those
encodings.
+
+### 3. Auto-detection for a mixed stream
+
+```json
+"stream.kafka.decoder.prop.jsonFormat": "AUTO"
+```
+
+Decodes text JSON, Smile, CBOR (with the self-describe tag), PostgreSQL
`jsonb`, and SQLite JSONB messages on
+the same topic, one classification per message.
+
+## Format notes
+
+### PostgreSQL `jsonb`
+
+Despite `jsonb`'s compact on-disk layout, that layout never leaves the server.
`jsonb_send` renders the value
+as text and emits a version byte (`1`) followed by the UTF-8 JSON text;
`jsonb_recv` reverses it. Every
+standard binary producer — the v3 extended-query protocol, `COPY ... WITH
(FORMAT binary)`, and logical
+replication via `pgoutput` — goes through that same function. This parser
strips the version byte and parses
+the text body, so its value types are identical to `TEXT`.
+
+### SQLite JSONB
+
+Implements the [SQLite JSONB](https://sqlite.org/jsonb.html) format (SQLite
3.45+), including the JSON5 element
+types (hex/`+`-prefixed integers, `Infinity`/`NaN` floats, and `\x`/`\'`/`\v`
string escapes). Per SQLite's
+validity rule, the top-level element must exactly fill the message; a payload
that declares a shorter size is
+rejected rather than decoded into a partial row.
+
+### Smile and CBOR
+
+Decoded with Jackson. Smile carries its 3-byte header by default (used for
`AUTO` detection). CBOR has no
+mandatory magic number, so `AUTO` recognizes it only when the optional `D9 D9
F7` self-describe tag is present;
+otherwise pin `jsonFormat: CBOR`.
+
+## Diagnostics
+
+When a message cannot be decoded, the decoder throws with a bounded, log-safe
description of the payload: a
+leading window rendered as text when it looks like text and as hex otherwise,
prefixed with the total byte
+length. Binary payloads therefore appear as a short hex prefix rather than an
unbounded dump of unreadable
+bytes.
diff --git a/pinot-plugins/pinot-input-format/pinot-json/pom.xml
b/pinot-plugins/pinot-input-format/pinot-json/pom.xml
index 306ae861116..2bdaea24550 100644
--- a/pinot-plugins/pinot-input-format/pinot-json/pom.xml
+++ b/pinot-plugins/pinot-input-format/pinot-json/pom.xml
@@ -35,6 +35,18 @@
<shade.phase.prop>package</shade.phase.prop>
</properties>
+ <dependencies>
+ <!-- Binary JSON encodings supported by the stream decoder. Versions
managed by jackson-bom. -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-smile</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-cbor</artifactId>
+ </dependency>
+ </dependencies>
+
<profiles>
<profile>
<id>pinot-fastdev</id>
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 08a1122e6c4..a83b6cfde05 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
@@ -18,36 +18,61 @@
*/
package org.apache.pinot.plugin.inputformat.json;
+import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Set;
+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;
import org.apache.pinot.spi.data.readers.RecordExtractor;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.stream.StreamMessageDecoder;
-import org.apache.pinot.spi.utils.JsonUtils;
/**
* An implementation of StreamMessageDecoder to read JSON records from a
stream.
+ *
+ * <p>Set the {@value #JSON_FORMAT_CONFIG_KEY} decoder property to pin the
payload encoding to one of
+ * {@code TEXT}, {@code POSTGRES_JSONB}, {@code SQLITE_JSONB}, {@code SMILE}
or {@code CBOR}. When unset the
+ * encoding is {@code TEXT}, the decoder's historical behavior.
+ *
+ * <p>Set it to {@code AUTO} to instead detect the encoding per message from
its leading magic / version bytes,
+ * falling back to text JSON. Detection is allocation-free and cannot
mis-route a well-formed text JSON
+ * document: a top-level <code>{</code> or <code>[</code> (optionally
after whitespace) collides with none
+ * of the binary signatures. It is opt-in rather than the default because it
is still a heuristic over a few
+ * leading bytes, so it may claim a corrupt message that text decoding would
have rejected outright.
+ * See {@link JsonPayloadFormat}.
*/
public class JSONMessageDecoder implements StreamMessageDecoder<byte[]> {
+ public static final String JSON_FORMAT_CONFIG_KEY = "jsonFormat";
+
+ /// Caps on the payload bytes echoed into a decode-failure message. Hex
renders two characters per byte, so
+ /// the binary cap is a quarter of the text one to keep both messages a
comparable length.
+ private static final int MAX_DIAGNOSTIC_TEXT_BYTES = 512;
+ private static final int MAX_DIAGNOSTIC_HEX_BYTES = 128;
+
private static final String JSON_RECORD_EXTRACTOR_CLASS =
"org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor";
private RecordExtractor<Map<String, Object>> _jsonRecordExtractor;
+ // For AUTO this resolves the concrete format per message; otherwise it is
the pinned format's parser.
+ private JsonPayloadParser _parser;
@Override
public void init(Map<String, String> props, Set<String> fieldsToRead, String
topicName)
throws Exception {
String recordExtractorClass = null;
+ String jsonFormat = null;
if (props != null) {
recordExtractorClass = props.get(RECORD_EXTRACTOR_CONFIG_KEY);
+ jsonFormat = props.get(JSON_FORMAT_CONFIG_KEY);
}
if (recordExtractorClass == null) {
recordExtractorClass = JSON_RECORD_EXTRACTOR_CLASS;
}
_jsonRecordExtractor =
PluginManager.get().createInstance(recordExtractorClass);
_jsonRecordExtractor.init(fieldsToRead, null);
+ _parser = JsonPayloadFormat.fromConfig(jsonFormat).getParser();
}
@Override
@@ -58,12 +83,52 @@ public class JSONMessageDecoder implements
StreamMessageDecoder<byte[]> {
@Override
public GenericRow decode(byte[] payload, int offset, int length, GenericRow
destination) {
try {
- // Parse directly to Map, avoiding intermediate JsonNode representation
for better performance
- Map<String, Object> jsonMap = JsonUtils.bytesToMap(payload, offset,
length);
+ // Parse directly to Map, avoiding an intermediate JsonNode
representation for better performance.
+ Map<String, Object> jsonMap = _parser.parse(payload, offset, length);
return _jsonRecordExtractor.extract(jsonMap, destination);
} catch (Exception e) {
throw new RuntimeException(
- "Caught exception while decoding JSON record with payload: " + new
String(payload, offset, length), e);
+ "Caught exception while decoding JSON record with payload: " +
describePayload(payload, offset, length), e);
+ }
+ }
+
+ /// Renders a bounded, log-safe description of a payload for an error
message.
+ ///
+ /// The payload may now be any of the binary encodings, so it is neither
necessarily text nor necessarily
+ /// small. A leading window is rendered as UTF-8 when it looks like text (no
control characters other than
+ /// whitespace) and as hex otherwise, so a binary message cannot flood the
log with megabytes of mojibake,
+ /// while a malformed text record still shows enough of itself to be
identified. The charset is always
+ /// explicit rather than the platform default.
+ private static String describePayload(byte[] payload, int offset, int
length) {
+ boolean text = looksLikeText(payload, offset, Math.min(length,
MAX_DIAGNOSTIC_TEXT_BYTES));
+ int previewLength = Math.min(length, text ? MAX_DIAGNOSTIC_TEXT_BYTES :
MAX_DIAGNOSTIC_HEX_BYTES);
+ StringBuilder description = new StringBuilder(32 + (text ? previewLength :
2 * previewLength));
+ description.append(length).append(" bytes: ");
+ if (text) {
+ // Truncation may split a multi-byte character; UTF-8 decoding
substitutes U+FFFD rather than throwing.
+ description.append(new String(payload, offset, previewLength,
StandardCharsets.UTF_8));
+ } else {
+ description.append("0x");
+ for (int i = 0; i < previewLength; i++) {
+ description.append(Character.forDigit((payload[offset + i] >> 4) &
0xF, 16))
+ .append(Character.forDigit(payload[offset + i] & 0xF, 16));
+ }
+ }
+ if (previewLength < length) {
+ description.append("...(truncated)");
+ }
+ return description.toString();
+ }
+
+ /// Whether the region contains no control characters other than JSON's
insignificant whitespace. Bytes at or
+ /// above 0x80 are accepted so non-ASCII UTF-8 text still renders as text.
+ private static boolean looksLikeText(byte[] payload, int offset, int length)
{
+ for (int i = offset, end = offset + length; i < end; i++) {
+ byte b = payload[i];
+ if (b >= 0 && b < 0x20 && b != '\t' && b != '\n' && b != '\r') {
+ return false;
+ }
}
+ return true;
}
}
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
new file mode 100644
index 00000000000..03813aebb25
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java
@@ -0,0 +1,40 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import java.util.Map;
+
+
+/// Resolves the concrete parser per message from the payload's leading magic
/ version bytes, then delegates.
+/// Backs [JsonPayloadFormat#AUTO], so the decoder can always call `parse`
without a null / mode check.
+class AutoDetectPayloadParser implements JsonPayloadParser {
+
+ /// Always {@code true}: this parser accepts any payload and resolves the
real format at [#parse] time. It is
+ /// deliberately excluded from the detection order it drives.
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ return true;
+ }
+
+ @Override
+ public Map<String, Object> parse(byte[] payload, int offset, int length)
+ throws Exception {
+ return JsonPayloadFormat.detectParser(payload, offset,
length).parse(payload, offset, length);
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java
new file mode 100644
index 00000000000..3ed57b00a90
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java
@@ -0,0 +1,49 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
+
+
+/// Parses <a href="https://www.rfc-editor.org/rfc/rfc8949.html";>CBOR</a> (RFC
8949) via Jackson. Values decode
+/// to the same Java types Jackson produces for text JSON (CBOR may
additionally emit `Float` and `byte[]`
+/// scalars, which Pinot handles downstream).
+///
+/// CBOR has no mandatory magic number, so AUTO detection only recognizes
payloads that carry the optional
+/// self-describe tag `0xD9D9F7` (RFC 8949 §3.4.6). CBOR streams without that
tag must select the format
+/// explicitly.
+class CborJsonPayloadParser extends JacksonPayloadParser {
+
+ // CBOR "Self-Described CBOR" tag 55799, encoded as the 3-byte prefix 0xD9
0xD9 0xF7 (RFC 8949 §3.4.6).
+ private static final int SELF_DESCRIBE_0 = 0xD9;
+ private static final int SELF_DESCRIBE_1 = 0xD9;
+ private static final int SELF_DESCRIBE_2 = 0xF7;
+
+ CborJsonPayloadParser() {
+ super(new CBORFactory());
+ }
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ return length >= 3
+ && (payload[offset] & 0xFF) == SELF_DESCRIBE_0
+ && (payload[offset + 1] & 0xFF) == SELF_DESCRIBE_1
+ && (payload[offset + 2] & 0xFF) == SELF_DESCRIBE_2;
+ }
+}
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
new file mode 100644
index 00000000000..711c55e2bec
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java
@@ -0,0 +1,45 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectReader;
+import java.util.Map;
+import org.apache.pinot.spi.utils.JsonUtils;
+
+
+/// Base for formats that Jackson can read directly given a binary
[JsonFactory] (Smile, CBOR). Subclasses only
+/// supply the factory and their magic-byte [#matches] check.
+///
+/// The [ObjectReader] is immutable and thread-safe, so a single instance is
shared across all decode calls.
+abstract class JacksonPayloadParser implements JsonPayloadParser {
+
+ private final ObjectReader _mapReader;
+
+ JacksonPayloadParser(JsonFactory factory) {
+ _mapReader = new
ObjectMapper(factory).readerFor(JsonUtils.MAP_TYPE_REFERENCE);
+ }
+
+ @Override
+ public Map<String, Object> parse(byte[] payload, int offset, int length)
+ throws Exception {
+ return _mapReader.readValue(payload, offset, length);
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java
new file mode 100644
index 00000000000..9e17c79cba5
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java
@@ -0,0 +1,89 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import java.util.Locale;
+import javax.annotation.Nullable;
+
+
+/// The payload encodings the JSON stream decoder understands, selected
through the `jsonFormat` decoder
+/// property.
+///
+/// An unset property means [#TEXT], preserving the decoder's historical
behavior exactly. [#AUTO] is opt-in:
+/// detection is a heuristic over a few leading bytes, and while it never
mis-routes a well-formed text JSON
+/// document, it can claim a corrupt message that text decoding would have
rejected outright. Turning that on
+/// for every existing stream would be a silent correctness change, so
operators ask for it explicitly.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes,
falling back to text JSON.
+ /// Opt-in; see the note on this enum.
+ AUTO(new AutoDetectPayloadParser()),
+ /// UTF-8 text JSON (the historical default).
+ TEXT(new TextJsonPayloadParser()),
+ /// PostgreSQL `jsonb` binary wire format (version byte + text JSON).
+ POSTGRES_JSONB(new PostgresJsonbPayloadParser()),
+ /// SQLite 3.45+ JSONB binary format.
+ SQLITE_JSONB(new SqliteJsonbPayloadParser()),
+ /// Jackson Smile binary JSON.
+ SMILE(new SmileJsonPayloadParser()),
+ /// CBOR (RFC 8949).
+ CBOR(new CborJsonPayloadParser());
+
+ // AUTO detection precedence: formats with strong, unambiguous magic bytes
first; text (the catch-all,
+ // detected only by a leading '{' / '[') last so it never shadows a binary
format. POSTGRES_JSONB precedes
+ // TEXT because its body *is* text JSON behind a version byte.
+ private static final JsonPayloadFormat[] DETECTION_ORDER =
+ {SMILE, CBOR, POSTGRES_JSONB, SQLITE_JSONB, TEXT};
+
+ private final JsonPayloadParser _parser;
+
+ JsonPayloadFormat(JsonPayloadParser parser) {
+ _parser = parser;
+ }
+
+ /// The parser for this format. For [#AUTO] this is a parser that resolves
the concrete format per message.
+ public JsonPayloadParser getParser() {
+ return _parser;
+ }
+
+ /// Resolves a configured format name (case-insensitive). An unset value
means [#TEXT], the decoder's
+ /// historical behavior; [#AUTO] must be requested explicitly.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return TEXT;
+ }
+ try {
+ return valueOf(value.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Unsupported jsonFormat '" + value + "'. Supported values: AUTO,
TEXT, POSTGRES_JSONB, SQLITE_JSONB, "
+ + "SMILE, CBOR", e);
+ }
+ }
+
+ /// Picks a parser for a payload by consulting each format's magic / version
bytes in [#DETECTION_ORDER],
+ /// defaulting to text JSON when nothing matches.
+ public static JsonPayloadParser detectParser(byte[] payload, int offset, int
length) {
+ for (JsonPayloadFormat format : DETECTION_ORDER) {
+ if (format._parser.matches(payload, offset, length)) {
+ return format._parser;
+ }
+ }
+ return TEXT._parser;
+ }
+}
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
new file mode 100644
index 00000000000..90b353cceb3
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java
@@ -0,0 +1,54 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import java.util.Map;
+
+
+/// Parses a stream payload encoded in a particular (text or binary) JSON
representation into a Jackson-style
+/// `Map<String, Object>` so the shared
+/// [org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor] can turn it
into a
+/// [org.apache.pinot.spi.data.readers.GenericRow].
+///
+/// Implementations must produce the same Java value contract Jackson produces
for text JSON, so the extractor
+/// can treat every format uniformly: `Boolean`, `Integer`, `Long`, `Double`,
`String`, `java.math.BigInteger`
+/// / `java.math.BigDecimal` for oversized / high-precision numbers,
`java.util.List` for arrays, and nested
+/// `Map<String, Object>` for objects. Binary formats may additionally yield
`Float` and `byte[]` scalars,
+/// which Pinot's downstream type conversion handles.
+///
+/// Implementations are stateless and must be thread-safe: a single instance
is shared across all decode calls.
+public interface JsonPayloadParser {
+
+ /// Cheap, allocation-free check of the payload's leading bytes to decide
whether this parser recognizes the
+ /// encoding. Used only by [JsonPayloadFormat#AUTO] detection; when a format
is configured explicitly the
+ /// corresponding parser is used without consulting this method.
+ ///
+ /// @param payload backing byte array
+ /// @param offset start offset of the record within {@code payload}
+ /// @param length number of bytes belonging to the record
+ /// @return {@code true} if the leading bytes match this format's magic /
version signature
+ boolean matches(byte[] payload, int offset, int length);
+
+ /// Parses the {@code [offset, offset + length)} region of {@code payload}
into a mutable
+ /// `Map<String, Object>` following the value contract described on this
interface.
+ ///
+ /// @throws Exception if the region is not valid for this format
+ Map<String, Object> parse(byte[] payload, int offset, int length)
+ throws Exception;
+}
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
new file mode 100644
index 00000000000..2a6a5b4c9af
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java
@@ -0,0 +1,58 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import java.util.Map;
+import org.apache.pinot.spi.utils.JsonUtils;
+
+
+/// Parses the PostgreSQL `jsonb` binary wire format: a single version byte
(currently `1`) followed by the
+/// value's **UTF-8 text** JSON representation.
+///
+/// Despite `jsonb`'s compact on-disk `JsonbContainer` layout, that layout
never leaves the server. PostgreSQL's
+/// `jsonb_send` (`src/backend/utils/adt/jsonb.c`) renders the value with
`JsonbToCString` and emits
+/// `pq_sendint8(version = 1)` followed by the text; `jsonb_recv` reverses it
and rejects any other version.
+/// Every standard binary producer — the v3 extended-query protocol, `COPY ...
WITH (FORMAT binary)`, and
+/// 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 {
+
+ /// The only version `jsonb_recv` accepts.
+ private static final byte JSONB_VERSION = 1;
+
+ @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
+ // claiming arbitrary binary payloads that merely happen to begin with
0x01.
+ return length >= 2 && payload[offset] == JSONB_VERSION &&
TextJsonPayloadParser.startsWithJsonDocument(payload,
+ offset + 1, length - 1);
+ }
+
+ @Override
+ public Map<String, Object> parse(byte[] payload, int offset, int length)
+ throws Exception {
+ 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/SmileJsonPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java
new file mode 100644
index 00000000000..eddfe470a05
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java
@@ -0,0 +1,44 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import com.fasterxml.jackson.dataformat.smile.SmileConstants;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+
+
+/// Parses <a
href="https://github.com/FasterXML/smile-format-specification";>Smile</a>,
Jackson's own binary
+/// JSON encoding. Values decode to the same Java types Jackson produces for
text JSON (Smile may additionally
+/// emit `Float` and `byte[]` scalars, which Pinot handles downstream).
+///
+/// Detection relies on the 3-byte Smile header `:)\n` (`0x3A 0x29 0x0A`) that
Jackson writes by default;
+/// header-less Smile payloads must select the format explicitly rather than
via AUTO detection.
+class SmileJsonPayloadParser extends JacksonPayloadParser {
+
+ SmileJsonPayloadParser() {
+ super(new SmileFactory());
+ }
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ return length >= 3
+ && payload[offset] == SmileConstants.HEADER_BYTE_1
+ && payload[offset + 1] == SmileConstants.HEADER_BYTE_2
+ && payload[offset + 2] == SmileConstants.HEADER_BYTE_3;
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java
new file mode 100644
index 00000000000..46041cc8ee5
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java
@@ -0,0 +1,508 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import com.google.common.collect.Maps;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.CharsetDecoder;
+import java.nio.charset.CodingErrorAction;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+
+/// Parses the <a href="https://sqlite.org/jsonb.html";>SQLite JSONB</a> binary
format (SQLite 3.45+).
+///
+/// Every element is a 1–9 byte header followed by a payload. The header's
first byte packs the element type in
+/// the low nibble and a payload-size descriptor in the high nibble;
descriptors 12–15 mean the size is a big-
+/// endian `uint8` / `uint16` / `uint32` / `uint64` in the following bytes,
and 0–11 are the size itself.
+///
+/// Numbers are stored as their ASCII text, so integers narrow to `Integer` /
`Long` / `BigInteger` and floats
+/// to `Double`, matching the text-JSON value contract. `TEXTJ` strings/labels
are JSON-unescaped (invalid
+/// escapes are rejected); `TEXT5` additionally accepts JSON5 escapes.
+///
+/// The decoder expects one row per payload, so the top-level element must be
an `OBJECT`.
+class SqliteJsonbPayloadParser implements JsonPayloadParser {
+
+ // Element types (low nibble of the header's first byte).
+ private static final int TYPE_NULL = 0;
+ private static final int TYPE_TRUE = 1;
+ private static final int TYPE_FALSE = 2;
+ private static final int TYPE_INT = 3; // canonical decimal integer,
ASCII
+ private static final int TYPE_INT5 = 4; // JSON5 integer (hex / leading
sign), ASCII
+ private static final int TYPE_FLOAT = 5; // canonical float, ASCII
+ private static final int TYPE_FLOAT5 = 6; // JSON5 float, ASCII
+ private static final int TYPE_TEXT = 7; // raw text, no escapes
+ private static final int TYPE_TEXTJ = 8; // text with JSON escapes
+ private static final int TYPE_TEXT5 = 9; // text with JSON5 escapes
+ private static final int TYPE_TEXTRAW = 10; // raw text needing quoting, no
escapes
+ private static final int TYPE_ARRAY = 11;
+ private static final int TYPE_OBJECT = 12;
+
+ private static final BigInteger INT_MIN =
BigInteger.valueOf(Integer.MIN_VALUE);
+ private static final BigInteger INT_MAX =
BigInteger.valueOf(Integer.MAX_VALUE);
+ private static final BigInteger LONG_MIN =
BigInteger.valueOf(Long.MIN_VALUE);
+ private static final BigInteger LONG_MAX =
BigInteger.valueOf(Long.MAX_VALUE);
+
+ // RFC 8259 number grammar for the canonical TYPE_INT / TYPE_FLOAT forms: an
optional minus, an integer part
+ // with no leading zeros, and (for floats) an optional fraction and
exponent. No leading '+', NaN, Infinity or
+ // hex -- the tokens Double.parseDouble / Long.parseLong would otherwise
accept.
+ private static final Pattern CANONICAL_INT =
Pattern.compile("-?(?:0|[1-9][0-9]*)");
+ private static final Pattern CANONICAL_NUMBER =
+ Pattern.compile("-?(?:0|[1-9][0-9]*)(?:\\.[0-9]+)?(?:[eE][+-]?[0-9]+)?");
+
+ // Bound nesting depth and numeric-token length to Jackson's
StreamReadConstraints defaults, so this binary
+ // path rejects the same pathological shapes text JSON already rejects via
JsonUtils.bytesToMap. The depth cap
+ // is the important one: each nesting level is only ~1-3 wire bytes, so
without it a small message can recurse
+ // deep enough to throw StackOverflowError -- an Error, not an Exception,
that escapes the decode-error
+ // handling and fails the whole consuming segment instead of dropping one
bad record.
+ private static final int MAX_NESTING_DEPTH = 1000;
+ private static final int MAX_NUMBER_LENGTH = 1000;
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ // The top-level element must be an OBJECT (low nibble == 12) to produce a
row. That never collides with
+ // text JSON, whose first byte is '{' (0x7B) or '[' (0x5B) -- both low
nibble 0x0B (ARRAY), never 0x0C.
+ if (length < 1 || (payload[offset] & 0x0F) != TYPE_OBJECT) {
+ return false;
+ }
+ // The nibble alone is a weak signal: one byte in sixteen of arbitrary
binary would claim the payload. Also
+ // require the object's declared size to exactly fill the payload -- the
same validity rule parse() enforces
+ // -- so AUTO cannot hand a corrupt message to this parser on the strength
of a single nibble.
+ int sizeDescriptor = (payload[offset] & 0xFF) >>> 4;
+ int headerLength = headerLength(sizeDescriptor);
+ if (length < headerLength) {
+ return false;
+ }
+ return declaredSize(payload, offset, sizeDescriptor) == (long) length -
headerLength;
+ }
+
+ /// Bytes occupied by an element header: the type/size-descriptor byte plus
the width of any explicit size
+ /// field. Descriptors 0-11 encode the size in the byte itself; 12-15
prepend a 1-, 2-, 4- or 8-byte
+ /// big-endian size.
+ ///
+ /// Shared by [#matches] and [#readElement] so detection and parsing can
never disagree about the layout.
+ private static int headerLength(int sizeDescriptor) {
+ switch (sizeDescriptor) {
+ case 12:
+ return 2;
+ case 13:
+ return 3;
+ case 14:
+ return 5;
+ case 15:
+ return 9;
+ default:
+ return 1;
+ }
+ }
+
+ /// Payload size declared by the element header starting at {@code offset}.
The caller must first ensure
+ /// [#headerLength] bytes are available. A descriptor-15 size with its sign
bit set yields a negative value,
+ /// which every caller rejects (it can neither be a valid length nor equal a
payload length).
+ private static long declaredSize(byte[] payload, int offset, int
sizeDescriptor) {
+ switch (sizeDescriptor) {
+ case 12:
+ return readUnsignedBE(payload, offset + 1, 1);
+ case 13:
+ return readUnsignedBE(payload, offset + 1, 2);
+ case 14:
+ return readUnsignedBE(payload, offset + 1, 4);
+ case 15:
+ return readUnsignedBE(payload, offset + 1, 8);
+ default:
+ return sizeDescriptor;
+ }
+ }
+
+ /// Reads {@code count} big-endian bytes as an unsigned value. An 8-byte
field with the sign bit set yields a
+ /// negative long, which never equals a payload length and is rejected by
the bounds checks.
+ private static long readUnsignedBE(byte[] payload, int offset, int count) {
+ long value = 0;
+ for (int i = 0; i < count; i++) {
+ value = (value << 8) | (payload[offset + i] & 0xFFL);
+ }
+ return value;
+ }
+
+ @Override
+ public Map<String, Object> parse(byte[] payload, int offset, int length) {
+ int limit = offset + length;
+ Cursor cursor = new Cursor(payload, offset, limit);
+ Object value = readElement(cursor, limit, 0);
+ if (!(value instanceof Map)) {
+ throw new IllegalArgumentException("Top-level SQLite JSONB element must
be an object");
+ }
+ // SQLite's validity rule: the outer element must exactly fill the BLOB.
Without this, a payload whose
+ // top-level element declares a short size (e.g. a bare 0x0C followed by
data) would decode to a partial --
+ // possibly empty -- row and silently discard the trailing bytes instead
of rejecting a corrupt message.
+ if (cursor._pos != limit) {
+ throw new IllegalArgumentException(
+ "Top-level SQLite JSONB element consumed " + (cursor._pos - offset)
+ " of " + length + " bytes");
+ }
+ //noinspection unchecked
+ return (Map<String, Object>) value;
+ }
+
+ /// Reads one element. {@code parentEnd} is the exclusive end of the
enclosing container (the whole payload at
+ /// the top level); an element may never extend past it, otherwise a nested
element could overrun its parent
+ /// while still fitting the payload and silently swallow its following
siblings.
+ private static Object readElement(Cursor cursor, int parentEnd, int depth) {
+ int header = cursor.peekUInt8();
+ int type = header & 0x0F;
+ int sizeDescriptor = header >>> 4;
+ int headerStart = cursor._pos;
+ // Decode the header through the same helpers matches() uses, so detection
and parsing cannot drift apart.
+ cursor.skip(headerLength(sizeDescriptor));
+ long payloadSize = declaredSize(cursor._buf, headerStart, sizeDescriptor);
+ int start = cursor._pos;
+ int end = cursor.boundedEnd(start, payloadSize, parentEnd);
+ Object result;
+ switch (type) {
+ case TYPE_NULL:
+ result = null;
+ break;
+ case TYPE_TRUE:
+ result = Boolean.TRUE;
+ break;
+ case TYPE_FALSE:
+ result = Boolean.FALSE;
+ break;
+ case TYPE_INT:
+ result = parseCanonicalInt(cursor.ascii(start, end));
+ break;
+ case TYPE_INT5:
+ result = parseInt5(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT:
+ result = parseCanonicalFloat(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT5:
+ result = parseJson5Float(cursor.ascii(start, end));
+ break;
+ case TYPE_TEXT:
+ case TYPE_TEXTRAW:
+ result = cursor.utf8(start, end);
+ break;
+ case TYPE_TEXTJ:
+ result = unescape(cursor.utf8(start, end), false);
+ break;
+ case TYPE_TEXT5:
+ result = unescape(cursor.utf8(start, end), true);
+ break;
+ case TYPE_ARRAY:
+ return readArray(cursor, end, depth);
+ case TYPE_OBJECT:
+ return readObject(cursor, end, depth);
+ default:
+ throw new IllegalArgumentException("Reserved/invalid SQLite JSONB
element type: " + type);
+ }
+ cursor._pos = end;
+ return result;
+ }
+
+ private static List<Object> readArray(Cursor cursor, int end, int depth) {
+ int childDepth = nextDepth(depth);
+ List<Object> list = new ArrayList<>();
+ while (cursor._pos < end) {
+ list.add(readElement(cursor, end, childDepth));
+ }
+ return list;
+ }
+
+ private static Map<String, Object> readObject(Cursor cursor, int end, int
depth) {
+ int childDepth = nextDepth(depth);
+ Map<String, Object> map = Maps.newHashMapWithExpectedSize(4);
+ while (cursor._pos < end) {
+ Object label = readElement(cursor, end, childDepth);
+ if (!(label instanceof String)) {
+ throw new IllegalArgumentException("SQLite JSONB object label must be
text");
+ }
+ if (cursor._pos >= end) {
+ throw new IllegalArgumentException("SQLite JSONB object is missing a
value for label: " + label);
+ }
+ map.put((String) label, readElement(cursor, end, childDepth));
+ }
+ return map;
+ }
+
+ /// Returns {@code depth + 1} for the children of a container, throwing once
nesting would exceed
+ /// [#MAX_NESTING_DEPTH]. Keeps deep nesting on the
`IllegalArgumentException` path (handled as a bad record)
+ /// rather than letting the recursion overflow the stack with a
`StackOverflowError`.
+ private static int nextDepth(int depth) {
+ if (depth >= MAX_NESTING_DEPTH) {
+ throw new IllegalArgumentException("SQLite JSONB nesting exceeds the
maximum depth of " + MAX_NESTING_DEPTH);
+ }
+ return depth + 1;
+ }
+
+ /// Parses a canonical (RFC 8259) integer, the SQLite `TYPE_INT` form.
Validated as strictly as text JSON:
+ /// the permissive tokens `Long.parseLong` / `BigInteger` accept -- a
leading `+` or leading zeros -- are
+ /// rejected, so a malformed `TYPE_INT` fails the record rather than
decoding to a value Jackson would refuse.
+ /// The JSON5 forms (hex, leading sign) live in [#parseInt5].
+ private static Object parseCanonicalInt(String text) {
+ if (!CANONICAL_INT.matcher(text).matches()) {
+ throw new IllegalArgumentException("Invalid canonical SQLite JSONB
integer: " + text);
+ }
+ return parseInt(text, 10);
+ }
+
+ private static Object parseInt(String text, int radix) {
+ // Fast path: values that fit a long (the overwhelming majority) avoid
BigInteger allocation.
+ try {
+ return narrowLong(Long.parseLong(text, radix));
+ } catch (NumberFormatException e) {
+ return narrow(new BigInteger(text, radix));
+ }
+ }
+
+ /// Narrows an integer to the smallest of `Integer` / `Long` / `BigInteger`
that holds it, mirroring the
+ /// Jackson text-JSON contract (`BigInteger` is later widened to
`BigDecimal` by the record extractor).
+ private static Object narrow(BigInteger value) {
+ if (value.compareTo(INT_MIN) >= 0 && value.compareTo(INT_MAX) <= 0) {
+ return value.intValue();
+ }
+ if (value.compareTo(LONG_MIN) >= 0 && value.compareTo(LONG_MAX) <= 0) {
+ return value.longValue();
+ }
+ return value;
+ }
+
+ private static Object parseInt5(String text) {
+ String digits = text;
+ boolean negative = false;
+ if (!digits.isEmpty() && (digits.charAt(0) == '+' || digits.charAt(0) ==
'-')) {
+ negative = digits.charAt(0) == '-';
+ digits = digits.substring(1);
+ }
+ Object magnitude;
+ if (digits.length() > 2 && digits.charAt(0) == '0' && (digits.charAt(1) ==
'x' || digits.charAt(1) == 'X')) {
+ magnitude = parseInt(digits.substring(2), 16);
+ } else {
+ magnitude = parseInt(digits, 10);
+ }
+ if (!negative) {
+ return magnitude;
+ }
+ return magnitude instanceof BigInteger ? narrow(((BigInteger)
magnitude).negate())
+ : narrowLong(-((Number) magnitude).longValue());
+ }
+
+ private static Object narrowLong(long value) {
+ // NOTE: keep these as separate returns. A `cond ? (int) value : value`
ternary would numerically promote
+ // the int branch to long, always boxing to Long.
+ if (value >= Integer.MIN_VALUE && value <= Integer.MAX_VALUE) {
+ return (int) value;
+ }
+ return value;
+ }
+
+ /// Parses a canonical (RFC 8259) float, the SQLite `TYPE_FLOAT` form.
`Double.parseDouble` is far more
+ /// permissive than JSON -- it accepts `NaN`, `Infinity`, a leading `+`,
Java hex floats and type suffixes --
+ /// so the text is first validated against the JSON number grammar. That
rejects the non-finite and other
+ /// non-canonical tokens exactly as Jackson would, rather than ingesting
them. The JSON5 forms live in
+ /// [#parseJson5Float].
+ private static Double parseCanonicalFloat(String text) {
+ if (!CANONICAL_NUMBER.matcher(text).matches()) {
+ throw new IllegalArgumentException("Invalid canonical SQLite JSONB
float: " + text);
+ }
+ return Double.parseDouble(text);
+ }
+
+ private static Double parseJson5Float(String text) {
+ // JSON5 permits a leading '+' and the tokens Infinity / NaN, all of which
Double.parseDouble accepts once
+ // the '+' is stripped.
+ String normalized = !text.isEmpty() && text.charAt(0) == '+' ?
text.substring(1) : text;
+ return Double.parseDouble(normalized);
+ }
+
+ /// Single-pass JSON string unescape. Returns the input unchanged (no
allocation) when it contains no
+ /// backslash. Rejects invalid escapes for `TEXTJ`; when {@code json5} is
set, also accepts the JSON5 escape
+ /// extensions (`\'`, `\v`, `\0`, `\xHH`, line continuations, and `\<char>`
passthrough).
+ private static String unescape(String content, boolean json5) {
+ int firstEscape = content.indexOf('\\');
+ if (firstEscape < 0) {
+ return content;
+ }
+ int n = content.length();
+ StringBuilder sb = new StringBuilder(n);
+ sb.append(content, 0, firstEscape);
+ int i = firstEscape;
+ while (i < n) {
+ char c = content.charAt(i++);
+ if (c != '\\') {
+ sb.append(c);
+ continue;
+ }
+ if (i >= n) {
+ throw new IllegalArgumentException("Dangling escape in SQLite JSONB
text");
+ }
+ char e = content.charAt(i++);
+ switch (e) {
+ case '"':
+ sb.append('"');
+ break;
+ case '\\':
+ sb.append('\\');
+ break;
+ case '/':
+ sb.append('/');
+ break;
+ case 'b':
+ sb.append('\b');
+ break;
+ case 'f':
+ sb.append('\f');
+ break;
+ case 'n':
+ sb.append('\n');
+ break;
+ case 'r':
+ sb.append('\r');
+ break;
+ case 't':
+ sb.append('\t');
+ break;
+ case 'u':
+ sb.append((char) parseHex(content, i, 4));
+ i += 4;
+ break;
+ default:
+ i = appendJson5Escape(sb, content, e, i, json5);
+ break;
+ }
+ }
+ return sb.toString();
+ }
+
+ private static int appendJson5Escape(StringBuilder sb, String content, char
e, int i, boolean json5) {
+ if (!json5) {
+ throw new IllegalArgumentException("Invalid JSON escape '\\" + e + "' in
SQLite JSONB text");
+ }
+ switch (e) {
+ case '\'':
+ sb.append('\'');
+ return i;
+ case 'v':
+ sb.append('\u000B');
+ return i;
+ case '0':
+ sb.append('\0');
+ return i;
+ case 'x':
+ sb.append((char) parseHex(content, i, 2));
+ return i + 2;
+ case '\n':
+ return i; // line continuation
+ case '\r':
+ return i < content.length() && content.charAt(i) == '\n' ? i + 1 : i;
// CRLF continuation
+ default:
+ sb.append(e); // \<char> -> char
+ return i;
+ }
+ }
+
+ private static int parseHex(String content, int start, int len) {
+ if (start + len > content.length()) {
+ throw new IllegalArgumentException("Truncated \\u / \\x escape in SQLite
JSONB text");
+ }
+ int value = 0;
+ for (int i = start; i < start + len; i++) {
+ int digit = Character.digit(content.charAt(i), 16);
+ if (digit < 0) {
+ throw new IllegalArgumentException("Invalid hex escape in SQLite JSONB
text");
+ }
+ value = (value << 4) | digit;
+ }
+ return value;
+ }
+
+ /// Sequential reader over a bounded region of a byte array. Multi-byte size
fields are big-endian per the
+ /// SQLite JSONB spec.
+ private static final class Cursor {
+ private final byte[] _buf;
+ private final int _limit;
+ private int _pos;
+
+ private Cursor(byte[] buf, int start, int limit) {
+ _buf = buf;
+ _pos = start;
+ _limit = limit;
+ }
+
+ /// The next byte, without advancing.
+ private int peekUInt8() {
+ if (_pos >= _limit) {
+ throw new IllegalArgumentException("Truncated SQLite JSONB payload");
+ }
+ return _buf[_pos] & 0xFF;
+ }
+
+ /// Advances past {@code count} bytes, verifying they are present.
Compares without adding to {@code _pos}
+ /// so the check cannot overflow.
+ private void skip(int count) {
+ if (count > _limit - _pos) {
+ throw new IllegalArgumentException("Truncated SQLite JSONB payload");
+ }
+ _pos += count;
+ }
+
+ /// Validates that a payload of {@code size} bytes starting at {@code
start} fits inside the enclosing
+ /// container (never merely inside the whole payload) and returns its
exclusive end. Compares without adding
+ /// to {@code start} so an adversarial `uint64` size cannot overflow past
the guard. A {@code start} already
+ /// past {@code parentEnd} — possible when a trailing element's header
straddles the boundary — makes
+ /// {@code parentEnd - start} negative and is therefore rejected too.
+ private int boundedEnd(int start, long size, int parentEnd) {
+ if (size < 0 || size > parentEnd - start) {
+ throw new IllegalArgumentException("SQLite JSONB element size exceeds
its enclosing container");
+ }
+ return start + (int) size;
+ }
+
+ /// Decodes a numeric token as ASCII. Numeric element types are this
method's only callers, so it also
+ /// enforces the numeric length cap here: `BigInteger(String)` parsing is
~O(n^2), and without a bound a
+ /// single ~1 MB run of digits would be a per-message CPU sink with no
analogue in text JSON, which caps
+ /// number length the same way via Jackson's StreamReadConstraints.
+ private String ascii(int start, int end) {
+ if (end - start > MAX_NUMBER_LENGTH) {
+ throw new IllegalArgumentException(
+ "SQLite JSONB numeric token exceeds the maximum length of " +
MAX_NUMBER_LENGTH);
+ }
+ return new String(_buf, start, end - start, StandardCharsets.US_ASCII);
+ }
+
+ /// Decodes text strictly as UTF-8. Unlike `new String(..., UTF_8)`, which
substitutes U+FFFD for malformed
+ /// bytes, a `CodingErrorAction.REPORT` decoder throws, so a corrupt JSONB
record fails rather than ingesting
+ /// mutated field names or values -- matching how text JSON / Jackson
reject invalid UTF-8.
+ private String utf8(int start, int end) {
+ CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder()
+ .onMalformedInput(CodingErrorAction.REPORT)
+ .onUnmappableCharacter(CodingErrorAction.REPORT);
+ try {
+ return decoder.decode(ByteBuffer.wrap(_buf, start, end -
start)).toString();
+ } catch (CharacterCodingException e) {
+ throw new IllegalArgumentException("Invalid UTF-8 in SQLite JSONB
text", e);
+ }
+ }
+ }
+}
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
new file mode 100644
index 00000000000..397fea32094
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java
@@ -0,0 +1,55 @@
+/**
+ * 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.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 {
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ return startsWithJsonDocument(payload, offset, length);
+ }
+
+ /// Whether the region's first non-whitespace byte opens a JSON document. A
decoder can only turn a top-level
+ /// object into a row, but `[` is tolerated so detection does not misfire on
top-level arrays.
+ ///
+ /// Shared with [PostgresJsonbPayloadParser], whose wire format is this same
text JSON behind a version byte.
+ static boolean startsWithJsonDocument(byte[] payload, int offset, int
length) {
+ for (int i = offset, end = offset + length; i < end; i++) {
+ byte b = payload[i];
+ // JSON insignificant whitespace: space, tab, LF, CR.
+ if (b == ' ' || b == '\t' || b == '\n' || b == '\r') {
+ continue;
+ }
+ return b == '{' || b == '[';
+ }
+ 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
new file mode 100644
index 00000000000..cbcfe94402b
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java
@@ -0,0 +1,263 @@
+/**
+ * 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.plugin.inputformat.json;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
+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.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// End-to-end coverage of {@link JSONMessageDecoder} decoding each configured
/ auto-detected payload format
+/// through the shared {@link JSONRecordExtractor} into a {@link GenericRow}.
+public class JSONMessageDecoderBinaryTest {
+
+ private static final Set<String> RICH_FIELDS = Set.of("name", "count",
"ratio");
+ private static final Set<String> SINGLE_FIELD = Set.of("a");
+
+ /// AUTO is opt-in, so detection tests must ask for it explicitly.
+ private static final Map<String, String> AUTO =
Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "AUTO");
+
+ private static byte[] smile(Map<String, Object> value)
+ throws Exception {
+ return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value);
+ }
+
+ private static byte[] cbor(Map<String, Object> value)
+ throws Exception {
+ CBORFactory factory = new CBORFactory();
+ factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER);
+ return new ObjectMapper(factory).writeValueAsBytes(value);
+ }
+
+ private static byte[] bytes(int... values) {
+ byte[] result = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (byte) values[i];
+ }
+ return result;
+ }
+
+ // Explicit UTF-8 so fixtures do not depend on the platform default charset.
+ private static final byte[] TEXT_DOC =
+
"{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes(StandardCharsets.UTF_8);
+
+ // Hand-built {"a": 1} fixtures matching the parser unit tests.
+ // SQLite: object(size 4){ text("a"), int("1") }.
+ private static final byte[] SQLITE_A1 = bytes(0x4C, 0x17, 0x61, 0x13, 0x31);
+ // PostgreSQL jsonb_send framing: version byte 1 followed by the text JSON
body.
+ private static final byte[] POSTGRES_A1 =
+ bytes(0x01, '{', '"', 'a', '"', ':', '1', '}');
+
+ private GenericRow decode(Map<String, String> props, Set<String> fields,
byte[] payload)
+ throws Exception {
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ decoder.init(props, fields, "topic");
+ return decoder.decode(payload, new GenericRow());
+ }
+
+ private void assertRich(GenericRow row) {
+ assertEquals(row.getValue("name"), "pinot");
+ assertEquals(row.getValue("count"), 7);
+ assertEquals(row.getValue("ratio"), 2.5);
+ }
+
+ private Map<String, Object> richDoc() {
+ return Map.of("name", "pinot", "count", 7, "ratio", 2.5);
+ }
+
+ /// An unset jsonFormat means TEXT -- the decoder's historical behavior,
with no detection at all.
+ @Test
+ public void testUnsetFormatIsText()
+ throws Exception {
+ assertRich(decode(Map.of(), RICH_FIELDS, TEXT_DOC));
+ }
+
+ /// 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
+ public void testUnsetFormatDoesNotAutoDetectBinary()
+ throws Exception {
+ assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD,
SQLITE_A1));
+ assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD,
POSTGRES_A1));
+ assertThrows(RuntimeException.class, () -> decode(Map.of(), RICH_FIELDS,
smile(richDoc())));
+ assertThrows(RuntimeException.class, () -> decode(Map.of(), RICH_FIELDS,
cbor(richDoc())));
+ // A lone 0x0C is a validly-exact empty SQLite object, but must not become
an empty row on a text stream.
+ assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD,
bytes(0x0C)));
+ }
+
+ @Test
+ public void testConfiguredSmile()
+ throws Exception {
+ assertRich(decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SMILE"), RICH_FIELDS, smile(richDoc())));
+ }
+
+ @Test
+ public void testConfiguredCbor()
+ throws Exception {
+ assertRich(decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"CBOR"), RICH_FIELDS, cbor(richDoc())));
+ }
+
+ @Test
+ public void testConfiguredSqlite()
+ throws Exception {
+ GenericRow row = decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SQLITE_JSONB"), SINGLE_FIELD,
+ SQLITE_A1);
+ assertEquals(row.getValue("a"), 1);
+ }
+
+ @Test
+ public void testConfiguredPostgres()
+ throws Exception {
+ GenericRow row = decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"POSTGRES_JSONB"), SINGLE_FIELD,
+ POSTGRES_A1);
+ assertEquals(row.getValue("a"), 1);
+ }
+
+ @Test
+ public void testAutoDetectsSmile()
+ throws Exception {
+ assertRich(decode(AUTO, RICH_FIELDS, smile(richDoc())));
+ }
+
+ @Test
+ public void testAutoDetectsCbor()
+ throws Exception {
+ assertRich(decode(AUTO, RICH_FIELDS, cbor(richDoc())));
+ }
+
+ @Test
+ public void testAutoDetectsSqlite()
+ throws Exception {
+ assertEquals(decode(AUTO, SINGLE_FIELD, SQLITE_A1).getValue("a"), 1);
+ }
+
+ @Test
+ public void testAutoDetectsPostgres()
+ throws Exception {
+ assertEquals(decode(AUTO, SINGLE_FIELD, POSTGRES_A1).getValue("a"), 1);
+ }
+
+ @Test
+ public void testAutoStillDecodesText()
+ throws Exception {
+ assertRich(decode(AUTO, RICH_FIELDS, TEXT_DOC));
+ }
+
+ @Test
+ public void testMalformedPayloadIsWrappedInRuntimeException() {
+ // A stream decoder routinely meets corrupt messages; decode() must
surface them as a RuntimeException
+ // rather than leaking the parser's own exception type.
+ byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF);
+ assertThrows(RuntimeException.class,
+ () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SMILE"), RICH_FIELDS, badSmile));
+ // Same via AUTO, which detects Smile from the header and then fails to
parse the body.
+ assertThrows(RuntimeException.class, () -> decode(AUTO, RICH_FIELDS,
badSmile));
+ // Truncated SQLite object.
+ assertThrows(RuntimeException.class,
+ () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SQLITE_JSONB"), SINGLE_FIELD,
+ bytes(0x5C, 0x17)));
+ }
+
+ @Test
+ public void
testSqliteTrailingBytesAreRejectedRatherThanIngestedAsAPartialRow() {
+ // A SQLite JSONB payload whose top-level element declares a short size
(here an empty object) used to
+ // decode to an empty row, silently discarding the trailing "a": 1.
Detection no longer claims it (the
+ // declared size does not fill the payload) and the parser rejects it
outright; either way the message
+ // must fail rather than yield a partial row.
+ byte[] shortObjectThenData = bytes(0x0C, 0x17, 0x61, 0x13, 0x31);
+ assertThrows(RuntimeException.class, () -> decode(AUTO, SINGLE_FIELD,
shortObjectThenData));
+ assertThrows(RuntimeException.class,
+ () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SQLITE_JSONB"), SINGLE_FIELD,
+ shortObjectThenData));
+ }
+
+ @Test
+ public void testUnsupportedFormatIsRejectedAtInit() {
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ assertThrows(IllegalArgumentException.class,
+ () -> decoder.init(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"bson"), SINGLE_FIELD, "topic"));
+ }
+
+ @Test
+ public void testDecodeFailureMessageIsBoundedAndCharsetSafe()
+ throws Exception {
+ // A malformed *text* payload keeps its readable rendering ...
+ String message = decodeFailureMessage(Map.of(),
"{\"name\":".getBytes(StandardCharsets.UTF_8));
+ assertTrue(message.contains("8 bytes: {\"name\":"), message);
+
+ // ... including non-ASCII UTF-8, since bytes >= 0x80 are not control
characters ...
+ message = decodeFailureMessage(Map.of(),
"{\"n\":\"café\"".getBytes(StandardCharsets.UTF_8));
+ assertTrue(message.contains("{\"n\":\"café\""), message);
+
+ // ... while a malformed *binary* payload is hex-encoded rather than
dumped as mojibake.
+ byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF);
+ message =
decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SMILE"), badSmile);
+ assertTrue(message.contains("7 bytes: 0x3a290a04ffffff"), message);
+
+ // ... an oversized text payload is truncated to the 512-byte window
instead of echoed in full ...
+ byte[] hugeText = new byte[8192];
+ Arrays.fill(hugeText, (byte) 'x');
+ message =
decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "TEXT"),
hugeText);
+ assertTrue(message.contains("8192 bytes: "), message);
+ assertTrue(message.contains("...(truncated)"), message);
+ assertEquals(preview(message), "x".repeat(512));
+
+ // ... and an oversized binary payload is capped at 128 hex-rendered bytes
(256 characters).
+ byte[] hugeBinary = new byte[8192];
+ Arrays.fill(hugeBinary, (byte) 0x01);
+ hugeBinary[0] = 0x3A;
+ hugeBinary[1] = 0x29;
+ hugeBinary[2] = 0x0A;
+ message =
decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY,
"SMILE"), hugeBinary);
+ assertTrue(message.contains("8192 bytes: 0x3a290a01"), message);
+ assertTrue(message.contains("...(truncated)"), message);
+ // "0x" prefix + 128 rendered bytes at two hex characters each.
+ assertEquals(preview(message).length(), 2 + 2 * 128, message);
+ }
+
+ /// The rendered payload window of a decode-failure message: what sits
between "<n> bytes: " and the
+ /// truncation marker. Isolates the assertion from the message's surrounding
prose.
+ private static String preview(String message) {
+ int start = message.indexOf("bytes: ") + "bytes: ".length();
+ int end = message.indexOf("...(truncated)");
+ return message.substring(start, end < 0 ? message.length() : end);
+ }
+
+ /// Decodes an intentionally malformed payload and returns the resulting
exception message.
+ private String decodeFailureMessage(Map<String, String> props, byte[]
payload)
+ throws Exception {
+ try {
+ decode(props, SINGLE_FIELD, payload);
+ throw new AssertionError("expected decode to fail");
+ } catch (RuntimeException e) {
+ return e.getMessage();
+ }
+ }
+}
diff --git
a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java
new file mode 100644
index 00000000000..fc3685ea5dc
--- /dev/null
+++
b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java
@@ -0,0 +1,670 @@
+/**
+ * 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.plugin.inputformat.json.format;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
+import com.fasterxml.jackson.dataformat.cbor.CBORGenerator;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+public class JsonPayloadFormatTest {
+
+ private static byte[] bytes(int... values) {
+ byte[] result = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (byte) values[i];
+ }
+ return result;
+ }
+
+ /// Explicit UTF-8 so fixtures do not depend on the platform default charset.
+ private static byte[] utf8(String value) {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+
+ /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text
JSON body.
+ private static byte[] postgres(String json) {
+ byte[] body = utf8(json);
+ byte[] result = new byte[body.length + 1];
+ result[0] = 1;
+ System.arraycopy(body, 0, result, 1, body.length);
+ return result;
+ }
+
+ private static byte[] smile(Object value)
+ throws Exception {
+ return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value);
+ }
+
+ private static byte[] cborSelfDescribed(Object value)
+ throws Exception {
+ CBORFactory factory = new CBORFactory();
+ factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER);
+ return new ObjectMapper(factory).writeValueAsBytes(value);
+ }
+
+ private static Map<String, Object> parse(JsonPayloadParser parser, byte[]
payload)
+ throws Exception {
+ return parser.parse(payload, 0, payload.length);
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // Config parsing
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testFromConfig() {
+ // An unset property means TEXT -- the decoder's historical behavior. AUTO
is opt-in, never implicit,
+ // because detection can claim a corrupt message that text decoding would
have rejected.
+ assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.TEXT);
+ assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.TEXT);
+ assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.TEXT);
+
+ assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT);
+ assertEquals(JsonPayloadFormat.fromConfig("auto"), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(" Smile "),
JsonPayloadFormat.SMILE);
+ assertEquals(JsonPayloadFormat.fromConfig("POSTGRES_JSONB"),
JsonPayloadFormat.POSTGRES_JSONB);
+ // AUTO resolves the concrete format per message, so every format has a
non-null parser.
+ assertSame(JsonPayloadFormat.AUTO.getParser().getClass(),
AutoDetectPayloadParser.class);
+ }
+
+ @Test
+ public void testFromConfigRejectsUnknownFormatAndKeepsTheCause() {
+ IllegalArgumentException e =
+ expectThrows(IllegalArgumentException.class, () ->
JsonPayloadFormat.fromConfig("bson"));
+ // The message names the offending value and the supported set ...
+ assertTrue(e.getMessage().contains("bson"), e.getMessage());
+ assertTrue(e.getMessage().contains("SQLITE_JSONB"), e.getMessage());
+ // ... and the underlying valueOf failure is preserved so the stack trace
still points at the parse.
+ assertNotNull(e.getCause());
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // Text
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testText()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser();
+ Map<String, Object> map = parse(parser, utf8(" {\"a\": 1, \"b\":
\"x\"}"));
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertTrue(parser.matches(utf8("{\"a\":1}"), 0, 7));
+ assertTrue(parser.matches(utf8(" [1,2]"), 0, 7));
+ assertFalse(parser.matches(bytes(0xFF), 0, 1));
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // Smile / CBOR (round-tripped through Jackson)
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testSmile()
+ throws Exception {
+ Map<String, Object> original = Map.of("name", "pinot", "count", 7,
"ratio", 2.5, "ok", true, "tags",
+ List.of("a", "b"));
+ byte[] encoded = smile(original);
+
+ JsonPayloadParser parser = JsonPayloadFormat.SMILE.getParser();
+ assertTrue(parser.matches(encoded, 0, encoded.length));
+ assertEquals(parse(parser, encoded), original);
+ }
+
+ @Test
+ public void testCbor()
+ throws Exception {
+ Map<String, Object> original = Map.of("name", "pinot", "count", 7,
"ratio", 2.5, "ok", true, "nested",
+ Map.of("x", 1));
+ byte[] encoded = cborSelfDescribed(original);
+
+ JsonPayloadParser parser = JsonPayloadFormat.CBOR.getParser();
+ assertTrue(parser.matches(encoded, 0, encoded.length));
+ assertEquals(parse(parser, encoded), original);
+ }
+
+ @Test
+ public void testCborWithoutSelfDescribeTagIsNotAutoDetected()
+ throws Exception {
+ byte[] noTag = new ObjectMapper(new
CBORFactory()).writeValueAsBytes(Map.of("name", "pinot"));
+ JsonPayloadParser parser = JsonPayloadFormat.CBOR.getParser();
+ // Not detectable without the tag ...
+ assertFalse(parser.matches(noTag, 0, noTag.length));
+ assertSame(JsonPayloadFormat.detectParser(noTag, 0,
noTag.length).getClass(), TextJsonPayloadParser.class);
+ // ... but explicitly configuring CBOR still decodes it.
+ assertEquals(parse(parser, noTag).get("name"), "pinot");
+ }
+
+ @Test
+ public void testBinaryFormatsPreserveFloatAndBytes()
+ throws Exception {
+ // Float and byte[] are scalars the binary formats can carry but text JSON
cannot. The parser contract
+ // documents that they pass through; guard against a regression that
silently upcasts Float -> Double.
+ Map<String, Object> original = new HashMap<>();
+ original.put("f", 1.5f);
+ original.put("bin", new byte[]{1, 2, 3});
+
+ for (byte[] encoded : List.of(smile(original),
cborSelfDescribed(original))) {
+ Map<String, Object> decoded =
parse(JsonPayloadFormat.detectParser(encoded, 0, encoded.length), encoded);
+ assertTrue(decoded.get("f") instanceof Float, "expected Float, got " +
decoded.get("f").getClass());
+ assertEquals(decoded.get("f"), 1.5f);
+ assertEquals((byte[]) decoded.get("bin"), new byte[]{1, 2, 3});
+ }
+ }
+
+ @Test
+ public void testJacksonFormatsRejectMalformedPayloads() {
+ // Valid Smile header, garbage body.
+ byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF);
+ assertThrows(Exception.class, () ->
parse(JsonPayloadFormat.SMILE.getParser(), badSmile));
+ // Self-describe tag then a truncated map.
+ byte[] badCbor = bytes(0xD9, 0xD9, 0xF7, 0xBF, 0x63);
+ assertThrows(Exception.class, () ->
parse(JsonPayloadFormat.CBOR.getParser(), badCbor));
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // PostgreSQL JSONB wire format: version byte + text JSON (jsonb_send /
jsonb_recv)
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testPostgresWireFormat()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser();
+ byte[] payload =
postgres("{\"a\":1,\"b\":\"x\",\"c\":1.5,\"d\":true,\"e\":null,\"f\":[1,2]}");
+ assertTrue(parser.matches(payload, 0, payload.length));
+
+ Map<String, Object> map = parse(parser, payload);
+ // Identical type contract to TEXT, since the body is plain text JSON.
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertEquals(map.get("c"), 1.5);
+ assertEquals(map.get("d"), true);
+ assertNull(map.get("e"));
+ assertEquals(map.get("f"), List.of(1, 2));
+ }
+
+ @Test
+ public void testPostgresBigIntegerNarrowing()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser();
+ Map<String, Object> map = parse(parser,
postgres("{\"n\":9999999999,\"h\":100000000000000000000}"));
+ assertEquals(map.get("n"), 9999999999L);
+ assertEquals(map.get("h"), new BigInteger("100000000000000000000"));
+ }
+
+ @Test
+ public void testPostgresErrors() {
+ JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser();
+ // Unsupported version byte (jsonb_recv rejects anything but 1).
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x02, '{', '}')));
+ // Version byte only.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x01)));
+ // Version byte followed by a non-JSON body is not claimed by detection.
+ assertFalse(parser.matches(bytes(0x01, 0xFF, 0xFF), 0, 3));
+ // The raw on-disk JsonbContainer layout is NOT the wire format and must
not be silently accepted.
+ assertFalse(parser.matches(bytes(0x01, 0x01, 0x00, 0x00, 0x20), 0, 5));
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // SQLite JSONB (hand-computed fixtures per https://sqlite.org/jsonb.html)
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testSqliteScalars()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+
+ // {"a": 1} -> object(size 4){ text("a"), int("1") }
+ assertEquals(parse(parser, bytes(0x4C, 0x17, 0x61, 0x13, 0x31)),
Map.of("a", 1));
+ // {"b": true} -> object(size 3){ text("b"), true(size 0) }
+ assertEquals(parse(parser, bytes(0x3C, 0x17, 0x62, 0x01)), Map.of("b",
true));
+ // {"f2": false} -> object(size 4){ text("f2"), false(size 0) }
+ assertEquals(parse(parser, bytes(0x4C, 0x27, 0x66, 0x32, 0x02)),
Map.of("f2", false));
+ // {"s": "hi"} -> object(size 5){ text("s"), text("hi") }
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x73, 0x27, 0x68, 0x69)),
Map.of("s", "hi"));
+ // {"r": "hi"} -> object(size 5){ text("r"), textraw("hi") }
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x72, 0x2A, 0x68, 0x69)),
Map.of("r", "hi"));
+ // {"f": 1.5} -> object(size 6){ text("f"), float("1.5") }
+ assertEquals(parse(parser, bytes(0x6C, 0x17, 0x66, 0x35, 0x31, 0x2E,
0x35)), Map.of("f", 1.5));
+ }
+
+ @Test
+ public void testSqliteJson5AndEscapes()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+
+ // {"h": 0x1F} -> object{ text("h"), int5("0x1F") } -> 31
+ assertEquals(parse(parser, bytes(0x7C, 0x17, 0x68, 0x44, 0x30, 0x78, 0x31,
0x46)), Map.of("h", 31));
+ // {"i": Infinity} -> object{ text("i"), float5("Infinity") }
+ assertEquals(parse(parser, bytes(0xBC, 0x17, 0x69, 0x86, 0x49, 0x6E, 0x66,
0x69, 0x6E, 0x69, 0x74, 0x79)),
+ Map.of("i", Double.POSITIVE_INFINITY));
+ // {"t": "a\nb"} -> object{ text("t"), textj("a\\nb") } -> newline
unescaped
+ assertEquals(parse(parser, bytes(0x7C, 0x17, 0x74, 0x48, 0x61, 0x5C, 0x6E,
0x62)), Map.of("t", "a\nb"));
+ // {"t": "\x41"} -> object{ text("t"), text5("\\x41") } -> "A" (JSON5 hex
escape)
+ assertEquals(parse(parser, bytes(0x7C, 0x17, 0x74, 0x49, 0x5C, 0x78, 0x34,
0x31)), Map.of("t", "A"));
+ }
+
+ @Test
+ public void testSqliteTextjRejectsInvalidEscape() {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // TEXTJ carries standard JSON escapes only; "\q" is invalid and must not
be silently passed through.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x6C, 0x17, 0x74, 0x38, 0x5C, 0x71)));
+ }
+
+ @Test
+ public void testSqliteNested()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+
+ // {"arr": [1, 2]} -> object(size 9){ text("arr"), array(size 4){
int("1"), int("2") } }
+ assertEquals(parse(parser, bytes(0x9C, 0x37, 0x61, 0x72, 0x72, 0x4B, 0x13,
0x31, 0x13, 0x32)),
+ Map.of("arr", List.of(1, 2)));
+
+ // {"o": {"x": null}} -> object(size 6){ text("o"), object(size 3){
text("x"), null } }
+ Map<String, Object> nested = parse(parser, bytes(0x6C, 0x17, 0x6F, 0x3C,
0x17, 0x78, 0x00));
+ //noinspection unchecked
+ Map<String, Object> inner = (Map<String, Object>) nested.get("o");
+ assertTrue(inner.containsKey("x"));
+ assertNull(inner.get("x"));
+ }
+
+ @Test
+ public void testSqliteIntNarrowing()
+ throws Exception {
+ assertEquals(sqliteSingleInt("42"), 42);
// fits int
+ assertEquals(sqliteSingleInt("9999999999"), 9999999999L);
// > int
+ assertEquals(sqliteSingleInt("99999999999999999999"), new
BigInteger("99999999999999999999")); // > long
+ }
+
+ /// Builds `{"n": <digits>}` as SQLite JSONB and returns the decoded value
of `n`.
+ private static Object sqliteSingleInt(String digits)
+ throws Exception {
+ byte[] digitBytes = digits.getBytes(StandardCharsets.US_ASCII);
+ // An element header carries its size in the high nibble when it fits
0..11, else via descriptor 12.
+ boolean wideInt = digitBytes.length > 11;
+ int valueElem = (wideInt ? 2 : 1) + digitBytes.length;
+ int objPayload = 2 + valueElem;
+ boolean wideObj = objPayload > 11;
+ byte[] payload = new byte[(wideObj ? 2 : 1) + objPayload];
+ int i = 0;
+ payload[i++] = (byte) (wideObj ? 0xCC : ((objPayload << 4) | 0x0C));
+ if (wideObj) {
+ payload[i++] = (byte) objPayload;
+ }
+ payload[i++] = (byte) 0x17; // text "n" (size 1)
+ payload[i++] = (byte) 'n';
+ payload[i++] = (byte) (wideInt ? 0xC3 : ((digitBytes.length << 4) | 0x03));
+ if (wideInt) {
+ payload[i++] = (byte) digitBytes.length;
+ }
+ System.arraycopy(digitBytes, 0, payload, i, digitBytes.length);
+ return parse(JsonPayloadFormat.SQLITE_JSONB.getParser(), payload).get("n");
+ }
+
+ @Test
+ public void testSqliteTwoByteSizeDescriptor()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // {"k": "<300 x 'a'>"} forces size descriptor 13 (2-byte big-endian size)
for both the text value and the
+ // enclosing object.
+ int textLen = 300;
+ int valueElem = 1 + 2 + textLen;
+ int objPayload = 2 + valueElem;
+ byte[] payload = new byte[3 + objPayload];
+ int i = 0;
+ payload[i++] = (byte) 0xDC; // OBJECT, size
descriptor 13
+ payload[i++] = (byte) (objPayload >>> 8);
+ payload[i++] = (byte) objPayload;
+ payload[i++] = (byte) 0x17; // text "k"
+ payload[i++] = (byte) 'k';
+ payload[i++] = (byte) 0xD7; // TEXT, size
descriptor 13
+ payload[i++] = (byte) (textLen >>> 8);
+ payload[i++] = (byte) textLen;
+ Arrays.fill(payload, i, i + textLen, (byte) 'a');
+ assertEquals(parse(parser, payload).get("k"), "a".repeat(textLen));
+ }
+
+ @Test
+ public void testSqliteErrors() {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // Reserved element type (low nibble 13).
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x0D)));
+ // Non-object top-level (a bare int).
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x13, 0x35)));
+ // Truncated: object claims 5 payload bytes but only 1 follows.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x5C, 0x17)));
+ // Object payload ends after a label, with no value for it.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x2C, 0x17, 0x61)));
+ // Object label must be a text element, not an int.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x3C, 0x13, 0x31, 0x01)));
+ }
+
+ @Test
+ public void testSqliteTopLevelElementMustExactlyFillThePayload()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // A bare empty object consumes the whole payload and is valid.
+ assertEquals(parse(parser, bytes(0x0C)), Map.of());
+
+ // SQLite's validity rule requires the outer element to exactly fill the
BLOB. Previously a top-level
+ // element declaring a short size decoded to a partial row and silently
dropped the trailing bytes:
+ // an empty object (size 0) followed by real data ...
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x0C, 0x17, 0x61, 0x13, 0x31)));
+ // ... and a well-formed {"a": 1} object with a stray trailing byte.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x4C, 0x17, 0x61, 0x13, 0x31, 0xFF)));
+ }
+
+ @Test
+ public void testSqliteWideSizeDescriptors()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // Descriptor 14: object size as a 4-byte big-endian int. Payload is
text("a") + int("1").
+ assertEquals(parse(parser, bytes(0xEC, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61,
0x13, 0x31)), Map.of("a", 1));
+ // Descriptor 15: object size as an 8-byte big-endian int.
+ assertEquals(parse(parser,
+ bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x17,
0x61, 0x13, 0x31)),
+ Map.of("a", 1));
+
+ // A uint64 size with the sign bit set decodes to a negative long and must
be rejected, not wrapped.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
+ bytes(0xFC, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x17)));
+ // A large positive size must be rejected by comparison, without
overflowing when added to the offset.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
+ bytes(0xFC, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x17)));
+ }
+
+ @Test
+ public void testSqliteJson5NumberVariants()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // {"h": -0x10} -> int5 with a sign and hex radix -> -16
+ assertEquals(parse(parser, bytes(0x8C, 0x17, 0x68, 0x54, 0x2D, 0x30, 0x78,
0x31, 0x30)), Map.of("h", -16));
+ // {"p": +1.5} -> float5 with a leading '+' (rejected by
Double.parseDouble until stripped)
+ assertEquals(parse(parser, bytes(0x7C, 0x17, 0x70, 0x46, 0x2B, 0x31, 0x2E,
0x35)), Map.of("p", 1.5));
+ // JSON5 floats may be non-finite: FLOAT5 "NaN" / "Infinity".
+ assertEquals(parse(parser, bytes(0x6C, 0x17, 0x78, 0x36, 0x4E, 0x61,
0x4E)).get("x"), Double.NaN);
+ assertEquals(parse(parser, bytes(0xBC, 0x17, 0x78, 0x86, 0x49, 0x6E, 0x66,
0x69, 0x6E, 0x69, 0x74, 0x79)).get("x"),
+ Double.POSITIVE_INFINITY);
+ }
+
+ @Test
+ public void testSqliteCanonicalNumbersRejectNonCanonicalTokens()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // TYPE_FLOAT is canonical RFC 8259 and must reject the permissive tokens
Double.parseDouble accepts, so a
+ // malformed float fails the record rather than ingesting a non-finite /
non-canonical value.
+ assertThrows(IllegalArgumentException.class, //
{"x": NaN}
+ () -> parse(parser, bytes(0x6C, 0x17, 0x78, 0x35, 0x4E, 0x61, 0x4E)));
+ assertThrows(IllegalArgumentException.class, //
{"x": Infinity}
+ () -> parse(parser, bytes(0xBC, 0x17, 0x78, 0x85, 0x49, 0x6E, 0x66,
0x69, 0x6E, 0x69, 0x74, 0x79)));
+ assertThrows(IllegalArgumentException.class, //
{"x": +1.5}
+ () -> parse(parser, bytes(0x7C, 0x17, 0x78, 0x45, 0x2B, 0x31, 0x2E,
0x35)));
+ assertThrows(IllegalArgumentException.class, //
{"x": 0x1p4} (Java hex float)
+ () -> parse(parser, bytes(0x8C, 0x17, 0x78, 0x55, 0x30, 0x78, 0x31,
0x70, 0x34)));
+
+ // TYPE_INT is likewise canonical: no leading '+' or leading zeros.
+ assertThrows(IllegalArgumentException.class, //
{"x": +5}
+ () -> parse(parser, bytes(0x5C, 0x17, 0x78, 0x23, 0x2B, 0x35)));
+ assertThrows(IllegalArgumentException.class, //
{"x": 007}
+ () -> parse(parser, bytes(0x6C, 0x17, 0x78, 0x33, 0x30, 0x30, 0x37)));
+
+ // The canonical happy paths still decode.
+ assertEquals(parse(parser, bytes(0x6C, 0x17, 0x78, 0x35, 0x31, 0x2E,
0x35)), Map.of("x", 1.5)); // 1.5
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x78, 0x23, 0x2D, 0x35)),
Map.of("x", -5)); // -5
+ }
+
+ @Test
+ public void testSqliteTextMustBeValidUtf8()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // Valid multibyte UTF-8 (é = 0xC3 0xA9) decodes.
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x78, 0x27, 0xC3, 0xA9)),
Map.of("x", "é"));
+ // A lone 0x80 (invalid UTF-8) in a value must fail the record rather than
becoming U+FFFD.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x4C, 0x17, 0x78, 0x17, 0x80)));
+ // Same for a field name.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
bytes(0x4C, 0x17, 0x80, 0x13, 0x31)));
+ }
+
+ // SQLite JSONB element type codes used by the builders below.
+ private static final int SQLITE_TYPE_INT = 3;
+ private static final int SQLITE_TYPE_TEXT = 7;
+ private static final int SQLITE_TYPE_ARRAY = 11;
+ private static final int SQLITE_TYPE_OBJECT = 12;
+
+ /// Builds a SQLite JSONB element (header + payload) whose declared size
exactly matches {@code payload},
+ /// choosing the narrowest size descriptor. Used to hand-build arbitrarily
large / deep valid fixtures.
+ private static byte[] sqliteElement(int type, byte[] payload) {
+ int size = payload.length;
+ byte[] header;
+ if (size <= 11) {
+ header = bytes((size << 4) | type);
+ } else if (size <= 0xFF) {
+ header = bytes((12 << 4) | type, size);
+ } else if (size <= 0xFFFF) {
+ header = bytes((13 << 4) | type, (size >>> 8) & 0xFF, size & 0xFF);
+ } else {
+ header = bytes((14 << 4) | type, (size >>> 24) & 0xFF, (size >>> 16) &
0xFF, (size >>> 8) & 0xFF, size & 0xFF);
+ }
+ byte[] out = new byte[header.length + payload.length];
+ System.arraycopy(header, 0, out, 0, header.length);
+ System.arraycopy(payload, 0, out, header.length, payload.length);
+ return out;
+ }
+
+ /// {@code {"<key>": <value element>}} as an exactly-filling SQLite JSONB
object.
+ private static byte[] sqliteObject(String key, byte[] valueElement) {
+ byte[] label = sqliteElement(SQLITE_TYPE_TEXT,
key.getBytes(StandardCharsets.US_ASCII));
+ byte[] payload = new byte[label.length + valueElement.length];
+ System.arraycopy(label, 0, payload, 0, label.length);
+ System.arraycopy(valueElement, 0, payload, label.length,
valueElement.length);
+ return sqliteElement(SQLITE_TYPE_OBJECT, payload);
+ }
+
+ /// {@code {"a": [[[ ... ]]]}} with {@code depth} nested arrays, innermost
empty. Each level is a valid,
+ /// exactly-filling element, so the whole payload is well-formed apart from
its depth.
+ private static byte[] sqliteNestedArrays(int depth) {
+ byte[] element = sqliteElement(SQLITE_TYPE_ARRAY, new byte[0]);
+ for (int i = 0; i < depth; i++) {
+ element = sqliteElement(SQLITE_TYPE_ARRAY, element);
+ }
+ return sqliteObject("a", element);
+ }
+
+ @Test
+ public void testSqliteRejectsExcessiveNestingDepth()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // Moderate nesting decodes -- confirms the fixture builder and normal
nesting are fine.
+ assertTrue(parse(parser, sqliteNestedArrays(50)).get("a") instanceof List);
+ // Past the depth cap it must fail as a bad record
(IllegalArgumentException), NOT recurse into a
+ // StackOverflowError, which is an Error that would escape decode-error
handling and fail the segment.
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
sqliteNestedArrays(1200)));
+ }
+
+ @Test
+ public void testSqliteRejectsOversizedNumericToken()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // A 1000-digit canonical integer is accepted (parity with Jackson's
number-length cap).
+ byte[] atLimit =
+ sqliteObject("n", sqliteElement(SQLITE_TYPE_INT,
"9".repeat(1000).getBytes(StandardCharsets.US_ASCII)));
+ assertTrue(parse(parser, atLimit).get("n") instanceof BigInteger);
+ // One digit past the cap is rejected before the ~O(n^2) BigInteger parse
runs.
+ byte[] overLimit =
+ sqliteObject("n", sqliteElement(SQLITE_TYPE_INT,
"9".repeat(1001).getBytes(StandardCharsets.US_ASCII)));
+ assertThrows(IllegalArgumentException.class, () -> parse(parser,
overLimit));
+ }
+
+ @Test
+ public void testSqliteTextEscapeVariants()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // TEXTJ \\u escape: {"u": "A"} -> "A"
+ assertEquals(parse(parser, bytes(0x9C, 0x17, 0x75, 0x68, 0x5C, 0x75, 0x30,
0x30, 0x34, 0x31)), Map.of("u", "A"));
+ // TEXT5 extensions, one payload each: \' -> ' , \v -> VT, \0 -> NUL, \q
-> q (passthrough)
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x27)),
Map.of("e", "'"));
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x76)),
Map.of("e", "\u000B"));
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x30)),
Map.of("e", "\0"));
+ assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x71)),
Map.of("e", "q"));
+ // TEXT5 line continuation: "a\<LF>b" -> "ab"
+ assertEquals(parse(parser, bytes(0x7C, 0x17, 0x65, 0x49, 0x61, 0x5C, 0x0A,
0x62)), Map.of("e", "ab"));
+ }
+
+ @Test
+ public void testSqliteNestedElementCannotOverrunItsContainer() {
+ JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // object(size 8){ text("a"), array(size 1){ text(size 4) }, "b": 7 }
+ // The array declares a 1-byte payload, but the text element inside it
declares 4 bytes -- which would run
+ // past the array's end and swallow the object's remaining "b": 7 pair.
Must be rejected, not silently
+ // truncated.
+ assertThrows(IllegalArgumentException.class,
+ () -> parse(parser, bytes(0x8C, 0x17, 0x61, 0x1B, 0x47, 0x17, 0x62,
0x13, 0x37)));
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // AUTO detection + cross-format equivalence
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testAutoDetection()
+ throws Exception {
+ assertSame(JsonPayloadFormat.detectParser(utf8("{\"a\":1}"), 0,
7).getClass(), TextJsonPayloadParser.class);
+ byte[] smile = smile(Map.of("a", 1));
+ assertSame(JsonPayloadFormat.detectParser(smile, 0,
smile.length).getClass(), SmileJsonPayloadParser.class);
+ byte[] cbor = cborSelfDescribed(Map.of("a", 1));
+ assertSame(JsonPayloadFormat.detectParser(cbor, 0,
cbor.length).getClass(), CborJsonPayloadParser.class);
+ byte[] pg = postgres("{\"a\":1}");
+ assertSame(JsonPayloadFormat.detectParser(pg, 0, pg.length).getClass(),
PostgresJsonbPayloadParser.class);
+ byte[] sqlite = bytes(0x3C, 0x17, 0x62, 0x01);
+ assertSame(JsonPayloadFormat.detectParser(sqlite, 0,
sqlite.length).getClass(), SqliteJsonbPayloadParser.class);
+ }
+
+ @Test
+ public void testSqliteDetectionRequiresAnExactlyFillingObject()
+ throws Exception {
+ JsonPayloadParser sqlite = JsonPayloadFormat.SQLITE_JSONB.getParser();
+ // The OBJECT low nibble alone would claim one arbitrary binary byte in
sixteen, so detection additionally
+ // requires the declared top-level size to exactly fill the payload.
+ assertTrue(sqlite.matches(bytes(0x4C, 0x17, 0x61, 0x13, 0x31), 0, 5));
// size 4, 4 bytes follow
+ assertFalse(sqlite.matches(bytes(0x4C, 0x17, 0x61), 0, 3));
// size 4, only 2 bytes follow
+ assertFalse(sqlite.matches(bytes(0x0C, 0x17, 0x61, 0x13, 0x31), 0, 5));
// size 0, but 4 bytes follow
+ assertFalse(sqlite.matches(bytes(0x4C, 0x17, 0x61, 0x13, 0x31, 0xFF), 0,
6)); // size 4, trailing byte
+
+ // Random binary that merely happens to carry the OBJECT nibble is no
longer claimed ...
+ assertFalse(sqlite.matches(bytes(0xFC, 0xDE, 0xAD, 0xBE, 0xEF), 0, 5));
+ assertSame(JsonPayloadFormat.detectParser(bytes(0x0C, 0x17, 0x61, 0x13,
0x31), 0, 5).getClass(),
+ TextJsonPayloadParser.class);
+
+ // ... but the degenerate empty object, which does exactly fill its
payload, still is.
+ assertTrue(sqlite.matches(bytes(0x0C), 0, 1));
+
+ // Every size descriptor participates in the same check. Each positive
case is the object {"a": 1} whose
+ // 4-byte payload is declared through a different-width size field, and
each negative case is the same
+ // payload with the declared size off by one.
+ // Descriptor 12: 1-byte size.
+ assertTrue(sqlite.matches(bytes(0xCC, 0x04, 0x17, 0x61, 0x13, 0x31), 0,
6));
+ assertFalse(sqlite.matches(bytes(0xCC, 0x05, 0x17, 0x61, 0x13, 0x31), 0,
6));
+ // Descriptor 13: 2-byte big-endian size.
+ assertTrue(sqlite.matches(bytes(0xDC, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31),
0, 7));
+ assertFalse(sqlite.matches(bytes(0xDC, 0x00, 0x05, 0x17, 0x61, 0x13,
0x31), 0, 7));
+ // A byte-order slip here would read 0x0400 rather than 0x0004.
+ assertFalse(sqlite.matches(bytes(0xDC, 0x04, 0x00, 0x17, 0x61, 0x13,
0x31), 0, 7));
+ // Descriptor 14: 4-byte big-endian size.
+ assertTrue(sqlite.matches(bytes(0xEC, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61,
0x13, 0x31), 0, 9));
+ assertFalse(sqlite.matches(bytes(0xEC, 0x00, 0x00, 0x00, 0x05, 0x17, 0x61,
0x13, 0x31), 0, 9));
+ // Descriptor 15: 8-byte big-endian size.
+ assertTrue(sqlite.matches(
+ bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x17,
0x61, 0x13, 0x31), 0, 13));
+ assertFalse(sqlite.matches(
+ bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, 0x17,
0x61, 0x13, 0x31), 0, 13));
+ // An 8-byte size with the sign bit set decodes negative and can never
equal a length.
+ assertFalse(sqlite.matches(bytes(0xFC, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF,
0xFF, 0xFF, 0x17), 0, 10));
+
+ // Truncated headers must not read past the payload.
+ assertFalse(sqlite.matches(bytes(0xCC), 0, 1));
+ assertFalse(sqlite.matches(bytes(0xDC, 0x00), 0, 2));
+ assertFalse(sqlite.matches(bytes(0xEC, 0x00), 0, 2));
+ assertFalse(sqlite.matches(bytes(0xFC, 0x00, 0x00, 0x00), 0, 4));
+
+ // Detection and parsing must agree: anything matches() claims here,
parse() accepts.
+ assertEquals(parse(sqlite, bytes(0xCC, 0x04, 0x17, 0x61, 0x13, 0x31)),
Map.of("a", 1));
+ assertEquals(parse(sqlite, bytes(0xDC, 0x00, 0x04, 0x17, 0x61, 0x13,
0x31)), Map.of("a", 1));
+ }
+
+ @Test
+ public void testAutoDetectionFallbackAndAdversarialInput() {
+ // Nothing matches -> TEXT fallback.
+ assertSame(JsonPayloadFormat.detectParser(bytes(0xFF), 0, 1).getClass(),
TextJsonPayloadParser.class);
+ // Empty payload -> TEXT fallback; matches() must not throw on too-short
input.
+ assertSame(JsonPayloadFormat.detectParser(bytes(), 0, 0).getClass(),
TextJsonPayloadParser.class);
+ // Truncated Smile header must not be claimed by the Smile parser.
+ assertSame(JsonPayloadFormat.detectParser(bytes(0x3A, 0x29), 0,
2).getClass(), TextJsonPayloadParser.class);
+ // A lone 0x01 with no JSON body must not be claimed by the Postgres
parser.
+ assertSame(JsonPayloadFormat.detectParser(bytes(0x01, 0x02), 0,
2).getClass(), TextJsonPayloadParser.class);
+ }
+
+ @Test
+ public void testCrossFormatEquivalence()
+ throws Exception {
+ // The same logical document must decode to an equal Map through every
format.
+ Map<String, Object> expected = new HashMap<>();
+ expected.put("count", 7);
+ expected.put("ok", true);
+
+ byte[] text = utf8("{\"count\":7,\"ok\":true}");
+ byte[] pg = postgres("{\"count\":7,\"ok\":true}");
+ byte[] smile = smile(expected);
+ byte[] cbor = new ObjectMapper(new
CBORFactory()).writeValueAsBytes(expected);
+ // object(size 12, descriptor 12){ text("count"), int("7"), text("ok"),
true }
+ byte[] sqlite = bytes(0xCC, 0x0C, 0x57, 0x63, 0x6F, 0x75, 0x6E, 0x74,
0x13, 0x37, 0x27, 0x6F, 0x6B, 0x01);
+
+ assertEquals(parse(JsonPayloadFormat.TEXT.getParser(), text), expected);
+ assertEquals(parse(JsonPayloadFormat.POSTGRES_JSONB.getParser(), pg),
expected);
+ assertEquals(parse(JsonPayloadFormat.SMILE.getParser(), smile), expected);
+ assertEquals(parse(JsonPayloadFormat.CBOR.getParser(), cbor), expected);
+ assertEquals(parse(JsonPayloadFormat.SQLITE_JSONB.getParser(), sqlite),
expected);
+ }
+
+ @Test
+ public void testAutoParserDelegates()
+ throws Exception {
+ JsonPayloadParser auto = JsonPayloadFormat.AUTO.getParser();
+ assertTrue(auto.matches(bytes(0x00), 0, 1));
+ assertEquals(parse(auto, postgres("{\"a\":1}")), Map.of("a", 1));
+ assertEquals(parse(auto, utf8("{\"a\":1}")), Map.of("a", 1));
+ assertEquals(parse(auto, smile(Map.of("a", 1))), Map.of("a", 1));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
