diff --git a/CHANGES.txt b/CHANGES.txt index 38b95c916b..f89d347673 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 05a1163faf..8bc2e917b8 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1530,6 +1530,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean, Memtable.Owner 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 diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index c81f9ab57f..e1c0a21834 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -718,6 +718,7 @@ public abstract class ReadCommand extends AbstractReadQuery metric.tombstoneScannedHistogram.update(tombstones); metric.liveScannedHistogram.update(liveRows); + metric.totalRowsRead.inc(liveRows); boolean warnTombstones = tombstones > warningThreshold && respectTombstoneThresholds; if (warnTombstones) diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index 9e00988da7..5330c47776 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/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 class TableMetrics 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"); diff --git a/test/unit/org/apache/cassandra/metrics/TableMetricsTest.java b/test/unit/org/apache/cassandra/metrics/TableMetricsTest.java index 5cb934af72..0025650940 100644 --- a/test/unit/org/apache/cassandra/metrics/TableMetricsTest.java +++ b/test/unit/org/apache/cassandra/metrics/TableMetricsTest.java @@ -24,6 +24,7 @@ import java.util.function.Supplier; 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 class TableMetricsTest 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() {