diff --git a/CHANGES.txt b/CHANGES.txt index b8157abc0c..43989f39d0 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -17,6 +17,7 @@ * 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) 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 2e8b7d8f71..c2c8efcb52 100644 --- a/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java +++ b/src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java @@ -503,7 +503,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. @@ -523,11 +526,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()) @@ -540,10 +543,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... @@ -571,9 +579,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(), @@ -766,7 +774,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) { @@ -794,7 +803,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 61f221c152..6e08bc5ba5 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java @@ -162,6 +162,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() {