[SkipRecovery] Fix get_json_object runtime path semantics [databricks] - #15850
[SkipRecovery] Fix get_json_object runtime path semantics [databricks]#15850wjxiz1992 wants to merge 7 commits into
Conversation
Signed-off-by: Allen Xu <allxu@nvidia.com>
Greptile SummaryThis PR moves JSON-path parsing into version-specific shims so GPU
Confidence Score: 5/5The PR appears safe to merge because no blocking failure remains. No blocking failure remains. Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
P[Literal JSON path] --> V{Spark shim}
V -->|Spark 3 except 3.5.3| L[Legacy parser]
V -->|Spark 3.5.3| D{Dataproc marker present?}
D -->|No| L
D -->|Yes| F[Fixed parser]
V -->|Spark 4| F
L --> I[Path instructions]
F --> I
I --> E{Any valid paths?}
E -->|Yes| J[JNI multi-path evaluation]
E -->|No| N[Construct null output columns]
Reviews (7): Last reviewed commit: "Merge branch 'main' into fix/14290-get-j..." | Re-trigger Greptile |
There was a problem hiding this comment.
Pull request overview
This pull request fixes GPU/CPU divergence for get_json_object when JSON paths contain quoted ? keys on some Spark 3.5 runtimes with vendor backports (e.g., Dataproc), by detecting the runtime’s Catalyst semantics once and selecting the matching JSON-path parser dialect (or failing closed to CPU when unknown).
Changes:
- Added a shared
GetJsonObjectRuntimeSemanticsclassifier and updated shims to exposequotedQuestionMarkSupport(Spark 4 hard-coded modern; pre-Spark-4 probes Catalyst at runtime). - Updated
GpuGetJsonObject/JsonPathParserto parse quoted names using the selected dialect, and fail closed to CPU when runtime semantics are unknown. - Added Scala unit tests plus Python integration tests for quoted
?keys and invalid-path batches (with GPU plan capture assertions).
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| tests/src/test/scala/com/nvidia/spark/rapids/JsonPathParserSuite.scala | New unit coverage for parser dialect behavior and runtime probe classification. |
| sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala | Spark 4 shim now reports modern quoted-? support. |
| sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/GetJsonObjectShim.scala | Pre-Spark-4 shim probes Catalyst once to detect quoted-? support at runtime. |
| sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GetJsonObjectRuntimeSemantics.scala | New shared helper to classify probe results (modern/legacy/unknown). |
| sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuGetJsonObject.scala | Plumbs dialect into parsing and expression conversion; adds fail-closed behavior and fixes empty multi-path handling. |
| integration_tests/src/main/python/get_json_test.py | Adds/updates ITs for quoted ? keys and invalid paths with executed-plan capture. |
Suppressed comments (1)
integration_tests/src/main/python/get_json_test.py:158
- This query projects multiple
get_json_objectexpressions, so it is expected to be combined intoGpuMultiGetJsonObject(and not necessarily containGpuGetJsonObjectnodes). AssertingGpuGetJsonObjecthere can cause a false failure if expression combining is enabled (default). AssertGpuMultiGetJsonObjectinstead.
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')
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| 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') |
There was a problem hiding this comment.
I kept the GpuGetJsonObject plan assertion. GpuEquivalentExpressions.replaceMultiExpressions runs inside GpuProjectExec.internalDoExecuteColumnar while binding a GpuTieredProject, so it does not rewrite the Spark executedPlan tree inspected by exist_classes. Changing both assertions to GpuMultiGetJsonObject made the focused Spark 3.3 GPU IT fail; restoring GpuGetJsonObject passed both tests (2 passed).
wjxiz1992
changed the title
Signed-off-by: Allen Xu <allxu@nvidia.com>
|
build |
|
|
||
| import org.apache.spark.unsafe.types.UTF8String | ||
|
|
||
| private[rapids] object GetJsonObjectRuntimeSemantics { |
There was a problem hiding this comment.
Personally GetJsonObjectRuntimeSemantics is not necessary, we can merge it with object GetJsonObjectShim.
It is not a good idea to put the expected result "QUESTION" and the Json test literal string {"?":"QUESTION"} into different classes/objects.
If you want keep GetJsonObjectRuntimeSemantics common, maybe we need callers to provide the expect results instead of the hardcode. e.g. def classifyQuotedQuestionMarkResult(result: => Any, expected: Any): Option[Boolean]
Signed-off-by: Allen Xu <allxu@nvidia.com>
|
build |
revans2
left a comment
revans2
left a comment
There was a problem hiding this comment.
I also am not thrilled with us trying to combine the different parser versions. For me if we are copying code from Spark, it would be nice to make sure it stays copied, even if it means more duplicate code. It just makes it that much simpler to maintain because we can diff it with the original code. I get in this case the change is very small, so for me it is just a nit.
| import org.apache.spark.unsafe.types.UTF8String | ||
|
|
||
| object GetJsonObjectShim { | ||
| private lazy val runtimeQuotedQuestionMarkSupport: Option[Boolean] = { |
There was a problem hiding this comment.
I personally hate this solution. We know what all of the versions of Spark are that we support and we should be running tests on them. Trying to figure it out at runtime to me is not a good solution. If you want to take this code and go run it on the different versions of Spark to get the result to put into the shim, then okay, but I don't want to ship with this.
The other alternative is we just fix it everywhere for OUR code. Yes we will not be bug for bug compatible, but I am fine with documenting it and just telling everyone that we are using the fixed version. That does not solve the problem of figuring out when this should be on and when it should be off ahead of time because we would need the tests to know. But it was just for dataproc and dataproc serverless. Why are we not just testing them to know.
There was a problem hiding this comment.
Thanks, Bobby. You were right that this should be established from the supported Dataproc runtime matrix instead of inferred by executing a Catalyst expression.
I ran the quoted question-mark case across every supported Classic and Serverless line:
| Environment | Version tested | Spark | Result |
|---|---|---|---|
| Classic 2.1 | 2.1.118 | 3.3.2 | NULL — legacy |
| Classic 2.2 | current 2.2, resolved to 2.2.84 | 3.5.3 | QUESTION — fixed |
| Classic 2.3 | 2.3.35 | 3.5.3 | QUESTION — fixed |
| Serverless 1.2 | 1.2.86 | 3.5.1 | NULL — legacy |
| Serverless 2.2 | 2.2.86 | 3.5.3 | QUESTION — fixed |
| Serverless 2.3 | 2.3.39 | 3.5.3 | QUESTION — fixed |
| Serverless 3.0 | current 3.0 line | 4.0.1 | QUESTION — fixed |
The exact Classic 2.2.86 image is retired, so I tested the currently resolvable Classic 2.2 image, which was 2.2.84. Every submitted Classic job reached DONE with YARN FINISHED, and every Serverless batch reached SUCCEEDED.
Based on those results, commit bc7d89bfa removes the semantic runtime probe and narrows the backport to the exact Spark 3.5.3 shim. Other Spark 3 shims remain legacy, and Spark 4 remains fixed. Within the 3.5.3 shim, spark.dataproc.engine is used only as a platform identity marker because vanilla Spark 3.5.3 and the Google-patched Spark 3.5.3 report the same Spark version but have different CPU behavior. It does not execute a query or infer semantics from a result, and it preserves CPU/GPU parity on vanilla Spark 3.5.3.
I also kept the legacy and fixed parser implementations as independent copies of the corresponding Spark sources, per your maintainability suggestion.
Finally, I built a custom JAR from this exact commit and validated it on Dataproc Classic 2.2.84 with Spark 3.5.3. Job 9fe997a7047f45e29cfa6f18663f07e0 completed DONE with YARN FINISHED: CPU=QUESTION, GPU=QUESTION, CPU_GPU_EQUAL=true, and the executed plan contained GpuProject with get_json_object.
Signed-off-by: Allen Xu <allxu@nvidia.com>
|
build |
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 <allxu@nvidia.com>
Signed-off-by: Allen Xu <allxu@nvidia.com>
|
build |
2 similar comments
|
build |
|
build |
Signed-off-by: Allen Xu <allxu@nvidia.com>
|
build |
1 similar comment
|
build |
wjxiz1992
changed the title
wjxiz1992
changed the title
wjxiz1992
changed the title
|
Hi @revans2, when you have a chance, could you please take another look at this PR? I’ve followed up on your review feedback. Thanks! |
JaCoCo production line coverage: +83 lines (
sql-plugin +83: shared parser +4, Spark 330 shim +18, Spark 353 shim +43, Spark 400 shim +18;JsonPathParserSuitefix-line intersection)Fixes #14290.
Description
Dataproc backported SPARK-46761 while retaining Spark 3.5.3. A Spark version check alone therefore cannot distinguish vanilla Spark 3.5.3 from the Google-patched runtime, and the previous revision inferred the behavior by evaluating a Catalyst expression at runtime.
This revision removes that semantic probe and uses the mapping established by testing every supported Dataproc line:
spark.dataproc.engineplatform marker identifies Dataproc. Vanilla Spark 3.5.3 remains legacy so CPU and GPU behavior stay aligned.The change is cold-path only. Non-3.5.3 shims select statically, and the 3.5.3 shim performs one lazy SparkConf marker lookup. No per-row, per-batch, or JNI work was added; parsed instructions remain cached.
Dataproc runtime matrix
The quoted question-mark key was evaluated with the CPU Spark expression on every supported Classic and Serverless line.
NULL— legacy2.2-ubuntu22, resolved to 2.2.84QUESTION— fixedQUESTION— fixedNULL— legacyQUESTION— fixedQUESTION— fixedQUESTION— fixedThe exact Classic 2.2.86 image is retired and was rejected before resource creation, so the Classic 2.2 check used the currently resolvable 2.2 image, 2.2.84. All submitted Classic probes reached
DONEwith YARNFINISHED; all Serverless batches reachedSUCCEEDED.Exact-head Dataproc GPU validation
A clean custom JAR built from commit
bc7d89bfa5aba4b3a558441994d331a4846b97abwas validated on Dataproc Classic 2.2.84-debian12 with Spark 3.5.3 and two Tesla T4 workers.69194905a24df75828273219dd9c72c55625b272a5dea6a49c6575cdf703ca909fe997a7047f45e29cfa6f18663f07e0:DONE, YARNFINISHED.QUESTION; GPU result:QUESTION;CPU_GPU_EQUAL=true.GpuProjectwithget_json_objectandGpuRange;GPU_PLAN_EVIDENCE=PASS.Local validation
JsonPathParserSuite: Spark build versions 330, 332, 340, 351, 353, 400, and 401 all passed withsucceeded 4, failed 0; Spark 400 passed with ANSI both off and on.BUILD SUCCESS.3 passed, 32630 deselected; CPU/GPU equality and GPU-plan capture passed.sql-plugin +83distinct covered production lines.AI assistance: This change and PR description were prepared with AI assistance.
Checklists
Documentation
Testing
(Please provide the names of the existing tests in the PR description.)
Performance