Merge branch 'cassandra-6.0' into trunk

* cassandra-6.0:
  Apply performance optimizations for rows merging logic

CHANGES.txt

@@ -6,6 +6,9 @@
  * 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)
  * Reduce allocations in DefaultQueryOptions (CASSANDRA-21467)
  * Allow unreserved keywords as user and identity names in USER and IDENTITY statements (CASSANDRA-21510)
@@ -30,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)
@@ -8667,4 +8670,3 @@ Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/
  * Added FlushPeriodInMinutes configuration parameter to force
    flushing of infrequently-updated ColumnFamilies
 
-

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

src/java/org/apache/cassandra/db/DeletionTime.java

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

src/java/org/apache/cassandra/db/marshal/AbstractTimeUUIDType.java

@@ -39,7 +39,7 @@ public abstract class AbstractTimeUUIDType<T> extends TemporalType<T>
 {
     AbstractTimeUUIDType()
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, 16);
     } // singleton
 
     @Override
@@ -193,12 +193,6 @@ public abstract class AbstractTimeUUIDType<T> extends TemporalType<T>
         return super.decomposeUntyped(value);
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 16;
-    }
-
     @Override
     public long toTimeInMillis(ByteBuffer value)
     {

src/java/org/apache/cassandra/db/marshal/AbstractType.java

@@ -90,11 +90,18 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     public final ComparisonType comparisonType;
     public final boolean isByteOrderComparable;
     public final ValueComparators comparatorSet;
+    private final int valueLengthIfFixed;
 
     protected AbstractType(ComparisonType comparisonType)
+    {
+        this(comparisonType, VARIABLE_LENGTH);
+    }
+
+    protected AbstractType(ComparisonType comparisonType, int valueLengthIfFixed)
     {
         this.comparisonType = comparisonType;
         this.isByteOrderComparable = comparisonType == ComparisonType.BYTE_ORDER;
+        this.valueLengthIfFixed = valueLengthIfFixed;
         reverseComparator = (o1, o2) -> AbstractType.this.compare(o2, o1);
         try
         {
@@ -384,12 +391,12 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     /**
      * Similar to {@link #isValueCompatibleWith(AbstractType)}, but takes into account {@link Cell} encoding.
      * In particular, this method doesn't consider two types serialization compatible if one of them has fixed
-     * length (overrides {@link #valueLengthIfFixed()}, and the other one doesn't.
+     * length, and the other one doesn't.
      */
     public boolean isSerializationCompatibleWith(AbstractType<?> previous)
     {
         return isValueCompatibleWith(previous)
-               && valueLengthIfFixed() == previous.valueLengthIfFixed()
+               && valueLengthIfFixed == previous.valueLengthIfFixed
                && isMultiCell() == previous.isMultiCell();
     }
 
@@ -498,7 +505,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
      */
     public int valueLengthIfFixed()
     {
-        return VARIABLE_LENGTH;
+        return valueLengthIfFixed;
     }
 
     /**
@@ -508,7 +515,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
      */
     public final boolean isValueLengthFixed()
     {
-        return valueLengthIfFixed() != VARIABLE_LENGTH;
+        return valueLengthIfFixed != VARIABLE_LENGTH;
     }
 
     /**
@@ -570,7 +577,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     public  <V> void writeValue(V value, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException
     {
         assert !isNull(value, accessor) : "bytes should not be null for type " + this;
-        int expectedValueLength = valueLengthIfFixed();
+        int expectedValueLength = valueLengthIfFixed;
         if (expectedValueLength >= 0)
         {
             int actualValueLength = accessor.size(value);
@@ -589,7 +596,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     public  <V> void writeValue(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException
     {
         assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
-        int expectedValueLength = valueLengthIfFixed();
+        int expectedValueLength = valueLengthIfFixed;
         if (expectedValueLength >= 0)
         {
             int actualValueLength = valueHolder.size(i);
@@ -613,7 +620,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     public <V> long writtenLength(V value, ValueAccessor<V> accessor)
     {
         assert !accessor.isEmpty(value) : "bytes should not be empty for type " + this;
-        return valueLengthIfFixed() >= 0
+        return valueLengthIfFixed >= 0
                ? accessor.size(value) // if the size is wrong, this will be detected in writeValue
                : accessor.sizeWithVIntLength(value);
     }
@@ -621,7 +628,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
     public <V> long writtenLength(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor)
     {
         assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
-        return valueLengthIfFixed() >= 0
+        return valueLengthIfFixed >= 0
                ? valueHolder.size(i) // if the size is wrong, this will be detected in writeValue
                : accessor.sizeWithVIntLength(valueHolder, i);
     }
@@ -643,7 +650,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
 
     public <V> V read(ValueAccessor<V> accessor, DataInputPlus in, int maxValueSize) throws IOException
     {
-        int length = valueLengthIfFixed();
+        int length = valueLengthIfFixed;
 
         if (length >= 0)
             return accessor.read(in, length);
@@ -664,7 +671,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
 
     public void skipValue(DataInputPlus in) throws IOException
     {
-        int length = valueLengthIfFixed();
+        int length = valueLengthIfFixed;
         if (length >= 0)
             in.skipBytesFully(length);
         else

src/java/org/apache/cassandra/db/marshal/BooleanType.java

@@ -36,7 +36,7 @@ public class BooleanType extends AbstractType<Boolean>
     private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance);
     private static final ByteBuffer MASKED_VALUE = instance.decompose(false);
 
-    BooleanType() {super(ComparisonType.CUSTOM);} // singleton
+    BooleanType() {super(ComparisonType.CUSTOM, 1);} // singleton
 
     @Override
     public boolean allowsEmpty()
@@ -127,12 +127,6 @@ public class BooleanType extends AbstractType<Boolean>
         return ARGUMENT_DESERIALIZER;
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 1;
-    }
-
     @Override
     public ByteBuffer getMaskedValue()
     {

src/java/org/apache/cassandra/db/marshal/DateType.java

@@ -49,7 +49,7 @@ public class DateType extends AbstractType<Date>
     private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance);
     private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
 
-    DateType() {super(ComparisonType.BYTE_ORDER);} // singleton
+    DateType() {super(ComparisonType.BYTE_ORDER, 8);} // singleton
 
     public boolean isEmptyValueMeaningless()
     {
@@ -143,12 +143,6 @@ public class DateType extends AbstractType<Date>
         return ARGUMENT_DESERIALIZER;
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 8;
-    }
-
     @Override
     public ByteBuffer getMaskedValue()
     {

src/java/org/apache/cassandra/db/marshal/DoubleType.java

@@ -40,7 +40,7 @@ public class DoubleType extends NumberType<Double>
 
     private static final ByteBuffer MASKED_VALUE = instance.decompose(0d);
 
-    DoubleType() {super(ComparisonType.CUSTOM);} // singleton
+    DoubleType() {super(ComparisonType.CUSTOM, 8);} // singleton
 
     @Override
     public boolean allowsEmpty()
@@ -145,12 +145,6 @@ public class DoubleType extends NumberType<Double>
         };
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 8;
-    }
-
     @Override
     public ByteBuffer add(Number left, Number right)
     {

src/java/org/apache/cassandra/db/marshal/EmptyType.java

@@ -71,7 +71,7 @@ public class EmptyType extends AbstractType<Void>
 
     public static final EmptyType instance = new EmptyType();
 
-    private EmptyType() {super(ComparisonType.CUSTOM);} // singleton
+    private EmptyType() {super(ComparisonType.CUSTOM, 0);} // singleton
 
     @Override
     public <V> ByteSource asComparableBytes(ValueAccessor<V> accessor, V data, ByteComparable.Version version)
@@ -137,12 +137,6 @@ public class EmptyType extends AbstractType<Void>
         throw new UnsupportedOperationException();
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 0;
-    }
-
     @Override
     public <V> long writtenLength(V value, ValueAccessor<V> accessor)
     {

src/java/org/apache/cassandra/db/marshal/FloatType.java

@@ -41,7 +41,7 @@ public class FloatType extends NumberType<Float>
 
     private static final ByteBuffer MASKED_VALUE = instance.decompose(0f);
 
-    FloatType() {super(ComparisonType.CUSTOM);} // singleton
+    FloatType() {super(ComparisonType.CUSTOM, 4);} // singleton
 
     @Override
     public boolean allowsEmpty()
@@ -146,12 +146,6 @@ public class FloatType extends NumberType<Float>
         };
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 4;
-    }
-
     @Override
     public ByteBuffer add(Number left, Number right)
     {

src/java/org/apache/cassandra/db/marshal/Int32Type.java

@@ -43,7 +43,7 @@ public class Int32Type extends NumberType<Integer>
 
     Int32Type()
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, 4);
     } // singleton
 
     @Override
@@ -152,12 +152,6 @@ public class Int32Type extends NumberType<Integer>
         };
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 4;
-    }
-
     @Override
     public ByteBuffer add(Number left, Number right)
     {

src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java

@@ -44,7 +44,7 @@ public class LexicalUUIDType extends AbstractType<UUID>
 
     LexicalUUIDType()
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, 16);
     } // singleton
 
     @Override
@@ -148,12 +148,6 @@ public class LexicalUUIDType extends AbstractType<UUID>
         return ARGUMENT_DESERIALIZER;
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 16;
-    }
-
     @Override
     public ByteBuffer getMaskedValue()
     {

src/java/org/apache/cassandra/db/marshal/LongType.java

@@ -41,7 +41,7 @@ public class LongType extends NumberType<Long>
 
     private static final ByteBuffer MASKED_VALUE = instance.decompose(0L);
 
-    LongType() {super(ComparisonType.CUSTOM);} // singleton
+    LongType() {super(ComparisonType.CUSTOM, 8);} // singleton
 
     @Override
     public boolean allowsEmpty()
@@ -170,12 +170,6 @@ public class LongType extends NumberType<Long>
         };
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 8;
-    }
-
     @Override
     public ByteBuffer add(Number left, Number right)
     {

src/java/org/apache/cassandra/db/marshal/MultiElementType.java

@@ -38,6 +38,11 @@ public abstract class MultiElementType<T> extends AbstractType<T>
         super(comparisonType);
     }
 
+    protected MultiElementType(ComparisonType comparisonType, int valueLengthIfFixed)
+    {
+        super(comparisonType, valueLengthIfFixed);
+    }
+
     /**
      * Returns the serialized representation of the value composed of the specified elements.
      *
@@ -133,4 +138,3 @@ public abstract class MultiElementType<T> extends AbstractType<T>
         throw new UnsupportedOperationException(this + " does not support retrieving elements by key or index");
     }
 }
-

src/java/org/apache/cassandra/db/marshal/NumberType.java

@@ -34,6 +34,11 @@ public abstract class NumberType<T extends Number> extends AbstractType<T>
         super(comparisonType);
     }
 
+    protected NumberType(ComparisonType comparisonType, int valueLengthIfFixed)
+    {
+        super(comparisonType, valueLengthIfFixed);
+    }
+
     /**
      * Checks if this type support floating point numbers.
      * @return {@code true} if this type support floating point numbers, {@code false} otherwise.

src/java/org/apache/cassandra/db/marshal/ReversedType.java

@@ -57,7 +57,7 @@ public class ReversedType<T> extends AbstractType<T>
 
     private ReversedType(AbstractType<T> baseType)
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, baseType.valueLengthIfFixed());
         this.baseType = baseType;
     }
 
@@ -181,12 +181,6 @@ public class ReversedType<T> extends AbstractType<T>
         return getInstance(baseType.withUpdatedUserType(udt));
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return baseType.valueLengthIfFixed();
-    }
-
     @Override
     public boolean isReversed()
     {

src/java/org/apache/cassandra/db/marshal/TemporalType.java

@@ -37,6 +37,11 @@ public abstract class TemporalType<T> extends AbstractType<T>
         super(comparisonType);
     }
 
+    protected TemporalType(ComparisonType comparisonType, int valueLengthIfFixed)
+    {
+        super(comparisonType, valueLengthIfFixed);
+    }
+
     /**
      * Returns the current temporal value.
      * @return the current temporal value.

src/java/org/apache/cassandra/db/marshal/TimestampType.java

@@ -53,7 +53,7 @@ public class TimestampType extends TemporalType<Date>
 
     private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
 
-    private TimestampType() {super(ComparisonType.CUSTOM);} // singleton
+    private TimestampType() {super(ComparisonType.CUSTOM, 8);} // singleton
 
     @Override
     public boolean allowsEmpty()
@@ -166,12 +166,6 @@ public class TimestampType extends TemporalType<Date>
         return TimestampSerializer.instance;
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 8;
-    }
-
     @Override
     protected void validateDuration(Duration duration)
     {

src/java/org/apache/cassandra/db/marshal/UUIDType.java

@@ -56,7 +56,7 @@ public class UUIDType extends AbstractType<UUID>
 
     UUIDType()
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, 16);
     }
 
     @Override
@@ -252,12 +252,6 @@ public class UUIDType extends AbstractType<UUID>
         return (uuid.get(6) & 0xf0) >> 4;
     }
 
-    @Override
-    public int valueLengthIfFixed()
-    {
-        return 16;
-    }
-
     @Override
     public ByteBuffer getMaskedValue()
     {

src/java/org/apache/cassandra/db/marshal/VectorType.java

@@ -82,20 +82,16 @@ public final class VectorType<T> extends MultiElementType<List<T>>
     public final AbstractType<T> elementType;
     public final int dimension;
     private final TypeSerializer<T> elementSerializer;
-    private final int valueLengthIfFixed;
     private final VectorSerializer serializer;
 
     private VectorType(AbstractType<T> elementType, int dimension)
     {
-        super(ComparisonType.CUSTOM);
+        super(ComparisonType.CUSTOM, valueLengthIfFixed(elementType, dimension));
         if (dimension <= 0)
             throw new InvalidRequestException(String.format("vectors may only have positive dimensions; given %d", dimension));
         this.elementType = elementType;
         this.dimension = dimension;
         this.elementSerializer = elementType.getSerializer();
-        this.valueLengthIfFixed = elementType.isValueLengthFixed() ?
-                                  elementType.valueLengthIfFixed() * dimension :
-                                  super.valueLengthIfFixed();
         this.serializer = elementType.isValueLengthFixed() ?
                           new FixedLengthSerializer() :
                           new VariableLengthSerializer();
@@ -126,10 +122,10 @@ public final class VectorType<T> extends MultiElementType<List<T>>
         return getSerializer().compareCustom(left, accessorL, right, accessorR);
     }
 
-    @Override
-    public int valueLengthIfFixed()
+    private static int valueLengthIfFixed(AbstractType<?> elementType, int dimension)
     {
-        return valueLengthIfFixed;
+        int elementLength = elementType.valueLengthIfFixed();
+        return elementLength >= 0 ? elementLength * dimension : elementLength;
     }
 
     @Override

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

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

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<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 class RowWithSource implements Row
         return row.hasDeletion(nowInSec);
     }
 
+    @Override
+    public long minLocalDeletionTime()
+    {
+        return row.minLocalDeletionTime();
+    }
+
     @Override
     public SearchIterator<ColumnMetadata, ColumnData> searchIterator()
     {

src/java/org/apache/cassandra/tcm/sequences/SingleNodeSequences.java

@@ -92,6 +92,7 @@ public interface SingleNodeSequences
         else if (InProgressSequences.isLeave(inProgress))
         {
             logger.info("Resuming decommission @ {} (current epoch = {}): {}", inProgress.latestModification, metadata.epoch, inProgress.status());
+            StorageService.instance.clearTransientMode();
         }
         else
         {

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

test/distributed/org/apache/cassandra/distributed/test/DecommissionTest.java

@@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.test;
 import java.io.IOException;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import net.bytebuddy.ByteBuddy;
 import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
@@ -141,6 +142,7 @@ public class DecommissionTest extends TestBaseImpl
 
     public static class BB
     {
+        static final AtomicBoolean injected = new AtomicBoolean(false);
         public static void install(ClassLoader classLoader, Integer num)
         {
             if (num == 2)
@@ -157,7 +159,7 @@ public class DecommissionTest extends TestBaseImpl
         public static void execute(NodeId leaving, PlacementDeltas startLeave, PlacementDeltas midLeave, PlacementDeltas finishLeave,
                                    @SuperCall Callable<?> zuper) throws ExecutionException, InterruptedException
         {
-            if (!StorageService.instance.isDecommissionFailed())
+            if (injected.compareAndSet(false, true))
                 throw new ExecutionException(new RuntimeException("simulated error in prepareUnbootstrapStreaming"));
 
             try

test/distributed/org/apache/cassandra/distributed/test/ring/DecommissionTest.java

@@ -53,8 +53,10 @@ import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
+import org.apache.cassandra.tcm.Epoch;
 import org.apache.cassandra.tcm.membership.NodeId;
 import org.apache.cassandra.tcm.membership.NodeVersion;
+import org.apache.cassandra.tcm.transformations.PrepareLeave;
 import org.apache.cassandra.tcm.transformations.Startup;
 import org.apache.cassandra.utils.CassandraVersion;
 import org.apache.cassandra.utils.FBUtilities;
@@ -62,6 +64,8 @@ import org.apache.cassandra.utils.FBUtilities;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
 import static org.apache.cassandra.distributed.api.Feature.NETWORK;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeCommit;
+import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
 import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
 import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
 import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate;
@@ -84,6 +88,43 @@ public class DecommissionTest extends TestBaseImpl
         }
     }
 
+    @Test
+    public void testOperationModeOnDecomResume() throws Exception
+    {
+        // Node 2's first decomission attempt fails mid-stream (injected by BB), leaving it in
+        // DECOMISSION_FAILED. On resume, operationMode() must transition back to LEAVING before
+        // MID_LEAVE is committed. We pause the CMS just before that commit to assert the mode
+        // in the window where streaming has finished but the epoch has not yet advanced.
+        try (Cluster cluster = builder().withNodes(3)
+                                        .withConfig(config -> config.with(NETWORK, GOSSIP))
+                                        .withInstanceInitializer(BB::install)
+                                        .start())
+        {
+            populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
+
+            IInvokableInstance cmsNode = cluster.get(1);
+            IInvokableInstance leavingNode = cluster.get(2);
+
+            // --force required: cie_internal keyspace has RF=3, decommission would fail replication check otherwise
+            leavingNode.nodetoolResult("decommission", "--force").asserts().failure();
+            leavingNode.runOnInstance(() -> assertEquals(StorageService.Mode.DECOMMISSION_FAILED, StorageService.instance.operationMode()));
+
+            Callable<Epoch> midLeavePaused = pauseBeforeCommit(cmsNode, e -> e instanceof PrepareLeave.MidLeave);
+
+            Thread resumeThread = new Thread(() -> leavingNode.nodetoolResult("decommission").asserts().success());
+            resumeThread.start();
+            midLeavePaused.call();
+
+            leavingNode.runOnInstance(() ->
+                assertEquals("operationMode during resumed decommission streaming should be LEAVING, not DECOMMISSION_FAILED",
+                             StorageService.Mode.LEAVING,
+                             StorageService.instance.operationMode()));
+
+            unpauseCommits(cmsNode);
+            resumeThread.join(TimeUnit.MINUTES.toMillis(2));
+        }
+    }
+
     @Test
     public void testAddressReuseAfterDecommission() throws IOException, ExecutionException, InterruptedException
     {

test/microbench/org/apache/cassandra/test/microbench/AbstractTypeSerializationBench.java

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.test.microbench;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.BooleanType;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.LongType;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.UUIDType;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+/**
+ * Measures value serialization when one call site sees the fixed- and variable-width types commonly used by a table.
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 5, time = 1)
+@Fork(value = 1, jvmArgsAppend = { "-Xmx1G", "-Djmh.executor=CUSTOM", "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor" })
+@Threads(1)
+@State(Scope.Thread)
+public class AbstractTypeSerializationBench
+{
+    private final AbstractType<?>[] types = { Int32Type.instance, LongType.instance, BooleanType.instance, UUIDType.instance,
+                                              UTF8Type.instance, Int32Type.instance, LongType.instance, UTF8Type.instance };
+    private final ByteBuffer[] values = { ByteBufferUtil.bytes(42), ByteBufferUtil.bytes(42L), ByteBuffer.wrap(new byte[] { 1 }),
+                                          ByteBuffer.wrap(new byte[16]), ByteBufferUtil.bytes("cassandra"), ByteBufferUtil.bytes(42),
+                                          ByteBufferUtil.bytes(42L), ByteBufferUtil.bytes("cassandra") };
+    private final ByteBuffer[] serializedValues = new ByteBuffer[types.length];
+    private final DataOutputBuffer out = new DataOutputBuffer();
+    private int index;
+
+    @Setup
+    public void setup() throws IOException
+    {
+        for (int i = 0; i < types.length; i++)
+        {
+            out.clear();
+            types[i].writeValue(values[i], out);
+            serializedValues[i] = out.asNewBuffer();
+        }
+    }
+
+    @Benchmark
+    public int writeValue() throws IOException
+    {
+        int i = index++ & (types.length - 1);
+        out.clear();
+        types[i].writeValue(values[i], out);
+        return out.getLength();
+    }
+
+    @Benchmark
+    public ByteBuffer readValue() throws IOException
+    {
+        int i = index++ & (types.length - 1);
+        return types[i].readBuffer(new DataInputBuffer(serializedValues[i], true));
+    }
+}

test/unit/org/apache/cassandra/db/marshal/AbstractTypeTest.java

@@ -739,6 +739,12 @@ public class AbstractTypeTest
             }});
     }
 
+    @Test
+    public void valueLengthIfFixedIsNotFinal() throws NoSuchMethodException
+    {
+        assertThat(Modifier.isFinal(AbstractType.class.getMethod("valueLengthIfFixed").getModifiers())).isFalse();
+    }
+
     @Test
     @SuppressWarnings({"rawtypes", "unchecked"})
     public void serde()

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