"
+ * Results will be written to "benchmarks/MetricsAggregationBenchmark-results.txt".
+ * }}}
+ */
+object MetricsAggregationBenchmark extends BenchmarkBase {
+
+ private def metricTrackingBenchmark(
+ timer: Benchmark.Timer,
+ numMetrics: Int,
+ numTasks: Int,
+ numStages: Int): Measurements = {
+ val conf = new SparkConf()
+ .set(LIVE_ENTITY_UPDATE_PERIOD, 0L)
+ .set(ASYNC_TRACKING_ENABLED, false)
+ val kvstore = new ElementTrackingStore(new InMemoryStore(), conf)
+ val listener = new SQLAppStatusListener(conf, kvstore, live = true)
+ val store = new SQLAppStatusStore(kvstore, Some(listener))
+
+ val metrics = (0 until numMetrics).map { i =>
+ new SQLMetricInfo(s"metric$i", i.toLong, "average")
+ }
+
+ val planInfo = new SparkPlanInfo(
+ getClass().getName(),
+ getClass().getName(),
+ Nil,
+ Map.empty,
+ metrics)
+
+ val idgen = new AtomicInteger()
+ val executionId = idgen.incrementAndGet()
+ val executionStart = SparkListenerSQLExecutionStart(
+ executionId,
+ getClass().getName(),
+ getClass().getName(),
+ getClass().getName(),
+ planInfo,
+ System.currentTimeMillis())
+
+ val executionEnd = SparkListenerSQLExecutionEnd(executionId, System.currentTimeMillis())
+
+ val properties = new Properties()
+ properties.setProperty(SQLExecution.EXECUTION_ID_KEY, executionId.toString)
+
+ timer.startTiming()
+ listener.onOtherEvent(executionStart)
+
+ val taskEventsTime = (0 until numStages).map { _ =>
+ val stageInfo = new StageInfo(idgen.incrementAndGet(), 0, getClass().getName(),
+ numTasks, Nil, Nil, getClass().getName())
+
+ val jobId = idgen.incrementAndGet()
+ val jobStart = SparkListenerJobStart(
+ jobId = jobId,
+ time = System.currentTimeMillis(),
+ stageInfos = Seq(stageInfo),
+ properties)
+
+ val stageStart = SparkListenerStageSubmitted(stageInfo)
+
+ val taskOffset = idgen.incrementAndGet().toLong
+ val taskEvents = (0 until numTasks).map { i =>
+ val info = new TaskInfo(
+ taskId = taskOffset + i.toLong,
+ index = i,
+ attemptNumber = 0,
+ // The following fields are not used.
+ launchTime = 0,
+ executorId = "",
+ host = "",
+ taskLocality = null,
+ speculative = false)
+ info.markFinished(TaskState.FINISHED, 1L)
+
+ val accumulables = (0 until numMetrics).map { mid =>
+ val acc = new LongAccumulator
+ acc.metadata = AccumulatorMetadata(mid, None, false)
+ acc.toInfo(Some(i.toLong), None)
+ }
+
+ info.setAccumulables(accumulables)
+
+ val start = SparkListenerTaskStart(stageInfo.stageId, stageInfo.attemptNumber, info)
+ val end = SparkListenerTaskEnd(stageInfo.stageId, stageInfo.attemptNumber,
+ taskType = "",
+ reason = null,
+ info,
+ new ExecutorMetrics(),
+ null)
+
+ (start, end)
+ }
+
+ val jobEnd = SparkListenerJobEnd(
+ jobId = jobId,
+ time = System.currentTimeMillis(),
+ JobSucceeded)
+
+ listener.onJobStart(jobStart)
+ listener.onStageSubmitted(stageStart)
+
+ val (_, _taskEventsTime) = Utils.timeTakenMs {
+ taskEvents.foreach { case (start, end) =>
+ listener.onTaskStart(start)
+ listener.onTaskEnd(end)
+ }
+ }
+
+ listener.onJobEnd(jobEnd)
+ _taskEventsTime
+ }
+
+ val (_, aggTime) = Utils.timeTakenMs {
+ listener.onOtherEvent(executionEnd)
+ val metrics = store.executionMetrics(executionId)
+ assert(metrics.size == numMetrics, s"${metrics.size} != $numMetrics")
+ }
+
+ timer.stopTiming()
+ kvstore.close()
+
+ Measurements(taskEventsTime, aggTime)
+ }
+
+ override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
+ val metricCount = 50
+ val taskCount = 100000
+ val stageCounts = Seq(1, 2, 3)
+
+ val benchmark = new Benchmark(
+ s"metrics aggregation ($metricCount metrics, $taskCount tasks per stage)", 1,
+ warmupTime = 0.seconds, output = output)
+
+ // Run this outside the measurement code so that classes are loaded and JIT is triggered,
+ // otherwise the first run tends to be much slower than others. Also because this benchmark is a
+ // bit weird and doesn't really map to what the Benchmark class expects, so it's a bit harder
+ // to use warmupTime and friends effectively.
+ stageCounts.foreach { count =>
+ metricTrackingBenchmark(new Benchmark.Timer(-1), metricCount, taskCount, count)
+ }
+
+ val measurements = mutable.HashMap[Int, Seq[Measurements]]()
+
+ stageCounts.foreach { count =>
+ benchmark.addTimerCase(s"$count stage(s)") { timer =>
+ val m = metricTrackingBenchmark(timer, metricCount, taskCount, count)
+ val all = measurements.getOrElse(count, Nil)
+ measurements(count) = all ++ Seq(m)
+ }
+ }
+
+ benchmark.run()
+
+ benchmark.out.printf("Stage Count Stage Proc. Time Aggreg. Time\n")
+ stageCounts.foreach { count =>
+ val data = measurements(count)
+ val eventsTimes = data.flatMap(_.taskEventsTimes)
+ val aggTimes = data.map(_.aggregationTime)
+
+ val msg = " %d %d %d\n".format(
+ count,
+ eventsTimes.sum / eventsTimes.size,
+ aggTimes.sum / aggTimes.size)
+ benchmark.out.printf(msg)
+ }
+ }
+
+ /**
+ * Finer-grained measurements of how long it takes to run some parts of the benchmark. This is
+ * collected by the benchmark method, so this collection slightly affects the overall benchmark
+ * results, but this data helps with seeing where the time is going, since this benchmark is
+ * triggering a whole lot of code in the listener class.
+ */
+ case class Measurements(
+ taskEventsTimes: Seq[Long],
+ aggregationTime: Long)
+}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
index 88864ccec7523..b8c0935b33a87 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListenerSuite.scala
@@ -79,9 +79,9 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
private def createStageInfo(stageId: Int, attemptId: Int): StageInfo = {
new StageInfo(stageId = stageId,
attemptId = attemptId,
+ numTasks = 8,
// The following fields are not used in tests
name = "",
- numTasks = 0,
rddInfos = Nil,
parentIds = Nil,
details = "")
@@ -94,8 +94,8 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
val info = new TaskInfo(
taskId = taskId,
attemptNumber = attemptNumber,
+ index = taskId.toInt,
// The following fields are not used in tests
- index = 0,
launchTime = 0,
executorId = "",
host = "",
@@ -190,6 +190,8 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
),
createProperties(executionId)))
listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(0, 0)))
+ listener.onTaskStart(SparkListenerTaskStart(0, 0, createTaskInfo(0, 0)))
+ listener.onTaskStart(SparkListenerTaskStart(0, 0, createTaskInfo(1, 0)))
assert(statusStore.executionMetrics(executionId).isEmpty)
@@ -217,6 +219,8 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
// Retrying a stage should reset the metrics
listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(0, 1)))
+ listener.onTaskStart(SparkListenerTaskStart(0, 1, createTaskInfo(0, 0)))
+ listener.onTaskStart(SparkListenerTaskStart(0, 1, createTaskInfo(1, 0)))
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
// (task id, stage id, stage attempt, accum updates)
@@ -260,6 +264,8 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
// Summit a new stage
listener.onStageSubmitted(SparkListenerStageSubmitted(createStageInfo(1, 0)))
+ listener.onTaskStart(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0)))
+ listener.onTaskStart(SparkListenerTaskStart(1, 0, createTaskInfo(1, 0)))
listener.onExecutorMetricsUpdate(SparkListenerExecutorMetricsUpdate("", Seq(
// (task id, stage id, stage attempt, accum updates)
@@ -490,8 +496,8 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
val statusStore = spark.sharedState.statusStore
val oldCount = statusStore.executionsList().size
- val expectedAccumValue = 12345
- val expectedAccumValue2 = 54321
+ val expectedAccumValue = 12345L
+ val expectedAccumValue2 = 54321L
val physicalPlan = MyPlan(sqlContext.sparkContext, expectedAccumValue, expectedAccumValue2)
val dummyQueryExecution = new QueryExecution(spark, LocalRelation()) {
override lazy val sparkPlan = physicalPlan
@@ -517,8 +523,9 @@ class SQLAppStatusListenerSuite extends SharedSparkSession with JsonTestUtils
val metrics = statusStore.executionMetrics(execId)
val driverMetric = physicalPlan.metrics("dummy")
val driverMetric2 = physicalPlan.metrics("dummy2")
- val expectedValue = SQLMetrics.stringValue(driverMetric.metricType, Seq(expectedAccumValue))
- val expectedValue2 = SQLMetrics.stringValue(driverMetric2.metricType, Seq(expectedAccumValue2))
+ val expectedValue = SQLMetrics.stringValue(driverMetric.metricType, Array(expectedAccumValue))
+ val expectedValue2 = SQLMetrics.stringValue(driverMetric2.metricType,
+ Array(expectedAccumValue2))
assert(metrics.contains(driverMetric.id))
assert(metrics(driverMetric.id) === expectedValue)
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
index 3c8ce0a3fc3e4..715534b0458dc 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala
@@ -450,15 +450,6 @@ class JDBCSuite extends QueryTest
urlWithUserAndPass, "TEST.PEOPLE", new Properties()).collect().length === 3)
}
- test("Basic API with illegal fetchsize") {
- val properties = new Properties()
- properties.setProperty(JDBCOptions.JDBC_BATCH_FETCH_SIZE, "-1")
- val e = intercept[IllegalArgumentException] {
- spark.read.jdbc(urlWithUserAndPass, "TEST.PEOPLE", properties).collect()
- }.getMessage
- assert(e.contains("Invalid value `-1` for parameter `fetchsize`"))
- }
-
test("Missing partition columns") {
withView("tempPeople") {
val e = intercept[IllegalArgumentException] {
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
index b28c6531d42b2..8021ef1a17a18 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCWriteSuite.scala
@@ -21,10 +21,12 @@ import java.sql.DriverManager
import java.util.Properties
import scala.collection.JavaConverters.propertiesAsScalaMapConverter
+import scala.collection.mutable.ArrayBuffer
import org.scalatest.BeforeAndAfter
import org.apache.spark.SparkException
+import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
import org.apache.spark.sql.{AnalysisException, DataFrame, Row, SaveMode}
import org.apache.spark.sql.catalyst.parser.ParseException
import org.apache.spark.sql.execution.datasources.jdbc.{JDBCOptions, JdbcUtils}
@@ -543,4 +545,57 @@ class JDBCWriteSuite extends SharedSparkSession with BeforeAndAfter {
}.getMessage
assert(errMsg.contains("Statement was canceled or the session timed out"))
}
+
+ test("metrics") {
+ val df = spark.createDataFrame(sparkContext.parallelize(arr2x2), schema2)
+ val df2 = spark.createDataFrame(sparkContext.parallelize(arr1x2), schema2)
+
+ runAndVerifyRecordsWritten(2) {
+ df.write.mode(SaveMode.Append).jdbc(url, "TEST.BASICCREATETEST", new Properties())
+ }
+
+ runAndVerifyRecordsWritten(1) {
+ df2.write.mode(SaveMode.Overwrite).jdbc(url, "TEST.BASICCREATETEST", new Properties())
+ }
+
+ runAndVerifyRecordsWritten(1) {
+ df2.write.mode(SaveMode.Overwrite).option("truncate", true)
+ .jdbc(url, "TEST.BASICCREATETEST", new Properties())
+ }
+
+ runAndVerifyRecordsWritten(0) {
+ intercept[AnalysisException] {
+ df2.write.mode(SaveMode.ErrorIfExists).jdbc(url, "TEST.BASICCREATETEST", new Properties())
+ }
+ }
+
+ runAndVerifyRecordsWritten(0) {
+ df.write.mode(SaveMode.Ignore).jdbc(url, "TEST.BASICCREATETEST", new Properties())
+ }
+ }
+
+ private def runAndVerifyRecordsWritten(expected: Long)(job: => Unit): Unit = {
+ assert(expected === runAndReturnMetrics(job, _.taskMetrics.outputMetrics.recordsWritten))
+ }
+
+ private def runAndReturnMetrics(job: => Unit, collector: (SparkListenerTaskEnd) => Long): Long = {
+ val taskMetrics = new ArrayBuffer[Long]()
+
+ // Avoid receiving earlier taskEnd events
+ sparkContext.listenerBus.waitUntilEmpty()
+
+ val listener = new SparkListener() {
+ override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
+ taskMetrics += collector(taskEnd)
+ }
+ }
+ sparkContext.addSparkListener(listener)
+
+ job
+
+ sparkContext.listenerBus.waitUntilEmpty()
+
+ sparkContext.removeSparkListener(listener)
+ taskMetrics.sum
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala
index b26d2556b2e36..09580b94056b4 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamingQueryManagerSuite.scala
@@ -17,6 +17,7 @@
package org.apache.spark.sql.streaming
+import java.io.File
import java.util.concurrent.CountDownLatch
import scala.concurrent.Future
@@ -28,7 +29,7 @@ import org.scalatest.time.Span
import org.scalatest.time.SpanSugar._
import org.apache.spark.SparkException
-import org.apache.spark.sql.Dataset
+import org.apache.spark.sql.{Dataset, Encoders}
import org.apache.spark.sql.execution.datasources.v2.StreamingDataSourceV2Relation
import org.apache.spark.sql.execution.streaming._
import org.apache.spark.sql.streaming.util.BlockingSource
@@ -242,6 +243,83 @@ class StreamingQueryManagerSuite extends StreamTest {
}
}
+ testQuietly("can't start a streaming query with the same name in the same session") {
+ val ds1 = makeDataset._2
+ val ds2 = makeDataset._2
+ val queryName = "abc"
+
+ val query1 = ds1.writeStream.format("noop").queryName(queryName).start()
+ try {
+ val e = intercept[IllegalArgumentException] {
+ ds2.writeStream.format("noop").queryName(queryName).start()
+ }
+ assert(e.getMessage.contains("query with that name is already active"))
+ } finally {
+ query1.stop()
+ }
+ }
+
+ testQuietly("can start a streaming query with the same name in a different session") {
+ val session2 = spark.cloneSession()
+
+ val ds1 = MemoryStream(Encoders.INT, spark.sqlContext).toDS()
+ val ds2 = MemoryStream(Encoders.INT, session2.sqlContext).toDS()
+ val queryName = "abc"
+
+ val query1 = ds1.writeStream.format("noop").queryName(queryName).start()
+ val query2 = ds2.writeStream.format("noop").queryName(queryName).start()
+
+ query1.stop()
+ query2.stop()
+ }
+
+ testQuietly("can't start multiple instances of the same streaming query in the same session") {
+ withTempDir { dir =>
+ val (ms1, ds1) = makeDataset
+ val (ms2, ds2) = makeDataset
+ val chkLocation = new File(dir, "_checkpoint").getCanonicalPath
+ val dataLocation = new File(dir, "data").getCanonicalPath
+
+ val query1 = ds1.writeStream.format("parquet")
+ .option("checkpointLocation", chkLocation).start(dataLocation)
+ ms1.addData(1, 2, 3)
+ try {
+ val e = intercept[IllegalStateException] {
+ ds2.writeStream.format("parquet")
+ .option("checkpointLocation", chkLocation).start(dataLocation)
+ }
+ assert(e.getMessage.contains("same id"))
+ } finally {
+ query1.stop()
+ }
+ }
+ }
+
+ testQuietly(
+ "can't start multiple instances of the same streaming query in the different sessions") {
+ withTempDir { dir =>
+ val session2 = spark.cloneSession()
+
+ val ms1 = MemoryStream(Encoders.INT, spark.sqlContext)
+ val ds2 = MemoryStream(Encoders.INT, session2.sqlContext).toDS()
+ val chkLocation = new File(dir, "_checkpoint").getCanonicalPath
+ val dataLocation = new File(dir, "data").getCanonicalPath
+
+ val query1 = ms1.toDS().writeStream.format("parquet")
+ .option("checkpointLocation", chkLocation).start(dataLocation)
+ ms1.addData(1, 2, 3)
+ try {
+ val e = intercept[IllegalStateException] {
+ ds2.writeStream.format("parquet")
+ .option("checkpointLocation", chkLocation).start(dataLocation)
+ }
+ assert(e.getMessage.contains("same id"))
+ } finally {
+ query1.stop()
+ }
+ }
+ }
+
/** Run a body of code by defining a query on each dataset */
private def withQueriesOn(datasets: Dataset[_]*)(body: Seq[StreamingQuery] => Unit): Unit = {
failAfter(streamingTimeout) {
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 4056be4769d21..7258978e3bada 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
@@ -17,19 +17,22 @@
package org.apache.spark.sql.hive.thriftserver.ui
+import java.net.URLEncoder
+import java.nio.charset.StandardCharsets.UTF_8
import java.util.Calendar
import javax.servlet.http.HttpServletRequest
-import scala.xml.Node
+import scala.collection.JavaConverters._
+import scala.xml.{Node, Unparsed}
import org.apache.commons.text.StringEscapeUtils
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{ExecutionInfo, ExecutionState, SessionInfo}
+import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{ExecutionInfo, SessionInfo}
import org.apache.spark.sql.hive.thriftserver.ui.ToolTips._
import org.apache.spark.ui._
import org.apache.spark.ui.UIUtils._
-
+import org.apache.spark.util.Utils
/** Page for Spark Web UI that shows statistics of the thrift server */
private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage("") with Logging {
@@ -69,45 +72,56 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
/** Generate stats of batch statements of the thrift server program */
private def generateSQLStatsTable(request: HttpServletRequest): Seq[Node] = {
+
val numStatement = listener.getExecutionList.size
+
val table = if (numStatement > 0) {
- val headerRow = 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(headerRow.length == tooltips.length)
- val dataRows = listener.getExecutionList.sortBy(_.startTimestamp).reverse
-
- def generateDataRow(info: ExecutionInfo): Seq[Node] = {
- val jobLink = info.jobId.map { id: String =>
-
- [{id}]
-
+
+ val sqlTableTag = "sqlstat"
+
+ val parameterOtherTable = request.getParameterMap().asScala
+ .filterNot(_._1.startsWith(sqlTableTag))
+ .map { case (name, vals) =>
+ name + "=" + vals(0)
}
- val detail = Option(info.detail).filter(!_.isEmpty).getOrElse(info.executePlan)
-
- | {info.userName} |
-
- {jobLink}
- |
- {info.groupId} |
- {formatDate(info.startTimestamp)} |
- {if (info.finishTimestamp > 0) formatDate(info.finishTimestamp)} |
- {if (info.closeTimestamp > 0) formatDate(info.closeTimestamp)} |
-
- {formatDurationOption(Some(info.totalTime(info.finishTimestamp)))} |
-
- {formatDurationOption(Some(info.totalTime(info.closeTimestamp)))} |
- {info.statement} |
- {info.state} |
- {errorMessageCell(detail)}
-
- }
- Some(UIUtils.listingTable(headerRow, generateDataRow,
- dataRows, false, None, Seq(null), false, tooltipHeaders = tooltips))
+ 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)
+
+ try {
+ Some(new SqlStatsPagedTable(
+ request,
+ parent,
+ listener.getExecutionList,
+ "sqlserver",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ parameterOtherTable,
+ sqlTableTag,
+ pageSize = sqlTablePageSize,
+ sortColumn = sqlTableSortColumn,
+ desc = sqlTableSortDesc
+ ).table(sqlTablePage))
+ } catch {
+ case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ Some(
+
Error while rendering job table:
+
+ {Utils.exceptionString(e)}
+
+
{errorSummary}{details} |
- }
-
/** Generate stats of batch sessions of the thrift server program */
private def generateSessionStatsTable(request: HttpServletRequest): Seq[Node] = {
val sessionList = listener.getSessionList
@@ -185,20 +175,262 @@ private[ui] class ThriftServerPage(parent: ThriftServerTab) extends WebUIPage(""
content
}
-
/**
* Returns a human-readable string representing a duration such as "5 second 35 ms"
*/
private def formatDurationOption(msOption: Option[Long]): String = {
msOption.map(formatDurationVerbose).getOrElse(emptyCell)
}
+}
+
+private[ui] class SqlStatsPagedTable(
+ request: HttpServletRequest,
+ parent: ThriftServerTab,
+ data: Seq[ExecutionInfo],
+ subPath: String,
+ basePath: String,
+ parameterOtherTable: Iterable[String],
+ sqlStatsTableTag: String,
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean) extends PagedTable[SqlStatsTableRow] {
+
+ override val dataSource = new SqlStatsTableDataSource(data, pageSize, sortColumn, desc)
+
+ private val parameterPath = s"$basePath/$subPath/?${parameterOtherTable.mkString("&")}"
+
+ override def tableId: String = sqlStatsTableTag
+
+ override def tableCssClass: String =
+ "table table-bordered table-condensed table-striped " +
+ "table-head-clickable table-cell-width-limited"
+
+ override def pageLink(page: Int): String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+ parameterPath +
+ s"&$pageNumberFormField=$page" +
+ s"&$sqlStatsTableTag.sort=$encodedSortColumn" +
+ s"&$sqlStatsTableTag.desc=$desc" +
+ s"&$pageSizeFormField=$pageSize"
+ }
+
+ override def pageSizeFormField: String = s"$sqlStatsTableTag.pageSize"
+
+ override def pageNumberFormField: String = s"$sqlStatsTableTag.page"
+
+ override def goButtonFormPath: String = {
+ val encodedSortColumn = URLEncoder.encode(sortColumn, UTF_8.name())
+ s"$parameterPath&$sqlStatsTableTag.sort=$encodedSortColumn&$sqlStatsTableTag.desc=$desc"
+ }
+
+ 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}
+
+ }
+
+ override def row(sqlStatsTableRow: SqlStatsTableRow): Seq[Node] = {
+ val info = sqlStatsTableRow.executionInfo
+ val startTime = info.startTimestamp
+ val executionTime = sqlStatsTableRow.executionTime
+ val duration = sqlStatsTableRow.duration
- /** Generate HTML table from string data */
- private def listingTable(headers: Seq[String], data: Seq[Seq[String]]) = {
- def generateDataRow(data: Seq[String]): Seq[Node] = {
- {data.map(d => | {d} | )}
+ def jobLinks(jobData: Seq[String]): Seq[Node] = {
+ jobData.map { jobId =>
+ [{jobId.toString}]
+ }
}
- UIUtils.listingTable(headers, generateDataRow, data, fixedWidth = true)
+
+
+ |
+ {info.userName}
+ |
+
+ {jobLinks(sqlStatsTableRow.jobId)}
+ |
+
+ {info.groupId}
+ |
+
+ {UIUtils.formatDate(startTime)}
+ |
+
+ {if (info.finishTimestamp > 0) formatDate(info.finishTimestamp)}
+ |
+
+ {if (info.closeTimestamp > 0) formatDate(info.closeTimestamp)}
+ |
+
+
+ {formatDurationVerbose(executionTime)}
+ |
+
+ {formatDurationVerbose(duration)}
+ |
+
+ {info.statement}
+ |
+
+ {info.state}
+ |
+ {errorMessageCell(sqlStatsTableRow.detail)}
+
}
+
+
+ private def errorMessageCell(errorMessage: String): Seq[Node] = {
+ val isMultiline = errorMessage.indexOf('\n') >= 0
+ val errorSummary = StringEscapeUtils.escapeHtml4(
+ if (isMultiline) {
+ errorMessage.substring(0, errorMessage.indexOf('\n'))
+ } else {
+ errorMessage
+ })
+ val details = if (isMultiline) {
+ // scalastyle:off
+
+ + details
+ ++
+
+ // scalastyle:on
+ } else {
+ ""
+ }
+
+ {errorSummary}{details}
+ |
+ }
+
+ private def jobURL(request: HttpServletRequest, jobId: String): String =
+ "%s/jobs/job/?id=%s".format(UIUtils.prependBaseUri(request, parent.basePath), jobId)
}
+ private[ui] class SqlStatsTableRow(
+ val jobId: Seq[String],
+ val duration: Long,
+ val executionTime: Long,
+ val executionInfo: ExecutionInfo,
+ val detail: String)
+
+ private[ui] class SqlStatsTableDataSource(
+ info: Seq[ExecutionInfo],
+ pageSize: Int,
+ sortColumn: String,
+ desc: Boolean) extends PagedDataSource[SqlStatsTableRow](pageSize) {
+
+ // Convert ExecutionInfo to SqlStatsTableRow 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 = info.map(sqlStatsTableRow).sorted(ordering(sortColumn, desc))
+
+ private var _slicedStartTime: Set[Long] = null
+
+ override def dataSize: Int = data.size
+
+ override def sliceData(from: Int, to: Int): Seq[SqlStatsTableRow] = {
+ val r = data.slice(from, to)
+ _slicedStartTime = r.map(_.executionInfo.startTimestamp).toSet
+ r
+ }
+
+ private def sqlStatsTableRow(executionInfo: ExecutionInfo): SqlStatsTableRow = {
+ val duration = executionInfo.totalTime(executionInfo.closeTimestamp)
+ val executionTime = executionInfo.totalTime(executionInfo.finishTimestamp)
+ val detail = Option(executionInfo.detail).filter(!_.isEmpty)
+ .getOrElse(executionInfo.executePlan)
+ val jobId = executionInfo.jobId.toSeq.sorted
+
+ new SqlStatsTableRow(jobId, duration, executionTime, executionInfo, detail)
+
+ }
+
+ /**
+ * Return Ordering according to sortColumn and desc.
+ */
+ private def ordering(sortColumn: String, desc: Boolean): Ordering[SqlStatsTableRow] = {
+ val ordering: Ordering[SqlStatsTableRow] = sortColumn match {
+ case "User" => Ordering.by(_.executionInfo.userName)
+ case "JobID" => Ordering by (_.jobId.headOption)
+ case "GroupID" => Ordering.by(_.executionInfo.groupId)
+ case "Start Time" => Ordering.by(_.executionInfo.startTimestamp)
+ case "Finish Time" => Ordering.by(_.executionInfo.finishTimestamp)
+ case "Close Time" => Ordering.by(_.executionInfo.closeTimestamp)
+ case "Execution Time" => Ordering.by(_.executionTime)
+ case "Duration" => Ordering.by(_.duration)
+ case "Statement" => Ordering.by(_.executionInfo.statement)
+ case "State" => Ordering.by(_.executionInfo.state)
+ case "Detail" => Ordering.by(_.detail)
+ case unknownColumn => throw new IllegalArgumentException(s"Unknown column: $unknownColumn")
+ }
+ if (desc) {
+ ordering.reverse
+ } else {
+ ordering
+ }
+ }
+
+ }
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 0aa0a2b8335d8..8b275f8f7be0d 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
@@ -20,15 +20,13 @@ package org.apache.spark.sql.hive.thriftserver.ui
import java.util.Calendar
import javax.servlet.http.HttpServletRequest
+import scala.collection.JavaConverters._
import scala.xml.Node
-import org.apache.commons.text.StringEscapeUtils
-
import org.apache.spark.internal.Logging
-import org.apache.spark.sql.hive.thriftserver.HiveThriftServer2.{ExecutionInfo, ExecutionState}
-import org.apache.spark.sql.hive.thriftserver.ui.ToolTips._
import org.apache.spark.ui._
import org.apache.spark.ui.UIUtils._
+import org.apache.spark.util.Utils
/** Page for Spark Web UI that shows statistics of jobs running in the thrift server */
private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
@@ -36,7 +34,6 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
private val listener = parent.listener
private val startTime = Calendar.getInstance().getTime()
- private val emptyCell = "-"
/** Render the page */
def render(request: HttpServletRequest): Seq[Node] = {
@@ -80,45 +77,52 @@ private[ui] class ThriftServerSessionPage(parent: ThriftServerTab)
.filter(_.sessionId == sessionID)
val numStatement = executionList.size
val table = if (numStatement > 0) {
- val headerRow = 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(headerRow.length == tooltips.length)
- val dataRows = executionList.sortBy(_.startTimestamp).reverse
-
- def generateDataRow(info: ExecutionInfo): Seq[Node] = {
- val jobLink = info.jobId.map { id: String =>
-
- [{id}]
-
+
+ val sqlTableTag = "sqlsessionstat"
+
+ val parameterOtherTable = request.getParameterMap().asScala
+ .filterNot(_._1.startsWith(sqlTableTag))
+ .map { case (name, vals) =>
+ name + "=" + vals(0)
}
- val detail = Option(info.detail).filter(!_.isEmpty).getOrElse(info.executePlan)
-
- | {info.userName} |
-
- {jobLink}
- |
- {info.groupId} |
- {formatDate(info.startTimestamp)} |
- {if (info.finishTimestamp > 0) formatDate(info.finishTimestamp)} |
- {if (info.closeTimestamp > 0) formatDate(info.closeTimestamp)} |
-
- {formatDurationOption(Some(info.totalTime(info.finishTimestamp)))}
- |
-
- {formatDurationOption(Some(info.totalTime(info.closeTimestamp)))}
- |
- {info.statement} |
- {info.state} |
- {errorMessageCell(detail)}
-
- }
- Some(UIUtils.listingTable(headerRow, generateDataRow,
- dataRows, false, None, Seq(null), false, tooltipHeaders = tooltips))
+ 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)
+
+ try {
+ Some(new SqlStatsPagedTable(
+ request,
+ parent,
+ executionList,
+ "sqlserver/session",
+ UIUtils.prependBaseUri(request, parent.basePath),
+ parameterOtherTable,
+ sqlTableTag,
+ pageSize = sqlTablePageSize,
+ sortColumn = sqlTableSortColumn,
+ desc = sqlTableSortDesc
+ ).table(sqlTablePage))
+ } catch {
+ case e@(_: IllegalArgumentException | _: IndexOutOfBoundsException) =>
+ Some(
+
Error while rendering job table:
+
+ {Utils.exceptionString(e)}
+
+
{errorSummary}{details} |
- }
-
- /**
- * Returns a human-readable string representing a duration such as "5 second 35 ms"
- */
- private def formatDurationOption(msOption: Option[Long]): String = {
- msOption.map(formatDurationVerbose).getOrElse(emptyCell)
- }
-
- /** Generate HTML table from string data */
- private def listingTable(headers: Seq[String], data: Seq[Seq[String]]) = {
- def generateDataRow(data: Seq[String]): Seq[Node] = {
- {data.map(d => | {d} | )}
- }
- UIUtils.listingTable(headers, generateDataRow, data, fixedWidth = true)
- }
}
diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerQueryTestSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerQueryTestSuite.scala
index abe91a280a14f..04b1de00ccbff 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerQueryTestSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerQueryTestSuite.scala
@@ -18,7 +18,7 @@
package org.apache.spark.sql.hive.thriftserver
import java.io.File
-import java.sql.{DriverManager, Statement, Timestamp}
+import java.sql.{DriverManager, SQLException, Statement, Timestamp}
import java.util.{Locale, MissingFormatArgumentException}
import scala.util.{Random, Try}
@@ -75,11 +75,6 @@ class ThriftServerQueryTestSuite extends SQLQueryTestSuite {
}
}
- override def sparkConf: SparkConf = super.sparkConf
- // Hive Thrift server should not executes SQL queries in an asynchronous way
- // because we may set session configuration.
- .set(HiveUtils.HIVE_THRIFT_SERVER_ASYNC, false)
-
override val isTestWithConfigSets = false
/** List of test cases to ignore, in lower cases. */
@@ -208,6 +203,12 @@ class ThriftServerQueryTestSuite extends SQLQueryTestSuite {
s"Exception did not match for query #$i\n${expected.sql}, " +
s"expected: ${expected.output}, but got: ${output.output}")
+ // SQLException should not exactly match. We only assert the result contains Exception.
+ case _ if output.output.startsWith(classOf[SQLException].getName) =>
+ assert(expected.output.contains("Exception"),
+ s"Exception did not match for query #$i\n${expected.sql}, " +
+ s"expected: ${expected.output}, but got: ${output.output}")
+
case _ =>
assertResult(expected.output, s"Result did not match for query #$i\n${expected.sql}") {
output.output
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
index 4cbc03d05c9e3..7b3fb68174234 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/CachedTableSuite.scala
@@ -300,7 +300,7 @@ class CachedTableSuite extends QueryTest with SQLTestUtils with TestHiveSingleto
val e = intercept[ParseException] {
sql(s"CACHE TABLE $db.cachedTable AS SELECT 1")
}.getMessage
- assert(e.contains("It is not allowed to add database prefix ") &&
+ assert(e.contains("It is not allowed to add catalog/namespace prefix ") &&
e.contains("to the table name in CACHE TABLE AS SELECT"))
}
}
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
index 116a62b2f6dc6..3e5834f33ea52 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/SQLQuerySuite.scala
@@ -42,7 +42,6 @@ import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.StaticSQLConf.GLOBAL_TEMP_DATABASE
import org.apache.spark.sql.test.SQLTestUtils
import org.apache.spark.sql.types._
-import org.apache.spark.unsafe.types.CalendarInterval
case class Nested1(f1: Nested2)
case class Nested2(f2: Nested3)
@@ -1183,51 +1182,6 @@ class SQLQuerySuite extends QueryTest with SQLTestUtils with TestHiveSingleton {
checkAnswer(sql("SELECT a.`c.b`, `b.$q`[0].`a@!.q`, `q.w`.`w.i&`[0] FROM t"), Row(1, 1, 1))
}
- test("Convert hive interval term into Literal of CalendarIntervalType") {
- checkAnswer(sql("select interval '0 0:0:0.1' day to second"),
- Row(CalendarInterval.fromString("interval 100 milliseconds")))
- checkAnswer(sql("select interval '10-9' year to month"),
- Row(CalendarInterval.fromString("interval 10 years 9 months")))
- checkAnswer(sql("select interval '20 15:40:32.99899999' day to hour"),
- Row(CalendarInterval.fromString("interval 2 weeks 6 days 15 hours")))
- checkAnswer(sql("select interval '20 15:40:32.99899999' day to minute"),
- Row(CalendarInterval.fromString("interval 2 weeks 6 days 15 hours 40 minutes")))
- checkAnswer(sql("select interval '20 15:40:32.99899999' day to second"),
- Row(CalendarInterval.fromString("interval 2 weeks 6 days 15 hours 40 minutes " +
- "32 seconds 998 milliseconds 999 microseconds")))
- checkAnswer(sql("select interval '15:40:32.99899999' hour to minute"),
- Row(CalendarInterval.fromString("interval 15 hours 40 minutes")))
- checkAnswer(sql("select interval '15:40.99899999' hour to second"),
- Row(CalendarInterval.fromString("interval 15 minutes 40 seconds 998 milliseconds " +
- "999 microseconds")))
- checkAnswer(sql("select interval '15:40' hour to second"),
- Row(CalendarInterval.fromString("interval 15 hours 40 minutes")))
- checkAnswer(sql("select interval '15:40:32.99899999' hour to second"),
- Row(CalendarInterval.fromString("interval 15 hours 40 minutes 32 seconds 998 milliseconds " +
- "999 microseconds")))
- checkAnswer(sql("select interval '20 40:32.99899999' minute to second"),
- Row(CalendarInterval.fromString("interval 2 weeks 6 days 40 minutes 32 seconds " +
- "998 milliseconds 999 microseconds")))
- checkAnswer(sql("select interval '40:32.99899999' minute to second"),
- Row(CalendarInterval.fromString("interval 40 minutes 32 seconds 998 milliseconds " +
- "999 microseconds")))
- checkAnswer(sql("select interval '40:32' minute to second"),
- Row(CalendarInterval.fromString("interval 40 minutes 32 seconds")))
- checkAnswer(sql("select interval '30' year"),
- Row(CalendarInterval.fromString("interval 30 years")))
- checkAnswer(sql("select interval '25' month"),
- Row(CalendarInterval.fromString("interval 25 months")))
- checkAnswer(sql("select interval '-100' day"),
- Row(CalendarInterval.fromString("interval -14 weeks -2 days")))
- checkAnswer(sql("select interval '40' hour"),
- Row(CalendarInterval.fromString("interval 1 days 16 hours")))
- checkAnswer(sql("select interval '80' minute"),
- Row(CalendarInterval.fromString("interval 1 hour 20 minutes")))
- checkAnswer(sql("select interval '299.889987299' second"),
- Row(CalendarInterval.fromString(
- "interval 4 minutes 59 seconds 889 milliseconds 987 microseconds")))
- }
-
test("specifying database name for a temporary view is not allowed") {
withTempPath { dir =>
withTempView("db.t") {