Skip to content

PHOENIX-6262 Bulk Load have a bug in lowercase tablename#1072

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wangchao316 wants to merge 2 commits into from
Closed

PHOENIX-6262 Bulk Load have a bug in lowercase tablename #1072

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4faab9d
PHOENIX-6050 Set properties is invalid in client
wangchao316 Jan 6, 2021
b003cfd
PHOENIX-6262 Bulk Load have a bug in lowercase tablename
wangchao316 Jan 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 43 additions & 21 deletions phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -23,13 +23,19 @@
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;
import java.util.concurrent.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.annotation.concurrent.GuardedBy;

import org.apache.phoenix.thirdparty.com.google.common.cache.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
Expand All@@ -40,6 +46,11 @@
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesImpl;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.thirdparty.com.google.common.cache.Cache;
import org.apache.phoenix.thirdparty.com.google.common.cache.CacheBuilder;
import org.apache.phoenix.thirdparty.com.google.common.cache.RemovalListener;
import org.apache.phoenix.thirdparty.com.google.common.cache.RemovalNotification;
import org.apache.phoenix.util.ReadOnlyProps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand DownExpand Up@@ -124,7 +135,8 @@ public void run() {
throw e;
}
} catch (SQLException e) {
throw new IllegalStateException("Unable to register " + PhoenixDriver.class.getName() + ": "+ e.getMessage());
throw new IllegalStateException("Unable to register "
+ PhoenixDriver.class.getName() + ": " + e.getMessage());
}
}

Expand All@@ -148,21 +160,23 @@ public PhoenixDriver() { // for Squirrel
}

