diff --git a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
index 2ba02ae769..755d479e92 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/cql3/SingleNodeTableWalkTest.java
@@ -66,10 +66,11 @@ import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ASTGenerators;
 import org.apache.cassandra.utils.AbstractTypeGenerators;
 import org.apache.cassandra.utils.AbstractTypeGenerators.TypeGenBuilder;
+import org.apache.cassandra.utils.AbstractTypeGenerators.TypeKind;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.CassandraGenerators.TableMetadataBuilder;
+import org.apache.cassandra.utils.Generators;
 import org.apache.cassandra.utils.ImmutableUniqueList;
-import org.quicktheories.generators.SourceDSL;
 
 import static accord.utils.Property.commands;
 import static accord.utils.Property.stateful;
@@ -78,6 +79,17 @@ import static org.apache.cassandra.utils.Generators.toGen;
 
 public class SingleNodeTableWalkTest extends StatefulASTBase
 {
+    private static final Gen<Gen<Boolean>> BOOLEAN_DISTRIBUTION = Gens.bools().mixedDistribution();
+    //TODO (coverage): COMPOSITE, DYNAMIC_COMPOSITE
+    private static final Gen<Gen<TypeKind>> TYPE_KIND_DISTRIBUTION = Gens.mixedDistribution(TypeKind.PRIMITIVE,
+                                                                                            TypeKind.SET, TypeKind.LIST, TypeKind.MAP,
+                                                                                            TypeKind.TUPLE, TypeKind.UDT,
+                                                                                            TypeKind.VECTOR
+    );
+    private static final Gen<Gen<AbstractType<?>>> PRIMITIVE_DISTRIBUTION = Gens.mixedDistribution(AbstractTypeGenerators.knownPrimitiveTypes()
+                                                                                                                         .stream()
+                                                                                                                         .filter(t -> !AbstractTypeGenerators.isUnsafeEquality(t))
+                                                                                                                         .collect(Collectors.toList()));
     private static final Logger logger = LoggerFactory.getLogger(SingleNodeTableWalkTest.class);
 
     protected void preCheck(Cluster cluster, Property.StatefulBuilder builder)
@@ -88,10 +100,20 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
         // CQL_DEBUG_APPLY_OPERATOR = true;
     }
 
-    protected TypeGenBuilder supportedTypes()
+    protected TypeGenBuilder supportedTypes(RandomSource rs)
     {
-        return AbstractTypeGenerators.withoutUnsafeEquality(AbstractTypeGenerators.builder()
-                                                                                  .withTypeKinds(AbstractTypeGenerators.TypeKind.PRIMITIVE));
+        return AbstractTypeGenerators.builder()
+                                     .withTypeKinds(Generators.fromGen(TYPE_KIND_DISTRIBUTION.next(rs)))
+                                     .withPrimitives(Generators.fromGen(PRIMITIVE_DISTRIBUTION.next(rs)))
+                                     .withUserTypeFields(AbstractTypeGenerators.UserTypeFieldsGen.simpleNames())
+                                     .withMaxDepth(1);
+    }
+
+    protected TypeGenBuilder supportedPrimaryColumnTypes(RandomSource rs)
+    {
+        return AbstractTypeGenerators.builder()
+                                     .withTypeKinds(TypeKind.PRIMITIVE)
+                                     .withPrimitives(Generators.fromGen(PRIMITIVE_DISTRIBUTION.next(rs)));
     }
 
     protected List<CreateIndexDDL.Indexer> supportedIndexers()
@@ -206,7 +228,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
             builder.value(pk, key.bufferAt(pks.indexOf(pk)));
 
 
-        List<Symbol> searchableColumns = state.nonPartitionColumns;
+        List<Symbol> searchableColumns = state.searchableNonPartitionColumns;
         Symbol symbol = rs.pick(searchableColumns);
 
         TreeMap<ByteBuffer, List<BytesPartitionState.PrimaryKey>> universe = state.model.index(ref, symbol);
@@ -363,7 +385,8 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
                                        .withCompression())
                      .withKeyspaceName(ks).withTableName("tbl")
                      .withSimpleColumnNames()
-                     .withDefaultTypeGen(supportedTypes())
+                     .withDefaultTypeGen(supportedTypes(rs))
+                     .withPrimaryColumnTypeGen(supportedPrimaryColumnTypes(rs))
                      .withPartitioner(Murmur3Partitioner.instance)
                      .build())
                .next(rs);
@@ -393,7 +416,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
     {
         protected final LinkedHashMap<Symbol, IndexedColumn> indexes;
         private final Gen<Mutation> mutationGen;
-        private final List<Symbol> nonPartitionColumns;
+        private final List<Symbol> searchableNonPartitionColumns;
         private final List<Symbol> searchableColumns;
         private final List<Symbol> nonPkIndexedColumns;
 
@@ -424,7 +447,8 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
                                                                   .withoutTransaction()
                                                                   .withoutTtl()
                                                                   .withoutTimestamp()
-                                                                  .withPartitions(SourceDSL.arbitrary().pick(uniquePartitions));
+                                                                  .withPartitions(Generators.fromGen(Gens.mixedDistribution(uniquePartitions).next(rs)))
+                                                                  .withColumnExpressions(e -> e.withOperators(Generators.fromGen(BOOLEAN_DISTRIBUTION.next(rs))));
             if (IGNORED_ISSUES.contains(KnownIssue.SAI_EMPTY_TYPE))
             {
                 model.factory.regularAndStaticColumns.stream()
@@ -438,16 +462,30 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
             }
             this.mutationGen = toGen(mutationGenBuilder.build());
 
-            nonPartitionColumns = ImmutableList.<Symbol>builder()
-                                               .addAll(model.factory.clusteringColumns)
-                                               .addAll(model.factory.staticColumns)
-                                               .addAll(model.factory.regularColumns)
-                                               .build();
+            var nonPartitionColumns = ImmutableList.<Symbol>builder()
+                                                   .addAll(model.factory.clusteringColumns)
+                                                   .addAll(model.factory.staticColumns)
+                                                   .addAll(model.factory.regularColumns)
+                                                   .build();
+            searchableNonPartitionColumns = nonPartitionColumns.stream()
+                                                               .filter(this::isSearchable)
+                                                               .collect(Collectors.toList());
             nonPkIndexedColumns = nonPartitionColumns.stream()
                                                      .filter(indexes::containsKey)
                                                      .collect(Collectors.toList());
 
-            searchableColumns = metadata.partitionKeyColumns().size() > 1 ?  model.factory.selectionOrder : nonPartitionColumns;
+            searchableColumns = (metadata.partitionKeyColumns().size() > 1 ? model.factory.selectionOrder : nonPartitionColumns)
+                                .stream()
+                                .filter(this::isSearchable)
+                                .collect(Collectors.toList());
+        }
+
+        private boolean isSearchable(Symbol symbol)
+        {
+            // See org.apache.cassandra.cql3.Operator.validateFor
+            // multi cell collections can only be searched if you search their elements, not the collection as a whole
+            //TODO (coverage): can you query for UDT fields?  its a single cell so you "should"?
+            return !(symbol.type().isMultiCell() && (symbol.type().isCollection() || symbol.type().isUDT()));
         }
 
         @Override
@@ -523,6 +561,8 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
             List<Symbol> allowedColumns = searchableColumns;
             if (hasMultiNodeMultiColumnAllowFilteringWithLocalWritesIssue())
                 allowedColumns = nonPkIndexedColumns;
