diff --git a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java index e61b00e8cf..b1b4dd4dc2 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java @@ -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) { diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index eef9d0c67b..7237cb9d4e 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -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(); } } diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 1f2d395324..018825910d 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -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; diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java index 018b094720..77c6df97a2 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java @@ -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 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 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 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 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 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 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; + } + }