diff --git a/CHANGES.txt b/CHANGES.txt index ac28d786ca..4d88aa96af 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 6d7bdbb200..41633150a9 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -226,7 +226,7 @@ public class SelectStatement implements CQLStatement return getRangeCommand(options, limitForQuery, now); List 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 diff --git a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java b/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java index 6ed635f1df..35d0971c10 100644 --- a/src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java +++ b/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 commands, ConsistencyLevel consistencyLevel, ClientState cState, boolean localQuery, PagingState state) + MultiPartitionPager(List 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 result = new ArrayList(); - int remainingThisQuery = pageSize; + int remainingThisQuery = Math.min(remaining, pageSize); while (remainingThisQuery > 0 && !isExhausted()) { // isExhausted has set us on the first non-exhausted pager diff --git a/src/java/org/apache/cassandra/service/pager/Pageable.java b/src/java/org/apache/cassandra/service/pager/Pageable.java index 3a69bf4677..d4986f7acc 100644 --- a/src/java/org/apache/cassandra/service/pager/Pageable.java +++ b/src/java/org/apache/cassandra/service/pager/Pageable.java @@ -30,9 +30,12 @@ public interface Pageable { public final List commands; - public ReadCommands(List commands) + public final int limitForQuery; + + public ReadCommands(List commands, int limitForQuery) { this.commands = commands; + this.limitForQuery = limitForQuery; } } } diff --git a/src/java/org/apache/cassandra/service/pager/QueryPagers.java b/src/java/org/apache/cassandra/service/pager/QueryPagers.java index c03e8ecda7..f933ccb026 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPagers.java +++ b/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) { diff --git a/test/unit/org/apache/cassandra/service/QueryPagerTest.java b/test/unit/org/apache/cassandra/service/QueryPagerTest.java index e71e97ac85..c78412f00d 100644 --- a/test/unit/org/apache/cassandra/service/QueryPagerTest.java +++ b/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() {{ add(sliceQuery("k1", "c2", "c6", 10)); add(sliceQuery("k4", "c3", "c5", 10)); - }})); + }}, 10)); List page;