Merge branch 'cassandra-6.0' into trunk

* cassandra-6.0:
  Support direct I/O for background SSTable writes

CHANGES.txt

@@ -4,6 +4,7 @@
  * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
  * Add a guardrail for misprepared statements (CASSANDRA-21139)
 Merged from 6.0:
+ * Support direct I/O for background SSTable writes (CASSANDRA-21134)
  * Relax assertion on partitioner instances in SinglePartitionReadCommand (CASSANDRA-21251)
  * Report cancelled read command execution to coordinator as a RequestFailure.TIMEOUT (CASSANDRA-21468)
  * Add policy for selecting CMS host when submitting commit request (CASSANDRA-21456)

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

@@ -364,6 +364,18 @@ public class Config
 
     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;
 
@@ -1304,8 +1316,7 @@ public class Config
         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

@@ -226,6 +226,8 @@ public class DatabaseDescriptor
 
     private static DiskAccessMode compactionReadDiskAccessMode;
 
+    private static DiskAccessMode backgroundWriteDiskAccessMode;
+
     private static AbstractCryptoProvider cryptoProvider;
     private static IAuthenticator authenticator;
     private static IAuthorizer authorizer;
@@ -909,6 +911,8 @@ public class DatabaseDescriptor
         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;
@@ -3423,6 +3427,71 @@ public class DatabaseDescriptor
         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;
+    }
+}

src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java

@@ -22,6 +22,7 @@ import java.io.EOFException;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.Channels;
+import java.nio.file.OpenOption;
 import java.util.Optional;
 import java.util.zip.CRC32;
 
@@ -46,36 +47,51 @@ import static org.apache.cassandra.utils.Throwables.merge;
 
 public class CompressedSequentialWriter extends SequentialWriter
 {
-    private final ChecksumWriter crcMetadata;
+    protected final ChecksumWriter crcMetadata;
 
     // holds offset in the file where current chunk should be written
     // changed only by flush() method where data buffer gets compressed and stored to the file
-    private long chunkOffset = 0;
+    protected long chunkOffset = 0;
 
     // index file writer (random I/O)
-    private final CompressionMetadata.Writer metadataWriter;
+    protected final CompressionMetadata.Writer metadataWriter;
     private final ICompressor compressor;
 
     // used to store compressed data
     private ByteBuffer compressed;
 
     // holds a number of already written chunks
-    private int chunkCount = 0;
+    protected int chunkCount = 0;
 
-    private long uncompressedSize = 0, compressedSize = 0;
+    protected long uncompressedSize = 0;
+    protected long compressedSize = 0;
 
-    private final MetadataCollector sstableMetadataCollector;
+    protected final MetadataCollector sstableMetadataCollector;
     private final CompressionDictionaryManager compressionDictionaryManager;
 
     private final ByteBuffer crcCheckBuffer = ByteBuffer.allocate(4);
-    private final Optional<File> digestFile;
+    protected final Optional<File> digestFile;
 
     private final int maxCompressedLength;
     private final boolean isDictionaryEnabled;
 
+    private static ByteBuffer allocateBuffer(CompressionParams parameters)
+    {
+        return parameters.getSstableCompressor().preferredBufferType().allocate(parameters.chunkLength());
+    }
+
+    private static SequentialWriterOption buildOption(SequentialWriterOption option, CompressionParams parameters)
+    {
+        return SequentialWriterOption.newBuilder()
+                                     .bufferSize(parameters.chunkLength())
+                                     .bufferType(parameters.getSstableCompressor().preferredBufferType())
+                                     .finishOnClose(option.finishOnClose())
+                                     .build();
+    }
+
     public CompressedSequentialWriter(File file,
                                       File offsetsFile,
-                                      File digestFile,
+                                      @Nullable File digestFile,
                                       SequentialWriterOption option,
                                       CompressionParams parameters,
                                       MetadataCollector sstableMetadataCollector)
@@ -83,33 +99,28 @@ public class CompressedSequentialWriter extends SequentialWriter
         this(file, offsetsFile, digestFile, option, parameters, sstableMetadataCollector, null);
     }
 
-
     /**
-     * Create CompressedSequentialWriter without digest file.
+     * Create CompressedSequentialWriter with optional compression dictionary and channel options.
      *
      * @param file File to write
      * @param offsetsFile File to write compression metadata
-     * @param digestFile File to write digest
+     * @param digestFile File to write digest, or null if not needed
      * @param option Write option (buffer size and type will be set the same as compression params)
      * @param parameters Compression parameters
      * @param sstableMetadataCollector Metadata collector
      * @param compressionDictionaryManager manages compression dictionary; null if absent
+     * @param extraOpenOptions additional options to pass to FileChannel.open (e.g., ExtendedOpenOption.DIRECT)
      */
     public CompressedSequentialWriter(File file,
                                       File offsetsFile,
-                                      File digestFile,
+                                      @Nullable File digestFile,
                                       SequentialWriterOption option,
                                       CompressionParams parameters,
                                       MetadataCollector sstableMetadataCollector,
-                                      @Nullable CompressionDictionaryManager compressionDictionaryManager)
+                                      @Nullable CompressionDictionaryManager compressionDictionaryManager,
+                                      OpenOption... extraOpenOptions)
     {
-        super(file, SequentialWriterOption.newBuilder()
-                            .bufferSize(option.bufferSize())
-                            .bufferType(option.bufferType())
-                            .bufferSize(parameters.chunkLength())
-                            .bufferType(parameters.getSstableCompressor().preferredBufferType())
-                            .finishOnClose(option.finishOnClose())
-                            .build());
+        super(file, allocateBuffer(parameters), buildOption(option, parameters), true, extraOpenOptions);
         ICompressor compressor = parameters.getSstableCompressor();
         this.digestFile = Optional.ofNullable(digestFile);
 
@@ -142,7 +153,16 @@ public class CompressedSequentialWriter extends SequentialWriter
         metadataWriter = CompressionMetadata.Writer.open(parameters, offsetsFile, compressionDictionary);
 
         this.sstableMetadataCollector = sstableMetadataCollector;
-        crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel)));
+        crcMetadata = createChecksumWriter();
+    }
+
+    /**
+     * Creates the {@link ChecksumWriter} for the chunk and full-file checksums. Invoked from the constructor,
+     * so overrides must not read subclass fields.
+     */
+    protected ChecksumWriter createChecksumWriter()
+    {
+        return new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel)));
     }
 
     @Override
@@ -178,7 +198,9 @@ public class CompressedSequentialWriter extends SequentialWriter
     @Override
     protected void flushData()
     {
-        seekToChunkStart(); // why is this necessary? seems like it should always be at chunk start in normal operation
+        // resetAndTruncate leaves fchannel.position() past EOF after its verification reads + truncate;
+        // re-seek so the next chunk lands at chunkOffset. No-op under linear writes.
+        seekToChunkStart();
 
         try
         {
@@ -216,25 +238,15 @@ public class CompressedSequentialWriter extends SequentialWriter
         }
         compressedSize += compressedLength;
 
-        try
-        {
-            // write an offset of the newly written chunk to the index file
-            metadataWriter.addOffset(chunkOffset);
-            chunkCount++;
+        // write an offset of the newly written chunk to the index file
+        metadataWriter.addOffset(chunkOffset);
+        chunkCount++;
 
-            // write out the compressed data
-            toWrite.flip();
-            channel.write(toWrite);
+        // write out the compressed data and checksum
+        toWrite.flip();
+        writeChunk(toWrite);
+        lastFlushOffset = uncompressedSize;
 
-            // write corresponding checksum
-            toWrite.rewind();
-            crcMetadata.appendDirect(toWrite, true);
-            lastFlushOffset = uncompressedSize;
-        }
-        catch (IOException e)
-        {
-            throw new FSWriteError(e, getPath());
-        }
         if (toWrite == buffer)
             buffer.position(uncompressedLength);
 
@@ -244,6 +256,20 @@ public class CompressedSequentialWriter extends SequentialWriter
             runPostFlush.accept(getLastFlushOffset());
     }
 
+    protected void writeChunk(ByteBuffer toWrite)
+    {
+        try
+        {
+            channel.write(toWrite);
+            toWrite.rewind();
+            crcMetadata.appendDirect(toWrite, true);
+        }
+        catch (IOException e)
+        {
+            throw new FSWriteError(e, getPath());
+        }
+    }
+
     public CompressionMetadata open(long overrideLength)
     {
         if (overrideLength <= 0)
@@ -358,10 +384,16 @@ public class CompressedSequentialWriter extends SequentialWriter
         }
     }
 
+    protected void writeDigestFile()
+    {
+        digestFile.ifPresent(crcMetadata::writeFullChecksum);
+    }
+
     /**
      * Seek to the offset where next compressed data chunk should be stored.
+     * Subclasses may override if they manage their own channel.
      */
