Merge 7dcfc6f343 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"/>

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

@@ -460,7 +460,7 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
         if(this == NONE)
             return 0;
 
-        return EMPTY_SIZE + BTree.sizeOfStructureOnHeap(columns);
+        return EMPTY_SIZE + BTree.sizeOnHeapOf(columns);
     }
 
     @Override

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
+    private static int valueLengthIfFixed(AbstractType<?> elementType, int dimension)
-    public int valueLengthIfFixed()
     {
-        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);
+        super(column, timestamp, ttl, localDeletionTimeUnsignedInteger, path);
-        this.timestamp = timestamp;
-        this.ttl = ttl;
-        this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
         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))
@@ -569,7 +574,7 @@ public class BTreeRow extends AbstractRow
                         + clustering.unsharedHeapSize()
                         + primaryKeyLivenessInfo.unsharedHeapSize()
                         + deletion.unsharedHeapSize()
-                        + BTree.sizeOfStructureOnHeap(btree);
+                        + BTree.sizeOnHeapOf(btree);
 
         return accumulate((cd, v) -> v + cd.unsharedHeapSize(), heapSize);
     }
@@ -581,7 +586,7 @@ public class BTreeRow extends AbstractRow
                         + clustering.unsharedHeapSizeExcludingData()
                         + primaryKeyLivenessInfo.unsharedHeapSize()
                         + deletion.unsharedHeapSize()
-                        + BTree.sizeOfStructureOnHeap(btree);
+                        + BTree.sizeOnHeapOf(btree);
 
         return accumulate((cd, v) -> v + cd.unsharedHeapSizeExcludingData(), heapSize);
     }
@@ -660,10 +665,20 @@ public class BTreeRow extends AbstractRow
                 {
                     // The update's deletion shadows part of the existing row. Those cells ARE owned by
                     // the memtable, so record their removal via retain().
-                    existingBtree = BTree.transformAndFilter(existingBtree, reconciler::retain);
+                    Object[] retained = BTree.transformAndFilter(existingBtree, reconciler::retain);
+                    if (existingBtree != retained)
+                    {
+                        reconcileF.onAllocatedOnHeap(BTree.sizeOnHeapOf(retained) - BTree.sizeOnHeapOf(existingBtree));
+                        existingBtree = retained;
+                    }
                 }
             }
             Object[] tree = BTree.update(existingBtree, updateBtree, ColumnData.comparator, reconciler);
+            // BTree.update and the reconciler only account the column data (cells and column-tree nodes); the row's
+            // own LivenessInfo/Deletion are not. When they change (e.g. a row tombstone supersedes a live row) the
+            // new objects become memtable-owned and the old ones are released, so account that delta here.
+            reconcileF.onAllocatedOnHeap((livenessInfo.unsharedHeapSize() + rowDeletion.unsharedHeapSize())
+                                         - (existing.primaryKeyLivenessInfo().unsharedHeapSize() + existing.deletion().unsharedHeapSize()));
             return new BTreeRow(existing.clustering, livenessInfo, rowDeletion, tree, minDeletionTime(tree, livenessInfo, deletion));
         }
     }

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

@@ -156,6 +156,7 @@ public abstract class ColumnData implements IMeasurableMemory
                     }
                     cells = BTree.update(existingTree, updateTree, existingComplex.column.cellComparator(), (UpdateFunction) reconciler);
                 }
+                onAllocatedOnHeap(maxComplexDeletion.unsharedHeapSize() - existingDeletion.unsharedHeapSize());
                 return new ComplexColumnData(existingComplex.column, cells, maxComplexDeletion);
             }
         }
@@ -217,8 +218,28 @@ public abstract class ColumnData implements IMeasurableMemory
                 ComplexColumnData existingComplex = (ComplexColumnData) existing;
                 if (activeDeletion.supersedes(existingComplex.complexDeletion()))
                 {
-                    Object[] cells = BTree.transformAndFilter(existingComplex.tree(), (ColumnData cd) -> removeShadowed(cd, recordDeletion));
+                    Object[] existingTree = existingComplex.tree();
-                    return BTree.isEmpty(cells) ? null : new ComplexColumnData(existingComplex.column, cells, DeletionTime.LIVE);
+                    Object[] cells = BTree.transformAndFilter(existingTree, (ColumnData cd) -> removeShadowed(cd, recordDeletion));
+                    ComplexColumnData result = BTree.isEmpty(cells) ? null
+                                                                    : new ComplexColumnData(existingComplex.column, cells, DeletionTime.LIVE);
+                    // The shadowed inner cells are released through recordDeletion.delete above, but that does not cover
+                    // the complex column's own structure: its cell tree (which can span multiple BTree nodes), its
+                    // complex deletion, and, when the column is dropped entirely, its wrapper. All were counted as
+                    // owned when the column was first written (ComplexColumnData.unsharedHeapSizeExcludingData), so release that
+                    // delta here. The rewritten column carries DeletionTime.LIVE, so the dropped complex
+                    // deletion's heap is released too, matching the swap Reconciler.merge accounts on its path.
+                    // On the update side (recordDeletion == noOp) this is a no-op, so skip it entirely.
+                    if (recordDeletion != ColumnData.noOp)
+                    {
+                        long structureBefore = ComplexColumnData.EMPTY_SIZE
+                                               + existingComplex.complexDeletion().unsharedHeapSize()
+                                               + BTree.sizeOnHeapOf(existingTree);
+                        long structureAfter = result == null ? 0 : ComplexColumnData.EMPTY_SIZE
+                                                                   + DeletionTime.LIVE.unsharedHeapSize()
+                                                                   + BTree.sizeOnHeapOf(cells);
+                        recordDeletion.onAllocatedOnHeap(structureAfter - structureBefore);
+                    }
+                    return result;
                 }
             }
 

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

