diff --git a/CHANGES.txt b/CHANGES.txt
index 2fc4f017b4..f48dc48bc6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.2
+ * Coordinator level metrics for read response and mutation row and column counts (CASSANDRA-18155)
  * Add CQL functions for dynamic data masking (CASSANDRA-17941)
  * Print friendly error when nodetool attempts to connect to uninitialized server (CASSANDRA-11537)
  * Use G1GC by default, and update default G1GC settings (CASSANDRA-18027)
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index c2992e6e4b..3aff5fe956 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -1064,6 +1064,8 @@ public class Config
 
     public volatile boolean sstable_read_rate_persistence_enabled = false;
 
+    public volatile boolean client_request_size_metrics_enabled = true;
+
     public volatile int max_top_size_partition_count = 10;
     public volatile int max_top_tombstone_partition_count = 10;
     public volatile DataStorageSpec.LongBytesBound min_tracked_partition_size = new DataStorageSpec.LongBytesBound("1MiB");
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index af11e89b74..1cf36a064b 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -4572,4 +4572,14 @@ public class DatabaseDescriptor
             conf.sstable_read_rate_persistence_enabled = enabled;
         }
     }
+
+    public static boolean getClientRequestSizeMetricsEnabled()
+    {
+        return conf.client_request_size_metrics_enabled;
+    }
+
+    public static void setClientRequestSizeMetricsEnabled(boolean enabled)
+    {
+        conf.client_request_size_metrics_enabled = enabled;
+    }
 }
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java
index c1d0c52ca7..146ec20557 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/ClusteringColumnRestrictions.java
@@ -228,5 +228,4 @@ final class ClusteringColumnRestrictions extends RestrictionSetWrapper
     {
         return restriction.isContains() || restriction.isLIKE() || index != restriction.getFirstColumn().position();
     }
-
 }
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java
index 7a5d5b964b..344498ce0e 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSet.java
@@ -105,6 +105,15 @@ final class RestrictionSet implements Restrictions, Iterable<SingleRestriction>
         return new ArrayList<>(restrictions.keySet());
     }
 
+    /**
+     * @return a direct reference to the key set from {@link #restrictions} with no defenseive copying
+     */
+    @Override
+    public Collection<ColumnMetadata> getColumnDefinitions()
+    {
+        return restrictions.keySet();
+    }
+
     @Override
     public void addFunctionsTo(List<Function> functions)
     {
@@ -332,7 +341,6 @@ final class RestrictionSet implements Restrictions, Iterable<SingleRestriction>
     /**
      * {@code Iterator} decorator that removes duplicates in an ordered one.
      *
-     * @param iterator the decorated iterator
      * @param <E> the iterator element type.
      */
     private static final class DistinctIterator<E> extends AbstractIterator<E>
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSetWrapper.java b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSetWrapper.java
index 9803adc459..8c89bca314 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSetWrapper.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/RestrictionSetWrapper.java
@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.cql3.restrictions;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 
@@ -57,6 +58,17 @@ class RestrictionSetWrapper implements Restrictions
         return restrictions.getColumnDefs();
     }
 
