Cassandra 19355: Rewrote CommandsForRanges so it doesn't store fully in memory

.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-19355_on_refactor-deps

src/java/org/apache/cassandra/config/AccordSpec.java

@@ -44,4 +44,9 @@ public class AccordSpec
     public DurationSpec.IntMillisecondsBound range_barrier_timeout = new DurationSpec.IntMillisecondsBound("2m");
 
     public volatile DurationSpec fast_path_update_delay = new DurationSpec.IntSecondsBound(5);
+
+    public volatile DurationSpec schedule_durability_frequency = new DurationSpec.IntSecondsBound(15);
+    public volatile DurationSpec durability_txnid_lag = new DurationSpec.IntSecondsBound(5);
+    public volatile DurationSpec shard_durability_cycle = new DurationSpec.IntMinutesBound(2);
+    public volatile DurationSpec global_durability_cycle = new DurationSpec.IntMinutesBound(10);
 }

src/java/org/apache/cassandra/config/CassandraRelevantProperties.java

@@ -541,7 +541,7 @@ public enum CassandraRelevantProperties
     TCM_UNSAFE_BOOT_WITH_CLUSTERMETADATA("cassandra.unsafe_boot_with_clustermetadata", null),
     TCM_USE_ATOMIC_LONG_PROCESSOR("cassandra.test.use_atomic_long_processor", "false"),
     TCM_USE_NO_OP_REPLICATOR("cassandra.test.use_no_op_replicator", "false"),
-
+    TEST_ACCORD_STORE_THREAD_CHECKS_ENABLED("cassandra.test.accord.store.thread_checks_enabled", "true"),
     TEST_BBFAILHELPER_ENABLED("test.bbfailhelper.enabled"),
     TEST_BLOB_SHARED_SEED("cassandra.test.blob.shared.seed"),
     TEST_BYTEMAN_TRANSFORMATIONS_DEBUG("cassandra.test.byteman.transformations.debug"),

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -4910,9 +4910,49 @@ public static long getAccordFastPathUpdateDelayMillis()
         return conf.accord.fast_path_update_delay.to(TimeUnit.MILLISECONDS);
     }
 
-    public static void setAccordFastPathUpdateDelayMillis(long millis)
+    public static void setAccordFastPathUpdateDelaySeconds(long seconds)
     {
-        conf.accord.fast_path_update_delay = new DurationSpec.IntMillisecondsBound(millis);
+        conf.accord.fast_path_update_delay = new DurationSpec.IntSecondsBound(seconds);
+    }
+
+    public static long getAccordScheduleDurabilityFrequency(TimeUnit unit)
+    {
+        return conf.accord.schedule_durability_frequency.to(unit);
+    }
+
+    public static void setAccordScheduleDurabilityFrequencySeconds(long seconds)
+    {
+        conf.accord.schedule_durability_frequency = new DurationSpec.IntSecondsBound(seconds);
+    }
+
+    public static long getAccordScheduleDurabilityTxnIdLag(TimeUnit unit)
+    {
+        return conf.accord.durability_txnid_lag.to(unit);
+    }
+
+    public static void setAccordScheduleDurabilityTxnIdLagSeconds(long seconds)
+    {
+        conf.accord.durability_txnid_lag = new DurationSpec.IntSecondsBound(seconds);
+    }
+
+    public static long getAccordGlobalDurabilityCycle(TimeUnit unit)
+    {
+        return conf.accord.global_durability_cycle.to(unit);
+    }
+
+    public static void setAccordGlobalDurabilityCycleSeconds(long seconds)
+    {
+        conf.accord.global_durability_cycle = new DurationSpec.IntSecondsBound(seconds);
+    }
+
+    public static long getAccordShardDurabilityCycle(TimeUnit unit)
+    {
+        return conf.accord.shard_durability_cycle.to(unit);
+    }
+
+    public static void setAccordShardDurabilityCycleSeconds(long seconds)
+    {
+        conf.accord.shard_durability_cycle = new DurationSpec.IntSecondsBound(seconds);
     }
 
     public static boolean getForceNewPreparedStatementBehaviour()

src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java

@@ -79,6 +79,10 @@
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.utils.FBUtilities;
 
