Adds the ability to use uncompressed chunks in compressed files

Triggered when size of compressed data surpasses a configurable
threshold.

Patch by Branimir Lambov; reviewed by Ropert Stupp for CASSANDRA-10520

CHANGES.txt

@@ -1,4 +1,5 @@
 4.0
+ * Adds the ability to use uncompressed chunks in compressed files (CASSANDRA-10520)
  * Don't flush sstables when streaming for incremental repair (CASSANDRA-13226)
  * Remove unused method (CASSANDRA-13227)
  * Fix minor bugs related to #9143 (CASSANDRA-13217)

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

@@ -60,6 +60,8 @@ public class CompressedSequentialWriter extends SequentialWriter
     private final ByteBuffer crcCheckBuffer = ByteBuffer.allocate(4);
     private final Optional<File> digestFile;
 
+    private final int maxCompressedLength;
+
     /**
      * Create CompressedSequentialWriter without digest file.
      *
@@ -90,6 +92,8 @@ public class CompressedSequentialWriter extends SequentialWriter
         // buffer for compression should be the same size as buffer itself
         compressed = compressor.preferredBufferType().allocate(compressor.initialCompressedBufferLength(buffer.capacity()));
 
+        maxCompressedLength = parameters.maxCompressedLength();
+
         /* Index File (-CompressionInfo.db component) and it's header */
         metadataWriter = CompressionMetadata.Writer.open(parameters, offsetsPath);
 
@@ -146,6 +150,12 @@ public class CompressedSequentialWriter extends SequentialWriter
 
         int compressedLength = compressed.position();
         uncompressedSize += buffer.position();
+        ByteBuffer toWrite = compressed;
+        if (compressedLength > maxCompressedLength)
+        {
+            toWrite = buffer;
+            compressedLength = buffer.position();
+        }
         compressedSize += compressedLength;
 
         try
@@ -155,18 +165,20 @@ public class CompressedSequentialWriter extends SequentialWriter
             chunkCount++;
 
             // write out the compressed data
-            compressed.flip();
-            channel.write(compressed);
+            toWrite.flip();
+            channel.write(toWrite);
 
             // write corresponding checksum
-            compressed.rewind();
-            crcMetadata.appendDirect(compressed, true);
+            toWrite.rewind();
+            crcMetadata.appendDirect(toWrite, true);
             lastFlushOffset += compressedLength + 4;
         }
         catch (IOException e)
         {
             throw new FSWriteError(e, getPath());
         }
+        if (toWrite == buffer)
+            buffer.position(compressedLength);
 
         // next chunk should be written right after current + length of the checksum (int)
         chunkOffset += compressedLength + 4;
@@ -228,7 +240,10 @@ public class CompressedSequentialWriter extends SequentialWriter
                 // Repopulate buffer from compressed data
                 buffer.clear();
                 compressed.flip();