+            if (IGNORED_ISSUES.contains(KnownIssue.SAI_AND_VECTOR_COLUMNS) && !indexes.isEmpty())
+                allowedColumns = allowedColumns.stream().filter(s -> !s.type().isVector()).collect(Collectors.toList());
             return allowedColumns;
         }
 
@@ -533,7 +573,7 @@ public class SingleNodeTableWalkTest extends StatefulASTBase
 
         public boolean allowPartitionQuery()
         {
-            return !(model.isEmpty() || nonPartitionColumns.isEmpty());
+            return !(model.isEmpty() || searchableNonPartitionColumns.isEmpty());
         }
 
         @Override
diff --git a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java
index 4c0a5f5cf5..d2fbb6edcc 100644
--- a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java
+++ b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModel.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.harry.model;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.BitSet;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -43,13 +44,16 @@ import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
 import accord.utils.Invariants;
+import org.apache.cassandra.cql3.ast.AssignmentOperator;
 import org.apache.cassandra.cql3.ast.Conditional;
 import org.apache.cassandra.cql3.ast.Conditional.Where.Inequality;
 import org.apache.cassandra.cql3.ast.Element;
 import org.apache.cassandra.cql3.ast.Expression;
 import org.apache.cassandra.cql3.ast.ExpressionEvaluator;
 import org.apache.cassandra.cql3.ast.FunctionCall;
+import org.apache.cassandra.cql3.ast.Literal;
 import org.apache.cassandra.cql3.ast.Mutation;
+import org.apache.cassandra.cql3.ast.Operator;
 import org.apache.cassandra.cql3.ast.Select;
 import org.apache.cassandra.cql3.ast.StandardVisitors;
 import org.apache.cassandra.cql3.ast.Symbol;
@@ -244,7 +248,10 @@ public class ASTSingleTableModel
                 // static columns to add in.  If we are doing something like += to a row that doesn't exist, we still update statics...
                 Map<Symbol, ByteBuffer> write = new HashMap<>();
                 for (Symbol col : Sets.intersection(factory.staticColumns.asSet(), set.keySet()))
-                    write.put(col, eval(set.get(col)));
+                {
+                    ByteBuffer current = partition.staticRow().get(col);
+                    write.put(col, eval(col, current, set.get(col)));
+                }
                 partition.setStaticColumns(write);
             }
             // table has clustering but non are in the write, so only pk/static can be updated
@@ -254,7 +261,10 @@ public class ASTSingleTableModel
             {
                 Map<Symbol, ByteBuffer> write = new HashMap<>();
                 for (Symbol col : Sets.intersection(factory.regularColumns.asSet(), set.keySet()))
-                    write.put(col, eval(set.get(col)));
+                {
+                    ByteBuffer current = partition.get(cd, col);
+                    write.put(col, eval(col, current, set.get(col)));
+                }
 
                 partition.setColumns(cd, write, false);
             }
@@ -493,6 +503,45 @@ public class ASTSingleTableModel
             if (actual.isEmpty()) sb.append("No rows returned");
             else sb.append("Missing rows:\n").append(table(columns, missing));
         }
+        if (!unexpected.isEmpty() && unexpected.size() == missing.size())
+        {
+            // good chance a column differs
+            StringBuilder finalSb = sb;
+            Runnable runOnce = new Runnable()
+            {
+                boolean ran = false;
+                @Override
+                public void run()
+                {
+                    if (ran) return;
+                    finalSb.append("\nPossible column conflicts:");
+                    ran = true;
+                }
+            };
+            for (var e : missing)
+            {
+                Row smallest = null;
+                BitSet smallestDiff = null;
+                for (var a : unexpected)
+                {
+                    BitSet diff = e.diff(a);
+                    if (smallestDiff == null || diff.cardinality() < smallestDiff.cardinality())
+                    {
+                        smallest = a;
+                        smallestDiff = diff;
+                    }
+                }
+                // if every column differs then ignore
+                if (smallestDiff.cardinality() == e.values.length)
+                    continue;
+                runOnce.run();
+                sb.append("\n\tExpected: ").append(e);
+                sb.append("\n\tDiff (expected over actual):\n");
+                Row eSmall = e.select(smallestDiff);
+                Row aSmall = smallest.select(smallestDiff);
+                sb.append(table(eSmall.columns, Arrays.asList(eSmall, aSmall)));
+            }
+        }
         if (sb != null)
         {
             sb.append("\nExpected:\n").append(table(columns, expected));
@@ -731,7 +780,7 @@ public class ASTSingleTableModel
         for (Expression e : conditions)
         {
             ByteBuffer expected = eval(e);
-            if (expected.equals(value))
+            if (expected != null && expected.equals(value))
                 return true;
         }
         return false;
@@ -893,13 +942,41 @@ public class ASTSingleTableModel
         return current.stream().map(BufferClustering::new).collect(Collectors.toList());
     }
 
+    private static ByteBuffer eval(Symbol col, @Nullable ByteBuffer current, Expression e)
+    {
+        if (!(e instanceof AssignmentOperator)) return eval(e);
+        // multi cell collections have the property that they do update even if the current value is null
+        boolean isFancy = col.type().isCollection() && col.type().isMultiCell();
+        if (current == null && !isFancy) return null; // null + ? == null
+        var assignment = (AssignmentOperator) e;
+        if (isFancy && current == null)
+        {
+            return assignment.kind == AssignmentOperator.Kind.SUBTRACT
+                   // if it doesn't exist, then there is nothing to subtract
+                   ? null
+                   : eval(assignment.right);
+        }
+        switch (assignment.kind)
+        {
+            case ADD:
+                return eval(new Operator(Operator.Kind.ADD, new Literal(current, e.type()), assignment.right));
+            case SUBTRACT:
+                return eval(new Operator(Operator.Kind.SUBTRACT, new Literal(current, e.type()), assignment.right));
+            default:
+                throw new UnsupportedOperationException(assignment.kind + ": " + assignment.toCQL());
+        }
+    }
+
+    @Nullable
     private static ByteBuffer eval(Expression e)
     {
-        return ExpressionEvaluator.tryEvalEncoded(e).get();
+        return ExpressionEvaluator.evalEncoded(e);
     }
 
     private static class Row
     {
+        private static final Row EMPTY = new Row(ImmutableUniqueList.empty(), ByteBufferUtil.EMPTY_ARRAY);
+
         private final ImmutableUniqueList<Symbol> columns;
         private final ByteBuffer[] values;
 
@@ -907,6 +984,8 @@ public class ASTSingleTableModel
         {
             this.columns = columns;
             this.values = values;
+            if (columns.size() != values.length)
+                throw new IllegalArgumentException("Columns " + columns + " should have the same size as values, but had " + values.length);
         }
 
         public String asCQL(Symbol symbol)
@@ -914,7 +993,9 @@ public class ASTSingleTableModel
             int offset = columns.indexOf(symbol);
             assert offset >= 0;
             ByteBuffer b = values[offset];
-            return (b == null || ByteBufferUtil.EMPTY_BYTE_BUFFER.equals(b)) ? "null" : symbol.type().asCQL3Type().toCQLLiteral(b);
+            if (b == null) return "null";
+            if (ByteBufferUtil.EMPTY_BYTE_BUFFER.equals(b)) return "<empty>";
+            return symbol.type().asCQL3Type().toCQLLiteral(b);
         }
 
         public List<String> asCQL()
