diff --git a/CHANGES.txt b/CHANGES.txt
index 1025510357..139eb064ea 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -25,13 +25,12 @@
  * CF id is changed to be non-deterministic. Data dir/key cache are created
    uniquely for CF id (CASSANDRA-5202)
  * New counters implementation (CASSANDRA-6504)
- * Replace UnsortedColumns and TreeMapBackedSortedColumns with rewritten
-   ArrayBackedSortedColumns (CASSANDRA-6630, CASSANDRA-6662)
+ * Replace UnsortedColumns, EmptyColumns, TreeMapBackedSortedColumns with new
+   ArrayBackedSortedColumns (CASSANDRA-6630, CASSANDRA-6662, CASSANDRA-6690)
  * Add option to use row cache with a given amount of rows (CASSANDRA-5357)
  * Avoid repairing already repaired data (CASSANDRA-5351)
  * Reject counter updates with USING TTL/TIMESTAMP (CASSANDRA-6649)
 
-
 2.0.6
  * Add compatibility for Hadoop 0.2.x (CASSANDRA-5201)
  * Fix EstimatedHistogram races (CASSANDRA-6682)
@@ -42,6 +41,9 @@
  * Correctly handle null with IF conditions and TTL (CASSANDRA-6623)
  * Account for range/row tombstones in tombstone drop
    time histogram (CASSANDRA-6522)
+ * Stop CommitLogSegment.close() from calling sync() (CASSANDRA-6652)
+ * Make commitlog failure handling configurable (CASSANDRA-6364)
+ * Avoid overlaps in LCS (CASSANDRA-6688)
 Merged from 1.2:
  * Fix broken streams when replacing with same IP (CASSANDRA-6622)
  * Fix upgradesstables NPE for non-CF-based indexes (CASSANDRA-6645)
@@ -49,6 +51,7 @@ Merged from 1.2:
  * Fix mean cells and mean row size per sstable calculations (CASSANDRA-6667)
  * Compact hints after partial replay to clean out tombstones (CASSANDRA-6666)
  * Log USING TTL/TIMESTAMP in a counter update warning (CASSANDRA-6649)
+ * Don't exchange schema between nodes with different versions (CASSANDRA-6695)
 
 
 2.0.5
diff --git a/src/java/org/apache/cassandra/db/AbstractThreadUnsafeSortedColumns.java b/src/java/org/apache/cassandra/db/AbstractThreadUnsafeSortedColumns.java
deleted file mode 100644
index 36b051b4b3..0000000000
--- a/src/java/org/apache/cassandra/db/AbstractThreadUnsafeSortedColumns.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.db;
-
-import org.apache.cassandra.config.CFMetaData;
-
-public abstract class AbstractThreadUnsafeSortedColumns extends ColumnFamily
-{
-    protected DeletionInfo deletionInfo;
-
-    public AbstractThreadUnsafeSortedColumns(CFMetaData metadata)
-    {
-        this(metadata, DeletionInfo.live());
-    }
-
-    protected AbstractThreadUnsafeSortedColumns(CFMetaData metadata, DeletionInfo deletionInfo)
-    {
-        super(metadata);
-        this.deletionInfo = deletionInfo;
-    }
-
-    public DeletionInfo deletionInfo()
-    {
-        return deletionInfo;
-    }
-
-    public void delete(DeletionTime delTime)
-    {
-        deletionInfo.add(delTime);
-    }
-
-    public void delete(DeletionInfo newInfo)
-    {
-        deletionInfo.add(newInfo);
-    }
-
-    protected void delete(RangeTombstone tombstone)
-    {
-        deletionInfo.add(tombstone, getComparator());
-    }
-
-    public void setDeletionInfo(DeletionInfo newInfo)
-    {
-        deletionInfo = newInfo;
-    }
-
-    /**
-     * Purges any tombstones with a local deletion time before gcBefore.
-     * @param gcBefore a timestamp (in seconds) before which tombstones should be purged
-     */
-    public void purgeTombstones(int gcBefore)
-    {
-        deletionInfo.purge(gcBefore);
-    }
-}
diff --git a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
index 8049d87950..eaebd11cd1 100644
--- a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
+++ b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
@@ -37,13 +37,13 @@ import org.apache.cassandra.db.filter.ColumnSlice;
  * main operations performed are iterating over the cells and adding cells
  * (especially if insertion is in sorted order).
  */