+    @Override
+    public Collection<ColumnMetadata> getColumnDefinitions()
+    {
+        return restrictions.getColumnDefinitions();
+    }
+
+    public RestrictionSet getRestrictionSet()
+    {
+        return restrictions;
+    }
+
     public void addFunctionsTo(List<Function> functions)
     {
         restrictions.addFunctionsTo(functions);
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/Restrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/Restrictions.java
index 77e0dd92fe..0ad7530e76 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/Restrictions.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/Restrictions.java
@@ -17,6 +17,7 @@
  */
 package org.apache.cassandra.cql3.restrictions;
 
+import java.util.Collection;
 import java.util.Set;
 
 import org.apache.cassandra.schema.ColumnMetadata;
@@ -34,6 +35,15 @@ public interface Restrictions extends Restriction
      */
     Set<Restriction> getRestrictions(ColumnMetadata columnDef);
 
+    /**
+     * This method exists in addition to {@link #getColumnDefs()} in case implementations want to
+     * provide columns definitions that are not strictly in position order.
+     */
+    default Collection<ColumnMetadata> getColumnDefinitions()
+    {
+        return getColumnDefs();
+    }
+
     /**
      * Checks if this <code>Restrictions</code> is empty or not.
      *
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/SingleRestriction.java b/src/java/org/apache/cassandra/cql3/restrictions/SingleRestriction.java
index 42b0b4ece7..989d5dabdf 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/SingleRestriction.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/SingleRestriction.java
@@ -51,6 +51,14 @@ public interface SingleRestriction extends Restriction
         return false;
     }
 
+    /**
+     * @return <code>true</code> if this restriction is based on equality comparison rather than a range or negation
+     */
+    default boolean isEqualityBased()
+    {
+        return isEQ() || isIN() || isContains();
+    }
+
     public default boolean isNotNull()
     {
         return false;
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
index 371bc589c3..b74d8adc43 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
@@ -390,6 +390,54 @@ public final class StatementRestrictions
                            .anyMatch(p -> ((SingleRestriction) p).isEQ());
     }
 
+    /**
+     * This method determines whether a specified column is restricted on equality or something equivalent, like IN.
+     * It can be used in conjunction with the columns selected by a query to determine which of those columns is 
+     * already bound by the client (and from its perspective, not retrieved by the database).
+     *
+     * @param column a column from the same table these restrictions are against
+     *
+     * @return <code>true</code> if the given column is restricted on equality
+     */
+    public boolean isEqualityRestricted(ColumnMetadata column)
+    {
+        if (column.kind == ColumnMetadata.Kind.PARTITION_KEY)
+        {
+            if (partitionKeyRestrictions.hasOnlyEqualityRestrictions())
+                for (ColumnMetadata restricted : partitionKeyRestrictions.getColumnDefinitions())
+                    if (restricted.name.equals(column.name))
+                        return true;
+        }
+        else if (column.kind == ColumnMetadata.Kind.CLUSTERING)
+        {
+            if (hasClusteringColumnsRestrictions())
+            {
+                for (SingleRestriction restriction : clusteringColumnsRestrictions.getRestrictionSet())
+                {
+                    if (restriction.isEqualityBased())
+                    {
+                        if (restriction.isMultiColumn())
+                        {
+                            for (ColumnMetadata restricted : restriction.getColumnDefs())
+                                if (restricted.name.equals(column.name))
+                                    return true;
+                        }
+                        else if (restriction.getFirstColumn().name.equals(column.name))
+                            return true;
+                    }
+                }
+            }
+        }
+        else if (hasNonPrimaryKeyRestrictions())
+        {
+            for (SingleRestriction restriction : nonPrimaryKeyRestrictions)
+                if (restriction.getFirstColumn().name.equals(column.name) && restriction.isEqualityBased())
+                    return true;
+        }
+
+        return false;
+    }
+
     /**
      * Returns the <code>Restrictions</code> for the specified type of columns.
      *
diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
index ab761ad9c1..b034db4482 100644
--- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
@@ -42,6 +42,7 @@ import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.db.rows.RowIterator;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.metrics.BatchMetrics;
+import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
 import org.apache.cassandra.service.*;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -442,6 +443,7 @@ public class BatchStatement implements CQLStatement
 
         boolean mutateAtomic = (isLogged() && mutations.size() > 1);
         StorageProxy.mutateWithTriggers(mutations, cl, mutateAtomic, queryStartNanoTime);
+        ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(mutations);
     }
 
     private void updatePartitionsPerBatchMetrics(int updatedPartitions)
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index c69cf4945a..c40109c607 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -33,6 +33,7 @@ import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.ReplicaLayout;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.ViewMetadata;
 import org.apache.cassandra.cql3.*;
@@ -51,6 +52,7 @@ import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.RowIterator;
 import org.apache.cassandra.db.view.View;
 import org.apache.cassandra.exceptions.*;
+import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.service.StorageProxy;
@@ -513,8 +515,13 @@ public abstract class ModificationStatement implements CQLStatement.SingleKeyspa
                          options.getNowInSeconds(queryState),
                          queryStartNanoTime);
         if (!mutations.isEmpty())
+        {
             StorageProxy.mutateWithTriggers(mutations, cl, false, queryStartNanoTime);
 
+            if (!SchemaConstants.isSystemKeyspace(metadata.keyspace))
+                ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(mutations);
+        }
+
         return null;
     }
 
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 75c12f065e..774fb3b77a 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -60,6 +60,7 @@ import org.apache.cassandra.db.rows.RowIterator;
 import org.apache.cassandra.db.view.View;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.exceptions.*;
+import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
 import org.apache.cassandra.index.IndexRegistry;
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.service.ClientState;
@@ -260,21 +261,30 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement
 
         if (options.isReadThresholdsEnabled())
             query.trackWarnings();
+        ResultMessage.Rows rows;
 
         if (aggregationSpec == null && (pageSize <= 0 || (query.limits().count() <= pageSize)))
-            return execute(query, options, state.getClientState(), selectors, nowInSec, userLimit, null, queryStartNanoTime);
+        {
+            rows = execute(query, options, state.getClientState(), selectors, nowInSec, userLimit, null, queryStartNanoTime);
+        }
+        else
+        {
+            QueryPager pager = getPager(query, options);
 
-        QueryPager pager = getPager(query, options);
+            rows = execute(state,
+                           Pager.forDistributedQuery(pager, cl, state.getClientState()),
+                           options,
+                           selectors,
+                           pageSize,
+                           nowInSec,
+                           userLimit,
+                           aggregationSpec,
+                           queryStartNanoTime);
+        }
+        if (!SchemaConstants.isSystemKeyspace(table.keyspace))
+            ClientRequestSizeMetrics.recordReadResponseMetrics(rows, restrictions, selection);
 
-        return execute(state,
-                       Pager.forDistributedQuery(pager, cl, state.getClientState()),
-                       options,
-                       selectors,
-                       pageSize,
-                       nowInSec,
-                       userLimit,
-                       aggregationSpec,
-                       queryStartNanoTime);
+        return rows;
     }
 
 
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
index 9f854c670a..7f2ca7d465 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
@@ -466,6 +466,62 @@ public class PartitionUpdate extends AbstractBTreePartition
         return marks;
     }
 
+    /**
+     *
+     * @return the estimated number of rows affected by this mutation 
+     */
+    public int affectedRowCount()
+    {
+        // If there is a partition-level deletion, we intend to delete at least one row.
+        if (!partitionLevelDeletion().isLive())
+            return 1;
+
+        int count = 0;
+
+        // Each range delete should correspond to at least one intended row deletion.
+        if (deletionInfo().hasRanges())
+            count += deletionInfo().rangeCount();
+
+        count += rowCount();
+
+        if (!staticRow().isEmpty())
+            count++;
+
+        return count;
+    }
+
+    /**
+     *
+     * @return the estimated total number of columns that either have live data or are covered by a delete
+     */
+    public int affectedColumnCount()
+    {
+        // If there is a partition-level deletion, we intend to delete at least the columns of one row.
+        if (!partitionLevelDeletion().isLive())
+            return metadata().regularAndStaticColumns().size();
+
+        int count = 0;
+
+        // Each range delete should correspond to at least one intended row deletion, and with it, its regular columns.
+        if (deletionInfo().hasRanges())
+            count += deletionInfo().rangeCount() * metadata().regularColumns().size();
+
+        for (Row row : this)
+        {
+            if (row.deletion().isLive())
+                // If the row is live, this will include simple tombstones as well as cells w/ actual data. 
+                count += row.columnCount();
+            else
+                // We have a row deletion, so account for the columns that might be deleted.
+                count += metadata().regularColumns().size();
+        }
+
+        if (!staticRow().isEmpty())
+            count += staticRow().columnCount();
+
+        return count;
+    }
+
     private static void addMarksForRow(Row row, List<CounterMark> marks)
     {
         for (Cell<?> cell : row.cells())
diff --git a/src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java b/src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java
new file mode 100644
index 0000000000..4def87ea16
--- /dev/null
+++ b/src/java/org/apache/cassandra/metrics/ClientRequestSizeMetrics.java
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.metrics;
+
+import java.util.Collection;
+
+import com.codahale.metrics.Counter;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.restrictions.StatementRestrictions;
+import org.apache.cassandra.cql3.selection.Selection;
+import org.apache.cassandra.db.IMutation;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.transport.messages.ResultMessage;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+
+public class ClientRequestSizeMetrics
+{
+    private static final String TYPE = "ClientRequestSize";
+
+    public static final Counter totalColumnsRead = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "ColumnsRead", null));
+    public static final Counter totalRowsRead = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "RowsRead", null));
+    public static final Counter totalColumnsWritten = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "ColumnsWritten", null));
+    public static final Counter totalRowsWritten = Metrics.counter(DefaultNameFactory.createMetricName(TYPE, "RowsWritten", null));
+
+    public static void recordReadResponseMetrics(ResultMessage.Rows rows, StatementRestrictions restrictions, Selection selection)
+    {
+        if (!DatabaseDescriptor.getClientRequestSizeMetricsEnabled())
+            return;
+
+        int rowCount = rows.result.size();
+        ClientRequestSizeMetrics.totalRowsRead.inc(rowCount);
+        
+        int nonRestrictedColumns = selection.getColumns().size();
+        
+        for (ColumnMetadata column : selection.getColumns())
+            if (restrictions.isEqualityRestricted(column))
+                nonRestrictedColumns--;
+            
+        long columnCount = (long) rowCount * nonRestrictedColumns;
+        ClientRequestSizeMetrics.totalColumnsRead.inc(columnCount);
+    }
+
+    public static void recordRowAndColumnCountMetrics(Collection<? extends IMutation> mutations)
+    {
+        if (!DatabaseDescriptor.getClientRequestSizeMetricsEnabled())
+            return;
+
+        int rowCount = 0;
+        int columnCount = 0;
+
+        for (IMutation mutation : mutations)
+        {
+            for (PartitionUpdate update : mutation.getPartitionUpdates())
+            {
+                columnCount += update.affectedColumnCount();
+                rowCount += update.affectedRowCount();
+            }
+        }
+
+        ClientRequestSizeMetrics.totalColumnsWritten.inc(columnCount);
+        ClientRequestSizeMetrics.totalRowsWritten.inc(rowCount);
+    }
+
+    public static void recordRowAndColumnCountMetrics(PartitionUpdate update)
+    {
+        if (!DatabaseDescriptor.getClientRequestSizeMetricsEnabled())
+            return;
+
+        ClientRequestSizeMetrics.totalColumnsWritten.inc(update.affectedColumnCount());
+        ClientRequestSizeMetrics.totalRowsWritten.inc(update.affectedRowCount());
+    }
+}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 52039c1d9f..0aafa9539f 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -108,6 +108,7 @@ import org.apache.cassandra.locator.ReplicaPlan;
 import org.apache.cassandra.locator.ReplicaPlans;
 import org.apache.cassandra.locator.Replicas;
 import org.apache.cassandra.metrics.CASClientRequestMetrics;
