[CARMEL-6400] Log detail sql execution metrics (#1162)

* [CARMEL-6400] Log detail sql execution metrics

* Fix code style

* Fix code style

Author: allenma

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/QueryPlanningTracker.scala

@@ -74,6 +74,20 @@ object QueryPlanningTracker {
     }
   }
 
+  class ExternalSystemCallSummary {
+    var callCount: Int = 0
+    var totalDuration: Long = 0
+
+    def updateCallInfo(duration: Long): Unit = {
+      callCount += 1
+      totalDuration += duration
+    }
+
+    override def toString: String = {
+      s"callCount=$callCount, totalDuration=${totalDuration}ms"
+    }
+  }
+
   /**
    * A thread local variable to implicitly pass the tracker around. This assumes the query planner
    * is single-threaded, and avoids passing the same tracker context in every function call.
@@ -104,6 +118,9 @@ class QueryPlanningTracker {
   // From a phase to its start time and end time, in ms.
   private val phasesMap = new java.util.HashMap[String, PhaseSummary]
 
+  // external system call tracking during planning
+  private val externalSystemCalls = new java.util.HashMap[String, ExternalSystemCallSummary]
+
   val queryPlanningContext = new QueryPlanningContext()
 
   var totalTimeNs = 0L
@@ -171,6 +188,20 @@ class QueryPlanningTracker {
 
   def getViewMap: java.util.HashMap[String, Int] = viewMap
 
+  /**
+   * track external system call information
+   * @param callName
+   * @param duration in ms
+   */
+  def recordExternalSystemCall(callName: String, duration: Long): Unit = {
+    var s = externalSystemCalls.get(callName)
+    if (s eq null) {
+      s = new ExternalSystemCallSummary
+      externalSystemCalls.put(callName, s)
+    }
+    s.updateCallInfo(duration)
+  }
+
   def inheritViewUsage(oldTracker: QueryPlanningTracker): Unit = {
     viewMap.putAll(oldTracker.getViewMap)
   }
