CNDB-13639: Adds optional prefetching for known sequential reads (#1692)

This introduces a `PrefetchingRebufferer` that, when enabled, prefetches
(read from the underlying reader) a configurable number of chunks in
advance (so only work on top of rebufferer factories that work with
chunks, i.e. not uncompressed mmap).

Prefetching must first be enabled globally by setting
`-Dcassandra.read_prefetching_size_kb` to the desired value. With that,
and assuming the disk access mode allows it (meaning, it's not
uncompressed mmap), then prefetching will be applied to reads that are
"clearly" sequential. Mostly, as of this patch, this means the sstable
"scanners" and `SortedStringTableCursor`, so compaction, SAI index
building and tools that read sstable fully (scrub, verifier) will
benefit from it.

Since rebufferers are synchronous, prefetching is done through a fixed
thread pool. The number of thread of that pool can be set with
`-Dcassandra.read_prefetching_threads` (but default to the number of
"processors").

The `-Dcassandra.read_prefetching_window` can also be set to define how
often prefetching is re-triggered. By default, when there is less than
half of the value of `-Dcassandra.read_prefetching_size_kb` prefetched,
then prefetching is triggered.

See riptano/cndb#13639 for additional
motivation.

I'll note that this patch is adapted from the similar DSE behavior.
However, there is a fair bit of modification since the DSE version was
relying on the asynchronous nature of rebufferers there (it also had a
few behavior I didn't fully understood and I didn't dig when it felt a
bit specific to DSE).

One point worth mentioning was that the DSE version was relying on the
ability of the underlying channel to create batches when multiple chunks
are prefetched. This is something that could have advantages and we
could try to add back over time, but it's a lot harder to see how to do
that in the non-async world of C*. Typically, in DSE, since all the
prefetches were initiated on the current thread (without blocking it),
building the "batch" was relatively easy, but it doesn't translate in
this patch. Not to mention that currently the underlying channels have
no batching options and those would have to be added (in a way that
trickled down to the channel implementation within CNDB). Anyway, as of
this patch, chunks are prefetched in parallel but with no batching
optimization. Which does mean the `window' parameter is probably not as
useful as for DSE, but I've kept it nonetheless as it's not exactly
adding complexity.

In the context of compaction in CNDB though, I think this mean that it
would kind of be ideal if we could use a relatively large "chunk size"
for the readers: we read the full file anyway, so there is no waste
there to use a large-ish chunk size, and it would kind of provide
batching for prefetching "for free" (in the sense that if we want to
prefetch, say, 128kb in advance, then we only need 2 chunks with a 64kb
chunk, but need a lot more with a 4kb chunk size).

---------

Co-authored-by: Branimir Lambov <[email protected]>

Author: pcmanus

src/java/org/apache/cassandra/cache/ChunkCache.java

@@ -52,6 +52,8 @@
 import org.apache.cassandra.io.util.ChannelProxy;
 import org.apache.cassandra.io.util.ChunkReader;
 import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.PrefetchingRebufferer;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.io.util.Rebufferer;
 import org.apache.cassandra.io.util.RebuffererFactory;
 import org.apache.cassandra.metrics.ChunkCacheMetrics;
@@ -688,6 +690,12 @@ public BufferHolder rebuffer(long position)
             }
         }
 
+        @Override
+        public int chunkSize()
+        {
+            return source.chunkSize();
+        }
+
         @Override
         public Rebufferer instantiateRebufferer()
         {

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -24,6 +24,7 @@
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.compress.AdaptiveCompressor;
 import org.apache.cassandra.io.compress.LZ4Compressor;
+import org.apache.cassandra.io.util.RandomAccessReader;
 import org.apache.cassandra.metrics.TableMetrics;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.sensors.SensorsFactory;
@@ -601,6 +602,29 @@ public enum CassandraRelevantProperties
      */
     SKIP_OPTIMAL_STREAMING_CANDIDATES_CALCULATION("cassandra.skip_optimal_streaming_candidates_calculation", "false"),
 
+    /**
+     * For reads that use prefetching (sequential ones), how much data is prefetced in kilobytes. If the value set is <= 0, then
+     * prefetching is disabled. Also see {@link #READ_PREFETCHING_WINDOW}.
+     * <p>
+     * This is disabled by default because read prefectching is already provided by the OS when working with local disks.
+     * But this is meant for tiered storage extensions, where prefetching may not be provided by the underlying
+     * (custom) "filesystem".
+     */
+    READ_PREFETCHING_SIZE_KB("cassandra.read_prefetching_size_kb", "-1"),
+
+    /**
+     * The window (on the prefetching size) used to triggered prefetching. The prefetching algorithm ensures that at least
+     * {@link #READ_PREFETCHING_WINDOW} * {@link #READ_PREFETCHING_SIZE_KB} prefetching has been requested, and if that's
+     * not the case it requests up to {@link #READ_PREFETCHING_SIZE_KB}.
+     */
+    READ_PREFETCHING_WINDOW("cassandra.read_prefetching_window", "0.5"),
+
+    /**
+     * Number of threads used for read prefetching (when enabled). If unsed, a default based on the number of processors
+     * is used.
+     */
+    READ_PREFETCHING_THREADS("cassandra.read_prefetching_threads"),
+
     /**
      * Allows custom implementation of {@link OperationContext.Factory} to optionally create and configure custom
      * {@link OperationContext} instances.

src/java/org/apache/cassandra/db/compaction/CompactionController.java

@@ -36,6 +36,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.FileDataInput;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.CompactionParams.TombstoneOption;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
@@ -352,6 +353,6 @@ protected boolean ignoreOverlaps()
 
     private FileDataInput openDataFile(SSTableReader reader)
     {
-        return limiter != null ? reader.openDataReader(limiter) : reader.openDataReader();
+        return limiter != null ? reader.openDataReader(limiter, ReadPattern.SEQUENTIAL) : reader.openDataReader(ReadPattern.SEQUENTIAL);
     }
 }

src/java/org/apache/cassandra/db/compaction/Scrubber.java

@@ -64,6 +64,7 @@
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.utils.AbstractIterator;
@@ -157,8 +158,8 @@ public Scrubber(CompactionRealm realm,
         // partition header (key or data size) is corrupt. (This means our position in the index file will be one
         // partition "ahead" of the data file.)
         this.dataFile = isOffline
-                        ? sstable.openDataReader()
-                        : sstable.openDataReader(CompactionManager.instance.getRateLimiter());
+                        ? sstable.openDataReader(ReadPattern.SEQUENTIAL)
+                        : sstable.openDataReader(CompactionManager.instance.getRateLimiter(), ReadPattern.SEQUENTIAL);
 
         try
         {

src/java/org/apache/cassandra/db/compaction/Verifier.java

@@ -63,6 +63,7 @@
 import org.apache.cassandra.io.util.FileInputStreamPlus;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ActiveRepairService;
 import org.apache.cassandra.service.StorageService;
@@ -139,8 +140,8 @@ public Verifier(@Nullable CompactionRealm realm, SSTableReader sstable, OutputHa
 
         this.fileAccessLock = new ReentrantReadWriteLock();
         this.dataFile = isOffline
-                        ? sstable.openDataReader()
-                        : sstable.openDataReader(CompactionManager.instance.getRateLimiter());
+                        ? sstable.openDataReader(ReadPattern.SEQUENTIAL)
+                        : sstable.openDataReader(CompactionManager.instance.getRateLimiter(), ReadPattern.SEQUENTIAL);
         this.verifyInfo = new VerifyInfo(dataFile, sstable, fileAccessLock.readLock());
         this.options = options;
         this.isOffline = isOffline;

src/java/org/apache/cassandra/index/sai/StorageAttachedIndexBuilder.java

@@ -58,6 +58,7 @@
 import org.apache.cassandra.io.sstable.format.RowIndexEntry;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ApproximateTime;
 import org.apache.cassandra.utils.ByteBufferUtil;
@@ -155,7 +156,7 @@ private boolean indexSSTable(SSTableReader sstable, Set<StorageAttachedIndex> in
 
         Set<Component> replacedComponents = new HashSet<>();
 
-        try (RandomAccessReader dataFile = sstable.openDataReader();
+        try (RandomAccessReader dataFile = sstable.openDataReader(ReadPattern.SEQUENTIAL);
              LifecycleTransaction txn = LifecycleTransaction.offline(OperationType.INDEX_BUILD, tracker.metadata))
         {
             perSSTableFileLock = shouldWritePerSSTableFiles(sstable, indexDescriptor, replacedComponents);

src/java/org/apache/cassandra/index/sasi/SASIIndexBuilder.java

@@ -44,6 +44,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.UUIDGen;
@@ -80,7 +81,7 @@ public void build()
             Map<ColumnMetadata, ColumnIndex> indexes = e.getValue();
 
             SSTableWatcher.instance.onIndexBuild(sstable, builtIndexes);
-            try (RandomAccessReader dataFile = sstable.openDataReader())
+            try (RandomAccessReader dataFile = sstable.openDataReader(ReadPattern.SEQUENTIAL))
             {
                 PerSSTableIndexWriter indexWriter = SASIIndex.newWriter(keyValidator, sstable.descriptor, indexes, OperationType.COMPACTION);
 

src/java/org/apache/cassandra/io/sstable/compaction/SortedStringTableCursor.java

@@ -42,6 +42,7 @@
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
@@ -83,17 +84,17 @@ public class SortedStringTableCursor implements SSTableCursor
 
     public SortedStringTableCursor(SSTableReader sstable)
     {
-        this(sstable, sstable.openDataReader(), null);
+        this(sstable, sstable.openDataReader(ReadPattern.SEQUENTIAL), null);
     }
 
     public SortedStringTableCursor(SSTableReader sstable, Range<Token> range)
     {
-        this(sstable, sstable.openDataReader(), range);
+        this(sstable, sstable.openDataReader(ReadPattern.SEQUENTIAL), range);
     }
 
     public SortedStringTableCursor(SSTableReader sstable, Range<Token> tokenRange, RateLimiter limiter)
     {
-        this(sstable, sstable.openDataReader(limiter), tokenRange);
+        this(sstable, sstable.openDataReader(limiter, ReadPattern.SEQUENTIAL), tokenRange);
     }
 
     public SortedStringTableCursor(SSTableReader sstable, RandomAccessReader dataFile, Range<Token> tokenRange)

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

@@ -125,6 +125,7 @@
 import org.apache.cassandra.io.util.FileOutputStreamPlus;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.io.util.SliceDescriptor;
 import org.apache.cassandra.metrics.RestorableMeter;
 import org.apache.cassandra.schema.CachingParams;
@@ -1966,7 +1967,9 @@ public ISSTableScanner getScanner(Iterator<AbstractBounds<PartitionPosition>> bo
 
     public FileDataInput getFileDataInput(long position)
     {
-        return dfile.createReader(position);
+        // While `FileDataInput` supports a `seek` method in practive, it's an interface predominently made for
+        // sequential access. If random access on the returned input is desired, caller should use `#openDataReader`
+        return dfile.createReader(position, ReadPattern.SEQUENTIAL);
     }
 
     /**
@@ -2321,21 +2324,21 @@ public StatsMetadata getSSTableMetadata()
         return sstableMetadata;
     }
 
-    public RandomAccessReader openDataReader(RateLimiter limiter)
+    public RandomAccessReader openDataReader(RateLimiter limiter, ReadPattern accessPattern)
     {
         assert limiter != null;
-        return dfile.createReader(limiter);
+        return dfile.createReader(limiter, accessPattern);
     }
 
-    public RandomAccessReader openDataReader()
+    public RandomAccessReader openDataReader(ReadPattern accessPattern)
     {
-        return dfile.createReader();
+        return dfile.createReader(accessPattern);
     }
 
-    public RandomAccessReader openIndexReader()
+    public RandomAccessReader openIndexReader(ReadPattern accessPattern)
     {
         if (ifile != null)
-            return ifile.createReader();
+            return ifile.createReader(accessPattern);
         return null;
     }
 

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

@@ -31,6 +31,7 @@
 import org.apache.cassandra.io.sstable.ISSTableScanner;
 import org.apache.cassandra.io.sstable.SSTableIdentityIterator;
 import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.ReadPattern;
 import org.apache.cassandra.schema.TableMetadata;
 
 import static org.apache.cassandra.io.sstable.format.SSTableReader.PartitionPositionBounds;
@@ -71,7 +72,7 @@ public SSTableSimpleScanner(SSTableReader sstable,
     {
         assert sstable != null;
 
-        this.dfile = sstable.openDataReader();
+        this.dfile = sstable.openDataReader(ReadPattern.SEQUENTIAL);
         this.sstable = sstable;
         this.sizeInBytes = boundsList.stream().mapToLong(ppb -> ppb.upperPosition - ppb.lowerPosition).sum();
         this.compressedSizeInBytes = sstable.compression ? sstable.onDiskSizeForPartitionPositions(boundsList) : sizeInBytes;