Merge 6791726dab into 10557d7ffe

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -212,6 +212,11 @@ public enum CassandraRelevantProperties
     DATA_OUTPUT_STREAM_PLUS_TEMP_BUFFER_SIZE("cassandra.data_output_stream_plus_temp_buffer_size", "8192"),
     DATA_RESPONSE_BUFFER_INITIAL_SIZE_MAX("cassandra.data_response_buffer_initial_size_max", "4096"),
     DATA_RESPONSE_BUFFER_INITIAL_SIZE_MIN("cassandra.data_response_buffer_initial_size_min", "128"),
+    /** higher in-memory limits used for ONE/LOCAL_ONE reads, when local result is consumed directly without waiting for other replicas **/
+    DATA_RESPONSE_IN_MEMORY_MAX_ROWS("cassandra.data_response_in_memory_max_rows", "128"),
+    DATA_RESPONSE_IN_MEMORY_MAX_ROWS_CL_ONE("cassandra.data_response_in_memory_max_rows_cl_one", "512"),
+    DATA_RESPONSE_IN_MEMORY_MAX_SIZE("cassandra.data_response_in_memory_max_size", "16KiB"),
+    DATA_RESPONSE_IN_MEMORY_MAX_SIZE_CL_ONE("cassandra.data_response_in_memory_max_size_cl_one", "64KiB"),
     DECAYING_ESTIMATED_HISTOGRAM_RESERVOIR_STRIPE_COUNT("cassandra.dehr_stripe_count", "2"),
     DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES("default.provide.overlapping.tombstones"),
     /** determinism properties for testing */

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

@@ -443,14 +443,27 @@ public abstract class ReadCommand extends AbstractReadQuery
     public abstract boolean isReversed();
 
     public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi)
+    {
+        return createResponse(iterator, rdi, false, false);
+    }
+
+    private ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead, boolean localReplicaOnly)
     {
         // validate that the sequence of RT markers is correct: open is followed by close, deletion times for both
         // ends equal, and there are no dangling RT bound in any partition.
         iterator = RTBoundValidator.validate(iterator, Stage.PROCESSED, true);
+        if (isDigestQuery())
+            return ReadResponse.createDigestResponse(iterator, this);
 
-        return isDigestQuery()
+        if (localRead && ReadResponse.inMemoryLocalResponseEnabled(localReplicaOnly))
-               ? ReadResponse.createDigestResponse(iterator, this)
+            return ReadResponse.createInMemoryDataResponse(iterator, this, rdi, localReplicaOnly);
-               : ReadResponse.createDataResponse(iterator, this, rdi);
+
+        return ReadResponse.createDataResponse(iterator, this, rdi);
+    }
+
+    public ReadResponse createLocalObjectResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localReplicaOnly)
+    {
+        return createResponse(iterator, rdi, true, localReplicaOnly);
     }
 
     public ReadResponse createEmptyResponse()

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

@@ -21,21 +21,30 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.NoSuchElementException;
 
 import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.db.rows.AbstractUnfilteredRowIterator;
 import org.apache.cassandra.db.rows.DeserializationHelper;
+import org.apache.cassandra.db.rows.RangeTombstoneMarker;
+import org.apache.cassandra.db.rows.Row;
 import org.apache.cassandra.db.rows.Rows;
+import org.apache.cassandra.db.rows.Unfiltered;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.db.rows.UnfilteredRowIterators;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.metrics.ReadResponseMetrics;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -49,10 +58,43 @@ public abstract class ReadResponse
     // Serializer for single partition read response
     public static final IVersionedSerializer<ReadResponse> serializer = new Serializer();
 
+    // Per-request limits for the in-memory portion of a local single-partition read response. Unfiltered objects are
+    // kept as an in-memory partition until either enabled limit is reached; the remainder is serialized into an
+    // overflow buffer. Each limit is disabled independently by setting it to <= 0.
+    //   - IN_MEMORY_MAX_ROWS: maximum number of in-memory Unfiltered objects (rows and range tombstone markers).
+    //   - IN_MEMORY_MAX_SIZE: maximum accumulated unshared heap size (in bytes) of the in-memory Unfiltered objects.
+    // ONE/LOCAL_ONE reads use separate higher limits: the local read is consumed directly
+    // without waiting for cross-replica interactions, so it is worth keeping more of it in memory.
+    static final int IN_MEMORY_MAX_ROWS = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_ROWS.getInt();
+    static final long IN_MEMORY_MAX_SIZE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_SIZE.getSizeInBytes();
+    static final int IN_MEMORY_MAX_ROWS_CL_ONE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_ROWS_CL_ONE.getInt();
+    static final long IN_MEMORY_MAX_SIZE_CL_ONE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_SIZE_CL_ONE.getSizeInBytes();
+
     protected ReadResponse()
     {
     }
 
+    private static int maxRows(boolean localReplicaOnly)
+    {
+        return localReplicaOnly ? IN_MEMORY_MAX_ROWS_CL_ONE : IN_MEMORY_MAX_ROWS;
+    }
+
+    private static long maxSize(boolean localReplicaOnly)
+    {
+        return localReplicaOnly ? IN_MEMORY_MAX_SIZE_CL_ONE : IN_MEMORY_MAX_SIZE;
+    }
+
+    static boolean inMemoryLocalResponseEnabled(boolean localReplicaOnly)
+    {
+        return maxRows(localReplicaOnly) > 0 || maxSize(localReplicaOnly) > 0;
+    }
+
+    @VisibleForTesting
+    int inMemoryUnfilteredCount()
+    {
+        return 0;
+    }
+
     public static ReadResponse createDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi)
     {
         return new LocalDataResponse(data, command, rdi);
@@ -63,9 +105,15 @@ public abstract class ReadResponse
         return new LocalDataResponse(data, command, NO_OP_REPAIRED_DATA_INFO);
     }
 
