You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "netudima (via GitHub)" <gi...@apache.org> on 2026/05/07 15:44:09 UTC
[PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
netudima opened a new pull request, #4789:
URL: https://github.com/apache/cassandra/pull/4789
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r4020545029
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +411,317 @@ public boolean isDigestResponse()
{
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
+ {
+ // An empty static row which is intentionally not the Rows.EMPTY_STATIC_ROW singleton
+ private static final Row PRESENT_EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
+
+ // 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ if (!iter.hasNext())
+ {
+ // Empty response
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+ }
+
+ ImmutableBTreePartition partition;
+ ByteBuffer serialized = null;
+ // Closing rowIter is our job, LimitedUnfilteredRowIterator deliberately never closes what it wraps.
+ // The only exception is the overflow path, where serialize() takes it over, hence no try-with-resources.
+ UnfilteredRowIterator rowIter = iter.next();
+ try
+ {
+ LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, inMemoryMaxRows, inMemoryMaxHeapSize);
+ 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);
+ rowIter = null;
+ }
+ }
+ finally
+ {
+ if (rowIter != null)
+ rowIter.close();
+ }
+
+ // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+ if (serialized != null)
+ return new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive());
+
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return 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 = unfilteredIteratorAsRead(prefix, command.columnFilter(), 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 inMemoryMaxRows;
+ private final long inMemoryMaxHeapSize;
+
+ 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ super(wrapped.metadata(),
+ wrapped.partitionKey(),
+ wrapped.partitionLevelDeletion(),
+ wrapped.columns(),
+ wrapped.staticRow(),
+ wrapped.isReverseOrder(),
+ wrapped.stats());
+ this.wrapped = wrapped;
+ this.inMemoryMaxRows = inMemoryMaxRows;
+ this.inMemoryMaxHeapSize = inMemoryMaxHeapSize;
+ }
+
+ boolean hasOverflow()
+ {
+ return hasOverflow;
+ }
+
+ boolean overflowedByRowLimit()
+ {
+ return overflowByRowLimit;
+ }
+
+ @Override
+ protected Unfiltered computeNext()
+ {
+ if (!wrapped.hasNext())
+ return endOfData();
+
+ if (!insideOpenMarker)
+ {
+ if (inMemoryMaxRows > 0 && unfilteredCount >= inMemoryMaxRows)
+ return overflow(true);
+ if (inMemoryMaxHeapSize > 0 && accumulatedHeapSize >= inMemoryMaxHeapSize)
+ return overflow(false);
+ }
+
+ Unfiltered next = wrapped.next();
+ if (next.isRangeTombstoneMarker())
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+ insideOpenMarker = marker.isOpen(isReverseOrder);
+ }
+ else
+ {
+ next = withoutShadowedData((Row) next);
+ }
+ unfilteredCount++;
+ // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+ if (inMemoryMaxHeapSize > 0)
+ accumulatedHeapSize += unsharedHeapSize(next);
Review Comment:
I think we just need to call it out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r4020538337
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +411,317 @@ public boolean isDigestResponse()
{
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
+ {
+ // An empty static row which is intentionally not the Rows.EMPTY_STATIC_ROW singleton
+ private static final Row PRESENT_EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
+
+ // 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ if (!iter.hasNext())
+ {
+ // Empty response
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+ }
+
+ ImmutableBTreePartition partition;
+ ByteBuffer serialized = null;
+ // Closing rowIter is our job, LimitedUnfilteredRowIterator deliberately never closes what it wraps.
+ // The only exception is the overflow path, where serialize() takes it over, hence no try-with-resources.
+ UnfilteredRowIterator rowIter = iter.next();
+ try
+ {
+ LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, inMemoryMaxRows, inMemoryMaxHeapSize);
+ 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);
+ rowIter = null;
+ }
+ }
+ finally
+ {
+ if (rowIter != null)
+ rowIter.close();
+ }
+
+ // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+ if (serialized != null)
+ return new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive());
+
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return 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 = unfilteredIteratorAsRead(prefix, command.columnFilter(), 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 inMemoryMaxRows;
+ private final long inMemoryMaxHeapSize;
+
+ 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ super(wrapped.metadata(),
+ wrapped.partitionKey(),
+ wrapped.partitionLevelDeletion(),
+ wrapped.columns(),
+ wrapped.staticRow(),
+ wrapped.isReverseOrder(),
+ wrapped.stats());
+ this.wrapped = wrapped;
+ this.inMemoryMaxRows = inMemoryMaxRows;
+ this.inMemoryMaxHeapSize = inMemoryMaxHeapSize;
+ }
+
+ boolean hasOverflow()
+ {
+ return hasOverflow;
+ }
+
+ boolean overflowedByRowLimit()
+ {
+ return overflowByRowLimit;
+ }
+
+ @Override
+ protected Unfiltered computeNext()
+ {
+ if (!wrapped.hasNext())
+ return endOfData();
+
+ if (!insideOpenMarker)
+ {
+ if (inMemoryMaxRows > 0 && unfilteredCount >= inMemoryMaxRows)
+ return overflow(true);
+ if (inMemoryMaxHeapSize > 0 && accumulatedHeapSize >= inMemoryMaxHeapSize)
+ return overflow(false);
+ }
+
+ Unfiltered next = wrapped.next();
+ if (next.isRangeTombstoneMarker())
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+ insideOpenMarker = marker.isOpen(isReverseOrder);
+ }
+ else
+ {
+ next = withoutShadowedData((Row) next);
+ }
+ unfilteredCount++;
+ // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+ if (inMemoryMaxHeapSize > 0)
+ accumulatedHeapSize += unsharedHeapSize(next);
Review Comment:
to measure heap size we have to consume an element from the original iterator, we cannot put it back easily. We do not keep it for a long time and almost immediate pass to serialization in case of overflow. Also we cannot deserialize a multi MiB row partially, so in any case we have to keep it in memory for while.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3229698449
##########
src/java/org/apache/cassandra/db/ReadCommand.java:
##########
@@ -436,14 +436,26 @@ public ReadCommand copyAsDigestQuery(Iterable<Replica> replicas)
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)
Review Comment:
NIT line break?
```suggestion
public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead)
```
##########
test/unit/org/apache/cassandra/db/ReadResponseTest.java:
##########
@@ -172,6 +191,212 @@ public void makeDigestDoesntConsiderRepairedDataInfo()
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);
+ }
Review Comment:
We are missing a test that has a RT marker between in memory rows and the overflow rows. I don't think we are handling this case correctly. We have a potential data resurrection issue here. Can we add this test?
```suggestion
}
@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);
}
```
The fix I suggest is in `org.apache.cassandra.db.ReadResponse.InMemoryDataResponse.LimitedUnfilteredRowIterator#computeNext` to ensure the RT open/close pairs are never split . We either keep them entirely in memory or in overflow:
```
private boolean insideOpenMarker = false;
@Override
protected Unfiltered computeNext()
{
if (!wrapped.hasNext())
return endOfData();
if (rowCount >= maxRows && !insideOpenMarker)
{
hasOverflow = true;
return endOfData();
}
Unfiltered next = wrapped.next();
if (next.isRangeTombstoneMarker())
{
RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
insideOpenMarker = marker.isOpen(isReverseOrder);
}
else
rowCount++;
return next;
}
```
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -288,8 +325,8 @@ static abstract class DataResponse extends 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;
Review Comment:
can we keep these as `private final` and instead provide a static method to `build` an `InMemoryDataResponse` object? something like this:
```
static InMemoryDataResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows)
{
ImmutableBTreePartition partition;
ByteBuffer overflow;
if (!iter.hasNext())
{
partition = null;
overflow = null;
}
else
{
try (UnfilteredRowIterator rowIter = iter.next())
{
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;
}
}
// 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());
}
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;
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3799440645
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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)
Review Comment:
addressed
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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)
Review Comment:
addressed
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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.
Review Comment:
addressed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3804613879
##########
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
Review Comment:
a counter for cases when a local single-partition read response kept in memory is added, to be able to see the ratio
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3756314490
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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.
Review Comment:
thanks, yes, the comment is out of date
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima closed pull request #4789: [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read
URL: https://github.com/apache/cassandra/pull/4789
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3237167612
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -288,8 +325,8 @@ static abstract class DataResponse extends 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;
Review Comment:
thanks, good idea, I was too focused on trying to keep the logic in a constructor and missed an option with a factory method.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r4024523502
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +411,317 @@ public boolean isDigestResponse()
{
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
+ {
+ // An empty static row which is intentionally not the Rows.EMPTY_STATIC_ROW singleton
+ private static final Row PRESENT_EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
+
+ // 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ if (!iter.hasNext())
+ {
+ // Empty response
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+ }
+
+ ImmutableBTreePartition partition;
+ ByteBuffer serialized = null;
+ // Closing rowIter is our job, LimitedUnfilteredRowIterator deliberately never closes what it wraps.
+ // The only exception is the overflow path, where serialize() takes it over, hence no try-with-resources.
+ UnfilteredRowIterator rowIter = iter.next();
+ try
+ {
+ LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, inMemoryMaxRows, inMemoryMaxHeapSize);
+ 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);
+ rowIter = null;
+ }
+ }
+ finally
+ {
+ if (rowIter != null)
+ rowIter.close();
+ }
+
+ // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+ if (serialized != null)
+ return new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive());
+
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return 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 = unfilteredIteratorAsRead(prefix, command.columnFilter(), 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 inMemoryMaxRows;
+ private final long inMemoryMaxHeapSize;
+
+ 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ super(wrapped.metadata(),
+ wrapped.partitionKey(),
+ wrapped.partitionLevelDeletion(),
+ wrapped.columns(),
+ wrapped.staticRow(),
+ wrapped.isReverseOrder(),
+ wrapped.stats());
+ this.wrapped = wrapped;
+ this.inMemoryMaxRows = inMemoryMaxRows;
+ this.inMemoryMaxHeapSize = inMemoryMaxHeapSize;
+ }
+
+ boolean hasOverflow()
+ {
+ return hasOverflow;
+ }
+
+ boolean overflowedByRowLimit()
+ {
+ return overflowByRowLimit;
+ }
+
+ @Override
+ protected Unfiltered computeNext()
+ {
+ if (!wrapped.hasNext())
+ return endOfData();
+
+ if (!insideOpenMarker)
+ {
+ if (inMemoryMaxRows > 0 && unfilteredCount >= inMemoryMaxRows)
+ return overflow(true);
+ if (inMemoryMaxHeapSize > 0 && accumulatedHeapSize >= inMemoryMaxHeapSize)
+ return overflow(false);
+ }
+
+ Unfiltered next = wrapped.next();
+ if (next.isRangeTombstoneMarker())
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+ insideOpenMarker = marker.isOpen(isReverseOrder);
+ }
+ else
+ {
+ next = withoutShadowedData((Row) next);
+ }
+ unfilteredCount++;
+ // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+ if (inMemoryMaxHeapSize > 0)
+ accumulatedHeapSize += unsharedHeapSize(next);
+ return next;
+ }
+
+ // A row read back from a buffer is rebuilt cell by cell on top of its row deletion using BTreeRow.Builder.
+ // The builder silently drops whatever that deletion covers.
+ // So, we rebuild here the row in the same way, to avoid a digest mismatch.
+ private static Row withoutShadowedData(Row row)
+ {
+ if (row.deletion().isLive())
+ return row;
+
+ Row.Builder builder = BTreeRow.sortedBuilder();
+ builder.newRow(row.clustering());
+ builder.addPrimaryKeyLivenessInfo(row.primaryKeyLivenessInfo());
+ builder.addRowDeletion(row.deletion());
+ for (ColumnData cd : row)
+ {
+ if (cd.column().isSimple())
+ {
+ builder.addCell((Cell<?>) cd);
+ }
+ else
+ {
+ ComplexColumnData complexData = (ComplexColumnData) cd;
+ if (!complexData.complexDeletion().isLive())
+ builder.addComplexDeletion(complexData.column(), complexData.complexDeletion());
+ for (Cell<?> cell : complexData)
+ builder.addCell(cell);
+ }
+ }
+ return builder.build();
+ }
+
+ 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 = unfilteredIteratorAsRead(partition, command.columnFilter(), command.isReversed());
+ return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter);
+ }
+
+ // Iterating an ImmutableBTreePartition partition does not give back the iterator the buffer deserialized read produced.
+ // The method adjusts the iterator to produce results in the same way as the deserialized path, to avoid a digest mismatch.
+ private static UnfilteredRowIterator unfilteredIteratorAsRead(ImmutableBTreePartition partition, ColumnFilter columnFilter, boolean reversed)
+ {
+ UnfilteredRowIterator iter = partition.unfilteredIterator(columnFilter, Slices.ALL, reversed);
+
+ // A partition with nothing in it is serialized as a flag and read back without columns and without a static row
+ if (iter.isEmpty())
+ {
+ iter.close();
+ return EmptyIterators.unfilteredRow(partition.metadata(), partition.partitionKey(), reversed);
+ }
+
+ Row staticRow = iter.staticRow() == Rows.EMPTY_STATIC_ROW && partition.staticRow() != Rows.EMPTY_STATIC_ROW
+ ? PRESENT_EMPTY_STATIC_ROW
+ : iter.staticRow();
+ // the static columns are serialized together with a static row only
+ RegularAndStaticColumns columns = staticRow == Rows.EMPTY_STATIC_ROW
+ ? new RegularAndStaticColumns(Columns.NONE, partition.columns().regulars)
+ : partition.columns();
+
+ if (staticRow == iter.staticRow() && columns.equals(iter.columns()))
+ return iter;
+
+ return new WrappingUnfilteredRowIterator()
+ {
+ @Override
+ public UnfilteredRowIterator wrapped()
+ {
+ return iter;
+ }
+
+ @Override
+ public RegularAndStaticColumns columns()
+ {
+ return columns;
+ }
+
+ @Override
+ public Row staticRow()
+ {
+ return staticRow;
+ }
+ };
+ }
+
+ @Override
+ protected ByteBuffer getSerializedData()
+ {
+ throw new UnsupportedOperationException("InMemoryDataResponse cannot be serialized over the network");
+ }
+ }
+
+ private static class SingletonUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator
Review Comment:
I was worrying about deriving of TableMetadata from iterator vs using the value from command (as deserialisation logic do) because it could introduce some discrepancies in digest as we had for columns but it seems there is no such issue for TableMetadata.., so I will double check it and remove the code duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3799459326
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +404,214 @@ public boolean isDigestResponse()
{
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())
Review Comment:
addressed, notes:
1) it does not reproduce with the current logic because BaseIterator.close has a protection against double close, at the same time it is a kind of implementation details, which is not guaranteed, so I've adjusted the logic to avoid a potential issue
2) I have not used a wrapping iterator to avoid extra object allocation as well as to not increase the number of types invoked on a hot path (extra risks to get metamorphic calls)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#issuecomment-5713688042
Merged as https://github.com/apache/cassandra/commit/72affbc20669f1b610dd2c997fc1f6c642f017c5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#issuecomment-5688590472
We need to call out this change in NEWS.txt also
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3243910632
##########
test/unit/org/apache/cassandra/db/ReadResponseTest.java:
##########
@@ -172,6 +191,212 @@ public void makeDigestDoesntConsiderRepairedDataInfo()
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);
+ }
Review Comment:
I've applied a very similar change + extended test coverage for this logic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "iamaleksey (via GitHub)" <gi...@apache.org>.
iamaleksey commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3750204089
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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)
Review Comment:
Ditto.
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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.
Review Comment:
Is this part of the comment still accurate?
##########
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java:
##########
@@ -207,6 +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"),
Review Comment:
With byte-size limits implemented, is there really value in doing both row + byte size? I would imagine that byte-size is strictly superior, with row-size being quite arbitrary in practice (think huge collections, for example).
##########
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java:
##########
@@ -207,6 +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"),
Review Comment:
It might be, I'm just struggling to see how. So this is a genuine question, and *not* a request to change disguised as a question.
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -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)
Review Comment:
This method name is ambiguous in context of the class and can be misinterpreted by fellow humans. Perhaps follow the naming of the related constants instead?
##########
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
Review Comment:
I would argue that:
1. These metrics belong in `TableMetrics`, and
2. It's currently not straightforward at all to calculate what the hit ratio is
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +404,214 @@ public boolean isDigestResponse()
{
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())
Review Comment:
I think you end up closing this iterator twice (first time eventually inside `UnfilteredPartitionIterators.Serializer#serialize()), and it has visible side-effects (metrics)).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3759918743
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +404,214 @@ public boolean isDigestResponse()
{
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())
Review Comment:
good catch, thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#issuecomment-5696085827
added and rebased on fresh trunk
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3237170672
##########
test/unit/org/apache/cassandra/db/ReadResponseTest.java:
##########
@@ -172,6 +191,212 @@ public void makeDigestDoesntConsiderRepairedDataInfo()
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);
+ }
Review Comment:
thank you for highlighting the issue!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r4019617506
##########
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java:
##########
@@ -207,6 +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 **/
Review Comment:
I think it's worth a comment that these limits won't be enforced across a range tombstone span (for correctness reasons).
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +411,317 @@ public boolean isDigestResponse()
{
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
+ {
+ // An empty static row which is intentionally not the Rows.EMPTY_STATIC_ROW singleton
+ private static final Row PRESENT_EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
+
+ // 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ if (!iter.hasNext())
+ {
+ // Empty response
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+ }
+
+ ImmutableBTreePartition partition;
+ ByteBuffer serialized = null;
+ // Closing rowIter is our job, LimitedUnfilteredRowIterator deliberately never closes what it wraps.
+ // The only exception is the overflow path, where serialize() takes it over, hence no try-with-resources.
+ UnfilteredRowIterator rowIter = iter.next();
+ try
+ {
+ LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, inMemoryMaxRows, inMemoryMaxHeapSize);
+ 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);
+ rowIter = null;
+ }
+ }
+ finally
+ {
+ if (rowIter != null)
+ rowIter.close();
+ }
+
+ // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+ if (serialized != null)
+ return new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive());
+
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return 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 = unfilteredIteratorAsRead(prefix, command.columnFilter(), 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 inMemoryMaxRows;
+ private final long inMemoryMaxHeapSize;
+
+ 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ super(wrapped.metadata(),
+ wrapped.partitionKey(),
+ wrapped.partitionLevelDeletion(),
+ wrapped.columns(),
+ wrapped.staticRow(),
+ wrapped.isReverseOrder(),
+ wrapped.stats());
+ this.wrapped = wrapped;
+ this.inMemoryMaxRows = inMemoryMaxRows;
+ this.inMemoryMaxHeapSize = inMemoryMaxHeapSize;
+ }
+
+ boolean hasOverflow()
+ {
+ return hasOverflow;
+ }
+
+ boolean overflowedByRowLimit()
+ {
+ return overflowByRowLimit;
+ }
+
+ @Override
+ protected Unfiltered computeNext()
+ {
+ if (!wrapped.hasNext())
+ return endOfData();
+
+ if (!insideOpenMarker)
+ {
+ if (inMemoryMaxRows > 0 && unfilteredCount >= inMemoryMaxRows)
+ return overflow(true);
+ if (inMemoryMaxHeapSize > 0 && accumulatedHeapSize >= inMemoryMaxHeapSize)
+ return overflow(false);
+ }
+
+ Unfiltered next = wrapped.next();
+ if (next.isRangeTombstoneMarker())
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+ insideOpenMarker = marker.isOpen(isReverseOrder);
+ }
+ else
+ {
+ next = withoutShadowedData((Row) next);
+ }
+ unfilteredCount++;
+ // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+ if (inMemoryMaxHeapSize > 0)
+ accumulatedHeapSize += unsharedHeapSize(next);
+ return next;
+ }
+
+ // A row read back from a buffer is rebuilt cell by cell on top of its row deletion using BTreeRow.Builder.
+ // The builder silently drops whatever that deletion covers.
+ // So, we rebuild here the row in the same way, to avoid a digest mismatch.
+ private static Row withoutShadowedData(Row row)
+ {
+ if (row.deletion().isLive())
+ return row;
+
+ Row.Builder builder = BTreeRow.sortedBuilder();
+ builder.newRow(row.clustering());
+ builder.addPrimaryKeyLivenessInfo(row.primaryKeyLivenessInfo());
+ builder.addRowDeletion(row.deletion());
+ for (ColumnData cd : row)
+ {
+ if (cd.column().isSimple())
+ {
+ builder.addCell((Cell<?>) cd);
+ }
+ else
+ {
+ ComplexColumnData complexData = (ComplexColumnData) cd;
+ if (!complexData.complexDeletion().isLive())
+ builder.addComplexDeletion(complexData.column(), complexData.complexDeletion());
+ for (Cell<?> cell : complexData)
+ builder.addCell(cell);
+ }
+ }
+ return builder.build();
+ }
+
+ 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 = unfilteredIteratorAsRead(partition, command.columnFilter(), command.isReversed());
+ return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter);
+ }
+
+ // Iterating an ImmutableBTreePartition partition does not give back the iterator the buffer deserialized read produced.
+ // The method adjusts the iterator to produce results in the same way as the deserialized path, to avoid a digest mismatch.
+ private static UnfilteredRowIterator unfilteredIteratorAsRead(ImmutableBTreePartition partition, ColumnFilter columnFilter, boolean reversed)
+ {
+ UnfilteredRowIterator iter = partition.unfilteredIterator(columnFilter, Slices.ALL, reversed);
+
+ // A partition with nothing in it is serialized as a flag and read back without columns and without a static row
+ if (iter.isEmpty())
+ {
+ iter.close();
+ return EmptyIterators.unfilteredRow(partition.metadata(), partition.partitionKey(), reversed);
+ }
+
+ Row staticRow = iter.staticRow() == Rows.EMPTY_STATIC_ROW && partition.staticRow() != Rows.EMPTY_STATIC_ROW
+ ? PRESENT_EMPTY_STATIC_ROW
+ : iter.staticRow();
+ // the static columns are serialized together with a static row only
+ RegularAndStaticColumns columns = staticRow == Rows.EMPTY_STATIC_ROW
+ ? new RegularAndStaticColumns(Columns.NONE, partition.columns().regulars)
+ : partition.columns();
+
+ if (staticRow == iter.staticRow() && columns.equals(iter.columns()))
+ return iter;
+
+ return new WrappingUnfilteredRowIterator()
+ {
+ @Override
+ public UnfilteredRowIterator wrapped()
+ {
+ return iter;
+ }
+
+ @Override
+ public RegularAndStaticColumns columns()
+ {
+ return columns;
+ }
+
+ @Override
+ public Row staticRow()
+ {
+ return staticRow;
+ }
+ };
+ }
+
+ @Override
+ protected ByteBuffer getSerializedData()
+ {
+ throw new UnsupportedOperationException("InMemoryDataResponse cannot be serialized over the network");
+ }
+ }
+
+ private static class SingletonUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator
Review Comment:
Can't we use `org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator` instead? Metadata is derived from `iter.metadata()`. and the implementation is almost 1:1 to this implementation.
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +411,317 @@ public boolean isDigestResponse()
{
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
+ {
+ // An empty static row which is intentionally not the Rows.EMPTY_STATIC_ROW singleton
+ private static final Row PRESENT_EMPTY_STATIC_ROW = BTreeRow.emptyRow(Clustering.STATIC_CLUSTERING);
+
+ // 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ if (!iter.hasNext())
+ {
+ // Empty response
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive());
+ }
+
+ ImmutableBTreePartition partition;
+ ByteBuffer serialized = null;
+ // Closing rowIter is our job, LimitedUnfilteredRowIterator deliberately never closes what it wraps.
+ // The only exception is the overflow path, where serialize() takes it over, hence no try-with-resources.
+ UnfilteredRowIterator rowIter = iter.next();
+ try
+ {
+ LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, inMemoryMaxRows, inMemoryMaxHeapSize);
+ 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);
+ rowIter = null;
+ }
+ }
+ finally
+ {
+ if (rowIter != null)
+ rowIter.close();
+ }
+
+ // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected.
+ if (serialized != null)
+ return new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive());
+
+ ReadResponseMetrics.inMemoryResponses.inc();
+ return 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 = unfilteredIteratorAsRead(prefix, command.columnFilter(), 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 inMemoryMaxRows;
+ private final long inMemoryMaxHeapSize;
+
+ 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 inMemoryMaxRows, long inMemoryMaxHeapSize)
+ {
+ super(wrapped.metadata(),
+ wrapped.partitionKey(),
+ wrapped.partitionLevelDeletion(),
+ wrapped.columns(),
+ wrapped.staticRow(),
+ wrapped.isReverseOrder(),
+ wrapped.stats());
+ this.wrapped = wrapped;
+ this.inMemoryMaxRows = inMemoryMaxRows;
+ this.inMemoryMaxHeapSize = inMemoryMaxHeapSize;
+ }
+
+ boolean hasOverflow()
+ {
+ return hasOverflow;
+ }
+
+ boolean overflowedByRowLimit()
+ {
+ return overflowByRowLimit;
+ }
+
+ @Override
+ protected Unfiltered computeNext()
+ {
+ if (!wrapped.hasNext())
+ return endOfData();
+
+ if (!insideOpenMarker)
+ {
+ if (inMemoryMaxRows > 0 && unfilteredCount >= inMemoryMaxRows)
+ return overflow(true);
+ if (inMemoryMaxHeapSize > 0 && accumulatedHeapSize >= inMemoryMaxHeapSize)
+ return overflow(false);
+ }
+
+ Unfiltered next = wrapped.next();
+ if (next.isRangeTombstoneMarker())
+ {
+ RangeTombstoneMarker marker = (RangeTombstoneMarker) next;
+ insideOpenMarker = marker.isOpen(isReverseOrder);
+ }
+ else
+ {
+ next = withoutShadowedData((Row) next);
+ }
+ unfilteredCount++;
+ // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed
+ if (inMemoryMaxHeapSize > 0)
+ accumulatedHeapSize += unsharedHeapSize(next);
Review Comment:
we accumulate first and then check later in line 558. I think there could be cases where we exceed the limits (and we could potentially have a multi MiB row). I think we should at least document this is intentional.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3756410654
##########
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java:
##########
@@ -207,6 +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"),
Review Comment:
My main worry is about the case when we have many small rows and the number of objects is high but total heap size is not that big, counting of objects is too complicated and introduces extra overheads, so as a trade-off I decided to preserve the row check... From runtime overhead point of view row limit is cheap (compared to heap size calculation which requires object graph traversal).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "netudima (via GitHub)" <gi...@apache.org>.
netudima commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3757114656
##########
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
Review Comment:
1. Yes, I was thinking about it but decided not to go in this way because we already have a lot of Table-level metrics and this one I suppose will be rarely used to introduce such overhead + it requires to add a lookup of CFS object to get metrics object. At the same time the overheads are small, so if you think it is beneficial to have per table granularity here, I do not see a problem to add it.
2. Yes, we have currently org.apache.cassandra.metrics.ClientRequestMetrics#localRequests which helps to estimate the number of local replica reads, but it does not differentiate the cases then the local replica was actually queried + the single partition case, so it makes sense to add an extra metric here to see a ratio. From overhead point of view it is better to have misses count to still have one metric update per request, while from usability point of view total probably is slightly easier to use..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] [CASSANDRA-21354][trunk] Avoid serialization and deserialization for coordinator-local single partition data read [cassandra]
Posted by "iamaleksey (via GitHub)" <gi...@apache.org>.
iamaleksey commented on code in PR #4789:
URL: https://github.com/apache/cassandra/pull/4789#discussion_r3750868089
##########
src/java/org/apache/cassandra/db/ReadResponse.java:
##########
@@ -354,6 +404,214 @@ public boolean isDigestResponse()
{
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())
Review Comment:
I think you end up closing this iterator twice (first time eventually inside `UnfilteredPartitionIterators.Serializer#serialize()`, and it has visible side-effects (metrics)).
##########
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java:
##########
@@ -207,6 +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"),
Review Comment:
Fair enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org