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

Merge branch 'cassandra-6.0' into trunk 

* cassandra-6.0:
  Apply performance optimizations for rows merging logic
This commit is contained in:
Dmitry Konstantinov 2026-07-29 23:40:30 +01:00
parent bcd9e662d2 1e3d43b2dd
commit e0e4feb120
9 changed files with 208 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -6,6 +6,7 @@
* Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
* Add a guardrail for misprepared statements (CASSANDRA-21139)
Merged from 6.0:
* Apply performance optimizations for rows merging logic (CASSANDRA-21524)
* Fix operationMode reporting DECOMMISSION_FAILED instead of LEAVING when resuming a failed decommission (CASSANDRA-21493)
* Avoid megamorphic calls when serializing and deserializing fixed-length values (CASSANDRA-21536)
* Avoid megamorphic calls for Cell.timestamp/ttl/path/localDeletionTimeAsUnsignedInt methods (CASSANDRA-21526)
@ -32,7 +33,7 @@ Merged from 6.0:
* Always send TCM commit failures as Messaging failures (CASSANDRA-21457)
* Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438)
* Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199)
* Don’t leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)
* Don't leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)
* Make nodetool abortbootstrap more robust (CASSANDRA-21235)
* Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234)
* Improve performance when deserializing cluster metadata (CASSANDRA-21224)

62
build.xml
View File

@ -651,6 +651,64 @@
<jflex file="${build.src.java}/org/apache/cassandra/index/sasi/analyzer/StandardTokenizerImpl.jflex" destdir="${build.src.gen-java}/" />
</target>
<!--
Copies a source class into ${build.src.gen-java} under a NEW class name while keeping it
in its ORIGINAL package, so it compiles into an independent class that still enjoys
package-private access to its neighbours.
This lets a hot call site use its own dedicated copy of a class (e.g. a RowMergeIterator
copy of MergeIterator) so that the virtual calls inside it stay mono-/bi-morphic and
remain inlinable, instead of turning megamorphic once every caller in the code base
funnels the shared original through the same bytecode.
The transformation is purely textual: every whole-word occurrence of the original simple
class name is rewritten to the new name. Matching on word boundaries keeps substrings
such as IMergeIterator intact. The package declaration is left untouched, so the copy
stays in the same package as the original and no imports need to change.
-->
<macrodef name="gen-java-copy">
<!-- Fully-qualified name of the class to copy, e.g. org.apache.cassandra.utils.MergeIterator -->
<attribute name="class"/>
<!-- New simple class name for the copy, e.g. RowMergeIterator -->
<attribute name="newName"/>
<sequential>
<local name="gen.src.rel"/>
<local name="gen.simple.name"/>
<!-- source file path relative to ${build.src.java}: dots -> slashes -->
<loadresource property="gen.src.rel">
<string value="@{class}"/>
<filterchain><tokenfilter><replaceregex pattern="\." replace="/" flags="g"/></tokenfilter></filterchain>
</loadresource>
<!-- original simple class name = last segment of the fully-qualified name -->
<loadresource property="gen.simple.name">
<string value="@{class}"/>
<filterchain><tokenfilter><replaceregex pattern="^.*\.([^.]+)$" replace="\1"/></tokenfilter></filterchain>
</loadresource>
<copy todir="${build.src.gen-java}" preservelastmodified="true">
<fileset dir="${build.src.java}" includes="${gen.src.rel}.java"/>
<!-- keep the original package directory, only rename the file -->
<mapper type="regexp" from="^(.*[\\/])[^\\/]+$" to="\1@{newName}.java"/>
<filterchain>
<tokenfilter>
<replaceregex pattern="\b${gen.simple.name}\b" replace="@{newName}" flags="g"/>
</tokenfilter>
</filterchain>
</copy>
</sequential>
</macrodef>
<!--
Generate the copies of hand-picked classes before compilation. Add a
<gen-java-copy class="..." newName="..."/> line here for every class that needs a
dedicated copy.
-->
<target name="gen-java-copies" depends="init"
description="Copy selected classes under new names (e.g. MergeIterator -> RowMergeIterator) into src/gen-java to avoid megamorphic calls">
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="RowMergeIterator"/>
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="UnfilteredMergeIterator"/>
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="ComplexCellMergeIterator"/>
</target>
<!-- create properties file with C version -->
<target name="_createVersionPropFile" depends="_get-git-sha,set-cqlsh-version,_set-build-date">
<taskdef name="propertyfile" classname="org.apache.tools.ant.taskdefs.optional.PropertyFile"/>
@ -710,7 +768,7 @@
</javac>
</target>
<target depends="init,gen-cql3-grammar,generate-cql-html,generate-jflex-java"
<target depends="init,gen-cql3-grammar,generate-cql-html,generate-jflex-java,gen-java-copies"
name="build-project">
<echo message="${ant.project.name}: ${ant.file}"/>
<!-- Order matters! -->
@ -2173,7 +2231,7 @@
</target>
<!-- Generate IDEA project description files -->
<target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,_createVersionPropFile" description="Generate IDEA files">
<target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,gen-java-copies,_createVersionPropFile" description="Generate IDEA files">
<delete dir=".idea"/>
<delete file="${eclipse.project.name}.iml"/>
<mkdir dir=".idea"/>

