CASSANDRA-20918 Add cursor-based low allocation optimized compaction implementation

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

@@ -232,6 +232,7 @@ public enum CassandraRelevantProperties
     DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
     /** In_JVM dtest property indicating that the test should use "latest" configuration */
     DTEST_JVM_DTESTS_USE_LATEST("jvm_dtests.latest"),
+    ENABLE_CURSOR_COMPACTION("cassandra.enable_cursor_compaction", "true"),
     ENABLE_DC_LOCAL_COMMIT("cassandra.enable_dc_local_commit", "true"),
     /**
      * Whether {@link org.apache.cassandra.db.ConsistencyLevel#NODE_LOCAL} should be allowed.

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

@@ -49,6 +49,7 @@
 
 import static org.apache.cassandra.config.CassandraRelevantProperties.AUTOCOMPACTION_ON_STARTUP_ENABLED;
 import static org.apache.cassandra.config.CassandraRelevantProperties.CASSANDRA_AVAILABLE_PROCESSORS;
+import static org.apache.cassandra.config.CassandraRelevantProperties.ENABLE_CURSOR_COMPACTION;
 import static org.apache.cassandra.config.CassandraRelevantProperties.FILE_CACHE_ENABLED;
 import static org.apache.cassandra.config.CassandraRelevantProperties.SKIP_PAXOS_REPAIR_ON_TOPOLOGY_CHANGE;
 import static org.apache.cassandra.config.CassandraRelevantProperties.SKIP_PAXOS_REPAIR_ON_TOPOLOGY_CHANGE_KEYSPACES;
@@ -644,6 +645,8 @@ public static class SSTableConfig
     @Replaces(oldName = "enable_drop_compact_storage", converter = Converters.IDENTITY, deprecated = true)
     public volatile boolean drop_compact_storage_enabled = false;
 
+    public boolean enable_cursor_compaction = ENABLE_CURSOR_COMPACTION.getBoolean();
+
     public volatile boolean use_statements_enabled = true;
 
     /**

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

@@ -4642,6 +4642,17 @@ public static void setTransientReplicationEnabledUnsafe(boolean enabled)
         conf.transient_replication_enabled = enabled;
     }
 
+    public static boolean enableCursorCompaction()
+    {
+        return conf.enable_cursor_compaction;
+    }
+
+    @VisibleForTesting
+    public static void setEnableCursorCompaction(boolean enable_cursor_compaction)
+    {
+        conf.enable_cursor_compaction = enable_cursor_compaction;
+    }
+
     public static boolean enableDropCompactStorage()
     {
         return conf.drop_compact_storage_enabled;

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

@@ -133,7 +133,7 @@ public String toString(TableMetadata metadata)
     /**
      * Serializer for Clustering object.
      * <p>
-     * Because every clustering in a given table must have the same size (ant that size cannot actually change once the table
+     * Because every clustering in a given table must have the same size (and that size cannot actually change once the table
      * has been defined), we don't record that size.
      */
     public static class Serializer

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

@@ -26,14 +26,17 @@
 import com.google.common.base.Joiner;
 import com.google.common.collect.ImmutableList;
 
+import org.apache.cassandra.io.sstable.ClusteringDescriptor;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.ByteBufferAccessor;
 import org.apache.cassandra.db.marshal.ValueAccessor;
 import org.apache.cassandra.db.rows.Row;
-import org.apache.cassandra.db.marshal.AbstractType;
-import org.apache.cassandra.serializers.MarshalException;
-
 import org.apache.cassandra.io.sstable.IndexInfo;
+import org.apache.cassandra.serializers.MarshalException;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 import org.apache.cassandra.utils.bytecomparable.ByteSource;
+import org.apache.cassandra.utils.vint.VIntCoding;
 
 import static org.apache.cassandra.utils.bytecomparable.ByteSource.EXCLUDED;
 import static org.apache.cassandra.utils.bytecomparable.ByteSource.NEXT_COMPONENT;
@@ -156,6 +159,126 @@ public <V1, V2> int compare(ClusteringPrefix<V1> c1, ClusteringPrefix<V2> c2)
         return s1 < s2 ? c1.kind().comparedToClustering : -c2.kind().comparedToClustering;
     }
 
