Merge 99161b2107 into 172099f4a8

(cherry picked from commit 451246e17aac5b0c00918666a27c7b8e4d992601)

debian/changelog

@@ -2,7 +2,7 @@ cassandra (4.0.21) unstable; urgency=medium
 
   * New release
 
- -- Stefan Miklosovic <smiklosovic@apache.org>  Mon, 13 Jul 2026 13:10:48 +0200
+ -- Stefan Miklosovic <smiklosovic@apache.org>  Tue, 28 Jul 2026 11:37:54 +0200
 
 cassandra (4.0.20) unstable; urgency=medium
 

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

@@ -436,7 +436,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/rows/BTreeRow.java

@@ -520,7 +520,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);
     }
@@ -588,10 +588,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/ColumnData.java

@@ -151,6 +151,7 @@ public abstract class ColumnData
                     }
                     cells = BTree.update(existingTree, updateTree, existingComplex.column.cellComparator(), (UpdateFunction) reconciler);
                 }
+                onAllocatedOnHeap(maxComplexDeletion.unsharedHeapSize() - existingDeletion.unsharedHeapSize());
                 return new ComplexColumnData(existingComplex.column, cells, maxComplexDeletion);
             }
         }
@@ -212,8 +213,28 @@ public abstract class ColumnData
                 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,7 +51,7 @@ 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("", "", "", SetType.getInstance(ByteType.instance, true)), NO_CELLS, new DeletionTime(0, 0)));
+    static final long EMPTY_SIZE = ObjectSizes.measure(new ComplexColumnData(ColumnMetadata.regularColumn("", "", "", SetType.getInstance(ByteType.instance, true)), NO_CELLS, new DeletionTime(0, 0)));
 
     // The cells for 'column' sorted by cell path.
     private final Object[] cells;
@@ -138,7 +138,7 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell<?>>
 
     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/utils/btree/BTree.java

@@ -947,19 +947,6 @@ public class BTree
         return ((int[]) tree[length - 1])[(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.
      *
@@ -2218,6 +2205,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 -- the per-node contribution to {@link #sizeOnHeapOf(Object[])}. The update builders use it to keep
+     * their running {@code allocated} total net: every newly built node adds its shallow heap and every source node it
+     * replaces subtracts its shallow heap, so reused subtrees cancel and the total equals the change in tree heap.
+     */
+    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();
@@ -2416,17 +2420,28 @@ public class BTree
          * @return the root of the constructed tree.
          */
         public Object[] completeBuild()
+        {
+            return completeBuild(null);
+        }
+
+        /**
+         * As {@link #completeBuild()}, but {@code rootUnode} is the source node at this (the root) level, so that an
+         * unchanged root is reused and a rebuilt root releases its predecessor from the running heap total -- the
+         * root-level analogue of the per-child accounting {@link Updater#updateRecursive} performs via drainAndPropagate.
+         */
+        public Object[] completeBuild(Object[] rootUnode)
         {
             LeafOrBranchBuilder level = this;
             while (true)
             {
                 if (!level.hasOverflow())
-                    return level.drain();
+                    return level.drainAndPropagate(rootUnode, null);
 
                 BranchBuilder parent = level.ensureParent();
-                level.drainAndPropagate(null, parent);
+                level.drainAndPropagate(rootUnode, parent);
                 if (level.savedBuffer != null)
                     Arrays.fill(level.savedBuffer, null);
+                rootUnode = null;
                 level = parent;
             }
         }
@@ -2508,6 +2523,7 @@ public class BTree
      */
     private static abstract class LeafBuilder extends LeafOrBranchBuilder
     {
+        static final long DISABLED = Long.MIN_VALUE;
         long allocated;
 
         LeafBuilder()
@@ -2516,6 +2532,12 @@ public class BTree
             buffer = new Object[MAX_KEYS];
         }
 
+        final void addAllocated(long bytes)
+        {
+            if (allocated != DISABLED)
+                allocated += bytes;
+        }
+
         /**
          * Add {@code nextKey} to the buffer, overflowing if necessary
          */
@@ -2673,8 +2695,7 @@ public class BTree
             int newPredecessorCount = predSize - steal;
             Object[] newPredecessor = new Object[newPredecessorCount | 1];
             System.arraycopy(pred, 0, newPredecessor, 0, newPredecessorCount);
-            if (allocated >= 0)
+            addAllocated(ObjectSizes.sizeOfReferenceArray(newPredecessorCount | 1));
-                allocated += ObjectSizes.sizeOfReferenceArray(newPredecessorCount | 1);
             ensureParent().addChildAndNextKey(newPredecessor, newPredecessorCount, pred[newPredecessorCount]);
             return newLeaf;
         }
@@ -2726,8 +2747,7 @@ public class BTree
         {
             // propagate the leaf we have saved in savedBuffer
             // precondition: savedLeafCount == MAX_KEYS
-            if (allocated >= 0)
+            addAllocated(ObjectSizes.sizeOfReferenceArray(MAX_KEYS));
-                allocated += ObjectSizes.sizeOfReferenceArray(MAX_KEYS);
             ensureParent().addChildAndNextKey(savedBuffer, MAX_KEYS, savedNextKey);
             savedBuffer = null;
             savedNextKey = null;
@@ -2749,6 +2769,9 @@ public class BTree
                 // we have too few items, so spread the two buffers across two new nodes
                 leaf = redistributeOverflowAndDrain();
                 sizeOfLeaf = MIN_KEYS;
+                // redistributeAndDrain accounted the propagated predecessor; account this returned leaf and release
+                // the source node it (together with the predecessor) replaces
+                addAllocated(ObjectSizes.sizeOfReferenceArray(MIN_KEYS) - (unode == null ? 0 : sizeOnHeapOfLeaf(unode)));
             }
             else if (!hasOverflow() && unode != null && count == sizeOfLeaf(unode) && areIdentical(buffer, 0, unode, 0, count))
             {
@@ -2764,8 +2787,9 @@ public class BTree
 
                 sizeOfLeaf = count;
                 leaf = drain();
-                if (allocated >= 0 && sizeOfLeaf > 0)
+                // account the new leaf (if any) and release the source it replaces, keeping the total a net delta
-                    allocated += ObjectSizes.sizeOfReferenceArray(sizeOfLeaf | 1) - (unode == null ? 0 : sizeOnHeapOfLeaf(unode));
+                addAllocated((sizeOfLeaf > 0 ? ObjectSizes.sizeOfReferenceArray(sizeOfLeaf | 1) : 0)
+                             - (unode == null ? 0 : sizeOnHeapOfLeaf(unode)));
             }
 
             count = 0;
@@ -2870,10 +2894,10 @@ public class BTree
          */
         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;
@@ -2928,8 +2952,7 @@ 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)
+            leaf.addAllocated(shallowHeapOf(savedBranch));
-                leaf.allocated += ObjectSizes.sizeOfReferenceArray(2 * (1 + savedBranchCount));
             ensureParent().addChildAndNextKey(savedBranch, savedBranchSize, savedBuffer[savedBranchCount]);
             savedNextKey = null;
 
@@ -3012,6 +3035,9 @@ public class BTree
             {
                 branch = redistributeOverflowAndDrain();
                 sizeOfBranch = sizeOfBranch(branch);
+                // redistributeOverflowAndDrain accounted the propagated branch; account this returned branch and
+                // release the source it (together with that branch) replaces
+                leaf.addAllocated(shallowHeapOf(branch) - (unode == null ? 0 : shallowHeapOf(unode)));
             }
             else
             {
@@ -3035,6 +3061,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)));
                 }
             }
 
