cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject [01/51] [partial] cassandra git commit: Storage engine refactor, a.k.a CASSANDRA-8099
Date Tue, 30 Jun 2015 10:47:25 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk 5c31a8633 -> a991b6481


http://git-wip-us.apache.org/repos/asf/cassandra/blob/a991b648/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
index c85b2e0..cf17df7 100644
--- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
+++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
@@ -21,119 +21,43 @@ package org.apache.cassandra.db;
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.nio.charset.CharacterCodingException;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-import java.util.SortedSet;
-import java.util.TreeSet;
-import java.util.UUID;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
+import java.util.*;
 
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Sets;
-
-import org.apache.cassandra.db.index.PerRowSecondaryIndexTest;
-import org.apache.cassandra.io.sstable.*;
-import org.apache.cassandra.io.sstable.format.SSTableReader;
-import org.apache.cassandra.io.sstable.format.SSTableWriter;
-import org.apache.commons.lang3.ArrayUtils;
-import org.apache.commons.lang3.StringUtils;
+import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
-import org.apache.cassandra.OrderedJUnit4ClassRunner;
-import org.apache.cassandra.SchemaLoader;
-import org.apache.cassandra.Util;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.google.common.collect.Iterators;
+import org.apache.cassandra.*;
 import org.apache.cassandra.config.*;
 import org.apache.cassandra.cql3.Operator;
-import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
-import org.apache.cassandra.db.composites.CellName;
-import org.apache.cassandra.db.composites.CellNameType;
-import org.apache.cassandra.db.composites.CellNames;
-import org.apache.cassandra.db.composites.Composites;
-import org.apache.cassandra.db.filter.ColumnSlice;
-import org.apache.cassandra.db.filter.IDiskAtomFilter;
-import org.apache.cassandra.db.filter.NamesQueryFilter;
-import org.apache.cassandra.db.filter.QueryFilter;
-import org.apache.cassandra.db.filter.SliceQueryFilter;
-import org.apache.cassandra.db.index.SecondaryIndex;
-import org.apache.cassandra.db.marshal.IntegerType;
-import org.apache.cassandra.db.marshal.LexicalUUIDType;
-import org.apache.cassandra.db.marshal.LongType;
-import org.apache.cassandra.db.marshal.UTF8Type;
-import org.apache.cassandra.dht.Bounds;
-import org.apache.cassandra.dht.ExcludingBounds;
-import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.IncludingExcludingBounds;
-import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.db.rows.*;
+import org.apache.cassandra.db.partitions.*;
+import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.db.partitions.PartitionIterator;
 import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
-import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.locator.SimpleStrategy;
 import org.apache.cassandra.metrics.ClearableHistogram;
-import org.apache.cassandra.service.ActiveRepairService;
-import org.apache.cassandra.service.StorageService;
-import org.apache.cassandra.thrift.SlicePredicate;
-import org.apache.cassandra.thrift.SliceRange;
-import org.apache.cassandra.thrift.ThriftValidation;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.WrappedRunnable;
-import org.apache.thrift.TException;
-
-import static org.apache.cassandra.Util.cellname;
-import static org.apache.cassandra.Util.column;
-import static org.apache.cassandra.Util.dk;
-import static org.apache.cassandra.Util.rp;
-import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
 
 @RunWith(OrderedJUnit4ClassRunner.class)
 public class ColumnFamilyStoreTest
 {
-    static byte[] bytes1, bytes2;
     public static final String KEYSPACE1 = "ColumnFamilyStoreTest1";
     public static final String KEYSPACE2 = "ColumnFamilyStoreTest2";
-    public static final String KEYSPACE3 = "ColumnFamilyStoreTest3";
-    public static final String KEYSPACE4 = "PerRowSecondaryIndex";
     public static final String CF_STANDARD1 = "Standard1";
     public static final String CF_STANDARD2 = "Standard2";
-    public static final String CF_STANDARD3 = "Standard3";
-    public static final String CF_STANDARD4 = "Standard4";
-    public static final String CF_STANDARD5 = "Standard5";
-    public static final String CF_STANDARDINT = "StandardInteger1";
     public static final String CF_SUPER1 = "Super1";
     public static final String CF_SUPER6 = "Super6";
-    public static final String CF_INDEX1 = "Indexed1";
-    public static final String CF_INDEX2 = "Indexed2";
-    public static final String CF_INDEX3 = "Indexed3";
-
-    static
-    {
-        Random random = new Random();
-        bytes1 = new byte[1024];
-        bytes2 = new byte[128];
-        random.nextBytes(bytes1);
-        random.nextBytes(bytes2);
-    }
 
     @BeforeClass
     public static void defineSchema() throws ConfigurationException
@@ -143,30 +67,24 @@ public class ColumnFamilyStoreTest
                                     SimpleStrategy.class,
                                     KSMetaData.optsWithRF(1),
                                     SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD1),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD3),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD4),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD5),
-                                    SchemaLoader.indexCFMD(KEYSPACE1, CF_INDEX1, true),
-                                    SchemaLoader.indexCFMD(KEYSPACE1, CF_INDEX2, false),
-                                    SchemaLoader.superCFMD(KEYSPACE1, CF_SUPER1, LongType.instance),
-                                    SchemaLoader.superCFMD(KEYSPACE1, CF_SUPER6, LexicalUUIDType.instance, UTF8Type.instance),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARDINT, IntegerType.instance));
+                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2));
+                                    // TODO: Fix superCFMD failing on legacy table creation. Seems to be applying composite comparator to partition key
+                                    // SchemaLoader.superCFMD(KEYSPACE1, CF_SUPER1, LongType.instance));
+                                    // SchemaLoader.superCFMD(KEYSPACE1, CF_SUPER6, "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", LexicalUUIDType.instance, UTF8Type.instance),
         SchemaLoader.createKeyspace(KEYSPACE2,
                                     SimpleStrategy.class,
                                     KSMetaData.optsWithRF(1),
-                                    SchemaLoader.standardCFMD(KEYSPACE2, CF_STANDARD1),
-                                    SchemaLoader.indexCFMD(KEYSPACE2, CF_INDEX1, true),
-                                    SchemaLoader.compositeIndexCFMD(KEYSPACE2, CF_INDEX2, true),
-                                    SchemaLoader.compositeIndexCFMD(KEYSPACE2, CF_INDEX3, true).gcGraceSeconds(0));
-        SchemaLoader.createKeyspace(KEYSPACE3,
-                                    SimpleStrategy.class,
-                                    KSMetaData.optsWithRF(5),
-                                    SchemaLoader.indexCFMD(KEYSPACE3, CF_INDEX1, true));
-        SchemaLoader.createKeyspace(KEYSPACE4,
-                                    SimpleStrategy.class,
-                                    KSMetaData.optsWithRF(1),
-                                    SchemaLoader.perRowIndexedCFMD(KEYSPACE4, "Indexed1"));
+                                    SchemaLoader.standardCFMD(KEYSPACE2, CF_STANDARD1));
+    }
+
+    @Before
+    public void truncateCFS()
+    {
+        Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1).truncateBlocking();
+        Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD2).truncateBlocking();
+        // Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_SUPER1).truncateBlocking();
+
+        Keyspace.open(KEYSPACE2).getColumnFamilyStore(CF_STANDARD1).truncateBlocking();
     }
 
     @Test
@@ -175,21 +93,23 @@ public class ColumnFamilyStoreTest
     {
         Keyspace keyspace = Keyspace.open(KEYSPACE1);
         ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);
-        cfs.truncateBlocking();
 
-        Mutation rm;
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("key1"));
-        rm.add(CF_STANDARD1, cellname("Column1"), ByteBufferUtil.bytes("asdf"), 0);
-        rm.applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 0, "key1")
+                .clustering("Column1")
+                .add("val", "asdf")
+                .build()
+                .applyUnsafe();
         cfs.forceBlockingFlush();
 
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("key1"));
-        rm.add(CF_STANDARD1, cellname("Column1"), ByteBufferUtil.bytes("asdf"), 1);
-        rm.applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 1, "key1")
+                .clustering("Column1")
+                .add("val", "asdf")
+                .build()
+                .applyUnsafe();
         cfs.forceBlockingFlush();
 
         ((ClearableHistogram)cfs.metric.sstablesPerReadHistogram.cf).clear(); // resets counts
-        cfs.getColumnFamily(Util.namesQueryFilter(cfs, Util.dk("key1"), "Column1"));
+        Util.getAll(Util.cmd(cfs, "key1").includeRow("c1").build());
         assertEquals(1, cfs.metric.sstablesPerReadHistogram.cf.getCount());
     }
 
@@ -198,788 +118,128 @@ public class ColumnFamilyStoreTest
     {
         Keyspace keyspace = Keyspace.open(KEYSPACE1);
         ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD1);
-        cfs.truncateBlocking();
 
         List<Mutation> rms = new LinkedList<>();