@@ -244,4 +275,23 @@ class QueryPlanningTracker {
      """.stripMargin
 
   }
+
+  def formattedPlanPhrases(): String = {
+    val phaseInfo = phases.toSeq.map {
+      case(phase, summary) =>
+        s"$phase: ${summary.toString}, duration(${summary.durationMs}ms)"
+    }.mkString(System.lineSeparator())
+
+    val externalSysCalls = externalSystemCalls.asScala.toSeq.map {
+      case(callName, callSummary) =>
+        s"$callName: ${callSummary.toString}"
+    }.mkString(System.lineSeparator())
+
+    s"""
+       |=== phases ===
+       |$phaseInfo
+       |=== external system calls ===
+       |$externalSysCalls
+       |""".stripMargin
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala

@@ -35,6 +35,7 @@ import org.apache.spark.internal.Logging
 import org.apache.spark.internal.io.SparkHadoopWriterUtils
 import org.apache.spark.metrics.source.HiveCatalogMetrics
 import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.QueryPlanningTracker
 import org.apache.spark.sql.execution.streaming.FileStreamSink
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types.StructType
@@ -158,8 +159,11 @@ class InMemoryFileIndex(
       fileStatusCache.putLeafFiles(path, leafFiles.toArray)
       output ++= leafFiles
     }
-    logInfo(s"It took ${(System.nanoTime() - startTime) / (1000 * 1000)} ms to list leaf files" +
+    val duration = (System.nanoTime() - startTime) / (1000 * 1000)
+    logInfo(s"It took ${duration} ms to list leaf files" +
       s" for ${paths.length} paths.")
+    QueryPlanningTracker.getCurrent.foreach(t =>
+      t.recordExternalSystemCall("listFiles", duration))
     output
   }
 }

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkDownloadDataOperation.scala

@@ -292,6 +292,7 @@ private[hive] class SparkDownloadDataOperation(
 
       setState(OperationState.FINISHED)
       logQueryInfo(s"Finished query [$statementId].")
+      logInfo(s"Plan details:${result.queryExecution.tracker.formattedPlanPhrases()}")
     } catch {
       case NonFatal(e) =>
         logQueryError(s"Error executing query [$statementId]", e)

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala

@@ -345,6 +345,8 @@ private[hive] class SparkExecuteStatementOperation(
         iter = executeForResults()
         dataTypes = result.schema.fields.map(_.dataType)
         logQueryInfo(s"Finished query [$statementId].")
+        logInfo(s"Query [$statementId] plan details:" +
+          s"${result.queryExecution.tracker.formattedPlanPhrases()}")
       }
     } catch {
       // Actually do need to catch Throwable as some failures don't inherit from Exception and

sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2Listener.scala

@@ -316,6 +316,24 @@ private[thriftserver] class HiveThriftServer2Listener(
     }
   }
 
+  def logExecutionMetrics(metrics: Option[QueryExecutionMetrics]): Unit = {
+    metrics.foreach(m => {
+      val stageStatsHeader = "stage start finish taskNum slowestTask(taskId execId " +
+        "taskDuration execRunTime launchDelay inputSize shuffleReadSize)"
+      val stats =
+        s"""
+           |Query [${m.groupId}] execution metrics:
+           |Task Num: ${m.taskNum}
+           |Task Time: ${m.totalTaskTime}
+           |Stage stats:
+           |$stageStatsHeader
+           |${m.stageStatsInfo()}
+       """.stripMargin
+      logInfo(stats)
+    }
+    )
+  }
+
   private def onQueryExit(e: SparkListenerThriftServerQueryExit): Unit = synchronized {
     val execInfo = executionList.get(e.id)
     if (execInfo == null) {
@@ -326,6 +344,7 @@ private[thriftserver] class HiveThriftServer2Listener(
       val session = sessionList.get(execInfo.sessionId)
       val metrics = Option(executionMetrics.get(execInfo.groupId))
       l.logQuery(e.id, execInfo, session, e.qlObjects, e.extInfo, metrics)
+      logExecutionMetrics(metrics)
     }
     executionMetrics.remove(execInfo.groupId)
     if (delayCleanExecution && executionList.get(e.id).state == ExecutionState.CLOSED) {
@@ -659,6 +678,7 @@ class QueryExecutionMetrics(val groupId: String, val sessionId: String,
 
     stageStats.foreach { stats =>
       stats.finishTaskNum += 1
+      stats.updateSlowestTaskMetrics(taskEnd)
       if (taskEnd.taskInfo.attemptNumber == 0 && !taskEnd.taskInfo.speculative) {
         val launchDelay = taskEnd.taskInfo.launchTime - stats.startTime
         if (launchDelay > maxLaunchDelay) maxLaunchDelay = launchDelay
@@ -695,6 +715,13 @@ class QueryExecutionMetrics(val groupId: String, val sessionId: String,
     lastExecutionEndTime = exec.time
   }
 
+  def stageStatsInfo(): String = {
+    stageStatsMap.toSeq.map { case ((stageId, attemptId), stageStats) =>
+      s"$stageId-$attemptId ${stageStats.startTime} ${stageStats.endTime} " +
+        s"${stageStats.totalTaskNum} (${stageStats.slowestTaskStatsInfo()})"
+    }.mkString(System.lineSeparator())
+  }
+
   override def toString(): String = {
     String.join(
       DELIMITER,
@@ -728,6 +755,8 @@ class StageStats(val name: String) {
   var maxShuffleWaitTime: Long = 0
   var maxInputWaitTime: Long = 0
 
+  var slowestTask: Option[SparkListenerTaskEnd] = None
+
   def updateTaskWaitResourceTime(waitResourceTime: Long): Unit = {
     if (waitResourceTime > maxWaitResourceTime) {
       maxWaitResourceTime = waitResourceTime
@@ -745,8 +774,50 @@ class StageStats(val name: String) {
       maxInputWaitTime = inputWaitTime
     }
   }
+
+  def updateSlowestTaskMetrics(taskEnd: SparkListenerTaskEnd): Unit = {
+    if (slowestTask.isEmpty || taskEnd.taskInfo.finishTime > slowestTask.get.taskInfo.finishTime) {
+      slowestTask = Some(taskEnd)
+    }
+  }
+
+  def slowestTaskStatsInfo(): String = {
+    def extractTaskStats(end: SparkListenerTaskEnd): TaskStats = {
+      TaskStats(
+        end.taskInfo.finishTime,
+        end.taskInfo.taskId,
+        end.taskInfo.executorId,
+        end.taskInfo.finishTime - end.taskInfo.launchTime,
+        end.taskMetrics.executorRunTime,
+        this.startTime - end.taskInfo.launchTime,
+        end.taskMetrics.inputMetrics.bytesRead,
+        end.taskMetrics.shuffleReadMetrics.totalBytesRead
+      )
+    }
+    slowestTask.map(taskEnd => {
+      val taskStats = extractTaskStats(taskEnd)
+      s"${taskStats.taskId} " +
+        s"${taskStats.execId} " +
+        s"${taskStats.duration} " +
+        s"${taskStats.executorRunTime} " +
+        s"${taskStats.launchDelay} " +
+        s"${taskStats.inputSize} " +
+        s"${taskStats.shuffleReadSize}"
+    }).getOrElse("")
+  }
 }
 
+case class TaskStats(
+  finishTime: Long,
+  taskId: Long,
+  execId: String,
+  duration: Long,
+  executorRunTime: Long,
+  launchDelay: Long,
+  inputSize: Long,
+  shuffleReadSize: Long
+)
+
 case class SqlWithStats(
   queryId: String,
   statement: String,

sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ui/HiveThriftServer2ListenerSuite.scala

@@ -221,6 +221,57 @@ class HiveThriftServer2ListenerSuite extends SparkFunSuite with BeforeAndAfter {
     assert(sqlWithStats.size == 1)
   }
 
+  test("log execution metrics") {
+    val (_, listener: HiveThriftServer2Listener) = createAppStatusStore(true)
+
+    val executionMetrics = new QueryExecutionMetrics("groupId", "sessionId")
+    executionMetrics.stageStart(createStageInfo(0, 0, 2))
+    executionMetrics.taskStart(SparkListenerTaskStart(0, 0, createTaskInfo(0, 0)))
+    executionMetrics.taskStart(SparkListenerTaskStart(0, 0, createTaskInfo(1, 0)))
+
+    val tm1 = new TaskMetrics
+    tm1.inputMetrics.setBytesRead(10000)
+    tm1.inputMetrics.setRecordsRead(1000)
+
+    val tm2 = new TaskMetrics
+    tm2.inputMetrics.setBytesRead(10000)
+    tm2.inputMetrics.setRecordsRead(1000)
+
+    executionMetrics.taskEnd(SparkListenerTaskEnd(
+      stageId = 0,
+      stageAttemptId = 0,
+      taskType = "",
+      reason = null,
+      createTaskInfo(0, 0),
+      new ExecutorMetrics,
+      tm1))
+    executionMetrics.taskEnd(SparkListenerTaskEnd(
+      stageId = 0,
+      stageAttemptId = 0,
+      taskType = "",
+      reason = null,
+      createTaskInfo(1, 0),
+      new ExecutorMetrics,
+      tm2))
+    executionMetrics.stageComplete(createStageInfo(0, 0, 2))
+
+    executionMetrics.stageStart(createStageInfo(1, 0, 1))
+    executionMetrics.taskStart(SparkListenerTaskStart(1, 0, createTaskInfo(0, 0)))
+    val tm3 = new TaskMetrics
+    tm3.inputMetrics.setBytesRead(10000)
+    tm3.inputMetrics.setRecordsRead(1000)
+    executionMetrics.taskEnd(SparkListenerTaskEnd(
+      stageId = 1,
+      stageAttemptId = 0,
+      taskType = "",
+      reason = null,
+      createTaskInfo(1, 0),
+      new ExecutorMetrics,
+      tm3))
+    executionMetrics.stageComplete(createStageInfo(1, 0, 1))
+    listener.logExecutionMetrics(Some(executionMetrics))
+  }
+
   private def createStageInfo(stageId: Int, attemptId: Int, numTasks: Int): StageInfo = {
     new StageInfo(stageId = stageId,
       attemptId = attemptId,

sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala

@@ -49,7 +49,7 @@ import org.apache.spark.{SparkConf, SparkException}
 import org.apache.spark.internal.Logging
 import org.apache.spark.metrics.source.HiveCatalogMetrics
 import org.apache.spark.sql.AnalysisException
-import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.{QueryPlanningTracker, TableIdentifier}
 import org.apache.spark.sql.catalyst.analysis.{NoSuchDatabaseException, NoSuchPartitionException}
 import org.apache.spark.sql.catalyst.catalog._
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
@@ -331,6 +331,8 @@ private[hive] class HiveClientImpl(
     val end = System.currentTimeMillis()
     HiveCatalogMetrics.updateCallDuration(call, end - start)
     HiveCatalogMetrics.incrementCallCount(call, 1)
+    QueryPlanningTracker.getCurrent.foreach(t =>
+      t.recordExternalSystemCall("hiveMetastoreRpc", end - start))
     t
   }