Make reading of range tombstones more reliable

Patch by Alex Petrov; reviewed by Benjamin Lerer for CASSANDRA-12811

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.13
+ * Make reading of range tombstones more reliable (CASSANDRA-12811)
  * Fix startup problems due to schema tables not completely flushed (CASSANDRA-12213)
  * Fix view builder bug that can filter out data on restart (CASSANDRA-13405)
  * Fix 2i page size calculation when there are no regular columns (CASSANDRA-13400)

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

@@ -736,13 +736,13 @@ private UnfilteredRowIterator queryMemtableAndSSTablesInTimestampOrder(ColumnFam
 
                 // We need to get the partition deletion and include it if it's live. In any case though, we're done with that sstable.
                 sstable.incrementReadCount();
-                try (UnfilteredRowIterator iter = sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift()))
+                try (UnfilteredRowIterator iter = filter.filter(sstable.iterator(partitionKey(), columnFilter(), filter.isReversed(), isForThrift())))
                 {
+                    sstablesIterated++;
                     if (!iter.partitionLevelDeletion().isLive())
-                    {
-                        sstablesIterated++;
                         result = add(UnfilteredRowIterators.noRowsIterator(iter.metadata(), iter.partitionKey(), Rows.EMPTY_STATIC_ROW, iter.partitionLevelDeletion(), filter.isReversed()), result, filter, sstable.isRepaired());
-                    }
+                    else
+                        result = add(iter, result, filter, sstable.isRepaired());
                 }
                 continue;
             }
