This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new cb89c0e4c56 Add experimental Fory JSON functions (#19223)
cb89c0e4c56 is described below
commit cb89c0e4c566005059d3b6b97ad306f2e159ae98
Author: Xiang Fu <[email protected]>
AuthorDate: Thu Aug 13 01:02:28 2026 -0700
Add experimental Fory JSON functions (#19223)
* Add experimental Fory JSON functions
* Optimize experimental Fory JSON extraction
* Benchmark Fory JSON function counterparts
* Address Fory JSON review feedback
* Clarify optional Fory runtime requirements
---------
Co-authored-by: Xiang Fu <[email protected]>
---
licenses-binary/LICENSE-kryo.txt | 2 +-
pinot-common/pom.xml | 18 +
.../common/function/ForyJsonPathExtractor.java | 388 +++++++++++++++++++++
.../common/function/TransformFunctionType.java | 2 +
.../common/function/scalar/JsonFunctions.java | 75 ++++
.../org/apache/pinot/sql/parsers/ParserUtils.java | 3 +
.../common/function/FastJsonPathExtractorTest.java | 1 +
.../function/ForyJsonLinkageFallbackTest.java | 143 ++++++++
.../common/function/ForyJsonPathFunctionsTest.java | 298 ++++++++++++++++
pinot-core/pom.xml | 5 +
.../JsonExtractScalarTransformFunction.java | 54 ++-
.../function/TransformFunctionFactory.java | 2 +
.../core/data/function/JsonFunctionsTest.java | 23 ++
.../pinot/core/function/FunctionRegistryTest.java | 2 +-
.../JsonExtractScalarTransformFunctionTest.java | 53 ++-
pinot-integration-tests/pom.xml | 5 +
.../integration/tests/custom/JsonPathTest.java | 47 ++-
.../resources/udf-test-results/all-functions.yaml | 16 +
pinot-perf/pom.xml | 4 +
.../pinot/perf/BenchmarkForyJsonFallback.java | 109 ++++++
.../perf/BenchmarkJsonExtractScalarQuery.java | 351 +++++++++++++++++++
.../pinot/perf/BenchmarkJsonPathExtraction.java | 241 +++++++++++--
.../apache/pinot/query/QueryCompilationTest.java | 6 +-
.../pinot/query/QueryEnvironmentTestBase.java | 2 +
.../query/runtime/queries/QueryRunnerTest.java | 4 +-
pom.xml | 6 +
26 files changed, 1819 insertions(+), 41 deletions(-)
diff --git a/licenses-binary/LICENSE-kryo.txt b/licenses-binary/LICENSE-kryo.txt
index bf531009ca5..a64798f5c0b 100644
--- a/licenses-binary/LICENSE-kryo.txt
+++ b/licenses-binary/LICENSE-kryo.txt
@@ -7,4 +7,4 @@ Redistribution and use in source and binary forms, with or
without modification,
* Redistributions in binary form must reproduce the above copyright notice,
this list of conditions and the following disclaimer in the documentation
and/or other materials provided with the distribution.
* Neither the name of Esoteric Software nor the names of its contributors may
be used to endorse or promote products derived from this software without
specific prior written permission.
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROF [...]
\ No newline at end of file
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
SERVICES; LOSS OF USE, DATA, OR PROF [...]
diff --git a/pinot-common/pom.xml b/pinot-common/pom.xml
index a36aea1d936..539625d7f2e 100644
--- a/pinot-common/pom.xml
+++ b/pinot-common/pom.xml
@@ -98,6 +98,19 @@
</java>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <configuration>
+ <artifactSet>
+ <excludes>
+ <!-- Experimental Fory support is activated only when
applications add Fory to their classpath. -->
+ <exclude>org.apache.fory:fory-json</exclude>
+ <exclude>org.apache.fory:fory-core</exclude>
+ </excludes>
+ </artifactSet>
+ </configuration>
+ </plugin>
<!-- Stage 1 of SQL parser codegen: FMPP plugs Pinot's custom grammar
(config.fmpp +
templates/Parser.jj) into Calcite's parser template and writes the
JavaCC grammar to
target/generated-sources/javacc/Parser.jj. Declared before
javacc-maven-plugin so it runs
@@ -280,6 +293,11 @@
<groupId>com.jayway.jsonpath</groupId>
<artifactId>json-path</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
+ <optional>true</optional>
+ </dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/ForyJsonPathExtractor.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/ForyJsonPathExtractor.java
new file mode 100644
index 00000000000..bfe32acc725
--- /dev/null
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/ForyJsonPathExtractor.java
@@ -0,0 +1,388 @@
+/**
+ * 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 bounded parser pool shared by all worker
threads. Initialization or runtime linkage
+/// failures permanently disable the optional path, allowing callers to fall
back to Jackson/Jayway. Jackson's
+/// document, token, nesting, field-name, string, and number constraints are
enforced while walking the document.
+/// The Fory runtime is an optional dependency; applications must add
`org.apache.fory:fory-json:1.6.0` (and its
+/// transitive `fory-core` dependency) to the application classpath to enable
this experimental path.
+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 static final Object FALLBACK_REQUIRED = new Object();
+
+ private ForyJsonPathExtractor() {
+ }
+
+ private static final class Holder {
+ @Nullable
+ private static final ForyJson STREAMING_PARSER = buildStreamingParser();
+ private static volatile boolean _streamingAvailable = STREAMING_PARSER !=
null;
+
+ private Holder() {
+ }
+
+ @Nullable
+ private static ForyJson buildStreamingParser() {
+ try {
+ return
ForyJson.builder().withCodegen(false).withAsyncCompilation(false)
+ .maxDepth(JACKSON_CONSTRAINTS.getMaxNestingDepth())
+ .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;
+ }
+
+ /// Returns whether an extracted value requires the reference parser for
exact container/coercion semantics.
+ public static boolean isFallbackRequired(@Nullable Object value) {
+ return value == FALLBACK_REQUIRED;
+ }
+
+ /// 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 configured limits are checked while
scanning. Callers should retry with the
+ /// reference parser when this method throws or
[#isFallbackRequired(Object)] returns `true`.
+ @Nullable
+ public static Object extract(String json, SimpleJsonPath path) {
+ if (JACKSON_CONSTRAINTS.hasMaxDocumentLength()
+ && json.length() > JACKSON_CONSTRAINTS.getMaxDocumentLength()) {
+ throw new IllegalArgumentException("JSON document requires Jackson
constraint validation");
+ }
+ if (!Holder._streamingAvailable) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ ForyJson parser = Holder.STREAMING_PARSER;
+ if (parser == null) {
+ throw new IllegalStateException("Fory JSON is unavailable");
+ }
+ 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;
+ context._tokenCount = 0;
+ 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._tokenCount = 0;
+ context._active = false;
+ }
+ }
+
+ private static void disable() {
+ Holder._streamingAvailable = false;
+ 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, PathContext context) {
+ String key = path.getKey(depth);
+ if (key != null) {
+ return readObjectPath(reader, path, depth, key, context);
+ }
+ return readArrayPath(reader, path, depth, path.getIndex(depth), context);
+ }
+
+ @Nullable
+ private static Object readObjectPath(JsonReader reader, SimpleJsonPath path,
int depth, String expectedKey,
+ PathContext context) {
+ if (reader.peekToken() != '{') {
+ skipValue(reader, context);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ countToken(context);
+ reader.expect('{');
+ if (reader.consume('}')) {
+ countToken(context);
+ return null;
+ }
+ Object result = null;
+ boolean more;
+ do {
+ countToken(context);
+ 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, context)
+ : readPath(reader, path, depth + 1, context);
+ } else {
+ skipValue(reader, context);
+ }
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ countToken(context);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readArrayPath(JsonReader reader, SimpleJsonPath path,
int depth, int expectedIndex,
+ PathContext context) {
+ if (reader.peekToken() != '[') {
+ skipValue(reader, context);
+ return null;
+ }
+ reader.enterDepth();
+ try {
+ countToken(context);
+ reader.expect('[');
+ if (reader.consume(']')) {
+ countToken(context);
+ return null;
+ }
+ Object result = null;
+ int index = 0;
+ boolean more;
+ do {
+ if (index == expectedIndex) {
+ result = depth + 1 == path.length() ? readScalar(reader, context)
+ : readPath(reader, path, depth + 1, context);
+ } else {
+ skipValue(reader, context);
+ }
+ index++;
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ countToken(context);
+ return result;
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ @Nullable
+ private static Object readScalar(JsonReader reader, PathContext context) {
+ char token = reader.peekToken();
+ if (token == '"') {
+ countToken(context);
+ 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') {
+ countToken(context);
+ return reader.readBoolean();
+ }
+ if (token == 'n') {
+ countToken(context);
+ reader.readNull();
+ return null;
+ }
+ if (token == '{' || token == '[') {
+ // Query scalar coercion has observable error/default behavior for
containers. Fully consume the value to keep
+ // malformed-tail and duplicate-key semantics, then signal a
stack-trace-free retry through Jayway.
+ skipValue(reader, context);
+ return FALLBACK_REQUIRED;
+ }
+ countToken(context);
+ 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, PathContext context) {
+ char token = reader.peekToken();
+ if (token == '{') {
+ skipObject(reader, context);
+ return;
+ }
+ if (token == '[') {
+ skipArray(reader, context);
+ return;
+ }
+ if (token == '"') {
+ countToken(context);
+ // 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;
+ }
+ countToken(context);
+ 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, PathContext context) {
+ reader.enterDepth();
+ try {
+ countToken(context);
+ reader.expect('{');
+ if (reader.consume('}')) {
+ countToken(context);
+ return;
+ }
+ boolean more;
+ do {
+ countToken(context);
+ 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, context);
+ more = reader.consumeCommaOrEndObject();
+ } while (more);
+ countToken(context);
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ private static void skipArray(JsonReader reader, PathContext context) {
+ reader.enterDepth();
+ try {
+ countToken(context);
+ reader.expect('[');
+ if (reader.consume(']')) {
+ countToken(context);
+ return;
+ }
+ boolean more;
+ do {
+ skipValue(reader, context);
+ more = reader.consumeCommaOrEndArray();
+ } while (more);
+ countToken(context);
+ } finally {
+ reader.exitDepth();
+ }
+ }
+
+ private static final class PathContext {
+ private final PathResult _marker = new PathResult();
+ private boolean _active;
+ private long _tokenCount;
+ @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, context);
+ 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 void countToken(PathContext context) {
+ if (JACKSON_CONSTRAINTS.hasMaxTokenCount()
+ && ++context._tokenCount > JACKSON_CONSTRAINTS.getMaxTokenCount()) {
+ throw new IllegalArgumentException("JSON token count exceeds Jackson's
configured limit");
+ }
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/TransformFunctionType.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/TransformFunctionType.java
index e884e526699..23c972d2684 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/TransformFunctionType.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/TransformFunctionType.java
@@ -107,6 +107,8 @@ public enum TransformFunctionType {
TransformFunctionType::jsonExtractScalarReturnTypeInference,
jsonExtractScalarOperandTypeChecker()),
JSON_EXTRACT_SCALAR_FIRST_MATCH("jsonExtractScalarFirstMatch",
TransformFunctionType::jsonExtractScalarReturnTypeInference,
jsonExtractScalarOperandTypeChecker()),
+ JSON_EXTRACT_SCALAR_FORY("jsonExtractScalarFory",
+ TransformFunctionType::jsonExtractScalarReturnTypeInference,
jsonExtractScalarOperandTypeChecker()),
JSON_EXTRACT_INDEX("jsonExtractIndex",
opBinding -> positionalReturnTypeInferenceFromStringLiteral(opBinding,
2, SqlTypeName.VARCHAR),
OperandTypes.family(
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/JsonFunctions.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/JsonFunctions.java
index df409185e4b..c8649a8add1 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/JsonFunctions.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/scalar/JsonFunctions.java
@@ -42,6 +42,7 @@ import java.util.UUID;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.function.FastJsonPathExtractor;
+import org.apache.pinot.common.function.ForyJsonPathExtractor;
import org.apache.pinot.common.function.JsonPathCache;
import org.apache.pinot.common.function.SimpleJsonPath;
import org.apache.pinot.spi.annotations.ScalarFunction;
@@ -129,6 +130,29 @@ public class JsonFunctions {
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) {
+ if (!(object instanceof String)) {
+ return jsonPath(object, jsonPath);
+ }
+ SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(jsonPath);
+ if (simpleJsonPath == null) {
+ return jsonPath(object, jsonPath);
+ }
+ try {
+ if (!ForyJsonPathExtractor.isAvailable()) {
+ return jsonPath(object, jsonPath);
+ }
+ Object value = ForyJsonPathExtractor.extract((String) object,
simpleJsonPath);
+ return ForyJsonPathExtractor.isFallbackRequired(value) ?
jsonPath(object, jsonPath) : value;
+ } catch (RuntimeException | LinkageError e) {
+ return jsonPath(object, jsonPath);
+ }
+ }
+
/// Returns `false` when the input is known to be non-extractable by a json
path without invoking the JSON
/// parser: a `null` value, or a string whose first non-whitespace character
cannot begin a JSON value. Used
/// by the default-value `jsonPath*` overloads to skip parsing plain-text
input (e.g. raw log lines) that
@@ -278,6 +302,24 @@ public class JsonFunctions {
}
}
+ /// **Experimental.** Fory-backed variant of [#jsonPathString(Object,
String, String)]. Fory parses JSON strings
+ /// for simple paths; unsupported inputs and parse failures fall back to
Jayway. This function can change or be
+ /// removed while the Fory integration is evaluated and must be explicitly
selected by name. The optional
+ /// `org.apache.fory:fory-json:1.6.0` runtime must be on the application
classpath to activate Fory; otherwise this
+ /// function uses Jayway.
+ @ScalarFunction(nullableParameters = true)
+ public static String jsonPathStringFory(@Nullable Object object, String
jsonPath, String defaultValue) {
+ if (!canExtractJsonPath(object)) {
+ return defaultValue;
+ }
+ try {
+ Object jsonValue = foryJsonPath(object, jsonPath);
+ return jsonValue != null ? jsonValueToString(jsonValue) : defaultValue;
+ } catch (Exception ignore) {
+ return defaultValue;
+ }
+ }
+
/// Extract from Json with path to Long
@ScalarFunction
public static long jsonPathLong(Object object, String jsonPath) {
@@ -352,6 +394,22 @@ public class JsonFunctions {
}
}
+ /// **Experimental.** Fory-backed variant of [#jsonPathLong(Object, String,
long)]. Unsupported inputs and parse
+ /// failures fall back to Jayway. This function can change or be removed
while the Fory integration is evaluated.
+ /// The optional `org.apache.fory:fory-json:1.6.0` runtime must be on the
application classpath to activate Fory.
+ @ScalarFunction(nullableParameters = true)
+ public static long jsonPathLongFory(@Nullable Object object, String
jsonPath, long defaultValue) {
+ if (!canExtractJsonPath(object)) {
+ return defaultValue;
+ }
+ try {
+ Object jsonValue = foryJsonPath(object, jsonPath);
+ return jsonValue != null ? jsonValueToLong(jsonValue) : defaultValue;
+ } catch (Exception ignore) {
+ return defaultValue;
+ }
+ }
+
/// Extract from Json with path to Double
@ScalarFunction
public static double jsonPathDouble(Object object, String jsonPath) {
@@ -425,6 +483,23 @@ public class JsonFunctions {
}
}
+ /// **Experimental.** Fory-backed variant of [#jsonPathDouble(Object,
String, double)]. Unsupported inputs and
+ /// parse failures fall back to Jayway. This function can change or be
removed while the Fory integration is
+ /// evaluated. The optional `org.apache.fory:fory-json:1.6.0` runtime must
be on the application classpath to
+ /// activate Fory.
+ @ScalarFunction(nullableParameters = true)
+ public static double jsonPathDoubleFory(@Nullable Object object, String
jsonPath, double defaultValue) {
+ if (!canExtractJsonPath(object)) {
+ return defaultValue;
+ }
+ try {
+ Object jsonValue = foryJsonPath(object, jsonPath);
+ return jsonValue != null ? jsonValueToDouble(jsonValue) : defaultValue;
+ } catch (Exception ignore) {
+ return defaultValue;
+ }
+ }
+
/// Extract an array of key-value maps to a map.
/// E.g. input: \[{"key": "k1", "value": "v1"}, {"key": "k2", "value":
"v2"}, {"key": "k3", "value": "v3"}\]
/// output: {"k1": "v1", "k2": "v2", "k3": "v3"}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/sql/parsers/ParserUtils.java
b/pinot-common/src/main/java/org/apache/pinot/sql/parsers/ParserUtils.java
index 77489f847ca..67c9bc8d67d 100644
--- a/pinot-common/src/main/java/org/apache/pinot/sql/parsers/ParserUtils.java
+++ b/pinot-common/src/main/java/org/apache/pinot/sql/parsers/ParserUtils.java
@@ -37,6 +37,9 @@ public class ParserUtils {
case "jsonextractscalarfirstmatch":
validateJsonExtractScalarFunction("jsonExtractScalarFirstMatch",
operands);
break;
+ case "jsonextractscalarfory":
+ validateJsonExtractScalarFunction("jsonExtractScalarFory", operands);
+ break;
case "jsonextractkey":
validateJsonExtractKeyFunction(operands);
break;
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/function/FastJsonPathExtractorTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/function/FastJsonPathExtractorTest.java
index 3212b705d1b..f4677acb672 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/function/FastJsonPathExtractorTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/function/FastJsonPathExtractorTest.java
@@ -397,6 +397,7 @@ public class FastJsonPathExtractorTest {
assertEquals(invoke("jsonPathStringFast", json, "$.user.country",
"DEFAULT"), "US");
assertEquals(invoke("jsonPathStringFirstMatch", json, "$.user.country",
"DEFAULT"), "US");
assertEquals(invoke("jsonPathStringFast", json, "$.missing", "DEFAULT"),
"DEFAULT");
+ assertEquals(invoke("jsonPathLongFast", json, "$.user.age", -7L), 41L);
assertEquals(invoke("jsonPathLongFirstMatch", json, "$.user.age", -7L),
41L);
assertEquals(invoke("jsonPathDoubleFast", json, "$.user.score", -7.5d),
9.5d);
/// A complex path must still resolve through the function by falling back
to Jayway, i.e. produce exactly
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonLinkageFallbackTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonLinkageFallbackTest.java
new file mode 100644
index 00000000000..2169d553724
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonLinkageFallbackTest.java
@@ -0,0 +1,143 @@
+/**
+ * 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.io.File;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.StringJoiner;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+import org.apache.pinot.common.function.scalar.JsonFunctions;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Verifies that an optional Fory runtime linkage failure does not break the
JSON functions.
+public class ForyJsonLinkageFallbackTest {
+ private static final String FALLBACK_ARGUMENT = "verifyFallback";
+ private static final String TOKEN_LIMIT_ARGUMENT = "verifyTokenLimit";
+
+ /// Child-process entry point used by the missing-Fory fallback tests.
+ public static void main(String[] arguments) {
+ if (arguments.length != 1) {
+ throw new IllegalArgumentException("Expected one child-process
verification argument");
+ }
+ if (TOKEN_LIMIT_ARGUMENT.equals(arguments[0])) {
+ verifyConfiguredTokenLimit();
+ return;
+ }
+ if (!FALLBACK_ARGUMENT.equals(arguments[0])) {
+ throw new IllegalArgumentException("Unknown child-process verification
argument: " + arguments[0]);
+ }
+ String actual = JsonFunctions.jsonPathStringFory("{\"v\":7}", "$.v",
"DEFAULT");
+ if (!"7".equals(actual)) {
+ throw new AssertionError("Expected Jayway fallback result 7, got: " +
actual);
+ }
+ long longValue = JsonFunctions.jsonPathLongFory("{\"v\":7}", "$.v", -1L);
+ if (longValue != 7L) {
+ throw new AssertionError("Expected Jayway fallback long 7, got: " +
longValue);
+ }
+ double doubleValue = JsonFunctions.jsonPathDoubleFory("{\"v\":7.5}",
"$.v", -1d);
+ if (doubleValue != 7.5d) {
+ throw new AssertionError("Expected Jayway fallback double 7.5, got: " +
doubleValue);
+ }
+ }
+
+ @Test
+ public void testMissingForyCoreFallsBack()
+ throws Exception {
+ runChild(false, true, FALLBACK_ARGUMENT);
+ }
+
+ @Test
+ public void testMissingForyJsonFallsBack()
+ throws Exception {
+ runChild(true, false, FALLBACK_ARGUMENT);
+ }
+
+ @Test
+ public void testConfiguredJacksonTokenLimit()
+ throws Exception {
+ runChild(false, false, TOKEN_LIMIT_ARGUMENT);
+ }
+
+ private static void verifyConfiguredTokenLimit() {
+ StreamReadConstraints constraints =
StreamReadConstraints.builder().maxTokenCount(3).build();
+ StreamReadConstraints.overrideDefaultStreamReadConstraints(constraints);
+ SimpleJsonPath path = SimpleJsonPath.compile("$.v");
+ if (path == null) {
+ throw new AssertionError("Expected a simple JSON path");
+ }
+ try {
+ ForyJsonPathExtractor.extract("{\"v\":7}", path);
+ throw new AssertionError("Expected Fory to enforce Jackson's configured
token limit");
+ } catch (IllegalArgumentException expected) {
+ // Expected: START_OBJECT, FIELD_NAME, VALUE_NUMBER_INT, END_OBJECT
exceeds the configured limit of three.
+ }
+ }
+
+ private static void runChild(boolean removeForyJson, boolean removeForyCore,
String childArgument)
+ throws Exception {
+ String separator = System.getProperty("path.separator");
+ StringJoiner childClassPath = new StringJoiner(separator);
+ boolean foundForyCore = false;
+ boolean foundForyJson = false;
+ for (String entry :
System.getProperty("java.class.path").split(Pattern.quote(separator))) {
+ String fileName = new File(entry).getName();
+ if (fileName.startsWith("fory-core-")) {
+ foundForyCore = true;
+ if (!removeForyCore) {
+ childClassPath.add(entry);
+ }
+ } else if (fileName.startsWith("fory-json-")) {
+ foundForyJson = true;
+ if (!removeForyJson) {
+ childClassPath.add(entry);
+ }
+ } else {
+ childClassPath.add(entry);
+ }
+ }
+ assertTrue(foundForyCore, "Test classpath does not contain fory-core");
+ assertTrue(foundForyJson, "Test classpath does not contain fory-json");
+
+ String javaExecutable = new File(new File(System.getProperty("java.home"),
"bin"), "java").getPath();
+ ProcessBuilder processBuilder = new ProcessBuilder(javaExecutable, "-cp",
childClassPath.toString(),
+ ForyJsonLinkageFallbackTest.class.getName(),
childArgument).redirectErrorStream(true);
+ processBuilder.environment().remove("JAVA_TOOL_OPTIONS");
+ processBuilder.environment().remove("JDK_JAVA_OPTIONS");
+
+ Process process = processBuilder.start();
+ boolean exited = process.waitFor(30, TimeUnit.SECONDS);
+ if (!exited) {
+ process.destroyForcibly();
+ process.waitFor(30, TimeUnit.SECONDS);
+ }
+ String output;
+ try (InputStream input = process.getInputStream()) {
+ output = new String(input.readAllBytes(), StandardCharsets.UTF_8);
+ }
+ assertTrue(exited, "Fallback child process timed out: " + output);
+ assertEquals(process.exitValue(), 0, output);
+ }
+}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonPathFunctionsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonPathFunctionsTest.java
new file mode 100644
index 00000000000..b57aadf7b9c
--- /dev/null
+++
b/pinot-common/src/test/java/org/apache/pinot/common/function/ForyJsonPathFunctionsTest.java
@@ -0,0 +1,298 @@
+/**
+ * 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.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.function.scalar.JsonFunctions;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
+import static org.testng.Assert.assertTrue;
+
+
+/// Differential and registry coverage for the opt-in Fory-backed JSON path
scalar functions.
+public class ForyJsonPathFunctionsTest {
+
+ @DataProvider(name = "stringCases")
+ public Object[][] stringCases() {
+ return new Object[][]{
+ {"{\"user\":{\"country\":\"US\"}}", "$.user.country"},
+ {"{\"items\":[{\"sku\":\"A1\"},{\"sku\":\"B7\"}]}", "$.items[1].sku"},
+ {"{\"v\":2147483648}", "$.v"},
+ {"{\"v\":9223372036854775808}", "$.v"},
+ {"{\"v\":1.2345678901234567}", "$.v"},
+ {"{\"v\":123456789012345678901234567890.123}", "$.v"},
+ {"{\"v\":true}", "$.v"},
+ {"{\"v\":\"café 😀\"}", "$.v"},
+ {"{\"v\":[1,2,3]}", "$.v"},
+ {"{\"v\":{\"nested\":1}}", "$.v"},
+ {"{\"v\":1,\"v\":2}", "$.v"}
+ };
+ }
+
+ @Test(dataProvider = "stringCases")
+ public void testStringParity(String json, String path) {
+ assertEquals(JsonFunctions.jsonPathStringFory(json, path, "DEFAULT"),
+ JsonFunctions.jsonPathString(json, path, "DEFAULT"));
+ }
+
+ @Test
+ public void testTypedFunctionsAndDefaults() {
+ String json =
"{\"n\":9223372036854775806,\"d\":19.75,\"numeric\":\"41\",\"nil\":null}";
+ assertEquals(JsonFunctions.jsonPathLongFory(json, "$.n", -1L),
+ JsonFunctions.jsonPathLong(json, "$.n", -1L));
+ assertEquals(JsonFunctions.jsonPathLongFory(json, "$.numeric", -1L), 41L);
+ assertEquals(JsonFunctions.jsonPathDoubleFory(json, "$.d", -1d),
+ JsonFunctions.jsonPathDouble(json, "$.d", -1d));
+ assertEquals(JsonFunctions.jsonPathStringFory(json, "$.missing",
"DEFAULT"), "DEFAULT");
+ assertEquals(JsonFunctions.jsonPathStringFory(json, "$.nil", "DEFAULT"),
"DEFAULT");
+ assertEquals(JsonFunctions.jsonPathLongFory("plain text", "$.n", -7L),
-7L);
+ assertEquals(JsonFunctions.jsonPathDoubleFory("{broken", "$.d", -7.5d),
-7.5d);
+ assertEquals(JsonFunctions.jsonPathStringFory(null, "$.v", "DEFAULT"),
"DEFAULT");
+ }
+
+ @Test
+ public void testStreamingParserIsAvailable() {
+ assertTrue(ForyJsonPathExtractor.isAvailable());
+ SimpleJsonPath path = SimpleJsonPath.compile("$.nested.value");
+ assertNotNull(path);
+
assertEquals(ForyJsonPathExtractor.extract("{\"n\":7,\"nested\":{\"value\":8}}",
path), 8L);
+ }
+
+ @Test
+ public void testJaywayFallbacks() {
+ String complex =
"{\"left\":{\"country\":\"US\"},\"right\":{\"country\":\"DE\"}}";
+ assertEquals(JsonFunctions.jsonPathStringFory(complex, "$..country",
"DEFAULT"),
+ JsonFunctions.jsonPathString(complex, "$..country", "DEFAULT"));
+
+ String trailingContent = "{\"v\":7} {\"ignored\":true}";
+ assertEquals(JsonFunctions.jsonPathStringFory(trailingContent, "$.v",
"DEFAULT"),
+ JsonFunctions.jsonPathString(trailingContent, "$.v", "DEFAULT"));
+
+ String malformedAfterMatch = "{\"v\":7,\"broken\":[}";
+ assertEquals(JsonFunctions.jsonPathStringFory(malformedAfterMatch, "$.v",
"DEFAULT"),
+ JsonFunctions.jsonPathString(malformedAfterMatch, "$.v", "DEFAULT"));
+
+ Map<String, Object> parsed = Map.of("v", Map.of("n", 42));
+ assertEquals(JsonFunctions.jsonPathLongFory(parsed, "$.v.n", -1L),
+ JsonFunctions.jsonPathLong(parsed, "$.v.n", -1L));
+
+ StringBuilder deepJson = new StringBuilder();
+ StringBuilder deepPath = new StringBuilder("$");
+ for (int i = 0; i < 25; i++) {
+ deepJson.append("{\"a\":");
+ deepPath.append(".a");
+ }
+ deepJson.append("\"value\"");
+ for (int i = 0; i < 25; i++) {
+ deepJson.append('}');
+ }
+ assertEquals(JsonFunctions.jsonPathStringFory(deepJson.toString(),
deepPath.toString(), "DEFAULT"), "value");
+ }
+
+ @Test
+ public void testJacksonConstraintFallbacks() {
+ StreamReadConstraints constraints = StreamReadConstraints.defaults();
+
+ String oversizedNumber = "1".repeat(constraints.getMaxNumberLength() + 1);
+ String numberDocument = "{\"oversized\":" + oversizedNumber + ",\"v\":7}";
+ assertEquals(JsonFunctions.jsonPathLongFory(numberDocument, "$.v", -1),
+ JsonFunctions.jsonPathLong(numberDocument, "$.v", -1));
+
+ String oversizedName = "n".repeat(constraints.getMaxNameLength() + 1);
+ String nameDocument = "{\"" + oversizedName + "\":1,\"v\":7}";
+ assertEquals(JsonFunctions.jsonPathLongFory(nameDocument, "$.v", -1),
+ JsonFunctions.jsonPathLong(nameDocument, "$.v", -1));
+ }
+
+ @Test
+ public void testStreamingExtractorNavigatesSimplePaths() {
+ assertEquals(extract("[{\"v\":1},{\"v\":2}]", "$[1].v"), 2L);
+ assertEquals(extract("{\"items\":[0,{\"detail\":{\"value\":3}}]}",
"$.items[1].detail.value"), 3L);
+ assertEquals(extract("{\"a-b\":{\"café\":4}}", "$['a-b'].café"), 4L);
+ }
+
+ @Test
+ public void testStreamingExtractorUsesLastDuplicate() {
+ assertEquals(extract("{\"v\":1,\"v\":2}", "$.v"), 2L);
+ assertEquals(extract("{\"a\":{\"v\":1},\"a\":{\"v\":2}}", "$.a.v"), 2L);
+ assertNull(extract("{\"a\":{\"v\":1},\"a\":7}", "$.a.v"));
+ assertNull(extract("{\"v\":1,\"v\":null}", "$.v"));
+ }
+
+ @Test
+ public void testStreamingExtractorReturnsNullForUnresolvedPaths() {
+ assertNull(extract("{\"a\":{\"v\":1}}", "$.missing"));
+ assertNull(extract("{\"a\":{\"v\":null}}", "$.a.v"));
+ assertNull(extract("{\"a\":7}", "$.a.v"));
+ assertNull(extract("{\"a\":[]}", "$.a[1]"));
+ }
+
+ @Test
+ public void
testStreamingExtractorSignalsContainerFallbackWithoutExceptions() {
+
assertTrue(ForyJsonPathExtractor.isFallbackRequired(extract("{\"v\":{\"n\":1}}",
"$.v")));
+
assertTrue(ForyJsonPathExtractor.isFallbackRequired(extract("{\"v\":[1,2]}",
"$.v")));
+ assertEquals(extract("{\"v\":{},\"v\":2}", "$.v"), 2L);
+
assertTrue(ForyJsonPathExtractor.isFallbackRequired(extract("{\"v\":2,\"v\":{}}",
"$.v")));
+ assertNull(extract("{\"v\":{},\"v\":null}", "$.v"));
+
+ SimpleJsonPath path = SimpleJsonPath.compile("$.v");
+ assertNotNull(path);
+ assertThrows(RuntimeException.class,
+ () -> ForyJsonPathExtractor.extract("{\"v\":{},\"broken\":[}", path));
+ assertEquals(ForyJsonPathExtractor.extract("{\"v\":3}", path), 3L);
+ }
+
+ @Test
+ public void testStreamingExtractorUsesJacksonNestingLimit() {
+ int depth = 25;
+ StringBuilder selectedJson = new StringBuilder();
+ StringBuilder selectedPath = new StringBuilder("$");
+ for (int i = 0; i < depth; i++) {
+ selectedJson.append("{\"a\":");
+ selectedPath.append(".a");
+ }
+ selectedJson.append('7');
+ selectedJson.append("}".repeat(depth));
+ assertEquals(extract(selectedJson.toString(), selectedPath.toString()),
7L);
+
+ String unrelated = "{\"selected\":1,\"deep\":" + "{\"a\":".repeat(depth) +
"7" + "}".repeat(depth)
+ + "}";
+ assertEquals(extract(unrelated, "$.selected"), 1L);
+
+ int maximumDepth = StreamReadConstraints.defaults().getMaxNestingDepth();
+ String maximumJson = "{\"a\":".repeat(maximumDepth) + "7" +
"}".repeat(maximumDepth);
+ assertEquals(extract(maximumJson, "$." + "a.".repeat(maximumDepth - 1) +
"a"), 7L);
+
+ String oversizedJson = "{\"a\":".repeat(maximumDepth + 1) + "7" +
"}".repeat(maximumDepth + 1);
+ SimpleJsonPath oversizedPath = SimpleJsonPath.compile("$." +
"a.".repeat(maximumDepth) + "a");
+ assertNotNull(oversizedPath);
+ assertThrows(RuntimeException.class, () ->
ForyJsonPathExtractor.extract(oversizedJson, oversizedPath));
+ }
+
+ @Test
+ public void testStreamingExtractorRejectsMalformedAndTrailingContent() {
+ SimpleJsonPath path = SimpleJsonPath.compile("$.v");
+ assertNotNull(path);
+ assertThrows(RuntimeException.class, () ->
ForyJsonPathExtractor.extract("{\"v\":1,\"broken\":[}", path));
+ assertEquals(ForyJsonPathExtractor.extract("{\"v\":2}", path), 2L);
+ assertThrows(RuntimeException.class, () ->
ForyJsonPathExtractor.extract("{\"v\":1} {\"ignored\":2}", path));
+ assertEquals(ForyJsonPathExtractor.extract("{\"v\":3}", path), 3L);
+ }
+
+ @Test
+ public void testStreamingExtractorJacksonConstraintBoundaries() {
+ StreamReadConstraints constraints = StreamReadConstraints.defaults();
+ SimpleJsonPath valuePath = SimpleJsonPath.compile("$.v");
+ assertNotNull(valuePath);
+
+ String maximumNumber = "1".repeat(constraints.getMaxNumberLength());
+ assertEquals(ForyJsonPathExtractor.extract("{\"v\":" + maximumNumber +
"}", valuePath).toString(),
+ maximumNumber);
+ String oversizedNumber = maximumNumber + '1';
+ assertThrows(RuntimeException.class,
+ () -> ForyJsonPathExtractor.extract("{\"v\":" + oversizedNumber + "}",
valuePath));
+ assertThrows(RuntimeException.class,
+ () -> ForyJsonPathExtractor.extract("{\"oversized\":" +
oversizedNumber + ",\"v\":7}", valuePath));
+
+ String maximumName = "n".repeat(constraints.getMaxNameLength());
+ assertEquals(ForyJsonPathExtractor.extract("{\"" + maximumName +
"\":1,\"v\":7}", valuePath), 7L);
+ String oversizedName = maximumName + 'n';
+ assertThrows(RuntimeException.class,
+ () -> ForyJsonPathExtractor.extract("{\"" + oversizedName +
"\":1,\"v\":7}", valuePath));
+ }
+
+ @Test
+ public void testStreamingExtractorDoesNotLeakContextAcrossThreads()
+ throws Exception {
+ int numThreads = 8;
+ ExecutorService executor = Executors.newFixedThreadPool(numThreads);
+ CountDownLatch ready = new CountDownLatch(numThreads);
+ CountDownLatch start = new CountDownLatch(1);
+ try {
+ List<Future<?>> futures = new ArrayList<>(numThreads);
+ for (int worker = 0; worker < numThreads; worker++) {
+ int expected = worker;
+ futures.add(executor.submit(() -> {
+ SimpleJsonPath workerPath = SimpleJsonPath.compile("$.worker");
+ SimpleJsonPath nestedPath =
SimpleJsonPath.compile("$.nested.worker");
+ assertNotNull(workerPath);
+ assertNotNull(nestedPath);
+ String json = "{\"worker\":" + expected + ",\"nested\":{\"worker\":"
+ (expected + 100) + "}}";
+ ready.countDown();
+ assertTrue(start.await(10, TimeUnit.SECONDS));
+ for (int iteration = 0; iteration < 100; iteration++) {
+ if ((iteration & 1) == 0) {
+ assertEquals(ForyJsonPathExtractor.extract(json, workerPath),
(long) expected);
+ } else {
+ assertEquals(ForyJsonPathExtractor.extract(json, nestedPath),
(long) expected + 100);
+ }
+ }
+ return null;
+ }));
+ }
+ assertTrue(ready.await(10, TimeUnit.SECONDS));
+ start.countDown();
+ for (Future<?> future : futures) {
+ future.get(30, TimeUnit.SECONDS);
+ }
+ } finally {
+ executor.shutdownNow();
+ executor.awaitTermination(10, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testFunctionsResolveThroughRegistry()
+ throws Exception {
+ String json = "{\"user\":{\"country\":\"US\",\"age\":41,\"score\":9.5}}";
+ assertEquals(invoke("jsonPathStringFory", json, "$.user.country",
"DEFAULT"), "US");
+ assertEquals(invoke("jsonPathLongFory", json, "$.user.age", -7L), 41L);
+ assertEquals(invoke("jsonPathDoubleFory", json, "$.user.score", -7.5d),
9.5d);
+ }
+
+ private static Object invoke(String name, Object... arguments)
+ throws Exception {
+ FunctionInfo functionInfo =
+
FunctionRegistry.lookupFunctionInfo(FunctionRegistry.canonicalize(name),
arguments.length);
+ assertNotNull(functionInfo, name + "/" + arguments.length + " is not
registered");
+ FunctionInvoker invoker = new FunctionInvoker(functionInfo);
+ Object[] copy = arguments.clone();
+ invoker.convertTypes(copy);
+ return invoker.invoke(copy);
+ }
+
+ private static Object extract(String json, String jsonPath) {
+ SimpleJsonPath path = SimpleJsonPath.compile(jsonPath);
+ assertNotNull(path);
+ return ForyJsonPathExtractor.extract(json, path);
+ }
+}
diff --git a/pinot-core/pom.xml b/pinot-core/pom.xml
index 7035b80f5bc..659781960cd 100644
--- a/pinot-core/pom.xml
+++ b/pinot-core/pom.xml
@@ -170,6 +170,11 @@
<artifactId>assertj-core</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<profiles>
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
index 21e0f3e0cde..78b054cb470 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunction.java
@@ -21,6 +21,7 @@ package org.apache.pinot.core.operator.transform.function;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
import com.jayway.jsonpath.Configuration;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.Option;
@@ -33,6 +34,7 @@ import java.util.Map;
import java.util.function.IntFunction;
import javax.annotation.Nullable;
import org.apache.pinot.common.function.FastJsonPathExtractor;
+import org.apache.pinot.common.function.ForyJsonPathExtractor;
import org.apache.pinot.common.function.JsonPathCache;
import org.apache.pinot.common.function.SimpleJsonPath;
import org.apache.pinot.core.operator.ColumnContext;
@@ -55,6 +57,7 @@ import org.roaringbitmap.RoaringBitmap;
/// | `jsonExtractScalar` | Builds the existing Jayway DOM; unchanged for
backward compatibility. |
/// | `jsonExtractScalarFast` | Uses [FastJsonPathExtractor] and scans the
full root value, preserving Jayway results. |
/// | `jsonExtractScalarFirstMatch` | Uses [FastJsonPathExtractor] and stops
when the addressed value is found. |
+/// | `jsonExtractScalarFory` | Experimental Fory streaming extraction with
per-row Jayway fallback. |
///
/// Each function reads a JSON document from `jsonField` for each row,
resolves the
/// [Stefan Goessner JsonPath](https://goessner.net/articles/JsonPath/)
expression against it, and
@@ -72,6 +75,12 @@ import org.roaringbitmap.RoaringBitmap;
/// 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`, `JSON`, and
`BIG_DECIMAL`. `BYTES` input, complex
+/// paths, containers / array result types, precision-sensitive results,
documents beyond Jackson's configured
+/// limits, and Fory failures use Jayway. Its name, supported envelope, and
implementation can change while the
+/// integration is evaluated. Fory is an optional runtime dependency; add
`org.apache.fory:fory-json:1.6.0` (and its
+/// transitive `fory-core` dependency) to the application classpath to
activate this experimental path.
///
/// **Arguments:**
/// - `jsonField` — single-value `STRING` or `BYTES` column / transform
expression containing JSON.
@@ -105,11 +114,13 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
public static final String FUNCTION_NAME = "jsonExtractScalar";
public static final String FAST_FUNCTION_NAME = "jsonExtractScalarFast";
public static final String FIRST_MATCH_FUNCTION_NAME =
"jsonExtractScalarFirstMatch";
+ public static final String FORY_FUNCTION_NAME = "jsonExtractScalarFory";
private enum ExtractionMode {
JAYWAY,
FAST,
- FIRST_MATCH
+ FIRST_MATCH,
+ FORY
}
// This ObjectMapper requires special configurations, hence we can't use
pinot JsonUtils here.
@@ -134,6 +145,7 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
private DataType _storedType;
private Object _defaultValue;
private boolean _defaultIsNull;
+ private boolean _foryEligible;
private TransformResultMetadata _resultMetadata;
public JsonExtractScalarTransformFunction() {
@@ -159,6 +171,13 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
}
}
+ /// Experimental Fory-backed variant of [JsonExtractScalarTransformFunction].
+ public static final class Fory extends JsonExtractScalarTransformFunction {
+ public Fory() {
+ super(FORY_FUNCTION_NAME, ExtractionMode.FORY);
+ }
+ }
+
@Override
public String getName() {
return _functionName;
@@ -241,6 +260,17 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
}
}
_resultMetadata = new TransformResultMetadata(_dataType, isSingleValue,
false);
+ DataType inputDataType = firstArgument.getResultMetadata().getDataType();
+ // STRING/JSON use the BigDecimal-preserving parser because non-string
JSON values are serialized to the output.
+ // Fory 1.6 materializes fractional numbers as Double, so enabling it here
would lose numeric precision.
+ _foryEligible = _extractionMode == ExtractionMode.FORY && _simpleJsonPath
!= null
+ && inputDataType == DataType.STRING && isSingleValue && _dataType !=
DataType.STRING
+ && _dataType != DataType.JSON && _dataType != DataType.BIG_DECIMAL;
+ }
+
+ @VisibleForTesting
+ boolean isForyEligible() {
+ return _foryEligible;
}
@Override
@@ -802,7 +832,7 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
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 || _extractionMode == ExtractionMode.FORY) {
return jaywayExtractor;
}
SimpleJsonPath[] paths = {_simpleJsonPath};
@@ -825,6 +855,26 @@ public class JsonExtractScalarTransformFunction extends
BaseTransformFunction {
if (_simpleJsonPath == null) {
return jaywayExtractor;
}
+ if (_extractionMode == ExtractionMode.FORY) {
+ if (!_foryEligible || useBigDecimal) {
+ return jaywayExtractor;
+ }
+ try {
+ if (!ForyJsonPathExtractor.isAvailable()) {
+ return jaywayExtractor;
+ }
+ } catch (LinkageError ignored) {
+ return jaywayExtractor;
+ }
+ return i -> {
+ try {
+ Object value = ForyJsonPathExtractor.extract(jsonStrings[i],
_simpleJsonPath);
+ return ForyJsonPathExtractor.isFallbackRequired(value) ?
jaywayExtractor.apply(i) : (T) value;
+ } catch (RuntimeException | LinkageError ignored) {
+ return jaywayExtractor.apply(i);
+ }
+ };
+ }
SimpleJsonPath[] paths = {_simpleJsonPath};
Object[] result = new Object[1];
boolean earlyExit = _extractionMode == ExtractionMode.FIRST_MATCH;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/TransformFunctionFactory.java
b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/TransformFunctionFactory.java
index 0d49547ca90..c4647f9a5bc 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/TransformFunctionFactory.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/operator/transform/function/TransformFunctionFactory.java
@@ -132,6 +132,8 @@ public class TransformFunctionFactory {
JsonExtractScalarTransformFunction.Fast.class);
typeToImplementation.put(TransformFunctionType.JSON_EXTRACT_SCALAR_FIRST_MATCH,
JsonExtractScalarTransformFunction.FirstMatch.class);
+ typeToImplementation.put(TransformFunctionType.JSON_EXTRACT_SCALAR_FORY,
+ JsonExtractScalarTransformFunction.Fory.class);
typeToImplementation.put(TransformFunctionType.JSON_EXTRACT_KEY,
JsonExtractKeyTransformFunction.class);
typeToImplementation.put(TransformFunctionType.TIME_CONVERT,
TimeConversionTransformFunction.class);
typeToImplementation.put(TransformFunctionType.DATE_TIME_CONVERT,
DateTimeConversionTransformFunction.class);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/data/function/JsonFunctionsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/data/function/JsonFunctionsTest.java
index fbe2efff8ed..bbaed22658a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/data/function/JsonFunctionsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/data/function/JsonFunctionsTest.java
@@ -133,6 +133,29 @@ public class JsonFunctionsTest {
inputs.add(new Object[]{
"json_path_double(jsonPathString, '$.k3.sub2')",
Lists.newArrayList("jsonPathString"), row11, 1.0
});
+
+ // Experimental Fory variants must resolve through the ingestion evaluator
with raw JSON strings. Parsed
+ // Map/List input would deliberately use their Jayway fallback and would
not exercise Fory initialization.
+ GenericRow row12 = new GenericRow();
+ row12.putValue("json", "{\"text\":\"value\",\"count\":10,\"ratio\":1.25}");
+ inputs.add(new Object[]{
+ "json_path_string_fory(json, '$.text', 'DEFAULT')",
Lists.newArrayList("json"), row12, "value"
+ });
+ inputs.add(new Object[]{
+ "json_path_long_fory(json, '$.count', -1)",
Lists.newArrayList("json"), row12, 10L
+ });
+ inputs.add(new Object[]{
+ "json_path_double_fory(json, '$.ratio', -1.0)",
Lists.newArrayList("json"), row12, 1.25
+ });
+ inputs.add(new Object[]{
+ "json_path_string_fast(json, '$.text', 'DEFAULT')",
Lists.newArrayList("json"), row12, "value"
+ });
+ inputs.add(new Object[]{
+ "json_path_long_fast(json, '$.count', -1)",
Lists.newArrayList("json"), row12, 10L
+ });
+ inputs.add(new Object[]{
+ "json_path_double_fast(json, '$.ratio', -1.0)",
Lists.newArrayList("json"), row12, 1.25
+ });
return inputs.toArray(new Object[0][]);
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java
index 37efd3c5b10..d1a0c076d46 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/function/FunctionRegistryTest.java
@@ -47,7 +47,7 @@ public class FunctionRegistryTest {
TransformFunctionType.IN, TransformFunctionType.NOT_IN,
TransformFunctionType.IS_TRUE,
TransformFunctionType.IS_NOT_TRUE, TransformFunctionType.IS_FALSE,
TransformFunctionType.IS_NOT_FALSE,
TransformFunctionType.JSON_EXTRACT_SCALAR,
TransformFunctionType.JSON_EXTRACT_SCALAR_FAST,
- TransformFunctionType.JSON_EXTRACT_SCALAR_FIRST_MATCH,
+ TransformFunctionType.JSON_EXTRACT_SCALAR_FIRST_MATCH,
TransformFunctionType.JSON_EXTRACT_SCALAR_FORY,
TransformFunctionType.JSON_EXTRACT_KEY,
TransformFunctionType.TIME_CONVERT,
TransformFunctionType.DATE_TIME_CONVERT_WINDOW_HOP,
TransformFunctionType.ARRAY_LENGTH,
TransformFunctionType.ARRAY_AVERAGE, TransformFunctionType.ARRAY_MIN,
TransformFunctionType.ARRAY_MAX,
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
index 421ded08d9c..f32114b00e0 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/operator/transform/function/JsonExtractScalarTransformFunctionTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.operator.transform.function;
+import com.fasterxml.jackson.core.StreamReadConstraints;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
@@ -55,7 +56,8 @@ public class JsonExtractScalarTransformFunctionTest extends
BaseTransformFunctio
private static final String[] JSON_EXTRACT_SCALAR_FUNCTIONS = {
JsonExtractScalarTransformFunction.FUNCTION_NAME,
JsonExtractScalarTransformFunction.FAST_FUNCTION_NAME,
- JsonExtractScalarTransformFunction.FIRST_MATCH_FUNCTION_NAME
+ JsonExtractScalarTransformFunction.FIRST_MATCH_FUNCTION_NAME,
+ JsonExtractScalarTransformFunction.FORY_FUNCTION_NAME
};
protected File _baseDir;
@@ -544,6 +546,8 @@ public class JsonExtractScalarTransformFunctionTest extends
BaseTransformFunctio
new Object[]{String.format("jsonExtractScalarFast(%s,
\"$.store.book[0].author\", 'String')", JSON_COLUMN)},
new Object[]{String.format("jsonExtractScalarFirstMatch(%s,
'$.store.book[0].author', \"String\")",
JSON_COLUMN)},
+ new Object[]{String.format("jsonExtractScalarFory(%s)", JSON_COLUMN)},
+ new Object[]{String.format("jsonExtractScalarFory(%s,
\"$.store.book[0].author\", 'String')", JSON_COLUMN)},
new Object[]{String.format("jsonExtractKey(%s, \"$.*\")",
JSON_COLUMN)},
new Object[]{String.format("json_extract_key(%s, \"$.*\")",
JSON_COLUMN)}};
//@formatter:on
@@ -718,6 +722,23 @@ public class JsonExtractScalarTransformFunctionTest
extends BaseTransformFunctio
DataType.STRING, "INT", "-1", 1);
}
+ @Test
+ public void testForyFallsBackForJacksonNumberConstraint() {
+ String oversizedNumber =
"1".repeat(StreamReadConstraints.defaults().getMaxNumberLength() + 1);
+ String json = "{\"oversized\":" + oversizedNumber + ",\"v\":7}";
+ assertJsonExtractScalar(JsonExtractScalarTransformFunction.FUNCTION_NAME,
json, DataType.STRING, "INT", "-1", -1);
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FORY_FUNCTION_NAME,
json, DataType.STRING, "INT", "-1",
+ -1);
+ }
+
+ @Test
+ public void testForyStreamingParserFallsBackForMalformedAndTrailingContent()
{
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FORY_FUNCTION_NAME,
+ "{\"v\":1,\"broken\":[}", DataType.STRING, "INT", "-1", -1);
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FORY_FUNCTION_NAME,
+ "{\"v\":7} {\"ignored\":true}", DataType.STRING, "INT", "-1", 7);
+ }
+
@Test
public void testFastExtractionFromBytesWithBigDecimal() {
byte[] json = "{\"label\":\"crème
brûlée\",\"v\":12345678901234567890.123456789}"
@@ -728,6 +749,36 @@ public class JsonExtractScalarTransformFunctionTest
extends BaseTransformFunctio
"BIG_DECIMAL", null, expected);
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FIRST_MATCH_FUNCTION_NAME,
hexEncodedJson,
DataType.BYTES, "BIG_DECIMAL", null, expected);
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FORY_FUNCTION_NAME,
hexEncodedJson, DataType.BYTES,
+ "BIG_DECIMAL", null, expected);
+
+ byte[] hostileJson = ("{\"v\":12345678901234567890.123456789,"
+ + "\"ignored\":1e999999999999}").getBytes(StandardCharsets.UTF_8);
+ String hostileHexEncodedJson = BytesUtils.toHexString(hostileJson);
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FAST_FUNCTION_NAME,
hostileHexEncodedJson,
+ DataType.BYTES, "BIG_DECIMAL", null, expected);
+
assertJsonExtractScalar(JsonExtractScalarTransformFunction.FIRST_MATCH_FUNCTION_NAME,
hostileHexEncodedJson,
+ DataType.BYTES, "BIG_DECIMAL", null, expected);
+ }
+
+ @Test
+ public void testForyEligibilityEnvelope() {
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "LONG", true);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "DOUBLE", true);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "STRING", false);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "JSON", false);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "BIG_DECIMAL", false);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$.v", "LONG_ARRAY", false);
+ assertForyEligibility(JSON_STRING_SV_COLUMN, "$..v", "LONG", false);
+ assertForyEligibility(BYTES_SV_COLUMN, "$.v", "LONG", false);
+ }
+
+ private void assertForyEligibility(String column, String path, String
resultType, boolean expected) {
+ ExpressionContext expression = RequestContextUtils.getExpression(
+ "jsonExtractScalarFory(" + column + ", '" + path + "', '" + resultType
+ "')");
+ TransformFunction transformFunction =
TransformFunctionFactory.get(expression, _dataSourceMap);
+ Assert.assertTrue(transformFunction instanceof
JsonExtractScalarTransformFunction.Fory);
+ Assert.assertEquals(((JsonExtractScalarTransformFunction)
transformFunction).isForyEligible(), expected);
}
@Test
diff --git a/pinot-integration-tests/pom.xml b/pinot-integration-tests/pom.xml
index 4b7904d40a8..8c26aaca0bd 100644
--- a/pinot-integration-tests/pom.xml
+++ b/pinot-integration-tests/pom.xml
@@ -552,6 +552,11 @@
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-timeseries-planner</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/JsonPathTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/JsonPathTest.java
index b5d4b79d030..29bb13ae8c1 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/JsonPathTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/JsonPathTest.java
@@ -60,12 +60,14 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
// path (which visits every doc) return the same result set but follow
visibly different code paths.
private static final int NUM_DISTINCT_K1 = 100;
private static final String MY_MAP_STR_FIELD_NAME = "myMapStr";
+ private static final String MY_MAP_NUMBER_STR_FIELD_NAME = "myMapNumberStr";
private static final String MY_MAP_BYTES_FIELD_NAME = "myMapBytes";
private static final String MY_MAP_STR_K1_FIELD_NAME = "myMapStr_k1";
private static final String MY_MAP_STR_K2_FIELD_NAME = "myMapStr_k2";
/// Derived columns that exercise the opt-in fast scalar functions through
the ingestion transform path.
private static final String MY_MAP_STR_K1_FAST_FIELD_NAME =
"myMapStr_k1_fast";
private static final String MY_MAP_STR_K1_FIRST_FIELD_NAME =
"myMapStr_k1_first";
+ private static final String MY_MAP_STR_K1_FORY_FIELD_NAME =
"myMapStr_k1_fory";
private static final String COMPLEX_MAP_STR_FIELD_NAME = "complexMapStr";
private static final String COMPLEX_MAP_STR_K3_FIELD_NAME =
"complexMapStr_k3";
@@ -92,11 +94,13 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
.setSchemaName(getTableName())
.addSingleValueDimension("myMap", DataType.STRING)
.addSingleValueDimension(MY_MAP_STR_FIELD_NAME, DataType.STRING)
+ .addSingleValueDimension(MY_MAP_NUMBER_STR_FIELD_NAME, DataType.STRING)
.addSingleValueDimension(MY_MAP_BYTES_FIELD_NAME, DataType.BYTES)
.addSingleValueDimension(MY_MAP_STR_K1_FIELD_NAME, DataType.STRING)
.addSingleValueDimension(MY_MAP_STR_K2_FIELD_NAME, DataType.STRING)
.addSingleValueDimension(MY_MAP_STR_K1_FAST_FIELD_NAME,
DataType.STRING)
.addSingleValueDimension(MY_MAP_STR_K1_FIRST_FIELD_NAME,
DataType.STRING)
+ .addSingleValueDimension(MY_MAP_STR_K1_FORY_FIELD_NAME,
DataType.STRING)
.addSingleValueDimension(COMPLEX_MAP_STR_FIELD_NAME, DataType.STRING)
.addMultiValueDimension(COMPLEX_MAP_STR_K3_FIELD_NAME, DataType.STRING)
.build();
@@ -111,6 +115,8 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
"jsonPathStringFast(" + MY_MAP_STR_FIELD_NAME + ", '$.k1',
'DEFAULT')"),
new TransformConfig(MY_MAP_STR_K1_FIRST_FIELD_NAME,
"jsonPathStringFirstMatch(" + MY_MAP_STR_FIELD_NAME + ", '$.k1',
'DEFAULT')"),
+ new TransformConfig(MY_MAP_STR_K1_FORY_FIELD_NAME,
+ "jsonPathStringFory(" + MY_MAP_STR_FIELD_NAME + ", '$.k1',
'DEFAULT')"),
new TransformConfig(COMPLEX_MAP_STR_K3_FIELD_NAME, "jsonPathArray(" +
COMPLEX_MAP_STR_FIELD_NAME + ", '$.k3')")
);
IngestionConfig ingestionConfig = new IngestionConfig();
@@ -129,6 +135,8 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
List<org.apache.avro.Schema.Field> fields = List.of(
new org.apache.avro.Schema.Field(MY_MAP_STR_FIELD_NAME,
org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING),
null, null),
+ new org.apache.avro.Schema.Field(MY_MAP_NUMBER_STR_FIELD_NAME,
+ org.apache.avro.Schema.create(org.apache.avro.Schema.Type.STRING),
null, null),
new org.apache.avro.Schema.Field(MY_MAP_BYTES_FIELD_NAME,
org.apache.avro.Schema.create(org.apache.avro.Schema.Type.BYTES),
null, null),
new org.apache.avro.Schema.Field(COMPLEX_MAP_STR_FIELD_NAME,
@@ -144,6 +152,7 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
GenericData.Record record = new GenericData.Record(avroSchema);
String myMapJson = JsonUtils.objectToString(map);
record.put(MY_MAP_STR_FIELD_NAME, myMapJson);
+ record.put(MY_MAP_NUMBER_STR_FIELD_NAME,
JsonUtils.objectToString(Map.of("n", i)));
record.put(MY_MAP_BYTES_FIELD_NAME,
ByteBuffer.wrap(myMapJson.getBytes(StandardCharsets.UTF_8)));
Map<String, Object> complexMap = new HashMap<>();
@@ -398,52 +407,68 @@ public class JsonPathTest extends
CustomDataQueryClusterIntegrationTest {
assertEquals(pinotResponse.get("totalDocs").asInt(), 0);
}
- /// End-to-end coverage for the opt-in fast functions. They take an `Object`
argument, so - like the
- /// existing `jsonPathString` - they are used through the ingestion
transform path, not as query-time scalars.
- /// `myMapStr_k1_fast` / `myMapStr_k1_first` are derived at ingestion via
`jsonPathStringFast` /
- /// `jsonPathStringFirstMatch`; this asserts, over real rows on both query
engines, that they equal the
- /// Jayway-derived `myMapStr_k1`. The rows have no duplicate keys and no
malformed content, so `FirstMatch`
- /// must also agree.
+ /// End-to-end coverage for the opt-in fast functions and the experimental
Fory function. They take an `Object`
+ /// argument, so - like the existing `jsonPathString` - they are used
through the ingestion transform path, not as
+ /// query-time scalars.
+ /// The additional columns are derived at ingestion via
`jsonPathStringFast`, `jsonPathStringFirstMatch`, and
+ /// `jsonPathStringFory`; this asserts, over real rows on both query
engines, that they equal the Jayway-derived
+ /// `myMapStr_k1`. The rows have no duplicate keys and no malformed content,
so `FirstMatch` must also agree.
@Test(dataProvider = "useBothQueryEngines")
void testFastScalarFunctions(boolean useMultiStageQueryEngine)
throws Exception {
setUseMultiStageQueryEngine(useMultiStageQueryEngine);
- String query = "SELECT myMapStr_k1, myMapStr_k1_fast, myMapStr_k1_first
FROM " + getTableName() + " LIMIT 1000";
+ String query = "SELECT myMapStr_k1, myMapStr_k1_fast, myMapStr_k1_first,
myMapStr_k1_fory FROM "
+ + getTableName() + " LIMIT 1000";
JsonNode rows = postQuery(query).get("resultTable").get("rows");
assertTrue(rows.size() > 0, "expected non-empty result set");
for (JsonNode row : rows) {
String jayway = row.get(0).asText();
assertEquals(row.get(1).asText(), jayway, "jsonPathStringFast must equal
Jayway jsonPathString");
assertEquals(row.get(2).asText(), jayway, "jsonPathStringFirstMatch must
equal Jayway on clean data");
+ assertEquals(row.get(3).asText(), jayway, "jsonPathStringFory must equal
Jayway jsonPathString");
}
}
- /// Query-time coverage for the typed transforms backed by the same fast
extractor. Both modes must match the
- /// existing `jsonExtractScalar` transform on this clean, duplicate-free
data set.
+ /// Query-time coverage for the opt-in typed transforms. Each mode must
match the existing `jsonExtractScalar`
+ /// transform on this clean, duplicate-free data set. The numeric query is
eligible for Fory's streaming fast path;
+ /// STRING results and BYTES input deliberately exercise its Jayway
compatibility fallback.
@Test(dataProvider = "useBothQueryEngines")
void testFastJsonExtractScalarTransforms(boolean useMultiStageQueryEngine)
throws Exception {
setUseMultiStageQueryEngine(useMultiStageQueryEngine);
String query = "SELECT jsonExtractScalar(myMapStr, '$.k1', 'STRING'), "
+ "jsonExtractScalarFast(myMapStr, '$.k1', 'STRING'), "
- + "jsonExtractScalarFirstMatch(myMapStr, '$.k1', 'STRING') FROM " +
getTableName() + " LIMIT 1000";
+ + "jsonExtractScalarFirstMatch(myMapStr, '$.k1', 'STRING'), "
+ + "jsonExtractScalarFory(myMapStr, '$.k1', 'STRING') FROM " +
getTableName() + " LIMIT 1000";
JsonNode rows = postQuery(query).get("resultTable").get("rows");
assertTrue(rows.size() > 0, "expected non-empty result set");
for (JsonNode row : rows) {
String jayway = row.get(0).asText();
assertEquals(row.get(1).asText(), jayway, "jsonExtractScalarFast must
equal jsonExtractScalar");
assertEquals(row.get(2).asText(), jayway, "jsonExtractScalarFirstMatch
must equal Jayway on clean data");
+ assertEquals(row.get(3).asText(), jayway, "jsonExtractScalarFory must
equal Jayway on clean data");
+ }
+
+ query = "SELECT jsonExtractScalar(myMapNumberStr, '$.n', 'LONG'), "
+ + "jsonExtractScalarFory(myMapNumberStr, '$.n', 'LONG') FROM " +
getTableName() + " LIMIT 1000";
+ rows = postQuery(query).get("resultTable").get("rows");
+ assertTrue(rows.size() > 0, "expected non-empty numeric result set");
+ for (JsonNode row : rows) {
+ assertEquals(row.get(1).asLong(), row.get(0).asLong(),
+ "streaming Fory extraction must equal jsonExtractScalar");
}
query = "SELECT jsonExtractScalar(myMapBytes, '$.k1', 'STRING'), "
+ "jsonExtractScalarFast(myMapBytes, '$.k1', 'STRING'), "
- + "jsonExtractScalarFirstMatch(myMapBytes, '$.k1', 'STRING') FROM " +
getTableName() + " LIMIT 1000";
+ + "jsonExtractScalarFirstMatch(myMapBytes, '$.k1', 'STRING'), "
+ + "jsonExtractScalarFory(myMapBytes, '$.k1', 'STRING') FROM " +
getTableName() + " LIMIT 1000";
rows = postQuery(query).get("resultTable").get("rows");
assertTrue(rows.size() > 0, "expected non-empty BYTES result set");
for (JsonNode row : rows) {
String jayway = row.get(0).asText();
assertEquals(row.get(1).asText(), jayway, "BYTES fast extraction must
equal jsonExtractScalar");
assertEquals(row.get(2).asText(), jayway, "BYTES first-match extraction
must equal Jayway on clean data");
+ assertEquals(row.get(3).asText(), jayway, "BYTES Fory fallback must
equal Jayway on clean data");
}
}
diff --git
a/pinot-integration-tests/src/test/resources/udf-test-results/all-functions.yaml
b/pinot-integration-tests/src/test/resources/udf-test-results/all-functions.yaml
index be6b3059d1c..5c44ac2f86b 100644
---
a/pinot-integration-tests/src/test/resources/udf-test-results/all-functions.yaml
+++
b/pinot-integration-tests/src/test/resources/udf-test-results/all-functions.yaml
@@ -997,6 +997,10 @@ jsonextractscalarfirstmatch:
scalar: null
transform:
"org.apache.pinot.core.operator.transform.function.JsonExtractScalarTransformFunction.FirstMatch"
udf: null
+jsonextractscalarfory:
+ scalar: null
+ transform:
"org.apache.pinot.core.operator.transform.function.JsonExtractScalarTransformFunction.Fory"
+ udf: null
jsonformat:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonFormat}"
transform: null
@@ -1031,6 +1035,10 @@ jsonpathdoublefirstmatch:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathDoubleFirstMatch}"
transform: null
udf: null
+jsonpathdoublefory:
+ scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathDoubleFory}"
+ transform: null
+ udf: null
jsonpathexists:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathExists}"
transform: null
@@ -1048,6 +1056,10 @@ jsonpathlongfirstmatch:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathLongFirstMatch}"
transform: null
udf: null
+jsonpathlongfory:
+ scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathLongFory}"
+ transform: null
+ udf: null
jsonpathstring:
scalar: "ArgumentCountBasedScalarFunction{[2:
org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathString,\
\ 3:
org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathString]}"
@@ -1061,6 +1073,10 @@ jsonpathstringfirstmatch:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathStringFirstMatch}"
transform: null
udf: null
+jsonpathstringfory:
+ scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonPathStringFory}"
+ transform: null
+ udf: null
jsonstringtoarray:
scalar:
"ArgumentCountBasedScalarFunction{org.apache.pinot.common.function.scalar.JsonFunctions.jsonStringToArray}"
transform: null
diff --git a/pinot-perf/pom.xml b/pinot-perf/pom.xml
index 68fd9816503..4009800d0c2 100644
--- a/pinot-perf/pom.xml
+++ b/pinot-perf/pom.xml
@@ -113,6 +113,10 @@
<groupId>org.openjdk.jmh</groupId>
<artifactId>jmh-core</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
+ </dependency>
<dependency>
<groupId>net.sf.jopt-simple</groupId>
<artifactId>jopt-simple</artifactId>
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkForyJsonFallback.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkForyJsonFallback.java
new file mode 100644
index 00000000000..317826fda2f
--- /dev/null
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkForyJsonFallback.java
@@ -0,0 +1,109 @@
+/**
+ * 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.concurrent.TimeUnit;
+import org.apache.pinot.common.function.ForyJsonPathExtractor;
+import org.apache.pinot.common.function.SimpleJsonPath;
+import org.apache.pinot.common.function.scalar.JsonFunctions;
+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.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;
+
+
+/// Compares normal, Fast, and Fory production scalar functions on values that
require a reference-parser fallback
+/// and on documents beyond Fory's former depth-20 limit. Run with multiple
JMH threads to expose parser-pool
+/// contention as well as per-row exception/fallback costs.
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@Warmup(iterations = 3, time = 2)
+@Measurement(iterations = 5, time = 3)
+@State(Scope.Thread)
+public class BenchmarkForyJsonFallback {
+ private static final String DEFAULT_VALUE = "DEFAULT";
+
+ @Param({"object", "array", "deepSelected", "deepUnrelated"})
+ private String _scenario;
+
+ private String _json;
+ private String _path;
+
+ @Setup
+ public void setUp() {
+ if (!ForyJsonPathExtractor.isAvailable()) {
+ throw new IllegalStateException("Fory JSON is unavailable; refusing to
publish fallback results as Fory");
+ }
+ switch (_scenario) {
+ case "object":
+ _json = "{\"v\":{\"n\":1}}";
+ _path = "$.v";
+ break;
+ case "array":
+ _json = "{\"v\":[1,2,3]}";
+ _path = "$.v";
+ break;
+ case "deepSelected":
+ _json = "{\"a\":".repeat(25) + "\"value\"" + "}".repeat(25);
+ _path = "$." + "a.".repeat(24) + "a";
+ break;
+ case "deepUnrelated":
+ _json = "{\"v\":\"value\",\"deep\":" + "{\"a\":".repeat(25) + "1" +
"}".repeat(25) + "}";
+ _path = "$.v";
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported scenario: " +
_scenario);
+ }
+
+ String expected = JsonFunctions.jsonPathString(_json, _path,
DEFAULT_VALUE);
+ if (!expected.equals(JsonFunctions.jsonPathStringFast(_json, _path,
DEFAULT_VALUE))
+ || !expected.equals(JsonFunctions.jsonPathStringFory(_json, _path,
DEFAULT_VALUE))) {
+ throw new IllegalStateException("JSON functions disagree for scenario: "
+ _scenario);
+ }
+ Object directResult = ForyJsonPathExtractor.extract(_json,
SimpleJsonPath.compile(_path));
+ boolean expectedFallback = _scenario.equals("object") ||
_scenario.equals("array");
+ if (ForyJsonPathExtractor.isFallbackRequired(directResult) !=
expectedFallback
+ || !expectedFallback && !expected.equals(directResult)) {
+ throw new IllegalStateException("Fory did not directly exercise the
expected path for scenario: " + _scenario);
+ }
+ }
+
+ @Benchmark
+ public String jsonPathStringJayway() {
+ return JsonFunctions.jsonPathString(_json, _path, DEFAULT_VALUE);
+ }
+
+ @Benchmark
+ public String jsonPathStringFast() {
+ return JsonFunctions.jsonPathStringFast(_json, _path, DEFAULT_VALUE);
+ }
+
+ @Benchmark
+ public String jsonPathStringFory() {
+ return JsonFunctions.jsonPathStringFory(_json, _path, DEFAULT_VALUE);
+ }
+}
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonExtractScalarQuery.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonExtractScalarQuery.java
new file mode 100644
index 00000000000..39a53ebc358
--- /dev/null
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonExtractScalarQuery.java
@@ -0,0 +1,351 @@
+/**
+ * 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.Objects;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.common.function.ForyJsonPathExtractor;
+import org.apache.pinot.common.function.SimpleJsonPath;
+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.
+/// Result dispatch is selected once during setup, outside the measured
per-row loop. The type-specific JSON literals
+/// have equal encoded lengths and occupy the same early/late field locations
so STRING, LONG, and DOUBLE comparisons
+/// do not accidentally measure different document layouts.
+@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_FORMAT = "{"
+ + "\"earlyValue\":%s,"
+ +
"\"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\"],"
+ + "\"lateValue\":%s} ";
+ private static final String LATE_FIELD_MARKER = "\"lateValue\":";
+
+ @Param({"early", "late"})
+ private String _fieldPosition;
+
+ @Param({"700", "8192", "65536"})
+ private int _documentBytes;
+
+ /// STRING is an intentional Jayway fallback for precision-safe coercion.
Add `-p _resultType=STRING` explicitly to
+ /// characterize that public-function fallback; default trials cover only
actual Fory streaming.
+ @Param({"LONG", "DOUBLE"})
+ private DataType _resultType;
+
+ /// Use `-p _pathResult=missing` to measure explicit defaults without
multiplying the default suite.
+ @Param({"hit"})
+ private String _pathResult;
+
+ private ValueBlock _valueBlock;
+ private JsonExtractScalarTransformFunction _jayway;
+ private JsonExtractScalarTransformFunction _fast;
+ private JsonExtractScalarTransformFunction _firstMatch;
+ private JsonExtractScalarTransformFunction _fory;
+
+ @Setup
+ public void setUp() {
+ if (_resultType != DataType.STRING &&
!ForyJsonPathExtractor.isAvailable()) {
+ throw new IllegalStateException("Fory JSON is unavailable; refusing to
publish fallback results as Fory");
+ }
+ String json = buildJson(_documentBytes, _resultType);
+ String[] jsonRows = new String[BLOCK_ROWS];
+ Arrays.fill(jsonRows, json);
+ TransformFunction input = new StringArrayTransformFunction(jsonRows);
+ boolean early = "early".equals(_fieldPosition);
+ boolean hit = "hit".equals(_pathResult);
+ String path = "$." + (early ? "early" : "late") + (hit ? "Value" :
"Ghost");
+ Object defaultValue = defaultValue(_resultType);
+ List<TransformFunction> arguments = List.of(input,
literal(DataType.STRING, path),
+ literal(DataType.STRING, _resultType.name()), literal(_resultType,
defaultValue));
+
+ _valueBlock = new FixedValueBlock(BLOCK_ROWS);
+ if (_resultType != DataType.STRING) {
+ Object directForyResult = ForyJsonPathExtractor.extract(json,
SimpleJsonPath.compile(path));
+ Object expectedForyResult = hit ? hitValue(_resultType, early) : null;
+ if (!Objects.equals(directForyResult, expectedForyResult)) {
+ throw new IllegalStateException("Direct Fory extraction produced an
unexpected " + _resultType + " result");
+ }
+ }
+ _jayway = initialize(new JsonExtractScalarTransformFunction(), arguments);
+ _fast = initialize(new JsonExtractScalarTransformFunction.Fast(),
arguments);
+ _firstMatch = initialize(new
JsonExtractScalarTransformFunction.FirstMatch(), arguments);
+ _fory = initialize(new JsonExtractScalarTransformFunction.Fory(),
arguments);
+
+ Object expectedRows = expectedRows(_resultType, hit ?
hitValue(_resultType, early) : defaultValue);
+ Object jaywayRows = apply(_jayway);
+ assertResultsEqual("jsonExtractScalar", expectedRows, jaywayRows);
+ assertResultsEqual("jsonExtractScalarFast", jaywayRows, apply(_fast));
+ assertResultsEqual("jsonExtractScalarFirstMatch", jaywayRows,
apply(_firstMatch));
+ assertResultsEqual("jsonExtractScalarFory", jaywayRows, apply(_fory));
+ }
+
+ private JsonExtractScalarTransformFunction
initialize(JsonExtractScalarTransformFunction function,
+ List<TransformFunction> arguments) {
+ function.init(arguments, Map.<String, ColumnContext>of(), false);
+ return function;
+ }
+
+ private Object apply(JsonExtractScalarTransformFunction function) {
+ switch (_resultType) {
+ case STRING:
+ return function.transformToStringValuesSV(_valueBlock);
+ case LONG:
+ return function.transformToLongValuesSV(_valueBlock);
+ case DOUBLE:
+ return function.transformToDoubleValuesSV(_valueBlock);
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ _resultType);
+ }
+ }
+
+ private static LiteralTransformFunction literal(DataType dataType, Object
value) {
+ return new LiteralTransformFunction(new LiteralContext(dataType, value));
+ }
+
+ private static String buildJson(int targetBytes, DataType resultType) {
+ String baseJson = String.format(BASE_JSON_FORMAT, jsonLiteral(resultType,
true), jsonLiteral(resultType, false));
+ if (baseJson.length() >= targetBytes) {
+ return baseJson;
+ }
+ int markerOffset = baseJson.indexOf(LATE_FIELD_MARKER);
+ String paddingPrefix = "\"padding\":\"";
+ String paddingSuffix = "\",";
+ int paddingLength = targetBytes - baseJson.length() -
paddingPrefix.length() - paddingSuffix.length();
+ if (paddingLength <= 0) {
+ return baseJson;
+ }
+ return baseJson.substring(0, markerOffset) + paddingPrefix +
"x".repeat(paddingLength) + paddingSuffix
+ + baseJson.substring(markerOffset);
+ }
+
+ private static String jsonLiteral(DataType resultType, boolean early) {
+ switch (resultType) {
+ case STRING:
+ return early ? "\"S\"" : "\"T\"";
+ case LONG:
+ return early ? "170" : "190";
+ case DOUBLE:
+ return early ? "1.7" : "1.9";
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ resultType);
+ }
+ }
+
+ private static Object hitValue(DataType resultType, boolean early) {
+ switch (resultType) {
+ case STRING:
+ return early ? "S" : "T";
+ case LONG:
+ return early ? 170L : 190L;
+ case DOUBLE:
+ return early ? 1.7d : 1.9d;
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ resultType);
+ }
+ }
+
+ private static Object defaultValue(DataType resultType) {
+ switch (resultType) {
+ case STRING:
+ return "DEFAULT";
+ case LONG:
+ return -1L;
+ case DOUBLE:
+ return -1d;
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ resultType);
+ }
+ }
+
+ private static Object expectedRows(DataType resultType, Object
expectedValue) {
+ switch (resultType) {
+ case STRING:
+ String[] stringValues = new String[BLOCK_ROWS];
+ Arrays.fill(stringValues, (String) expectedValue);
+ return stringValues;
+ case LONG:
+ long[] longValues = new long[BLOCK_ROWS];
+ Arrays.fill(longValues, (Long) expectedValue);
+ return longValues;
+ case DOUBLE:
+ double[] doubleValues = new double[BLOCK_ROWS];
+ Arrays.fill(doubleValues, (Double) expectedValue);
+ return doubleValues;
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ resultType);
+ }
+ }
+
+ private void assertResultsEqual(String functionName, Object expected, Object
actual) {
+ boolean equal;
+ switch (_resultType) {
+ case STRING:
+ equal = Arrays.equals((String[]) expected, (String[]) actual);
+ break;
+ case LONG:
+ equal = Arrays.equals((long[]) expected, (long[]) actual);
+ break;
+ case DOUBLE:
+ equal = Arrays.equals((double[]) expected, (double[]) actual);
+ break;
+ default:
+ throw new IllegalStateException("Unsupported benchmark result type: "
+ _resultType);
+ }
+ if (!equal) {
+ throw new IllegalStateException(functionName + " produced an unexpected
" + _resultType + " result for "
+ + _fieldPosition + '/' + _pathResult);
+ }
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(BLOCK_ROWS)
+ public Object queryJayway() {
+ return apply(_jayway);
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(BLOCK_ROWS)
+ public Object queryFast() {
+ return apply(_fast);
+ }
+
+ /// Auxiliary comparator: unlike the primary parity-preserving variants,
FirstMatch intentionally has weaker
+ /// duplicate-key and malformed-tail semantics.
+ @Benchmark
+ @OperationsPerInvocation(BLOCK_ROWS)
+ public Object queryFirstMatch() {
+ return apply(_firstMatch);
+ }
+
+ /// For STRING, this measures the production Fory function's intentional
Jayway fallback rather than Fory parsing.
+ @Benchmark
+ @OperationsPerInvocation(BLOCK_ROWS)
+ public Object queryFory() {
+ return apply(_fory);
+ }
+
+ public static void main(String[] arguments)
+ throws Exception {
+ Options options = new
OptionsBuilder().include(BenchmarkJsonExtractScalarQuery.class.getSimpleName()).build();
+ new Runner(options).run();
+ }
+
+ private static final class StringArrayTransformFunction extends
BaseTransformFunction {
+ private final String[] _values;
+
+ private StringArrayTransformFunction(String[] values) {
+ _values = values;
+ }
+
+ @Override
+ public String getName() {
+ return "stringArrayInput";
+ }
+
+ @Override
+ public TransformResultMetadata getResultMetadata() {
+ return STRING_METADATA;
+ }
+
+ @Override
+ public String[] transformToStringValuesSV(ValueBlock valueBlock) {
+ return _values;
+ }
+ }
+
+ private static final class FixedValueBlock implements ValueBlock {
+ private final int _numDocs;
+
+ private FixedValueBlock(int numDocs) {
+ _numDocs = numDocs;
+ }
+
+ @Override
+ public int getNumDocs() {
+ return _numDocs;
+ }
+
+ @Override
+ public int[] getDocIds() {
+ return null;
+ }
+
+ @Override
+ public BlockValSet getBlockValueSet(ExpressionContext expression) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlockValSet getBlockValueSet(String column) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public BlockValSet getBlockValueSet(String[] paths) {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java
index effb8925962..2e1b8fa6d84 100644
---
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java
+++
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java
@@ -19,6 +19,7 @@
package org.apache.pinot.perf;
import com.jayway.jsonpath.Configuration;
+import com.jayway.jsonpath.DocumentContext;
import com.jayway.jsonpath.JsonPath;
import com.jayway.jsonpath.Option;
import com.jayway.jsonpath.ParseContext;
@@ -26,7 +27,9 @@ import com.jayway.jsonpath.Predicate;
import com.jayway.jsonpath.spi.json.JacksonJsonProvider;
import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider;
import java.util.concurrent.TimeUnit;
+import org.apache.fory.json.ForyJson;
import org.apache.pinot.common.function.FastJsonPathExtractor;
+import org.apache.pinot.common.function.ForyJsonPathExtractor;
import org.apache.pinot.common.function.SimpleJsonPath;
import org.apache.pinot.common.function.scalar.JsonFunctions;
import org.openjdk.jmh.annotations.Benchmark;
@@ -47,31 +50,36 @@ import org.openjdk.jmh.runner.options.OptionsBuilder;
/// Compares JsonPath extraction through Jayway (today's implementation, which
builds a full Jackson DOM of the
-/// document and then walks to the field) against [FastJsonPathExtractor].
+/// document and then walks to the field), Fory's streaming and dynamic-tree
parsers, and [FastJsonPathExtractor].
///
/// The `fieldPosition` parameter puts the extracted field either near the
start or at the very end of a
-/// ~700 byte nested event payload, because that is what decides whether early
exit can pay off.
+/// nested event payload, because that is what decides whether early exit can
pay off. `documentBytes` pads the
+/// payload immediately before the late field so parser scaling is visible
without changing the addressed values.
///
-/// The single-column benchmarks are also the per-row cost of
`jsonExtractScalar`: that transform function
-/// does exactly `parseContext.parse(row).read(jsonPath)` per row, so
measuring the extraction in isolation
-/// measures it without the surrounding `ValueBlock` scaffolding.
+/// The scalar-function methods measure the production `jsonPath*` ingestion
wrappers. The lower-level single-column
+/// methods isolate parser/traversal strategies;
[BenchmarkJsonExtractScalarQuery] is the authoritative query-transform
+/// measurement.
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@Fork(1)
@Warmup(iterations = 3, time = 2)
@Measurement(iterations = 5, time = 3)
-@State(Scope.Benchmark)
+@State(Scope.Thread)
public class BenchmarkJsonPathExtraction {
private static final Predicate[] NO_PREDICATES = new Predicate[0];
+ private static final ForyJson FORY_JSON = ForyJson.builder().build();
+ private static final String STRING_DEFAULT = "DEFAULT";
+ private static final long LONG_DEFAULT = -1L;
+ private static final double DOUBLE_DEFAULT = -1.25d;
/// Exactly the context `JsonExtractScalarTransformFunction` and
`JsonFunctions` use.
private static final ParseContext PARSE_CONTEXT = JsonPath.using(
new Configuration.ConfigurationBuilder().jsonProvider(new
JacksonJsonProvider())
.mappingProvider(new
JacksonMappingProvider()).options(Option.SUPPRESS_EXCEPTIONS).build());
- private static final String JSON = "{"
+ private static final String BASE_JSON = "{"
+ "\"ts\":1719878400123,"
- +
"\"user\":{\"id\":\"u-19283\",\"country\":\"US\",\"tier\":\"gold\",\"age\":41},"
+ +
"\"user\":{\"id\":\"u-19283\",\"country\":\"US\",\"score\":17.25,\"age\":41},"
+
"\"event\":{\"name\":\"checkout\",\"cart\":[{\"sku\":\"A1\",\"qty\":2,\"price\":19.99},"
+
"{\"sku\":\"B7\",\"qty\":1,\"price\":149.5},{\"sku\":\"C3\",\"qty\":5,\"price\":3.25}],"
+ "\"total\":352.73,\"currency\":\"USD\"},"
@@ -81,17 +89,30 @@ public class BenchmarkJsonPathExtraction {
+
"\"tags\":[\"mobile\",\"ios\",\"returning\",\"promo-eligible\",\"newsletter\"],"
+
"\"session\":{\"id\":\"s-aaaabbbbccccdddd\",\"start\":1719878300000,\"pages\":14,\"referrer\":"
+ "\"https://example.com/landing?utm_source=x&utm_medium=y\"},";
- + "\"trailer\":{\"country\":\"DE\",\"note\":\"last field in the
document\"}"
+ + "\"trailer\":{\"country\":\"DE\",\"double\":19.25,\"long\":19}"
+ "}";
/// Four derived columns, spread through the document, as an ingestion
`transformConfigs` would pull.
private static final String[] FOUR_PATHS = {"$.user.country",
"$.event.currency", "$.device.os", "$.geo.city"};
+ private static final String COMPLEX_PATH = "$.event.cart[*].sku";
@Param({"early", "late"})
private String _fieldPosition;
+ @Param({"700", "8192", "65536"})
+ private int _documentBytes;
+
+ /// Use `-p _valueCase=missing` to measure the explicit-default path without
doubling the default benchmark suite.
+ @Param({"hit"})
+ private String _valueCase;
+
+ private String _json;
private String _path;
+ private String _longPath;
+ private String _doublePath;
private SimpleJsonPath _simplePath;
+ private SimpleJsonPath _simpleLongPath;
+ private SimpleJsonPath _simpleDoublePath;
private SimpleJsonPath[] _simpleFourPaths;
private Object[] _fourResults;
@@ -104,53 +125,211 @@ public class BenchmarkJsonPathExtraction {
@Setup(Level.Trial)
public void setUp() {
- _path = "early".equals(_fieldPosition) ? "$.user.country" :
"$.trailer.country";
+ if (!ForyJsonPathExtractor.isAvailable()) {
+ throw new IllegalStateException("Fory JSON is unavailable; refusing to
publish fallback results as Fory");
+ }
+ _json = buildJson(_documentBytes);
+ boolean missing;
+ if ("hit".equals(_valueCase)) {
+ missing = false;
+ } else if ("missing".equals(_valueCase)) {
+ missing = true;
+ } else {
+ throw new IllegalArgumentException("Unsupported value case: " +
_valueCase);
+ }
+ if (missing) {
+ _path = "$.missing";
+ _longPath = _path;
+ _doublePath = _path;
+ } else if ("early".equals(_fieldPosition)) {
+ _path = "$.user.country";
+ _longPath = "$.user.age";
+ _doublePath = "$.user.score";
+ } else {
+ _path = "$.trailer.country";
+ _longPath = "$.trailer.long";
+ _doublePath = "$.trailer.double";
+ }
_simplePath = SimpleJsonPath.compile(_path);
+ _simpleLongPath = SimpleJsonPath.compile(_longPath);
+ _simpleDoublePath = SimpleJsonPath.compile(_doublePath);
_simpleFourPaths = new SimpleJsonPath[FOUR_PATHS.length];
for (int i = 0; i < FOUR_PATHS.length; i++) {
_simpleFourPaths[i] = SimpleJsonPath.compile(FOUR_PATHS[i]);
}
_fourResults = new Object[FOUR_PATHS.length];
+
+ String expectedString = missing ? STRING_DEFAULT :
"early".equals(_fieldPosition) ? "US" : "DE";
+ long expectedLong = missing ? LONG_DEFAULT :
"early".equals(_fieldPosition) ? 41L : 19L;
+ double expectedDouble = missing ? DOUBLE_DEFAULT :
"early".equals(_fieldPosition) ? 17.25d : 19.25d;
+ verifyScalarFunctions(expectedString, expectedLong, expectedDouble);
+ for (String path : FOUR_PATHS) {
+ String jayway = JsonFunctions.jsonPathString(_json, path, "");
+ String fory = JsonFunctions.jsonPathStringFory(_json, path, "");
+ if (!jayway.equals(fory)) {
+ throw new IllegalStateException("Fory result does not match Jayway for
" + path);
+ }
+ }
+ String expectedCart = "[\"A1\",\"B7\",\"C3\"]";
+ if (!expectedCart.equals(JsonFunctions.jsonPathStringFory(_json,
COMPLEX_PATH, ""))) {
+ throw new IllegalStateException("Complex-path fallback does not match
the expected cart");
+ }
+ }
+
+ private void verifyScalarFunctions(String expectedString, long expectedLong,
double expectedDouble) {
+ Object extractedString = ForyJsonPathExtractor.extract(_json, _simplePath);
+ Object extractedLong = ForyJsonPathExtractor.extract(_json,
_simpleLongPath);
+ Object extractedDouble = ForyJsonPathExtractor.extract(_json,
_simpleDoublePath);
+ if ("missing".equals(_valueCase)) {
+ if (extractedString != null || extractedLong != null || extractedDouble
!= null) {
+ throw new IllegalStateException("Fory returned a value for a missing
benchmark path");
+ }
+ } else if (!expectedString.equals(extractedString) || ((Number)
extractedLong).longValue() != expectedLong
+ || Double.compare(((Number) extractedDouble).doubleValue(),
expectedDouble) != 0) {
+ throw new IllegalStateException("Direct Fory extraction produced an
unexpected benchmark result");
+ }
+
+ String jaywayString = JsonFunctions.jsonPathString(_json, _path,
STRING_DEFAULT);
+ String fastString = JsonFunctions.jsonPathStringFast(_json, _path,
STRING_DEFAULT);
+ String foryString = JsonFunctions.jsonPathStringFory(_json, _path,
STRING_DEFAULT);
+ if (!expectedString.equals(jaywayString) ||
!jaywayString.equals(fastString) || !jaywayString.equals(foryString)) {
+ throw new IllegalStateException("JSON string extractors disagree for " +
_path);
+ }
+
+ long jaywayLong = JsonFunctions.jsonPathLong(_json, _longPath,
LONG_DEFAULT);
+ long fastLong = JsonFunctions.jsonPathLongFast(_json, _longPath,
LONG_DEFAULT);
+ long foryLong = JsonFunctions.jsonPathLongFory(_json, _longPath,
LONG_DEFAULT);
+ if (jaywayLong != expectedLong || fastLong != jaywayLong || foryLong !=
jaywayLong) {
+ throw new IllegalStateException("JSON long extractors disagree for " +
_longPath);
+ }
+
+ double jaywayDouble = JsonFunctions.jsonPathDouble(_json, _doublePath,
DOUBLE_DEFAULT);
+ double fastDouble = JsonFunctions.jsonPathDoubleFast(_json, _doublePath,
DOUBLE_DEFAULT);
+ double foryDouble = JsonFunctions.jsonPathDoubleFory(_json, _doublePath,
DOUBLE_DEFAULT);
+ if (Double.compare(jaywayDouble, expectedDouble) != 0 ||
Double.compare(fastDouble, jaywayDouble) != 0
+ || Double.compare(foryDouble, jaywayDouble) != 0) {
+ throw new IllegalStateException("JSON double extractors disagree for " +
_doublePath);
+ }
+ }
+
+ private static String buildJson(int targetBytes) {
+ if (BASE_JSON.length() >= targetBytes) {
+ return BASE_JSON;
+ }
+ String marker = "\"trailer\":";
+ int markerOffset = BASE_JSON.indexOf(marker);
+ String paddingPrefix = "\"padding\":\"";
+ String paddingSuffix = "\",";
+ if (BASE_JSON.length() + paddingPrefix.length() + paddingSuffix.length()
>= targetBytes) {
+ return BASE_JSON;
+ }
+ 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);
}
@Benchmark
public Object jaywayOneColumn() {
- return PARSE_CONTEXT.parse(JSON).read(_path, NO_PREDICATES);
+ return PARSE_CONTEXT.parse(_json).read(_path, NO_PREDICATES);
+ }
+
+ /// Fory dynamic-tree parsing plus Jayway traversal, retained as a
non-production comparison baseline.
+ @Benchmark
+ public Object foryOneColumn() {
+ Object root = FORY_JSON.fromJson(_json, Object.class);
+ return PARSE_CONTEXT.parse(root).read(_path, NO_PREDICATES);
}
@Benchmark
public Object fastOneColumnFullScan() {
- return FastJsonPathExtractor.extract(JSON, _simplePath, false, false);
+ return FastJsonPathExtractor.extract(_json, _simplePath, false, false);
}
@Benchmark
public Object fastOneColumnEarlyExit() {
- return FastJsonPathExtractor.extract(JSON, _simplePath, false, true);
+ return FastJsonPathExtractor.extract(_json, _simplePath, false, true);
}
/// The existing Jayway scalar function (applicability check + String
coercion).
@Benchmark
public String jsonPathStringJayway() {
- return JsonFunctions.jsonPathString(JSON, _path, "");
+ return JsonFunctions.jsonPathString(_json, _path, STRING_DEFAULT);
}
- /// The opt-in fast scalar function, full scan (exact parity).
+ /// The production Fory scalar function (applicability check + String
coercion).
+ @Benchmark
+ public String jsonPathStringFory() {
+ return JsonFunctions.jsonPathStringFory(_json, _path, STRING_DEFAULT);
+ }
+
+ /// The opt-in fast scalar function, full scan with fallback for unsupported
or failed extraction.
@Benchmark
public String jsonPathStringFast() {
- return JsonFunctions.jsonPathStringFast(JSON, _path, "");
+ return JsonFunctions.jsonPathStringFast(_json, _path, STRING_DEFAULT);
}
/// The opt-in fast scalar function, early exit / first match.
@Benchmark
public String jsonPathStringFirstMatch() {
- return JsonFunctions.jsonPathStringFirstMatch(JSON, _path, "");
+ return JsonFunctions.jsonPathStringFirstMatch(_json, _path,
STRING_DEFAULT);
+ }
+
+ /// The existing Jayway scalar function (applicability check + long
coercion).
+ @Benchmark
+ public long jsonPathLongJayway() {
+ return JsonFunctions.jsonPathLong(_json, _longPath, LONG_DEFAULT);
+ }
+
+ /// The opt-in fast scalar function, full scan with fallback for unsupported
or failed extraction.
+ @Benchmark
+ public long jsonPathLongFast() {
+ return JsonFunctions.jsonPathLongFast(_json, _longPath, LONG_DEFAULT);
+ }
+
+ /// The production Fory scalar function (applicability check + long
coercion).
+ @Benchmark
+ public long jsonPathLongFory() {
+ return JsonFunctions.jsonPathLongFory(_json, _longPath, LONG_DEFAULT);
+ }
+
+ /// The existing Jayway scalar function (applicability check + double
coercion).
+ @Benchmark
+ public double jsonPathDoubleJayway() {
+ return JsonFunctions.jsonPathDouble(_json, _doublePath, DOUBLE_DEFAULT);
+ }
+
+ /// The opt-in fast scalar function, full scan with fallback for unsupported
or failed extraction.
+ @Benchmark
+ public double jsonPathDoubleFast() {
+ return JsonFunctions.jsonPathDoubleFast(_json, _doublePath,
DOUBLE_DEFAULT);
+ }
+
+ /// The production Fory scalar function (applicability check + double
coercion).
+ @Benchmark
+ public double jsonPathDoubleFory() {
+ return JsonFunctions.jsonPathDoubleFory(_json, _doublePath,
DOUBLE_DEFAULT);
+ }
+
+ @Benchmark
+ public String jsonPathStringJaywayComplex() {
+ return JsonFunctions.jsonPathString(_json, COMPLEX_PATH, "");
+ }
+
+ @Benchmark
+ public String jsonPathStringFastComplex() {
+ return JsonFunctions.jsonPathStringFast(_json, COMPLEX_PATH, "");
+ }
+
+ @Benchmark
+ public String jsonPathStringForyComplex() {
+ return JsonFunctions.jsonPathStringFory(_json, COMPLEX_PATH, "");
}
@Benchmark
public Object jaywayFourColumns() {
Object last = null;
for (String path : FOUR_PATHS) {
- last = PARSE_CONTEXT.parse(JSON).read(path, NO_PREDICATES);
+ last = PARSE_CONTEXT.parse(_json).read(path, NO_PREDICATES);
}
return last;
}
@@ -159,14 +338,36 @@ public class BenchmarkJsonPathExtraction {
public Object fastFourColumnsSeparatePasses() {
Object last = null;
for (SimpleJsonPath path : _simpleFourPaths) {
- last = FastJsonPathExtractor.extract(JSON, path, false, false);
+ last = FastJsonPathExtractor.extract(_json, path, false, false);
}
return last;
}
@Benchmark
public Object[] fastFourColumnsSinglePass() {
- FastJsonPathExtractor.extract(JSON, _simpleFourPaths, _fourResults, false,
false);
+ FastJsonPathExtractor.extract(_json, _simpleFourPaths, _fourResults,
false, false);
return _fourResults;
}
+
+ /// Models four independent scalar expressions: each call streams over the
document again.
+ @Benchmark
+ public String foryFourColumnsSeparateParses() {
+ String last = null;
+ for (String path : FOUR_PATHS) {
+ last = JsonFunctions.jsonPathStringFory(_json, path, "");
+ }
+ return last;
+ }
+
+ /// Upper bound for a future parse-sharing optimization; Pinot does not
currently expose this execution shape.
+ @Benchmark
+ public Object foryFourColumnsSingleParse() {
+ Object root = FORY_JSON.fromJson(_json, Object.class);
+ DocumentContext context = PARSE_CONTEXT.parse(root);
+ Object last = null;
+ for (String path : FOUR_PATHS) {
+ last = context.read(path, NO_PREDICATES);
+ }
+ return last;
+ }
}
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
index fb278aef2d1..4416e47bb52 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
@@ -146,7 +146,7 @@ public class QueryCompilationTest extends
QueryEnvironmentTestBase {
/// `SqlNode` in that position: operand checking runs before
`PinotEvaluateLiteralRule` folds constant
/// expressions, so an argument such as `CONCAT('$.', 'foo')` folds to a
literal and plans and executes
/// correctly. Requiring [org.apache.calcite.sql.type.OperandTypes#LITERAL]
there would reject these queries,
- /// which plan and execute successfully on master. Regression guard for all
three JSON scalar transforms, which
+ /// which plan and execute successfully on master. Regression guard for all
JSON scalar transforms, which
/// share one operand checker;
`QueryRunnerTest#provideTestSqlWithExecutionException` covers the end-to-end
half,
/// asserting that a folded path is actually applied on the leaf stage.
///
@@ -154,8 +154,8 @@ public class QueryCompilationTest extends
QueryEnvironmentTestBase {
/// for return-type inference to see it.
@Test
public void testJsonExtractScalarAcceptsFoldableJsonPath() {
- List<String> functions =
- List.of("JSON_EXTRACT_SCALAR", "JSON_EXTRACT_SCALAR_FAST",
"JSON_EXTRACT_SCALAR_FIRST_MATCH");
+ List<String> functions = List.of("JSON_EXTRACT_SCALAR",
"JSON_EXTRACT_SCALAR_FAST",
+ "JSON_EXTRACT_SCALAR_FIRST_MATCH", "JSON_EXTRACT_SCALAR_FORY");
for (String function : functions) {
for (String path : List.of("CONCAT('$.', 'foo')", "CAST('$.foo' AS
VARCHAR)", "UPPER('$.foo')")) {
String query = "SELECT " + function + "(col1, " + path + ", 'INT')
FROM a";
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryEnvironmentTestBase.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryEnvironmentTestBase.java
index e26dd6222e2..fa42e993f27 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryEnvironmentTestBase.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryEnvironmentTestBase.java
@@ -258,6 +258,8 @@ public class QueryEnvironmentTestBase {
new Object[]{"SELECT JSON_EXTRACT_SCALAR_FIRST_MATCH(col1, '$.foo',
'LONG', '0') FROM a"},
new Object[]{"SELECT JSON_EXTRACT_SCALAR_FIRST_MATCH(col1, '$.foo',
'STRING_ARRAY') FROM a"},
new Object[]{"SELECT JSON_EXTRACT_SCALAR_FIRST_MATCH(col1, '$.foo',
'TIMESTAMP_ARRAY') FROM a"},
+ new Object[]{"SELECT JSON_EXTRACT_SCALAR_FORY(col1, '$.foo', 'LONG',
'0') FROM a"},
+ new Object[]{"SELECT JSON_EXTRACT_SCALAR_FORY(col1, '$.foo',
'DOUBLE_ARRAY') FROM a"},
new Object[]{"SELECT ts_timestamp FROM a WHERE ts_timestamp BETWEEN
TIMESTAMP '2016-01-01 00:00:00' AND "
+ "TIMESTAMP '2016-01-01 10:00:00'"},
new Object[]{"SELECT ts_timestamp FROM a WHERE ts_timestamp >=
CAST(1454284798000 AS TIMESTAMP)"},
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
index f00dd6413f2..00121f59d66 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTest.java
@@ -330,9 +330,9 @@ public class QueryRunnerTest extends QueryRunnerTestBase {
testCases.add(new Object[]{
"SELECT CAST(jsonExtractScalar(col1, CONCAT('pa', 'th'), 'INT') AS
INT) FROM a", "Cannot resolve JSON path"});
// - the flip side: a jsonPath that cannot fold to a literal is still
rejected, on the leaf stage rather than
- // during validation. Covers all three variants, which share the
operand type checker.
+ // during validation. Covers all four variants, which share the
operand type checker.
for (String jsonExtractScalar : new String[]{
- "jsonExtractScalar", "jsonExtractScalarFast",
"jsonExtractScalarFirstMatch"
+ "jsonExtractScalar", "jsonExtractScalarFast",
"jsonExtractScalarFirstMatch", "jsonExtractScalarFory"
}) {
testCases.add(new Object[]{
"SELECT " + jsonExtractScalar + "(col1, col2, 'INT') FROM a",
diff --git a/pom.xml b/pom.xml
index f28203a1964..f249ac32891 100644
--- a/pom.xml
+++ b/pom.xml
@@ -191,6 +191,7 @@
<helix.version>2.0.1</helix.version>
<zkclient.version>0.11</zkclient.version>
<jackson.version>2.22.1</jackson.version>
+ <fory.version>1.6.0</fory.version>
<zookeeper.version>3.9.5</zookeeper.version>
<async-http-client.version>3.0.12</async-http-client.version>
<jersey.version>2.48</jersey.version>
@@ -1432,6 +1433,11 @@
<artifactId>json-path</artifactId>
<version>${jsonpath.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.fory</groupId>
+ <artifactId>fory-json</artifactId>
+ <version>${fory.version}</version>
+ </dependency>
<dependency>
<groupId>net.minidev</groupId>
<artifactId>json-smart</artifactId>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
