PHOENIX-5838 Add Histograms for Table level Metrics

phoenix-core/pom.xml

@@ -494,6 +494,10 @@
  <artifactId>curator-framework</artifactId>
  <version>${curator.version}</version>
  </dependency>
+ <dependency>
+ <groupId>org.hdrhistogram</groupId>
+ <artifactId>HdrHistogram</artifactId>
+ </dependency>
  </dependencies>
 
  <profiles>

phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java

@@ -57,6 +57,9 @@
 import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT;
 import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_COMMIT_TIME;
 import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SCAN_BYTES;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
@@ -400,6 +403,50 @@ private static void assertMutationTableMetrics(final boolean isUpsert, final Str
  }
  }
 
+ private void assertHistogramMetricsForMutations(String tableName, boolean isUpsert,
+ long ltCount, long szCount, boolean verifyMetricValues) {
+ LatencyHistogram ltHisto;
+ SizeHistogram szHisto;
+ if (isUpsert) {
+ ltHisto = TableMetricsManager.getUpsertLatencyHistogramForTable(tableName);
+ szHisto = TableMetricsManager.getUpsertSizeHistogramForTable(tableName);
+ } else {
+ ltHisto = TableMetricsManager.getDeleteLatencyHistogramForTable(tableName);
+ szHisto = TableMetricsManager.getDeleteSizeHistogramForTable(tableName);
+ }
+ assertNotNull(ltHisto);
+ assertNotNull(szHisto);
+ assertEquals(ltCount, ltHisto.getHistogram().getTotalCount());
+ assertEquals(szCount, szHisto.getHistogram().getTotalCount());
+
+ // If we are just comparing one data point then we can compare with table metrics
+ // or global metrics but if there are multiple data points then we can't compare histogram
+ // data points with global metrics.
+ if (verifyMetricValues) {
+ long sqlTime;
+ if (isUpsert) {
+ sqlTime = getMetricFromTableMetrics(tableName, MetricType.UPSERT_SQL_QUERY_TIME);
+ } else {
+ sqlTime = getMetricFromTableMetrics(tableName, MetricType.DELETE_SQL_QUERY_TIME);
+
+ }
+ long commitTime = getMetricFromTableMetrics(tableName, MetricType.MUTATION_COMMIT_TIME);
+ // Latency metric for mutation is sum of time spent in executeMutation
+ // and PhoenixConnection#commit time.
+ long totalCommitTimeFromMetrics = sqlTime + commitTime;
+
+ // Histogram#maxValue is the last value in the bucket. So we can't compare directly
+ // maxValue with totalCommitTimeFromMetrics.
+ Assert.assertTrue(ltHisto.getHistogram().valuesAreEquivalent(totalCommitTimeFromMetrics,
+ ltHisto.getHistogram().getMaxValue()));
+
+ long mutationBytesFromGlobalMetrics = GLOBAL_MUTATION_BYTES.getMetric().getValue();
+ Assert.assertTrue(szHisto.getHistogram().valuesAreEquivalent(mutationBytesFromGlobalMetrics,
+ szHisto.getHistogram().getMaxValue()));
+ }
+
+ }
+
  /**
  * Checks that if the metric is of the passed in type, it has the expected value
  * (based on the CompareOp). If the metric type is different than checkType, ignore
@@ -1181,25 +1228,166 @@ private static void assertMetricValue(Metric m, MetricType checkType, long compa
  assertNotNull("Failed to get a connection!", conn);
  // Get write metrics before closing the connection since that clears those metrics
  Map<MetricType, Long>
- writeMutMetrics =
- getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
+writeMutMetrics =
+getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName);
  conn.close();
  // 1 regular upsert + numAtomicUpserts
  // 2 mutations (regular and atomic on the same row in the same batch will be split)
  assertMutationTableMetrics(true, tableName, 1 + numAtomicUpserts, 0, 0, true, 2, 0, 0, 2, 0,
- writeMutMetrics, conn);
+writeMutMetrics, conn);
  assertEquals(numAtomicUpserts, getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_SQL_COUNTER));
  assertTrue(getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_COMMIT_TIME) > 0);
  }
  }
 
+ @Test
+ public void testHistogramMetricsForMutations() throws Exception {
+ String tableName = generateUniqueName();
+ // Reset table level metrics to capture histogram metrics for upsert.
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, true, true, 10, true, conn, false);
+ }
+ // Metrics will be reset after creation of table so below we will get latency
+ // just for upsert queries.
+ // Since we are recording latency histograms after every executeMutation method and
+ // since we are not batch upserting, it will record histogram event after every upsert.
+ assertHistogramMetricsForMutations(tableName, true, 1, 1, true);
+
+ // Reset table histograms as well as global metrics
+ PhoenixRuntime.clearTableLevelMetrics();
+ PhoenixMetricsIT.resetGlobalMetrics();
+ try (Connection connection = getConnFromTestDriver();
+ Statement statement = connection.createStatement()) {
+ String delete = "DELETE FROM " + tableName;
+ statement.execute(delete);
+ connection.commit();
+ }
+ // Verify metrics for delete mutations
+ assertHistogramMetricsForMutations(tableName, false, 1, 1, true);
+ PhoenixRuntime.clearTableLevelMetrics();
+ }
+
+ @Test
+ public void testHistogramMetricsForMutationsAutoCommitTrue() throws Exception {
+ String tableName = generateUniqueName();
+ // Reset table level metrics to capture histogram metrics for upsert.
+ try (Connection conn = getConnFromTestDriver()) {
+ conn.setAutoCommit(true);
+ createTableAndInsertValues(tableName, true, true, 10, false, conn, false);
+ }
+ // Metrics will be reset after creation of table so below we will get latency
+ // just for upsert queries.
+ // Since we are recording latency histograms after every executeMutation method and
+ // since we are not batch upserting, it will record histogram event after every upsert.
+ assertHistogramMetricsForMutations(tableName, true, 10, 10, false);
+
+ // Reset table histograms as well as global metrics
+ PhoenixRuntime.clearTableLevelMetrics();
+ PhoenixMetricsIT.resetGlobalMetrics();
+ try (Connection connection = getConnFromTestDriver();
+ Statement statement = connection.createStatement()) {
+ connection.setAutoCommit(true);
+ String delete = "DELETE FROM " + tableName;
+ statement.execute(delete);
+ }
+ // Verify metrics for delete mutations. We won't get any data point for
+ // size histogram since delete happened on server side using ServerSelectDeleteMutationPlan.
+ assertHistogramMetricsForMutations(tableName, false,1, 0, false);
+ PhoenixRuntime.clearTableLevelMetrics();
+ }
+
+ @Test
+ public void testHistogramMetricsForQueries() throws Exception {
+ String tableName = generateUniqueName();
+ // Reset table level metrics to capture histogram metrics for select queries.
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, true, true, 10, true, conn, true);
+ }
+ // Reset table metrics as well as global metrics
+ PhoenixRuntime.clearTableLevelMetrics();
+ PhoenixMetricsIT.resetGlobalMetrics();
+ try (Connection conn = getConnFromTestDriver();
+ Statement statement = conn.createStatement()) {
+ String select = "SELECT * FROM " + tableName;
+ ResultSet resultSet = statement.executeQuery(select);
+ while (resultSet.next()) {
+ // do nothing
+ }
+ resultSet.close();
+ } // conn close will close the rs at which point we will increment the scan_bytes counter
+
+ // Verify that value from histogram is equal to metric from global metrics.
+ LatencyHistogram ltHisto = TableMetricsManager.getQueryLatencyHistogramForTable(tableName);
+ SizeHistogram szHisto = TableMetricsManager.getQuerySizeHistogramForTable(tableName);
+
+ assertHistogramMetricsForQueries(tableName, ltHisto, szHisto, 1, 1);
+ }
+
+ @Test
+ public void testHistogramMetricsForRangeScan() throws Exception {
+ String tableName = generateUniqueName();
+ // Reset table level metrics to capture histogram metrics for select queries.
+ try (Connection conn = getConnFromTestDriver()) {
+ createTableAndInsertValues(tableName, true, true, 10, true, conn, true);
+ }
+ // Reset global metrics and table level metrics.
+ PhoenixMetricsIT.resetGlobalMetrics();
+ PhoenixRuntime.clearTableLevelMetrics();
+ try (Connection conn = getConnFromTestDriver();
+ Statement statement = conn.createStatement()) {
+ String select = "SELECT * FROM " + tableName;
+ ResultSet resultSet = statement.executeQuery(select);
+ while (resultSet.next()) {
+ // do nothing
+ }
+ } // conn close will close the rs at which point we will increment the scan_bytes counter
+
+ // Make sure that point lookup histograms are empty since this is a range scan query.
+ LatencyHistogram pointLookupLtHisto =
+ TableMetricsManager.getPointLookupLatencyHistogramForTable(tableName);
+ SizeHistogram pointLookupSzHisto =
+ TableMetricsManager.getPointLookupSizeHistogramForTable(tableName);
+ Assert.assertEquals(0, pointLookupLtHisto.getHistogram().getTotalCount());
+ Assert.assertEquals(0, pointLookupSzHisto.getHistogram().getTotalCount());
+
+ LatencyHistogram ltHistogram =
+ TableMetricsManager.getRangeScanLatencyHistogramForTable(tableName);
+ Assert.assertEquals(1, ltHistogram.getHistogram().getTotalCount());
+ SizeHistogram sizeHistogram =
+ TableMetricsManager.getRangeScanSizeHistogramForTable(tableName);
+ Assert.assertEquals(1, sizeHistogram.getHistogram().getTotalCount());
+
+ // Verify that value from histogram is equal to metric from global metrics.
+ assertHistogramMetricsForQueries(tableName, ltHistogram, sizeHistogram, 1, 1);
+ }
+
  private Connection getConnFromTestDriver() throws SQLException {
  Connection conn = DriverManager.getConnection(url);
  assertTrue(conn.unwrap(PhoenixConnection.class)
  .getQueryServices() instanceof PhoenixMetricsTestingQueryServices);
  return conn;
  }
 
+ // Verify that there is a histogram counter for the operation and verify with table level metrics
+ private void assertHistogramMetricsForQueries(String tableName, LatencyHistogram ltHistogram,
+ SizeHistogram sizeHistogram, int ltCount, int szCount) {
+ Assert.assertEquals(ltCount, ltHistogram.getHistogram().getTotalCount());
+ Assert.assertEquals(szCount, sizeHistogram.getHistogram().getTotalCount());
+
+ // Get latency metrics from table level metrics
+ Long queryTime = GLOBAL_QUERY_TIME.getMetric().getValue();
+ long rsNextTime = getMetricFromTableMetrics(tableName, MetricType.RESULT_SET_TIME_MS);
+ // Latency for queries is sum of time spent in executeQuery phase and rs.next phase.
+ long totalLatency = queryTime + rsNextTime;
+ long maxLtValue = ltHistogram.getHistogram().getMaxValue();
+ Assert.assertTrue(ltHistogram.getHistogram().valuesAreEquivalent(totalLatency, maxLtValue));
+
+ Long scanBytes = GLOBAL_SCAN_BYTES.getMetric().getValue();
+ long maxSzValue = sizeHistogram.getHistogram().getMaxValue();
+ Assert.assertTrue(sizeHistogram.getHistogram().valuesAreEquivalent(scanBytes, maxSzValue));
+ }
+
+
  private long getMetricFromTableMetrics(String tableName, MetricType type) {
  Long value = TableMetricsManager.getMetricValue(tableName, type);
  Assert.assertNotNull(value);

phoenix-core/src/main/java/org/apache/phoenix/compile/UpsertCompiler.java

@@ -1038,7 +1038,7 @@ private static void throwIfNotUpdatable(TableRef tableRef, Set<PColumn> overlapV
  }
  }
 
-private class ServerUpsertSelectMutationPlan implements MutationPlan {
+public class ServerUpsertSelectMutationPlan implements MutationPlan {
  private final QueryPlan queryPlan;
  private final TableRef tableRef;
  private final QueryPlan originalQueryPlan;

phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java

@@ -38,6 +38,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -164,6 +165,7 @@ public class MutationState implements SQLCloseable {
  private final MutationMetricQueue mutationMetricQueue;
  private ReadMetricQueue readMetricQueue;
 
+ private Map<String, Long> timeInExecuteMutationMap = new HashMap<>();
  private static boolean allUpsertsMutations = true;
  private static boolean allDeletesMutations = true;
 
@@ -1496,6 +1498,16 @@ public List<Mutation> getMutationList() {
  TableMetricsManager.updateMetricsMethod(htableNameStr, allUpsertsMutations ? UPSERT_AGGREGATE_FAILURE_SQL_COUNTER :
  DELETE_AGGREGATE_FAILURE_SQL_COUNTER, 1);
  }
+ // Update size and latency histogram metrics.
+ TableMetricsManager.updateSizeHistogramMetricsForMutations(htableNameStr,
+ committedMutationsMetric.getTotalMutationsSizeBytes().getValue(), allUpsertsMutations);
+ Long latency = timeInExecuteMutationMap.get(htableNameStr);
+ if (latency == null) {
+ latency = 0l;
+ }
+ latency += mutationCommitTime;
+ TableMetricsManager.updateLatencyHistogramForMutations(htableNameStr,
+ latency, allUpsertsMutations);
 
  }
  resetAllMutationState();
@@ -2174,4 +2186,17 @@ public MutationMetricQueue getMutationMetricQueue() {
  return mutationMetricQueue;
  }
 
+ public void addExecuteMutationTime(long time, String tableName) {
+ Long timeSpent = timeInExecuteMutationMap.get(tableName);
+ if (timeSpent == null) {
+ timeSpent = 0l;
+ }
+ timeSpent += time;
+ timeInExecuteMutationMap.put(tableName, timeSpent);
+ }
+
+ public void resetExecuteMutationTimeMap() {
+ timeInExecuteMutationMap.clear();
+ }
+
 }

phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java

@@ -730,7 +730,11 @@ public void commit() throws SQLException {
  @Override
  public Void call() throws SQLException {
  checkOpen();
- mutationState.commit();
+ try {
+ mutationState.commit();
+ } finally {
+ mutationState.resetExecuteMutationTimeMap();
+ }
  return null;
  }
  }, Tracing.withTracing(this, "committing mutations"));

phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java

@@ -906,6 +906,20 @@ private void updateTableLevelReadMetrics(String tableName, boolean isPointLookup
  metricsFromOverallQuery.put(tableName, overAllReadMetrics);
  TableMetricsManager.pushMetricsFromConnInstanceMethod(metricsFromOverallQuery);
  if (readMetrics.get(tableName) != null) {
+ Long scanBytes = readMetrics.get(tableName).get(MetricType.SCAN_BYTES);
+ if (scanBytes == null) {
+ scanBytes = 0L;
+ }
+ TableMetricsManager.updateHistogramMetricsForQueryScanBytes(
+ scanBytes, tableName, isPointLookup);
+ Long timeSpentInRSNext = overAllReadMetrics.get(MetricType.RESULT_SET_TIME_MS);
+
+ if (timeSpentInRSNext == null) {
+ timeSpentInRSNext = 0l;
+ }
+ timeSpentInRSNext += queryTime;
+ TableMetricsManager.updateHistogramMetricsForQueryLatency(tableName, timeSpentInRSNext, isPointLookup);
+
  TableMetricsManager.updateMetricsMethod(tableName, this.exception == null ?
  MetricType.SELECT_AGGREGATE_SUCCESS_SQL_COUNTER :
  MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER, 1);

phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java

@@ -619,6 +619,17 @@ public Integer call() throws SQLException {
  TableMetricsManager.updateMetricsMethod(tableName, isUpsert ? UPSERT_AGGREGATE_FAILURE_SQL_COUNTER:
  DELETE_AGGREGATE_FAILURE_SQL_COUNTER, 1);
  }
+ if (plan instanceof DeleteCompiler.ServerSelectDeleteMutationPlan
+ || plan instanceof UpsertCompiler.ServerUpsertSelectMutationPlan) {
+ TableMetricsManager.updateLatencyHistogramForMutations(
+ tableName, executeMutationTimeSpent, false);
+ // We won't have size histograms for delete mutations when auto commit is set to true and
+ // if plan is of ServerSelectDeleteMutationPlan or ServerUpsertSelectMutationPlan
+ // since the update happens on server.
+ } else {
+ state.addExecuteMutationTime(
+ executeMutationTimeSpent, tableName);
+ }
  }
  }
 

phoenix-core/src/main/java/org/apache/phoenix/mapreduce/util/PhoenixConfigurationUtil.java

@@ -878,4 +878,27 @@ public static boolean isMRSnapshotManagedExternally(final Configuration configur
  return isSnapshotRestoreManagedExternally;
  }
 
+ /**
+ * Get the value of the <code>name</code> property as a set of comma-delimited
+ * <code>long</code> values.
+ * If no such property exists, null is returned.
+ * Hadoop Configuration object has support for getting ints delimited by comma
+ * but doesn't support for long.
+ * @param name property name
+ * @return property value interpreted as an array of comma-delimited
+ * <code>long</code> values
+ */
+ public static long[] getLongs(Configuration conf, String name) {
+ String[] strings = conf.getTrimmedStrings(name);
+ // Configuration#getTrimmedStrings will never return null.
+ // If key is not found, it will return empty array.
+ if (strings.length == 0) {
+ return null;
+ }
+ long[] longs = new long[strings.length];
+ for (int i = 0; i < strings.length; i++) {
+ longs[i] = Long.parseLong(strings[i]);
+ }
+ return longs;
+ }
 }