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 d9010434ad06938d6e1dcdbdf7c78a94a646faf5 Author: Ariel Weisberg <[email protected]> AuthorDate: Tue Aug 11 19:22:52 2026 -0400 Fix reversed rows in forwarded consensus reads CasForwardResponse materializes the replica coordinator's result into a FilteredPartition, which normalizes rows to clustering order, then rebuilt the iterator with rowIterator(false). A reversed slice forwarded from a non-replica coordinator came back ascending. Capture isReverseOrder() while materializing and carry it over the wire in a new IS_REVERSED flag. CAS is unaffected, CQL3CasRequest.readCommand never reverses. --- .../service/paxos/CasForwardResponse.java | 83 +++++--- .../MutationTrackingCasForwardingTest.java | 219 +++++++++++++++++++++ .../cassandra/service/paxos/CasForwardingTest.java | 99 +++++++++- 3 files changed, 371 insertions(+), 30 deletions(-) diff --git a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java index a996fa45a9..457b732606 100644 --- a/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java +++ b/src/java/org/apache/cassandra/service/paxos/CasForwardResponse.java @@ -53,6 +53,13 @@ import static org.apache.cassandra.db.rows.DeserializationHelper.Flag.FROM_REMOT public class CasForwardResponse { private final FilteredPartition result; + + /** + * Direction {@link #result} was read in. {@link FilteredPartition} always stores rows in clustering + * order, so a reversed slice needs its direction carried alongside. + */ + private final boolean reversed; + public final CassandraException exception; @Nonnull @@ -60,52 +67,74 @@ public class CasForwardResponse public CasForwardResponse(RowIterator result, List<String> warnings) { - this(materialize(result), null, warnings); + this(Materialized.of(result), null, warnings); } public CasForwardResponse(PartitionIterator result, List<String> warnings) { - this(materialize(result), null, warnings); + this(Materialized.of(result), null, warnings); } public CasForwardResponse(CassandraException exception, List<String> warnings) { - this((FilteredPartition) null, exception, warnings); + this(Materialized.NONE, exception, warnings); + } + + private CasForwardResponse(Materialized result, CassandraException exception, List<String> warnings) + { + this(result.partition, result.reversed, exception, warnings); } - private CasForwardResponse(FilteredPartition result, CassandraException exception, List<String> warnings) + private CasForwardResponse(FilteredPartition result, boolean reversed, CassandraException exception, List<String> warnings) { this.result = result; + this.reversed = reversed; this.exception = exception; this.warnings = warnings == null ? Collections.emptyList() : warnings; } - private static FilteredPartition materialize(RowIterator rows) + /** A materialized result and the direction it was read in, taken before the iterator is consumed. */ + private static class Materialized { - if (rows == null) - return null; + private static final Materialized NONE = new Materialized(null, false); + + private final FilteredPartition partition; + private final boolean reversed; - try (RowIterator toClose = rows) + private Materialized(FilteredPartition partition, boolean reversed) { - return new FilteredPartition(toClose); + this.partition = partition; + this.reversed = reversed; } - } - private static FilteredPartition materialize(PartitionIterator partitions) - { - if (partitions == null) - return null; + private static Materialized of(RowIterator rows) + { + if (rows == null) + return NONE; + + try (RowIterator toClose = rows) + { + boolean reversed = toClose.isReverseOrder(); + return new Materialized(new FilteredPartition(toClose), reversed); + } + } - try (PartitionIterator toClose = partitions) + private static Materialized of(PartitionIterator partitions) { - 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; + if (partitions == null) + return NONE; + + try (PartitionIterator toClose = partitions) + { + if (!toClose.hasNext()) + return NONE; + + Materialized materialized = of(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; + } } } @@ -121,7 +150,7 @@ public class CasForwardResponse public RowIterator rowIterator() { - return result == null ? null : result.rowIterator(false); + return result == null ? null : result.rowIterator(reversed); } public PartitionIterator partitionIterator() @@ -137,6 +166,8 @@ public class CasForwardResponse private static final int HAS_RESULT = 0x01; private static final int HAS_EXCEPTION = 0x02; private static final int HAS_WARNINGS = 0x04; + /** Rows are always written in clustering order, so this records the direction asked for. */ + private static final int IS_REVERSED = 0x08; @Override public void serialize(CasForwardResponse response, DataOutputPlus out, int version) throws IOException @@ -144,6 +175,7 @@ public class CasForwardResponse int flags = (response.hasResult() ? HAS_RESULT : 0) | (response.exception != null ? HAS_EXCEPTION : 0) | (!response.warnings.isEmpty() ? HAS_WARNINGS : 0) + | (response.reversed ? IS_REVERSED : 0) ; out.write(flags); @@ -171,6 +203,7 @@ public class CasForwardResponse boolean hasResult = (flags & HAS_RESULT) != 0; boolean hasException = (flags & HAS_EXCEPTION) != 0; boolean hasWarnings = (flags & HAS_WARNINGS) != 0; + boolean reversed = (flags & IS_REVERSED) != 0; FilteredPartition result = null; if (hasResult) @@ -194,7 +227,7 @@ public class CasForwardResponse if (hasWarnings) warnings = CollectionSerializers.deserializeList(in, version, StringSerializer.instance); - return new CasForwardResponse(result, exception, warnings); + return new CasForwardResponse(result, reversed, exception, warnings); } @Override 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 9fb9f8110b..1955749acc 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingCasForwardingTest.java @@ -20,7 +20,12 @@ package org.apache.cassandra.distributed.test.tracking; import java.util.Arrays; import java.util.Comparator; +import java.util.HashSet; import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import com.google.common.collect.Iterators; import org.junit.Test; import org.slf4j.Logger; @@ -31,9 +36,11 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.shared.AssertUtils; import org.apache.cassandra.distributed.test.TestBaseImpl; import org.apache.cassandra.gms.Gossiper; import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.net.Verb; import org.apache.cassandra.replication.CoordinatorLogId; import org.apache.cassandra.replication.MutationSummary; import org.apache.cassandra.replication.MutationTrackingService; @@ -61,6 +68,13 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl private static final String CONDITIONAL_INSERT_CQL = "INSERT INTO " + KEYSPACE + ".tbl (k, v) VALUES (1, 1) IF NOT EXISTS"; + /** Partition read back by the forwarded SERIAL read tests, and how many rows it holds. */ + private static final int READ_KEY = 1; + private static final int READ_ROWS = 4; + + /** Forwarded consensus reads seen by the message filter, asserted per query by the read helpers. */ + private final AtomicInteger consensusReadForwards = new AtomicInteger(); + @Test public void testCasForwardingPaxosV1() throws Throwable { @@ -85,6 +99,211 @@ public class MutationTrackingCasForwardingTest extends TestBaseImpl testCasForwarding("v2", true); // replica coordinator } + @Test + public void testForwardedSerialReadOrderingPaxosV1() throws Throwable + { + testForwardedSerialReadOrdering("v1"); + } + + @Test + public void testForwardedSerialReadOrderingPaxosV2() throws Throwable + { + testForwardedSerialReadOrdering("v2"); + } + + /** + * A forwarded SERIAL read can return a whole partition, unlike CAS, so a reversed slice has to come + * back from the replica coordinator in the order it was read in. + */ + private void testForwardedSerialReadOrdering(String paxosVariant) throws Throwable + { + try (Cluster cluster = Cluster.build(4) + .withConfig(cfg -> cfg.with(Feature.NETWORK) + .with(Feature.GOSSIP) + .set("paxos_variant", paxosVariant)) + .start()) + { + cluster.schemaChange(withKeyspace("CREATE KEYSPACE %s WITH replication = " + + "{'class': 'SimpleStrategy', 'replication_factor': 3} " + + "AND replication_type='tracked';")); + cluster.schemaChange(withKeyspace("CREATE TABLE %s.ascending (k int, c int, v int, PRIMARY KEY (k, c));")); + cluster.schemaChange(withKeyspace("CREATE TABLE %s.descending (k int, c int, v int, PRIMARY KEY (k, c)) " + + "WITH CLUSTERING ORDER BY (c DESC);")); + + int coordinator = nonReplicaCoordinator(cluster, "ascending", READ_KEY); + logger.info("DEBUG testForwardedSerialReadOrdering: Using non-replica coordinator: " + coordinator); + + for (int c = 1; c <= READ_ROWS; c++) + { + cluster.coordinator(coordinator).execute(withKeyspace("INSERT INTO %s.ascending (k, c, v) VALUES (?, ?, ?)"), + ConsistencyLevel.ALL, READ_KEY, c, c); + cluster.coordinator(coordinator).execute(withKeyspace("INSERT INTO %s.descending (k, c, v) VALUES (?, ?, ?)"), + ConsistencyLevel.ALL, READ_KEY, c, c); + } + + // Count the forwards, so each assertion below can prove the read left the coordinator + cluster.filters() + .verbs(Verb.CONSENSUS_READ_FORWARD_REQ.id) + .messagesMatching((from, to, message) -> { + consensusReadForwards.incrementAndGet(); + return false; // count only, deliver as normal + }) + .drop(); + + // Controls: rows are stored in the clustering order these read, so the direction cannot show + assertSerialReadOrder(cluster, coordinator, "ascending", "", 1, 2, 3, 4); + assertSerialReadOrder(cluster, coordinator, "descending", "", 4, 3, 2, 1); + + // Reversed slices, the reads that come back the wrong way round without the direction + assertSerialReadOrder(cluster, coordinator, "ascending", " ORDER BY c DESC", 4, 3, 2, 1); + assertSerialReadOrder(cluster, coordinator, "descending", " ORDER BY c ASC", 1, 2, 3, 4); + + // Explicit orderings that agree with the clustering order + assertSerialReadOrder(cluster, coordinator, "ascending", " ORDER BY c ASC", 1, 2, 3, 4); + assertSerialReadOrder(cluster, coordinator, "descending", " ORDER BY c DESC", 4, 3, 2, 1); + + // An empty result carries no partition, so no direction is consulted + int missingKey = unwrittenNonReplicaKey(cluster, "ascending", coordinator); + assertForwardedReadIsEmpty(cluster, coordinator, "ascending", "k = " + missingKey, ""); + assertForwardedReadIsEmpty(cluster, coordinator, "descending", "k = " + missingKey, " ORDER BY c ASC"); + // A partition that exists, sliced so that it selects no rows + assertForwardedReadIsEmpty(cluster, coordinator, "ascending", "k = " + READ_KEY + " AND c > 100", " ORDER BY c DESC"); + + // Paging does more than reorder: the next page's boundary comes from these rows as they + // stream past, so a page handed back ascending repeats or skips rows + assertPagedSerialReadOrder(cluster, coordinator, "ascending", " ORDER BY c DESC", 4, 3, 2, 1); + assertPagedSerialReadOrder(cluster, coordinator, "descending", " ORDER BY c ASC", 1, 2, 3, 4); + } + } + + /** + * Asserts the forwarded SERIAL read returns the partition in the order the same query returns it at a + * non-serial consistency, which does not forward. + */ + private void assertSerialReadOrder(Cluster cluster, int coordinator, String table, String ordering, int... expected) + { + String cql = withKeyspace("SELECT c FROM %s." + table + " WHERE k = " + READ_KEY) + ordering; + assertForwardedReadMatches(cluster, coordinator, cql, expectedRows(expected)); + } + + private void assertForwardedReadIsEmpty(Cluster cluster, int coordinator, String table, String predicate, String ordering) + { + String cql = withKeyspace("SELECT c FROM %s." + table + " WHERE " + predicate) + ordering; + assertForwardedReadMatches(cluster, coordinator, cql, new Object[0][]); + } + + /** + * Checks the forwarded result against the ordinary read path. The forward count is asserted per query, + * since one assertion at the end would be satisfied by the control reads alone. + */ + private void assertForwardedReadMatches(Cluster cluster, int coordinator, String cql, Object[][] expectedRows) + { + int beforeReference = consensusReadForwards.get(); + assertRowsInOrder(cql, ConsistencyLevel.ALL, cluster, coordinator, expectedRows); + assertEquals("A non-serial read should not have been forwarded: " + cql, + beforeReference, consensusReadForwards.get()); + + for (ConsistencyLevel serial : new ConsistencyLevel[]{ ConsistencyLevel.SERIAL, ConsistencyLevel.LOCAL_SERIAL }) + { + int before = consensusReadForwards.get(); + assertRowsInOrder(cql, serial, cluster, coordinator, expectedRows); + assertTrue('"' + cql + "\" at " + serial + " should have been forwarded to a replica coordinator", + consensusReadForwards.get() > before); + } + } + + /** The same read paged in twos, so the direction has to hold within a page and across the boundary. */ + private void assertPagedSerialReadOrder(Cluster cluster, int coordinator, String table, String ordering, int... expected) + { + String cql = withKeyspace("SELECT c FROM %s." + table + " WHERE k = " + READ_KEY) + ordering; + Object[][] expectedRows = expectedRows(expected); + + for (ConsistencyLevel consistencyLevel : new ConsistencyLevel[]{ ConsistencyLevel.ALL, ConsistencyLevel.SERIAL }) + { + Object[][] actual = Iterators.toArray(cluster.coordinator(coordinator).executeWithPaging(cql, consistencyLevel, 2), + Object[].class); + try + { + AssertUtils.assertRows(actual, expectedRows); + } + catch (AssertionError e) + { + throw new AssertionError('"' + cql + "\" paged at " + consistencyLevel + ": " + e.getMessage(), e); + } + } + } + + private static Object[][] expectedRows(int... expected) + { + Object[][] expectedRows = new Object[expected.length][]; + for (int i = 0; i < expected.length; i++) + expectedRows[i] = AssertUtils.row(expected[i]); + return expectedRows; + } + + private void assertRowsInOrder(String cql, ConsistencyLevel consistencyLevel, Cluster cluster, int coordinator, Object[][] expectedRows) + { + Object[][] actual = cluster.coordinator(coordinator).execute(cql, consistencyLevel); + try + { + AssertUtils.assertRows(actual, expectedRows); + } + catch (AssertionError e) + { + throw new AssertionError('"' + cql + "\" at " + consistencyLevel + ": " + e.getMessage(), e); + } + } + + /** + * With RF=3 across four nodes exactly one node is not a replica for the key, and a non-replica + * coordinator is the only one that forwards. + */ + private int nonReplicaCoordinator(Cluster cluster, String tableName, int key) + { + Set<Integer> replicaNodes = replicaNodes(cluster, tableName, key); + + for (int node = 1; node <= cluster.size(); node++) + { + if (!replicaNodes.contains(node)) + return node; + } + + throw new AssertionError("Every node is a replica for key " + key + ", nothing would be forwarded: " + replicaNodes); + } + + /** + * A key nothing has been written to that the given coordinator is not a replica for, so that reading + * it both forwards and comes back empty. Only {@link #READ_KEY} is ever written. + */ + private int unwrittenNonReplicaKey(Cluster cluster, String tableName, int coordinator) + { + for (int key = READ_KEY + 1; key <= READ_KEY + 100; key++) + { + if (!replicaNodes(cluster, tableName, key).contains(coordinator)) + return key; + } + + throw new AssertionError("Found no unwritten key that node " + coordinator + " is not a replica for"); + } + + private Set<Integer> replicaNodes(Cluster cluster, String tableName, int key) + { + String keyspaceName = KEYSPACE; + String replicaEndpoints = cluster.get(1).callOnInstance( + () -> String.join(",", StorageService.instance.getNaturalEndpointsWithPort(keyspaceName, tableName, Integer.toString(key)))); + + Set<Integer> replicaNodes = new HashSet<>(); + for (String endpoint : replicaEndpoints.split(",")) + { + // Addresses arrive as "127.0.0.3:7000" or "/127.0.0.3:7000" + int colonIndex = endpoint.indexOf(':'); + String hostPart = colonIndex > 0 ? endpoint.substring(0, colonIndex) : endpoint; + replicaNodes.add(Integer.parseInt(hostPart.substring(hostPart.lastIndexOf('.') + 1))); + } + + return replicaNodes; + } + private void testCasForwarding(String paxosVariant, boolean useReplicaCoordinator) throws Throwable { try (Cluster cluster = Cluster.build(4) diff --git a/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java b/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java index 8697e7f0d9..36401d9e6d 100644 --- a/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java +++ b/test/unit/org/apache/cassandra/service/paxos/CasForwardingTest.java @@ -25,6 +25,8 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.ExecutionException; +import com.google.common.collect.Lists; + import org.junit.BeforeClass; import org.junit.Test; @@ -36,7 +38,9 @@ import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.EmptyIterators; import org.apache.cassandra.db.RegularAndStaticColumns; import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.Slices; import org.apache.cassandra.db.WriteType; +import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.partitions.PartitionIterator; import org.apache.cassandra.db.partitions.PartitionIterators; import org.apache.cassandra.db.partitions.PartitionUpdate; @@ -162,6 +166,49 @@ public class CasForwardingTest twoRowValues(), Collections.emptyList()); } + /** + * {@link org.apache.cassandra.db.partitions.FilteredPartition} always stores rows in clustering order, + * so a reversed multi-row result has to carry the direction it was read in. + */ + @Test + public void testReversedResultRoundTrip() throws IOException + { + // Guard the fixture: a reversed result that matched the ascending one would assert nothing + List<String> ascending = twoRowValues(false); + List<String> descending = twoRowValues(true); + assertEquals("Reversed fixture should be the ascending rows in reverse", + Lists.reverse(ascending), descending); + + assertRoundTrips(new CasForwardResponse(twoRowResult(true), null), descending, true, Collections.emptyList()); + } + + @Test + public void testReversedConsensusReadResultRoundTrip() throws IOException + { + // The read forwarding verb reaches the same payload through the PartitionIterator constructor + assertRoundTrips(new CasForwardResponse(PartitionIterators.singletonIterator(twoRowResult(true)), null), + twoRowValues(true), true, Collections.emptyList()); + } + + /** A handler reads the result locally before messaging serializes it, so reading must not consume it. */ + @Test + public void testReversedResultIsReadableRepeatedly() throws IOException + { + CasForwardResponse response = new CasForwardResponse(twoRowResult(true), null); + + List<String> expected = twoRowValues(true); + for (int i = 0; i < 3; i++) + { + assertEquals("partitionIterator() read " + i, expected, rowValues(response.partitionIterator())); + try (RowIterator rows = response.rowIterator()) + { + assertTrue("rowIterator() read " + i + " should still be reversed", rows.isReverseOrder()); + } + } + + assertRoundTrips(response, expected, true, Collections.emptyList()); + } + @Test public void testResultAndWarningsRoundTrip() throws IOException { @@ -296,7 +343,15 @@ public class CasForwardingTest List<String> expectedRows, List<String> expectedWarnings) throws IOException { - assertRows("Result", expectedRows, response); + return assertRoundTrips(response, expectedRows, false, expectedWarnings); + } + + private static CasForwardResponse assertRoundTrips(CasForwardResponse response, + List<String> expectedRows, + boolean expectedReversed, + List<String> expectedWarnings) throws IOException + { + assertRows("Result", expectedRows, expectedReversed, response); byte[] bytes = serializeCheckingSize(response); assertArrayEquals("Repeated serialization should produce identical bytes", @@ -310,7 +365,7 @@ public class CasForwardingTest assertEquals("Success should survive the round trip", response.isSuccess(), deserialized.isSuccess()); assertEquals("Warnings should survive the round trip", expectedWarnings, deserialized.warnings); - assertRows("Deserialized result", expectedRows, deserialized); + assertRows("Deserialized result", expectedRows, expectedReversed, deserialized); byte[] reserialized = serializeCheckingSize(deserialized); // A deserialized exception picks up the stack frames of the deserialize call, so only the @@ -334,7 +389,7 @@ public class CasForwardingTest } } - private static void assertRows(String what, List<String> expectedRows, CasForwardResponse response) + private static void assertRows(String what, List<String> expectedRows, boolean expectedReversed, CasForwardResponse response) { if (expectedRows == null) { @@ -346,10 +401,36 @@ public class CasForwardingTest { assertTrue(what + " should be present", response.hasResult()); assertEquals(what + " rows should match", expectedRows, rowValues(response.partitionIterator())); + assertEquals(what + " should report the direction it was read in", + expectedReversed, isReverseOrder(response.rowIterator())); + assertEquals(what + " partition iterator should report the direction it was read in", + expectedReversed, isReverseOrder(response.partitionIterator())); + } + } + + private static boolean isReverseOrder(RowIterator rows) + { + try (RowIterator toClose = rows) + { + return toClose.isReverseOrder(); + } + } + + private static boolean isReverseOrder(PartitionIterator partitions) + { + try (PartitionIterator toClose = partitions) + { + assertTrue("Result should contain a partition", toClose.hasNext()); + return isReverseOrder(toClose.next()); } } private static RowIterator twoRowResult() + { + return twoRowResult(false); + } + + private static RowIterator twoRowResult(boolean reversed) { TableMetadata metadata = Schema.instance.getTableMetadata(KEYSPACE1, CF_STANDARD1); @@ -360,12 +441,20 @@ public class CasForwardingTest .clustering("c2").add("val", "v2") .buildUpdate())); - return UnfilteredRowIterators.filter(update.unfilteredIterator(), FBUtilities.nowInSeconds()); + // all(update.columns()) rather than all(metadata) keeps the non-reversed fixture byte-identical to + // the no-arg unfilteredIterator() this replaced, whose selection lands in the serialization header + return UnfilteredRowIterators.filter(update.unfilteredIterator(ColumnFilter.all(update.columns()), Slices.ALL, reversed), + FBUtilities.nowInSeconds()); } private static List<String> twoRowValues() { - return rowValues(PartitionIterators.singletonIterator(twoRowResult())); + return twoRowValues(false); + } + + private static List<String> twoRowValues(boolean reversed) + { + return rowValues(PartitionIterators.singletonIterator(twoRowResult(reversed))); } private static List<String> rowValues(PartitionIterator partitions) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
