From e3f39cb7cf111039c893f8df8178d3ac1de4eaa4 Mon Sep 17 00:00:00 2001 From: Artem Gavrilov <37815717+Artem3213212@users.noreply.github.com> Date: Thu, 28 Aug 2025 10:00:43 +0300 Subject: [PATCH 1/5] hide .pgpass values (#9) * hide .pgpass values --- .../pxf/plugins/jdbc/JdbcBasePlugin.java | 1227 +++++++++-------- .../pxf/plugins/jdbc/JdbcAccessorTest.java | 2 + .../pxf/plugins/jdbc/JdbcBasePluginTest.java | 140 +- .../jdbc/JdbcBasePluginTestInitialize.java | 11 + 4 files changed, 776 insertions(+), 604 deletions(-) diff --git a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java index 6a500a1ac..1ded13ff9 100644 --- a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java +++ b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java @@ -1,598 +1,629 @@ -package org.apache.cloudberry.pxf.plugins.jdbc; - -/* - * 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. - */ - -import org.apache.commons.lang.StringUtils; -import org.apache.hadoop.conf.Configuration; -import org.apache.cloudberry.pxf.api.model.BasePlugin; -import org.apache.cloudberry.pxf.api.model.RequestContext; -import org.apache.cloudberry.pxf.api.security.SecureLogin; -import org.apache.cloudberry.pxf.api.utilities.ColumnDescriptor; -import org.apache.cloudberry.pxf.api.utilities.SpringContext; -import org.apache.cloudberry.pxf.api.utilities.Utilities; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.DbProduct; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.HiveJdbcUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.security.PrivilegedExceptionAction; -import java.sql.Connection; -import java.sql.DatabaseMetaData; -import java.sql.PreparedStatement; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.stream.Collectors; - -import static org.apache.cloudberry.pxf.api.security.SecureLogin.CONFIG_KEY_SERVICE_USER_IMPERSONATION; - -/** - * JDBC tables plugin (base class) - *

- * Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}. - */ -public class JdbcBasePlugin extends BasePlugin { - - private static final Logger LOG = LoggerFactory.getLogger(JdbcBasePlugin.class); - - // '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754 - private static final int DEFAULT_BATCH_SIZE = 100; - private static final int DEFAULT_FETCH_SIZE = 1000; - // MySQL fetches all data in memory first unless streaming is enabled by setting fetchSize to Integer.MIN_VALUE - // see https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-implementation-notes.html - private static final int DEFAULT_MYSQL_FETCH_SIZE = Integer.MIN_VALUE; - private static final int DEFAULT_POOL_SIZE = 1; - - // configuration parameter names - private static final String JDBC_DRIVER_PROPERTY_NAME = "jdbc.driver"; - private static final String JDBC_URL_PROPERTY_NAME = "jdbc.url"; - private static final String JDBC_USER_PROPERTY_NAME = "jdbc.user"; - private static final String JDBC_PASSWORD_PROPERTY_NAME = "jdbc.password"; - private static final String JDBC_SESSION_PROPERTY_PREFIX = "jdbc.session.property."; - private static final String JDBC_CONNECTION_PROPERTY_PREFIX = "jdbc.connection.property."; - - // connection parameter names - private static final String JDBC_CONNECTION_TRANSACTION_ISOLATION = "jdbc.connection.transactionIsolation"; - - // statement properties - private static final String JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME = "jdbc.statement.batchSize"; - private static final String JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME = "jdbc.statement.fetchSize"; - private static final String JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME = "jdbc.statement.queryTimeout"; - - // connection pool properties - private static final String JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME = "jdbc.pool.enabled"; - private static final String JDBC_CONNECTION_POOL_PROPERTY_PREFIX = "jdbc.pool.property."; - private static final String JDBC_POOL_QUALIFIER_PROPERTY_NAME = "jdbc.pool.qualifier"; - - // DDL option names - private static final String JDBC_DRIVER_OPTION_NAME = "JDBC_DRIVER"; - private static final String JDBC_URL_OPTION_NAME = "DB_URL"; - - private static final String FORBIDDEN_SESSION_PROPERTY_CHARACTERS = ";\n\b\0"; - private static final String QUERY_NAME_PREFIX = "query:"; - private static final int QUERY_NAME_PREFIX_LENGTH = QUERY_NAME_PREFIX.length(); - - private static final String HIVE_URL_PREFIX = "jdbc:hive2://"; - private static final String HIVE_DEFAULT_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver"; - private static final String MYSQL_DRIVER_PREFIX = "com.mysql."; - private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange"; - - private enum TransactionIsolation { - READ_UNCOMMITTED(1), - READ_COMMITTED(2), - REPEATABLE_READ(4), - SERIALIZABLE(8), - NOT_PROVIDED(-1); - - private final int isolationLevel; - - TransactionIsolation(int transactionIsolation) { - isolationLevel = transactionIsolation; - } - - public int getLevel() { - return isolationLevel; - } - - public static TransactionIsolation typeOf(String str) { - return valueOf(str); - } - } - - // JDBC parameters from config file or specified in DDL - - private String jdbcUrl; - - protected String tableName; - - // Write batch size - protected int batchSize; - protected boolean batchSizeIsSetByUser = false; - - // Read batch size - protected int fetchSize; - - // Thread pool size - protected int poolSize; - - // Query timeout. - protected Integer queryTimeout; - - // Quote columns setting set by user (three values are possible) - protected Boolean quoteColumns = null; - - // Environment variables to SET before query execution - protected Map sessionConfiguration = new HashMap<>(); - - // Properties object to pass to JDBC Driver when connection is created - protected Properties connectionConfiguration = new Properties(); - - // Transaction isolation level that a user can configure - private TransactionIsolation transactionIsolation = TransactionIsolation.NOT_PROVIDED; - - // Columns description - protected List columns = null; - - // Name of query to execute for read flow (optional) - protected String queryName; - - // connection pool fields - private boolean isConnectionPoolUsed; - private Properties poolConfiguration; - private String poolQualifier; - - private final ConnectionManager connectionManager; - private final SecureLogin secureLogin; - - // Flag which is used when the year might contain more than 4 digits in `date` or 'timestamp' - protected boolean isDateWideRange; - - static { - // Deprecated as of Oct 22, 2019 in version 5.9.2+ - Configuration.addDeprecation("pxf.impersonation.jdbc", - CONFIG_KEY_SERVICE_USER_IMPERSONATION, - "The property \"pxf.impersonation.jdbc\" has been deprecated in favor of \"pxf.service.user.impersonation\"."); - } - - /** - * Creates a new instance with default (singleton) instances of - * ConnectionManager and SecureLogin. - */ - JdbcBasePlugin() { - this(SpringContext.getBean(ConnectionManager.class), SpringContext.getBean(SecureLogin.class)); - } - - /** - * Creates a new instance with the given ConnectionManager and ConfigurationFactory - * - * @param connectionManager connection manager instance - */ - JdbcBasePlugin(ConnectionManager connectionManager, SecureLogin secureLogin) { - this.connectionManager = connectionManager; - this.secureLogin = secureLogin; - } - - @Override - public void afterPropertiesSet() { - // Required parameter. Can be auto-overwritten by user options - String jdbcDriver = configuration.get(JDBC_DRIVER_PROPERTY_NAME); - assertMandatoryParameter(jdbcDriver, JDBC_DRIVER_PROPERTY_NAME, JDBC_DRIVER_OPTION_NAME); - try { - LOG.debug("JDBC driver: '{}'", jdbcDriver); - Class.forName(jdbcDriver); - } catch (ClassNotFoundException e) { - throw new RuntimeException(e); - } - - // Required parameter. Can be auto-overwritten by user options - jdbcUrl = configuration.get(JDBC_URL_PROPERTY_NAME); - assertMandatoryParameter(jdbcUrl, JDBC_URL_PROPERTY_NAME, JDBC_URL_OPTION_NAME); - - // Required metadata - String dataSource = context.getDataSource(); - if (StringUtils.isBlank(dataSource)) { - throw new IllegalArgumentException("Data source must be provided"); - } - - // Determine if the datasource is a table name or a query name - if (dataSource.startsWith(QUERY_NAME_PREFIX)) { - queryName = dataSource.substring(QUERY_NAME_PREFIX_LENGTH); - if (StringUtils.isBlank(queryName)) { - throw new IllegalArgumentException(String.format("Query name is not provided in data source [%s]", dataSource)); - } - LOG.debug("Query name is {}", queryName); - } else { - tableName = dataSource; - LOG.debug("Table name is {}", tableName); - } - - // Required metadata - columns = context.getTupleDescription(); - - // Optional parameters - batchSizeIsSetByUser = configuration.get(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME) != null; - if (context.getRequestType() == RequestContext.RequestType.WRITE_BRIDGE) { - batchSize = configuration.getInt(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, DEFAULT_BATCH_SIZE); - - if (batchSize == 0) { - batchSize = 1; // if user set to 0, it is the same as batchSize of 1 - } else if (batchSize < 0) { - throw new IllegalArgumentException(String.format( - "Property %s has incorrect value %s : must be a non-negative integer", JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, batchSize)); - } - } - - // determine fetchSize for read operations, with different default values for MySQL driver and all others - int defaultFetchSize = jdbcDriver.startsWith(MYSQL_DRIVER_PREFIX) ? DEFAULT_MYSQL_FETCH_SIZE : DEFAULT_FETCH_SIZE; - fetchSize = configuration.getInt(JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME, defaultFetchSize); - LOG.debug("Will be using fetchSize {}", fetchSize); - - poolSize = context.getOption("POOL_SIZE", DEFAULT_POOL_SIZE); - - String queryTimeoutString = configuration.get(JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME); - if (StringUtils.isNotBlank(queryTimeoutString)) { - try { - queryTimeout = Integer.parseUnsignedInt(queryTimeoutString); - } catch (NumberFormatException e) { - throw new IllegalArgumentException(String.format( - "Property %s has incorrect value %s : must be a non-negative integer", - JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME, queryTimeoutString), e); - } - } - - // Optional parameter. The default value is null - String quoteColumnsRaw = context.getOption("QUOTE_COLUMNS"); - if (quoteColumnsRaw != null) { - quoteColumns = Boolean.parseBoolean(quoteColumnsRaw); - } - - // Optional parameter. The default value is empty map - sessionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_SESSION_PROPERTY_PREFIX)); - // Check forbidden symbols - // Note: PreparedStatement enables us to skip this check: its values are distinct from its SQL code - // However, SET queries cannot be executed this way. This is why we do this check - if (sessionConfiguration.entrySet().stream() - .anyMatch( - entry -> - StringUtils.containsAny( - entry.getKey(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS - ) || - StringUtils.containsAny( - entry.getValue(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS - ) - ) - ) { - throw new IllegalArgumentException("Some session configuration parameter contains forbidden characters"); - } - if (LOG.isDebugEnabled()) { - LOG.debug("Session configuration: {}", - sessionConfiguration.entrySet().stream() - .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") - .collect(Collectors.joining(", ")) - ); - } - - // Optional parameter. The default value is empty map - connectionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_PROPERTY_PREFIX)); - - // Optional parameter. The default value depends on the database - String transactionIsolationString = configuration.get(JDBC_CONNECTION_TRANSACTION_ISOLATION, "NOT_PROVIDED"); - transactionIsolation = TransactionIsolation.typeOf(transactionIsolationString); - - // Set optional user parameter, taking into account impersonation setting for the server. - String jdbcUser = configuration.get(JDBC_USER_PROPERTY_NAME); - boolean impersonationEnabledForServer = configuration.getBoolean(CONFIG_KEY_SERVICE_USER_IMPERSONATION, false); - LOG.debug("JDBC impersonation is {}enabled for server {}", impersonationEnabledForServer ? "" : "not ", context.getServerName()); - if (impersonationEnabledForServer) { - if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { - // secure impersonation for Hive JDBC driver requires setting URL fragment that cannot be overwritten by properties - String updatedJdbcUrl = HiveJdbcUtils.updateImpersonationPropertyInHiveJdbcUrl(jdbcUrl, context.getUser()); - LOG.debug("Replaced JDBC URL {} with {}", jdbcUrl, updatedJdbcUrl); - jdbcUrl = updatedJdbcUrl; - } else { - // the jdbcUser is the GPDB user - jdbcUser = context.getUser(); - } - } - if (jdbcUser != null) { - LOG.debug("Effective JDBC user {}", jdbcUser); - connectionConfiguration.setProperty("user", jdbcUser); - } else { - LOG.debug("JDBC user has not been set"); - } - - if (LOG.isDebugEnabled()) { - LOG.debug("Connection configuration: {}", - connectionConfiguration.entrySet().stream() - .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") - .collect(Collectors.joining(", ")) - ); - } - - // This must be the last parameter parsed, as we output connectionConfiguration earlier - // Optional parameter. By default, corresponding connectionConfiguration property is not set - if (jdbcUser != null) { - String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); - if (jdbcPassword != null) { - LOG.debug("Connection password: {}", ConnectionManager.maskPassword(jdbcPassword)); - connectionConfiguration.setProperty("password", jdbcPassword); - } - } - - // connection pool is optional, enabled by default - isConnectionPoolUsed = configuration.getBoolean(JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME, true); - LOG.debug("Connection pool is {}enabled", isConnectionPoolUsed ? "" : "not "); - if (isConnectionPoolUsed) { - poolConfiguration = new Properties(); - // for PXF upgrades where jdbc-site template has not been updated, make sure there're sensible defaults - poolConfiguration.setProperty("maximumPoolSize", "15"); - poolConfiguration.setProperty("connectionTimeout", "30000"); - poolConfiguration.setProperty("idleTimeout", "30000"); - poolConfiguration.setProperty("minimumIdle", "0"); - // apply values read from the template - poolConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_POOL_PROPERTY_PREFIX)); - - // packaged Hive JDBC Driver does not support connection.isValid() method, so we need to force set - // connectionTestQuery parameter in this case, unless already set by the user - if (jdbcUrl.startsWith(HIVE_URL_PREFIX) && HIVE_DEFAULT_DRIVER_CLASS.equals(jdbcDriver) && poolConfiguration.getProperty("connectionTestQuery") == null) { - poolConfiguration.setProperty("connectionTestQuery", "SELECT 1"); - } - - // get the qualifier for connection pool, if configured. Might be used when connection session authorization is employed - // to switch effective user once connection is established - poolQualifier = configuration.get(JDBC_POOL_QUALIFIER_PROPERTY_NAME); - } - - // Optional parameter to determine if the year might contain more than 4 digits in `date` or 'timestamp'. - // The default value is false. - isDateWideRange = configuration.getBoolean(JDBC_DATE_WIDE_RANGE, false); - } - - /** - * Open a new JDBC connection - * - * @return {@link Connection} - * @throws SQLException if a database access or connection error occurs - */ - public Connection getConnection() throws SQLException { - LOG.debug("Requesting a new JDBC connection. URL={} table={} txid:seg={}:{}", jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); - - Connection connection = null; - try { - connection = getConnectionInternal(); - LOG.debug("Obtained a JDBC connection {} for URL={} table={} txid:seg={}:{}", connection, jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); - - prepareConnection(connection); - } catch (Exception e) { - closeConnection(connection); - if (e instanceof SQLException) { - throw (SQLException) e; - } else { - String msg = e.getMessage(); - if (msg == null) { - Throwable t = e.getCause(); - if (t != null) msg = t.getMessage(); - } - throw new SQLException(msg, e); - } - } - - return connection; - } - - /** - * Prepare a JDBC PreparedStatement - * - * @param connection connection to use for creating the statement - * @param query query to execute - * @return PreparedStatement - * @throws SQLException if a database access error occurs - */ - public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException { - if ((connection == null) || (query == null)) { - throw new IllegalArgumentException("The provided query or connection is null"); - } - PreparedStatement statement = connection.prepareStatement(query); - if (queryTimeout != null) { - LOG.debug("Setting query timeout to {} seconds", queryTimeout); - statement.setQueryTimeout(queryTimeout); - } - return statement; - } - - /** - * Close a JDBC statement and underlying {@link Connection} - * - * @param statement statement to close - * @throws SQLException throws when a SQLException occurs - */ - public static void closeStatementAndConnection(Statement statement) throws SQLException { - if (statement == null) { - LOG.warn("Call to close statement and connection is ignored as statement provided was null"); - return; - } - - SQLException exception = null; - Connection connection = null; - - try { - connection = statement.getConnection(); - } catch (SQLException e) { - LOG.error("Exception when retrieving Connection from Statement", e); - exception = e; - } - - try { - LOG.debug("Closing statement for connection {}", connection); - statement.close(); - } catch (SQLException e) { - LOG.error("Exception when closing Statement", e); - exception = e; - } - - try { - closeConnection(connection); - } catch (SQLException e) { - LOG.error(String.format("Exception when closing connection %s", connection), e); - exception = e; - } - - if (exception != null) { - throw exception; - } - } - - /** - * For a Kerberized Hive JDBC connection, it creates a connection as the loginUser. - * Otherwise, it returns a new connection. - * - * @return for a Kerberized Hive JDBC connection, returns a new connection as the loginUser. - * Otherwise, it returns a new connection. - * @throws Exception throws when an error occurs - */ - private Connection getConnectionInternal() throws Exception { - Configuration configuration = context.getConfiguration(); - if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { - return secureLogin.getLoginUser(context, configuration).doAs((PrivilegedExceptionAction) () -> - connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier)); - } else { - return connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier); - } - } - - /** - * Close a JDBC connection - * - * @param connection connection to close - * @throws SQLException throws when a SQLException occurs - */ - protected static void closeConnection(Connection connection) throws SQLException { - if (connection == null) { - LOG.warn("Call to close connection is ignored as connection provided was null"); - return; - } - try { - if (!connection.isClosed() && - connection.getMetaData().supportsTransactions() && - !connection.getAutoCommit()) { - - LOG.debug("Committing transaction (as part of connection.close()) on connection {}", connection); - connection.commit(); - } - } finally { - try { - LOG.debug("Closing connection {}", connection); - connection.close(); - } catch (Exception e) { - // ignore - LOG.warn(String.format("Failed to close JDBC connection %s, ignoring the error.", connection), e); - } - } - } - - /** - * Prepare JDBC connection by setting session-level variables in external database - * - * @param connection {@link Connection} to prepare - */ - private void prepareConnection(Connection connection) throws SQLException { - if (connection == null) { - throw new IllegalArgumentException("The provided connection is null"); - } - - DatabaseMetaData metadata = connection.getMetaData(); - - // Handle optional connection transaction isolation level - if (transactionIsolation != TransactionIsolation.NOT_PROVIDED) { - // user wants to set isolation level explicitly - if (metadata.supportsTransactionIsolationLevel(transactionIsolation.getLevel())) { - LOG.debug("Setting transaction isolation level to {} on connection {}", transactionIsolation.toString(), connection); - connection.setTransactionIsolation(transactionIsolation.getLevel()); - } else { - throw new RuntimeException( - String.format("Transaction isolation level %s is not supported", transactionIsolation.toString()) - ); - } - } - - // Disable autocommit - if (metadata.supportsTransactions()) { - LOG.debug("Setting autoCommit to false on connection {}", connection); - connection.setAutoCommit(false); - } - - // Prepare session (process sessionConfiguration) - if (!sessionConfiguration.isEmpty()) { - DbProduct dbProduct = DbProduct.getDbProduct(metadata.getDatabaseProductName()); - - try (Statement statement = connection.createStatement()) { - for (Map.Entry e : sessionConfiguration.entrySet()) { - String sessionQuery = dbProduct.buildSessionQuery(e.getKey(), e.getValue()); - LOG.debug("Executing statement {} on connection {}", sessionQuery, connection); - statement.execute(sessionQuery); - } - } - } - } - - /** - * Asserts whether a given parameter has non-empty value, throws IllegalArgumentException otherwise - * - * @param value value to check - * @param paramName parameter name - * @param optionName name of the option for a given parameter - */ - private void assertMandatoryParameter(String value, String paramName, String optionName) { - if (StringUtils.isBlank(value)) { - throw new IllegalArgumentException(String.format( - "Required parameter %s is missing or empty in jdbc-site.xml and option %s is not specified in table definition.", paramName, optionName) - ); - } - } - - /** - * Constructs a mapping of configuration and includes all properties that start with the specified - * configuration prefix. Property names in the mapping are trimmed to remove the configuration prefix. - * This is a method from Hadoop's Configuration class ported here to make older and custom versions of Hadoop - * work with JDBC profile. - * - * @param configuration configuration map - * @param confPrefix configuration prefix - * @return mapping of configuration properties with prefix stripped - */ - private Map getPropsWithPrefix(Configuration configuration, String confPrefix) { - Map configMap = new HashMap<>(); - for (Map.Entry stringStringEntry : configuration) { - String propertyName = stringStringEntry.getKey(); - if (propertyName.startsWith(confPrefix)) { - // do not use value from the iterator as it might not come with variable substitution - String value = configuration.get(propertyName); - String keyName = propertyName.substring(confPrefix.length()); - configMap.put(keyName, value); - } - } - return configMap; - } - -} +package org.apache.cloudberry.pxf.plugins.jdbc; + +/* + * 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. + */ + +import org.apache.commons.lang.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.cloudberry.pxf.api.model.BasePlugin; +import org.apache.cloudberry.pxf.api.model.RequestContext; +import org.apache.cloudberry.pxf.api.security.SecureLogin; +import org.apache.cloudberry.pxf.api.utilities.ColumnDescriptor; +import org.apache.cloudberry.pxf.api.utilities.SpringContext; +import org.apache.cloudberry.pxf.api.utilities.Utilities; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.DbProduct; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.HiveJdbcUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.URI; +import java.net.URISyntaxException; +import java.security.PrivilegedExceptionAction; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + +import static org.apache.cloudberry.pxf.api.security.SecureLogin.CONFIG_KEY_SERVICE_USER_IMPERSONATION; + +/** + * JDBC tables plugin (base class) + *

+ * Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}. + */ +public class JdbcBasePlugin extends BasePlugin { + + private static final Logger LOG = LoggerFactory.getLogger(JdbcBasePlugin.class); + + // '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754 + private static final int DEFAULT_BATCH_SIZE = 100; + private static final int DEFAULT_FETCH_SIZE = 1000; + // MySQL fetches all data in memory first unless streaming is enabled by setting fetchSize to Integer.MIN_VALUE + // see https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-implementation-notes.html + private static final int DEFAULT_MYSQL_FETCH_SIZE = Integer.MIN_VALUE; + private static final int DEFAULT_POOL_SIZE = 1; + + // configuration parameter names + private static final String JDBC_DRIVER_PROPERTY_NAME = "jdbc.driver"; + private static final String JDBC_URL_PROPERTY_NAME = "jdbc.url"; + private static final String JDBC_USER_PROPERTY_NAME = "jdbc.user"; + private static final String JDBC_PASSWORD_PROPERTY_NAME = "jdbc.password"; + private static final String JDBC_SESSION_PROPERTY_PREFIX = "jdbc.session.property."; + private static final String JDBC_CONNECTION_PROPERTY_PREFIX = "jdbc.connection.property."; + + // connection parameter names + private static final String JDBC_CONNECTION_TRANSACTION_ISOLATION = "jdbc.connection.transactionIsolation"; + + // statement properties + private static final String JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME = "jdbc.statement.batchSize"; + private static final String JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME = "jdbc.statement.fetchSize"; + private static final String JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME = "jdbc.statement.queryTimeout"; + + // connection pool properties + private static final String JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME = "jdbc.pool.enabled"; + private static final String JDBC_CONNECTION_POOL_PROPERTY_PREFIX = "jdbc.pool.property."; + private static final String JDBC_POOL_QUALIFIER_PROPERTY_NAME = "jdbc.pool.qualifier"; + + // DDL option names + private static final String JDBC_DRIVER_OPTION_NAME = "JDBC_DRIVER"; + private static final String JDBC_URL_OPTION_NAME = "DB_URL"; + + private static final String FORBIDDEN_SESSION_PROPERTY_CHARACTERS = ";\n\b\0"; + private static final String QUERY_NAME_PREFIX = "query:"; + private static final int QUERY_NAME_PREFIX_LENGTH = QUERY_NAME_PREFIX.length(); + + private static final String HIVE_URL_PREFIX = "jdbc:hive2://"; + private static final String HIVE_DEFAULT_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver"; + private static final String MYSQL_DRIVER_PREFIX = "com.mysql."; + private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange"; + + private enum TransactionIsolation { + READ_UNCOMMITTED(1), + READ_COMMITTED(2), + REPEATABLE_READ(4), + SERIALIZABLE(8), + NOT_PROVIDED(-1); + + private final int isolationLevel; + + TransactionIsolation(int transactionIsolation) { + isolationLevel = transactionIsolation; + } + + public int getLevel() { + return isolationLevel; + } + + public static TransactionIsolation typeOf(String str) { + return valueOf(str); + } + } + + // JDBC parameters from config file or specified in DDL + + private String jdbcUrl; + + protected String tableName; + + // Write batch size + protected int batchSize; + protected boolean batchSizeIsSetByUser = false; + + // Read batch size + protected int fetchSize; + + // Thread pool size + protected int poolSize; + + // Query timeout. + protected Integer queryTimeout; + + // Quote columns setting set by user (three values are possible) + protected Boolean quoteColumns = null; + + // Environment variables to SET before query execution + protected Map sessionConfiguration = new HashMap<>(); + + // Properties object to pass to JDBC Driver when connection is created + protected Properties connectionConfiguration = new Properties(); + + // Transaction isolation level that a user can configure + private TransactionIsolation transactionIsolation = TransactionIsolation.NOT_PROVIDED; + + // Columns description + protected List columns = null; + + // Name of query to execute for read flow (optional) + protected String queryName; + + // connection pool fields + private boolean isConnectionPoolUsed; + private Properties poolConfiguration; + private String poolQualifier; + + private final ConnectionManager connectionManager; + private final SecureLogin secureLogin; + + // Flag which is used when the year might contain more than 4 digits in `date` or 'timestamp' + protected boolean isDateWideRange; + + static { + // Deprecated as of Oct 22, 2019 in version 5.9.2+ + Configuration.addDeprecation("pxf.impersonation.jdbc", + CONFIG_KEY_SERVICE_USER_IMPERSONATION, + "The property \"pxf.impersonation.jdbc\" has been deprecated in favor of \"pxf.service.user.impersonation\"."); + } + + /** + * Creates a new instance with default (singleton) instances of + * ConnectionManager and SecureLogin. + */ + JdbcBasePlugin() { + this(SpringContext.getBean(ConnectionManager.class), SpringContext.getBean(SecureLogin.class)); + } + + /** + * Creates a new instance with the given ConnectionManager and ConfigurationFactory + * + * @param connectionManager connection manager instance + */ + JdbcBasePlugin(ConnectionManager connectionManager, SecureLogin secureLogin) { + this.connectionManager = connectionManager; + this.secureLogin = secureLogin; + } + + @Override + public void afterPropertiesSet() { + // Required parameter. Can be auto-overwritten by user options + String jdbcDriver = configuration.get(JDBC_DRIVER_PROPERTY_NAME); + assertMandatoryParameter(jdbcDriver, JDBC_DRIVER_PROPERTY_NAME, JDBC_DRIVER_OPTION_NAME); + try { + LOG.debug("JDBC driver: '{}'", jdbcDriver); + Class.forName(jdbcDriver); + } catch (ClassNotFoundException e) { + throw new RuntimeException(e); + } + + // Required parameter. Can be auto-overwritten by user options + jdbcUrl = configuration.get(JDBC_URL_PROPERTY_NAME); + assertMandatoryParameter(jdbcUrl, JDBC_URL_PROPERTY_NAME, JDBC_URL_OPTION_NAME); + + // Required metadata + String dataSource = context.getDataSource(); + if (StringUtils.isBlank(dataSource)) { + throw new IllegalArgumentException("Data source must be provided"); + } + + // Determine if the datasource is a table name or a query name + if (dataSource.startsWith(QUERY_NAME_PREFIX)) { + queryName = dataSource.substring(QUERY_NAME_PREFIX_LENGTH); + if (StringUtils.isBlank(queryName)) { + throw new IllegalArgumentException(String.format("Query name is not provided in data source [%s]", dataSource)); + } + LOG.debug("Query name is {}", queryName); + } else { + tableName = dataSource; + LOG.debug("Table name is {}", tableName); + } + + // Required metadata + columns = context.getTupleDescription(); + + // Optional parameters + batchSizeIsSetByUser = configuration.get(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME) != null; + if (context.getRequestType() == RequestContext.RequestType.WRITE_BRIDGE) { + batchSize = configuration.getInt(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, DEFAULT_BATCH_SIZE); + + if (batchSize == 0) { + batchSize = 1; // if user set to 0, it is the same as batchSize of 1 + } else if (batchSize < 0) { + throw new IllegalArgumentException(String.format( + "Property %s has incorrect value %s : must be a non-negative integer", JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, batchSize)); + } + } + + // determine fetchSize for read operations, with different default values for MySQL driver and all others + int defaultFetchSize = jdbcDriver.startsWith(MYSQL_DRIVER_PREFIX) ? DEFAULT_MYSQL_FETCH_SIZE : DEFAULT_FETCH_SIZE; + fetchSize = configuration.getInt(JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME, defaultFetchSize); + LOG.debug("Will be using fetchSize {}", fetchSize); + + poolSize = context.getOption("POOL_SIZE", DEFAULT_POOL_SIZE); + + String queryTimeoutString = configuration.get(JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME); + if (StringUtils.isNotBlank(queryTimeoutString)) { + try { + queryTimeout = Integer.parseUnsignedInt(queryTimeoutString); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(String.format( + "Property %s has incorrect value %s : must be a non-negative integer", + JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME, queryTimeoutString), e); + } + } + + // Optional parameter. The default value is null + String quoteColumnsRaw = context.getOption("QUOTE_COLUMNS"); + if (quoteColumnsRaw != null) { + quoteColumns = Boolean.parseBoolean(quoteColumnsRaw); + } + + // Optional parameter. The default value is empty map + sessionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_SESSION_PROPERTY_PREFIX)); + // Check forbidden symbols + // Note: PreparedStatement enables us to skip this check: its values are distinct from its SQL code + // However, SET queries cannot be executed this way. This is why we do this check + if (sessionConfiguration.entrySet().stream() + .anyMatch( + entry -> + StringUtils.containsAny( + entry.getKey(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS + ) || + StringUtils.containsAny( + entry.getValue(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS + ) + ) + ) { + throw new IllegalArgumentException("Some session configuration parameter contains forbidden characters"); + } + if (LOG.isDebugEnabled()) { + LOG.debug("Session configuration: {}", + sessionConfiguration.entrySet().stream() + .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") + .collect(Collectors.joining(", ")) + ); + } + + // Optional parameter. The default value is empty map + connectionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_PROPERTY_PREFIX)); + + // Optional parameter. The default value depends on the database + String transactionIsolationString = configuration.get(JDBC_CONNECTION_TRANSACTION_ISOLATION, "NOT_PROVIDED"); + transactionIsolation = TransactionIsolation.typeOf(transactionIsolationString); + + boolean hasUserInUrl = false; + boolean hasPasswordInUrl = false; + try { + URI uri = new URI(jdbcUrl); + String query = uri.getQuery(); + if (query != null && !query.isEmpty()) { + hasUserInUrl = query.contains("user="); + hasPasswordInUrl = query.contains("password="); + } + String rawUserInfo = uri.getRawUserInfo(); + if (rawUserInfo != null && !rawUserInfo.isEmpty()) { + hasUserInUrl = true; + hasPasswordInUrl = rawUserInfo.contains("/"); // https://www.orafaq.com/wiki/JDBC + } + } catch (URISyntaxException e) { + // If the URL is malformed, it's also an invalid argument + throw new IllegalArgumentException("Invalid JDBC URL format provided: " + jdbcUrl, e); + } + + // Set optional user parameter, taking into account impersonation setting for the server. + String jdbcUser = configuration.get(JDBC_USER_PROPERTY_NAME); + boolean impersonationEnabledForServer = configuration.getBoolean(CONFIG_KEY_SERVICE_USER_IMPERSONATION, false); + LOG.debug("JDBC impersonation is {}enabled for server {}", impersonationEnabledForServer ? "" : "not ", context.getServerName()); + if (impersonationEnabledForServer) { + if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { + // secure impersonation for Hive JDBC driver requires setting URL fragment that cannot be overwritten by properties + String updatedJdbcUrl = HiveJdbcUtils.updateImpersonationPropertyInHiveJdbcUrl(jdbcUrl, context.getUser()); + LOG.debug("Replaced JDBC URL {} with {}", jdbcUrl, updatedJdbcUrl); + jdbcUrl = updatedJdbcUrl; + } else { + // the jdbcUser is the GPDB user + jdbcUser = context.getUser(); + } + } + if (jdbcUser != null && !jdbcUser.isEmpty()) { + LOG.debug("Effective JDBC user {}", jdbcUser); + connectionConfiguration.setProperty("user", jdbcUser); + } else if (!hasUserInUrl) { + LOG.debug("JDBC user has not been set"); + throw new IllegalArgumentException("JDBC user has not been set"); + } + + if (LOG.isDebugEnabled()) { + LOG.debug("Connection configuration: {}", + connectionConfiguration.entrySet().stream() + .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") + .collect(Collectors.joining(", ")) + ); + } + + // This must be the last parameter parsed, as we output connectionConfiguration earlier + // Optional parameter. By default, corresponding connectionConfiguration property is not set + boolean passwordSetted = false; + if (jdbcUser != null) { + String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); + if (jdbcPassword != null && !jdbcPassword.isEmpty()) { + LOG.debug("Connection password: {}", ConnectionManager.maskPassword(jdbcPassword)); + connectionConfiguration.setProperty("password", jdbcPassword); + passwordSetted = true; + } + } + if (!passwordSetted && !hasPasswordInUrl) { + throw new IllegalArgumentException("JDBC password has not been set"); + } + + // connection pool is optional, enabled by default + isConnectionPoolUsed = configuration.getBoolean(JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME, true); + LOG.debug("Connection pool is {}enabled", isConnectionPoolUsed ? "" : "not "); + if (isConnectionPoolUsed) { + poolConfiguration = new Properties(); + // for PXF upgrades where jdbc-site template has not been updated, make sure there're sensible defaults + poolConfiguration.setProperty("maximumPoolSize", "15"); + poolConfiguration.setProperty("connectionTimeout", "30000"); + poolConfiguration.setProperty("idleTimeout", "30000"); + poolConfiguration.setProperty("minimumIdle", "0"); + // apply values read from the template + poolConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_POOL_PROPERTY_PREFIX)); + + // packaged Hive JDBC Driver does not support connection.isValid() method, so we need to force set + // connectionTestQuery parameter in this case, unless already set by the user + if (jdbcUrl.startsWith(HIVE_URL_PREFIX) && HIVE_DEFAULT_DRIVER_CLASS.equals(jdbcDriver) && poolConfiguration.getProperty("connectionTestQuery") == null) { + poolConfiguration.setProperty("connectionTestQuery", "SELECT 1"); + } + + // get the qualifier for connection pool, if configured. Might be used when connection session authorization is employed + // to switch effective user once connection is established + poolQualifier = configuration.get(JDBC_POOL_QUALIFIER_PROPERTY_NAME); + } + + // Optional parameter to determine if the year might contain more than 4 digits in `date` or 'timestamp'. + // The default value is false. + isDateWideRange = configuration.getBoolean(JDBC_DATE_WIDE_RANGE, false); + } + + /** + * Open a new JDBC connection + * + * @return {@link Connection} + * @throws SQLException if a database access or connection error occurs + */ + public Connection getConnection() throws SQLException { + LOG.debug("Requesting a new JDBC connection. URL={} table={} txid:seg={}:{}", jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); + + Connection connection = null; + try { + connection = getConnectionInternal(); + LOG.debug("Obtained a JDBC connection {} for URL={} table={} txid:seg={}:{}", connection, jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); + + prepareConnection(connection); + } catch (Exception e) { + closeConnection(connection); + if (e instanceof SQLException) { + throw (SQLException) e; + } else { + String msg = e.getMessage(); + if (msg == null) { + Throwable t = e.getCause(); + if (t != null) msg = t.getMessage(); + } + throw new SQLException(msg, e); + } + } + + return connection; + } + + /** + * Prepare a JDBC PreparedStatement + * + * @param connection connection to use for creating the statement + * @param query query to execute + * @return PreparedStatement + * @throws SQLException if a database access error occurs + */ + public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException { + if ((connection == null) || (query == null)) { + throw new IllegalArgumentException("The provided query or connection is null"); + } + PreparedStatement statement = connection.prepareStatement(query); + if (queryTimeout != null) { + LOG.debug("Setting query timeout to {} seconds", queryTimeout); + statement.setQueryTimeout(queryTimeout); + } + return statement; + } + + /** + * Close a JDBC statement and underlying {@link Connection} + * + * @param statement statement to close + * @throws SQLException throws when a SQLException occurs + */ + public static void closeStatementAndConnection(Statement statement) throws SQLException { + if (statement == null) { + LOG.warn("Call to close statement and connection is ignored as statement provided was null"); + return; + } + + SQLException exception = null; + Connection connection = null; + + try { + connection = statement.getConnection(); + } catch (SQLException e) { + LOG.error("Exception when retrieving Connection from Statement", e); + exception = e; + } + + try { + LOG.debug("Closing statement for connection {}", connection); + statement.close(); + } catch (SQLException e) { + LOG.error("Exception when closing Statement", e); + exception = e; + } + + try { + closeConnection(connection); + } catch (SQLException e) { + LOG.error(String.format("Exception when closing connection %s", connection), e); + exception = e; + } + + if (exception != null) { + throw exception; + } + } + + /** + * For a Kerberized Hive JDBC connection, it creates a connection as the loginUser. + * Otherwise, it returns a new connection. + * + * @return for a Kerberized Hive JDBC connection, returns a new connection as the loginUser. + * Otherwise, it returns a new connection. + * @throws Exception throws when an error occurs + */ + private Connection getConnectionInternal() throws Exception { + Configuration configuration = context.getConfiguration(); + if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { + return secureLogin.getLoginUser(context, configuration).doAs((PrivilegedExceptionAction) () -> + connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier)); + } else { + return connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier); + } + } + + /** + * Close a JDBC connection + * + * @param connection connection to close + * @throws SQLException throws when a SQLException occurs + */ + protected static void closeConnection(Connection connection) throws SQLException { + if (connection == null) { + LOG.warn("Call to close connection is ignored as connection provided was null"); + return; + } + try { + if (!connection.isClosed() && + connection.getMetaData().supportsTransactions() && + !connection.getAutoCommit()) { + + LOG.debug("Committing transaction (as part of connection.close()) on connection {}", connection); + connection.commit(); + } + } finally { + try { + LOG.debug("Closing connection {}", connection); + connection.close(); + } catch (Exception e) { + // ignore + LOG.warn(String.format("Failed to close JDBC connection %s, ignoring the error.", connection), e); + } + } + } + + /** + * Prepare JDBC connection by setting session-level variables in external database + * + * @param connection {@link Connection} to prepare + */ + private void prepareConnection(Connection connection) throws SQLException { + if (connection == null) { + throw new IllegalArgumentException("The provided connection is null"); + } + + DatabaseMetaData metadata = connection.getMetaData(); + + // Handle optional connection transaction isolation level + if (transactionIsolation != TransactionIsolation.NOT_PROVIDED) { + // user wants to set isolation level explicitly + if (metadata.supportsTransactionIsolationLevel(transactionIsolation.getLevel())) { + LOG.debug("Setting transaction isolation level to {} on connection {}", transactionIsolation.toString(), connection); + connection.setTransactionIsolation(transactionIsolation.getLevel()); + } else { + throw new RuntimeException( + String.format("Transaction isolation level %s is not supported", transactionIsolation.toString()) + ); + } + } + + // Disable autocommit + if (metadata.supportsTransactions()) { + LOG.debug("Setting autoCommit to false on connection {}", connection); + connection.setAutoCommit(false); + } + + // Prepare session (process sessionConfiguration) + if (!sessionConfiguration.isEmpty()) { + DbProduct dbProduct = DbProduct.getDbProduct(metadata.getDatabaseProductName()); + + try (Statement statement = connection.createStatement()) { + for (Map.Entry e : sessionConfiguration.entrySet()) { + String sessionQuery = dbProduct.buildSessionQuery(e.getKey(), e.getValue()); + LOG.debug("Executing statement {} on connection {}", sessionQuery, connection); + statement.execute(sessionQuery); + } + } + } + } + + /** + * Asserts whether a given parameter has non-empty value, throws IllegalArgumentException otherwise + * + * @param value value to check + * @param paramName parameter name + * @param optionName name of the option for a given parameter + */ + private void assertMandatoryParameter(String value, String paramName, String optionName) { + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException(String.format( + "Required parameter %s is missing or empty in jdbc-site.xml and option %s is not specified in table definition.", paramName, optionName) + ); + } + } + + /** + * Constructs a mapping of configuration and includes all properties that start with the specified + * configuration prefix. Property names in the mapping are trimmed to remove the configuration prefix. + * This is a method from Hadoop's Configuration class ported here to make older and custom versions of Hadoop + * work with JDBC profile. + * + * @param configuration configuration map + * @param confPrefix configuration prefix + * @return mapping of configuration properties with prefix stripped + */ + private Map getPropsWithPrefix(Configuration configuration, String confPrefix) { + Map configMap = new HashMap<>(); + for (Map.Entry stringStringEntry : configuration) { + String propertyName = stringStringEntry.getKey(); + if (propertyName.startsWith(confPrefix)) { + // do not use value from the iterator as it might not come with variable substitution + String value = configuration.get(propertyName); + String keyName = propertyName.substring(confPrefix.length()); + configMap.put(keyName, value); + } + } + return configMap; + } + +} + + + diff --git a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcAccessorTest.java b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcAccessorTest.java index e659fbc67..0f1ed990b 100644 --- a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcAccessorTest.java +++ b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcAccessorTest.java @@ -58,6 +58,8 @@ public void setup() { accessor = new JdbcAccessor(mockConnectionManager, mockSecureLogin); configuration = new Configuration(); + configuration.set("jdbc.user", "test-user"); + configuration.set("jdbc.password", "test-password"); context = new RequestContext(); context.setConfig("default"); context.setDataSource("test-table"); diff --git a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java index c2a1baed2..db6cf0bc3 100644 --- a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java +++ b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java @@ -23,6 +23,7 @@ import org.apache.cloudberry.pxf.api.model.RequestContext; import org.apache.cloudberry.pxf.api.security.SecureLogin; import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -71,9 +72,19 @@ public class JdbcBasePluginTest { private RequestContext context; private Properties poolProps; + private Properties getDefaultConnectionProperties() { + Properties properties = new Properties(); + properties.setProperty("user", "test-user"); + properties.setProperty("password", "test-password"); + return properties; + } + @BeforeEach public void before() { configuration = new Configuration(); + configuration.set("jdbc.user", "test-user"); + configuration.set("jdbc.password", "test-password"); + context = new RequestContext(); context.setConfig("default"); context.setDataSource("test-table"); @@ -331,6 +342,123 @@ public void testGetPreparedStatementDoesNotSetQueryTimeoutIfNotSpecified() throw verify(mockStatement, never()).setQueryTimeout(anyInt()); } + @Test + public void testGetConnectionErrorWithoutPassword1() throws SQLException { + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url"); + configuration.set("jdbc.password", ""); + + context.setServerName("test-server"); + + try { + getPlugin(mockConnectionManager, mockSecureLogin, context); + } catch (IllegalArgumentException e) { + assertEquals("JDBC password has not been set", e.getMessage()); + return; + } + Assertions.fail("Expected an exception to be thrown due to missing password, but no exception was thrown."); + } + + @Test + public void testGetConnectionErrorWithoutPassword2() throws SQLException { + configuration = new Configuration(); + configuration.set("jdbc.user", "test-user"); + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url"); + context.setConfiguration(configuration); + + context.setServerName("test-server"); + + try { + getPlugin(mockConnectionManager, mockSecureLogin, context); + } catch (IllegalArgumentException e) { + assertEquals("JDBC password has not been set", e.getMessage()); + return; + } + Assertions.fail("Expected an exception to be thrown due to missing password, but no exception was thrown."); + } + + @Test + public void testGetConnectionWithPasswordInUrl() throws SQLException { + configuration = new Configuration(); + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url?password=test-password"); + configuration.set("jdbc.user", "test-user"); + + context.setServerName("test-server"); + context.setConfiguration(configuration); + + when(mockConnectionManager.getConnection(any(), any(), any(), anyBoolean(), any(), any())).thenReturn(mockConnection); + when(mockConnection.getMetaData()).thenReturn(mockMetaData); + + JdbcBasePlugin plugin = getPlugin(mockConnectionManager, mockSecureLogin, context); + Connection conn = plugin.getConnection(); + + assertSame(mockConnection, conn); + + Properties properties = new Properties(); + properties.setProperty("user", "test-user"); + + verify(mockConnectionManager).getConnection("test-server", "test-url?password=test-password", properties, true, poolProps, null); + } + + @Test + public void testGetConnectionErrorWithoutUser1() throws SQLException { + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url"); + configuration.set("jdbc.user", ""); + configuration.set("jdbc.password", "test-password"); + + context.setServerName("test-server"); + + try { + getPlugin(mockConnectionManager, mockSecureLogin, context); + } catch (IllegalArgumentException e) { + assertEquals("JDBC user has not been set", e.getMessage()); + return; + } + Assertions.fail("Expected an exception to be thrown due to missing user, but no exception was thrown."); + } + + @Test + public void testGetConnectionErrorWithoutUser2() throws SQLException { + configuration = new Configuration(); + configuration.set("jdbc.password", "test-password"); + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url"); + context.setConfiguration(configuration); + + context.setServerName("test-server"); + + try { + getPlugin(mockConnectionManager, mockSecureLogin, context); + } catch (IllegalArgumentException e) { + assertEquals("JDBC user has not been set", e.getMessage()); + return; + } + Assertions.fail("Expected an exception to be thrown due to missing user, but no exception was thrown."); + } + + @Test + public void testGetConnectionWithUserAndPasswordInUrl() throws SQLException { + configuration = new Configuration(); + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "test-url?user=test-user&password=test-password"); + context.setConfiguration(configuration); + + context.setServerName("test-server"); + + when(mockConnectionManager.getConnection(any(), any(), any(), anyBoolean(), any(), any())).thenReturn(mockConnection); + when(mockConnection.getMetaData()).thenReturn(mockMetaData); + + JdbcBasePlugin plugin = getPlugin(mockConnectionManager, mockSecureLogin, context); + Connection conn = plugin.getConnection(); + + assertSame(mockConnection, conn); + + verify(mockConnectionManager).getConnection("test-server", "test-url?user=test-user&password=test-password", new Properties(), true, poolProps, null); + } + @Test public void testGetConnectionNoConnPropsPoolDisabled() throws SQLException { context.setServerName("test-server"); @@ -346,14 +474,14 @@ public void testGetConnectionNoConnPropsPoolDisabled() throws SQLException { assertSame(mockConnection, conn); - verify(mockConnectionManager).getConnection("test-server", "test-url", new Properties(), false, null, null); + verify(mockConnectionManager).getConnection("test-server", "test-url", getDefaultConnectionProperties(), false, null, null); } @Test public void testGetConnectionConnPropsPoolDisabled() throws SQLException { context.setServerName("test-server"); - Properties connProps = new Properties(); + Properties connProps = getDefaultConnectionProperties(); connProps.setProperty("foo", "foo-val"); connProps.setProperty("bar", "bar-val"); @@ -392,7 +520,7 @@ public void testGetConnectionConnPropsPoolEnabledNoPoolProps() throws SQLExcepti assertSame(mockConnection, conn); - Properties connProps = new Properties(); + Properties connProps = getDefaultConnectionProperties(); connProps.setProperty("foo", "foo-val"); connProps.setProperty("bar", "bar-val"); @@ -419,7 +547,7 @@ public void testGetConnectionConnPropsPoolEnabledWithQualifier() throws SQLExcep assertSame(mockConnection, conn); - Properties connProps = new Properties(); + Properties connProps = getDefaultConnectionProperties(); connProps.setProperty("foo", "foo-val"); connProps.setProperty("bar", "bar-val"); @@ -447,7 +575,7 @@ public void testGetConnectionConnPropsPoolEnabledPoolProps() throws SQLException assertSame(mockConnection, conn); - Properties connProps = new Properties(); + Properties connProps = getDefaultConnectionProperties(); connProps.setProperty("foo", "foo-val"); connProps.setProperty("bar", "bar-val"); @@ -478,7 +606,7 @@ public void testGetConnectionConnPropsPoolDisabledPoolProps() throws SQLExceptio assertSame(mockConnection, conn); - Properties connProps = new Properties(); + Properties connProps = getDefaultConnectionProperties(); connProps.setProperty("foo", "foo-val"); connProps.setProperty("bar", "bar-val"); diff --git a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTestInitialize.java b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTestInitialize.java index 80f1a76f7..8d71b8d5d 100644 --- a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTestInitialize.java +++ b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTestInitialize.java @@ -123,6 +123,8 @@ private Configuration makeConfiguration() { Configuration configuration = new Configuration(); configuration.set("jdbc.driver", JDBC_DRIVER); configuration.set("jdbc.url", JDBC_URL); + configuration.set("jdbc.user", "test-user"); + configuration.set("jdbc.password", "test-password"); return configuration; } @@ -430,6 +432,8 @@ public void testConnectionConfiguration() throws Exception { Properties expected = new Properties(); expected.setProperty(CONFIG_PROPERTIES_KEYS[0], "v1"); expected.setProperty(CONFIG_PROPERTIES_KEYS[1], "v2"); + expected.setProperty("user", "test-user"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -449,6 +453,7 @@ public void testUser() throws Exception { // Checks Properties expected = new Properties(); expected.setProperty("user", "user"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -469,6 +474,7 @@ public void testUserWithImpersonation() throws Exception { // Checks Properties expected = new Properties(); expected.setProperty("user", "proxy"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -490,6 +496,7 @@ public void testUserWithImpersonationOverwrite() throws Exception { // Checks Properties expected = new Properties(); expected.setProperty("user", "proxy"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -511,6 +518,7 @@ public void testUserWithoutImpersonationNoOverwrite() throws Exception { // Checks Properties expected = new Properties(); expected.setProperty("user", "user"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -531,6 +539,7 @@ public void testUserDefaultImpersonationNoOverwrite() throws Exception { // Checks Properties expected = new Properties(); expected.setProperty("user", "user"); + expected.setProperty("password", "test-password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } @@ -571,6 +580,8 @@ public void testPassword() throws Exception { // Checks Properties expected = new Properties(); + expected.setProperty("user", "test-user"); + expected.setProperty("password", "password"); assertEntrySetEquals(expected.entrySet(), ((Properties) getInternalState(plugin, "connectionConfiguration")).entrySet()); } From cb2f3307325ed40db62b98385df97a154ba9b920 Mon Sep 17 00:00:00 2001 From: Nikolay Antonov Date: Wed, 24 Dec 2025 12:35:15 +0500 Subject: [PATCH 2/5] Do not enforce username/passwords for non-postgresql datasources (#17) `java.net.URI` is not proper tool to parse JDBC connections strings: * it fails to parse schemas that contains ':'. For example, `jdbc:postgres://` and `jdbc:sqlserver://` will fail to parse. * Also it won't work with mircosoft jdbc that doesn't use `?` as separator between hostname and query parameters[1]. For example, in jdbcUrl `jdbc:sqlserver://localhost:1433;databaseName=db1;user=admin` it will not extract `user` field. **Solution**: Use `Driver.parseURL` for postgresql and do not check other drivers. [1] https://learn.microsoft.com/en-us/sql/connect/jdbc/using-the-jdbc-driver?view=sql-server-ver17#make-a-simple-connection-to-a-database --- .../pxf/plugins/jdbc/JdbcBasePlugin.java | 72 ++++++++++--------- .../pxf/plugins/jdbc/JdbcBasePluginTest.java | 47 +++++++----- 2 files changed, 67 insertions(+), 52 deletions(-) diff --git a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java index 1ded13ff9..1cda61ed4 100644 --- a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java +++ b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java @@ -30,12 +30,11 @@ import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; import org.apache.cloudberry.pxf.plugins.jdbc.utils.DbProduct; import org.apache.cloudberry.pxf.plugins.jdbc.utils.HiveJdbcUtils; +import org.postgresql.Driver; +import org.postgresql.PGProperty; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.net.URI; -import java.net.URISyntaxException; import java.security.PrivilegedExceptionAction; import java.sql.Connection; import java.sql.DatabaseMetaData; @@ -100,6 +99,7 @@ public class JdbcBasePlugin extends BasePlugin { private static final String HIVE_DEFAULT_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver"; private static final String MYSQL_DRIVER_PREFIX = "com.mysql."; private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange"; + private static final String POSTGRESQL_URL_PREFIX = "jdbc:postgresql://"; private enum TransactionIsolation { READ_UNCOMMITTED(1), @@ -303,25 +303,6 @@ public void afterPropertiesSet() { String transactionIsolationString = configuration.get(JDBC_CONNECTION_TRANSACTION_ISOLATION, "NOT_PROVIDED"); transactionIsolation = TransactionIsolation.typeOf(transactionIsolationString); - boolean hasUserInUrl = false; - boolean hasPasswordInUrl = false; - try { - URI uri = new URI(jdbcUrl); - String query = uri.getQuery(); - if (query != null && !query.isEmpty()) { - hasUserInUrl = query.contains("user="); - hasPasswordInUrl = query.contains("password="); - } - String rawUserInfo = uri.getRawUserInfo(); - if (rawUserInfo != null && !rawUserInfo.isEmpty()) { - hasUserInUrl = true; - hasPasswordInUrl = rawUserInfo.contains("/"); // https://www.orafaq.com/wiki/JDBC - } - } catch (URISyntaxException e) { - // If the URL is malformed, it's also an invalid argument - throw new IllegalArgumentException("Invalid JDBC URL format provided: " + jdbcUrl, e); - } - // Set optional user parameter, taking into account impersonation setting for the server. String jdbcUser = configuration.get(JDBC_USER_PROPERTY_NAME); boolean impersonationEnabledForServer = configuration.getBoolean(CONFIG_KEY_SERVICE_USER_IMPERSONATION, false); @@ -340,9 +321,6 @@ public void afterPropertiesSet() { if (jdbcUser != null && !jdbcUser.isEmpty()) { LOG.debug("Effective JDBC user {}", jdbcUser); connectionConfiguration.setProperty("user", jdbcUser); - } else if (!hasUserInUrl) { - LOG.debug("JDBC user has not been set"); - throw new IllegalArgumentException("JDBC user has not been set"); } if (LOG.isDebugEnabled()) { @@ -355,18 +333,15 @@ public void afterPropertiesSet() { // This must be the last parameter parsed, as we output connectionConfiguration earlier // Optional parameter. By default, corresponding connectionConfiguration property is not set - boolean passwordSetted = false; + String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); if (jdbcUser != null) { - String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); if (jdbcPassword != null && !jdbcPassword.isEmpty()) { LOG.debug("Connection password: {}", ConnectionManager.maskPassword(jdbcPassword)); connectionConfiguration.setProperty("password", jdbcPassword); - passwordSetted = true; } } - if (!passwordSetted && !hasPasswordInUrl) { - throw new IllegalArgumentException("JDBC password has not been set"); - } + + assertJDBC(jdbcUrl, jdbcUser, jdbcPassword); // connection pool is optional, enabled by default isConnectionPoolUsed = configuration.getBoolean(JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME, true); @@ -429,6 +404,38 @@ public Connection getConnection() throws SQLException { return connection; } + /** + * Asserts whether all required JDBC URL params/properties set, throws IllegalArgumentException otherwise + * + * @param jdbcUrl connection string + * @param username username from properties (outside of connection string) + * @param password password from properties (outside of connection string) + */ + private void assertJDBC(String jdbcUrl, String username, String password) { + if (!StringUtils.startsWith(jdbcUrl, POSTGRESQL_URL_PREFIX)) { + return; + } + + // For PostgreSQL we enforce users to specify username and password + Properties urlProperties = Driver.parseURL(jdbcUrl, null); + if (urlProperties == null) { + throw new IllegalArgumentException("Invalid PostgreSQL JDBC URL format provided: " + jdbcUrl); + } + + String urlUsername = urlProperties.getProperty(PGProperty.USER.getName()); + if (StringUtils.isEmpty(username) && StringUtils.isEmpty(urlUsername)) { + LOG.debug("PostgreSQL JDBC user has not been set"); + throw new IllegalArgumentException("PostgreSQL JDBC user has not been set"); + } + + String urlPassword = urlProperties.getProperty(PGProperty.PASSWORD.getName()); + if (StringUtils.isEmpty(password) && StringUtils.isEmpty(urlPassword)) { + throw new IllegalArgumentException("PostgreSQL JDBC password has not been set"); + } + + // Ok. It is fine to work with this postgres datasource +} + /** * Prepare a JDBC PreparedStatement * @@ -624,6 +631,3 @@ private Map getPropsWithPrefix(Configuration configuration, Stri } } - - - diff --git a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java index db6cf0bc3..2bcb781dc 100644 --- a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java +++ b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java @@ -344,8 +344,8 @@ public void testGetPreparedStatementDoesNotSetQueryTimeoutIfNotSpecified() throw @Test public void testGetConnectionErrorWithoutPassword1() throws SQLException { - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url"); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/test-url"); configuration.set("jdbc.password", ""); context.setServerName("test-server"); @@ -353,7 +353,7 @@ public void testGetConnectionErrorWithoutPassword1() throws SQLException { try { getPlugin(mockConnectionManager, mockSecureLogin, context); } catch (IllegalArgumentException e) { - assertEquals("JDBC password has not been set", e.getMessage()); + assertEquals("PostgreSQL JDBC password has not been set", e.getMessage()); return; } Assertions.fail("Expected an exception to be thrown due to missing password, but no exception was thrown."); @@ -362,9 +362,9 @@ public void testGetConnectionErrorWithoutPassword1() throws SQLException { @Test public void testGetConnectionErrorWithoutPassword2() throws SQLException { configuration = new Configuration(); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/test-url"); configuration.set("jdbc.user", "test-user"); - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url"); context.setConfiguration(configuration); context.setServerName("test-server"); @@ -372,17 +372,28 @@ public void testGetConnectionErrorWithoutPassword2() throws SQLException { try { getPlugin(mockConnectionManager, mockSecureLogin, context); } catch (IllegalArgumentException e) { - assertEquals("JDBC password has not been set", e.getMessage()); + assertEquals("PostgreSQL JDBC password has not been set", e.getMessage()); return; } Assertions.fail("Expected an exception to be thrown due to missing password, but no exception was thrown."); } + @Test + public void testGetConnectionErrorWithoutPassword3() throws SQLException { + configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.url", "jdbc:sqlserver://localhost:1433;databaseName=db1;user=admin"); + + context.setServerName("test-server"); + + // non-PostgreSQL drivers are not subject to credential validation + getPlugin(mockConnectionManager, mockSecureLogin, context); + } + @Test public void testGetConnectionWithPasswordInUrl() throws SQLException { configuration = new Configuration(); - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url?password=test-password"); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/test-url?password=test-password"); configuration.set("jdbc.user", "test-user"); context.setServerName("test-server"); @@ -399,13 +410,13 @@ public void testGetConnectionWithPasswordInUrl() throws SQLException { Properties properties = new Properties(); properties.setProperty("user", "test-user"); - verify(mockConnectionManager).getConnection("test-server", "test-url?password=test-password", properties, true, poolProps, null); + verify(mockConnectionManager).getConnection("test-server", "jdbc:postgresql://example.com/test-url?password=test-password", properties, true, poolProps, null); } @Test public void testGetConnectionErrorWithoutUser1() throws SQLException { - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url"); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/test-url"); configuration.set("jdbc.user", ""); configuration.set("jdbc.password", "test-password"); @@ -414,7 +425,7 @@ public void testGetConnectionErrorWithoutUser1() throws SQLException { try { getPlugin(mockConnectionManager, mockSecureLogin, context); } catch (IllegalArgumentException e) { - assertEquals("JDBC user has not been set", e.getMessage()); + assertEquals("PostgreSQL JDBC user has not been set", e.getMessage()); return; } Assertions.fail("Expected an exception to be thrown due to missing user, but no exception was thrown."); @@ -424,8 +435,8 @@ public void testGetConnectionErrorWithoutUser1() throws SQLException { public void testGetConnectionErrorWithoutUser2() throws SQLException { configuration = new Configuration(); configuration.set("jdbc.password", "test-password"); - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url"); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/test-url"); context.setConfiguration(configuration); context.setServerName("test-server"); @@ -433,7 +444,7 @@ public void testGetConnectionErrorWithoutUser2() throws SQLException { try { getPlugin(mockConnectionManager, mockSecureLogin, context); } catch (IllegalArgumentException e) { - assertEquals("JDBC user has not been set", e.getMessage()); + assertEquals("PostgreSQL JDBC user has not been set", e.getMessage()); return; } Assertions.fail("Expected an exception to be thrown due to missing user, but no exception was thrown."); @@ -442,8 +453,8 @@ public void testGetConnectionErrorWithoutUser2() throws SQLException { @Test public void testGetConnectionWithUserAndPasswordInUrl() throws SQLException { configuration = new Configuration(); - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); - configuration.set("jdbc.url", "test-url?user=test-user&password=test-password"); + configuration.set("jdbc.driver", "org.postgresql.Driver"); + configuration.set("jdbc.url", "jdbc:postgresql://example.com/db?user=test-user&password=test-password"); context.setConfiguration(configuration); context.setServerName("test-server"); @@ -456,7 +467,7 @@ public void testGetConnectionWithUserAndPasswordInUrl() throws SQLException { assertSame(mockConnection, conn); - verify(mockConnectionManager).getConnection("test-server", "test-url?user=test-user&password=test-password", new Properties(), true, poolProps, null); + verify(mockConnectionManager).getConnection("test-server", "jdbc:postgresql://example.com/db?user=test-user&password=test-password", new Properties(), true, poolProps, null); } @Test From 93cbd30fd455e7c59bf8860378b28dc333330279 Mon Sep 17 00:00:00 2001 From: Nikolay Antonov Date: Fri, 17 Jul 2026 17:05:37 +0300 Subject: [PATCH 3/5] CLRF -> LF --- .../pxf/plugins/jdbc/JdbcBasePlugin.java | 1266 ++++++++--------- 1 file changed, 633 insertions(+), 633 deletions(-) diff --git a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java index 1cda61ed4..0ce2a8777 100644 --- a/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java +++ b/server/pxf-jdbc/src/main/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePlugin.java @@ -1,633 +1,633 @@ -package org.apache.cloudberry.pxf.plugins.jdbc; - -/* - * 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. - */ - -import org.apache.commons.lang.StringUtils; -import org.apache.hadoop.conf.Configuration; -import org.apache.cloudberry.pxf.api.model.BasePlugin; -import org.apache.cloudberry.pxf.api.model.RequestContext; -import org.apache.cloudberry.pxf.api.security.SecureLogin; -import org.apache.cloudberry.pxf.api.utilities.ColumnDescriptor; -import org.apache.cloudberry.pxf.api.utilities.SpringContext; -import org.apache.cloudberry.pxf.api.utilities.Utilities; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.DbProduct; -import org.apache.cloudberry.pxf.plugins.jdbc.utils.HiveJdbcUtils; -import org.postgresql.Driver; -import org.postgresql.PGProperty; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.security.PrivilegedExceptionAction; -import java.sql.Connection; -import java.sql.DatabaseMetaData; -import java.sql.PreparedStatement; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Properties; -import java.util.stream.Collectors; - -import static org.apache.cloudberry.pxf.api.security.SecureLogin.CONFIG_KEY_SERVICE_USER_IMPERSONATION; - -/** - * JDBC tables plugin (base class) - *

- * Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}. - */ -public class JdbcBasePlugin extends BasePlugin { - - private static final Logger LOG = LoggerFactory.getLogger(JdbcBasePlugin.class); - - // '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754 - private static final int DEFAULT_BATCH_SIZE = 100; - private static final int DEFAULT_FETCH_SIZE = 1000; - // MySQL fetches all data in memory first unless streaming is enabled by setting fetchSize to Integer.MIN_VALUE - // see https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-implementation-notes.html - private static final int DEFAULT_MYSQL_FETCH_SIZE = Integer.MIN_VALUE; - private static final int DEFAULT_POOL_SIZE = 1; - - // configuration parameter names - private static final String JDBC_DRIVER_PROPERTY_NAME = "jdbc.driver"; - private static final String JDBC_URL_PROPERTY_NAME = "jdbc.url"; - private static final String JDBC_USER_PROPERTY_NAME = "jdbc.user"; - private static final String JDBC_PASSWORD_PROPERTY_NAME = "jdbc.password"; - private static final String JDBC_SESSION_PROPERTY_PREFIX = "jdbc.session.property."; - private static final String JDBC_CONNECTION_PROPERTY_PREFIX = "jdbc.connection.property."; - - // connection parameter names - private static final String JDBC_CONNECTION_TRANSACTION_ISOLATION = "jdbc.connection.transactionIsolation"; - - // statement properties - private static final String JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME = "jdbc.statement.batchSize"; - private static final String JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME = "jdbc.statement.fetchSize"; - private static final String JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME = "jdbc.statement.queryTimeout"; - - // connection pool properties - private static final String JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME = "jdbc.pool.enabled"; - private static final String JDBC_CONNECTION_POOL_PROPERTY_PREFIX = "jdbc.pool.property."; - private static final String JDBC_POOL_QUALIFIER_PROPERTY_NAME = "jdbc.pool.qualifier"; - - // DDL option names - private static final String JDBC_DRIVER_OPTION_NAME = "JDBC_DRIVER"; - private static final String JDBC_URL_OPTION_NAME = "DB_URL"; - - private static final String FORBIDDEN_SESSION_PROPERTY_CHARACTERS = ";\n\b\0"; - private static final String QUERY_NAME_PREFIX = "query:"; - private static final int QUERY_NAME_PREFIX_LENGTH = QUERY_NAME_PREFIX.length(); - - private static final String HIVE_URL_PREFIX = "jdbc:hive2://"; - private static final String HIVE_DEFAULT_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver"; - private static final String MYSQL_DRIVER_PREFIX = "com.mysql."; - private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange"; - private static final String POSTGRESQL_URL_PREFIX = "jdbc:postgresql://"; - - private enum TransactionIsolation { - READ_UNCOMMITTED(1), - READ_COMMITTED(2), - REPEATABLE_READ(4), - SERIALIZABLE(8), - NOT_PROVIDED(-1); - - private final int isolationLevel; - - TransactionIsolation(int transactionIsolation) { - isolationLevel = transactionIsolation; - } - - public int getLevel() { - return isolationLevel; - } - - public static TransactionIsolation typeOf(String str) { - return valueOf(str); - } - } - - // JDBC parameters from config file or specified in DDL - - private String jdbcUrl; - - protected String tableName; - - // Write batch size - protected int batchSize; - protected boolean batchSizeIsSetByUser = false; - - // Read batch size - protected int fetchSize; - - // Thread pool size - protected int poolSize; - - // Query timeout. - protected Integer queryTimeout; - - // Quote columns setting set by user (three values are possible) - protected Boolean quoteColumns = null; - - // Environment variables to SET before query execution - protected Map sessionConfiguration = new HashMap<>(); - - // Properties object to pass to JDBC Driver when connection is created - protected Properties connectionConfiguration = new Properties(); - - // Transaction isolation level that a user can configure - private TransactionIsolation transactionIsolation = TransactionIsolation.NOT_PROVIDED; - - // Columns description - protected List columns = null; - - // Name of query to execute for read flow (optional) - protected String queryName; - - // connection pool fields - private boolean isConnectionPoolUsed; - private Properties poolConfiguration; - private String poolQualifier; - - private final ConnectionManager connectionManager; - private final SecureLogin secureLogin; - - // Flag which is used when the year might contain more than 4 digits in `date` or 'timestamp' - protected boolean isDateWideRange; - - static { - // Deprecated as of Oct 22, 2019 in version 5.9.2+ - Configuration.addDeprecation("pxf.impersonation.jdbc", - CONFIG_KEY_SERVICE_USER_IMPERSONATION, - "The property \"pxf.impersonation.jdbc\" has been deprecated in favor of \"pxf.service.user.impersonation\"."); - } - - /** - * Creates a new instance with default (singleton) instances of - * ConnectionManager and SecureLogin. - */ - JdbcBasePlugin() { - this(SpringContext.getBean(ConnectionManager.class), SpringContext.getBean(SecureLogin.class)); - } - - /** - * Creates a new instance with the given ConnectionManager and ConfigurationFactory - * - * @param connectionManager connection manager instance - */ - JdbcBasePlugin(ConnectionManager connectionManager, SecureLogin secureLogin) { - this.connectionManager = connectionManager; - this.secureLogin = secureLogin; - } - - @Override - public void afterPropertiesSet() { - // Required parameter. Can be auto-overwritten by user options - String jdbcDriver = configuration.get(JDBC_DRIVER_PROPERTY_NAME); - assertMandatoryParameter(jdbcDriver, JDBC_DRIVER_PROPERTY_NAME, JDBC_DRIVER_OPTION_NAME); - try { - LOG.debug("JDBC driver: '{}'", jdbcDriver); - Class.forName(jdbcDriver); - } catch (ClassNotFoundException e) { - throw new RuntimeException(e); - } - - // Required parameter. Can be auto-overwritten by user options - jdbcUrl = configuration.get(JDBC_URL_PROPERTY_NAME); - assertMandatoryParameter(jdbcUrl, JDBC_URL_PROPERTY_NAME, JDBC_URL_OPTION_NAME); - - // Required metadata - String dataSource = context.getDataSource(); - if (StringUtils.isBlank(dataSource)) { - throw new IllegalArgumentException("Data source must be provided"); - } - - // Determine if the datasource is a table name or a query name - if (dataSource.startsWith(QUERY_NAME_PREFIX)) { - queryName = dataSource.substring(QUERY_NAME_PREFIX_LENGTH); - if (StringUtils.isBlank(queryName)) { - throw new IllegalArgumentException(String.format("Query name is not provided in data source [%s]", dataSource)); - } - LOG.debug("Query name is {}", queryName); - } else { - tableName = dataSource; - LOG.debug("Table name is {}", tableName); - } - - // Required metadata - columns = context.getTupleDescription(); - - // Optional parameters - batchSizeIsSetByUser = configuration.get(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME) != null; - if (context.getRequestType() == RequestContext.RequestType.WRITE_BRIDGE) { - batchSize = configuration.getInt(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, DEFAULT_BATCH_SIZE); - - if (batchSize == 0) { - batchSize = 1; // if user set to 0, it is the same as batchSize of 1 - } else if (batchSize < 0) { - throw new IllegalArgumentException(String.format( - "Property %s has incorrect value %s : must be a non-negative integer", JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, batchSize)); - } - } - - // determine fetchSize for read operations, with different default values for MySQL driver and all others - int defaultFetchSize = jdbcDriver.startsWith(MYSQL_DRIVER_PREFIX) ? DEFAULT_MYSQL_FETCH_SIZE : DEFAULT_FETCH_SIZE; - fetchSize = configuration.getInt(JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME, defaultFetchSize); - LOG.debug("Will be using fetchSize {}", fetchSize); - - poolSize = context.getOption("POOL_SIZE", DEFAULT_POOL_SIZE); - - String queryTimeoutString = configuration.get(JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME); - if (StringUtils.isNotBlank(queryTimeoutString)) { - try { - queryTimeout = Integer.parseUnsignedInt(queryTimeoutString); - } catch (NumberFormatException e) { - throw new IllegalArgumentException(String.format( - "Property %s has incorrect value %s : must be a non-negative integer", - JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME, queryTimeoutString), e); - } - } - - // Optional parameter. The default value is null - String quoteColumnsRaw = context.getOption("QUOTE_COLUMNS"); - if (quoteColumnsRaw != null) { - quoteColumns = Boolean.parseBoolean(quoteColumnsRaw); - } - - // Optional parameter. The default value is empty map - sessionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_SESSION_PROPERTY_PREFIX)); - // Check forbidden symbols - // Note: PreparedStatement enables us to skip this check: its values are distinct from its SQL code - // However, SET queries cannot be executed this way. This is why we do this check - if (sessionConfiguration.entrySet().stream() - .anyMatch( - entry -> - StringUtils.containsAny( - entry.getKey(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS - ) || - StringUtils.containsAny( - entry.getValue(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS - ) - ) - ) { - throw new IllegalArgumentException("Some session configuration parameter contains forbidden characters"); - } - if (LOG.isDebugEnabled()) { - LOG.debug("Session configuration: {}", - sessionConfiguration.entrySet().stream() - .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") - .collect(Collectors.joining(", ")) - ); - } - - // Optional parameter. The default value is empty map - connectionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_PROPERTY_PREFIX)); - - // Optional parameter. The default value depends on the database - String transactionIsolationString = configuration.get(JDBC_CONNECTION_TRANSACTION_ISOLATION, "NOT_PROVIDED"); - transactionIsolation = TransactionIsolation.typeOf(transactionIsolationString); - - // Set optional user parameter, taking into account impersonation setting for the server. - String jdbcUser = configuration.get(JDBC_USER_PROPERTY_NAME); - boolean impersonationEnabledForServer = configuration.getBoolean(CONFIG_KEY_SERVICE_USER_IMPERSONATION, false); - LOG.debug("JDBC impersonation is {}enabled for server {}", impersonationEnabledForServer ? "" : "not ", context.getServerName()); - if (impersonationEnabledForServer) { - if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { - // secure impersonation for Hive JDBC driver requires setting URL fragment that cannot be overwritten by properties - String updatedJdbcUrl = HiveJdbcUtils.updateImpersonationPropertyInHiveJdbcUrl(jdbcUrl, context.getUser()); - LOG.debug("Replaced JDBC URL {} with {}", jdbcUrl, updatedJdbcUrl); - jdbcUrl = updatedJdbcUrl; - } else { - // the jdbcUser is the GPDB user - jdbcUser = context.getUser(); - } - } - if (jdbcUser != null && !jdbcUser.isEmpty()) { - LOG.debug("Effective JDBC user {}", jdbcUser); - connectionConfiguration.setProperty("user", jdbcUser); - } - - if (LOG.isDebugEnabled()) { - LOG.debug("Connection configuration: {}", - connectionConfiguration.entrySet().stream() - .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") - .collect(Collectors.joining(", ")) - ); - } - - // This must be the last parameter parsed, as we output connectionConfiguration earlier - // Optional parameter. By default, corresponding connectionConfiguration property is not set - String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); - if (jdbcUser != null) { - if (jdbcPassword != null && !jdbcPassword.isEmpty()) { - LOG.debug("Connection password: {}", ConnectionManager.maskPassword(jdbcPassword)); - connectionConfiguration.setProperty("password", jdbcPassword); - } - } - - assertJDBC(jdbcUrl, jdbcUser, jdbcPassword); - - // connection pool is optional, enabled by default - isConnectionPoolUsed = configuration.getBoolean(JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME, true); - LOG.debug("Connection pool is {}enabled", isConnectionPoolUsed ? "" : "not "); - if (isConnectionPoolUsed) { - poolConfiguration = new Properties(); - // for PXF upgrades where jdbc-site template has not been updated, make sure there're sensible defaults - poolConfiguration.setProperty("maximumPoolSize", "15"); - poolConfiguration.setProperty("connectionTimeout", "30000"); - poolConfiguration.setProperty("idleTimeout", "30000"); - poolConfiguration.setProperty("minimumIdle", "0"); - // apply values read from the template - poolConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_POOL_PROPERTY_PREFIX)); - - // packaged Hive JDBC Driver does not support connection.isValid() method, so we need to force set - // connectionTestQuery parameter in this case, unless already set by the user - if (jdbcUrl.startsWith(HIVE_URL_PREFIX) && HIVE_DEFAULT_DRIVER_CLASS.equals(jdbcDriver) && poolConfiguration.getProperty("connectionTestQuery") == null) { - poolConfiguration.setProperty("connectionTestQuery", "SELECT 1"); - } - - // get the qualifier for connection pool, if configured. Might be used when connection session authorization is employed - // to switch effective user once connection is established - poolQualifier = configuration.get(JDBC_POOL_QUALIFIER_PROPERTY_NAME); - } - - // Optional parameter to determine if the year might contain more than 4 digits in `date` or 'timestamp'. - // The default value is false. - isDateWideRange = configuration.getBoolean(JDBC_DATE_WIDE_RANGE, false); - } - - /** - * Open a new JDBC connection - * - * @return {@link Connection} - * @throws SQLException if a database access or connection error occurs - */ - public Connection getConnection() throws SQLException { - LOG.debug("Requesting a new JDBC connection. URL={} table={} txid:seg={}:{}", jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); - - Connection connection = null; - try { - connection = getConnectionInternal(); - LOG.debug("Obtained a JDBC connection {} for URL={} table={} txid:seg={}:{}", connection, jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); - - prepareConnection(connection); - } catch (Exception e) { - closeConnection(connection); - if (e instanceof SQLException) { - throw (SQLException) e; - } else { - String msg = e.getMessage(); - if (msg == null) { - Throwable t = e.getCause(); - if (t != null) msg = t.getMessage(); - } - throw new SQLException(msg, e); - } - } - - return connection; - } - - /** - * Asserts whether all required JDBC URL params/properties set, throws IllegalArgumentException otherwise - * - * @param jdbcUrl connection string - * @param username username from properties (outside of connection string) - * @param password password from properties (outside of connection string) - */ - private void assertJDBC(String jdbcUrl, String username, String password) { - if (!StringUtils.startsWith(jdbcUrl, POSTGRESQL_URL_PREFIX)) { - return; - } - - // For PostgreSQL we enforce users to specify username and password - Properties urlProperties = Driver.parseURL(jdbcUrl, null); - if (urlProperties == null) { - throw new IllegalArgumentException("Invalid PostgreSQL JDBC URL format provided: " + jdbcUrl); - } - - String urlUsername = urlProperties.getProperty(PGProperty.USER.getName()); - if (StringUtils.isEmpty(username) && StringUtils.isEmpty(urlUsername)) { - LOG.debug("PostgreSQL JDBC user has not been set"); - throw new IllegalArgumentException("PostgreSQL JDBC user has not been set"); - } - - String urlPassword = urlProperties.getProperty(PGProperty.PASSWORD.getName()); - if (StringUtils.isEmpty(password) && StringUtils.isEmpty(urlPassword)) { - throw new IllegalArgumentException("PostgreSQL JDBC password has not been set"); - } - - // Ok. It is fine to work with this postgres datasource -} - - /** - * Prepare a JDBC PreparedStatement - * - * @param connection connection to use for creating the statement - * @param query query to execute - * @return PreparedStatement - * @throws SQLException if a database access error occurs - */ - public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException { - if ((connection == null) || (query == null)) { - throw new IllegalArgumentException("The provided query or connection is null"); - } - PreparedStatement statement = connection.prepareStatement(query); - if (queryTimeout != null) { - LOG.debug("Setting query timeout to {} seconds", queryTimeout); - statement.setQueryTimeout(queryTimeout); - } - return statement; - } - - /** - * Close a JDBC statement and underlying {@link Connection} - * - * @param statement statement to close - * @throws SQLException throws when a SQLException occurs - */ - public static void closeStatementAndConnection(Statement statement) throws SQLException { - if (statement == null) { - LOG.warn("Call to close statement and connection is ignored as statement provided was null"); - return; - } - - SQLException exception = null; - Connection connection = null; - - try { - connection = statement.getConnection(); - } catch (SQLException e) { - LOG.error("Exception when retrieving Connection from Statement", e); - exception = e; - } - - try { - LOG.debug("Closing statement for connection {}", connection); - statement.close(); - } catch (SQLException e) { - LOG.error("Exception when closing Statement", e); - exception = e; - } - - try { - closeConnection(connection); - } catch (SQLException e) { - LOG.error(String.format("Exception when closing connection %s", connection), e); - exception = e; - } - - if (exception != null) { - throw exception; - } - } - - /** - * For a Kerberized Hive JDBC connection, it creates a connection as the loginUser. - * Otherwise, it returns a new connection. - * - * @return for a Kerberized Hive JDBC connection, returns a new connection as the loginUser. - * Otherwise, it returns a new connection. - * @throws Exception throws when an error occurs - */ - private Connection getConnectionInternal() throws Exception { - Configuration configuration = context.getConfiguration(); - if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { - return secureLogin.getLoginUser(context, configuration).doAs((PrivilegedExceptionAction) () -> - connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier)); - } else { - return connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier); - } - } - - /** - * Close a JDBC connection - * - * @param connection connection to close - * @throws SQLException throws when a SQLException occurs - */ - protected static void closeConnection(Connection connection) throws SQLException { - if (connection == null) { - LOG.warn("Call to close connection is ignored as connection provided was null"); - return; - } - try { - if (!connection.isClosed() && - connection.getMetaData().supportsTransactions() && - !connection.getAutoCommit()) { - - LOG.debug("Committing transaction (as part of connection.close()) on connection {}", connection); - connection.commit(); - } - } finally { - try { - LOG.debug("Closing connection {}", connection); - connection.close(); - } catch (Exception e) { - // ignore - LOG.warn(String.format("Failed to close JDBC connection %s, ignoring the error.", connection), e); - } - } - } - - /** - * Prepare JDBC connection by setting session-level variables in external database - * - * @param connection {@link Connection} to prepare - */ - private void prepareConnection(Connection connection) throws SQLException { - if (connection == null) { - throw new IllegalArgumentException("The provided connection is null"); - } - - DatabaseMetaData metadata = connection.getMetaData(); - - // Handle optional connection transaction isolation level - if (transactionIsolation != TransactionIsolation.NOT_PROVIDED) { - // user wants to set isolation level explicitly - if (metadata.supportsTransactionIsolationLevel(transactionIsolation.getLevel())) { - LOG.debug("Setting transaction isolation level to {} on connection {}", transactionIsolation.toString(), connection); - connection.setTransactionIsolation(transactionIsolation.getLevel()); - } else { - throw new RuntimeException( - String.format("Transaction isolation level %s is not supported", transactionIsolation.toString()) - ); - } - } - - // Disable autocommit - if (metadata.supportsTransactions()) { - LOG.debug("Setting autoCommit to false on connection {}", connection); - connection.setAutoCommit(false); - } - - // Prepare session (process sessionConfiguration) - if (!sessionConfiguration.isEmpty()) { - DbProduct dbProduct = DbProduct.getDbProduct(metadata.getDatabaseProductName()); - - try (Statement statement = connection.createStatement()) { - for (Map.Entry e : sessionConfiguration.entrySet()) { - String sessionQuery = dbProduct.buildSessionQuery(e.getKey(), e.getValue()); - LOG.debug("Executing statement {} on connection {}", sessionQuery, connection); - statement.execute(sessionQuery); - } - } - } - } - - /** - * Asserts whether a given parameter has non-empty value, throws IllegalArgumentException otherwise - * - * @param value value to check - * @param paramName parameter name - * @param optionName name of the option for a given parameter - */ - private void assertMandatoryParameter(String value, String paramName, String optionName) { - if (StringUtils.isBlank(value)) { - throw new IllegalArgumentException(String.format( - "Required parameter %s is missing or empty in jdbc-site.xml and option %s is not specified in table definition.", paramName, optionName) - ); - } - } - - /** - * Constructs a mapping of configuration and includes all properties that start with the specified - * configuration prefix. Property names in the mapping are trimmed to remove the configuration prefix. - * This is a method from Hadoop's Configuration class ported here to make older and custom versions of Hadoop - * work with JDBC profile. - * - * @param configuration configuration map - * @param confPrefix configuration prefix - * @return mapping of configuration properties with prefix stripped - */ - private Map getPropsWithPrefix(Configuration configuration, String confPrefix) { - Map configMap = new HashMap<>(); - for (Map.Entry stringStringEntry : configuration) { - String propertyName = stringStringEntry.getKey(); - if (propertyName.startsWith(confPrefix)) { - // do not use value from the iterator as it might not come with variable substitution - String value = configuration.get(propertyName); - String keyName = propertyName.substring(confPrefix.length()); - configMap.put(keyName, value); - } - } - return configMap; - } - -} +package org.apache.cloudberry.pxf.plugins.jdbc; + +/* + * 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. + */ + +import org.apache.commons.lang.StringUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.cloudberry.pxf.api.model.BasePlugin; +import org.apache.cloudberry.pxf.api.model.RequestContext; +import org.apache.cloudberry.pxf.api.security.SecureLogin; +import org.apache.cloudberry.pxf.api.utilities.ColumnDescriptor; +import org.apache.cloudberry.pxf.api.utilities.SpringContext; +import org.apache.cloudberry.pxf.api.utilities.Utilities; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.ConnectionManager; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.DbProduct; +import org.apache.cloudberry.pxf.plugins.jdbc.utils.HiveJdbcUtils; +import org.postgresql.Driver; +import org.postgresql.PGProperty; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.security.PrivilegedExceptionAction; +import java.sql.Connection; +import java.sql.DatabaseMetaData; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; + +import static org.apache.cloudberry.pxf.api.security.SecureLogin.CONFIG_KEY_SERVICE_USER_IMPERSONATION; + +/** + * JDBC tables plugin (base class) + *

+ * Implemented subclasses: {@link JdbcAccessor}, {@link JdbcResolver}. + */ +public class JdbcBasePlugin extends BasePlugin { + + private static final Logger LOG = LoggerFactory.getLogger(JdbcBasePlugin.class); + + // '100' is a recommended value: https://docs.oracle.com/cd/E11882_01/java.112/e16548/oraperf.htm#JJDBC28754 + private static final int DEFAULT_BATCH_SIZE = 100; + private static final int DEFAULT_FETCH_SIZE = 1000; + // MySQL fetches all data in memory first unless streaming is enabled by setting fetchSize to Integer.MIN_VALUE + // see https://dev.mysql.com/doc/connector-j/8.0/en/connector-j-reference-implementation-notes.html + private static final int DEFAULT_MYSQL_FETCH_SIZE = Integer.MIN_VALUE; + private static final int DEFAULT_POOL_SIZE = 1; + + // configuration parameter names + private static final String JDBC_DRIVER_PROPERTY_NAME = "jdbc.driver"; + private static final String JDBC_URL_PROPERTY_NAME = "jdbc.url"; + private static final String JDBC_USER_PROPERTY_NAME = "jdbc.user"; + private static final String JDBC_PASSWORD_PROPERTY_NAME = "jdbc.password"; + private static final String JDBC_SESSION_PROPERTY_PREFIX = "jdbc.session.property."; + private static final String JDBC_CONNECTION_PROPERTY_PREFIX = "jdbc.connection.property."; + + // connection parameter names + private static final String JDBC_CONNECTION_TRANSACTION_ISOLATION = "jdbc.connection.transactionIsolation"; + + // statement properties + private static final String JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME = "jdbc.statement.batchSize"; + private static final String JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME = "jdbc.statement.fetchSize"; + private static final String JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME = "jdbc.statement.queryTimeout"; + + // connection pool properties + private static final String JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME = "jdbc.pool.enabled"; + private static final String JDBC_CONNECTION_POOL_PROPERTY_PREFIX = "jdbc.pool.property."; + private static final String JDBC_POOL_QUALIFIER_PROPERTY_NAME = "jdbc.pool.qualifier"; + + // DDL option names + private static final String JDBC_DRIVER_OPTION_NAME = "JDBC_DRIVER"; + private static final String JDBC_URL_OPTION_NAME = "DB_URL"; + + private static final String FORBIDDEN_SESSION_PROPERTY_CHARACTERS = ";\n\b\0"; + private static final String QUERY_NAME_PREFIX = "query:"; + private static final int QUERY_NAME_PREFIX_LENGTH = QUERY_NAME_PREFIX.length(); + + private static final String HIVE_URL_PREFIX = "jdbc:hive2://"; + private static final String HIVE_DEFAULT_DRIVER_CLASS = "org.apache.hive.jdbc.HiveDriver"; + private static final String MYSQL_DRIVER_PREFIX = "com.mysql."; + private static final String JDBC_DATE_WIDE_RANGE = "jdbc.date.wideRange"; + private static final String POSTGRESQL_URL_PREFIX = "jdbc:postgresql://"; + + private enum TransactionIsolation { + READ_UNCOMMITTED(1), + READ_COMMITTED(2), + REPEATABLE_READ(4), + SERIALIZABLE(8), + NOT_PROVIDED(-1); + + private final int isolationLevel; + + TransactionIsolation(int transactionIsolation) { + isolationLevel = transactionIsolation; + } + + public int getLevel() { + return isolationLevel; + } + + public static TransactionIsolation typeOf(String str) { + return valueOf(str); + } + } + + // JDBC parameters from config file or specified in DDL + + private String jdbcUrl; + + protected String tableName; + + // Write batch size + protected int batchSize; + protected boolean batchSizeIsSetByUser = false; + + // Read batch size + protected int fetchSize; + + // Thread pool size + protected int poolSize; + + // Query timeout. + protected Integer queryTimeout; + + // Quote columns setting set by user (three values are possible) + protected Boolean quoteColumns = null; + + // Environment variables to SET before query execution + protected Map sessionConfiguration = new HashMap<>(); + + // Properties object to pass to JDBC Driver when connection is created + protected Properties connectionConfiguration = new Properties(); + + // Transaction isolation level that a user can configure + private TransactionIsolation transactionIsolation = TransactionIsolation.NOT_PROVIDED; + + // Columns description + protected List columns = null; + + // Name of query to execute for read flow (optional) + protected String queryName; + + // connection pool fields + private boolean isConnectionPoolUsed; + private Properties poolConfiguration; + private String poolQualifier; + + private final ConnectionManager connectionManager; + private final SecureLogin secureLogin; + + // Flag which is used when the year might contain more than 4 digits in `date` or 'timestamp' + protected boolean isDateWideRange; + + static { + // Deprecated as of Oct 22, 2019 in version 5.9.2+ + Configuration.addDeprecation("pxf.impersonation.jdbc", + CONFIG_KEY_SERVICE_USER_IMPERSONATION, + "The property \"pxf.impersonation.jdbc\" has been deprecated in favor of \"pxf.service.user.impersonation\"."); + } + + /** + * Creates a new instance with default (singleton) instances of + * ConnectionManager and SecureLogin. + */ + JdbcBasePlugin() { + this(SpringContext.getBean(ConnectionManager.class), SpringContext.getBean(SecureLogin.class)); + } + + /** + * Creates a new instance with the given ConnectionManager and ConfigurationFactory + * + * @param connectionManager connection manager instance + */ + JdbcBasePlugin(ConnectionManager connectionManager, SecureLogin secureLogin) { + this.connectionManager = connectionManager; + this.secureLogin = secureLogin; + } + + @Override + public void afterPropertiesSet() { + // Required parameter. Can be auto-overwritten by user options + String jdbcDriver = configuration.get(JDBC_DRIVER_PROPERTY_NAME); + assertMandatoryParameter(jdbcDriver, JDBC_DRIVER_PROPERTY_NAME, JDBC_DRIVER_OPTION_NAME); + try { + LOG.debug("JDBC driver: '{}'", jdbcDriver); + Class.forName(jdbcDriver); + } catch (ClassNotFoundException e) { + throw new RuntimeException(e); + } + + // Required parameter. Can be auto-overwritten by user options + jdbcUrl = configuration.get(JDBC_URL_PROPERTY_NAME); + assertMandatoryParameter(jdbcUrl, JDBC_URL_PROPERTY_NAME, JDBC_URL_OPTION_NAME); + + // Required metadata + String dataSource = context.getDataSource(); + if (StringUtils.isBlank(dataSource)) { + throw new IllegalArgumentException("Data source must be provided"); + } + + // Determine if the datasource is a table name or a query name + if (dataSource.startsWith(QUERY_NAME_PREFIX)) { + queryName = dataSource.substring(QUERY_NAME_PREFIX_LENGTH); + if (StringUtils.isBlank(queryName)) { + throw new IllegalArgumentException(String.format("Query name is not provided in data source [%s]", dataSource)); + } + LOG.debug("Query name is {}", queryName); + } else { + tableName = dataSource; + LOG.debug("Table name is {}", tableName); + } + + // Required metadata + columns = context.getTupleDescription(); + + // Optional parameters + batchSizeIsSetByUser = configuration.get(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME) != null; + if (context.getRequestType() == RequestContext.RequestType.WRITE_BRIDGE) { + batchSize = configuration.getInt(JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, DEFAULT_BATCH_SIZE); + + if (batchSize == 0) { + batchSize = 1; // if user set to 0, it is the same as batchSize of 1 + } else if (batchSize < 0) { + throw new IllegalArgumentException(String.format( + "Property %s has incorrect value %s : must be a non-negative integer", JDBC_STATEMENT_BATCH_SIZE_PROPERTY_NAME, batchSize)); + } + } + + // determine fetchSize for read operations, with different default values for MySQL driver and all others + int defaultFetchSize = jdbcDriver.startsWith(MYSQL_DRIVER_PREFIX) ? DEFAULT_MYSQL_FETCH_SIZE : DEFAULT_FETCH_SIZE; + fetchSize = configuration.getInt(JDBC_STATEMENT_FETCH_SIZE_PROPERTY_NAME, defaultFetchSize); + LOG.debug("Will be using fetchSize {}", fetchSize); + + poolSize = context.getOption("POOL_SIZE", DEFAULT_POOL_SIZE); + + String queryTimeoutString = configuration.get(JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME); + if (StringUtils.isNotBlank(queryTimeoutString)) { + try { + queryTimeout = Integer.parseUnsignedInt(queryTimeoutString); + } catch (NumberFormatException e) { + throw new IllegalArgumentException(String.format( + "Property %s has incorrect value %s : must be a non-negative integer", + JDBC_STATEMENT_QUERY_TIMEOUT_PROPERTY_NAME, queryTimeoutString), e); + } + } + + // Optional parameter. The default value is null + String quoteColumnsRaw = context.getOption("QUOTE_COLUMNS"); + if (quoteColumnsRaw != null) { + quoteColumns = Boolean.parseBoolean(quoteColumnsRaw); + } + + // Optional parameter. The default value is empty map + sessionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_SESSION_PROPERTY_PREFIX)); + // Check forbidden symbols + // Note: PreparedStatement enables us to skip this check: its values are distinct from its SQL code + // However, SET queries cannot be executed this way. This is why we do this check + if (sessionConfiguration.entrySet().stream() + .anyMatch( + entry -> + StringUtils.containsAny( + entry.getKey(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS + ) || + StringUtils.containsAny( + entry.getValue(), FORBIDDEN_SESSION_PROPERTY_CHARACTERS + ) + ) + ) { + throw new IllegalArgumentException("Some session configuration parameter contains forbidden characters"); + } + if (LOG.isDebugEnabled()) { + LOG.debug("Session configuration: {}", + sessionConfiguration.entrySet().stream() + .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") + .collect(Collectors.joining(", ")) + ); + } + + // Optional parameter. The default value is empty map + connectionConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_PROPERTY_PREFIX)); + + // Optional parameter. The default value depends on the database + String transactionIsolationString = configuration.get(JDBC_CONNECTION_TRANSACTION_ISOLATION, "NOT_PROVIDED"); + transactionIsolation = TransactionIsolation.typeOf(transactionIsolationString); + + // Set optional user parameter, taking into account impersonation setting for the server. + String jdbcUser = configuration.get(JDBC_USER_PROPERTY_NAME); + boolean impersonationEnabledForServer = configuration.getBoolean(CONFIG_KEY_SERVICE_USER_IMPERSONATION, false); + LOG.debug("JDBC impersonation is {}enabled for server {}", impersonationEnabledForServer ? "" : "not ", context.getServerName()); + if (impersonationEnabledForServer) { + if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { + // secure impersonation for Hive JDBC driver requires setting URL fragment that cannot be overwritten by properties + String updatedJdbcUrl = HiveJdbcUtils.updateImpersonationPropertyInHiveJdbcUrl(jdbcUrl, context.getUser()); + LOG.debug("Replaced JDBC URL {} with {}", jdbcUrl, updatedJdbcUrl); + jdbcUrl = updatedJdbcUrl; + } else { + // the jdbcUser is the GPDB user + jdbcUser = context.getUser(); + } + } + if (jdbcUser != null && !jdbcUser.isEmpty()) { + LOG.debug("Effective JDBC user {}", jdbcUser); + connectionConfiguration.setProperty("user", jdbcUser); + } + + if (LOG.isDebugEnabled()) { + LOG.debug("Connection configuration: {}", + connectionConfiguration.entrySet().stream() + .map(entry -> "'" + entry.getKey() + "'='" + entry.getValue() + "'") + .collect(Collectors.joining(", ")) + ); + } + + // This must be the last parameter parsed, as we output connectionConfiguration earlier + // Optional parameter. By default, corresponding connectionConfiguration property is not set + String jdbcPassword = configuration.get(JDBC_PASSWORD_PROPERTY_NAME); + if (jdbcUser != null) { + if (jdbcPassword != null && !jdbcPassword.isEmpty()) { + LOG.debug("Connection password: {}", ConnectionManager.maskPassword(jdbcPassword)); + connectionConfiguration.setProperty("password", jdbcPassword); + } + } + + assertJDBC(jdbcUrl, jdbcUser, jdbcPassword); + + // connection pool is optional, enabled by default + isConnectionPoolUsed = configuration.getBoolean(JDBC_CONNECTION_POOL_ENABLED_PROPERTY_NAME, true); + LOG.debug("Connection pool is {}enabled", isConnectionPoolUsed ? "" : "not "); + if (isConnectionPoolUsed) { + poolConfiguration = new Properties(); + // for PXF upgrades where jdbc-site template has not been updated, make sure there're sensible defaults + poolConfiguration.setProperty("maximumPoolSize", "15"); + poolConfiguration.setProperty("connectionTimeout", "30000"); + poolConfiguration.setProperty("idleTimeout", "30000"); + poolConfiguration.setProperty("minimumIdle", "0"); + // apply values read from the template + poolConfiguration.putAll(getPropsWithPrefix(configuration, JDBC_CONNECTION_POOL_PROPERTY_PREFIX)); + + // packaged Hive JDBC Driver does not support connection.isValid() method, so we need to force set + // connectionTestQuery parameter in this case, unless already set by the user + if (jdbcUrl.startsWith(HIVE_URL_PREFIX) && HIVE_DEFAULT_DRIVER_CLASS.equals(jdbcDriver) && poolConfiguration.getProperty("connectionTestQuery") == null) { + poolConfiguration.setProperty("connectionTestQuery", "SELECT 1"); + } + + // get the qualifier for connection pool, if configured. Might be used when connection session authorization is employed + // to switch effective user once connection is established + poolQualifier = configuration.get(JDBC_POOL_QUALIFIER_PROPERTY_NAME); + } + + // Optional parameter to determine if the year might contain more than 4 digits in `date` or 'timestamp'. + // The default value is false. + isDateWideRange = configuration.getBoolean(JDBC_DATE_WIDE_RANGE, false); + } + + /** + * Open a new JDBC connection + * + * @return {@link Connection} + * @throws SQLException if a database access or connection error occurs + */ + public Connection getConnection() throws SQLException { + LOG.debug("Requesting a new JDBC connection. URL={} table={} txid:seg={}:{}", jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); + + Connection connection = null; + try { + connection = getConnectionInternal(); + LOG.debug("Obtained a JDBC connection {} for URL={} table={} txid:seg={}:{}", connection, jdbcUrl, tableName, context.getTransactionId(), context.getSegmentId()); + + prepareConnection(connection); + } catch (Exception e) { + closeConnection(connection); + if (e instanceof SQLException) { + throw (SQLException) e; + } else { + String msg = e.getMessage(); + if (msg == null) { + Throwable t = e.getCause(); + if (t != null) msg = t.getMessage(); + } + throw new SQLException(msg, e); + } + } + + return connection; + } + + /** + * Asserts whether all required JDBC URL params/properties set, throws IllegalArgumentException otherwise + * + * @param jdbcUrl connection string + * @param username username from properties (outside of connection string) + * @param password password from properties (outside of connection string) + */ + private void assertJDBC(String jdbcUrl, String username, String password) { + if (!StringUtils.startsWith(jdbcUrl, POSTGRESQL_URL_PREFIX)) { + return; + } + + // For PostgreSQL we enforce users to specify username and password + Properties urlProperties = Driver.parseURL(jdbcUrl, null); + if (urlProperties == null) { + throw new IllegalArgumentException("Invalid PostgreSQL JDBC URL format provided: " + jdbcUrl); + } + + String urlUsername = urlProperties.getProperty(PGProperty.USER.getName()); + if (StringUtils.isEmpty(username) && StringUtils.isEmpty(urlUsername)) { + LOG.debug("PostgreSQL JDBC user has not been set"); + throw new IllegalArgumentException("PostgreSQL JDBC user has not been set"); + } + + String urlPassword = urlProperties.getProperty(PGProperty.PASSWORD.getName()); + if (StringUtils.isEmpty(password) && StringUtils.isEmpty(urlPassword)) { + throw new IllegalArgumentException("PostgreSQL JDBC password has not been set"); + } + + // Ok. It is fine to work with this postgres datasource +} + + /** + * Prepare a JDBC PreparedStatement + * + * @param connection connection to use for creating the statement + * @param query query to execute + * @return PreparedStatement + * @throws SQLException if a database access error occurs + */ + public PreparedStatement getPreparedStatement(Connection connection, String query) throws SQLException { + if ((connection == null) || (query == null)) { + throw new IllegalArgumentException("The provided query or connection is null"); + } + PreparedStatement statement = connection.prepareStatement(query); + if (queryTimeout != null) { + LOG.debug("Setting query timeout to {} seconds", queryTimeout); + statement.setQueryTimeout(queryTimeout); + } + return statement; + } + + /** + * Close a JDBC statement and underlying {@link Connection} + * + * @param statement statement to close + * @throws SQLException throws when a SQLException occurs + */ + public static void closeStatementAndConnection(Statement statement) throws SQLException { + if (statement == null) { + LOG.warn("Call to close statement and connection is ignored as statement provided was null"); + return; + } + + SQLException exception = null; + Connection connection = null; + + try { + connection = statement.getConnection(); + } catch (SQLException e) { + LOG.error("Exception when retrieving Connection from Statement", e); + exception = e; + } + + try { + LOG.debug("Closing statement for connection {}", connection); + statement.close(); + } catch (SQLException e) { + LOG.error("Exception when closing Statement", e); + exception = e; + } + + try { + closeConnection(connection); + } catch (SQLException e) { + LOG.error(String.format("Exception when closing connection %s", connection), e); + exception = e; + } + + if (exception != null) { + throw exception; + } + } + + /** + * For a Kerberized Hive JDBC connection, it creates a connection as the loginUser. + * Otherwise, it returns a new connection. + * + * @return for a Kerberized Hive JDBC connection, returns a new connection as the loginUser. + * Otherwise, it returns a new connection. + * @throws Exception throws when an error occurs + */ + private Connection getConnectionInternal() throws Exception { + Configuration configuration = context.getConfiguration(); + if (Utilities.isSecurityEnabled(configuration) && StringUtils.startsWith(jdbcUrl, HIVE_URL_PREFIX)) { + return secureLogin.getLoginUser(context, configuration).doAs((PrivilegedExceptionAction) () -> + connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier)); + } else { + return connectionManager.getConnection(context.getServerName(), jdbcUrl, connectionConfiguration, isConnectionPoolUsed, poolConfiguration, poolQualifier); + } + } + + /** + * Close a JDBC connection + * + * @param connection connection to close + * @throws SQLException throws when a SQLException occurs + */ + protected static void closeConnection(Connection connection) throws SQLException { + if (connection == null) { + LOG.warn("Call to close connection is ignored as connection provided was null"); + return; + } + try { + if (!connection.isClosed() && + connection.getMetaData().supportsTransactions() && + !connection.getAutoCommit()) { + + LOG.debug("Committing transaction (as part of connection.close()) on connection {}", connection); + connection.commit(); + } + } finally { + try { + LOG.debug("Closing connection {}", connection); + connection.close(); + } catch (Exception e) { + // ignore + LOG.warn(String.format("Failed to close JDBC connection %s, ignoring the error.", connection), e); + } + } + } + + /** + * Prepare JDBC connection by setting session-level variables in external database + * + * @param connection {@link Connection} to prepare + */ + private void prepareConnection(Connection connection) throws SQLException { + if (connection == null) { + throw new IllegalArgumentException("The provided connection is null"); + } + + DatabaseMetaData metadata = connection.getMetaData(); + + // Handle optional connection transaction isolation level + if (transactionIsolation != TransactionIsolation.NOT_PROVIDED) { + // user wants to set isolation level explicitly + if (metadata.supportsTransactionIsolationLevel(transactionIsolation.getLevel())) { + LOG.debug("Setting transaction isolation level to {} on connection {}", transactionIsolation.toString(), connection); + connection.setTransactionIsolation(transactionIsolation.getLevel()); + } else { + throw new RuntimeException( + String.format("Transaction isolation level %s is not supported", transactionIsolation.toString()) + ); + } + } + + // Disable autocommit + if (metadata.supportsTransactions()) { + LOG.debug("Setting autoCommit to false on connection {}", connection); + connection.setAutoCommit(false); + } + + // Prepare session (process sessionConfiguration) + if (!sessionConfiguration.isEmpty()) { + DbProduct dbProduct = DbProduct.getDbProduct(metadata.getDatabaseProductName()); + + try (Statement statement = connection.createStatement()) { + for (Map.Entry e : sessionConfiguration.entrySet()) { + String sessionQuery = dbProduct.buildSessionQuery(e.getKey(), e.getValue()); + LOG.debug("Executing statement {} on connection {}", sessionQuery, connection); + statement.execute(sessionQuery); + } + } + } + } + + /** + * Asserts whether a given parameter has non-empty value, throws IllegalArgumentException otherwise + * + * @param value value to check + * @param paramName parameter name + * @param optionName name of the option for a given parameter + */ + private void assertMandatoryParameter(String value, String paramName, String optionName) { + if (StringUtils.isBlank(value)) { + throw new IllegalArgumentException(String.format( + "Required parameter %s is missing or empty in jdbc-site.xml and option %s is not specified in table definition.", paramName, optionName) + ); + } + } + + /** + * Constructs a mapping of configuration and includes all properties that start with the specified + * configuration prefix. Property names in the mapping are trimmed to remove the configuration prefix. + * This is a method from Hadoop's Configuration class ported here to make older and custom versions of Hadoop + * work with JDBC profile. + * + * @param configuration configuration map + * @param confPrefix configuration prefix + * @return mapping of configuration properties with prefix stripped + */ + private Map getPropsWithPrefix(Configuration configuration, String confPrefix) { + Map configMap = new HashMap<>(); + for (Map.Entry stringStringEntry : configuration) { + String propertyName = stringStringEntry.getKey(); + if (propertyName.startsWith(confPrefix)) { + // do not use value from the iterator as it might not come with variable substitution + String value = configuration.get(propertyName); + String keyName = propertyName.substring(confPrefix.length()); + configMap.put(keyName, value); + } + } + return configMap; + } + +} From c63dc68745fb6b147d1b010232028cb3ed1400ce Mon Sep 17 00:00:00 2001 From: Nikolay Antonov Date: Fri, 17 Jul 2026 17:10:22 +0300 Subject: [PATCH 4/5] fix test --- .../apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java index 2bcb781dc..8ff6010c2 100644 --- a/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java +++ b/server/pxf-jdbc/src/test/java/org/apache/cloudberry/pxf/plugins/jdbc/JdbcBasePluginTest.java @@ -380,7 +380,7 @@ public void testGetConnectionErrorWithoutPassword2() throws SQLException { @Test public void testGetConnectionErrorWithoutPassword3() throws SQLException { - configuration.set("jdbc.driver", "org.greenplum.pxf.plugins.jdbc.FakeJdbcDriver"); + configuration.set("jdbc.driver", "org.apache.cloudberry.pxf.plugins.jdbc.FakeJdbcDriver"); configuration.set("jdbc.url", "jdbc:sqlserver://localhost:1433;databaseName=db1;user=admin"); context.setServerName("test-server"); From ed5b36897f5023fcc7c1019302932919a92bcdc2 Mon Sep 17 00:00:00 2001 From: Nikolay Antonov Date: Fri, 17 Jul 2026 20:10:20 +0300 Subject: [PATCH 5/5] an attempt to fix test --- automation/Makefile | 8 ++++---- .../pxf/automation/applications/PXFApplication.java | 8 ++++---- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/automation/Makefile b/automation/Makefile index 053135562..ea35770c0 100755 --- a/automation/Makefile +++ b/automation/Makefile @@ -141,8 +141,8 @@ sync_jdbc_config: cp $(TEMPLATES_DIR)/jdbc-site.xml $(PXF_BASE_SERVERS)/database/; \ sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_DRIVER_CLASS_NAME|org.postgresql.Driver|" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_URL|jdbc:postgresql://$(JDBC_HOST):$(JDBC_PORT)/pxfautomation|" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ - sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_USER||" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ - sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_PASSWORD||" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ + sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_USER|gpadmin|" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ + sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_PASSWORD|gpadmin|" $(PXF_BASE_SERVERS)/database/jdbc-site.xml; \ cp $(PXF_BASE_SERVERS)/database/jdbc-site.xml $(PXF_BASE_SERVERS)/database/testuser-user.xml; \ sed $(SED_OPTS) "s|pxfautomation|template1|" $(PXF_BASE_SERVERS)/database/testuser-user.xml; \ fi @@ -152,8 +152,8 @@ sync_jdbc_config: cp $(TEMPLATES_DIR)/jdbc-site.xml $(PXF_BASE_SERVERS)/db-session-params/; \ sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_DRIVER_CLASS_NAME|org.postgresql.Driver|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_URL|jdbc:postgresql://$(JDBC_HOST):$(JDBC_PORT)/pxfautomation|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ - sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_USER||" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ - sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_PASSWORD||" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ + sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_USER|gpadmin|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ + sed $(SED_OPTS) "s|YOUR_DATABASE_JDBC_PASSWORD|gpadmin|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ sed $(SED_OPTS) "s||jdbc.session.property.client_min_messagesdebug1|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ sed $(SED_OPTS) "s||jdbc.session.property.default_statistics_target123|" $(PXF_BASE_SERVERS)/db-session-params/jdbc-site.xml; \ fi diff --git a/automation/src/main/java/org/apache/cloudberry/pxf/automation/applications/PXFApplication.java b/automation/src/main/java/org/apache/cloudberry/pxf/automation/applications/PXFApplication.java index ea67fb28f..dc9070a4b 100644 --- a/automation/src/main/java/org/apache/cloudberry/pxf/automation/applications/PXFApplication.java +++ b/automation/src/main/java/org/apache/cloudberry/pxf/automation/applications/PXFApplication.java @@ -52,8 +52,8 @@ public void configureJdbcServers() throws IOException, InterruptedException { "cp ${TEMPLATES_DIR}/jdbc-site.xml ${PXF_BASE_SERVERS}/database/", "sed -i 's|YOUR_DATABASE_JDBC_DRIVER_CLASS_NAME|org.postgresql.Driver|' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", "sed -i 's|YOUR_DATABASE_JDBC_URL|jdbc:postgresql://localhost:7000/pxfautomation|' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", - "sed -i 's|YOUR_DATABASE_JDBC_USER||' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", - "sed -i 's|YOUR_DATABASE_JDBC_PASSWORD||' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", + "sed -i 's|YOUR_DATABASE_JDBC_USER|gpadmin|' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", + "sed -i 's|YOUR_DATABASE_JDBC_PASSWORD|gpadmin|' ${PXF_BASE_SERVERS}/database/jdbc-site.xml", "cp ${PXF_BASE_SERVERS}/database/jdbc-site.xml ${PXF_BASE_SERVERS}/database/testuser-user.xml", "sed -i 's|pxfautomation|template1|' ${PXF_BASE_SERVERS}/database/testuser-user.xml", "cp /home/gpadmin/workspace/cloudberry-pxf/automation/src/test/resources/report.sql ${PXF_BASE_SERVERS}/database/", @@ -62,8 +62,8 @@ public void configureJdbcServers() throws IOException, InterruptedException { "cp ${TEMPLATES_DIR}/jdbc-site.xml ${PXF_BASE_SERVERS}/db-session-params/", "sed -i 's|YOUR_DATABASE_JDBC_DRIVER_CLASS_NAME|org.postgresql.Driver|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", "sed -i 's|YOUR_DATABASE_JDBC_URL|jdbc:postgresql://localhost:7000/pxfautomation|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", - "sed -i 's|YOUR_DATABASE_JDBC_USER||' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", - "sed -i 's|YOUR_DATABASE_JDBC_PASSWORD||' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", + "sed -i 's|YOUR_DATABASE_JDBC_USER|gpadmin|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", + "sed -i 's|YOUR_DATABASE_JDBC_PASSWORD|gpadmin|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", "sed -i 's||jdbc.session.property.client_min_messagesdebug1|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml", "sed -i 's||jdbc.session.property.default_statistics_target123|' ${PXF_BASE_SERVERS}/db-session-params/jdbc-site.xml",