-public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
+public class ArrayBackedSortedColumns extends ColumnFamily
 {
-    private static final int INITIAL_CAPACITY = 10;
-
     private static final Cell[] EMPTY_ARRAY = new Cell[0];
+    private static final int MINIMAL_CAPACITY = 10;
 
     private final boolean reversed;
+    private DeletionInfo deletionInfo;
 
     private Cell[] cells;
     private int size;
@@ -61,7 +61,8 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
     {
         super(metadata);
         this.reversed = reversed;
-        this.cells = new Cell[INITIAL_CAPACITY];
+        this.deletionInfo = DeletionInfo.live();
+        this.cells = EMPTY_ARRAY;
         this.size = 0;
         this.sortedSize = 0;
     }
@@ -70,6 +71,7 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
     {
         super(original.metadata);
         this.reversed = original.reversed;
+        this.deletionInfo = DeletionInfo.live(); // this is INTENTIONALLY not set to original.deletionInfo.
         this.cells = Arrays.copyOf(original.cells, original.size);
         this.size = original.size;
         this.sortedSize = original.sortedSize;
@@ -211,9 +213,11 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
      */
     private void internalAdd(Cell cell)
     {
-        // Resize the backing array if we hit the capacity
-        if (cells.length == size)
+        if (cells == EMPTY_ARRAY)
+            cells = new Cell[MINIMAL_CAPACITY];
+        else if (cells.length == size)
             cells = Arrays.copyOf(cells, size * 3 / 2 + 1);
+
         cells[size++] = cell;
     }
 
@@ -318,6 +322,40 @@ public class ArrayBackedSortedColumns extends AbstractThreadUnsafeSortedColumns
         size = sortedSize = 0;
     }
 
+    public DeletionInfo deletionInfo()
+    {
+        return deletionInfo;
+    }
+
+    public void delete(DeletionTime delTime)
+    {
+        deletionInfo.add(delTime);
+    }
+
+    public void delete(DeletionInfo newInfo)
+    {
+        deletionInfo.add(newInfo);
+    }
+
+    protected void delete(RangeTombstone tombstone)
+    {
+        deletionInfo.add(tombstone, getComparator());
+    }
+
+    public void setDeletionInfo(DeletionInfo newInfo)
+    {
+        deletionInfo = newInfo;
+    }
+
+    /**
+     * Purges any tombstones with a local deletion time before gcBefore.
+     * @param gcBefore a timestamp (in seconds) before which tombstones should be purged
+     */
+    public void purgeTombstones(int gcBefore)
+    {
+        deletionInfo.purge(gcBefore);
+    }
+
     public Iterable<CellName> getColumnNames()
     {
         maybeSortCells();
diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java
index 66e8e331ba..3437410931 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamily.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamily.java
@@ -389,14 +389,6 @@ public abstract class ColumnFamily implements Iterable<Cell>, IRowCacheEntry
         return cf1.diff(cf2);
     }
 
-    public void resolve(ColumnFamily cf)
-    {
-        // Row _does_ allow null CF objects :(  seems a necessary evil for efficiency
-        if (cf == null)
-            return;
-        addAll(cf);
-    }
-
     public ColumnStats getColumnStats()
     {
         long minTimestampSeen = deletionInfo().isLive() ? Long.MAX_VALUE : deletionInfo().minTimestamp();
diff --git a/src/java/org/apache/cassandra/db/EmptyColumns.java b/src/java/org/apache/cassandra/db/EmptyColumns.java
deleted file mode 100644
index fa6ea1b6ea..0000000000
--- a/src/java/org/apache/cassandra/db/EmptyColumns.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- * 
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- * 
- *   http://www.apache.org/licenses/LICENSE-2.0
- * 
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- * 
- */
-package org.apache.cassandra.db;
-
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Iterator;
-
-import com.google.common.collect.Iterators;
-
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.db.composites.CellName;
-import org.apache.cassandra.db.filter.ColumnSlice;
-
-public class EmptyColumns extends AbstractThreadUnsafeSortedColumns
-{
-    public static final Factory<EmptyColumns> factory = new Factory<EmptyColumns>()
-    {
-        public EmptyColumns create(CFMetaData metadata, boolean insertReversed)
-        {
-            assert !insertReversed;
-            return new EmptyColumns(metadata, DeletionInfo.live());
-        }
-    };
-
-    public EmptyColumns(CFMetaData metadata, DeletionInfo info)
-    {
-        super(metadata, info);
-    }
-
-    public ColumnFamily cloneMe()
-    {
-        return new EmptyColumns(metadata, deletionInfo);
-    }
-
-    public void clear()
-    {
-    }
-
-    public Factory<EmptyColumns> getFactory()
-    {
-        return factory;
-    }
-
-    public void addColumn(Cell cell)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    public void addAll(ColumnFamily cm)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    public Cell getColumn(CellName name)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    public Iterable<CellName> getColumnNames()
-    {
-        return Collections.emptyList();
-    }
-
-    public Collection<Cell> getSortedColumns()
-    {
-        return Collections.emptyList();
-    }
-
-    public Collection<Cell> getReverseSortedColumns()
-    {
-        return Collections.emptyList();
-    }
-
-    public int getColumnCount()
-    {
-        return 0;
-    }
-
-    public Iterator<Cell> iterator(ColumnSlice[] slices)
-    {
-        return Iterators.emptyIterator();
-    }
-
-    public Iterator<Cell> reverseIterator(ColumnSlice[] slices)
-    {
-        return Iterators.emptyIterator();
-    }
-
-    public boolean isInsertReversed()
-    {
-        return false;
-    }
-}
diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java
index bb7dcefefe..3663380868 100644
--- a/src/java/org/apache/cassandra/db/Mutation.java
+++ b/src/java/org/apache/cassandra/db/Mutation.java
@@ -191,7 +191,7 @@ public class Mutation implements IMutation
             // not in the case where it wasn't there indeed.
             ColumnFamily cf = modifications.put(entry.getKey(), entry.getValue());
             if (cf != null)
-                entry.getValue().resolve(cf);
+                entry.getValue().addAll(cf);
         }
     }
 
