LegacyLayout should handle paging states that cross a collection column

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.19
+ * LegacyLayout should handle paging states that cross a collection column (CASSANDRA-15201)
  * Prevent RuntimeException when username or password is empty/null (CASSANDRA-15198)
  * Multiget thrift query returns null records after digest mismatch (CASSANDRA-14812)
  * Skipping illegal legacy cells can break reverse iteration of indexed partitions (CASSANDRA-15178)

src/java/org/apache/cassandra/db/LegacyLayout.java

@@ -24,6 +24,7 @@
 import java.security.MessageDigest;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.apache.cassandra.cql3.ColumnIdentifier;
 import org.apache.cassandra.cql3.SuperColumnCompatibility;
@@ -235,36 +236,45 @@ private static LegacyBound decodeBound(CFMetaData metadata, ByteBuffer bound, bo
         assert !isStatic ||
                 (components.size() >= clusteringSize
                         && all(components.subList(0, clusteringSize), ByteBufferUtil.EMPTY_BYTE_BUFFER::equals));
+
         ColumnDefinition collectionName = null;
         if (components.size() > clusteringSize)
         {
             // For a deletion, there can be more components than the clustering size only in the case this is the
             // bound of a collection range tombstone. In such a case, there is exactly one more component, and that
-            // component is the name of the collection being selected/deleted.
-            // If the bound is not part of a deletion, it is from slice query filter. In this scnario, the column name
-            // may be a valid, non-collection column or it may be an empty buffer, representing a row marker. In either
-            // case, this needn't be included in the returned bound, so we pop the last element from the components
-            // list but ensure that the collection name remains null.
-
-            assert clusteringSize + 1 == components.size() && !metadata.isCompactTable();
-            // pop the final element from the back of the list of clusterings
-            ByteBuffer columnNameBytes = components.remove(clusteringSize);
-            if (isDeletion)
+            // component is the name of the collection being deleted, since we do not support collection range deletions.
+            // If the bound is not part of a deletion, it is from slice query filter. The column name may be:
+            //   - a valid, non-collection column; in this case we expect a single extra component
+            //   - an empty buffer, representing a row marker; in this case we also expect a single extra empty component
+            //   - a valid collection column and the first part of a cell path; in this case we expect exactly two extra components
+            // In any of these slice cases, these items are unnecessary for the bound we construct,
+            // so we can simply remove them, after corroborating we have encountered one of these scenario.
+            assert !metadata.isCompactTable() : toDebugHex(components);
+
+            // In all cases, the element straight after the clusterings should contain the name of a column.
+            if (components.size() > clusteringSize + 1)
             {
-                collectionName = metadata.getColumnDefinition(columnNameBytes);
-                if (collectionName == null || !collectionName.isComplex())
-                {
-                    collectionName = metadata.getDroppedColumnDefinition(columnNameBytes, isStatic);
-                    // if no record of the column having ever existed is found, something is badly wrong
-                    if (collectionName == null)
-                        throw new RuntimeException("Unknown collection column " + UTF8Type.instance.getString(columnNameBytes) + " during deserialization");
-
-                    // if we do have a record of dropping this column but it wasn't previously complex, use a fake
-                    // column definition for safety (see the comment on the constant declaration for details)
-                    if (!collectionName.isComplex())
-                        collectionName = INVALID_DROPPED_COMPLEX_SUBSTITUTE_COLUMN;
-                }
+                // we accept bounds from paging state that occur inside a complex column - in this case, we expect
+                // two excess components, the first of which is a column name, the second a key into the collection
+                if (isDeletion)
+                    throw new IllegalArgumentException("Invalid bound " + toDebugHex(components) + ": deletion can have at most one extra component");
+
+                if (clusteringSize + 2 != components.size())
+                    throw new IllegalArgumentException("Invalid bound " + toDebugHex(components) + ": complex slices require exactly two extra components");
+
+                // decode simply to verify that we have (or may have had) a complex column; we assume the collection key is valid
+                decodeBoundLookupComplexColumn(metadata, components, clusteringSize, isStatic);
+                components.remove(clusteringSize + 1);
+            }
+            else if (isDeletion)
+            {
+                collectionName = decodeBoundLookupComplexColumn(metadata, components, clusteringSize, isStatic);
+            }
+            else if (components.get(clusteringSize).hasRemaining())
+            {
+                decodeBoundVerifySimpleColumn(metadata, components, clusteringSize, isStatic);
             }
+            components.remove(clusteringSize);
         }
 
         boolean isInclusive;
@@ -292,6 +302,48 @@ private static LegacyBound decodeBound(CFMetaData metadata, ByteBuffer bound, bo
         return new LegacyBound(sb, isStatic, collectionName);
     }
 
+    // finds the simple column definition associated with components.get(clusteringSize)
+    // if no such columns exists, or ever existed, we throw an exception; if we do not know, we return a dummy column definition
+    private static ColumnDefinition decodeBoundLookupComplexColumn(CFMetaData metadata, List<ByteBuffer> components, int clusteringSize, boolean isStatic)
+    {
+        ByteBuffer columnNameBytes = components.get(clusteringSize);
+        ColumnDefinition columnName = metadata.getColumnDefinition(columnNameBytes);
+        if (columnName == null || !columnName.isComplex())
+        {
+            columnName = metadata.getDroppedColumnDefinition(columnNameBytes, isStatic);
+            // if no record of the column having ever existed is found, something is badly wrong
+            if (columnName == null)
+                throw new IllegalArgumentException("Invalid bound " + toDebugHex(components) + ": expected complex column at position " + clusteringSize);
+
+            // if we do have a record of dropping this column but it wasn't previously complex, use a fake
+            // column definition for safety (see the comment on the constant declaration for details)
+            if (!columnName.isComplex())
+                columnName = INVALID_DROPPED_COMPLEX_SUBSTITUTE_COLUMN;
+        }
+
+        return columnName;
+    }
+
+    // finds the simple column definition associated with components.get(clusteringSize)
+    // if no such columns exists, and definitely never existed, we throw an exception
+    private static void decodeBoundVerifySimpleColumn(CFMetaData metadata, List<ByteBuffer> components, int clusteringSize, boolean isStatic)
+    {
+        ByteBuffer columnNameBytes = components.get(clusteringSize);
+        ColumnDefinition columnName = metadata.getColumnDefinition(columnNameBytes);
+        if (columnName == null || !columnName.isSimple())
+        {
+            columnName = metadata.getDroppedColumnDefinition(columnNameBytes, isStatic);
+            // if no record of the column having ever existed is found, something is badly wrong
+            if (columnName == null)
+                throw new IllegalArgumentException("Invalid bound " + toDebugHex(components) + ": expected simple column at position " + clusteringSize);
+        }
+    }
+
+    private static String toDebugHex(Collection<ByteBuffer> buffers)
+    {
+        return buffers.stream().map(ByteBufferUtil::bytesToHex).collect(Collectors.joining());
+    }
+
     public static ByteBuffer encodeBound(CFMetaData metadata, Slice.Bound bound, boolean isStart)
     {
         if (bound == Slice.Bound.BOTTOM || bound == Slice.Bound.TOP || metadata.comparator.size() == 0)

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

@@ -38,6 +38,7 @@
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.serializers.Int32Serializer;
 import org.apache.cassandra.serializers.UTF8Serializer;
+import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.utils.FBUtilities;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -57,6 +58,7 @@
 import org.apache.cassandra.dht.Murmur3Partitioner;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.Hex;
 
 import static org.junit.Assert.*;
 
@@ -291,4 +293,56 @@ public void testCollectionDeletionRoundTripForDroppedColumn() throws Throwable
         }
     }
 
