CASSANDRA-21094 Advise sequential access on descriptors owned by bulk SSTable scans and streaming

CHANGES.txt

@@ -1,4 +1,5 @@
 7.0
+ * Advise buffered background SSTable scans and whole-file streaming reads as sequential on Linux; bulk scans now use a private descriptor that bypasses the shared chunk cache (CASSANDRA-21094)
  * Allow CQLSSTableWriter to specify SSTable id generator to use (CASSANDRA-21012)
  * Reject LIKE patterns with a wildcard (%) anywhere other than the start or end (CASSANDRA-21068)
  * Support pluggable default role initialization (CASSANDRA-21546)

src/java/org/apache/cassandra/db/streaming/CassandraCompressedStreamWriter.java

@@ -34,6 +34,7 @@
 import org.apache.cassandra.streaming.StreamSession;
 import org.apache.cassandra.streaming.StreamingDataOutputPlus;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.NativeLibrary;
 
 /**
  * CassandraStreamWriter for compressed SSTable.
@@ -68,6 +69,10 @@ public void write(StreamingDataOutputPlus out) throws IOException
             // we want to send continuous chunks together to minimise reads from disk and network writes
             List<Section> sections = fuseAdjacentChunks(compressionInfo.chunks());
 
+            // Sequential readahead overshoots the end of every section, so only advise a whole-file stream.
+            if (sections.size() == 1 && isWholeFileSection(sections.get(0).start, sections.get(0).end, fc.size()))
+                NativeLibrary.trySetSequential(fc.getFileDescriptor(), fc.filePath());
+
             int sectionIdx = 0;
 
             // stream each of the required sections of the file

src/java/org/apache/cassandra/db/streaming/CassandraStreamWriter.java

@@ -39,6 +39,7 @@
 import org.apache.cassandra.streaming.StreamingDataOutputPlus;
 import org.apache.cassandra.streaming.async.StreamCompressionSerializer;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.NativeLibrary;
 import org.apache.cassandra.utils.memory.BufferPools;
 
 import static org.apache.cassandra.net.MessagingService.current_version;
@@ -83,6 +84,10 @@ public void write(StreamingDataOutputPlus out) throws IOException
         try(ChannelProxy proxy = sstable.getDataChannel().newChannel();
             ChecksumValidator validator = sstable.maybeGetChecksumValidator())
         {
+            // Sequential readahead overshoots the end of every section, so only advise a whole-file stream.
+            SSTableReader.PartitionPositionBounds only = sections.size() == 1 ? sections.iterator().next() : null;
+            if (only != null && isWholeFileSection(only.lowerPosition, only.upperPosition, proxy.size()))
+                NativeLibrary.trySetSequential(proxy.getFileDescriptor(), proxy.filePath());
             int bufferSize = validator == null ? DEFAULT_CHUNK_SIZE: validator.chunkSize;
 
             // setting up data compression stream
@@ -127,6 +132,12 @@ protected long totalSize()
         return totalSize;
     }
 
+    /** True when a single stream section spans the whole file: the case worth advising as sequential. */
+    protected static boolean isWholeFileSection(long start, long end, long fileSize)
+    {
+        return start == 0 && end == fileSize;
+    }
+
     /**
      * Sequentially read bytes from the file and write them to the output stream
      *

src/java/org/apache/cassandra/db/streaming/ComponentContext.java

@@ -31,6 +31,7 @@
 import org.apache.cassandra.io.sstable.SSTable;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.NativeLibrary;
 
 public class ComponentContext implements AutoCloseable
 {
@@ -80,6 +81,7 @@ public FileChannel channel(Descriptor descriptor, Component component, long size
 
         assert size == channel.size() : String.format("Entire sstable streaming expects %s file size to be %s but got %s.",
                                                       component, size, channel.size());
+        NativeLibrary.trySetSequential(NativeLibrary.getfd(channel), toTransfer.path());
         return channel;
     }
 

src/java/org/apache/cassandra/io/sstable/SSTableCursorReader.java

@@ -608,7 +608,7 @@ private SSTableCursorReader(SSTableReader reader, TableMetadata metadata, Ref<SS
         serializationHeader = reader.header;
         sstableHasDroppedColumns = anyDroppedColumn(deserializationHelper, serializationHeader);
 
-        dataReader = reader.openDataReaderForScan(diskAccessMode);
+        dataReader = reader.openDataReaderForBulkScan(diskAccessMode);
         // the HEADER decides whether this sstable can contain static rows: after
         // ALTER TABLE ... DROP of the last static column, current metadata has no static
         // columns but older sstables legitimately still carry static rows

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

@@ -1421,42 +1421,58 @@ public StatsMetadata getSSTableMetadata()
 
     public RandomAccessReader openDataReader()
     {
-        return openDataReaderInternal(null, null, false);
+        return openDataReaderInternal(null, null, false, false);
     }
 
     public RandomAccessReader openDataReader(RateLimiter limiter)
     {
         assert limiter != null;
-        return openDataReaderInternal(null, limiter, false);
+        return openDataReaderInternal(null, limiter, false, false);
     }
 
     public RandomAccessReader openDataReader(DiskAccessMode diskAccessMode)
     {
-        return openDataReaderInternal(diskAccessMode, null, false);
+        return openDataReaderInternal(diskAccessMode, null, false, false);
     }
 
     public RandomAccessReader openDataReaderForScan()
     {
-        return openDataReaderInternal(null, null, true);
+        return openDataReaderInternal(null, null, true, false);
     }
 
-    public RandomAccessReader openDataReaderForScan(DiskAccessMode diskAccessMode)
+    /**
+     * A background bulk scan (compaction, cleanup, streaming, offline tools) reads the file once, so it may take
+     * a private descriptor advised as sequential. User range reads must use {@link #openDataReaderForScan()}.
+     */
+    public RandomAccessReader openDataReaderForBulkScan(DiskAccessMode diskAccessMode)
     {
-        return openDataReaderInternal(diskAccessMode, null, true);
+        return openDataReaderInternal(diskAccessMode, null, true, true);
     }
 
     private RandomAccessReader openDataReaderInternal(@Nullable DiskAccessMode diskAccessMode,
                                                       @Nullable RateLimiter limiter,
-                                                      boolean forScan)
-    {
-        if (canReuseDfile(diskAccessMode))
+                                                      boolean forScan,
+                                                      boolean bulkScan)
+    {
+        boolean reuseDfile = canReuseDfile(diskAccessMode);
+        DiskAccessMode effectiveMode = reuseDfile ? dfile.diskAccessMode() : diskAccessMode;
+        // Mmap reads access the file through MmappedRegions, not the descriptor's read()/readahead path, so
+        // fadvise has nothing to act on there; direct I/O bypasses the page cache, so advice is a no-op too.
+        boolean advise = bulkScan && effectiveMode != DiskAccessMode.mmap && effectiveMode != DiskAccessMode.direct;
+        if (reuseDfile && !advise)
             return dfile.createReader(limiter, forScan, OnReaderClose.RETAIN_FILE_OPEN);
 
         FileHandle handle = dfile.toBuilder()
-                                 .withDiskAccessMode(diskAccessMode)
+                                 .withDiskAccessMode(effectiveMode)
+                                 // A temporary handle must not reopen through the writer's closed regions cache,
+                                 // nor own and invalidate the SSTable's shared chunk cache.
+                                 .withMmappedRegionsCache(null)
+                                 .withChunkCache(null)
                                  .complete();
         try
         {
+            if (advise)
+                NativeLibrary.trySetSequential(handle.channel.getFileDescriptor(), handle.path());
             return handle.createReader(limiter, forScan, OnReaderClose.CLOSE_FILE);
         }
         catch (Throwable t)

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

@@ -78,7 +78,7 @@ public SSTableSimpleScanner(SSTableReader sstable,
     {
         assert sstable != null;
 
-        this.dfile = sstable.openDataReaderForScan(diskAccessMode);
+        this.dfile = sstable.openDataReaderForBulkScan(diskAccessMode);
         this.sstable = sstable;
         this.tableMetadata = sstable.metadata();
         this.sizeInBytes = boundsList.stream().mapToLong(ppb -> ppb.upperPosition - ppb.lowerPosition).sum();

src/java/org/apache/cassandra/utils/NativeLibrary.java

@@ -273,6 +273,44 @@ public static void trySkipCache(int fd, long offset, int len, String path)
         }
     }
 
+    /**
+     * Advises the kernel that this descriptor will be read sequentially; affects the open file description,
+     * so advise only an exclusively owned read fd, never one shared with random readers.
+     */
+    public static void trySetSequential(int fd, String path)
+    {
+        if (fd < 0)
+            return;
+
+        try
+        {
+            if (osType == LINUX)
+            {
+                int result = wrappedLibrary.callPosixFadvise(fd, 0, 0, POSIX_FADV_SEQUENTIAL);
+                if (result != 0)
+                    NoSpamLogger.log(
+                            logger,
+                            NoSpamLogger.Level.WARN,
+                            10,
+                            TimeUnit.MINUTES,
+                            "Failed trySetSequential on file: {} Error: " + wrappedLibrary.callStrerror(result).getString(0),
+                            path);
+            }
+        }
+        catch (UnsatisfiedLinkError e)
+        {
+            // if JNA is unavailable just skipping
+        }
+        catch (RuntimeException e)
+        {
+            if (!(e instanceof LastErrorException))
+                throw e;
+
+            NoSpamLogger.log(logger, NoSpamLogger.Level.WARN, 10, TimeUnit.MINUTES,
+                             "posix_fadvise({}, SEQUENTIAL) failed, errno ({}).", fd, errno(e));
+        }
+    }
+
     public static int tryFcntl(int fd, int command, int flags)
     {
         // fcntl return value may or may not be useful, depending on the command