21460 cursor bti

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

@@ -645,6 +645,12 @@ public enum CassandraRelevantProperties
     TEST_CASSANDRA_SKIP_SYNC("cassandra.skip_sync"),
     TEST_CASSANDRA_SUITENAME("suitename", "suitename_IS_UNDEFINED"),
     TEST_CASSANDRA_TESTTAG("cassandra.testtag", "cassandra.testtag_IS_UNDEFINED"),
+    /** Generated examples for the ClusteringDescriptorPrefixView property test. */
+    TEST_CLUSTERING_PREFIX_VIEW_EXAMPLES("cassandra.test.clustering_prefix_view.examples", "1000"),
+    TEST_COMPACTION_BURN_CHECKPOINT("cassandra.test.compaction_burn_checkpoint", "build/test/logs/burn-checkpoint.tsv"),
+    TEST_COMPACTION_BURN_MINUTES("cassandra.test.compaction_burn_minutes", "0"),
+    TEST_COMPACTION_BURN_SCALE("cassandra.test.compaction_burn_scale"),
+    TEST_COMPACTION_BURN_TARGET_ROWS("cassandra.test.compaction_burn_target_rows", "0"),
     TEST_COMPRESSION("cassandra.test.compression"),
     TEST_COMPRESSION_ALGO("cassandra.test.compression.algo", "lz4"),
     TEST_DEBUG_REF_COUNT("cassandra.debugrefcount"),
@@ -658,8 +664,20 @@ public enum CassandraRelevantProperties
     TEST_DIFFERENTIAL_BIGVOLUME_ROUNDS("cassandra.test.differential.bigvolume.rounds", "20"),
     TEST_DIFFERENTIAL_BIGVOLUME_ROWS_PER_ROUND("cassandra.test.differential.bigvolume.rows_per_round", "100"),
     TEST_DIFFERENTIAL_BIGVOLUME_VALUE_PADDING("cassandra.test.differential.bigvolume.value_padding", "200"),
+    /**
+     * Padding-byte width of the block-boundary sweeps in EdgeCaseDifferentialCompactionTest. It must
+     * exceed the per-row serialization overhead, plus one range tombstone marker for the marker sweep.
+     * Both sweeps fail naming this property if it stops being wide enough to bracket the cut.
+     */
+    TEST_DIFFERENTIAL_BLOCK_BOUNDARY_SWEEP("cassandra.test.differential.block_boundary.sweep", "160"),
     /** Number of generated examples the randomized differential soak runs; must be > 0. */
     TEST_DIFFERENTIAL_EXAMPLES("cassandra.test.differential.examples"),
+    /**
+     * Upper bound of the randomized soak's per-example row count for its wide partitions; the floor is
+     * a quarter of it. Zero disables the wide partitions, which then fails the soak's own promoted-index
+     * assertion.
+     */
+    TEST_DIFFERENTIAL_HUB_ROWS_PER_ROUND("cassandra.test.differential.hub_rows_per_round", "120"),
     /**
      * Preserves a failed differential comparison's captured sstables for post-mortem instead of deleting
      * them. Off by default: the burn scenarios' captures are multi-GB and would fill a CI disk.
@@ -671,6 +689,13 @@ public enum CassandraRelevantProperties
     TEST_DIFFERENTIAL_LARGEPARTITION_VALUE_PADDING("cassandra.test.differential.largepartition.value_padding", "240"),
     /** Seed for the randomized differential soak; defaults to the wall clock, logged per example. */
     TEST_DIFFERENTIAL_SEED("cassandra.test.differential.seed"),