+    @Test
+    public void testDecodeLegacyPagedRangeCommandSerializer() throws IOException
+    {
+        /*
+         Run on 2.1
+         public static void main(String[] args) throws IOException, ConfigurationException
+         {
+             Gossiper.instance.start((int) (System.currentTimeMillis() / 1000));
+             Keyspace.setInitialized();
+             CFMetaData cfMetaData = CFMetaData.sparseCFMetaData("ks", "cf", UTF8Type.instance)
+             .addColumnDefinition(new ColumnDefinition("ks", "cf", new ColumnIdentifier("v", true), SetType.getInstance(Int32Type.instance, false), null, null, null, null, ColumnDefinition.Kind.REGULAR));
+             KSMetaData ksMetaData = KSMetaData.testMetadata("ks", SimpleStrategy.class, KSMetaData.optsWithRF(3), cfMetaData);
+             MigrationManager.announceNewKeyspace(ksMetaData);
+             RowPosition position = RowPosition.ForKey.get(ByteBufferUtil.EMPTY_BYTE_BUFFER, new Murmur3Partitioner());
+             SliceQueryFilter filter = new IdentityQueryFilter();
+             Composite cellName = CellNames.compositeSparseWithCollection(new ByteBuffer[0], Int32Type.instance.decompose(1), new ColumnIdentifier("v", true), false);
+             try (DataOutputBuffer buffer = new DataOutputBuffer(1024))
+             {
+                 PagedRangeCommand command = new PagedRangeCommand("ks", "cf", 1, AbstractBounds.bounds(position, true, position, true), filter, cellName, filter.finish(), Collections.emptyList(), 1, true);
+                 PagedRangeCommand.serializer.serialize(command, buffer, MessagingService.current_version);
+                 System.out.println(Hex.bytesToHex(buffer.toByteArray()));
+             }
+         }
+         */
+
+        DatabaseDescriptor.setDaemonInitialized();
+        Keyspace.setInitialized();
+        CFMetaData table = CFMetaData.Builder.create("ks", "cf")
+                                             .addPartitionKey("k", Int32Type.instance)
+                                             .addRegularColumn("v", SetType.getInstance(Int32Type.instance, true))
+                                             .build();
+        SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1));
+        MigrationManager.announceNewColumnFamily(table);
+
+        byte[] bytes = Hex.hexToBytes("00026b73000263660000000000000001fffffffe01000000088000000000000000010000000880000000000000000000000100000000007fffffffffffffff000b00017600000400000001000000000000000000000101");
+        ReadCommand.legacyPagedRangeCommandSerializer.deserialize(new DataInputBuffer(bytes), MessagingService.VERSION_21);
+    }
+
+    @Test
+    public void testDecodeCollectionPageBoundary()
+    {
+        CFMetaData table = CFMetaData.Builder.create("ks", "cf")
+                                             .addPartitionKey("k", Int32Type.instance)
+                                             .addRegularColumn("v", SetType.getInstance(Int32Type.instance, true))
+                                             .build();
+
+        ColumnDefinition v = table.getColumnDefinition(new ColumnIdentifier("v", false));
+        ByteBuffer bound = LegacyLayout.encodeCellName(table, Clustering.EMPTY, v.name.bytes, Int32Type.instance.decompose(1));
+
+        LegacyLayout.decodeSliceBound(table, bound, true);
+    }
+
 }