Fix invalid cell value skipping when reading from disk

patch by Jacek Lewandowski; reviewed by Sylvain Lebresne for CASSANDRA-16223

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0.23:
+ * Fix invalid cell value skipping when reading from disk (CASSANDRA-16223)
  * Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146)
  * Fix OOM when terminating repair session (CASSANDRA-15902)
  * Avoid marking shutting down nodes as up after receiving gossip shutdown message (CASSANDRA-16094)

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

@@ -514,12 +514,12 @@ public class SelectStatement implements CQLStatement
         // Note that we use the total limit for every key, which is potentially inefficient.
         // However, IN + LIMIT is not a very sensible choice.
         List<SinglePartitionReadCommand> commands = new ArrayList<>(keys.size());
+        ColumnFilter columnFilter = createColumnFilter(options);
         for (ByteBuffer key : keys)
         {
             QueryProcessor.validateKey(key);
             DecoratedKey dk = cfm.decorateKey(ByteBufferUtil.clone(key));
-            ColumnFilter cf = (cfm.isSuper() && cfm.isDense()) ? SuperColumnCompatibility.getColumnFilter(cfm, options, restrictions.getSuperColumnRestrictions()) : queriedColumns;
+            commands.add(SinglePartitionReadCommand.create(cfm, nowInSec, columnFilter, rowFilter, limit, dk, filter));
-            commands.add(SinglePartitionReadCommand.create(cfm, nowInSec, cf, rowFilter, limit, dk, filter));
         }
 
         return new SinglePartitionReadCommand.Group(commands, limit);
@@ -581,7 +581,7 @@ public class SelectStatement implements CQLStatement
             return ReadQuery.EMPTY;
 
         PartitionRangeReadCommand command =
-            PartitionRangeReadCommand.create(false, cfm, nowInSec, queriedColumns, rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter));
+            PartitionRangeReadCommand.create(false, cfm, nowInSec, createColumnFilter(options), rowFilter, limit, new DataRange(keyBounds, clusteringIndexFilter));
 
         // If there's a secondary index that the command can use, have it validate the request parameters.
         command.maybeValidateIndex();
@@ -870,6 +870,13 @@ public class SelectStatement implements CQLStatement
         Collections.sort(cqlRows.rows, orderingComparator);
     }
 
+    private ColumnFilter createColumnFilter(QueryOptions options)
+    {
+        return (cfm.isSuper() && cfm.isDense())
+               ? SuperColumnCompatibility.getColumnFilter(cfm, options, restrictions.getSuperColumnRestrictions())
+               : queriedColumns;
+    }
+
     public static class RawStatement extends CFStatement
     {
         public final Parameters parameters;

test/unit/org/apache/cassandra/cql3/validation/ThriftIntegrationTest.java

@@ -936,7 +936,7 @@ public class ThriftIntegrationTest extends ThriftCQLTester
 
     private void flushAll()
     {
-        for (String cfName : new String[]{ currentTable(), currentSparseTable(), currentCounterTable() })
+        for (String cfName : new String[]{ currentTable(), currentSparseTable(), currentCounterTable(), currentDenseTable() })
-            Keyspace.open(KEYSPACE).getColumnFamilyStore(cfName);
+            Keyspace.open(KEYSPACE).getColumnFamilyStore(cfName).forceBlockingFlush();
     }
 }