3
src/java/org/apache/cassandra/db/DeletionTime.java
View File

@ -178,7 +178,8 @@ public abstract class DeletionTime implements Comparable<DeletionTime>, IMeasura
public boolean deletes(Cell<?> cell)
{
return deletes(cell.timestamp());
// check for LIVE first to avoid a potential cell megamorphic call
return markedForDeleteAt() != MARKED_FOR_DELETE_AT_LIVE && deletes(cell.timestamp());
}
public boolean deletes(long timestamp)

5
src/java/org/apache/cassandra/db/rows/BTreeRow.java
View File

@ -439,6 +439,11 @@ public class BTreeRow extends AbstractRow
return nowInSec >= minLocalDeletionTime;
}
public long minLocalDeletionTime()
{
return minLocalDeletionTime;
}
public boolean hasInvalidDeletions()
{
if (primaryKeyLivenessInfo().isExpiring() && (primaryKeyLivenessInfo().ttl() < 0 || primaryKeyLivenessInfo().localExpirationTime() < 0))

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

@ -20,7 +20,6 @@ package org.apache.cassandra.db.rows;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
@ -44,9 +43,10 @@ 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.ComplexCellMergeIterator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.RowMergeIterator;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
@ -221,6 +221,15 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
*/
public boolean hasDeletion(long nowInSec);
/**
* The smallest local deletion time of all the data in this row (row deletion, primary key liveness, cells and
* complex deletions), or {@link Cell#MAX_DELETION_TIME} if the row has no deletion nor expiring data.
* <p>
* Unlike {@link #hasDeletion(long)}, this value is independent of the current time. In particular, a value of
* {@link Cell#MAX_DELETION_TIME} guarantees the row carries neither tombstones nor expiring data.
*/
public long minLocalDeletionTime();
/**
* An iterator to efficiently search data for a given column.
*
@ -762,6 +771,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
LivenessInfo rowInfo = LivenessInfo.EMPTY;
Deletion rowDeletion = Deletion.LIVE;
int columnsCountEstimation = 0;
// Track the smallest local deletion time across all inputs: if none of them carries any deletion or
// expiring data (i.e. this stays at MAX_DELETION_TIME), the merged row can't either, so we can hand the
// value to BTreeRow.create() below and skip the full btree scan it would otherwise do to recompute it.
long minDeletionTime = Cell.MAX_DELETION_TIME;
for (Row row : rows)
{
if (row == null)
@ -771,6 +785,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
rowInfo = row.primaryKeyLivenessInfo();
if (row.deletion().supersedes(rowDeletion))
rowDeletion = row.deletion();
minDeletionTime = Math.min(minDeletionTime, row.minLocalDeletionTime());
columnDataIterators.add(row.iterator());
columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
}
if (rowDeletion.isShadowedBy(rowInfo))
@ -784,25 +803,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (activeDeletion.deletes(rowInfo))
rowInfo = LivenessInfo.EMPTY;
int columnsCountEstimation = 0;
for (Row row : rows)
{
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);
Iterator<ColumnData> merged = RowMergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
while (merged.hasNext())
{
ColumnData data = merged.next();
@ -819,8 +825,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer))
{
return BTreeRow.create(clustering, rowInfo, rowDeletion,
BTree.build(it, dataBufferSize, UpdateFunction.noOp()));
Object[] tree = BTree.build(it, dataBufferSize, UpdateFunction.noOp());
// If none of the merged rows had any deletion or expiring data, neither does the result, so we can
// pass the already-known min local deletion time and avoid rescanning the whole btree to recompute it.
return minDeletionTime == Cell.MAX_DELETION_TIME
? BTreeRow.create(clustering, rowInfo, rowDeletion, tree, Cell.MAX_DELETION_TIME)
: BTreeRow.create(clustering, rowInfo, rowDeletion, tree);
}
}
@ -841,10 +851,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
return rows;
}
private static class ColumnDataReducer extends MergeIterator.Reducer<ColumnData, ColumnData>
private static class ColumnDataReducer extends RowMergeIterator.Reducer<ColumnData, ColumnData>
{
private ColumnMetadata column;
private final List<ColumnData> versions;
private final ColumnData[] versions;
private int versionsSize;
private DeletionTime activeDeletion;
@ -854,7 +865,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
public ColumnDataReducer(int size, boolean hasComplex)
{
this.versions = new ArrayList<>(size);
this.versions = new ColumnData[size];
this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null;
this.complexCells = hasComplex ? new ArrayList<>(size) : null;
this.cellReducer = new CellReducer();
@ -870,20 +881,24 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (useColumnMetadata(data.column()))
column = data.column();
versions.add(data);
versions[versionsSize++] = data;
}
/**
* Determines it the {@code ColumnMetadata} is the one that should be used.
* @param dataColumn the {@code ColumnMetadata} to use.
* @return {@code true} if the {@code ColumnMetadata} is the one that should be used, {@code false} otherwise.
* Determines whether {@code dataColumn} should replace the currently selected column metadata,
* i.e. whether no column has been selected yet or {@code dataColumn} is a newer version.
* @param dataColumn the candidate {@code ColumnMetadata} to evaluate.
* @return {@code true} if {@code dataColumn} should be used, {@code false} otherwise.
*/
private boolean useColumnMetadata(ColumnMetadata dataColumn)
{
if (column == null)
ColumnMetadata currentColumn = column;
if (currentColumn == null)
return true;
if (currentColumn == dataColumn)
return false;
return ColumnMetadataVersionComparator.INSTANCE.compare(column, dataColumn) < 0;
return ColumnMetadataVersionComparator.INSTANCE.compare(currentColumn, dataColumn) < 0;
}
protected ColumnData getReduced()
@ -891,9 +906,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (column.isSimple())
{
Cell<?> merged = null;
for (int i=0, isize=versions.size(); i<isize; i++)
for (int i = 0; i < versionsSize; i++)
{
Cell<?> cell = (Cell<?>) versions.get(i);
Cell<?> cell = (Cell<?>) versions[i];
if (!activeDeletion.deletes(cell))
merged = merged == null ? cell : Cells.reconcile(merged, cell);
}
@ -904,9 +919,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
complexBuilder.newColumn(column);
complexCells.clear();
DeletionTime complexDeletion = DeletionTime.LIVE;
for (int i=0, isize=versions.size(); i<isize; i++)
for (int i = 0; i < versionsSize; i++)
{
ColumnData data = versions.get(i);
ColumnData data = versions[i];
ComplexColumnData cd = (ComplexColumnData)data;
if (cd.complexDeletion().supersedes(complexDeletion))
complexDeletion = cd.complexDeletion();
@ -923,7 +938,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
cellReducer.setActiveDeletion(activeDeletion);
}
Iterator<Cell<?>> cells = MergeIterator.get(complexCells, Cell.comparator, cellReducer);
Iterator<Cell<?>> cells = ComplexCellMergeIterator.get(complexCells, Cell.comparator, cellReducer);
while (cells.hasNext())
{
Cell<?> merged = cells.next();
@ -937,11 +952,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
protected void onKeyChange()
{
column = null;
versions.clear();
Arrays.fill(versions, 0, versionsSize, null);
versionsSize = 0;
}
}
private static class CellReducer extends MergeIterator.Reducer<Cell<?>, Cell<?>>
private static class CellReducer extends ComplexCellMergeIterator.Reducer<Cell<?>, Cell<?>>
{
private DeletionTime activeDeletion;
private Cell<?> merged;

6
src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java
View File

@ -38,7 +38,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.IMergeIterator;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.UnfilteredMergeIterator;
/**
* Static methods to work with atom iterators.
@ -415,7 +415,7 @@ public abstract class UnfilteredRowIterators
reversed,
EncodingStats.merge(iterators, UnfilteredRowIterator::stats));
this.mergeIterator = MergeIterator.get(iterators,
this.mergeIterator = UnfilteredMergeIterator.get(iterators,
reversed ? metadata.comparator.reversed() : metadata.comparator,
new MergeReducer(iterators.size(), reversed, listener));
this.listener = listener;
@ -540,7 +540,7 @@ public abstract class UnfilteredRowIterators
listener.close();
}
private class MergeReducer extends MergeIterator.Reducer<Unfiltered, Unfiltered>
private class MergeReducer extends UnfilteredMergeIterator.Reducer<Unfiltered, Unfiltered>
{
private final MergeListener listener;

6
src/java/org/apache/cassandra/index/sai/utils/RowWithSource.java
View File

@ -213,6 +213,12 @@ public class RowWithSource implements Row
return row.hasDeletion(nowInSec);
}
@Override
public long minLocalDeletionTime()
{
return row.minLocalDeletionTime();
}
@Override
public SearchIterator<ColumnMetadata, ColumnData> searchIterator()
{

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

@ -21,6 +21,8 @@ import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
import net.nicoulaj.compilecommand.annotations.DontInline;
import accord.utils.Invariants;
/** Merges sorted input iterators which individually contain unique items. */
@ -307,9 +309,36 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
heap[currIdx] = heap[nextIdx];
currIdx = nextIdx;
}
// If the candidate has no children in the binary-heap section it is already in its final position, so we can
// skip the (rarely needed) sink below. nextIdx is the candidate's left child; anything >= size means there
// are no children. The single-child (nextIdx == size - 1) and two-children cases still have to go through
// sinkInBinaryHeap, which compares against the child(ren) and may swap or update the equalParent flag.
nextIdx = (currIdx * 2) - (sortedSectionSize - 1);
if (nextIdx >= size)
{
heap[currIdx] = candidate;
return;
}
// The candidate did not settle within the sorted section; sink it through the binary-heap section. This is
// rarely reached for typical narrow, lightly-overlapping merges, so it is split into its own method to keep
// the hot sorted-section path above small enough to be inlined into advance().
sinkInBinaryHeap(candidate, currIdx, size, sortedSectionSize);
}
/**
* Sink {@code candidate} down the binary-heap section of the queue (the part below the sorted section), pulling
* up the lighter child at each level, and place it in its final position.
*
* Split out of {@link #replaceAndSink} so that the common case, where the candidate settles within the sorted
* section, stays compact and inlinable.
*/
@DontInline
private void sinkInBinaryHeap(Candidate<In> candidate, int currIdx, final int size, final int sortedSectionSize)
{
// If size <= SORTED_SECTION_SIZE, nextIdx below will be no less than size,
// because currIdx == sortedSectionSize == size - 1 and nextIdx becomes
// (size - 1) * 2) - (size - 1 - 1) == size.
int nextIdx;
// Advance in the binary heap, pulling up the lighter element from the two at each level.
while ((nextIdx = (currIdx * 2) - (sortedSectionSize - 1)) + 1 < size)

51
test/unit/org/apache/cassandra/db/rows/RowsTest.java
View File

@ -282,6 +282,57 @@ public class RowsTest
}
}
private static Row liveRow(Clustering<?> c, long ts, ByteBuffer vVal)
{
return rowWithCell(c, ts, BufferCell.live(v, ts, vVal));
}
private static Row rowWithCell(Clustering<?> c, long ts, Cell<?> cell)
{
Row.Builder builder = createBuilder(c);
builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(ts));
builder.addCell(cell);
return builder.build();
}
private static Row mergeRows(Row... rows)
{
boolean hasComplex = false;
for (Row row : rows)
hasComplex |= row.hasComplex();
Row.Merger merger = new Row.Merger(rows.length, hasComplex);
for (int i = 0; i < rows.length; i++)
merger.add(i, rows[i]);
return merger.merge(DeletionTime.LIVE);
}
@Test
public void testMergerMinLocalDeletionTime()
{
long now = FBUtilities.nowInSeconds();
long ts = secondToTs(now);
// All inputs are free of deletions and expiring data, so the merged row must be too. This is the fast path in
// Row.Merger#merge that reuses Cell.MAX_DELETION_TIME instead of recomputing it by scanning the merged btree.
Row mergedLive = mergeRows(liveRow(c1, ts, BB1), liveRow(c1, ts + 1, BB2));
Assert.assertEquals(Cell.MAX_DELETION_TIME, mergedLive.minLocalDeletionTime());
Assert.assertFalse(mergedLive.hasDeletion(now));
// One input carries an expiring cell that wins reconciliation (higher timestamp): the merged row must keep
// its expiration time rather than being reported as deletion-free.
Cell<?> expiringCell = BufferCell.expiring(v, ts + 2, 100, now, BB3);
Row mergedExpiring = mergeRows(liveRow(c1, ts, BB1), rowWithCell(c1, ts + 2, expiringCell));
Assert.assertEquals(expiringCell.localDeletionTime(), mergedExpiring.minLocalDeletionTime());
Assert.assertFalse(mergedExpiring.hasDeletion(now));
Assert.assertTrue(mergedExpiring.hasDeletion(expiringCell.localDeletionTime()));
// Same for a tombstone cell winning reconciliation: the merged row must report a deletion.
Cell<?> tombstoneCell = BufferCell.tombstone(v, ts + 2, now);
Row mergedTombstone = mergeRows(liveRow(c1, ts, BB1), rowWithCell(c1, ts + 2, tombstoneCell));
Assert.assertEquals(Long.MIN_VALUE, mergedTombstone.minLocalDeletionTime());
Assert.assertTrue(mergedTombstone.hasDeletion(now));
}
@Test
public void diff()
{