-        Mutation rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("key1"));
-        rm.add(CF_STANDARD1, cellname("Column1"), ByteBufferUtil.bytes("asdf"), 0);
-        rm.add(CF_STANDARD1, cellname("Column2"), ByteBufferUtil.bytes("asdf"), 0);
-        rms.add(rm);
+        rms.add(new RowUpdateBuilder(cfs.metadata, 0, "key1")
+                .clustering("Column1")
+                .add("val", "asdf")
+                .build());
+
         Util.writeColumnFamily(rms);
 
         List<SSTableReader> ssTables = keyspace.getAllSSTables();
         assertEquals(1, ssTables.size());
         ssTables.get(0).forceFilterFailures();
-        ColumnFamily cf = cfs.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk("key2"), CF_STANDARD1, System.currentTimeMillis()));
-        assertNull(cf);
+        Util.assertEmpty(Util.cmd(cfs, "key2").build());
     }
 
     @Test
     public void testEmptyRow() throws Exception
     {
         Keyspace keyspace = Keyspace.open(KEYSPACE1);
-        final ColumnFamilyStore store = keyspace.getColumnFamilyStore(CF_STANDARD2);
-        Mutation rm;
+        final ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_STANDARD2);
 
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("key1"));
-        rm.delete(CF_STANDARD2, System.currentTimeMillis());
-        rm.applyUnsafe();
+        RowUpdateBuilder.deleteRow(cfs.metadata, FBUtilities.timestampMicros(), "key1", "Column1").applyUnsafe();
 
         Runnable r = new WrappedRunnable()
         {
             public void runMayThrow() throws IOException
             {
-                QueryFilter sliceFilter = QueryFilter.getSliceFilter(Util.dk("key1"), CF_STANDARD2, Composites.EMPTY, Composites.EMPTY, false, 1, System.currentTimeMillis());
-                ColumnFamily cf = store.getColumnFamily(sliceFilter);
-                assertTrue(cf.isMarkedForDelete());
-                assertFalse(cf.hasColumns());
-
-                QueryFilter namesFilter = Util.namesQueryFilter(store, Util.dk("key1"), "a");
-                cf = store.getColumnFamily(namesFilter);
-                assertTrue(cf.isMarkedForDelete());
-                assertFalse(cf.hasColumns());
+                Row toCheck = Util.getOnlyRowUnfiltered(Util.cmd(cfs, "key1").build());
+                Iterator<Cell> iter = toCheck.iterator();
+                assert(Iterators.size(iter) == 0);
             }
         };
 
