natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-5.0' into cassandra-6.0 

* cassandra-5.0:
  Ensure SAI sends range tombstones to the coordinator for queries on static columns
This commit is contained in:
Caleb Rackliffe 2026-04-27 13:03:35 -05:00
parent ae418d2a1c 8093c3c9ce
commit 98f0a34f07
3 changed files with 59 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

25
src/java/org/apache/cassandra/index/sai/plan/StorageAttachedIndexSearcher.java
View File

@ -503,7 +503,10 @@ public class StorageAttachedIndexSearcher implements Index.Searcher
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 class StorageAttachedIndexSearcher implements Index.Searcher
}
}
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 @@ 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<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 class StorageAttachedIndexSearcher implements Index.Searcher
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 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();
}

41
test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java
View File

@ -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()
{