Merge branch 'cassandra-2.2' into cassandra-3.0

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

@@ -1179,7 +1179,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         Set<SSTableReader> results = null;
         for (SSTableReader sstable : sstables)
         {
-            Set<SSTableReader> overlaps = ImmutableSet.copyOf(view.sstablesInBounds(sstableSet, AbstractBounds.bounds(sstable.first, true, sstable.last, true)));
+            Set<SSTableReader> overlaps = ImmutableSet.copyOf(view.sstablesInBounds(sstableSet, sstable.first, sstable.last));
             results = results == null ? overlaps : Sets.union(results, overlaps).immutableCopy();
         }
         results = Sets.difference(results, ImmutableSet.copyOf(sstables));

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

@@ -87,9 +87,10 @@ public class SizeEstimatesRecorder extends MigrationListener implements Runnable
     @SuppressWarnings("resource")
     private void recordSizeEstimates(ColumnFamilyStore table, Collection<Range<Token>> localRanges)
     {
+        List<Range<Token>> unwrappedRanges = Range.normalize(localRanges);
         // for each local primary range, estimate (crudely) mean partition size and partitions count.
         Map<Range<Token>, Pair<Long, Long>> estimates = new HashMap<>(localRanges.size());
-        for (Range<Token> range : localRanges)
+        for (Range<Token> range : unwrappedRanges)
         {
             // filter sstables that have partitions in this range.
             Refs<SSTableReader> refs = null;

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

@@ -176,12 +176,19 @@ public class View
         return String.format("View(pending_count=%d, sstables=%s, compacting=%s)", liveMemtables.size() + flushingMemtables.size() - 1, sstables, compacting);
     }
 
-    public Iterable<SSTableReader> sstablesInBounds(SSTableSet sstableSet, AbstractBounds<PartitionPosition> rowBounds)
+    /**
+     * Returns the sstables that have any partition between {@code left} and {@code right}, when both bounds are taken inclusively.
+     * The interval formed by {@code left} and {@code right} shouldn't wrap.
+     */
+    public Iterable<SSTableReader> sstablesInBounds(SSTableSet sstableSet, PartitionPosition left, PartitionPosition right)
     {
+        assert !AbstractBounds.strictlyWrapsAround(left, right);
+
         if (intervalTree.isEmpty())
             return Collections.emptyList();
-        PartitionPosition stopInTree = rowBounds.right.isMinimum() ? intervalTree.max() : rowBounds.right;
-        return select(sstableSet, intervalTree.search(Interval.create(rowBounds.left, stopInTree)));
+
+        PartitionPosition stopInTree = right.isMinimum() ? intervalTree.max() : right;
+        return select(sstableSet, intervalTree.search(Interval.create(left, stopInTree)));
     }
 
     public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet)
@@ -210,7 +217,12 @@ public class View
      */
     public static Function<View, Iterable<SSTableReader>> select(SSTableSet sstableSet, AbstractBounds<PartitionPosition> rowBounds)
     {
-        return (view) -> view.sstablesInBounds(sstableSet, rowBounds);
+        // Note that View.sstablesInBounds always includes it's bound while rowBounds may not. This is ok however
+        // because the fact we restrict the sstables returned by this function is an optimization in the first
+        // place and the returned sstables will (almost) never cover *exactly* rowBounds anyway. It's also
+        // *very* unlikely that a sstable is included *just* because we consider one of the bound inclusively
+        // instead of exclusively, so the performance impact is negligible in practice.
+        return (view) -> view.sstablesInBounds(sstableSet, rowBounds.left, rowBounds.right);
     }
 
     // METHODS TO CONSTRUCT FUNCTIONS FOR MODIFYING A VIEW:

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

@@ -20,6 +20,7 @@ package org.apache.cassandra.dht;
 import java.io.DataInput;
 import java.io.IOException;
 import java.io.Serializable;
+import java.util.Collection;
 import java.util.List;
 
 import org.apache.cassandra.db.DecoratedKey;
@@ -72,6 +73,30 @@ public abstract class AbstractBounds<T extends RingPosition<T>> implements Seria
     public abstract boolean inclusiveLeft();
     public abstract boolean inclusiveRight();
 
