cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject [2/2] git commit: Single-pass compaction patch by jbellis; reviewed by marcuse for CASSANDRA-4180
Date Tue, 07 May 2013 19:22:21 GMT
Single-pass compaction
patch by jbellis; reviewed by marcuse for CASSANDRA-4180


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/a9bd531b
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/a9bd531b
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/a9bd531b

Branch: refs/heads/trunk
Commit: a9bd531bc9a286cc8e81a800aaa29c1fead62dcd
Parents: 3ca160e
Author: Jonathan Ellis <jbellis@apache.org>
Authored: Tue May 7 14:20:53 2013 -0500
Committer: Jonathan Ellis <jbellis@apache.org>
Committed: Tue May 7 14:21:23 2013 -0500

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 src/java/org/apache/cassandra/db/Column.java       |   27 +-
 .../org/apache/cassandra/db/ColumnFamilyStore.java |    4 +-
 src/java/org/apache/cassandra/db/ColumnIndex.java  |   46 +++-
 src/java/org/apache/cassandra/db/OnDiskAtom.java   |    5 +-
 .../apache/cassandra/db/RowIteratorFactory.java    |    3 +-
 .../columniterator/ICountableColumnIterator.java   |   25 --
 .../db/columniterator/IndexedSliceReader.java      |   29 +-
 .../db/columniterator/SSTableNamesIterator.java    |   10 +-
 .../db/columniterator/SimpleSliceReader.java       |   32 +--
 .../db/compaction/AbstractCompactedRow.java        |   18 +-
 .../cassandra/db/compaction/CompactionManager.java |   53 ++--
 .../cassandra/db/compaction/CompactionTask.java    |    5 +-
 .../db/compaction/LazilyCompactedRow.java          |  111 ++------
 .../db/compaction/ParallelCompactionIterable.java  |   28 +--
 .../cassandra/db/compaction/PrecompactedRow.java   |   40 +---
 .../apache/cassandra/db/compaction/Scrubber.java   |   49 ++--
 .../apache/cassandra/io/sstable/Descriptor.java    |    3 +
 .../io/sstable/SSTableBoundedScanner.java          |   95 ------
 .../io/sstable/SSTableIdentityIterator.java        |   99 ++-----
 .../apache/cassandra/io/sstable/SSTableReader.java |   13 +-
 .../cassandra/io/sstable/SSTableScanner.java       |  184 +++++-------
 .../apache/cassandra/io/sstable/SSTableWriter.java |  134 ++++-----
 .../cassandra/streaming/IncomingStreamReader.java  |   33 +--
 .../org/apache/cassandra/tools/SSTableExport.java  |   54 ++--
 test/unit/org/apache/cassandra/SchemaLoader.java   |   21 +-
 test/unit/org/apache/cassandra/Util.java           |   18 --
 test/unit/org/apache/cassandra/db/ScrubTest.java   |  222 ---------------
 .../apache/cassandra/db/SerializationsTest.java    |    3 +-
 .../cassandra/db/compaction/CompactionsTest.java   |    3 +-
 .../cassandra/io/LazilyCompactedRowTest.java       |   44 ++--
 .../apache/cassandra/io/sstable/SSTableTest.java   |  112 --------
 .../cassandra/service/LeaveAndBootstrapTest.java   |    2 +-
 .../org/apache/cassandra/service/MoveTest.java     |    2 +-
 .../org/apache/cassandra/service/RemoveTest.java   |    2 +-
 35 files changed, 424 insertions(+), 1106 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 60accb1..4cd7b6b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.0
+ * Single-pass compaction (CASSANDRA-4180)
  * Removed token range bisection (CASSANDRA-5518)
  * Removed compatibility with pre-1.2.5 sstables and network messages
    (CASSANDRA-5511)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/Column.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/Column.java b/src/java/org/apache/cassandra/db/Column.java
index bf4f5b3..a8c1f94 100644
--- a/src/java/org/apache/cassandra/db/Column.java
+++ b/src/java/org/apache/cassandra/db/Column.java
@@ -27,6 +27,8 @@ import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
+import com.google.common.collect.AbstractIterator;
+
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.db.marshal.*;
 import org.apache.cassandra.io.sstable.Descriptor;
@@ -50,33 +52,34 @@ public class Column implements OnDiskAtom
         return OnDiskAtom.Serializer.instance;
     }
 
+    /**
+     * For 2.0-formatted sstables (where column count is not stored), @param count should be Integer.MAX_VALUE,
+     * and we will look for the end-of-row column name marker instead of relying on that.
+     */
     public static Iterator<OnDiskAtom> onDiskIterator(final DataInput in, final int count, final ColumnSerializer.Flag flag, final int expireBefore, final Descriptor.Version version)
     {
-        return new Iterator<OnDiskAtom>()
+        return new AbstractIterator<OnDiskAtom>()
         {
             int i = 0;
 
-            public boolean hasNext()
+            protected OnDiskAtom computeNext()
             {
-                return i < count;
-            }
+                if (i++ >= count)
+                    return endOfData();
 
-            public OnDiskAtom next()
-            {
-                ++i;
+                OnDiskAtom atom;
                 try
                 {
-                    return onDiskSerializer().deserializeFromSSTable(in, flag, expireBefore, version);
+                    atom = onDiskSerializer().deserializeFromSSTable(in, flag, expireBefore, version);
                 }
                 catch (IOException e)
                 {
                     throw new IOError(e);
                 }
-            }
+                if (atom == null)
+                    return endOfData();
 
-            public void remove()
-            {
-                throw new UnsupportedOperationException();
+                return atom;
             }
         };
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index b07d97c..bbbb554 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -1480,6 +1480,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             final CloseableIterator<Row> iterator = RowIteratorFactory.getIterator(view.memtables, view.sstables, startWith, stopAt, filter, this);
 
+            // todo this could be pushed into SSTableScanner
             return new AbstractScanIterator()
             {
                 protected Row computeNext()
@@ -1498,7 +1499,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
                     if (!range.contains(key))
                         return computeNext();
 
-                    logger.trace("scanned {}", key);
+                    if (logger.isTraceEnabled())
+                        logger.trace("scanned {}", metadata.getKeyValidator().getString(key.key));
 
                     return current;
                 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/ColumnIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ColumnIndex.java b/src/java/org/apache/cassandra/db/ColumnIndex.java
index 57bd995..bcc5c2f 100644
--- a/src/java/org/apache/cassandra/db/ColumnIndex.java
+++ b/src/java/org/apache/cassandra/db/ColumnIndex.java
@@ -26,6 +26,7 @@ import com.google.common.annotations.VisibleForTesting;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.io.sstable.IndexHelper;
+import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class ColumnIndex
 {
@@ -65,13 +66,21 @@ public class ColumnIndex
         private final DataOutput output;
         private final RangeTombstone.Tracker tombstoneTracker;
         private int atomCount;
+        private final ByteBuffer key;
+        private final DeletionInfo deletionInfo;
 
         public Builder(ColumnFamily cf,
                        ByteBuffer key,
                        DataOutput output,
                        boolean fromStream)
         {
-            this.indexOffset = rowHeaderSize(key, cf.deletionInfo());
+            assert cf != null;
+            assert key != null;
+            assert output != null;
+
+            this.key = key;
+            deletionInfo = cf.deletionInfo();
+            this.indexOffset = rowHeaderSize(key, deletionInfo);
             this.result = new ColumnIndex(new ArrayList<IndexHelper.IndexInfo>());
             this.output = output;
             this.tombstoneTracker = fromStream ? null : new RangeTombstone.Tracker(cf.getComparator());
@@ -94,9 +103,7 @@ public class ColumnIndex
             // TODO fix constantSize when changing the nativeconststs.
             int keysize = key.remaining();
             return typeSizes.sizeof((short) keysize) + keysize          // Row key
-                 + typeSizes.sizeof(0L)                                 // Row data size
-                 + DeletionTime.serializer.serializedSize(delInfo.getTopLevelDeletion(), typeSizes)
-                 + typeSizes.sizeof(0);                                 // Column count
+                 + DeletionTime.serializer.serializedSize(delInfo.getTopLevelDeletion(), typeSizes);
         }
 
         public RangeTombstone.Tracker tombstoneTracker()
@@ -119,6 +126,7 @@ public class ColumnIndex
          */
         public ColumnIndex build(ColumnFamily cf) throws IOException
         {
+            // cf has disentangled the columns and range tombstones, we need to re-interleave them in comparator order
             Iterator<RangeTombstone> rangeIter = cf.deletionInfo().rangeIterator();
             RangeTombstone tombstone = rangeIter.hasNext() ? rangeIter.next() : null;
             Comparator<ByteBuffer> comparator = cf.getComparator();
@@ -138,15 +146,22 @@ public class ColumnIndex
                 add(tombstone);
                 tombstone = rangeIter.hasNext() ? rangeIter.next() : null;
             }
-            return build();
+            ColumnIndex index = build();
+
+            finish();
+
+            return index;
         }
 
         public ColumnIndex build(Iterable<OnDiskAtom> columns) throws IOException
         {
             for (OnDiskAtom c : columns)
                 add(c);
+            ColumnIndex index = build();
+
+            finish();
 
-            return build();
+            return index;
         }
 
         public void add(OnDiskAtom column) throws IOException
@@ -177,8 +192,8 @@ public class ColumnIndex
                 lastBlockClosing = column;
             }
 
-            if (output != null)
-                atomSerializer.serializeForSSTable(column, output);
+            maybeWriteRowHeader();
+            atomSerializer.serializeForSSTable(column, output);
 
             // TODO: Should deal with removing unneeded tombstones
             if (tombstoneTracker != null)
@@ -187,6 +202,15 @@ public class ColumnIndex
             lastColumn = column;
         }
 