-    public static ReadResponse createSimpleDataResponse(UnfilteredPartitionIterator data, ColumnFilter selection)
+    static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, boolean oneOrLocalOne)
     {
-        return new LocalDataResponse(data, selection);
+        return createInMemoryDataResponse(data, command, rdi, maxRows(oneOrLocalOne), maxSize(oneOrLocalOne));
+    }
+
+    @VisibleForTesting
+    static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize)
+    {
+        return InMemoryDataResponse.build(data, command, rdi, maxRows, maxHeapSize);
     }
 
     @VisibleForTesting
@@ -240,9 +288,11 @@ public abstract class ReadResponse
                   DeserializationHelper.Flag.LOCAL);
         }
 
-        private LocalDataResponse(UnfilteredPartitionIterator iter, ColumnFilter selection)
+        // Wraps an already-serialized response buffer; used when a local in-memory response outgrows its limits and
+        // is serialized fully (see InMemoryDataResponse.build).
+        private LocalDataResponse(ByteBuffer data, ByteBuffer repairedDataDigest, boolean isRepairedDigestConclusive)
         {
-            super(build(iter, selection), null, false, MessagingService.current_version, DeserializationHelper.Flag.LOCAL);
+            super(data, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL);
         }
 
         private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection)
@@ -354,6 +404,214 @@ public abstract class ReadResponse
         {
             return false;
         }
+
+        protected ByteBuffer getSerializedData()
+        {
+            return data;
+        }
+    }
+
+    // built on the coordinator for local single-partition reads; never sent over the network.
+    // Holds the whole response as an in-memory object graph. Only used when the response fits within the per-request
+    // limits; larger responses are serialized in full and returned as an ordinary LocalDataResponse instead
+    private static class InMemoryDataResponse extends DataResponse
+    {
+        // the whole response as an in-memory partition; null if the response was empty
+        private final ImmutableBTreePartition partition;
+
+        static ReadResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize)
+        {
+            if (!iter.hasNext())
+            {
+                // Empty response
+                return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+            }
+
+            ImmutableBTreePartition partition;
+            ByteBuffer serialized;
+            try (UnfilteredRowIterator rowIter = iter.next())
+            {
+                LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows, maxHeapSize);
+                partition = ImmutableBTreePartition.create(limitedIter);
+
+                if (limitedIter.hasOverflow())
+                {
+                    // if a per-request limit is crossed then fall back to the ordinary serialized representation.
+                    (limitedIter.overflowedByRowLimit() ? ReadResponseMetrics.inMemoryRowLimitHits
+                                                        : ReadResponseMetrics.inMemorySizeLimitHits).inc();
+                    serialized = serialize(command, partition, rowIter);
+                }
+                else
+                {
+                    serialized = null;
+                }
+            }
+
+            // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+            return serialized != null
+                   ? new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive())
+                   : new InMemoryDataResponse(partition, rdi.getDigest(), rdi.isConclusive());
+        }
+
+        // Serializes the full response (the already-consumed in-memory prefix followed by the remaining rows) into a
+        // buffer in the intra-node partition format, so it can be read back like any other serialized DataResponse.
+        private static ByteBuffer serialize(ReadCommand command, ImmutableBTreePartition prefix, UnfilteredRowIterator suffix)
+        {
+            UnfilteredRowIterator prefixIter = prefix.unfilteredIterator(command.columnFilter(), Slices.ALL, suffix.isReverseOrder());
+            UnfilteredRowIterator combined = UnfilteredRowIterators.concat(prefixIter, suffix);
+            UnfilteredPartitionIterator partitionIter = new SingletonUnfilteredPartitionIterator(command.metadata(), combined);
+            return LocalDataResponse.build(partitionIter, command.columnFilter());
+        }
+
+        private InMemoryDataResponse(ImmutableBTreePartition partition,
+                                     ByteBuffer repairedDataDigest, boolean isRepairedDigestConclusive)
+        {
+            super(null, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL);
+            this.partition = partition;
+        }
+
+        @Override
+        @VisibleForTesting
+        int inMemoryUnfilteredCount()
+        {
+            if (partition == null)
+                return 0;
+            int count = 0;
+            try (UnfilteredRowIterator iter = partition.unfilteredIterator())
+            {
+                while (iter.hasNext()) { iter.next(); count++; }
+            }
+            return count;
+        }
+
+        // Decorating iterator that stops once a per-request limit is reached and records whether overflow occurred and which limit caused it.
+        // Does NOT close the wrapped iterator on close() so the caller can continue reading overflow rows.
+        private static class LimitedUnfilteredRowIterator extends AbstractUnfilteredRowIterator
+        {
+            private final UnfilteredRowIterator wrapped;
+            private final int maxRows;
+            private final long maxHeapSize;
+
+            private int unfilteredCount = 0;
+            private long accumulatedHeapSize = 0;
+            private boolean hasOverflow = false;
+            private boolean overflowByRowLimit = false;
+            private boolean insideOpenMarker = false;
+
+            private LimitedUnfilteredRowIterator(UnfilteredRowIterator wrapped, int maxRows, long maxHeapSize)
+            {
+                super(wrapped.metadata(),
+                      wrapped.partitionKey(),
+                      wrapped.partitionLevelDeletion(),
+                      wrapped.columns(),
+                      wrapped.staticRow(),
+                      wrapped.isReverseOrder(),
+                      wrapped.stats());
+                this.wrapped = wrapped;
+                this.maxRows = maxRows;
+                this.maxHeapSize = maxHeapSize;
+            }
+
+            boolean hasOverflow()
+            {
+                return hasOverflow;
+            }
+
+            boolean overflowedByRowLimit()
+            {
+                return overflowByRowLimit;
+            }
+
+            @Override
+            protected Unfiltered computeNext()
+            {
+                if (!wrapped.hasNext())
+                    return endOfData();
+
+                if (!insideOpenMarker)
+                {
+                    if (maxRows > 0 && unfilteredCount >= maxRows)
+                        return overflow(true);
+                    if (maxHeapSize > 0 && accumulatedHeapSize >= maxHeapSize)
+                        return overflow(false);
+                }
+
+                Unfiltered next = wrapped.next();
+                if (next.isRangeTombstoneMarker())
+                {
+                    RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+                    insideOpenMarker = marker.isOpen(isReverseOrder);
+                }
+                unfilteredCount++;
+                // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+                if (maxHeapSize > 0)
+                    accumulatedHeapSize += unsharedHeapSize(next);
+                return next;
+            }
+
+            private Unfiltered overflow(boolean byRowLimit)
+            {
+                hasOverflow = true;
+                overflowByRowLimit = byRowLimit;
+                return endOfData();
+            }
+
+            private static long unsharedHeapSize(Unfiltered unfiltered)
+            {
+                return unfiltered.isRow()
+                       ? ((Row) unfiltered).unsharedHeapSize()
+                       : ((RangeTombstoneMarker) unfiltered).unsharedHeapSize();
+            }
+        }
+
+        @Override
+        public UnfilteredPartitionIterator makeIterator(ReadCommand command)
+        {
+            if (partition == null)
+                return EmptyIterators.unfilteredPartition(command.metadata());
+
+            UnfilteredRowIterator inMemoryIter = partition.unfilteredIterator(command.columnFilter(), Slices.ALL, command.isReversed());
+            return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter);
+        }
+
+        @Override
+        protected ByteBuffer getSerializedData()
+        {
+            throw new UnsupportedOperationException("InMemoryDataResponse cannot be serialized over the network");
+        }
+    }
+
+    private static class SingletonUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator
+    {
+        private final TableMetadata metadata;
+        private final UnfilteredRowIterator partition;
+        private boolean returned = false;
+
+        private SingletonUnfilteredPartitionIterator(TableMetadata metadata, UnfilteredRowIterator partition)
+        {
+            this.metadata = metadata;
+            this.partition = partition;
+        }
+
+        public TableMetadata metadata() { return metadata; }
+
+        public boolean hasNext()
+        {
+            return !returned;
+        }
+
+        public UnfilteredRowIterator next()
+        {
+            if (returned) throw new NoSuchElementException();
+            returned = true;
+            return partition;
+        }
+
+        @Override
+        public void close()
+        {
+            partition.close();
+        }
     }
 
     private static class Serializer implements IVersionedSerializer<ReadResponse>
