diff --git a/CHANGES.txt b/CHANGES.txt index c524258496..4434d242ec 100644 --- a/CHANGES.txt +++ b/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 diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 5b0b8856cf..ce07215e11 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/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 keys, final AbstractBounds range, int maxResults) + private void getKeyRange(List 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 keys = new ArrayList(); - 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 rows = new ArrayList(keys.size()); final QueryPath queryPath = new QueryPath(columnFamily_, super_column, null); final SortedSet columnNameSet = new TreeSet(getComparator()); diff --git a/src/java/org/apache/cassandra/dht/Range.java b/src/java/org/apache/cassandra/dht/Range.java index c6428a73d8..f5d832630a 100644 --- a/src/java/org/apache/cassandra/dht/Range.java +++ b/src/java/org/apache/cassandra/dht/Range.java @@ -131,10 +131,10 @@ public class Range extends AbstractBounds implements Comparable, Serializ */ public Set 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); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 16c4c3531f..e2d5c93f1a 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/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 getRestrictedRanges(AbstractBounds queryRange) + private static List getRestrictedRanges(final AbstractBounds queryRange) { TokenMetadata tokenMetadata = StorageService.instance.getTokenMetadata(); + if (logger.isDebugEnabled()) + logger.debug("computing restricted ranges for query " + queryRange); + List ranges = new ArrayList(); // 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 comparator = new Comparator() { + // 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; } diff --git a/test/system/test_server.py b/test/system/test_server.py index 8e9ac16913..a8e82b8b42 100644 --- a/test/system/test_server.py +++ b/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'])