@@ -925,6 +1006,40 @@ public class ASTSingleTableModel
             return human;
         }
 
+        public BitSet diff(Row other)
+        {
+            if (!columns.equals(other.columns))
+                throw new UnsupportedOperationException("Columns do not match: expected " + columns + " but given " + other.columns);
+            int maxLength = Math.max(values.length, other.values.length);
+            int minLength = Math.min(values.length, other.values.length);
+            BitSet set = new BitSet(maxLength);
+            for (int i = 0; i < minLength; i++)
+            {
+                ByteBuffer a = values[i];
+                ByteBuffer b = other.values[i];
+                if (!Objects.equals(a, b))
+                    set.set(i);
+            }
+            for (int i = minLength; i < maxLength; i++)
+                set.set(i);
+            return set;
+        }
+
+        public Row select(BitSet selection)
+        {
+            if (selection.isEmpty()) return EMPTY;
+            var names = ImmutableUniqueList.<Symbol>builder(selection.cardinality());
+            ByteBuffer[] copy = new ByteBuffer[selection.cardinality()];
+            int offset = 0;
+            for (int i = 0; i < this.values.length; i++)
+            {
+                if (!selection.get(i)) continue;
+                names.add(this.columns.get(i));
+                copy[offset++] = this.values[i];
+            }
+            return new Row(names.build(), copy);
+        }
+
         @Override
         public boolean equals(Object o)
         {
diff --git a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModelTest.java b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModelTest.java
index 16f4d66819..a04425f827 100644
--- a/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModelTest.java
+++ b/test/harry/main/org/apache/cassandra/harry/model/ASTSingleTableModelTest.java
@@ -23,6 +23,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.TreeMap;
 import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
@@ -34,9 +36,11 @@ import java.util.stream.Stream;
 import com.google.common.collect.ImmutableList;
 import org.junit.Test;
 
+import org.apache.cassandra.cql3.ast.AssignmentOperator;
 import org.apache.cassandra.cql3.ast.Bind;
 import org.apache.cassandra.cql3.ast.Conditional.Where.Inequality;
 import org.apache.cassandra.cql3.ast.FunctionCall;
+import org.apache.cassandra.cql3.ast.Literal;
 import org.apache.cassandra.cql3.ast.Mutation;
 import org.apache.cassandra.cql3.ast.Select;
 import org.apache.cassandra.cql3.ast.Symbol;
@@ -48,7 +52,10 @@ import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.InetAddressType;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.marshal.LexicalUUIDType;
+import org.apache.cassandra.db.marshal.ListType;
+import org.apache.cassandra.db.marshal.MapType;
 import org.apache.cassandra.db.marshal.ReversedType;
+import org.apache.cassandra.db.marshal.SetType;
 import org.apache.cassandra.db.marshal.ShortType;
 import org.apache.cassandra.db.marshal.TimestampType;
 import org.apache.cassandra.dht.Murmur3Partitioner;
@@ -67,6 +74,9 @@ public class ASTSingleTableModelTest
 
     private static final EnumSet<Inequality> RANGE_INEQUALITY = EnumSet.of(Inequality.LESS_THAN, Inequality.LESS_THAN_EQ,
                                                                            Inequality.GREATER_THAN, Inequality.GREATER_THAN_EQ);
+    public static final ListType<Integer> LIST_INT = ListType.getInstance(Int32Type.instance, true);
+    public static final SetType<Integer> SET_INT = SetType.getInstance(Int32Type.instance, true);
+    public static final MapType<Integer, Integer> MAP_INT = MapType.getInstance(Int32Type.instance, Int32Type.instance, true);
 
     @Test
     public void singlePartition()
@@ -296,12 +306,12 @@ public class ASTSingleTableModelTest
         String pk0 = "'e44b:bdaf:aeb:f68b:1cff:ecbd:8b54:2295'";
         ByteBuffer pk0BB = InetAddressType.instance.asCQL3Type().fromCQLLiteral(pk0);
 
-        Short row1 = Short.valueOf((short) -14407);
+        Short row1 = (short) -14407;
         ByteBuffer row1BB = ShortType.instance.decompose(row1);
         String row1V1 = "0x00000000000049008a00000000000000";
         ByteBuffer row1V1BB = LexicalUUIDType.instance.asCQL3Type().fromCQLLiteral(row1V1);
 
-        Short row2 = Short.valueOf((short) ((short) 18175 - (short) 23847));
+        Short row2 = (short) ((short) 18175 - (short) 23847);
         ByteBuffer row2BB = ShortType.instance.decompose(row2);
         String row2V0 = "'1989-01-11T15:00:30.950Z'";
         ByteBuffer row2V0BB = TimestampType.instance.asCQL3Type().fromCQLLiteral(row2V0);
@@ -324,7 +334,7 @@ public class ASTSingleTableModelTest
                              .build());
 
         model.validate(new ByteBuffer[][]{ new ByteBuffer[]{ pk0BB, row1BB, null, row1V1BB } }, selectPk);
-        model.validate(new ByteBuffer[0][], selectColumn);
+        model.validate(EMPTY, selectColumn);
 
 
         model.update(Mutation.insert(metadata)
@@ -339,7 +349,7 @@ public class ASTSingleTableModelTest
         new ByteBuffer[]{ pk0BB, row1BB, null, row1V1BB },
         }, selectPk);
 
-        model.validate(new ByteBuffer[0][], selectColumn);
+        model.validate(EMPTY, selectColumn);
     }
 
     @Test
@@ -565,6 +575,116 @@ public class ASTSingleTableModelTest
                                     .build());
     }
 
