Merge 591740370d into 10557d7ffe

Patch by C. Scott Andreas; reviewed by Sam Tunnicliffe for CASSANDRA-21501

CHANGES.txt

@@ -6,6 +6,10 @@
  * 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)
  * Reduce allocations in DefaultQueryOptions (CASSANDRA-21467)
@@ -29,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)
@@ -8666,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"/>

pylib/cqlshlib/cqlshmain.py

@@ -1035,7 +1035,7 @@ class Shell(cmd.Cmd):
 
         # print row data
         for row in formatted_values:
-            line = ' | '.join(col.rjust(w, color=self.color) for (col, w) in zip(row, widths))
+            line = ' | '.join(col.just(w, color=self.color) for (col, w) in zip(row, widths))
             self.writeresult(' ' + line)
 
         if tty:

pylib/cqlshlib/displaying.py

@@ -14,6 +14,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from enum import Enum
 from collections import defaultdict
 
 RED = '\033[0;1;31m'
@@ -27,12 +28,17 @@ DARK_MAGENTA = '\033[0;35m'
 ANSI_RESET = '\033[0m'
 
 
-def colorme(bval, colormap, colorkey):
+class Alignment(Enum):
+    LEFT = 1
+    RIGHT = 2
+
+
+def colorme(bval, colormap, colorkey, alignment=None):
     if colormap is NO_COLOR_MAP:
         return bval
     if colormap is None:
         colormap = DEFAULT_VALUE_COLORS
-    return FormattedValue(bval, colormap[colorkey] + bval + colormap['reset'])
+    return FormattedValue(bval, colormap[colorkey] + bval + colormap['reset'], alignment=alignment)
 
 
 def get_str(val):
@@ -43,7 +49,7 @@ def get_str(val):
 
 class FormattedValue:
 
-    def __init__(self, strval, coloredval=None, displaywidth=None):
+    def __init__(self, strval, coloredval=None, displaywidth=None, alignment=Alignment.RIGHT):
         self.strval = strval
         if coloredval is None:
             coloredval = strval
@@ -53,6 +59,7 @@ class FormattedValue:
         # displaywidth is useful for display of special unicode characters
         # with
         self.displaywidth = displaywidth
+        self.alignment = alignment
 
     def __len__(self):
         return len(self.strval)
@@ -63,6 +70,11 @@ class FormattedValue:
         else:
             return ''
 