+        private void maybeWriteRowHeader() throws IOException
+        {
+            if (lastColumn == null)
+            {
+                ByteBufferUtil.writeWithShortLength(key, output);
+                DeletionInfo.serializer().serializeForSSTable(deletionInfo, output);
+            }
+        }
+
         public ColumnIndex build()
         {
             // all columns were GC'd after all
@@ -204,5 +228,11 @@ public class ColumnIndex
             assert result.columnsIndex.size() > 0;
             return result;
         }
+
+        public void finish() throws IOException
+        {
+            if (!deletionInfo.equals(DeletionInfo.LIVE))
+                maybeWriteRowHeader();
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/OnDiskAtom.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/OnDiskAtom.java b/src/java/org/apache/cassandra/db/OnDiskAtom.java
index 7cf7914..2fdb7ad 100644
--- a/src/java/org/apache/cassandra/db/OnDiskAtom.java
+++ b/src/java/org/apache/cassandra/db/OnDiskAtom.java
@@ -73,7 +73,10 @@ public interface OnDiskAtom
         {
             ByteBuffer name = ByteBufferUtil.readWithShortLength(in);
             if (name.remaining() <= 0)
-                throw ColumnSerializer.CorruptColumnException.create(in, name);
+            {
+                // SSTableWriter.END_OF_ROW
+                return null;
+            }
 
             int b = in.readUnsignedByte();
             if ((b & ColumnSerializer.RANGE_TOMBSTONE_MASK) != 0)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/RowIteratorFactory.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/RowIteratorFactory.java b/src/java/org/apache/cassandra/db/RowIteratorFactory.java
index 1bc506b..4588146 100644
--- a/src/java/org/apache/cassandra/db/RowIteratorFactory.java
+++ b/src/java/org/apache/cassandra/db/RowIteratorFactory.java
@@ -69,8 +69,7 @@ public class RowIteratorFactory
 
         for (SSTableReader sstable : sstables)
         {
-            final SSTableScanner scanner = sstable.getScanner(filter);
-            scanner.seekTo(startWith);
+            final SSTableScanner scanner = sstable.getScanner(filter, startWith);
             iterators.add(scanner);
         }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/columniterator/ICountableColumnIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/ICountableColumnIterator.java b/src/java/org/apache/cassandra/db/columniterator/ICountableColumnIterator.java
deleted file mode 100644
index 0b5fa97..0000000
--- a/src/java/org/apache/cassandra/db/columniterator/ICountableColumnIterator.java
+++ /dev/null
@@ -1,25 +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.columniterator;
-
-public interface ICountableColumnIterator extends OnDiskAtomIterator
-{
-    public int getColumnCount();
-
-    public void reset();
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
index ffbca1a..ccf9149 100644
--- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java
@@ -74,16 +74,15 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
         {
             Descriptor.Version version = sstable.descriptor.version;
             this.indexes = indexEntry.columnsIndex();
+            emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
             if (indexes.isEmpty())
             {
-                setToRowStart(sstable, indexEntry, input);
-                this.emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
+                setToRowStart(indexEntry, input);
                 emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version));
                 fetcher = new SimpleBlockFetcher();
             }
             else
             {
-                emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
                 emptyColumnFamily.delete(indexEntry.deletionTime());
                 fetcher = new IndexedBlockFetcher(indexEntry.position);
             }
@@ -98,19 +97,20 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
     /**
      * Sets the seek position to the start of the row for column scanning.
      */
-    private void setToRowStart(SSTableReader reader, RowIndexEntry indexEntry, FileDataInput input) throws IOException
+    private void setToRowStart(RowIndexEntry rowEntry, FileDataInput in) throws IOException
     {
-        if (input == null)
+        if (in == null)
         {
-            this.file = sstable.getFileDataInput(indexEntry.position);
+            this.file = sstable.getFileDataInput(rowEntry.position);
         }
         else
         {
-            this.file = input;
-            input.seek(indexEntry.position);
+            this.file = in;
+            in.seek(rowEntry.position);
         }
         sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(file));
-        file.readLong();
+        if (sstable.descriptor.version.hasRowSizeAndColumnCount)
+            file.readLong();
     }
 
     public ColumnFamily getColumnFamily()
@@ -210,7 +210,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
     private class IndexedBlockFetcher extends BlockFetcher
     {
         // where this row starts
-        private final long basePosition;
+        private final long columnsStart;
 
         // the index entry for the next block to deserialize
         private int nextIndexIdx = -1;
@@ -222,10 +222,10 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
         // may still match a slice
         private final Deque<OnDiskAtom> prefetched;
 
-        public IndexedBlockFetcher(long basePosition)
+        public IndexedBlockFetcher(long columnsStart)
         {
             super(-1);
-            this.basePosition = basePosition;
+            this.columnsStart = columnsStart;
             this.prefetched = reversed ? new ArrayDeque<OnDiskAtom>() : null;
             setNextSlice();
         }
@@ -330,7 +330,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
             IndexInfo currentIndex = indexes.get(lastDeserializedBlock);
 
             /* seek to the correct offset to the data, and calculate the data size */
-            long positionToSeek = basePosition + currentIndex.offset;
+            long positionToSeek = columnsStart + currentIndex.offset;
 
             // With new promoted indexes, our first seek in the data file will happen at that point.
             if (file == null)
@@ -420,7 +420,8 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA
             // We remenber when we are whithin a slice to avoid some comparison
             boolean inSlice = false;
 
-            Iterator<OnDiskAtom> atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, file.readInt(), sstable.descriptor.version);
+            int columnCount = sstable.descriptor.version.hasRowSizeAndColumnCount ? file.readInt() : Integer.MAX_VALUE;
+            Iterator<OnDiskAtom> atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, columnCount, sstable.descriptor.version);
             while (atomIterator.hasNext())
             {
                 OnDiskAtom column = atomIterator.next();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
index bc0af30..f0e6dc7 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
@@ -115,7 +115,8 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
 
             DecoratedKey keyInDisk = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(file));
             assert keyInDisk.equals(key) : String.format("%s != %s in %s", keyInDisk, key, file.getPath());
-            file.readLong();
+            if (sstable.descriptor.version.hasRowSizeAndColumnCount)
+                file.readLong();
         }
 
         indexList = indexEntry.columnsIndex();
@@ -142,7 +143,8 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
         List<OnDiskAtom> result = new ArrayList<OnDiskAtom>();
         if (indexList.isEmpty())
         {
-            readSimpleColumns(file, columns, result);
+            int columnCount = sstable.descriptor.version.hasRowSizeAndColumnCount ? file.readInt() : Integer.MAX_VALUE;
+            readSimpleColumns(file, columns, result, columnCount);
         }
         else
         {
@@ -153,9 +155,9 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement
         iter = result.iterator();
     }
 
