diff --git a/core/src/main/scala/org/apache/spark/ui/PagedTable.scala b/core/src/main/scala/org/apache/spark/ui/PagedTable.scala index 008dcc6200d3..a002af70a919 100644 --- a/core/src/main/scala/org/apache/spark/ui/PagedTable.scala +++ b/core/src/main/scala/org/apache/spark/ui/PagedTable.scala @@ -17,8 +17,9 @@ package org.apache.spark.ui -import java.net.URLDecoder +import java.net.{URLDecoder, URLEncoder} import java.nio.charset.StandardCharsets.UTF_8 +import javax.servlet.http.HttpServletRequest import scala.collection.JavaConverters._ import scala.xml.{Node, Unparsed} @@ -297,4 +298,102 @@ private[spark] trait PagedTable[T] { * Returns the submission path for the "go to page #" form. */ def goButtonFormPath: String + + /** + * Returns parameters of other tables in the page. + */ + def getParameterOtherTable(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 getTableParameters( + 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) + } + + /** + * Check if given sort column is valid or not. If invalid then an exception is thrown. + */ + def isSortColumnValid( + headerInfo: Seq[(String, Boolean, Option[String])], + sortColumn: String): Unit = { + if (!headerInfo.filter(_._2).map(_._1).contains(sortColumn)) { + throw new IllegalArgumentException(s"Unknown column: $sortColumn") + } + } + + def headerRow( + 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 + + + + + {header} {Unparsed(arrow)} + + + + } else { + if (sortable) { + val headerLink = Unparsed( + parameterPath + + s"&$tableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + + s"&$tableTag.pageSize=$pageSize" + + s"#$headerId") + + + + + {header} + + + + } else { + + + {header} + + + } + } + } + } + + {row} + + } } diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala index 0b362201a784..716d604ca31b 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/AllJobsPage.scala @@ -22,7 +22,6 @@ import java.nio.charset.StandardCharsets.UTF_8 import java.util.Date import javax.servlet.http.HttpServletRequest -import scala.collection.JavaConverters._ import scala.collection.mutable.ListBuffer import scala.xml._ @@ -211,45 +210,24 @@ private[ui] class AllJobsPage(parent: JobsTab, store: AppStatusStore) extends We jobTag: String, jobs: Seq[v1.JobData], killEnabled: Boolean): Seq[Node] = { - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(jobTag)) - .map(para => para._1 + "=" + para._2(0)) val someJobHasJobGroup = jobs.exists(_.jobGroup.isDefined) val jobIdTitle = if (someJobHasJobGroup) "Job Id (Job Group)" else "Job Id" - - val parameterJobPage = request.getParameter(jobTag + ".page") - val parameterJobSortColumn = request.getParameter(jobTag + ".sort") - val parameterJobSortDesc = request.getParameter(jobTag + ".desc") - val parameterJobPageSize = request.getParameter(jobTag + ".pageSize") - - val jobPage = Option(parameterJobPage).map(_.toInt).getOrElse(1) - val jobSortColumn = Option(parameterJobSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse(jobIdTitle) - val jobSortDesc = Option(parameterJobSortDesc).map(_.toBoolean).getOrElse( - // New jobs should be shown above old jobs by default. - jobSortColumn == jobIdTitle - ) - val jobPageSize = Option(parameterJobPageSize).map(_.toInt).getOrElse(100) - + val jobPage = Option(request.getParameter(jobTag + ".page")).map(_.toInt).getOrElse(1) val currentTime = System.currentTimeMillis() try { new JobPagedTable( + request, store, jobs, tableHeaderId, jobTag, UIUtils.prependBaseUri(request, parent.basePath), "jobs", // subPath - parameterOtherTable, killEnabled, currentTime, - jobIdTitle, - pageSize = jobPageSize, - sortColumn = jobSortColumn, - desc = jobSortDesc + jobIdTitle ).table(jobPage) } catch { case e @ (_ : IllegalArgumentException | _ : IndexOutOfBoundsException) => @@ -493,21 +471,19 @@ private[ui] class JobDataSource( } private[ui] class JobPagedTable( + request: HttpServletRequest, store: AppStatusStore, data: Seq[v1.JobData], tableHeaderId: String, jobTag: String, basePath: String, subPath: String, - parameterOtherTable: Iterable[String], killEnabled: Boolean, currentTime: Long, - jobIdTitle: String, - pageSize: Int, - sortColumn: String, - desc: Boolean + jobIdTitle: String ) extends PagedTable[JobTableRowData] { - val parameterPath = basePath + s"/$subPath/?" + parameterOtherTable.mkString("&") + private val (sortColumn, desc, pageSize) = getTableParameters(request, jobTag, jobIdTitle) + private val parameterPath = basePath + s"/$subPath/?" + getParameterOtherTable(request, jobTag) override def tableId: String = jobTag + "-table" @@ -544,90 +520,22 @@ private[ui] class JobPagedTable( } override def headers: Seq[Node] = { - // Information for each header: title, cssClass, and sortable - val jobHeadersAndCssClasses: Seq[(String, String, Boolean, Option[String])] = + // Information for each header: title, sortable, tooltip + val jobHeadersAndCssClasses: Seq[(String, Boolean, Option[String])] = Seq( - (jobIdTitle, "", true, None), - ("Description", "", true, None), - ("Submitted", "", true, None), - ("Duration", "", true, Some("Elapsed time since the job was submitted " + + (jobIdTitle, true, None), + ("Description", true, None), + ("Submitted", true, None), + ("Duration", true, Some("Elapsed time since the job was submitted " + "until execution completion of all its stages.")), - ("Stages: Succeeded/Total", "", false, None), - ("Tasks (for all stages): Succeeded/Total", "", false, None) + ("Stages: Succeeded/Total", false, None), + ("Tasks (for all stages): Succeeded/Total", false, None) ) - if (!jobHeadersAndCssClasses.filter(_._3).map(_._1).contains(sortColumn)) { - throw new IllegalArgumentException(s"Unknown column: $sortColumn") - } + isSortColumnValid(jobHeadersAndCssClasses, sortColumn) - val headerRow: Seq[Node] = { - jobHeadersAndCssClasses.map { case (header, cssClass, sortable, tooltip) => - if (header == sortColumn) { - val headerLink = Unparsed( - parameterPath + - s"&$jobTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$jobTag.desc=${!desc}" + - s"&$jobTag.pageSize=$pageSize" + - s"#$tableHeaderId") - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - - - { - if (tooltip.nonEmpty) { - - {header} {Unparsed(arrow)} - - } else { - - {header} {Unparsed(arrow)} - - } - } - - - } else { - if (sortable) { - val headerLink = Unparsed( - parameterPath + - s"&$jobTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$jobTag.pageSize=$pageSize" + - s"#$tableHeaderId") - - - - { - if (tooltip.nonEmpty) { - - {header} - - } else { - - {header} - - } - } - - - } else { - - { - if (tooltip.nonEmpty) { - - {header} - - } else { - - {header} - - } - } - - } - } - } - } - {headerRow} + headerRow(jobHeadersAndCssClasses, desc, pageSize, sortColumn, parameterPath, + jobTag, tableHeaderId) } override def row(jobTableRow: JobTableRowData): Seq[Node] = { diff --git a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala index a29483b5d5a5..f9e84c2b2f4e 100644 --- a/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala +++ b/core/src/main/scala/org/apache/spark/ui/jobs/StageTable.scala @@ -22,7 +22,6 @@ import java.nio.charset.StandardCharsets.UTF_8 import java.util.Date import javax.servlet.http.HttpServletRequest -import scala.collection.JavaConverters._ import scala.xml._ import org.apache.commons.text.StringEscapeUtils @@ -43,24 +42,8 @@ private[ui] class StageTableBase( isFairScheduler: Boolean, killEnabled: Boolean, isFailedStage: Boolean) { - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(stageTag)) - .map(para => para._1 + "=" + para._2(0)) - - val parameterStagePage = request.getParameter(stageTag + ".page") - val parameterStageSortColumn = request.getParameter(stageTag + ".sort") - val parameterStageSortDesc = request.getParameter(stageTag + ".desc") - val parameterStagePageSize = request.getParameter(stageTag + ".pageSize") - - val stagePage = Option(parameterStagePage).map(_.toInt).getOrElse(1) - val stageSortColumn = Option(parameterStageSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse("Stage Id") - val stageSortDesc = Option(parameterStageSortDesc).map(_.toBoolean).getOrElse( - // New stages should be shown above old jobs by default. - stageSortColumn == "Stage Id" - ) - val stagePageSize = Option(parameterStagePageSize).map(_.toInt).getOrElse(100) + + val stagePage = Option(request.getParameter(stageTag + ".page")).map(_.toInt).getOrElse(1) val currentTime = System.currentTimeMillis() @@ -75,11 +58,7 @@ private[ui] class StageTableBase( isFairScheduler, killEnabled, currentTime, - stagePageSize, - stageSortColumn, - stageSortDesc, isFailedStage, - parameterOtherTable, request ).table(stagePage) } catch { @@ -131,11 +110,7 @@ private[ui] class StagePagedTable( isFairScheduler: Boolean, killEnabled: Boolean, currentTime: Long, - pageSize: Int, - sortColumn: String, - desc: Boolean, isFailedStage: Boolean, - parameterOtherTable: Iterable[String], request: HttpServletRequest) extends PagedTable[StageTableRowData] { override def tableId: String = stageTag + "-table" @@ -148,8 +123,10 @@ private[ui] class StagePagedTable( override def pageNumberFormField: String = stageTag + ".page" + private val (sortColumn, desc, pageSize) = getTableParameters(request, stageTag, "Stage Id") + val parameterPath = UIUtils.prependBaseUri(request, basePath) + s"/$subPath/?" + - parameterOtherTable.mkString("&") + getParameterOtherTable(request, stageTag) override val dataSource = new StageDataSource( store, @@ -176,76 +153,27 @@ private[ui] class StagePagedTable( } override def headers: Seq[Node] = { - // stageHeadersAndCssClasses has three parts: header title, tooltip information, and sortable. + // stageHeadersAndCssClasses has three parts: header title, sortable and tooltip information. // The tooltip information could be None, which indicates it does not have a tooltip. - // Otherwise, it has two parts: tooltip text, and position (true for left, false for default). - val stageHeadersAndCssClasses: Seq[(String, String, Boolean)] = - Seq(("Stage Id", null, true)) ++ - {if (isFairScheduler) {Seq(("Pool Name", null, true))} else Seq.empty} ++ + val stageHeadersAndCssClasses: Seq[(String, Boolean, Option[String])] = + Seq(("Stage Id", true, None)) ++ + {if (isFairScheduler) {Seq(("Pool Name", true, None))} else Seq.empty} ++ Seq( - ("Description", null, true), - ("Submitted", null, true), - ("Duration", ToolTips.DURATION, true), - ("Tasks: Succeeded/Total", null, false), - ("Input", ToolTips.INPUT, true), - ("Output", ToolTips.OUTPUT, true), - ("Shuffle Read", ToolTips.SHUFFLE_READ, true), - ("Shuffle Write", ToolTips.SHUFFLE_WRITE, true) + ("Description", true, None), + ("Submitted", true, None), + ("Duration", true, Some(ToolTips.DURATION)), + ("Tasks: Succeeded/Total", false, None), + ("Input", true, Some(ToolTips.INPUT)), + ("Output", true, Some(ToolTips.OUTPUT)), + ("Shuffle Read", true, Some(ToolTips.SHUFFLE_READ)), + ("Shuffle Write", true, Some(ToolTips.SHUFFLE_WRITE)) ) ++ - {if (isFailedStage) {Seq(("Failure Reason", null, false))} else Seq.empty} - - if (!stageHeadersAndCssClasses.filter(_._3).map(_._1).contains(sortColumn)) { - throw new IllegalArgumentException(s"Unknown column: $sortColumn") - } + {if (isFailedStage) {Seq(("Failure Reason", false, None))} else Seq.empty} - val headerRow: Seq[Node] = { - stageHeadersAndCssClasses.map { case (header, tooltip, sortable) => - val headerSpan = if (null != tooltip && !tooltip.isEmpty) { - - {header} - - } else { - {header} - } + isSortColumnValid(stageHeadersAndCssClasses, sortColumn) - if (header == sortColumn) { - val headerLink = Unparsed( - parameterPath + - s"&$stageTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$stageTag.desc=${!desc}" + - s"&$stageTag.pageSize=$pageSize") + - s"#$tableHeaderId" - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - - - {headerSpan} -  {Unparsed(arrow)} - - - - } else { - if (sortable) { - val headerLink = Unparsed( - parameterPath + - s"&$stageTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$stageTag.pageSize=$pageSize") + - s"#$tableHeaderId" - - - - {headerSpan} - - - } else { - - {headerSpan} - - } - } - } - } - {headerRow} + headerRow(stageHeadersAndCssClasses, desc, pageSize, sortColumn, parameterPath, + stageTag, tableHeaderId) } override def row(data: StageTableRowData): Seq[Node] = { diff --git a/core/src/main/scala/org/apache/spark/ui/storage/RDDPage.scala b/core/src/main/scala/org/apache/spark/ui/storage/RDDPage.scala index 04f8d8edd4d5..97f3cf534fb2 100644 --- a/core/src/main/scala/org/apache/spark/ui/storage/RDDPage.scala +++ b/core/src/main/scala/org/apache/spark/ui/storage/RDDPage.scala @@ -35,15 +35,7 @@ private[ui] class RDDPage(parent: SparkUITab, store: AppStatusStore) extends Web val parameterId = request.getParameter("id") require(parameterId != null && parameterId.nonEmpty, "Missing id parameter") - val parameterBlockPage = request.getParameter("block.page") - val parameterBlockSortColumn = request.getParameter("block.sort") - val parameterBlockSortDesc = request.getParameter("block.desc") - val parameterBlockPageSize = request.getParameter("block.pageSize") - - val blockPage = Option(parameterBlockPage).map(_.toInt).getOrElse(1) - val blockSortColumn = Option(parameterBlockSortColumn).getOrElse("Block Name") - val blockSortDesc = Option(parameterBlockSortDesc).map(_.toBoolean).getOrElse(false) - val blockPageSize = Option(parameterBlockPageSize).map(_.toInt).getOrElse(100) + val blockPage = Option(request.getParameter("block.page")).map(_.toInt).getOrElse(1) val rddId = parameterId.toInt val rddStorageInfo = try { @@ -60,11 +52,10 @@ private[ui] class RDDPage(parent: SparkUITab, store: AppStatusStore) extends Web val blockTableHTML = try { val _blockTable = new BlockPagedTable( + request, + "block", UIUtils.prependBaseUri(request, parent.basePath) + s"/storage/rdd/?id=${rddId}", rddStorageInfo.partitions.get, - blockPageSize, - blockSortColumn, - blockSortDesc, store.executorList(true)) _blockTable.table(blockPage) } catch { @@ -216,21 +207,22 @@ private[ui] class BlockDataSource( } private[ui] class BlockPagedTable( + request: HttpServletRequest, + rddTag: String, basePath: String, rddPartitions: Seq[RDDPartitionInfo], - pageSize: Int, - sortColumn: String, - desc: Boolean, executorSummaries: Seq[ExecutorSummary]) extends PagedTable[BlockTableRowData] { + private val (sortColumn, desc, pageSize) = getTableParameters(request, rddTag, "Block Name") + override def tableId: String = "rdd-storage-by-block-table" override def tableCssClass: String = "table table-bordered table-sm table-striped table-head-clickable" - override def pageSizeFormField: String = "block.pageSize" + override def pageSizeFormField: String = s"$rddTag.pageSize" - override def pageNumberFormField: String = "block.page" + override def pageNumberFormField: String = s"$rddTag.page" override val dataSource: BlockDataSource = new BlockDataSource( rddPartitions, @@ -254,46 +246,16 @@ private[ui] class BlockPagedTable( } override def headers: Seq[Node] = { - val blockHeaders = Seq( + val blockHeaders: Seq[(String, Boolean, Option[String])] = Seq( "Block Name", "Storage Level", "Size in Memory", "Size on Disk", - "Executors") + "Executors").map(x => (x, true, None)) - if (!blockHeaders.contains(sortColumn)) { - throw new IllegalArgumentException(s"Unknown column: $sortColumn") - } + isSortColumnValid(blockHeaders, sortColumn) - val headerRow: Seq[Node] = { - blockHeaders.map { header => - if (header == sortColumn) { - val headerLink = Unparsed( - basePath + - s"&block.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&block.desc=${!desc}" + - s"&block.pageSize=$pageSize") - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - - {header} -  {Unparsed(arrow)} - - - } else { - val headerLink = Unparsed( - basePath + - s"&block.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&block.pageSize=$pageSize") - - - {header} - - - } - } - } - {headerRow} + headerRow(blockHeaders, desc, pageSize, sortColumn, basePath, rddTag, "block") } override def row(block: BlockTableRowData): Seq[Node] = { diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/AllExecutionsPage.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/AllExecutionsPage.scala index 5ce052a0ae99..8c23f2cbb86b 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/AllExecutionsPage.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ui/AllExecutionsPage.scala @@ -21,9 +21,8 @@ import java.net.URLEncoder import java.nio.charset.StandardCharsets.UTF_8 import javax.servlet.http.HttpServletRequest -import scala.collection.JavaConverters._ import scala.collection.mutable -import scala.xml.{Node, NodeSeq, Unparsed} +import scala.xml.{Node, NodeSeq} import org.apache.spark.JobExecutionStatus import org.apache.spark.internal.Logging @@ -159,26 +158,8 @@ private[ui] class AllExecutionsPage(parent: SQLTab) extends WebUIPage("") with L showSucceededJobs: Boolean, showFailedJobs: Boolean): Seq[Node] = { - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(executionTag)) - .map { case (name, vals) => - name + "=" + vals(0) - } - - val parameterExecutionPage = request.getParameter(s"$executionTag.page") - val parameterExecutionSortColumn = request.getParameter(s"$executionTag.sort") - val parameterExecutionSortDesc = request.getParameter(s"$executionTag.desc") - val parameterExecutionPageSize = request.getParameter(s"$executionTag.pageSize") - - val executionPage = Option(parameterExecutionPage).map(_.toInt).getOrElse(1) - val executionSortColumn = Option(parameterExecutionSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse("ID") - val executionSortDesc = Option(parameterExecutionSortDesc).map(_.toBoolean).getOrElse( - // New executions should be shown above old executions by default. - executionSortColumn == "ID" - ) - val executionPageSize = Option(parameterExecutionPageSize).map(_.toInt).getOrElse(100) + val executionPage = + Option(request.getParameter(s"$executionTag.page")).map(_.toInt).getOrElse(1) val tableHeaderId = executionTag // "running", "completed" or "failed" @@ -191,11 +172,7 @@ private[ui] class AllExecutionsPage(parent: SQLTab) extends WebUIPage("") with L executionTag, UIUtils.prependBaseUri(request, parent.basePath), "SQL", // subPath - parameterOtherTable, currentTime, - pageSize = executionPageSize, - sortColumn = executionSortColumn, - desc = executionSortDesc, showRunningJobs, showSucceededJobs, showFailedJobs).table(executionPage) @@ -219,15 +196,13 @@ private[ui] class ExecutionPagedTable( executionTag: String, basePath: String, subPath: String, - parameterOtherTable: Iterable[String], currentTime: Long, - pageSize: Int, - sortColumn: String, - desc: Boolean, showRunningJobs: Boolean, showSucceededJobs: Boolean, showFailedJobs: Boolean) extends PagedTable[ExecutionTableRowData] { + private val (sortColumn, desc, pageSize) = getTableParameters(request, executionTag, "ID") + override val dataSource = new ExecutionDataSource( request, parent, @@ -241,7 +216,8 @@ private[ui] class ExecutionPagedTable( showSucceededJobs, showFailedJobs) - private val parameterPath = s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}" + private val parameterPath = + s"$basePath/$subPath/?${getParameterOtherTable(request, executionTag)}" override def tableId: String = s"$executionTag-table" @@ -269,83 +245,32 @@ private[ui] class ExecutionPagedTable( } override def headers: Seq[Node] = { - // Information for each header: title, sortable - val executionHeadersAndCssClasses: Seq[(String, Boolean)] = + // Information for each header: title, sortable, tooltip + val executionHeadersAndCssClasses: Seq[(String, Boolean, Option[String])] = Seq( - ("ID", true), - ("Description", true), - ("Submitted", true), - ("Duration", true)) ++ { + ("ID", true, None), + ("Description", true, None), + ("Submitted", true, None), + ("Duration", true, Some("Time from query submission to completion (or if still executing," + + "time since submission)"))) ++ { if (showRunningJobs && showSucceededJobs && showFailedJobs) { Seq( - ("Running Job IDs", true), - ("Succeeded Job IDs", true), - ("Failed Job IDs", true)) + ("Running Job IDs", true, None), + ("Succeeded Job IDs", true, None), + ("Failed Job IDs", true, None)) } else if (showSucceededJobs && showFailedJobs) { Seq( - ("Succeeded Job IDs", true), - ("Failed Job IDs", true)) + ("Succeeded Job IDs", true, None), + ("Failed Job IDs", true, None)) } else { - Seq(("Job IDs", true)) + Seq(("Job IDs", true, None)) } } - val sortableColumnHeaders = executionHeadersAndCssClasses.filter { - case (_, sortable) => sortable - }.map { case (title, _) => title } - - require(sortableColumnHeaders.contains(sortColumn), s"Unknown column: $sortColumn") - - val headerRow: Seq[Node] = { - executionHeadersAndCssClasses.map { case (header, sortable) => - if (header == sortColumn) { - val headerLink = Unparsed( - parameterPath + - s"&$executionTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$executionTag.desc=${!desc}" + - s"&$executionTag.pageSize=$pageSize" + - s"#$tableHeaderId") - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - - - {header} -  {Unparsed(arrow)} - - - - } else { - if (sortable) { - val headerLink = Unparsed( - parameterPath + - s"&$executionTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$executionTag.pageSize=$pageSize" + - s"#$tableHeaderId") - - - - {if (header == "Duration") { - - {header} - - } else { - {header} - }} - - - } else { - - {header} - - } - } - } - } - - {headerRow} - + isSortColumnValid(executionHeadersAndCssClasses, sortColumn) + + headerRow(executionHeadersAndCssClasses, desc, pageSize, sortColumn, parameterPath, + executionTag, tableHeaderId) } override def row(executionTableRow: ExecutionTableRowData): Seq[Node] = { diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPage.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPage.scala index 7fb755c292b3..1f2d4b1b8777 100644 --- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPage.scala +++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerPage.scala @@ -22,8 +22,7 @@ import java.nio.charset.StandardCharsets.UTF_8 import java.util.Calendar import javax.servlet.http.HttpServletRequest -import scala.collection.JavaConverters._ -import scala.xml.{Node, Unparsed} +import scala.xml.Node import org.apache.commons.text.StringEscapeUtils @@ -78,26 +77,8 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("" val sqlTableTag = "sqlstat" - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(sqlTableTag)) - .map { case (name, vals) => - name + "=" + vals(0) - } - - val parameterSqlTablePage = request.getParameter(s"$sqlTableTag.page") - val parameterSqlTableSortColumn = request.getParameter(s"$sqlTableTag.sort") - val parameterSqlTableSortDesc = request.getParameter(s"$sqlTableTag.desc") - val parameterSqlPageSize = request.getParameter(s"$sqlTableTag.pageSize") - - val sqlTablePage = Option(parameterSqlTablePage).map(_.toInt).getOrElse(1) - val sqlTableSortColumn = Option(parameterSqlTableSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse("Start Time") - val sqlTableSortDesc = Option(parameterSqlTableSortDesc).map(_.toBoolean).getOrElse( - // New executions should be shown above old executions by default. - sqlTableSortColumn == "Start Time" - ) - val sqlTablePageSize = Option(parameterSqlPageSize).map(_.toInt).getOrElse(100) + val sqlTablePage = + Option(request.getParameter(s"$sqlTableTag.page")).map(_.toInt).getOrElse(1) try { Some(new SqlStatsPagedTable( @@ -106,12 +87,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("" store.getExecutionList, "sqlserver", UIUtils.prependBaseUri(request, parent.basePath), - parameterOtherTable, - sqlTableTag, - pageSize = sqlTablePageSize, - sortColumn = sqlTableSortColumn, - desc = sqlTableSortDesc - ).table(sqlTablePage)) + sqlTableTag).table(sqlTablePage)) } catch { case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) => Some(
@@ -146,26 +122,8 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("" val sessionTableTag = "sessionstat" - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(sessionTableTag)) - .map { case (name, vals) => - name + "=" + vals(0) - } - - val parameterSessionTablePage = request.getParameter(s"$sessionTableTag.page") - val parameterSessionTableSortColumn = request.getParameter(s"$sessionTableTag.sort") - val parameterSessionTableSortDesc = request.getParameter(s"$sessionTableTag.desc") - val parameterSessionPageSize = request.getParameter(s"$sessionTableTag.pageSize") - - val sessionTablePage = Option(parameterSessionTablePage).map(_.toInt).getOrElse(1) - val sessionTableSortColumn = Option(parameterSessionTableSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse("Start Time") - val sessionTableSortDesc = Option(parameterSessionTableSortDesc).map(_.toBoolean).getOrElse( - // New session should be shown above old session by default. - (sessionTableSortColumn == "Start Time") - ) - val sessionTablePageSize = Option(parameterSessionPageSize).map(_.toInt).getOrElse(100) + val sessionTablePage = + Option(request.getParameter(s"$sessionTableTag.page")).map(_.toInt).getOrElse(1) try { Some(new SessionStatsPagedTable( @@ -174,11 +132,7 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("" store.getSessionList, "sqlserver", UIUtils.prependBaseUri(request, parent.basePath), - parameterOtherTable, - sessionTableTag, - pageSize = sessionTablePageSize, - sortColumn = sessionTableSortColumn, - desc = sessionTableSortDesc + sessionTableTag ).table(sessionTablePage)) } catch { case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) => @@ -216,15 +170,15 @@ private[ui] class SqlStatsPagedTable( data: Seq[ExecutionInfo], subPath: String, basePath: String, - parameterOtherTable: Iterable[String], - sqlStatsTableTag: String, - pageSize: Int, - sortColumn: String, - desc: Boolean) extends PagedTable[SqlStatsTableRow] { + sqlStatsTableTag: String) extends PagedTable[SqlStatsTableRow] { + + private val (sortColumn, desc, pageSize) = + getTableParameters(request, sqlStatsTableTag, "Start Time") override val dataSource = new SqlStatsTableDataSource(data, pageSize, sortColumn, desc) - private val parameterPath = s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}" + private val parameterPath = + s"$basePath/$subPath/?${getParameterOtherTable(request, sqlStatsTableTag)}" override def tableId: String = sqlStatsTableTag @@ -238,7 +192,8 @@ private[ui] class SqlStatsPagedTable( s"&$pageNumberFormField=$page" + s"&$sqlStatsTableTag.sort=$encodedSortColumn" + s"&$sqlStatsTableTag.desc=$desc" + - s"&$pageSizeFormField=$pageSize" + s"&$pageSizeFormField=$pageSize" + + s"#$sqlStatsTableTag" } override def pageSizeFormField: String = s"$sqlStatsTableTag.pageSize" @@ -247,73 +202,29 @@ private[ui] class SqlStatsPagedTable( override def goButtonFormPath: String = { val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name()) - s"$parameterPath&$sqlStatsTableTag.sort=$encodedSortColumn&$sqlStatsTableTag.desc=$desc" + s"$parameterPath&$sqlStatsTableTag.sort=$encodedSortColumn" + + s"&$sqlStatsTableTag.desc=$desc#$sqlStatsTableTag" } override def headers: Seq[Node] = { - val sqlTableHeaders = Seq("User", "JobID", "GroupID", "Start Time", "Finish Time", - "Close Time", "Execution Time", "Duration", "Statement", "State", "Detail") - - val tooltips = Seq(None, None, None, None, Some(THRIFT_SERVER_FINISH_TIME), - Some(THRIFT_SERVER_CLOSE_TIME), Some(THRIFT_SERVER_EXECUTION), - Some(THRIFT_SERVER_DURATION), None, None, None) - - assert(sqlTableHeaders.length == tooltips.length) - - val headerRow: Seq[Node] = { - sqlTableHeaders.zip(tooltips).map { case (header, tooltip) => - if (header == sortColumn) { - val headerLink = Unparsed( - parameterPath + - s"&$sqlStatsTableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$sqlStatsTableTag.desc=${!desc}" + - s"&$sqlStatsTableTag.pageSize=$pageSize" + - s"#$sqlStatsTableTag") - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - if (tooltip.nonEmpty) { - - - - {header} {Unparsed(arrow)} - - - - } else { - - - {header} {Unparsed(arrow)} - - - } - } else { - val headerLink = Unparsed( - parameterPath + - s"&$sqlStatsTableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$sqlStatsTableTag.pageSize=$pageSize" + - s"#$sqlStatsTableTag") - - if(tooltip.nonEmpty) { - - - - {header} - - - - } else { - - - {header} - - - } - } - } - } - - {headerRow} - + val sqlTableHeadersAndTooltips: Seq[(String, Boolean, Option[String])] = + Seq( + ("User", true, None), + ("JobID", true, None), + ("GroupID", true, None), + ("Start Time", true, None), + ("Finish Time", true, Some(THRIFT_SERVER_FINISH_TIME)), + ("Close Time", true, Some(THRIFT_SERVER_CLOSE_TIME)), + ("Execution Time", true, Some(THRIFT_SERVER_EXECUTION)), + ("Duration", true, Some(THRIFT_SERVER_DURATION)), + ("Statement", true, None), + ("State", true, None), + ("Detail", true, None)) + + isSortColumnValid(sqlTableHeadersAndTooltips, sortColumn) + + headerRow(sqlTableHeadersAndTooltips, desc, pageSize, sortColumn, parameterPath, + sqlStatsTableTag, sqlStatsTableTag) } override def row(sqlStatsTableRow: SqlStatsTableRow): Seq[Node] = { @@ -391,15 +302,15 @@ private[ui] class SessionStatsPagedTable( data: Seq[SessionInfo], subPath: String, basePath: String, - parameterOtherTable: Iterable[String], - sessionStatsTableTag: String, - pageSize: Int, - sortColumn: String, - desc: Boolean) extends PagedTable[SessionInfo] { + sessionStatsTableTag: String) extends PagedTable[SessionInfo] { + + private val (sortColumn, desc, pageSize) = + getTableParameters(request, sessionStatsTableTag, "Start Time") override val dataSource = new SessionStatsTableDataSource(data, pageSize, sortColumn, desc) - private val parameterPath = s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}" + private val parameterPath = + s"$basePath/$subPath/?${getParameterOtherTable(request, sessionStatsTableTag)}" override def tableId: String = sessionStatsTableTag @@ -413,7 +324,8 @@ private[ui] class SessionStatsPagedTable( s"&$pageNumberFormField=$page" + s"&$sessionStatsTableTag.sort=$encodedSortColumn" + s"&$sessionStatsTableTag.desc=$desc" + - s"&$pageSizeFormField=$pageSize" + s"&$pageSizeFormField=$pageSize" + + s"#$sessionStatsTableTag" } override def pageSizeFormField: String = s"$sessionStatsTableTag.pageSize" @@ -422,70 +334,25 @@ private[ui] class SessionStatsPagedTable( override def goButtonFormPath: String = { val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name()) - s"$parameterPath&$sessionStatsTableTag.sort=$encodedSortColumn&$sessionStatsTableTag.desc=$desc" + s"$parameterPath&$sessionStatsTableTag.sort=$encodedSortColumn" + + s"&$sessionStatsTableTag.desc=$desc#$sessionStatsTableTag" } override def headers: Seq[Node] = { - val sessionTableHeaders = - Seq("User", "IP", "Session ID", "Start Time", "Finish Time", "Duration", "Total Execute") - - val tooltips = Seq(None, None, None, None, None, Some(THRIFT_SESSION_DURATION), - Some(THRIFT_SESSION_TOTAL_EXECUTE)) - assert(sessionTableHeaders.length == tooltips.length) - val colWidthAttr = s"${100.toDouble / sessionTableHeaders.size}%" - - val headerRow: Seq[Node] = { - sessionTableHeaders.zip(tooltips).map { case (header, tooltip) => - if (header == sortColumn) { - val headerLink = Unparsed( - parameterPath + - s"&$sessionStatsTableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$sessionStatsTableTag.desc=${!desc}" + - s"&$sessionStatsTableTag.pageSize=$pageSize" + - s"#$sessionStatsTableTag") - val arrow = if (desc) "▾" else "▴" // UP or DOWN - - - { - if (tooltip.nonEmpty) { - - {header} {Unparsed(arrow)} - - } else { - - {header} {Unparsed(arrow)} - - } - } - - - - } else { - val headerLink = Unparsed( - parameterPath + - s"&$sessionStatsTableTag.sort=${URLEncoder.encode(header, UTF_8.name())}" + - s"&$sessionStatsTableTag.pageSize=$pageSize" + - s"#$sessionStatsTableTag") - - - - { - if (tooltip.nonEmpty) { - - {header} - - } else { - {header} - } - } - - - } - } - } - - {headerRow} - + val sessionTableHeadersAndTooltips: Seq[(String, Boolean, Option[String])] = + Seq( + ("User", true, None), + ("IP", true, None), + ("Session ID", true, None), + ("Start Time", true, None), + ("Finish Time", true, None), + ("Duration", true, Some(THRIFT_SESSION_DURATION)), + ("Total Execute", true, Some(THRIFT_SESSION_TOTAL_EXECUTE))) + + isSortColumnValid(sessionTableHeadersAndTooltips, sortColumn) + + headerRow(sessionTableHeadersAndTooltips, desc, pageSize, sortColumn, + parameterPath, sessionStatsTableTag, sessionStatsTableTag) } override def row(session: SessionInfo): Seq[Node] = { diff --git a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerSessionPage.scala b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerSessionPage.scala index 2d7adf552738..87165cc8cac4 100644 --- a/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerSessionPage.scala +++ b/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/ThriftServerSessionPage.scala @@ -19,7 +19,6 @@ package org.apache.spark.sql.hive.thriftserver.ui import javax.servlet.http.HttpServletRequest -import scala.collection.JavaConverters._ import scala.xml.Node import org.apache.spark.internal.Logging @@ -77,26 +76,8 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab) val sqlTableTag = "sqlsessionstat" - val parameterOtherTable = request.getParameterMap().asScala - .filterNot(_._1.startsWith(sqlTableTag)) - .map { case (name, vals) => - name + "=" + vals(0) - } - - val parameterSqlTablePage = request.getParameter(s"$sqlTableTag.page") - val parameterSqlTableSortColumn = request.getParameter(s"$sqlTableTag.sort") - val parameterSqlTableSortDesc = request.getParameter(s"$sqlTableTag.desc") - val parameterSqlPageSize = request.getParameter(s"$sqlTableTag.pageSize") - - val sqlTablePage = Option(parameterSqlTablePage).map(_.toInt).getOrElse(1) - val sqlTableSortColumn = Option(parameterSqlTableSortColumn).map { sortColumn => - UIUtils.decodeURLParameter(sortColumn) - }.getOrElse("Start Time") - val sqlTableSortDesc = Option(parameterSqlTableSortDesc).map(_.toBoolean).getOrElse( - // New executions should be shown above old executions by default. - sqlTableSortColumn == "Start Time" - ) - val sqlTablePageSize = Option(parameterSqlPageSize).map(_.toInt).getOrElse(100) + val sqlTablePage = + Option(request.getParameter(s"$sqlTableTag.page")).map(_.toInt).getOrElse(1) try { Some(new SqlStatsPagedTable( @@ -105,11 +86,7 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab) executionList, "sqlserver/session", UIUtils.prependBaseUri(request, parent.basePath), - parameterOtherTable, - sqlTableTag, - pageSize = sqlTablePageSize, - sortColumn = sqlTableSortColumn, - desc = sqlTableSortDesc + sqlTableTag ).table(sqlTablePage)) } catch { case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) =>