+import static accord.primitives.Txn.Kind.EphemeralRead;
+import static accord.primitives.Txn.Kind.Read;
+import static org.apache.cassandra.config.Config.NonSerialWriteStrategy.accord;
+import static org.apache.cassandra.config.DatabaseDescriptor.getNonSerialWriteStrategy;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
 import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue;
@@ -321,7 +325,8 @@ public Txn createTxn(ClientState state, QueryOptions options)
             List<TxnNamedRead> reads = createNamedReads(options, state, ImmutableMap.of(), keySet::add);
             Keys txnKeys = toKeys(keySet);
             TxnRead read = createTxnRead(reads, txnKeys, null);
-            return new Txn.InMemory(txnKeys, read, TxnQuery.ALL);
+            Txn.Kind kind = txnKeys.size() == 1 && getNonSerialWriteStrategy() == accord ? EphemeralRead : Read;
+            return new Txn.InMemory(kind, txnKeys, read, TxnQuery.ALL, null);
         }
         else
         {

src/java/org/apache/cassandra/db/DiskBoundaryManager.java

@@ -194,7 +194,7 @@ private static List<PartitionPosition> getDiskBoundaries(RangesAtEndpoint replic
         List<PartitionPosition> diskBoundaries = new ArrayList<>();
         for (int i = 0; i < boundaries.size() - 1; i++)
             diskBoundaries.add(boundaries.get(i).maxKeyBound());
-        diskBoundaries.add(partitioner.getMaximumToken().maxKeyBound());
+        diskBoundaries.add(partitioner.getMaximumTokenForSplitting().maxKeyBound());
         return diskBoundaries;
     }
 }

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

@@ -98,10 +98,10 @@
 import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
 import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.maybeDropTruncatedCommandColumns;
 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.TimestampsForKeyColumns.max_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.deserializeRouteOrNull;
@@ -221,11 +221,8 @@ private Transformation<UnfilteredRowIterator> purger(ColumnFamilyStore cfs, Supp
         if (isAccordTimestampsForKey(cfs))
             return new AccordTimestampsForKeyPurger(accordService);
 
-        if (isAccordDepsCommandsForKey(cfs))
-            return new AccordCommandsForKeyPurger(AccordKeyspace.DepsCommandsForKeysAccessor, accordService);
-
-        if (isAccordAllCommandsForKey(cfs))
-            return new AccordCommandsForKeyPurger(AccordKeyspace.AllCommandsForKeysAccessor, accordService);
+        if (isAccordCommandsForKey(cfs))
+            return new AccordCommandsForKeyPurger(AccordKeyspace.CommandsForKeysAccessor, accordService);
 
         throw new IllegalArgumentException("Unhandled accord table: " + cfs.keyspace.getName() + '.' + cfs.name);
     }
@@ -936,22 +933,14 @@ protected Row applyToRow(Row row)
                 lastWriteCell = null;
             }
 
-            Cell<?> maxTimestampCell = row.getCell(max_timestamp);
-            Timestamp max_timestamp = deserializeTimestampOrNull(maxTimestampCell);
-            if (max_timestamp != null && max_timestamp.compareTo(redundantBeforeTxnId) < 0)
-            {
-                maxTimestampCell = null;
-            }
-
             // No need to emit a tombstone as earlier versions of the row will also be nulled out
             // when compacted later or loaded into a commands for key
             if (lastExecuteMicrosCell == null &&
                 lastExecuteCell == null &&
-                lastWriteCell == null &&
-                maxTimestampCell == null)
+                lastWriteCell == null)
                 return null;
 
-            return truncateTimestampsForKeyRow(nowInSec, row, lastExecuteMicrosCell, lastExecuteCell, lastWriteCell, maxTimestampCell);
+            return truncateTimestampsForKeyRow(nowInSec, row, lastExecuteMicrosCell, lastExecuteCell, lastWriteCell);
         }
 
         @Override
@@ -997,11 +986,10 @@ protected Row applyToRow(Row row)
                 return row;
 
             TxnId redundantBeforeTxnId = redundantBeforeEntry.shardRedundantBefore();
-            Timestamp timestamp = accessor.getTimestamp(row);
-            if (timestamp != null && timestamp.compareTo(redundantBeforeTxnId) < 0)
-                return null;
+            if (redundantBeforeTxnId.equals(TxnId.NONE))
+                return row;
 
