diff --git a/LICENSE-binary b/LICENSE-binary index 8d688bb0936e..9430db9a15f7 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -213,7 +213,9 @@ com.dynatrace.hash4j:hash4j:0.30.0 com.fasterxml.jackson.core:jackson-annotations:2.21 com.fasterxml.jackson.core:jackson-core:2.21.1 com.fasterxml.jackson.core:jackson-databind:2.21.1 +com.fasterxml.jackson.dataformat:jackson-dataformat-cbor:2.21.1 com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.21.1 +com.fasterxml.jackson.dataformat:jackson-dataformat-smile:2.21.1 com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:2.21.1 com.fasterxml.jackson.datatype:jackson-datatype-guava:2.21.1 com.fasterxml.jackson.datatype:jackson-datatype-jdk8:2.21.1 diff --git a/pinot-plugins/pinot-input-format/pinot-json/README.md b/pinot-plugins/pinot-input-format/pinot-json/README.md new file mode 100644 index 000000000000..0893ef81fde3 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/README.md @@ -0,0 +1,170 @@ + + +# Pinot JSON input format + +This plugin reads JSON records into Pinot. It provides: + +- `JSONMessageDecoder` — a `StreamMessageDecoder` for real-time (streaming) ingestion. +- `JSONRecordReader` — a `RecordReader` for batch ingestion of newline-delimited JSON files. +- `JSONRecordExtractor` — the shared extractor that turns a parsed JSON object into a Pinot `GenericRow`. + +By default the stream decoder reads **UTF-8 text JSON**, exactly as it always has. It can additionally decode +several **binary** JSON encodings, selected with the `jsonFormat` decoder property described below. + +> The `jsonFormat` property applies to the **stream decoder** (`JSONMessageDecoder`) only. Batch ingestion via +> `JSONRecordReader` always reads text JSON. + +## `jsonFormat` decoder property + +Set the decoder property `jsonFormat` (full stream-config key `stream..decoder.prop.jsonFormat`) to one +of: + +| `jsonFormat` | Payload encoding | +|------------------|-------------------------------------------------------------------------| +| *(unset)* / `TEXT` | UTF-8 text JSON. The historical behavior; unchanged. | +| `POSTGRES_JSONB` | PostgreSQL `jsonb` binary wire format (version byte + text JSON). | +| `SQLITE_JSONB` | SQLite 3.45+ JSONB binary format. | +| `SMILE` | [Jackson Smile](https://github.com/FasterXML/smile-format-specification) binary JSON. | +| `CBOR` | [CBOR](https://www.rfc-editor.org/rfc/rfc8949.html) (RFC 8949). | +| `AUTO` | Detect the encoding per message from its leading bytes (opt-in). | + +The value is case-insensitive. An unrecognized value fails table creation with a message listing the supported +values. + +Every format decodes to the same value contract, so the same table config, schema, and transform functions +work regardless of the wire encoding. Numbers narrow to `INT` / `LONG` / `BIG_DECIMAL`, decimals to `DOUBLE`, +and objects/arrays to nested JSON — just as for text JSON. (The binary formats can additionally carry native +`float` and `byte[]` scalars, which Pinot converts to the target column type.) + +## Default behavior and `AUTO` + +An unset `jsonFormat` means `TEXT`, so **existing tables are completely unaffected** by this feature. + +`AUTO` is opt-in rather than the default. Detection is a heuristic over a few leading bytes; it never +mis-routes a well-formed text JSON document (a top-level `{` or `[`, optionally after whitespace, matches none +of the binary signatures), but it can claim a *corrupt* message that text decoding would otherwise have +rejected. Turning that on implicitly for every existing stream would be a silent behavior change, so you ask +for it explicitly. + +When `AUTO` is set, each message is classified in this order, falling back to text JSON: + +| Order | Format | Signature | +|-------|------------------|------------------------------------------------------------------------------| +| 1 | `SMILE` | 3-byte header `3A 29 0A` (`:)\n`). | +| 2 | `CBOR` | Self-describe tag `D9 D9 F7` (RFC 8949 §3.4.6). CBOR **without** the tag is not detected. | +| 3 | `POSTGRES_JSONB` | Version byte `01` followed by a JSON document start. | +| 4 | `SQLITE_JSONB` | A top-level object whose declared size exactly fills the payload. | +| 5 | `TEXT` | First non-whitespace byte is `{` or `[`. | + +Pin an explicit format instead of `AUTO` when you know the encoding: it skips detection and, for formats +without a strong magic number (notably tag-less CBOR), is the only way to decode them. + +The SQLite signature is intentionally narrow, but not zero-risk: a message consisting of the single byte +`0x0C` is a valid, exactly-filling empty SQLite object, so under `AUTO` a stray `0x0C` decodes to an empty row +rather than being rejected. This is the heuristic nature of `AUTO`; pin `TEXT` (or your actual format) to avoid +it entirely. + +## Examples + +All examples use a Kafka real-time table; substitute your stream type (`stream..…`) as needed. + +### 1. Text JSON (default — no configuration needed) + +```json +{ + "streamConfigMaps": [ + { + "streamType": "kafka", + "stream.kafka.topic.name": "events", + "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder", + "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory", + "stream.kafka.broker.list": "localhost:9092" + } + ] +} +``` + +### 2. A pinned binary format (SQLite JSONB) + +Add a single decoder property: + +```json +"stream.kafka.decoder.prop.jsonFormat": "SQLITE_JSONB" +``` + +Full block: + +```json +{ + "streamConfigMaps": [ + { + "streamType": "kafka", + "stream.kafka.topic.name": "events", + "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder", + "stream.kafka.decoder.prop.jsonFormat": "SQLITE_JSONB", + "stream.kafka.consumer.factory.class.name": "org.apache.pinot.plugin.stream.kafka30.KafkaConsumerFactory", + "stream.kafka.broker.list": "localhost:9092" + } + ] +} +``` + +Swap the value for `POSTGRES_JSONB`, `SMILE`, or `CBOR` to decode those encodings. + +### 3. Auto-detection for a mixed stream + +```json +"stream.kafka.decoder.prop.jsonFormat": "AUTO" +``` + +Decodes text JSON, Smile, CBOR (with the self-describe tag), PostgreSQL `jsonb`, and SQLite JSONB messages on +the same topic, one classification per message. + +## Format notes + +### PostgreSQL `jsonb` + +Despite `jsonb`'s compact on-disk layout, that layout never leaves the server. `jsonb_send` renders the value +as text and emits a version byte (`1`) followed by the UTF-8 JSON text; `jsonb_recv` reverses it. Every +standard binary producer — the v3 extended-query protocol, `COPY ... WITH (FORMAT binary)`, and logical +replication via `pgoutput` — goes through that same function. This parser strips the version byte and parses +the text body, so its value types are identical to `TEXT`. + +### SQLite JSONB + +Implements the [SQLite JSONB](https://sqlite.org/jsonb.html) format (SQLite 3.45+), including the JSON5 element +types (hex/`+`-prefixed integers, `Infinity`/`NaN` floats, and `\x`/`\'`/`\v` string escapes). Per SQLite's +validity rule, the top-level element must exactly fill the message; a payload that declares a shorter size is +rejected rather than decoded into a partial row. + +### Smile and CBOR + +Decoded with Jackson. Smile carries its 3-byte header by default (used for `AUTO` detection). CBOR has no +mandatory magic number, so `AUTO` recognizes it only when the optional `D9 D9 F7` self-describe tag is present; +otherwise pin `jsonFormat: CBOR`. + +## Diagnostics + +When a message cannot be decoded, the decoder throws with a bounded, log-safe description of the payload: a +leading window rendered as text when it looks like text and as hex otherwise, prefixed with the total byte +length. Binary payloads therefore appear as a short hex prefix rather than an unbounded dump of unreadable +bytes. diff --git a/pinot-plugins/pinot-input-format/pinot-json/pom.xml b/pinot-plugins/pinot-input-format/pinot-json/pom.xml index 306ae8611162..2bdaea245508 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/pom.xml +++ b/pinot-plugins/pinot-input-format/pinot-json/pom.xml @@ -35,6 +35,18 @@ package + + + + com.fasterxml.jackson.dataformat + jackson-dataformat-smile + + + com.fasterxml.jackson.dataformat + jackson-dataformat-cbor + + + pinot-fastdev diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java index 08a1122e6c4e..a83b6cfde05a 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java @@ -18,36 +18,61 @@ */ package org.apache.pinot.plugin.inputformat.json; +import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.Set; +import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadFormat; +import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadParser; import org.apache.pinot.spi.data.readers.GenericRow; import org.apache.pinot.spi.data.readers.RecordExtractor; import org.apache.pinot.spi.plugin.PluginManager; import org.apache.pinot.spi.stream.StreamMessageDecoder; -import org.apache.pinot.spi.utils.JsonUtils; /** * An implementation of StreamMessageDecoder to read JSON records from a stream. + * + *

