From 548316593898c3b2169986d27eae7a6624bd873d Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Thu, 20 Feb 2025 17:54:29 -0600 Subject: [PATCH] Avoid limit on RFP fetch in the case of an unresolved static row patch by Caleb Rackliffe; reviewed by David Capwell and Zhao Yang for CASSANDRA-20323 --- CHANGES.txt | 1 + .../reads/ReplicaFilteringProtection.java | 15 +- .../test/ReplicaFilteringProtectionTest.java | 19 --- .../sai/ReplicaFilteringWithStaticsTest.java | 131 ++++++++++++++++++ .../test/sai/StrictFilteringTest.java | 20 --- 5 files changed, 142 insertions(+), 44 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java diff --git a/CHANGES.txt b/CHANGES.txt index 06c119f1ff..b01f18babc 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.4 + * Avoid limit on RFP fetch in the case of an unresolved static row (CASSANDRA-20323) * Include materialized views to the output of DESCRIBE TABLE statements (CASSANDRA-20365) * Heap and GC jvm flags improvements (CASSANDRA-20296) * Fix unparseable YAML in default cassandra.yaml when uncommented for downstream tooling (CASSANDRA-20359) diff --git a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java index 7c57245bb1..c66c2007d6 100644 --- a/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java +++ b/src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java @@ -230,7 +230,7 @@ public class ReplicaFilteringProtection> // Even if there are no completely missing rows, replicas may still be silent about individual // columns, so we need to check for divergence at the column level: - for (ColumnMetadata column : columns) + for (ColumnMetadata column : merged.isStatic() ? columns.statics : columns.regulars) { Arrays.fill(silentColumnAt, false); boolean allSilent = true; @@ -538,14 +538,19 @@ public class ReplicaFilteringProtection> Tracing.trace("Requesting {} rows in partition {} from {} for replica filtering protection", clusterings.size(), key, source); - // build the read command taking into account that we could be requesting only in the static row - DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE; - ClusteringIndexFilter filter = unresolvedStatic ? command.clusteringIndexFilter(key) : new ClusteringIndexNamesFilter(clusterings, command.isReversed()); + // If there is an unresolved static column, we must fetch the entire partition, as static column predicates + // may produce row matches across the entire partition. If there are only non-static rows to complete, we + // query the partition specifically for the corresponding cluterings by name. In either case, we do not + // provide a limit. (In the unresolved static case, we have no way of knowing how many stale rows we might + // read on a silent replica before finding a live one.) + ClusteringIndexFilter filter = unresolvedStatic ? command.clusteringIndexFilter(key) + : new ClusteringIndexNamesFilter(clusterings, command.isReversed()); + SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(), command.nowInSec(), command.columnFilter(), RowFilter.none(), - limits, + DataLimits.NONE, key, filter); diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java index 1cff326f41..fd8110cba7 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java @@ -27,7 +27,6 @@ import org.junit.Test; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.distributed.Cluster; -import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.IInvokableInstance; import org.apache.cassandra.distributed.api.SimpleQueryResult; import org.apache.cassandra.exceptions.OverloadedException; @@ -72,24 +71,6 @@ public class ReplicaFilteringProtectionTest extends TestBaseImpl cluster.close(); } - @Test - public void testMissingStaticRowWithNonStaticExpression() - { - cluster.schemaChange(withKeyspace("CREATE TABLE %s.single_predicate (pk0 int, ck0 int, ck1 int, s0 int static, s1 int static, v0 int, PRIMARY KEY (pk0, ck0, ck1)) " + - "WITH CLUSTERING ORDER BY (ck0 ASC, ck1 DESC) AND read_repair = 'NONE'")); - - cluster.get(1).executeInternal(withKeyspace("INSERT INTO %s.single_predicate (pk0, ck0, ck1, s0, s1, v0) " + - "VALUES (0, 1, 2, 3, 4, 5) USING TIMESTAMP 1")); - cluster.get(2).executeInternal(withKeyspace("UPDATE %s.single_predicate USING TIMESTAMP 2 SET s0 = 6, s1 = 7, v0 = 8 " + - "WHERE pk0 = 0 AND ck0 = 9 AND ck1 = 10")); - - // Node 2 will not produce a match for the static row. Make sure that replica filtering protection does not - // fetch the entire partition, which could let non-matching rows slip through combined with the fact that we - // don't post-filter at the coordinator with no regular column predicates in the query. - String select = withKeyspace("SELECT pk0, ck0, ck1, s0, s1 FROM %s.single_predicate WHERE ck1 = 2 ALLOW FILTERING"); - assertRows(cluster.coordinator(1).execute(select, ConsistencyLevel.ALL), row(0, 1, 2, 6, 7)); - } - @Test public void testMissedUpdatesBelowCachingWarnThreshold() { diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java new file mode 100644 index 0000000000..59cc1cc6a5 --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/ReplicaFilteringWithStaticsTest.java @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.distributed.test.sai; + +import java.io.IOException; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +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 static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL; +import static org.apache.cassandra.distributed.api.Feature.GOSSIP; +import static org.apache.cassandra.distributed.api.Feature.NETWORK; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; + +public class ReplicaFilteringWithStaticsTest extends TestBaseImpl +{ + private static Cluster CLUSTER; + + @BeforeClass + public static void setUpCluster() throws IOException + { + CLUSTER = init(Cluster.build(3).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start()); + } + + @Test + public void testStaticMatchWithPartitionDelete() + { + testStaticMatchWithPartitionDelete(false); + } + + @Test + public void testStaticMatchWithPartitionDeleteSAI() + { + testStaticMatchWithPartitionDelete(true); + } + + public void testStaticMatchWithPartitionDelete(boolean sai) + { + String table = "static_and_delete" + (sai ? "_sai" : ""); + CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s." + table + " (pk0 boolean, ck0 ascii, s1 tinyint static, v0 boolean, PRIMARY KEY (pk0, ck0)) " + + "WITH CLUSTERING ORDER BY (ck0 ASC) AND read_repair = 'NONE'")); + + if (sai) + { + CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s." + table + "(s1) USING 'sai'")); + SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE); + } + + CLUSTER.get(3).executeInternal(withKeyspace("UPDATE %s." + table + " USING TIMESTAMP 1 SET v0 = false WHERE pk0 = true AND ck0 = 'D'")); + CLUSTER.get(1).executeInternal(withKeyspace("DELETE FROM %s." + table + " USING TIMESTAMP 2 WHERE pk0 = true")); + + CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, s1, v0) VALUES (true, 'G', -114, true) USING TIMESTAMP 3")); + CLUSTER.get(3).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0) VALUES (true, 'F') USING TIMESTAMP 4")); + CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, s1, v0) VALUES (true, 'C', 17, true) USING TIMESTAMP 5")); + + // This update to the static column creates matches across all previously written live rows in the partition. + // When RFP sees the unresolved static row, it must read enough data from the silent replicas at nodes 1 and 3 + // to find all potential matches. With a page size of 1, reading only 1 row from node 3 will return the row at + // ck = 'D', as the partition delete never made it to node 3. This means we'll ignore the live result node 3 has + // at ck = 'F', because node 1 will produce a result at ck = 'G', and that determines the next paging cursor. + CLUSTER.get(2).executeInternal(withKeyspace("UPDATE %s." + table + " USING TIMESTAMP 6 SET s1 = 1, v0 = false WHERE pk0 = true AND ck0 = 'A'")); + + String select = withKeyspace("SELECT ck0 FROM %s." + table + " WHERE s1 = 1" + (sai ? "" : " ALLOW FILTERING" )); + assertRows(CLUSTER.coordinator(1).executeWithPaging(select, ALL, 1), row("A"), row("C"), row("F"), row("G")); + } + + @Test + public void testMissingStaticRowWithNonStaticExpression() + { + testMissingStaticRowWithNonStaticExpression(false); + } + + @Test + public void testMissingStaticRowWithNonStaticExpressionSAI() + { + testMissingStaticRowWithNonStaticExpression(true); + } + + public void testMissingStaticRowWithNonStaticExpression(boolean sai) + { + String table = "single_predicate" + (sai ? "_sai" : ""); + CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s." + table + " (pk0 int, ck0 int, ck1 int, s0 int static, s1 int static, v0 int, PRIMARY KEY (pk0, ck0, ck1)) " + + "WITH CLUSTERING ORDER BY (ck0 ASC, ck1 DESC) AND read_repair = 'NONE'")); + + if (sai) + { + CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s." + table + "(ck1) USING 'sai'")); + SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE); + } + + CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s." + table + " (pk0, ck0, ck1, s0, s1, v0) " + + "VALUES (0, 1, 2, 3, 4, 5) USING TIMESTAMP 1")); + CLUSTER.get(2).executeInternal(withKeyspace("UPDATE %s." + table + " USING TIMESTAMP 2 SET s0 = 6, s1 = 7, v0 = 8 " + + "WHERE pk0 = 0 AND ck0 = 9 AND ck1 = 10")); + + // Node 2 will not produce a match for the static row. Make sure that replica filtering protection does not + // fetch the entire partition, which could let non-matching rows slip through combined with the fact that we + // don't post-filter at the coordinator with no regular column predicates in the query. + String select = withKeyspace("SELECT pk0, ck0, ck1, s0, s1 FROM %s." + table + " WHERE ck1 = 2" + (sai ? "" : " ALLOW FILTERING")); + assertRows(CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL), row(0, 1, 2, 6, 7)); + } + + @AfterClass + public static void shutDownCluster() + { + if (CLUSTER != null) + CLUSTER.close(); + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java b/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java index 5ef92bb9ef..6ec80fd0ae 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/sai/StrictFilteringTest.java @@ -53,26 +53,6 @@ public class StrictFilteringTest extends TestBaseImpl CLUSTER = init(Cluster.build(2).withConfig(config -> config.set("hinted_handoff_enabled", false).with(GOSSIP).with(NETWORK)).start()); } - @Test - public void testMissingStaticRowWithNonStaticExpression() - { - CLUSTER.schemaChange(withKeyspace("CREATE TABLE %s.single_predicate (pk0 int, ck0 int, ck1 int, s0 int static, s1 int static, v0 int, PRIMARY KEY (pk0, ck0, ck1)) " + - "WITH CLUSTERING ORDER BY (ck0 ASC, ck1 DESC) AND read_repair = 'NONE'")); - CLUSTER.schemaChange(withKeyspace("CREATE INDEX ON %s.single_predicate(ck1) USING 'sai'")); - SAIUtil.waitForIndexQueryable(CLUSTER, KEYSPACE); - - CLUSTER.get(1).executeInternal(withKeyspace("INSERT INTO %s.single_predicate (pk0, ck0, ck1, s0, s1, v0) " + - "VALUES (0, 1, 2, 3, 4, 5) USING TIMESTAMP 1")); - CLUSTER.get(2).executeInternal(withKeyspace("UPDATE %s.single_predicate USING TIMESTAMP 2 SET s0 = 6, s1 = 7, v0 = 8 " + - "WHERE pk0 = 0 AND ck0 = 9 AND ck1 = 10")); - - // Node 2 will not produce a match for the static row. Make sure that replica filtering protection does not - // fetch the entire partition, which could let non-matching rows slip through combined with the fact that we - // don't post-filter at the coordinator with no regular column predicates in the query. - String select = withKeyspace("SELECT pk0, ck0, ck1, s0, s1 FROM %s.single_predicate WHERE ck1 = 2 ALLOW FILTERING"); - assertRows(CLUSTER.coordinator(1).execute(select, ConsistencyLevel.ALL), row(0, 1, 2, 6, 7)); - } - @Test public void shouldDegradeToUnionOnSingleStatic() {