IndexOutOfBoundsException when accessing partition where the column was deleted

src/java/org/apache/cassandra/db/filter/RowFilter.java

@@ -224,7 +224,7 @@ protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
 
                 // Short-circuit all partitions that won't match based on static and partition keys
                 for (Expression e : partitionLevelExpressions)
-                    if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow()))
+                    if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
                     {
                         partition.close();
                         return null;
@@ -251,7 +251,7 @@ public Row applyToRow(Row row)
                     return null;
 
                 for (Expression e : rowLevelExpressions)
-                    if (!e.isSatisfiedBy(metadata, pk, purged))
+                    if (!e.isSatisfiedBy(metadata, pk, purged, nowInSec))
                         return null;
 
                 return row;
@@ -303,7 +303,7 @@ public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey,
 
         for (Expression e : expressions)
         {
-            if (!e.isSatisfiedBy(metadata, partitionKey, purged))
+            if (!e.isSatisfiedBy(metadata, partitionKey, purged, nowInSec))
                 return false;
         }
         return true;
@@ -515,9 +515,9 @@ public void validateForIndexing()
          * (i.e. it should come from a RowIterator).
          * @return whether the row is satisfied by this expression.
          */
-        public abstract boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row);
+        public abstract boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec);
 
-        protected ByteBuffer getValue(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        protected ByteBuffer getValue(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             switch (column.kind)
             {
@@ -529,7 +529,7 @@ protected ByteBuffer getValue(TableMetadata metadata, DecoratedKey partitionKey,
                     return row.clustering().bufferAt(column.position());
                 default:
                     Cell<?> cell = row.getCell(column);
-                    return cell == null ? null : cell.buffer();
+                    return Cell.getValidCellBuffer(cell, nowInSec);
             }
         }
 
@@ -690,7 +690,8 @@ public static class SimpleExpression extends Expression
             super(column, operator, value);
         }
 
-        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        @Override
+        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             // We support null conditions for LWT (in ColumnCondition) but not for RowFilter.
             // TODO: we should try to merge both code someday.
@@ -704,7 +705,7 @@ public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey,
                 // representation. See CASSANDRA-11629
                 if (column.type.isCounter())
                 {
-                    ByteBuffer foundValue = getValue(metadata, partitionKey, row);
+                    ByteBuffer foundValue = getValue(metadata, partitionKey, row, nowInSec);
                     if (foundValue == null)
                         return false;
 
@@ -714,7 +715,7 @@ public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey,
                 else
                 {
                     // Note that CQL expression are always of the form 'x < 4', i.e. the tested value is on the left.
-                    ByteBuffer foundValue = getValue(metadata, partitionKey, row);
+                    ByteBuffer foundValue = getValue(metadata, partitionKey, row, nowInSec);
                     return foundValue != null && operator.isSatisfiedBy(column.type, foundValue, value);
                 }
             }
@@ -729,7 +730,7 @@ else if (operator.appliesToCollectionElements() || operator.appliesToMapKeys())
                 }
                 else
                 {
-                    ByteBuffer foundValue = getValue(metadata, partitionKey, row);
+                    ByteBuffer foundValue = getValue(metadata, partitionKey, row, nowInSec);
                     return foundValue != null && operator.isSatisfiedBy(column.type, foundValue, value);
                 }
             }
@@ -814,7 +815,7 @@ public ByteBuffer getIndexValue()
             return CompositeType.build(ByteBufferAccessor.instance, key, value);
         }
 
-        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             assert key != null;
             // We support null conditions for LWT (in ColumnCondition) but not for RowFilter.
@@ -832,7 +833,7 @@ public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey,
             }
             else
             {
-                ByteBuffer serializedMap = getValue(metadata, partitionKey, row);
+                ByteBuffer serializedMap = getValue(metadata, partitionKey, row, nowInSec);
                 if (serializedMap == null)
                     return false;
 
@@ -933,7 +934,7 @@ protected Kind kind()
         }
 
         // Filtering by custom expressions isn't supported yet, so just accept any row
-        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             return true;
         }

src/java/org/apache/cassandra/db/rows/Cell.java

