Merge branch 'cassandra-3.0' into cassandra-3.5

CHANGES.txt

@@ -1,5 +1,6 @@
 3.5
 Merged from 3.0:
+ * Support streaming pre-3.0 sstables (CASSANDRA-10990)
  * Add backpressure to compressed or encrypted commit log (CASSANDRA-10971)
  * SSTableExport supports secondary index tables (CASSANDRA-11330)
  * Fix sstabledump to include missing info in debug output (CASSANDRA-11321)

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

@@ -588,6 +588,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
          // clear ephemeral snapshots that were not properly cleared last session (CASSANDRA-7357)
         clearEphemeralSnapshots(directories);
 
+        directories.removeTemporaryDirectories();
+
         logger.trace("Removing temporary or obsoleted files from unfinished operations for table {}", metadata.cfName);
         LifecycleTransaction.removeUnfinishedLeftovers(metadata);
 

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

@@ -95,9 +95,11 @@ public class Directories
 
     public static final String BACKUPS_SUBDIR = "backups";
     public static final String SNAPSHOT_SUBDIR = "snapshots";
+    public static final String TMP_SUBDIR = "tmp";
     public static final String SECONDARY_INDEX_NAME_SEPARATOR = ".";
 
     public static final DataDirectory[] dataDirectories;
+
     static
     {
         String[] locations = DatabaseDescriptor.getAllDataFileLocations();
@@ -335,6 +337,34 @@ public class Directories
         return getLocationForDisk(getWriteableLocation(writeSize));
     }
 
+    /**
+     * Returns a temporary subdirectory on non-blacklisted data directory
+     * that _currently_ has {@code writeSize} bytes as usable space.
+     * This method does not create the temporary directory.
+     *
+     * @throws IOError if all directories are blacklisted.
+     */
+    public File getTemporaryWriteableDirectoryAsFile(long writeSize)
+    {
+        File location = getLocationForDisk(getWriteableLocation(writeSize));
+        if (location == null)
+            return null;
+        return new File(location, TMP_SUBDIR);
+    }
+
+    public void removeTemporaryDirectories()
+    {
+        for (File dataDir : dataPaths)
+        {
+            File tmpDir = new File(dataDir, TMP_SUBDIR);
+            if (tmpDir.exists())
+            {
+                logger.debug("Removing temporary directory {}", tmpDir);
+                FileUtils.deleteRecursive(tmpDir);
+            }
+        }
+    }
+
     /**
      * Returns a non-blacklisted data directory that _currently_ has {@code writeSize} bytes as usable space.
      *

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

@@ -70,6 +70,11 @@ public class SerializationHeader
         this.typeMap = typeMap;
     }
 
+    public static SerializationHeader makeWithoutStats(CFMetaData metadata)
+    {
+        return new SerializationHeader(true, metadata, metadata.partitionColumns(), EncodingStats.NO_STATS);
+    }
+
     public static SerializationHeader forKeyCache(CFMetaData metadata)
     {
         // We don't save type information in the key cache (we could change

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

@@ -46,62 +46,77 @@ public class Serializers
     // unecessary (since IndexInfo.Serializer won't depend on the metadata either).
     public ISerializer<ClusteringPrefix> indexEntryClusteringPrefixSerializer(final Version version, final SerializationHeader header)
     {
-        if (!version.storeRows())
+        if (!version.storeRows() || header ==  null) //null header indicates streaming from pre-3.0 sstables
         {
-            return new ISerializer<ClusteringPrefix>()
-            {
-                public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
-                {
-                    // We should only use this for reading old sstable, never write new ones.
-                    throw new UnsupportedOperationException();
-                }
-
-                public ClusteringPrefix deserialize(DataInputPlus in) throws IOException
-                {
-                    // We're reading the old cellname/composite
-                    ByteBuffer bb = ByteBufferUtil.readWithShortLength(in);
-                    assert bb.hasRemaining(); // empty cellnames were invalid
-
-                    int clusteringSize = metadata.clusteringColumns().size();
-                    // If the table has no clustering column, then the cellname will just be the "column" name, which we ignore here.
-                    if (clusteringSize == 0)
-                        return Clustering.EMPTY;
-
-                    if (!metadata.isCompound())
-                        return Clustering.make(bb);
-
-                    List<ByteBuffer> components = CompositeType.splitName(bb);
-                    byte eoc = CompositeType.lastEOC(bb);
-
-                    if (eoc == 0 || components.size() >= clusteringSize)
-                    {
-                        // That's a clustering.
-                        if (components.size() > clusteringSize)
-                            components = components.subList(0, clusteringSize);
-
-                        return Clustering.make(components.toArray(new ByteBuffer[clusteringSize]));
-                    }
-                    else
-                    {
-                        // It's a range tombstone bound. It is a start since that's the only part we've ever included
-                        // in the index entries.
-                        Slice.Bound.Kind boundKind = eoc > 0
-                                                   ? Slice.Bound.Kind.EXCL_START_BOUND
-                                                   : Slice.Bound.Kind.INCL_START_BOUND;
-
-                        return Slice.Bound.create(boundKind, components.toArray(new ByteBuffer[components.size()]));
-                    }
-                }
-
-                public long serializedSize(ClusteringPrefix clustering)
-                {
-                    // We should only use this for reading old sstable, never write new ones.
-                    throw new UnsupportedOperationException();
-                }
-            };
+            return oldFormatSerializer(version);
         }
 
+        return newFormatSerializer(version, header);
+    }
+
+    private ISerializer<ClusteringPrefix> oldFormatSerializer(final Version version)
+    {
         return new ISerializer<ClusteringPrefix>()
+        {
+            SerializationHeader newHeader = SerializationHeader.makeWithoutStats(metadata);
+
+            public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
+            {
+                //we deserialize in the old format and serialize in the new format
+                ClusteringPrefix.serializer.serialize(clustering, out,
+                                                      version.correspondingMessagingVersion(),
+                                                      newHeader.clusteringTypes());
+            }
+
+            public ClusteringPrefix deserialize(DataInputPlus in) throws IOException
+            {
+                // We're reading the old cellname/composite
+                ByteBuffer bb = ByteBufferUtil.readWithShortLength(in);
+                assert bb.hasRemaining(); // empty cellnames were invalid
+
+                int clusteringSize = metadata.clusteringColumns().size();
+                // If the table has no clustering column, then the cellname will just be the "column" name, which we ignore here.
+                if (clusteringSize == 0)
+                    return Clustering.EMPTY;
+
+                if (!metadata.isCompound())
+                    return Clustering.make(bb);
+
+                List<ByteBuffer> components = CompositeType.splitName(bb);
+                byte eoc = CompositeType.lastEOC(bb);
+
+                if (eoc == 0 || components.size() >= clusteringSize)
+                {
+                    // That's a clustering.
+                    if (components.size() > clusteringSize)
+                        components = components.subList(0, clusteringSize);
+
+                    return Clustering.make(components.toArray(new ByteBuffer[clusteringSize]));
+                }
+                else
+                {
+                    // It's a range tombstone bound. It is a start since that's the only part we've ever included
+                    // in the index entries.
+                    Slice.Bound.Kind boundKind = eoc > 0
+                                                 ? Slice.Bound.Kind.EXCL_START_BOUND
+                                                 : Slice.Bound.Kind.INCL_START_BOUND;
+
+                    return Slice.Bound.create(boundKind, components.toArray(new ByteBuffer[components.size()]));
+                }
+            }
+
+            public long serializedSize(ClusteringPrefix clustering)
+            {
+                return ClusteringPrefix.serializer.serializedSize(clustering, version.correspondingMessagingVersion(),
+                                                                  newHeader.clusteringTypes());
+            }
+        };
+    }
+
+
+    private ISerializer<ClusteringPrefix> newFormatSerializer(final Version version, final SerializationHeader header)
+    {
+        return new ISerializer<ClusteringPrefix>() //Reading and writing from/to the new sstable format
         {
             public void serialize(ClusteringPrefix clustering, DataOutputPlus out) throws IOException
             {
@@ -119,4 +134,5 @@ public class Serializers
             }
         };
     }
-}
+
+}

src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java

@@ -29,7 +29,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.IndexHelper;
 import org.apache.cassandra.io.util.FileDataInput;
-import org.apache.cassandra.io.util.FileMark;
+import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 abstract class AbstractSSTableIterator implements UnfilteredRowIterator
@@ -415,7 +415,7 @@ abstract class AbstractSSTableIterator implements UnfilteredRowIterator
         private int currentIndexIdx;
 
         // Marks the beginning of the block corresponding to currentIndexIdx.
-        private FileMark mark;
+        private DataPosition mark;
 
         public IndexState(Reader reader, ClusteringComparator comparator, RowIndexEntry indexEntry, boolean reversed)
         {

src/java/org/apache/cassandra/db/commitlog/EncryptedFileSegmentInputStream.java

@@ -3,8 +3,8 @@ package org.apache.cassandra.db.commitlog;
 import java.io.DataInput;
 import java.nio.ByteBuffer;
 
+import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.io.util.FileDataInput;
-import org.apache.cassandra.io.util.FileMark;
 import org.apache.cassandra.io.util.FileSegmentInputStream;
 
 /**
@@ -60,7 +60,7 @@ public class EncryptedFileSegmentInputStream extends FileSegmentInputStream impl
         throw new UnsupportedOperationException();
     }
 
-    public long bytesPastMark(FileMark mark)
+    public long bytesPastMark(DataPosition mark)
     {
         throw new UnsupportedOperationException();
     }

src/java/org/apache/cassandra/hints/ChecksummedDataInput.java

@@ -22,13 +22,9 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.zip.CRC32;
 
-import org.apache.cassandra.io.FSError;
-import org.apache.cassandra.io.FSReadError;
-import org.apache.cassandra.io.compress.ICompressor;
 import org.apache.cassandra.io.util.ChannelProxy;
-import org.apache.cassandra.io.util.FileMark;
+import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.io.util.RandomAccessReader;
-import org.apache.cassandra.schema.CompressionParams;
 
 /**
  * A {@link RandomAccessReader} wrapper that calctulates the CRC in place.
@@ -48,7 +44,7 @@ public class ChecksummedDataInput extends RandomAccessReader.RandomAccessReaderW
     private boolean crcUpdateDisabled;
 
     private long limit;
-    private FileMark limitMark;
+    private DataPosition limitMark;
 
     protected ChecksummedDataInput(Builder builder)
     {

src/java/org/apache/cassandra/hints/HintMessage.java

@@ -33,7 +33,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
-import org.apache.cassandra.utils.BytesReadTracker;
+import org.apache.cassandra.io.util.TrackedDataInputPlus;
 import org.apache.cassandra.utils.UUIDSerializer;
 
 /**
@@ -117,7 +117,7 @@ public final class HintMessage
             UUID hostId = UUIDSerializer.serializer.deserialize(in, version);
 
             long hintSize = in.readUnsignedVInt();
-            BytesReadTracker countingIn = new BytesReadTracker(in);
+            TrackedDataInputPlus countingIn = new TrackedDataInputPlus(in);
             try
             {
                 return new HintMessage(hostId, Hint.serializer.deserialize(countingIn, version));

src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java

@@ -32,7 +32,7 @@ import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.sstable.CorruptSSTableException;
 import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
 import org.apache.cassandra.io.util.DataIntegrityMetadata;
-import org.apache.cassandra.io.util.FileMark;
+import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.SequentialWriter;
 import org.apache.cassandra.schema.CompressionParams;
@@ -153,7 +153,7 @@ public class CompressedSequentialWriter extends SequentialWriter
     }
 
     @Override
-    public FileMark mark()
+    public DataPosition mark()
     {
         if (!buffer.hasRemaining())
             doFlush(0);
@@ -161,7 +161,7 @@ public class CompressedSequentialWriter extends SequentialWriter
     }
 
     @Override
-    public synchronized void resetAndTruncate(FileMark mark)
+    public synchronized void resetAndTruncate(DataPosition mark)
     {
         assert mark instanceof CompressedFileWriterMark;
 
@@ -306,7 +306,7 @@ public class CompressedSequentialWriter extends SequentialWriter
     /**
      * Class to hold a mark to the position of the file
      */
