Backport CASSANDRA-18760

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

@@ -226,7 +226,7 @@ private void addColumn(KeyspaceMetadata keyspace,
             {
                 // After #8099, not safe to re-add columns of incompatible types - until *maybe* deser logic with dropped
                 // columns is pushed deeper down the line. The latter would still be problematic in cases of schema races.
-                if (!type.isValueCompatibleWith(droppedColumn.type))
+                if (!type.isSerializationCompatibleWith(droppedColumn.type))
                 {
                     throw ire("Cannot re-add previously dropped column '%s' of type %s, incompatible with previous type %s",
                               name,

src/java/org/apache/cassandra/db/marshal/AbstractType.java

@@ -382,6 +382,18 @@ protected boolean isValueCompatibleWithInternal(AbstractType<?> otherType)
         return isCompatibleWith(otherType);
     }
 
+    /**
+     * Similar to {@link #isValueCompatibleWith(AbstractType)}, but takes into account {@link Cell} encoding.
+     * In particular, this method doesn't consider two types serialization compatible if one of them has fixed
+     * length (overrides {@link #valueLengthIfFixed()}, and the other one doesn't.
+     */
+    public boolean isSerializationCompatibleWith(AbstractType<?> previous)
+    {
+        return isValueCompatibleWith(previous)
+               && valueLengthIfFixed() == previous.valueLengthIfFixed()
+               && isMultiCell() == previous.isMultiCell();
+    }
+
     /**
      * An alternative comparison function used by CollectionsType in conjunction with CompositeType.
      *

src/java/org/apache/cassandra/db/marshal/CollectionType.java

@@ -157,7 +157,7 @@ protected boolean isCompatibleWhenFrozenWith(AbstractType<?> previous)
         // nameComparator() is hard-coded), but this method is not so performance sensitive that it's worth bothering.
         CollectionType<?> prev = (CollectionType<?>)previous;
         return nameComparator().isCompatibleWith(prev.nameComparator())
-               && valueComparator().isCompatibleWith(prev.valueComparator());
+               && valueComparator().isSerializationCompatibleWith(prev.valueComparator());
     }
 
     @Override
@@ -168,7 +168,7 @@ protected boolean isCompatibleWhenNonFrozenWith(AbstractType<?> previous)
         // for sorting so value-compatibility is enough.
         CollectionType<?> prev = (CollectionType<?>)previous;
         return nameComparator().isCompatibleWith(prev.nameComparator())
-               && valueComparator().isValueCompatibleWith(prev.valueComparator());
+               && valueComparator().isSerializationCompatibleWith(prev.valueComparator());
     }
 
     @Override
@@ -177,7 +177,7 @@ protected boolean isValueCompatibleWhenFrozenWith(AbstractType<?> previous)
         // When frozen, the full collection is a blob, so value-compatibility is all we care for everything.
         CollectionType<?> prev = (CollectionType<?>)previous;
         return nameComparator().isValueCompatibleWith(prev.nameComparator())
-               && valueComparator().isValueCompatibleWith(prev.valueComparator());
+               && valueComparator().isSerializationCompatibleWith(prev.valueComparator());
     }
 
     public CQL3Type asCQL3Type()

test/unit/org/apache/cassandra/cql3/validation/operations/AlterTest.java

@@ -30,6 +30,7 @@
 import org.apache.cassandra.db.memtable.TrieMemtable;
 import org.apache.cassandra.dht.OrderPreservingPartitioner;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.SyntaxException;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.locator.TokenMetadata;
@@ -445,6 +446,37 @@ public void testDowngradeToCompact() throws Throwable
         alterTable("alter table %s add v1 int");
     }
 
+    @Test(expected = InvalidRequestException.class)
+    public void testDropFixedAddVariable() throws Throwable
+    {
+        createTable("create table %s (k int, c int, v int, PRIMARY KEY (k, c))");
+        execute("alter table %s drop v");
+        execute("alter table %s add v varint");
+    }
+
+    @Test(expected = InvalidRequestException.class)
+    public void testDropFixedCollectionAddVariableCollection() throws Throwable
+    {
+        createTable("create table %s (k int, c int, v list<int>, PRIMARY KEY (k, c))");
+        execute("alter table %s drop v");
+        execute("alter table %s add v list<varint>");
+    }
+
+    @Test(expected = InvalidRequestException.class)
+    public void testDropSimpleAddComplex() throws Throwable
+    {
+        createTable("create table %s (k int, c int, v set<text>, PRIMARY KEY (k, c))");
+        execute("alter table %s drop v");
+        execute("alter table %s add v blob");
+    }
+
+    @Test(expected = SyntaxException.class)
+    public void renameToEmptyTest() throws Throwable
+    {
+        createTable("CREATE TABLE %s (k int, c1 int, v int, PRIMARY KEY (k, c1))");
+        execute("ALTER TABLE %s RENAME c1 TO \"\"");
+    }
+
     @Test
     // tests CASSANDRA-9565
     public void testDoubleWith() throws Throwable