CNDB-15594:Add AbstractBounds#intersects and fix Range#intersects to handle full range

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

@@ -24,7 +24,6 @@
 import org.apache.cassandra.db.PartitionPosition;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.SSTableWriter;
 
 public interface ShardTracker
@@ -71,4 +70,9 @@ default void applyTokenSpaceCoverage(SSTableWriter writer)
         if (writer.first != null)
             writer.setTokenSpaceCoverage(rangeSpanned(writer.first, writer.last));
     }
+
+    static ShardTracker createPreassignedBoundaries(Token[] sortedTokens)
+    {
+        return new SimpleShardTracker(sortedTokens);
+    }
 }

src/java/org/apache/cassandra/dht/AbstractBounds.java

@@ -147,6 +147,22 @@ private String format(T value, AbstractType<?> keyValidator)
     public abstract boolean isStartInclusive();
     public abstract boolean isEndInclusive();
 
+    public boolean intersects(AbstractBounds<T> other)
+    {
+        // If one is a Range, it may be wraparound, thus we must defer to its implementation of intersects.
+        if (other instanceof Range)
+            return other.intersects(this);
+
+        int cmp = other.right.isMinimum() ? -1 : left.compareTo(other.right);
+        if (cmp > 0 || (cmp == 0 && (!inclusiveLeft() || !other.inclusiveRight())))
+            return false;
+        cmp = right.isMinimum() ? 1 : right.compareTo(other.left);
+        if (cmp < 0 || (cmp == 0 && (!inclusiveRight() || !other.inclusiveLeft())))
+            return false;
+
+        return true;
+    }
+
     public abstract AbstractBounds<T> withNewRight(T newRight);
 
     public static class AbstractBoundsSerializer<T extends RingPosition<T>> implements IPartitionerDependentSerializer<AbstractBounds<T>>

src/java/org/apache/cassandra/dht/Range.java

@@ -118,6 +118,7 @@ public boolean intersects(Range<T> that)
         return intersectionWith(that).size() > 0;
     }
 
+    @Override
     public boolean intersects(AbstractBounds<T> that)
     {
         // implemented for cleanup compaction membership test, so only Range + Bounds are supported for now
@@ -164,7 +165,12 @@ public boolean intersects(Bounds<T> that)
         // Same punishment than in Bounds.contains(), we must be carefull if that.left == that.right as
         // as new Range<T>(that.left, that.right) will then cover the full ring which is not what we
         // want.
-        return contains(that.left) || (!that.left.equals(that.right) && intersects(new Range<T>(that.left, that.right)));
+        if (contains(that.left))
+            return true;
+        else if (that.left.equals(that.right))
+            return that.right.isMinimum(); // [x, x] denotes a single point only, but [min, min] covers the full token range
+        else
+            return intersects(new Range<T>(that.left, that.right));
     }
 
     public static boolean intersects(Iterable<Range<Token>> l, Iterable<Range<Token>> r)

test/unit/org/apache/cassandra/dht/AbstractBoundsQuickTest.java

@@ -0,0 +1,120 @@
+/*
+ * 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.dht;
+
+import org.junit.Test;
+
+import org.quicktheories.core.Gen;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.quicktheories.QuickTheory.qt;
+import static org.quicktheories.generators.SourceDSL.integers;
+import static org.quicktheories.generators.SourceDSL.longs;
+
+public class AbstractBoundsQuickTest
+{
+    private static final long MAX_TOKEN = Murmur3Partitioner.MAXIMUM;
+
+    @Test
+    public void testIntersects()
+    {
+        qt().forAll(bounds(), bounds())
+            .check((r1, r2) -> {
+                boolean intersects = r1.intersects(r2);
+                // Check commutativity
+                assertThat(r2.intersects(r1)).isEqualTo(intersects);
+
+                assertThat(intersects).isEqualTo(stupidIntersects(r1, r2));
+
+                return true;
+            });
+    }
+
+    boolean stupidIntersects(AbstractBounds<Token> l, AbstractBounds<Token> r)
+    {
+        if (isPoint(l))
+            return l.left.isMinimum() || r.contains(l.left);
+        if (isPoint(r))
+            return r.left.isMinimum() || l.contains(r.left);
+
+        // Range.intersects is already tested
+        return toRange(l).intersects(r);
+    }
+
+    private static Range<Token> toRange(AbstractBounds<Token> bounds)
+    {
+        if (bounds instanceof Range)
+            return (Range<Token>) bounds;
+
+        Token l = bounds.left;
+        if (bounds.inclusiveLeft())
+            l = new Murmur3Partitioner.LongToken(l.getLongValue() - 1);
+        Token r = bounds.right;
+        if (!bounds.inclusiveRight())
+            r = new Murmur3Partitioner.LongToken(r.getLongValue() - 1);
+
+        return new Range<>(l, r);
+    }
+
+    private static boolean isPoint(AbstractBounds<Token> l)
+    {
+        return l.inclusiveLeft() && l.inclusiveRight() && l.left.equals(l.right);
+    }
+
+    private Gen<AbstractBounds<Token>> bounds()
+    {
+        return longs().between(0, MAX_TOKEN)
+                      .zip(integers().between(0, 6), this::createBoundary) // 14% chance min
+               .zip(longs().between(0, MAX_TOKEN)
+                           .zip(integers().between(-1, 6), this::createBoundary), // 12.5% chance point, 12.5% chance min
+                    this::createAbstractBounds);
+    }
+
+    private AbstractBounds.Boundary<Token> createBoundary(long pos, int minOrInclusive)
+    {
+        if (minOrInclusive < 0)
+            return null;    // point bounds
+        Token t;
+        if (minOrInclusive == 0)
+            t = Murmur3Partitioner.instance.getMinimumToken();
+        else
+            t = new Murmur3Partitioner.LongToken(pos);
+
+        return new AbstractBounds.Boundary(t, minOrInclusive % 2 == 0);
+    }
+
+    private AbstractBounds<Token> createAbstractBounds(AbstractBounds.Boundary<Token> left, AbstractBounds.Boundary<Token> right)
+    {
+        if (right == null)
+        {
+            return AbstractBounds.bounds(left.boundary, true, left.boundary, true);
+        }
+
+        if (!left.inclusive && right.inclusive)
+            return new Range<>(left.boundary, right.boundary); // ranges can be wraparound
+
+        // For other cases, create a normal bounds
+        int cmp = left.boundary.compareTo(right.boundary);
+        if (cmp < 0)
+            return AbstractBounds.bounds(left, right);
+        else if (cmp > 0)
+            return AbstractBounds.bounds(right, left);
+        else
+            return AbstractBounds.bounds(left.boundary, true, left.boundary, true);
+    }
+}