Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder#18953
Conversation
There was a problem hiding this comment.
Pull request overview
This PR extends the pinot-json stream decoder to support multiple binary JSON payload encodings (PostgreSQL jsonb wire framing, SQLite JSONB, Smile, CBOR) in addition to UTF-8 text JSON by introducing a payload-format abstraction with optional per-message auto-detection.
Changes:
- Introduces
JsonPayloadParser+JsonPayloadFormat(AUTOdefault) and wiresJSONMessageDecoderto parse via the selected parser. - Adds concrete payload parsers for TEXT, PostgreSQL jsonb (version byte + text body), SQLite JSONB (hand-rolled), Smile, and CBOR (Jackson-backed).
- Adds unit/integration-style tests for per-format decoding and AUTO detection, plus module dependencies for Jackson Smile/CBOR and updates
LICENSE-binary.
Reviewed changes
Copilot reviewed 14 out of 14 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java | Adds jsonFormat config, parser selection, and uses parser in decode() |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java | Defines the parser contract (matches + parse) for text/binary payloads |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java | Adds enum of formats, config parsing, and AUTO detection order |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java | Implements AUTO parser delegating to detected concrete parser |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java | Text JSON parser delegating to JsonUtils.bytesToMap |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java | Parses Postgres jsonb wire framing (version byte + text JSON) |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java | Implements SQLite JSONB decoder with container-bound validation |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java | Shared base for Jackson-backed binary formats using ObjectReader |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java | Smile parser + header-based AUTO detection |
| pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java | CBOR parser + self-describe-tag-based AUTO detection |
| pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java | End-to-end decode → GenericRow tests for configured and AUTO formats |
| pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java | Unit tests for parsers, detection, equivalence, and error cases |
| pinot-plugins/pinot-input-format/pinot-json/pom.xml | Adds Jackson Smile + CBOR dependencies |
| LICENSE-binary | Records new binary dependencies |
Comments suppressed due to low confidence (1)
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java:83
- The exception wrapper always builds
new String(payload, offset, length)using the platform default charset. With the new binary payload formats (Smile/CBOR/SQLite JSONB), this can allocate a very large string of unreadable data, potentially throw due to invalid byte sequences, and can leak raw binary into logs. Consider logging only metadata (length/offset) plus a small hex prefix instead.
} catch (Exception e) {
throw new RuntimeException(
"Caught exception while decoding JSON record with payload: " + new String(payload, offset, length), e);
}
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #18953 +/- ##
============================================
+ Coverage 65.05% 65.09% +0.03%
Complexity 1403 1403
============================================
Files 3399 3407 +8
Lines 212808 213087 +279
Branches 33568 33606 +38
============================================
+ Hits 138443 138701 +258
+ Misses 63223 63217 -6
- Partials 11142 11169 +27
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
All review comments are addressed. Summary of the two follow-up commits:
Also taking Copilot's low-confidence note on the exception message, which is a real problem now that payloads can be binary: Module is at 65 tests, all passing; spotless / checkstyle / license clean. |
|
Two more commits, then CI green on
Final state: 66 tests in the module, all green; spotless / checkstyle / license clean; all 13 CI checks passing. |
xiangfu0
left a comment
xiangfu0
left a comment
There was a problem hiding this comment.
Found one high-signal issue; see inline comment.
Jackie-Jiang
added
ingestion
xiangfu0
force-pushed
the
claude/jsonb-pinot-json-decoder-5af7ba
branch
from
56555c4 to
976ec7d
Compare
xiangfu0
left a comment
xiangfu0
left a comment
There was a problem hiding this comment.
Found two high-signal SQLite JSONB fail-fast correctness issues; see inline comments.
|
Added a user manual with examples in It documents the (The branch was also rebased onto latest master; CI is green on the pre-README head |
xiangfu0
force-pushed
the
claude/jsonb-pinot-json-decoder-5af7ba
branch
from
f270bee to
b6e2008
Compare
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.
…eir 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.
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.
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.
…sted 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.
…act 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.
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.
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.
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.
xiangfu0
force-pushed
the
claude/jsonb-pinot-json-decoder-5af7ba
branch
from
b6e2008 to
9d164f7
Compare
|
Reviewed this focusing on the two hand-rolled binary parsers (which parse untrusted stream bytes on the ingestion hot path). Overall this is careful work — bounds arithmetic is defensive (compare-without-add to avoid overflow, sign-bit One blocker, plus a few smaller notes. 🔴 Blocker: unbounded recursion in
|
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.
|
@gortiz thanks for the careful read — the recursion blocker was a real hole. Addressed in 🔴 Blocker — unbounded recursion / 🟡 Numeric-token length. Fixed for parity. 🟡 Bare 🟡 Trailing bytes after Smile/CBOR. No change, as you noted — it matches the existing text-JSON behavior via Module is at 72 tests; spotless/checkstyle/license clean. (Two CI compat jobs are red on an unrelated Apache Kafka-tarball mirror download failure in test setup, not this change — auto-retries have been re-hitting the mirror outage.) 🤖 Addressed by Claude Code |
gortiz
left a comment
gortiz
left a comment
There was a problem hiding this comment.
Verified the fix in ed8144ac against the recursion blocker I raised.
Depth cap — readElement/readArray/readObject now thread a depth counter and nextDepth() throws IllegalArgumentException once nesting would exceed MAX_NESTING_DEPTH = 1000 (Jackson's StreamReadConstraints default). Depth increments exactly once per container boundary before descending, and label-position recursion is bounded by the same cap, so deep nesting now flows through the normal decode-error handling instead of escalating to a StackOverflowError that fails the whole consuming segment. testSqliteRejectsExcessiveNestingDepth genuinely distinguishes fixed from broken — a 1200-deep exactly-filling payload asserts IllegalArgumentException (a StackOverflowError would not match), and a 50-deep one still decodes.
Numeric-token length — capped at 1000 chars in Cursor.ascii() (the numeric-only accessor) before the ~O(n²) BigInteger parse; tested at and past the limit.
Bare 0x0C under AUTO — documented in the README caveats, the right disposition for an inherent heuristic property.
All three items from my review are resolved and the blocker is fixed at the root. LGTM.
On trailing bytes: agree with leaving Smile/CBOR consistent with text JSON rather than tightening only the binary paths. FAIL_ON_TRAILING_TOKENS applied uniformly across all Jackson-backed formats would be a reasonable separate follow-up, but not required here.
Summary
JSONMessageDecoderonly understood UTF-8 text JSON. This adds a pluggable payload-format layer so a stream can also carry PostgreSQL jsonb, SQLite JSONB, Smile, or CBOR, selected via a newjsonFormatdecoder property.Every format decodes to the same
Map<String, Object>value contract Jackson produces for text JSON, so the existingJSONRecordExtractorhandles all of them unchanged.Configuration
jsonFormatTEXTPOSTGRES_JSONBjsonbwire formatSQLITE_JSONBSMILECBORAUTOBackward compatibility
An unset
jsonFormatmeansTEXT, so existing tables are byte-for-byte unchanged. No existing config keys, SPI signatures, or serialization formats change.AUTOis deliberately opt-in rather than the default. It never mis-routes a well-formed text JSON document — a top-level{/[(optionally after whitespace) collides with none of the binary signatures — but it is still a heuristic over a few leading bytes, so it can claim a corrupt message that text decoding would have rejected outright. Enabling that for every existing stream would be a silent correctness change. Detection order is Smile → CBOR → Postgres → SQLite → text.To keep
AUTOitself safe, SQLite detection requires more than the OBJECT low nibble (which one arbitrary binary byte in sixteen would satisfy): the top-level object's declared size must exactly fill the payload, the same validity rule the parser enforces.Notes on the two hand-rolled decoders
PostgreSQL jsonb's wire format is not its on-disk
JsonbContainerlayout.jsonb_sendrenders the value withJsonbToCStringand emits a version byte followed by text JSON;jsonb_recvreverses it. Every standard binary producer — the v3 extended-query protocol,COPY ... WITH (FORMAT binary), and logical replication viapgoutput— routes through that same send function. So this parser strips the version byte and parses the text body, which also makes its type contract identical toTEXT.SQLite JSONB is a genuine binary format, decoded per spec including the JSON5 element types (
INT5,FLOAT5,TEXTJ,TEXT5,TEXTRAW) and all four size descriptors. SQLite's validity rule that the top-level element exactly fill the BLOB is enforced, so a short-declared element cannot yield a partial (or empty) row. Nested elements are bounded by their enclosing container rather than merely by the payload, so a crafted element cannot overrun its parent and swallow later siblings.TEXTJrejects invalid escapes rather than passing them through; string unescaping is single-pass with a no-allocation fast path.Documentation
A user manual for the plugin ships in the PR at
pinot-plugins/pinot-input-format/pinot-json/README.md: thejsonFormatvalues, theTEXTdefault and opt-inAUTO(with the per-message detection order and signatures), real-time table config examples for text / a pinned binary format /AUTO, per-format notes, and the decode-failure diagnostic.Dependencies
Adds
jackson-dataformat-smileandjackson-dataformat-cbor. Versions are managed by the existingjackson-bomimport (no version overrides). Both are recorded inLICENSE-binary.Testing
68 tests pass in the module. Coverage includes:
Integer/Long/BigIntegernarrowing tiers,Double, andFloat/byte[]passthrough for the binary formats).Map.jsonFormatdecodes text and rejects Smile, CBOR, Postgres, SQLite, and a bare0x0c, rather than silently detecting them.0xFF, truncated Smile header, lone0x01), that tag-less CBOR is not auto-detected but still decodes when pinned, and that SQLite detection accepts only exactly-filling objects.TEXTJescape, and end-to-enddecode()wrapping parse failures in a bounded, charset-safeRuntimeException.decode()→GenericRowfor every format, both pinned and auto-detected.Release notes
Adds the
jsonFormatdecoder property toJSONMessageDecoder, supportingTEXT(default),POSTGRES_JSONB,SQLITE_JSONB,SMILE,CBOR, and an opt-inAUTOthat detects the encoding per message. Existing tables are unaffected: an unset property behaves exactly as before.