You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@phoenix.apache.org by sa...@apache.org on 2026/02/08 18:29:00 UTC
(phoenix) branch master updated: PHOENIX-7729: Identify top N slowest parallel scans for metrics reporting (#2337)
This is an automated email from the ASF dual-hosted git repository.
sanjeet pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/phoenix.git
The following commit(s) were added to refs/heads/master by this push:
new 5cd2b7487a PHOENIX-7729: Identify top N slowest parallel scans for metrics reporting (#2337)
5cd2b7487a is described below
commit 5cd2b7487ad40c510d9dc4c0a489bc466999d52a
Author: sanjeet006py <36...@users.noreply.github.com>
AuthorDate: Sun Feb 8 23:58:54 2026 +0530
PHOENIX-7729: Identify top N slowest parallel scans for metrics reporting (#2337)
---
.../apache/phoenix/compile/StatementContext.java | 83 ++
.../apache/phoenix/iterate/ParallelIterators.java | 7 +-
.../phoenix/iterate/ScanningResultIterator.java | 111 +--
.../apache/phoenix/iterate/SerialIterators.java | 3 +-
.../phoenix/iterate/UnionResultIterators.java | 4 +-
.../ParallelPhoenixNullComparingResultSet.java | 8 +
.../phoenix/jdbc/ParallelPhoenixResultSet.java | 8 +
.../org/apache/phoenix/jdbc/PhoenixConnection.java | 17 +
.../phoenix/jdbc/PhoenixMonitoredResultSet.java | 5 +
.../org/apache/phoenix/jdbc/PhoenixResultSet.java | 7 +
.../phoenix/monitoring/GlobalClientMetrics.java | 25 +
.../org/apache/phoenix/monitoring/MetricType.java | 91 +-
.../phoenix/monitoring/OverAllQueryMetrics.java | 14 +-
.../apache/phoenix/monitoring/ReadMetricQueue.java | 15 +-
.../phoenix/monitoring/ScanMetricsHolder.java | 182 ++--
.../monitoring/SlowestScanMetricsQueue.java | 81 ++
.../phoenix/monitoring/TopNTreeMultiMap.java | 196 +++++
.../org/apache/phoenix/query/QueryServices.java | 5 +
.../apache/phoenix/query/QueryServicesOptions.java | 2 +
.../java/org/apache/phoenix/util/JDBCUtil.java | 48 +-
.../org/apache/phoenix/util/PhoenixRuntime.java | 25 +
.../org/apache/phoenix/end2end/ServerPagingIT.java | 10 +-
.../end2end/ServerPagingWithRegionMovesIT.java | 31 +-
.../phoenix/jdbc/ParallelPhoenixConnectionIT.java | 77 ++
.../phoenix/monitoring/SlowestScanMetricsIT.java | 955 +++++++++++++++++++++
.../phoenix/compile/SlowestScanMetricsTest.java | 406 +++++++++
.../phoenix/compat/hbase/CompatScanMetrics.java | 38 +-
.../phoenix/compat/hbase/CompatScanMetrics.java | 38 +-
.../phoenix/compat/hbase/CompatScanMetrics.java | 38 +-
.../phoenix/compat/hbase/CompatScanMetrics.java | 61 +-
30 files changed, 2246 insertions(+), 345 deletions(-)
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/StatementContext.java b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/StatementContext.java
index 85f1ae3a34..1053dac1ea 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/compile/StatementContext.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/compile/StatementContext.java
@@ -19,8 +19,11 @@ package org.apache.phoenix.compile;
import java.sql.SQLException;
import java.text.Format;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Deque;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -33,6 +36,9 @@ import org.apache.phoenix.jdbc.PhoenixStatement;
import org.apache.phoenix.log.QueryLogger;
import org.apache.phoenix.monitoring.OverAllQueryMetrics;
import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.ScanMetricsHolder;
+import org.apache.phoenix.monitoring.SlowestScanMetricsQueue;
+import org.apache.phoenix.monitoring.TopNTreeMultiMap;
import org.apache.phoenix.parse.SelectStatement;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.query.QueryServices;
@@ -52,6 +58,8 @@ import org.apache.phoenix.util.ReadOnlyProps;
import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
import org.apache.phoenix.thirdparty.com.google.common.collect.Sets;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* Class that keeps common state used across processing the various clauses in a top level JDBC
* statement such as SELECT, UPSERT, DELETE, etc.
@@ -91,6 +99,8 @@ public class StatementContext {
private Integer totalSegmentsValue;
private boolean hasRowSizeFunction = false;
private boolean hasRawRowSizeFunction = false;
+ private final SlowestScanMetricsQueue slowestScanMetricsQueue;
+ private final int slowestScanMetricsCount;
public StatementContext(PhoenixStatement statement) {
this(statement, new Scan());
@@ -116,6 +126,8 @@ public class StatementContext {
this.subqueryResults = context.subqueryResults;
this.readMetricsQueue = context.readMetricsQueue;
this.overAllQueryMetrics = context.overAllQueryMetrics;
+ this.slowestScanMetricsQueue = context.slowestScanMetricsQueue;
+ this.slowestScanMetricsCount = context.slowestScanMetricsCount;
this.queryLogger = context.queryLogger;
this.isClientSideUpsertSelect = context.isClientSideUpsertSelect;
this.isUncoveredIndex = context.isUncoveredIndex;
@@ -180,6 +192,10 @@ public class StatementContext {
this.readMetricsQueue = new ReadMetricQueue(isRequestMetricsEnabled, connection.getLogLevel());
this.overAllQueryMetrics =
new OverAllQueryMetrics(isRequestMetricsEnabled, connection.getLogLevel());
+ this.slowestScanMetricsCount = connection.getSlowestScanMetricsCount();
+ this.slowestScanMetricsQueue = slowestScanMetricsCount > 0
+ ? new SlowestScanMetricsQueue()
+ : SlowestScanMetricsQueue.NOOP_SLOWEST_SCAN_METRICS_QUEUE;
this.retryingPersistentCache = Maps.<Long, Boolean> newHashMap();
this.hasFirstValidResult = new AtomicBoolean(false);
this.subStatementContexts = Sets.newHashSet();
@@ -475,4 +491,71 @@ public class StatementContext {
public void setTotalSegmentsValue(Integer totalSegmentsValue) {
this.totalSegmentsValue = totalSegmentsValue;
}
+
+ /**
+ * FOR INTERNAL USE ONLY.
+ * @return the slowest scan metrics queue
+ */
+ public SlowestScanMetricsQueue getSlowestScanMetricsQueue() {
+ return slowestScanMetricsQueue;
+ }
+
+ public List<List<JsonObject>> getTopNSlowestScanMetrics() {
+ if (slowestScanMetricsCount <= 0) {
+ return Collections.emptyList();
+ }
+ TopNTreeMultiMap<Long, List<ScanMetricsHolder>> slowestScanMetricsHolders =
+ TopNTreeMultiMap.getInstance(slowestScanMetricsCount, (s1, s2) -> Long.compare(s2, s1));
+ getSlowestScanMetricsUtil(this, new ArrayDeque<>(), 0, slowestScanMetricsHolders);
+ List<List<JsonObject>> topNSlowestScanMetricsHolders =
+ new ArrayList<>(slowestScanMetricsHolders.size());
+ for (Map.Entry<Long, List<ScanMetricsHolder>> entry : slowestScanMetricsHolders.entries()) {
+ List<JsonObject> scanMetricsHolderJsonObjects = new ArrayList<>(entry.getValue().size());
+ for (ScanMetricsHolder scanMetricsHolder : entry.getValue()) {
+ scanMetricsHolderJsonObjects.add(scanMetricsHolder.toJson());
+ }
+ topNSlowestScanMetricsHolders.add(scanMetricsHolderJsonObjects);
+ }
+ return topNSlowestScanMetricsHolders;
+ }
+
+ private static void getSlowestScanMetricsUtil(StatementContext node,
+ Deque<ScanMetricsHolder> currentScanMetricsHolders, long sumOfMillisBetweenNexts,
+ TopNTreeMultiMap<Long, List<ScanMetricsHolder>> topNSlowestScanMetricsHolders) {
+ Iterator<ScanMetricsHolder> currentScanMetricsHolderIterator =
+ node.getSlowestScanMetricsQueue().getIterator();
+ if (currentScanMetricsHolderIterator.hasNext()) {
+ do {
+ ScanMetricsHolder currentScanMetricsHolder = currentScanMetricsHolderIterator.next();
+ long currentMillisBetweenNexts = currentScanMetricsHolder.getCostOfScan();
+ long newSumOfMillisBetweenNexts = sumOfMillisBetweenNexts + currentMillisBetweenNexts;
+
+ // Add at tail of the dequeue
+ currentScanMetricsHolders.addLast(currentScanMetricsHolder);
+
+ traverseSlowestScanMetricsTree(node, currentScanMetricsHolders, newSumOfMillisBetweenNexts,
+ topNSlowestScanMetricsHolders);
+ currentScanMetricsHolders.removeLast();
+ } while (currentScanMetricsHolderIterator.hasNext());
+ } else {
+ traverseSlowestScanMetricsTree(node, currentScanMetricsHolders, sumOfMillisBetweenNexts,
+ topNSlowestScanMetricsHolders);
+ }
+ }
+
+ private static void traverseSlowestScanMetricsTree(StatementContext node,
+ Deque<ScanMetricsHolder> currentScanMetricsHolders, long sumOfMillisBetweenNexts,
+ TopNTreeMultiMap<Long, List<ScanMetricsHolder>> topNSlowestScanMetricsHolders) {
+ Set<StatementContext> subContexts = node.getSubStatementContexts();
+ if (subContexts == null || subContexts.isEmpty()) {
+ topNSlowestScanMetricsHolders.put(sumOfMillisBetweenNexts,
+ () -> new ArrayList<>(currentScanMetricsHolders));
+ } else {
+ // Process sub-contexts
+ for (StatementContext sub : subContexts) {
+ getSlowestScanMetricsUtil(sub, currentScanMetricsHolders, sumOfMillisBetweenNexts,
+ topNSlowestScanMetricsHolders);
+ }
+ }
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index 3530e1d4ad..4d30d09722 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -31,6 +31,7 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.cache.ServerCacheClient.ServerCache;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.hbase.index.util.ImmutableBytesPtr;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.job.JobManager.JobCallable;
import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.monitoring.ScanMetricsHolder;
@@ -115,8 +116,10 @@ public class ParallelIterators extends BaseResultIterators {
context.getConnection().getQueryServices().getRenewLeaseThresholdMilliSeconds();
for (final ScanLocator scanLocation : scanLocations) {
final Scan scan = scanLocation.getScan();
- final ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics,
- physicalTableName, scan, context.getConnection().getLogLevel());
+ PhoenixConnection connection = context.getConnection();
+ final ScanMetricsHolder scanMetricsHolder =
+ ScanMetricsHolder.getInstance(readMetrics, physicalTableName, scan,
+ connection.getLogLevel(), connection.isScanMetricsByRegionEnabled());
final TaskExecutionMetricsHolder taskMetrics =
new TaskExecutionMetricsHolder(readMetrics, physicalTableName);
final TableResultIterator tableResultItr =
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
index 420d0d570f..28fead5048 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
@@ -17,31 +17,7 @@
*/
package org.apache.phoenix.iterate;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.BYTES_IN_REMOTE_RESULTS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.NOT_SERVING_REGION_EXCEPTION_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.REGIONS_SCANNED_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.REMOTE_RPC_CALLS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.REMOTE_RPC_RETRIES_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.RPC_CALLS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ScanMetrics.RPC_RETRIES_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME;
import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_BYTES_IN_REMOTE_RESULTS;
-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_NOT_SERVING_REGION_EXCEPTION;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_REMOTE_RPC_CALLS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_REMOTE_RPC_RETRIES;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_ROWS_FILTERED;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_ROWS_SCANNED;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_RPC_CALLS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_RPC_RETRIES;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_HBASE_COUNT_SCANNED_REGIONS;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_PAGED_ROWS_COUNTER;
-import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SCAN_BYTES;
import static org.apache.phoenix.util.ScanUtil.isDummy;
import java.io.IOException;
@@ -53,12 +29,11 @@ import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.util.ReflectionUtils;
-import org.apache.phoenix.compat.hbase.CompatScanMetrics;
import org.apache.phoenix.compile.ExplainPlanAttributes.ExplainPlanAttributesBuilder;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
-import org.apache.phoenix.monitoring.CombinableMetric;
+import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.monitoring.GlobalClientMetrics;
import org.apache.phoenix.monitoring.ScanMetricsHolder;
import org.apache.phoenix.query.QueryServices;
@@ -120,88 +95,20 @@ public class ScanningResultIterator implements ResultIterator {
updateMetrics();
}
- private void changeMetric(CombinableMetric metric, Long value) {
- if (value != null) {
- metric.change(value);
- }
- }
-
- private void changeMetric(GlobalClientMetrics metric, Long value) {
- if (value != null) {
- metric.update(value);
- }
- }
-
private void updateMetrics() {
if (scanMetricsEnabled && !scanMetricsUpdated) {
ScanMetrics scanMetrics = scanner.getScanMetrics();
Map<String, Long> scanMetricsMap = scanMetrics.getMetricsMap();
scanMetricsHolder.setScanMetricMap(scanMetricsMap);
-
- changeMetric(scanMetricsHolder.getCountOfRPCcalls(),
- scanMetricsMap.get(RPC_CALLS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRemoteRPCcalls(),
- scanMetricsMap.get(REMOTE_RPC_CALLS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getSumOfMillisSecBetweenNexts(),
- scanMetricsMap.get(MILLIS_BETWEEN_NEXTS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfNSRE(),
- scanMetricsMap.get(NOT_SERVING_REGION_EXCEPTION_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfBytesInResults(),
- scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfBytesInRemoteResults(),
- scanMetricsMap.get(BYTES_IN_REMOTE_RESULTS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRegions(),
- scanMetricsMap.get(REGIONS_SCANNED_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRPCRetries(),
- scanMetricsMap.get(RPC_RETRIES_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRemoteRPCRetries(),
- scanMetricsMap.get(REMOTE_RPC_RETRIES_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRowsScanned(),
- scanMetricsMap.get(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRowsFiltered(),
- scanMetricsMap.get(COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfBytesScanned(),
- scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
- changeMetric(scanMetricsHolder.getCountOfRowsPaged(), dummyRowCounter);
- changeMetric(scanMetricsHolder.getFsReadTime(),
- CompatScanMetrics.getFsReadTime(scanMetricsMap));
- changeMetric(scanMetricsHolder.getCountOfBytesReadFromFS(),
- CompatScanMetrics.getBytesReadFromFs(scanMetricsMap));
- changeMetric(scanMetricsHolder.getCountOfBytesReadFromMemstore(),
- CompatScanMetrics.getBytesReadFromMemstore(scanMetricsMap));
- changeMetric(scanMetricsHolder.getCountOfBytesReadFromBlockcache(),
- CompatScanMetrics.getBytesReadFromBlockCache(scanMetricsMap));
- changeMetric(scanMetricsHolder.getCountOfBlockReadOps(),
- CompatScanMetrics.getBlockReadOpsCount(scanMetricsMap));
- changeMetric(scanMetricsHolder.getRpcScanProcessingTime(),
- CompatScanMetrics.getRpcScanProcessingTime(scanMetricsMap));
- changeMetric(scanMetricsHolder.getRpcScanQueueWaitTime(),
- CompatScanMetrics.getRpcScanQueueWaitTime(scanMetricsMap));
-
- changeMetric(GLOBAL_SCAN_BYTES, scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_RPC_CALLS, scanMetricsMap.get(RPC_CALLS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_REMOTE_RPC_CALLS,
- scanMetricsMap.get(REMOTE_RPC_CALLS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_MILLS_BETWEEN_NEXTS,
- scanMetricsMap.get(MILLIS_BETWEEN_NEXTS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_NOT_SERVING_REGION_EXCEPTION,
- scanMetricsMap.get(NOT_SERVING_REGION_EXCEPTION_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_BYTES_REGION_SERVER_RESULTS,
- scanMetricsMap.get(BYTES_IN_RESULTS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_BYTES_IN_REMOTE_RESULTS,
- scanMetricsMap.get(BYTES_IN_REMOTE_RESULTS_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_SCANNED_REGIONS,
- scanMetricsMap.get(REGIONS_SCANNED_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_RPC_RETRIES, scanMetricsMap.get(RPC_RETRIES_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_REMOTE_RPC_RETRIES,
- scanMetricsMap.get(REMOTE_RPC_RETRIES_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_ROWS_SCANNED,
- scanMetricsMap.get(COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME));
- changeMetric(GLOBAL_HBASE_COUNT_ROWS_FILTERED,
- scanMetricsMap.get(COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME));
-
- changeMetric(GLOBAL_PAGED_ROWS_COUNTER, dummyRowCounter);
+ scanMetricsHolder.populateMetrics(dummyRowCounter);
+ GlobalClientMetrics.populateMetrics(scanMetricsMap, dummyRowCounter);
+ PhoenixConnection connection = context.getConnection();
+ int slowestScanMetricsCount = connection.getSlowestScanMetricsCount();
+ if (slowestScanMetricsCount > 0) {
+ scanMetricsHolder.setScanMetricsByRegion(scanMetrics.collectMetricsByRegion());
+ context.getSlowestScanMetricsQueue().add(scanMetricsHolder);
+ }
scanMetricsUpdated = true;
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/SerialIterators.java b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
index f7e5c1b1a6..8b23073396 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
@@ -185,8 +185,9 @@ public class SerialIterators extends BaseResultIterators {
currentScan.setAttribute(BaseScannerRegionObserverConstants.SCAN_OFFSET,
PInteger.INSTANCE.toBytes(remainingOffset));
}
+ PhoenixConnection connection = context.getConnection();
ScanMetricsHolder scanMetricsHolder = ScanMetricsHolder.getInstance(readMetrics, tableName,
- currentScan, context.getConnection().getLogLevel());
+ currentScan, connection.getLogLevel(), connection.isScanMetricsByRegionEnabled());
TableResultIterator itr = new TableResultIterator(mutationState, currentScan,
scanMetricsHolder, renewLeaseThreshold, plan, scanGrouper, caches, maxQueryEndTime);
PeekingResultIterator peekingItr =
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
index d29a087a3c..e5b981e7a6 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
@@ -54,6 +54,7 @@ public class UnionResultIterators implements ResultIterators {
readMetricsList = Lists.newArrayListWithCapacity(nPlans);
overAllQueryMetricsList = Lists.newArrayListWithCapacity(nPlans);
for (QueryPlan plan : plans) {
+ parentStmtCtx.addSubStatementContext(plan.getContext());
readMetricsList.add(plan.getContext().getReadMetricsQueue());
overAllQueryMetricsList.add(plan.getContext().getOverallQueryMetrics());
iterators.add(LookAheadResultIterator.wrap(plan.iterator()));
@@ -102,11 +103,12 @@ public class UnionResultIterators implements ResultIterators {
private void setMetricsInParentContext() {
ReadMetricQueue parentCtxReadMetrics = parentStmtCtx.getReadMetricsQueue();
for (ReadMetricQueue readMetrics : readMetricsList) {
- parentCtxReadMetrics.combineReadMetrics(readMetrics);
+ parentCtxReadMetrics.combineReadMetrics(readMetrics, true);
}
OverAllQueryMetrics parentCtxQueryMetrics = parentStmtCtx.getOverallQueryMetrics();
for (OverAllQueryMetrics metric : overAllQueryMetricsList) {
parentCtxQueryMetrics.combine(metric);
+ metric.reset();
}
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSet.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSet.java
index 8f50817de0..8ae6889b2e 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSet.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixNullComparingResultSet.java
@@ -33,11 +33,14 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.util.JDBCUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* ResultSet suitable for truly immutable use cases that do not delete data and do not query data
* the does not exist. Returns a non-nullvalue when possible. Checks result from both the underlying
@@ -240,6 +243,11 @@ public class ParallelPhoenixNullComparingResultSet extends DelegateResultSet
return metrics;
}
+ @Override
+ public List<List<JsonObject>> getTopNSlowestScanMetrics() {
+ return JDBCUtil.getTopNSlowestScanMetrics(rs);
+ }
+
@Override
public Map<MetricType, Long> getOverAllRequestReadMetrics() {
Map<MetricType, Long> metrics;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixResultSet.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixResultSet.java
index 136222bd30..f177947793 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixResultSet.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/ParallelPhoenixResultSet.java
@@ -29,12 +29,15 @@ import java.util.concurrent.CompletionException;
import java.util.function.Function;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.util.JDBCUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.phoenix.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* ParallelPhoenixResultSet that provides the standard wait until at least one cluster completes. We
* close the idle result set to release server resources asynchronously.
@@ -133,6 +136,11 @@ public class ParallelPhoenixResultSet extends DelegateResultSet
return metrics;
}
+ @Override
+ public List<List<JsonObject>> getTopNSlowestScanMetrics() {
+ return JDBCUtil.getTopNSlowestScanMetrics(rs);
+ }
+
@Override
public Map<MetricType, Long> getOverAllRequestReadMetrics() {
Map<MetricType, Long> metrics;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index 7f2612fa7f..44ab9c99ae 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -187,6 +187,8 @@ public class PhoenixConnection
private String sourceOfOperation;
private volatile SQLException reasonForClose;
private static final String[] CONNECTION_PROPERTIES;
+ private final int slowestScanMetricsCount;
+ private final boolean isScanMetricsByRegionEnabled;
private final ConcurrentLinkedQueue<PhoenixConnection> childConnections =
new ConcurrentLinkedQueue<>();
@@ -357,6 +359,13 @@ public class PhoenixConnection
.get(QueryServices.AUDIT_LOG_LEVEL, QueryServicesOptions.DEFAULT_AUDIT_LOGGING_LEVEL));
this.isRequestLevelMetricsEnabled =
JDBCUtil.isCollectingRequestLevelMetricsEnabled(url, info, this.services.getProps());
+ int slowestScanMetricsCount =
+ JDBCUtil.getSlowestScanMetricsCount(url, this.info, this.services.getProps());
+ this.slowestScanMetricsCount = this.isRequestLevelMetricsEnabled ? slowestScanMetricsCount : 0;
+ boolean isScanMetricsByRegionEnabled =
+ JDBCUtil.isScanMetricsByRegionEnabled(url, this.info, this.services.getProps());
+ this.isScanMetricsByRegionEnabled =
+ this.slowestScanMetricsCount > 0 ? isScanMetricsByRegionEnabled : false;
this.mutationState = mutationState == null
? newMutationState(maxSize, maxSizeBytes)
: new MutationState(mutationState, this);
@@ -1489,4 +1498,12 @@ public class PhoenixConnection
public void setActivityLogger(ConnectionActivityLogger connectionActivityLogger) {
this.connectionActivityLogger = connectionActivityLogger;
}
+
+ public int getSlowestScanMetricsCount() {
+ return slowestScanMetricsCount;
+ }
+
+ public boolean isScanMetricsByRegionEnabled() {
+ return isScanMetricsByRegionEnabled;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixMonitoredResultSet.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixMonitoredResultSet.java
index f244ce3a8e..a8ee805340 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixMonitoredResultSet.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixMonitoredResultSet.java
@@ -18,13 +18,18 @@
package org.apache.phoenix.jdbc;
import java.sql.ResultSet;
+import java.util.List;
import java.util.Map;
import org.apache.phoenix.monitoring.MetricType;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
public interface PhoenixMonitoredResultSet extends ResultSet {
Map<String, Map<MetricType, Long>> getReadMetrics();
+ List<List<JsonObject>> getTopNSlowestScanMetrics();
+
Map<MetricType, Long> getOverAllRequestReadMetrics();
void resetMetrics();
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
index 5b9b8f0af8..6b8ae717de 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
@@ -106,6 +106,8 @@ import org.apache.phoenix.thirdparty.com.google.common.base.Strings;
import org.apache.phoenix.thirdparty.com.google.common.base.Throwables;
import org.apache.phoenix.thirdparty.com.google.common.primitives.Bytes;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* JDBC ResultSet implementation of Phoenix. Currently only the following data types are supported:
* - String - Date - Time - Timestamp - BigDecimal - Double - Float - Int - Short - Long - Binary -
@@ -1496,6 +1498,11 @@ public class PhoenixResultSet implements PhoenixMonitoredResultSet, SQLCloseable
return one.aggregate();
}
+ @Override
+ public List<List<JsonObject>> getTopNSlowestScanMetrics() {
+ return context.getTopNSlowestScanMetrics();
+ }
+
@Override
public Map<MetricType, Long> getOverAllRequestReadMetrics() {
return overAllQueryMetrics.publish();
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
index 7275684221..89f04f8827 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/GlobalClientMetrics.java
@@ -84,6 +84,8 @@ import static org.apache.phoenix.monitoring.MetricType.TASK_REJECTED_COUNTER;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.metrics.Gauge;
import org.apache.hadoop.hbase.metrics.MetricRegistries;
import org.apache.hadoop.hbase.metrics.MetricRegistry;
@@ -174,6 +176,7 @@ public enum GlobalClientMetrics {
QueryServicesOptions.withDefaults().isGlobalMetricsEnabled();
private MetricType metricType;
private GlobalMetric metric;
+ private static List<GlobalClientMetrics> hbaseScanMetrics;
static {
try {
@@ -182,6 +185,13 @@ public enum GlobalClientMetrics {
MetricRegistry metricRegistry = createMetricRegistry();
registerPhoenixMetricsToRegistry(metricRegistry);
GlobalMetricRegistriesAdapter.getInstance().registerMetricRegistry(metricRegistry);
+ hbaseScanMetrics = new ArrayList<>();
+ for (GlobalClientMetrics globalMetric : GlobalClientMetrics.values()) {
+ MetricType metricType = globalMetric.getMetricType();
+ if (metricType != null && !StringUtils.isEmpty(metricType.getHBaseMetricName())) {
+ hbaseScanMetrics.add(globalMetric);
+ }
+ }
}
} catch (Throwable t) {
LOGGER.error("Failed to initialize GlobalClientMetrics", t);
@@ -271,4 +281,19 @@ public enum GlobalClientMetrics {
return isGlobalMetricsEnabled;
}
+ private static void changeMetric(GlobalClientMetrics metric, Long value) {
+ if (value != null) {
+ metric.update(value);
+ }
+ }
+
+ public static void populateMetrics(Map<String, Long> scanMetricsMap, Long dummyRowCounter) {
+ // PAGED_ROWS_COUNTER is not a HBase scan metric but is captured by Phoenix client
+ // for each scan, so we need to handle it separately.
+ changeMetric(GLOBAL_PAGED_ROWS_COUNTER, dummyRowCounter);
+ for (GlobalClientMetrics globalMetric : hbaseScanMetrics) {
+ String hbaseMetricName = globalMetric.getMetricType().getHBaseMetricName();
+ changeMetric(globalMetric, scanMetricsMap.get(hbaseMetricName));
+ }
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
index 85cdb2d3bf..ff80705c0d 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java
@@ -17,6 +17,13 @@
*/
package org.apache.phoenix.monitoring;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
+import org.apache.phoenix.compat.hbase.CompatScanMetrics;
import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.schema.types.PDataType;
import org.apache.phoenix.schema.types.PLong;
@@ -175,7 +182,8 @@ public enum MetricType {
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),
+ SCAN_BYTES("sb", "Number of bytes read by scans", LogLevel.OFF, PLong.INSTANCE,
+ ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME),
SELECT_SQL_COUNTER("sc", "Counter for number of sql queries", LogLevel.OFF, PLong.INSTANCE),
// task metrics
TASK_QUEUE_WAIT_TIME("tw",
@@ -246,33 +254,43 @@ public enum MetricType {
"Number of StaleMetadataCacheException encountered.", LogLevel.DEBUG, PLong.INSTANCE),
// hbase metrics
- 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_RPC_CALLS("rp", "Number of RPC calls", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.RPC_CALLS_METRIC_NAME),
+ COUNT_REMOTE_RPC_CALLS("rr", "Number of remote RPC calls", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.REMOTE_RPC_CALLS_METRIC_NAME),
COUNT_MILLS_BETWEEN_NEXTS("n", "Sum of milliseconds between sequential next calls",
- LogLevel.DEBUG, PLong.INSTANCE),
+ LogLevel.DEBUG, PLong.INSTANCE, ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME),
COUNT_NOT_SERVING_REGION_EXCEPTION("nsr", "Number of NotServingRegionException caught",
- LogLevel.DEBUG, PLong.INSTANCE),
+ LogLevel.DEBUG, PLong.INSTANCE, ScanMetrics.NOT_SERVING_REGION_EXCEPTION_METRIC_NAME),
COUNT_BYTES_REGION_SERVER_RESULTS("rs", "Number of bytes in Result objects from region servers",
- LogLevel.DEBUG, PLong.INSTANCE),
+ LogLevel.DEBUG, PLong.INSTANCE, ScanMetrics.BYTES_IN_RESULTS_METRIC_NAME),
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),
- FS_READ_TIME("frt", "Time spent in filesystem read", LogLevel.DEBUG, PLong.INSTANCE),
- BYTES_READ_FROM_FS("brff", "Number of bytes read from filesystem", LogLevel.DEBUG,
- PLong.INSTANCE),
+ "Number of bytes in Result objects from remote region servers", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.BYTES_IN_REMOTE_RESULTS_METRIC_NAME),
+ COUNT_SCANNED_REGIONS("rg", "Number of regions scanned", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.REGIONS_SCANNED_METRIC_NAME),
+ COUNT_RPC_RETRIES("rpr", "Number of RPC retries", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.RPC_RETRIES_METRIC_NAME),
+ COUNT_REMOTE_RPC_RETRIES("rrr", "Number of remote RPC retries", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.REMOTE_RPC_RETRIES_METRIC_NAME),
+ COUNT_ROWS_SCANNED("ws", "Number of rows scanned", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.COUNT_OF_ROWS_SCANNED_KEY_METRIC_NAME),
+ COUNT_ROWS_FILTERED("wf", "Number of rows filtered", LogLevel.DEBUG, PLong.INSTANCE,
+ ScanMetrics.COUNT_OF_ROWS_FILTERED_KEY_METRIC_NAME),
+ FS_READ_TIME("frt", "Time spent in filesystem read", LogLevel.DEBUG, PLong.INSTANCE,
+ CompatScanMetrics.FS_READ_TIME_METRIC_NAME),
+ BYTES_READ_FROM_FS("brff", "Number of bytes read from filesystem", LogLevel.DEBUG, PLong.INSTANCE,
+ CompatScanMetrics.BYTES_READ_FROM_FS_METRIC_NAME),
BYTES_READ_FROM_MEMSTORE("brfm", "Number of bytes read from memstore", LogLevel.DEBUG,
- PLong.INSTANCE),
+ PLong.INSTANCE, CompatScanMetrics.BYTES_READ_FROM_MEMSTORE_METRIC_NAME),
BYTES_READ_FROM_BLOCKCACHE("brfc", "Number of bytes read from blockcache", LogLevel.DEBUG,
- PLong.INSTANCE),
- BLOCK_READ_OPS_COUNT("broc", "Number of block read operations", LogLevel.DEBUG, PLong.INSTANCE),
+ PLong.INSTANCE, CompatScanMetrics.BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME),
+ BLOCK_READ_OPS_COUNT("broc", "Number of block read operations", LogLevel.DEBUG, PLong.INSTANCE,
+ CompatScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME),
RPC_SCAN_PROCESSING_TIME("spt", "Time spent in RPC scan processing", LogLevel.DEBUG,
- PLong.INSTANCE),
+ PLong.INSTANCE, CompatScanMetrics.RPC_SCAN_PROCESSING_TIME_METRIC_NAME),
RPC_SCAN_QUEUE_WAIT_TIME("sqwt", "Time spent in RPC scan queue wait", LogLevel.DEBUG,
- PLong.INSTANCE),
+ PLong.INSTANCE, CompatScanMetrics.RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME),
COUNTER_METADATA_INCONSISTENCY("mi", "Number of times the metadata inconsistencies ",
LogLevel.DEBUG, PLong.INSTANCE),
NUM_SYSTEM_TABLE_RPC_SUCCESS("nstrs", "Number of successful system table RPC calls",
@@ -341,12 +359,32 @@ public enum MetricType {
private final String shortName;
private LogLevel logLevel;
private PDataType dataType;
+ private final String hbaseMetricName;
+ // SCAN_BYTES and COUNT_BYTES_REGION_SERVER_RESULTS are mapped to same HBase metric but we will
+ // only keeping mapping for one of them.
+ private static Map<String, MetricType> hbaseToPhoenixMetricMap = new HashMap<>();
+ private static List<MetricType> hbaseScanMetrics = new ArrayList<>();
+
+ static {
+ for (MetricType metric : MetricType.values()) {
+ if (!StringUtils.isEmpty(metric.hbaseMetricName)) {
+ hbaseToPhoenixMetricMap.put(metric.hbaseMetricName, metric);
+ hbaseScanMetrics.add(metric);
+ }
+ }
+ }
+
+ MetricType(String shortName, String description, LogLevel logLevel, PDataType dataType) {
+ this(shortName, description, logLevel, dataType, StringUtils.EMPTY);
+ }
- private MetricType(String shortName, String description, LogLevel logLevel, PDataType dataType) {
+ MetricType(String shortName, String description, LogLevel logLevel, PDataType dataType,
+ String hbaseMetricName) {
this.shortName = shortName;
this.description = description;
this.logLevel = logLevel;
this.dataType = dataType;
+ this.hbaseMetricName = hbaseMetricName;
}
public String description() {
@@ -369,6 +407,10 @@ public enum MetricType {
return name();
}
+ public String getHBaseMetricName() {
+ return hbaseMetricName;
+ }
+
public boolean isLoggingEnabled(LogLevel connectionLogLevel) {
return logLevel() != LogLevel.OFF && (logLevel().ordinal() <= connectionLogLevel.ordinal());
}
@@ -386,4 +428,11 @@ public enum MetricType {
return buffer.toString();
}
+ public static MetricType getMetricType(String hbaseMetricName) {
+ return hbaseToPhoenixMetricMap.get(hbaseMetricName);
+ }
+
+ public static List<MetricType> getHbaseScanMetrics() {
+ return hbaseScanMetrics;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
index 101f01674a..173b9af51e 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/OverAllQueryMetrics.java
@@ -127,15 +127,15 @@ public class OverAllQueryMetrics {
}
public void setQueryCompilerTimeMS(long time) {
- queryCompilerTimeMS.set(time);
+ queryCompilerTimeMS.change(time);
}
public void setQueryOptimizerTimeMS(long time) {
- queryOptimizerTimeMS.set(time);
+ queryOptimizerTimeMS.change(time);
}
public void setQueryResultItrTimeMS(long time) {
- queryResultItrSetTimeMS.set(time);
+ queryResultItrSetTimeMS.change(time);
}
public void startQuery() {
@@ -175,7 +175,7 @@ public class OverAllQueryMetrics {
}
public void setQueryParsingTimeMS(long time) {
- queryParsingTimeMS.set(time);
+ queryParsingTimeMS.change(time);
}
public Map<MetricType, Long> publish() {
@@ -231,9 +231,9 @@ public class OverAllQueryMetrics {
numParallelScans.combine(metric.numParallelScans);
wallClockTimeMS.combine(metric.wallClockTimeMS);
resultSetTimeMS.combine(metric.resultSetTimeMS);
- queryCompilerTimeMS.combine(queryCompilerTimeMS);
- queryOptimizerTimeMS.combine(queryOptimizerTimeMS);
- queryResultItrSetTimeMS.combine(queryResultItrSetTimeMS);
+ queryCompilerTimeMS.combine(metric.queryCompilerTimeMS);
+ queryOptimizerTimeMS.combine(metric.queryOptimizerTimeMS);
+ queryResultItrSetTimeMS.combine(metric.queryResultItrSetTimeMS);
queryParsingTimeMS.combine(metric.queryParsingTimeMS);
return this;
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
index b6692bd624..5ae4edb6de 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ReadMetricQueue.java
@@ -112,22 +112,25 @@ public class ReadMetricQueue {
return combinedMetric;
}
+ public ReadMetricQueue combineReadMetrics(ReadMetricQueue other) {
+ return combineReadMetrics(other, false);
+ }
+
/**
* Combine the metrics. This method should only be called in a single threaded manner when the two
* metric holders are not getting modified.
*/
- public ReadMetricQueue combineReadMetrics(ReadMetricQueue other) {
+ public ReadMetricQueue combineReadMetrics(ReadMetricQueue other, boolean reset) {
ConcurrentMap<MetricKey, Queue<CombinableMetric>> otherMetricsMap = other.metricsMap;
for (Entry<MetricKey, Queue<CombinableMetric>> entry : otherMetricsMap.entrySet()) {
MetricKey key = entry.getKey();
Queue<CombinableMetric> otherQueue = entry.getValue();
- CombinableMetric combinedMetric = null;
+ CombinableMetric combinedMetric = new CombinableMetricImpl(key.type);
// combine the metrics corresponding to this metric key before putting it in the queue.
for (CombinableMetric m : otherQueue) {
- if (combinedMetric == null) {
- combinedMetric = m;
- } else {
- combinedMetric.combine(m);
+ combinedMetric.combine(m);
+ if (reset) {
+ m.reset();
}
}
if (combinedMetric != null) {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
index 0c827602af..0630e1456c 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/ScanMetricsHolder.java
@@ -39,115 +39,100 @@ import static org.apache.phoenix.monitoring.MetricType.RPC_SCAN_QUEUE_WAIT_TIME;
import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
import java.io.IOException;
+import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
+import org.apache.hadoop.hbase.client.metrics.ScanMetricsRegionInfo;
import org.apache.hadoop.hbase.util.JsonMapper;
import org.apache.phoenix.log.LogLevel;
+import org.apache.hbase.thirdparty.com.google.gson.JsonArray;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
public class ScanMetricsHolder {
- private final CombinableMetric countOfRPCcalls;
- private final CombinableMetric countOfRemoteRPCcalls;
- private final CombinableMetric sumOfMillisSecBetweenNexts;
- private final CombinableMetric countOfNSRE;
- private final CombinableMetric countOfBytesInResults;
- private final CombinableMetric countOfBytesInRemoteResults;
- private final CombinableMetric countOfRegions;
- private final CombinableMetric countOfRPCRetries;
- private final CombinableMetric countOfRemoteRPCRetries;
- private final CombinableMetric countOfRowsScanned;
- private final CombinableMetric countOfRowsFiltered;
- private final CombinableMetric countOfBytesScanned;
- private final CombinableMetric countOfRowsPaged;
- private final CombinableMetric fsReadTime;
- private final CombinableMetric countOfBytesReadFromFS;
- private final CombinableMetric countOfBytesReadFromMemstore;
- private final CombinableMetric countOfBytesReadFromBlockcache;
- private final CombinableMetric countOfBlockReadOps;
- private final CombinableMetric rpcScanProcessingTime;
- private final CombinableMetric rpcScanQueueWaitTime;
private Map<String, Long> scanMetricMap;
+ private Map<ScanMetricsRegionInfo, Map<String, Long>> scanMetricsByRegion;
private Object scan;
+ private final String tableName;
+ private final Map<MetricType, CombinableMetric> metricTypeToMetric;
private static final ScanMetricsHolder NO_OP_INSTANCE =
- new ScanMetricsHolder(new ReadMetricQueue(false, LogLevel.OFF), "", null);
+ new ScanMetricsHolder(new ReadMetricQueue(false, LogLevel.OFF), "", null, false);
public static ScanMetricsHolder getInstance(ReadMetricQueue readMetrics, String tableName,
Scan scan, LogLevel connectionLogLevel) {
+ return ScanMetricsHolder.getInstance(readMetrics, tableName, scan, connectionLogLevel, false);
+ }
+
+ public static ScanMetricsHolder getInstance(ReadMetricQueue readMetrics, String tableName,
+ Scan scan, LogLevel connectionLogLevel, boolean isScanMetricsByRegionEnabled) {
if (connectionLogLevel == LogLevel.OFF && !readMetrics.isRequestMetricsEnabled()) {
return NO_OP_INSTANCE;
}
- scan.setScanMetricsEnabled(true);
- return new ScanMetricsHolder(readMetrics, tableName, scan);
+ return new ScanMetricsHolder(readMetrics, tableName, scan, isScanMetricsByRegionEnabled);
}
- private ScanMetricsHolder(ReadMetricQueue readMetrics, String tableName, Scan scan) {
+ private ScanMetricsHolder(ReadMetricQueue readMetrics, String tableName, Scan scan,
+ boolean isScanMetricsByRegionEnabled) {
readMetrics.addScanHolder(this);
+ this.tableName = tableName;
this.scan = scan;
- countOfRPCcalls = readMetrics.allotMetric(COUNT_RPC_CALLS, tableName);
- countOfRemoteRPCcalls = readMetrics.allotMetric(COUNT_REMOTE_RPC_CALLS, tableName);
- sumOfMillisSecBetweenNexts = readMetrics.allotMetric(COUNT_MILLS_BETWEEN_NEXTS, tableName);
- countOfNSRE = readMetrics.allotMetric(COUNT_NOT_SERVING_REGION_EXCEPTION, tableName);
- countOfBytesInResults = readMetrics.allotMetric(COUNT_BYTES_REGION_SERVER_RESULTS, tableName);
- countOfBytesInRemoteResults = readMetrics.allotMetric(COUNT_BYTES_IN_REMOTE_RESULTS, tableName);
- countOfRegions = readMetrics.allotMetric(COUNT_SCANNED_REGIONS, tableName);
- countOfRPCRetries = readMetrics.allotMetric(COUNT_RPC_RETRIES, tableName);
- countOfRemoteRPCRetries = readMetrics.allotMetric(COUNT_REMOTE_RPC_RETRIES, tableName);
- countOfRowsScanned = readMetrics.allotMetric(COUNT_ROWS_SCANNED, tableName);
- countOfRowsFiltered = readMetrics.allotMetric(COUNT_ROWS_FILTERED, tableName);
- countOfBytesScanned = readMetrics.allotMetric(SCAN_BYTES, tableName);
- countOfRowsPaged = readMetrics.allotMetric(PAGED_ROWS_COUNTER, tableName);
- fsReadTime = readMetrics.allotMetric(FS_READ_TIME, tableName);
- countOfBytesReadFromFS = readMetrics.allotMetric(BYTES_READ_FROM_FS, tableName);
- countOfBytesReadFromMemstore = readMetrics.allotMetric(BYTES_READ_FROM_MEMSTORE, tableName);
- countOfBytesReadFromBlockcache = readMetrics.allotMetric(BYTES_READ_FROM_BLOCKCACHE, tableName);
- countOfBlockReadOps = readMetrics.allotMetric(BLOCK_READ_OPS_COUNT, tableName);
- rpcScanProcessingTime = readMetrics.allotMetric(RPC_SCAN_PROCESSING_TIME, tableName);
- rpcScanQueueWaitTime = readMetrics.allotMetric(RPC_SCAN_QUEUE_WAIT_TIME, tableName);
+ if (scan != null) {
+ scan.setScanMetricsEnabled(true);
+ scan.setEnableScanMetricsByRegion(isScanMetricsByRegionEnabled);
+ }
+ metricTypeToMetric = new HashMap<>();
+ metricTypeToMetric.put(PAGED_ROWS_COUNTER,
+ readMetrics.allotMetric(PAGED_ROWS_COUNTER, tableName));
+ for (MetricType metric : MetricType.getHbaseScanMetrics()) {
+ metricTypeToMetric.put(metric, readMetrics.allotMetric(metric, tableName));
+ }
}
public CombinableMetric getCountOfRemoteRPCcalls() {
- return countOfRemoteRPCcalls;
+ return metricTypeToMetric.get(COUNT_REMOTE_RPC_CALLS);
}
public CombinableMetric getSumOfMillisSecBetweenNexts() {
- return sumOfMillisSecBetweenNexts;
+ return metricTypeToMetric.get(COUNT_MILLS_BETWEEN_NEXTS);
}
public CombinableMetric getCountOfNSRE() {
- return countOfNSRE;
+ return metricTypeToMetric.get(COUNT_NOT_SERVING_REGION_EXCEPTION);
}
public CombinableMetric getCountOfBytesInRemoteResults() {
- return countOfBytesInRemoteResults;
+ return metricTypeToMetric.get(COUNT_BYTES_IN_REMOTE_RESULTS);
}
public CombinableMetric getCountOfRegions() {
- return countOfRegions;
+ return metricTypeToMetric.get(COUNT_SCANNED_REGIONS);
}
public CombinableMetric getCountOfRPCRetries() {
- return countOfRPCRetries;
+ return metricTypeToMetric.get(COUNT_RPC_RETRIES);
}
public CombinableMetric getCountOfRemoteRPCRetries() {
- return countOfRemoteRPCRetries;
+ return metricTypeToMetric.get(COUNT_REMOTE_RPC_RETRIES);
}
public CombinableMetric getCountOfRowsFiltered() {
- return countOfRowsFiltered;
+ return metricTypeToMetric.get(COUNT_ROWS_FILTERED);
}
public CombinableMetric getCountOfRPCcalls() {
- return countOfRPCcalls;
+ return metricTypeToMetric.get(COUNT_RPC_CALLS);
}
public CombinableMetric getCountOfBytesInResults() {
- return countOfBytesInResults;
+ return metricTypeToMetric.get(COUNT_BYTES_REGION_SERVER_RESULTS);
}
public CombinableMetric getCountOfRowsScanned() {
- return countOfRowsScanned;
+ return metricTypeToMetric.get(COUNT_ROWS_SCANNED);
}
public Map<String, Long> getScanMetricMap() {
@@ -155,45 +140,114 @@ public class ScanMetricsHolder {
}
public CombinableMetric getCountOfBytesScanned() {
- return countOfBytesScanned;
+ return metricTypeToMetric.get(SCAN_BYTES);
}
public CombinableMetric getCountOfRowsPaged() {
- return countOfRowsPaged;
+ return metricTypeToMetric.get(PAGED_ROWS_COUNTER);
}
public CombinableMetric getFsReadTime() {
- return fsReadTime;
+ return metricTypeToMetric.get(FS_READ_TIME);
}
public CombinableMetric getCountOfBytesReadFromFS() {
- return countOfBytesReadFromFS;
+ return metricTypeToMetric.get(BYTES_READ_FROM_FS);
}
public CombinableMetric getCountOfBytesReadFromMemstore() {
- return countOfBytesReadFromMemstore;
+ return metricTypeToMetric.get(BYTES_READ_FROM_MEMSTORE);
}
public CombinableMetric getCountOfBytesReadFromBlockcache() {
- return countOfBytesReadFromBlockcache;
+ return metricTypeToMetric.get(BYTES_READ_FROM_BLOCKCACHE);
}
public CombinableMetric getCountOfBlockReadOps() {
- return countOfBlockReadOps;
+ return metricTypeToMetric.get(BLOCK_READ_OPS_COUNT);
}
public CombinableMetric getRpcScanProcessingTime() {
- return rpcScanProcessingTime;
+ return metricTypeToMetric.get(RPC_SCAN_PROCESSING_TIME);
}
public CombinableMetric getRpcScanQueueWaitTime() {
- return rpcScanQueueWaitTime;
+ return metricTypeToMetric.get(RPC_SCAN_QUEUE_WAIT_TIME);
}
public void setScanMetricMap(Map<String, Long> scanMetricMap) {
this.scanMetricMap = scanMetricMap;
}
+ public void
+ setScanMetricsByRegion(Map<ScanMetricsRegionInfo, Map<String, Long>> scanMetricsByRegion) {
+ this.scanMetricsByRegion = scanMetricsByRegion;
+ }
+
+ public void populateMetrics(Long dummyRowCounter) {
+ for (Map.Entry<MetricType, CombinableMetric> entry : metricTypeToMetric.entrySet()) {
+ String hbaseMetricName = entry.getKey().getHBaseMetricName();
+ CombinableMetric metric = entry.getValue();
+ // PAGED_ROWS_COUNTER is not a HBase scan metric but is captured by Phoenix client
+ // for each scan, so we need to handle it separately.
+ if (metric.getMetricType() == MetricType.PAGED_ROWS_COUNTER) {
+ changeMetric(metric, dummyRowCounter);
+ } else if (hbaseMetricName != null && !hbaseMetricName.isEmpty()) {
+ changeMetric(metric, scanMetricMap.get(hbaseMetricName));
+ }
+ }
+ }
+
+ private void changeMetric(CombinableMetric metric, Long value) {
+ if (value != null) {
+ metric.change(value);
+ }
+ }
+
+ public Long getCostOfScan() {
+ if (scanMetricMap == null || scanMetricMap.isEmpty()) {
+ return Long.MAX_VALUE;
+ }
+ return scanMetricMap.get(ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME);
+ }
+
+ public JsonObject toJson() {
+ JsonObject tableJson = new JsonObject();
+ if (scanMetricMap == null || scanMetricMap.isEmpty()) {
+ return tableJson;
+ }
+ if (scanMetricsByRegion != null && !scanMetricsByRegion.isEmpty()) {
+ tableJson.addProperty("table", tableName);
+ JsonArray regionMetrics = new JsonArray();
+ for (Map.Entry<ScanMetricsRegionInfo, Map<String, Long>> entry : scanMetricsByRegion
+ .entrySet()) {
+ JsonObject regionJson = new JsonObject();
+ ScanMetricsRegionInfo regionInfo = entry.getKey();
+ regionJson.addProperty("region", regionInfo.getEncodedRegionName());
+ regionJson.addProperty("server", regionInfo.getServerName().toString());
+ for (Map.Entry<String, Long> scanMetricEntry : entry.getValue().entrySet()) {
+ MetricType metricType = MetricType.getMetricType(scanMetricEntry.getKey());
+ if (metricType == null) {
+ continue;
+ }
+ regionJson.addProperty(metricType.shortName(), scanMetricEntry.getValue());
+ }
+ regionMetrics.add(regionJson);
+ }
+ tableJson.add("regions", regionMetrics);
+ } else {
+ tableJson.addProperty("table", tableName);
+ for (Map.Entry<String, Long> scanMetricEntry : scanMetricMap.entrySet()) {
+ MetricType metricType = MetricType.getMetricType(scanMetricEntry.getKey());
+ if (metricType == null) {
+ continue;
+ }
+ tableJson.addProperty(metricType.shortName(), scanMetricEntry.getValue());
+ }
+ }
+ return tableJson;
+ }
+
@Override
public String toString() {
try {
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/SlowestScanMetricsQueue.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/SlowestScanMetricsQueue.java
new file mode 100644
index 0000000000..dfd384b7d6
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/SlowestScanMetricsQueue.java
@@ -0,0 +1,81 @@
+/*
+ * 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 java.util.Collections;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.concurrent.ConcurrentLinkedDeque;
+
+/**
+ * Maintains a queue of {@link ScanMetricsHolder} instances, one instance per HBase scan created by
+ * Phoenix client. The class exposes an iterator which returns ScanMetricsHolder instances in
+ * reverse order of insertion as the last inserted instance is more likely to correspond to the
+ * slowest HBase scan.
+ * <p>
+ * The insertion of ScanMetricsHolder instances to the queue is thread-safe, so multiple parallel
+ * scans can concurrently add their ScanMetricsHolder instances to the queue.
+ */
+public class SlowestScanMetricsQueue {
+
+ /**
+ * A no-op implementation that ignores all additions and returns an empty iterator.
+ */
+ public static final SlowestScanMetricsQueue NOOP_SLOWEST_SCAN_METRICS_QUEUE =
+ new SlowestScanMetricsQueue(null) {
+ @Override
+ public void add(ScanMetricsHolder scanMetricsHolder) {
+ }
+
+ @Override
+ public Iterator<ScanMetricsHolder> getIterator() {
+ return Collections.emptyIterator();
+ }
+ };
+
+ private final Deque<ScanMetricsHolder> slowestScanMetricsQueue;
+
+ private SlowestScanMetricsQueue(Deque<ScanMetricsHolder> slowestScanMetricsQueue) {
+ this.slowestScanMetricsQueue = slowestScanMetricsQueue;
+ }
+
+ /**
+ * Creates a new SlowestScanMetricsQueue with an empty concurrent deque.
+ */
+ public SlowestScanMetricsQueue() {
+ this.slowestScanMetricsQueue = new ConcurrentLinkedDeque<>();
+ }
+
+ /**
+ * Adds a {@link ScanMetricsHolder} instance to the queue. This method is thread-safe and can be
+ * called concurrently by multiple threads.
+ * @param scanMetricsHolder the scan metrics holder to add to the queue
+ */
+ public void add(ScanMetricsHolder scanMetricsHolder) {
+ this.slowestScanMetricsQueue.add(scanMetricsHolder);
+ }
+
+ /**
+ * Returns an iterator that traverses the queue in reverse order of insertion (LIFO). The most
+ * recently added {@link ScanMetricsHolder} will be returned first.
+ * @return an iterator over the scan metrics holders in reverse insertion order
+ */
+ public Iterator<ScanMetricsHolder> getIterator() {
+ return this.slowestScanMetricsQueue.descendingIterator();
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/TopNTreeMultiMap.java b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/TopNTreeMultiMap.java
new file mode 100644
index 0000000000..f1443cdd87
--- /dev/null
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/TopNTreeMultiMap.java
@@ -0,0 +1,196 @@
+/*
+ * 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 java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import javax.annotation.Nullable;
+
+import org.apache.phoenix.thirdparty.com.google.common.base.Supplier;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ForwardingListMultimap;
+import org.apache.phoenix.thirdparty.com.google.common.collect.ListMultimap;
+import org.apache.phoenix.thirdparty.com.google.common.collect.Multimap;
+import org.apache.phoenix.thirdparty.com.google.common.collect.MultimapBuilder;
+
+/**
+ * A bounded "Top-N" ListMultimap that maintains a fixed maximum number of entries (N = maxSize).
+ * Keys are sorted using the provided comparator, and values for each key are stored in an
+ * ArrayList.
+ * <p>
+ * When adding a new entry to a full map (size >= maxSize), the implementation evicts the last value
+ * associated with the highest key to make room, but only if the new entry's key is lower than the
+ * current highest key. If the new key is greater than or equal to the highest key, the new entry is
+ * rejected.
+ * @param <K> the type of keys
+ * @param <V> the type of values
+ */
+public class TopNTreeMultiMap<K, V> extends ForwardingListMultimap<K, V> {
+
+ private final ListMultimap<K, V> delegate;
+ private final int maxSize;
+ private final Comparator<? super K> keyComparator;
+
+ public static <K, V> TopNTreeMultiMap<K, V> getInstance(int maxSize,
+ Comparator<? super K> keyComparator) {
+ if (maxSize <= 0) {
+ throw new IllegalArgumentException("maxSize must be positive");
+ }
+ return new TopNTreeMultiMap<>(maxSize, keyComparator);
+ }
+
+ /**
+ * Creates a new bounded multimap with keys sorted according to the provided comparator.
+ * @param maxSize the maximum total number of key-value pairs this map can hold (must be
+ * positive)
+ * @param keyComparator the comparator used to order keys in this multimap
+ */
+ private TopNTreeMultiMap(int maxSize, Comparator<? super K> keyComparator) {
+ this.maxSize = maxSize;
+ this.keyComparator = keyComparator;
+ // Use a TreeMap for sorted keys, with ArrayList for storing multiple values per key
+ this.delegate = MultimapBuilder.treeKeys(keyComparator).arrayListValues().build();
+ }
+
+ @Override
+ protected ListMultimap<K, V> delegate() {
+ return delegate;
+ }
+
+ /**
+ * Returns the comparator used to order the keys in this multimap.
+ * @return the key comparator
+ */
+ public Comparator<? super K> keyComparator() {
+ return keyComparator;
+ }
+
+ /**
+ * Returns the backing key set as a SortedSet view (not a copy).
+ * <p>
+ * Time Complexity: O(1) - Returns a live view of the keys backed by the internal TreeMap.
+ * <p>
+ * The returned SortedSet supports efficient operations:
+ * <ul>
+ * <li>isEmpty(): O(1)</li>
+ * <li>last(): O(log k) where k = number of unique keys</li>
+ * </ul>
+ * @return a sorted view of the keys in this multimap
+ */
+ private SortedSet<K> sortedKeySet() {
+ // This cast is safe according to MultimapBuilder.treeKeys() documentation
+ return (SortedSet<K>) delegate.keySet();
+ }
+
+ /**
+ * Adds a key-value pair to this multimap, potentially evicting an existing entry if the map is
+ * full.
+ * <p>
+ * If the map is at capacity and the new key is smaller than the current highest key, the last
+ * value associated with the highest key is evicted to make room for the new entry. If the new key
+ * is greater than or equal to the highest key, the new entry is rejected.
+ * <p>
+ * Time Complexity:
+ * <ul>
+ * <li>When not full: O(log k) for TreeMap insertion, where k = number of unique keys</li>
+ * <li>When full: O(log k) for finding last key + O(1) for list removal + O(log k) for insertion =
+ * O(log k) overall</li>
+ * </ul>
+ * @param key the key to add
+ * @param valueSupplier a supplier that provides the value to add
+ * @return true if the multimap was modified (entry was added), false otherwise
+ */
+ public boolean put(@Nullable K key, Supplier<V> valueSupplier) {
+ // Fast path: if the map has not reached capacity, add the entry directly
+ // Note: delegate.size() is O(1), delegate.put() is O(log k) for TreeMap
+ if (delegate.size() < maxSize) {
+ return delegate.put(key, valueSupplier.get());
+ }
+
+ // Map is at capacity. Identify the highest key to determine if eviction is needed
+ SortedSet<K> keySet = sortedKeySet();
+ K lastKey = keySet.last();
+
+ // Compare the new key to the current highest key
+ int comp = keyComparator.compare(key, lastKey);
+ if (comp >= 0) {
+ // Reject: new key is greater than or equal to the highest key
+ return false;
+ }
+
+ // New key is smaller than the highest key. Evict the last value from the highest key.
+ List<V> lastValues = delegate.get(lastKey);
+
+ // Remove the last value associated with the highest key.
+ // Important: The list returned by get() is a live view, so this removal:
+ // 1. Automatically decrements delegate.size()
+ // 2. Removes the key entirely if this was its only value
+ lastValues.remove(lastValues.size() - 1);
+
+ // Insert the new entry (this increments size back to maxSize)
+ return delegate.put(key, valueSupplier.get());
+ }
+
+ /**
+ * Adds a key-value pair to this multimap, delegating to the Supplier-based put method. See
+ * {@link #put(Object, Supplier)} for details on the eviction behavior.
+ * @param key the key to add
+ * @param value the value to add
+ * @return true if the multimap was modified (entry was added), false otherwise
+ */
+ @Override
+ public boolean put(@Nullable K key, @Nullable V value) {
+ return put(key, () -> value);
+ }
+
+ /**
+ * Adds all values from the provided iterable to the specified key, applying the bounded put logic
+ * for each value individually.
+ * @param key the key to associate with the values
+ * @param values the iterable of values to add
+ * @return true if the multimap was modified by this operation, false otherwise
+ */
+ @Override
+ public boolean putAll(@Nullable K key, Iterable<? extends V> values) {
+ boolean changed = false;
+ for (V value : values) {
+ if (put(key, value)) {
+ changed = true;
+ }
+ }
+ return changed;
+ }
+
+ /**
+ * Adds all entries from the specified multimap to this multimap, applying the bounded put logic
+ * for each entry individually.
+ * @param multimap the multimap containing entries to add
+ * @return true if this multimap was modified by this operation, false otherwise
+ */
+ @Override
+ public boolean putAll(Multimap<? extends K, ? extends V> multimap) {
+ boolean changed = false;
+ for (Map.Entry<? extends K, ? extends V> entry : multimap.entries()) {
+ if (put(entry.getKey(), entry.getValue())) {
+ changed = true;
+ }
+ }
+ return changed;
+ }
+}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
index dc5d042e46..8adc461a38 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServices.java
@@ -637,6 +637,11 @@ public interface QueryServices extends SQLCloseable {
"phoenix.uncovered.index.threads.keepalive.sec";
String USE_BLOOMFILTER_FOR_MULTIKEY_POINTLOOKUP = "phoenix.bloomfilter.multikey.pointlookup";
+ // Value of N for the top N slowest scan metrics to be collected. Accepts integer values.
+ String SLOWEST_SCAN_METRICS_COUNT = "phoenix.slowest.scan.metrics.count";
+ // Whether to collect region name and server name in the slowest scan metrics. Accepts boolean
+ // values.
+ String SCAN_METRICS_BY_REGION_ENABLED = "phoenix.scan.metrics.by.region.enabled";
/**
* Get executor service used for parallel scans
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
index 2af299aa2c..7fdd24fe70 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/query/QueryServicesOptions.java
@@ -506,6 +506,8 @@ public class QueryServicesOptions {
public static final int DEFAULT_PHOENIX_UNCOVERED_INDEX_MAX_POOL_SIZE = 512;
public static final int DEFAULT_PHOENIX_UNCOVERED_INDEX_KEEP_ALIVE_TIME_SEC = 60; // 1min
+ public static final int DEFAULT_SLOWEST_SCAN_METRICS_COUNT = 0;
+ public static final boolean DEFAULT_SCAN_METRICS_BY_REGION_ENABLED = false;
private final Configuration config;
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/JDBCUtil.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/JDBCUtil.java
index 6822822cf1..97b8193808 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/JDBCUtil.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/JDBCUtil.java
@@ -21,7 +21,10 @@ import static org.apache.phoenix.thirdparty.com.google.common.collect.Maps.newHa
import static org.apache.phoenix.util.PhoenixRuntime.ANNOTATION_ATTRIB_PREFIX;
import edu.umd.cs.findbugs.annotations.NonNull;
+import java.sql.ResultSet;
import java.sql.SQLException;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import javax.annotation.Nullable;
@@ -29,6 +32,7 @@ import org.apache.hadoop.hbase.client.Consistency;
import org.apache.phoenix.jdbc.AbstractRPCConnectionInfo;
import org.apache.phoenix.jdbc.ClusterRoleRecord;
import org.apache.phoenix.jdbc.ConnectionInfo;
+import org.apache.phoenix.jdbc.PhoenixMonitoredResultSet;
import org.apache.phoenix.jdbc.ZKConnectionInfo;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
@@ -37,6 +41,8 @@ import org.apache.phoenix.schema.PNameFactory;
import org.apache.phoenix.thirdparty.com.google.common.base.Preconditions;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* Utilities for JDBC
*/
@@ -139,13 +145,35 @@ public class JDBCUtil {
return (scnStr == null ? null : Long.parseLong(scnStr));
}
+ private static int getIntFromConnectionProperty(String url, Properties props,
+ ReadOnlyProps config, String connPropertyName, String configName, int defaultValue) {
+ String valueString = findProperty(url, props, connPropertyName);
+ return valueString == null
+ ? config.getInt(configName, defaultValue)
+ : Integer.parseInt(valueString);
+ }
+
public static int getMutateBatchSize(String url, Properties info, ReadOnlyProps props)
throws SQLException {
- String batchSizeStr = findProperty(url, info, PhoenixRuntime.UPSERT_BATCH_SIZE_ATTRIB);
- return (batchSizeStr == null
- ? props.getInt(QueryServices.MUTATE_BATCH_SIZE_ATTRIB,
- QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE)
- : Integer.parseInt(batchSizeStr));
+ return getIntFromConnectionProperty(url, info, props, PhoenixRuntime.UPSERT_BATCH_SIZE_ATTRIB,
+ QueryServices.MUTATE_BATCH_SIZE_ATTRIB, QueryServicesOptions.DEFAULT_MUTATE_BATCH_SIZE);
+ }
+
+ public static int getSlowestScanMetricsCount(String url, Properties info, ReadOnlyProps props)
+ throws SQLException {
+ return getIntFromConnectionProperty(url, info, props, QueryServices.SLOWEST_SCAN_METRICS_COUNT,
+ QueryServices.SLOWEST_SCAN_METRICS_COUNT,
+ QueryServicesOptions.DEFAULT_SLOWEST_SCAN_METRICS_COUNT);
+ }
+
+ public static boolean isScanMetricsByRegionEnabled(String url, Properties info,
+ ReadOnlyProps props) throws SQLException {
+ String scanMetricsByRegionEnabledStr =
+ findProperty(url, info, QueryServices.SCAN_METRICS_BY_REGION_ENABLED);
+ return scanMetricsByRegionEnabledStr == null
+ ? props.getBoolean(QueryServices.SCAN_METRICS_BY_REGION_ENABLED,
+ QueryServicesOptions.DEFAULT_SCAN_METRICS_BY_REGION_ENABLED)
+ : Boolean.parseBoolean(scanMetricsByRegionEnabledStr);
}
public static long getMutateBatchSizeBytes(String url, Properties info, ReadOnlyProps props)
@@ -275,4 +303,14 @@ public class JDBCUtil {
}
return formatUrl(url);
}
+
+ public static List<List<JsonObject>> getTopNSlowestScanMetrics(ResultSet rs) {
+ List<List<JsonObject>> metrics;
+ if (rs != null) {
+ metrics = ((PhoenixMonitoredResultSet) rs).getTopNSlowestScanMetrics();
+ } else {
+ metrics = Collections.emptyList();
+ }
+ return metrics;
+ }
}
diff --git a/phoenix-core-client/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java b/phoenix-core-client/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
index b1fc00cc08..311b4c163e 100644
--- a/phoenix-core-client/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
+++ b/phoenix-core-client/src/main/java/org/apache/phoenix/util/PhoenixRuntime.java
@@ -99,6 +99,7 @@ import org.apache.phoenix.thirdparty.org.apache.commons.cli.Options;
import org.apache.phoenix.thirdparty.org.apache.commons.cli.ParseException;
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
/**
* Collection of non JDBC compliant utility methods
@@ -1547,6 +1548,30 @@ public class PhoenixRuntime {
return resultSet.getReadMetrics();
}
+ /**
+ * Method to expose the top N slowest scan metrics for a given result set. A typical pattern is:
+ *
+ * <pre>
+ * {@code
+ * List<List<JsonObject>> slowestScanMetrics = null;
+ * try (ResultSet rs = stmt.executeQuery()) {
+ * while(rs.next()) {
+ * .....
+ * }
+ * slowestScanMetrics = PhoenixRuntime.getTopNSlowestScanMetrics(rs);
+ * }
+ * }
+ * </pre>
+ *
+ * @param rs the result set to get the slowest scan metrics
+ * @return a list of lists of {@link JsonObject} objects containing the metrics for top N slowest
+ * scans
+ */
+ public static List<List<JsonObject>> getTopNSlowestScanMetrics(ResultSet rs) throws SQLException {
+ PhoenixMonitoredResultSet resultSet = rs.unwrap(PhoenixMonitoredResultSet.class);
+ return resultSet.getTopNSlowestScanMetrics();
+ }
+
@Deprecated
// use getRequestReadMetricInfo
public static Map<String, Map<String, Long>> getRequestReadMetrics(ResultSet rs)
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingIT.java
index b3565f3376..b86793badb 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingIT.java
@@ -76,7 +76,7 @@ public class ServerPagingIT extends ParallelStatsDisabledIT {
setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
}
- private void assertServerPagingMetric(String tableName, ResultSet rs, boolean isPaged)
+ private void assertServerPagingMetric(String tableName, ResultSet rs, boolean isDummyRowReturned)
throws SQLException {
Map<String, Map<MetricType, Long>> metrics = PhoenixRuntime.getRequestReadMetricInfo(rs);
for (Map.Entry<String, Map<MetricType, Long>> entry : metrics.entrySet()) {
@@ -84,7 +84,7 @@ public class ServerPagingIT extends ParallelStatsDisabledIT {
Map<MetricType, Long> metricValues = entry.getValue();
Long pagedRowsCntr = metricValues.get(MetricType.PAGED_ROWS_COUNTER);
assertNotNull(pagedRowsCntr);
- if (isPaged) {
+ if (isDummyRowReturned) {
assertTrue(String.format("Got %d", pagedRowsCntr.longValue()), pagedRowsCntr > 0);
} else {
assertTrue(String.format("Got %d", pagedRowsCntr.longValue()), pagedRowsCntr == 0);
@@ -582,7 +582,11 @@ public class ServerPagingIT extends ParallelStatsDisabledIT {
assertTrue(rs.next());
assertEquals(STRINGS[i - 1], rs.getString(1));
}
- assertServerPagingMetric(tablename, rs, true);
+ // To ensure PAGED_ROWS_COUNTER is incremented correctly, we need to close the result set.
+ rs.close();
+ // No dummy rows are returned though paging happens as we only hit the condition of paging
+ // filter being stopped with a valid row.
+ assertServerPagingMetric(tablename, rs, false);
limit = 1;
offset = 1;
rs = conn.createStatement().executeQuery(
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingWithRegionMovesIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingWithRegionMovesIT.java
index 0c57cd3043..3b7a818e5f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingWithRegionMovesIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ServerPagingWithRegionMovesIT.java
@@ -142,7 +142,7 @@ public class ServerPagingWithRegionMovesIT extends ParallelStatsDisabledIT {
countOfValidResults = 0;
}
- private void assertServerPagingMetric(String tableName, ResultSet rs, boolean isPaged)
+ private void assertServerPagingMetric(String tableName, ResultSet rs, boolean isDummyRowReturned)
throws SQLException {
Map<String, Map<MetricType, Long>> metrics = PhoenixRuntime.getRequestReadMetricInfo(rs);
for (Map.Entry<String, Map<MetricType, Long>> entry : metrics.entrySet()) {
@@ -150,7 +150,7 @@ public class ServerPagingWithRegionMovesIT extends ParallelStatsDisabledIT {
Map<MetricType, Long> metricValues = entry.getValue();
Long pagedRowsCntr = metricValues.get(MetricType.PAGED_ROWS_COUNTER);
assertNotNull(pagedRowsCntr);
- if (isPaged) {
+ if (isDummyRowReturned) {
assertTrue(String.format("Got %d", pagedRowsCntr), pagedRowsCntr > 0);
} else {
assertEquals(String.format("Got %d", pagedRowsCntr), 0, (long) pagedRowsCntr);
@@ -503,7 +503,18 @@ public class ServerPagingWithRegionMovesIT extends ParallelStatsDisabledIT {
assertTrue(rs.next());
assertEquals(STRINGS[i - 1], rs.getString(1));
}
- assertServerPagingMetric(tablename, rs, false);
+ // To ensure PAGED_ROWS_COUNTER is incremented correctly, we need to close the result set.
+ rs.close();
+ // Dummy rows are returned as because regions are being moved so, cells have been flushed to
+ // HFile. When doing a full table scan and cells are flushed to HFile, we first read a row
+ // with rowkey as startkey of the scan and cell value as lastRowCol for that rowkey as
+ // startkey is not included. This row is filtered out by HBase store scanner but as paging
+ // filter has stopped so, RegionScanner ends up returning a empty row which leads to
+ // PagingRegionScanner returning a dummy row. When region moves, same logic is repeated as
+ // HBase client scanner tracks the last read rowkey and sets that as startkey of the scan when
+ // scan is reopened but startkey is not included. Thus, we will see dummy row is returned by
+ // PagingRegionScanner when first row of query is read or first row after region move is read.
+ assertServerPagingMetric(tablename, rs, true);
limit = 1;
offset = 1;
rs = conn.createStatement().executeQuery(
@@ -678,8 +689,18 @@ public class ServerPagingWithRegionMovesIT extends ParallelStatsDisabledIT {
assertTrue(rs.next());
assertEquals(STRINGS[i - 1], rs.getString(1));
}
- // no paging when serial offset
- assertServerPagingMetric(tablename, rs, false);
+ // To ensure PAGED_ROWS_COUNTER is incremented correctly, we need to close the result set.
+ rs.close();
+ // Dummy rows are returned as because regions are being moved so, cells have been flushed to
+ // HFile. When doing a full table scan and cells are flushed to HFile, we first read a row
+ // with rowkey as startkey of the scan and cell value as lastRowCol for that rowkey as
+ // startkey is not included. This row is filtered out by HBase store scanner but as paging
+ // filter has stopped so, RegionScanner ends up returning a empty row which leads to
+ // PagingRegionScanner returning a dummy row. When region moves, same logic is repeated as
+ // HBase client scanner tracks the last read rowkey and sets that as startkey of the scan when
+ // scan is reopened but startkey is not included. Thus, we will see dummy row is returned by
+ // PagingRegionScanner when first row of query is read or first row after region move is read.
+ assertServerPagingMetric(tablename, rs, true);
limit = 1;
offset = 1;
rs = conn.createStatement().executeQuery(
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
index 655cbbcfaa..4a0c518efe 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/jdbc/ParallelPhoenixConnectionIT.java
@@ -46,6 +46,7 @@ import static org.apache.phoenix.query.QueryServices.CQSI_THREAD_POOL_MAX_THREAD
import static org.apache.phoenix.query.QueryServices.CQSI_THREAD_POOL_METRICS_ENABLED;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -69,6 +70,7 @@ import java.util.concurrent.ThreadPoolExecutor;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.RetriesExhaustedWithDetailsException;
import org.apache.hadoop.hbase.util.VersionInfo;
import org.apache.hadoop.test.GenericTestUtils;
@@ -76,6 +78,7 @@ import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
import org.apache.phoenix.exception.MutationBlockedIOException;
import org.apache.phoenix.jdbc.ClusterRoleRecord.ClusterRole;
import org.apache.phoenix.jdbc.HighAvailabilityTestingUtility.HBaseTestingUtilityPair;
+import org.apache.phoenix.jdbc.ParallelPhoenixResultSetFactory.ParallelPhoenixResultSetType;
import org.apache.phoenix.jdbc.PhoenixStatement.Operation;
import org.apache.phoenix.log.LogLevel;
import org.apache.phoenix.monitoring.GlobalMetric;
@@ -83,13 +86,16 @@ import org.apache.phoenix.monitoring.HTableThreadPoolHistograms;
import org.apache.phoenix.monitoring.HTableThreadPoolMetricsManager;
import org.apache.phoenix.monitoring.HistogramDistribution;
import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.monitoring.SlowestScanMetricsIT;
import org.apache.phoenix.query.ConnectionQueryServices;
import org.apache.phoenix.query.ConnectionQueryServicesImpl;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.util.PhoenixRuntime;
import org.apache.phoenix.util.QueryUtil;
+import org.apache.phoenix.util.TestUtil;
import org.junit.AfterClass;
import org.junit.Assert;
+import org.junit.Assume;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Rule;
@@ -100,6 +106,9 @@ import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.apache.hbase.thirdparty.com.google.gson.JsonArray;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
/**
* Test failover basics for {@link ParallelPhoenixConnection}.
*/
@@ -709,6 +718,68 @@ public class ParallelPhoenixConnectionIT {
}
}
+ @Test
+ public void testTopNSlowestScanMetrics() throws Exception {
+ testTopNSlowestScanMetrics(ParallelPhoenixResultSetType.PARALLEL_PHOENIX_RESULT_SET);
+ }
+
+ @Test
+ public void testTopNSlowestScanMetricsWithNullComparingResultSet() throws Exception {
+ testTopNSlowestScanMetrics(
+ ParallelPhoenixResultSetType.PARALLEL_PHOENIX_NULL_COMPARING_RESULT_SET);
+ }
+
+ private void testTopNSlowestScanMetrics(ParallelPhoenixResultSetType rsType) throws Exception {
+ Assume.assumeTrue(VersionInfo.compareVersion(VersionInfo.getVersion(), "2.6.3") > 0);
+ try (Connection conn = getParallelConnection()) {
+ try (Statement stmt = conn.createStatement()) {
+ stmt.executeUpdate(String.format("UPSERT INTO %s VALUES(%d, 1984)", tableName, 0));
+ conn.commit();
+ TestUtil.flush(CLUSTERS.getHBaseCluster1(), TableName.valueOf(tableName));
+ TestUtil.flush(CLUSTERS.getHBaseCluster2(), TableName.valueOf(tableName));
+ }
+ waitForCompletion(conn);
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(1));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ props.setProperty(ParallelPhoenixResultSetFactory.PHOENIX_PARALLEL_RESULTSET_TYPE,
+ rsType.getName());
+ try (Connection conn = getParallelConnection(props)) {
+ try (Statement stmt = conn.createStatement()) {
+ ResultSet rs = stmt.executeQuery(String.format("SELECT * FROM %s", tableName));
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ if (rsType == ParallelPhoenixResultSetType.PARALLEL_PHOENIX_RESULT_SET) {
+ assertTrue(rs instanceof ParallelPhoenixResultSet);
+ } else {
+ assertTrue(rs instanceof ParallelPhoenixNullComparingResultSet);
+ }
+ JsonArray slowestScanMetricsJsonArray =
+ SlowestScanMetricsIT.getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as its scan of single region only.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as it's a simple query so, only data block is read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
/**
* Test Phoenix connection metrics when no metrics have been generated
*/
@@ -907,6 +978,12 @@ public class ParallelPhoenixConnectionIT {
return DriverManager.getConnection(CLUSTERS.getJdbcHAUrl(), clientProperties);
}
+ private Connection getParallelConnection(Properties props) throws SQLException {
+ Properties combinedProps = new Properties(clientProperties);
+ combinedProps.putAll(props);
+ return DriverManager.getConnection(CLUSTERS.getJdbcHAUrl(), combinedProps);
+ }
+
void waitForCompletion(Connection conn) throws Exception {
ParallelPhoenixContext context = ((ParallelPhoenixConnection) conn).getContext();
Thread.sleep(200);
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/SlowestScanMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/SlowestScanMetricsIT.java
new file mode 100644
index 0000000000..d79edeb958
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/SlowestScanMetricsIT.java
@@ -0,0 +1,955 @@
+/*
+ * 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 static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.util.VersionInfo;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.end2end.NeedsOwnMiniClusterTest;
+import org.apache.phoenix.jdbc.PhoenixResultSet;
+import org.apache.phoenix.query.BaseTest;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.MetaDataUtil;
+import org.apache.phoenix.util.PhoenixRuntime;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.TestUtil;
+import org.junit.Assume;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Maps;
+
+import org.apache.hbase.thirdparty.com.google.gson.JsonArray;
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
+@Category(NeedsOwnMiniClusterTest.class)
+public class SlowestScanMetricsIT extends BaseTest {
+ private static final Random RAND = new Random(11);
+
+ @BeforeClass
+ public static void setup() throws Exception {
+ Assume.assumeTrue(VersionInfo.compareVersion(VersionInfo.getVersion(), "2.6.3") > 0);
+ Map<String, String> props = Maps.newHashMapWithExpectedSize(3);
+ props.put(QueryServices.COLLECT_REQUEST_LEVEL_METRICS, "true");
+ setUpTestDriver(new ReadOnlyProps(props.entrySet().iterator()));
+ }
+
+ @Test
+ public void testSinglePointLookupQuery() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String sql = "SELECT * FROM " + tableName + " WHERE k1 = 1 AND k2 = 'a'";
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTableAndUpsertData(conn, tableName, "");
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as its single point lookup though topN is 2.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple single point lookup and a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 2 as it's a single point lookup so, bloom filter block will also be read along with
+ // data block.
+ assertEquals(2, regionJson.get("broc").getAsLong());
+ }
+ }
+
+ @Test
+ public void testMultiplePointsLookupQuery() throws Exception {
+ int topN = 1;
+ String tableName = generateUniqueName();
+ String sql = "SELECT * FROM " + tableName + " WHERE (k1, k2) IN ((1, 'a'), (2, 'b'), (3, 'c'))";
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(3, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 2 as its a multi-point lookup query and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple single point lookup and a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as it's a multi-point lookup query so, only data block will be read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testUnionAllQuery() throws Exception {
+ int topN = 2;
+ String tableName1 = generateUniqueName();
+ String tableName2 = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName1, "");
+ createTableAndUpsertData(conn, tableName2, "");
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT * FROM " + tableName1 + " WHERE k1 = 1 AND k2 = 'a'"
+ + " UNION ALL SELECT * FROM " + tableName2 + " WHERE k1 = 1 AND k2 = 'a'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(2, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 2 as its UNION of 2 single point lookup queries and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a union of 2 single point lookup queries and not a query
+ // with subquery or JOIN operation. Two HBase tables are scanned and there is one inner
+ // array for each table.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ String tableName = groupJson.get("table").getAsString();
+ assertTrue(tableName1.equals(tableName) || tableName2.equals(tableName));
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 2 as each query in UNION is a single point lookup so, bloom filter block will
+ // also be read along with data block.
+ assertEquals(2, regionJson.get("broc").getAsLong());
+ }
+
+ StatementContext stmtCtx = rs.unwrap(PhoenixResultSet.class).getContext();
+
+ // Verify that existing read metrics are combined correctly.
+ ReadMetricQueue readMetricsQueue = stmtCtx.getReadMetricsQueue();
+ Map<String, Map<MetricType, Long>> expectedMetrics = readMetricsQueue.aggregate();
+ Map<String, Map<MetricType, Long>> actualMetrics =
+ PhoenixRuntime.getRequestReadMetricInfo(rs);
+ assertEquals(expectedMetrics, actualMetrics);
+
+ // Verify that overall query metrics are published correctly.
+ Map<MetricType, Long> expectedOverallQueryMetrics =
+ stmtCtx.getOverallQueryMetrics().publish();
+ Map<MetricType, Long> actualOverallQueryMetrics =
+ PhoenixRuntime.getOverAllReadRequestMetricInfo(rs);
+ assertEquals(expectedOverallQueryMetrics, actualOverallQueryMetrics);
+ }
+ }
+
+ @Test
+ public void testQueryContainingSubquery() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String subqueryTableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "");
+ createTableAndUpsertData(conn, subqueryTableName, "");
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT * FROM " + tableName + " WHERE k1 IN (SELECT k1 FROM "
+ + subqueryTableName + " WHERE k2 IN ('a', 'b'))";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(2, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as it's a query containing subquery and both outer query and
+ // subquery are scanning single regions.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 2 as there is one region scan metrics for outer query and one for
+ // subquery.
+ assertEquals(2, groupArray.size());
+ for (int i = 0; i < groupArray.size(); i++) {
+ JsonObject groupJson = groupArray.get(i).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ String hbaseTableName = groupJson.get("table").getAsString();
+ assertTrue(hbaseTableName.equals(tableName) || hbaseTableName.equals(subqueryTableName));
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as both outer query and subquery are scanning multiple rows so, only data block
+ // will be read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testQueryWithSelfJoin() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE TABLE " + tableName
+ + " (k1 INTEGER NOT NULL, k2 varchar NOT NULL, v1 VARCHAR, v2 VARCHAR, CONSTRAINT PK PRIMARY KEY (k1, k2))");
+ conn.commit();
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'a', 'a1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'b', 'b1', 'b2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'c', 'c1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'd', 'd1', 'd2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'e', 'e1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'f', 'f1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (5, 'g', 'g1', 'v2')");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(tableName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql =
+ "SELECT a.k1 as k1, b.k2 as k2, b.v1 as v1, a.total_count as total_count FROM (SELECT k1, COUNT(*) as total_count FROM "
+ + tableName + " WHERE k1 IN (1, 3) GROUP BY k1) a JOIN (SELECT k1, k2, v1 FROM "
+ + tableName + " WHERE k1 IN (1, 3) AND k2 = 'a') b ON a.k1 = b.k1";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Same sized as inner array.
+ long[] brocValues = new long[2];
+ long[] brfcValues = new long[2];
+ // Outer array has size 1 as it's a JOIN query and both left and right tables are scanning
+ // single regions.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 2 as there is one region scan metrics for left table and one for
+ // right table.
+ assertEquals(2, groupArray.size());
+ for (int i = 0; i < groupArray.size(); i++) {
+ JsonObject groupJson = groupArray.get(i).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ String hbaseTableName = groupJson.get("table").getAsString();
+ assertEquals(tableName, hbaseTableName);
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ brocValues[i] = regionJson.get("broc").getAsLong();
+ brfcValues[i] = regionJson.get("brfc").getAsLong();
+ }
+
+ // On scanning the left table, the data blocks get loaded into block cache. So, left table
+ // reads data from file system and right table reads data from block cache. As both left and
+ // right tables are scanning single regions, only 1 data block will be read.
+ assertEquals(0, brocValues[0]);
+ assertEquals(1, brocValues[1]);
+ assertTrue(brfcValues[0] > 0);
+ assertEquals(0, brfcValues[1]);
+ }
+ }
+
+ @Test
+ public void testUnionAllAggregateQuery() throws Exception {
+ int topN = 2;
+ String tableName1 = generateUniqueName();
+ String tableName2 = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName1, "");
+ createTableAndUpsertData(conn, tableName2, "");
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT MAX(v1) FROM (SELECT k1, v1 FROM " + tableName1
+ + " UNION ALL SELECT k1, v1 FROM " + tableName2 + ") GROUP BY k1 HAVING MAX(v1) > 'c1'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 2 as its UNION of 2 queries each scanning single region
+ // and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a union of 2 queries each scanning single region and not a
+ // query with subquery or JOIN operation. Two HBase tables are scanned and there is one
+ // inner array for each table.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(2, groupJson.size());
+ String tableName = groupJson.get("table").getAsString();
+ assertTrue(tableName1.equals(tableName) || tableName2.equals(tableName));
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as each query in UNIONed query is doing full table scan so, only data block is
+ // read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+
+ StatementContext stmtCtx = rs.unwrap(PhoenixResultSet.class).getContext();
+ StatementContext subStmtCtx = stmtCtx.getSubStatementContexts().iterator().next();
+
+ // Verify that existing read metrics are combined correctly.
+ ReadMetricQueue readMetricsQueue = subStmtCtx.getReadMetricsQueue();
+ Map<String, Map<MetricType, Long>> expectedMetrics = readMetricsQueue.aggregate();
+ Map<String, Map<MetricType, Long>> actualMetrics =
+ PhoenixRuntime.getRequestReadMetricInfo(rs);
+ assertEquals(expectedMetrics, actualMetrics);
+
+ // Verify that overall query metrics are published correctly.
+ Map<MetricType, Long> expectedOverallQueryMetrics =
+ stmtCtx.getOverallQueryMetrics().publish();
+ Map<MetricType, Long> actualOverallQueryMetrics =
+ PhoenixRuntime.getOverAllReadRequestMetricInfo(rs);
+ assertEquals(expectedOverallQueryMetrics, actualOverallQueryMetrics);
+ }
+ }
+
+ @Test
+ public void testConstantQuery() throws Exception {
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(2));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT 'This is a constant query'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array contains empty inner array as its a constant query not needing to scan any
+ // HBase table.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ assertEquals(0, groupArray.size());
+ }
+ }
+
+ @Test
+ public void testRequestLevelMetricsDisabled() throws Exception {
+ String tableName = generateUniqueName();
+ Properties props = new Properties();
+ props.setProperty(PhoenixRuntime.REQUEST_METRIC_ATTRIB, "false");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ createTableAndUpsertData(conn, tableName, "");
+ Statement stmt = conn.createStatement();
+ String sql = "SELECT * FROM " + tableName + " WHERE k1 = 1 AND k2 = 'a'";
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+ // No slowest scan metrics are returned as request level metrics are disabled.
+ assertTrue(slowestScanMetricsJsonArray.isEmpty());
+ }
+ }
+
+ @Test
+ public void testSlowestScanMetricsDisabled() throws Exception {
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "");
+ Statement stmt = conn.createStatement();
+ String sql = "SELECT * FROM " + tableName + " WHERE k1 = 1 AND k2 = 'a'";
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+ // No slowest scan metrics are returned as slowest scan metrics are disabled.
+ assertEquals(0, slowestScanMetricsJsonArray.size());
+ }
+ }
+
+ @Test
+ public void testHBaseScanMetricsByRegionDisabled() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql =
+ "SELECT * FROM " + tableName + " WHERE (k1, k2) IN ((1, 'a'), (2, 'b'), (3, 'c'))";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(3, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 2 as its a multi-point lookup query and topN is 2 even though 3
+ // parallel scans are generated.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ String hbaseTableName = groupJson.get("table").getAsString();
+ assertEquals(tableName, hbaseTableName);
+ // No region name and server are returned as HBase scan metrics by region are disabled.
+ assertNull(groupJson.get("regions"));
+ assertNull(groupJson.get("server"));
+ assertNull(groupJson.get("region"));
+ // BROC is 1 as it's a multi-point lookup query so, only data block is read.
+ assertEquals(1, groupJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testAggregateQueryWithoutGroupBy() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT MAX(v1) FROM " + tableName;
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 2 as aggregate query did a full table scan and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple aggregate query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as it's a full table scan so, only data block is read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testAggregateQueryWithGroupBy() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "BLOOMFILTER='NONE'");
+ Statement stmt = conn.createStatement();
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'a', 'a1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'b', 'b1', 'b2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'c', 'c1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'd', 'd1', 'd2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'e', 'e1', 'v2')");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(tableName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT MAX(v1) FROM " + tableName + " WHERE v2 = 'v2' GROUP BY k1";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(2, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as its scan of single region only.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple aggregate query with group by and not a query with
+ // subquery or JOIN operation.
+ assertEquals(1, groupArray.size());
+ for (int i = 0; i < groupArray.size(); i++) {
+ JsonObject groupJson = groupArray.get(i).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // Though it's a full table scan but still broc is 2 as there are 2 HFiles, 1 per commit
+ // call, and 1 data block is read from each HFile.
+ assertEquals(2, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testAggregateQueryWithGroupByAndOrderBy() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "BLOOMFILTER='NONE'");
+ Statement stmt = conn.createStatement();
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'a', 'a1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'b', 'b1', 'b2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'c', 'c1', 'v2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'd', 'd1', 'd2')");
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'e', 'e1', 'v2')");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(tableName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT v2, MAX(v1) FROM " + tableName + " GROUP BY v2 ORDER BY v2";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ // Read total 5 rows from 2 HFiles.
+ assertEquals(5, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as its scan of single region only.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple aggregate query with group by and not a query with
+ // subquery or JOIN operation.
+ assertEquals(1, groupArray.size());
+ for (int i = 0; i < groupArray.size(); i++) {
+ JsonObject groupJson = groupArray.get(i).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // Though it's a full table scan but still broc is 2 as there are 2 HFiles, 1 per commit
+ // call, and one data block is read from each HFile.
+ assertEquals(2, regionJson.get("broc").getAsLong());
+ }
+ }
+ }
+
+ @Test
+ public void testQueryOnCoveredIndex() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE INDEX " + indexName + " ON " + tableName + " (v1) INCLUDE (v2)");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(indexName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT v1, v2 FROM " + tableName + " WHERE v1 = 'a1'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ int totalBroc = 0;
+ // Outer array has size 2 as the base table is salted so, salt bucket no. of parallel scans
+ // are generated and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(indexName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ totalBroc += regionJson.get("broc").getAsLong();
+ }
+ // Total BROC is 1 as data is read from only one region.
+ assertEquals(1, totalBroc);
+ }
+ }
+
+ @Test
+ public void testQueryOnUncoveredIndex() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String indexName = generateUniqueName();
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE UNCOVERED INDEX " + indexName + " ON " + tableName + " (v1)");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(indexName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT v1, v2 FROM " + tableName + " WHERE v1 = 'a1'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ int totalBroc = 0;
+ // Outer array has size 2 as the base table is salted so, salt bucket no. of parallel scans
+ // are generated and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(indexName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ totalBroc += regionJson.get("broc").getAsLong();
+ }
+ // Total BROC is 2 as data is read from one region of index table and one region of data table
+ // as index is uncovered.
+ assertEquals(2, totalBroc);
+ }
+ }
+
+ @Test
+ public void testQueryOnView() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String viewName = generateUniqueName();
+ assertNotEquals(viewName, tableName);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "");
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE VIEW " + viewName + " AS SELECT * FROM " + tableName + " WHERE k1 = 1");
+ conn.commit();
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT * FROM " + viewName;
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ // Outer array has size 1 as its scan of single region only.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ // BROC is 1 as it's not a single point lookup so, only data block is read.
+ assertEquals(1, regionJson.get("broc").getAsLong());
+ }
+ }
+
+ @Test
+ public void testQueryOnViewIndex() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ String viewName = generateUniqueName();
+ String viewIndexName = generateUniqueName();
+ String viewIndexPhysicalName = MetaDataUtil.getViewIndexPhysicalName(tableName);
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "SALT_BUCKETS=3");
+ Statement stmt = conn.createStatement();
+ stmt.execute("CREATE VIEW " + viewName + " AS SELECT * FROM " + tableName + " WHERE k1 = 1");
+ conn.commit();
+ stmt.execute("CREATE INDEX " + viewIndexName + " ON " + viewName + " (v1) INCLUDE (v2)");
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(viewIndexPhysicalName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ String sql = "SELECT v2 FROM " + viewName + " WHERE v1 = 'a1'";
+ Statement stmt = conn.createStatement();
+ ResultSet rs = stmt.executeQuery(sql);
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertEquals(1, rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ int totalBroc = 0;
+ // Outer array has size 2 as the base table is salted so, salt bucket no. of parallel scans
+ // are generated and topN is 2.
+ assertEquals(topN, slowestScanMetricsJsonArray.size());
+ for (int i = 0; i < topN; i++) {
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(i).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(viewIndexPhysicalName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ totalBroc += regionJson.get("broc").getAsLong();
+ }
+ // Total BROC is 1 as data is read from single region of view index table.
+ assertEquals(1, totalBroc);
+ }
+ }
+
+ @Test
+ public void testMultiKeyPointLookupWithPaging() throws Exception {
+ int topN = 2;
+ String tableName = generateUniqueName();
+ int totalRows = 10000;
+ try (Connection conn = DriverManager.getConnection(getUrl())) {
+ createTableAndUpsertData(conn, tableName, "");
+ String dml = "UPSERT INTO " + tableName + " VALUES(?, ?, ?, ?)";
+ PreparedStatement ps = conn.prepareStatement(dml);
+ for (int i = 0; i < totalRows; ++i) {
+ ps.setInt(1, i % 3);
+ ps.setString(2, "k2_" + i);
+ ps.setString(3, "v1_" + i);
+ ps.setString(4, "v2_" + i % 10);
+ ps.executeUpdate();
+ if (i != 0 && i % 100 == 0) {
+ conn.commit();
+ }
+ }
+ conn.commit();
+ TestUtil.flush(getUtility(), TableName.valueOf(tableName));
+ }
+ Properties props = new Properties();
+ props.setProperty(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(topN));
+ props.setProperty(QueryServices.SCAN_METRICS_BY_REGION_ENABLED, "true");
+ props.setProperty(QueryServices.PHOENIX_SERVER_PAGE_SIZE_MS, String.valueOf(5));
+ try (Connection conn = DriverManager.getConnection(getUrl(), props)) {
+ int rowKeyCount = 1000;
+ List<String> inList =
+ Stream.generate(() -> "(?, ?)").limit(rowKeyCount).collect(Collectors.toList());
+ String dql = String.format("select k1, k2 from %s where (k1, k2) IN (%s)", tableName,
+ String.join(",", inList));
+ PreparedStatement ps = conn.prepareStatement(dql);
+ for (int i = 0; i < rowKeyCount; i++) {
+ ps.setInt(2 * i + 1, RAND.nextInt(3));
+ if (RAND.nextBoolean()) {
+ ps.setString(2 * i + 2, "k2_" + RAND.nextInt(totalRows));
+ } else {
+ // generate a non-existing row key
+ ps.setString(2 * i + 2, "k2_" + 78123);
+ }
+ }
+ ResultSet rs = ps.executeQuery();
+ int rowCount = 0;
+ while (rs.next()) {
+ rowCount++;
+ }
+ assertTrue(rowKeyCount >= rowCount);
+ JsonArray slowestScanMetricsJsonArray = getSlowestScanMetricsJsonArray(rs);
+
+ long brocFromSlowestScanMetrics;
+ // Outer array has size 1 as its scan of single region only.
+ assertEquals(1, slowestScanMetricsJsonArray.size());
+ JsonArray groupArray = slowestScanMetricsJsonArray.get(0).getAsJsonArray();
+ // Inner array has size 1 as it's a simple query and not a query with subquery or
+ // JOIN operation.
+ assertEquals(1, groupArray.size());
+ JsonObject groupJson = groupArray.get(0).getAsJsonObject();
+ assertEquals(tableName, groupJson.get("table").getAsString());
+ JsonArray regionsArray = groupJson.get("regions").getAsJsonArray();
+ assertEquals(1, regionsArray.size());
+ JsonObject regionJson = regionsArray.get(0).getAsJsonObject();
+ assertNotNull(regionJson.get("region"));
+ assertNotNull(regionJson.get("server"));
+ brocFromSlowestScanMetrics = regionJson.get("broc").getAsLong();
+ assertTrue(regionJson.get("rp").getAsLong() > 1);
+
+ Map<String, Map<MetricType, Long>> metrics = PhoenixRuntime.getRequestReadMetricInfo(rs);
+ long brocFromReadMetrics = metrics.get(tableName).get(MetricType.BLOCK_READ_OPS_COUNT);
+ assertEquals(brocFromReadMetrics, brocFromSlowestScanMetrics);
+ long pagedRowsCounter = metrics.get(tableName).get(MetricType.PAGED_ROWS_COUNTER);
+ assertTrue(pagedRowsCounter > 0);
+ }
+ }
+
+ private void createTableAndUpsertData(Connection conn, String tableName, String ddlOptions)
+ throws Exception {
+ String createTableSql = "CREATE TABLE " + tableName
+ + " (k1 INTEGER NOT NULL, k2 varchar NOT NULL, v1 VARCHAR, v2 VARCHAR, CONSTRAINT PK PRIMARY KEY (k1, k2)) "
+ + ddlOptions;
+ try (Statement stmt = conn.createStatement()) {
+ stmt.execute(createTableSql);
+ conn.commit();
+ stmt.execute("UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (1, 'a', 'a1', 'a2')");
+ stmt.execute(
+ "UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (2, 'b', 'b1000', 'b2000')");
+ stmt.execute(
+ "UPSERT INTO " + tableName + " (k1, k2, v1, v2) VALUES (3, 'c', 'c1000000', 'c2000000')");
+ conn.commit();
+ }
+ TestUtil.flush(getUtility(), TableName.valueOf(tableName));
+ }
+
+ /**
+ * Returns a JSON array of arrays of JSON maps. Each JSON map contains name of HBase table scanned
+ * and list of scan metrics for each region scanned along with region and server name. The list of
+ * scan metrics in JSON map will have more than one entry if region moved b/w consecutive RPC
+ * calls.
+ * @param rs ResultSet from which to get the slowest scan metrics
+ * @return JSON array of arrays of JSON maps
+ */
+ public static JsonArray getSlowestScanMetricsJsonArray(ResultSet rs) throws Exception {
+ List<List<JsonObject>> slowestScanMetrics = PhoenixRuntime.getTopNSlowestScanMetrics(rs);
+ JsonArray jsonArray = new JsonArray();
+ for (List<JsonObject> group : slowestScanMetrics) {
+ JsonArray groupArray = new JsonArray();
+ for (JsonObject scanMetrics : group) {
+ groupArray.add(scanMetrics);
+ }
+ jsonArray.add(groupArray);
+ }
+ System.out.println("Slowest scan metrics JSON array: " + jsonArray);
+ return jsonArray;
+ }
+}
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/compile/SlowestScanMetricsTest.java b/phoenix-core/src/test/java/org/apache/phoenix/compile/SlowestScanMetricsTest.java
new file mode 100644
index 0000000000..ab31b1d443
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/compile/SlowestScanMetricsTest.java
@@ -0,0 +1,406 @@
+/*
+ * 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.compile;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
+import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.jdbc.PhoenixStatement;
+import org.apache.phoenix.log.LogLevel;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.ScanMetricsHolder;
+import org.apache.phoenix.query.ConnectionQueryServices;
+import org.apache.phoenix.query.QueryServices;
+import org.apache.phoenix.util.ReadOnlyProps;
+import org.junit.Test;
+
+import org.apache.hbase.thirdparty.com.google.gson.JsonObject;
+
+/**
+ * Unit test for {@link StatementContext#getTopNSlowestScanMetrics()}. This test validates the logic
+ * of computing top N slowest scan metrics in isolation.
+ */
+public class SlowestScanMetricsTest {
+
+ /**
+ * Test that when slowestScanMetricsCount is 0 or negative, an empty list is returned.
+ */
+ @Test
+ public void testWithZeroOrNegativeCount() {
+ StatementContext context = createStatementContext(0);
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+ assertTrue(result.isEmpty());
+
+ context = createStatementContext(-1);
+ result = context.getTopNSlowestScanMetrics();
+ assertTrue(result.isEmpty());
+ }
+
+ /**
+ * Test with a single scan metrics holder.
+ */
+ @Test
+ public void testWithSingleScanMetric() {
+ StatementContext context = createStatementContext(5);
+ long millisBetweenNexts = 100L;
+
+ // Add a single scan metric
+ ScanMetricsHolder scanMetric = createScanMetricsHolder("TABLE1", millisBetweenNexts);
+ context.getSlowestScanMetricsQueue().add(scanMetric);
+
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+
+ assertEquals(1, result.size());
+ assertEquals(1, result.get(0).size());
+ assertEquals("TABLE1", result.get(0).get(0).get("table").getAsString());
+ assertEquals(millisBetweenNexts, result.get(0).get(0).get("n").getAsLong());
+ }
+
+ /**
+ * Test with multiple scan metrics holders - verify the sorted order with slowest one first.
+ */
+ @Test
+ public void testWithMultipleScanMetrics_Sorted() {
+ int topN = 3;
+ StatementContext context = createStatementContext(topN);
+
+ // Add scan metrics with different times
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE1", 100L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE2", 500L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE3", 200L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE4", 300L));
+
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+
+ // Should return top 3 slowest: 500, 300, 200 in this order only.
+ assertEquals(topN, result.size());
+
+ // Verify the results are sorted by time in descending order
+ assertEquals(500L, result.get(0).get(0).get("n").getAsLong());
+ assertEquals(300L, result.get(1).get(0).get("n").getAsLong());
+ assertEquals(200L, result.get(2).get(0).get("n").getAsLong());
+ }
+
+ /**
+ * Test with fewer scan metrics holders than the topN limit.
+ */
+ @Test
+ public void testWithFewerMetricsThanTopN() {
+ StatementContext context = createStatementContext(10);
+
+ // Add only 3 metrics when topN is 10
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE1", 100L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE2", 300L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE3", 200L));
+
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+
+ assertEquals(3, result.size());
+
+ // Verify sorting
+ assertEquals(300L, result.get(0).get(0).get("n").getAsLong());
+ assertEquals(200L, result.get(1).get(0).get("n").getAsLong());
+ assertEquals(100L, result.get(2).get(0).get("n").getAsLong());
+ }
+
+ /**
+ * Test with sub-statement contexts (e.g., subqueries or joins). Verifies that scan metrics from
+ * sub-contexts are included and aggregated correctly. The algorithm only returns paths that reach
+ * leaf nodes (contexts with no sub-contexts).
+ */
+ @Test
+ public void testWithSubStatementContexts() {
+ StatementContext mainContext = createStatementContext(5);
+ String mainTableName = "MAIN_TABLE";
+ String subTableName = "SUB_TABLE";
+
+ // Add scan metrics holder to main context
+ mainContext.getSlowestScanMetricsQueue().add(createScanMetricsHolder(mainTableName, 100L));
+
+ // Create a sub-context (e.g., for a subquery) - this is a leaf node
+ StatementContext subContext1 = createStatementContext(5);
+ subContext1.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName, 200L));
+ mainContext.addSubStatementContext(subContext1);
+
+ List<List<JsonObject>> result = mainContext.getTopNSlowestScanMetrics();
+
+ // Should have 1 result with the full path: main+sub (100+200=300)
+ // The algorithm only returns paths to leaf nodes
+ assertEquals(1, result.size());
+
+ // Verify the path contains both scan metrics holders
+ List<JsonObject> path = result.get(0);
+ assertEquals(2, path.size());
+
+ // Verify the metrics
+ assertEquals(mainTableName, path.get(0).get("table").getAsString());
+ assertEquals(100L, path.get(0).get("n").getAsLong());
+ assertEquals(subTableName, path.get(1).get("table").getAsString());
+ assertEquals(200L, path.get(1).get("n").getAsLong());
+ }
+
+ /**
+ * Test with multiple levels of sub-contexts (nested subqueries).
+ */
+ @Test
+ public void testWithNestedSubContexts() {
+ StatementContext mainContext = createStatementContext(10);
+ String mainTableName = "MAIN";
+ String subTableName1 = "SUB1";
+ String subTableName2 = "SUB2";
+
+ // Top level statement context
+ mainContext.getSlowestScanMetricsQueue().add(createScanMetricsHolder(mainTableName, 50L));
+
+ // First level sub-contexts
+ StatementContext subContext11 = createStatementContext(10);
+ subContext11.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName1, 100L));
+ StatementContext subContext12 = createStatementContext(10);
+ subContext12.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName1, 200L));
+
+ // Second level sub-context
+ StatementContext subContext2 = createStatementContext(10);
+ subContext2.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName2, 150L));
+
+ subContext11.addSubStatementContext(subContext2);
+ mainContext.addSubStatementContext(subContext11);
+ mainContext.addSubStatementContext(subContext12);
+
+ List<List<JsonObject>> result = mainContext.getTopNSlowestScanMetrics();
+
+ // Should have multiple paths through the tree
+ // The longest path would be: main(50) + sub1(100) + sub2(150) = 300ms
+ assertEquals(2, result.size());
+
+ // The slowest path should have all three scan metrics
+ List<JsonObject> slowestPath = result.get(0);
+ assertEquals(3, slowestPath.size());
+ assertEquals(mainTableName, slowestPath.get(0).get("table").getAsString());
+ assertEquals(50L, slowestPath.get(0).get("n").getAsLong());
+ assertEquals(subTableName1, slowestPath.get(1).get("table").getAsString());
+ assertEquals(100L, slowestPath.get(1).get("n").getAsLong());
+ assertEquals(subTableName2, slowestPath.get(2).get("table").getAsString());
+ assertEquals(150L, slowestPath.get(2).get("n").getAsLong());
+
+ // The second slowest path should have two scan metrics
+ List<JsonObject> secondSlowestPath = result.get(1);
+ assertEquals(2, secondSlowestPath.size());
+ assertEquals(mainTableName, secondSlowestPath.get(0).get("table").getAsString());
+ assertEquals(50L, secondSlowestPath.get(0).get("n").getAsLong());
+ assertEquals(subTableName1, secondSlowestPath.get(1).get("table").getAsString());
+ assertEquals(200L, secondSlowestPath.get(1).get("n").getAsLong());
+ }
+
+ /**
+ * Test with empty sub-context at in between level.
+ */
+ @Test
+ public void testWithEmptySubContextAtInBetweenLevel() {
+ StatementContext mainContext = createStatementContext(2);
+ String mainTableName = "MAIN";
+ String subTableName11 = "SUB11";
+ String subTableName21 = "SUB21";
+ String subTableName22 = "SUB22";
+
+ // Main statement context
+ mainContext.getSlowestScanMetricsQueue().add(createScanMetricsHolder(mainTableName, 100L));
+
+ // First level sub-contexts
+ StatementContext subContext11 = createStatementContext(2);
+ subContext11.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName11, 200L));
+ subContext11.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName11, 250L));
+ // Empty sub-context at in between level
+ StatementContext subContext12 = createStatementContext(2);
+
+ // Second level sub-contexts
+ StatementContext subContext21 = createStatementContext(2);
+ subContext21.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName21, 150L));
+ StatementContext subContext22 = createStatementContext(2);
+ subContext22.getSlowestScanMetricsQueue().add(createScanMetricsHolder(subTableName22, 600L));
+
+ mainContext.addSubStatementContext(subContext11);
+ mainContext.addSubStatementContext(subContext12);
+ subContext11.addSubStatementContext(subContext21);
+ subContext12.addSubStatementContext(subContext22);
+
+ List<List<JsonObject>> result = mainContext.getTopNSlowestScanMetrics();
+
+ assertEquals(2, result.size());
+
+ // Verify the results
+ List<JsonObject> slowestPath = result.get(0);
+ assertEquals(2, slowestPath.size());
+ assertEquals(mainTableName, slowestPath.get(0).get("table").getAsString());
+ assertEquals(100L, slowestPath.get(0).get("n").getAsLong());
+ assertEquals(subTableName22, slowestPath.get(1).get("table").getAsString());
+ assertEquals(600L, slowestPath.get(1).get("n").getAsLong());
+
+ // The second slowest path should have three scan metrics
+ List<JsonObject> secondSlowestPath = result.get(1);
+ assertEquals(3, secondSlowestPath.size());
+ assertEquals(mainTableName, secondSlowestPath.get(0).get("table").getAsString());
+ assertEquals(100L, secondSlowestPath.get(0).get("n").getAsLong());
+ assertEquals(subTableName11, secondSlowestPath.get(1).get("table").getAsString());
+ assertEquals(250L, secondSlowestPath.get(1).get("n").getAsLong());
+ assertEquals(subTableName21, secondSlowestPath.get(2).get("table").getAsString());
+ assertEquals(150L, secondSlowestPath.get(2).get("n").getAsLong());
+ }
+
+ /**
+ * Test with multiple ScanMetricsHolders having the same millisBetweenNexts, verifying that
+ * eviction works correctly when the underlying TopNTreeMultiMap is full. The TopNTreeMultiMap
+ * uses a reverse comparator (larger values sorted first), so it keeps the top N highest values
+ * and evicts from the smallest values when full.
+ */
+ @Test
+ public void testEvictionWithDuplicateMillisBetweenNexts() {
+ // Create context with capacity for 5 scan metrics
+ int topN = 5;
+ StatementContext context = createStatementContext(topN);
+
+ // Add scan metrics where two of them have the same millisBetweenNexts (200L)
+ // With reverse comparator, these are sorted as: 500L, 400L, 300L, 200L, 200L, 100L
+ // When capacity is 5, the smallest values (100L) will be at the "end" of the sorted map
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE1", 500L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE2", 400L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE3", 300L));
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE4", 200L)); // First with
+ // 200L
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE5", 200L)); // Second
+ // with
+ // 200L
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE6", 100L));
+
+ // Verify we have 5 scan metrics (100L should already be rejected)
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+ assertEquals(5, result.size());
+
+ // Count how many ScanMetricsHolders with 200L are present
+ long count200L = result.stream().flatMap(List::stream)
+ .filter(json -> Long.valueOf(200L).equals(json.get("n").getAsLong())).count();
+
+ // Both 200L entries should still be there since 100L is the smallest
+ assertEquals(2, count200L);
+
+ // Now add a new ScanMetricsHolder with 250L which should cause eviction
+ // Since 250L > 200L (smallest in map), and map is full, one of the 200L entries should be
+ // evicted
+ context.getSlowestScanMetricsQueue().add(createScanMetricsHolder("TABLE7", 250L));
+
+ // Get the updated results
+ result = context.getTopNSlowestScanMetrics();
+
+ // Should still have 5 entries total
+ assertEquals(5, result.size());
+
+ // Verify the new ScanMetricsHolder with 250L is present
+ boolean has250L = result.stream().flatMap(List::stream)
+ .anyMatch(json -> Long.valueOf(250L).equals(json.get("n").getAsLong()));
+ assertTrue(has250L);
+
+ // Verify that only ONE of the two ScanMetricsHolders with 200L remains
+ // (one should have been evicted as it was the last value of the smallest key)
+ count200L = result.stream().flatMap(List::stream)
+ .filter(json -> Long.valueOf(200L).equals(json.get("n").getAsLong())).count();
+ assertEquals(1, count200L);
+
+ // Verify the larger values are still present
+ boolean has500L = result.stream().flatMap(List::stream)
+ .anyMatch(json -> Long.valueOf(500L).equals(json.get("n").getAsLong()));
+ assertTrue(has500L);
+
+ boolean has400L = result.stream().flatMap(List::stream)
+ .anyMatch(json -> Long.valueOf(400L).equals(json.get("n").getAsLong()));
+ assertTrue(has400L);
+
+ boolean has300L = result.stream().flatMap(List::stream)
+ .anyMatch(json -> Long.valueOf(300L).equals(json.get("n").getAsLong()));
+ assertTrue(has300L);
+
+ // Verify all remaining entries have millisBetweenNexts >= 200L (the minimum)
+ long minMillis = result.stream().flatMap(List::stream)
+ .mapToLong(json -> json.get("n").getAsLong()).min().orElse(Long.MAX_VALUE);
+ assertEquals(200L, minMillis);
+ }
+
+ /**
+ * Test with empty slowest scan metrics queue.
+ */
+ @Test
+ public void testWithEmptyQueue() {
+ StatementContext context = createStatementContext(5);
+ // Don't add any scan metrics
+
+ List<List<JsonObject>> result = context.getTopNSlowestScanMetrics();
+
+ // Should return empty result since the slowest scan metrics queue is empty
+ assertEquals(1, result.size());
+ assertEquals(0, result.get(0).size());
+ }
+
+ /**
+ * Helper method to create a StatementContext with a specific slowestScanMetricsCount.
+ */
+ private StatementContext createStatementContext(int slowestScanMetricsCount) {
+ // Mock PhoenixConnection
+ PhoenixConnection mockConnection = mock(PhoenixConnection.class);
+ ConnectionQueryServices mockQueryServices = mock(ConnectionQueryServices.class);
+
+ // Set up the configuration properties
+ Map<String, String> props = new HashMap<>();
+ props.put(QueryServices.SLOWEST_SCAN_METRICS_COUNT, String.valueOf(slowestScanMetricsCount));
+ ReadOnlyProps readOnlyProps = new ReadOnlyProps(props);
+
+ when(mockConnection.getQueryServices()).thenReturn(mockQueryServices);
+ when(mockQueryServices.getProps()).thenReturn(readOnlyProps);
+ when(mockConnection.getSlowestScanMetricsCount()).thenReturn(slowestScanMetricsCount);
+ when(mockConnection.isRequestLevelMetricsEnabled()).thenReturn(true);
+ when(mockConnection.getLogLevel()).thenReturn(LogLevel.INFO);
+
+ // Mock PhoenixStatement
+ PhoenixStatement mockStatement = mock(PhoenixStatement.class);
+ when(mockStatement.getConnection()).thenReturn(mockConnection);
+ when(mockStatement.getParameters()).thenReturn(Collections.emptyList());
+
+ // Create StatementContext
+ return new StatementContext(mockStatement, new Scan());
+ }
+
+ /**
+ * Helper method to create a ScanMetricsHolder with a specific table name and millis between
+ * nexts.
+ */
+ private ScanMetricsHolder createScanMetricsHolder(String tableName, long millisBetweenNexts) {
+ ReadMetricQueue readMetricQueue = new ReadMetricQueue(true, LogLevel.DEBUG);
+ ScanMetricsHolder holder =
+ ScanMetricsHolder.getInstance(readMetricQueue, tableName, new Scan(), LogLevel.DEBUG);
+ Map<String, Long> scanMetrics = new HashMap<>();
+ scanMetrics.put(ScanMetrics.MILLIS_BETWEEN_NEXTS_METRIC_NAME, millisBetweenNexts);
+ holder.setScanMetricMap(scanMetrics);
+ return holder;
+ }
+}
diff --git a/phoenix-hbase-compat-2.5.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java b/phoenix-hbase-compat-2.5.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
index 93dcc0cf93..fd7641f033 100644
--- a/phoenix-hbase-compat-2.5.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
+++ b/phoenix-hbase-compat-2.5.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
@@ -17,10 +17,18 @@
*/
package org.apache.phoenix.compat.hbase;
-import java.util.Map;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
public class CompatScanMetrics {
+ public static final String FS_READ_TIME_METRIC_NAME = "FS_READ_TIME";
+ public static final String BYTES_READ_FROM_FS_METRIC_NAME = "BYTES_READ_FROM_FS";
+ public static final String BYTES_READ_FROM_MEMSTORE_METRIC_NAME = "BYTES_READ_FROM_MEMSTORE";
+ public static final String BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME =
+ "BYTES_READ_FROM_BLOCK_CACHE";
+ public static final String BLOCK_READ_OPS_COUNT_METRIC_NAME = "BLOCK_READ_OPS_COUNT";
+ public static final String RPC_SCAN_PROCESSING_TIME_METRIC_NAME = "RPC_SCAN_PROCESSING_TIME";
+ public static final String RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME = "RPC_SCAN_QUEUE_WAIT_TIME";
+
private CompatScanMetrics() {
// Not to be instantiated
}
@@ -29,51 +37,23 @@ public class CompatScanMetrics {
return false;
}
- public static Long getFsReadTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getFsReadTime(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromFs(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromFs(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromMemstore(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromMemstore(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromBlockCache(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromBlockCache(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBlockReadOpsCount(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBlockReadOpsCount(ScanMetrics scanMetrics) {
return 0L;
}
-
- public static Long getRpcScanProcessingTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
- public static Long getRpcScanQueueWaitTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
}
diff --git a/phoenix-hbase-compat-2.5.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java b/phoenix-hbase-compat-2.5.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
index 93dcc0cf93..fd7641f033 100644
--- a/phoenix-hbase-compat-2.5.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
+++ b/phoenix-hbase-compat-2.5.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
@@ -17,10 +17,18 @@
*/
package org.apache.phoenix.compat.hbase;
-import java.util.Map;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
public class CompatScanMetrics {
+ public static final String FS_READ_TIME_METRIC_NAME = "FS_READ_TIME";
+ public static final String BYTES_READ_FROM_FS_METRIC_NAME = "BYTES_READ_FROM_FS";
+ public static final String BYTES_READ_FROM_MEMSTORE_METRIC_NAME = "BYTES_READ_FROM_MEMSTORE";
+ public static final String BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME =
+ "BYTES_READ_FROM_BLOCK_CACHE";
+ public static final String BLOCK_READ_OPS_COUNT_METRIC_NAME = "BLOCK_READ_OPS_COUNT";
+ public static final String RPC_SCAN_PROCESSING_TIME_METRIC_NAME = "RPC_SCAN_PROCESSING_TIME";
+ public static final String RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME = "RPC_SCAN_QUEUE_WAIT_TIME";
+
private CompatScanMetrics() {
// Not to be instantiated
}
@@ -29,51 +37,23 @@ public class CompatScanMetrics {
return false;
}
- public static Long getFsReadTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getFsReadTime(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromFs(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromFs(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromMemstore(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromMemstore(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromBlockCache(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromBlockCache(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBlockReadOpsCount(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBlockReadOpsCount(ScanMetrics scanMetrics) {
return 0L;
}
-
- public static Long getRpcScanProcessingTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
- public static Long getRpcScanQueueWaitTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
}
diff --git a/phoenix-hbase-compat-2.6.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java b/phoenix-hbase-compat-2.6.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
index 93dcc0cf93..fd7641f033 100644
--- a/phoenix-hbase-compat-2.6.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
+++ b/phoenix-hbase-compat-2.6.0/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
@@ -17,10 +17,18 @@
*/
package org.apache.phoenix.compat.hbase;
-import java.util.Map;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
public class CompatScanMetrics {
+ public static final String FS_READ_TIME_METRIC_NAME = "FS_READ_TIME";
+ public static final String BYTES_READ_FROM_FS_METRIC_NAME = "BYTES_READ_FROM_FS";
+ public static final String BYTES_READ_FROM_MEMSTORE_METRIC_NAME = "BYTES_READ_FROM_MEMSTORE";
+ public static final String BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME =
+ "BYTES_READ_FROM_BLOCK_CACHE";
+ public static final String BLOCK_READ_OPS_COUNT_METRIC_NAME = "BLOCK_READ_OPS_COUNT";
+ public static final String RPC_SCAN_PROCESSING_TIME_METRIC_NAME = "RPC_SCAN_PROCESSING_TIME";
+ public static final String RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME = "RPC_SCAN_QUEUE_WAIT_TIME";
+
private CompatScanMetrics() {
// Not to be instantiated
}
@@ -29,51 +37,23 @@ public class CompatScanMetrics {
return false;
}
- public static Long getFsReadTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getFsReadTime(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromFs(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromFs(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromMemstore(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromMemstore(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBytesReadFromBlockCache(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBytesReadFromBlockCache(ScanMetrics scanMetrics) {
return 0L;
}
- public static Long getBlockReadOpsCount(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
public static Long getBlockReadOpsCount(ScanMetrics scanMetrics) {
return 0L;
}
-
- public static Long getRpcScanProcessingTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
-
- public static Long getRpcScanQueueWaitTime(Map<String, Long> scanMetrics) {
- return 0L;
- }
}
diff --git a/phoenix-hbase-compat-2.6.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java b/phoenix-hbase-compat-2.6.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
index 4026ceed05..9a8e2f8cd6 100644
--- a/phoenix-hbase-compat-2.6.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
+++ b/phoenix-hbase-compat-2.6.4/src/main/java/org/apache/phoenix/compat/hbase/CompatScanMetrics.java
@@ -17,18 +17,25 @@
*/
package org.apache.phoenix.compat.hbase;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.BYTES_READ_FROM_FS_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.BYTES_READ_FROM_MEMSTORE_METRIC_NAME;
-import static org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME;
-
-import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.hadoop.hbase.client.metrics.ScanMetrics;
import org.apache.hadoop.hbase.client.metrics.ServerSideScanMetrics;
public class CompatScanMetrics {
+ public static final String FS_READ_TIME_METRIC_NAME =
+ ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME;
+ public static final String BYTES_READ_FROM_FS_METRIC_NAME =
+ ServerSideScanMetrics.BYTES_READ_FROM_FS_METRIC_NAME;
+ public static final String BYTES_READ_FROM_MEMSTORE_METRIC_NAME =
+ ServerSideScanMetrics.BYTES_READ_FROM_MEMSTORE_METRIC_NAME;
+ public static final String BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME =
+ ServerSideScanMetrics.BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME;
+ public static final String BLOCK_READ_OPS_COUNT_METRIC_NAME =
+ ServerSideScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME;
+ public static final String RPC_SCAN_PROCESSING_TIME_METRIC_NAME =
+ ServerSideScanMetrics.RPC_SCAN_PROCESSING_TIME_METRIC_NAME;
+ public static final String RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME =
+ ServerSideScanMetrics.RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME;
private CompatScanMetrics() {
// Not to be instantiated
@@ -38,53 +45,25 @@ public class CompatScanMetrics {
return true;
}
- public static Long getFsReadTime(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME, 0L);
- }
-
public static Long getFsReadTime(ScanMetrics scanMetrics) {
- return getCounterValue(scanMetrics, FS_READ_TIME_METRIC_NAME);
- }
-
- public static Long getBytesReadFromFs(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.BYTES_READ_FROM_FS_METRIC_NAME, 0L);
+ return getCounterValue(scanMetrics, ServerSideScanMetrics.FS_READ_TIME_METRIC_NAME);
}
public static Long getBytesReadFromFs(ScanMetrics scanMetrics) {
- return getCounterValue(scanMetrics, BYTES_READ_FROM_FS_METRIC_NAME);
- }
-
- public static Long getBytesReadFromMemstore(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.BYTES_READ_FROM_MEMSTORE_METRIC_NAME, 0L);
+ return getCounterValue(scanMetrics, ServerSideScanMetrics.BYTES_READ_FROM_FS_METRIC_NAME);
}
public static Long getBytesReadFromMemstore(ScanMetrics scanMetrics) {
- return getCounterValue(scanMetrics, BYTES_READ_FROM_MEMSTORE_METRIC_NAME);
- }
-
- public static Long getBytesReadFromBlockCache(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME,
- 0L);
+ return getCounterValue(scanMetrics, ServerSideScanMetrics.BYTES_READ_FROM_MEMSTORE_METRIC_NAME);
}
public static Long getBytesReadFromBlockCache(ScanMetrics scanMetrics) {
- return getCounterValue(scanMetrics, BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME);
- }
-
- public static Long getBlockReadOpsCount(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME, 0L);
+ return getCounterValue(scanMetrics,
+ ServerSideScanMetrics.BYTES_READ_FROM_BLOCK_CACHE_METRIC_NAME);
}
public static Long getBlockReadOpsCount(ScanMetrics scanMetrics) {
- return getCounterValue(scanMetrics, BLOCK_READ_OPS_COUNT_METRIC_NAME);
- }
-
- public static Long getRpcScanProcessingTime(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.RPC_SCAN_PROCESSING_TIME_METRIC_NAME, 0L);
- }
-
- public static Long getRpcScanQueueWaitTime(Map<String, Long> scanMetrics) {
- return scanMetrics.getOrDefault(ServerSideScanMetrics.RPC_SCAN_QUEUE_WAIT_TIME_METRIC_NAME, 0L);
+ return getCounterValue(scanMetrics, ServerSideScanMetrics.BLOCK_READ_OPS_COUNT_METRIC_NAME);
}
private static Long getCounterValue(ScanMetrics scanMetrics, String metricName) {