Merge branch 'cassandra-2.1' into trunk

CHANGES.txt

@@ -19,6 +19,7 @@
  * Optimise NativeCell comparisons (CASSANDRA-6755)
 
 2.1.0-rc3
+ * Consider expiry when reconciling otherwise equal cells (CASSANDRA-7403)
  * Introduce CQL support for stress tool (CASSANDRA-6146)
  * Fix ClassCastException processing expired messages (CASSANDRA-7496)
  * Fix prepared marker for collections inside UDT (CASSANDRA-7472)

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

@@ -120,16 +120,12 @@ public abstract class AbstractCell implements Cell
 
     public Cell reconcile(Cell cell)
     {
-        // tombstones take precedence.  (if both are tombstones, then it doesn't matter which one we use.)
-        if (!isLive())
-            return timestamp() < cell.timestamp() ? cell : this;
-        if (!cell.isLive())
-            return timestamp() > cell.timestamp() ? this : cell;
-        // break ties by comparing values.
-        if (timestamp() == cell.timestamp())
-            return value().compareTo(cell.value()) < 0 ? cell : this;
-        // neither is tombstoned and timestamps are different
-        return timestamp() < cell.timestamp() ? cell : this;
+        long ts1 = timestamp(), ts2 = cell.timestamp();
+        if (ts1 != ts2)
+            return ts1 < ts2 ? cell : this;
+        if (isLive() != cell.isLive())
+            return isLive() ? cell : this;
+        return value().compareTo(cell.value()) < 0 ? cell : this;
     }
 
     @Override

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

@@ -142,6 +142,28 @@ public class BufferExpiringCell extends BufferCell implements ExpiringCell
             throw new MarshalException("The local expiration time should not be negative");
     }
 
+    public Cell reconcile(Cell cell)
+    {
+        long ts1 = timestamp(), ts2 = cell.timestamp();
+        if (ts1 != ts2)
+            return ts1 < ts2 ? cell : this;
+        // we should prefer tombstones
+        if (cell instanceof DeletedCell)
+            return cell;
+        // however if we're both ExpiringCells, we should prefer the one with the longest ttl
+        // (really in preference _always_ to the value comparison)
+        int c = value().compareTo(cell.value());
+        if (c != 0)
+            return c < 0 ? cell : this;
+        if (cell instanceof ExpiringCell)
+        {
+            int let1 = localExpirationTime, let2 = cell.getLocalDeletionTime();
+            if (let1 < let2)
+                return cell;
+        }
+        return this;
+    }
+
     @Override
     public boolean equals(Cell cell)
     {

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

@@ -128,6 +128,28 @@ public class NativeExpiringCell extends NativeCell implements ExpiringCell
         FBUtilities.updateWithInt(digest, getTimeToLive());
     }
 
+    public Cell reconcile(Cell cell)
+    {
+        long ts1 = timestamp(), ts2 = cell.timestamp();
+        if (ts1 != ts2)
+            return ts1 < ts2 ? cell : this;
+        // we should prefer tombstones
+        if (cell instanceof DeletedCell)
+            return cell;
+        // however if we're both ExpiringCells, we should prefer the one with the longest ttl
+        // (really in preference _always_ to the value comparison)
+        int c = value().compareTo(cell.value());
+        if (c != 0)
+            return c < 0 ? cell : this;
+        if (cell instanceof ExpiringCell)
+        {
+            int let1 = getLocalDeletionTime(), let2 = cell.getLocalDeletionTime();
+            if (let1 < let2)
+                return cell;
+        }
+        return this;
+    }
+
     public boolean equals(Cell cell)
     {
         return cell instanceof ExpiringCell && equals((ExpiringCell) cell);

test/unit/org/apache/cassandra/db/CellTest.java

@@ -0,0 +1,76 @@
+package org.apache.cassandra.db;
+
+import org.junit.Test;
+
+import junit.framework.Assert;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.concurrent.OpOrder;
+import org.apache.cassandra.utils.memory.NativeAllocator;
+import org.apache.cassandra.utils.memory.NativePool;
+
+public class CellTest
+{
+
+    private static final OpOrder order = new OpOrder();
+    private static NativeAllocator allocator = new NativePool(Integer.MAX_VALUE, Integer.MAX_VALUE, 1f, null).newAllocator();
+
+    @Test
+    public void testExpiringCellReconile()
+    {
+        // equal
+        Assert.assertEquals(0, testExpiring("a", "a", 1, 1, null, null, null, null));
+
+        // newer timestamp
+        Assert.assertEquals(-1, testExpiring("a", "a", 2, 1, null, null, 1L, null));
+        Assert.assertEquals(-1, testExpiring("a", "a", 2, 1, null, "b", 1L, 2));
+
+        // newer TTL
+        Assert.assertEquals(-1, testExpiring("a", "a", 1, 2, null, null, null, 1));
+        Assert.assertEquals(1, testExpiring("a", "a", 1, 2, null, "b", null, 1));
+
+        // newer value
+        Assert.assertEquals(-1, testExpiring("a", "b", 2, 1, null, "a", null, null));
+        Assert.assertEquals(-1, testExpiring("a", "b", 2, 1, null, "a", null, 2));
+    }
+
+    private int testExpiring(String n1, String v1, long t1, int et1, String n2, String v2, Long t2, Integer et2)
+    {
+        if (n2 == null)
+            n2 = n1;
+        if (v2 == null)
+            v2 = v1;
+        if (t2 == null)
+            t2 = t1;
+        if (et2 == null)
+            et2 = et1;
+        int result = testExpiring(n1, v1, t1, et1, false, n2, v2, t2, et2, false);
+        Assert.assertEquals(result, testExpiring(n1, v1, t1, et1, false, n2, v2, t2, et2, true));
+        Assert.assertEquals(result, testExpiring(n1, v1, t1, et1, true, n2, v2, t2, et2, false));
+        Assert.assertEquals(result, testExpiring(n1, v1, t1, et1, true, n2, v2, t2, et2, true));
+        return result;
+    }
+
+    private int testExpiring(String n1, String v1, long t1, int et1, boolean native1, String n2, String v2, long t2, int et2, boolean native2)
+    {
+        Cell c1 = expiring(n1, v1, t1, et1, native1);
+        Cell c2 = expiring(n2, v2, t2, et2, native2);
+        return reconcile(c1, c2);
+    }
+
+    int reconcile(Cell c1, Cell c2)
+    {
+        if (c1.reconcile(c2) == c1)
+            return c2.reconcile(c1) == c1 ? -1 : 0;
+        return c2.reconcile(c1) == c2 ? 1 : 0;
+    }
+
+    private Cell expiring(String name, String value, long timestamp, int expirationTime, boolean nativeCell)
+    {
+        ExpiringCell cell = new BufferExpiringCell(Util.cellname(name), ByteBufferUtil.bytes(value), timestamp, 1, expirationTime);
+        if (nativeCell)
+            cell = new NativeExpiringCell(allocator, order.getCurrent(), cell);
+        return cell;
+    }
+
+}