+    @Test
+    public void assignmentOperator()
+    {
+        // not testing if assignment / operators are "corrrect", other tests can cover that
+        // the goal of this test is to test the plumbing and null handling within the model
+        TableMetadata metadata = defaultTable()
+                                 .addPartitionKeyColumn("pk", Int32Type.instance)
+                                 .addStaticColumn("s", Int32Type.instance)
+                                 .addRegularColumn("r", Int32Type.instance)
+                                 .build();
+        ASTSingleTableModel model = new ASTSingleTableModel(metadata);
+
+        // pk=0 doesn't exist, so s/r are null; so the operation should end with a null... this shouldn't create the partition
+        model.update(Mutation.update(metadata)
+                             .value("pk", 0)
+                             .set("s", subtract(42))
+                             .set("r", subtract(42))
+                             .build());
+
+        model.validate(EMPTY, Select.builder(metadata).build());
+
+        model.update(Mutation.insert(metadata).value("pk", 0).value("s", 40).value("r", 40).build());
+        model.update(Mutation.update(metadata)
+                             .value("pk", 0)
+                             .set("s", subtract(42))
+                             .set("r", subtract(42))
+                             .build());
+
+        model.validate(rows(row(metadata, 0, -2, -2)), Select.builder(metadata).build());
+    }
+
+    @Test
+    public void assignmentOperatorMultiCellCollections()
+    {
+        // not testing if assignment / operators are "corrrect", other tests can cover that
+        // the goal of this test is to test the plumbing and null handling within the model
+        TableMetadata metadata = defaultTable()
+                                 .addPartitionKeyColumn("pk", Int32Type.instance)
+                                 .addStaticColumn("s0", LIST_INT)
+                                 .addStaticColumn("r0", LIST_INT)
+                                 .addStaticColumn("s1", SET_INT)
+                                 .addStaticColumn("r1", SET_INT)
+                                 .addStaticColumn("s2", MAP_INT)
+                                 .addStaticColumn("r2", MAP_INT)
+                                 .build();
+        ASTSingleTableModel model = new ASTSingleTableModel(metadata);
+
+        // pk=0 doesn't exist, so s/r are null; but these are multi cell collections, so the update happens!
+        model.update(Mutation.update(metadata)
+                             .value("pk", 0)
+                             .set("s0", add(List.of(42)))
+                             .set("r0", add(List.of(42)))
+                             .set("s1", add(Set.of(42)))
+                             .set("r1", add(Set.of(42)))
+                             .set("s2", add(Map.of(42, 42)))
+                             .set("r2", add(Map.of(42, 42)))
+                             .build());
+
+        // Expected:
+        //pk | r0   | r1   | r2       | s0   | s1   | s2
+        //0  | [42] | {42} | {42: 42} | [42] | {42} | {42: 42}
+        model.validate(rows(row(metadata, 0, List.of(42), Set.of(42), Map.of(42, 42), List.of(42), Set.of(42), Map.of(42, 42))), Select.builder(metadata).build());
+
+        // add to existing
+        model.update(Mutation.update(metadata)
+                             .value("pk", 0)
+                             .set("s0", add(List.of(42)))
+                             .set("r0", add(List.of(42)))
+                             .set("s1", add(Set.of(0)))
+                             .set("r1", add(Set.of(0)))
+                             .set("s2", add(Map.of(42, 0)))
+                             .set("r2", add(Map.of(42, 0)))
+                             .build());
+        model.validate(rows(row(metadata, 0, List.of(42, 42), Set.of(0, 42), Map.of(42, 0), List.of(42, 42), Set.of(0, 42), Map.of(42, 0))), Select.builder(metadata).build());
+    }
+
+    private static ByteBuffer[][] rows(ByteBuffer[]... rows)
+    {
+        return rows;
+    }
+
+    private static ByteBuffer[] row(TableMetadata metadata, Object... values)
+    {
+        ByteBuffer[] row = new ByteBuffer[values.length];
+        var it = metadata.allColumnsInSelectOrder();
+        for (int i = 0; i < values.length && it.hasNext(); i++)
+            row[i] = it.next().type.decomposeUntyped(values[i]);
+        return row;
+    }
+
+    private static AssignmentOperator subtract(int value)
+    {
+        return new AssignmentOperator(AssignmentOperator.Kind.SUBTRACT, Literal.of(value));
+    }
+
+    private static AssignmentOperator add(List<Integer> value)
+    {
+        return new AssignmentOperator(AssignmentOperator.Kind.ADD, new Literal(value, LIST_INT));
+    }
+
+    private static AssignmentOperator add(Set<Integer> value)
+    {
+        return new AssignmentOperator(AssignmentOperator.Kind.ADD, new Literal(value, SET_INT));
+    }
+
+    private static AssignmentOperator add(Map<Integer, Integer> value)
+    {
+        return new AssignmentOperator(AssignmentOperator.Kind.ADD, new Literal(value, MAP_INT));
+    }
+
     private static TableMetadata.Builder defaultTable()
     {
         return TableMetadata.builder("ks", "tbl")
diff --git a/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java b/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java
index 6b8f61259d..a10524968a 100644
--- a/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java
+++ b/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java
@@ -125,6 +125,12 @@ public class BytesPartitionState
         long cd = factory.clusteringCache.deflate(clustering);
         long[] vds = toDescriptor(factory.regularColumns, values);
         state.writeRegular(cd, vds, MagicConstants.NO_TIMESTAMP, writePrimaryKeyLiveness);
+
+        // UDT's have the ability to "update" that triggers a delete; this allows creating an "empty" row.
+        // When an empty row exists without liveness info, then purge the row
+        var row = state.rows.get(cd);
+        if (row.isEmpty() && !row.hasPrimaryKeyLivenessInfo)
+            state.delete(cd, MagicConstants.NO_TIMESTAMP);
     }
 
     private long[] toDescriptor(ImmutableUniqueList<Symbol> positions, Map<Symbol, ByteBuffer> values)
@@ -135,8 +141,14 @@ public class BytesPartitionState
             Symbol column = positions.get(i);
             if (values.containsKey(column))
             {
-                long vd = factory.valueCache.deflate(new Value(column.type(), values.get(column)));
-                vds[i] = vd;
+                ByteBuffer value = values.get(column);
+                // user type is the only multi cell type that allows <empty> so this check should be fine; can expand if we find more cases
+                if (value == null || !value.hasRemaining() && (column.type().isUDT() && column.type().isMultiCell()))
+                {
+                    vds[i] = MagicConstants.NIL_DESCR;
+                    continue;
+                }
+                vds[i] = factory.valueCache.deflate(new Value(column.type(), value));
             }
             else
             {
@@ -197,6 +209,13 @@ public class BytesPartitionState
         return toRow(rowState);
     }
 
+    @Nullable
+    public ByteBuffer get(Clustering<ByteBuffer> clustering, Symbol column)
+    {
+        Row row = get(clustering);
+        return row == null ? null : row.get(column);
+    }
+
     private Row toRow(PartitionState.RowState rowState)
     {
         Clustering<ByteBuffer> clustering;
@@ -548,8 +567,8 @@ public class BytesPartitionState
 
         private Value(AbstractType<?> type, ByteBuffer value)
         {
-            this.type = type;
-            this.value = value;
+            this.type = Objects.requireNonNull(type);
+            this.value = Objects.requireNonNull(value);
         }
 
         @Override
diff --git a/test/unit/org/apache/cassandra/cql3/KnownIssue.java b/test/unit/org/apache/cassandra/cql3/KnownIssue.java
index a1924a91f9..b1c2c09d66 100644
--- a/test/unit/org/apache/cassandra/cql3/KnownIssue.java
+++ b/test/unit/org/apache/cassandra/cql3/KnownIssue.java
@@ -39,6 +39,8 @@ public enum KnownIssue
                    "Some types allow empty bytes, but define them as meaningless.  AF can be used to query them using <, <=, and =; but SAI can not"),
     AF_MULTI_NODE_MULTI_COLUMN_AND_NODE_LOCAL_WRITES("https://issues.apache.org/jira/browse/CASSANDRA-19007",
                                                      "When doing multi node/multi column queries, AF can miss data when the nodes are not in-sync"),
+    SAI_AND_VECTOR_COLUMNS("https://issues.apache.org/jira/browse/CASSANDRA-20464",
+                           "When doing an SAI query, if the where clause also contains a vector column bad results can be produced")
     ;
 
     KnownIssue(String url, String description)
diff --git a/test/unit/org/apache/cassandra/cql3/ast/AssignmentOperator.java b/test/unit/org/apache/cassandra/cql3/ast/AssignmentOperator.java
index f72fceb817..0ffb65411b 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/AssignmentOperator.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/AssignmentOperator.java
@@ -53,31 +53,30 @@ public class AssignmentOperator implements Expression
         this.right = right;
     }
 
