Skip to content

Reduce allocations + copies due to rebuffering in LocalDataResponse row serialization #4707

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
cscotta wants to merge 2 commits into
base: trunk
Choose a base branch
from
+14 −2
Open

Reduce allocations + copies due to rebuffering in LocalDataResponse row serialization #4707

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1f5c87c
Reduce allocations + copies due to rebuffering in LocalDataResponse r…
cscotta Apr 5, 2026
df95df1
Eliminate duplicate call to estimatedResponseBytes.get(); return outp…
cscotta Apr 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 14 additions & 2 deletions src/java/org/apache/cassandra/db/ReadResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ExpMovingAverage;
import org.apache.cassandra.utils.MovingAverage;

import static org.apache.cassandra.db.RepairedDataInfo.NO_OP_REPAIRED_DATA_INFO;

Expand Down Expand Up @@ -224,6 +226,9 @@ public boolean isDigestResponse()
// built on the owning node responding to a query
private static class LocalDataResponse extends DataResponse
{
// Moving average of response sizes, used to set initial size of output buffer.
private static final MovingAverage estimatedResponseBytes = ExpMovingAverage.decayBy1000();

private LocalDataResponse(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi)
{
super(build(iter, command.columnFilter()),
Expand All @@ -239,10 +244,17 @@ private LocalDataResponse(UnfilteredPartitionIterator iter, ColumnFilter selecti

private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection)
{
try (DataOutputBuffer buffer = new DataOutputBuffer())
// Size output buffer to 10% above the moving average to absorb minor variance and limit rebuffering.
double estimatedResponseSize = estimatedResponseBytes.get();
double bufferSizeEstimate = Double.isNaN(estimatedResponseSize) ? 128 : estimatedResponseSize;
int initialBufferSize = (int) (bufferSizeEstimate * 1.1);

try (DataOutputBuffer buffer = new DataOutputBuffer(initialBufferSize))
{
UnfilteredPartitionIterators.serializerForIntraNode().serialize(iter, selection, buffer, MessagingService.current_version);
return buffer.buffer();
estimatedResponseBytes.update(buffer.position());

return buffer.buffer(false);
}
catch (IOException e)
{
Expand Down