+import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
 import org.apache.cassandra.metrics.DenylistMetrics;
 import org.apache.cassandra.metrics.ReadRepairMetrics;
 import org.apache.cassandra.metrics.StorageMetrics;
@@ -362,6 +363,9 @@ public class StorageProxy implements StorageProxyMBean
                 // Create the desired updates
                 PartitionUpdate updates = request.makeUpdates(current, clientState, ballot);
 
+                // Update the metrics before triggers potentially add mutations.
+                ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(updates);
+
                 long size = updates.dataSize();
                 casWriteMetrics.mutationSize.update(size);
                 writeMetricsForLevel(consistencyForPaxos).mutationSize.update(size);
@@ -1714,8 +1718,8 @@ public class StorageProxy implements StorageProxyMBean
             // we build this ONLY to perform the sufficiency check that happens on construction
             ReplicaPlans.forWrite(keyspace, cm.consistency(), tk, ReplicaPlans.writeAll);
 
-            // This host isn't a replica, so mark the request as being remote. If this host is a 
-            // replica, applyCounterMutationOnCoordinator() in the branch above will call performWrite(), and 
+            // This host isn't a replica, so mark the request as being remote. If this host is a
+            // replica, applyCounterMutationOnCoordinator() in the branch above will call performWrite(), and
             // there we'll mark a local request against the metrics.
             writeMetrics.remoteRequests.mark();
 