diff --git a/src/java/org/apache/cassandra/db/SystemKeyspace.java b/src/java/org/apache/cassandra/db/SystemKeyspace.java
index f8637c1f92..1a00cc0fc9 100644
--- a/src/java/org/apache/cassandra/db/SystemKeyspace.java
+++ b/src/java/org/apache/cassandra/db/SystemKeyspace.java
@@ -810,7 +810,7 @@ public class SystemKeyspace
             return new PaxosState(key, metadata);
         UntypedResultSet.Row row = results.one();
         Commit promised = row.has("in_progress_ballot")
-                        ? new Commit(key, row.getUUID("in_progress_ballot"), EmptyColumns.factory.create(metadata))
+                        ? new Commit(key, row.getUUID("in_progress_ballot"), ArrayBackedSortedColumns.factory.create(metadata))
                         : Commit.emptyCommit(key, metadata);
         // either we have both a recently accepted ballot and update or we have neither
         Commit accepted = row.has("proposal")
diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
index 8715a906f8..2a596ea065 100644
--- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
@@ -77,7 +77,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
         try
         {
             this.indexes = indexEntry.columnsIndex();
-            emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
+            emptyColumnFamily = ArrayBackedSortedColumns.factory.create(sstable.metadata);
             if (indexes.isEmpty())
             {
                 setToRowStart(indexEntry, input);
diff --git a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
index 862fad2c2a..7fb48e3ea3 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
@@ -71,7 +71,7 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt
             if (version.hasRowSizeAndColumnCount)
                 file.readLong();
 
-            emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
+            emptyColumnFamily = ArrayBackedSortedColumns.factory.create(sstable.metadata);
             emptyColumnFamily.delete(DeletionTime.serializer.deserialize(file));
             int columnCount = version.hasRowSizeAndColumnCount ? file.readInt() : Integer.MAX_VALUE;
             atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, columnCount, sstable.descriptor.version);
diff --git a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
index 89181e6ab0..53be7657c2 100644
--- a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
@@ -81,7 +81,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow
         // containing `key` outside of the set of sstables involved in this compaction.
         maxPurgeableTimestamp = controller.maxPurgeableTimestamp(key);
 
-        emptyColumnFamily = EmptyColumns.factory.create(controller.cfs.metadata);
+        emptyColumnFamily = ArrayBackedSortedColumns.factory.create(controller.cfs.metadata);
         emptyColumnFamily.delete(maxRowTombstone);
         if (maxRowTombstone.markedForDeleteAt < maxPurgeableTimestamp)
             emptyColumnFamily.purgeTombstones(controller.gcBefore);
diff --git a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
index cab726d8ea..05b838d363 100644
--- a/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
+++ b/src/java/org/apache/cassandra/db/compaction/LeveledManifest.java
@@ -173,23 +173,6 @@ public class LeveledManifest
         }
     }
 
-    /**
-     * if the number of SSTables in the current compacted set *by itself* exceeds the target level's
-     * (regardless of the level's current contents), find an empty level instead
-     */
-    private int skipLevels(int newLevel, Iterable<SSTableReader> added)
-    {
-        // Note that we now check if the sstables included in the compaction, *before* the compaction, fit in the next level.
-        // This is needed since we need to decide before the actual compaction what level they will be in.
-        // This should be safe, we might skip levels where the compacted data could have fit but that should be ok.
-        while (maxBytesForLevel(newLevel) < SSTableReader.getTotalBytes(added)
-               && getLevel(newLevel + 1).isEmpty())
-        {
-            newLevel++;
-        }
-        return newLevel;
-    }
-
     public synchronized void replace(Collection<SSTableReader> removed, Collection<SSTableReader> added)
     {
         assert !removed.isEmpty(); // use add() instead of promote when adding new sstables
@@ -560,7 +543,10 @@ public class LeveledManifest
 
                 for (SSTableReader newCandidate : overlappedL0)
                 {
-                    candidates.add(newCandidate);
+                    // overlappedL0 could contain sstables that are not in compactingL0, but do overlap
+                    // other sstables that are
+                    if (overlapping(newCandidate, compactingL0).isEmpty())
+                        candidates.add(newCandidate);
                     remaining.remove(newCandidate);
                 }
 
@@ -683,7 +669,6 @@ public class LeveledManifest
         else
         {
             newLevel = minimumLevel == maximumLevel ? maximumLevel + 1 : maximumLevel;
-            newLevel = skipLevels(newLevel, sstables);
             assert newLevel > 0;
         }
         return newLevel;
diff --git a/src/java/org/apache/cassandra/db/filter/ColumnSlice.java b/src/java/org/apache/cassandra/db/filter/ColumnSlice.java
index f5ea49ac71..02a5a0c1cb 100644
--- a/src/java/org/apache/cassandra/db/filter/ColumnSlice.java
+++ b/src/java/org/apache/cassandra/db/filter/ColumnSlice.java
@@ -138,55 +138,6 @@ public class ColumnSlice
         }
     }
 
