CNDB-18906: Implement a trie serialization format for partition updates

conf/cassandra.yaml

@@ -791,6 +791,12 @@ memtable:
   configurations:
     skiplist:
       class_name: SkipListMemtable
+    trie_stage1:
+      class_name: TrieMemtableStage1
+    trie_stage2:
+      class_name: TrieMemtableStage2
+    trie_stage3:
+      class_name: TrieMemtableStage3
     trie:
       class_name: TrieMemtable
     default:

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

@@ -421,7 +421,7 @@ public enum CassandraRelevantProperties
      * @deprecated remove when cndb no longer supports bdp/6.8-cndb
      */
     @Deprecated(since = "5.0")
-    DS_CURRENT_MESSAGING_VERSION("ds.current_messaging_version", Integer.toString(MessagingService.VERSION_DS_20)),
+    DS_CURRENT_MESSAGING_VERSION("ds.current_messaging_version", Integer.toString(MessagingService.VERSION_DS_21)),
     DTEST_API_LOG_TOPOLOGY("cassandra.dtest.api.log.topology"),
     /** This property indicates if the code is running under the in-jvm dtest framework */
     DTEST_IS_IN_JVM_DTEST("org.apache.cassandra.dtest.is_in_jvm_dtest"),
@@ -762,10 +762,10 @@ public enum CassandraRelevantProperties
 
     /** Class used to discover/load the proper SAI index components file for a given sstable. */
     SAI_ANN_USE_SYNTHETIC_SCORE("cassandra.sai.ann_use_synthetic_score", "false"),
-    
+
     /** The current version of the SAI on-disk index format. */
     SAI_CURRENT_VERSION("cassandra.sai.latest.version", "ec"),
-    
+
     SAI_CUSTOM_COMPONENTS_DISCOVERY_CLASS("cassandra.sai.custom_components_discovery_class"),
     SAI_ENABLE_EDGES_CACHE("cassandra.sai.enable_edges_cache", "false"),
     SAI_ENABLE_GENERAL_ORDER_BY("cassandra.sai.general_order_by", "true"),
@@ -907,7 +907,6 @@ public enum CassandraRelevantProperties
     // NVQ number of subvectors. This isn't really expected to change much so we're only exposing
     // it as a global variable in case it's needed.
     SAI_VECTOR_NVQ_NUM_SUB_VECTORS("cassandra.sai.vector.nvq_num_sub_vectors", "2"),
-
     // The allowed ratio of extra rows (that map to "holes" in the ordinal space) to total rows indexed in the graph
     // Higher percentages will result in more memory utilized to store the extra postings mappings and larger graph
     // file sizes to store the empty nodes.
@@ -923,7 +922,6 @@ public enum CassandraRelevantProperties
      * build a potential result set for search-then-sort query execution.
      */
     SAI_VECTOR_SEARCH_MAX_MATERIALIZE_KEYS("cassandra.sai.vector_search.max_materialized_keys", "16000"),
-
     /** Controls the maximum top-k limit for vector search */
     SAI_VECTOR_SEARCH_MAX_TOP_K("cassandra.sai.vector_search.max_top_k", "1000"),
     SAI_VECTOR_USE_PRUNING_DEFAULT("cassandra.sai.jvector.use_pruning_default", "false"),
@@ -1098,6 +1096,8 @@ public enum CassandraRelevantProperties
     // i.e. that all replicas except for at most one in the cluster (across all DCs) must accept the write for it to be successful.
     THREE_MEANS_ALL_BUT_ONE("dse.consistency_level.three_means_all_but_one", "false"),
     TOLERATE_SSTABLE_SIZE("cassandra.tolerate_sstable_size"),
+    /** To be used for tests: whether trie cursors should be verified for correctness. */
+    TRIE_DEBUG("cassandra.debug_tries"),
     /**
      * Allows to set custom current trie index format. This node will produce sstables in this format.
      */

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

