You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by ja...@apache.org on 2018/05/22 20:39:44 UTC
[01/22] phoenix git commit: PHOENIX-4685 Properly handle connection
caching for Phoenix inside RegionServers(Rajeshbabu)
Repository: phoenix
Updated Branches:
refs/heads/omid2 d0f98a020 -> 2015345a0
PHOENIX-4685 Properly handle connection caching for Phoenix inside RegionServers(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/4082c73e
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/4082c73e
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/4082c73e
Branch: refs/heads/omid2
Commit: 4082c73ee23d901642d8c5bc45ececfcf5e50ede
Parents: d0f98a0
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Tue May 8 12:06:49 2018 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Tue May 8 12:06:49 2018 +0530
----------------------------------------------------------------------
.../DelegateRegionCoprocessorEnvironment.java | 7 +-
.../UngroupedAggregateRegionObserver.java | 14 +-
.../org/apache/phoenix/hbase/index/Indexer.java | 19 +--
.../hbase/index/write/IndexWriterUtils.java | 27 +---
.../index/PhoenixTransactionalIndexer.java | 18 +--
.../org/apache/phoenix/util/ServerUtil.java | 141 ++++++++++++++++---
6 files changed, 142 insertions(+), 84 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionCoprocessorEnvironment.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionCoprocessorEnvironment.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionCoprocessorEnvironment.java
index 284d53c..a791f4a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionCoprocessorEnvironment.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/DelegateRegionCoprocessorEnvironment.java
@@ -32,6 +32,7 @@ import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
/**
* Class to encapsulate {@link RegionCoprocessorEnvironment} for phoenix coprocessors. Often we
@@ -44,10 +45,10 @@ public class DelegateRegionCoprocessorEnvironment implements RegionCoprocessorEn
private RegionCoprocessorEnvironment delegate;
private HTableFactory tableFactory;
- public DelegateRegionCoprocessorEnvironment(Configuration config, RegionCoprocessorEnvironment delegate) {
- this.config = config;
+ public DelegateRegionCoprocessorEnvironment(RegionCoprocessorEnvironment delegate, ConnectionType connectionType) {
+ this.config = ServerUtil.ConnectionFactory.getTypeSpecificConfiguration(connectionType, delegate.getConfiguration());
this.delegate = delegate;
- this.tableFactory = ServerUtil.getDelegateHTableFactory(this, config);
+ this.tableFactory = ServerUtil.getDelegateHTableFactory(this, connectionType);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index 6bee65c..14213f4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -144,6 +144,7 @@ import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
import org.apache.phoenix.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -225,14 +226,7 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
upsertSelectConfig.setClass(RpcControllerFactory.CUSTOM_CONTROLLER_CONF_KEY,
InterRegionServerIndexRpcControllerFactory.class, RpcControllerFactory.class);
- compactionConfig = PropertiesUtil.cloneConfig(e.getConfiguration());
- // lower the number of rpc retries, so we don't hang the compaction
- compactionConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
- e.getConfiguration().getInt(QueryServices.METADATA_WRITE_RETRIES_NUMBER,
- QueryServicesOptions.DEFAULT_METADATA_WRITE_RETRIES_NUMBER));
- compactionConfig.setInt(HConstants.HBASE_CLIENT_PAUSE,
- e.getConfiguration().getInt(QueryServices.METADATA_WRITE_RETRY_PAUSE,
- QueryServicesOptions.DEFAULT_METADATA_WRITE_RETRY_PAUSE));
+ compactionConfig = ServerUtil.getCompactionConfig(e.getConfiguration());
// For retries of index write failures, use the same # of retries as the rebuilder
indexWriteConfig = PropertiesUtil.cloneConfig(e.getConfiguration());
@@ -984,7 +978,9 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
InternalScanner internalScanner = scanner;
try {
long clientTimeStamp = EnvironmentEdgeManager.currentTimeMillis();
- DelegateRegionCoprocessorEnvironment compactionConfEnv = new DelegateRegionCoprocessorEnvironment(compactionConfig, c.getEnvironment());
+ DelegateRegionCoprocessorEnvironment compactionConfEnv =
+ new DelegateRegionCoprocessorEnvironment(c.getEnvironment(),
+ ConnectionType.COMPACTION_CONNECTION);
StatisticsCollector stats = StatisticsCollectorFactory.createStatisticsCollector(
compactionConfEnv, table.getNameAsString(), clientTimeStamp,
store.getFamily().getName());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index 7325cd8..115182b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -94,6 +94,7 @@ import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
@@ -219,25 +220,11 @@ public class Indexer extends BaseRegionObserver {
this.builder = new IndexBuildManager(env);
// Clone the config since it is shared
- Configuration clonedConfig = PropertiesUtil.cloneConfig(e.getConfiguration());
- /*
- * Set the rpc controller factory so that the HTables used by IndexWriter would
- * set the correct priorities on the remote RPC calls.
- */
- clonedConfig.setClass(RpcControllerFactory.CUSTOM_CONTROLLER_CONF_KEY,
- InterRegionServerIndexRpcControllerFactory.class, RpcControllerFactory.class);
- // lower the number of rpc retries. We inherit config from HConnectionManager#setServerSideHConnectionRetries,
- // which by default uses a multiplier of 10. That is too many retries for our synchronous index writes
- clonedConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
- env.getConfiguration().getInt(INDEX_WRITER_RPC_RETRIES_NUMBER,
- DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER));
- clonedConfig.setInt(HConstants.HBASE_CLIENT_PAUSE, env.getConfiguration()
- .getInt(INDEX_WRITER_RPC_PAUSE, DEFAULT_INDEX_WRITER_RPC_PAUSE));
- DelegateRegionCoprocessorEnvironment indexWriterEnv = new DelegateRegionCoprocessorEnvironment(clonedConfig, env);
+ DelegateRegionCoprocessorEnvironment indexWriterEnv = new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION);
// setup the actual index writer
this.writer = new IndexWriter(indexWriterEnv, serverName + "-index-writer");
- this.rowLockWaitDuration = clonedConfig.getInt("hbase.rowlock.wait.duration",
+ this.rowLockWaitDuration = env.getConfiguration().getInt("hbase.rowlock.wait.duration",
DEFAULT_ROWLOCK_WAIT_DURATION);
this.lockManager = new LockManager();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java
index 0d3004f..ef53b9f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/write/IndexWriterUtils.java
@@ -26,6 +26,7 @@ import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
public class IndexWriterUtils {
@@ -50,9 +51,9 @@ public class IndexWriterUtils {
* threads. Currently, HBase doesn't support a custom thread-pool to back the HTable via the
* coprocesor hooks, so we can't modify this behavior.
*/
- private static final String INDEX_WRITER_PER_TABLE_THREADS_CONF_KEY =
+ public static final String INDEX_WRITER_PER_TABLE_THREADS_CONF_KEY =
"index.writer.threads.pertable.max";
- private static final int DEFAULT_NUM_PER_TABLE_THREADS = Integer.MAX_VALUE;
+ public static final int DEFAULT_NUM_PER_TABLE_THREADS = Integer.MAX_VALUE;
/** Configuration key that HBase uses to set the max number of threads for an HTable */
public static final String HTABLE_THREAD_KEY = "hbase.htable.threads.max";
@@ -79,19 +80,7 @@ public class IndexWriterUtils {
}
public static HTableFactory getDefaultDelegateHTableFactory(CoprocessorEnvironment env) {
- // create a simple delegate factory, setup the way we need
- Configuration conf = PropertiesUtil.cloneConfig(env.getConfiguration());
- setHTableThreads(conf);
- return ServerUtil.getDelegateHTableFactory(env, conf);
- }
-
- private static void setHTableThreads(Configuration conf) {
- // set the number of threads allowed per table.
- int htableThreads =
- conf.getInt(IndexWriterUtils.INDEX_WRITER_PER_TABLE_THREADS_CONF_KEY,
- IndexWriterUtils.DEFAULT_NUM_PER_TABLE_THREADS);
- LOG.trace("Creating HTableFactory with " + htableThreads + " threads for each HTable.");
- IndexManagementUtil.setIfNotSet(conf, HTABLE_THREAD_KEY, htableThreads);
+ return ServerUtil.getDelegateHTableFactory(env, ConnectionType.INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS);
}
/**
@@ -99,12 +88,8 @@ public class IndexWriterUtils {
* instead to avoid tying up the handler
*/
public static HTableFactory getNoRetriesHTableFactory(CoprocessorEnvironment env) {
- Configuration conf = PropertiesUtil.cloneConfig(env.getConfiguration());
- setHTableThreads(conf);
- // note in HBase 2+, numTries = numRetries + 1
- // in prior versions, numTries = numRetries
- conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
- return ServerUtil.getDelegateHTableFactory(env, conf);
+ return ServerUtil.getDelegateHTableFactory(env,
+ ConnectionType.INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS_NO_RETRIES);
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
index bdfcaff..02296c9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/index/PhoenixTransactionalIndexer.java
@@ -58,6 +58,7 @@ import org.apache.phoenix.trace.util.NullSpan;
import org.apache.phoenix.transaction.PhoenixTransactionContext;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.ServerUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionType;
import org.apache.phoenix.util.TransactionUtil;
/**
@@ -94,22 +95,7 @@ public class PhoenixTransactionalIndexer extends BaseRegionObserver {
Configuration conf = e.getConfiguration();
String serverName = env.getRegionServerServices().getServerName().getServerName();
codec = new PhoenixIndexCodec(conf, env.getRegion().getRegionInfo().getStartKey(), env.getRegion().getRegionInfo().getEndKey(), env.getRegionInfo().getTable().getName());
- // Clone the config since it is shared
- Configuration clonedConfig = PropertiesUtil.cloneConfig(e.getConfiguration());
- /*
- * Set the rpc controller factory so that the HTables used by IndexWriter would
- * set the correct priorities on the remote RPC calls.
- */
- clonedConfig.setClass(RpcControllerFactory.CUSTOM_CONTROLLER_CONF_KEY,
- InterRegionServerIndexRpcControllerFactory.class, RpcControllerFactory.class);
- // lower the number of rpc retries. We inherit config from HConnectionManager#setServerSideHConnectionRetries,
- // which by default uses a multiplier of 10. That is too many retries for our synchronous index writes
- clonedConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
- env.getConfiguration().getInt(INDEX_WRITER_RPC_RETRIES_NUMBER,
- DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER));
- clonedConfig.setInt(HConstants.HBASE_CLIENT_PAUSE, env.getConfiguration()
- .getInt(INDEX_WRITER_RPC_PAUSE, DEFAULT_INDEX_WRITER_RPC_PAUSE));
- DelegateRegionCoprocessorEnvironment indexWriterEnv = new DelegateRegionCoprocessorEnvironment(clonedConfig, env);
+ DelegateRegionCoprocessorEnvironment indexWriterEnv = new DelegateRegionCoprocessorEnvironment(env, ConnectionType.INDEX_WRITER_CONNECTION);
// setup the actual index writer
// For transactional tables, we keep the index active upon a write failure
// since we have the all versus none behavior for transactions. Also, we
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4082c73e/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 4b3cc43..2dab076 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -17,11 +17,17 @@
*/
package org.apache.phoenix.util;
+import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.DEFAULT_INDEX_WRITER_RPC_PAUSE;
+import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER;
+import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.INDEX_WRITER_RPC_PAUSE;
+import static org.apache.phoenix.hbase.index.write.IndexWriterUtils.INDEX_WRITER_RPC_RETRIES_NUMBER;
+
import java.io.IOException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
+import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -35,12 +41,15 @@ import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.client.CoprocessorHConnection;
import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
+import org.apache.hadoop.hbase.ipc.controller.InterRegionServerIndexRpcControllerFactory;
import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.RegionServerServices;
@@ -52,8 +61,13 @@ import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.hbase.index.table.CoprocessorHTableFactory;
import org.apache.phoenix.hbase.index.table.HTableFactory;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
import org.apache.phoenix.hbase.index.util.VersionUtil;
+import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.StaleRegionBoundaryCacheException;
+import org.jboss.netty.util.internal.ConcurrentHashMap;
@SuppressWarnings("deprecation")
@@ -269,12 +283,12 @@ public class ServerUtil {
endKey) < 0));
}
- public static HTableFactory getDelegateHTableFactory(CoprocessorEnvironment env, Configuration conf) {
+ public static HTableFactory getDelegateHTableFactory(CoprocessorEnvironment env, ConnectionType connectionType) {
if (env instanceof RegionCoprocessorEnvironment) {
RegionCoprocessorEnvironment e = (RegionCoprocessorEnvironment) env;
RegionServerServices services = e.getRegionServerServices();
if (services instanceof HRegionServer) {
- return new CoprocessorHConnectionTableFactory(conf, (HRegionServer) services);
+ return new CoprocessorHConnectionTableFactory(env.getConfiguration(), (HRegionServer) services, connectionType);
}
}
return new CoprocessorHTableFactory(env);
@@ -286,44 +300,133 @@ public class ServerUtil {
* https://issues.apache.org/jira/browse/HBASE-18359
*/
public static class CoprocessorHConnectionTableFactory implements HTableFactory {
- @GuardedBy("CoprocessorHConnectionTableFactory.this")
- private HConnection connection;
private final Configuration conf;
private final HRegionServer server;
+ private final ConnectionType connectionType;
- CoprocessorHConnectionTableFactory(Configuration conf, HRegionServer server) {
+ CoprocessorHConnectionTableFactory(Configuration conf, HRegionServer server, ConnectionType connectionType) {
this.conf = conf;
this.server = server;
+ this.connectionType = connectionType;
}
- private synchronized HConnection getConnection(Configuration conf) throws IOException {
- if (connection == null || connection.isClosed()) {
- connection = new CoprocessorHConnection(conf, server);
- }
- return connection;
+ private ClusterConnection getConnection() throws IOException {
+ return ConnectionFactory.getConnection(connectionType, conf, server);
}
@Override
public HTableInterface getTable(ImmutableBytesPtr tablename) throws IOException {
- return getConnection(conf).getTable(tablename.copyBytesIfNecessary());
+ return getConnection().getTable(tablename.copyBytesIfNecessary());
}
@Override
public synchronized void shutdown() {
- try {
- if (connection != null && !connection.isClosed()) {
- connection.close();
- }
- } catch (Throwable e) {
- LOG.warn("Error while trying to close the HConnection used by CoprocessorHConnectionTableFactory", e);
- }
+ // We need not close the cached connections as they are shared across the server.
}
@Override
public HTableInterface getTable(ImmutableBytesPtr tablename, ExecutorService pool)
throws IOException {
- return getConnection(conf).getTable(tablename.copyBytesIfNecessary(), pool);
+ return getConnection().getTable(tablename.copyBytesIfNecessary(), pool);
+ }
+ }
+
+ public static enum ConnectionType {
+ COMPACTION_CONNECTION,
+ INDEX_WRITER_CONNECTION,
+ INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS,
+ INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS_NO_RETRIES,
+ DEFAULT_SERVER_CONNECTION;
+ }
+
+ public static class ConnectionFactory {
+
+ private static Map<ConnectionType, ClusterConnection> connections =
+ new ConcurrentHashMap<ConnectionType, ClusterConnection>();
+
+ public static ClusterConnection getConnection(final ConnectionType connectionType, final Configuration conf, final HRegionServer server) throws IOException {
+ ClusterConnection connection = null;
+ if((connection = connections.get(connectionType)) == null) {
+ synchronized (CoprocessorHConnectionTableFactory.class) {
+ if(connections.get(connectionType) == null) {
+ connection = new CoprocessorHConnection(conf, server);
+ connections.put(connectionType, connection);
+ return connection;
+ }
+ }
+ }
+ return connection;
}
+
+ public static Configuration getTypeSpecificConfiguration(ConnectionType connectionType, Configuration conf) {
+ switch (connectionType) {
+ case COMPACTION_CONNECTION:
+ return getCompactionConfig(conf);
+ case DEFAULT_SERVER_CONNECTION:
+ return conf;
+ case INDEX_WRITER_CONNECTION:
+ return getIndexWriterConnection(conf);
+ case INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS:
+ return getIndexWriterConfigurationWithCustomThreads(conf);
+ case INDEX_WRITER_CONNECTION_WITH_CUSTOM_THREADS_NO_RETRIES:
+ return getNoRetriesIndexWriterConfigurationWithCustomThreads(conf);
+ default:
+ return conf;
+ }
+ }
+ }
+
+ public static Configuration getCompactionConfig(Configuration conf) {
+ Configuration compactionConfig = PropertiesUtil.cloneConfig(conf);
+ // lower the number of rpc retries, so we don't hang the compaction
+ compactionConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+ conf.getInt(QueryServices.METADATA_WRITE_RETRIES_NUMBER,
+ QueryServicesOptions.DEFAULT_METADATA_WRITE_RETRIES_NUMBER));
+ compactionConfig.setInt(HConstants.HBASE_CLIENT_PAUSE,
+ conf.getInt(QueryServices.METADATA_WRITE_RETRY_PAUSE,
+ QueryServicesOptions.DEFAULT_METADATA_WRITE_RETRY_PAUSE));
+ return compactionConfig;
+ }
+
+ public static Configuration getIndexWriterConnection(Configuration conf) {
+ Configuration clonedConfig = PropertiesUtil.cloneConfig(conf);
+ /*
+ * Set the rpc controller factory so that the HTables used by IndexWriter would
+ * set the correct priorities on the remote RPC calls.
+ */
+ clonedConfig.setClass(RpcControllerFactory.CUSTOM_CONTROLLER_CONF_KEY,
+ InterRegionServerIndexRpcControllerFactory.class, RpcControllerFactory.class);
+ // lower the number of rpc retries. We inherit config from HConnectionManager#setServerSideHConnectionRetries,
+ // which by default uses a multiplier of 10. That is too many retries for our synchronous index writes
+ clonedConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+ conf.getInt(INDEX_WRITER_RPC_RETRIES_NUMBER,
+ DEFAULT_INDEX_WRITER_RPC_RETRIES_NUMBER));
+ clonedConfig.setInt(HConstants.HBASE_CLIENT_PAUSE, conf
+ .getInt(INDEX_WRITER_RPC_PAUSE, DEFAULT_INDEX_WRITER_RPC_PAUSE));
+ return clonedConfig;
+ }
+
+ public static Configuration getIndexWriterConfigurationWithCustomThreads(Configuration conf) {
+ Configuration clonedConfig = PropertiesUtil.cloneConfig(conf);
+ setHTableThreads(clonedConfig);
+ return clonedConfig;
+ }
+
+ private static void setHTableThreads(Configuration conf) {
+ // set the number of threads allowed per table.
+ int htableThreads =
+ conf.getInt(IndexWriterUtils.INDEX_WRITER_PER_TABLE_THREADS_CONF_KEY,
+ IndexWriterUtils.DEFAULT_NUM_PER_TABLE_THREADS);
+ IndexManagementUtil.setIfNotSet(conf, IndexWriterUtils.HTABLE_THREAD_KEY, htableThreads);
+ }
+
+ public static Configuration getNoRetriesIndexWriterConfigurationWithCustomThreads(Configuration conf) {
+ Configuration clonedConf = getIndexWriterConfigurationWithCustomThreads(conf);
+ // note in HBase 2+, numTries = numRetries + 1
+ // in prior versions, numTries = numRetries
+ clonedConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
+ return clonedConf;
+
}
}
[21/22] phoenix git commit: PHOENIX-4685 Properly handle connection
caching for Phoenix inside RegionServers(addendum) (James Taylor)
Posted by ja...@apache.org.
PHOENIX-4685 Properly handle connection caching for Phoenix inside RegionServers(addendum) (James Taylor)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/52432db4
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/52432db4
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/52432db4
Branch: refs/heads/omid2
Commit: 52432db457303583105c8b74375e5415327d7975
Parents: d51353a
Author: Ankit Singhal <an...@gmail.com>
Authored: Wed May 16 17:37:36 2018 -0700
Committer: Ankit Singhal <an...@gmail.com>
Committed: Wed May 16 17:37:36 2018 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/util/ServerUtil.java | 23 +++++++++++++++-----
.../java/org/apache/phoenix/query/BaseTest.java | 13 +++++++----
2 files changed, 26 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/52432db4/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 891839a..09701c5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -31,8 +31,6 @@ import java.util.concurrent.ExecutorService;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
-import javax.annotation.concurrent.GuardedBy;
-
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
@@ -43,15 +41,14 @@ import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.client.CoprocessorHConnection;
-import org.apache.hadoop.hbase.client.HConnection;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.ipc.controller.InterRegionServerIndexRpcControllerFactory;
-import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.regionserver.Region;
import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.HashJoinCacheNotFoundException;
@@ -318,8 +315,8 @@ public class ServerUtil {
}
@Override
- public synchronized void shutdown() {
- // We need not close the cached connections as they are shared across the server.
+ public void shutdown() {
+ ConnectionFactory.shutdown();
}
@Override
@@ -342,6 +339,20 @@ public class ServerUtil {
private static Map<ConnectionType, ClusterConnection> connections =
new ConcurrentHashMap<ConnectionType, ClusterConnection>();
+ public static void shutdown() {
+ synchronized (CoprocessorHConnectionTableFactory.class) {
+ for (ClusterConnection connection : connections.values()) {
+ try {
+ connection.close();
+ } catch (IOException e) {
+ LOG.warn("Unable to close coprocessor connection", e);
+ }
+ }
+ connections.clear();
+ }
+ }
+
+
public static ClusterConnection getConnection(final ConnectionType connectionType, final Configuration conf, final HRegionServer server) throws IOException {
ClusterConnection connection = null;
if((connection = connections.get(connectionType)) == null) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/52432db4/phoenix-core/src/test/java/org/apache/phoenix/query/BaseTest.java
----------------------------------------------------------------------
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 0ea63e7..f49d291 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
@@ -137,6 +137,7 @@ import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.util.ServerUtil.ConnectionFactory;
import org.junit.ClassRule;
import org.junit.rules.TemporaryFolder;
import org.slf4j.Logger;
@@ -477,10 +478,14 @@ public abstract class BaseTest {
} catch (Throwable t) {
logger.error("Exception caught when shutting down mini cluster", t);
} finally {
- logger.info(
- "Time in seconds spent in shutting down mini cluster with "
- + numTables + " tables: "
- + (System.currentTimeMillis() - startTime) / 1000);
+ try {
+ ConnectionFactory.shutdown();
+ } finally {
+ logger.info(
+ "Time in seconds spent in shutting down mini cluster with "
+ + numTables + " tables: "
+ + (System.currentTimeMillis() - startTime) / 1000);
+ }
}
}
}
[22/22] phoenix git commit: PHOENIX-4701 Write client-side metrics
asynchronously to SYSTEM.LOG(addendum 2)
Posted by ja...@apache.org.
PHOENIX-4701 Write client-side metrics asynchronously to SYSTEM.LOG(addendum 2)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/2015345a
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/2015345a
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/2015345a
Branch: refs/heads/omid2
Commit: 2015345a023f0adb59174443ec1328bb1399f11b
Parents: 52432db
Author: James Taylor <jt...@salesforce.com>
Authored: Wed May 16 21:37:02 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed May 16 21:37:02 2018 -0700
----------------------------------------------------------------------
.../src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/2015345a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java
index 1160f4e..460b22f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryWithLimitIT.java
@@ -59,9 +59,9 @@ public class QueryWithLimitIT extends BaseUniqueNamesOwnClusterIT {
// Must update config before starting server
props.put(QueryServices.STATS_GUIDEPOST_WIDTH_BYTES_ATTRIB, Long.toString(50));
props.put(QueryServices.QUEUE_SIZE_ATTRIB, Integer.toString(1));
- props.put(QueryServices.DROP_METADATA_ATTRIB, Boolean.TRUE.toString());
- props.put(QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB, Integer.toString(0)); // Prevents RejectedExecutionException when deleting sequences
+ props.put(QueryServices.SEQUENCE_SALT_BUCKETS_ATTRIB, Integer.toString(0)); // Prevents RejectedExecutionException when creatomg sequence table
props.put(QueryServices.THREAD_POOL_SIZE_ATTRIB, Integer.toString(4));
+ props.put(QueryServices.LOG_SALT_BUCKETS_ATTRIB, Integer.toString(0)); // Prevents RejectedExecutionException when creating log table
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
[12/22] phoenix git commit: PHOENIX-4685 Properly handle connection
caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Posted by ja...@apache.org.
PHOENIX-4685 Properly handle connection caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/39b92bf9
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/39b92bf9
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/39b92bf9
Branch: refs/heads/omid2
Commit: 39b92bf9e8d9cae46b1fa230d91ac04a8e49e629
Parents: 72fa874
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Fri May 11 23:20:52 2018 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Fri May 11 23:20:52 2018 +0530
----------------------------------------------------------------------
.../src/main/java/org/apache/phoenix/util/ServerUtil.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/39b92bf9/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 9190373..891839a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -346,8 +346,8 @@ public class ServerUtil {
ClusterConnection connection = null;
if((connection = connections.get(connectionType)) == null) {
synchronized (CoprocessorHConnectionTableFactory.class) {
- if(connections.get(connectionType) == null) {
- connection = new CoprocessorHConnection(conf, server);
+ if((connection = connections.get(connectionType)) == null) {
+ connection = new CoprocessorHConnection(getTypeSpecificConfiguration(connectionType, conf), server);
connections.put(connectionType, connection);
return connection;
}
@@ -405,7 +405,7 @@ public class ServerUtil {
}
public static Configuration getIndexWriterConfigurationWithCustomThreads(Configuration conf) {
- Configuration clonedConfig = PropertiesUtil.cloneConfig(conf);
+ Configuration clonedConfig = getIndexWriterConnection(conf);
setHTableThreads(clonedConfig);
return clonedConfig;
}
[05/22] phoenix git commit: PHOENIX-4623 Inconsistent physical view
index name
Posted by ja...@apache.org.
PHOENIX-4623 Inconsistent physical view index name
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/4d6f12e6
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/4d6f12e6
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/4d6f12e6
Branch: refs/heads/omid2
Commit: 4d6f12e6097c8173abc799ef6cb3275c161e110c
Parents: 4849fbe
Author: James Taylor <jt...@salesforce.com>
Authored: Tue May 8 16:29:41 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed May 9 13:36:32 2018 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/end2end/BaseViewIT.java | 7 +++----
.../java/org/apache/phoenix/end2end/ViewIT.java | 2 +-
.../phoenix/schema/stats/StatsCollectorIT.java | 3 +--
.../org/apache/phoenix/util/MetaDataUtil.java | 18 +++++-------------
.../java/org/apache/phoenix/util/UpgradeUtil.java | 2 +-
5 files changed, 11 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4d6f12e6/phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java
index 478b234..5728a20 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/BaseViewIT.java
@@ -46,7 +46,6 @@ import org.apache.phoenix.util.MetaDataUtil;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ScanUtil;
import org.apache.phoenix.util.SchemaUtil;
-import org.apache.phoenix.util.TestUtil;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
@@ -66,7 +65,7 @@ public abstract class BaseViewIT extends ParallelStatsEnabledIT {
if (transactional) {
optionBuilder.append(" TRANSACTIONAL=true ");
}
- this.schemaName = TestUtil.DEFAULT_SCHEMA_NAME;
+ this.schemaName = "S_" + generateUniqueName();
this.tableDDLOptions = optionBuilder.toString();
this.tableName = "T_" + generateUniqueName();
this.fullTableName = SchemaUtil.getTableName(schemaName, tableName);
@@ -166,7 +165,7 @@ public abstract class BaseViewIT extends ParallelStatsEnabledIT {
ResultSet rs;
Connection conn = DriverManager.getConnection(getUrl());
String viewIndexName1 = "I_" + generateUniqueName();
- String viewIndexPhysicalName = MetaDataUtil.getViewIndexName(schemaName, tableName);
+ String viewIndexPhysicalName = MetaDataUtil.getViewIndexPhysicalName(fullTableName);
if (localIndex) {
conn.createStatement().execute("CREATE LOCAL INDEX " + viewIndexName1 + " on " + viewName + "(k3)");
} else {
@@ -231,7 +230,7 @@ public abstract class BaseViewIT extends ParallelStatsEnabledIT {
rs = conn.createStatement().executeQuery("EXPLAIN " + query);
String physicalTableName;
if (localIndex) {
- physicalTableName = tableName;
+ physicalTableName = fullTableName;
assertEquals("CLIENT PARALLEL "+ (saltBuckets == null ? 1 : saltBuckets) +"-WAY RANGE SCAN OVER " + fullTableName +" [" + (2) + ",'foo']\n"
+ " SERVER FILTER BY FIRST KEY ONLY\n"
+ "CLIENT MERGE SORT",QueryUtil.getExplainPlan(rs));
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4d6f12e6/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
index 279bbd7..34292ba 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ViewIT.java
@@ -69,7 +69,7 @@ public class ViewIT extends BaseViewIT {
String ddl = "CREATE TABLE " + fullTableName + " (k INTEGER NOT NULL PRIMARY KEY, v1 DATE) "+ tableDDLOptions;
conn.createStatement().execute(ddl);
String fullParentViewName = "V_" + generateUniqueName();
- ddl = "CREATE VIEW " + fullParentViewName + " (v2 VARCHAR) AS SELECT * FROM " + tableName + " WHERE k > 5";
+ ddl = "CREATE VIEW " + fullParentViewName + " (v2 VARCHAR) AS SELECT * FROM " + fullTableName + " WHERE k > 5";
conn.createStatement().execute(ddl);
try {
conn.createStatement().execute("UPSERT INTO " + fullParentViewName + " VALUES(1)");
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4d6f12e6/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
index c424f45..09d28f8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/schema/stats/StatsCollectorIT.java
@@ -48,7 +48,6 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.phoenix.coprocessor.MetaDataRegionObserver;
import org.apache.phoenix.coprocessor.UngroupedAggregateRegionObserver;
import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
import org.apache.phoenix.jdbc.PhoenixConnection;
@@ -785,7 +784,7 @@ public abstract class StatsCollectorIT extends BaseUniqueNamesOwnClusterIT {
String viewIndex = "VI_" + generateUniqueName();
ddl = "CREATE INDEX " + viewIndex + " ON " + view + " (b)";
conn.createStatement().execute(ddl);
- String viewIndexTableName = MetaDataUtil.getViewIndexTableName(baseTable);
+ String viewIndexTableName = MetaDataUtil.getViewIndexPhysicalName(baseTable);
statsCollector = getDefaultStatsCollectorForTable(viewIndexTableName);
statsCollector.init();
assertEquals(defaultGuidePostWidth, statsCollector.getGuidePostDepth());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4d6f12e6/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
index 2a0c8f0..1dda818 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/MetaDataUtil.java
@@ -356,23 +356,15 @@ public class MetaDataUtil {
return getIndexPhysicalName(physicalTableName, VIEW_INDEX_TABLE_PREFIX);
}
- public static String getViewIndexTableName(String tableName) {
- return VIEW_INDEX_TABLE_PREFIX + tableName;
- }
-
- public static String getViewIndexSchemaName(String schemaName) {
- return schemaName;
- }
-
- public static String getViewIndexName(String schemaName, String tableName) {
- return SchemaUtil.getTableName(getViewIndexSchemaName(schemaName), getViewIndexTableName(tableName));
+ public static String getViewIndexPhysicalName(String physicalTableName) {
+ return getIndexPhysicalName(physicalTableName, VIEW_INDEX_TABLE_PREFIX);
}
- public static byte[] getIndexPhysicalName(byte[] physicalTableName, String indexPrefix) {
- return getIndexPhysicalName(Bytes.toString(physicalTableName), indexPrefix).getBytes();
+ private static byte[] getIndexPhysicalName(byte[] physicalTableName, String indexPrefix) {
+ return Bytes.toBytes(getIndexPhysicalName(Bytes.toString(physicalTableName), indexPrefix));
}
- public static String getIndexPhysicalName(String physicalTableName, String indexPrefix) {
+ private static String getIndexPhysicalName(String physicalTableName, String indexPrefix) {
if (physicalTableName.contains(QueryConstants.NAMESPACE_SEPARATOR)) {
String schemaName = SchemaUtil.getSchemaNameFromFullName(physicalTableName,
QueryConstants.NAMESPACE_SEPARATOR);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4d6f12e6/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
index 63f2183..b127408 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/UpgradeUtil.java
@@ -600,7 +600,7 @@ public class UpgradeUtil {
viewPTable = PhoenixRuntime.getTable(globalConnection, fullViewName);
globalConnection.createStatement().execute(String.format(disableIndexDDL, indexName, fullViewName));
}
- String indexPhysicalTableName = MetaDataUtil.getViewIndexTableName(viewPTable.getPhysicalName().getString());
+ String indexPhysicalTableName = MetaDataUtil.getViewIndexPhysicalName(viewPTable.getPhysicalName().getString());
if (physicalTables.add(indexPhysicalTableName)) {
final TableName tableName = TableName.valueOf(indexPhysicalTableName);
if(admin.tableExists(tableName)) {
[07/22] phoenix git commit: PHOENIX-4678 ndexScrutinyTool generates
malformed query due to incorrect table name(s) (Josh Elser)
Posted by ja...@apache.org.
PHOENIX-4678 ndexScrutinyTool generates malformed query due to incorrect table name(s) (Josh Elser)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/7c20fdad
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/7c20fdad
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/7c20fdad
Branch: refs/heads/omid2
Commit: 7c20fdad9f100e5fbd36eb5e803603354ed2b180
Parents: 4d6f12e
Author: James Taylor <jt...@salesforce.com>
Authored: Wed May 9 12:50:22 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed May 9 13:37:10 2018 -0700
----------------------------------------------------------------------
phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7c20fdad/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index b43f54e..92a2cde 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -1132,7 +1132,7 @@ public class SchemaUtil {
* @return the byte representation of the HTable
*/
public static String getQualifiedTableName(String schemaName, String tableName) {
- if (schemaName != null) {
+ if (schemaName != null && !schemaName.isEmpty()) {
return String.format("%s.%s", normalizeIdentifier(schemaName),
normalizeIdentifier(tableName));
} else {
[19/22] phoenix git commit: PHOENIX-4737 Use position as column
qualifier for APPEND_ONLY_SCHEMA
Posted by ja...@apache.org.
PHOENIX-4737 Use position as column qualifier for APPEND_ONLY_SCHEMA
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/a72aad33
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/a72aad33
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/a72aad33
Branch: refs/heads/omid2
Commit: a72aad3369789129196cf180da7e757a71034fcb
Parents: f73f98c
Author: James Taylor <jt...@salesforce.com>
Authored: Sun May 13 17:36:21 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon May 14 21:53:35 2018 -0700
----------------------------------------------------------------------
.../end2end/ColumnEncodedBytesPropIT.java | 60 ++++++++++++++++++++
.../coprocessor/MetaDataEndpointImpl.java | 9 ++-
.../apache/phoenix/schema/MetaDataClient.java | 11 ++--
3 files changed, 75 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a72aad33/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedBytesPropIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedBytesPropIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedBytesPropIT.java
index 3b129f5..c85ff6e 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedBytesPropIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ColumnEncodedBytesPropIT.java
@@ -17,22 +17,29 @@
*/
package org.apache.phoenix.end2end;
+import static org.apache.phoenix.query.QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
+import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.query.QueryConstants;
+import org.apache.phoenix.schema.PColumn;
import org.apache.phoenix.schema.PTable;
import org.apache.phoenix.schema.PTable.QualifierEncodingScheme;
import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.PropertiesUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.junit.Test;
@@ -109,4 +116,57 @@ public class ColumnEncodedBytesPropIT extends ParallelStatsDisabledIT {
}
}
+ @Test
+ public void testAppendOnlySchema() throws SQLException {
+ Properties props = PropertiesUtil.deepCopy(TEST_PROPERTIES);
+ String dataTableFullName = SchemaUtil.getTableName("", generateUniqueName());
+ String view1 = SchemaUtil.getTableName("", generateUniqueName());
+ String view2 = SchemaUtil.getTableName("", generateUniqueName());
+ try (Connection conn = DriverManager.getConnection(getUrl(), props);) {
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE IMMUTABLE TABLE " + dataTableFullName +
+ " (id varchar not null, v1 varchar " +
+ " CONSTRAINT pk PRIMARY KEY (id)) COLUMN_ENCODED_BYTES=2, APPEND_ONLY_SCHEMA=true, UPDATE_CACHE_FREQUENCY=NEVER");
+ stmt.execute("ALTER TABLE " + dataTableFullName + " ADD v2 varchar");
+
+ stmt.execute("CREATE VIEW " + view1 + "(v3 varchar, v4 varchar)" +
+ " AS SELECT * FROM " + dataTableFullName + " WHERE v1='a'");
+ stmt.execute("CREATE VIEW " + view2 + "(v3 bigint, v4 integer)" +
+ " AS SELECT * FROM " + dataTableFullName + " WHERE v1='b'");
+ PTable v1 = PhoenixRuntime.getTable(conn, view1);
+ PTable v2 = PhoenixRuntime.getTable(conn, view1);
+ assertEquals(v1.getColumns().size(), v2.getColumns().size());
+ for (int i = 1; i < v1.getColumns().size(); i++) {
+ PColumn c1 = v1.getColumns().get(i);
+ PColumn c2 = v2.getColumns().get(i);
+ assertEquals(ENCODED_CQ_COUNTER_INITIAL_VALUE + i - Math.abs(Short.MIN_VALUE), Bytes.toShort(c1.getColumnQualifierBytes()));
+ assertEquals(ENCODED_CQ_COUNTER_INITIAL_VALUE + i - Math.abs(Short.MIN_VALUE), Bytes.toShort(c2.getColumnQualifierBytes()));
+ }
+
+ // add one more column to confirm disallowed now
+ try {
+ stmt.execute("ALTER TABLE " + dataTableFullName + " ADD v5 varchar");
+ fail();
+ }
+ catch (SQLException e) {
+ assertEquals(SQLExceptionCode.CANNOT_MUTATE_TABLE.getErrorCode(), e.getErrorCode());
+ }
+
+ conn.setAutoCommit(true);
+ stmt.execute("UPSERT INTO " + view1 + " VALUES('a','a','c','d','e')");
+ stmt.execute("UPSERT INTO " + view2 + " VALUES('b','b','c',1, 2)");
+ ResultSet rs1 = stmt.executeQuery("SELECT * FROM " + view1);
+ assertTrue(rs1.next());
+ assertEquals("a",rs1.getString(1));
+ assertEquals("d",rs1.getString(4));
+ assertEquals("e",rs1.getString(5));
+ assertFalse(rs1.next());
+ ResultSet rs2 = stmt.executeQuery("SELECT * FROM " + view2);
+ assertTrue(rs2.next());
+ assertEquals("b",rs2.getString(1));
+ assertEquals(1L,rs2.getLong(4));
+ assertEquals(2,rs2.getInt(5));
+ assertFalse(rs2.next());
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a72aad33/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
index b77f113..ff62c92 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/MetaDataEndpointImpl.java
@@ -2564,7 +2564,14 @@ public class MetaDataEndpointImpl extends MetaDataProtocol implements Coprocesso
+ Bytes.toString(schema) + ", table:"
+ Bytes.toString(table));
continue;
- }
+ }
+ /*
+ * Disallow adding columns to a base table with APPEND_ONLY_SCHEMA since this
+ * creates a gap in the column positions for every view (PHOENIX-4737).
+ */
+ if (!columnPutsForBaseTable.isEmpty() && view.isAppendOnlySchema()) {
+ return new MetaDataMutationResult(MutationCode.UNALLOWED_TABLE_MUTATION, EnvironmentEdgeManager.currentTimeMillis(), basePhysicalTable);
+ }
ColumnOrdinalPositionUpdateList ordinalPositionList = new ColumnOrdinalPositionUpdateList();
List<PColumn> viewPkCols = new ArrayList<>(view.getPKColumns());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/a72aad33/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 8f14680..4bdc5ff 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -93,6 +93,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TYPE;
import static org.apache.phoenix.query.QueryConstants.BASE_TABLE_BASE_COLUMN_COUNT;
import static org.apache.phoenix.query.QueryConstants.DEFAULT_COLUMN_FAMILY;
+import static org.apache.phoenix.query.QueryConstants.ENCODED_CQ_COUNTER_INITIAL_VALUE;
import static org.apache.phoenix.query.QueryServices.DROP_METADATA_ATTRIB;
import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_DROP_METADATA;
import static org.apache.phoenix.query.QueryServicesOptions.DEFAULT_RUN_UPDATE_STATS_ASYNC;
@@ -2376,7 +2377,9 @@ public class MetaDataClient {
cqCounterFamily = defaultFamilyName != null ? defaultFamilyName : DEFAULT_COLUMN_FAMILY;
}
}
- Integer encodedCQ = isPkColumn ? null : cqCounter.getNextQualifier(cqCounterFamily);
+ // Use position as column qualifier if APPEND_ONLY_SCHEMA to prevent gaps in
+ // the column encoding (PHOENIX-4737).
+ Integer encodedCQ = isPkColumn ? null : isAppendOnlySchema ? Integer.valueOf(ENCODED_CQ_COUNTER_INITIAL_VALUE + position) : cqCounter.getNextQualifier(cqCounterFamily);
byte[] columnQualifierBytes = null;
try {
columnQualifierBytes = EncodedColumnsUtil.getColumnQualifierBytes(columnDefName.getColumnName(), encodedCQ, encodingScheme, isPkColumn);
@@ -2387,7 +2390,7 @@ public class MetaDataClient {
.setTableName(tableName).build().buildException();
}
PColumn column = newColumn(position++, colDef, pkConstraint, defaultFamilyName, false, columnQualifierBytes, isImmutableRows);
- if (cqCounter.increment(cqCounterFamily)) {
+ if (!isAppendOnlySchema && cqCounter.increment(cqCounterFamily)) {
changedCqCounters.put(cqCounterFamily, cqCounter.getNextQualifier(cqCounterFamily));
}
if (SchemaUtil.isPKColumn(column)) {
@@ -3350,8 +3353,8 @@ public class MetaDataClient {
} else {
familyName = defaultColumnFamily;
}
- encodedCQ = cqCounterToUse.getNextQualifier(familyName);
- if (cqCounterToUse.increment(familyName)) {
+ encodedCQ = table.isAppendOnlySchema() ? Integer.valueOf(ENCODED_CQ_COUNTER_INITIAL_VALUE + position) : cqCounterToUse.getNextQualifier(familyName);
+ if (!table.isAppendOnlySchema() && cqCounterToUse.increment(familyName)) {
changedCqCounters.put(familyName,
cqCounterToUse.getNextQualifier(familyName));
}
[09/22] phoenix git commit: PHOENIX-3163 Split during global index
creation may cause ERROR 201 error (Sergey Soldatov)
Posted by ja...@apache.org.
PHOENIX-3163 Split during global index creation may cause ERROR 201 error (Sergey Soldatov)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/c27418b3
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/c27418b3
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/c27418b3
Branch: refs/heads/omid2
Commit: c27418b3089bd92f2056ef208eeca153b7779b7b
Parents: 7f91665
Author: James Taylor <jt...@salesforce.com>
Authored: Thu May 10 12:31:58 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Thu May 10 13:14:01 2018 -0700
----------------------------------------------------------------------
.../phoenix/compile/StatementContext.java | 9 +++
.../apache/phoenix/compile/UpsertCompiler.java | 1 +
.../phoenix/iterate/TableResultIterator.java | 71 +++++++++++---------
3 files changed, 50 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c27418b3/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
index 3e5c8f2..3ea5dd5 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
@@ -85,6 +85,7 @@ public class StatementContext {
private final ReadMetricQueue readMetricsQueue;
private final OverAllQueryMetrics overAllQueryMetrics;
private QueryLogger queryLogger;
+ private boolean isClientSideUpsertSelect;
public StatementContext(PhoenixStatement statement) {
this(statement, new Scan());
@@ -316,5 +317,13 @@ public class StatementContext {
public QueryLogger getQueryLogger() {
return queryLogger;
}
+
+ public boolean isClientSideUpsertSelect() {
+ return isClientSideUpsertSelect;
+ }
+
+ public void setClientSideUpsertSelect(boolean isClientSideUpsertSelect) {
+ this.isClientSideUpsertSelect = isClientSideUpsertSelect;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c27418b3/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
index 22119a3..30f0c18 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java
@@ -1252,6 +1252,7 @@ public class UpsertCompiler {
this.useServerTimestamp = useServerTimestamp;
this.maxSize = maxSize;
this.maxSizeBytes = maxSizeBytes;
+ queryPlan.getContext().setClientSideUpsertSelect(true);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/c27418b3/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
index f6902cc..8c80c28 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
@@ -41,7 +41,6 @@ import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.phoenix.cache.ServerCacheClient;
import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.coprocessor.HashJoinCacheNotFoundException;
@@ -92,7 +91,7 @@ public class TableResultIterator implements ResultIterator {
@GuardedBy("renewLeaseLock")
private long renewLeaseTime = 0;
-
+
private final Lock renewLeaseLock = new ReentrantLock();
private int retry;
@@ -114,12 +113,12 @@ public class TableResultIterator implements ResultIterator {
public static enum RenewLeaseStatus {
RENEWED, NOT_RENEWED, CLOSED, UNINITIALIZED, THRESHOLD_NOT_REACHED, LOCK_NOT_ACQUIRED, NOT_SUPPORTED
};
-
+
public TableResultIterator(MutationState mutationState, Scan scan, ScanMetricsHolder scanMetricsHolder,
long renewLeaseThreshold, QueryPlan plan, ParallelScanGrouper scanGrouper) throws SQLException {
this(mutationState, scan, scanMetricsHolder, renewLeaseThreshold, plan, scanGrouper, null);
}
-
+
public TableResultIterator(MutationState mutationState, Scan scan, ScanMetricsHolder scanMetricsHolder,
long renewLeaseThreshold, QueryPlan plan, ParallelScanGrouper scanGrouper,Map<ImmutableBytesPtr,ServerCache> caches) throws SQLException {
this.scan = scan;
@@ -133,7 +132,7 @@ public class TableResultIterator implements ResultIterator {
this.hashCacheClient = new HashCacheClient(plan.getContext().getConnection());
this.caches = caches;
this.retry=plan.getContext().getConnection().getQueryServices().getProps()
- .getInt(QueryConstants.HASH_JOIN_CACHE_RETRIES, QueryConstants.DEFAULT_HASH_JOIN_CACHE_RETRIES);
+ .getInt(QueryConstants.HASH_JOIN_CACHE_RETRIES, QueryConstants.DEFAULT_HASH_JOIN_CACHE_RETRIES);
}
@Override
@@ -156,7 +155,7 @@ public class TableResultIterator implements ResultIterator {
}
}
-
+
@Override
public Tuple next() throws SQLException {
try {
@@ -172,7 +171,7 @@ public class TableResultIterator implements ResultIterator {
try {
throw ServerUtil.parseServerException(e);
} catch(StaleRegionBoundaryCacheException | HashJoinCacheNotFoundException e1) {
- if(ScanUtil.isNonAggregateScan(scan)) {
+ if(ScanUtil.isNonAggregateScan(scan) && plan.getContext().getAggregationManager().isEmpty()) {
// For non aggregate queries if we get stale region boundary exception we can
// continue scanning from the next value of lasted fetched result.
Scan newScan = ScanUtil.newScan(scan);
@@ -189,34 +188,44 @@ public class TableResultIterator implements ResultIterator {
}
}
plan.getContext().getConnection().getQueryServices().clearTableRegionCache(htable.getTableName());
- if (e1 instanceof HashJoinCacheNotFoundException) {
- logger.debug(
- "Retrying when Hash Join cache is not found on the server ,by sending the cache again");
- if (retry <= 0) {
- throw e1;
- }
- retry--;
- try {
- Long cacheId = ((HashJoinCacheNotFoundException) e1).getCacheId();
+ if (e1 instanceof HashJoinCacheNotFoundException) {
+ logger.debug(
+ "Retrying when Hash Join cache is not found on the server ,by sending the cache again");
+ if (retry <= 0) {
+ throw e1;
+ }
+ retry--;
+ try {
+ Long cacheId = ((HashJoinCacheNotFoundException) e1).getCacheId();
- ServerCache cache = caches == null ? null :
- caches.get(new ImmutableBytesPtr(Bytes.toBytes(cacheId)));
+ ServerCache cache = caches == null ? null :
+ caches.get(new ImmutableBytesPtr(Bytes.toBytes(cacheId)));
- if (!hashCacheClient.addHashCacheToServer(newScan.getStartRow(),
- cache, plan.getTableRef().getTable())) {
- throw e1;
- }
- this.scanIterator = ((BaseQueryPlan) plan).iterator(caches, scanGrouper, newScan);
+ if (!hashCacheClient.addHashCacheToServer(newScan.getStartRow(),
+ cache, plan.getTableRef().getTable())) {
+ throw e1;
+ }
+ this.scanIterator = ((BaseQueryPlan) plan).iterator(caches, scanGrouper, newScan);
- } catch (Exception e2) {
- throw new SQLException(e2);
- }
- } else {
- this.scanIterator = plan.iterator(scanGrouper, newScan);
- }
+ } catch (Exception ex) {
+ throw ServerUtil.parseServerException(ex);
+ }
+ } else {
+ try {
+ if(plan.getContext().isClientSideUpsertSelect()) {
+ if(ScanUtil.isLocalIndex(newScan)) {
+ throw e;
+ }
+ this.scanIterator =
+ new ScanningResultIterator(htable.getScanner(newScan), newScan, scanMetricsHolder);
+ } else {
+ this.scanIterator = plan.iterator(scanGrouper, newScan);
+ }
+ } catch (IOException ex) {
+ throw ServerUtil.parseServerException(ex);
+ }
+ }
lastTuple = scanIterator.next();
- } else {
- throw e;
}
}
}
[18/22] phoenix git commit: PHOENIX-3163 Split during global index
creation may cause ERROR 201 error (addendum)
Posted by ja...@apache.org.
PHOENIX-3163 Split during global index creation may cause ERROR 201 error (addendum)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5b982f83
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5b982f83
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5b982f83
Branch: refs/heads/omid2
Commit: 5b982f83a1b7f8663d01bdb24353f547b969b91d
Parents: a72aad3
Author: James Taylor <jt...@salesforce.com>
Authored: Mon May 14 21:50:30 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Mon May 14 21:53:35 2018 -0700
----------------------------------------------------------------------
.../main/java/org/apache/phoenix/iterate/TableResultIterator.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5b982f83/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
index 8c80c28..06f612a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
@@ -226,6 +226,8 @@ public class TableResultIterator implements ResultIterator {
}
}
lastTuple = scanIterator.next();
+ } else {
+ throw e;
}
}
}
[17/22] phoenix git commit: PHOENIX-4701 Write client-side metrics
asynchronously to SYSTEM.LOG
Posted by ja...@apache.org.
PHOENIX-4701 Write client-side metrics asynchronously to SYSTEM.LOG
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/f73f98c7
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/f73f98c7
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/f73f98c7
Branch: refs/heads/omid2
Commit: f73f98c7a77fbe7e41930fc23b9fa3850fc0d260
Parents: d175c64
Author: Ankit Singhal <an...@gmail.com>
Authored: Mon May 14 13:37:53 2018 -0700
Committer: Ankit Singhal <an...@gmail.com>
Committed: Mon May 14 13:37:53 2018 -0700
----------------------------------------------------------------------
bin/hbase-site.xml | 4 +
.../apache/phoenix/end2end/QueryLoggerIT.java | 76 ++++++----
.../phoenix/monitoring/PhoenixMetricsIT.java | 28 ++--
.../phoenix/compile/StatementContext.java | 4 +-
.../apache/phoenix/execute/MutationState.java | 4 +-
.../phoenix/iterate/ChunkedResultIterator.java | 2 +-
.../phoenix/iterate/ParallelIterators.java | 3 +-
.../apache/phoenix/iterate/SerialIterators.java | 3 +-
.../apache/phoenix/jdbc/PhoenixConnection.java | 10 +-
.../apache/phoenix/jdbc/PhoenixResultSet.java | 54 ++++---
.../apache/phoenix/jdbc/PhoenixStatement.java | 21 ++-
.../java/org/apache/phoenix/log/LogLevel.java | 2 +-
.../java/org/apache/phoenix/log/LogWriter.java | 6 +-
.../org/apache/phoenix/log/QueryLogInfo.java | 38 +++--
.../org/apache/phoenix/log/QueryLogState.java | 22 ---
.../org/apache/phoenix/log/QueryLogger.java | 74 +++++++---
.../org/apache/phoenix/log/QueryLoggerUtil.java | 62 ++++----
.../org/apache/phoenix/log/QueryStatus.java | 22 +++
.../org/apache/phoenix/log/RingBufferEvent.java | 38 +++--
.../phoenix/log/RingBufferEventTranslator.java | 21 ++-
.../org/apache/phoenix/log/TableLogWriter.java | 144 +++++++++++--------
.../phoenix/mapreduce/PhoenixRecordReader.java | 13 +-
.../phoenix/monitoring/MemoryMetricsHolder.java | 1 -
.../apache/phoenix/monitoring/MetricType.java | 123 ++++++++++------
.../apache/phoenix/monitoring/MetricUtil.java | 30 ++++
.../phoenix/monitoring/MutationMetricQueue.java | 18 ++-
.../phoenix/monitoring/OverAllQueryMetrics.java | 21 ++-
.../phoenix/monitoring/ReadMetricQueue.java | 27 ++--
.../phoenix/monitoring/ScanMetricsHolder.java | 9 +-
.../monitoring/SpoolingMetricsHolder.java | 3 +-
.../monitoring/TaskExecutionMetricsHolder.java | 4 +-
.../query/ConnectionQueryServicesImpl.java | 7 +-
.../query/ConnectionlessQueryServicesImpl.java | 7 +-
.../apache/phoenix/query/QueryConstants.java | 124 ++--------------
.../org/apache/phoenix/query/QueryServices.java | 1 +
.../phoenix/query/QueryServicesOptions.java | 1 +
.../apache/phoenix/schema/MetaDataClient.java | 2 +-
.../java/org/apache/phoenix/util/QueryUtil.java | 7 +-
.../org/apache/phoenix/util/SchemaUtil.java | 4 +
.../iterate/SpoolingResultIteratorTest.java | 7 +-
.../hive/mapreduce/PhoenixRecordReader.java | 3 +-
41 files changed, 570 insertions(+), 480 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/bin/hbase-site.xml
----------------------------------------------------------------------
diff --git a/bin/hbase-site.xml b/bin/hbase-site.xml
index 0ab9fd8..2f360e2 100644
--- a/bin/hbase-site.xml
+++ b/bin/hbase-site.xml
@@ -24,4 +24,8 @@
<name>hbase.regionserver.wal.codec</name>
<value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value>
</property>
+ <property>
+ <name>phoenix.log.level</name>
+ <value>DEBUG</value>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryLoggerIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryLoggerIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryLoggerIT.java
index 940ba6f..618d7d9 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryLoggerIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/QueryLoggerIT.java
@@ -31,7 +31,6 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.START_TIME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TOTAL_EXECUTION_TIME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USER;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -53,8 +52,10 @@ import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDriver;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.log.LogLevel;
-import org.apache.phoenix.log.QueryLogState;
+import org.apache.phoenix.log.QueryStatus;
import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.EnvironmentEdge;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.ReadOnlyProps;
import org.junit.BeforeClass;
@@ -77,6 +78,19 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
}
+ private static class MyClock extends EnvironmentEdge {
+ public volatile long time;
+
+ public MyClock (long time) {
+ this.time = time;
+ }
+
+ @Override
+ public long currentTime() {
+ return time;
+ }
+ }
+
@Test
public void testDebugLogs() throws Exception {
@@ -97,12 +111,13 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
ResultSet explainRS = conn.createStatement().executeQuery("Explain " + query);
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
- rs = conn.createStatement().executeQuery(logQuery);
- boolean foundQueryLog = false;
int delay = 5000;
// sleep for sometime to let query log committed
Thread.sleep(delay);
+ rs = conn.createStatement().executeQuery(logQuery);
+ boolean foundQueryLog = false;
+
while (rs.next()) {
if (rs.getString(QUERY_ID).equals(queryId)) {
foundQueryLog = true;
@@ -113,11 +128,9 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
assertEquals(rs.getString(GLOBAL_SCAN_DETAILS), context.getScan().toJSON());
assertEquals(rs.getLong(NO_OF_RESULTS_ITERATED), 10);
assertEquals(rs.getString(QUERY), query);
- assertEquals(rs.getString(QUERY_STATUS), QueryLogState.COMPLETED.toString());
- assertTrue(System.currentTimeMillis() - rs.getTimestamp(START_TIME).getTime() > delay);
+ assertEquals(rs.getString(QUERY_STATUS), QueryStatus.COMPLETED.toString());
assertEquals(rs.getString(TENANT_ID), null);
- assertTrue(rs.getString(TOTAL_EXECUTION_TIME) != null);
- assertTrue(rs.getString(SCAN_METRICS_JSON).contains("scanMetrics"));
+ assertTrue(rs.getString(SCAN_METRICS_JSON)==null);
assertEquals(rs.getString(EXCEPTION_TRACE),null);
}else{
//confirm we are not logging system queries
@@ -140,7 +153,10 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
String query = "SELECT * FROM " + tableName;
int count=100;
for (int i = 0; i < count; i++) {
- conn.createStatement().executeQuery(query);
+ ResultSet rs = conn.createStatement().executeQuery(query);
+ while(rs.next()){
+
+ }
}
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
@@ -178,12 +194,12 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
}
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
- rs = conn.createStatement().executeQuery(logQuery);
- boolean foundQueryLog = false;
int delay = 5000;
// sleep for sometime to let query log committed
Thread.sleep(delay);
+ rs = conn.createStatement().executeQuery(logQuery);
+ boolean foundQueryLog = false;
while (rs.next()) {
if (rs.getString(QUERY_ID).equals(queryId)) {
foundQueryLog = true;
@@ -191,12 +207,10 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
assertEquals(rs.getString(CLIENT_IP), InetAddress.getLocalHost().getHostAddress());
assertEquals(rs.getString(EXPLAIN_PLAN), null);
assertEquals(rs.getString(GLOBAL_SCAN_DETAILS),null);
- assertEquals(rs.getLong(NO_OF_RESULTS_ITERATED), 0);
+ assertEquals(rs.getLong(NO_OF_RESULTS_ITERATED), 10);
assertEquals(rs.getString(QUERY), query);
- assertEquals(rs.getString(QUERY_STATUS),null);
- assertTrue(System.currentTimeMillis() - rs.getTimestamp(START_TIME).getTime() > delay);
+ assertEquals(rs.getString(QUERY_STATUS),QueryStatus.COMPLETED.toString());
assertEquals(rs.getString(TENANT_ID), null);
- assertTrue(rs.getString(TOTAL_EXECUTION_TIME) == null);
}
}
assertTrue(foundQueryLog);
@@ -222,12 +236,12 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
}
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
- rs = conn.createStatement().executeQuery(logQuery);
- boolean foundQueryLog = false;
int delay = 5000;
// sleep for sometime to let query log committed
Thread.sleep(delay);
+ rs = conn.createStatement().executeQuery(logQuery);
+ boolean foundQueryLog = false;
while (rs.next()) {
if (rs.getString(QUERY_ID).equals(queryId)) {
foundQueryLog = true;
@@ -255,7 +269,9 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
props.setProperty(QueryServices.LOG_LEVEL, loglevel.name());
Connection conn = DriverManager.getConnection(getUrl(),props);
assertEquals(conn.unwrap(PhoenixConnection.class).getLogLevel(),loglevel);
-
+ final MyClock clock = new MyClock(100);
+ EnvironmentEdgeManager.injectEdge(clock);
+ try{
String query = "SELECT * FROM " + tableName +" where V = ?";
PreparedStatement pstmt = conn.prepareStatement(query);
@@ -270,12 +286,12 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
ResultSet explainRS = conn.createStatement()
.executeQuery("Explain " + "SELECT * FROM " + tableName + " where V = 'value5'");
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
- rs = conn.createStatement().executeQuery(logQuery);
- boolean foundQueryLog = false;
int delay = 5000;
-
+
// sleep for sometime to let query log committed
Thread.sleep(delay);
+ rs = conn.createStatement().executeQuery(logQuery);
+ boolean foundQueryLog = false;
while (rs.next()) {
if (rs.getString(QUERY_ID).equals(queryId)) {
foundQueryLog = true;
@@ -286,14 +302,18 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
assertEquals(rs.getString(GLOBAL_SCAN_DETAILS), context.getScan().toJSON());
assertEquals(rs.getLong(NO_OF_RESULTS_ITERATED), 1);
assertEquals(rs.getString(QUERY), query);
- assertEquals(rs.getString(QUERY_STATUS), QueryLogState.COMPLETED.toString());
- assertTrue(System.currentTimeMillis() - rs.getTimestamp(START_TIME).getTime() > delay);
+ assertEquals(rs.getString(QUERY_STATUS), QueryStatus.COMPLETED.toString());
+ assertTrue(LogLevel.TRACE == loglevel ? rs.getString(SCAN_METRICS_JSON).contains("scanMetrics")
+ : rs.getString(SCAN_METRICS_JSON) == null);
+ assertEquals(rs.getTimestamp(START_TIME).getTime(),100);
assertEquals(rs.getString(TENANT_ID), null);
- assertTrue(rs.getString(TOTAL_EXECUTION_TIME) != null);
}
}
assertTrue(foundQueryLog);
conn.close();
+ }finally{
+ EnvironmentEdgeManager.injectEdge(null);
+ }
}
@@ -315,14 +335,14 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
assertEquals(e.getErrorCode(), SQLExceptionCode.TABLE_UNDEFINED.getErrorCode());
}
String logQuery = "SELECT * FROM " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"";
- ResultSet rs = conn.createStatement().executeQuery(logQuery);
- boolean foundQueryLog = false;
int delay = 5000;
// sleep for sometime to let query log committed
Thread.sleep(delay);
+ ResultSet rs = conn.createStatement().executeQuery(logQuery);
+ boolean foundQueryLog = false;
while (rs.next()) {
- if (QueryLogState.FAILED.name().equals(rs.getString(QUERY_STATUS))) {
+ if (QueryStatus.FAILED.name().equals(rs.getString(QUERY_STATUS))) {
foundQueryLog = true;
assertEquals(rs.getString(USER), System.getProperty("user.name"));
assertEquals(rs.getString(CLIENT_IP), InetAddress.getLocalHost().getHostAddress());
@@ -331,8 +351,6 @@ public class QueryLoggerIT extends BaseUniqueNamesOwnClusterIT {
assertEquals(rs.getLong(NO_OF_RESULTS_ITERATED), 0);
assertEquals(rs.getString(QUERY), query);
assertTrue(rs.getString(EXCEPTION_TRACE).contains(SQLExceptionCode.TABLE_UNDEFINED.getMessage()));
- assertTrue(System.currentTimeMillis() - rs.getTimestamp(START_TIME).getTime() > delay);
- assertTrue(rs.getString(TOTAL_EXECUTION_TIME) != null);
}
}
assertTrue(foundQueryLog);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
index e45ddcd..73cdf0a 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
@@ -10,6 +10,10 @@
package org.apache.phoenix.monitoring;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_FAILED_QUERY_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_BYTES_REGION_SERVER_RESULTS;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_MILLS_BETWEEN_NEXTS;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_RPC_CALLS;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_SCANNED_REGIONS;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HCONNECTIONS_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_FAILED_COUNT;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_SIZE;
@@ -29,19 +33,6 @@ import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SELECT_SQ
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SPOOL_FILE_COUNTER;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_END_TO_END_TIME;
import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_EXECUTION_TIME;
-
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_RPC_CALLS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_REMOTE_RPC_CALLS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_MILLS_BETWEEN_NEXTS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_NOT_SERVING_REGION_EXCEPTION;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_BYTES_REGION_SERVER_RESULTS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_BYTES_IN_REMOTE_RESULTS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_SCANNED_REGIONS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_RPC_RETRIES;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_REMOTE_RPC_RETRIES;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_ROWS_SCANNED;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_ROWS_FILTERED;
-
import static org.apache.phoenix.monitoring.MetricType.MEMORY_CHUNK_BYTES;
import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
import static org.apache.phoenix.monitoring.MetricType.TASK_EXECUTED_COUNTER;
@@ -73,11 +64,12 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
import org.apache.phoenix.exception.SQLExceptionCode;
+import org.apache.phoenix.jdbc.LoggingPhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.jdbc.PhoenixDriver;
import org.apache.phoenix.jdbc.PhoenixMetricsLog;
import org.apache.phoenix.jdbc.PhoenixResultSet;
-import org.apache.phoenix.jdbc.LoggingPhoenixConnection;
+import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.ReadOnlyProps;
@@ -109,6 +101,7 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
props.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, String.valueOf(true));
// disable renewing leases as this will force spooling to happen.
props.put(QueryServices.RENEW_LEASE_ENABLED, String.valueOf(false));
+ props.put(QueryServices.LOG_LEVEL, LogLevel.DEBUG.toString());
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
// need the non-test driver for some tests that check number of hconnections, etc.
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
@@ -377,6 +370,7 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
insertRowsInTable(tableName, numRows);
Properties props = new Properties();
props.setProperty(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, "false");
+ props.setProperty(QueryServices.LOG_LEVEL, LogLevel.OFF.name());
Connection conn = DriverManager.getConnection(getUrl(), props);
ResultSet rs = conn.createStatement().executeQuery("SELECT * FROM " + tableName);
while (rs.next()) {}
@@ -706,7 +700,7 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
private void changeInternalStateForTesting(PhoenixResultSet rs) {
// get and set the internal state for testing purposes.
- ReadMetricQueue testMetricsQueue = new TestReadMetricsQueue(true);
+ ReadMetricQueue testMetricsQueue = new TestReadMetricsQueue(LogLevel.DEBUG);
StatementContext ctx = (StatementContext)Whitebox.getInternalState(rs, "context");
Whitebox.setInternalState(ctx, "readMetricsQueue", testMetricsQueue);
Whitebox.setInternalState(rs, "readMetricsQueue", testMetricsQueue);
@@ -772,8 +766,8 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
private class TestReadMetricsQueue extends ReadMetricQueue {
- public TestReadMetricsQueue(boolean isRequestMetricsEnabled) {
- super(isRequestMetricsEnabled);
+ public TestReadMetricsQueue(LogLevel connectionLogLevel) {
+ super(connectionLogLevel);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
index 3ea5dd5..4358ee3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
@@ -135,8 +135,8 @@ public class StatementContext {
this.dataColumns = this.currentTable == null ? Collections.<PColumn, Integer> emptyMap() : Maps
.<PColumn, Integer> newLinkedHashMap();
this.subqueryResults = Maps.<SelectStatement, Object> newHashMap();
- this.readMetricsQueue = new ReadMetricQueue(isRequestMetricsEnabled);
- this.overAllQueryMetrics = new OverAllQueryMetrics(isRequestMetricsEnabled);
+ this.readMetricsQueue = new ReadMetricQueue(connection.getLogLevel());
+ this.overAllQueryMetrics = new OverAllQueryMetrics(connection.getLogLevel());
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index 18f4fea..f3a383e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -39,8 +39,8 @@ import javax.annotation.concurrent.Immutable;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.client.Delete;
-import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
@@ -1181,7 +1181,7 @@ public class MutationState implements SQLCloseable {
numFailedMutations = uncommittedStatementIndexes.length;
GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations);
} finally {
- MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
+ MutationMetric mutationsMetric = new MutationMetric(connection.getLogLevel(),numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
try {
if (cache!=null)
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
index 8595fd4..acb6c04 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
@@ -147,7 +147,7 @@ public class ChunkedResultIterator implements PeekingResultIterator {
String tableName = tableRef.getTable().getPhysicalName().getString();
ReadMetricQueue readMetrics = context.getReadMetricsQueue();
ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics, tableName, scan,
- readMetrics.isRequestMetricsEnabled());
+ context.getConnection().getLogLevel());
long renewLeaseThreshold = context.getConnection().getQueryServices().getRenewLeaseThresholdMilliSeconds();
ResultIterator singleChunkResultIterator =
new SingleChunkResultIterator(new TableResultIterator(mutationState, scan,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 41d278d..262ae44 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -107,11 +107,10 @@ public class ParallelIterators extends BaseResultIterators {
context.getOverallQueryMetrics().updateNumParallelScans(numScans);
GLOBAL_NUM_PARALLEL_SCANS.update(numScans);
final long renewLeaseThreshold = context.getConnection().getQueryServices().getRenewLeaseThresholdMilliSeconds();
- boolean isRequestMetricsEnabled = readMetrics.isRequestMetricsEnabled();
for (final ScanLocator scanLocation : scanLocations) {
final Scan scan = scanLocation.getScan();
final ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics, physicalTableName,
- scan, isRequestMetricsEnabled);
+ scan, context.getConnection().getLogLevel());
final TaskExecutionMetricsHolder taskMetrics = new TaskExecutionMetricsHolder(readMetrics, physicalTableName);
final TableResultIterator tableResultItr =
context.getConnection().getTableResultIteratorFactory().newIterator(
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
index c13fcdb..1693421 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
@@ -171,7 +171,6 @@ public class SerialIterators extends BaseResultIterators {
return EMPTY_ITERATOR;
}
ReadMetricQueue readMetrics = context.getReadMetricsQueue();
- boolean isRequestMetricsEnabled = readMetrics.isRequestMetricsEnabled();
while (index < scans.size()) {
Scan currentScan = scans.get(index++);
if (remainingOffset != null) {
@@ -179,7 +178,7 @@ public class SerialIterators extends BaseResultIterators {
}
ScanMetricsHolder scanMetricsHolder =
ScanMetricsHolder.getInstance(readMetrics, tableName, currentScan,
- isRequestMetricsEnabled);
+ context.getConnection().getLogLevel());
TableResultIterator itr =
new TableResultIterator(mutationState, currentScan, scanMetricsHolder,
renewLeaseThreshold, plan, scanGrouper, caches);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index d3626f8..312d17b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -364,9 +364,10 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
function.getTenantId())));
}
};
- this.isRequestLevelMetricsEnabled = JDBCUtil
- .isCollectingRequestLevelMetricsEnabled(url, info,
- this.services.getProps());
+ this.logLevel= LogLevel.valueOf(this.services.getProps().get(QueryServices.LOG_LEVEL,
+ QueryServicesOptions.DEFAULT_LOGGING_LEVEL));
+ this.isRequestLevelMetricsEnabled = JDBCUtil.isCollectingRequestLevelMetricsEnabled(url, info,
+ this.services.getProps());
this.mutationState = mutationState == null ? newMutationState(maxSize,
maxSizeBytes) : new MutationState(mutationState);
this.metaData = metaData;
@@ -380,8 +381,7 @@ public class PhoenixConnection implements Connection, MetaDataMutated, SQLClosea
this.scannerQueue = new LinkedBlockingQueue<>();
this.tableResultIteratorFactory = new DefaultTableResultIteratorFactory();
this.isRunningUpgrade = isRunningUpgrade;
- this.logLevel= LogLevel.valueOf(this.services.getProps().get(QueryServices.LOG_LEVEL,
- QueryServicesOptions.DEFAULT_LOGGING_LEVEL));
+
this.logSamplingRate = Double.parseDouble(this.services.getProps().get(QueryServices.LOG_SAMPLE_RATE,
QueryServicesOptions.DEFAULT_LOG_SAMPLE_RATE));
GLOBAL_OPEN_PHOENIX_CONNECTIONS.increment();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
index 153fa08..84816a0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
@@ -51,8 +51,8 @@ import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.log.QueryLogInfo;
-import org.apache.phoenix.log.QueryLogState;
import org.apache.phoenix.log.QueryLogger;
+import org.apache.phoenix.log.QueryStatus;
import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.monitoring.OverAllQueryMetrics;
import org.apache.phoenix.monitoring.ReadMetricQueue;
@@ -76,8 +76,6 @@ import org.apache.phoenix.util.SQLCloseable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
@@ -133,9 +131,9 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
private Long count = 0L;
- private QueryLogState logStatus = QueryLogState.COMPLETED;
+ private Object exception;
+
- private RuntimeException exception;
public PhoenixResultSet(ResultIterator resultIterator, RowProjector rowProjector, StatementContext ctx) throws SQLException {
this.rowProjector = rowProjector;
@@ -144,7 +142,7 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
this.statement = context.getStatement();
this.readMetricsQueue = context.getReadMetricsQueue();
this.overAllQueryMetrics = context.getOverallQueryMetrics();
- this.queryLogger = context.getQueryLogger();
+ this.queryLogger = context.getQueryLogger() != null ? context.getQueryLogger() : QueryLogger.NO_OP_INSTANCE;
}
@Override
@@ -181,6 +179,19 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
statement.getResultSets().remove(this);
overAllQueryMetrics.endQuery();
overAllQueryMetrics.stopResultSetWatch();
+ if (!queryLogger.isSynced()) {
+ if(this.exception==null){
+ queryLogger.log(QueryLogInfo.QUERY_STATUS_I,QueryStatus.COMPLETED.toString());
+ }
+ queryLogger.log(QueryLogInfo.NO_OF_RESULTS_ITERATED_I, count);
+ if (queryLogger.isDebugEnabled()) {
+ queryLogger.log(QueryLogInfo.SCAN_METRICS_JSON_I,
+ readMetricsQueue.getScanMetricsHolderList().toString());
+ readMetricsQueue.getScanMetricsHolderList().clear();
+ }
+ // if not already synced , like closing before result set exhausted
+ queryLogger.sync(getReadMetrics(), getOverAllRequestReadMetrics());
+ }
}
}
@@ -799,36 +810,33 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable {
}
rowProjector.reset();
} catch (RuntimeException e) {
- this.logStatus=QueryLogState.FAILED;
// FIXME: Expression.evaluate does not throw SQLException
// so this will unwrap throws from that.
+ queryLogger.log(QueryLogInfo.QUERY_STATUS_I, QueryStatus.FAILED.toString());
+ if (queryLogger.isDebugEnabled()) {
+ queryLogger.log(QueryLogInfo.EXCEPTION_TRACE_I, Throwables.getStackTraceAsString(e));
+ }
this.exception = e;
if (e.getCause() instanceof SQLException) {
throw (SQLException) e.getCause();
}
throw e;
}finally{
- if (currentRow == null && queryLogger != null ) {
+ if (this.exception!=null) {
+ queryLogger.log(QueryLogInfo.NO_OF_RESULTS_ITERATED_I, count);
if (queryLogger.isDebugEnabled()) {
- Builder<QueryLogInfo, Object> queryLogBuilder = ImmutableMap.builder();
- queryLogBuilder.put(QueryLogInfo.NO_OF_RESULTS_ITERATED_I, count);
- queryLogBuilder.put(QueryLogInfo.TOTAL_EXECUTION_TIME_I,
- System.currentTimeMillis() - queryLogger.getStartTime());
- queryLogBuilder.put(QueryLogInfo.SCAN_METRICS_JSON_I,
+ queryLogger.log(QueryLogInfo.SCAN_METRICS_JSON_I,
readMetricsQueue.getScanMetricsHolderList().toString());
- if (this.exception != null) {
- queryLogBuilder.put(QueryLogInfo.EXCEPTION_TRACE_I,
- Throwables.getStackTraceAsString(this.exception));
- }
readMetricsQueue.getScanMetricsHolderList().clear();
- queryLogger.log(logStatus, queryLogBuilder.build());
+ }
+ if (queryLogger != null) {
+ queryLogger.sync(getReadMetrics(), getOverAllRequestReadMetrics());
}
}
- }
- if (currentRow == null) {
-
- overAllQueryMetrics.endQuery();
- overAllQueryMetrics.stopResultSetWatch();
+ if (currentRow == null) {
+ overAllQueryMetrics.endQuery();
+ overAllQueryMetrics.stopResultSetWatch();
+ }
}
return currentRow != null;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index 25b9fb0..015f04c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -93,7 +93,7 @@ import org.apache.phoenix.iterate.MaterializedResultIterator;
import org.apache.phoenix.iterate.ParallelScanGrouper;
import org.apache.phoenix.iterate.ResultIterator;
import org.apache.phoenix.log.QueryLogInfo;
-import org.apache.phoenix.log.QueryLogState;
+import org.apache.phoenix.log.QueryStatus;
import org.apache.phoenix.log.QueryLogger;
import org.apache.phoenix.log.QueryLoggerUtil;
import org.apache.phoenix.optimize.Cost;
@@ -190,8 +190,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.math.IntMath;
@@ -319,10 +317,8 @@ public class PhoenixStatement implements Statement, SQLCloseable {
StatementContext context = plan.getContext();
context.setQueryLogger(queryLogger);
if(queryLogger.isDebugEnabled()){
- Builder<QueryLogInfo, Object> queryLogBuilder = ImmutableMap.builder();
- queryLogBuilder.put(QueryLogInfo.EXPLAIN_PLAN_I, QueryUtil.getExplainPlan(resultIterator));
- queryLogBuilder.put(QueryLogInfo.GLOBAL_SCAN_DETAILS_I, context.getScan()!=null?context.getScan().toString():null);
- queryLogger.log(QueryLogState.COMPILED, queryLogBuilder.build());
+ queryLogger.log(QueryLogInfo.EXPLAIN_PLAN_I, QueryUtil.getExplainPlan(resultIterator));
+ queryLogger.log(QueryLogInfo.GLOBAL_SCAN_DETAILS_I, context.getScan()!=null?context.getScan().toString():null);
}
context.getOverallQueryMetrics().startQuery();
PhoenixResultSet rs = newResultSet(resultIterator, plan.getProjector(), plan.getContext());
@@ -351,6 +347,7 @@ public class PhoenixStatement implements Statement, SQLCloseable {
}
throw e;
}catch (RuntimeException e) {
+
// FIXME: Expression.evaluate does not throw SQLException
// so this will unwrap throws from that.
if (e.getCause() instanceof SQLException) {
@@ -367,11 +364,9 @@ public class PhoenixStatement implements Statement, SQLCloseable {
}, PhoenixContextExecutor.inContext());
}catch (Exception e) {
if (queryLogger.isDebugEnabled()) {
- Builder<QueryLogInfo, Object> queryLogBuilder = ImmutableMap.builder();
- queryLogBuilder.put(QueryLogInfo.TOTAL_EXECUTION_TIME_I,
- System.currentTimeMillis() - queryLogger.getStartTime());
- queryLogBuilder.put(QueryLogInfo.EXCEPTION_TRACE_I, Throwables.getStackTraceAsString(e));
- queryLogger.log(QueryLogState.FAILED, queryLogBuilder.build());
+ queryLogger.log(QueryLogInfo.EXCEPTION_TRACE_I, Throwables.getStackTraceAsString(e));
+ queryLogger.log(QueryLogInfo.QUERY_STATUS_I, QueryStatus.FAILED.toString());
+ queryLogger.sync(null, null);
}
Throwables.propagateIfInstanceOf(e, SQLException.class);
Throwables.propagate(e);
@@ -1781,7 +1776,7 @@ public class PhoenixStatement implements Statement, SQLCloseable {
}
QueryLogger queryLogger = QueryLogger.getInstance(connection,isSystemTable);
QueryLoggerUtil.logInitialDetails(queryLogger, connection.getTenantId(),
- connection.getQueryServices(), sql, queryLogger.getStartTime(), getParameters());
+ connection.getQueryServices(), sql, getParameters());
return queryLogger;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/LogLevel.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/LogLevel.java b/phoenix-core/src/main/java/org/apache/phoenix/log/LogLevel.java
index 5792658..269b4f4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/LogLevel.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/LogLevel.java
@@ -18,5 +18,5 @@
package org.apache.phoenix.log;
public enum LogLevel {
- OFF, INFO, DEBUG, TRACE
+ OFF,INFO, DEBUG, TRACE
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/LogWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/LogWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/log/LogWriter.java
index 817f9ec..dab03e7 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/LogWriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/LogWriter.java
@@ -31,16 +31,18 @@ public interface LogWriter {
* @param event
* @throws SQLException
* @throws IOException
+ * @throws ClassNotFoundException
*/
- void write(RingBufferEvent event) throws SQLException, IOException;
+ void write(RingBufferEvent event) throws SQLException, IOException, ClassNotFoundException;
/**
* will be called when disruptor is getting shutdown
*
* @throws IOException
+ * @throws SQLException
*/
- void close() throws IOException;
+ void close() throws IOException, SQLException;
/**
* if writer is closed and cannot write further event
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogInfo.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogInfo.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogInfo.java
index 87de267..fb38ba2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogInfo.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogInfo.java
@@ -28,8 +28,8 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_ID;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_STATUS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCAN_METRICS_JSON;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.START_TIME;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TOTAL_EXECUTION_TIME;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USER;
import org.apache.phoenix.schema.types.PDataType;
@@ -40,29 +40,27 @@ import org.apache.phoenix.schema.types.PVarchar;
public enum QueryLogInfo {
- CLIENT_IP_I(CLIENT_IP, QueryLogState.STARTED, LogLevel.INFO, PVarchar.INSTANCE),
- QUERY_I(QUERY,QueryLogState.STARTED, LogLevel.INFO,PVarchar.INSTANCE),
- BIND_PARAMETERS_I(BIND_PARAMETERS,QueryLogState.STARTED, LogLevel.TRACE,PVarchar.INSTANCE),
- QUERY_ID_I(QUERY_ID,QueryLogState.STARTED, LogLevel.INFO,PVarchar.INSTANCE),
- TENANT_ID_I(TENANT_ID,QueryLogState.STARTED, LogLevel.INFO,PVarchar.INSTANCE),
- START_TIME_I(START_TIME,QueryLogState.STARTED, LogLevel.INFO,PTimestamp.INSTANCE),
- USER_I(USER,QueryLogState.STARTED, LogLevel.INFO,PVarchar.INSTANCE),
- EXPLAIN_PLAN_I(EXPLAIN_PLAN,QueryLogState.COMPILED, LogLevel.DEBUG,PVarchar.INSTANCE),
- GLOBAL_SCAN_DETAILS_I(GLOBAL_SCAN_DETAILS,QueryLogState.COMPILED, LogLevel.DEBUG,PVarchar.INSTANCE),
- NO_OF_RESULTS_ITERATED_I(NO_OF_RESULTS_ITERATED,QueryLogState.COMPLETED, LogLevel.DEBUG,PLong.INSTANCE),
- EXCEPTION_TRACE_I(EXCEPTION_TRACE,QueryLogState.COMPLETED, LogLevel.DEBUG,PVarchar.INSTANCE),
- QUERY_STATUS_I(QUERY_STATUS,QueryLogState.COMPLETED, LogLevel.DEBUG,PVarchar.INSTANCE),
- TOTAL_EXECUTION_TIME_I(TOTAL_EXECUTION_TIME,QueryLogState.COMPLETED, LogLevel.DEBUG,PLong.INSTANCE),
- SCAN_METRICS_JSON_I(SCAN_METRICS_JSON,QueryLogState.COMPLETED, LogLevel.DEBUG,PVarchar.INSTANCE);
+ CLIENT_IP_I(CLIENT_IP, LogLevel.INFO, PVarchar.INSTANCE),
+ QUERY_I(QUERY, LogLevel.INFO,PVarchar.INSTANCE),
+ BIND_PARAMETERS_I(BIND_PARAMETERS, LogLevel.TRACE,PVarchar.INSTANCE),
+ QUERY_ID_I(QUERY_ID, LogLevel.INFO,PVarchar.INSTANCE),
+ TENANT_ID_I(TENANT_ID, LogLevel.INFO,PVarchar.INSTANCE),
+ START_TIME_I(START_TIME, LogLevel.INFO,PTimestamp.INSTANCE),
+ USER_I(USER, LogLevel.INFO,PVarchar.INSTANCE),
+ EXPLAIN_PLAN_I(EXPLAIN_PLAN,LogLevel.DEBUG,PVarchar.INSTANCE),
+ GLOBAL_SCAN_DETAILS_I(GLOBAL_SCAN_DETAILS, LogLevel.DEBUG,PVarchar.INSTANCE),
+ NO_OF_RESULTS_ITERATED_I(NO_OF_RESULTS_ITERATED, LogLevel.INFO,PLong.INSTANCE),
+ EXCEPTION_TRACE_I(EXCEPTION_TRACE, LogLevel.DEBUG,PVarchar.INSTANCE),
+ QUERY_STATUS_I(QUERY_STATUS, LogLevel.INFO,PVarchar.INSTANCE),
+ SCAN_METRICS_JSON_I(SCAN_METRICS_JSON, LogLevel.TRACE,PVarchar.INSTANCE),
+ TABLE_NAME_I(TABLE_NAME, LogLevel.DEBUG,PVarchar.INSTANCE);
public final String columnName;
- public final QueryLogState logState;
public final LogLevel logLevel;
public final PDataType dataType;
- private QueryLogInfo(String columnName, QueryLogState logState, LogLevel logLevel, PDataType dataType) {
+ private QueryLogInfo(String columnName, LogLevel logLevel, PDataType dataType) {
this.columnName = columnName;
- this.logState=logState;
this.logLevel=logLevel;
this.dataType=dataType;
}
@@ -71,10 +69,6 @@ public enum QueryLogInfo {
return columnName;
}
- public QueryLogState getLogState() {
- return logState;
- }
-
public LogLevel getLogLevel() {
return logLevel;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogState.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogState.java
deleted file mode 100644
index e27f0e8..0000000
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogState.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.phoenix.log;
-
-public enum QueryLogState {
- STARTED, PLAN, COMPILED, EXECUTION, COMPLETED,FAILED
-}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogger.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogger.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogger.java
index b2fb235..ef5559c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogger.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLogger.java
@@ -17,13 +17,17 @@
*/
package org.apache.phoenix.log;
+import java.util.Map;
import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.util.EnvironmentEdgeManager;
import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
import io.netty.util.internal.ThreadLocalRandom;
@@ -34,15 +38,17 @@ public class QueryLogger {
private final ThreadLocal<RingBufferEventTranslator> threadLocalTranslator = new ThreadLocal<>();
private QueryLoggerDisruptor queryDisruptor;
private String queryId;
- private Long startTime;
private LogLevel logLevel;
+ private Builder<QueryLogInfo, Object> queryLogBuilder = ImmutableMap.builder();
+ private boolean isSynced;
private static final Log LOG = LogFactory.getLog(QueryLoggerDisruptor.class);
private QueryLogger(PhoenixConnection connection) {
this.queryId = UUID.randomUUID().toString();
this.queryDisruptor = connection.getQueryServices().getQueryDisruptor();
- this.startTime = System.currentTimeMillis();
logLevel = connection.getLogLevel();
+ log(QueryLogInfo.QUERY_ID_I, queryId);
+ log(QueryLogInfo.START_TIME_I, EnvironmentEdgeManager.currentTimeMillis());
}
private QueryLogger() {
@@ -58,21 +64,32 @@ public class QueryLogger {
return result;
}
- private static final QueryLogger NO_OP_INSTANCE = new QueryLogger() {
+ public static final QueryLogger NO_OP_INSTANCE = new QueryLogger() {
@Override
- public void log(QueryLogState logState, ImmutableMap<QueryLogInfo, Object> map) {
+ public void log(QueryLogInfo queryLogInfo, Object info) {
}
-
+
@Override
- public boolean isDebugEnabled(){
+ public boolean isDebugEnabled() {
return false;
}
-
+
@Override
- public boolean isInfoEnabled(){
+ public boolean isInfoEnabled() {
return false;
}
+
+ @Override
+ public void sync(
+ Map<String, Map<MetricType, Long>> readMetrics, Map<MetricType, Long> overAllMetrics) {
+
+ }
+
+ @Override
+ public boolean isSynced(){
+ return true;
+ }
};
public static QueryLogger getInstance(PhoenixConnection connection, boolean isSystemTable) {
@@ -82,14 +99,14 @@ public class QueryLogger {
}
/**
- * Add query log in the table, columns will be logged depending upon the connection logLevel
- * @param logState State of the query
- * @param map Value of the map should be in format of the corresponding data type
+ * Add query log in the table, columns will be logged depending upon the connection logLevel
*/
- public void log(QueryLogState logState, ImmutableMap<QueryLogInfo, Object> map) {
- final RingBufferEventTranslator translator = getCachedTranslator();
- translator.setQueryInfo(logState, map, logLevel);
- publishLogs(translator);
+ public void log(QueryLogInfo queryLogInfo, Object info) {
+ try {
+ queryLogBuilder.put(queryLogInfo, info);
+ } catch (Exception e) {
+ LOG.warn("Unable to add log info because of " + e.getMessage());
+ }
}
private boolean publishLogs(RingBufferEventTranslator translator) {
@@ -102,13 +119,6 @@ public class QueryLogger {
}
/**
- * Start time when the logger was started, if {@link LogLevel#OFF} then it's the current time
- */
- public Long getStartTime() {
- return startTime != null ? startTime : System.currentTimeMillis();
- }
-
- /**
* Is debug logging currently enabled?
* Call this method to prevent having to perform expensive operations (for example, String concatenation) when the log level is more than debug.
*/
@@ -117,7 +127,8 @@ public class QueryLogger {
}
private boolean isLevelEnabled(LogLevel logLevel){
- return this.logLevel != null ? logLevel.ordinal() <= this.logLevel.ordinal() : false;
+ return this.logLevel != null && logLevel != LogLevel.OFF ? logLevel.ordinal() <= this.logLevel.ordinal()
+ : false;
}
/**
@@ -142,4 +153,21 @@ public class QueryLogger {
return this.queryId;
}
+
+ public void sync(Map<String, Map<MetricType, Long>> readMetrics, Map<MetricType, Long> overAllMetrics) {
+ if (!isSynced) {
+ isSynced = true;
+ final RingBufferEventTranslator translator = getCachedTranslator();
+ translator.setQueryInfo(logLevel, queryLogBuilder.build(), readMetrics, overAllMetrics);
+ publishLogs(translator);
+ }
+ }
+
+ /**
+ * Is Synced already
+ */
+ public boolean isSynced(){
+ return this.isSynced;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
index d5c4878..21917ed 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
@@ -25,48 +25,36 @@ import org.apache.commons.lang.StringUtils;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.schema.PName;
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableMap.Builder;
-
public class QueryLoggerUtil {
+
public static void logInitialDetails(QueryLogger queryLogger, PName tenantId, ConnectionQueryServices queryServices,
- String query, long startTime, List<Object> bindParameters) {
+ String query, List<Object> bindParameters) {
try {
- queryLogger.log(QueryLogState.STARTED,
- getInitialDetails(tenantId, queryServices, query, startTime, bindParameters));
+ String clientIP;
+ try {
+ clientIP = InetAddress.getLocalHost().getHostAddress();
+ } catch (UnknownHostException e) {
+ clientIP = "UnknownHost";
+ }
+
+ if (clientIP != null) {
+ queryLogger.log(QueryLogInfo.CLIENT_IP_I, clientIP);
+ }
+ if (query != null) {
+ queryLogger.log(QueryLogInfo.QUERY_I, query);
+ }
+ if (bindParameters != null) {
+ queryLogger.log(QueryLogInfo.BIND_PARAMETERS_I, StringUtils.join(bindParameters, ","));
+ }
+ if (tenantId != null) {
+ queryLogger.log(QueryLogInfo.TENANT_ID_I, tenantId.getString());
+ }
+
+ queryLogger.log(QueryLogInfo.USER_I, queryServices.getUserName() != null ? queryServices.getUserName()
+ : queryServices.getUser().getShortName());
} catch (Exception e) {
- // Ignore for now
- }
-
- }
-
- private static ImmutableMap<QueryLogInfo, Object> getInitialDetails(PName tenantId,
- ConnectionQueryServices queryServices, String query, long startTime, List<Object> bindParameters) {
- Builder<QueryLogInfo, Object> queryLogBuilder = ImmutableMap.builder();
- String clientIP;
- try {
- clientIP = InetAddress.getLocalHost().getHostAddress();
- } catch (UnknownHostException e) {
- clientIP = "UnknownHost";
+ // Ignore
}
-
- if (clientIP != null) {
- queryLogBuilder.put(QueryLogInfo.CLIENT_IP_I, clientIP);
- }
- if (query != null) {
- queryLogBuilder.put(QueryLogInfo.QUERY_I, query);
- }
- queryLogBuilder.put(QueryLogInfo.START_TIME_I, startTime);
- if (bindParameters != null) {
- queryLogBuilder.put(QueryLogInfo.BIND_PARAMETERS_I, StringUtils.join(bindParameters, ","));
- }
- if (tenantId != null) {
- queryLogBuilder.put(QueryLogInfo.TENANT_ID_I, tenantId.getString());
- }
-
- queryLogBuilder.put(QueryLogInfo.USER_I, queryServices.getUserName() != null ? queryServices.getUserName()
- : queryServices.getUser().getShortName());
- return queryLogBuilder.build();
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/QueryStatus.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryStatus.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryStatus.java
new file mode 100644
index 0000000..0e634c1
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryStatus.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.log;
+
+public enum QueryStatus {
+ COMPILED, COMPLETED,FAILED
+}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEvent.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEvent.java b/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEvent.java
index 96e4bf9..8854e68 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEvent.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEvent.java
@@ -17,14 +17,19 @@
*/
package org.apache.phoenix.log;
+import java.util.Map;
+
+import org.apache.phoenix.monitoring.MetricType;
+
import com.google.common.collect.ImmutableMap;
import com.lmax.disruptor.EventFactory;
class RingBufferEvent {
private String queryId;
- private QueryLogState logState;
private LogLevel connectionLogLevel;
private ImmutableMap<QueryLogInfo, Object> queryInfo;
+ private Map<String, Map<MetricType, Long>> readMetrics;
+ private Map<MetricType, Long> overAllMetrics;
public static final Factory FACTORY = new Factory();
@@ -40,7 +45,6 @@ import com.lmax.disruptor.EventFactory;
}
public void clear() {
- this.logState=null;
this.queryInfo=null;
this.queryId=null;
}
@@ -53,10 +57,6 @@ import com.lmax.disruptor.EventFactory;
public static Factory getFactory() {
return FACTORY;
}
-
- public QueryLogState getLogState() {
- return logState;
- }
public void setQueryInfo(ImmutableMap<QueryLogInfo, Object> queryInfo) {
this.queryInfo=queryInfo;
@@ -73,12 +73,6 @@ import com.lmax.disruptor.EventFactory;
}
- public void setLogState(QueryLogState logState) {
- this.logState=logState;
-
- }
-
-
public LogLevel getConnectionLogLevel() {
return connectionLogLevel;
}
@@ -88,6 +82,26 @@ import com.lmax.disruptor.EventFactory;
this.connectionLogLevel = connectionLogLevel;
}
+
+ public Map<String, Map<MetricType, Long>> getReadMetrics() {
+ return readMetrics;
+ }
+
+
+ public void setReadMetrics(Map<String, Map<MetricType, Long>> readMetrics) {
+ this.readMetrics = readMetrics;
+ }
+
+
+ public Map<MetricType, Long> getOverAllMetrics() {
+ return overAllMetrics;
+ }
+
+
+ public void setOverAllMetrics(Map<MetricType, Long> overAllMetrics) {
+ this.overAllMetrics = overAllMetrics;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEventTranslator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEventTranslator.java b/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEventTranslator.java
index 653ddd6..742f8e1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEventTranslator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/RingBufferEventTranslator.java
@@ -17,14 +17,19 @@
*/
package org.apache.phoenix.log;
+import java.util.Map;
+
+import org.apache.phoenix.monitoring.MetricType;
+
import com.google.common.collect.ImmutableMap;
import com.lmax.disruptor.EventTranslator;
class RingBufferEventTranslator implements EventTranslator<RingBufferEvent> {
private String queryId;
- private QueryLogState logState;
private ImmutableMap<QueryLogInfo, Object> queryInfo;
private LogLevel connectionLogLevel;
+ private Map<String, Map<MetricType, Long>> readMetrics;
+ private Map<MetricType, Long> overAllMetrics;
public RingBufferEventTranslator(String queryId) {
this.queryId=queryId;
@@ -34,20 +39,22 @@ class RingBufferEventTranslator implements EventTranslator<RingBufferEvent> {
public void translateTo(RingBufferEvent event, long sequence) {
event.setQueryId(queryId);
event.setQueryInfo(queryInfo);
- event.setLogState(logState);
+ event.setReadMetrics(readMetrics);
+ event.setOverAllMetrics(overAllMetrics);
event.setConnectionLogLevel(connectionLogLevel);
clear();
}
private void clear() {
- setQueryInfo(null,null,null);
+ setQueryInfo(null,null,null,null);
}
- public void setQueryInfo(QueryLogState logState, ImmutableMap<QueryLogInfo, Object> queryInfo,
- LogLevel connectionLogLevel) {
+ public void setQueryInfo(LogLevel logLevel, ImmutableMap<QueryLogInfo, Object> queryInfo, Map<String, Map<MetricType, Long>> readMetrics,
+ Map<MetricType, Long> overAllMetrics) {
this.queryInfo = queryInfo;
- this.logState = logState;
- this.connectionLogLevel = connectionLogLevel;
+ this.connectionLogLevel = logLevel;
+ this.readMetrics = readMetrics;
+ this.overAllMetrics=overAllMetrics;
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/log/TableLogWriter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/TableLogWriter.java b/phoenix-core/src/main/java/org/apache/phoenix/log/TableLogWriter.java
index c102855..0209951 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/TableLogWriter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/TableLogWriter.java
@@ -21,23 +21,17 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCH
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE;
import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
import java.sql.SQLException;
-import java.util.Map.Entry;
+import java.util.HashMap;
+import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hbase.client.Connection;
-import org.apache.hadoop.hbase.client.ConnectionFactory;
-import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Table;
-import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.phoenix.expression.Determinism;
-import org.apache.phoenix.expression.LiteralExpression;
-import org.apache.phoenix.query.QueryConstants;
-import org.apache.phoenix.util.ByteUtil;
-import org.apache.phoenix.util.SchemaUtil;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.util.QueryUtil;
import com.google.common.collect.ImmutableMap;
@@ -49,75 +43,111 @@ public class TableLogWriter implements LogWriter {
private static final Log LOG = LogFactory.getLog(LogWriter.class);
private Connection connection;
private boolean isClosed;
- private Table table;
+ private PreparedStatement upsertStatement;
private Configuration config;
+ private Map<MetricType,Integer> metricOrdinals=new HashMap<MetricType,Integer>();
public TableLogWriter(Configuration configuration) {
- this.config = configuration;
- try {
- this.connection = ConnectionFactory.createConnection(configuration);
- table = this.connection.getTable(SchemaUtil.getPhysicalTableName(
- SchemaUtil.getTableNameAsBytes(SYSTEM_CATALOG_SCHEMA, SYSTEM_LOG_TABLE), config));
- } catch (Exception e) {
- LOG.warn("Unable to initiate LogWriter for writing query logs to table");
+ this.config=configuration;
+ }
+
+ private PreparedStatement buildUpsertStatement(Connection conn) throws SQLException {
+ StringBuilder buf = new StringBuilder("UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"(");
+ int queryLogEntries=0;
+ for (QueryLogInfo info : QueryLogInfo.values()) {
+ buf.append(info.columnName);
+ buf.append(',');
+ queryLogEntries++;
+ }
+ for (MetricType metric : MetricType.values()) {
+ if (metric.logLevel() != LogLevel.OFF) {
+ metricOrdinals.put(metric, ++queryLogEntries);
+ buf.append(metric.columnName());
+ buf.append(',');
+ }
+ }
+ buf.setLength(buf.length()-1);
+ buf.append(") VALUES (");
+ for (int i = 0; i < QueryLogInfo.values().length; i++) {
+ buf.append("?,");
}
+ for (MetricType metric : MetricType.values()) {
+ if (metric.logLevel() != LogLevel.OFF) {
+ buf.append("?,");
+ }
+ }
+ buf.setLength(buf.length()-1);
+ buf.append(")");
+ return conn.prepareStatement(buf.toString());
}
@Override
- public void write(RingBufferEvent event) throws SQLException, IOException {
- if(isClosed()){
+ public void write(RingBufferEvent event) throws SQLException, IOException, ClassNotFoundException {
+ if (isClosed()) {
LOG.warn("Unable to commit query log as Log committer is already closed");
return;
}
- if (table == null || connection == null) {
- LOG.warn("Unable to commit query log as connection was not initiated ");
- return;
+ if (connection == null) {
+ synchronized (this) {
+ if (connection == null) {
+ connection = QueryUtil.getConnectionForQueryLog(this.config);
+ this.upsertStatement = buildUpsertStatement(connection);
+ }
+ }
}
- ImmutableMap<QueryLogInfo, Object> queryInfo=event.getQueryInfo();
- ImmutableBytesWritable ptr = new ImmutableBytesWritable();
- Put put =new Put(Bytes.toBytes(event.getQueryId()));
- for (Entry<QueryLogInfo, Object> entry : queryInfo.entrySet()) {
- if (entry.getKey().logLevel.ordinal() <= event.getConnectionLogLevel().ordinal()) {
- LiteralExpression expression = LiteralExpression.newConstant(entry.getValue(), entry.getKey().dataType,
- Determinism.ALWAYS);
- expression.evaluate(null, ptr);
- put.addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES, Bytes.toBytes(entry.getKey().columnName),
- ByteUtil.copyKeyBytesIfNecessary(ptr));
+
+ ImmutableMap<QueryLogInfo, Object> queryInfoMap = event.getQueryInfo();
+ for (QueryLogInfo info : QueryLogInfo.values()) {
+ if (queryInfoMap.containsKey(info) && info.logLevel.ordinal() <= event.getConnectionLogLevel().ordinal()) {
+ upsertStatement.setObject(info.ordinal() + 1, queryInfoMap.get(info));
+ } else {
+ upsertStatement.setObject(info.ordinal() + 1, null);
}
}
-
- if (QueryLogInfo.QUERY_STATUS_I.logLevel.ordinal() <= event.getConnectionLogLevel().ordinal()
- && (event.getLogState() == QueryLogState.COMPLETED || event.getLogState() == QueryLogState.FAILED)) {
- LiteralExpression expression = LiteralExpression.newConstant(event.getLogState().toString(),
- QueryLogInfo.QUERY_STATUS_I.dataType, Determinism.ALWAYS);
- expression.evaluate(null, ptr);
- put.addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES,
- Bytes.toBytes(QueryLogInfo.QUERY_STATUS_I.columnName), ByteUtil.copyKeyBytesIfNecessary(ptr));
+ Map<MetricType, Long> overAllMetrics = event.getOverAllMetrics();
+ Map<String, Map<MetricType, Long>> readMetrics = event.getReadMetrics();
+
+ for (MetricType metric : MetricType.values()) {
+ if (overAllMetrics != null && overAllMetrics.containsKey(metric)
+ && metric.isLoggingEnabled(event.getConnectionLogLevel())) {
+ upsertStatement.setObject(metricOrdinals.get(metric), overAllMetrics.get(metric));
+ } else {
+ if (metric.logLevel() != LogLevel.OFF) {
+ upsertStatement.setObject(metricOrdinals.get(metric), null);
+ }
+ }
+ }
+
+ if (readMetrics != null && !readMetrics.isEmpty()) {
+ for (Map.Entry<String, Map<MetricType, Long>> entry : readMetrics.entrySet()) {
+ upsertStatement.setObject(QueryLogInfo.TABLE_NAME_I.ordinal() + 1, entry.getKey());
+ for (MetricType metric : entry.getValue().keySet()) {
+ if (metric.isLoggingEnabled(event.getConnectionLogLevel())) {
+ upsertStatement.setObject(metricOrdinals.get(metric), entry.getValue().get(metric));
+ }
+ }
+ upsertStatement.executeUpdate();
+ }
+ } else {
+ upsertStatement.executeUpdate();
}
- put.addColumn(QueryConstants.DEFAULT_COLUMN_FAMILY_BYTES, QueryConstants.EMPTY_COLUMN_BYTES, QueryConstants.EMPTY_COLUMN_VALUE_BYTES);
- table.put(put);
-
+ connection.commit();
}
@Override
public void close() throws IOException {
- if(isClosed()){
- return;
- }
- isClosed=true;
+ if (isClosed()) { return; }
+ isClosed = true;
try {
- if (table != null) {
- table.close();
- }
- if (connection != null && !connection.isClosed()) {
- //It should internally close all the statements
+ if (connection != null) {
+ // It should internally close all the statements
connection.close();
}
- } catch (IOException e) {
+ } catch (SQLException e) {
// TODO Ignore?
}
}
-
+
public boolean isClosed(){
return isClosed;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
index ec1b451..58c048b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/PhoenixRecordReader.java
@@ -35,7 +35,15 @@ import org.apache.hadoop.util.ReflectionUtils;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.coprocessor.BaseScannerRegionObserver;
-import org.apache.phoenix.iterate.*;
+import org.apache.phoenix.iterate.ConcatResultIterator;
+import org.apache.phoenix.iterate.LookAheadResultIterator;
+import org.apache.phoenix.iterate.MapReduceParallelScanGrouper;
+import org.apache.phoenix.iterate.PeekingResultIterator;
+import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.iterate.RoundRobinResultIterator;
+import org.apache.phoenix.iterate.SequenceResultIterator;
+import org.apache.phoenix.iterate.TableResultIterator;
+import org.apache.phoenix.iterate.TableSnapshotResultIterator;
import org.apache.phoenix.jdbc.PhoenixResultSet;
import org.apache.phoenix.mapreduce.util.PhoenixConfigurationUtil;
import org.apache.phoenix.monitoring.ReadMetricQueue;
@@ -112,7 +120,6 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
services.clearTableRegionCache(tableNameBytes);
long renewScannerLeaseThreshold = queryPlan.getContext().getConnection().getQueryServices().getRenewLeaseThresholdMilliSeconds();
- boolean isRequestMetricsEnabled = readMetrics.isRequestMetricsEnabled();
for (Scan scan : scans) {
// For MR, skip the region boundary check exception if we encounter a split. ref: PHOENIX-2599
scan.setAttribute(BaseScannerRegionObserver.SKIP_REGION_BOUNDARY_CHECK, Bytes.toBytes(true));
@@ -120,7 +127,7 @@ public class PhoenixRecordReader<T extends DBWritable> extends RecordReader<Null
PeekingResultIterator peekingResultIterator;
ScanMetricsHolder scanMetricsHolder =
ScanMetricsHolder.getInstance(readMetrics, tableName, scan,
- isRequestMetricsEnabled);
+ queryPlan.getContext().getConnection().getLogLevel());
if (snapshotName != null) {
// result iterator to read snapshots
final TableSnapshotResultIterator tableSnapshotResultIterator = new TableSnapshotResultIterator(configuration, scan,
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MemoryMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MemoryMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MemoryMetricsHolder.java
index 0e82ce4..daa0bba 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MemoryMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MemoryMetricsHolder.java
@@ -26,7 +26,6 @@ import static org.apache.phoenix.monitoring.MetricType.MEMORY_WAIT_TIME;
public class MemoryMetricsHolder {
private final CombinableMetric memoryChunkSizeMetric;
private final CombinableMetric memoryWaitTimeMetric;
- public static final MemoryMetricsHolder NO_OP_INSTANCE = new MemoryMetricsHolder(new ReadMetricQueue(false), null);
public MemoryMetricsHolder(ReadMetricQueue readMetrics, String tableName) {
this.memoryChunkSizeMetric = readMetrics.allotMetric(MEMORY_CHUNK_BYTES, tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
index ef6eceb..8e1de66 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricType.java
@@ -17,62 +17,74 @@
*/
package org.apache.phoenix.monitoring;
+import org.apache.phoenix.log.LogLevel;
+import org.apache.phoenix.schema.types.PDataType;
+import org.apache.phoenix.schema.types.PLong;
+
+
+/**
+ * Keeping {@link LogLevel#OFF} for metrics which are calculated globally only and doesn't need to be logged in SYSTEM.LOG
+ */
public enum MetricType {
- NO_OP_METRIC("no", "No op metric"),
+ NO_OP_METRIC("no", "No op metric",LogLevel.OFF, PLong.INSTANCE),
// mutation (write) related metrics
- MUTATION_BATCH_SIZE("ms", "Number of mutations in the batch"),
- MUTATION_BYTES("mb", "Size of mutations in bytes"),
- MUTATION_COMMIT_TIME("mt", "Time it took to commit a batch of mutations"),
- MUTATION_BATCH_FAILED_SIZE("mfs", "Number of mutations that failed to be committed"),
- MUTATION_SQL_COUNTER("msc", "Counter for number of mutation sql statements"),
+ MUTATION_BATCH_SIZE("ms", "Number of mutations in the batch",LogLevel.OFF, PLong.INSTANCE),
+ MUTATION_BYTES("mb", "Size of mutations in bytes",LogLevel.OFF, PLong.INSTANCE),
+ MUTATION_COMMIT_TIME("mt", "Time it took to commit a batch of mutations",LogLevel.OFF, PLong.INSTANCE),
+ MUTATION_BATCH_FAILED_SIZE("mfs", "Number of mutations that failed to be committed",LogLevel.OFF, PLong.INSTANCE),
+ MUTATION_SQL_COUNTER("msc", "Counter for number of mutation sql statements",LogLevel.OFF, PLong.INSTANCE),
// query (read) related metrics
- QUERY_TIME("qt", "Query times"),
- QUERY_TIMEOUT_COUNTER("qo", "Number of times query timed out"),
- QUERY_FAILED_COUNTER("qf", "Number of times query failed"),
- NUM_PARALLEL_SCANS("ps", "Number of scans that were executed in parallel"),
- SCAN_BYTES("sb", "Number of bytes read by scans"),
- SELECT_SQL_COUNTER("sc", "Counter for number of sql queries"),
+ QUERY_TIME("qt", "Query times",LogLevel.OFF, PLong.INSTANCE),
+ QUERY_TIMEOUT_COUNTER("qo", "Number of times query timed out",LogLevel.DEBUG, PLong.INSTANCE),
+ QUERY_FAILED_COUNTER("qf", "Number of times query failed",LogLevel.DEBUG, PLong.INSTANCE),
+ NUM_PARALLEL_SCANS("ps", "Number of scans that were executed in parallel",LogLevel.DEBUG, PLong.INSTANCE),
+ SCAN_BYTES("sb", "Number of bytes read by scans",LogLevel.OFF, PLong.INSTANCE),
+ SELECT_SQL_COUNTER("sc", "Counter for number of sql queries",LogLevel.OFF, PLong.INSTANCE),
// task metrics
- TASK_QUEUE_WAIT_TIME("tw", "Time in milliseconds tasks had to wait in the queue of the thread pool executor"),
- TASK_END_TO_END_TIME("tee", "Time in milliseconds spent by tasks from creation to completion"),
- TASK_EXECUTION_TIME("tx", "Time in milliseconds tasks took to execute"),
- TASK_EXECUTED_COUNTER("te", "Counter for number of tasks submitted to the thread pool executor"),
- TASK_REJECTED_COUNTER("tr", "Counter for number of tasks that were rejected by the thread pool executor"),
+ TASK_QUEUE_WAIT_TIME("tw", "Time in milliseconds tasks had to wait in the queue of the thread pool executor",LogLevel.DEBUG, PLong.INSTANCE),
+ TASK_END_TO_END_TIME("tee", "Time in milliseconds spent by tasks from creation to completion",LogLevel.DEBUG, PLong.INSTANCE),
+ TASK_EXECUTION_TIME("tx", "Time in milliseconds tasks took to execute",LogLevel.DEBUG, PLong.INSTANCE),
+ TASK_EXECUTED_COUNTER("te", "Counter for number of tasks submitted to the thread pool executor",LogLevel.DEBUG, PLong.INSTANCE),
+ TASK_REJECTED_COUNTER("tr", "Counter for number of tasks that were rejected by the thread pool executor",LogLevel.DEBUG, PLong.INSTANCE),
// spool metrics
- SPOOL_FILE_SIZE("ss", "Size of spool files created in bytes"),
- SPOOL_FILE_COUNTER("sn", "Number of spool files created"),
+ SPOOL_FILE_SIZE("ss", "Size of spool files created in bytes",LogLevel.DEBUG, PLong.INSTANCE),
+ SPOOL_FILE_COUNTER("sn", "Number of spool files created",LogLevel.DEBUG, PLong.INSTANCE),
// misc metrics
- MEMORY_CHUNK_BYTES("mc", "Number of bytes allocated by the memory manager"),
- MEMORY_WAIT_TIME("mw", "Number of milliseconds threads needed to wait for memory to be allocated through memory manager"),
- CACHE_REFRESH_SPLITS_COUNTER("cr", "Number of times cache was refreshed because of splits"),
- WALL_CLOCK_TIME_MS("tq", "Wall clock time elapsed for the overall query execution"),
- RESULT_SET_TIME_MS("tn", "Wall clock time elapsed for reading all records using resultSet.next()"),
- OPEN_PHOENIX_CONNECTIONS_COUNTER("o", "Number of open phoenix connections"),
- QUERY_SERVICES_COUNTER("cqs", "Number of ConnectionQueryServicesImpl instantiated"),
- HCONNECTIONS_COUNTER("h", "Number of HConnections created by phoenix driver"),
+ MEMORY_CHUNK_BYTES("mc", "Number of bytes allocated by the memory manager",LogLevel.DEBUG, PLong.INSTANCE),
+ MEMORY_WAIT_TIME("mw", "Number of milliseconds threads needed to wait for memory to be allocated through memory manager",LogLevel.DEBUG, PLong.INSTANCE),
+ CACHE_REFRESH_SPLITS_COUNTER("cr", "Number of times cache was refreshed because of splits",LogLevel.DEBUG, PLong.INSTANCE),
+ WALL_CLOCK_TIME_MS("tq", "Wall clock time elapsed for the overall query execution",LogLevel.INFO, PLong.INSTANCE),
+ RESULT_SET_TIME_MS("tn", "Wall clock time elapsed for reading all records using resultSet.next()",LogLevel.INFO, PLong.INSTANCE),
+ OPEN_PHOENIX_CONNECTIONS_COUNTER("o", "Number of open phoenix connections",LogLevel.OFF, PLong.INSTANCE),
+ QUERY_SERVICES_COUNTER("cqs", "Number of ConnectionQueryServicesImpl instantiated",LogLevel.OFF, PLong.INSTANCE),
+ HCONNECTIONS_COUNTER("h", "Number of HConnections created by phoenix driver",LogLevel.OFF, PLong.INSTANCE),
PHOENIX_CONNECTIONS_THROTTLED_COUNTER("ct", "Number of client Phoenix connections prevented from opening " +
- "because there are already too many to that target cluster."),
- PHOENIX_CONNECTIONS_ATTEMPTED_COUNTER("ca","Number of requests for Phoenix connections, whether successful or not."),
+ "because there are already too many to that target cluster.",LogLevel.OFF, PLong.INSTANCE),
+ PHOENIX_CONNECTIONS_ATTEMPTED_COUNTER("ca","Number of requests for Phoenix connections, whether successful or not.",LogLevel.OFF, PLong.INSTANCE),
// hbase metrics
- COUNT_RPC_CALLS("rp", "Number of RPC calls"),
- COUNT_REMOTE_RPC_CALLS("rr", "Number of remote RPC calls"),
- COUNT_MILLS_BETWEEN_NEXTS("n", "Sum of milliseconds between sequential next calls"),
- COUNT_NOT_SERVING_REGION_EXCEPTION("nsr", "Number of NotServingRegionException caught"),
- COUNT_BYTES_REGION_SERVER_RESULTS("rs", "Number of bytes in Result objects from region servers"),
- COUNT_BYTES_IN_REMOTE_RESULTS("rrs", "Number of bytes in Result objects from remote region servers"),
- COUNT_SCANNED_REGIONS("rg", "Number of regions scanned"),
- COUNT_RPC_RETRIES("rpr", "Number of RPC retries"),
- COUNT_REMOTE_RPC_RETRIES("rrr", "Number of remote RPC retries"),
- COUNT_ROWS_SCANNED("ws", "Number of rows scanned"),
- COUNT_ROWS_FILTERED("wf", "Number of rows filtered");
+ COUNT_RPC_CALLS("rp", "Number of RPC calls",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_REMOTE_RPC_CALLS("rr", "Number of remote RPC calls",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_MILLS_BETWEEN_NEXTS("n", "Sum of milliseconds between sequential next calls",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_NOT_SERVING_REGION_EXCEPTION("nsr", "Number of NotServingRegionException caught",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_BYTES_REGION_SERVER_RESULTS("rs", "Number of bytes in Result objects from region servers",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_BYTES_IN_REMOTE_RESULTS("rrs", "Number of bytes in Result objects from remote region servers",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_SCANNED_REGIONS("rg", "Number of regions scanned",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_RPC_RETRIES("rpr", "Number of RPC retries",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_REMOTE_RPC_RETRIES("rrr", "Number of remote RPC retries",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_ROWS_SCANNED("ws", "Number of rows scanned",LogLevel.DEBUG, PLong.INSTANCE),
+ COUNT_ROWS_FILTERED("wf", "Number of rows filtered",LogLevel.DEBUG,PLong.INSTANCE);
private final String description;
private final String shortName;
+ private LogLevel logLevel;
+ private PDataType dataType;
- private MetricType(String shortName, String description) {
+ private MetricType(String shortName, String description, LogLevel logLevel, PDataType dataType) {
this.shortName = shortName;
this.description = description;
+ this.logLevel=logLevel;
+ this.dataType=dataType;
}
public String description() {
@@ -82,5 +94,34 @@ public enum MetricType {
public String shortName() {
return shortName;
}
+
+ public LogLevel logLevel() {
+ return logLevel;
+ }
+
+ public PDataType dataType() {
+ return dataType;
+ }
+
+ public String columnName() {
+ return name();
+ }
+
+ public boolean isLoggingEnabled(LogLevel connectionLogLevel){
+ return logLevel() != LogLevel.OFF && (logLevel().ordinal() <= connectionLogLevel.ordinal());
+ }
+ public static String getMetricColumnsDetails() {
+ StringBuilder buffer=new StringBuilder();
+ for(MetricType metric:MetricType.values()){
+ if (metric.logLevel() != LogLevel.OFF) {
+ buffer.append(metric.columnName());
+ buffer.append(" ");
+ buffer.append(metric.dataType.getSqlTypeName());
+ buffer.append(",");
+ }
+ }
+ return buffer.toString();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
new file mode 100644
index 0000000..1e5ac08
--- /dev/null
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.phoenix.monitoring;
+
+import org.apache.phoenix.log.LogLevel;
+import org.apache.phoenix.monitoring.CombinableMetric.NoOpRequestMetric;
+
+public class MetricUtil {
+
+ public static CombinableMetric getCombinableMetric(LogLevel connectionLogLevel, MetricType type) {
+ if (!type.isLoggingEnabled(connectionLogLevel)) { return NoOpRequestMetric.INSTANCE; }
+ return new CombinableMetricImpl(type);
+ }
+
+}
[14/22] phoenix git commit: Revert "PHOENIX-4726 save index build
timestamp -- for SYNC case only"
Posted by ja...@apache.org.
Revert "PHOENIX-4726 save index build timestamp -- for SYNC case only"
This reverts commit 7ebba49d000bb221a8de45e70b4c0afb3b9355a3.
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/09d0b3a5
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/09d0b3a5
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/09d0b3a5
Branch: refs/heads/omid2
Commit: 09d0b3a5fd2bffd5dc9a54a84b71e92686fecea7
Parents: 7ebba49
Author: Vincent Poon <vi...@apache.org>
Authored: Mon May 14 10:25:16 2018 -0700
Committer: Vincent Poon <vi...@apache.org>
Committed: Mon May 14 10:25:16 2018 -0700
----------------------------------------------------------------------
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 1 -
.../org/apache/phoenix/schema/MetaDataClient.java | 17 -----------------
2 files changed, 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/09d0b3a5/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 2e12ca2..55de772 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -316,7 +316,6 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String ASYNC_CREATED_DATE = "ASYNC_CREATED_DATE";
public static final String SEQUENCE_TABLE_TYPE = SYSTEM_SEQUENCE_TABLE;
- public static final String SYNC_INDEX_CREATED_DATE = "SYNC_INDEX_CREATED_DATE";
public static final String SYSTEM_MUTEX_TABLE_NAME = "MUTEX";
public static final String SYSTEM_MUTEX_NAME = SchemaUtil.getTableName(QueryConstants.SYSTEM_SCHEMA_NAME, SYSTEM_MUTEX_TABLE_NAME);
public static final TableName SYSTEM_MUTEX_HBASE_TABLE_NAME = TableName.valueOf(SYSTEM_MUTEX_NAME);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/09d0b3a5/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 20ac732..009289b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -74,7 +74,6 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.RETURN_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SALT_BUCKETS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SORT_ORDER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.STORE_NULLS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYNC_INDEX_CREATED_DATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
@@ -264,14 +263,6 @@ public class MetaDataClient {
TABLE_NAME + "," +
ASYNC_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
") VALUES (?, ?, ?, ?)";
-
- private static final String SET_INDEX_SYNC_CREATED_DATE =
- "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
- TENANT_ID + "," +
- TABLE_SCHEM + "," +
- TABLE_NAME + "," +
- SYNC_INDEX_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
- ") VALUES (?, ?, ?, ?)";
private static final String CREATE_TABLE =
"UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
TENANT_ID + "," +
@@ -2687,14 +2678,6 @@ public class MetaDataClient {
setAsync.setString(3, tableName);
setAsync.setDate(4, asyncCreatedDate);
setAsync.execute();
- } else {
- Date syncCreatedDate = new Date(EnvironmentEdgeManager.currentTimeMillis());
- PreparedStatement setSync = connection.prepareStatement(SET_INDEX_SYNC_CREATED_DATE);
- setSync.setString(1, tenantIdStr);
- setSync.setString(2, schemaName);
- setSync.setString(3, tableName);
- setSync.setDate(4, syncCreatedDate);
- setSync.execute();
}
tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
connection.rollback();
[16/22] phoenix git commit: PHOENIX-4701 Write client-side metrics
asynchronously to SYSTEM.LOG
Posted by ja...@apache.org.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
index 3de2be1..1256f5c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
@@ -27,6 +27,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
+import org.apache.phoenix.log.LogLevel;
+
/**
* Queue that tracks various writes/mutations related phoenix request metrics.
*/
@@ -81,12 +83,16 @@ public class MutationMetricQueue {
* Class that holds together the various metrics associated with mutations.
*/
public static class MutationMetric {
- private final CombinableMetric numMutations = new CombinableMetricImpl(MUTATION_BATCH_SIZE);
- private final CombinableMetric mutationsSizeBytes = new CombinableMetricImpl(MUTATION_BYTES);
- private final CombinableMetric totalCommitTimeForMutations = new CombinableMetricImpl(MUTATION_COMMIT_TIME);
- private final CombinableMetric numFailedMutations = new CombinableMetricImpl(MUTATION_BATCH_FAILED_SIZE);
-
- public MutationMetric(long numMutations, long mutationsSizeBytes, long commitTimeForMutations, long numFailedMutations) {
+ private final CombinableMetric numMutations;;
+ private final CombinableMetric mutationsSizeBytes;
+ private final CombinableMetric totalCommitTimeForMutations;
+ private final CombinableMetric numFailedMutations;
+
+ public MutationMetric(LogLevel connectionLogLevel, long numMutations, long mutationsSizeBytes, long commitTimeForMutations, long numFailedMutations) {
+ this.numMutations = MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BATCH_SIZE);
+ this.mutationsSizeBytes =MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BYTES);
+ this.totalCommitTimeForMutations =MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_COMMIT_TIME);
+ this.numFailedMutations = MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BATCH_FAILED_SIZE);
this.numMutations.change(numMutations);
this.mutationsSizeBytes.change(mutationsSizeBytes);
this.totalCommitTimeForMutations.change(commitTimeForMutations);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
index b995267..3121ecd 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
@@ -27,7 +27,7 @@ import static org.apache.phoenix.monitoring.MetricType.WALL_CLOCK_TIME_MS;
import java.util.HashMap;
import java.util.Map;
-import org.apache.phoenix.monitoring.CombinableMetric.NoOpRequestMetric;
+import org.apache.phoenix.log.LogLevel;
/**
* Class that represents the overall metrics associated with a query being executed by the phoenix.
@@ -42,16 +42,15 @@ public class OverAllQueryMetrics {
private final CombinableMetric queryFailed;
private final CombinableMetric cacheRefreshedDueToSplits;
- public OverAllQueryMetrics(boolean isMetricsEnabled) {
- queryWatch = new MetricsStopWatch(isMetricsEnabled);
- resultSetWatch = new MetricsStopWatch(isMetricsEnabled);
- numParallelScans = isMetricsEnabled ? new CombinableMetricImpl(NUM_PARALLEL_SCANS) : NoOpRequestMetric.INSTANCE;
- wallClockTimeMS = isMetricsEnabled ? new CombinableMetricImpl(WALL_CLOCK_TIME_MS) : NoOpRequestMetric.INSTANCE;
- resultSetTimeMS = isMetricsEnabled ? new CombinableMetricImpl(RESULT_SET_TIME_MS) : NoOpRequestMetric.INSTANCE;
- queryTimedOut = isMetricsEnabled ? new CombinableMetricImpl(QUERY_TIMEOUT_COUNTER) : NoOpRequestMetric.INSTANCE;
- queryFailed = isMetricsEnabled ? new CombinableMetricImpl(QUERY_FAILED_COUNTER) : NoOpRequestMetric.INSTANCE;
- cacheRefreshedDueToSplits = isMetricsEnabled ? new CombinableMetricImpl(CACHE_REFRESH_SPLITS_COUNTER)
- : NoOpRequestMetric.INSTANCE;
+ public OverAllQueryMetrics(LogLevel connectionLogLevel) {
+ queryWatch = new MetricsStopWatch(WALL_CLOCK_TIME_MS.isLoggingEnabled(connectionLogLevel));
+ resultSetWatch = new MetricsStopWatch(RESULT_SET_TIME_MS.isLoggingEnabled(connectionLogLevel));
+ numParallelScans = MetricUtil.getCombinableMetric(connectionLogLevel, NUM_PARALLEL_SCANS);
+ wallClockTimeMS = MetricUtil.getCombinableMetric(connectionLogLevel, WALL_CLOCK_TIME_MS);
+ resultSetTimeMS = MetricUtil.getCombinableMetric(connectionLogLevel, RESULT_SET_TIME_MS);
+ queryTimedOut = MetricUtil.getCombinableMetric(connectionLogLevel, QUERY_TIMEOUT_COUNTER);
+ queryFailed = MetricUtil.getCombinableMetric(connectionLogLevel, QUERY_FAILED_COUNTER);
+ cacheRefreshedDueToSplits = MetricUtil.getCombinableMetric(connectionLogLevel, CACHE_REFRESH_SPLITS_COUNTER);
}
public void updateNumParallelScans(long numParallelScans) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
index c008635..fa81e9c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
@@ -25,6 +25,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import javax.annotation.Nonnull;
+import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.monitoring.CombinableMetric.NoOpRequestMetric;
import com.google.common.annotations.VisibleForTesting;
@@ -40,19 +41,23 @@ public class ReadMetricQueue {
private final List<ScanMetricsHolder> scanMetricsHolderList = new ArrayList<ScanMetricsHolder>();
- private final boolean isRequestMetricsEnabled;
- public ReadMetricQueue(boolean isRequestMetricsEnabled) {
- this.isRequestMetricsEnabled = isRequestMetricsEnabled;
+ private LogLevel connectionLogLevel;
+
+ public ReadMetricQueue(LogLevel connectionLogLevel) {
+ this.connectionLogLevel = connectionLogLevel;
}
public CombinableMetric allotMetric(MetricType type, String tableName) {
- if (!isRequestMetricsEnabled) { return NoOpRequestMetric.INSTANCE; }
- MetricKey key = new MetricKey(type, tableName);
- Queue<CombinableMetric> q = getMetricQueue(key);
- CombinableMetric metric = getMetric(type);
- q.offer(metric);
- return metric;
+ if (type.isLoggingEnabled(connectionLogLevel)) {
+ MetricKey key = new MetricKey(type, tableName);
+ Queue<CombinableMetric> q = getMetricQueue(key);
+ CombinableMetric metric = getMetric(type);
+ q.offer(metric);
+ return metric;
+ } else {
+ return NoOpRequestMetric.INSTANCE;
+ }
}
@VisibleForTesting
@@ -173,10 +178,6 @@ public class ReadMetricQueue {
return q;
}
- public boolean isRequestMetricsEnabled() {
- return isRequestMetricsEnabled;
- }
-
public void addScanHolder(ScanMetricsHolder holder){
scanMetricsHolderList.add(holder);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
index 9125cd8..494b3e6 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
@@ -34,6 +34,7 @@ import java.util.Map;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.JsonMapper;
+import org.apache.phoenix.log.LogLevel;
public class ScanMetricsHolder {
@@ -52,13 +53,11 @@ public class ScanMetricsHolder {
private Object scan;
private static final ScanMetricsHolder NO_OP_INSTANCE =
- new ScanMetricsHolder(new ReadMetricQueue(false), "",null);
+ new ScanMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "",null);
public static ScanMetricsHolder getInstance(ReadMetricQueue readMetrics, String tableName,
- Scan scan, boolean isRequestMetricsEnabled) {
- if (!isRequestMetricsEnabled) {
- return NO_OP_INSTANCE;
- }
+ Scan scan, LogLevel connectionLogLevel) {
+ if (connectionLogLevel == LogLevel.OFF) { return NO_OP_INSTANCE; }
scan.setScanMetricsEnabled(true);
return new ScanMetricsHolder(readMetrics, tableName, scan);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
index 4373887..699982f 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.monitoring;
+import org.apache.phoenix.log.LogLevel;
/**
* Class that encapsulates the various metrics associated with the spooling done by phoenix as part of servicing a
@@ -26,7 +27,7 @@ public class SpoolingMetricsHolder {
private final CombinableMetric spoolFileSizeMetric;
private final CombinableMetric numSpoolFileMetric;
- public static final SpoolingMetricsHolder NO_OP_INSTANCE = new SpoolingMetricsHolder(new ReadMetricQueue(false), "");
+ public static final SpoolingMetricsHolder NO_OP_INSTANCE = new SpoolingMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "");
public SpoolingMetricsHolder(ReadMetricQueue readMetrics, String tableName) {
this.spoolFileSizeMetric = readMetrics.allotMetric(MetricType.SPOOL_FILE_SIZE, tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
index 98ff57c..6117b40 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
@@ -23,6 +23,8 @@ import static org.apache.phoenix.monitoring.MetricType.TASK_EXECUTION_TIME;
import static org.apache.phoenix.monitoring.MetricType.TASK_QUEUE_WAIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.TASK_REJECTED_COUNTER;
+import org.apache.phoenix.log.LogLevel;
+
/**
* Class to encapsulate the various metrics associated with submitting and executing a task to the phoenix client
@@ -35,7 +37,7 @@ public class TaskExecutionMetricsHolder {
private final CombinableMetric taskExecutionTime;
private final CombinableMetric numTasks;
private final CombinableMetric numRejectedTasks;
- public static final TaskExecutionMetricsHolder NO_OP_INSTANCE = new TaskExecutionMetricsHolder(new ReadMetricQueue(false), "");
+ public static final TaskExecutionMetricsHolder NO_OP_INSTANCE = new TaskExecutionMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "");
public TaskExecutionMetricsHolder(ReadMetricQueue readMetrics, String tableName) {
taskQueueWaitTime = readMetrics.allotMetric(TASK_QUEUE_WAIT_TIME, tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
index 1d3feed..8b7b708 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionQueryServicesImpl.java
@@ -2471,7 +2471,12 @@ public class ConnectionQueryServicesImpl extends DelegateQueryServices implement
// Available for testing
protected String getLogTableDDL() {
- return QueryConstants.CREATE_LOG_METADATA;
+ return setSystemLogDDLProperties(QueryConstants.CREATE_LOG_METADATA);
+ }
+
+ private String setSystemLogDDLProperties(String ddl) {
+ return String.format(ddl, props.getInt(LOG_SALT_BUCKETS_ATTRIB, QueryServicesOptions.DEFAULT_LOG_SALT_BUCKETS));
+
}
private String setSystemDDLProperties(String ddl) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
index 3531a87..a7fad8e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/ConnectionlessQueryServicesImpl.java
@@ -164,7 +164,12 @@ public class ConnectionlessQueryServicesImpl extends DelegateQueryServices imple
}
protected String getLogTableDDL() {
- return QueryConstants.CREATE_LOG_METADATA;
+ return setSystemLogDDLProperties(QueryConstants.CREATE_LOG_METADATA);
+ }
+
+ private String setSystemLogDDLProperties(String ddl) {
+ return String.format(ddl, props.getInt(LOG_SALT_BUCKETS_ATTRIB, QueryServicesOptions.DEFAULT_LOG_SALT_BUCKETS));
+
}
private String setSystemDDLProperties(String ddl) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
index 65806ae..52abed0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryConstants.java
@@ -18,112 +18,7 @@
package org.apache.phoenix.query;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.APPEND_ONLY_SCHEMA;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ARG_POSITION;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ARRAY_SIZE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.AUTO_PARTITION_SEQ;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BASE_COLUMN_COUNT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BIND_PARAMETERS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.BUFFER_LENGTH;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CACHE_SIZE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CHAR_OCTET_LENGTH;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLASS_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CLIENT_IP;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_COUNT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_DEF;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_FAMILY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_QUALIFIER;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_QUALIFIER_COUNTER;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.COLUMN_SIZE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CURRENT_VALUE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.CYCLE_FLAG;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TABLE_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DATA_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DECIMAL_DIGITS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DEFAULT_COLUMN_FAMILY_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DEFAULT_VALUE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.DISABLE_WAL;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ENCODING_SCHEME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.EXCEPTION_TRACE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.EXPLAIN_PLAN;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.FUNCTION_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GLOBAL_SCAN_DETAILS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POSTS_ROW_COUNT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POSTS_WIDTH;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.GUIDE_POST_KEY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_ROWS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IMMUTABLE_STORAGE_SCHEME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INCREMENT_BY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_DISABLE_TIMESTAMP;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_STATE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.INDEX_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_ARRAY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_AUTOINCREMENT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_CONSTANT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_NAMESPACE_MAPPED;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_NULLABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_ROW_TIMESTAMP;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.IS_VIEW_REFERENCED;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.JAR_PATH;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.KEY_SEQ;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LAST_STATS_UPDATE_TIME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LIMIT_REACHED_FLAG;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.LINK_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MAX_VALUE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MIN_VALUE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.MULTI_TENANT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NO_OF_RESULTS_ITERATED;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NULLABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NUM_ARGS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.NUM_PREC_RADIX;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.ORDINAL_POSITION;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PHYSICAL_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.PK_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.QUERY_STATUS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REF_GENERATION;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.REMARKS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.RETURN_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SALT_BUCKETS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCAN_METRICS_JSON;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCOPE_CATALOG;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCOPE_SCHEMA;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SCOPE_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SELF_REFERENCING_COL_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SEQUENCE_SCHEMA;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SORT_ORDER;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SOURCE_DATA_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SQL_DATA_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SQL_DATETIME_SUB;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.START_TIME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.START_WITH;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.STORE_NULLS;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_STATS_TABLE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SCHEM;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_SEQ_NUM;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TABLE_TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TENANT_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TOTAL_EXECUTION_TIME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTIONAL;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TRANSACTION_PROVIDER;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_NAME;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.TYPE_SEQUENCE;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.UPDATE_CACHE_FREQUENCY;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USER;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.USE_STATS_FOR_PARALLELIZATION;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_CONSTANT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_INDEX_ID;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_STATEMENT;
-import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.VIEW_TYPE;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.*;
import java.math.BigDecimal;
@@ -134,9 +29,11 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
+import org.apache.phoenix.monitoring.MetricType;
import org.apache.phoenix.schema.MetaDataSplitPolicy;
import org.apache.phoenix.schema.PName;
import org.apache.phoenix.schema.PNameFactory;
+import org.apache.phoenix.schema.PTable.ImmutableStorageScheme;
import org.apache.phoenix.schema.PTable.QualifierEncodingScheme;
import org.apache.phoenix.schema.SortOrder;
import org.apache.phoenix.schema.SystemFunctionSplitPolicy;
@@ -412,28 +309,31 @@ public interface QueryConstants {
PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE;
public static final String CREATE_LOG_METADATA =
- "CREATE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"(\n" +
+ "CREATE IMMUTABLE TABLE " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_LOG_TABLE + "\"(\n" +
// Pk columns
+ START_TIME + " DECIMAL, \n" +
+ TABLE_NAME + " VARCHAR, \n" +
+ QUERY_ID + " VARCHAR NOT NULL,\n" +
TENANT_ID + " VARCHAR ," +
- QUERY_ID + " VARCHAR NOT NULL,\n" +
USER + " VARCHAR , \n" +
CLIENT_IP + " VARCHAR, \n" +
// Function metadata (will be null for argument row)
QUERY + " VARCHAR, \n" +
EXPLAIN_PLAN + " VARCHAR, \n" +
// Argument metadata (will be null for function row)
- START_TIME + " TIMESTAMP, \n" +
- TOTAL_EXECUTION_TIME + " BIGINT, \n" +
NO_OF_RESULTS_ITERATED + " BIGINT, \n" +
QUERY_STATUS + " VARCHAR, \n" +
EXCEPTION_TRACE + " VARCHAR, \n" +
GLOBAL_SCAN_DETAILS + " VARCHAR, \n" +
BIND_PARAMETERS + " VARCHAR, \n" +
SCAN_METRICS_JSON + " VARCHAR, \n" +
- " CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (QUERY_ID))\n" +
+ MetricType.getMetricColumnsDetails()+"\n"+
+ " CONSTRAINT " + SYSTEM_TABLE_PK_NAME + " PRIMARY KEY (START_TIME, TABLE_NAME, QUERY_ID))\n" +
+ PhoenixDatabaseMetaData.SALT_BUCKETS + "=%s,\n"+
PhoenixDatabaseMetaData.TRANSACTIONAL + "=" + Boolean.FALSE+ ",\n" +
HColumnDescriptor.TTL + "=" + MetaDataProtocol.DEFAULT_LOG_TTL+",\n"+
- PhoenixDatabaseMetaData.COLUMN_ENCODED_BYTES +" = 0";
+ TableProperty.IMMUTABLE_STORAGE_SCHEME.toString() + " = " + ImmutableStorageScheme.SINGLE_CELL_ARRAY_WITH_OFFSETS.name() + ",\n" +
+ TableProperty.COLUMN_ENCODED_BYTES.toString()+" = 1";
public static final byte[] OFFSET_FAMILY = "f_offset".getBytes();
public static final byte[] OFFSET_COLUMN = "c_offset".getBytes();
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
index db0b10b..559d165 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -197,6 +197,7 @@ public interface QueryServices extends SQLCloseable {
public static final String COMMIT_STATS_ASYNC = "phoenix.stats.commit.async";
// Maximum size in bytes taken up by cached table stats in the client
public static final String STATS_MAX_CACHE_SIZE = "phoenix.stats.cache.maxSize";
+ public static final String LOG_SALT_BUCKETS_ATTRIB = "phoenix.log.saltBuckets";
public static final String SEQUENCE_SALT_BUCKETS_ATTRIB = "phoenix.sequence.saltBuckets";
public static final String COPROCESSOR_PRIORITY_ATTRIB = "phoenix.coprocessor.priority";
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 0e6e89f..d708785 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -356,6 +356,7 @@ public class QueryServicesOptions {
public static final boolean DEFAULT_COST_BASED_OPTIMIZER_ENABLED = false;
public static final String DEFAULT_LOGGING_LEVEL = LogLevel.OFF.name();
public static final String DEFAULT_LOG_SAMPLE_RATE = "1.0";
+ public static final int DEFAULT_LOG_SALT_BUCKETS = 32;
private final Configuration config;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 20ac732..8f14680 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -2286,7 +2286,7 @@ public class MetaDataClient {
}
}
// System tables have hard-coded column qualifiers. So we can't use column encoding for them.
- else if (!SchemaUtil.isSystemTable(Bytes.toBytes(SchemaUtil.getTableName(schemaName, tableName)))) {
+ else if (!SchemaUtil.isSystemTable(Bytes.toBytes(SchemaUtil.getTableName(schemaName, tableName)))|| SchemaUtil.isLogTable(schemaName, tableName)) {
/*
* Indexes inherit the storage scheme of the parent data tables. Otherwise, we always attempt to
* create tables with encoded column names.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
index 9d2e53c..94cbfea 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/QueryUtil.java
@@ -436,7 +436,7 @@ public final class QueryUtil {
}
return url;
}
-
+
private static int getInt(String key, int defaultValue, Properties props, Configuration conf) {
if (conf == null) {
Preconditions.checkNotNull(props);
@@ -487,5 +487,10 @@ public final class QueryUtil {
public static String getViewPartitionClause(String partitionColumnName, long autoPartitionNum) {
return partitionColumnName + " " + toSQL(CompareOp.EQUAL) + " " + autoPartitionNum;
}
+
+ public static Connection getConnectionForQueryLog(Configuration config) throws ClassNotFoundException, SQLException {
+ //we don't need this connection to upgrade anything or start dispatcher
+ return getConnectionOnServer(config);
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
index 92a2cde..dd00a69 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/SchemaUtil.java
@@ -1170,6 +1170,10 @@ public class SchemaUtil {
Cell isNamespaceMappedCell = currentResult.getColumnLatestCell(TABLE_FAMILY_BYTES, IS_NAMESPACE_MAPPED_BYTES);
return isNamespaceMappedCell!=null && (boolean) PBoolean.INSTANCE.toObject(isNamespaceMappedCell.getValue());
}
+
+ public static boolean isLogTable(String schemaName, String tableName) {
+ return PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA.equals(schemaName) && PhoenixDatabaseMetaData.SYSTEM_LOG_TABLE.equals(tableName);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
index e0a731d..d858e72 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
@@ -24,10 +24,12 @@ import java.util.Arrays;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.memory.DelegatingMemoryManager;
import org.apache.phoenix.memory.GlobalMemoryManager;
import org.apache.phoenix.memory.MemoryManager;
import org.apache.phoenix.monitoring.MemoryMetricsHolder;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.monitoring.SpoolingMetricsHolder;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.tuple.SingleKeyValueTuple;
@@ -54,7 +56,10 @@ public class SpoolingResultIteratorTest {
};
MemoryManager memoryManager = new DelegatingMemoryManager(new GlobalMemoryManager(threshold));
- ResultIterator scanner = new SpoolingResultIterator(SpoolingMetricsHolder.NO_OP_INSTANCE, MemoryMetricsHolder.NO_OP_INSTANCE, iterator, memoryManager, threshold, maxSizeSpool,"/tmp");
+ ResultIterator scanner = new SpoolingResultIterator(
+ SpoolingMetricsHolder.NO_OP_INSTANCE,
+ new MemoryMetricsHolder(new ReadMetricQueue(LogLevel.OFF), ""), iterator, memoryManager, threshold,
+ maxSizeSpool, "/tmp");
AssertResults.assertResults(scanner, expectedResults);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/f73f98c7/phoenix-hive/src/main/java/org/apache/phoenix/hive/mapreduce/PhoenixRecordReader.java
----------------------------------------------------------------------
diff --git a/phoenix-hive/src/main/java/org/apache/phoenix/hive/mapreduce/PhoenixRecordReader.java b/phoenix-hive/src/main/java/org/apache/phoenix/hive/mapreduce/PhoenixRecordReader.java
index c10c4d1..2ff8aca 100644
--- a/phoenix-hive/src/main/java/org/apache/phoenix/hive/mapreduce/PhoenixRecordReader.java
+++ b/phoenix-hive/src/main/java/org/apache/phoenix/hive/mapreduce/PhoenixRecordReader.java
@@ -112,11 +112,10 @@ public class PhoenixRecordReader<T extends DBWritable> implements
String tableName = queryPlan.getTableRef().getTable().getPhysicalName().getString();
long renewScannerLeaseThreshold = queryPlan.getContext().getConnection()
.getQueryServices().getRenewLeaseThresholdMilliSeconds();
- boolean isRequestMetricsEnabled = readMetrics.isRequestMetricsEnabled();
for (Scan scan : scans) {
scan.setAttribute(BaseScannerRegionObserver.SKIP_REGION_BOUNDARY_CHECK, Bytes
.toBytes(true));
- ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics, tableName, scan, isRequestMetricsEnabled);
+ ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics, tableName, scan, ctx.getConnection().getLogLevel());
final TableResultIterator tableResultIterator = new TableResultIterator(
queryPlan.getContext().getConnection().getMutationState(), scan, scanMetricsHolder,
renewScannerLeaseThreshold, queryPlan, MapReduceParallelScanGrouper.getInstance());
[13/22] phoenix git commit: PHOENIX-4726 save index build timestamp
-- for SYNC case only
Posted by ja...@apache.org.
PHOENIX-4726 save index build timestamp -- for SYNC case only
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/7ebba49d
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/7ebba49d
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/7ebba49d
Branch: refs/heads/omid2
Commit: 7ebba49d000bb221a8de45e70b4c0afb3b9355a3
Parents: 39b92bf
Author: Vincent Poon <vi...@apache.org>
Authored: Mon May 14 10:10:32 2018 -0700
Committer: Vincent Poon <vi...@apache.org>
Committed: Mon May 14 10:11:48 2018 -0700
----------------------------------------------------------------------
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 1 +
.../org/apache/phoenix/schema/MetaDataClient.java | 17 +++++++++++++++++
2 files changed, 18 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7ebba49d/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 55de772..2e12ca2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -316,6 +316,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String ASYNC_CREATED_DATE = "ASYNC_CREATED_DATE";
public static final String SEQUENCE_TABLE_TYPE = SYSTEM_SEQUENCE_TABLE;
+ public static final String SYNC_INDEX_CREATED_DATE = "SYNC_INDEX_CREATED_DATE";
public static final String SYSTEM_MUTEX_TABLE_NAME = "MUTEX";
public static final String SYSTEM_MUTEX_NAME = SchemaUtil.getTableName(QueryConstants.SYSTEM_SCHEMA_NAME, SYSTEM_MUTEX_TABLE_NAME);
public static final TableName SYSTEM_MUTEX_HBASE_TABLE_NAME = TableName.valueOf(SYSTEM_MUTEX_NAME);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7ebba49d/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 009289b..20ac732 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -74,6 +74,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.RETURN_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SALT_BUCKETS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SORT_ORDER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.STORE_NULLS;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYNC_INDEX_CREATED_DATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
@@ -263,6 +264,14 @@ public class MetaDataClient {
TABLE_NAME + "," +
ASYNC_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
") VALUES (?, ?, ?, ?)";
+
+ private static final String SET_INDEX_SYNC_CREATED_DATE =
+ "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
+ TENANT_ID + "," +
+ TABLE_SCHEM + "," +
+ TABLE_NAME + "," +
+ SYNC_INDEX_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
+ ") VALUES (?, ?, ?, ?)";
private static final String CREATE_TABLE =
"UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
TENANT_ID + "," +
@@ -2678,6 +2687,14 @@ public class MetaDataClient {
setAsync.setString(3, tableName);
setAsync.setDate(4, asyncCreatedDate);
setAsync.execute();
+ } else {
+ Date syncCreatedDate = new Date(EnvironmentEdgeManager.currentTimeMillis());
+ PreparedStatement setSync = connection.prepareStatement(SET_INDEX_SYNC_CREATED_DATE);
+ setSync.setString(1, tenantIdStr);
+ setSync.setString(2, schemaName);
+ setSync.setString(3, tableName);
+ setSync.setDate(4, syncCreatedDate);
+ setSync.execute();
}
tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
connection.rollback();
[10/22] phoenix git commit: Fix PhoenixMRJobSubmitter submits
duplicate MR jobs for an index build from indexTool
Posted by ja...@apache.org.
Fix PhoenixMRJobSubmitter submits duplicate MR jobs for an index build from indexTool
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/5dc873dd
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5dc873dd
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5dc873dd
Branch: refs/heads/omid2
Commit: 5dc873dd250d83b66b36d6a883531dac78e1dced
Parents: c27418b
Author: Xu Cang <xc...@salesforce.com>
Authored: Mon May 7 16:21:46 2018 -0700
Committer: Thomas D'Silva <td...@apache.org>
Committed: Thu May 10 15:38:17 2018 -0700
----------------------------------------------------------------------
.../phoenix/mapreduce/index/IndexTool.java | 6 ++--
.../index/automation/PhoenixAsyncIndex.java | 2 +-
.../index/automation/PhoenixMRJobSubmitter.java | 2 +-
.../index/automated/MRJobSubmitterTest.java | 30 +++++++++++++++-----
4 files changed, 28 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5dc873dd/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
index 671e4cf..e3aa729 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/IndexTool.java
@@ -123,7 +123,7 @@ public class IndexTool extends Configured implements Tool {
private static final Option SNAPSHOT_OPTION = new Option("snap", "snapshot", false,
"If specified, uses Snapshots for async index building (optional)");
private static final Option HELP_OPTION = new Option("h", "help", false, "Help");
- public static final String INDEX_JOB_NAME_TEMPLATE = "PHOENIX_%s_INDX_%s";
+ public static final String INDEX_JOB_NAME_TEMPLATE = "PHOENIX_%s.%s_INDX_%s";
private Options getOptions() {
final Options options = new Options();
@@ -373,9 +373,9 @@ public class IndexTool extends Configured implements Tool {
PhoenixRuntime.generateColumnInfo(connection, qIndexTable, indexColumns);
ColumnInfoToStringEncoderDecoder.encode(configuration, columnMetadataList);
fs = outputPath.getFileSystem(configuration);
- fs.delete(outputPath, true);
+ fs.delete(outputPath, true);
- final String jobName = String.format(INDEX_JOB_NAME_TEMPLATE, pdataTable.getName().toString(), indexTable);
+ final String jobName = String.format(INDEX_JOB_NAME_TEMPLATE, schemaName, dataTable, indexTable);
final Job job = Job.getInstance(configuration, jobName);
job.setJarByClass(IndexTool.class);
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5dc873dd/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixAsyncIndex.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixAsyncIndex.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixAsyncIndex.java
index 3e88cd0..a61e49a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixAsyncIndex.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixAsyncIndex.java
@@ -59,7 +59,7 @@ public class PhoenixAsyncIndex {
}
public String getJobName() {
- return String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE, dataTableName, tableName);
+ return String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE, tableSchem, dataTableName, tableName);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5dc873dd/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
index d86802a..3e20bd2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
@@ -216,7 +216,7 @@ public class PhoenixMRJobSubmitter {
indexInfo.setTableSchem(rs.getString(PhoenixDatabaseMetaData.TABLE_SCHEM));
indexInfo.setTableName(rs.getString(PhoenixDatabaseMetaData.TABLE_NAME));
candidateIndexes.put(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- indexInfo.getDataTableName(), indexInfo.getTableName()), indexInfo);
+ indexInfo.getTableSchem(), indexInfo.getDataTableName(), indexInfo.getTableName()), indexInfo);
}
return candidateIndexes;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/5dc873dd/phoenix-core/src/test/java/org/apache/phoenix/index/automated/MRJobSubmitterTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/index/automated/MRJobSubmitterTest.java b/phoenix-core/src/test/java/org/apache/phoenix/index/automated/MRJobSubmitterTest.java
index 5ed1837..3a4de4c 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/index/automated/MRJobSubmitterTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/index/automated/MRJobSubmitterTest.java
@@ -43,18 +43,20 @@ public class MRJobSubmitterTest {
PhoenixAsyncIndex index1 = new PhoenixAsyncIndex();
index1.setDataTableName("DT1");
index1.setTableName("IT1");
+ index1.setTableSchem("NEW_SCHEM1");
index1.setIndexType(IndexType.LOCAL);
candidateJobs.put(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- index1.getDataTableName(), index1.getTableName()), index1);
+ index1.getTableSchem(), index1.getDataTableName(), index1.getTableName()), index1);
PhoenixAsyncIndex index2 = new PhoenixAsyncIndex();
index2.setDataTableName("DT2");
index2.setTableName("IT2");
+ index2.setTableSchem("NEW_SCHEM2");
index2.setIndexType(IndexType.LOCAL);
candidateJobs.put(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- index2.getDataTableName(), index2.getTableName()), index2);
+ index2.getTableSchem(), index2.getDataTableName(), index2.getTableName()), index2);
}
@Test
@@ -71,6 +73,20 @@ public class MRJobSubmitterTest {
}
@Test
+ public void testIndexJobsName() throws IOException {
+ // Verify index job name contains schem name, not only table name.
+ PhoenixAsyncIndex index = new PhoenixAsyncIndex();
+ index.setDataTableName("MyDataTable");
+ index.setTableName("MyTableName");
+ index.setTableSchem("MySchem");
+ index.setIndexType(IndexType.LOCAL);
+
+ String jobName = String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
+ index.getTableSchem(), index.getDataTableName(), index.getTableName());
+ assertEquals("PHOENIX_MySchem.MyDataTable_INDX_MyTableName", jobName);
+ }
+
+ @Test
public void testGlobalIndexJobsForSubmission() throws IOException {
// Set the index type to GLOBAL
@@ -91,7 +107,7 @@ public class MRJobSubmitterTest {
// Mark one job as running
submittedJobs.add(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- jobs[0].getDataTableName(), jobs[0].getTableName()));
+ jobs[0].getTableSchem(), jobs[0].getDataTableName(), jobs[0].getTableName()));
PhoenixMRJobSubmitter submitter = new PhoenixMRJobSubmitter();
Set<PhoenixAsyncIndex> jobsToSubmit =
@@ -110,9 +126,9 @@ public class MRJobSubmitterTest {
// Mark all the candidate jobs as running/in-progress
submittedJobs.add(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- jobs[0].getDataTableName(), jobs[0].getTableName()));
+ jobs[0].getTableSchem(), jobs[0].getDataTableName(), jobs[0].getTableName()));
submittedJobs.add(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- jobs[1].getDataTableName(), jobs[1].getTableName()));
+ jobs[1].getTableSchem(), jobs[1].getDataTableName(), jobs[1].getTableName()));
PhoenixMRJobSubmitter submitter = new PhoenixMRJobSubmitter();
Set<PhoenixAsyncIndex> jobsToSubmit =
@@ -126,9 +142,9 @@ public class MRJobSubmitterTest {
candidateJobs.clear();
// Add some dummy running jobs to the submitted list
submittedJobs.add(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- "d1", "i1"));
+ "s1", "d1", "i1"));
submittedJobs.add(String.format(IndexTool.INDEX_JOB_NAME_TEMPLATE,
- "d2", "i2"));
+ "s2", "d2", "i2"));
PhoenixMRJobSubmitter submitter = new PhoenixMRJobSubmitter();
Set<PhoenixAsyncIndex> jobsToSubmit =
submitter.getJobsToSubmit(candidateJobs, submittedJobs);
[11/22] phoenix git commit: PHOENIX-4734 SQL Query with an RVC
expression lexographically higher than all values in an OR clause causes
query to blow up
Posted by ja...@apache.org.
PHOENIX-4734 SQL Query with an RVC expression lexographically higher than all values in an OR clause causes query to blow up
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/72fa8749
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/72fa8749
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/72fa8749
Branch: refs/heads/omid2
Commit: 72fa8749eb443b6fb9cb6436b06891642eff0440
Parents: 5dc873d
Author: Thomas D'Silva <td...@apache.org>
Authored: Thu May 10 16:30:36 2018 -0700
Committer: Thomas D'Silva <td...@apache.org>
Committed: Fri May 11 09:33:35 2018 -0700
----------------------------------------------------------------------
.../phoenix/end2end/RowValueConstructorIT.java | 120 +++++++++++++++++++
.../org/apache/phoenix/compile/ScanRanges.java | 5 +
2 files changed, 125 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/72fa8749/phoenix-core/src/it/java/org/apache/phoenix/end2end/RowValueConstructorIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/RowValueConstructorIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/RowValueConstructorIT.java
index df7603a..fb04261 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/RowValueConstructorIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/RowValueConstructorIT.java
@@ -48,6 +48,7 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
+import java.util.List;
import java.util.Properties;
import org.apache.phoenix.util.DateUtil;
@@ -57,6 +58,9 @@ import org.apache.phoenix.util.QueryUtil;
import org.apache.phoenix.util.SchemaUtil;
import org.junit.Test;
+import com.google.common.base.Joiner;
+import com.google.common.collect.Lists;
+
public class RowValueConstructorIT extends ParallelStatsDisabledIT {
@@ -1646,4 +1650,120 @@ public class RowValueConstructorIT extends ParallelStatsDisabledIT {
assertEquals("value", rs.getString(3));
assertFalse(rs.next());
}
+
+ @Test
+ /**
+ * Verifies that a query with a RVC expression lexographically higher than all values in an OR
+ * clause causes query works see PHOENIX-4734
+ */
+ public void testRVCWithAndClause() throws Exception {
+ final int numItemsInClause = 5;
+ Properties tenantProps = new Properties();
+ tenantProps.setProperty(PhoenixRuntime.TENANT_ID_ATTRIB, "tenant1");
+ String fullTableName = SchemaUtil.getTableName("S", "T_" + generateUniqueName());
+ String fullViewName = SchemaUtil.getTableName("S", "V_" + generateUniqueName());
+ try (Connection tenantConn = DriverManager.getConnection(getUrl(), tenantProps)) {
+ tenantConn.setAutoCommit(false);
+ createBaseTableAndTenantView(tenantConn, fullTableName, fullViewName);
+ loadDataIntoTenantView(tenantConn, fullViewName);
+ List<String> objectIdsList =
+ selectObjectIdsForInClause(tenantConn, fullViewName, numItemsInClause);
+ StringBuilder querySb = generateQueryToTest(numItemsInClause, fullViewName);
+ PreparedStatement ps = tenantConn.prepareStatement(querySb.toString());
+ int numbBindVarsSet = 0;
+ String objectId = null;
+ for (int i = 0; i < numItemsInClause; i++) {
+ objectId = objectIdsList.get(i);
+ ps.setString((i + 1), objectId);
+ numbBindVarsSet++;
+ }
+ assertEquals(numItemsInClause, numbBindVarsSet);
+ assertEquals("v1000", objectId);
+ ps.setString(numItemsInClause + 1, "z00");
+ ps.setString(numItemsInClause + 2, "v1000"); // This value must match or be
+ // lexographically higher than the highest
+ // value in the IN clause
+ // Query should execute and return 0 results
+ ResultSet rs = ps.executeQuery();
+ assertFalse(
+ "Query should return no results as IN clause and RVC clause are disjoint sets",
+ rs.next());
+ }
+ }
+
+ private StringBuilder generateQueryToTest(int numItemsInClause, String fullViewName) {
+ StringBuilder querySb =
+ new StringBuilder("SELECT OBJECT_ID,OBJECT_DATA2,OBJECT_DATA FROM " + fullViewName);
+ querySb.append(" WHERE ((");
+ List<String> orClauses = Lists.newArrayList();
+ for (int i = 1; i < (numItemsInClause + 1); i++) {
+ orClauses.add("OBJECT_ID = ?");
+ }
+ querySb.append(Joiner.on(" OR ").join(orClauses));
+ querySb.append(") AND (KEY_PREFIX,OBJECT_ID) > (?,?)) ORDER BY OBJECT_ID LIMIT 125");
+ System.out.println(querySb);
+ return querySb;
+ }
+
+ private List<String> selectObjectIdsForInClause(Connection tenantConn, String fullViewName,
+ int numItemsInClause) throws SQLException {
+ String sqlForObjIds =
+ "SELECT OBJECT_ID FROM " + fullViewName + " ORDER BY OBJECT_ID LIMIT "
+ + numItemsInClause;
+ PreparedStatement ps = tenantConn.prepareStatement(sqlForObjIds);
+ ResultSet rs = ps.executeQuery();
+ List<String> objectIdsList = Lists.newArrayList();
+ System.out.println("ObjectIds: ");
+ while (rs.next()) {
+ System.out.println("Object Id: " + rs.getString("OBJECT_ID"));
+ objectIdsList.add(rs.getString("OBJECT_ID"));
+ }
+ assertEquals(numItemsInClause, objectIdsList.size());
+ return objectIdsList;
+ }
+
+ private void loadDataIntoTenantView(Connection tenantConn, String fullViewName)
+ throws SQLException {
+ for (int i = 0; i < 2000; i++) {
+ String objectId = "v" + i;
+ String upsert =
+ "UPSERT INTO " + fullViewName
+ + " (OBJECT_ID, OBJECT_DATA, OBJECT_DATA2) VALUES ('" + objectId
+ + "', 'data','data2')";
+ PreparedStatement ps = tenantConn.prepareStatement(upsert);
+ ps.executeUpdate();
+ }
+ tenantConn.commit();
+
+ // Validate Data was loaded correctly
+ PreparedStatement selectStatement =
+ tenantConn.prepareStatement(
+ "SELECT OBJECT_ID FROM " + fullViewName + " ORDER BY OBJECT_ID");
+ ResultSet rs = selectStatement.executeQuery();
+ int count = 0;
+ while (rs.next()) {
+ count++;
+ }
+ assertEquals(2000, count);
+ }
+
+ private void createBaseTableAndTenantView(Connection tenantConn, String fulTableName,
+ String fullViewName) throws SQLException {
+ String ddl =
+ "CREATE TABLE IF NOT EXISTS " + fulTableName + " (TENANT_ID CHAR(15) NOT NULL,"
+ + " KEY_PREFIX CHAR(3) NOT NULL," + " CREATED_DATE DATE,"
+ + " CREATED_BY CHAR(15)," + " SYSTEM_MODSTAMP DATE"
+ + " CONSTRAINT PK PRIMARY KEY (TENANT_ID,KEY_PREFIX)"
+ + ") VERSIONS=1, MULTI_TENANT=true, IMMUTABLE_ROWS=TRUE, REPLICATION_SCOPE=1";
+ createTestTable(getUrl(), ddl);
+
+ String tenantViewDdl =
+ "CREATE VIEW IF NOT EXISTS " + fullViewName + " (OBJECT_ID VARCHAR(18) NOT NULL, "
+ + "OBJECT_DATA VARCHAR(131072), " + "OBJECT_DATA2 VARCHAR(131072), "
+ + "CONSTRAINT PK PRIMARY KEY (OBJECT_ID)) " + "AS SELECT * FROM "
+ + fulTableName + " WHERE KEY_PREFIX = 'z00'";
+ // Get tenant specific connection
+ PreparedStatement stmt2 = tenantConn.prepareStatement(tenantViewDdl);
+ stmt2.execute();
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/72fa8749/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java
index 8c71248..019f15d 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/ScanRanges.java
@@ -105,6 +105,11 @@ public class ScanRanges {
keyRanges.add(KeyRange.getKeyRange(key));
}
}
+ // while doing a point look up if after intersecting with the MinMaxrange there are
+ // no more keyranges left then just return
+ if (keyRanges.isEmpty()) {
+ return NOTHING;
+ }
ranges = Collections.singletonList(keyRanges);
useSkipScan = keyRanges.size() > 1;
// Treat as binary if descending because we've got a separator byte at the end
[02/22] phoenix git commit: PHOENIX-4685 Properly handle connection
caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Posted by ja...@apache.org.
PHOENIX-4685 Properly handle connection caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d38822ed
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d38822ed
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d38822ed
Branch: refs/heads/omid2
Commit: d38822ed7cc6a39578a2423556a036c5d48d8540
Parents: 4082c73
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Tue May 8 20:50:25 2018 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Tue May 8 20:50:25 2018 +0530
----------------------------------------------------------------------
phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d38822ed/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index 2dab076..d80b6fc 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -27,7 +27,6 @@ import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
-import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
[04/22] phoenix git commit: PHOENIX-4732 Improve config defaults for
async secondary index MR jobs
Posted by ja...@apache.org.
PHOENIX-4732 Improve config defaults for async secondary index MR jobs
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/4849fbe2
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/4849fbe2
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/4849fbe2
Branch: refs/heads/omid2
Commit: 4849fbe2b7160d9cf0e1f05b93ae39b186d3f144
Parents: 7e5c387
Author: James Taylor <jt...@salesforce.com>
Authored: Tue May 8 15:26:44 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed May 9 13:35:52 2018 -0700
----------------------------------------------------------------------
.../index/automation/PhoenixMRJobSubmitter.java | 48 ++++++++++++++++++--
.../apache/phoenix/util/PhoenixMRJobUtil.java | 18 +++-----
2 files changed, 52 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4849fbe2/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
index 9c447e8..d86802a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/mapreduce/index/automation/PhoenixMRJobSubmitter.java
@@ -45,6 +45,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
+import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
import org.apache.phoenix.mapreduce.index.IndexTool;
@@ -53,8 +54,8 @@ import org.apache.phoenix.schema.PTable.IndexType;
import org.apache.phoenix.schema.PTableType;
import org.apache.phoenix.schema.types.PDate;
import org.apache.phoenix.util.PhoenixMRJobUtil;
-import org.apache.phoenix.util.UpgradeUtil;
import org.apache.phoenix.util.PhoenixMRJobUtil.MR_SCHEDULER_TYPE;
+import org.apache.phoenix.util.UpgradeUtil;
import org.apache.phoenix.util.ZKBasedMasterElectionUtil;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONObject;
@@ -69,8 +70,28 @@ public class PhoenixMRJobSubmitter {
// Lock to elect a master node that submits the Phoenix Secondary Index MR Jobs
private static final String PHOENIX_LOCKS_PARENT =
"/phoenix/automated-mr-index-build-leader-election";
+ public static final String PHOENIX_MR_CLIENT_SCANNER_TIMEOUT_PERIOD =
+ "phoenix.mr.client.scanner.timeout.period";
+ public static final String PHOENIX_MR_RPC_TIMEOUT =
+ "phoenix.mr.rpc.timeout";
+ public static final String PHOENIX_MR_TASK_TIMEOUT =
+ "phoenix.mr.task.timeout";
+ public static final String PHOENIX_MR_CLIENT_RETRIES_NUMBER =
+ "phoenix.mr.client.retries.number";
+ public static final String PHOENIX_MR_CLIENT_PAUSE =
+ "phoenix.mr.client.retries.number";
+ public static final String PHOENIX_MR_ZK_RECOVERY_RETRY =
+ "phoenix.mr.zk.recovery.retry";
private static final String AUTO_INDEX_BUILD_LOCK_NAME = "ActiveStandbyElectorLock";
-
+ private static final int DEFAULT_TIMEOUT_IN_MILLIS = 600000;
+ public static final int DEFAULT_MR_CLIENT_SCANNER_TIMEOUT_PERIOD = DEFAULT_TIMEOUT_IN_MILLIS;
+ public static final int DEFAULT_MR_RPC_TIMEOUT = DEFAULT_TIMEOUT_IN_MILLIS;
+ public static final int DEFAULT_MR_TASK_TIMEOUT = DEFAULT_TIMEOUT_IN_MILLIS;
+ // Reduced HBase/Zookeeper Client Retries
+ public static final int DEFAULT_MR_CLIENT_RETRIES_NUMBER = 10;
+ public static final int DEFAULT_MR_CLIENT_PAUSE = 1000;
+ public static final int DEFAULT_MR_ZK_RECOVERY_RETRY = 1;
+
public static final String CANDIDATE_INDEX_INFO_QUERY = "SELECT "
+ PhoenixDatabaseMetaData.INDEX_TYPE + ","
+ PhoenixDatabaseMetaData.DATA_TABLE_NAME + ", "
@@ -103,7 +124,28 @@ public class PhoenixMRJobSubmitter {
}
this.conf = conf;
- PhoenixMRJobUtil.updateTimeoutsToFailFast(conf);
+ // Have Phoenix specific properties for defaults to enable potential override
+ conf.setLong(HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,
+ conf.getLong(PHOENIX_MR_CLIENT_SCANNER_TIMEOUT_PERIOD,
+ DEFAULT_MR_CLIENT_SCANNER_TIMEOUT_PERIOD));
+ conf.setLong(HConstants.HBASE_RPC_TIMEOUT_KEY,
+ conf.getLong(PHOENIX_MR_RPC_TIMEOUT,
+ DEFAULT_MR_RPC_TIMEOUT));
+ conf.setLong(MRJobConfig.TASK_TIMEOUT,
+ conf.getLong(PHOENIX_MR_TASK_TIMEOUT,
+ DEFAULT_MR_TASK_TIMEOUT));
+
+ // Reduced HBase Client Retries
+ conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
+ conf.getInt(PHOENIX_MR_CLIENT_RETRIES_NUMBER,
+ DEFAULT_MR_CLIENT_RETRIES_NUMBER));
+ conf.setInt(HConstants.HBASE_CLIENT_PAUSE,
+ conf.getInt(PHOENIX_MR_CLIENT_PAUSE,
+ DEFAULT_MR_CLIENT_PAUSE));
+ conf.setInt("zookeeper.recovery.retry",
+ conf.getInt(PHOENIX_MR_ZK_RECOVERY_RETRY,
+ DEFAULT_MR_ZK_RECOVERY_RETRY));
+
String schedulerType =
conf.get(PhoenixMRJobUtil.PHOENIX_MR_SCHEDULER_TYPE_NAME,
MR_SCHEDULER_TYPE.NONE.toString());
http://git-wip-us.apache.org/repos/asf/phoenix/blob/4849fbe2/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixMRJobUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixMRJobUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixMRJobUtil.java
index f12d49d..13eae98 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixMRJobUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/PhoenixMRJobUtil.java
@@ -52,19 +52,17 @@ public class PhoenixMRJobUtil {
private static final String ACTIVE_STANDBY_ELECTOR_LOCK = "ActiveStandbyElectorLock";
private static final String RM_APPS_GET_ENDPOINT = "/ws/v1/cluster/apps";
- // Reduced HBase Client Retries
- private static final int CLIENT_RETRIES_NUMBER = 2;
- private static final long CLIENT_PAUSE_TIME = 1000;
- private static final int ZOOKEEPER_RECOVERY_RETRY_COUNT = 1;
-
public static final String PHOENIX_INDEX_MR_QUEUE_NAME_PROPERTY =
"phoenix.index.mr.scheduler.capacity.queuename";
public static final String PHOENIX_INDEX_MR_MAP_MEMORY_PROPERTY =
"phoenix.index.mr.scheduler.capacity.mapMemoryMB";
+ public static final String PHOENIX_MR_CONCURRENT_MAP_LIMIT_PROPERTY =
+ "phoenix.mr.concurrent.map.limit";
// Default MR Capacity Scheduler Configurations for Phoenix MR Index Build
// Jobs
public static final String DEFAULT_QUEUE_NAME = "default";
+ public static final int DEFAULT_MR_CONCURRENT_MAP_LIMIT = 20;
public static final int DEFAULT_MAP_MEMROY_MB = 5120;
public static final String XMX_OPT = "-Xmx";
@@ -208,12 +206,6 @@ public class PhoenixMRJobUtil {
return rmPort;
}
- public static void updateTimeoutsToFailFast(Configuration conf) {
- conf.set("hbase.client.retries.number", String.valueOf(CLIENT_RETRIES_NUMBER));
- conf.set("zookeeper.recovery.retry", String.valueOf(ZOOKEEPER_RECOVERY_RETRY_COUNT));
- conf.set("hbase.client.pause", String.valueOf(CLIENT_PAUSE_TIME));
- }
-
/**
* This method set the configuration values for Capacity scheduler.
* @param conf - Configuration to which Capacity Queue information to be added
@@ -221,6 +213,10 @@ public class PhoenixMRJobUtil {
public static void updateCapacityQueueInfo(Configuration conf) {
conf.set(MRJobConfig.QUEUE_NAME,
conf.get(PHOENIX_INDEX_MR_QUEUE_NAME_PROPERTY, DEFAULT_QUEUE_NAME));
+
+ conf.setInt(MRJobConfig.JOB_RUNNING_MAP_LIMIT,
+ conf.getInt(PHOENIX_MR_CONCURRENT_MAP_LIMIT_PROPERTY, DEFAULT_MR_CONCURRENT_MAP_LIMIT));
+
int mapMemoryMB = conf.getInt(PHOENIX_INDEX_MR_MAP_MEMORY_PROPERTY, DEFAULT_MAP_MEMROY_MB);
conf.setInt(MRJobConfig.MAP_MEMORY_MB, mapMemoryMB);
[03/22] phoenix git commit: PHOENIX-4685 Properly handle connection
caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Posted by ja...@apache.org.
PHOENIX-4685 Properly handle connection caching for Phoenix inside RegionServers-addendum(Rajeshbabu)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/7e5c3871
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/7e5c3871
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/7e5c3871
Branch: refs/heads/omid2
Commit: 7e5c3871dc55e00d800c8a2cf17f44d45cb8a194
Parents: d38822e
Author: Rajeshbabu Chintaguntla <ra...@apache.org>
Authored: Wed May 9 18:13:13 2018 +0530
Committer: Rajeshbabu Chintaguntla <ra...@apache.org>
Committed: Wed May 9 18:13:13 2018 +0530
----------------------------------------------------------------------
.../src/main/java/org/apache/phoenix/util/ServerUtil.java | 3 +--
1 file changed, 1 insertion(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e5c3871/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
index d80b6fc..9190373 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/util/ServerUtil.java
@@ -26,6 +26,7 @@ import java.io.IOException;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -66,8 +67,6 @@ import org.apache.phoenix.hbase.index.write.IndexWriterUtils;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.StaleRegionBoundaryCacheException;
-import org.jboss.netty.util.internal.ConcurrentHashMap;
-
@SuppressWarnings("deprecation")
public class ServerUtil {
[20/22] phoenix git commit: PHOENIX-4701 Write client-side metrics
asynchronously to SYSTEM.LOG(addendum)
Posted by ja...@apache.org.
PHOENIX-4701 Write client-side metrics asynchronously to SYSTEM.LOG(addendum)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d51353a4
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d51353a4
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d51353a4
Branch: refs/heads/omid2
Commit: d51353a4cb45b3f0b448eedc921c6a2ca27208a7
Parents: 5b982f8
Author: Ankit Singhal <an...@gmail.com>
Authored: Wed May 16 17:22:26 2018 -0700
Committer: Ankit Singhal <an...@gmail.com>
Committed: Wed May 16 17:22:26 2018 -0700
----------------------------------------------------------------------
.../phoenix/monitoring/PhoenixMetricsIT.java | 7 +++----
.../apache/phoenix/compile/StatementContext.java | 4 ++--
.../org/apache/phoenix/execute/MutationState.java | 4 ++--
.../phoenix/monitoring/CombinableMetric.java | 9 +++++++++
.../phoenix/monitoring/CombinableMetricImpl.java | 14 +++++++++++---
.../org/apache/phoenix/monitoring/MetricUtil.java | 4 ++--
.../phoenix/monitoring/MutationMetricQueue.java | 18 ++++++------------
.../phoenix/monitoring/OverAllQueryMetrics.java | 14 +++++++-------
.../phoenix/monitoring/ReadMetricQueue.java | 14 ++++++++++----
.../phoenix/monitoring/ScanMetricsHolder.java | 4 ++--
.../phoenix/monitoring/SpoolingMetricsHolder.java | 2 +-
.../monitoring/TaskExecutionMetricsHolder.java | 2 +-
.../iterate/SpoolingResultIteratorTest.java | 2 +-
13 files changed, 57 insertions(+), 41 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
index 73cdf0a..f13391f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java
@@ -101,7 +101,6 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
props.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, String.valueOf(true));
// disable renewing leases as this will force spooling to happen.
props.put(QueryServices.RENEW_LEASE_ENABLED, String.valueOf(false));
- props.put(QueryServices.LOG_LEVEL, LogLevel.DEBUG.toString());
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
// need the non-test driver for some tests that check number of hconnections, etc.
DriverManager.registerDriver(PhoenixDriver.INSTANCE);
@@ -700,7 +699,7 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
private void changeInternalStateForTesting(PhoenixResultSet rs) {
// get and set the internal state for testing purposes.
- ReadMetricQueue testMetricsQueue = new TestReadMetricsQueue(LogLevel.DEBUG);
+ ReadMetricQueue testMetricsQueue = new TestReadMetricsQueue(LogLevel.OFF,true);
StatementContext ctx = (StatementContext)Whitebox.getInternalState(rs, "context");
Whitebox.setInternalState(ctx, "readMetricsQueue", testMetricsQueue);
Whitebox.setInternalState(rs, "readMetricsQueue", testMetricsQueue);
@@ -766,8 +765,8 @@ public class PhoenixMetricsIT extends BaseUniqueNamesOwnClusterIT {
private class TestReadMetricsQueue extends ReadMetricQueue {
- public TestReadMetricsQueue(LogLevel connectionLogLevel) {
- super(connectionLogLevel);
+ public TestReadMetricsQueue(LogLevel connectionLogLevel, boolean isRequestMetricsEnabled) {
+ super(isRequestMetricsEnabled, connectionLogLevel);
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
index 4358ee3..fe60bb9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/compile/StatementContext.java
@@ -135,8 +135,8 @@ public class StatementContext {
this.dataColumns = this.currentTable == null ? Collections.<PColumn, Integer> emptyMap() : Maps
.<PColumn, Integer> newLinkedHashMap();
this.subqueryResults = Maps.<SelectStatement, Object> newHashMap();
- this.readMetricsQueue = new ReadMetricQueue(connection.getLogLevel());
- this.overAllQueryMetrics = new OverAllQueryMetrics(connection.getLogLevel());
+ this.readMetricsQueue = new ReadMetricQueue(isRequestMetricsEnabled,connection.getLogLevel());
+ this.overAllQueryMetrics = new OverAllQueryMetrics(isRequestMetricsEnabled,connection.getLogLevel());
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index f3a383e..52e490e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -38,6 +38,7 @@ import javax.annotation.Nonnull;
import javax.annotation.concurrent.Immutable;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.phoenix.util.KeyValueUtil;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.HTableInterface;
@@ -97,7 +98,6 @@ import org.apache.phoenix.transaction.TransactionFactory;
import org.apache.phoenix.transaction.TransactionFactory.Provider;
import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.IndexUtil;
-import org.apache.phoenix.util.KeyValueUtil;
import org.apache.phoenix.util.LogUtil;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.SQLCloseable;
@@ -1181,7 +1181,7 @@ public class MutationState implements SQLCloseable {
numFailedMutations = uncommittedStatementIndexes.length;
GLOBAL_MUTATION_BATCH_FAILED_COUNT.update(numFailedMutations);
} finally {
- MutationMetric mutationsMetric = new MutationMetric(connection.getLogLevel(),numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
+ MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, mutationCommitTime, numFailedMutations);
mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
try {
if (cache!=null)
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetric.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetric.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetric.java
index db51c56..103fd14 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetric.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetric.java
@@ -28,6 +28,8 @@ public interface CombinableMetric extends Metric {
String getPublishString();
CombinableMetric combine(CombinableMetric metric);
+
+ CombinableMetric clone();
public class NoOpRequestMetric implements CombinableMetric {
@@ -70,7 +72,14 @@ public interface CombinableMetric extends Metric {
@Override
public void decrement() {}
+
+ @Override
+ public CombinableMetric clone(){
+ return INSTANCE;
+ }
}
+
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetricImpl.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetricImpl.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetricImpl.java
index 88049c7..bd22418 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetricImpl.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/CombinableMetricImpl.java
@@ -17,8 +17,6 @@
*/
package org.apache.phoenix.monitoring;
-import static com.google.common.base.Preconditions.checkArgument;
-
public class CombinableMetricImpl implements CombinableMetric {
private final Metric metric;
@@ -26,6 +24,10 @@ public class CombinableMetricImpl implements CombinableMetric {
public CombinableMetricImpl(MetricType type) {
metric = new NonAtomicMetric(type);
}
+
+ private CombinableMetricImpl(Metric metric) {
+ this.metric = metric;
+ }
@Override
public MetricType getMetricType() {
@@ -64,7 +66,6 @@ public class CombinableMetricImpl implements CombinableMetric {
@Override
public CombinableMetric combine(CombinableMetric metric) {
- checkArgument(this.getClass().equals(metric.getClass()));
this.metric.change(metric.getValue());
return this;
}
@@ -73,5 +74,12 @@ public class CombinableMetricImpl implements CombinableMetric {
public void decrement() {
metric.decrement();
}
+
+ @Override
+ public CombinableMetric clone(){
+ NonAtomicMetric metric = new NonAtomicMetric(this.metric.getMetricType());
+ metric.change(this.metric.getValue());
+ return new CombinableMetricImpl(metric);
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
index 1e5ac08..e792c08 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MetricUtil.java
@@ -22,8 +22,8 @@ import org.apache.phoenix.monitoring.CombinableMetric.NoOpRequestMetric;
public class MetricUtil {
- public static CombinableMetric getCombinableMetric(LogLevel connectionLogLevel, MetricType type) {
- if (!type.isLoggingEnabled(connectionLogLevel)) { return NoOpRequestMetric.INSTANCE; }
+ public static CombinableMetric getCombinableMetric(boolean isRequestMetricsEnabled, LogLevel connectionLogLevel, MetricType type) {
+ if (!type.isLoggingEnabled(connectionLogLevel) && !isRequestMetricsEnabled) { return NoOpRequestMetric.INSTANCE; }
return new CombinableMetricImpl(type);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
index 1256f5c..08cf239 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java
@@ -27,7 +27,6 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
-import org.apache.phoenix.log.LogLevel;
/**
* Queue that tracks various writes/mutations related phoenix request metrics.
@@ -83,17 +82,12 @@ public class MutationMetricQueue {
* Class that holds together the various metrics associated with mutations.
*/
public static class MutationMetric {
- private final CombinableMetric numMutations;;
- private final CombinableMetric mutationsSizeBytes;
- private final CombinableMetric totalCommitTimeForMutations;
- private final CombinableMetric numFailedMutations;
-
- public MutationMetric(LogLevel connectionLogLevel, long numMutations, long mutationsSizeBytes, long commitTimeForMutations, long numFailedMutations) {
- this.numMutations = MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BATCH_SIZE);
- this.mutationsSizeBytes =MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BYTES);
- this.totalCommitTimeForMutations =MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_COMMIT_TIME);
- this.numFailedMutations = MetricUtil.getCombinableMetric(connectionLogLevel,MUTATION_BATCH_FAILED_SIZE);
- this.numMutations.change(numMutations);
+ private final CombinableMetric numMutations = new CombinableMetricImpl(MUTATION_BATCH_SIZE);
+ private final CombinableMetric mutationsSizeBytes = new CombinableMetricImpl(MUTATION_BYTES);
+ private final CombinableMetric totalCommitTimeForMutations = new CombinableMetricImpl(MUTATION_COMMIT_TIME);
+ private final CombinableMetric numFailedMutations = new CombinableMetricImpl(MUTATION_BATCH_FAILED_SIZE);
+
+ public MutationMetric(long numMutations, long mutationsSizeBytes, long commitTimeForMutations, long numFailedMutations) { this.numMutations.change(numMutations);
this.mutationsSizeBytes.change(mutationsSizeBytes);
this.totalCommitTimeForMutations.change(commitTimeForMutations);
this.numFailedMutations.change(numFailedMutations);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
index 3121ecd..9a2f426 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
@@ -42,15 +42,15 @@ public class OverAllQueryMetrics {
private final CombinableMetric queryFailed;
private final CombinableMetric cacheRefreshedDueToSplits;
- public OverAllQueryMetrics(LogLevel connectionLogLevel) {
+ public OverAllQueryMetrics(boolean isRequestMetricsEnabled, LogLevel connectionLogLevel) {
queryWatch = new MetricsStopWatch(WALL_CLOCK_TIME_MS.isLoggingEnabled(connectionLogLevel));
resultSetWatch = new MetricsStopWatch(RESULT_SET_TIME_MS.isLoggingEnabled(connectionLogLevel));
- numParallelScans = MetricUtil.getCombinableMetric(connectionLogLevel, NUM_PARALLEL_SCANS);
- wallClockTimeMS = MetricUtil.getCombinableMetric(connectionLogLevel, WALL_CLOCK_TIME_MS);
- resultSetTimeMS = MetricUtil.getCombinableMetric(connectionLogLevel, RESULT_SET_TIME_MS);
- queryTimedOut = MetricUtil.getCombinableMetric(connectionLogLevel, QUERY_TIMEOUT_COUNTER);
- queryFailed = MetricUtil.getCombinableMetric(connectionLogLevel, QUERY_FAILED_COUNTER);
- cacheRefreshedDueToSplits = MetricUtil.getCombinableMetric(connectionLogLevel, CACHE_REFRESH_SPLITS_COUNTER);
+ numParallelScans = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, NUM_PARALLEL_SCANS);
+ wallClockTimeMS = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, WALL_CLOCK_TIME_MS);
+ resultSetTimeMS = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, RESULT_SET_TIME_MS);
+ queryTimedOut = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_TIMEOUT_COUNTER);
+ queryFailed = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, QUERY_FAILED_COUNTER);
+ cacheRefreshedDueToSplits = MetricUtil.getCombinableMetric(isRequestMetricsEnabled,connectionLogLevel, CACHE_REFRESH_SPLITS_COUNTER);
}
public void updateNumParallelScans(long numParallelScans) {
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
index fa81e9c..6186f6b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
@@ -44,12 +44,15 @@ public class ReadMetricQueue {
private LogLevel connectionLogLevel;
- public ReadMetricQueue(LogLevel connectionLogLevel) {
+ private boolean isRequestMetricsEnabled;
+
+ public ReadMetricQueue(boolean isRequestMetricsEnabled, LogLevel connectionLogLevel) {
+ this.isRequestMetricsEnabled = isRequestMetricsEnabled;
this.connectionLogLevel = connectionLogLevel;
}
public CombinableMetric allotMetric(MetricType type, String tableName) {
- if (type.isLoggingEnabled(connectionLogLevel)) {
+ if (type.isLoggingEnabled(connectionLogLevel) || isRequestMetricsEnabled) {
MetricKey key = new MetricKey(type, tableName);
Queue<CombinableMetric> q = getMetricQueue(key);
CombinableMetric metric = getMetric(type);
@@ -95,7 +98,8 @@ public class ReadMetricQueue {
int size = metrics.size();
if (size == 0) { throw new IllegalArgumentException("Metrics collection needs to have at least one element"); }
Iterator<CombinableMetric> itr = metrics.iterator();
- CombinableMetric combinedMetric = itr.next();
+ //Clone first metric for combining so that aggregate always give consistent result
+ CombinableMetric combinedMetric = itr.next().clone();
while (itr.hasNext()) {
combinedMetric = combinedMetric.combine(itr.next());
}
@@ -186,6 +190,8 @@ public class ReadMetricQueue {
return scanMetricsHolderList;
}
-
+ public boolean isRequestMetricsEnabled() {
+ return isRequestMetricsEnabled;
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
index 494b3e6..dd6603e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
@@ -53,11 +53,11 @@ public class ScanMetricsHolder {
private Object scan;
private static final ScanMetricsHolder NO_OP_INSTANCE =
- new ScanMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "",null);
+ new ScanMetricsHolder(new ReadMetricQueue(false,LogLevel.OFF), "",null);
public static ScanMetricsHolder getInstance(ReadMetricQueue readMetrics, String tableName,
Scan scan, LogLevel connectionLogLevel) {
- if (connectionLogLevel == LogLevel.OFF) { return NO_OP_INSTANCE; }
+ if (connectionLogLevel == LogLevel.OFF && !readMetrics.isRequestMetricsEnabled()) { return NO_OP_INSTANCE; }
scan.setScanMetricsEnabled(true);
return new ScanMetricsHolder(readMetrics, tableName, scan);
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
index 699982f..8662f1a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/SpoolingMetricsHolder.java
@@ -27,7 +27,7 @@ public class SpoolingMetricsHolder {
private final CombinableMetric spoolFileSizeMetric;
private final CombinableMetric numSpoolFileMetric;
- public static final SpoolingMetricsHolder NO_OP_INSTANCE = new SpoolingMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "");
+ public static final SpoolingMetricsHolder NO_OP_INSTANCE = new SpoolingMetricsHolder(new ReadMetricQueue(false,LogLevel.OFF), "");
public SpoolingMetricsHolder(ReadMetricQueue readMetrics, String tableName) {
this.spoolFileSizeMetric = readMetrics.allotMetric(MetricType.SPOOL_FILE_SIZE, tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
index 6117b40..531a6b3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TaskExecutionMetricsHolder.java
@@ -37,7 +37,7 @@ public class TaskExecutionMetricsHolder {
private final CombinableMetric taskExecutionTime;
private final CombinableMetric numTasks;
private final CombinableMetric numRejectedTasks;
- public static final TaskExecutionMetricsHolder NO_OP_INSTANCE = new TaskExecutionMetricsHolder(new ReadMetricQueue(LogLevel.OFF), "");
+ public static final TaskExecutionMetricsHolder NO_OP_INSTANCE = new TaskExecutionMetricsHolder(new ReadMetricQueue(false,LogLevel.OFF), "");
public TaskExecutionMetricsHolder(ReadMetricQueue readMetrics, String tableName) {
taskQueueWaitTime = readMetrics.allotMetric(TASK_QUEUE_WAIT_TIME, tableName);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d51353a4/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java b/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
index d858e72..0e5684c 100644
--- a/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
+++ b/phoenix-core/src/test/java/org/apache/phoenix/iterate/SpoolingResultIteratorTest.java
@@ -58,7 +58,7 @@ public class SpoolingResultIteratorTest {
MemoryManager memoryManager = new DelegatingMemoryManager(new GlobalMemoryManager(threshold));
ResultIterator scanner = new SpoolingResultIterator(
SpoolingMetricsHolder.NO_OP_INSTANCE,
- new MemoryMetricsHolder(new ReadMetricQueue(LogLevel.OFF), ""), iterator, memoryManager, threshold,
+ new MemoryMetricsHolder(new ReadMetricQueue(false,LogLevel.OFF), ""), iterator, memoryManager, threshold,
maxSizeSpool, "/tmp");
AssertResults.assertResults(scanner, expectedResults);
}
[08/22] phoenix git commit: PHOENIX-4733 NPE while running sql
through file using psql
Posted by ja...@apache.org.
PHOENIX-4733 NPE while running sql through file using psql
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/7f916652
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/7f916652
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/7f916652
Branch: refs/heads/omid2
Commit: 7f916652298da0a7e4bca9c29a96f01000636e57
Parents: cc0de5d
Author: Ankit Singhal <an...@gmail.com>
Authored: Wed May 9 14:11:29 2018 -0700
Committer: Ankit Singhal <an...@gmail.com>
Committed: Wed May 9 14:11:29 2018 -0700
----------------------------------------------------------------------
.../org/apache/phoenix/log/QueryLoggerUtil.java | 24 ++++++++++++++------
1 file changed, 17 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7f916652/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
index 2f22931..d5c4878 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/log/QueryLoggerUtil.java
@@ -30,10 +30,14 @@ import com.google.common.collect.ImmutableMap.Builder;
public class QueryLoggerUtil {
- public static void logInitialDetails(QueryLogger queryLogger, PName tenantId,
- ConnectionQueryServices queryServices, String query, long startTime, List<Object> bindParameters) {
- queryLogger.log(QueryLogState.STARTED,
- getInitialDetails(tenantId, queryServices, query, startTime, bindParameters));
+ public static void logInitialDetails(QueryLogger queryLogger, PName tenantId, ConnectionQueryServices queryServices,
+ String query, long startTime, List<Object> bindParameters) {
+ try {
+ queryLogger.log(QueryLogState.STARTED,
+ getInitialDetails(tenantId, queryServices, query, startTime, bindParameters));
+ } catch (Exception e) {
+ // Ignore for now
+ }
}
@@ -46,15 +50,21 @@ public class QueryLoggerUtil {
} catch (UnknownHostException e) {
clientIP = "UnknownHost";
}
- queryLogBuilder.put(QueryLogInfo.CLIENT_IP_I, clientIP);
- queryLogBuilder.put(QueryLogInfo.QUERY_I, query);
+
+ if (clientIP != null) {
+ queryLogBuilder.put(QueryLogInfo.CLIENT_IP_I, clientIP);
+ }
+ if (query != null) {
+ queryLogBuilder.put(QueryLogInfo.QUERY_I, query);
+ }
queryLogBuilder.put(QueryLogInfo.START_TIME_I, startTime);
if (bindParameters != null) {
- queryLogBuilder.put(QueryLogInfo.BIND_PARAMETERS_I, StringUtils.join(bindParameters,","));
+ queryLogBuilder.put(QueryLogInfo.BIND_PARAMETERS_I, StringUtils.join(bindParameters, ","));
}
if (tenantId != null) {
queryLogBuilder.put(QueryLogInfo.TENANT_ID_I, tenantId.getString());
}
+
queryLogBuilder.put(QueryLogInfo.USER_I, queryServices.getUserName() != null ? queryServices.getUserName()
: queryServices.getUser().getShortName());
return queryLogBuilder.build();
[15/22] phoenix git commit: PHOENIX-4726 save sync index build start
timestamp
Posted by ja...@apache.org.
PHOENIX-4726 save sync index build start timestamp
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/d175c646
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/d175c646
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/d175c646
Branch: refs/heads/omid2
Commit: d175c646562c7b3e5e944a81cab38157dc385915
Parents: 09d0b3a
Author: Xu Cang <xc...@salesforce.com>
Authored: Mon May 7 14:25:42 2018 -0700
Committer: Vincent Poon <vi...@apache.org>
Committed: Mon May 14 10:25:24 2018 -0700
----------------------------------------------------------------------
.../phoenix/jdbc/PhoenixDatabaseMetaData.java | 1 +
.../org/apache/phoenix/schema/MetaDataClient.java | 17 +++++++++++++++++
2 files changed, 18 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d175c646/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index 55de772..2e12ca2 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -316,6 +316,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData {
public static final String ASYNC_CREATED_DATE = "ASYNC_CREATED_DATE";
public static final String SEQUENCE_TABLE_TYPE = SYSTEM_SEQUENCE_TABLE;
+ public static final String SYNC_INDEX_CREATED_DATE = "SYNC_INDEX_CREATED_DATE";
public static final String SYSTEM_MUTEX_TABLE_NAME = "MUTEX";
public static final String SYSTEM_MUTEX_NAME = SchemaUtil.getTableName(QueryConstants.SYSTEM_SCHEMA_NAME, SYSTEM_MUTEX_TABLE_NAME);
public static final TableName SYSTEM_MUTEX_HBASE_TABLE_NAME = TableName.valueOf(SYSTEM_MUTEX_NAME);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/d175c646/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
index 009289b..20ac732 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/MetaDataClient.java
@@ -74,6 +74,7 @@ import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.RETURN_TYPE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SALT_BUCKETS;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SORT_ORDER;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.STORE_NULLS;
+import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYNC_INDEX_CREATED_DATE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_SCHEMA;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_CATALOG_TABLE;
import static org.apache.phoenix.jdbc.PhoenixDatabaseMetaData.SYSTEM_FUNCTION_TABLE;
@@ -263,6 +264,14 @@ public class MetaDataClient {
TABLE_NAME + "," +
ASYNC_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
") VALUES (?, ?, ?, ?)";
+
+ private static final String SET_INDEX_SYNC_CREATED_DATE =
+ "UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
+ TENANT_ID + "," +
+ TABLE_SCHEM + "," +
+ TABLE_NAME + "," +
+ SYNC_INDEX_CREATED_DATE + " " + PDate.INSTANCE.getSqlTypeName() +
+ ") VALUES (?, ?, ?, ?)";
private static final String CREATE_TABLE =
"UPSERT INTO " + SYSTEM_CATALOG_SCHEMA + ".\"" + SYSTEM_CATALOG_TABLE + "\"( " +
TENANT_ID + "," +
@@ -2678,6 +2687,14 @@ public class MetaDataClient {
setAsync.setString(3, tableName);
setAsync.setDate(4, asyncCreatedDate);
setAsync.execute();
+ } else {
+ Date syncCreatedDate = new Date(EnvironmentEdgeManager.currentTimeMillis());
+ PreparedStatement setSync = connection.prepareStatement(SET_INDEX_SYNC_CREATED_DATE);
+ setSync.setString(1, tenantIdStr);
+ setSync.setString(2, schemaName);
+ setSync.setString(3, tableName);
+ setSync.setDate(4, syncCreatedDate);
+ setSync.execute();
}
tableMetaData.addAll(connection.getMutationState().toMutations(timestamp).next().getSecond());
connection.rollback();
[06/22] phoenix git commit: PHOENIX-4646 The data exceeds the max
capacity for the data type error for valid scenarios (Sergey Soldatov)
Posted by ja...@apache.org.
PHOENIX-4646 The data exceeds the max capacity for the data type error for valid scenarios (Sergey Soldatov)
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/cc0de5d2
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/cc0de5d2
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/cc0de5d2
Branch: refs/heads/omid2
Commit: cc0de5d274e56205051dc5968281f5152e035ab2
Parents: 7c20fda
Author: James Taylor <jt...@salesforce.com>
Authored: Wed May 9 13:12:26 2018 -0700
Committer: James Taylor <jt...@salesforce.com>
Committed: Wed May 9 13:37:10 2018 -0700
----------------------------------------------------------------------
.../apache/phoenix/end2end/UpsertSelectIT.java | 19 +++++++++++++++++++
.../apache/phoenix/schema/types/PVarchar.java | 10 ++++++----
2 files changed, 25 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc0de5d2/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
index 275d72d..e6feea6 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/UpsertSelectIT.java
@@ -1486,6 +1486,25 @@ public class UpsertSelectIT extends ParallelStatsDisabledIT {
}
}
+ @Test // See https://issues.apache.org/jira/browse/PHOENIX-4646
+ public void testLengthLimitedVarchar() throws Exception {
+ String tableName1 = generateUniqueName();
+ String tableName2 = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ conn.setAutoCommit(true);
+ conn.createStatement().execute("create table " + tableName1 + "(name varchar(160) primary key, id varchar(120), address varchar(160))");
+ conn.createStatement().execute("create table " + tableName2 + "(name varchar(160) primary key, id varchar(10), address varchar(10))");
+ conn.createStatement().execute("upsert into " + tableName1 + " values('test','test','test')");
+ conn.createStatement().execute("upsert into " + tableName2 + " select * from " + tableName1);
+ ResultSet rs = conn.createStatement().executeQuery("select * from " + tableName2);
+ assertTrue(rs.next());
+ assertEquals("test", rs.getString(1));
+ assertEquals("test", rs.getString(2));
+ assertEquals("test", rs.getString(2));
+ assertFalse(rs.next());
+ }
+ }
+
private static Connection getTenantConnection(String tenantId) throws Exception {
Properties props = PropertiesUtil.deepCopy(TestUtil.TEST_PROPERTIES);
props.setProperty(TENANT_ID_ATTRIB, tenantId);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/cc0de5d2/phoenix-core/src/main/java/org/apache/phoenix/schema/types/PVarchar.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/schema/types/PVarchar.java b/phoenix-core/src/main/java/org/apache/phoenix/schema/types/PVarchar.java
index fb5a045..a99925c 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/schema/types/PVarchar.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/schema/types/PVarchar.java
@@ -102,15 +102,17 @@ public class PVarchar extends PDataType<String> {
SortOrder sortOrder, Integer maxLength, Integer scale,
Integer desiredMaxLength, Integer desiredScale) {
if (ptr.getLength() != 0 && desiredMaxLength != null) {
- if (maxLength == null) {
+ if (maxLength == null || maxLength > desiredMaxLength) {
if (value != null) { // Use value if provided
maxLength = value.toString().length();
} else {
- coerceBytes(ptr, value, srcType, maxLength, scale, sortOrder, desiredMaxLength, desiredScale, sortOrder, true);
- maxLength = StringUtil.calculateUTF8Length(ptr.get(), ptr.getOffset(), ptr.getLength(), sortOrder);
+ coerceBytes(ptr, value, srcType, maxLength, scale, sortOrder, desiredMaxLength,
+ desiredScale, sortOrder, true);
+ maxLength = StringUtil
+ .calculateUTF8Length(ptr.get(), ptr.getOffset(), ptr.getLength(), sortOrder);
}
+ return maxLength <= desiredMaxLength;
}
- return maxLength <= desiredMaxLength;
}
return true;
}