support copyAndAddIntervals in IntervalTree for faster replaceFlushed View update

CHANGES.txt

@@ -1,4 +1,5 @@
 4.1.8
+ * support copyAndAddIntervals in IntervalTree for faster replaceFlushed View update (CASSANDRA-20164)
  * Add nodetool checktokenmetadata command that checks TokenMetadata is insync with Gossip endpointState (CASSANDRA-18758)
  * Backport Java 11 support for Simulator (CASSANDRA-17178/CASSANDRA-19935)
  * Equality check for Paxos.Electorate should not depend on collection types (CASSANDRA-19935)

src/java/org/apache/cassandra/db/lifecycle/SSTableIntervalTree.java

@@ -40,6 +40,11 @@ public class SSTableIntervalTree extends IntervalTree<PartitionPosition, SSTable
         super(intervals);
     }
 
+    SSTableIntervalTree(int count, IntervalNode head)
+    {
+        super(count, head);
+    }
+
     public static SSTableIntervalTree empty()
     {
         return EMPTY;
@@ -50,6 +55,14 @@ public static SSTableIntervalTree build(Iterable<SSTableReader> sstables)
         return new SSTableIntervalTree(buildIntervals(sstables));
     }
 
+    public static SSTableIntervalTree addSSTables(SSTableIntervalTree tree, Iterable<SSTableReader> sstables)
+    {
+        List<Interval<PartitionPosition, SSTableReader>> intervals = buildIntervals(sstables);
+        SSTableIntervalTree newTree = new SSTableIntervalTree(tree.count + intervals.size(),
+                                                              tree.head.copyAndAddIntervals(intervals));
+        return newTree;
+    }
+
     public static List<Interval<PartitionPosition, SSTableReader>> buildIntervals(Iterable<SSTableReader> sstables)
     {
         List<Interval<PartitionPosition, SSTableReader>> intervals = new ArrayList<>(Iterables.size(sstables));

src/java/org/apache/cassandra/db/lifecycle/View.java

@@ -336,9 +336,18 @@ public View apply(View view)
                     return new View(view.liveMemtables, flushingMemtables, view.sstablesMap,
                                     view.compactingMap, view.intervalTree);
 
+                // here we're adding sstables only
                 Map<SSTableReader, SSTableReader> sstableMap = replace(view.sstablesMap, emptySet(), flushed);
-                return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap,
-                                SSTableIntervalTree.build(sstableMap.keySet()));
+                if (!view.intervalTree.isEmpty())
+                {
+                    return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap,
+                                    SSTableIntervalTree.addSSTables(view.intervalTree, flushed));
+                }
+                else
+                {
+                    return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap,
+                                    SSTableIntervalTree.build(sstableMap.keySet()));
+                }
             }
         };
     }

src/java/org/apache/cassandra/utils/IntervalTree.java

@@ -41,8 +41,14 @@ public class IntervalTree<C extends Comparable<? super C>, D, I extends Interval
     @SuppressWarnings("unchecked")
     private static final IntervalTree EMPTY_TREE = new IntervalTree(null);
 
-    private final IntervalNode head;
-    private final int count;
+    protected final IntervalNode head;
+    protected final int count;
+
+    protected IntervalTree(int count, IntervalNode head)
+    {
+        this.head = head;
+        this.count = count;
+    }
 
     protected IntervalTree(Collection<I> intervals)
     {
@@ -142,7 +148,7 @@ public final int hashCode()
         return result;
     }
 
-    private class IntervalNode
+    protected class IntervalNode
     {
         final C center;
         final C low;
@@ -217,6 +223,17 @@ else if (candidate.min.compareTo(center) > 0)
             }
         }
 
+        public IntervalNode(C center, C low, C high, List<I> intersectsLeft, List<I> intersectsRight, IntervalNode left, IntervalNode right)
+        {
+            this.center = center;
+            this.low = low;
+            this.high = high;
+            this.intersectsLeft = intersectsLeft;
+            this.intersectsRight = intersectsRight;
+            this.left = left;
+            this.right = right;
+        }
+
         void searchInternal(Interval<C, D> searchInterval, List<D> results)
         {
             if (center.compareTo(searchInterval.min) < 0)
@@ -256,6 +273,54 @@ else if (center.compareTo(searchInterval.max) > 0)
                     right.searchInternal(searchInterval, results);
             }
         }