-    public static class NavigableMapIterator extends AbstractIterator<Cell>
-    {
-        private final NavigableMap<CellName, Cell> map;
-        private final ColumnSlice[] slices;
-
-        private int idx = 0;
-        private Iterator<Cell> currentSlice;
-
-        public NavigableMapIterator(NavigableMap<CellName, Cell> map, ColumnSlice[] slices)
-        {
-            this.map = map;
-            this.slices = slices;
-        }
-
-        protected Cell computeNext()
-        {
-            if (currentSlice == null)
-            {
-                if (idx >= slices.length)
-                    return endOfData();
-
-                ColumnSlice slice = slices[idx++];
-                // Note: we specialize the case of start == "" and finish = "" because it is slightly more efficient, but also they have a specific
-                // meaning (namely, they always extend to the beginning/end of the range).
-                if (slice.start.isEmpty())
-                {
-                    if (slice.finish.isEmpty())
-                        currentSlice = map.values().iterator();
-                    else
-                        currentSlice = map.headMap(new FakeCellName(slice.finish), true).values().iterator();
-                }
-                else if (slice.finish.isEmpty())
-                {
-                    currentSlice = map.tailMap(new FakeCellName(slice.start), true).values().iterator();
-                }
-                else
-                {
-                    currentSlice = map.subMap(new FakeCellName(slice.start), true, new FakeCellName(slice.finish), true).values().iterator();
-                }
-            }
-
-            if (currentSlice.hasNext())
-                return currentSlice.next();
-
-            currentSlice = null;
-            return computeNext();
-        }
-    }
-
     public static class NavigableSetIterator extends AbstractIterator<Cell>
     {
         private final NavigableSet<Cell> set;
diff --git a/src/java/org/apache/cassandra/db/index/composites/CompositesSearcher.java b/src/java/org/apache/cassandra/db/index/composites/CompositesSearcher.java
index fad3d506e6..41c9f410c9 100644
--- a/src/java/org/apache/cassandra/db/index/composites/CompositesSearcher.java
+++ b/src/java/org/apache/cassandra/db/index/composites/CompositesSearcher.java
@@ -274,7 +274,7 @@ public class CompositesSearcher extends SecondaryIndexSearcher
 
                         if (data == null)
                             data = ArrayBackedSortedColumns.factory.create(baseCfs.metadata);
-                        data.resolve(newData);
+                        data.addAll(newData);
                         columnsCount += dataFilter.lastCounted();
                     }
                  }
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
index 4e23ac2845..11f2f71f91 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
@@ -99,7 +99,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
 
         try
         {
-            columnFamily = EmptyColumns.factory.create(metadata);
+            columnFamily = ArrayBackedSortedColumns.factory.create(metadata);
             columnFamily.delete(DeletionTime.serializer.deserialize(in));
             columnCount = dataVersion.hasRowSizeAndColumnCount ? in.readInt() : Integer.MAX_VALUE;
             atomIterator = columnFamily.metadata().getOnDiskIterator(in, columnCount, dataVersion);
diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java
index 179e9554fe..b69ac102ef 100644
--- a/src/java/org/apache/cassandra/service/CassandraDaemon.java
+++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java
@@ -56,6 +56,7 @@ import org.apache.cassandra.metrics.StorageMetrics;
 import org.apache.cassandra.thrift.ThriftServer;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.CLibrary;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Mx4jTool;
 import org.apache.cassandra.utils.Pair;
 
@@ -331,7 +332,8 @@ public class CassandraDaemon
             }
         }
 
-        waitForGossipToSettle();
+        if (!FBUtilities.getBroadcastAddress().equals(InetAddress.getLoopbackAddress()))
+            waitForGossipToSettle();
 
         // Thift
         InetAddress rpcAddr = DatabaseDescriptor.getRpcAddress();
