You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by sl...@apache.org on 2026/10/09 09:06:24 UTC
(auron) branch master updated: [AURON #2559] Add native conversion diagnostics to Auron SQL UI. (#2560)
This is an automated email from the ASF dual-hosted git repository.
slfan1989 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/auron.git
The following commit(s) were added to refs/heads/master by this push:
new 73300dc4 [AURON #2559] Add native conversion diagnostics to Auron SQL UI. (#2560)
73300dc4 is described below
commit 73300dc4a75386692a01d74064aff44c584612ed
Author: slfan1989 <55...@users.noreply.github.com>
AuthorDate: Fri Oct 9 17:06:18 2026 +0800
[AURON #2559] Add native conversion diagnostics to Auron SQL UI. (#2560)
### Which issue does this PR close?
Closes #2559
### Rationale for this change
The Auron tab currently displays build information but does not explain
which operators were converted to native execution or why others
remained in Spark.
This change makes conversion diagnostics available in the Spark UI,
including separate snapshots for AQE planning stages, so users can
investigate fallback behavior alongside relevant configuration and
metrics.
### What changes are included in this PR?
- Add a paginated SQL execution list with lifecycle information and
conversion counts from the latest snapshot.
- Capture per-operator conversion status, fallback reasons, physical
plans, and selected effective configuration values.
- Retain up to 20 recent snapshots per execution, collapsing consecutive
identical snapshots.
- Distinguish native operators, recorded fallbacks, other Spark
operators, and wrappers. Exclude wrappers from conversion counts.
- Display available Spark SQL operator metrics and links to SQL
execution details.
### Are there any user-facing changes?
Yes.
The Auron tab now provides SQL execution summaries and detailed
conversion diagnostics when `spark.auron.ui.enabled=true`.
#### SQL execution overview
The Auron tab summarizes native operators, recorded fallbacks, other
Spark operators, and snapshot counts for each SQL execution. Executions
without conversion snapshots explicitly show "Not available".
<img width="1280" height="720" alt="image"
src="https://github.com/user-attachments/assets/4f4b9a2e-300d-4727-aeeb-1c355152ef51"
/>
#### Actionable fallback reasons
Per-operator diagnostics distinguish configuration-disabled conversion
from unsupported operators. In this example, native data writing is
disabled, while WriteFiles and Range remain in Spark.
<img width="1280" height="660" alt="image"
src="https://github.com/user-attachments/assets/6ffb71b2-70dd-4e9e-8219-8f6a9ddf26fc"
/>
#### Configuration captured at conversion time
Selected effective configuration values are recorded with each snapshot,
including the overall native writing switch and the Parquet/ORC-specific
switches.
<img width="1280" height="720" alt="image"
src="https://github.com/user-attachments/assets/f7ce9298-b4d1-4bdd-a2fa-85dc2e62ffd8"
/>
#### AQE snapshots and execution metrics
Separate snapshots capture different AQE conversion stages. Counts apply
to each snapshot, and wrappers are tracked separately. Available Spark
SQL metrics are shown with their original units and aggregation
semantics.
<img width="1280" height="700" alt="image"
src="https://github.com/user-attachments/assets/34e708ad-69a4-462a-ac11-e0f9a22863c2"
/>
### How was this patch tested?
- Passed `./dev/reformat --check`, including Spark 3.0–4.1 compilation
checks.
- Passed all 9 tests in `AuronDiagnosticsSuite` and
`BuildInfoInSparkUISuite`, covering event serialization, lifecycle
handling, retention, pagination, shared plan traversal, HTML escaping,
wrapper classification, native execution, UDF fallback, and disabled
Auron.
### Was this patch authored or co-authored using generative AI tooling?
- [x] Yes
- [ ] No
`Generated-by: OpenAI Codex (GPT-6)`
ASF guidance: https://www.apache.org/legal/generative-tooling.html
---
.../org/apache/auron/spark/ui/AuronEvent.scala | 11 ++
.../sql/execution/ui/AuronAllExecutionsPage.scala | 18 +-
.../sql/execution/ui/AuronDiagnosticsView.scala | 158 ++++++++++++++++
.../execution/ui/AuronSQLAppStatusListener.scala | 63 ++++++-
.../sql/execution/ui/AuronSQLAppStatusStore.scala | 55 ++++++
.../sql/execution/BuildInfoInSparkUISuite.scala | 53 +++++-
.../sql/execution/ui/AuronDiagnosticsSuite.scala | 203 +++++++++++++++++++++
.../spark/sql/auron/AuronPlanDiagnostics.scala | 140 ++++++++++++++
.../sql/auron/AuronSparkSessionExtension.scala | 8 +
9 files changed, 702 insertions(+), 7 deletions(-)
diff --git a/auron-spark-ui/src/main/scala/org/apache/auron/spark/ui/AuronEvent.scala b/auron-spark-ui/src/main/scala/org/apache/auron/spark/ui/AuronEvent.scala
index b0dd2fdd..af86a3f5 100644
--- a/auron-spark-ui/src/main/scala/org/apache/auron/spark/ui/AuronEvent.scala
+++ b/auron-spark-ui/src/main/scala/org/apache/auron/spark/ui/AuronEvent.scala
@@ -23,3 +23,14 @@ import org.apache.spark.scheduler.SparkListenerEvent
sealed trait AuronEvent extends SparkListenerEvent {}
case class AuronBuildInfoEvent(info: mutable.LinkedHashMap[String, String]) extends AuronEvent {}
+
+/** A conversion snapshot; AQE can publish several snapshots for one SQL execution. */
+case class AuronPlanDiagnosticEvent(
+ executionId: Long,
+ timestamp: Long,
+ physicalPlan: String,
+ nodes: Seq[AuronPlanDiagnosticNode],
+ configurations: Map[String, String])
+ extends AuronEvent
+
+case class AuronPlanDiagnosticNode(id: Int, name: String, status: String, reason: String)
diff --git a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala
index 3523d4ad..293aed48 100644
--- a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala
+++ b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala
@@ -29,12 +29,26 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
@sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
override def render(request: javax.servlet.http.HttpServletRequest): Seq[Node] = {
- UIUtils.headerSparkPage(request, "Auron", buildInfoSummary(sqlStore.buildInfo()), parent)
+ UIUtils.headerSparkPage(
+ request,
+ "Auron",
+ buildInfoSummary(sqlStore.buildInfo()) ++ new AuronDiagnosticsView(sqlStore).render(
+ request.getParameter("id"),
+ request.getParameter("page"),
+ UIUtils.prependBaseUri(request, parent.basePath)),
+ parent)
}
@sparkver("4.0 / 4.1 / 4.2")
override def render(request: jakarta.servlet.http.HttpServletRequest): Seq[Node] = {
- UIUtils.headerSparkPage(request, "Auron", buildInfoSummary(sqlStore.buildInfo()), parent)
+ UIUtils.headerSparkPage(
+ request,
+ "Auron",
+ buildInfoSummary(sqlStore.buildInfo()) ++ new AuronDiagnosticsView(sqlStore).render(
+ request.getParameter("id"),
+ request.getParameter("page"),
+ UIUtils.prependBaseUri(request, parent.basePath)),
+ parent)
}
private def propertyHeader = Seq("Name", "Value")
diff --git a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsView.scala b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsView.scala
new file mode 100644
index 00000000..7e7119e5
--- /dev/null
+++ b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsView.scala
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.spark.sql.execution.ui
+
+import scala.util.Try
+import scala.xml.NodeSeq
+
+import org.apache.spark.ui.UIUtils
+
+/** Shared rendering for live and historical executions, independent of servlet API versions. */
+private[ui] class AuronDiagnosticsView(sqlStore: AuronSQLAppStatusStore) {
+ def render(id: String, page: String, base: String): NodeSeq = {
+ if (id != null) {
+ Try(id.toLong).toOption.filter(_ >= 0).flatMap(sqlStore.execution) match {
+ case Some(execution) => executionDetails(execution, base)
+ case None =>
+ <div class="alert alert-info">Execution unavailable or no longer retained.</div>
+ }
+ } else {
+ val size = 20
+ val count = sqlStore.executionsCount()
+ val last = math.max(1L, (count + size - 1) / size).min(Int.MaxValue / size).toInt
+ val current = Try(Option(page).getOrElse("1").toInt).getOrElse(1).max(1).min(last)
+ val rows = sqlStore.executions((current - 1) * size, size)
+ <div>
+ <h3>SQL conversion diagnostics ({count})</h3>
+ <p>Counts describe the latest captured conversion snapshot, not the entire AQE query.</p>
+ <table class="table table-bordered table-striped">
+ <thead><tr><th>SQL ID</th><th>Description</th><th>Lifecycle</th>
+ <th>Native</th><th>Fallback with reason</th><th>Other Spark</th><th>Snapshots</th></tr></thead>
+ <tbody>{
+ rows.map { e =>
+ val nodes = e.snapshots.lastOption.toSeq.flatMap(_.nodes)
+ def count(status: String): String = {
+ if (e.snapshots.isEmpty) "Not available" else nodes.count(_.status == status).toString
+ }
+ <tr><td><a href={s"$base/auron/?id=${e.executionId}"}>{e.executionId}</a></td>
+ <td>{e.description}</td><td>{
+ if (e.endedAt > 0) "Finished" else "Running / planning"
+ }</td>
+ <td>{count("Native")}</td>
+ <td>{count("Fallback")}</td>
+ <td>{count("Spark (no reason recorded)")}</td>
+ <td>{e.snapshots.size}</td></tr>
+ }
+ }</tbody>
+ </table>
+ {if (rows.isEmpty) <p>No SQL executions recorded.</p> else NodeSeq.Empty}
+ <p><a href={s"$base/auron/?page=${math.max(1, current - 1)}"}>Previous</a>
+ {s" - Page $current of $last - "}
+ <a href={s"$base/auron/?page=${math.min(last, current + 1)}"}>Next</a></p>
+ </div>
+ }
+ }
+
+ private def executionDetails(e: AuronExecutionUIData, base: String): NodeSeq = {
+ val metrics = sqlStore.metrics(e.executionId)
+ <div>
+ <p><a href={s"$base/auron/"}>All executions</a>{" | "}
+ <a href={
+ s"$base/SQL/execution/?id=${e.executionId}"
+ }>Spark SQL execution, jobs and stages</a></p>
+ <h3>SQL {e.executionId}: {e.description}</h3>
+ <p>{
+ if (e.startedAt > 0 && e.endedAt > 0) {
+ s"Elapsed: ${e.endedAt - e.startedAt} ms"
+ } else {
+ "Execution timing is not available yet."
+ }
+ }</p>
+ <h4>Conversion snapshots</h4>
+ <p>The latest 20 distinct snapshots are retained. AQE may convert individual stages separately.
+ Node IDs are local to each snapshot; wrappers are excluded from conversion counts.
+ Other Spark means no fallback reason was recorded; it does not indicate native conversion.
+ Plans are limited to 65,536 characters and individual reasons to 8,192 characters.</p>
+ {
+ if (e.snapshots.isEmpty) {
+ <p>No conversion snapshot was captured for this execution.</p>
+ } else {
+ e.snapshots.reverse.map { snapshot =>
+ <details><summary>Snapshot {new java.util.Date(snapshot.timestamp).toString}
+ {
+ s" (${snapshot.nodes.count(_.status == "Native")} native, " +
+ s"${snapshot.nodes.count(_.status == "Fallback")} fallback, " +
+ s"${snapshot.nodes.count(_.status == "Spark (no reason recorded)")} other Spark, " +
+ s"${snapshot.nodes.count(_.status == "Wrapper")} wrappers)"
+ }</summary>
+ <h4>Operator diagnostics</h4>
+ <table class="table table-bordered"><thead><tr><th>ID</th><th>Operator</th>
+ <th>Conversion</th><th>Reason</th></tr></thead>
+ <tbody>{
+ snapshot.nodes.map(n =>
+ <tr><td>{n.id}</td><td>{n.name}</td>
+ <td>{n.status}</td><td><pre>{
+ if (n.status == "Wrapper") {
+ "Planning/execution wrapper; excluded from conversion counts. See child operators."
+ } else {
+ n.reason
+ }
+ }</pre></td></tr>)
+ }</tbody></table>
+ <h4>Physical plan</h4><pre>{snapshot.physicalPlan}</pre>
+ <h4>Configuration at conversion</h4>
+ {
+ UIUtils.listingTable(
+ propertyHeader,
+ propertyRow,
+ snapshot.configurations.toSeq.sortBy(_._1),
+ fixedWidth = true)
+ }
+ </details>
+ }
+ }
+ }
+ <h4>Execution metrics</h4>
+ <p>Metrics reported by Spark, including available native timing, row counts, memory,
+ spill and shuffle measurements. Values retain Spark's units and aggregation semantics;
+ they are not summed across operators. Metric IDs belong to Spark's plan graph.</p>
+ {
+ if (metrics.isEmpty) {
+ <p>Metrics are not available yet or have been removed by Spark retention.</p>
+ } else {
+ <table class="table table-bordered"><thead><tr><th>Operator</th><th>Metric</th>
+ <th>Value</th></tr></thead><tbody>{
+ metrics.map { case (operator, name, value) =>
+ <tr><td>{operator}</td><td>{name}</td><td><pre>{value}</pre></td></tr>
+ }
+ }</tbody></table>
+ }
+ }
+ </div>
+ }
+
+ private def propertyHeader = Seq("Name", "Value")
+
+ private def propertyRow(kv: (String, String)) = <tr>
+ <td>
+ {kv._1}
+ </td> <td>
+ {kv._2}
+ </td>
+ </tr>
+
+}
diff --git a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala
index 651234ac..762b678c 100644
--- a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala
+++ b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala
@@ -16,20 +16,38 @@
*/
package org.apache.spark.sql.execution.ui
-import scala.annotation.nowarn
+import scala.collection.JavaConverters._
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.internal.Logging
import org.apache.spark.scheduler.{SparkListener, SparkListenerEvent}
import org.apache.spark.status.ElementTrackingStore
-import org.apache.auron.spark.ui.AuronBuildInfoEvent
+import org.apache.auron.spark.ui.{AuronBuildInfoEvent, AuronPlanDiagnosticEvent}
-@nowarn("cat=unused") // conf temporarily unused
class AuronSQLAppStatusListener(conf: SparkConf, kvstore: ElementTrackingStore)
extends SparkListener
with Logging {
+ private val store = new AuronSQLAppStatusStore(kvstore)
+ private val retained = math.max(0, conf.getInt("spark.sql.ui.retainedExecutions", 1000))
+
+ kvstore.addTrigger(classOf[AuronExecutionUIData], retained) { count =>
+ val iter = kvstore.view(classOf[AuronExecutionUIData]).closeableIterator()
+ val ids =
+ try {
+ iter.asScala
+ .filter(_.endedAt > 0)
+ .take(math.max(0L, count - retained).toInt)
+ .map(_.executionId)
+ .toVector
+ } finally iter.close()
+ ids.foreach(id => kvstore.delete(classOf[AuronExecutionUIData], id))
+ }
+
+ private def current(id: Long): AuronExecutionUIData =
+ store.execution(id).getOrElse(new AuronExecutionUIData(id, "", 0L, 0L, Seq.empty))
+
def getAuronBuildInfo(): Long = {
kvstore.count(classOf[AuronBuildInfoUIData])
}
@@ -41,6 +59,45 @@ class AuronSQLAppStatusListener(conf: SparkConf, kvstore: ElementTrackingStore)
override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
case e: AuronBuildInfoEvent => onAuronBuildInfo(e)
+ case e: SparkListenerSQLExecutionStart =>
+ val old = current(e.executionId)
+ kvstore.write(
+ new AuronExecutionUIData(
+ e.executionId,
+ e.description,
+ e.time,
+ old.endedAt,
+ old.snapshots),
+ true)
+ case e: SparkListenerSQLExecutionEnd =>
+ val old = current(e.executionId)
+ kvstore.write(
+ new AuronExecutionUIData(
+ e.executionId,
+ old.description,
+ old.startedAt,
+ e.time,
+ old.snapshots),
+ true)
+ case e: AuronPlanDiagnosticEvent =>
+ val old = current(e.executionId)
+ // Keep bounded history; repeated identical snapshots do not consume space.
+ val snapshots =
+ if (old.snapshots.lastOption.exists(s =>
+ s.physicalPlan == e.physicalPlan && s.nodes == e.nodes &&
+ s.configurations == e.configurations)) {
+ old.snapshots
+ } else {
+ (old.snapshots :+ e).takeRight(20)
+ }
+ kvstore.write(
+ new AuronExecutionUIData(
+ e.executionId,
+ old.description,
+ old.startedAt,
+ old.endedAt,
+ snapshots),
+ true)
case _ => // Ignore
}
diff --git a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala
index 0c44644f..1df53d8e 100644
--- a/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala
+++ b/auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala
@@ -16,14 +16,57 @@
*/
package org.apache.spark.sql.execution.ui
+import scala.collection.JavaConverters._
import scala.util.control.NonFatal
import com.fasterxml.jackson.annotation.JsonIgnore
import org.apache.spark.internal.Logging
import org.apache.spark.util.kvstore.{KVIndex, KVStore}
+import org.apache.auron.spark.ui.AuronPlanDiagnosticEvent
+
class AuronSQLAppStatusStore(store: KVStore) extends Logging {
+ private val sparkStore = new SQLAppStatusStore(store)
+
+ def execution(id: Long): Option[AuronExecutionUIData] = {
+ try Some(store.read(classOf[AuronExecutionUIData], id))
+ catch { case _: NoSuchElementException => None }
+ }
+
+ def executionsCount(): Long = store.count(classOf[AuronExecutionUIData])
+
+ def executions(offset: Int, limit: Int): Seq[AuronExecutionUIData] = {
+ val iter = store
+ .view(classOf[AuronExecutionUIData])
+ .reverse()
+ .skip(offset)
+ .max(limit)
+ .closeableIterator()
+ try iter.asScala.toVector
+ finally iter.close()
+ }
+
+ def metrics(id: Long): Seq[(String, String, String)] = {
+ try {
+ val values = sparkStore.executionMetrics(id)
+ sparkStore
+ .planGraph(id)
+ .allNodes
+ .flatMap { node =>
+ node.metrics.map { metric =>
+ (
+ s"${node.id}: ${node.name}",
+ metric.name,
+ values.getOrElse(metric.accumulatorId, "Not available yet"))
+ }
+ }
+ .toVector
+ } catch {
+ case _: NoSuchElementException => Seq.empty
+ }
+ }
+
def buildInfo(): Option[AuronBuildInfoUIData] = {
val kClass = classOf[AuronBuildInfoUIData]
// KVStore throws when the record doesn't exist; treat missing data as "no build info".
@@ -43,3 +86,15 @@ class AuronBuildInfoUIData(val info: Seq[(String, String)]) {
@KVIndex
def id: String = classOf[AuronBuildInfoUIData].getName()
}
+
+/** Stored independently of Spark's SQL records so retention uses the matching record count. */
+class AuronExecutionUIData(
+ val executionId: Long,
+ val description: String,
+ val startedAt: Long,
+ val endedAt: Long,
+ val snapshots: Seq[AuronPlanDiagnosticEvent]) {
+ @JsonIgnore
+ @KVIndex
+ def id: Long = executionId
+}
diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala
index 08a4474b..398f04c9 100644
--- a/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala
+++ b/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala
@@ -20,7 +20,7 @@ import java.io.File
import org.apache.spark.SparkConf
import org.apache.spark.sql.AuronQueryTest
-import org.apache.spark.sql.execution.ui.AuronSQLAppStatusListener
+import org.apache.spark.sql.execution.ui.{AuronSQLAppStatusListener, AuronSQLAppStatusStore}
import org.apache.spark.util.Utils
import org.apache.auron.BaseAuronSQLSuite
@@ -39,7 +39,8 @@ class BuildInfoInSparkUISuite extends AuronQueryTest with BaseAuronSQLSuite {
}
override def afterAll(): Unit = {
- Utils.deleteRecursively(testDir)
+ try super.afterAll()
+ finally Utils.deleteRecursively(testDir)
}
test("test build info in spark UI ") {
@@ -50,4 +51,52 @@ class BuildInfoInSparkUISuite extends AuronQueryTest with BaseAuronSQLSuite {
assert(listener.getAuronBuildInfo() == 1)
}
+ test("native conversion publishes diagnostics to the UI store") {
+ withTable("auron_ui_diagnostics") {
+ sql("create table auron_ui_diagnostics (value int) using parquet")
+ sql("insert into auron_ui_diagnostics values (1), (2)")
+ sql("select sum(value) from auron_ui_diagnostics").collect()
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ val store = new AuronSQLAppStatusStore(spark.sparkContext.statusStore.store)
+ val executions = store.executions(0, 100)
+ val snapshots = executions.flatMap(_.snapshots)
+ assert(executions.exists(e => store.metrics(e.executionId).nonEmpty))
+ assert(snapshots.nonEmpty)
+ assert(snapshots.exists(_.nodes.exists(_.status == "Native")))
+ assert(snapshots.forall(_.physicalPlan.nonEmpty))
+ assert(snapshots.exists(_.configurations.contains("spark.sql.session.timeZone")))
+ assert(snapshots.exists(_.configurations.contains("spark.auron.memoryFraction")))
+ assert(
+ snapshots.forall(
+ _.configurations
+ .get("spark.auron.enable.data.writing")
+ .contains("false")))
+ assert(
+ snapshots.forall(_.configurations.contains("spark.auron.enable.data.writing.parquet")))
+ assert(snapshots.forall(_.configurations.contains("spark.auron.enable.data.writing.orc")))
+
+ val beforeFallback = store.executions(0, 100).map(_.executionId).toSet
+ spark.udf.register("auron_ui_passthrough", (value: Int) => value)
+ sql("select auron_ui_passthrough(value) from auron_ui_diagnostics").collect()
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ val fallbackSnapshots = store
+ .executions(0, 100)
+ .filterNot(e => beforeFallback.contains(e.executionId))
+ .flatMap(_.snapshots)
+ assert(fallbackSnapshots.exists(_.nodes.exists(n =>
+ n.status == "Fallback" && n.reason.nonEmpty)))
+
+ val beforeDisabled = store.executions(0, 100).map(_.executionId).toSet
+ withSQLConf("spark.auron.enabled" -> "false") {
+ sql("select sum(value) from auron_ui_diagnostics").collect()
+ }
+ spark.sparkContext.listenerBus.waitUntilEmpty()
+ assert(
+ store
+ .executions(0, 100)
+ .filterNot(e => beforeDisabled.contains(e.executionId))
+ .forall(_.snapshots.isEmpty))
+ }
+ }
+
}
diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsSuite.scala
new file mode 100644
index 00000000..eccf2c0a
--- /dev/null
+++ b/spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/ui/AuronDiagnosticsSuite.scala
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.spark.sql.execution.ui
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.sql.auron.{AuronConvertStrategy, AuronPlanDiagnostics, ForceNativeExecutionWrapper}
+import org.apache.spark.sql.execution.{LocalTableScanExec, UnionExec}
+import org.apache.spark.status.ElementTrackingStore
+import org.apache.spark.util.JsonProtocol
+import org.apache.spark.util.kvstore.InMemoryStore
+
+import org.apache.auron.spark.ui.{AuronPlanDiagnosticEvent, AuronPlanDiagnosticNode}
+import org.apache.auron.sparkver
+
+class AuronDiagnosticsSuite extends SparkFunSuite {
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3")
+ private def writeEvent(event: org.apache.spark.scheduler.SparkListenerEvent): String =
+ org.json4s.jackson.JsonMethods.compact(JsonProtocol.sparkEventToJson(event))
+
+ @sparkver("3.4 / 3.5 / 4.0 / 4.1 / 4.2")
+ private def writeEvent(event: org.apache.spark.scheduler.SparkListenerEvent): String =
+ JsonProtocol.sparkEventToJsonString(event)
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3")
+ private def readEvent(json: String): org.apache.spark.scheduler.SparkListenerEvent =
+ JsonProtocol.sparkEventFromJson(org.json4s.jackson.JsonMethods.parse(json))
+
+ @sparkver("3.4 / 3.5 / 4.0 / 4.1 / 4.2")
+ private def readEvent(json: String): org.apache.spark.scheduler.SparkListenerEvent =
+ JsonProtocol.sparkEventFromJson(json)
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def emptyScan(): LocalTableScanExec = LocalTableScanExec(Nil, Nil)
+
+ @sparkver("4.0 / 4.1 / 4.2")
+ private def emptyScan(): LocalTableScanExec = LocalTableScanExec(Nil, Nil, None)
+
+ private def snapshot(id: Long, index: Int = 0): AuronPlanDiagnosticEvent =
+ AuronPlanDiagnosticEvent(
+ id,
+ index.toLong,
+ s"plan $index",
+ Seq(AuronPlanDiagnosticNode(1, "Project", "Fallback", "unsupported expression")),
+ Map("spark.sql.session.timeZone" -> "UTC"))
+
+ test("diagnostic events survive JSON replay and out-of-order lifecycle events") {
+ val conf = new SparkConf(false)
+ val kv = new ElementTrackingStore(new InMemoryStore, conf)
+ try {
+ val listener = new AuronSQLAppStatusListener(conf, kv)
+ val store = new AuronSQLAppStatusStore(kv)
+ val event = snapshot(7)
+ val replayed = readEvent(writeEvent(event))
+ assert(replayed == event)
+ listener.onOtherEvent(replayed)
+ listener.onOtherEvent(SparkListenerSQLExecutionEnd(7, 200))
+ // This also checks that a late start does not erase an end or plan snapshot.
+ listener.onOtherEvent(
+ readEvent("""{"Event":"org.apache.spark.sql.execution.ui.SparkListenerSQLExecutionStart",
+ "executionId":7,"description":"query","details":"","physicalPlanDescription":"",
+ "time":100,"modifiedConfigs":{},"jobTags":[]}"""))
+ val stored = store.execution(7).get
+ assert(stored.description == "query")
+ assert(stored.startedAt == 100 && stored.endedAt == 200)
+ assert(stored.snapshots == Seq(event))
+ assert(stored.snapshots.head.nodes.head.reason == "unsupported expression")
+ } finally kv.close()
+ }
+
+ test("bounded snapshots preserve AQE updates and pagination reads the requested page") {
+ val conf = new SparkConf(false)
+ val kv = new ElementTrackingStore(new InMemoryStore, conf)
+ try {
+ val listener = new AuronSQLAppStatusListener(conf, kv)
+ val store = new AuronSQLAppStatusStore(kv)
+ (0 until 25).foreach(i => listener.onOtherEvent(snapshot(1, i)))
+ listener.onOtherEvent(snapshot(1, 24))
+ assert(store.execution(1).get.snapshots.size == 20)
+ assert(store.execution(1).get.snapshots.head.physicalPlan == "plan 5")
+ (2L to 5L).foreach(id => listener.onOtherEvent(snapshot(id)))
+ assert(store.executionsCount() == 5)
+ assert(store.executions(1, 2).map(_.executionId) == Seq(4L, 3L))
+ assert(store.execution(99).isEmpty)
+ assert(store.metrics(99).isEmpty)
+ } finally kv.close()
+ }
+
+ test("shared plan instances are visited once and operator tags are left unchanged") {
+ val leaf = emptyScan()
+ val tag = org.apache.spark.sql.catalyst.trees.TreeNodeTag[Int]("operatorId")
+ leaf.setTagValue(tag, 1234)
+ leaf.setTagValue(AuronConvertStrategy.neverConvertReasonTag, "test fallback")
+ val plan = UnionExec(Seq(leaf, leaf))
+ val nodes = AuronPlanDiagnostics.collect(plan)
+ assert(nodes.size == 2)
+ assert(nodes.count(_.reason == "test fallback") == 1)
+ assert(leaf.getTagValue(tag).contains(1234))
+ }
+
+ test("retention deletes completed diagnostics but preserves active executions") {
+ val conf = new SparkConf(false)
+ .set("spark.sql.ui.retainedExecutions", "2")
+ .set("spark.appStateStore.asyncTracking.enable", "false")
+ val kv = new ElementTrackingStore(new InMemoryStore, conf)
+ try {
+ val listener = new AuronSQLAppStatusListener(conf, kv)
+ val store = new AuronSQLAppStatusStore(kv)
+ listener.onOtherEvent(snapshot(0))
+ (1L to 4L).foreach { id =>
+ listener.onOtherEvent(snapshot(id))
+ listener.onOtherEvent(SparkListenerSQLExecutionEnd(id, 100))
+ }
+ assert(store.execution(0).isDefined)
+ assert(store.executionsCount() == 2)
+ assert(store.execution(4).isDefined)
+ } finally kv.close()
+ }
+
+ test("diagnostic HTML escapes plan content and accepts malformed navigation parameters") {
+ val conf = new SparkConf(false)
+ val kv = new ElementTrackingStore(new InMemoryStore, conf)
+ try {
+ val listener = new AuronSQLAppStatusListener(conf, kv)
+ listener.onOtherEvent(
+ snapshot(1).copy(
+ physicalPlan = "<script>alert(1)</script>",
+ nodes = Seq(
+ AuronPlanDiagnosticNode(1, "AdaptiveSparkPlan", "Wrapper", "old unsupported reason"),
+ AuronPlanDiagnosticNode(2, "Range", "Spark (no reason recorded)", ""))))
+ val view = new AuronDiagnosticsView(new AuronSQLAppStatusStore(kv))
+ val details = view.render("1", null, "/history/app-1").toString
+ assert(details.contains("<script>"))
+ assert(!details.contains("<script>"))
+ assert(details.contains("0 native, 0 fallback, 1 other Spark, 1 wrappers"))
+ assert(details.contains("Planning/execution wrapper; excluded from conversion counts."))
+ assert(!details.contains("old unsupported reason"))
+ assert(details.contains("/history/app-1/SQL/execution/?id=1"))
+ assert(view.render(null, "invalid", "").text.contains("Page 1"))
+ assert(view.render(null, "-12", "").text.contains("Page 1"))
+ assert(view.render("bad", null, "").text.contains("unavailable"))
+ } finally kv.close()
+ }
+
+ test("execution records retain nested diagnostic data through KVStore serialization") {
+ val serializer = new org.apache.spark.status.KVUtils.KVStoreScalaSerializer
+ val record = new AuronExecutionUIData(5, "query", 10, 20, Seq(snapshot(5)))
+ val restored =
+ serializer.deserialize(serializer.serialize(record), classOf[AuronExecutionUIData])
+ assert(restored.executionId == 5)
+ assert(restored.snapshots == record.snapshots)
+ assert(restored.description == record.description)
+ }
+
+ test("reused exchanges expose their child without counting a shared subtree twice") {
+ val leaf = emptyScan()
+ val exchange = org.apache.spark.sql.execution.exchange.BroadcastExchangeExec(
+ org.apache.spark.sql.catalyst.plans.physical.IdentityBroadcastMode,
+ leaf)
+ val reused = org.apache.spark.sql.execution.exchange.ReusedExchangeExec(Nil, exchange)
+ reused.setTagValue(AuronConvertStrategy.neverConvertReasonTag, "unsupported wrapper")
+ val nodes = AuronPlanDiagnostics.collect(UnionExec(Seq(reused, exchange)))
+ assert(nodes.size == 4)
+ assert(nodes.count(_.name == leaf.nodeName) == 1)
+ assert(nodes.find(_.name == reused.nodeName).get.status == "Wrapper")
+ assert(nodes.find(_.name == reused.nodeName).get.reason.isEmpty)
+ assert(
+ reused
+ .getTagValue(AuronConvertStrategy.neverConvertReasonTag)
+ .contains("unsupported wrapper"))
+ }
+
+ test(
+ "Auron execution wrappers are excluded from native counts and preserve child diagnostics") {
+ val leaf = emptyScan()
+ leaf.setTagValue(AuronConvertStrategy.neverConvertReasonTag, "test fallback")
+ val native = org.apache.spark.sql.execution.auron.plan.NativeUnionExec(Seq(leaf), Nil)
+ val wrapper = ForceNativeExecutionWrapper(native)
+ val nodes = AuronPlanDiagnostics.collect(UnionExec(Seq(wrapper, native)))
+ assert(nodes.size == 4)
+ assert(nodes.count(_.status == "Native") == 1)
+ assert(nodes.count(_.status == "Wrapper") == 1)
+ assert(
+ nodes
+ .find(_.name == wrapper.nodeName)
+ .exists(n => n.status == "Wrapper" && n.reason.isEmpty))
+ assert(nodes.count(_.reason == "test fallback") == 1)
+ }
+
+}
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronPlanDiagnostics.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronPlanDiagnostics.scala
new file mode 100644
index 00000000..91dbfdf2
--- /dev/null
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronPlanDiagnostics.scala
@@ -0,0 +1,140 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.spark.sql.auron
+
+import java.util.IdentityHashMap
+
+import scala.collection.mutable.ArrayBuffer
+import scala.util.control.NonFatal
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.{InputAdapter, ReusedSubqueryExec, SparkPlan, SQLExecution, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
+import org.apache.spark.sql.execution.ui.AuronEventUtils
+import org.apache.spark.util.Utils
+
+import org.apache.auron.configuration.AuronConfiguration
+import org.apache.auron.spark.configuration.SparkAuronConfiguration
+import org.apache.auron.spark.ui.{AuronPlanDiagnosticEvent, AuronPlanDiagnosticNode}
+
+/** Diagnostics must neither mutate Spark's operator IDs nor prevent query execution. */
+object AuronPlanDiagnostics extends Logging {
+ private val configurationKeys = Seq(
+ "spark.auron.enabled",
+ "spark.auron.enable",
+ "spark.auron.parquet.enable.pageFiltering",
+ "spark.auron.parquet.enable.bloomFilter",
+ "spark.auron.process.vmrss.memoryFraction",
+ "spark.auron.ui.enabled",
+ "spark.shuffle.manager",
+ "spark.sql.adaptive.enabled",
+ "spark.sql.session.timeZone",
+ "spark.sql.ansi.enabled",
+ "spark.sql.shuffle.partitions",
+ "spark.memory.offHeap.enabled",
+ "spark.memory.offHeap.size")
+
+ def collect(plan: SparkPlan): Seq[AuronPlanDiagnosticNode] = {
+ val visited = new IdentityHashMap[SparkPlan, java.lang.Boolean]()
+ val nodes = ArrayBuffer.empty[AuronPlanDiagnosticNode]
+ def visit(p: SparkPlan): Unit = {
+ if (visited.put(p, java.lang.Boolean.TRUE) != null) {
+ return
+ }
+ val reason = p.getTagValue(AuronConvertStrategy.neverConvertReasonTag).getOrElse("")
+ val status = p match {
+ case _: AuronConverters.ForceNativeExecutionWrapperBase => "Wrapper"
+ case _: NativeSupports => "Native"
+ case _: AdaptiveSparkPlanExec | _: QueryStageExec | _: ReusedExchangeExec |
+ _: ReusedSubqueryExec | _: InputAdapter | _: WholeStageCodegenExec =>
+ "Wrapper"
+ case _ if reason.nonEmpty => "Fallback"
+ case _ => "Spark (no reason recorded)"
+ }
+ nodes += AuronPlanDiagnosticNode(
+ nodes.size + 1,
+ p.nodeName,
+ status,
+ if (status == "Wrapper") "" else reason)
+ p match {
+ case a: AdaptiveSparkPlanExec => visit(a.executedPlan)
+ case q: QueryStageExec => visit(q.plan)
+ case r: ReusedExchangeExec => visit(r.child)
+ case r: ReusedSubqueryExec => visit(r.child)
+ case _ => p.children.foreach(visit)
+ }
+ p.innerChildren.foreach {
+ case child: SparkPlan => visit(child)
+ case _ =>
+ }
+ }
+ visit(plan)
+ nodes.toVector
+ }
+
+ def publish(session: SparkSession, plan: SparkPlan): Unit = {
+ if (!SparkAuronConfiguration.AURON_ENABLED.get() ||
+ !SparkAuronConfiguration.UI_ENABLED.get()) {
+ return
+ }
+ val sc = session.sparkContext
+ Option(sc.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)).foreach { id =>
+ try {
+ // Only non-secret diagnostic settings are captured, never arbitrary session options.
+ val effectiveAuronSettings = Seq(
+ SparkAuronConfiguration.AURON_ENABLED,
+ SparkAuronConfiguration.UI_ENABLED,
+ SparkAuronConfiguration.ENABLE_SHUFFLE_EXCHANGE,
+ SparkAuronConfiguration.ENABLE_SCAN_PARQUET,
+ SparkAuronConfiguration.ENABLE_DATA_WRITING,
+ SparkAuronConfiguration.ENABLE_DATA_WRITING_PARQUET,
+ SparkAuronConfiguration.ENABLE_DATA_WRITING_ORC,
+ SparkAuronConfiguration.PARQUET_ENABLE_PAGE_FILTERING,
+ SparkAuronConfiguration.PARQUET_ENABLE_BLOOM_FILTER,
+ SparkAuronConfiguration.PROCESS_MEMORY_FRACTION,
+ SparkAuronConfiguration.SPILL_COMPRESSION_CODEC,
+ SparkAuronConfiguration.RSS_SPILL_MEMORY_FRACTION,
+ SparkAuronConfiguration.RSS_SPILL_MEMORY_SIZE,
+ SparkAuronConfiguration.ON_HEAP_SPILL_MEM_FRACTION,
+ SparkAuronConfiguration.ORC_BATCH_SIZE,
+ AuronConfiguration.BATCH_SIZE,
+ AuronConfiguration.MEMORY_FRACTION,
+ AuronConfiguration.SUGGESTED_BATCH_MEM_SIZE)
+ .map(option => s"spark.${option.key}" -> option.get().toString)
+ .toMap
+ val settings = configurationKeys.flatMap { key =>
+ session.conf.getOption(key).orElse(sc.getConf.getOption(key)).map(key -> _)
+ }.toMap ++ effectiveAuronSettings
+ val redaction = session.sessionState.conf.stringRedactionPattern
+ val nodes =
+ collect(plan).map(n => n.copy(reason = Utils.redact(redaction, n.reason).take(8192)))
+ AuronEventUtils.post(
+ sc,
+ AuronPlanDiagnosticEvent(
+ id.toLong,
+ System.currentTimeMillis(),
+ Utils.redact(redaction, plan.treeString).take(65536),
+ nodes,
+ settings))
+ } catch {
+ case NonFatal(e) => logWarning("Unable to record Auron plan diagnostics", e)
+ }
+ }
+ }
+}
diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
index 77228475..85e65e00 100644
--- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
+++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala
@@ -101,4 +101,12 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
}
}
}
+
+ override def postColumnarTransitions: Rule[SparkPlan] = new Rule[SparkPlan] {
+ override def apply(plan: SparkPlan): SparkPlan = {
+ AuronPlanDiagnostics.publish(sparkSession, plan)
+ plan
+ }
+ }
+
}