Set the {@value #JSON_FORMAT_CONFIG_KEY} decoder property to pin the payload encoding to one of + * {@code TEXT}, {@code POSTGRES_JSONB}, {@code SQLITE_JSONB}, {@code SMILE} or {@code CBOR}. When unset the + * encoding is {@code TEXT}, the decoder's historical behavior. + * + *

Set it to {@code AUTO} to instead detect the encoding per message from its leading magic / version bytes, + * falling back to text JSON. Detection is allocation-free and cannot mis-route a well-formed text JSON + * document: a top-level { or [ (optionally after whitespace) collides with none + * of the binary signatures. It is opt-in rather than the default because it is still a heuristic over a few + * leading bytes, so it may claim a corrupt message that text decoding would have rejected outright. + * See {@link JsonPayloadFormat}. */ public class JSONMessageDecoder implements StreamMessageDecoder { + public static final String JSON_FORMAT_CONFIG_KEY = "jsonFormat"; + + /// Caps on the payload bytes echoed into a decode-failure message. Hex renders two characters per byte, so + /// the binary cap is a quarter of the text one to keep both messages a comparable length. + private static final int MAX_DIAGNOSTIC_TEXT_BYTES = 512; + private static final int MAX_DIAGNOSTIC_HEX_BYTES = 128; + private static final String JSON_RECORD_EXTRACTOR_CLASS = "org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor"; private RecordExtractor> _jsonRecordExtractor; + // For AUTO this resolves the concrete format per message; otherwise it is the pinned format's parser. + private JsonPayloadParser _parser; @Override public void init(Map props, Set fieldsToRead, String topicName) throws Exception { String recordExtractorClass = null; + String jsonFormat = null; if (props != null) { recordExtractorClass = props.get(RECORD_EXTRACTOR_CONFIG_KEY); + jsonFormat = props.get(JSON_FORMAT_CONFIG_KEY); } if (recordExtractorClass == null) { recordExtractorClass = JSON_RECORD_EXTRACTOR_CLASS; } _jsonRecordExtractor = PluginManager.get().createInstance(recordExtractorClass); _jsonRecordExtractor.init(fieldsToRead, null); + _parser = JsonPayloadFormat.fromConfig(jsonFormat).getParser(); } @Override @@ -58,12 +83,52 @@ public GenericRow decode(byte[] payload, GenericRow destination) { @Override public GenericRow decode(byte[] payload, int offset, int length, GenericRow destination) { try { - // Parse directly to Map, avoiding intermediate JsonNode representation for better performance - Map jsonMap = JsonUtils.bytesToMap(payload, offset, length); + // Parse directly to Map, avoiding an intermediate JsonNode representation for better performance. + Map jsonMap = _parser.parse(payload, offset, length); return _jsonRecordExtractor.extract(jsonMap, destination); } catch (Exception e) { throw new RuntimeException( - "Caught exception while decoding JSON record with payload: " + new String(payload, offset, length), e); + "Caught exception while decoding JSON record with payload: " + describePayload(payload, offset, length), e); + } + } + + /// Renders a bounded, log-safe description of a payload for an error message. + /// + /// The payload may now be any of the binary encodings, so it is neither necessarily text nor necessarily + /// small. A leading window is rendered as UTF-8 when it looks like text (no control characters other than + /// whitespace) and as hex otherwise, so a binary message cannot flood the log with megabytes of mojibake, + /// while a malformed text record still shows enough of itself to be identified. The charset is always + /// explicit rather than the platform default. + private static String describePayload(byte[] payload, int offset, int length) { + boolean text = looksLikeText(payload, offset, Math.min(length, MAX_DIAGNOSTIC_TEXT_BYTES)); + int previewLength = Math.min(length, text ? MAX_DIAGNOSTIC_TEXT_BYTES : MAX_DIAGNOSTIC_HEX_BYTES); + StringBuilder description = new StringBuilder(32 + (text ? previewLength : 2 * previewLength)); + description.append(length).append(" bytes: "); + if (text) { + // Truncation may split a multi-byte character; UTF-8 decoding substitutes U+FFFD rather than throwing. + description.append(new String(payload, offset, previewLength, StandardCharsets.UTF_8)); + } else { + description.append("0x"); + for (int i = 0; i < previewLength; i++) { + description.append(Character.forDigit((payload[offset + i] >> 4) & 0xF, 16)) + .append(Character.forDigit(payload[offset + i] & 0xF, 16)); + } + } + if (previewLength < length) { + description.append("...(truncated)"); + } + return description.toString(); + } + + /// Whether the region contains no control characters other than JSON's insignificant whitespace. Bytes at or + /// above 0x80 are accepted so non-ASCII UTF-8 text still renders as text. + private static boolean looksLikeText(byte[] payload, int offset, int length) { + for (int i = offset, end = offset + length; i < end; i++) { + byte b = payload[i]; + if (b >= 0 && b < 0x20 && b != '\t' && b != '\n' && b != '\r') { + return false; + } } + return true; } } diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java new file mode 100644 index 000000000000..03813aebb254 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/AutoDetectPayloadParser.java @@ -0,0 +1,40 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import java.util.Map; + + +/// Resolves the concrete parser per message from the payload's leading magic / version bytes, then delegates. +/// Backs [JsonPayloadFormat#AUTO], so the decoder can always call `parse` without a null / mode check. +class AutoDetectPayloadParser implements JsonPayloadParser { + + /// Always {@code true}: this parser accepts any payload and resolves the real format at [#parse] time. It is + /// deliberately excluded from the detection order it drives. + @Override + public boolean matches(byte[] payload, int offset, int length) { + return true; + } + + @Override + public Map parse(byte[] payload, int offset, int length) + throws Exception { + return JsonPayloadFormat.detectParser(payload, offset, length).parse(payload, offset, length); + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java new file mode 100644 index 000000000000..3ed57b00a900 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/CborJsonPayloadParser.java @@ -0,0 +1,49 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import com.fasterxml.jackson.dataformat.cbor.CBORFactory; + + +/// Parses CBOR (RFC 8949) via Jackson. Values decode +/// to the same Java types Jackson produces for text JSON (CBOR may additionally emit `Float` and `byte[]` +/// scalars, which Pinot handles downstream). +/// +/// CBOR has no mandatory magic number, so AUTO detection only recognizes payloads that carry the optional +/// self-describe tag `0xD9D9F7` (RFC 8949 §3.4.6). CBOR streams without that tag must select the format +/// explicitly. +class CborJsonPayloadParser extends JacksonPayloadParser { + + // CBOR "Self-Described CBOR" tag 55799, encoded as the 3-byte prefix 0xD9 0xD9 0xF7 (RFC 8949 §3.4.6). + private static final int SELF_DESCRIBE_0 = 0xD9; + private static final int SELF_DESCRIBE_1 = 0xD9; + private static final int SELF_DESCRIBE_2 = 0xF7; + + CborJsonPayloadParser() { + super(new CBORFactory()); + } + + @Override + public boolean matches(byte[] payload, int offset, int length) { + return length >= 3 + && (payload[offset] & 0xFF) == SELF_DESCRIBE_0 + && (payload[offset + 1] & 0xFF) == SELF_DESCRIBE_1 + && (payload[offset + 2] & 0xFF) == SELF_DESCRIBE_2; + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java new file mode 100644 index 000000000000..711c55e2bec8 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JacksonPayloadParser.java @@ -0,0 +1,45 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.ObjectReader; +import java.util.Map; +import org.apache.pinot.spi.utils.JsonUtils; + + +/// Base for formats that Jackson can read directly given a binary [JsonFactory] (Smile, CBOR). Subclasses only +/// supply the factory and their magic-byte [#matches] check. +/// +/// The [ObjectReader] is immutable and thread-safe, so a single instance is shared across all decode calls. +abstract class JacksonPayloadParser implements JsonPayloadParser { + + private final ObjectReader _mapReader; + + JacksonPayloadParser(JsonFactory factory) { + _mapReader = new ObjectMapper(factory).readerFor(JsonUtils.MAP_TYPE_REFERENCE); + } + + @Override + public Map parse(byte[] payload, int offset, int length) + throws Exception { + return _mapReader.readValue(payload, offset, length); + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java new file mode 100644 index 000000000000..9e17c79cba5e --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java @@ -0,0 +1,89 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import java.util.Locale; +import javax.annotation.Nullable; + + +/// The payload encodings the JSON stream decoder understands, selected through the `jsonFormat` decoder +/// property. +/// +/// An unset property means [#TEXT], preserving the decoder's historical behavior exactly. [#AUTO] is opt-in: +/// detection is a heuristic over a few leading bytes, and while it never mis-routes a well-formed text JSON +/// document, it can claim a corrupt message that text decoding would have rejected outright. Turning that on +/// for every existing stream would be a silent correctness change, so operators ask for it explicitly. +public enum JsonPayloadFormat { + /// Detect the encoding per message from its leading magic / version bytes, falling back to text JSON. + /// Opt-in; see the note on this enum. + AUTO(new AutoDetectPayloadParser()), + /// UTF-8 text JSON (the historical default). + TEXT(new TextJsonPayloadParser()), + /// PostgreSQL `jsonb` binary wire format (version byte + text JSON). + POSTGRES_JSONB(new PostgresJsonbPayloadParser()), + /// SQLite 3.45+ JSONB binary format. + SQLITE_JSONB(new SqliteJsonbPayloadParser()), + /// Jackson Smile binary JSON. + SMILE(new SmileJsonPayloadParser()), + /// CBOR (RFC 8949). + CBOR(new CborJsonPayloadParser()); + + // AUTO detection precedence: formats with strong, unambiguous magic bytes first; text (the catch-all, + // detected only by a leading '{' / '[') last so it never shadows a binary format. POSTGRES_JSONB precedes + // TEXT because its body *is* text JSON behind a version byte. + private static final JsonPayloadFormat[] DETECTION_ORDER = + {SMILE, CBOR, POSTGRES_JSONB, SQLITE_JSONB, TEXT}; + + private final JsonPayloadParser _parser; + + JsonPayloadFormat(JsonPayloadParser parser) { + _parser = parser; + } + + /// The parser for this format. For [#AUTO] this is a parser that resolves the concrete format per message. + public JsonPayloadParser getParser() { + return _parser; + } + + /// Resolves a configured format name (case-insensitive). An unset value means [#TEXT], the decoder's + /// historical behavior; [#AUTO] must be requested explicitly. + public static JsonPayloadFormat fromConfig(@Nullable String value) { + if (value == null || value.trim().isEmpty()) { + return TEXT; + } + try { + return valueOf(value.trim().toUpperCase(Locale.ROOT)); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException( + "Unsupported jsonFormat '" + value + "'. Supported values: AUTO, TEXT, POSTGRES_JSONB, SQLITE_JSONB, " + + "SMILE, CBOR", e); + } + } + + /// Picks a parser for a payload by consulting each format's magic / version bytes in [#DETECTION_ORDER], + /// defaulting to text JSON when nothing matches. + public static JsonPayloadParser detectParser(byte[] payload, int offset, int length) { + for (JsonPayloadFormat format : DETECTION_ORDER) { + if (format._parser.matches(payload, offset, length)) { + return format._parser; + } + } + return TEXT._parser; + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java new file mode 100644 index 000000000000..90b353cceb31 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadParser.java @@ -0,0 +1,54 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import java.util.Map; + + +/// Parses a stream payload encoded in a particular (text or binary) JSON representation into a Jackson-style +/// `Map` so the shared +/// [org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor] can turn it into a +/// [org.apache.pinot.spi.data.readers.GenericRow]. +/// +/// Implementations must produce the same Java value contract Jackson produces for text JSON, so the extractor +/// can treat every format uniformly: `Boolean`, `Integer`, `Long`, `Double`, `String`, `java.math.BigInteger` +/// / `java.math.BigDecimal` for oversized / high-precision numbers, `java.util.List` for arrays, and nested +/// `Map` for objects. Binary formats may additionally yield `Float` and `byte[]` scalars, +/// which Pinot's downstream type conversion handles. +/// +/// Implementations are stateless and must be thread-safe: a single instance is shared across all decode calls. +public interface JsonPayloadParser { + + /// Cheap, allocation-free check of the payload's leading bytes to decide whether this parser recognizes the + /// encoding. Used only by [JsonPayloadFormat#AUTO] detection; when a format is configured explicitly the + /// corresponding parser is used without consulting this method. + /// + /// @param payload backing byte array + /// @param offset start offset of the record within {@code payload} + /// @param length number of bytes belonging to the record + /// @return {@code true} if the leading bytes match this format's magic / version signature + boolean matches(byte[] payload, int offset, int length); + + /// Parses the {@code [offset, offset + length)} region of {@code payload} into a mutable + /// `Map` following the value contract described on this interface. + /// + /// @throws Exception if the region is not valid for this format + Map parse(byte[] payload, int offset, int length) + throws Exception; +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java new file mode 100644 index 000000000000..2a6a5b4c9af4 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/PostgresJsonbPayloadParser.java @@ -0,0 +1,58 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import java.util.Map; +import org.apache.pinot.spi.utils.JsonUtils; + + +/// Parses the PostgreSQL `jsonb` binary wire format: a single version byte (currently `1`) followed by the +/// value's **UTF-8 text** JSON representation. +/// +/// Despite `jsonb`'s compact on-disk `JsonbContainer` layout, that layout never leaves the server. PostgreSQL's +/// `jsonb_send` (`src/backend/utils/adt/jsonb.c`) renders the value with `JsonbToCString` and emits +/// `pq_sendint8(version = 1)` followed by the text; `jsonb_recv` reverses it and rejects any other version. +/// Every standard binary producer — the v3 extended-query protocol, `COPY ... WITH (FORMAT binary)`, and +/// logical replication via `pgoutput` — routes through that same send function, so the version-byte + text +/// framing is what a stream actually carries. +/// +/// Because the body is ordinary text JSON, values decode through [JsonUtils#bytesToMap] and therefore follow +/// exactly the same type contract as [TextJsonPayloadParser]. +class PostgresJsonbPayloadParser implements JsonPayloadParser { + + /// The only version `jsonb_recv` accepts. + private static final byte JSONB_VERSION = 1; + + @Override + public boolean matches(byte[] payload, int offset, int length) { + // Version byte followed by a text JSON document. Requiring the JSON start character keeps this from + // claiming arbitrary binary payloads that merely happen to begin with 0x01. + return length >= 2 && payload[offset] == JSONB_VERSION && TextJsonPayloadParser.startsWithJsonDocument(payload, + offset + 1, length - 1); + } + + @Override + public Map parse(byte[] payload, int offset, int length) + throws Exception { + if (length < 2 || payload[offset] != JSONB_VERSION) { + throw new IllegalArgumentException("Payload is not a version-1 PostgreSQL jsonb value"); + } + return JsonUtils.bytesToMap(payload, offset + 1, length - 1); + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java new file mode 100644 index 000000000000..eddfe470a05e --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SmileJsonPayloadParser.java @@ -0,0 +1,44 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import com.fasterxml.jackson.dataformat.smile.SmileConstants; +import com.fasterxml.jackson.dataformat.smile.SmileFactory; + + +/// Parses Smile, Jackson's own binary +/// JSON encoding. Values decode to the same Java types Jackson produces for text JSON (Smile may additionally +/// emit `Float` and `byte[]` scalars, which Pinot handles downstream). +/// +/// Detection relies on the 3-byte Smile header `:)\n` (`0x3A 0x29 0x0A`) that Jackson writes by default; +/// header-less Smile payloads must select the format explicitly rather than via AUTO detection. +class SmileJsonPayloadParser extends JacksonPayloadParser { + + SmileJsonPayloadParser() { + super(new SmileFactory()); + } + + @Override + public boolean matches(byte[] payload, int offset, int length) { + return length >= 3 + && payload[offset] == SmileConstants.HEADER_BYTE_1 + && payload[offset + 1] == SmileConstants.HEADER_BYTE_2 + && payload[offset + 2] == SmileConstants.HEADER_BYTE_3; + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java new file mode 100644 index 000000000000..46041cc8ee55 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/SqliteJsonbPayloadParser.java @@ -0,0 +1,508 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import com.google.common.collect.Maps; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.CharsetDecoder; +import java.nio.charset.CodingErrorAction; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; + + +/// Parses the SQLite JSONB binary format (SQLite 3.45+). +/// +/// Every element is a 1–9 byte header followed by a payload. The header's first byte packs the element type in +/// the low nibble and a payload-size descriptor in the high nibble; descriptors 12–15 mean the size is a big- +/// endian `uint8` / `uint16` / `uint32` / `uint64` in the following bytes, and 0–11 are the size itself. +/// +/// Numbers are stored as their ASCII text, so integers narrow to `Integer` / `Long` / `BigInteger` and floats +/// to `Double`, matching the text-JSON value contract. `TEXTJ` strings/labels are JSON-unescaped (invalid +/// escapes are rejected); `TEXT5` additionally accepts JSON5 escapes. +/// +/// The decoder expects one row per payload, so the top-level element must be an `OBJECT`. +class SqliteJsonbPayloadParser implements JsonPayloadParser { + + // Element types (low nibble of the header's first byte). + private static final int TYPE_NULL = 0; + private static final int TYPE_TRUE = 1; + private static final int TYPE_FALSE = 2; + private static final int TYPE_INT = 3; // canonical decimal integer, ASCII + private static final int TYPE_INT5 = 4; // JSON5 integer (hex / leading sign), ASCII + private static final int TYPE_FLOAT = 5; // canonical float, ASCII + private static final int TYPE_FLOAT5 = 6; // JSON5 float, ASCII + private static final int TYPE_TEXT = 7; // raw text, no escapes + private static final int TYPE_TEXTJ = 8; // text with JSON escapes + private static final int TYPE_TEXT5 = 9; // text with JSON5 escapes + private static final int TYPE_TEXTRAW = 10; // raw text needing quoting, no escapes + private static final int TYPE_ARRAY = 11; + private static final int TYPE_OBJECT = 12; + + private static final BigInteger INT_MIN = BigInteger.valueOf(Integer.MIN_VALUE); + private static final BigInteger INT_MAX = BigInteger.valueOf(Integer.MAX_VALUE); + private static final BigInteger LONG_MIN = BigInteger.valueOf(Long.MIN_VALUE); + private static final BigInteger LONG_MAX = BigInteger.valueOf(Long.MAX_VALUE); + + // RFC 8259 number grammar for the canonical TYPE_INT / TYPE_FLOAT forms: an optional minus, an integer part + // with no leading zeros, and (for floats) an optional fraction and exponent. No leading '+', NaN, Infinity or + // hex -- the tokens Double.parseDouble / Long.parseLong would otherwise accept. + private static final Pattern CANONICAL_INT = Pattern.compile("-?(?:0|[1-9][0-9]*)"); + private static final Pattern CANONICAL_NUMBER = + Pattern.compile("-?(?:0|[1-9][0-9]*)(?:\\.[0-9]+)?(?:[eE][+-]?[0-9]+)?"); + + // Bound nesting depth and numeric-token length to Jackson's StreamReadConstraints defaults, so this binary + // path rejects the same pathological shapes text JSON already rejects via JsonUtils.bytesToMap. The depth cap + // is the important one: each nesting level is only ~1-3 wire bytes, so without it a small message can recurse + // deep enough to throw StackOverflowError -- an Error, not an Exception, that escapes the decode-error + // handling and fails the whole consuming segment instead of dropping one bad record. + private static final int MAX_NESTING_DEPTH = 1000; + private static final int MAX_NUMBER_LENGTH = 1000; + + @Override + public boolean matches(byte[] payload, int offset, int length) { + // The top-level element must be an OBJECT (low nibble == 12) to produce a row. That never collides with + // text JSON, whose first byte is '{' (0x7B) or '[' (0x5B) -- both low nibble 0x0B (ARRAY), never 0x0C. + if (length < 1 || (payload[offset] & 0x0F) != TYPE_OBJECT) { + return false; + } + // The nibble alone is a weak signal: one byte in sixteen of arbitrary binary would claim the payload. Also + // require the object's declared size to exactly fill the payload -- the same validity rule parse() enforces + // -- so AUTO cannot hand a corrupt message to this parser on the strength of a single nibble. + int sizeDescriptor = (payload[offset] & 0xFF) >>> 4; + int headerLength = headerLength(sizeDescriptor); + if (length < headerLength) { + return false; + } + return declaredSize(payload, offset, sizeDescriptor) == (long) length - headerLength; + } + + /// Bytes occupied by an element header: the type/size-descriptor byte plus the width of any explicit size + /// field. Descriptors 0-11 encode the size in the byte itself; 12-15 prepend a 1-, 2-, 4- or 8-byte + /// big-endian size. + /// + /// Shared by [#matches] and [#readElement] so detection and parsing can never disagree about the layout. + private static int headerLength(int sizeDescriptor) { + switch (sizeDescriptor) { + case 12: + return 2; + case 13: + return 3; + case 14: + return 5; + case 15: + return 9; + default: + return 1; + } + } + + /// Payload size declared by the element header starting at {@code offset}. The caller must first ensure + /// [#headerLength] bytes are available. A descriptor-15 size with its sign bit set yields a negative value, + /// which every caller rejects (it can neither be a valid length nor equal a payload length). + private static long declaredSize(byte[] payload, int offset, int sizeDescriptor) { + switch (sizeDescriptor) { + case 12: + return readUnsignedBE(payload, offset + 1, 1); + case 13: + return readUnsignedBE(payload, offset + 1, 2); + case 14: + return readUnsignedBE(payload, offset + 1, 4); + case 15: + return readUnsignedBE(payload, offset + 1, 8); + default: + return sizeDescriptor; + } + } + + /// Reads {@code count} big-endian bytes as an unsigned value. An 8-byte field with the sign bit set yields a + /// negative long, which never equals a payload length and is rejected by the bounds checks. + private static long readUnsignedBE(byte[] payload, int offset, int count) { + long value = 0; + for (int i = 0; i < count; i++) { + value = (value << 8) | (payload[offset + i] & 0xFFL); + } + return value; + } + + @Override + public Map parse(byte[] payload, int offset, int length) { + int limit = offset + length; + Cursor cursor = new Cursor(payload, offset, limit); + Object value = readElement(cursor, limit, 0); + if (!(value instanceof Map)) { + throw new IllegalArgumentException("Top-level SQLite JSONB element must be an object"); + } + // SQLite's validity rule: the outer element must exactly fill the BLOB. Without this, a payload whose + // top-level element declares a short size (e.g. a bare 0x0C followed by data) would decode to a partial -- + // possibly empty -- row and silently discard the trailing bytes instead of rejecting a corrupt message. + if (cursor._pos != limit) { + throw new IllegalArgumentException( + "Top-level SQLite JSONB element consumed " + (cursor._pos - offset) + " of " + length + " bytes"); + } + //noinspection unchecked + return (Map) value; + } + + /// Reads one element. {@code parentEnd} is the exclusive end of the enclosing container (the whole payload at + /// the top level); an element may never extend past it, otherwise a nested element could overrun its parent + /// while still fitting the payload and silently swallow its following siblings. + private static Object readElement(Cursor cursor, int parentEnd, int depth) { + int header = cursor.peekUInt8(); + int type = header & 0x0F; + int sizeDescriptor = header >>> 4; + int headerStart = cursor._pos; + // Decode the header through the same helpers matches() uses, so detection and parsing cannot drift apart. + cursor.skip(headerLength(sizeDescriptor)); + long payloadSize = declaredSize(cursor._buf, headerStart, sizeDescriptor); + int start = cursor._pos; + int end = cursor.boundedEnd(start, payloadSize, parentEnd); + Object result; + switch (type) { + case TYPE_NULL: + result = null; + break; + case TYPE_TRUE: + result = Boolean.TRUE; + break; + case TYPE_FALSE: + result = Boolean.FALSE; + break; + case TYPE_INT: + result = parseCanonicalInt(cursor.ascii(start, end)); + break; + case TYPE_INT5: + result = parseInt5(cursor.ascii(start, end)); + break; + case TYPE_FLOAT: + result = parseCanonicalFloat(cursor.ascii(start, end)); + break; + case TYPE_FLOAT5: + result = parseJson5Float(cursor.ascii(start, end)); + break; + case TYPE_TEXT: + case TYPE_TEXTRAW: + result = cursor.utf8(start, end); + break; + case TYPE_TEXTJ: + result = unescape(cursor.utf8(start, end), false); + break; + case TYPE_TEXT5: + result = unescape(cursor.utf8(start, end), true); + break; + case TYPE_ARRAY: + return readArray(cursor, end, depth); + case TYPE_OBJECT: + return readObject(cursor, end, depth); + default: + throw new IllegalArgumentException("Reserved/invalid SQLite JSONB element type: " + type); + } + cursor._pos = end; + return result; + } + + private static List readArray(Cursor cursor, int end, int depth) { + int childDepth = nextDepth(depth); + List list = new ArrayList<>(); + while (cursor._pos < end) { + list.add(readElement(cursor, end, childDepth)); + } + return list; + } + + private static Map readObject(Cursor cursor, int end, int depth) { + int childDepth = nextDepth(depth); + Map map = Maps.newHashMapWithExpectedSize(4); + while (cursor._pos < end) { + Object label = readElement(cursor, end, childDepth); + if (!(label instanceof String)) { + throw new IllegalArgumentException("SQLite JSONB object label must be text"); + } + if (cursor._pos >= end) { + throw new IllegalArgumentException("SQLite JSONB object is missing a value for label: " + label); + } + map.put((String) label, readElement(cursor, end, childDepth)); + } + return map; + } + + /// Returns {@code depth + 1} for the children of a container, throwing once nesting would exceed + /// [#MAX_NESTING_DEPTH]. Keeps deep nesting on the `IllegalArgumentException` path (handled as a bad record) + /// rather than letting the recursion overflow the stack with a `StackOverflowError`. + private static int nextDepth(int depth) { + if (depth >= MAX_NESTING_DEPTH) { + throw new IllegalArgumentException("SQLite JSONB nesting exceeds the maximum depth of " + MAX_NESTING_DEPTH); + } + return depth + 1; + } + + /// Parses a canonical (RFC 8259) integer, the SQLite `TYPE_INT` form. Validated as strictly as text JSON: + /// the permissive tokens `Long.parseLong` / `BigInteger` accept -- a leading `+` or leading zeros -- are + /// rejected, so a malformed `TYPE_INT` fails the record rather than decoding to a value Jackson would refuse. + /// The JSON5 forms (hex, leading sign) live in [#parseInt5]. + private static Object parseCanonicalInt(String text) { + if (!CANONICAL_INT.matcher(text).matches()) { + throw new IllegalArgumentException("Invalid canonical SQLite JSONB integer: " + text); + } + return parseInt(text, 10); + } + + private static Object parseInt(String text, int radix) { + // Fast path: values that fit a long (the overwhelming majority) avoid BigInteger allocation. + try { + return narrowLong(Long.parseLong(text, radix)); + } catch (NumberFormatException e) { + return narrow(new BigInteger(text, radix)); + } + } + + /// Narrows an integer to the smallest of `Integer` / `Long` / `BigInteger` that holds it, mirroring the + /// Jackson text-JSON contract (`BigInteger` is later widened to `BigDecimal` by the record extractor). + private static Object narrow(BigInteger value) { + if (value.compareTo(INT_MIN) >= 0 && value.compareTo(INT_MAX) <= 0) { + return value.intValue(); + } + if (value.compareTo(LONG_MIN) >= 0 && value.compareTo(LONG_MAX) <= 0) { + return value.longValue(); + } + return value; + } + + private static Object parseInt5(String text) { + String digits = text; + boolean negative = false; + if (!digits.isEmpty() && (digits.charAt(0) == '+' || digits.charAt(0) == '-')) { + negative = digits.charAt(0) == '-'; + digits = digits.substring(1); + } + Object magnitude; + if (digits.length() > 2 && digits.charAt(0) == '0' && (digits.charAt(1) == 'x' || digits.charAt(1) == 'X')) { + magnitude = parseInt(digits.substring(2), 16); + } else { + magnitude = parseInt(digits, 10); + } + if (!negative) { + return magnitude; + } + return magnitude instanceof BigInteger ? narrow(((BigInteger) magnitude).negate()) + : narrowLong(-((Number) magnitude).longValue()); + } + + private static Object narrowLong(long value) { + // NOTE: keep these as separate returns. A `cond ? (int) value : value` ternary would numerically promote + // the int branch to long, always boxing to Long. + if (value >= Integer.MIN_VALUE && value <= Integer.MAX_VALUE) { + return (int) value; + } + return value; + } + + /// Parses a canonical (RFC 8259) float, the SQLite `TYPE_FLOAT` form. `Double.parseDouble` is far more + /// permissive than JSON -- it accepts `NaN`, `Infinity`, a leading `+`, Java hex floats and type suffixes -- + /// so the text is first validated against the JSON number grammar. That rejects the non-finite and other + /// non-canonical tokens exactly as Jackson would, rather than ingesting them. The JSON5 forms live in + /// [#parseJson5Float]. + private static Double parseCanonicalFloat(String text) { + if (!CANONICAL_NUMBER.matcher(text).matches()) { + throw new IllegalArgumentException("Invalid canonical SQLite JSONB float: " + text); + } + return Double.parseDouble(text); + } + + private static Double parseJson5Float(String text) { + // JSON5 permits a leading '+' and the tokens Infinity / NaN, all of which Double.parseDouble accepts once + // the '+' is stripped. + String normalized = !text.isEmpty() && text.charAt(0) == '+' ? text.substring(1) : text; + return Double.parseDouble(normalized); + } + + /// Single-pass JSON string unescape. Returns the input unchanged (no allocation) when it contains no + /// backslash. Rejects invalid escapes for `TEXTJ`; when {@code json5} is set, also accepts the JSON5 escape + /// extensions (`\'`, `\v`, `\0`, `\xHH`, line continuations, and `\` 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 + return i; + } + } + + private static int parseHex(String content, int start, int len) { + if (start + len > content.length()) { + throw new IllegalArgumentException("Truncated \\u / \\x escape in SQLite JSONB text"); + } + int value = 0; + for (int i = start; i < start + len; i++) { + int digit = Character.digit(content.charAt(i), 16); + if (digit < 0) { + throw new IllegalArgumentException("Invalid hex escape in SQLite JSONB text"); + } + value = (value << 4) | digit; + } + return value; + } + + /// Sequential reader over a bounded region of a byte array. Multi-byte size fields are big-endian per the + /// SQLite JSONB spec. + private static final class Cursor { + private final byte[] _buf; + private final int _limit; + private int _pos; + + private Cursor(byte[] buf, int start, int limit) { + _buf = buf; + _pos = start; + _limit = limit; + } + + /// The next byte, without advancing. + private int peekUInt8() { + if (_pos >= _limit) { + throw new IllegalArgumentException("Truncated SQLite JSONB payload"); + } + return _buf[_pos] & 0xFF; + } + + /// Advances past {@code count} bytes, verifying they are present. Compares without adding to {@code _pos} + /// so the check cannot overflow. + private void skip(int count) { + if (count > _limit - _pos) { + throw new IllegalArgumentException("Truncated SQLite JSONB payload"); + } + _pos += count; + } + + /// Validates that a payload of {@code size} bytes starting at {@code start} fits inside the enclosing + /// container (never merely inside the whole payload) and returns its exclusive end. Compares without adding + /// to {@code start} so an adversarial `uint64` size cannot overflow past the guard. A {@code start} already + /// past {@code parentEnd} — possible when a trailing element's header straddles the boundary — makes + /// {@code parentEnd - start} negative and is therefore rejected too. + private int boundedEnd(int start, long size, int parentEnd) { + if (size < 0 || size > parentEnd - start) { + throw new IllegalArgumentException("SQLite JSONB element size exceeds its enclosing container"); + } + return start + (int) size; + } + + /// Decodes a numeric token as ASCII. Numeric element types are this method's only callers, so it also + /// enforces the numeric length cap here: `BigInteger(String)` parsing is ~O(n^2), and without a bound a + /// single ~1 MB run of digits would be a per-message CPU sink with no analogue in text JSON, which caps + /// number length the same way via Jackson's StreamReadConstraints. + private String ascii(int start, int end) { + if (end - start > MAX_NUMBER_LENGTH) { + throw new IllegalArgumentException( + "SQLite JSONB numeric token exceeds the maximum length of " + MAX_NUMBER_LENGTH); + } + return new String(_buf, start, end - start, StandardCharsets.US_ASCII); + } + + /// Decodes text strictly as UTF-8. Unlike `new String(..., UTF_8)`, which substitutes U+FFFD for malformed + /// bytes, a `CodingErrorAction.REPORT` decoder throws, so a corrupt JSONB record fails rather than ingesting + /// mutated field names or values -- matching how text JSON / Jackson reject invalid UTF-8. + private String utf8(int start, int end) { + CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder() + .onMalformedInput(CodingErrorAction.REPORT) + .onUnmappableCharacter(CodingErrorAction.REPORT); + try { + return decoder.decode(ByteBuffer.wrap(_buf, start, end - start)).toString(); + } catch (CharacterCodingException e) { + throw new IllegalArgumentException("Invalid UTF-8 in SQLite JSONB text", e); + } + } + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java new file mode 100644 index 000000000000..397fea320943 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/TextJsonPayloadParser.java @@ -0,0 +1,55 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import java.util.Map; +import org.apache.pinot.spi.utils.JsonUtils; + + +/// Parses UTF-8 text JSON, the historical [org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder] +/// behavior. Delegates to [JsonUtils#bytesToMap] so the produced value types exactly match the rest of Pinot. +class TextJsonPayloadParser implements JsonPayloadParser { + + @Override + public boolean matches(byte[] payload, int offset, int length) { + return startsWithJsonDocument(payload, offset, length); + } + + /// Whether the region's first non-whitespace byte opens a JSON document. A decoder can only turn a top-level + /// object into a row, but `[` is tolerated so detection does not misfire on top-level arrays. + /// + /// Shared with [PostgresJsonbPayloadParser], whose wire format is this same text JSON behind a version byte. + static boolean startsWithJsonDocument(byte[] payload, int offset, int length) { + for (int i = offset, end = offset + length; i < end; i++) { + byte b = payload[i]; + // JSON insignificant whitespace: space, tab, LF, CR. + if (b == ' ' || b == '\t' || b == '\n' || b == '\r') { + continue; + } + return b == '{' || b == '['; + } + return false; + } + + @Override + public Map parse(byte[] payload, int offset, int length) + throws Exception { + return JsonUtils.bytesToMap(payload, offset, length); + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java new file mode 100644 index 000000000000..cbcfe94402be --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java @@ -0,0 +1,263 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.cbor.CBORFactory; +import com.fasterxml.jackson.dataformat.cbor.CBORGenerator; +import com.fasterxml.jackson.dataformat.smile.SmileFactory; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Map; +import java.util.Set; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertThrows; +import static org.testng.Assert.assertTrue; + + +/// End-to-end coverage of {@link JSONMessageDecoder} decoding each configured / auto-detected payload format +/// through the shared {@link JSONRecordExtractor} into a {@link GenericRow}. +public class JSONMessageDecoderBinaryTest { + + private static final Set RICH_FIELDS = Set.of("name", "count", "ratio"); + private static final Set SINGLE_FIELD = Set.of("a"); + + /// AUTO is opt-in, so detection tests must ask for it explicitly. + private static final Map AUTO = Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "AUTO"); + + private static byte[] smile(Map value) + throws Exception { + return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value); + } + + private static byte[] cbor(Map value) + throws Exception { + CBORFactory factory = new CBORFactory(); + factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER); + return new ObjectMapper(factory).writeValueAsBytes(value); + } + + private static byte[] bytes(int... values) { + byte[] result = new byte[values.length]; + for (int i = 0; i < values.length; i++) { + result[i] = (byte) values[i]; + } + return result; + } + + // Explicit UTF-8 so fixtures do not depend on the platform default charset. + private static final byte[] TEXT_DOC = + "{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes(StandardCharsets.UTF_8); + + // Hand-built {"a": 1} fixtures matching the parser unit tests. + // SQLite: object(size 4){ text("a"), int("1") }. + private static final byte[] SQLITE_A1 = bytes(0x4C, 0x17, 0x61, 0x13, 0x31); + // PostgreSQL jsonb_send framing: version byte 1 followed by the text JSON body. + private static final byte[] POSTGRES_A1 = + bytes(0x01, '{', '"', 'a', '"', ':', '1', '}'); + + private GenericRow decode(Map props, Set 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 richDoc() { + return Map.of("name", "pinot", "count", 7, "ratio", 2.5); + } + + /// An unset jsonFormat means TEXT -- the decoder's historical behavior, with no detection at all. + @Test + public void testUnsetFormatIsText() + throws Exception { + assertRich(decode(Map.of(), RICH_FIELDS, TEXT_DOC)); + } + + /// An unset jsonFormat must not silently auto-detect binary payloads: those streams keep failing as they did + /// before this feature existed, rather than being decoded (or worse, ingested as a partial row). + @Test + public void testUnsetFormatDoesNotAutoDetectBinary() + throws Exception { + assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD, SQLITE_A1)); + assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD, POSTGRES_A1)); + assertThrows(RuntimeException.class, () -> decode(Map.of(), RICH_FIELDS, smile(richDoc()))); + assertThrows(RuntimeException.class, () -> decode(Map.of(), RICH_FIELDS, cbor(richDoc()))); + // A lone 0x0C is a validly-exact empty SQLite object, but must not become an empty row on a text stream. + assertThrows(RuntimeException.class, () -> decode(Map.of(), SINGLE_FIELD, bytes(0x0C))); + } + + @Test + public void testConfiguredSmile() + throws Exception { + assertRich(decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SMILE"), RICH_FIELDS, smile(richDoc()))); + } + + @Test + public void testConfiguredCbor() + throws Exception { + assertRich(decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "CBOR"), RICH_FIELDS, cbor(richDoc()))); + } + + @Test + public void testConfiguredSqlite() + throws Exception { + GenericRow row = decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SQLITE_JSONB"), SINGLE_FIELD, + SQLITE_A1); + assertEquals(row.getValue("a"), 1); + } + + @Test + public void testConfiguredPostgres() + throws Exception { + GenericRow row = decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "POSTGRES_JSONB"), SINGLE_FIELD, + POSTGRES_A1); + assertEquals(row.getValue("a"), 1); + } + + @Test + public void testAutoDetectsSmile() + throws Exception { + assertRich(decode(AUTO, RICH_FIELDS, smile(richDoc()))); + } + + @Test + public void testAutoDetectsCbor() + throws Exception { + assertRich(decode(AUTO, RICH_FIELDS, cbor(richDoc()))); + } + + @Test + public void testAutoDetectsSqlite() + throws Exception { + assertEquals(decode(AUTO, SINGLE_FIELD, SQLITE_A1).getValue("a"), 1); + } + + @Test + public void testAutoDetectsPostgres() + throws Exception { + assertEquals(decode(AUTO, SINGLE_FIELD, POSTGRES_A1).getValue("a"), 1); + } + + @Test + public void testAutoStillDecodesText() + throws Exception { + assertRich(decode(AUTO, RICH_FIELDS, TEXT_DOC)); + } + + @Test + public void testMalformedPayloadIsWrappedInRuntimeException() { + // A stream decoder routinely meets corrupt messages; decode() must surface them as a RuntimeException + // rather than leaking the parser's own exception type. + byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF); + assertThrows(RuntimeException.class, + () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SMILE"), RICH_FIELDS, badSmile)); + // Same via AUTO, which detects Smile from the header and then fails to parse the body. + assertThrows(RuntimeException.class, () -> decode(AUTO, RICH_FIELDS, badSmile)); + // Truncated SQLite object. + assertThrows(RuntimeException.class, + () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SQLITE_JSONB"), SINGLE_FIELD, + bytes(0x5C, 0x17))); + } + + @Test + public void testSqliteTrailingBytesAreRejectedRatherThanIngestedAsAPartialRow() { + // A SQLite JSONB payload whose top-level element declares a short size (here an empty object) used to + // decode to an empty row, silently discarding the trailing "a": 1. Detection no longer claims it (the + // declared size does not fill the payload) and the parser rejects it outright; either way the message + // must fail rather than yield a partial row. + byte[] shortObjectThenData = bytes(0x0C, 0x17, 0x61, 0x13, 0x31); + assertThrows(RuntimeException.class, () -> decode(AUTO, SINGLE_FIELD, shortObjectThenData)); + assertThrows(RuntimeException.class, + () -> decode(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SQLITE_JSONB"), SINGLE_FIELD, + shortObjectThenData)); + } + + @Test + public void testUnsupportedFormatIsRejectedAtInit() { + JSONMessageDecoder decoder = new JSONMessageDecoder(); + assertThrows(IllegalArgumentException.class, + () -> decoder.init(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "bson"), SINGLE_FIELD, "topic")); + } + + @Test + public void testDecodeFailureMessageIsBoundedAndCharsetSafe() + throws Exception { + // A malformed *text* payload keeps its readable rendering ... + String message = decodeFailureMessage(Map.of(), "{\"name\":".getBytes(StandardCharsets.UTF_8)); + assertTrue(message.contains("8 bytes: {\"name\":"), message); + + // ... including non-ASCII UTF-8, since bytes >= 0x80 are not control characters ... + message = decodeFailureMessage(Map.of(), "{\"n\":\"café\"".getBytes(StandardCharsets.UTF_8)); + assertTrue(message.contains("{\"n\":\"café\""), message); + + // ... while a malformed *binary* payload is hex-encoded rather than dumped as mojibake. + byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF); + message = decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SMILE"), badSmile); + assertTrue(message.contains("7 bytes: 0x3a290a04ffffff"), message); + + // ... an oversized text payload is truncated to the 512-byte window instead of echoed in full ... + byte[] hugeText = new byte[8192]; + Arrays.fill(hugeText, (byte) 'x'); + message = decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "TEXT"), hugeText); + assertTrue(message.contains("8192 bytes: "), message); + assertTrue(message.contains("...(truncated)"), message); + assertEquals(preview(message), "x".repeat(512)); + + // ... and an oversized binary payload is capped at 128 hex-rendered bytes (256 characters). + byte[] hugeBinary = new byte[8192]; + Arrays.fill(hugeBinary, (byte) 0x01); + hugeBinary[0] = 0x3A; + hugeBinary[1] = 0x29; + hugeBinary[2] = 0x0A; + message = decodeFailureMessage(Map.of(JSONMessageDecoder.JSON_FORMAT_CONFIG_KEY, "SMILE"), hugeBinary); + assertTrue(message.contains("8192 bytes: 0x3a290a01"), message); + assertTrue(message.contains("...(truncated)"), message); + // "0x" prefix + 128 rendered bytes at two hex characters each. + assertEquals(preview(message).length(), 2 + 2 * 128, message); + } + + /// The rendered payload window of a decode-failure message: what sits between " bytes: " and the + /// truncation marker. Isolates the assertion from the message's surrounding prose. + private static String preview(String message) { + int start = message.indexOf("bytes: ") + "bytes: ".length(); + int end = message.indexOf("...(truncated)"); + return message.substring(start, end < 0 ? message.length() : end); + } + + /// Decodes an intentionally malformed payload and returns the resulting exception message. + private String decodeFailureMessage(Map props, byte[] payload) + throws Exception { + try { + decode(props, SINGLE_FIELD, payload); + throw new AssertionError("expected decode to fail"); + } catch (RuntimeException e) { + return e.getMessage(); + } + } +} diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java new file mode 100644 index 000000000000..fc3685ea5dc9 --- /dev/null +++ b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java @@ -0,0 +1,670 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.plugin.inputformat.json.format; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.dataformat.cbor.CBORFactory; +import com.fasterxml.jackson.dataformat.cbor.CBORGenerator; +import com.fasterxml.jackson.dataformat.smile.SmileFactory; +import java.math.BigInteger; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertSame; +import static org.testng.Assert.assertThrows; +import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; + + +public class JsonPayloadFormatTest { + + private static byte[] bytes(int... values) { + byte[] result = new byte[values.length]; + for (int i = 0; i < values.length; i++) { + result[i] = (byte) values[i]; + } + return result; + } + + /// Explicit UTF-8 so fixtures do not depend on the platform default charset. + private static byte[] utf8(String value) { + return value.getBytes(StandardCharsets.UTF_8); + } + + /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text JSON body. + private static byte[] postgres(String json) { + byte[] body = utf8(json); + byte[] result = new byte[body.length + 1]; + result[0] = 1; + System.arraycopy(body, 0, result, 1, body.length); + return result; + } + + private static byte[] smile(Object value) + throws Exception { + return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value); + } + + private static byte[] cborSelfDescribed(Object value) + throws Exception { + CBORFactory factory = new CBORFactory(); + factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER); + return new ObjectMapper(factory).writeValueAsBytes(value); + } + + private static Map parse(JsonPayloadParser parser, byte[] payload) + throws Exception { + return parser.parse(payload, 0, payload.length); + } + + // ------------------------------------------------------------------------------------------------------ + // Config parsing + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testFromConfig() { + // An unset property means TEXT -- the decoder's historical behavior. AUTO is opt-in, never implicit, + // because detection can claim a corrupt message that text decoding would have rejected. + assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.TEXT); + assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.TEXT); + assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.TEXT); + + assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT); + assertEquals(JsonPayloadFormat.fromConfig("auto"), JsonPayloadFormat.AUTO); + assertEquals(JsonPayloadFormat.fromConfig(" Smile "), JsonPayloadFormat.SMILE); + assertEquals(JsonPayloadFormat.fromConfig("POSTGRES_JSONB"), JsonPayloadFormat.POSTGRES_JSONB); + // AUTO resolves the concrete format per message, so every format has a non-null parser. + assertSame(JsonPayloadFormat.AUTO.getParser().getClass(), AutoDetectPayloadParser.class); + } + + @Test + public void testFromConfigRejectsUnknownFormatAndKeepsTheCause() { + IllegalArgumentException e = + expectThrows(IllegalArgumentException.class, () -> JsonPayloadFormat.fromConfig("bson")); + // The message names the offending value and the supported set ... + assertTrue(e.getMessage().contains("bson"), e.getMessage()); + assertTrue(e.getMessage().contains("SQLITE_JSONB"), e.getMessage()); + // ... and the underlying valueOf failure is preserved so the stack trace still points at the parse. + assertNotNull(e.getCause()); + } + + // ------------------------------------------------------------------------------------------------------ + // Text + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testText() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser(); + Map map = parse(parser, utf8(" {\"a\": 1, \"b\": \"x\"}")); + assertEquals(map.get("a"), 1); + assertEquals(map.get("b"), "x"); + assertTrue(parser.matches(utf8("{\"a\":1}"), 0, 7)); + assertTrue(parser.matches(utf8(" [1,2]"), 0, 7)); + assertFalse(parser.matches(bytes(0xFF), 0, 1)); + } + + // ------------------------------------------------------------------------------------------------------ + // Smile / CBOR (round-tripped through Jackson) + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testSmile() + throws Exception { + Map original = Map.of("name", "pinot", "count", 7, "ratio", 2.5, "ok", true, "tags", + List.of("a", "b")); + byte[] encoded = smile(original); + + JsonPayloadParser parser = JsonPayloadFormat.SMILE.getParser(); + assertTrue(parser.matches(encoded, 0, encoded.length)); + assertEquals(parse(parser, encoded), original); + } + + @Test + public void testCbor() + throws Exception { + Map original = Map.of("name", "pinot", "count", 7, "ratio", 2.5, "ok", true, "nested", + Map.of("x", 1)); + byte[] encoded = cborSelfDescribed(original); + + JsonPayloadParser parser = JsonPayloadFormat.CBOR.getParser(); + assertTrue(parser.matches(encoded, 0, encoded.length)); + assertEquals(parse(parser, encoded), original); + } + + @Test + public void testCborWithoutSelfDescribeTagIsNotAutoDetected() + throws Exception { + byte[] noTag = new ObjectMapper(new CBORFactory()).writeValueAsBytes(Map.of("name", "pinot")); + JsonPayloadParser parser = JsonPayloadFormat.CBOR.getParser(); + // Not detectable without the tag ... + assertFalse(parser.matches(noTag, 0, noTag.length)); + assertSame(JsonPayloadFormat.detectParser(noTag, 0, noTag.length).getClass(), TextJsonPayloadParser.class); + // ... but explicitly configuring CBOR still decodes it. + assertEquals(parse(parser, noTag).get("name"), "pinot"); + } + + @Test + public void testBinaryFormatsPreserveFloatAndBytes() + throws Exception { + // Float and byte[] are scalars the binary formats can carry but text JSON cannot. The parser contract + // documents that they pass through; guard against a regression that silently upcasts Float -> Double. + Map original = new HashMap<>(); + original.put("f", 1.5f); + original.put("bin", new byte[]{1, 2, 3}); + + for (byte[] encoded : List.of(smile(original), cborSelfDescribed(original))) { + Map decoded = parse(JsonPayloadFormat.detectParser(encoded, 0, encoded.length), encoded); + assertTrue(decoded.get("f") instanceof Float, "expected Float, got " + decoded.get("f").getClass()); + assertEquals(decoded.get("f"), 1.5f); + assertEquals((byte[]) decoded.get("bin"), new byte[]{1, 2, 3}); + } + } + + @Test + public void testJacksonFormatsRejectMalformedPayloads() { + // Valid Smile header, garbage body. + byte[] badSmile = bytes(0x3A, 0x29, 0x0A, 0x04, 0xFF, 0xFF, 0xFF); + assertThrows(Exception.class, () -> parse(JsonPayloadFormat.SMILE.getParser(), badSmile)); + // Self-describe tag then a truncated map. + byte[] badCbor = bytes(0xD9, 0xD9, 0xF7, 0xBF, 0x63); + assertThrows(Exception.class, () -> parse(JsonPayloadFormat.CBOR.getParser(), badCbor)); + } + + // ------------------------------------------------------------------------------------------------------ + // PostgreSQL JSONB wire format: version byte + text JSON (jsonb_send / jsonb_recv) + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testPostgresWireFormat() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser(); + byte[] payload = postgres("{\"a\":1,\"b\":\"x\",\"c\":1.5,\"d\":true,\"e\":null,\"f\":[1,2]}"); + assertTrue(parser.matches(payload, 0, payload.length)); + + Map map = parse(parser, payload); + // Identical type contract to TEXT, since the body is plain text JSON. + assertEquals(map.get("a"), 1); + assertEquals(map.get("b"), "x"); + assertEquals(map.get("c"), 1.5); + assertEquals(map.get("d"), true); + assertNull(map.get("e")); + assertEquals(map.get("f"), List.of(1, 2)); + } + + @Test + public void testPostgresBigIntegerNarrowing() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser(); + Map map = parse(parser, postgres("{\"n\":9999999999,\"h\":100000000000000000000}")); + assertEquals(map.get("n"), 9999999999L); + assertEquals(map.get("h"), new BigInteger("100000000000000000000")); + } + + @Test + public void testPostgresErrors() { + JsonPayloadParser parser = JsonPayloadFormat.POSTGRES_JSONB.getParser(); + // Unsupported version byte (jsonb_recv rejects anything but 1). + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x02, '{', '}'))); + // Version byte only. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x01))); + // Version byte followed by a non-JSON body is not claimed by detection. + assertFalse(parser.matches(bytes(0x01, 0xFF, 0xFF), 0, 3)); + // The raw on-disk JsonbContainer layout is NOT the wire format and must not be silently accepted. + assertFalse(parser.matches(bytes(0x01, 0x01, 0x00, 0x00, 0x20), 0, 5)); + } + + // ------------------------------------------------------------------------------------------------------ + // SQLite JSONB (hand-computed fixtures per https://sqlite.org/jsonb.html) + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testSqliteScalars() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + + // {"a": 1} -> object(size 4){ text("a"), int("1") } + assertEquals(parse(parser, bytes(0x4C, 0x17, 0x61, 0x13, 0x31)), Map.of("a", 1)); + // {"b": true} -> object(size 3){ text("b"), true(size 0) } + assertEquals(parse(parser, bytes(0x3C, 0x17, 0x62, 0x01)), Map.of("b", true)); + // {"f2": false} -> object(size 4){ text("f2"), false(size 0) } + assertEquals(parse(parser, bytes(0x4C, 0x27, 0x66, 0x32, 0x02)), Map.of("f2", false)); + // {"s": "hi"} -> object(size 5){ text("s"), text("hi") } + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x73, 0x27, 0x68, 0x69)), Map.of("s", "hi")); + // {"r": "hi"} -> object(size 5){ text("r"), textraw("hi") } + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x72, 0x2A, 0x68, 0x69)), Map.of("r", "hi")); + // {"f": 1.5} -> object(size 6){ text("f"), float("1.5") } + assertEquals(parse(parser, bytes(0x6C, 0x17, 0x66, 0x35, 0x31, 0x2E, 0x35)), Map.of("f", 1.5)); + } + + @Test + public void testSqliteJson5AndEscapes() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + + // {"h": 0x1F} -> object{ text("h"), int5("0x1F") } -> 31 + assertEquals(parse(parser, bytes(0x7C, 0x17, 0x68, 0x44, 0x30, 0x78, 0x31, 0x46)), Map.of("h", 31)); + // {"i": Infinity} -> object{ text("i"), float5("Infinity") } + assertEquals(parse(parser, bytes(0xBC, 0x17, 0x69, 0x86, 0x49, 0x6E, 0x66, 0x69, 0x6E, 0x69, 0x74, 0x79)), + Map.of("i", Double.POSITIVE_INFINITY)); + // {"t": "a\nb"} -> object{ text("t"), textj("a\\nb") } -> newline unescaped + assertEquals(parse(parser, bytes(0x7C, 0x17, 0x74, 0x48, 0x61, 0x5C, 0x6E, 0x62)), Map.of("t", "a\nb")); + // {"t": "\x41"} -> object{ text("t"), text5("\\x41") } -> "A" (JSON5 hex escape) + assertEquals(parse(parser, bytes(0x7C, 0x17, 0x74, 0x49, 0x5C, 0x78, 0x34, 0x31)), Map.of("t", "A")); + } + + @Test + public void testSqliteTextjRejectsInvalidEscape() { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // TEXTJ carries standard JSON escapes only; "\q" is invalid and must not be silently passed through. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x6C, 0x17, 0x74, 0x38, 0x5C, 0x71))); + } + + @Test + public void testSqliteNested() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + + // {"arr": [1, 2]} -> object(size 9){ text("arr"), array(size 4){ int("1"), int("2") } } + assertEquals(parse(parser, bytes(0x9C, 0x37, 0x61, 0x72, 0x72, 0x4B, 0x13, 0x31, 0x13, 0x32)), + Map.of("arr", List.of(1, 2))); + + // {"o": {"x": null}} -> object(size 6){ text("o"), object(size 3){ text("x"), null } } + Map nested = parse(parser, bytes(0x6C, 0x17, 0x6F, 0x3C, 0x17, 0x78, 0x00)); + //noinspection unchecked + Map inner = (Map) nested.get("o"); + assertTrue(inner.containsKey("x")); + assertNull(inner.get("x")); + } + + @Test + public void testSqliteIntNarrowing() + throws Exception { + assertEquals(sqliteSingleInt("42"), 42); // fits int + assertEquals(sqliteSingleInt("9999999999"), 9999999999L); // > int + assertEquals(sqliteSingleInt("99999999999999999999"), new BigInteger("99999999999999999999")); // > long + } + + /// Builds `{"n": }` as SQLite JSONB and returns the decoded value of `n`. + private static Object sqliteSingleInt(String digits) + throws Exception { + byte[] digitBytes = digits.getBytes(StandardCharsets.US_ASCII); + // An element header carries its size in the high nibble when it fits 0..11, else via descriptor 12. + boolean wideInt = digitBytes.length > 11; + int valueElem = (wideInt ? 2 : 1) + digitBytes.length; + int objPayload = 2 + valueElem; + boolean wideObj = objPayload > 11; + byte[] payload = new byte[(wideObj ? 2 : 1) + objPayload]; + int i = 0; + payload[i++] = (byte) (wideObj ? 0xCC : ((objPayload << 4) | 0x0C)); + if (wideObj) { + payload[i++] = (byte) objPayload; + } + payload[i++] = (byte) 0x17; // text "n" (size 1) + payload[i++] = (byte) 'n'; + payload[i++] = (byte) (wideInt ? 0xC3 : ((digitBytes.length << 4) | 0x03)); + if (wideInt) { + payload[i++] = (byte) digitBytes.length; + } + System.arraycopy(digitBytes, 0, payload, i, digitBytes.length); + return parse(JsonPayloadFormat.SQLITE_JSONB.getParser(), payload).get("n"); + } + + @Test + public void testSqliteTwoByteSizeDescriptor() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // {"k": "<300 x 'a'>"} forces size descriptor 13 (2-byte big-endian size) for both the text value and the + // enclosing object. + int textLen = 300; + int valueElem = 1 + 2 + textLen; + int objPayload = 2 + valueElem; + byte[] payload = new byte[3 + objPayload]; + int i = 0; + payload[i++] = (byte) 0xDC; // OBJECT, size descriptor 13 + payload[i++] = (byte) (objPayload >>> 8); + payload[i++] = (byte) objPayload; + payload[i++] = (byte) 0x17; // text "k" + payload[i++] = (byte) 'k'; + payload[i++] = (byte) 0xD7; // TEXT, size descriptor 13 + payload[i++] = (byte) (textLen >>> 8); + payload[i++] = (byte) textLen; + Arrays.fill(payload, i, i + textLen, (byte) 'a'); + assertEquals(parse(parser, payload).get("k"), "a".repeat(textLen)); + } + + @Test + public void testSqliteErrors() { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // Reserved element type (low nibble 13). + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x0D))); + // Non-object top-level (a bare int). + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x13, 0x35))); + // Truncated: object claims 5 payload bytes but only 1 follows. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x5C, 0x17))); + // Object payload ends after a label, with no value for it. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x2C, 0x17, 0x61))); + // Object label must be a text element, not an int. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x3C, 0x13, 0x31, 0x01))); + } + + @Test + public void testSqliteTopLevelElementMustExactlyFillThePayload() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // A bare empty object consumes the whole payload and is valid. + assertEquals(parse(parser, bytes(0x0C)), Map.of()); + + // SQLite's validity rule requires the outer element to exactly fill the BLOB. Previously a top-level + // element declaring a short size decoded to a partial row and silently dropped the trailing bytes: + // an empty object (size 0) followed by real data ... + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x0C, 0x17, 0x61, 0x13, 0x31))); + // ... and a well-formed {"a": 1} object with a stray trailing byte. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x4C, 0x17, 0x61, 0x13, 0x31, 0xFF))); + } + + @Test + public void testSqliteWideSizeDescriptors() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // Descriptor 14: object size as a 4-byte big-endian int. Payload is text("a") + int("1"). + assertEquals(parse(parser, bytes(0xEC, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31)), Map.of("a", 1)); + // Descriptor 15: object size as an 8-byte big-endian int. + assertEquals(parse(parser, + bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31)), + Map.of("a", 1)); + + // A uint64 size with the sign bit set decodes to a negative long and must be rejected, not wrapped. + assertThrows(IllegalArgumentException.class, () -> parse(parser, + bytes(0xFC, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x17))); + // A large positive size must be rejected by comparison, without overflowing when added to the offset. + assertThrows(IllegalArgumentException.class, () -> parse(parser, + bytes(0xFC, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x17))); + } + + @Test + public void testSqliteJson5NumberVariants() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // {"h": -0x10} -> int5 with a sign and hex radix -> -16 + assertEquals(parse(parser, bytes(0x8C, 0x17, 0x68, 0x54, 0x2D, 0x30, 0x78, 0x31, 0x30)), Map.of("h", -16)); + // {"p": +1.5} -> float5 with a leading '+' (rejected by Double.parseDouble until stripped) + assertEquals(parse(parser, bytes(0x7C, 0x17, 0x70, 0x46, 0x2B, 0x31, 0x2E, 0x35)), Map.of("p", 1.5)); + // JSON5 floats may be non-finite: FLOAT5 "NaN" / "Infinity". + assertEquals(parse(parser, bytes(0x6C, 0x17, 0x78, 0x36, 0x4E, 0x61, 0x4E)).get("x"), Double.NaN); + assertEquals(parse(parser, bytes(0xBC, 0x17, 0x78, 0x86, 0x49, 0x6E, 0x66, 0x69, 0x6E, 0x69, 0x74, 0x79)).get("x"), + Double.POSITIVE_INFINITY); + } + + @Test + public void testSqliteCanonicalNumbersRejectNonCanonicalTokens() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // TYPE_FLOAT is canonical RFC 8259 and must reject the permissive tokens Double.parseDouble accepts, so a + // malformed float fails the record rather than ingesting a non-finite / non-canonical value. + assertThrows(IllegalArgumentException.class, // {"x": NaN} + () -> parse(parser, bytes(0x6C, 0x17, 0x78, 0x35, 0x4E, 0x61, 0x4E))); + assertThrows(IllegalArgumentException.class, // {"x": Infinity} + () -> parse(parser, bytes(0xBC, 0x17, 0x78, 0x85, 0x49, 0x6E, 0x66, 0x69, 0x6E, 0x69, 0x74, 0x79))); + assertThrows(IllegalArgumentException.class, // {"x": +1.5} + () -> parse(parser, bytes(0x7C, 0x17, 0x78, 0x45, 0x2B, 0x31, 0x2E, 0x35))); + assertThrows(IllegalArgumentException.class, // {"x": 0x1p4} (Java hex float) + () -> parse(parser, bytes(0x8C, 0x17, 0x78, 0x55, 0x30, 0x78, 0x31, 0x70, 0x34))); + + // TYPE_INT is likewise canonical: no leading '+' or leading zeros. + assertThrows(IllegalArgumentException.class, // {"x": +5} + () -> parse(parser, bytes(0x5C, 0x17, 0x78, 0x23, 0x2B, 0x35))); + assertThrows(IllegalArgumentException.class, // {"x": 007} + () -> parse(parser, bytes(0x6C, 0x17, 0x78, 0x33, 0x30, 0x30, 0x37))); + + // The canonical happy paths still decode. + assertEquals(parse(parser, bytes(0x6C, 0x17, 0x78, 0x35, 0x31, 0x2E, 0x35)), Map.of("x", 1.5)); // 1.5 + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x78, 0x23, 0x2D, 0x35)), Map.of("x", -5)); // -5 + } + + @Test + public void testSqliteTextMustBeValidUtf8() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // Valid multibyte UTF-8 (é = 0xC3 0xA9) decodes. + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x78, 0x27, 0xC3, 0xA9)), Map.of("x", "é")); + // A lone 0x80 (invalid UTF-8) in a value must fail the record rather than becoming U+FFFD. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x4C, 0x17, 0x78, 0x17, 0x80))); + // Same for a field name. + assertThrows(IllegalArgumentException.class, () -> parse(parser, bytes(0x4C, 0x17, 0x80, 0x13, 0x31))); + } + + // SQLite JSONB element type codes used by the builders below. + private static final int SQLITE_TYPE_INT = 3; + private static final int SQLITE_TYPE_TEXT = 7; + private static final int SQLITE_TYPE_ARRAY = 11; + private static final int SQLITE_TYPE_OBJECT = 12; + + /// Builds a SQLite JSONB element (header + payload) whose declared size exactly matches {@code payload}, + /// choosing the narrowest size descriptor. Used to hand-build arbitrarily large / deep valid fixtures. + private static byte[] sqliteElement(int type, byte[] payload) { + int size = payload.length; + byte[] header; + if (size <= 11) { + header = bytes((size << 4) | type); + } else if (size <= 0xFF) { + header = bytes((12 << 4) | type, size); + } else if (size <= 0xFFFF) { + header = bytes((13 << 4) | type, (size >>> 8) & 0xFF, size & 0xFF); + } else { + header = bytes((14 << 4) | type, (size >>> 24) & 0xFF, (size >>> 16) & 0xFF, (size >>> 8) & 0xFF, size & 0xFF); + } + byte[] out = new byte[header.length + payload.length]; + System.arraycopy(header, 0, out, 0, header.length); + System.arraycopy(payload, 0, out, header.length, payload.length); + return out; + } + + /// {@code {"": }} as an exactly-filling SQLite JSONB object. + private static byte[] sqliteObject(String key, byte[] valueElement) { + byte[] label = sqliteElement(SQLITE_TYPE_TEXT, key.getBytes(StandardCharsets.US_ASCII)); + byte[] payload = new byte[label.length + valueElement.length]; + System.arraycopy(label, 0, payload, 0, label.length); + System.arraycopy(valueElement, 0, payload, label.length, valueElement.length); + return sqliteElement(SQLITE_TYPE_OBJECT, payload); + } + + /// {@code {"a": [[[ ... ]]]}} with {@code depth} nested arrays, innermost empty. Each level is a valid, + /// exactly-filling element, so the whole payload is well-formed apart from its depth. + private static byte[] sqliteNestedArrays(int depth) { + byte[] element = sqliteElement(SQLITE_TYPE_ARRAY, new byte[0]); + for (int i = 0; i < depth; i++) { + element = sqliteElement(SQLITE_TYPE_ARRAY, element); + } + return sqliteObject("a", element); + } + + @Test + public void testSqliteRejectsExcessiveNestingDepth() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // Moderate nesting decodes -- confirms the fixture builder and normal nesting are fine. + assertTrue(parse(parser, sqliteNestedArrays(50)).get("a") instanceof List); + // Past the depth cap it must fail as a bad record (IllegalArgumentException), NOT recurse into a + // StackOverflowError, which is an Error that would escape decode-error handling and fail the segment. + assertThrows(IllegalArgumentException.class, () -> parse(parser, sqliteNestedArrays(1200))); + } + + @Test + public void testSqliteRejectsOversizedNumericToken() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // A 1000-digit canonical integer is accepted (parity with Jackson's number-length cap). + byte[] atLimit = + sqliteObject("n", sqliteElement(SQLITE_TYPE_INT, "9".repeat(1000).getBytes(StandardCharsets.US_ASCII))); + assertTrue(parse(parser, atLimit).get("n") instanceof BigInteger); + // One digit past the cap is rejected before the ~O(n^2) BigInteger parse runs. + byte[] overLimit = + sqliteObject("n", sqliteElement(SQLITE_TYPE_INT, "9".repeat(1001).getBytes(StandardCharsets.US_ASCII))); + assertThrows(IllegalArgumentException.class, () -> parse(parser, overLimit)); + } + + @Test + public void testSqliteTextEscapeVariants() + throws Exception { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // TEXTJ \\u escape: {"u": "A"} -> "A" + assertEquals(parse(parser, bytes(0x9C, 0x17, 0x75, 0x68, 0x5C, 0x75, 0x30, 0x30, 0x34, 0x31)), Map.of("u", "A")); + // TEXT5 extensions, one payload each: \' -> ' , \v -> VT, \0 -> NUL, \q -> q (passthrough) + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x27)), Map.of("e", "'")); + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x76)), Map.of("e", "\u000B")); + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x30)), Map.of("e", "\0")); + assertEquals(parse(parser, bytes(0x5C, 0x17, 0x65, 0x29, 0x5C, 0x71)), Map.of("e", "q")); + // TEXT5 line continuation: "a\b" -> "ab" + assertEquals(parse(parser, bytes(0x7C, 0x17, 0x65, 0x49, 0x61, 0x5C, 0x0A, 0x62)), Map.of("e", "ab")); + } + + @Test + public void testSqliteNestedElementCannotOverrunItsContainer() { + JsonPayloadParser parser = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // object(size 8){ text("a"), array(size 1){ text(size 4) }, "b": 7 } + // The array declares a 1-byte payload, but the text element inside it declares 4 bytes -- which would run + // past the array's end and swallow the object's remaining "b": 7 pair. Must be rejected, not silently + // truncated. + assertThrows(IllegalArgumentException.class, + () -> parse(parser, bytes(0x8C, 0x17, 0x61, 0x1B, 0x47, 0x17, 0x62, 0x13, 0x37))); + } + + // ------------------------------------------------------------------------------------------------------ + // AUTO detection + cross-format equivalence + // ------------------------------------------------------------------------------------------------------ + + @Test + public void testAutoDetection() + throws Exception { + assertSame(JsonPayloadFormat.detectParser(utf8("{\"a\":1}"), 0, 7).getClass(), TextJsonPayloadParser.class); + byte[] smile = smile(Map.of("a", 1)); + assertSame(JsonPayloadFormat.detectParser(smile, 0, smile.length).getClass(), SmileJsonPayloadParser.class); + byte[] cbor = cborSelfDescribed(Map.of("a", 1)); + assertSame(JsonPayloadFormat.detectParser(cbor, 0, cbor.length).getClass(), CborJsonPayloadParser.class); + byte[] pg = postgres("{\"a\":1}"); + assertSame(JsonPayloadFormat.detectParser(pg, 0, pg.length).getClass(), PostgresJsonbPayloadParser.class); + byte[] sqlite = bytes(0x3C, 0x17, 0x62, 0x01); + assertSame(JsonPayloadFormat.detectParser(sqlite, 0, sqlite.length).getClass(), SqliteJsonbPayloadParser.class); + } + + @Test + public void testSqliteDetectionRequiresAnExactlyFillingObject() + throws Exception { + JsonPayloadParser sqlite = JsonPayloadFormat.SQLITE_JSONB.getParser(); + // The OBJECT low nibble alone would claim one arbitrary binary byte in sixteen, so detection additionally + // requires the declared top-level size to exactly fill the payload. + assertTrue(sqlite.matches(bytes(0x4C, 0x17, 0x61, 0x13, 0x31), 0, 5)); // size 4, 4 bytes follow + assertFalse(sqlite.matches(bytes(0x4C, 0x17, 0x61), 0, 3)); // size 4, only 2 bytes follow + assertFalse(sqlite.matches(bytes(0x0C, 0x17, 0x61, 0x13, 0x31), 0, 5)); // size 0, but 4 bytes follow + assertFalse(sqlite.matches(bytes(0x4C, 0x17, 0x61, 0x13, 0x31, 0xFF), 0, 6)); // size 4, trailing byte + + // Random binary that merely happens to carry the OBJECT nibble is no longer claimed ... + assertFalse(sqlite.matches(bytes(0xFC, 0xDE, 0xAD, 0xBE, 0xEF), 0, 5)); + assertSame(JsonPayloadFormat.detectParser(bytes(0x0C, 0x17, 0x61, 0x13, 0x31), 0, 5).getClass(), + TextJsonPayloadParser.class); + + // ... but the degenerate empty object, which does exactly fill its payload, still is. + assertTrue(sqlite.matches(bytes(0x0C), 0, 1)); + + // Every size descriptor participates in the same check. Each positive case is the object {"a": 1} whose + // 4-byte payload is declared through a different-width size field, and each negative case is the same + // payload with the declared size off by one. + // Descriptor 12: 1-byte size. + assertTrue(sqlite.matches(bytes(0xCC, 0x04, 0x17, 0x61, 0x13, 0x31), 0, 6)); + assertFalse(sqlite.matches(bytes(0xCC, 0x05, 0x17, 0x61, 0x13, 0x31), 0, 6)); + // Descriptor 13: 2-byte big-endian size. + assertTrue(sqlite.matches(bytes(0xDC, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31), 0, 7)); + assertFalse(sqlite.matches(bytes(0xDC, 0x00, 0x05, 0x17, 0x61, 0x13, 0x31), 0, 7)); + // A byte-order slip here would read 0x0400 rather than 0x0004. + assertFalse(sqlite.matches(bytes(0xDC, 0x04, 0x00, 0x17, 0x61, 0x13, 0x31), 0, 7)); + // Descriptor 14: 4-byte big-endian size. + assertTrue(sqlite.matches(bytes(0xEC, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31), 0, 9)); + assertFalse(sqlite.matches(bytes(0xEC, 0x00, 0x00, 0x00, 0x05, 0x17, 0x61, 0x13, 0x31), 0, 9)); + // Descriptor 15: 8-byte big-endian size. + assertTrue(sqlite.matches( + bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31), 0, 13)); + assertFalse(sqlite.matches( + bytes(0xFC, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, 0x17, 0x61, 0x13, 0x31), 0, 13)); + // An 8-byte size with the sign bit set decodes negative and can never equal a length. + assertFalse(sqlite.matches(bytes(0xFC, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x17), 0, 10)); + + // Truncated headers must not read past the payload. + assertFalse(sqlite.matches(bytes(0xCC), 0, 1)); + assertFalse(sqlite.matches(bytes(0xDC, 0x00), 0, 2)); + assertFalse(sqlite.matches(bytes(0xEC, 0x00), 0, 2)); + assertFalse(sqlite.matches(bytes(0xFC, 0x00, 0x00, 0x00), 0, 4)); + + // Detection and parsing must agree: anything matches() claims here, parse() accepts. + assertEquals(parse(sqlite, bytes(0xCC, 0x04, 0x17, 0x61, 0x13, 0x31)), Map.of("a", 1)); + assertEquals(parse(sqlite, bytes(0xDC, 0x00, 0x04, 0x17, 0x61, 0x13, 0x31)), Map.of("a", 1)); + } + + @Test + public void testAutoDetectionFallbackAndAdversarialInput() { + // Nothing matches -> TEXT fallback. + assertSame(JsonPayloadFormat.detectParser(bytes(0xFF), 0, 1).getClass(), TextJsonPayloadParser.class); + // Empty payload -> TEXT fallback; matches() must not throw on too-short input. + assertSame(JsonPayloadFormat.detectParser(bytes(), 0, 0).getClass(), TextJsonPayloadParser.class); + // Truncated Smile header must not be claimed by the Smile parser. + assertSame(JsonPayloadFormat.detectParser(bytes(0x3A, 0x29), 0, 2).getClass(), TextJsonPayloadParser.class); + // A lone 0x01 with no JSON body must not be claimed by the Postgres parser. + assertSame(JsonPayloadFormat.detectParser(bytes(0x01, 0x02), 0, 2).getClass(), TextJsonPayloadParser.class); + } + + @Test + public void testCrossFormatEquivalence() + throws Exception { + // The same logical document must decode to an equal Map through every format. + Map expected = new HashMap<>(); + expected.put("count", 7); + expected.put("ok", true); + + byte[] text = utf8("{\"count\":7,\"ok\":true}"); + byte[] pg = postgres("{\"count\":7,\"ok\":true}"); + byte[] smile = smile(expected); + byte[] cbor = new ObjectMapper(new CBORFactory()).writeValueAsBytes(expected); + // object(size 12, descriptor 12){ text("count"), int("7"), text("ok"), true } + byte[] sqlite = bytes(0xCC, 0x0C, 0x57, 0x63, 0x6F, 0x75, 0x6E, 0x74, 0x13, 0x37, 0x27, 0x6F, 0x6B, 0x01); + + assertEquals(parse(JsonPayloadFormat.TEXT.getParser(), text), expected); + assertEquals(parse(JsonPayloadFormat.POSTGRES_JSONB.getParser(), pg), expected); + assertEquals(parse(JsonPayloadFormat.SMILE.getParser(), smile), expected); + assertEquals(parse(JsonPayloadFormat.CBOR.getParser(), cbor), expected); + assertEquals(parse(JsonPayloadFormat.SQLITE_JSONB.getParser(), sqlite), expected); + } + + @Test + public void testAutoParserDelegates() + throws Exception { + JsonPayloadParser auto = JsonPayloadFormat.AUTO.getParser(); + assertTrue(auto.matches(bytes(0x00), 0, 1)); + assertEquals(parse(auto, postgres("{\"a\":1}")), Map.of("a", 1)); + assertEquals(parse(auto, utf8("{\"a\":1}")), Map.of("a", 1)); + assertEquals(parse(auto, smile(Map.of("a", 1))), Map.of("a", 1)); + } +}