[CARMEL-6085] Show metrics after running analyze event command (#1012)

Author: fenzhu

sql/core/src/main/scala/org/apache/spark/sql/execution/command/AnalyzeEventCommand.scala

@@ -57,6 +57,11 @@ case class AnalyzeEventCommand(file: String, prefix: String,
 
   override lazy val metrics: Map[String, SQLMetric] = getMetrics
 
+  override val output: Seq[Attribute] = {
+    AttributeReference("analyze_event_item", StringType, nullable = false)() ::
+      AttributeReference("analyze_event_value", StringType, nullable = false)() :: Nil
+  }
+
   private def getMetrics: Map[String, SQLMetric] = {
     Map[String, SQLMetric](
       BasicWriteJobStatsTracker.NUM_FILES_KEY ->
@@ -75,6 +80,8 @@ case class AnalyzeEventCommand(file: String, prefix: String,
       throw new SparkException(s"Analyze event command should only be executed in " +
         s"reserved-test queue, current queue is $currentQueue.")
     }
+    val result = new ArrayBuffer[Row]
+    result += Row("Input Parameter", s"file=$file, prefix=$prefix, from=$from, to=$to")
 
     val eventLogDir = sparkSession.sparkContext.getConf.get(EVENT_LOG_DIR)
     val hadoopConf = sparkSession.sessionState.newHadoopConf()
@@ -85,7 +92,9 @@ case class AnalyzeEventCommand(file: String, prefix: String,
     val candidateLogFiles: Seq[FileStatus] = AnalyzeEventCommand.getCandidateEventLogFiles(
       sparkSession, fs, qualifiedEventLogParentPath, file, prefix, from, to)
     logInfo(s"Analyze ${candidateLogFiles.length} Candidate EventLog Files.")
-    if (candidateLogFiles.isEmpty) return ArrayBuffer.empty
+    result += Row("Analyzed Files", s"${candidateLogFiles.length}")
+    if (candidateLogFiles.isEmpty) return result
+
     val skipTaskParsing = sparkSession.sessionState.conf.getConfString(
       "spark.sql.workload.task.parse.skip", "true").toBoolean
     val maxPartition = sparkSession.sessionState.conf.getConfString(
@@ -130,7 +139,7 @@ case class AnalyzeEventCommand(file: String, prefix: String,
       rdd.unpersist()
       sparkContext.setJobDescription(previousDescription)
     }
-    Seq.empty[Row]
+    result
   }
 
   private def analyzeHugeLog(pathStr: String, task_event_count: Long, partitions: Int,