-        KeyspaceTest.reTest(store, r);
-    }
-
-    @Test
-    public void testSkipStartKey()
-    {
-        ColumnFamilyStore cfs = insertKey1Key2();
-
-        IPartitioner p = StorageService.getPartitioner();
-        List<Row> result = cfs.getRangeSlice(Util.range(p, "key1", "key2"),
-                                             null,
-                                             Util.namesFilter(cfs, "asdf"),
-                                             10);
-        assertEquals(1, result.size());
-        assert result.get(0).key.getKey().equals(ByteBufferUtil.bytes("key2"));
-    }
-
-    @Test
-    public void testIndexScan()
-    {
-        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_INDEX1);
-        Mutation rm;
-        CellName nobirthdate = cellname("notbirthdate");
-        CellName birthdate = cellname("birthdate");
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k2"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k3"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k4aaaa"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(3L), 0);
-        rm.applyUnsafe();
-
-        // basic single-expression query
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = cfs.search(range, clause, filter, 100);
-
-        assert rows != null;
-        assert rows.size() == 2 : StringUtils.join(rows, ",");
-
-        String key = new String(rows.get(0).key.getKey().array(), rows.get(0).key.getKey().position(), rows.get(0).key.getKey().remaining());
-        assert "k1".equals( key ) : key;
-
-        key = new String(rows.get(1).key.getKey().array(), rows.get(1).key.getKey().position(), rows.get(1).key.getKey().remaining());
-        assert "k3".equals(key) : key;
-
-        assert ByteBufferUtil.bytes(1L).equals( rows.get(0).cf.getColumn(birthdate).value());
-        assert ByteBufferUtil.bytes(1L).equals( rows.get(1).cf.getColumn(birthdate).value());
-
-        // add a second expression
-        IndexExpression expr2 = new IndexExpression(ByteBufferUtil.bytes("notbirthdate"), Operator.GTE, ByteBufferUtil.bytes(2L));
-        clause = Arrays.asList(expr, expr2);
-        rows = cfs.search(range, clause, filter, 100);
-
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = new String(rows.get(0).key.getKey().array(), rows.get(0).key.getKey().position(), rows.get(0).key.getKey().remaining());
-        assert "k3".equals( key );
-
-        // same query again, but with resultset not including the subordinate expression
-        rows = cfs.search(range, clause, Util.namesFilter(cfs, "birthdate"), 100);
-
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = new String(rows.get(0).key.getKey().array(), rows.get(0).key.getKey().position(), rows.get(0).key.getKey().remaining());
-        assert "k3".equals( key );
-
-        assert rows.get(0).cf.getColumnCount() == 1 : rows.get(0).cf;
-
-        // once more, this time with a slice rowset that needs to be expanded
-        SliceQueryFilter emptyFilter = new SliceQueryFilter(Composites.EMPTY, Composites.EMPTY, false, 0);
-        rows = cfs.search(range, clause, emptyFilter, 100);
-
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = new String(rows.get(0).key.getKey().array(), rows.get(0).key.getKey().position(), rows.get(0).key.getKey().remaining());
-        assert "k3".equals( key );
-
-        assertFalse(rows.get(0).cf.hasColumns());
-
-        // query with index hit but rejected by secondary clause, with a small enough count that just checking count
-        // doesn't tell the scan loop that it's done
-        IndexExpression expr3 = new IndexExpression(ByteBufferUtil.bytes("notbirthdate"), Operator.EQ, ByteBufferUtil.bytes(-1L));
-        clause = Arrays.asList(expr, expr3);
-        rows = cfs.search(range, clause, filter, 100);
-
-        assert rows.isEmpty();
-    }
-
-    @Test
-    public void testLargeScan()
-    {
-        Mutation rm;
-        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_INDEX1);
-        for (int i = 0; i < 100; i++)
-        {
-            rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("key" + i));
-            rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(34L), 0);
-            rm.add(CF_INDEX1, cellname("notbirthdate"), ByteBufferUtil.bytes((long) (i % 2)), 0);
-            rm.applyUnsafe();
-        }
-
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(34L));
-        IndexExpression expr2 = new IndexExpression(ByteBufferUtil.bytes("notbirthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr, expr2);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = cfs.search(range, clause, filter, 100);
-
-        assert rows != null;
-        assert rows.size() == 50 : rows.size();
-        Set<DecoratedKey> keys = new HashSet<DecoratedKey>();
-        // extra check that there are no duplicate results -- see https://issues.apache.org/jira/browse/CASSANDRA-2406
-        for (Row row : rows)
-            keys.add(row.key);
-        assert rows.size() == keys.size();
-    }
-
-    @Test
-    public void testIndexDeletions() throws IOException
-    {
-        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE3).getColumnFamilyStore(CF_INDEX1);
-        Mutation rm;
-
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        String key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-        // delete the column directly
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.delete(CF_INDEX1, cellname("birthdate"), 1);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.isEmpty();
-
-        // verify that it's not being indexed under the deletion column value either
-        Cell deletion = rm.getColumnFamilies().iterator().next().iterator().next();
-        ByteBuffer deletionLong = ByteBufferUtil.bytes((long) ByteBufferUtil.toInt(deletion.value()));
-        IndexExpression expr0 = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, deletionLong);
-        List<IndexExpression> clause0 = Arrays.asList(expr0);
-        rows = cfs.search(range, clause0, filter, 100);
-        assert rows.isEmpty();
-
-        // resurrect w/ a newer timestamp
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 2);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-        // verify that row and delete w/ older timestamp does nothing
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.delete(CF_INDEX1, 1);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-        // similarly, column delete w/ older timestamp should do nothing
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.delete(CF_INDEX1, cellname("birthdate"), 1);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-        // delete the entire row (w/ newer timestamp this time)
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.delete(CF_INDEX1, 3);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.isEmpty() : StringUtils.join(rows, ",");
-
-        // make sure obsolete mutations don't generate an index entry
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 3);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.isEmpty() : StringUtils.join(rows, ",");
-
-        // try insert followed by row delete in the same mutation
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 1);
-        rm.delete(CF_INDEX1, 2);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.isEmpty() : StringUtils.join(rows, ",");
-
-        // try row delete followed by insert in the same mutation
-        rm = new Mutation(KEYSPACE3, ByteBufferUtil.bytes("k1"));
-        rm.delete(CF_INDEX1, 3);
-        rm.add(CF_INDEX1, cellname("birthdate"), ByteBufferUtil.bytes(1L), 4);
-        rm.applyUnsafe();
-        rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-    }
-
-    @Test
-    public void testIndexUpdate() throws IOException
-    {
-        Keyspace keyspace = Keyspace.open(KEYSPACE2);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_INDEX1);
-        CellName birthdate = cellname("birthdate");
-
-        // create a row and update the birthdate value, test that the index query fetches the new version
-        Mutation rm;
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 1);
-        rm.applyUnsafe();
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(2L), 2);
-        rm.applyUnsafe();
-
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = cfs.search(range, clause, filter, 100);
-        assert rows.size() == 0;
-
-        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(2L));
-        clause = Arrays.asList(expr);
-        rows = keyspace.getColumnFamilyStore(CF_INDEX1).search(range, clause, filter, 100);
-        String key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-        // update the birthdate value with an OLDER timestamp, and test that the index ignores this
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(3L), 0);
-        rm.applyUnsafe();
-
-        rows = keyspace.getColumnFamilyStore(CF_INDEX1).search(range, clause, filter, 100);
-        key = ByteBufferUtil.string(rows.get(0).key.getKey());
-        assert "k1".equals( key );
-
-    }
-
-    @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
-        Mutation rm;
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("k100"));
-        rm.add("Indexed1", cellname("birthdate"), ByteBufferUtil.bytes(100L), 1, 1000);
-        rm.applyUnsafe();
-
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.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 Mutation(KEYSPACE2, ByteBufferUtil.bytes("k100"));
-        rm.add("Indexed1", cellname("birthdate"), ByteBufferUtil.bytes(100L), 1, 1000);
-        rm.applyUnsafe();
-
-        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 Mutation(KEYSPACE2, ByteBufferUtil.bytes("k101"));
-        rm.add("Indexed1", cellname("birthdate"), ByteBufferUtil.bytes(101L), 1, 1000);
-        rm.applyUnsafe();
-
-        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.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 Mutation(KEYSPACE2, ByteBufferUtil.bytes("k101"));
-        rm.add("Indexed1", cellname("birthdate"), ByteBufferUtil.bytes(102L), 1, 1000);
-        rm.applyUnsafe();
-        // 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"), Operator.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
-    {
-        String keySpace = KEYSPACE2;
-        String cfName = CF_INDEX1;
-
-        Keyspace keyspace = Keyspace.open(keySpace);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.truncateBlocking();
-
-        ByteBuffer rowKey = ByteBufferUtil.bytes("k1");
-        CellName colName = cellname("birthdate"); 
-        ByteBuffer val1 = ByteBufferUtil.bytes(1L);
-        ByteBuffer val2 = ByteBufferUtil.bytes(2L);
-
-        // create a row and update the "birthdate" value, test that the index query fetches this version
-        Mutation rm;
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, colName, val1, 0);
-        rm.applyUnsafe();
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, val1);
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(1, rows.size());
-
-        // force a flush, so our index isn't being read from a memtable
-        keyspace.getColumnFamilyStore(cfName).forceBlockingFlush();
-
-        // now apply another update, but force the index update to be skipped
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, colName, val2, 1);
-        keyspace.apply(rm, true, false);
-
-        // Now searching the index for either the old or new value should return 0 rows
-        // because the new value was not indexed and the old value should be ignored
-        // (and in fact purged from the index cf).
-        // first check for the old value
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-        // now check for the updated value
-        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, val2);
-        clause = Arrays.asList(expr);
-        filter = new IdentityQueryFilter();
-        range = Util.range("", "");
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-
-        // now, reset back to the original value, still skipping the index update, to
-        // make sure the value was expunged from the index when it was discovered to be inconsistent
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, colName, ByteBufferUtil.bytes(1L), 3);
-        keyspace.apply(rm, true, false);
-
-        expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        clause = Arrays.asList(expr);
-        filter = new IdentityQueryFilter();
-        range = Util.range("", "");
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-    }
-
-    @Test
-    public void testDeleteOfInconsistentValuesFromCompositeIndex() throws Exception
-    {
-        String keySpace = KEYSPACE2;
-        String cfName = CF_INDEX2;
-
-        Keyspace keyspace = Keyspace.open(keySpace);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.truncateBlocking();
-
-        ByteBuffer rowKey = ByteBufferUtil.bytes("k1");
-        ByteBuffer clusterKey = ByteBufferUtil.bytes("ck1");
-        ByteBuffer colName = ByteBufferUtil.bytes("col1"); 
-
-        CellNameType baseComparator = cfs.getComparator();
-        CellName compositeName = baseComparator.makeCellName(clusterKey, colName);
-
-        ByteBuffer val1 = ByteBufferUtil.bytes("v1");
-        ByteBuffer val2 = ByteBufferUtil.bytes("v2");
-
-        // create a row and update the author value
-        Mutation rm;
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, compositeName, val1, 0);
-        rm.applyUnsafe();
-
-        // test that the index query fetches this version
-        IndexExpression expr = new IndexExpression(colName, Operator.EQ, val1);
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(1, rows.size());
-
-        // force a flush and retry the query, so our index isn't being read from a memtable
-        keyspace.getColumnFamilyStore(cfName).forceBlockingFlush();
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(1, rows.size());
-
-        // now apply another update, but force the index update to be skipped
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, compositeName, val2, 1);
-        keyspace.apply(rm, true, false);
-
-        // Now searching the index for either the old or new value should return 0 rows
-        // because the new value was not indexed and the old value should be ignored
-        // (and in fact purged from the index cf).
-        // first check for the old value
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-        // now check for the updated value
-        expr = new IndexExpression(colName, Operator.EQ, val2);
-        clause = Arrays.asList(expr);
-        filter = new IdentityQueryFilter();
-        range = Util.range("", "");
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-
-        // now, reset back to the original value, still skipping the index update, to
-        // make sure the value was expunged from the index when it was discovered to be inconsistent
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, compositeName, val1, 2);
-        keyspace.apply(rm, true, false);
-
-        expr = new IndexExpression(colName, Operator.EQ, val1);
-        clause = Arrays.asList(expr);
-        filter = new IdentityQueryFilter();
-        range = Util.range("", "");
-        rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-    }
-
-    // See CASSANDRA-6098
-    @Test
-    public void testDeleteCompositeIndex() throws Exception
-    {
-        String keySpace = KEYSPACE2;
-        String cfName = CF_INDEX3; // has gcGrace 0
-
-        Keyspace keyspace = Keyspace.open(keySpace);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.truncateBlocking();
-
-        ByteBuffer rowKey = ByteBufferUtil.bytes("k1");
-        ByteBuffer clusterKey = ByteBufferUtil.bytes("ck1");
-        ByteBuffer colName = ByteBufferUtil.bytes("col1");
-
-        CellNameType baseComparator = cfs.getComparator();
-        CellName compositeName = baseComparator.makeCellName(clusterKey, colName);
-
-        ByteBuffer val1 = ByteBufferUtil.bytes("v2");
-
-        // Insert indexed value.
-        Mutation rm;
-        rm = new Mutation(keySpace, rowKey);
-        rm.add(cfName, compositeName, val1, 0);
-        rm.applyUnsafe();
-
-        // Now delete the value and flush too.
-        rm = new Mutation(keySpace, rowKey);
-        rm.delete(cfName, 1);
-        rm.applyUnsafe();
-
-        // We want the data to be gcable, but even if gcGrace == 0, we still need to wait 1 second
-        // since we won't gc on a tie.
-        try { Thread.sleep(1000); } catch (Exception e) {}
-
-        // Read the index and we check we do get no value (and no NPE)
-        // Note: the index will return the entry because it hasn't been deleted (we
-        // haven't read yet nor compacted) but the data read itself will return null
-        IndexExpression expr = new IndexExpression(colName, Operator.EQ, val1);
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = keyspace.getColumnFamilyStore(cfName).search(range, clause, filter, 100);
-        assertEquals(0, rows.size());
-    }
-
-    // See CASSANDRA-2628
-    @Test
-    public void testIndexScanWithLimitOne()
-    {
-        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_INDEX1);
-        Mutation rm;
-
-        CellName nobirthdate = cellname("notbirthdate");
-        CellName birthdate = cellname("birthdate");
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("kk1"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("kk2"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("kk3"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("kk4"));
-        rm.add(CF_INDEX1, nobirthdate, ByteBufferUtil.bytes(2L), 0);
-        rm.add(CF_INDEX1, birthdate, ByteBufferUtil.bytes(1L), 0);
-        rm.applyUnsafe();
-
-        // basic single-expression query
-        IndexExpression expr1 = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        IndexExpression expr2 = new IndexExpression(ByteBufferUtil.bytes("notbirthdate"), Operator.GT, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr1, expr2);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        Range<RowPosition> range = Util.range("", "");
-        List<Row> rows = cfs.search(range, clause, filter, 1);
-
-        assert rows != null;
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-    }
-
-    @Test
-    public void testIndexCreate() throws IOException, InterruptedException, ExecutionException
-    {
-        Keyspace keyspace = Keyspace.open(KEYSPACE1);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_INDEX2);
-
-        // create a row and update the birthdate value, test that the index query fetches the new version
-        Mutation rm;
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k1"));
-        rm.add(CF_INDEX2, cellname("birthdate"), ByteBufferUtil.bytes(1L), 1);
-        rm.applyUnsafe();
-
-        ColumnDefinition old = cfs.metadata.getColumnDefinition(ByteBufferUtil.bytes("birthdate"));
-        ColumnDefinition cd = ColumnDefinition.regularDef(cfs.metadata, old.name.bytes, old.type, null).setIndex("birthdate_index", IndexType.KEYS, null);
-        Future<?> future = cfs.indexManager.addIndexedColumn(cd);
-        future.get();
-        // we had a bug (CASSANDRA-2244) where index would get created but not flushed -- check for that
-        assert cfs.indexManager.getIndexForColumn(cd.name.bytes).getIndexCfs().getSSTables().size() > 0;
-
-        queryBirthdate(keyspace);
-
-        // validate that drop clears it out & rebuild works (CASSANDRA-2320)
-        SecondaryIndex indexedCfs = cfs.indexManager.getIndexForColumn(ByteBufferUtil.bytes("birthdate"));
-        cfs.indexManager.removeIndexedColumn(ByteBufferUtil.bytes("birthdate"));
-        assert !indexedCfs.isIndexBuilt(ByteBufferUtil.bytes("birthdate"));
-
-        // rebuild & re-query
-        future = cfs.indexManager.addIndexedColumn(cd);
-        future.get();
-        queryBirthdate(keyspace);
-    }
-
-    private void queryBirthdate(Keyspace keyspace) throws CharacterCodingException
-    {
-        IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes("birthdate"), Operator.EQ, ByteBufferUtil.bytes(1L));
-        List<IndexExpression> clause = Arrays.asList(expr);
-        IDiskAtomFilter filter = new IdentityQueryFilter();
-        List<Row> rows = keyspace.getColumnFamilyStore(CF_INDEX2).search(Util.range("", ""), clause, filter, 100);
-        assert rows.size() == 1 : StringUtils.join(rows, ",");
-        assertEquals("k1", ByteBufferUtil.string(rows.get(0).key.getKey()));
-    }
-
-    @Test
-    public void testCassandra6778() throws CharacterCodingException
-    {
-        String cfname = CF_STANDARDINT;
-        Keyspace keyspace = Keyspace.open(KEYSPACE1);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfname);
-
-        // insert two columns that represent the same integer but have different binary forms (the
-        // second one is padded with extra zeros)
-        Mutation rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k1"));
-        CellName column1 = cellname(ByteBuffer.wrap(new byte[]{1}));
-        rm.add(cfname, column1, ByteBufferUtil.bytes("data1"), 1);
-        rm.applyUnsafe();
-        cfs.forceBlockingFlush();
-
-        rm = new Mutation(KEYSPACE1, ByteBufferUtil.bytes("k1"));
-        CellName column2 = cellname(ByteBuffer.wrap(new byte[]{0, 0, 1}));
-        rm.add(cfname, column2, ByteBufferUtil.bytes("data2"), 2);
-        rm.applyUnsafe();
-        cfs.forceBlockingFlush();
-
-        // fetch by the first column name; we should get the second version of the column value
-        SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(
-            KEYSPACE1, ByteBufferUtil.bytes("k1"), cfname, System.currentTimeMillis(),
-            new NamesQueryFilter(FBUtilities.singleton(column1, cfs.getComparator())));
-
-        ColumnFamily cf = cmd.getRow(keyspace).cf;
-        assertEquals(1, cf.getColumnCount());
-        Cell cell = cf.getColumn(column1);
-        assertEquals("data2", ByteBufferUtil.string(cell.value()));
-        assertEquals(column2, cell.name());
-
-        // fetch by the second column name; we should get the second version of the column value
-        cmd = new SliceByNamesReadCommand(
-            KEYSPACE1, ByteBufferUtil.bytes("k1"), cfname, System.currentTimeMillis(),
-            new NamesQueryFilter(FBUtilities.singleton(column2, cfs.getComparator())));
-
-        cf = cmd.getRow(keyspace).cf;
-        assertEquals(1, cf.getColumnCount());
-        cell = cf.getColumn(column2);
-        assertEquals("data2", ByteBufferUtil.string(cell.value()));
-        assertEquals(column2, cell.name());
-    }
-
-    @Test
-    public void testInclusiveBounds()
-    {
-        ColumnFamilyStore cfs = insertKey1Key2();
-
-        List<Row> result = cfs.getRangeSlice(Util.bounds("key1", "key2"),
-                                             null,
-                                             Util.namesFilter(cfs, "asdf"),
-                                             10);
-        assertEquals(2, result.size());
-        assert result.get(0).key.getKey().equals(ByteBufferUtil.bytes("key1"));
-    }
-
-    @Test
-    public void testDeleteSuperRowSticksAfterFlush() throws Throwable
-    {
-        String keyspaceName = KEYSPACE1;
-        String cfName= CF_SUPER1;
-        ByteBuffer scfName = ByteBufferUtil.bytes("SuperDuper");
-        Keyspace keyspace = Keyspace.open(keyspaceName);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        DecoratedKey key = Util.dk("flush-resurrection");
-
-        // create an isolated sstable.
-        putColsSuper(cfs, key, scfName,
-                new BufferCell(cellname(1L), ByteBufferUtil.bytes("val1"), 1),
-                new BufferCell(cellname(2L), ByteBufferUtil.bytes("val2"), 1),
-                new BufferCell(cellname(3L), ByteBufferUtil.bytes("val3"), 1));
-        cfs.forceBlockingFlush();
-
-        // insert, don't flush.
-        putColsSuper(cfs, key, scfName,
-                new BufferCell(cellname(4L), ByteBufferUtil.bytes("val4"), 1),
-                new BufferCell(cellname(5L), ByteBufferUtil.bytes("val5"), 1),
-                new BufferCell(cellname(6L), ByteBufferUtil.bytes("val6"), 1));
-
-        // verify insert.
-        final SlicePredicate sp = new SlicePredicate();
-        sp.setSlice_range(new SliceRange());
-        sp.getSlice_range().setCount(100);
-        sp.getSlice_range().setStart(ArrayUtils.EMPTY_BYTE_ARRAY);
-        sp.getSlice_range().setFinish(ArrayUtils.EMPTY_BYTE_ARRAY);
-
-        assertRowAndColCount(1, 6, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
-
-        // delete
-        Mutation rm = new Mutation(keyspace.getName(), key.getKey());
-        rm.deleteRange(cfName, SuperColumns.startOf(scfName), SuperColumns.endOf(scfName), 2);
-        rm.applyUnsafe();
-
-        // verify delete.
-        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
-
-        // flush
-        cfs.forceBlockingFlush();
-
-        // re-verify delete.
-        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
-
-        // late insert.
-        putColsSuper(cfs, key, scfName,
-                new BufferCell(cellname(4L), ByteBufferUtil.bytes("val4"), 1L),
-                new BufferCell(cellname(7L), ByteBufferUtil.bytes("val7"), 1L));
-
-        // re-verify delete.
-        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
-
-        // make sure new writes are recognized.
-        putColsSuper(cfs, key, scfName,
-                new BufferCell(cellname(3L), ByteBufferUtil.bytes("val3"), 3),
-                new BufferCell(cellname(8L), ByteBufferUtil.bytes("val8"), 3),
-                new BufferCell(cellname(9L), ByteBufferUtil.bytes("val9"), 3));
-        assertRowAndColCount(1, 3, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
-    }
-
-    private static void assertRowAndColCount(int rowCount, int colCount, boolean isDeleted, Collection<Row> rows) throws CharacterCodingException
-    {
-        assert rows.size() == rowCount : "rowcount " + rows.size();
-        for (Row row : rows)
-        {
-            assert row.cf != null : "cf was null";
-            assert row.cf.getColumnCount() == colCount : "colcount " + row.cf.getColumnCount() + "|" + str(row.cf);
-            if (isDeleted)
-                assert row.cf.isMarkedForDelete() : "cf not marked for delete";
-        }
-    }
-
-    private static String str(ColumnFamily cf) throws CharacterCodingException
-    {
-        StringBuilder sb = new StringBuilder();
-        for (Cell col : cf.getSortedColumns())
-            sb.append(String.format("(%s,%s,%d),", ByteBufferUtil.string(col.name().toByteBuffer()), ByteBufferUtil.string(col.value()), col.timestamp()));
-        return sb.toString();
-    }
-
-    private static void putColsSuper(ColumnFamilyStore cfs, DecoratedKey key, ByteBuffer scfName, Cell... cols) throws Throwable
-    {
-        ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfs.keyspace.getName(), cfs.name);
-        for (Cell col : cols)
-            cf.addColumn(col.withUpdatedName(CellNames.compositeDense(scfName, col.name().toByteBuffer())));
-        Mutation rm = new Mutation(cfs.keyspace.getName(), key.getKey(), cf);
-        rm.applyUnsafe();
-    }
-
-    private static void putColsStandard(ColumnFamilyStore cfs, DecoratedKey key, Cell... cols) throws Throwable
-    {
-        ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfs.keyspace.getName(), cfs.name);
-        for (Cell col : cols)
-            cf.addColumn(col);
-        Mutation rm = new Mutation(cfs.keyspace.getName(), key.getKey(), cf);
-        rm.applyUnsafe();
-    }
+        reTest(cfs, r);
+    }
+
+    // TODO: Implement this once we have hooks to super columns available in CQL context
+//    @Test
+//    public void testDeleteSuperRowSticksAfterFlush() throws Throwable
+//    {
+//        String keyspaceName = KEYSPACE1;
+//        String cfName= CF_SUPER1;
+//
+//        Keyspace keyspace = Keyspace.open(keyspaceName);
+//        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
+//
+//        ByteBuffer scfName = ByteBufferUtil.bytes("SuperDuper");
+//        DecoratedKey key = Util.dk("flush-resurrection");
+//
+//        // create an isolated sstable.
+//        putColSuper(cfs, key, 0, ByteBufferUtil.bytes("val"), ByteBufferUtil.bytes(1L), ByteBufferUtil.bytes(1L), ByteBufferUtil.bytes("val1"));
+
+//        putColsSuper(cfs, key, scfName,
+//                new BufferCell(cellname(1L), ByteBufferUtil.bytes("val1"), 1),
+//                new BufferCell(cellname(2L), ByteBufferUtil.bytes("val2"), 1),
+//                new BufferCell(cellname(3L), ByteBufferUtil.bytes("val3"), 1));
+//        cfs.forceBlockingFlush();
+//
+//        // insert, don't flush.
+//        putColsSuper(cfs, key, scfName,
+//                new BufferCell(cellname(4L), ByteBufferUtil.bytes("val4"), 1),
+//                new BufferCell(cellname(5L), ByteBufferUtil.bytes("val5"), 1),
+//                new BufferCell(cellname(6L), ByteBufferUtil.bytes("val6"), 1));
+//
+//        // verify insert.
+//        final SlicePredicate sp = new SlicePredicate();
+//        sp.setSlice_range(new SliceRange());
+//        sp.getSlice_range().setCount(100);
+//        sp.getSlice_range().setStart(ArrayUtils.EMPTY_BYTE_ARRAY);
+//        sp.getSlice_range().setFinish(ArrayUtils.EMPTY_BYTE_ARRAY);
+//
+//        assertRowAndColCount(1, 6, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
+//
+//        // delete
+//        Mutation rm = new Mutation(keyspace.getName(), key.getKey());
+//        rm.deleteRange(cfName, SuperColumns.startOf(scfName), SuperColumns.endOf(scfName), 2);
+//        rm.applyUnsafe();
+//
+//        // verify delete.
+//        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
+//
+//        // flush
+//        cfs.forceBlockingFlush();
+//
+//        // re-verify delete.
+//        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
+//
+//        // late insert.
+//        putColsSuper(cfs, key, scfName,
+//                new BufferCell(cellname(4L), ByteBufferUtil.bytes("val4"), 1L),
+//                new BufferCell(cellname(7L), ByteBufferUtil.bytes("val7"), 1L));
+//
+//        // re-verify delete.
+//        assertRowAndColCount(1, 0, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
+//
+//        // make sure new writes are recognized.
+//        putColsSuper(cfs, key, scfName,
+//                new BufferCell(cellname(3L), ByteBufferUtil.bytes("val3"), 3),
+//                new BufferCell(cellname(8L), ByteBufferUtil.bytes("val8"), 3),
+//                new BufferCell(cellname(9L), ByteBufferUtil.bytes("val9"), 3));
+//        assertRowAndColCount(1, 3, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, scfName), 100));
+//    }
+
+//    private static void assertRowAndColCount(int rowCount, int colCount, boolean isDeleted, Collection<Row> rows) throws CharacterCodingException
+//    {
+//        assert rows.size() == rowCount : "rowcount " + rows.size();
+//        for (Row row : rows)
+//        {
+//            assert row.cf != null : "cf was null";
+//            assert row.cf.getColumnCount() == colCount : "colcount " + row.cf.getColumnCount() + "|" + str(row.cf);
+//            if (isDeleted)
+//                assert row.cf.isMarkedForDelete() : "cf not marked for delete";
+//        }
+//    }
+//
+//    private static String str(ColumnFamily cf) throws CharacterCodingException
+//    {
+//        StringBuilder sb = new StringBuilder();
+//        for (Cell col : cf.getSortedColumns())
+//            sb.append(String.format("(%s,%s,%d),", ByteBufferUtil.string(col.name().toByteBuffer()), ByteBufferUtil.string(col.value()), col.timestamp()));
+//        return sb.toString();
+//    }
 
     @Test
     public void testDeleteStandardRowSticksAfterFlush() throws Throwable
