[SPARK-57626][SQL] Share repeated nested JSON path parsing

### What changes were proposed in this pull request?

This PR extends the internal shared parser introduced by [SPARK-47670](https://issues.apache.org/jira/browse/SPARK-47670) and #56547 from simple top-level fields to repeated literal named object paths.

The proposed changes:

- Parse literal named JSON paths into field-name segments. Both dot notation such as `$.payload.user.id` and quoted names such as `$['payload']['user.name']` are supported.
- Build a runtime path trie so multiple nested fields can be extracted in one streaming JSON scan.
- Preserve the first non-null duplicate-key match independently for each requested path, including duplicate parent objects and non-object intermediate values.
- Keep ancestor and descendant paths in separate shared parses. For example, `$.a` never shares a parse with `$.a.b`.
- Build all greedy prefix-free groups in one optimizer invocation, avoiding repeated fixed-point iterations for parallel prefix chains.
- Retain the existing top-level fast path so top-level sharing does not pay for runtime trie traversal.
- Leave dynamic paths, wildcards, array subscripts, and paths deeper than 64 named fields on their existing independent `GetJsonObject` evaluation.
- Continue using the existing internal, default-disabled `spark.sql.optimizer.getJsonObjectSharedParsing.enabled` configuration. The existing JSON expression optimization must also be enabled.
- Extend optimizer, runtime, code-generation, malformed-input, and microbenchmark coverage for nested paths.

For example, these prefix-free paths share one parse:

```text
$.payload.user.id
$.payload.user.name
$.payload.request_id
```

For this ordered set of ancestor/descendant paths:

```text
$.a
$.a.b
$.x
$.x.y
```

the optimizer creates two prefix-free groups in one invocation:

```text
group 1: $.a,   $.x
group 2: $.a.b, $.x.y
```

Unsupported forms such as `$.items[0].id`, `$.payload.*`, and paths supplied by another column remain unchanged.

### Why are the changes needed?

[SPARK-57626](https://issues.apache.org/jira/browse/SPARK-57626) follows up on the initial shared-parsing optimization. The initial implementation intentionally supports only simple top-level fields, so repeated literal nested paths still parse the same JSON independently.

For example, consider an `events` table whose `json` column contains:

```json
{"payload":{"user":{"id":123,"name":"alice"},"request_id":"r-1"}}
```

An existing query might run:

```sql
SELECT
  get_json_object(json, '$.payload.user.id') AS user_id,
  get_json_object(json, '$.payload.user.name') AS user_name,
  get_json_object(json, '$.payload.request_id') AS request_id
FROM events;
```

Before this PR, Spark parses each row's `json` value independently for all three nested extractions. With shared parsing enabled, Catalyst rewrites them to one internal `MultiGetJsonObject`, so the input is scanned once and all three values are returned from that scan. The SQL and its results do not change.

### Does this PR introduce _any_ user-facing change?

Yes, within the unreleased `master` branch only. When the existing internal, default-disabled `spark.sql.optimizer.getJsonObjectSharedParsing.enabled` configuration is enabled, eligible repeated simple nested `get_json_object` paths now use shared parsing; previously only top-level paths were eligible.

There is no new API, expression, configuration, or query migration. Result semantics remain unchanged for malformed input, duplicate keys, nulls, non-object intermediate values, and rendering failures. With the flag disabled, existing analyzed and optimized plans remain unchanged. Released Spark versions are unaffected.

### How was this patch tested?

The following compilation, suites, and style checks passed on JDK 17:

```bash
build/sbt "catalyst/compile" "catalyst/Test/compile" "sql/Test/compile"
build/sbt "catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.OptimizeJsonExprsSuite"
build/sbt "sql/testOnly org.apache.spark.sql.JsonFunctionsSuite"
build/sbt "hive/Test/testOnly org.apache.spark.sql.configaudit.SparkConfigBindingPolicySuite"
build/sbt "catalyst/scalastyle" "catalyst/Test/scalastyle" "sql/Test/scalastyle"
git diff --check
```

The complete `OptimizeJsonExprsSuite` passed 24 tests, the complete `JsonFunctionsSuite` passed 106 tests, and `SparkConfigBindingPolicySuite` passed 3 tests. The coverage includes nested sharing, both prefix-conflict directions, one-pass grouping of parallel prefix chains, a 2,000-path projection, unsupported paths, the depth limit, default-off plan equivalence, duplicate keys, nulls, malformed and single-quoted JSON, non-object intermediate values, rendering failures, and whole-stage code generation.

The microbenchmark and its committed result files were regenerated with Spark's `Run benchmarks` GitHub Actions workflow for [JDK 17](https://github.com/sunchao/spark/actions/runs/28035861052), [JDK 21](https://github.com/sunchao/spark/actions/runs/28035861792), and [JDK 25](https://github.com/sunchao/spark/actions/runs/28035859413). All three runs passed and committed their JDK-specific results to this branch.

Best JDK 17 times for 200,000 cached rows with 32 fields under a nested `payload` object:

| Nested paths extracted | Shared parsing off | Shared parsing on | Relative speedup |
| ---: | ---: | ---: | ---: |
| 2 | 1,498 ms | 894 ms | 1.7x |
| 4 | 2,969 ms | 1,129 ms | 2.6x |
| 8 | 5,841 ms | 1,378 ms | 4.2x |
| 16 | 11,856 ms | 1,926 ms | 6.2x |

These are GitHub-hosted expression-scaling measurements, not end-to-end production-job results. The complete output, including the JDK 21 and JDK 25 runs, is recorded in the three `SharedJsonParseBenchmark` result files.

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

Closes #56685 from sunchao/codex/SPARK-57626-shared-nested-json-paths.

Lead-authored-by: Chao Sun <chao@openai.com>
Co-authored-by: sunchao <sunchao@users.noreply.github.com>
Signed-off-by: Chao Sun <chao@openai.com>

Author: sunchao

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/json/JsonExpressionEvalUtils.scala

@@ -18,6 +18,7 @@ package org.apache.spark.sql.catalyst.expressions.json
 
 import java.io.{ByteArrayOutputStream, CharArrayWriter, StringWriter}
 
+import scala.collection.mutable
 import scala.util.parsing.combinator.RegexParsers
 
 import com.fasterxml.jackson.core._
@@ -575,24 +576,30 @@ case class GetJsonObjectEvaluator(cachedPath: UTF8String) {
 }
 
 /**
- * Evaluates multiple simple top-level JSON fields in one parse.
+ * Evaluates multiple simple named JSON paths in one parse.
  */
 case class MultiGetJsonObjectEvaluator(
-    fieldNames: Seq[String],
-    fallbackPaths: Seq[UTF8String]) {
+    fallbackPaths: Seq[UTF8String],
+    namedPaths: Seq[Seq[String]]) {
   import SharedFactory._
 
-  require(
-    fieldNames.nonEmpty &&
-      fieldNames.distinct.length == fieldNames.length &&
-      fallbackPaths.length == fieldNames.length)
+  require(fallbackPaths.nonEmpty && namedPaths.length == fallbackPaths.length)
 
   @transient
-  private lazy val fieldToOrdinal: Map[String, Int] = fieldNames.zipWithIndex.toMap
+  private lazy val useTopLevelFastPath: Boolean =
+    namedPaths.forall(_.length == 1) && namedPaths.distinct.length == namedPaths.length
+
+  @transient
+  private lazy val topLevelFieldToOrdinal: Map[String, Int] =
+    namedPaths.zipWithIndex.map { case (path, ordinal) => path.head -> ordinal }.toMap
+
+  @transient
+  private lazy val pathTrie: MultiGetJsonObjectEvaluator.PathTrieNode =
+    MultiGetJsonObjectEvaluator.buildPathTrie(namedPaths)
 
   @transient
   private lazy val nullRow: InternalRow =
-    new GenericInternalRow(Array.ofDim[Any](fieldNames.length))
+    new GenericInternalRow(Array.ofDim[Any](fallbackPaths.length))
 
   @transient
   private lazy val fallbackEvaluators: Seq[GetJsonObjectEvaluator] =
@@ -611,34 +618,18 @@ case class MultiGetJsonObjectEvaluator(
   def evaluate(json: UTF8String): InternalRow = {
     if (json == null) return null
 
-    val values = Array.ofDim[Any](fieldNames.length)
-    val matched = Array.ofDim[Boolean](fieldNames.length)
+    val values = Array.ofDim[Any](fallbackPaths.length)
+    val matched = Array.ofDim[Boolean](fallbackPaths.length)
 
     try {
       val validObject = Utils.tryWithResource(
         CreateJacksonParser.utf8String(jsonFactory, json)) { parser =>
         if (parser.nextToken() != JsonToken.START_OBJECT) {
           false
+        } else if (useTopLevelFastPath) {
+          extractTopLevelObject(parser, values, matched)
         } else {
-          var token = parser.nextToken()
-          while (token != null && token != JsonToken.END_OBJECT) {
-            if (token == JsonToken.FIELD_NAME) {
-              val fieldName = parser.currentName
-              val ordinal = fieldToOrdinal.get(fieldName).filter(!matched(_))
-              val valueToken = parser.nextToken()
-              if (ordinal.nonEmpty && valueToken != JsonToken.VALUE_NULL) {
-                val index = ordinal.get
-                matched(index) = true
-                copyCurrentStructure(parser).foreach(value => values(index) = value)
-              } else {
-                parser.skipChildren()
-              }
-            } else {
-              parser.skipChildren()
-            }
-            token = parser.nextToken()
-          }
-          token == JsonToken.END_OBJECT
+          extractObject(parser, pathTrie, values, matched)
         }
       }
       if (validObject) {
@@ -647,15 +638,88 @@ case class MultiGetJsonObjectEvaluator(
         nullRow
       }
     } catch {
-      // Every simple top-level legacy extraction scans through the root object's closing token,
-      // so a syntax failure makes every sibling null without needing per-path reparsing.
+      // Every simple named legacy extraction scans through the root object's closing token, so a
+      // syntax failure makes every sibling null without needing per-path reparsing.
       case _: JsonParseException => nullRow
-      // A parser-side rendering failure can leave the shared token stream unusable. Reparse each
-      // path with the legacy evaluator so one bad selected value cannot erase sibling results.
+      // A parser-side rendering failure, such as a string-length constraint violation, can leave
+      // the shared token stream unusable. Reparse each path with the legacy evaluator so one bad
+      // selected value cannot erase independent sibling results.
       case _: JsonProcessingException => fallback(json)
     }
   }
 
+  private def extractTopLevelObject(
+      parser: JsonParser,
+      values: Array[Any],
+      matched: Array[Boolean]): Boolean = {
+    var token = parser.nextToken()
+    while (token != null && token != JsonToken.END_OBJECT) {
+      if (token == JsonToken.FIELD_NAME) {
+        val ordinal = topLevelFieldToOrdinal.get(parser.currentName).filter(!matched(_))
+        val valueToken = parser.nextToken()
+        if (ordinal.nonEmpty && valueToken != JsonToken.VALUE_NULL) {
+          val index = ordinal.get
+          matched(index) = true
+          copyCurrentStructure(parser).foreach(value => values(index) = value)
+        } else {
+          parser.skipChildren()
+        }
+      } else {
+        parser.skipChildren()
+      }
+      token = parser.nextToken()
+    }
+    token == JsonToken.END_OBJECT
+  }
+
+  private def extractObject(
+      parser: JsonParser,
+      node: MultiGetJsonObjectEvaluator.PathTrieNode,
+      values: Array[Any],
+      matched: Array[Boolean]): Boolean = {
+    var valid = true
+    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 valueToken = parser.nextToken()
+        if (child.nonEmpty && valueToken != JsonToken.VALUE_NULL) {
+          valid = extractValue(parser, child.get, values, matched)
+        } else {
+          parser.skipChildren()
+        }
+      } else {
+        parser.skipChildren()
+      }
+      if (valid) {
+        token = parser.nextToken()
+      }
+    }
+    valid && token == JsonToken.END_OBJECT
+  }
+
+  private def extractValue(
+      parser: JsonParser,
+      node: MultiGetJsonObjectEvaluator.PathTrieNode,
+      values: Array[Any],
+      matched: Array[Boolean]): Boolean = {
+    // Optimizer-generated paths are deduplicated. Multiple ordinals defensively support
+    // directly constructed internal expressions with duplicate paths.
+    if (node.terminalOrdinals.nonEmpty) {
+      node.terminalOrdinals.foreach { ordinal => matched(ordinal) = true }
+      val value = copyCurrentStructure(parser)
+      value.foreach { result =>
+        node.terminalOrdinals.foreach { ordinal => values(ordinal) = result }
+      }
+      true
+    } else if (parser.currentToken == JsonToken.START_OBJECT) {
+      extractObject(parser, node, values, matched)
+    } else {
+      parser.skipChildren()
+      true
+    }
+  }
+
   private def copyCurrentStructure(parser: JsonParser): Option[UTF8String] = {
     outputBuffer.reset()
     var renderingFailed = false
@@ -726,3 +790,43 @@ case class MultiGetJsonObjectEvaluator(
     if (renderingFailed) None else Some(UTF8String.fromBytes(outputBuffer.toByteArray))
   }
 }
+
+object MultiGetJsonObjectEvaluator {
+  private final class MutablePathTrieNode {
+    val terminalOrdinals: mutable.ArrayBuffer[Int] = mutable.ArrayBuffer.empty
+    val children: mutable.LinkedHashMap[String, MutablePathTrieNode] = mutable.LinkedHashMap.empty
+
+    def freeze(): PathTrieNode = {
+      require(
+        terminalOrdinals.isEmpty || children.isEmpty,
+        "Shared JSON paths must not be prefixes of one another")
+      val frozenChildren = children.iterator.map { case (name, child) =>
+        name -> child.freeze()
+      }.toMap
+      val ordinals = (terminalOrdinals.iterator ++
+        frozenChildren.valuesIterator.flatMap(_.descendantOrdinals.iterator)).toArray
+      PathTrieNode(terminalOrdinals.toArray, frozenChildren, ordinals)
+    }
+  }
+
+  private case class PathTrieNode(
+      terminalOrdinals: Array[Int],
+      children: Map[String, PathTrieNode],
+      descendantOrdinals: Array[Int]) {
+    def hasUnmatched(matched: Array[Boolean]): Boolean = {
+      descendantOrdinals.exists(index => !matched(index))
+    }
+  }
+
+  private def buildPathTrie(paths: Seq[Seq[String]]): PathTrieNode = {
+    val root = new MutablePathTrieNode
+    paths.zipWithIndex.foreach { case (path, ordinal) =>
+      var node = root
+      path.foreach { fieldName =>
+        node = node.children.getOrElseUpdate(fieldName, new MutablePathTrieNode)
+      }
+      node.terminalOrdinals += ordinal
+    }
+    root.freeze()
+  }
+}

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/jsonExpressions.scala

@@ -142,10 +142,16 @@ case class GetJsonObject(json: Expression, path: Expression)
 }
 
 object GetJsonObject {
-  private[sql] def simpleTopLevelField(path: UTF8String): Option[String] = {
+  import PathInstruction._
+
+  private[sql] def simpleNamedPath(path: UTF8String): Option[Seq[String]] = {
     try {
-      Option(path).flatMap(value => JsonPathParser.parse(value.toString)).collect {
-        case List(PathInstruction.Key, PathInstruction.Named(fieldName)) => fieldName
+      Option(path).flatMap(value => JsonPathParser.parse(value.toString)).flatMap { instructions =>
+        val names = instructions.grouped(2).map {
+          case List(Key, Named(fieldName)) => Some(fieldName)
+          case _ => None
+        }.toSeq
+        if (names.nonEmpty && names.forall(_.isDefined)) Some(names.flatten) else None
       }
     } catch {
       // Numeric subscripts are parsed as Long and can overflow before the parser returns None.
@@ -155,28 +161,25 @@ object GetJsonObject {
 }
 
 /**
- * Extracts multiple simple top-level fields from a JSON string in one parse. This is an internal
- * expression used to share sibling [[GetJsonObject]] expressions; unsupported JSON paths remain
- * as independent GetJsonObject expressions.
+ * 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
+ * prefix-conflicting JSON paths remain as independent GetJsonObject expressions.
  */
 case class MultiGetJsonObject(
     json: Expression,
-    fieldNames: Seq[String],
     fallbackPaths: Seq[String])
   extends UnaryExpression
   with ExpectsInputTypes {
 
-  require(
-    fieldNames.nonEmpty &&
-      fieldNames.distinct.length == fieldNames.length &&
-      fallbackPaths.length == fieldNames.length)
+  // OptimizeCsvJsonExprs caps shared path depth to keep evaluator recursion stack-safe.
+  require(fallbackPaths.nonEmpty)
 
   override def child: Expression = json
 
   override def inputTypes: Seq[AbstractDataType] =
     Seq(StringTypeWithCollation(supportsTrimCollation = true))
 
-  override lazy val dataType: DataType = StructType(fieldNames.indices.map { index =>
+  override lazy val dataType: DataType = StructType(fallbackPaths.indices.map { index =>
     StructField(s"_$index", StringType, nullable = true)
   })
 
@@ -189,10 +192,17 @@ 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 {
+      throw new IllegalArgumentException(s"Unsupported shared JSON path: $path")
+    }
+  }
+
   @transient
   private lazy val evaluator = MultiGetJsonObjectEvaluator(
-    fieldNames,
-    fallbackPaths.map(UTF8String.fromString))
+    fallbackPaths.map(UTF8String.fromString),
+    namedPaths)
 
   override def eval(input: InternalRow): Any = {
     evaluator.evaluate(json.eval(input).asInstanceOf[UTF8String])