CASSANDRA-21605: Improve algorithmic complexity for finding intersecting SSTables in LCS

src/java/org/apache/cassandra/db/compaction/LeveledGenerations.java

@@ -27,6 +27,7 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
+import java.util.NavigableSet;
 import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 
@@ -39,6 +40,8 @@
 
 import org.apache.cassandra.io.sstable.SSTableIdFactory;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.dht.Bounds;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_STRICT_LCS_CHECKS;
@@ -245,6 +248,59 @@ Set<SSTableReader> allSSTables()
         return builder.build();
     }
 
+    /**
+     * Returns all SSTables in the given level that overlap with the provided sstable.
+     * For levels >= 1, SSTables are strictly disjoint and ordered in a TreeSet, enabling
+     * an O(log N + K) binary search instead of an O(N) linear scan across the entire generation.
+     */
+    Set<SSTableReader> getOverlapping(int level, SSTableReader sstable)
+    {
+        if (level >= levelCount() || level < 0)
+            throw new ArrayIndexOutOfBoundsException("Invalid generation " + level + " - maximum is " + (levelCount() - 1));
+
+        if (level == 0)
+        {
+            Set<SSTableReader> overlapped = new HashSet<>();
+            Bounds<Token> targetBounds = new Bounds<>(sstable.getFirst().getToken(), sstable.getLast().getToken());
+            for (SSTableReader candidate : l0)
+            {
+                Bounds<Token> candidateBounds = new Bounds<>(candidate.getFirst().getToken(), candidate.getLast().getToken());
+                if (candidateBounds.intersects(targetBounds))
+                    overlapped.add(candidate);
+            }
+            return overlapped;
+        }
+
+        TreeSet<SSTableReader> sortedLevel = levels[level - 1];
+        if (sortedLevel.isEmpty())
+            return Collections.emptySet();
+
+        Set<SSTableReader> overlapped = new HashSet<>();
+        Token start = sstable.getFirst().getToken();
+        Token end = sstable.getLast().getToken();
+        Bounds<Token> targetBounds = new Bounds<>(start, end);
+
+        SSTableReader floor = sortedLevel.floor(sstable);
+        if (floor != null)
+        {
+            Bounds<Token> floorBounds = new Bounds<>(floor.getFirst().getToken(), floor.getLast().getToken());
+            if (floorBounds.intersects(targetBounds))
+                overlapped.add(floor);
+        }
+
+        NavigableSet<SSTableReader> tail = floor != null ? sortedLevel.tailSet(floor, false) : sortedLevel;
+        for (SSTableReader candidate : tail)
+        {
+            if (candidate.getFirst().getToken().compareTo(end) > 0)
+                break;
+
+            Bounds<Token> candidateBounds = new Bounds<>(candidate.getFirst().getToken(), candidate.getLast().getToken());
+            if (candidateBounds.intersects(targetBounds))
+                overlapped.add(candidate);
+        }
+        return overlapped;
+    }
+
     /**
      * given a level with sstables with first tokens [0, 10, 20, 30] and a lastCompactedSSTable with last = 15, we will
      * return an Iterator over [20, 30, 0, 10].

src/java/org/apache/cassandra/db/compaction/LeveledManifest.java

@@ -575,12 +575,11 @@ private Collection<SSTableReader> getCandidatesFor(int level)
 
         // look for a non-suspect keyspace to compact with, starting with where we left off last time,
         // and wrapping back to the beginning of the generation if necessary
-        Map<SSTableReader, Bounds<Token>> sstablesNextLevel = genBounds(generations.get(level + 1));
         Iterator<SSTableReader> levelIterator = generations.wrappingIterator(level, lastCompactedSSTables[level]);
         while (levelIterator.hasNext())
         {
             SSTableReader sstable = levelIterator.next();
-            Set<SSTableReader> candidates = Sets.union(Collections.singleton(sstable), overlappingWithBounds(sstable, sstablesNextLevel));
+            Set<SSTableReader> candidates = Sets.union(Collections.singleton(sstable), generations.getOverlapping(level + 1, sstable));
 
             if (Iterables.any(candidates, SSTableReader::isMarkedSuspect))
                 continue;

test/unit/org/apache/cassandra/db/compaction/LeveledGenerationsTest.java

@@ -21,6 +21,7 @@
 import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Set;
 
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
@@ -41,6 +42,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.fail;
+import static org.junit.Assert.assertTrue;
 
 public class LeveledGenerationsTest extends CQLTester
 {
@@ -163,6 +165,65 @@ public void testFillLevels()
         {}
     }
 
+    @Test
+    public void testGetOverlapping()
+    {
+        ColumnFamilyStore cfs = MockSchema.newCFS();
+        LeveledGenerations gens = new LeveledGenerations();
+
+        // Populate Level 2 with disjoint sstables: [0, 10], [20, 30], [40, 50], [60, 70], [80, 90]
+        SSTableReader s1 = MockSchema.sstable(1, 5, true, 0, 10, 2, cfs);
+        SSTableReader s2 = MockSchema.sstable(2, 5, true, 20, 30, 2, cfs);
+        SSTableReader s3 = MockSchema.sstable(3, 5, true, 40, 50, 2, cfs);
+        SSTableReader s4 = MockSchema.sstable(4, 5, true, 60, 70, 2, cfs);
+        SSTableReader s5 = MockSchema.sstable(5, 5, true, 80, 90, 2, cfs);
+        gens.addAll(Lists.newArrayList(s1, s2, s3, s4, s5));
+
+        // 1. Overlap across multiple tables [25, 65]: should find s2 [20, 30], s3 [40, 50], s4 [60, 70]
+        SSTableReader targetMulti = MockSchema.sstable(10, 5, true, 25, 65, 1, cfs);
+        Set<SSTableReader> overlapping = gens.getOverlapping(2, targetMulti);
+        assertEquals(3, overlapping.size());
+        assertTrue(overlapping.contains(s2));
+        assertTrue(overlapping.contains(s3));
+        assertTrue(overlapping.contains(s4));
+        assertFalse(overlapping.contains(s1));
+        assertFalse(overlapping.contains(s5));
+
+        // 2. Exact match with s3 [40, 50]: should find only s3
+        SSTableReader targetExact = MockSchema.sstable(11, 5, true, 40, 50, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetExact);
+        assertEquals(Collections.singleton(s3), overlapping);
+
+        // 3. In gap between s2 and s3 [32, 38]: should find 0
+        SSTableReader targetGap = MockSchema.sstable(12, 5, true, 32, 38, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetGap);
+        assertTrue(overlapping.isEmpty());
+
+        // 4. Completely before all tables [-50, -10]: should find 0
+        SSTableReader targetBefore = MockSchema.sstable(13, 5, true, -50, -10, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetBefore);
+        assertTrue(overlapping.isEmpty());
+
+        // 5. Completely after all tables [100, 150]: should find 0
+        SSTableReader targetAfter = MockSchema.sstable(14, 5, true, 100, 150, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetAfter);
+        assertTrue(overlapping.isEmpty());
+
+        // 6. Overlap with predecessor floor only [-5, 5]: should find s1 [0, 10]
+        SSTableReader targetFloor = MockSchema.sstable(15, 5, true, -5, 5, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetFloor);
+        assertEquals(Collections.singleton(s1), overlapping);
+
+        // 7. Overlap with successor only [85, 95]: should find s5 [80, 90]
+        SSTableReader targetSuccessor = MockSchema.sstable(16, 5, true, 85, 95, 1, cfs);
+        overlapping = gens.getOverlapping(2, targetSuccessor);
+        assertEquals(Collections.singleton(s5), overlapping);
+
+        // 8. Empty level: should return empty set
+        overlapping = gens.getOverlapping(3, targetExact);
+        assertTrue(overlapping.isEmpty());
+    }
+
     private void assertIter(Iterator<SSTableReader> iter, long first, long last, int expectedCount)
     {
         List<SSTableReader> drained = Lists.newArrayList(iter);