Allow recovery from the cases when CQL-created compact sense tables have bytes in EmptyType columns.

Patch by Alex Petrov; reviewed by Sylvain Lebresne for CASSANDRA-15778.

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

@@ -615,6 +615,27 @@ public final class CFMetaData
                         this);
     }
 
+    public CFMetaData copyWithNewCompactValueType(AbstractType<?> type)
+    {
+        assert isDense && compactValueColumn.type instanceof EmptyType && partitionColumns.size() == 1;
+        return copyOpts(new CFMetaData(ksName,
+                                       cfName,
+                                       cfId,
+                                       isSuper,
+                                       isCounter,
+                                       isDense,
+                                       isCompound,
+                                       isView,
+                                       copy(partitionKeyColumns),
+                                       copy(clusteringColumns),
+                                       PartitionColumns.of(compactValueColumn.withNewType(type)),
+                                       partitioner,
+                                       superCfKeyColumn,
+                                       superCfValueColumn),
+                        this);
+    }
+
+
     private static List<ColumnDefinition> copy(List<ColumnDefinition> l)
     {
         List<ColumnDefinition> copied = new ArrayList<>(l.size());

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

@@ -30,6 +30,8 @@ import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.marshal.EmptyType;
 import org.apache.cassandra.db.view.View;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.schema.IndexMetadata;
@@ -109,7 +111,26 @@ public class AlterTableStatement extends SchemaAlteringStatement
         switch (oType)
         {
             case ALTER:
-                throw new InvalidRequestException("Altering of types is not allowed");
+                // We do not support altering of types and only allow this to for people who have already one
+                // through the upgrade of 2.x CQL-created SSTables with Thrift writes, affected by CASSANDRA-15778.
+                if (meta.isDense()
+                    && meta.compactValueColumn().equals(def)
+                    && meta.compactValueColumn().type instanceof EmptyType
+                    && validator != null)
+                {
+                    if (validator.getType() instanceof BytesType)
+                    {
+                        cfm = meta.copyWithNewCompactValueType(validator.getType());
+                        break;
+                    }
+
+                    throw new InvalidRequestException(String.format("Compact value type can only be changed to BytesType, but %s was given.",
+                                                                    validator.getType()));
+                }
+                else
+                {
+                    throw new InvalidRequestException("Altering of types is not allowed");
+                }
             case ADD:
                 assert columnName != null;
                 if (meta.isDense())

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

@@ -394,7 +394,11 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
     public void writeValue(ByteBuffer value, DataOutputPlus out) throws IOException
     {
         assert value.hasRemaining();
-        if (valueLengthIfFixed() >= 0)
+        int valueLengthIfFixed = valueLengthIfFixed();
+        assert valueLengthIfFixed < 0 || value.remaining() == valueLengthIfFixed : String.format("Expected exactly %d bytes, but was %d",
+                                                                                                 valueLengthIfFixed, value.remaining());
+
+        if (valueLengthIfFixed >= 0)
             out.write(value);
         else
             ByteBufferUtil.writeWithVIntLength(value, out);
@@ -403,7 +407,11 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>
     public long writtenLength(ByteBuffer value)
     {
         assert value.hasRemaining();
-        return valueLengthIfFixed() >= 0
+        int valueLengthIfFixed = valueLengthIfFixed();
+        assert valueLengthIfFixed < 0 || value.remaining() == valueLengthIfFixed : String.format("Expected exactly %d bytes, but was %d",
+                                                                                                 valueLengthIfFixed, value.remaining());
+
+        return valueLengthIfFixed >= 0
              ? value.remaining()
              : TypeSizes.sizeofWithVIntLength(value);
     }

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

@@ -632,9 +632,12 @@ public final class LegacySchemaMigrator
         }
         else
         {
-            // For dense compact tables, we get here if we don't have a compact value column, in which case we should add it
-            // (we use EmptyType to recognize that the compact value was not declared by the use (see CreateTableStatement too))
-            defs.add(ColumnDefinition.regularDef(ksName, cfName, names.defaultCompactValueName(), EmptyType.instance));
+            // For dense compact tables, we get here if we don't have a compact value column, in which case we should add it.
+            // We use EmptyType to recognize that the compact value was not declared by the user (see CreateTableStatement).
+            // If user made any writes to this column, compact value column should be initialized as bytes (see CASSANDRA-15778).
+            AbstractType<?> compactColumnType = Boolean.getBoolean("cassandra.init_dense_table_compact_value_as_bytes")
+                                                ? BytesType.instance : EmptyType.instance;
+            defs.add(ColumnDefinition.regularDef(ksName, cfName, names.defaultCompactValueName(), compactColumnType));
         }
     }
 

src/java/org/apache/cassandra/serializers/EmptySerializer.java

@@ -40,7 +40,11 @@ public class EmptySerializer implements TypeSerializer<Void>
     public void validate(ByteBuffer bytes) throws MarshalException
     {
         if (bytes.remaining() > 0)
-            throw new MarshalException("EmptyType only accept empty values");
+        {
+            throw new MarshalException("EmptyType only accept empty values. " +
+                                       "A non-empty value can be a result of a Thrift write into CQL-created dense table. " +
+                                       "See CASSANDRA-15778 for details.");
+        }
     }
 
     public String toString(Void value)

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_cql_created_dense_table_with_bytes/legacy_tables-legacy_ka_cql_created_dense_table_with_bytes-ka-1-Digest.sha1

@@ -0,0 +1 @@
+3389985016

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_cql_created_dense_table_with_bytes/legacy_tables-legacy_ka_cql_created_dense_table_with_bytes-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Digest.sha1
+CompressionInfo.db
+Summary.db
+Statistics.db
+Data.db
+Index.db
+TOC.txt
+Filter.db

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_cql_created_dense_table_with_int/legacy_tables-legacy_ka_cql_created_dense_table_with_int-ka-1-Digest.sha1

@@ -0,0 +1 @@
+1334250623

test/data/legacy-sstables/ka/legacy_tables/legacy_ka_cql_created_dense_table_with_int/legacy_tables-legacy_ka_cql_created_dense_table_with_int-ka-1-TOC.txt

@@ -0,0 +1,8 @@
+Digest.sha1
+CompressionInfo.db
+Summary.db
+Statistics.db
+Data.db
+Index.db
+TOC.txt
+Filter.db

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

@@ -23,6 +23,7 @@ import org.junit.Test;
 import org.apache.cassandra.cql3.CQLTester;
 import org.apache.cassandra.db.ColumnFamilyStore;
 import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.IntegerType;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.SyntaxException;
@@ -506,4 +507,25 @@ public class AlterTest extends CQLTester
         assertInvalidMessage("Cannot rename unknown column column1 in keyspace",
                              "ALTER TABLE %s RENAME column1 TO column2");
     }
