When updating a multi cell collection element, if the update is rejected then the shared Row.Builder is not freed causing all future mutations to be rejected

patch by David Capwell; reviewed by Bernardo Botella Corbi, Caleb Rackliffe, Dmitry Konstantinov for CASSANDRA-21055

Author: dcapwell

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * When updating a multi cell collection element, if the update is rejected then the shared Row.Builder is not freed causing all future mutations to be rejected (CASSANDRA-21055)
  * Schema annotations escape validation on CREATE and ALTER DDL statements (CASSANDRA-21046)
  * Calculate once and cache the result of ModificationStatement#requiresRead as a perf optimization (CASSANDRA-21040)
  * Update system schema tables with new distributed keyspace on upgrade (CASSANDRA-20872)

src/java/org/apache/cassandra/cql3/UpdateParameters.java

@@ -62,9 +62,6 @@ public class UpdateParameters
     // Holds data for operations that require a read-before-write. Will be null otherwise.
     private final Map<DecoratedKey, Partition> prefetchedRows;
 
-    private Row.Builder staticBuilder;
-    private Row.Builder regularBuilder;
-
     // The builder currently in use. Will alias either staticBuilder or regularBuilder, which are themselves built lazily.
     private Row.Builder builder;
 
@@ -108,20 +105,8 @@ public <V> void newRow(Clustering<V> clustering) throws InvalidRequestException
                     throw new InvalidRequestException("Invalid empty or null value for column " + metadata.clusteringColumns().get(0).name);
             }
         }
-
-        if (clustering == Clustering.STATIC_CLUSTERING)
-        {
-            if (staticBuilder == null)
-                staticBuilder = BTreeRow.pooledUnsortedBuilder();
-            builder = staticBuilder;
-        }
-        else
-        {
-            if (regularBuilder == null)
-                regularBuilder = BTreeRow.pooledUnsortedBuilder();
-            builder = regularBuilder;
-        }
-
+        assert builder == null : "newRow called without building the previous row";
+        builder = BTreeRow.pooledUnsortedBuilder();
         builder.newRow(clustering);
     }
 

src/java/org/apache/cassandra/utils/caching/TinyThreadLocalPool.java

@@ -18,10 +18,12 @@
 
 package org.apache.cassandra.utils.caching;
 