private Cache<ConnectionInfo, ConnectionQueryServices> initializeConnectionCache() {
Configuration config = HBaseFactoryProvider.getConfigurationFactory().getConfiguration();
Configuration config = HBaseFactoryProvider.getConfigurationFactory()
.getConfiguration();
int maxCacheDuration = config.getInt(QueryServices.CLIENT_CONNECTION_CACHE_MAX_DURATION_MILLISECONDS,
QueryServicesOptions.DEFAULT_CLIENT_CONNECTION_CACHE_MAX_DURATION);
RemovalListener<ConnectionInfo, ConnectionQueryServices> cacheRemovalListener =
new RemovalListener<ConnectionInfo, ConnectionQueryServices>() {
@Override
public void onRemoval(RemovalNotification<ConnectionInfo, ConnectionQueryServices> notification) {
String connInfoIdentifier = notification.getKey().toString();
LOGGER.debug("Expiring " + connInfoIdentifier + " because of "
+ notification.getCause().name());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Expiring " + connInfoIdentifier + " because of "
+ notification.getCause().name());
}

try {
notification.getValue().close();
}
catch (SQLException se) {
} catch (SQLException se) {
LOGGER.error("Error while closing expired cache connection " + connInfoIdentifier, se);
}
}
Expand All@@ -182,7 +196,7 @@ public void onRemoval(RemovalNotification<ConnectionInfo, ConnectionQueryService


@Override
public QueryServices getQueryServices() throws SQLException {
public QueryServices getQueryServices(final Properties info) throws SQLException {
try {
lockInterruptibly(LockMode.READ);
checkClosed();
Expand All@@ -193,8 +207,18 @@ public QueryServices getQueryServices() throws SQLException {
if (result == null) {
synchronized(this) {
result = services;
if(result == null) {
services = result = new QueryServicesImpl(getDefaultProps());
if (result == null) {
Configuration config = HBaseFactoryProvider.getConfigurationFactory().getConfiguration();
if (info != null) {
for (Object key : info.keySet()) {
config.set((String) key, info.getProperty((String) key));
}
}
ReadOnlyProps props = new ReadOnlyProps(config.iterator());
QueryServicesOptions queryServicesOptions = QueryServicesOptions.withDefaults()
.setAll(props);
result = new QueryServicesImpl(getDefaultProps(), queryServicesOptions);
services = result;
}
}
}
Expand All@@ -213,7 +237,7 @@ public boolean acceptsURL(String url) throws SQLException {
@Override
public Connection connect(String url, Properties info) throws SQLException {
if (!acceptsURL(url)) {
return null;
return null;
}
try {
lockInterruptibly(LockMode.READ);
Expand All@@ -232,20 +256,20 @@ protected ConnectionQueryServices getConnectionQueryServices(String url, final P
ConnectionInfo connInfo = ConnectionInfo.create(url);
SQLException sqlE = null;
boolean success = false;
final QueryServices services = getQueryServices();
final QueryServices queryServices = getQueryServices(info);
ConnectionQueryServices connectionQueryServices = null;
// Also performs the Kerberos login if the URL/properties request this
final ConnectionInfo normalizedConnInfo = connInfo.normalize(services.getProps(), info);
final ConnectionInfo normalizedConnInfo = connInfo.normalize(queryServices.getProps(), info);
try {
connectionQueryServices =
connectionQueryServicesCache.get(normalizedConnInfo, new Callable<ConnectionQueryServices>() {
@Override
public ConnectionQueryServices call() throws Exception {
ConnectionQueryServices connectionQueryServices;
if (normalizedConnInfo.isConnectionless()) {
connectionQueryServices = new ConnectionlessQueryServicesImpl(services, normalizedConnInfo, info);
connectionQueryServices = new ConnectionlessQueryServicesImpl(queryServices, normalizedConnInfo, info);
} else {
connectionQueryServices = new ConnectionQueryServicesImpl(services, normalizedConnInfo, info);
connectionQueryServices = new ConnectionQueryServicesImpl(queryServices, normalizedConnInfo, info);
}

return connectionQueryServices;
Expand All@@ -254,17 +278,15 @@ public ConnectionQueryServices call() throws Exception {

connectionQueryServices.init(url, info);
success = true;
} catch (ExecutionException ee){
} catch (ExecutionException ee){
if (ee.getCause() instanceof SQLException) {
sqlE = (SQLException) ee.getCause();
} else {
throw new SQLException(ee);
}
}
catch (SQLException e) {
} catch (SQLException e) {
sqlE = e;
}
finally {
} finally {
if (!success) {
// Remove from map, as initialization failed
connectionQueryServicesCache.invalidate(normalizedConnInfo);
Expand Down
85 changes: 48 additions & 37 deletions phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -53,8 +53,6 @@
import org.apache.phoenix.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;



/**
*
* Abstract base class for JDBC Driver implementation of Phoenix
Expand DownExpand Up@@ -91,8 +89,15 @@ public abstract class PhoenixEmbeddedDriver implements Driver, SQLCloseable {
protected ReadOnlyProps getDefaultProps() {
return DEFAULT_PROPS;
}

abstract public QueryServices getQueryServices() throws SQLException;

/**
* get query services
* @param info properties info
* @return query services
* @throws SQLException if failed to get query service.
*/
public abstract QueryServices getQueryServices(Properties info)
throws SQLException;

@Override
public boolean acceptsURL(String url) throws SQLException {
Expand DownExpand Up@@ -139,11 +144,11 @@ public Connection connect(String url, Properties info) throws SQLException {
}

protected final Connection createConnection(String url, Properties info) throws SQLException {
Properties augmentedInfo = PropertiesUtil.deepCopy(info);
augmentedInfo.putAll(getDefaultProps().asMap());
ConnectionQueryServices connectionServices = getConnectionQueryServices(url, augmentedInfo);
PhoenixConnection connection = connectionServices.connect(url, augmentedInfo);
return connection;
Properties augmentedInfo = PropertiesUtil.deepCopy(info);
augmentedInfo.putAll(getDefaultProps().asMap());
ConnectionQueryServices connectionServices = getConnectionQueryServices(url, augmentedInfo);
PhoenixConnection connection = connectionServices.connect(url, augmentedInfo);
return connection;
}

/**
Expand DownExpand Up@@ -199,14 +204,19 @@ public static class ConnectionInfo {
private static final Object KERBEROS_LOGIN_LOCK = new Object();
private static final char WINDOWS_SEPARATOR_CHAR = '\\';
private static final String REALM_EQUIVALENCY_WARNING_MSG = "Provided principal does not contan a realm and the default realm cannot be determined. Ignoring realm equivalency check.";

private static SQLException getMalFormedUrlException(String url) {
return new SQLExceptionInfo.Builder(SQLExceptionCode.MALFORMED_CONNECTION_URL)
.setMessage(url).build().buildException();
}

public String getZookeeperConnectionString() {
return getZookeeperQuorum() + ":" + getPort();
}

/**
* get zookeeper connection info
* @return
*/
public String getZookeeperConnectionString() {
return getZookeeperQuorum() + ":" + getPort();
}

/**
* Detect url with quorum:1,quorum:2 as HBase does not handle different port numbers
Expand DownExpand Up@@ -240,8 +250,8 @@ public static ConnectionInfo create(String url) throws SQLException {
String[] tokens = new String[5];
String token = null;
while (tokenizer.hasMoreTokens() &&
!(token=tokenizer.nextToken()).equals(TERMINATOR) &&
tokenizer.hasMoreTokens() && nTokens < tokens.length) {
!(token = tokenizer.nextToken()).equals(TERMINATOR)
&& tokenizer.hasMoreTokens() && nTokens < tokens.length) {
token = tokenizer.nextToken();
// This would mean we have an empty string for a token which is illegal
if (DELIMITERS.contains(token)) {
Expand All@@ -253,11 +263,11 @@ public static ConnectionInfo create(String url) throws SQLException {
if (tokenizer.hasMoreTokens() && !TERMINATOR.equals(token)) {
String extraToken = tokenizer.nextToken();
if (WINDOWS_SEPARATOR_CHAR == extraToken.charAt(0)) {
String prevToken = tokens[nTokens - 1];
tokens[nTokens - 1] = prevToken + ":" + extraToken;
if (tokenizer.hasMoreTokens() && !(token=tokenizer.nextToken()).equals(TERMINATOR)) {
throw getMalFormedUrlException(url);
}
String prevToken = tokens[nTokens - 1];
tokens[nTokens - 1] = prevToken + ":" + extraToken;
if (tokenizer.hasMoreTokens() && !(token = tokenizer.nextToken()).equals(TERMINATOR)) {
throw getMalFormedUrlException(url);
}
} else {
throw getMalFormedUrlException(url);
}
Expand DownExpand Up@@ -304,7 +314,7 @@ public static ConnectionInfo create(String url) throws SQLException {
}
}
}
return new ConnectionInfo(quorum,port,rootNode, principal, keytabFile);
return new ConnectionInfo(quorum,port,rootNode, principal, keytabFile);
}

public ConnectionInfo normalize(ReadOnlyProps props, Properties info) throws SQLException {
Expand DownExpand Up@@ -347,15 +357,15 @@ public ConnectionInfo normalize(ReadOnlyProps props, Properties info) throws SQL
throw new SQLExceptionInfo.Builder(SQLExceptionCode.MALFORMED_CONNECTION_URL)
.setMessage("Root node may not be specified when using the connectionless url \"" + this.toString() + "\"").build().buildException();
}
if(principal == null){
if(principal == null){
if (!isConnectionless) {
principal = props.get(QueryServices.HBASE_CLIENT_PRINCIPAL);
}
principal = props.get(QueryServices.HBASE_CLIENT_PRINCIPAL);
}
}
if(keytab == null){
if (!isConnectionless) {
keytab = props.get(QueryServices.HBASE_CLIENT_KEYTAB);
}
if(keytab == null){
if (!isConnectionless) {
keytab = props.get(QueryServices.HBASE_CLIENT_KEYTAB);
}
}
if (!isConnectionless()) {
boolean credsProvidedInUrl = null != principal && null != keytab;
Expand DownExpand Up@@ -463,7 +473,7 @@ static boolean isSameName(String currentName, String newName, String hostname, S
private Configuration getConfiguration(ReadOnlyProps props, Properties info, String principal, String keytab) {
final Configuration config = HBaseFactoryProvider.getConfigurationFactory().getConfiguration();
// Add QueryServices properties
for (Entry<String,String> entry : props) {
for (Entry<String,String> entry : props) {
config.set(entry.getKey(), entry.getValue());
}
// Add any user-provided properties (via DriverManager)
Expand DownExpand Up@@ -508,7 +518,7 @@ public ConnectionInfo(String zookeeperQuorum, Integer port, String rootNode, Str
}

public ConnectionInfo(String zookeeperQuorum, Integer port, String rootNode) {
this(zookeeperQuorum, port, rootNode, null, null);
this(zookeeperQuorum, port, rootNode, null, null);
}

/**
Expand DownExpand Up@@ -608,12 +618,12 @@ public boolean equals(Object obj) {
}

@Override
public String toString() {
return zookeeperQuorum + (port == null ? "" : ":" + port)
+ (rootNode == null ? "" : ":" + rootNode)
+ (principal == null ? "" : ":" + principal)
+ (keytab == null ? "" : ":" + keytab);
}
public String toString() {
return zookeeperQuorum + (port == null ? "" : ":" + port)
+ (rootNode == null ? "" : ":" + rootNode)
+ (principal == null ? "" : ":" + principal)
+ (keytab == null ? "" : ":" + keytab);
}

public String toUrl() {
return PhoenixRuntime.JDBC_PROTOCOL + PhoenixRuntime.JDBC_PROTOCOL_SEPARATOR
Expand All@@ -628,7 +638,8 @@ private static ConnectionInfo defaultConnectionInfo(String url) throws SQLExcept
throw getMalFormedUrlException(url);
}
String clientPort = config.get(HConstants.ZOOKEEPER_CLIENT_PORT);
Integer port = clientPort==null ? null : Integer.parseInt(clientPort);
Integer port = clientPort == null ? null
: Integer.parseInt(clientPort);
if (port == null || port < 0) {
throw getMalFormedUrlException(url);
}
Expand Down
Loading