From 1e3d43b2ddb22ebcdd5a835a55b2a58e8977d9db Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 16:09:46 +0100 Subject: [PATCH] Apply performance optimizations for rows merging logic duplicate MergeIterator class to avoid megamorphic calls in hot path for Row/ComplexCell and UnfilteredRowIterators disable inlining for sinkInBinaryHeap, move typical case outside to improve hot path inlining replace List for versions with array fast path for same column metadata in cell merger avoid potential megamorphic cell method invocation in ALIVE DeletionTime minDeletionTime calculation optimization for merge logic do not add an empty iterator, merge 2 loops into 1 patch by Dmitry Konstantinov; reviewed by Francisco Guerrero for CASSANDRA-21524 --- CHANGES.txt | 1 + build.xml | 62 +++++++++++++- .../org/apache/cassandra/db/DeletionTime.java | 3 +- .../apache/cassandra/db/rows/BTreeRow.java | 5 ++ .../org/apache/cassandra/db/rows/Row.java | 84 +++++++++++-------- .../db/rows/UnfilteredRowIterators.java | 6 +- .../index/sai/utils/RowWithSource.java | 6 ++ .../apache/cassandra/utils/MergeIterator.java | 29 +++++++ .../apache/cassandra/db/rows/RowsTest.java | 51 +++++++++++ 9 files changed, 207 insertions(+), 40 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 09a7cd544f..5fe13105fd 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 6.0-alpha2 + * 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) diff --git a/build.xml b/build.xml index 4783737abe..93625fbc53 100644 --- a/build.xml +++ b/build.xml @@ -651,6 +651,64 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -710,7 +768,7 @@ - @@ -2165,7 +2223,7 @@ - + diff --git a/src/java/org/apache/cassandra/db/DeletionTime.java b/src/java/org/apache/cassandra/db/DeletionTime.java index 1d54a00fff..c220d1316b 100644 --- a/src/java/org/apache/cassandra/db/DeletionTime.java +++ b/src/java/org/apache/cassandra/db/DeletionTime.java @@ -178,7 +178,8 @@ public abstract class DeletionTime implements Comparable, 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) diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java index cdc2e973c3..82cf43c6a7 100644 --- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java +++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java @@ -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)) diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 9c281408ff..4137c42ca2 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -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, 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. + *

+ * 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, 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, 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, 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 merged = MergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer); + Iterator merged = RowMergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer); while (merged.hasNext()) { ColumnData data = merged.next(); @@ -819,8 +825,12 @@ public interface Row extends Unfiltered, Iterable, IMeasurableMemory try (BulkIterator 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, IMeasurableMemory return rows; } - private static class ColumnDataReducer extends MergeIterator.Reducer + private static class ColumnDataReducer extends RowMergeIterator.Reducer { private ColumnMetadata column; - private final List versions; + private final ColumnData[] versions; + private int versionsSize; private DeletionTime activeDeletion; @@ -854,7 +865,7 @@ public interface Row extends Unfiltered, Iterable, 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, 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, IMeasurableMemory if (column.isSimple()) { Cell merged = null; - for (int i=0, isize=versions.size(); i 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, IMeasurableMemory complexBuilder.newColumn(column); complexCells.clear(); DeletionTime complexDeletion = DeletionTime.LIVE; - for (int i=0, isize=versions.size(); i, IMeasurableMemory cellReducer.setActiveDeletion(activeDeletion); } - Iterator> cells = MergeIterator.get(complexCells, Cell.comparator, cellReducer); + Iterator> cells = ComplexCellMergeIterator.get(complexCells, Cell.comparator, cellReducer); while (cells.hasNext()) { Cell merged = cells.next(); @@ -937,11 +952,12 @@ public interface Row extends Unfiltered, Iterable, IMeasurableMemory protected void onKeyChange() { column = null; - versions.clear(); + Arrays.fill(versions, 0, versionsSize, null); + versionsSize = 0; } } - private static class CellReducer extends MergeIterator.Reducer, Cell> + private static class CellReducer extends ComplexCellMergeIterator.Reducer, Cell> { private DeletionTime activeDeletion; private Cell merged; diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java index bc4bfd15ba..a306841892 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java @@ -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 + private class MergeReducer extends UnfilteredMergeIterator.Reducer { private final MergeListener listener; diff --git a/src/java/org/apache/cassandra/index/sai/utils/RowWithSource.java b/src/java/org/apache/cassandra/index/sai/utils/RowWithSource.java index 33f5f07deb..6b13a92236 100644 --- a/src/java/org/apache/cassandra/index/sai/utils/RowWithSource.java +++ b/src/java/org/apache/cassandra/index/sai/utils/RowWithSource.java @@ -213,6 +213,12 @@ public class RowWithSource implements Row return row.hasDeletion(nowInSec); } + @Override + public long minLocalDeletionTime() + { + return row.minLocalDeletionTime(); + } + @Override public SearchIterator searchIterator() { diff --git a/src/java/org/apache/cassandra/utils/MergeIterator.java b/src/java/org/apache/cassandra/utils/MergeIterator.java index b156e20a0a..dff1677d61 100644 --- a/src/java/org/apache/cassandra/utils/MergeIterator.java +++ b/src/java/org/apache/cassandra/utils/MergeIterator.java @@ -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 extends AbstractIterator 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 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) diff --git a/test/unit/org/apache/cassandra/db/rows/RowsTest.java b/test/unit/org/apache/cassandra/db/rows/RowsTest.java index 06bf701a2f..5412f82ee0 100644 --- a/test/unit/org/apache/cassandra/db/rows/RowsTest.java +++ b/test/unit/org/apache/cassandra/db/rows/RowsTest.java @@ -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() {