This is an automated email from the ASF dual-hosted git repository. aweisberg pushed a commit to branch cep-45-mutation-tracking in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 763e1fedf43b2c6a225ae78d2620db625eee1b87 Author: Ariel Weisberg <[email protected]> AuthorDate: Wed Aug 12 11:53:07 2026 -0400 Refuse multi-partition serial reads before forwarding An IN over partition keys builds one query per partition, but a forwarded consensus read carries one command and returns one partition, so checkAndForwardConsensusReadIfNeeded answered with queries.get(0) alone. Both paxos implementations refused such a group below the forwarding hook, and accord refused nothing, so the outcome depended on the protocol and on whether the coordinator was a replica. Refuse it in readWithConsensusInternal, above forwarding and the protocol choice, and delete the two copies below. Conditional updates and batches already reject IN and multiple partitions. --- .../org/apache/cassandra/service/StorageProxy.java | 7 ++- .../service/paxos/CasForwardResponse.java | 10 +++- .../cassandra/service/paxos/CasForwarding.java | 20 +++---- .../service/paxos/ConsensusReadForwardRequest.java | 3 +- .../org/apache/cassandra/service/paxos/Paxos.java | 2 - .../MutationTrackingCasForwardingTest.java | 68 +++++++++++++++++----- 6 files changed, 77 insertions(+), 33 deletions(-) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 9d2f51f2ec..70db523926 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -2725,6 +2725,10 @@ public class StorageProxy implements StorageProxyMBean private static PartitionIterator readWithConsensusInternal(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime, boolean alreadyForwarded) throws InvalidRequestException, UnavailableException, ReadFailureException, ReadTimeoutException { + // Refused here, above forwarding and the protocol choice, so every coordinator rejects identically + if (group.queries.size() > 1) + throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one partition at a time"); + // Check if this consensus read needs to be forwarded to a replica coordinator for tracked keyspaces CasForwarding.Forwarded<PartitionIterator> forwarded = CasForwarding.checkAndForwardConsensusReadIfNeeded(group, consistencyLevel, @@ -2861,9 +2865,6 @@ public class StorageProxy implements StorageProxyMBean throws InvalidRequestException, UnavailableException, ReadFailureException, ReadTimeoutException { long start = nanoTime(); - if (group.queries.size() > 1) - throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one partition at a time"); - SinglePartitionReadCommand command = group.queries.get(0); TableMetadata metadata = command.metadata(); DecoratedKey key = command.partitionKey(); diff --git a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java index bd99628ecc..a996fa45a9 100644 --- a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java +++ b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java @@ -42,6 +42,7 @@ import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.CollectionSerializers; import org.apache.cassandra.utils.StringSerializer; +import static com.google.common.base.Preconditions.checkState; import static org.apache.cassandra.db.SerializationHeader.StableHeaderSerializer.STABLE; import static org.apache.cassandra.db.rows.DeserializationHelper.Flag.FROM_REMOTE; @@ -97,7 +98,14 @@ public class CasForwardResponse try (PartitionIterator toClose = partitions) { - return toClose.hasNext() ? materialize(toClose.next()) : null; + if (!toClose.hasNext()) + return null; + + FilteredPartition materialized = materialize(toClose.next()); + // Serial reads are single partition, enforced in StorageProxy.readWithConsensusInternal. + // Asked only after the partition above is drained, per the note in PartitionIterators. + checkState(!toClose.hasNext(), "Forwarded read response cannot carry more than one partition"); + return materialized; } } diff --git a/src/java/org/apache/cassandra/service/paxos/CasForwarding.java b/src/java/org/apache/cassandra/service/paxos/CasForwarding.java index cc388e80f3..60bc74e778 100644 --- a/src/java/org/apache/cassandra/service/paxos/CasForwarding.java +++ b/src/java/org/apache/cassandra/service/paxos/CasForwarding.java @@ -61,6 +61,7 @@ import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.utils.FBUtilities; +import static com.google.common.base.Preconditions.checkState; import static org.apache.cassandra.net.Verb.CONSENSUS_READ_FORWARD_REQ; public class CasForwarding @@ -226,19 +227,18 @@ public class CasForwarding boolean alreadyForwarded) throws UnavailableException, ReadFailureException, ReadTimeoutException { - if (group.queries.isEmpty()) - return null; + // StorageProxy.readWithConsensusInternal already refused groups of more than one query + checkState(group.queries.size() == 1, "Consensus read forwarding requires a single partition, got %s", group.queries.size()); - // Use the first command to determine keyspace and key for replica planning - SinglePartitionReadCommand firstCommand = group.queries.get(0); - String keyspaceName = firstCommand.metadata().keyspace; + SinglePartitionReadCommand command = group.queries.get(0); + String keyspaceName = command.metadata().keyspace; Keyspace keyspace = Keyspace.openIfExists(keyspaceName); if (keyspace == null) throw new KeyspaceNotDefinedException("Keyspace " + keyspaceName + " does not exist"); ClusterMetadata cm = ClusterMetadata.current(); - if (!MigrationRouter.shouldUseTracked(cm, firstCommand)) + if (!MigrationRouter.shouldUseTracked(cm, command)) return null; // Not tracked, no forwarding needed // Property to disable top-level forwarding for testing @@ -246,7 +246,7 @@ public class CasForwarding return null; // Check if current coordinator is not a replica - Token tk = firstCommand.partitionKey().getToken(); + Token tk = command.partitionKey().getToken(); EndpointsForToken allReplicas = ReplicaLayout.forTokenWriteLiveAndDown(cm, keyspace, tk) .all(); EndpointsForToken liveReplicas = allReplicas.filter(FailureDetector.isReplicaAlive); @@ -261,7 +261,7 @@ public class CasForwarding if (alreadyForwarded) { logger.error("Received forwarded consensus read for keyspace {} key {} but local node {} is not a replica. Replicas are: {}", - keyspaceName, firstCommand.partitionKey(), localEndpoint, allReplicas); + keyspaceName, command.partitionKey(), localEndpoint, allReplicas); Tracing.trace("ERROR: Received forwarded consensus read but local node is not a replica"); throw new RuntimeException("Forwarded consensus read received by non-replica node " + localEndpoint); } @@ -274,8 +274,8 @@ public class CasForwarding EndpointsForToken sortedReplicas = DatabaseDescriptor.getNodeProximity().sortedByProximity(localEndpoint, liveReplicas); InetAddressAndPort replicaCoordinator = sortedReplicas.get(0).endpoint(); - // Create forward request - consensus reads only have a single command - ConsensusReadForwardRequest forwardRequest = new ConsensusReadForwardRequest(firstCommand, consistencyLevel); + // Create forward request + ConsensusReadForwardRequest forwardRequest = new ConsensusReadForwardRequest(command, consistencyLevel); Message<ConsensusReadForwardRequest> message = Message.out(CONSENSUS_READ_FORWARD_REQ, forwardRequest); try diff --git a/src/java/org/apache/cassandra/service/paxos/ConsensusReadForwardRequest.java b/src/java/org/apache/cassandra/service/paxos/ConsensusReadForwardRequest.java index 923897e690..d376285c3a 100644 --- a/src/java/org/apache/cassandra/service/paxos/ConsensusReadForwardRequest.java +++ b/src/java/org/apache/cassandra/service/paxos/ConsensusReadForwardRequest.java @@ -32,7 +32,8 @@ import org.apache.cassandra.io.util.DataOutputPlus; * This is used when the original coordinator is not a replica but needs to * execute a consensus read for a tracked keyspace that requires proper coordination. * - * Consensus reads only ever contain a single read command. + * Serial reads are single partition, enforced in StorageProxy.readWithConsensusInternal, so one command + * goes out and one partition's rows come back. */ public class ConsensusReadForwardRequest { diff --git a/src/java/org/apache/cassandra/service/paxos/Paxos.java b/src/java/org/apache/cassandra/service/paxos/Paxos.java index dd88bae1bd..af9a0f6ec9 100644 --- a/src/java/org/apache/cassandra/service/paxos/Paxos.java +++ b/src/java/org/apache/cassandra/service/paxos/Paxos.java @@ -932,8 +932,6 @@ public class Paxos throws InvalidRequestException, UnavailableException, ReadFailureException, ReadTimeoutException { long start = nanoTime(); - if (group.queries.size() > 1) - throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one partition at a time"); long deadline = requestTime.computeDeadline(DatabaseDescriptor.getReadRpcTimeout(NANOSECONDS)); int failedAttemptsDueToContention = 0; diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java index 5fd1e8393e..9fb9f8110b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java @@ -18,6 +18,8 @@ package org.apache.cassandra.distributed.test.tracking; +import java.util.Arrays; +import java.util.Comparator; import java.util.List; import org.junit.Test; @@ -41,8 +43,12 @@ import org.apache.cassandra.schema.ReplicationType; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.service.StorageService; +import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows; +import static org.apache.cassandra.distributed.shared.AssertUtils.row; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.getOnlyLogId; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.summaryIdSpace; +import static org.apache.cassandra.distributed.test.tracking.PaxosMigrationTestUtils.assertCasApplied; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -130,20 +136,7 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl logger.info("DEBUG testCasForwarding: Using replica coordinator: " + coordinatorNode); } else { // Find the non-replica node - coordinatorNode = -1; - for (int i = 1; i <= 4; i++) { - boolean isReplica = false; - for (int replicaNode : replicaNodes) { - if (i == replicaNode) { - isReplica = true; - break; - } - } - if (!isReplica) { - coordinatorNode = i; - break; - } - } + coordinatorNode = nonReplicaNode(replicaNodes); logger.info("DEBUG testCasForwarding: Using non-replica coordinator: " + coordinatorNode); } @@ -175,7 +168,9 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl // Perform CAS operation from determined coordinator // This should trigger forwarding if coordinator is not a replica - cluster.coordinator(coordinatorNode).execute(CONDITIONAL_INSERT_CQL, ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM); + Object[][] casResult = cluster.coordinator(coordinatorNode).execute(CONDITIONAL_INSERT_CQL, ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM); + // A forwarded CAS has to hand its outcome back, not just its side effect + assertCasApplied(casResult); // Verify that unblocked replica nodes have the mutation tracked for (int replicaNode : replicaNodes) { @@ -267,6 +262,47 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl for (int i = 1; i < mutationIds.length; i++) { assertEquals("All replicas should have same mutation ID", mutationIds[0], mutationIds[i]); } + + // A CAS that does not apply hands back the row that stopped it. Left until here so the + // tracking assertions above see only the mutation the applying CAS wrote. + Object[][] notAppliedResult = cluster.coordinator(coordinatorNode).execute(CONDITIONAL_INSERT_CQL, ConsistencyLevel.SERIAL, ConsistencyLevel.QUORUM); + assertRows(notAppliedResult, row(false, 1, 1)); + + // A serial read spanning partitions is refused wherever it is coordinated. Before the fix a + // non-replica coordinator answered it with partition 1 alone, no error and no warning. + cluster.coordinator(coordinatorNode).execute(withKeyspace("INSERT INTO %s.tbl (k, v) VALUES (2, 2)"), ConsistencyLevel.ALL); + String multiPartitionRead = withKeyspace("SELECT * FROM %s.tbl WHERE k IN (1, 2)"); + for (int node : new int[]{ replicaNodes[0], nonReplicaNode(replicaNodes) }) + { + assertThatThrownBy(() -> cluster.coordinator(node).execute(multiPartitionRead, ConsistencyLevel.SERIAL)) + .describedAs("SERIAL read spanning two partitions, coordinated by node " + node) + .hasMessageContaining("may only be requested for one partition at a time"); + + // Both partitions are readable without SERIAL. Sort by key, they arrive in token order + Object[][] bothPartitions = cluster.coordinator(node).execute(multiPartitionRead, ConsistencyLevel.ALL); + Arrays.sort(bothPartitions, Comparator.comparingInt(partitionRow -> (int) partitionRow[0])); + assertRows(bothPartitions, row(1, 1), row(2, 2)); + } + } + } + + /** The one node of four that is not a replica for key 1, and so has to forward. */ + private static int nonReplicaNode(int[] replicaNodes) + { + for (int node = 1; node <= 4; node++) + { + boolean isReplica = false; + for (int replicaNode : replicaNodes) + { + if (node == replicaNode) + { + isReplica = true; + break; + } + } + if (!isReplica) + return node; } + throw new AssertionError("Expected one of the four nodes to not be a replica for key 1"); } -} \ No newline at end of file +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