-    protected static class CompressedFileWriterMark implements FileMark
+    protected static class CompressedFileWriterMark implements DataPosition
     {
         // chunk offset in the compressed file
         final long chunkOffset;

src/java/org/apache/cassandra/io/sstable/SSTableSimpleIterator.java

@@ -21,6 +21,7 @@ import java.io.IOException;
 import java.io.IOError;
 import java.util.Iterator;
 
+import org.apache.cassandra.io.util.RewindableDataInput;
 import org.apache.cassandra.utils.AbstractIterator;
 
 import org.apache.cassandra.config.CFMetaData;
@@ -29,7 +30,7 @@ import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.FileDataInput;
-import org.apache.cassandra.io.util.FileMark;
+import org.apache.cassandra.io.util.DataPosition;
 import org.apache.cassandra.net.MessagingService;
 
 /**
@@ -113,11 +114,9 @@ public abstract class SSTableSimpleIterator extends AbstractIterator<Unfiltered>
                 // need to extract them. Which imply 2 passes (one to extract the static, then one for other value).
                 if (metadata.isStaticCompactTable())
                 {
-                    // Because we don't support streaming from old file version, the only case we should get there is for compaction,
-                    // where the DataInput should be a file based one.
-                    assert in instanceof FileDataInput;
-                    FileDataInput file = (FileDataInput)in;
-                    FileMark mark = file.mark();
+                    assert in instanceof RewindableDataInput;
+                    RewindableDataInput file = (RewindableDataInput)in;
+                    DataPosition mark = file.mark();
                     Row staticRow = LegacyLayout.extractStaticColumns(metadata, file, metadata.partitionColumns().statics);
                     file.reset(mark);
 

src/java/org/apache/cassandra/io/sstable/format/RangeAwareSSTableWriter.java

@@ -42,7 +42,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
     private final long estimatedKeys;
     private final long repairedAt;
     private final SSTableFormat.Type format;
-    private final SerializationHeader.Component header;
+    private final SerializationHeader header;
     private final LifecycleTransaction txn;
     private int currentIndex = -1;
     public final ColumnFamilyStore cfs;
@@ -50,7 +50,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
     private final List<SSTableReader> finishedReaders = new ArrayList<>();
     private SSTableMultiWriter currentWriter = null;
 
-    public RangeAwareSSTableWriter(ColumnFamilyStore cfs, long estimatedKeys, long repairedAt, SSTableFormat.Type format, int sstableLevel, long totalSize, LifecycleTransaction txn, SerializationHeader.Component header) throws IOException
+    public RangeAwareSSTableWriter(ColumnFamilyStore cfs, long estimatedKeys, long repairedAt, SSTableFormat.Type format, int sstableLevel, long totalSize, LifecycleTransaction txn, SerializationHeader header) throws IOException
     {
         directories = cfs.getDirectories().getWriteableLocations();
         this.sstableLevel = sstableLevel;
@@ -67,7 +67,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
             if (localDir == null)
                 throw new IOException("Insufficient disk space to store " + totalSize + " bytes");
             Descriptor desc = Descriptor.fromFilename(cfs.getSSTablePath(cfs.getDirectories().getLocationForDisk(localDir), format));
-            currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, sstableLevel, header.toHeader(cfs.metadata), txn);
+            currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, sstableLevel, header, txn);
         }
     }
 
@@ -89,7 +89,7 @@ public class RangeAwareSSTableWriter implements SSTableMultiWriter
                 finishedWriters.add(currentWriter);
 
             Descriptor desc = Descriptor.fromFilename(cfs.getSSTablePath(cfs.getDirectories().getLocationForDisk(directories[currentIndex])), format);
-            currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, sstableLevel, header.toHeader(cfs.metadata), txn);
+            currentWriter = cfs.createSSTableMultiWriter(desc, estimatedKeys, repairedAt, sstableLevel, header, txn);
         }
     }
 

src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java

@@ -428,7 +428,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
                                              System.currentTimeMillis(),
                                              statsMetadata,
                                              OpenReason.NORMAL,
-                                             header.toHeader(metadata));
+                                             header == null? null : header.toHeader(metadata));
 
         // special implementation of load to use non-pooled SegmentedFile builders
         try(SegmentedFile.Builder ibuilder = new BufferedSegmentedFile.Builder();

src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java

@@ -83,7 +83,7 @@ public abstract class SSTableWriter extends SSTable implements Transactional
         this.keyCount = keyCount;
         this.repairedAt = repairedAt;
         this.metadataCollector = metadataCollector;
-        this.header = header;
+        this.header = header != null ? header : SerializationHeader.makeWithoutStats(metadata); //null header indicates streaming from pre-3.0 sstable
         this.rowIndexEntrySerializer = descriptor.version.getSSTableFormat().getIndexSerializer(metadata, descriptor.version, header);
         this.observers = observers == null ? Collections.emptySet() : observers;
     }

src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java

@@ -57,7 +57,7 @@ public class BigTableWriter extends SSTableWriter
     private final SegmentedFile.Builder dbuilder;
     protected final SequentialWriter dataFile;
     private DecoratedKey lastWrittenKey;
-    private FileMark dataMark;
+    private DataPosition dataMark;
 
     public BigTableWriter(Descriptor descriptor, 
                           Long keyCount, 
@@ -372,7 +372,7 @@ public class BigTableWriter extends SSTableWriter
         public final SegmentedFile.Builder builder;
         public final IndexSummaryBuilder summary;
         public final IFilter bf;
-        private FileMark mark;
+        private DataPosition mark;
 
         IndexWriter(long keyCount, final SequentialWriter dataFile)
         {

src/java/org/apache/cassandra/io/util/BytesReadTracker.java

@@ -0,0 +1,30 @@
+/*
+ * 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.util;
+
+public interface BytesReadTracker
+{
+    public long getBytesRead();
+
+    /**
+     * reset counter to @param count
+     */
+    public void reset(long count);
+
+}

