From 8093c3c9ce73cc736fedb489577f0deb8f9e49bd Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Thu, 23 Apr 2026 18:17:53 -0500 Subject: [PATCH] Ensure SAI sends range tombstones to the coordinator for queries on static columns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit patch by Caleb Rackliffe; reviewed by Andres de la Peña and David Capwell for CASSANDRA-21332 --- CHANGES.txt | 1 + .../plan/StorageAttachedIndexSearcher.java | 25 +++++++---- .../sai/ReplicaFilteringWithStaticsTest.java | 41 +++++++++++++++++++ 3 files changed, 59 insertions(+), 8 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index bca3b56a37..e197805c11 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.9 + * 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) Merged from 4.0: diff --git a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java index 7d16f33990..d937100c54 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java +++ b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java @@ -502,7 +502,10 @@ public class StorageAttachedIndexSearcher implements Index.Searcher queryContext.partitionsRead++; queryContext.checkpoint(); - List 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 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. @@ -522,11 +525,11 @@ public class StorageAttachedIndexSearcher implements Index.Searcher } } - private static List filterPartition(UnfilteredRowIterator partition, FilterTree tree, QueryContext context) + private static List filterPartition(UnfilteredRowIterator partition, FilterTree tree, QueryContext context, boolean matchTombstones) { Row staticRow = partition.staticRow(); DecoratedKey partitionKey = partition.partitionKey(); - List matches = new ArrayList<>(); + List matches = new ArrayList<>(); boolean hasMatch = false; while (partition.hasNext()) @@ -539,10 +542,15 @@ public class StorageAttachedIndexSearcher implements Index.Searcher 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... @@ -570,9 +578,9 @@ public class StorageAttachedIndexSearcher implements Index.Searcher private static class SinglePartitionIterator extends AbstractUnfilteredRowIterator { - private final Iterator rows; + private final Iterator rows; - public SinglePartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator rows) + public SinglePartitionIterator(UnfilteredRowIterator partition, Row staticRow, Iterator rows) { super(partition.metadata(), partition.partitionKey(), @@ -765,7 +773,8 @@ public class StorageAttachedIndexSearcher implements Index.Searcher queryContext.partitionsRead++; queryContext.checkpoint(); - List clusters = filterPartition(partition, filterTree, queryContext); + // Scored queries do not involve RFP, and therefore can ignore range tombstones. + List clusters = filterPartition(partition, filterTree, queryContext, false); if (clusters == null) { @@ -793,7 +802,7 @@ public class StorageAttachedIndexSearcher implements Index.Searcher 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(); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java index 1892157712..6351193f21 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java @@ -158,6 +158,47 @@ public class ReplicaFilteringWithStaticsTest extends TestBaseImpl 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() {