From 6a979dc67ff155a8bc34e8b52ebaa9698ba7a537 Mon Sep 17 00:00:00 2001 From: Tanuj Khurana Date: Thu, 20 May 2021 17:08:11 -0700 Subject: [PATCH 1/2] PHOENIX-6474 Client and server metrics for atomic upserts --- .../PhoenixTableLevelMetricsIT.java | 46 +++++++++++++++++++ .../apache/phoenix/execute/MutationState.java | 25 ++++++++-- .../hbase/index/IndexRegionObserver.java | 7 +++ .../apache/phoenix/jdbc/PhoenixStatement.java | 10 ++++ .../apache/phoenix/monitoring/MetricType.java | 6 +++ .../monitoring/MutationMetricQueue.java | 13 ++++-- .../monitoring/TableClientMetrics.java | 8 +++- 7 files changed, 107 insertions(+), 8 deletions(-) diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java index b1215cd0599..2c1aaf925e9 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixTableLevelMetricsIT.java @@ -56,6 +56,8 @@ import static org.apache.phoenix.exception.SQLExceptionCode.DATA_EXCEEDS_MAX_CAPACITY; import static org.apache.phoenix.exception.SQLExceptionCode.GET_TABLE_REGIONS_FAIL; import static org.apache.phoenix.exception.SQLExceptionCode.OPERATION_TIMED_OUT; +import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_COMMIT_TIME; +import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_SQL_COUNTER; import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER; import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_SUCCESS_SQL_COUNTER; import static org.apache.phoenix.monitoring.MetricType.DELETE_BATCH_FAILED_COUNTER; @@ -1148,6 +1150,50 @@ private static void assertMetricValue(Metric m, MetricType checkType, long compa } } + @Test public void testTableLevelMetricsForAtomicUpserts() throws Throwable { + String tableName = generateUniqueName(); + Connection conn = null; + Throwable exception = null; + int numAtomicUpserts = 4; + try { + conn = getConnFromTestDriver(); + String ddl = "create table " + tableName + "(pk varchar primary key, counter1 bigint)"; + conn.createStatement().execute(ddl); + String dml; + ResultSet rs; + dml = String.format("UPSERT INTO %s VALUES('a', 0)", tableName); + conn.createStatement().execute(dml); + dml = String.format("UPSERT INTO %s VALUES('a', 0) ON DUPLICATE KEY UPDATE counter1 = counter1 + 1", tableName); + for (int i = 0; i < numAtomicUpserts; ++i) { + conn.createStatement().execute(dml); + } + conn.commit(); + String dql = String.format("SELECT counter1 FROM %s WHERE counter1 > 0", tableName); + rs = conn.createStatement().executeQuery(dql); + assertTrue(rs.next()); + assertEquals(4, rs.getInt(1)); + }catch (Throwable t) { + exception = t; + } finally { + // Otherwise the test fails with an error from assertions below instead of the real exception + if (exception != null) { + throw exception; + } + assertNotNull("Failed to get a connection!", conn); + // Get write metrics before closing the connection since that clears those metrics + Map + writeMutMetrics = + getWriteMetricInfoForMutationsSinceLastReset(conn).get(tableName); + conn.close(); + // 1 regular upsert + numAtomicUpserts + // 2 mutations (regular and atomic on the same row in the same batch will be split) + assertMutationTableMetrics(true, tableName, 1 + numAtomicUpserts, 0, 0, true, 2, 0, 0, 2, 0, + writeMutMetrics, conn); + assertEquals(numAtomicUpserts, getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_SQL_COUNTER)); + assertTrue(getMetricFromTableMetrics(tableName, ATOMIC_UPSERT_COMMIT_TIME) > 0); + } + } + private Connection getConnFromTestDriver() throws SQLException { Connection conn = DriverManager.getConnection(url); assertTrue(conn.unwrap(PhoenixConnection.class) 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 1efbcf047f3..68c0d0234a3 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 @@ -955,6 +955,7 @@ static MutationBytes calculateMutationSize(List mutations, boolean u long tempSize; long deleteSize = 0, deleteCounter = 0; long upsertsize = 0, upsertCounter = 0; + long atomicUpsertsize = 0; if (GlobalClientMetrics.isMetricsEnabled()) { for (Mutation mutation : mutations) { tempSize = KeyValueUtil.calculateMutationDiskSize(mutation); @@ -966,6 +967,9 @@ static MutationBytes calculateMutationSize(List mutations, boolean u }else if(mutation instanceof Put) { upsertsize += tempSize; upsertCounter++; + if (mutation.getAttribute(PhoenixIndexBuilder.ATOMIC_OP_ATTRIB) != null) { + atomicUpsertsize += tempSize; + } allDeletesMutations = false; } else { allUpsertsMutations = false; @@ -976,7 +980,7 @@ static MutationBytes calculateMutationSize(List mutations, boolean u if (updateGlobalClientMetrics) { GLOBAL_MUTATION_BYTES.update(byteSize); } - return new MutationBytes(deleteCounter, deleteSize, byteSize, upsertCounter, upsertsize); + return new MutationBytes(deleteCounter, deleteSize, byteSize, upsertCounter, upsertsize, atomicUpsertsize); } public long getBatchSizeBytes() { @@ -994,14 +998,16 @@ public static final class MutationBytes { private long totalMutationBytes; private long upsertMutationCounter; private long upsertMutationBytes; + private long atomicUpsertMutationBytes; // needed to calculate atomic upsert commit time - public MutationBytes(long deleteMutationCounter, long deleteMutationBytes, long totalMutationBytes, long - upsertMutationCounter, long upsertMutationBytes) { + public MutationBytes(long deleteMutationCounter, long deleteMutationBytes, long totalMutationBytes, + long upsertMutationCounter, long upsertMutationBytes, long atomicUpsertMutationBytes) { this.deleteMutationCounter = deleteMutationCounter; this.deleteMutationBytes = deleteMutationBytes; this.totalMutationBytes = totalMutationBytes; this.upsertMutationCounter = upsertMutationCounter; this.upsertMutationBytes = upsertMutationBytes; + this.atomicUpsertMutationBytes = atomicUpsertMutationBytes; } @@ -1024,6 +1030,8 @@ public long getUpsertMutationCounter() { public long getUpsertMutationBytes() { return upsertMutationBytes; } + + public long getAtomicUpsertMutationBytes() { return atomicUpsertMutationBytes; } } public enum MutationMetadataType { @@ -1542,7 +1550,7 @@ public static MutationMetricQueue.MutationMetric updateMutationBatchFailureMetri // in case we are dealing with all deletes for a non-transactional table, since there is a // bug in sendMutations where we don't get the correct value for numFailedMutations when // we don't use transactions - return new MutationMetricQueue.MutationMetric(0, 0, 0, 0, 0, + return new MutationMetricQueue.MutationMetric(0, 0, 0, 0, 0, 0, allDeletesMutations && !isTransactional ? numDeleteMutationsInBatch : numFailedMutations, 0, 0, 0, 0, numUpsertMutationsInBatch, @@ -1571,6 +1579,8 @@ static MutationMetric getCommittedMutationsMetric( long numFailedPhase3Mutations, long mutationCommitTime) { long committedUpsertMutationBytes = totalMutationBytesObject == null ? 0 : totalMutationBytesObject.getUpsertMutationBytes(); + long committedAtomicUpsertMutationBytes = totalMutationBytesObject == null ? 0: + totalMutationBytesObject.getAtomicUpsertMutationBytes(); long committedDeleteMutationBytes = totalMutationBytesObject == null ? 0 : totalMutationBytesObject.getDeleteMutationBytes(); long committedUpsertMutationCounter = totalMutationBytesObject == null ? 0 : @@ -1580,6 +1590,7 @@ static MutationMetric getCommittedMutationsMetric( long committedTotalMutationBytes = totalMutationBytesObject == null ? 0 : totalMutationBytesObject.getTotalMutationBytes(); long upsertMutationCommitTime = 0L; + long atomicUpsertMutationCommitTime = 0L; long deleteMutationCommitTime = 0L; if (totalMutationBytesObject != null && numFailedMutations != 0) { @@ -1592,6 +1603,8 @@ static MutationMetric getCommittedMutationsMetric( calculateMutationSize(uncommittedMutationsList, false); committedUpsertMutationBytes -= uncommittedMutationBytesObject.getUpsertMutationBytes(); + committedAtomicUpsertMutationBytes -= + uncommittedMutationBytesObject.getAtomicUpsertMutationBytes(); committedDeleteMutationBytes -= uncommittedMutationBytesObject.getDeleteMutationBytes(); committedUpsertMutationCounter -= @@ -1606,6 +1619,9 @@ static MutationMetric getCommittedMutationsMetric( upsertMutationCommitTime = (long)Math.floor((double)(committedUpsertMutationBytes * mutationCommitTime)/ committedTotalMutationBytes); + atomicUpsertMutationCommitTime = + (long)Math.floor((double)(committedAtomicUpsertMutationBytes * mutationCommitTime)/ + committedTotalMutationBytes); deleteMutationCommitTime = (long)Math.ceil((double)(committedDeleteMutationBytes * mutationCommitTime)/ committedTotalMutationBytes); @@ -1614,6 +1630,7 @@ static MutationMetric getCommittedMutationsMetric( committedUpsertMutationBytes, committedDeleteMutationBytes, upsertMutationCommitTime, + atomicUpsertMutationCommitTime, deleteMutationCommitTime, 0, // num failed mutations have been counted already in updateMutationBatchFailureMetrics() committedUpsertMutationCounter, diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java index a70b39a1bd2..96b4f18c493 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/IndexRegionObserver.java @@ -1101,19 +1101,26 @@ public void preBatchMutateWithExceptions(ObserverContext> aggregate() { publishedMetricsForTable.put(metric.getDeleteMutationsSizeBytes().getMetricType(), metric.getDeleteMutationsSizeBytes().getValue()); publishedMetricsForTable.put(metric.getCommitTimeForMutations().getMetricType(), metric.getCommitTimeForMutations().getValue()); publishedMetricsForTable.put(metric.getTotalCommitTimeForUpserts().getMetricType(), metric.getTotalCommitTimeForUpserts().getValue()); + publishedMetricsForTable.put(metric.getTotalCommitTimeForAtomicUpserts().getMetricType(), metric.getTotalCommitTimeForAtomicUpserts().getValue()); publishedMetricsForTable.put(metric.getTotalCommitTimeForDeletes().getMetricType(), metric.getTotalCommitTimeForDeletes().getValue()); publishedMetricsForTable.put(metric.getNumFailedMutations().getMetricType(), metric.getNumFailedMutations().getValue()); publishedMetricsForTable.put(metric.getNumOfIndexCommitFailedMutations().getMetricType(), metric.getNumOfIndexCommitFailedMutations().getValue()); @@ -110,6 +112,7 @@ public static class MutationMetric { private final CombinableMetric totalCommitTimeForMutations = new CombinableMetricImpl(MUTATION_COMMIT_TIME); private final CombinableMetric numFailedMutations = new CombinableMetricImpl(MUTATION_BATCH_FAILED_SIZE); private final CombinableMetric totalCommitTimeForUpserts = new CombinableMetricImpl(UPSERT_COMMIT_TIME); + private final CombinableMetric totalCommitTimeForAtomicUpserts = new CombinableMetricImpl(ATOMIC_UPSERT_COMMIT_TIME); private final CombinableMetric totalCommitTimeForDeletes = new CombinableMetricImpl(DELETE_COMMIT_TIME); private final CombinableMetric upsertMutationsSizeBytes = new CombinableMetricImpl(UPSERT_MUTATION_BYTES); private final CombinableMetric deleteMutationsSizeBytes = new CombinableMetricImpl(DELETE_MUTATION_BYTES); @@ -124,17 +127,18 @@ public static class MutationMetric { INDEX_COMMIT_FAILURE_SIZE); public static final MutationMetric EMPTY_METRIC = - new MutationMetric(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); public MutationMetric(long numMutations, long upsertMutationsSizeBytes, - long deleteMutationsSizeBytes, long commitTimeForUpserts, long commitTimeForDeletes, - long numFailedMutations, long upsertMutationSqlCounterSuccess, + 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) { this.numMutations.change(numMutations); this.totalCommitTimeForUpserts.change(commitTimeForUpserts); + this.totalCommitTimeForAtomicUpserts.change(commitTimeForAtomicUpserts); this.totalCommitTimeForDeletes.change(commitTimeForDeletes); this.totalCommitTimeForMutations.change(commitTimeForUpserts + commitTimeForDeletes); this.numFailedMutations.change(numFailedMutations); @@ -154,6 +158,8 @@ public CombinableMetric getTotalCommitTimeForUpserts() { return totalCommitTimeForUpserts; } + public CombinableMetric getTotalCommitTimeForAtomicUpserts() { return totalCommitTimeForAtomicUpserts; } + public CombinableMetric getTotalCommitTimeForDeletes() { return totalCommitTimeForDeletes; } @@ -213,6 +219,7 @@ public CombinableMetric getDeleteBatchFailedCounter() { public void combineMetric(MutationMetric other) { this.numMutations.combine(other.numMutations); this.totalCommitTimeForUpserts.combine(other.totalCommitTimeForUpserts); + this.totalCommitTimeForAtomicUpserts.combine(other.totalCommitTimeForAtomicUpserts); this.totalCommitTimeForDeletes.combine(other.totalCommitTimeForDeletes); this.totalCommitTimeForMutations.combine(other.totalCommitTimeForMutations); this.numFailedMutations.combine(other.numFailedMutations); diff --git a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java index b640f296a84..13ef8562bc4 100644 --- a/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java +++ b/phoenix-core/src/main/java/org/apache/phoenix/monitoring/TableClientMetrics.java @@ -68,6 +68,9 @@ import static org.apache.phoenix.monitoring.MetricType.DELETE_AGGREGATE_FAILURE_SQL_COUNTER; import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_SUCCESS_SQL_COUNTER; import static org.apache.phoenix.monitoring.MetricType.SELECT_AGGREGATE_FAILURE_SQL_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_COMMIT_TIME; +import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_SQL_COUNTER; +import static org.apache.phoenix.monitoring.MetricType.ATOMIC_UPSERT_SQL_QUERY_TIME; /** * This is used by TableMetricsManager class to store instance of @@ -121,7 +124,10 @@ public enum TableMetrics { DELETE_AGGREGATE_SUCCESS_SQL_COUNTER), TABLE_DELETE_AGGREGATE_FAILURE_SQL_COUNTER( DELETE_AGGREGATE_FAILURE_SQL_COUNTER), TABLE_SELECT_AGGREGATE_SUCCESS_SQL_COUNTER( SELECT_AGGREGATE_SUCCESS_SQL_COUNTER), TABLE_SELECT_AGGREGATE_FAILURE_SQL_COUNTER( - SELECT_AGGREGATE_FAILURE_SQL_COUNTER); + SELECT_AGGREGATE_FAILURE_SQL_COUNTER), + TABLE_ATOMIC_UPSERT_SQL_COUNTER(ATOMIC_UPSERT_SQL_COUNTER), + TABLE_ATOMIC_UPSERT_COMMIT_TIME(ATOMIC_UPSERT_COMMIT_TIME), + TABLE_ATOMIC_UPSERT_SQL_QUERY_TIME(ATOMIC_UPSERT_SQL_QUERY_TIME); private final MetricType metricType; private PhoenixTableMetric metric; From 2756b7da83734e6b6dfa59a2bfcca3c137a93545 Mon Sep 17 00:00:00 2001 From: Tanuj Khurana Date: Mon, 24 May 2021 10:29:54 -0700 Subject: [PATCH 2/2] Fixed failing tests related to metrics --- .../org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java | 2 +- .../it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java index 7c589452613..45b3561a0b8 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/BasePhoenixMetricsIT.java @@ -113,7 +113,7 @@ static void assertMutationMetrics(String tableName, int numRows, boolean isUpser String t = entry.getKey(); assertEquals("Table names didn't match!", tableName, t); Map p = entry.getValue(); - assertEquals("There should have been fifteen metrics", 15, p.size()); + assertEquals("There should have been sixteen metrics", 16, p.size()); boolean mutationBatchSizePresent = false; boolean mutationCommitTimePresent = false; boolean mutationBytesPresent = false; diff --git a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java index dc27dee8423..32d2f678164 100644 --- a/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java +++ b/phoenix-core/src/it/java/org/apache/phoenix/monitoring/PhoenixMetricsIT.java @@ -487,7 +487,7 @@ public void testMetricsForUpsert() throws Exception { String t = entry.getKey(); assertEquals("Table names didn't match!", tableName, t); Map p = entry.getValue(); - assertEquals("There should have been five metrics", 15, p.size()); + assertEquals("There should have been sixteen metrics", 16, p.size()); boolean mutationBatchSizePresent = false; boolean mutationCommitTimePresent = false; boolean mutationBytesPresent = false;