[CASSANDRA-21524][trunk] Row merging logic optimizations

build.xml

@@ -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! -->

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)

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))

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;
@@ -44,9 +43,10 @@
 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 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)
@@ -771,6 +785,11 @@ public Row merge(DeletionTime activeDeletion)
                     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 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];
 
             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 Row merge(DeletionTime activeDeletion)
 
             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 Row[] mergedRows()
             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 @@ private static class ColumnDataReducer extends MergeIterator.Reducer<ColumnData,
 
             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,7 +881,7 @@ public void reduce(int idx, ColumnData data)
                 if (useColumnMetadata(data.column()))
                     column = data.column();
 
-                versions.add(data);
+                versions[versionsSize++] = data;
             }
 
             /**
@@ -880,20 +891,23 @@ 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()
             {
                 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 +918,9 @@ protected ColumnData getReduced()
                     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 +937,7 @@ protected ColumnData getReduced()
                         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 +951,12 @@ protected ColumnData getReduced()
             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;

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<Unfiltered, Unfiltered>
+        private class MergeReducer extends UnfilteredMergeIterator.Reducer<Unfiltered, Unfiltered>
         {
             private final MergeListener listener;
 

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

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<In> 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<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)