Merge branch 'cassandra-6.0' into trunk

* cassandra-6.0:
  Ensure SAI sends range tombstones to the coordinator for queries on static columns

Author: maedhroz

CHANGES.txt

@@ -20,6 +20,7 @@ Merged from 6.0:
  * Fix a removed TTLed row re-appearance in a materialized view after a cursor compaction (CASSANDRA-21152)
  * Rework ZSTD dictionary compression logic to create a trainer per training (CASSANDRA-21209)
 Merged from 5.0:
+ * Ensure SAI sends range tombstones to the coordinator for queries on static columns (CASSANDRA-21332)
 Merged from 4.1:
  * Add Paxos v2 option and informatin in cassandra.yaml (CASSANDRA-21316)
  * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290)

src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java

@@ -503,7 +503,10 @@ private UnfilteredRowIterator queryStorageAndFilter(List<PrimaryKey> keys)
                 queryContext.partitionsRead++;
                 queryContext.checkpoint();
 
-                List<Row> filtered = filterPartition(partition, filterTree, queryContext);
+                // If there is an unresolved static expression during RFP, completion reads to silent replicas will 
+                // fetch entire partitions. If we don't return range tombstones in this initial read, there may not be 
+                // enough information at the coordinator for RFP to shadow logically deleted rows from those replicas.
+                List<Unfiltered> filtered = filterPartition(partition, filterTree, queryContext, command.rowFilter().hasStaticExpression());
 
                 // Note that we record the duration of the read after post-filtering, which actually
                 // materializes the rows from disk.
@@ -523,11 +526,11 @@ public void close()
         }
     }
 
-    private static List<Row> filterPartition(UnfilteredRowIterator partition, FilterTree tree, QueryContext context)
+    private static List<Unfiltered> filterPartition(UnfilteredRowIterator partition, FilterTree tree, QueryContext context, boolean matchTombstones)
     {
         Row staticRow = partition.staticRow();
         DecoratedKey partitionKey = partition.partitionKey();
-        List<Row> matches = new ArrayList<>();
+        List<Unfiltered> matches = new ArrayList<>();
         boolean hasMatch = false;
 
         while (partition.hasNext())
@@ -540,10 +543,15 @@ private static List<Row> filterPartition(UnfilteredRowIterator partition, Filter
 
                 if (tree.isSatisfiedBy(partitionKey, (Row) unfiltered, staticRow))
                 {
-                    matches.add((Row) unfiltered);
+                    matches.add(unfiltered);
                     hasMatch = true;
                 }
             }
+            else if (matchTombstones && unfiltered.isRangeTombstoneMarker())
+            {
+                // Note that range tombstones do not constitute matches, and will be discarded if no actual rows match.
+                matches.add(unfiltered);
+            }
         }
 
         // We may not have any non-static row data to filter...
@@ -571,9 +579,9 @@ private static List<Row> filterPartition(UnfilteredRowIterator partition, Filter
 
     private static class SinglePartitionIterator extends AbstractUnfilteredRowIterator
     {
-        private final Iterator<Row> rows;
+        private final Iterator<Unfiltered> rows;
 
-        public SinglePartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator<Row> rows)
+        public SinglePartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator<Unfiltered> rows)
         {
             super(partition.metadata(),
                   partition.partitionKey(),
@@ -766,7 +774,8 @@ public UnfilteredRowIterator readAndValidatePartition(PrimaryKey pk, List<Primar
                 queryContext.partitionsRead++;
                 queryContext.checkpoint();
 
-                List<Row> clusters = filterPartition(partition, filterTree, queryContext);
+                // Scored queries do not involve RFP, and therefore can ignore range tombstones.
+                List<Unfiltered> clusters = filterPartition(partition, filterTree, queryContext, false);
 
                 if (clusters == null)
                 {
@@ -794,7 +803,7 @@ public UnfilteredRowIterator readAndValidatePartition(PrimaryKey pk, List<Primar
                         processedKeys.add(pk);
                         return null;
                     }
-                    representativeRow = clusters.get(0);
+                    representativeRow = (Row) clusters.get(0);
                     assert clusters.size() == 1 : "Expect 1 result row, but got: " + clusters.size();
                 }
 

test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java

@@ -162,6 +162,47 @@ public void testMissingStaticRowWithNonStaticExpression(boolean sai)
         assertRows(CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL), row(0, 1, 2, 6, 7));
     }
 
+    @Test
+    public void testRangeTombstoneWithStaticSAI()
+    {
+        testRangeTombstoneWithStatic(true);
+    }
+
+    @Test
+    public void testRangeTombstoneWithStatic()
+    {
+        testRangeTombstoneWithStatic(false);
+    }
+
+    private void testRangeTombstoneWithStatic(boolean sai)
+    {
+        String table = "range_tombstone_with_static" + (sai ? "_sai" : "");
+        CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s." + table + " (pk0 int, ck0 boolean, ck1 double, s1 int static, v0 boolean," + " PRIMARY KEY (pk0, ck0, ck1)) WITH read_repair = 'NONE'"));
+        disableCompaction(CLUSTER, KEYSPACE, table);
+
+        if (sai)
+        {
+            CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s." + table + "(s1) USING 'sai'"));
+            SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
+        }
+
+        // Node 3 gets a row at ck0=false with an old s1 value. This will be locally live but globally dead once the range tombstone on node 2 is considered.
+        CLUSTER.get(3).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, ck1, s1, v0) VALUES (1, false, 1.0, 99, false) USING TIMESTAMP 1"));
+
+        // Node 2 gets a range tombstone covering all rows (ck0 is boolean, <= true covers everything). Nodes 1 and 3 never receive this.
+        CLUSTER.get(2).executeInternal(withKeyspace("DELETE FROM %s." + table + " USING TIMESTAMP 2 WHERE pk0 = 1 AND ck0 <= true"));
+
+        // Node 2 also gets a new surviving row at ck0=true with a new s1 value. This is the only row that should be visible after reconciliation.
+        CLUSTER.get(2).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, ck1, s1, v0) VALUES (1, true, 5.0, 42, true) USING TIMESTAMP 3"));
+
+        // The query on s1=42 should return only the single surviving row, and the first pass input to RFP get exactly that.
+        // However, since there is an unresolved static, RFP completion reads for nodes 1 and 3 must read the entire partition.
+        // This returns the rows written before the deletion, which never arrived there. If the range tombstone from
+        // node 2 is not included in the first pass query, it will not be available to shadow the logically deleted rows.
+        String select = withKeyspace("SELECT ck0, ck1 FROM %s." + table + " WHERE s1 = 42" + (sai ? "" : " ALLOW FILTERING"));
+        assertRows(CLUSTER.coordinator(1).executeWithPaging(select, ALL, 1), row(true, 5.0));
+    }
+
     @AfterClass
     public static void shutDownCluster()
     {