src/java/org/apache/cassandra/io/util/DataPosition.java

@@ -17,4 +17,5 @@
  */
 package org.apache.cassandra.io.util;
 
-public interface FileMark {}
+public interface DataPosition
+{}

src/java/org/apache/cassandra/io/util/FileDataInput.java

@@ -20,7 +20,7 @@ package org.apache.cassandra.io.util;
 import java.io.Closeable;
 import java.io.IOException;
 
-public interface FileDataInput extends DataInputPlus, Closeable
+public interface FileDataInput extends RewindableDataInput, Closeable
 {
     String getPath();
 
@@ -30,11 +30,5 @@ public interface FileDataInput extends DataInputPlus, Closeable
 
     void seek(long pos) throws IOException;
 
-    FileMark mark();
-
-    void reset(FileMark mark) throws IOException;
-
-    long bytesPastMark(FileMark mark);
-
     long getFilePointer();
 }

src/java/org/apache/cassandra/io/util/FileSegmentInputStream.java

@@ -68,17 +68,23 @@ public class FileSegmentInputStream extends DataInputBuffer implements FileDataI
         buffer.position((int) (pos - offset));
     }
 
-    public FileMark mark()
+    @Override
+    public boolean markSupported()
+    {
+        return false;
+    }
+
+    public DataPosition mark()
     {
         throw new UnsupportedOperationException();
     }
 
-    public void reset(FileMark mark)
+    public void reset(DataPosition mark)
     {
         throw new UnsupportedOperationException();
     }
 
-    public long bytesPastMark(FileMark mark)
+    public long bytesPastMark(DataPosition mark)
     {
         return 0;
     }

src/java/org/apache/cassandra/io/util/RandomAccessReader.java

@@ -200,19 +200,19 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa
         return bytes;
     }
 
-    public FileMark mark()
+    public DataPosition mark()
     {
         markedPointer = current();
         return new BufferedRandomAccessFileMark(markedPointer);
     }
 
-    public void reset(FileMark mark)
+    public void reset(DataPosition mark)
     {
         assert mark instanceof BufferedRandomAccessFileMark;
         seek(((BufferedRandomAccessFileMark) mark).pointer);
     }
 