+
+        public IntervalNode copyAndAddIntervals(List<I> intervals)
+        {
+            return copyAndAddIntervalsHelper(this, intervals);
+        }
+
+        private IntervalNode copyAndAddIntervalsHelper(IntervalNode root, List<I> intervals)
+        {
+            if (intervals.isEmpty())
+                return root;
+            if (root == null)
+                return new IntervalNode(intervals);
+
+            List<I> leftSegment = new ArrayList<>();
+            List<I> rightSegment = new ArrayList<>();
+            C newLow = root.low;
+            C newHigh = root.high;
+            List<I> newIntersectsLeft = new ArrayList<>(root.intersectsLeft);
+            List<I> newIntersectsRight = new ArrayList<>(root.intersectsRight);
+            for (I i : intervals)
+            {
+                newLow = newLow.compareTo(i.min) < 0 ? newLow : i.min;
+                newHigh = newHigh.compareTo(i.max) > 0 ? newHigh : i.max;
+                if (i.max.compareTo(root.center) < 0)
+                {
+                    leftSegment.add(i);
+                }
+                else if (i.min.compareTo(root.center) > 0)
+                {
+                    rightSegment.add(i);
+                }
+                else
+                {
+                    int leftIdx = Interval.<C, D>minOrdering().binarySearchAsymmetric(newIntersectsLeft, i.min, Op.CEIL);
+                    newIntersectsLeft.add(leftIdx, i);
+
+                    int rightIdx = Interval.<C, D>maxOrdering().binarySearchAsymmetric(newIntersectsRight, i.max, Op.HIGHER);
+                    newIntersectsRight.add(rightIdx, i);
+                }
+            }
+            return new IntervalNode(root.center,
+                                    newLow,
+                                    newHigh,
+                                    newIntersectsLeft,
+                                    newIntersectsRight,
+                                    copyAndAddIntervalsHelper(root.left, leftSegment),
+                                    copyAndAddIntervalsHelper(root.right, rightSegment));
+        }
     }
 
     private class TreeIterator extends AbstractIterator<I>

test/unit/org/apache/cassandra/utils/IntervalTreeTest.java

@@ -195,4 +195,38 @@ public long serializedSize(String v)
 
         assertEquals(intervals, intervals2);
     }
+
+    @Test
+    public void testCopyAndAddIntervals()
+    {
+        List<Interval<Integer, Void>> intervals = new ArrayList<Interval<Integer, Void>>();
+
+        intervals.add(Interval.<Integer, Void>create(-300, -200));
+        intervals.add(Interval.<Integer, Void>create(-3, -2));
+        intervals.add(Interval.<Integer, Void>create(1, 2));
+        intervals.add(Interval.<Integer, Void>create(3, 6));
+        intervals.add(Interval.<Integer, Void>create(2, 4));
+        intervals.add(Interval.<Integer, Void>create(5, 7));
+        intervals.add(Interval.<Integer, Void>create(4, 6));
+        intervals.add(Interval.<Integer, Void>create(15, 20));
+        intervals.add(Interval.<Integer, Void>create(49, 60));
+
+
+        IntervalTree<Integer, Void, Interval<Integer, Void>> it = IntervalTree.build(intervals);
+
+        List<Interval<Integer, Void>> intervalsToAdd = new ArrayList<>();
+        intervalsToAdd.add(Interval.create(1, 3));
+        intervalsToAdd.add(Interval.create(8, 9));
+        intervalsToAdd.add(Interval.create(40, 50));
+
+        it = new IntervalTree<>(it.count, it.head.copyAndAddIntervals(intervalsToAdd));
+
+        assertEquals(3, it.search(Interval.<Integer, Void>create(4, 4)).size());
+        assertEquals(4, it.search(Interval.<Integer, Void>create(4, 5)).size());
+        assertEquals(7, it.search(Interval.<Integer, Void>create(-1, 10)).size());
+        assertEquals(0, it.search(Interval.<Integer, Void>create(-1, -1)).size());
+        assertEquals(5, it.search(Interval.<Integer, Void>create(1, 4)).size());
+        assertEquals(2, it.search(Interval.<Integer, Void>create(0, 1)).size());
+        assertEquals(0, it.search(Interval.<Integer, Void>create(10, 12)).size());
+    }
 }