This is an automated email from the ASF dual-hosted git repository.
sunchao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 64876d2f9668 [SPARK-57990][SQL] Share array-index JSON paths across
coalesce
64876d2f9668 is described below
commit 64876d2f9668a935516c647565b95f6928cc27ab
Author: Chao Sun <[email protected]>
AuthorDate: Tue Jul 7 11:56:36 2026 -0700
[SPARK-57990][SQL] Share array-index JSON paths across coalesce
## Why are the changes needed?
[SPARK-47670](https://issues.apache.org/jira/browse/SPARK-47670) introduced
opt-in shared parsing for repeated `get_json_object` calls over the same JSON
input. [SPARK-57626](https://issues.apache.org/jira/browse/SPARK-57626)
extended the optimization from top-level fields to repeated literal nested
object paths.
Array indexes and conditional expressions remain outside the shared-parsing
path. As described in
[SPARK-57990](https://issues.apache.org/jira/browse/SPARK-57990), this leaves a
common object-or-array payload pattern parsing the same JSON repeatedly:
```sql
SELECT
coalesce(
get_json_object(json, '$.model'),
get_json_object(json, '$[0].model')) AS model,
coalesce(
get_json_object(json, '$.request_id'),
get_json_object(json, '$[0].request_id')) AS request_id
FROM events
```
For an array-shaped row, each `coalesce` evaluates both extractions. The
parsing cost therefore continues to scale with the number of projected fields.
## What changes were proposed in this PR?
This PR extends the existing internal shared parser and optimizer rewrite
to array-index paths and safe object-or-array `coalesce` fallbacks.
The changes:
- Represent eligible literal JSON paths as named-key and nonnegative
array-index segments. This supports paths such as `$.items[0].id`,
`$[0].model`, and `$.matrix[0][1]`.
- Extend the runtime path trie to traverse object and array roots, nested
arrays, and mixed object/array paths in one streaming parse.
- Share compatible `coalesce` branches when every branch is an eligible
`get_json_object`, optionally wrapped in casts, and every branch reads the same
JSON attribute.
- Select only the first object-root and first array-root branch from each
eligible `coalesce`. These root shapes are mutually exclusive, while later
same-root fallbacks remain lazy.
- Keep each selected object/array pair together as one atomic grouping unit
when prefix conflicts require multiple shared-parser groups.
- Preserve legacy behavior for duplicate keys and parents, nulls, missing
indexes, malformed or trailing JSON, non-container intermediate values, raw
strings, rendering failures, and ancestor/descendant prefix conflicts. Terminal
array-index JSON null remains the string `"null"`, while object-key JSON null
remains SQL null.
- Leave dynamic paths, wildcards, negative, invalid, or overflowing
indexes, paths deeper than 64 segments, guarded conditionals, and unsafe
`coalesce` shapes on independent `get_json_object` evaluation.
- Update the existing configuration description and add optimizer, runtime,
code-generation, malformed-input, prefix-grouping, and benchmark coverage.
This reuses the existing internal, default-disabled
`spark.sql.optimizer.getJsonObjectSharedParsing.enabled` configuration. It does
not add a new expression, configuration, or query migration.
## How was this PR tested?
The following checks passed locally on JDK 17:
```bash
build/sbt \
"catalyst/testOnly
org.apache.spark.sql.catalyst.optimizer.OptimizeJsonExprsSuite"
build/sbt \
"sql/testOnly org.apache.spark.sql.JsonFunctionsSuite"
build/sbt \
"catalyst/scalastyle" \
"catalyst/Test/scalastyle" \
"sql/scalastyle" \
"sql/Test/scalastyle"
git diff --check
```
- `OptimizeJsonExprsSuite`: 31 passed.
- `JsonFunctionsSuite`: 111 passed.
- All four Scalastyle targets completed with zero errors and zero warnings.
- `git diff --check` passed.
The added coverage includes array and mixed object/array paths, safe and
unsafe `coalesce` shapes, cast preservation, atomic prefix grouping, malformed
input, duplicate keys and parents, JSON null behavior, missing indexes,
unsupported paths, and whole-stage code generation.
A microbenchmark was added for 200,000 cached rows, 32 fields, and a 50/50
mix of object and one-element-array payloads, projecting 2, 4, 8, and 16
object/array fallback fields. Spark's GitHub Actions benchmark workflow is
generating the Linux result files before the PR is marked ready for review:
- [JDK 17
benchmark](https://github.com/sunchao/spark/actions/runs/28842645967)
- [JDK 21
benchmark](https://github.com/sunchao/spark/actions/runs/28842650758)
- [JDK 25
benchmark](https://github.com/sunchao/spark/actions/runs/28842650019)
## Does this PR introduce any user-facing change?
Yes, within the unreleased `master` branch only. When the existing internal
shared-parsing configuration is enabled, eligible repeated array-index paths
and object-or-array `coalesce` fallbacks now use one shared parse. Query
results and public APIs are unchanged. With the configuration disabled,
analyzed and optimized plans remain unchanged. Released Spark versions are
unaffected.
## Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #57069 from sunchao/dev/chao/codex/json-array-coalesce-master.
Lead-authored-by: Chao Sun <[email protected]>
Co-authored-by: sunchao <[email protected]>
Co-authored-by: Chao Sun <[email protected]>
Signed-off-by: Chao Sun <[email protected]>
---
.../expressions/json/JsonExpressionEvalUtils.scala | 102 ++++++---
.../sql/catalyst/expressions/jsonExpressions.scala | 24 ++-
.../catalyst/optimizer/OptimizeCsvJsonExprs.scala | 214 ++++++++++++++-----
.../org/apache/spark/sql/internal/SQLConf.scala | 5 +-
.../optimizer/OptimizeJsonExprsSuite.scala | 227 ++++++++++++++++++++-
.../SharedJsonParseBenchmark-jdk21-results.txt | 60 ++++--
.../SharedJsonParseBenchmark-jdk25-results.txt | 76 ++++---
.../SharedJsonParseBenchmark-results.txt | 60 ++++--
.../org/apache/spark/sql/JsonFunctionsSuite.scala | 147 ++++++++++++-
.../json/SharedJsonParseBenchmark.scala | 39 ++++
10 files changed, 804 insertions(+), 150 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
index b5d31af45585..e80b4a355b72 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala
@@ -26,7 +26,7 @@ import com.fasterxml.jackson.core.json.JsonReadFeature
import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{ExprUtils,
GenericInternalRow}
+import org.apache.spark.sql.catalyst.expressions.{ExprUtils,
GenericInternalRow, GetJsonObject}
import org.apache.spark.sql.catalyst.json.{CreateJacksonParser,
JacksonGenerator, JacksonParser, JsonInferSchema, JSONOptions}
import org.apache.spark.sql.catalyst.util.{ArrayData, FailFastMode,
FailureSafeParser, MapData, PermissiveMode}
import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -576,26 +576,31 @@ case class GetJsonObjectEvaluator(cachedPath: UTF8String)
{
}
/**
- * Evaluates multiple simple named JSON paths in one parse.
+ * Evaluates multiple simple object-key and array-index JSON paths in one
parse.
*/
case class MultiGetJsonObjectEvaluator(
fallbackPaths: Seq[UTF8String],
- namedPaths: Seq[Seq[String]]) {
+ simplePaths: Seq[Seq[GetJsonObject.SimpleJsonPathSegment]]) {
import SharedFactory._
- require(fallbackPaths.nonEmpty && namedPaths.length == fallbackPaths.length)
+ require(fallbackPaths.nonEmpty && simplePaths.length == fallbackPaths.length)
@transient
private lazy val useTopLevelFastPath: Boolean =
- namedPaths.forall(_.length == 1) && namedPaths.distinct.length ==
namedPaths.length
+ simplePaths.forall {
+ case Seq(_: GetJsonObject.NamedPathSegment) => true
+ case _ => false
+ } && simplePaths.distinct.length == simplePaths.length
@transient
private lazy val topLevelFieldToOrdinal: Map[String, Int] =
- namedPaths.zipWithIndex.map { case (path, ordinal) => path.head -> ordinal
}.toMap
+ simplePaths.zipWithIndex.map { case (path, ordinal) =>
+ path.head.asInstanceOf[GetJsonObject.NamedPathSegment].name -> ordinal
+ }.toMap
@transient
private lazy val pathTrie: MultiGetJsonObjectEvaluator.PathTrieNode =
- MultiGetJsonObjectEvaluator.buildPathTrie(namedPaths)
+ MultiGetJsonObjectEvaluator.buildPathTrie(simplePaths)
@transient
private lazy val nullRow: InternalRow =
@@ -622,23 +627,30 @@ case class MultiGetJsonObjectEvaluator(
val matched = Array.ofDim[Boolean](fallbackPaths.length)
try {
- val validObject = Utils.tryWithResource(
+ val validRoot = Utils.tryWithResource(
CreateJacksonParser.utf8String(jsonFactory, json)) { parser =>
- if (parser.nextToken() != JsonToken.START_OBJECT) {
- false
- } else if (useTopLevelFastPath) {
- extractTopLevelObject(parser, values, matched)
- } else {
- extractObject(parser, pathTrie, values, matched)
+ parser.nextToken() match {
+ case JsonToken.START_OBJECT if pathTrie.namedChildren.isEmpty =>
+ false
+ case JsonToken.START_OBJECT if useTopLevelFastPath =>
+ extractTopLevelObject(parser, values, matched)
+ case JsonToken.START_OBJECT =>
+ extractObject(parser, pathTrie, values, matched)
+ case JsonToken.START_ARRAY if pathTrie.indexedChildren.isEmpty =>
+ false
+ case JsonToken.START_ARRAY =>
+ extractArray(parser, pathTrie, values, matched)
+ case _ =>
+ false
}
}
- if (validObject) {
+ if (validRoot) {
new GenericInternalRow(values)
} else {
nullRow
}
} catch {
- // Every simple named legacy extraction scans through the root object's
closing token, so a
+ // Every eligible legacy extraction scans through its root container's
closing token, so a
// syntax failure makes every sibling null without needing per-path
reparsing.
case _: JsonParseException => nullRow
// A parser-side rendering failure, such as a string-length constraint
violation, can leave
@@ -681,7 +693,7 @@ case class MultiGetJsonObjectEvaluator(
var token = parser.nextToken()
while (valid && token != null && token != JsonToken.END_OBJECT) {
if (token == JsonToken.FIELD_NAME) {
- val child =
node.children.get(parser.currentName).filter(_.hasUnmatched(matched))
+ val child =
node.namedChildren.get(parser.currentName).filter(_.hasUnmatched(matched))
val valueToken = parser.nextToken()
if (child.nonEmpty && valueToken != JsonToken.VALUE_NULL) {
valid = extractValue(parser, child.get, values, matched)
@@ -698,6 +710,29 @@ case class MultiGetJsonObjectEvaluator(
valid && token == JsonToken.END_OBJECT
}
+ private def extractArray(
+ parser: JsonParser,
+ node: MultiGetJsonObjectEvaluator.PathTrieNode,
+ values: Array[Any],
+ matched: Array[Boolean]): Boolean = {
+ var valid = true
+ var index = 0L
+ var token = parser.nextToken()
+ while (valid && token != null && token != JsonToken.END_ARRAY) {
+ val child =
node.indexedChildren.get(index).filter(_.hasUnmatched(matched))
+ if (child.nonEmpty) {
+ valid = extractValue(parser, child.get, values, matched)
+ } else {
+ parser.skipChildren()
+ }
+ if (valid) {
+ token = parser.nextToken()
+ index += 1
+ }
+ }
+ valid && token == JsonToken.END_ARRAY
+ }
+
private def extractValue(
parser: JsonParser,
node: MultiGetJsonObjectEvaluator.PathTrieNode,
@@ -714,6 +749,8 @@ case class MultiGetJsonObjectEvaluator(
true
} else if (parser.currentToken == JsonToken.START_OBJECT) {
extractObject(parser, node, values, matched)
+ } else if (parser.currentToken == JsonToken.START_ARRAY) {
+ extractArray(parser, node, values, matched)
} else {
parser.skipChildren()
true
@@ -794,36 +831,49 @@ case class MultiGetJsonObjectEvaluator(
object MultiGetJsonObjectEvaluator {
private final class MutablePathTrieNode {
val terminalOrdinals: mutable.ArrayBuffer[Int] = mutable.ArrayBuffer.empty
- val children: mutable.LinkedHashMap[String, MutablePathTrieNode] =
mutable.LinkedHashMap.empty
+ val namedChildren: mutable.LinkedHashMap[String, MutablePathTrieNode] =
+ mutable.LinkedHashMap.empty
+ val indexedChildren: mutable.LinkedHashMap[Long, MutablePathTrieNode] =
+ mutable.LinkedHashMap.empty
def freeze(): PathTrieNode = {
require(
- terminalOrdinals.isEmpty || children.isEmpty,
+ terminalOrdinals.isEmpty || (namedChildren.isEmpty &&
indexedChildren.isEmpty),
"Shared JSON paths must not be prefixes of one another")
- val frozenChildren = children.iterator.map { case (name, child) =>
+ val frozenNamedChildren = namedChildren.iterator.map { case (name,
child) =>
name -> child.freeze()
}.toMap
+ val frozenIndexedChildren = indexedChildren.iterator.map { case (index,
child) =>
+ index -> child.freeze()
+ }.toMap
val ordinals = (terminalOrdinals.iterator ++
-
frozenChildren.valuesIterator.flatMap(_.descendantOrdinals.iterator)).toArray
- PathTrieNode(terminalOrdinals.toArray, frozenChildren, ordinals)
+
frozenNamedChildren.valuesIterator.flatMap(_.descendantOrdinals.iterator) ++
+
frozenIndexedChildren.valuesIterator.flatMap(_.descendantOrdinals.iterator)).toArray
+ PathTrieNode(
+ terminalOrdinals.toArray, frozenNamedChildren, frozenIndexedChildren,
ordinals)
}
}
private case class PathTrieNode(
terminalOrdinals: Array[Int],
- children: Map[String, PathTrieNode],
+ namedChildren: Map[String, PathTrieNode],
+ indexedChildren: Map[Long, PathTrieNode],
descendantOrdinals: Array[Int]) {
def hasUnmatched(matched: Array[Boolean]): Boolean = {
descendantOrdinals.exists(index => !matched(index))
}
}
- private def buildPathTrie(paths: Seq[Seq[String]]): PathTrieNode = {
+ private def buildPathTrie(
+ paths: Seq[Seq[GetJsonObject.SimpleJsonPathSegment]]): PathTrieNode = {
val root = new MutablePathTrieNode
paths.zipWithIndex.foreach { case (path, ordinal) =>
var node = root
- path.foreach { fieldName =>
- node = node.children.getOrElseUpdate(fieldName, new
MutablePathTrieNode)
+ path.foreach {
+ case GetJsonObject.NamedPathSegment(fieldName) =>
+ node = node.namedChildren.getOrElseUpdate(fieldName, new
MutablePathTrieNode)
+ case GetJsonObject.IndexedPathSegment(index) =>
+ node = node.indexedChildren.getOrElseUpdate(index, new
MutablePathTrieNode)
}
node.terminalOrdinals += ordinal
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
index 471916c68860..b4f41fda42f0 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala
@@ -145,25 +145,31 @@ case class GetJsonObject(json: Expression, path:
Expression)
object GetJsonObject {
import PathInstruction._
- private[sql] def simpleNamedPath(path: UTF8String): Option[Seq[String]] = {
+ private[sql] sealed trait SimpleJsonPathSegment
+ private[sql] case class NamedPathSegment(name: String) extends
SimpleJsonPathSegment
+ private[sql] case class IndexedPathSegment(index: Long) extends
SimpleJsonPathSegment
+
+ private[sql] def simplePath(path: UTF8String):
Option[Seq[SimpleJsonPathSegment]] = {
try {
Option(path).flatMap(value =>
JsonPathParser.parse(value.toString)).flatMap { instructions =>
- val names = instructions.grouped(2).map {
- case List(Key, Named(fieldName)) => Some(fieldName)
+ val segments = instructions.grouped(2).map {
+ case List(Key, Named(fieldName)) => Some(NamedPathSegment(fieldName))
+ case List(Subscript, Index(index)) if index >= 0 =>
Some(IndexedPathSegment(index))
case _ => None
}.toSeq
- if (names.nonEmpty && names.forall(_.isDefined)) Some(names.flatten)
else None
+ if (segments.nonEmpty && segments.forall(_.isDefined))
Some(segments.flatten) else None
}
} catch {
// Numeric subscripts are parsed as Long and can overflow before the
parser returns None.
case _: NumberFormatException => None
}
}
+
}
/**
- * Extracts multiple simple named paths from a JSON string in one parse. This
is an internal
- * expression used to share sibling [[GetJsonObject]] expressions; unsupported
and
+ * Extracts multiple simple object-key and array-index paths from a JSON
string in one parse. This
+ * is an internal expression used to share sibling [[GetJsonObject]]
expressions; unsupported and
* prefix-conflicting JSON paths remain as independent GetJsonObject
expressions.
*/
case class MultiGetJsonObject(
@@ -194,8 +200,8 @@ case class MultiGetJsonObject(
final override val nodePatterns: Seq[TreePattern] = Seq(GET_JSON_OBJECT)
@transient
- private lazy val namedPaths = fallbackPaths.map { path =>
- GetJsonObject.simpleNamedPath(UTF8String.fromString(path)).getOrElse {
+ private lazy val simplePaths = fallbackPaths.map { path =>
+ GetJsonObject.simplePath(UTF8String.fromString(path)).getOrElse {
throw new IllegalArgumentException(s"Unsupported shared JSON path:
$path")
}
}
@@ -203,7 +209,7 @@ case class MultiGetJsonObject(
@transient
private lazy val evaluator = MultiGetJsonObjectEvaluator(
fallbackPaths.map(UTF8String.fromString),
- namedPaths)
+ simplePaths)
override def eval(input: InternalRow): Any = {
evaluator.evaluate(json.eval(input).asInstanceOf[UTF8String])
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
index a95c6807bc2c..ec4ac257b304 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeCsvJsonExprs.scala
@@ -44,29 +44,48 @@ import org.apache.spark.unsafe.types.UTF8String
object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
private def nameOfCorruptRecord = conf.columnNameOfCorruptRecord
+ private type SimpleJsonPath = Seq[GetJsonObject.SimpleJsonPathSegment]
+ private type SharedJsonCandidate = (GetJsonObject, SimpleJsonPath, String)
+ private type SharedJsonCandidateUnit = Seq[SharedJsonCandidate]
+ private type RequestedJsonPath = (SimpleJsonPath, String)
+ private type RequestedJsonPathUnit = Seq[RequestedJsonPath]
+
private case class SharedJsonFields(
json: Expression,
- paths: Seq[Seq[String]],
+ paths: Seq[SimpleJsonPath],
alias: Alias) {
- val ordinalMapping: Map[Seq[String], Int] = paths.zipWithIndex.toMap
+ val ordinalMapping: Map[SimpleJsonPath, Int] = paths.zipWithIndex.toMap
}
private final class SelectedJsonPathTrieNode {
var isTerminal: Boolean = false
var hasSelectedPathInSubtree: Boolean = false
- val children: mutable.HashMap[String, SelectedJsonPathTrieNode] =
mutable.HashMap.empty
+ val children: mutable.HashMap[GetJsonObject.SimpleJsonPathSegment,
SelectedJsonPathTrieNode] =
+ mutable.HashMap.empty
}
private final class SelectedJsonPathGroup {
val root = new SelectedJsonPathTrieNode
- val paths = mutable.ArrayBuffer.empty[(Seq[String], String)]
+ val paths = mutable.ArrayBuffer.empty[RequestedJsonPath]
- def tryAdd(candidate: (Seq[String], String)): Boolean = {
- val added = addIfNoPrefixConflict(root, candidate._1)
- if (added) {
- paths += candidate
+ def tryAdd(unit: RequestedJsonPathUnit): Boolean = {
+ val uniquePaths = mutable.LinkedHashMap.empty[SimpleJsonPath, String]
+ unit.foreach { case (path, fallbackPath) =>
+ uniquePaths.getOrElseUpdate(path, fallbackPath)
+ }
+ val newPaths = uniquePaths.iterator.filterNot { case (path, _) =>
+ containsExactPath(root, path)
+ }.toSeq
+ val conflictsWithGroup = newPaths.exists { case (path, _) =>
+ hasPrefixConflict(root, path)
+ }
+ if (!conflictsWithGroup) {
+ newPaths.foreach { case (path, _) => commitPath(root, path) }
+ paths ++= newPaths
+ true
+ } else {
+ false
}
- added
}
}
@@ -74,46 +93,69 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
// paths retain their existing independent GetJsonObject evaluation.
private val maxSharedJsonPathDepth = 64
- /** Inserts a path unless a previously selected path is its ancestor or
descendant. */
- private def addIfNoPrefixConflict(
+ private def containsExactPath(
root: SelectedJsonPathTrieNode,
- path: Seq[String]): Boolean = {
+ path: SimpleJsonPath): Boolean = {
var node = root
- val visited = mutable.ArrayBuffer(root)
var index = 0
- var hasSelectedPrefix = false
- while (index < path.length && !hasSelectedPrefix) {
+ while (index < path.length) {
+ node.children.get(path(index)) match {
+ case Some(child) =>
+ node = child
+ index += 1
+ case None =>
+ return false
+ }
+ }
+ node.isTerminal
+ }
+
+ private def hasPrefixConflict(
+ root: SelectedJsonPathTrieNode,
+ path: SimpleJsonPath): Boolean = {
+ var node = root
+ var index = 0
+ while (index < path.length) {
if (node.isTerminal) {
- hasSelectedPrefix = true
- } else {
- node = node.children.getOrElseUpdate(path(index), new
SelectedJsonPathTrieNode)
- visited += node
- index += 1
+ return true
+ }
+ node.children.get(path(index)) match {
+ case Some(child) =>
+ node = child
+ index += 1
+ case None =>
+ return false
}
}
+ !node.isTerminal && node.hasSelectedPathInSubtree
+ }
- val conflicts = hasSelectedPrefix || node.hasSelectedPathInSubtree
- if (!conflicts) {
- node.isTerminal = true
- visited.foreach(_.hasSelectedPathInSubtree = true)
+ private def commitPath(root: SelectedJsonPathTrieNode, path:
SimpleJsonPath): Unit = {
+ var node = root
+ val visited = mutable.ArrayBuffer(root)
+ path.foreach { segment =>
+ node = node.children.getOrElseUpdate(segment, new
SelectedJsonPathTrieNode)
+ visited += node
}
- !conflicts
+ node.isTerminal = true
+ visited.foreach(_.hasSelectedPathInSubtree = true)
}
- // First-fit produces the same groups as repeated greedy optimizer passes,
but does so in one
- // invocation so parallel prefix chains do not consume one fixed-point
iteration per depth.
+ // First-fit builds every prefix-free group in one invocation. Coalesce
candidates are added as
+ // atomic, prefix-free units so their mutually exclusive object/array paths
always use the same
+ // shared parse.
private def groupNonConflictingPaths(
- paths: Iterable[(Seq[String], String)]): Seq[Seq[(Seq[String], String)]]
= {
+ units: Iterable[RequestedJsonPathUnit]): Seq[Seq[RequestedJsonPath]] = {
val groups = mutable.ArrayBuffer.empty[SelectedJsonPathGroup]
- paths.foreach { candidate =>
+ units.foreach { unit =>
var added = false
val iterator = groups.iterator
while (!added && iterator.hasNext) {
- added = iterator.next().tryAdd(candidate)
+ added = iterator.next().tryAdd(unit)
}
if (!added) {
val group = new SelectedJsonPathGroup
- require(group.tryAdd(candidate))
+ require(group.tryAdd(unit))
groups += group
}
}
@@ -154,36 +196,35 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
}
/**
- * Share simple named GetJsonObject paths without changing the
Hive-compatible semantics of
- * wildcards or array subscripts. [[MultiGetJsonObject]] preserves the first
non-null
+ * Share simple named and array-index GetJsonObject paths without changing
the Hive-compatible
+ * semantics of wildcards. [[MultiGetJsonObject]] preserves the first
non-null
* duplicate-key match used by GetJsonObject, unlike JsonTuple.
Prefix-conflicting paths are
* placed in separate shared parses so each path retains independent legacy
evaluation.
*/
private def shareGetJsonObjects(project: Project): Project = {
- val candidates = project.projectList.flatMap(collectGetJsonObjectFields)
+ val candidateUnits =
project.projectList.flatMap(collectGetJsonObjectFields)
val groups = mutable.ArrayBuffer.empty[
- (Expression, mutable.ArrayBuffer[(Seq[String], String)])]
+ (Expression, mutable.ArrayBuffer[RequestedJsonPathUnit])]
val groupsByHash = mutable.HashMap.empty[
- Int, mutable.ArrayBuffer[(Expression, mutable.ArrayBuffer[(Seq[String],
String)])]]
+ Int, mutable.ArrayBuffer[(Expression,
mutable.ArrayBuffer[RequestedJsonPathUnit])]]
- candidates.foreach { case (getJsonObject, pathSegments, path) =>
+ candidateUnits.foreach { unit =>
+ val getJsonObject = unit.head._1
val bucket = groupsByHash.getOrElseUpdate(
getJsonObject.json.semanticHash(), mutable.ArrayBuffer.empty)
bucket.find(_._1.semanticEquals(getJsonObject.json)) match {
- case Some((_, fields)) => fields += pathSegments -> path
+ case Some((_, fields)) =>
+ fields += unit.map { case (_, pathSegments, path) => pathSegments ->
path }
case None =>
- val group = getJsonObject.json -> mutable.ArrayBuffer(pathSegments
-> path)
+ val requestedUnit = unit.map { case (_, pathSegments, path) =>
pathSegments -> path }
+ val group = getJsonObject.json -> mutable.ArrayBuffer(requestedUnit)
bucket += group
groups += group
}
}
- val sharedFields = groups.flatMap { case (json, requestedFields) =>
- val paths = mutable.LinkedHashMap.empty[Seq[String], String]
- requestedFields.foreach { case (pathSegments, path) =>
- paths.getOrElseUpdate(pathSegments, path)
- }
- groupNonConflictingPaths(paths).flatMap { nonConflictingPaths =>
+ val sharedFields = groups.flatMap { case (json, requestedUnits) =>
+ groupNonConflictingPaths(requestedUnits).flatMap { nonConflictingPaths =>
if (nonConflictingPaths.length > 1) {
val pathSegments = nonConflictingPaths.map(_._1)
val alias = Alias(
@@ -212,17 +253,22 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
}
private def collectGetJsonObjectFields(
- expression: Expression): Seq[(GetJsonObject, Seq[String], String)] = {
+ expression: Expression): Seq[SharedJsonCandidateUnit] = {
expression match {
case getJsonObject @ GetJsonObject(_: Attribute, Literal(path:
UTF8String, StringType))
if getJsonObject.deterministic =>
- GetJsonObject.simpleNamedPath(path)
+ GetJsonObject.simplePath(path)
.filter(_.length <= maxSharedJsonPathDepth)
- .map(pathSegments => (getJsonObject, pathSegments,
path.toString)).toSeq
+ .map { pathSegments =>
+ Seq((getJsonObject, pathSegments, path.toString))
+ }.toSeq
case _: GetJsonObject =>
Nil
+ case coalesce: Coalesce =>
+ eligibleCoalesceBranches(coalesce).map(_.map(_._2)).toSeq
+
case other =>
getJsonObjectTraversalChild(other).toSeq.flatMap(collectGetJsonObjectFields)
}
@@ -234,7 +280,7 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
expression match {
case getJsonObject @ GetJsonObject(json, Literal(path: UTF8String,
StringType)) =>
val replacement = for {
- pathSegments <- GetJsonObject.simpleNamedPath(path)
+ pathSegments <- GetJsonObject.simplePath(path)
shared <- sharedFieldsByHash.getOrElse(json.semanticHash(),
Nil).find { candidate =>
candidate.json.semanticEquals(json) &&
candidate.ordinalMapping.contains(pathSegments)
}
@@ -244,6 +290,32 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
case _: GetJsonObject =>
expression
+ case coalesce: Coalesce =>
+ eligibleCoalesceBranches(coalesce).map { branches =>
+ val selectedBranches = branches.toMap
+ val firstCandidate = branches.head._2
+ val shared = sharedFieldsByHash
+ .getOrElse(firstCandidate._1.json.semanticHash(), Nil)
+ .find { candidate =>
+ candidate.json.semanticEquals(firstCandidate._1.json) &&
+ branches.forall { case (_, branchCandidate) =>
+ val (_, pathSegments, _) = branchCandidate
+ candidate.ordinalMapping.contains(pathSegments)
+ }
+ }
+ shared.map { sharedFields =>
+ val pairSharedFields = Map(
+ firstCandidate._1.json.semanticHash() -> Seq(sharedFields))
+ coalesce.withNewChildren(coalesce.children.zipWithIndex.map { case
(child, index) =>
+ if (selectedBranches.contains(index)) {
+ rewriteGetJsonObjectFields(child, pairSharedFields)
+ } else {
+ child
+ }
+ })
+ }.getOrElse(coalesce)
+ }.getOrElse(coalesce)
+
case other =>
getJsonObjectTraversalChild(other).map { child =>
other.withNewChildren(
@@ -252,6 +324,50 @@ object OptimizeCsvJsonExprs extends Rule[LogicalPlan] {
}
}
+ /**
+ * Returns the first object-root and first array-root parser calls from a
coalesce only when every
+ * branch is a GetJsonObject, optionally wrapped in casts, and all branches
read the same input
+ * attribute. Only one root shape can match a given input. Later same-shape
fallbacks and all
+ * casts remain in the outer coalesce, preserving lazy evaluation and branch
ordering.
+ */
+ private def eligibleCoalesceBranches(
+ coalesce: Coalesce): Option[Seq[(Int, SharedJsonCandidate)]] = {
+ def eligibleBranch(expression: Expression): Option[SharedJsonCandidate] = {
+ expression match {
+ case cast: Cast => eligibleBranch(cast.child)
+ case getJsonObject @ GetJsonObject(_: Attribute, Literal(path:
UTF8String, StringType))
+ if getJsonObject.deterministic =>
+ GetJsonObject.simplePath(path)
+ .filter(_.length <= maxSharedJsonPathDepth)
+ .map(pathSegments => (getJsonObject, pathSegments, path.toString))
+ case _ => None
+ }
+ }
+
+ val branches = coalesce.children.map(eligibleBranch)
+ if (branches.nonEmpty && branches.forall(_.isDefined)) {
+ val candidates = branches.zipWithIndex.map { case (candidate, index) =>
+ index -> candidate.get
+ }
+ val sameInput = candidates.tail.forall { case (_, candidate) =>
+ candidate._1.json.semanticEquals(candidates.head._2._1.json)
+ }
+ val firstNamed = candidates.find { case (_, candidate) =>
+ candidate._2.head.isInstanceOf[GetJsonObject.NamedPathSegment]
+ }
+ val firstIndexed = candidates.find { case (_, candidate) =>
+ candidate._2.head.isInstanceOf[GetJsonObject.IndexedPathSegment]
+ }
+ if (sameInput && firstNamed.isDefined && firstIndexed.isDefined) {
+ Some(Seq(firstNamed.get, firstIndexed.get).sortBy(_._1))
+ } else {
+ None
+ }
+ } else {
+ None
+ }
+ }
+
private def getJsonObjectTraversalChild(expression: Expression):
Option[Expression] = {
expression match {
case _: ConditionalExpression | _: And | _: Or | _: In | _: TryEval |
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index f41337682378..9a9d5522ce6b 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -3891,8 +3891,9 @@ object SQLConf {
buildConf("spark.sql.optimizer.getJsonObjectSharedParsing.enabled")
.internal()
.doc(s"When true and '${JSON_EXPRESSION_OPTIMIZATION.key}' is also true,
the optimizer " +
- "replaces repeated simple named get_json_object paths over the same
input " +
- "with one shared parse.")
+ "replaces repeated simple named and array-index get_json_object paths
over the same " +
+ "input with one shared parse, including mutually exclusive
object/array coalesce " +
+ "branches.")
.version("4.3.0")
.withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
.booleanConf
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
index cb551b772ef3..dd0f3dd01327 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJsonExprsSuite.scala
@@ -228,7 +228,10 @@ class OptimizeJsonExprsSuite extends PlanTest with
ExpressionEvalHelper {
assert(!new SQLConf().getJsonObjectSharedParsingEnabled)
val query = testRelation2.select(
GetJsonObject($"json", Literal("$.a")).as("a"),
- GetJsonObject($"json", Literal("$.b")).as("b"))
+ GetJsonObject($"json", Literal("$.b")).as("b"),
+ Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.c")),
+ GetJsonObject($"json", Literal("$[0].c")))).as("c"))
withSQLConf(SQLConf.GET_JSON_OBJECT_SHARED_PARSING_ENABLED.key -> "false")
{
comparePlans(Optimizer.execute(query.analyze), query.analyze)
@@ -271,16 +274,205 @@ class OptimizeJsonExprsSuite extends PlanTest with
ExpressionEvalHelper {
}
}
- test("SPARK-57626: leave prefix-conflicting and unsupported paths
independent") {
+ test("SPARK-57990: share simple array-index get_json_object paths") {
+ val query = testRelation2.select(
+ GetJsonObject($"json", Literal("$[0].a")).as("first_a"),
+ GetJsonObject($"json", Literal("$.items[1].b")).as("second_b"),
+ GetJsonObject($"json", Literal("$[2]")).as("third"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ optimized match {
+ case Project(projectList, Project(innerProjectList, _: LocalRelation)) =>
+ val sharedAlias = innerProjectList.collectFirst {
+ case alias @ Alias(_: MultiGetJsonObject, "_shared_json_paths") =>
alias
+ }.getOrElse(fail(s"Missing shared JSON paths in plan:\n$optimized"))
+ val shared = sharedAlias.child.asInstanceOf[MultiGetJsonObject]
+ assert(shared.fallbackPaths == Seq("$[0].a", "$.items[1].b", "$[2]"))
+
+ val sharedAttr = sharedAlias.toAttribute
+ val extractedFields = projectList.flatMap(_.collect {
+ case getStructField: GetStructField
+ if getStructField.child.semanticEquals(sharedAttr) =>
getStructField
+ })
+ assert(extractedFields.map(_.ordinal) == Seq(0, 1, 2))
+
+ case _ =>
+ fail(s"Expected shared array-index JSON paths below the project, but
found:\n$optimized")
+ }
+ }
+
+ test("SPARK-57990: share get_json_object paths within coalesce") {
+ val query = testRelation2.select(
+ Coalesce(Seq(
+ Cast(GetJsonObject($"json", Literal("$.a")), IntegerType),
+ Cast(GetJsonObject($"json", Literal("$[0].a")), IntegerType))).as("a"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ optimized match {
+ case Project(Seq(Alias(coalesce: Coalesce, "a")),
+ Project(innerProjectList, _: LocalRelation)) =>
+ val sharedAlias = innerProjectList.collectFirst {
+ case alias @ Alias(_: MultiGetJsonObject, "_shared_json_paths") =>
alias
+ }.getOrElse(fail(s"Missing shared JSON paths in plan:\n$optimized"))
+ val shared = sharedAlias.child.asInstanceOf[MultiGetJsonObject]
+ assert(shared.fallbackPaths == Seq("$.a", "$[0].a"))
+
+ val sharedAttr = sharedAlias.toAttribute
+ assert(coalesce.children.zipWithIndex.forall { case (child, ordinal) =>
+ child match {
+ case Cast(getStructField: GetStructField, IntegerType, _, _) =>
+ getStructField.child.semanticEquals(sharedAttr) &&
getStructField.ordinal == ordinal
+ case _ => false
+ }
+ })
+
+ case _ =>
+ fail(s"Expected shared JSON paths within coalesce, but
found:\n$optimized")
+ }
+ }
+
+ test("SPARK-57990: share only the first get_json_object path for each
coalesce root shape") {
+ val query = testRelation2.select(
+ Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ GetJsonObject($"json", Literal("$.b")),
+ GetJsonObject($"json", Literal("$[0].a")),
+ GetJsonObject($"json", Literal("$[0].b")))).as("value"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ optimized match {
+ case Project(Seq(Alias(coalesce: Coalesce, "value")),
+ Project(innerProjectList, _: LocalRelation)) =>
+ val sharedAlias = innerProjectList.collectFirst {
+ case alias @ Alias(_: MultiGetJsonObject, "_shared_json_paths") =>
alias
+ }.getOrElse(fail(s"Missing shared JSON paths in plan:\n$optimized"))
+ val shared = sharedAlias.child.asInstanceOf[MultiGetJsonObject]
+ assert(shared.fallbackPaths == Seq("$.a", "$[0].a"))
+
+ val sharedAttr = sharedAlias.toAttribute
+ coalesce.children.zipWithIndex.foreach {
+ case (getStructField: GetStructField, 0) =>
+ assert(getStructField.child.semanticEquals(sharedAttr))
+ assert(getStructField.ordinal == 0)
+ case (getStructField: GetStructField, 2) =>
+ assert(getStructField.child.semanticEquals(sharedAttr))
+ assert(getStructField.ordinal == 1)
+ case (GetJsonObject(_, Literal(path: UTF8String, StringType)), 1) =>
+ assert(path.toString == "$.b")
+ case (GetJsonObject(_, Literal(path: UTF8String, StringType)), 3) =>
+ assert(path.toString == "$[0].b")
+ case (child, index) =>
+ fail(s"Unexpected coalesce branch $index after sharing: $child")
+ }
+
+ case _ =>
+ fail(s"Expected selectively shared coalesce paths, but
found:\n$optimized")
+ }
+ }
+
+ test("SPARK-57990: keep coalesce path pairs atomic across prefix groups") {
+ // $.a conflicts with $.a.x but not $[0].x, so both coalesce paths must
move together.
+ val query = testRelation2.select(
+ GetJsonObject($"json", Literal("$.a")).as("a"),
+ Coalesce(Seq(
+ GetJsonObject($"json", Literal("$[0].x")),
+ GetJsonObject($"json", Literal("$.a.x")))).as("x"),
+ Coalesce(Seq(
+ GetJsonObject($"json", Literal("$[0].y")),
+ GetJsonObject($"json", Literal("$.a.y")))).as("y"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ optimized match {
+ case Project(projectList, Project(innerProjectList, _: LocalRelation)) =>
+ val sharedAliases = innerProjectList.collect {
+ case alias @ Alias(_: MultiGetJsonObject, "_shared_json_paths") =>
alias
+ }
+ assert(sharedAliases.length == 1)
+ val sharedAlias = sharedAliases.head
+ val shared = sharedAlias.child.asInstanceOf[MultiGetJsonObject]
+ assert(shared.fallbackPaths == Seq("$[0].x", "$.a.x", "$[0].y",
"$.a.y"))
+
+ val sharedAttr = sharedAlias.toAttribute
+ Seq("x", "y").foreach { name =>
+ val coalesce = projectList.collectFirst {
+ case Alias(expression: Coalesce, aliasName) if aliasName == name
=> expression
+ }.getOrElse(fail(s"Missing $name coalesce in plan:\n$optimized"))
+ assert(coalesce.children.forall {
+ case getStructField: GetStructField =>
+ getStructField.child.semanticEquals(sharedAttr)
+ case _ => false
+ })
+ }
+
+ assert(projectList.exists {
+ case Alias(GetJsonObject(_, Literal(path: UTF8String, StringType)),
"a") =>
+ path.toString == "$.a"
+ case _ => false
+ })
+
+ case _ =>
+ fail(s"Expected atomic shared coalesce paths, but found:\n$optimized")
+ }
+ }
+
+ test("SPARK-57990: share get_json_object paths within coalesce below an
outer cast") {
+ val query = testRelation2.select(
+ Cast(Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ GetJsonObject($"json", Literal("$[0].a")))), IntegerType).as("a"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ val shared = optimized.collect {
+ case Project(projectList, _) => projectList.collectFirst {
+ case Alias(shared: MultiGetJsonObject, "_shared_json_paths") => shared
+ }
+ }.flatten.headOption.getOrElse(fail(s"Missing shared JSON paths in
plan:\n$optimized"))
+ assert(shared.fallbackPaths == Seq("$.a", "$[0].a"))
+
+ optimized match {
+ case Project(Seq(Alias(Cast(coalesce: Coalesce, IntegerType, _, _),
"a")), _) =>
+ assert(coalesce.children.forall(_.isInstanceOf[GetStructField]))
+ case _ => fail(s"Expected cast around rewritten coalesce, but
found:\n$optimized")
+ }
+ }
+
+ test("SPARK-57990: do not share unsafe coalesce branches") {
+ val otherJsonAttr = $"other_json".string
+ val relation = LocalRelation(jsonAttr, otherJsonAttr)
+ val queries = Seq(
+ relation.select(Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ GetJsonObject($"json", Literal("$.b")))).as("value")),
+ relation.select(Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ Literal("fallback"),
+ GetJsonObject($"json", Literal("$.b")))).as("value")),
+ relation.select(Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ GetJsonObject($"other_json", Literal("$.b")))).as("value")),
+ relation.select(Coalesce(Seq(
+ GetJsonObject($"json", Literal("$.a")),
+ GetJsonObject($"json", Literal("$.items[*]")))).as("value")))
+
+ queries.foreach { query =>
+ val optimized = Optimizer.execute(query.analyze)
+ assert(!optimized.exists { plan =>
+ plan.expressions.exists(_.exists(_.isInstanceOf[MultiGetJsonObject]))
+ })
+ comparePlans(optimized, query.analyze)
+ }
+ }
+
+ test("SPARK-57990: leave prefix-conflicting and unsupported paths
independent") {
val deepPath = (1 to 65).map(index => s"field$index").mkString("$.", ".",
"")
- val legacyPaths = Seq("$.a.b", "$.items[0].id", "$.a.*", deepPath)
+ val legacyPaths = Seq("$.a.b", "$.a.*", deepPath)
val query = testRelation2.select(
GetJsonObject($"json", Literal("$.a")).as("a"),
GetJsonObject($"json", Literal(legacyPaths(0))).as("nested"),
GetJsonObject($"json", Literal("$.c.d")).as("d"),
- GetJsonObject($"json", Literal(legacyPaths(1))).as("array"),
- GetJsonObject($"json", Literal(legacyPaths(2))).as("wildcard"),
- GetJsonObject($"json", Literal(legacyPaths(3))).as("deep"),
+ GetJsonObject($"json", Literal("$.items[0].id")).as("array"),
+ GetJsonObject($"json", Literal(legacyPaths(1))).as("wildcard"),
+ GetJsonObject($"json", Literal(legacyPaths(2))).as("deep"),
GetJsonObject($"json", Literal("$.e")).as("e"))
val optimized = Optimizer.execute(query.analyze)
@@ -290,7 +482,7 @@ class OptimizeJsonExprsSuite extends PlanTest with
ExpressionEvalHelper {
}
}.flatten.headOption.getOrElse(fail(s"Missing shared JSON paths in
plan:\n$optimized"))
.child.asInstanceOf[MultiGetJsonObject]
- assert(shared.fallbackPaths == Seq("$.a", "$.c.d", "$.e"))
+ assert(shared.fallbackPaths == Seq("$.a", "$.c.d", "$.items[0].id", "$.e"))
val remainingPaths = optimized.expressions.flatMap(_.collect {
case GetJsonObject(_, Literal(path: UTF8String, StringType)) =>
path.toString
@@ -319,6 +511,27 @@ class OptimizeJsonExprsSuite extends PlanTest with
ExpressionEvalHelper {
}))
}
+ test("SPARK-57990: keep array-index prefix conflicts independent") {
+ val query = testRelation2.select(
+ GetJsonObject($"json", Literal("$[0].a")).as("a"),
+ GetJsonObject($"json", Literal("$[0]")).as("first"),
+ GetJsonObject($"json", Literal("$[0].b")).as("b"),
+ GetJsonObject($"json", Literal("$[1]")).as("second"))
+ val optimized = Optimizer.execute(query.analyze)
+
+ val shared = optimized.collect {
+ case Project(projectList, _) => projectList.collectFirst {
+ case alias @ Alias(_: MultiGetJsonObject, "_shared_json_paths") =>
alias
+ }
+ }.flatten.headOption.getOrElse(fail(s"Missing shared JSON paths in
plan:\n$optimized"))
+ .child.asInstanceOf[MultiGetJsonObject]
+ assert(shared.fallbackPaths == Seq("$[0].a", "$[0].b", "$[1]"))
+ assert(optimized.expressions.exists(_.exists {
+ case GetJsonObject(_, Literal(path: UTF8String, StringType)) =>
path.toString == "$[0]"
+ case _ => false
+ }))
+ }
+
test("SPARK-57626: share parallel prefix chains in one optimizer
invocation") {
def prefixChain(root: String): Seq[String] = (1 to 9).map { depth =>
(1 to depth).map(index => s"$root$index").mkString("$.", ".", "")
diff --git a/sql/core/benchmarks/SharedJsonParseBenchmark-jdk21-results.txt
b/sql/core/benchmarks/SharedJsonParseBenchmark-jdk21-results.txt
index f4e2d3112fbb..f8d596422dc1 100644
--- a/sql/core/benchmarks/SharedJsonParseBenchmark-jdk21-results.txt
+++ b/sql/core/benchmarks/SharedJsonParseBenchmark-jdk21-results.txt
@@ -6,56 +6,84 @@ OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1625 1667
46 0.1 8125.0 1.0X
-shared parsing on 910 933
20 0.2 4552.2 1.8X
+shared parsing off 1685 1714
25 0.1 8426.5 1.0X
+shared parsing on 965 977
15 0.2 4824.9 1.7X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 3154 3178
21 0.1 15770.7 1.0X
-shared parsing on 1130 1178
45 0.2 5649.4 2.8X
+shared parsing off 3308 3389
81 0.1 16539.7 1.0X
+shared parsing on 1116 1124
7 0.2 5579.1 3.0X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 6067 6175
93 0.0 30333.8 1.0X
-shared parsing on 1234 1265
27 0.2 6170.1 4.9X
+shared parsing off 6565 6597
51 0.0 32824.1 1.0X
+shared parsing on 1341 1346
6 0.1 6706.1 4.9X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------
-shared parsing off 11938 11946
13 0.0 59690.5 1.0X
-shared parsing on 1699 1709
15 0.1 8493.5 7.0X
+shared parsing off 13148 13178
30 0.0 65742.5 1.0X
+shared parsing on 1800 1820
18 0.1 8999.8 7.3X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1600 1607
10 0.1 8001.8 1.0X
-shared parsing on 873 900
25 0.2 4367.2 1.8X
+shared parsing off 1565 1572
6 0.1 7822.6 1.0X
+shared parsing on 917 922
7 0.2 4587.4 1.7X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 3120 3156
35 0.1 15600.5 1.0X
-shared parsing on 1102 1115
18 0.2 5509.8 2.8X
+shared parsing off 3044 3068
30 0.1 15221.3 1.0X
+shared parsing on 1125 1131
11 0.2 5626.8 2.7X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 6362 6408
46 0.0 31809.2 1.0X
-shared parsing on 1347 1367
19 0.1 6733.2 4.7X
+shared parsing off 6062 6070
7 0.0 30310.2 1.0X
+shared parsing on 1355 1373
22 0.1 6772.6 4.5X
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 12772
12831 54 0.0 63861.1 1.0X
-shared parsing on 1852
1868 15 0.1 9260.4 6.9X
+shared parsing off 12061
12098 34 0.0 60305.8 1.0X
+shared parsing on 1913
1919 5 0.1 9565.3 6.3X
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 2 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 2456
2466 10 0.1 12279.2 1.0X
+shared parsing on 1002
1009 9 0.2 5009.5 2.5X
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 4 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 4835
4844 9 0.0 24174.8 1.0X
+shared parsing on 1276
1293 17 0.2 6381.5 3.8X
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 8 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 9578
9604 27 0.0 47890.8 1.0X
+shared parsing on 1466
1478 11 0.1 7329.2 6.5X
+
+OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 16 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 19267
19315 50 0.0 96336.7 1.0X
+shared parsing on 1975
1992 15 0.1 9874.2 9.8X
diff --git a/sql/core/benchmarks/SharedJsonParseBenchmark-jdk25-results.txt
b/sql/core/benchmarks/SharedJsonParseBenchmark-jdk25-results.txt
index 5ab0fd777a7f..5efc7ab149ce 100644
--- a/sql/core/benchmarks/SharedJsonParseBenchmark-jdk25-results.txt
+++ b/sql/core/benchmarks/SharedJsonParseBenchmark-jdk25-results.txt
@@ -3,59 +3,87 @@ Benchmark for sharing repeated get_json_object parsing
================================================================================================
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1567 1580
18 0.1 7835.8 1.0X
-shared parsing on 927 933
6 0.2 4634.1 1.7X
+shared parsing off 1583 1612
36 0.1 7914.7 1.0X
+shared parsing on 956 985
37 0.2 4780.2 1.7X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 3002 3015
14 0.1 15008.5 1.0X
-shared parsing on 1187 1218
28 0.2 5933.7 2.5X
+shared parsing off 3082 3090
8 0.1 15407.6 1.0X
+shared parsing on 1087 1096
8 0.2 5435.4 2.8X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 6027 6030
2 0.0 30136.0 1.0X
-shared parsing on 1290 1305
16 0.2 6449.2 4.7X
+shared parsing off 6098 6123
29 0.0 30491.0 1.0X
+shared parsing on 1300 1321
28 0.2 6498.6 4.7X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------
-shared parsing off 12003 12020
25 0.0 60015.7 1.0X
-shared parsing on 1724 1728
5 0.1 8620.3 7.0X
+shared parsing off 12234 12259
42 0.0 61168.6 1.0X
+shared parsing on 1744 1751
7 0.1 8717.7 7.0X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1620 1637
16 0.1 8098.1 1.0X
-shared parsing on 966 968
2 0.2 4831.3 1.7X
+shared parsing off 1615 1627
13 0.1 8074.2 1.0X
+shared parsing on 931 953
19 0.2 4655.2 1.7X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 3107 3136
28 0.1 15532.8 1.0X
-shared parsing on 1185 1193
7 0.2 5925.4 2.6X
+shared parsing off 3164 3178
14 0.1 15822.3 1.0X
+shared parsing on 1258 1262
4 0.2 6287.7 2.5X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 6162 6178
21 0.0 30809.9 1.0X
-shared parsing on 1405 1421
14 0.1 7025.9 4.4X
+shared parsing off 6272 6283
10 0.0 31361.5 1.0X
+shared parsing on 1473 1476
3 0.1 7366.8 4.3X
OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
-AMD EPYC 9V74 80-Core Processor
+AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 12341
12350 12 0.0 61704.0 1.0X
-shared parsing on 1834
1846 10 0.1 9171.9 6.7X
+shared parsing off 12533
12549 20 0.0 62667.3 1.0X
+shared parsing on 1990
2010 20 0.1 9950.0 6.3X
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 2 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 2203
2221 18 0.1 11012.6 1.0X
+shared parsing on 989
999 10 0.2 4945.7 2.2X
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 4 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 4413
4420 7 0.0 22065.9 1.0X
+shared parsing on 1187
1200 13 0.2 5933.0 3.7X
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 8 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 8692
8705 14 0.0 43459.9 1.0X
+shared parsing on 1375
1389 16 0.1 6873.5 6.3X
+
+OpenJDK 64-Bit Server VM 25.0.3+9-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 16 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 17609
17639 28 0.0 88043.8 1.0X
+shared parsing on 1906
1913 11 0.1 9527.6 9.2X
diff --git a/sql/core/benchmarks/SharedJsonParseBenchmark-results.txt
b/sql/core/benchmarks/SharedJsonParseBenchmark-results.txt
index 6c90a4479c3a..f7982bc9c311 100644
--- a/sql/core/benchmarks/SharedJsonParseBenchmark-results.txt
+++ b/sql/core/benchmarks/SharedJsonParseBenchmark-results.txt
@@ -6,56 +6,84 @@ OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux
6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1577 1587
15 0.1 7883.7 1.0X
-shared parsing on 932 946
16 0.2 4661.0 1.7X
+shared parsing off 1550 1571
25 0.1 7750.2 1.0X
+shared parsing on 874 889
13 0.2 4370.5 1.8X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 3005 3052
41 0.1 15025.5 1.0X
-shared parsing on 1074 1075
2 0.2 5370.1 2.8X
+shared parsing off 2940 2948
12 0.1 14701.0 1.0X
+shared parsing on 1018 1026
7 0.2 5090.3 2.9X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------------------------------------
-shared parsing off 6090 6104
12 0.0 30448.1 1.0X
-shared parsing on 1315 1321
8 0.2 6575.3 4.6X
+shared parsing off 5820 5833
20 0.0 29099.7 1.0X
+shared parsing on 1256 1257
2 0.2 6280.4 4.6X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 fields: Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------
-shared parsing off 12144 12159
17 0.0 60719.4 1.0X
-shared parsing on 1840 1848
7 0.1 9201.7 6.6X
+shared parsing off 11588 11623
38 0.0 57942.2 1.0X
+shared parsing on 1736 1742
8 0.1 8681.5 6.7X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 2 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 1498 1509
9 0.1 7492.3 1.0X
-shared parsing on 894 895
1 0.2 4471.6 1.7X
+shared parsing off 1585 1595
9 0.1 7922.5 1.0X
+shared parsing on 914 917
3 0.2 4572.3 1.7X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 4 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 2969 2978
8 0.1 14845.4 1.0X
-shared parsing on 1129 1131
4 0.2 5644.8 2.6X
+shared parsing off 3134 3151
26 0.1 15672.4 1.0X
+shared parsing on 1155 1159
6 0.2 5774.6 2.7X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 8 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
--------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 5841 5847
9 0.0 29206.9 1.0X
-shared parsing on 1378 1384
7 0.1 6889.6 4.2X
+shared parsing off 6302 6311
14 0.0 31509.0 1.0X
+shared parsing on 1377 1382
10 0.1 6883.6 4.6X
OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
AMD EPYC 7763 64-Core Processor
get_json_object extracting 16 of 32 nested fields: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
---------------------------------------------------------------------------------------------------------------------------------
-shared parsing off 11856
11904 64 0.0 59278.4 1.0X
-shared parsing on 1926
1936 12 0.1 9630.4 6.2X
+shared parsing off 12228
12247 16 0.0 61139.6 1.0X
+shared parsing on 1901
1907 8 0.1 9504.9 6.4X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 2 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 2199
2203 4 0.1 10994.3 1.0X
+shared parsing on 999
1001 2 0.2 4995.2 2.2X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 4 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 4322
4339 16 0.0 21608.3 1.0X
+shared parsing on 1239
1249 9 0.2 6195.2 3.5X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 8 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+-------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 8652
8658 6 0.0 43257.7 1.0X
+shared parsing on 1397
1400 3 0.1 6987.2 6.2X
+
+OpenJDK 64-Bit Server VM 17.0.19+10-LTS on Linux 6.17.0-1018-azure
+AMD EPYC 7763 64-Core Processor
+get_json_object coalescing 16 object/array field paths: Best Time(ms) Avg
Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative
+--------------------------------------------------------------------------------------------------------------------------------------
+shared parsing off 17561
17574 22 0.0 87802.5 1.0X
+shared parsing on 1908
1908 1 0.1 9537.9 9.2X
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
index 396b86144f4d..4bbd3b533d34 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/JsonFunctionsSuite.scala
@@ -27,7 +27,8 @@ import com.fasterxml.jackson.core.StreamReadConstraints
import org.apache.spark.{SparkException, SparkRuntimeException}
import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.{JsonToStructs, Literal,
MultiGetJsonObject}
+import org.apache.spark.sql.catalyst.expressions.{GetJsonObject,
JsonToStructs, Literal,
+ MultiGetJsonObject}
import org.apache.spark.sql.catalyst.expressions.Cast._
import org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils
import
org.apache.spark.sql.catalyst.util.TimestampNanosTestUtils.foreachNanosPrecision
@@ -38,6 +39,7 @@ import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.sql.types.DayTimeIntervalType.{DAY, HOUR, MINUTE,
SECOND}
import org.apache.spark.sql.types.YearMonthIntervalType.{MONTH, YEAR}
+import org.apache.spark.unsafe.types.UTF8String
class JsonFunctionsSuite extends SharedSparkSession {
import testImplicits._
@@ -167,6 +169,149 @@ class JsonFunctionsSuite extends SharedSparkSession {
Row("single", "7", "8", "9")))
}
+ test("SPARK-57990: share simple array-index get_json_object paths") {
+ val input = Seq[String](
+ """[{"name":"root-zero","value":"raw"},{"name":"root-one"}]""",
+
"""{"items":[{"name":"item-zero"},{"name":"item-one","value":{"x":1}}],""" +
+ """"matrix":[[0,"one"]]}""",
+ """{"items":[null,{"name":"after-null","value":[1,2]}],""" +
+ """"matrix":[[0,null]]}""",
+ """{"items":[{"name":"first"}],"items":[{"name":"second-zero"},""" +
+ """{"name":"second-one"}]}""",
+ """{"items":[{"name":null,"name":"after-null"},""" +
+ """{"name":"first","name":"second"}]}""",
+ """{"items":[{"name":"before"},{"bad":"\q"}],"other":1}""",
+ """[{"name":"before"},{"bad":"\q"}]""",
+ """{'items':[{'name':'single'},{'value':7}]}""",
+ """[]""",
+ """{}""",
+ """1""",
+ null)
+
+ def result(jsonOptimization: Boolean, sharedParsing: Boolean): Seq[Row] = {
+ var rows = Seq.empty[Row]
+ withSQLConf(
+ SQLConf.JSON_EXPRESSION_OPTIMIZATION.key ->
jsonOptimization.toString,
+ SQLConf.GET_JSON_OBJECT_SHARED_PARSING_ENABLED.key ->
sharedParsing.toString) {
+ val query = input.toDF("json").select(
+ get_json_object($"json", "$[0].name"),
+ get_json_object($"json", "$[1].name"),
+ get_json_object($"json", "$[0].value"),
+ get_json_object($"json", "$.items[0].name"),
+ get_json_object($"json", "$.items[1].name"),
+ get_json_object($"json", "$.items[1].value"),
+ get_json_object($"json", "$.matrix[0][1]"),
+ get_json_object($"json", "$.items[9].name"))
+ val hasSharedParsing = query.queryExecution.optimizedPlan.exists {
plan =>
+ plan.expressions.exists(_.exists(_.isInstanceOf[MultiGetJsonObject]))
+ }
+ if (jsonOptimization) {
+ assert(hasSharedParsing == sharedParsing)
+ }
+ rows = query.collect().toSeq
+ }
+ rows
+ }
+
+ val legacy = result(jsonOptimization = false, sharedParsing = false)
+ assert(result(jsonOptimization = true, sharedParsing = false) == legacy)
+ assert(result(jsonOptimization = true, sharedParsing = true) == legacy)
+ assert(legacy.take(8) == Seq(
+ Row("root-zero", "root-one", "raw", null, null, null, null, null),
+ Row(null, null, null, "item-zero", "item-one", "{\"x\":1}", "one", null),
+ Row(null, null, null, null, "after-null", "[1,2]", "null", null),
+ Row(null, null, null, "first", "second-one", null, null, null),
+ Row(null, null, null, "after-null", "first", null, null, null),
+ Row(null, null, null, null, null, null, null, null),
+ Row(null, null, null, null, null, null, null, null),
+ Row(null, null, null, "single", null, "7", null, null)))
+ }
+
+ test("SPARK-57990: shared get_json_object preserves object-or-array coalesce
semantics") {
+ val input = Seq[String](
+ """{"name":"object","nested":[{"name":"object-nested"}]}""",
+ """[{"name":"array","nested":[{"name":"array-nested"}]},{"other":1}]""",
+ """{"name":null,"nested":[null]}""",
+ """[null]""",
+ """1""",
+ """[{"name":"before"},{"bad":"\q"}]""",
+ """{"name":"before","bad":"\q"}""",
+ null)
+
+ def result(sharedParsing: Boolean): Seq[Row] = {
+ var rows = Seq.empty[Row]
+ withSQLConf(
+ SQLConf.JSON_EXPRESSION_OPTIMIZATION.key -> "true",
+ SQLConf.GET_JSON_OBJECT_SHARED_PARSING_ENABLED.key ->
sharedParsing.toString) {
+ val query = input.toDF("json").select(
+ coalesce(
+ get_json_object($"json", "$.name"),
+ get_json_object($"json", "$[0].name")),
+ coalesce(
+ get_json_object($"json", "$.nested[0].name"),
+ get_json_object($"json", "$[0].nested[0].name")))
+ val hasSharedParsing = query.queryExecution.optimizedPlan.exists {
plan =>
+ plan.expressions.exists(_.exists(_.isInstanceOf[MultiGetJsonObject]))
+ }
+ assert(hasSharedParsing == sharedParsing)
+ rows = query.collect().toSeq
+ }
+ rows
+ }
+
+ val legacy = result(sharedParsing = false)
+ assert(result(sharedParsing = true) == legacy)
+ assert(legacy == Seq(
+ Row("object", "object-nested"),
+ Row("array", "array-nested"),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null),
+ Row(null, null)))
+ }
+
+ test("SPARK-57990: shared get_json_object validates simple array-index
paths") {
+ import GetJsonObject.{IndexedPathSegment, NamedPathSegment}
+
+ def simplePath(path: String):
Option[Seq[GetJsonObject.SimpleJsonPathSegment]] = {
+ GetJsonObject.simplePath(UTF8String.fromString(path))
+ }
+
+ assert(simplePath("$.a[0]['b'][2]") == Some(Seq(
+ NamedPathSegment("a"), IndexedPathSegment(0), NamedPathSegment("b"),
+ IndexedPathSegment(2))))
+ Seq("$", "$[*]", "$.a[*]", "$.a[-1]", "$.a[1x]", "$[9223372036854775808]")
+ .foreach(path => assert(simplePath(path).isEmpty, path))
+
+ val expression = MultiGetJsonObject(
+ Literal("""["raw",{"x":1},[2,3],null]"""),
+ Seq("$[0]", "$[1]", "$[1]", "$[2][1]", "$[3]", "$[9]"))
+ val row = expression.eval().asInstanceOf[InternalRow]
+ assert(row.getUTF8String(0).toString == "raw")
+ assert(row.getUTF8String(1).toString == "{\"x\":1}")
+ assert(row.getUTF8String(2).toString == "{\"x\":1}")
+ assert(row.getUTF8String(3).toString == "3")
+ assert(row.getUTF8String(4).toString == "null")
+ assert(row.isNullAt(5))
+ assert(row.getUTF8String(4) == GetJsonObject(
+ Literal("""["raw",{"x":1},[2,3],null]"""), Literal("$[3]")).eval())
+
+ val nestedNull = MultiGetJsonObject(
+ Literal("""{"a":[null],"b":1}"""), Seq("$.a[0]", "$.b")).eval()
+ .asInstanceOf[InternalRow]
+ assert(nestedNull.getUTF8String(0).toString == "null")
+ assert(nestedNull.getUTF8String(0) == GetJsonObject(
+ Literal("""{"a":[null],"b":1}"""), Literal("$.a[0]")).eval())
+ assert(nestedNull.getUTF8String(1).toString == "1")
+
+ val prefixConflict = MultiGetJsonObject(
+ Literal("""[{"x":1}]"""), Seq("$[0]", "$[0].x"))
+ val error = intercept[IllegalArgumentException](prefixConflict.eval())
+ assert(error.getMessage.contains("must not be prefixes"))
+ }
+
test("SPARK-57626: shared nested get_json_object isolates value rendering
failures") {
val invalidSurrogate = "\\" + "uD800"
val input = Seq(
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/SharedJsonParseBenchmark.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/SharedJsonParseBenchmark.scala
index 48aa5842858b..8a9f74b66b91 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/SharedJsonParseBenchmark.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/json/SharedJsonParseBenchmark.scala
@@ -109,6 +109,45 @@ object SharedJsonParseBenchmark extends SqlBasedBenchmark {
}
nestedData.unpersist()
+
+ val payload = struct(Seq.tabulate(fieldCount) { index =>
+ fieldValue.as(s"field_$index")
+ }: _*)
+ val mixedObjectArrayData = spark.range(0, rows, 1, 4)
+ .select(when(($"id" % 2) === 0, to_json(payload))
+ .otherwise(to_json(array(payload)))
+ .as("json"))
+ .cache()
+ mixedObjectArrayData.count()
+
+ Seq(2, 4, 8, 16).foreach { selectedFieldCount =>
+ val pathBenchmark = new Benchmark(
+ s"get_json_object coalescing $selectedFieldCount object/array field
paths",
+ rows,
+ output = output)
+
+ def extractPaths(sharedParsing: Boolean): Unit = {
+ withSQLConf(
+ SQLConf.JSON_EXPRESSION_OPTIMIZATION.key -> "true",
+ SQLConf.GET_JSON_OBJECT_SHARED_PARSING_ENABLED.key ->
sharedParsing.toString) {
+ mixedObjectArrayData.select(Seq.tabulate(selectedFieldCount) {
index =>
+ coalesce(
+ get_json_object($"json", s"$$.field_$index"),
+ get_json_object($"json", s"$$[0].field_$index"))
+ }: _*).noop()
+ }
+ }
+
+ pathBenchmark.addCase("shared parsing off", 3) { _ =>
+ extractPaths(sharedParsing = false)
+ }
+ pathBenchmark.addCase("shared parsing on", 3) { _ =>
+ extractPaths(sharedParsing = true)
+ }
+ pathBenchmark.run()
+ }
+
+ mixedObjectArrayData.unpersist()
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
