diff --git a/CHANGES.txt b/CHANGES.txt index 01c4c288d5..40e5b703f3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -61,6 +61,7 @@ Merged from 2.0: * Extend triggers to support CAS updates (CASSANDRA-6882) * Static columns with IF NOT EXISTS don't always work as expected (CASSANDRA-6873) * Add CqlRecordReader to take advantage of native CQL pagination (CASSANDRA-6311) + * Fix paging with SELECT DISTINCT (CASSANDRA-6857) Merged from 1.2: * Add UNLOGGED, COUNTER options to BATCH documentation (CASSANDRA-6816) * add extra SSL cipher suites (CASSANDRA-6613) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 845352d51d..10418123d5 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1965,10 +1965,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean Composite columnStop, List rowFilter, int maxResults, + boolean countCQL3Rows, long now) { DataRange dataRange = new DataRange.Paging(keyRange, columnRange, columnStart, columnStop, metadata.comparator); - return ExtendedFilter.create(this, dataRange, rowFilter, maxResults, true, now); + return ExtendedFilter.create(this, dataRange, rowFilter, maxResults, countCQL3Rows, now); } public List getRangeSlice(AbstractBounds range, diff --git a/src/java/org/apache/cassandra/db/PagedRangeCommand.java b/src/java/org/apache/cassandra/db/PagedRangeCommand.java index 5c8f3bae7e..f2d81b9626 100644 --- a/src/java/org/apache/cassandra/db/PagedRangeCommand.java +++ b/src/java/org/apache/cassandra/db/PagedRangeCommand.java @@ -40,6 +40,7 @@ public class PagedRangeCommand extends AbstractRangeCommand public final Composite start; public final Composite stop; public final int limit; + private final boolean countCQL3Rows; public PagedRangeCommand(String keyspace, String columnFamily, @@ -49,12 +50,14 @@ public class PagedRangeCommand extends AbstractRangeCommand Composite start, Composite stop, List rowFilter, - int limit) + int limit, + boolean countCQL3Rows) { super(keyspace, columnFamily, timestamp, keyRange, predicate, rowFilter); this.start = start; this.stop = stop; this.limit = limit; + this.countCQL3Rows = countCQL3Rows; } public MessageOut createMessage() @@ -74,7 +77,8 @@ public class PagedRangeCommand extends AbstractRangeCommand newStart, newStop, rowFilter, - limit); + limit, + countCQL3Rows); } public AbstractRangeCommand withUpdatedLimit(int newLimit) @@ -87,7 +91,8 @@ public class PagedRangeCommand extends AbstractRangeCommand start, stop, rowFilter, - newLimit); + newLimit, + countCQL3Rows); } public int limit() @@ -97,14 +102,14 @@ public class PagedRangeCommand extends AbstractRangeCommand public boolean countCQL3Rows() { - return true; + return countCQL3Rows; } public List executeLocally() { ColumnFamilyStore cfs = Keyspace.open(keyspace).getColumnFamilyStore(columnFamily); - ExtendedFilter exFilter = cfs.makeExtendedFilter(keyRange, (SliceQueryFilter)predicate, start, stop, rowFilter, limit, timestamp); + ExtendedFilter exFilter = cfs.makeExtendedFilter(keyRange, (SliceQueryFilter)predicate, start, stop, rowFilter, limit, countCQL3Rows(), timestamp); if (cfs.indexManager.hasIndexFor(rowFilter)) return cfs.search(exFilter); else @@ -146,6 +151,8 @@ public class PagedRangeCommand extends AbstractRangeCommand } out.writeInt(cmd.limit); + if (version >= MessagingService.VERSION_21) + out.writeBoolean(cmd.countCQL3Rows); } public PagedRangeCommand deserialize(DataInput in, int version) throws IOException @@ -174,7 +181,10 @@ public class PagedRangeCommand extends AbstractRangeCommand } int limit = in.readInt(); - return new PagedRangeCommand(keyspace, columnFamily, timestamp, keyRange, predicate, start, stop, rowFilter, limit); + boolean countCQL3Rows = version >= MessagingService.VERSION_21 + ? in.readBoolean() + : predicate.compositesToGroup >= 0 || predicate.count != 1; // See #6857 + return new PagedRangeCommand(keyspace, columnFamily, timestamp, keyRange, predicate, start, stop, rowFilter, limit, countCQL3Rows); } public long serializedSize(PagedRangeCommand cmd, int version) @@ -203,6 +213,8 @@ public class PagedRangeCommand extends AbstractRangeCommand } size += TypeSizes.NATIVE.sizeof(cmd.limit); + if (version >= MessagingService.VERSION_21) + size += TypeSizes.NATIVE.sizeof(cmd.countCQL3Rows); return size; } } diff --git a/src/java/org/apache/cassandra/service/pager/QueryPagers.java b/src/java/org/apache/cassandra/service/pager/QueryPagers.java index f186e693d1..04702d0875 100644 --- a/src/java/org/apache/cassandra/service/pager/QueryPagers.java +++ b/src/java/org/apache/cassandra/service/pager/QueryPagers.java @@ -71,8 +71,14 @@ public class QueryPagers else { assert command instanceof RangeSliceCommand; - // We can never be sure a range slice won't need paging - return true; + RangeSliceCommand rsc = (RangeSliceCommand)command; + // We don't support paging for thrift in general because the way thrift RangeSliceCommand count rows + // independently of cells makes things harder (see RangeSliceQueryPager). The one case where we do + // get a RangeSliceCommand from CQL3 without the countCQL3Rows flag set is for DISTINCT. In that case + // however, the underlying sliceQueryFilter count is 1, so that the RSC limit is still a limit on the + // number of CQL3 rows returned. + assert rsc.countCQL3Rows || (rsc.predicate instanceof SliceQueryFilter && ((SliceQueryFilter)rsc.predicate).count == 1); + return rsc.maxResults > pageSize; } } diff --git a/src/java/org/apache/cassandra/service/pager/RangeSliceQueryPager.java b/src/java/org/apache/cassandra/service/pager/RangeSliceQueryPager.java index 7e79ffea8d..b03cc69f50 100644 --- a/src/java/org/apache/cassandra/service/pager/RangeSliceQueryPager.java +++ b/src/java/org/apache/cassandra/service/pager/RangeSliceQueryPager.java @@ -81,7 +81,8 @@ public class RangeSliceQueryPager extends AbstractQueryPager start, sf.finish(), command.rowFilter, - pageSize); + pageSize, + command.countCQL3Rows); return localQuery ? pageCmd.executeLocally() diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java index c705f01a2d..70fc12a8fd 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java @@ -1188,7 +1188,7 @@ public class ColumnFamilyStoreTest extends SchemaLoader cellname("c2"), false, 0); - rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds(ka, kc), sf, cellname("c2"), cellname("c1"), null, 2, System.currentTimeMillis())); + rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds(ka, kc), sf, cellname("c2"), cellname("c1"), null, 2, true, System.currentTimeMillis())); assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows); iter = rows.iterator(); row1 = iter.next(); @@ -1196,7 +1196,7 @@ public class ColumnFamilyStoreTest extends SchemaLoader assertColumnNames(row1, "c2"); assertColumnNames(row2, "c1"); - rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds(kb, kc), sf, cellname("c1"), cellname("c1"), null, 10, System.currentTimeMillis())); + rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds(kb, kc), sf, cellname("c1"), cellname("c1"), null, 10, true, System.currentTimeMillis())); assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows); iter = rows.iterator(); row1 = iter.next();