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

Reduce memory allocations in row merge logic 

org.apache.cassandra.utils.MergeIterator.Candidate - has Comparator field which is the same for all Candidates under an MergeIterator instance, we can move the field to MergeIterator level. Candidate is 4.8% of all allocations and we can save 20% (~1% of all allocations) of it for JDK 21 with compressed references on.

 switch from ArrayList to array in org.apache.cassandra.db.rows.Row.Merger
The ArrayList iterator costs 0.72% and BulkIterator.Adapter costs 0.35%. So, in total we can save here ~1% of total allocations.

patch by Dmitry Konstantinov; reviewed by Francisco Guerrero for CASSANDRA-21359
This commit is contained in:
Dmitry Konstantinov 2026-05-10 14:10:56 +01:00
parent b3bc7c019f
commit 71e8b7c213
3 changed files with 54 additions and 17 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 memory allocations in row merge logic (CASSANDRA-21359)
* Restore option to avoid hint transfer during decommission (CASSANDRA-21341)
* Add an offline cluster metadata tool (CASSANDRA-19151)
* Accord: Tail Latency Improvements (CASSANDRA-21361)

49
src/java/org/apache/cassandra/db/rows/Row.java
View File

@ -43,11 +43,13 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import org.apache.cassandra.utils.memory.Cloner;
/**
@ -716,7 +718,10 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
private int rowsToMerge;
private int lastRowSet = -1;
private final List<ColumnData> dataBuffer = new ArrayList<>();
private static final ColumnData[] EMPTY_DATA_BUFFER = new ColumnData[0];
private ColumnData[] dataBuffer = EMPTY_DATA_BUFFER;
private int dataBufferSize;
private final ColumnDataReducer columnDataReducer;
public Merger(int size, boolean hasComplex)
@ -728,7 +733,8 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
public void clear()
{
dataBuffer.clear();
Arrays.fill(dataBuffer, 0, dataBufferSize, null);
dataBufferSize = 0;
Arrays.fill(rows, null);
columnDataIterators.clear();
rowsToMerge = 0;
@ -778,8 +784,22 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (activeDeletion.deletes(rowInfo))
rowInfo = LivenessInfo.EMPTY;
int columnsCountEstimation = 0;
for (Row row : rows)
columnDataIterators.add(row == null ? Collections.emptyIterator() : row.iterator());
{
if (row != null)
{
columnDataIterators.add(row.iterator());
columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
}
else
{
columnDataIterators.add(Collections.emptyIterator());
}
}
// try to estimate and set a potential target capacity
if (dataBuffer.length < columnsCountEstimation)
dataBuffer = new ColumnData[columnsCountEstimation];
columnDataReducer.setActiveDeletion(activeDeletion);
Iterator<ColumnData> merged = MergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
@ -787,13 +807,28 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
{
ColumnData data = merged.next();
if (data != null)
dataBuffer.add(data);
{
ensureDataBufferCapacity();
dataBuffer[dataBufferSize++] = data;
}
}
// Because some data might have been shadowed by the 'activeDeletion', we could have an empty row
return rowInfo.isEmpty() && rowDeletion.isLive() && dataBuffer.isEmpty()
? null
: BTreeRow.create(clustering, rowInfo, rowDeletion, BTree.build(dataBuffer));
if (rowInfo.isEmpty() && rowDeletion.isLive() && dataBufferSize == 0)
return null;
try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer))
{
return BTreeRow.create(clustering, rowInfo, rowDeletion,
BTree.build(it, dataBufferSize, UpdateFunction.noOp()));
}
}
private void ensureDataBufferCapacity()
{
if (dataBufferSize == dataBuffer.length)
// increase capacity by 50%, use 4 as a default capacity
dataBuffer = Arrays.copyOf(dataBuffer, Math.max(dataBuffer.length + (dataBuffer.length >> 1), 4));
}
public Clustering<?> mergedClustering()

21
src/java/org/apache/cassandra/utils/MergeIterator.java
View File

@ -149,6 +149,8 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
{
protected final Candidate<In>[] heap;
private final Comparator<? super In> comp;
/** Number of non-exhausted iterators. */
int size;
@ -174,9 +176,10 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
this.heap = heap;
size = 0;
this.comp = comp;
for (int i = 0; i < iters.size(); i++)
{
Candidate<In> candidate = new Candidate<>(i, iters.get(i), comp);
Candidate<In> candidate = new Candidate<>(i, iters.get(i));
heap[size++] = candidate;
}
needingAdvance = size;
@ -292,7 +295,7 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
{
if (!heap[nextIdx].equalParent) // if we were greater then an (or were the) equal parent, we are >= the child
{
int cmp = candidate.compareTo(heap[nextIdx]);
int cmp = candidate.compareTo(heap[nextIdx], comp);
if (cmp <= 0)
{
heap[nextIdx].equalParent = cmp == 0;
@ -316,12 +319,12 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
if (!heap[nextIdx + 1].equalParent)
{
// pick the smallest of the two children
int siblingCmp = heap[nextIdx + 1].compareTo(heap[nextIdx]);
int siblingCmp = heap[nextIdx + 1].compareTo(heap[nextIdx], comp);
if (siblingCmp < 0)
++nextIdx;
// if we're smaller than this, we are done, and must only restore the heap and equalParent properties
int cmp = candidate.compareTo(heap[nextIdx]);
int cmp = candidate.compareTo(heap[nextIdx], comp);
if (cmp <= 0)
{
if (cmp == 0)
@ -362,7 +365,7 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
// ... but sometimes we will have one last child to compare against, that has no siblings
if (!heap[nextIdx].equalParent)
{
int cmp = candidate.compareTo(heap[nextIdx]);
int cmp = candidate.compareTo(heap[nextIdx], comp);
if (cmp <= 0)
{
heap[nextIdx].equalParent = cmp == 0;
@ -377,19 +380,17 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
}
// Holds and is comparable by the head item of an iterator it owns
protected static final class Candidate<In> implements Comparable<Candidate<In>>
protected static final class Candidate<In>
{
private final Iterator<? extends In> iter;
private final Comparator<? super In> comp;
private final int idx;
private In item;
private In lowerBound;
boolean equalParent;
public Candidate(int idx, Iterator<? extends In> iter, Comparator<? super In> comp)
public Candidate(int idx, Iterator<? extends In> iter)
{
this.iter = iter;
this.comp = comp;
this.idx = idx;
this.lowerBound = iter instanceof IteratorWithLowerBound ? ((IteratorWithLowerBound<In>)iter).lowerBound() : null;
}
@ -410,7 +411,7 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
return this;
}
public int compareTo(Candidate<In> that)
int compareTo(Candidate<In> that, Comparator<? super In> comp)
{
assert this.item != null && that.item != null;
int ret = comp.compare(this.item, that.item);