-    private void readSimpleColumns(FileDataInput file, SortedSet<ByteBuffer> columnNames, List<OnDiskAtom> result) throws IOException
+    private void readSimpleColumns(FileDataInput file, SortedSet<ByteBuffer> columnNames, List<OnDiskAtom> result, int columnCount) throws IOException
     {
-        Iterator<OnDiskAtom> atomIterator = cf.metadata().getOnDiskIterator(file, file.readInt(), sstable.descriptor.version);
+        Iterator<OnDiskAtom> atomIterator = cf.metadata().getOnDiskIterator(file, columnCount, sstable.descriptor.version);
         int n = 0;
         while (atomIterator.hasNext())
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
index efbb92c..d0d74c6 100644
--- a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
+++ b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java
@@ -22,6 +22,8 @@ import java.nio.ByteBuffer;
 import java.util.Iterator;
 
 import com.google.common.collect.AbstractIterator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.marshal.AbstractType;
@@ -30,21 +32,22 @@ import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.IndexHelper;
 import org.apache.cassandra.io.sstable.SSTableReader;
 import org.apache.cassandra.io.util.FileDataInput;
-import org.apache.cassandra.io.util.FileMark;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAtomIterator
 {
+    private static final Logger logger = LoggerFactory.getLogger(SimpleSliceReader.class);
+
     private final FileDataInput file;
     private final boolean needsClosing;
     private final ByteBuffer finishColumn;
     private final AbstractType<?> comparator;
     private final ColumnFamily emptyColumnFamily;
-    private FileMark mark;
     private final Iterator<OnDiskAtom> atomIterator;
 
     public SimpleSliceReader(SSTableReader sstable, RowIndexEntry indexEntry, FileDataInput input, ByteBuffer finishColumn)
     {
+        logger.debug("Slicing {}", sstable);
         this.finishColumn = finishColumn;
         this.comparator = sstable.metadata.comparator;
         try
@@ -61,16 +64,17 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt
                 this.needsClosing = false;
             }
 
+            Descriptor.Version version = sstable.descriptor.version;
+
             // Skip key and data size
             ByteBufferUtil.skipShortLength(file);
-            file.readLong();
-
-            Descriptor.Version version = sstable.descriptor.version;
+            if (version.hasRowSizeAndColumnCount)
+                file.readLong();
 
             emptyColumnFamily = EmptyColumns.factory.create(sstable.metadata);
-            emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version));
-            atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, file.readInt(), version);
-            mark = file.mark();
+            emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, sstable.descriptor.version));
+            int columnCount = version.hasRowSizeAndColumnCount ? file.readInt() : Integer.MAX_VALUE;
+            atomIterator = emptyColumnFamily.metadata().getOnDiskIterator(file, columnCount, sstable.descriptor.version);
         }
         catch (IOException e)
         {
@@ -84,20 +88,10 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt
         if (!atomIterator.hasNext())
             return endOfData();
 
-        OnDiskAtom column;
-        try
-        {
-            file.reset(mark);
-            column = atomIterator.next();
-        }
-        catch (IOException e)
-        {
-            throw new CorruptSSTableException(e, file.getPath());
-        }
+        OnDiskAtom column = atomIterator.next();
         if (finishColumn.remaining() > 0 && comparator.compare(column.name(), finishColumn) > 0)
             return endOfData();
 
-        mark = file.mark();
         return column;
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
index 56e9353..cd2952a 100644
--- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactedRow.java
@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.security.MessageDigest;
 
 import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.RowIndexEntry;
 import org.apache.cassandra.io.sstable.ColumnStats;
 import org.apache.cassandra.db.DeletionInfo;
 import org.apache.cassandra.db.ColumnIndex;
@@ -48,7 +49,7 @@ public abstract class AbstractCompactedRow implements Closeable
      *
      * write() may change internal state; it is NOT valid to call write() or update() a second time.
      */
-    public abstract long write(DataOutput out) throws IOException;
+    public abstract RowIndexEntry write(long currentPosition, DataOutput out) throws IOException;
 
     /**
      * update @param digest with the data bytes of the row (not including row key or row size).
@@ -59,23 +60,8 @@ public abstract class AbstractCompactedRow implements Closeable
     public abstract void update(MessageDigest digest);
 
     /**
-     * @return true if there are no columns in the row AND there are no row-level tombstones to be preserved
-     */
-    public abstract boolean isEmpty();
-
-    /**
      * @return aggregate information about the columns in this row.  Some fields may
      * contain default values if computing them value would require extra effort we're not willing to make.
      */
     public abstract ColumnStats columnStats();
-
-    /**
-     * @return the compacted row deletion infos.
-     */
-    public abstract DeletionInfo deletionInfo();
-
-    /**
-     * @return the column index for this row.
-     */
-    public abstract ColumnIndex index();
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
index 877d349..758068c 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionManager.java
@@ -482,9 +482,6 @@ public class CompactionManager implements CompactionManagerMBean
             if (logger.isDebugEnabled())
                 logger.debug("Expected bloom filter size : " + expectedBloomFilterSize);
 
-            SSTableWriter writer = null;
-            SSTableReader newSstable = null;
-
             logger.info("Cleaning up " + sstable);
             // Calculate the expected compacted filesize
             long expectedRangeFileSize = cfs.getExpectedCompactedFileSize(Arrays.asList(sstable), OperationType.CLEANUP);
@@ -498,6 +495,11 @@ public class CompactionManager implements CompactionManagerMBean
 
             CleanupInfo ci = new CleanupInfo(sstable, scanner);
             metrics.beginCompaction(ci);
+            SSTableWriter writer = createWriter(cfs,
+                                                compactionFileLocation,
+                                                expectedBloomFilterSize,
+                                                sstable);
+            SSTableReader newSstable = null;
             try
             {
                 while (scanner.hasNext())
@@ -508,11 +510,8 @@ public class CompactionManager implements CompactionManagerMBean
                     if (Range.isInRanges(row.getKey().token, ranges))
                     {
                         AbstractCompactedRow compactedRow = controller.getCompactedRow(row);
-                        if (compactedRow.isEmpty())
-                            continue;
-                        writer = maybeCreateWriter(cfs, OperationType.CLEANUP, compactionFileLocation, expectedBloomFilterSize, writer, sstable);
-                        writer.append(compactedRow);
-                        totalkeysWritten++;
+                        if (writer.append(compactedRow) != null)
+                            totalkeysWritten++;
                     }
                     else
                     {
@@ -553,13 +552,14 @@ public class CompactionManager implements CompactionManagerMBean
                         }
                     }
                 }
-                if (writer != null)
+                if (totalkeysWritten > 0)
                     newSstable = writer.closeAndOpenReader(sstable.maxDataAge);
+                else
+                    writer.abort();
             }
             catch (Throwable e)
             {
-                if (writer != null)
-                    writer.abort();
+                writer.abort();
                 throw Throwables.propagate(e);
             }
             finally
@@ -589,23 +589,17 @@ public class CompactionManager implements CompactionManagerMBean
         }
     }
 
