Add cursor based optimized compaction path (WIP)

Adds a compaction implementation utilizing new fixed allocation SSTable reader/writer implementations, and other purpose built code, leading to improved efficiencies.

patch by Nitsan Wakart; reviewed by Branimir Lambov, +TBD for CASSANDRA-20918

Author: Nitsan Wakart

build.xml

@@ -1414,6 +1414,7 @@
         <jvmarg value="-Dcassandra.config=file:///${scm_none_yaml}"/>
         <jvmarg value="-Dcassandra.test.storage_compatibility_mode=NONE"/>
         <jvmarg value="-Dcassandra.skip_sync=true" />
+        <jvmarg value="-Dcassandra.cursor_compaction_enabled=false" />
       </testmacrohelper>
     </sequential>
   </macrodef>

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

@@ -193,6 +193,7 @@ public enum CassandraRelevantProperties
     CONSISTENT_RANGE_MOVEMENT("cassandra.consistent.rangemovement", "true"),
     CONSISTENT_SIMULTANEOUS_MOVES_ALLOW("cassandra.consistent.simultaneousmoves.allow"),
     CRYPTO_PROVIDER_CLASS_NAME("cassandra.crypto_provider_class_name"),
+    CURSOR_COMPACTION_ENABLED("cassandra.cursor_compaction_enabled", "true"),
     CUSTOM_DISK_ERROR_HANDLER("cassandra.custom_disk_error_handler"),
     CUSTOM_GUARDRAILS_CONFIG_PROVIDER_CLASS("cassandra.custom_guardrails_config_provider_class"),
     CUSTOM_QUERY_HANDLER_CLASS("cassandra.custom_query_handler_class"),

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.CURSOR_COMPACTION_ENABLED;
 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;
@@ -655,6 +656,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 cursor_compaction_enabled = CURSOR_COMPACTION_ENABLED.getBoolean();
+
     public volatile boolean use_statements_enabled = true;
 
     /**

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

@@ -4683,6 +4683,17 @@ public static void setTransientReplicationEnabledUnsafe(boolean enabled)
         conf.transient_replication_enabled = enabled;
     }
 
+    public static boolean cursorCompactionEnabled()
+    {
+        return conf.cursor_compaction_enabled;
+    }
+
+    @VisibleForTesting
+    public static void setCursorCompactionEnabled(boolean cursor_compaction_enabled)
+    {
+        conf.cursor_compaction_enabled = cursor_compaction_enabled;
+    }
+
     public static boolean enableDropCompactStorage()
     {
         return conf.drop_compact_storage_enabled;

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

@@ -25,7 +25,7 @@
 
 public class BufferDecoratedKey extends DecoratedKey
 {
-    private final ByteBuffer key;
+    protected ByteBuffer key;
 
     public BufferDecoratedKey(Token token, ByteBuffer key)
     {

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,107 @@ 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
+        {
+            for (int clusteringIndex = 0; clusteringIndex < size; clusteringIndex++)
+            {
+                if (clusteringIndex % 32 == 0)
+                {
+                    clusteringBlock1 = VIntCoding.readUnsignedVInt(c1);
+                    clusteringBlock2 = VIntCoding.readUnsignedVInt(c2);
+                }
+
+                AbstractType<?> type = types[clusteringIndex];
+
+                byte v1Flags = (byte) (clusteringBlock1 & 0b11);
+                byte v2Flags = (byte) (clusteringBlock2 & 0b11);
+
+                // both values are present
+                if ((v1Flags|v2Flags) == 0)
+                {
+                    boolean isByteOrderComparable = type.isByteOrderComparable;
+                    int vlen1,vlen2;
+                    if (type.isValueLengthFixed())
+                    {
+                        vlen1 = vlen2 = type.valueLengthIfFixed();
+                    }
+                    else
+                    {
+                        vlen1 = VIntCoding.readUnsignedVInt32(c1);
+                        vlen2 = VIntCoding.readUnsignedVInt32(c2);
+                    }
+                    int v1Limit = c1.position() + vlen1;
+                    if (v1Limit > limit1)
+                        throw new IllegalArgumentException("Value limit exceeds buffer limit.");
+                    c1.limit(v1Limit);
+                    int v2Limit = c2.position() + vlen2;
+                    if (v2Limit > limit2)
+                        throw new IllegalArgumentException("Value limit exceeds buffer limit.");
+                    c2.limit(v2Limit);
+                    int cmp = isByteOrderComparable ?
+                                ByteBufferUtil.compareUnsigned(c1, c2) :
+                                type.compareCustom(c1, ByteBufferAccessor.instance, c2, ByteBufferAccessor.instance);
+                    if (cmp != 0)
+                        return cmp;
+                    c1.position(v1Limit);
+                    c2.position(v2Limit);
+                    c1.limit(limit1);
+                    c2.limit(limit2);
+                }
+                // present > not present
+                else
+                {
+                    // null (0b10) is smaller than empty (0b01) which is smaller than valued (0b00);
+                    // compare swapped arguments to reverse the order
+                    int cmp = Long.compare(v2Flags, v1Flags);
+                    if (cmp != 0)
+                        return cmp;
+                    // null/empty == null/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

@@ -2329,6 +2329,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)