Merge branch 'cassandra-2.1' into trunk

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)

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

@@ -1965,10 +1965,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
                                              Composite columnStop,
                                              List<IndexExpression> 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<Row> getRangeSlice(AbstractBounds<RowPosition> range,

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<IndexExpression> 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<PagedRangeCommand> 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<Row> 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;
         }
     }

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;
         }
     }
 

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()

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<RowPosition>(ka, kc), sf, cellname("c2"), cellname("c1"), null, 2, System.currentTimeMillis()));
+        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(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<RowPosition>(kb, kc), sf, cellname("c1"), cellname("c1"), null, 10, System.currentTimeMillis()));
+        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(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();