Add Direct IO support for compaction reads

This change adds support for using O_DIRECT (direct I/O) when reading SSTables
during compaction. Direct I/O bypasses the OS page cache, which can be beneficial
for compaction workloads where data is typically read once and not accessed again
soon after.

Key changes:
- Add DiskAccessMode.direct option for scan operations
- Introduce DirectThreadLocalByteBufferHolder and DirectThreadLocalReadAheadBuffer
  for aligned buffer management required by O_DIRECT
- Add ByteBufferHolder interface to abstract buffer management
- Modify FileHandle to support building direct I/O readers via toBuilder()
- Add startup check to verify Direct IO support on configured data directories
- Enable read-ahead for uncompressed data in Direct CompressedChunkReader
- Move direct IO support logic into FileHandle (from SSTableReader)
- Add comprehensive tests for direct I/O chunk readers and buffer holders

patch by Sam Lightfoot; reviewed by Ariel Weisberg and Maxwell Guo for CASSANDRA-19987

Author: samueldlightfoot

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Direct I/O support for compaction reads (CASSANDRA-19987)
  * Support custom StartupCheck implementations via SPI (CASSANDRA-21093)
  * Make sstableexpiredblockers support human-readable output with SSTable sizes (CASSANDRA-20448)
  * Enhance nodetool compactionhistory to report more compaction properities (CASSANDRA-20081)

conf/cassandra.yaml

@@ -688,6 +688,11 @@ commitlog_segment_size: 32MiB
 # The default setting is legacy when the storage compatibility is set to 4 or auto otherwise.
 commitlog_disk_access_mode: legacy
 
+# Set the disk access mode for reading SSTables during compaction. The allowed values are:
+# - auto: inherit from disk_access_mode (default)
+# - direct: use direct I/O for compaction reads, bypassing the OS page cache
+# compaction_read_disk_access_mode: auto
+
 # Compression to apply to SSTables as they flush for compressed tables.
 # Note that tables without compression enabled do not respect this flag.
 #

conf/cassandra_latest.yaml

@@ -695,6 +695,11 @@ commitlog_segment_size: 32MiB
 # The default setting is legacy when the storage compatibility is set to 4 or auto otherwise.
 commitlog_disk_access_mode: auto
 
+# Set the disk access mode for reading SSTables during compaction. The allowed values are:
+# - auto: inherit from disk_access_mode (default)
+# - direct: use direct I/O for compaction reads, bypassing the OS page cache
+# compaction_read_disk_access_mode: auto
+
 # Compression to apply to SSTables as they flush for compressed tables.
 # Note that tables without compression enabled do not respect this flag.
 #

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

@@ -428,6 +428,7 @@ public static class SSTableConfig
     public FlushCompression flush_compression = FlushCompression.fast;
     public int commitlog_max_compression_buffers_in_pool = 3;
     public DiskAccessMode commitlog_disk_access_mode = DiskAccessMode.legacy;
+    public DiskAccessMode compaction_read_disk_access_mode = DiskAccessMode.auto;
     @Replaces(oldName = "periodic_commitlog_sync_lag_block_in_ms", converter = Converters.MILLIS_DURATION_INT, deprecated = true)
     public DurationSpec.IntMillisecondsBound periodic_commitlog_sync_lag_block;
     public TransparentDataEncryptionOptions transparent_data_encryption_options = new TransparentDataEncryptionOptions();

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

@@ -221,6 +221,8 @@ public class DatabaseDescriptor
 
     private static DiskAccessMode commitLogWriteDiskAccessMode;
 
+    private static DiskAccessMode compactionReadDiskAccessMode;
+
     private static AbstractCryptoProvider cryptoProvider;
     private static IAuthenticator authenticator;
     private static IAuthorizer authorizer;
@@ -662,6 +664,21 @@ else if (conf.disk_access_mode == DiskAccessMode.direct)
         }
         logger.info("DiskAccessMode is {}, indexAccessMode is {}", conf.disk_access_mode, indexAccessMode);
 
+        if (DiskAccessMode.auto == conf.compaction_read_disk_access_mode)
+        {
+            compactionReadDiskAccessMode = conf.disk_access_mode;
+        }
+        else if (DiskAccessMode.direct == conf.compaction_read_disk_access_mode)
+        {
+            compactionReadDiskAccessMode = DiskAccessMode.direct;
+        }
+        else
+        {
+            throw new IllegalArgumentException("Unsupported disk access mode for compaction_read_disk_access_mode " +
+                                               "(options: direct/auto) " + conf.compaction_read_disk_access_mode);
+        }
+        logger.info("compaction_read_disk_access_mode resolved to: {}", compactionReadDiskAccessMode);
+
         /* phi convict threshold for FailureDetector */
         if (conf.phi_convict_threshold < 5 || conf.phi_convict_threshold > 16)
         {
@@ -1773,7 +1790,7 @@ private static Pair<DiskAccessMode, Boolean> resolveCommitLogWriteDiskAccessMode
 
                 File commitLogLocationDir = new File(commitLogLocation);
                 PathUtils.createDirectoriesIfNotExists(commitLogLocationDir.toPath());
-                directIOSupported = FileUtils.getBlockSize(commitLogLocationDir) > 0;
+                directIOSupported = FileUtils.isDirectIOSupported(commitLogLocationDir);
             }
             catch (IOError | ConfigurationException ex)
             {
@@ -3301,6 +3318,18 @@ public static void setCommitLogSegmentSize(int sizeMebibytes)
         conf.commitlog_segment_size = new DataStorageSpec.IntMebibytesBound(sizeMebibytes);
     }
 
