fix duplicate rows being read during mapreduce. patch by jbellis; reviewed by Jeremy Hanna for CASSANDRA-1042

git-svn-id: https://svn.apache.org/repos/asf/cassandra/branches/cassandra-0.6@965604 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -12,6 +12,7 @@
  * log errors in gossip instead of re-throwing (CASSANDRA-1289)
  * avoid aborting commitlog replay prematurely if a flushed-but-
    not-removed commitlog segment is encountered (CASSANDRA-1297)
+ * fix duplicate rows being read during mapreduce (CASSANDRA-1142)
 
 
 0.6.3

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

@@ -895,7 +895,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
        range_slice.  still opens one randomaccessfile per key, which sucks.  something like compactioniterator
        would be better.
      */
-    private boolean getKeyRange(List<String> keys, final AbstractBounds range, int maxResults)
+    private void getKeyRange(List<String> keys, final AbstractBounds range, int maxResults)
     throws IOException, ExecutionException, InterruptedException
     {
         final DecoratedKey startWith = new DecoratedKey(range.left, null);
@@ -974,21 +974,22 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             {
                 if (!stopAt.isEmpty() && stopAt.compareTo(current) < 0)
                 {
-                    return true;
+                    return;
                 }
 
                 if (range instanceof Bounds || !first || !current.equals(startWith))
                 {
+                    if (logger_.isDebugEnabled())
+                        logger_.debug("scanned " + current.key + " with token of " + StorageService.getPartitioner().getToken(current.key));
                     keys.add(current.key);
                 }
                 first = false;
 
                 if (keys.size() >= maxResults)
                 {
-                    return true;
+                    return;
                 }
             }
-            return false;
         }
         finally
         {
@@ -1017,23 +1018,10 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     throws IOException, ExecutionException, InterruptedException
     {
         List<String> keys = new ArrayList<String>();
-        boolean completed;
-        if ((range instanceof Bounds || !((Range)range).isWrapAround()))
-        {
-            completed = getKeyRange(keys, range, keyMax);
-        }
-        else
-        {
-            // wrapped range
-            Token min = StorageService.getPartitioner().getMinimumToken();
-            Range first = new Range(range.left, min);
-            completed = getKeyRange(keys, first, keyMax);
-            if (!completed && min.compareTo(range.right) < 0)
-            {
-                Range second = new Range(min, range.right);
-                getKeyRange(keys, second, keyMax);
-            }
-        }
+        assert range instanceof Bounds
+               || (!((Range)range).isWrapAround() || range.right.equals(StorageService.getPartitioner().getMinimumToken()))
+               : range;
+        getKeyRange(keys, range, keyMax);
         List<Row> rows = new ArrayList<Row>(keys.size());
         final QueryPath queryPath =  new QueryPath(columnFamily_, super_column, null);
         final SortedSet<byte[]> columnNameSet = new TreeSet<byte[]>(getComparator());

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

@@ -131,10 +131,10 @@ public class Range extends AbstractBounds implements Comparable<Range>, Serializ
      */
     public Set<Range> intersectionWith(Range that)
     {
-        if (this.contains(that))
-            return rangeSet(that);
         if (that.contains(this))
             return rangeSet(this);
+        if (this.contains(that))
+            return rangeSet(that);
 
         boolean thiswraps = isWrapAround(left, right);
         boolean thatwraps = isWrapAround(that.left, that.right);

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -659,10 +659,13 @@ public class StorageProxy implements StorageProxyMBean
      *     D, but we don't want any other results from it until after the (D, T] range.  Unwrapping so that
      *     the ranges we consider are (D, T], (T, MIN], (MIN, D] fixes this.
      */
-    private static List<AbstractBounds> getRestrictedRanges(AbstractBounds queryRange)
+    private static List<AbstractBounds> getRestrictedRanges(final AbstractBounds queryRange)
     {
         TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata();
 
+        if (logger.isDebugEnabled())
+            logger.debug("computing restricted ranges for query " + queryRange);
+
         List<AbstractBounds> ranges = new ArrayList<AbstractBounds>();
         // for each node, compute its intersection with the query range, and add its unwrapped components to our list
         for (Token nodeToken : tokenMetadata.sortedTokens())
@@ -682,14 +685,23 @@ public class StorageProxy implements StorageProxyMBean
         // re-sort ranges in ring order, post-unwrapping
         Comparator<AbstractBounds> comparator = new Comparator<AbstractBounds>()
         {
+            // no restricted ranges will overlap so we don't need to worry about inclusive vs exclusive left,
+            // just sort by raw token position.
             public int compare(AbstractBounds o1, AbstractBounds o2)
             {
-                // no restricted ranges will overlap so we don't need to worry about inclusive vs exclusive left,
-                // just sort by raw token position.
-                return o1.left.compareTo(o2.left);
+                // sort in order that the original query range would see them.
+                int queryOrder1 = queryRange.left.compareTo(o1.left);
+                int queryOrder2 = queryRange.left.compareTo(o2.left);
+                if (queryOrder1 < queryOrder2)
+                    return -1; // o1 comes after query start, o2 wraps to after
+                if (queryOrder1 > queryOrder2)
+                    return 1; // o2 comes after query start, o1 wraps to after
+                return o1.left.compareTo(o2.left); // o1 and o2 are on the same side of query start
             }
         };
         Collections.sort(ranges, comparator);
+        if (logger.isDebugEnabled())
+            logger.debug("Sorted ranges are [" + StringUtils.join(ranges, ", ") + "]");
 
         return ranges;
     }

test/system/test_server.py

@@ -906,6 +906,26 @@ class TestMutations(CassandraTester):
         assert result[1].columns[0].column.name == 'col1'
         
     
+    def test_wrapped_range_slices(self):
+        def copp_token(key):
+            # I cheated and generated this from Java
+            return {'a': '00530000000100000001', 
+                    'b': '00540000000100000001', 
+                    'c': '00550000000100000001',
+                    'd': '00560000000100000001', 
+                    'e': '00580000000100000001'}[key]
+        for key in ['a', 'b', 'c', 'd', 'e']:
+            for cname in ['col1', 'col2', 'col3', 'col4', 'col5']:
+                client.insert('Keyspace1', key, ColumnPath('Standard1', column=cname), 'v-' + cname, 0, ConsistencyLevel.ONE)
+        cp = ColumnParent('Standard1')
+
+        result = client.get_range_slices("Keyspace1", cp, SlicePredicate(column_names=['col1', 'col3']), KeyRange(start_token=copp_token('e'), end_token=copp_token('e')), ConsistencyLevel.ONE)
+        assert [row.key for row in result] == ['a', 'b', 'c', 'd', 'e',], [row.key for row in result]
+
+        result = client.get_range_slices("Keyspace1", cp, SlicePredicate(column_names=['col1', 'col3']), KeyRange(start_token=copp_token('c'), end_token=copp_token('c')), ConsistencyLevel.ONE)
+        assert [row.key for row in result] == ['d', 'e', 'a', 'b', 'c',], [row.key for row in result]
+        
+
     def test_get_slice_by_names(self):
         _insert_range()
         p = SlicePredicate(column_names=['c1', 'c2'])