Merge branch 'cassandra-2.0' into cassandra-2.1

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
	src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java
	src/java/org/apache/cassandra/service/pager/QueryPagers.java

CHANGES.txt

@@ -41,6 +41,7 @@
  * Log failed host when preparing incremental repair (CASSANDRA-8228)
  * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
 Merged from 2.0:
+ * Fix paging for multi-partition IN queries (CASSANDRA-8408)
  * Fix MOVED_NODE topology event never being emitted when a node
    moves its token (CASSANDRA-8373)
  * Fix validation of indexes in COMPACT tables (CASSANDRA-8156)

src/java/org/apache/cassandra/cql3/statements/SelectStatement.java

@@ -226,7 +226,7 @@ public class SelectStatement implements CQLStatement
             return getRangeCommand(options, limitForQuery, now);
 
         List<ReadCommand> commands = getSliceCommands(options, limitForQuery, now);
-        return commands == null ? null : new Pageable.ReadCommands(commands);
+        return commands == null ? null : new Pageable.ReadCommands(commands, limitForQuery);
     }
 
     public Pageable getPageableCommand(QueryOptions options) throws RequestValidationException

src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java

@@ -47,7 +47,7 @@ class MultiPartitionPager implements QueryPager
     private int remaining;
     private int current;
 
-    MultiPartitionPager(List<ReadCommand> commands, ConsistencyLevel consistencyLevel, ClientState cState, boolean localQuery, PagingState state)
+    MultiPartitionPager(List<ReadCommand> commands, ConsistencyLevel consistencyLevel, ClientState cState, boolean localQuery, PagingState state, int limitForQuery)
     {
         int i = 0;
         // If it's not the beginning (state != null), we need to find where we were and skip previous commands
@@ -77,7 +77,8 @@ class MultiPartitionPager implements QueryPager
                 throw new IllegalArgumentException("All commands must have the same timestamp or weird results may happen.");
             pagers[j - i] = makePager(command, consistencyLevel, cState, localQuery, null);
         }
-        remaining = state == null ? computeRemaining(pagers) : state.remaining;
+
+        remaining = state == null ? limitForQuery : state.remaining;
     }
 
     private static SinglePartitionPager makePager(ReadCommand command, ConsistencyLevel consistencyLevel, ClientState cState, boolean localQuery, PagingState state)
@@ -87,14 +88,6 @@ class MultiPartitionPager implements QueryPager
              : new NamesQueryPager((SliceByNamesReadCommand)command, consistencyLevel, cState, localQuery);
     }
 
-    private static int computeRemaining(SinglePartitionPager[] pagers)
-    {
-        long remaining = 0;
-        for (SinglePartitionPager pager : pagers)
-            remaining += pager.maxRemaining();
-        return remaining > Integer.MAX_VALUE ? Integer.MAX_VALUE : (int)remaining;
-    }
-
     public PagingState state()
     {
         // Sets current to the first non-exhausted pager
@@ -124,7 +117,7 @@ class MultiPartitionPager implements QueryPager
     {
         List<Row> result = new ArrayList<Row>();
 
-        int remainingThisQuery = pageSize;
+        int remainingThisQuery = Math.min(remaining, pageSize);
         while (remainingThisQuery > 0 && !isExhausted())
         {
             // isExhausted has set us on the first non-exhausted pager

src/java/org/apache/cassandra/service/pager/Pageable.java

@@ -30,9 +30,12 @@ public interface Pageable
     {
         public final List<ReadCommand> commands;
 
-        public ReadCommands(List<ReadCommand> commands)
+        public final int limitForQuery;
+
+        public ReadCommands(List<ReadCommand> commands, int limitForQuery)
         {
             this.commands = commands;
+            this.limitForQuery = limitForQuery;
         }
     }
 }

src/java/org/apache/cassandra/service/pager/QueryPagers.java

@@ -99,7 +99,7 @@ public class QueryPagers
             if (commands.size() == 1)
                 return pager(commands.get(0), consistencyLevel, cState, local, state);
 
-            return new MultiPartitionPager(commands, consistencyLevel, cState, local, state);
+            return new MultiPartitionPager(commands, consistencyLevel, cState, local, state, ((Pageable.ReadCommands) command).limitForQuery);
         }
         else if (command instanceof ReadCommand)
         {

test/unit/org/apache/cassandra/service/QueryPagerTest.java

@@ -242,7 +242,7 @@ public class QueryPagerTest extends SchemaLoader
         QueryPager pager = QueryPagers.localPager(new Pageable.ReadCommands(new ArrayList<ReadCommand>() {{
             add(sliceQuery("k1", "c2", "c6", 10));
             add(sliceQuery("k4", "c3", "c5", 10));
-        }}));
+        }}, 10));
 
         List<Row> page;