@@ -378,7 +636,7 @@ public abstract class ReadResponse
                 ByteBufferUtil.writeWithVIntLength(response.repairedDataDigest(), out);
                 out.writeBoolean(response.isRepairedDigestConclusive());
 
-                ByteBuffer data = ((DataResponse)response).data;
+                ByteBuffer data = ((DataResponse)response).getSerializedData();
                 ByteBufferUtil.writeWithVIntLength(data, out);
             }
         }
@@ -418,7 +676,7 @@ public abstract class ReadResponse
                 // In theory, we should deserialize/re-serialize if the version asked is different from the current
                 // version as the content could have a different serialization format. So far though, we haven't made
                 // change to partition iterators serialization since 3.0 so we skip this.
-                ByteBuffer data = ((DataResponse)response).data;
+                ByteBuffer data = ((DataResponse)response).getSerializedData();
                 size += ByteBufferUtil.serializedSizeWithVIntLength(data);
             }
             return size;

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

@@ -146,6 +146,7 @@ public class CassandraMetricsRegistry extends MetricRegistry
                                    .add(MutualTlsMetrics.TYPE_NAME)
                                    .add(PaxosMetrics.TYPE_NAME)
                                    .add(ReadRepairMetrics.TYPE_NAME)
+                                   .add(ReadResponseMetrics.TYPE_NAME)
                                    .add(RepairMetrics.TYPE_NAME)
                                    .add(RowIndexEntry.TYPE_NAME)
                                    .add(StorageMetrics.TYPE_NAME)

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