@@ -989,1292 +249,339 @@ public class ColumnFamilyStoreTest
         String cfName = CF_STANDARD1;
         Keyspace keyspace = Keyspace.open(keyspaceName);
         ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        DecoratedKey key = Util.dk("f-flush-resurrection");
 
-        SlicePredicate sp = new SlicePredicate();
-        sp.setSlice_range(new SliceRange());
-        sp.getSlice_range().setCount(100);
-        sp.getSlice_range().setStart(ArrayUtils.EMPTY_BYTE_ARRAY);
-        sp.getSlice_range().setFinish(ArrayUtils.EMPTY_BYTE_ARRAY);
+        ByteBuffer col = ByteBufferUtil.bytes("val");
+        ByteBuffer val = ByteBufferUtil.bytes("val1");
 
         // insert
-        putColsStandard(cfs, key, column("col1", "val1", 1), column("col2", "val2", 1));
-        assertRowAndColCount(1, 2, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        ColumnDefinition newCol = new ColumnDefinition(cfs.metadata, ByteBufferUtil.bytes("val2"), AsciiType.instance, 1, ColumnDefinition.Kind.REGULAR);
+        new RowUpdateBuilder(cfs.metadata, 0, "key1").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 0, "key2").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        assertRangeCount(cfs, col, val, 2);
 
         // flush.
         cfs.forceBlockingFlush();
 
         // insert, don't flush
