TEZ-4325: Counter for sort time

tez-api/src/main/java/org/apache/tez/common/counters/TaskCounter.java

@@ -215,6 +215,11 @@ public enum TaskCounter {
    */
   MERGE_PHASE_TIME,
 
+  /**
+   * Time taken to sort data in milliseconds.
+   */
+  SORT_TIME,
+
   /**
    * First event received from source relative to task start time.
    *

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/ExternalSorter.java

@@ -162,6 +162,9 @@ public void progress() {
   protected final TezCounter numAdditionalSpills;
   // Number of files offered via shuffle-handler to consumers.
   protected final TezCounter numShuffleChunks;
+  // Time (in ms) spent with sorting (includes only the actual sorting time, doesn't include Span related operations)
+  protected final TezCounter sortTimeMsCounter;
+
   // How partition stats should be reported.
   final ReportPartitionStats reportPartitionStats;
 
@@ -220,6 +223,7 @@ public ExternalSorter(OutputContext outputContext, Configuration conf, int numOu
     additionalSpillBytesRead = outputContext.getCounters().findCounter(TaskCounter.ADDITIONAL_SPILLS_BYTES_READ);
     numAdditionalSpills = outputContext.getCounters().findCounter(TaskCounter.ADDITIONAL_SPILL_COUNT);
     numShuffleChunks = outputContext.getCounters().findCounter(TaskCounter.SHUFFLE_CHUNK_COUNT);
+    sortTimeMsCounter = outputContext.getCounters().findCounter(TaskCounter.SORT_TIME);
 
     // compression
     this.codec = CodecUtils.getCodec(conf);

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/PipelinedSorter.java

@@ -966,8 +966,10 @@ public SpanIterator sort(IndexedSorter sorter) {
       if(length() > 1) {
         sorter.sort(this, 0, length(), progressable);
       }
+      long sortTime = System.currentTimeMillis() - start;
       LOG.info(outputContext.getDestinationVertexName() + ": " + "done sorting span=" + index + ", length=" + length() + ", "
-          + "time=" + (System.currentTimeMillis() - start));
+          + "time=" + sortTime);
+      sortTimeMsCounter.increment(sortTime);
       return new SpanIterator((SortSpan)this);
     }
 

tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/dflt/DefaultSorter.java

@@ -859,7 +859,11 @@ protected void sortAndSpill(long sameKeyCount, long totalKeysCount)
       throws IOException, InterruptedException {
     final int mstart = getMetaStart();
     final int mend = getMetaEnd();
+
+    long start = System.currentTimeMillis();
     sorter.sort(this, mstart, mend, progressable);
+    sortTimeMsCounter.increment(System.currentTimeMillis() - start);
+
     spill(mstart, mend, sameKeyCount, totalKeysCount);
   }