phoenix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From pbo...@apache.org
Subject [05/13] phoenix git commit: PHOENIX-4701 Write client-side metrics asynchronously to SYSTEM.LOG
Date Sat, 19 May 2018 18:58:40 GMT
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/5ba5f819
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/5ba5f819
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/5ba5f819

Branch: refs/heads/4.x-cdh5.13
Commit: 5ba5f8195716de3deda7037bd0fcf41260d26179
Parents: 88cbbeb
Author: Ankit Singhal <ankitsinghal59@gmail.com>
Authored: Mon May 14 13:39:59 2018 -0700
Committer: James Taylor <jtaylor@salesforce.com>
Committed: Wed May 16 21:48:05 2018 -0700

----------------------------------------------------------------------
 .../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 +-
 40 files changed, 566 insertions(+), 480 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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 ca3e6d0..b823829 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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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/5ba5f819/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);
+    }
+
+}


Mime
View raw message