@@ -238,6 +238,26 @@ public static long decodeLocalDeletionTime(long localDeletionTime, int ttl, Dese
                                            // timestamp on expiry.
     }
 
+    /**
+     * Validates a cell's liveliness, tombstone status, and buffer contents.
+     *
+     * @param cell         The cell to validate.
+     * @param nowInSeconds The current time in seconds.
+     * @return A ByteBuffer (including valid empty buffers) if valid, or null otherwise.
+     */
+    public static ByteBuffer getValidCellBuffer(Cell<?> cell, long nowInSeconds) {
+        if (cell == null || cell.isTombstone()) {
+            return null;
+        }
+
+        if (!cell.isLive(nowInSeconds)) {
+            return null;
+        }
+
+        // Allow valid empty buffers
+        return cell.buffer();
+    }
+
     /**
      * The serialization format for cell is:
      *     [ flags ][ timestamp ][ deletion time ][    ttl    ][ path size ][ path ][ value size ][ value ]

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

@@ -2554,6 +2554,35 @@ public void filteringOnStaticColumnTest() throws Throwable
         });
     }
 
+    @Test
+    public void filteringOnDeletedStaticColumnValue() throws Throwable {
+        // Create table with int-only columns
+        createTable("CREATE TABLE %s (pk0 int, pk1 int, ck0 int, ck1 int, s0 tinyint static, v0 int, v1 int, PRIMARY KEY ((pk0, pk1), ck0, ck1))");
+
+        // Insert rows
+        execute("INSERT INTO %s (pk0, pk1, s0, ck0, ck1, v0, v1) VALUES (?, ?, ?, ?, ?, ?, ?)", 1000, 2000, (byte) 126, 100, 1, 20, 30);
+        execute("INSERT INTO %s (pk0, pk1, s0, ck0, ck1, v0, v1) VALUES (?, ?, ?, ?, ?, ?, ?)", 1000, 2000, (byte) 125, 200, 2, 40, 50);
+        execute("INSERT INTO %s (pk0, pk1, s0, ck0, ck1, v0, v1) VALUES (?, ?, ?, ?, ?, ?, ?)", 1000, 3000, (byte) 122, 300, 3, 60, 70);
+        execute("DELETE s0,v0,v1 FROM %s WHERE pk0=1000 AND pk1=2000 and ck0=100 and ck1=1");
+
+        beforeAndAfterFlush(() -> {
+            // Verify the columns are deleted
+            assertRows(execute("SELECT pk0, pk1, s0, ck0, ck1, v0, v1 FROM %s WHERE s0=? ALLOW FILTERING", (byte) 122),
+                       row(1000, 3000, (byte) 122, 300, 3, 60, 70));
+        });
+
+        execute("DELETE v0 FROM %s WHERE pk0=1000 AND pk1=3000 AND ck0=300 AND ck1=3");
+
+        beforeAndAfterFlush(() -> {
+            assertRows(execute("SELECT pk0, pk1, s0, ck0, ck1, v0, v1 FROM %s WHERE s0=? ALLOW FILTERING", (byte) 122),
+                       row(1000, 3000, (byte) 122, 300, 3, null, 70));
+
+            assertRows(execute("SELECT pk0, pk1, s0, ck0, ck1, v0, v1 FROM %s WHERE pk0=1000 AND pk1=3000 AND ck0=300 AND ck1=3"),
+                       row(1000, 3000, (byte) 122, 300, 3, null, 70));
+        });
+
+    }
+
     @Test
     public void containsFilteringOnNonClusteringColumn() throws Throwable {
         createTable("CREATE TABLE %s (a int, b int, c int, d list<int>, PRIMARY KEY (a, b, c))");

test/unit/org/apache/cassandra/index/sai/plan/OperationTest.java

@@ -515,7 +515,7 @@ public Kind kind()
         }
 
         @Override
-        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             throw new UnsupportedOperationException();
         }

test/unit/org/apache/cassandra/index/sasi/plan/OperationTest.java

@@ -651,7 +651,7 @@ protected Kind kind()
         }
 
         @Override
-        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row)
+        public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
         {
             throw new UnsupportedOperationException();
         }