+    /**
+     * Whether {@code left} and {@code right} forms a wrapping interval, that is if unwrapping wouldn't be a no-op.
+     * <p>
+     * Note that the semantic is slightly different from {@link Range#isWrapAround()} in the sense that if both
+     * {@code right} are minimal (for the partitioner), this methods return false (doesn't wrap) while
+     * {@link Range#isWrapAround()} returns true (does wrap). This is confusing and we should fix it by
+     * refactoring/rewriting the whole AbstractBounds hierarchy with cleaner semantics, but we don't want to risk
+     * breaking something by changing {@link Range#isWrapAround()} in the meantime.
+     */
+    public static <T extends RingPosition<T>> boolean strictlyWrapsAround(T left, T right)
+    {
+        return !(left.compareTo(right) <= 0 || right.isMinimum());
+    }
+
+    public static <T extends RingPosition<T>> boolean noneStrictlyWrapsAround(Collection<AbstractBounds<T>> bounds)
+    {
+        for (AbstractBounds<T> b : bounds)
+        {
+            if (strictlyWrapsAround(b.left, b.right))
+                return false;
+        }
+        return true;
+    }
+
     @Override
     public int hashCode()
     {

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

@@ -32,7 +32,7 @@ public class Bounds<T extends RingPosition<T>> extends AbstractBounds<T>
     {
         super(left, right);
         // unlike a Range, a Bounds may not wrap
-        assert left.compareTo(right) <= 0 || right.isMinimum() : "[" + left + "," + right + "]";
+        assert !strictlyWrapsAround(left, right) : "[" + left + "," + right + "]";
     }
 
     public boolean contains(T position)

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

@@ -31,7 +31,7 @@ public class ExcludingBounds<T extends RingPosition<T>> extends AbstractBounds<T
     {
         super(left, right);
         // unlike a Range, an ExcludingBounds may not wrap, nor be empty
-        assert left.compareTo(right) < 0 || right.isMinimum() : "(" + left + "," + right + ")";
+        assert !strictlyWrapsAround(left, right) && (right.isMinimum() || left.compareTo(right) != 0) : "(" + left + "," + right + ")";
     }
 
     public boolean contains(T position)

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

@@ -32,7 +32,7 @@ public class IncludingExcludingBounds<T extends RingPosition<T>> extends Abstrac
         super(left, right);
         // unlike a Range, an IncludingExcludingBounds may not wrap, nor have
         // right == left unless the right is the min token
-        assert left.compareTo(right) < 0 || right.isMinimum() : "[" + left + "," + right + ")";
+        assert !strictlyWrapsAround(left, right) && (right.isMinimum() || left.compareTo(right) != 0) : "(" + left + "," + right + ")";
     }
 
     public boolean contains(T position)

src/java/org/apache/cassandra/streaming/StreamSession.java

@@ -36,7 +36,6 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.PartitionPosition;
-import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.gms.*;
@@ -320,15 +319,22 @@ public class StreamSession implements IEndpointStateChangeSubscriber
         {
             for (ColumnFamilyStore cfStore : stores)
             {
-                final List<AbstractBounds<PartitionPosition>> rowBoundsList = new ArrayList<>(ranges.size());
+                final List<Range<PartitionPosition>> keyRanges = new ArrayList<>(ranges.size());
                 for (Range<Token> range : ranges)
-                    rowBoundsList.add(Range.makeRowRange(range));
+                    keyRanges.add(Range.makeRowRange(range));
                 refs.addAll(cfStore.selectAndReference(view -> {
                     Set<SSTableReader> sstables = Sets.newHashSet();
-                    for (AbstractBounds<PartitionPosition> rowBounds : rowBoundsList)
+                    for (Range<PartitionPosition> keyRange : keyRanges)
                     {
-                        for (SSTableReader sstable : view.sstablesInBounds(SSTableSet.CANONICAL, rowBounds))
+                        // keyRange excludes its start, while sstableInBounds is inclusive (of both start and end).
+                        // This is fine however, because keyRange has been created from a token range through Range.makeRowRange (see above).
+                        // And that later method uses the Token.maxKeyBound() method to creates the range, which return a "fake" key that
+                        // sort after all keys having the token. That "fake" key cannot however be equal to any real key, so that even
+                        // including keyRange.left will still exclude any key having the token of the original token range, and so we're
+                        // still actually selecting what we wanted.
+                        for (SSTableReader sstable : view.sstablesInBounds(SSTableSet.CANONICAL, keyRange.left, keyRange.right))
                         {
+                            // sstableInBounds may contain early opened sstables
                             if (!isIncremental || !sstable.isRepaired())
                                 sstables.add(sstable);
                         }

test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java

@@ -63,13 +63,15 @@ public class ViewTest
             {
                 PartitionPosition min = MockSchema.readerBounds(i);
                 PartitionPosition max = MockSchema.readerBounds(j);
-                for (boolean minInc : new boolean[] { true, false} )
+                for (boolean minInc : new boolean[] { true })//, false} )
                 {
-                    for (boolean maxInc : new boolean[] { true, false} )
+                    for (boolean maxInc : new boolean[] { true })//, false} )
                     {
                         if (i == j && !(minInc && maxInc))
                             continue;
-                        List<SSTableReader> r = ImmutableList.copyOf(initialView.sstablesInBounds(SSTableSet.LIVE, AbstractBounds.bounds(min, minInc, max, maxInc)));
+
+                        AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(min, minInc, max, maxInc);
+                        List<SSTableReader> r = ImmutableList.copyOf(initialView.sstablesInBounds(SSTableSet.LIVE,bounds.left, bounds.right));
                         Assert.assertEquals(String.format("%d(%s) %d(%s)", i, minInc, j, maxInc), j - i + (minInc ? 0 : -1) + (maxInc ? 1 : 0), r.size());
                     }
                 }