From 4faab9d52de2c408d99467e610ed2c0791278be6 Mon Sep 17 00:00:00 2001 From: wangchao316 <576749262@qq.com> Date: Wed, 6 Jan 2021 09:20:16 +0800 Subject: [PATCH 1/2] PHOENIX-6050 Set properties is invalid in client --- .../apache/phoenix/jdbc/PhoenixDriver.java | 64 ++++++--- .../phoenix/jdbc/PhoenixEmbeddedDriver.java | 85 +++++++----- .../phoenix/query/QueryServicesImpl.java | 15 +- .../phoenix/jdbc/PhoenixDriverTest.java | 27 ++-- .../phoenix/jdbc/PhoenixTestDriver.java | 17 ++- .../org/apache/phoenix/query/BaseTest.java | 128 +++++++++--------- 6 files changed, 190 insertions(+), 146 deletions(-) diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java index 21264acd65e..33b060af825 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDriver.java @@ -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; @@ -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; @@ -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()); } } @@ -148,7 +160,8 @@ public PhoenixDriver() { // for Squirrel } private Cache 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 cacheRemovalListener = @@ -156,13 +169,14 @@ private Cache initializeConnectionCache @Override public void onRemoval(RemovalNotification 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); } } @@ -182,7 +196,7 @@ public void onRemoval(RemovalNotification() { @@ -243,9 +267,9 @@ protected ConnectionQueryServices getConnectionQueryServices(String url, final P 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; @@ -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); diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java index 989475e3e2f..eaa1cd5e5c8 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixEmbeddedDriver.java @@ -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 @@ -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 { @@ -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; } /** @@ -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 @@ -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)) { @@ -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); } @@ -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 { @@ -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; @@ -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 entry : props) { + for (Entry entry : props) { config.set(entry.getKey(), entry.getValue()); } // Add any user-provided properties (via DriverManager) @@ -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); } /** @@ -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 @@ -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); } diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesImpl.java index 564da60e6ad..ed58ea123a0 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesImpl.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesImpl.java @@ -19,10 +19,6 @@ import org.apache.phoenix.util.ReadOnlyProps; - - - - /** * * Real implementation of QueryServices for use in runtime and perf testing @@ -31,8 +27,13 @@ * @since 0.1 */ public final class QueryServicesImpl extends BaseQueryServicesImpl { - - public QueryServicesImpl(ReadOnlyProps defaultProps) { - super(defaultProps, QueryServicesOptions.withDefaults()); + + /** + * init query services + * @param defaultProps default props + * @param queryServicesOptions options info + */ + public QueryServicesImpl(ReadOnlyProps defaultProps, final QueryServicesOptions queryServicesOptions) { + super(defaultProps, queryServicesOptions); } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixDriverTest.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixDriverTest.java index e7afb30ee29..bd02ee82bd9 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixDriverTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixDriverTest.java @@ -62,8 +62,8 @@ public void testFirstConnectionWhenUrlHasTenantId() throws Exception { @Test public void testMaxMutationSizeSetCorrectly() throws SQLException { Properties connectionProperties = new Properties(); - connectionProperties.setProperty(QueryServices.MAX_MUTATION_SIZE_ATTRIB,"100"); - connectionProperties.setProperty(QueryServices.IMMUTABLE_ROWS_ATTRIB,"100"); + connectionProperties.setProperty(QueryServices.MAX_MUTATION_SIZE_ATTRIB, "100"); + connectionProperties.setProperty(QueryServices.IMMUTABLE_ROWS_ATTRIB, "100"); Connection connection = DriverManager.getConnection(getUrl(), connectionProperties); PreparedStatement stmt = connection.prepareStatement("upsert into " + ATABLE + " (organization_id, entity_id, a_integer) values (?,?,?)"); @@ -83,7 +83,7 @@ public void testMaxMutationSizeSetCorrectly() throws SQLException { @Test public void testMaxMutationSizeInBytesSetCorrectly() throws Exception { Properties connectionProperties = new Properties(); - connectionProperties.setProperty(QueryServices.MUTATE_BATCH_SIZE_BYTES_ATTRIB,"100"); + connectionProperties.setProperty(QueryServices.MUTATE_BATCH_SIZE_BYTES_ATTRIB, "100"); PhoenixConnection connection = (PhoenixConnection) DriverManager.getConnection(getUrl(), connectionProperties); assertEquals(100L, connection.getMutateBatchSizeBytes()); assertEquals(100L, connection.getMutationState().getBatchSizeBytes()); @@ -134,11 +134,20 @@ public void testDisallowIsolationLevel() throws SQLException { @Test public void testInvalidURL() throws Exception { - Class.forName(PhoenixDriver.class.getName()); - try { - DriverManager.getConnection("any text whatever you want to put here"); - fail("Should have failed due to invalid driver"); - } catch(Exception e) { - } + Class.forName(PhoenixDriver.class.getName()); + try { + DriverManager.getConnection("any text whatever you want to put here"); + fail("Should have failed due to invalid driver"); + } catch(Exception e) { + } + } + + @Test + public void testSetPropertiesInvalid() throws SQLException { + Properties connectionProperties = new Properties(); + connectionProperties.setProperty(QueryServices.THREAD_POOL_SIZE_ATTRIB, "300"); + PhoenixDriver phoenixDriver = new PhoenixDriver(); + QueryServices services = phoenixDriver.getQueryServices(connectionProperties); + assertEquals(300, services.getExecutor().getCorePoolSize()); } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java index f9fa9f8acda..31df70c9987 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/jdbc/PhoenixTestDriver.java @@ -32,8 +32,6 @@ import org.apache.phoenix.query.QueryServicesTestImpl; import org.apache.phoenix.util.ReadOnlyProps; - - /** * * JDBC Driver implementation of Phoenix for testing. @@ -44,7 +42,6 @@ */ @ThreadSafe public class PhoenixTestDriver extends PhoenixEmbeddedDriver { - @GuardedBy("this") private ConnectionQueryServices connectionQueryServices; private final ReadOnlyProps overrideProps; @@ -66,7 +63,7 @@ public PhoenixTestDriver(ReadOnlyProps props) { } @Override - public synchronized QueryServices getQueryServices() { + public synchronized QueryServices getQueryServices(Properties info) { checkClosed(); return queryServices; } @@ -86,7 +83,9 @@ public synchronized Connection connect(String url, Properties info) throws SQLEx @Override // public for testing public synchronized ConnectionQueryServices getConnectionQueryServices(String url, Properties info) throws SQLException { checkClosed(); - if (connectionQueryServices != null) { return connectionQueryServices; } + if (connectionQueryServices != null) { + return connectionQueryServices; + } ConnectionInfo connInfo = ConnectionInfo.create(url); if (connInfo.isConnectionless()) { connectionQueryServices = new ConnectionlessQueryServicesImpl(queryServices, connInfo, info); @@ -110,13 +109,17 @@ public synchronized void close() throws SQLException { } closed = true; try { - if (connectionQueryServices != null) connectionQueryServices.close(); + if (connectionQueryServices != null) { + connectionQueryServices.close(); + } } finally { ThreadPoolExecutor executor = queryServices.getExecutor(); try { queryServices.close(); } finally { - if (executor != null) executor.shutdownNow(); + if (executor != null) { + executor.shutdownNow(); + } connectionQueryServices = null; } } diff --git a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java index 48947278537..4240f976ab5 100644 --- a/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java +++ b/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java @@ -184,22 +184,22 @@ public abstract class BaseTest { public static final String DRIVER_CLASS_NAME_ATTRIB = "phoenix.driver.class.name"; private static final double ZERO = 1e-9; - private static final Map tableDDLMap; + private static final Map tableDDLMap; private static final Logger LOGGER = LoggerFactory.getLogger(BaseTest.class); @ClassRule public static TemporaryFolder tmpFolder = new TemporaryFolder(); - private static final int dropTableTimeout = 300; // 5 mins should be long enough. - private static final ThreadFactory factory = new ThreadFactoryBuilder().setDaemon(true) + private static final int DROP_TABLE_TIMEOUT = 300; // 5 mins should be long enough. + private static final ThreadFactory FACTORY = new ThreadFactoryBuilder().setDaemon(true) .setNameFormat("DROP-TABLE-BASETEST" + "-thread-%s").build(); private static final ExecutorService dropHTableService = Executors - .newSingleThreadExecutor(factory); + .newSingleThreadExecutor(FACTORY); @ClassRule public static final SystemExitRule SYSTEM_EXIT_RULE = new SystemExitRule(); static { - ImmutableMap.Builder builder = ImmutableMap.builder(); - builder.put(ENTITY_HISTORY_TABLE_NAME,"create table " + ENTITY_HISTORY_TABLE_NAME + + ImmutableMap.Builder builder = ImmutableMap.builder(); + builder.put(ENTITY_HISTORY_TABLE_NAME, "create table " + ENTITY_HISTORY_TABLE_NAME + " (organization_id char(15) not null,\n" + " parent_id char(15) not null,\n" + " created_date date not null,\n" + @@ -208,7 +208,7 @@ public abstract class BaseTest { " new_value varchar,\n" + //create table shouldn't blow up if the last column definition ends with a comma. " CONSTRAINT pk PRIMARY KEY (organization_id, parent_id, created_date, entity_history_id)\n" + ")"); - builder.put(ENTITY_HISTORY_SALTED_TABLE_NAME,"create table " + ENTITY_HISTORY_SALTED_TABLE_NAME + + builder.put(ENTITY_HISTORY_SALTED_TABLE_NAME, "create table " + ENTITY_HISTORY_SALTED_TABLE_NAME + " (organization_id char(15) not null,\n" + " parent_id char(15) not null,\n" + " created_date date not null,\n" + @@ -217,7 +217,7 @@ public abstract class BaseTest { " new_value varchar\n" + " CONSTRAINT pk PRIMARY KEY (organization_id, parent_id, created_date, entity_history_id))\n" + " SALT_BUCKETS = 4"); - builder.put(ATABLE_NAME,"create table " + ATABLE_NAME + + builder.put(ATABLE_NAME, "create table " + ATABLE_NAME + " (organization_id char(15) not null, \n" + " entity_id char(15) not null,\n" + " a_string varchar(100),\n" + @@ -259,7 +259,7 @@ public abstract class BaseTest { + " a_unsigned_double unsigned_double \n" + " CONSTRAINT pk PRIMARY KEY (organization_id, entity_id)\n" + ")"); - builder.put(BTABLE_NAME,"create table " + BTABLE_NAME + + builder.put(BTABLE_NAME, "create table " + BTABLE_NAME + " (a_string varchar not null, \n" + " a_id char(3) not null,\n" + " b_string varchar not null, \n" + @@ -270,7 +270,7 @@ public abstract class BaseTest { " d_string varchar(3),\n" + " e_string char(10)\n" + " CONSTRAINT my_pk PRIMARY KEY (a_string,a_id,b_string,a_integer,c_string))"); - builder.put(TABLE_WITH_SALTING,"create table " + TABLE_WITH_SALTING + + builder.put(TABLE_WITH_SALTING, "create table " + TABLE_WITH_SALTING + " (a_integer integer not null, \n" + " a_string varchar not null, \n" + " a_id char(3) not null,\n" + @@ -278,30 +278,30 @@ public abstract class BaseTest { " b_integer integer \n" + " CONSTRAINT pk PRIMARY KEY (a_integer, a_string, a_id))\n" + " SALT_BUCKETS = 4"); - builder.put(STABLE_NAME,"create table " + STABLE_NAME + + builder.put(STABLE_NAME, "create table " + STABLE_NAME + " (id char(1) not null primary key,\n" + " \"value\" integer)"); - builder.put(PTSDB_NAME,"create table " + PTSDB_NAME + + builder.put(PTSDB_NAME, "create table " + PTSDB_NAME + " (inst varchar null,\n" + " host varchar null,\n" + " date date not null,\n" + " val decimal(31,10)\n" + " CONSTRAINT pk PRIMARY KEY (inst, host, date))"); - builder.put(PTSDB2_NAME,"create table " + PTSDB2_NAME + + builder.put(PTSDB2_NAME, "create table " + PTSDB2_NAME + " (inst varchar(10) not null,\n" + " date date not null,\n" + " val1 decimal,\n" + " val2 decimal(31,10),\n" + " val3 decimal\n" + " CONSTRAINT pk PRIMARY KEY (inst, date))"); - builder.put(PTSDB3_NAME,"create table " + PTSDB3_NAME + + builder.put(PTSDB3_NAME, "create table " + PTSDB3_NAME + " (host varchar(10) not null,\n" + " date date not null,\n" + " val1 decimal,\n" + " val2 decimal(31,10),\n" + " val3 decimal\n" + " CONSTRAINT pk PRIMARY KEY (host DESC, date DESC))"); - builder.put(FUNKY_NAME,"create table " + FUNKY_NAME + + builder.put(FUNKY_NAME, "create table " + FUNKY_NAME + " (\"foo!\" varchar not null primary key,\n" + " \"1\".\"#@$\" varchar, \n" + " \"1\".\"foo.bar-bas\" varchar, \n" + @@ -310,7 +310,7 @@ public abstract class BaseTest { " \"1\".\"value\" integer,\n" + " \"1\".\"_blah^\" varchar)" ); - builder.put(MULTI_CF_NAME,"create table " + MULTI_CF_NAME + + builder.put(MULTI_CF_NAME, "create table " + MULTI_CF_NAME + " (id char(15) not null primary key,\n" + " a.unique_user_count integer,\n" + " b.unique_org_count integer,\n" + @@ -319,14 +319,14 @@ public abstract class BaseTest { " e.cpu_utilization decimal(31,10),\n" + " f.response_time bigint,\n" + " g.response_time bigint)"); - builder.put(HBASE_DYNAMIC_COLUMNS,"create table " + HBASE_DYNAMIC_COLUMNS + + builder.put(HBASE_DYNAMIC_COLUMNS, "create table " + HBASE_DYNAMIC_COLUMNS + " (entry varchar not null," + " F varchar," + " A.F1v1 varchar," + " A.F1v2 varchar," + " B.F2v1 varchar" + " CONSTRAINT pk PRIMARY KEY (entry))\n"); - builder.put(PRODUCT_METRICS_NAME,"create table " + PRODUCT_METRICS_NAME + + builder.put(PRODUCT_METRICS_NAME, "create table " + PRODUCT_METRICS_NAME + " (organization_id char(15) not null," + " date date not null," + " feature char(1) not null," + @@ -339,7 +339,7 @@ public abstract class BaseTest { " region varchar,\n" + " unset_column decimal(31,10)\n" + " CONSTRAINT pk PRIMARY KEY (organization_id, \"DATE\", feature, UNIQUE_USERS))"); - builder.put(CUSTOM_ENTITY_DATA_FULL_NAME,"create table " + CUSTOM_ENTITY_DATA_FULL_NAME + + builder.put(CUSTOM_ENTITY_DATA_FULL_NAME, "create table " + CUSTOM_ENTITY_DATA_FULL_NAME + " (organization_id char(15) not null, \n" + " key_prefix char(3) not null,\n" + " custom_entity_data_id char(12) not null,\n" + @@ -367,9 +367,9 @@ public abstract class BaseTest { " b.val8 varchar,\n" + " b.val9 varchar\n" + " CONSTRAINT pk PRIMARY KEY (organization_id, key_prefix, custom_entity_data_id))"); - builder.put("IntKeyTest","create table IntKeyTest" + + builder.put("IntKeyTest", "create table IntKeyTest" + " (i integer not null primary key)"); - builder.put("IntIntKeyTest","create table IntIntKeyTest" + + builder.put("IntIntKeyTest", "create table IntIntKeyTest" + " (i integer not null primary key, j integer)"); builder.put("PKIntValueTest", "create table PKIntValueTest" + " (pk integer not null primary key)"); @@ -386,9 +386,9 @@ public abstract class BaseTest { builder.put("KVBigIntValueTest", "create table KVBigIntValueTest" + " (pk integer not null primary key,\n" + " kv bigint)\n"); - builder.put(SUM_DOUBLE_NAME,"create table SumDoubleTest" + + builder.put(SUM_DOUBLE_NAME, "create table SumDoubleTest" + " (id varchar not null primary key, d DOUBLE, f FLOAT, ud UNSIGNED_DOUBLE, uf UNSIGNED_FLOAT, i integer, de decimal)"); - builder.put(BINARY_NAME,"create table " + BINARY_NAME + + builder.put(BINARY_NAME, "create table " + BINARY_NAME + " (a_binary BINARY(16) not null, \n" + " b_binary BINARY(16), \n" + " a_varbinary VARBINARY, \n" + @@ -550,7 +550,7 @@ private static String initMiniCluster(Configuration conf, ReadOnlyProps override try { long startTime = System.currentTimeMillis(); utility.startMiniCluster(NUM_SLAVES_BASE); - long startupTime = System.currentTimeMillis()-startTime; + long startupTime = System.currentTimeMillis() - startTime; LOGGER.info("HBase minicluster startup complete in {} ms", startupTime); return getLocalClusterUrl(utility); } catch (Throwable t) { @@ -587,8 +587,8 @@ private static void setTestConfigForDistribuedCluster(Configuration conf, ReadOn private static void setDefaultTestConfig(Configuration conf, ReadOnlyProps overrideProps) throws Exception { ConfigUtil.setReplicationConfigIfAbsent(conf); - QueryServices services = newTestDriver(overrideProps).getQueryServices(); - for (Entry entry : services.getProps()) { + QueryServices services = newTestDriver(overrideProps).getQueryServices(new Properties()); + for (Entry entry : services.getProps()) { conf.set(entry.getKey(), entry.getValue()); } //no point doing sanity checks when running tests. @@ -599,7 +599,7 @@ private static void setDefaultTestConfig(Configuration conf, ReadOnlyProps overr conf.setLong(HConstants.ZOOKEEPER_TICK_TIME, 6 * 1000); // override any defaults based on overrideProps - for (Entry entry : overrideProps) { + for (Entry entry : overrideProps) { conf.set(entry.getKey(), entry.getValue()); } } @@ -652,7 +652,7 @@ public static Configuration setUpConfigForMiniCluster(Configuration conf, ReadOn private static PhoenixTestDriver newTestDriver(ReadOnlyProps props) throws Exception { PhoenixTestDriver newDriver; String driverClassName = props.get(DRIVER_CLASS_NAME_ATTRIB); - if(isDistributedClusterModeEnabled(config)) { + if (isDistributedClusterModeEnabled(config)) { HashMap distPropMap = new HashMap<>(1); distPropMap.put(DROP_METADATA_ATTRIB, Boolean.TRUE.toString()); props = new ReadOnlyProps(props, distPropMap.entrySet().iterator()); @@ -746,13 +746,13 @@ protected static void ensureTableCreated(String url, String tableName, String ta protected static void ensureTableCreated(String url, String tableName, String tableDDLType, byte[][] splits, Long ts, String tableDDLOptions) throws SQLException { String ddl = tableDDLMap.get(tableDDLType); - if(!tableDDLType.equals(tableName)) { - ddl = ddl.replace(tableDDLType, tableName); + if (!tableDDLType.equals(tableName)) { + ddl = ddl.replace(tableDDLType, tableName); } - if (tableDDLOptions!=null) { + if (tableDDLOptions != null) { ddl += tableDDLOptions; } - createSchema(url,tableName, ts); + createSchema(url, tableName, ts); createTestTable(url, ddl, splits, ts); } @@ -799,7 +799,7 @@ public static void freeResourcesIfBeyondThreshold() throws Exception { if (TABLE_COUNTER.get() > TEARDOWN_THRESHOLD) { int numTables = TABLE_COUNTER.get(); TABLE_COUNTER.set(0); - if(isDistributedClusterModeEnabled(config)) { + if (isDistributedClusterModeEnabled(config)) { LOGGER.info( "Deleting old tables on distributed cluster because number of tables is likely greater than " + TEARDOWN_THRESHOLD); @@ -846,7 +846,7 @@ protected static void createTestTable(String url, String ddl, byte[][] splits, L for (int i = 0; i < splits.length; i++) { buf.append("'").append(Bytes.toString(splits[i])).append("'").append(","); } - buf.setCharAt(buf.length()-1, ')'); + buf.setCharAt(buf.length() - 1, ')'); } ddl = buf.toString(); Properties props = new Properties(); @@ -951,8 +951,7 @@ private static void deletePriorTables(long ts, String tenantId, String url) thro } } } - } - finally { + } finally { conn.close(); } } @@ -1128,19 +1127,19 @@ protected static String initATableValues(String tenantId, byte[][] splits, Date } protected static String initATableValues(String tableName, String tenantId, byte[][] splits, Date date, Long ts, String url, String tableDDLOptions) throws Exception { - if(tableName == null) { + if (tableName == null) { tableName = generateUniqueName(); } String tableDDLType = ATABLE_NAME; if (ts == null) { ensureTableCreated(url, tableName, tableDDLType, splits, null, tableDDLOptions); } else { - ensureTableCreated(url, tableName, tableDDLType, splits, ts-5, tableDDLOptions); + ensureTableCreated(url, tableName, tableDDLType, splits, ts - 5, tableDDLOptions); } Properties props = new Properties(); if (ts != null) { - props.setProperty(CURRENT_SCN_ATTRIB, Long.toString(ts-3)); + props.setProperty(CURRENT_SCN_ATTRIB, Long.toString(ts - 3)); } try (Connection conn = DriverManager.getConnection(url, props)) { @@ -1360,7 +1359,7 @@ private static String initEntityHistoryTableValues(String tableName, String tena if (ts == null) { ensureTableCreated(url, tableName, ENTITY_HISTORY_TABLE_NAME, splits, null); } else { - ensureTableCreated(url, tableName, ENTITY_HISTORY_TABLE_NAME, splits, ts-2, null); + ensureTableCreated(url, tableName, ENTITY_HISTORY_TABLE_NAME, splits, ts - 2, null); } Properties props = new Properties(); @@ -1470,7 +1469,7 @@ protected static String initSaltedEntityHistoryTableValues(String tableName, Str if (ts == null) { ensureTableCreated(url, tableName, ENTITY_HISTORY_SALTED_TABLE_NAME, splits, null); } else { - ensureTableCreated(url, tableName, ENTITY_HISTORY_SALTED_TABLE_NAME, splits, ts-2, null); + ensureTableCreated(url, tableName, ENTITY_HISTORY_SALTED_TABLE_NAME, splits, ts - 2, null); } Properties props = new Properties(); @@ -1607,13 +1606,13 @@ public Void call() throws Exception { return null; } }); - future.get(dropTableTimeout, TimeUnit.SECONDS); + future.get(DROP_TABLE_TIMEOUT, TimeUnit.SECONDS); success = true; } catch (TimeoutException e) { throw new SQLExceptionInfo.Builder(SQLExceptionCode.OPERATION_TIMED_OUT) .setMessage( "Not able to disable and delete table " + tableName.getNameAsString() - + " in " + dropTableTimeout + " seconds.").build().buildException(); + + " in " + DROP_TABLE_TIMEOUT + " seconds.").build().buildException(); } catch (Exception e) { throw e; } @@ -1629,7 +1628,7 @@ public static void assertOneOfValuesEqualsResultSet(ResultSet rs, List result = Lists.newArrayList(); for (int i = 0; i < rs.getMetaData().getColumnCount(); i++) { - result.add(rs.getObject(i+1)); + result.add(rs.getObject(i + 1)); } results.add(result); } @@ -1668,7 +1667,7 @@ public static void assertValuesEqualsResultSet(ResultSet rs, List> while (rs.next() && errorResult == null) { List result = Lists.newArrayList(); for (int i = 0; i < rs.getMetaData().getColumnCount(); i++) { - result.add(rs.getObject(i+1)); + result.add(rs.getObject(i + 1)); } if (!expectedResults.contains(result)) { errorResult = result; @@ -1685,7 +1684,7 @@ public static HBaseTestingUtility getUtility() { } public static void upsertRows(Connection conn, String fullTableName, int numRows) throws SQLException { - for (int i=1; i<=numRows; ++i) { + for (int i = 1; i <= numRows; ++i) { upsertRow(conn, fullTableName, i, false); } } @@ -1694,8 +1693,8 @@ public static void upsertRow(Connection conn, String fullTableName, int index, b String upsert = "UPSERT INTO " + fullTableName + " VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; PreparedStatement stmt = conn.prepareStatement(upsert); - stmt.setString(1, firstRowInBatch ? "firstRowInBatch_" : "" + "varchar"+index); - stmt.setString(2, "char"+index); + stmt.setString(1, firstRowInBatch ? "firstRowInBatch_" : "" + "varchar" + index); + stmt.setString(2, "char" + index); stmt.setInt(3, index); stmt.setLong(4, index); stmt.setBigDecimal(5, new BigDecimal(index)); @@ -1703,15 +1702,15 @@ public static void upsertRow(Connection conn, String fullTableName, int index, b stmt.setDate(6, date); stmt.setString(7, "varchar_a"); stmt.setString(8, "chara"); - stmt.setInt(9, index+1); - stmt.setLong(10, index+1); - stmt.setBigDecimal(11, new BigDecimal(index+1)); + stmt.setInt(9, index + 1); + stmt.setLong(10, index + 1); + stmt.setBigDecimal(11, new BigDecimal(index + 1)); stmt.setDate(12, date); stmt.setString(13, "varchar_b"); stmt.setString(14, "charb"); - stmt.setInt(15, index+2); - stmt.setLong(16, index+2); - stmt.setBigDecimal(17, new BigDecimal(index+2)); + stmt.setInt(15, index + 2); + stmt.setLong(16, index + 2); + stmt.setBigDecimal(17, new BigDecimal(index + 2)); stmt.setDate(18, date); stmt.executeUpdate(); } @@ -1822,7 +1821,7 @@ private static void verifySequence(String tenantID, String sequenceName, String ResultSet rs = phxConn.createStatement().executeQuery(ddl); - if(exists) { + if (exists) { assertTrue(rs.next()); assertEquals(value, rs.getLong(4)); } else { @@ -1846,7 +1845,7 @@ protected static void splitTableSync(Admin admin, TableName hbaseTableName, byte splitSuccessful = false; } } - if(splitSuccessful) { + if (splitSuccessful) { return; } } @@ -1886,14 +1885,14 @@ protected static void splitTable(TableName fullTableName, List splitPoin assertFalse("Balancer must be off", master.isBalancerOn()); AssignmentManager am = master.getAssignmentManager(); // No need to split on the first splitPoint since the end key of region boundaries are exclusive - for (int i=1; i regionInfoList = admin.getRegions(fullTableName); assertEquals(splitPoints.size(), regionInfoList.size()); HashMap> serverToRegionsList = Maps.newHashMapWithExpectedSize(NUM_SLAVES_BASE); Deque availableRegionServers = new ArrayDeque(NUM_SLAVES_BASE); - for (int i=0; i tableRegions = @@ -1913,8 +1912,8 @@ protected static void splitTable(TableName fullTableName, List splitPoin availableRegionServers.isEmpty()); for (Entry> entry : serverToRegionsList.entrySet()) { List regions = entry.getValue(); - if (regions.size()>1) { - for (int i=1; i< regions.size(); ++i) { + if (regions.size() > 1) { + for (int i = 1; i < regions.size(); ++i) { moveRegion(regions.get(i), entry.getKey(), availableRegionServers.pop()); } } @@ -1930,9 +1929,8 @@ protected static void splitTable(TableName fullTableName, List splitPoin ServerName serverName = am.getRegionStates().getRegionServerOfRegion(hRegionInfo); if (!serverNames.contains(serverName)) { serverNames.add(serverName); - } - else { - fail("Multiple regions on "+serverName.getServerName()); + } else { + fail("Multiple regions on " + serverName.getServerName()); } } } @@ -1946,14 +1944,14 @@ protected static void splitTable(TableName fullTableName, List splitPoin protected static void splitSystemCatalog(Map> tenantToTableAndViewMap) throws Exception { List splitPoints = Lists.newArrayListWithExpectedSize(5); // add the rows keys of the table or view metadata rows - Set schemaNameSet=Sets.newHashSetWithExpectedSize(15); + Set schemaNameSet = Sets.newHashSetWithExpectedSize(15); for (Entry> entrySet : tenantToTableAndViewMap.entrySet()) { String tenantId = entrySet.getKey(); for (String fullName : entrySet.getValue()) { String schemaName = SchemaUtil.getSchemaNameFromFullName(fullName); // we don't allow SYSTEM.CATALOG to split within a schema, so to ensure each table // or view is on a separate region they need to have a unique tenant and schema name - assertTrue("Schema names of tables/view must be unique ", schemaNameSet.add(tenantId+"."+schemaName)); + assertTrue("Schema names of tables/view must be unique ", schemaNameSet.add(tenantId + "." + schemaName)); String tableName = SchemaUtil.getTableNameFromFullName(fullName); splitPoints.add( SchemaUtil.getTableKey(tenantId, "".equals(schemaName) ? null : schemaName, tableName)); From b003cfddd80b85ad5b6aedce84bcf80b74ebf330 Mon Sep 17 00:00:00 2001 From: wangchao316 <576749262@qq.com> Date: Thu, 7 Jan 2021 10:29:09 +0800 Subject: [PATCH 2/2] PHOENIX-6262 Bulk Load have a bug in lowercase tablename --- .../mapreduce/AbstractBulkLoadTool.java | 78 +++++----- .../apache/phoenix/util/PhoenixRuntime.java | 139 ++++++++++-------- 2 files changed, 120 insertions(+), 97 deletions(-) diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java index 2abe1965b94..46637db14d5 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/AbstractBulkLoadTool.java @@ -75,8 +75,7 @@ * Base tool for running MapReduce-based ingests of data. */ public abstract class AbstractBulkLoadTool extends Configured implements Tool { - - protected static final Logger LOGGER = LoggerFactory.getLogger(AbstractBulkLoadTool.class); + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractBulkLoadTool.class); static final Option ZK_QUORUM_OPT = new Option("z", "zookeeper", true, "Supply zookeeper connection details (optional)"); static final Option INPUT_PATH_OPT = new Option("i", "input", true, "Input path(s) (comma-separated, mandatory)"); @@ -97,7 +96,8 @@ public abstract class AbstractBulkLoadTool extends Configured implements Tool { * @param conf job configuration */ protected abstract void configureOptions(CommandLine cmdLine, List importColumns, - Configuration conf) throws SQLException; + Configuration conf) throws SQLException; + protected abstract void setupJob(Job job); protected Options getOptions() { @@ -188,13 +188,13 @@ private boolean isStartWithTwoDoubleQuotes (String name) { boolean start = false; boolean end = false; if (name != null && name.length() > 1) { - int length = name.length(); - start = name.substring(0,2).equals("\"\""); - end = name.substring(length-2, length).equals("\"\""); - if (start && !end) { - throw new IllegalArgumentException("Invalid table/schema name " + name + - ". Please check if name end with two double quotes."); - } + int length = name.length(); + start = name.substring(0, 2).equals("\"\""); + end = name.substring(length - 2, length).equals("\"\""); + if (start && !end) { + throw new IllegalArgumentException("Invalid table/schema name " + name + + ". Please check if name end with two double quotes."); + } } return start; } @@ -213,19 +213,31 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception { } boolean quotedSchemaName = isStartWithTwoDoubleQuotes(schemaName); if (quotedSchemaName) { - schemaName = schemaName.substring(1,schemaName.length() - 1); + schemaName = schemaName.substring(1, schemaName.length() - 1); } String qualifiedTableName = SchemaUtil.getQualifiedTableName(schemaName, tableName); String qualifiedIndexTableName = null; - if (indexTableName != null){ + if (indexTableName != null) { qualifiedIndexTableName = SchemaUtil.getQualifiedTableName(schemaName, indexTableName); } + + // remove "", beacause system.catalog include in no Quotation + String removeMarksTableName = qualifiedTableName; + if (removeMarksTableName != null) { + removeMarksTableName = removeMarksTableName.replace("\"", ""); + } + if (tableName != null) { + tableName = tableName.replace("\"", ""); + } + if (schemaName != null) { + schemaName = schemaName.replace("\"", ""); + } if (cmdLine.hasOption(ZK_QUORUM_OPT.getOpt())) { // ZK_QUORUM_OPT is optional, but if it's there, use it for both the conn and the job. String zkQuorum = cmdLine.getOptionValue(ZK_QUORUM_OPT.getOpt()); PhoenixDriver.ConnectionInfo info = PhoenixDriver.ConnectionInfo.create(zkQuorum); LOGGER.info("Configuring HBase connection to {}", info); - for (Map.Entry entry : info.asProps()) { + for (Map.Entry entry : info.asProps()) { if (LOGGER.isDebugEnabled()) { LOGGER.debug("Setting {} = {}", entry.getKey(), entry.getValue()); } @@ -242,7 +254,7 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception { LOGGER.debug("Reading columns from {} :: {}", ((PhoenixConnection) conn).getURL(), qualifiedTableName); } - List importColumns = buildImportColumns(conn, cmdLine, qualifiedTableName); + List importColumns = buildImportColumns(conn, cmdLine, removeMarksTableName); Preconditions.checkNotNull(importColumns); Preconditions.checkArgument(!importColumns.isEmpty(), "Column info list is empty"); FormatToBytesWritableMapper.configureColumnInfoList(conf, importColumns); @@ -269,10 +281,10 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception { } List tablesToBeLoaded = new ArrayList(); - PTable table = PhoenixRuntime.getTable(conn, qualifiedTableName); + PTable table = PhoenixRuntime.getTable(conn, removeMarksTableName); tablesToBeLoaded.add(new TargetTableRef(qualifiedTableName, table.getPhysicalName().getString())); boolean hasLocalIndexes = false; - for(PTable index: table.getIndexes()) { + for (PTable index: table.getIndexes()) { if (index.getIndexType() == IndexType.LOCAL) { hasLocalIndexes = qualifiedIndexTableName == null ? true : index.getTableName().getString() @@ -281,18 +293,18 @@ private int loadData(Configuration conf, CommandLine cmdLine) throws Exception { } } // using conn after it's been closed... o.O - tablesToBeLoaded.addAll(getIndexTables(conn, qualifiedTableName)); + tablesToBeLoaded.addAll(getIndexTables(conn, removeMarksTableName)); // When loading a single index table, check index table name is correct - if (qualifiedIndexTableName != null){ + if (qualifiedIndexTableName != null) { TargetTableRef targetIndexRef = null; - for (TargetTableRef tmpTable : tablesToBeLoaded){ + for (TargetTableRef tmpTable : tablesToBeLoaded) { if (tmpTable.getLogicalName().compareToIgnoreCase(qualifiedIndexTableName) == 0) { targetIndexRef = tmpTable; break; } } - if (targetIndexRef == null){ + if (targetIndexRef == null) { throw new IllegalStateException("Bulk Loader error: index table " + qualifiedIndexTableName + " doesn't exist"); } @@ -325,8 +337,8 @@ public int submitJob(final Configuration conf, final String qualifiedTableName, try(org.apache.hadoop.hbase.client.Connection hbaseConn = ConnectionFactory.createConnection(job.getConfiguration())) { RegionLocator regionLocator = null; - if(hasLocalIndexes) { - try{ + if (hasLocalIndexes) { + try { regionLocator = hbaseConn.getRegionLocator( TableName.valueOf(qualifiedTableName)); splitKeysBeforeJob = regionLocator.getStartKeys(); @@ -357,7 +369,7 @@ public int submitJob(final Configuration conf, final String qualifiedTableName, try { regionLocator = hbaseConn.getRegionLocator( TableName.valueOf(qualifiedTableName)); - if(!IndexUtil.matchingSplitKeys(splitKeysBeforeJob, + if (!IndexUtil.matchingSplitKeys(splitKeysBeforeJob, regionLocator.getStartKeys())) { LOGGER.error("The table " + qualifiedTableName + " has local indexes and" + " there is split key mismatch before and after running" @@ -370,22 +382,22 @@ public int submitJob(final Configuration conf, final String qualifiedTableName, } } LOGGER.info("Loading HFiles from {}", outputPath); - completebulkload(conf,outputPath,tablesToBeLoaded); + completebulkload(conf, outputPath, tablesToBeLoaded); LOGGER.info("Removing output directory {}", outputPath); - if(!outputPath.getFileSystem(conf).delete(outputPath, true)) { + if (!outputPath.getFileSystem(conf).delete(outputPath, true)) { LOGGER.error("Failed to delete the output directory {}", outputPath); } return 0; } else { - return -1; - } - } + return -1; + } + } } - private void completebulkload(Configuration conf,Path outputPath , List tablesToBeLoaded) throws Exception { + private void completebulkload(Configuration conf, Path outputPath , List tablesToBeLoaded) throws Exception { Set tableNames = new HashSet<>(tablesToBeLoaded.size()); - for(TargetTableRef table : tablesToBeLoaded) { - if(tableNames.contains(table.getPhysicalName())){ + for (TargetTableRef table : tablesToBeLoaded) { + if (tableNames.contains(table.getPhysicalName())) { continue; } tableNames.add(table.getPhysicalName()); @@ -430,7 +442,7 @@ List buildImportColumns(Connection conn, CommandLine cmdLine, * @throws java.sql.SQLException */ private void validateTable(Connection conn, String schemaName, - String tableName) throws SQLException { + String tableName) throws SQLException { ResultSet rs = conn.getMetaData().getColumns( null, StringUtil.escapeLike(schemaName), @@ -460,7 +472,7 @@ private List getIndexTables(Connection conn, String qualifiedTab throws SQLException { PTable table = PhoenixRuntime.getTable(conn, qualifiedTableName); List indexTables = new ArrayList(); - for(PTable indexTable : table.getIndexes()){ + for (PTable indexTable : table.getIndexes()) { indexTables.add(new TargetTableRef(indexTable.getName().getString(), indexTable .getPhysicalName().getString())); } diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java index e844158b054..929b720107e 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java @@ -267,10 +267,14 @@ public static void main(String [] args) { System.out.println("Starting upgrading table:" + srcTable + "... please don't kill it in between!!"); UpgradeUtil.upgradeTable(conn, srcTable); } else if (execCmd.isUpgrade()) { - if (conn.getClientInfo(PhoenixRuntime.CURRENT_SCN_ATTRIB) != null) { throw new SQLException( - "May not specify the CURRENT_SCN property when upgrading"); } - if (conn.getClientInfo(PhoenixRuntime.TENANT_ID_ATTRIB) != null) { throw new SQLException( - "May not specify the TENANT_ID_ATTRIB property when upgrading"); } + if (conn.getClientInfo(PhoenixRuntime.CURRENT_SCN_ATTRIB) != null) { + throw new SQLException( + "May not specify the CURRENT_SCN property when upgrading"); + } + if (conn.getClientInfo(PhoenixRuntime.TENANT_ID_ATTRIB) != null) { + throw new SQLException( + "May not specify the TENANT_ID_ATTRIB property when upgrading"); + } if (execCmd.getInputFiles().isEmpty()) { List tablesNeedingUpgrade = UpgradeUtil.getPhysicalTablesWithDescRowKey(conn); if (tablesNeedingUpgrade.isEmpty()) { @@ -290,14 +294,13 @@ public static void main(String [] args) { } else { UpgradeUtil.upgradeDescVarLengthRowKeys(conn, execCmd.getInputFiles(), execCmd.isBypassUpgrade()); } - } else if(execCmd.isLocalIndexUpgrade()) { + } else if (execCmd.isLocalIndexUpgrade()) { UpgradeUtil.upgradeLocalIndexes(conn); } else { for (String inputFile : execCmd.getInputFiles()) { if (inputFile.endsWith(SQL_FILE_EXT)) { PhoenixRuntime.executeStatements(conn, new FileReader(inputFile), Collections.emptyList()); } else if (inputFile.endsWith(CSV_FILE_EXT)) { - String tableName = execCmd.getTableName(); if (tableName == null) { tableName = SchemaUtil.normalizeIdentifier( @@ -344,7 +347,7 @@ private PhoenixRuntime() { * @throws IOException * @throws SQLException */ - public static int executeStatements(Connection conn, Reader reader, List binds) throws IOException,SQLException { + public static int executeStatements(Connection conn, Reader reader, List binds) throws IOException, SQLException { PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); // Turn auto commit to true when running scripts in case there's DML pconn.setAutoCommit(true); @@ -360,7 +363,7 @@ public static int executeStatements(Connection conn, Reader reader, List */ @Deprecated public static List getUncommittedData(Connection conn) throws SQLException { - Iterator>> iterator = getUncommittedDataIterator(conn); + Iterator>> iterator = getUncommittedDataIterator(conn); if (iterator.hasNext()) { return iterator.next().getSecond(); } @@ -374,7 +377,7 @@ public static List getUncommittedData(Connection conn) throws SQLException * @return the list of HBase mutations for uncommitted data * @throws SQLException */ - public static Iterator>> getUncommittedDataIterator(Connection conn) throws SQLException { + public static Iterator>> getUncommittedDataIterator(Connection conn) throws SQLException { return getUncommittedDataIterator(conn, false); } @@ -385,10 +388,10 @@ public static Iterator>> getUncommittedDataIterator(Conne * @return the list of HBase mutations for uncommitted data * @throws SQLException */ - public static Iterator>> getUncommittedDataIterator(Connection conn, boolean includeMutableIndexes) throws SQLException { + public static Iterator>> getUncommittedDataIterator(Connection conn, boolean includeMutableIndexes) throws SQLException { final PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); - final Iterator>> iterator = pconn.getMutationState().toMutations(includeMutableIndexes); - return new Iterator>>() { + final Iterator>> iterator = pconn.getMutationState().toMutations(includeMutableIndexes); + return new Iterator>>() { @Override public boolean hasNext() { @@ -397,7 +400,7 @@ public boolean hasNext() { @Override public Pair> next() { - Pair> pair = iterator.next(); + Pair> pair = iterator.next(); List keyValues = Lists.newArrayListWithExpectedSize(pair.getSecond().size() * 5); // Guess-timate 5 key values per row for (Mutation mutation : pair.getSecond()) { for (List keyValueList : mutation.getFamilyCellMap().values()) { @@ -407,7 +410,7 @@ public Pair> next() { } } Collections.sort(keyValues, pconn.getKeyValueBuilder().getKeyValueComparator()); - return new Pair>(pair.getFirst(),keyValues); + return new Pair>(pair.getFirst(), keyValues); } @Override @@ -424,12 +427,11 @@ public static PTable getTableNoCache(Connection conn, String name) throws SQLExc PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); MetaDataMutationResult result = new MetaDataClient(pconn).updateCache(pconn.getTenantId(), schemaName, tableName, true); - if(result.getMutationCode() != MutationCode.TABLE_ALREADY_EXISTS) { + if (result.getMutationCode() != MutationCode.TABLE_ALREADY_EXISTS) { throw new TableNotFoundException(schemaName, tableName); } return result.getTable(); - } /** @@ -447,6 +449,10 @@ public static PTable getTableNoCache(Connection conn, String name) throws SQLExc */ public static PTable getTable(Connection conn, String name) throws SQLException { PTable table = null; + // remove "", beacause system.catalog include in no Quotation + if (name != null) { + name = name.replace("\"", ""); + } PhoenixConnection pconn = conn.unwrap(PhoenixConnection.class); try { table = pconn.getTable(new PTableKey(pconn.getTenantId(), name)); @@ -535,10 +541,10 @@ public static List generateColumnInfo(Connection conn, Set unresolvedColumnNames = new TreeSet(); if (columns == null || columns.isEmpty()) { // use all columns in the table - int offset = (table.getBucketNum() == null ? 0 : 1); - for (int i = offset; i < table.getColumns().size(); i++) { - PColumn pColumn = table.getColumns().get(i); - columnInfoList.add(PhoenixRuntime.getColumnInfo(pColumn)); + int offset = (table.getBucketNum() == null ? 0 : 1); + for (int i = offset; i < table.getColumns().size(); i++) { + PColumn pColumn = table.getColumns().get(i); + columnInfoList.add(PhoenixRuntime.getColumnInfo(pColumn)); } } else { // Leave "null" as indication to skip b/c it doesn't exist @@ -555,25 +561,31 @@ public static List generateColumnInfo(Connection conn, } } // if there exists columns that cannot be resolved, error out. - if (unresolvedColumnNames.size()>0) { + if (unresolvedColumnNames.size() > 0) { StringBuilder exceptionMessage = new StringBuilder(); boolean first = true; exceptionMessage.append("Unable to resolve these column names:\n"); for (String col : unresolvedColumnNames) { - if (first) first = false; - else exceptionMessage.append(","); + if (first) { + first = false; + } else { + exceptionMessage.append(","); + } exceptionMessage.append(col); } exceptionMessage.append("\nAvailable columns with column families:\n"); first = true; for (PColumn pColumn : table.getColumns()) { - if (first) first = false; - else exceptionMessage.append(","); + if (first) { + first = false; + } else { + exceptionMessage.append(","); + } exceptionMessage.append(pColumn.toString()); } throw new SQLException(exceptionMessage.toString()); - } - return columnInfoList; + } + return columnInfoList; } /** @@ -585,17 +597,17 @@ public static List generateColumnInfo(Connection conn, * @throws SQLException if parameters are null or if column is not found or if column is ambiguous. */ public static ColumnInfo getColumnInfo(PTable table, String columnName) throws SQLException { - if (table==null) { + if (table == null) { throw new SQLException("Table must not be null."); } - if (columnName==null) { + if (columnName == null) { throw new SQLException("columnName must not be null."); } PColumn pColumn = null; if (columnName.contains(QueryConstants.NAME_SEPARATOR)) { String[] tokens = columnName.split(QueryConstants.NAME_SEPARATOR_REGEX); - if (tokens.length!=2) { - throw new SQLException(String.format("Unable to process column %s, expected family-qualified name.",columnName)); + if (tokens.length != 2) { + throw new SQLException(String.format("Unable to process column %s, expected family-qualified name.", columnName)); } String familyName = tokens[0]; String familyColumn = tokens[1]; @@ -682,8 +694,8 @@ public static ExecutionCommand parseArgs(String[] args) { "but instead have always provided data up to the full max length of the column. See PHOENIX-2067 " + "and PHOENIX-2120 for more information. "); Option mapNamespaceOption = new Option("m", "map-namespace", true, - "Used to map table to a namespace matching with schema, require "+ QueryServices.IS_NAMESPACE_MAPPING_ENABLED + - " to be enabled"); + "Used to map table to a namespace matching with schema, require " + QueryServices.IS_NAMESPACE_MAPPING_ENABLED + + " to be enabled"); Option localIndexUpgradeOption = new Option("l", "local-index-upgrade", false, "Used to upgrade local index data by moving index data from separate table to " + "separate column families in the same table."); @@ -711,7 +723,7 @@ public static ExecutionCommand parseArgs(String[] args) { ExecutionCommand execCmd = new ExecutionCommand(); execCmd.connectionString = ""; - if(cmdLine.hasOption(mapNamespaceOption.getOpt())){ + if (cmdLine.hasOption(mapNamespaceOption.getOpt())) { execCmd.mapNamespace = true; execCmd.srcTable = validateTableName(cmdLine.getOptionValue(mapNamespaceOption.getOpt())); } @@ -758,7 +770,7 @@ public static ExecutionCommand parseArgs(String[] args) { } execCmd.isBypassUpgrade = true; } - if(cmdLine.hasOption(localIndexUpgradeOption.getOpt())) { + if (cmdLine.hasOption(localIndexUpgradeOption.getOpt())) { execCmd.localIndexUpgrade = true; } @@ -797,7 +809,6 @@ private static String validateTableName(String tableName) { } else { return tableName; } - } private static char getCharacter(String s) { @@ -1028,11 +1039,11 @@ public static String getArraySqlTypeName(@Nullable Integer maxLength, @Nullable private static String appendMaxLengthAndScale(@Nullable Integer maxLength, @Nullable Integer scale, String sqlTypeName) { if (maxLength != null) { - sqlTypeName = sqlTypeName + "(" + maxLength; - if (scale != null) { - sqlTypeName = sqlTypeName + "," + scale; // has both max length and scale. For ex- decimal(10,2) - } - sqlTypeName = sqlTypeName + ")"; + sqlTypeName = sqlTypeName + "(" + maxLength; + if (scale != null) { + sqlTypeName = sqlTypeName + "," + scale; // has both max length and scale. For ex- decimal(10,2) + } + sqlTypeName = sqlTypeName + ")"; } return sqlTypeName; } @@ -1139,8 +1150,8 @@ public static Object[] decodeValues(Connection conn, String fullTableName, byte[ kvSchema.iterator(ptr); int i = 0; List values = new ArrayList(); - while(hasValue = kvSchema.next(ptr, i, maxOffset, valueSet) != null) { - if(hasValue) { + while (hasValue = kvSchema.next(ptr, i, maxOffset, valueSet) != null) { + if (hasValue) { values.add(kvSchema.getField(i).getDataType().toObject(ptr)); } i++; @@ -1203,8 +1214,8 @@ public static Object[] decodeColumnValues(Connection conn, String fullTableName, kvSchema.iterator(ptr); int i = 0; List values = new ArrayList(); - while(hasValue = kvSchema.next(ptr, i, maxOffset, valueSet) != null) { - if(hasValue) { + while (hasValue = kvSchema.next(ptr, i, maxOffset, valueSet) != null) { + if (hasValue) { values.add(kvSchema.getField(i).getDataType().toObject(ptr)); } i++; @@ -1249,10 +1260,10 @@ private static List getPColumns(PTable table, List @Deprecated private static PColumn getPColumn(PTable table, @Nullable String familyName, String columnName) throws SQLException { - if (table==null) { + if (table == null) { throw new SQLException("Table must not be null."); } - if (columnName==null) { + if (columnName == null) { throw new SQLException("columnName must not be null."); } // normalize and remove quotes from family and column names before looking up. @@ -1284,10 +1295,10 @@ private static List getColumns(PTable table, List> } private static PColumn getColumn(PTable table, @Nullable String familyName, String columnName) throws SQLException { - if (table==null) { + if (table == null) { throw new SQLException("Table must not be null."); } - if (columnName==null) { + if (columnName == null) { throw new SQLException("columnName must not be null."); } // normalize and remove quotes from family and column names before looking up. @@ -1377,22 +1388,22 @@ public static boolean areGlobalClientMetricsBeingCollected() { } private static Map createMetricMap(Map metricInfoMap) { - Map metricMap = Maps.newHashMapWithExpectedSize(metricInfoMap.size()); - for (Entry entry : metricInfoMap.entrySet()) { - metricMap.put(entry.getKey().shortName(), entry.getValue()); - } - return metricMap; - } + Map metricMap = Maps.newHashMapWithExpectedSize(metricInfoMap.size()); + for (Entry entry : metricInfoMap.entrySet()) { + metricMap.put(entry.getKey().shortName(), entry.getValue()); + } + return metricMap; + } - private static Map> transformMetrics(Map> metricMap) { - Function, Map> func = new Function, Map>() { - @Override - public Map apply(Map map) { - return createMetricMap(map); - } - }; - return Maps.transformValues(metricMap, func); - } + private static Map> transformMetrics(Map> metricMap) { + Function, Map> func = new Function, Map>() { + @Override + public Map apply(Map map) { + return createMetricMap(map); + } + }; + return Maps.transformValues(metricMap, func); + } /** * Method to expose the metrics associated with performing reads using the passed result set. A typical pattern is: