CASSANDRA-21134: Direct I/O for background SSTable writes

Adds an opt-in O_DIRECT write path for background SSTable producers,
bypassing the OS page cache for data that is unlikely to be re-read
soon after being written. Memtable flushes remain buffered.

Enabled via two new YAML knobs:
 - background_write_disk_access_mode: standard (default) | direct
 - direct_write_buffer_size: 256KiB (default; aligned up to FS block
   size, auto-grown to chunk_length)

The path is gated by config, table compression being enabled, and an
OperationType allowlist in DataComponent. The allowlist is exhaustive:
any new OperationType with writesData=true that is not classified will
fail static initialization.

Operations on the DIO path: COMPACTION, MAJOR_COMPACTION,
TOMBSTONE_COMPACTION, ANTICOMPACTION, GARBAGE_COLLECT, CLEANUP,
UPGRADE_SSTABLES, WRITE, STREAM (chunked receiver only).

Operations off the DIO path:
 - FLUSH (policy: just-flushed data is hot, keep in page cache)
 - SCRUB (correctness: tryAppend needs mark/resetAndTruncate)
 - Zero-Copy Streaming (bypasses DataComponent.buildWriter)
 - Uncompressed writers (only CompressedSequentialWriter has a DIO
   subclass in this change)

StartupChecks fails fast if 'direct' is requested on a platform/FS
that does not support O_DIRECT.

patch by Sam Lightfoot; reviewed by <reviewers> for CASSANDRA-21134

Author: samueldlightfoot

conf/cassandra.yaml

@@ -693,6 +693,21 @@ commitlog_disk_access_mode: legacy
 # - direct: use direct I/O for compaction reads, bypassing the OS page cache
 # compaction_read_disk_access_mode: auto
 
+# Set the disk access mode for writing compressed SSTables during background operations
+# (compaction, streaming, scrub, cleanup, repair, etc.). The allowed values are:
+# - standard: use buffered I/O (default)
+# - direct: use direct I/O, bypassing the OS page cache
+# Note: Only applies to compressed tables. Uncompressed tables always use buffered I/O.
+# Note: Memtable flushes always use buffered I/O regardless of this setting, as flushed
+# data benefits from page cache for subsequent reads.
+# background_write_disk_access_mode: standard
+
+# Preferred buffer size for Direct IO background writes. Will be aligned up to filesystem
+# block size. If a table's compression chunk_length exceeds this value, the buffer will
+# auto-expand to fit. Larger buffers reduce syscall overhead but increase memory usage
+# per compaction thread.
+# direct_write_buffer_size: 256KiB
+
 # 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

@@ -362,6 +362,17 @@ public MemtableOptions()
 
     public DataStorageSpec.IntKibibytesBound compressed_read_ahead_buffer_size = new DataStorageSpec.IntKibibytesBound("256KiB");
 
+    // Direct IO for background SSTable writes (compaction, streaming, scrub, cleanup, etc.)
+    // When 'direct' is set, background writes bypass the OS page cache using O_DIRECT.
+    // Memtable flushes always use buffered I/O regardless of this setting.
+    // Default is 'standard' (buffered I/O) - users must opt-in to Direct IO
+    public DiskAccessMode background_write_disk_access_mode = DiskAccessMode.standard;
+
+    // Preferred buffer size for Direct IO background writes. Will be aligned up to filesystem block size.
+    // If a table's compression chunk_length exceeds this value, the buffer will auto-expand to fit.
+    // Larger buffers reduce syscall overhead but increase memory usage per compaction thread.
+    public DataStorageSpec.IntKibibytesBound direct_write_buffer_size = new DataStorageSpec.IntKibibytesBound("256KiB");
+
     // fraction of free disk space available for compaction after min free space is subtracted
     public volatile Double max_space_usable_for_compactions_in_percentage = .95;
 

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

@@ -224,6 +224,8 @@ public class DatabaseDescriptor
 
     private static DiskAccessMode compactionReadDiskAccessMode;
 
+    private static DiskAccessMode backgroundWriteDiskAccessMode;
+
     private static AbstractCryptoProvider cryptoProvider;
     private static IAuthenticator authenticator;
     private static IAuthorizer authorizer;
@@ -897,6 +899,10 @@ else if (conf.repair_session_space.toMebibytes() > (int) (Runtime.getRuntime().m
         if (conf.hints_directory.equals(conf.saved_caches_directory))
             throw new ConfigurationException("saved_caches_directory must not be the same as the hints_directory", false);
 
+        initializeBackgroundWriteDiskAccessMode();
+        if (backgroundWriteDiskAccessMode != conf.background_write_disk_access_mode)
+            logger.info("background_write_disk_access_mode resolved to: {}", backgroundWriteDiskAccessMode);
+
         if (conf.memtable_flush_writers == 0)
         {
             conf.memtable_flush_writers = conf.data_file_directories.length == 1 ? 2 : 1;
@@ -3406,6 +3412,85 @@ public static void initializeCommitLogDiskAccessMode()
         commitLogWriteDiskAccessMode = accessModeDirectIoPair.left;
     }
 
+    /**
+     * Return background write disk access mode.
+     */
+    public static DiskAccessMode getBackgroundWriteDiskAccessMode()
+    {
+        return backgroundWriteDiskAccessMode;
+    }
+
+    @VisibleForTesting
+    public static void setBackgroundWriteDiskAccessMode(DiskAccessMode diskAccessMode)
+    {
+        backgroundWriteDiskAccessMode = diskAccessMode;
+        conf.background_write_disk_access_mode = diskAccessMode;
+    }
+
+    public static DataStorageSpec.IntKibibytesBound getDirectWriteBufferSize()
+    {
+        return conf.direct_write_buffer_size;
+    }
+
+    @VisibleForTesting
+    public static void initializeBackgroundWriteDiskAccessMode()
+    {
+        DiskAccessMode providedMode = conf.background_write_disk_access_mode;
+
+        // For 'auto', default to standard (conservative, safe default)
+        if (providedMode == DiskAccessMode.auto)
+        {
+            providedMode = DiskAccessMode.standard;
+        }
+
+        // Validate Direct IO is supported on ALL data directories if requested
+        if (providedMode == DiskAccessMode.direct)
+        {
+            // DataStorageSpec already rejects negatives at parse time; zero is the remaining
+            // nonsense value. The writer's Math.max would silently coerce it to minRequiredSize,
+            // which masks a likely operator mistake — fail fast instead.
+            if (conf.direct_write_buffer_size.toBytes() <= 0)
+                throw new ConfigurationException("direct_write_buffer_size must be > 0 when background_write_disk_access_mode is 'direct'. " +
+                                                 "Got: " + conf.direct_write_buffer_size, false);
+
+            // Only check Direct IO support if not running as a tool
+            if (!toolInitialized)
+            {
+                List<String> unsupportedLocations = new ArrayList<>();
+
+                for (String dataDir : conf.data_file_directories)
+                {
+                    try
+                    {
+                        File dataDirFile = new File(dataDir);
+                        PathUtils.createDirectoriesIfNotExists(dataDirFile.toPath());
+
+                        if (!FileUtils.isDirectIOSupported(dataDirFile))
+                        {
+                            unsupportedLocations.add(dataDir);
+                        }
+                    }
+                    catch (RuntimeException e)
+                    {
+                        logger.warn("Unable to determine Direct IO support for data directory {}: {}", dataDir, e.getMessage());
+                        unsupportedLocations.add(dataDir + " (check failed: " + e.getMessage() + ")");
+                    }
+                }
+
+                if (!unsupportedLocations.isEmpty())
+                {
+                    throw new ConfigurationException(
+                        String.format("background_write_disk_access_mode is set to 'direct', but the following data directories " +
+                                      "do not support Direct I/O: %s. Either change background_write_disk_access_mode to 'standard' " +
+                                      "in cassandra.yaml, or ensure all data directories are on filesystems that support Direct I/O.",
+                                      unsupportedLocations), false);
+                }
+            }
+        }
+
+        backgroundWriteDiskAccessMode = providedMode;
+    }
+
     public static String getSavedCachesLocation()
     {
         return conf.saved_caches_directory;

src/java/org/apache/cassandra/io/DirectIoSupport.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.io;
+
+/**
+ * Classifies an operation's eligibility for a direct-IO (O_DIRECT) data path, encoding both
+ * the answer and the rationale class. Consumers maintain their own per-operation classification
+ * and apply this alongside their own gates (e.g. compression, configuration mode);
+ * {@link #SUPPORTED} is necessary but not sufficient.
+ */
+public enum DirectIoSupport
+{
+    /** Eligible for the direct-IO data path. */
+    SUPPORTED,
+
+    /**
+     * The direct-IO path is mechanically incompatible with this operation. Removing this
+     * exclusion requires code changes, not policy.
+     */
+    UNSUPPORTED_CORRECTNESS,
+
+    /**
+     * Direct IO would work, but is deliberately disabled for performance or cache-residency
+     * reasons. Removing this exclusion requires re-evaluating the policy, not code changes.
+     */
+    UNSUPPORTED_POLICY,
+
+    /**
+     * The operation does not exercise this data path (e.g. a sentinel, or an op that does not
+     * read or write through the consuming component). The gate is moot.
+     */
+    NOT_APPLICABLE;
+
+    public boolean isSupported()
+    {
+        return this == SUPPORTED;
+    }
+}