[SPARK-34399][SQL] Add commit duration to SQL tab's graph node.

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

@@ -192,6 +192,10 @@ class BasicWriteJobStatsTracker(
     new BasicWriteTaskStatsTracker(serializableHadoopConf.value)
   }
 
+  override def processCommitDuration(duration: Long): Unit = {
+    metrics(BasicWriteJobStatsTracker.DURATION_FILE_COMMIT).set(duration)
+  }
+
   override def processStats(stats: Seq[WriteTaskStats]): Unit = {
     val sparkContext = SparkContext.getActive.get
     var partitionsSet: mutable.Set[InternalRow] = mutable.HashSet.empty
@@ -223,6 +227,7 @@ object BasicWriteJobStatsTracker {
   private val NUM_OUTPUT_BYTES_KEY = "numOutputBytes"
   private val NUM_OUTPUT_ROWS_KEY = "numOutputRows"
   private val NUM_PARTS_KEY = "numParts"
+  private val DURATION_FILE_COMMIT = "durationCommit"
   /** XAttr key of the data length header added in HADOOP-17414. */
   val FILE_LENGTH_XATTR = "header.x-hadoop-s3a-magic-data-length"
 
@@ -232,7 +237,8 @@ object BasicWriteJobStatsTracker {
       NUM_FILES_KEY -> SQLMetrics.createMetric(sparkContext, "number of written files"),
       NUM_OUTPUT_BYTES_KEY -> SQLMetrics.createSizeMetric(sparkContext, "written output"),
       NUM_OUTPUT_ROWS_KEY -> SQLMetrics.createMetric(sparkContext, "number of output rows"),
-      NUM_PARTS_KEY -> SQLMetrics.createMetric(sparkContext, "number of dynamic part")
+      NUM_PARTS_KEY -> SQLMetrics.createMetric(sparkContext, "number of dynamic part"),
+      DURATION_FILE_COMMIT-> SQLMetrics.createTimingMetric(sparkContext, "duration of commit files")
     )
   }
 }

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

@@ -221,7 +221,7 @@ object FileFormatWriter extends Logging {
       val (_, duration) = Utils.timeTakenMs { committer.commitJob(job, commitMsgs) }
       logInfo(s"Write Job ${description.uuid} committed. Elapsed time: $duration ms.")
 
-      processStats(description.statsTrackers, ret.map(_.summary.stats))
+      processStats(description.statsTrackers, ret.map(_.summary.stats), duration)
       logInfo(s"Finished processing stats for write job ${description.uuid}.")
 
       // return a set of all the partition paths that were updated during this job
@@ -304,7 +304,8 @@ object FileFormatWriter extends Logging {
    */
   private[datasources] def processStats(
       statsTrackers: Seq[WriteJobStatsTracker],
-      statsPerTask: Seq[Seq[WriteTaskStats]])
+      statsPerTask: Seq[Seq[WriteTaskStats]],
+      duration: Long)
   : Unit = {
 
     val numStatsTrackers = statsTrackers.length
@@ -320,6 +321,7 @@ object FileFormatWriter extends Logging {
       statsTrackers.map(_ => Seq.empty)
     }
 
+    statsTrackers.foreach(_.processCommitDuration(duration))
     statsTrackers.zip(statsPerTracker).foreach {
       case (statsTracker, stats) => statsTracker.processStats(stats)
     }

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

@@ -103,6 +103,11 @@ trait WriteJobStatsTracker extends Serializable {
    */
   def newTaskInstance(): WriteTaskStatsTracker
 
+  /**
+   * Process the duration of commit all files.
+   */
+  def processCommitDuration(duration: Long): Unit
+
   /**
    * Process the given collection of stats computed during this job.
    * E.g. aggregate them, write them to memory / disk, issue warnings, whatever.

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

@@ -36,7 +36,7 @@ class FileBatchWrite(
     val (_, duration) = Utils.timeTakenMs { committer.commitJob(job, results.map(_.commitMsg)) }
     logInfo(s"Write Job ${description.uuid} committed. Elapsed time: $duration ms.")
 
-    processStats(description.statsTrackers, results.map(_.summary.stats))
+    processStats(description.statsTrackers, results.map(_.summary.stats), duration)
     logInfo(s"Finished processing stats for write job ${description.uuid}.")
   }