xiangfu0 commented on code in PR #18953:
URL: https://github.com/apache/pinot/pull/18953#discussion_r3556923567
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java:
##########
@@ -20,34 +20,50 @@
import java.util.Map;
import java.util.Set;
+import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadFormat;
+import org.apache.pinot.plugin.inputformat.json.format.JsonPayloadParser;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordExtractor;
import org.apache.pinot.spi.plugin.PluginManager;
import org.apache.pinot.spi.stream.StreamMessageDecoder;
-import org.apache.pinot.spi.utils.JsonUtils;
/**
* An implementation of StreamMessageDecoder to read JSON records from a
stream.
+ *
+ * <p>Set the {@value #JSON_FORMAT_CONFIG_KEY} decoder property to pin the
payload encoding to one of
+ * {@code TEXT}, {@code POSTGRES_JSONB}, {@code SQLITE_JSONB}, {@code SMILE}
or {@code CBOR}.
+ *
+ * <p>When unset (equivalently {@code AUTO}) the encoding is detected per
message from its leading magic /
+ * version bytes, falling back to text JSON. Detection is allocation-free and
cannot mis-route a well-formed
+ * text JSON document: a top-level {@code {} / {@code [} (optionally after
whitespace) collides with none of the
+ * binary signatures. Pin {@code TEXT} to skip detection entirely. See {@link
JsonPayloadFormat}.
Review Comment:
Fixed in `4b7401f` — replaced with `<code>{</code>` / `<code>[</code>`
HTML entities so the inline tag no longer terminates early; verified with `mvn
javadoc:javadoc`.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormat.java:
##########
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.json.format;
+
+import java.util.Locale;
+import javax.annotation.Nullable;
+
+
+/// The payload encodings the JSON stream decoder understands. Selected
through the `jsonFormat` decoder
+/// property; [#AUTO] (the default) sniffs each message's leading bytes.
+public enum JsonPayloadFormat {
+ /// Detect the encoding per message from its leading magic / version bytes,
falling back to text JSON.
+ AUTO(new AutoDetectPayloadParser()),
+ /// UTF-8 text JSON (the historical default).
+ TEXT(new TextJsonPayloadParser()),
+ /// PostgreSQL `jsonb` binary wire format (version byte + text JSON).
+ POSTGRES_JSONB(new PostgresJsonbPayloadParser()),
+ /// SQLite 3.45+ JSONB binary format.
+ SQLITE_JSONB(new SqliteJsonbPayloadParser()),
+ /// Jackson Smile binary JSON.
+ SMILE(new SmileJsonPayloadParser()),
+ /// CBOR (RFC 8949).
+ CBOR(new CborJsonPayloadParser());
+
+ // AUTO detection precedence: formats with strong, unambiguous magic bytes
first; text (the catch-all,
+ // detected only by a leading '{' / '[') last so it never shadows a binary
format. POSTGRES_JSONB precedes
+ // TEXT because its body *is* text JSON behind a version byte.
+ private static final JsonPayloadFormat[] DETECTION_ORDER =
+ {SMILE, CBOR, POSTGRES_JSONB, SQLITE_JSONB, TEXT};
+
+ private final JsonPayloadParser _parser;
+
+ JsonPayloadFormat(JsonPayloadParser parser) {
+ _parser = parser;
+ }
+
+ /// The parser for this format. For [#AUTO] this is a parser that resolves
the concrete format per message.
+ public JsonPayloadParser getParser() {
+ return _parser;
+ }
+
+ /// Resolves a configured format name (case-insensitive), returning [#AUTO]
when unset.
+ public static JsonPayloadFormat fromConfig(@Nullable String value) {
+ if (value == null || value.trim().isEmpty()) {
+ return AUTO;
+ }
+ try {
+ return valueOf(value.trim().toUpperCase(Locale.ROOT));
+ } catch (IllegalArgumentException e) {
+ throw new IllegalArgumentException(
+ "Unsupported jsonFormat '" + value + "'. Supported values: AUTO,
TEXT, POSTGRES_JSONB, SQLITE_JSONB, "
+ + "SMILE, CBOR");
+ }
Review Comment:
Fixed in `4b7401f` — now `throw new IllegalArgumentException("Unsupported
jsonFormat ...", e)`, and `ed03fea` adds a test asserting the cause survives
(the previous `assertThrows` passed even without it).
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderBinaryTest.java:
##########
@@ -0,0 +1,172 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.json;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
+import com.fasterxml.jackson.dataformat.cbor.CBORGenerator;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertThrows;
+
+
+/// End-to-end coverage of {@link JSONMessageDecoder} decoding each configured
/ auto-detected payload format
+/// through the shared {@link JSONRecordExtractor} into a {@link GenericRow}.
+public class JSONMessageDecoderBinaryTest {
+
+ private static final Set<String> RICH_FIELDS = Set.of("name", "count",
"ratio");
+ private static final Set<String> SINGLE_FIELD = Set.of("a");
+
+ private static byte[] smile(Map<String, Object> value)
+ throws Exception {
+ return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value);
+ }
+
+ private static byte[] cbor(Map<String, Object> value)
+ throws Exception {
+ CBORFactory factory = new CBORFactory();
+ factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER);
+ return new ObjectMapper(factory).writeValueAsBytes(value);
+ }
+
+ private static byte[] bytes(int... values) {
+ byte[] result = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (byte) values[i];
+ }
+ return result;
+ }
+
+ // Hand-built {"a": 1} fixtures matching the parser unit tests.
+ // SQLite: object(size 4){ text("a"), int("1") }.
+ private static final byte[] SQLITE_A1 = bytes(0x4C, 0x17, 0x61, 0x13, 0x31);
+ // PostgreSQL jsonb_send framing: version byte 1 followed by the text JSON
body.
+ private static final byte[] POSTGRES_A1 =
+ bytes(0x01, '{', '"', 'a', '"', ':', '1', '}');
+
+ private GenericRow decode(Map<String, String> props, Set<String> fields,
byte[] payload)
+ throws Exception {
+ JSONMessageDecoder decoder = new JSONMessageDecoder();
+ decoder.init(props, fields, "topic");
+ return decoder.decode(payload, new GenericRow());
+ }
+
+ private void assertRich(GenericRow row) {
+ assertEquals(row.getValue("name"), "pinot");
+ assertEquals(row.getValue("count"), 7);
+ assertEquals(row.getValue("ratio"), 2.5);
+ }
+
+ private Map<String, Object> richDoc() {
+ return Map.of("name", "pinot", "count", 7, "ratio", 2.5);
+ }
+
+ @Test
+ public void testDefaultIsText()
+ throws Exception {
+ assertRich(decode(Map.of(), RICH_FIELDS,
"{\"name\":\"pinot\",\"count\":7,\"ratio\":2.5}".getBytes()));
+ }
Review Comment:
Fixed in `4b7401f` — renamed to `testUnsetFormatAutoDetectsText`, with a doc
comment noting the default is AUTO rather than a pinned TEXT format.
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
##########
pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/format/JsonPayloadFormatTest.java:
##########
@@ -0,0 +1,422 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.plugin.inputformat.json.format;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.cbor.CBORFactory;
+import com.fasterxml.jackson.dataformat.cbor.CBORGenerator;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+public class JsonPayloadFormatTest {
+
+ private static byte[] bytes(int... values) {
+ byte[] result = new byte[values.length];
+ for (int i = 0; i < values.length; i++) {
+ result[i] = (byte) values[i];
+ }
+ return result;
+ }
+
+ /// PostgreSQL `jsonb_send` framing: version byte 1 followed by the text
JSON body.
+ private static byte[] postgres(String json) {
+ byte[] body = json.getBytes(StandardCharsets.UTF_8);
+ byte[] result = new byte[body.length + 1];
+ result[0] = 1;
+ System.arraycopy(body, 0, result, 1, body.length);
+ return result;
+ }
+
+ private static byte[] smile(Object value)
+ throws Exception {
+ return new ObjectMapper(new SmileFactory()).writeValueAsBytes(value);
+ }
+
+ private static byte[] cborSelfDescribed(Object value)
+ throws Exception {
+ CBORFactory factory = new CBORFactory();
+ factory.enable(CBORGenerator.Feature.WRITE_TYPE_HEADER);
+ return new ObjectMapper(factory).writeValueAsBytes(value);
+ }
+
+ private static Map<String, Object> parse(JsonPayloadParser parser, byte[]
payload)
+ throws Exception {
+ return parser.parse(payload, 0, payload.length);
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // Config parsing
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testFromConfig() {
+ assertEquals(JsonPayloadFormat.fromConfig(null), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(""), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig(" "), JsonPayloadFormat.AUTO);
+ assertEquals(JsonPayloadFormat.fromConfig("text"), JsonPayloadFormat.TEXT);
+ assertEquals(JsonPayloadFormat.fromConfig(" Smile "),
JsonPayloadFormat.SMILE);
+ assertEquals(JsonPayloadFormat.fromConfig("POSTGRES_JSONB"),
JsonPayloadFormat.POSTGRES_JSONB);
+ // AUTO resolves the concrete format per message, so every format has a
non-null parser.
+ assertSame(JsonPayloadFormat.AUTO.getParser().getClass(),
AutoDetectPayloadParser.class);
+ assertThrows(IllegalArgumentException.class, () ->
JsonPayloadFormat.fromConfig("bson"));
+ }
+
+ //
------------------------------------------------------------------------------------------------------
+ // Text
+ //
------------------------------------------------------------------------------------------------------
+
+ @Test
+ public void testText()
+ throws Exception {
+ JsonPayloadParser parser = JsonPayloadFormat.TEXT.getParser();
+ Map<String, Object> map = parse(parser, " {\"a\": 1, \"b\":
\"x\"}".getBytes());
+ assertEquals(map.get("a"), 1);
+ assertEquals(map.get("b"), "x");
+ assertTrue(parser.matches("{\"a\":1}".getBytes(), 0, 7));
+ assertTrue(parser.matches(" [1,2]".getBytes(), 0, 7));
Review Comment:
Fixed in `4b7401f` — added a `utf8(String)` helper and routed every fixture
through it; `JSONMessageDecoderBinaryTest` uses a `StandardCharsets.UTF_8`
`TEXT_DOC` constant. (Left the pre-existing `line.getBytes()` in
`JSONMessageDecoderTest`, untouched by this PR.)
_🤖 Addressed by [Claude Code](https://claude.com/claude-code)_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
