From 4eef4499ba24dd5168fdd759f4e2587892e4394b Mon Sep 17 00:00:00 2001
From: Benedict Elliott Smith <benedict@apache.org>
Date: Tue, 24 Mar 2015 13:35:01 +0000
Subject: [PATCH] expand CASSANDRA-8946 behaviour to all SSTableScanner
 constructors

---
 .../cassandra/io/sstable/SSTableScanner.java  | 98 +++++++++----------
 1 file changed, 44 insertions(+), 54 deletions(-)

diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
index 46ddc24542..51631079c6 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
@@ -87,42 +87,7 @@ public class SSTableScanner implements ISSTableScanner
         this.dataRange = dataRange;
 
         List<AbstractBounds<RowPosition>> boundsList = new ArrayList<>(2);
-        if (dataRange.isWrapAround())
-        {
-            if (dataRange.stopKey().compareTo(sstable.first) >= 0)
-            {
-                // since we wrap, we must contain the whole sstable prior to stopKey()
-                Boundary<RowPosition> left = new Boundary<RowPosition>(sstable.first, true);
-                Boundary<RowPosition> right;
-                right = dataRange.keyRange().rightBoundary();
-                right = minRight(right, sstable.last, true);
-                if (!isEmpty(left, right))
-                    boundsList.add(AbstractBounds.bounds(left, right));
-            }
-            if (dataRange.startKey().compareTo(sstable.last) <= 0)
-            {
-                // since we wrap, we must contain the whole sstable after dataRange.startKey()
-                Boundary<RowPosition> right = new Boundary<RowPosition>(sstable.last, true);
-                Boundary<RowPosition> left;
-                left = dataRange.keyRange().leftBoundary();
-                left = maxLeft(left, sstable.first, true);
-                if (!isEmpty(left, right))
-                    boundsList.add(AbstractBounds.bounds(left, right));
-            }
-        }
-        else
-        {
-            assert dataRange.startKey().compareTo(dataRange.stopKey()) <= 0 || dataRange.stopKey().isMinimum();
-            Boundary<RowPosition> left, right;
-            left = dataRange.keyRange().leftBoundary();
-            right = dataRange.keyRange().rightBoundary();
-            left = maxLeft(left, sstable.first, true);
-            // apparently isWrapAround() doesn't count Bounds that extend to the limit (min) as wrapping
-            right = dataRange.stopKey().isMinimum() ? new Boundary<RowPosition>(sstable.last, true)
-                                                    : minRight(right, sstable.last, true);
-            if (!isEmpty(left, right))
-                boundsList.add(AbstractBounds.bounds(left, right));
-        }
+        addRange(dataRange.keyRange(), boundsList);
         this.rangeIterator = boundsList.iterator();
     }
 
@@ -140,28 +105,53 @@ public class SSTableScanner implements ISSTableScanner
         this.sstable = sstable;
         this.dataRange = null;
 
-        List<Range<Token>> normalized = Range.normalize(tokenRanges);
-        List<AbstractBounds<RowPosition>> boundsList = new ArrayList<>(normalized.size());
-        for (Range<Token> range : normalized)
-        {
-            // cap our ranges by the start/end of the sstable
-            RowPosition right = range.right.maxKeyBound(sstable.partitioner);
-            if (right.compareTo(sstable.last) > 0)
-                right = sstable.last;
-
-            RowPosition left = range.left.maxKeyBound(sstable.partitioner);
-            if (left.compareTo(sstable.first) < 0)
-            {
-                if (sstable.first.compareTo(right) <= 0)
-                    boundsList.add(new Bounds<>(sstable.first, right, sstable.partitioner));
-            }
-            else if (left.compareTo(right) < 0)
-                boundsList.add(new Range<>(left, right, sstable.partitioner));
-        }
+        List<AbstractBounds<RowPosition>> boundsList = new ArrayList<>(tokenRanges.size());
+        for (Range<Token> range : Range.normalize(tokenRanges))
+            addRange(range.toRowBounds(), boundsList);
 
         this.rangeIterator = boundsList.iterator();
     }
 
+    private void addRange(AbstractBounds<RowPosition> requested, List<AbstractBounds<RowPosition>> boundsList)
+    {
+        if (requested instanceof Range && ((Range)requested).isWrapAround())
+        {
+            if (requested.right.compareTo(sstable.first) >= 0)
+            {
+                // since we wrap, we must contain the whole sstable prior to stopKey()
+                Boundary<RowPosition> left = new Boundary<RowPosition>(sstable.first, true);
+                Boundary<RowPosition> right;
+                right = requested.rightBoundary();
+                right = minRight(right, sstable.last, true);
+                if (!isEmpty(left, right))
+                    boundsList.add(AbstractBounds.bounds(left, right));
+            }
+            if (requested.left.compareTo(sstable.last) <= 0)
+            {
+                // since we wrap, we must contain the whole sstable after dataRange.startKey()
+                Boundary<RowPosition> right = new Boundary<RowPosition>(sstable.last, true);
+                Boundary<RowPosition> left;
+                left = requested.leftBoundary();
+                left = maxLeft(left, sstable.first, true);
+                if (!isEmpty(left, right))
+                    boundsList.add(AbstractBounds.bounds(left, right));
+            }
+        }
+        else
+        {
+            assert requested.left.compareTo(requested.right) <= 0 || requested.right.isMinimum();
+            Boundary<RowPosition> left, right;
+            left = requested.leftBoundary();
+            right = requested.rightBoundary();
+            left = maxLeft(left, sstable.first, true);
+            // apparently isWrapAround() doesn't count Bounds that extend to the limit (min) as wrapping
+            right = requested.right.isMinimum() ? new Boundary<RowPosition>(sstable.last, true)
+                                                    : minRight(right, sstable.last, true);
+            if (!isEmpty(left, right))
+                boundsList.add(AbstractBounds.bounds(left, right));
+        }
+    }
+
     private void seekToCurrentRangeStart()
     {
         long indexPosition = sstable.getIndexScanPosition(currentRange.left);