This is an automated email from the ASF dual-hosted git repository. morningman pushed a commit to branch branch-1.2-lts in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6efb9a9c0d7d49e84972125d8ca4cb266e7d7a68 Author: Tiewei Fang <[email protected]> AuthorDate: Tue Mar 7 08:51:10 2023 +0800 [enhencement](jdbc catalog) Use Druid instead of HikariCP in JdbcClient (#17395) This pr does three things: 1. Use Druid instead of HikariCP in JdbcClient 2. when download udf jar, add the name of the jar package after the local file name. 3. refactor some jdbcResource code --- be/src/runtime/user_function_cache.cpp | 24 +++++-- be/src/runtime/user_function_cache.h | 4 +- be/src/vec/exec/vjdbc_connector.cpp | 1 + fe/fe-core/pom.xml | 5 +- .../org/apache/doris/catalog/JdbcResource.java | 81 ++++++++-------------- .../java/org/apache/doris/catalog/OdbcTable.java | 22 ++---- .../doris/catalog/external/JdbcExternalTable.java | 1 + .../org/apache/doris/external/jdbc/JdbcClient.java | 29 ++++---- fe/java-udf/pom.xml | 3 +- fe/pom.xml | 8 +-- 10 files changed, 82 insertions(+), 96 deletions(-) diff --git a/be/src/runtime/user_function_cache.cpp b/be/src/runtime/user_function_cache.cpp index ff2f575bc2..693a4672de 100644 --- a/be/src/runtime/user_function_cache.cpp +++ b/be/src/runtime/user_function_cache.cpp @@ -252,14 +252,15 @@ Status UserFunctionCache::_get_cache_entry(int64_t fid, const std::string& url, const std::string& checksum, UserFunctionCacheEntry** output_entry, LibType type) { UserFunctionCacheEntry* entry = nullptr; + std::string file_name = _get_file_name_from_url(url); { std::lock_guard<std::mutex> l(_cache_lock); auto it = _entry_map.find(fid); if (it != _entry_map.end()) { entry = it->second; } else { - entry = new UserFunctionCacheEntry(fid, checksum, _make_lib_file(fid, checksum, type), - type); + entry = new UserFunctionCacheEntry( + fid, checksum, _make_lib_file(fid, checksum, type, file_name), type); entry->ref(); _entry_map.emplace(fid, entry); } @@ -376,6 +377,17 @@ std::string UserFunctionCache::_get_real_url(const std::string& url) { return url; } +std::string UserFunctionCache::_get_file_name_from_url(const std::string& url) const { + std::string file_name; + size_t last_slash_pos = url.find_last_of('/'); + if (last_slash_pos != std::string::npos) { + file_name = url.substr(last_slash_pos + 1, url.size()); + } else { + file_name = url; + } + return file_name; +} + // entry's lock must be held Status UserFunctionCache::_load_cache_entry_internal(UserFunctionCacheEntry* entry) { RETURN_IF_ERROR(dynamic_open(entry->lib_file.c_str(), &entry->lib_handle)); @@ -384,12 +396,12 @@ Status UserFunctionCache::_load_cache_entry_internal(UserFunctionCacheEntry* ent } std::string UserFunctionCache::_make_lib_file(int64_t function_id, const std::string& checksum, - LibType type) { + LibType type, const std::string& file_name) { int shard = function_id % kLibShardNum; std::stringstream ss; ss << _lib_dir << '/' << shard << '/' << function_id << '.' << checksum; if (type == LibType::JAR) { - ss << ".jar"; + ss << '.' << file_name; } else { ss << ".so"; } @@ -417,6 +429,7 @@ Status UserFunctionCache::check_jar(int64_t fid, const std::string& url, const std::string& checksum) { UserFunctionCacheEntry* entry = nullptr; Status st = Status::OK(); + std::string file_name = _get_file_name_from_url(url); { std::lock_guard<std::mutex> l(_cache_lock); auto it = _entry_map.find(fid); @@ -424,7 +437,8 @@ Status UserFunctionCache::check_jar(int64_t fid, const std::string& url, entry = it->second; } else { entry = new UserFunctionCacheEntry( - fid, checksum, _make_lib_file(fid, checksum, LibType::JAR), LibType::JAR); + fid, checksum, _make_lib_file(fid, checksum, LibType::JAR, file_name), + LibType::JAR); entry->ref(); _entry_map.emplace(fid, entry); } diff --git a/be/src/runtime/user_function_cache.h b/be/src/runtime/user_function_cache.h index 59f0a1deb9..b12edab9e8 100644 --- a/be/src/runtime/user_function_cache.h +++ b/be/src/runtime/user_function_cache.h @@ -80,10 +80,12 @@ private: Status _download_lib(const std::string& url, UserFunctionCacheEntry* entry); Status _load_cache_entry_internal(UserFunctionCacheEntry* entry); - std::string _make_lib_file(int64_t function_id, const std::string& checksum, LibType type); + std::string _make_lib_file(int64_t function_id, const std::string& checksum, LibType type, + const std::string& file_name); void _destroy_cache_entry(UserFunctionCacheEntry* entry); std::string _get_real_url(const std::string& url); + std::string _get_file_name_from_url(const std::string& url) const; private: std::string _lib_dir; diff --git a/be/src/vec/exec/vjdbc_connector.cpp b/be/src/vec/exec/vjdbc_connector.cpp index 5ded70ed2f..e9caf93647 100644 --- a/be/src/vec/exec/vjdbc_connector.cpp +++ b/be/src/vec/exec/vjdbc_connector.cpp @@ -140,6 +140,7 @@ Status JdbcConnector::open(RuntimeState* state, bool read) { std::abs((int64_t)hash_str(_conn_param.resource_name)), _conn_param.driver_path, _conn_param.driver_checksum, &local_location)); } + VLOG_QUERY << "driver local path = " << local_location; TJdbcExecutorCtorParams ctor_params; ctor_params.__set_statement(_sql_str); diff --git a/fe/fe-core/pom.xml b/fe/fe-core/pom.xml index 16b9e8db23..0f69faa571 100644 --- a/fe/fe-core/pom.xml +++ b/fe/fe-core/pom.xml @@ -767,9 +767,10 @@ under the License. <version>3.10.1</version> </dependency> + <!-- https://mvnrepository.com/artifact/com.alibaba/druid --> <dependency> - <groupId>com.zaxxer</groupId> - <artifactId>HikariCP</artifactId> + <groupId>com.alibaba</groupId> + <artifactId>druid</artifactId> </dependency> <!-- for aliyun dlf --> diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java index ce9baf316c..1cede92b9f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java @@ -26,6 +26,7 @@ import org.apache.doris.common.proc.BaseProcResult; import org.apache.doris.common.util.Util; import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.gson.annotations.SerializedName; @@ -39,7 +40,6 @@ import java.net.URI; import java.net.URISyntaxException; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; -import java.util.List; import java.util.Map; @@ -84,13 +84,23 @@ public class JdbcResource extends Resource { public static final String ONLY_SPECIFIED_DATABASE = "only_specified_database"; public static final String LOWER_CASE_TABLE_NAMES = "lower_case_table_names"; public static final String CHECK_SUM = "checksum"; - - private static final List<String> OPTIONAL_PROPERTIES = Lists.newArrayList( + private static final ImmutableList<String> ALL_PROPERTIES = new ImmutableList.Builder<String>().add( + JDBC_URL, + USER, + PASSWORD, + DRIVER_CLASS, + DRIVER_URL, + TYPE, + ONLY_SPECIFIED_DATABASE, + LOWER_CASE_TABLE_NAMES + ).build(); + private static final ImmutableList<String> OPTIONAL_PROPERTIES = new ImmutableList.Builder<String>().add( ONLY_SPECIFIED_DATABASE, LOWER_CASE_TABLE_NAMES - ); + ).build(); // The default value of optional properties + // if one optional property is not specified, will use default value private static final Map<String, String> OPTIONAL_PROPERTIES_DEFAULT_VALUE = Maps.newHashMap(); static { @@ -116,29 +126,12 @@ public class JdbcResource extends Resource { this.configs = configs; } - public JdbcResource getCopiedResource() { - return new JdbcResource(name, Maps.newHashMap(configs)); - } - - private void checkProperties(String propertiesKey) throws DdlException { - // check the properties key - String value = configs.get(propertiesKey); - if (value == null) { - throw new DdlException("JdbcResource Missing " + propertiesKey + " in properties"); - } - } - @Override public void modifyProperties(Map<String, String> properties) throws DdlException { // modify properties - replaceIfEffectiveValue(this.configs, DRIVER_URL, properties.get(DRIVER_URL)); - replaceIfEffectiveValue(this.configs, DRIVER_CLASS, properties.get(DRIVER_CLASS)); - replaceIfEffectiveValue(this.configs, JDBC_URL, properties.get(JDBC_URL)); - replaceIfEffectiveValue(this.configs, USER, properties.get(USER)); - replaceIfEffectiveValue(this.configs, PASSWORD, properties.get(PASSWORD)); - replaceIfEffectiveValue(this.configs, TYPE, properties.get(TYPE)); - replaceIfEffectiveValue(this.configs, ONLY_SPECIFIED_DATABASE, properties.get(ONLY_SPECIFIED_DATABASE)); - replaceIfEffectiveValue(this.configs, LOWER_CASE_TABLE_NAMES, properties.get(LOWER_CASE_TABLE_NAMES)); + for (String propertyKey : ALL_PROPERTIES) { + replaceIfEffectiveValue(this.configs, propertyKey, properties.get(propertyKey)); + } this.configs.put(JDBC_URL, handleJdbcUrl(getProperty(JDBC_URL))); super.modifyProperties(properties); } @@ -147,14 +140,9 @@ public class JdbcResource extends Resource { public void checkProperties(Map<String, String> properties) throws AnalysisException { Map<String, String> copiedProperties = Maps.newHashMap(properties); // check properties - copiedProperties.remove(DRIVER_URL); - copiedProperties.remove(DRIVER_CLASS); - copiedProperties.remove(JDBC_URL); - copiedProperties.remove(USER); - copiedProperties.remove(PASSWORD); - copiedProperties.remove(TYPE); - copiedProperties.remove(ONLY_SPECIFIED_DATABASE); - copiedProperties.remove(LOWER_CASE_TABLE_NAMES); + for (String propertyKey : ALL_PROPERTIES) { + copiedProperties.remove(propertyKey); + } if (!copiedProperties.isEmpty()) { throw new AnalysisException("Unknown JDBC catalog resource properties: " + copiedProperties); } @@ -164,30 +152,19 @@ public class JdbcResource extends Resource { protected void setProperties(Map<String, String> properties) throws DdlException { Preconditions.checkState(properties != null); for (String key : properties.keySet()) { - switch (key) { - case DRIVER_URL: - case JDBC_URL: - case USER: - case PASSWORD: - case TYPE: - case DRIVER_CLASS: - case ONLY_SPECIFIED_DATABASE: // optional argument - case LOWER_CASE_TABLE_NAMES: // optional argument - break; - default: - throw new DdlException("JDBC resource Property of " + key + " is unknown"); + if (!ALL_PROPERTIES.contains(key)) { + throw new DdlException("JDBC resource Property of " + key + " is unknown"); } } configs = properties; handleOptionalArguments(); - checkProperties(DRIVER_URL); - checkProperties(DRIVER_CLASS); - checkProperties(JDBC_URL); - checkProperties(USER); - checkProperties(PASSWORD); - checkProperties(TYPE); - checkProperties(ONLY_SPECIFIED_DATABASE); - checkProperties(LOWER_CASE_TABLE_NAMES); + // check properties + for (String property : ALL_PROPERTIES) { + String value = configs.get(property); + if (value == null) { + throw new DdlException("JdbcResource Missing " + property + " in properties"); + } + } this.configs.put(JDBC_URL, handleJdbcUrl(getProperty(JDBC_URL))); configs.put(CHECK_SUM, computeObjectChecksum(getProperty(DRIVER_URL))); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OdbcTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OdbcTable.java index 5372c56312..b02939119a 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OdbcTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OdbcTable.java @@ -76,29 +76,15 @@ public class OdbcTable extends Table { private static String mysqlProperName(String name) { // In JdbcExternalTable, the name contains databaseName, like: db.table // So, we should split db and table, then switch to `db`.`table`. - String[] fields = name.split("\\."); - String result = ""; - for (int i = 0; i < fields.length; ++i) { - if (i != 0) { - result += "."; - } - result += ("`" + fields[i] + "`"); - } - return result; + List<String> list = Arrays.asList(name.split("\\.")); + return list.stream().map(s -> "`" + s + "`").collect(Collectors.joining(".")); } private static String mssqlProperName(String name) { // In JdbcExternalTable, the name contains databaseName, like: db.table // So, we should split db and table, then switch to [db].[table]. - String[] fields = name.split("\\."); - String result = ""; - for (int i = 0; i < fields.length; ++i) { - if (i != 0) { - result += "."; - } - result += ("[" + fields[i] + "]"); - } - return result; + List<String> list = Arrays.asList(name.split("\\.")); + return list.stream().map(s -> "[" + s + "]").collect(Collectors.joining(".")); } private static String psqlProperName(String name) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/JdbcExternalTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/JdbcExternalTable.java index cb7c51bb64..03c145da8c 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/external/JdbcExternalTable.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/external/JdbcExternalTable.java @@ -87,6 +87,7 @@ public class JdbcExternalTable extends ExternalTable { jdbcTable.setJdbcPasswd(jdbcCatalog.getJdbcPasswd()); jdbcTable.setDriverClass(jdbcCatalog.getDriverClass()); jdbcTable.setDriverUrl(jdbcCatalog.getDriverUrl()); + jdbcTable.setResourceName(jdbcCatalog.getResource()); jdbcTable.setCheckSum(jdbcCatalog.getCheckSum()); return jdbcTable; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java b/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java index 1196f67018..1b4f5bd076 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java +++ b/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java @@ -26,10 +26,9 @@ import org.apache.doris.common.Config; import org.apache.doris.common.DdlException; import org.apache.doris.common.util.Util; +import com.alibaba.druid.pool.DruidDataSource; import com.google.common.collect.Lists; import com.google.common.collect.Maps; -import com.zaxxer.hikari.HikariConfig; -import com.zaxxer.hikari.HikariDataSource; import lombok.Data; import lombok.Getter; import org.apache.logging.log4j.LogManager; @@ -57,8 +56,7 @@ public class JdbcClient { private URLClassLoader classLoader = null; - private HikariDataSource dataSource = null; - + private DruidDataSource dataSource = null; private boolean isOnlySpecifiedDatabase = false; private boolean isLowerCaseTableNames = false; @@ -82,21 +80,26 @@ public class JdbcClient { // and URLClassLoader may load the jar package directly into memory URL[] urls = {new URL(JdbcResource.getFullDriverUrl(driverUrl))}; // set parent ClassLoader to null, we can achieve class loading isolation. - classLoader = URLClassLoader.newInstance(urls, null); + ClassLoader parent = getClass().getClassLoader(); + ClassLoader classLoader = URLClassLoader.newInstance(urls, parent); + LOG.debug("parent ClassLoader: {}, old ClassLoader: {}, class Loader: {}.", + parent, oldClassLoader, classLoader); Thread.currentThread().setContextClassLoader(classLoader); - HikariConfig config = new HikariConfig(); - config.setDriverClassName(driverClass); - config.setJdbcUrl(jdbcUrl); - config.setUsername(jdbcUser); - config.setPassword(password); - config.setMaximumPoolSize(1); + dataSource = new DruidDataSource(); + dataSource.setDriverClassLoader(classLoader); + dataSource.setDriverClassName(driverClass); + dataSource.setUrl(jdbcUrl); + dataSource.setUsername(jdbcUser); + dataSource.setPassword(password); + dataSource.setMinIdle(1); + dataSource.setInitialSize(2); + dataSource.setMaxActive(5); // set connection timeout to 5s. // The default is 30s, which is too long. // Because when querying information_schema db, BE will call thrift rpc(default timeout is 30s) // to FE to get schema info, and may create connection here, if we set it too long and the url is invalid, // it may cause the thrift rpc timeout. - config.setConnectionTimeout(5000); - dataSource = new HikariDataSource(config); + dataSource.setMaxWait(5000); } catch (MalformedURLException e) { throw new JdbcClientException("MalformedURLException to load class about " + driverUrl, e); } finally { diff --git a/fe/java-udf/pom.xml b/fe/java-udf/pom.xml index f8d164d4c9..163fa5c0ae 100644 --- a/fe/java-udf/pom.xml +++ b/fe/java-udf/pom.xml @@ -90,12 +90,13 @@ under the License. <artifactId>ojdbc6</artifactId> <version>11.2.0.4</version> </dependency> + <!-- https://mvnrepository.com/artifact/com.alibaba/druid --> <dependency> <groupId>com.alibaba</groupId> <artifactId>druid</artifactId> - <version>1.2.5</version> </dependency> + <!-- https://mvnrepository.com/artifact/org.junit.jupiter/junit-jupiter-engine --> <dependency> <groupId>org.junit.jupiter</groupId> diff --git a/fe/pom.xml b/fe/pom.xml index e715cf3c9a..c5015f4ee9 100644 --- a/fe/pom.xml +++ b/fe/pom.xml @@ -187,7 +187,7 @@ under the License. <commons-io.version>2.6</commons-io.version> <json-simple.version>1.1.1</json-simple.version> <junit.version>5.8.2</junit.version> - <hikaricp.version>3.4.5</hikaricp.version> + <druid.version>1.2.5</druid.version> <thrift.version>0.13.0</thrift.version> <log4j2.version>2.18.0</log4j2.version> <metrics-core.version>4.0.2</metrics-core.version> @@ -990,9 +990,9 @@ under the License. <version>${dlf-metastore-client-hive.version}</version> </dependency> <dependency> - <groupId>com.zaxxer</groupId> - <artifactId>HikariCP</artifactId> - <version>${hikaricp.version}</version> + <groupId>com.alibaba</groupId> + <artifactId>druid</artifactId> + <version>${druid.version}</version> </dependency> <!-- https://mvnrepository.com/artifact/joda-time/joda-time --> <dependency> --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