@@ -3329,7 +3357,7 @@ public class BTree
 
         FastBuilder()
         {
-            allocated = -1;
+            allocated = DISABLED;
         } // disable allocation tracking
 
         public void add(V value)
@@ -3437,7 +3465,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();
             for (int i = 0; i < leafDepth; ++i)
@@ -3446,9 +3474,9 @@ public class BTree
             Insert ik = this.insert.next();
             ik = updateRecursive(ik, update, null, builder);
             assert ik == null;
-            Object[] result = builder.completeBuild();
+            Object[] result = builder.completeBuild(update);
 
-            if (allocated > 0)
+            if (allocated != DISABLED)
                 updateF.onAllocatedOnHeap(allocated);
 
             return result;
@@ -3633,7 +3661,7 @@ public class BTree
 
         AbstractTransformer()
         {
-            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/unit/org/apache/cassandra/db/partitions/PartitionRowAccountingTest.java

@@ -0,0 +1,501 @@
+/*
+ * 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 java.util.function.Supplier;
+
+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.Keyspace;
+import org.apache.cassandra.db.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</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;
+            conf.memtable_cleanup_threshold = 0.8f; // to reduce risk of memtable switch during a test
+        }
+        catch (ReflectiveOperationException e)
+        {
+            throw new RuntimeException(e);
+        }
+        CQLTester.prepareServer();
+    }
+
+    private static void flushAllToBaseline()
+    {
+        for (Keyspace ks : Keyspace.all())
+            for (ColumnFamilyStore c : ks.getColumnFamilyStores())
+                c.forceBlockingFlush();
+    }
+
+    private static final int MAX_ROUND_ATTEMPTS = 10;
+
+    /**
+     * Runs one round of a test -- the writes plus the ownership measurement taken at their end -- and returns that
+     * measurement, guaranteeing it was read from the very memtables the round started writing into.
+     * <p>
+     * Ownership is only comparable while everything the round wrote is still held by the memtable it is measured from,
+     * but a flush can roll a memtable at any moment (memtable pressure, the commitlog, another test's flush). So
+     * instead of failing when that happens we discard the round and repeat it: every attempt starts from a freshly
+     * flushed baseline, and replaying the identical writes (identical timestamps) rebuilds the identical state in the
+     * new memtables. The measurement is taken inside {@code round}, before the memtables are re-checked, so a roll
+     * between the writes and the measurement is caught too.
+     */
+    private <T> T onStableMemtables(Supplier<T> round, ColumnFamilyStore... cfss)
+    {
+        for (int attempt = 1; attempt <= MAX_ROUND_ATTEMPTS; attempt++)
+        {
+            flushAllToBaseline();
+            Memtable[] started = currentMemtables(cfss);
+
+            T measurement = round.get();
+
+            if (stillCurrent(started, cfss))
+                return measurement;
+
+            logger.info("attempt " + attempt + '/' + MAX_ROUND_ATTEMPTS +
+                        " discarded: a memtable rolled (flushed) mid-round, repeating the round");
+        }
+        throw new AssertionError("a memtable rolled (flushed) mid-round in each of the " + MAX_ROUND_ATTEMPTS +
+                                 " attempts -- ownership measurement never valid");
+    }
+
+    private static Memtable[] currentMemtables(ColumnFamilyStore... cfss)
+    {
+        Memtable[] memtables = new Memtable[cfss.length];
+        for (int i = 0; i < cfss.length; i++)
+            memtables[i] = cfss[i].getTracker().getView().getCurrentMemtable();
+        return memtables;
+    }
+
+    private static boolean stillCurrent(Memtable[] started, ColumnFamilyStore... cfss)
+    {
+        for (int i = 0; i < cfss.length; i++)
+            if (started[i] != cfss[i].getTracker().getView().getCurrentMemtable())
+                return false;
+        return true;
+    }
+
+    /**
+     * The on/off-heap ownership of the retain- and removeShadowed-path memtables, captured in one measurement.
+     */
+    private static class Owns
+    {
+        final long onHeapRetain, offHeapRetain, onHeapShadow, offHeapShadow;
+
+        Owns(ColumnFamilyStore cfsRetain, ColumnFamilyStore cfsShadow)
+        {
+            onHeapRetain = ownsOnHeapNow(cfsRetain);
+            offHeapRetain = ownsOffHeapNow(cfsRetain);
+            onHeapShadow = ownsOnHeapNow(cfsShadow);
+            offHeapShadow = ownsOffHeapNow(cfsShadow);
+        }
+    }
+
+    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();
+        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
+
+        Owns owns = onStableMemtables(() -> {
+            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));
+            }
+            return new Owns(cfsRetain, cfsShadow);
+        }, cfsRetain, cfsShadow);
+
+        long onHeapRetain = owns.onHeapRetain, onHeapShadow = owns.onHeapShadow;
+        logger.info("retain-path onHeap=" + onHeapRetain + " offHeap=" + owns.offHeapRetain +
+                    "; removeShadowed-path onHeap=" + onHeapShadow + " offHeap=" + owns.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
+
+        Owns owns = onStableMemtables(() -> {
+            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));
+            }
+            return new Owns(cfsRetain, cfsShadow);
+        }, cfsRetain, cfsShadow);
+
+        long onHeapRetain = owns.onHeapRetain, onHeapShadow = owns.onHeapShadow;
+        logger.info("collection retain-path onHeap=" + onHeapRetain + " offHeap=" + owns.offHeapRetain +
+                    "; removeShadowed-path onHeap=" + onHeapShadow + " offHeap=" + owns.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
+
+        Owns owns = onStableMemtables(() -> {
+            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));
+            }
+            return new Owns(cfsRetain, cfsShadow);
+        }, cfsRetain, cfsShadow);
+
+        long onHeapRetain = owns.onHeapRetain, onHeapShadow = owns.onHeapShadow;
+        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 cfs.getTracker().getView().getCurrentMemtable().getAllocator().onHeap().owns();
+    }
+
+    private static long ownsOffHeapNow(ColumnFamilyStore cfs)
+    {
+        return cfs.getTracker().getView().getCurrentMemtable().getAllocator().offHeap().owns();
+    }
+
+    /**
+     * 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())
+        {
+            assertThat(mt.getAllocator().onHeap().owns())
+            .as("ON-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+            assertThat(mt.getAllocator().offHeap().owns())
+            .as("OFF-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+        }
+    }
+}

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

@@ -0,0 +1,170 @@
+/*
+ * 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.Memtable;
+import org.apache.cassandra.db.marshal.UTF8Type;
+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();
+        assertOwnsNonNegative(cfs, "after flush");
+    }
+
+    private static long ownsOnHeapNow(ColumnFamilyStore cfs)
+    {
+        return cfs.getTracker().getView().getCurrentMemtable().getAllocator().onHeap().owns();
+    }
+
+    private static long ownsOffHeapNow(ColumnFamilyStore cfs)
+    {
+        return cfs.getTracker().getView().getCurrentMemtable().getAllocator().offHeap().owns();
+    }
+
+
+    /**
+     * {@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())
+        {
+            assertThat(mt.getAllocator().onHeap().owns())
+            .as("ON-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+            assertThat(mt.getAllocator().offHeap().owns())
+            .as("OFF-heap owns went NEGATIVE [" + step + "]")
+            .isGreaterThanOrEqualTo(0L);
+        }
+    }
+}

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<Integer, Integer>()
+        {
+            @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);
+    }
+}