CASSANDRA-21134: Fix O_DIRECT short reads on early-open of compaction SSTables

Early-open publishes readers before commit pads/truncates the file. Under
O_DIRECT this exposed not-yet-durable or block-padded tails, so early
readers short-read past EOF (CorruptBlockException/CorruptSSTableException).
Fix all three early-open paths inside the DIO writer subclass:

- final early-open: override syncInternal() to flush the buffered sub-block
  tail, restoring channel/buffer cursors so the commit-time pad+truncate
  still yields a byte-identical file.
- preemptive early-open (SSTableRewriter.maybeReopenEarly): report the
  durable uncompressed offset by tracking each staged chunk's
  {compressedEnd, uncompressedEnd} and advancing only over chunks whose
  compressed bytes sit below fchannel.position(). No extra I/O.
- scan early-open: truncate to actualDataSize at openFinalEarly so a
  scanner's snapshotted file size already equals the committed size.

Author: samueldlightfoot

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

@@ -19,8 +19,10 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.IntConsumer;
+import java.util.function.LongConsumer;
 
 import javax.annotation.Nullable;
 
@@ -34,6 +36,7 @@
 
 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;
@@ -78,6 +81,12 @@ public class DirectCompressedSequentialWriter extends CompressedSequentialWriter
 
     private final int blockSize;
 
+    // Chunks staged in writeBuffer but not yet on disk, in write order, each as {compressedEnd, uncompressedEnd}.
+    // Drained as whole blocks reach disk so the post-flush listener can report a durable offset (see
+    // setPostFlushListener). A chunk costs one small array per ~chunkLength written — negligible churn.
+    private final ArrayDeque<long[]> stagedChunkBoundaries = new ArrayDeque<>();
+    private long durableUncompressedOffset = 0;
+
     public DirectCompressedSequentialWriter(File file,
                                             File offsetsFile,
                                             @Nullable File digestFile,
@@ -148,6 +157,27 @@ protected void seekToChunkStart()
         // resetAndTruncate (the parent's reason for this seek) is unsupported under DIO.
     }
 
+    // openFinalEarly() publishes a reader right after sync(), before commit. The inherited sync flushes
+    // only whole blocks, so under O_DIRECT the last chunk's sub-block tail is still buffered and that
+    // reader short-reads past EOF — flush the tail here too.
+    @Override
+    protected void syncInternal()
+    {
+        doFlush(0);
+        flushBufferedTailForEarlyOpen();
+        syncDataOnlyInternal();
+    }
+
+    // flushData fires this per chunk with the staged (uncompressed) offset, but under O_DIRECT a chunk is
+    // only readable once its block reaches disk. Preemptive early-open (markDataSynced) publishes a reader
+    // as soon as this offset covers its boundary, so reporting the staged offset would expose chunks still
+    // in writeBuffer. Report the durable offset instead; the boundary simply waits for the next block flush.
+    @Override
+    public void setPostFlushListener(LongConsumer postFlush)
+    {
+        super.setPostFlushListener(stagedOffset -> postFlush.accept(durableUncompressedOffset()));
+    }
+
     @Override
     protected void writeChunk(ByteBuffer toWrite)
     {
@@ -164,6 +194,7 @@ protected void writeChunk(ByteBuffer toWrite)
         crcMetadata.appendDirect(toWrite, true);
 
         actualDataSize = chunkOffset + chunkLength + CRC_LENGTH;
+        stagedChunkBoundaries.add(new long[]{ actualDataSize, uncompressedSize });
     }
 
     private void writeToAlignedBuffer(ByteBuffer data)
@@ -229,6 +260,57 @@ private void flushCompleteBlocks()
         }
     }
 