@@ -835,9 +835,6 @@ private ClusteringIndexNamesFilter reduceFilter(ClusteringIndexNamesFilter filte
         NavigableSet<Clustering> toRemove = null;
         for (Clustering clustering : clusterings)
         {
-            if (!searchIter.hasNext())
-                break;
-
             Row row = searchIter.next(clustering);
             if (row == null || !canRemoveRow(row, columns.regulars, sstableTimestamp))
                 continue;

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

@@ -176,7 +176,9 @@ public Unfiltered next()
 
     public UnfilteredRowIterator getUnfilteredRowIterator(final ColumnFilter columnFilter, final Partition partition)
     {
+        final Iterator<Clustering> clusteringIter = clusteringsInQueryOrder.iterator();
         final SearchIterator<Clustering, Row> searcher = partition.searchIterator(columnFilter, reversed);
+
         return new AbstractUnfilteredRowIterator(partition.metadata(),
                                         partition.partitionKey(),
                                         partition.partitionLevelDeletion(),
@@ -185,11 +187,9 @@ public UnfilteredRowIterator getUnfilteredRowIterator(final ColumnFilter columnF
                                         reversed,
                                         partition.stats())
         {
-            private final Iterator<Clustering> clusteringIter = clusteringsInQueryOrder.iterator();
-
             protected Unfiltered computeNext()
             {
-                while (clusteringIter.hasNext() && searcher.hasNext())
+                while (clusteringIter.hasNext())
                 {
                     Row row = searcher.next(clusteringIter.next());
                     if (row != null)

src/java/org/apache/cassandra/db/partitions/AbstractBTreePartition.java

@@ -139,11 +139,6 @@ public SearchIterator<Clustering, Row> searchIterator(final ColumnFilter columns
             private final SearchIterator<Clustering, Row> rawIter = new BTreeSearchIterator<>(current.tree, metadata.comparator, desc(reversed));
             private final DeletionTime partitionDeletion = current.deletionInfo.getPartitionDeletion();
 
-            public boolean hasNext()
-            {
-                return rawIter.hasNext();
-            }
-
             public Row next(Clustering clustering)
             {
                 if (clustering == Clustering.STATIC_CLUSTERING)

src/java/org/apache/cassandra/utils/IndexedSearchIterator.java

@@ -19,6 +19,11 @@
 
 public interface IndexedSearchIterator<K, V> extends SearchIterator<K, V>
 {
+    /**
+     * @return true if iterator has any elements left, false otherwise
+     */
+    public boolean hasNext();
+
     /**
      * @return the value just recently returned by next()
      * @throws java.util.NoSuchElementException if next() returned null

src/java/org/apache/cassandra/utils/SearchIterator.java

@@ -19,8 +19,6 @@
 
 public interface SearchIterator<K, V>
 {
-    public boolean hasNext();
-
     /**
      * Searches "forwards" (in direction of travel) in the iterator for the required key;
      * if this or any key greater has already been returned by the iterator, the method may

test/unit/org/apache/cassandra/cql3/validation/operations/DeleteTest.java

@@ -43,15 +43,89 @@ public class DeleteTest extends CQLTester
     @Test
     public void testRangeDeletion() throws Throwable
     {
-        createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b, c))");
+        testRangeDeletion(true, true);
+        testRangeDeletion(false, true);
+        testRangeDeletion(true, false);
+        testRangeDeletion(false, false);
+    }
 
+    private void testRangeDeletion(boolean flushData, boolean flushTombstone) throws Throwable
+    {
+        createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b, c))");
         execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", 1, 1, 1, 1);
-        flush();
+        flush(flushData);
         execute("DELETE FROM %s WHERE a=? AND b=?", 1, 1);
-        flush();
+        flush(flushTombstone);
         assertEmpty(execute("SELECT * FROM %s WHERE a=? AND b=? AND c=?", 1, 1, 1));
     }
 
+    @Test
+    public void testDeleteRange() throws Throwable
+    {
+        testDeleteRange(true, true);
+        testDeleteRange(false, true);
+        testDeleteRange(true, false);
+        testDeleteRange(false, false);
+    }
+
+    private void testDeleteRange(boolean flushData, boolean flushTombstone) throws Throwable
+    {
+        createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a, b))");
+
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 1, 1, 1);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 1, 2);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 2, 3);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 3, 4);
+        flush(flushData);
+
+        execute("DELETE FROM %s WHERE a = ? AND b >= ?", 2, 2);
+        flush(flushTombstone);
+
+        assertRowsIgnoringOrder(execute("SELECT * FROM %s"),
+                                row(1, 1, 1),
+                                row(2, 1, 2));
+
+        assertRows(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 1),
+                   row(2, 1, 2));
+        assertEmpty(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 2));
+        assertEmpty(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 3));
+    }
+
+    @Test
+    public void testCrossMemSSTableMultiColumn() throws Throwable
+    {
+        testCrossMemSSTableMultiColumn(true, true);
+        testCrossMemSSTableMultiColumn(false, true);
+        testCrossMemSSTableMultiColumn(true, false);
+        testCrossMemSSTableMultiColumn(false, false);
+    }
+
+    private void testCrossMemSSTableMultiColumn(boolean flushData, boolean flushTombstone) throws Throwable
+    {
+        createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a, b))");
+
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 1, 1, 1);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 1, 2);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 2, 2);
+        execute("INSERT INTO %s (a, b, c) VALUES (?, ?, ?)", 2, 3, 3);
+        flush(flushData);
+
+        execute("DELETE FROM %s WHERE a = ? AND (b) = (?)", 2, 2);
+        execute("DELETE FROM %s WHERE a = ? AND (b) = (?)", 2, 3);
+
+        flush(flushTombstone);
+
+        assertRowsIgnoringOrder(execute("SELECT * FROM %s"),
+                                row(1, 1, 1),
+                                row(2, 1, 2));
+
+        assertRows(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 1),
+                   row(2, 1, 2));
+        assertEmpty(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 2));
+        assertEmpty(execute("SELECT * FROM %s WHERE a = ? AND b = ?", 2, 3));
+    }
+
+
     /**
      * Test simple deletion and in particular check for #4193 bug
      * migrated from cql_tests.py:TestCQL.deletion_test()
@@ -440,7 +514,7 @@ private void testDeleteWithOneClusteringColumns(boolean forceFlush) throws Throw
                 assertEmpty(execute("SELECT * FROM %s WHERE partitionKey = ? AND clustering = ?", 0, 1));
             }
 
-            execute("DELETE FROM %s WHERE partitionKey = ? AND (clustering) = (?)", 0, 1);
+            execute("DELETE FROM %s WHERE partitionKey = ? AND clustering = ?", 0, 1);
             flush(forceFlush);
             assertEmpty(execute("SELECT value FROM %s WHERE partitionKey = ? AND clustering = ?", 0, 1));
 

test/unit/org/apache/cassandra/db/partition/PartitionImplementationTest.java

@@ -326,7 +326,7 @@ void testSearchIterator(NavigableSet<Clusterable> sortedContent, Partition parti
         int pos = reversed ? KEY_RANGE : 0;
         int mul = reversed ? -1 : 1;
         boolean started = false;
-        while (searchIter.hasNext())
+        while (pos < KEY_RANGE)
         {
             int skip = rand.nextInt(KEY_RANGE / 10);
             pos += skip * mul;