natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization 

patch by C. Scott Andreas; reviewed by Caleb Rackliffe and Dmitry Konstantinov for CASSANDRA-21285

Co-authored-by: C. Scott Andreas <cscotta@users.noreply.github.com>
Co-authored-by: Caleb Rackliffe <calebrackliffe@gmail.com>
This commit is contained in:
C. Scott Andreas 2026-04-04 17:34:06 -07:00 committed by Caleb Rackliffe
parent 6a665f3ee4
commit a657f4bef9
3 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
6.0-alpha2
* Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization (CASSANDRA-21285)
* Implement a guardrail for client driver versions (CASSANDRA-21146)
* Enable IAuthenticator to declare supported and alterable role options (CASSANDRA-20834)
* Avoid capturing lambda allocation in UnfilteredSerializer.deserializeRowBody (CASSANDRA-21289)

2
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -205,6 +205,8 @@ public enum CassandraRelevantProperties
/** Controls the type of bufffer (heap/direct) used for shared scratch buffers */
DATA_OUTPUT_BUFFER_ALLOCATE_TYPE("cassandra.dob.allocate_type"),
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"),
DECAYING_ESTIMATED_HISTOGRAM_RESERVOIR_STRIPE_COUNT("cassandra.dehr_stripe_count", "2"),
DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES("default.provide.overlapping.tombstones"),
/** determinism properties for testing */

23
src/java/org/apache/cassandra/db/ReadResponse.java
View File

@ -24,6 +24,7 @@ import java.util.List;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
@ -38,6 +39,8 @@ import org.apache.cassandra.io.util.DataOutputPlus;
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;
@ -224,6 +227,11 @@ public abstract class ReadResponse
// built on the owning node responding to a query
private static class LocalDataResponse extends DataResponse
{
// Exponential moving average of response sizes, used to set initial size of output buffer.
private static final MovingAverage estimatedResponseBytes = ExpMovingAverage.decayBy1000();
private static final int bufferInitialSizeMin = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MIN.getInt();
private static final int bufferInitialSizeMax = CassandraRelevantProperties.DATA_RESPONSE_BUFFER_INITIAL_SIZE_MAX.getInt();
private LocalDataResponse(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi)
{
super(build(iter, command.columnFilter()),
@ -239,10 +247,21 @@ public abstract class ReadResponse
private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection)
{
try (DataOutputBuffer buffer = new DataOutputBuffer())
int initialBufferSize = bufferInitialSizeMin;
if (bufferInitialSizeMax > bufferInitialSizeMin)
{
// 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) ? bufferInitialSizeMin : estimatedResponseSize * 1.1;
initialBufferSize = Math.min((int) bufferSizeEstimate, bufferInitialSizeMax);
}
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)
{