+    // Truncate to actualDataSize so an early-open scan reader, which snapshots the physical file size once and
+    // bounds its reads by it, isn't shrunk under by commit's truncate into short-reading its last block. (Point
+    // reads are immune: they bound by compressedFileLength.) O_DIRECT forces the pad-to-a-block before the
+    // truncate. Restoring the cursors leaves the writer untouched so commit reproduces the identical file and
+    // writes resume at resumeChannelPos (always <= actualDataSize, so the truncate never strands the offset).
+    private void flushBufferedTailForEarlyOpen()
+    {
+        int logicalPos = writeBuffer.position();
+        if (logicalPos == 0)
+            return;
+
+        try
+        {
+            long resumeChannelPos = fchannel.position();
+            writeAlignedBlocks(BitUtil.align(logicalPos, blockSize));
+            fchannel.truncate(actualDataSize);
+
+            fchannel.position(resumeChannelPos);
+            writeBuffer.limit(writeBuffer.capacity());
+            writeBuffer.position(logicalPos);
+        }
+        catch (IOException e)
+        {
+            throw new FSWriteError(e, getPath());
+        }
+    }
+
+    // Advance over chunks whose compressed bytes now sit entirely below the on-disk boundary
+    // (fchannel.position() — only whole blocks are written). Chunks flush in order, so a single drain
+    // suffices and the offset is monotonic.
+    private long durableUncompressedOffset()
+    {
+        long onDisk;
+        try
+        {
+            onDisk = fchannel.position();
+        }
+        catch (IOException e)
+        {
+            throw new FSReadError(e, getPath());
+        }
+
+        long[] chunk;
+        while ((chunk = stagedChunkBoundaries.peek()) != null && chunk[0] <= onDisk)
+        {
+            durableUncompressedOffset = chunk[1];
+            stagedChunkBoundaries.poll();
+        }
+        return durableUncompressedOffset;
+    }
+
     private void flushFinalWithPadding()
     {
         int logicalPos = writeBuffer.position();

test/unit/org/apache/cassandra/io/compress/DirectCompressedSequentialWriterTest.java

@@ -810,6 +810,181 @@ public void testOnDiskFilePointerTracksActualDataSize() throws IOException
         }
     }
 