@@ -51,11 +51,11 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
 {
     static final Cell<?>[] NO_CELLS = new Cell<?>[0];
 
-    private static final long EMPTY_SIZE = ObjectSizes.measure(new ComplexColumnData(ColumnMetadata.regularColumn("",
+    static final long EMPTY_SIZE = ObjectSizes.measure(new ComplexColumnData(ColumnMetadata.regularColumn("",
-                                                                                                                  "",
+                                                                                                          "",
-                                                                                                                  "",
+                                                                                                          "",
-                                                                                                                  SetType.getInstance(ByteType.instance, true),
+                                                                                                          SetType.getInstance(ByteType.instance, true),
-                                                                                                                  ColumnMetadata.NO_UNIQUE_ID),
+                                                                                                          ColumnMetadata.NO_UNIQUE_ID),
                                                                                      NO_CELLS,
                                                                                      DeletionTime.build(0, 0)));
 
@@ -151,7 +151,7 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
     @Override
     public long unsharedHeapSizeExcludingData()
     {
-        long heapSize = EMPTY_SIZE + BTree.sizeOnHeapOf(cells);
+        long heapSize = EMPTY_SIZE + BTree.sizeOnHeapOf(cells) + complexDeletion.unsharedHeapSize();
         // TODO: this can be turned into a simple multiplication, at least while we have only one Cell implementation
         for (Cell<?> cell : this)
             heapSize += cell.unsharedHeapSizeExcludingData();

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,
+                Object[] tree = BTree.build(it, dataBufferSize, UpdateFunction.noOp());
-                                       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.
+             * Determines whether {@code dataColumn} should replace the currently selected column metadata,
-             * @param dataColumn the {@code ColumnMetadata} to use.
+             * i.e. whether no column has been selected yet or {@code dataColumn} is a newer version.
-             * @return {@code true} if the {@code ColumnMetadata} is the one that should be used, {@code false} otherwise.
+             * @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)

src/java/org/apache/cassandra/utils/btree/BTree.java

@@ -86,6 +86,7 @@ public class BTree
     public static final int MIN_KEYS = BRANCH_FACTOR / 2 - 1;
     public static final int MAX_KEYS = BRANCH_FACTOR - 1;
     public static final long STOP_SENTINEL_VALUE = Long.MAX_VALUE;
+    private static final long MAX_KEYS_REF_ARRAY_SIZE = ObjectSizes.sizeOfReferenceArray(MAX_KEYS);
 
     // An empty BTree Leaf - which is the same as an empty BTree
     private static final Object[] EMPTY_LEAF = new Object[1];
@@ -979,19 +980,6 @@ public class BTree
         return sizeMap(tree)[(length / 2) - 1];
     }
 
-    public static long sizeOfStructureOnHeap(Object[] tree)
-    {
-        if (tree == EMPTY_LEAF)
-            return 0;
-
-        long size = ObjectSizes.sizeOfArray(tree);
-        if (isLeaf(tree))
-            return size;
-        for (int i = getChildStart(tree); i < getChildEnd(tree); i++)
-            size += sizeOfStructureOnHeap((Object[]) tree[i]);
-        return size;
-    }
-
     /**
      * Checks is the node is a leaf.
      *
@@ -2331,6 +2319,23 @@ public class BTree
         return ObjectSizes.sizeOfArray(tree);
     }
 
+    /**
+     * The heap occupied by a single node (its backing array, plus the sizeMap for a branch), <em>not</em> including
+     * its children. This is the per-node contribution to {@link #sizeOnHeapOf(Object[])}; the builders use it to
+     * keep their running {@code allocated} total net: every newly retained node adds its shallow heap and every
+     * source node it replaces subtracts its shallow heap, so reused subtrees cancel.
+     */
+    private static long shallowHeapOf(Object[] node)
+    {
+        if (isEmpty(node))
+            return 0;
+
+        long size = ObjectSizes.sizeOfArray(node);
+        if (!isLeaf(node))
+            size += ObjectSizes.sizeOfArray(sizeMap(node));
+        return size;
+    }
+
     // Arbitrary boundaries
     private static Object POSITIVE_INFINITY = new Object();
     private static Object NEGATIVE_INFINITY = new Object();
@@ -2646,6 +2651,7 @@ public class BTree
      */
     private static abstract class LeafBuilder extends LeafOrBranchBuilder
     {
+        static final long DISABLED = Long.MIN_VALUE;
         long allocated;
 
         LeafBuilder()
@@ -2654,6 +2660,13 @@ public class BTree
             buffer = new Object[MAX_KEYS];
         }
 
+        /** Adjust the running heap total, when accounting is enabled */
+        final void addAllocated(long bytes)
+        {
+            if (allocated != DISABLED)
+                allocated += bytes;
+        }
+
         /**
          * Add {@code nextKey} to the buffer, overflowing if necessary
          */
@@ -2830,8 +2843,11 @@ public class BTree
             int newPredecessorCount = predSize - steal;
             Object[] newPredecessor = new Object[newPredecessorCount | 1];
             System.arraycopy(pred, 0, newPredecessor, 0, newPredecessorCount);
-            if (allocated >= 0)
+            // we replace the single source leaf with two new leaves (newPredecessor + newLeaf); account both and
-                allocated += ObjectSizes.sizeOfReferenceArray(newPredecessorCount | 1);
+            // release the source we consumed, so the running total stays a net delta
+            addAllocated(ObjectSizes.sizeOfArray(newPredecessor)
+                         + ObjectSizes.sizeOfArray(newLeaf)
+                         - (sourceNode == null ? 0 : sizeOnHeapOfLeaf(sourceNode)));
             ensureParent().addChildAndNextKey(newPredecessor, newPredecessorCount, pred[newPredecessorCount]);
             return newLeaf;
         }
@@ -2883,8 +2899,7 @@ public class BTree
         {
             // propagate the leaf we have saved in savedBuffer
             // precondition: savedLeafCount == MAX_KEYS
-            if (allocated >= 0)
+            addAllocated(MAX_KEYS_REF_ARRAY_SIZE);
-                allocated += ObjectSizes.sizeOfReferenceArray(MAX_KEYS);
             ensureParent().addChildAndNextKey(savedBuffer, MAX_KEYS, savedNextKey);
             savedBuffer = null;
             savedNextKey = null;
@@ -2921,9 +2936,7 @@ public class BTree
                     propagateOverflow();
 
                 sizeOfLeaf = count;
-                leaf = drain();
+                leaf = drain(); // drain() accounts the new leaf and releases the source it replaces
-                if (allocated >= 0 && sizeOfLeaf > 0)
-                    allocated += ObjectSizes.sizeOfReferenceArray(sizeOfLeaf | 1) - (sourceNode == null ? 0 : sizeOnHeapOfLeaf(sourceNode));
             }
 
             count = 0;
@@ -2941,19 +2954,27 @@ public class BTree
             // the number of children here may be smaller than MIN_KEYS if this is the root node
             assert !hasOverflow();
             if (count == 0)
+            {
+                // defensive: unreachable from BTree.update
+                if (sourceNode != null)
+                    addAllocated(-sizeOnHeapOfLeaf(sourceNode));
+                clearSourceNode();
                 return empty();
+            }
 
             Object[] newLeaf;
             if (sourceNode != null
                 && count == sizeOfLeaf(sourceNode)
                 && areIdentical(buffer, 0, sourceNode, 0, count))
             {
-                newLeaf = sourceNode;
+                newLeaf = sourceNode; // reused unchanged: neither allocated nor released
             }
             else
             {
                 newLeaf = new Object[count | 1];
                 System.arraycopy(buffer, 0, newLeaf, 0, count);
+                // account the new leaf and release the source it replaces, before clearSourceNode() drops it
+                addAllocated(ObjectSizes.sizeOfArray(newLeaf) - (sourceNode == null ? 0 : sizeOnHeapOfLeaf(sourceNode)));
             }
             count = 0;
             clearSourceNode();
@@ -3054,10 +3075,10 @@ public class BTree
          */
         final void propagateOverflow()
         {
-            // propagate the leaf we have saved in leaf().savedBuffer
+            // propagate the branch we have saved in savedBuffer; it is a brand new node, so account it in full
-            if (leaf.allocated >= 0)
+            // (array + sizeMap) with nothing to release
-                leaf.allocated += ObjectSizes.sizeOfReferenceArray(2 * (1 + MAX_KEYS));
             int size = setOverflowSizeMap(savedBuffer, MAX_KEYS);
+            leaf.addAllocated(shallowHeapOf(savedBuffer));
             ensureParent().addChildAndNextKey(savedBuffer, size, savedNextKey);
             savedBuffer = null;
             savedNextKey = null;
@@ -3114,8 +3135,11 @@ public class BTree
             System.arraycopy(savedBuffer, 0, savedBranch, 0, savedBranchCount);
             System.arraycopy(savedBuffer, MAX_KEYS, savedBranch, savedBranchCount, savedBranchCount + 1);
             int savedBranchSize = setOverflowSizeMap(savedBranch, savedBranchCount);
-            if (leaf.allocated >= 0)
+            // we replace the single source branch with two new branches (savedBranch + newBranch); account both
-                leaf.allocated += ObjectSizes.sizeOfReferenceArray(2 * (1 + savedBranchCount));
+            // (array + sizeMap) and release the source we consumed
+            leaf.addAllocated(shallowHeapOf(savedBranch)
+                              + shallowHeapOf(newBranch)
+                              - (sourceNode == null ? 0 : shallowHeapOf(sourceNode)));
             ensureParent().addChildAndNextKey(savedBranch, savedBranchSize, savedBuffer[savedBranchCount]);
             savedNextKey = null;
 
@@ -3224,6 +3248,8 @@ public class BTree
                     System.arraycopy(buffer, 0, branch, 0, count);
                     System.arraycopy(buffer, MAX_KEYS, branch, count, count + 1);
                     sizeOfBranch = setDrainSizeMap(unode, usz, branch, count);
+                    // account the new branch (array + sizeMap) and release the source branch it replaces
+                    leaf.addAllocated(shallowHeapOf(branch) - (unode == null ? 0 : shallowHeapOf(unode)));
                 }
             }
 
