Big 6996 enable task progress logging

common/src/java/org/apache/hadoop/hive/conf/HiveConf.java

@@ -3741,13 +3741,20 @@ public static enum ConfVars {
         "hive.tez.exec.inplace.progress",
         true,
         "Updates tez job execution progress in-place in the terminal when hive-cli is used."),
+    HIVE_SESSION_STATE_TIMESTAMP(
+        "hive.session.state.timestamp",
+        false,
+        "If true, adds timestamp prefix to status messages like 'Launching Job' and 'Moving data to directory'."),
     HIVE_SERVER2_INPLACE_PROGRESS(
         "hive.server2.in.place.progress",
         true,
         "Allows hive server 2 to send progress bar update information. This is currently available"
             + " only if the execution engine is tez."),
     TEZ_DAG_STATUS_CHECK_INTERVAL("hive.tez.dag.status.check.interval", "500ms",
       new TimeValidator(TimeUnit.MILLISECONDS), "Interval between subsequent DAG status invocation."),
+    TEZ_PROGRESS_PRINT_INTERVAL("hive.tez.progress.print.interval", "3s",
+      new TimeValidator(TimeUnit.MILLISECONDS), 
+      "Interval between printing progress status for Map/Reducer tasks. Progress is printed at this fixed interval regardless of status changes."),
     SPARK_EXEC_INPLACE_PROGRESS("hive.spark.exec.inplace.progress", true,
         "Updates spark job execution progress in-place in the terminal."),
     TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION("hive.tez.container.max.java.heap.fraction", 0.8f,

ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezTask.java

@@ -270,14 +270,25 @@ public int execute(DriverContext driverContext) {
         }
       }
 
-      if (LOG.isInfoEnabled() && counters != null
+      if (counters != null && console != null
           && (HiveConf.getBoolVar(conf, HiveConf.ConfVars.TEZ_EXEC_SUMMARY) ||
           Utilities.isPerfOrAboveLogging(conf))) {
-        for (CounterGroup group: counters) {
-          LOG.info(group.getDisplayName() +":");
-          for (TezCounter counter: group) {
-            LOG.info("   "+counter.getDisplayName()+": "+counter.getValue());
+        try {
+          for (CounterGroup group : counters) {
+            if (group != null) {
+              String groupName = group.getDisplayName();
+              console.printInfoNoTimestamp(groupName != null ? groupName + ":" : "Unknown Group:");
+              for (TezCounter counter : group) {
+                if (counter != null) {
+                  String counterName = counter.getDisplayName();
+                  console.printInfoNoTimestamp("   " + (counterName != null ? counterName : "Unknown") 
+                      + ": " + counter.getValue());
+                }
+              }
+            }
           }
+        } catch (Exception e) {
+          LOG.warn("Failed to print counters to console. Ignoring.", e);
         }
       }
     } catch (Exception e) {

ql/src/java/org/apache/hadoop/hive/ql/exec/tez/monitoring/RenderStrategy.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
 
 class RenderStrategy {
 
@@ -39,17 +40,20 @@ interface UpdateFunction {
   }
 
   private abstract static class BaseUpdateFunction implements UpdateFunction {
-    private static final int PRINT_INTERVAL = 3000;
-
     final TezJobMonitor monitor;
     private final PerfLogger perfLogger;
+    private final long printInterval;
 
     private long lastPrintTime = 0L;
     private String lastReport = null;
 
     BaseUpdateFunction(TezJobMonitor monitor) {
       this.monitor = monitor;
       perfLogger = SessionState.getPerfLogger();
+      this.printInterval = HiveConf.getTimeVar(
+          SessionState.get().getConf(), 
+          HiveConf.ConfVars.TEZ_PROGRESS_PRINT_INTERVAL, 
+          TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -64,8 +68,7 @@ public void update(DAGStatus status, Map<String, Progress> vertexProgressMap) {
     }
 
     private boolean showReport(String report) {
-      return !report.equals(lastReport)
-          || System.currentTimeMillis() >= lastPrintTime + PRINT_INTERVAL;
+      return System.currentTimeMillis() >= lastPrintTime + printInterval;
     }
 
     /*

ql/src/java/org/apache/hadoop/hive/ql/session/SessionState.java

@@ -1225,6 +1225,37 @@ public void printInfo(String info, String detail) {
      * @param isSilent If true then the message will not be printed to the info stream
      */
     public void printInfo(String info, String detail, boolean isSilent) {
+      if (!isSilent) {
+        SessionState ss = SessionState.get();
+        boolean addTimestamp = ss != null && ss.getConf() != null 
+            && HiveConf.getBoolVar(ss.getConf(), HiveConf.ConfVars.HIVE_SESSION_STATE_TIMESTAMP);
+        if (addTimestamp) {
+          String timestamp = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new java.util.Date());
+          getInfoStream().println(timestamp + "\t" + info);
+        } else {
+          getInfoStream().println(info);
+        }
+      }
+      LOG.info(info + StringUtils.defaultString(detail));
+    }
+
+    /**
+     * Logs info into the log file and console without timestamp prefix.
+     * Use this for output like counters where timestamps are not desired.
+     * @param info The log message
+     */
+    public void printInfoNoTimestamp(String info) {
+      printInfoNoTimestamp(info, null, getIsSilent());
+    }
+
+    /**
+     * Logs info into the log file and console without timestamp prefix.
+     * Use this for output like counters where timestamps are not desired.
+     * @param info The log message
+     * @param detail Extra detail to log which will be not printed if null
+     * @param isSilent If true then the message will not be printed to the info stream
+     */
+    public void printInfoNoTimestamp(String info, String detail, boolean isSilent) {
       if (!isSilent) {
         getInfoStream().println(info);
       }