Fix missing rows when reading 2.1 SSTables with static columns in 3.0

patch by Aleksey Yeschenko; reviewed by Blake Eggleston for
CASSANDRA-14873

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.18
+ * Fix missing rows when reading 2.1 SSTables with static columns in 3.0 (CASSANDRA-14873)
  * Move TWCS message 'No compaction necessary for bucket size' to Trace level (CASSANDRA-14884)
  * Sstable min/max metadata can cause data loss (CASSANDRA-14861)
  * Dropped columns can cause reverse sstable iteration to return prematurely (CASSANDRA-14838)

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

@@ -79,6 +79,9 @@ public class Serializers
                 if (clusteringSize == 0)
                     return Clustering.EMPTY;
 
+                if (metadata.isCompound() && CompositeType.isStaticName(bb))
+                    return Clustering.STATIC_CLUSTERING;
+
                 if (!metadata.isCompound())
                     return new Clustering(bb);
 

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

@@ -0,0 +1 @@
+4060752841

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

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

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

@@ -242,6 +242,44 @@ public class LegacySSTableTest
         Assert.assertEquals(forward.size(), reverse.size());
     }
 
+    @Test
+    public void test14873() throws Exception
+    {
+        /*
+         * When reading 2.1 sstables in 3.0 in reverse order it's possible to wrongly return an empty result set if the
+         * partition being read has a static row, and the read is performed backwards.
+         */
+
+        /*
+         * Contents of the SSTable (column_index_size_in_kb: 1) below:
+         *
+         * insert into legacy_tables.legacy_ka_14873 (pkc, sc)     values (0, 0);
+         * insert into legacy_tables.legacy_ka_14873 (pkc, cc, rc) values (0, 5, '5555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555555');
+         * insert into legacy_tables.legacy_ka_14873 (pkc, cc, rc) values (0, 4, '4444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444444');
+         * insert into legacy_tables.legacy_ka_14873 (pkc, cc, rc) values (0, 3, '3333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333333');
+         * insert into legacy_tables.legacy_ka_14873 (pkc, cc, rc) values (0, 2, '2222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222222');
+         * insert into legacy_tables.legacy_ka_14873 (pkc, cc, rc) values (0, 1, '1111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111');
+         */
+
+        String ddl =
+            "CREATE TABLE legacy_tables.legacy_ka_14873 ("
+            + "pkc int, cc int, sc int static, rc text, PRIMARY KEY (pkc, cc)"
+            + ") WITH CLUSTERING ORDER BY (cc DESC) AND compaction = {'enabled' : 'false', 'class' : 'LeveledCompactionStrategy'};";
+        QueryProcessor.executeInternal(ddl);
+        loadLegacyTable("legacy_%s_14873%s", "ka", "");
+
+        UntypedResultSet forward =
+            QueryProcessor.executeOnceInternal(
+                String.format("SELECT * FROM legacy_tables.legacy_ka_14873 WHERE pkc = 0 AND cc > 0 ORDER BY cc DESC;"));
+
+        UntypedResultSet reverse =
+            QueryProcessor.executeOnceInternal(
+                String.format("SELECT * FROM legacy_tables.legacy_ka_14873 WHERE pkc = 0 AND cc > 0 ORDER BY cc ASC;"));
+
+        Assert.assertEquals(5, forward.size());
+        Assert.assertEquals(5, reverse.size());
+    }
+
     @Test
     public void testMultiBlockRangeTombstones() throws Exception
     {