-    public static SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs,
-                                                  OperationType compactionType,
-                                                  File compactionFileLocation,
-                                                  int expectedBloomFilterSize,
-                                                  SSTableWriter writer,
-                                                  SSTableReader sstable)
+    public static SSTableWriter createWriter(ColumnFamilyStore cfs,
+                                             File compactionFileLocation,
+                                             int expectedBloomFilterSize,
+                                             SSTableReader sstable)
     {
-        if (writer == null)
-        {
-            FileUtils.createDirectory(compactionFileLocation);
-            writer = new SSTableWriter(cfs.getTempSSTablePath(compactionFileLocation),
-                                       expectedBloomFilterSize,
-                                       cfs.metadata,
-                                       cfs.partitioner,
-                                       SSTableMetadata.createCollector(Collections.singleton(sstable), sstable.getSSTableLevel()));
-        }
-        return writer;
+        FileUtils.createDirectory(compactionFileLocation);
+        return new SSTableWriter(cfs.getTempSSTablePath(compactionFileLocation),
+                                 expectedBloomFilterSize,
+                                 cfs.metadata,
+                                 cfs.partitioner,
+                                 SSTableMetadata.createCollector(Collections.singleton(sstable), sstable.getSSTableLevel()));
     }
 
     /**
@@ -661,10 +655,7 @@ public class CompactionManager implements CompactionManagerMBean
                 if (ci.isStopRequested())
                     throw new CompactionInterruptedException(ci.getCompactionInfo());
                 AbstractCompactedRow row = iter.next();
-                if (row.isEmpty())
-                    row.close();
-                else
-                    validator.add(row);
+                validator.add(row);
             }
             validator.complete();
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
index 3b4fdc8..e32956b 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java
@@ -156,7 +156,8 @@ public class CompactionTask extends AbstractCompactionTask
                     throw new CompactionInterruptedException(ci.getCompactionInfo());
 
                 AbstractCompactedRow row = iter.next();
-                if (row.isEmpty())
+                RowIndexEntry indexEntry = writer.append(row);
+                if (indexEntry == null)
                 {
                     controller.invalidateCachedRow(row.key);
                     row.close();
@@ -166,8 +167,6 @@ public class CompactionTask extends AbstractCompactionTask
                 // If the row is cached, we call removeDeleted on at read time it to have coherent query returns,
                 // but if the row is not pushed out of the cache, obsolete tombstones will persist indefinitely.
                 controller.removeDeletedInCache(row.key);
-
-                RowIndexEntry indexEntry = writer.append(row);
                 totalkeysWritten++;
 
                 if (DatabaseDescriptor.getPreheatKeyCache())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
index b799f0d..444ee11 100644
--- a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java
@@ -29,13 +29,14 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.columniterator.ICountableColumnIterator;
 import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.db.index.SecondaryIndexManager;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.io.sstable.ColumnStats;
 import org.apache.cassandra.io.sstable.SSTable;
+import org.apache.cassandra.io.sstable.SSTableWriter;
 import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.MergeIterator;
 import org.apache.cassandra.utils.StreamingHistogram;
 
@@ -45,35 +46,30 @@ import org.apache.cassandra.utils.StreamingHistogram;
  * of the rows being compacted, and merging them as it does so.  So the most we have
  * in memory at a time is the bloom filter, the index, and one column from each
  * pre-compaction row.
- *
- * When write() or update() is called, a second pass is made over the pre-compaction
- * rows to write the merged columns or update the hash, again with at most one column
- * from each row deserialized at a time.
  */
 public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable<OnDiskAtom>
 {
     private static Logger logger = LoggerFactory.getLogger(LazilyCompactedRow.class);
 
-    private final List<? extends ICountableColumnIterator> rows;
+    private final List<? extends OnDiskAtomIterator> rows;
     private final CompactionController controller;
     private final boolean shouldPurge;
     private ColumnFamily emptyColumnFamily;
     private Reducer reducer;
-    private final ColumnStats columnStats;
-    private long columnSerializedSize;
+    private ColumnStats columnStats;
     private boolean closed;
     private ColumnIndex.Builder indexBuilder;
-    private ColumnIndex columnsIndex;
     private final SecondaryIndexManager.Updater indexer;
+    private long maxDelTimestamp;
 
-    public LazilyCompactedRow(CompactionController controller, List<? extends ICountableColumnIterator> rows)
+    public LazilyCompactedRow(CompactionController controller, List<? extends OnDiskAtomIterator> rows)
     {
         super(rows.get(0).getKey());
         this.rows = rows;
         this.controller = controller;
         indexer = controller.cfs.indexManager.updaterFor(key);
 
-        long maxDelTimestamp = Long.MIN_VALUE;
+        maxDelTimestamp = Long.MIN_VALUE;
         for (OnDiskAtomIterator row : rows)
         {
             ColumnFamily cf = row.getColumnFamily();
@@ -85,58 +81,45 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
                 emptyColumnFamily.delete(cf);
         }
         this.shouldPurge = controller.shouldPurge(key, maxDelTimestamp);
+    }
+
+    public RowIndexEntry write(long currentPosition, DataOutput out) throws IOException
+    {
+        assert !closed;
 
+        ColumnIndex columnsIndex;
         try
         {
-            indexAndWrite(null);
+            indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.key, out);
+            columnsIndex = indexBuilder.build(this);
+            if (columnsIndex.columnsIndex.isEmpty())
+            {
+                boolean cfIrrelevant = shouldPurge
+                                       ? ColumnFamilyStore.removeDeletedCF(emptyColumnFamily, controller.gcBefore) == null
+                                       : !emptyColumnFamily.isMarkedForDelete(); // tombstones are relevant
+                if (cfIrrelevant)
+                    return null;
+            }
         }
         catch (IOException e)
         {
             throw new RuntimeException(e);
         }
-        // reach into the reducer used during iteration to get column count, size, max column timestamp
+        // reach into the reducer (created during iteration) to get column count, size, max column timestamp
         // (however, if there are zero columns, iterator() will not be called by ColumnIndexer and reducer will be null)
-        columnStats = new ColumnStats(reducer == null ? 0 : reducer.columns, 
-                                      reducer == null ? Long.MAX_VALUE : reducer.minTimestampSeen, 
+        columnStats = new ColumnStats(reducer == null ? 0 : reducer.columns,
+                                      reducer == null ? Long.MAX_VALUE : reducer.minTimestampSeen,
                                       reducer == null ? maxDelTimestamp : Math.max(maxDelTimestamp, reducer.maxTimestampSeen),
                                       reducer == null ? Integer.MIN_VALUE : reducer.maxLocalDeletionTimeSeen,
                                       reducer == null ? new StreamingHistogram(SSTable.TOMBSTONE_HISTOGRAM_BIN_SIZE) : reducer.tombstones
         );
-        columnSerializedSize = reducer == null ? 0 : reducer.serializedSize;
         reducer = null;
-    }
-
-    private void indexAndWrite(DataOutput out) throws IOException
-    {
-        this.indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.key, out);
-        this.columnsIndex = indexBuilder.build(this);
-    }
-
-    public long write(DataOutput out) throws IOException
-    {
-        assert !closed;
-
-        DataOutputBuffer clockOut = new DataOutputBuffer();
-        DeletionInfo.serializer().serializeForSSTable(emptyColumnFamily.deletionInfo(), clockOut);
-
-        long dataSize = clockOut.getLength() + columnSerializedSize;
-        if (logger.isDebugEnabled())
-            logger.debug(String.format("clock / column sizes are %s / %s", clockOut.getLength(), columnSerializedSize));
-        assert dataSize > 0;
-        out.writeLong(dataSize);
-        out.write(clockOut.getData(), 0, clockOut.getLength());
-        out.writeInt(indexBuilder.writtenAtomCount());
 
-        // We rebuild the column index uselessly, but we need to do that because range tombstone markers depend
-        // on indexing. If we're able to remove the two-phase compaction, we'll avoid that.
-        indexAndWrite(out);
-
-        long secondPassColumnSize = reducer == null ? 0 : reducer.serializedSize;
-        assert secondPassColumnSize == columnSerializedSize
-               : "originally calculated column size of " + columnSerializedSize + " but now it is " + secondPassColumnSize;
+        out.writeShort(SSTableWriter.END_OF_ROW);
 
         close();
-        return dataSize;
+
+        return RowIndexEntry.create(currentPosition, emptyColumnFamily.deletionInfo().getTopLevelDeletion(), columnsIndex);
     }
 
     public void update(MessageDigest digest)
@@ -150,7 +133,6 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
         try
         {
             DeletionInfo.serializer().serializeForSSTable(emptyColumnFamily.deletionInfo(), out);
-            out.writeInt(columnStats.columnCount);
             digest.update(out.getData(), 0, out.getLength());
         }
         catch (IOException e)
@@ -158,30 +140,14 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
             throw new AssertionError(e);
         }
 
+        // initialize indexBuilder for the benefit of its tombstoneTracker, used by our reducing iterator
+        indexBuilder = new ColumnIndex.Builder(emptyColumnFamily, key.key, out);
         Iterator<OnDiskAtom> iter = iterator();
         while (iter.hasNext())
-        {
             iter.next().updateDigest(digest);
-        }
         close();
     }
 