+    def just(self, width, fill=' ', color=False):
+        if self.alignment == Alignment.LEFT:
+            return self.ljust(width, fill, color)
+        return self.rjust(width, fill, color)
+
     def ljust(self, width, fill=' ', color=False):
         """
         Similar to self.strval.ljust(width), but takes expected terminal

pylib/cqlshlib/formatting.py

@@ -26,7 +26,7 @@ from collections import defaultdict
 
 from cassandra.cqltypes import EMPTY
 from cassandra.util import datetime_from_timestamp
-from .displaying import colorme, get_str, FormattedValue, DEFAULT_VALUE_COLORS, NO_COLOR_MAP
+from .displaying import colorme, get_str, FormattedValue, DEFAULT_VALUE_COLORS, NO_COLOR_MAP, Alignment
 
 UNICODE_CONTROLCHARS_RE = re.compile(r'[\x00-\x1f\x7f-\xa0]')
 CONTROLCHARS_RE = re.compile(r'[\x00-\x1f\x7f-\xff]')
@@ -65,7 +65,7 @@ def format_by_type(val, cqltype, encoding, colormap=None, addcolor=False,
     if nullval is None:
         nullval = default_null_placeholder
     if val is None:
-        return colorme(nullval, colormap, 'error')
+        return colorme(nullval, colormap, 'error', alignment=Alignment.LEFT)
     if addcolor is False:
         colormap = empty_colormap
     elif colormap is None:
@@ -80,7 +80,7 @@ def format_by_type(val, cqltype, encoding, colormap=None, addcolor=False,
                         boolean_styles=boolean_styles)
 
 
-def color_text(bval, colormap, displaywidth=None):
+def color_text(bval, colormap, displaywidth=None, alignment=None):
     # note that here, we render natural backslashes as just backslashes,
     # in the same color as surrounding text, when using color. When not
     # using color, we need to double up the backslashes, so it's not
@@ -95,7 +95,7 @@ def color_text(bval, colormap, displaywidth=None):
     coloredval = colormap['text'] + bits_to_turn_red_re.sub(tbr, bval) + colormap['reset']
     if colormap['text']:
         displaywidth -= bval.count(r'\\')
-    return FormattedValue(bval, coloredval, displaywidth)
+    return FormattedValue(bval, coloredval, displaywidth, alignment)
 
 
 DEFAULT_NANOTIME_FORMAT = '%H:%M:%S.%N'
@@ -209,7 +209,7 @@ def format_value_default(val, colormap, **_):
     val = str(val)
     escapedval = val.replace('\\', '\\\\')
     bval = CONTROLCHARS_RE.sub(_show_control_chars, escapedval)
-    return bval if colormap is NO_COLOR_MAP else color_text(bval, colormap)
+    return bval if colormap is NO_COLOR_MAP else color_text(bval, colormap, alignment=Alignment.RIGHT)
 
 
 # Mapping cql type base names ("int", "map", etc) to formatter functions,
@@ -249,7 +249,7 @@ class BlobType:
 @formatter_for('BlobType')
 def format_value_blob(val, colormap, **_):
     bval = '0x' + val.hex()
-    return colorme(bval, colormap, 'blob')
+    return colorme(bval, colormap, 'blob', alignment=Alignment.LEFT)
 
 
 formatter_for('bytearray')(format_value_blob)
@@ -257,23 +257,23 @@ formatter_for('buffer')(format_value_blob)
 formatter_for('blob')(format_value_blob)
 
 
-def format_python_formatted_type(val, colormap, color, quote=False):
+def format_python_formatted_type(val, colormap, color, quote=False, alignment=None):
     bval = str(val)
     if quote:
         bval = "'%s'" % bval
-    return colorme(bval, colormap, color)
+    return colorme(bval, colormap, color, alignment)
 
 
 @formatter_for('Decimal')
 def format_value_decimal(val, float_precision, colormap, decimal_sep=None, thousands_sep=None, **_):
     if (decimal_sep and decimal_sep != '.') or thousands_sep:
         return format_floating_point_type(val, colormap, float_precision, decimal_sep, thousands_sep)
-    return format_python_formatted_type(val, colormap, 'decimal')
+    return format_python_formatted_type(val, colormap, 'decimal', alignment=Alignment.RIGHT)
 
 
 @formatter_for('UUID')
 def format_value_uuid(val, colormap, **_):
-    return format_python_formatted_type(val, colormap, 'uuid')
+    return format_python_formatted_type(val, colormap, 'uuid', alignment=Alignment.LEFT)
 
 
 formatter_for('timeuuid')(format_value_uuid)
@@ -281,14 +281,14 @@ formatter_for('timeuuid')(format_value_uuid)
 
 @formatter_for('inet')
 def formatter_value_inet(val, colormap, quote=False, **_):
-    return format_python_formatted_type(val, colormap, 'inet', quote=quote)
+    return format_python_formatted_type(val, colormap, 'inet', quote=quote, alignment=Alignment.LEFT)
 
 
 @formatter_for('bool')
 def format_value_boolean(val, colormap, boolean_styles=None, **_):
     if boolean_styles:
         val = boolean_styles[0] if val else boolean_styles[1]
-    return format_python_formatted_type(val, colormap, 'boolean')
+    return format_python_formatted_type(val, colormap, 'boolean', alignment=Alignment.LEFT)
 
 
 formatter_for('boolean')(format_value_boolean)
@@ -318,7 +318,7 @@ def format_floating_point_type(val, colormap, float_precision, decimal_sep=None,
             if decimal_sep:
                 bval = bval.replace('.', decimal_sep)
 
-    return colorme(bval, colormap, 'float')
+    return colorme(bval, colormap, 'float', alignment=Alignment.RIGHT)
 
 
 formatter_for('float')(format_floating_point_type)
@@ -329,7 +329,7 @@ def format_integer_type(val, colormap, thousands_sep=None, **_):
     # base-10 only for now; support others?
     bval = format_integer_with_thousands_sep(val, thousands_sep) if thousands_sep else str(val)
     bval = str(bval)
-    return colorme(bval, colormap, 'int')
+    return colorme(bval, colormap, 'int', alignment=Alignment.RIGHT)
 
 
 def format_integer_with_thousands_sep(val, thousands_sep=','):
@@ -357,7 +357,7 @@ def format_value_timestamp(val, colormap, date_time_format, quote=False, **_):
 
     if quote:
         bval = "'%s'" % bval
-    return colorme(bval, colormap, 'timestamp')
+    return colorme(bval, colormap, 'timestamp', alignment=Alignment.LEFT)
 
 
 formatter_for('timestamp')(format_value_timestamp)
@@ -399,17 +399,18 @@ def round_microseconds(val):
 
 @formatter_for('Date')
 def format_value_date(val, colormap, **_):
-    return format_python_formatted_type(val, colormap, 'date')
+    return format_python_formatted_type(val, colormap, 'date', alignment=Alignment.LEFT)
 
 
 @formatter_for('Time')
 def format_value_time(val, colormap, **_):
-    return format_python_formatted_type(val, colormap, 'time')
+    return format_python_formatted_type(val, colormap, 'time', alignment=Alignment.LEFT)
 
 
 @formatter_for('Duration')
 def format_value_duration(val, colormap, **_):
-    return format_python_formatted_type(duration_as_str(val.months, val.days, val.nanoseconds), colormap, 'duration')
+    return format_python_formatted_type(duration_as_str(val.months, val.days, val.nanoseconds),
+                                        colormap,'duration', alignment=Alignment.RIGHT)
 
 
 def duration_as_str(months, days, nanoseconds):
@@ -485,7 +486,9 @@ def format_value_text(val, encoding, colormap, quote=False, **_):
     bval = escapedval
     if quote:
         bval = "'{}'".format(bval)
-    return bval if colormap is NO_COLOR_MAP else color_text(bval, colormap, wcwidth.wcswidth(bval))
+    if colormap is NO_COLOR_MAP:
+        return bval
+    return color_text(bval, colormap, wcwidth.wcswidth(bval), alignment=Alignment.LEFT)
 
 
 # name alias
@@ -510,7 +513,7 @@ def format_simple_collection(val, cqltype, lbracket, rbracket, encoding,
                    for s in (lbracket, ', ', rbracket)]
     coloredval = lb + sep.join(sval.coloredval for sval in subs) + rb
     displaywidth = 2 * len(subs) + sum(sval.displaywidth for sval in subs)
-    return FormattedValue(bval, coloredval, displaywidth)
+    return FormattedValue(bval, coloredval, displaywidth, alignment=Alignment.RIGHT)
 
 
 @formatter_for('list')
@@ -562,7 +565,7 @@ def format_value_map(val, cqltype, encoding, colormap, date_time_format, float_p
         + comma.join(k.coloredval + colon + v.coloredval for (k, v) in subs) \
         + rb
     displaywidth = 4 * len(subs) + sum(k.displaywidth + v.displaywidth for (k, v) in subs)
-    return FormattedValue(bval, coloredval, displaywidth)
+    return FormattedValue(bval, coloredval, displaywidth, alignment=Alignment.RIGHT)
 
 
 formatter_for('OrderedDict')(format_value_map)
@@ -575,7 +578,7 @@ def format_value_utype(val, cqltype, encoding, colormap, date_time_format, float
                        decimal_sep, thousands_sep, boolean_styles, **_):
     def format_field_value(v, t):
         if v is None:
-            return colorme(nullval, colormap, 'error')
+            return colorme(nullval, colormap, 'error', alignment=Alignment.RIGHT)
         return format_value(v, cqltype=t, encoding=encoding, colormap=colormap,
                             date_time_format=date_time_format, float_precision=float_precision,
                             nullval=nullval, quote=True, decimal_sep=decimal_sep,
@@ -596,7 +599,7 @@ def format_value_utype(val, cqltype, encoding, colormap, date_time_format, float
         + comma.join(k.coloredval + colon + v.coloredval for (k, v) in subs) \
         + rb
     displaywidth = 4 * len(subs) + sum(k.displaywidth + v.displaywidth for (k, v) in subs)
-    return FormattedValue(bval, coloredval, displaywidth)
+    return FormattedValue(bval, coloredval, displaywidth, alignment=Alignment.RIGHT)
 
 
 NANOS_PER_MICRO = 1000

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/AbstractCell.java

@@ -61,7 +61,18 @@ public abstract class AbstractCell<V> extends Cell<V>
 
     public boolean isTombstone()
     {
-        return localDeletionTime() != NO_DELETION_TIME && ttl() == NO_TTL;
+        return isTombstone(localDeletionTime());
+    }
+
+    public long minDeletionTime()
+    {
+        long localDeletionTime = localDeletionTime();
+        return isTombstone(localDeletionTime) ? Long.MIN_VALUE : localDeletionTime;
+    }
+
+    private boolean isTombstone(long localDeletionTime)
+    {
+        return localDeletionTime != NO_DELETION_TIME && ttl() == NO_TTL;
     }
 
     public boolean isExpiring()

src/java/org/apache/cassandra/db/rows/ArrayCell.java

@@ -31,17 +31,12 @@ import org.apache.cassandra.utils.memory.ByteBufferCloner;
 
 import static org.apache.cassandra.utils.ByteArrayUtil.EMPTY_BYTE_ARRAY;
 
-public class ArrayCell extends AbstractCell<byte[]>
+public class ArrayCell extends HeapAbstractCell<byte[]>
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new ArrayCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, EMPTY_BYTE_ARRAY, null));
 
     // Careful: Adding vars here has an impact on memtable size
-    private final long timestamp;
-    private final int ttl;
-    private final int localDeletionTimeUnsignedInteger;
-
     private final byte[] value;
-    private final CellPath path;
 
     // Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not
     // available.
@@ -52,12 +47,8 @@ public class ArrayCell extends AbstractCell<byte[]>
 
     public ArrayCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, byte[] value, CellPath path)
     {
-        super(column);
-        this.timestamp = timestamp;
-        this.ttl = ttl;
-        this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
+        super(column, timestamp, ttl, localDeletionTimeUnsignedInteger, path);
         this.value = value;
-        this.path = path;
     }
 
     public static ArrayCell live(ColumnMetadata column, long timestamp, byte[] value, CellPath path)
@@ -71,16 +62,6 @@ public class ArrayCell extends AbstractCell<byte[]>
         return new ArrayCell(column, timestamp, ttl, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl), value, path);
     }
 
-    public long timestamp()
-    {
-        return timestamp;
-    }
-
-    public int ttl()
-    {
-        return ttl;
-    }
-
     public byte[] value()
     {
         return value;
@@ -91,10 +72,6 @@ public class ArrayCell extends AbstractCell<byte[]>
         return ByteArrayAccessor.instance;
     }
 
-    public CellPath path()
-    {
-        return path;
-    }
 
     public Cell<?> withUpdatedColumn(ColumnMetadata newColumn)
     {
@@ -144,9 +121,4 @@ public class ArrayCell extends AbstractCell<byte[]>
         return EMPTY_SIZE + ObjectSizes.sizeOfArray(value) - value.length + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
     }
 
-    @Override
-    protected int localDeletionTimeAsUnsignedInt()
-    {
-        return localDeletionTimeUnsignedInteger;
-    }
 }

src/java/org/apache/cassandra/db/rows/BTreeRow.java

@@ -172,7 +172,7 @@ public class BTreeRow extends AbstractRow
 
     private static long minDeletionTime(Cell<?> cell)
     {
-        return cell.isTombstone() ? Long.MIN_VALUE : cell.localDeletionTime();
+        return cell.minDeletionTime();
     }
 
     private static long minDeletionTime(LivenessInfo info)
@@ -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/BufferCell.java

@@ -30,17 +30,12 @@ import org.apache.cassandra.utils.memory.ByteBufferCloner;
 
 import static java.lang.String.format;
 
-public class BufferCell extends AbstractCell<ByteBuffer>
+public class BufferCell extends HeapAbstractCell<ByteBuffer>
 {
     private static final long EMPTY_SIZE = ObjectSizes.measure(new BufferCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, null));
 
     // Careful: Adding vars here has an impact on memtable size
-    private final long timestamp;
-    private final int ttl;
-    private final int localDeletionTimeUnsignedInteger;
-
     private final ByteBuffer value;
-    private final CellPath path;
 
     // Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not
     // available.
@@ -51,14 +46,10 @@ public class BufferCell extends AbstractCell<ByteBuffer>
 
     public BufferCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, ByteBuffer value, CellPath path)
     {
-        super(column);
+        super(column, timestamp, ttl, localDeletionTimeUnsignedInteger, path);
         assert !column.isPrimaryKeyColumn();
         assert column.isComplex() == (path != null) : format("Column %s.%s(%s: %s) isComplex: %b with cellpath: %s", column.ksName, column.cfName, column.name, column.type.toString(), column.isComplex(), path);
-        this.timestamp = timestamp;
-        this.ttl = ttl;
-        this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
         this.value = value;
-        this.path = path;
     }
 
     public static BufferCell live(ColumnMetadata column, long timestamp, ByteBuffer value)
@@ -92,16 +83,6 @@ public class BufferCell extends AbstractCell<ByteBuffer>
         return new BufferCell(column, timestamp, NO_TTL, nowInSec, ByteBufferUtil.EMPTY_BYTE_BUFFER, path);
     }
 
-    public long timestamp()
-    {
-        return timestamp;
-    }
-
-    public int ttl()
-    {
-        return ttl;
-    }
-
     public ByteBuffer value()
     {
         return value;
@@ -112,10 +93,6 @@ public class BufferCell extends AbstractCell<ByteBuffer>
         return ByteBufferAccessor.instance;
     }
 
-    public CellPath path()
-    {
-        return path;
-    }
 
     public Cell<?> withUpdatedColumn(ColumnMetadata newColumn)
     {
@@ -163,10 +140,4 @@ public class BufferCell extends AbstractCell<ByteBuffer>
     {
         return EMPTY_SIZE + ObjectSizes.sizeOnHeapExcludingDataOf(value) + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
     }
-
-    @Override
-    protected int localDeletionTimeAsUnsignedInt()
-    {
-        return localDeletionTimeUnsignedInteger;
-    }
 }

src/java/org/apache/cassandra/db/rows/Cell.java

@@ -150,6 +150,8 @@ public abstract class Cell<V> extends ColumnData
         return deletionTimeUnsignedIntegerToLong(localDeletionTimeAsUnsignedInt());
     }
 
+    public abstract long minDeletionTime();
+
     /**
      * Whether the cell is a tombstone or not.
      *

src/java/org/apache/cassandra/db/rows/HeapAbstractCell.java

@@ -0,0 +1,64 @@
+/*
+ * 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.db.rows;
+
+import org.apache.cassandra.schema.ColumnMetadata;
+
+public abstract class HeapAbstractCell<V> extends AbstractCell<V>
+{
+    // Careful: Adding vars here has an impact on memtable size
+    protected final long timestamp;
+    protected final int ttl;
+    protected final int localDeletionTimeUnsignedInteger;
+
+    protected final CellPath path;
+
+    protected HeapAbstractCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, CellPath path)
+    {
+        super(column);
+        this.timestamp = timestamp;
+        this.ttl = ttl;
+        this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
+        this.path = path;
+    }
+
+    @Override
+    public long timestamp()
+    {
+        return timestamp;
+    }
+
+    @Override
+    public int ttl()
+    {
+        return ttl;
+    }
+
+    @Override
+    protected int localDeletionTimeAsUnsignedInt()
+    {
+        return localDeletionTimeUnsignedInteger;
+    }
+
+    @Override
+    public CellPath path()
+    {
+        return path;
+    }
+}

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/CellWithSource.java

@@ -103,6 +103,12 @@ public class CellWithSource<T> extends Cell<T>
         return cell.localDeletionTime();
     }
 
+    @Override
+    public long minDeletionTime()
+    {
+        return cell.minDeletionTime();
+    }
+
     @Override
     public boolean isTombstone()
     {

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/distributed/org/apache/cassandra/distributed/test/tcm/CMSShutdownTest.java

@@ -50,7 +50,7 @@ public class CMSShutdownTest extends TestBaseImpl
         // This test simulates a CMS node attempting to commit an entry to the log but being unable
         // to obtain consensus from other CMS members while it is also shutting down itself.
         try (Cluster cluster = Cluster.build(2)
-                                      .withConfig(c -> c.with(Feature.values()))
+                                      .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
                                       .withInstanceInitializer(BBHelper::install)
                                       .start())
         {
@@ -67,8 +67,15 @@ public class CMSShutdownTest extends TestBaseImpl
             // latch ensures that every commit will fail as if unable to obtain consensus
             // from other CMS members
             State.latch.countDown();
-            for (; ;)
-                ClusterMetadataService.instance().commit(TriggerSnapshot.instance);
+            try
+            {
+                while (!Thread.currentThread().isInterrupted())
+                    ClusterMetadataService.instance().commit(TriggerSnapshot.instance);
+            }
+            catch (Throwable t)
+            {
+                // Commits fail by design; an interrupt or a stopped processor on shutdown ends them.
+            }
         }
 
         public static void scheduleCommits()

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