-    public static EnumSet<Kind> supportsOperators(AbstractType<?> type)
+    public static EnumSet<Kind> supportsOperators(AbstractType<?> type, boolean isTransaction)
     {
         type = type.unwrap();
         EnumSet<Kind> result = EnumSet.noneOf(Kind.class);
+        if (type instanceof CollectionType && type.isMultiCell())
+        {
+            if (type instanceof SetType || type instanceof ListType)
+                return EnumSet.of(Kind.ADD, Kind.SUBTRACT);
+            if (type instanceof MapType)
+            {
+                // map supports subtract, but not map - map; only map - set!
+                // since this is annoying to support, for now dropping -
+                return EnumSet.of(Kind.ADD);
+            }
+            throw new AssertionError("Unexpected collection type: " + type);
+        }
+        if (!isTransaction)
+            return result; // only multi-cell collections can be updated outside of transactions
         for (Operator.Kind supported : Operator.supportsOperators(type))
         {
             Kind kind = toKind(supported);
             if (kind != null)
                 result.add(kind);
         }
-        if (result.isEmpty())
-        {
-            if (type instanceof CollectionType && type.isMultiCell())
-            {
-                if (type instanceof SetType || type instanceof ListType)
-                    return EnumSet.of(Kind.ADD, Kind.SUBTRACT);
-                if (type instanceof MapType)
-                {
-                    // map supports subtract, but not map - map; only map - set!
-                    // since this is annoying to support, for now dropping -
-                    return EnumSet.of(Kind.ADD);
-                }
-                throw new AssertionError("Unexpected collection type: " + type);
-            }
-        }
         return result;
     }
 
diff --git a/test/unit/org/apache/cassandra/cql3/ast/Conditional.java b/test/unit/org/apache/cassandra/cql3/ast/Conditional.java
index 7fdcd17bea..52f79bb1dc 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/Conditional.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/Conditional.java
@@ -433,7 +433,7 @@ public interface Conditional extends Expression
             return sub.isEmpty();
         }
 
-        private Builder add(Conditional conditional)
+        public Builder add(Conditional conditional)
         {
             sub.add(conditional);
             return this;
diff --git a/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java b/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java
index 0984ee4620..ce86d6bbb5 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/CreateIndexDDL.java
@@ -141,7 +141,7 @@ public class CreateIndexDDL implements Element
         public EnumSet<QueryType> supportedQueries(AbstractType<?> type)
         {
             type = type.unwrap();
-            if (IndexTermType.isEqOnlyType(type))
+            if (IndexTermType.isEqOnlyType(type) || type.isCollection() || type.isUDT() || type.isTuple())
                 return EnumSet.of(QueryType.Eq);
             return EnumSet.allOf(QueryType.class);
         }
diff --git a/test/unit/org/apache/cassandra/cql3/ast/ExpressionEvaluator.java b/test/unit/org/apache/cassandra/cql3/ast/ExpressionEvaluator.java
index 8270b438f4..34acb843c7 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/ExpressionEvaluator.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/ExpressionEvaluator.java
@@ -21,107 +21,139 @@ package org.apache.cassandra.cql3.ast;
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
-import java.util.Optional;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import javax.annotation.Nullable;
 
 import org.apache.cassandra.db.marshal.AbstractType;
 
-import static java.util.Optional.of;
-
 public class ExpressionEvaluator
 {
-    public static Optional<Object> tryEval(Expression e)
+    @Nullable
+    public static Object eval(Expression e)
     {
         if (e instanceof Value)
-            return of(((Value) e).value());
+            return ((Value) e).value();
         if (e instanceof TypeHint)
-            return tryEval(((TypeHint) e).e);
+            return eval(((TypeHint) e).e);
         if (e instanceof Operator)
-            return tryEval((Operator) e);
-        return Optional.empty();
+            return eval((Operator) e);
+        throw new UnsupportedOperationException("Unexpected expression type " + e.getClass() + ": " + e.toCQL());
     }
 
-    public static Optional<Object> tryEval(Operator e)
+    @Nullable
+    public static Object eval(Operator e)
     {
+        Object lhs = eval(e.left);
+        if (lhs instanceof ByteBuffer)
+            lhs = e.left.type().compose((ByteBuffer) lhs);
+        Object rhs = eval(e.right);
+        if (rhs instanceof ByteBuffer)
+            rhs = e.right.type().compose((ByteBuffer) rhs);
         switch (e.kind)
         {
             case ADD:
             {
-                var lhsOpt = tryEval(e.left);
-                var rhsOpt = tryEval(e.right);
-                if (lhsOpt.isEmpty() || rhsOpt.isEmpty())
-                    return Optional.empty();
-                Object lhs = lhsOpt.get();
-                Object rhs = rhsOpt.get();
                 if (lhs instanceof Byte)
-                    return of((byte) (((Byte) lhs) + ((Byte) rhs)));
+                    return (byte) (((Byte) lhs) + ((Byte) rhs));
                 if (lhs instanceof Short)
-                    return of((short) (((Short) lhs) + ((Short) rhs)));
+                    return (short) (((Short) lhs) + ((Short) rhs));
                 if (lhs instanceof Integer)
-                    return of((int) (((Integer) lhs) + ((Integer) rhs)));
+                    return (int) (((Integer) lhs) + ((Integer) rhs));
                 if (lhs instanceof Long)
-                    return of((long) (((Long) lhs) + ((Long) rhs)));
+                    return (long) (((Long) lhs) + ((Long) rhs));
                 if (lhs instanceof Float)
-                    return of((float) (((Float) lhs) + ((Float) rhs)));
+                    return (float) (((Float) lhs) + ((Float) rhs));
                 if (lhs instanceof Double)
-                    return of((double) (((Double) lhs) + ((Double) rhs)));
+                    return (double) (((Double) lhs) + ((Double) rhs));
                 if (lhs instanceof BigInteger)
-                    return of(((BigInteger) lhs).add((BigInteger) rhs));
+                    return ((BigInteger) lhs).add((BigInteger) rhs);
                 if (lhs instanceof BigDecimal)
-                    return of(((BigDecimal) lhs).add((BigDecimal) rhs));
+                    return ((BigDecimal) lhs).add((BigDecimal) rhs);
                 if (lhs instanceof String)
-                    return of(lhs.toString() + rhs.toString());
+                    return lhs.toString() + rhs.toString();
+                if (lhs instanceof Set)
+                {
+                    Set<Object> accum = new HashSet<>((Set<Object>) lhs);
+                    accum.addAll((Set<Object>) rhs);
+                    return accum;
+                }
+                if (lhs instanceof List)
+                {
+                    List<Object> accum = new ArrayList<>((List<Object>) lhs);
+                    accum.addAll((List<Object>) rhs);
+                    return accum;
+                }
+                if (lhs instanceof Map)
+                {
+                    Map<Object, Object> accum = new HashMap<>((Map<Object, Object>) lhs);
+                    accum.putAll((Map<Object, Object>) rhs);
+                    return accum;
+                }
                 throw new UnsupportedOperationException("Unexpected type: " + lhs.getClass());
             }
             case SUBTRACT:
             {
-                var lhsOpt = tryEval(e.left);
-                var rhsOpt = tryEval(e.right);
-                if (lhsOpt.isEmpty() || rhsOpt.isEmpty())
-                    return Optional.empty();
-                Object lhs = lhsOpt.get();
-                Object rhs = rhsOpt.get();
                 if (lhs instanceof Byte)
-                    return of((byte) (((Byte) lhs) - ((Byte) rhs)));
+                    return (byte) (((Byte) lhs) - ((Byte) rhs));
                 if (lhs instanceof Short)
-                    return of((short) (((Short) lhs) - ((Short) rhs)));
+                    return (short) (((Short) lhs) - ((Short) rhs));
                 if (lhs instanceof Integer)
-                    return of((int) (((Integer) lhs) - ((Integer) rhs)));
+                    return (int) (((Integer) lhs) - ((Integer) rhs));
                 if (lhs instanceof Long)
-                    return of((long) (((Long) lhs) - ((Long) rhs)));
+                    return (long) (((Long) lhs) - ((Long) rhs));
                 if (lhs instanceof Float)
-                    return of((float) (((Float) lhs) - ((Float) rhs)));
+                    return (float) (((Float) lhs) - ((Float) rhs));
                 if (lhs instanceof Double)
-                    return of((double) (((Double) lhs) - ((Double) rhs)));
+                    return (double) (((Double) lhs) - ((Double) rhs));
                 if (lhs instanceof BigInteger)
-                    return of(((BigInteger) lhs).subtract((BigInteger) rhs));
+                    return ((BigInteger) lhs).subtract((BigInteger) rhs);
                 if (lhs instanceof BigDecimal)
-                    return of(((BigDecimal) lhs).subtract((BigDecimal) rhs));
+                    return ((BigDecimal) lhs).subtract((BigDecimal) rhs);
+                if (lhs instanceof Set)
+                {
+                    Set<Object> accum = new HashSet<>((Set<Object>) lhs);
+                    accum.removeAll((Set<Object>) rhs);
+                    return accum.isEmpty() ? null : accum;
+                }
+                if (lhs instanceof List)
+                {
+                    List<Object> accum = new ArrayList<>((List<Object>) lhs);
+                    accum.removeAll((List<Object>) rhs);
+                    return accum.isEmpty() ? null : accum;
+                }
+                if (lhs instanceof Map)
+                {
+                    // rhs is a Set<Object> as CQL doesn't allow removing if the key and value both match
+                    Map<Object, Object> accum = new HashMap<>((Map<Object, Object>) lhs);
+                    ((Set<Object>) rhs).forEach(accum::remove);
+                    return accum.isEmpty() ? null : accum;
+                }
                 throw new UnsupportedOperationException("Unexpected type: " + lhs.getClass());
             }
             case MULTIPLY:
             {
-                var lhsOpt = tryEval(e.left);
-                var rhsOpt = tryEval(e.right);
-                if (lhsOpt.isEmpty() || rhsOpt.isEmpty())
-                    return Optional.empty();
-                Object lhs = lhsOpt.get();
-                Object rhs = rhsOpt.get();
                 if (lhs instanceof Byte)
-                    return of((byte) (((Byte) lhs) * ((Byte) rhs)));
+                    return (byte) (((Byte) lhs) * ((Byte) rhs));
                 if (lhs instanceof Short)
-                    return of((short) (((Short) lhs) * ((Short) rhs)));
+                    return (short) (((Short) lhs) * ((Short) rhs));
                 if (lhs instanceof Integer)
-                    return of((int) (((Integer) lhs) * ((Integer) rhs)));
+                    return (int) (((Integer) lhs) * ((Integer) rhs));
                 if (lhs instanceof Long)
-                    return of((long) (((Long) lhs) * ((Long) rhs)));
+                    return (long) (((Long) lhs) * ((Long) rhs));
                 if (lhs instanceof Float)
-                    return of((float) (((Float) lhs) * ((Float) rhs)));
+                    return (float) (((Float) lhs) * ((Float) rhs));
                 if (lhs instanceof Double)
-                    return of((double) ((Double) lhs) * ((Double) rhs));
+                    return (double) ((Double) lhs) * ((Double) rhs);
                 if (lhs instanceof BigInteger)
-                    return of(((BigInteger) lhs).multiply((BigInteger) rhs));
+                    return ((BigInteger) lhs).multiply((BigInteger) rhs);
                 if (lhs instanceof BigDecimal)
-                    return of(((BigDecimal) lhs).multiply((BigDecimal) rhs));
+                    return ((BigDecimal) lhs).multiply((BigDecimal) rhs);
                 throw new UnsupportedOperationException("Unexpected type: " + lhs.getClass());
             }
             default:
@@ -129,18 +161,12 @@ public class ExpressionEvaluator
         }
     }
 
-    public static Optional<ByteBuffer> tryEvalEncoded(Expression e)
+    @Nullable
+    public static ByteBuffer evalEncoded(Expression e)
     {
-        return tryEval(e).map(v -> {
-            if (v instanceof ByteBuffer) return (ByteBuffer) v;
-            try
-            {
-                return ((AbstractType) e.type()).decompose(v);
-            }
-            catch (Throwable t)
-            {
-                throw t;
-            }
-        });
+        Object v = eval(e);
+        if (v == null) return null;
+        if (v instanceof ByteBuffer) return (ByteBuffer) v;
+        return ((AbstractType) e.type()).decompose(v);
     }
 }
diff --git a/test/unit/org/apache/cassandra/cql3/ast/Mutation.java b/test/unit/org/apache/cassandra/cql3/ast/Mutation.java
index c21508f4d2..9126d8cbda 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/Mutation.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/Mutation.java
@@ -720,6 +720,12 @@ WHERE PK_column_conditions
             return set(new Symbol(column, Int32Type.instance), Bind.of(value));
         }
 
+        public UpdateBuilder set(String column, Expression expression)
+        {
+            Symbol symbol = new Symbol(metadata.getColumn(new ColumnIdentifier(column, true)));
+            return set(symbol, expression);
+        }
+
         public UpdateBuilder set(String column, String value)
         {
             Symbol symbol = new Symbol(metadata.getColumn(new ColumnIdentifier(column, true)));
diff --git a/test/unit/org/apache/cassandra/cql3/ast/Select.java b/test/unit/org/apache/cassandra/cql3/ast/Select.java
index 7fa497e16c..28134dde8e 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/Select.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/Select.java
@@ -378,6 +378,13 @@ FROM [keyspace_name.] table_name
             return (T) this;
         }
 
+        public T where(Conditional conditional)
+        {
+            where = new Conditional.Builder();
+            where.add(conditional);
+            return (T) this;
+        }
+
         @Override
         public T where(Expression ref, Conditional.Where.Inequality kind, Expression expression)
         {
diff --git a/test/unit/org/apache/cassandra/cql3/ast/StandardVisitors.java b/test/unit/org/apache/cassandra/cql3/ast/StandardVisitors.java
index 854c096c07..4cbf3d989f 100644
--- a/test/unit/org/apache/cassandra/cql3/ast/StandardVisitors.java
+++ b/test/unit/org/apache/cassandra/cql3/ast/StandardVisitors.java
@@ -50,7 +50,7 @@ public class StandardVisitors
         public Expression visit(Expression e)
         {
             if (!(e instanceof Operator)) return e;
-            return new Bind(ExpressionEvaluator.tryEval((Operator) e).get(), e.type());
+            return new Bind(ExpressionEvaluator.eval((Operator) e), e.type());
         }
     };
 
diff --git a/test/unit/org/apache/cassandra/utils/ASTGenerators.java b/test/unit/org/apache/cassandra/utils/ASTGenerators.java
index a02bf65610..47b2267de3 100644
--- a/test/unit/org/apache/cassandra/utils/ASTGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/ASTGenerators.java
@@ -35,6 +35,7 @@ import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import javax.annotation.Nullable;
@@ -98,7 +99,7 @@ public class ASTGenerators
         throw new AssertionError("Unsupported map type: " + map.getClass());
     }
 
-    public static Gen<AssignmentOperator> assignmentOperatorGen(EnumSet<AssignmentOperator.Kind> allowed, Expression right)
+    private static Gen<AssignmentOperator> assignmentOperatorGen(EnumSet<AssignmentOperator.Kind> allowed, Expression right)
     {
         if (allowed.isEmpty())
             throw new IllegalArgumentException("Unable to create a operator gen for empty set of allowed operators");
@@ -187,6 +188,12 @@ public class ASTGenerators
             return this;
         }
 
+        public ExpressionBuilder withOperators(Gen<Boolean> useOperator)
+        {
+            this.useOperator = Objects.requireNonNull(useOperator);
+            return this;
+        }
+
         public ExpressionBuilder withoutOperators()
         {
             useOperator = i -> false;
@@ -375,6 +382,13 @@ public class ASTGenerators
                 columnExpressions.put(symbol, new ExpressionBuilder(symbol.type()));
         }
 
+        public MutationGenBuilder withColumnExpressions(Consumer<ExpressionBuilder> fn)
+        {
+            for (Symbol symbol : allColumns)
+                fn.accept(columnExpressions.get(symbol));
+            return this;
+        }
+
         public MutationGenBuilder allowEmpty(Symbol symbol)
         {
             columnExpressions.get(symbol).allowEmpty();
@@ -770,12 +784,12 @@ public class ASTGenerators
                     }
                 }
             }
-            if (kind == Mutation.Kind.UPDATE && isTransaction)
+            if (kind == Mutation.Kind.UPDATE)
             {
                 for (Symbol c : new ArrayList<>(columnsToGenerate))
                 {
                     var useOperator = columnExpressions.get(c).useOperator;
-                    EnumSet<AssignmentOperator.Kind> additionOperatorAllowed = AssignmentOperator.supportsOperators(c.type());
+                    EnumSet<AssignmentOperator.Kind> additionOperatorAllowed = AssignmentOperator.supportsOperators(c.type(), isTransaction);
                     if (!additionOperatorAllowed.isEmpty() && useOperator.generate(rnd))
                     {
                         Expression expression = columnExpressions.get(c).build().generate(rnd);
diff --git a/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java b/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java
index 902c37e92d..ea9a128233 100644
--- a/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java
+++ b/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java
@@ -26,7 +26,6 @@ import java.util.Comparator;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -165,12 +164,13 @@ public final class AbstractTypeGenerators
     ).collect(Collectors.toMap(t -> t.type, t -> t));
     // NOTE not supporting reversed as CQL doesn't allow nested reversed types
     // when generating part of the clustering key, it would be good to allow reversed types as the top level
-    private static final Gen<AbstractType<?>> PRIMITIVE_TYPE_GEN;
-    static
+    private static final Gen<AbstractType<?>> PRIMITIVE_TYPE_GEN = SourceDSL.arbitrary().pick(knownPrimitiveTypes());
+
+    public static List<AbstractType<?>> knownPrimitiveTypes()
     {
         ArrayList<AbstractType<?>> types = new ArrayList<>(PRIMITIVE_TYPE_DATA_GENS.keySet());
         types.sort(Comparator.comparing(a -> a.getClass().getName()));
-        PRIMITIVE_TYPE_GEN = SourceDSL.arbitrary().pick(types);
+        return types;
     }
 
     private static final Set<Class<? extends AbstractType>> NON_PRIMITIVE_TYPES = ImmutableSet.<Class<? extends AbstractType>>builder()
@@ -244,6 +244,14 @@ public final class AbstractTypeGenerators
         return () -> PRIMITIVE_TYPE_DATA_GENS.put(type, original);
     }
 
+    public static boolean isUnsafeEquality(AbstractType<?> type)
+    {
+        return type == EmptyType.instance
+               || type == DurationType.instance
+               || type == DecimalType.instance
+               || type == CounterColumnType.instance;
+    }
+
     public static TypeGenBuilder withoutUnsafeEquality(TypeGenBuilder builder)
     {
         // make sure to keep UNSAFE_EQUALITY in-sync
@@ -281,6 +289,7 @@ public final class AbstractTypeGenerators
         private Predicate<AbstractType<?>> typeFilter = null;
         private Gen<String> udtName = null;
         private Gen<Boolean> multiCellGen = BOOLEAN_GEN;
+        private UserTypeFieldsGen fieldNamesGen = UserTypeFieldsGen.random();
 
         public TypeGenBuilder()
         {
@@ -289,19 +298,27 @@ public final class AbstractTypeGenerators
         public TypeGenBuilder(TypeGenBuilder other)
         {
             maxDepth = other.maxDepth;
-            kinds = other.kinds == null ? null : EnumSet.copyOf(other.kinds);
+            kinds = other.kinds;
             typeKindGen = other.typeKindGen;
             defaultSizeGen = other.defaultSizeGen;
             vectorSizeGen = other.vectorSizeGen;
             tupleSizeGen = other.tupleSizeGen;
-            udtName = other.udtName;
             udtSizeGen = other.udtSizeGen;
+            compositeSizeGen = other.compositeSizeGen;
             primitiveGen = other.primitiveGen;
+            compositeElementGen = other.compositeElementGen;
             userTypeKeyspaceGen = other.userTypeKeyspaceGen;
             defaultSetKeyFunc = other.defaultSetKeyFunc;
-            compositeElementGen = other.compositeElementGen;
-            compositeSizeGen = other.compositeSizeGen;
             typeFilter = other.typeFilter;
+            udtName = other.udtName;
+            multiCellGen = other.multiCellGen;
+            fieldNamesGen = other.fieldNamesGen;
+        }
+
+        public TypeGenBuilder withUserTypeFields(UserTypeFieldsGen fieldNamesGen)
+        {
+            this.fieldNamesGen = fieldNamesGen;
+            return this;
         }
 
         public TypeGenBuilder withMultiCell(Gen<Boolean> multiCellGen)
@@ -406,6 +423,13 @@ public final class AbstractTypeGenerators
             return this;
         }
 
+        public TypeGenBuilder withPrimitives(Gen<AbstractType<?>> gen)
+        {
+            // any previous filters will be ignored...
+            primitiveGen = Objects.requireNonNull(gen);
+            return this;
+        }
+
         public TypeGenBuilder withMaxDepth(int value)
         {
             this.maxDepth = value;
@@ -509,7 +533,7 @@ public final class AbstractTypeGenerators
                     case TUPLE:
                         return tupleTypeGen(atBottom ? primitiveGen : buildRecursive(maxDepth, level - 1, typeKindGen, SourceDSL.arbitrary().constant(false)), tupleSizeGen != null ? tupleSizeGen : defaultSizeGen).generate(rnd);
                     case UDT:
-                        return userTypeGen(next.get(), udtSizeGen != null ? udtSizeGen : defaultSizeGen, userTypeKeyspaceGen, udtName, multiCellGen).generate(rnd);
+                        return userTypeGen(fieldNamesGen, next.get(), udtSizeGen != null ? udtSizeGen : defaultSizeGen, userTypeKeyspaceGen, udtName, multiCellGen).generate(rnd);
                     case VECTOR:
                     {
                         Gen<Integer> sizeGen = vectorSizeGen != null ? vectorSizeGen : defaultSizeGen;
@@ -762,27 +786,47 @@ public final class AbstractTypeGenerators
         OVERRIDE_KEYSPACE.remove();
     }
 
+    public interface UserTypeFieldsGen
+    {
+        List<FieldIdentifier> generate(RandomnessSource rnd, int size);
+
+        static UserTypeFieldsGen random()
+        {
+            Gen<FieldIdentifier> fieldNameGen = IDENTIFIER_GEN.map(FieldIdentifier::forQuoted);
+            return (rnd, size) -> Generators.uniqueList(fieldNameGen, i -> size).generate(rnd);
+        }
+
+        static UserTypeFieldsGen simpleNames()
+        {
+            return (rnd, size) -> {
+                List<FieldIdentifier> output = new ArrayList<>(size);
+                for (int i = 0; i < size; i++)
+                    output.add(FieldIdentifier.forUnquoted("f" + i));
+                return output;
+            };
+        }
+    }
+
     public static Gen<UserType> userTypeGen(Gen<AbstractType<?>> elementGen, Gen<Integer> sizeGen, Gen<String> ksGen, Gen<String> nameGen, Gen<Boolean> multiCellGen)
     {
-        Gen<FieldIdentifier> fieldNameGen = IDENTIFIER_GEN.map(FieldIdentifier::forQuoted);
+        return userTypeGen(UserTypeFieldsGen.random(), elementGen, sizeGen, ksGen, nameGen, multiCellGen);
+    }
+
+    public static Gen<UserType> userTypeGen(UserTypeFieldsGen fieldNamesGen, Gen<AbstractType<?>> elementGen, Gen<Integer> sizeGen, Gen<String> ksGen, Gen<String> nameGen, Gen<Boolean> multiCellGen)
+    {
         return rnd -> {
             boolean multiCell = multiCellGen.generate(rnd);
             int numElements = sizeGen.generate(rnd);
             List<AbstractType<?>> fieldTypes = new ArrayList<>(numElements);
-            LinkedHashSet<FieldIdentifier> fieldNames = new LinkedHashSet<>(numElements);
+            List<FieldIdentifier> fieldNames = fieldNamesGen.generate(rnd, numElements);
             String ks = OVERRIDE_KEYSPACE.get();
             if (ks == null)
                 ks = ksGen.generate(rnd);
             String name = nameGen.generate(rnd);
             ByteBuffer nameBB = AsciiType.instance.decompose(name);
 
-            Gen<FieldIdentifier> distinctNameGen = filter(fieldNameGen, 30, e -> !fieldNames.contains(e));
-            // UDTs don't allow duplicate names, so make sure all names are unique
             for (int i = 0; i < numElements; i++)
             {
-                FieldIdentifier fieldName = distinctNameGen.generate(rnd);
-                fieldNames.add(fieldName);
-
                 AbstractType<?> element = elementGen.generate(rnd);
                 element = multiCell ? element.freeze() : element.unfreeze();
                 // a UDT cannot contain a non-frozen UDT; as defined by CreateType
@@ -790,7 +834,7 @@ public final class AbstractTypeGenerators
                     element = element.freeze();
                 fieldTypes.add(element);
             }
-            return new UserType(ks, nameBB, new ArrayList<>(fieldNames), fieldTypes, multiCell);
+            return new UserType(ks, nameBB, fieldNames, fieldTypes, multiCell);
         };
     }
 
diff --git a/test/unit/org/apache/cassandra/utils/Generators.java b/test/unit/org/apache/cassandra/utils/Generators.java
index 5f99421240..6bb7f56a8d 100644
--- a/test/unit/org/apache/cassandra/utils/Generators.java
+++ b/test/unit/org/apache/cassandra/utils/Generators.java
@@ -27,7 +27,6 @@ import java.sql.Timestamp;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
-import java.util.Comparator;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.List;
@@ -38,10 +37,13 @@ import java.util.concurrent.TimeUnit;
 import java.util.function.Predicate;
 
 import com.google.common.collect.Range;
+import com.google.common.collect.Sets;
 import org.apache.commons.lang3.ArrayUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import accord.utils.DefaultRandom;
+import accord.utils.RandomSource;
 import org.apache.cassandra.cql3.ReservedKeywords;
 import org.quicktheories.core.Gen;
 import org.quicktheories.core.RandomnessSource;
@@ -492,13 +494,20 @@ public final class Generators
         };
     }
 
-    public static <T extends Comparable<? super T>> Gen<List<T>> uniqueList(Gen<T> gen, Gen<Integer> sizeGen)
+    public static <T> Gen<List<T>> uniqueList(Gen<T> gen, Gen<Integer> sizeGen)
     {
-        return set(gen, sizeGen).map(t -> {
-            List<T> list = new ArrayList<>(t);
-            list.sort(Comparator.naturalOrder());
-            return list;
-        });
+        return rnd -> {
+            int size = sizeGen.generate(rnd);
+            Set<T> set = Sets.newHashSetWithExpectedSize(size);
+            List<T> output = new ArrayList<>(size);
+            for (int i = 0; i < size; i++)
+            {
+                T value;
+                while (!set.add(value = gen.generate(rnd))) {}
+                output.add(value);
+            }
+            return output;
+        };
     }
 
     public static <T> Gen<T> cached(Gen<T> gen)
@@ -612,6 +621,14 @@ public final class Generators
         };
     }
 
