Support multi-cell columns in cursor compaction

Cursor compaction now handles non-frozen collections and UDTs, supports UCS
parallelize_output_shards, and applies the collection guardrails.

patch by Jon Haddad; reviewed by Josh McKenzie for CASSANDRA-21463

Author: rustyrazorblade

CHANGES.txt

@@ -1,4 +1,5 @@
 6.0-alpha3
+ * Support multi-cell columns in cursor compaction (CASSANDRA-21463)
  * Optimize authorization logic for BatchStatement and TransactionStatement when a single table is updated (CASSANDRA-21606)
  * Upgrade lz4-java to 1.11.2 (CASSANDRA-21564)
  * Encode the CQL as raw bytes to avoid utf string serialization length limitation (CASSANDRA-21599)

conf/cassandra_latest.yaml

@@ -695,6 +695,9 @@ commitlog_segment_size: 32MiB
 # The default setting is legacy when the storage compatibility is set to 4 or auto otherwise.
 commitlog_disk_access_mode: auto
 
+# Experimental.  New garbage free compaction path.  Does not yet support collections, counters, BTI.
+cursor_compaction_enabled: true
+
 # Set the disk access mode for reading SSTables during compaction. The allowed values are:
 # - auto: inherit from disk_access_mode (default)
 # - direct: use direct I/O for compaction reads, bypassing the OS page cache

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

@@ -663,6 +663,9 @@ 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"),
+    /** 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"),
     TEST_DRIVER_CONNECTION_TIMEOUT_MS("cassandra.test.driver.connection_timeout_ms", "5000"),
     TEST_DRIVER_READ_TIMEOUT_MS("cassandra.test.driver.read_timeout_ms", "12000"),
     TEST_ENCRYPTION("cassandra.test.encryption", "false"),

src/java/org/apache/cassandra/db/ClusteringComparator.java

@@ -249,12 +249,10 @@ private static int compare(AbstractType<?>[] types, ByteBuffer c1, ByteBuffer c2
                     int cmp = Long.compare(v2Flags, v1Flags);
                     if (cmp != 0)
                     {
-                        // Null ordering is type-independent: nulls sort first even on reversed
-                        // (DESC) columns, mirroring ClusteringComparator.compareComponent which
-                        // handles nulls before consulting the type. EMPTY vs VALUED, however, is
-                        // decided by the type: every base type sorts empty before values, so
-                        // ReversedType sorts empty AFTER values (it swaps operands around the
-                        // base comparison) — flip the flag-derived order for reversed columns.
+                        // Nulls are excluded because null ordering is type-independent:
+                        // compareComponent handles a null before it consults the type. Empty
+                        // against valued is not — every base type sorts empty first, so a
+                        // reversed type sorts it last, and the flag-derived order must flip.
                         if ((v1Flags & CLUSTERING_VALUE_FLAG_NULL) == 0 && (v2Flags & CLUSTERING_VALUE_FLAG_NULL) == 0 && type.isReversed())
                             cmp = -cmp;
                         return cmp;

src/java/org/apache/cassandra/db/DeletionTime.java

@@ -177,18 +177,15 @@ public boolean deletes(LivenessInfo info)
     }
 
     /**
-     * Whether this deletion shadows a cell written at {@code timestamp}. The LIVE short-circuit is
-     * load-bearing rather than an optimisation: {@link #deletes(long)} answers TRUE for
-     * {@link LivenessInfo#NO_TIMESTAMP}, because Long.MIN_VALUE is both that sentinel and LIVE's
-     * {@code markedForDeleteAt} — so a LIVE deletion would shadow a cell carrying it.
+     * Whether this deletion shadows a cell written at {@code timestamp}.
      *
-     * {@link #deletes(LivenessInfo)} has no such guard, and must not grow one. Two reasons, and the second is
-     * the load-bearing one: for ROW liveness the answer is behaviour-neutral, since an empty row liveness has
-     * nothing to shadow and "deleted" and "absent" coincide; and the iterator path applies that same unguarded
-     * form to row liveness of its own accord ({@code Row.Merger.merge}, {@code BTreeRow}), so guarding it would
-     * move the reference as well as the cursor. The guard belongs to the cell contract alone, which is why
-     * {@link #deletes(Cell)} carries it too; that overload keeps its own copy because the check has to precede
-     * the virtual {@code timestamp()} read it guards.
+     * LIVE stores {@code Long.MIN_VALUE} as its {@code markedForDeleteAt}, and that is also
+     * {@link LivenessInfo#NO_TIMESTAMP}. The LIVE check is therefore required: without it a LIVE
+     * deletion shadows a cell that carries no timestamp.
+     *
+     * Do not add the same check to {@link #deletes(LivenessInfo)}. The iterator path applies the
+     * unguarded form to row liveness ({@code Row.Merger.merge}, {@code BTreeRow}), so a guard there
+     * would move the reference as well as the cursor.
      */
     public boolean deletesCellAt(long timestamp)
     {

src/java/org/apache/cassandra/db/ReusableLivenessInfo.java

@@ -62,21 +62,9 @@ public boolean isExpired()
     }
 
     /**
-     * The ROW contract, matching the row references: empty liveness is not live
-     * ({@link LivenessInfo#EMPTY}, whose {@code isLive} is {@code !isEmpty()}), an
-     * {@link #EXPIRED_LIVENESS_TTL} marker is never live ({@link LivenessInfo.ExpiredLivenessInfo}), and
-     * otherwise an expiring liveness is live until its expiration second.
-     *
-     * It differs from the cell contract on three inputs, which is why cell liveness has its own type;
-     * {@code ReusableCellLivenessInfo} carries that reading.
-     *
-     * The old code here read the CELL contract instead, treating an {@code EXPIRED_LIVENESS_TTL} marker
-     * (view maintenance's PK-shadow tombstone) as live until its expiration second rather than never. The
-     * flip is inert on this branch: the marker's {@code localExpirationTime} is stamped with
-     * {@code nowInSec} when the view mutation is applied, and compaction only observes the row after that
-     * mutation has been applied and flushed, so compaction-time {@code nowInSec} is never less than the
-     * stamped one — the old read could never actually return {@code true} for it either. Exercised by
-     * {@code MaterializedViewDifferentialCompactionTest}.
+     * The ROW contract, matching the row references: empty liveness is not live, an
+     * {@link #EXPIRED_LIVENESS_TTL} marker is never live, and otherwise an expiring liveness is live
+     * until its expiration second. {@code ReusableCellLivenessInfo} carries the cell reading.
      */
     @Override
     public boolean isLive(long nowInSec)

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

