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

Merge branch 'cassandra-2.1' into trunk 

Conflicts:
	src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
	test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
This commit is contained in:
Benedict Elliott Smith 2015-03-04 16:31:59 +00:00
parent 5346ce700c a1e2978f9f
commit 36bd31d057
4 changed files with 134 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
View File

@ -204,6 +204,10 @@ public class CompressedRandomAccessReader extends RandomAccessReader
// buffer offset is always aligned
bufferOffset = position & ~(buffer.capacity() - 1);
buffer.position((int) (position - bufferOffset));
// the length() can be provided at construction time, to override the true (uncompressed) length of the file;
// this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length
if (bufferOffset + buffer.limit() > length())
buffer.limit((int)(length() - bufferOffset));
}
catch (CorruptBlockException e)
{
@ -269,6 +273,10 @@ public class CompressedRandomAccessReader extends RandomAccessReader
// buffer offset is always aligned
bufferOffset = position & ~(buffer.capacity() - 1);
buffer.position((int) (position - bufferOffset));
// the length() can be provided at construction time, to override the true (uncompressed) length of the file;
// this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length
if (bufferOffset + buffer.limit() > length())
buffer.limit((int)(length() - bufferOffset));
}
catch (CorruptBlockException e)
{

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

@ -59,6 +59,9 @@ import org.apache.cassandra.utils.Pair;
*/
public class CompressionMetadata
{
// dataLength can represent either the true length of the file
// or some shorter value, in the case we want to impose a shorter limit on readers
// (when early opening, we want to ensure readers cannot read past fully written sections)
public final long dataLength;
public final long compressedFileLength;
private final Memory chunkOffsets;
@ -331,33 +334,39 @@ public class CompressionMetadata
public CompressionMetadata open(long dataLength, long compressedLength, OpenType type)
{
SafeMemory offsets = this.offsets;
SafeMemory offsets;
int count = this.count;
switch (type)
{
case FINAL: case SHARED_FINAL:
// maybe resize the data
if (this.offsets.size() != count * 8L)
{
offsets = this.offsets.copy(count * 8L);
// release our reference to the original shared data;
// we don't do this if not resizing since we must pass out existing
// reference onto our caller
// finalize the size of memory used if it won't now change;
// unnecessary if already correct size
SafeMemory tmp = this.offsets.copy(count * 8L);
this.offsets.free();
this.offsets = tmp;
}
// null out our reference to the original shared data to catch accidental reuse
// note that since noone is writing to this Writer while we open it, null:ing out this.offsets is safe
this.offsets = null;
if (type == OpenType.SHARED_FINAL)
// we will use the data again, so stash our resized data back, and take an extra reference to it
this.offsets = offsets.sharedCopy();
{
offsets = this.offsets.sharedCopy();
}
else
{
offsets = this.offsets;
// null out our reference to the original shared data to catch accidental reuse
// note that since noone is writing to this Writer while we open it, null:ing out this.offsets is safe
this.offsets = null;
}
break;
case SHARED:
offsets = this.offsets.sharedCopy();
// we should only be opened on a compression data boundary; truncate our size to this boundary
assert dataLength % parameters.chunkLength() == 0;
count = (int) (dataLength / parameters.chunkLength());
if (dataLength % parameters.chunkLength() != 0)
count++;
// grab our actual compressed length from the next offset from our the position we're opened to
if (count < this.count)
compressedLength = offsets.getLong(count * 8L);
@ -411,7 +420,8 @@ public class CompressionMetadata
public void abort()
{
offsets.close();
if (offsets != null)
offsets.close();
}
}

2
src/java/org/apache/cassandra/io/util/RandomAccessReader.java
View File

@ -47,6 +47,8 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp
// channel linked with the file, used to retrieve data and force updates.
protected final FileChannel channel;
// this can be overridden at construction to a value shorter than the true length of the file;
// if so, it acts as an imposed limit on reads, rather than a convenience property
private final long fileLength;
protected final PoolingSegmentedFile owner;

154
test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.io.sstable;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.ThreadLocalRandom;
import javax.annotation.Nullable;
@ -212,30 +213,34 @@ public class SSTableRewriterTest extends SchemaLoader
for (int i = 0; i < 100; i++)
cf.addColumn(Util.cellname(i), ByteBuffer.allocate(1000), 1);
File dir = cfs.directories.getDirectoryForNewSSTables();
SSTableWriter writer = getWriter(cfs, dir);
for (int i = 0; i < 500; i++)
writer.append(StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(i)), cf);
SSTableReader s = writer.openEarly(1000);
assertFileCounts(dir.list(), 2, 3);
for (int i = 500; i < 1000; i++)
writer.append(StorageService.getPartitioner().decorateKey(ByteBufferUtil.bytes(i)), cf);
SSTableReader s2 = writer.openEarly(1000);
assertTrue(s != s2);
assertFileCounts(dir.list(), 2, 3);
s.markObsolete();
s.selfRef().release();
s2.selfRef().release();
Thread.sleep(1000);
assertFileCounts(dir.list(), 0, 3);
writer.abort();
Thread.sleep(1000);
int datafiles = assertFileCounts(dir.list(), 0, 0);
assertEquals(datafiles, 0);
validateCFS(cfs);
try
{
for (int i = 0; i < 1000; i++)
writer.append(StorageService.getPartitioner().decorateKey(random(i, 10)), cf);
SSTableReader s = writer.openEarly(1000);
assertFileCounts(dir.list(), 2, 2);
for (int i = 1000; i < 2000; i++)
writer.append(StorageService.getPartitioner().decorateKey(random(i, 10)), cf);
SSTableReader s2 = writer.openEarly(1000);
assertTrue(s.last.compareTo(s2.last) < 0);
assertFileCounts(dir.list(), 2, 2);
s.markObsolete();
s.selfRef().release();
s2.selfRef().release();
Thread.sleep(1000);
assertFileCounts(dir.list(), 0, 2);
writer.abort();
Thread.sleep(1000);
int datafiles = assertFileCounts(dir.list(), 0, 0);
assertEquals(datafiles, 0);
validateCFS(cfs);
}
catch (Throwable t)
{
writer.abort();
throw t;
}
}
@Test
@ -316,17 +321,23 @@ public class SSTableRewriterTest extends SchemaLoader
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
}
List<SSTableReader> sstables = rewriter.finish();
assertEquals(files, sstables.size());
assertEquals(files, cfs.getSSTables().size());
assertEquals(1, cfs.getDataTracker().getView().shadowed.size());
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
assertEquals(files, cfs.getSSTables().size());
assertEquals(0, cfs.getDataTracker().getView().shadowed.size());
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
catch (Throwable t)
{
rewriter.abort();
throw t;
}
List<SSTableReader> sstables = rewriter.finish();
assertEquals(files, sstables.size());
assertEquals(files, cfs.getSSTables().size());
assertEquals(1, cfs.getDataTracker().getView().shadowed.size());
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
assertEquals(files, cfs.getSSTables().size());
assertEquals(0, cfs.getDataTracker().getView().shadowed.size());
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
@ -431,6 +442,11 @@ public class SSTableRewriterTest extends SchemaLoader
{
test.run(scanner, controller, s, cfs, rewriter);
}
catch (Throwable t)
{
rewriter.abort();
throw t;
}
Thread.sleep(1000);
assertEquals(startSize, cfs.metric.liveDiskSpaceUsed.getCount());
@ -463,7 +479,7 @@ public class SSTableRewriterTest extends SchemaLoader
while(scanner.hasNext())
{
rewriter.append(new LazilyCompactedRow(controller, Arrays.asList(scanner.next())));
if (rewriter.currentWriter().getOnDiskFilePointer() > 25000000)
if (rewriter.currentWriter().getFilePointer() > 25000000)
{
rewriter.switchWriter(getWriter(cfs, s.descriptor.directory));
files++;
@ -477,11 +493,17 @@ public class SSTableRewriterTest extends SchemaLoader
break;
}
}
Thread.sleep(1000);
assertEquals(files - 1, cfs.getSSTables().size()); // we never wrote anything to the last file
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
catch (Throwable t)
{
rewriter.abort();
throw t;
}
Thread.sleep(1000);
assertEquals(files - 1, cfs.getSSTables().size()); // we never wrote anything to the last file
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
@Test
@ -513,14 +535,20 @@ public class SSTableRewriterTest extends SchemaLoader
assertEquals(cfs.getSSTables().size(), files); // we have one original file plus the ones we have switched out.
}
}
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
cfs.truncateBlocking();
Thread.sleep(1000); // make sure the deletion tasks have run etc
validateCFS(cfs);
}
catch (Throwable t)
{
rewriter.abort();
throw t;
}
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
cfs.truncateBlocking();
Thread.sleep(1000); // make sure the deletion tasks have run etc
validateCFS(cfs);
}
@Test
@ -552,14 +580,20 @@ public class SSTableRewriterTest extends SchemaLoader
files++;
}
}
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
assertEquals(files, sstables.size());
assertEquals(files, cfs.getSSTables().size());
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
catch (Throwable t)
{
rewriter.abort();
throw t;
}
List<SSTableReader> sstables = rewriter.finish();
cfs.getDataTracker().markCompactedSSTablesReplaced(compacting, sstables, OperationType.COMPACTION);
assertEquals(files, sstables.size());
assertEquals(files, cfs.getSSTables().size());
Thread.sleep(1000);
assertFileCounts(s.descriptor.directory.list(), 0, 0);
validateCFS(cfs);
}
@Test
public void testSSTableSplit() throws InterruptedException
@ -739,7 +773,7 @@ public class SSTableRewriterTest extends SchemaLoader
{
ArrayBackedSortedColumns cf = ArrayBackedSortedColumns.factory.create(cfs.metadata);
for (int i = 0; i < count / 100; i++)
cf.addColumn(Util.cellname(i), ByteBuffer.allocate(1000), 1);
cf.addColumn(Util.cellname(i), random(0, 1000), 1);
File dir = cfs.directories.getDirectoryForNewSSTables();
String filename = cfs.getTempSSTablePath(dir);
@ -781,7 +815,9 @@ public class SSTableRewriterTest extends SchemaLoader
int datacount = 0;
for (String f : files)
{
if (f.contains("tmplink-"))
if (f.endsWith("-CRC.db"))
continue;
if (f.contains("-tmplink-"))
tmplinkcount++;
else if (f.contains("tmp-"))
tmpcount++;
@ -798,4 +834,14 @@ public class SSTableRewriterTest extends SchemaLoader
String filename = cfs.getTempSSTablePath(directory);
return SSTableWriter.create(filename, 0, 0);
}
private ByteBuffer random(int i, int size)
{
byte[] bytes = new byte[size + 4];
ThreadLocalRandom.current().nextBytes(bytes);
ByteBuffer r = ByteBuffer.wrap(bytes);
r.putInt(0, i);
return r;
}
}