This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 8e8dd5155da HIVE-29583: Profile JDBC executions in Metastore (#6464)
8e8dd5155da is described below
commit 8e8dd5155da4e74c966b419e388cd4b66b118892
Author: dengzh <[email protected]>
AuthorDate: Fri Jul 24 11:48:37 2026 +0800
HIVE-29583: Profile JDBC executions in Metastore (#6464)
---
.../hadoop/hive/metastore/conf/MetastoreConf.java | 10 +
.../hadoop/hive/metastore/HMSHandlerContext.java | 62 ++++
.../hadoop/hive/metastore/RetryingHMSHandler.java | 16 +
.../metastore/datasource/DataSourceProvider.java | 34 ++-
.../datasource/DbCPDataSourceProvider.java | 24 +-
.../datasource/HikariCPDataSourceProvider.java | 19 +-
.../metastore/datasource/MetastoreConnection.java | 338 +++++++++++++++++++++
.../metastore/datasource/MetastoreStatement.java | 213 +++++++++++++
.../metastore/datasource/ProfilingDataSource.java | 103 +++++++
.../metastore/handler/AbstractRequestHandler.java | 27 +-
.../hive/metastore/metrics/MetricsConstants.java | 2 +
.../datasource/TestProfilingDataSource.java | 164 ++++++++++
12 files changed, 972 insertions(+), 40 deletions(-)
diff --git
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
index acd5929b8b9..3a4d7a173d4 100644
---
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
+++
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
@@ -652,6 +652,16 @@ public enum ConfVars {
"hive.txn.acid.metrics.delta.pct.threshold", 0.01f,
"Percentage (fractional) size of the delta files relative to the base
directory. Deltas smaller than this threshold " +
"count as small deltas. Default 0.01 = 1%.)"),
+
METASTORE_JDBC_SLOW_QUERY_THRESHOLD("metastore.jdbc.execution.logSlowQueriesThreshold",
"metastore.jdbc.execution.logSlowQueriesThreshold",
+ 3000, TimeUnit.MILLISECONDS, "Log the slow jdbc query that Metastore
has been waiting for the result beyond the threshold(ms), " +
+ "should turn on the metastore.profile.jdbc.execution first"),
+ METASTORE_PROFILE_JDBC_EXECUTION("metastore.profile.jdbc.execution",
"metastore.profile.jdbc.execution", false,
+ "Turn on to profile JDBC executions at the statement layer (slow-query
logging, per-query metrics\n" +
+ "about metastore.jdbc.profile.thrift.apis, and aggregate JDBC
summaries for Thrift APIs )."),
+ METASTORE_PROFILE_JDBC_THRIFT_APIS("metastore.jdbc.profile.thrift.apis",
"metastore.jdbc.profile.thrift.apis",
+ "get_table_req,get_database_req",
+ "Thrift API method names for which to record the per-query metrics.\n"
+
+ "Per-query slow detection applies to all JDBC executions while
profiling is on."),
COMPACTOR_INITIATOR_ON("metastore.compactor.initiator.on",
"hive.compactor.initiator.on", true,
"Whether to run the initiator thread on this metastore instance or
not.\n" +
"Set this to true on one instance of the Thrift metastore service
as part of turning\n" +
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandlerContext.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandlerContext.java
index d52755786d4..90dd0a4a7c9 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandlerContext.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/HMSHandlerContext.java
@@ -20,11 +20,14 @@
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.handler.BaseHandler;
import org.apache.hadoop.hive.metastore.txn.TxnStore;
import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.slf4j.Logger;
/**
* When one hms client connects in, we create a handler context for it.
@@ -55,6 +58,57 @@ public final class HMSHandlerContext {
private Map<String, com.codahale.metrics.Timer.Context> timerContexts = new
HashMap<>();
+ // Unique ID of current thrift call
+ private CallCtx callCtx;
+
+ public static final class CallCtx {
+ private final String methodName;
+ private long totalTime;
+ private int queryCount;
+ private long maxJdbcTimeMs;
+
+ public CallCtx(String methodName) {
+ this.methodName = methodName;
+ }
+
+ public String methodName() {
+ return methodName;
+ }
+
+ public void recordJdbcExecution(long timeSpentMs) {
+ queryCount++;
+ totalTime += timeSpentMs;
+ maxJdbcTimeMs = Math.max(maxJdbcTimeMs, timeSpentMs);
+ }
+
+ public int getQueryCount() {
+ return queryCount;
+ }
+
+ public long getTotalTime() {
+ return totalTime;
+ }
+
+ public long getMaxJdbcTimeMs() {
+ return maxJdbcTimeMs;
+ }
+
+ public void logJdbcSummary(Logger Log, Configuration configuration) {
+ if (queryCount == 0) {
+ return;
+ }
+ Log.debug("{} jdbc summary: {} queries, {} ms total, {} ms max per
query",
+ methodName, queryCount, totalTime, maxJdbcTimeMs);
+ long slowThreshold = MetastoreConf.getTimeVar(configuration,
+ MetastoreConf.ConfVars.METASTORE_JDBC_SLOW_QUERY_THRESHOLD,
TimeUnit.MILLISECONDS);
+ if (slowThreshold > 0 && totalTime > slowThreshold) {
+ Log.warn("{} jdbc summary: {} queries, {} ms total, {} ms max per
query (no single query exceeded "
+ + "the slow threshold, but aggregate JDBC time is high)",
+ methodName, queryCount, totalTime, maxJdbcTimeMs);
+ }
+ }
+ }
+
private HMSHandlerContext() {
}
@@ -89,6 +143,14 @@ public static Map<String,
com.codahale.metrics.Timer.Context> getTimerContexts()
return context.get().timerContexts;
}
+ public static Optional<CallCtx> getCallCtx() {
+ return Optional.ofNullable(context.get().callCtx);
+ }
+
+ public static void setCallCtx(CallCtx ctx) {
+ context.get().callCtx = ctx;
+ }
+
public static void setRawStore(RawStore rawStore) {
context.get().rawStore = rawStore;
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RetryingHMSHandler.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RetryingHMSHandler.java
index 0d52107ad09..2620177aaa7 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RetryingHMSHandler.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/RetryingHMSHandler.java
@@ -21,6 +21,7 @@
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.UndeclaredThrowableException;
+import java.util.Optional;
import java.util.concurrent.TimeUnit;
import javax.jdo.JDOException;
@@ -44,10 +45,12 @@ public class RetryingHMSHandler extends
AbstractHMSHandlerProxy {
private final long retryInterval;
private final int retryLimit;
+ private final boolean local;
public RetryingHMSHandler(Configuration conf, IHMSHandler baseHandler,
boolean local)
throws MetaException {
super(conf, baseHandler, local);
+ this.local = local;
retryInterval = MetastoreConf.getTimeVar(origConf,
ConfVars.HMS_HANDLER_INTERVAL, TimeUnit.MILLISECONDS);
retryLimit = MetastoreConf.getIntVar(origConf,
ConfVars.HMS_HANDLER_ATTEMPTS);
@@ -87,9 +90,22 @@ public Result invokeInternal(final Object proxy, final
Method method, final Obje
}
Object object = null;
boolean isStarted = Deadline.startTimer(method.getName());
+ boolean clearCtxCall = false;
try {
+ if (!local) {
+ Optional<HMSHandlerContext.CallCtx> callCtx =
HMSHandlerContext.getCallCtx();
+ if (callCtx.isEmpty()) {
+ HMSHandlerContext.setCallCtx(new
HMSHandlerContext.CallCtx(method.getName()));
+ clearCtxCall = true;
+ }
+ }
object = method.invoke(baseHandler, args);
} finally {
+ if (clearCtxCall) {
+ HMSHandlerContext.getCallCtx()
+ .ifPresent(ctx -> ctx.logJdbcSummary(LOG, getActiveConf()));
+ HMSHandlerContext.setCallCtx(null);
+ }
if (isStarted) {
Deadline.stopTimer();
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.java
index 78fa406444b..f5baa8ad2ce 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.java
@@ -20,11 +20,15 @@
import java.io.Closeable;
import java.io.IOException;
import java.sql.SQLException;
+import java.util.Map;
import java.util.Properties;
+import java.util.function.Consumer;
import javax.sql.DataSource;
import com.google.common.collect.Iterables;
+
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.DatabaseProduct;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
public interface DataSourceProvider {
@@ -43,7 +47,14 @@ default DataSource create(Configuration hdpConfig) throws
SQLException {
* @param maxPoolSize the maximum size of the connection pool
* @return the new connection pool
*/
- DataSource create(Configuration hdpConfig, int maxPoolSize) throws
SQLException;
+ default DataSource create(Configuration hdpConfig, int maxPoolSize) throws
SQLException {
+ return maybeWrapForProfiling(hdpConfig, createPool(hdpConfig,
maxPoolSize));
+ }
+
+ /**
+ * Creates the underlying connection pool without JDBC profiling decoration.
+ */
+ DataSource createPool(Configuration hdpConfig, int maxPoolSize) throws
SQLException;
/**
* Get the declared pooling type string. This is used to check against the
constant in
@@ -76,14 +87,33 @@ static String getMetastoreJdbcPasswd(Configuration conf)
throws SQLException {
}
}
- static String getMetastoreJdbcDriverUrl(Configuration conf) throws
SQLException {
+ static String getMetastoreJdbcDriverUrl(Configuration conf) {
return MetastoreConf.getVar(conf, MetastoreConf.ConfVars.CONNECT_URL_KEY);
}
+ static DataSource maybeWrapForProfiling(Configuration conf, DataSource
dataSource) {
+ if (MetastoreConf.getBoolVar(conf,
MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_EXECUTION)) {
+ return new ProfilingDataSource(dataSource, conf);
+ }
+ return dataSource;
+ }
+
static String getDataSourceName(Configuration conf) {
return conf.get(DataSourceNameConfigurator.DATA_SOURCE_NAME);
}
+ static void preparePool(Configuration configuration, Consumer<String>
initSql,
+ Consumer<Map.Entry<String, String>> dataSourceProps) {
+ String url = MetastoreConf.getVar(configuration,
MetastoreConf.ConfVars.CONNECT_URL_KEY);
+ DatabaseProduct dbProduct = DatabaseProduct.determineDatabaseProduct(url,
configuration);
+ String s = dbProduct.getPrepareTxnStmt();
+ if (s != null) {
+ initSql.accept(s);
+ }
+ Map<String, String> properties = dbProduct.getDataSourceProperties();
+ properties.entrySet().forEach(dataSourceProps);
+ }
+
class DataSourceNameConfigurator implements Closeable {
static final String DATA_SOURCE_NAME =
"metastore.DataSourceProvider.pool.name";
private final Configuration configuration;
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java
index 92160541a50..aa0e6c00a54 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java
@@ -20,7 +20,6 @@
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collections;
-import java.util.Map;
import javax.sql.DataSource;
@@ -32,7 +31,6 @@
import org.apache.commons.pool2.impl.BaseObjectPoolConfig;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.metastore.DatabaseProduct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -58,7 +56,7 @@ public class DbCPDataSourceProvider implements
DataSourceProvider {
@SuppressWarnings({ "rawtypes", "unchecked" })
@Override
- public DataSource create(Configuration hdpConfig, int maxPoolSize) throws
SQLException {
+ public DataSource createPool(Configuration hdpConfig, int maxPoolSize)
throws SQLException {
String poolName = DataSourceProvider.getDataSourceName(hdpConfig);
LOG.info("Creating dbcp connection pool for the MetaStore, maxPoolSize:
{}, name: {}", maxPoolSize, poolName);
@@ -73,12 +71,6 @@ public DataSource create(Configuration hdpConfig, int
maxPoolSize) throws SQLExc
dbcpDs.setDefaultReadOnly(false);
dbcpDs.setDefaultAutoCommit(true);
- DatabaseProduct dbProduct =
DatabaseProduct.determineDatabaseProduct(driverUrl, hdpConfig);
- Map<String, String> props = dbProduct.getDataSourceProperties();
- for (Map.Entry<String, String> kv : props.entrySet()) {
- dbcpDs.setConnectionProperties(kv.getKey() + "=" + kv.getValue());
- }
-
long connectionTimeout = hdpConfig.getLong(CONNECTION_TIMEOUT_PROPERTY,
30000L);
int connectionMaxIlde = hdpConfig.getInt(CONNECTION_MAX_IDLE_PROPERTY, 8);
int connectionMinIlde = hdpConfig.getInt(CONNECTION_MIN_IDLE_PROPERTY, 0);
@@ -127,9 +119,17 @@ public DataSource create(Configuration hdpConfig, int
maxPoolSize) throws SQLExc
objectPool.setSoftMinEvictableIdleDuration(Duration.ofMillis(600 *
1000));
}
}
- String stmt = dbProduct.getPrepareTxnStmt();
- if (stmt != null) {
-
poolableConnFactory.setConnectionInitSql(Collections.singletonList(stmt));
+ StringBuilder connectionProperties = new StringBuilder();
+ DataSourceProvider.preparePool(hdpConfig,
+ stmt ->
poolableConnFactory.setConnectionInitSql(Collections.singletonList(stmt)),
+ kv -> {
+ if (!connectionProperties.isEmpty()) {
+ connectionProperties.append(';');
+ }
+
connectionProperties.append(kv.getKey()).append('=').append(kv.getValue());
+ });
+ if (!connectionProperties.isEmpty()) {
+ dbcpDs.setConnectionProperties(connectionProperties.toString());
}
return new PoolingDataSource(objectPool);
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/HikariCPDataSourceProvider.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/HikariCPDataSourceProvider.java
index 91f1ea3d557..9958d5303fa 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/HikariCPDataSourceProvider.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/HikariCPDataSourceProvider.java
@@ -21,7 +21,6 @@
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.metastore.DatabaseProduct;
import org.apache.hadoop.hive.metastore.metrics.Metrics;
import org.apache.hadoop.hive.metastore.utils.StringUtils;
import org.slf4j.Logger;
@@ -29,7 +28,6 @@
import javax.sql.DataSource;
import java.sql.SQLException;
-import java.util.Map;
import java.util.Properties;
/**
@@ -42,7 +40,7 @@ public class HikariCPDataSourceProvider implements
DataSourceProvider {
static final String HIKARI = "hikaricp";
@Override
- public DataSource create(Configuration hdpConfig, int maxPoolSize) throws
SQLException {
+ public DataSource createPool(Configuration hdpConfig, int maxPoolSize)
throws SQLException {
String poolName = DataSourceProvider.getDataSourceName(hdpConfig);
LOG.info("Creating Hikari connection pool for the MetaStore, maxPoolSize:
{}, name: {}", maxPoolSize, poolName);
@@ -77,19 +75,8 @@ public DataSource create(Configuration hdpConfig, int
maxPoolSize) throws SQLExc
config.setMinimumIdle(Math.min(maxPoolSize, minimumIdle));
}
- DatabaseProduct dbProduct =
DatabaseProduct.determineDatabaseProduct(driverUrl, hdpConfig);
-
- String s = dbProduct.getPrepareTxnStmt();
- if (s!= null) {
- config.setConnectionInitSql(s);
- }
-
- Map<String, String> props = dbProduct.getDataSourceProperties();
-
- for ( Map.Entry<String, String> kv : props.entrySet()) {
- config.addDataSourceProperty(kv.getKey(), kv.getValue());
- }
-
+ DataSourceProvider.preparePool(hdpConfig, s ->
config.setConnectionInitSql(s),
+ entry -> config.addDataSourceProperty(entry.getKey(),
entry.getValue()));
return new HikariDataSource(initMetrics(config));
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreConnection.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreConnection.java
new file mode 100644
index 00000000000..22e7795838e
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreConnection.java
@@ -0,0 +1,338 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.datasource;
+
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.CallableStatement;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.NClob;
+import java.sql.PreparedStatement;
+import java.sql.SQLClientInfoException;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import java.sql.SQLXML;
+import java.sql.Savepoint;
+import java.sql.Statement;
+import java.sql.Struct;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+
+import org.apache.hadoop.conf.Configuration;
+
+public final class MetastoreConnection implements Connection {
+ private final Configuration configuration;
+ private final Connection delegate;
+
+ MetastoreConnection(Connection connection, Configuration conf) {
+ this.delegate = Objects.requireNonNull(connection, "delegate connection");
+ this.configuration = Objects.requireNonNull(conf, "configuration");
+ }
+
+ @Override
+ public Statement createStatement() throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.createStatement(), null);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.prepareStatement(sql), sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.prepareCall(sql), sql);
+ }
+
+ @Override
+ public String nativeSQL(String sql) throws SQLException {
+ return delegate.nativeSQL(sql);
+ }
+
+ @Override
+ public void setAutoCommit(boolean autoCommit) throws SQLException {
+ delegate.setAutoCommit(autoCommit);
+ }
+
+ @Override
+ public boolean getAutoCommit() throws SQLException {
+ return delegate.getAutoCommit();
+ }
+
+ @Override
+ public void commit() throws SQLException {
+ delegate.commit();
+ }
+
+ @Override
+ public void rollback() throws SQLException {
+ delegate.rollback();
+ }
+
+ @Override
+ public void close() throws SQLException {
+ delegate.close();
+ }
+
+ @Override
+ public boolean isClosed() throws SQLException {
+ return delegate.isClosed();
+ }
+
+ @Override
+ public DatabaseMetaData getMetaData() throws SQLException {
+ return delegate.getMetaData();
+ }
+
+ @Override
+ public void setReadOnly(boolean readOnly) throws SQLException {
+ delegate.setReadOnly(readOnly);
+ }
+
+ @Override
+ public boolean isReadOnly() throws SQLException {
+ return delegate.isReadOnly();
+ }
+
+ @Override
+ public void setCatalog(String catalog) throws SQLException {
+ delegate.setCatalog(catalog);
+ }
+
+ @Override
+ public String getCatalog() throws SQLException {
+ return delegate.getCatalog();
+ }
+
+ @Override
+ public void setTransactionIsolation(int level) throws SQLException {
+ delegate.setTransactionIsolation(level);
+ }
+
+ @Override
+ public int getTransactionIsolation() throws SQLException {
+ return delegate.getTransactionIsolation();
+ }
+
+ @Override
+ public SQLWarning getWarnings() throws SQLException {
+ return delegate.getWarnings();
+ }
+
+ @Override
+ public void clearWarnings() throws SQLException {
+ delegate.clearWarnings();
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int
resultSetConcurrency) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.createStatement(resultSetType, resultSetConcurrency), null);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int
resultSetConcurrency)
+ throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.prepareStatement(sql, resultSetType, resultSetConcurrency),
sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int
resultSetConcurrency) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.prepareCall(sql, resultSetType, resultSetConcurrency), sql);
+ }
+
+ @Override
+ public Map<String, Class<?>> getTypeMap() throws SQLException {
+ return delegate.getTypeMap();
+ }
+
+ @Override
+ public void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+ delegate.setTypeMap(map);
+ }
+
+ @Override
+ public void setHoldability(int holdability) throws SQLException {
+ delegate.setHoldability(holdability);
+ }
+
+ @Override
+ public int getHoldability() throws SQLException {
+ return delegate.getHoldability();
+ }
+
+ @Override
+ public Savepoint setSavepoint() throws SQLException {
+ return delegate.setSavepoint();
+ }
+
+ @Override
+ public Savepoint setSavepoint(String name) throws SQLException {
+ return delegate.setSavepoint(name);
+ }
+
+ @Override
+ public void rollback(Savepoint savepoint) throws SQLException {
+ delegate.rollback(savepoint);
+ }
+
+ @Override
+ public void releaseSavepoint(Savepoint savepoint) throws SQLException {
+ delegate.releaseSavepoint(savepoint);
+ }
+
+ @Override
+ public Statement createStatement(int resultSetType, int
resultSetConcurrency, int resultSetHoldability)
+ throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.createStatement(resultSetType, resultSetConcurrency,
resultSetHoldability), null);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int resultSetType, int
resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.prepareStatement(sql, resultSetType, resultSetConcurrency,
resultSetHoldability), sql);
+ }
+
+ @Override
+ public CallableStatement prepareCall(String sql, int resultSetType, int
resultSetConcurrency,
+ int resultSetHoldability) throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
+ delegate.prepareCall(sql, resultSetType, resultSetConcurrency,
resultSetHoldability), sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys)
throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.prepareStatement(sql, autoGeneratedKeys), sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, int[] columnIndexes)
throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.prepareStatement(sql, columnIndexes), sql);
+ }
+
+ @Override
+ public PreparedStatement prepareStatement(String sql, String[] columnNames)
throws SQLException {
+ return MetastoreStatement.getProxyStatement(configuration,
delegate.prepareStatement(sql, columnNames), sql);
+ }
+
+ @Override
+ public Clob createClob() throws SQLException {
+ return delegate.createClob();
+ }
+
+ @Override
+ public Blob createBlob() throws SQLException {
+ return delegate.createBlob();
+ }
+
+ @Override
+ public NClob createNClob() throws SQLException {
+ return delegate.createNClob();
+ }
+
+ @Override
+ public SQLXML createSQLXML() throws SQLException {
+ return delegate.createSQLXML();
+ }
+
+ @Override
+ public boolean isValid(int timeout) throws SQLException {
+ return delegate.isValid(timeout);
+ }
+
+ @Override
+ public void setClientInfo(String name, String value) throws
SQLClientInfoException {
+ delegate.setClientInfo(name, value);
+ }
+
+ @Override
+ public void setClientInfo(Properties properties) throws
SQLClientInfoException {
+ delegate.setClientInfo(properties);
+ }
+
+ @Override
+ public String getClientInfo(String name) throws SQLException {
+ return delegate.getClientInfo(name);
+ }
+
+ @Override
+ public Properties getClientInfo() throws SQLException {
+ return delegate.getClientInfo();
+ }
+
+ @Override
+ public Array createArrayOf(String typeName, Object[] elements) throws
SQLException {
+ return delegate.createArrayOf(typeName, elements);
+ }
+
+ @Override
+ public Struct createStruct(String typeName, Object[] attributes) throws
SQLException {
+ return delegate.createStruct(typeName, attributes);
+ }
+
+ @Override
+ public void setSchema(String schema) throws SQLException {
+ delegate.setSchema(schema);
+ }
+
+ @Override
+ public String getSchema() throws SQLException {
+ return delegate.getSchema();
+ }
+
+ @Override
+ public void abort(Executor executor) throws SQLException {
+ delegate.abort(executor);
+ }
+
+ @Override
+ public void setNetworkTimeout(Executor executor, int milliseconds) throws
SQLException {
+ delegate.setNetworkTimeout(executor, milliseconds);
+ }
+
+ @Override
+ public int getNetworkTimeout() throws SQLException {
+ return delegate.getNetworkTimeout();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ if (iface.isInstance(this)) {
+ return iface.cast(this);
+ }
+ return delegate.unwrap(iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return iface.isInstance(this) || delegate.isWrapperFor(iface);
+ }
+
+ public Configuration getConfiguration() {
+ return configuration;
+ }
+}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java
new file mode 100644
index 00000000000..051b251bd82
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java
@@ -0,0 +1,213 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.datasource;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Timer;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.lang.reflect.UndeclaredThrowableException;
+import java.sql.Statement;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.lang3.ClassUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HMSHandlerContext;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.metrics.Metrics;
+import org.apache.hadoop.hive.metastore.metrics.MetricsConstants;
+import org.apache.hadoop.hive.metastore.utils.JavaUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@SuppressWarnings("unchecked")
+public final class MetastoreStatement implements InvocationHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(MetastoreStatement.class);
+ static final String EXEC_HOOK = "metastore.jdbc.execution.hook";
+ static final Set<String> QUERY_EXECUTION =
+ Set.of("executeQuery", "executeUpdate", "execute", "executeBatch");
+
+ private final String rawSql;
+ private final Statement delegate;
+ private final Configuration configuration;
+ private final MetastoreStatementHook hook;
+ private final long slowQueryThreshold;
+
+ private MetastoreStatement(Configuration conf, Statement statement, String
rawSql) {
+ this.configuration = Objects.requireNonNull(conf);
+ this.rawSql = rawSql;
+ this.delegate = Objects.requireNonNull(statement);
+ this.hook = resolveHook(conf);
+ this.slowQueryThreshold = MetastoreConf.getTimeVar(configuration,
+ MetastoreConf.ConfVars.METASTORE_JDBC_SLOW_QUERY_THRESHOLD,
TimeUnit.MILLISECONDS);
+ }
+
+ private MetastoreStatementHook resolveHook(Configuration conf) {
+ String className = conf.get(EXEC_HOOK, "");
+ if (StringUtils.isEmpty(className)) {
+ return new ThriftApiProfiler(configuration);
+ }
+ try {
+ return JavaUtils.newInstance(JavaUtils.getClass(className.trim(),
MetastoreStatementHook.class),
+ new Class[] {Configuration.class}, new Object[] {conf});
+ } catch (MetaException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ }
+ }
+
+ public static <T extends Statement> T getProxyStatement(Configuration
configuration, T delegate, String rawSql) {
+ MetastoreStatement handler = new MetastoreStatement(configuration,
delegate, rawSql);
+ return (T) Proxy.newProxyInstance(JavaUtils.getClassLoader(),
+ ClassUtils.getAllInterfaces(delegate.getClass()).toArray(new
Class[0]), handler);
+ }
+
+ @Override
+ public Object invoke(Object proxy, Method method, Object[] args) throws
Throwable {
+ Timer.Context ctx = null;
+ try {
+ boolean monitor = hook.profile(rawSql, method, args);
+ if (Metrics.getRegistry() != null && monitor) {
+ String metricName = hook.getMetricName(method);
+ Timer timer = Metrics.getOrCreateTimer(metricName);
+ if (timer != null) {
+ ctx = timer.time();
+ }
+ }
+ long start = System.currentTimeMillis();
+ hook.preRun(method, args);
+ Object result = method.invoke(delegate, args);
+ hook.postRun(method, args, result);
+ long timeSpent = System.currentTimeMillis() - start;
+ if (monitor) {
+ String statement = rawSql != null ? rawSql
+ : (args != null && args.length > 0 ? (String) args[0] : "no sql
found");
+ LOG.debug("Jdbc query: {} completed in {} ms", statement, timeSpent);
+ }
+ if (isQueryExecution(method)) {
+ HMSHandlerContext.getCallCtx().ifPresent(callCtx ->
callCtx.recordJdbcExecution(timeSpent));
+ logExecution(timeSpent, rawSql, method, args);
+ }
+ return result;
+ } catch (InvocationTargetException | UndeclaredThrowableException e) {
+ throw e.getCause() != null ? e.getCause() : e;
+ } finally {
+ if (ctx != null) {
+ ctx.stop();
+ }
+ }
+ }
+
+ private void logExecution(long timeSpent, String sql, Method method,
Object[] args) {
+ if (isSlowExecution(timeSpent)) {
+ Object[] printableArgs = args;
+ if (args != null && args.length > 10) {
+ printableArgs = new Object[10];
+ System.arraycopy(args, 0, printableArgs, 0, 7);
+ System.arraycopy(args, args.length - 2, printableArgs, 8, 2);
+ printableArgs[7] = "....";
+ }
+ LOG.warn("Slow execution detected, method: {}, time taken: {} ms, args
size: {}, args: {}{}",
+ method.getName(), timeSpent,
+ args != null ? args.length : 0, Arrays.toString(printableArgs), sql
!= null ? ", sql: " + sql : "");
+ if (Metrics.getRegistry() != null) {
+ Counter slowQueries =
Metrics.getOrCreateCounter(MetricsConstants.JDBC_SLOW_QUERIES);
+ slowQueries.inc();
+ }
+ }
+ }
+
+ private boolean isSlowExecution(long timeSpent) {
+ return slowQueryThreshold > 0 && timeSpent > slowQueryThreshold;
+ }
+
+ private static boolean isQueryExecution(Method method) {
+ return QUERY_EXECUTION.contains(method.getName());
+ }
+
+ public interface MetastoreStatementHook {
+ /**
+ * Whether should monitor the current call, this method gives a chance to
profile a specific pattern of queries.
+ * For example, we use {@link ThriftApiProfiler} to profile the queries
originated from a set of specific APIs.
+ * @param sql The sql being executed, it might be null for {@link
Statement#execute}, for this case
+ * need to obtain the sql from args, the method input.
+ * @param method Method which is being called
+ * @param args The method input
+ * @return true for profiling this call, false otherwise
+ */
+ boolean profile(String sql, Method method, Object[] args);
+
+ String getMetricName(Method method);
+
+ default void preRun(Method method, Object[] args) {
+ }
+
+ default void postRun(Method method, Object[] args, Object result) {
+ }
+ }
+
+ /**
+ * This class is used to profile the statement originated from specific
thrift API calls
+ */
+ public static class ThriftApiProfiler implements MetastoreStatementHook {
+ private final Configuration configuration;
+ private final Set<String> profiledApis;
+
+ public ThriftApiProfiler(Configuration configuration) {
+ this.configuration = Objects.requireNonNull(configuration);
+ this.profiledApis = getProfiledApis();
+ }
+
+ private Set<String> getProfiledApis() {
+ String thriftApis = MetastoreConf.getVar(configuration,
+ MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_THRIFT_APIS);
+ Set<String> profiledApis = new HashSet<>();
+ for (String thriftApi : thriftApis.split(",")) {
+ String trimmedThriftApi = thriftApi.trim();
+ if (!trimmedThriftApi.isEmpty()) {
+ profiledApis.add(trimmedThriftApi);
+ }
+ }
+ return profiledApis;
+ }
+
+ @Override
+ public boolean profile(String sql, Method method, Object[] args) {
+ if (!isQueryExecution(method) || profiledApis.isEmpty()) {
+ return false;
+ }
+ Optional<HMSHandlerContext.CallCtx> ctxCall =
HMSHandlerContext.getCallCtx();
+ return ctxCall.isPresent() &&
profiledApis.contains(ctxCall.get().methodName());
+ }
+
+ @Override
+ public String getMetricName(Method method) {
+ return MetricsConstants.JDBC_EXECUTION + method.getName();
+ }
+ }
+}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/ProfilingDataSource.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/ProfilingDataSource.java
new file mode 100644
index 00000000000..f142de5d7ef
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/ProfilingDataSource.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.datasource;
+
+import java.io.PrintWriter;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.Objects;
+import java.util.logging.Logger;
+
+import javax.sql.DataSource;
+
+import org.apache.hadoop.conf.Configuration;
+
+/**
+ * Decorates a {@link DataSource} to wrap checked-out connections for JDBC
profiling.
+ */
+public class ProfilingDataSource implements DataSource, AutoCloseable {
+
+ private final DataSource delegate;
+ private final Configuration configuration;
+
+ public ProfilingDataSource(DataSource delegate, Configuration configuration)
{
+ this.delegate = Objects.requireNonNull(delegate, "delegate datasource");
+ this.configuration = Objects.requireNonNull(configuration,
"configuration");
+ }
+
+ public DataSource getDelegate() {
+ return delegate;
+ }
+
+ @Override
+ public Connection getConnection() throws SQLException {
+ return new MetastoreConnection(delegate.getConnection(), configuration);
+ }
+
+ @Override
+ public Connection getConnection(String username, String password) throws
SQLException {
+ return new MetastoreConnection(delegate.getConnection(username, password),
configuration);
+ }
+
+ @Override
+ public PrintWriter getLogWriter() throws SQLException {
+ return delegate.getLogWriter();
+ }
+
+ @Override
+ public void setLogWriter(PrintWriter out) throws SQLException {
+ delegate.setLogWriter(out);
+ }
+
+ @Override
+ public void setLoginTimeout(int seconds) throws SQLException {
+ delegate.setLoginTimeout(seconds);
+ }
+
+ @Override
+ public int getLoginTimeout() throws SQLException {
+ return delegate.getLoginTimeout();
+ }
+
+ @Override
+ public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+ return delegate.getParentLogger();
+ }
+
+ @Override
+ public <T> T unwrap(Class<T> iface) throws SQLException {
+ if (iface.isInstance(this)) {
+ return iface.cast(this);
+ }
+ return delegate.unwrap(iface);
+ }
+
+ @Override
+ public boolean isWrapperFor(Class<?> iface) throws SQLException {
+ return iface.isInstance(this) || delegate.isWrapperFor(iface);
+ }
+
+ @Override
+ public void close() throws Exception {
+ if (delegate instanceof AutoCloseable) {
+ ((AutoCloseable) delegate).close();
+ }
+ }
+}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java
index 0f078f20040..01881a8721e 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/handler/AbstractRequestHandler.java
@@ -25,6 +25,7 @@
import java.io.IOException;
import java.lang.reflect.Modifier;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CancellationException;
@@ -41,6 +42,7 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.reflect.MethodUtils;
+import org.apache.hadoop.hive.metastore.HMSHandlerContext;
import org.apache.hadoop.hive.metastore.IHMSHandler;
import org.apache.hadoop.hive.metastore.api.AsyncOperationResp;
import org.apache.hadoop.hive.metastore.api.MetaException;
@@ -139,21 +141,26 @@ private Future<Result> executeRequest() {
timerContext = null;
}
+ Optional<HMSHandlerContext.CallCtx> callCtx =
HMSHandlerContext.getCallCtx();
Future<Result> resultFuture = executor.submit(() -> {
+ if (async) {
+ callCtx.ifPresent(HMSHandlerContext::setCallCtx);
+ }
A resultV = null;
- beforeExecute();
try {
- resultV = execute();
- } finally {
+ beforeExecute();
try {
- if (async) {
- REQUEST_CLEANER.schedule(() -> ID_TO_HANDLER.remove(id), 1,
TimeUnit.HOURS);
- }
- afterExecute(resultV);
+ resultV = execute();
} finally {
- if (timerContext != null) {
- timerContext.stop();
- }
+ afterExecute(resultV);
+ }
+ } finally {
+ if (async) {
+ HMSHandlerContext.clear();
+ REQUEST_CLEANER.schedule(() -> ID_TO_HANDLER.remove(id), 1,
TimeUnit.HOURS);
+ }
+ if (timerContext != null) {
+ timerContext.stop();
}
}
return async ? resultV.shrinkIfNecessary() : resultV;
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metrics/MetricsConstants.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metrics/MetricsConstants.java
index 46512ab3c4b..cd8f36452de 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metrics/MetricsConstants.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metrics/MetricsConstants.java
@@ -71,6 +71,8 @@ public class MetricsConstants {
public static final String DELETE_TOTAL_PARTITIONS =
"delete_total_count_partitions";
public static final String DIRECTSQL_ERRORS = "directsql_errors";
+ public static final String JDBC_SLOW_QUERIES = "jdbc_slow_queries";
+ public static final String JDBC_EXECUTION = "jdbc_";
public static final String JVM_PAUSE_INFO = "jvm.pause.info-threshold";
public static final String JVM_PAUSE_WARN = "jvm.pause.warn-threshold";
diff --git
a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/datasource/TestProfilingDataSource.java
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/datasource/TestProfilingDataSource.java
new file mode 100644
index 00000000000..78ae1aa13af
--- /dev/null
+++
b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/datasource/TestProfilingDataSource.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hive.metastore.datasource;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Timer;
+import com.zaxxer.hikari.HikariDataSource;
+
+import javax.sql.DataSource;
+import java.lang.reflect.Method;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.dbcp2.PoolingDataSource;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.HMSHandlerContext;
+import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.metrics.Metrics;
+import org.apache.hadoop.hive.metastore.metrics.MetricsConstants;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(MetastoreUnitTest.class)
+public class TestProfilingDataSource {
+ private Configuration conf;
+ private Counter slowQuery;
+
+ @Before
+ public void init() {
+ conf = MetastoreConf.newMetastoreConf();
+ conf.set(MetastoreStatement.EXEC_HOOK,
MetastoreStatementTestHook.class.getName());
+ MetastoreConf.setTimeVar(conf,
MetastoreConf.ConfVars.METASTORE_JDBC_SLOW_QUERY_THRESHOLD, 200,
TimeUnit.MILLISECONDS);
+ MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.METRICS_ENABLED,
true);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_EXECUTION, true);
+ MetastoreConf.setVar(conf,
MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_THRIFT_APIS,
"test_metastore_statement");
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_USER_NAME,
"dummyUser");
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.PWD, "dummyPass");
+ conf.unset(MetastoreConf.ConfVars.CONNECTION_POOLING_TYPE.getVarname());
+
+ Metrics.initialize(conf);
+ slowQuery = Metrics.getOrCreateCounter(MetricsConstants.JDBC_SLOW_QUERIES);
+ }
+
+ @Test
+ public void testDefaultHikariCp() throws Exception {
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_POOLING_TYPE,
HikariCPDataSourceProvider.HIKARI);
+
+ DataSourceProvider dsp =
DataSourceProviderFactory.tryGetDataSourceProviderOrNull(conf);
+ Assert.assertNotNull(dsp);
+ DataSource ds = dsp.create(conf);
+ Assert.assertTrue(ds instanceof ProfilingDataSource);
+ Assert.assertTrue(((ProfilingDataSource) ds).getDelegate() instanceof
HikariDataSource);
+ try (Connection connection = ds.getConnection()) {
+ verify(connection);
+ }
+ }
+
+ @Test
+ public void testDbCpDataSource() throws Exception {
+ MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_POOLING_TYPE,
DbCPDataSourceProvider.DBCP);
+
+ DataSourceProvider dsp =
DataSourceProviderFactory.tryGetDataSourceProviderOrNull(conf);
+ Assert.assertNotNull(dsp);
+ DataSource ds = dsp.create(conf);
+ Assert.assertTrue(ds instanceof ProfilingDataSource);
+ Assert.assertTrue(((ProfilingDataSource) ds).getDelegate() instanceof
PoolingDataSource);
+ try (Connection connection = ds.getConnection()) {
+ verify(connection);
+ }
+ }
+
+ private void verify(Connection connection) throws Exception {
+ //
Assert.assertTrue(connection.unwrap(MetastoreConnection.class).delegate()
instanceof EmbedConnection);
+ long slowNum = slowQuery.getCount();
+ Timer timer =
Metrics.getOrCreateTimer(MetastoreStatementTestHook.TEST_METRIC_NAME);
+ Assert.assertNotNull(timer);
+ long timeCount = timer.getCount();
+ try (AutoCloseable sleep =
MetastoreStatementTestHook.testConnection("test_metastore_statement", 300,
connection)) {
+ try (Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery("VALUES 1")) {
+ Assert.assertTrue(rs.next());
+ }
+ HMSHandlerContext.getCallCtx().ifPresent(ctx -> {
+ Assert.assertEquals(1, ctx.getQueryCount());
+ Assert.assertTrue(ctx.getTotalTime() >= 300);
+ Assert.assertTrue(ctx.getMaxJdbcTimeMs() >= 300);
+ });
+ }
+ Assert.assertEquals(slowNum + 1, slowQuery.getCount());
+ Assert.assertEquals(timeCount + 1, timer.getCount());
+ Assert.assertTrue(timer.getSnapshot().getMean() >
TimeUnit.MILLISECONDS.toNanos(300));
+
+ try (AutoCloseable sleep =
MetastoreStatementTestHook.testConnection("test_statement_outside", 300,
connection)) {
+ try (Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery("VALUES 1")) {
+ Assert.assertTrue(rs.next());
+ }
+ }
+ Assert.assertEquals(slowNum + 2, slowQuery.getCount());
+ Assert.assertEquals(timeCount + 1, timer.getCount());
+ }
+
+ public static class MetastoreStatementTestHook extends
MetastoreStatement.ThriftApiProfiler {
+ static final String TEST_METRIC_NAME = "MetastoreStatementTestHook_" +
System.currentTimeMillis();
+ static final String SLEEP_MILLIS = "MetastoreStatementTestHook.sleep.ms";
+ private final Configuration configuration;
+
+ public MetastoreStatementTestHook(Configuration configuration) {
+ super(configuration);
+ this.configuration = configuration;
+ }
+
+ @Override
+ public void preRun(Method method, Object[] args) {
+ long sleepMs = configuration.getLong(SLEEP_MILLIS, 0);
+ if (sleepMs > 0 &&
+ MetastoreStatement.QUERY_EXECUTION.contains(method.getName())) {
+ try {
+ Thread.sleep(sleepMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ @Override
+ public String getMetricName(Method method) {
+ return TEST_METRIC_NAME;
+ }
+
+ public static AutoCloseable testConnection(String method, long sleepMs,
Connection connection)
+ throws Exception {
+ HMSHandlerContext.setCallCtx(new HMSHandlerContext.CallCtx(method));
+ Configuration configuration =
connection.unwrap(MetastoreConnection.class).getConfiguration();
+ configuration.set(SLEEP_MILLIS, sleepMs + "");
+ return () -> {
+ HMSHandlerContext.setCallCtx(null);
+ configuration.unset(SLEEP_MILLIS);
+ };
+ }
+ }
+}
