address comments

sql/core/src/main/scala/org/apache/spark/sql/execution/Sort.scala

@@ -89,11 +89,11 @@ case class Sort(
       // Remember spill data size of this task before execute this operator so that we can
       // figure out how many bytes we spilled for this operator.
       val spillSizeBefore = metrics.memoryBytesSpilled
-      val beforeSort = System.currentTimeMillis()
+      val beforeSort = System.nanoTime()
 
       val sortedIterator = sorter.sort(iter.asInstanceOf[Iterator[UnsafeRow]])
 
-      sortingTime += System.currentTimeMillis() - beforeSort
+      sortingTime += (System.nanoTime() - beforeSort) >> 20
       peakMemory += sorter.getPeakMemoryUsage
       spillSize += metrics.memoryBytesSpilled - spillSizeBefore
       metrics.incPeakExecutionMemory(sorter.getPeakMemoryUsage)
@@ -147,13 +147,13 @@ case class Sort(
     val beforeSortMs = ctx.freshName("beforeSortMs")
     s"""
        | if ($needToSort) {
-       |   long $beforeSortMs = System.currentTimeMillis();
+       |   long $beforeSortMs = System.nanoTime();
        |   long $spillSizeBefore = $metrics.memoryBytesSpilled();
        |
        |   $addToSorter();
        |   $sortedIterator = $sorterVariable.sort();
        |
-       |   $sortingTime.add(System.currentTimeMillis() - $beforeSortMs);
+       |   $sortingTime.add((System.nanoTime() - $beforeSortMs) >> 20);
        |   $peakMemory.add($sorterVariable.getPeakMemoryUsage());
        |   $spillSize.add($metrics.memoryBytesSpilled() - $spillSizeBefore);
        |   $metrics.incPeakExecutionMemory($sorterVariable.getPeakMemoryUsage());

sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/TungstenAggregate.scala

@@ -218,9 +218,9 @@ case class TungstenAggregate(
     s"""
        | while (!$initAgg) {
        |   $initAgg = true;
-       |   long $beforeAgg = System.currentTimeMillis();
+       |   long $beforeAgg = System.nanoTime();
        |   $doAgg();
-       |   $aggTime.add(System.currentTimeMillis() - $beforeAgg);
+       |   $aggTime.add((System.nanoTime() - $beforeAgg) >> 20);
        |
        |   // output the result
        |   ${genResult.trim}
@@ -544,9 +544,9 @@ case class TungstenAggregate(
     s"""
      if (!$initAgg) {
        $initAgg = true;
-       long $beforeAgg = System.currentTimeMillis();
+       long $beforeAgg = System.nanoTime();
        $doAgg();
-       $aggTime.add(System.currentTimeMillis() - $beforeAgg);
+       $aggTime.add((System.nanoTime() - $beforeAgg) >> 20);
      }
 
      // output the result

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchange.scala

@@ -69,20 +69,20 @@ case class BroadcastExchange(
       // This will run in another thread. Set the execution id so that we can connect these jobs
       // with the correct execution.
       SQLExecution.withExecutionId(sparkContext, executionId) {
-        val beforeCollect = System.currentTimeMillis()
+        val beforeCollect = System.nanoTime()
         // Note that we use .executeCollect() because we don't want to convert data to Scala types
         val input: Array[InternalRow] = child.executeCollect()
-        val beforeBuild = System.currentTimeMillis()
-        longMetric("collectTime") += beforeBuild - beforeCollect
+        val beforeBuild = System.nanoTime()
+        longMetric("collectTime") += (beforeBuild - beforeCollect) >> 20
         longMetric("dataSize") += input.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
 
         // Construct and broadcast the relation.
         val relation = mode.transform(input)
-        val beforeBroadcast = System.currentTimeMillis()
-        longMetric("buildTime") += beforeBroadcast - beforeBuild
+        val beforeBroadcast = System.nanoTime()
+        longMetric("buildTime") += (beforeBroadcast - beforeBuild) >> 20
 
         val broadcasted = sparkContext.broadcast(relation)
-        longMetric("broadcastTime") += System.currentTimeMillis() - beforeBroadcast
+        longMetric("broadcastTime") += (System.nanoTime() - beforeBroadcast) >> 20
         broadcasted
       }
     }(BroadcastExchange.executionContext)