+
+    @Test
+    public void testAlterTableAlterType() throws Throwable
+    {
+        createTable("CREATE TABLE %s (a int, b int, PRIMARY KEY (a,b)) WITH COMPACT STORAGE");
+        assertInvalidMessage(String.format("Compact value type can only be changed to BytesType, but %s was given.",
+                                           IntegerType.instance),
+                             "ALTER TABLE %s ALTER value TYPE 'org.apache.cassandra.db.marshal.IntegerType'");
+
+        execute("ALTER TABLE %s ALTER value TYPE 'org.apache.cassandra.db.marshal.BytesType'");
+
+        createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a,b)) WITH COMPACT STORAGE");
+        assertInvalidMessage("Altering of types is not allowed",
+                             "ALTER TABLE %s ALTER c TYPE 'org.apache.cassandra.db.marshal.BytesType'");
+
+        createTable("CREATE TABLE %s (a int, value int, PRIMARY KEY (a,value)) WITH COMPACT STORAGE");
+        assertInvalidMessage("Altering of types is not allowed",
+                             "ALTER TABLE %s ALTER value TYPE 'org.apache.cassandra.db.marshal.IntegerType'");
+        execute("ALTER TABLE %s ALTER value1 TYPE 'org.apache.cassandra.db.marshal.BytesType'");
+    }
+
 }

