check Sort metrics values

Author: seancxmao

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsSuite.scala

@@ -197,10 +197,23 @@ class SQLMetricsSuite extends SparkFunSuite with SQLMetricsTestUtils with Shared
     // Assume the execution plan with node id is
     // Sort(nodeId = 0)
     //   Exchange(nodeId = 1)
-    //     LocalTableScan(nodeId = 2)
+    //     Project(nodeId = 2)
+    //       LocalTableScan(nodeId = 3)
+    // Because of SPARK-25267, ConvertToLocalRelation is disabled in the test cases of sql/core,
+    // so Project here is not collapsed into LocalTableScan.
     val df = Seq(1, 3, 2).toDF("id").sort('id)
-    testSparkPlanMetrics(df, 2, Map.empty)
-    assert(df.queryExecution.executedPlan.find(_.isInstanceOf[SortExec]).isDefined)
+    val metrics = getSparkPlanMetrics(df, 2, Set(0))
+    val sortMetrics = metrics.get.get(0).get
+    // Check node 0 is Sort node
+    val operatorName = sortMetrics._1
+    assert(operatorName == "Sort")
+    // Check metrics values
+    val sortTimeStr = sortMetrics._2.get("sort time total (min, med, max)").get.toString
+    timingMetricStats(sortTimeStr).foreach { case (sortTime, _) => assert(sortTime >= 0) }
+    val peakMemoryStr = sortMetrics._2.get("peak memory total (min, med, max)").get.toString
+    sizeMetricStats(peakMemoryStr).foreach { case (peakMemory, _) => assert(peakMemory > 0) }
+    val spillSizeStr = sortMetrics._2.get("spill size total (min, med, max)").get.toString
+    sizeMetricStats(spillSizeStr).foreach { case (spillSize, _) => assert(spillSize >= 0) }
   }
 
   test("SortMergeJoin metrics") {

sql/core/src/test/scala/org/apache/spark/sql/execution/metric/SQLMetricsTestUtils.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.execution.metric
 
 import java.io.File
+import java.util.regex.Pattern
 
 import scala.collection.mutable.HashMap
 
@@ -198,6 +199,51 @@ trait SQLMetricsTestUtils extends SQLTestUtils {
       }
     }
   }
+
+  private def metricStats(metricStr: String): Seq[String] = {
+    val sum = metricStr.substring(0, metricStr.indexOf("(")).stripPrefix("\n").stripSuffix(" ")
+    val minMedMax = metricStr.substring(metricStr.indexOf("(") + 1, metricStr.indexOf(")"))
+      .split(", ").toSeq
+    (sum +: minMedMax)
+  }
+
+  private def stringToBytes(str: String): (Float, String) = {
+    val matcher = Pattern.compile("([0-9]+(\\.[0-9]+)?) (EB|PB|TB|GB|MB|KB|B)").matcher(str)
+    if (matcher.matches()) {
+      (matcher.group(1).toFloat, matcher.group(3))
+    } else {
+      throw new NumberFormatException("Failed to parse byte string: " + str)
+    }
+  }
+
+  private def stringToDuration(str: String): (Float, String) = {
+    val matcher = Pattern.compile("([0-9]+(\\.[0-9]+)?) (ms|s|m|h)").matcher(str)
+    if (matcher.matches()) {
+      (matcher.group(1).toFloat, matcher.group(3))
+    } else {
+      throw new NumberFormatException("Failed to parse time string: " + str)
+    }
+  }
+
+  /**
+   * Convert a size metric string to a sequence of stats, including sum, min, med and max in order,
+   * each a tuple of (value, unit).
+   * @param metricStr size metric string, e.g. "\n96.2 MB (32.1 MB, 32.1 MB, 32.1 MB)"
+   * @return A sequence of stats, e.g. ((96.2,MB), (32.1,MB), (32.1,MB), (32.1,MB))
+   */
+  protected def sizeMetricStats(metricStr: String): Seq[(Float, String)] = {
+    metricStats(metricStr).map(stringToBytes)
+  }
+
+  /**
+   * Convert a timing metric string to a sequence of stats, including sum, min, med and max in
+   * order, each a tuple of (value, unit).
+   * @param metricStr timing metric string, e.g. "\n2.0 ms (1.0 ms, 1.0 ms, 1.0 ms)"
+   * @return A sequence of stats, e.g. ((2.0,ms), (1.0,ms), (1.0,ms), (1.0,ms))
+   */
+  protected def timingMetricStats(metricStr: String): Seq[(Float, String)] = {
+    metricStats(metricStr).map(stringToDuration)
+  }
 }