-    public boolean isEmpty()
-    {
-        boolean cfIrrelevant = shouldPurge
-                             ? ColumnFamilyStore.removeDeletedCF(emptyColumnFamily, controller.gcBefore) == null
-                             : !emptyColumnFamily.isMarkedForDelete(); // tombstones are relevant
-        return cfIrrelevant && columnStats.columnCount == 0;
-    }
-
-    public int getEstimatedColumnCount()
-    {
-        int n = 0;
-        for (ICountableColumnIterator row : rows)
-            n += row.getColumnCount();
-        return n;
-    }
-
     public AbstractType<?> getComparator()
     {
         return emptyColumnFamily.getComparator();
@@ -189,8 +155,6 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
 
     public Iterator<OnDiskAtom> iterator()
     {
-        for (ICountableColumnIterator row : rows)
-            row.reset();
         reducer = new Reducer();
         Iterator<OnDiskAtom> iter = MergeIterator.get(rows, getComparator().onDiskAtomComparator, reducer);
         return Iterators.filter(iter, Predicates.notNull());
@@ -217,19 +181,6 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
         closed = true;
     }
 
-    public DeletionInfo deletionInfo()
-    {
-        return emptyColumnFamily.deletionInfo();
-    }
-
-    /**
-     * @return the column index for this row.
-     */
-    public ColumnIndex index()
-    {
-        return columnsIndex;
-    }
-
     private class Reducer extends MergeIterator.Reducer<OnDiskAtom, OnDiskAtom>
     {
         // all columns reduced together will have the same name, so there will only be one column

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
index 718b1e3..c0bce30 100644
--- a/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
+++ b/src/java/org/apache/cassandra/db/compaction/ParallelCompactionIterable.java
@@ -33,7 +33,7 @@ import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
 import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.columniterator.ICountableColumnIterator;
+import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
 import org.apache.cassandra.utils.*;
 
@@ -164,7 +164,7 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
                 return new CompactedRowContainer(rows.get(0).getKey(), executor.submit(new MergeTask(rawRows)));
             }
 
-            List<ICountableColumnIterator> iterators = new ArrayList<ICountableColumnIterator>(rows.size());
+            List<OnDiskAtomIterator> iterators = new ArrayList<OnDiskAtomIterator>(rows.size());
             for (RowContainer container : rows)
                 iterators.add(container.row == null ? container.wrapper : new DeserializedColumnIterator(container.row));
             return new CompactedRowContainer(new LazilyCompactedRow(controller, iterators));
@@ -203,7 +203,7 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
             }
         }
 
-        private class DeserializedColumnIterator implements ICountableColumnIterator
+        private class DeserializedColumnIterator implements OnDiskAtomIterator
         {
             private final Row row;
             private Iterator<Column> iter;
@@ -224,16 +224,6 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
                 return row.key;
             }
 
-            public int getColumnCount()
-            {
-                return row.cf.getColumnCount();
-            }
-
-            public void reset()
-            {
-                iter = row.cf.iterator();
-            }
-
             public void close() throws IOException {}
 
             public boolean hasNext()
@@ -319,7 +309,7 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
     /**
      * a wrapper around SSTII that notifies the given condition when it is closed
      */
-    private static class NotifyingSSTableIdentityIterator implements ICountableColumnIterator
+    private static class NotifyingSSTableIdentityIterator implements OnDiskAtomIterator
     {
         private final SSTableIdentityIterator wrapped;
         private final Condition condition;
@@ -340,16 +330,6 @@ public class ParallelCompactionIterable extends AbstractCompactionIterable
             return wrapped.getKey();
         }
 
-        public int getColumnCount()
-        {
-            return wrapped.getColumnCount();
-        }
-
-        public void reset()
-        {
-            wrapped.reset();
-        }
-
         public void close() throws IOException
         {
             wrapped.close();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java b/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
index 1e7285a..1b5d4a9 100644
--- a/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
+++ b/src/java/org/apache/cassandra/db/compaction/PrecompactedRow.java
@@ -31,6 +31,7 @@ import org.apache.cassandra.db.filter.IDiskAtomFilter;
 import org.apache.cassandra.db.index.SecondaryIndexManager;
 import org.apache.cassandra.io.sstable.ColumnStats;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
+import org.apache.cassandra.io.sstable.SSTableWriter;
 import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.utils.CloseableIterator;
 import org.apache.cassandra.utils.FBUtilities;
@@ -42,7 +43,6 @@ import org.apache.cassandra.utils.MergeIterator;
 public class PrecompactedRow extends AbstractCompactedRow
 {
     private final ColumnFamily compactedCf;
-    private ColumnIndex columnIndex;
 
     /** it is caller's responsibility to call removeDeleted + removeOldShards from the cf before calling this constructor */
     public PrecompactedRow(DecoratedKey key, ColumnFamily cf)
@@ -152,21 +152,12 @@ public class PrecompactedRow extends AbstractCompactedRow
         filter.collectReducedColumns(returnCF, reduced, CompactionManager.NO_GC);
     }
 
-    public long write(DataOutput out) throws IOException
+    public RowIndexEntry write(long currentPosition, DataOutput out) throws IOException
     {
-        assert compactedCf != null;
-        DataOutputBuffer buffer = new DataOutputBuffer();
-        ColumnIndex.Builder builder = new ColumnIndex.Builder(compactedCf, key.key, buffer);
-        columnIndex = builder.build(compactedCf);
-
-        TypeSizes typeSizes = TypeSizes.NATIVE;
-        long delSize = DeletionTime.serializer.serializedSize(compactedCf.deletionInfo().getTopLevelDeletion(), typeSizes);
-        long dataSize = buffer.getLength() + delSize + typeSizes.sizeof(0);
-        out.writeLong(dataSize);
-        DeletionInfo.serializer().serializeForSSTable(compactedCf.deletionInfo(), out);
-        out.writeInt(builder.writtenAtomCount());
-        out.write(buffer.getData(), 0, buffer.getLength());
-        return dataSize;
+        if (compactedCf == null)
+            return null;
+
+        return SSTableWriter.rawAppend(compactedCf, currentPosition, key, out);
     }
 
     public void update(MessageDigest digest)
@@ -176,7 +167,6 @@ public class PrecompactedRow extends AbstractCompactedRow
         try
         {
             DeletionInfo.serializer().serializeForSSTable(compactedCf.deletionInfo(), buffer);
-            buffer.writeInt(compactedCf.getColumnCount());
             digest.update(buffer.getData(), 0, buffer.getLength());
         }
         catch (IOException e)
@@ -186,11 +176,6 @@ public class PrecompactedRow extends AbstractCompactedRow
         compactedCf.updateDigest(digest);
     }
 
-    public boolean isEmpty()
-    {
-        return compactedCf == null;
-    }
-
     public ColumnStats columnStats()
     {
         return compactedCf.getColumnStats();
@@ -207,18 +192,5 @@ public class PrecompactedRow extends AbstractCompactedRow
         return compactedCf;
     }
 
-    public DeletionInfo deletionInfo()
-    {
-        return compactedCf.deletionInfo();
-    }
-
-    /**
-     * @return the column index for this row.
-     */
-    public ColumnIndex index()
-    {
-        return columnIndex;
-    }
-
     public void close() { }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 421385a..5809741 100644
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
@@ -111,7 +111,7 @@ public class Scrubber implements Closeable
             }
 
             // TODO errors when creating the writer may leave empty temp files.
-            writer = CompactionManager.maybeCreateWriter(cfs, OperationType.SCRUB, destination, expectedBloomFilterSize, null, sstable);
+            writer = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, sstable);
 
             AbstractCompactedRow prevRow = null;
 
@@ -168,25 +168,21 @@ public class Scrubber implements Closeable
                         throw new IOError(new IOException("Unable to read row key from data file"));
                     if (dataSize > dataFile.length())
                         throw new IOError(new IOException("Impossible row size " + dataSize));
-                    SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStart, dataSize, true);
+
+                    SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
                     AbstractCompactedRow compactedRow = controller.getCompactedRow(row);
-                    if (compactedRow.isEmpty())
+                    if (prevRow != null && acrComparator.compare(prevRow, compactedRow) >= 0)
                     {
-                        emptyRows++;
+                        outOfOrderRows.add(compactedRow);
+                        outputHandler.warn(String.format("Out of order row detected (%s found after %s)", compactedRow.key, prevRow.key));
+                        continue;
                     }