@@ -3245,6 +3271,9 @@ public class BTree
             assert !hasOverflow();
             if (count == 0)
             {
+                // defensive: unreachable from BTree.update
+                if (sourceNode != null)
+                    leaf.addAllocated(-shallowHeapOf(sourceNode));
                 clearSourceNode();
                 hasRightChild = false;
                 return (Object[]) buffer[MAX_KEYS];
@@ -3256,7 +3285,7 @@ public class BTree
                 && areIdentical(buffer, 0, sourceNode, 0, count)
                 && areIdentical(buffer, MAX_KEYS, sourceNode, count, count + 1))
             {
-                branch = sourceNode;
+                branch = sourceNode; // reused unchanged: neither allocated nor released
             }
             else
             {
@@ -3273,6 +3302,9 @@ public class BTree
                     System.arraycopy(buffer, MAX_KEYS, branch, count, count + 1);
                 }
                 setDrainSizeMap(null, -1, branch, count);
+                // account the new branch (array + sizeMap) and release the source branch it replaces,
+                // before clearSourceNode() drops it
+                leaf.addAllocated(shallowHeapOf(branch) - (sourceNode == null ? 0 : shallowHeapOf(sourceNode)));
             }
 
             count = 0;
@@ -3506,6 +3538,7 @@ public class BTree
                 branch.inUse = false;
                 branch = branch.parent;
             }
+            leaf().clearSourceNode();
             Invariants.require(branch == null || (branch.count == 0 && !branch.hasRightChild));
         }
 
@@ -3554,7 +3587,7 @@ public class BTree
 
         FastBuilder()
         {
-            allocated = -1;
+            allocated = DISABLED;
         } // disable allocation tracking
 
         public void add(V value)
@@ -3667,7 +3700,7 @@ public class BTree
             this.insert.init(insert);
             this.updateF = updateF;
             this.comparator = comparator;
-            this.allocated = isSimple(updateF) ? -1 : 0;
+            this.allocated = isSimple(updateF) ? DISABLED : 0;
             int leafDepth = BTree.depth(update) - 1;
             LeafOrBranchBuilder builder = leaf();
             Invariants.require(builder.isEmpty());
@@ -3678,12 +3711,17 @@ public class BTree
                 builder = branch;
             }
 
+            // record the root as the top builder's source, so that when the root node is rebuilt its old version is
+            // released from the running heap total (and an unchanged root can be reused), mirroring the per-node
+            // accounting performed for every child level
+            builder.setSourceNode(update);
+
             Insert ik = this.insert.next();
             ik = updateRecursive(ik, update, null, builder);
             assert ik == null;
             Object[] result = builder.completeBuild();
 
-            if (allocated > 0)
+            if (allocated != DISABLED)
                 updateF.onAllocatedOnHeap(allocated);
 
             return result;
@@ -3864,7 +3902,7 @@ public class BTree
 
         AbstractSeekingTransformer()
         {
-            allocated = -1;
+            allocated = DISABLED;
             ensureParent();
             parent.inUse = false;
         }

src/java/org/apache/cassandra/utils/btree/UpdateFunction.java

