DSP-24600 & CASSANDRA-20485 Fixes & test additions for handling of UDTs across versions

All big formats now have implicitly frozen tuples. Originally this was only an issue up to `me`, but C*'s `na` and `nb` formats are also implicitly frozen, are compatibility against them is required.
Fix language around hasExplicitlyFrozenTuples versus hasImplicitlyFrozenTuples in TrieIndexFormat.

Extends SSTableHeaderFix to only run when dropped multi-cell columns are detected.  Add a StartupCheck preventing upgrades from <3.0.25 and <3.11.11, only used by HCD, to reduce the liklihood of unfixed sstable headers being introduced into a running cluster (it is best if the operator manually fixes these headers before the upgrade to HCD 1.x (CC4 / C* 4.0 based).

Additional DSE tries legacy sstable data has been added in addition to the backport described below.

Backport of CASSANDRA-20485, as follows:
Add LegacySSTableTest data for past sstable formats

Also add test method testVerifyOldTupleSSTables for frozen tuples and dropping of them.

The test data also adds different versions of me format sstables from C* version 3.0.25 and 3.11.11 as they do (unfortunately) differ. Test sstable files have had their sequence generation ids changed to represent the server version used to create them, while this shouldn't be necessary it is for debug purposes when sstable formats have changed between versions.

 patch by Roxana Neophytou, Mick Semb Wever; reviewed by Dmitry Konstantinov for CASSANDRA-20485

Co-authored-by: mck <[email protected]>

Author: roxananeo

src/java/org/apache/cassandra/io/sstable/SSTableHeaderFix.java

@@ -23,6 +23,7 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.EnumSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -50,13 +51,16 @@
 import org.apache.cassandra.db.marshal.DynamicCompositeType;
 import org.apache.cassandra.db.marshal.ListType;
 import org.apache.cassandra.db.marshal.MapType;
+import org.apache.cassandra.db.marshal.MultiCellCapableType;
 import org.apache.cassandra.db.marshal.SetType;
 import org.apache.cassandra.db.marshal.TupleType;
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.io.sstable.metadata.MetadataComponent;
 import org.apache.cassandra.io.sstable.metadata.MetadataType;
 import org.apache.cassandra.io.util.File;
 import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.schema.DroppedColumn;
+import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.SchemaConstants;
@@ -84,7 +88,19 @@ public static void fixNonFrozenUDTIfUpgradeFrom30()
         CassandraVersion previousVersion = new CassandraVersion(previousVersionString);
         if (previousVersion.major != 3 || previousVersion.minor > 0)
         {
-            // Not an upgrade from 3.0 to 3.x, nothing to do here
+            // Not an upgrade from 3.0, nothing to do here
+            return;
+        }
+
+        boolean hasDroppedFrozenColumn = false;
+        for (KeyspaceMetadata keyspace : Schema.instance.getUserKeyspaces())
+            for (TableMetadata tableMetadata : keyspace.tablesAndViews())
+                if (!tableMetadata.isIndex() && !tableMetadata.isView() && !tableMetadata.isVirtual())
+                    hasDroppedFrozenColumn |= hasDroppedFrozenColumn(tableMetadata.droppedColumns.values());
+
+        if (!hasDroppedFrozenColumn)
+        {
+            logger.info("No dropped frozen columns found - NOT fixing sstable metadata serialization-headers");
             return;
         }
 
@@ -313,6 +329,16 @@ private static Stream<Path> maybeExpandDirectory(Path path)
                                    .map(File::toPath);
     }
 
+    private static boolean hasDroppedFrozenColumn(Collection<DroppedColumn> droppedColumns)
+    {
+        boolean hasDroppedFrozenColumn = false;
+        for (DroppedColumn droppedColumn : droppedColumns)
+            hasDroppedFrozenColumn |= droppedColumn.column.type.equals(droppedColumn.column.type.freeze())
+                    && droppedColumn.column.type instanceof MultiCellCapableType;
+
+        return hasDroppedFrozenColumn;
+    }
+
     private void processSSTable(Descriptor desc)
     {
         if (desc.cfname.indexOf('.') != -1)
@@ -332,6 +358,12 @@ private void processSSTable(Descriptor desc)
             return;
         }
 
+        if (!hasDroppedFrozenColumn(tableMetadata.droppedColumns.values()))
+        {
+            info.accept(String.format("Table %s.%s has no dropped frozen columns - NOT checking sstable %s", desc.ksname, desc.cfname, desc));
+            return;
+        }
+
         Set<Component> components = SSTable.discoverComponentsFor(desc);
         if (components.stream().noneMatch(c -> c.type == Component.Type.STATS))
         {
@@ -577,8 +609,8 @@ private AbstractType<?> fixType(Descriptor desc, ByteBuffer name, AbstractType<?
                 info.accept(String.format("sstable %s: Column '%s' needs to be updated from type '%s' to '%s'",
                                           desc,
                                           logColumnName(name),
-                                          typeInHeader.asCQL3Type(),
-                                          fixedType.asCQL3Type()));
+                                          typeInHeader.asCQL3Type().toSchemaString(),
+                                          fixedType.asCQL3Type().toSchemaString()));
             return fixedType;
         }
 

src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java

@@ -380,7 +380,7 @@ public boolean hasMaxColumnValueLengths()
         @Override
         public boolean hasImplicitlyFrozenTuples()
         {
-            return false;
+            return true;
         }
 
         @Override

src/java/org/apache/cassandra/io/sstable/format/trieindex/TrieIndexFormat.java

@@ -334,7 +334,7 @@ static class TrieIndexVersion extends Version
         private final boolean hasMaxColumnValueLengths;
 
         private final int correspondingMessagingVersion;
-        private final boolean hasExplicitlyFrozenTuples;
+        private final boolean hasImplicitlyFrozenTuples;
 
         private final ByteComparable.Version byteComparableVersion;
 
@@ -348,7 +348,7 @@ static class TrieIndexVersion extends Version
             hasOriginatingHostId = version.matches("(a[d-z])|(b[b-z])") || version.compareTo("ca") >= 0;
             hasMaxColumnValueLengths = version.matches("b[a-z]"); // DSE only field
             correspondingMessagingVersion = version.compareTo("ca") >= 0 ? MessagingService.VERSION_DS_10 : MessagingService.VERSION_3014;
-            hasExplicitlyFrozenTuples = version.compareTo("cc") < 0 || version.compareTo("da") >= 0; // we don't know if what DA is going to be eventually, but it is almost certain it will not include explicitly frozen tuples
+            hasImplicitlyFrozenTuples = version.compareTo("cc") < 0 || version.compareTo("da") >= 0; // `da` is found in C* 5.0 and CC `main-5.0`, and both have implicitly frozen tuples
             byteComparableVersion = version.compareTo("ca") >= 0 ? ByteComparable.Version.OSS41 : ByteComparable.Version.LEGACY;
         }
 
@@ -501,7 +501,7 @@ public boolean hasIsTransient()
         @Override
         public boolean hasImplicitlyFrozenTuples()
         {
-            return hasExplicitlyFrozenTuples;
+            return hasImplicitlyFrozenTuples;
         }
 
         @Override

test/data/legacy-sstables/aa/legacy_tables/legacy_aa_tuple/.keep

@@ -0,0 +1 @@
+1407325515