+    public static int compare(ClusteringDescriptor c1, ClusteringDescriptor c2)
+    {
+        final int c1Size = c1.clusteringColumnsBound();
+        final int c2Size = c2.clusteringColumnsBound();
+        final int minColumns = Math.min(c1Size, c2Size);
+
+        final int cmp = compare(c1.clusteringTypes(), c1.clusteringBuffer(), c2.clusteringBuffer(), minColumns);
+        if (cmp != 0)
+            return cmp;
+
+        final ClusteringPrefix.Kind c1Kind = c1.clusteringKind();
+        final ClusteringPrefix.Kind c2Kind = c2.clusteringKind();
+        if (c1Size == c2Size)
+        {
+            return ClusteringPrefix.Kind.compare(c1Kind, c2Kind);
+        }
+
+        return c1Size < c2Size ? c1Kind.comparedToClustering : -c2Kind.comparedToClustering;
+    }
+
+    public static int compare(AbstractType<?>[] types, ByteBuffer c1, ByteBuffer c2) {
+        return compare(types, c1, c2, types.length);
+    }
+
+    private static int compare(AbstractType<?>[] types, ByteBuffer c1, ByteBuffer c2, int size)
+    {
+        long clusteringBlock1 = 0;
+        long clusteringBlock2 = 0;
+        final int position1 = c1.position();
+        final int position2 = c2.position();
+        final int limit1 = c1.limit();
+        final int limit2 = c2.limit();
+        try
+        {
+            int ofst1 = position1;
+            int ofst2 = position2;
+            for (int clusteringIndex = 0; clusteringIndex < size; clusteringIndex++)
+            {
+                if (clusteringIndex % 32 == 0)
+                {
+                    clusteringBlock1 = VIntCoding.getUnsignedVInt(c1, ofst1, limit1);
+                    ofst1 += VIntCoding.computeUnsignedVIntSize(clusteringBlock1);
+                    clusteringBlock2 = VIntCoding.getUnsignedVInt(c2, ofst2, limit2);
+                    ofst2 += VIntCoding.computeUnsignedVIntSize(clusteringBlock2);
+                }
+
+                AbstractType<?> type = types[clusteringIndex];
+
+                boolean v1Present = (clusteringBlock1 & 0b11) == 0;
+                boolean v2Present = (clusteringBlock2 & 0b11) == 0;
+
+                if (v1Present && v2Present)
+                {
+                    boolean isByteOrderComparable = type.isByteOrderComparable;
+                    int vlen1,vlen2;
+                    if (type.isValueLengthFixed())
+                    {
+                        vlen1 = vlen2 = type.valueLengthIfFixed();
+                    }
+                    else
+                    {
+                        vlen1 = VIntCoding.getUnsignedVInt32(c1, ofst1, limit1);
+                        ofst1 += VIntCoding.computeUnsignedVIntSize(vlen1);
+                        vlen2 = VIntCoding.getUnsignedVInt32(c2, ofst2, limit2);
+                        ofst2 += VIntCoding.computeUnsignedVIntSize(vlen2);
+                    }
+                    int v1Limit = ofst1 + vlen1;
+                    if (v1Limit > limit1)
+                        throw new IllegalArgumentException("Value limit exceeds buffer limit.");
+                    c1.position(ofst1).limit(v1Limit);
+                    int v2Limit = ofst2 + vlen2;
+                    if (v2Limit > limit2)
+                        throw new IllegalArgumentException("Value limit exceeds buffer limit.");
+                    c2.position(ofst2).limit(v2Limit);
+                    int cmp = isByteOrderComparable ?
+                                ByteBufferUtil.compareUnsigned(c1, c2) :
+                                type.compareCustom(c1, ByteBufferAccessor.instance, c2, ByteBufferAccessor.instance);
+                    if (cmp != 0)
+                        return cmp;
+                    c1.limit(limit1);
+                    c2.limit(limit2);
+                    ofst1 += vlen1;
+                    ofst2 += vlen2;
+                }
+                // present > not present
+                else if (v1Present && !v2Present)
+                {
+                    return 1;
+                }
+                else if (!v1Present && v2Present)
+                {
+                    return -1;
+                }
+                else
+                {
+                    boolean v1Null = (clusteringBlock1 & 0b10) == 0;
+                    boolean v2Null = (clusteringBlock2 & 0b10) == 0;
+                    // empty > null
+                    if (!v1Null && v2Null)
+                    {
+                        return 1;
+                    }
+                    else if (v1Null && !v2Null)
+                    {
+                        return -1;
+                    }
+                    // empty == empty, continue...
+                }
+                clusteringBlock1 = clusteringBlock1 >>> 2;
+                clusteringBlock2 = clusteringBlock2 >>> 2;
+            }
+        }
+        finally
+        {
+            c1.position(position1).limit(limit1);
+            c2.position(position2).limit(limit2);
+        }
+        return 0;
+    }
+
     public <V1, V2> int compare(Clustering<V1> c1, Clustering<V2> c2)
     {
         return compare(c1, c2, size());

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

@@ -519,7 +519,7 @@ <V> long valuesWithoutSizeSerializedSize(ClusteringPrefix<V> clustering, int ver
             return result;
         }
 
-        byte[][] deserializeValuesWithoutSize(DataInputPlus in, int size, int version, List<AbstractType<?>> types) throws IOException
+        public byte[][] deserializeValuesWithoutSize(DataInputPlus in, int size, int version, List<AbstractType<?>> types) throws IOException
         {
             // Callers of this method should handle the case where size = 0 (in all case we want to return a special value anyway).
             assert size > 0;

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

@@ -2323,6 +2323,11 @@ public boolean shouldIgnoreGcGraceForKey(DecoratedKey dk)
         return partitionKeySetIgnoreGcGrace.contains(dk);
     }
 
+    public boolean shouldIgnoreGcGraceForAnyKey()
+    {
+        return !partitionKeySetIgnoreGcGrace.isEmpty();
+    }
+
     public static Iterable<ColumnFamilyStore> all()
     {
         List<Iterable<ColumnFamilyStore>> stores = new ArrayList<>(Schema.instance.getKeyspaces().size());

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

@@ -33,6 +33,8 @@
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.db.rows.ColumnData;
 import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.db.rows.SerializationHelper;
+import org.apache.cassandra.db.rows.UnfilteredSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.ColumnMetadata;
@@ -529,6 +531,7 @@ public void serializeSubset(Collection<ColumnMetadata> columns, Columns superset
             int supersetCount = superset.size();
             if (columnCount == supersetCount)
             {
+                /** This is prevented by caller for row serialization: {@link UnfilteredSerializer#serializeRowBody(Row, int, SerializationHelper, DataOutputPlus)}*/
                 out.writeUnsignedVInt32(0);
             }
             else if (supersetCount < 64)

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

@@ -114,7 +114,7 @@ public ByteSource asComparableBytes(Version version)
         // The OSS50 version avoids this by adding a terminator.
         return ByteSource.withTerminatorMaybeLegacy(version,
                                                     ByteSource.END_OF_STREAM,
-                                                    token.asComparableBytes(version),
+                                                    getToken().asComparableBytes(version),
                                                     keyComparableBytes(version));
     }
 
@@ -127,7 +127,7 @@ public ByteComparable asComparableBound(boolean before)
 
             return ByteSource.withTerminator(
                     before ? ByteSource.LT_NEXT_COMPONENT : ByteSource.GT_NEXT_COMPONENT,
-                    token.asComparableBytes(version),
+                    getToken().asComparableBytes(version),
                     keyComparableBytes(version));
         };
     }

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

@@ -19,16 +19,16 @@
 
 public interface DeletionPurger
 {
-    public static final DeletionPurger PURGE_ALL = (ts, ldt) -> true;
+    DeletionPurger PURGE_ALL = (ts, ldt) -> true;
 
-    public boolean shouldPurge(long timestamp, long localDeletionTime);
+    boolean shouldPurge(long timestamp, long localDeletionTime);
 
-    public default boolean shouldPurge(DeletionTime dt)
+    default boolean shouldPurge(DeletionTime dt)
     {
         return !dt.isLive() && shouldPurge(dt.markedForDeleteAt(), dt.localDeletionTime());
     }
 
-    public default boolean shouldPurge(LivenessInfo liveness, long nowInSec)
+    default boolean shouldPurge(LivenessInfo liveness, long nowInSec)
     {
         return !liveness.isLive(nowInSec) && shouldPurge(liveness.timestamp(), liveness.localExpirationTime());
     }