@@ -44,7 +44,7 @@ public interface UpdateFunction<K, V>
     V merge(V replacing, K update);
 
     /**
-     * @param heapSize extra heap space allocated (over previous tree)
+     * @param heapSize heap space signed delta allocated (over previous tree), can be negative
      */
     void onAllocatedOnHeap(long heapSize);
 

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 (; ;)
+            try
-                ClusterMetadataService.instance().commit(TriggerSnapshot.instance);
+            {
+                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/partitions/PartitionRowAccountingTest.java

@@ -0,0 +1,423 @@
+/*
+ * 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.partitions;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.utils.btree.BTree;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Row-level analogue of {@link SetCellAccountingTest}. Where that test grows/resets a single {@code set<text>}
+ * column on one row, this test grows/resets the <em>rows of a single partition</em>:
+ * <ul>
+ *   <li>adding a row is the analogue of adding a set element (both merge a new entry into a BTree -- here the
+ *       partition's row tree, via {@link BTreePartitionUpdater#makeMergedPartition} →
+ *       {@code BTree.update(current.tree, update.holder().tree, ...)});</li>
+ *   <li>a whole-partition tombstone is the analogue of a {@code set = {...}} override: it shadows every previously
+ *       written row, just as a collection assignment shadows every previously written element.</li>
+ * </ul>
+ * It runs a grow/reset op mix on one partition with strictly increasing timestamps and asserts the memtable's
+ * on/off-heap ownership never goes negative -- i.e. the update accounting never releases more than it allocated.
+ */
+public class PartitionRowAccountingTest extends CQLTester
+{
+    private static final Logger logger = LoggerFactory.getLogger(PartitionRowAccountingTest.class);
+
+    @BeforeClass
+    public static void setUpClass()
+    {
+        ServerTestUtils.daemonInitialization();
+        try
+        {
+            Field confField = DatabaseDescriptor.class.getDeclaredField("conf");
+            confField.setAccessible(true);
+            Config conf = (Config) confField.get(null);
+            conf.memtable_allocation_type = Config.MemtableAllocationType.offheap_objects;
+        }
+        catch (ReflectiveOperationException e)
+        {
+            throw new RuntimeException(e);
+        }
+        CQLTester.prepareServer();
+    }
+
+    private ColumnFamilyStore createTestTable()
+    {
+        createTable("CREATE TABLE %s (" +
+                    "    pk text," +
+                    "    ck int," +
+                    "    last_contact timestamp," +
+                    "    namespace text," +
+                    "    properties text," +
+                    "    state text," +
+                    "    v text," +
+                    "    PRIMARY KEY (pk, ck))");
+        return getCurrentColumnFamilyStore();
+    }
+
+    @Test
+    public void largePartitionGrowShrinkKeepOwnsNonNegative()
+    {
+        ColumnFamilyStore cfs = createTestTable();
+        String tbl = KEYSPACE + '.' + currentTable();
+
+        final int rowCount = BTree.MAX_KEYS + 1;
+        final int ops = 2_000;
+        final AtomicLong ts = new AtomicLong(1_000_000L);
+
+        for (int op = 0; op < ops; op++)
+        {
+            String cql;
+            int expectedLiveRows;
+            if (op % 2 == 0)
+            {
+                // grow: add the full set of rows (each row ~ a set element)
+                cql = growBatch(tbl, 0, rowCount, ts.incrementAndGet());
+                expectedLiveRows = rowCount;
+            }
+            else
+            {
+                // reset: a whole-partition tombstone (~ "set = {...}") followed by a smaller set of rows.
+                // the tombstone is at an earlier timestamp than the re-inserted rows so the latter survive,
+                // while every previously written row (at an earlier timestamp) is shadowed by the tombstone.
+                long deleteTs = ts.incrementAndGet();
+                long insertTs = ts.incrementAndGet();
+                cql = resetBatch(tbl, 0, rowCount / 2, deleteTs, insertTs);
+                expectedLiveRows = rowCount / 2;
+            }
+
+            QueryProcessor.executeInternal(cql);
+
+            UntypedResultSet rs = QueryProcessor.executeInternal("SELECT ck FROM " + tbl + " WHERE pk = 'test'");
+            int liveRows = 0;
+            if (rs != null)
+                for (UntypedResultSet.Row ignored : rs)
+                    liveRows++;
+
+            if (op % 100 == 0)
+                logger.info("== op=" + op +
+                            ", liveRows= " + liveRows +
+                            ", heapSize= " + ownsOnHeapNow(cfs) +
+                            ", offheapSize= " + ownsOffHeapNow(cfs));
+
+            // the grow/reset analogue must actually oscillate the visible row count, proving the partition
+            // tombstone shadows the prior rows just as a set override shadows prior elements
+            assertThat(liveRows).as("visible rows after op=" + op).isEqualTo(expectedLiveRows);
+            assertOwnsNonNegative(cfs, "after op=" + op);
+        }
+
+        cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+        assertOwnsNonNegative(cfs, "after flush");
+    }
+
+    /**
+     * Builds two logically identical partitions -- a deleted wide row -- via two merge paths and requires their
+     * on-heap ownership to be exactly equal:
+     * <ul>
+     *   <li>retain path: insert a row's cells, then delete the row (the tombstone shadows the existing, owned cells);</li>
+     *   <li>removeShadowed path: delete the row first, then insert cells the tombstone already shadows.</li>
+     * </ul>
+     * Both end with the same columns, row tombstone and empty cell tree, so correct accounting leaves them owning the
+     * same on-heap. Only the retain path can leak: if {@code BTreeRow.merge} fails to release the shrunk column tree or
+     * the row's liveness/deletion delta, the retain partition owns more -- an inflation a non-negative-ownership check
+     * cannot catch. Off-heap is not compared: under {@code offheap_objects} the retain path has written the cells into
+     * the off-heap slab, only reclaimed at flush, so it legitimately owns more off-heap.
+     */
+    @Test
+    public void rowTombstoneOverExistingRowDoesNotInflateOwnership()
+    {
+        ColumnFamilyStore cfsRetain = createWideTable();
+        String tblRetain = KEYSPACE + '.' + currentTable();
+        ColumnFamilyStore cfsShadow = createWideTable();
+        String tblShadow = KEYSPACE + '.' + currentTable();
+
+        final int rows = 50;
+        final long insertTs = 1000L;
+        final long deleteTs = 2000L; // newer than insertTs, so the tombstone shadows the cells
+
+        for (int ck = 0; ck < rows; ck++)
+        {
+            // retain path: write the row, then delete it -> the tombstone shadows existing (owned) cells
+            QueryProcessor.executeInternal(wideInsert(tblRetain, ck, insertTs));
+            QueryProcessor.executeInternal(rowDelete(tblRetain, ck, deleteTs));
+
+            // removeShadowed path: delete first, then write cells the tombstone already shadows
+            QueryProcessor.executeInternal(rowDelete(tblShadow, ck, deleteTs));
+            QueryProcessor.executeInternal(wideInsert(tblShadow, ck, insertTs));
+        }
+
+        long onHeapRetain = ownsOnHeapNow(cfsRetain), onHeapShadow = ownsOnHeapNow(cfsShadow);
+        long offHeapRetain = ownsOffHeapNow(cfsRetain), offHeapShadow = ownsOffHeapNow(cfsShadow);
+        logger.info("retain-path onHeap=" + onHeapRetain + " offHeap=" + offHeapRetain +
+                    "; removeShadowed-path onHeap=" + onHeapShadow + " offHeap=" + offHeapShadow);
+
+        assertOwnsNonNegative(cfsRetain, "retain path");
+        assertOwnsNonNegative(cfsShadow, "removeShadowed path");
+
+        assertThat(onHeapRetain)
+        .as("writing then deleting a row must leave the memtable owning exactly the same on-heap (%d bytes) as never " +
+            "effectively writing it (%d bytes) -- the shrunk column tree and the row's liveness/deletion delta must be " +
+            "accounted", onHeapRetain, onHeapShadow)
+        .isEqualTo(onHeapShadow);
+    }
+
+    /**
+     * Same as {@link #rowTombstoneOverExistingRowDoesNotInflateOwnership} but the shadowed data lives in a
+     * <em>complex</em> (collection) column whose internal cell tree spans multiple BTree nodes (branch + leaves).
+     * <p>
+     * The retain path ({@code BTreeRow.merge} → {@code reconciler::retain} → {@code ColumnData.removeShadowed} for the
+     * {@code set} column) drops every collection cell when the row tombstone supersedes them, but must also release the
+     * collection's internal tree node structure -- which was counted as memtable-owned when the row was first inserted
+     * ({@code ComplexColumnData.unsharedHeapSizeExcludingData} includes {@code sizeOnHeapOf(tree)}). If that internal
+     * structure is not released, the retain partition owns more on-heap than the logically identical removeShadowed
+     * partition.
+     */
+    @Test
+    public void rowTombstoneOverExistingCollectionDoesNotInflateOwnership()
+    {
+        ColumnFamilyStore cfsRetain = createCollectionTable();
+        String tblRetain = KEYSPACE + '.' + currentTable();
+        ColumnFamilyStore cfsShadow = createCollectionTable();
+        String tblShadow = KEYSPACE + '.' + currentTable();
+
+        final int rows = 50;
+        final long insertTs = 1000L;
+        final long deleteTs = 2000L; // newer than insertTs, so the tombstone shadows the collection cells
+
+        for (int ck = 0; ck < rows; ck++)
+        {
+            // retain path: write the row (large collection), then delete it -> the tombstone shadows existing (owned)
+            // cells and the collection's multi-node internal tree
+            QueryProcessor.executeInternal(setInsert(tblRetain, ck, insertTs));
+            QueryProcessor.executeInternal(rowDelete(tblRetain, ck, deleteTs));
+
+            // removeShadowed path: delete first, then write cells the tombstone already shadows
+            QueryProcessor.executeInternal(rowDelete(tblShadow, ck, deleteTs));
+            QueryProcessor.executeInternal(setInsert(tblShadow, ck, insertTs));
+        }
+
+        long onHeapRetain = ownsOnHeapNow(cfsRetain), onHeapShadow = ownsOnHeapNow(cfsShadow);
+        long offHeapRetain = ownsOffHeapNow(cfsRetain), offHeapShadow = ownsOffHeapNow(cfsShadow);
+        logger.info("collection retain-path onHeap=" + onHeapRetain + " offHeap=" + offHeapRetain +
+                    "; removeShadowed-path onHeap=" + onHeapShadow + " offHeap=" + offHeapShadow);
+
+        assertOwnsNonNegative(cfsRetain, "retain path");
+        assertOwnsNonNegative(cfsShadow, "removeShadowed path");
+
+        assertThat(onHeapRetain)
+        .as("writing then deleting a row with a multi-node collection must leave the memtable owning exactly the same " +
+            "on-heap (%d bytes) as never effectively writing it (%d bytes) -- the collection's freed internal tree " +
+            "structure must be accounted", onHeapRetain, onHeapShadow)
+        .isEqualTo(onHeapShadow);
+    }
+
+    /**
+     * Isolates the complex column's own tombstone heap, which the broader
+     * {@link #rowTombstoneOverExistingCollectionDoesNotInflateOwnership} masks behind a large surviving cell tree.
+     * <p>
+     * A whole-collection delete ({@code DELETE s ...}) writes a {@link org.apache.cassandra.db.rows.ComplexColumnData}
+     * carrying only a <em>non-LIVE</em> complex deletion and an <em>empty</em> cell tree, so its memtable-owned heap
+     * ({@code unsharedHeapSizeExcludingData}) reduces to {@code EMPTY_SIZE + complexDeletion.unsharedHeapSize()} -- the
+     * cell tree contributes nothing. When a newer row tombstone supersedes it, {@code ColumnData.removeShadowed} must
+     * release that complex deletion's heap ({@code DeletionTime.EMPTY_SIZE}) and not just the wrapper. If that term is
+     * dropped, the retain path strands exactly one {@code DeletionTime.EMPTY_SIZE} per row and owns more on-heap than
+     * the logically identical removeShadowed path -- a gap the large-tree test cannot attribute and a non-negative
+     * check cannot see. This is the drop-side half of the complex-deletion heap accounting; the swap-side (merge) half
+     * is exercised exhaustively by {@code AtomicBTreePartitionMemtableAccountingTest}.
+     */
+    @Test
+    public void rowTombstoneOverCollectionDeletionReleasesComplexDeletionHeap()
+    {
+        ColumnFamilyStore cfsRetain = createCollectionTable();
+        String tblRetain = KEYSPACE + '.' + currentTable();
+        ColumnFamilyStore cfsShadow = createCollectionTable();
+        String tblShadow = KEYSPACE + '.' + currentTable();
+
+        final int rows = 200;
+        final long collectionDeleteTs = 1000L;
+        final long rowDeleteTs = 2000L; // newer, so the row tombstone supersedes the collection tombstone
+
+        for (int ck = 0; ck < rows; ck++)
+        {
+            // retain path: write a collection tombstone (an owned complex deletion over an empty cell tree),
+            // then delete the row -> the row tombstone supersedes and releases it via reconciler::retain
+            QueryProcessor.executeInternal(collectionDelete(tblRetain, ck, collectionDeleteTs));
+            QueryProcessor.executeInternal(rowDelete(tblRetain, ck, rowDeleteTs));
+
+            // removeShadowed path: delete the row first, then write a collection tombstone it already shadows
+            QueryProcessor.executeInternal(rowDelete(tblShadow, ck, rowDeleteTs));
+            QueryProcessor.executeInternal(collectionDelete(tblShadow, ck, collectionDeleteTs));
+        }
+
+        long onHeapRetain = ownsOnHeapNow(cfsRetain), onHeapShadow = ownsOnHeapNow(cfsShadow);
+        logger.info("collection-tombstone retain-path onHeap=" + onHeapRetain +
+                    "; removeShadowed-path onHeap=" + onHeapShadow);
+
+        assertOwnsNonNegative(cfsRetain, "retain path");
+        assertOwnsNonNegative(cfsShadow, "removeShadowed path");
+
+        assertThat(onHeapRetain)
+        .as("a row tombstone superseding a collection tombstone must release the complex deletion's own heap " +
+            "(DeletionTime.EMPTY_SIZE per row): retain owns %d, removeShadowed owns %d", onHeapRetain, onHeapShadow)
+        .isEqualTo(onHeapShadow);
+    }
+
+    private static String collectionDelete(String tbl, int ck, long timestamp)
+    {
+        return "DELETE s FROM " + tbl + " USING TIMESTAMP " + timestamp + " WHERE pk = 'test' AND ck = " + ck;
+    }
+
+    private static long ownsOnHeapNow(ColumnFamilyStore cfs)
+    {
+        return Memtable.getMemoryUsage(cfs.getTracker().getView().getCurrentMemtable()).ownsOnHeap;
+    }
+
+    private static long ownsOffHeapNow(ColumnFamilyStore cfs)
+    {
+        return Memtable.getMemoryUsage(cfs.getTracker().getView().getCurrentMemtable()).ownsOffHeap;
+    }
+
+    /**
+     * A batch that inserts rows with clustering keys in [from,to), all at {@code timestamp}.
+     */
+    private static String growBatch(String tbl, int from, int to, long timestamp)
+    {
+        StringBuilder sb = new StringBuilder("BEGIN UNLOGGED BATCH\n");
+        for (int ck = from; ck < to; ck++)
+            appendInsert(sb, tbl, ck, timestamp);
+        return sb.append("APPLY BATCH;").toString();
+    }
+
+    /**
+     * A batch that first deletes the whole partition at {@code deleteTs} (a partition tombstone, the analogue of a
+     * {@code set = {...}} override), then inserts rows with clustering keys in [from,to) at {@code insertTs > deleteTs}.
+     */
+    private static String resetBatch(String tbl, int from, int to, long deleteTs, long insertTs)
+    {
+        StringBuilder sb = new StringBuilder("BEGIN UNLOGGED BATCH\n");
+        sb.append("DELETE FROM ").append(tbl).append(" USING TIMESTAMP ").append(deleteTs)
+          .append(" WHERE pk = 'test';\n");
+        for (int ck = from; ck < to; ck++)
+            appendInsert(sb, tbl, ck, insertTs);
+        return sb.append("APPLY BATCH;").toString();
+    }
+
+    private static void appendInsert(StringBuilder sb, String tbl, int ck, long timestamp)
+    {
+        sb.append("INSERT INTO ").append(tbl)
+          .append(" (pk, ck, namespace, properties, state, v) VALUES ('test', ").append(ck)
+          .append(", '").append(elemName(ck)).append('\'')
+          .append(", '").append(elemName(ck)).append('\'')
+          .append(", '").append(elemName(ck)).append('\'')
+          .append(", '").append(elemName(ck)).append('\'')
+          .append(") USING TIMESTAMP ").append(timestamp).append(";\n");
+    }
+
+    private static String elemName(int i)
+    {
+        return 'e' + String.format("%05d", i);
+    }
+
+    private static final int WIDE_COLS = BTree.MAX_KEYS + 1; // one more than fits in a leaf, so each row's column tree spans multiple nodes
+
+    private ColumnFamilyStore createWideTable()
+    {
+        StringBuilder sb = new StringBuilder("CREATE TABLE %s (pk text, ck int");
+        for (int i = 0; i < WIDE_COLS; i++)
+            sb.append(", ").append(colName(i)).append(" int");
+        sb.append(", PRIMARY KEY (pk, ck))");
+        createTable(sb.toString());
+        return getCurrentColumnFamilyStore();
+    }
+
+    private static String wideInsert(String tbl, int ck, long timestamp)
+    {
+        StringBuilder names = new StringBuilder("pk, ck");
+        StringBuilder vals = new StringBuilder("'test', ").append(ck);
+        for (int i = 0; i < WIDE_COLS; i++)
+        {
+            names.append(", ").append(colName(i));
+            vals.append(", ").append(i);
+        }
+        return "INSERT INTO " + tbl + " (" + names + ") VALUES (" + vals + ") USING TIMESTAMP " + timestamp;
+    }
+
+    private static String rowDelete(String tbl, int ck, long timestamp)
+    {
+        return "DELETE FROM " + tbl + " USING TIMESTAMP " + timestamp + " WHERE pk = 'test' AND ck = " + ck;
+    }
+
+    // enough elements that each row's collection cell tree spans multiple nodes (branches + leaves), so the freed
+    // internal structure is non-trivial
+    private static final int SET_ELEMENTS = 4 * (BTree.MAX_KEYS + 1);
+
+    private ColumnFamilyStore createCollectionTable()
+    {
+        createTable("CREATE TABLE %s (pk text, ck int, s set<text>, PRIMARY KEY (pk, ck))");
+        return getCurrentColumnFamilyStore();
+    }
+
+    private static String setInsert(String tbl, int ck, long timestamp)
+    {
+        StringBuilder set = new StringBuilder("{");
+        for (int i = 0; i < SET_ELEMENTS; i++)
+        {
+            if (i > 0)
+                set.append(", ");
+            set.append('\'').append(elemName(i)).append('\'');
+        }
+        set.append('}');
+        return "INSERT INTO " + tbl + " (pk, ck, s) VALUES ('test', " + ck + ", " + set + ") USING TIMESTAMP " + timestamp;
+    }
+
+    private static String colName(int i)
+    {
+        return "c" + String.format("%03d", i);
+    }
+
+    private static void assertOwnsNonNegative(ColumnFamilyStore cfs, String step)
+    {
+        for (Memtable mt : cfs.getTracker().getView().getAllMemtables())
+        {
+            Memtable.MemoryUsage usage = Memtable.getMemoryUsage(mt);
+            assertThat(usage.ownsOnHeap)
+            .as("ON-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+            assertThat(usage.ownsOffHeap)
+            .as("OFF-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+        }
+    }
+}

test/unit/org/apache/cassandra/db/partitions/SetCellAccountingTest.java

@@ -0,0 +1,171 @@
+/*
+ * 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.partitions;
+
+import java.lang.reflect.Field;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.Config;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.memtable.Memtable;
+import org.apache.cassandra.utils.btree.BTree;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Runs a grow/reset op mix on one partition's {@code set<text>} column with strictly
+ * increasing timestamps, then asserts the memtable's on/off-heap ownership never goes negative.
+ */
+public class SetCellAccountingTest extends CQLTester
+{
+    private static final Logger logger = LoggerFactory.getLogger(SetCellAccountingTest.class);
+
+    @BeforeClass
+    public static void setUpClass()
+    {
+        ServerTestUtils.daemonInitialization();
+        try
+        {
+            Field confField = DatabaseDescriptor.class.getDeclaredField("conf");
+            confField.setAccessible(true);
+            Config conf = (Config) confField.get(null);
+            conf.memtable_allocation_type = Config.MemtableAllocationType.offheap_objects;
+        }
+        catch (ReflectiveOperationException e)
+        {
+            throw new RuntimeException(e);
+        }
+        CQLTester.prepareServer();
+    }
+
+    private ColumnFamilyStore createTestTable()
+    {
+        createTable("CREATE TABLE %s (" +
+                    "    name text PRIMARY KEY," +
+                    "    last_contact timestamp," +
+                    "    namespace text," +
+                    "    partitioner text," +
+                    "    properties text," +
+                    "    state text," +
+                    "    seed_hosts set<text>)");
+        return getCurrentColumnFamilyStore();
+    }
+
+    @Test
+    public void largeSetGrowShrinkKeepOwnsNonNegative()
+    {
+        ColumnFamilyStore cfs = createTestTable();
+
+        final int setSize = BTree.MAX_KEYS + 1;
+        final int ops = 10_000;
+        final AtomicLong ts = new AtomicLong(1_000_000L);
+
+        for (int op = 0; op < ops; op++)
+        {
+            long t = ts.incrementAndGet();
+            String cql;
+            if (op % 2 == 0)
+            {
+                cql = "UPDATE %s USING TIMESTAMP " + t + " SET seed_hosts = seed_hosts + " +
+                      rangeSet(0, setSize) + " WHERE name = 'test'";
+            }
+            else
+            {
+                cql = "UPDATE %s USING TIMESTAMP " + t + " SET seed_hosts = " +
+                      rangeSet(0, setSize / 2) + " WHERE name = 'test'";
+            }
+
+
+            QueryProcessor.executeInternal(formatQuery(cql));
+            UntypedResultSet rs = QueryProcessor.executeInternal(
+            formatQuery("SELECT seed_hosts FROM %s WHERE name = 'test'"));
+            Set<String> seeds = (rs == null || rs.isEmpty() || !rs.one().has("seed_hosts"))
+                                ? null
+                                : rs.one().getSet("seed_hosts", UTF8Type.instance);
+
+            if (op % 100 == 0)
+                logger.info("== op=" + op +
+                            ", seedsSize= " + (seeds != null ? seeds.size() : "0") +
+                            ", heapSize= " + ownsOnHeapNow(cfs) +
+                            ", offheapSize= " + ownsOffHeapNow(cfs) +
+                            ", seed_hosts=" + seeds);
+            assertOwnsNonNegative(cfs, "after op=" + op);
+
+        }
+        cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+        assertOwnsNonNegative(cfs, "after flush");
+    }
+
+    private static long ownsOnHeapNow(ColumnFamilyStore cfs)
+    {
+        return Memtable.getMemoryUsage(cfs.getTracker().getView().getCurrentMemtable()).ownsOnHeap;
+    }
+
+    private static long ownsOffHeapNow(ColumnFamilyStore cfs)
+    {
+        return Memtable.getMemoryUsage(cfs.getTracker().getView().getCurrentMemtable()).ownsOffHeap;
+    }
+
+
+    /**
+     * {@code {'e00000','e00001',...}} for indices [from,to).
+     */
+    private static String rangeSet(int from, int to)
+    {
+        if (to <= from) return "{}";
+        StringBuilder sb = new StringBuilder("{");
+        for (int i = from; i < to; i++)
+        {
+            if (i > from) sb.append(',');
+            sb.append('\'').append(elemName(i)).append('\'');
+        }
+        return sb.append('}').toString();
+    }
+
+    private static String elemName(int i)
+    {
+        return 'e' + String.format("%05d", i);
+    }
+
+    private static void assertOwnsNonNegative(ColumnFamilyStore cfs, String step)
+    {
+        for (Memtable mt : cfs.getTracker().getView().getAllMemtables())
+        {
+            Memtable.MemoryUsage usage = Memtable.getMemoryUsage(mt);
+            assertThat(usage.ownsOnHeap)
+            .as("ON-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+            assertThat(usage.ownsOffHeap)
+            .as("OFF-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+        }
+    }
+}

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

test/unit/org/apache/cassandra/utils/btree/BTreeUpdateHeapAccountingTest.java

@@ -0,0 +1,276 @@
+/*
+ * 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.utils.btree;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Random;
+import java.util.TreeSet;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.utils.BulkIterator;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Asserts that {@link BTree#update}'s heap accounting -- the total reported through
+ * {@link UpdateFunction#onAllocatedOnHeap(long)} -- matches the actual on-heap structure of the resulting tree, as
+ * measured by {@link BTree#sizeOnHeapOf(Object[])}. {@code onAllocatedOnHeap} reports the extra heap allocated over
+ * the previous tree, so summed over a sequence of updates from empty it must equal {@code sizeOnHeapOf(result)}.
+ * <p>
+ * The scenarios cover the {@code Updater} paths that matter for accounting (coverage verified with JaCoCo): many tiny
+ * disjoint inserts; a large contiguous run inserted in one update (forcing a node to overflow more than once before
+ * draining); re-inserting existing keys (exercising the merge path); and a height-4 tree.
+ */
+public class BTreeUpdateHeapAccountingTest
+{
+    private static final Logger logger = LoggerFactory.getLogger(BTreeUpdateHeapAccountingTest.class);
+
+    private static final Comparator<Integer> CMP = Integer::compare;
+
+    private enum Scenario { SMALL_INSERTS, BLOCK_INSERTS, OVERLAPPING, DEEP }
+
+    /**
+     * A non-simple (i.e. not {@link UpdateFunction.Simple}) update function, so that {@code BTree.update} actually
+     * performs heap accounting. It is an identity merge over the keys, so the only heap it reports is that of the
+     * BTree node arrays themselves -- exactly what {@code sizeOnHeapOf} measures. It also counts merge invocations
+     * so the test can assert the overlapping scenario really exercised the merge path.
+     */
+    private static final class AccountingUpdateFunction implements UpdateFunction<Integer, Integer>
+    {
+        long reported = 0;
+        long merges = 0;
+
+        @Override
+        public Integer insert(Integer insert)
+        {
+            return insert;
+        }
+
+        @Override
+        public Integer merge(Integer replacing, Integer update)
+        {
+            merges++;
+            return update;
+        }
+
+        @Override
+        public void onAllocatedOnHeap(long heapSize)
+        {
+            reported += heapSize;
+        }
+    }
+
+    @Test
+    public void reportedAllocationShouldMatchHeapSizeOfResult()
+    {
+        Trial[] trials = {
+            runTrial(Scenario.SMALL_INSERTS, 1L, 3000, 3),
+            runTrial(Scenario.SMALL_INSERTS, 42L, 2500, 100),
+            runTrial(Scenario.BLOCK_INSERTS, 7L, 12000, 250),
+            runTrial(Scenario.OVERLAPPING, 99L, 4000, 40),
+            runTrial(Scenario.DEEP, 12345L, 40000, 200),
+        };
+
+        List<String> failures = new ArrayList<>();
+        for (Trial r : trials)
+        {
+            logger.info(String.format("%-13s seed=%-6d keys=%-6d reported=%-10d sizeOnHeapOf=%-9d gap=%-10d  [over=+%d/%du, under=-%d/%du]  merges=%d height=%d",
+                                      r.scenario, r.seed, r.distinctKeys, r.reported, r.actualHeap, r.reported - r.actualHeap,
+                                      r.sumOver, r.updatesOver, r.sumUnder, r.updatesUnder, r.merges, r.height));
+            if (r.reported != r.actualHeap)
+                failures.add(String.format("%s (seed=%d): reported=%d but sizeOnHeapOf(result)=%d (net gap=%d; over=+%d, under=-%d; merges=%d)",
+                                            r.scenario, r.seed, r.reported, r.actualHeap, r.reported - r.actualHeap,
+                                            r.sumOver, r.sumUnder, r.merges));
+        }
+
+        assertTrue("BTree.update's onAllocatedOnHeap total does not match BTree.sizeOnHeapOf(result):\n  "
+                   + String.join("\n  ", failures),
+                   failures.isEmpty());
+    }
+
+    private static Trial runTrial(Scenario scenario, long seed, int numKeys, int maxBatch)
+    {
+        Random random = new Random(seed);
+
+        // distinct keys inserted in random order, in random-sized batches; depending on the scenario some batches
+        // are large contiguous runs (to force multi-overflow) and some re-insert existing keys (to force merge).
+        List<Integer> order = new ArrayList<>(numKeys);
+        for (int i = 0; i < numKeys; i++)
+            order.add(i);
+        Collections.shuffle(order, random);
+
+        AccountingUpdateFunction fn = new AccountingUpdateFunction();
+        Object[] tree = BTree.empty();
+        // reference set of the keys the tree should contain, used for the end-of-trial sanity check
+        TreeSet<Integer> model = new TreeSet<>();
+
+        Trial trial = new Trial();
+        int pos = 0; // number of distinct keys consumed from order[] so far, i.e. order[0..pos) are in the tree
+        boolean blockDone = false;
+        // run until every distinct key has been inserted; OVERLAPPING additionally keeps going until the merge path
+        // has been exercised at least 50 times (re-inserts only start producing merges once keys are in the tree)
+        while (pos < numKeys || (scenario == Scenario.OVERLAPPING && fn.merges < 50))
+        {
+            TreeSet<Integer> batchSet = new TreeSet<>(CMP);
+
+            int block = 4000; // large enough to overflow not just a leaf but a whole branch >1x in a single update
+            if (scenario == Scenario.BLOCK_INSERTS && !blockDone && pos > maxBatch && pos < numKeys - block)
+            {
+                // a single update inserting a long contiguous run between two existing keys -- this overflows the
+                // same leaf, and its parent branch, more than once within one update.
+                for (int i = 0; i < block && pos < numKeys; i++)
+                    batchSet.add(order.get(pos++));
+                blockDone = true;
+            }
+            else
+            {
+                int batch = 1 + random.nextInt(maxBatch);
+                for (int i = 0; i < batch; i++)
+                {
+                    // OVERLAPPING re-inserts an already-inserted key (one from order[0..pos)) to drive the merge
+                    // path: roughly one entry in three while fresh keys remain, and every entry once they run out
+                    boolean reInsert = scenario == Scenario.OVERLAPPING && pos > 0
+                                       && (pos >= numKeys || random.nextInt(3) == 0);
+                    if (reInsert)
+                        batchSet.add(order.get(random.nextInt(pos)));
+                    else if (pos < numKeys)
+                        batchSet.add(order.get(pos++));
+                }
+                if (batchSet.isEmpty() && pos < numKeys)
+                    batchSet.add(order.get(pos++));
+            }
+            if (batchSet.isEmpty())
+                break;
+
+            Integer[] keys = batchSet.toArray(new Integer[0]); // already sorted (TreeSet w/ CMP)
+            Object[] insert = BTree.build(BulkIterator.of(keys), keys.length, UpdateFunction.noOp());
+
+            // per-update contract: reported delta == sizeOnHeapOf(after) - sizeOnHeapOf(before)
+            long heapBefore = BTree.sizeOnHeapOf(tree);
+            long reportedBefore = fn.reported;
+            tree = BTree.update(tree, insert, CMP, fn);
+            long err = (fn.reported - reportedBefore) - (BTree.sizeOnHeapOf(tree) - heapBefore);
+            if (err > 0) { trial.updatesOver++; trial.sumOver += err; }
+            else if (err < 0) { trial.updatesUnder++; trial.sumUnder += -err; }
+
+            model.addAll(batchSet);
+        }
+
+        // sanity: the tree must contain exactly the model, in order
+        assertEquals(model.size(), BTree.size(tree));
+        Integer prev = null;
+        int n = 0;
+        for (Integer key : BTree.<Integer>iterable(tree))
+        {
+            if (prev != null)
+                assertTrue("tree not sorted", key > prev);
+            assertTrue("unexpected key " + key, model.contains(key));
+            prev = key;
+            n++;
+        }
+        assertEquals(model.size(), n);
+
+        trial.scenario = scenario;
+        trial.seed = seed;
+        trial.distinctKeys = model.size();
+        trial.reported = fn.reported;
+        trial.merges = fn.merges;
+        trial.actualHeap = BTree.sizeOnHeapOf(tree);
+        trial.height = BTree.depth(tree);
+        return trial;
+    }
+
+    private static final class Trial
+    {
+        Scenario scenario;
+        long seed;
+        int distinctKeys;
+        long reported;
+        long merges;
+        long actualHeap;
+        int height;
+        int updatesOver;
+        long sumOver;
+        int updatesUnder;
+        long sumUnder;
+    }
+
+    /**
+     * Verifies that re-inserting every key that already exists reports a net heap delta of exactly zero,
+     * and that {@code onAllocatedOnHeap} is actually invoked (not silently skipped by an {@code allocated > 0}
+     * guard). The call must happen so that callers accumulating signed deltas see the confirmation, and so
+     * that a negative allocated value cannot permanently disable accounting by crossing the -1 sentinel.
+     */
+    @Test
+    public void netZeroDeltaOnAllocatedOnHeapIsInvoked()
+    {
+        int numKeys = BTree.MAX_KEYS * BTree.MAX_KEYS;
+        Integer[] keys = new Integer[numKeys];
+        for (int i = 0; i < numKeys; i++)
+            keys[i] = i;
+
+        Object[] tree = BTree.build(BulkIterator.of(keys), numKeys, UpdateFunction.noOp());
+        Object[] insert = BTree.build(BulkIterator.of(keys), numKeys, UpdateFunction.noOp());
+
+        // Use a counting variant that records invocation count separately from the accumulated total
+        final long[] invocations = { 0 };
+        final long[] total = { 0 };
+
+        UpdateFunction<Integer, Integer> fn = new UpdateFunction<>()
+        {
+            @Override
+            public Integer insert(Integer i)
+            {
+                return i;
+            }
+
+            @Override
+            public Integer merge(Integer existing, Integer update)
+            {
+                return update;
+            }
+
+            @Override
+            public void onAllocatedOnHeap(long delta)
+            {
+                invocations[0]++;
+                total[0] += delta;
+            }
+        };
+
+        Object[] result = BTree.update(tree, insert, CMP, fn);
+
+        assertSame("full re-insert of identical keys must reuse the existing tree root", tree, result);
+
+        assertEquals("reported net delta must be zero for a full re-insert of identical keys",
+                     0L, total[0]);
+
+        assertTrue("onAllocatedOnHeap must be invoked even for a net-zero update; " +
+                   "skipping it conflates 'tracking disabled' (-1 sentinel) with 'net zero result' (0)",
+                   invocations[0] >= 1);
+    }
+}