need to include column container's deletion status when determining whether to include a column in the live count.

patch by jbellis; reviewed by Jun Rao for CASSANDRA-386

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

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

@@ -125,6 +125,11 @@ public final class Column implements IColumn
         return timestamp;
     }
 
+    public long mostRecentChangeAt()
+    {
+        return timestamp;
+    }
+
     public int size()
     {
         /*

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

@@ -27,6 +27,7 @@ public interface IColumn
     public static short UtfPrefix_ = 2;
     public boolean isMarkedForDelete();
     public long getMarkedForDeleteAt();
+    public long mostRecentChangeAt();
     public byte[] name();
     public int size();
     public int serializedSize();

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

@@ -27,5 +27,8 @@ public interface IColumnContainer
 {
     public void addColumn(IColumn column);
 
+    public boolean isMarkedForDelete();
+    public long getMarkedForDeleteAt();
+
     public AbstractType getComparator();
 }

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

@@ -139,13 +139,25 @@ public final class SuperColumn implements IColumn, IColumnContainer
     public long timestamp(byte[] columnName)
     {
     	IColumn column = columns_.get(columnName);
-    	if ( column instanceof SuperColumn )
-    		throw new UnsupportedOperationException("A super column cannot hold other super columns.");
+    	assert column instanceof Column;
     	if ( column != null )
     		return column.timestamp();
     	throw new IllegalArgumentException("Timestamp was requested for a column that does not exist.");
     }
 
+    public long mostRecentChangeAt()
+    {
+        long max = Long.MIN_VALUE;
+        for (IColumn column : columns_.values())
+        {
+            if (column.mostRecentChangeAt() > max)
+            {
+                max = column.mostRecentChangeAt();
+            }
+        }
+        return max;
+    }
+
     public byte[] value()
     {
     	throw new UnsupportedOperationException("This operation is not supported for Super Columns.");

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

@@ -27,6 +27,7 @@ import java.util.Iterator;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.log4j.Logger;
 import org.apache.commons.collections.comparators.ReverseComparator;
 import org.apache.commons.collections.iterators.ReverseListIterator;
 import org.apache.commons.collections.IteratorUtils;
@@ -38,6 +39,8 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 
 public class SliceQueryFilter extends QueryFilter
 {
+    private static Logger logger = Logger.getLogger(SliceQueryFilter.class);
+
     public final byte[] start, finish;
     public final boolean reversed;
     public final int count;
@@ -104,19 +107,35 @@ public class SliceQueryFilter extends QueryFilter
 
         while (reducedColumns.hasNext())
         {
-            IColumn column = reducedColumns.next();
             if (liveColumns >= count)
                 break;
+
+            IColumn column = reducedColumns.next();
+            if (logger.isDebugEnabled())
+                logger.debug("collecting " + column.getString(comparator));
+
             if (finish.length > 0
                 && ((!reversed && comparator.compare(column.name(), finish) > 0))
                     || (reversed && comparator.compare(column.name(), finish) < 0))
                 break;
 
-            if (!column.isMarkedForDelete())
+            // only count live columns towards the `count` criteria
+            if (!column.isMarkedForDelete()
+                && (!container.isMarkedForDelete()
+                    || column.mostRecentChangeAt() > container.getMarkedForDeleteAt()))
+            {
                 liveColumns++;
+            }
 
-            if (!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore)
+            // but we need to add all non-gc-able columns to the result for read repair:
+            // the column itself must be not gc-able, (1)
+            // and if its container is deleted, the column must be changed more recently than the container tombstone (2)
+            // (since otherwise, the only thing repair cares about is the container tombstone)
+            if ((!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore) // (1)
+                && (!container.isMarkedForDelete() || column.mostRecentChangeAt() > container.getMarkedForDeleteAt())) // (2)
+            {
                 container.addColumn(column);
+            }
         }
     }
 }

test/system/test_server.py

@@ -235,6 +235,23 @@ class TestMutations(CassandraTester):
                  for result in client.get_slice('Keyspace2', 'key1', column_parent, p, ConsistencyLevel.ONE)]
         assert slice == [Column(L[2].bytes, 'value2', 2)], slice
         
+    def test_long_remove(self):
+        column_parent = ColumnParent('StandardLong1')
+        sp = SlicePredicate(slice_range=SliceRange('', '', False, 1))
+
+        for i in xrange(10):
+            path = ColumnPath('StandardLong1', column=_i64(i))
+
+            client.insert('Keyspace1', 'key1', path, 'value1', 10 * i, ConsistencyLevel.ONE)
+            client.remove('Keyspace1', 'key1', ColumnPath('StandardLong1'), 10 * i + 1, ConsistencyLevel.ONE)
+            slice = client.get_slice('Keyspace1', 'key1', column_parent, sp, ConsistencyLevel.ONE)
+            assert slice == [], slice
+            # resurrect
+            client.insert('Keyspace1', 'key1', path, 'value2', 10 * i + 2, ConsistencyLevel.ONE)
+            slice = [result.column
+                     for result in client.get_slice('Keyspace1', 'key1', column_parent, sp, ConsistencyLevel.ONE)]
+            assert slice == [Column(_i64(i), 'value2', 10 * i + 2)], (slice, i)
+        
     def test_batch_insert(self):
         _insert_batch(False)
         time.sleep(0.1)