@@ -3207,4 +3211,16 @@ public class StorageProxy implements StorageProxyMBean
     {
         DatabaseDescriptor.setSStableReadRatePersistenceEnabled(enabled);
     }
+
+    @Override
+    public boolean getClientRequestSizeMetricsEnabled()
+    {
+        return DatabaseDescriptor.getClientRequestSizeMetricsEnabled();
+    }
+
+    @Override
+    public void setClientRequestSizeMetricsEnabled(boolean enabled)
+    {
+        DatabaseDescriptor.setClientRequestSizeMetricsEnabled(enabled);
+    }
 }
diff --git a/src/java/org/apache/cassandra/service/StorageProxyMBean.java b/src/java/org/apache/cassandra/service/StorageProxyMBean.java
index 4a3adfd5bd..4ba41d6340 100644
--- a/src/java/org/apache/cassandra/service/StorageProxyMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageProxyMBean.java
@@ -141,4 +141,7 @@ public interface StorageProxyMBean
 
     boolean getSStableReadRatePersistenceEnabled();
     void setSStableReadRatePersistenceEnabled(boolean enabled);
+
+    boolean getClientRequestSizeMetricsEnabled();
+    void setClientRequestSizeMetricsEnabled(boolean enabled);
 }
diff --git a/src/java/org/apache/cassandra/service/paxos/Paxos.java b/src/java/org/apache/cassandra/service/paxos/Paxos.java
index 5bdb75c78f..c8bf3b9579 100644
--- a/src/java/org/apache/cassandra/service/paxos/Paxos.java
+++ b/src/java/org/apache/cassandra/service/paxos/Paxos.java
@@ -44,6 +44,7 @@ import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.ReplicaLayout;
 import org.apache.cassandra.locator.ReplicaLayout.ForTokenWrite;
 import org.apache.cassandra.locator.ReplicaPlan.ForRead;
