Add rowsMutatedPerWriteHistogram metric to track rows mutated per write request

patch by Piotrek Walczak; reviewed by Dmitry Konstantinov, Stefan Miklosovic for CASSANDRA-21320

Author: walcp1

CHANGES.txt

@@ -1,4 +1,5 @@
 6.0-alpha2
+ * Add rowsMutatedPerWriteHistogram metric to track rows mutated per write request (CASSANDRA-21320)
  * Add TotalRowsRead and TotalRowsMutated counters to TableMetrics for accurate per-table row throughput tracking (CASSANDRA-21321)
  * Move exception handling of SPI startup checks when iterating over them (CASSANDRA-21409)
  * Avoid type lookup in SerializationHeader#getType if schema and SSTable are aligned (CASSANDRA-21402)

src/java/org/apache/cassandra/db/ColumnFamilyStore.java

@@ -1530,7 +1530,9 @@ public void apply(PartitionUpdate update, CassandraWriteContext context, boolean
                 metric.topWritePartitionSize.addSample(key.getKey(), update.dataSize());
             StorageHook.instance.reportWrite(metadata.id, update);
             metric.writeLatency.addNano(nanoTime() - start);
-            metric.totalRowsMutated.inc(update.affectedRowCount());
+            int affectedRows = update.affectedRowCount();
+            metric.totalRowsMutated.inc(affectedRows);
+            metric.rowsMutatedPerWriteHistogram.update(affectedRows);
             // CASSANDRA-11117 - certain resolution paths on memtable put can result in very
             // large time deltas, either through a variety of sentinel timestamps (used for empty values, ensuring
             // a minimal write, etc). This limits the time delta to the max value the histogram

src/java/org/apache/cassandra/db/virtual/TableMetricTables.java

@@ -78,6 +78,7 @@ public static Collection<VirtualTable> getAll(String name)
             new HistogramTableMetric(name, "tombstones_per_read", t -> t.tombstoneScannedHistogram.cf),
             new HistogramTableMetric(name, "purgeable_tombstones_per_read", t -> t.purgeableTombstoneScannedHistogram.cf),
             new HistogramTableMetric(name, "rows_per_read", t -> t.liveScannedHistogram.cf),
+            new HistogramTableMetric(name, "rows_per_write", t -> t.rowsMutatedPerWriteHistogram.cf),
             new StorageTableMetric(name, "disk_usage", (TableMetrics t) -> t.totalDiskSpaceUsed),
             new StorageTableMetric(name, "max_partition_size", (TableMetrics t) -> t.maxPartitionSize),
             new StorageTableMetric(name, "max_sstable_size", (TableMetrics t) -> t.maxSSTableSize),

src/java/org/apache/cassandra/metrics/KeyspaceMetrics.java

@@ -89,6 +89,8 @@ public class KeyspaceMetrics
     public final Histogram purgeableTombstoneScannedHistogram;
     /** Live cells scanned in queries on this Keyspace */
     public final Histogram liveScannedHistogram;
+    /** Rows mutated in writes on this Keyspace */
+    public final Histogram rowsMutatedPerWriteHistogram;
     /** Column update time delta on this Keyspace */
     public final Histogram colUpdateTimeDeltaHistogram;
     /** time taken acquiring the partition lock for materialized view updates on this keyspace */
@@ -258,6 +260,7 @@ public KeyspaceMetrics(final Keyspace ks)
         tombstoneScannedHistogram = createKeyspaceHistogram("TombstoneScannedHistogram", false);
         purgeableTombstoneScannedHistogram = createKeyspaceHistogram("PurgeableTombstoneScannedHistogram", false);
         liveScannedHistogram = createKeyspaceHistogram("LiveScannedHistogram", false);
+        rowsMutatedPerWriteHistogram = createKeyspaceHistogram("RowsMutatedPerWriteHistogram", false);
         colUpdateTimeDeltaHistogram = createKeyspaceHistogram("ColUpdateTimeDeltaHistogram", false);
         viewLockAcquireTime = createKeyspaceTimer("ViewLockAcquireTime");
         viewReadTime = createKeyspaceTimer("ViewReadTime");

src/java/org/apache/cassandra/metrics/TableMetrics.java

@@ -170,6 +170,8 @@ public class TableMetrics
     public final Counter totalRowsRead;
     /** Total number of rows mutated in writes to this CF (cumulative counter, suitable for rate/windowed calculations) */
     public final Counter totalRowsMutated;
+    /** Rows mutated in writes on this CF */
+    public final TableHistogram rowsMutatedPerWriteHistogram;
     /** Column update time delta on this CF */
     public final TableHistogram colUpdateTimeDeltaHistogram;
     /** time taken acquiring the partition lock for materialized view updates for this table */
@@ -818,6 +820,7 @@ public Long getValue()
         liveScannedHistogram = createTableHistogram("LiveScannedHistogram", cfs.keyspace.metric.liveScannedHistogram, false);
         totalRowsRead = createTableCounter("TotalRowsRead");
         totalRowsMutated = createTableCounter("TotalRowsMutated");
+        rowsMutatedPerWriteHistogram = createTableHistogram("RowsMutatedPerWriteHistogram", cfs.keyspace.metric.rowsMutatedPerWriteHistogram, false);
         colUpdateTimeDeltaHistogram = createTableHistogram("ColUpdateTimeDeltaHistogram", cfs.keyspace.metric.colUpdateTimeDeltaHistogram, false);
         coordinatorReadLatency = createTableTimer("CoordinatorReadLatency");
         coordinatorScanLatency = createTableTimer("CoordinatorScanLatency");

test/distributed/org/apache/cassandra/distributed/test/metric/TableMetricTest.java

@@ -62,7 +62,7 @@ public class TableMetricTest extends TestBaseImpl
         ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(false);
     }
     private static volatile Map<String, Collection<String>> SYSTEM_TABLES = null;