test/unit/org/apache/cassandra/db/ScrubTest.java

@@ -42,6 +42,7 @@ import org.apache.cassandra.db.compaction.CompactionManager;
 import org.apache.cassandra.db.compaction.OperationType;
 import org.apache.cassandra.db.compaction.Scrubber;
 import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
+import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.marshal.LongType;
 import org.apache.cassandra.db.marshal.UUIDType;
 import org.apache.cassandra.db.partitions.Partition;
@@ -509,7 +510,7 @@ public class ScrubTest
         QueryProcessor.process("CREATE TABLE \"Keyspace1\".test_scrub_validation (a text primary key, b int)", ConsistencyLevel.ONE);
         ColumnFamilyStore cfs2 = keyspace.getColumnFamilyStore("test_scrub_validation");
 
-        new Mutation(UpdateBuilder.create(cfs2.metadata, "key").newRow().add("b", LongType.instance.decompose(1L)).build()).apply();
+        new Mutation(UpdateBuilder.create(cfs2.metadata, "key").newRow().add("b", Int32Type.instance.decompose(1)).build()).apply();
         cfs2.forceBlockingFlush();
 
         CompactionManager.instance.performScrub(cfs2, false, false, 2);

test/unit/org/apache/cassandra/db/marshal/EmptyTypeTest.java

@@ -88,7 +88,7 @@ public class EmptyTypeTest
             Assert.fail("compose is expected to reject non-empty values, but did not");
         }
         catch (MarshalException e) {
-            Assert.assertEquals("EmptyType only accept empty values", e.getMessage());
+            Assert.assertTrue(e.getMessage().startsWith("EmptyType only accept empty values"));
         }
     }
 }

test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java

@@ -790,6 +790,43 @@ public class LegacySSTableTest
         cfs.loadNewSSTables();
     }
 
+
+    /**
+     * Test for CASSANDRA-15778
+     */
+    @Test
+    public void testReadLegacyCqlCreatedTableWithBytes() throws Exception {
+        String table = "legacy_ka_cql_created_dense_table_with_bytes";
+        QueryProcessor.executeInternal("CREATE TABLE legacy_tables." + table + " (" +
+                                       " k int," +
+                                       " v text," +
+                                       " PRIMARY KEY(k, v)) WITH COMPACT STORAGE");
+        loadLegacyTable("legacy_%s_cql_created_dense_table_with_bytes%s", "ka", "");
+        QueryProcessor.executeInternal("ALTER TABLE legacy_tables." + table + " ALTER value TYPE 'org.apache.cassandra.db.marshal.BytesType';");
+        UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM legacy_tables." + table);
+        Assert.assertNotNull(rs);
+        assertEquals(1, rs.size());
+        assertEquals(ByteBufferUtil.bytes("byte string"), rs.one().getBytes("value"));
+    }
+
+    /**
+     * Test for CASSANDRA-15778
+     */
+    @Test
+    public void testReadLegacyCqlCreatedTableWithInt() throws Exception {
+        String table = "legacy_ka_cql_created_dense_table_with_int";
+        QueryProcessor.executeInternal("CREATE TABLE legacy_tables." + table + " (" +
+                                       " k int," +
+                                       " v text," +
+                                       " PRIMARY KEY(k, v)) WITH COMPACT STORAGE");
+        loadLegacyTable("legacy_%s_cql_created_dense_table_with_int%s", "ka", "");
+        QueryProcessor.executeInternal("ALTER TABLE legacy_tables." + table + " ALTER value TYPE 'org.apache.cassandra.db.marshal.BytesType';");
+        UntypedResultSet rs = QueryProcessor.executeInternal("SELECT * FROM legacy_tables." + table);
+        Assert.assertNotNull(rs);
+        assertEquals(1, rs.size());
+        assertEquals(ByteBufferUtil.bytes(0xaabbcc), rs.one().getBytes("value"));
+    }
+
     /**
      * Generates sstables for 8 CQL tables (see {@link #createTables(String)}) in <i>current</i>
      * sstable format (version) into {@code test/data/legacy-sstables/VERSION}, where