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

Author: architjainjain

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

@@ -159,6 +159,12 @@ public void render(ProgressMonitor monitor) {
         monitor.rows().stream().map(row -> VERTEX_FORMAT.formatted(row.toArray())).collect(Collectors.toList());
     reprintMultiLine(StringUtils.join(printReady, "\n"));
 
+    // Display queue metrics if available (may be multi-line: queue name + metrics)
+    String queueMetrics = monitor.queueMetrics();
+    if (queueMetrics != null && !queueMetrics.isEmpty()) {
+      reprintMultiLine(queueMetrics);
+    }
+
     // -------------------------------------------------------------------------------
     // VERTICES: 03/04            [=================>>-----] 86%  ELAPSED TIME: 1.71 s
     // -------------------------------------------------------------------------------

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 String queueMetrics() {
+      return "";
+    }
   };
 
   List<String> headers();
@@ -65,4 +70,6 @@ public double progressedPercentage() {
   String executionStatus();
 
   double progressedPercentage();
+
+  String queueMetrics();
 }

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

@@ -3982,6 +3982,12 @@ public static enum ConfVars {
     HIVE_SERVER2_TEZ_QUEUE_ACCESS_CHECK("hive.server2.tez.queue.access.check", false,
         "Whether to check user access to explicitly specified YARN queues. " +
           "yarn.resourcemanager.webapp.address must be configured to use this."),
+    HIVE_TEZ_QUEUE_METRICS_REFRESH_INTERVAL("hive.tez.queue.metrics.refresh.interval", "0s",
+        new TimeValidator(TimeUnit.SECONDS),
+        "Interval for refreshing YARN queue resource metrics during Tez query execution. " +
+        "When set to a positive value (e.g. 10s), displays real-time memory, vCore, capacity " +
+        "and application metrics for the YARN queue being used. " +
+        "Set to 0 or negative to disable. Minimum effective value is 1 second."),
     HIVE_SERVER2_TEZ_SESSION_LIFETIME("hive.server2.tez.session.lifetime", "162h",
         new TimeValidator(TimeUnit.HOURS),
         "The lifetime of the Tez sessions launched by HS2 when default sessions are enabled.\n" +

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

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
 import org.apache.hadoop.hive.ql.wm.WmContext;
 import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.tez.client.TezClient;
 import org.apache.tez.dag.api.TezException;
 import org.apache.tez.dag.api.client.DAGStatus;
@@ -86,6 +87,7 @@ public String toString() {
 
   HiveConf getConf();
   TezClient getTezClient();
+  YarnClient getYarnClient();
   boolean isOpen();
   boolean isOpening();
   boolean getDoAsEnabled();

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

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hive.ql.wm.WmContext;
 import org.apache.hadoop.hive.registry.impl.TezAmInstance;
 import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.tez.client.TezClient;
 import org.apache.tez.dag.api.TezException;
 import org.apache.tez.dag.api.client.DAGStatus;
@@ -337,6 +338,11 @@ public TezClient getTezClient() {
     return baseSession.getTezClient();
   }
 
+  @Override
+  public YarnClient getYarnClient() {
+    return baseSession.getYarnClient();
+  }
+
   @Override
   public boolean isOpening() {
     return baseSession.isOpening();

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

@@ -73,6 +73,7 @@
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.client.api.YarnClient;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.tez.client.TezClient;
 import org.apache.tez.common.TezUtils;
@@ -117,6 +118,7 @@ public class TezSessionState implements TezSession {
   Path tezScratchDir;
   protected LocalResource appJarLr;
   private TezClient session;
+  private YarnClient yarnClient;
   private Future<TezClient> sessionFuture;
   /** Console used for user feedback during async session opening. */
   private LogHelper console;
@@ -750,6 +752,17 @@ public void close(boolean keepDagFilesDir) throws Exception {
           closeClient(asyncSession);
         }
       }
+
+      // Stop YarnClient if it was initialized
+      if (yarnClient != null) {
+        try {
+          LOG.info("Stopping YarnClient for session: {}", sessionId);
+          yarnClient.stop();
+          yarnClient = null;
+        } catch (Exception e) {
+          LOG.warn("Error stopping YarnClient for session {}: {}", sessionId, e.getMessage());
+        }
+      }
     } finally {
       try {
         cleanupScratchDir();
@@ -795,6 +808,20 @@ public String getSessionId() {
 
   protected final void setTezClient(TezClient session) {
     this.session = session;
+
+    // Initialize YarnClient for queue metrics collection
+    if (session != null && yarnClient == null) {
+      try {
+        yarnClient = YarnClient.createYarnClient();
+        yarnClient.init(conf);
+        yarnClient.start();
+        LOG.info("YarnClient initialized for session: {}", sessionId);
+      } catch (Exception e) {
+        LOG.warn("Failed to initialize YarnClient for metrics collection: {}", e.getMessage());
+        LOG.debug("Full exception for YarnClient initialization failure", e);
+        yarnClient = null;
+      }
+    }
   }
 
   @Override
@@ -820,6 +847,11 @@ public TezClient getTezClient() {
     return session;
   }
 
+  @Override
+  public YarnClient getYarnClient() {
+    return yarnClient;
+  }
+
   @Override
   public LocalResource getAppJarLr() {
     return appJarLr;