CNDB-15995 Add CC 4.0 schema backward compatibility mode

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

@@ -173,7 +173,7 @@ private TableParams build(TableParams.Builder builder)
 
         if (hasOption(MEMTABLE))
         {
-            // Handle deserialzation of Astra/CC 4.0 schema with memtable option as a map
+            // Handle deserialization of Astra/CC 4.0 schema with memtable option as a map
             if (properties.get(MEMTABLE.toString()) instanceof Map)
             {
                 String memtableClass = getMap(MEMTABLE)
@@ -183,15 +183,38 @@ private TableParams build(TableParams.Builder builder)
                              .map(Map.Entry::getValue)
                              .findFirst()
                              .orElse(null);
-                // Not exhaustive, but avoids raising an error upgrading from a CC 4.0 schema
+
                 if (memtableClass == null)
+                {
                     builder.memtable(MemtableParams.get(null));
-                else if ("SkipListMemtable".equalsIgnoreCase(memtableClass))
-                    builder.memtable(MemtableParams.get("skiplist"));
-                else if ("PersistentMemoryMemtable".equalsIgnoreCase(memtableClass))
-                    builder.memtable(MemtableParams.get("persistent_memory"));
+                }
+                // Only process as a known memtable if it's in the standard package or is a short name (no package qualifier)
+                else if (memtableClass.startsWith("org.apache.cassandra.db.memtable.") || !memtableClass.contains("."))
+                {
+                    // Extract short class name for comparison against known types
+                    String shortClassName = memtableClass.contains(".")
+                                            ? memtableClass.substring(memtableClass.lastIndexOf('.') + 1)
+                                            : memtableClass;
+
+                    if ("SkipListMemtable".equalsIgnoreCase(shortClassName))
+                        builder.memtable(MemtableParams.get("skiplist"));
+                    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 ("ShardedSkipListMemtable".equalsIgnoreCase(shortClassName))
+                        builder.memtable(MemtableParams.get("skiplist_sharded"));
+                    else
+                        // Unknown short name or unknown class in standard package - use as configuration key
+                        builder.memtable(MemtableParams.get(shortClassName));
+                }
                 else
-                    builder.memtable(MemtableParams.get("trie"));
+                {
+                    // Custom fully qualified class name from a different package.
+                    builder.memtable(MemtableParams.get(memtableClass));
+                }
             }
             else
                 builder.memtable(MemtableParams.get(getString(MEMTABLE)));

src/java/org/apache/cassandra/schema/MemtableParams.java

@@ -69,6 +69,37 @@ public Memtable.Factory factory()
         return factory;
     }
 
+    /**
+     * Returns a map representation of the memtable configuration for backward compatibility with CC 4.0.
+     * This is used when outputting schema in a format compatible with CC 4.0.
+     *
+     * CC 4.0 accepts both short class names (e.g., 'TrieMemtable') and fully qualified names
+     * (e.g., 'org.apache.cassandra.db.memtable.TrieMemtable'). For standard Cassandra memtables
+     * in the org.apache.cassandra.db.memtable package, we use short names and for custom memtables
+     * from other packages, we preserve the fully qualified class name.
+     */
+    public Map<String, String> toMapForCC4()
+    {
+        ParameterizedClass definition = CONFIGURATION_DEFINITIONS.get(configurationKey);
+        if (definition != null && definition.class_name != null)
+        {
+            Map<String, String> map = new HashMap<>();
+            String className = definition.class_name;
+
+            if (className.startsWith("org.apache.cassandra.db.memtable."))
+            {
+                className = className.substring("org.apache.cassandra.db.memtable.".length());
+            }
+
+            map.put("class", className);
+            if (definition.parameters != null)
+                map.putAll(definition.parameters);
+            return map;
+        }
+        // Fallback for unknown configurations
+        return ImmutableMap.of("class", configurationKey);
+    }
+
     @Override
     public String toString()
     {

src/java/org/apache/cassandra/schema/TableParams.java

@@ -21,6 +21,8 @@
 import java.util.Map;
 import java.util.Map.Entry;
 
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.cassandra.utils.StorageCompatibilityMode;
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Objects;
 import com.google.common.collect.ImmutableMap;
@@ -69,6 +71,9 @@ public String toString()
         }
     }
 
