From a292ccb35c729e64ccfc5e6e98ba56b0380def81 Mon Sep 17 00:00:00 2001 From: Xiang Fu Date: Wed, 15 Jul 2026 22:49:39 -0700 Subject: [PATCH] Add streaming jsonPath*Fast / jsonPath*FirstMatch extractor functions Ingestion JsonPath extraction goes through Jayway (parse(json).read(path)), which builds a full Jackson DOM of the document and only then walks to the field - once per row per derived column, a major ingestion CPU cost. This adds a streaming Jackson-based extractor for simple linear paths ($ followed only by .key, ['key'], [int]), exposed as opt-in scalar functions so each derived column chooses independently, in two clearly named families: jsonPathStringFast/jsonPathLongFast/jsonPathDoubleFast(object, path, default) - streaming, full scan; identical result to the existing Jayway jsonPath* functions, just faster. jsonPathStringFirstMatch/...FirstMatch(object, path, default) - streaming, stops at the addressed field; faster still when the field is early, at the cost of two documented behavior changes on undefined/corrupt input (duplicate keys resolve to the first occurrence; a document malformed strictly after the field is not rejected). Both resolve a simple linear path in a single streaming pass and fall back to Jayway for complex paths (wildcards, deep scan, filters, unions, slices, negative indices) and non-JSON input, and also fall back to Jayway on any streaming exception, so an unforeseen streaming bug can only cost a second parse on a row, never a wrong result. The existing jsonPath* functions and the jsonExtractScalar transform are unchanged. Server-local compute only - no wire, segment, or config-schema change. Components: - SimpleJsonPath: compiles/caches the "simple linear chain" JsonPath subset, returning null (-> Jayway) for anything else. Plain ConcurrentHashMap cache (lock-free reads) rather than a size-bounded Guava cache that locks per read. - StreamingJsonPathExtractor: single forward pass with skipChildren() on non-addressed subtrees; single- and multi-path overloads (the multi-path single-pass form and byte[] overloads are engine support for follow-ups). - JsonFunctions: the six opt-in scalar functions above. Testing: - StreamingJsonPathExtractorTest: differential test against the exact Jayway config it replaces (value level and through the Fast functions), a hand- picked matrix (missing paths, invalid JSON, null/empty/whitespace/BOM, type mismatches, nested arrays, unicode, duplicate keys, big/precise numbers, dotted keys), randomized fuzz incl. a dedicated BigDecimal-context fuzz, the one documented full-scan divergence pinned, the FirstMatch duplicate-key divergence pinned, and a FunctionRegistry resolution check. - JsonPathTest: derives columns via the new functions through ingestion transform configs and asserts they equal the Jayway-derived column on both query engines. - BenchmarkJsonPathExtraction (pinot-perf): Jayway vs streaming, single- and multi-column. Note on one full-scan divergence: Jayway materializes the whole document, so it rejects a value it can lex but not materialize (a string over maxStringLength, or a float with an int-overflow exponent under BigDecimal) even outside the addressed path; the streaming extractor skips such subtrees and returns the requested value. It never returns a different value for the addressed path. --- .../pinot/common/function/SimpleJsonPath.java | 217 +++++ .../function/StreamingJsonPathExtractor.java | 326 ++++++++ .../common/function/scalar/JsonFunctions.java | 142 ++++ .../StreamingJsonPathExtractorTest.java | 756 ++++++++++++++++++ .../tests/custom/JsonPathTest.java | 31 + .../perf/BenchmarkJsonPathExtraction.java | 174 ++++ 6 files changed, 1646 insertions(+) create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/function/SimpleJsonPath.java create mode 100644 pinot-common/src/main/java/org/apache/pinot/common/function/StreamingJsonPathExtractor.java create mode 100644 pinot-common/src/test/java/org/apache/pinot/common/function/StreamingJsonPathExtractorTest.java create mode 100644 pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java diff --git a/pinot-common/src/main/java/org/apache/pinot/common/function/SimpleJsonPath.java b/pinot-common/src/main/java/org/apache/pinot/common/function/SimpleJsonPath.java new file mode 100644 index 000000000000..2204ec5c6700 --- /dev/null +++ b/pinot-common/src/main/java/org/apache/pinot/common/function/SimpleJsonPath.java @@ -0,0 +1,217 @@ +/** + * 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.google.common.annotations.VisibleForTesting; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nullable; + + +/// A JsonPath expression restricted to a *simple linear chain*: a `$` root followed only by object-key and +/// array-index steps. Such a path addresses at most one value and can be resolved by a single forward pass +/// over the JSON document, which is what [StreamingJsonPathExtractor] does. +/// +/// Accepted syntax: +/// - `.name` where `name` is made of letters, digits, `_` or `-` +/// - `['name']` or `["name"]` — the quoted key is taken literally (a `.` inside it does **not** nest) +/// - `[123]` — a non-negative array index +/// +/// Everything else compiles to `null` and must fall back to Jayway: wildcards (`*`), deep scan (`..`), +/// filters (`[?(...)]`), unions (`['a','b']`), slices (`[1:2]`), functions, negative indices (`[-1]`, which +/// Jayway resolves from the end and therefore needs the array length), bracket keys containing a backslash +/// (Jayway's bracket-escape handling is not reproducible here), a bare `$`, and any path not starting +/// with `$`. +/// +/// Compilation results are cached per literal path string; in practice a path is a query or transform-config +/// literal, so the cache is warm after the first row. +/// +/// The cache is a plain [ConcurrentHashMap] rather than a Guava `CacheBuilder.maximumSize` cache on purpose: +/// this is read on the per-row path, and Guava's size-bounded cache takes a segment lock on every *read* to +/// maintain its LRU recency queue, which measurably contends across query threads. Growth is bounded instead +/// by simply not admitting new entries past [#CACHE_MAXIMUM_SIZE] - a path that misses a full cache is +/// re-parsed, which is a few dozen character comparisons, so the fallback is cheap and there is no eviction +/// policy to maintain. +/// +/// Immutable and thread-safe. +public final class SimpleJsonPath { + private static final int CACHE_MAXIMUM_SIZE = 10_000; + + /// Sentinel for "not a simple linear path". Distinguishable from any real instance because a real one always + /// has at least one segment (a bare `$` is deliberately rejected). + private static final SimpleJsonPath NOT_SIMPLE = new SimpleJsonPath(new String[0], new int[0]); + + private static final ConcurrentHashMap CACHE = new ConcurrentHashMap<>(); + + /// Segment `i` is an object key when `_keys[i] != null`, and an array index (`_indices[i]`) otherwise. + private final String[] _keys; + private final int[] _indices; + + private SimpleJsonPath(String[] keys, int[] indices) { + _keys = keys; + _indices = indices; + } + + /// Returns the compiled path, or `null` when `jsonPath` is `null` or is not a simple linear chain, in which + /// case the caller must fall back to Jayway. A `null` path is deliberately reported the same way rather than + /// raising, so that the fast path never changes which exception a caller sees - Jayway raises its own. + @Nullable + public static SimpleJsonPath compile(@Nullable String jsonPath) { + if (jsonPath == null) { + return null; + } + SimpleJsonPath compiled = CACHE.get(jsonPath); + if (compiled == null) { + compiled = parse(jsonPath); + // Bound the cache by refusing new entries when full: query paths are attacker-controllable, and re-parsing + // is cheap enough that an eviction policy would cost more than the miss it saves. + if (CACHE.size() < CACHE_MAXIMUM_SIZE) { + CACHE.putIfAbsent(jsonPath, compiled); + } + } + return compiled == NOT_SIMPLE ? null : compiled; + } + + int length() { + return _keys.length; + } + + /// Returns the object key at `depth`, or `null` when that segment is an array index. + @Nullable + String getKey(int depth) { + return _keys[depth]; + } + + /// Returns the array index at `depth`. Only meaningful when [#getKey] returns `null` for the same depth. + int getIndex(int depth) { + return _indices[depth]; + } + + boolean matchesKey(int depth, String key) { + return key.equals(_keys[depth]); + } + + boolean matchesIndex(int depth, int index) { + return _keys[depth] == null && _indices[depth] == index; + } + + @VisibleForTesting + static SimpleJsonPath parse(String jsonPath) { + int length = jsonPath.length(); + if (length < 2 || jsonPath.charAt(0) != '$') { + return NOT_SIMPLE; + } + List keys = new ArrayList<>(4); + List indices = new ArrayList<>(4); + int pos = 1; + while (pos < length) { + char c = jsonPath.charAt(pos); + if (c == '.') { + int start = ++pos; + while (pos < length && isNameChar(jsonPath.charAt(pos))) { + pos++; + } + if (pos == start) { + // "..", ".*", a trailing "." or a name starting with an unsupported character. + return NOT_SIMPLE; + } + keys.add(jsonPath.substring(start, pos)); + indices.add(-1); + } else if (c == '[') { + pos++; + if (pos == length) { + return NOT_SIMPLE; + } + char quote = jsonPath.charAt(pos); + if (quote == '\'' || quote == '"') { + int start = ++pos; + while (pos < length && jsonPath.charAt(pos) != quote) { + if (jsonPath.charAt(pos) == '\\') { + return NOT_SIMPLE; + } + pos++; + } + if (pos == length) { + return NOT_SIMPLE; + } + String key = jsonPath.substring(start, pos); + // Skip the closing quote; anything other than ']' next means a union such as ['a','b']. + if (++pos == length || jsonPath.charAt(pos++) != ']') { + return NOT_SIMPLE; + } + keys.add(key); + indices.add(-1); + } else { + int start = pos; + while (pos < length && isDigit(jsonPath.charAt(pos))) { + pos++; + } + // Rejects [*], [?(...)], [-1], [1,2] and [1:2] because none of them close right after the digits. + if (pos == start || pos == length || jsonPath.charAt(pos++) != ']') { + return NOT_SIMPLE; + } + long index; + try { + index = Long.parseLong(jsonPath, start, pos - 1, 10); + } catch (NumberFormatException e) { + return NOT_SIMPLE; + } + if (index > Integer.MAX_VALUE) { + return NOT_SIMPLE; + } + keys.add(null); + indices.add((int) index); + } + } else { + return NOT_SIMPLE; + } + } + if (keys.isEmpty()) { + return NOT_SIMPLE; + } + int size = keys.size(); + int[] indexArray = new int[size]; + for (int i = 0; i < size; i++) { + indexArray[i] = indices.get(i); + } + return new SimpleJsonPath(keys.toArray(new String[0]), indexArray); + } + + private static boolean isNameChar(char c) { + return Character.isLetterOrDigit(c) || c == '_' || c == '-'; + } + + private static boolean isDigit(char c) { + return c >= '0' && c <= '9'; + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("$"); + for (int i = 0; i < _keys.length; i++) { + if (_keys[i] != null) { + sb.append("['").append(_keys[i]).append("']"); + } else { + sb.append('[').append(_indices[i]).append(']'); + } + } + return sb.toString(); + } +} diff --git a/pinot-common/src/main/java/org/apache/pinot/common/function/StreamingJsonPathExtractor.java b/pinot-common/src/main/java/org/apache/pinot/common/function/StreamingJsonPathExtractor.java new file mode 100644 index 000000000000..2a0a8bd821c4 --- /dev/null +++ b/pinot-common/src/main/java/org/apache/pinot/common/function/StreamingJsonPathExtractor.java @@ -0,0 +1,326 @@ +/** + * 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.JsonFactory; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonToken; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.common.base.Preconditions; +import com.jayway.jsonpath.InvalidJsonException; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import javax.annotation.Nullable; + + +/// Resolves one or more [SimpleJsonPath]s against a JSON document in a single forward pass of a Jackson +/// [JsonParser], materializing only the addressed values. Subtrees that no path descends into are consumed +/// with `skipChildren()`, so no intermediate `Map` / `List` is ever built for them. This replaces +/// `JsonPath.parse(json).read(path)`, which builds a full Jackson DOM of the document and only then walks to +/// the field. +/// +/// **Semantics.** The results are byte-for-byte identical to Jayway configured as Pinot configures it +/// (`JacksonJsonProvider` + `JacksonMappingProvider` + `Option.SUPPRESS_EXCEPTIONS`), including: +/// - a missing key, an out-of-range index, indexing a non-array, descending into a scalar, and an explicit +/// JSON `null` all resolve to `null`; +/// - duplicate object keys resolve to the **last** occurrence, and a later occurrence discards whatever an +/// earlier one resolved to, at every depth; +/// - a container leaf is returned as a `LinkedHashMap` / `ArrayList` with the same element types; +/// - numbers become `Integer` / `Long` / `BigInteger` by magnitude, and `Double` (or `BigDecimal` when +/// `useBigDecimal` is set) for floating-point literals; +/// - a document that is malformed anywhere inside its root value raises [InvalidJsonException], even when the +/// addressed field was already seen; content *after* the root value is ignored, exactly as +/// `ObjectMapper.readValue` ignores it (`FAIL_ON_TRAILING_TOKENS` is off by default). +/// +/// **The one place it deliberately differs.** Jayway builds a DOM of the *whole* document, so it materializes +/// and validates every value in it - including fields the path never addresses. This extractor skips those +/// subtrees, so a value that Jackson can lex but cannot *materialize* raises in Jayway and not here. Two such +/// values exist, and both only matter when they sit **outside** the addressed path: +/// 1. a string longer than `StreamReadConstraints` `maxStringLength` (20,000,000 chars); +/// 2. when `useBigDecimal` is set, a float literal whose exponent overflows an `int` (e.g. `1e999999999999`), +/// which `BigDecimal` cannot represent. +/// Jayway rejects such a document outright; this extractor returns the requested value. When the offending +/// value *is* the addressed one, both raise, so the results still agree. +/// +/// This is inherent to the optimization - not materializing what was not asked for is precisely what makes it +/// fast, and matching Jayway here would mean materializing every string in every document. The extractor never +/// returns a *different value* for the addressed path; it only declines to fail on values the caller did not +/// ask about, and it is the safer of the two (it never allocates the 20 MB string or the pathological +/// `BigDecimal`). +/// +/// **Early exit.** When the caller passes `earlyExit`, the pass stops as soon as every requested path has +/// resolved to a non-null value. That is much faster when the target fields appear early in the document, at +/// the cost of two behavior changes, since the tail is never read: +/// 1. duplicate keys resolve to the **first** occurrence rather than the last; +/// 2. a document malformed strictly after the last addressed field yields the extracted value instead of +/// raising [InvalidJsonException]. +/// Both inputs are undefined (RFC 8259) or corrupt, and the switch is off by default. +/// +/// **Applicability.** Callers must first check [#canExtract]: the document's first non-whitespace character +/// has to be `{` or `[`. Bare scalar documents, `null`, empty / whitespace-only input, a leading byte-order +/// mark and plain text all take the Jayway path, which reproduces its exact exceptions for them. +/// +/// Stateless and thread-safe; the shared [JsonFactory] and [ObjectMapper]s are safe for concurrent use. +public final class StreamingJsonPathExtractor { + /// The live-path set is tracked as a bitmask, so a single call cannot resolve more paths than a `long` has + /// bits. Callers with more paths must batch them. + public static final int MAX_PATHS = Long.SIZE; + + /// Deliberately the plain, unconfigured factory and mappers that Jayway's `JacksonJsonProvider` builds, so + /// that stream-read constraints (nesting depth, string length) and number handling match exactly. + private static final JsonFactory FACTORY = new JsonFactory(); + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final ObjectMapper MAPPER_WITH_BIG_DECIMAL = + new ObjectMapper().configure(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS, true); + + private StreamingJsonPathExtractor() { + } + + /// Returns `true` when `json`'s first non-whitespace character starts a JSON object or array, which is the + /// only shape this extractor handles. Everything else must go through Jayway. + public static boolean canExtract(@Nullable String json) { + if (json == null) { + return false; + } + for (int i = 0, n = json.length(); i < n; i++) { + char c = json.charAt(i); + if (c == ' ' || c == '\t' || c == '\n' || c == '\r') { + continue; + } + return c == '{' || c == '['; + } + return false; + } + + /// UTF-8 overload of [#canExtract(String)]. A byte-order mark is rejected, matching the `String` overload. + public static boolean canExtract(@Nullable byte[] json) { + if (json == null) { + return false; + } + for (byte b : json) { + if (b == ' ' || b == '\t' || b == '\n' || b == '\r') { + continue; + } + return b == '{' || b == '['; + } + return false; + } + + /// Resolves a single `path` against `json`, returning the addressed value or `null` when it resolves to a + /// missing key, an out-of-range index, a type mismatch, or an explicit JSON `null` (matching Jayway under + /// `SUPPRESS_EXCEPTIONS`). Raises [InvalidJsonException] if the document is malformed inside its root value. + /// The two short-lived arrays cost a few ns against a ~900ns parse, so the single-path case reuses the + /// multi-path walk rather than duplicating it; split out a dedicated single-path walk only if a profile says so. + @Nullable + public static Object extract(String json, SimpleJsonPath path, boolean useBigDecimal, boolean earlyExit) { + Object[] result = new Object[1]; + try (JsonParser parser = FACTORY.createParser(json)) { + extract(parser, new SimpleJsonPath[]{path}, result, useBigDecimal, earlyExit); + } catch (IOException e) { + throw new InvalidJsonException(e); + } + return result[0]; + } + + /// UTF-8 overload of [#extract(String, SimpleJsonPath, boolean, boolean)] with the same return and exception + /// contract. + @Nullable + public static Object extract(byte[] json, SimpleJsonPath path, boolean useBigDecimal, boolean earlyExit) { + Object[] result = new Object[1]; + try (JsonParser parser = FACTORY.createParser(json)) { + extract(parser, new SimpleJsonPath[]{path}, result, useBigDecimal, earlyExit); + } catch (IOException e) { + throw new InvalidJsonException(e); + } + return result[0]; + } + + /// Resolves every path in `paths` in one pass, writing path `i`'s value into `out[i]`. `out` must be at + /// least as long as `paths` and is fully overwritten for the first `paths.length` entries. + public static void extract(String json, SimpleJsonPath[] paths, Object[] out, boolean useBigDecimal, + boolean earlyExit) { + try (JsonParser parser = FACTORY.createParser(json)) { + extract(parser, paths, out, useBigDecimal, earlyExit); + } catch (IOException e) { + throw new InvalidJsonException(e); + } + } + + private static void extract(JsonParser parser, SimpleJsonPath[] paths, Object[] out, boolean useBigDecimal, + boolean earlyExit) + throws IOException { + int numPaths = paths.length; + Preconditions.checkArgument(numPaths > 0 && numPaths <= MAX_PATHS, "Expected 1 to %s paths, got: %s", MAX_PATHS, + numPaths); + Preconditions.checkArgument(out.length >= numPaths, "Expected an output array of at least %s entries, got: %s", + numPaths, out.length); + for (int i = 0; i < numPaths; i++) { + out[i] = null; + } + JsonToken token = parser.nextToken(); + if (token != JsonToken.START_OBJECT && token != JsonToken.START_ARRAY) { + // Guarded by canExtract(); a bare scalar addresses nothing under a non-empty linear path. + return; + } + long live = numPaths == MAX_PATHS ? -1L : (1L << numPaths) - 1; + walk(parser, paths, live, 0, out, useBigDecimal, earlyExit); + } + + /// Walks the container the parser is currently positioned on, resolving the paths in `live` whose segment at + /// `depth` matches. Consumes the container's closing token, unless it returns `true`, which means every path + /// has resolved and the caller should stop reading immediately (early-exit mode only). + private static boolean walk(JsonParser parser, SimpleJsonPath[] paths, long live, int depth, Object[] out, + boolean useBigDecimal, boolean earlyExit) + throws IOException { + boolean isObject = parser.currentToken() == JsonToken.START_OBJECT; + int index = 0; + while (true) { + JsonToken token = parser.nextToken(); + if (token == JsonToken.END_OBJECT || token == JsonToken.END_ARRAY) { + return false; + } + long leaves = 0; + long children = 0; + if (isObject) { + String name = parser.currentName(); + parser.nextToken(); + for (long mask = live; mask != 0; mask &= mask - 1) { + int i = Long.numberOfTrailingZeros(mask); + SimpleJsonPath path = paths[i]; + if (!path.matchesKey(depth, name)) { + continue; + } + // Jackson keeps the last of a set of duplicate keys, so a later occurrence must discard whatever an + // earlier one resolved to - including values resolved deeper down inside it. + out[i] = null; + if (depth == path.length() - 1) { + leaves |= 1L << i; + } else { + children |= 1L << i; + } + } + } else { + int elementIndex = index++; + for (long mask = live; mask != 0; mask &= mask - 1) { + int i = Long.numberOfTrailingZeros(mask); + SimpleJsonPath path = paths[i]; + if (!path.matchesIndex(depth, elementIndex)) { + continue; + } + if (depth == path.length() - 1) { + leaves |= 1L << i; + } else { + children |= 1L << i; + } + } + } + + JsonToken value = parser.currentToken(); + boolean container = value == JsonToken.START_OBJECT || value == JsonToken.START_ARRAY; + if (leaves != 0) { + // At least one path ends here. Materialize the value once; any path that wants to descend further + // (e.g. both `$.a` and `$.a.b` were requested) navigates the materialized value instead of the stream. + Object materialized = readValue(parser, value, useBigDecimal); + for (long mask = leaves; mask != 0; mask &= mask - 1) { + out[Long.numberOfTrailingZeros(mask)] = materialized; + } + for (long mask = children; mask != 0; mask &= mask - 1) { + int i = Long.numberOfTrailingZeros(mask); + out[i] = navigate(materialized, paths[i], depth + 1); + } + if (earlyExit && allResolved(out, paths.length)) { + return true; + } + } else if (children != 0 && container) { + if (walk(parser, paths, children, depth + 1, out, useBigDecimal, earlyExit)) { + return true; + } + } else { + // Either nothing matches here, or a path wants to descend into a scalar - which resolves to null. + parser.skipChildren(); + } + } + } + + /// Resolves the remaining segments of `path`, from `depth` onwards, against an already-materialized value. + /// Mirrors Jayway's behavior on the same DOM: any type mismatch, missing key or out-of-range index is `null`. + @Nullable + private static Object navigate(@Nullable Object node, SimpleJsonPath path, int depth) { + for (int d = depth, n = path.length(); d < n; d++) { + String key = path.getKey(d); + if (key != null) { + if (!(node instanceof Map)) { + return null; + } + node = ((Map) node).get(key); + } else { + if (!(node instanceof List)) { + return null; + } + List list = (List) node; + int index = path.getIndex(d); + if (index >= list.size()) { + return null; + } + node = list.get(index); + } + } + return node; + } + + /// Reads the value the parser is positioned on, leaving the parser on that value's last token. Containers are + /// materialized through the same `ObjectMapper` configuration Jayway uses, so their contents match exactly. + @Nullable + private static Object readValue(JsonParser parser, JsonToken token, boolean useBigDecimal) + throws IOException { + switch (token) { + case VALUE_STRING: + return parser.getText(); + case VALUE_NUMBER_INT: + return parser.getNumberValue(); + case VALUE_NUMBER_FLOAT: + return useBigDecimal ? parser.getDecimalValue() : parser.getNumberValue(); + case VALUE_TRUE: + return Boolean.TRUE; + case VALUE_FALSE: + return Boolean.FALSE; + case VALUE_NULL: + return null; + case START_OBJECT: + case START_ARRAY: + return (useBigDecimal ? MAPPER_WITH_BIG_DECIMAL : MAPPER).readValue(parser, Object.class); + default: + throw new IllegalStateException("Unexpected JSON token: " + token); + } + } + + /// A path whose value is an explicit JSON `null` never counts as resolved, so early exit degrades to a full + /// pass for those documents. That is the conservative direction: the extra scan can only add exceptions and + /// later duplicates that a full pass would have found anyway. + private static boolean allResolved(Object[] out, int numPaths) { + for (int i = 0; i < numPaths; i++) { + if (out[i] == null) { + return false; + } + } + return true; + } +} 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 eac8e4e4af4a..b96f321d1c24 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 @@ -38,6 +38,8 @@ import javax.annotation.Nullable; import org.apache.commons.lang3.StringUtils; import org.apache.pinot.common.function.JsonPathCache; +import org.apache.pinot.common.function.SimpleJsonPath; +import org.apache.pinot.common.function.StreamingJsonPathExtractor; import org.apache.pinot.spi.annotations.ScalarFunction; import org.apache.pinot.spi.utils.JsonUtils; @@ -105,6 +107,33 @@ public static Object jsonPath(Object object, String jsonPath) { return PARSE_CONTEXT.parse(object).read(jsonPath, NO_PREDICATES); } + /** + * Resolves {@code jsonPath} against {@code object} using the streaming extractor + * ({@link StreamingJsonPathExtractor}) when {@code object} is a JSON {@code String} and the path is a simple + * linear chain, and falls back to the Jayway {@link #jsonPath} implementation otherwise, so a caller sees the + * same value and the same exceptions either way. Backs the opt-in {@code earlyExit} extractor overloads below. + * See {@link StreamingJsonPathExtractor} for the one edge case where the streaming and Jayway results differ. + */ + @Nullable + private static Object jsonPathStreaming(Object object, String jsonPath, boolean useBigDecimal, boolean earlyExit) { + if (object instanceof String) { + String json = (String) object; + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(jsonPath); + if (simpleJsonPath != null && StreamingJsonPathExtractor.canExtract(json)) { + try { + return StreamingJsonPathExtractor.extract(json, simpleJsonPath, useBigDecimal, earlyExit); + } catch (Exception e) { + // Any failure in the streaming path degrades to the reference Jayway implementation, so an unforeseen + // streaming bug can never change a caller's result - at worst it costs a second parse on that row. + // Malformed JSON also lands here: Jayway then raises the same exception the streaming path would have. + return jsonPath(json, jsonPath); + } + } + } + // Complex path, non-JSON input, or an already-parsed container: the existing Jayway implementation. + return jsonPath(object, jsonPath); + } + /** * Returns {@code false} when the input is known to be non-extractable by a json path without invoking the JSON * parser: a {@code null} value, or a string whose first non-whitespace character cannot begin a JSON value. Used @@ -221,6 +250,45 @@ public static String jsonPathString(@Nullable Object object, String jsonPath, St } } + /** + * Opt-in streaming variant of {@link #jsonPathString(Object, String, String)} with identical results: a simple + * linear path over a JSON string is resolved in a single forward pass instead of building a full Jayway DOM + * (see {@link StreamingJsonPathExtractor}), and complex paths / non-JSON input fall back to Jayway. Faster; + * choose this when the value matters and the result must match {@code jsonPathString}. + */ + @ScalarFunction(nullableParameters = true) + public static String jsonPathStringFast(@Nullable Object object, String jsonPath, String defaultValue) { + return jsonPathStringStreaming(object, jsonPath, defaultValue, false); + } + + /** + * Like {@link #jsonPathStringFast} but stops the pass at the addressed field. Faster still when the field + * appears early, at the cost of two behavior changes on undefined/corrupt input: duplicate object keys resolve + * to the first occurrence (Jayway and {@code jsonPathString} take the last), and a document that is + * malformed strictly after the addressed field is not rejected. Choose this only when the source JSON is + * known to be well-formed and free of duplicate keys. + */ + @ScalarFunction(nullableParameters = true) + public static String jsonPathStringFirstMatch(@Nullable Object object, String jsonPath, String defaultValue) { + return jsonPathStringStreaming(object, jsonPath, defaultValue, true); + } + + private static String jsonPathStringStreaming(@Nullable Object object, String jsonPath, String defaultValue, + boolean earlyExit) { + if (!canExtractJsonPath(object)) { + return defaultValue; + } + try { + Object jsonValue = jsonPathStreaming(object, jsonPath, false, earlyExit); + if (jsonValue instanceof String) { + return (String) jsonValue; + } + return jsonValue == null ? defaultValue : JsonUtils.objectToString(jsonValue); + } catch (Exception ignore) { + return defaultValue; + } + } + /** * Extract from Json with path to Long */ @@ -251,6 +319,43 @@ public static long jsonPathLong(@Nullable Object object, String jsonPath, long d } } + /** + * Opt-in streaming variant of {@link #jsonPathLong(Object, String, long)} with identical results; see + * {@link #jsonPathStringFast} for the streaming/Jayway-fallback contract. + */ + @ScalarFunction(nullableParameters = true) + public static long jsonPathLongFast(@Nullable Object object, String jsonPath, long defaultValue) { + return jsonPathLongStreaming(object, jsonPath, defaultValue, false); + } + + /** + * Early-exit streaming variant of {@link #jsonPathLong(Object, String, long)}; see + * {@link #jsonPathStringFirstMatch} for the two behavior changes and when to use it. + */ + @ScalarFunction(nullableParameters = true) + public static long jsonPathLongFirstMatch(@Nullable Object object, String jsonPath, long defaultValue) { + return jsonPathLongStreaming(object, jsonPath, defaultValue, true); + } + + private static long jsonPathLongStreaming(@Nullable Object object, String jsonPath, long defaultValue, + boolean earlyExit) { + if (!canExtractJsonPath(object)) { + return defaultValue; + } + try { + Object jsonValue = jsonPathStreaming(object, jsonPath, false, earlyExit); + if (jsonValue == null) { + return defaultValue; + } + if (jsonValue instanceof Number) { + return ((Number) jsonValue).longValue(); + } + return Long.parseLong(jsonValue.toString()); + } catch (Exception ignore) { + return defaultValue; + } + } + /** * Extract from Json with path to Double */ @@ -281,6 +386,43 @@ public static double jsonPathDouble(@Nullable Object object, String jsonPath, do } } + /** + * Opt-in streaming variant of {@link #jsonPathDouble(Object, String, double)} with identical results; see + * {@link #jsonPathStringFast} for the streaming/Jayway-fallback contract. + */ + @ScalarFunction(nullableParameters = true) + public static double jsonPathDoubleFast(@Nullable Object object, String jsonPath, double defaultValue) { + return jsonPathDoubleStreaming(object, jsonPath, defaultValue, false); + } + + /** + * Early-exit streaming variant of {@link #jsonPathDouble(Object, String, double)}; see + * {@link #jsonPathStringFirstMatch} for the two behavior changes and when to use it. + */ + @ScalarFunction(nullableParameters = true) + public static double jsonPathDoubleFirstMatch(@Nullable Object object, String jsonPath, double defaultValue) { + return jsonPathDoubleStreaming(object, jsonPath, defaultValue, true); + } + + private static double jsonPathDoubleStreaming(@Nullable Object object, String jsonPath, double defaultValue, + boolean earlyExit) { + if (!canExtractJsonPath(object)) { + return defaultValue; + } + try { + Object jsonValue = jsonPathStreaming(object, jsonPath, false, earlyExit); + if (jsonValue == null) { + return defaultValue; + } + if (jsonValue instanceof Number) { + return ((Number) jsonValue).doubleValue(); + } + return Double.parseDouble(jsonValue.toString()); + } 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"}] diff --git a/pinot-common/src/test/java/org/apache/pinot/common/function/StreamingJsonPathExtractorTest.java b/pinot-common/src/test/java/org/apache/pinot/common/function/StreamingJsonPathExtractorTest.java new file mode 100644 index 000000000000..6f789adaf081 --- /dev/null +++ b/pinot-common/src/test/java/org/apache/pinot/common/function/StreamingJsonPathExtractorTest.java @@ -0,0 +1,756 @@ +/** + * 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.JsonFactory; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.StreamReadFeature; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.jayway.jsonpath.Configuration; +import com.jayway.jsonpath.JsonPath; +import com.jayway.jsonpath.Option; +import com.jayway.jsonpath.ParseContext; +import com.jayway.jsonpath.Predicate; +import com.jayway.jsonpath.spi.json.JacksonJsonProvider; +import com.jayway.jsonpath.spi.mapper.JacksonMappingProvider; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.Random; +import java.util.StringJoiner; +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.assertTrue; +import static org.testng.Assert.fail; + + +/** + * Differential test for {@link StreamingJsonPathExtractor} and the opt-in streaming scalar overloads, with + * Jayway as the oracle. + *

+ * Every case runs the same {json, path} pair through the streaming extractor and through the exact Jayway + * configuration it replaces, and asserts the two produce the same value - or the same exception type. The + * oracle is the real production Jayway config ({@link #PLAIN_CONTEXT} / {@link #BIG_DECIMAL_CONTEXT}) and the + * existing Jayway scalar overloads, so it cannot drift away from what Pinot actually does today. + */ +public class StreamingJsonPathExtractorTest { + private static final Predicate[] NO_PREDICATES = new Predicate[0]; + + /** Rejects the two inputs early exit is allowed to disagree on: duplicate keys and a malformed / trailing tail. */ + private static final JsonFactory STRICT_FACTORY = + JsonFactory.builder().enable(StreamReadFeature.STRICT_DUPLICATE_DETECTION).build(); + + /** Mirrors {@code JsonExtractScalarTransformFunction.JSON_PARSER_CONTEXT}. */ + private static final ParseContext PLAIN_CONTEXT = JsonPath.using( + new Configuration.ConfigurationBuilder().jsonProvider(new JacksonJsonProvider()) + .mappingProvider(new JacksonMappingProvider()).options(Option.SUPPRESS_EXCEPTIONS).build()); + + /** + * Mirrors {@code JsonExtractScalarTransformFunction.JSON_PARSER_CONTEXT_WITH_BIG_DECIMAL}, the one production + * context that is not reachable through {@link JsonFunctions}. + */ + private static final ParseContext BIG_DECIMAL_CONTEXT = JsonPath.using( + new Configuration.ConfigurationBuilder().jsonProvider(new JacksonJsonProvider( + new ObjectMapper().configure(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS, true))) + .mappingProvider(new JacksonMappingProvider()).options(Option.SUPPRESS_EXCEPTIONS).build()); + + static { + // JsonFunctions registers Pinot's JsonPath cache in its static initializer, and Jayway rejects that + // registration once its own cache has been read. This test reads through Jayway directly, so load + // JsonFunctions first or its class initialization fails. + JsonFunctions.jsonPathExists("{}", "$.x"); + } + + /** The value a call produced, or the type of the exception it threw. */ + private static final class Outcome { + final Object _value; + final Class _thrown; + + private Outcome(Object value, Class thrown) { + _value = value; + _thrown = thrown; + } + + static Outcome of(ThrowingSupplier supplier) { + try { + return new Outcome(supplier.get(), null); + } catch (Exception e) { + return new Outcome(null, e.getClass()); + } + } + + @Override + public boolean equals(Object o) { + if (!(o instanceof Outcome)) { + return false; + } + Outcome other = (Outcome) o; + return _thrown == other._thrown && Objects.deepEquals(_value, other._value); + } + + @Override + public int hashCode() { + return Objects.hash(_thrown, _value); + } + + @Override + public String toString() { + if (_thrown != null) { + return "threw " + _thrown.getSimpleName(); + } + return _value == null ? "null" : _value.getClass().getSimpleName() + "(" + _value + ")"; + } + } + + @FunctionalInterface + private interface ThrowingSupplier { + Object get() + throws Exception; + } + + /** + * Each typed extractor that has an opt-in streaming overload, paired with its existing Jayway overload. The + * two must return the same thing for every input; that is what makes the streaming overload a safe drop-in. + */ + private enum Fn { + STRING((o, p) -> JsonFunctions.jsonPathString(o, p, "DEFAULT"), + (o, p) -> JsonFunctions.jsonPathStringFast(o, p, "DEFAULT")), + LONG((o, p) -> JsonFunctions.jsonPathLong(o, p, -7L), + (o, p) -> JsonFunctions.jsonPathLongFast(o, p, -7L)), + DOUBLE((o, p) -> JsonFunctions.jsonPathDouble(o, p, -7.5d), + (o, p) -> JsonFunctions.jsonPathDoubleFast(o, p, -7.5d)); + + private final Invocation _jayway; + private final Invocation _streaming; + + Fn(Invocation jayway, Invocation streaming) { + _jayway = jayway; + _streaming = streaming; + } + + Outcome jayway(String json, String path) { + return Outcome.of(() -> _jayway.invoke(json, path)); + } + + Outcome streaming(String json, String path) { + return Outcome.of(() -> _streaming.invoke(json, path)); + } + } + + @FunctionalInterface + private interface Invocation { + Object invoke(Object json, String path) + throws Exception; + } + + /** + * Asserts full parity with Jayway at two levels: the extractor itself against the exact Jayway config it + * replaces (value level, covering raw values and containers), and every opt-in streaming scalar overload + * against its existing Jayway overload (function level, covering the coercion and fallback wiring). + */ + private static void assertParity(String json, String path) { + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(path); + if (simpleJsonPath != null && StreamingJsonPathExtractor.canExtract(json)) { + Outcome jayway = Outcome.of(() -> PLAIN_CONTEXT.parse(json).read(path, NO_PREDICATES)); + Outcome streaming = Outcome.of(() -> StreamingJsonPathExtractor.extract(json, simpleJsonPath, false, false)); + assertEquals(streaming, jayway, + String.format("extractor mismatch for json=<%s> path=<%s>: streaming=%s jayway=%s", json, path, streaming, + jayway)); + } + for (Fn fn : Fn.values()) { + Outcome jayway = fn.jayway(json, path); + Outcome streaming = fn.streaming(json, path); + assertEquals(streaming, jayway, + String.format("%s mismatch for json=<%s> path=<%s>: streaming=%s jayway=%s", fn, json, path, streaming, + jayway)); + } + } + + @DataProvider(name = "parityCases") + public Object[][] parityCases() { + return new Object[][]{ + // Scalar leaves, all JSON types. + {"{\"a\":\"s\"}", "$.a"}, + {"{\"a\":1}", "$.a"}, + {"{\"a\":true}", "$.a"}, + {"{\"a\":false}", "$.a"}, + {"{\"a\":null}", "$.a"}, + + // Missing paths, at every depth and shape. + {"{\"a\":1}", "$.b"}, + {"{\"a\":1}", "$.b.c"}, + {"{\"a\":1}", "$.a.b.c.d"}, + {"{\"a\":{\"b\":{}}}", "$.a.b.c"}, + {"{\" a\":1}", "$.a"}, + + // Type mismatches: descend into a scalar / array, index a non-array, out of range. + {"{\"a\":5}", "$.a.b"}, + {"{\"a\":\"xy\"}", "$.a[0]"}, + {"{\"a\":[1,2]}", "$.a.b"}, + {"{\"a\":{\"x\":1}}", "$.a[0]"}, + {"{\"a\":[10,20]}", "$.a[5]"}, + {"{\"a\":null}", "$.a.b"}, + {"{\"a\":null}", "$.a[0]"}, + {"[1,2,3]", "$.a"}, + {"{\"a\":1}", "$[0]"}, + + // Roots. + {"[10,20,30]", "$[0]"}, + {"[10,20,30]", "$[5]"}, + {"{\"a\":1}", "$['a']"}, + {"{\"a\":1}", "$[\"a\"]"}, + + // Nested arrays and arrays of objects. + {"{\"a\":[{\"b\":7}]}", "$.a[0].b"}, + {"{\"a\":[{\"b\":7}]}", "$.a[1].b"}, + {"{\"a\":[[1,2]]}", "$.a[0][1]"}, + {"{\"a\":[[1,2]]}", "$.a[0][9]"}, + {"{\"a\":[10,20]}", "$.a[01]"}, + {"{\"a\":[10,20]}", "$.a[ 1 ]"}, + {"[[{\"x\":[5]}]]", "$[0][0].x[0]"}, + + // Container leaves. + {"{\"a\":{\"x\":1,\"y\":[1,2]}}", "$.a"}, + {"{\"a\":[1,\"two\",null,{\"b\":3}]}", "$.a"}, + {"{\"a\":[]}", "$.a"}, + {"{\"a\":{}}", "$.a"}, + {"{\"a\":{\"b\":{\"c\":{\"d\":[1,{\"e\":2}]}}}}", "$.a.b.c"}, + + // Deep nesting. + {"{\"a\":{\"b\":{\"c\":{\"d\":{\"e\":{\"f\":42}}}}}}", "$.a.b.c.d.e.f"}, + + // Dotted / quoted / empty keys. + {"{\"a.b\":1}", "$['a.b']"}, + {"{\"a\":{\"b\":1}}", "$['a.b']"}, + {"{\"a\":{\"b\":1}}", "$.a['b']"}, + {"{\"a\":{\"b\":1}}", "$['a'].b"}, + {"{\"\":1}", "$['']"}, + {"{\"a b\":1}", "$['a b']"}, + {"{\"a\\\"b\":1}", "$['a\"b']"}, + {"{\"a'b\":1}", "$[\"a'b\"]"}, + {"{\"a-b\":1}", "$.a-b"}, + {"{\"a_b\":1}", "$.a_b"}, + {"{\"0\":1}", "$['0']"}, + + // Duplicate keys: last occurrence wins, at every depth, and a later one discards the earlier subtree. + {"{\"a\":1,\"a\":2}", "$.a"}, + {"{\"a\":{\"x\":1},\"a\":{\"x\":2}}", "$.a.x"}, + {"{\"a\":{\"x\":1},\"a\":{\"y\":2}}", "$.a.x"}, + {"{\"a\":{\"x\":1},\"a\":5}", "$.a.x"}, + {"{\"a\":5,\"a\":{\"x\":1}}", "$.a.x"}, + {"{\"a\":{\"x\":1,\"x\":2}}", "$.a.x"}, + {"{\"a\":[1],\"a\":[2]}", "$.a[0]"}, + {"{\"a\":1,\"a\":null}", "$.a"}, + {"{\"a\":null,\"a\":1}", "$.a"}, + + // Numbers: every boundary the untyped Jackson deserializer switches on. + {"{\"a\":2147483647}", "$.a"}, + {"{\"a\":2147483648}", "$.a"}, + {"{\"a\":-2147483648}", "$.a"}, + {"{\"a\":-2147483649}", "$.a"}, + {"{\"a\":9223372036854775807}", "$.a"}, + {"{\"a\":9223372036854775808}", "$.a"}, + {"{\"a\":-9223372036854775809}", "$.a"}, + {"{\"a\":1.5}", "$.a"}, + {"{\"a\":1.0}", "$.a"}, + {"{\"a\":-0.0}", "$.a"}, + {"{\"a\":0.1}", "$.a"}, + {"{\"a\":1E2}", "$.a"}, + {"{\"a\":1e400}", "$.a"}, + {"{\"a\":-1e400}", "$.a"}, + {"{\"a\":1e-400}", "$.a"}, + {"{\"a\":123456789012345678901234567890.123}", "$.a"}, + {"{\"a\":[1,2147483648,1.5]}", "$.a"}, + + // Unicode escapes, surrogate pairs, control characters, escaped keys. + {"{\"a\":\"caf\\u00e9\"}", "$.a"}, + {"{\"a\":\"\\ud83d\\ude00\"}", "$.a"}, + {"{\"a\":\"\\u0000\"}", "$.a"}, + {"{\"a\":\"line\\nbreak\\ttab\\\\slash\\\"quote\"}", "$.a"}, + {"{\"\\u0061\":1}", "$.a"}, + {"{\"a\":\"\\/\"}", "$.a"}, + + // Invalid JSON: before, at and after the addressed field; and structurally broken. + {"{\"a\":1,\"b\":}", "$.a"}, + {"{\"b\":,\"a\":1}", "$.a"}, + {"{\"a\":1,\"b\":[1,", "$.a"}, + {"{\"a\":1,", "$.a"}, + {"{\"a\":1,}", "$.a"}, + {"{\"a\":01}", "$.a"}, + {"{\"a\":NaN}", "$.a"}, + {"{'a':1}", "$.a"}, + {"{a:1}", "$.a"}, + {"{\"a\":1,\"b\":\"\\uZZZZ\"}", "$.a"}, + {"[1,2", "$[0]"}, + {"{", "$.a"}, + + // Content after the root value is ignored, exactly as ObjectMapper.readValue ignores it. + {"{\"a\":1} xyz", "$.a"}, + {"{\"a\":1}{\"a\":2}", "$.a"}, + {"[1,2] xyz", "$[0]"}, + + // Non-JSON, empty, whitespace, BOM, bare scalars: all must take the Jayway path. + {"hello world", "$.a"}, + {"", "$.a"}, + {" ", "$.a"}, + {"{\"a\":1}", "$.a"}, + {"5", "$.a"}, + {"\"str\"", "$.a"}, + {"true", "$.a"}, + {"null", "$.a"}, + {"nullx", "$.a"}, + {" {\"a\":1} ", "$.a"}, + + // Paths the fast path must decline, so these assert the fallback is wired correctly. + {"{\"a\":{\"b\":1}}", "$..b"}, + {"{\"a\":{\"b\":1}}", "$.*"}, + {"{\"a\":[1,2,3]}", "$.a[*]"}, + {"{\"a\":[1,2,3]}", "$.a[-1]"}, + {"{\"a\":[1,2,3]}", "$.a[0:2]"}, + {"{\"a\":[1,2,3]}", "$.a.length()"}, + {"{\"a\":[{\"b\":1},{\"b\":2}]}", "$.a[?(@.b==2)]"}, + {"{\"a\":1,\"b\":2}", "$['a','b']"}, + {"{\"a\":1}", "$"}, + {"{\"a\":1}", "a"}, + + // String coercions that only bite the typed wrappers. + {"{\"a\":\"123\"}", "$.a"}, + {"{\"a\":\"12.5\"}", "$.a"}, + {"{\"a\":\"not a number\"}", "$.a"}, + {"{\"a\":{\"b\":1}}", "$.a"}, + }; + } + + @Test(dataProvider = "parityCases") + public void testParityWithJayway(String json, String path) { + assertParity(json, path); + } + + /** + * The {@code USE_BIG_DECIMAL_FOR_FLOATS} context used by {@code jsonExtractScalar} for STRING and BIG_DECIMAL + * results is not reachable through {@link JsonFunctions}, so it gets its own oracle. + */ + @Test(dataProvider = "parityCases") + public void testBigDecimalParityWithJayway(String json, String path) { + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(path); + if (simpleJsonPath == null || !StreamingJsonPathExtractor.canExtract(json)) { + return; + } + Outcome jayway = Outcome.of(() -> BIG_DECIMAL_CONTEXT.parse(json).read(path, NO_PREDICATES)); + Outcome streaming = Outcome.of(() -> StreamingJsonPathExtractor.extract(json, simpleJsonPath, true, false)); + assertEquals(streaming, jayway, + String.format("big-decimal mismatch for json=<%s> path=<%s>: streaming=%s jayway=%s", json, path, streaming, + jayway)); + } + + /** + * The {@code byte[]} overloads back {@code jsonExtractScalar} over a BYTES column, which reaches + * {@code parseContext.parseUtf8} rather than {@code parse}. They get their own oracle. + */ + @Test(dataProvider = "parityCases") + public void testBytesParityWithJayway(String json, String path) { + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(path); + byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8); + assertEquals(StreamingJsonPathExtractor.canExtract(jsonBytes), StreamingJsonPathExtractor.canExtract(json), + "canExtract disagrees between String and byte[] for json=<" + json + ">"); + if (simpleJsonPath == null || !StreamingJsonPathExtractor.canExtract(jsonBytes)) { + return; + } + Outcome jayway = Outcome.of(() -> PLAIN_CONTEXT.parseUtf8(jsonBytes).read(path, NO_PREDICATES)); + Outcome streaming = Outcome.of(() -> StreamingJsonPathExtractor.extract(jsonBytes, simpleJsonPath, false, false)); + assertEquals(streaming, jayway, + String.format("bytes mismatch for json=<%s> path=<%s>: streaming=%s jayway=%s", json, path, streaming, jayway)); + } + + /** Multibyte UTF-8 in keys and values, which only the {@code byte[]} path can get wrong. */ + @Test + public void testBytesParityWithMultibyteContent() { + for (String[] jsonAndPath : new String[][]{ + {"{\"a\":\"café\"}", "$.a"}, {"{\"café\":1}", "$.café"}, {"{\"café\":1}", "$['café']"}, + {"{\"a\":\"日本語\"}", "$.a"}, {"{\"日本\":{\"語\":7}}", "$.日本.語"}, {"{\"a\":\"😀\"}", "$.a"}, + {"{\"a\":\"😀\",\"b\":2}", "$.b"} + }) { + testBytesParityWithJayway(jsonAndPath[0], jsonAndPath[1]); + assertParity(jsonAndPath[0], jsonAndPath[1]); + } + } + + /** + * The opt-in streaming overloads are new SQL-visible signatures, so resolve and invoke them through the + * {@link FunctionRegistry} the query/ingestion path uses. This is what a Java-level call cannot prove: that + * the four-argument overloads register under their own arity and that a {@code boolean} literal converts. + */ + @Test + public void testStreamingFunctionsResolveThroughFunctionRegistry() + throws Exception { + String json = "{\"user\":{\"country\":\"US\",\"age\":41,\"score\":9.5}}"; + 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("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 + // what the existing Jayway function produces (here the deep-scan list, stringified). + assertEquals(invoke("jsonPathStringFast", json, "$..country", "DEFAULT"), + JsonFunctions.jsonPathString(json, "$..country", "DEFAULT")); + } + + /** + * Function-level coverage of the {@code FirstMatch} (early-exit) functions: they equal {@code Fast}/Jayway on + * clean data, and diverge to the first occurrence on the documented duplicate-key case. + */ + @Test + public void testFirstMatchFunctions() { + String clean = "{\"a\":{\"b\":\"x\"},\"n\":{\"m\":7}}"; + assertEquals(JsonFunctions.jsonPathStringFirstMatch(clean, "$.a.b", "d"), + JsonFunctions.jsonPathStringFast(clean, "$.a.b", "d")); + assertEquals(JsonFunctions.jsonPathLongFirstMatch(clean, "$.n.m", -1L), + JsonFunctions.jsonPathLongFast(clean, "$.n.m", -1L)); + + String duplicateKeys = "{\"a\":1,\"a\":2}"; + assertEquals(JsonFunctions.jsonPathLong(duplicateKeys, "$.a", -1L), 2L, "Jayway takes the last occurrence"); + assertEquals(JsonFunctions.jsonPathLongFast(duplicateKeys, "$.a", -1L), 2L, "Fast matches Jayway"); + assertEquals(JsonFunctions.jsonPathLongFirstMatch(duplicateKeys, "$.a", -1L), 1L, "FirstMatch takes the first"); + } + + private static Object invoke(String name, Object... arguments) + throws Exception { + FunctionInfo functionInfo = FunctionRegistry.getFunctionInfo(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); + } + + /** Invalid UTF-8 must raise the same exception type Jayway raises. */ + @Test + public void testBytesParityWithInvalidUtf8() { + byte[] invalid = {'{', '"', 'a', '"', ':', '"', (byte) 0xC3, '"', '}'}; + SimpleJsonPath path = SimpleJsonPath.compile("$.a"); + assertNotNull(path); + assertTrue(StreamingJsonPathExtractor.canExtract(invalid)); + Outcome jayway = Outcome.of(() -> PLAIN_CONTEXT.parseUtf8(invalid).read("$.a", NO_PREDICATES)); + Outcome streaming = Outcome.of(() -> StreamingJsonPathExtractor.extract(invalid, path, false, false)); + assertEquals(streaming, jayway); + } + + /** The {@code long} live-path bitmask special-cases a full 64-path batch; anything more must be rejected. */ + @Test + public void testMaxPathsBoundary() { + String json = "{\"k0\":0,\"k63\":63,\"k64\":64}"; + SimpleJsonPath[] paths = new SimpleJsonPath[StreamingJsonPathExtractor.MAX_PATHS]; + for (int i = 0; i < paths.length; i++) { + paths[i] = SimpleJsonPath.compile("$.k" + i); + } + Object[] out = new Object[paths.length]; + StreamingJsonPathExtractor.extract(json, paths, out, false, false); + assertEquals(out[0], 0); + assertEquals(out[63], 63); + assertNull(out[1]); + + SimpleJsonPath[] tooMany = new SimpleJsonPath[StreamingJsonPathExtractor.MAX_PATHS + 1]; + Arrays.fill(tooMany, paths[0]); + try { + StreamingJsonPathExtractor.extract(json, tooMany, new Object[tooMany.length], false, false); + fail("must reject more paths than the bitmask can hold"); + } catch (IllegalArgumentException e) { + // Expected. + } + } + + /** + * Pins the one documented divergence from Jayway in full-scan mode. Jayway materializes the whole document, so + * a value it can lex but not materialize makes it reject the document even when the path never addresses that + * value; the streaming extractor skips such subtrees and returns the requested value. Matching Jayway here + * would mean materializing every string in every document, which is exactly the cost this class exists to + * avoid. Asserted explicitly so it can never widen unnoticed. + */ + @Test + public void testUnmaterializableValueOutsideThePathDivergesFromJayway() { + SimpleJsonPath path = SimpleJsonPath.compile("$.a"); + assertNotNull(path); + + // (1) float literal whose exponent overflows an int, only unrepresentable under useBigDecimal. + String hostileFloat = "{\"other\":1e999999999999,\"a\":1}"; + Outcome jaywayBigDecimal = Outcome.of(() -> BIG_DECIMAL_CONTEXT.parse(hostileFloat).read("$.a", NO_PREDICATES)); + assertNotNull(jaywayBigDecimal._thrown, "Jayway must still reject the hostile float under useBigDecimal"); + assertEquals(StreamingJsonPathExtractor.extract(hostileFloat, path, true, false), 1, + "streaming skips the unaddressed subtree and returns the addressed value"); + // The plain (double) context represents it as Infinity, so there is no divergence there. + assertParity(hostileFloat, "$.a"); + // When the hostile literal IS the addressed value, both raise - parity holds. + String addressedHostileFloat = "{\"a\":1e999999999999}"; + try { + StreamingJsonPathExtractor.extract(addressedHostileFloat, path, true, false); + fail("addressing the hostile float must still raise under useBigDecimal"); + } catch (Exception e) { + // Expected, same as Jayway. + } + + // (2) string longer than StreamReadConstraints.maxStringLength (20,000,000), outside the path. + String hostileString = "{\"a\":\"ERROR\",\"other\":\"" + "x".repeat(20_000_001) + "\"}"; + Outcome jaywayLongString = Outcome.of(() -> PLAIN_CONTEXT.parse(hostileString).read("$.a", NO_PREDICATES)); + assertNotNull(jaywayLongString._thrown, "Jayway must still reject the over-long unaddressed string"); + assertEquals(StreamingJsonPathExtractor.extract(hostileString, path, false, false), "ERROR", + "streaming never materializes the unaddressed string, so it does not hit the limit"); + } + + /** + * The {@code useBigDecimal} context backs {@code jsonExtractScalar(..., 'STRING')} and {@code 'BIG_DECIMAL'} - + * the most common result types - but is unreachable through {@link JsonFunctions}, so the other fuzzers never + * touch it. Fuzz it directly against its own Jayway oracle. + */ + @Test + public void testBigDecimalFuzzAgainstJayway() { + Random random = new Random(4242L); + for (int i = 0; i < 20_000; i++) { + String json = randomJson(random, 0); + String path = randomPath(random); + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(path); + if (simpleJsonPath == null || !StreamingJsonPathExtractor.canExtract(json)) { + continue; + } + Outcome jayway = Outcome.of(() -> BIG_DECIMAL_CONTEXT.parse(json).read(path, NO_PREDICATES)); + Outcome streaming = Outcome.of(() -> StreamingJsonPathExtractor.extract(json, simpleJsonPath, true, false)); + assertEquals(streaming, jayway, + String.format("big-decimal fuzz mismatch for json=<%s> path=<%s>", json, path)); + } + } + + @Test + public void testCompileAcceptsSimpleLinearPaths() { + for (String path : new String[]{ + "$.a", "$.a.b", "$['a']", "$[\"a\"]", "$[0]", "$.a[0]", "$.a[0].b", "$['a.b']", "$.a['b'][2]", "$['']", + "$.a-b", "$.a_b", "$.abc123", "$[2147483647]", + // Jayway reads a leading zero as a plain decimal index, so this is index 1, not a rejected literal. + "$[01]" + }) { + assertNotNull(SimpleJsonPath.compile(path), path); + } + } + + @Test + public void testCompileRejectsEverythingElse() { + for (String path : new String[]{ + "$", "$..a", "$.*", "$..*", "$.a[*]", "$.a[-1]", "$.a[0:2]", "$.a[:2]", "$.a[1,2]", "$['a','b']", + "$.a[?(@.b==2)]", "$.a.length()", "$.a.", "$..", "a", "a.b", "", "$.", "$[", "$[']", "$['a", "$['a'", + "$['a\\'b']", "$['a']extra", "$.a b", "$[2147483648]", "$[ 1 ]", "$.@", "$.a$b" + }) { + assertNull(SimpleJsonPath.compile(path), path); + } + } + + /** The two behavior changes early exit buys, asserted explicitly so they can never regress silently. */ + @Test + public void testEarlyExitDivergences() { + String duplicateKeys = "{\"a\":1,\"a\":2}"; + String malformedTail = "{\"a\":1,\"b\":}"; + SimpleJsonPath path = SimpleJsonPath.compile("$.a"); + assertNotNull(path); + + assertEquals(StreamingJsonPathExtractor.extract(duplicateKeys, path, false, false), 2, + "full pass takes the last key"); + try { + StreamingJsonPathExtractor.extract(malformedTail, path, false, false); + fail("full pass must reject a document malformed inside its root value"); + } catch (Exception e) { + // Expected: same as Jayway. + } + + assertEquals(StreamingJsonPathExtractor.extract(duplicateKeys, path, false, true), 1, + "early exit takes the first key"); + assertEquals(StreamingJsonPathExtractor.extract(malformedTail, path, false, true), 1, + "early exit never reads the malformed tail"); + } + + /** + * A JSON document that parses cleanly with duplicate-key detection on, and has nothing after its root value, is + * exactly a document for which early exit provably cannot change the answer. Those must match a full pass. + */ + private static boolean isStrictAndSingleRooted(String json) { + try (JsonParser parser = STRICT_FACTORY.createParser(json)) { + while (parser.nextToken() != null) { + // Reading every token also rejects trailing content, which a full pass would ignore but never see. + } + return true; + } catch (Exception e) { + return false; + } + } + + private static void assertEarlyExitParity(String json, String path) { + if (!isStrictAndSingleRooted(json)) { + return; + } + SimpleJsonPath simpleJsonPath = SimpleJsonPath.compile(path); + if (simpleJsonPath == null || !StreamingJsonPathExtractor.canExtract(json)) { + return; + } + Outcome fullPass = Outcome.of(() -> StreamingJsonPathExtractor.extract(json, simpleJsonPath, false, false)); + Outcome earlyExit = Outcome.of(() -> StreamingJsonPathExtractor.extract(json, simpleJsonPath, false, true)); + assertEquals(earlyExit, fullPass, String.format("early-exit mismatch for json=<%s> path=<%s>", json, path)); + } + + /** Everything but a duplicate key or a malformed tail must be identical whether or not early exit is on. */ + @Test(dataProvider = "parityCases") + public void testEarlyExitParityExceptKnownDivergences(String json, String path) { + assertEarlyExitParity(json, path); + } + + @Test + public void testEarlyExitFuzzParityExceptKnownDivergences() { + Random random = new Random(556677L); + for (int i = 0; i < 20_000; i++) { + assertEarlyExitParity(randomJson(random, 0), randomPath(random)); + } + } + + @Test + public void testMultiPathMatchesSinglePath() { + String json = "{\"user\":{\"id\":7,\"country\":\"US\"},\"tags\":[\"a\",\"b\"],\"total\":1.5,\"nil\":null}"; + String[] paths = {"$.user.country", "$.user.id", "$.tags[1]", "$.total", "$.nil", "$.missing", "$.user", + "$.user.missing"}; + SimpleJsonPath[] compiled = new SimpleJsonPath[paths.length]; + for (int i = 0; i < paths.length; i++) { + compiled[i] = SimpleJsonPath.compile(paths[i]); + assertNotNull(compiled[i], paths[i]); + } + Object[] batch = new Object[paths.length]; + StreamingJsonPathExtractor.extract(json, compiled, batch, false, false); + for (int i = 0; i < paths.length; i++) { + assertEquals(batch[i], StreamingJsonPathExtractor.extract(json, compiled[i], false, false), paths[i]); + assertEquals(batch[i], PLAIN_CONTEXT.parse(json).read(paths[i], NO_PREDICATES), paths[i]); + } + } + + /** A path that is a strict prefix of another must not consume the value the longer path needs. */ + @Test + public void testMultiPathWithOverlappingPrefixes() { + String json = "{\"a\":{\"b\":{\"c\":1}}}"; + SimpleJsonPath[] compiled = { + SimpleJsonPath.compile("$.a"), SimpleJsonPath.compile("$.a.b"), SimpleJsonPath.compile("$.a.b.c"), + SimpleJsonPath.compile("$.a.b.missing") + }; + Object[] batch = new Object[compiled.length]; + StreamingJsonPathExtractor.extract(json, compiled, batch, false, false); + assertEquals(batch[0].toString(), "{b={c=1}}"); + assertEquals(batch[1].toString(), "{c=1}"); + assertEquals(batch[2], 1); + assertNull(batch[3]); + } + + @Test + public void testFuzzAgainstJayway() { + Random random = new Random(20260709L); + for (int i = 0; i < 20_000; i++) { + String json = randomJson(random, 0); + String path = randomPath(random); + assertParity(json, path); + } + } + + /** Same corpus, but the paths are generated to actually hit something more often than not. */ + @Test + public void testFuzzWithDuplicateKeysAndDeepNesting() { + Random random = new Random(981723L); + for (int i = 0; i < 20_000; i++) { + StringJoiner object = new StringJoiner(",", "{", "}"); + int numFields = 1 + random.nextInt(4); + for (int f = 0; f < numFields; f++) { + // A small key alphabet makes duplicate keys frequent, which is the case early exit gets wrong. + object.add("\"" + KEYS[random.nextInt(3)] + "\":" + randomJson(random, 2)); + } + assertParity(object.toString(), randomPath(random)); + } + } + + private static final String[] KEYS = {"a", "b", "c", "a-b", "a.b", ""}; + private static final String[] SCALARS = { + "1", "0", "-1", "2147483648", "9223372036854775808", "1.5", "-0.0", "1e400", "1E2", + "123456789012345678901234567890.123", "true", "false", "null", "\"\"", "\"s\"", "\"caf\\u00e9\"", + "\"\\ud83d\\ude00\"", "\"123\"" + }; + + private static String randomJson(Random random, int depth) { + int choice = depth >= 3 ? 2 : random.nextInt(5); + if (choice == 0) { + StringJoiner object = new StringJoiner(",", "{", "}"); + int numFields = random.nextInt(4); + for (int i = 0; i < numFields; i++) { + object.add("\"" + KEYS[random.nextInt(KEYS.length)] + "\":" + randomJson(random, depth + 1)); + } + return object.toString(); + } + if (choice == 1) { + StringJoiner array = new StringJoiner(",", "[", "]"); + int numElements = random.nextInt(4); + for (int i = 0; i < numElements; i++) { + array.add(randomJson(random, depth + 1)); + } + return array.toString(); + } + return SCALARS[random.nextInt(SCALARS.length)]; + } + + private static String randomPath(Random random) { + StringBuilder path = new StringBuilder("$"); + int numSegments = 1 + random.nextInt(4); + for (int i = 0; i < numSegments; i++) { + switch (random.nextInt(4)) { + case 0: + path.append('.').append(KEYS[random.nextInt(3)]); + break; + case 1: + path.append("['").append(KEYS[random.nextInt(KEYS.length)]).append("']"); + break; + case 2: + path.append('[').append(random.nextInt(4)).append(']'); + break; + default: + path.append("[\"").append(KEYS[random.nextInt(KEYS.length)]).append("\"]"); + break; + } + } + return path.toString(); + } + + /** The fuzz corpora must actually reach the fast path, otherwise the test proves nothing. */ + @Test + public void testFuzzCorpusExercisesTheFastPath() { + Random random = new Random(20260709L); + List handled = new ArrayList<>(); + for (int i = 0; i < 20_000; i++) { + String json = randomJson(random, 0); + String path = randomPath(random); + if (SimpleJsonPath.compile(path) != null && StreamingJsonPathExtractor.canExtract(json)) { + handled.add(path); + } + } + assertTrue(handled.size() > 5_000, "fuzz corpus only reached the fast path " + handled.size() + " times"); + } +} 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 94af87e0bb7b..edb5299965bc 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,6 +60,9 @@ public class JsonPathTest extends CustomDataQueryClusterIntegrationTest { private static final String MY_MAP_STR_FIELD_NAME = "myMapStr"; 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 streaming 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 COMPLEX_MAP_STR_FIELD_NAME = "complexMapStr"; private static final String COMPLEX_MAP_STR_K3_FIELD_NAME = "complexMapStr_k3"; @@ -88,6 +91,8 @@ public Schema createSchema() { .addSingleValueDimension(MY_MAP_STR_FIELD_NAME, DataType.STRING) .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(COMPLEX_MAP_STR_FIELD_NAME, DataType.STRING) .addMultiValueDimension(COMPLEX_MAP_STR_K3_FIELD_NAME, DataType.STRING) .build(); @@ -98,6 +103,10 @@ public TableConfig createOfflineTableConfig() { List transformConfigs = List.of( new TransformConfig(MY_MAP_STR_K1_FIELD_NAME, "jsonPathString(" + MY_MAP_STR_FIELD_NAME + ", '$.k1')"), new TransformConfig(MY_MAP_STR_K2_FIELD_NAME, "jsonPathString(" + MY_MAP_STR_FIELD_NAME + ", '$.k2')"), + new TransformConfig(MY_MAP_STR_K1_FAST_FIELD_NAME, + "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(COMPLEX_MAP_STR_K3_FIELD_NAME, "jsonPathArray(" + COMPLEX_MAP_STR_FIELD_NAME + ", '$.k3')") ); IngestionConfig ingestionConfig = new IngestionConfig(); @@ -381,6 +390,28 @@ void testFailedQuery(boolean useMultiStageQueryEngine) assertEquals(pinotResponse.get("totalDocs").asInt(), 0); } + /** + * End-to-end coverage for the opt-in streaming functions. They take an {@code Object} argument, so - like the + * existing {@code jsonPathString} - they are used through the ingestion transform path, not as query-time + * scalars. {@code myMapStr_k1_fast} / {@code myMapStr_k1_first} are derived at ingestion via + * {@code jsonPathStringFast} / {@code jsonPathStringFirstMatch}; this asserts, over real rows on both query + * engines, that they equal the Jayway-derived {@code myMapStr_k1}. The rows have no duplicate keys and no + * malformed content, so {@code FirstMatch} must also agree. + */ + @Test(dataProvider = "useBothQueryEngines") + void testStreamingScalarFunctions(boolean useMultiStageQueryEngine) + throws Exception { + setUseMultiStageQueryEngine(useMultiStageQueryEngine); + String query = "SELECT myMapStr_k1, myMapStr_k1_fast, myMapStr_k1_first 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"); + } + } + @Test public void testJsonPathCache() { Cache cache = CacheProvider.getCache(); 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 new file mode 100644 index 000000000000..aff2fe6d8ada --- /dev/null +++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonPathExtraction.java @@ -0,0 +1,174 @@ +/** + * 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 com.jayway.jsonpath.Configuration; +import com.jayway.jsonpath.JsonPath; +import com.jayway.jsonpath.Option; +import com.jayway.jsonpath.ParseContext; +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.pinot.common.function.SimpleJsonPath; +import org.apache.pinot.common.function.StreamingJsonPathExtractor; +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.Level; +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; +import org.openjdk.jmh.runner.Runner; +import org.openjdk.jmh.runner.options.ChainedOptionsBuilder; +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 {@link StreamingJsonPathExtractor}. + *

+ * The {@code 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. + *

+ * The single-column benchmarks are also the per-row cost of {@code jsonExtractScalar}: that transform function + * does exactly {@code parseContext.parse(row).read(jsonPath)} per row, so measuring the extraction in isolation + * measures it without the surrounding {@code ValueBlock} scaffolding. + */ +@BenchmarkMode(Mode.Throughput) +@OutputTimeUnit(TimeUnit.SECONDS) +@Fork(1) +@Warmup(iterations = 3, time = 2) +@Measurement(iterations = 5, time = 3) +@State(Scope.Benchmark) +public class BenchmarkJsonPathExtraction { + private static final Predicate[] NO_PREDICATES = new Predicate[0]; + + /** Exactly the context {@code JsonExtractScalarTransformFunction} and {@code 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 = "{" + + "\"ts\":1719878400123," + + "\"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\":149.5},{\"sku\":\"C3\",\"qty\":5,\"price\":3.25}]," + + "\"total\":352.73,\"currency\":\"USD\"}," + + "\"device\":{\"os\":\"iOS\",\"version\":\"17.4.1\",\"model\":\"iPhone15,3\",\"screen\":{\"w\":1179," + + "\"h\":2556}}," + + "\"geo\":{\"lat\":37.7749,\"lon\":-122.4194,\"city\":\"San Francisco\",\"region\":\"CA\"}," + + "\"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\"}" + + "}"; + + /** Four derived columns, spread through the document, as an ingestion {@code transformConfigs} would pull. */ + private static final String[] FOUR_PATHS = {"$.user.country", "$.event.currency", "$.device.os", "$.geo.city"}; + + @Param({"early", "late"}) + private String _fieldPosition; + + private String _path; + private SimpleJsonPath _simplePath; + private SimpleJsonPath[] _simpleFourPaths; + private Object[] _fourResults; + + public static void main(String[] args) + throws Exception { + ChainedOptionsBuilder opt = + new OptionsBuilder().include(BenchmarkJsonPathExtraction.class.getSimpleName()).shouldDoGC(true); + new Runner(opt.build()).run(); + } + + @Setup(Level.Trial) + public void setUp() { + _path = "early".equals(_fieldPosition) ? "$.user.country" : "$.trailer.country"; + _simplePath = SimpleJsonPath.compile(_path); + _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]; + } + + @Benchmark + public Object jaywayOneColumn() { + return PARSE_CONTEXT.parse(JSON).read(_path, NO_PREDICATES); + } + + @Benchmark + public Object streamingOneColumnFullScan() { + return StreamingJsonPathExtractor.extract(JSON, _simplePath, false, false); + } + + @Benchmark + public Object streamingOneColumnEarlyExit() { + return StreamingJsonPathExtractor.extract(JSON, _simplePath, false, true); + } + + /** The existing Jayway scalar function (applicability check + String coercion). */ + @Benchmark + public String jsonPathStringJayway() { + return JsonFunctions.jsonPathString(JSON, _path, ""); + } + + /** The opt-in streaming scalar function, full scan (exact parity). */ + @Benchmark + public String jsonPathStringFast() { + return JsonFunctions.jsonPathStringFast(JSON, _path, ""); + } + + /** The opt-in streaming scalar function, early exit / first match. */ + @Benchmark + public String jsonPathStringFirstMatch() { + return JsonFunctions.jsonPathStringFirstMatch(JSON, _path, ""); + } + + @Benchmark + public Object jaywayFourColumns() { + Object last = null; + for (String path : FOUR_PATHS) { + last = PARSE_CONTEXT.parse(JSON).read(path, NO_PREDICATES); + } + return last; + } + + @Benchmark + public Object streamingFourColumnsSeparatePasses() { + Object last = null; + for (SimpleJsonPath path : _simpleFourPaths) { + last = StreamingJsonPathExtractor.extract(JSON, path, false, false); + } + return last; + } + + @Benchmark + public Object[] streamingFourColumnsSinglePass() { + StreamingJsonPathExtractor.extract(JSON, _simpleFourPaths, _fourResults, false, false); + return _fourResults; + } +}