Merge branch 'cassandra-2.1' into trunk

Conflicts:
	src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
	src/java/org/apache/cassandra/io/util/CompressedPoolingSegmentedFile.java
	src/java/org/apache/cassandra/io/util/RandomAccessReader.java
	src/java/org/apache/cassandra/io/util/SegmentedFile.java

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)

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();
 

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;
+        if (overrideLength <= 0)
-        return metadataWriter.open(originalSize, chunkOffset, finishType);
+            return metadataWriter.open(originalSize, chunkOffset, isFinal ? FINAL : SHARED_FINAL);
+        return metadataWriter.open(overrideLength, chunkOffset, SHARED);
     }
 
     @Override

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;
+            // i.e. FinishType == EARLY; we will use the RefCountedMemory in possibly multiple instances
-            if (finishType.isFinal)
+            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
+                case FINAL: case SHARED_FINAL:
-                offsets = this.offsets.copy(count * 8L);
+                    // maybe resize the data
-                this.offsets.unreference();
+                    if (this.offsets.size() != count * 8L)
-            }
+                    {
-            else
+                        offsets = this.offsets.copy(count * 8L);
-            {
+                        // release our reference to the original shared data;
-                offsets = this.offsets;
+                        // 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);
         }

src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java

@@ -1257,11 +1257,7 @@ public abstract class SSTableReader extends SSTable implements RefCounted<SSTabl
 
             long left = getPosition(leftBound, Operator.GT).position;
             long right = (rightBound.compareTo(last) > 0)
-                         ? (openReason == OpenReason.EARLY
+                         ? uncompressedLength()
-                            // 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())
                          : getPosition(rightBound, Operator.GT).position;
 
             if (left == right)

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 ifile = iwriter.builder.complete(link.filenameFor(Component.PRIMARY_INDEX), boundary.indexLength);
-        SegmentedFile dfile = dbuilder.complete(link.filenameFor(Component.DATA), FinishType.EARLY);
+        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 ifile = iwriter.builder.complete(desc.filenameFor(Component.PRIMARY_INDEX), finishType.isFinal);
-        SegmentedFile dfile = dbuilder.complete(desc.filenameFor(Component.DATA), finishType);
+        SegmentedFile dfile = dbuilder.complete(desc.filenameFor(Component.DATA), finishType.isFinal);
         SSTableReader sstable = SSTableReader.internalOpen(desc.asType(Descriptor.Type.FINAL),
                                                            components,
                                                            this.metadata,

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

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

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

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

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<Long> 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];
 

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)

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, long overrideLength, boolean useDirectBuffer, PoolingSegmentedFile owner) throws FileNotFoundException
-    protected RandomAccessReader(File file, int bufferSize, 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();
+            try
-        }
+            {
-        catch (IOException e)
+                fileLength = channel.size();
-        {
+            }
-            throw new FSReadError(e, filePath);
+            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)

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

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