diff --git a/common/src/java/org/apache/hadoop/hive/common/metrics/common/MetricsConstant.java b/common/src/java/org/apache/hadoop/hive/common/metrics/common/MetricsConstant.java index b0d2b85..9dc96f9 100644 --- a/common/src/java/org/apache/hadoop/hive/common/metrics/common/MetricsConstant.java +++ b/common/src/java/org/apache/hadoop/hive/common/metrics/common/MetricsConstant.java @@ -57,4 +57,8 @@ public static final String DELETE_TOTAL_PARTITIONS = "delete_total_count_partitions"; public static final String DIRECTSQL_ERRORS = "directsql_errors"; + + // The number of Hive operations that are waiting to enter the compile block + public static final String WAITING_COMPILE_OPS = "waiting_compile_ops"; + } \ No newline at end of file diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java index 3fecc5c..8a33472 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java @@ -39,6 +39,9 @@ import org.apache.commons.lang.StringUtils; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.hive.common.ValidTxnList; +import org.apache.hadoop.hive.common.metrics.common.Metrics; +import org.apache.hadoop.hive.common.metrics.common.MetricsConstant; +import org.apache.hadoop.hive.common.metrics.common.MetricsFactory; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.conf.HiveConf.ConfVars; import org.apache.hadoop.hive.conf.HiveVariableSource; @@ -1147,6 +1150,17 @@ public CommandProcessorResponse compileAndRespond(String command) { private static final ReentrantLock globalCompileLock = new ReentrantLock(); private int compileInternal(String command) { int ret; + + Metrics metrics = MetricsFactory.getInstance(); + if (metrics != null) { + try { + metrics.incrementCounter(MetricsConstant.WAITING_COMPILE_OPS, 1); + } catch (IOException e) { + // This won't happen if we are using the newer CodaHale metrics. Same for below. + LOG.warn("Error while incrementing metrics counter.", e); + } + } + final ReentrantLock compileLock = tryAcquireCompileLock(isParallelEnabled, command); if (compileLock == null) { @@ -1154,6 +1168,13 @@ private int compileInternal(String command) { } try { + if (metrics != null) { + try { + metrics.decrementCounter(MetricsConstant.WAITING_COMPILE_OPS, 1); + } catch (IOException e) { + LOG.warn("Error while decrementing metrics counter.", e); + } + } ret = compile(command); } finally { compileLock.unlock();