You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by if...@apache.org on 2025/07/23 07:47:07 UTC
(cassandra) branch trunk updated: Add a way to patch accord journal commands via virtual tables
This is an automated email from the ASF dual-hosted git repository.
ifesdjeen pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new 810dd55128 Add a way to patch accord journal commands via virtual tables
810dd55128 is described below
commit 810dd55128b2fc087b1002982c80087f37a963fe
Author: Alex Petrov <ol...@gmail.com>
AuthorDate: Mon Jul 7 16:50:12 2025 +0200
Add a way to patch accord journal commands via virtual tables
Patch by Alex Petrov; reviewed by Benedict Elliott Smith for CASSANDRA-20765
---
.../cassandra/db/virtual/AccordDebugKeyspace.java | 153 +++++++++++++++++----
.../io/sstable/indexsummary/IndexSummary.java | 2 +-
.../cassandra/service/accord/AccordJournal.java | 11 ++
.../accord/AccordJournalValueSerializers.java | 2 +-
.../db/virtual/AccordDebugKeyspaceTest.java | 81 ++++++++++-
5 files changed, 218 insertions(+), 31 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
index 685e44a061..a9f6edc26a 100644
--- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.db.virtual;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
@@ -44,18 +45,23 @@ import accord.api.TraceEventType;
import accord.impl.CommandChange;
import accord.impl.progresslog.DefaultProgressLog;
import accord.impl.progresslog.TxnStateKind;
+import accord.local.Cleanup;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
+import accord.local.Commands;
import accord.local.DurableBefore;
import accord.local.MaxConflicts;
import accord.local.RejectBefore;
+import accord.local.SafeCommand;
+import accord.local.SafeCommandStore;
import accord.local.StoreParticipants;
import accord.local.durability.ShardDurability;
import accord.primitives.Participants;
import accord.primitives.Status;
import accord.primitives.TxnId;
import accord.utils.Invariants;
+import accord.utils.async.AsyncChains;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.DataRange;
@@ -96,8 +102,8 @@ import static accord.local.RedundantStatus.Property.LOCALLY_WITNESSED;
import static accord.local.RedundantStatus.Property.MAJORITY_APPLIED;
import static accord.local.RedundantStatus.Property.PRE_BOOTSTRAP;
import static accord.local.RedundantStatus.Property.SHARD_APPLIED;
-import static java.lang.String.format;
import static com.google.common.collect.ImmutableList.toImmutableList;
+import static java.lang.String.format;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
import static org.apache.cassandra.schema.SchemaConstants.VIRTUAL_ACCORD_DEBUG;
@@ -114,8 +120,9 @@ public class AccordDebugKeyspace extends VirtualKeyspace
public static final String REDUNDANT_BEFORE = "redundant_before";
public static final String REJECT_BEFORE = "reject_before";
public static final String TXN_BLOCKED_BY = "txn_blocked_by";
- public static final String TXN = "txn";
public static final String JOURNAL = "journal";
+ public static final String TXN = "txn";
+ public static final String TXN_UPDATE = "txn_update";
public static final String TXN_TRACE = "txn_trace";
public static final String TXN_TRACES = "txn_traces";
@@ -135,6 +142,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
new TxnBlockedByTable(),
new TxnTable(),
new JournalTable(),
+ new TxnUpdateTable(),
new TxnTraceTable(),
new TxnTracesTable()
));
@@ -820,21 +828,21 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
Command command = e.command;
ds.row(txnIdStr, e.commandStoreId)
- .column("save_status", Objects.toString(command.saveStatus()))
- .column("route", Objects.toString(command.route()))
+ .column("save_status", toStringOrNull(command.saveStatus()))
+ .column("route", toStringOrNull(command.route()))
.column("participants_owns", toStr(command, StoreParticipants::owns, StoreParticipants::stillOwns))
.column("participants_touches", toStr(command, StoreParticipants::touches, StoreParticipants::stillTouches))
- .column("participants_has_touched", Objects.toString(command.participants().hasTouched()))
+ .column("participants_has_touched", toStringOrNull(command.participants().hasTouched()))
.column("participants_executes", toStr(command, StoreParticipants::executes, StoreParticipants::stillExecutes))
.column("participants_waits_on", toStr(command, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn))
- .column("durability", Objects.toString(command.durability()))
- .column("execute_at", Objects.toString(command.executeAt()))
- .column("executes_at_least", Objects.toString(command.executesAtLeast()))
- .column("txn", Objects.toString(command.partialTxn()))
- .column("deps", Objects.toString(command.partialDeps()))
- .column("waiting_on", Objects.toString(command.waitingOn()))
- .column("writes", Objects.toString(command.writes()))
- .column("result", Objects.toString(command.result()));
+ .column("durability", toStringOrNull(command.durability()))
+ .column("execute_at", toStringOrNull(command.executeAt()))
+ .column("executes_at_least", toStringOrNull(command.executesAtLeast()))
+ .column("txn", toStringOrNull(command.partialTxn()))
+ .column("deps", toStringOrNull(command.partialDeps()))
+ .column("waiting_on", toStringOrNull(command.waitingOn()))
+ .column("writes", toStringOrNull(command.writes()))
+ .column("result", toStringOrNull(command.result()));
}
return ds.getPartition(partitionKey);
@@ -926,20 +934,20 @@ public class AccordDebugKeyspace extends VirtualKeyspace
StoreParticipants participants = b.participants();
if (participants == null) participants = StoreParticipants.empty(txnId);
ds.row(txnIdStr, e.commandStoreId, e.segment, e.position)
- .column("save_status", Objects.toString(b.saveStatus()))
- .column("route", Objects.toString(participants.route()))
+ .column("save_status", toStringOrNull(b.saveStatus()))
+ .column("route", toStringOrNull(participants.route()))
.column("participants_owns", toStr(participants, StoreParticipants::owns, StoreParticipants::stillOwns))
.column("participants_touches", toStr(participants, StoreParticipants::touches, StoreParticipants::stillTouches))
- .column("participants_has_touched", Objects.toString(participants.hasTouched()))
+ .column("participants_has_touched", toStringOrNull(participants.hasTouched()))
.column("participants_executes", toStr(participants, StoreParticipants::executes, StoreParticipants::stillExecutes))
.column("participants_waits_on", toStr(participants, StoreParticipants::waitsOn, StoreParticipants::stillWaitsOn))
- .column("durability", Objects.toString(b.durability()))
- .column("execute_at", Objects.toString(b.executeAt()))
- .column("executes_at_least", Objects.toString(b.executesAtLeast()))
- .column("txn", Objects.toString(b.partialTxn()))
- .column("deps", Objects.toString(b.partialDeps()))
- .column("writes", Objects.toString(b.writes()))
- .column("result", Objects.toString(b.result()));
+ .column("durability", toStringOrNull(b.durability()))
+ .column("execute_at", toStringOrNull(b.executeAt()))
+ .column("executes_at_least", toStringOrNull(b.executesAtLeast()))
+ .column("txn", toStringOrNull(b.partialTxn()))
+ .column("deps", toStringOrNull(b.partialDeps()))
+ .column("writes", toStringOrNull(b.writes()))
+ .column("result", toStringOrNull(b.result()));
}
return ds.getPartition(partitionKey);
@@ -952,6 +960,98 @@ public class AccordDebugKeyspace extends VirtualKeyspace
}
}
+ /**
+ * Write-only virtual table for updating Accord command cleanup states, both in CommandStore and persist the change in Journal.
+ *
+ * Use nonnegative command_store_id to update a command on a specific store.
+ *
+ * Example queries:
+ *
+ * UPDATE system_virtual_schema.accord_debug_txn_update SET cleanup = 'TRUNCATE' WHERE txn_id = '[11,1751902116570000,146(KW),1]' AND command_store_id = 5;
+ *
+ */
+ // Had to be separate from the "regular" journal table since it does not have segment and position, and command store id is inferred
+ // TODO (required): add access control
+ public static final class TxnUpdateTable extends AbstractMutableVirtualTable implements AbstractVirtualTable.DataSet
+ {
+ private static Set<String> ALLOWED = new HashSet<>(Arrays.asList("ERASE_VESTIGIAL", "INVALIDATE"));
+ private TxnUpdateTable()
+ {
+ super(parse(VIRTUAL_ACCORD_DEBUG, TXN_UPDATE,
+ "Update Accord Command State",
+ "CREATE TABLE %s (\n" +
+ " txn_id text,\n" +
+ " command_store_id int,\n" +
+ " cleanup text,\n" +
+ " PRIMARY KEY (txn_id, command_store_id)" +
+ ')', UTF8Type.instance));
+ }
+
+ @Override
+ public DataSet data()
+ {
+ throw new UnsupportedOperationException("Patch Journal is a write-only table");
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return true;
+ }
+
+ @Override
+ public Partition getPartition(DecoratedKey partitionKey)
+ {
+ throw new UnsupportedOperationException("Patch Journal is a write-only table");
+ }
+
+ @Override
+ public Iterator<Partition> getPartitions(DataRange range)
+ {
+ throw new UnsupportedOperationException("Patch Journal is a write-only table");
+ }
+
+
+ @Override
+ protected void applyColumnUpdate(ColumnValues partitionKey, ColumnValues clusteringColumns, Optional<ColumnValue> columnValue)
+ {
+ TxnId txnId = TxnId.parse(partitionKey.value(0));
+ Invariants.require(columnValue.isPresent());
+ Cleanup cleanup;
+ switch ((String) columnValue.get().value())
+ {
+ default: throw new IllegalStateException(String.format("Unknown Cleanup %s. Allowed: %s", columnValue.get().value(), ALLOWED));
+ case "ERASE_VESTIGIAL": cleanup = Cleanup.VESTIGIAL; break;
+ case "INVALIDATE": cleanup = Cleanup.INVALIDATE; break;
+ }
+
+ int commandStoreId = clusteringColumns.value(0);
+ Invariants.require(commandStoreId >= 0);
+
+ AccordService accord = (AccordService) AccordService.instance();
+ AsyncChains.awaitUninterruptibly(accord.node()
+ .commandStores()
+ .forId(commandStoreId)
+ .submit(txnId, new Function<SafeCommandStore, Void>()
+ {
+ @Override
+ public Void apply(SafeCommandStore safeStore)
+ {
+ SafeCommand safeCommand = safeStore.unsafeGet(txnId);
+ Command command = safeCommand.current();
+ Command updated = Commands.purge(safeStore,
+ command,
+ command.participants(),
+ cleanup,
+ true);
+ safeCommand.update(safeStore, updated);
+ return null;
+ }
+ })
+ .beginAsResult());
+ }
+ }
+
public static class TxnBlockedByTable extends AbstractVirtualTable
{
enum Reason { Self, Txn, Key }
@@ -1111,4 +1211,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace
try { return TraceEventType.valueOf(LocalizeString.toUpperCaseLocalized(input, Locale.ENGLISH)); }
catch (Throwable t) { throw invalidRequest("event_type must be one of %s; received %s", TraceEventType.values(), input); }
}
+
+ private static String toStringOrNull(Object o)
+ {
+ if (o == null)
+ return null;
+ return Objects.toString(o);
+ }
}
diff --git a/src/java/org/apache/cassandra/io/sstable/indexsummary/IndexSummary.java b/src/java/org/apache/cassandra/io/sstable/indexsummary/IndexSummary.java
index 39377cc01f..6173d8ddd7 100644
--- a/src/java/org/apache/cassandra/io/sstable/indexsummary/IndexSummary.java
+++ b/src/java/org/apache/cassandra/io/sstable/indexsummary/IndexSummary.java
@@ -56,7 +56,7 @@ import static org.apache.cassandra.io.sstable.Downsampling.BASE_SAMPLING_LEVEL;
* Layout of Memory for index summaries:
*
* There are two sections:
- * 1. A "header" containing the offset into `bytes` of entries in the summary summary data, consisting of
+ * 1. A "header" containing the offset into `bytes` of entries in the summary data, consisting of
* one four byte position for each entry in the summary. This allows us do simple math in getIndex()
* to find the position in the Memory to start reading the actual index summary entry.
* (This is necessary because keys can have different lengths.)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index dfb4636c4c..cd64259cd1 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -349,6 +349,17 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
journal.onDurable(pointer, onFlush);
}
+ public void patchCommand(int commandStoreId, TxnId txnId, Cleanup cleanup, @Nullable Runnable onFlush)
+ {
+ Builder change = new Builder(txnId);
+ change.maybeCleanup(false, cleanup);
+
+ JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
+ RecordPointer pointer = journal.asyncWrite(key, (out, userVersion) -> change.serialize(out, Version.fromVersion(configuration().userVersion())));
+ if (onFlush != null)
+ journal.onDurable(pointer, onFlush);
+ }
+
@Override
public Iterator<AccordTopologyUpdate.ImmutableTopoloyImage> replayTopologies()
{
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
index c7b3a77c4b..7df9e27d52 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
@@ -342,4 +342,4 @@ public class AccordJournalValueSerializers
into.update(AccordTopologyUpdate.RangesForEpochSerializer.instance.deserialize(in));
}
}
-}
\ No newline at end of file
+}
diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
index 56abfe10e8..039444d66e 100644
--- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
+++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
@@ -26,6 +26,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.function.BiPredicate;
+import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
@@ -146,7 +147,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
try
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
- var accord = accord();
+ AccordService accord = accord();
DatabaseDescriptor.getAccord().fetch_txn = "1s";
TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
@@ -210,7 +211,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
public void completedTxn() throws ExecutionException, InterruptedException
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
- var accord = accord();
+ AccordService accord = accord();
TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
AsyncChains.getBlocking(accord.node().coordinate(id, txn));
@@ -218,18 +219,19 @@ public class AccordDebugKeyspaceTest extends CQLTester
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied"));
- assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), "null"));
+ assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
}
@Test
public void inflight() throws ExecutionException, InterruptedException
{
+ ProtocolModifiers.Toggles.setPermitLocalExecution(false);
AccordMsgFilter filter = new AccordMsgFilter();
MessagingService.instance().outboundSink.add(filter);
try
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
- var accord = accord();
+ AccordService accord = accord();
TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
@@ -241,7 +243,6 @@ public class AccordDebugKeyspaceTest extends CQLTester
accord.node().coordinate(id, txn);
filter.preAccept.awaitThrowUncheckedOnInterrupt();
-
assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
@@ -251,6 +252,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
}
finally
{
+ filter.reset();
MessagingService.instance().outboundSink.remove(filter);
}
}
@@ -265,7 +267,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
try
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
- var accord = accord();
+ AccordService accord = accord();
TxnId first = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
@@ -301,6 +303,73 @@ public class AccordDebugKeyspaceTest extends CQLTester
}
finally
{
+ filter.reset();
+ MessagingService.instance().outboundSink.remove(filter);
+ }
+ }
+
+ @Test
+ public void patchJournalVestigialTest()
+ {
+ testPatchJournal("ERASE_VESTIGIAL", "Vestigial");
+ }
+
+ @Test
+ public void patchJournalInvalidateTest()
+ {
+ testPatchJournal("INVALIDATE", "Invalidated");
+ }
+
+ @Test
+ public void patchJournalTruncateTest()
+ {
+ try
+ {
+ testPatchJournal("ERASE", "Erased");
+ Assert.fail("Should have thrown");
+ }
+ catch (Throwable t)
+ {
+ Assert.assertTrue(t.getMessage().contains("Unknown"));
+ }
+ }
+
+ private void testPatchJournal(String cleanupAction, String expectedStatus)
+ {
+ AccordMsgFilter filter = new AccordMsgFilter();
+ MessagingService.instance().outboundSink.add(filter);
+ try
+ {
+ String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
+ String insertTxn = String.format("BEGIN TRANSACTION\n" +
+ " INSERT INTO %s.%s (k, c, v) VALUES (?, ?, ?);\n" +
+ "COMMIT TRANSACTION",
+ KEYSPACE,
+ tableName);
+ AccordService accord = accord();
+ TxnId id = accord.node().nextTxnId(Txn.Kind.Write, Routable.Domain.Key);
+ accord.node().coordinate(id, createTxn(insertTxn, 0, 0, 0));
+
+ filter.preAccept.awaitThrowUncheckedOnInterrupt();
+ String QUERY_JOURNAL = String.format("SELECT txn_id, save_status, command_store_id FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);
+ String QUERY_TXN = String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN);
+
+ UntypedResultSet rs = execute(QUERY_JOURNAL, id.toString());
+ assertRows(rs, row(id.toString(), "PreAccepted", anyNonNull()));
+
+ int commandStoreId = rs.one().getInt("command_store_id");
+ String PATCH_JOURNAL = String.format("UPDATE %s.%s SET cleanup = ? WHERE txn_id=? AND command_store_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_UPDATE);
+ execute(PATCH_JOURNAL, cleanupAction, id.toString(), commandStoreId);
+
+ assertRows(execute(QUERY_TXN, id.toString()),
+ row(id.toString(), expectedStatus));
+ assertRows(execute(QUERY_JOURNAL, id.toString()),
+ row(id.toString(), "PreAccepted", commandStoreId),
+ row(id.toString(), expectedStatus, commandStoreId));
+ }
+ finally
+ {
+ filter.reset();
MessagingService.instance().outboundSink.remove(filter);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org