HCD-241: Preserve user-defined types during keyspace schema migration

When updating keyspace schemas, UDTs from the previous schema are now
preserved if they don't exist in the new schema.
This prevents issues where inherited tables depend on types that would
otherwise be lost during schema transformations.

Author: szymon-miezal

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

@@ -20,6 +20,9 @@
 
 import java.util.Optional;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.exceptions.AlreadyExistsException;
 import org.apache.cassandra.exceptions.ConfigurationException;
@@ -31,6 +34,8 @@
  */
 public class SchemaTransformations
 {
+    private static final Logger logger = LoggerFactory.getLogger(SchemaTransformations.class);
+
     /**
      * Creates a schema transformation that adds the provided keyspace.
      *
@@ -227,6 +232,22 @@ public Keyspaces apply(Keyspaces schema)
                             updatedKeyspace = updatedKeyspace.withSwapped(updatedKeyspace.tables.with(updatedBuilder.build()));
                         }
                     }
+
+                    if (curKeyspace.types != null)
+                    {
+                        for (UserType currType : curKeyspace.types)
+                        {
+                            UserType desiredType = updatedKeyspace.types.getNullable(currType.name);
+                            // if the type exist we keep the existing definition, otherwise we inherit the type
+                            // the motivation behind it is that there might be tables (inherited above) that depend on it
+                            if (desiredType == null)
+                            {
+                                logger.debug("Preserving type {} for keyspace {}", currType.getNameAsString(), curKeyspace.name);
+                                updatedKeyspace = updatedKeyspace.withSwapped(updatedKeyspace.types.with(currType));
+                            }
+                        }
+                    }
+
                 }
                 return schema.withAddedOrUpdated(updatedKeyspace);
             }

test/unit/org/apache/cassandra/schema/MigrationManagerTest.java

@@ -20,6 +20,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 
@@ -40,7 +41,9 @@
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
 import org.apache.cassandra.db.marshal.ByteType;
 import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.UserType;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.io.sstable.Component;
 import org.apache.cassandra.io.sstable.Descriptor;
@@ -51,7 +54,9 @@
 import static org.apache.cassandra.Util.throwAssert;
 import static org.apache.cassandra.cql3.CQLTester.assertRows;
 import static org.apache.cassandra.cql3.CQLTester.row;
+import static org.apache.cassandra.cql3.FieldIdentifier.forUnquoted;
 import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS;
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -617,6 +622,32 @@ public void testEvolveSystemKeyspaceChanged()
         assertEquals(keyspace1, diff.altered.get(0).after);
     }
 
+    @Test
+    public void testInheritTypesFromPreviousKeyspace()
+    {
+        // given
+        TableMetadata table0 = addTestTable("ks2", "t", "");
+        UserType udt = new UserType("ks2",
+                                    bytes("PreviousType"),
+                                    List.of(forUnquoted("a"), forUnquoted("b")),
+                                    List.of(Int32Type.instance, Int32Type.instance),
+                                    true);
+        KeyspaceMetadata previousKeyspace = KeyspaceMetadata.create("ks2", KeyspaceParams.simple(1), Tables.of(table0), Views.none(), Types.of(udt), UserFunctions.none());
+        KeyspaceMetadata currentKeyspace = KeyspaceMetadata.create("ks2", KeyspaceParams.simple(1), Tables.of(table0));
+
+        // when
+        SchemaTransformation transformation = SchemaTransformations.updateSystemKeyspace(currentKeyspace, 1);
+        Keyspaces after = transformation.apply(Keyspaces.of(previousKeyspace));
+
+        // then
+        KeyspaceMetadata keyspaceAfterTransformation = after.getNullable("ks2");
+        assertNotNull(keyspaceAfterTransformation);
+        assertEquals(currentKeyspace.types.stream().count() + 1, keyspaceAfterTransformation.types.stream().count());
+        UserType inheritedUdt = keyspaceAfterTransformation.types.getNullable(udt.name);
+        assertNotNull(inheritedUdt);
+        assertEquals(udt, inheritedUdt);
+    }
+
     private TableMetadata addTestTable(String ks, String cf, String comment)
     {
         return