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 2015/06/27 01:29:05 UTC
[3/4] phoenix git commit: PHOENIX-1819 Build a framework to capture
and report phoenix client side request level metrics
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
index 857a952..57fa25a 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/HashJoinPlan.java
@@ -17,6 +17,7 @@
*/
package org.apache.phoenix.execute;
+import static org.apache.phoenix.monitoring.TaskExecutionMetricsHolder.NO_OP_INSTANCE;
import static org.apache.phoenix.util.LogUtil.addCustomAnnotations;
import java.sql.SQLException;
@@ -54,6 +55,7 @@ import org.apache.phoenix.jdbc.PhoenixConnection;
import org.apache.phoenix.job.JobManager.JobCallable;
import org.apache.phoenix.join.HashCacheClient;
import org.apache.phoenix.join.HashJoinInfo;
+import org.apache.phoenix.monitoring.TaskExecutionMetricsHolder;
import org.apache.phoenix.parse.FilterableStatement;
import org.apache.phoenix.parse.ParseNode;
import org.apache.phoenix.parse.SQLParser;
@@ -140,6 +142,11 @@ public class HashJoinPlan extends DelegateQueryPlan {
public Object getJobId() {
return HashJoinPlan.this;
}
+
+ @Override
+ public TaskExecutionMetricsHolder getTaskExecutionMetric() {
+ return NO_OP_INSTANCE;
+ }
}));
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
index 9ffa135..0de7aa3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/MutationState.java
@@ -17,6 +17,10 @@
*/
package org.apache.phoenix.execute;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BATCH_SIZE;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_BYTES;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_COMMIT_TIME;
+
import java.io.IOException;
import java.sql.SQLException;
import java.util.Arrays;
@@ -39,7 +43,11 @@ import org.apache.phoenix.index.IndexMaintainer;
import org.apache.phoenix.index.IndexMetaDataCacheClient;
import org.apache.phoenix.index.PhoenixIndexCodec;
import org.apache.phoenix.jdbc.PhoenixConnection;
-import org.apache.phoenix.monitoring.PhoenixMetrics;
+import org.apache.phoenix.monitoring.GlobalClientMetrics;
+import org.apache.phoenix.monitoring.MutationMetricQueue;
+import org.apache.phoenix.monitoring.MutationMetricQueue.MutationMetric;
+import org.apache.phoenix.monitoring.MutationMetricQueue.NoOpMutationMetricsQueue;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.query.QueryConstants;
import org.apache.phoenix.schema.IllegalDataException;
import org.apache.phoenix.schema.MetaDataClient;
@@ -65,9 +73,6 @@ import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sun.istack.NotNull;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.MUTATION_BYTES;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.MUTATION_BATCH_SIZE;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.MUTATION_COMMIT_TIME;
/**
*
@@ -85,11 +90,17 @@ public class MutationState implements SQLCloseable {
private final Map<TableRef, Map<ImmutableBytesPtr,RowMutationState>> mutations;
private long sizeOffset;
private int numRows = 0;
+ private final MutationMetricQueue mutationMetricQueue;
+ private ReadMetricQueue readMetricQueue;
- MutationState(long maxSize, PhoenixConnection connection, Map<TableRef, Map<ImmutableBytesPtr,RowMutationState>> mutations) {
+ MutationState(long maxSize, PhoenixConnection connection,
+ Map<TableRef, Map<ImmutableBytesPtr, RowMutationState>> mutations) {
this.maxSize = maxSize;
this.connection = connection;
this.mutations = mutations;
+ boolean isMetricsEnabled = connection.isRequestLevelMetricsEnabled();
+ this.mutationMetricQueue = isMetricsEnabled ? new MutationMetricQueue()
+ : NoOpMutationMetricsQueue.NO_OP_MUTATION_METRICS_QUEUE;
}
public MutationState(long maxSize, PhoenixConnection connection) {
@@ -108,6 +119,12 @@ public class MutationState implements SQLCloseable {
throwIfTooBig();
}
+ public static MutationState emptyMutationState(long maxSize, PhoenixConnection connection) {
+ MutationState state = new MutationState(maxSize, connection, Collections.<TableRef, Map<ImmutableBytesPtr,RowMutationState>>emptyMap());
+ state.sizeOffset = 0;
+ return state;
+ }
+
private void throwIfTooBig() {
if (numRows > maxSize) {
// TODO: throw SQLException ?
@@ -120,17 +137,18 @@ public class MutationState implements SQLCloseable {
}
/**
- * Combine a newer mutation with this one, where in the event of overlaps,
- * the newer one will take precedence.
- * @param newMutation the newer mutation
+ * Combine a newer mutation with this one, where in the event of overlaps, the newer one will take precedence.
+ * Combine any metrics collected for the newer mutation.
+ *
+ * @param newMutationState the newer mutation state
*/
- public void join(MutationState newMutation) {
- if (this == newMutation) { // Doesn't make sense
+ public void join(MutationState newMutationState) {
+ if (this == newMutationState) { // Doesn't make sense
return;
}
- this.sizeOffset += newMutation.sizeOffset;
+ this.sizeOffset += newMutationState.sizeOffset;
// Merge newMutation with this one, keeping state from newMutation for any overlaps
- for (Map.Entry<TableRef, Map<ImmutableBytesPtr,RowMutationState>> entry : newMutation.mutations.entrySet()) {
+ for (Map.Entry<TableRef, Map<ImmutableBytesPtr,RowMutationState>> entry : newMutationState.mutations.entrySet()) {
// Replace existing entries for the table with new entries
TableRef tableRef = entry.getKey();
PTable table = tableRef.getTable();
@@ -168,6 +186,12 @@ public class MutationState implements SQLCloseable {
}
}
}
+ mutationMetricQueue.combineMetricQueues(newMutationState.mutationMetricQueue);
+ if (readMetricQueue == null) {
+ readMetricQueue = newMutationState.readMetricQueue;
+ } else if (readMetricQueue != null && newMutationState.readMetricQueue != null) {
+ readMetricQueue.combineReadMetrics(newMutationState.readMetricQueue);
+ }
throwIfTooBig();
}
@@ -332,18 +356,15 @@ public class MutationState implements SQLCloseable {
return timeStamps;
}
- private static void logMutationSize(HTableInterface htable, List<Mutation> mutations, PhoenixConnection connection) {
+ private static long calculateMutationSize(List<Mutation> mutations) {
long byteSize = 0;
- int keyValueCount = 0;
- if (PhoenixMetrics.isMetricsEnabled() || logger.isDebugEnabled()) {
+ if (GlobalClientMetrics.isMetricsEnabled()) {
for (Mutation mutation : mutations) {
byteSize += mutation.heapSize();
}
- MUTATION_BYTES.update(byteSize);
- if (logger.isDebugEnabled()) {
- logger.debug(LogUtil.addCustomAnnotations("Sending " + mutations.size() + " mutations for " + Bytes.toString(htable.getTableName()) + " with " + keyValueCount + " key values of total size " + byteSize + " bytes", connection));
- }
}
+ GLOBAL_MUTATION_BYTES.update(byteSize);
+ return byteSize;
}
@SuppressWarnings("deprecation")
@@ -352,126 +373,134 @@ public class MutationState implements SQLCloseable {
byte[] tenantId = connection.getTenantId() == null ? null : connection.getTenantId().getBytes();
long[] serverTimeStamps = validate();
Iterator<Map.Entry<TableRef, Map<ImmutableBytesPtr,RowMutationState>>> iterator = this.mutations.entrySet().iterator();
-
// add tracing for this operation
- TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables");
- Span span = trace.getSpan();
- while (iterator.hasNext()) {
- Map.Entry<TableRef, Map<ImmutableBytesPtr,RowMutationState>> entry = iterator.next();
- Map<ImmutableBytesPtr,RowMutationState> valuesMap = entry.getValue();
- TableRef tableRef = entry.getKey();
- PTable table = tableRef.getTable();
- table.getIndexMaintainers(tempPtr, connection);
- boolean hasIndexMaintainers = tempPtr.getLength() > 0;
- boolean isDataTable = true;
- long serverTimestamp = serverTimeStamps[i++];
- Iterator<Pair<byte[],List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, serverTimestamp, false);
- while (mutationsIterator.hasNext()) {
- Pair<byte[],List<Mutation>> pair = mutationsIterator.next();
- byte[] htableName = pair.getFirst();
- List<Mutation> mutations = pair.getSecond();
-
- //create a span per target table
- //TODO maybe we can be smarter about the table name to string here?
- Span child = Tracing.child(span,"Writing mutation batch for table: "+Bytes.toString(htableName));
+ try (TraceScope trace = Tracing.startNewSpan(connection, "Committing mutations to tables")) {
+ Span span = trace.getSpan();
+ while (iterator.hasNext()) {
+ Map.Entry<TableRef, Map<ImmutableBytesPtr,RowMutationState>> entry = iterator.next();
+ // at this point we are going through mutations for each table
- int retryCount = 0;
- boolean shouldRetry = false;
- do {
- ServerCache cache = null;
- if (hasIndexMaintainers && isDataTable) {
- byte[] attribValue = null;
- byte[] uuidValue;
- if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, tempPtr.getLength())) {
- IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
- cache = client.addIndexMetadataCache(mutations, tempPtr);
- child.addTimelineAnnotation("Updated index metadata cache");
- uuidValue = cache.getId();
- // If we haven't retried yet, retry for this case only, as it's possible that
- // a split will occur after we send the index metadata cache to all known
- // region servers.
- shouldRetry = true;
- } else {
- attribValue = ByteUtil.copyKeyBytesIfNecessary(tempPtr);
- uuidValue = ServerCacheClient.generateId();
- }
- // Either set the UUID to be able to access the index metadata from the cache
- // or set the index metadata directly on the Mutation
- for (Mutation mutation : mutations) {
- if (tenantId != null) {
- mutation.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
- }
- mutation.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
- if (attribValue != null) {
- mutation.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
- }
- }
- }
-
- SQLException sqlE = null;
- HTableInterface hTable = connection.getQueryServices().getTable(htableName);
- try {
- logMutationSize(hTable, mutations, connection);
- MUTATION_BATCH_SIZE.update(mutations.size());
- long startTime = System.currentTimeMillis();
- child.addTimelineAnnotation("Attempt " + retryCount);
- hTable.batch(mutations);
- child.stop();
- long duration = System.currentTimeMillis() - startTime;
- MUTATION_COMMIT_TIME.update(duration);
- shouldRetry = false;
- if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("Total time for batch call of " + mutations.size() + " mutations into " + table.getName().getString() + ": " + duration + " ms", connection));
- } catch (Exception e) {
- SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
- if (inferredE != null) {
- if (shouldRetry && retryCount == 0 && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND.getErrorCode()) {
- // Swallow this exception once, as it's possible that we split after sending the index metadata
- // and one of the region servers doesn't have it. This will cause it to have it the next go around.
- // If it fails again, we don't retry.
- String msg = "Swallowing exception and retrying after clearing meta cache on connection. " + inferredE;
- logger.warn(LogUtil.addCustomAnnotations(msg, connection));
- connection.getQueryServices().clearTableRegionCache(htableName);
+ Map<ImmutableBytesPtr,RowMutationState> valuesMap = entry.getValue();
+ // above is mutations for a table where the first part is the row key and the second part is column values.
- // add a new child span as this one failed
- child.addTimelineAnnotation(msg);
- child.stop();
- child = Tracing.child(span,"Failed batch, attempting retry");
+ TableRef tableRef = entry.getKey();
+ PTable table = tableRef.getTable();
+ table.getIndexMaintainers(tempPtr, connection);
+ boolean hasIndexMaintainers = tempPtr.getLength() > 0;
+ boolean isDataTable = true;
+ long serverTimestamp = serverTimeStamps[i++];
+ Iterator<Pair<byte[],List<Mutation>>> mutationsIterator = addRowMutations(tableRef, valuesMap, serverTimestamp, false);
+ // above returns an iterator of pair where the first
+ while (mutationsIterator.hasNext()) {
+ Pair<byte[],List<Mutation>> pair = mutationsIterator.next();
+ byte[] htableName = pair.getFirst();
+ List<Mutation> mutations = pair.getSecond();
- continue;
+ //create a span per target table
+ //TODO maybe we can be smarter about the table name to string here?
+ Span child = Tracing.child(span,"Writing mutation batch for table: "+Bytes.toString(htableName));
+
+ int retryCount = 0;
+ boolean shouldRetry = false;
+ do {
+ ServerCache cache = null;
+ if (hasIndexMaintainers && isDataTable) {
+ byte[] attribValue = null;
+ byte[] uuidValue;
+ if (IndexMetaDataCacheClient.useIndexMetadataCache(connection, mutations, tempPtr.getLength())) {
+ IndexMetaDataCacheClient client = new IndexMetaDataCacheClient(connection, tableRef);
+ cache = client.addIndexMetadataCache(mutations, tempPtr);
+ child.addTimelineAnnotation("Updated index metadata cache");
+ uuidValue = cache.getId();
+ // If we haven't retried yet, retry for this case only, as it's possible that
+ // a split will occur after we send the index metadata cache to all known
+ // region servers.
+ shouldRetry = true;
+ } else {
+ attribValue = ByteUtil.copyKeyBytesIfNecessary(tempPtr);
+ uuidValue = ServerCacheClient.generateId();
+ }
+ // Either set the UUID to be able to access the index metadata from the cache
+ // or set the index metadata directly on the Mutation
+ for (Mutation mutation : mutations) {
+ if (tenantId != null) {
+ mutation.setAttribute(PhoenixRuntime.TENANT_ID_ATTRIB, tenantId);
+ }
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_UUID, uuidValue);
+ if (attribValue != null) {
+ mutation.setAttribute(PhoenixIndexCodec.INDEX_MD, attribValue);
+ }
}
- e = inferredE;
}
- sqlE = new CommitException(e, getUncommittedSattementIndexes());
- } finally {
+
+ SQLException sqlE = null;
+ HTableInterface hTable = connection.getQueryServices().getTable(htableName);
try {
- hTable.close();
- } catch (IOException e) {
- if (sqlE != null) {
- sqlE.setNextException(ServerUtil.parseServerException(e));
- } else {
- sqlE = ServerUtil.parseServerException(e);
+ long numMutations = mutations.size();
+ GLOBAL_MUTATION_BATCH_SIZE.update(numMutations);
+
+ long startTime = System.currentTimeMillis();
+ child.addTimelineAnnotation("Attempt " + retryCount);
+ hTable.batch(mutations);
+ child.stop();
+ shouldRetry = false;
+ long mutationCommitTime = System.currentTimeMillis() - startTime;
+ GLOBAL_MUTATION_COMMIT_TIME.update(mutationCommitTime);
+
+ long mutationSizeBytes = calculateMutationSize(mutations);
+ MutationMetric mutationsMetric = new MutationMetric(numMutations, mutationSizeBytes, mutationCommitTime);
+ mutationMetricQueue.addMetricsForTable(Bytes.toString(htableName), mutationsMetric);
+ } catch (Exception e) {
+ SQLException inferredE = ServerUtil.parseServerExceptionOrNull(e);
+ if (inferredE != null) {
+ if (shouldRetry && retryCount == 0 && inferredE.getErrorCode() == SQLExceptionCode.INDEX_METADATA_NOT_FOUND.getErrorCode()) {
+ // Swallow this exception once, as it's possible that we split after sending the index metadata
+ // and one of the region servers doesn't have it. This will cause it to have it the next go around.
+ // If it fails again, we don't retry.
+ String msg = "Swallowing exception and retrying after clearing meta cache on connection. " + inferredE;
+ logger.warn(LogUtil.addCustomAnnotations(msg, connection));
+ connection.getQueryServices().clearTableRegionCache(htableName);
+
+ // add a new child span as this one failed
+ child.addTimelineAnnotation(msg);
+ child.stop();
+ child = Tracing.child(span,"Failed batch, attempting retry");
+
+ continue;
+ }
+ e = inferredE;
}
+ sqlE = new CommitException(e, getUncommittedStatementIndexes());
} finally {
try {
- if (cache != null) {
- cache.close();
+ hTable.close();
+ } catch (IOException e) {
+ if (sqlE != null) {
+ sqlE.setNextException(ServerUtil.parseServerException(e));
+ } else {
+ sqlE = ServerUtil.parseServerException(e);
}
} finally {
- if (sqlE != null) {
- throw sqlE;
+ try {
+ if (cache != null) {
+ cache.close();
+ }
+ } finally {
+ if (sqlE != null) {
+ throw sqlE;
+ }
}
}
}
- }
- } while (shouldRetry && retryCount++ < 1);
- isDataTable = false;
- }
- if (tableRef.getTable().getType() != PTableType.INDEX) {
- numRows -= entry.getValue().size();
+ } while (shouldRetry && retryCount++ < 1);
+ isDataTable = false;
+ }
+ if (tableRef.getTable().getType() != PTableType.INDEX) {
+ numRows -= entry.getValue().size();
+ }
+ iterator.remove(); // Remove batches as we process them
}
- iterator.remove(); // Remove batches as we process them
}
- trace.close();
assert(numRows==0);
assert(this.mutations.isEmpty());
}
@@ -481,7 +510,7 @@ public class MutationState implements SQLCloseable {
numRows = 0;
}
- private int[] getUncommittedSattementIndexes() {
+ private int[] getUncommittedStatementIndexes() {
int[] result = new int[0];
for (Map<ImmutableBytesPtr, RowMutationState> rowMutations : mutations.values()) {
for (RowMutationState rowMutationState : rowMutations.values()) {
@@ -533,12 +562,23 @@ public class MutationState implements SQLCloseable {
int[] getStatementIndexes() {
return statementIndexes;
}
-
+
void join(RowMutationState newRow) {
getColumnValues().putAll(newRow.getColumnValues());
statementIndexes = joinSortedIntArrays(statementIndexes, newRow.getStatementIndexes());
}
-
+ }
+
+ public ReadMetricQueue getReadMetricQueue() {
+ return readMetricQueue;
+ }
+ public void setReadMetricQueue(ReadMetricQueue readMetricQueue) {
+ this.readMetricQueue = readMetricQueue;
}
+
+ public MutationMetricQueue getMutationMetricQueue() {
+ return mutationMetricQueue;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/execute/UnionPlan.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/execute/UnionPlan.java b/phoenix-core/src/main/java/org/apache/phoenix/execute/UnionPlan.java
index 031b58b..2bed3a0 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/execute/UnionPlan.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/execute/UnionPlan.java
@@ -49,7 +49,7 @@ public class UnionPlan implements QueryPlan {
private final FilterableStatement statement;
private final ParameterMetaData paramMetaData;
private final OrderBy orderBy;
- private final StatementContext context;
+ private final StatementContext parentContext;
private final Integer limit;
private final GroupBy groupBy;
private final RowProjector projector;
@@ -59,7 +59,7 @@ public class UnionPlan implements QueryPlan {
public UnionPlan(StatementContext context, FilterableStatement statement, TableRef table, RowProjector projector,
Integer limit, OrderBy orderBy, GroupBy groupBy, List<QueryPlan> plans, ParameterMetaData paramMetaData) throws SQLException {
- this.context = context;
+ this.parentContext = context;
this.statement = statement;
this.tableRef = table;
this.projector = projector;
@@ -128,7 +128,7 @@ public class UnionPlan implements QueryPlan {
}
public final ResultIterator iterator(final List<? extends SQLCloseable> dependencies) throws SQLException {
- this.iterators = new UnionResultIterators(plans);
+ this.iterators = new UnionResultIterators(plans, parentContext);
ResultIterator scanner;
boolean isOrdered = !orderBy.getOrderByExpressions().isEmpty();
@@ -175,7 +175,7 @@ public class UnionPlan implements QueryPlan {
@Override
public StatementContext getContext() {
- return context;
+ return parentContext;
}
@Override
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
index 6a3847b..43731cb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/BaseResultIterators.java
@@ -18,8 +18,8 @@
package org.apache.phoenix.iterate;
import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.EXPECTED_UPPER_REGION_KEY;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.FAILED_QUERY;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.QUERY_TIMEOUT;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_FAILED_QUERY_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIMEOUT_COUNTER;
import static org.apache.phoenix.util.ByteUtil.EMPTY_BYTE_ARRAY;
import java.sql.SQLException;
@@ -540,12 +540,13 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
} catch (ExecutionException e) {
try { // Rethrow as SQLException
throw ServerUtil.parseServerException(e);
- } catch (StaleRegionBoundaryCacheException e2) {
+ } catch (StaleRegionBoundaryCacheException e2) {
// Catch only to try to recover from region boundary cache being out of date
List<List<Pair<Scan,Future<PeekingResultIterator>>>> newFutures = Lists.newArrayListWithExpectedSize(2);
if (!clearedCache) { // Clear cache once so that we rejigger job based on new boundaries
services.clearTableRegionCache(physicalTableName);
clearedCache = true;
+ context.getOverallQueryMetrics().cacheRefreshedDueToSplits();
}
// Resubmit just this portion of work again
Scan oldScan = scanPair.getFirst();
@@ -582,7 +583,8 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
success = true;
return iterators;
} catch (TimeoutException e) {
- QUERY_TIMEOUT.increment();
+ context.getOverallQueryMetrics().queryTimedOut();
+ GLOBAL_QUERY_TIMEOUT_COUNTER.increment();
// thrown when a thread times out waiting for the future.get() call to return
toThrow = new SQLExceptionInfo.Builder(SQLExceptionCode.OPERATION_TIMED_OUT)
.setMessage(". Query couldn't be completed in the alloted time: " + queryTimeOut + " ms")
@@ -616,7 +618,8 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
}
} finally {
if (toThrow != null) {
- FAILED_QUERY.increment();
+ GLOBAL_FAILED_QUERY_COUNTER.increment();
+ context.getOverallQueryMetrics().queryFailed();
throw toThrow;
}
}
@@ -639,7 +642,7 @@ public abstract class BaseResultIterators extends ExplainTable implements Result
if (futurePair != null) {
Future<PeekingResultIterator> future = futurePair.getSecond();
if (future != null) {
- cancelledWork |= future.cancel(false);
+ future.cancel(false);
}
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
index e1ee8db..f272e55 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ChunkedResultIterator.java
@@ -19,6 +19,7 @@
package org.apache.phoenix.iterate;
import static org.apache.phoenix.coprocessor.BaseScannerRegionObserver.STARTKEY_OFFSET;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
import java.sql.SQLException;
import java.util.List;
@@ -66,18 +67,17 @@ public class ChunkedResultIterator implements PeekingResultIterator {
}
@Override
- public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan) throws SQLException {
- scanner.close(); //close the iterator since we don't need it anymore.
+ public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan, String tableName) throws SQLException {
if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("ChunkedResultIteratorFactory.newIterator over " + tableRef.getTable().getName().getString() + " with " + scan, ScanUtil.getCustomAnnotations(scan)));
return new ChunkedResultIterator(delegateFactory, context, tableRef, scan,
context.getConnection().getQueryServices().getProps().getLong(
QueryServices.SCAN_RESULT_CHUNK_SIZE,
- QueryServicesOptions.DEFAULT_SCAN_RESULT_CHUNK_SIZE));
+ QueryServicesOptions.DEFAULT_SCAN_RESULT_CHUNK_SIZE), scanner);
}
}
- public ChunkedResultIterator(ParallelIteratorFactory delegateIteratorFactory,
- StatementContext context, TableRef tableRef, Scan scan, long chunkSize) throws SQLException {
+ private ChunkedResultIterator(ParallelIteratorFactory delegateIteratorFactory,
+ StatementContext context, TableRef tableRef, Scan scan, long chunkSize, ResultIterator scanner) throws SQLException {
this.delegateIteratorFactory = delegateIteratorFactory;
this.context = context;
this.tableRef = tableRef;
@@ -87,9 +87,9 @@ public class ChunkedResultIterator implements PeekingResultIterator {
// to get parallel scans kicked off in separate threads. If we delay this,
// we'll get serialized behavior (see PHOENIX-
if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("Get first chunked result iterator over " + tableRef.getTable().getName().getString() + " with " + scan, ScanUtil.getCustomAnnotations(scan)));
- ResultIterator singleChunkResultIterator = new SingleChunkResultIterator(
- new TableResultIterator(context, tableRef, scan), chunkSize);
- resultIterator = delegateIteratorFactory.newIterator(context, singleChunkResultIterator, scan);
+ ResultIterator singleChunkResultIterator = new SingleChunkResultIterator(scanner, chunkSize);
+ String tableName = tableRef.getTable().getPhysicalName().getString();
+ resultIterator = delegateIteratorFactory.newIterator(context, singleChunkResultIterator, scan, tableName);
}
@Override
@@ -118,9 +118,10 @@ public class ChunkedResultIterator implements PeekingResultIterator {
scan = ScanUtil.newScan(scan);
scan.setStartRow(ByteUtil.copyKeyBytesIfNecessary(lastKey));
if (logger.isDebugEnabled()) logger.debug(LogUtil.addCustomAnnotations("Get next chunked result iterator over " + tableRef.getTable().getName().getString() + " with " + scan, ScanUtil.getCustomAnnotations(scan)));
+ String tableName = tableRef.getTable().getPhysicalName().getString();
ResultIterator singleChunkResultIterator = new SingleChunkResultIterator(
- new TableResultIterator(context, tableRef, scan), chunkSize);
- resultIterator = delegateIteratorFactory.newIterator(context, singleChunkResultIterator, scan);
+ new TableResultIterator(context, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName)), chunkSize);
+ resultIterator = delegateIteratorFactory.newIterator(context, singleChunkResultIterator, scan, tableName);
}
return resultIterator;
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorFactory.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorFactory.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorFactory.java
index df8f658..f25e373 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorFactory.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIteratorFactory.java
@@ -25,10 +25,10 @@ import org.apache.phoenix.compile.StatementContext;
public interface ParallelIteratorFactory {
public static ParallelIteratorFactory NOOP_FACTORY = new ParallelIteratorFactory() {
@Override
- public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan)
+ public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan, String physicalTableName)
throws SQLException {
return LookAheadResultIterator.wrap(scanner);
}
};
- PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan) throws SQLException;
+ PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan, String physicalTableName) throws SQLException;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
index be10c20..2dfbfe3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ParallelIterators.java
@@ -17,7 +17,7 @@
*/
package org.apache.phoenix.iterate;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.PARALLEL_SCANS;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_NUM_PARALLEL_SCANS;
import java.sql.SQLException;
import java.util.Collections;
@@ -30,6 +30,10 @@ import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.job.JobManager.JobCallable;
+import org.apache.phoenix.monitoring.MetricType;
+import org.apache.phoenix.monitoring.CombinableMetric;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.TaskExecutionMetricsHolder;
import org.apache.phoenix.trace.util.Tracing;
import org.apache.phoenix.util.LogUtil;
import org.apache.phoenix.util.ScanUtil;
@@ -79,19 +83,25 @@ public class ParallelIterators extends BaseResultIterators {
// Shuffle so that we start execution across many machines
// before we fill up the thread pool
Collections.shuffle(scanLocations);
- PARALLEL_SCANS.update(scanLocations.size());
+ ReadMetricQueue readMetrics = context.getReadMetricsQueue();
+ final String physicalTableName = tableRef.getTable().getPhysicalName().getString();
+ int numScans = scanLocations.size();
+ context.getOverallQueryMetrics().updateNumParallelScans(numScans);
+ GLOBAL_NUM_PARALLEL_SCANS.update(numScans);
for (ScanLocator scanLocation : scanLocations) {
final Scan scan = scanLocation.getScan();
+ final CombinableMetric scanMetrics = readMetrics.allotMetric(MetricType.SCAN_BYTES, physicalTableName);
+ final TaskExecutionMetricsHolder taskMetrics = new TaskExecutionMetricsHolder(readMetrics, physicalTableName);
Future<PeekingResultIterator> future = executor.submit(Tracing.wrap(new JobCallable<PeekingResultIterator>() {
-
+
@Override
public PeekingResultIterator call() throws Exception {
long startTime = System.currentTimeMillis();
- ResultIterator scanner = new TableResultIterator(context, tableRef, scan);
+ ResultIterator scanner = new TableResultIterator(context, tableRef, scan, scanMetrics);
if (logger.isDebugEnabled()) {
logger.debug(LogUtil.addCustomAnnotations("Id: " + scanId + ", Time: " + (System.currentTimeMillis() - startTime) + "ms, Scan: " + scan, ScanUtil.getCustomAnnotations(scan)));
}
- PeekingResultIterator iterator = iteratorFactory.newIterator(context, scanner, scan);
+ PeekingResultIterator iterator = iteratorFactory.newIterator(context, scanner, scan, physicalTableName);
// Fill the scanner's cache. This helps reduce latency since we are parallelizing the I/O needed.
iterator.peek();
@@ -109,6 +119,11 @@ public class ParallelIterators extends BaseResultIterators {
public Object getJobId() {
return ParallelIterators.this;
}
+
+ @Override
+ public TaskExecutionMetricsHolder getTaskExecutionMetric() {
+ return taskMetrics;
+ }
}, "Parallel scanner for table: " + tableRef.getTable().getName().getString()));
// Add our future in the right place so that we can concatenate the
// results of the inner futures versus merge sorting across all of them.
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
index 4a9ad3e..92ac570 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/RoundRobinResultIterator.java
@@ -18,7 +18,7 @@
package org.apache.phoenix.iterate;
import static com.google.common.base.Preconditions.checkArgument;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.FAILED_QUERY;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_FAILED_QUERY_COUNTER;
import java.sql.SQLException;
import java.util.ArrayList;
@@ -268,7 +268,7 @@ public class RoundRobinResultIterator implements ResultIterator {
}
} finally {
if (toThrow != null) {
- FAILED_QUERY.increment();
+ GLOBAL_FAILED_QUERY_COUNTER.increment();
throw toThrow;
}
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
index fd65d0c..b722794 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/ScanningResultIterator.java
@@ -17,7 +17,7 @@
*/
package org.apache.phoenix.iterate;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.SCAN_BYTES;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SCAN_BYTES;
import java.io.IOException;
import java.sql.SQLException;
@@ -28,15 +28,20 @@ import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.KeyValueUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
-import org.apache.phoenix.monitoring.PhoenixMetrics;
+import org.apache.phoenix.monitoring.CombinableMetric.NoOpRequestMetric;
+import org.apache.phoenix.monitoring.GlobalClientMetrics;
+import org.apache.phoenix.monitoring.CombinableMetric;
import org.apache.phoenix.schema.tuple.ResultTuple;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.ServerUtil;
public class ScanningResultIterator implements ResultIterator {
private final ResultScanner scanner;
- public ScanningResultIterator(ResultScanner scanner) {
+ private final CombinableMetric scanMetrics;
+
+ public ScanningResultIterator(ResultScanner scanner, CombinableMetric scanMetrics) {
this.scanner = scanner;
+ this.scanMetrics = scanMetrics;
}
@Override
@@ -66,17 +71,18 @@ public class ScanningResultIterator implements ResultIterator {
return "ScanningResultIterator [scanner=" + scanner + "]";
}
- private static void calculateScanSize(Result result) {
- if (PhoenixMetrics.isMetricsEnabled()) {
- if (result != null) {
- Cell[] cells = result.rawCells();
- long scanResultSize = 0;
- for (Cell cell : cells) {
- KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
- scanResultSize += kv.heapSize();
- }
- SCAN_BYTES.update(scanResultSize);
- }
- }
- }
+ private void calculateScanSize(Result result) {
+ if (GlobalClientMetrics.isMetricsEnabled() || scanMetrics != NoOpRequestMetric.INSTANCE) {
+ if (result != null) {
+ Cell[] cells = result.rawCells();
+ long scanResultSize = 0;
+ for (Cell cell : cells) {
+ KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
+ scanResultSize += kv.heapSize();
+ }
+ scanMetrics.change(scanResultSize);
+ GLOBAL_SCAN_BYTES.update(scanResultSize);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
index 6b3b5e3..516d73e 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SerialIterators.java
@@ -17,6 +17,8 @@
*/
package org.apache.phoenix.iterate;
+import static org.apache.phoenix.monitoring.MetricType.SCAN_BYTES;
+
import java.sql.SQLException;
import java.util.Collections;
import java.util.List;
@@ -29,11 +31,9 @@ import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.compile.QueryPlan;
import org.apache.phoenix.iterate.TableResultIterator.ScannerCreation;
import org.apache.phoenix.job.JobManager.JobCallable;
+import org.apache.phoenix.monitoring.TaskExecutionMetricsHolder;
import org.apache.phoenix.trace.util.Tracing;
-import org.apache.phoenix.util.LogUtil;
import org.apache.phoenix.util.ScanUtil;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
@@ -48,7 +48,6 @@ import com.google.common.collect.Lists;
* @since 0.1
*/
public class SerialIterators extends BaseResultIterators {
- private static final Logger logger = LoggerFactory.getLogger(SerialIterators.class);
private static final String NAME = "SERIAL";
private final ParallelIteratorFactory iteratorFactory;
@@ -74,18 +73,15 @@ public class SerialIterators extends BaseResultIterators {
Scan lastScan = scans.get(scans.size()-1);
final Scan overallScan = ScanUtil.newScan(firstScan);
overallScan.setStopRow(lastScan.getStopRow());
+ final String tableName = tableRef.getTable().getPhysicalName().getString();
+ final TaskExecutionMetricsHolder taskMetrics = new TaskExecutionMetricsHolder(context.getReadMetricsQueue(), tableName);
Future<PeekingResultIterator> future = executor.submit(Tracing.wrap(new JobCallable<PeekingResultIterator>() {
-
@Override
public PeekingResultIterator call() throws Exception {
List<PeekingResultIterator> concatIterators = Lists.newArrayListWithExpectedSize(scans.size());
for (final Scan scan : scans) {
- long startTime = System.currentTimeMillis();
- ResultIterator scanner = new TableResultIterator(context, tableRef, scan, ScannerCreation.DELAYED);
- if (logger.isDebugEnabled()) {
- logger.debug(LogUtil.addCustomAnnotations("Id: " + scanId + ", Time: " + (System.currentTimeMillis() - startTime) + "ms, Scan: " + scan, ScanUtil.getCustomAnnotations(scan)));
- }
- concatIterators.add(iteratorFactory.newIterator(context, scanner, scan));
+ ResultIterator scanner = new TableResultIterator(context, tableRef, scan, context.getReadMetricsQueue().allotMetric(SCAN_BYTES, tableName), ScannerCreation.DELAYED);
+ concatIterators.add(iteratorFactory.newIterator(context, scanner, scan, tableName));
}
PeekingResultIterator concatIterator = ConcatResultIterator.newIterator(concatIterators);
allIterators.add(concatIterator);
@@ -101,6 +97,11 @@ public class SerialIterators extends BaseResultIterators {
public Object getJobId() {
return SerialIterators.this;
}
+
+ @Override
+ public TaskExecutionMetricsHolder getTaskExecutionMetric() {
+ return taskMetrics;
+ }
}, "Serial scanner for table: " + tableRef.getTable().getName().getString()));
// Add our singleton Future which will execute serially
nestedFutures.add(Collections.singletonList(new Pair<Scan,Future<PeekingResultIterator>>(overallScan,future)));
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SpoolingResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SpoolingResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SpoolingResultIterator.java
index f49bce5..540b410 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/SpoolingResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/SpoolingResultIterator.java
@@ -17,6 +17,11 @@
*/
package org.apache.phoenix.iterate;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MEMORY_CHUNK_BYTES;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MEMORY_WAIT_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SPOOL_FILE_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SPOOL_FILE_SIZE;
+
import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -34,6 +39,9 @@ import org.apache.hadoop.io.WritableUtils;
import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.memory.MemoryManager;
import org.apache.phoenix.memory.MemoryManager.MemoryChunk;
+import org.apache.phoenix.monitoring.MemoryMetricsHolder;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
+import org.apache.phoenix.monitoring.SpoolingMetricsHolder;
import org.apache.phoenix.query.QueryServices;
import org.apache.phoenix.query.QueryServicesOptions;
import org.apache.phoenix.schema.tuple.ResultTuple;
@@ -42,8 +50,6 @@ import org.apache.phoenix.util.ByteUtil;
import org.apache.phoenix.util.ResultUtil;
import org.apache.phoenix.util.ServerUtil;
import org.apache.phoenix.util.TupleUtil;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.NUM_SPOOL_FILE;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.SPOOL_FILE_SIZE;
@@ -56,7 +62,10 @@ import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.SPOOL_FILE
* @since 0.1
*/
public class SpoolingResultIterator implements PeekingResultIterator {
+
private final PeekingResultIterator spoolFrom;
+ private final SpoolingMetricsHolder spoolMetrics;
+ private final MemoryMetricsHolder memoryMetrics;
public static class SpoolingResultIteratorFactory implements ParallelIteratorFactory {
private final QueryServices services;
@@ -65,14 +74,16 @@ public class SpoolingResultIterator implements PeekingResultIterator {
this.services = services;
}
@Override
- public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan) throws SQLException {
- return new SpoolingResultIterator(scanner, services);
+ public PeekingResultIterator newIterator(StatementContext context, ResultIterator scanner, Scan scan, String physicalTableName) throws SQLException {
+ ReadMetricQueue readRequestMetric = context.getReadMetricsQueue();
+ SpoolingMetricsHolder spoolMetrics = new SpoolingMetricsHolder(readRequestMetric, physicalTableName);
+ MemoryMetricsHolder memoryMetrics = new MemoryMetricsHolder(readRequestMetric, physicalTableName);
+ return new SpoolingResultIterator(spoolMetrics, memoryMetrics, scanner, services);
}
-
}
- public SpoolingResultIterator(ResultIterator scanner, QueryServices services) throws SQLException {
- this (scanner, services.getMemoryManager(),
+ private SpoolingResultIterator(SpoolingMetricsHolder spoolMetrics, MemoryMetricsHolder memoryMetrics, ResultIterator scanner, QueryServices services) throws SQLException {
+ this (spoolMetrics, memoryMetrics, scanner, services.getMemoryManager(),
services.getProps().getInt(QueryServices.SPOOL_THRESHOLD_BYTES_ATTRIB, QueryServicesOptions.DEFAULT_SPOOL_THRESHOLD_BYTES),
services.getProps().getLong(QueryServices.MAX_SPOOL_TO_DISK_BYTES_ATTRIB, QueryServicesOptions.DEFAULT_MAX_SPOOL_TO_DISK_BYTES),
services.getProps().get(QueryServices.SPOOL_DIRECTORY, QueryServicesOptions.DEFAULT_SPOOL_DIRECTORY));
@@ -87,9 +98,15 @@ public class SpoolingResultIterator implements PeekingResultIterator {
* the memory manager) is exceeded.
* @throws SQLException
*/
- SpoolingResultIterator(ResultIterator scanner, MemoryManager mm, final int thresholdBytes, final long maxSpoolToDisk, final String spoolDirectory) throws SQLException {
+ SpoolingResultIterator(SpoolingMetricsHolder sMetrics, MemoryMetricsHolder mMetrics, ResultIterator scanner, MemoryManager mm, final int thresholdBytes, final long maxSpoolToDisk, final String spoolDirectory) throws SQLException {
+ this.spoolMetrics = sMetrics;
+ this.memoryMetrics = mMetrics;
boolean success = false;
+ long startTime = System.currentTimeMillis();
final MemoryChunk chunk = mm.allocate(0, thresholdBytes);
+ long waitTime = System.currentTimeMillis() - startTime;
+ GLOBAL_MEMORY_WAIT_TIME.update(waitTime);
+ memoryMetrics.getMemoryWaitTimeMetric().change(waitTime);
DeferredFileOutputStream spoolTo = null;
try {
// Can't be bigger than int, since it's the max of the above allocation
@@ -97,8 +114,11 @@ public class SpoolingResultIterator implements PeekingResultIterator {
spoolTo = new DeferredFileOutputStream(size, "ResultSpooler",".bin", new File(spoolDirectory)) {
@Override
protected void thresholdReached() throws IOException {
- super.thresholdReached();
- chunk.close();
+ try {
+ super.thresholdReached();
+ } finally {
+ chunk.close();
+ }
}
};
DataOutputStream out = new DataOutputStream(spoolTo);
@@ -116,9 +136,14 @@ public class SpoolingResultIterator implements PeekingResultIterator {
byte[] data = spoolTo.getData();
chunk.resize(data.length);
spoolFrom = new InMemoryResultIterator(data, chunk);
+ GLOBAL_MEMORY_CHUNK_BYTES.update(data.length);
+ memoryMetrics.getMemoryChunkSizeMetric().change(data.length);
} else {
- NUM_SPOOL_FILE.increment();
- SPOOL_FILE_SIZE.update(spoolTo.getFile().length());
+ long sizeOfSpoolFile = spoolTo.getFile().length();
+ GLOBAL_SPOOL_FILE_SIZE.update(sizeOfSpoolFile);
+ GLOBAL_SPOOL_FILE_COUNTER.increment();
+ spoolMetrics.getNumSpoolFileMetric().increment();
+ spoolMetrics.getSpoolFileSizeMetric().change(sizeOfSpoolFile);
spoolFrom = new OnDiskResultIterator(spoolTo.getFile());
}
success = true;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
index ea13dfd..6f040d1 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/TableResultIterator.java
@@ -24,6 +24,7 @@ import java.util.List;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.monitoring.CombinableMetric;
import org.apache.phoenix.schema.TableRef;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.util.Closeables;
@@ -44,9 +45,10 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
private final Scan scan;
private final HTableInterface htable;
private volatile ResultIterator delegate;
-
- public TableResultIterator(StatementContext context, TableRef tableRef) throws SQLException {
- this(context, tableRef, context.getScan());
+ private final CombinableMetric scanMetrics;
+
+ public TableResultIterator(StatementContext context, TableRef tableRef, CombinableMetric scanMetrics) throws SQLException {
+ this(context, tableRef, context.getScan(), scanMetrics);
}
/*
@@ -62,7 +64,7 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
delegate = this.delegate;
if (delegate == null) {
try {
- this.delegate = delegate = isClosing ? ResultIterator.EMPTY_ITERATOR : new ScanningResultIterator(htable.getScanner(scan));
+ this.delegate = delegate = isClosing ? ResultIterator.EMPTY_ITERATOR : new ScanningResultIterator(htable.getScanner(scan), scanMetrics);
} catch (IOException e) {
Closeables.closeQuietly(htable);
throw ServerUtil.parseServerException(e);
@@ -73,13 +75,14 @@ public class TableResultIterator extends ExplainTable implements ResultIterator
return delegate;
}
- public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan) throws SQLException {
- this(context, tableRef, scan, ScannerCreation.IMMEDIATE);
+ public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan, CombinableMetric scanMetrics) throws SQLException {
+ this(context, tableRef, scan, scanMetrics, ScannerCreation.IMMEDIATE);
}
- public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan, ScannerCreation creationMode) throws SQLException {
+ public TableResultIterator(StatementContext context, TableRef tableRef, Scan scan, CombinableMetric scanMetrics, ScannerCreation creationMode) throws SQLException {
super(context, tableRef);
this.scan = scan;
+ this.scanMetrics = scanMetrics;
htable = context.getConnection().getQueryServices().getTable(tableRef.getTable().getPhysicalName().getBytes());
if (creationMode == ScannerCreation.IMMEDIATE) {
getDelegate(false);
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java b/phoenix-core/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
index b7c8b21..2296982 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/iterate/UnionResultIterators.java
@@ -22,6 +22,9 @@ import java.util.List;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.phoenix.compile.QueryPlan;
+import org.apache.phoenix.compile.StatementContext;
+import org.apache.phoenix.monitoring.OverAllQueryMetrics;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.query.KeyRange;
import org.apache.phoenix.util.ServerUtil;
@@ -39,14 +42,22 @@ public class UnionResultIterators implements ResultIterators {
private final List<List<Scan>> scans;
private final List<PeekingResultIterator> iterators;
private final List<QueryPlan> plans;
-
- public UnionResultIterators(List<QueryPlan> plans) throws SQLException {
+ private final List<ReadMetricQueue> readMetricsList;
+ private final List<OverAllQueryMetrics> overAllQueryMetricsList;
+ private boolean closed;
+ private final StatementContext parentStmtCtx;
+ public UnionResultIterators(List<QueryPlan> plans, StatementContext parentStmtCtx) throws SQLException {
+ this.parentStmtCtx = parentStmtCtx;
this.plans = plans;
int nPlans = plans.size();
iterators = Lists.newArrayListWithExpectedSize(nPlans);
splits = Lists.newArrayListWithExpectedSize(nPlans * 30);
scans = Lists.newArrayListWithExpectedSize(nPlans * 10);
+ readMetricsList = Lists.newArrayListWithCapacity(nPlans);
+ overAllQueryMetricsList = Lists.newArrayListWithCapacity(nPlans);
for (QueryPlan plan : this.plans) {
+ readMetricsList.add(plan.getContext().getReadMetricsQueue());
+ overAllQueryMetricsList.add(plan.getContext().getOverallQueryMetrics());
iterators.add(LookAheadResultIterator.wrap(plan.iterator()));
splits.addAll(plan.getSplits());
scans.addAll(plan.getScans());
@@ -59,32 +70,47 @@ public class UnionResultIterators implements ResultIterators {
}
@Override
- public void close() throws SQLException {
- SQLException toThrow = null;
- try {
- if (iterators != null) {
- for (int index=0; index < iterators.size(); index++) {
- PeekingResultIterator iterator = iterators.get(index);
- try {
- iterator.close();
- } catch (Exception e) {
- if (toThrow == null) {
- toThrow = ServerUtil.parseServerException(e);
- } else {
- toThrow.setNextException(ServerUtil.parseServerException(e));
+ public void close() throws SQLException {
+ if (!closed) {
+ closed = true;
+ SQLException toThrow = null;
+ try {
+ if (iterators != null) {
+ for (int index=0; index < iterators.size(); index++) {
+ PeekingResultIterator iterator = iterators.get(index);
+ try {
+ iterator.close();
+ } catch (Exception e) {
+ if (toThrow == null) {
+ toThrow = ServerUtil.parseServerException(e);
+ } else {
+ toThrow.setNextException(ServerUtil.parseServerException(e));
+ }
}
}
}
- }
- } catch (Exception e) {
- toThrow = ServerUtil.parseServerException(e);
- } finally {
- if (toThrow != null) {
- throw toThrow;
+ } catch (Exception e) {
+ toThrow = ServerUtil.parseServerException(e);
+ } finally {
+ setMetricsInParentContext();
+ if (toThrow != null) {
+ throw toThrow;
+ }
}
}
}
-
+
+ private void setMetricsInParentContext() {
+ ReadMetricQueue parentCtxReadMetrics = parentStmtCtx.getReadMetricsQueue();
+ for (ReadMetricQueue readMetrics : readMetricsList) {
+ parentCtxReadMetrics.combineReadMetrics(readMetrics);
+ }
+ OverAllQueryMetrics parentCtxQueryMetrics = parentStmtCtx.getOverallQueryMetrics();
+ for (OverAllQueryMetrics metric : overAllQueryMetricsList) {
+ parentCtxQueryMetrics.combine(metric);
+ }
+ }
+
@Override
public List<List<Scan>> getScans() {
return scans;
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
index e33e8ee..a6479f3 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixConnection.java
@@ -122,7 +122,7 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
private final Properties info;
private List<SQLCloseable> statements = new ArrayList<SQLCloseable>();
private final Map<PDataType<?>, Format> formatters = new HashMap<>();
- private MutationState mutationState;
+ private final MutationState mutationState;
private final int mutateBatchSize;
private final Long scn;
private boolean isAutoCommit = false;
@@ -136,8 +136,9 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
private boolean isClosed = false;
private Sampler<?> sampler;
private boolean readOnly = false;
- private Map<String, String> customTracingAnnotations = emptyMap();
-
+ private Map<String, String> customTracingAnnotations = emptyMap();
+ private final boolean isRequestLevelMetricsEnabled;
+
static {
Tracing.addTraceMetricsSource();
}
@@ -232,6 +233,7 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
! Objects.equal(tenantId, function.getTenantId()));
}
};
+ this.isRequestLevelMetricsEnabled = JDBCUtil.isCollectingRequestLevelMetricsEnabled(url, info, this.services.getProps());
this.mutationState = newMutationState(maxSize);
this.metaData = metaData.pruneTables(pruner);
this.metaData = metaData.pruneFunctions(pruner);
@@ -433,6 +435,7 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
return;
}
try {
+ clearMetrics();
try {
if (traceScope != null) {
traceScope.close();
@@ -853,4 +856,23 @@ public class PhoenixConnection implements Connection, org.apache.phoenix.jdbc.Jd
public void setTraceScope(TraceScope traceScope) {
this.traceScope = traceScope;
}
+
+ public Map<String, Map<String, Long>> getMutationMetrics() {
+ return mutationState.getMutationMetricQueue().aggregate();
+ }
+
+ public Map<String, Map<String, Long>> getReadMetrics() {
+ return mutationState.getReadMetricQueue() != null ? mutationState.getReadMetricQueue().aggregate() : Collections.<String, Map<String, Long>>emptyMap();
+ }
+
+ public boolean isRequestLevelMetricsEnabled() {
+ return isRequestLevelMetricsEnabled;
+ }
+
+ public void clearMetrics() {
+ mutationState.getMutationMetricQueue().clearMetrics();
+ if (mutationState.getReadMetricQueue() != null) {
+ mutationState.getReadMetricQueue().clearMetrics();
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
index d1b3b27..2dd8af4 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixDatabaseMetaData.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.hbase.util.Bytes;
import org.apache.phoenix.compile.ColumnProjector;
import org.apache.phoenix.compile.ExpressionProjector;
import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.coprocessor.MetaDataProtocol;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
@@ -311,7 +312,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
public static final int CLIENT_KEY_VALUE_BUILDER_THRESHOLD = VersionUtil.encodeVersion("0", "94", "14");
PhoenixDatabaseMetaData(PhoenixConnection connection) throws SQLException {
- this.emptyResultSet = new PhoenixResultSet(ResultIterator.EMPTY_ITERATOR, RowProjector.EMPTY_PROJECTOR, new PhoenixStatement(connection));
+ this.emptyResultSet = new PhoenixResultSet(ResultIterator.EMPTY_ITERATOR, RowProjector.EMPTY_PROJECTOR, new StatementContext(new PhoenixStatement(connection), false));
this.connection = connection;
}
@@ -509,11 +510,10 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
public PhoenixStatement newStatement(PhoenixConnection connection) {
return new PhoenixStatement(connection) {
@Override
- protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector)
- throws SQLException {
- return new PhoenixResultSet(
- new TenantColumnFilteringIterator(iterator, projector),
- projector, this);
+ protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector,
+ StatementContext context) throws SQLException {
+ return new PhoenixResultSet(new TenantColumnFilteringIterator(iterator, projector),
+ projector, context);
}
};
}
@@ -523,7 +523,12 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
}
return stmt.executeQuery(buf.toString());
}
-
+
+// private ColumnResolver getColumnResolverForCatalogTable() throws SQLException {
+// TableRef tableRef = new TableRef(getTable(connection, SYSTEM_CATALOG_NAME));
+// return FromCompiler.getResolver(tableRef);
+// }
+
/**
* Filters the tenant id column out of a column metadata result set (thus, where each row is a column definition).
* The tenant id is by definition the first column of the primary key, but the primary key does not necessarily
@@ -1007,7 +1012,7 @@ public class PhoenixDatabaseMetaData implements DatabaseMetaData, org.apache.pho
}
@Override
public ResultSet getTableTypes() throws SQLException {
- return new PhoenixResultSet(new MaterializedResultIterator(TABLE_TYPE_TUPLES), TABLE_TYPE_ROW_PROJECTOR, new PhoenixStatement(connection));
+ return new PhoenixResultSet(new MaterializedResultIterator(TABLE_TYPE_TUPLES), TABLE_TYPE_ROW_PROJECTOR, new StatementContext(new PhoenixStatement(connection), false));
}
/**
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
index 8ee56ea..da06370 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixResultSet.java
@@ -39,16 +39,21 @@ import java.sql.Time;
import java.sql.Timestamp;
import java.text.Format;
import java.util.Calendar;
+import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.util.Pair;
import org.apache.phoenix.compile.ColumnProjector;
import org.apache.phoenix.compile.RowProjector;
+import org.apache.phoenix.compile.StatementContext;
import org.apache.phoenix.exception.SQLExceptionCode;
import org.apache.phoenix.exception.SQLExceptionInfo;
import org.apache.phoenix.iterate.ResultIterator;
+import org.apache.phoenix.monitoring.OverAllQueryMetrics;
+import org.apache.phoenix.monitoring.ReadMetricQueue;
import org.apache.phoenix.schema.tuple.ResultTuple;
import org.apache.phoenix.schema.tuple.Tuple;
import org.apache.phoenix.schema.types.PBoolean;
@@ -109,18 +114,25 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable, org.apache.pho
private final ResultIterator scanner;
private final RowProjector rowProjector;
private final PhoenixStatement statement;
+ private final StatementContext context;
+ private final ReadMetricQueue readMetricsQueue;
+ private final OverAllQueryMetrics overAllQueryMetrics;
private final ImmutableBytesWritable ptr = new ImmutableBytesWritable();
private Tuple currentRow = BEFORE_FIRST;
private boolean isClosed = false;
private boolean wasNull = false;
-
- public PhoenixResultSet(ResultIterator resultIterator, RowProjector rowProjector, PhoenixStatement statement) throws SQLException {
+ private boolean firstRecordRead = false;
+
+ public PhoenixResultSet(ResultIterator resultIterator, RowProjector rowProjector, StatementContext ctx) throws SQLException {
this.rowProjector = rowProjector;
this.scanner = resultIterator;
- this.statement = statement;
+ this.context = ctx;
+ this.statement = context.getStatement();
+ this.readMetricsQueue = context.getReadMetricsQueue();
+ this.overAllQueryMetrics = context.getOverallQueryMetrics();
}
-
+
@Override
public boolean absolute(int row) throws SQLException {
throw new SQLFeatureNotSupportedException();
@@ -147,14 +159,14 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable, org.apache.pho
@Override
public void close() throws SQLException {
- if (isClosed) {
- return;
- }
+ if (isClosed) { return; }
try {
scanner.close();
} finally {
isClosed = true;
statement.getResultSets().remove(this);
+ overAllQueryMetrics.endQuery();
+ overAllQueryMetrics.stopResultSetWatch();
}
}
@@ -754,6 +766,10 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable, org.apache.pho
public boolean next() throws SQLException {
checkOpen();
try {
+ if (!firstRecordRead) {
+ firstRecordRead = true;
+ overAllQueryMetrics.startResultSetWatch();
+ }
currentRow = scanner.next();
rowProjector.reset();
} catch (RuntimeException e) {
@@ -764,6 +780,10 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable, org.apache.pho
}
throw e;
}
+ if (currentRow == null) {
+ overAllQueryMetrics.endQuery();
+ overAllQueryMetrics.stopResultSetWatch();
+ }
return currentRow != null;
}
@@ -1261,4 +1281,18 @@ public class PhoenixResultSet implements ResultSet, SQLCloseable, org.apache.pho
public ResultIterator getUnderlyingIterator() {
return scanner;
}
+
+ public Map<String, Map<String, Long>> getReadMetrics() {
+ return readMetricsQueue.aggregate();
+ }
+
+ public Map<String, Long> getOverAllRequestReadMetrics() {
+ return overAllQueryMetrics.publish();
+ }
+
+ public void resetMetrics() {
+ readMetricsQueue.clearMetrics();
+ overAllQueryMetrics.reset();
+ }
+
}
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
index 51fb027..9b67c22 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/jdbc/PhoenixStatement.java
@@ -17,9 +17,9 @@
*/
package org.apache.phoenix.jdbc;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.MUTATION_COUNT;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.QUERY_COUNT;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.QUERY_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_QUERY_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_SELECT_SQL_COUNTER;
import java.io.IOException;
import java.io.Reader;
@@ -213,8 +213,8 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
return resultSets;
}
- protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector) throws SQLException {
- return new PhoenixResultSet(iterator, projector, this);
+ protected PhoenixResultSet newResultSet(ResultIterator iterator, RowProjector projector, StatementContext context) throws SQLException {
+ return new PhoenixResultSet(iterator, projector, context);
}
protected boolean execute(final CompilableStatement stmt) throws SQLException {
@@ -232,7 +232,7 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
}
protected PhoenixResultSet executeQuery(final CompilableStatement stmt) throws SQLException {
- QUERY_COUNT.increment();
+ GLOBAL_SELECT_SQL_COUNTER.increment();
try {
return CallRunner.run(
new CallRunner.CallableThrowable<PhoenixResultSet, SQLException>() {
@@ -250,7 +250,9 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
String explainPlan = QueryUtil.getExplainPlan(resultIterator);
logger.debug(LogUtil.addCustomAnnotations("Explain plan: " + explainPlan, connection));
}
- PhoenixResultSet rs = newResultSet(resultIterator, plan.getProjector());
+ StatementContext context = plan.getContext();
+ context.getOverallQueryMetrics().startQuery();
+ PhoenixResultSet rs = newResultSet(resultIterator, plan.getProjector(), context);
resultSets.add(rs);
setLastQueryPlan(plan);
setLastResultSet(rs);
@@ -269,7 +271,7 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
// Regardless of whether the query was successfully handled or not,
// update the time spent so far. If needed, we can separate out the
// success times and failure times.
- QUERY_TIME.update(System.currentTimeMillis() - startTime);
+ GLOBAL_QUERY_TIME.update(System.currentTimeMillis() - startTime);
}
}
}, PhoenixContextExecutor.inContext());
@@ -285,7 +287,7 @@ public class PhoenixStatement implements Statement, SQLCloseable, org.apache.pho
SQLExceptionCode.READ_ONLY_CONNECTION).
build().buildException();
}
- MUTATION_COUNT.increment();
+ GLOBAL_MUTATION_SQL_COUNTER.increment();
try {
return CallRunner
.run(
http://git-wip-us.apache.org/repos/asf/phoenix/blob/7e29d57b/phoenix-core/src/main/java/org/apache/phoenix/job/JobManager.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/job/JobManager.java b/phoenix-core/src/main/java/org/apache/phoenix/job/JobManager.java
index 31ef742..7406e46 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/job/JobManager.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/job/JobManager.java
@@ -17,11 +17,11 @@
*/
package org.apache.phoenix.job;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.REJECTED_TASK_COUNT;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.CountMetric.TASK_COUNT;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.TASK_END_TO_END_TIME;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.TASK_EXECUTION_TIME;
-import static org.apache.phoenix.monitoring.PhoenixMetrics.SizeMetric.TASK_QUEUE_WAIT_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_REJECTED_TASK_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_END_TO_END_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_EXECUTED_COUNTER;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_EXECUTION_TIME;
+import static org.apache.phoenix.monitoring.GlobalClientMetrics.GLOBAL_TASK_QUEUE_WAIT_TIME;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
@@ -36,6 +36,10 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.Nullable;
+
+import org.apache.phoenix.monitoring.TaskExecutionMetricsHolder;
+
import com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
*
@@ -63,6 +67,7 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
public static interface JobRunnable<T> extends Runnable {
public Object getJobId();
+ public TaskExecutionMetricsHolder getTaskExecutionMetric();
}
public static ThreadPoolExecutor createThreadPoolExec(int keepAliveMs, int size, int queueSize, boolean useInstrumentedThreadPool) {
@@ -117,13 +122,17 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
*/
static class JobFutureTask<T> extends FutureTask<T> {
private final Object jobId;
+ @Nullable
+ private final TaskExecutionMetricsHolder taskMetric;
public JobFutureTask(Runnable r, T t) {
super(r, t);
if(r instanceof JobRunnable){
this.jobId = ((JobRunnable)r).getJobId();
+ this.taskMetric = ((JobRunnable)r).getTaskExecutionMetric();
} else {
this.jobId = this;
+ this.taskMetric = null;
}
}
@@ -132,8 +141,10 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
// FIXME: this fails when executor used by hbase
if (c instanceof JobCallable) {
this.jobId = ((JobCallable<T>) c).getJobId();
+ this.taskMetric = ((JobCallable<T>) c).getTaskExecutionMetric();
} else {
this.jobId = this;
+ this.taskMetric = null;
}
}
@@ -187,6 +198,7 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
*/
public static interface JobCallable<T> extends Callable<T> {
public Object getJobId();
+ public TaskExecutionMetricsHolder getTaskExecutionMetric();
}
@@ -224,27 +236,40 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
private final RejectedExecutionHandler rejectedExecHandler = new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
- REJECTED_TASK_COUNT.increment();
+ TaskExecutionMetricsHolder metrics = getRequestMetric(r);
+ if (metrics != null) {
+ metrics.getNumRejectedTasks().increment();
+ }
+ GLOBAL_REJECTED_TASK_COUNTER.increment();
throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + executor.toString());
}
};
- public InstrumentedThreadPoolExecutor(String threadPoolName, int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
- BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
+ public InstrumentedThreadPoolExecutor(String threadPoolName, int corePoolSize, int maximumPoolSize,
+ long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
setRejectedExecutionHandler(rejectedExecHandler);
}
@Override
public void execute(Runnable task) {
- TASK_COUNT.increment();
+ TaskExecutionMetricsHolder metrics = getRequestMetric(task);
+ if (metrics != null) {
+ metrics.getNumTasks().increment();
+ }
+ GLOBAL_TASK_EXECUTED_COUNTER.increment();
super.execute(task);
}
@Override
protected void beforeExecute(Thread worker, Runnable task) {
InstrumentedJobFutureTask instrumentedTask = (InstrumentedJobFutureTask)task;
- TASK_QUEUE_WAIT_TIME.update(System.currentTimeMillis() - instrumentedTask.getTaskSubmissionTime());
+ long queueWaitTime = System.currentTimeMillis() - instrumentedTask.getTaskSubmissionTime();
+ GLOBAL_TASK_QUEUE_WAIT_TIME.update(queueWaitTime);
+ TaskExecutionMetricsHolder metrics = getRequestMetric(task);
+ if (metrics != null) {
+ metrics.getTaskQueueWaitTime().change(queueWaitTime);
+ }
super.beforeExecute(worker, instrumentedTask);
}
@@ -254,10 +279,21 @@ public class JobManager<T> extends AbstractRoundRobinQueue<T> {
try {
super.afterExecute(instrumentedTask, t);
} finally {
- TASK_EXECUTION_TIME.update(System.currentTimeMillis() - instrumentedTask.getTaskExecutionStartTime());
- TASK_END_TO_END_TIME.update(System.currentTimeMillis() - instrumentedTask.getTaskSubmissionTime());
+ long taskExecutionTime = System.currentTimeMillis() - instrumentedTask.getTaskExecutionStartTime();
+ long endToEndTaskTime = System.currentTimeMillis() - instrumentedTask.getTaskSubmissionTime();
+ TaskExecutionMetricsHolder metrics = getRequestMetric(task);
+ if (metrics != null) {
+ metrics.getTaskExecutionTime().change(taskExecutionTime);
+ metrics.getTaskEndToEndTime().change(endToEndTaskTime);
+ }
+ GLOBAL_TASK_EXECUTION_TIME.update(taskExecutionTime);
+ GLOBAL_TASK_END_TO_END_TIME.update(endToEndTaskTime);
}
}
+
+ private static TaskExecutionMetricsHolder getRequestMetric(Runnable task) {
+ return ((JobFutureTask)task).taskMetric;
+ }
}
}