@@ -458,7 +460,6 @@ public class CassandraDaemon
         destroy();
     }
 
-
     private void waitForGossipToSettle()
     {
         int forceAfter = Integer.getInteger("cassandra.skip_wait_for_gossip_to_settle", -1);
@@ -502,7 +503,7 @@ public class CassandraDaemon
         if (totalPolls > GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED)
             logger.info("Gossip settled after {} extra polls; proceeding", totalPolls - GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED);
         else
-            logger.debug("Gossip settled after {} extra polls; proceeding", totalPolls - GOSSIP_SETTLE_POLL_SUCCESSES_REQUIRED);
+            logger.info("No gossip backlog; proceeding");
     }
 
     public static void stop(String[] args)
diff --git a/src/java/org/apache/cassandra/service/MigrationManager.java b/src/java/org/apache/cassandra/service/MigrationManager.java
index 39e52f3394..6e5a17f7dc 100644
--- a/src/java/org/apache/cassandra/service/MigrationManager.java
+++ b/src/java/org/apache/cassandra/service/MigrationManager.java
@@ -132,11 +132,11 @@ public class MigrationManager
     private static boolean shouldPullSchemaFrom(InetAddress endpoint)
     {
         /*
-         * Don't request schema from nodes with a higher major (may have incompatible schema)
+         * Don't request schema from nodes with a differnt or unknonw major version (may have incompatible schema)
          * Don't request schema from fat clients
          */
         return MessagingService.instance().knowsVersion(endpoint)
-                && MessagingService.instance().getVersion(endpoint) <= MessagingService.current_version
+                && MessagingService.instance().getVersion(endpoint) == MessagingService.current_version
                 && !Gossiper.instance.isFatClient(endpoint);
     }
 
@@ -302,15 +302,13 @@ public class MigrationManager
 
         for (InetAddress endpoint : Gossiper.instance.getLiveMembers())
         {
-            if (endpoint.equals(FBUtilities.getBroadcastAddress()))
-                continue; // we've dealt with localhost already
-
-            // don't send schema to the nodes with the versions older than current major
-            if (MessagingService.instance().getVersion(endpoint) < MessagingService.current_version)
-                continue;
-
-            pushSchemaMutation(endpoint, schema);
+            // only push schema to nodes with known and equal versions
+            if (!endpoint.equals(FBUtilities.getBroadcastAddress()) &&
+                    MessagingService.instance().knowsVersion(endpoint) &&
+                    MessagingService.instance().getVersion(endpoint) == MessagingService.current_version)
+                pushSchemaMutation(endpoint, schema);
         }
+
         return f;
     }
 
