Support 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: 1MiB (default; aligned up to FS block
   size, auto-grown to fit a worst-case compressed chunk)

The path is gated by config, table compression being enabled, and an
OperationType allowlist in DataComponent. The allowlist is exhaustive
over OperationType: any new value left unclassified fails static
initialization.

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

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 Ariel Weisberg, Dmitry Konstantinov for CASSANDRA-21134

Author: samueldlightfoot

conf/cassandra.yaml

@@ -693,6 +693,25 @@ 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, cleanup, repair, etc.). The allowed values are:
+# - standard: use buffered I/O (default)
+# - direct: use direct I/O, bypassing the OS page cache
+# Only applies to compressed tables. Uncompressed tables always use buffered I/O.
+# Memtable flushes always use buffered I/O regardless of this setting, as flushed
+# data typically benefits from page cache for subsequent reads (workloads that
+# rarely read recently-flushed data, e.g. write-heavy time series, may see less
+# benefit from caching flush output).
+# background_write_disk_access_mode: standard
+
+# Size of the in-memory staging buffer for Direct IO background writes. Trades off syscall
+# frequency against per-flush blocking latency on the compaction thread.
+# Aligned up to filesystem block size; auto-expands to fit a single compressed chunk + CRC
+# + one block when chunk_length exceeds this value.
+# Allocated per concurrent background writer (e.g. one per active compaction thread,
+# streaming session, etc.), so total off-heap usage scales with concurrency.
+# direct_write_buffer_size: 1MiB
+
 # 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

@@ -700,6 +700,21 @@ commitlog_disk_access_mode: auto
 # - 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, cleanup, repair, etc.). The allowed values are:
+# - standard: use buffered I/O (default)
+# - direct: use direct I/O, bypassing the OS page cache
+# Only applies to compressed tables. Uncompressed tables always use buffered I/O.
+# 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: direct
+
+# Size of the in-memory staging buffer for Direct IO background writes. Trades off syscall
+# frequency against per-flush blocking latency on the compaction thread.
+# Aligned up to filesystem block size; auto-expands to fit a single compressed chunk + CRC
+# + one block when chunk_length exceeds this value.
+# direct_write_buffer_size: 1MiB
+
 # 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,18 @@ public MemtableOptions()
 
     public DataStorageSpec.IntKibibytesBound compressed_read_ahead_buffer_size = new DataStorageSpec.IntKibibytesBound("256KiB");
 
+    // Direct IO for background SSTable writes (compaction, streaming, 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;
+
+    // Size of the in-memory staging buffer for Direct IO background writes. Trades off syscall
+    // frequency against per-flush blocking latency on the compaction thread.
+    // Aligned up to filesystem block size; auto-expands to fit a single compressed chunk + CRC
+    // + one block when chunk_length exceeds this value.
+    public DataStorageSpec.IntKibibytesBound direct_write_buffer_size = new DataStorageSpec.IntKibibytesBound("1MiB");
+
     // 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;
 
@@ -1275,8 +1287,7 @@ public enum DiskAccessMode
         legacy,
 
         /**
-         * Direct-I/O is enabled for commitlog disk only.
-         * When adding support for direct IO, update {@link org.apache.cassandra.service.StartupChecks#checkKernelBug1057843}
+         * When adding support for Direct I/O, update {@link org.apache.cassandra.service.StartupChecks#checkKernelBug1057843}
          */
         direct
     }

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,8 @@ 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 (conf.memtable_flush_writers == 0)
         {
             conf.memtable_flush_writers = conf.data_file_directories.length == 1 ? 2 : 1;
@@ -3406,6 +3410,71 @@ public static void initializeCommitLogDiskAccessMode()
         commitLogWriteDiskAccessMode = accessModeDirectIoPair.left;
     }
 
+    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;
+    }
+
+    /**
+     * The directories opened with O_DIRECT for writing. Startup checks that must reason about O_DIRECT
+     * write targets (e.g. the kernel-bug 1057843 check) derive their path set from here.
+     */
+    public static Set<Path> getDirectIOWritePaths()
+    {
+        Set<Path> paths = new HashSet<>();
+
+        if (getCommitLogWriteDiskAccessMode() == DiskAccessMode.direct)
+            paths.add(new File(getCommitLogLocation()).toPath());
+
+        if (getBackgroundWriteDiskAccessMode() == DiskAccessMode.direct)
+            for (String dataDir : getAllDataFileLocations())
+                paths.add(new File(dataDir).toPath());
+
+        return paths;
+    }
+
+    @VisibleForTesting
+    public static void initializeBackgroundWriteDiskAccessMode()
+    {
+        DiskAccessMode providedMode = conf.background_write_disk_access_mode;
+
+        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);
+
+            // Create the data directories up front (as we do for the commit log) so the kernel-bug 1057843
+            // startup check can stat each O_DIRECT write target. Direct I/O support is validated separately
+            // by the directio_support startup check.
+            if (!toolInitialized)
+                for (String dataDir : getAllDataFileLocations())
+                    PathUtils.createDirectoriesIfNotExists(new File(dataDir).toPath());
+        }
+        else if (providedMode != DiskAccessMode.standard)
+        {
+            throw new ConfigurationException("Unsupported disk access mode for background_write_disk_access_mode " +
+                                             "(options: standard/direct): " + providedMode, false);
+        }
+
+        backgroundWriteDiskAccessMode = providedMode;
+    }
+
     public static String getSavedCachesLocation()
     {
         return conf.saved_caches_directory;

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

@@ -0,0 +1,55 @@
+/*
+ * 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 never constructs a data-file writer, so the direct-IO question does not apply.
+     * Exists so a consumer's classification can be total over its operation enum rather than partial.
+     */
+    NOT_A_WRITER;
+
+    public boolean isSupported()
+    {
+        return this == SUPPORTED;
+    }
+}