@@ -3991,10 +3991,10 @@ public static void setFileCacheEnabled(boolean enabled)
 
     public static int getFileCacheSizeInMiB()
     {
-        if (conf.file_cache_size == null)
+        if (conf == null || conf.file_cache_size == null)
         {
-            // In client mode the value is not set.
-            assert DatabaseDescriptor.isClientInitialized();
+            // In client mode or uninitialized unit tests the value is not set.
+            assert conf == null || DatabaseDescriptor.isClientInitialized();
             return 0;
         }
 
@@ -4014,21 +4014,21 @@ public static void enableChunkCache(int sizeInMB)
 
     public static int getNetworkingCacheSizeInMiB()
     {
-        if (conf.networking_cache_size == null)
+        if (conf == null || conf.networking_cache_size == null)
         {
-            // In client mode the value is not set.
-            assert DatabaseDescriptor.isClientInitialized();
+            // In client mode or uninitialized unit tests the value is not set.
+            assert conf == null || DatabaseDescriptor.isClientInitialized();
             return 0;
         }
         return conf.networking_cache_size.toMebibytes();
     }
 
     public static boolean getFileCacheRoundUp()
     {
-        if (conf.file_cache_round_up == null)
+        if (conf == null || conf.file_cache_round_up == null)
         {
-            // In client mode the value is not set.
-            assert DatabaseDescriptor.isClientInitialized();
+            // In client mode or uninitialized unit tests the value is not set.
+            assert conf == null || DatabaseDescriptor.isClientInitialized();
             return false;
         }
 

src/java/org/apache/cassandra/cql3/UpdateParameters.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 
 import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.db.*;
@@ -103,13 +104,13 @@ public <V> void newRow(Clustering<V> clustering) throws InvalidRequestException
         if (clustering == Clustering.STATIC_CLUSTERING)
         {
             if (staticBuilder == null)
-                staticBuilder = BTreeRow.unsortedBuilder();
+                staticBuilder = PartitionUpdate.rowBuilder(metadata, true, false);
             builder = staticBuilder;
         }
         else
         {
             if (regularBuilder == null)
-                regularBuilder = BTreeRow.unsortedBuilder();
+                regularBuilder = PartitionUpdate.rowBuilder(metadata, false, false);
             builder = regularBuilder;
         }
 

src/java/org/apache/cassandra/cql3/statements/BatchStatement.java

@@ -494,7 +494,7 @@ private void updatePerBatchMetrics(Collection<? extends IMutation> mutations)
             for (PartitionUpdate update : mutation.getPartitionUpdates())
             {
                 for (Row row : update.rows())
-                    nrUpdatedColumns += row.columns().size();
+                    nrUpdatedColumns += row.columnCount();
             }
         }
         metrics.update(type, nrUpdatedPartitions, nrUpdatedColumns);

src/java/org/apache/cassandra/cql3/statements/schema/TableAttributes.java

@@ -202,8 +202,8 @@ else if ("PersistentMemoryMemtable".equalsIgnoreCase(shortClassName))
                         builder.memtable(MemtableParams.get("persistent_memory"));
                     else if ("TrieMemtable".equalsIgnoreCase(shortClassName))
                         builder.memtable(MemtableParams.get("trie"));
-                    else if ("TrieMemtableStage1".equalsIgnoreCase(shortClassName))
-                        builder.memtable(MemtableParams.get("trie"));
+                    else if (shortClassName.matches("TrieMemtableStage\\d"))
+                        builder.memtable(MemtableParams.get(shortClassName.replaceFirst("TrieMemtableStage", "trie_stage")));
                     else if ("ShardedSkipListMemtable".equalsIgnoreCase(shortClassName))
                         builder.memtable(MemtableParams.get("skiplist_sharded"));
                     else

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

@@ -2919,43 +2919,46 @@ private SSTableMultiWriter writeMemtableRanges(Supplier<Collection<Range<Partiti
         if (current.isClean())
             return null;
 
-        List<Memtable.FlushablePartitionSet<?>> dataSets = new ArrayList<>(ranges.size());
-        IntervalSet.Builder<CommitLogPosition> commitLogIntervals = new IntervalSet.Builder();
-        long keys = 0;
-        for (Range<PartitionPosition> range : ranges)
+        try (OpOrder.Group protectData = readOrdering.start())
         {
-            Memtable.FlushablePartitionSet<?> dataSet = current.getFlushSet(range.left, range.right);
-            dataSets.add(dataSet);
-            commitLogIntervals.add(dataSet.commitLogLowerBound(), dataSet.commitLogUpperBound());
-            keys += dataSet.partitionCount();
-        }
-        if (keys == 0)
-            return null;
+            List<Memtable.FlushablePartitionSet<?>> dataSets = new ArrayList<>(ranges.size());
+            IntervalSet.Builder<CommitLogPosition> commitLogIntervals = new IntervalSet.Builder();
+            long keys = 0;
+            for (Range<PartitionPosition> range : ranges)
+            {
+                Memtable.FlushablePartitionSet<?> dataSet = current.getFlushSet(range.left, range.right);
+                dataSets.add(dataSet);
+                commitLogIntervals.add(dataSet.commitLogLowerBound(), dataSet.commitLogUpperBound());
+                keys += dataSet.partitionCount();
+            }
+            if (keys == 0)
+                return null;
 
-        // TODO: Can we write directly to stream, skipping disk?
-        Memtable.FlushablePartitionSet<?> firstDataSet = dataSets.get(0);
-        SSTableMultiWriter writer = createSSTableMultiWriter(newSSTableDescriptor(directories.getDirectoryForNewSSTables()),
-                                                             keys,
-                                                             0,
-                                                             repairSessionID,
-                                                             false,
-                                                             commitLogIntervals.build(),
-                                                             new SerializationHeader(true,
-                                                                                     firstDataSet.metadata(),
-                                                                                     firstDataSet.columns(),
-                                                                                     firstDataSet.encodingStats()),
-                                                             DO_NOT_TRACK);
-        try
-        {
-            for (Memtable.FlushablePartitionSet<?> dataSet : dataSets)
-                new Flushing.FlushRunnable(dataSet, writer, metric, false).call();  // executes on this thread
+            // TODO: Can we write directly to stream, skipping disk?
+            Memtable.FlushablePartitionSet<?> firstDataSet = dataSets.get(0);
+            SSTableMultiWriter writer = createSSTableMultiWriter(newSSTableDescriptor(directories.getDirectoryForNewSSTables()),
+                                                                 keys,
+                                                                 0,
+                                                                 repairSessionID,
+                                                                 false,
+                                                                 commitLogIntervals.build(),
+                                                                 new SerializationHeader(true,
+                                                                                         firstDataSet.metadata(),
+                                                                                         firstDataSet.columns(),
+                                                                                         firstDataSet.encodingStats()),
+                                                                 DO_NOT_TRACK);
+            try
+            {
+                for (Memtable.FlushablePartitionSet<?> dataSet : dataSets)
+                    new Flushing.FlushRunnable(dataSet, writer, metric, false).call();  // executes on this thread
 
-            return writer;
-        }
-        catch (Error | RuntimeException t)
-        {
-            writer.abort(t);
-            throw t;
+                return writer;
+            }
+            catch (Error | RuntimeException t)
+            {
+                writer.abort(t);
+                throw t;
+            }
         }
     }
 

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

@@ -76,6 +76,7 @@
 import static org.apache.cassandra.net.MessagingService.VERSION_DS_11;
 import static org.apache.cassandra.net.MessagingService.VERSION_DS_12;
 import static org.apache.cassandra.net.MessagingService.VERSION_DS_20;
+import static org.apache.cassandra.net.MessagingService.VERSION_DS_21;
 import static org.apache.cassandra.net.MessagingService.VERSION_40;
 import static org.apache.cassandra.net.MessagingService.VERSION_50;
 import static org.apache.cassandra.net.MessagingService.VERSION_DSE_68;
@@ -490,6 +491,7 @@ public long getTimeout(TimeUnit unit)
     private int serializedSizeDS11;
     private int serializedSizeDS12;
     private int serializedSizeDS20;
+    private int serializedSizeDS21;
     private int serializedSizeDSE68;
 
     public int serializedSize(int version)
@@ -520,6 +522,10 @@ public int serializedSize(int version)
                 if (serializedSizeDS20 == 0)
                     serializedSizeDS20 = (int) serializer.serializedSize(this, VERSION_DS_20);
                 return serializedSizeDS20;
+            case VERSION_DS_21:
+                if (serializedSizeDS21 == 0)
+                    serializedSizeDS21 = (int) serializer.serializedSize(this, VERSION_DS_21);
+                return serializedSizeDS21;
             case VERSION_DSE_68:
                 if (serializedSizeDSE68 == 0)
                     serializedSizeDSE68 = (int) serializer.serializedSize(this, VERSION_DSE_68);

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

@@ -24,6 +24,7 @@
 
 import org.apache.cassandra.cache.IMeasurableMemory;
 import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.CellData;
 import org.apache.cassandra.io.ISerializer;
 import org.apache.cassandra.io.sstable.format.Version;
 import org.apache.cassandra.io.util.DataInputPlus;
@@ -36,7 +37,7 @@
 /**
  * Information on deletion of a storage engine object.
  */
-public class DeletionTime implements Comparable<DeletionTime>, IMeasurableMemory
+public class DeletionTime implements Comparable<DeletionTime>, IMeasurableMemory, IDataSize
 {
     public static final long EMPTY_SIZE = ObjectSizes.measure(new DeletionTime(0, 0));
 
@@ -68,9 +69,9 @@ static DeletionTime buildUnsafeWithUnsignedInteger(long markedForDeleteAt, int l
                 : new DeletionTime(markedForDeleteAt, localDeletionTimeUnsignedInteger);
     }
 
-    private DeletionTime(long markedForDeleteAt, long localDeletionTime)
+    protected DeletionTime(long markedForDeleteAt, long localDeletionTime)
     {
-        this(markedForDeleteAt, Cell.deletionTimeLongToUnsignedInteger(localDeletionTime));
+        this(markedForDeleteAt, CellData.deletionTimeLongToUnsignedInteger(localDeletionTime));
     }
 
     private DeletionTime(long markedForDeleteAt, int localDeletionTimeUnsignedInteger)
@@ -95,7 +96,7 @@ public long markedForDeleteAt()
      */
     public long localDeletionTime()
     {
-        return Cell.deletionTimeUnsignedIntegerToLong(localDeletionTimeUnsignedInteger);
+        return CellData.deletionTimeUnsignedIntegerToLong(localDeletionTimeUnsignedInteger);
     }
 
     /**
@@ -143,7 +144,7 @@ public final int hashCode()
     @Override
     public String toString()
     {
-        return this == LIVE ? "LIVE" : String.format("deletedAt=%d, localDeletion=%d", markedForDeleteAt(), localDeletionTime());
+        return this.isLive() ? "LIVE" : String.format("deletedAt=%d, localDeletion=%d", markedForDeleteAt(), localDeletionTime());
     }
 
     public int compareTo(DeletionTime dt)
@@ -170,7 +171,7 @@ public boolean deletes(LivenessInfo info)
         return deletes(info.timestamp());
     }
 
-    public boolean deletes(Cell<?> cell)
+    public boolean deletes(CellData<?, ?> cell)
     {
         return deletes(cell.timestamp());
     }

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

@@ -28,6 +28,7 @@
 import org.apache.cassandra.config.CassandraRelevantProperties;
 import org.apache.cassandra.db.rows.BufferCell;
 import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.CellData;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.service.ClientWarn;
@@ -79,7 +80,7 @@ public static void maybeApplyExpirationDateOverflowPolicy(TableMetadata metadata
 
         // Check for localExpirationTime overflow (CASSANDRA-14092) to apply a policy if needed
         long nowInSecs = currentTimeMillis() / 1000;
-        if (((long) ttl + nowInSecs) > Cell.getVersionedMaxDeletiontionTime())
+        if (((long) ttl + nowInSecs) > CellData.getVersionedMaxDeletiontionTime())
         {
             switch (policy)
             {
@@ -121,13 +122,13 @@ public static long computeLocalExpirationTime(long nowInSec, int timeToLive)
     {
 
         long localExpirationTime = (long) (nowInSec + timeToLive);
-        long cellMaxDeletionTime = Cell.getVersionedMaxDeletiontionTime();
+        long cellMaxDeletionTime = CellData.getVersionedMaxDeletiontionTime();
         return localExpirationTime <= cellMaxDeletionTime ? localExpirationTime : cellMaxDeletionTime;
     }
 
     private static String getMaxExpirationDateTS()
     {
-        return Cell.getVersionedMaxDeletiontionTime() == Cell.MAX_DELETION_TIME_2038_LEGACY_CAP ? "2038-01-19T03:14:06+00:00"
-                                                                                                : "2106-02-07T06:28:13+00:00";
+        return CellData.getVersionedMaxDeletiontionTime() == CellData.MAX_DELETION_TIME_2038_LEGACY_CAP ? "2038-01-19T03:14:06+00:00"
+                                                                                                        : "2106-02-07T06:28:13+00:00";
     }
 }

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

@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db;
+
+/// Shared interface for providing data size information
+public interface IDataSize
+{
+    int dataSize();
+}