uncleGen commented on a change in pull request #28485:
URL: https://github.com/apache/spark/pull/28485#discussion_r422801194
##########
File path:
sql/core/src/main/scala/org/apache/spark/sql/streaming/ui/StreamingQueryPage.scala
##########
@@ -35,11 +40,218 @@ private[ui] class StreamingQueryPage(parent:
StreamingQueryTab)
SparkUIUtils.headerSparkPage(request, "Streaming Query", content, parent)
}
- def generateDataRow(request: HttpServletRequest, queryActive: Boolean)
- (query: StreamingQueryUIData): Seq[Node] = {
+ private def generateStreamingQueryTable(request: HttpServletRequest):
Seq[Node] = {
+ val (activeQueries, inactiveQueries) = parent.statusListener.allQueryStatus
+ .partition(_.isActive)
+
+ val content = mutable.ListBuffer[Node]()
+ // show active queries table only if there is at least one active query
+ if (activeQueries.nonEmpty) {
+ // scalastyle:off
+ content ++=
+ <span id="active" class="collapse-aggregated-activeQueries
collapse-table"
+
onClick="collapseTable('collapse-aggregated-activeQueries','aggregated-activeQueries')">
+ <h5 id="activequeries">
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a>Active Streaming Queries ({activeQueries.length})</a>
+ </h5>
+ </span> ++
+ <div>
+ <ul class="aggregated-activeQueries collapsible-table">
+ {queryTable(activeQueries, request, "active")}
+ </ul>
+ </div>
+ // scalastyle:on
+ }
+ // show active queries table only if there is at least one completed query
+ if (inactiveQueries.nonEmpty) {
+ // scalastyle:off
+ content ++=
+ <span id="completed" class="collapse-aggregated-completedQueries
collapse-table"
+
onClick="collapseTable('collapse-aggregated-completedQueries','aggregated-completedQueries')">
+ <h5 id="completedqueries">
+ <span class="collapse-table-arrow arrow-open"></span>
+ <a>Completed Streaming Queries ({inactiveQueries.length})</a>
+ </h5>
+ </span> ++
+ <div>
+ <ul class="aggregated-completedQueries collapsible-table">
+ {queryTable(inactiveQueries, request, "completed")}
+ </ul>
+ </div>
+ // scalastyle:on
+ }
+ content
+ }
+
+ private def queryTable(data: Seq[StreamingQueryUIData], request:
HttpServletRequest,
+ tableTag: String): Seq[Node] = {
+
+ val isActive = if (tableTag.contains("active")) true else false
+ val parameterOtherTable = request.getParameterMap.asScala
+ .filterNot(_._1.startsWith(tableTag))
+ .map { case (name, vals) =>
+ name + "=" + vals(0)
+ }
+
+ val parameterPage = request.getParameter(s"$tableTag.page")
+ val parameterSortColumn = request.getParameter(s"$tableTag.sort")
+ val parameterSortDesc = request.getParameter(s"$tableTag.desc")
+ val parameterPageSize = request.getParameter(s"$tableTag.pageSize")
+
+ val page = Option(parameterPage).map(_.toInt).getOrElse(1)
+ val sortColumn = Option(parameterSortColumn).map { sortColumn =>
+ SparkUIUtils.decodeURLParameter(sortColumn)
+ }.getOrElse("Start Time")
+ val sortDesc =
Option(parameterSortDesc).map(_.toBoolean).getOrElse(sortColumn == "Start Time")
+ val pageSize = Option(parameterPageSize).map(_.toInt).getOrElse(100)
+
+ try {
+ new StreamingQueryPagedTable(
+ request,
+ parent,
+ data,
+ tableTag,
+ pageSize,
+ sortColumn,
+ sortDesc,
+ isActive,
+ parameterOtherTable,
+ SparkUIUtils.prependBaseUri(request, parent.basePath),
+ "StreamingQuery"
+ ).table(page)
+ } catch {
+ case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ <div class="alert alert-error">
+ <p>Error while rendering execution table:</p>
+ <pre>
+ {Utils.exceptionString(e)}
+ </pre>
+ </div>
+ }
+ }
+}
+
+class StreamingQueryPagedTable(
+ request: HttpServletRequest,
+ parent: StreamingQueryTab,
+ data: Seq[StreamingQueryUIData],
+ tableTag: String,
+ pageSize: Int,
+ sortColumn: String,
+ sortDesc: Boolean,
+ isActive: Boolean,
+ parameterOtherTable: Iterable[String],
+ basePath: String,
+ subPath: String) extends PagedTable[StructuredStreamingRow] {
+
+ private val parameterPath =
s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}"
+ private val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+
+ override def tableId: String = s"$tableTag-table"
+
+ override def tableCssClass: String =
+ "table table-bordered table-sm table-striped table-head-clickable
table-cell-width-limited"
+
+ override def pageSizeFormField: String = s"$tableTag.pageSize"
+
+ override def pageNumberFormField: String = s"$tableTag.page"
+
+ override def pageLink(page: Int): String = {
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$tableTag.sort=$encodedSortColumn" +
+ s"&$tableTag.desc=$sortDesc" +
+ s"&$pageSizeFormField=$pageSize" +
+ s"#$tableTag"
Review comment:
nit: indent
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
