natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.1' into trunk

This commit is contained in:
Yuki Morishita 2015-11-17 19:31:51 -06:00
parent aa898a0a6a 5be1f663b6
commit aff6994c8f
6 changed files with 93 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

30
src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
View File

@ -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<Pair<Long, Long>> sections)
{
long size = 0;
long lastOffset = -1;
for (Pair<Long, Long> 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

55
src/java/org/apache/cassandra/streaming/messages/FileMessageHeader.java
View File

@ -41,7 +41,7 @@ import org.apache.cassandra.utils.UUIDSerializer;
*/
public class FileMessageHeader
{
public static IVersionedSerializer<FileMessageHeader> 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<Pair<Long, Long>> 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<Pair<Long, Long>> 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<Long, Long> 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<FileMessageHeader>
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

2
src/java/org/apache/cassandra/streaming/messages/IncomingFileMessage.java
View File

@ -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

14
src/java/org/apache/cassandra/streaming/messages/OutgoingFileMessage.java
View File

@ -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);
}

7
test/unit/org/apache/cassandra/streaming/compression/CompressedInputStreamTest.java
View File

@ -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;