+    public static DiskAccessMode getCompactionReadDiskAccessMode()
+    {
+        return compactionReadDiskAccessMode;
+    }
+
+    @VisibleForTesting
+    public static void setCompactionReadDiskAccessMode(DiskAccessMode scanDiskAccessMode)
+    {
+        compactionReadDiskAccessMode = scanDiskAccessMode;
+        conf.compaction_read_disk_access_mode = scanDiskAccessMode;
+    }
+
     /**
      * Return commitlog disk access mode.
      */

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

@@ -32,6 +32,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
 import org.apache.cassandra.db.SerializationHeader;
@@ -262,7 +263,7 @@ public ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Ra
         try
         {
             for (SSTableReader sstable : sstables)
-                scanners.add(sstable.getScanner(ranges));
+                scanners.add(sstable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode()));
         }
         catch (Throwable t)
         {

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

@@ -1767,7 +1767,7 @@ public ISSTableScanner getScanner(SSTableReader sstable)
                 {
                     rangesToScan = Collections2.filter(ranges, range -> !transientRanges.contains(range));
                 }
-                return sstable.getScanner(rangesToScan);
+                return sstable.getScanner(rangesToScan, DatabaseDescriptor.getCompactionReadDiskAccessMode());
             }
 
             @Override
@@ -1790,7 +1790,7 @@ public Full(ColumnFamilyStore cfs, Collection<Range<Token>> ranges, long nowInSe
             @Override
             public ISSTableScanner getScanner(SSTableReader sstable)
             {
-                return sstable.getScanner();
+                return sstable.getScanner(DatabaseDescriptor.getCompactionReadDiskAccessMode());
             }
 
             @Override

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

@@ -44,6 +44,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
@@ -343,7 +344,7 @@ public ScannerList getScanners(Collection<SSTableReader> sstables, Collection<Ra
                 {
                     // L0 makes no guarantees about overlapping-ness.  Just create a direct scanner for each
                     for (SSTableReader sstable : byLevel.get(level))
-                        scanners.add(sstable.getScanner(ranges));
+                        scanners.add(sstable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode()));
                 }
                 else
                 {
@@ -445,7 +446,7 @@ public LeveledScanner(TableMetadata metadata, Collection<SSTableReader> sstables
             sstableIterator = this.sstables.iterator();
             assert sstableIterator.hasNext(); // caller should check intersecting first
             SSTableReader currentSSTable = sstableIterator.next();
-            currentScanner = currentSSTable.getScanner(ranges);
+            currentScanner = currentSSTable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode());
         }
 
         @Override
@@ -498,7 +499,7 @@ protected UnfilteredRowIterator computeNext()
                     return endOfData();
                 }
                 SSTableReader currentSSTable = sstableIterator.next();
-                currentScanner = currentSSTable.getScanner(ranges);
+                currentScanner = currentSSTable.getScanner(ranges, DatabaseDescriptor.getCompactionReadDiskAccessMode());
             }
         }
 

src/java/org/apache/cassandra/index/accord/SSTableIndex.java

@@ -76,7 +76,7 @@ public static SSTableIndex create(IndexDescriptor id) throws IOException
     {
         Map<IndexComponent, FileHandle> files = new EnumMap<>(IndexComponent.class);
         for (IndexComponent c : id.getLiveComponents())
-            files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped(true).complete());
+            files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped().complete());
         List<Segment> segments = RouteIndexFormat.readSegments(files);
         files.remove(IndexComponent.SEGMENT).close();
         files.remove(IndexComponent.METADATA).close();

src/java/org/apache/cassandra/index/sai/disk/format/IndexDescriptor.java

@@ -269,7 +269,7 @@ public FileHandle createPerSSTableFileHandle(IndexComponent indexComponent, Thro
             if (logger.isTraceEnabled())
                 logger.trace(logMessage("Opening file handle for {} ({})"), file, FBUtilities.prettyPrintMemory(file.length()));
 
-            return new FileHandle.Builder(file).mmapped(true).complete();
+            return new FileHandle.Builder(file).mmapped().complete();
         }
         catch (Throwable t)
         {
@@ -291,7 +291,7 @@ public FileHandle createPerIndexFileHandle(IndexComponent indexComponent, IndexI
             if (logger.isTraceEnabled())
                 logger.trace(logMessage("Opening file handle for {} ({})"), file, FBUtilities.prettyPrintMemory(file.length()));
 
-            return new FileHandle.Builder(file).mmapped(true).complete();
+            return new FileHandle.Builder(file).mmapped().complete();
         }
         catch (Throwable t)
         {