You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/07/10 00:02:42 UTC
[PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
xiangfu0 opened a new pull request, #18953:
URL: https://github.com/apache/pinot/pull/18953
## Summary
`JSONMessageDecoder` only 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 new `jsonFormat` decoder property.
Every format decodes to the same `Map<String, Object>` value contract Jackson produces for text JSON, so the existing `JSONRecordExtractor` handles all of them unchanged.
## Configuration
| `jsonFormat` | Payload |
|---|---|
| *(unset)* / `AUTO` | Detected per message from leading magic/version bytes; falls back to text |
| `TEXT` | UTF-8 text JSON (historical behavior) |
| `POSTGRES_JSONB` | PostgreSQL `jsonb` wire format |
| `SQLITE_JSONB` | SQLite 3.45+ JSONB |
| `SMILE` | Jackson Smile |
| `CBOR` | CBOR (RFC 8949) |
```
"stream.kafka.decoder.prop.jsonFormat": "SQLITE_JSONB"
```
## Backward compatibility
The default changes from "always text" to `AUTO`. This is **behavior-preserving for text JSON**: detection is allocation-free and cannot mis-route a well-formed text document, because a top-level `{` / `[` (optionally after whitespace) collides with none of the binary signatures — Smile `3A 29 0A`, CBOR `D9 D9 F7`, Postgres `0x01`+JSON, SQLite object nibble `0x?C`. Anything unrecognized falls back to text. Operators who want to skip detection entirely can pin `jsonFormat: TEXT`.
Detection order is Smile → CBOR → Postgres → SQLite → text. No existing config keys, SPI signatures, or serialization formats change.
## 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 — the v3 extended-query protocol, `COPY ... WITH (FORMAT binary)`, and logical replication via `pgoutput` — 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 to `TEXT`.
**SQLite JSONB** is a genuine binary format, decoded per [spec](https://sqlite.org/jsonb.html) including the JSON5 element types (`INT5`, `FLOAT5`, `TEXTJ`, `TEXT5`, `TEXTRAW`) and all four size descriptors. Nested elements are bounded by their enclosing container rather than merely by the payload, so a crafted element cannot overrun its parent and silently swallow later sibling key/value pairs. `TEXTJ` rejects invalid escapes rather than passing them through; string unescaping is single-pass with a no-allocation fast path.
## Dependencies
Adds `jackson-dataformat-smile` and `jackson-dataformat-cbor`. Versions are managed by the existing `jackson-bom` import (no version overrides). Both are recorded in `LICENSE-binary`.
## Testing
59 tests pass in the module (34 new). Coverage includes:
- Per-format decoding with **type-strict** assertions (`Integer` / `Long` / `BigInteger` narrowing tiers, `Double`, and `Float` / `byte[]` passthrough for the binary formats).
- **Cross-format equivalence** — one logical document decoded through all five formats yields an equal `Map`.
- **AUTO detection**, including the fallback path and adversarial input (empty payload, `0xFF`, truncated Smile header, lone `0x01`), and that tag-less CBOR is *not* auto-detected but still decodes when pinned.
- **Negative / malformed input** — truncated payloads, reserved element types, non-object top level, unsupported version byte, nested-element container overrun, invalid `TEXTJ` escape, and end-to-end `decode()` wrapping parse failures in `RuntimeException`.
- End-to-end `decode()` → `GenericRow` for every format, both pinned and auto-detected.
## Release notes
Adds the `jsonFormat` decoder property to `JSONMessageDecoder`; unset behaves as `AUTO` (per-message format detection, text-JSON fallback).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556923567
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java:
##########
@@ -20,34 +20,50 @@
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}.
+ *
+ * <p>When unset (equivalently {@code 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 {@code {} / {@code [} (optionally after whitespace) collides with none of the
+ * binary signatures. Pin {@code TEXT} to skip detection entirely. See {@link JsonPayloadFormat}.
Review Comment:
Fixed in `4b7401f` — replaced with `<code>{</code>` / `<code>[</code>` HTML entities so the inline tag no longer terminates early; verified with `mvn javadoc:javadoc`.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * 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; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON.
+ 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), returning [#AUTO] when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
+ }
+ 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");
+ }
Review Comment:
Fixed in `4b7401f` — now `throw new IllegalArgumentException("Unsupported jsonFormat ...", e)`, and `ed03fea` adds a test asserting the cause survives (the previous `assertThrows` passed even without it).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java:
##########
@@ -0,0 +1,172 @@
+/**
+ * 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.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;
+
+
+/// 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");
+
+ 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;
+ }
+
+ // 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);
+ }
+
+ @Test
+ public void testDefaultIsText()
+ throws Exception {
+ assertRich(decode(Map.of(), RICH_FIELDS, "{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes()));
+ }
Review Comment:
Fixed in `4b7401f` — renamed to `testUnsetFormatAutoDetectsText`, with a doc comment noting the default is AUTO rather than a pinned TEXT format.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java:
##########
@@ -0,0 +1,422 @@
+/**
+ * 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.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+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;
+ }
+
+ /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text JSON body.
+ private static byte[] postgres(String json) {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ 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() {
+ assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT);
+ 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);
+ assertThrows(IllegalArgumentException.class, () -> JsonPayloadFormat.fromConfig("bson"));
+ }
+
+ // ------------------------------------------------------------------------------------------------------
+ // Text
+ // ------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testText()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser();
+ Map<String, Object> map = parse(parser, " {\"a\": 1, \"b\": \"x\"}".getBytes());
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertTrue(parser.matches("{\"a\":1}".getBytes(), 0, 7));
+ assertTrue(parser.matches(" [1,2]".getBytes(), 0, 7));
Review Comment:
Fixed in `4b7401f` — added a `utf8(String)` helper and routed every fixture through it; `JSONMessageDecoderBinaryTest` uses a `StandardCharsets.UTF_8` `TEXT_DOC` constant. (Left the pre-existing `line.getBytes()` in `JSONMessageDecoderTest`, untouched by this PR.)
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3555564561
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java:
##########
@@ -0,0 +1,172 @@
+/**
+ * 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.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;
+
+
+/// 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");
+
+ 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;
+ }
+
+ // 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);
+ }
+
+ @Test
+ public void testDefaultIsText()
+ throws Exception {
+ assertRich(decode(Map.of(), RICH_FIELDS, "{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes()));
+ }
Review Comment:
The test name `testDefaultIsText` is misleading now that the decoder default is `AUTO` (per-message detection) rather than a pinned TEXT format. The assertion is still correct (AUTO should decode text), so renaming the test will reduce confusion.
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java:
##########
@@ -20,34 +20,50 @@
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}.
+ *
+ * <p>When unset (equivalently {@code 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 {@code {} / {@code [} (optionally after whitespace) collides with none of the
+ * binary signatures. Pin {@code TEXT} to skip detection entirely. See {@link JsonPayloadFormat}.
Review Comment:
The class Javadoc has a malformed nested {@code ...} construct: `{@code {} / {@code [}` will break Javadoc rendering (and can fail doclint) because `}` terminates the tag early and the second tag is missing its closing brace. Use explicit character literals instead (e.g., `'{', '['`) or escape braces.
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * 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; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON.
+ 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), returning [#AUTO] when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
+ }
+ 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");
+ }
Review Comment:
`fromConfig()` wraps the `valueOf()` failure but drops the original exception as the cause, which makes debugging invalid configs harder (the stack trace no longer points at the actual parse failure). It’s usually better to keep the original exception as the cause.
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java:
##########
@@ -0,0 +1,422 @@
+/**
+ * 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.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+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;
+ }
+
+ /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text JSON body.
+ private static byte[] postgres(String json) {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ 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() {
+ assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT);
+ 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);
+ assertThrows(IllegalArgumentException.class, () -> JsonPayloadFormat.fromConfig("bson"));
+ }
+
+ // ------------------------------------------------------------------------------------------------------
+ // Text
+ // ------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testText()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser();
+ Map<String, Object> map = parse(parser, " {\"a\": 1, \"b\": \"x\"}".getBytes());
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertTrue(parser.matches("{\"a\":1}".getBytes(), 0, 7));
+ assertTrue(parser.matches(" [1,2]".getBytes(), 0, 7));
Review Comment:
These tests use `String#getBytes()` without specifying a charset, which makes them depend on the platform default charset. Using `StandardCharsets.UTF_8` keeps the fixtures deterministic across environments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556950783
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * 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; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON.
+ 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), returning [#AUTO] when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
Review Comment:
Agreed — you're right on both counts, and I'd been keeping AUTO as the default on the strength of an earlier "detect when unconfigured" request. Fixed in `68af742`, taking both remedies you offered:
**1. Unset `jsonFormat` now means `TEXT`.** `fromConfig(null)` returns `TEXT`, so existing tables are byte-for-byte unchanged; `AUTO` must be requested explicitly. New test `testUnsetFormatDoesNotAutoDetectBinary` asserts an unset format decodes text and *rejects* Smile, CBOR, Postgres, SQLite and a bare `0x0c`.
**2. SQLite detection now sits behind a stronger validity gate.** `matches()` additionally requires the top-level object's declared size to exactly fill the payload — the same rule `parse()` enforces — so the single nibble can no longer hand a corrupt message to the parser. It is allocation-free and covers all four size descriptors, including an 8-byte size whose sign bit is set. `testSqliteDetectionRequiresAnExactlyFillingObject` pins it: `0x0c`-then-data and `0xfc deadbeef` are no longer claimed and fall back to text.
The lone `0x0c` you identified does remain a validly-exact empty object under an *explicit* `SQLITE_JSONB`/`AUTO`, which is correct for the format — but it can no longer reach a text stream. PR description and release-note line updated to match.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3561254017
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,433 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @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);
+ 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 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 = parseInt(cursor.ascii(start, end), 10);
+ break;
+ case TYPE_INT5:
+ result = parseInt5(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT:
+ case TYPE_FLOAT5:
+ result = parseFloat(cursor.ascii(start, end));
Review Comment:
Fixed in `f270bee`. `TYPE_FLOAT` now validates against the RFC 8259 number grammar (`-?(?:0|[1-9][0-9]*)(?:\.[0-9]+)?(?:[eE][+-]?[0-9]+)?`) before `Double.parseDouble`, so `NaN`, `Infinity`, a leading `+`, and Java hex floats are rejected; the permissive parser is kept only for `TYPE_FLOAT5`.
Following the same principle ("permissive only for the *5 types"), I applied it to canonical `TYPE_INT` as well, which had the analogous leniency via `Long.parseLong`/`BigInteger` (leading `+`, leading zeros). Happy to narrow it back to `TYPE_FLOAT` only if you'd rather keep this change scoped to what you flagged.
Regressions in `testSqliteCanonicalNumbersRejectNonCanonicalTokens`: canonical `NaN`/`Infinity`/`+1.5`/hex-float and `+5`/`007` throw, while `testSqliteJson5NumberVariants` confirms `FLOAT5` still accepts `NaN`/`Infinity`/`+1.5` and the canonical happy paths still decode.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,433 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @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);
+ 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 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 = parseInt(cursor.ascii(start, end), 10);
+ break;
+ case TYPE_INT5:
+ result = parseInt5(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT:
+ case TYPE_FLOAT5:
+ result = parseFloat(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);
+ case TYPE_OBJECT:
+ return readObject(cursor, end);
+ 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) {
+ List<Object> list = new ArrayList<>();
+ while (cursor._pos < end) {
+ list.add(readElement(cursor, end));
+ }
+ return list;
+ }
+
+ private static Map<String, Object> readObject(Cursor cursor, int end) {
+ Map<String, Object> map = Maps.newHashMapWithExpectedSize(4);
+ while (cursor._pos < end) {
+ Object label = readElement(cursor, end);
+ 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));
+ }
+ return map;
+ }
+
+ 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;
+ }
+
+ private static Double parseFloat(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;
+ }
+
+ private String ascii(int start, int end) {
+ return new String(_buf, start, end - start, StandardCharsets.US_ASCII);
+ }
+
+ private String utf8(int start, int end) {
+ return new String(_buf, start, end - start, StandardCharsets.UTF_8);
Review Comment:
Fixed in `f270bee`. Text is now decoded with a `CharsetDecoder` configured `onMalformedInput(CodingErrorAction.REPORT)` / `onUnmappableCharacter(REPORT)` instead of `new String(..., UTF_8)`, so invalid UTF-8 throws `IllegalArgumentException` and fails the record rather than substituting U+FFFD into a field name or value.
Regression in `testSqliteTextMustBeValidUtf8`: valid multibyte UTF-8 (`é` = `0xC3 0xA9`) still decodes, while a lone `0x80` in either a value or a field name is rejected.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3558089941
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,433 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @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);
+ 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 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 = parseInt(cursor.ascii(start, end), 10);
+ break;
+ case TYPE_INT5:
+ result = parseInt5(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT:
+ case TYPE_FLOAT5:
+ result = parseFloat(cursor.ascii(start, end));
Review Comment:
`TYPE_FLOAT` is the canonical JSON numeric form, but this path shares the permissive JSON5 parser used for `TYPE_FLOAT5`. `Double.parseDouble()` accepts values such as `NaN`, `Infinity`, leading `+`, and Java hex floats, so malformed SQLite JSONB can ingest values that text JSON/Jackson would reject instead of failing the record. Please split `TYPE_FLOAT` onto an RFC-8259-style number validator and keep the permissive handling only for `TYPE_FLOAT5`, with regressions for non-finite canonical floats.
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,433 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @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);
+ 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 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 = parseInt(cursor.ascii(start, end), 10);
+ break;
+ case TYPE_INT5:
+ result = parseInt5(cursor.ascii(start, end));
+ break;
+ case TYPE_FLOAT:
+ case TYPE_FLOAT5:
+ result = parseFloat(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);
+ case TYPE_OBJECT:
+ return readObject(cursor, end);
+ 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) {
+ List<Object> list = new ArrayList<>();
+ while (cursor._pos < end) {
+ list.add(readElement(cursor, end));
+ }
+ return list;
+ }
+
+ private static Map<String, Object> readObject(Cursor cursor, int end) {
+ Map<String, Object> map = Maps.newHashMapWithExpectedSize(4);
+ while (cursor._pos < end) {
+ Object label = readElement(cursor, end);
+ 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));
+ }
+ return map;
+ }
+
+ 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;
+ }
+
+ private static Double parseFloat(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;
+ }
+
+ private String ascii(int start, int end) {
+ return new String(_buf, start, end - start, StandardCharsets.US_ASCII);
+ }
+
+ private String utf8(int start, int end) {
+ return new String(_buf, start, end - start, StandardCharsets.UTF_8);
Review Comment:
SQLite JSONB text payloads are UTF-8, but `new String(..., UTF_8)` replaces malformed byte sequences instead of failing. That lets a corrupt JSONB record ingest mutated field names or values where text JSON/Jackson would reject the message, which is silent data corruption on the stream path. Please decode with a `CharsetDecoder` configured with `CodingErrorAction.REPORT` and add an invalid-UTF-8 regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4937793538
Added a user manual with examples in `23ea2d4`: [`pinot-plugins/pinot-input-format/pinot-json/README.md`](https://github.com/apache/pinot/blob/claude/jsonb-pinot-json-decoder-5af7ba/pinot-plugins/pinot-input-format/pinot-json/README.md).
It documents the `jsonFormat` decoder property — the supported values, the `TEXT` default and opt-in `AUTO` detection (with the per-message detection order and each format's signature), real-time table `streamConfigMaps` examples for text, a pinned binary format, and `AUTO`, plus per-format notes (Postgres wire framing, SQLite JSON5 + exact-fill, Smile/CBOR magic bytes) and the bounded decode-failure diagnostic.
(The branch was also rebased onto latest master; CI is green on the pre-README head `976ec7d`, and this doc-only commit fast-forwards on top.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "gortiz (via GitHub)" <gi...@apache.org>.
gortiz commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4958882684
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 `uint64` sizes rejected), UTF-8 uses `CodingErrorAction.REPORT` so corrupt bytes fail rather than silently substitute, canonical numbers are regex-validated before the permissive `parse*`, exact-fill + container-bounded nesting are enforced, and backward compatibility (unset → `TEXT`, `AUTO` opt-in) is preserved and tested. Nice test coverage too.
One blocker, plus a few smaller notes.
## 🔴 Blocker: unbounded recursion in `SqliteJsonbPayloadParser` → `StackOverflowError` takes the whole consuming segment to `ERROR`
`readElement` → `readArray`/`readObject` → `readElement` recurses one frame per nesting level with **no depth cap** (`SqliteJsonbPayloadParser.java:162,206-208,216-237`). Each nesting level costs only ~1–3 bytes on the wire (a `TYPE_ARRAY`/`TYPE_OBJECT` header plus its size field), so an ordinary stream message of a few KB — well under Kafka's default 1 MB `max.message.bytes` — produces thousands of stack frames and throws `StackOverflowError`.
The problem is not the throw itself, it's that `StackOverflowError` is an `Error`, not an `Exception`, so it escapes every decode-error guard on the path and is **not** handled as a bad record:
- `JSONMessageDecoder.decode` catches only `catch (Exception e)` (`JSONMessageDecoder.java:89`) — so it's *not* wrapped in the bounded `RuntimeException` this PR documents.
- `StreamDataDecoderImpl.decode` also catches only `catch (Exception e)` (`StreamDataDecoderImpl.java:96`) — so the normal "drop the row / stop-on-decode-error" handling in `RealtimeSegmentDataManager.processStreamEvents` never runs.
- The `Error` propagates out of `consumeLoop` into the consumer thread's `catch (Throwable e)` in `RealtimeSegmentDataManager.run()` (line ~1017), which sets `_state = State.ERROR`, drops `LLC_PARTITION_CONSUMING` to 0, and returns.
Net effect: **a single malformed `SQLITE_JSONB` message moves the entire consuming segment into `ERROR` state** and stops the partition, instead of dropping one row. Because the offset hasn't advanced past the message, it behaves as a poison pill requiring manual intervention. Every *other* malformed input in this parser correctly throws `IllegalArgumentException` (handled gracefully) — this recursion path is the one that escalates to segment failure, which contradicts the error-handling contract the rest of the parser upholds.
Reachable via pinned `SQLITE_JSONB` (a deeply-nested but otherwise valid, exactly-filling top-level object) and via `AUTO` (same payload passes `matches()`'s exact-fill check). The Jackson-backed Smile/CBOR paths are **not** affected — Jackson's `StreamReadConstraints` caps nesting depth (~1000) by default. Default `TEXT` is unaffected.
Suggested fix: cap recursion depth in `readElement` (mirroring Jackson's ~1000 default) and throw `IllegalArgumentException` past the limit, so it flows through the existing decode-error handling. A deep-nesting regression test should accompany it — the current tests only exercise shallow nesting (`testSqliteNested`, `testSqliteNestedElementCannotOverrunItsContainer`).
## 🟡 Smaller notes (non-blocking)
- **No cap on numeric-token length.** `parseInt` falls back to `new BigInteger(text)` for tokens that overflow `long` (`SqliteJsonbPayloadParser.java:250-257`), and `BigInteger(String)` is ~O(n²). A single huge numeric token (bounded only by the payload size, so up to ~1 MB of digits) is a per-message CPU cost with no analogue in Jackson, which caps number length at ~1000 chars via `StreamReadConstraints`. Worth a length bound for parity.
- **A bare `0x0C` decodes to an empty row** under `SQLITE_JSONB`/`AUTO` (`matches()` treats it as an exactly-filling empty OBJECT; `parse()` returns an empty map). Internally consistent and acknowledged by the "AUTO is a heuristic" framing, but under `AUTO` a stray `0x0C` byte becomes an empty record rather than a rejected one — worth a line in the README's AUTO caveats.
- **Trailing bytes after a valid Smile/CBOR object are ignored** (no `FAIL_ON_TRAILING_TOKENS`). This matches the existing text-JSON behavior via `JsonUtils.bytesToMap`, so it's not a regression — noting only for completeness.
The Postgres parser (version-byte strip + text-JSON body, with `matches()` requiring both the version byte and a JSON start char) looks correct to me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4959023781
@gortiz thanks for the careful read — the recursion blocker was a real hole. Addressed in `ed8144ac`.
**🔴 Blocker — unbounded recursion / `StackOverflowError` → segment ERROR.** Fixed. `readElement`/`readArray`/`readObject` now thread a depth counter and reject nesting past **1000** (Jackson `StreamReadConstraints`' default) with `IllegalArgumentException`, so deep nesting flows through the normal decode-error handling and drops the one bad record instead of escalating to a `StackOverflowError` that escapes `catch (Exception)` and takes the consuming segment to ERROR. Exactly your diagnosis — the `Error`-vs-`Exception` distinction is what made it a poison pill. Added `testSqliteRejectsExcessiveNestingDepth` (builds a 1200-deep exactly-filling payload, asserts `IllegalArgumentException`, not SOE; a 50-deep one still decodes), since the prior tests only exercised shallow nesting.
**🟡 Numeric-token length.** Fixed for parity. `BigInteger(String)` is ~O(n²) and was bounded only by payload size; numeric tokens are now capped at **1000** chars (matching Jackson's number-length limit) before the parse runs. Added `testSqliteRejectsOversizedNumericToken` (1000 digits OK → `BigInteger`, 1001 rejected).
**🟡 Bare `0x0C` → empty row under `AUTO`.** Documented. Added a line to the README's AUTO caveats noting a stray `0x0C` is a valid exactly-filling empty object and decodes to an empty row under `AUTO` — pin `TEXT`/your format to avoid it.
**🟡 Trailing bytes after Smile/CBOR.** No change, as you noted — it matches the existing text-JSON behavior via `JsonUtils.bytesToMap` (no `FAIL_ON_TRAILING_TOKENS`), so tightening only Smile/CBOR would make the binary paths *stricter* than text and inconsistent. Happy to add `FAIL_ON_TRAILING_TOKENS` across all Jackson-backed formats (text included) as a follow-up if you'd prefer that direction.
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](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556958596
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,383 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ // Top-level element must be an OBJECT (low nibble == 12) to produce a row. This never collides with text
+ // JSON, whose first byte is '{' (0x7B) or '[' (0x5B) -- both low nibble 0x0B (ARRAY), never 0x0C.
+ return length >= 1 && (payload[offset] & 0x0F) == TYPE_OBJECT;
+ }
+
+ @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);
Review Comment:
Fixed in `4b7401f` — `parse()` enforces SQLite's exact-fill rule (rejects a short-declared top-level element and trailing bytes), with regression tests at the parser and `JSONMessageDecoder` levels. `68af742` additionally moves the same rule into `matches()`, so AUTO no longer claims such a payload at all.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556731370
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * 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; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON.
+ 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), returning [#AUTO] when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
Review Comment:
Defaulting unset `jsonFormat` to AUTO makes the weak SQLite detector active for every existing JSON stream. `SqliteJsonbPayloadParser.matches()` claims any payload whose first byte has the OBJECT low nibble, and the parser treats a lone `0x0c` as a valid empty object, so a corrupt/non-JSON message that previously failed text JSON decoding can now be ingested as an empty row. That is a backward-compat/data-correctness change for existing deployments. Please keep the unset default on TEXT and require explicit `AUTO`/`SQLITE_JSONB`, or make SQLite auto-detection opt-in behind a stronger validity gate.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556309771
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,383 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ // Top-level element must be an OBJECT (low nibble == 12) to produce a row. This never collides with text
+ // JSON, whose first byte is '{' (0x7B) or '[' (0x5B) -- both low nibble 0x0B (ARRAY), never 0x0C.
+ return length >= 1 && (payload[offset] & 0x0F) == TYPE_OBJECT;
+ }
+
+ @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);
Review Comment:
Confirmed and fixed in 4b7401f — good catch, this was a silent-corruption bug.
Reproduced it exactly as you described: `0x0C` (OBJECT, size descriptor 0 → payload size 0) followed by real data parsed to `{}` and dropped the trailing bytes. Because `matches()` claims any payload whose first byte carries the OBJECT nibble, AUTO made it reachable for arbitrary input, so a corrupt stream message became an empty ingested row rather than a rejected one.
`parse()` now enforces SQLite's exact-fill rule and reports how much was consumed:
```java
if (cursor._pos != limit) {
throw new IllegalArgumentException(
"Top-level SQLite JSONB element consumed " + (cursor._pos - offset) + " of " + length + " bytes");
}
```
Regression tests at both levels:
- `testSqliteTopLevelElementMustExactlyFillThePayload` — the bare-`0x0C`-then-data case, a well-formed `{"a":1}` with one stray trailing byte, and that a lone `0x0C` (legitimately empty object, consumes the whole payload) still parses to `{}`.
- `testSqliteTrailingBytesAreRejectedRatherThanIngestedAsAPartialRow` — end-to-end through `JSONMessageDecoder`, via both AUTO and a pinned `SQLITE_JSONB`, asserting the message fails instead of yielding a partial row.
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java:
##########
@@ -20,34 +20,50 @@
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}.
+ *
+ * <p>When unset (equivalently {@code 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 {@code {} / {@code [} (optionally after whitespace) collides with none of the
+ * binary signatures. Pin {@code TEXT} to skip detection entirely. See {@link JsonPayloadFormat}.
Review Comment:
Fixed in 4b7401f. You're right that `{@code {}` terminates the inline tag at the first `}`. Switched to HTML entities so it renders and passes doclint:
```
* text JSON document: a top-level <code>{</code> or <code>[</code> (optionally after whitespace) collides
```
Verified with `mvn javadoc:javadoc` on the module — no errors.
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * 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; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON.
+ 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), returning [#AUTO] when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
+ }
+ 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");
+ }
Review Comment:
Fixed in 4b7401f — now `throw new IllegalArgumentException("Unsupported jsonFormat ...", e)`, preserving the `valueOf` failure as the cause.
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java:
##########
@@ -0,0 +1,172 @@
+/**
+ * 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.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;
+
+
+/// 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");
+
+ 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;
+ }
+
+ // 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);
+ }
+
+ @Test
+ public void testDefaultIsText()
+ throws Exception {
+ assertRich(decode(Map.of(), RICH_FIELDS, "{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes()));
+ }
Review Comment:
Fixed in 4b7401f. Renamed to `testUnsetFormatAutoDetectsText`, with a doc comment making the intent explicit: the default is AUTO (per-message detection), and text JSON must still decode through it.
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java:
##########
@@ -0,0 +1,422 @@
+/**
+ * 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.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+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;
+ }
+
+ /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text JSON body.
+ private static byte[] postgres(String json) {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ 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() {
+ assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT);
+ 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);
+ assertThrows(IllegalArgumentException.class, () -> JsonPayloadFormat.fromConfig("bson"));
+ }
+
+ // ------------------------------------------------------------------------------------------------------
+ // Text
+ // ------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testText()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser();
+ Map<String, Object> map = parse(parser, " {\"a\": 1, \"b\": \"x\"}".getBytes());
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertTrue(parser.matches("{\"a\":1}".getBytes(), 0, 7));
+ assertTrue(parser.matches(" [1,2]".getBytes(), 0, 7));
Review Comment:
Fixed in 4b7401f. Added a `utf8(String)` helper and routed every fixture through it; `postgres(...)` now builds on it too. Did the same in `JSONMessageDecoderBinaryTest` via a `TEXT_DOC` constant using `StandardCharsets.UTF_8`.
(Left the pre-existing `line.getBytes()` in `JSONMessageDecoderTest` alone — untouched by this PR.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #18953:
URL: https://github.com/apache/pinot/pull/18953
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4971387186
Opened the pinot-docs follow-up for this change:
- https://github.com/pinot-contrib/pinot-docs/pull/913
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4930967043
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18953?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `0%` with `220 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 56.86%. Comparing base ([`9ceca0f`](https://app.codecov.io/gh/apache/pinot/commit/9ceca0f18e99894eaf09eceabbbec561eaebbf24?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3d89565`](https://app.codecov.io/gh/apache/pinot/commit/3d895655a90c29a3265c74240ca9009a20ab3e52?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 24 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18953?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...utformat/json/format/SqliteJsonbPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FSqliteJsonbPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L1NxbGl0ZUpzb25iUGF5bG9hZFBhcnNlci5qYXZh) | 0.00% | [168 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...gin/inputformat/json/format/JsonPayloadFormat.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FJsonPayloadFormat.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L0pzb25QYXlsb2FkRm9ybWF0LmphdmE=) | 0.00% | [21 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...inputformat/json/format/TextJsonPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FTextJsonPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L1RleHRKc29uUGF5bG9hZFBhcnNlci5qYXZh) | 0.00% | [9 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...format/json/format/PostgresJsonbPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FPostgresJsonbPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L1Bvc3RncmVzSnNvbmJQYXlsb2FkUGFyc2VyLmphdmE=) | 0.00% | [5 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...ot/plugin/inputformat/json/JSONMessageDecoder.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2FJSONMessageDecoder.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vSlNPTk1lc3NhZ2VEZWNvZGVyLmphdmE=) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../inputformat/json/format/JacksonPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FJacksonPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L0phY2tzb25QYXlsb2FkUGFyc2VyLmphdmE=) | 0.00% | [4 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...putformat/json/format/AutoDetectPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FAutoDetectPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L0F1dG9EZXRlY3RQYXlsb2FkUGFyc2VyLmphdmE=) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...inputformat/json/format/CborJsonPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FCborJsonPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L0Nib3JKc29uUGF5bG9hZFBhcnNlci5qYXZh) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...nputformat/json/format/SmileJsonPayloadParser.java](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&filepath=pinot-plugins%2Fpinot-input-format%2Fpinot-json%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fplugin%2Finputformat%2Fjson%2Fformat%2FSmileJsonPayloadParser.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtcGx1Z2lucy9waW5vdC1pbnB1dC1mb3JtYXQvcGlub3QtanNvbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvcGx1Z2luL2lucHV0Zm9ybWF0L2pzb24vZm9ybWF0L1NtaWxlSnNvblBheWxvYWRQYXJzZXIuamF2YQ==) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18953?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (9ceca0f) and HEAD (3d89565). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (9ceca0f) | HEAD (3d89565) |
>|------|------|------|
>|java-21|5|1|
>|unittests|2|1|
>|temurin|5|1|
>|unittests2|1|0|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #18953 +/- ##
============================================
- Coverage 64.81% 56.86% -7.95%
+ Complexity 1347 1 -1346
============================================
Files 3396 2613 -783
Lines 212503 153964 -58539
Branches 33484 25104 -8380
============================================
- Hits 137732 87557 -50175
+ Misses 63610 58938 -4672
+ Partials 11161 7469 -3692
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <0.00%> (-7.95%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <0.00%> (-7.95%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <0.00%> (-7.95%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.86% <0.00%> (+0.08%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18953/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/18953?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4931926095
All review comments are addressed. Summary of the two follow-up commits:
**`4b7401f` — review fixes**
- **@xiangfu0's SQLite exact-fill bug** (the substantive one). Reproduced: `0x0C` (OBJECT, payload size 0) followed by real data decoded to `{}` and silently dropped the trailing bytes. `matches()` claims any payload whose first byte carries the OBJECT nibble, so AUTO made this reachable for arbitrary input — a corrupt stream message became an empty ingested row. `parse()` now enforces SQLite's exact-fill rule, with regression tests at both the parser and `JSONMessageDecoder` levels (and confirming a lone `0x0C` empty object still parses).
- Malformed `{@code {}` Javadoc construct (verified with `mvn javadoc:javadoc`).
- `fromConfig()` now preserves the `valueOf` failure as the exception cause.
- `testDefaultIsText` → `testUnsetFormatAutoDetectsText` (the default is AUTO, not pinned TEXT).
- Test fixtures pinned to `StandardCharsets.UTF_8`.
- Coverage for previously untested SQLite branches: size descriptors 14/15 including oversized-`uint64` rejection, signed hex `INT5`, `FLOAT5` with a leading `+`, the `\u` escape, the `TEXT5` escape extensions, and the missing-value / non-text-label errors.
**`5289a94` — bounded, charset-safe decode diagnostic**
Also taking Copilot's low-confidence note on the exception message, which is a real problem now that payloads can be binary: `new String(payload, offset, length)` used the **platform default charset** and echoed the entire payload, so a large Smile/CBOR/JSONB message was dumped into the log as unbounded mojibake. It now renders at most the first 128 bytes — as UTF-8 when they contain no control characters other than whitespace, as hex otherwise — and always reports the true length. Text payloads keep their readable rendering; binary payloads become a short hex prefix.
Module is at 65 tests, all passing; spotless / checkstyle / license clean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556136517
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java:
##########
@@ -0,0 +1,383 @@
+/**
+ * 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.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+
+/// 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);
+
+ @Override
+ public boolean matches(byte[] payload, int offset, int length) {
+ // Top-level element must be an OBJECT (low nibble == 12) to produce a row. This never collides with text
+ // JSON, whose first byte is '{' (0x7B) or '[' (0x5B) -- both low nibble 0x0B (ARRAY), never 0x0C.
+ return length >= 1 && (payload[offset] & 0x0F) == TYPE_OBJECT;
+ }
+
+ @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);
Review Comment:
This parser accepts a top-level JSONB element even when it consumes only a prefix of the message. SQLite's JSONB validity rule requires the outer element to exactly fill the BLOB; without that check, AUTO can classify a payload like `0x0C ...` as SQLite JSONB and ingest it as an empty object, or accept an object whose declared size drops trailing key/value bytes. That silently produces partial rows instead of rejecting a corrupt stream message. Please reject when the cursor position after `readElement` is not `limit`, and add a trailing-bytes regression test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Support binary JSON payload formats (PostgreSQL jsonb, SQLite JSONB, Smile, CBOR) in the JSON stream decoder [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #18953:
URL: https://github.com/apache/pinot/pull/18953#issuecomment-4932370171
Two more commits, then CI green on `ed03fea` (13/13).
- **`32aa635`** — split the decode-failure diagnostic into a 512-byte text window and a 128-byte hex window. My first cut used one 128-byte cap, which 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 gave. Hex costs two characters per byte, so the two caps produce messages of comparable length.
- **`ed03fea`** — assert two behaviors that were claimed above but not actually tested. The `assertThrows` on `fromConfig` passed even before the cause was chained, and every `describePayload` fixture was pure ASCII or contained a control byte forcing the hex path, so the "bytes >= 0x80 still render as text" branch was never exercised. Both now have real assertions.
Final state: 66 tests in the module, all green; spotless / checkstyle / license clean; all 13 CI checks passing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org