-    public long bytesPastMark(FileMark mark)
+    public long bytesPastMark(DataPosition mark)
     {
         assert mark instanceof BufferedRandomAccessFileMark;
         long bytes = current() - ((BufferedRandomAccessFileMark) mark).pointer;
@@ -262,7 +262,7 @@ public class RandomAccessReader extends RebufferingInputStream implements FileDa
     /**
      * Class to hold a mark to the position of the file
      */
-    protected static class BufferedRandomAccessFileMark implements FileMark
+    protected static class BufferedRandomAccessFileMark implements DataPosition
     {
         final long pointer;
 

src/java/org/apache/cassandra/io/util/RewindableDataInput.java

@@ -0,0 +1,30 @@
+/*
+ * 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.util;
+
+import java.io.IOException;
+
+public interface RewindableDataInput extends DataInputPlus
+{
+    DataPosition mark();
+
+    void reset(DataPosition mark) throws IOException;
+
+    long bytesPastMark(DataPosition mark);
+}

src/java/org/apache/cassandra/io/util/RewindableDataInputStreamPlus.java

@@ -0,0 +1,569 @@
+/*
+ * 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.util;
+
+import java.io.Closeable;
+import java.io.File;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.apache.cassandra.utils.Throwables.maybeFail;
+import static org.apache.cassandra.utils.Throwables.merge;
+
+/**
+ * Adds mark/reset functionality to another input stream by caching read bytes to a memory buffer and
+ * spilling to disk if necessary.
+ *
+ * When the stream is marked via {@link this#mark()} or {@link this#mark(int)}, up to
+ * <code>maxMemBufferSize</code> will be cached in memory (heap). If more than
+ * <code>maxMemBufferSize</code> bytes are read while the stream is marked, the
+ * following bytes are cached on the <code>spillFile</code> for up to <code>maxDiskBufferSize</code>.
+ *
+ * Please note that successive calls to {@link this#mark()} and {@link this#reset()} will write
+ * sequentially to the same <code>spillFile</code> until <code>maxDiskBufferSize</code> is reached.
+ * At this point, if less than <code>maxDiskBufferSize</code> bytes are currently cached on the
+ * <code>spillFile</code>, the remaining bytes are written to the beginning of the file,
+ * treating the <code>spillFile</code> as a circular buffer.
+ *
+ * If more than <code>maxMemBufferSize + maxDiskBufferSize</code> are cached while the stream is marked,
+ * the following {@link this#reset()} invocation will throw a {@link IllegalStateException}.
+ *
+ */
+public class RewindableDataInputStreamPlus extends FilterInputStream implements RewindableDataInput, Closeable
+{
+    private boolean marked = false;
+    private boolean exhausted = false;
+    private AtomicBoolean closed = new AtomicBoolean(false);
+
+    protected int memAvailable = 0;
+    protected int diskTailAvailable = 0;
+    protected int diskHeadAvailable = 0;
+
+    private final File spillFile;
+    private final int initialMemBufferSize;
+    private final int maxMemBufferSize;
+    private final int maxDiskBufferSize;
+
+    private volatile byte memBuffer[];
+    private int memBufferSize;
+    private RandomAccessFile spillBuffer;
+
+    private final DataInputPlus dataReader;
+
+    public RewindableDataInputStreamPlus(InputStream in, int initialMemBufferSize, int maxMemBufferSize,
+                                         File spillFile, int maxDiskBufferSize)
+    {
+        super(in);
+        dataReader = new DataInputStreamPlus(this);
+        this.initialMemBufferSize = initialMemBufferSize;
+        this.maxMemBufferSize = maxMemBufferSize;
+        this.spillFile = spillFile;
+        this.maxDiskBufferSize = maxDiskBufferSize;
+    }
+
+    /* RewindableDataInput methods */
+
+    /**
+     * Marks the current position of a stream to return to this position later via the {@link this#reset(DataPosition)} method.
+     * @return An empty @link{DataPosition} object
+     */
+    public DataPosition mark()
+    {
+        mark(0);
+        return new RewindableDataInputPlusMark();
+    }
+
+    /**
+     * Rewinds to the previously marked position via the {@link this#mark()} method.
+     * @param mark it's not possible to return to a custom position, so this parameter is ignored.
+     * @throws IOException if an error ocurs while resetting
+     */
+    public void reset(DataPosition mark) throws IOException
+    {
+        reset();
+    }
+
+    public long bytesPastMark(DataPosition mark)
+    {
+        return maxMemBufferSize - memAvailable + (diskTailAvailable == -1? 0 : maxDiskBufferSize - diskHeadAvailable - diskTailAvailable);
+    }
+
+
+    protected static class RewindableDataInputPlusMark implements DataPosition
+    {
+    }
+
+    /* InputStream methods */
+
+    public boolean markSupported()
+    {
+        return true;
+    }
+
+    /**
+     * Marks the current position of a stream to return to this position
+     * later via the {@link this#reset()} method.
+     * @param readlimit the maximum amount of bytes to cache
+     */
+    public synchronized void mark(int readlimit)
+    {
+        if (marked)
+            throw new IllegalStateException("Cannot mark already marked stream.");
+
+        if (memAvailable > 0 || diskHeadAvailable > 0 || diskTailAvailable > 0)
+            throw new IllegalStateException("Can only mark stream after reading previously marked data.");
+
+        marked = true;
+        memAvailable = maxMemBufferSize;
+        diskHeadAvailable = -1;
+        diskTailAvailable = -1;
+    }
+
+    public synchronized void reset() throws IOException
+    {
+        if (!marked)
+            throw new IOException("Must call mark() before calling reset().");
+
+        if (exhausted)
+            throw new IOException(String.format("Read more than capacity: %d bytes.", maxMemBufferSize + maxDiskBufferSize));
+
+        memAvailable = maxMemBufferSize - memAvailable;
+        memBufferSize = memAvailable;
+
+        if (diskTailAvailable == -1)
+        {
+            diskHeadAvailable = 0;
+            diskTailAvailable = 0;
+        }
+        else
+        {
+            int initialPos = diskTailAvailable > 0 ? 0 : (int)getIfNotClosed(spillBuffer).getFilePointer();
+            int diskMarkpos = initialPos + diskHeadAvailable;
+            getIfNotClosed(spillBuffer).seek(diskMarkpos);
+
+            diskHeadAvailable = diskMarkpos - diskHeadAvailable;
+            diskTailAvailable = (maxDiskBufferSize - diskTailAvailable) - diskMarkpos;
+        }
+
+        marked = false;
+    }
+
+    public int available() throws IOException
+    {
+
+        return super.available() + (marked? 0 : memAvailable + diskHeadAvailable + diskTailAvailable);
+    }
+
+    public int read() throws IOException
+    {
+        int read = readOne();
+        if (read == -1)
+            return read;
+
+        if (marked)
+        {
+            //mark exhausted
+            if (isExhausted(1))
+            {
+                exhausted = true;
+                return read;
+            }
+
+            writeOne(read);
+        }
+
+        return read;
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException
+    {
+        int readBytes = readMulti(b, off, len);
+        if (readBytes == -1)
+            return readBytes;
+
+        if (marked)
+        {
+            //check we have space on buffer
+            if (isExhausted(readBytes))
+            {
+                exhausted = true;
+                return readBytes;
+            }
+
+            writeMulti(b, off, readBytes);
+        }
+
+        return readBytes;
+    }
+
+    private void maybeCreateDiskBuffer() throws IOException
+    {
+        if (spillBuffer == null)
+        {
+            if (!spillFile.getParentFile().exists())
+                spillFile.getParentFile().mkdirs();
+            spillFile.createNewFile();
+
+            this.spillBuffer = new RandomAccessFile(spillFile, "rw");
+        }
+    }
+
+
+    private int readOne() throws IOException
+    {
+        if (!marked)
+        {
+            if (memAvailable > 0)
+            {
+                int pos = memBufferSize - memAvailable;
+                memAvailable--;
+                return getIfNotClosed(memBuffer)[pos] & 0xff;
+            }
+
+            if (diskTailAvailable > 0 || diskHeadAvailable > 0)
+            {
+                int read = getIfNotClosed(spillBuffer).read();
+                if (diskTailAvailable > 0)
+                    diskTailAvailable--;
+                else if (diskHeadAvailable > 0)
+                    diskHeadAvailable++;
+                if (diskTailAvailable == 0)
+                    spillBuffer.seek(0);
+                return read;
+            }
+        }
+
+        return getIfNotClosed(in).read();
+    }
+
+    private boolean isExhausted(int readBytes)
+    {
+        return exhausted || readBytes > memAvailable + (long)(diskTailAvailable == -1? maxDiskBufferSize : diskTailAvailable + diskHeadAvailable);
+    }
+
+    private int readMulti(byte[] b, int off, int len) throws IOException
+    {
+        int readBytes = 0;
+        if (!marked)
+        {
+            if (memAvailable > 0)
+            {
+                readBytes += memAvailable < len ? memAvailable : len;
+                int pos = memBufferSize - memAvailable;
+                System.arraycopy(memBuffer, pos, b, off, readBytes);
+                memAvailable -= readBytes;
+                off += readBytes;
+                len -= readBytes;
+            }
+            if (len > 0 && diskTailAvailable > 0)
+            {
+                int readFromTail = diskTailAvailable < len? diskTailAvailable : len;
+                getIfNotClosed(spillBuffer).read(b, off, readFromTail);
+                readBytes += readFromTail;
+                diskTailAvailable -= readFromTail;
+                off += readFromTail;
+                len -= readFromTail;
+                if (diskTailAvailable == 0)
+                    spillBuffer.seek(0);
+            }
+            if (len > 0 && diskHeadAvailable > 0)
+            {
+                int readFromHead = diskHeadAvailable < len? diskHeadAvailable : len;
+                getIfNotClosed(spillBuffer).read(b, off, readFromHead);
+                readBytes += readFromHead;
+                diskHeadAvailable -= readFromHead;
+                off += readFromHead;
+                len -= readFromHead;
+            }
+        }
+
+        if (len > 0)
+            readBytes += getIfNotClosed(in).read(b, off, len);
+
+        return readBytes;
+    }
+
+    private void writeMulti(byte[] b, int off, int len) throws IOException
+    {
+        if (memAvailable > 0)
+        {
+            if (memBuffer == null)
+                memBuffer = new byte[initialMemBufferSize];
+            int pos = maxMemBufferSize - memAvailable;
+            int memWritten = memAvailable < len? memAvailable : len;
+            if (pos + memWritten >= getIfNotClosed(memBuffer).length)
+                growMemBuffer(pos, memWritten);
+            System.arraycopy(b, off, memBuffer, pos, memWritten);
+            off += memWritten;
+            len -= memWritten;
+            memAvailable -= memWritten;
+        }
+
+        if (len > 0)
+        {
+            if (diskTailAvailable == -1)
+            {
+                maybeCreateDiskBuffer();
+                diskHeadAvailable = (int)spillBuffer.getFilePointer();
+                diskTailAvailable = maxDiskBufferSize - diskHeadAvailable;
+            }
+
+            if (len > 0 && diskTailAvailable > 0)
+            {
+                int diskTailWritten = diskTailAvailable < len? diskTailAvailable : len;
+                getIfNotClosed(spillBuffer).write(b, off, diskTailWritten);
+                off += diskTailWritten;
+                len -= diskTailWritten;
+                diskTailAvailable -= diskTailWritten;
+                if (diskTailAvailable == 0)
+                    spillBuffer.seek(0);
+            }
+
+            if (len > 0 && diskTailAvailable > 0)
+            {
+                int diskHeadWritten = diskHeadAvailable < len? diskHeadAvailable : len;
+                getIfNotClosed(spillBuffer).write(b, off, diskHeadWritten);
+            }
+        }
+    }
+
+    private void writeOne(int value) throws IOException
+    {
+        if (memAvailable > 0)
+        {
+            if (memBuffer == null)
+                memBuffer = new byte[initialMemBufferSize];
+            int pos = maxMemBufferSize - memAvailable;
+            if (pos == getIfNotClosed(memBuffer).length)
+                growMemBuffer(pos, 1);
+            getIfNotClosed(memBuffer)[pos] = (byte)value;
+            memAvailable--;
+            return;
+        }
+
+        if (diskTailAvailable == -1)
+        {
+            maybeCreateDiskBuffer();
+            diskHeadAvailable = (int)spillBuffer.getFilePointer();
+            diskTailAvailable = maxDiskBufferSize - diskHeadAvailable;
+        }
+
+        if (diskTailAvailable > 0 || diskHeadAvailable > 0)
+        {
+            getIfNotClosed(spillBuffer).write(value);
+            if (diskTailAvailable > 0)
+                diskTailAvailable--;
+            else if (diskHeadAvailable > 0)
+                diskHeadAvailable--;
+            if (diskTailAvailable == 0)
+                spillBuffer.seek(0);
+            return;
+        }
+    }
+
+    public int read(byte[] b) throws IOException
+    {
+        return read(b, 0, b.length);
+    }
+
+    private void growMemBuffer(int pos, int writeSize)
+    {
+        int newSize = Math.min(2 * (pos + writeSize), maxMemBufferSize);
+        byte newBuffer[] = new byte[newSize];
+        System.arraycopy(memBuffer, 0, newBuffer, 0, (int)pos);
+        memBuffer = newBuffer;
+    }
+
+    public long skip(long n) throws IOException
+    {
+        long skipped = 0;
+
+        if (marked)
+        {
+            //if marked, we need to cache skipped bytes
+            while (n-- > 0 && read() != -1)
+            {
+                skipped++;
+            }
+            return skipped;
+        }
+
+        if (memAvailable > 0)
+        {
+            skipped += memAvailable < n ? memAvailable : n;
+            memAvailable -= skipped;
+            n -= skipped;
+        }
+        if (n > 0 && diskTailAvailable > 0)
+        {
+            int skipFromTail = diskTailAvailable < n? diskTailAvailable : (int)n;
+            getIfNotClosed(spillBuffer).skipBytes(skipFromTail);
+            diskTailAvailable -= skipFromTail;
+            skipped += skipFromTail;
+            n -= skipFromTail;
+            if (diskTailAvailable == 0)
+                spillBuffer.seek(0);
+        }
+        if (n > 0 && diskHeadAvailable > 0)
+        {
+            int skipFromHead = diskHeadAvailable < n? diskHeadAvailable : (int)n;
+            getIfNotClosed(spillBuffer).skipBytes(skipFromHead);
+            diskHeadAvailable -= skipFromHead;
+            skipped += skipFromHead;
+            n -= skipFromHead;
+        }
+
+        if (n > 0)
+            skipped += getIfNotClosed(in).skip(n);
+
+        return skipped;
+    }
+
+    private <T> T getIfNotClosed(T in) throws IOException {
+        if (closed.get())
+            throw new IOException("Stream closed");
+        return in;
+    }
+
+    public void close() throws IOException
+    {
+        close(true);
+    }
+
+    public void close(boolean closeUnderlying) throws IOException
+    {
+        if (closed.compareAndSet(false, true))
+        {
+            Throwable fail = null;
+            if (closeUnderlying)
+            {
+                try
+                {
+                    super.close();
+                }
+                catch (IOException e)
+                {
+                    fail = merge(fail, e);
+                }
+            }
+            try
+            {
+                if (spillBuffer != null)
+                {
+                    this.spillBuffer.close();
+                    this.spillBuffer = null;
+                }
+            } catch (IOException e)
+            {
+                fail = merge(fail, e);
+            }
+            try {
+                if (spillFile.exists())
+                {
+                    spillFile.delete();
+                }
+            }
+            catch (Throwable e)
+            {
+                fail = merge(fail, e);
+            }
+            maybeFail(fail, IOException.class);
+        }
+    }
+
+    /* DataInputPlus methods */
+
+    public void readFully(byte[] b) throws IOException
+    {
+        dataReader.readFully(b);
+    }
+
+    public void readFully(byte[] b, int off, int len) throws IOException
+    {
+        dataReader.readFully(b, off, len);
+    }
+
+    public int skipBytes(int n) throws IOException
+    {
+        return dataReader.skipBytes(n);
+    }
+
+    public boolean readBoolean() throws IOException
+    {
+        return dataReader.readBoolean();
+    }
+
+    public byte readByte() throws IOException
+    {
+        return dataReader.readByte();
+    }
+
+    public int readUnsignedByte() throws IOException
+    {
+        return dataReader.readUnsignedByte();
+    }
+
+    public short readShort() throws IOException
+    {
+        return dataReader.readShort();
+    }
+
+    public int readUnsignedShort() throws IOException
+    {
+        return dataReader.readUnsignedShort();
+    }
+
+    public char readChar() throws IOException
+    {
+        return dataReader.readChar();
+    }
+
+    public int readInt() throws IOException
+    {
+        return dataReader.readInt();
+    }
+
+    public long readLong() throws IOException
+    {
+        return dataReader.readLong();
+    }
+
+    public float readFloat() throws IOException
+    {
+        return dataReader.readFloat();
+    }
+
+    public double readDouble() throws IOException
+    {
+        return dataReader.readDouble();
+    }
+
+    public String readLine() throws IOException
+    {
+        return dataReader.readLine();
+    }
+
+    public String readUTF() throws IOException
+    {
+        return dataReader.readUTF();
+    }
+}

src/java/org/apache/cassandra/io/util/SequentialWriter.java

@@ -304,7 +304,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
         return bufferOffset + (buffer == null ? 0 : buffer.position());
     }
 
-    public FileMark mark()
+    public DataPosition mark()
     {
         return new BufferedFileWriterMark(current());
     }
@@ -313,7 +313,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
      * Drops all buffered data that's past the limits of our new file mark + buffer capacity, or syncs and truncates
      * the underlying file to the marked position
      */
-    public void resetAndTruncate(FileMark mark)
+    public void resetAndTruncate(DataPosition mark)
     {
         assert mark instanceof BufferedFileWriterMark;
 
@@ -411,7 +411,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
     /**
      * Class to hold a mark to the position of the file
      */
-    protected static class BufferedFileWriterMark implements FileMark
+    protected static class BufferedFileWriterMark implements DataPosition
     {
         final long pointer;
 

src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java

@@ -15,24 +15,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.cassandra.utils;
+package org.apache.cassandra.io.util;
 
 import java.io.DataInput;
 import java.io.DataInputStream;
 import java.io.IOException;
 
-import org.apache.cassandra.io.util.DataInputPlus;
-
 /**
  * This class is to track bytes read from given DataInput
  */
-public class BytesReadTracker implements DataInputPlus
+public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker
 {
-
     private long bytesRead;
     final DataInput source;
 
-    public BytesReadTracker(DataInput source)
+    public TrackedDataInputPlus(DataInput source)
     {
         this.source = source;
     }

src/java/org/apache/cassandra/io/util/TrackedInputStream.java

@@ -0,0 +1,76 @@
+/*
+ * 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.util;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * This class is to track bytes read from given DataInputStream
+ */
+public class TrackedInputStream extends FilterInputStream implements BytesReadTracker
+{
+    private long bytesRead;
+
+    public TrackedInputStream(InputStream source)
+    {
+        super(source);
+    }
+
+    public long getBytesRead()
+    {
+        return bytesRead;
+    }
+
+    /**
+     * reset counter to @param count
+     */
+    public void reset(long count)
+    {
+        bytesRead = count;
+    }
+
+    public int read() throws IOException
+    {
+        int read = super.read();
+        bytesRead += 1;
+        return read;
+    }
+
+    public int read(byte[] b, int off, int len) throws IOException
+    {
+        int read = super.read(b, off, len);
+        bytesRead += read;
+        return read;
+    }
+
+    public int read(byte[] b) throws IOException
+    {
+        int read = super.read(b);
+        bytesRead += read;
+        return read;
+    }
+
+    public long skip(long n) throws IOException
+    {
+        long skip = super.skip(n);
+        bytesRead += skip;
+        return skip;
+    }
+}

src/java/org/apache/cassandra/service/StorageService.java

@@ -4366,6 +4366,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return Collections.unmodifiableList(keyspaceNamesList);
     }
 
+
     public List<String> getNonSystemKeyspaces()
     {
         List<String> keyspaceNamesList = new ArrayList<>(Schema.instance.getNonSystemKeyspaces());

src/java/org/apache/cassandra/streaming/StreamReader.java

@@ -40,10 +40,13 @@ import org.apache.cassandra.io.sstable.SSTableSimpleIterator;
 import org.apache.cassandra.io.sstable.format.RangeAwareSSTableWriter;
 import org.apache.cassandra.io.sstable.format.SSTableFormat;
 import org.apache.cassandra.io.sstable.format.Version;
+import org.apache.cassandra.io.util.RewindableDataInputStreamPlus;
 import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.streaming.messages.FileMessageHeader;
 import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.cassandra.utils.BytesReadTracker;
+import org.apache.cassandra.io.util.TrackedInputStream;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 
 
@@ -103,9 +106,9 @@ public class StreamReader
                      session.planId(), fileSeqNum, session.peer, repairedAt, totalSize, cfs.keyspace.getName(),
                      cfs.getColumnFamilyName());
 
-        DataInputStream dis = new DataInputStream(new LZFInputStream(Channels.newInputStream(channel)));
-        BytesReadTracker in = new BytesReadTracker(dis);
-        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata, in, inputVersion, header.toHeader(cfs.metadata));
+        TrackedInputStream in = new TrackedInputStream(new LZFInputStream(Channels.newInputStream(channel)));
+        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata, in, inputVersion, getHeader(cfs.metadata),
+                                                                 totalSize, session.planId());
         SSTableMultiWriter writer = null;
         try
         {
@@ -124,17 +127,27 @@ public class StreamReader
         {
             if (deserializer != null)
                 logger.warn("[Stream {}] Error while reading partition {} from stream on ks='{}' and table='{}'.",
-                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(), cfs.getColumnFamilyName());
+                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(), cfs.getTableName());
             if (writer != null)
             {
                 writer.abort(e);
             }
-            drain(dis, in.getBytesRead());
+            drain(in, in.getBytesRead());
             if (e instanceof IOException)
                 throw (IOException) e;
             else
                 throw Throwables.propagate(e);
         }
+        finally
+        {
+            if (deserializer != null)
+                deserializer.cleanup();
+        }
+    }
+
+    protected SerializationHeader getHeader(CFMetaData metadata)
+    {
+        return header != null? header.toHeader(metadata) : null; //pre-3.0 sstable have no SerializationHeader
     }
 
     protected SSTableMultiWriter createWriter(ColumnFamilyStore cfs, long totalSize, long repairedAt, SSTableFormat.Type format) throws IOException
@@ -143,7 +156,7 @@ public class StreamReader
         if (localDir == null)
             throw new IOException("Insufficient disk space to store " + totalSize + " bytes");
 
-        RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, format, sstableLevel, totalSize, session.getTransaction(cfId), header);
+        RangeAwareSSTableWriter writer = new RangeAwareSSTableWriter(cfs, estimatedKeys, repairedAt, format, sstableLevel, totalSize, session.getTransaction(cfId), getHeader(cfs.metadata));
         StreamHook.instance.reportIncomingFile(cfs, writer, session, fileSeqNum);
         return writer;
     }