@@ -27,13 +27,13 @@
 import org.apache.cassandra.db.AbstractCompactionController;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Directories;
+import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;
 import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.utils.TimeUUID;
 
 abstract class AbstractCompactionPipeline extends CompactionInfo.Holder implements AutoCloseable {
-    // Which pipeline each compaction selected, as a delta across one compaction. Read by
-    // CompactionPipelineCounts (test), which documents why supportability alone cannot establish it.
+    // Pipeline selection counts, read only by CompactionPipelineCounts in the test tree.
     private static final LongAdder CURSOR_PIPELINES_CREATED = new LongAdder();
     private static final LongAdder ITERATOR_PIPELINES_CREATED = new LongAdder();
 
@@ -66,27 +66,70 @@ static AbstractCompactionPipeline create(
         return new IteratorCompactionPipeline(task, type, scanners, controller, nowInSec, compactionId);
     }
 
+    final CompactionTask task;
+    long totalKeysWritten;
+    CompactionAwareWriter writer;
+
+    AbstractCompactionPipeline(CompactionTask task)
+    {
+        this.task = task;
+    }
+
+    /**
+     * The object that does the work. It owns the progress and the stop flag this pipeline reports,
+     * so a subclass must not keep a second copy of either.
+     */
+    abstract CompactionInfo.Holder delegate();
+
+    @Override
+    public CompactionInfo getCompactionInfo()
+    {
+        return delegate().getCompactionInfo();
+    }
+
+    @Override
+    public boolean isGlobal()
+    {
+        return delegate().isGlobal();
+    }
+
+    public void stop()
+    {
+        delegate().stop();
+    }
+
     abstract boolean processNextPartitionKey() throws IOException;
 
     public abstract long[] getMergedRowCounts();
 
     public abstract long getTotalSourceCQLRows();
 
-    public abstract long getTotalKeysWritten();
-
     public abstract long getTotalBytesScanned();
 
-    public abstract AutoCloseable openWriterResource(ColumnFamilyStore cfs,
-                                                     Directories directories,
-                                                     ILifecycleTransaction transaction,
-                                                     Set<SSTableReader> nonExpiredSSTables);
-
     @Override
     public abstract void close() throws IOException;
 
-    public abstract Collection<SSTableReader> finishWriting();
+    public AutoCloseable openWriterResource(ColumnFamilyStore cfs,
+                                            Directories directories,
+                                            ILifecycleTransaction transaction,
+                                            Set<SSTableReader> nonExpiredSSTables)
+    {
+        this.writer = task.getCompactionAwareWriter(cfs, directories, transaction, nonExpiredSSTables);
+        return writer;
+    }
 
-    public abstract long estimatedKeys();
+    public Collection<SSTableReader> finishWriting()
+    {
+        return writer.finish();
+    }
+
+    public long estimatedKeys()
+    {
+        return writer.estimatedKeys();
+    }
 
-    public abstract void stop();
+    public long getTotalKeysWritten()
+    {
+        return totalKeysWritten;
+    }
 }

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

