diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java index 0bf298f795..cdd3dc5cbe 100644 --- a/src/java/org/apache/cassandra/db/Column.java +++ b/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() { /* diff --git a/src/java/org/apache/cassandra/db/IColumn.java b/src/java/org/apache/cassandra/db/IColumn.java index 8cbc782fde..b83d49afc8 100644 --- a/src/java/org/apache/cassandra/db/IColumn.java +++ b/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(); diff --git a/src/java/org/apache/cassandra/db/IColumnContainer.java b/src/java/org/apache/cassandra/db/IColumnContainer.java index 9f6c0703b3..c22f1c8c46 100644 --- a/src/java/org/apache/cassandra/db/IColumnContainer.java +++ b/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(); } diff --git a/src/java/org/apache/cassandra/db/SuperColumn.java b/src/java/org/apache/cassandra/db/SuperColumn.java index fc5af7b68c..59f5ccd2ae 100644 --- a/src/java/org/apache/cassandra/db/SuperColumn.java +++ b/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."); diff --git a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java b/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java index e76062c663..cf8b1754fd 100644 --- a/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java +++ b/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); + } } } } diff --git a/test/system/test_server.py b/test/system/test_server.py index 1e17f03fbe..77573ff79b 100644 --- a/test/system/test_server.py +++ b/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)