@@ -0,0 +1,42 @@
+/*
+ * 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 com.codahale.metrics.Counter;
+
+import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics;
+
+public class ReadResponseMetrics
+{
+    public static final String TYPE_NAME = "ReadResponse";
+    private static final MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME);
+
+    /**
+     * Incremented when a local single-partition read response hit the per-request in-memory row-count limit
+     */
+    public static final Counter inMemoryRowLimitHits = Metrics.counter(factory.createMetricName("InMemoryRowLimitHits"));
+
+    /**
+     * Incremented when a local single-partition read response hit the per-request in-memory heap-size limit.
+     * The row-count limit is checked first, so this
+     * counts only responses that passed the row-count limit (or have it disabled) but crossed the size limit.
+     */
+    public static final Counter inMemorySizeLimitHits = Metrics.counter(factory.createMetricName("InMemorySizeLimitHits"));
+
+    public static void init() {}
+}

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -127,6 +127,7 @@ 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.ReadResponseMetrics;
 import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.net.ArtificialLatency;
 import org.apache.cassandra.net.ForwardingInfo;
@@ -311,6 +312,7 @@ public class StorageProxy implements StorageProxyMBean
 
 
         ReadRepairMetrics.init();
+        ReadResponseMetrics.init();
 
         if (!Paxos.isLinearizable())
         {
@@ -2753,7 +2755,14 @@ public class StorageProxy implements StorageProxyMBean
                 try (ReadExecutionController controller = command.executionController(trackRepairedStatus);
                      UnfilteredPartitionIterator iterator = command.executeLocally(controller))
                 {
-                    response = command.createResponse(iterator, controller.getRepairedDataInfo());
+                    if (command.isLimitedToOnePartition() && !command.isDigestQuery())
+                    {
+                        ConsistencyLevel cl = handler.consistencyLevel();
+                        boolean localReplicaOnly = cl == ConsistencyLevel.ONE || cl == ConsistencyLevel.LOCAL_ONE;
+                        response = command.createLocalObjectResponse(iterator, controller.getRepairedDataInfo(), localReplicaOnly);
+                    }
+                    else
+                        response = command.createResponse(iterator, controller.getRepairedDataInfo());
                 }
                 catch (RejectException e)
                 {

src/java/org/apache/cassandra/service/reads/ReadCallback.java

@@ -30,6 +30,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.MessageParams;
 import org.apache.cassandra.db.PartitionRangeReadCommand;
 import org.apache.cassandra.db.ReadCommand;
@@ -101,6 +102,11 @@ public class ReadCallback<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
         return replicaPlan.get();
     }
 
+    public ConsistencyLevel consistencyLevel()
+    {
+        return replicaPlan().consistencyLevel();
+    }
+
     public boolean await(long commandTimeout, TimeUnit unit)
     {
         return awaitUntil(requestTime.computeDeadline(unit.toNanos(commandTimeout)));

test/unit/org/apache/cassandra/db/ReadResponseTest.java

@@ -20,6 +20,8 @@ package org.apache.cassandra.db;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Random;
 
 import org.junit.Before;
@@ -27,16 +29,28 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.RowFilter;
 import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
+import org.apache.cassandra.db.rows.BTreeRow;
+import org.apache.cassandra.db.rows.BufferCell;
+import org.apache.cassandra.db.rows.RangeTombstoneMarker;
+import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.db.rows.Unfiltered;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
+import org.apache.cassandra.db.rows.WrappingUnfilteredRowIterator;
 import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.metrics.ReadResponseMetrics;
 import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -50,6 +64,7 @@ public class ReadResponseTest
 {
     private final Random random = new Random();
     private TableMetadata metadata;
+    private TableMetadata metadataWithClustering;
 
     @BeforeClass
     public static void beforeClass()
@@ -60,13 +75,20 @@ public class ReadResponseTest
     @Before
     public void setup()
     {
-
         metadata = TableMetadata.builder("ks", "t1")
                                 .offline()
                                 .addPartitionKeyColumn("p", Int32Type.instance)
                                 .addRegularColumn("v", Int32Type.instance)
                                 .partitioner(Murmur3Partitioner.instance)
                                 .build();
+
+        metadataWithClustering = TableMetadata.builder("ks", "t2")
+                                              .offline()
+                                              .addPartitionKeyColumn("p", Int32Type.instance)
+                                              .addClusteringColumn("c", Int32Type.instance)
+                                              .addRegularColumn("v", Int32Type.instance)
+                                              .partitioner(Murmur3Partitioner.instance)
+                                              .build();
     }
 
     @Test
@@ -172,6 +194,505 @@ public class ReadResponseTest
         assertEquals(response1.digest(command1), response2.digest(command2));
     }
 
+    @Test
+    public void inMemoryResponseEmptyIteratorMatchesLocalDataResponse()
+    {
+        ReadCommand command = command(key(), metadata);
+        StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+
+        ReadResponse localResponse = command.createResponse(EmptyIterators.unfilteredPartition(metadata), rdi);
+        ReadResponse inMemoryResponse = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false);
+
+        assertIteratorsEqual(command, localResponse, inMemoryResponse);
+    }
+
+    @Test
+    public void inMemoryResponseWithRowsMatchesLocalDataResponse()
+    {
+        int key = key();
+        ReadCommand command = command(key, metadata);
+        StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+
+        DecoratedKey dk = metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key));
+        Row row = buildRow(metadata, dk);
+        PartitionUpdate update = PartitionUpdate.singleRowUpdate(metadata, dk, row);
+
+        ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi);
+        ReadResponse inMemoryResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, false);
+
+        assertIteratorsEqual(command, localResponse, inMemoryResponse);
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void inMemoryResponseCannotBeSerialized()
+    {
+        ReadCommand command = command(key(), metadata);
+        StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+        ReadResponse response = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false);
+
+        try (DataOutputBuffer out = new DataOutputBuffer())
+        {
+            ReadResponse.serializer.serialize(response, out, MessagingService.current_version);
+        }
+        catch (IOException e)
+        {
+            fail("Unexpected IOException: " + e.getMessage());
+        }
+    }
+
+    @Test
+    public void inMemoryResponseUsesHigherLimitsForOneConsistency()
+    {
+        int key = key();
+        ReadCommand command = command(key, metadataWithClustering);
+        StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+        // More rows than the default row limit (128) but fewer than the ONE/LOCAL_ONE limit (512).
+        int rowCount = 150;
+        PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rowCount);
+
+        ReadResponse otherResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, false);
+        ReadResponse oneResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, true);
+
+        // The default-tier limits overflow this partition; ONE/LOCAL_ONE keeps more of it in memory.
+        assertTrue("default tier should overflow", otherResponse.inMemoryUnfilteredCount() < rowCount);
+        assertTrue("ONE/LOCAL_ONE tier should keep more rows in memory",
+                   oneResponse.inMemoryUnfilteredCount() > otherResponse.inMemoryUnfilteredCount());
+    }
+
+    @Test
+    public void inMemoryResponseCountsRowLimitHitOnOverflow()
+    {
+        // Row limit 1, size limit disabled: the rows beyond the first overflow, hitting the row-count limit.
+        limitHitAssertion()
+            .rows(5).maxRows(1)
+            .expectRowLimitHit()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseCountsSizeLimitHitOnOverflow()
+    {
+        // Row limit disabled, tiny size limit: the rows beyond the first overflow, hitting the size limit.
+        limitHitAssertion()
+            .rows(5).maxSize(1)
+            .expectSizeLimitHit()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseCountsRowLimitFirstWhenBothLimitsCrossed()
+    {
+        // Both limits would be crossed; the row-count limit is checked first, so only it is counted.
+        limitHitAssertion()
+            .rows(5).maxRows(1).maxSize(1)
+            .expectRowLimitHit()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseDoesNotCountLimitHitWhenAllInMemory()
+    {
+        // Limits above the response size: nothing overflows.
+        limitHitAssertion()
+            .rows(3).maxRows(100).maxSize(Long.MAX_VALUE)
+            .expectNoLimitHit()
+            .verify();
+    }
+
+    private LimitHitAssertionBuilder limitHitAssertion()
+    {
+        return new LimitHitAssertionBuilder();
+    }
+
+    /**
+     * Builds an in-memory response for a partition of {@link #rows} rows with the configured per-request limits
+     * ({@code 0} disables a limit) and asserts which limit-hit counter(s) were incremented (each by one).
+     */
+    private class LimitHitAssertionBuilder
+    {
+        private int rows = 0;
+        private int maxRows = 0;   // 0 = row-count limit disabled
+        private long maxSize = 0;  // 0 = heap-size limit disabled
+        private boolean expectRowHit = false;
+        private boolean expectSizeHit = false;
+
+        LimitHitAssertionBuilder rows(int rows) { this.rows = rows; return this; }
+        LimitHitAssertionBuilder maxRows(int maxRows) { this.maxRows = maxRows; return this; }
+        LimitHitAssertionBuilder maxSize(long maxSize) { this.maxSize = maxSize; return this; }
+        LimitHitAssertionBuilder expectRowLimitHit() { this.expectRowHit = true; return this; }
+        LimitHitAssertionBuilder expectSizeLimitHit() { this.expectSizeHit = true; return this; }
+        LimitHitAssertionBuilder expectNoLimitHit() { this.expectRowHit = false; this.expectSizeHit = false; return this; }
+
+        void verify()
+        {
+            int key = key();
+            ReadCommand command = command(key, metadataWithClustering);
+            StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+            PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rows);
+
+            long rowHitsBefore = ReadResponseMetrics.inMemoryRowLimitHits.getCount();
+            long sizeHitsBefore = ReadResponseMetrics.inMemorySizeLimitHits.getCount();
+            ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows, maxSize);
+            assertEquals("unexpected row limit hit count", rowHitsBefore + (expectRowHit ? 1 : 0), ReadResponseMetrics.inMemoryRowLimitHits.getCount());
+            assertEquals("unexpected size limit hit count", sizeHitsBefore + (expectSizeHit ? 1 : 0), ReadResponseMetrics.inMemorySizeLimitHits.getCount());
+        }
+    }
+
+    @Test
+    public void inMemoryResponseWithOverflowMatchesLocalDataResponse()
+    {
+        // Row limit crossed: the whole response is serialized.
+        inMemoryAssertion()
+            .maxRows(2).rows(5)
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseAllRowsInMemoryWhenUnderLimit()
+    {
+        inMemoryAssertion()
+            .maxRows(10).rows(3)
+            .expectInMemory()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseByteLimitWithOverflow()
+    {
+        // Row limit disabled; the byte limit is crossed, so the whole response is serialized.
+        inMemoryAssertion()
+            .maxBytesAsSizeOfFirstNRows(2).rows(5)
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseByteLimitAllRowsInMemoryWhenUnderLimit()
+    {
+        // Byte limit sized above the whole response: it stays in memory.
+        inMemoryAssertion()
+            .maxBytesAsSizeOfFirstNRows(10).rows(3)
+            .expectInMemory()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseByteLimitReachedBeforeRowLimit()
+    {
+        // Row limit (10) is not reached; the byte limit (sized for 2 rows) is crossed first, so the response is serialized.
+        inMemoryAssertion()
+            .maxRows(10).maxBytesAsSizeOfFirstNRows(2).rows(5)
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseRowLimitReachedBeforeByteLimit()
+    {
+        // Byte limit (sized for all 5 rows) is not reached; the row limit (2) is crossed first, so the response is serialized.
+        inMemoryAssertion()
+            .maxRows(2).maxBytesAsSizeOfFirstNRows(5).rows(5)
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseKeepsEverythingWhenBothLimitsDisabled()
+    {
+        // Both limits disabled: the whole partition is kept in memory.
+        inMemoryAssertion()
+            .maxRows(0).rows(4)
+            .expectInMemory()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseCapturesRepairedDigestAfterIteratorIsConsumed()
+    {
+        // RepairedDataInfo updates its digest lazily as rows are consumed via withRepairedDataInfo
+        // transformations; this test uses a stub that simulates the same timing.
+        int key = key();
+        ReadCommand command = command(key, metadataWithClustering);
+        PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, 3);
+
+        ByteBuffer expectedDigest = digest();
+        // Returns EMPTY before any iteration; returns expectedDigest once the iterator has been consumed.
+        LazyRepairedDataInfo rdi = new LazyRepairedDataInfo(expectedDigest);
+
+        // large heap budget (row limit disabled) so all rows are kept in memory; this test only cares about digest capture timing
+        ReadResponse response = ReadResponse.createInMemoryDataResponse(lazyRdiWrappedIterator(update, rdi), command, rdi, 0, Long.MAX_VALUE);
+
+        assertTrue("digest should be non-empty after iterator was consumed", rdi.wasConsumedBeforeDigestCaptured());
+        assertEquals(expectedDigest, response.repairedDataDigest());
+        assertTrue(response.isRepairedDigestConclusive());
+    }
+
+    @Test
+    public void inMemoryResponseWithRangeTombstoneAllInMemory()
+    {
+        // A single range tombstone (open+close markers) well under the limit: kept in memory.
+        inMemoryAssertion()
+            .maxRows(10).rows(0)
+            .withTombstone(rangeTombstone(0, 5))
+            .expectInMemory()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithRangeTombstoneOverflow()
+    {
+        // Rows plus a trailing range tombstone cross the row limit: the whole response is serialized.
+        inMemoryAssertion()
+            .maxRows(2).rows(5)
+            .withTombstone(rangeTombstone(6, 9))
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithTombstonesOnlyOverflow()
+    {
+        // Two range tombstones (open+close markers each) cross the row limit: the whole response is serialized.
+        inMemoryAssertion()
+            .maxRows(2).rows(0)
+            .withTombstone(rangeTombstone(0, 3))
+            .withTombstone(rangeTombstone(5, 8))
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithRangeTombstoneAmongRowsOverflow()
+    {
+        // A range tombstone at [1, 2) interleaved with rows 0-4, crossing the row limit: serialized in full.
+        inMemoryAssertion()
+            .maxRows(2).rows(5)
+            .withTombstone(rangeTombstone(1, 2))
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithRangeTombstoneAmongRowsOverflowReversed()
+    {
+        // Same as above with reversed read order; the range tombstone is at [2, 3).
+        inMemoryAssertion()
+            .maxRows(2).rows(5).reversed()
+            .withTombstone(rangeTombstone(2, 3))
+            .expectSerialized()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestamps()
+    {
+        // Three overlapping RTs: [0,3) ts=100, [1,4) ts=200, [2,5) ts=50 form a single continuous open range
+        // (bound, boundary, boundary, bound markers). Emission cannot stop while inside the open marker, so even
+        // with a row limit of 2 the whole response is kept in memory.
+        inMemoryAssertion()
+            .maxRows(2).rows(5)
+            .withTombstone(rangeTombstone(0, 3).timestamp(100))
+            .withTombstone(rangeTombstone(1, 4).timestamp(200))
+            .withTombstone(rangeTombstone(2, 5).timestamp(50))
+            .expectInMemory()
+            .verify();
+    }
+
+    @Test
+    public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestampsReversed()
+    {
+        // Same tombstones as above with reversed read order; still kept in memory (open marker cannot be split).
+        inMemoryAssertion()
+            .maxRows(2).rows(5).reversed()
+            .withTombstone(rangeTombstone(0, 3).timestamp(100))
+            .withTombstone(rangeTombstone(1, 4).timestamp(200))
+            .withTombstone(rangeTombstone(2, 5).timestamp(50))
+            .expectInMemory()
+            .verify();
+    }
+
+    private InMemoryAssertionBuilder inMemoryAssertion()
+    {
+        return new InMemoryAssertionBuilder();
+    }
+
+    private class InMemoryAssertionBuilder
+    {
+        private int maxRows = 0;                 // per-request row limit; 0 = disabled
+        private Integer maxBytes = null; // if set, the byte limit is sized to hold this many leading Unfiltered objects; null = disabled
+        private int rows = 0;
+        private boolean reversed = false;
+        private final List<RangeTombstoneSpec> tombstones = new ArrayList<>();
+        // true = the whole response is expected to be kept as an object graph, false = serialized in full
+        private Boolean expectInMemory = null;
+
+        InMemoryAssertionBuilder maxRows(int maxRows) { this.maxRows = maxRows; return this; }
+        // Enables the byte limit, sizing it to exactly hold the first n Unfiltered objects the read produces
+        InMemoryAssertionBuilder maxBytesAsSizeOfFirstNRows(int n) {this.maxBytes = n; return this; }
+        InMemoryAssertionBuilder rows(int rows) { this.rows = rows; return this; }
+        InMemoryAssertionBuilder reversed() { this.reversed = true; return this; }
+        InMemoryAssertionBuilder withTombstone(RangeTombstoneSpec rt) { tombstones.add(rt); return this; }
+        // The response stays within the limits and is kept in memory as an object graph.
+        InMemoryAssertionBuilder expectInMemory() { this.expectInMemory = true; return this; }
+        // A limit is crossed, so the whole response is serialized into a buffer.
+        InMemoryAssertionBuilder expectSerialized() { this.expectInMemory = false; return this; }
+
+        void verify()
+        {
+            int partitionKey = key();
+            ReadCommand command = command(partitionKey, metadataWithClustering, reversed);
+            StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true);
+
+            PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadataWithClustering, ByteBufferUtil.bytes(partitionKey)).timestamp(0);
+            for (int i = 0; i < rows; i++)
+                builder.row(i).add("v", i);
+            for (RangeTombstoneSpec rt : tombstones)
+            {
+                Slice slice = Slice.make(Clustering.make(ByteBufferUtil.bytes(rt.start)),
+                                         Clustering.make(ByteBufferUtil.bytes(rt.end)));
+                builder.addRangeTombstone(new RangeTombstone(slice,
+                                                             DeletionTime.build(rt.markedForDeleteAt, FBUtilities.nowInSeconds())
+                                          )
+                );
+            }
+            PartitionUpdate update = builder.build();
+
+            long maxHeapSize = maxBytes == null ? 0 : heapSizeOfFirstNRows(update, reversed, maxBytes);
+
+            ReadResponse localResponse = command.createResponse(singlePartitionIterator(update, reversed), rdi);
+            ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update, reversed), command, rdi, maxRows, maxHeapSize);
+
+            // The reconstructed response must match the plain serialized response regardless of representation.
+            List<String> expected = collectUnfiltered(command, localResponse);
+            assertEquals(expected, collectUnfiltered(command, inMemoryResponse));
+
+            // When kept in memory the whole response is an object graph (inMemoryUnfilteredCount == all unfiltereds);
+            // when serialized nothing is kept in memory (inMemoryUnfilteredCount == 0).
+            if (expectInMemory != null)
+                assertEquals("unexpected inMemoryUnfilteredCount",
+                             expectInMemory ? expected.size() : 0,
+                             inMemoryResponse.inMemoryUnfilteredCount());
+        }
+    }
+
+    /**
+     * Returns a heap size of the first {@code n} Unfiltered objects the read of
+     * {@code update} produces. Feeding this to the in-memory response keeps those {@code n} objects in memory (the
+     * (n+1)th crosses the >= threshold and overflows), which lets the assertions above express the byte-based split
+     * as a count of leading Unfiltered objects even though the limit itself is a heap size.
+     */
+    private long heapSizeOfFirstNRows(PartitionUpdate update, boolean reversed, int n)
+    {
+        if (n <= 0)
+            return 0;
+
+        long size = 0;
+        int count = 0;
+        try (UnfilteredRowIterator rowIter = update.unfilteredIterator(ColumnFilter.SelectionColumnFilter.all(update.columns()), Slices.ALL, reversed))
+        {
+            while (rowIter.hasNext() && count < n)
+            {
+                size += unsharedHeapSize(rowIter.next());
+                count++;
+            }
+        }
+        return size;
+    }
+
+    private static long unsharedHeapSize(Unfiltered unfiltered)
+    {
+        return unfiltered.isRow()
+               ? ((Row) unfiltered).unsharedHeapSize()
+               : ((RangeTombstoneMarker) unfiltered).unsharedHeapSize();
+    }
+
+    private static RangeTombstoneSpec rangeTombstone(int start, int end)
+    {
+        return new RangeTombstoneSpec(start, end);
+    }
+
+    private static class RangeTombstoneSpec
+    {
+        final int start;
+        final int end;
+        long markedForDeleteAt;
+
+        RangeTombstoneSpec(int start, int end)
+        {
+            this.start = start;
+            this.end = end;
+        }
+
+        RangeTombstoneSpec timestamp(long ts)
+        {
+            this.markedForDeleteAt = ts;
+            return this;
+        }
+    }
+
+    private void assertIteratorsEqual(ReadCommand command, ReadResponse expected, ReadResponse actual)
+    {
+        List<String> expectedUnfiltered = collectUnfiltered(command, expected);
+        List<String> actualUnfiltered = collectUnfiltered(command, actual);
+        assertEquals(expectedUnfiltered, actualUnfiltered);
+    }
+
+    private List<String> collectUnfiltered(ReadCommand command, ReadResponse response)
+    {
+        List<String> result = new ArrayList<>();
+        try (UnfilteredPartitionIterator iter = response.makeIterator(command))
+        {
+            while (iter.hasNext())
+            {
+                try (UnfilteredRowIterator partition = iter.next())
+                {
+                    while (partition.hasNext())
+                        result.add(partition.next().toString(partition.metadata(), true));
+                }
+            }
+        }
+        return result;
+    }
+
+    private Row buildRow(TableMetadata metadata, DecoratedKey key)
+    {
+        ColumnMetadata col = metadata.getColumn(ByteBufferUtil.bytes("v"));
+        Clustering<?> clustering = Clustering.EMPTY;
+        return BTreeRow.singleCellRow(clustering, BufferCell.live(col, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(42)));
+    }
+
+    private PartitionUpdate buildMultiRowUpdate(TableMetadata metadata, int partitionKey, int rowCount)
+    {
+        PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadata, ByteBufferUtil.bytes(partitionKey)).timestamp(0);
+        for (int i = 0; i < rowCount; i++)
+            builder.row(i).add("v", i);
+        return builder.build();
+    }
+
+    private UnfilteredPartitionIterator singlePartitionIterator(PartitionUpdate update)
+    {
+        return singlePartitionIterator(update, false);
+    }
+    private UnfilteredPartitionIterator singlePartitionIterator(PartitionUpdate update, boolean reversed)
+    {
+        UnfilteredRowIterator rowIter = update.unfilteredIterator(ColumnFilter.SelectionColumnFilter.all(update.columns()), Slices.ALL, reversed);
+        return new AbstractUnfilteredPartitionIterator()
+        {
+            private boolean returned = false;
+
+            public TableMetadata metadata() { return update.metadata(); }
+
+            public boolean hasNext() { return !returned; }
+
+            public UnfilteredRowIterator next()
+            {
+                returned = true;
+                return rowIter;
+            }
+        };
+    }
+
     private void verifySerDe(ReadResponse response) {
         // check that roundtripping through ReadResponse.serializer behaves as expected
         for (MessagingService.Version version : MessagingService.Version.supportedVersions())
@@ -214,12 +735,18 @@ public class ReadResponseTest
 
     private ReadCommand digestCommand(int key, TableMetadata metadata)
     {
-        return new StubReadCommand(key, metadata, true);
+        return new StubReadCommand(key, metadata, true, false);
     }
 
     private ReadCommand command(int key, TableMetadata metadata)
     {
-        return new StubReadCommand(key, metadata, false);
+        return command(key, metadata, false);
+    }
+
+    private ReadCommand command(int key, TableMetadata metadata, boolean reversed)
+    {
+        return new StubReadCommand(key, metadata, false, reversed);
+
     }
 
     private static class StubRepairedDataInfo extends RepairedDataInfo
@@ -247,9 +774,84 @@ public class ReadResponseTest
         }
     }
 
+    /**
+     * Simulates a RepairedDataInfo that updates its digest lazily as rows are consumed.
+     * Returns EMPTY_BYTE_BUFFER until the partition iterator wrapping it is fully consumed,
+     * at which point getDigest() returns the provided expected digest.
+     * This lets us verify that InMemoryDataResponse captures the digest *after* consumption.
+     */
+    private static class LazyRepairedDataInfo extends RepairedDataInfo
+    {
+        private final ByteBuffer finalDigest;
+        private boolean iteratorConsumed = false;
+        private boolean digestCapturedAfterConsumption = false;
+
+        LazyRepairedDataInfo(ByteBuffer finalDigest)
+        {
+            super(null);
+            this.finalDigest = finalDigest;
+        }
+
+        void markConsumed()
+        {
+            iteratorConsumed = true;
+        }
+
+        boolean wasConsumedBeforeDigestCaptured()
+        {
+            return digestCapturedAfterConsumption;
+        }
+
+        @Override
+        public ByteBuffer getDigest()
+        {
+            if (iteratorConsumed)
+                digestCapturedAfterConsumption = true;
+            return iteratorConsumed ? finalDigest : ByteBufferUtil.EMPTY_BYTE_BUFFER;
+        }
+
+        @Override
+        public boolean isConclusive()
+        {
+            return true;
+        }
+    }
+
+    private UnfilteredPartitionIterator lazyRdiWrappedIterator(PartitionUpdate update, LazyRepairedDataInfo rdi)
+    {
+        // Wraps the row iterator so that close() marks rdi as consumed, simulating RepairedDataInfo
+        // updating its digest lazily after the partition has been fully read.
+        UnfilteredRowIterator baseRowIter = update.unfilteredIterator();
+        UnfilteredRowIterator wrappedRowIter = new WrappingUnfilteredRowIterator()
+        {
+            public UnfilteredRowIterator wrapped() { return baseRowIter; }
+
+            @Override
+            public void close()
+            {
+                rdi.markConsumed();
+                baseRowIter.close();
+            }
+        };
+        return new AbstractUnfilteredPartitionIterator()
+        {
+            private boolean returned = false;
+
+            public TableMetadata metadata() { return update.metadata(); }
+
+            public boolean hasNext() { return !returned; }
+
+            public UnfilteredRowIterator next()
+            {
+                returned = true;
+                return wrappedRowIter;
+            }
+        };
+    }
+
     private static class StubReadCommand extends SinglePartitionReadCommand
     {
-        StubReadCommand(int key, TableMetadata metadata, boolean isDigest)
+        StubReadCommand(int key, TableMetadata metadata, boolean isDigest, boolean reversed)
         {
             super(metadata.epoch,
                   isDigest,
@@ -262,11 +864,11 @@ public class ReadResponseTest
                   RowFilter.none(),
                   DataLimits.NONE,
                   metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key)),
-                  null,
+                  new ClusteringIndexSliceFilter(Slices.ALL, reversed),
                   null,
                   false,
                   null);
-           
+
         }
 
         @Override