You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2026/04/14 15:49:38 UTC
(cassandra) branch trunk updated: Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization
This is an automated email from the ASF dual-hosted git repository.
maedhroz pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new a657f4bef9 Reduce allocations and array copies due to buffer resizing in LocalDataResponse during row serialization
a657f4bef9 is described below
commit a657f4bef9f5f3a0aed98e81fdd07134581be283
Author: C. Scott Andreas <cs...@users.noreply.github.com>
AuthorDate: Sat Apr 4 17:34:06 2026 -0700
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 <cs...@users.noreply.github.com>
Co-authored-by: Caleb Rackliffe <ca...@gmail.com>
---
CHANGES.txt | 1 +
.../config/CassandraRelevantProperties.java | 2 ++
src/java/org/apache/cassandra/db/ReadResponse.java | 23 ++++++++++++++++++++--
3 files changed, 24 insertions(+), 2 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 84812a1cad..7b3c3c86be 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -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)
diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
index 8a60d44797..c4adaf0c2a 100644
--- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
+++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
@@ -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 */
diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java
index 65e17a6920..7edd743ad4 100644
--- a/src/java/org/apache/cassandra/db/ReadResponse.java
+++ b/src/java/org/apache/cassandra/db/ReadResponse.java
@@ -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)
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org