, 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()
{