-                compressor.uncompress(compressed, buffer);
+                if (chunkSize <= maxCompressedLength)
+                    compressor.uncompress(compressed, buffer);
+                else
+                    buffer.put(compressed);
             }
             catch (IOException e)
             {

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

@@ -84,12 +84,22 @@ public class CompressionMetadata
      */
     public static CompressionMetadata create(String dataFilePath)
     {
-        Descriptor desc = Descriptor.fromFilename(dataFilePath);
-        return new CompressionMetadata(desc.filenameFor(Component.COMPRESSION_INFO), new File(dataFilePath).length());
+        return createWithLength(dataFilePath, new File(dataFilePath).length());
+    }
+
+    public static CompressionMetadata createWithLength(String dataFilePath, long compressedLength)
+    {
+        return new CompressionMetadata(Descriptor.fromFilename(dataFilePath), compressedLength);
     }
 
     @VisibleForTesting
-    public CompressionMetadata(String indexFilePath, long compressedLength)
+    public CompressionMetadata(Descriptor desc, long compressedLength)
+    {
+        this(desc.filenameFor(Component.COMPRESSION_INFO), compressedLength, desc.version.hasMaxCompressedLength());
+    }
+
+    @VisibleForTesting
+    public CompressionMetadata(String indexFilePath, long compressedLength, boolean hasMaxCompressedSize)
     {
         this.indexFilePath = indexFilePath;
 
@@ -105,9 +115,12 @@ public class CompressionMetadata
                 options.put(key, value);
             }
             int chunkLength = stream.readInt();
+            int maxCompressedSize = Integer.MAX_VALUE;
+            if (hasMaxCompressedSize)
+                maxCompressedSize = stream.readInt();
             try
             {
-                parameters = new CompressionParams(compressorName, chunkLength, options);
+                parameters = new CompressionParams(compressorName, chunkLength, maxCompressedSize, options);
             }
             catch (ConfigurationException e)
             {
@@ -150,6 +163,11 @@ public class CompressionMetadata
         return parameters.chunkLength();
     }
 
+    public int maxCompressedLength()
+    {
+        return parameters.maxCompressedLength();
+    }
+
     /**
      * Returns the amount of memory in bytes used off heap.
      * @return the amount of memory in bytes used off heap
@@ -350,6 +368,7 @@ public class CompressionMetadata
 
                 // store the length of the chunk
                 out.writeInt(parameters.chunkLength());
+                out.writeInt(parameters.maxCompressedLength());
                 // store position and reserve a place for uncompressed data length and chunks count
                 out.writeLong(dataLength);
                 out.writeInt(chunks);

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

@@ -51,6 +51,8 @@ public abstract class Version
 
     public abstract boolean hasCommitLogIntervals();
 
+    public abstract boolean hasMaxCompressedLength();
+
     public abstract boolean hasPendingRepair();
 
     public String getVersion()

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

@@ -112,7 +112,7 @@ public class BigFormat implements SSTableFormat
     // we always incremented the major version.
     static class BigVersion extends Version
     {
-        public static final String current_version = "md";
+        public static final String current_version = "na";
         public static final String earliest_supported_version = "ma";
 
         // ma (3.0.0): swap bf hash order
@@ -120,6 +120,8 @@ public class BigFormat implements SSTableFormat
         // mb (3.0.7, 3.7): commit log lower bound included
         // mc (3.0.8, 3.9): commit log intervals included
         // md (3.0.9, 3.10): pending repair session included
+
+        // na (4.0.0): uncompressed chunks
         //
         // NOTE: when adding a new version, please add that to LegacySSTableTest, too.
 
@@ -127,6 +129,7 @@ public class BigFormat implements SSTableFormat
         public final int correspondingMessagingVersion;
         private final boolean hasCommitLogLowerBound;
         private final boolean hasCommitLogIntervals;
+        public final boolean hasMaxCompressedLength;
         private final boolean hasPendingRepair;
 
         BigVersion(String version)
@@ -138,6 +141,7 @@ public class BigFormat implements SSTableFormat
 
             hasCommitLogLowerBound = version.compareTo("mb") >= 0;
             hasCommitLogIntervals = version.compareTo("mc") >= 0;
+            hasMaxCompressedLength = version.compareTo("na") >= 0;
             hasPendingRepair = version.compareTo("md") >= 0;
         }
 
@@ -181,5 +185,11 @@ public class BigFormat implements SSTableFormat
         {
             return isCompatible() && version.charAt(0) == current_version.charAt(0);
         }
+
+        @Override
+        public boolean hasMaxCompressedLength()
+        {
+            return hasMaxCompressedLength;
+        }
     }
 }

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

@@ -34,11 +34,13 @@ import org.apache.cassandra.utils.ChecksumType;
 public abstract class CompressedChunkReader extends AbstractReaderFileProxy implements ChunkReader
 {
     final CompressionMetadata metadata;
+    final int maxCompressedLength;
 
     protected CompressedChunkReader(ChannelProxy channel, CompressionMetadata metadata)
     {
         super(channel, metadata.dataLength);
         this.metadata = metadata;
+        this.maxCompressedLength = metadata.maxCompressedLength();
         assert Integer.bitCount(metadata.chunkLength()) == 1; //must be a power of two
     }
 
@@ -48,6 +50,11 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
         return metadata.parameters.getCrcCheckChance();
     }
 
+    public boolean maybeCheckCrc()
+    {
+        return metadata.parameters.maybeCheckCrc();
+    }
+
     @Override
     public String toString()
     {
@@ -90,7 +97,8 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
 
         public ByteBuffer allocateBuffer()
         {
-            return allocateBuffer(metadata.compressor().initialCompressedBufferLength(metadata.chunkLength()));
+            return allocateBuffer(Math.min(maxCompressedLength,
+                                           metadata.compressor().initialCompressedBufferLength(metadata.chunkLength())));
         }
 
         public ByteBuffer allocateBuffer(int size)
@@ -108,54 +116,58 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
                 assert position <= fileLength;
 
                 CompressionMetadata.Chunk chunk = metadata.chunkFor(position);
-                ByteBuffer compressed = compressedHolder.get();
-
-                if (compressed.capacity() < chunk.length)
+                if (chunk.length <= maxCompressedLength)
                 {
-                    compressed = allocateBuffer(chunk.length);
-                    compressedHolder.set(compressed);
+                    ByteBuffer compressed = compressedHolder.get();
+                    assert compressed.capacity() >= chunk.length;
+                    compressed.clear().limit(chunk.length);
+                    if (channel.read(compressed, chunk.offset) != chunk.length)
+                        throw new CorruptBlockException(channel.filePath(), chunk);
+
+                    compressed.flip();
+                    uncompressed.clear();
+
+                    try
+                    {
+                        metadata.compressor().uncompress(compressed, uncompressed);
+                    }
+                    catch (IOException e)
+                    {
+                        throw new CorruptBlockException(channel.filePath(), chunk, e);
+                    }
+                    maybeCheckCrc(chunk, compressed);
                 }
                 else
                 {
-                    compressed.clear();
-                }
-
-                compressed.limit(chunk.length);
-                if (channel.read(compressed, chunk.offset) != chunk.length)
-                    throw new CorruptBlockException(channel.filePath(), chunk);
-
-                compressed.flip();
-                uncompressed.clear();
-
-                try
-                {
-                    metadata.compressor().uncompress(compressed, uncompressed);
-                }
-                catch (IOException e)
-                {
-                    throw new CorruptBlockException(channel.filePath(), chunk, e);
-                }
-                finally
-                {
-                    uncompressed.flip();
-                }
-
-                if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble())
-                {
-                    compressed.rewind();
-                    int checksum = (int) ChecksumType.CRC32.of(compressed);
-
-                    compressed.clear().limit(Integer.BYTES);
-                    if (channel.read(compressed, chunk.offset + chunk.length) != Integer.BYTES
-                                || compressed.getInt(0) != checksum)
+                    uncompressed.position(0).limit(chunk.length);
+                    if (channel.read(uncompressed, chunk.offset) != chunk.length)
                         throw new CorruptBlockException(channel.filePath(), chunk);
+                    maybeCheckCrc(chunk, uncompressed);
                 }
+                uncompressed.flip();
             }
             catch (CorruptBlockException e)
             {
+                // Make sure reader does not see stale data.
+                uncompressed.position(0).limit(0);
                 throw new CorruptSSTableException(e, channel.filePath());
             }
         }
+
+        void maybeCheckCrc(CompressionMetadata.Chunk chunk, ByteBuffer content) throws CorruptBlockException
+        {
+            if (metadata.parameters.maybeCheckCrc())
+            {
+                content.flip();
+                int checksum = (int) ChecksumType.CRC32.of(content);
+
+                ByteBuffer scratch = compressedHolder.get(); // This may match content. That's ok, we no longer need it.
+                scratch.clear().limit(Integer.BYTES);
+                if (channel.read(scratch, chunk.offset + chunk.length) != Integer.BYTES
+                            || scratch.getInt(0) != checksum)
+                    throw new CorruptBlockException(channel.filePath(), chunk);
+            }
+        }
     }
 
     public static class Mmap extends CompressedChunkReader
@@ -190,18 +202,18 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
 
                 try
                 {
-                    metadata.compressor().uncompress(compressedChunk, uncompressed);
+                    if (chunk.length <= maxCompressedLength)
+                        metadata.compressor().uncompress(compressedChunk, uncompressed);
+                    else
+                        uncompressed.put(compressedChunk);
                 }
                 catch (IOException e)
                 {
                     throw new CorruptBlockException(channel.filePath(), chunk, e);
                 }
-                finally
-                {
-                    uncompressed.flip();
-                }
+                uncompressed.flip();
 
-                if (getCrcCheckChance() > ThreadLocalRandom.current().nextDouble())
+                if (maybeCheckCrc())
                 {
                     compressedChunk.position(chunkOffset).limit(chunkOffset + chunk.length);
 
@@ -214,6 +226,8 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
             }
             catch (CorruptBlockException e)
             {
+                // Make sure reader does not see stale data.
+                uncompressed.position(0).limit(0);
                 throw new CorruptSSTableException(e, channel.filePath());
             }
 

src/java/org/apache/cassandra/schema/CompressionParams.java

@@ -23,10 +23,13 @@ import java.lang.reflect.Method;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.ThreadLocalRandom;
 
 import com.google.common.collect.ImmutableMap;
+
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,6 +40,7 @@ import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.compress.*;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.streaming.messages.StreamMessage;
 
 import static java.lang.String.format;
 
@@ -50,14 +54,18 @@ public final class CompressionParams
     private static volatile boolean hasLoggedCrcCheckChanceWarning;
 
     public static final int DEFAULT_CHUNK_LENGTH = 65536;
+    public static final double DEFAULT_MIN_COMPRESS_RATIO = 1.1;
     public static final IVersionedSerializer<CompressionParams> serializer = new Serializer();
 
     public static final String CLASS = "class";
     public static final String CHUNK_LENGTH_IN_KB = "chunk_length_in_kb";
     public static final String ENABLED = "enabled";
+    public static final String MIN_COMPRESS_RATIO = "min_compress_ratio";
 
     public static final CompressionParams DEFAULT = new CompressionParams(LZ4Compressor.create(Collections.<String, String>emptyMap()),
                                                                           DEFAULT_CHUNK_LENGTH,
+                                                                          calcMaxCompressedLength(DEFAULT_CHUNK_LENGTH, DEFAULT_MIN_COMPRESS_RATIO),
+                                                                          DEFAULT_MIN_COMPRESS_RATIO,
                                                                           Collections.emptyMap());
 
     private static final String CRC_CHECK_CHANCE_WARNING = "The option crc_check_chance was deprecated as a compression option. " +
@@ -68,7 +76,9 @@ public final class CompressionParams
     @Deprecated public static final String CRC_CHECK_CHANCE = "crc_check_chance";
 
     private final ICompressor sstableCompressor;
-    private final Integer chunkLength;
+    private final int chunkLength;
+    private final int maxCompressedLength;  // In content we store max length to avoid rounding errors causing compress/decompress mismatch.
+    private final double minCompressRatio;  // In configuration we store min ratio, the input parameter.
     private final ImmutableMap<String, String> otherOptions; // Unrecognized options, can be used by the compressor
 
     private volatile double crcCheckChance = 1.0;
@@ -94,9 +104,10 @@ public final class CompressionParams
             sstableCompressionClass = removeSstableCompressionClass(options);
         }
 
-        Integer chunkLength = removeChunkLength(options);
+        int chunkLength = removeChunkLength(options);
+        double minCompressRatio = removeMinCompressRatio(options);
 
-        CompressionParams cp = new CompressionParams(sstableCompressionClass, chunkLength, options);
+        CompressionParams cp = new CompressionParams(sstableCompressionClass, options, chunkLength, minCompressRatio);
         cp.validate();
 
         return cp;
@@ -109,54 +120,83 @@ public final class CompressionParams
 
     public static CompressionParams noCompression()
     {
-        return new CompressionParams((ICompressor) null, DEFAULT_CHUNK_LENGTH, Collections.emptyMap());
+        return new CompressionParams((ICompressor) null, DEFAULT_CHUNK_LENGTH, Integer.MAX_VALUE, 0.0, Collections.emptyMap());
     }
 
     public static CompressionParams snappy()
     {
-        return snappy(null);
+        return snappy(DEFAULT_CHUNK_LENGTH);
     }
 
-    public static CompressionParams snappy(Integer chunkLength)
+    public static CompressionParams snappy(int chunkLength)
     {
-        return new CompressionParams(SnappyCompressor.instance, chunkLength, Collections.emptyMap());
+        return snappy(chunkLength, DEFAULT_MIN_COMPRESS_RATIO);
+    }
+
+    public static CompressionParams snappy(int chunkLength, double minCompressRatio)
+    {
+        return new CompressionParams(SnappyCompressor.instance, chunkLength, calcMaxCompressedLength(chunkLength, minCompressRatio), minCompressRatio, Collections.emptyMap());
     }
 
     public static CompressionParams deflate()
     {
-        return deflate(null);
+        return deflate(DEFAULT_CHUNK_LENGTH);
     }
 
-    public static CompressionParams deflate(Integer chunkLength)
+    public static CompressionParams deflate(int chunkLength)
     {
-        return new CompressionParams(DeflateCompressor.instance, chunkLength, Collections.emptyMap());
+        return new CompressionParams(DeflateCompressor.instance, chunkLength, Integer.MAX_VALUE, 0.0, Collections.emptyMap());
     }
 
     public static CompressionParams lz4()
     {
-        return lz4(null);
+        return lz4(DEFAULT_CHUNK_LENGTH);
     }
 
-    public static CompressionParams lz4(Integer chunkLength)
+    public static CompressionParams lz4(int chunkLength)
     {
-        return new CompressionParams(LZ4Compressor.create(Collections.emptyMap()), chunkLength, Collections.emptyMap());
+        return lz4(chunkLength, Integer.MAX_VALUE);
     }
 
-    public CompressionParams(String sstableCompressorClass, Integer chunkLength, Map<String, String> otherOptions) throws ConfigurationException
+    public static CompressionParams lz4(int chunkLength, int maxCompressedLength)
     {
-        this(createCompressor(parseCompressorClass(sstableCompressorClass), otherOptions), chunkLength, otherOptions);
+        return new CompressionParams(LZ4Compressor.create(Collections.emptyMap()), chunkLength, maxCompressedLength, calcMinCompressRatio(chunkLength, maxCompressedLength), Collections.emptyMap());
     }
 
-    private CompressionParams(ICompressor sstableCompressor, Integer chunkLength, Map<String, String> otherOptions) throws ConfigurationException
+    public CompressionParams(String sstableCompressorClass, Map<String, String> otherOptions, int chunkLength, double minCompressRatio) throws ConfigurationException
+    {
+        this(createCompressor(parseCompressorClass(sstableCompressorClass), otherOptions), chunkLength, calcMaxCompressedLength(chunkLength, minCompressRatio), minCompressRatio, otherOptions);
+    }
+
+    static int calcMaxCompressedLength(int chunkLength, double minCompressRatio)
+    {
+        return (int) Math.ceil(Math.min(chunkLength / minCompressRatio, Integer.MAX_VALUE));
+    }
+
+    public CompressionParams(String sstableCompressorClass, int chunkLength, int maxCompressedLength, Map<String, String> otherOptions) throws ConfigurationException
+    {
+        this(createCompressor(parseCompressorClass(sstableCompressorClass), otherOptions), chunkLength, maxCompressedLength, calcMinCompressRatio(chunkLength, maxCompressedLength), otherOptions);
+    }
+
+    static double calcMinCompressRatio(int chunkLength, int maxCompressedLength)
+    {
+        if (maxCompressedLength == Integer.MAX_VALUE)
+            return 0;
+        return chunkLength * 1.0 / maxCompressedLength;
+    }
+
+    private CompressionParams(ICompressor sstableCompressor, int chunkLength, int maxCompressedLength, double minCompressRatio, Map<String, String> otherOptions) throws ConfigurationException
     {
         this.sstableCompressor = sstableCompressor;
         this.chunkLength = chunkLength;
         this.otherOptions = ImmutableMap.copyOf(otherOptions);
+        this.minCompressRatio = minCompressRatio;
+        this.maxCompressedLength = maxCompressedLength;
     }
 
     public CompressionParams copy()
     {
-        return new CompressionParams(sstableCompressor, chunkLength, otherOptions);
+        return new CompressionParams(sstableCompressor, chunkLength, maxCompressedLength, minCompressRatio, otherOptions);
     }
 
     /**
@@ -184,7 +224,12 @@ public final class CompressionParams
 
     public int chunkLength()
     {
-        return chunkLength == null ? DEFAULT_CHUNK_LENGTH : chunkLength;
+        return chunkLength;
+    }
+
+    public int maxCompressedLength()
+    {
+        return maxCompressedLength;
     }
 
     private static Class<?> parseCompressorClass(String className) throws ConfigurationException
@@ -312,7 +357,7 @@ public final class CompressionParams
      * @param options the options
      * @return the chunk length value
      */
-    private static Integer removeChunkLength(Map<String, String> options)
+    private static int removeChunkLength(Map<String, String> options)
     {
         if (options.containsKey(CHUNK_LENGTH_IN_KB))
         {
@@ -339,7 +384,23 @@ public final class CompressionParams
             return parseChunkLength(options.remove(CHUNK_LENGTH_KB));
         }
 
-        return null;
+        return DEFAULT_CHUNK_LENGTH;
+    }
+
+    /**
+     * Removes the min compress ratio option from the specified set of option.
+     *
+     * @param options the options
+     * @return the min compress ratio, used to calculate max chunk size to write compressed
+     */
+    private static double removeMinCompressRatio(Map<String, String> options)
+    {
+        String ratio = options.remove(MIN_COMPRESS_RATIO);
+        if (ratio != null)
+        {
+            return Double.parseDouble(ratio);
+        }
+        return DEFAULT_MIN_COMPRESS_RATIO;
     }
 
     /**
@@ -420,25 +481,14 @@ public final class CompressionParams
     public void validate() throws ConfigurationException
     {
         // if chunk length was not set (chunkLength == null), this is fine, default will be used
-        if (chunkLength != null)
-        {
-            if (chunkLength <= 0)
-                throw new ConfigurationException("Invalid negative or null " + CHUNK_LENGTH_IN_KB);
+        if (chunkLength <= 0)
+            throw new ConfigurationException("Invalid negative or null " + CHUNK_LENGTH_IN_KB);
 
-            int c = chunkLength;
-            boolean found = false;
-            while (c != 0)
-            {
-                if ((c & 0x01) != 0)
-                {
-                    if (found)
-                        throw new ConfigurationException(CHUNK_LENGTH_IN_KB + " must be a power of 2");
-                    else
-                        found = true;
-                }
-                c >>= 1;
-            }
-        }
+        if ((chunkLength & (chunkLength - 1)) != 0)
+            throw new ConfigurationException(CHUNK_LENGTH_IN_KB + " must be a power of 2");
+
+        if (maxCompressedLength < 0)
+            throw new ConfigurationException("Invalid negative " + MIN_COMPRESS_RATIO);
     }
 
     public Map<String, String> asMap()
@@ -449,6 +499,7 @@ public final class CompressionParams
         Map<String, String> options = new HashMap<>(otherOptions);
         options.put(CLASS, sstableCompressor.getClass().getName());
         options.put(CHUNK_LENGTH_IN_KB, chunkLengthInKB());
+        options.put(MIN_COMPRESS_RATIO, String.valueOf(minCompressRatio));
 
         return options;
     }
@@ -468,6 +519,12 @@ public final class CompressionParams
         return crcCheckChance;
     }
 
+    public boolean maybeCheckCrc()
+    {
+        double checkChance = getCrcCheckChance();
+        return checkChance > 0d && checkChance > ThreadLocalRandom.current().nextDouble();
+    }
+
     @Override
     public boolean equals(Object obj)
     {
@@ -510,6 +567,11 @@ public final class CompressionParams
                 out.writeUTF(entry.getValue());
             }
             out.writeInt(parameters.chunkLength());
+            if (version >= StreamMessage.VERSION_40)
+                out.writeInt(parameters.maxCompressedLength);
+            else
+                if (parameters.maxCompressedLength != Integer.MAX_VALUE)
+                    throw new UnsupportedOperationException("Cannot stream SSTables with uncompressed chunks to pre-4.0 nodes.");
         }
 
         public CompressionParams deserialize(DataInputPlus in, int version) throws IOException
@@ -524,10 +586,14 @@ public final class CompressionParams
                 options.put(key, value);
             }
             int chunkLength = in.readInt();
+            int minCompressRatio = Integer.MAX_VALUE;   // Earlier Cassandra cannot use uncompressed chunks.
+            if (version >= StreamMessage.VERSION_40)
+                minCompressRatio = in.readInt();
+
             CompressionParams parameters;
             try
             {
-                parameters = new CompressionParams(compressorName, chunkLength, options);
+                parameters = new CompressionParams(compressorName, chunkLength, minCompressRatio, options);
             }
             catch (ConfigurationException e)
             {
@@ -546,6 +612,8 @@ public final class CompressionParams
                 size += TypeSizes.sizeof(entry.getValue());
             }
             size += TypeSizes.sizeof(parameters.chunkLength());
+            if (version >= StreamMessage.VERSION_40)
+                size += TypeSizes.sizeof(parameters.maxCompressedLength());
             return size;
         }
     }

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

@@ -148,12 +148,16 @@ public abstract class AbstractReadExecutor
 
     private static ReadRepairDecision newReadRepairDecision(TableMetadata metadata)
     {
-        double chance = ThreadLocalRandom.current().nextDouble();
-        if (metadata.params.readRepairChance > chance)
-            return ReadRepairDecision.GLOBAL;
+        if (metadata.params.readRepairChance > 0d ||
+            metadata.params.dcLocalReadRepairChance > 0)
+        {
+            double chance = ThreadLocalRandom.current().nextDouble();
+            if (metadata.params.readRepairChance > chance)
+                return ReadRepairDecision.GLOBAL;
 
-        if (metadata.params.dcLocalReadRepairChance > chance)
-            return ReadRepairDecision.DC_LOCAL;
+            if (metadata.params.dcLocalReadRepairChance > chance)
+                return ReadRepairDecision.DC_LOCAL;
+        }
 
         return ReadRepairDecision.NONE;
     }

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

@@ -160,11 +160,18 @@ public class CompressedInputStream extends InputStream
     private void decompress(byte[] compressed) throws IOException
     {
         // uncompress
-        validBufferBytes = info.parameters.getSstableCompressor().uncompress(compressed, 0, compressed.length - checksumBytes.length, buffer, 0);
+        if (compressed.length - checksumBytes.length < info.parameters.maxCompressedLength())
+            validBufferBytes = info.parameters.getSstableCompressor().uncompress(compressed, 0, compressed.length - checksumBytes.length, buffer, 0);
+        else
+        {
+            validBufferBytes = compressed.length - checksumBytes.length;
+            System.arraycopy(compressed, 0, buffer, 0, validBufferBytes);
+        }
         totalCompressedBytesRead += compressed.length;
 
         // validate crc randomly
-        if (this.crcCheckChanceSupplier.get() > ThreadLocalRandom.current().nextDouble())
+        double crcCheckChance = this.crcCheckChanceSupplier.get();
+        if (crcCheckChance > 0d && crcCheckChance > ThreadLocalRandom.current().nextDouble())
         {
             int checksum = (int) checksumType.of(compressed, 0, compressed.length - checksumBytes.length);
 

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

@@ -221,7 +221,7 @@ public class FileMessageHeader
             List<Pair<Long, Long>> sections = new ArrayList<>(count);
             for (int k = 0; k < count; k++)
                 sections.add(Pair.create(in.readLong(), in.readLong()));
-            CompressionInfo compressionInfo = CompressionInfo.serializer.deserialize(in, MessagingService.current_version);
+            CompressionInfo compressionInfo = CompressionInfo.serializer.deserialize(in, version);
             long repairedAt = in.readLong();
             int sstableLevel = in.readInt();
             SerializationHeader.Component header =  SerializationHeader.serializer.deserialize(sstableVersion, in);
@@ -246,8 +246,7 @@ public class FileMessageHeader
             size += CompressionInfo.serializer.serializedSize(header.compressionInfo, version);
             size += TypeSizes.sizeof(header.sstableLevel);
 
-            if (version >= StreamMessage.VERSION_30)
-                size += SerializationHeader.serializer.serializedSize(header.version, header.header);
+            size += SerializationHeader.serializer.serializedSize(header.version, header.header);
 
             return size;
         }

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

@@ -33,8 +33,8 @@ import org.apache.cassandra.streaming.StreamSession;
 public abstract class StreamMessage
 {
     /** Streaming protocol version */
-    public static final int VERSION_30 = 4;
-    public static final int CURRENT_VERSION = VERSION_30;
+    public static final int VERSION_40 = 5;
+    public static final int CURRENT_VERSION = VERSION_40;
 
     private transient volatile boolean sent = false;
 

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+1337396261

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-big-TOC.txt

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

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-bti-Digest.crc32

@@ -0,0 +1 @@
+2666613329

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust/na-1-bti-TOC.txt

@@ -0,0 +1,8 @@
+Data.db
+Filter.db
+Statistics.db
+CompressionInfo.db
+Partitions.db
+TOC.txt
+Rows.db
+Digest.crc32

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+2454456604

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-big-TOC.txt

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

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-bti-Digest.crc32

@@ -0,0 +1 @@
+874495544

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_compact/na-1-bti-TOC.txt

@@ -0,0 +1,8 @@
+Data.db
+Filter.db
+Statistics.db
+CompressionInfo.db
+Partitions.db
+TOC.txt
+Rows.db
+Digest.crc32

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+41575897

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-big-TOC.txt

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

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-bti-Digest.crc32

@@ -0,0 +1 @@
+2912620103

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter/na-1-bti-TOC.txt

@@ -0,0 +1,8 @@
+Data.db
+Filter.db
+Statistics.db
+CompressionInfo.db
+Partitions.db
+TOC.txt
+Rows.db
+Digest.crc32

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+4050499673

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-big-TOC.txt

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

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-bti-Digest.crc32

@@ -0,0 +1 @@
+1995270006

test/data/legacy-sstables/na/legacy_tables/legacy_na_clust_counter_compact/na-1-bti-TOC.txt

@@ -0,0 +1,8 @@
+Data.db
+Filter.db
+Statistics.db
+CompressionInfo.db
+Partitions.db
+TOC.txt
+Rows.db
+Digest.crc32

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+243065670

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-big-TOC.txt

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

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-bti-Digest.crc32

@@ -0,0 +1 @@
+4102718625

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple/na-1-bti-TOC.txt

@@ -0,0 +1,8 @@
+Data.db
+Filter.db
+Statistics.db
+CompressionInfo.db
+Partitions.db
+TOC.txt
+Rows.db
+Digest.crc32

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_compact/na-1-big-Digest.crc32

@@ -0,0 +1 @@
+576034138

test/data/legacy-sstables/na/legacy_tables/legacy_na_simple_compact/na-1-big-TOC.txt

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