Copilot commented on code in PR #19223:
URL: https://github.com/apache/pinot/pull/19223#discussion_r3769293921
##########
pinot-common/src/main/java/org/apache/pinot/common/function/scalar/JsonFunctions.java:
##########
@@ -129,6 +130,25 @@ private static Object fastJsonPath(Object object, String
jsonPath, boolean useBi
return jsonPath(object, jsonPath);
}
+ /// Resolves a simple path over a JSON string with Fory's streaming reader.
Complex paths, already-parsed inputs,
+ /// container results, and Fory extraction failures use the existing Jayway
implementation so the experimental
+ /// functions below retain the current behavior outside Fory's supported
envelope.
+ @Nullable
+ private static Object foryJsonPath(Object object, String jsonPath) {
+ SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(jsonPath);
+ if (!(object instanceof String) || simpleJsonPath == null) {
+ return jsonPath(object, jsonPath);
+ }
Review Comment:
`foryJsonPath()` compiles `SimpleJsonPath` even when the input is not a
`String` (e.g., already-parsed Map/List input), then immediately falls back to
Jayway. This adds unnecessary work on the fallback path and is inconsistent
with `fastJsonPath()` which only compiles for String inputs.
##########
pinot-common/src/main/java/org/apache/pinot/common/function/ForyJsonPathExtractor.java:
##########
@@ -0,0 +1,439 @@
+/**
+ * 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.common.function;
+
+import com.fasterxml.jackson.core.StreamReadConstraints;
+import java.util.concurrent.atomic.AtomicBoolean;
+import javax.annotation.Nullable;
+import org.apache.fory.json.ForyJson;
+import org.apache.fory.json.codec.JsonValueCodec;
+import org.apache.fory.json.reader.JsonReader;
+import org.apache.fory.json.reader.Latin1JsonReader;
+import org.apache.fory.json.reader.Utf16JsonReader;
+import org.apache.fory.json.reader.Utf8JsonReader;
+import org.apache.fory.json.writer.StringJsonWriter;
+import org.apache.fory.json.writer.Utf8JsonWriter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Shared Fory JSON parser for opt-in JSON-path implementations.
+///
+/// Streaming extraction uses one single-state parser per worker thread to
avoid shared-pool contention. Initialization
+/// or runtime linkage failures permanently disable the optional path,
allowing callers to fall back to
+/// Jackson/Jayway. Jackson's default scalar token limits are enforced while
walking the document, and inputs outside
+/// Fory's safe nesting depth fall back to the reference parser.
+public final class ForyJsonPathExtractor {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ForyJsonPathExtractor.class);
+ private static final StreamReadConstraints JACKSON_CONSTRAINTS =
StreamReadConstraints.defaults();
+ private static final ThreadLocal<PathContext> PATH_CONTEXT =
ThreadLocal.withInitial(PathContext::new);
+ private static final AtomicBoolean UNAVAILABLE_WARNING_LOGGED = new
AtomicBoolean();
+
+ private ForyJsonPathExtractor() {
+ }
+
+ private static final class Holder {
+ private static volatile boolean _streamingAvailable;
+ private static final ThreadLocal<ForyJson> STREAMING_PARSER =
ThreadLocal.withInitial(() -> {
+ if (!_streamingAvailable) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ ForyJson parser = buildStreamingParser();
+ if (parser == null) {
+ _streamingAvailable = false;
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ return parser;
+ });
+
+ static {
+ ForyJson parser = buildStreamingParser();
+ _streamingAvailable = parser != null;
+ if (parser != null) {
+ STREAMING_PARSER.set(parser);
+ }
+ }
+
+ private Holder() {
+ }
+
+ @Nullable
+ private static ForyJson buildStreamingParser() {
+ try {
+ return
ForyJson.builder().withCodegen(false).withAsyncCompilation(false).withConcurrencyLevel(1)
+ .registerCodec(PathResult.class, PathCodec.INSTANCE).build();
+ } catch (RuntimeException | LinkageError e) {
+ logUnavailable(e);
+ return null;
+ }
+ }
+ }
+
+ /// Returns whether the optional Fory runtime initialized successfully.
+ public static boolean isAvailable() {
+ return Holder._streamingAvailable;
+ }
+
+ /// Extracts a simple path with Fory's streaming reader without
materializing the complete JSON tree.
+ ///
+ /// Unrelated values are still fully consumed so malformed input and
duplicate-key last-wins behavior match the
+ /// reference parser. Jackson's nesting, field-name, string, and number
limits are checked while scanning. Callers
+ /// should retry with the reference parser when this method throws.
+ @Nullable
+ public static Object extract(String json, SimpleJsonPath path) {
+ if ((JACKSON_CONSTRAINTS.hasMaxDocumentLength()
+ && json.length() > JACKSON_CONSTRAINTS.getMaxDocumentLength())
+ || (JACKSON_CONSTRAINTS.hasMaxTokenCount() &&
requiresJacksonFallback(json))) {
+ throw new IllegalArgumentException("JSON document requires Jackson
constraint validation");
+ }
Review Comment:
`requiresJacksonFallback(json)` also checks max nesting depth / token sizes,
but it is only invoked when `JACKSON_CONSTRAINTS.hasMaxTokenCount()` is true.
That makes the nesting-depth and length checks inside `requiresJacksonFallback`
effectively unreachable under the default Jackson constraints configuration,
and contradicts the class comment about enforcing Jackson constraints while
scanning.
##########
pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonExtractScalarQuery.java:
##########
@@ -0,0 +1,218 @@
+/**
+ * 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.perf;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.request.context.LiteralContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.operator.ColumnContext;
+import org.apache.pinot.core.operator.blocks.ValueBlock;
+import org.apache.pinot.core.operator.transform.TransformResultMetadata;
+import org.apache.pinot.core.operator.transform.function.BaseTransformFunction;
+import
org.apache.pinot.core.operator.transform.function.JsonExtractScalarTransformFunction;
+import
org.apache.pinot.core.operator.transform.function.LiteralTransformFunction;
+import org.apache.pinot.core.operator.transform.function.TransformFunction;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+
+/// Measures the actual query-side `jsonExtractScalar*` ValueBlock loop. Each
invocation processes a 128-row block;
+/// JMH normalizes throughput and allocation to one row via
[OperationsPerInvocation]. Input projection is represented
+/// by a pre-materialized String array so the comparison isolates JSON
extraction, type coercion, and result writing.
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 3)
+@State(Scope.Thread)
+public class BenchmarkJsonExtractScalarQuery {
+ private static final int BLOCK_ROWS = 128;
+ private static final TransformResultMetadata STRING_METADATA =
+ new TransformResultMetadata(DataType.STRING, true, false);
+ private static final String BASE_JSON = "{"
+ + "\"earlyMetric\":17,"
+ +
"\"user\":{\"id\":\"u-19283\",\"country\":\"US\",\"tier\":\"gold\",\"age\":41},"
+ +
"\"event\":{\"name\":\"checkout\",\"cart\":[{\"sku\":\"A1\",\"qty\":2,\"price\":19.99},"
+ +
"{\"sku\":\"B7\",\"qty\":1,\"price\":129.0},{\"sku\":\"C3\",\"qty\":4,\"price\":3.5}],"
+ + "\"currency\":\"USD\",\"coupon\":null},"
+ +
"\"device\":{\"os\":\"macOS\",\"version\":\"14.5\",\"browser\":\"Chrome\","
+ + "\"screen\":{\"width\":1728,\"height\":1117}},"
+ + "\"geo\":{\"city\":\"San
Francisco\",\"region\":\"CA\",\"country\":\"US\","
+ + "\"coordinates\":[-122.4194,37.7749]},"
+ +
"\"attributes\":{\"campaign\":\"summer-sale\",\"referrer\":\"search\",\"experiment\":\"checkout-v2\"},"
+ + "\"flags\":[\"returning\",\"subscribed\",\"beta\"],"
+ + "\"lateMetric\":19} ";
+
+ @Param({"early", "late"})
+ private String _fieldPosition;
+
+ @Param({"700", "8192", "65536"})
+ private int _documentBytes;
+
+ private ValueBlock _valueBlock;
+ private JsonExtractScalarTransformFunction _jayway;
+ private JsonExtractScalarTransformFunction _fast;
+ private JsonExtractScalarTransformFunction _firstMatch;
+ private JsonExtractScalarTransformFunction _fory;
+
+ @Setup
+ public void setUp() {
+ String json = buildJson(_documentBytes);
+ String[] jsonRows = new String[BLOCK_ROWS];
+ Arrays.fill(jsonRows, json);
+ TransformFunction input = new StringArrayTransformFunction(jsonRows);
+ String path = "early".equals(_fieldPosition) ? "$.earlyMetric" :
"$.lateMetric";
+ List<TransformFunction> arguments = List.of(input, literal(path),
literal("LONG"));
+
+ _valueBlock = new FixedValueBlock(BLOCK_ROWS);
+ _jayway = new JsonExtractScalarTransformFunction();
+ _fast = new JsonExtractScalarTransformFunction.Fast();
+ _firstMatch = new JsonExtractScalarTransformFunction.FirstMatch();
+ _fory = new JsonExtractScalarTransformFunction.Fory();
+ for (JsonExtractScalarTransformFunction function : List.of(_jayway, _fast,
_firstMatch, _fory)) {
+ function.init(arguments, Map.<String, ColumnContext>of(), false);
+ long expected = "early".equals(_fieldPosition) ? 17L : 19L;
+ long[] values = function.transformToLongValuesSV(_valueBlock);
+ if (values.length < BLOCK_ROWS || values[0] != expected ||
values[BLOCK_ROWS - 1] != expected) {
+ throw new IllegalStateException(function.getName() + " produced an
unexpected query result");
+ }
+ }
+ }
+
+ private static LiteralTransformFunction literal(String value) {
+ return new LiteralTransformFunction(new LiteralContext(DataType.STRING,
value));
+ }
+
+ private static String buildJson(int targetBytes) {
+ if (BASE_JSON.length() >= targetBytes) {
+ return BASE_JSON;
+ }
+ String marker = "\"lateMetric\":";
+ int markerOffset = BASE_JSON.indexOf(marker);
+ String paddingPrefix = "\"padding\":\"";
+ String paddingSuffix = "\",";
+ int paddingLength = targetBytes - BASE_JSON.length() -
paddingPrefix.length() - paddingSuffix.length();
+ return BASE_JSON.substring(0, markerOffset) + paddingPrefix +
"x".repeat(paddingLength) + paddingSuffix
+ + BASE_JSON.substring(markerOffset);
Review Comment:
`buildJson(int targetBytes)` can compute a negative `paddingLength` when
`targetBytes` is slightly larger than `BASE_JSON.length()` but smaller than
`BASE_JSON.length() + paddingPrefix.length() + paddingSuffix.length()`, causing
`"x".repeat(paddingLength)` to throw. `BenchmarkJsonPathExtraction#buildJson`
already guards this case; this benchmark should do the same to keep the helper
robust if the base JSON changes or params are adjusted.
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java:
##########
@@ -72,6 +74,10 @@
/// keys to the first non-null occurrence and does not validate malformed
content after the resolved value. Use it
/// only for well-formed, duplicate-free JSON. `Fast` scans the full root
value and retains Jayway's last-key-wins
/// and malformed-document behavior; see [FastJsonPathExtractor] for one
documented unaddressed-value edge case.
+/// `jsonExtractScalarFory` is experimental and must be selected explicitly.
It accelerates simple paths over
+/// `STRING` input for scalar result types other than `STRING` and
`BIG_DECIMAL`. `BYTES` input, complex paths,
+/// containers / array result types, precision-sensitive results, deeply
nested documents, and Fory failures use
+/// Jayway. Its name, supported envelope, and implementation can change while
the integration is evaluated.
Review Comment:
The Javadoc says `jsonExtractScalarFory` accelerates scalar result types
other than `STRING` and `BIG_DECIMAL`, but the implementation enables the Fory
extractor for `STRING` results as well (it only gates on `useBigDecimal`). This
makes the doc misleading for callers trying to understand when the Fory path is
actually used.
--
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]