-    private static Set<String> TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency");
+    private static Set<String> TABLE_METRIC_NAMES = ImmutableSet.of("WriteLatency", "RowsMutatedPerWriteHistogram");
 
     /**
      * Makes sure that all system tables have the expected metrics

test/unit/org/apache/cassandra/metrics/TableMetricsTest.java

@@ -269,6 +269,32 @@ private void assertMetricRows(String metricName, Counter counter)
                                String.valueOf(counter.getCount())));
     }
 
+    @Test
+    public void testRowsMutatedPerWriteHistogram()
+    {
+        ColumnFamilyStore cfs = recreateTable();
+        assertEquals(0, cfs.metric.rowsMutatedPerWriteHistogram.cf.getCount());
+
+        session.execute(String.format("INSERT INTO %s.%s (id, val1, val2) VALUES (1, 'a', 'b')", KEYSPACE, TABLE));
+        session.execute(String.format("INSERT INTO %s.%s (id, val1, val2) VALUES (1, 'c', 'd')", KEYSPACE, TABLE));
+
+        assertEquals(2, cfs.metric.rowsMutatedPerWriteHistogram.cf.getCount());
+        assertGreaterThan(cfs.metric.rowsMutatedPerWriteHistogram.cf.getSnapshot().getMax(), 0);
+
+        String histogramMedian = Double.toString(cfs.metric.rowsMutatedPerWriteHistogram.cf.getSnapshot().getMedian());
+
+        assertRowsContains(cluster, session.execute("SELECT * FROM system_metrics.table_group"),
+                           row("org.apache.cassandra.metrics.Table.RowsMutatedPerWriteHistogram.junit.tablemetricstest",
+                               "junit.tablemetricstest",
+                               "histogram",
+                               histogramMedian));
+        assertRowsContains(cluster, session.execute("SELECT * FROM system_metrics.column_family_group"),
+                           row("org.apache.cassandra.metrics.ColumnFamily.RowsMutatedPerWriteHistogram.junit.tablemetricstest",
+                               "junit.tablemetricstest",
+                               "histogram",
+                               histogramMedian));
+    }
+
     @Test
     public void testLoggedPartitionsPerBatch()
     {