+    @VisibleForTesting
+    public static StorageCompatibilityMode storageCompatibilityModeOverride = null;
+
     public final String comment;
     public final boolean allowAutoSnapshot;
     public final double bloomFilterFpChance;
@@ -285,12 +290,23 @@ public String toString()
 
     public void appendCqlTo(CqlBuilder builder, boolean isView)
     {
+        StorageCompatibilityMode compatibilityMode = storageCompatibilityModeOverride != null
+                                                      ? storageCompatibilityModeOverride
+                                                      : StorageCompatibilityMode.current();
+        boolean usePre50Schema = compatibilityMode.isBefore(5);
+
         // option names should be in alphabetical order
         builder.append("additional_write_policy = ").appendWithSingleQuotes(additionalWritePolicy.toString())
-               .newLine()
-               .append("AND allow_auto_snapshot = ").append(allowAutoSnapshot)
-               .newLine()
-               .append("AND bloom_filter_fp_chance = ").append(bloomFilterFpChance)
+               .newLine();
+
+        // Exclude allow_auto_snapshot in backward compatibility mode (new in 5.0)
+        if (!usePre50Schema)
+        {
+            builder.append("AND allow_auto_snapshot = ").append(allowAutoSnapshot)
+                   .newLine();
+        }
+
+        builder.append("AND bloom_filter_fp_chance = ").append(bloomFilterFpChance)
                .newLine()
                .append("AND caching = ").append(caching.asMap())
                .newLine()
@@ -301,9 +317,15 @@ public void appendCqlTo(CqlBuilder builder, boolean isView)
                .append("AND compaction = ").append(compaction.asMap())
                .newLine()
                .append("AND compression = ").append(compression.asMap())
-               .newLine()
-               .append("AND memtable = ").appendWithSingleQuotes(memtable.configurationKey())
-               .newLine()
+               .newLine();
+
+        // Use map format for pre-5.0 compatibility, string format for 5.0
+        if (usePre50Schema)
+            builder.append("AND memtable = ").append(memtable.toMapForCC4());
+        else
+            builder.append("AND memtable = ").appendWithSingleQuotes(memtable.configurationKey());
+
+        builder.newLine()
                .append("AND crc_check_chance = ").append(crcCheckChance)
                .newLine();
 
@@ -320,10 +342,16 @@ public void appendCqlTo(CqlBuilder builder, boolean isView)
                                                    false)
                .newLine()
                .append("AND gc_grace_seconds = ").append(gcGraceSeconds)
-               .newLine()
-               .append("AND incremental_backups = ").append(incrementalBackups)
-               .newLine()
-               .append("AND max_index_interval = ").append(maxIndexInterval)
+               .newLine();
+
+        // Exclude incremental_backups in backward compatibility mode (new in 5.0)
+        if (!usePre50Schema)
+        {
+            builder.append("AND incremental_backups = ").append(incrementalBackups)
+                   .newLine();
+        }
+
+        builder.append("AND max_index_interval = ").append(maxIndexInterval)
                .newLine()
                .append("AND memtable_flush_period_in_ms = ").append(memtableFlushPeriodInMs)
                .newLine()

src/java/org/apache/cassandra/utils/StorageCompatibilityMode.java

@@ -35,6 +35,11 @@ public enum StorageCompatibilityMode
      */
     CASSANDRA_4(4),
 
+    /**
+     * Same as {@link #CASSANDRA_4}, but allows the use of BTI format in {@link #validateSstableFormat}.
+     */
+    CC_4(4),
+
     /**
      * Use the storage formats of the current version, but disabling features that are not compatible with any
      * not-upgraded nodes in the cluster. Use this during rolling upgrades to a new major Cassandra version. Once all