-            return row;
+            return CommandsForKeysAccessor.withoutRedundantCommands(partitionKey, row, redundantBeforeTxnId);
         }
 
         @Override
@@ -1057,8 +1045,7 @@ private static boolean requiresAccordSpecificPurger(ColumnFamilyStore cfs)
         return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) &&
                ImmutableSet.of(AccordKeyspace.COMMANDS,
                                AccordKeyspace.TIMESTAMPS_FOR_KEY,
-                               AccordKeyspace.DEPS_COMMANDS_FOR_KEY,
-                               AccordKeyspace.ALL_COMMANDS_FOR_KEY)
+                               AccordKeyspace.COMMANDS_FOR_KEY)
                            .contains(cfs.getTableName());
     }
 
@@ -1077,13 +1064,8 @@ private static boolean isAccordTimestampsForKey(ColumnFamilyStore cfs)
         return isAccordTable(cfs, AccordKeyspace.TIMESTAMPS_FOR_KEY);
     }
 
-    private static boolean isAccordDepsCommandsForKey(ColumnFamilyStore cfs)
-    {
-        return isAccordTable(cfs, AccordKeyspace.DEPS_COMMANDS_FOR_KEY);
-    }
-
-    private static boolean isAccordAllCommandsForKey(ColumnFamilyStore cfs)
+    private static boolean isAccordCommandsForKey(ColumnFamilyStore cfs)
     {
-        return isAccordTable(cfs, AccordKeyspace.ALL_COMMANDS_FOR_KEY);
+        return isAccordTable(cfs, AccordKeyspace.COMMANDS_FOR_KEY);
     }
 }

src/java/org/apache/cassandra/dht/ByteOrderedPartitioner.java

@@ -144,6 +144,12 @@ public double size(Token next)
 
         @Override
         public Token nextValidToken()
+        {
+            throw new UnsupportedOperationException(String.format("Token type %s does not support token allocation.",
+                                                                  getClass().getSimpleName()));
+        }
+
+        public Token increaseSlightly()
         {
             // find first byte we can increment
             int i = token.length - 1;
@@ -180,8 +186,6 @@ public Token decreaseSlightly()
             if (i == -1)
             {
                 byte[] newToken = Arrays.copyOf(token, token.length - 1);
-                if (newToken.length > 0)
-                    newToken[newToken.length - 1] = (byte)-1;
                 return new BytesToken(newToken);
             }
 

src/java/org/apache/cassandra/dht/IPartitioner.java

@@ -83,9 +83,9 @@ static void validate(AbstractBounds<?> bounds)
      * The biggest token for this partitioner, unlike getMinimumToken, this token is actually used and users wanting to
      * include all tokens need to do getMaximumToken().maxKeyBound()
      *
-     * Not implemented for the ordered partitioners
+     * THIS IS NOT SAFE FOR PURPOSES BESIDES SPLITTING/BALANCING
      */
-    default Token getMaximumToken()
+    default Token getMaximumTokenForSplitting()
     {
         throw new UnsupportedOperationException("If you are using a splitting partitioner, getMaximumToken has to be implemented");
     }

src/java/org/apache/cassandra/dht/Murmur3Partitioner.java

@@ -463,7 +463,7 @@ public AbstractType<?> getTokenValidator()
         return LongType.instance;
     }
 
-    public Token getMaximumToken()
+    public Token getMaximumTokenForSplitting()
     {
         return new LongToken(Long.MAX_VALUE);
     }

src/java/org/apache/cassandra/dht/OrderPreservingPartitioner.java

@@ -140,7 +140,7 @@ public StringToken getMinimumToken()
         return MINIMUM;
     }
 
-    public StringToken getMaximumToken()
+    public StringToken getMaximumTokenForSplitting()
     {
         return MAXIMUM;
     }

src/java/org/apache/cassandra/dht/RandomPartitioner.java

@@ -369,7 +369,7 @@ public Map<Token, Float> describeOwnership(List<Token> sortedTokens)
         return ownerships;
     }
 
-    public Token getMaximumToken()
+    public Token getMaximumTokenForSplitting()
     {
         return new BigIntegerToken(MAXIMUM);
     }

src/java/org/apache/cassandra/dht/Splitter.java