+    /**
+     * Covers the early-open read path: openFinalEarly() publishes a reader after sync() but before
+     * commit, and under O_DIRECT the last chunk's sub-block tail is still buffered then, so without the
+     * sync() tail-flush the reader short-reads past EOF. The other tests miss this window — they all read
+     * post-finish, once the tail is flushed. Tiny sub-block outputs (e.g. system/IndexInfo) are the prod shape.
+     */
+    @Test
+    public void testEarlyOpenAfterSyncReadsBackData() throws IOException
+    {
+        for (int size : new int[]{ 1, 64, 298, 1024, 4096, DEFAULT_CHUNK_LENGTH * 3 + 137 })
+            assertEarlyOpenReadsBack(compressible(size), CompressionParams.lz4());
+    }
+
+    // Compresses well below one block, so after sync() the whole chunk+CRC is still buffered and the
+    // on-disk file is empty — the worst case for the early-open read.
+    private static byte[] compressible(int size)
+    {
+        byte[] data = new byte[size];
+        for (int i = 0; i < size; i++)
+            data[i] = (byte) (i % 7);
+        return data;
+    }
+
+    private static void assertEarlyOpenReadsBack(byte[] payload, CompressionParams params) throws IOException
+    {
+        File dataFile = FileUtils.createTempFile("early_open_direct", ".db");
+        File metadataFile = new File(dataFile.absolutePath() + ".metadata");
+        try
+        {
+            MetadataCollector collector = newCollector();
+            try (DirectCompressedSequentialWriter writer = new DirectCompressedSequentialWriter(
+                dataFile, metadataFile, null, SequentialWriterOption.DEFAULT, params, collector, null))
+            {
+                writer.write(payload);
+
+                // Mirror openFinalEarly: sync(), then open(0) == openDataFile(NO_LENGTH_OVERRIDE).
+                writer.sync();
+                try (CompressionMetadata md = writer.open(0);
+                     FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete();
+                     RandomAccessReader reader = fh.createReader())
+                {
+                    assertEquals("early-open length mismatch, size=" + payload.length, payload.length, reader.length());
+                    byte[] readBack = new byte[payload.length];
+                    reader.readFully(readBack);
+                    assertArrayEquals("early-open read-back mismatch, size=" + payload.length, payload, readBack);
+                }
+
+                writer.finish();
+            }
+        }
+        finally
+        {
+            dataFile.tryDelete();
+            metadataFile.tryDelete();
+        }
+    }
+
+    /**
+     * Covers the preemptive early-open read path (SSTableRewriter.maybeReopenEarly): unlike openFinalEarly
+     * it never calls sync(). The partition index advances its readable boundary off the writer's post-flush
+     * offset, then a reader is published over that boundary mid-compaction. Under O_DIRECT only whole blocks
+     * reach disk during flushData, so the writer must report the durable offset, not the staged one —
+     * otherwise the reader short-reads chunks still parked in writeBuffer. testEarlyOpenAfterSyncReadsBackData
+     * does not cover this — its sync() flushes the tail; this window has no sync at all.
+     */
+    @Test
+    public void testPreemptiveOpenReadsBackSyncedData() throws IOException
+    {
+        // Many small chunks that compress far below one block: nothing reaches disk before finish, so the
+        // writer must report 0 synced (the prod system/IndexInfo shape) rather than the staged length.
+        assertSyncedOffsetIsReadable(compressible(DEFAULT_CHUNK_LENGTH * 4 + 137), CompressionParams.lz4());
+
+        // Several MiB of incompressible data forces real block flushes mid-write, so the synced boundary
+        // must advance past zero (preemptive open still works) yet never cross the still-buffered tail.
+        long synced = assertSyncedOffsetIsReadable(incompressible(4 << 20), CompressionParams.lz4());
+        assertTrue("expected the synced boundary to advance for multi-MiB data", synced > 0);
+    }
+
+    // Mirrors how a preemptive early-open reader is bounded: whatever offset the writer reports as synced
+    // (via the post-flush listener that drives PartitionIndexBuilder.markDataSynced) must be fully readable
+    // from disk, with no sync()/finish() in between. Returns that synced offset.
+    private static long assertSyncedOffsetIsReadable(byte[] payload, CompressionParams params) throws IOException
+    {
+        File dataFile = FileUtils.createTempFile("preemptive_open_direct", ".db");
+        File metadataFile = new File(dataFile.absolutePath() + ".metadata");
+        try
+        {
+            MetadataCollector collector = newCollector();
+            try (DirectCompressedSequentialWriter writer = new DirectCompressedSequentialWriter(
+                dataFile, metadataFile, null, SequentialWriterOption.DEFAULT, params, collector, null))
+            {
+                long[] syncedOffset = { 0 };
+                writer.setPostFlushListener(offset -> syncedOffset[0] = offset);
+
+                writer.write(payload);
+
+                int readable = (int) syncedOffset[0];
+                if (readable > 0)
+                {
+                    try (CompressionMetadata md = writer.open(readable);
+                         FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete();
+                         RandomAccessReader reader = fh.createReader())
+                    {
+                        byte[] readBack = new byte[readable];
+                        reader.readFully(readBack);
+                        assertArrayEquals("preemptive-open read-back mismatch", Arrays.copyOf(payload, readable), readBack);
+                    }
+                }
+
+                writer.finish();
+                return syncedOffset[0];
+            }
+        }
+        finally
+        {
+            dataFile.tryDelete();
+            metadataFile.tryDelete();
+        }
+    }
+
+    private static byte[] incompressible(int size)
+    {
+        byte[] data = new byte[size];
+        new Random(42).nextBytes(data);
+        return data;
+    }
+
+    /**
+     * Regression test for the early-open SCAN read path, which the point-read early-open tests
+     * ({@link #testEarlyOpenAfterSyncReadsBackData}, {@link #testPreemptiveOpenReadsBackSyncedData}) miss: a
+     * scanner snapshots the data file's size when it opens, so if openFinalEarly leaves the file padded and
+     * commit later truncates it, the live scanner short-reads its last block (CorruptSSTableException — the
+     * system/local_metadata_log failure seen under dtest-latest). Tiny sub-block outputs are the worst case.
+     */
+    @Test
+    public void testEarlyOpenScanReadsBackAfterCommitTruncate() throws IOException
+    {
+        for (int size : new int[]{ 1, 298, 1024, DEFAULT_CHUNK_LENGTH * 3 + 137 })
+            assertEarlyOpenScanReadsBackAcrossCommit(compressible(size), CompressionParams.lz4());
+    }
+
+    // Ordering is the whole point: the scanner must open before finish() so it caches the file size while
+    // padded, then finish() truncates underneath it.
+    private static void assertEarlyOpenScanReadsBackAcrossCommit(byte[] payload, CompressionParams params) throws IOException
+    {
+        File dataFile = FileUtils.createTempFile("early_open_scan_direct", ".db");
+        File metadataFile = new File(dataFile.absolutePath() + ".metadata");
+        try
+        {
+            MetadataCollector collector = newCollector();
+            try (DirectCompressedSequentialWriter writer = new DirectCompressedSequentialWriter(
+                dataFile, metadataFile, null, SequentialWriterOption.DEFAULT, params, collector, null))
+            {
+                writer.write(payload);
+                writer.sync();
+                try (CompressionMetadata md = writer.open(0);
+                     FileHandle fh = new FileHandle.Builder(dataFile).withCompressionMetadata(md).complete();
+                     // forScan=true engages the read-ahead path; the scanner snapshots channel.size() here.
+                     RandomAccessReader scanReader = fh.createReader(null, true))
+                {
+                    writer.finish();
+
+                    byte[] readBack = new byte[payload.length];
+                    scanReader.readFully(readBack);
+                    assertArrayEquals("early-open scan read-back mismatch, size=" + payload.length, payload, readBack);
+                }
+            }
+        }
+        finally
+        {
+            dataFile.tryDelete();
+            metadataFile.tryDelete();
+        }
+    }
+
     private void testWriteAndRead(String testName, int dataSize, CompressionParams params) throws IOException
     {
         File dataFile = FileUtils.createTempFile(testName + "_direct", ".db");