This is an automated email from the ASF dual-hosted git repository.
ulysses-you 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 f2f6bde39089 [SPARK-57987][SQL] Expose modifiedConfigs and node
description in SQL REST API
f2f6bde39089 is described below
commit f2f6bde390896a8f4d7ab43e069473e6b56066d2
Author: Xiduo You <[email protected]>
AuthorDate: Wed Jul 8 14:01:41 2026 +0800
[SPARK-57987][SQL] Expose modifiedConfigs and node description in SQL REST
API
### What changes were proposed in this pull request?
This PR exposes two additional fields in the SQL REST API
(`/api/v1/applications/[app-id]/sql/[execution-id]`):
- `modifiedConfigs` on `ExecutionData` -- the SQL configs modified for the
execution, sourced from `SQLExecutionUIData.modifiedConfigs`.
- `desc` on `Node` -- the physical plan node description, sourced from
`SparkPlanGraphNode.desc`.
### Why are the changes needed?
The SQL REST API already surfaces plan and metric information, but callers
cannot see which configs were modified for a given execution, nor the per-node
description string. Exposing these makes the REST API self-sufficient for
tooling that inspects executions without scraping the Web UI.
### Does this PR introduce _any_ user-facing change?
Yes. Two new fields are added to the SQL REST API response:
`modifiedConfigs` (map) on the execution object and `desc` (string) on each
plan node. Both default to empty/absent, so existing consumers are unaffected.
### How was this patch tested?
Extended `SqlResourceSuite` to populate and assert both fields (node `desc`
via the `details = true` cases, and `modifiedConfigs` via
`verifyExpectedExecutionData`). All 7 tests in the suite pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Closes #57066 from ulysses-you/sqlresource.
Authored-by: Xiduo You <[email protected]>
Signed-off-by: Xiduo You <[email protected]>
---
project/MimaExcludes.scala | 6 ++-
.../spark/status/api/v1/sql/SqlResource.scala | 6 ++-
.../org/apache/spark/status/api/v1/sql/api.scala | 20 +++++----
.../spark/status/api/v1/sql/SqlResourceSuite.scala | 51 ++++++++++++++++++----
4 files changed, 62 insertions(+), 21 deletions(-)
diff --git a/project/MimaExcludes.scala b/project/MimaExcludes.scala
index 442912429aa4..245e45abb68d 100644
--- a/project/MimaExcludes.scala
+++ b/project/MimaExcludes.scala
@@ -43,7 +43,11 @@ object MimaExcludes {
ProblemFilters.exclude[FinalMethodProblem](
"org.apache.spark.ml.recommendation.ALS.intermediateStorageLevel"),
ProblemFilters.exclude[FinalMethodProblem](
- "org.apache.spark.ml.recommendation.ALS.getIntermediateStorageLevel")
+ "org.apache.spark.ml.recommendation.ALS.getIntermediateStorageLevel"),
+ // [SPARK-57987] Add desc field to the SQL REST API Node case class
+
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.apply"),
+
ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.spark.status.api.v1.sql.Node.copy"),
+
ProblemFilters.exclude[MissingTypesProblem]("org.apache.spark.status.api.v1.sql.Node$")
)
// Exclude rules for 4.2.x from 4.1.0
diff --git
a/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/SqlResource.scala
b/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/SqlResource.scala
index b2c734ec27ac..223b19f09dda 100644
---
a/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/SqlResource.scala
+++
b/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/SqlResource.scala
@@ -266,7 +266,8 @@ private[v1] class SqlResource extends BaseAppResource {
edges,
if (exec.queryId != null) exec.queryId.toString else null,
exec.errorMessage.orNull,
- exec.rootExecutionId)
+ exec.rootExecutionId,
+ exec.modifiedConfigs)
}
private def printableMetrics(allNodes: collection.Seq[SparkPlanGraphNode],
@@ -286,7 +287,8 @@ private[v1] class SqlResource extends BaseAppResource {
val wholeStageCodegenId = nodeIdAndWSCGIdMap.get(node.id).flatten
val metrics =
node.metrics.flatMap(m => getMetric(metricValues, m.accumulatorId,
m.name.trim))
- Node(nodeId = node.id, nodeName = node.name.trim, wholeStageCodegenId,
metrics)
+ Node(nodeId = node.id, nodeName = node.name.trim, wholeStageCodegenId,
metrics,
+ desc = node.desc)
}
nodes.sortBy(_.nodeId).reverse
diff --git
a/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/api.scala
b/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/api.scala
index 723b6ba311d6..9eee17b4c129 100644
--- a/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/api.scala
+++ b/sql/core/src/main/scala/org/apache/spark/status/api/v1/sql/api.scala
@@ -20,6 +20,15 @@ import java.util.Date
import org.apache.spark.sql.execution.ui.SparkPlanGraphEdge
+case class Metric private[spark] (name: String, value: String)
+
+case class Node private[spark] (
+ nodeId: Long,
+ nodeName: String,
+ wholeStageCodegenId: Option[Long] = None,
+ metrics: collection.Seq[Metric],
+ desc: String = null)
+
class ExecutionData private[spark] (
val id: Long,
val status: String,
@@ -34,12 +43,5 @@ class ExecutionData private[spark] (
val edges: collection.Seq[SparkPlanGraphEdge],
val queryId: String = null,
val errorMessage: String = null,
- val rootExecutionId: Long = -1)
-
-case class Node private[spark](
- nodeId: Long,
- nodeName: String,
- wholeStageCodegenId: Option[Long] = None,
- metrics: collection.Seq[Metric])
-
-case class Metric private[spark] (name: String, value: String)
+ val rootExecutionId: Long = -1,
+ val modifiedConfigs: Map[String, String] = Map.empty)
diff --git
a/sql/core/src/test/scala/org/apache/spark/status/api/v1/sql/SqlResourceSuite.scala
b/sql/core/src/test/scala/org/apache/spark/status/api/v1/sql/SqlResourceSuite.scala
index e100a74e30c8..f2a3812b5930 100644
---
a/sql/core/src/test/scala/org/apache/spark/status/api/v1/sql/SqlResourceSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/status/api/v1/sql/SqlResourceSuite.scala
@@ -25,6 +25,7 @@ import org.scalatest.PrivateMethodTester
import org.apache.spark.{JobExecutionStatus, SparkFunSuite}
import org.apache.spark.sql.execution.ui.{SparkPlanGraph,
SparkPlanGraphCluster, SparkPlanGraphEdge, SparkPlanGraphNode,
SQLExecutionUIData, SQLPlanMetric}
+import org.apache.spark.status.api.v1.JacksonMessageWriter
object SqlResourceSuite {
@@ -38,16 +39,21 @@ object SqlResourceSuite {
val SIZE_OF_FILES_READ = "size of files read"
val PLAN_DESCRIPTION = "== Physical Plan ==\nCollectLimit (3)\n+- * Filter
(2)\n +- Scan text..."
val DESCRIPTION = "csv at MyDataFrames.scala:57"
+ val WHOLE_STAGE_CODEGEN_1_DESC = "WholeStageCodegen (1)"
+ val FILTER_DESC = "Filter (isnotnull(value#1))"
+ val SCAN_TEXT_DESC = "Scan text [value#1]"
+ val MODIFIED_CONFIGS: Map[String, String] =
+ Map("spark.sql.shuffle.partitions" -> "200", "spark.sql.adaptive.enabled"
-> "true")
val nodeIdAndWSCGIdMap: Map[Long, Option[Long]] = Map(1L -> Some(1L))
- val filterNode = new SparkPlanGraphNode(1, FILTER, "",
+ val filterNode = new SparkPlanGraphNode(1, FILTER, FILTER_DESC,
metrics = Seq(SQLPlanMetric(NUMBER_OF_OUTPUT_ROWS, 1, "")))
val nodes: Seq[SparkPlanGraphNode] = Seq(
- new SparkPlanGraphCluster(0, WHOLE_STAGE_CODEGEN_1, "",
+ new SparkPlanGraphCluster(0, WHOLE_STAGE_CODEGEN_1,
WHOLE_STAGE_CODEGEN_1_DESC,
nodes = ArrayBuffer(filterNode),
metrics = Seq(SQLPlanMetric(DURATION, 0, ""))),
- new SparkPlanGraphNode(2, SCAN_TEXT, "",
+ new SparkPlanGraphNode(2, SCAN_TEXT, SCAN_TEXT_DESC,
metrics = Seq(
SQLPlanMetric(METADATA_TIME, 2, ""),
SQLPlanMetric(NUMBER_OF_FILES_READ, 3, ""),
@@ -85,7 +91,7 @@ object SqlResourceSuite {
description = DESCRIPTION,
details = "",
physicalPlanDescription = PLAN_DESCRIPTION,
- Map.empty,
+ MODIFIED_CONFIGS,
metrics = metrics,
submissionTime = 1586768888233L,
completionTime = Some(new Date(1586768888999L)),
@@ -101,26 +107,31 @@ object SqlResourceSuite {
private def getNodes(): Seq[Node] = {
val node = Node(0, WHOLE_STAGE_CODEGEN_1,
- wholeStageCodegenId = None, metrics = Seq(Metric(DURATION, "0 ms")))
+ wholeStageCodegenId = None, metrics = Seq(Metric(DURATION, "0 ms")),
+ desc = WHOLE_STAGE_CODEGEN_1_DESC)
val node2 = Node(1, FILTER,
- wholeStageCodegenId = Some(1), metrics =
Seq(Metric(NUMBER_OF_OUTPUT_ROWS, "1")))
+ wholeStageCodegenId = Some(1), metrics =
Seq(Metric(NUMBER_OF_OUTPUT_ROWS, "1")),
+ desc = FILTER_DESC)
val node3 = Node(2, SCAN_TEXT, wholeStageCodegenId = None,
metrics = Seq(Metric(METADATA_TIME, "2 ms"),
Metric(NUMBER_OF_FILES_READ, "1"),
Metric(NUMBER_OF_OUTPUT_ROWS, "1"),
- Metric(SIZE_OF_FILES_READ, "330.0 B")))
+ Metric(SIZE_OF_FILES_READ, "330.0 B")),
+ desc = SCAN_TEXT_DESC)
// reverse order because of supporting execution order by aligning with
Spark-UI
Seq(node3, node2, node)
}
private def getExpectedNodesWhenWholeStageCodegenIsOff(): Seq[Node] = {
- val node = Node(1, FILTER, metrics = Seq(Metric(NUMBER_OF_OUTPUT_ROWS,
"1")))
+ val node = Node(1, FILTER, metrics = Seq(Metric(NUMBER_OF_OUTPUT_ROWS,
"1")),
+ desc = FILTER_DESC)
val node2 = Node(2, SCAN_TEXT,
metrics = Seq(Metric(METADATA_TIME, "2 ms"),
Metric(NUMBER_OF_FILES_READ, "1"),
Metric(NUMBER_OF_OUTPUT_ROWS, "1"),
- Metric(SIZE_OF_FILES_READ, "330.0 B")))
+ Metric(SIZE_OF_FILES_READ, "330.0 B")),
+ desc = SCAN_TEXT_DESC)
// reverse order because of supporting execution order by aligning with
Spark-UI
Seq(node2, node)
@@ -145,6 +156,7 @@ object SqlResourceSuite {
assert(executionData.queryId == "efe98ba7-1532-491e-9b4f-4be621cef37c")
assert(executionData.errorMessage == null)
assert(executionData.rootExecutionId == 1)
+ assert(executionData.modifiedConfigs == MODIFIED_CONFIGS)
}
}
@@ -247,4 +259,25 @@ class SqlResourceSuite extends SparkFunSuite with
PrivateMethodTester {
assert(executionData.errorMessage == null)
assert(executionData.rootExecutionId == -1)
}
+
+ test("SPARK-57987: JSON serialization of default modifiedConfigs and node
desc") {
+ val mapper = new JacksonMessageWriter().mapper
+ val nodeWithEmptyDesc = Node(0, SCAN_TEXT, metrics = Seq.empty)
+ val executionData = new ExecutionData(
+ id = 0,
+ status = "COMPLETED",
+ description = DESCRIPTION,
+ planDescription = "",
+ submissionTime = new Date(),
+ duration = 0,
+ runningJobIds = Seq.empty,
+ successJobIds = Seq.empty,
+ failedJobIds = Seq.empty,
+ nodes = Seq(nodeWithEmptyDesc),
+ edges = Seq.empty)
+ val executionJson =
mapper.writeValueAsString(executionData).replaceAll("\\s", "")
+ assert(executionJson.contains("\"modifiedConfigs\":{}"))
+ assert(executionJson.contains(
+ "\"nodes\":[{\"nodeId\":0,\"nodeName\":\"Scantext\",\"metrics\":[]}]"))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
