[SPARK-27142][SQL] Provide REST API for SQL information

sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusStore.scala

@@ -34,14 +34,16 @@ import org.apache.spark.util.kvstore.{KVIndex, KVStore}
  * Provides a view of a KVStore with methods that make it easy to query SQL-specific state. There's
  * no state kept in this class, so it's ok to have multiple instances of it in an application.
  */
-class SQLAppStatusStore(
-    store: KVStore,
-    val listener: Option[SQLAppStatusListener] = None) {
+class SQLAppStatusStore(store: KVStore, val listener: Option[SQLAppStatusListener] = None) {
 
   def executionsList(): Seq[SQLExecutionUIData] = {
     store.view(classOf[SQLExecutionUIData]).asScala.toSeq
   }
 
+  def executionsList(offset: Int, length: Int): Seq[SQLExecutionUIData] = {
+    store.view(classOf[SQLExecutionUIData]).skip(offset).max(length).asScala.toSeq
+  }
+
   def execution(executionId: Long): Option[SQLExecutionUIData] = {
     try {
       Some(store.read(classOf[SQLExecutionUIData], executionId))
@@ -120,7 +122,10 @@ class SparkPlanGraphClusterWrapper(
     val metrics: Seq[SQLPlanMetric]) {
 
   def toSparkPlanGraphCluster(): SparkPlanGraphCluster = {
-    new SparkPlanGraphCluster(id, name, desc,
+    new SparkPlanGraphCluster(
+      id,
+      name,
+      desc,
       new ArrayBuffer() ++ nodes.map(_.toSparkPlanGraphNode()),
       metrics)
   }
@@ -139,7 +144,4 @@ class SparkPlanGraphNodeWrapper(
 
 }
 
-case class SQLPlanMetric(
-    name: String,
-    accumulatorId: Long,
-    metricType: String)
+case class SQLPlanMetric(name: String, accumulatorId: Long, metricType: String)

sql/core/src/main/scala/org/apache/spark/status/api/v1/ApiSqlRootResource.scala

@@ -0,0 +1,27 @@
+/*
+ * 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.status.api.v1
+
+import javax.ws.rs.Path
+
+@Path("/v1")
+private[v1] class ApiSqlRootResource extends ApiRequestContext {
+
+  @Path("applications/{appId}/sql")
+  def sqlList(): Class[SqlResource] = classOf[SqlResource]
+}

sql/core/src/main/scala/org/apache/spark/status/api/v1/SqlResource.scala

@@ -0,0 +1,101 @@
+/*
+ * 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.status.api.v1
+
+import java.util.Date
+import javax.ws.rs.{DefaultValue, GET, Path, PathParam, Produces, QueryParam}
+import javax.ws.rs.core.MediaType
+
+import org.apache.spark.JobExecutionStatus
+import org.apache.spark.sql.execution.ui.{SQLAppStatusStore, SQLExecutionUIData, SQLPlanMetric}
+import org.apache.spark.ui.UIUtils
+
+@Produces(Array(MediaType.APPLICATION_JSON))
+private[v1] class SqlResource extends BaseAppResource {
+
+  @GET
+  def sqlList(
+      @DefaultValue("false") @QueryParam("details") details: Boolean,
+      @DefaultValue("0") @QueryParam("offset") offset: Int,
+      @DefaultValue("20") @QueryParam("length") length: Int): Seq[ExecutionData] = {
+    withUI { ui =>
+      val sqlStore = new SQLAppStatusStore(ui.store.store)
+      sqlStore.executionsList(offset, length).map(prepareExecutionData(_, details))
+    }
+  }
+
+  @GET
+  @Path("{executionId:\\d+}")
+  def sql(
+      @PathParam("executionId") execId: Long,
+      @DefaultValue("false") @QueryParam("details") details: Boolean): ExecutionData = {
+    withUI { ui =>
+      val sqlStore = new SQLAppStatusStore(ui.store.store)
+      sqlStore
+        .execution(execId)
+        .map(prepareExecutionData(_, details))
+        .getOrElse(throw new NotFoundException("unknown id: " + execId))
+    }
+  }
+
+  private def printableMetrics(
+      metrics: Seq[SQLPlanMetric],
+      metricValues: Map[Long, String]): Seq[Metrics] = {
+    metrics.map(metric =>
+      Metrics(metric.name, metricValues.get(metric.accumulatorId).getOrElse("")))
+  }
+
+  private def prepareExecutionData(exec: SQLExecutionUIData, details: Boolean): ExecutionData = {
+    var running = Seq[Int]()
+    var completed = Seq[Int]()
+    var failed = Seq[Int]()
+
+    exec.jobs.foreach {
+      case (id, JobExecutionStatus.RUNNING) =>
+        running = running :+ id
+      case (id, JobExecutionStatus.SUCCEEDED) =>
+        completed = completed :+ id
+      case (id, JobExecutionStatus.FAILED) =>
+        failed = failed :+ id
+      case _ =>
+    }
+
+    val status = if (exec.jobs.size == completed.size) {
+      "COMPLETED"
+    } else if (failed.nonEmpty) {
+      "FAILED"
+    } else {
+      "RUNNING"
+    }
+
+    val duration = exec.completionTime.getOrElse(new Date()).getTime - exec.submissionTime
+    val planDetails = if (details) exec.physicalPlanDescription else ""
+    val metrics = if (details) printableMetrics(exec.metrics, exec.metricValues) else Seq.empty
+    new ExecutionData(
+      exec.executionId,
+      status,
+      exec.description,
+      planDetails,
+      metrics,
+      new Date(exec.submissionTime),
+      duration,
+      running,
+      completed,
+      failed)
+  }
+}

sql/core/src/main/scala/org/apache/spark/status/api/v1/api.scala

@@ -0,0 +1,33 @@
+/*
+ * 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.status.api.v1
+
+import java.util.Date
+
+class ExecutionData private[spark] (
+    val id: Long,
+    val status: String,
+    val description: String,
+    val planDescription: String,
+    val metrics: Seq[Metrics],
+    val submissionTime: Date,
+    val duration: Long,
+    val runningJobIds: Seq[Int],
+    val successJobIds: Seq[Int],
+    val failedJobIds: Seq[Int])
+
+case class Metrics private[spark] (metricName: String, metricValue: String)