+import org.apache.cassandra.metrics.ClientRequestSizeMetrics;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -701,6 +702,9 @@ public class Paxos
                     // TODO "turn null updates into delete?" - what does this TODO even mean?
                     PartitionUpdate updates = request.makeUpdates(current, clientState, begin.ballot);
 
+                    // Update the metrics before triggers potentially add mutations.
+                    ClientRequestSizeMetrics.recordRowAndColumnCountMetrics(updates);
+
                     // Apply triggers to cas updates. A consideration here is that
                     // triggers emit Mutations, and so a given trigger implementation
                     // may generate mutations for partitions other than the one this
diff --git a/test/unit/org/apache/cassandra/metrics/ClientRequestRowAndColumnMetricsTest.java b/test/unit/org/apache/cassandra/metrics/ClientRequestRowAndColumnMetricsTest.java
new file mode 100644
index 0000000000..cb4b38897f
--- /dev/null
+++ b/test/unit/org/apache/cassandra/metrics/ClientRequestRowAndColumnMetricsTest.java
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.metrics;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.List;
+
+import com.google.common.collect.ImmutableList;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.statements.BatchStatement;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.service.StorageProxy;
+import org.apache.cassandra.service.paxos.Paxos;
+import org.apache.cassandra.transport.SimpleClient;
+import org.apache.cassandra.transport.messages.BatchMessage;
+import org.apache.cassandra.transport.messages.QueryMessage;
+
+import static org.junit.Assert.assertEquals;
+
+import static org.apache.cassandra.transport.ProtocolVersion.CURRENT;
+
+public class ClientRequestRowAndColumnMetricsTest extends CQLTester
+{
+    @BeforeClass
+    public static void setup()
+    {
+        requireNetwork();
+    }
+
+    @Before
+    public void clearMetrics()
+    {
+        ClientRequestSizeMetrics.totalRowsRead.dec(ClientRequestSizeMetrics.totalRowsRead.getCount());
+        ClientRequestSizeMetrics.totalColumnsRead.dec(ClientRequestSizeMetrics.totalColumnsRead.getCount());
+        ClientRequestSizeMetrics.totalRowsWritten.dec(ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        ClientRequestSizeMetrics.totalColumnsWritten.dec(ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+        StorageProxy.instance.setClientRequestSizeMetricsEnabled(true);
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForMultiRowPartitionSelection() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1");
+
+        assertEquals(2, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key is provided by the client in the request, so we don't consider those columns as read.
+        assertEquals(4, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithOnlyPartitionKeyInSelect() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT pk FROM %s WHERE pk = 1");
+
+        assertEquals(2, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key is provided by the client in the request, so we don't consider that column read.
+        assertEquals(0, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithOnlyClusteringKeyInSelect() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT ck FROM %s WHERE pk = 1");
+
+        assertEquals(2, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldNotRecordReadMetricsWhenDisabled() throws Throwable
+    {
+        StorageProxy.instance.setClientRequestSizeMetricsEnabled(false);
+
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1");
+
+        assertEquals(0, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        assertEquals(0, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithSingleRowSelection() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND ck = 1");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // Both the partition key and clustering key are provided by the client in the request.
+        assertEquals(1, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithSliceRestriction() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND ck > 0");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key is selected, but the restriction over the clustering key is a slice.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithINRestrictionSinglePartition() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND ck IN (0, 1)");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key and clustering key are both selected.
+        assertEquals(1, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsWithINRestrictionMultiplePartitions() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 3)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (4, 5, 6)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk IN (1, 4)");
+
+        assertEquals(2, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key is selected, but there is no clustering restriction.
+        assertEquals(4, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForMultiColumnClusteringRestriction() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck1 int, ck2 int, ck3 int, v int, PRIMARY KEY (pk, ck1, ck2, ck3))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck1, ck2, ck3, v) VALUES (1, 2, 3, 4, 6)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND ck1 = 2 AND (ck2, ck3) = (3, 4)");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The entire primary key is selected, so only one value is actually read.
+        assertEquals(1, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForClusteringSlice() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck1 int, ck2 int, ck3 int, v int, PRIMARY KEY (pk, ck1, ck2, ck3))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck1, ck2, ck3, v) VALUES (1, 2, 3, 4, 6)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND ck1 = 2 AND ck2 = 3 AND ck3 >= 4");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The last clustering key element isn't bound, so count it as being read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForTokenAndClusteringSlice() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck1 int, ck2 int, ck3 int, v int, PRIMARY KEY (pk, ck1, ck2, ck3))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck1, ck2, ck3, v) VALUES (1, 2, 3, 4, 6)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE token(pk) = token(1) AND ck1 = 2 AND ck2 = 3 AND ck3 >= 4 ALLOW FILTERING");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // Last clustering is a slice, and the partition key is restricted on token, so count them as read.
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForSingleValueRow() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(1, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldNotRecordWriteMetricsWhenDisabled() throws Throwable
+    {
+        StorageProxy.instance.setClientRequestSizeMetricsEnabled(false);
+
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+
+        assertEquals(0, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(0, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForMultiValueRow() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int, v3 int)");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, v1, v2, v3) VALUES (1, 2, 3, 4)");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForBatch() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int)");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+
+            String first = String.format("INSERT INTO %s.%s (pk, v1, v2) VALUES (1, 10, 100)", KEYSPACE, currentTable());
+            String second = String.format("INSERT INTO %s.%s (pk, v1, v2) VALUES (2, 20, 200)", KEYSPACE, currentTable());
+
+            List<List<ByteBuffer>> values = ImmutableList.of(Collections.emptyList(), Collections.emptyList());
+            BatchMessage batch = new BatchMessage(BatchStatement.Type.LOGGED, ImmutableList.of(first, second), values, QueryOptions.DEFAULT);
+            client.execute(batch);
+
+            // The metrics should reflect the batch as a single write operation with multiple rows and columns.
+            assertEquals(2, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            assertEquals(4, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForCellDeletes() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int, v3 int)");
+
+        executeNet(CURRENT, "DELETE v1, v2, v3 FROM %s WHERE pk = 1");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForCellNulls() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int, v3 int)");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, v1, v2, v3) VALUES (1, null, null, null)");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForSingleStaticInsert() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v0 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v0, v1, v2) VALUES (0, 1, 2, 3, 4)");
+
+        assertEquals(2, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForBatchedStaticInserts() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v0 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+
+            String first = String.format("INSERT INTO %s.%s (pk, ck, v0, v1, v2) VALUES (0, 1, 2, 3, 4)", KEYSPACE, currentTable());
+            String second = String.format("INSERT INTO %s.%s (pk, ck, v0, v1, v2) VALUES (0, 2, 3, 5, 6)", KEYSPACE, currentTable());
+
+            List<List<ByteBuffer>> values = ImmutableList.of(Collections.emptyList(), Collections.emptyList());
+            BatchMessage batch = new BatchMessage(BatchStatement.Type.LOGGED, ImmutableList.of(first, second), values, QueryOptions.DEFAULT);
+            client.execute(batch);
+
+            // Two normal rows and the single static row:
+            assertEquals(3, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            // Two normal columns per insert, and then one columns for the static row:
+            assertEquals(5, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForRowDelete() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v0 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "DELETE FROM %s WHERE pk = 1 AND ck = 1");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        // The columns metric should account for all regular columns.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForRangeDelete() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v0 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "DELETE FROM %s WHERE pk = 1 AND ck > 1");
+
+        // The range delete is intended to delete at least one row, but that is only a lower bound.
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        // The columns metric should account for all regular columns.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForPartitionDelete() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v0 int static, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "DELETE FROM %s WHERE pk = 1");
+
+        // A partition deletion intends to delete at least one row.
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+        // If we delete one row, we intended to delete all its regular and static columns.
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForIntraRowBatch() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+
+            String first = String.format("INSERT INTO %s.%s (pk, ck, v1, v2) VALUES (1, 2, 3, 4)", KEYSPACE, currentTable());
+            String second = String.format("DELETE FROM %s.%s WHERE pk = 1 AND ck > 1", KEYSPACE, currentTable());
+
+            List<List<ByteBuffer>> values = ImmutableList.of(Collections.emptyList(), Collections.emptyList());
+            BatchMessage batch = new BatchMessage(BatchStatement.Type.LOGGED, ImmutableList.of(first, second), values, QueryOptions.DEFAULT);
+            client.execute(batch);
+
+            // Both operations affect the same row, but writes and deletes are distinct.
+            assertEquals(2, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            assertEquals(4, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForIfNotExistsV1() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v1);
+        shouldRecordWriteMetricsForIfNotExists();
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForIfNotExistsV2() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v2);
+        shouldRecordWriteMetricsForIfNotExists();
+    }
+
+    public void shouldRecordWriteMetricsForIfNotExists() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int, v3 int)");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+            client.execute(new QueryMessage(String.format("INSERT INTO %s.%s (pk, v1, v2, v3) VALUES (1, 2, 3, 4) IF NOT EXISTS", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+            // We read internally, but don't reflect that in our read metrics.
+            assertEquals(0, ClientRequestSizeMetrics.totalRowsRead.getCount());
+            assertEquals(0, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForCASV1() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v1);
+        shouldRecordWriteMetricsForCAS();
+    }
+
+    @Test
+    public void shouldRecordWriteMetricsForCASV2() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v2);
+        shouldRecordWriteMetricsForCAS();
+    }
+
+    public void shouldRecordWriteMetricsForCAS() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int)");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+            client.execute(new QueryMessage(String.format("INSERT INTO %s.%s (pk, v1, v2) VALUES (1, 2, 3)", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            assertEquals(2, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+            client.execute(new QueryMessage(String.format("UPDATE %s.%s SET v2 = 4 WHERE pk = 1 IF v1 = 2", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            assertEquals(3, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+            // We read internally, but don't reflect that in our read metrics.
+            assertEquals(0, ClientRequestSizeMetrics.totalRowsRead.getCount());
+            assertEquals(0, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+        }
+    }
+
+    @Test
+    public void shouldNotRecordWriteMetricsForFailedCASV1() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v1);
+        shouldNotRecordWriteMetricsForFailedCAS();
+    }
+
+    @Test
+    public void shouldNotRecordWriteMetricsForFailedCASV2() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v2);
+        shouldNotRecordWriteMetricsForFailedCAS();
+    }
+
+    public void shouldNotRecordWriteMetricsForFailedCAS() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int PRIMARY KEY, v1 int, v2 int)");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+            client.execute(new QueryMessage(String.format("INSERT INTO %s.%s (pk, v1, v2) VALUES (1, 2, 3)", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            assertEquals(1, ClientRequestSizeMetrics.totalRowsWritten.getCount());
+            assertEquals(2, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+            client.execute(new QueryMessage(String.format("UPDATE %s.%s SET v2 = 4 WHERE pk = 1 IF v1 = 4", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            // We didn't actually write anything, so don't reflect a write against the metrics.
+            assertEquals(2, ClientRequestSizeMetrics.totalColumnsWritten.getCount());
+
+            // Don't reflect in our read metrics the result returned to the client by the failed CAS write. 
+            assertEquals(0, ClientRequestSizeMetrics.totalRowsRead.getCount());
+            assertEquals(0, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordReadMetricsOnSerialReadV1() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v1);
+        shouldRecordReadMetricsOnSerialRead();
+    }
+
+    @Test
+    public void shouldRecordReadMetricsOnSerialReadV2() throws Exception
+    {
+        Paxos.setPaxosVariant(Config.PaxosVariant.v2);
+        shouldRecordReadMetricsOnSerialRead();
+    }
+
+    public void shouldRecordReadMetricsOnSerialRead() throws Exception
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        try (SimpleClient client = new SimpleClient(nativeAddr.getHostAddress(), nativePort, CURRENT))
+        {
+            client.connect(false);
+            client.execute(new QueryMessage(String.format("INSERT INTO %s.%s (pk, ck, v) VALUES (1, 1, 1)", KEYSPACE, currentTable()), QueryOptions.DEFAULT));
+
+            QueryMessage query = new QueryMessage(String.format("SELECT * FROM %s.%s WHERE pk = 1 AND ck = 1", KEYSPACE, currentTable()),
+                                                  QueryOptions.forInternalCalls(ConsistencyLevel.SERIAL, Collections.emptyList()));
+            client.execute(query);
+
+            assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+
+            // Both the partition key and clustering key are provided by the client in the request.
+            assertEquals(1, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+        }
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForGlobalIndexQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+        createIndex("CREATE INDEX on %s (v)");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (2, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE v = 1");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The index search term is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForPartitionRestrictedIndexQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+        createIndex("CREATE INDEX on %s (v)");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE pk = 1 AND v = 1");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The partition key and index search term are provided by the client, so we don't consider those columns read.
+        assertEquals(1, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForClusteringKeyIndexQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+        createIndex("CREATE INDEX on %s (ck)");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE ck = 2");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The index search term is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForFilteringQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (2, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE v = 1 ALLOW FILTERING");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The filtering term is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForRangeFilteringQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (2, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE v > 1 ALLOW FILTERING");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The value column is restricted over a range, not bound to a particular value.
+        assertEquals(3, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForINFilteringQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, 1)");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (2, 2, 2)");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE v IN (1) ALLOW FILTERING");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The filtering term is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+
+    @Test
+    public void shouldRecordReadMetricsForContainsQuery() throws Throwable
+    {
+        createTable("CREATE TABLE %s (pk int, ck int, v set<int>, PRIMARY KEY (pk, ck))");
+
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (1, 1, {1, 2, 3} )");
+        executeNet(CURRENT, "INSERT INTO %s (pk, ck, v) VALUES (2, 2, {4, 5, 6})");
+        executeNet(CURRENT, "SELECT * FROM %s WHERE v CONTAINS 1 ALLOW FILTERING");
+
+        assertEquals(1, ClientRequestSizeMetrics.totalRowsRead.getCount());
+        // The filtering term is provided by the client in the request, so we don't consider that column read.
+        assertEquals(2, ClientRequestSizeMetrics.totalColumnsRead.getCount());
+    }
+}