-                    else
-                    {
-                        if (prevRow != null && acrComparator.compare(prevRow, compactedRow) >= 0)
-                        {
-                            outOfOrderRows.add(compactedRow);
-                            outputHandler.warn(String.format("Out of order row detected (%s found after %s)", compactedRow.key, prevRow.key));
-                            continue;
-                        }
 
-                        writer.append(compactedRow);
-                        prevRow = compactedRow;
+                    if (writer.append(compactedRow) == null)
+                        emptyRows++;
+                    else
                         goodRows++;
-                    }
+                    prevRow = compactedRow;
                     if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex)
                         outputHandler.warn("Index file contained a different key or row size; using key from data file");
                 }
@@ -204,24 +200,19 @@ public class Scrubber implements Closeable
                         key = sstable.partitioner.decorateKey(currentIndexKey);
                         try
                         {
-                            SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataStartFromIndex, dataSizeFromIndex, true);
+                            SSTableIdentityIterator row = new SSTableIdentityIterator(sstable, dataFile, key, dataSizeFromIndex, true);
                             AbstractCompactedRow compactedRow = controller.getCompactedRow(row);
-                            if (compactedRow.isEmpty())
+                            if (prevRow != null && acrComparator.compare(prevRow, compactedRow) >= 0)
                             {
-                                emptyRows++;
+                                outOfOrderRows.add(compactedRow);
+                                outputHandler.warn(String.format("Out of order row detected (%s found after %s)", compactedRow.key, prevRow.key));
+                                continue;
                             }
+                            if (writer.append(compactedRow) == null)
+                                emptyRows++;
                             else
