You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@auron.apache.org by "guixiaowen (via GitHub)" <gi...@apache.org> on 2026/04/01 15:40:40 UTC
[PR] [AURON #1471] Follow #1409 Add the conversion execution list in the S… [auron]
guixiaowen opened a new pull request, #2150:
URL: https://github.com/apache/auron/pull/2150
…park UI.
# Which issue does this PR close?
Closes #<issue_number>
# Rationale for this change
# What changes are included in this PR?
# Are there any user-facing changes?
# How was this patch tested?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-5964219419
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3216116376
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
Review Comment:
`executionPage` is parsed with `_.toInt` outside the `try` block, so a non-integer `auron.page` query parameter will throw `NumberFormatException` and break the whole page render. Parse these request params defensively (e.g., `toIntOption` or catch `NumberFormatException`) and fall back to a safe default.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala:
##########
@@ -36,8 +39,41 @@ class AuronSQLAppStatusStore(store: KVStore) extends Logging {
None
}
}
+
+ private def viewToSeq[T](view: KVStoreView[T]): Seq[T] = {
+ Utils.tryWithResource(view.closeableIterator())(iter => iter.asScala.toList)
+ }
+
+ def executionsList(): Seq[AuronSQLExecutionUIData] = {
+ viewToSeq(store.view(classOf[AuronSQLExecutionUIData]))
+ }
+
+ def executionsList(offset: Int, length: Int): Seq[AuronSQLExecutionUIData] = {
+ viewToSeq(store.view(classOf[AuronSQLExecutionUIData]).skip(offset).max(length))
+ }
+
+ def execution(executionId: Long): Option[AuronSQLExecutionUIData] = {
+ try {
+ Some(store.read(classOf[AuronSQLExecutionUIData], executionId))
+ } catch {
+ case _: NoSuchElementException => None
+ }
+ }
+
+ def executionsCount(): Long = {
+ store.count(classOf[AuronSQLExecutionUIData])
+ }
}
+@KVIndex("executionId")
+class AuronSQLExecutionUIData(
+ @KVIndexParam val executionId: Long,
+ val description: String,
+ val numAuronNodes: Int,
+ val numFallbackNodes: Int,
+ val fallbackDescription: String,
+ val fallbackNodeToReason: Seq[(String, String)]) {}
Review Comment:
`@KVIndex("executionId")` is applied to the class, but Spark's `KVIndex` annotation is intended for indexed *fields/getters* (and `@KVIndexParam` already defines the key). This is likely to fail compilation with "annotation type not applicable". Remove the class-level `@KVIndex` and, if you need an index, annotate an appropriate getter/field instead.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
Review Comment:
The table header links are built with `scala.xml.Unparsed` using raw request parameters (`parameterPath` comes from `request.getParameterMap`). This bypasses XML escaping and can allow HTML/JS injection (XSS) via crafted query parameters. Build `href` values as normal strings (let XML escape) and URL-encode parameter names/values instead of using `Unparsed`.
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala:
##########
@@ -99,4 +100,36 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
}
}
}
+
+ override def postColumnarTransitions: Rule[SparkPlan] = {
+ new Rule[SparkPlan] {
+ override def apply(sparkPlan: SparkPlan): SparkPlan = {
+ if (SparkEnv.get.conf
+ .get(SparkAuronConfiguration.UI_ENABLED.key, "true")
+ .equals("true")) {
Review Comment:
`postColumnarTransitions` always traverses and stringifies the full SparkPlan when `spark.auron.ui.enabled` is true, even when `spark.auron.enabled` is false (in which case no conversion happens). This adds avoidable overhead to every query. Consider also gating this block on `SparkAuronConfiguration.AURON_ENABLED.get()` and using the typed boolean accessor for `UI_ENABLED` instead of a string equality check.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>
+ {concat.toString()}
+ </pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}
+ </a>
+ } else {
+ <a href={executionURL(execution.executionId)}>
+ {execution.executionId}
+ </a>
+ }
+
+ <div>
+ {desc}{details}
+ </div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+ }
+
+ @sparkver("4.0 / 4.1")
+ private[ui] class AuronExecutionPagedTable40(
+ request: jakarta.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: jakarta.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: jakarta.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>{concat.toString()}</pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}</a>
+ } else {
+ <a href={executionURL(execution.executionId)}>{execution.executionId}</a>
+ }
+
+ <div>{desc}{details}</div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+ }
+}
+
+private[ui] class AuronExecutionTableRowData(val executionUIData: AuronSQLExecutionUIData)
+
+private[ui] class AuronExecutionDataSource(
+ executionData: Seq[AuronSQLExecutionUIData],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean)
+ extends PagedDataSource[AuronExecutionTableRowData](pageSize) {
+
+ // Convert ExecutionData to ExecutionTableRowData which contains the final contents to show
+ // in the table so that we can avoid creating duplicate contents during sorting the data
+ private val data = executionData.map(executionRow).sorted(ordering(sortColumn, desc))
+
+ override def dataSize: Int = data.size
+
+ override def sliceData(from: Int, to: Int): Seq[AuronExecutionTableRowData] =
+ data.slice(from, to)
+
+ private def executionRow(
+ executionUIData: AuronSQLExecutionUIData): AuronExecutionTableRowData = {
+ new AuronExecutionTableRowData(executionUIData)
+ }
+
+ /** Return Ordering according to sortColumn and desc. */
+ private def ordering(
+ sortColumn: String,
+ desc: Boolean): Ordering[AuronExecutionTableRowData] = {
+ val ordering: Ordering[AuronExecutionTableRowData] = sortColumn match {
+ case "ID" => Ordering.by(_.executionUIData.executionId)
+ case "Description" => Ordering.by(_.executionUIData.fallbackDescription)
Review Comment:
Sorting for the "Description" column uses `executionUIData.fallbackDescription`, but the rendered description text comes from `executionUIData.description`. This makes sorting appear incorrect and can be very expensive if `fallbackDescription` contains the full physical plan. Sort by the same field that is displayed in the column.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -29,12 +36,22 @@ 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()) ++
+ buildExecutionsListSummary(sqlStore.executionsList(), request),
+ parent)
Review Comment:
The page always loads `sqlStore.executionsList()` (all retained executions) and then sorts/paginates in memory. With `UI_RETAINED_EXECUTIONS` set high, this can make the Auron tab slow and memory-heavy. Consider paging directly from KVStore using `executionsCount()` + `executionsList(offset, length)` (and/or KVStore sorting) instead of materializing the full list each render.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
Review Comment:
The fallback node formatting/parsing relies on fixed-width substrings (`substring(0, 3)` / `substring(4)`), which breaks once operator ids exceed 999 (e.g. "1000 ..."). Parse the id/name by splitting on the first space (or store id and nodeName separately) instead of assuming 3 digits.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala:
##########
@@ -40,10 +47,61 @@ class AuronSQLAppStatusListener(conf: SparkConf, kvstore: ElementTrackingStore)
}
override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case e: SparkListenerSQLExecutionStart => onSQLExecutionStart(e)
+ case e: SparkListenerSQLExecutionEnd => onSQLExtensionEnd(e)
case e: AuronBuildInfoEvent => onAuronBuildInfo(e)
+ case e: AuronPlanFallbackEvent => onAuronPlanFallback(e)
case _ => // Ignore
}
+ private def onAuronPlanFallback(event: AuronPlanFallbackEvent): Unit = {
+ val description = executionIdToDescription.get(event.executionId)
+ if (description.isDefined) {
+ val uiData = new AuronSQLExecutionUIData(
+ event.executionId,
+ description.get,
+ event.numAuronNodes,
+ event.numFallbackNodes,
+ event.physicalPlanDescription,
+ event.fallbackNodeToReason.toSeq.sortBy(_._1))
Review Comment:
`AuronSQLExecutionUIData` stores `event.physicalPlanDescription` into a field named `fallbackDescription`, which is confusing given the UI also has a fallback summary. Renaming this field (and corresponding UI references) to something like `physicalPlanDescription` will make the persisted data schema and UI code much easier to follow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "guixiaowen (via GitHub)" <gi...@apache.org>.
guixiaowen commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4332590888
@ShreyeshArangath sh Would you mind helping to review the code? I’d really appreciate it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "cxzl25 (via GitHub)" <gi...@apache.org>.
cxzl25 closed pull request #2150: [AURON #1471] Add the conversion execution list in the SparkUI
URL: https://github.com/apache/auron/pull/2150
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870683
##########
spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala:
##########
@@ -49,4 +49,13 @@ class BuildInfoInSparkUISuite extends AuronQueryTest with BaseAuronSQLSuite {
assert(listener.getAuronBuildInfo() == 1)
}
+ test("test convert table in spark UI ") {
+ withTable("t1") {
+ sql(
+ "create table t1 using parquet PARTITIONED BY (part) as select 1 as c1, 2 as c2, 'test test' as part")
+ val df = sql("select * from t1")
+ checkAnswer(df, Seq(Row(1, 2, "test test")))
+ }
+ }
Review Comment:
This test executes a query but doesn't assert that any Spark UI conversion/execution data was produced (so it can pass even if the new feature is broken). Consider asserting that AuronSQLExecutionUIData entries are written to the status KVStore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "guixiaowen (via GitHub)" <gi...@apache.org>.
guixiaowen commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4787026780
> @guixiaowen did you get a chance to address some of my comments?
@ShreyeshArangath Thank you very much for the code review. I've been a bit busy these past couple of days, but I will complete the revisions as soon as possible. Thank you again for reviewing the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3025617354
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +95,376 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+}
+
+private[ui] class AuronExecutionPagedTable(
+ request: HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) = getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(request: HttpServletRequest, tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>{concat.toString()}</pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}</a>
+ } else {
+ <a href={executionURL(execution.executionId)}>{execution.executionId}</a>
+ }
+
+ <div>{desc}{details}</div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+}
+
+private[ui] class AuronExecutionTableRowData(val executionUIData: AuronSQLExecutionUIData)
+
+private[ui] class AuronExecutionDataSource(
+ executionData: Seq[AuronSQLExecutionUIData],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean)
+ extends PagedDataSource[AuronExecutionTableRowData](pageSize) {
+
+ // Convert ExecutionData to ExecutionTableRowData which contains the final contents to show
+ // in the table so that we can avoid creating duplicate contents during sorting the data
+ private val data = executionData.map(executionRow).sorted(ordering(sortColumn, desc))
+
+ override def dataSize: Int = data.size
+
+ override def sliceData(from: Int, to: Int): Seq[AuronExecutionTableRowData] =
+ data.slice(from, to)
+
+ private def executionRow(
+ executionUIData: AuronSQLExecutionUIData): AuronExecutionTableRowData = {
+ new AuronExecutionTableRowData(executionUIData)
+ }
+
+ /** Return Ordering according to sortColumn and desc. */
+ private def ordering(
+ sortColumn: String,
+ desc: Boolean): Ordering[AuronExecutionTableRowData] = {
+ val ordering: Ordering[AuronExecutionTableRowData] = sortColumn match {
+ case "ID" => Ordering.by(_.executionUIData.executionId)
+ case "Description" => Ordering.by(_.executionUIData.fallbackDescription)
+ case "Num Auron Nodes" => Ordering.by(_.executionUIData.numAuronNodes)
+ case "Num Fallback Nodes" => Ordering.by(_.executionUIData.numFallbackNodes)
+ case unknownColumn => throw new IllegalArgumentException(s"Unknown column: $unknownColumn")
+ }
Review Comment:
Sorting by the “Description” column uses `fallbackDescription` (physical plan text) rather than `executionUIData.description`. This makes sorting behavior inconsistent with the displayed column label/content. Update the ordering for `"Description"` to use the `description` field.
##########
spark-extension-shims-spark/src/test/scala/org/apache/spark/sql/execution/BuildInfoInSparkUISuite.scala:
##########
@@ -49,4 +49,13 @@ class BuildInfoInSparkUISuite extends AuronQueryTest with BaseAuronSQLSuite {
assert(listener.getAuronBuildInfo() == 1)
}
+ test("test convert table in spark UI ") {
+ withTable("t1") {
+ sql(
+ "create table t1 using parquet PARTITIONED BY (part) as select 1 as c1, 2 as c2, 'test test' as part")
+ val df = sql("select * from t1")
+ checkAnswer(df, Seq(Row(1, 2, "test test")))
+ }
+ }
Review Comment:
This test exercises table creation/querying but does not assert the new Spark UI functionality (that `AuronPlanFallbackEvent` is posted and persisted as `AuronSQLExecutionUIData`, and that counts/reasons/plan text are stored). Add assertions against the `AuronSQLAppStatusListener`/KVStore (or `AuronSQLAppStatusStore`) to verify at least one execution record is written with expected fields (executionId, num nodes, fallback map/plan description).
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala:
##########
@@ -99,4 +100,36 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
}
}
}
+
+ override def postColumnarTransitions: Rule[SparkPlan] = {
+ new Rule[SparkPlan] {
+ override def apply(sparkPlan: SparkPlan): SparkPlan = {
+ if (SparkEnv.get.conf
+ .get(SparkAuronConfiguration.UI_ENABLED.key, "true")
+ .equals("true")) {
+ val sc = sparkSession.sparkContext
+ val executionId = sc.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+ if (executionId == null) {
+ logDebug(s"Unknown execution id for plan: $sparkPlan")
+ return sparkPlan
+ }
+ val concat = new PlanStringConcat()
+ concat.append("== Physical Plan ==\n")
+
+ val (numAuronNodes, fallbackNodeToReason) =
+ AuronExplainUtils.processPlan(sparkPlan, concat.append)
+
+ val event = AuronPlanFallbackEvent(
+ executionId.toLong,
+ numAuronNodes,
+ fallbackNodeToReason.size,
+ concat.toString(),
+ fallbackNodeToReason)
Review Comment:
Two issues here can cause unexpected failures/behavior: (1) UI enablement is checked via a case-sensitive string comparison; prefer a boolean conf read (or a safer parse) so values like `TRUE` / `true` behave consistently. (2) `executionId.toLong` can throw if the local property is non-numeric; guard the parse and skip posting the event (with debug logging) if it cannot be parsed.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala:
##########
@@ -16,11 +16,14 @@
*/
package org.apache.spark.sql.execution.ui
+import scala.jdk.CollectionConverters.asScalaIteratorConverter
Review Comment:
`scala.jdk.CollectionConverters` is Scala 2.13+ only. This repo appears to support Spark 3.x as well (typically Scala 2.12), so this import can break compilation for Spark 3 builds. Prefer `scala.collection.JavaConverters._` / `JavaConverters.asScalaIteratorConverter` in Scala 2.12, or introduce a small compatibility shim (or Spark-version-specific source) to select the right converters per build target.
```suggestion
import scala.collection.JavaConverters._
```
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -29,12 +37,22 @@ 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()) ++
+ buildExecutionsListSummary(sqlStore.executionsList(), request),
+ parent)
Review Comment:
The page currently calls `sqlStore.executionsList()` and passes the full result into the table, and the datasource later sorts that full in-memory sequence. For applications with many SQL executions this can make the Auron tab slow and memory-heavy. Prefer a KVStore-backed paged datasource: use `executionsCount()` for the total, and fetch only the requested page via `executionsList(offset, length)` (and ideally push sorting down to the KVStore view when possible), mirroring Spark’s own UI pattern.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +95,376 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+}
+
+private[ui] class AuronExecutionPagedTable(
+ request: HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) = getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(request: HttpServletRequest, tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
Review Comment:
Query parameter values are concatenated into URLs without URL-encoding. This can break navigation when values contain `&`, `=`, spaces, or other reserved characters. Encode both keys and values (or use a Spark UI helper if available) when rebuilding the query string.
```suggestion
.map { case (key, values) =>
val encodedKey = URLEncoder.encode(key, UTF_8.name())
val value = if (values != null && values.nonEmpty) values(0) else ""
val encodedValue = URLEncoder.encode(value, UTF_8.name())
s"$encodedKey=$encodedValue"
}
```
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +95,376 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+}
+
+private[ui] class AuronExecutionPagedTable(
+ request: HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) = getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(request: HttpServletRequest, tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>{concat.toString()}</pre>
+ </div>
+ } else {
+ Nil
+ }
Review Comment:
The “+details” expansion is currently hidden whenever `execution.description` is empty/null, even though fallback details (reasons/plan text) may still exist and would be useful. Gate `details` on the presence of fallback/plan content (e.g., `fallbackDescription` / `fallbackNodeToReason.nonEmpty`) instead of `description`.
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronExplainUtils.scala:
##########
@@ -0,0 +1,268 @@
+/*
+ * 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.Collections.newSetFromMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.auron.AuronConvertStrategy.neverConvertReasonTag
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.{BaseSubqueryExec, InputAdapter, ReusedSubqueryExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.ExplainUtils.getOpId
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.command.ExecutedCommandExec
+import org.apache.spark.sql.execution.exchange.{Exchange, ReusedExchangeExec}
+
+import org.apache.auron.sparkver
+
+object AuronExplainUtils {
+ private def generateOperatorIDs(
+ plan: QueryPlan[_],
+ startOperatorID: Int,
+ visited: java.util.Set[QueryPlan[_]],
+ reusedExchanges: ArrayBuffer[ReusedExchangeExec],
+ addReusedExchanges: Boolean): Int = {
+ var currentOperationID = startOperatorID
+ if (plan.isInstanceOf[BaseSubqueryExec]) {
+ return currentOperationID
+ }
+
+ def setOpId(plan: QueryPlan[_]): Unit = if (!visited.contains(plan)) {
+ plan match {
+ case r: ReusedExchangeExec if addReusedExchanges =>
+ reusedExchanges.append(r)
+ case _ =>
+ }
+ visited.add(plan)
+ currentOperationID += 1
+ plan.setTagValue(TreeNodeTag[Int]("operatorId"), currentOperationID)
+ }
+
+ plan.foreachUp {
+ case _: WholeStageCodegenExec =>
+ case _: InputAdapter =>
+ case p: AdaptiveSparkPlanExec =>
+ currentOperationID = generateOperatorIDs(
+ p.executedPlan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case p: QueryStageExec =>
+ currentOperationID = generateOperatorIDs(
+ p.plan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case other: QueryPlan[_] =>
+ setOpId(other)
+ currentOperationID = other.innerChildren.foldLeft(currentOperationID) { (curId, plan) =>
+ generateOperatorIDs(plan, curId, visited, reusedExchanges, addReusedExchanges)
+ }
+ }
+ currentOperationID
+ }
+
+ private def getSubqueries(
+ plan: => QueryPlan[_],
+ subqueries: ArrayBuffer[(SparkPlan, Expression, BaseSubqueryExec)]): Unit = {
+ plan.foreach {
+ case a: AdaptiveSparkPlanExec =>
+ getSubqueries(a.executedPlan, subqueries)
+ case q: QueryStageExec =>
+ getSubqueries(q.plan, subqueries)
+ case p: SparkPlan =>
+ p.expressions.foreach(_.collect { case e: PlanExpression[_] =>
+ e.plan match {
+ case s: BaseSubqueryExec =>
+ subqueries += ((p, e, s))
+ getSubqueries(s, subqueries)
+ case _ =>
+ }
+ })
+ }
+ }
+
+ private def processPlanSkippingSubqueries[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit): Unit = {
+ try {
+
+ QueryPlan.append(plan, append, verbose = false, addSuffix = false, printOperatorId = true)
+
+ append("\n")
+ } catch {
+ case e: AnalysisException => append(e.toString)
+ }
+ }
+
+ private def collectFallbackNodes(plan: QueryPlan[_]): (Int, Map[String, String]) = {
+ var numAuronNodes = 0
+ val fallbackNodeToReason = new mutable.HashMap[String, String]
+
+ def collect(tmp: QueryPlan[_]): Unit = {
+ tmp.foreachUp {
+ case p: ExecutedCommandExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ case p: AdaptiveSparkPlanExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.executedPlan)
+ case p: QueryStageExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.plan)
+ case p: NativeSupports =>
+ numAuronNodes += 1
+ p.innerChildren.foreach(collect)
+ case p: SparkPlan =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ p.innerChildren.foreach(collect)
+ case _ =>
+ }
+ }
+
+ collect(plan)
Review Comment:
`collectFallbackNodes` traverses the plan multiple times because it uses `foreachUp` (which already walks the subtree) and also recursively calls `collect(...)` on children/subplans inside the match arms. This will over-count `numAuronNodes` and can duplicate fallback collection work (and can degrade significantly on large plans). Refactor to do a single traversal (e.g., one `foreachUp`/`foreach` walk) and handle adaptive/query-stage subplans without re-traversing already-visited nodes (use a visited identity set if needed).
```suggestion
// Use an identity-based visited set to avoid re-traversing shared subplans
val visited =
newSetFromMap[QueryPlan[_]](
new java.util.IdentityHashMap[QueryPlan[_], java.lang.Boolean]()
)
def traverse(root: QueryPlan[_]): Unit = {
// Skip traversal if we've already seen this plan instance
if (!visited.add(root)) {
return
}
root.foreachUp {
case p: ExecutedCommandExec =>
handleVanillaSparkPlan(p, fallbackNodeToReason)
case p: AdaptiveSparkPlanExec =>
handleVanillaSparkPlan(p, fallbackNodeToReason)
// Traverse the executed plan as a separate root, guarding with visited set
traverse(p.executedPlan)
case p: QueryStageExec =>
handleVanillaSparkPlan(p, fallbackNodeToReason)
// Traverse the underlying stage plan as a separate root, guarding with visited set
traverse(p.plan)
case p: NativeSupports =>
numAuronNodes += 1
case p: SparkPlan =>
handleVanillaSparkPlan(p, fallbackNodeToReason)
case _ =>
}
}
traverse(plan)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870662
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>
+ {concat.toString()}
+ </pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}
+ </a>
+ } else {
+ <a href={executionURL(execution.executionId)}>
+ {execution.executionId}
+ </a>
+ }
+
+ <div>
+ {desc}{details}
+ </div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+ }
+
+ @sparkver("4.0 / 4.1")
+ private[ui] class AuronExecutionPagedTable40(
+ request: jakarta.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: jakarta.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: jakarta.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>{concat.toString()}</pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}</a>
+ } else {
+ <a href={executionURL(execution.executionId)}>{execution.executionId}</a>
+ }
+
+ <div>{desc}{details}</div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+ }
+}
+
+private[ui] class AuronExecutionTableRowData(val executionUIData: AuronSQLExecutionUIData)
+
+private[ui] class AuronExecutionDataSource(
+ executionData: Seq[AuronSQLExecutionUIData],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean)
+ extends PagedDataSource[AuronExecutionTableRowData](pageSize) {
+
+ // Convert ExecutionData to ExecutionTableRowData which contains the final contents to show
+ // in the table so that we can avoid creating duplicate contents during sorting the data
+ private val data = executionData.map(executionRow).sorted(ordering(sortColumn, desc))
+
+ override def dataSize: Int = data.size
+
+ override def sliceData(from: Int, to: Int): Seq[AuronExecutionTableRowData] =
+ data.slice(from, to)
+
+ private def executionRow(
+ executionUIData: AuronSQLExecutionUIData): AuronExecutionTableRowData = {
+ new AuronExecutionTableRowData(executionUIData)
+ }
+
+ /** Return Ordering according to sortColumn and desc. */
+ private def ordering(
+ sortColumn: String,
+ desc: Boolean): Ordering[AuronExecutionTableRowData] = {
+ val ordering: Ordering[AuronExecutionTableRowData] = sortColumn match {
+ case "ID" => Ordering.by(_.executionUIData.executionId)
+ case "Description" => Ordering.by(_.executionUIData.fallbackDescription)
Review Comment:
The table column is labeled "Description", but sorting uses `fallbackDescription` (physical plan text) instead of the user-facing `description` shown in the column. This makes sorting behavior incorrect/unintuitive.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "guixiaowen (via GitHub)" <gi...@apache.org>.
guixiaowen commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4324470143
> Thanks for implementing this! Being able to see why stages weren't converted directly in the SparkUI is super helpful for debugging.
@yew1eb ye Thank you very much for helping review the code. I really appreciate it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870594
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
Review Comment:
Parsing the page number with `.toInt` can throw (e.g., `auron.page=abc`) and fail the whole page render. The try/catch below won't catch it because it happens before the `try` block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870462
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronSparkSessionExtension.scala:
##########
@@ -99,4 +100,36 @@ case class AuronColumnarOverrides(sparkSession: SparkSession) extends ColumnarRu
}
}
}
+
+ override def postColumnarTransitions: Rule[SparkPlan] = {
+ new Rule[SparkPlan] {
+ override def apply(sparkPlan: SparkPlan): SparkPlan = {
+ if (SparkEnv.get.conf
+ .get(SparkAuronConfiguration.UI_ENABLED.key, "true")
+ .equals("true")) {
Review Comment:
UI enablement is checked by reading SparkConf as a string and comparing to "true". This bypasses the existing boolean accessor (SparkAuronConfiguration.UI_ENABLED.get()) used elsewhere and can mis-handle non-canonical boolean values (e.g., "TRUE").
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "yew1eb (via GitHub)" <gi...@apache.org>.
yew1eb commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4321121558
Thanks for implementing this! Being able to see why stages weren't converted directly in the SparkUI is super helpful for debugging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870508
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronExplainUtils.scala:
##########
@@ -0,0 +1,268 @@
+/*
+ * 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.Collections.newSetFromMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.auron.AuronConvertStrategy.neverConvertReasonTag
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.{BaseSubqueryExec, InputAdapter, ReusedSubqueryExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.ExplainUtils.getOpId
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.command.ExecutedCommandExec
+import org.apache.spark.sql.execution.exchange.{Exchange, ReusedExchangeExec}
+
+import org.apache.auron.sparkver
+
+object AuronExplainUtils {
+ private def generateOperatorIDs(
+ plan: QueryPlan[_],
+ startOperatorID: Int,
+ visited: java.util.Set[QueryPlan[_]],
+ reusedExchanges: ArrayBuffer[ReusedExchangeExec],
+ addReusedExchanges: Boolean): Int = {
+ var currentOperationID = startOperatorID
+ if (plan.isInstanceOf[BaseSubqueryExec]) {
+ return currentOperationID
+ }
+
+ def setOpId(plan: QueryPlan[_]): Unit = if (!visited.contains(plan)) {
+ plan match {
+ case r: ReusedExchangeExec if addReusedExchanges =>
+ reusedExchanges.append(r)
+ case _ =>
+ }
+ visited.add(plan)
+ currentOperationID += 1
+ plan.setTagValue(TreeNodeTag[Int]("operatorId"), currentOperationID)
+ }
+
+ plan.foreachUp {
+ case _: WholeStageCodegenExec =>
+ case _: InputAdapter =>
+ case p: AdaptiveSparkPlanExec =>
+ currentOperationID = generateOperatorIDs(
+ p.executedPlan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case p: QueryStageExec =>
+ currentOperationID = generateOperatorIDs(
+ p.plan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case other: QueryPlan[_] =>
+ setOpId(other)
+ currentOperationID = other.innerChildren.foldLeft(currentOperationID) { (curId, plan) =>
+ generateOperatorIDs(plan, curId, visited, reusedExchanges, addReusedExchanges)
+ }
+ }
+ currentOperationID
+ }
+
+ private def getSubqueries(
+ plan: => QueryPlan[_],
+ subqueries: ArrayBuffer[(SparkPlan, Expression, BaseSubqueryExec)]): Unit = {
+ plan.foreach {
+ case a: AdaptiveSparkPlanExec =>
+ getSubqueries(a.executedPlan, subqueries)
+ case q: QueryStageExec =>
+ getSubqueries(q.plan, subqueries)
+ case p: SparkPlan =>
+ p.expressions.foreach(_.collect { case e: PlanExpression[_] =>
+ e.plan match {
+ case s: BaseSubqueryExec =>
+ subqueries += ((p, e, s))
+ getSubqueries(s, subqueries)
+ case _ =>
+ }
+ })
+ }
+ }
+
+ private def processPlanSkippingSubqueries[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit): Unit = {
+ try {
+
+ QueryPlan.append(plan, append, verbose = false, addSuffix = false, printOperatorId = true)
+
+ append("\n")
+ } catch {
+ case e: AnalysisException => append(e.toString)
+ }
+ }
+
+ private def collectFallbackNodes(plan: QueryPlan[_]): (Int, Map[String, String]) = {
+ var numAuronNodes = 0
+ val fallbackNodeToReason = new mutable.HashMap[String, String]
+
+ def collect(tmp: QueryPlan[_]): Unit = {
+ tmp.foreachUp {
+ case p: ExecutedCommandExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ case p: AdaptiveSparkPlanExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.executedPlan)
+ case p: QueryStageExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.plan)
+ case p: NativeSupports =>
+ numAuronNodes += 1
+ p.innerChildren.foreach(collect)
+ case p: SparkPlan =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ p.innerChildren.foreach(collect)
+ case _ =>
+ }
+ }
Review Comment:
collectFallbackNodes() double-traverses the plan (uses foreachUp and also recursively calls collect on children). This will over-count numAuronNodes and can repeatedly process the same nodes, producing incorrect UI metrics.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #2150: [AURON #1471] Add the conversion execution list in the SparkUI
URL: https://github.com/apache/auron/pull/2150
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "guixiaowen (via GitHub)" <gi...@apache.org>.
guixiaowen commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4510395116
@cxzl25 Would you mind helping to review the code? I’d really appreciate it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3163018637
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
Review Comment:
nit: This name did not read clean to me, thoughts on something like this
```suggestion
class AuronExecutionPagedTableSpark3x
```
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -29,12 +36,22 @@ 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()) ++
+ buildExecutionsListSummary(sqlStore.executionsList(), request),
Review Comment:
nit: let's add a helper were we can arrange all the elements within an Auron view, so we can re-use it and have a consolidated place were we can design this page?
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala:
##########
@@ -36,8 +39,41 @@ class AuronSQLAppStatusStore(store: KVStore) extends Logging {
None
}
}
+
+ private def viewToSeq[T](view: KVStoreView[T]): Seq[T] = {
+ Utils.tryWithResource(view.closeableIterator())(iter => iter.asScala.toList)
+ }
+
+ def executionsList(): Seq[AuronSQLExecutionUIData] = {
+ viewToSeq(store.view(classOf[AuronSQLExecutionUIData]))
+ }
+
+ def executionsList(offset: Int, length: Int): Seq[AuronSQLExecutionUIData] = {
+ viewToSeq(store.view(classOf[AuronSQLExecutionUIData]).skip(offset).max(length))
+ }
+
+ def execution(executionId: Long): Option[AuronSQLExecutionUIData] = {
+ try {
+ Some(store.read(classOf[AuronSQLExecutionUIData], executionId))
+ } catch {
+ case _: NoSuchElementException => None
+ }
+ }
+
+ def executionsCount(): Long = {
+ store.count(classOf[AuronSQLExecutionUIData])
+ }
}
+@KVIndex("executionId")
+class AuronSQLExecutionUIData(
+ @KVIndexParam val executionId: Long,
+ val description: String,
Review Comment:
Storing the entire plan here might be expensive for large plans. The structured fallbackNodeToReason is the bit that's actually unique and adds value. I'm not sure if the full plan needs to live in `AuronSQLExecutionUIData` at all. One thing we could do is `+details` popover can link to Spark's existing `/SQL/execution/?id=`... page (which already retains its own plan, deduped at the Spark layer).
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusStore.scala:
##########
@@ -36,8 +39,41 @@ class AuronSQLAppStatusStore(store: KVStore) extends Logging {
None
}
}
+
+ private def viewToSeq[T](view: KVStoreView[T]): Seq[T] = {
+ Utils.tryWithResource(view.closeableIterator())(iter => iter.asScala.toList)
+ }
+
+ def executionsList(): Seq[AuronSQLExecutionUIData] = {
Review Comment:
executionsList() loads everything into memory and the page sorts in memory `AuronSQLAppStatusStore.scala:42-44 (iter.asScala.toList)` and `AuronAllExecutionsPage.scala:560-561 (executionData.map(...).sorted(...))`. The paginated overload `executionsList(offset, length)` exists but isn't used. With 1000 retained executions × MB-scale plan strings, this could have non-trivial effect on the heap? I think the cap helps, but pagination at the store layer would help more? WDYT?
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronExplainUtils.scala:
##########
@@ -0,0 +1,268 @@
+/*
+ * 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.Collections.newSetFromMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.auron.AuronConvertStrategy.neverConvertReasonTag
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.{BaseSubqueryExec, InputAdapter, ReusedSubqueryExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.ExplainUtils.getOpId
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.command.ExecutedCommandExec
+import org.apache.spark.sql.execution.exchange.{Exchange, ReusedExchangeExec}
+
+import org.apache.auron.sparkver
+
+object AuronExplainUtils {
+ private def generateOperatorIDs(
+ plan: QueryPlan[_],
+ startOperatorID: Int,
+ visited: java.util.Set[QueryPlan[_]],
+ reusedExchanges: ArrayBuffer[ReusedExchangeExec],
+ addReusedExchanges: Boolean): Int = {
+ var currentOperationID = startOperatorID
+ if (plan.isInstanceOf[BaseSubqueryExec]) {
+ return currentOperationID
+ }
+
+ def setOpId(plan: QueryPlan[_]): Unit = if (!visited.contains(plan)) {
+ plan match {
+ case r: ReusedExchangeExec if addReusedExchanges =>
+ reusedExchanges.append(r)
+ case _ =>
+ }
+ visited.add(plan)
+ currentOperationID += 1
+ plan.setTagValue(TreeNodeTag[Int]("operatorId"), currentOperationID)
+ }
+
+ plan.foreachUp {
+ case _: WholeStageCodegenExec =>
+ case _: InputAdapter =>
+ case p: AdaptiveSparkPlanExec =>
+ currentOperationID = generateOperatorIDs(
+ p.executedPlan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case p: QueryStageExec =>
+ currentOperationID = generateOperatorIDs(
+ p.plan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case other: QueryPlan[_] =>
+ setOpId(other)
+ currentOperationID = other.innerChildren.foldLeft(currentOperationID) { (curId, plan) =>
+ generateOperatorIDs(plan, curId, visited, reusedExchanges, addReusedExchanges)
+ }
+ }
+ currentOperationID
+ }
+
+ private def getSubqueries(
+ plan: => QueryPlan[_],
+ subqueries: ArrayBuffer[(SparkPlan, Expression, BaseSubqueryExec)]): Unit = {
+ plan.foreach {
+ case a: AdaptiveSparkPlanExec =>
+ getSubqueries(a.executedPlan, subqueries)
+ case q: QueryStageExec =>
+ getSubqueries(q.plan, subqueries)
+ case p: SparkPlan =>
+ p.expressions.foreach(_.collect { case e: PlanExpression[_] =>
+ e.plan match {
+ case s: BaseSubqueryExec =>
+ subqueries += ((p, e, s))
+ getSubqueries(s, subqueries)
+ case _ =>
+ }
+ })
+ }
+ }
+
+ private def processPlanSkippingSubqueries[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit): Unit = {
+ try {
+
+ QueryPlan.append(plan, append, verbose = false, addSuffix = false, printOperatorId = true)
+
+ append("\n")
+ } catch {
+ case e: AnalysisException => append(e.toString)
+ }
+ }
+
+ private def collectFallbackNodes(plan: QueryPlan[_]): (Int, Map[String, String]) = {
+ var numAuronNodes = 0
+ val fallbackNodeToReason = new mutable.HashMap[String, String]
+
+ def collect(tmp: QueryPlan[_]): Unit = {
+ tmp.foreachUp {
+ case p: ExecutedCommandExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ case p: AdaptiveSparkPlanExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.executedPlan)
+ case p: QueryStageExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.plan)
+ case p: NativeSupports =>
+ numAuronNodes += 1
+ p.innerChildren.foreach(collect)
+ case p: SparkPlan =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ p.innerChildren.foreach(collect)
+ case _ =>
+ }
+ }
+
+ collect(plan)
+ (numAuronNodes, fallbackNodeToReason.toMap)
+ }
+
+ def handleVanillaSparkPlan(
+ p: SparkPlan,
+ fallbackNodeToReason: mutable.HashMap[String, String]): Unit = {
+ if (p.getTagValue(neverConvertReasonTag).isDefined) {
+ addFallbackNodeWithReason(p, p.getTagValue(neverConvertReasonTag).get, fallbackNodeToReason)
+ }
+ }
+
+ def addFallbackNodeWithReason(
+ p: SparkPlan,
+ reason: String,
+ fallbackNodeToReason: mutable.HashMap[String, String]): Unit = {
+ p.getTagValue(TreeNodeTag[Int]("operatorId")).foreach { opId =>
+ // e.g., 002 project, it is used to help analysis by `substring(4)`
+ val formattedNodeName = f"$opId%03d ${p.nodeName}"
+ fallbackNodeToReason.put(formattedNodeName, reason)
+ }
+ }
+
+ def processPlan[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit,
+ collectFallbackFunc: Option[QueryPlan[_] => (Int, Map[String, String])] = None)
+ : (Int, Map[String, String]) = synchronized {
Review Comment:
Does this need to be synchronized?
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
Review Comment:
Same here
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala:
##########
@@ -40,10 +47,61 @@ class AuronSQLAppStatusListener(conf: SparkConf, kvstore: ElementTrackingStore)
}
override def onOtherEvent(event: SparkListenerEvent): Unit = event match {
+ case e: SparkListenerSQLExecutionStart => onSQLExecutionStart(e)
+ case e: SparkListenerSQLExecutionEnd => onSQLExtensionEnd(e)
case e: AuronBuildInfoEvent => onAuronBuildInfo(e)
+ case e: AuronPlanFallbackEvent => onAuronPlanFallback(e)
case _ => // Ignore
}
+ private def onAuronPlanFallback(event: AuronPlanFallbackEvent): Unit = {
+ val description = executionIdToDescription.get(event.executionId)
+ if (description.isDefined) {
+ val uiData = new AuronSQLExecutionUIData(
+ event.executionId,
+ description.get,
+ event.numAuronNodes,
+ event.numFallbackNodes,
+ event.physicalPlanDescription,
+ event.fallbackNodeToReason.toSeq.sortBy(_._1))
+ kvstore.write(uiData)
+ } else {
+ executionIdToFallbackEvent.put(event.executionId, event.copy())
Review Comment:
do we need to do event.copy(), the event should be immutable right?
##########
spark-extension/src/main/scala/org/apache/spark/sql/auron/AuronExplainUtils.scala:
##########
@@ -0,0 +1,268 @@
+/*
+ * 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.Collections.newSetFromMap
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.auron.AuronConvertStrategy.neverConvertReasonTag
+import org.apache.spark.sql.catalyst.expressions.{Expression, PlanExpression}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.catalyst.trees.TreeNodeTag
+import org.apache.spark.sql.execution.{BaseSubqueryExec, InputAdapter, ReusedSubqueryExec, SparkPlan, WholeStageCodegenExec}
+import org.apache.spark.sql.execution.ExplainUtils.getOpId
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.command.ExecutedCommandExec
+import org.apache.spark.sql.execution.exchange.{Exchange, ReusedExchangeExec}
+
+import org.apache.auron.sparkver
+
+object AuronExplainUtils {
+ private def generateOperatorIDs(
+ plan: QueryPlan[_],
+ startOperatorID: Int,
+ visited: java.util.Set[QueryPlan[_]],
+ reusedExchanges: ArrayBuffer[ReusedExchangeExec],
+ addReusedExchanges: Boolean): Int = {
+ var currentOperationID = startOperatorID
+ if (plan.isInstanceOf[BaseSubqueryExec]) {
+ return currentOperationID
+ }
+
+ def setOpId(plan: QueryPlan[_]): Unit = if (!visited.contains(plan)) {
+ plan match {
+ case r: ReusedExchangeExec if addReusedExchanges =>
+ reusedExchanges.append(r)
+ case _ =>
+ }
+ visited.add(plan)
+ currentOperationID += 1
+ plan.setTagValue(TreeNodeTag[Int]("operatorId"), currentOperationID)
+ }
+
+ plan.foreachUp {
+ case _: WholeStageCodegenExec =>
+ case _: InputAdapter =>
+ case p: AdaptiveSparkPlanExec =>
+ currentOperationID = generateOperatorIDs(
+ p.executedPlan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case p: QueryStageExec =>
+ currentOperationID = generateOperatorIDs(
+ p.plan,
+ currentOperationID,
+ visited,
+ reusedExchanges,
+ addReusedExchanges)
+ setOpId(p)
+ case other: QueryPlan[_] =>
+ setOpId(other)
+ currentOperationID = other.innerChildren.foldLeft(currentOperationID) { (curId, plan) =>
+ generateOperatorIDs(plan, curId, visited, reusedExchanges, addReusedExchanges)
+ }
+ }
+ currentOperationID
+ }
+
+ private def getSubqueries(
+ plan: => QueryPlan[_],
+ subqueries: ArrayBuffer[(SparkPlan, Expression, BaseSubqueryExec)]): Unit = {
+ plan.foreach {
+ case a: AdaptiveSparkPlanExec =>
+ getSubqueries(a.executedPlan, subqueries)
+ case q: QueryStageExec =>
+ getSubqueries(q.plan, subqueries)
+ case p: SparkPlan =>
+ p.expressions.foreach(_.collect { case e: PlanExpression[_] =>
+ e.plan match {
+ case s: BaseSubqueryExec =>
+ subqueries += ((p, e, s))
+ getSubqueries(s, subqueries)
+ case _ =>
+ }
+ })
+ }
+ }
+
+ private def processPlanSkippingSubqueries[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit): Unit = {
+ try {
+
+ QueryPlan.append(plan, append, verbose = false, addSuffix = false, printOperatorId = true)
+
+ append("\n")
+ } catch {
+ case e: AnalysisException => append(e.toString)
+ }
+ }
+
+ private def collectFallbackNodes(plan: QueryPlan[_]): (Int, Map[String, String]) = {
+ var numAuronNodes = 0
+ val fallbackNodeToReason = new mutable.HashMap[String, String]
+
+ def collect(tmp: QueryPlan[_]): Unit = {
+ tmp.foreachUp {
+ case p: ExecutedCommandExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ case p: AdaptiveSparkPlanExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.executedPlan)
+ case p: QueryStageExec =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ collect(p.plan)
+ case p: NativeSupports =>
+ numAuronNodes += 1
+ p.innerChildren.foreach(collect)
+ case p: SparkPlan =>
+ handleVanillaSparkPlan(p, fallbackNodeToReason)
+ p.innerChildren.foreach(collect)
+ case _ =>
+ }
+ }
+
+ collect(plan)
+ (numAuronNodes, fallbackNodeToReason.toMap)
+ }
+
+ def handleVanillaSparkPlan(
+ p: SparkPlan,
+ fallbackNodeToReason: mutable.HashMap[String, String]): Unit = {
+ if (p.getTagValue(neverConvertReasonTag).isDefined) {
+ addFallbackNodeWithReason(p, p.getTagValue(neverConvertReasonTag).get, fallbackNodeToReason)
+ }
+ }
+
+ def addFallbackNodeWithReason(
+ p: SparkPlan,
+ reason: String,
+ fallbackNodeToReason: mutable.HashMap[String, String]): Unit = {
+ p.getTagValue(TreeNodeTag[Int]("operatorId")).foreach { opId =>
+ // e.g., 002 project, it is used to help analysis by `substring(4)`
+ val formattedNodeName = f"$opId%03d ${p.nodeName}"
+ fallbackNodeToReason.put(formattedNodeName, reason)
+ }
+ }
+
+ def processPlan[T <: QueryPlan[T]](
+ plan: T,
+ append: String => Unit,
+ collectFallbackFunc: Option[QueryPlan[_] => (Int, Map[String, String])] = None)
+ : (Int, Map[String, String]) = synchronized {
+ try {
+ val operators = newSetFromMap[QueryPlan[_]](new java.util.IdentityHashMap())
+ val reusedExchanges = ArrayBuffer.empty[ReusedExchangeExec]
+
+ var currentOperatorID = 0
+ currentOperatorID =
+ generateOperatorIDs(plan, currentOperatorID, operators, reusedExchanges, true)
+
+ val subqueries = ArrayBuffer.empty[(SparkPlan, Expression, BaseSubqueryExec)]
+ getSubqueries(plan, subqueries)
+
+ currentOperatorID = subqueries.foldLeft(currentOperatorID) { (curId, plan) =>
+ generateOperatorIDs(plan._3.child, curId, operators, reusedExchanges, true)
+ }
+
+ val optimizedOutExchanges = ArrayBuffer.empty[Exchange]
+ reusedExchanges.foreach { reused =>
+ val child = reused.child
+ if (!operators.contains(child)) {
+ optimizedOutExchanges.append(child)
+ currentOperatorID =
+ generateOperatorIDs(child, currentOperatorID, operators, reusedExchanges, false)
+ }
+ }
+
+ processPlanSkippingSubqueries(plan, append)
+
+ var i = 0
+ for (sub <- subqueries) {
+ if (i == 0) {
+ append("\n===== Subqueries =====\n\n")
+ }
+ i = i + 1
+ append(
+ s"Subquery:$i Hosting operator id = " +
+ s"${getOpId(sub._1)} Hosting Expression = ${sub._2}\n")
+
+ if (!sub._3.isInstanceOf[ReusedSubqueryExec]) {
+ processPlanSkippingSubqueries(sub._3.child, append)
+ }
+ append("\n")
+ }
+
+ i = 0
+ optimizedOutExchanges.foreach { exchange =>
+ if (i == 0) {
+ append("\n===== Adaptively Optimized Out Exchanges =====\n\n")
+ }
+ i = i + 1
+ append(s"Subplan:$i\n")
+ processPlanSkippingSubqueries[SparkPlan](exchange, append)
+ append("\n")
+ }
+
+ (subqueries.filter(!_._3.isInstanceOf[ReusedSubqueryExec]).map(_._3.child) :+ plan)
+ .map { plan =>
+ if (collectFallbackFunc.isEmpty) {
+ collectFallbackNodes(plan)
+ } else {
+ collectFallbackFunc.get.apply(plan)
+ }
+ }
+ .reduce((a, b) => (a._1 + b._1, a._2 ++ b._2))
+ } finally {
+ removeTags(plan)
+ }
+ }
+
+ @sparkver("3.1/ 3.2 / 3.3/ 3.4/ 3.5/ 4.0/ 4.1")
+ private def removeTags(plan: QueryPlan[_]): Unit = {
+ def remove(p: QueryPlan[_], children: Seq[QueryPlan[_]]): Unit = {
+ p.unsetTagValue(TreeNodeTag[Int]("operatorId"))
+ children.foreach(removeTags)
+ }
+
+ plan.foreach {
+ case p: AdaptiveSparkPlanExec => remove(p, Seq(p.executedPlan))
+ case p: QueryStageExec => remove(p, Seq(p.plan))
+ case plan: QueryPlan[_] => remove(plan, plan.innerChildren)
+ }
+ }
+
+ @sparkver("3.0")
+ private def removeTags(plan: QueryPlan[_]): Unit = {
Review Comment:
I think the operation here is quadratic: `AuronExplainUtils.scala:L204-238`. `plan.foreach { case ... => remove(p, p.innerChildren) }` already visits every node; inside remove, `children.foreach(removeTags)` recurses again from each visited node. Each subtree is visited `O(depth)` times. For a plan with N nodes and depth D, this is `O(N·D)` instead of `O(N)`. Replace with a `single plan.foreach(_.unsetTagValue(...))`plus an explicit walk of `AdaptiveSparkPlanExec.executedPlan` / `QueryStageExec.plan` if those aren't reached by the default foreach
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2150:
URL: https://github.com/apache/auron/pull/2150#discussion_r3414870569
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
Review Comment:
Parsing the page number with `.toInt` can throw (e.g., `auron.page=abc`) and fail the whole page render. The try/catch below won't catch it because it happens before the `try` block.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronSQLAppStatusListener.scala:
##########
@@ -16,20 +16,27 @@
*/
package org.apache.spark.sql.execution.ui
-import scala.annotation.nowarn
+import scala.collection.mutable
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.spark.sql.internal.StaticSQLConf.UI_RETAINED_EXECUTIONS
+import org.apache.spark.status.{ElementTrackingStore, KVUtils}
-import org.apache.auron.spark.ui.AuronBuildInfoEvent
+import org.apache.auron.spark.ui.{AuronBuildInfoEvent, AuronPlanFallbackEvent}
-@nowarn("cat=unused") // conf temporarily unused
class AuronSQLAppStatusListener(conf: SparkConf, kvstore: ElementTrackingStore)
extends SparkListener
with Logging {
+ private val executionIdToDescription = new mutable.HashMap[Long, String]
+ private val executionIdToFallbackEvent = new mutable.HashMap[Long, AuronPlanFallbackEvent]
+
+ kvstore.addTrigger(classOf[SQLExecutionUIData], conf.get[Int](UI_RETAINED_EXECUTIONS)) {
+ count => cleanupExecutions(count)
+ }
Review Comment:
The KVStore cleanup trigger is registered on SQLExecutionUIData, but cleanupExecutions() deletes AuronSQLExecutionUIData. If AuronSQLExecutionUIData is missing for some SQL executions, this can delete the wrong records (or delete too aggressively). Register the trigger on AuronSQLExecutionUIData so the count passed into cleanup matches the data being deleted.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
Review Comment:
Query parameters from the request are concatenated into URLs without URL-encoding. Because these values are user-controlled, this can lead to broken links and potential injection/XSS issues when rendered into href attributes.
##########
auron-spark-ui/src/main/scala/org/apache/spark/sql/execution/ui/AuronAllExecutionsPage.scala:
##########
@@ -77,4 +94,612 @@ private[ui] class AuronAllExecutionsPage(parent: AuronSQLTab) extends WebUIPage(
}
}
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: javax.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable30(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("4.0 / 4.1")
+ private def buildExecutionsListSummary(
+ executionsList: Seq[AuronSQLExecutionUIData],
+ request: jakarta.servlet.http.HttpServletRequest): NodeSeq = {
+ val content = {
+ val _content = mutable.ListBuffer[Node]()
+ val executionPage =
+ Option(request.getParameter("auron.page")).map(_.toInt).getOrElse(1)
+
+ val auronPageTable =
+ try {
+ new AuronExecutionPagedTable40(
+ request,
+ parent,
+ executionsList,
+ "auron",
+ "auron",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ "auron").table(executionPage)
+ } catch {
+ case e @ (_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+
+ _content ++=
+ <span id="auron" class="collapse-aggregated-runningExecutions collapse-table"
+ onClick="collapseTable('collapse-aggregated-runningExecutions',
+ 'aggregated-runningExecutions')">
+ <h4>
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a href="#auron">
+ Queries:
+ </a>{executionsList.size}
+ </h4>
+ </span> ++
+ <div class="aggregated-runningExecutions collapsible-table">
+ {auronPageTable}
+ </div>
+
+ _content
+ }
+ content ++=
+ <script>
+ function clickDetail(details) {{
+ details.parentNode.querySelector('.stage-details').classList.toggle('collapsed')
+ }}
+ </script>
+ }
+
+ @sparkver("3.0 / 3.1 / 3.2 / 3.3 / 3.4 / 3.5")
+ class AuronExecutionPagedTable30(
+ request: javax.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
+ }
+
+ /**
+ * Returns parameter of this table.
+ */
+ def getAuronTableParameters(
+ request: javax.servlet.http.HttpServletRequest,
+ tableTag: String,
+ defaultSortColumn: String): (String, Boolean, Int) = {
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+ val sortColumn = Option(parameterSortColumn)
+ .map { sortColumn =>
+ UIUtils.decodeURLParameter(sortColumn)
+ }
+ .getOrElse(defaultSortColumn)
+ val desc =
+ Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == defaultSortColumn)
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ (sortColumn, desc, pageSize)
+ }
+
+ override def pageSizeFormField: String = s"$executionTag.pageSize"
+
+ override def pageNumberFormField: String = s"$executionTag.page"
+
+ override def goButtonFormPath: String =
+ s"$parameterPath&$executionTag.sort=$encodedSortColumn&$executionTag.desc=$desc#$tableHeaderId"
+
+ // Information for each header: title, sortable, tooltip
+ private val headerInfo: Seq[(String, Boolean, Option[String])] = {
+ Seq(
+ ("ID", true, None),
+ ("Description", true, None),
+ ("Num Auron Nodes", true, None),
+ ("Num Fallback Nodes", true, None))
+ }
+
+ override def headers: Seq[Node] = {
+ isAuronSortColumnValid(headerInfo, sortColumn)
+
+ headerAuronRow(
+ headerInfo,
+ desc,
+ pageSize,
+ sortColumn,
+ parameterPath,
+ executionTag,
+ tableHeaderId)
+ }
+
+ def headerAuronRow(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ desc: Boolean,
+ pageSize: Int,
+ sortColumn: String,
+ parameterPath: String,
+ tableTag: String,
+ headerId: String): Seq[Node] = {
+ val row: Seq[Node] = {
+ headerInfo.map { case (header, sortable, tooltip) =>
+ if (header == sortColumn) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.desc=${!desc}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+ val arrow = if (desc) "▾" else "▴" // UP or DOWN
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header} {Unparsed(arrow)}
+ </span>
+ </a>
+ </th>
+ } else {
+ if (sortable) {
+ val headerLink = Unparsed(
+ parameterPath +
+ s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" +
+ s"&$tableTag.pageSize=$pageSize" +
+ s"#$headerId")
+
+ <th>
+ <a href={headerLink}>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </a>
+ </th>
+ } else {
+ <th>
+ <span data-toggle="tooltip" data-placement="top" title={tooltip.getOrElse("")}>
+ {header}
+ </span>
+ </th>
+ }
+ }
+ }
+ }
+ <thead>
+ <tr>
+ {row}
+ </tr>
+ </thead>
+ }
+
+ def isAuronSortColumnValid(
+ headerInfo: Seq[(String, Boolean, Option[String])],
+ sortColumn: String): Unit = {
+ if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) {
+ throw new IllegalArgumentException(s"Unknown column: $sortColumn")
+ }
+ }
+
+ override def row(executionTableRow: AuronExecutionTableRowData): Seq[Node] = {
+ val executionUIData = executionTableRow.executionUIData
+
+ <tr>
+ <td>
+ {executionUIData.executionId.toString}
+ </td>
+ <td>
+ {descriptionCell(executionUIData)}
+ </td>
+ <td sorttable_customkey={executionUIData.numAuronNodes.toString}>
+ {executionUIData.numAuronNodes.toString}
+ </td>
+ <td sorttable_customkey={executionUIData.numFallbackNodes.toString}>
+ {executionUIData.numFallbackNodes.toString}
+ </td>
+ </tr>
+ }
+
+ private def descriptionCell(execution: AuronSQLExecutionUIData): Seq[Node] = {
+ val details = if (execution.description != null && execution.description.nonEmpty) {
+ val concat = new PlanStringConcat()
+ concat.append("== Fallback Summary ==\n")
+ val fallbackSummary = execution.fallbackNodeToReason
+ .map { case (name, reason) =>
+ val id = name.substring(0, 3)
+ val nodeName = name.substring(4)
+ s"(${id.toInt}) $nodeName: $reason"
+ }
+ .mkString("\n")
+ concat.append(fallbackSummary)
+ if (execution.fallbackNodeToReason.isEmpty) {
+ concat.append("No fallback nodes")
+ }
+ concat.append("\n\n")
+ concat.append(execution.fallbackDescription)
+
+ <span onclick="this.parentNode.querySelector('.stage-details').classList.toggle('collapsed')"
+ class="expand-details">
+ +details
+ </span> ++
+ <div class="stage-details collapsed">
+ <pre>
+ {concat.toString()}
+ </pre>
+ </div>
+ } else {
+ Nil
+ }
+
+ val desc = if (execution.description != null && execution.description.nonEmpty) {
+ <a href={executionURL(execution.executionId)} class="description-input">
+ {execution.description}
+ </a>
+ } else {
+ <a href={executionURL(execution.executionId)}>
+ {execution.executionId}
+ </a>
+ }
+
+ <div>
+ {desc}{details}
+ </div>
+ }
+
+ private def executionURL(executionID: Long): String =
+ s"${UIUtils.prependBaseUri(request, parent.basePath)}/SQL/execution/?id=$executionID"
+ }
+
+ @sparkver("4.0 / 4.1")
+ private[ui] class AuronExecutionPagedTable40(
+ request: jakarta.servlet.http.HttpServletRequest,
+ parent: AuronSQLTab,
+ data: Seq[AuronSQLExecutionUIData],
+ tableHeaderId: String,
+ executionTag: String,
+ basePath: String,
+ subPath: String)
+ extends PagedTable[AuronExecutionTableRowData] {
+
+ private val (sortColumn, desc, pageSize) =
+ getAuronTableParameters(request, executionTag, "ID")
+
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override val dataSource = new AuronExecutionDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath =
+ s"$basePath/$subPath/?${getAuronParameterOtherTable(request, executionTag)}"
+
+ override def tableId: String = s"$executionTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$executionTag.sort=$encodedSortColumn" +
+ s"&$executionTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableHeaderId"
+ }
+
+ /**
+ * Returns parameters of other tables in the page.
+ */
+ def getAuronParameterOtherTable(
+ request: jakarta.servlet.http.HttpServletRequest,
+ tableTag: String): String = {
+ request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map(parameter => parameter._1 + "=" + parameter._2(0))
+ .mkString("&")
Review Comment:
Query parameters from the request are concatenated into URLs without URL-encoding. Because these values are user-controlled, this can lead to broken links and potential injection/XSS issues when rendered into href attributes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "ShreyeshArangath (via GitHub)" <gi...@apache.org>.
ShreyeshArangath commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4712368074
@guixiaowen did you get a chance to address some of my comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [AURON #1471] Add the conversion execution list in the SparkUI [auron]
Posted by "guixiaowen (via GitHub)" <gi...@apache.org>.
guixiaowen commented on PR #2150:
URL: https://github.com/apache/auron/pull/2150#issuecomment-4414754231
@richox Could you please review the code again? It was already reviewed before, but there were merge conflicts that have now been resolved. Thank you.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@auron.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org