IndexOutOfBoundsException when accessing partition where the column was deleted

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

@@ -73,6 +73,7 @@
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkBindValueSet;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
+import static org.apache.cassandra.utils.FBUtilities.nowInSeconds;
 
 /**
  * A filter on which rows a given query should include or exclude.
@@ -529,7 +530,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, nowInSeconds());
             }
         }
 

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))");