yashmayya commented on code in PR #19223:
URL: https://github.com/apache/pinot/pull/19223#discussion_r3770314522
##########
pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java:
##########
@@ -802,7 +817,7 @@ private <T> IntFunction<T> getResultExtractor(ValueBlock
valueBlock, ParseContex
if (_jsonFieldTransformFunction.getResultMetadata().getDataType() ==
DataType.BYTES) {
byte[][] jsonBytes =
_jsonFieldTransformFunction.transformToBytesValuesSV(valueBlock);
IntFunction<T> jaywayExtractor = i ->
parseContext.parseUtf8(jsonBytes[i]).read(_jsonPath);
- if (_simpleJsonPath == null) {
+ if (_simpleJsonPath == null || useBigDecimal || _extractionMode ==
ExtractionMode.FORY) {
Review Comment:
**Blocker.** This line changes the two existing functions, not only the new
one.
Before this PR, BYTES input with a `BIG_DECIMAL` result type used
`FastJsonPathExtractor`. The new `useBigDecimal` condition sends that case to
Jayway instead.
The STRING branch at line 868 still passes `useBigDecimal` to the fast
extractor. This difference between the two branches looks accidental.
To gate only Fory, write `_simpleJsonPath == null || _extractionMode ==
ExtractionMode.FORY`. If the change is intentional, please add a test and a
note in the PR description.
##########
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(() -> {
Review Comment:
**Blocker.** Each thread that calls this function keeps its own `ForyJson`
instance forever.
I measured the live heap on your branch. 64 threads added 14.28 MB, which is
about 220 KB for each thread. A control group of 64 idle threads added 0.53 MB.
This ThreadLocal is never removed. A Pinot server has many query threads,
and ingestion adds one thread for each consuming partition. The total reaches
tens of megabytes for an experimental function.
`ForyJson` is already thread-safe and pooled through `withConcurrencyLevel`.
One shared instance, with a concurrency level that matches the thread pool,
avoids this multiplication. The comment above says that a shared pool causes
contention. The PR gives no measurement for that claim.
The static block at line 69 also calls `STREAMING_PARSER.set(parser)`. This
pins one parser to the arbitrary thread that loads the class first, and that
thread can be a Helix or ZooKeeper callback thread that never uses it.
##########
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");
+ }
+ if (!Holder._streamingAvailable) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ ForyJson parser = Holder.STREAMING_PARSER.get();
+ PathContext context = PATH_CONTEXT.get();
+ if (context._active) {
+ throw new IllegalStateException("Fory JSON path extraction is not
reentrant");
+ }
+ context._active = true;
+ context._path = path;
+ context._result = null;
+ try {
+ parser.fromJson(json, PathResult.class);
+ return context._result;
+ } catch (LinkageError e) {
+ disable();
+ logUnavailable(e);
+ throw new IllegalStateException("Fory JSON became unavailable", e);
+ } finally {
+ context._path = null;
+ context._result = null;
+ context._active = false;
+ }
+ }
+
+ private static void disable() {
+ Holder._streamingAvailable = false;
+ Holder.STREAMING_PARSER.remove();
+ PATH_CONTEXT.remove();
+ }
+
+ private static void logUnavailable(Throwable cause) {
+ if (UNAVAILABLE_WARNING_LOGGED.compareAndSet(false, true)) {
+ LOGGER.warn("Experimental Fory JSON support is unavailable; falling back
to Jackson/Jayway", cause);
+ }
+ }
+
+ private static Object readPath(JsonReader reader, SimpleJsonPath path, int
depth) {
+ String key = path.getKey(depth);
+ if (key != null) {
+ return readObjectPath(reader, path, depth, key);
+ }
+ return readArrayPath(reader, path, depth, path.getIndex(depth));
+ }
+
+ @Nullable
+ private static Object readObjectPath(JsonReader reader, SimpleJsonPath path,
int depth, String expectedKey) {
+ if (reader.peekToken() != '{') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('{');
+ if (reader.consume('}')) {
+ return null;
+ }
+ Object result = null;
+ boolean more;
+ do {
+ String fieldName = reader.readFieldName();
+ if (fieldName.length() > JACKSON_CONSTRAINTS.getMaxNameLength()) {
+ throw new IllegalArgumentException("JSON field name exceeds
Jackson's configured limit");
+ }
+ reader.expect(':');
+ if (expectedKey.equals(fieldName)) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readArrayPath(JsonReader reader, SimpleJsonPath path,
int depth, int expectedIndex) {
+ if (reader.peekToken() != '[') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('[');
+ if (reader.consume(']')) {
+ return null;
+ }
+ Object result = null;
+ int index = 0;
+ boolean more;
+ do {
+ if (index == expectedIndex) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ index++;
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readScalar(JsonReader reader) {
+ char token = reader.peekToken();
+ if (token == '"') {
+ String value = reader.readString();
+ if (value.length() > JACKSON_CONSTRAINTS.getMaxStringLength()) {
+ throw new IllegalArgumentException("JSON string exceeds Jackson's
configured limit");
+ }
+ return value;
+ }
+ if (token == 't' || token == 'f') {
+ return reader.readBoolean();
+ }
+ if (token == 'n') {
+ reader.readNull();
+ return null;
+ }
+ if (token == '{' || token == '[') {
+ // Query scalar coercion has observable error/default behavior for
containers. Let Jayway produce the exact
+ // reference value rather than materializing a Fory container on this
uncommon path.
+ throw new IllegalArgumentException("Container result requires reference
JSON parsing");
+ }
+ int start = reader.position();
+ Number value = reader.readNumber();
+ if (reader.position() - start > JACKSON_CONSTRAINTS.getMaxNumberLength()) {
+ throw new IllegalArgumentException("JSON number exceeds Jackson's
configured limit");
+ }
+ return value;
+ }
+
+ private static void skipValue(JsonReader reader) {
+ char token = reader.peekToken();
+ if (token == '{') {
+ skipObject(reader);
+ return;
+ }
+ if (token == '[') {
+ skipArray(reader);
+ return;
+ }
+ if (token == '"') {
+ // Fory 1.6's skipValue() computes an FNV hash over every character. Its
string decoder uses packed scans and
+ // is substantially faster even when the decoded value is discarded. An
upstream fast-skip API could remove
+ // this temporary allocation in a future Fory version.
+ String value = reader.readString();
+ if (value.length() > JACKSON_CONSTRAINTS.getMaxStringLength()) {
+ throw new IllegalArgumentException("JSON string exceeds Jackson's
configured limit");
+ }
+ return;
+ }
+ int start = reader.position();
+ reader.skipValue();
+ int rawLength = reader.position() - start;
+ if (token != 't' && token != 'f' && token != 'n'
+ && rawLength > JACKSON_CONSTRAINTS.getMaxNumberLength()) {
+ throw new IllegalArgumentException("JSON number exceeds Jackson's
configured limit");
+ }
+ }
+
+ private static void skipObject(JsonReader reader) {
+ reader.enterDepth();
+ try {
+ reader.expect('{');
+ if (reader.consume('}')) {
+ return;
+ }
+ boolean more;
+ do {
+ String fieldName = reader.readFieldName();
+ if (fieldName.length() > JACKSON_CONSTRAINTS.getMaxNameLength()) {
+ throw new IllegalArgumentException("JSON field name exceeds
Jackson's configured limit");
+ }
+ reader.expect(':');
+ skipValue(reader);
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ private static void skipArray(JsonReader reader) {
+ reader.enterDepth();
+ try {
+ reader.expect('[');
+ if (reader.consume(']')) {
+ return;
+ }
+ boolean more;
+ do {
+ skipValue(reader);
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ private static final class PathContext {
+ private final PathResult _marker = new PathResult();
+ private boolean _active;
+ @Nullable
+ private SimpleJsonPath _path;
+ @Nullable
+ private Object _result;
+ }
+
+ private static final class PathResult {
+ }
+
+ private static final class PathCodec implements JsonValueCodec<PathResult> {
+ private static final PathCodec INSTANCE = new PathCodec();
+
+ @Override
+ public PathResult readLatin1(Latin1JsonReader reader) {
+ return read(reader);
+ }
+
+ @Override
+ public PathResult readUtf16(Utf16JsonReader reader) {
+ return read(reader);
+ }
+
+ @Override
+ public PathResult readUtf8(Utf8JsonReader reader) {
+ return read(reader);
+ }
+
+ private static PathResult read(JsonReader reader) {
+ PathContext context = PATH_CONTEXT.get();
+ SimpleJsonPath path = context._path;
+ if (!context._active || path == null) {
+ throw new IllegalStateException("Missing JSON path extraction
context");
+ }
+ context._result = readPath(reader, path, 0);
+ return context._marker;
+ }
+
+ @Override
+ public void writeString(StringJsonWriter writer, PathResult value) {
+ throw new UnsupportedOperationException("PathResult is read-only");
+ }
+
+ @Override
+ public void writeUtf8(Utf8JsonWriter writer, PathResult value) {
+ throw new UnsupportedOperationException("PathResult is read-only");
+ }
+ }
+
+ private static boolean requiresJacksonFallback(String json) {
Review Comment:
This method never runs with the default Jackson configuration.
The caller at line 102 invokes it only when `hasMaxTokenCount()` is true. I
printed the defaults on JDK 25.0.3 with Jackson 2.22.1:
```
hasMaxDocumentLength=false maxDocumentLength=-1
hasMaxTokenCount=false maxTokenCount=-1
maxNestingDepth=1000 maxNameLength=50000 maxStringLength=20000000
maxNumberLength=1000
```
Both parts of the guard at line 100 are always false. These 70 lines are
dead code, and they explain the 50.9% patch coverage that Codecov reports for
this file.
The class Javadoc also says that Jackson's nesting limit is enforced while
the parser walks the document. It is not. Only Fory's limit of 20 applies.
Please delete this method, or read the constraints from the same
`JsonFactory` that the Jayway path uses, so that the guard becomes reachable
and testable.
##########
pinot-common/pom.xml:
##########
@@ -280,6 +280,10 @@
<groupId>com.jayway.jsonpath</groupId>
<artifactId>json-path</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
Review Comment:
**Blocker.** This adds a hard dependency to `pinot-common`. Every Pinot
module and every shaded jar then contains `fory-json` and `fory-core`, which
are about 3.6 MB together.
The PR describes all four functions as experimental, and says that they can
change or be removed.
`fory-core` also calls a terminally deprecated JDK method. The first call
prints this on JDK 25.0.3:
```
WARNING: A terminally deprecated method in sun.misc.Unsafe has been called
WARNING: sun.misc.Unsafe::staticFieldOffset has been called by
org.apache.fory.platform.internal._Lookup
WARNING: sun.misc.Unsafe::staticFieldOffset will be removed in a future
release
```
Your fallback handles the future removal correctly, because
`buildStreamingParser` catches `RuntimeException` and `LinkageError`. The
warning still reaches the logs of every role.
Please consider a separate optional module for this dependency until the
evaluation is complete. A module keeps the experiment out of the default
distribution.
##########
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");
+ }
+ if (!Holder._streamingAvailable) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ ForyJson parser = Holder.STREAMING_PARSER.get();
+ PathContext context = PATH_CONTEXT.get();
+ if (context._active) {
+ throw new IllegalStateException("Fory JSON path extraction is not
reentrant");
+ }
+ context._active = true;
+ context._path = path;
+ context._result = null;
+ try {
+ parser.fromJson(json, PathResult.class);
+ return context._result;
+ } catch (LinkageError e) {
+ disable();
+ logUnavailable(e);
+ throw new IllegalStateException("Fory JSON became unavailable", e);
+ } finally {
+ context._path = null;
+ context._result = null;
+ context._active = false;
+ }
+ }
+
+ private static void disable() {
+ Holder._streamingAvailable = false;
+ Holder.STREAMING_PARSER.remove();
+ PATH_CONTEXT.remove();
+ }
+
+ private static void logUnavailable(Throwable cause) {
+ if (UNAVAILABLE_WARNING_LOGGED.compareAndSet(false, true)) {
+ LOGGER.warn("Experimental Fory JSON support is unavailable; falling back
to Jackson/Jayway", cause);
+ }
+ }
+
+ private static Object readPath(JsonReader reader, SimpleJsonPath path, int
depth) {
+ String key = path.getKey(depth);
+ if (key != null) {
+ return readObjectPath(reader, path, depth, key);
+ }
+ return readArrayPath(reader, path, depth, path.getIndex(depth));
+ }
+
+ @Nullable
+ private static Object readObjectPath(JsonReader reader, SimpleJsonPath path,
int depth, String expectedKey) {
+ if (reader.peekToken() != '{') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('{');
+ if (reader.consume('}')) {
+ return null;
+ }
+ Object result = null;
+ boolean more;
+ do {
+ String fieldName = reader.readFieldName();
+ if (fieldName.length() > JACKSON_CONSTRAINTS.getMaxNameLength()) {
+ throw new IllegalArgumentException("JSON field name exceeds
Jackson's configured limit");
+ }
+ reader.expect(':');
+ if (expectedKey.equals(fieldName)) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readArrayPath(JsonReader reader, SimpleJsonPath path,
int depth, int expectedIndex) {
+ if (reader.peekToken() != '[') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('[');
+ if (reader.consume(']')) {
+ return null;
+ }
+ Object result = null;
+ int index = 0;
+ boolean more;
+ do {
+ if (index == expectedIndex) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ index++;
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readScalar(JsonReader reader) {
+ char token = reader.peekToken();
+ if (token == '"') {
+ String value = reader.readString();
+ if (value.length() > JACKSON_CONSTRAINTS.getMaxStringLength()) {
+ throw new IllegalArgumentException("JSON string exceeds Jackson's
configured limit");
+ }
+ return value;
+ }
+ if (token == 't' || token == 'f') {
+ return reader.readBoolean();
+ }
+ if (token == 'n') {
+ reader.readNull();
+ return null;
+ }
+ if (token == '{' || token == '[') {
+ // Query scalar coercion has observable error/default behavior for
containers. Let Jayway produce the exact
+ // reference value rather than materializing a Fory container on this
uncommon path.
+ throw new IllegalArgumentException("Container result requires reference
JSON parsing");
+ }
+ int start = reader.position();
+ Number value = reader.readNumber();
+ if (reader.position() - start > JACKSON_CONSTRAINTS.getMaxNumberLength()) {
+ throw new IllegalArgumentException("JSON number exceeds Jackson's
configured limit");
+ }
+ return value;
+ }
+
+ private static void skipValue(JsonReader reader) {
+ char token = reader.peekToken();
+ if (token == '{') {
+ skipObject(reader);
+ return;
+ }
+ if (token == '[') {
+ skipArray(reader);
+ return;
+ }
+ if (token == '"') {
+ // Fory 1.6's skipValue() computes an FNV hash over every character. Its
string decoder uses packed scans and
+ // is substantially faster even when the decoded value is discarded. An
upstream fast-skip API could remove
+ // this temporary allocation in a future Fory version.
+ String value = reader.readString();
+ if (value.length() > JACKSON_CONSTRAINTS.getMaxStringLength()) {
+ throw new IllegalArgumentException("JSON string exceeds Jackson's
configured limit");
+ }
+ return;
+ }
+ int start = reader.position();
+ reader.skipValue();
+ int rawLength = reader.position() - start;
+ if (token != 't' && token != 'f' && token != 'n'
+ && rawLength > JACKSON_CONSTRAINTS.getMaxNumberLength()) {
+ throw new IllegalArgumentException("JSON number exceeds Jackson's
configured limit");
+ }
+ }
+
+ private static void skipObject(JsonReader reader) {
+ reader.enterDepth();
Review Comment:
`skipObject` and `skipArray` also increase the depth counter. So a deep
subtree that the path never enters still disables the fast path for the whole
row.
`ForyJson.DEFAULT_MAX_DEPTH` is 20, and the builder at line 79 does not
change it. Jayway allows 1000 levels.
I verified this on your branch. The document `{"a":1,"b":<21 nested
objects>}` with path `$.a` throws `ForyJsonException: JSON max depth 20
exceeded at JSON position 106`.
A limit of 20 is low for real JSON, and one deep field anywhere in the
document removes the benefit for every row. Please raise the limit with the
builder, or write the number 20 in the class Javadoc.
##########
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
Review Comment:
This paragraph does not match the code. `getResultExtractor` gates only
`useBigDecimal`. `STRING` and `JSON` result types both use Fory.
For a `STRING` result type the fast path works and is faster, so the text is
too modest there. For a `JSON` result type the value is usually an object or an
array, so Fory throws for each row and Jayway parses the document a second time.
##########
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");
+ }
+ if (!Holder._streamingAvailable) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ ForyJson parser = Holder.STREAMING_PARSER.get();
+ PathContext context = PATH_CONTEXT.get();
+ if (context._active) {
+ throw new IllegalStateException("Fory JSON path extraction is not
reentrant");
+ }
+ context._active = true;
+ context._path = path;
+ context._result = null;
+ try {
+ parser.fromJson(json, PathResult.class);
+ return context._result;
+ } catch (LinkageError e) {
+ disable();
+ logUnavailable(e);
+ throw new IllegalStateException("Fory JSON became unavailable", e);
+ } finally {
+ context._path = null;
+ context._result = null;
+ context._active = false;
+ }
+ }
+
+ private static void disable() {
+ Holder._streamingAvailable = false;
+ Holder.STREAMING_PARSER.remove();
+ PATH_CONTEXT.remove();
+ }
+
+ private static void logUnavailable(Throwable cause) {
+ if (UNAVAILABLE_WARNING_LOGGED.compareAndSet(false, true)) {
+ LOGGER.warn("Experimental Fory JSON support is unavailable; falling back
to Jackson/Jayway", cause);
+ }
+ }
+
+ private static Object readPath(JsonReader reader, SimpleJsonPath path, int
depth) {
+ String key = path.getKey(depth);
+ if (key != null) {
+ return readObjectPath(reader, path, depth, key);
+ }
+ return readArrayPath(reader, path, depth, path.getIndex(depth));
+ }
+
+ @Nullable
+ private static Object readObjectPath(JsonReader reader, SimpleJsonPath path,
int depth, String expectedKey) {
+ if (reader.peekToken() != '{') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('{');
+ if (reader.consume('}')) {
+ return null;
+ }
+ Object result = null;
+ boolean more;
+ do {
+ String fieldName = reader.readFieldName();
+ if (fieldName.length() > JACKSON_CONSTRAINTS.getMaxNameLength()) {
+ throw new IllegalArgumentException("JSON field name exceeds
Jackson's configured limit");
+ }
+ reader.expect(':');
+ if (expectedKey.equals(fieldName)) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readArrayPath(JsonReader reader, SimpleJsonPath path,
int depth, int expectedIndex) {
+ if (reader.peekToken() != '[') {
+ skipValue(reader);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ reader.expect('[');
+ if (reader.consume(']')) {
+ return null;
+ }
+ Object result = null;
+ int index = 0;
+ boolean more;
+ do {
+ if (index == expectedIndex) {
+ result = depth + 1 == path.length() ? readScalar(reader) :
readPath(reader, path, depth + 1);
+ } else {
+ skipValue(reader);
+ }
+ index++;
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readScalar(JsonReader reader) {
+ char token = reader.peekToken();
+ if (token == '"') {
+ String value = reader.readString();
+ if (value.length() > JACKSON_CONSTRAINTS.getMaxStringLength()) {
+ throw new IllegalArgumentException("JSON string exceeds Jackson's
configured limit");
+ }
+ return value;
+ }
+ if (token == 't' || token == 'f') {
+ return reader.readBoolean();
+ }
+ if (token == 'n') {
+ reader.readNull();
+ return null;
+ }
+ if (token == '{' || token == '[') {
+ // Query scalar coercion has observable error/default behavior for
containers. Let Jayway produce the exact
+ // reference value rather than materializing a Fory container on this
uncommon path.
+ throw new IllegalArgumentException("Container result requires reference
JSON parsing");
Review Comment:
**Blocker.** The fallback throws a new exception for each row. The Fory
function is then slower than the function that it replaces.
I measured this on your branch with JDK 25.0.3, 200,000 iterations for each
case, best of two runs after warmup:
| document | path | Fory | Jayway | result |
| --- | --- | ---: | ---: | --- |
| `{"a":"v","b":"w"}` | `$.a` | 0.06 us/op | 0.35 us/op | 5.8x faster |
| `{"a":{"b":1},"b":"w"}` | `$.a` | 1.92 us/op | 0.51 us/op | **3.8x
slower** |
| `{"a":[1,2,3],"b":"w"}` | `$.a` | 1.63 us/op | 0.48 us/op | **3.4x
slower** |
| 25-level nested document | `$.k` x25 | 5.24 us/op | 2.19 us/op | **2.4x
slower** |
The cost has two parts. `new IllegalArgumentException(...)` fills in a stack
trace, and then Jayway parses the same document a second time.
Two changes remove most of this cost. First, return a sentinel value from
`readScalar` instead of a new exception, because a container result is normal
control flow and not an error. Second, route `JSON` result types to Jayway once
in `init()`, because `_dataType` is known there.
--
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]
