Merge branch 'cassandra-2.1' into trunk

CHANGES.txt

@@ -83,6 +83,8 @@
  * Log failed host when preparing incremental repair (CASSANDRA-8228)
  * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
 Merged from 2.0:
+ * Fix non-distinct results in DISTNCT queries on static columns when
+   paging is enabled (CASSANDRA-8087)
  * Move all hints related tasks to hints internal executor (CASSANDRA-8285)
  * Fix paging for multi-partition IN queries (CASSANDRA-8408)
  * Fix MOVED_NODE topology event never being emitted when a node

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

@@ -1993,12 +1993,12 @@ public List<Row> getRangeSlice(final AbstractBounds<RowPosition> range,
     /**
      * Allows generic range paging with the slice column filter.
      * Typically, suppose we have rows A, B, C ... Z having each some columns in [1, 100].
-     * And suppose we want to page throught the query that for all rows returns the columns
+     * And suppose we want to page through the query that for all rows returns the columns
      * within [25, 75]. For that, we need to be able to do a range slice starting at (row r, column c)
      * and ending at (row Z, column 75), *but* that only return columns in [25, 75].
      * That is what this method allows. The columnRange is the "window" of  columns we are interested
      * in each row, and columnStart (resp. columnEnd) is the start (resp. end) for the first
-     * (resp. end) requested row.
+     * (resp. last) requested row.
      */
     public ExtendedFilter makeExtendedFilter(AbstractBounds<RowPosition> keyRange,
                                              SliceQueryFilter columnRange,
@@ -2038,6 +2038,8 @@ public ExtendedFilter makeExtendedFilter(AbstractBounds<RowPosition> range,
             assert columnFilter instanceof SliceQueryFilter;
             SliceQueryFilter sfilter = (SliceQueryFilter)columnFilter;
             assert sfilter.slices.length == 1;
+            // create a new SliceQueryFilter that selects all cells, but pass the original slice start and finish
+            // through to DataRange.Paging to be used on the first and last partitions
             SliceQueryFilter newFilter = new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, sfilter.isReversed(), sfilter.count);
             dataRange = new DataRange.Paging(range, newFilter, sfilter.start(), sfilter.finish(), metadata.comparator);
         }

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

@@ -112,24 +112,37 @@ public boolean selectsFullRowFor(ByteBuffer rowKey)
         return selectFullRow;
     }
 
+    /**
+     * Returns a column filter that should be used for a particular row key.  Note that in the case of paging,
+     * slice starts and ends may change depending on the row key.
+     */
     public IDiskAtomFilter columnFilter(ByteBuffer rowKey)
     {
         return columnFilter;
     }
 
+    /**
+     * Sets a new limit on the number of (grouped) cells to fetch. This is currently only used when the query limit applies
+     * to CQL3 rows.
+     */
     public void updateColumnsLimit(int count)
     {
         columnFilter.updateColumnsLimit(count);
     }
 
     public static class Paging extends DataRange
     {
+        // The slice of columns that we want to fetch for each row, ignoring page start/end issues.
         private final SliceQueryFilter sliceFilter;
         private final Comparator<Composite> comparator;
-        private final Composite columnStart;
-        private final Composite columnFinish;
 
-        private Paging(AbstractBounds<RowPosition> range, SliceQueryFilter filter, Composite columnStart, Composite columnFinish, Comparator<Composite> comparator)
+        // used to restrict the start of the slice for the first partition in the range
+        private final Composite firstPartitionColumnStart;
+
+        // used to restrict the end of the slice for the last partition in the range
+        private final Composite lastPartitionColumnFinish;
+
+        private Paging(AbstractBounds<RowPosition> range, SliceQueryFilter filter, Composite firstPartitionColumnStart, Composite lastPartitionColumnFinish, Comparator<Composite> comparator)
         {
             super(range, filter);
 
@@ -139,8 +152,8 @@ private Paging(AbstractBounds<RowPosition> range, SliceQueryFilter filter, Compo
 
             this.sliceFilter = filter;
             this.comparator = comparator;
-            this.columnStart = columnStart;
-            this.columnFinish = columnFinish;
+            this.firstPartitionColumnStart = firstPartitionColumnStart;
+            this.lastPartitionColumnFinish = lastPartitionColumnFinish;
         }
 
         public Paging(AbstractBounds<RowPosition> range, SliceQueryFilter filter, Composite columnStart, Composite columnFinish, CellNameType comparator)
@@ -184,11 +197,11 @@ public IDiskAtomFilter columnFilter(ByteBuffer rowKey)
 
         private ColumnSlice[] slicesForKey(ByteBuffer key)
         {
-            // We don't call that until it's necessary, so assume we have to do some hard work
+            // Also note that firstPartitionColumnStart and lastPartitionColumnFinish, when used, only "restrict" the filter slices,
             // it doesn't expand on them. As such, we can ignore the case where they are empty and we do
             // as it screw up with the logic below (see #6592)
-            Composite newStart = equals(startKey(), key) && !columnStart.isEmpty() ? columnStart : null;
-            Composite newFinish = equals(stopKey(), key) && !columnFinish.isEmpty() ? columnFinish : null;
+            Composite newStart = equals(startKey(), key) && !firstPartitionColumnStart.isEmpty() ? firstPartitionColumnStart : null;
+            Composite newFinish = equals(stopKey(), key) && !lastPartitionColumnFinish.isEmpty() ? lastPartitionColumnFinish : null;
 
             List<ColumnSlice> newSlices = new ArrayList<ColumnSlice>(sliceFilter.slices.length); // in the common case, we'll have the same number of slices
 

src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java

@@ -302,7 +302,11 @@ public void setStart(Composite start)
 
     public int lastCounted()
     {
-        return columnCounter == null ? 0 : columnCounter.live();
+        // If we have a slice limit set, columnCounter.live() can overcount by one because we have to call
+        // columnCounter.count() before we can tell if we've exceeded the slice limit (and accordingly, should not
+        // add the cells to returned container).  To deal with this overcounting, we take the min of the slice
+        // limit and the counter's count.
+        return columnCounter == null ? 0 : Math.min(columnCounter.live(), count);
     }
 
     public int lastIgnored()