diff --git a/CHANGES.txt b/CHANGES.txt index 97ac3d1cb2..2710ed306c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -19,6 +19,7 @@ Merged from 2.2: * (Hadoop) fix splits calculation (CASSANDRA-10640) * (Hadoop) ensure that Cluster instances are always closed (CASSANDRA-10058) Merged from 2.1: + * Create compression chunk for sending file only (CASSANDRA-10680) * Forbid compact clustering column type changes in ALTER TABLE (CASSANDRA-8879) * Reject incremental repair with subrange repair (CASSANDRA-10422) * Add a nodetool command to refresh size_estimates (CASSANDRA-9579) diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index ae482ff608..101f72284e 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -240,6 +240,36 @@ public class CompressionMetadata return new Chunk(chunkOffset, (int) (nextChunkOffset - chunkOffset - 4)); // "4" bytes reserved for checksum } + /** + * @param sections Collection of sections in uncompressed file. Should not contain sections that overlap each other. + * @return Total chunk size in bytes for given sections including checksum. + */ + public long getTotalSizeForSections(Collection> sections) + { + long size = 0; + long lastOffset = -1; + for (Pair section : sections) + { + int startIndex = (int) (section.left / parameters.chunkLength()); + int endIndex = (int) (section.right / parameters.chunkLength()); + endIndex = section.right % parameters.chunkLength() == 0 ? endIndex - 1 : endIndex; + for (int i = startIndex; i <= endIndex; i++) + { + long offset = i * 8L; + long chunkOffset = chunkOffsets.getLong(offset); + if (chunkOffset > lastOffset) + { + lastOffset = chunkOffset; + long nextChunkOffset = offset + 8 == chunkOffsetsSize + ? compressedFileLength + : chunkOffsets.getLong(offset + 8); + size += (nextChunkOffset - chunkOffset); + } + } + } + return size; + } + /** * @param sections Collection of sections in uncompressed file * @return Array of chunks which corresponds to given sections of uncompressed file, sorted by chunk offset diff --git a/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java b/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java index 123b983c9d..e1e13b7bde 100644 --- a/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java +++ b/src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java @@ -41,7 +41,7 @@ import org.apache.cassandra.utils.UUIDSerializer; */ public class FileMessageHeader { - public static IVersionedSerializer serializer = new FileMessageHeaderSerializer(); + public static FileMessageHeaderSerializer serializer = new FileMessageHeaderSerializer(); public final UUID cfId; public final int sequenceNumber; @@ -52,7 +52,13 @@ public class FileMessageHeader public final SSTableFormat.Type format; public final long estimatedKeys; public final List> sections; + /** + * Compression info for SSTable to send. Can be null if SSTable is not compressed. + * On sender, this field is always null to avoid holding large number of Chunks. + * Use compressionMetadata instead. + */ public final CompressionInfo compressionInfo; + private final CompressionMetadata compressionMetadata; public final long repairedAt; public final int sstableLevel; public final SerializationHeader.Component header; @@ -75,11 +81,41 @@ public class FileMessageHeader this.estimatedKeys = estimatedKeys; this.sections = sections; this.compressionInfo = compressionInfo; + this.compressionMetadata = null; this.repairedAt = repairedAt; this.sstableLevel = sstableLevel; this.header = header; } + public FileMessageHeader(UUID cfId, + int sequenceNumber, + Version version, + SSTableFormat.Type format, + long estimatedKeys, + List> sections, + CompressionMetadata compressionMetadata, + long repairedAt, + int sstableLevel, + SerializationHeader.Component header) + { + this.cfId = cfId; + this.sequenceNumber = sequenceNumber; + this.version = version; + this.format = format; + this.estimatedKeys = estimatedKeys; + this.sections = sections; + this.compressionInfo = null; + this.compressionMetadata = compressionMetadata; + this.repairedAt = repairedAt; + this.sstableLevel = sstableLevel; + this.header = header; + } + + public boolean isCompressed() + { + return compressionInfo != null || compressionMetadata != null; + } + /** * @return total file size to transfer in bytes */ @@ -92,6 +128,10 @@ public class FileMessageHeader for (CompressionMetadata.Chunk chunk : compressionInfo.chunks) size += chunk.length + 4; // 4 bytes for CRC } + else if (compressionMetadata != null) + { + size = compressionMetadata.getTotalSizeForSections(sections); + } else { for (Pair section : sections) @@ -110,7 +150,7 @@ public class FileMessageHeader sb.append(", format: ").append(format); sb.append(", estimated keys: ").append(estimatedKeys); sb.append(", transfer size: ").append(size()); - sb.append(", compressed?: ").append(compressionInfo != null); + sb.append(", compressed?: ").append(isCompressed()); sb.append(", repairedAt: ").append(repairedAt); sb.append(", level: ").append(sstableLevel); sb.append(')'); @@ -134,9 +174,9 @@ public class FileMessageHeader return result; } - static class FileMessageHeaderSerializer implements IVersionedSerializer + static class FileMessageHeaderSerializer { - public void serialize(FileMessageHeader header, DataOutputPlus out, int version) throws IOException + public CompressionInfo serialize(FileMessageHeader header, DataOutputPlus out, int version) throws IOException { UUIDSerializer.serializer.serialize(header.cfId, out, version); out.writeInt(header.sequenceNumber); @@ -156,12 +196,17 @@ public class FileMessageHeader out.writeLong(section.left); out.writeLong(section.right); } - CompressionInfo.serializer.serialize(header.compressionInfo, out, version); + // construct CompressionInfo here to avoid holding large number of Chunks on heap. + CompressionInfo compressionInfo = null; + if (header.compressionMetadata != null) + compressionInfo = new CompressionInfo(header.compressionMetadata.getChunksForSections(header.sections), header.compressionMetadata.parameters); + CompressionInfo.serializer.serialize(compressionInfo, out, version); out.writeLong(header.repairedAt); out.writeInt(header.sstableLevel); if (version >= StreamMessage.VERSION_30) SerializationHeader.serializer.serialize(header.version, header.header, out); + return compressionInfo; } public FileMessageHeader deserialize(DataInputPlus in, int version) throws IOException diff --git a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java index 19f9e126a9..d881d43cd1 100644 --- a/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java +++ b/src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java @@ -42,7 +42,7 @@ public class IncomingFileMessage extends StreamMessage { DataInputPlus input = new DataInputStreamPlus(Channels.newInputStream(in)); FileMessageHeader header = FileMessageHeader.serializer.deserialize(input, version); - StreamReader reader = header.compressionInfo == null ? new StreamReader(header, session) + StreamReader reader = !header.isCompressed() ? new StreamReader(header, session) : new CompressedStreamReader(header, session); try diff --git a/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java index 82e662032f..f10b42e1d8 100644 --- a/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java +++ b/src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java @@ -62,19 +62,13 @@ public class OutgoingFileMessage extends StreamMessage SSTableReader sstable = ref.get(); filename = sstable.getFilename(); - CompressionInfo compressionInfo = null; - if (sstable.compression) - { - CompressionMetadata meta = sstable.getCompressionMetadata(); - compressionInfo = new CompressionInfo(meta.getChunksForSections(sections), meta.parameters); - } this.header = new FileMessageHeader(sstable.metadata.cfId, sequenceNumber, sstable.descriptor.version, sstable.descriptor.formatType, estimatedKeys, sections, - compressionInfo, + sstable.compression ? sstable.getCompressionMetadata() : null, repairedAt, keepSSTableLevel ? sstable.getSSTableLevel() : 0, sstable.header == null ? null : sstable.header.toComponent()); @@ -87,13 +81,13 @@ public class OutgoingFileMessage extends StreamMessage return; } - FileMessageHeader.serializer.serialize(header, out, version); + CompressionInfo compressionInfo = FileMessageHeader.serializer.serialize(header, out, version); final SSTableReader reader = ref.get(); - StreamWriter writer = header.compressionInfo == null ? + StreamWriter writer = compressionInfo == null ? new StreamWriter(reader, header.sections, session) : new CompressedStreamWriter(reader, header.sections, - header.compressionInfo, session); + compressionInfo, session); writer.write(out); } diff --git a/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java b/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java index db05a3e800..5646592bd0 100644 --- a/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java +++ b/test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java @@ -34,6 +34,8 @@ import org.apache.cassandra.streaming.compress.CompressionInfo; import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.Pair; +import static org.junit.Assert.assertEquals; + /** */ public class CompressedInputStreamTest @@ -83,6 +85,11 @@ public class CompressedInputStreamTest sections.add(Pair.create(position, position + 8)); } CompressionMetadata.Chunk[] chunks = comp.getChunksForSections(sections); + long totalSize = comp.getTotalSizeForSections(sections); + long expectedSize = 0; + for (CompressionMetadata.Chunk c : chunks) + expectedSize += c.length + 4; + assertEquals(expectedSize, totalSize); // buffer up only relevant parts of file int size = 0;