PHOENIX-7485: Add metrics for time taken during mutation plan creation and execution

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

@@ -1724,7 +1724,7 @@ public static MutationMetricQueue.MutationMetric updateMutationBatchFailureMetri
                 numUpsertMutationsInBatch,
                 allUpsertsMutations ? 1 : 0,
                 numDeleteMutationsInBatch,
-                allDeletesMutations ? 1 : 0);
+                allDeletesMutations ? 1 : 0, 0, 0, 0, 0, 0, 0);
     }
 
     /**
@@ -1807,7 +1807,7 @@ static MutationMetric getCommittedMutationsMetric(
                 committedDeleteMutationCounter,
                 committedTotalMutationBytes,
                 numFailedPhase3Mutations,
-                0, 0, 0, 0 );
+                0, 0, 0, 0, 0, 0, 0, 0, 0, 0);
     }
 
     private void filterIndexCheckerMutations(Map<TableInfo, List<Mutation>> mutationMap,

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

@@ -43,6 +43,7 @@
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_SQL_QUERY_TIME;
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_SUCCESS_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MutationMetricQueue.MutationMetric;
 
 import java.io.File;
 import java.io.IOException;
@@ -606,15 +607,17 @@ private Pair<Integer, ResultSet> executeMutation(final CompilableStatement stmt,
                            new CallRunner.CallableThrowable<Pair<Integer, ResultSet>, SQLException>() {
                         @Override
                             public Pair<Integer, ResultSet> call() throws SQLException {
+                            final long startExecuteMutationTime = EnvironmentEdgeManager.timeMarkerInNanos();
                             boolean success = false;
                             String tableName = null;
                             boolean isUpsert = false;
                             boolean isAtomicUpsert = false;
                             boolean isDelete = false;
                             MutationState state = null;
                             MutationPlan plan = null;
-                            final long startExecuteMutationTime = EnvironmentEdgeManager.currentTimeMillis();
                             clearResultSet();
+                            long mutationPlanCreationTimeInNs = 0;
+                            long mutationPlanExecutionTimeInNs = 0;
                             try {
                                 PhoenixConnection conn = getConnection();
                                 if (conn.getQueryServices().isUpgradeRequired() && !conn.isRunningUpgrade()
@@ -624,6 +627,7 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                 state = connection.getMutationState();
                                 isUpsert = stmt instanceof ExecutableUpsertStatement;
                                 isDelete = stmt instanceof ExecutableDeleteStatement;
+                                long mutationPlanCreationStartTimeMarker = EnvironmentEdgeManager.timeMarkerInNanos();
                                 if (isDelete && connection.getAutoCommit() &&
                                         returnResult == ReturnResult.ROW) {
                                     // used only if single row deletion needs to atomically
@@ -635,6 +639,8 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                     plan = stmt.compilePlan(PhoenixStatement.this,
                                             Sequence.ValueOp.VALIDATE_SEQUENCE);
                                 }
+                                mutationPlanCreationTimeInNs = EnvironmentEdgeManager.timeMarkerInNanos() -
+                                        mutationPlanCreationStartTimeMarker;
                                 isAtomicUpsert = isUpsert && ((ExecutableUpsertStatement)stmt).getOnDupKeyPairs() != null;
                                 if (plan.getTargetRef() != null && plan.getTargetRef().getTable() != null) {
                                     if (!Strings.isNullOrEmpty(plan.getTargetRef().getTable().getPhysicalName().toString())) {
@@ -648,7 +654,10 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                 state.sendUncommitted(tableRefs);
                                 state.checkpointIfNeccessary(plan);
                                 checkIfDDLStatementandMutationState(stmt, state);
+                                long mutationPlanExecutionStartTimeMarker = EnvironmentEdgeManager.timeMarkerInNanos();
                                 MutationState lastState = plan.execute();
+                                mutationPlanExecutionTimeInNs = EnvironmentEdgeManager.timeMarkerInNanos() -
+                                        mutationPlanExecutionStartTimeMarker;
                                 state.join(lastState);
                                 // Unfortunately, JDBC uses an int for update count, so we
                                 // just max out at Integer.MAX_VALUE
@@ -718,8 +727,10 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                             MUTATION_SQL_COUNTER, 1);
                                     // Only count dml operations
                                     if (isUpsert || isDelete) {
-                                        long executeMutationTimeSpent =
-                                                EnvironmentEdgeManager.currentTimeMillis() - startExecuteMutationTime;
+                                        long executeMutationTimeSpentInNs =
+                                                EnvironmentEdgeManager.timeMarkerInNanos() - startExecuteMutationTime;
+                                        // This will ensure existing use cases of metrics are not broken.
+                                        long executeMutationTimeSpent = PhoenixRuntime.convertTimeInNsToMs(executeMutationTimeSpentInNs);
 
                                         TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
                                                 UPSERT_SQL_COUNTER : DELETE_SQL_COUNTER, 1);
@@ -731,6 +742,19 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                             TableMetricsManager.updateMetricsMethod(tableName,
                                                 ATOMIC_UPSERT_SQL_QUERY_TIME, executeMutationTimeSpent);
                                         }
+                                        MutationMetric stagedMutationMetric;
+                                        if (isUpsert) {
+                                            stagedMutationMetric = getUncommittedMutationMetric(
+                                                    mutationPlanCreationTimeInNs, mutationPlanExecutionTimeInNs,
+                                                    0, 0,
+                                                    executeMutationTimeSpentInNs, 0);
+                                        }
+                                        else {
+                                            stagedMutationMetric = getUncommittedMutationMetric(0, 0,
+                                                    mutationPlanCreationTimeInNs, mutationPlanExecutionTimeInNs,
+                                                    0, executeMutationTimeSpentInNs);
+                                        }
+                                        state.getMutationMetricQueue().addMetricsForTable(tableName, stagedMutationMetric);
 
                                         if (success) {
                                             TableMetricsManager.updateMetricsMethod(tableName, isUpsert ?
@@ -753,6 +777,7 @@ public Pair<Integer, ResultSet> call() throws SQLException {
                                             state.addExecuteMutationTime(
                                                     executeMutationTimeSpent, tableName);
                                         }
+
                                     }
                                 }
 
@@ -773,6 +798,25 @@ public Pair<Integer, ResultSet> call() throws SQLException {
         }
     }
 
+    /**
+     * Get mutation metrics for executeMutation call i.e. before the mutation are committed.
+     * All the times are in nano seconds.
+     * @param upsertMutationPlanCreationTime Time taken to create the upsert mutation plan.
+     * @param upsertMutationPlanExecutionTime Time taken to execute the upsert mutation plan.
+     * @param deleteMutationPlanCreationTime Time taken to create the delete mutation plan.
+     * @param deleteMutationPlanExecutionTime Time taken to execute the delete mutation plan.
+     * @param upsertExecuteMutationTime Time taken by upsert in executeMutation call.
+     * @param deleteExecuteMutationTime Time taken by delete in executeMutation call.
+     * @return MutationMetric object.
+     */
+    private MutationMetric getUncommittedMutationMetric(long upsertMutationPlanCreationTime, long upsertMutationPlanExecutionTime,
+                                                        long deleteMutationPlanCreationTime, long deleteMutationPlanExecutionTime,
+                                                        long upsertExecuteMutationTime, long deleteExecuteMutationTime) {
+        return new MutationMetric(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
+                upsertMutationPlanCreationTime, upsertMutationPlanExecutionTime, deleteMutationPlanCreationTime,
+                deleteMutationPlanExecutionTime, upsertExecuteMutationTime, deleteExecuteMutationTime);
+    }
+
     /**
      * Get different Result if the row is atomically deleted.
      *

phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MetricType.java

@@ -88,6 +88,24 @@ public enum MetricType {
     DELETE_BATCH_FAILED_COUNTER("dbfc", "Number of delete mutation batches that failed to be committed",
             LogLevel.OFF, PLong.INSTANCE),
 
+    UPSERT_PLAN_CREATION_TIME("upct", "Time taken to create the upsert mutation plan in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
+    UPSERT_PLAN_EXECUTION_TIME("upet", "Time taken to execute the upsert mutation plan in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
+    UPSERT_EXECUTE_MUTATION_TIME("uemt", "Time taken by upsert in executeMutation in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
+    DELETE_PLAN_CREATION_TIME("dpct", "Time taken to create the delete mutation plan in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
+    DELETE_PLAN_EXECUTION_TIME("dpet", "Time taken to execute the delete mutation plan in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
+    DELETE_EXECUTE_MUTATION_TIME("demt", "Time taken by delete in executeMutation in ns",
+            LogLevel.OFF, PLong.INSTANCE),
+
     // select-specific query (read) metrics updated during executeQuery
     SELECT_SUCCESS_SQL_COUNTER("sss", "Counter for number of select sql queries that successfully"
             + " passed the executeQuery phase", LogLevel.OFF, PLong.INSTANCE),

phoenix-core-client/src/main/java/org/apache/phoenix/monitoring/MutationMetricQueue.java

@@ -21,13 +21,19 @@
 import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_SIZE;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_COMMIT_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_EXECUTE_MUTATION_TIME;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_BYTES;
 import static org.apache.phoenix.monitoring.MetricType.DELETE_MUTATION_SQL_COUNTER;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_PLAN_CREATION_TIME;
+import static org.apache.phoenix.monitoring.MetricType.DELETE_PLAN_EXECUTION_TIME;
 import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_FAILED_SIZE;
 import static org.apache.phoenix.monitoring.MetricType.MUTATION_BATCH_SIZE;
 import static org.apache.phoenix.monitoring.MetricType.MUTATION_BYTES;
 import static org.apache.phoenix.monitoring.MetricType.MUTATION_COMMIT_TIME;
 import static org.apache.phoenix.monitoring.MetricType.INDEX_COMMIT_FAILURE_SIZE;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_EXECUTE_MUTATION_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_PLAN_CREATION_TIME;
+import static org.apache.phoenix.monitoring.MetricType.UPSERT_PLAN_EXECUTION_TIME;
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_COUNTER;
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_BATCH_FAILED_SIZE;
 import static org.apache.phoenix.monitoring.MetricType.UPSERT_COMMIT_TIME;
@@ -93,7 +99,12 @@ public Map<String, Map<MetricType, Long>> aggregate() {
             publishedMetricsForTable.put(metric.getUpsertBatchFailedCounter().getMetricType(), metric.getUpsertBatchFailedCounter().getValue());
             publishedMetricsForTable.put(metric.getDeleteBatchFailedSize().getMetricType(), metric.getDeleteBatchFailedSize().getValue());
             publishedMetricsForTable.put(metric.getDeleteBatchFailedCounter().getMetricType(), metric.getDeleteBatchFailedCounter().getValue());
-
+            publishedMetricsForTable.put(metric.getUpsertMutationPlanCreationTime().getMetricType(), metric.getUpsertMutationPlanCreationTime().getValue());
+            publishedMetricsForTable.put(metric.getUpsertMutationPlanExecutionTime().getMetricType(), metric.getUpsertMutationPlanExecutionTime().getValue());
+            publishedMetricsForTable.put(metric.getDeleteMutationPlanCreationTime().getMetricType(), metric.getDeleteMutationPlanCreationTime().getValue());
+            publishedMetricsForTable.put(metric.getDeleteMutationPlanExecutionTime().getMetricType(), metric.getDeleteMutationPlanExecutionTime().getValue());
+            publishedMetricsForTable.put(metric.getUpsertExecuteMutationTime().getMetricType(), metric.getUpsertExecuteMutationTime().getValue());
+            publishedMetricsForTable.put(metric.getDeleteExecuteMutationTime().getMetricType(), metric.getDeleteExecuteMutationTime().getValue());
         }
         return publishedMetrics;
     }
@@ -125,16 +136,26 @@ public static class MutationMetric {
         private final CombinableMetric numOfIndexCommitFailMutations = new CombinableMetricImpl(
                 INDEX_COMMIT_FAILURE_SIZE);
 
+        private final CombinableMetric upsertMutationPlanCreationTime = new CombinableMetricImpl(UPSERT_PLAN_CREATION_TIME);
+        private final CombinableMetric upsertMutationPlanExecutionTime = new CombinableMetricImpl(UPSERT_PLAN_EXECUTION_TIME);
+        private final CombinableMetric deleteMutationPlanCreationTime = new CombinableMetricImpl(DELETE_PLAN_CREATION_TIME);
+        private final CombinableMetric deleteMutationPlanExecutionTime = new CombinableMetricImpl(DELETE_PLAN_EXECUTION_TIME);
+        private final CombinableMetric upsertExecuteMutationTime = new CombinableMetricImpl(UPSERT_EXECUTE_MUTATION_TIME);
+        private final CombinableMetric deleteExecuteMutationTime = new CombinableMetricImpl(DELETE_EXECUTE_MUTATION_TIME);
+
+
         public static final MutationMetric EMPTY_METRIC =
-                new MutationMetric(0,0,0,0, 0, 0,0,0,0,0,0,0,0,0,0);
+                new MutationMetric(0,0,0,0, 0, 0,0,0,0,0,0,0,0,0,0, 0, 0, 0, 0, 0, 0);
 
         public MutationMetric(long numMutations, long upsertMutationsSizeBytes,
-                long deleteMutationsSizeBytes, long commitTimeForUpserts, long commitTimeForAtomicUpserts,
-                long commitTimeForDeletes, long numFailedMutations, long upsertMutationSqlCounterSuccess,
-                long deleteMutationSqlCounterSuccess, long totalMutationBytes,
-                long numOfPhase3Failed, long upsertBatchFailedSize,
-                long upsertBatchFailedCounter, long deleteBatchFailedSize,
-                long deleteBatchFailedCounter) {
+                              long deleteMutationsSizeBytes, long commitTimeForUpserts, long commitTimeForAtomicUpserts,
+                              long commitTimeForDeletes, long numFailedMutations, long upsertMutationSqlCounterSuccess,
+                              long deleteMutationSqlCounterSuccess, long totalMutationBytes,
+                              long numOfPhase3Failed, long upsertBatchFailedSize,
+                              long upsertBatchFailedCounter, long deleteBatchFailedSize,
+                              long deleteBatchFailedCounter, long upsertMutationPlanCreationTime,
+                              long upsertMutationPlanExecutionTime, long deleteMutationPlanCreationTime,
+                              long deleteMutationPlanExecutionTime, long upsertExecuteMutationTime, long deleteExecuteMutationTime) {
             this.numMutations.change(numMutations);
             this.totalCommitTimeForUpserts.change(commitTimeForUpserts);
             this.totalCommitTimeForAtomicUpserts.change(commitTimeForAtomicUpserts);
@@ -151,6 +172,12 @@ public MutationMetric(long numMutations, long upsertMutationsSizeBytes,
             this.upsertBatchFailedCounter.change(upsertBatchFailedCounter);
             this.deleteBatchFailedSize.change(deleteBatchFailedSize);
             this.deleteBatchFailedCounter.change(deleteBatchFailedCounter);
+            this.upsertMutationPlanCreationTime.change(upsertMutationPlanCreationTime);
+            this.upsertMutationPlanExecutionTime.change(upsertMutationPlanExecutionTime);
+            this.deleteMutationPlanCreationTime.change(deleteMutationPlanCreationTime);
+            this.deleteMutationPlanExecutionTime.change(deleteMutationPlanExecutionTime);
+            this.upsertExecuteMutationTime.change(upsertExecuteMutationTime);
+            this.deleteExecuteMutationTime.change(deleteExecuteMutationTime);
         }
 
         public CombinableMetric getTotalCommitTimeForUpserts() {
@@ -215,6 +242,30 @@ public CombinableMetric getDeleteBatchFailedCounter() {
             return deleteBatchFailedCounter;
         }
 
+        public CombinableMetric getUpsertMutationPlanCreationTime() {
+            return upsertMutationPlanCreationTime;
+        }
+
+        public CombinableMetric getUpsertMutationPlanExecutionTime() {
+            return upsertMutationPlanExecutionTime;
+        }
+
+        public CombinableMetric getDeleteMutationPlanCreationTime() {
+            return deleteMutationPlanCreationTime;
+        }
+
+        public CombinableMetric getDeleteMutationPlanExecutionTime() {
+            return deleteMutationPlanExecutionTime;
+        }
+
+        public CombinableMetric getUpsertExecuteMutationTime() {
+            return upsertExecuteMutationTime;
+        }
+
+        public CombinableMetric getDeleteExecuteMutationTime() {
+            return deleteExecuteMutationTime;
+        }
+
         public void combineMetric(MutationMetric other) {
             this.numMutations.combine(other.numMutations);
             this.totalCommitTimeForUpserts.combine(other.totalCommitTimeForUpserts);
@@ -232,6 +283,12 @@ public void combineMetric(MutationMetric other) {
             this.upsertBatchFailedCounter.combine(other.upsertBatchFailedCounter);
             this.deleteBatchFailedSize.combine(other.deleteBatchFailedSize);
             this.deleteBatchFailedCounter.combine(other.deleteBatchFailedCounter);
+            this.upsertMutationPlanCreationTime.combine(other.upsertMutationPlanCreationTime);
+            this.upsertMutationPlanExecutionTime.combine(other.upsertMutationPlanExecutionTime);
+            this.deleteMutationPlanCreationTime.combine(other.deleteMutationPlanCreationTime);
+            this.deleteMutationPlanExecutionTime.combine(other.deleteMutationPlanExecutionTime);
+            this.upsertExecuteMutationTime.combine(other.upsertExecuteMutationTime);
+            this.deleteExecuteMutationTime.combine(other.deleteExecuteMutationTime);
         }
 
     }