+import accord.utils.Invariants;
 import io.netty.util.concurrent.FastThreadLocal;
 
 public class TinyThreadLocalPool<V> extends FastThreadLocal<TinyThreadLocalPool.TinyPool<V>>
 {
+    private static final boolean DEBUG = Invariants.debug();
     protected TinyPool<V> initialValue()
     {
         return new TinyPool<>();
@@ -46,10 +48,23 @@ public void offer(V value)
         }
         private void offerSafe(V value)
         {
+            if (DEBUG)
+                checkOfferSafe(value);
             if (val1 == null) val1 = value;
             else if (val2 == null) val2 = value;
             else if (val3 == null) val3 = value;
         }
+
+        private void checkOfferSafe(V value)
+        {
+            if (val1 == value)
+                throw new IllegalStateException("Double offer");
+            if (val2 == value)
+                throw new IllegalStateException("Double offer");
+            if (val3 == value)
+                throw new IllegalStateException("Double offer");
+        }
+
         public V poll()
         {
             Object result;

test/distributed/org/apache/cassandra/distributed/test/TestBaseImpl.java

@@ -94,6 +94,11 @@
 // checkstyle: suppress below 'blockSystemPropertyUsage'
 public class TestBaseImpl extends DistributedTestBase
 {
+    static
+    {
+        System.setProperty("accord.debug", "true"); // checkstyle: suppress nearby 'blockSystemPropertyUsage'
+    }
+
     private static final Logger logger = LoggerFactory.getLogger(TestBaseImpl.class);
 
     public static final Object[][] EMPTY_ROWS = new Object[0][];

test/distributed/org/apache/cassandra/distributed/test/cql3/AccordInteropMultiNodeTableWalkBase.java

@@ -117,6 +117,13 @@ protected void createTable(TableMetadata metadata)
             return super.command(rs, mutation, annotate);
         }
 
+        @Override
+        protected boolean allowListElementAccessForUpdateSet()
+        {
+            // See CASSANDRA-20828
+            return false;
+        }
+
         @Override
         protected boolean allowUsingTimestamp()
         {

test/distributed/org/apache/cassandra/distributed/test/cql3/CasMultiNodeTableWalkBase.java

@@ -75,6 +75,13 @@ private State(RandomSource rs, Cluster cluster)
             super(rs, cluster);
         }
 
+        @Override
+        protected boolean allowUsingTimestamp()
+        {
+            // Paxos doesn't allow USING TIMESTAMP
+            return false;
+        }
+
         @Override
         protected Gen<Mutation> toMutationGen(ASTGenerators.MutationGenBuilder mutationGenBuilder)
         {

test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java

@@ -447,6 +447,7 @@ public State(RandomSource rs, Cluster cluster)
             ASTGenerators.MutationGenBuilder mutationGenBuilder = new ASTGenerators.MutationGenBuilder(metadata)
                                                                   .withTxnSafe()
                                                                   .withColumnExpressions(e -> e.withOperators(Generators.fromGen(BOOLEAN_DISTRIBUTION.next(rs))))
+                                                                  .withListElementAccessForUpdateSet(allowListElementAccessForUpdateSet())
                                                                   .withIgnoreIssues(IGNORED_ISSUES);
 
             // Run the test with and without bound partitions

test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTokenConflictTest.java

@@ -386,6 +386,7 @@ class State extends CommonState
             this.mutationGen = toGen(new ASTGenerators.MutationGenBuilder(metadata)
                                      .withTxnSafe()
                                      .withPartitions(SourceDSL.arbitrary().pick(uniquePartitions))
+                                     .withListElementAccessForUpdateSet(allowListElementAccessForUpdateSet())
                                      .withIgnoreIssues(IGNORED_ISSUES)
                                      .build());
         }

test/distributed/org/apache/cassandra/distributed/test/cql3/StatefulASTBase.java

@@ -27,11 +27,13 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.annotation.Nullable;
 
 import com.google.common.collect.ImmutableList;
+import org.assertj.core.api.ThrowableAssert;
 import org.slf4j.Logger;
 
 import accord.utils.Gen;
@@ -105,6 +107,11 @@ public class StatefulASTBase extends TestBaseImpl
      */
     protected static boolean CQL_DEBUG_APPLY_OPERATOR = false;
 
+    /**
+     * Allows for overriding the CQL format logic, default is none (single line), but can create custom ones to help with the history
+     */
+    protected static Supplier<CQLFormatter> CQL_FORMATTER = () -> CQLFormatter.None.instance;
+
     protected static final Gen<Gen<Boolean>> BOOL_DISTRIBUTION = Gens.bools().mixedDistribution();
     protected static final Gen<Gen<Conditional.Where.Inequality>> LESS_THAN_DISTRO = Gens.mixedDistribution(Stream.of(Conditional.Where.Inequality.values())
                                                                                                                   .filter(i -> i == Conditional.Where.Inequality.LESS_THAN || i == Conditional.Where.Inequality.LESS_THAN_EQ)
@@ -196,7 +203,7 @@ protected <S extends BaseState> Property.StatefulSuccess<S, Void> onSuccess(Logg
     protected static <S extends BaseState> Property.Command<S, Void, ?> compactTable(RandomSource rs, S state)
     {
         return new Property.SimpleCommand<>("nodetool compact " + state.metadata.keyspace + ' ' + state.metadata.name, s2 -> {
-            state.cluster.forEach(i -> i.nodetoolResult("compact", s2.metadata.keyspace, s2.metadata.name).asserts().success());
+            s2.cluster.forEach(i -> i.nodetoolResult("compact", s2.metadata.keyspace, s2.metadata.name).asserts().success());
             s2.compact();
         });
     }
@@ -475,6 +482,12 @@ private String createKeyspaceCQL(String ks)
             return command(rs, select, null);
         }
 
+        protected boolean allowListElementAccessForUpdateSet()
+        {
+            // this requires a read at the same CL, but the model is global level and not per-node level, so can't handle
+            return mutationCl() != ConsistencyLevel.NODE_LOCAL;
+        }
+
         protected boolean allowRepair()
         {
             return false;
@@ -586,18 +599,24 @@ protected ConsistencyLevel mutationCl()
         {
             var inst = selectInstance(rs);
             String postfix = "on " + inst;
-            if (mutation.isCas())
-            {
+            @Nullable
+            Consumer<ThrowableAssert.ThrowingCallable> shouldRaiseThrowable = model.shouldReject(mutation);
+            if (shouldRaiseThrowable != null)
+                postfix += ", should reject";
+            else if (mutation.isCas())
                 postfix += ", would apply " + model.shouldApply(mutation);
-                // CAS doesn't allow timestamps
-                mutation = mutation.withoutTimestamp();
-            }
+
             if (annotate == null) annotate = postfix;
             else                  annotate += ", " + postfix;
-            Mutation finalMutation = mutation;
             return new Property.SimpleCommand<>(humanReadable(mutation, annotate), s -> {
-                var result = s.executeQuery(inst, Integer.MAX_VALUE, s.mutationCl(), finalMutation);
-                s.model.updateAndValidate(result, finalMutation);
+                if (shouldRaiseThrowable != null)
+                {
+                    shouldRaiseThrowable.accept(() -> s.executeQuery(inst, Integer.MAX_VALUE, s.mutationCl(), mutation));
+                    s.mutation();
+                    return;
+                }
+                var result = s.executeQuery(inst, Integer.MAX_VALUE, s.mutationCl(), mutation);
+                s.model.updateAndValidate(result, mutation);
                 s.mutation();
             });
         }
@@ -611,15 +630,26 @@ protected ConsistencyLevel mutationCl()
         {
             var inst = selectInstance(rs);
             String postfix = "on " + inst;
-            if (model.isConditional(txn))
+            @Nullable
+            Consumer<ThrowableAssert.ThrowingCallable> shouldRaiseThrowable = model.shouldReject(txn);
+            if (shouldRaiseThrowable != null)
+                postfix += ", should reject";
+            else if (model.isConditional(txn))
                 postfix += ", would apply " + model.shouldApply(txn);
             if (annotate == null) annotate = postfix;
             else annotate += ", " + postfix;
 
             return new Property.SimpleCommand<>(humanReadable(txn, annotate), s -> {
                 boolean hasMutation = txn.ifBlock.isPresent() || !txn.mutations.isEmpty();
                 ConsistencyLevel cl = hasMutation ? s.mutationCl() : s.selectCl();
-                s.model.updateAndValidate(s.executeQuery(inst, Integer.MAX_VALUE, cl, txn), txn);
+                if (shouldRaiseThrowable != null)
+                {
+                    shouldRaiseThrowable.accept(() -> s.executeQuery(inst, Integer.MAX_VALUE, cl, txn));
+                }
+                else
+                {
+                    s.model.updateAndValidate(s.executeQuery(inst, Integer.MAX_VALUE, cl, txn), txn);
+                }
                 if (hasMutation)
                     s.mutation();
             });
@@ -706,7 +736,7 @@ protected String humanReadable(Statement stmt, @Nullable String annotate)
         {
             // With UTF-8 some chars can cause printing issues leading to error messages that don't reproduce the original issue.
             // To avoid this problem, always escape the CQL so nothing gets lost
-            String cql = StringUtils.escapeControlChars(stmt.visit(debug).toCQL(CQLFormatter.None.instance));
+            String cql = StringUtils.escapeControlChars(stmt.visit(debug).toCQL(CQL_FORMATTER.get()));
             if (annotate != null)
                 cql += " -- " + annotate;
             return cql;

test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java

@@ -148,7 +148,11 @@ private static Spec createSchemaSpec(RandomSource rs, Cluster cluster)
     private static CommandGen<Spec> cqlOperations(Spec spec)
     {
         Gen<Statement> select = (Gen<Statement>) (Gen<?>) fromQT(new ASTGenerators.SelectGenBuilder(spec.metadata).withLimit1().build());
-        Gen<Statement> mutation = (Gen<Statement>) (Gen<?>) fromQT(new ASTGenerators.MutationGenBuilder(spec.metadata).withTxnSafe().disallowUpdateMultiplePartitionKeys().build());
+        Gen<Statement> mutation = (Gen<Statement>) (Gen<?>) fromQT(new ASTGenerators.MutationGenBuilder(spec.metadata)
+                                                                   .withTxnSafe()
+                                                                   .disallowUpdateMultiplePartitionKeys() //TODO (coverage): this is something Accord should support, so should remove and make sure accord is updated
+                                                                   .disallowListElementAccessForUpdateSet() //TODO (coverage): CASSANDRA-20828 found an issue with multi cell list type timestamp handling, so make sure accord doesn't hit this
+                                                                   .build());
         Gen<Statement> txn = (Gen<Statement>) (Gen<?>) fromQT(new ASTGenerators.TxnGenBuilder(spec.metadata).build());
         Map<Gen<Statement>, Integer> operations = new LinkedHashMap<>();
         operations.put(select, 1);