-    private void seekToChunkStart()
+    protected void seekToChunkStart()
     {
         if (getOnDiskFilePointer() != chunkOffset)
         {
@@ -429,7 +461,7 @@ public class CompressedSequentialWriter extends SequentialWriter
         protected void doPrepare()
         {
             syncInternal();
-            digestFile.ifPresent(crcMetadata::writeFullChecksum);
+            writeDigestFile();
             sstableMetadataCollector.addCompressionRatio(compressedSize, uncompressedSize);
             metadataWriter.finalizeLength(current(), chunkCount).prepareToCommit();
         }

src/java/org/apache/cassandra/io/compress/DirectCompressedSequentialWriter.java

@@ -0,0 +1,427 @@
+/*
+ * 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.compress;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.IntConsumer;
+import java.util.function.LongConsumer;
+
+import javax.annotation.Nullable;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.sun.nio.file.ExtendedOpenOption;
+
+import org.agrona.BitUtil;
+import org.agrona.BufferUtil;
+import org.agrona.collections.LongArrayQueue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.compression.CompressionDictionaryManager;
+import org.apache.cassandra.io.FSReadError;
+import org.apache.cassandra.io.FSWriteError;
+import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
+import org.apache.cassandra.io.util.ChecksumWriter;
+import org.apache.cassandra.io.util.DataPosition;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.SequentialWriter;
+import org.apache.cassandra.io.util.SequentialWriterOption;
+import org.apache.cassandra.metrics.StorageMetrics;
+import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.Throwables;
+import org.apache.cassandra.utils.memory.MemoryUtil;
+
+import sun.nio.ch.DirectBuffer;
+
+import static org.apache.cassandra.utils.Throwables.merge;
+
+/**
+ * Uses O_DIRECT to bypass the OS page cache, reducing memory pressure during compaction.
+ * <p>
+ * O_DIRECT requires all writes to be block-aligned, so compressed chunks are accumulated in an aligned buffer.
+ * Only complete blocks are flushed; at finalization, remaining data is padded, written, and the file is
+ * truncated to actual size.
+ */
+public class DirectCompressedSequentialWriter extends CompressedSequentialWriter
+{
+    private static final Logger logger = LoggerFactory.getLogger(DirectCompressedSequentialWriter.class);
+
+    // Warn once per JVM that an undersized buffer was coerced up — operator-visible without per-SSTable spam.
+    private static final AtomicBoolean undersizedBufferWarned = new AtomicBoolean(false);
+
+    private static final int CRC_LENGTH = Integer.BYTES;
+
+    // capacity >= maxChunkWrite + CRC_LENGTH + blockSize: a chunk is staged in a single put, so it must fit
+    // contiguously even after a flush carries over up to blockSize - 1 unaligned bytes.
+    private ByteBuffer writeBuffer;
+    private long actualDataSize = 0;
+
+    private boolean dataFinalized = false;
+
+    // Chunks staged in writeBuffer but not yet on disk, as flat {compressedEnd, uncompressedEnd} pairs in
+    // write order: offer 2 in writeChunk, poll 2 in durableUncompressedOffset, nothing else, or pairing
+    // breaks. Lets the post-flush listener report a durable offset. Offsets are never negative, so -1 is
+    // a safe empty sentinel.
+    private final LongArrayQueue stagedChunkBoundaries = new LongArrayQueue(-1L);
+    private long durableUncompressedOffset = 0;
+
+    private final int blockSize;
+
+    // Bytes registered with StorageMetrics.directWriteBufferBytes at allocation; subtracted at cleanup so the
+    // gauge balances even on the abort path. 0 until writeBuffer is successfully allocated.
+    private int directBufferBytes = 0;
+
+    public DirectCompressedSequentialWriter(File file,
+                                            File offsetsFile,
+                                            @Nullable File digestFile,
+                                            SequentialWriterOption option,
+                                            CompressionParams parameters,
+                                            MetadataCollector sstableMetadataCollector,
+                                            @Nullable CompressionDictionaryManager compressionDictionaryManager)
+    {
+        super(file, offsetsFile, digestFile, option, parameters, sstableMetadataCollector, compressionDictionaryManager, ExtendedOpenOption.DIRECT);
+
+        // super() opened the O_DIRECT FileChannel and allocated parent buffers; if anything below throws
+        // the caller never gets a reference to clean them up, so abort the txn proxy ourselves.
+        try
+        {
+            this.blockSize = FileUtils.getBlockSize(file.parent());
+            if (blockSize <= 0)
+                throw new IllegalStateException("Unable to determine filesystem block size for Direct IO. " +
+                                                "Block size: " + blockSize);
+
+            if (!BitUtil.isPowerOfTwo(blockSize))
+                throw new IllegalStateException("Filesystem block size must be a power of two for Direct IO. " +
+                                                "Block size: " + blockSize);
+
+            int configuredSize = DatabaseDescriptor.getDirectWriteBufferSize().toBytes();
+            int maxChunkWrite = parameters.getSstableCompressor().initialCompressedBufferLength(parameters.chunkLength());
+            int minRequiredSize = maxChunkWrite + CRC_LENGTH + blockSize;
+            if (configuredSize < minRequiredSize && undersizedBufferWarned.compareAndSet(false, true))
+                logger.warn("direct_write_buffer_size ({} bytes) is below the minimum required for SSTable {} " +
+                            "(worst-case chunk {} + CRC 4 + blockSize {} = {} bytes); using the minimum. " +
+                            "Increase direct_write_buffer_size in cassandra.yaml to silence this warning.",
+                            configuredSize, file, maxChunkWrite, blockSize, minRequiredSize);
+            int bufferSize = BitUtil.align(Math.max(configuredSize, minRequiredSize), blockSize);
+
+            this.writeBuffer = BufferUtil.allocateDirectAligned(bufferSize, blockSize);
+            this.directBufferBytes = bufferSize;
+            StorageMetrics.directWriteBufferBytes.inc(bufferSize);
+            StorageMetrics.directWriteBuffersAllocated.mark();
+        }
+        catch (Throwable t)
+        {
+            Throwable merged = t;
+            try { merged = abort(t); }
+            catch (Throwable t2) { t.addSuppressed(t2); }
+            Throwables.maybeFail(merged);
+            // Unreachable: maybeFail(non-null) always throws. Present for definite-assignment of `blockSize`.
+            throw new AssertionError("Throwables.maybeFail should have thrown", merged);
+        }
+    }
+
+    // Invoked from super()'s constructor before writeBuffer exists; safe because we only capture the
+    // writeCrcToAlignedBuffer reference, which isn't called until the first chunk is written.
+    @Override
+    protected ChecksumWriter createChecksumWriter()
+    {
+        return new DirectChecksumWriter(this::writeCrcToAlignedBuffer);
+    }
+
+    // Parent reads fchannel.position(), which lags by the bytes staged in writeBuffer.
+    // getEstimatedOnDiskBytesWritten is intentionally NOT overridden: parent returns chunkOffset,
+    // which already reflects the eventual on-disk size.
+    @Override
+    public long getOnDiskFilePointer()
+    {
+        return actualDataSize;
+    }
+
+    @Override
+    protected void seekToChunkStart()
+    {
+        // No-op: bytes staged in writeBuffer would be skipped by a seek, leaving a hole.
+        // resetAndTruncate (the parent's reason for this seek) is unsupported here.
+    }
+
+    // openFinalEarly() publishes a reader right after sync(), but the inherited sync flushes only whole
+    // blocks — the last chunk's sub-block tail would stay buffered and the reader would short-read.
+    // Finalize first so the reader sees the complete file.
+    @Override
+    protected void syncInternal()
+    {
+        finalizeDataFile();
+        syncDataOnlyInternal();
+    }
+
+    // Brings the file to its final on-disk form exactly once. openFinalEarly's sync and commit's doPrepare
+    // both land here; whichever runs first does the work, the other is a no-op. Safe because the writer is
+    // switched out right after openFinalEarly, so nothing writes once the file is final.
+    private void finalizeDataFile()
+    {
+        if (dataFinalized)
+            return;
+
+        doFlush(0);
+        flushFinalWithPadding();
+        dataFinalized = true;
+    }
+
+    // The parent feeds the post-flush listener getLastFlushOffset() (== uncompressedSize), which counts bytes
+    // staged in writeBuffer that O_DIRECT has not yet written. Report instead the uncompressed offset of the
+    // last chunk whose compressed bytes are fully on disk, so a preemptive reader never short-reads past EOF.
+    @Override
+    public void setPostFlushListener(LongConsumer postFlush)
+    {
+        super.setPostFlushListener(stagedOffset -> postFlush.accept(durableUncompressedOffset()));
+    }
+
+    // fchannel.position() counts only whole blocks written, so chunks ending at or below it are durable.
+    // Chunks flush in order, so draining from the head suffices and the offset is monotonic.
+    private long durableUncompressedOffset()
+    {
+        long onDisk;
+        try
+        {
+            onDisk = fchannel.position();
+        }
+        catch (IOException e)
+        {
+            throw new FSReadError(e, getPath());
+        }
+
+        long compressedEnd;
+        while ((compressedEnd = stagedChunkBoundaries.peekLong()) != stagedChunkBoundaries.nullValue()
+               && compressedEnd <= onDisk)
+        {
+            stagedChunkBoundaries.pollLong();
+            durableUncompressedOffset = stagedChunkBoundaries.pollLong();
+        }
+        return durableUncompressedOffset;
+    }
+
+    @Override
+    protected void writeChunk(ByteBuffer toWrite)
+    {
+        Preconditions.checkState(!dataFinalized, "writeChunk after finalizeDataFile() under O_DIRECT");
+
+        Preconditions.checkArgument(toWrite.position() == 0,
+                                    "writeChunk requires a flipped buffer (position == 0), got position=%s",
+                                    toWrite.position());
+
+        int chunkLength = toWrite.remaining();
+
+        writeToAlignedBuffer(toWrite);
+
+        // writeToAlignedBuffer drained toWrite; rewind so appendDirect can re-read it for the CRC.
+        toWrite.rewind();
+        crcMetadata.appendDirect(toWrite, true);
+
+        actualDataSize = chunkOffset + chunkLength + CRC_LENGTH;
+
+        stagedChunkBoundaries.offerLong(actualDataSize);
+        stagedChunkBoundaries.offerLong(uncompressedSize);
+    }
+
+    private void writeToAlignedBuffer(ByteBuffer data)
+    {
+        int dataLength = data.remaining();
+
+        // Buffer is sized for worst-case chunk + CRC + blockSize, so a flush always frees enough room.
+        if (writeBuffer.position() + dataLength > writeBuffer.capacity())
+            flushCompleteBlocks();
+
+        writeBuffer.put(data);
+    }
+
+    private void writeCrcToAlignedBuffer(int crcValue)
+    {
+        // After flush, leftover < blockSize, so there's always room for the CRC trailer.
+        if (writeBuffer.position() + CRC_LENGTH > writeBuffer.capacity())
+            flushCompleteBlocks();
+
+        writeBuffer.putInt(crcValue);
+    }
+
+    // FileChannel.write may write fewer bytes than requested, and a partial write would short the file
+    // and desync the callers' leftover/truncate bookkeeping.
+    private void writeAlignedBlocks(int flushLimit) throws IOException
+    {
+        writeBuffer.position(0);
+        writeBuffer.limit(flushLimit);
+        while (writeBuffer.hasRemaining())
+            fchannel.write(writeBuffer);
+    }
+
+    private void flushCompleteBlocks()
+    {
+        // writeAlignedBlocks rewinds position() to 0 to drain, so snapshot the cursor first.
+        int logicalPos = writeBuffer.position();
+
+        // Align down: O_DIRECT cannot write partial blocks
+        int flushLimit = logicalPos & -blockSize;
+
+        // Callers flush only on overflow, which the sizing floor (maxChunkWrite + CRC + blockSize)
+        // guarantees cannot happen before a whole block is buffered.
+        Preconditions.checkState(flushLimit > 0,
+                                 "flush requested with no complete block buffered; writeBuffer undersized");
+
+        try
+        {
+            writeAlignedBlocks(flushLimit);
+
+            int leftover = logicalPos - flushLimit;
+            if (leftover > 0)
+            {
+                writeBuffer.limit(logicalPos);
+                writeBuffer.position(flushLimit);
+                writeBuffer.compact();
+            }
+            else
+            {
+                writeBuffer.clear();
+            }
+        }
+        catch (IOException e)
+        {
+            throw new FSWriteError(e, getPath());
+        }
+    }
+
+    private void flushFinalWithPadding()
+    {
+        int logicalPos = writeBuffer.position();
+
+        if (logicalPos == 0)
+            return;
+
+        try
+        {
+            int flushLimit = BitUtil.align(logicalPos, blockSize);
+
+            ByteBufferUtil.writeZeroes(writeBuffer, flushLimit - logicalPos);
+
+            writeAlignedBlocks(flushLimit);
+
+            // O_DIRECT required padding; truncate back to actual data size.
+            fchannel.truncate(actualDataSize);
+        }
+        catch (IOException e)
+        {
+            throw new FSWriteError(e, getPath());
+        }
+    }
+
+    // Gated out for SCRUB in DataComponent.buildWriter; these throws are a canary if the gate is bypassed.
+    @Override
+    public DataPosition mark()
+    {
+        throw new UnsupportedOperationException("mark() not supported under O_DIRECT");
+    }
+
+    @Override
+    public synchronized void resetAndTruncate(DataPosition mark)
+    {
+        throw new UnsupportedOperationException("resetAndTruncate() not supported under O_DIRECT");
+    }
+
+    @VisibleForTesting
+    ByteBuffer writeBuffer()
+    {
+        return writeBuffer;
+    }
+
+    @VisibleForTesting
+    static void resetUndersizedBufferWarned()
+    {
+        undersizedBufferWarned.set(false);
+    }
+
+    protected class DirectTransactionalProxy extends CompressedSequentialWriter.TransactionalProxy
+    {
+        @Override
+        protected void doPrepare()
+        {
+            finalizeDataFile();
+            syncDataOnlyInternal();
+            writeDigestFile();
+            sstableMetadataCollector.addCompressionRatio(compressedSize, uncompressedSize);
+            metadataWriter.finalizeLength(current(), chunkCount).prepareToCommit();
+        }
+
+        @Override
+        protected Throwable doPreCleanup(Throwable accumulate)
+        {
+            if (writeBuffer != null)
+            {
+                try
+                {
+                    // allocateDirectAligned returns a slice; clean the backing buffer via the attachment.
+                    MemoryUtil.clean((ByteBuffer) ((DirectBuffer) writeBuffer).attachment());
+                }
+                catch (Throwable t)
+                {
+                    accumulate = merge(accumulate, t);
+                }
+                StorageMetrics.directWriteBufferBytes.dec(directBufferBytes);
+                writeBuffer = null;
+            }
+
+            return super.doPreCleanup(accumulate);
+        }
+    }
+
+    @Override
+    protected SequentialWriter.TransactionalProxy txnProxy()
+    {
+        return new DirectTransactionalProxy();
+    }
+
+    /**
+     * Routes the per-chunk CRC into the block-aligned writeBuffer instead of the channel, reusing
+     * ChecksumWriter's bookkeeping rather than duplicating it. Only the CRC trailer flows through here;
+     * {@link #writeChunk} stages the chunk data.
+     */
+    private static final class DirectChecksumWriter extends ChecksumWriter
+    {
+        private final IntConsumer alignedSink;
+
+        DirectChecksumWriter(IntConsumer alignedSink)
+        {
+            this.alignedSink = alignedSink;
+        }
+
+        @Override
+        protected void writeIncrementalInt(int value)
+        {
+            alignedSink.accept(value);
+        }
+
+        @Override
+        public void writeChunkSize(int length)
+        {
+            throw new UnsupportedOperationException("writeChunkSize is unused on the compressed O_DIRECT path");
+        }
+    }
+}

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

@@ -18,10 +18,18 @@
 
 package org.apache.cassandra.io.sstable.format;
 
+import java.util.EnumMap;
+
+import com.google.common.annotations.VisibleForTesting;
+
+import org.apache.cassandra.config.Config.DiskAccessMode;
 import org.apache.cassandra.config.Config.FlushCompression;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.compression.CompressionDictionaryManager;
+import org.apache.cassandra.io.DirectIoSupport;
 import org.apache.cassandra.io.compress.CompressedSequentialWriter;
+import org.apache.cassandra.io.compress.DirectCompressedSequentialWriter;
 import org.apache.cassandra.io.compress.ICompressor;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
@@ -32,8 +40,61 @@ import org.apache.cassandra.io.util.SequentialWriterOption;
 import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.TableMetadata;
 
+import static org.apache.cassandra.io.DirectIoSupport.NOT_A_WRITER;
+import static org.apache.cassandra.io.DirectIoSupport.SUPPORTED;
+import static org.apache.cassandra.io.DirectIoSupport.UNSUPPORTED_CORRECTNESS;
+import static org.apache.cassandra.io.DirectIoSupport.UNSUPPORTED_POLICY;
+
 public class DataComponent
 {
+    private static final EnumMap<OperationType, DirectIoSupport> DIRECT_WRITE_SUPPORT = buildDirectWriteSupport();
+
+    private static EnumMap<OperationType, DirectIoSupport> buildDirectWriteSupport()
+    {
+        EnumMap<OperationType, DirectIoSupport> m = new EnumMap<>(OperationType.class);
+
+        // append()-only writers; safe under O_DIRECT.
+        m.put(OperationType.CLEANUP, SUPPORTED);
+        m.put(OperationType.UPGRADE_SSTABLES, SUPPORTED);
+        m.put(OperationType.MAJOR_COMPACTION, SUPPORTED);
+        m.put(OperationType.GARBAGE_COLLECT, SUPPORTED);
+        m.put(OperationType.WRITE, SUPPORTED);
+        m.put(OperationType.ANTICOMPACTION, SUPPORTED);
+        m.put(OperationType.COMPACTION, SUPPORTED);
+        m.put(OperationType.TOMBSTONE_COMPACTION, SUPPORTED);
+        m.put(OperationType.STREAM, SUPPORTED);
+        // writesData==false yet these still reach buildWriter() (RELOCATE via relocateSSTables, UNKNOWN via
+        // the offline sstablesplit tool), which is why classification can't key off writesData. CASSANDRA-21134.
+        m.put(OperationType.RELOCATE, SUPPORTED);
+        m.put(OperationType.UNKNOWN, SUPPORTED);
+
+        // tryAppend() needs mark()/resetAndTruncate(), unsupported under O_DIRECT.
+        m.put(OperationType.SCRUB, UNSUPPORTED_CORRECTNESS);
+
+        // Flushed data is hot; keep it in the page cache.
+        m.put(OperationType.FLUSH, UNSUPPORTED_POLICY);
+
+        // These never construct a data-file writer (read-only, rewrite a non-data component, or write their
+        // own files), so they never reach buildWriter(). Classified only to keep the map total.
+        m.put(OperationType.P0, NOT_A_WRITER);
+        m.put(OperationType.VERIFY, NOT_A_WRITER);
+        m.put(OperationType.VALIDATION, NOT_A_WRITER);
+        m.put(OperationType.INDEX_BUILD, NOT_A_WRITER);
+        m.put(OperationType.VIEW_BUILD, NOT_A_WRITER);
+        m.put(OperationType.INDEX_SUMMARY, NOT_A_WRITER);
+        m.put(OperationType.KEY_CACHE_SAVE, NOT_A_WRITER);
+        m.put(OperationType.ROW_CACHE_SAVE, NOT_A_WRITER);
+        m.put(OperationType.COUNTER_CACHE_SAVE, NOT_A_WRITER);
+
+        // Total over the enum so a newly-added OperationType fails here at class-load instead of throwing
+        // from buildWriter() in production (CASSANDRA-21134: RELOCATE slipped through the old writesData gate).
+        for (OperationType op : OperationType.values())
+            if (!m.containsKey(op))
+                throw new AssertionError("Missing direct-write classification for " + op
+                                         + " — every OperationType must be SUPPORTED, UNSUPPORTED_*, or NOT_A_WRITER");
+        return m;
+    }
+
     public static SequentialWriter buildWriter(Descriptor descriptor,
                                                TableMetadata metadata,
                                                SequentialWriterOption options,
@@ -44,15 +105,29 @@ public class DataComponent
     {
         if (metadata.params.compression.isEnabled())
         {
-            final CompressionParams compressionParams = buildCompressionParams(metadata, operationType, flushCompression);
+            CompressionParams compressionParams = buildCompressionParams(metadata, operationType, flushCompression);
 
-            return new CompressedSequentialWriter(descriptor.fileFor(Components.DATA),
-                                                  descriptor.fileFor(Components.COMPRESSION_INFO),
-                                                  descriptor.fileFor(Components.DIGEST),
-                                                  options,
-                                                  compressionParams,
-                                                  metadataCollector,
-                                                  compressionDictionaryManager);
+            if (DatabaseDescriptor.getBackgroundWriteDiskAccessMode() == DiskAccessMode.direct
+                && isDirectWriteSupported(operationType))
+            {
+                return new DirectCompressedSequentialWriter(descriptor.fileFor(Components.DATA),
+                                                            descriptor.fileFor(Components.COMPRESSION_INFO),
+                                                            descriptor.fileFor(Components.DIGEST),
+                                                            options,
+                                                            compressionParams,
+                                                            metadataCollector,
+                                                            compressionDictionaryManager);
+            }
+            else
+            {
+                return new CompressedSequentialWriter(descriptor.fileFor(Components.DATA),
+                                                      descriptor.fileFor(Components.COMPRESSION_INFO),
+                                                      descriptor.fileFor(Components.DIGEST),
+                                                      options,
+                                                      compressionParams,
+                                                      metadataCollector,
+                                                      compressionDictionaryManager);
+            }
         }
         else
         {
@@ -103,4 +178,14 @@ public class DataComponent
         }
         return compressionParams;
     }
+
+    @VisibleForTesting
+    static boolean isDirectWriteSupported(OperationType operationType)
+    {
+        DirectIoSupport support = DIRECT_WRITE_SUPPORT.get(operationType);
+        if (support == null)
+            throw new IllegalStateException("OperationType " + operationType + " has no direct-write classification");
+        return support.isSupported();
+    }
+
 }

src/java/org/apache/cassandra/io/util/ChecksumWriter.java

@@ -40,6 +40,20 @@ public class ChecksumWriter
         this.incrementalOut = incrementalOut;
     }
 
+    // Subclasses with their own CRC sink (see writeIncrementalInt) use this ctor; they must not call
+    // writeChunkSize, which needs incrementalOut.
+    protected ChecksumWriter()
+    {
+        this.incrementalOut = null;
+    }
+
+    // Seam so subclasses can redirect the per-chunk CRC int without re-implementing appendDirect's
+    // checksum bookkeeping.
+    protected void writeIncrementalInt(int value) throws IOException
+    {
+        incrementalOut.writeInt(value);
+    }
+
     public void writeChunkSize(int length)
     {
         try
@@ -69,7 +83,7 @@ public class ChecksumWriter
             toAppend.reset();
 
             int incrementalChecksumValue = (int) incrementalChecksum.getValue();
-            incrementalOut.writeInt(incrementalChecksumValue);
+            writeIncrementalInt(incrementalChecksumValue);
 
             fullChecksum.update(toAppend);
             if (checksumIncrementalResult)

src/java/org/apache/cassandra/io/util/FileUtils.java

@@ -41,6 +41,7 @@ import java.util.Collections;
 import java.util.EnumSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -783,7 +784,14 @@ public final class FileUtils
         }
     }
 
+    private static final ConcurrentHashMap<String, Integer> blockSizeByDirectory = new ConcurrentHashMap<>();
+
     public static int getBlockSize(File directory)
+    {
+        return blockSizeByDirectory.computeIfAbsent(directory.absolutePath(), ignored -> probeBlockSize(directory));
+    }
+
+    private static int probeBlockSize(File directory)
     {
         File f = FileUtils.createTempFile("block-size-test", ".tmp", directory);
         try

src/java/org/apache/cassandra/io/util/SequentialWriter.java

@@ -20,7 +20,11 @@ package org.apache.cassandra.io.util;
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.FileChannel;
+import java.nio.file.OpenOption;
 import java.nio.file.StandardOpenOption;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
+import java.util.Set;
 import java.util.function.LongConsumer;
 
 import org.apache.cassandra.io.FSReadError;
@@ -109,25 +113,36 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
     }
 
     // TODO: we should specify as a parameter if we permit an existing file or not
-    private static FileChannel openChannel(File file)
+    private static FileChannel openChannel(File file, OpenOption... extraOptions)
     {
         try
         {
+            Set<OpenOption> options = new LinkedHashSet<>(Arrays.asList(StandardOpenOption.READ, StandardOpenOption.WRITE));
+            options.addAll(Arrays.asList(extraOptions));
+
             if (file.exists())
             {
-                return FileChannel.open(file.toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE);
+                return FileChannel.open(file.toPath(), options.toArray(OpenOption[]::new));
             }
             else
             {
-                FileChannel channel = FileChannel.open(file.toPath(), StandardOpenOption.READ, StandardOpenOption.WRITE, StandardOpenOption.CREATE_NEW);
+                options.add(StandardOpenOption.CREATE_NEW);
+                FileChannel channel = FileChannel.open(file.toPath(), options.toArray(OpenOption[]::new));
                 try
                 {
                     SyncUtil.trySyncDir(file.parent());
                 }
                 catch (Throwable t)
                 {
-                    try { channel.close(); }
-                    catch (Throwable t2) { t.addSuppressed(t2); }
+                    try
+                    {
+                        channel.close();
+                    }
+                    catch (Throwable t2)
+                    {
+                        t.addSuppressed(t2);
+                    }
+                    throw t;
                 }
                 return channel;
             }
@@ -170,15 +185,14 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
         this(file, option.allocateBuffer(), option, strictFlushing);
     }
 
-    protected SequentialWriter(File file, ByteBuffer buffer, SequentialWriterOption option, boolean strictFlushing)
+    protected SequentialWriter(File file, ByteBuffer buffer, SequentialWriterOption option, boolean strictFlushing,
+                               OpenOption... extraOpenOptions)
     {
-        super(openChannel(file), buffer);
+        super(openChannel(file, extraOpenOptions), buffer);
         this.strictFlushing = strictFlushing;
-        this.fchannel = (FileChannel)channel;
-
+        this.fchannel = (FileChannel) this.channel;
         this.file = file;
         this.filePath = file.absolutePath();
-
         this.option = option;
     }
 

src/java/org/apache/cassandra/metrics/StorageMetrics.java

@@ -53,6 +53,9 @@ public class StorageMetrics
     public static final Counter startupOpsForInvalidToken = Metrics.counter(factory.createMetricName("StartupOpsForInvalidToken"));
     public static final Meter bootstrapFilesThroughputMetric = Metrics.meter(factory.createMetricName("BootstrapFilesThroughput"));
 
+    public static final Counter directWriteBufferBytes = Metrics.counter(factory.createMetricName("DirectWriteBufferBytes"));
+    public static final Meter directWriteBuffersAllocated = Metrics.meter(factory.createMetricName("DirectWriteBuffersAllocated"));
+
     private static Gauge<Long> createSummingGauge(String name, ToLongFunction<KeyspaceMetrics> extractor)
     {
         return Metrics.register(factory.createMetricName(name),

src/java/org/apache/cassandra/schema/CompressionParams.java

@@ -197,9 +197,15 @@ public final class CompressionParams
 
     @VisibleForTesting
     public static CompressionParams noop()
+    {
+        return noop(DEFAULT_CHUNK_LENGTH);
+    }
+
+    @VisibleForTesting
+    public static CompressionParams noop(int chunkLength)
     {
         NoopCompressor compressor = NoopCompressor.create(Collections.emptyMap());
-        return new CompressionParams(compressor, DEFAULT_CHUNK_LENGTH, Integer.MAX_VALUE, DEFAULT_MIN_COMPRESS_RATIO, Collections.emptyMap());
+        return new CompressionParams(compressor, chunkLength, Integer.MAX_VALUE, DEFAULT_MIN_COMPRESS_RATIO, Collections.emptyMap());
     }
 
     public CompressionParams(String sstableCompressorClass, Map<String, String> otherOptions, int chunkLength, double minCompressRatio) throws ConfigurationException

src/java/org/apache/cassandra/service/StartupChecks.java

@@ -258,11 +258,7 @@ public class StartupChecks
             if (!FBUtilities.isLinux)
                 return;
 
-            Set<Path> directIOWritePaths = new HashSet<>();
-            if (DatabaseDescriptor.getCommitLogWriteDiskAccessMode() == Config.DiskAccessMode.direct)
-                directIOWritePaths.add(new File(DatabaseDescriptor.getCommitLogLocation()).toPath());
-            // Note: Data directories for direct IO compaction reads are checked in checkDirectIOSupport.
-            // This check is specifically for direct IO writes which are currently only supported for commit log.
+            Set<Path> directIOWritePaths = DatabaseDescriptor.getDirectIOWritePaths();
 
             if (!directIOWritePaths.isEmpty() && IGNORE_KERNEL_BUG_1057843_CHECK.getBoolean())
             {
@@ -887,21 +883,27 @@ public class StartupChecks
             if (configuration.isDisabled(name()))
                 return;
 
-            // Only check if compaction_read_disk_access_mode is direct
-            if (DatabaseDescriptor.getCompactionReadDiskAccessMode() != Config.DiskAccessMode.direct)
+            boolean directReads = DatabaseDescriptor.getCompactionReadDiskAccessMode() == Config.DiskAccessMode.direct;
+            boolean directWrites = DatabaseDescriptor.getBackgroundWriteDiskAccessMode() == Config.DiskAccessMode.direct;
+
+            if (!directReads && !directWrites)
                 return;
 
             List<String> unsupportedLocations = findDirectIOUnsupportedLocations(DatabaseDescriptor.getAllDataFileLocations());
 
             if (!unsupportedLocations.isEmpty())
             {
+                String configuredModes = directReads && directWrites
+                    ? "compaction reads and background writes"
+                    : directReads ? "compaction reads" : "background writes";
+
                 throw new StartupException(StartupException.ERR_WRONG_DISK_STATE,
-                                           String.format("Direct I/O is configured for compaction reads (compaction_read_disk_access_mode=direct), " +
+                                           String.format("Direct I/O is configured for %s, " +
                                                          "but the following data directories do not support Direct I/O: %s. " +
-                                                         "Either change compaction_read_disk_access_mode to 'standard' in cassandra.yaml, " +
+                                                         "Either change the disk access mode to 'standard' in cassandra.yaml, " +
                                                          "or ensure all data directories are on filesystems that support Direct I/O. " +
                                                          "Network filesystems (NFS, CIFS) and some virtual filesystems do not support Direct I/O.",
-                                                         unsupportedLocations));
+                                                         configuredModes, unsupportedLocations));
             }
         }
     };

test/conf/latest_diff.yaml

@@ -46,6 +46,8 @@ memtable_allocation_type: offheap_objects
 
 commitlog_disk_access_mode: auto
 
+background_write_disk_access_mode: direct
+
 trickle_fsync: true
 
 sstable:

test/distributed/org/apache/cassandra/distributed/impl/InstanceConfig.java

@@ -151,6 +151,8 @@ public class InstanceConfig implements IInstanceConfig
 
                 .set("commitlog_disk_access_mode", "auto")
 
+                .set("background_write_disk_access_mode", "direct")
+
                 .set("trickle_fsync", "true")
 
                 .set("sstable", Map.of(

test/distributed/org/apache/cassandra/distributed/test/StreamingDirectWriteTest.java

@@ -0,0 +1,82 @@
+/*
+ * 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.distributed.test;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Random;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.metrics.StorageMetrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class StreamingDirectWriteTest extends TestBaseImpl
+{
+    @Test
+    public void testChunkedStreamReceiverWithDirectWrites() throws IOException
+    {
+        // ZCS streaming bypasses DataComponent.buildWriter; disable to exercise the chunked receiver path.
+        try (Cluster cluster = init(Cluster.build(2)
+                                           .withConfig(c -> c.with(Feature.NETWORK, Feature.GOSSIP)
+                                                             .set("stream_entire_sstables", false)
+                                                             .set("background_write_disk_access_mode", "direct"))
+                                           .start()))
+        {
+            cluster.schemaChange(withKeyspace("CREATE TABLE %s.t (k text PRIMARY KEY, v blob) WITH compression = {'enabled':'true', 'class':'LZ4Compressor'};"));
+            cluster.stream().forEach(i -> i.nodetoolResult("disableautocompaction", KEYSPACE).asserts().success());
+
+            IInvokableInstance node1 = cluster.get(1);
+            IInvokableInstance node2 = cluster.get(2);
+
+            // ~800 KiB of incompressible data spread across many flushes — multiple sstables to stream chunked.
+            int rowCount = 100;
+            int valBytes = 8 * 1024;
+            byte[] val = new byte[valBytes];
+            new Random(42).nextBytes(val);
+            ByteBuffer blob = ByteBuffer.wrap(val);
+            for (int i = 0; i < rowCount; i++)
+            {
+                node1.executeInternal(withKeyspace("INSERT INTO %s.t (k, v) VALUES (?, ?)"), "k" + i, blob);
+                if ((i % 20) == 19)
+                    node1.flush(KEYSPACE);
+            }
+            node1.flush(KEYSPACE);
+
+            // Monotonic meter: a snapshot before/after rebuild proves the DIO writer engaged on the
+            // receiver, regardless of timing or GC — unlike the inc/dec byte gauge, which races to 0.
+            // NB: explicit lambda, not a method ref — a ref binds the non-serializable Meter and fails
+            // to ship across the instance boundary; the lambda reads the static field on the receiver.
+            long writersBefore = node2.callOnInstance(() -> StorageMetrics.directWriteBuffersAllocated.getCount());
+
+            node2.nodetoolResult("rebuild", "--keyspace", KEYSPACE).asserts().success();
+
+            Object[][] rows = node2.executeInternal(withKeyspace("SELECT k FROM %s.t"));
+            assertThat(rows.length).isEqualTo(rowCount);
+
+            long writersAfter = node2.callOnInstance(() -> StorageMetrics.directWriteBuffersAllocated.getCount());
+            assertThat(writersAfter)
+                .describedAs("DIO writer must engage for STREAM op on the chunked receiver")
+                .isGreaterThan(writersBefore);
+        }
+    }
+}

test/unit/org/apache/cassandra/config/DatabaseDescriptorTest.java

@@ -24,10 +24,13 @@ import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
 import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.function.Consumer;
 
 import com.google.common.base.Throwables;
@@ -41,6 +44,7 @@ import org.mockito.MockedStatic;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.distributed.shared.WithProperties;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.io.util.PathUtils;
 import org.apache.cassandra.security.EncryptionContext;
@@ -1074,4 +1078,116 @@ public class DatabaseDescriptorTest
         Config config = DatabaseDescriptor.loadConfig();
         Assert.assertEquals(config.max_value_size.toMebibytes() * 1024 * 1024, DatabaseDescriptor.getMaxValueSize());
     }
+
+    @Test
+    public void testInitializeBackgroundWriteDiskAccessModeRejectsZeroBufferSize()
+    {
+        Config conf = DatabaseDescriptor.getRawConfig();
+        Config.DiskAccessMode savedMode = conf.background_write_disk_access_mode;
+        DataStorageSpec.IntKibibytesBound savedBufferSize = conf.direct_write_buffer_size;
+        try
+        {
+            conf.background_write_disk_access_mode = Config.DiskAccessMode.direct;
+            conf.direct_write_buffer_size = new DataStorageSpec.IntKibibytesBound("0KiB");
+
+            try
+            {
+                DatabaseDescriptor.initializeBackgroundWriteDiskAccessMode();
+                fail("expected ConfigurationException for direct_write_buffer_size == 0");
+            }
+            catch (ConfigurationException expected)
+            {
+                Assert.assertTrue("expected message to mention direct_write_buffer_size, got: " + expected.getMessage(),
+                                  expected.getMessage().contains("direct_write_buffer_size"));
+            }
+        }
+        finally
+        {
+            conf.background_write_disk_access_mode = savedMode;
+            conf.direct_write_buffer_size = savedBufferSize;
+        }
+    }
+
+    @Test
+    public void testInitializeBackgroundWriteDiskAccessModeRejectsUnsupportedModes()
+    {
+        // Everything outside standard/direct must be rejected, including 'auto': it has no
+        // Direct I/O auto-detection for background writes yet, so it must fail rather than
+        // silently resolving to standard (unlike commitlog_disk_access_mode).
+        Config conf = DatabaseDescriptor.getRawConfig();
+        Config.DiskAccessMode savedMode = conf.background_write_disk_access_mode;
+        try
+        {
+            EnumSet<Config.DiskAccessMode> supported = EnumSet.of(Config.DiskAccessMode.standard,
+                                                                  Config.DiskAccessMode.direct);
+            for (Config.DiskAccessMode mode : EnumSet.complementOf(supported))
+            {
+                conf.background_write_disk_access_mode = mode;
+                assertThatExceptionOfType(ConfigurationException.class)
+                    .isThrownBy(DatabaseDescriptor::initializeBackgroundWriteDiskAccessMode)
+                    .withMessageContaining("background_write_disk_access_mode");
+            }
+        }
+        finally
+        {
+            conf.background_write_disk_access_mode = savedMode;
+        }
+    }
+
+    @Test
+    public void testGetDirectIOWritePaths() throws IOException
+    {
+        Config conf = DatabaseDescriptor.getRawConfig();
+        Config.DiskAccessMode savedCommitLogMode = DatabaseDescriptor.getCommitLogWriteDiskAccessMode();
+        Config.DiskAccessMode savedBgWriteMode = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
+        String savedCommitLogLocation = DatabaseDescriptor.getCommitLogLocation();
+        String[] savedDataDirs = conf.data_file_directories;
+        String savedLocalSystemDir = conf.local_system_data_file_directory;
+        try
+        {
+            Path baseDir = Files.createTempDirectory("testGetDirectIOWritePaths");
+            String dataDir1 = baseDir.resolve("dio-data-1").toString();
+            String dataDir2 = baseDir.resolve("dio-data-2").toString();
+            String commitLogDir = baseDir.resolve("dio-commitlog").toString();
+
+            conf.local_system_data_file_directory = null;
+            conf.data_file_directories = new String[]{ dataDir1, dataDir2 };
+            DatabaseDescriptor.setCommitLogLocation(commitLogDir);
+
+            Path commitLogPath = new File(commitLogDir).toPath();
+            Set<Path> dataPaths = new HashSet<>();
+            for (String dir : DatabaseDescriptor.getAllDataFileLocations())
+                dataPaths.add(new File(dir).toPath());
+
+            // Neither write path uses Direct I/O.
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(Config.DiskAccessMode.standard);
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(Config.DiskAccessMode.standard);
+            assertThat(DatabaseDescriptor.getDirectIOWritePaths()).isEmpty();
+
+            // Only the commit log uses Direct I/O.
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(Config.DiskAccessMode.direct);
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(Config.DiskAccessMode.standard);
+            assertThat(DatabaseDescriptor.getDirectIOWritePaths()).containsExactly(commitLogPath);
+
+            // Only background SSTable writes use Direct I/O, which must enumerate every data directory.
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(Config.DiskAccessMode.standard);
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(Config.DiskAccessMode.direct);
+            assertThat(DatabaseDescriptor.getDirectIOWritePaths()).containsExactlyInAnyOrderElementsOf(dataPaths);
+
+            // Both write paths use Direct I/O: the union of the commit log and data directories.
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(Config.DiskAccessMode.direct);
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(Config.DiskAccessMode.direct);
+            Set<Path> expectedUnion = new HashSet<>(dataPaths);
+            expectedUnion.add(commitLogPath);
+            assertThat(DatabaseDescriptor.getDirectIOWritePaths()).containsExactlyInAnyOrderElementsOf(expectedUnion);
+        }
+        finally
+        {
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(savedCommitLogMode);
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(savedBgWriteMode);
+            DatabaseDescriptor.setCommitLogLocation(savedCommitLogLocation);
+            conf.data_file_directories = savedDataDirs;
+            conf.local_system_data_file_directory = savedLocalSystemDir;
+        }
+    }
 }

test/unit/org/apache/cassandra/cql3/CQLTester.java

@@ -3758,6 +3758,11 @@ public abstract class CQLTester
             setupFileSystem();
 
             CQLTester.prePrepareServer();
+
+            // The in-memory (jimfs) filesystem installed above cannot do Direct I/O: its FileStore has no
+            // getBlockSize() and it rejects ExtendedOpenOption.DIRECT. cassandra_latest.yaml enables
+            // background_write_disk_access_mode: direct, which would otherwise crash every flush+compact here.
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(Config.DiskAccessMode.standard);
         }
 
         protected static void setupFileSystem()

test/unit/org/apache/cassandra/db/compaction/AntiCompactionTest.java

@@ -18,12 +18,15 @@
 package org.apache.cassandra.db.compaction;
 
 import java.io.IOException;
+import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.function.Predicate;
 
@@ -55,12 +58,15 @@ import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.io.sstable.Descriptor;
 import org.apache.cassandra.io.sstable.ISSTableScanner;
 import org.apache.cassandra.io.sstable.SSTableTxnWriter;
+import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.io.util.DirectIoTestUtils;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.RangesAtEndpoint;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.repair.NoSuchRepairSessionException;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.MockSchema;
 import org.apache.cassandra.schema.Schema;
@@ -78,6 +84,7 @@ import static org.apache.cassandra.service.ActiveRepairService.NO_PENDING_REPAIR
 import static org.apache.cassandra.service.ActiveRepairService.UNREPAIRED_SSTABLE;
 import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
 import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -89,6 +96,7 @@ public class AntiCompactionTest
 {
     private static final String KEYSPACE1 = "AntiCompactionTest";
     private static final String CF = "AntiCompactionTest";
+    private static final String CF_COMPRESSED = "AntiCompactionCompressed";
     private static final Collection<Range<Token>> NO_RANGES = Collections.emptyList();
 
     private static TableMetadata metadata;
@@ -101,7 +109,10 @@ public class AntiCompactionTest
     {
         SchemaLoader.prepareServer();
         metadata = SchemaLoader.standardCFMD(KEYSPACE1, CF).build();
-        SchemaLoader.createKeyspace(KEYSPACE1, KeyspaceParams.simple(1), metadata);
+        TableMetadata compressedMetadata = SchemaLoader.standardCFMD(KEYSPACE1, CF_COMPRESSED)
+                                                       .compression(CompressionParams.lz4())
+                                                       .build();
+        SchemaLoader.createKeyspace(KEYSPACE1, KeyspaceParams.simple(1), metadata, compressedMetadata);
         cfs = Schema.instance.getColumnFamilyStoreInstance(metadata.id);
         local = InetAddressAndPort.getByName("127.0.0.1");
     }
@@ -114,11 +125,11 @@ public class AntiCompactionTest
         store.truncateBlocking();
     }
 
-    private void registerParentRepairSession(TimeUUID sessionID, Iterable<Range<Token>> ranges, long repairedAt, TimeUUID pendingRepair) throws IOException
+    private void registerParentRepairSession(TimeUUID sessionID, ColumnFamilyStore store, Iterable<Range<Token>> ranges, long repairedAt, TimeUUID pendingRepair) throws IOException
     {
         ActiveRepairService.instance().registerParentRepairSession(sessionID,
                                                                    InetAddressAndPort.getByName("10.0.0.1"),
-                                                                   Lists.newArrayList(cfs), ImmutableSet.copyOf(ranges),
+                                                                   Lists.newArrayList(store), ImmutableSet.copyOf(ranges),
                                                                    pendingRepair != null || repairedAt != UNREPAIRED_SSTABLE,
                                                                    repairedAt, true, PreviewKind.NONE);
     }
@@ -157,7 +168,7 @@ public class AntiCompactionTest
         {
             if (txn == null)
                 throw new IllegalStateException();
-            registerParentRepairSession(sessionID, ranges.ranges(), FBUtilities.nowInSeconds(), sessionID);
+            registerParentRepairSession(sessionID, store, ranges.ranges(), FBUtilities.nowInSeconds(), sessionID);
             CompactionManager.instance.performAnticompaction(store, ranges, refs, txn, sessionID, () -> false);
         }
 
@@ -233,6 +244,72 @@ public class AntiCompactionTest
         assertOnDiskState(store, 3);
     }
 
+    @Test
+    public void testAntiCompactionWithCompressedTableAndDirectWrites() throws Throwable
+    {
+        Keyspace keyspace = Keyspace.open(KEYSPACE1);
+        ColumnFamilyStore store = keyspace.getColumnFamilyStore(CF_COMPRESSED);
+        try
+        {
+            store.disableAutoCompaction();
+            assertTrue("CF must be compressed for DIO to engage",
+                       store.metadata().params.compression.isEnabled());
+
+            long ts = 1L;
+            populateDeterministic(store, ts);
+            SSTableStats baselineStats = antiCompactRanges(store, atEndpoint(range(0, 4), NO_RANGES));
+            assertEquals(2, baselineStats.numLiveSSTables);
+            assertEquals(4, baselineStats.pendingKeys);
+            assertEquals(6, baselineStats.unrepairedKeys);
+            Map<Boolean, byte[]> baselineBytes = captureDataBytesByPendingRepair(store);
+            store.truncateBlocking();
+
+            populateDeterministic(store, ts);
+            SSTableStats directStats = DirectIoTestUtils.withDirectWrites(() ->
+                antiCompactRanges(store, atEndpoint(range(0, 4), NO_RANGES)));
+            assertEquals(2, directStats.numLiveSSTables);
+            assertEquals(4, directStats.pendingKeys);
+            assertEquals(6, directStats.unrepairedKeys);
+            Map<Boolean, byte[]> directBytes = captureDataBytesByPendingRepair(store);
+
+            assertArrayEquals("pending-repair data file must match standard-writer output",
+                              baselineBytes.get(true), directBytes.get(true));
+            assertArrayEquals("unrepaired data file must match standard-writer output",
+                              baselineBytes.get(false), directBytes.get(false));
+        }
+        finally
+        {
+            store.truncateBlocking();
+        }
+    }
+
+    private void populateDeterministic(ColumnFamilyStore store, long ts)
+    {
+        TableMetadata md = store.metadata();
+        for (int i = 0; i < 10; i++)
+        {
+            new RowUpdateBuilder(md, ts, Integer.toString(i))
+                .clustering("c")
+                .add("val", "val")
+                .build()
+                .applyUnsafe();
+        }
+        Util.flush(store);
+    }
+
+    private Map<Boolean, byte[]> captureDataBytesByPendingRepair(ColumnFamilyStore store) throws IOException
+    {
+        Map<Boolean, byte[]> bytes = new HashMap<>();
+        for (SSTableReader sstable : store.getLiveSSTables())
+        {
+            byte[] prev = bytes.put(sstable.isPendingRepair(),
+                                    Files.readAllBytes(sstable.descriptor.fileFor(Components.DATA).toPath()));
+            assertTrue("expected one sstable per pending-repair status; duplicate at " + sstable.isPendingRepair(),
+                       prev == null);
+        }
+        return bytes;
+    }
+
     @Test
     public void antiCompactOneTransOnly() throws Exception
     {
@@ -257,7 +334,7 @@ public class AntiCompactionTest
         List<Range<Token>> ranges = Arrays.asList(range);
         Collection<SSTableReader> sstables = cfs.getLiveSSTables();
         TimeUUID parentRepairSession = nextTimeUUID();
-        registerParentRepairSession(parentRepairSession, ranges, UNREPAIRED_SSTABLE, nextTimeUUID());
+        registerParentRepairSession(parentRepairSession, cfs, ranges, UNREPAIRED_SSTABLE, nextTimeUUID());
         try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION);
              Refs<SSTableReader> refs = Refs.ref(sstables))
         {
@@ -386,7 +463,7 @@ public class AntiCompactionTest
         Range<Token> range = new Range<Token>(new BytesToken("/".getBytes()), new BytesToken("9999".getBytes()));
         List<Range<Token>> ranges = Arrays.asList(range);
         TimeUUID pendingRepair = nextTimeUUID();
-        registerParentRepairSession(pendingRepair, ranges, UNREPAIRED_SSTABLE, pendingRepair);
+        registerParentRepairSession(pendingRepair, store, ranges, UNREPAIRED_SSTABLE, pendingRepair);
 
         try (LifecycleTransaction txn = store.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION);
              Refs<SSTableReader> refs = Refs.ref(sstables))
@@ -420,7 +497,7 @@ public class AntiCompactionTest
         Range<Token> range = new Range<Token>(new BytesToken("-1".getBytes()), new BytesToken("-10".getBytes()));
         List<Range<Token>> ranges = Arrays.asList(range);
         TimeUUID parentRepairSession = nextTimeUUID();
-        registerParentRepairSession(parentRepairSession, ranges, UNREPAIRED_SSTABLE, null);
+        registerParentRepairSession(parentRepairSession, store, ranges, UNREPAIRED_SSTABLE, null);
         boolean gotException = false;
         try (LifecycleTransaction txn = store.getTracker().tryModify(sstables, OperationType.ANTICOMPACTION);
              Refs<SSTableReader> refs = Refs.ref(sstables))

test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java

@@ -18,13 +18,17 @@
 */
 package org.apache.cassandra.db.compaction;
 
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.Random;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
@@ -53,10 +57,16 @@ import org.apache.cassandra.db.RowUpdateBuilder;
 import org.apache.cassandra.db.SinglePartitionReadCommand;
 import org.apache.cassandra.db.Slice;
 import org.apache.cassandra.db.Slices;
+import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;
+import org.apache.cassandra.db.compaction.writers.MaxSSTableSizeWriter;
 import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
+import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.ValueAccessors;
 import org.apache.cassandra.db.partitions.FilteredPartition;
 import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
@@ -77,6 +87,7 @@ import org.apache.cassandra.io.sstable.format.SSTableFormat.Components;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.metadata.StatsMetadata;
 import org.apache.cassandra.schema.CompactionParams;
+import org.apache.cassandra.schema.CompressionParams;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.SchemaTestUtil;
 import org.apache.cassandra.schema.TableMetadata;
@@ -97,6 +108,7 @@ public class CompactionsTest
     private static final String CF_STANDARD2 = "Standard2";
     private static final String CF_STANDARD3 = "Standard3";
     private static final String CF_STANDARD4 = "Standard4";
+    private static final String CF_COMPRESSED_STANDARD1 = "CF_COMPRESSED_STANDARD1";
 
     @Parameterized.Parameter(0)
     public DiskAccessMode compactionReadDiskAccessMode;
@@ -104,25 +116,33 @@ public class CompactionsTest
     @Parameterized.Parameter(1)
     public boolean cursorCompactionEnabled;
 
-    @Parameterized.Parameters(name = "diskAccessMode={0},cursor={1}")
+    @Parameterized.Parameter(2)
+    public DiskAccessMode backgroundWriteDiskAccessMode;
+
+    @Parameterized.Parameters(name = "diskAccessMode={0},cursor={1},backgroundWriteMode={2}")
     public static Collection<Object[]> params()
     {
-        return Arrays.asList(new Object[]{ DiskAccessMode.standard, true },
-                             new Object[]{ DiskAccessMode.standard, false },
-                             new Object[]{ DiskAccessMode.direct, true },
-                             new Object[]{ DiskAccessMode.direct, false });
+        // One direct-write cell instead of cross-multiplying: uncompressed CFs ignore the write mode.
+        return Arrays.asList(new Object[]{ DiskAccessMode.standard, true, DiskAccessMode.standard },
+                             new Object[]{ DiskAccessMode.standard, false, DiskAccessMode.standard },
+                             new Object[]{ DiskAccessMode.direct, true, DiskAccessMode.standard },
+                             new Object[]{ DiskAccessMode.direct, false, DiskAccessMode.standard },
+                             new Object[]{ DiskAccessMode.standard, false, DiskAccessMode.direct });
     }
 
     private DiskAccessMode originalDiskAccessMode;
     private boolean originalCursorCompactionEnabled;
+    private DiskAccessMode originalBackgroundWriteDiskAccessMode;
 
     @Before
     public void setCompactionParams()
     {
         originalDiskAccessMode = DatabaseDescriptor.getCompactionReadDiskAccessMode();
         originalCursorCompactionEnabled = DatabaseDescriptor.cursorCompactionEnabled();
+        originalBackgroundWriteDiskAccessMode = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
         DatabaseDescriptor.setCompactionReadDiskAccessMode(compactionReadDiskAccessMode);
         DatabaseDescriptor.setCursorCompactionEnabled(cursorCompactionEnabled);
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(backgroundWriteDiskAccessMode);
     }
 
     @After
@@ -130,6 +150,7 @@ public class CompactionsTest
     {
         DatabaseDescriptor.setCompactionReadDiskAccessMode(originalDiskAccessMode);
         DatabaseDescriptor.setCursorCompactionEnabled(originalCursorCompactionEnabled);
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(originalBackgroundWriteDiskAccessMode);
     }
 
     @BeforeClass
@@ -149,7 +170,10 @@ public class CompactionsTest
                                                 .compaction(CompactionParams.stcs(compactionOptions)),
                                     SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD2),
                                     SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD3),
-                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD4));
+                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_STANDARD4),
+                                    SchemaLoader.standardCFMD(KEYSPACE1, CF_COMPRESSED_STANDARD1,
+                                                              0, AsciiType.instance, BytesType.instance, AsciiType.instance)
+                                                .compression(CompressionParams.lz4()));
     }
 
     public static long populate(String ks, String cf, int startRowKey, int endRowKey, int ttl)