-        putColsStandard(cfs, key, column("col3", "val3", 1), column("col4", "val4", 1));
-        assertRowAndColCount(1, 4, false, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        new RowUpdateBuilder(cfs.metadata, 1, "key3").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 1, "key4").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        assertRangeCount(cfs, col, val, 4);
 
         // delete (from sstable and memtable)
-        Mutation rm = new Mutation(keyspace.getName(), key.getKey());
-        rm.delete(cfs.name, 2);
-        rm.applyUnsafe();
+        RowUpdateBuilder.deleteRow(cfs.metadata, 5, "key1", "Column1").applyUnsafe();
+        RowUpdateBuilder.deleteRow(cfs.metadata, 5, "key3", "Column1").applyUnsafe();
 
         // verify delete
-        assertRowAndColCount(1, 0, true, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        assertRangeCount(cfs, col, val, 2);
 
         // flush
         cfs.forceBlockingFlush();
 
         // re-verify delete. // first breakage is right here because of CASSANDRA-1837.
-        assertRowAndColCount(1, 0, true, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        assertRangeCount(cfs, col, val, 2);
 
         // simulate a 'late' insertion that gets put in after the deletion. should get inserted, but fail on read.
-        putColsStandard(cfs, key, column("col5", "val5", 1), column("col2", "val2", 1));
+        new RowUpdateBuilder(cfs.metadata, 2, "key1").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 2, "key3").clustering("Column1").add("val", "val1").build().applyUnsafe();
 
         // should still be nothing there because we deleted this row. 2nd breakage, but was undetected because of 1837.
-        assertRowAndColCount(1, 0, true, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        assertRangeCount(cfs, col, val, 2);
 
         // make sure that new writes are recognized.
-        putColsStandard(cfs, key, column("col6", "val6", 3), column("col7", "val7", 3));
-        assertRowAndColCount(1, 2, true, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
+        new RowUpdateBuilder(cfs.metadata, 10, "key5").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        new RowUpdateBuilder(cfs.metadata, 10, "key6").clustering("Column1").add("val", "val1").build().applyUnsafe();
+        assertRangeCount(cfs, col, val, 4);
 
         // and it remains so after flush. (this wasn't failing before, but it's good to check.)
         cfs.forceBlockingFlush();
-        assertRowAndColCount(1, 2, true, cfs.getRangeSlice(Util.range("f", "g"), null, ThriftValidation.asIFilter(sp, cfs.metadata, null), 100));
-    }
-
-
-    private ColumnFamilyStore insertKey1Key2()
-    {
-        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE2).getColumnFamilyStore(CF_STANDARD1);
-        List<Mutation> rms = new LinkedList<>();
-        Mutation rm;
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("key1"));
-        rm.add(CF_STANDARD1, cellname("Column1"), ByteBufferUtil.bytes("asdf"), 0);
-        rms.add(rm);
-        Util.writeColumnFamily(rms);
-
-        rm = new Mutation(KEYSPACE2, ByteBufferUtil.bytes("key2"));
-        rm.add(CF_STANDARD1, cellname("Column1"), ByteBufferUtil.bytes("asdf"), 0);
-        rms.add(rm);
-        return Util.writeColumnFamily(rms);
+        assertRangeCount(cfs, col, val, 4);
     }
 
     @Test
     public void testBackupAfterFlush() throws Throwable
     {
-        ColumnFamilyStore cfs = insertKey1Key2();
+        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE2).getColumnFamilyStore(CF_STANDARD1);
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key1")).clustering("Column1").add("val", "asdf").build().applyUnsafe();
+        cfs.forceBlockingFlush();
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key2")).clustering("Column1").add("val", "asdf").build().applyUnsafe();
+        cfs.forceBlockingFlush();
 
         for (int version = 1; version <= 2; ++version)
         {
             Descriptor existing = new Descriptor(cfs.directories.getDirectoryForNewSSTables(), KEYSPACE2, CF_STANDARD1, version, Descriptor.Type.FINAL);
             Descriptor desc = new Descriptor(Directories.getBackupsDirectory(existing), KEYSPACE2, CF_STANDARD1, version, Descriptor.Type.FINAL);
             for (Component c : new Component[]{ Component.DATA, Component.PRIMARY_INDEX, Component.FILTER, Component.STATS })
-                assertTrue("can not find backedup file:" + desc.filenameFor(c), new File(desc.filenameFor(c)).exists());
+                assertTrue("Cannot find backed-up file:" + desc.filenameFor(c), new File(desc.filenameFor(c)).exists());
         }
     }
 
