saihemanth-cloudera commented on code in PR #6464: URL: https://github.com/apache/hive/pull/6464#discussion_r3306338295
########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreDriver.java: ########## @@ -0,0 +1,177 @@ +/* + * 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.google.common.annotations.VisibleForTesting; + +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.DriverPropertyInfo; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.util.ArrayList; +import java.util.Enumeration; +import java.util.List; +import java.util.Properties; +import java.util.logging.Logger; +import java.util.regex.Pattern; + +import org.apache.commons.lang3.math.NumberUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.utils.MetastoreVersionInfo; +import org.slf4j.LoggerFactory; + +import static java.sql.DriverManager.registerDriver; + +public class MetastoreDriver implements Driver { + private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(MetastoreDriver.class); + private static final String URL_PREFIX = "jdbc:metastore://"; + private static final Configuration configuration; + private static int majorVersion = -1; + private static int minorVersion = -1; + private static volatile Driver delegateDriver; + static { + try { + registerDriver(new MetastoreDriver()); + String versionString = MetastoreVersionInfo.getVersion(); + String[] versionNums = versionString.split("\\."); + if (NumberUtils.isNumber(versionNums[0])) { + majorVersion = Integer.parseInt(versionNums[0]); + } + if (versionNums.length >1 && NumberUtils.isNumber(versionNums[1])) { + minorVersion = Integer.parseInt(versionNums[1]); + } + configuration = MetastoreConf.newMetastoreConf(); + } catch (Exception e) { + throw new RuntimeException("Failed to register Metastore driver", e); + } + } + + private synchronized static Driver + findRegisteredDriver(String jdbcUrl, String driverClassName) throws SQLException { + if (delegateDriver != null && delegateDriver.acceptsURL(jdbcUrl)) { + // Use the cached driver + return delegateDriver; + } + List<Driver> candidates = new ArrayList<>(); + for (Enumeration<Driver> drivers = DriverManager.getDrivers(); drivers.hasMoreElements();) { + Driver driver = drivers.nextElement(); + try { + if (driver.acceptsURL(jdbcUrl)) { + candidates.add(driver); + } + } catch (Exception e) { + } + } + + if (candidates.isEmpty()) { + Class<Driver> driverClz = tryLoadDriver(driverClassName, Thread.currentThread().getContextClassLoader(), Review Comment: Should we check 'driverClassName' for null or move this into try block? ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreDriver.java: ########## @@ -0,0 +1,177 @@ +/* + * 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.google.common.annotations.VisibleForTesting; + +import java.sql.Connection; +import java.sql.Driver; +import java.sql.DriverManager; +import java.sql.DriverPropertyInfo; +import java.sql.SQLException; +import java.sql.SQLFeatureNotSupportedException; +import java.util.ArrayList; +import java.util.Enumeration; +import java.util.List; +import java.util.Properties; +import java.util.logging.Logger; +import java.util.regex.Pattern; + +import org.apache.commons.lang3.math.NumberUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.utils.MetastoreVersionInfo; +import org.slf4j.LoggerFactory; + +import static java.sql.DriverManager.registerDriver; + +public class MetastoreDriver implements Driver { + private static final org.slf4j.Logger LOG = LoggerFactory.getLogger(MetastoreDriver.class); + private static final String URL_PREFIX = "jdbc:metastore://"; + private static final Configuration configuration; + private static int majorVersion = -1; + private static int minorVersion = -1; + private static volatile Driver delegateDriver; + static { + try { + registerDriver(new MetastoreDriver()); + String versionString = MetastoreVersionInfo.getVersion(); + String[] versionNums = versionString.split("\\."); + if (NumberUtils.isNumber(versionNums[0])) { + majorVersion = Integer.parseInt(versionNums[0]); + } + if (versionNums.length >1 && NumberUtils.isNumber(versionNums[1])) { + minorVersion = Integer.parseInt(versionNums[1]); + } + configuration = MetastoreConf.newMetastoreConf(); + } catch (Exception e) { + throw new RuntimeException("Failed to register Metastore driver", e); + } + } + + private synchronized static Driver + findRegisteredDriver(String jdbcUrl, String driverClassName) throws SQLException { + if (delegateDriver != null && delegateDriver.acceptsURL(jdbcUrl)) { + // Use the cached driver + return delegateDriver; + } + List<Driver> candidates = new ArrayList<>(); + for (Enumeration<Driver> drivers = DriverManager.getDrivers(); drivers.hasMoreElements();) { + Driver driver = drivers.nextElement(); + try { + if (driver.acceptsURL(jdbcUrl)) { + candidates.add(driver); + } + } catch (Exception e) { + } + } + + if (candidates.isEmpty()) { + Class<Driver> driverClz = tryLoadDriver(driverClassName, Thread.currentThread().getContextClassLoader(), + MetastoreDriver.class.getClassLoader()); + if (driverClz != null) { + try { + Driver driver = driverClz.getDeclaredConstructor().newInstance(); + if (!driver.acceptsURL(jdbcUrl)) { + throw new RuntimeException("Driver " + driverClassName + " cannot accept jdbcUrl"); + } + candidates.add(driver); + } catch (Exception e) { + LOG.warn("Failed to create instance of driver class {}", driverClassName, e); Review Comment: I think we should also throw a run time exception when we fail to create/register driver class instead of warning. ########## standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java: ########## @@ -648,6 +648,13 @@ 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_QUERIES("metastore.jdbc.execution.logSlowQueriesThreshold", "metastore.jdbc.execution.logSlowQueriesThreshold", + 3000, "Log the slow jdbc query that Metastore has been waiting for the result beyond the threshold(ms), " + Review Comment: nit: Also provide "TimeUnit.MILLISECONDS" time unit in the config. ########## standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java: ########## @@ -648,6 +648,13 @@ 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_QUERIES("metastore.jdbc.execution.logSlowQueriesThreshold", "metastore.jdbc.execution.logSlowQueriesThreshold", Review Comment: ```suggestion METASTORE_JDBC_SLOW_QUERIES("metastore.jdbc.execution.logSlowQueriesThreshold", "hive.metastore.jdbc.execution.logSlowQueriesThreshold", ``` Same with other configs ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java: ########## @@ -0,0 +1,263 @@ +/* + * 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.MetricRegistry; +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 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; + +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.logSlowExecution; +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.isSlowExecution; + +@SuppressWarnings("unchecked") +public final class MetastoreStatement implements InvocationHandler { + private static final Logger LOG = LoggerFactory.getLogger(MetastoreStatement.class); + private static final ThreadLocal<HMSHandlerContext.CallCtx> CALL_CTX = new ThreadLocal<>(); + static final String EXEC_HOOK = "metastore.jdbc.execution.hook"; + + private final String rawSql; + private final Statement delegate; + private final Configuration configuration; + private final MetastoreStatementHook hook; + + private MetastoreStatement(Configuration conf, Statement statement, String rawSql) { + this.configuration = Objects.requireNonNull(conf); + this.rawSql = rawSql; + this.delegate = Objects.requireNonNull(statement); + String className = conf.get(EXEC_HOOK, ""); + if (StringUtils.isEmpty(className)) { + hook = new JdbcProfilerUtils(conf); + } else { + try { + hook = 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); + } + + private void logSummary(boolean monitor) { + Optional<HMSHandlerContext.CallCtx> ctxCall = HMSHandlerContext.getCallCtx(); + HMSHandlerContext.CallCtx previousCall = CALL_CTX.get(); + if (ctxCall.isPresent()) { + if (previousCall == null) { + if (monitor) { + CALL_CTX.set(ctxCall.get()); + } + } else if (!ctxCall.get().equals(previousCall)) { Review Comment: ```suggestion } else if (previousCall != null && !ctxCall.get().equals(previousCall)) { ``` ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.java: ########## @@ -77,13 +85,54 @@ static String getMetastoreJdbcPasswd(Configuration conf) throws SQLException { } static String getMetastoreJdbcDriverUrl(Configuration conf) throws SQLException { + if (MetastoreConf.getBoolVar(conf, MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_EXECUTION)) { + return MetastoreDriver.getMetastoreDbUrl(conf); + } return MetastoreConf.getVar(conf, MetastoreConf.ConfVars.CONNECT_URL_KEY); } + static void addJdbcWrapperProperties(Configuration configuration, Properties properties) + throws SQLException { + if (MetastoreConf.getBoolVar(configuration, MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_EXECUTION)) { + try { + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + configuration.writeXml(outputStream); + properties.put(CONF_PROPERTY, Base64.getEncoder().encodeToString(outputStream.toByteArray())); Review Comment: By converting the entire config map into a Base64 string and attaching it to the JDBC Connection Properties, some sensitive info in hadoop config are exposed to the JDBC driver layer. I think, we should create a targeted whitelist. We should instantiate a clean, empty Configuration object, copy only the specific configuration parameters that start with `metastore.jdbc.` or `metastore.profile.`, and serialize that stripped-down config. ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java: ########## @@ -0,0 +1,263 @@ +/* + * 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.MetricRegistry; +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 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; + +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.logSlowExecution; +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.isSlowExecution; + +@SuppressWarnings("unchecked") +public final class MetastoreStatement implements InvocationHandler { + private static final Logger LOG = LoggerFactory.getLogger(MetastoreStatement.class); + private static final ThreadLocal<HMSHandlerContext.CallCtx> CALL_CTX = new ThreadLocal<>(); + static final String EXEC_HOOK = "metastore.jdbc.execution.hook"; + + private final String rawSql; + private final Statement delegate; + private final Configuration configuration; + private final MetastoreStatementHook hook; + + private MetastoreStatement(Configuration conf, Statement statement, String rawSql) { + this.configuration = Objects.requireNonNull(conf); + this.rawSql = rawSql; + this.delegate = Objects.requireNonNull(statement); + String className = conf.get(EXEC_HOOK, ""); + if (StringUtils.isEmpty(className)) { + hook = new JdbcProfilerUtils(conf); + } else { + try { + hook = 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); + } + + private void logSummary(boolean monitor) { + Optional<HMSHandlerContext.CallCtx> ctxCall = HMSHandlerContext.getCallCtx(); + HMSHandlerContext.CallCtx previousCall = CALL_CTX.get(); + if (ctxCall.isPresent()) { + if (previousCall == null) { + if (monitor) { + CALL_CTX.set(ctxCall.get()); + } + } else if (!ctxCall.get().equals(previousCall)) { + // we approach the end of previous thrift call + long totalSpent = previousCall.totalTime().get(); + LOG.debug("{} took {} ms to complete all queries", previousCall.methodName(), totalSpent); + if (isSlowExecution(configuration, totalSpent)) { + LOG.warn("{} took {} ms to complete all queries", previousCall.methodName(), totalSpent); + } + if (monitor) { + CALL_CTX.set(ctxCall.get()); + } else { + CALL_CTX.remove(); + } + } + } + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + Timer.Context ctx = null; + try { + boolean monitor = hook.profile(rawSql, method, args); + logSummary(monitor); + if (Metrics.getRegistry() != null && monitor) { + String metricName = hook.getMetricName(method, args); + 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 (CALL_CTX.get() != null) { + CALL_CTX.get().totalTime().addAndGet(timeSpent); + } + } + logSlowExecution(timeSpent, configuration, rawSql, method, args); + return result; + } catch (InvocationTargetException | UndeclaredThrowableException e) { + throw e.getCause(); + } finally { + if (ctx != null) { + ctx.stop(); + } + } + } + + 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 JdbcProfilerUtils} 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, Object[] args); + /** + * Invoked before the method call + * @param method Method which is being called + * @param args The method input + */ + default void preRun(Method method, Object[] args) { + + } + + /** + * Invoked post the method call + * @param method Method which is being called + * @param args The method input + * @param result The execution result from the call + */ + default void postRun(Method method, Object[] args, Object result) { + + } + } + + /** + * This class is used to profile the underlying statement originated from specific thrift API calls + */ + public static class JdbcProfilerUtils implements MetastoreStatement.MetastoreStatementHook { + private static final Set<String> PROFILED_APIS = new HashSet<>(); + static final Set<String> QUERY_EXECUTION = + Set.of("executeQuery", "executeUpdate", "execute", "executeBatch"); + private static volatile boolean initialized = false; + private static long logSlowQueriesThreshold; + + public JdbcProfilerUtils(Configuration configuration) { + initialize(Objects.requireNonNull(configuration)); + } + + private static void initialize(Configuration configuration) { + if (!initialized) { + synchronized (JdbcProfilerUtils.class) { + if (!initialized) { + initialized = true; + logSlowQueriesThreshold = MetastoreConf.getLongVar(configuration, + MetastoreConf.ConfVars.METASTORE_JDBC_SLOW_QUERIES); + if (logSlowQueriesThreshold > 0) { + LOG.info("The slow query log enabled, will log the query that takes more than {} ms", + logSlowQueriesThreshold); + } + String thriftApis = MetastoreConf.getVar(configuration, + MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_THRIFT_APIS); + for (String thriftApi : thriftApis.split(",")) { + String trimmedThriftApi = thriftApi.trim(); + if (!trimmedThriftApi.isEmpty()) { + PROFILED_APIS.add(trimmedThriftApi); + } + } + } + } + } + } + + public static void logSlowExecution(long timeSpent, Configuration configuration, + String sql, Method method, Object[] args) { + if (isSlowExecution(configuration, 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 : ""); + MetricRegistry registry = Metrics.getRegistry(); + if (registry != null) { + Counter slowQueries = Metrics.getOrCreateCounter(MetricsConstants.JDBC_SLOW_QUERIES); + slowQueries.inc(); + } + } + } + + public static boolean isSlowExecution(Configuration configuration, long timeSpent) { + initialize(configuration); + return logSlowQueriesThreshold > 0 && timeSpent > logSlowQueriesThreshold; + } + + @Override + public boolean profile(String sql, Method method, Object[] args) { + if (PROFILED_APIS.isEmpty() || !QUERY_EXECUTION.contains(method.getName())) { + // no api configured to profile + return false; + } + Optional<HMSHandlerContext.CallCtx> ctxCall = HMSHandlerContext.getCallCtx(); + if (ctxCall.isPresent()) { + String call = ctxCall.get().methodName(); + return PROFILED_APIS.contains(call); + } + return false; + } + + @Override + public String getMetricName(Method method, Object[] args) { Review Comment: nit: remove 'args' argument ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DbCPDataSourceProvider.java: ########## @@ -127,9 +120,21 @@ 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(); + Properties jdbcWrapperProperties = new Properties(); + DataSourceProvider.addJdbcWrapperProperties(hdpConfig, jdbcWrapperProperties); + jdbcWrapperProperties.forEach((key, value) -> + connectionProperties.append(key).append('=').append(value).append(';')); Review Comment: Converting a massive config to an XML byte array and then a Base64 string consumes significant heap space. Also, keeping this giant string permanently pinned in memory inside the connection pool properties array is wasteful. Addressing standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/DataSourceProvider.javaL#100 would also address this performance/memory concern. ########## standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/datasource/MetastoreStatement.java: ########## @@ -0,0 +1,263 @@ +/* + * 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.MetricRegistry; +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 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; + +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.logSlowExecution; +import static org.apache.hadoop.hive.metastore.datasource.MetastoreStatement.JdbcProfilerUtils.isSlowExecution; + +@SuppressWarnings("unchecked") +public final class MetastoreStatement implements InvocationHandler { + private static final Logger LOG = LoggerFactory.getLogger(MetastoreStatement.class); + private static final ThreadLocal<HMSHandlerContext.CallCtx> CALL_CTX = new ThreadLocal<>(); + static final String EXEC_HOOK = "metastore.jdbc.execution.hook"; + + private final String rawSql; + private final Statement delegate; + private final Configuration configuration; + private final MetastoreStatementHook hook; + + private MetastoreStatement(Configuration conf, Statement statement, String rawSql) { + this.configuration = Objects.requireNonNull(conf); + this.rawSql = rawSql; + this.delegate = Objects.requireNonNull(statement); + String className = conf.get(EXEC_HOOK, ""); + if (StringUtils.isEmpty(className)) { + hook = new JdbcProfilerUtils(conf); + } else { + try { + hook = 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); + } + + private void logSummary(boolean monitor) { + Optional<HMSHandlerContext.CallCtx> ctxCall = HMSHandlerContext.getCallCtx(); + HMSHandlerContext.CallCtx previousCall = CALL_CTX.get(); + if (ctxCall.isPresent()) { + if (previousCall == null) { + if (monitor) { + CALL_CTX.set(ctxCall.get()); + } + } else if (!ctxCall.get().equals(previousCall)) { + // we approach the end of previous thrift call + long totalSpent = previousCall.totalTime().get(); + LOG.debug("{} took {} ms to complete all queries", previousCall.methodName(), totalSpent); + if (isSlowExecution(configuration, totalSpent)) { + LOG.warn("{} took {} ms to complete all queries", previousCall.methodName(), totalSpent); + } + if (monitor) { + CALL_CTX.set(ctxCall.get()); + } else { + CALL_CTX.remove(); + } + } + } + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + Timer.Context ctx = null; + try { + boolean monitor = hook.profile(rawSql, method, args); + logSummary(monitor); + if (Metrics.getRegistry() != null && monitor) { + String metricName = hook.getMetricName(method, args); + 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 (CALL_CTX.get() != null) { + CALL_CTX.get().totalTime().addAndGet(timeSpent); + } + } + logSlowExecution(timeSpent, configuration, rawSql, method, args); + return result; + } catch (InvocationTargetException | UndeclaredThrowableException e) { + throw e.getCause(); + } finally { + if (ctx != null) { + ctx.stop(); + } + } + } + + 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 JdbcProfilerUtils} 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, Object[] args); + /** + * Invoked before the method call + * @param method Method which is being called + * @param args The method input + */ + default void preRun(Method method, Object[] args) { + + } + + /** + * Invoked post the method call + * @param method Method which is being called + * @param args The method input + * @param result The execution result from the call + */ + default void postRun(Method method, Object[] args, Object result) { + + } + } + + /** + * This class is used to profile the underlying statement originated from specific thrift API calls + */ + public static class JdbcProfilerUtils implements MetastoreStatement.MetastoreStatementHook { + private static final Set<String> PROFILED_APIS = new HashSet<>(); + static final Set<String> QUERY_EXECUTION = + Set.of("executeQuery", "executeUpdate", "execute", "executeBatch"); + private static volatile boolean initialized = false; + private static long logSlowQueriesThreshold; + + public JdbcProfilerUtils(Configuration configuration) { + initialize(Objects.requireNonNull(configuration)); + } + + private static void initialize(Configuration configuration) { + if (!initialized) { + synchronized (JdbcProfilerUtils.class) { + if (!initialized) { + initialized = true; + logSlowQueriesThreshold = MetastoreConf.getLongVar(configuration, + MetastoreConf.ConfVars.METASTORE_JDBC_SLOW_QUERIES); + if (logSlowQueriesThreshold > 0) { + LOG.info("The slow query log enabled, will log the query that takes more than {} ms", + logSlowQueriesThreshold); + } + String thriftApis = MetastoreConf.getVar(configuration, + MetastoreConf.ConfVars.METASTORE_PROFILE_JDBC_THRIFT_APIS); + for (String thriftApi : thriftApis.split(",")) { + String trimmedThriftApi = thriftApi.trim(); + if (!trimmedThriftApi.isEmpty()) { + PROFILED_APIS.add(trimmedThriftApi); + } + } + } + } + } + } + + public static void logSlowExecution(long timeSpent, Configuration configuration, + String sql, Method method, Object[] args) { + if (isSlowExecution(configuration, 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 : ""); + MetricRegistry registry = Metrics.getRegistry(); + if (registry != null) { + Counter slowQueries = Metrics.getOrCreateCounter(MetricsConstants.JDBC_SLOW_QUERIES); + slowQueries.inc(); + } + } + } + + public static boolean isSlowExecution(Configuration configuration, long timeSpent) { + initialize(configuration); + return logSlowQueriesThreshold > 0 && timeSpent > logSlowQueriesThreshold; + } + + @Override + public boolean profile(String sql, Method method, Object[] args) { Review Comment: We are not using 'sql' argument, we can drop it from argument list. ########## standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java: ########## @@ -648,6 +648,13 @@ 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_QUERIES("metastore.jdbc.execution.logSlowQueriesThreshold", "metastore.jdbc.execution.logSlowQueriesThreshold", + 3000, "Log the slow jdbc query that Metastore has been waiting for the result beyond the threshold(ms), " + + "should enable the metastore.profile.jdbc.execution first"), + METASTORE_PROFILE_JDBC_EXECUTION("metastore.profile.jdbc.execution", "metastore.profile.jdbc.execution", true, Review Comment: Shouldn't the default value be false? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
