From 4ef1b5bbf3e659ac7524b28e7f77b9c6de5b3307 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Thu, 7 May 2026 10:59:42 +0100 Subject: [PATCH 1/5] Avoid serialization and deserialization for coordinator-local single partition data read Currently, when we execute a local read we fetch data from SSTables and Memtables using a merging iterator and write it to a byte buffer. Later when we combine a CQL response we deserialize the data back to iterate over them as a part of coordinator logic. So, we allocate rows and cells twice here, during the read from SSTables/Memtables and during the deserialization by coordinator logic if we read data locally (it is a typical scenario because usually drivers are sending requests to replicas). The idea of optimization: if we do a single partition read of a small number of rows we can keep the data in memory and avoid double row objects allocation. A system property is used to limit number of rows we keep in memory in this scenario (to avoid too much pressure on GC due to extended lifetime for these objects and promoting them to an old generation). The property also allows to disable the logic in case of any issues. We cannot get a number of rows in advance, so we read first N rows to memory and if we still have something then we serialize the remaining to a buffer and then concatenate iterators for the in-memory rows + deserialized one when we need to iterate over the result patch by Dmitry Konstantinov; reviewed by TBD for CASSANDRA-21354 --- .../config/CassandraRelevantProperties.java | 1 + .../org/apache/cassandra/db/ReadCommand.java | 18 +- .../org/apache/cassandra/db/ReadResponse.java | 195 +++++++++++++++- .../cassandra/service/StorageProxy.java | 5 +- .../apache/cassandra/db/ReadResponseTest.java | 215 +++++++++++++++++- 5 files changed, 421 insertions(+), 13 deletions(-) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index c72577e7ec..ad475a1427 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -207,6 +207,7 @@ 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"), + DATA_RESPONSE_IN_MEMORY_MAX_ROWS("cassandra.data_response_in_memory_max_rows", "128"), DECAYING_ESTIMATED_HISTOGRAM_RESERVOIR_STRIPE_COUNT("cassandra.dehr_stripe_count", "2"), DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES("default.provide.overlapping.tombstones"), /** determinism properties for testing */ diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index d161fb315d..5bc4171187 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -436,14 +436,26 @@ public abstract class ReadCommand extends AbstractReadQuery public abstract boolean isReversed(); public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi) + { + return createResponse(iterator, rdi, false); + } + public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead) { // 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() - ? ReadResponse.createDigestResponse(iterator, this) - : ReadResponse.createDataResponse(iterator, this, rdi); + if (localRead && ReadResponse.IN_MEMORY_MAX_ROWS > 0) + return ReadResponse.createInMemoryDataResponse(iterator, this, rdi); + + return ReadResponse.createDataResponse(iterator, this, rdi); + } + + public ReadResponse createLocalObjectResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi) + { + return createResponse(iterator, rdi, true); } public ReadResponse createEmptyResponse() diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 7edd743ad4..4b60932fb1 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -21,16 +21,23 @@ 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.Rows; +import org.apache.cassandra.db.rows.Unfiltered; import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIteratorSerializer; +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; @@ -49,6 +56,8 @@ public abstract class ReadResponse // Serializer for single partition read response public static final IVersionedSerializer serializer = new Serializer(); + static final int IN_MEMORY_MAX_ROWS = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_ROWS.getInt(); + protected ReadResponse() { } @@ -63,9 +72,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) { - return new LocalDataResponse(data, selection); + return createInMemoryDataResponse(data, command, rdi, IN_MEMORY_MAX_ROWS); + } + + @VisibleForTesting + static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows) + { + return new InMemoryDataResponse(data, command, rdi, maxRows); } @VisibleForTesting @@ -229,6 +244,9 @@ public abstract class ReadResponse { // Exponential moving average of response sizes, used to set initial size of output buffer. private static final MovingAverage estimatedResponseBytes = ExpMovingAverage.decayBy1000(); + + // Exponential moving average of overflow(!) sizes, used to set initial size of output buffer. + private static final MovingAverage estimatedOverflowBytes = ExpMovingAverage.decayBy1000(); private static final int bufferInitialSizeMin = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MIN.getInt(); private static final int bufferInitialSizeMax = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MAX.getInt(); @@ -240,9 +258,28 @@ public abstract class ReadResponse DeserializationHelper.Flag.LOCAL); } - private LocalDataResponse(UnfilteredPartitionIterator iter, ColumnFilter selection) + private static ByteBuffer buildOverflow(UnfilteredRowIterator rowIter, ColumnFilter selection) { - super(build(iter, selection), null, false, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + int initialBufferSize = bufferInitialSizeMin; + + if (bufferInitialSizeMax > bufferInitialSizeMin) + { + double estimatedOverflowSize = estimatedOverflowBytes.get(); + double bufferSizeEstimate = Double.isNaN(estimatedOverflowSize) ? bufferInitialSizeMin : estimatedOverflowSize * 1.1; + initialBufferSize = Math.min((int) bufferSizeEstimate, bufferInitialSizeMax); + } + + try (DataOutputBuffer buffer = new DataOutputBuffer(initialBufferSize)) + { + // NOTE: we use UnfilteredRowIteratorSerializer here, not UnfilteredPartitionIterators + UnfilteredRowIteratorSerializer.serializer.serialize(rowIter, selection, buffer, MessagingService.current_version); + estimatedOverflowBytes.update(buffer.position()); + return buffer.buffer(false); + } + catch (IOException e) + { + throw new RuntimeException(e); + } } private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection) @@ -354,6 +391,152 @@ 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 + private static class InMemoryDataResponse extends DataResponse + { + // rows up to the max-rows limit stored as an in-memory partition; null if the response was empty + private final ImmutableBTreePartition partition; + // rows beyond the max-rows limit serialized to a buffer; null if all rows fit in memory + private final ByteBuffer overflow; + + private InMemoryDataResponse(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows) + { + super(null, rdi.getDigest(), rdi.isConclusive(), MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + + if (!iter.hasNext()) + { + this.partition = null; + this.overflow = null; + return; + } + + try (UnfilteredRowIterator rowIter = iter.next()) + { + LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows); + this.partition = ImmutableBTreePartition.create(limitedIter); + // Uses buildOverflow (row-iterator level) to match the UnfilteredRowIteratorSerializer + // deserializer used in makeIterator. + this.overflow = limitedIter.hasOverflow() + ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) + : null; + } + } + + // Decorating iterator that stops after maxRows Unfiltered objects and records whether overflow occurred. + // 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 int rowCount = 0; + private boolean hasOverflow = false; + + private LimitedUnfilteredRowIterator(UnfilteredRowIterator wrapped, int maxRows) + { + super(wrapped.metadata(), + wrapped.partitionKey(), + wrapped.partitionLevelDeletion(), + wrapped.columns(), + wrapped.staticRow(), + wrapped.isReverseOrder(), + wrapped.stats()); + this.wrapped = wrapped; + this.maxRows = maxRows; + } + + boolean hasOverflow() + { + return hasOverflow; + } + + @Override + protected Unfiltered computeNext() + { + if (rowCount >= maxRows) + { + hasOverflow = wrapped.hasNext(); + return endOfData(); + } + if (!wrapped.hasNext()) + return endOfData(); + rowCount++; + return wrapped.next(); + } + } + + @Override + public UnfilteredPartitionIterator makeIterator(ReadCommand command) + { + if (partition == null) + return EmptyIterators.unfilteredPartition(command.metadata()); + + UnfilteredRowIterator inMemoryIter = partition.unfilteredIterator(command.columnFilter(), Slices.ALL, command.isReversed()); + + if (overflow == null) + return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter); + + // Deserialize the overflow and concatenate with the in-memory part. + // Buffer closing is not needed and actually is incorrect + // because the result iterator uses the buffer context + DataInputBuffer in = new DataInputBuffer(overflow, true); + try + { + UnfilteredRowIterator overflowIter = UnfilteredRowIteratorSerializer.serializer.deserialize( + in, MessagingService.current_version, command.metadata(), command.columnFilter(), DeserializationHelper.Flag.LOCAL); + UnfilteredRowIterator combined = UnfilteredRowIterators.concat(inMemoryIter, overflowIter); + return new SingletonUnfilteredPartitionIterator(command.metadata(), combined); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + @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 @@ -378,7 +561,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 +601,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; diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 5c1ebc9ed5..c267123b95 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -2748,7 +2748,10 @@ 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()) + response = command.createLocalObjectResponse(iterator, controller.getRepairedDataInfo()); + else + response = command.createResponse(iterator, controller.getRepairedDataInfo()); } catch (RejectException e) { diff --git a/test/unit/org/apache/cassandra/db/ReadResponseTest.java b/test/unit/org/apache/cassandra/db/ReadResponseTest.java index ebe1cf0322..03956bfc5f 100644 --- a/test/unit/org/apache/cassandra/db/ReadResponseTest.java +++ b/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,24 @@ 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.Row; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; 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.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 +60,7 @@ public class ReadResponseTest { private final Random random = new Random(); private TableMetadata metadata; + private TableMetadata metadataWithClustering; @BeforeClass public static void beforeClass() @@ -60,13 +71,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 +190,192 @@ 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); + + 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); + + 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); + + try (DataOutputBuffer out = new DataOutputBuffer()) + { + ReadResponse.serializer.serialize(response, out, MessagingService.current_version); + } + catch (IOException e) + { + fail("Unexpected IOException: " + e.getMessage()); + } + } + + @Test + public void inMemoryResponseWithOverflowMatchesLocalDataResponse() + { + // 5 rows, only 2 fit in memory — 3 overflow into the serialized buffer + testMultipleRows(2, 5); + } + + @Test + public void inMemoryResponseAllRowsInMemoryWhenUnderLimit() + { + // 3 rows, limit is 10 — all fit in memory with no overflow + testMultipleRows(10, 3); + } + + @Test + public void inMemoryResponseWithZeroMaxRowsUsesOnlyOverflow() + { + // all rows go directly into the overflow buffer + testMultipleRows(0, 3); + } + + private void testMultipleRows(int maxRows, int rows) + { + int key = key(); + ReadCommand command = command(key, metadataWithClustering); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + + PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rows); + + ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); + ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows); + + assertIteratorsEqual(command, localResponse, inMemoryResponse); + } + + @Test + public void inMemoryResponseWithRangeTombstoneOnlyAllInMemory() + { + // Partition contains only a range tombstone (no rows); limit is large enough that nothing overflows + testWithTombstones(10, 0, 0, 5); + } + + @Test + public void inMemoryResponseWithRangeTombstoneOnlyOverflow() + { + // 5 rows at clusterings 0-4 plus a range tombstone covering [6, 9]; + // rows 0-1 go in-memory, rows 2-4 and the RT go to overflow + testWithTombstones(2, 5, 6, 9); + } + + @Test + public void inMemoryResponseWithRangeTombstoneBetweenInMemoryAndOverflow() + { + // RT at [1, 2] sits between the in-memory rows (0) and the overflow rows (3-4) + testWithTombstones(1, 5, 1, 2); + } + + private void testWithTombstones(int maxRows, int rows, int rtStart, int rtEnd) + { + int key = key(); + ReadCommand command = command(key, metadataWithClustering); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + + PartitionUpdate update = buildUpdateWithRowsAndRangeTombstone(metadataWithClustering, key, rows, rtStart, rtEnd); + + ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); + ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows); + + assertIteratorsEqual(command, localResponse, inMemoryResponse); + } + + + private PartitionUpdate buildUpdateWithRowsAndRangeTombstone(TableMetadata metadata, int partitionKey, + int rowCount, int rtStart, int rtEnd) + { + PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadata, ByteBufferUtil.bytes(partitionKey)).timestamp(0); + for (int i = 0; i < rowCount; i++) + builder.row(i).add("v", i); + builder.addRangeTombstone().start(rtStart).end(rtEnd); + return builder.build(); + } + + private void assertIteratorsEqual(ReadCommand command, ReadResponse expected, ReadResponse actual) + { + List expectedUnfiltered = collectUnfiltered(command, expected); + List actualUnfiltered = collectUnfiltered(command, actual); + assertEquals(expectedUnfiltered, actualUnfiltered); + } + + private List collectUnfiltered(ReadCommand command, ReadResponse response) + { + List 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) + { + UnfilteredRowIterator rowIter = update.unfilteredIterator(); + 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()) @@ -222,6 +426,11 @@ public class ReadResponseTest return new StubReadCommand(key, metadata, false); } + private ReadCommand command(int key) + { + return command(key, metadata); + } + private static class StubRepairedDataInfo extends RepairedDataInfo { private final ByteBuffer repairedDigest; @@ -262,11 +471,11 @@ public class ReadResponseTest RowFilter.none(), DataLimits.NONE, metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key)), - null, + new ClusteringIndexSliceFilter(Slices.ALL, false), null, false, null); - + } @Override From e32230a34dc07ecd8f20ac8f2c8b1460c71ef796 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Thu, 7 May 2026 21:14:54 +0100 Subject: [PATCH 2/5] fix RepairedDataInfo propagation --- .../org/apache/cassandra/db/ReadResponse.java | 35 ++++--- .../apache/cassandra/db/ReadResponseTest.java | 96 +++++++++++++++++++ 2 files changed, 117 insertions(+), 14 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 4b60932fb1..5d1cdeb344 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -325,8 +325,8 @@ public abstract class ReadResponse // TODO: can the digest be calculated over the raw bytes now? // The response, serialized in the current messaging version private final ByteBuffer data; - private final ByteBuffer repairedDataDigest; - private final boolean isRepairedDigestConclusive; + protected ByteBuffer repairedDataDigest; + protected boolean isRepairedDigestConclusive; private final int dataSerializationVersion; private final DeserializationHelper.Flag flag; @@ -408,25 +408,32 @@ public abstract class ReadResponse private InMemoryDataResponse(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows) { - super(null, rdi.getDigest(), rdi.isConclusive(), MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + // pass fake values for rdi.getDigest(), rdi.isConclusive(), they will be calculated and set later + super(null, null, false, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); if (!iter.hasNext()) { this.partition = null; this.overflow = null; - return; + } + else + { + try (UnfilteredRowIterator rowIter = iter.next()) + { + LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows); + this.partition = ImmutableBTreePartition.create(limitedIter); + // Uses buildOverflow (row-iterator level) to match the UnfilteredRowIteratorSerializer + // deserializer used in makeIterator. + this.overflow = limitedIter.hasOverflow() + ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) + : null; + } } - try (UnfilteredRowIterator rowIter = iter.next()) - { - LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows); - this.partition = ImmutableBTreePartition.create(limitedIter); - // Uses buildOverflow (row-iterator level) to match the UnfilteredRowIteratorSerializer - // deserializer used in makeIterator. - this.overflow = limitedIter.hasOverflow() - ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) - : null; - } + // Capture digest after consuming the iterator so that any updates made by + // RepairedDataInfo transformations are reflected in the digest. + this.repairedDataDigest = rdi.getDigest(); + this.isRepairedDigestConclusive = rdi.isConclusive(); } // Decorating iterator that stops after maxRows Unfiltered objects and records whether overflow occurred. diff --git a/test/unit/org/apache/cassandra/db/ReadResponseTest.java b/test/unit/org/apache/cassandra/db/ReadResponseTest.java index 03956bfc5f..99705312ab 100644 --- a/test/unit/org/apache/cassandra/db/ReadResponseTest.java +++ b/test/unit/org/apache/cassandra/db/ReadResponseTest.java @@ -41,6 +41,7 @@ import org.apache.cassandra.db.rows.BTreeRow; import org.apache.cassandra.db.rows.BufferCell; import org.apache.cassandra.db.rows.Row; 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; @@ -271,6 +272,26 @@ public class ReadResponseTest assertIteratorsEqual(command, localResponse, inMemoryResponse); } + @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); + + ReadResponse response = ReadResponse.createInMemoryDataResponse(lazyRdiWrappedIterator(update, rdi), command, rdi, 10); + + assertTrue("digest should be non-empty after iterator was consumed", rdi.wasConsumedBeforeDigestCaptured()); + assertEquals(expectedDigest, response.repairedDataDigest()); + assertTrue(response.isRepairedDigestConclusive()); + } + @Test public void inMemoryResponseWithRangeTombstoneOnlyAllInMemory() { @@ -456,6 +477,81 @@ 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) From 04740b6f46f74b34f6e0d714ae3916a44a06d925 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Wed, 13 May 2026 21:38:05 +0100 Subject: [PATCH 3/5] adress simple review comments --- .../org/apache/cassandra/db/ReadCommand.java | 1 + .../org/apache/cassandra/db/ReadResponse.java | 35 +++++++++++-------- 2 files changed, 22 insertions(+), 14 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 5bc4171187..f332a7ad14 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -439,6 +439,7 @@ public abstract class ReadCommand extends AbstractReadQuery { return createResponse(iterator, rdi, false); } + public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead) { // validate that the sequence of RT markers is correct: open is followed by close, deletion times for both diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 5d1cdeb344..50bfa0f2fb 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -80,7 +80,7 @@ public abstract class ReadResponse @VisibleForTesting static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows) { - return new InMemoryDataResponse(data, command, rdi, maxRows); + return InMemoryDataResponse.build(data, command, rdi, maxRows); } @VisibleForTesting @@ -325,8 +325,8 @@ public abstract class ReadResponse // TODO: can the digest be calculated over the raw bytes now? // The response, serialized in the current messaging version private final ByteBuffer data; - protected ByteBuffer repairedDataDigest; - protected boolean isRepairedDigestConclusive; + private final ByteBuffer repairedDataDigest; + private final boolean isRepairedDigestConclusive; private final int dataSerializationVersion; private final DeserializationHelper.Flag flag; @@ -406,34 +406,41 @@ public abstract class ReadResponse // rows beyond the max-rows limit serialized to a buffer; null if all rows fit in memory private final ByteBuffer overflow; - private InMemoryDataResponse(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows) + static InMemoryDataResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows) { - // pass fake values for rdi.getDigest(), rdi.isConclusive(), they will be calculated and set later - super(null, null, false, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + ImmutableBTreePartition partition; + ByteBuffer overflow; if (!iter.hasNext()) { - this.partition = null; - this.overflow = null; + partition = null; + overflow = null; } else { try (UnfilteredRowIterator rowIter = iter.next()) { LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows); - this.partition = ImmutableBTreePartition.create(limitedIter); + partition = ImmutableBTreePartition.create(limitedIter); // Uses buildOverflow (row-iterator level) to match the UnfilteredRowIteratorSerializer // deserializer used in makeIterator. - this.overflow = limitedIter.hasOverflow() - ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) - : null; + overflow = limitedIter.hasOverflow() + ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) + : null; } } // Capture digest after consuming the iterator so that any updates made by // RepairedDataInfo transformations are reflected in the digest. - this.repairedDataDigest = rdi.getDigest(); - this.isRepairedDigestConclusive = rdi.isConclusive(); + return new InMemoryDataResponse(partition, overflow, rdi.getDigest(), rdi.isConclusive()); + } + + private InMemoryDataResponse(ImmutableBTreePartition partition, ByteBuffer overflow, + ByteBuffer repairedDataDigest, boolean isRepairedDigestConclusive) + { + super(null, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + this.partition = partition; + this.overflow = overflow; } // Decorating iterator that stops after maxRows Unfiltered objects and records whether overflow occurred. From 39931bbdacd27afc0c0ee2c76209642caad28c97 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Thu, 14 May 2026 17:13:41 +0100 Subject: [PATCH 4/5] address review comments --- .../org/apache/cassandra/db/ReadResponse.java | 46 +++- .../apache/cassandra/db/ReadResponseTest.java | 209 +++++++++++++----- 2 files changed, 195 insertions(+), 60 deletions(-) diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 50bfa0f2fb..5a57f715f9 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -33,6 +33,7 @@ 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.Rows; import org.apache.cassandra.db.rows.Unfiltered; import org.apache.cassandra.db.rows.UnfilteredRowIterator; @@ -62,6 +63,12 @@ public abstract class ReadResponse { } + @VisibleForTesting + int inMemoryUnfilteredCount() + { + return 0; + } + public static ReadResponse createDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi) { return new LocalDataResponse(data, command, rdi); @@ -443,14 +450,29 @@ public abstract class ReadResponse this.overflow = overflow; } + @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 after maxRows Unfiltered objects and records whether overflow occurred. // 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 int rowCount = 0; + private int unfilteredCount = 0; private boolean hasOverflow = false; + private boolean insideOpenMarker = false; private LimitedUnfilteredRowIterator(UnfilteredRowIterator wrapped, int maxRows) { @@ -473,15 +495,23 @@ public abstract class ReadResponse @Override protected Unfiltered computeNext() { - if (rowCount >= maxRows) - { - hasOverflow = wrapped.hasNext(); - return endOfData(); - } if (!wrapped.hasNext()) return endOfData(); - rowCount++; - return wrapped.next(); + + if (unfilteredCount >= maxRows && !insideOpenMarker) + { + hasOverflow = true; + return endOfData(); + } + + Unfiltered next = wrapped.next(); + if (next.isRangeTombstoneMarker()) + { + RangeTombstoneMarker marker = (RangeTombstoneMarker) next; + insideOpenMarker = marker.isOpen(isReverseOrder); + } + unfilteredCount++; + return next; } } diff --git a/test/unit/org/apache/cassandra/db/ReadResponseTest.java b/test/unit/org/apache/cassandra/db/ReadResponseTest.java index 99705312ab..5f39e0d1ce 100644 --- a/test/unit/org/apache/cassandra/db/ReadResponseTest.java +++ b/test/unit/org/apache/cassandra/db/ReadResponseTest.java @@ -240,36 +240,28 @@ public class ReadResponseTest @Test public void inMemoryResponseWithOverflowMatchesLocalDataResponse() { - // 5 rows, only 2 fit in memory — 3 overflow into the serialized buffer - testMultipleRows(2, 5); + inMemoryAssertion() + .maxRows(2).rows(5) + .expectInMemoryUnfilteredCount(2) + .verify(); } @Test public void inMemoryResponseAllRowsInMemoryWhenUnderLimit() { - // 3 rows, limit is 10 — all fit in memory with no overflow - testMultipleRows(10, 3); + inMemoryAssertion() + .maxRows(10).rows(3) + .expectInMemoryUnfilteredCount(3) + .verify(); } @Test public void inMemoryResponseWithZeroMaxRowsUsesOnlyOverflow() { - // all rows go directly into the overflow buffer - testMultipleRows(0, 3); - } - - private void testMultipleRows(int maxRows, int rows) - { - int key = key(); - ReadCommand command = command(key, metadataWithClustering); - StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); - - PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rows); - - ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); - ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows); - - assertIteratorsEqual(command, localResponse, inMemoryResponse); + inMemoryAssertion() + .maxRows(0).rows(3) + .expectInMemoryUnfilteredCount(0) + .verify(); } @Test @@ -295,48 +287,156 @@ public class ReadResponseTest @Test public void inMemoryResponseWithRangeTombstoneOnlyAllInMemory() { - // Partition contains only a range tombstone (no rows); limit is large enough that nothing overflows - testWithTombstones(10, 0, 0, 5); + inMemoryAssertion() + .maxRows(10).rows(0) + .withTombstone(rangeTombstone(0, 5)) + .expectInMemoryUnfilteredCount(2) + .verify(); } @Test public void inMemoryResponseWithRangeTombstoneOnlyOverflow() { - // 5 rows at clusterings 0-4 plus a range tombstone covering [6, 9]; // rows 0-1 go in-memory, rows 2-4 and the RT go to overflow - testWithTombstones(2, 5, 6, 9); + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(6, 9)) + .expectInMemoryUnfilteredCount(2) + .verify(); + } + + @Test + public void inMemoryResponseWithTombstonesOnlyOverflow() + { + // Each RT produces an open+close marker pair (2 unfiltered objects). + // maxRows=2 fits exactly the first RT; the second overflows. + inMemoryAssertion() + .maxRows(2).rows(0) + .withTombstone(rangeTombstone(0, 3)) + .withTombstone(rangeTombstone(5, 8)) + .expectInMemoryUnfilteredCount(2) + .verify(); } @Test public void inMemoryResponseWithRangeTombstoneBetweenInMemoryAndOverflow() { - // RT at [1, 2] sits between the in-memory rows (0) and the overflow rows (3-4) - testWithTombstones(1, 5, 1, 2); + // RT at [1, 2) sits between the in-memory rows (0) and the overflow rows (3-4) + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(1, 2)) + .expectInMemoryUnfilteredCount(3) + .verify(); } - private void testWithTombstones(int maxRows, int rows, int rtStart, int rtEnd) + @Test + public void inMemoryResponseWithRangeTombstoneBetweenInMemoryAndOverflowReversed() { - int key = key(); - ReadCommand command = command(key, metadataWithClustering); - StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); - - PartitionUpdate update = buildUpdateWithRowsAndRangeTombstone(metadataWithClustering, key, rows, rtStart, rtEnd); - - ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); - ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows); - - assertIteratorsEqual(command, localResponse, inMemoryResponse); + // RT at [2, 3) sits between the in-memory rows (4) and the overflow rows (0-1) + inMemoryAssertion() + .maxRows(2).rows(5).reversed() + .withTombstone(rangeTombstone(2, 3)) + .expectInMemoryUnfilteredCount(3) + .verify(); } - - private PartitionUpdate buildUpdateWithRowsAndRangeTombstone(TableMetadata metadata, int partitionKey, - int rowCount, int rtStart, int rtEnd) + @Test + public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestamps() { - PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadata, ByteBufferUtil.bytes(partitionKey)).timestamp(0); - for (int i = 0; i < rowCount; i++) - builder.row(i).add("v", i); - builder.addRangeTombstone().start(rtStart).end(rtEnd); - return builder.build(); + // Three overlapping RTs: [0,3) ts=100, [1,4) ts=200, [2,5) ts=50 + // which are transformed into: bound, boundary, boundary, bound markers + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(0, 3).timestamp(100)) + .withTombstone(rangeTombstone(1, 4).timestamp(200)) + .withTombstone(rangeTombstone(2, 5).timestamp(50)) + .expectInMemoryUnfilteredCount(4) + .verify(); + } + + @Test + public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestampsReversed() + { + // Same tombstones as above with reversed read order + inMemoryAssertion() + .maxRows(2).rows(5).reversed() + .withTombstone(rangeTombstone(0, 3).timestamp(100)) + .withTombstone(rangeTombstone(1, 4).timestamp(200)) + .withTombstone(rangeTombstone(2, 5).timestamp(50)) + .expectInMemoryUnfilteredCount(4) + .verify(); + } + + private InMemoryAssertionBuilder inMemoryAssertion() + { + return new InMemoryAssertionBuilder(); + } + + private class InMemoryAssertionBuilder + { + private int maxRows = Integer.MAX_VALUE; + private int rows = 0; + private boolean reversed = false; + private final List tombstones = new ArrayList<>(); + private Integer expectInMemoryUnfilteredCount = null; + + InMemoryAssertionBuilder maxRows(int maxRows) { this.maxRows = maxRows; 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; } + InMemoryAssertionBuilder expectInMemoryUnfilteredCount(int count) { this.expectInMemoryUnfilteredCount = count; 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(); + + ReadResponse localResponse = command.createResponse(singlePartitionIterator(update, reversed), rdi); + ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update, reversed), command, rdi, maxRows); + + assertIteratorsEqual(command, localResponse, inMemoryResponse); + if (expectInMemoryUnfilteredCount != null) + assertEquals("Unxpected inMemoryUnfilteredCount", (int) expectInMemoryUnfilteredCount, inMemoryResponse.inMemoryUnfilteredCount()); + } + } + + 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) @@ -380,7 +480,11 @@ public class ReadResponseTest private UnfilteredPartitionIterator singlePartitionIterator(PartitionUpdate update) { - UnfilteredRowIterator rowIter = update.unfilteredIterator(); + 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; @@ -439,17 +543,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) + private ReadCommand command(int key, TableMetadata metadata, boolean reversed) { - return command(key, metadata); + return new StubReadCommand(key, metadata, false, reversed); + } private static class StubRepairedDataInfo extends RepairedDataInfo @@ -554,7 +659,7 @@ public class ReadResponseTest 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, @@ -567,7 +672,7 @@ public class ReadResponseTest RowFilter.none(), DataLimits.NONE, metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key)), - new ClusteringIndexSliceFilter(Slices.ALL, false), + new ClusteringIndexSliceFilter(Slices.ALL, reversed), null, false, null); From 6791726dab8556c76db9c30ffd7884050017696a Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Mon, 13 Jul 2026 18:56:28 +0100 Subject: [PATCH 5/5] address review comments: - add size limit - add separate limits for ONE/LOCAL_ONE - keep only in-memory or serialized version, without mix to simplify code also add limit crossing counter metrics to improve observability --- .../config/CassandraRelevantProperties.java | 4 + .../org/apache/cassandra/db/ReadCommand.java | 12 +- .../org/apache/cassandra/db/ReadResponse.java | 185 +++++++----- .../metrics/CassandraMetricsRegistry.java | 1 + .../metrics/ReadResponseMetrics.java | 42 +++ .../cassandra/service/StorageProxy.java | 8 +- .../cassandra/service/reads/ReadCallback.java | 6 + .../apache/cassandra/db/ReadResponseTest.java | 264 +++++++++++++++--- 8 files changed, 402 insertions(+), 120 deletions(-) create mode 100644 src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index ad475a1427..60dfd27d95 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -207,7 +207,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 */ diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index f332a7ad14..ad8edb5d08 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -437,10 +437,10 @@ public abstract class ReadCommand extends AbstractReadQuery public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi) { - return createResponse(iterator, rdi, false); + return createResponse(iterator, rdi, false, false); } - public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead) + 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. @@ -448,15 +448,15 @@ public abstract class ReadCommand extends AbstractReadQuery if (isDigestQuery()) return ReadResponse.createDigestResponse(iterator, this); - if (localRead && ReadResponse.IN_MEMORY_MAX_ROWS > 0) - return ReadResponse.createInMemoryDataResponse(iterator, this, rdi); + if (localRead && ReadResponse.inMemoryLocalResponseEnabled(localReplicaOnly)) + return ReadResponse.createInMemoryDataResponse(iterator, this, rdi, localReplicaOnly); return ReadResponse.createDataResponse(iterator, this, rdi); } - public ReadResponse createLocalObjectResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi) + public ReadResponse createLocalObjectResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localReplicaOnly) { - return createResponse(iterator, rdi, true); + return createResponse(iterator, rdi, true, localReplicaOnly); } public ReadResponse createEmptyResponse() diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 5a57f715f9..b35917412c 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -34,16 +34,17 @@ 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.UnfilteredRowIteratorSerializer; 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; @@ -57,12 +58,37 @@ public abstract class ReadResponse // Serializer for single partition read response public static final IVersionedSerializer 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() { @@ -79,15 +105,15 @@ public abstract class ReadResponse return new LocalDataResponse(data, command, NO_OP_REPAIRED_DATA_INFO); } - static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi) + static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, boolean oneOrLocalOne) { - return createInMemoryDataResponse(data, command, rdi, IN_MEMORY_MAX_ROWS); + return createInMemoryDataResponse(data, command, rdi, maxRows(oneOrLocalOne), maxSize(oneOrLocalOne)); } @VisibleForTesting - static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows) + static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize) { - return InMemoryDataResponse.build(data, command, rdi, maxRows); + return InMemoryDataResponse.build(data, command, rdi, maxRows, maxHeapSize); } @VisibleForTesting @@ -251,9 +277,6 @@ public abstract class ReadResponse { // Exponential moving average of response sizes, used to set initial size of output buffer. private static final MovingAverage estimatedResponseBytes = ExpMovingAverage.decayBy1000(); - - // Exponential moving average of overflow(!) sizes, used to set initial size of output buffer. - private static final MovingAverage estimatedOverflowBytes = ExpMovingAverage.decayBy1000(); private static final int bufferInitialSizeMin = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MIN.getInt(); private static final int bufferInitialSizeMax = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MAX.getInt(); @@ -265,28 +288,11 @@ public abstract class ReadResponse DeserializationHelper.Flag.LOCAL); } - private static ByteBuffer buildOverflow(UnfilteredRowIterator rowIter, 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) { - int initialBufferSize = bufferInitialSizeMin; - - if (bufferInitialSizeMax > bufferInitialSizeMin) - { - double estimatedOverflowSize = estimatedOverflowBytes.get(); - double bufferSizeEstimate = Double.isNaN(estimatedOverflowSize) ? bufferInitialSizeMin : estimatedOverflowSize * 1.1; - initialBufferSize = Math.min((int) bufferSizeEstimate, bufferInitialSizeMax); - } - - try (DataOutputBuffer buffer = new DataOutputBuffer(initialBufferSize)) - { - // NOTE: we use UnfilteredRowIteratorSerializer here, not UnfilteredPartitionIterators - UnfilteredRowIteratorSerializer.serializer.serialize(rowIter, selection, buffer, MessagingService.current_version); - estimatedOverflowBytes.update(buffer.position()); - return buffer.buffer(false); - } - catch (IOException e) - { - throw new RuntimeException(e); - } + super(data, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); } private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection) @@ -405,49 +411,63 @@ public abstract class ReadResponse } } - // built on the coordinator for local single-partition reads; never sent over the network + // 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 { - // rows up to the max-rows limit stored as an in-memory partition; null if the response was empty + // the whole response as an in-memory partition; null if the response was empty private final ImmutableBTreePartition partition; - // rows beyond the max-rows limit serialized to a buffer; null if all rows fit in memory - private final ByteBuffer overflow; - static InMemoryDataResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows) + static ReadResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize) { - ImmutableBTreePartition partition; - ByteBuffer overflow; - if (!iter.hasNext()) { - partition = null; - overflow = null; + // Empty response + return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive()); } - else + + ImmutableBTreePartition partition; + ByteBuffer serialized; + try (UnfilteredRowIterator rowIter = iter.next()) { - try (UnfilteredRowIterator rowIter = iter.next()) + LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows, maxHeapSize); + partition = ImmutableBTreePartition.create(limitedIter); + + if (limitedIter.hasOverflow()) { - LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows); - partition = ImmutableBTreePartition.create(limitedIter); - // Uses buildOverflow (row-iterator level) to match the UnfilteredRowIteratorSerializer - // deserializer used in makeIterator. - overflow = limitedIter.hasOverflow() - ? LocalDataResponse.buildOverflow(rowIter, command.columnFilter()) - : null; + // 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 the iterator so that any updates made by - // RepairedDataInfo transformations are reflected in the digest. - return new InMemoryDataResponse(partition, overflow, rdi.getDigest(), rdi.isConclusive()); + // 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()); } - private InMemoryDataResponse(ImmutableBTreePartition partition, ByteBuffer overflow, + // 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; - this.overflow = overflow; } @Override @@ -464,17 +484,21 @@ public abstract class ReadResponse return count; } - // Decorating iterator that stops after maxRows Unfiltered objects and records whether overflow occurred. + // 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) + private LimitedUnfilteredRowIterator(UnfilteredRowIterator wrapped, int maxRows, long maxHeapSize) { super(wrapped.metadata(), wrapped.partitionKey(), @@ -485,6 +509,7 @@ public abstract class ReadResponse wrapped.stats()); this.wrapped = wrapped; this.maxRows = maxRows; + this.maxHeapSize = maxHeapSize; } boolean hasOverflow() @@ -492,16 +517,23 @@ public abstract class ReadResponse return hasOverflow; } + boolean overflowedByRowLimit() + { + return overflowByRowLimit; + } + @Override protected Unfiltered computeNext() { if (!wrapped.hasNext()) return endOfData(); - if (unfilteredCount >= maxRows && !insideOpenMarker) + if (!insideOpenMarker) { - hasOverflow = true; - return endOfData(); + if (maxRows > 0 && unfilteredCount >= maxRows) + return overflow(true); + if (maxHeapSize > 0 && accumulatedHeapSize >= maxHeapSize) + return overflow(false); } Unfiltered next = wrapped.next(); @@ -511,8 +543,25 @@ public abstract class ReadResponse 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 @@ -522,25 +571,7 @@ public abstract class ReadResponse return EmptyIterators.unfilteredPartition(command.metadata()); UnfilteredRowIterator inMemoryIter = partition.unfilteredIterator(command.columnFilter(), Slices.ALL, command.isReversed()); - - if (overflow == null) - return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter); - - // Deserialize the overflow and concatenate with the in-memory part. - // Buffer closing is not needed and actually is incorrect - // because the result iterator uses the buffer context - DataInputBuffer in = new DataInputBuffer(overflow, true); - try - { - UnfilteredRowIterator overflowIter = UnfilteredRowIteratorSerializer.serializer.deserialize( - in, MessagingService.current_version, command.metadata(), command.columnFilter(), DeserializationHelper.Flag.LOCAL); - UnfilteredRowIterator combined = UnfilteredRowIterators.concat(inMemoryIter, overflowIter); - return new SingletonUnfilteredPartitionIterator(command.metadata(), combined); - } - catch (IOException e) - { - throw new RuntimeException(e); - } + return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter); } @Override diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 7c69c77d88..312f449f13 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/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) diff --git a/src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java b/src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java new file mode 100644 index 0000000000..a138bd5871 --- /dev/null +++ b/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() {} +} diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index c267123b95..bc9a308a4a 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -128,6 +128,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.ForwardingInfo; import org.apache.cassandra.net.Message; @@ -310,6 +311,7 @@ public class StorageProxy implements StorageProxyMBean ReadRepairMetrics.init(); + ReadResponseMetrics.init(); if (!Paxos.isLinearizable()) { @@ -2749,7 +2751,11 @@ public class StorageProxy implements StorageProxyMBean UnfilteredPartitionIterator iterator = command.executeLocally(controller)) { if (command.isLimitedToOnePartition() && !command.isDigestQuery()) - response = command.createLocalObjectResponse(iterator, controller.getRepairedDataInfo()); + { + 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()); } diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java index 7f68b3e4d5..10b19d23a3 100644 --- a/src/java/org/apache/cassandra/service/reads/ReadCallback.java +++ b/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, 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))); diff --git a/test/unit/org/apache/cassandra/db/ReadResponseTest.java b/test/unit/org/apache/cassandra/db/ReadResponseTest.java index 5f39e0d1ce..e148d8a49b 100644 --- a/test/unit/org/apache/cassandra/db/ReadResponseTest.java +++ b/test/unit/org/apache/cassandra/db/ReadResponseTest.java @@ -39,13 +39,16 @@ 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; @@ -198,7 +201,7 @@ public class ReadResponseTest 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); + ReadResponse inMemoryResponse = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false); assertIteratorsEqual(command, localResponse, inMemoryResponse); } @@ -215,7 +218,7 @@ public class ReadResponseTest PartitionUpdate update = PartitionUpdate.singleRowUpdate(metadata, dk, row); ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); - ReadResponse inMemoryResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi); + ReadResponse inMemoryResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, false); assertIteratorsEqual(command, localResponse, inMemoryResponse); } @@ -225,7 +228,7 @@ public class ReadResponseTest { ReadCommand command = command(key(), metadata); StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); - ReadResponse response = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi); + ReadResponse response = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false); try (DataOutputBuffer out = new DataOutputBuffer()) { @@ -237,12 +240,111 @@ public class ReadResponseTest } } + @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) - .expectInMemoryUnfilteredCount(2) + .expectSerialized() .verify(); } @@ -251,16 +353,57 @@ public class ReadResponseTest { inMemoryAssertion() .maxRows(10).rows(3) - .expectInMemoryUnfilteredCount(3) + .expectInMemory() .verify(); } @Test - public void inMemoryResponseWithZeroMaxRowsUsesOnlyOverflow() + public void inMemoryResponseByteLimitWithOverflow() { + // Row limit disabled; the byte limit is crossed, so the whole response is serialized. inMemoryAssertion() - .maxRows(0).rows(3) - .expectInMemoryUnfilteredCount(0) + .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(); } @@ -277,7 +420,8 @@ public class ReadResponseTest // Returns EMPTY before any iteration; returns expectedDigest once the iterator has been consumed. LazyRepairedDataInfo rdi = new LazyRepairedDataInfo(expectedDigest); - ReadResponse response = ReadResponse.createInMemoryDataResponse(lazyRdiWrappedIterator(update, rdi), command, rdi, 10); + // 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()); @@ -285,85 +429,86 @@ public class ReadResponseTest } @Test - public void inMemoryResponseWithRangeTombstoneOnlyAllInMemory() + 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)) - .expectInMemoryUnfilteredCount(2) + .expectInMemory() .verify(); } @Test - public void inMemoryResponseWithRangeTombstoneOnlyOverflow() + public void inMemoryResponseWithRangeTombstoneOverflow() { - // rows 0-1 go in-memory, rows 2-4 and the RT go to overflow + // Rows plus a trailing range tombstone cross the row limit: the whole response is serialized. inMemoryAssertion() .maxRows(2).rows(5) .withTombstone(rangeTombstone(6, 9)) - .expectInMemoryUnfilteredCount(2) + .expectSerialized() .verify(); } @Test public void inMemoryResponseWithTombstonesOnlyOverflow() { - // Each RT produces an open+close marker pair (2 unfiltered objects). - // maxRows=2 fits exactly the first RT; the second overflows. + // 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)) - .expectInMemoryUnfilteredCount(2) + .expectSerialized() .verify(); } @Test - public void inMemoryResponseWithRangeTombstoneBetweenInMemoryAndOverflow() + public void inMemoryResponseWithRangeTombstoneAmongRowsOverflow() { - // RT at [1, 2) sits between the in-memory rows (0) and the overflow rows (3-4) + // 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)) - .expectInMemoryUnfilteredCount(3) + .expectSerialized() .verify(); } @Test - public void inMemoryResponseWithRangeTombstoneBetweenInMemoryAndOverflowReversed() + public void inMemoryResponseWithRangeTombstoneAmongRowsOverflowReversed() { - // RT at [2, 3) sits between the in-memory rows (4) and the overflow rows (0-1) + // Same as above with reversed read order; the range tombstone is at [2, 3). inMemoryAssertion() .maxRows(2).rows(5).reversed() .withTombstone(rangeTombstone(2, 3)) - .expectInMemoryUnfilteredCount(3) + .expectSerialized() .verify(); } @Test public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestamps() { - // Three overlapping RTs: [0,3) ts=100, [1,4) ts=200, [2,5) ts=50 - // which are transformed into: bound, boundary, boundary, bound markers + // 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)) - .expectInMemoryUnfilteredCount(4) + .expectInMemory() .verify(); } @Test public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestampsReversed() { - // Same tombstones as above with reversed read order + // 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)) - .expectInMemoryUnfilteredCount(4) + .expectInMemory() .verify(); } @@ -374,17 +519,24 @@ public class ReadResponseTest private class InMemoryAssertionBuilder { - private int maxRows = Integer.MAX_VALUE; + 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 tombstones = new ArrayList<>(); - private Integer expectInMemoryUnfilteredCount = null; + // 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; } - InMemoryAssertionBuilder expectInMemoryUnfilteredCount(int count) { this.expectInMemoryUnfilteredCount = count; 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() { @@ -406,15 +558,55 @@ public class ReadResponseTest } PartitionUpdate update = builder.build(); - ReadResponse localResponse = command.createResponse(singlePartitionIterator(update, reversed), rdi); - ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update, reversed), command, rdi, maxRows); + long maxHeapSize = maxBytes == null ? 0 : heapSizeOfFirstNRows(update, reversed, maxBytes); - assertIteratorsEqual(command, localResponse, inMemoryResponse); - if (expectInMemoryUnfilteredCount != null) - assertEquals("Unxpected inMemoryUnfilteredCount", (int) expectInMemoryUnfilteredCount, inMemoryResponse.inMemoryUnfilteredCount()); + 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 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);