@@ -183,6 +196,13 @@ public class StreamReader
 
     public static class StreamDeserializer extends UnmodifiableIterator<Unfiltered> implements UnfilteredRowIterator
     {
+        public static final int INITIAL_MEM_BUFFER_SIZE = Integer.getInteger("cassandra.streamdes.initial_mem_buffer_size", 32768);
+        public static final int MAX_MEM_BUFFER_SIZE = Integer.getInteger("cassandra.streamdes.max_mem_buffer_size", 1048576);
+        public static final int MAX_SPILL_FILE_SIZE = Integer.getInteger("cassandra.streamdes.max_spill_file_size", Integer.MAX_VALUE);
+
+        public static final String BUFFER_FILE_PREFIX = "buf";
+        public static final String BUFFER_FILE_SUFFIX = "dat";
+
         private final CFMetaData metadata;
         private final DataInputPlus in;
         private final SerializationHeader header;
@@ -194,11 +214,20 @@ public class StreamReader
         private Row staticRow;
         private IOException exception;
 
-        public StreamDeserializer(CFMetaData metadata, DataInputPlus in, Version version, SerializationHeader header)
+        public StreamDeserializer(CFMetaData metadata, InputStream in, Version version, SerializationHeader header,
+                                  long totalSize, UUID sessionId) throws IOException
         {
-            assert version.storeRows() : "We don't allow streaming from pre-3.0 nodes";
             this.metadata = metadata;
-            this.in = in;
+            // streaming pre-3.0 sstables require mark/reset support from source stream
+            if (version.correspondingMessagingVersion() < MessagingService.VERSION_30)
+            {
+                logger.trace("Initializing rewindable input stream for reading legacy sstable with {} bytes with following " +
+                             "parameters: initial_mem_buffer_size={}, max_mem_buffer_size={}, max_spill_file_size={}.",
+                             totalSize, INITIAL_MEM_BUFFER_SIZE, MAX_MEM_BUFFER_SIZE, MAX_SPILL_FILE_SIZE);
+                File bufferFile = getTempBufferFile(metadata, totalSize, sessionId);
+                this.in = new RewindableDataInputStreamPlus(in, INITIAL_MEM_BUFFER_SIZE, MAX_MEM_BUFFER_SIZE, bufferFile, MAX_SPILL_FILE_SIZE);
+            } else
+                this.in = new DataInputPlus.DataInputStreamPlus(in);
             this.helper = new SerializationHelper(metadata, version.correspondingMessagingVersion(), SerializationHelper.Flag.PRESERVE_SIZE);
             this.header = header;
         }
@@ -290,5 +319,41 @@ public class StreamReader
         public void close()
         {
         }
+
+        /* We have a separate cleanup method because sometimes close is called before exhausting the
+           StreamDeserializer (for instance, when enclosed in an try-with-resources wrapper, such as in
+           BigTableWriter.append()).
+         */
+        public void cleanup()
+        {
+            if (in instanceof RewindableDataInputStreamPlus)
+            {
+                try
+                {
+                    ((RewindableDataInputStreamPlus) in).close(false);
+                }
+                catch (IOException e)
+                {
+                    logger.warn("Error while closing RewindableDataInputStreamPlus.", e);
+                }
+            }
+        }
+
+        private static File getTempBufferFile(CFMetaData metadata, long totalSize, UUID sessionId) throws IOException
+        {
+            ColumnFamilyStore cfs = Keyspace.open(metadata.ksName).getColumnFamilyStore(metadata.cfName);
+            if (cfs == null)
+            {
+                // schema was dropped during streaming
+                throw new RuntimeException(String.format("CF %s.%s was dropped during streaming", metadata.ksName, metadata.cfName));
+            }
+
+            long maxSize = Math.min(MAX_SPILL_FILE_SIZE, totalSize);
+            File tmpDir = cfs.getDirectories().getTemporaryWriteableDirectoryAsFile(maxSize);
+            if (tmpDir == null)
+                throw new IOException(String.format("No sufficient disk space to stream legacy sstable from {}.{}. " +
+                                                         "Required disk space: %s.", FBUtilities.prettyPrintMemory(maxSize)));
+            return new File(tmpDir, String.format("%s-%s.%s", BUFFER_FILE_PREFIX, sessionId, BUFFER_FILE_SUFFIX));
+        }
     }
 }

