HIVE-27126: queue level resource stats for YARN RM.

common/src/java/org/apache/hadoop/hive/common/log/ProgressMonitor.java

@@ -52,6 +52,11 @@ public String executionStatus() {
     public double progressedPercentage() {
       return 0;
     }
+
+    @Override
+    public List<String> getResourceLines() {
+      return Collections.emptyList();
+    }
   };
 
   List<String> headers();
@@ -65,4 +70,13 @@ public double progressedPercentage() {
   String executionStatus();
 
   double progressedPercentage();
+
+  /**
+   * Returns zero or more supplementary resource-info lines to render after the progress bar.
+   * Implementations that have no resource info should return an empty list (the default).
+   * Lines are rendered as-is by the render strategy — no additional formatting is applied.
+   */
+  default List<String> getResourceLines() {
+    return Collections.emptyList();
+  }
 }

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

@@ -4859,6 +4859,10 @@ public static enum ConfVars {
             + " 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_YARN_METRICS_REFRESH_INTERVAL("hive.tez.yarn.metrics.refresh.interval", "5000ms",
+      new TimeValidator(TimeUnit.MILLISECONDS),
+      "Interval between YARN queue/cluster resource metric refreshes shown in the Tez progress monitor. "
+          + "Set to 0 to disable YARN resource display."),
     TEZ_CONTAINER_MAX_JAVA_HEAP_FRACTION("hive.tez.container.max.java.heap.fraction", 0.8f,
         "This is to override the tez setting with the same name"),
     TEZ_TASK_SCALE_MEMORY_RESERVE_FRACTION_MIN("hive.tez.task.scale.memory.reserve-fraction.min",

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

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.metastore.api.WMFullResourcePlan;
 import org.apache.hadoop.hive.metastore.api.WMTrigger;
 import org.apache.hadoop.hive.ql.exec.tez.TezSessionPoolSession.Manager;
+import org.apache.hadoop.hive.ql.exec.tez.monitoring.YarnMetricsHelper;
 import org.apache.hadoop.hive.ql.metadata.HiveException;
 import org.apache.hadoop.hive.ql.session.SessionState;
 import org.apache.hadoop.hive.ql.wm.ExecutionTrigger;
@@ -397,6 +398,8 @@ public void stop() throws Exception {
 
     metrics.stop();
     instance = null;
+    // Release the shared YarnClient used for queue resource metrics display.
+    YarnMetricsHelper.shutdown();
   }
 
   /**

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

@@ -31,6 +31,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.StringWriter;
+import java.util.List;
 import java.util.Map;
 import java.util.SortedSet;
 import java.util.TreeSet;
@@ -63,7 +64,11 @@ private abstract static class BaseUpdateFunction implements UpdateFunction {
 
     @Override
     public void update(DAGStatus status, Map<String, Progress> vertexProgressMap) {
-      renderProgress(monitor.progressMonitor(status, vertexProgressMap));
+      ProgressMonitor progressMonitor = monitor.progressMonitor(status, vertexProgressMap);
+      renderProgress(progressMonitor);
+      // Render supplementary resource lines (e.g. YARN queue/cluster metrics).
+      // getResourceLines() returns an empty list when metrics are unavailable — no cast needed.
+      renderResourceLines(progressMonitor.getResourceLines());
       String report = getReport(vertexProgressMap);
       if (showReport(report)) {
         renderReport(report);
@@ -137,14 +142,23 @@ private String getReport(Map<String, Progress> progressMap) {
       return reportBuffer.toString();
     }
 
+    /** Renders the primary progress table (vertex rows + progress bar). */
     abstract void renderProgress(ProgressMonitor progressMonitor);
 
+    /**
+     * Renders supplementary resource lines returned by
+     * {@link ProgressMonitor#getResourceLines()}.
+     * Called after {@link #renderProgress} with whatever the monitor returned —
+     * may be an empty list, in which case nothing should be printed.
+     */
+    abstract void renderResourceLines(List<String> lines);
+
     abstract void renderReport(String report);
   }
 
   /**
-   * this adds the required progress update to the session state that is used by HS2 to send the
-   * same information to beeline client when requested.
+   * Used by HiveServer2: stores the monitor in SessionState so beeline can poll it,
+   * and logs resource lines to the server log.
    */
   static class LogToFileFunction extends BaseUpdateFunction {
     private static final Logger LOGGER = LoggerFactory.getLogger(LogToFileFunction.class);
@@ -164,6 +178,13 @@ public void renderProgress(ProgressMonitor progressMonitor) {
       SessionState.get().updateProgressMonitor(progressMonitor);
     }
 
+    @Override
+    public void renderResourceLines(List<String> lines) {
+      if (hiveServer2InPlaceProgressEnabled) {
+        lines.forEach(LOGGER::info);
+      }
+    }
+
     @Override
     public void renderReport(String report) {
       if (hiveServer2InPlaceProgressEnabled) {
@@ -176,8 +197,7 @@ public void renderReport(String report) {
   }
 
   /**
-   * This used when we want the progress update to printed in the same process typically used via
-   * hive-cli mode.
+   * Used by hive-cli: renders the progress table directly to the terminal in-place.
    */
   static class InPlaceUpdateFunction extends BaseUpdateFunction {
     /**
@@ -196,6 +216,12 @@ public void renderProgress(ProgressMonitor progressMonitor) {
       inPlaceUpdate.render(progressMonitor);
     }
 
+    @Override
+    public void renderResourceLines(List<String> lines) {
+      lines.forEach(line ->
+          InPlaceUpdate.reprintLine(SessionState.LogHelper.getInfoStream(), line));
+    }
+
     @Override
     public void renderReport(String report) {
       monitor.console.logInfo(report);

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

@@ -119,6 +119,10 @@ public static void initShutdownHook() {
   private final RenderStrategy.UpdateFunction updateFunction;
   // compile time tez counters
   private final TezCounters counters;
+  /** Live YARN metrics helper; may be null when the feature is disabled or YARN is unavailable. */
+  private final YarnMetricsHelper yarnMetricsHelper;
+  /** YARN queue name for this query. */
+  private final String yarnQueueName;
 
   public TezJobMonitor(TezSession session, List<BaseWork> topSortedWorks, final DAGClient dagClient, HiveConf conf,
       DAG dag, Context ctx, final TezCounters counters, PerfLogger perfLogger) {
@@ -134,6 +138,9 @@ public TezJobMonitor(TezSession session, List<BaseWork> topSortedWorks, final DA
     this.counters = counters;
     this.shouldCollectSummaryString = conf.getBoolVar(HiveConf.ConfVars.HIVE_QUERY_HISTORY_ENABLED) &&
         conf.getBoolVar(ConfVars.HIVE_QUERY_HISTORY_EXEC_SUMMARY_ENABLED);
+    String queueName = session.getQueueName();
+    this.yarnQueueName = (queueName != null && !queueName.isEmpty()) ? queueName : "default";
+    this.yarnMetricsHelper = YarnMetricsHelper.getInstance(conf);
   }
 
   private RenderStrategy.UpdateFunction updateFunction() {
@@ -479,6 +486,7 @@ private static boolean hasInterruptedException(Throwable e) {
     return false;
   }
 
+
   /**
    * killRunningJobs tries to terminate execution of all
    * currently running tez queries. No guarantees, best effort only.
@@ -516,7 +524,7 @@ public String getDiagnostics() {
   ProgressMonitor progressMonitor(DAGStatus status, Map<String, Progress> progressMap) {
     try {
       return new TezProgressMonitor(dagClient, status, topSortedWorks, progressMap, console,
-          executionStartTime);
+          executionStartTime, yarnMetricsHelper, yarnQueueName);
     } catch (IOException | TezException e) {
       console.printInfo("Getting  Progress Information: " + e.getMessage() + " stack trace: " +
           ExceptionUtils.getStackTrace(e));

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

@@ -26,6 +26,7 @@
 import org.apache.tez.dag.api.client.Progress;
 import org.apache.tez.dag.api.client.VertexStatus;
 
+import javax.annotation.Nullable;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -44,20 +45,41 @@ public class TezProgressMonitor implements ProgressMonitor {
   private final SessionState.LogHelper console;
   private final long executionStartTime;
   private final DAGStatus status;
+  /** Optional helper for fetching live YARN resource metrics; may be null if disabled. */
+  @Nullable
+  private final YarnMetricsHelper yarnMetricsHelper;
+  /** YARN queue name resolved at construction time; never null. */
+  private final String yarnQueueName;
   Map<String, VertexStatus> vertexStatusMap = new HashMap<>();
   Map<String, VertexProgress> progressCountsMap = new HashMap<>();
 
   /**
    * Try to get most the data required from dagClient in the constructor itself so that even after
-   * the tez job has finished this object can be used for later use.s
+   * the tez job has finished this object can be used for later use.
    */
   TezProgressMonitor(DAGClient dagClient, DAGStatus status, List<BaseWork> topSortedWork,
       Map<String, Progress> progressMap, SessionState.LogHelper console, long executionStartTime)
       throws IOException, TezException {
+    this(dagClient, status, topSortedWork, progressMap, console, executionStartTime,
+        null, YarnMetricsHelper.NA);
+  }
+
+  /**
+   * Full constructor that includes YARN metrics support.
+   *
+   * @param yarnMetricsHelper  live YARN helper; may be {@code null} to disable YARN metrics.
+   * @param yarnQueueName      name of the YARN queue used for this query.
+   */
+  TezProgressMonitor(DAGClient dagClient, DAGStatus status, List<BaseWork> topSortedWork,
+      Map<String, Progress> progressMap, SessionState.LogHelper console, long executionStartTime,
+      @Nullable YarnMetricsHelper yarnMetricsHelper, String yarnQueueName)
+      throws IOException, TezException {
     this.status = status;
     this.topSortedWork = topSortedWork;
     this.console = console;
     this.executionStartTime = executionStartTime;
+    this.yarnMetricsHelper = yarnMetricsHelper;
+    this.yarnQueueName = yarnQueueName != null ? yarnQueueName : YarnMetricsHelper.NA;
     for (Map.Entry<String, Progress> entry : progressMap.entrySet()) {
       String vertexName = entry.getKey();
       progressCountsMap.put(vertexName, new VertexProgress(entry.getValue(), status.getState()));
@@ -330,4 +352,20 @@ public int hashCode() {
   public DAGStatus getStatus() {
     return status;
   }
+
+  /**
+   * Overrides {@link ProgressMonitor#getResourceLines()} to supply real-time YARN queue
+   * and cluster resource lines when a {@link YarnMetricsHelper} is available.
+   *
+   * <p>Called by the render strategies — no {@code instanceof} cast required there.
+   * Returns an empty list (no YARN lines) when metrics are unavailable or disabled.
+   */
+  @Override
+  public List<String> getResourceLines() {
+    if (yarnMetricsHelper == null) {
+      return Collections.emptyList();
+    }
+    YarnMetricsHelper.ResourceInfo info = yarnMetricsHelper.getResourceInfo(yarnQueueName);
+    return Arrays.asList(info.formatQueueLine(), info.formatClusterLine());
+  }
 }