PHOENIX-4870 LoggingPhoenixConnection should log metrics when AutoCommit is set to True.
Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/6ea2110b Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/6ea2110b Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/6ea2110b Branch: refs/heads/4.x-cdh5.15 Commit: 6ea2110bacd97943b17dbbe4484bf8a6da9dde7a Parents: bb297e7 Author: s.kadam <s.ka...@gus.com> Authored: Thu Sep 6 01:00:03 2018 +0100 Committer: Pedro Boado <pbo...@apache.org> Committed: Wed Oct 17 22:49:38 2018 +0100 ---------------------------------------------------------------------- .../monitoring/PhoenixLoggingMetricsIT.java | 61 +++++++++++++++++++- .../phoenix/jdbc/LoggingPhoenixConnection.java | 37 +++++++----- .../jdbc/LoggingPhoenixPreparedStatement.java | 25 +++++++- .../phoenix/jdbc/LoggingPhoenixStatement.java | 28 +++++++-- 4 files changed, 125 insertions(+), 26 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ea2110b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java index 5d5524c..483d341 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixLoggingMetricsIT.java @@ -102,7 +102,8 @@ public class PhoenixLoggingMetricsIT extends BasePhoenixMetricsIT { public void testPhoenixMetricsLoggedOnCommit() throws Exception { // run SELECT to verify read metrics are logged String query = "SELECT * FROM " + tableName1; - verifyQueryLevelMetricsLogging(query); + ResultSet rs = upsertRows(query); + verifyQueryLevelMetricsLogging(query, rs); // run UPSERT SELECT to verify mutation metrics are logged String upsertSelect = "UPSERT INTO " + tableName2 + " SELECT * FROM " + tableName1; @@ -140,7 +141,9 @@ public class PhoenixLoggingMetricsIT extends BasePhoenixMetricsIT { public void testPhoenixMetricsLoggedOnClose() throws Exception { // run SELECT to verify read metrics are logged String query = "SELECT * FROM " + tableName1; - verifyQueryLevelMetricsLogging(query); + + ResultSet rs = upsertRows(query); + verifyQueryLevelMetricsLogging(query, rs); // run UPSERT SELECT to verify mutation metrics are logged String upsertSelect = "UPSERT INTO " + tableName2 + " SELECT * FROM " + tableName1; @@ -164,13 +167,61 @@ public class PhoenixLoggingMetricsIT extends BasePhoenixMetricsIT { mutationReadMetricsMap.size() == 0); } + /** + * This test is added to verify if metrics are being logged in case + * auto commit is set to true. + */ + @Test + public void testPhoenixMetricsLoggedOnAutoCommitTrue() throws Exception { + loggedConn.setAutoCommit(true); + + String query = "SELECT * FROM " + tableName1; + ResultSet rs = upsertRows(query); + verifyQueryLevelMetricsLogging(query, rs); + + // run UPSERT SELECT to verify mutation metrics are logged + String upsertSelect = "UPSERT INTO " + tableName2 + " SELECT * FROM " + tableName1; + loggedConn.createStatement().executeUpdate(upsertSelect); + + assertTrue("Mutation write metrics are not logged for " + tableName2, + mutationWriteMetricsMap.get(tableName2).size() > 0); + assertTrue("Mutation read metrics are not found for " + tableName1, + mutationReadMetricsMap.get(tableName1).size() > 0); + + clearAllTestMetricMaps(); + + loggedConn.createStatement().execute(query); + assertTrue("Read metrics found for " + tableName1, + mutationReadMetricsMap.size() == 0); + loggedConn.createStatement().execute(upsertSelect); + + assertTrue("Mutation write metrics are not logged for " + tableName2 + + " in createStatement",mutationWriteMetricsMap.get(tableName2).size() > 0); + assertTrue("Mutation read metrics are not found for " + tableName1 + + " in createStatement",mutationReadMetricsMap.get(tableName1).size() > 0); + + clearAllTestMetricMaps(); + + loggedConn.prepareStatement(query).executeQuery(); + assertTrue("Read metrics found for " + tableName1, + mutationReadMetricsMap.size() == 0); + + loggedConn.prepareStatement(upsertSelect).executeUpdate(); + assertTrue("Mutation write metrics are not logged for " + tableName2 + + " in prepareStatement",mutationWriteMetricsMap.get(tableName2).size() > 0); + assertTrue("Mutation read metrics are not found for " + tableName1 + + " in prepareStatement",mutationReadMetricsMap.get(tableName1).size() > 0); + + + } + private ResultSet executeAndGetResultSet(String query) throws Exception { Statement stmt = loggedConn.createStatement(); stmt.execute(query); return stmt.getResultSet(); } - private void verifyQueryLevelMetricsLogging(String query) throws SQLException { + private ResultSet upsertRows(String query) throws SQLException { Statement stmt = loggedConn.createStatement(); ResultSet rs = stmt.executeQuery(query); assertTrue(rs instanceof LoggingPhoenixResultSet); @@ -180,6 +231,10 @@ public class PhoenixLoggingMetricsIT extends BasePhoenixMetricsIT { } rs.close(); assertTrue(rowsRetrievedCounter == NUM_ROWS); + return rs; + } + + private void verifyQueryLevelMetricsLogging(String query , ResultSet rs) throws SQLException { assertTrue("Read metrics for not found for " + tableName1, requestReadMetricsMap.get(tableName1).size() > 0); assertTrue("Logged query doesn't match actual query", loggedSql.equals(query)); http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ea2110b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixConnection.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixConnection.java index 37917e2..af0f803 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixConnection.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixConnection.java @@ -40,14 +40,16 @@ public class LoggingPhoenixConnection extends DelegateConnection { @Override public Statement createStatement() throws SQLException { - return new LoggingPhoenixStatement(super.createStatement(), phoenixMetricsLog); + return new LoggingPhoenixStatement(super.createStatement(), phoenixMetricsLog, + this); } @Override public Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException { return new LoggingPhoenixStatement( - super.createStatement(resultSetType, resultSetConcurrency), phoenixMetricsLog); + super.createStatement(resultSetType, resultSetConcurrency), phoenixMetricsLog, + this); } @Override @@ -55,13 +57,13 @@ public class LoggingPhoenixConnection extends DelegateConnection { int resultSetHoldability) throws SQLException { return new LoggingPhoenixStatement( super.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability), - phoenixMetricsLog); + phoenixMetricsLog, this); } @Override public PreparedStatement prepareStatement(String sql) throws SQLException { return new LoggingPhoenixPreparedStatement(super.prepareStatement(sql), - phoenixMetricsLog, sql); + phoenixMetricsLog, sql, this); } @Override @@ -69,52 +71,57 @@ public class LoggingPhoenixConnection extends DelegateConnection { int resultSetConcurrency) throws SQLException { return new LoggingPhoenixPreparedStatement( super.prepareStatement(sql, resultSetType, resultSetConcurrency), - phoenixMetricsLog, sql); + phoenixMetricsLog, sql, this); } @Override public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability) throws SQLException { return new LoggingPhoenixPreparedStatement(super.prepareStatement(sql, resultSetType, - resultSetConcurrency, resultSetHoldability), phoenixMetricsLog, sql); + resultSetConcurrency, resultSetHoldability), phoenixMetricsLog, sql, this); } @Override public PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException { return new LoggingPhoenixPreparedStatement(super.prepareStatement(sql, autoGeneratedKeys), - phoenixMetricsLog, sql); + phoenixMetricsLog, sql, this); } @Override public PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException { return new LoggingPhoenixPreparedStatement(super.prepareStatement(sql, columnIndexes), - phoenixMetricsLog, sql); + phoenixMetricsLog, sql, this); } @Override public PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException { return new LoggingPhoenixPreparedStatement(super.prepareStatement(sql, columnNames), - phoenixMetricsLog, sql); + phoenixMetricsLog, sql, this); } @Override public void commit() throws SQLException { super.commit(); - phoenixMetricsLog.logWriteMetricsfoForMutationsSinceLastReset(PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn)); - phoenixMetricsLog.logReadMetricInfoForMutationsSinceLastReset(PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(conn)); - PhoenixRuntime.resetMetrics(conn); + loggingMetricsHelper(); } @Override public void close() throws SQLException { try { - phoenixMetricsLog.logWriteMetricsfoForMutationsSinceLastReset(PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn)); - phoenixMetricsLog.logReadMetricInfoForMutationsSinceLastReset(PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(conn)); - PhoenixRuntime.resetMetrics(conn); + loggingMetricsHelper(); } finally { super.close(); } } + + public void loggingMetricsHelper() throws SQLException { + + phoenixMetricsLog.logWriteMetricsfoForMutationsSinceLastReset( + PhoenixRuntime.getWriteMetricInfoForMutationsSinceLastReset(conn)); + phoenixMetricsLog.logReadMetricInfoForMutationsSinceLastReset( + PhoenixRuntime.getReadMetricInfoForMutationsSinceLastReset(conn)); + PhoenixRuntime.resetMetrics(conn); + } } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ea2110b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixPreparedStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixPreparedStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixPreparedStatement.java index dbeea0d..12edde9 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixPreparedStatement.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixPreparedStatement.java @@ -21,16 +21,21 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.SQLFeatureNotSupportedException; +import java.sql.Connection; + public class LoggingPhoenixPreparedStatement extends DelegatePreparedStatement { private PhoenixMetricsLog phoenixMetricsLog; private String sql; + private Connection conn; - public LoggingPhoenixPreparedStatement(PreparedStatement stmt, PhoenixMetricsLog phoenixMetricsLog, String sql) { + public LoggingPhoenixPreparedStatement(PreparedStatement stmt, + PhoenixMetricsLog phoenixMetricsLog, String sql, Connection conn) { super(stmt); this.phoenixMetricsLog = phoenixMetricsLog; this.sql = sql; + this.conn = conn; } @Override @@ -40,7 +45,9 @@ public class LoggingPhoenixPreparedStatement extends DelegatePreparedStatement { @Override public ResultSet executeQuery() throws SQLException { - return new LoggingPhoenixResultSet(super.executeQuery(), phoenixMetricsLog, sql); + ResultSet rs = new LoggingPhoenixResultSet(super.executeQuery(), phoenixMetricsLog, sql); + this.loggingAutoCommitHelper(); + return rs; } @Override @@ -50,10 +57,22 @@ public class LoggingPhoenixPreparedStatement extends DelegatePreparedStatement { return (resultSet == null) ? null : new LoggingPhoenixResultSet(resultSet, phoenixMetricsLog, sql); } + + @Override + public int executeUpdate() throws SQLException { + int res = super.executeUpdate(); + this.loggingAutoCommitHelper(); + return res; + } @Override public ResultSet getGeneratedKeys() throws SQLException { return new LoggingPhoenixResultSet(super.getGeneratedKeys(), phoenixMetricsLog, sql); } - + + private void loggingAutoCommitHelper() throws SQLException { + if(conn.getAutoCommit() && (conn instanceof LoggingPhoenixConnection)) { + ((LoggingPhoenixConnection)conn).loggingMetricsHelper(); + } + } } http://git-wip-us.apache.org/repos/asf/phoenix/blob/6ea2110b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixStatement.java ---------------------------------------------------------------------- diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixStatement.java index de33893..d31f521 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixStatement.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/LoggingPhoenixStatement.java @@ -17,6 +17,8 @@ */ package org.apache.phoenix.jdbc; + +import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; @@ -25,28 +27,38 @@ public class LoggingPhoenixStatement extends DelegateStatement { private PhoenixMetricsLog phoenixMetricsLog; private String sql; - - public LoggingPhoenixStatement(Statement stmt, PhoenixMetricsLog phoenixMetricsLog) { + private Connection conn; + + public LoggingPhoenixStatement(Statement stmt, PhoenixMetricsLog phoenixMetricsLog, Connection conn) { super(stmt); this.phoenixMetricsLog = phoenixMetricsLog; + this.conn = conn; } @Override public boolean execute(String sql) throws SQLException { + boolean result; this.sql = sql; - return super.execute(sql); + result = super.execute(sql); + this.loggingAutoCommitHelper(); + return result; } @Override public ResultSet executeQuery(String sql) throws SQLException { this.sql = sql; - return new LoggingPhoenixResultSet(super.executeQuery(sql), phoenixMetricsLog, this.sql); + ResultSet rs = new LoggingPhoenixResultSet(super.executeQuery(sql), phoenixMetricsLog, this.sql); + this.loggingAutoCommitHelper(); + return rs; } @Override public int executeUpdate(String sql) throws SQLException { + int result; this.sql = sql; - return super.executeUpdate(sql); + result = super.executeUpdate(sql); + this.loggingAutoCommitHelper(); + return result; } @Override @@ -62,4 +74,10 @@ public class LoggingPhoenixStatement extends DelegateStatement { return new LoggingPhoenixResultSet(super.getGeneratedKeys(), phoenixMetricsLog, this.sql); } + private void loggingAutoCommitHelper() throws SQLException { + if(conn.getAutoCommit() && (conn instanceof LoggingPhoenixConnection)) { + ((LoggingPhoenixConnection)conn).loggingMetricsHelper(); + } + } + }
