From ce930acf8a63998b1c20f2f78afbac2386d28251 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 16:09:46 +0100 Subject: [PATCH 1/9] do not add an empty iterator, merge 2 loops into 1 --- src/java/org/apache/cassandra/db/rows/Row.java | 18 ++++-------------- 1 file changed, 4 insertions(+), 14 deletions(-) diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 9c281408ffbd..41ee08acf061 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 @@ 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; @@ -762,6 +761,7 @@ public Row merge(DeletionTime activeDeletion) LivenessInfo rowInfo = LivenessInfo.EMPTY; Deletion rowDeletion = Deletion.LIVE; + int columnsCountEstimation = 0; for (Row row : rows) { if (row == null) @@ -771,6 +771,9 @@ public Row merge(DeletionTime activeDeletion) rowInfo = row.primaryKeyLivenessInfo(); if (row.deletion().supersedes(rowDeletion)) rowDeletion = row.deletion(); + + columnDataIterators.add(row.iterator()); + columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount()); } if (rowDeletion.isShadowedBy(rowInfo)) @@ -784,19 +787,6 @@ public Row merge(DeletionTime activeDeletion) 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]; From 487c4268dfca8452a512107937b3c80bd294c4d8 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 17:17:06 +0100 Subject: [PATCH 2/9] duplicate MergeIterator class to avoid megamorphic calls in hot path --- build.xml | 58 ++++++++++++++++++- .../org/apache/cassandra/db/rows/Row.java | 5 +- 2 files changed, 60 insertions(+), 3 deletions(-) diff --git a/build.xml b/build.xml index 7240153109a9..0f091ca07b96 100644 --- a/build.xml +++ b/build.xml @@ -651,6 +651,62 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -710,7 +766,7 @@ - diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 41ee08acf061..848d2b3f4072 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -47,6 +47,7 @@ import org.apache.cassandra.utils.MergeIterator; import org.apache.cassandra.utils.ObjectSizes; import org.apache.cassandra.utils.SearchIterator; +import org.apache.cassandra.utils.RowMergeIterator; import org.apache.cassandra.utils.btree.BTree; import org.apache.cassandra.utils.btree.UpdateFunction; import org.apache.cassandra.utils.memory.Cloner; @@ -792,7 +793,7 @@ public Row merge(DeletionTime activeDeletion) 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(); @@ -831,7 +832,7 @@ public Row[] mergedRows() return rows; } - private static class ColumnDataReducer extends MergeIterator.Reducer + private static class ColumnDataReducer extends RowMergeIterator.Reducer { private ColumnMetadata column; private final List versions; From 5c5344a5d9fc46ae86a833d2ee997497868d7330 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 18:12:26 +0100 Subject: [PATCH 3/9] minDeletionTime calculation optimization for merge logic --- .../apache/cassandra/db/rows/BTreeRow.java | 5 ++ .../org/apache/cassandra/db/rows/Row.java | 23 ++++++++- .../index/sai/utils/RowWithSource.java | 6 +++ .../apache/cassandra/db/rows/RowsTest.java | 51 +++++++++++++++++++ 4 files changed, 83 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java index b8df8450c136..3ed0c9bdf239 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 boolean hasDeletion(long nowInSec) 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 848d2b3f4072..7e8fade84596 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -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. * @@ -763,6 +772,10 @@ public Row merge(DeletionTime activeDeletion) 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) @@ -773,6 +786,8 @@ public Row merge(DeletionTime activeDeletion) if (row.deletion().supersedes(rowDeletion)) rowDeletion = row.deletion(); + minDeletionTime = Math.min(minDeletionTime, row.minLocalDeletionTime()); + columnDataIterators.add(row.iterator()); columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount()); } @@ -810,8 +825,12 @@ public Row merge(DeletionTime activeDeletion) 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); } } 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 33f5f07deb08..6b13a922368d 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 boolean hasDeletion(long nowInSec) return row.hasDeletion(nowInSec); } + @Override + public long minLocalDeletionTime() + { + return row.minLocalDeletionTime(); + } + @Override public SearchIterator searchIterator() { diff --git a/test/unit/org/apache/cassandra/db/rows/RowsTest.java b/test/unit/org/apache/cassandra/db/rows/RowsTest.java index 06bf701a2f76..5412f82ee0cf 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 static void addExpectedCells(Set>> dst, Cell merged } } + 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() { From 6ecc27485b5b2887e6f39b85d2d64ec6b823062e Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 19:33:39 +0100 Subject: [PATCH 4/9] avoid cell invocation in ALIVE DeletionTime --- src/java/org/apache/cassandra/db/DeletionTime.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/java/org/apache/cassandra/db/DeletionTime.java b/src/java/org/apache/cassandra/db/DeletionTime.java index 1d54a00fff3f..c220d1316b05 100644 --- a/src/java/org/apache/cassandra/db/DeletionTime.java +++ b/src/java/org/apache/cassandra/db/DeletionTime.java @@ -178,7 +178,8 @@ public boolean deletes(LivenessInfo info) 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) From da9066584c782d132cee28fb2a60d2086432b460 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 21:51:07 +0100 Subject: [PATCH 5/9] fast path for same column metadata in merger --- src/java/org/apache/cassandra/db/rows/Row.java | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 7e8fade84596..f3234a6cacf4 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -890,10 +890,13 @@ public void reduce(int idx, ColumnData data) */ 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() From 0b5fb2469e961053cf0247ddef1a8c020fc8dc7c Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Tue, 14 Jul 2026 22:10:28 +0100 Subject: [PATCH 6/9] replace List for versions with array --- src/java/org/apache/cassandra/db/rows/Row.java | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index f3234a6cacf4..1650a3fa86f8 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -854,7 +854,8 @@ public Row[] mergedRows() private static class ColumnDataReducer extends RowMergeIterator.Reducer { private ColumnMetadata column; - private final List versions; + private final ColumnData[] versions; + private int versionsSize; private DeletionTime activeDeletion; @@ -864,7 +865,7 @@ private static class ColumnDataReducer extends RowMergeIterator.Reducer(size); + this.versions = new ColumnData[size]; this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null; this.complexCells = hasComplex ? new ArrayList<>(size) : null; this.cellReducer = new CellReducer(); @@ -880,7 +881,7 @@ public void reduce(int idx, ColumnData data) if (useColumnMetadata(data.column())) column = data.column(); - versions.add(data); + versions[versionsSize++] = data; } /** @@ -904,9 +905,9 @@ protected ColumnData getReduced() 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); } @@ -917,9 +918,9 @@ protected ColumnData getReduced() complexBuilder.newColumn(column); complexCells.clear(); DeletionTime complexDeletion = DeletionTime.LIVE; - for (int i=0, isize=versions.size(); i Date: Wed, 15 Jul 2026 00:24:24 +0100 Subject: [PATCH 7/9] disable inlining for sinkInBinaryHeap, move typical case outside --- .../apache/cassandra/utils/MergeIterator.java | 29 +++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/src/java/org/apache/cassandra/utils/MergeIterator.java b/src/java/org/apache/cassandra/utils/MergeIterator.java index b156e20a0a6f..dff1677d6173 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.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 @@ private void replaceAndSink(Candidate candidate, int currIdx) 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) From 74bd6ea759c9e4a3debc9d5a99b6ccba6946fe4b Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Thu, 16 Jul 2026 09:17:14 +0100 Subject: [PATCH 8/9] add copy to avoid megamorphic calls for UnfilteredRowIterators too fix an import order --- build.xml | 1 + src/java/org/apache/cassandra/db/rows/Row.java | 2 +- .../apache/cassandra/db/rows/UnfilteredRowIterators.java | 6 +++--- 3 files changed, 5 insertions(+), 4 deletions(-) diff --git a/build.xml b/build.xml index 0f091ca07b96..8771f1a15e01 100644 --- a/build.xml +++ b/build.xml @@ -705,6 +705,7 @@ + diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 1650a3fa86f8..c49a3c9b0b17 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -46,8 +46,8 @@ 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.RowMergeIterator; +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; diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIterators.java index bc4bfd15ba81..a306841892a0 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.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 @@ private UnfilteredRowMergeIterator(TableMetadata metadata, 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 void close() listener.close(); } - private class MergeReducer extends MergeIterator.Reducer + private class MergeReducer extends UnfilteredMergeIterator.Reducer { private final MergeListener listener; From a0741082d8f21b8cd5f99d30f95c3362136ec326 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Thu, 16 Jul 2026 20:46:22 +0100 Subject: [PATCH 9/9] add copy to avoid megamorphic calls for ComplexCell merge too --- build.xml | 1 + src/java/org/apache/cassandra/db/rows/Row.java | 6 +++--- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/build.xml b/build.xml index 8771f1a15e01..23454bdd819d 100644 --- a/build.xml +++ b/build.xml @@ -706,6 +706,7 @@ description="Copy selected classes under new names (e.g. MergeIterator -> RowMergeIterator) into src/gen-java to avoid megamorphic calls"> + diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index c49a3c9b0b17..2d7fbcebf52b 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -43,8 +43,8 @@ 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; @@ -937,7 +937,7 @@ protected ColumnData getReduced() 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(); @@ -956,7 +956,7 @@ protected void onKeyChange() } } - private static class CellReducer extends MergeIterator.Reducer, Cell> + private static class CellReducer extends ComplexCellMergeIterator.Reducer, Cell> { private DeletionTime activeDeletion; private Cell merged;