diff --git a/CHANGES.txt b/CHANGES.txt index 3dd723a538..aa04be3fd9 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -58,6 +58,7 @@ 2.1.4 + * Ensure SSTableReader.last corresponds exactly with the file end (CASSANDRA-8750) * Make SSTableWriter.openEarly more robust and obvious (CASSANDRA-8747) * Enforce SSTableReader.first/last (CASSANDRA-8744) * Cleanup SegmentedFile API (CASSANDRA-8749) diff --git a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java index d6482997bb..e61b00e8cf 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java @@ -80,7 +80,7 @@ public class CompressedRandomAccessReader extends RandomAccessReader protected CompressedRandomAccessReader(String dataFilePath, CompressionMetadata metadata, PoolingSegmentedFile owner) throws FileNotFoundException { - super(new File(dataFilePath), metadata.chunkLength(), metadata.compressor().useDirectOutputByteBuffers(), owner); + super(new File(dataFilePath), metadata.chunkLength(), metadata.compressedFileLength, metadata.compressor().useDirectOutputByteBuffers(), owner); this.metadata = metadata; checksum = new Adler32(); diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java index 0f3599dc97..7b72105bbc 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java @@ -33,6 +33,10 @@ import org.apache.cassandra.io.util.DataIntegrityMetadata; import org.apache.cassandra.io.util.FileMark; import org.apache.cassandra.io.util.SequentialWriter; +import static org.apache.cassandra.io.compress.CompressionMetadata.Writer.OpenType.FINAL; +import static org.apache.cassandra.io.compress.CompressionMetadata.Writer.OpenType.SHARED; +import static org.apache.cassandra.io.compress.CompressionMetadata.Writer.OpenType.SHARED_FINAL; + public class CompressedSequentialWriter extends SequentialWriter { private final DataIntegrityMetadata.ChecksumWriter crcMetadata; @@ -142,10 +146,11 @@ public class CompressedSequentialWriter extends SequentialWriter runPostFlush.run(); } - public CompressionMetadata open(SSTableWriter.FinishType finishType) + public CompressionMetadata open(long overrideLength, boolean isFinal) { - assert finishType != SSTableWriter.FinishType.NORMAL || current == originalSize; - return metadataWriter.open(originalSize, chunkOffset, finishType); + if (overrideLength <= 0) + return metadataWriter.open(originalSize, chunkOffset, isFinal ? FINAL : SHARED_FINAL); + return metadataWriter.open(overrideLength, chunkOffset, SHARED); } @Override diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index 890aa0a280..55de306935 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -323,18 +323,56 @@ public class CompressionMetadata } } - public CompressionMetadata open(long dataLength, long compressedLength, SSTableWriter.FinishType finishType) + static enum OpenType { - RefCountedMemory offsets; - if (finishType.isFinal) + // i.e. FinishType == EARLY; we will use the RefCountedMemory in possibly multiple instances + SHARED, + // i.e. FinishType == EARLY, but the sstable has been completely written, so we can + // finalise the contents and size of the memory, but must retain a reference to it + SHARED_FINAL, + // i.e. FinishType == NORMAL or FINISH_EARLY, i.e. we have actually finished writing the table + // and will never need to open the metadata again, so we can release any references to it here + FINAL + } + + public CompressionMetadata open(long dataLength, long compressedLength, OpenType type) + { + RefCountedMemory offsets = this.offsets; + int count = this.count; + switch (type) { - // we now know how many offsets we have and can resize the offsets properly - offsets = this.offsets.copy(count * 8L); - this.offsets.unreference(); - } - else - { - offsets = this.offsets; + 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 + this.offsets.unreference(); + } + // null out our reference to the original shared data to catch accidental reuse + 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; + this.offsets.reference(); + } + break; + + case SHARED: + + // 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()); + // 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 * 8); + break; + + default: + throw new AssertionError(); } return new CompressionMetadata(filePath, parameters, offsets, count * 8L, dataLength, compressedLength); } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index 35bb109303..aaf19b2850 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -1257,11 +1257,7 @@ public abstract class SSTableReader extends SSTable implements RefCounted 0) - ? (openReason == OpenReason.EARLY - // if opened early, we overlap with the old sstables by one key, so we know that the last - // (and further) key(s) will be streamed from these if necessary - ? getPosition(last, Operator.GT, false, true).position - : uncompressedLength()) + ? uncompressedLength() : getPosition(rightBound, Operator.GT).position; if (left == right) diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index 89ecd9965d..d411faffba 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java @@ -341,8 +341,8 @@ public class BigTableWriter extends SSTableWriter assert boundary.indexLength > 0 && boundary.dataLength > 0; Descriptor link = makeTmpLinks(); // open the reader early, giving it a FINAL descriptor type so that it is indistinguishable for other consumers - SegmentedFile ifile = iwriter.builder.complete(link.filenameFor(Component.PRIMARY_INDEX), FinishType.EARLY); - SegmentedFile dfile = dbuilder.complete(link.filenameFor(Component.DATA), FinishType.EARLY); + SegmentedFile ifile = iwriter.builder.complete(link.filenameFor(Component.PRIMARY_INDEX), boundary.indexLength); + SegmentedFile dfile = dbuilder.complete(link.filenameFor(Component.DATA), boundary.dataLength); SSTableReader sstable = SSTableReader.internalOpen(descriptor.asType(Descriptor.Type.FINAL), components, metadata, partitioner, ifile, @@ -378,8 +378,8 @@ public class BigTableWriter extends SSTableWriter desc = makeTmpLinks(); // finalize in-memory state for the reader - SegmentedFile ifile = iwriter.builder.complete(desc.filenameFor(Component.PRIMARY_INDEX), finishType); - SegmentedFile dfile = dbuilder.complete(desc.filenameFor(Component.DATA), finishType); + SegmentedFile ifile = iwriter.builder.complete(desc.filenameFor(Component.PRIMARY_INDEX), finishType.isFinal); + SegmentedFile dfile = dbuilder.complete(desc.filenameFor(Component.DATA), finishType.isFinal); SSTableReader sstable = SSTableReader.internalOpen(desc.asType(Descriptor.Type.FINAL), components, this.metadata, diff --git a/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java index 6e65834e91..372da0328e 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedPoolingSegmentedFile.java @@ -45,9 +45,10 @@ public class BufferedPoolingSegmentedFile extends PoolingSegmentedFile // only one segment in a standard-io file } - public SegmentedFile complete(String path, SSTableWriter.FinishType finishType) + public SegmentedFile complete(String path, long overrideLength, boolean isFinal) { - long length = new File(path).length(); + assert !isFinal || overrideLength <= 0; + long length = overrideLength > 0 ? overrideLength : new File(path).length(); return new BufferedPoolingSegmentedFile(path, length); } } diff --git a/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java index 6eb687682f..c31a88b5b4 100644 --- a/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/BufferedSegmentedFile.java @@ -52,9 +52,10 @@ public class BufferedSegmentedFile extends SegmentedFile // only one segment in a standard-io file } - public SegmentedFile complete(String path, SSTableWriter.FinishType finishType) + public SegmentedFile complete(String path, long overrideLength, boolean isFinal) { - long length = new File(path).length(); + assert !isFinal || overrideLength <= 0; + long length = overrideLength > 0 ? overrideLength : new File(path).length(); return new BufferedSegmentedFile(path, length); } } diff --git a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java index 3ea0d6bf5b..40a54dc268 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java @@ -23,7 +23,6 @@ import org.apache.cassandra.io.compress.CompressedRandomAccessReader; import org.apache.cassandra.io.compress.CompressedSequentialWriter; import org.apache.cassandra.io.compress.CompressedThrottledReader; import org.apache.cassandra.io.compress.CompressionMetadata; -import org.apache.cassandra.io.sstable.format.SSTableWriter; public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile implements ICompressedFile { @@ -68,9 +67,9 @@ public class CompressedPoolingSegmentedFile extends PoolingSegmentedFile impleme // only one segment in a standard-io file } - public SegmentedFile complete(String path, SSTableWriter.FinishType finishType) + public SegmentedFile complete(String path, long overrideLength, boolean isFinal) { - return new CompressedPoolingSegmentedFile(path, metadata(path, finishType)); + return new CompressedPoolingSegmentedFile(path, metadata(path, overrideLength, isFinal)); } } diff --git a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java index 545b7995b3..b37ef1e936 100644 --- a/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/CompressedSegmentedFile.java @@ -73,17 +73,18 @@ public class CompressedSegmentedFile extends SegmentedFile implements ICompresse // only one segment in a standard-io file } - protected CompressionMetadata metadata(String path, SSTableWriter.FinishType finishType) + protected CompressionMetadata metadata(String path, long overrideLength, boolean isFinal) { if (writer == null) return CompressionMetadata.create(path); - return writer.open(finishType); + return writer.open(overrideLength, isFinal); } - public SegmentedFile complete(String path, SSTableWriter.FinishType finishType) + public SegmentedFile complete(String path, long overrideLength, boolean isFinal) { - return new CompressedSegmentedFile(path, metadata(path, finishType)); + assert !isFinal || overrideLength <= 0; + return new CompressedSegmentedFile(path, metadata(path, overrideLength, isFinal)); } } diff --git a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java index a863548ded..eec6581553 100644 --- a/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/MmappedSegmentedFile.java @@ -183,31 +183,38 @@ public class MmappedSegmentedFile extends SegmentedFile } } - public SegmentedFile complete(String path, SSTableWriter.FinishType finishType) + public SegmentedFile complete(String path, long overrideLength, boolean isFinal) { - long length = new File(path).length(); + assert !isFinal || overrideLength <= 0; + long length = overrideLength > 0 ? overrideLength : new File(path).length(); // create the segments - return new MmappedSegmentedFile(path, length, createSegments(path)); + return new MmappedSegmentedFile(path, length, createSegments(path, length)); } - private Segment[] createSegments(String path) + private Segment[] createSegments(String path, long length) { RandomAccessFile raf; - long length; try { raf = new RandomAccessFile(path, "r"); - length = raf.length(); } catch (IOException e) { throw new RuntimeException(e); } + // if we're early finishing a range that doesn't span multiple segments, but the finished file now does, + // we remove these from the end (we loop incase somehow this spans multiple segments, but that would + // be a loco dataset + while (length < boundaries.get(boundaries.size() - 1)) + boundaries.remove(boundaries.size() -1); + // add a sentinel value == length List boundaries = new ArrayList<>(this.boundaries); if (length != boundaries.get(boundaries.size() - 1)) boundaries.add(length); + + int segcount = boundaries.size() - 1; Segment[] segments = new Segment[segcount]; diff --git a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java b/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java index d3c90c706e..4ab98afad5 100644 --- a/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/PoolingSegmentedFile.java @@ -67,7 +67,7 @@ public abstract class PoolingSegmentedFile extends SegmentedFile protected RandomAccessReader createPooledReader() { - return RandomAccessReader.open(new File(path), this); + return RandomAccessReader.open(new File(path), length, this); } public void recycle(RandomAccessReader reader) diff --git a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java index 6bff3784c2..1f2d395324 100644 --- a/src/java/org/apache/cassandra/io/util/RandomAccessReader.java +++ b/src/java/org/apache/cassandra/io/util/RandomAccessReader.java @@ -53,10 +53,9 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp protected RandomAccessReader(File file, int bufferSize, PoolingSegmentedFile owner) throws FileNotFoundException { - this(file, bufferSize, false, owner); + this(file, bufferSize, -1, false, owner); } - - protected RandomAccessReader(File file, int bufferSize, boolean useDirectBuffer, PoolingSegmentedFile owner) throws FileNotFoundException + protected RandomAccessReader(File file, int bufferSize, long overrideLength, boolean useDirectBuffer, PoolingSegmentedFile owner) throws FileNotFoundException { this.owner = owner; @@ -76,14 +75,19 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp throw new IllegalArgumentException("bufferSize must be positive"); // we can cache file length in read-only mode - try + long fileLength = overrideLength; + if (fileLength <= 0) { - fileLength = channel.size(); - } - catch (IOException e) - { - throw new FSReadError(e, filePath); + try + { + fileLength = channel.size(); + } + catch (IOException e) + { + throw new FSReadError(e, filePath); + } } + this.fileLength = fileLength; buffer = allocateBuffer(bufferSize, useDirectBuffer); buffer.limit(0); } @@ -96,22 +100,32 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp : ByteBuffer.allocateDirect(size); } - public static RandomAccessReader open(File file, PoolingSegmentedFile owner) + public static RandomAccessReader open(File file, long overrideSize, PoolingSegmentedFile owner) { - return open(file, DEFAULT_BUFFER_SIZE, owner); + return open(file, DEFAULT_BUFFER_SIZE, overrideSize, owner); } public static RandomAccessReader open(File file) { - return open(file, DEFAULT_BUFFER_SIZE, null); + return open(file, -1L); + } + + public static RandomAccessReader open(File file, long overrideSize) + { + return open(file, DEFAULT_BUFFER_SIZE, overrideSize, null); } @VisibleForTesting static RandomAccessReader open(File file, int bufferSize, PoolingSegmentedFile owner) + { + return open(file, bufferSize, -1L, owner); + } + + private static RandomAccessReader open(File file, int bufferSize, long overrideSize, PoolingSegmentedFile owner) { try { - return new RandomAccessReader(file, bufferSize, owner); + return new RandomAccessReader(file, bufferSize, overrideSize, false, owner); } catch (IOException e) { @@ -143,12 +157,16 @@ public class RandomAccessReader extends AbstractDataInput implements FileDataInp try { channel.position(bufferOffset); // setting channel position - while (buffer.hasRemaining()) + long limit = bufferOffset; + while (buffer.hasRemaining() && limit < fileLength) { int n = channel.read(buffer); if (n < 0) break; + limit = bufferOffset + buffer.position(); } + if (limit > fileLength) + buffer.position((int)(fileLength - bufferOffset)); buffer.flip(); } catch (IOException e) diff --git a/src/java/org/apache/cassandra/io/util/SegmentedFile.java b/src/java/org/apache/cassandra/io/util/SegmentedFile.java index 8f760da62b..1341fa948b 100644 --- a/src/java/org/apache/cassandra/io/util/SegmentedFile.java +++ b/src/java/org/apache/cassandra/io/util/SegmentedFile.java @@ -25,13 +25,13 @@ import java.nio.MappedByteBuffer; import java.util.Iterator; import java.util.NoSuchElementException; +import com.google.common.base.Throwables; import com.google.common.util.concurrent.RateLimiter; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.io.FSReadError; import org.apache.cassandra.io.compress.CompressedSequentialWriter; -import org.apache.cassandra.io.sstable.format.SSTableWriter; import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.concurrent.RefCounted; import org.apache.cassandra.utils.concurrent.SharedCloseableImpl; @@ -96,13 +96,13 @@ public abstract class SegmentedFile extends SharedCloseableImpl public RandomAccessReader createReader() { - return RandomAccessReader.open(new File(path)); + return RandomAccessReader.open(new File(path), length); } public RandomAccessReader createThrottledReader(RateLimiter limiter) { assert limiter != null; - return ThrottledReader.open(new File(path), limiter); + return ThrottledReader.open(new File(path), length, limiter); } public FileDataInput getSegment(long position) @@ -156,11 +156,21 @@ public abstract class SegmentedFile extends SharedCloseableImpl * Called after all potential boundaries have been added to apply this Builder to a concrete file on disk. * @param path The file on disk. */ - public abstract SegmentedFile complete(String path, SSTableWriter.FinishType openType); + protected abstract SegmentedFile complete(String path, long overrideLength, boolean isFinal); public SegmentedFile complete(String path) { - return complete(path, SSTableWriter.FinishType.NORMAL); + return complete(path, -1, true); + } + + public SegmentedFile complete(String path, boolean isFinal) + { + return complete(path, -1, isFinal); + } + + public SegmentedFile complete(String path, long overrideLength) + { + return complete(path, overrideLength, false); } public void serializeBounds(DataOutput out) throws IOException diff --git a/src/java/org/apache/cassandra/io/util/ThrottledReader.java b/src/java/org/apache/cassandra/io/util/ThrottledReader.java index b9b645a4cf..570d77713c 100644 --- a/src/java/org/apache/cassandra/io/util/ThrottledReader.java +++ b/src/java/org/apache/cassandra/io/util/ThrottledReader.java @@ -30,9 +30,9 @@ public class ThrottledReader extends RandomAccessReader { private final RateLimiter limiter; - protected ThrottledReader(File file, RateLimiter limiter) throws FileNotFoundException + protected ThrottledReader(File file, long overrideLength, RateLimiter limiter) throws FileNotFoundException { - super(file, RandomAccessReader.DEFAULT_BUFFER_SIZE, null); + super(file, RandomAccessReader.DEFAULT_BUFFER_SIZE, overrideLength, false, null); this.limiter = limiter; } @@ -42,11 +42,11 @@ public class ThrottledReader extends RandomAccessReader super.reBuffer(); } - public static ThrottledReader open(File file, RateLimiter limiter) + public static ThrottledReader open(File file, long overrideLength, RateLimiter limiter) { try { - return new ThrottledReader(file, limiter); + return new ThrottledReader(file, overrideLength, limiter); } catch (FileNotFoundException e) {