fix not including tombstone-only keys in keyRange.

patch by jbellis; reviewed by Evan Weaver for CASSANDRA-286

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@794475 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -530,7 +530,12 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
      */
     static ColumnFamily removeDeleted(ColumnFamily cf)
     {
-        return removeDeleted(cf, (int) (System.currentTimeMillis() / 1000) - DatabaseDescriptor.getGcGraceInSeconds());
+        return removeDeleted(cf, getDefaultGCBefore());
+    }
+
+    public static int getDefaultGCBefore()
+    {
+        return (int)(System.currentTimeMillis() / 1000) - DatabaseDescriptor.getGcGraceInSeconds();
     }
 
     static ColumnFamily removeDeleted(ColumnFamily cf, int gcBefore)
@@ -1431,12 +1436,17 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return getColumnFamily(new TimeQueryFilter(key, columnParent, since));    
     }
 
+    public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException
+    {
+        return getColumnFamily(filter, getDefaultGCBefore());
+    }
+
     /**
      * get a list of columns starting from a given column, in a specified order.
      * only the latest version of a column is returned.
      * @return null if there is no data and no tombstones; otherwise a ColumnFamily
      */
-    public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException
+    public ColumnFamily getColumnFamily(QueryFilter filter, int gcBefore) throws IOException
     {
         // if we are querying subcolumns of a supercolumn, fetch the supercolumn with NQF, then filter in-memory.
         if (filter.path.superColumnName != null)
@@ -1450,7 +1460,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
                     filter.filterSuperColumn((SuperColumn) column);
                 }
             }
-            return removeDeleted(cf);
+            return removeDeleted(cf, gcBefore);
         }
 
         // we are querying top-level columns, do a merging fetch with indexes.
@@ -1500,9 +1510,9 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
             if (!collated.hasNext())
                 return null;
 
-            filter.collectColumns(returnCF, collated);
+            filter.collectColumns(returnCF, collated, gcBefore);
 
-            return removeDeleted(returnCF);
+            return removeDeleted(returnCF, gcBefore); // collect does a first pass but doesn't try to recognize e.g. the entire CF being tombstoned
         }
         finally
         {

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

@@ -711,7 +711,7 @@ public class Table
                 }
                 // make sure there is actually non-tombstone content associated w/ this key
                 // TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable)
-                if (ColumnFamilyStore.removeDeleted(cfs.getColumnFamily(new SliceQueryFilter(current, new QueryPath(cfName), "", "", true, 1)), Integer.MAX_VALUE) != null)
+                if (cfs.getColumnFamily(new SliceQueryFilter(current, new QueryPath(cfName), "", "", true, 1), Integer.MAX_VALUE) != null)
                 {
                     keys.add(current);
                 }

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

@@ -49,11 +49,12 @@ public class NamesQueryFilter extends QueryFilter
         }
     }
 
-    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns)
+    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns, int gcBefore)
     {
         for (IColumn column : reducedColumns)
         {
-            returnCF.addColumn(column);
+            if (!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore)
+                returnCF.addColumn(column);
         }
     }
 }

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

@@ -36,7 +36,7 @@ public abstract class QueryFilter
      * by the filter code, which should have some limit on the number of columns
      * to avoid running out of memory on large rows.
      */
-    public abstract void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns);
+    public abstract void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns, int gcBefore);
 
     /**
      * subcolumns of a supercolumn are unindexed, so to pick out parts of those we operate in-memory.
@@ -55,7 +55,7 @@ public abstract class QueryFilter
         };
     }
     
-    public void collectColumns(final ColumnFamily returnCF, Iterator collatedColumns)
+    public void collectColumns(final ColumnFamily returnCF, Iterator collatedColumns, int gcBefore)
     {
         // define a 'reduced' iterator that merges columns w/ the same name, which
         // greatly simplifies computing liveColumns in the presence of tombstones.
@@ -81,7 +81,7 @@ public abstract class QueryFilter
             }
         };
 
-        collectColumns(returnCF, reduced);
+        collectColumns(returnCF, reduced, gcBefore);
     }
 
     public String getColumnFamilyName()

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

@@ -47,7 +47,7 @@ public class SliceQueryFilter extends QueryFilter
         return isAscending ? comparator : new ReverseComparator(comparator);
     }
 
-    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns)
+    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns, int gcBefore)
     {
         int liveColumns = 0;
 
@@ -59,10 +59,12 @@ public class SliceQueryFilter extends QueryFilter
                 && ((isAscending && column.name().compareTo(finish) > 0))
                     || (!isAscending && column.name().compareTo(finish) < 0))
                 break;
+
             if (!column.isMarkedForDelete())
                 liveColumns++;
 
-            returnCF.addColumn(column);
+            if (!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore)
+                returnCF.addColumn(column);
         }
     }
 }

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

@@ -33,11 +33,12 @@ public class TimeQueryFilter extends QueryFilter
         return ColumnComparatorFactory.timestampComparator_;
     }
 
-    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns)
+    public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns, int gcBefore)
     {
         for (IColumn column : reducedColumns)
         {
-            returnCF.addColumn(column);
+            if (!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore)
+                returnCF.addColumn(column);
         }
     }
 

test/system/test_server.py

@@ -273,7 +273,15 @@ class TestMutations(CassandraTester):
         client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c1'), 1, True)
         client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c2'), 1, True)
         actual = client.get_key_range('Table1', 'Standard1', '', '', 1000)
-        assert not actual, actual
+        assert actual == [], actual
+
+    def test_range_with_remove_cf(self):
+        _insert_simple()
+        assert client.get_key_range('Table1', 'Standard1', 'key1', '', 1000) == ['key1']
+
+        client.remove('Table1', 'key1', ColumnPathOrParent('Standard1'), 1, True)
+        actual = client.get_key_range('Table1', 'Standard1', '', '', 1000)
+        assert actual == [], actual
 
     def test_range_collation(self):
         for key in ['-a', '-b', 'a', 'b'] + [str(i) for i in xrange(100)]: