From 6167e436c302b41569e2a53c08c2351a8ac04be2 Mon Sep 17 00:00:00 2001 From: David Capwell Date: Mon, 21 Aug 2023 13:32:42 -0700 Subject: [PATCH] CEP-15 (C*): when loading commands that have empty waiting_on, make sure not to loose the partial deps (#3590) patch by David Capwell; reviewed by Aleksey Yeschenko for CASSANDRA-18783 --- modules/accord | 2 +- .../service/accord/AccordKeyspace.java | 4 +- .../service/accord/AccordKeyspaceTest.java | 90 +++++++++++++++++++ .../service/accord/AccordTestUtils.java | 9 ++ 4 files changed, 102 insertions(+), 3 deletions(-) create mode 100644 test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java diff --git a/modules/accord b/modules/accord index 91336705bd..8c7a3c9ef4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 91336705bde8332954e849219d73205d68fa168a +Subproject commit 8c7a3c9ef4209d635b186189e17a2d9e728e9871 diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index d845c74c53..5f31f86276 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -636,10 +636,10 @@ public class AccordKeyspace return bytes != null && !ByteBufferAccessor.instance.isEmpty(bytes) ? deserialize(bytes, serializer) : null; } - private static WaitingOn deserializeWaitingOn(Deps deps, ByteBuffer bytes) throws IOException + private static WaitingOn deserializeWaitingOn(@Nullable Deps deps, @Nullable ByteBuffer bytes) throws IOException { if (bytes == null || !bytes.hasRemaining()) - return WaitingOn.EMPTY; + return deps == null ? WaitingOn.EMPTY : WaitingOn.none(deps); return WaitingOnSerializer.deserialize(deps, new DataInputBuffer(bytes, false)); } diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java new file mode 100644 index 0000000000..d868850845 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java @@ -0,0 +1,90 @@ +/* + * 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.service.accord; + +import java.util.Collections; +import java.util.concurrent.atomic.AtomicLong; + +import org.junit.Test; + +import accord.api.RoutingKey; +import accord.local.Command; +import accord.local.CommonAttributes; +import accord.local.Node; +import accord.local.SaveStatus; +import accord.local.Status; +import accord.primitives.Ballot; +import accord.primitives.Deps; +import accord.primitives.FullRoute; +import accord.primitives.KeyDeps; +import accord.primitives.Keys; +import accord.primitives.PartialTxn; +import accord.primitives.RangeDeps; +import accord.primitives.Ranges; +import accord.primitives.Routable; +import accord.primitives.Timestamp; +import accord.primitives.Txn; +import accord.primitives.TxnId; +import org.apache.cassandra.cql3.CQLTester; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.service.accord.api.AccordRoutingKey; +import org.assertj.core.api.Assertions; + +import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn; +import static org.apache.cassandra.service.accord.AccordTestUtils.wrapInTxn; + +public class AccordKeyspaceTest extends CQLTester.InMemory +{ + private static final Ranges GLOBAL_SCOPE = Ranges.of(new TokenRange(AccordRoutingKey.SentinelKey.min(KEYSPACE), AccordRoutingKey.SentinelKey.max(KEYSPACE))); + + @Test + public void serde() + { + AtomicLong now = new AtomicLong(); + + String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c))"); + + AccordCommandStore store = AccordTestUtils.createAccordCommandStore(now::incrementAndGet, KEYSPACE, tableName); + + TxnId id = new TxnId(Timestamp.fromValues(1, 42, new Node.Id(1)), Txn.Kind.Read, Routable.Domain.Key); + + Txn txn = createTxn(wrapInTxn(String.format("SELECT * FROM %s.%s WHERE k=? LIMIT 1", KEYSPACE, tableName)), Collections.singletonList(42)); + + PartialTxn partialTxn = txn.slice(GLOBAL_SCOPE, true); + RoutingKey routingKey = partialTxn.keys().get(0).asKey().toUnseekable(); + FullRoute route = partialTxn.keys().toRoute(routingKey); + Deps deps = new Deps(KeyDeps.none((Keys) txn.keys()), RangeDeps.NONE); + + + CommonAttributes.Mutable common = new CommonAttributes.Mutable(id); + common.partialTxn(partialTxn); + common.route(route); + common.partialDeps(deps.slice(GLOBAL_SCOPE)); + common.durability(Status.Durability.NotDurable); + Command.WaitingOn waitingOn = Command.WaitingOn.none(deps.slice(GLOBAL_SCOPE)); + Command.Committed committed = Command.SerializerSupport.committed(common, SaveStatus.Committed, id, Ballot.ZERO, Ballot.ZERO, waitingOn); + + AccordSafeCommand safeCommand = new AccordSafeCommand(AccordTestUtils.loaded(id, null)); + safeCommand.set(committed); + Mutation mutation = AccordKeyspace.getCommandMutation(store, safeCommand, 42); + mutation.apply(); + + Assertions.assertThat(AccordKeyspace.loadCommand(store, id)).isEqualTo(committed); + } +} \ No newline at end of file diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index c8a5872d9f..628a094257 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -230,6 +230,15 @@ public class AccordTestUtils } + public static String wrapInTxn(String query) + { + if (!query.endsWith(";")) + query += ";"; + return "BEGIN TRANSACTION\n" + + query + + "\nCOMMIT TRANSACTION"; + } + public static Txn createTxn(String query) { return createTxn(query, QueryOptions.DEFAULT);