diff --git a/src/java/org/apache/cassandra/service/ReadCallback.java b/src/java/org/apache/cassandra/service/ReadCallback.java
index 0c0d50cd9a..c43feaa953 100644
--- a/src/java/org/apache/cassandra/service/ReadCallback.java
+++ b/src/java/org/apache/cassandra/service/ReadCallback.java
@@ -98,10 +98,8 @@ public class ReadCallback<TMessage, TResolved> implements IAsyncCallback<TMessag
         if (!await(command.getTimeout(), TimeUnit.MILLISECONDS))
         {
             // Same as for writes, see AbstractWriteResponseHandler
-            int acks = received;
-            if (resolver.isDataPresent() && acks >= blockfor)
-                acks = blockfor - 1;
             ReadTimeoutException ex = new ReadTimeoutException(consistencyLevel, received, blockfor, resolver.isDataPresent());
+
             if (logger.isDebugEnabled())
                 logger.debug("Read timeout: {}", ex.toString());
             throw ex;
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 8c3119290f..05fdd61dd9 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -232,7 +232,7 @@ public class StorageProxy implements StorageProxyMBean
             {
                 Tracing.trace("CAS precondition {} does not match current values {}", conditions, current);
                 // We should not return null as this means success
-                return current == null ? EmptyColumns.factory.create(metadata) : current;
+                return current == null ? ArrayBackedSortedColumns.factory.create(metadata) : current;
             }
 
             // finish the paxos round w/ the desired updates
@@ -603,7 +603,7 @@ public class StorageProxy implements StorageProxyMBean
 
     private static void asyncRemoveFromBatchlog(Collection<InetAddress> endpoints, UUID uuid)
     {
-        ColumnFamily cf = EmptyColumns.factory.create(Schema.instance.getCFMetaData(Keyspace.SYSTEM_KS, SystemKeyspace.BATCHLOG_CF));
+        ColumnFamily cf = ArrayBackedSortedColumns.factory.create(Schema.instance.getCFMetaData(Keyspace.SYSTEM_KS, SystemKeyspace.BATCHLOG_CF));
         cf.delete(new DeletionInfo(FBUtilities.timestampMicros(), (int) (System.currentTimeMillis() / 1000)));
         AbstractWriteResponseHandler handler = new WriteResponseHandler(endpoints,
                                                                         Collections.<InetAddress>emptyList(),
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 958e7d4729..cca104ed78 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -843,7 +843,6 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
             if (!current.isEmpty())
                 for (InetAddress existing : current)
                     Gossiper.instance.replacedEndpoint(existing);
-            logger.info("Startup completed! Now serving reads.");
             assert tokenMetadata.sortedTokens().size() > 0;
 
             Auth.setup();
diff --git a/src/java/org/apache/cassandra/service/pager/QueryPagers.java b/src/java/org/apache/cassandra/service/pager/QueryPagers.java
index 32ff86c468..f186e693d1 100644
--- a/src/java/org/apache/cassandra/service/pager/QueryPagers.java
+++ b/src/java/org/apache/cassandra/service/pager/QueryPagers.java
@@ -147,7 +147,7 @@ public class QueryPagers
                 {
                     List<Row> rows = pager.fetchPage(pageSize);
                     ColumnFamily cf = rows.isEmpty() ? null : rows.get(0).cf;
-                    return cf == null ? EmptyColumns.factory.create(cfs.metadata) : cf;
+                    return cf == null ? ArrayBackedSortedColumns.factory.create(cfs.metadata) : cf;
                 }
                 catch (Exception e)
                 {
diff --git a/src/java/org/apache/cassandra/service/paxos/Commit.java b/src/java/org/apache/cassandra/service/paxos/Commit.java
index aae9b72591..6abfe4cb03 100644
--- a/src/java/org/apache/cassandra/service/paxos/Commit.java
+++ b/src/java/org/apache/cassandra/service/paxos/Commit.java
@@ -57,7 +57,7 @@ public class Commit
 
     public static Commit newPrepare(ByteBuffer key, CFMetaData metadata, UUID ballot)
     {
-        return new Commit(key, ballot, EmptyColumns.factory.create(metadata));
+        return new Commit(key, ballot, ArrayBackedSortedColumns.factory.create(metadata));
     }
 
     public static Commit newProposal(ByteBuffer key, UUID ballot, ColumnFamily update)
@@ -67,7 +67,7 @@ public class Commit
 
     public static Commit emptyCommit(ByteBuffer key, CFMetaData metadata)
     {
-        return new Commit(key, UUIDGen.minTimeUUID(0), EmptyColumns.factory.create(metadata));
+        return new Commit(key, UUIDGen.minTimeUUID(0), ArrayBackedSortedColumns.factory.create(metadata));
     }
 
     public boolean isAfter(Commit other)
diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java
index 92a0b4a90b..7dc86dae44 100644
--- a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java
+++ b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java
@@ -195,12 +195,6 @@ public class RangeTombstoneListTest
 
     @Test
     public void addAllTest()
-    {
-        //addAllTest(false);
-        addAllTest(true);
-    }
-
-    private void addAllTest(boolean doMerge)
     {
         RangeTombstoneList l1 = new RangeTombstoneList(cmp, 0);
         l1.add(rt(0, 4, 5));
diff --git a/test/unit/org/apache/cassandra/db/RowTest.java b/test/unit/org/apache/cassandra/db/RowTest.java
index 4a686c5f1a..9a912850ec 100644
--- a/test/unit/org/apache/cassandra/db/RowTest.java
+++ b/test/unit/org/apache/cassandra/db/RowTest.java
@@ -59,7 +59,7 @@ public class RowTest extends SchemaLoader
         cf2.addColumn(column("one", "B", 1));
         cf2.addColumn(column("two", "C", 1));
 
-        cf1.resolve(cf2);
+        cf1.addAll(cf2);
         assert Arrays.equals(cf1.getColumn(CellNames.simpleDense(ByteBufferUtil.bytes("one"))).value().array(), "B".getBytes());
         assert Arrays.equals(cf1.getColumn(CellNames.simpleDense(ByteBufferUtil.bytes("two"))).value().array(), "C".getBytes());
     }
diff --git a/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java b/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java
index a11a128c57..c7470f8c1d 100644
--- a/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java
+++ b/test/unit/org/apache/cassandra/locator/OldNetworkTopologyStrategyTest.java
@@ -181,7 +181,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
         BigIntegerToken newToken = new BigIntegerToken("21267647932558653966460912964485513216");
         BigIntegerToken[] tokens = initTokens();
         BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
-        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
+        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
 
         assertEquals(ranges.left.iterator().next().left, tokensAfterMove[movingNodeIdx]);
         assertEquals(ranges.left.iterator().next().right, tokens[movingNodeIdx]);
@@ -198,7 +198,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
         BigIntegerToken newToken = new BigIntegerToken("35267647932558653966460912964485513216");
         BigIntegerToken[] tokens = initTokens();
         BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
-        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
+        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
 
         assertEquals("No data should be streamed", ranges.left.size(), 0);
         assertEquals(ranges.right.iterator().next().left, tokens[movingNodeIdx]);
@@ -216,7 +216,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
         BigIntegerToken newToken = new BigIntegerToken("90070591730234615865843651857942052864");
         BigIntegerToken[] tokens = initTokens();
         BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
-        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
+        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
 
         // sort the results, so they can be compared
         Range[] toStream = ranges.left.toArray(new Range[0]);
@@ -248,7 +248,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
         BigIntegerToken newToken = new BigIntegerToken("52535295865117307932921825928971026432");
         BigIntegerToken[] tokens = initTokens();
         BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
-        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
+        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
 
 
         // sort the results, so they can be compared
@@ -280,7 +280,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
         BigIntegerToken newToken = new BigIntegerToken("158873535527910577765226390751398592512");
         BigIntegerToken[] tokens = initTokens();
         BigIntegerToken[] tokensAfterMove = initTokensAfterMove(tokens, movingNodeIdx, newToken);
-        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx, newToken);
+        Pair<Set<Range<Token>>, Set<Range<Token>>> ranges = calculateStreamAndFetchRanges(tokens, tokensAfterMove, movingNodeIdx);
 
         Range[] toStream = ranges.left.toArray(new Range[0]);
         Range[] toFetch = ranges.right.toArray(new Range[0]);
@@ -350,7 +350,7 @@ public class OldNetworkTopologyStrategyTest extends SchemaLoader
 
     }
 
-    private Pair<Set<Range<Token>>, Set<Range<Token>>> calculateStreamAndFetchRanges(BigIntegerToken[] tokens, BigIntegerToken[] tokensAfterMove, int movingNodeIdx, BigIntegerToken newToken) throws UnknownHostException
+    private Pair<Set<Range<Token>>, Set<Range<Token>>> calculateStreamAndFetchRanges(BigIntegerToken[] tokens, BigIntegerToken[] tokensAfterMove, int movingNodeIdx) throws UnknownHostException
     {
         RackInferringSnitch endpointSnitch = new RackInferringSnitch();
 
diff --git a/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT-jar-with-dependencies.jar b/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT-jar-with-dependencies.jar
deleted file mode 100644
index 1f4dafddfb..0000000000
Binary files a/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT-jar-with-dependencies.jar and /dev/null differ
diff --git a/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT.jar b/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT.jar
deleted file mode 100644
index c0d4242725..0000000000
Binary files a/tools/lib/cassandra-driver-core-2.0.0-rc2-SNAPSHOT.jar and /dev/null differ
diff --git a/tools/lib/cassandra-driver-core-2.0.0-rc3-SNAPSHOT.jar b/tools/lib/cassandra-driver-core-2.0.0-rc3-SNAPSHOT.jar
new file mode 100644
index 0000000000..54a175f01f
Binary files /dev/null and b/tools/lib/cassandra-driver-core-2.0.0-rc3-SNAPSHOT.jar differ
diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
index d85f01062f..94824ec0db 100644
--- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java
+++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
@@ -155,9 +155,7 @@ public class StressAction implements Runnable
 
     private boolean hasAverageImprovement(List<StressMetrics> results, int count, double minImprovement)
     {
-        if (results.size() < count + 1)
-            return true;
-        return averageImprovement(results, count) >= minImprovement;
+        return results.size() < count + 1 || averageImprovement(results, count) >= minImprovement;
     }
 
     private double averageImprovement(List<StressMetrics> results, int count)
@@ -389,7 +387,7 @@ public class StressAction implements Runnable
             int batchSize = (int) (operations / (1 << 19));
             if (batchSize < 20)
                 batchSize = 20;
-            ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<Work>(
+            ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<>(
                     (int) ((operations / batchSize)
                   + (operations % batchSize == 0 ? 0 : 1))
             );
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
index 7733ed6889..59079560d0 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
@@ -27,17 +27,17 @@ public class DataGenStringDictionary extends DataGen
     @Override
     public void generate(ByteBuffer fill, long index, ByteBuffer seed)
     {
-        fill(fill, 0);
+        fill(fill);
     }
 
     @Override
     public void generate(List<ByteBuffer> fills, long index, ByteBuffer seed)
     {
         for (int i = 0 ; i < fills.size() ; i++)
-            fill(fills.get(0), i);
+            fill(fills.get(0));
     }
 
-    private void fill(ByteBuffer fill, int column)
+    private void fill(ByteBuffer fill)
     {
         fill.clear();
         byte[] trg = fill.array();
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
index 4c5bb89b53..62ea9220bc 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
@@ -50,7 +50,7 @@ public class DataGenStringRepeats extends DataGen
 
     private byte[] getData(long index, int column, ByteBuffer seed)
     {
-        final long key = (column * repeatFrequency) + ((seed == null ? index : Math.abs(seed.hashCode())) % repeatFrequency);
+        final long key = ((long)column * repeatFrequency) + ((seed == null ? index : Math.abs(seed.hashCode())) % repeatFrequency);
         byte[] r = cache.get(key);
         if (r != null)
             return r;
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java
index b68ab3ca29..6749eae80c 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java
@@ -21,6 +21,8 @@ public class RowGenDistributedSize extends RowGen
     final ByteBuffer[] ret;
     final int[] sizes;
 
+    final boolean isDeterministic;
+
     public RowGenDistributedSize(DataGen dataGenerator, Distribution countDistribution, Distribution sizeDistribution)
     {
         super(dataGenerator);
@@ -28,6 +30,8 @@ public class RowGenDistributedSize extends RowGen
         this.sizeDistribution = sizeDistribution;
         ret = new ByteBuffer[(int) countDistribution.maxValue()];
         sizes = new int[ret.length];
+        this.isDeterministic = dataGen.isDeterministic() && countDistribution.maxValue() == countDistribution.minValue()
+            && sizeDistribution.minValue() == sizeDistribution.maxValue();
     }
 
     ByteBuffer getBuffer(int size)
@@ -78,7 +82,7 @@ public class RowGenDistributedSize extends RowGen
     @Override
     public boolean isDeterministic()
     {
-        return false;
+        return isDeterministic;
     }
 
 }
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
index bd2f131136..b17f52042a 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
@@ -192,8 +192,12 @@ public abstract class CqlOperation<V> extends Operation
             if (result.length != expect.size())
                 return false;
             for (int i = 0 ; i < result.length ; i++)
-                if (!expect.get(i).equals(Arrays.asList(result[i])))
+            {
+                List<ByteBuffer> resultRow = Arrays.asList(result[i]);
+                resultRow = resultRow.subList(1, resultRow.size());
+                if (expect.get(i) != null && !expect.get(i).equals(resultRow))
                     return false;
+            }
             return true;
         }
     }
@@ -521,20 +525,19 @@ public abstract class CqlOperation<V> extends Operation
                 @Override
                 public ByteBuffer[][] apply(ResultMessage result)
                 {
-                    if (result instanceof ResultMessage.Rows)
+                    if (!(result instanceof ResultMessage.Rows))
+                        return new ByteBuffer[0][];
+
+                    ResultMessage.Rows rows = ((ResultMessage.Rows) result);
+                    ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
+                    for (int i = 0 ; i < r.length ; i++)
                     {
-                        ResultMessage.Rows rows = ((ResultMessage.Rows) result);
-                        ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
-                        for (int i = 0 ; i < r.length ; i++)
-                        {
-                            List<ByteBuffer> row = rows.result.rows.get(i);
-                            r[i] = new ByteBuffer[row.size()];
-                            for (int j = 0 ; j < row.size() ; j++)
-                                r[i][j] = row.get(j);
-                        }
-                        return r;
+                        List<ByteBuffer> row = rows.result.rows.get(i);
+                        r[i] = new ByteBuffer[row.size()];
+                        for (int j = 0 ; j < row.size() ; j++)
+                            r[i][j] = row.get(j);
                     }
-                    return new ByteBuffer[0][];
+                    return r;
                 }
             };
         }
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
index 6cef0bfc6f..e2647bb9e7 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
@@ -73,9 +73,10 @@ public class SettingsKey implements Serializable
 
     public KeyGen newKeyGen()
     {
-        if (range != null)
-            return new KeyGen(new DataGenHexFromOpIndex(range[0], range[1]), keySize);
-        return new KeyGen(new DataGenHexFromDistribution(distribution.get()), keySize);
+        return new KeyGen(range == null
+                            ? new DataGenHexFromDistribution(distribution.get())
+                            : new DataGenHexFromOpIndex(range[0], range[1]),
+                          keySize);
     }
 
     // CLI Utility Methods
diff --git a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
index cf37040665..7bde900623 100644
--- a/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
+++ b/tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
@@ -62,7 +62,9 @@ public class JavaDriverClient
     public void connect(ProtocolOptions.Compression compression) throws Exception
     {
         Cluster.Builder clusterBuilder = Cluster.builder()
-                .addContactPoint(host).withPort(port);
+                                                .addContactPoint(host)
+                                                .withPort(port)
+                                                .withoutMetrics(); // The driver uses metrics 3 with conflict with our version
         clusterBuilder.withCompression(compression);
         if (encryptionOptions.enabled)
         {
@@ -142,7 +144,6 @@ public class JavaDriverClient
 
     public void disconnect()
     {
-        FBUtilities.waitOnFuture(cluster.shutdown());
+        cluster.close();
     }
-
-}
\ No newline at end of file
+}