@@ -422,6 +446,75 @@ public class CompactionsTest
         assertEquals(keys, k);
     }
 
+    @Test
+    public void testCompactionWithSizeLimitedRewriter() throws Exception
+    {
+        Keyspace keyspace = Keyspace.open(KEYSPACE1);
+        ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF_COMPRESSED_STANDARD1);
+        cfs.clearUnsafe();
+        cfs.disableAutoCompaction();
+
+        assertTrue("CF_COMPRESSED_STANDARD1 must be compressed for DIO to engage",
+                   cfs.metadata().params.compression.isEnabled());
+        assertEquals(backgroundWriteDiskAccessMode, DatabaseDescriptor.getBackgroundWriteDiskAccessMode());
+
+        // ~800 KiB of incompressible random data across 200 partitions -> >=2 output sstables at 64 KiB cap.
+        int valSize = 4096;
+        byte[] val = new byte[valSize];
+        new Random(42).nextBytes(val);
+        TableMetadata table = cfs.metadata();
+        long timestamp = System.currentTimeMillis();
+        for (int i = 0; i < 200; i++)
+        {
+            DecoratedKey key = Util.dk(String.format("%05d", i));
+            new RowUpdateBuilder(table, timestamp, key.getKey())
+                .clustering("0")
+                .add("val", ByteBuffer.wrap(val))
+                .build()
+                .applyUnsafe();
+        }
+        Util.flush(cfs);
+        assertEquals(1, cfs.getLiveSSTables().size());
+
+        Set<SSTableReader> originals = new HashSet<>(cfs.getLiveSSTables());
+        long maxSSTableSize = 64L * 1024;
+
+        LifecycleTransaction txn = cfs.getTracker().tryModify(originals, OperationType.COMPACTION);
+        CompactionTask task = new CompactionTask(cfs, txn, FBUtilities.nowInSeconds())
+        {
+            @Override
+            public CompactionAwareWriter getCompactionAwareWriter(ColumnFamilyStore cfs,
+                                                                  Directories directories,
+                                                                  ILifecycleTransaction transaction,
+                                                                  Set<SSTableReader> nonExpiredSSTables)
+            {
+                return new MaxSSTableSizeWriter(cfs, directories, transaction, nonExpiredSSTables, maxSSTableSize, 0);
+            }
+        };
+        task.execute(CompactionManager.instance.active);
+
+        Set<SSTableReader> result = cfs.getLiveSSTables();
+        assertTrue("expected segment rotation to produce >= 2 SSTables, got " + result.size(), result.size() >= 2);
+
+        int partitionsRead = 0;
+        for (SSTableReader sstable : result)
+        {
+            try (ISSTableScanner scanner = sstable.getScanner())
+            {
+                while (scanner.hasNext())
+                {
+                    try (UnfilteredRowIterator it = scanner.next())
+                    {
+                        while (it.hasNext())
+                            it.next();
+                        partitionsRead++;
+                    }
+                }
+            }
+        }
+        assertEquals(200, partitionsRead);
+    }
+
     private void testDontPurgeAccidentally(String k, String cfname) throws InterruptedException
     {
         // This test catches the regression of CASSANDRA-2786

test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterDaemonTest.java

@@ -18,18 +18,39 @@
 
 package org.apache.cassandra.io.sstable;
 
+import java.util.Arrays;
+import java.util.Collection;
+
+import org.junit.After;
+import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
 
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.ServerTestUtils;
+import org.apache.cassandra.config.Config.DiskAccessMode;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.service.StorageService;
 
+@RunWith(Parameterized.class)
 public class CQLSSTableWriterDaemonTest extends CQLSSTableWriterTest
 {
+    @Parameterized.Parameter
+    public DiskAccessMode backgroundWriteMode;
+
+    @Parameterized.Parameters(name = "backgroundWriteMode={0}")
+    public static Collection<Object[]> modes()
+    {
+        return Arrays.asList(new Object[]{ DiskAccessMode.standard },
+                             new Object[]{ DiskAccessMode.direct });
+    }
+
+    private DiskAccessMode originalBackgroundWriteMode;
+
     @BeforeClass
     public static void setup() throws Exception
     {
@@ -41,4 +62,17 @@ public class CQLSSTableWriterDaemonTest extends CQLSSTableWriterTest
         MessagingService.instance().waitUntilListeningUnchecked();
         StorageService.instance.initServer();
     }
+
+    @Before
+    public void setBackgroundWriteMode()
+    {
+        originalBackgroundWriteMode = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(backgroundWriteMode);
+    }
+
+    @After
+    public void restoreBackgroundWriteMode()
+    {
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(originalBackgroundWriteMode);
+    }
 }

test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java

@@ -25,6 +25,7 @@ import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.LinkedHashSet;
@@ -202,6 +203,42 @@ public abstract class CQLSSTableWriterTest
         }
     }
 