src/java/org/apache/cassandra/streaming/compress/CompressedStreamReader.java

@@ -24,7 +24,6 @@ import java.nio.channels.ReadableByteChannel;
 
 import com.google.common.base.Throwables;
 
-import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.io.sstable.SSTableMultiWriter;
 import org.apache.cassandra.io.sstable.format.RangeAwareSSTableWriter;
 
@@ -39,7 +38,7 @@ import org.apache.cassandra.streaming.ProgressInfo;
 import org.apache.cassandra.streaming.StreamReader;
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.messages.FileMessageHeader;
-import org.apache.cassandra.utils.BytesReadTracker;
+import org.apache.cassandra.io.util.TrackedInputStream;
 import org.apache.cassandra.utils.Pair;
 
 /**
@@ -84,8 +83,10 @@ public class CompressedStreamReader extends StreamReader
 
         CompressedInputStream cis = new CompressedInputStream(Channels.newInputStream(channel), compressionInfo,
                                                               inputVersion.compressedChecksumType(), cfs::getCrcCheckChance);
-        BytesReadTracker in = new BytesReadTracker(new DataInputStream(cis));
-        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata, in, inputVersion, header.toHeader(cfs.metadata));
+        TrackedInputStream in = new TrackedInputStream(cis);
+
+        StreamDeserializer deserializer = new StreamDeserializer(cfs.metadata, in, inputVersion, getHeader(cfs.metadata),
+                                                                 totalSize, session.planId());
         SSTableMultiWriter writer = null;
         try
         {
@@ -116,17 +117,22 @@ public class CompressedStreamReader extends StreamReader
         {
             if (deserializer != null)
                 logger.warn("[Stream {}] Error while reading partition {} from stream on ks='{}' and table='{}'.",
-                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(), cfs.getColumnFamilyName());
+                            session.planId(), deserializer.partitionKey(), cfs.keyspace.getName(), cfs.getTableName());
             if (writer != null)
             {
                 writer.abort(e);
             }
-            drain(cis, in.getBytesRead());
+            drain(in, in.getBytesRead());
             if (e instanceof IOException)
                 throw (IOException) e;
             else
                 throw Throwables.propagate(e);
         }
+        finally
+        {
+            if (deserializer != null)
+                deserializer.cleanup();
+        }
     }
 
     @Override

src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java

@@ -204,7 +204,7 @@ public class FileMessageHeader
             out.writeLong(header.repairedAt);
             out.writeInt(header.sstableLevel);
 
-            if (version >= StreamMessage.VERSION_30)
+            if (version >= StreamMessage.VERSION_30 && header.version.storeRows())
                 SerializationHeader.serializer.serialize(header.version, header.header, out);
             return compressionInfo;
         }
@@ -227,7 +227,7 @@ public class FileMessageHeader
             CompressionInfo compressionInfo = CompressionInfo.serializer.deserialize(in, MessagingService.current_version);
             long repairedAt = in.readLong();
             int sstableLevel = in.readInt();
-            SerializationHeader.Component header = version >= StreamMessage.VERSION_30
+            SerializationHeader.Component header = version >= StreamMessage.VERSION_30 && sstableVersion.storeRows()
                                                  ? SerializationHeader.serializer.deserialize(sstableVersion, in)
                                                  : null;
 

src/java/org/apache/cassandra/tools/nodetool/Repair.java

@@ -56,7 +56,7 @@ public class Repair extends NodeToolCmd
     private boolean localDC = false;
 
     @Option(title = "specific_dc", name = {"-dc", "--in-dc"}, description = "Use -dc to repair specific datacenters")
-    private List<String> specificDataCenters = new ArrayList<>();
+    private List<String> specificDataCenters = new ArrayList<>();;
 
     @Option(title = "specific_host", name = {"-hosts", "--in-hosts"}, description = "Use -hosts to repair specific hosts")
     private List<String> specificHosts = new ArrayList<>();

src/java/org/apache/cassandra/utils/CloseableIterator.java

@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.utils;
 
-import java.io.Closeable;
 import java.util.Iterator;
 
 // so we can instantiate anonymous classes implementing both interfaces

test/data/legacy-sstables/jb/legacy_tables/legacy_jb_clust_compact/legacy_tables-legacy_jb_clust_compact-jb-1-TOC.txt

@@ -0,0 +1,7 @@
+Data.db
+CompressionInfo.db
+Index.db
+Summary.db
+TOC.txt
+Statistics.db
+Filter.db

test/data/legacy-sstables/jb/legacy_tables/legacy_jb_clust_counter_compact/legacy_tables-legacy_jb_clust_counter_compact-jb-1-TOC.txt

@@ -0,0 +1,7 @@
+Data.db
+CompressionInfo.db
+Index.db
+Summary.db
+TOC.txt
+Statistics.db
+Filter.db

test/data/legacy-sstables/jb/legacy_tables/legacy_jb_simple_compact/legacy_tables-legacy_jb_simple_compact-jb-1-TOC.txt

@@ -0,0 +1,7 @@
+Data.db
+CompressionInfo.db
+Index.db
+Summary.db
+TOC.txt
+Statistics.db
+Filter.db

test/data/legacy-sstables/jb/legacy_tables/legacy_jb_simple_counter_compact/legacy_tables-legacy_jb_simple_counter_compact-jb-1-TOC.txt

@@ -0,0 +1,7 @@
+Data.db
+CompressionInfo.db
+Index.db
+Summary.db
+TOC.txt
+Statistics.db
+Filter.db

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_clust_compact/legacy_tables-legacy_ka_clust_compact-ka-1-Digest.sha1

@@ -0,0 +1 @@
+1331331706

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_clust_compact/legacy_tables-legacy_ka_clust_compact-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Filter.db
+TOC.txt
+Statistics.db
+Summary.db
+Index.db
+Data.db
+Digest.sha1
+CompressionInfo.db

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_clust_counter_compact/legacy_tables-legacy_ka_clust_counter_compact-ka-1-Digest.sha1

@@ -0,0 +1 @@
+2793875907

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_clust_counter_compact/legacy_tables-legacy_ka_clust_counter_compact-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Filter.db
+TOC.txt
+Statistics.db
+Summary.db
+Index.db
+Data.db
+Digest.sha1
+CompressionInfo.db

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_simple_compact/legacy_tables-legacy_ka_simple_compact-ka-1-Digest.sha1

@@ -0,0 +1 @@
+606280675

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_simple_compact/legacy_tables-legacy_ka_simple_compact-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Filter.db
+TOC.txt
+Statistics.db
+Summary.db
+Index.db
+Data.db
+Digest.sha1
+CompressionInfo.db

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_simple_counter_compact/legacy_tables-legacy_ka_simple_counter_compact-ka-1-Digest.sha1

@@ -0,0 +1 @@
+616768162

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_simple_counter_compact/legacy_tables-legacy_ka_simple_counter_compact-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Filter.db
+TOC.txt
+Statistics.db
+Summary.db
+Index.db
+Data.db
+Digest.sha1
+CompressionInfo.db

test/data/legacy-sstables/la/legacy_tables/legacy_la_clust_compact/la-1-big-Digest.adler32

@@ -0,0 +1 @@
+1372047449

test/data/legacy-sstables/la/legacy_tables/legacy_la_clust_compact/la-1-big-TOC.txt

@@ -0,0 +1,8 @@
+TOC.txt
+Statistics.db
+Digest.adler32
+CompressionInfo.db
+Summary.db
+Data.db
+Filter.db
+Index.db