@@ -50,7 +50,7 @@ protected BigInteger tokensInRange(Range<Token> range)
     {
         //full range case
         if (range.left.equals(range.right))
-            return tokensInRange(new Range(partitioner.getMinimumToken(), partitioner.getMaximumToken()));
+            return tokensInRange(new Range(partitioner.getMinimumToken(), partitioner.getMaximumTokenForSplitting()));
 
         BigInteger totalTokens = BigInteger.ZERO;
         for (Range<Token> unwrapped : range.unwrap())
@@ -95,7 +95,7 @@ public double positionInRange(Token token, Range<Token> range)
     {
         //full range case
         if (range.left.equals(range.right))
-            return positionInRange(token, new Range(partitioner.getMinimumToken(), partitioner.getMaximumToken()));
+            return positionInRange(token, new Range(partitioner.getMinimumToken(), partitioner.getMaximumTokenForSplitting()));
 
         // leftmost token means we are on position 0.0
         if (token.equals(range.left))
@@ -115,7 +115,7 @@ public double positionInRange(Token token, Range<Token> range)
     public List<Token> splitOwnedRanges(int parts, List<WeightedRange> weightedRanges, boolean dontSplitRanges)
     {
         if (weightedRanges.isEmpty() || parts == 1)
-            return Collections.singletonList(partitioner.getMaximumToken());
+            return Collections.singletonList(partitioner.getMaximumTokenForSplitting());
 
         BigInteger totalTokens = BigInteger.ZERO;
         for (WeightedRange weightedRange : weightedRanges)
@@ -126,7 +126,7 @@ public List<Token> splitOwnedRanges(int parts, List<WeightedRange> weightedRange
         BigInteger perPart = totalTokens.divide(BigInteger.valueOf(parts));
         // the range owned is so tiny we can't split it:
         if (perPart.equals(BigInteger.ZERO))
-            return Collections.singletonList(partitioner.getMaximumToken());
+            return Collections.singletonList(partitioner.getMaximumTokenForSplitting());
 
         if (dontSplitRanges)
             return splitOwnedRangesNoPartialRanges(weightedRanges, perPart, parts);
@@ -155,7 +155,7 @@ else if (partsLeft == 1)
             }
             sum = sum.add(currentRangeWidth);
         }
-        boundaries.set(boundaries.size() - 1, partitioner.getMaximumToken());
+        boundaries.set(boundaries.size() - 1, partitioner.getMaximumTokenForSplitting());
 
         assert boundaries.size() == parts : boundaries.size() + "!=" + parts + " " + boundaries + ":" + weightedRanges;
         return boundaries;
@@ -192,7 +192,7 @@ private List<Token> splitOwnedRangesNoPartialRanges(List<WeightedRange> weighted
             }
             i++;
         }
-        boundaries.add(partitioner.getMaximumToken());
+        boundaries.add(partitioner.getMaximumTokenForSplitting());
         return boundaries;
     }
 
@@ -202,7 +202,7 @@ private List<Token> splitOwnedRangesNoPartialRanges(List<WeightedRange> weighted
      */
     private Token token(Token t)
     {
-        return t.equals(partitioner.getMinimumToken()) ? partitioner.getMaximumToken() : t;
+        return t.equals(partitioner.getMinimumToken()) ? partitioner.getMaximumTokenForSplitting() : t;
     }
 
     /**

src/java/org/apache/cassandra/dht/Token.java

@@ -249,7 +249,12 @@ public long getLongValue()
     abstract public Token nextValidToken();
 
     /**
-     * Returns a token that is slightly less than this.
+     * Returns a token that is slightly more than this. This is NOT guaranteed to be the directly following token.
+     */
+    public Token increaseSlightly() { return nextValidToken(); }
+
+    /**
+     * Returns a token that is slightly less than this. This is NOT guaranteed to be the directly preceding token.
      */
     abstract public Token decreaseSlightly();
 

src/java/org/apache/cassandra/index/sai/QueryContext.java

@@ -37,7 +37,7 @@ public class QueryContext
 {
     private static final boolean DISABLE_TIMEOUT = SAI_TEST_DISABLE_TIMEOUT.getBoolean();
 
-    private final ReadCommand readCommand;
+    public final ReadCommand readCommand;
     private final long queryStartTimeNanos;
 
     public final long executionQuotaNano;