+    /**
+     * Reads every row of a captured output back through a routed slice, so the BTI row trie is
+     * exercised as an index rather than only compared as bytes. On by default; skipped in scale mode.
+     */
+    TEST_DIFFERENTIAL_SLICE_READBACK("cassandra.test.differential.slice_readback", "true"),
+    /** Per-partition slice cap for the read-back; the default clears the widest current scenario. */
+    TEST_DIFFERENTIAL_SLICE_READBACK_MAX_ROWS("cassandra.test.differential.slice_readback.max_rows", "5000"),
     /** Column counts for the pathological wide-table differential test. */
     TEST_DIFFERENTIAL_WIDE_REGULARS("cassandra.test.differential.wide.regulars", "1800"),
     TEST_DIFFERENTIAL_WIDE_STATICS("cassandra.test.differential.wide.statics", "200"),

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

@@ -41,6 +41,7 @@
 import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.CELL_END;
 import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.CELL_HEADER_START;
 import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.CELL_VALUE_START;
+import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.DONE;
 import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.UNFILTERED_END;
 import static org.apache.cassandra.io.sstable.SSTableCursorReader.State.isState;
 
@@ -86,6 +87,12 @@ public StatefulCursor(SSTableReader reader, Collection<PartitionPositionBounds>
 
     public int readPartitionHeader()
     {
+        // Rejected here rather than in readPartitionHeader(PartitionDescriptor), which is past the
+        // swap below: a DONE cursor has no next partition, and rotating the descriptors on a call
+        // that cannot succeed leaves prev holding content the write side never wrote.
+        if (state() == DONE)
+            throw new IllegalStateException("readPartitionHeader() on a cursor that is DONE");
+
         // A range never spans a partition, so one left open belongs to the partition that ended.
         // Reporting it here names that partition; carrying the flag forward would blame the next
         // partition's first start bound instead, and would hide an unmatched close in it.

src/java/org/apache/cassandra/db/compaction/writers/CompactionAwareWriter.java

@@ -156,6 +156,16 @@ public final String getSStableDirectoryPath() throws IOException
         return sstableDirectoryPath;
     }
 
+    /**
+     * Publishes an early-opened partial sstable once enough has been written since the last one.
+     * {@link #append} gets this from {@link SSTableRewriter#append}; the cursor path, which does not append,
+     * calls it directly on each partition boundary.
+     */
+    public final void maybeReopenEarly(DecoratedKey key)
+    {
+        sstableWriter.maybeReopenEarly(key);
+    }
+
     @Override
     protected Throwable doPostCleanup(Throwable accumulate)
     {

src/java/org/apache/cassandra/dht/ReusableDecoratedKey.java

@@ -19,8 +19,10 @@
 package org.apache.cassandra.dht;
 
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 
 import org.apache.cassandra.db.BufferDecoratedKey;
+import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 public abstract class ReusableDecoratedKey extends BufferDecoratedKey
@@ -46,6 +48,25 @@ public void copyKey(ByteBuffer newKey)
         recalculateToken();
     }
 
+    public void copyKey(byte[] newKey, int length)
+    {
+        maybeResizeKey(length);
+        System.arraycopy(newKey, 0, keyBytes, 0, length);
+        keyLength = length;
+        key.limit(length);
+        recalculateToken();
+    }
+
+    /**
+     * Always a copy, token included: the next copyKey overwrites the bytes and moves the token, so
+     * this key is never safe to retain as it is.
+     */
+    @Override
+    public DecoratedKey retainable()
+    {
+        return getToken().getPartitioner().decorateKey(ByteBuffer.wrap(Arrays.copyOf(keyBytes, keyLength)));
+    }
+
     /** WARNING: retains ref to external buffer */
     public void shadowKey(ByteBuffer newKey, byte[] newKeyBytes, int newKeyLength)
     {

src/java/org/apache/cassandra/io/sstable/BigCursorIndexWriter.java

@@ -27,6 +27,7 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.ClusteringPrefix;
 import org.apache.cassandra.db.DeletionTime;
+import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.io.FSWriteError;
 import org.apache.cassandra.io.sstable.format.big.BigFormatPartitionWriter;
 import org.apache.cassandra.io.sstable.format.big.BigTableWriter;
@@ -43,6 +44,7 @@
  */
 public class BigCursorIndexWriter extends CursorIndexWriter
 {
+    private final BigTableWriter writer;
     private final BigTableWriter.IndexWriter indexWriter;
     private final DeletionTime.Serializer deletionTimeSerializer;
     // The garbage-free add() overload exists only on the concrete BloomFilter. With
@@ -59,9 +61,11 @@ public class BigCursorIndexWriter extends CursorIndexWriter
     private int rowIndexEntryOffset;
     private final int indexBlockThreshold;
 
-    public BigCursorIndexWriter(BigTableWriter.IndexWriter indexWriter,
+    public BigCursorIndexWriter(BigTableWriter writer,
+                                BigTableWriter.IndexWriter indexWriter,
                                 DeletionTime.Serializer deletionTimeSerializer)
     {
+        this.writer = writer;
         this.indexWriter = indexWriter;
         this.deletionTimeSerializer = deletionTimeSerializer;
         this.indexBlockThreshold = DatabaseDescriptor.getColumnIndexSize(BigFormatPartitionWriter.DEFAULT_GRANULARITY);
@@ -158,8 +162,8 @@ private void writeClusteringToRowIndexEntries(ClusteringDescriptor clustering) t
     }
 
     @Override
-    public void endPartition(byte[] key, int keyLength, int headerLength,
-                             DeletionTime partitionDeletionTime, long partitionEnd,
+    public void endPartition(org.apache.cassandra.db.DecoratedKey decoratedKey, byte[] key, int keyLength,
+                             int headerLength, DeletionTime partitionDeletionTime, long partitionEnd,
                              ClusteringDescriptor lastName) throws IOException
     {
         /**
@@ -171,6 +175,8 @@ public void endPartition(byte[] key, int keyLength, int headerLength,
         if (bloomFilter != null)
             bloomFilter.add(key, 0, keyLength, reusableIndexes);
         long indexStart = indexFileWriter.position();
+        int columnIndexCount = 0;
+        int indexedPartSize = 0;
         try
         {
             ByteArrayUtil.writeWithShortLength(key, 0, keyLength, indexFileWriter);
@@ -212,6 +218,10 @@ public void endPartition(byte[] key, int keyLength, int headerLength,
 
                 int entriesAndOffsetsSize = rowIndexEntries.getLength() + rowIndexEntriesOffsets.size() * 4;
                 assert entriesAndOffsetsSize > 0;
+                columnIndexCount = rowIndexEntriesOffsets.size();
+                // What RowIndexEntry calls indexedPartSize: the entries and their offsets, without the
+                // header fields that entriesAndOffsetsSize also counts.
+                indexedPartSize = endOfEntries + rowIndexEntriesOffsets.size() * 4;
                 indexFileWriter.writeUnsignedVInt32(entriesAndOffsetsSize); // size != 0
                 // copy the header elements
                 indexFileWriter.write(rowIndexEntries.getData(), endOfEntries, rowIndexEntries.getLength() - endOfEntries);
@@ -227,6 +237,13 @@ public void endPartition(byte[] key, int keyLength, int headerLength,
         {
             throw new FSWriteError(e, indexFileWriter.getPath());
         }
-        indexWriter.summary.maybeAddEntry(key, 0, keyLength, indexStart);
+        // indexEnd and partitionEnd feed the readable boundary that openEarly needs; without them the
+        // preemptive reopen has nothing to publish and never fires.
+        indexWriter.summary.maybeAddEntry(decoratedKey, key, 0, keyLength,
+                                          indexStart, indexFileWriter.position(), partitionEnd);
+
+        // The entry starts after the key, which was written at indexStart with a short length prefix.
+        writer.maybeCacheKey(decoratedKey, partitionStart, indexStart + TypeSizes.SHORT_SIZE + keyLength,
+                             partitionDeletionTime, headerLength, columnIndexCount, indexedPartSize);
     }
 }