+    @Test
+    public void testCompressedWriteAndReadBack() throws Exception
+    {
+        String schema = "CREATE TABLE " + qualifiedTable + " ("
+                        + "  k int PRIMARY KEY,"
+                        + "  v text"
+                        + ") WITH compression = {'enabled':'true', 'class':'LZ4Compressor'}";
+        String insert = "INSERT INTO " + qualifiedTable + " (k, v) VALUES (?, ?)";
+
+        try (CQLSSTableWriter writer = CQLSSTableWriter.builder()
+                                                       .inDirectory(dataDir)
+                                                       .forTable(schema)
+                                                       .using(insert)
+                                                       .build())
+        {
+            for (int i = 0; i < 50; i++)
+                writer.addRow(i, "row-" + i);
+        }
+
+        loadSSTables(dataDir, keyspace, table);
+
+        if (verifyDataAfterLoading)
+        {
+            UntypedResultSet rs = QueryProcessor.executeInternal("SELECT k, v FROM " + qualifiedTable);
+            assertEquals(50, rs.size());
+
+            Set<Integer> seen = new HashSet<>();
+            for (UntypedResultSet.Row row : rs)
+            {
+                int k = row.getInt("k");
+                assertTrue("duplicate key " + k, seen.add(k));
+                assertEquals("row-" + k, row.getString("v"));
+            }
+        }
+    }
+
     private void validateFilesAreInFormat(SSTableFormat<?, ?> format) throws IOException
     {
         try (Stream<Path> dataFilePaths = Files.list(dataDir.toPath()).filter(p -> p.toString().endsWith("Data.db")))

test/unit/org/apache/cassandra/io/sstable/format/DataComponentDirectWriteSelectionTest.java

@@ -0,0 +1,182 @@
+/*
+* 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.sstable.format;
+
+import java.nio.file.Files;
+import java.util.Collections;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.config.Config.DiskAccessMode;
+import org.apache.cassandra.config.Config.FlushCompression;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ClusteringComparator;
+import org.apache.cassandra.db.compaction.OperationType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.io.compress.CompressedSequentialWriter;
+import org.apache.cassandra.io.compress.DirectCompressedSequentialWriter;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SequenceBasedSSTableId;
+import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
+import org.apache.cassandra.io.util.ChecksummedSequentialWriter;
+import org.apache.cassandra.io.util.DirectIoTestUtils;
+import org.apache.cassandra.io.util.File;
+import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.io.util.SequentialWriter;
+import org.apache.cassandra.io.util.SequentialWriterOption;
+import org.apache.cassandra.schema.CompressionParams;
+import org.apache.cassandra.schema.TableMetadata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class DataComponentDirectWriteSelectionTest
+{
+    private static final String KS = "ks_dio_selection";
+    private static final String CF = "cf_dio_selection";
+
+    private File tmpDir;
+    private int nextId;
+
+    @BeforeClass
+    public static void setupClass()
+    {
+        DatabaseDescriptor.daemonInitialization();
+    }
+
+    @Before
+    public void setUp() throws Exception
+    {
+        tmpDir = new File(Files.createTempDirectory("dio_selection"));
+        nextId = 0;
+    }
+
+    @After
+    public void tearDown()
+    {
+        if (tmpDir != null)
+            FileUtils.deleteRecursive(tmpDir);
+    }
+
+    @Test
+    public void testWriterSelectionMatchesAllowListForEveryOp() throws Exception
+    {
+        DirectIoTestUtils.withDirectWrites(() -> {
+            for (OperationType op : OperationType.values())
+            {
+                boolean expectDirect = DataComponent.isDirectWriteSupported(op);
+                try (SequentialWriter w = build(compressedMetadata(), op))
+                {
+                    assertEquals("buildWriter selection must match the direct-write allow-list for " + op
+                                 + ", got " + w.getClass().getSimpleName(),
+                                 expectDirect, w instanceof DirectCompressedSequentialWriter);
+                }
+            }
+        });
+    }
+
+    // The allow-list loop test takes its oracle from isDirectWriteSupported, so it can't catch a bad
+    // reclassification of these ops. Pin their intent independently.
+    @Test
+    public void testScrubAndFlushAreNeverDirectWriteSupported()
+    {
+        assertFalse("SCRUB must never use direct writes (needs mark()/resetAndTruncate())",
+                    DataComponent.isDirectWriteSupported(OperationType.SCRUB));
+        assertFalse("FLUSH must never use direct writes (hot output stays in the page cache)",
+                    DataComponent.isDirectWriteSupported(OperationType.FLUSH));
+    }
+
+    // RELOCATE and UNKNOWN are writesData==false but DO reach buildWriter (relocateSSTables / offline
+    // sstablesplit). Pin them SUPPORTED so they can't silently regress to NOT_A_WRITER (CASSANDRA-21134).
+    @Test
+    public void testRelocateAndUnknownAreDirectWriteSupported()
+    {
+        assertTrue("RELOCATE rewrites SSTables via a compaction task and must use direct writes",
+                   DataComponent.isDirectWriteSupported(OperationType.RELOCATE));
+        assertTrue("UNKNOWN reaches buildWriter via the offline sstablesplit tool and must use direct writes",
+                   DataComponent.isDirectWriteSupported(OperationType.UNKNOWN));
+    }
+
+    @Test
+    public void testUncompressedAlwaysUsesStandardWriter() throws Exception
+    {
+        DirectIoTestUtils.withDirectWrites(() -> {
+            for (OperationType op : OperationType.values())
+            {
+                try (SequentialWriter w = build(uncompressedMetadata(), op))
+                {
+                    assertTrue("Uncompressed tables must use ChecksummedSequentialWriter for " + op + ", got " + w.getClass().getSimpleName(),
+                               w instanceof ChecksummedSequentialWriter);
+                }
+            }
+        });
+    }
+
+    @Test
+    public void testBufferedModeNeverPicksDirectWriter()
+    {
+        DiskAccessMode original = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(DiskAccessMode.standard);
+        try
+        {
+            for (OperationType op : OperationType.values())
+            {
+                try (SequentialWriter w = build(compressedMetadata(), op))
+                {
+                    assertTrue("Buffered mode must not produce DirectCompressedSequentialWriter for " + op + ", got " + w.getClass().getSimpleName(),
+                               w instanceof CompressedSequentialWriter && !(w instanceof DirectCompressedSequentialWriter));
+                }
+            }
+        }
+        finally
+        {
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(original);
+        }
+    }
+
+    private SequentialWriter build(TableMetadata metadata, OperationType op)
+    {
+        Descriptor descriptor = new Descriptor(tmpDir, KS, CF, new SequenceBasedSSTableId(nextId++),
+                                               DatabaseDescriptor.getSelectedSSTableFormat());
+        MetadataCollector collector = new MetadataCollector(new ClusteringComparator(Collections.singletonList(BytesType.instance)));
+        // finishOnClose(false) routes try-with-resources close through abort, avoiding finish() side effects on an empty writer.
+        SequentialWriterOption options = SequentialWriterOption.newBuilder().finishOnClose(false).build();
+        return DataComponent.buildWriter(descriptor, metadata, options, collector, op, FlushCompression.fast, null);
+    }
+
+    private static TableMetadata compressedMetadata()
+    {
+        return TableMetadata.builder(KS, CF)
+                            .addPartitionKeyColumn("k", BytesType.instance)
+                            .compression(CompressionParams.lz4())
+                            .build();
+    }
+
+    private static TableMetadata uncompressedMetadata()
+    {
+        return TableMetadata.builder(KS, CF)
+                            .addPartitionKeyColumn("k", BytesType.instance)
+                            .compression(CompressionParams.noCompression())
+                            .build();
+    }
+}

test/unit/org/apache/cassandra/io/util/DirectIoTestUtils.java

@@ -0,0 +1,49 @@
+/*
+* 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.util;
+
+import java.util.concurrent.Callable;
+
+import org.apache.cassandra.config.Config.DiskAccessMode;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.utils.Throwables.ThrowingRunnable;
+
+public final class DirectIoTestUtils
+{
+    private DirectIoTestUtils() {}
+
+    public static void withDirectWrites(ThrowingRunnable body) throws Exception
+    {
+        withDirectWrites(() -> { body.run(); return null; });
+    }
+
+    public static <T> T withDirectWrites(Callable<T> body) throws Exception
+    {
+        DiskAccessMode original = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
+        DatabaseDescriptor.setBackgroundWriteDiskAccessMode(DiskAccessMode.direct);
+        try
+        {
+            return body.call();
+        }
+        finally
+        {
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(original);
+        }
+    }
+}

test/unit/org/apache/cassandra/service/StartupChecksTest.java

@@ -314,12 +314,23 @@ public class StartupChecksTest
     {
         Assume.assumeTrue(DatabaseDescriptor.getCommitLogCompression() == null); // we would not be able to enable direct io otherwise
         Assume.assumeTrue("Skipping this test on non-Linux OS", FBUtilities.isLinux);
-        testKernelBug1057843Check("ext4", DiskAccessMode.direct, new Semver("6.1.63.1-generic"), false);
-        testKernelBug1057843Check("ext4", DiskAccessMode.direct, new Semver("6.1.64.1-generic"), true);
-        testKernelBug1057843Check("ext4", DiskAccessMode.direct, new Semver("6.1.65.1-generic"), true);
-        testKernelBug1057843Check("ext4", DiskAccessMode.direct, new Semver("6.1.66.1-generic"), false);
-        testKernelBug1057843Check("tmpfs", DiskAccessMode.direct, new Semver("6.1.64.1-generic"), false);
-        testKernelBug1057843Check("ext4", DiskAccessMode.mmap, new Semver("6.1.64.1-generic"), false);
+
+        // Commit log direct writes
+        testKernelBug1057843Check("ext4", DiskAccessMode.direct, DiskAccessMode.standard, new Semver("6.1.63.1-generic"), false);
+        testKernelBug1057843Check("ext4", DiskAccessMode.direct, DiskAccessMode.standard, new Semver("6.1.64.1-generic"), true);
+        testKernelBug1057843Check("ext4", DiskAccessMode.direct, DiskAccessMode.standard, new Semver("6.1.65.1-generic"), true);
+        testKernelBug1057843Check("ext4", DiskAccessMode.direct, DiskAccessMode.standard, new Semver("6.1.66.1-generic"), false);
+        testKernelBug1057843Check("tmpfs", DiskAccessMode.direct, DiskAccessMode.standard, new Semver("6.1.64.1-generic"), false);
+        testKernelBug1057843Check("ext4", DiskAccessMode.mmap, DiskAccessMode.standard, new Semver("6.1.64.1-generic"), false);
+
+        // Background (data dir) direct writes hit the same check; kernel/fs-type logic is swept above, so
+        // here we only prove the data-dir branch reaches it (ext4 trips, tmpfs filtered out). Commit log is
+        // mmap, not standard: standard is rejected for the commit log unless compression/encryption is on.
+        testKernelBug1057843Check("ext4", DiskAccessMode.mmap, DiskAccessMode.direct, new Semver("6.1.64.1-generic"), true);
+        testKernelBug1057843Check("tmpfs", DiskAccessMode.mmap, DiskAccessMode.direct, new Semver("6.1.64.1-generic"), false);
+
+        // Both commit log and background data dir direct writes at once
+        testKernelBug1057843Check("ext4", DiskAccessMode.direct, DiskAccessMode.direct, new Semver("6.1.64.1-generic"), true);
     }
 
     @SuppressWarnings("unchecked")
@@ -563,19 +574,25 @@ public class StartupChecksTest
         }
     }
 
-    private void testKernelBug1057843Check(String fsType, DiskAccessMode diskAccessMode, Semver kernelVersion, boolean expectToFail) throws Exception
+    private void testKernelBug1057843Check(String fsType,
+                                           DiskAccessMode commitLogMode,
+                                           DiskAccessMode backgroundMode,
+                                           Semver kernelVersion,
+                                           boolean expectToFail) throws Exception
     {
         String commitLogLocation = Files.createTempDirectory("testKernelBugCheck").toString();
 
         String savedCommitLogLocation = DatabaseDescriptor.getCommitLogLocation();
         DiskAccessMode savedCommitLogWriteDiskAccessMode = DatabaseDescriptor.getCommitLogWriteDiskAccessMode();
+        DiskAccessMode savedBackgroundWriteDiskAccessMode = DatabaseDescriptor.getBackgroundWriteDiskAccessMode();
         SystemInfo savedSystemInfo = FBUtilities.getSystemInfo();
         try
         {
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(backgroundMode);
             DatabaseDescriptor.setCommitLogLocation(commitLogLocation);
-            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(diskAccessMode);
+            DatabaseDescriptor.setCommitLogWriteDiskAccessMode(commitLogMode);
             DatabaseDescriptor.initializeCommitLogDiskAccessMode();
-            assertThat(DatabaseDescriptor.getCommitLogWriteDiskAccessMode()).isEqualTo(diskAccessMode);
+            assertThat(DatabaseDescriptor.getCommitLogWriteDiskAccessMode()).isEqualTo(commitLogMode);
             FBUtilities.setSystemInfoSupplier(() -> new SystemInfo()
             {
                 @Override
@@ -584,7 +601,15 @@ public class StartupChecksTest
                     return kernelVersion;
                 }
             });
-            withPathOverriddingFileSystem(Map.of(commitLogLocation, fsType), () -> {
+
+            // Override the filesystem type for every path getDirectIOWritePaths() may return (commit log and
+            // all data dirs) so the check sees only our synthetic type, never the real test filesystem.
+            Map<String, String> pathOverrides = new HashMap<>();
+            pathOverrides.put(commitLogLocation, fsType);
+            for (String dataDir : DatabaseDescriptor.getAllDataFileLocations())
+                pathOverrides.put(dataDir, fsType);
+
+            withPathOverriddingFileSystem(pathOverrides, () -> {
                 if (expectToFail)
                     assertThatExceptionOfType(StartupException.class).isThrownBy(() -> StartupChecks.checkKernelBug1057843.execute(options));
                 else
@@ -597,6 +622,7 @@ public class StartupChecksTest
             DatabaseDescriptor.setCommitLogLocation(savedCommitLogLocation);
             DatabaseDescriptor.setCommitLogWriteDiskAccessMode(savedCommitLogWriteDiskAccessMode);
             DatabaseDescriptor.initializeCommitLogDiskAccessMode();
+            DatabaseDescriptor.setBackgroundWriteDiskAccessMode(savedBackgroundWriteDiskAccessMode);
             FBUtilities.setSystemInfoSupplier(() -> savedSystemInfo);
         }
     }