-                            {
-                                if (prevRow != null && acrComparator.compare(prevRow, compactedRow) >= 0)
-                                {
-                                    outOfOrderRows.add(compactedRow);
-                                    outputHandler.warn(String.format("Out of order row detected (%s found after %s)", compactedRow.key, prevRow.key));
-                                    continue;
-                                }
-                                writer.append(compactedRow);
-                                prevRow = compactedRow;
                                 goodRows++;
-                            }
+                            prevRow = compactedRow;
                         }
                         catch (Throwable th2)
                         {
@@ -266,7 +257,7 @@ public class Scrubber implements Closeable
 
         if (!outOfOrderRows.isEmpty())
         {
-            SSTableWriter inOrderWriter = CompactionManager.maybeCreateWriter(cfs, OperationType.SCRUB, destination, expectedBloomFilterSize, null, sstable);
+            SSTableWriter inOrderWriter = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, sstable);
             for (AbstractCompactedRow row : outOfOrderRows)
                 inOrderWriter.append(row);
             newInOrderSstable = inOrderWriter.closeAndOpenReader(sstable.maxDataAge);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
index e2a7a18..fdc99b9 100644
--- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java
+++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java
@@ -53,6 +53,7 @@ public class Descriptor
         //               into this new format)
         //             tracks max local deletiontime in sstable metadata
         //             records bloom_filter_fp_chance in metadata component
+        //             remove data size and column count from data file (CASSANDRA-4180)
 
         public static final Version CURRENT = new Version(current_version);
 
@@ -63,6 +64,7 @@ public class Descriptor
         public final boolean tracksMaxLocalDeletionTime;
         public final boolean hasBloomFilterFPChance;
         public final boolean offHeapSummaries;
+        public final boolean hasRowSizeAndColumnCount;
 
         public Version(String version)
         {
@@ -72,6 +74,7 @@ public class Descriptor
             hasSuperColumns = version.compareTo("ja") < 0;
             hasBloomFilterFPChance = version.compareTo("ja") >= 0;
             offHeapSummaries = version.compareTo("ja") >= 0;
+            hasRowSizeAndColumnCount = version.compareTo("ja") < 0;
         }
 
         /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
deleted file mode 100644
index 4febea6..0000000
--- a/src/java/org/apache/cassandra/io/sstable/SSTableBoundedScanner.java
+++ /dev/null
@@ -1,95 +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.io.sstable;
-
-import java.util.Arrays;
-import java.util.Iterator;
-
-import com.google.common.util.concurrent.RateLimiter;
-
-import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
-import org.apache.cassandra.db.RowPosition;
-import org.apache.cassandra.utils.Pair;
-
-/**
- * A SSTableScanner that only reads key in a given range (for validation compaction).
- */
-public class SSTableBoundedScanner extends SSTableScanner
-{
-    private final Iterator<Pair<Long, Long>> rangeIterator;
-    private Pair<Long, Long> currentRange;
-
-    SSTableBoundedScanner(SSTableReader sstable, Iterator<Pair<Long, Long>> rangeIterator, RateLimiter limiter)
-    {
-        super(sstable, limiter);
-        assert rangeIterator.hasNext(); // use EmptyCompactionScanner otherwise
-        this.rangeIterator = rangeIterator;
-        currentRange = rangeIterator.next();
-        dfile.seek(currentRange.left);
-    }
-
-    /*
-     * This shouldn't be used with a bounded scanner as it could put the
-     * bounded scanner outside it's range.
-     */
-    @Override
-    public void seekTo(RowPosition seekKey)
-    {
-        throw new UnsupportedOperationException();
-    }
-
-    @Override
-    public boolean hasNext()
-    {
-        if (iterator == null)
-            iterator = exhausted ? Arrays.asList(new OnDiskAtomIterator[0]).iterator() : new BoundedKeyScanningIterator();
-        return iterator.hasNext();
-    }
-
-    @Override
-    public OnDiskAtomIterator next()
-    {
-        if (iterator == null)
-            iterator = exhausted ? Arrays.asList(new OnDiskAtomIterator[0]).iterator() : new BoundedKeyScanningIterator();
-        return iterator.next();
-    }
-
-    protected class BoundedKeyScanningIterator extends KeyScanningIterator
-    {
-        @Override
-        public boolean hasNext()
-        {
-            if (!super.hasNext())
-                return false;
-
-            if (finishedAt < currentRange.right)
-                return true;
-
-            if (rangeIterator.hasNext())
-            {
-                currentRange = rangeIterator.next();
-                finishedAt = currentRange.left; // next() will seek for us
-                return true;
-            }
-            else
-            {
-                return false;
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
index 8e636c5..d35e14d 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
@@ -25,30 +25,25 @@ import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.columniterator.ICountableColumnIterator;
+import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.db.marshal.MarshalException;
 import org.apache.cassandra.io.util.RandomAccessReader;
-import org.apache.cassandra.utils.BytesReadTracker;
 
-public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterator>, ICountableColumnIterator
+public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterator>, OnDiskAtomIterator
 {
     private static final Logger logger = LoggerFactory.getLogger(SSTableIdentityIterator.class);
 
     private final DecoratedKey key;
-    private final DataInput input;
-    private final long dataStart;
-    public final long dataSize;
+    private final DataInput in;
+    public final long dataSize; // we [still] require this so compaction can tell if it's safe to read the row into memory
     public final ColumnSerializer.Flag flag;
 
     private final ColumnFamily columnFamily;
     private final int columnCount;
-    private final long columnPosition;
 
     private final Iterator<OnDiskAtom> atomIterator;
     private final Descriptor.Version dataVersion;
 
-    private final BytesReadTracker inputWithTracker; // tracks bytes read
-
     // Used by lazilyCompactedRow, so that we see the same things when deserializing the first and second time
     private final int expireBefore;
 
@@ -60,13 +55,12 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
      * @param sstable SSTable we are reading ffrom.
      * @param file Reading using this file.
      * @param key Key of this row.
-     * @param dataStart Data for this row starts at this pos.
      * @param dataSize length of row data
      * @throws IOException
      */
-    public SSTableIdentityIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key, long dataStart, long dataSize)
+    public SSTableIdentityIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key, long dataSize)
     {
-        this(sstable, file, key, dataStart, dataSize, false);
+        this(sstable, file, key, dataSize, false);
     }
 
     /**
@@ -74,39 +68,29 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
      * @param sstable SSTable we are reading ffrom.
      * @param file Reading using this file.
      * @param key Key of this row.
-     * @param dataStart Data for this row starts at this pos.
      * @param dataSize length of row data
      * @param checkData if true, do its best to deserialize and check the coherence of row data
      */
-    public SSTableIdentityIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key, long dataStart, long dataSize, boolean checkData)
+    public SSTableIdentityIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key, long dataSize, boolean checkData)
     {
-        this(sstable.metadata, file, file.getPath(), key, dataStart, dataSize, checkData, sstable, ColumnSerializer.Flag.LOCAL);
-    }
-
-    // Must only be used against current file format
-    public SSTableIdentityIterator(CFMetaData metadata, DataInput file, String filename, DecoratedKey key, long dataStart, long dataSize, ColumnSerializer.Flag flag)
-    {
-        this(metadata, file, filename, key, dataStart, dataSize, false, null, flag);
+        this(sstable.metadata, file, file.getPath(), key, dataSize, checkData, sstable, ColumnSerializer.Flag.LOCAL);
     }
 
     // sstable may be null *if* checkData is false
     // If it is null, we assume the data is in the current file format
     private SSTableIdentityIterator(CFMetaData metadata,
-                                    DataInput input,
+                                    DataInput in,
                                     String filename,
                                     DecoratedKey key,
-                                    long dataStart,
                                     long dataSize,
                                     boolean checkData,
                                     SSTableReader sstable,
                                     ColumnSerializer.Flag flag)
     {
         assert !checkData || (sstable != null);
-        this.input = input;
+        this.in = in;
         this.filename = filename;
-        this.inputWithTracker = new BytesReadTracker(input);
         this.key = key;
-        this.dataStart = dataStart;
         this.dataSize = dataSize;
         this.expireBefore = (int)(System.currentTimeMillis() / 1000);
         this.flag = flag;
@@ -115,21 +99,10 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
 
         try
         {
-            if (input instanceof RandomAccessReader)
-            {
-                RandomAccessReader file = (RandomAccessReader) input;
-                file.seek(this.dataStart);
-                if (dataStart + dataSize > file.length())
-                    throw new IOException(String.format("dataSize of %s starting at %s would be larger than file %s length %s",
-                                          dataSize, dataStart, file.getPath(), file.length()));
-            }
-
             columnFamily = EmptyColumns.factory.create(metadata);
-            columnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(inputWithTracker, dataVersion));
-
-            columnCount = inputWithTracker.readInt();
-            atomIterator = columnFamily.metadata().getOnDiskIterator(inputWithTracker, columnCount, dataVersion);
-            columnPosition = dataStart + inputWithTracker.getBytesRead();
+            columnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(in, dataVersion));
+            columnCount = dataVersion.hasRowSizeAndColumnCount ? in.readInt() : Integer.MAX_VALUE;
+            atomIterator = columnFamily.metadata().getOnDiskIterator(in, columnCount, dataVersion);
         }
         catch (IOException e)
         {
@@ -151,7 +124,18 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
 
     public boolean hasNext()
     {
-        return inputWithTracker.getBytesRead() < dataSize;
+        try
+        {
+            return atomIterator.hasNext();
+        }
+        catch (IOError e)
+        {
+            // catch here b/c atomIterator is an AbstractIterator; hasNext reads the value
+            if (e.getCause() instanceof IOException)
+                throw new CorruptSSTableException((IOException)e.getCause(), filename);
+            else
+                throw e;
+        }
     }
 
     public OnDiskAtom next()
@@ -163,13 +147,6 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
                 atom.validateFields(columnFamily.metadata());
             return atom;
         }
-        catch (IOError e)
-        {
-            if (e.getCause() instanceof IOException)
-                throw new CorruptSSTableException((IOException)e.getCause(), filename);
-            else
-                throw e;
-        }
         catch (MarshalException me)
         {
             throw new CorruptSSTableException(me, filename);
@@ -189,9 +166,9 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
     public String getPath()
     {
         // if input is from file, then return that path, otherwise it's from streaming
-        if (input instanceof RandomAccessReader)
+        if (in instanceof RandomAccessReader)
         {
-            RandomAccessReader file = (RandomAccessReader) input;
+            RandomAccessReader file = (RandomAccessReader) in;
             return file.getPath();
         }
         else
@@ -202,10 +179,9 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
 
     public ColumnFamily getColumnFamilyWithColumns(ColumnFamily.Factory containerFactory)
     {
-        assert inputWithTracker.getBytesRead() == headerSize();
         ColumnFamily cf = columnFamily.cloneMeShallow(containerFactory, false);
         // since we already read column count, just pass that value and continue deserialization
-        columnFamily.serializer.deserializeColumnsFromSSTable(inputWithTracker, cf, columnCount, flag, expireBefore, dataVersion);
+        columnFamily.serializer.deserializeColumnsFromSSTable(in, cf, columnCount, flag, expireBefore, dataVersion);
         if (validateColumns)
         {
             try
@@ -220,28 +196,9 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
         return cf;
     }
 
-    private long headerSize()
-    {
-        return columnPosition - dataStart;
-    }
-
     public int compareTo(SSTableIdentityIterator o)
     {
         return key.compareTo(o.key);
     }
 
-    public void reset()
-    {
-        if (!(input instanceof RandomAccessReader))
-            throw new UnsupportedOperationException();
-
-        RandomAccessReader file = (RandomAccessReader) input;
-        file.seek(columnPosition);
-        inputWithTracker.reset(headerSize());
-    }
-
-    public int getColumnCount()
-    {
-        return columnCount;
-    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
index a09f65b..baba472 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
@@ -1000,10 +1000,15 @@ public class SSTableReader extends SSTable
      */
     public SSTableScanner getScanner(QueryFilter filter)
     {
-        return new SSTableScanner(this, filter);
+        return new SSTableScanner(this, filter, null);
     }
 
-   /**
+    public SSTableScanner getScanner(QueryFilter filter, RowPosition startWith)
+    {
+        return new SSTableScanner(this, filter, startWith, null);
+    }
+
+    /**
     * I/O SSTableScanner
     * @return A Scanner for seeking over the rows of the SSTable.
     */
@@ -1014,7 +1019,7 @@ public class SSTableReader extends SSTable
 
    public SSTableScanner getScanner(RateLimiter limiter)
    {
-       return new SSTableScanner(this, limiter);
+       return new SSTableScanner(this, null, limiter);
    }
 
    /**
@@ -1030,7 +1035,7 @@ public class SSTableReader extends SSTable
 
         Iterator<Pair<Long, Long>> rangeIterator = getPositionsForRanges(Collections.singletonList(range)).iterator();
         if (rangeIterator.hasNext())
-            return new SSTableBoundedScanner(this, rangeIterator, limiter);
+            return new SSTableScanner(this, null, range, limiter);
         else
             return new EmptyCompactionScanner(getFilename());
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/a9bd531b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
index 38b9e65..fb52a02 100644
--- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
+++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java
@@ -18,11 +18,12 @@
 package org.apache.cassandra.io.sstable;
 
 import java.io.IOException;
-import java.util.Arrays;
 import java.util.Iterator;
 
 import com.google.common.util.concurrent.RateLimiter;
 
+import com.google.common.collect.AbstractIterator;
+
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.RowIndexEntry;
 import org.apache.cassandra.db.RowPosition;
@@ -31,6 +32,8 @@ import org.apache.cassandra.db.columniterator.LazyColumnIterator;
 import org.apache.cassandra.db.columniterator.OnDiskAtomIterator;
 import org.apache.cassandra.db.compaction.ICompactionScanner;
 import org.apache.cassandra.db.filter.QueryFilter;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -40,77 +43,83 @@ public class SSTableScanner implements ICompactionScanner
     protected final RandomAccessReader dfile;
     protected final RandomAccessReader ifile;
     public final SSTableReader sstable;
-    private OnDiskAtomIterator row;
-    protected boolean exhausted = false;
     protected Iterator<OnDiskAtomIterator> iterator;
     private final QueryFilter filter;
+    private long stopAt;
 
     /**
-     * @param sstable SSTable to scan.
-     * @param limiter
+     * @param sstable SSTable to scan; must not be null
+     * @param filter filter to use when scanning the columns; may be null
+     * @param limiter background i/o RateLimiter; may be null
      */
-    SSTableScanner(SSTableReader sstable, RateLimiter limiter)
+    SSTableScanner(SSTableReader sstable, QueryFilter filter, RateLimiter limiter)
     {
-        this.dfile = limiter == null ? sstable.openDataReader() : sstable.openDataReader(limiter);
-        this.ifile = sstable.openIndexReader();
-        this.sstable = sstable;
-        this.filter = null;
-    }
+        assert sstable != null;
 
-    /**
-     * @param sstable SSTable to scan.
-     * @param filter filter to use when scanning the columns
-     */
-    SSTableScanner(SSTableReader sstable, QueryFilter filter)
-    {
-        this.dfile = sstable.openDataReader();
+        this.dfile = limiter == null ? sstable.openDataReader() : sstable.openDataReader(limiter);
         this.ifile = sstable.openIndexReader();
         this.sstable = sstable;
         this.filter = filter;
+        stopAt = dfile.length();
     }
 
-    public void close() throws IOException
+    public SSTableScanner(SSTableReader sstable, QueryFilter filter, RowPosition startWith, RateLimiter limiter)
     {
-        FileUtils.close(dfile, ifile);
-    }
+        this(sstable, filter, limiter);
+
+        long indexPosition = sstable.getIndexScanPosition(startWith);
+        // -1 means the key is before everything in the sstable. So just start from the beginning.
+        if (indexPosition == -1)
+            indexPosition = 0;
+        ifile.seek(indexPosition);
 
-    public void seekTo(RowPosition seekKey)
-    {
         try
         {
-            long indexPosition = sstable.getIndexScanPosition(seekKey);
-            // -1 means the key is before everything in the sstable. So just start from the beginning.
-            if (indexPosition == -1)
-                indexPosition = 0;
-
-            ifile.seek(indexPosition);
-
             while (!ifile.isEOF())
             {
-                long startPosition = ifile.getFilePointer();
+                indexPosition = ifile.getFilePointer();
                 DecoratedKey indexDecoratedKey = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(ifile));
-                int comparison = indexDecoratedKey.compareTo(seekKey);
+                int comparison = indexDecoratedKey.compareTo(startWith);
                 if (comparison >= 0)
                 {
                     // Found, just read the dataPosition and seek into index and data files
                     long dataPosition = ifile.readLong();
-                    ifile.seek(startPosition);
+                    ifile.seek(indexPosition);
                     dfile.seek(dataPosition);
-                    row = null;
-                    return;
+                    break;
                 }
                 else
                 {
                     RowIndexEntry.serializer.skip(ifile);
                 }
             }
-            exhausted = true;
         }
         catch (IOException e)
         {
             sstable.markSuspect();
-            throw new CorruptSSTableException(e, ifile.getPath());
+            throw new CorruptSSTableException(e, sstable.getFilename());
         }
+
+    }
+
+    public SSTableScanner(SSTableReader sstable, QueryFilter filter, Range<Token> range, RateLimiter limiter)
+    {
+        this(sstable, filter, range.toRowBounds().left, limiter);
+
+        if (range.isWrapAround())
+        {
+            stopAt = dfile.length();
+        }
+        else
+        {
+            RowIndexEntry position = sstable.getPosition(range.toRowBounds().right, SSTableReader.Operator.GT);
+            stopAt = position == null ? dfile.length() : position.position;
+        }
+    }
+
+    public void close() throws IOException
+    {
+        FileUtils.close(dfile, ifile);
     }
 
     public long getLengthInBytes()
@@ -131,14 +140,14 @@ public class SSTableScanner implements ICompactionScanner
     public boolean hasNext()
     {
         if (iterator == null)
-            iterator = exhausted ? Arrays.asList(new OnDiskAtomIterator[0]).iterator() : createIterator();
+            iterator = createIterator();
         return iterator.hasNext();
     }
 
     public OnDiskAtomIterator next()
     {
         if (iterator == null)
-            iterator = exhausted ? Arrays.asList(new OnDiskAtomIterator[0]).iterator() : createIterator();
+            iterator = createIterator();
         return iterator.next();
     }
 
@@ -149,76 +158,24 @@ public class SSTableScanner implements ICompactionScanner
 
     private Iterator<OnDiskAtomIterator> createIterator()
     {
-        return filter == null ? new KeyScanningIterator() : new FilteredKeyScanningIterator();
-    }
-
-    protected class KeyScanningIterator implements Iterator<OnDiskAtomIterator>
-    {
-        protected long finishedAt;
-
-        public boolean hasNext()
-        {
-            if (row == null)
-                return !dfile.isEOF();
-            return finishedAt < dfile.length();
-        }
-
-        public OnDiskAtomIterator next()
-        {
-            try
-            {
-                if (row != null)
-                    dfile.seek(finishedAt);
-                assert !dfile.isEOF();
-
-                // Read data header
-                DecoratedKey key = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(dfile));
-                long dataSize = dfile.readLong();
-                long dataStart = dfile.getFilePointer();
-                finishedAt = dataStart + dataSize;
-
-                row = new SSTableIdentityIterator(sstable, dfile, key, dataStart, dataSize);
-                return row;
-            }
-            catch (IOException e)
-            {
-                sstable.markSuspect();
-                throw new CorruptSSTableException(e, dfile.getPath());
-            }
-        }
-
-        public void remove()
-        {
-            throw new UnsupportedOperationException();
-        }
-
-        @Override
-        public String toString()
-        {
-            return getClass().getSimpleName() + "(" + "finishedAt:" + finishedAt + ")";
-        }
+        return new KeyScanningIterator();
     }
 
-    protected class FilteredKeyScanningIterator implements Iterator<OnDiskAtomIterator>
+    protected class KeyScanningIterator extends AbstractIterator<OnDiskAtomIterator>
     {
-        protected DecoratedKey nextKey;
-        protected RowIndexEntry nextEntry;
-
-        public boolean hasNext()
-        {
-            if (row == null)
-                return !ifile.isEOF();
-            return nextKey != null;
-        }
+        private DecoratedKey nextKey;
+        private RowIndexEntry nextEntry;
+        private DecoratedKey currentKey;
+        private RowIndexEntry currentEntry;
 
-        public OnDiskAtomIterator next()
+        protected OnDiskAtomIterator computeNext()
         {
             try
             {
-                final DecoratedKey currentKey;
-                final RowIndexEntry currentEntry;
+                if (ifile.isEOF() && nextKey == null)
+                    return endOfData();
 
-                if (row == null)
+                if (currentKey == null)
                 {
                     currentKey = sstable.partitioner.decorateKey(ByteBufferUtil.readWithShortLength(ifile));
                     currentEntry = RowIndexEntry.serializer.deserialize(ifile, sstable.descriptor.version);
@@ -229,6 +186,10 @@ public class SSTableScanner implements ICompactionScanner
                     currentEntry = nextEntry;
                 }
 
+                assert currentEntry.position <= stopAt;
+                if (currentEntry.position == stopAt)
+                    return endOfData();
+
                 if (ifile.isEOF())
                 {
                     nextKey = null;
@@ -241,7 +202,18 @@ public class SSTableScanner implements ICompactionScanner
                 }
 
                 assert !dfile.isEOF();
-                return row = new LazyColumnIterator(currentKey, new IColumnIteratorFactory()
+
+                if (filter == null)
+                {
+                    dfile.seek(currentEntry.position);
+                    ByteBufferUtil.readWithShortLength(dfile); // key
+                    if (sstable.descriptor.version.hasRowSizeAndColumnCount)
+                        dfile.readLong();
+                    long dataSize = (nextEntry == null ? dfile.length() : nextEntry.position) - dfile.getFilePointer();
+                    return new SSTableIdentityIterator(sstable, dfile, currentKey, dataSize);
+                }
+
+                return new LazyColumnIterator(currentKey, new IColumnIteratorFactory()
                 {
                     public OnDiskAtomIterator create()
                     {
@@ -252,14 +224,9 @@ public class SSTableScanner implements ICompactionScanner
             catch (IOException e)
             {
                 sstable.markSuspect();
-                throw new CorruptSSTableException(e, ifile.getPath());
+                throw new CorruptSSTableException(e, sstable.getFilename());
             }
         }
-
-        public void remove()
-        {
-            throw new UnsupportedOperationException();
-        }
     }
 
     @Override
@@ -269,7 +236,6 @@ public class SSTableScanner implements ICompactionScanner
                "dfile=" + dfile +
                " ifile=" + ifile +
                " sstable=" + sstable +
-               " exhausted=" + exhausted +
                ")";
     }
 }


Mime
View raw message