-    // CASSANDRA-3467.  the key here is that supercolumn and subcolumn comparators are different
-    @Test
-    public void testSliceByNamesCommandOnUUIDTypeSCF() throws Throwable
-    {
-        String keyspaceName = KEYSPACE1;
-        String cfName = CF_SUPER6;
-        ByteBuffer superColName = LexicalUUIDType.instance.fromString("a4ed3562-0e8e-4b41-bdfd-c45a2774682d");
-        Keyspace keyspace = Keyspace.open(keyspaceName);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        DecoratedKey key = Util.dk("slice-get-uuid-type");
-
-        // Insert a row with one supercolumn and multiple subcolumns
-        putColsSuper(cfs, key, superColName, new BufferCell(cellname("a"), ByteBufferUtil.bytes("A"), 1),
-                                             new BufferCell(cellname("b"), ByteBufferUtil.bytes("B"), 1));
-
-        // Get the entire supercolumn like normal
-        ColumnFamily cfGet = cfs.getColumnFamily(QueryFilter.getIdentityFilter(key, cfName, System.currentTimeMillis()));
-        assertEquals(ByteBufferUtil.bytes("A"), cfGet.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a"))).value());
-        assertEquals(ByteBufferUtil.bytes("B"), cfGet.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b"))).value());
-
-        // Now do the SliceByNamesCommand on the supercolumn, passing both subcolumns in as columns to get
-        SortedSet<CellName> sliceColNames = new TreeSet<CellName>(cfs.metadata.comparator);
-        sliceColNames.add(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a")));
-        sliceColNames.add(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b")));
-        SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(keyspaceName, key.getKey(), cfName, System.currentTimeMillis(), new NamesQueryFilter(sliceColNames));
-        ColumnFamily cfSliced = cmd.getRow(keyspace).cf;
-
-        // Make sure the slice returns the same as the straight get
-        assertEquals(ByteBufferUtil.bytes("A"), cfSliced.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a"))).value());
-        assertEquals(ByteBufferUtil.bytes("B"), cfSliced.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b"))).value());
-    }
-
-    @Test
-    public void testSliceByNamesCommandOldMetadata() throws Throwable
-    {
-        String keyspaceName = KEYSPACE1;
-        String cfName= CF_STANDARD1;
-        DecoratedKey key = Util.dk("slice-name-old-metadata");
-        CellName cname = cellname("c1");
-        Keyspace keyspace = Keyspace.open(keyspaceName);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.truncateBlocking();
-
-        // Create a cell a 'high timestamp'
-        putColsStandard(cfs, key, new BufferCell(cname, ByteBufferUtil.bytes("a"), 2));
+    // TODO: Fix once we have working supercolumns in 8099
+//    // CASSANDRA-3467.  the key here is that supercolumn and subcolumn comparators are different
+//    @Test
+//    public void testSliceByNamesCommandOnUUIDTypeSCF() throws Throwable
+//    {
+//        String keyspaceName = KEYSPACE1;
+//        String cfName = CF_SUPER6;
+//        ByteBuffer superColName = LexicalUUIDType.instance.fromString("a4ed3562-0e8e-4b41-bdfd-c45a2774682d");
+//        Keyspace keyspace = Keyspace.open(keyspaceName);
+//        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
+//        DecoratedKey key = Util.dk("slice-get-uuid-type");
+//
+//        // Insert a row with one supercolumn and multiple subcolumns
+//        putColsSuper(cfs, key, superColName, new BufferCell(cellname("a"), ByteBufferUtil.bytes("A"), 1),
+//                                             new BufferCell(cellname("b"), ByteBufferUtil.bytes("B"), 1));
+//
+//        // Get the entire supercolumn like normal
+//        ColumnFamily cfGet = cfs.getColumnFamily(QueryFilter.getIdentityFilter(key, cfName, System.currentTimeMillis()));
+//        assertEquals(ByteBufferUtil.bytes("A"), cfGet.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a"))).value());
+//        assertEquals(ByteBufferUtil.bytes("B"), cfGet.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b"))).value());
+//
+//        // Now do the SliceByNamesCommand on the supercolumn, passing both subcolumns in as columns to get
+//        SortedSet<CellName> sliceColNames = new TreeSet<CellName>(cfs.metadata.comparator);
+//        sliceColNames.add(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a")));
+//        sliceColNames.add(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b")));
+//        SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(keyspaceName, key.getKey(), cfName, System.currentTimeMillis(), new NamesQueryFilter(sliceColNames));
+//        ColumnFamily cfSliced = cmd.getRow(keyspace).cf;
+//
+//        // Make sure the slice returns the same as the straight get
+//        assertEquals(ByteBufferUtil.bytes("A"), cfSliced.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("a"))).value());
+//        assertEquals(ByteBufferUtil.bytes("B"), cfSliced.getColumn(CellNames.compositeDense(superColName, ByteBufferUtil.bytes("b"))).value());
+//    }
+
+    // TODO: fix once SSTableSimpleWriter's back in
+//    @Test
+//    public void testRemoveUnfinishedCompactionLeftovers() throws Throwable
+//    {
+//        String ks = KEYSPACE1;
+//        String cf = CF_STANDARD3; // should be empty
+//
+//        final CFMetaData cfmeta = Schema.instance.getCFMetaData(ks, cf);
+//        Directories dir = new Directories(cfmeta);
+//        ByteBuffer key = bytes("key");
+//
+//        // 1st sstable
+//        SSTableSimpleWriter writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(), cfmeta, StorageService.getPartitioner());
+//        writer.newRow(key);
+//        writer.addColumn(bytes("col"), bytes("val"), 1);
+//        writer.close();
+//
+//        Map<Descriptor, Set<Component>> sstables = dir.sstableLister().list();
+//        assertEquals(1, sstables.size());
+//
+//        Map.Entry<Descriptor, Set<Component>> sstableToOpen = sstables.entrySet().iterator().next();
+//        final SSTableReader sstable1 = SSTableReader.open(sstableToOpen.getKey());
+//
+//        // simulate incomplete compaction
+//        writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(),
+//                                         cfmeta, StorageService.getPartitioner())
+//        {
+//            protected SSTableWriter getWriter()
+//            {
+//                MetadataCollector collector = new MetadataCollector(cfmeta.comparator);
+//                collector.addAncestor(sstable1.descriptor.generation); // add ancestor from previously written sstable
+//                return SSTableWriter.create(createDescriptor(directory, metadata.ksName, metadata.cfName, DatabaseDescriptor.getSSTableFormat()),
+//                        0L,
+//                        ActiveRepairService.UNREPAIRED_SSTABLE,
+//                        metadata,
+//                        DatabaseDescriptor.getPartitioner(),
+//                        collector);
+//            }
+//        };
+//        writer.newRow(key);
+//        writer.addColumn(bytes("col"), bytes("val"), 1);
+//        writer.close();
+//
+//        // should have 2 sstables now
+//        sstables = dir.sstableLister().list();
+//        assertEquals(2, sstables.size());
+//
+//        SSTableReader sstable2 = SSTableReader.open(sstable1.descriptor);
+//        UUID compactionTaskID = SystemKeyspace.startCompaction(
+//                Keyspace.open(ks).getColumnFamilyStore(cf),
+//                Collections.singleton(sstable2));
+//
+//        Map<Integer, UUID> unfinishedCompaction = new HashMap<>();
+//        unfinishedCompaction.put(sstable1.descriptor.generation, compactionTaskID);
+//        ColumnFamilyStore.removeUnfinishedCompactionLeftovers(cfmeta, unfinishedCompaction);
+//
+//        // 2nd sstable should be removed (only 1st sstable exists in set of size 1)
+//        sstables = dir.sstableLister().list();
+//        assertEquals(1, sstables.size());
+//        assertTrue(sstables.containsKey(sstable1.descriptor));
+//
+//        Map<Pair<String, String>, Map<Integer, UUID>> unfinished = SystemKeyspace.getUnfinishedCompactions();
+//        assertTrue(unfinished.isEmpty());
+//        sstable1.selfRef().release();
+//        sstable2.selfRef().release();
+//    }
+
+    // TODO: Fix once SSTableSimpleWriter's back in
+    // @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-6086">CASSANDRA-6086</a>
+//    @Test
+//    public void testFailedToRemoveUnfinishedCompactionLeftovers() throws Throwable
+//    {
+//        final String ks = KEYSPACE1;
+//        final String cf = CF_STANDARD4; // should be empty
+//
+//        final CFMetaData cfmeta = Schema.instance.getCFMetaData(ks, cf);
+//        Directories dir = new Directories(cfmeta);
+//        ByteBuffer key = bytes("key");
+//
+//        // Write SSTable generation 3 that has ancestors 1 and 2
+//        final Set<Integer> ancestors = Sets.newHashSet(1, 2);
+//        SSTableSimpleWriter writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(),
+//                                                cfmeta, StorageService.getPartitioner())
+//        {
+//            protected SSTableWriter getWriter()
+//            {
+//                MetadataCollector collector = new MetadataCollector(cfmeta.comparator);
+//                for (int ancestor : ancestors)
+//                    collector.addAncestor(ancestor);
+//                String file = new Descriptor(directory, ks, cf, 3, Descriptor.Type.TEMP).filenameFor(Component.DATA);
+//                return SSTableWriter.create(Descriptor.fromFilename(file),
+//                        0L,
+//                        ActiveRepairService.UNREPAIRED_SSTABLE,
+//                        metadata,
+//                        StorageService.getPartitioner(),
+//                        collector);
+//            }
+//        };
+//        writer.newRow(key);
+//        writer.addColumn(bytes("col"), bytes("val"), 1);
+//        writer.close();
+//
+//        Map<Descriptor, Set<Component>> sstables = dir.sstableLister().list();
+//        assert sstables.size() == 1;
+//
+//        Map.Entry<Descriptor, Set<Component>> sstableToOpen = sstables.entrySet().iterator().next();
+//        final SSTableReader sstable1 = SSTableReader.open(sstableToOpen.getKey());
+//
+//        // simulate we don't have generation in compaction_history
+//        Map<Integer, UUID> unfinishedCompactions = new HashMap<>();
+//        UUID compactionTaskID = UUID.randomUUID();
+//        for (Integer ancestor : ancestors)
+//            unfinishedCompactions.put(ancestor, compactionTaskID);
+//        ColumnFamilyStore.removeUnfinishedCompactionLeftovers(cfmeta, unfinishedCompactions);
+//
+//        // SSTable should not be deleted
+//        sstables = dir.sstableLister().list();
+//        assert sstables.size() == 1;
+//        assert sstables.containsKey(sstable1.descriptor);
+//    }
+
+    // TODO: Fix once SSTableSimpleWriter's back in
+//    @Test
+//    public void testLoadNewSSTablesAvoidsOverwrites() throws Throwable
+//    {
+//        String ks = KEYSPACE1;
+//        String cf = CF_STANDARD1;
+//        ColumnFamilyStore cfs = Keyspace.open(ks).getColumnFamilyStore(cf);
+//        SSTableDeletingTask.waitForDeletions();
+//
+//        final CFMetaData cfmeta = Schema.instance.getCFMetaData(ks, cf);
+//        Directories dir = new Directories(cfs.metadata);
+//
+//        // clear old SSTables (probably left by CFS.clearUnsafe() calls in other tests)
+//        for (Map.Entry<Descriptor, Set<Component>> entry : dir.sstableLister().list().entrySet())
+//        {
+//            for (Component component : entry.getValue())
+//            {
+//                FileUtils.delete(entry.getKey().filenameFor(component));
+//            }
+//        }
+//
+//        // sanity check
+//        int existingSSTables = dir.sstableLister().list().keySet().size();
+//        assert existingSSTables == 0 : String.format("%d SSTables unexpectedly exist", existingSSTables);
+//
+//        ByteBuffer key = bytes("key");
+//
+//        SSTableSimpleWriter writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(),
+//                                                             cfmeta, StorageService.getPartitioner())
+//        {
+//            @Override
+//            protected SSTableWriter getWriter()
+//            {
+//                // hack for reset generation
+//                generation.set(0);
+//                return super.getWriter();
+//            }
+//        };
+//        writer.newRow(key);
+//        writer.addColumn(bytes("col"), bytes("val"), 1);
+//        writer.close();
+//
+//        writer = new SSTableSimpleWriter(dir.getDirectoryForNewSSTables(),
+//                                         cfmeta, StorageService.getPartitioner());
+//        writer.newRow(key);
+//        writer.addColumn(bytes("col"), bytes("val"), 1);
+//        writer.close();
+//
+//        Set<Integer> generations = new HashSet<>();
+//        for (Descriptor descriptor : dir.sstableLister().list().keySet())
+//            generations.add(descriptor.generation);
+//
+//        // we should have two generations: [1, 2]
+//        assertEquals(2, generations.size());
+//        assertTrue(generations.contains(1));
+//        assertTrue(generations.contains(2));
+//
+//        assertEquals(0, cfs.getSSTables().size());
+//
+//        // start the generation counter at 1 again (other tests have incremented it already)
+//        cfs.resetFileIndexGenerator();
+//
+//        boolean incrementalBackupsEnabled = DatabaseDescriptor.isIncrementalBackupsEnabled();
+//        try
+//        {
+//            // avoid duplicate hardlinks to incremental backups
+//            DatabaseDescriptor.setIncrementalBackupsEnabled(false);
+//            cfs.loadNewSSTables();
+//        }
+//        finally
+//        {
+//            DatabaseDescriptor.setIncrementalBackupsEnabled(incrementalBackupsEnabled);
+//        }
+//
+//        assertEquals(2, cfs.getSSTables().size());
+//        generations = new HashSet<>();
+//        for (Descriptor descriptor : dir.sstableLister().list().keySet())
+//            generations.add(descriptor.generation);
+//
+//        // normally they would get renamed to generations 1 and 2, but since those filenames already exist,
+//        // they get skipped and we end up with generations 3 and 4
+//        assertEquals(2, generations.size());
+//        assertTrue(generations.contains(3));
+//        assertTrue(generations.contains(4));
+//    }
+
+    public void reTest(ColumnFamilyStore cfs, Runnable verify) throws Exception
+    {
+        verify.run();
         cfs.forceBlockingFlush();
-
-        // Nuke the metadata and reload that sstable
-        Collection<SSTableReader> ssTables = cfs.getSSTables();
-        assertEquals(1, ssTables.size());
-        cfs.clearUnsafe();
-        assertEquals(0, cfs.getSSTables().size());
-
-        new File(ssTables.iterator().next().descriptor.filenameFor(Component.STATS)).delete();
-        cfs.loadNewSSTables();
-
-        // Add another cell with a lower timestamp
-        putColsStandard(cfs, key, new BufferCell(cname, ByteBufferUtil.bytes("b"), 1));
-
-        // Test fetching the cell by name returns the first cell
-        SliceByNamesReadCommand cmd = new SliceByNamesReadCommand(keyspaceName, key.getKey(), cfName, System.currentTimeMillis(), new NamesQueryFilter(FBUtilities.singleton(cname, cfs.getComparator())));
-        ColumnFamily cf = cmd.getRow(keyspace).cf;
-        Cell cell = cf.getColumn(cname);
-        assert cell.value().equals(ByteBufferUtil.bytes("a")) : "expecting a, got " + ByteBufferUtil.string(cell.value());
-
-        Keyspace.clear(KEYSPACE1); // CASSANDRA-7195
+        verify.run();
     }
 
-    private static void assertTotalColCount(Collection<Row> rows, int expectedCount)
+    private void assertRangeCount(ColumnFamilyStore cfs, ByteBuffer col, ByteBuffer val, int count)
     {
-        int columns = 0;
-        for (Row row : rows)
-        {
-            columns += row.getLiveCount(new SliceQueryFilter(ColumnSlice.ALL_COLUMNS_ARRAY, false, expectedCount), System.currentTimeMillis());
-        }
-        assert columns == expectedCount : "Expected " + expectedCount + " live columns but got " + columns + ": " + rows;
+        assertRangeCount(cfs, cfs.metadata.getColumnDefinition(col), val, count);
     }
 
-
-    @Test
-    public void testRangeSliceColumnsLimit() throws Throwable
+    private void assertRangeCount(ColumnFamilyStore cfs, ColumnDefinition col, ByteBuffer val, int count)
     {
-        String keyspaceName = KEYSPACE1;
-        String cfName = CF_STANDARD1;
-        Keyspace keyspace = Keyspace.open(keyspaceName);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.clearUnsafe();
 
-        Cell[] cols = new Cell[5];
-        for (int i = 0; i < 5; i++)
-            cols[i] = column("c" + i, "value", 1);
-
-        putColsStandard(cfs, Util.dk("a"), cols[0], cols[1], cols[2], cols[3], cols[4]);
-        putColsStandard(cfs, Util.dk("b"), cols[0], cols[1]);
-        putColsStandard(cfs, Util.dk("c"), cols[0], cols[1], cols[2], cols[3]);
-        cfs.forceBlockingFlush();
-
-        SlicePredicate sp = new SlicePredicate();
-        sp.setSlice_range(new SliceRange());
-        sp.getSlice_range().setCount(1);
-        sp.getSlice_range().setStart(ArrayUtils.EMPTY_BYTE_ARRAY);
-        sp.getSlice_range().setFinish(ArrayUtils.EMPTY_BYTE_ARRAY);
-
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              3,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            3);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              5,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            5);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              8,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            8);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              10,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            10);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              100,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            11);
-
-        // Check that when querying by name, we always include all names for a
-        // gien row even if it means returning more columns than requested (this is necesseray for CQL)
-        sp = new SlicePredicate();
-        sp.setColumn_names(Arrays.asList(
-            ByteBufferUtil.bytes("c0"),
-            ByteBufferUtil.bytes("c1"),
-            ByteBufferUtil.bytes("c2")
-        ));
-
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              1,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            3);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              4,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            5);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              5,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            5);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              6,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            8);
-        assertTotalColCount(cfs.getRangeSlice(Util.range("", ""),
-                                              null,
-                                              ThriftValidation.asIFilter(sp, cfs.metadata, null),
-                                              100,
-                                              System.currentTimeMillis(),
-                                              true,
-                                              false),
-                            8);
-    }
-
-    @Test
-    public void testRangeSlicePaging() throws Throwable
-    {
-        String keyspaceName = KEYSPACE1;
-        String cfName = CF_STANDARD1;
-        Keyspace keyspace = Keyspace.open(keyspaceName);
-        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
-        cfs.clearUnsafe();
-
-        Cell[] cols = new Cell[4];
-        for (int i = 0; i < 4; i++)
-            cols[i] = column("c" + i, "value", 1);
-
-        DecoratedKey ka = Util.dk("a");
-        DecoratedKey kb = Util.dk("b");
-        DecoratedKey kc = Util.dk("c");
-
-        RowPosition min = Util.rp("");
-
-        putColsStandard(cfs, ka, cols[0], cols[1], cols[2], cols[3]);
-        putColsStandard(cfs, kb, cols[0], cols[1], cols[2]);
-        putColsStandard(cfs, kc, cols[0], cols[1], cols[2], cols[3]);
-        cfs.forceBlockingFlush();
-
-        SlicePredicate sp = new SlicePredicate();
-        sp.setSlice_range(new SliceRange());
-        sp.getSlice_range().setCount(1);
-        sp.getSlice_range().setStart(ArrayUtils.EMPTY_BYTE_ARRAY);
-        sp.getSlice_range().setFinish(ArrayUtils.EMPTY_BYTE_ARRAY);
-
-        Collection<Row> rows;
-        Row row, row1, row2;
-        IDiskAtomFilter filter = ThriftValidation.asIFilter(sp, cfs.metadata, null);
-
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(Util.range("", ""), filter, null, 3, true, true, System.currentTimeMillis()));
-        assert rows.size() == 1 : "Expected 1 row, got " + toString(rows);
-        row = rows.iterator().next();
-        assertColumnNames(row, "c0", "c1", "c2");
-
-        sp.getSlice_range().setStart(ByteBufferUtil.getArray(ByteBufferUtil.bytes("c2")));
-        filter = ThriftValidation.asIFilter(sp, cfs.metadata, null);
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(ka, min), filter, null, 3, true, true, System.currentTimeMillis()));
-        assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows);
-        Iterator<Row> iter = rows.iterator();
-        row1 = iter.next();
-        row2 = iter.next();
-        assertColumnNames(row1, "c2", "c3");
-        assertColumnNames(row2, "c0");
-
-        sp.getSlice_range().setStart(ByteBufferUtil.getArray(ByteBufferUtil.bytes("c0")));
-        filter = ThriftValidation.asIFilter(sp, cfs.metadata, null);
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(row2.key, min), filter, null, 3, true, true, System.currentTimeMillis()));
-        assert rows.size() == 1 : "Expected 1 row, got " + toString(rows);
-        row = rows.iterator().next();
-        assertColumnNames(row, "c0", "c1", "c2");
-
-        sp.getSlice_range().setStart(ByteBufferUtil.getArray(ByteBufferUtil.bytes("c2")));
-        filter = ThriftValidation.asIFilter(sp, cfs.metadata, null);
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(row.key, min), filter, null, 3, true, true, System.currentTimeMillis()));
-        assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows);
-        iter = rows.iterator();
-        row1 = iter.next();
-        row2 = iter.next();
-        assertColumnNames(row1, "c2");
-        assertColumnNames(row2, "c0", "c1");
-
-        // Paging within bounds
-        SliceQueryFilter sf = new SliceQueryFilter(cellname("c1"),
-                                                   cellname("c2"),
-                                                   false,
-                                                   0);
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(ka, kc), sf, cellname("c2"), cellname("c1"), null, 2, true, System.currentTimeMillis()));
-        assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows);
-        iter = rows.iterator();
-        row1 = iter.next();
-        row2 = iter.next();
-        assertColumnNames(row1, "c2");
-        assertColumnNames(row2, "c1");
-
-        rows = cfs.getRangeSlice(cfs.makeExtendedFilter(new Bounds<RowPosition>(kb, kc), sf, cellname("c1"), cellname("c1"), null, 10, true, System.currentTimeMillis()));
-        assert rows.size() == 2 : "Expected 2 rows, got " + toString(rows);
-        iter = rows.iterator();
-        row1 = iter.next();
-        row2 = iter.next();
-        assertColumnNames(row1, "c1", "c2");
-        assertColumnNames(row2, "c1");
-    }
-
-    private static String toString(Collection<Row> rows)
-    {
-        try
+        int found = 0;
+        if (count != 0)
         {
-            StringBuilder sb = new StringBuilder();
-            for (Row row : rows)
+            for (FilteredPartition partition : Util.getAll(Util.cmd(cfs).filterOn(col.name.toString(), Operator.EQ, val).build()))
             {
-                sb.append("{");
-                sb.append(ByteBufferUtil.string(row.key.getKey()));
-                sb.append(":");
-                if (row.cf != null && !row.cf.isEmpty())
+                for (Row r : partition)
                 {
-                    for (Cell c : row.cf)
-                        sb.append(" ").append(row.cf.getComparator().getString(c.name()));
+                    if (r.getCell(col).value().equals(val))
+                        ++found;
                 }
-                sb.append("} ");
-            }
-            return sb.toString();
-        }
-        catch (Exception e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    private static void assertColumnNames(Row row, String ... columnNames) throws Exception
-    {
-        if (row == null || row.cf == null)
-            throw new AssertionError("The row should not be empty");
-
-        Iterator<Cell> columns = row.cf.getSortedColumns().iterator();
-        Iterator<String> names = Arrays.asList(columnNames).iterator();
-
-        while (columns.hasNext())
-        {
-            Cell c = columns.next();
-            assert names.hasNext() : "Got more columns that expected (first un

<TRUNCATED>

Mime
View raw message