@@ -19,55 +19,21 @@
 package org.apache.cassandra.db.compaction;
 
 import java.io.IOException;
-import java.util.Collection;
-import java.util.Set;
 
 import org.apache.cassandra.db.AbstractCompactionController;
-import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.Directories;
-import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter;
-import org.apache.cassandra.db.lifecycle.ILifecycleTransaction;
-import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.utils.TimeUUID;
 
 class CursorCompactionPipeline extends AbstractCompactionPipeline {
     final CursorCompactor cursorCompactor;
-    final CompactionTask task;
-    long totalKeysWritten;
-    CompactionAwareWriter writer;
 
     CursorCompactionPipeline(CompactionTask task, OperationType type, AbstractCompactionStrategy.ScannerList scanners, AbstractCompactionController controller, long nowInSec, TimeUUID compactionId) {
-        this.task = task;
+        super(task);
         cursorCompactor = new CursorCompactor(type, scanners.scanners, controller, nowInSec, compactionId);
     }
 
-    public AutoCloseable openWriterResource(ColumnFamilyStore cfs,
-                                            Directories directories,
-                                            ILifecycleTransaction transaction,
-                                            Set<SSTableReader> nonExpiredSSTables) {
-        this.writer = task.getCompactionAwareWriter(cfs, directories, transaction, nonExpiredSSTables);
-        return writer;
-    }
-
-
-    @Override
-    public Collection<SSTableReader> finishWriting() {
-        return writer.finish();
-    }
-
-    @Override
-    public long estimatedKeys() {
-        return writer.estimatedKeys();
-    }
-
     @Override
-    public CompactionInfo getCompactionInfo() {
-        return cursorCompactor.getCompactionInfo();
-    }
-
-    @Override
-    public boolean isGlobal() {
-        return cursorCompactor.isGlobal();
+    CompactionInfo.Holder delegate() {
+        return cursorCompactor;
     }
 
     @Override
@@ -90,11 +56,6 @@ public long getTotalSourceCQLRows() {
         return cursorCompactor.getTotalSourceCQLRows();
     }
 
-    @Override
-    public long getTotalKeysWritten() {
-        return totalKeysWritten;
-    }
-
     @Override
     public long getTotalBytesScanned() {
         return cursorCompactor.getTotalBytesScanned();
@@ -104,9 +65,4 @@ public long getTotalBytesScanned() {
     public void close() throws IOException {
         cursorCompactor.close();
     }
-
-    @Override
-    public void stop() {
-        cursorCompactor.stop();
-    }
 }