fix updating index when value is changed. patch by jbellis; tested by Tyler Hobbs for CASSANDRA-1373

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

CHANGES.txt

@@ -10,6 +10,7 @@ dev
  * merge StorageProxy.mutate, mutateBlocking (CASSANDRA-1396)
  * faster UUIDType, LongType comparisons (CASSANDRA-1386, 1393)
  * fix setting read_repair_chance from CLI addColumnFamily (CASSANDRA-1399)
+ * fix updates to indexed columns (CASSANDRA-1373)
 
 
 0.7-beta1

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

@@ -1091,10 +1091,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
             byte[] dataKey = null;
             int n = 0;
-            Iterator<byte[]> iter = indexRow.getColumnNames().iterator();
+            for (IColumn column : indexRow.getSortedColumns())
-            while (iter.hasNext())
             {
-                dataKey = iter.next();
+                if (column.isMarkedForDelete())
+                    continue;
+                dataKey = column.name();
                 n++;
                 DecoratedKey dk = partitioner_.decorateKey(dataKey);
                 if (!range.right.equals(partitioner_.getMinimumToken()) && range.right.compareTo(dk.token) < 0)
@@ -1436,7 +1437,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         return ColumnFamily.create(indexedColumns_.get(column).metadata);
     }
 
-    public DecoratedKey getIndexKeyFor(byte[] name, byte[] value)
+    public DecoratedKey<LocalToken> getIndexKeyFor(byte[] name, byte[] value)
     {
         return indexedColumns_.get(name).partitioner_.decorateKey(value);
     }

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

@@ -32,6 +32,7 @@ import com.google.common.collect.Iterables;
 
 import org.apache.cassandra.config.*;
 import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.dht.LocalToken;
 import org.apache.cassandra.io.sstable.SSTableDeletingReference;
 import org.apache.cassandra.io.sstable.SSTableReader;
 import org.apache.cassandra.io.util.FileUtils;
@@ -389,7 +390,7 @@ public class Table
                         for (byte[] columnName : mutatedIndexedColumns)
                         {
                             IColumn column = columnFamily.getColumn(columnName);
-                            DecoratedKey valueKey = cfs.getIndexKeyFor(columnName, column.value());
+                            DecoratedKey<LocalToken> valueKey = cfs.getIndexKeyFor(columnName, column.value());
                             ColumnFamily cf = cfs.newIndexedColumnFamily(columnName);
                             cf.addColumn(new Column(mutation.key(), ArrayUtils.EMPTY_BYTE_ARRAY, column.clock()));
                             applyCF(cfs.getIndexedColumnFamilyStore(columnName), valueKey, cf, memtablesToFlush);
@@ -403,10 +404,10 @@ public class Table
                             {
                                 byte[] columnName = entry.getKey();
                                 IColumn column = entry.getValue();
-                                DecoratedKey valueKey = cfs.getIndexKeyFor(columnName, column.value());
+                                DecoratedKey<LocalToken> valueKey = cfs.getIndexKeyFor(columnName, column.value());
                                 ColumnFamily cf = cfs.newIndexedColumnFamily(columnName);
                                 cf.deleteColumn(mutation.key(), localDeletionTime, column.clock());
-                                applyCF(cfs, valueKey, cf, memtablesToFlush);
+                                applyCF(cfs.getIndexedColumnFamilyStore(columnName), valueKey, cf, memtablesToFlush);
                             }
                         }
                     }

test/conf/cassandra.yaml

@@ -80,7 +80,6 @@ keyspaces:
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 1
       column_families:
-
         - name: Standard1
 
         - name: Standard3
@@ -92,6 +91,12 @@ keyspaces:
           column_type: Super
           compare_subcolumns_with: TimeUUIDType
 
+        - name: Indexed1
+          column_metadata:
+            - name: birthdate
+              validator_class: LongType
+              index_type: KEYS
+
     - name: Keyspace3
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 5

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

@@ -189,6 +189,33 @@ public class ColumnFamilyStoreTest extends CleanupHelper
         assert Arrays.equals(FBUtilities.toByteArray(1L), rows.get(1).cf.getColumn("birthdate".getBytes("UTF8")).value());
     }
 
+    @Test
+    public void testIndexUpdate() throws IOException
+    {
+        RowMutation rm;
+
+        rm = new RowMutation("Keyspace2", "k1".getBytes());
+        rm.add(new QueryPath("Indexed1", null, "birthdate".getBytes("UTF8")), FBUtilities.toByteArray(1L), new TimestampClock(1));
+        rm.apply();
+
+        rm = new RowMutation("Keyspace2", "k1".getBytes());
+        rm.add(new QueryPath("Indexed1", null, "birthdate".getBytes("UTF8")), FBUtilities.toByteArray(2L), new TimestampClock(2));
+        rm.apply();
+
+        IndexExpression expr = new IndexExpression("birthdate".getBytes("UTF8"), IndexOperator.EQ, FBUtilities.toByteArray(1L));
+        IndexClause clause = new IndexClause(Arrays.asList(expr), ArrayUtils.EMPTY_BYTE_ARRAY, 100);
+        IFilter filter = new IdentityQueryFilter();
+        IPartitioner p = StorageService.getPartitioner();
+        Range range = new Range(p.getMinimumToken(), p.getMinimumToken());
+        List<Row> rows = Table.open("Keyspace2").getColumnFamilyStore("Indexed1").scan(clause, range, filter);
+        assert rows.size() == 0;
+
+        expr = new IndexExpression("birthdate".getBytes("UTF8"), IndexOperator.EQ, FBUtilities.toByteArray(2L));
+        clause = new IndexClause(Arrays.asList(expr), ArrayUtils.EMPTY_BYTE_ARRAY, 100);
+        rows = Table.open("Keyspace2").getColumnFamilyStore("Indexed1").scan(clause, range, filter);
+        assert Arrays.equals("k1".getBytes(), rows.get(0).key.key);
+    }
+
     private ColumnFamilyStore insertKey1Key2() throws IOException, ExecutionException, InterruptedException
     {
         List<RowMutation> rms = new LinkedList<RowMutation>();