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 trunk 

* cassandra-5.0:
  Avoid purging deletions in RowFilter when reconciliation is required
This commit is contained in:
Caleb Rackliffe 2025-04-10 22:33:21 -05:00
parent 0e408e6175 725e4ba3eb
commit 30f1429e6d
5 changed files with 51 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -171,6 +171,7 @@
* Add the ability to disable bulk loading of SSTables (CASSANDRA-18781)
* Clean up obsolete functions and simplify cql_version handling in cqlsh (CASSANDRA-18787)
Merged from 5.0:
* Avoid purging deletions in RowFilter when reconciliation is required (CASSANDRA-20541)
* Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338)
* Upgrade com.datastax.cassandra:cassandra-driver-core:3.11.5 to org.apache.cassandra:cassandra-driver-core:3.12.1 (CASSANDRA-17231)
* Update netty to 4.1.119.Final and netty-tcnative to 2.0.70.Final (CASSANDRA-20314)

5
src/java/org/apache/cassandra/db/filter/RowFilter.java
View File

@ -256,7 +256,10 @@ public class RowFilter implements Iterable<RowFilter.Expression>
@Override
public Row applyToRow(Row row)
{
Row purged = row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
// If we purge deletions when reconciliation is required, we hide information replica filtering
// protection would require to filter rows that are no longer matches are the coordinator.
Row purged = needsReconciliation() ? row : row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
if (purged == null)
return null;

6
test/distributed/org/apache/cassandra/distributed/test/TableLevelIncrementalBackupsTest.java
View File

@ -141,12 +141,6 @@ public class TableLevelIncrementalBackupsTest extends TestBaseImpl
cluster.get(i).flush(keyspace);
}
private void disableCompaction(Cluster cluster, String keyspace, String table)
{
for (int i = 1; i < cluster.size() + 1; i++)
cluster.get(i).nodetool("disableautocompaction", keyspace, table);
}
private static void assertBackupSSTablesCount(Cluster cluster, int expectedTablesCount, boolean enable, String ks, String... tableNames)
{
for (int i = 1; i < cluster.size() + 1; i++)

6
test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java
View File

@ -240,4 +240,10 @@ public class TestBaseImpl extends DistributedTestBase
// in real live repair is needed in this case, but in the test case it doesn't matter if the tables loose
// anything, so ignoring repair to speed up the tests.
}
protected static void disableCompaction(Cluster cluster, String keyspace, String table)
{
for (int i = 1; i < cluster.size() + 1; i++)
cluster.get(i).nodetool("disableautocompaction", keyspace, table);
}
}

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

@ -27,6 +27,7 @@ import org.junit.Test;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.test.TestBaseImpl;
import org.apache.cassandra.distributed.test.cql3.MultiNodeTableWalkWithoutReadRepairTest;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
@ -44,6 +45,45 @@ public class ReplicaFilteringWithStaticsTest extends TestBaseImpl
CLUSTER = init(Cluster.build(3).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start());
}
@Test
public void testRowFilterDeletePurging()
{
testRowFilterDeletePurging(false);
}
@Test
public void testRowFilterDeletePurgingSAI()
{
testRowFilterDeletePurging(true);
}
/**
* Originally discovered by {@link MultiNodeTableWalkWithoutReadRepairTest} with seed 6640281155419111674
*/
public void testRowFilterDeletePurging(boolean sai)
{
String table = "row_filtering_delete_purging" + (sai ? "_sai" : "");
CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s." + table + " (pk0 double, ck0 boolean, s0 ascii static, v0 ascii, " +
"PRIMARY KEY (pk0, ck0)) WITH CLUSTERING ORDER BY (ck0 DESC) AND read_repair = 'NONE'"));
disableCompaction(CLUSTER, KEYSPACE, table);
if (sai)
{
CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s." + table + "(s0) USING 'sai'"));
SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE);
}
CLUSTER.get(3).executeInternal(withKeyspace("UPDATE %s." + table + " USING TIMESTAMP 1 SET s0='foo', v0='c' WHERE pk0 = 2.9 AND ck0 IN (false, true)"));
// This delete must be resolved by RFP to eliminate the row with ck0 = true from node 3:
CLUSTER.get(1).executeInternal(withKeyspace("DELETE FROM %s." + table + " USING TIMESTAMP 2 WHERE pk0 = 2.9 AND ck0 = true"));
CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, s0, v0) VALUES (2.9, false, 'bar', 'xyz') USING TIMESTAMP 3"));
String select = withKeyspace("SELECT ck0 FROM %s." + table + " WHERE s0 = 'bar' ALLOW FILTERING");
assertRows(CLUSTER.coordinator(1).executeWithPaging(select, ALL, 100), row(false));
}
@Test
public void testStaticMatchWithPartitionDelete()
{