First dumb of logic to get route index running on the journal

.gitmodules

@@ -1,4 +1,4 @@
 [submodule "modules/accord"]
 	path = modules/accord
-	url = https://github.com/apache/cassandra-accord.git
-	branch = trunk
+	url = https://github.com/dcapwell/cassandra-accord.git
+	branch = CASSANDRA-20144

src/java/org/apache/cassandra/db/compaction/CompactionIterator.java

@@ -35,16 +35,10 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import accord.api.Agent;
 import accord.local.Cleanup;
 import accord.local.CommandStores;
-import accord.local.CommandStores.RangesForEpoch;
 import accord.local.DurableBefore;
 import accord.local.RedundantBefore;
-import accord.local.StoreParticipants;
-import accord.primitives.SaveStatus;
-import accord.primitives.Status;
-import accord.primitives.Status.Durability;
 import accord.primitives.Timestamp;
 import accord.primitives.TxnId;
 import accord.utils.Invariants;
@@ -67,7 +61,6 @@
 import org.apache.cassandra.db.partitions.PurgeFunction;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
-import org.apache.cassandra.db.rows.BTreeRow;
 import org.apache.cassandra.db.rows.Cell;
 import org.apache.cassandra.db.rows.RangeTombstoneBoundMarker;
 import org.apache.cassandra.db.rows.RangeTombstoneMarker;
@@ -96,8 +89,7 @@
 import org.apache.cassandra.service.accord.AccordJournalValueSerializers;
 import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer;
 import org.apache.cassandra.service.accord.AccordKeyspace;
-import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
-import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
+
 import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
 import org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyRows;
 import org.apache.cassandra.service.accord.AccordService;
@@ -111,23 +103,15 @@
 import org.apache.cassandra.utils.TimeUUID;
 
 import static accord.local.Cleanup.ERASE;
-import static accord.local.Cleanup.TRUNCATE_WITH_OUTCOME;
-import static accord.local.Cleanup.shouldCleanupPartial;
 import static com.google.common.base.Preconditions.checkState;
 import static java.util.concurrent.TimeUnit.MICROSECONDS;
 import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
 import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
-import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.expungePartial;
-import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.saveStatusOnly;
-import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.truncatedApply;
 import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeysAccessor;
 import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_executed_micros;
 import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_executed_timestamp;
 import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_write_timestamp;
 import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyRows.truncateTimestampsForKeyRow;
-import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeDurabilityOrNull;
-import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeParticipantsOrNull;
-import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeSaveStatusOrNull;
 import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeTimestampOrNull;
 
 /**
@@ -239,8 +223,6 @@ private Transformation<UnfilteredRowIterator> purger(ColumnFamilyStore cfs, Supp
         if (!requiresAccordSpecificPurger(cfs))
             return new Purger(controller, nowInSec);
 
-        if (isAccordCommands(cfs))
-            return new AccordCommandsPurger(accordService);
         if (isAccordTimestampsForKey(cfs))
             return new AccordTimestampsForKeyPurger(accordService);
         if (isAccordJournal(cfs))
@@ -805,98 +787,6 @@ protected Row applyToRow(Row row)
         }
     }
 
-    class AccordCommandsPurger extends AbstractPurger
-    {
-        final Agent agent;
-        final Int2ObjectHashMap<RedundantBefore> redundantBefores;
-        final Int2ObjectHashMap<DurableBefore> durableBefores;
-        final Int2ObjectHashMap<RangesForEpoch> ranges;
-
-        int storeId;
-        TxnId txnId;
-
-        AccordCommandsPurger(Supplier<IAccordService> accordService)
-        {
-            IAccordService service = accordService.get();
-            IAccordService.CompactionInfo compactionInfo = service.getCompactionInfo();
-            this.agent = service.agent();
-            this.redundantBefores = compactionInfo.redundantBefores;
-            this.ranges = compactionInfo.ranges;
-            this.durableBefores = compactionInfo.durableBefores;
-        }
-
-        protected void beginPartition(UnfilteredRowIterator partition)
-        {
-            ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
-            storeId = CommandRows.getStoreId(partitionKeyComponents);
-            txnId = CommandRows.getTxnId(partitionKeyComponents);
-        }
-
-        @Override
-        protected Row applyToRow(Row row)
-        {
-            updateProgress();
-
-            RedundantBefore redundantBefore = redundantBefores.get(storeId);
-            DurableBefore durableBefore = durableBefores.get(storeId);
-            // TODO (expected): if the store has been retired, this should return null
-            if (redundantBefore == null)
-                return row;
-
-            Cell durabilityCell = row.getCell(CommandsColumns.durability);
-            Durability durability = deserializeDurabilityOrNull(durabilityCell);
-            Cell executeAtCell = row.getCell(CommandsColumns.execute_at);
-            Cell participantsCell = row.getCell(CommandsColumns.participants);
-            StoreParticipants participants = deserializeParticipantsOrNull(participantsCell);
-            Cell statusCell = row.getCell(CommandsColumns.status);
-            SaveStatus saveStatus = deserializeSaveStatusOrNull(statusCell);
-
-            if (saveStatus == null)
-                return row;
-
-            if (saveStatus.is(Status.Invalidated))
-                return saveStatusOnly(saveStatus, row, nowInSec);
-
-            Cleanup cleanup = shouldCleanupPartial(agent, txnId, saveStatus, durability, participants,
-                                                   redundantBefore, durableBefore);
-            switch (cleanup)
-            {
-                default: throw new AssertionError(String.format("Unexpected cleanup task: %s", cleanup));
-                case EXPUNGE:
-                    return null;
-
-                case EXPUNGE_PARTIAL:
-                    return expungePartial(row, durabilityCell, executeAtCell, participantsCell);
-
-                case ERASE:
-                    // Emit a tombstone so if this is slicing the command and making it not possible to determine if it
-                    // can be truncated later it can still be dropped via the tombstone.
-                    // Eventually the tombstone can be dropped by `durableBefore.min(txnId) == Universal`
-                    // We can still encounter sliced command state just because compaction inputs are random
-                    return BTreeRow.emptyDeletedRow(row.clustering(), new Row.Deletion(DeletionTime.build(row.primaryKeyLivenessInfo().timestamp(), nowInSec), false));
-
-                case VESTIGIAL:
-                case INVALIDATE:
-                    return saveStatusOnly(cleanup.appliesIfNot, row, nowInSec);
-
-                case TRUNCATE_WITH_OUTCOME:
-                case TRUNCATE:
-                    return truncatedApply(cleanup.appliesIfNot, row, nowInSec, durability, durabilityCell, executeAtCell, participantsCell, cleanup == TRUNCATE_WITH_OUTCOME);
-
-                case NO:
-                    // TODO (required): when we port this to journal, make sure to expunge extra fields beyond those we need to retain
-                    return row;
-            }
-        }
-
-        @Override
-        protected Row applyToStatic(Row row)
-        {
-            checkState(row.isStatic() && row.isEmpty());
-            return row;
-        }
-    }
-
     class AccordTimestampsForKeyPurger extends AbstractPurger
     {
         final Int2ObjectHashMap<RedundantBefore> redundantBefores;
@@ -1225,8 +1115,7 @@ private static boolean isPaxos(ColumnFamilyStore cfs)
     private static boolean requiresAccordSpecificPurger(ColumnFamilyStore cfs)
     {
         return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) &&
-               ImmutableSet.of(AccordKeyspace.COMMANDS,
-                               AccordKeyspace.TIMESTAMPS_FOR_KEY,
+               ImmutableSet.of(AccordKeyspace.TIMESTAMPS_FOR_KEY,
                                AccordKeyspace.JOURNAL,
                                AccordKeyspace.COMMANDS_FOR_KEY)
                            .contains(cfs.getTableName());
@@ -1242,11 +1131,6 @@ private static boolean isAccordJournal(ColumnFamilyStore cfs)
         return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && cfs.name.startsWith(AccordKeyspace.JOURNAL);
     }
 
-    private static boolean isAccordCommands(ColumnFamilyStore cfs)
-    {
-        return isAccordTable(cfs, AccordKeyspace.COMMANDS);
-    }
-
     private static boolean isAccordTimestampsForKey(ColumnFamilyStore cfs)
     {
         return isAccordTable(cfs, AccordKeyspace.TIMESTAMPS_FOR_KEY);

src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java

@@ -214,7 +214,7 @@ public DataSet data()
             {
                 try (AccordExecutor.ExclusiveGlobalCaches cache = executor.lockCaches())
                 {
-                    addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS, cache.commands.statsSnapshot());
+                    addRow(ds, executor.executorId(), "commands", cache.commands.statsSnapshot());
                     addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS_FOR_KEY, cache.commandsForKey.statsSnapshot());
                     addRow(ds, executor.executorId(), AccordKeyspace.TIMESTAMPS_FOR_KEY, cache.timestampsForKey.statsSnapshot());
                 }

src/java/org/apache/cassandra/index/accord/IndexMetrics.java

@@ -40,7 +40,7 @@ public class IndexMetrics
     private final String indexName;
     public final Timer memtableIndexWriteLatency;
 
-    public IndexMetrics(RouteIndex index)
+    public IndexMetrics(ParticipantsJournalIndex index)
     {
         this.ks = index.baseCfs().getKeyspaceName();
         this.table = index.baseCfs().name;