Add TotalRowsRead and TotalRowsMutated counters to TableMetrics for accurate per-table row throughput tracking

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

Author: pwalczak

CHANGES.txt

@@ -1,4 +1,5 @@
 6.0-alpha2
+ * 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)
  * Replace LongAdder in metric-like logic with ThreadLocalCounter (CASSANDRA-21400)

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

@@ -1530,6 +1530,7 @@ 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());
             // 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/ReadCommand.java

@@ -718,6 +718,7 @@ public void onClose()
 
                 metric.tombstoneScannedHistogram.update(tombstones);
                 metric.liveScannedHistogram.update(liveRows);
+                metric.totalRowsRead.inc(liveRows);
 
                 boolean warnTombstones = tombstones > warningThreshold && respectTombstoneThresholds;
                 if (warnTombstones)

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

@@ -166,6 +166,10 @@ public class TableMetrics
     public final TableHistogram purgeableTombstoneScannedHistogram;
     /** Live rows scanned in queries on this CF */
     public final TableHistogram liveScannedHistogram;
+    /** Total number of live rows read from this CF (cumulative counter, suitable for rate/windowed calculations) */
+    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;
     /** Column update time delta on this CF */
     public final TableHistogram colUpdateTimeDeltaHistogram;
     /** time taken acquiring the partition lock for materialized view updates for this table */
@@ -812,6 +816,8 @@ public Long getValue()
         tombstoneScannedHistogram = createTableHistogram("TombstoneScannedHistogram", cfs.keyspace.metric.tombstoneScannedHistogram, false);
         purgeableTombstoneScannedHistogram = createTableHistogram("PurgeableTombstoneScannedHistogram", cfs.keyspace.metric.purgeableTombstoneScannedHistogram, true);
         liveScannedHistogram = createTableHistogram("LiveScannedHistogram", cfs.keyspace.metric.liveScannedHistogram, false);
+        totalRowsRead = createTableCounter("TotalRowsRead");
+        totalRowsMutated = createTableCounter("TotalRowsMutated");
         colUpdateTimeDeltaHistogram = createTableHistogram("ColUpdateTimeDeltaHistogram", cfs.keyspace.metric.colUpdateTimeDeltaHistogram, false);
         coordinatorReadLatency = createTableTimer("CoordinatorReadLatency");
         coordinatorScanLatency = createTableTimer("CoordinatorScanLatency");

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

@@ -24,6 +24,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import com.codahale.metrics.Counter;
 import com.datastax.driver.core.BatchStatement;
 import com.datastax.driver.core.Cluster;
 import com.datastax.driver.core.PreparedStatement;
@@ -213,6 +214,61 @@ public void testPreparedStatementsExecuted()
         assertGreaterThan(cfs.metric.coordinatorWriteLatency.getMeanRate(), 0);
     }
 
+    @Test
+    public void testRowsMutatedCounter()
+    {
+        ColumnFamilyStore cfs = recreateTable();
+        assertEquals(0, cfs.metric.totalRowsMutated.getCount());
+
+        // Each INSERT touches exactly one row
+        session.execute(String.format("INSERT INTO %s.%s (id, val1, val2) VALUES (1, 'a', 'b')", KEYSPACE, TABLE));
+        assertEquals(1, cfs.metric.totalRowsMutated.getCount());
+
+        session.execute(String.format("INSERT INTO %s.%s (id, val1, val2) VALUES (2, 'c', 'd')", KEYSPACE, TABLE));
+        assertEquals(2, cfs.metric.totalRowsMutated.getCount());
+
+        // Batch of 3 rows — counter should jump by 3
+        executeBatch(false, 3, 1);
+        assertEquals(5, cfs.metric.totalRowsMutated.getCount());
+
+        assertMetricRows("TotalRowsMutated", cfs.metric.totalRowsMutated);
+    }
+
+    @Test
+    public void testRowsReadCounter()
+    {
+        ColumnFamilyStore cfs = recreateTable();
+        assertEquals(0, cfs.metric.totalRowsRead.getCount());
+
+        // Seed some rows
+        for (int i = 0; i < 5; i++)
+            session.execute(String.format("INSERT INTO %s.%s (id, val1, val2) VALUES (%d, 'v%d', 'x')", KEYSPACE, TABLE, i, i));
+
+        // Full-table scan should touch all 5 rows
+        session.execute(String.format("SELECT * FROM %s.%s", KEYSPACE, TABLE));
+        assertEquals(5, cfs.metric.totalRowsRead.getCount());
+
+        // Single-partition read touches 1 row
+        session.execute(String.format("SELECT * FROM %s.%s WHERE id = 0", KEYSPACE, TABLE));
+        assertEquals(6, cfs.metric.totalRowsRead.getCount());
+
+        assertMetricRows("TotalRowsRead", cfs.metric.totalRowsRead);
+    }
+
+    private void assertMetricRows(String metricName, Counter counter)
+    {
+        assertRowsContains(cluster, session.execute("SELECT * FROM system_metrics.table_group"),
+                           row("org.apache.cassandra.metrics.Table." + metricName + ".junit.tablemetricstest",
+                               "junit.tablemetricstest",
+                               "counter",
+                               String.valueOf(counter.getCount())));
+        assertRowsContains(cluster, session.execute("SELECT * FROM system_metrics.column_family_group"),
+                           row("org.apache.cassandra.metrics.ColumnFamily." + metricName + ".junit.tablemetricstest",
+                               "junit.tablemetricstest",
+                               "counter",
+                               String.valueOf(counter.getCount())));
+    }
+
     @Test
     public void testLoggedPartitionsPerBatch()
     {