From 8919c92bb8e0369332332cfb496f4b6bb8fbc880 Mon Sep 17 00:00:00 2001 From: Allen Xu Date: Fri, 28 Aug 2026 17:42:26 +0800 Subject: [PATCH 1/5] fix: match get_json_object runtime path semantics Signed-off-by: Allen Xu --- .../src/main/python/get_json_test.py | 37 +++++- .../spark/rapids/GpuGetJsonObject.scala | 118 +++++++++++++----- .../shims/GetJsonObjectRuntimeSemantics.scala | 33 +++++ .../rapids/shims/GetJsonObjectShim.scala | 25 ++-- .../rapids/shims/GetJsonObjectShim.scala | 12 +- .../spark/rapids/JsonPathParserSuite.scala | 85 +++++++++++++ 6 files changed, 252 insertions(+), 58 deletions(-) create mode 100644 sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala create mode 100644 tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala diff --git a/integration_tests/src/main/python/get_json_test.py b/integration_tests/src/main/python/get_json_test.py index 2b4218692a6..8fc37df0955 100644 --- a/integration_tests/src/main/python/get_json_test.py +++ b/integration_tests/src/main/python/get_json_test.py @@ -14,12 +14,13 @@ import pytest -from asserts import assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect, with_gpu_session +from asserts import (assert_cpu_and_gpu_are_equal_collect_with_capture, + assert_gpu_and_cpu_are_equal_collect, assert_gpu_fallback_collect, + with_gpu_session) from data_gen import * from pyspark.sql.types import * from marks import * from spark_init_internal import spark_version -from conftest import is_dataproc_runtime, is_dataproc_serverless_runtime from spark_session import is_before_spark_400, is_databricks113_or_later, is_databricks_runtime def mk_json_str_gen(pattern): @@ -123,15 +124,39 @@ def test_get_json_object_normalize_non_string_output(): f.col('jsonStr'), f.get_json_object('jsonStr', '$'))) -@pytest.mark.xfail(condition=is_dataproc_runtime() or is_dataproc_serverless_runtime(), - reason="https://github.com/NVIDIA/spark-rapids/issues/14290") def test_get_json_object_quoted_question(): schema = StructType([StructField("jsonStr", StringType())]) data = [[r'{"?":"QUESTION"}']] - assert_gpu_and_cpu_are_equal_collect( + assert_cpu_and_gpu_are_equal_collect_with_capture( + lambda spark: spark.createDataFrame(data,schema=schema).select( + f.get_json_object('jsonStr',r'''$['?']''').alias('question')), + exist_classes='GpuGetJsonObject') + + +def test_multi_get_json_object_quoted_question(): + schema = StructType([StructField("jsonStr", StringType())]) + data = [[r'{"?":"QUESTION","a?b":"EMBEDDED","outer":{"?":"NESTED"}}']] + + assert_cpu_and_gpu_are_equal_collect_with_capture( lambda spark: spark.createDataFrame(data,schema=schema).select( - f.get_json_object('jsonStr',r'''$['?']''').alias('question'))) + f.get_json_object('jsonStr',r'''$['?']''').alias('question'), + f.get_json_object('jsonStr',r'''$['a?b']''').alias('embedded'), + f.get_json_object('jsonStr',r'''$.outer['?']''').alias('nested')), + exist_classes='GpuProjectExec,GpuGetJsonObject') + + +def test_multi_get_json_object_all_invalid_paths(): + schema = StructType([StructField("jsonStr", StringType())]) + data = [['{"a":"A"}']] + + assert_cpu_and_gpu_are_equal_collect_with_capture( + lambda spark: spark.createDataFrame(data,schema=schema).selectExpr( + 'get_json_object(jsonStr, CAST(NULL AS STRING)) AS null_path', + 'get_json_object(jsonStr, "$[") AS malformed_path', + 'get_json_object(jsonStr, "not_a_path") AS missing_root'), + exist_classes='GpuProjectExec,GpuGetJsonObject') + def test_get_json_object_escaped_string_data(): schema = StructType([StructField("jsonStr", StringType())]) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala index bd6760f6765..82d1cd7abc3 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala @@ -68,10 +68,13 @@ object JsonPathParser extends RegexParsers { Subscript :: operand :: Nil } + private val legacyNamedPartRegexp = "[^\\'\\?]+" + private val fixedNamedPartRegexp = "[^\\']+" + // parse `.name` or `['name']` child expressions - def named: Parser[List[PathInstruction]] = + private def named(partRegexpInNamed: String): Parser[List[PathInstruction]] = for { - name <- '.' ~> "[^\\.\\[]+".r | "['" ~> GetJsonObjectShim.partRegexpInNamed.r <~ "']" + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> partRegexpInNamed.r <~ "']" } yield { Key :: Named(name) :: Nil } @@ -80,16 +83,25 @@ object JsonPathParser extends RegexParsers { def wildcard: Parser[List[PathInstruction]] = (".*" | "['*']") ^^^ List(Wildcard) - def node: Parser[List[PathInstruction]] = + private def node(partRegexpInNamed: String): Parser[List[PathInstruction]] = wildcard | - named | + named(partRegexpInNamed) | subscript - val expression: Parser[List[PathInstruction]] = { - phrase(root ~> rep(node) ^^ (x => x.flatten)) + private def pathExpression(partRegexpInNamed: String): Parser[List[PathInstruction]] = { + phrase(root ~> rep(node(partRegexpInNamed)) ^^ (x => x.flatten)) } - def parse(str: String): Option[List[PathInstruction]] = { + private lazy val legacyPathExpression = pathExpression(legacyNamedPartRegexp) + private lazy val fixedPathExpression = pathExpression(fixedNamedPartRegexp) + + def parse(str: String, allowQuestionMarkInQuotedName: Boolean): + Option[List[PathInstruction]] = { + val expression = if (allowQuestionMarkInQuotedName) { + fixedPathExpression + } else { + legacyPathExpression + } this.parseAll(expression, str) match { case Success(result, _) => Some(result) @@ -156,6 +168,29 @@ object JsonPathParser extends RegexParsers { } } +object GpuGetJsonObjectMeta { + private[rapids] val UNKNOWN_QUESTION_MARK_SUPPORT_REASON = + "Could not determine whether this Spark runtime accepts question marks in quoted " + + "get_json_object paths" + + private[rapids] def parseLiteralPath( + value: Any, + allowQuestionMarkInQuotedName: Boolean): Option[List[PathInstruction]] = { + Option(value).map(_.asInstanceOf[UTF8String].toString).flatMap { path => + JsonPathParser.parse(path, allowQuestionMarkInQuotedName) + } + } + + private[rapids] def unsupportedReason( + quotedQuestionMarkSupport: Option[Boolean]): Option[String] = { + if (quotedQuestionMarkSupport.isDefined) { + None + } else { + Some(UNKNOWN_QUESTION_MARK_SUPPORT_REASON) + } + } +} + class GpuGetJsonObjectMeta( expr: GetJsonObject, conf: RapidsConf, @@ -163,22 +198,31 @@ class GpuGetJsonObjectMeta( rule: DataFromReplacementRule ) extends BinaryExprMeta[GetJsonObject](expr, conf, parent, rule) { + private val quotedQuestionMarkSupport = GetJsonObjectShim.quotedQuestionMarkSupport + override def tagExprForGpu(): Unit = { - val lit = GpuOverrides.extractLit(expr.right) - lit.foreach { l => - val instructions = JsonPathParser.parse(l.value.asInstanceOf[UTF8String].toString) - val updated = instructions.map(JsonPathParser.filterInstructionsForJni) - if (updated.exists(JsonPathParser.fallbackCheck)) { - willNotWorkOnGpu(s"get_json_object on GPU does not support more " + - s"than ${JsonPathParser.MAX_PATH_DEPTH} nested paths." + - instructions.map(i => s" (Found ${i.length})").getOrElse("")) + GpuGetJsonObjectMeta.unsupportedReason(quotedQuestionMarkSupport).foreach(willNotWorkOnGpu) + quotedQuestionMarkSupport.foreach { allowQuestionMark => + val lit = GpuOverrides.extractLit(expr.right) + lit.foreach { l => + val instructions = + GpuGetJsonObjectMeta.parseLiteralPath(l.value, allowQuestionMark) + val updated = instructions.map(JsonPathParser.filterInstructionsForJni) + if (updated.exists(JsonPathParser.fallbackCheck)) { + willNotWorkOnGpu(s"get_json_object on GPU does not support more " + + s"than ${JsonPathParser.MAX_PATH_DEPTH} nested paths." + + instructions.map(i => s" (Found ${i.length})").getOrElse("")) + } } } } override def convertToGpu(lhs: Expression, rhs: Expression): GpuExpression = { + val allowQuestionMark = quotedQuestionMarkSupport.getOrElse { + throw new IllegalStateException(GpuGetJsonObjectMeta.UNKNOWN_QUESTION_MARK_SUPPORT_REASON) + } GpuGetJsonObject(lhs, rhs)( - conf.testGetJsonObjectSavePath, conf.testGetJsonObjectSaveRows) + conf.testGetJsonObjectSavePath, conf.testGetJsonObjectSaveRows, allowQuestionMark) } } @@ -216,13 +260,15 @@ case class GpuMultiGetJsonObject(json: Expression, val validPaths = validPathsWithIndexes.map(_._1) withResource(new Array[ColumnVector](validPaths.length)) { validPathColumns => withResource(json.columnarEval(batch)) { input => - // Last argument -1 indicates to use automatically calculated parallelism - withResource(JSONUtils.getJsonObjectMultiplePaths(input.getBase, - java.util.Arrays.asList(validPaths: _*), 4 * targetBatchSize, - -1)) { chunkedResult => - chunkedResult.foreach { cr => - validPathColumns(validPathsIndex) = cr.incRefCount() - validPathsIndex += 1 + if (validPaths.nonEmpty) { + // Last argument -1 indicates to use automatically calculated parallelism + withResource(JSONUtils.getJsonObjectMultiplePaths(input.getBase, + java.util.Arrays.asList(validPaths: _*), 4 * targetBatchSize, + -1)) { chunkedResult => + chunkedResult.foreach { cr => + validPathColumns(validPathsIndex) = cr.incRefCount() + validPathsIndex += 1 + } } } @@ -311,7 +357,8 @@ class GetJsonObjectCombiner(private val exp: GpuGetJsonObject) extends GpuExpres case u: UTF8String => u.toString case _ => null.asInstanceOf[String] } - val pathInstructions = parseJsonPath(str) + val pathInstructions = + parseJsonPath(str, e.allowQuestionMarkInQuotedName) if (hasSeparateWildcard(pathInstructions)) { // If has separate wildcard path, should return all nulls None @@ -336,18 +383,22 @@ class GetJsonObjectCombiner(private val exp: GpuGetJsonObject) extends GpuExpres } object GpuGetJsonObject { - def parseJsonPath(path: GpuScalar): Option[List[PathInstruction]] = { + def parseJsonPath( + path: GpuScalar, + allowQuestionMarkInQuotedName: Boolean): Option[List[PathInstruction]] = { if (path.isValid) { val pathStr = path.getValue.toString - JsonPathParser.parse(pathStr) + JsonPathParser.parse(pathStr, allowQuestionMarkInQuotedName) } else { None } } - def parseJsonPath(pathStr: String): Option[List[PathInstruction]] = { + def parseJsonPath( + pathStr: String, + allowQuestionMarkInQuotedName: Boolean): Option[List[PathInstruction]] = { if (pathStr != null) { - JsonPathParser.parse(pathStr) + JsonPathParser.parse(pathStr, allowQuestionMarkInQuotedName) } else { None } @@ -392,7 +443,8 @@ case class GpuGetJsonObject( json: Expression, path: Expression)( val savePathForVerify: Option[String], - val saveRowsForVerify: Int) + val saveRowsForVerify: Int, + val allowQuestionMarkInQuotedName: Boolean) extends GpuBinaryExpressionArgsAnyScalar with ExpectsInputTypes with GpuCombinable { @@ -405,8 +457,10 @@ case class GpuGetJsonObject( } val seed = System.nanoTime() - override def otherCopyArgs: Seq[AnyRef] = Seq(savePathForVerify, - saveRowsForVerify.asInstanceOf[java.lang.Integer]) + override def otherCopyArgs: Seq[AnyRef] = Seq( + savePathForVerify, + saveRowsForVerify.asInstanceOf[java.lang.Integer], + allowQuestionMarkInQuotedName.asInstanceOf[java.lang.Boolean]) override def left: Expression = json override def right: Expression = path @@ -420,7 +474,7 @@ case class GpuGetJsonObject( override def doColumnar(lhs: GpuColumnVector, rhs: GpuScalar): ColumnVector = { val fromGpu = cachedInstructions.getOrElse { - val pathInstructions = parseJsonPath(rhs) + val pathInstructions = parseJsonPath(rhs, allowQuestionMarkInQuotedName) val checkedPathInstructions = if (hasSeparateWildcard(pathInstructions)) { // If has separate wildcard path, should return all nulls None diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala new file mode 100644 index 00000000000..12afe227761 --- /dev/null +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala @@ -0,0 +1,33 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed 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 com.nvidia.spark.rapids.shims + +import scala.util.Try + +import org.apache.spark.unsafe.types.UTF8String + +private[rapids] object GetJsonObjectRuntimeSemantics { + private val ExpectedQuestionMarkValue = "QUESTION" + + def classifyQuotedQuestionMarkResult(result: => Any): Option[Boolean] = { + Try(result).toOption match { + case Some(null) => Some(false) + case Some(value: UTF8String) if value.toString == ExpectedQuestionMarkValue => Some(true) + case _ => None + } + } +} diff --git a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index c8ab49e07f0..44475710b50 100644 --- a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -41,17 +41,22 @@ spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.shims +import org.apache.spark.sql.catalyst.expressions.{GetJsonObject, Literal} +import org.apache.spark.sql.types.StringType +import org.apache.spark.unsafe.types.UTF8String + object GetJsonObjectShim { + private lazy val runtimeQuotedQuestionMarkSupport: Option[Boolean] = { + val json = Literal.create(UTF8String.fromString("""{"?":"QUESTION"}"""), StringType) + val path = Literal.create(UTF8String.fromString("$['?']"), StringType) + GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult { + GetJsonObject(json, path).eval(null) + } + } + /** - * Return a shim string for a part in named Regexp. - * For Spark versions before 400, named Regexp is: - * name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\'\\?]+".r <~ "']" - * For Spark versions 400 and 400+, named Regexp is: - * name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" - * This is the shim to distinct "[^\\'\\?]+" and "[^\\']+" - * - * "[^\\'\\?]+" : One or more chars which are not: ' or ? - * "[^\\']+" : One or more chars which are not: ' + * Detect whether this Spark runtime includes SPARK-46761 semantics. Some vendors backported the + * fix without changing the upstream Spark version, so a version check is not sufficient. */ - def partRegexpInNamed: String = "[^\\'\\?]+" + def quotedQuestionMarkSupport: Option[Boolean] = runtimeQuotedQuestionMarkSupport } diff --git a/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index efb91ce6c2c..61e158e4f45 100644 --- a/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -32,15 +32,7 @@ package com.nvidia.spark.rapids.shims object GetJsonObjectShim { /** - * Return a shim string for a part in named Regexp. - * For Spark versions before 400, named Regexp is: - * name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\'\\?]+".r <~ "']" - * For Spark versions 400 and 400+, named Regexp is: - * name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" - * This is the shim to distinct "[^\\'\\?]+" and "[^\\']+" - * - * "[^\\'\\?]+" : One or more chars which are not: ' or ? - * "[^\\']+" : One or more chars which are not: ' + * Spark 4 includes SPARK-46761, which accepts question marks in quoted path names. */ - def partRegexpInNamed: String = "[^\\']+" + def quotedQuestionMarkSupport: Option[Boolean] = Some(true) } diff --git a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala new file mode 100644 index 00000000000..87e3fb5806e --- /dev/null +++ b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala @@ -0,0 +1,85 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. + * + * Licensed 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 com.nvidia.spark.rapids + +import com.nvidia.spark.rapids.PathInstruction.{Key, Named} +import com.nvidia.spark.rapids.shims.{GetJsonObjectRuntimeSemantics, GetJsonObjectShim} +import org.scalatest.funsuite.AnyFunSuite + +import org.apache.spark.sql.catalyst.expressions.{GetJsonObject, Literal} +import org.apache.spark.sql.types.StringType +import org.apache.spark.unsafe.types.UTF8String + +class JsonPathParserSuite extends AnyFunSuite { + private val questionMarkPath = List(Key, Named("?")) + + test("quoted question marks follow the selected parser dialect") { + val fixedCases = Seq( + "$['?']" -> List(Key, Named("?")), + "$['a?b']" -> List(Key, Named("a?b")), + "$.outer['?']" -> List(Key, Named("outer"), Key, Named("?"))) + + fixedCases.foreach { case (path, expected) => + assert(JsonPathParser.parse(path, allowQuestionMarkInQuotedName = true) === Some(expected)) + assert(JsonPathParser.parse(path, allowQuestionMarkInQuotedName = false).isEmpty) + } + } + + test("unquoted and malformed paths are independent of the selected parser dialect") { + Seq(true, false).foreach { allowQuestionMark => + assert(JsonPathParser.parse("$.?", allowQuestionMark) === Some(questionMarkPath)) + assert(JsonPathParser.parse("$['ordinary']", allowQuestionMark) === + Some(List(Key, Named("ordinary")))) + assert(JsonPathParser.parse("$['']", allowQuestionMark).isEmpty) + assert(JsonPathParser.parse("$['unterminated]", allowQuestionMark).isEmpty) + } + } + + test("quoted question mark probe result is classified fail closed") { + assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + UTF8String.fromString("QUESTION")) === Some(true)) + assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult(null) === Some(false)) + assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + UTF8String.fromString("unexpected")).isEmpty) + assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + throw new RuntimeException("probe failed")).isEmpty) + } + + test("unknown runtime semantics require CPU fallback") { + assert(GpuGetJsonObjectMeta.unsupportedReason(None) === + Some(GpuGetJsonObjectMeta.UNKNOWN_QUESTION_MARK_SUPPORT_REASON)) + assert(GpuGetJsonObjectMeta.unsupportedReason(Some(true)).isEmpty) + assert(GpuGetJsonObjectMeta.unsupportedReason(Some(false)).isEmpty) + } + + test("literal path parsing handles null without changing the selected dialect") { + assert(GpuGetJsonObjectMeta.parseLiteralPath( + null, allowQuestionMarkInQuotedName = true).isEmpty) + assert(GpuGetJsonObjectMeta.parseLiteralPath( + null, allowQuestionMarkInQuotedName = false).isEmpty) + } + + test("shim capability matches the active Spark CPU expression") { + val json = Literal.create(UTF8String.fromString("""{"?":"QUESTION"}"""), StringType) + val path = Literal.create(UTF8String.fromString("$['?']"), StringType) + val cpuResult = GetJsonObject(json, path).eval(null) + val expected = GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult(cpuResult) + + assert(expected.isDefined) + assert(GetJsonObjectShim.quotedQuestionMarkSupport === expected) + } +} From 9a7812b817d6706e436b87f893f3a5cf1b65b694 Mon Sep 17 00:00:00 2001 From: Allen Xu Date: Tue, 1 Sep 2026 16:29:03 +0800 Subject: [PATCH 2/5] docs: clarify get_json_object runtime semantics Signed-off-by: Allen Xu --- .../scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala | 2 ++ .../rapids/shims/GetJsonObjectRuntimeSemantics.scala | 8 ++++++++ 2 files changed, 10 insertions(+) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala index 82d1cd7abc3..628404e0822 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala @@ -68,6 +68,8 @@ object JsonPathParser extends RegexParsers { Subscript :: operand :: Nil } + // SPARK-46761 made '?' valid in bracket-quoted names. Some pre-Spark-4 vendor runtimes + // backported that change, so retain both variants and select one from the runtime probe. private val legacyNamedPartRegexp = "[^\\'\\?]+" private val fixedNamedPartRegexp = "[^\\']+" diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala index 12afe227761..5451b983712 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala @@ -23,6 +23,14 @@ import org.apache.spark.unsafe.types.UTF8String private[rapids] object GetJsonObjectRuntimeSemantics { private val ExpectedQuestionMarkValue = "QUESTION" + /** + * Classify the result of evaluating the quoted-question-mark probe with the active CPU + * Catalyst implementation. + * + * @param result lazily evaluated probe result; evaluation failures are treated as unknown + * @return `Some(true)` for the SPARK-46761 result, `Some(false)` for the legacy null result, + * or `None` for an unexpected value or evaluation failure + */ def classifyQuotedQuestionMarkResult(result: => Any): Option[Boolean] = { Try(result).toOption match { case Some(null) => Some(false) From 802b94a04a0e854de88fd7c135b1ffcb6ee8a56b Mon Sep 17 00:00:00 2001 From: Allen Xu Date: Wed, 2 Sep 2026 12:33:31 +0800 Subject: [PATCH 3/5] Address JSON runtime semantics review Signed-off-by: Allen Xu --- .../shims/GetJsonObjectRuntimeSemantics.scala | 7 +++---- .../spark/rapids/shims/GetJsonObjectShim.scala | 8 ++++---- .../spark/rapids/JsonPathParserSuite.scala | 17 +++++++++++------ 3 files changed, 18 insertions(+), 14 deletions(-) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala index 5451b983712..8da7269fc0e 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala @@ -21,20 +21,19 @@ import scala.util.Try import org.apache.spark.unsafe.types.UTF8String private[rapids] object GetJsonObjectRuntimeSemantics { - private val ExpectedQuestionMarkValue = "QUESTION" - /** * Classify the result of evaluating the quoted-question-mark probe with the active CPU * Catalyst implementation. * * @param result lazily evaluated probe result; evaluation failures are treated as unknown + * @param expected expected string value when quoted question marks are supported * @return `Some(true)` for the SPARK-46761 result, `Some(false)` for the legacy null result, * or `None` for an unexpected value or evaluation failure */ - def classifyQuotedQuestionMarkResult(result: => Any): Option[Boolean] = { + def classifyQuotedQuestionMarkResult(result: => Any, expected: String): Option[Boolean] = { Try(result).toOption match { case Some(null) => Some(false) - case Some(value: UTF8String) if value.toString == ExpectedQuestionMarkValue => Some(true) + case Some(value: UTF8String) if value.toString == expected => Some(true) case _ => None } } diff --git a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index 44475710b50..0bfc3d9994d 100644 --- a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -47,11 +47,11 @@ import org.apache.spark.unsafe.types.UTF8String object GetJsonObjectShim { private lazy val runtimeQuotedQuestionMarkSupport: Option[Boolean] = { - val json = Literal.create(UTF8String.fromString("""{"?":"QUESTION"}"""), StringType) + val expected = "QUESTION" + val json = Literal.create(UTF8String.fromString(s"""{"?":"$expected"}"""), StringType) val path = Literal.create(UTF8String.fromString("$['?']"), StringType) - GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult { - GetJsonObject(json, path).eval(null) - } + GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + GetJsonObject(json, path).eval(null), expected) } /** diff --git a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala index 87e3fb5806e..fca2bfdfd3a 100644 --- a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala +++ b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala @@ -50,13 +50,15 @@ class JsonPathParserSuite extends AnyFunSuite { } test("quoted question mark probe result is classified fail closed") { + val expected = "QUESTION" assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( - UTF8String.fromString("QUESTION")) === Some(true)) - assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult(null) === Some(false)) + UTF8String.fromString(expected), expected) === Some(true)) assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( - UTF8String.fromString("unexpected")).isEmpty) + null, expected) === Some(false)) assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( - throw new RuntimeException("probe failed")).isEmpty) + UTF8String.fromString("unexpected"), expected).isEmpty) + assert(GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + throw new RuntimeException("probe failed"), expected).isEmpty) } test("unknown runtime semantics require CPU fallback") { @@ -74,10 +76,13 @@ class JsonPathParserSuite extends AnyFunSuite { } test("shim capability matches the active Spark CPU expression") { - val json = Literal.create(UTF8String.fromString("""{"?":"QUESTION"}"""), StringType) + val expectedValue = "QUESTION" + val json = Literal.create( + UTF8String.fromString(s"""{"?":"$expectedValue"}"""), StringType) val path = Literal.create(UTF8String.fromString("$['?']"), StringType) val cpuResult = GetJsonObject(json, path).eval(null) - val expected = GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult(cpuResult) + val expected = GetJsonObjectRuntimeSemantics.classifyQuotedQuestionMarkResult( + cpuResult, expectedValue) assert(expected.isDefined) assert(GetJsonObjectShim.quotedQuestionMarkSupport === expected) From 034fa1f4c563f9b46d9bc10fb2568936c05d2fd4 Mon Sep 17 00:00:00 2001 From: Allen Xu Date: Thu, 3 Sep 2026 11:22:20 +0800 Subject: [PATCH 4/5] Keep JSON path parsers independently diffable Keep the legacy Spark 3 and SPARK-46761 parser implementations as independent shim-local copies so each remains directly diffable with Apache Spark. Select the known Dataproc backport from spark.dataproc.engine without probing Catalyst behavior. Performance: This changes only driver-side literal JSON path parsing. Parser selection is cached once, and no executor row or batch loop is affected. Signed-off-by: Allen Xu --- .../spark/rapids/GpuGetJsonObject.scala | 60 +------- .../rapids/shims/GetJsonObjectShim.scala | 129 ++++++++++++++++-- .../rapids/shims/GetJsonObjectShim.scala | 62 ++++++++- .../spark/rapids/JsonPathParserSuite.scala | 23 ++-- 4 files changed, 190 insertions(+), 84 deletions(-) diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala index 0017b7df4a9..d08bb6e5957 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala @@ -17,7 +17,6 @@ package com.nvidia.spark.rapids import scala.collection.mutable -import scala.util.parsing.combinator.RegexParsers import ai.rapids.cudf.ColumnVector import com.nvidia.spark.rapids.Arm.{closeOnExcept, withResource} @@ -46,7 +45,7 @@ object PathInstruction { case class Named(name: String) extends PathInstruction } -object JsonPathParser extends RegexParsers { +object JsonPathParser { // Mirrors JSONUtils.MAX_PATH_DEPTH from spark-rapids-jni (get_json_object.hpp). // Duplicated here to avoid triggering JNI native library loading during // Driver-side plan conversion (see github.com/NVIDIA/spark-rapids/issues/14184). @@ -54,63 +53,8 @@ object JsonPathParser extends RegexParsers { import PathInstruction._ - def root: Parser[Char] = '$' - - def long: Parser[Long] = "\\d+".r ^? { - case x => x.toLong - } - - // parse `[*]` and `[123]` subscripts - def subscript: Parser[List[PathInstruction]] = - for { - operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' - } yield { - Subscript :: operand :: Nil - } - - // parse `.name` or `['name']` child expressions - private def named(partRegexpInNamed: String): Parser[List[PathInstruction]] = - for { - name <- '.' ~> "[^\\.\\[]+".r | "['" ~> partRegexpInNamed.r <~ "']" - } yield { - Key :: Named(name) :: Nil - } - - // child wildcards: `..`, `.*` or `['*']` - def wildcard: Parser[List[PathInstruction]] = - (".*" | "['*']") ^^^ List(Wildcard) - - private def node(partRegexpInNamed: String): Parser[List[PathInstruction]] = - wildcard | - named(partRegexpInNamed) | - subscript - - private def pathExpression(partRegexpInNamed: String): Parser[List[PathInstruction]] = { - phrase(root ~> rep(node(partRegexpInNamed)) ^^ (x => x.flatten)) - } - - private lazy val expression = pathExpression(GetJsonObjectShim.partRegexpInNamed) - - private def parseExpression( - expression: Parser[List[PathInstruction]], - str: String): Option[List[PathInstruction]] = { - this.parseAll(expression, str) match { - case Success(result, _) => - Some(result) - - case _ => - None - } - } - - private[rapids] def parseWithNamedPartRegexp( - str: String, - partRegexpInNamed: String): Option[List[PathInstruction]] = { - parseExpression(pathExpression(partRegexpInNamed), str) - } - def parse(str: String): Option[List[PathInstruction]] = { - parseExpression(expression, str) + GetJsonObjectShim.parse(str) } def filterInstructionsForJni(instructions: List[PathInstruction]): List[PathInstruction] = diff --git a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index 311b0d09190..322f72ac2ac 100644 --- a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -41,18 +41,125 @@ spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.shims +import scala.util.parsing.combinator.RegexParsers + +import com.nvidia.spark.rapids.PathInstruction + import org.apache.spark.{SparkConf, SparkEnv} object GetJsonObjectShim { private val DATAPROC_ENGINE_KEY = "spark.dataproc.engine" - private val LEGACY_NAMED_PART_REGEXP = "[^\\'\\?]+" - private val FIXED_NAMED_PART_REGEXP = "[^\\']+" - private[rapids] def partRegexpInNamed(conf: SparkConf): String = { - if (conf.contains(DATAPROC_ENGINE_KEY)) { - FIXED_NAMED_PART_REGEXP + // Copied from Apache Spark 3.5.3 JsonPathParser in jsonExpressions.scala. + private object LegacyJsonPathParser extends RegexParsers { + import com.nvidia.spark.rapids.PathInstruction._ + + def root: Parser[Char] = '$' + + def long: Parser[Long] = "\\d+".r ^? { + case x => x.toLong + } + + // parse `[*]` and `[123]` subscripts + def subscript: Parser[List[PathInstruction]] = + for { + operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' + } yield { + Subscript :: operand :: Nil + } + + // parse `.name` or `['name']` child expressions + def named: Parser[List[PathInstruction]] = + for { + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\'\\?]+".r <~ "']" + } yield { + Key :: Named(name) :: Nil + } + + // child wildcards: `..`, `.*` or `['*']` + def wildcard: Parser[List[PathInstruction]] = + (".*" | "['*']") ^^^ List(Wildcard) + + def node: Parser[List[PathInstruction]] = + wildcard | + named | + subscript + + val expression: Parser[List[PathInstruction]] = { + phrase(root ~> rep(node) ^^ (x => x.flatten)) + } + + def parse(str: String): Option[List[PathInstruction]] = { + this.parseAll(expression, str) match { + case Success(result, _) => + Some(result) + + case _ => + None + } + } + } + + // Copied from Apache Spark 4.0.0 JsonPathParser after SPARK-46761. + private object FixedJsonPathParser extends RegexParsers { + import com.nvidia.spark.rapids.PathInstruction._ + + def root: Parser[Char] = '$' + + def long: Parser[Long] = "\\d+".r ^? { + case x => x.toLong + } + + // parse `[*]` and `[123]` subscripts + def subscript: Parser[List[PathInstruction]] = + for { + operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' + } yield { + Subscript :: operand :: Nil + } + + // parse `.name` or `['name']` child expressions + def named: Parser[List[PathInstruction]] = + for { + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" + } yield { + Key :: Named(name) :: Nil + } + + // child wildcards: `..`, `.*` or `['*']` + def wildcard: Parser[List[PathInstruction]] = + (".*" | "['*']") ^^^ List(Wildcard) + + def node: Parser[List[PathInstruction]] = + wildcard | + named | + subscript + + val expression: Parser[List[PathInstruction]] = { + phrase(root ~> rep(node) ^^ (x => x.flatten)) + } + + def parse(str: String): Option[List[PathInstruction]] = { + this.parseAll(expression, str) match { + case Success(result, _) => + Some(result) + + case _ => + None + } + } + } + + private def useFixedParser(conf: SparkConf): Boolean = conf.contains(DATAPROC_ENGINE_KEY) + + private lazy val activeParserUsesFixedSemantics = + Option(SparkEnv.get).exists(env => useFixedParser(env.conf)) + + private[rapids] def parse(str: String, conf: SparkConf): Option[List[PathInstruction]] = { + if (useFixedParser(conf)) { + FixedJsonPathParser.parse(str) } else { - LEGACY_NAMED_PART_REGEXP + LegacyJsonPathParser.parse(str) } } @@ -60,9 +167,11 @@ object GetJsonObjectShim { * Spark 3.x uses the legacy quoted-name parser, except on Dataproc classic and Serverless * runtimes. Dataproc sets `spark.dataproc.engine` and has backported SPARK-46761. */ - def partRegexpInNamed: String = { - Option(SparkEnv.get) - .map(env => partRegexpInNamed(env.conf)) - .getOrElse(LEGACY_NAMED_PART_REGEXP) + def parse(str: String): Option[List[PathInstruction]] = { + if (activeParserUsesFixedSemantics) { + FixedJsonPathParser.parse(str) + } else { + LegacyJsonPathParser.parse(str) + } } } diff --git a/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index c90b79d3863..0f33441b674 100644 --- a/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -30,15 +30,71 @@ spark-rapids-shim-json-lines ***/ package com.nvidia.spark.rapids.shims +import scala.util.parsing.combinator.RegexParsers + +import com.nvidia.spark.rapids.PathInstruction + import org.apache.spark.SparkConf object GetJsonObjectShim { - private val FIXED_NAMED_PART_REGEXP = "[^\\']+" + // Copied from Apache Spark 4.0.0 JsonPathParser after SPARK-46761. + private object JsonPathParser extends RegexParsers { + import com.nvidia.spark.rapids.PathInstruction._ + + def root: Parser[Char] = '$' + + def long: Parser[Long] = "\\d+".r ^? { + case x => x.toLong + } + + // parse `[*]` and `[123]` subscripts + def subscript: Parser[List[PathInstruction]] = + for { + operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' + } yield { + Subscript :: operand :: Nil + } + + // parse `.name` or `['name']` child expressions + def named: Parser[List[PathInstruction]] = + for { + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" + } yield { + Key :: Named(name) :: Nil + } + + // child wildcards: `..`, `.*` or `['*']` + def wildcard: Parser[List[PathInstruction]] = + (".*" | "['*']") ^^^ List(Wildcard) + + def node: Parser[List[PathInstruction]] = + wildcard | + named | + subscript + + val expression: Parser[List[PathInstruction]] = { + phrase(root ~> rep(node) ^^ (x => x.flatten)) + } + + def parse(str: String): Option[List[PathInstruction]] = { + this.parseAll(expression, str) match { + case Success(result, _) => + Some(result) + + case _ => + None + } + } + } - private[rapids] def partRegexpInNamed(conf: SparkConf): String = FIXED_NAMED_PART_REGEXP + private[rapids] def parse( + str: String, + _conf: SparkConf): Option[List[PathInstruction]] = { + JsonPathParser.parse(str) + } /** * Spark 4 includes SPARK-46761, which accepts question marks in quoted path names. */ - def partRegexpInNamed: String = FIXED_NAMED_PART_REGEXP + def parse(str: String): Option[List[PathInstruction]] = JsonPathParser.parse(str) } diff --git a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala index ea4dae91a85..5e4b9c69948 100644 --- a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala +++ b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala @@ -30,30 +30,27 @@ class JsonPathParserSuite extends AnyFunSuite { test("Dataproc configuration selects quoted question mark support") { val dataprocConf = new SparkConf(false).set("spark.dataproc.engine", "default") - val dataprocRegexp = GetJsonObjectShim.partRegexpInNamed(dataprocConf) val fixedCases = Seq( "$['?']" -> List(Key, Named("?")), "$['a?b']" -> List(Key, Named("a?b")), "$.outer['?']" -> List(Key, Named("outer"), Key, Named("?"))) fixedCases.foreach { case (path, expected) => - assert(JsonPathParser.parseWithNamedPartRegexp(path, dataprocRegexp) === Some(expected)) + assert(GetJsonObjectShim.parse(path, dataprocConf) === Some(expected)) } } test("unquoted and malformed paths are independent of the configured platform") { - val vanillaRegexp = GetJsonObjectShim.partRegexpInNamed(new SparkConf(false)) - val dataprocRegexp = GetJsonObjectShim.partRegexpInNamed( - new SparkConf(false).set("spark.dataproc.engine", "default")) + val vanillaConf = new SparkConf(false) + val dataprocConf = new SparkConf(false).set("spark.dataproc.engine", "default") - Seq(vanillaRegexp, dataprocRegexp).distinct.foreach { partRegexpInNamed => - assert(JsonPathParser.parseWithNamedPartRegexp("$.?", partRegexpInNamed) === + Seq(vanillaConf, dataprocConf).foreach { conf => + assert(GetJsonObjectShim.parse("$.?", conf) === Some(questionMarkPath)) - assert(JsonPathParser.parseWithNamedPartRegexp("$['ordinary']", partRegexpInNamed) === + assert(GetJsonObjectShim.parse("$['ordinary']", conf) === Some(List(Key, Named("ordinary")))) - assert(JsonPathParser.parseWithNamedPartRegexp("$['']", partRegexpInNamed).isEmpty) - assert(JsonPathParser.parseWithNamedPartRegexp( - "$['unterminated]", partRegexpInNamed).isEmpty) + assert(GetJsonObjectShim.parse("$['']", conf).isEmpty) + assert(GetJsonObjectShim.parse("$['unterminated]", conf).isEmpty) } } @@ -72,9 +69,9 @@ class JsonPathParserSuite extends AnyFunSuite { case None => None case other => fail(s"Unexpected CPU get_json_object result: $other") } - val vanillaRegexp = GetJsonObjectShim.partRegexpInNamed(new SparkConf(false)) + val vanillaConf = new SparkConf(false) - assert(JsonPathParser.parseWithNamedPartRegexp("$['?']", vanillaRegexp) === + assert(GetJsonObjectShim.parse("$['?']", vanillaConf) === expectedInstructions) assert(JsonPathParser.parse("$['?']") === expectedInstructions) } From bc7d89bfa5aba4b3a558441994d331a4846b97ab Mon Sep 17 00:00:00 2001 From: Allen Xu Date: Thu, 3 Sep 2026 14:33:49 +0800 Subject: [PATCH 5/5] Map Dataproc JSON path semantics by Spark shim Signed-off-by: Allen Xu --- .../rapids/shims/GetJsonObjectShim.scala | 84 +--------- .../rapids/shims/GetJsonObjectShim.scala | 155 ++++++++++++++++++ .../spark/rapids/JsonPathParserSuite.scala | 16 +- 3 files changed, 175 insertions(+), 80 deletions(-) create mode 100644 sql-plugin/src/main/spark353/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala diff --git a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala index 322f72ac2ac..5cc2fb84c32 100644 --- a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala +++ b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -31,7 +31,6 @@ {"spark": "350db143"} {"spark": "351"} {"spark": "352"} -{"spark": "353"} {"spark": "354"} {"spark": "355"} {"spark": "356"} @@ -45,13 +44,11 @@ import scala.util.parsing.combinator.RegexParsers import com.nvidia.spark.rapids.PathInstruction -import org.apache.spark.{SparkConf, SparkEnv} +import org.apache.spark.SparkConf object GetJsonObjectShim { - private val DATAPROC_ENGINE_KEY = "spark.dataproc.engine" - // Copied from Apache Spark 3.5.3 JsonPathParser in jsonExpressions.scala. - private object LegacyJsonPathParser extends RegexParsers { + private object JsonPathParser extends RegexParsers { import com.nvidia.spark.rapids.PathInstruction._ def root: Parser[Char] = '$' @@ -100,78 +97,13 @@ object GetJsonObjectShim { } } - // Copied from Apache Spark 4.0.0 JsonPathParser after SPARK-46761. - private object FixedJsonPathParser extends RegexParsers { - import com.nvidia.spark.rapids.PathInstruction._ - - def root: Parser[Char] = '$' - - def long: Parser[Long] = "\\d+".r ^? { - case x => x.toLong - } - - // parse `[*]` and `[123]` subscripts - def subscript: Parser[List[PathInstruction]] = - for { - operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' - } yield { - Subscript :: operand :: Nil - } - - // parse `.name` or `['name']` child expressions - def named: Parser[List[PathInstruction]] = - for { - name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" - } yield { - Key :: Named(name) :: Nil - } - - // child wildcards: `..`, `.*` or `['*']` - def wildcard: Parser[List[PathInstruction]] = - (".*" | "['*']") ^^^ List(Wildcard) - - def node: Parser[List[PathInstruction]] = - wildcard | - named | - subscript - - val expression: Parser[List[PathInstruction]] = { - phrase(root ~> rep(node) ^^ (x => x.flatten)) - } - - def parse(str: String): Option[List[PathInstruction]] = { - this.parseAll(expression, str) match { - case Success(result, _) => - Some(result) - - case _ => - None - } - } - } - - private def useFixedParser(conf: SparkConf): Boolean = conf.contains(DATAPROC_ENGINE_KEY) - - private lazy val activeParserUsesFixedSemantics = - Option(SparkEnv.get).exists(env => useFixedParser(env.conf)) - - private[rapids] def parse(str: String, conf: SparkConf): Option[List[PathInstruction]] = { - if (useFixedParser(conf)) { - FixedJsonPathParser.parse(str) - } else { - LegacyJsonPathParser.parse(str) - } - } + private[rapids] def parse( + str: String, + _conf: SparkConf): Option[List[PathInstruction]] = JsonPathParser.parse(str) /** - * Spark 3.x uses the legacy quoted-name parser, except on Dataproc classic and Serverless - * runtimes. Dataproc sets `spark.dataproc.engine` and has backported SPARK-46761. + * Spark 3.x uses the legacy quoted-name parser. The Dataproc runtimes that backport + * SPARK-46761 use the Spark 3.5.3-specific shim instead. */ - def parse(str: String): Option[List[PathInstruction]] = { - if (activeParserUsesFixedSemantics) { - FixedJsonPathParser.parse(str) - } else { - LegacyJsonPathParser.parse(str) - } - } + def parse(str: String): Option[List[PathInstruction]] = JsonPathParser.parse(str) } diff --git a/sql-plugin/src/main/spark353/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala b/sql-plugin/src/main/spark353/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala new file mode 100644 index 00000000000..8d02a4767d6 --- /dev/null +++ b/sql-plugin/src/main/spark353/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala @@ -0,0 +1,155 @@ +/* + * Copyright (c) 2025-2026, NVIDIA CORPORATION. + * + * Licensed 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. + */ + +/*** spark-rapids-shim-json-lines +{"spark": "353"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids.shims + +import scala.util.parsing.combinator.RegexParsers + +import com.nvidia.spark.rapids.PathInstruction + +import org.apache.spark.{SparkConf, SparkEnv} + +object GetJsonObjectShim { + private val DATAPROC_ENGINE_KEY = "spark.dataproc.engine" + + // Copied from Apache Spark 3.5.3 JsonPathParser in jsonExpressions.scala. + private object LegacyJsonPathParser extends RegexParsers { + import com.nvidia.spark.rapids.PathInstruction._ + + def root: Parser[Char] = '$' + + def long: Parser[Long] = "\\d+".r ^? { + case x => x.toLong + } + + // parse `[*]` and `[123]` subscripts + def subscript: Parser[List[PathInstruction]] = + for { + operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' + } yield { + Subscript :: operand :: Nil + } + + // parse `.name` or `['name']` child expressions + def named: Parser[List[PathInstruction]] = + for { + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\'\\?]+".r <~ "']" + } yield { + Key :: Named(name) :: Nil + } + + // child wildcards: `..`, `.*` or `['*']` + def wildcard: Parser[List[PathInstruction]] = + (".*" | "['*']") ^^^ List(Wildcard) + + def node: Parser[List[PathInstruction]] = + wildcard | + named | + subscript + + val expression: Parser[List[PathInstruction]] = { + phrase(root ~> rep(node) ^^ (x => x.flatten)) + } + + def parse(str: String): Option[List[PathInstruction]] = { + this.parseAll(expression, str) match { + case Success(result, _) => + Some(result) + + case _ => + None + } + } + } + + // Copied from Apache Spark 4.0.0 JsonPathParser after SPARK-46761. + private object FixedJsonPathParser extends RegexParsers { + import com.nvidia.spark.rapids.PathInstruction._ + + def root: Parser[Char] = '$' + + def long: Parser[Long] = "\\d+".r ^? { + case x => x.toLong + } + + // parse `[*]` and `[123]` subscripts + def subscript: Parser[List[PathInstruction]] = + for { + operand <- '[' ~> ('*' ^^^ Wildcard | long ^^ Index) <~ ']' + } yield { + Subscript :: operand :: Nil + } + + // parse `.name` or `['name']` child expressions + def named: Parser[List[PathInstruction]] = + for { + name <- '.' ~> "[^\\.\\[]+".r | "['" ~> "[^\\']+".r <~ "']" + } yield { + Key :: Named(name) :: Nil + } + + // child wildcards: `..`, `.*` or `['*']` + def wildcard: Parser[List[PathInstruction]] = + (".*" | "['*']") ^^^ List(Wildcard) + + def node: Parser[List[PathInstruction]] = + wildcard | + named | + subscript + + val expression: Parser[List[PathInstruction]] = { + phrase(root ~> rep(node) ^^ (x => x.flatten)) + } + + def parse(str: String): Option[List[PathInstruction]] = { + this.parseAll(expression, str) match { + case Success(result, _) => + Some(result) + + case _ => + None + } + } + } + + private def useFixedParser(conf: SparkConf): Boolean = conf.contains(DATAPROC_ENGINE_KEY) + + private lazy val activeParserUsesFixedSemantics = + Option(SparkEnv.get).exists(env => useFixedParser(env.conf)) + + private[rapids] def parse(str: String, conf: SparkConf): Option[List[PathInstruction]] = { + if (useFixedParser(conf)) { + FixedJsonPathParser.parse(str) + } else { + LegacyJsonPathParser.parse(str) + } + } + + /** + * Dataproc classic 2.2/2.3 and Serverless 2.2/2.3 use Spark 3.5.3 builds with + * SPARK-46761 backported. Vanilla Spark 3.5.3 keeps the legacy parser. + */ + def parse(str: String): Option[List[PathInstruction]] = { + if (activeParserUsesFixedSemantics) { + FixedJsonPathParser.parse(str) + } else { + LegacyJsonPathParser.parse(str) + } + } +} diff --git a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala index 5e4b9c69948..d43507de162 100644 --- a/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala +++ b/tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala @@ -28,15 +28,23 @@ import org.apache.spark.unsafe.types.UTF8String class JsonPathParserSuite extends AnyFunSuite { private val questionMarkPath = List(Key, Named("?")) - test("Dataproc configuration selects quoted question mark support") { + // Classic 2.1 and Serverless 1.2 use the legacy parser through Spark 3.3.2/3.5.1. + // Classic 2.2/2.3 and Serverless 2.2/2.3 use patched Spark 3.5.3; Spark 4 is fixed. + private val dataprocUsesFixedParser = { + val sparkVersion = org.apache.spark.SPARK_VERSION + sparkVersion.startsWith("3.5.3") || sparkVersion.split('.').head.toInt >= 4 + } + + test("supported Dataproc shims select measured quoted question mark semantics") { val dataprocConf = new SparkConf(false).set("spark.dataproc.engine", "default") - val fixedCases = Seq( + val questionMarkCases = Seq( "$['?']" -> List(Key, Named("?")), "$['a?b']" -> List(Key, Named("a?b")), "$.outer['?']" -> List(Key, Named("outer"), Key, Named("?"))) - fixedCases.foreach { case (path, expected) => - assert(GetJsonObjectShim.parse(path, dataprocConf) === Some(expected)) + questionMarkCases.foreach { case (path, expected) => + val expectedResult = if (dataprocUsesFixedParser) Some(expected) else None + assert(GetJsonObjectShim.parse(path, dataprocConf) === expectedResult) } }