Optimize RangeTombstoneList copying with copy-on-write array sharing (CASSANDRA-21492)

CHANGES.txt

@@ -1,5 +1,6 @@
 7.0
  * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975)
+ * Optimize RangeTombstoneList copying with copy-on-write array sharing (CASSANDRA-21492)
  * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394)
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)

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

@@ -48,6 +48,13 @@
  * <p>
  * The only use of the local deletion time is to know when a given tombstone can
  * be purged, which will be done by the purge() method.
+ * <p>
+ * <b>Copy-on-Write (COW) Invariant:</b>
+ * This class implements a copy-on-write optimization. The {@link #copy()} method creates a new instance 
+ * that shares the four parallel backing arrays ({@code starts}, {@code ends}, {@code markedAts}, 
+ * and {@code delTimesUnsignedIntegers}) with the original instance and marks both instances as shared ({@code shared = true}).
+ * To ensure safe copy independence, any operation that performs in-place mutation of these backing arrays 
+ * MUST call {@link #isolate()} before writing to them for the first time. All backing-array writes are guarded this way.
  */
 public class RangeTombstoneList implements Iterable<RangeTombstone>, IMeasurableMemory
 {
@@ -64,6 +71,19 @@ public class RangeTombstoneList implements Iterable<RangeTombstone>, IMeasurable
 
     private long boundaryHeapSize;
     private int size;
+    private boolean shared;
+
+    private void isolate()
+    {
+        if (shared)
+        {
+            starts = Arrays.copyOf(starts, starts.length);
+            ends = Arrays.copyOf(ends, ends.length);
+            markedAts = Arrays.copyOf(markedAts, markedAts.length);
+            delTimesUnsignedIntegers = Arrays.copyOf(delTimesUnsignedIntegers, delTimesUnsignedIntegers.length);
+            shared = false;
+        }
+    }
 
     private RangeTombstoneList(ClusteringComparator comparator,
                                ClusteringBound<?>[] starts,
@@ -103,14 +123,28 @@ public ClusteringComparator comparator()
         return comparator;
     }
 
+    /**
+     * Creates a copy-on-write duplicate of this list.
+     * <p>
+     * This method shares the four backing arrays ({@code starts}, {@code ends}, {@code markedAts}, and
+     * {@code delTimesUnsignedIntegers}) between the original list and the returned copy, marking both instances
+     * as shared ({@code shared = true}). To maintain independent states, any subsequent in-place mutation on
+     * either instance must first trigger {@link #isolate()} to copy and isolate the backing arrays before writing.
+     *
+     * @return a copy-on-write copy of this list.
+     */
     public RangeTombstoneList copy()
     {
-        return new RangeTombstoneList(comparator,
-                                      Arrays.copyOf(starts, size),
-                                      Arrays.copyOf(ends, size),
-                                      Arrays.copyOf(markedAts, size),
-                                      Arrays.copyOf(delTimesUnsignedIntegers, size),
-                                      boundaryHeapSize, size);
+        this.shared = true;
+        RangeTombstoneList copy = new RangeTombstoneList(comparator,
+                                                         starts,
+                                                         ends,
+                                                         markedAts,
+                                                         delTimesUnsignedIntegers,
+                                                         boundaryHeapSize,
+                                                         size);
+        copy.shared = true;
+        return copy;
     }
 
     public RangeTombstoneList clone(ByteBufferCloner cloner)
@@ -188,7 +222,14 @@ public void addAll(RangeTombstoneList tombstones)
 
         if (isEmpty())
         {
-            copyArrays(tombstones, this);
+            tombstones.shared = true;
+            this.starts = tombstones.starts;
+            this.ends = tombstones.ends;
+            this.markedAts = tombstones.markedAts;
+            this.delTimesUnsignedIntegers = tombstones.delTimesUnsignedIntegers;
+            this.size = tombstones.size;
+            this.boundaryHeapSize = tombstones.boundaryHeapSize;
+            this.shared = true;
             return;
         }
 
@@ -324,12 +365,14 @@ public void collectStats(EncodingStats.Collector collector)
 
     public void updateAllTimestamp(long timestamp)
     {
+        isolate();
         for (int i = 0; i < size; i++)
             markedAts[i] = timestamp;
     }
 
     public void updateAllTimestampAndLocalDeletionTime(long timestamp, long localDeletionTime)
     {
+        isolate();
         int unsignedLocalDeletionTime = Cell.deletionTimeLongToUnsignedInteger(localDeletionTime);
         for (int i = 0; i < size; i++)
         {
@@ -527,17 +570,6 @@ public final int hashCode()
         return result;
     }
 
-    private static void copyArrays(RangeTombstoneList src, RangeTombstoneList dst)
-    {
-        dst.grow(src.size);
-        System.arraycopy(src.starts, 0, dst.starts, 0, src.size);
-        System.arraycopy(src.ends, 0, dst.ends, 0, src.size);
-        System.arraycopy(src.markedAts, 0, dst.markedAts, 0, src.size);
-        System.arraycopy(src.delTimesUnsignedIntegers, 0, dst.delTimesUnsignedIntegers, 0, src.size);
-        dst.size = src.size;
-        dst.boundaryHeapSize = src.boundaryHeapSize;
-    }
-
     /*
      * Inserts a new element starting at index i. This method assumes that:
      *    ends[i-1] <= start < ends[i]
@@ -552,6 +584,7 @@ private static void copyArrays(RangeTombstoneList src, RangeTombstoneList dst)
      */
     private void insertFrom(int i, ClusteringBound<?> start, ClusteringBound<?> end, long markedAt, int delTimeUnsignedInternal)
     {
+        isolate();
         while (i < size)
         {
             assert start.isStart() && end.isEnd();
@@ -677,8 +710,12 @@ private void addInternal(int i, ClusteringBound<?> start, ClusteringBound<?> end
 
         if (size == capacity())
             growToFree(i);
-        else if (i < size)
-            moveElements(i);
+        else
+        {
+            isolate();
+            if (i < size)
+                moveElements(i);
+        }
 
         setInternal(i, start, end, markedAt, delTimeUnsignedInteger);
         size++;
@@ -698,21 +735,13 @@ private void growToFree(int i)
         grow(i, newLength);
     }
 
-    /*
-     * Grow the arrays to match newLength capacity.
-     */
-    private void grow(int newLength)
-    {
-        if (capacity() < newLength)
-            grow(-1, newLength);
-    }
-
     private void grow(int i, int newLength)
     {
         starts = grow(starts, size, newLength, i);
         ends = grow(ends, size, newLength, i);
         markedAts = grow(markedAts, size, newLength, i);
         delTimesUnsignedIntegers = grow(delTimesUnsignedIntegers, size, newLength, i);
+        shared = false;
     }
 
     private static ClusteringBound<?>[] grow(ClusteringBound<?>[] a, int size, int newLength, int i)

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

@@ -0,0 +1,87 @@
+/*
+ * 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.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.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+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.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ClusteringComparator;
+import org.apache.cassandra.db.DeletionTime;
+import org.apache.cassandra.db.RangeTombstone;
+import org.apache.cassandra.db.RangeTombstoneList;
+import org.apache.cassandra.db.Slice;
+import org.apache.cassandra.db.marshal.Int32Type;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1, jvmArgsAppend = "-Xmx512M")
+@Threads(1)
+@State(Scope.Benchmark)
+public class RangeTombstoneListBench
+{
+    private ClusteringComparator comparator;
+    private RangeTombstoneList existing;
+    private RangeTombstone tombstone;
+
+    @Param({"10", "100", "1000"})
+    private int size;
+
+    @Setup(Level.Trial)
+    public void setup()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        comparator = new ClusteringComparator(Int32Type.instance);
+        existing = new RangeTombstoneList(comparator, size);
+        for (int i = 0; i < size; i++)
+        {
+            existing.add(new RangeTombstone(Slice.make(comparator.make(i * 2), comparator.make(i * 2 + 1)), DeletionTime.build(1, 1)));
+        }
+        tombstone = new RangeTombstone(Slice.make(comparator.make(100000), comparator.make(100001)), DeletionTime.build(1, 1));
+    }
+
+    @Benchmark
+    public RangeTombstoneList benchCopyOnly()
+    {
+        return existing.copy();
+    }
+
+    @Benchmark
+    public RangeTombstoneList benchCopyAndAdd()
+    {
+        RangeTombstoneList copy = existing.copy();
+        copy.add(tombstone);
+        return copy;
+    }
+}

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

@@ -0,0 +1,83 @@
+/*
+ * 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.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.Level;
+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.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ClusteringComparator;
+import org.apache.cassandra.db.DeletionTime;
+import org.apache.cassandra.db.RangeTombstone;
+import org.apache.cassandra.db.RangeTombstoneList;
+import org.apache.cassandra.db.Slice;
+import org.apache.cassandra.db.marshal.Int32Type;
+
+@BenchmarkMode(Mode.AverageTime)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1, jvmArgsAppend = "-Xmx512M")
+@Threads(1)
+@State(Scope.Benchmark)
+public class RangeTombstoneListSize10Bench
+{
+    private ClusteringComparator comparator;
+    private RangeTombstoneList existing;
+    private RangeTombstone tombstone;
+
+    @Setup(Level.Trial)
+    public void setup()
+    {
+        DatabaseDescriptor.daemonInitialization();
+        comparator = new ClusteringComparator(Int32Type.instance);
+        existing = new RangeTombstoneList(comparator, 10);
+        for (int i = 0; i < 10; i++)
+        {
+            existing.add(new RangeTombstone(Slice.make(comparator.make(i * 2), comparator.make(i * 2 + 1)), DeletionTime.build(1, 1)));
+        }
+        tombstone = new RangeTombstone(Slice.make(comparator.make(100000), comparator.make(100001)), DeletionTime.build(1, 1));
+    }
+
+    @Benchmark
+    public RangeTombstoneList benchCopyOnly()
+    {
+        return existing.copy();
+    }
+
+    @Benchmark
+    public RangeTombstoneList benchCopyAndAdd()
+    {
+        RangeTombstoneList copy = existing.copy();
+        copy.add(tombstone);
+        return copy;
+    }
+}