Merge branch 'cassandra-1.2' into cassandra-2.0

Conflicts:
	CHANGES.txt
	src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java

CHANGES.txt

@@ -16,8 +16,10 @@
  * Fix infinite loop on exception while streaming (CASSANDRA-7330)
  * Reference sstables before populating key cache (CASSANDRA-7234)
 Merged from 1.2:
+ * Don't insert tombstones that hide indexed values into 2i (CASSANDRA-7268)
  * Track metrics at a keyspace level (CASSANDRA-6539)
- * Add replace_address_first_boot flag to only replace if not bootstrapped (CASSANDRA-7356)
+ * Add replace_address_first_boot flag to only replace if not bootstrapped
+   (CASSANDRA-7356)
  * Enable keepalive for native protocol (CASSANDRA-7380)
  * Check internal addresses for seeds (CASSANDRA-6523)
  * Fix potential / by 0 in HHOM page size calculation (CASSANDRA-7354)

src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java

@@ -632,7 +632,14 @@ public class SecondaryIndexManager
                     // where the row is invisible to both queries (the opposite seems preferable); see CASSANDRA-5540
                     if (!column.isMarkedForDelete(System.currentTimeMillis()))
                         ((PerColumnSecondaryIndex) index).insert(key.key, column);
-                    ((PerColumnSecondaryIndex) index).delete(key.key, oldColumn);
+
+                    // Usually we want to delete the old value from the index, except when
+                    // name/value/timestamp are all equal, but the columns themselves
+                    // are not (as is the case when overwriting expiring columns with
+                    // identical values and ttl) Then, we don't want to delete as the
+                    // tombstone will hide the new value we just inserted; see CASSANDRA-7268
+                    if (shouldCleanupOldValue(oldColumn, column))
+                        ((PerColumnSecondaryIndex) index).delete(key.key, oldColumn);
                 }
             }
         }
@@ -652,5 +659,21 @@ public class SecondaryIndexManager
             for (SecondaryIndex index : rowLevelIndexMap.values())
                 ((PerRowSecondaryIndex) index).index(key.key, cf);
         }
+
+        private boolean shouldCleanupOldValue(Column oldColumn, Column newColumn)
+        {
+            // If any one of name/value/timestamp are different, then we
+            // should delete from the index. If not, then we can infer that
+            // at least one of the columns is an ExpiringColumn and that the
+            // difference is in the expiry time. In this case, we don't want to
+            // delete the old value from the index as the tombstone we insert
+            // will just hide the inserted value.
+            // Completely identical columns (including expiring columns with
+            // identical ttl & localExpirationTime) will not get this far due
+            // to the oldColumn.equals(newColumn) in StandardUpdater.update
+            return !oldColumn.name().equals(newColumn.name())
+                || !oldColumn.value().equals(newColumn.value())
+                || oldColumn.timestamp() != newColumn.timestamp();
+        }
     }
 }

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

@@ -25,6 +25,7 @@ import java.nio.charset.CharacterCodingException;
 import java.util.*;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import com.google.common.base.Function;
 import com.google.common.collect.Iterables;
@@ -423,6 +424,60 @@ public class ColumnFamilyStoreTest extends SchemaLoader
 
     }
 
+    @Test
+    public void testIndexUpdateOverwritingExpiringColumns() throws Exception
+    {
+        // see CASSANDRA-7268
+        Keyspace keyspace = Keyspace.open("Keyspace2");
+
+        // create a row and update the birthdate value with an expiring column
+        RowMutation rm;
+        rm = new RowMutation("Keyspace2", ByteBufferUtil.bytes("k100"));
+        rm.add("Indexed1", ByteBufferUtil.bytes("birthdate"), ByteBufferUtil.bytes(100L), 1, 1000);
+        rm.apply();
+
+        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), IndexOperator.EQ, ByteBufferUtil.bytes(100L));
+        List<IndexExpression> clause = Arrays.asList(expr);
+        IDiskAtomFilter filter = new IdentityQueryFilter();
+        Range<RowPosition> range = Util.range("", "");
+        List<Row> rows = keyspace.getColumnFamilyStore("Indexed1").search(range, clause, filter, 100);
+        assertEquals(1, rows.size());
+
+        // requires a 1s sleep because we calculate local expiry time as (now() / 1000) + ttl
+        TimeUnit.SECONDS.sleep(1);
+
+        // now overwrite with the same name/value/ttl, but the local expiry time will be different
+        rm = new RowMutation("Keyspace2", ByteBufferUtil.bytes("k100"));
+        rm.add("Indexed1", ByteBufferUtil.bytes("birthdate"), ByteBufferUtil.bytes(100L), 1, 1000);
+        rm.apply();
+
+        rows = keyspace.getColumnFamilyStore("Indexed1").search(range, clause, filter, 100);
+        assertEquals(1, rows.size());
+
+        // check that modifying the indexed value using the same timestamp behaves as expected
+        rm = new RowMutation("Keyspace2", ByteBufferUtil.bytes("k101"));
+        rm.add("Indexed1", ByteBufferUtil.bytes("birthdate"), ByteBufferUtil.bytes(101L), 1, 1000);
+        rm.apply();
+
+        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), IndexOperator.EQ, ByteBufferUtil.bytes(101L));
+        clause = Arrays.asList(expr);
+        rows = keyspace.getColumnFamilyStore("Indexed1").search(range, clause, filter, 100);
+        assertEquals(1, rows.size());
+
+        TimeUnit.SECONDS.sleep(1);
+        rm = new RowMutation("Keyspace2", ByteBufferUtil.bytes("k101"));
+        rm.add("Indexed1", ByteBufferUtil.bytes("birthdate"), ByteBufferUtil.bytes(102L), 1, 1000);
+        rm.apply();
+        // search for the old value
+        rows = keyspace.getColumnFamilyStore("Indexed1").search(range, clause, filter, 100);
+        assertEquals(0, rows.size());
+        // and for the new
+        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), IndexOperator.EQ, ByteBufferUtil.bytes(102L));
+        clause = Arrays.asList(expr);
+        rows = keyspace.getColumnFamilyStore("Indexed1").search(range, clause, filter, 100);
+        assertEquals(1, rows.size());
+    }
+
     @Test
     public void testDeleteOfInconsistentValuesInKeysIndex() throws Exception
     {