DISTINCT queries failing with range tombstones

Author: krummas

src/java/org/apache/cassandra/db/rows/LazilyInitializedUnfilteredRowIterator.java

@@ -32,8 +32,8 @@
 public abstract class LazilyInitializedUnfilteredRowIterator extends AbstractIterator<Unfiltered> implements UnfilteredRowIterator
 {
     private final DecoratedKey partitionKey;
-
     private UnfilteredRowIterator iterator;
+    private boolean closed = false;
 
     public LazilyInitializedUnfilteredRowIterator(DecoratedKey partitionKey)
     {
@@ -97,15 +97,16 @@ protected Unfiltered computeNext()
 
     public void close()
     {
+        // don't use iterator == null as indicator if this is closed since we might re-initialize after closing it in that case.
+        closed = true;
         if (iterator != null)
-        {
             iterator.close();
-            iterator = null;
-        }
     }
 
     public boolean isOpen()
     {
-        return iterator != null;
+        if (closed)
+            return false;
+        return iterator != null; // for backwards compatibility - if `maybeInit` has not been run on this class, consider it not open, for example SSTableExport seems to rely on this
     }
 }

test/distributed/org/apache/cassandra/distributed/test/DistinctReadTest.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+
+public class DistinctReadTest extends TestBaseImpl
+{
+    @Test
+    public void test() throws IOException
+    {
+        try (Cluster cluster = init(Cluster.build()
+                                           .withNodes(1)
+                                           .start()))
+        {
+            cluster.schemaChange(withKeyspace("CREATE TABLE %s.tbl (id int, ck int, x int, PRIMARY KEY (id, ck))"));
+            cluster.coordinator(1).execute(withKeyspace("DELETE FROM %s.tbl USING TIMESTAMP 100 WHERE id = 1 AND ck < 10 "), ConsistencyLevel.ONE);
+            cluster.coordinator(1).execute(withKeyspace("INSERT INTO %s.tbl (id, ck, x) VALUES (1, 5, 7) USING TIMESTAMP 101"), ConsistencyLevel.ONE);
+            cluster.get(1).flush(KEYSPACE);
+            // all these failed before fix;
+            cluster.coordinator(1).execute(withKeyspace("select distinct id from %s.tbl where token(id) > " + Long.MIN_VALUE), ConsistencyLevel.ONE);
+            cluster.coordinator(1).execute(withKeyspace("select distinct id from %s.tbl where id > 0 allow filtering"), ConsistencyLevel.ONE);
+            cluster.coordinator(1).execute(withKeyspace("select id from %s.tbl where token(id) > " + Long.MIN_VALUE +" PER PARTITION LIMIT 1"), ConsistencyLevel.ONE);
+        }
+    }
+}

test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java

@@ -169,26 +169,28 @@ private SSTableStats antiCompactRanges(ColumnFamilyStore store, RangesAtEndpoint
             {
                 while (scanner.hasNext())
                 {
-                    UnfilteredRowIterator row = scanner.next();
-                    Token token = row.partitionKey().getToken();
-                    if (sstable.isPendingRepair() && !sstable.isTransient())
+                    try (UnfilteredRowIterator row = scanner.next())
                     {
-                        assertTrue(fullContains.test(token));
-                        assertFalse(transContains.test(token));
-                        stats.pendingKeys++;
-                    }
-                    else if (sstable.isPendingRepair() && sstable.isTransient())
-                    {
-
-                        assertTrue(transContains.test(token));
-                        assertFalse(fullContains.test(token));
-                        stats.transKeys++;
-                    }
-                    else
-                    {
-                        assertFalse(fullContains.test(token));
-                        assertFalse(transContains.test(token));
-                        stats.unrepairedKeys++;
+                        Token token = row.partitionKey().getToken();
+                        if (sstable.isPendingRepair() && !sstable.isTransient())
+                        {
+                            assertTrue(fullContains.test(token));
+                            assertFalse(transContains.test(token));
+                            stats.pendingKeys++;
+                        }
+                        else if (sstable.isPendingRepair() && sstable.isTransient())
+                        {
+
+                            assertTrue(transContains.test(token));
+                            assertFalse(fullContains.test(token));
+                            stats.transKeys++;
+                        }
+                        else
+                        {
+                            assertFalse(fullContains.test(token));
+                            assertFalse(transContains.test(token));
+                            stats.unrepairedKeys++;
+                        }
                     }
                 }
             }

test/unit/org/apache/cassandra/db/compaction/LeveledCompactionStrategyTest.java

@@ -51,6 +51,7 @@
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
+import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
@@ -286,7 +287,12 @@ public void testCompactionProgress() throws Exception
         ISSTableScanner scanner = scanners.get(0);
         // scan through to the end
         while (scanner.hasNext())
-            scanner.next();
+        {
+            try (UnfilteredRowIterator ignored = scanner.next())
+            {
+                // just close the iterator
+            }
+        }
 
         // scanner.getCurrentPosition should be equal to total bytes of L1 sstables
         assertEquals(scanner.getCurrentPosition(), SSTableReader.getTotalUncompressedBytes(sstables));

test/unit/org/apache/cassandra/db/compaction/TTLExpiryTest.java

@@ -258,8 +258,10 @@ public void testNoExpire() throws InterruptedException, IOException
         assertTrue(scanner.hasNext());
         while(scanner.hasNext())
         {
-            UnfilteredRowIterator iter = scanner.next();
-            assertEquals(Util.dk(noTTLKey), iter.partitionKey());
+            try (UnfilteredRowIterator iter = scanner.next())
+            {
+                assertEquals(Util.dk(noTTLKey), iter.partitionKey());
+            }
         }
         scanner.close();
     }

test/unit/org/apache/cassandra/db/rows/ThrottledUnfilteredIteratorTest.java

@@ -214,8 +214,6 @@ else if (ck1 == ck2 - 1) // cell tombstone
         {
             try (UnfilteredRowIterator rowIterator = scanner.next())
             {
-                // only 1 partition data
-                assertFalse(scanner.hasNext());
                 List<Unfiltered> expectedUnfiltereds = new ArrayList<>();
                 rowIterator.forEachRemaining(expectedUnfiltereds::add);
 
@@ -227,15 +225,17 @@ else if (ck1 == ck2 - 1) // cell tombstone
                         assertTrue(scannerForThrottle.hasNext());
                         try (UnfilteredRowIterator rowIteratorForThrottle = scannerForThrottle.next())
                         {
-                            assertFalse(scannerForThrottle.hasNext());
                             verifyThrottleIterator(expectedUnfiltereds,
                                                    rowIteratorForThrottle,
                                                    new ThrottledUnfilteredIterator(rowIteratorForThrottle, throttle),
                                                    throttle);
                         }
+                        assertFalse(scannerForThrottle.hasNext());
                     }
                 }
             }
+            // only 1 partition data
+            assertFalse(scanner.hasNext());
         }
     }
 

test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java

@@ -345,7 +345,12 @@ public void testSingleDataRangeWithMovedStart() throws IOException
         // full range scan
         ISSTableScanner scanner = sstable.getScanner();
         for (int i = 4; i < 10; i++)
-            assertEquals(toKey(i), new String(scanner.next().partitionKey().getKey().array()));
+        {
+            try (UnfilteredRowIterator row = scanner.next())
+            {
+                assertEquals(toKey(i), new String(row.partitionKey().getKey().array()));
+            }
+        }
 
         scanner.close();