+    public static <T> org.quicktheories.core.Gen<T> fromGen(accord.utils.Gen<T> accord)
+    {
+        return rnd -> {
+            RandomSource rs = new DefaultRandom(rnd.next(Constraint.none()));
+            return accord.next(rs);
+        };
+    }
+
     public static Gen<TimeUUID> timeUUID()
     {
         ZonedDateTime now = ZonedDateTime.of(2020, 8, 20,
diff --git a/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java b/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java
index 73e7da9a91..00fabea136 100644
--- a/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java
+++ b/test/unit/org/apache/cassandra/utils/ImmutableUniqueList.java
@@ -32,6 +32,8 @@ import org.agrona.collections.Object2IntHashMap;
 
 public class ImmutableUniqueList<T> extends AbstractList<T> implements RandomAccess
 {
+    private static final ImmutableUniqueList<Object> EMPTY = ImmutableUniqueList.builder().build();
+
     private final T[] values;
     private final Object2IntHashMap<T> indexLookup;
     private transient AsSet asSet = null;
@@ -46,6 +48,16 @@ public class ImmutableUniqueList<T> extends AbstractList<T> implements RandomAcc
         return new Builder<>();
     }
 
+    public static <T> Builder<T> builder(int expectedSize)
+    {
+        return new Builder<>(expectedSize);
+    }
+
+    public static <T> ImmutableUniqueList<T> empty()
+    {
+        return (ImmutableUniqueList<T>) EMPTY;
+    }
+
     public AsSet asSet()
     {
         if (asSet != null) return asSet;
@@ -85,10 +97,20 @@ public class ImmutableUniqueList<T> extends AbstractList<T> implements RandomAcc
 
     public static final class Builder<T> extends AbstractSet<T>
     {
-        private final List<T> values = new ArrayList<>();
+        private final List<T> values;
         private final Object2IntHashMap<T> indexLookup = new Object2IntHashMap<>(-1);
         private int idx;
 
+        public Builder()
+        {
+            this.values = new ArrayList<>();
+        }
+
+        public Builder(int expectedSize)
+        {
+            this.values = new ArrayList<>(expectedSize);
+        }
+
         public Builder<T> mayAddAll(Collection<? extends T> values)
         {
             addAll(values);