From d726f1414b1457ab09a40adc6e92dd2f3f16a8a9 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 2 Jul 2026 15:13:21 -0700 Subject: [PATCH 01/16] changes --- src/antlr/Lexer.g | 1 + src/antlr/Parser.g | 23 ++- .../cql3/statements/TransactionStatement.java | 162 ++++++++++++------ .../cql3/transactions/ConditionStatement.java | 16 +- .../service/accord/txn/TxnCondition.java | 42 ++++- .../service/accord/txn/TxnUpdate.java | 105 ++++++++++++ 6 files changed, 289 insertions(+), 60 deletions(-) diff --git a/src/antlr/Lexer.g b/src/antlr/Lexer.g index c7065f7351b5..a87a70027ad9 100644 --- a/src/antlr/Lexer.g +++ b/src/antlr/Lexer.g @@ -128,6 +128,7 @@ K_ALLOW: A L L O W; K_FILTERING: F I L T E R I N G; K_IF: I F; K_THEN: T H E N; +K_ELSE: E L S E; K_END: E N D; K_IS: I S; K_CONTAINS: C O N T A I N S; diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g index 8794d00d02a9..f3458281f5d0 100644 --- a/src/antlr/Parser.g +++ b/src/antlr/Parser.g @@ -783,20 +783,35 @@ batchTxnStatement returns [TransactionStatement.Parsed expr] List assignments = new ArrayList<>(); SelectStatement.RawStatement select = null; List returning = null; - List updates = new ArrayList<>(); + List> conditions = new ArrayList<>(); + List> updates = new ArrayList<>(); + List e = new ArrayList<>(); + e.add(new ConditionStatement.Raw(null, ConditionStatement.Kind.ELSE, null)); } : K_BEGIN K_TRANSACTION ( let=letStatement ';' { assignments.add(let); })* ( ( (selectStatement) => s=selectStatement ';' { select = s; }) | ( K_SELECT drs=rowDataReferences ';' { returning = drs; }) )? - ( K_IF conditions=txnConditions K_THEN { isTxnConditional = true; } )? - ( upd=batchStatementObjective ';' { updates.add(upd); } )* - ( {!isTxnConditional}? (K_COMMIT K_TRANSACTION) | {isTxnConditional}? (K_END K_IF K_COMMIT K_TRANSACTION)) + ( + K_IF c=txnConditions K_THEN { isTxnConditional = true; } u=updateStatements { conditions.add(c); updates.add(u); } + ( K_ELSE K_IF c=txnConditions K_THEN u=updateStatements { conditions.add(c); updates.add(u); } )* + ( K_ELSE u=updateStatements { conditions.add(e); updates.add(u); } )? + K_END K_IF + )? + ( (batchStatementObjective) => u=updateStatements { updates.add(u); } )? + (K_COMMIT K_TRANSACTION) { $expr = new TransactionStatement.Parsed(assignments, select, returning, updates, conditions, references); } ; finally { isParsingTxn = false; } +updateStatements returns [List updates] + @init { + updates = new ArrayList(); + } + : (upd=batchStatementObjective ';' { updates.add(upd); })* + ; + rowDataReferences returns [List refs] : r1=rowDataReference { refs = new ArrayList(); refs.add(r1); } (',' rN=rowDataReference { refs.add(rN); })* ; diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 97a7fd32f2c6..8c7df6dc53cd 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -20,6 +20,7 @@ import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; @@ -45,6 +46,7 @@ import accord.primitives.Keys; import accord.primitives.Routable.Domain; import accord.primitives.Txn; +import accord.utils.Invariants; import org.apache.cassandra.audit.AuditLogContext; import org.apache.cassandra.audit.AuditLogEntryType; @@ -98,6 +100,7 @@ import org.apache.cassandra.transport.Dispatcher; import org.apache.cassandra.transport.messages.ResultMessage; import org.apache.cassandra.utils.NoSpamLogger; +import org.apache.cassandra.utils.Pair; import static accord.primitives.Txn.Kind.Read; import static com.google.common.base.Preconditions.checkArgument; @@ -155,25 +158,32 @@ public NamedSelect(int name, SelectStatement select) private final NamedSelect returningSelect; private final List returningReferences; private final List updates; + private final List> groupedUpdates; private final List conditions; + private final List> groupedConditions; private final VariableSpecifications bindVariables; private final ResultSet.ResultMetadata resultMetadata; private long minEpoch = Epoch.EMPTY.getEpoch(); + // Every index of List corresponds with an index of groupedConditions. + // If the size of groupedUpdates is larger than the size of groupedConditions this means that + // we have updates that are not part of a condition public TransactionStatement(List assignments, NamedSelect returningSelect, List returningReferences, - List updates, - List conditions, + List> groupedUpdates, + List> groupedConditions, // Each List represents the condition in IF, ELSE IF, ELSE VariableSpecifications bindVariables) { this.assignments = assignments; this.returningSelect = returningSelect; this.returningReferences = returningReferences; - this.updates = updates; - this.conditions = conditions; + this.updates = groupedUpdates.stream().flatMap(Collection::stream).collect(Collectors.toList()); + this.groupedUpdates = groupedUpdates; + this.conditions = groupedConditions.stream().flatMap(Collection::stream).collect(Collectors.toList()); + this.groupedConditions = groupedConditions; this.bindVariables = bindVariables; if (returningSelect != null) @@ -331,7 +341,7 @@ private List createNamedReads(QueryOptions options, @Nullable Int2 return reads; } - TxnCondition createCondition(QueryOptions options) + TxnCondition createCondition(List conditions, QueryOptions options) { if (conditions.isEmpty()) return TxnCondition.none(); @@ -376,31 +386,50 @@ private Keys toKeys(SortedSet keySet) return new Keys(keySet); } - List createWriteFragments(ClientState state, QueryOptions options, Map autoReads, TableMetadatasAndKeys.KeyCollector keyCollector) + List> createWriteFragments(ClientState state, QueryOptions options, Map autoReads, TableMetadatasAndKeys.KeyCollector keyCollector) { - // check that within a transaction we don't have multiple updates to the same primary key, column pair - HashMap seenColumns = new HashMap<>(); + List> groupedWriteFragments = new ArrayList<>(groupedUpdates.size()); + + // This is an over approximation of all the possible seen columns + HashMap totalSeenColumns = new HashMap<>(); - List fragments = new ArrayList<>(updates.size()); int idx = 0; - for (ModificationStatement modification : updates) + for (int groupedUpdatesIdx = 0; groupedUpdatesIdx < groupedUpdates.size(); groupedUpdatesIdx++) { - minEpoch = Math.max(minEpoch, modification.metadata().epoch.getEpoch()); - List writeFragments = modification.getTxnWriteFragment(idx, state, options, keyCollector); - fragments.addAll(writeFragments); - validateOnlyModifyPrimaryKeyColumnPairOnce(seenColumns, modification, writeFragments); + List updates = groupedUpdates.get(groupedUpdatesIdx); + HashMap seenColumns = new HashMap<>(); - if (modification.allReferenceOperations().stream().anyMatch(ReferenceOperation::requiresRead)) + List fragments = new ArrayList<>(updates.size()); + for (ModificationStatement modification : updates) { - // Reads are not merged by partition here due to potentially differing columns retrieved, etc. - int partitionName = txnDataName(AUTO_READ, idx); - if (!autoReads.containsKey(partitionName)) - autoReads.put(partitionName, new NamedSelect(partitionName, modification.createSelectForTxn())); - } + minEpoch = Math.max(minEpoch, modification.metadata().epoch.getEpoch()); + List writeFragments = modification.getTxnWriteFragment(idx, state, options, keyCollector); + fragments.addAll(writeFragments); + // TODO: When adding support for consecutive IF statements, we need to revisit CASSANDRA-21136, currently we perform this check per branch since only one branch can be executed + // In the case where we have a trailing update, we use an overapproximation of all the seen columns we + // have accumulated within each branch to perform the validation + if (groupedUpdatesIdx == groupedUpdates.size() - 1 && groupedUpdates.size() > groupedConditions.size()) + validateOnlyModifyPrimaryKeyColumnPairOnce(totalSeenColumns, modification, writeFragments); + else + { + validateOnlyModifyPrimaryKeyColumnPairOnce(seenColumns, modification, writeFragments); + totalSeenColumns.putAll(seenColumns); + } + + if (modification.allReferenceOperations().stream().anyMatch(ReferenceOperation::requiresRead)) + { + // Reads are not merged by partition here due to potentially differing columns retrieved, etc. + int partitionName = txnDataName(AUTO_READ, idx); + if (!autoReads.containsKey(partitionName)) + autoReads.put(partitionName, new NamedSelect(partitionName, modification.createSelectForTxn())); + } - idx++; + idx++; + } + groupedWriteFragments.add(fragments); } - return fragments; + + return groupedWriteFragments; } private static void validateOnlyModifyPrimaryKeyColumnPairOnce(HashMap seenColumns, @@ -520,9 +549,28 @@ public Txn createTxn(ClientState state, QueryOptions options) else { Int2ObjectHashMap autoReads = new Int2ObjectHashMap<>(); - List writeFragments = createWriteFragments(state, options, autoReads, keyCollector); + + TxnCondition[] conditions = new TxnCondition[groupedConditions.size()]; + // Groups each fragment with an index corresponding to the TxnCondition it is associated with + List> conditionFragments = new ArrayList<>(); + List> groupedFragments = createWriteFragments(state, options, autoReads, keyCollector); + + int idx = 0; + while (idx < groupedConditions.size()) + { + conditions[idx] = createCondition(groupedConditions.get(idx), options); + List correspondingFragments = groupedFragments.get(idx); + for (TxnWrite.Fragment fragment : correspondingFragments) + conditionFragments.add(Pair.create(fragment, idx)); + idx++; + } + + List noneConditionFragments = new ArrayList<>(); + if (idx < groupedFragments.size()) + noneConditionFragments.addAll(groupedFragments.get(idx)); + List reads = createNamedReads(options, autoReads, keyCollector); - if (writeFragments.isEmpty()) // ModificationStatement yield no Mutation (DELETE WHERE pk=0 AND c < 0 AND c > 0 -- matches no keys; so has no mutation) + if (conditionFragments.isEmpty() && noneConditionFragments.isEmpty()) // ModificationStatement yield no Mutation (DELETE WHERE pk=0 AND c < 0 AND c > 0 -- matches no keys; so has no mutation) { // cleanup memory keyCollector.clear(); @@ -531,7 +579,8 @@ public Txn createTxn(ClientState state, QueryOptions options) } ConsistencyLevel commitCL = consistencyLevelForAccordCommit(cm, tables, keyCollector, options.getConsistency()); Keys keys = keyCollector.build(); - AccordUpdate update = new TxnUpdate(tables, writeFragments, createCondition(options), commitCL, PreserveTimestamp.no); + + AccordUpdate update = new TxnUpdate(tables, conditions, conditionFragments, noneConditionFragments, commitCL, PreserveTimestamp.no); TxnRead read = createTxnRead(tables, reads, null, Domain.Key); return new Txn.InMemory(keys, read, TxnQuery.ALL, update, new TableMetadatasAndKeys(tables, keys)); } @@ -720,15 +769,15 @@ public static class Parsed extends QualifiedStatement.Composite private final List assignments; private final SelectStatement.RawStatement select; private final List returning; - private final List updates; - private final List conditions; + private final List> updates; + private final List> conditions; private final List dataReferences; public Parsed(List assignments, SelectStatement.RawStatement select, List returning, - List updates, - List conditions, + List> updates, + List> conditions, List dataReferences) { this.assignments = assignments; @@ -742,7 +791,7 @@ public Parsed(List assignments, @Override protected Iterable getStatements() { - Iterable group = Iterables.concat(assignments, updates); + Iterable group = Iterables.concat(assignments, updates.stream().flatMap(Collection::stream).collect(Collectors.toList())); if (select != null) group = Iterables.concat(group, Collections.singleton(select)); return group; @@ -752,7 +801,7 @@ protected Iterable getStatements() public CQLStatement prepare(ClientState state) { checkFalse(updates.isEmpty() && returning == null && select == null, EMPTY_TRANSACTION_MESSAGE); - + Invariants.require(updates.size() == conditions.size() + 1 || updates.size() == conditions.size()); if (select != null || returning != null) checkTrue(select != null ^ returning != null, "Cannot specify both a full SELECT and a SELECT w/ LET references."); @@ -804,30 +853,43 @@ public CQLStatement prepare(ClientState state) .collect(Collectors.toList()); } - List preparedUpdates = new ArrayList<>(updates.size()); - - // check for any read-before-write updates - for (int i = 0; i < updates.size(); i++) - { - ModificationStatement.Parsed parsed = updates.get(i); + List> preparedUpdates = new ArrayList<>(updates.size()); + for (int updateIdx = 0; updateIdx < updates.size(); updateIdx++) { + List preparedUpdate = new ArrayList<>(updates.get(updateIdx).size()); + + // check for any read-before-write updates + for (int i = 0; i < updates.get(updateIdx).size(); i++) + { + ModificationStatement.Parsed parsed = updates.get(updateIdx).get(i); - ModificationStatement prepared = parsed.prepare(state, bindVariables); - checkTrue(prepared.metadata().isAccordEnabled(), TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, prepared.type, prepared.source); - checkFalse(prepared.metadata().params.pendingDrop, TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, prepared.type, prepared.source); - checkFalse(prepared.hasConditions(), NO_CONDITIONS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); - checkFalse(prepared.isTimestampSet(), NO_TIMESTAMPS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); - checkFalse(prepared.attrs.isTimeToLiveSet(), NO_TTLS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); + ModificationStatement prepared = parsed.prepare(state, bindVariables); + checkTrue(prepared.metadata().isAccordEnabled(), TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, prepared.type, prepared.source); + checkFalse(prepared.metadata().params.pendingDrop, TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, prepared.type, prepared.source); + checkFalse(prepared.hasConditions(), NO_CONDITIONS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); + checkFalse(prepared.isTimestampSet(), NO_TIMESTAMPS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); + checkFalse(prepared.attrs.isTimeToLiveSet(), NO_TTLS_IN_UPDATES_MESSAGE, prepared.type, prepared.source); - if (prepared.metadata().isCounter()) - throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, prepared.type, prepared.source); + if (prepared.metadata().isCounter()) + throw invalidRequest(NO_COUNTERS_IN_TXNS_MESSAGE, prepared.type, prepared.source); - preparedUpdates.add(prepared); + preparedUpdate.add(prepared); + } + + preparedUpdates.add(preparedUpdate); } - List preparedConditions = new ArrayList<>(conditions.size()); - for (ConditionStatement.Raw condition : conditions) - // TODO: If we eventually support IF ks.function(ref) THEN, the keyspace will have to be provided here - preparedConditions.add(condition.prepare("[txn]", bindVariables)); + List> preparedConditions = new ArrayList<>(conditions.size()); + for (int conditionIdx = 0; conditionIdx < conditions.size(); conditionIdx++) + { + List preparedCondition = new ArrayList<>(conditions.get(conditionIdx).size()); + for (ConditionStatement.Raw condition : conditions.get(conditionIdx)) + { + // TODO: If we eventually support IF ks.function(ref) THEN, the keyspace will have to be provided here + preparedCondition.add(condition.prepare("[txn]", bindVariables)); + } + + preparedConditions.add(preparedCondition); + } return new TransactionStatement(preparedAssignments, returningSelect, returningReferences, preparedUpdates, preparedConditions, bindVariables); } diff --git a/src/java/org/apache/cassandra/cql3/transactions/ConditionStatement.java b/src/java/org/apache/cassandra/cql3/transactions/ConditionStatement.java index 7f19d834d934..c1dc96ccac26 100644 --- a/src/java/org/apache/cassandra/cql3/transactions/ConditionStatement.java +++ b/src/java/org/apache/cassandra/cql3/transactions/ConditionStatement.java @@ -40,13 +40,14 @@ public enum Kind GT(TxnCondition.Kind.GREATER_THAN, TxnCondition.Kind.LESS_THAN), GTE(TxnCondition.Kind.GREATER_THAN_OR_EQUAL, TxnCondition.Kind.LESS_THAN_OR_EQUAL), LT(TxnCondition.Kind.LESS_THAN, TxnCondition.Kind.GREATER_THAN), - LTE(TxnCondition.Kind.LESS_THAN_OR_EQUAL, TxnCondition.Kind.GREATER_THAN_OR_EQUAL); - + LTE(TxnCondition.Kind.LESS_THAN_OR_EQUAL, TxnCondition.Kind.GREATER_THAN_OR_EQUAL), + ELSE(TxnCondition.Kind.ELSE, null); + // TODO: Support for IN, CONTAINS, CONTAINS KEY private final TxnCondition.Kind kind; private final TxnCondition.Kind reversedKind; - + Kind(TxnCondition.Kind kind, TxnCondition.Kind reversedKind) { this.kind = kind; @@ -80,8 +81,8 @@ public static class Raw public Raw(Term.Raw lhs, Kind kind, Term.Raw rhs) { - Preconditions.checkArgument(lhs != null); - Preconditions.checkArgument((rhs == null) == (kind == Kind.IS_NOT_NULL || kind == Kind.IS_NULL)); + Preconditions.checkArgument(lhs != null || kind == Kind.ELSE); + Preconditions.checkArgument((rhs == null) == (kind == Kind.IS_NOT_NULL || kind == Kind.IS_NULL || kind == Kind.ELSE)); this.lhs = lhs; this.kind = kind; this.rhs = rhs; @@ -89,6 +90,9 @@ public Raw(Term.Raw lhs, Kind kind, Term.Raw rhs) public ConditionStatement prepare(String keyspace, VariableSpecifications bindVariables) { + if (kind == Kind.ELSE) + return new ConditionStatement(null, kind, null, false); + if (rhs == null) { // In the IS NULL/IS NOT NULL case, the reference will always be on the LHS @@ -134,6 +138,8 @@ public TxnCondition createCondition(QueryOptions options) { switch (kind) { + case ELSE: + return TxnCondition.Else.instance; case IS_NOT_NULL: case IS_NULL: return new TxnCondition.Exists(reference.toTxnReference(options), kind.toTxnKind(reversed)); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java b/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java index 380566155ff7..846c86d90a68 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java @@ -117,7 +117,8 @@ public enum Kind GREATER_THAN_OR_EQUAL(">=", Operator.GTE), LESS_THAN("<", Operator.LT), LESS_THAN_OR_EQUAL("<=", Operator.LTE), - COLUMN_CONDITIONS("COLUMN_CONDITIONS", null); + COLUMN_CONDITIONS("COLUMN_CONDITIONS", null), + ELSE("ELSE", null); @Nonnull private final String symbol; @@ -152,6 +153,8 @@ private ConditionSerializer serializer() return None.serializer; case COLUMN_CONDITIONS: return ColumnConditionsAdapter.serializer; + case ELSE: + return Else.serializer; default: throw new IllegalArgumentException("No serializer exists for kind " + this); } @@ -231,6 +234,43 @@ public static TxnCondition none() return None.instance; } + public static class Else extends TxnCondition + { + public static final Else instance = new Else(); + + private Else() + { + super(Kind.ELSE); + } + + @Override + public String toString() + { + return kind.toString(); + } + + @Override + public void collect(TableMetadatas.Collector collector) + { + } + + @Override + public boolean applies(TxnData data) + { + return true; + } + + private static final ConditionSerializer serializer = new ConditionSerializer<>() + { + @Override + public void serialize(Else condition, TableMetadatas tables, DataOutputPlus out) {} + @Override + public Else deserialize(TableMetadatas tables, DataInputPlus in, Kind kind) { return instance; } + @Override + public long serializedSize(Else condition, TableMetadatas tables) { return 0; } + }; + } + public static class Exists extends TxnCondition { private static final Set KINDS = ImmutableSet.of(Kind.IS_NOT_NULL, Kind.IS_NULL); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index 720c28847a3c..7be96fb60ff7 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.Objects; import java.util.function.Function; +import java.util.stream.Collectors; import javax.annotation.Nullable; @@ -293,6 +294,11 @@ public long serializedSize(Block t, TableMetadatasAndKeys p) final BlockFragment[] fragments; final ConditionalBlock[] conditionalBlocks; + // Must ensure the following invariants when constructing a Block + // (1) Every fragment in fragments is sorted by key order + // (2) conditionalBlocks occurs in the same order as the IF/ELSE IF/ELSE statements as the order of the array + // determines the order in which the conditions get evaluated + // (3) Within each ConditionalBlock, fragmentIds is in sorted order Block(BlockFragment[] fragments, ConditionalBlock[] conditionalBlocks) { this.fragments = fragments; @@ -576,6 +582,105 @@ public TxnUpdate(TableMetadatas tables, List fragments, TxnCondition c this.preserveTimestamps = preserveTimestamps; } + + public TxnUpdate(TableMetadatas tables, TxnCondition[] conditions, List> conditionFragments, List noneConditionFragment, @Nullable ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) + { + requireArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL)); + this.tables = tables; + + List> sortedConditionFragments = conditionFragments.stream().sorted((l, r) -> Fragment.compareKeys(l.left(), r.left())).collect(Collectors.toList()); + List sortedNoneConditionFragment = noneConditionFragment.stream().sorted(Fragment::compareKeys).collect(Collectors.toList()); + List sortedAllFragments = mergeSortedFragments(sortedConditionFragments.stream().map(Pair::left).collect(Collectors.toList()), sortedNoneConditionFragment); + + this.keys = Keys.of(sortedAllFragments, fragment -> fragment.key); + + List blocks = new ArrayList<>(); + int nextConditionIndex = 0; + if (conditions.length != 0) + { + Pair pair = generateConditionBlock(conditions, sortedConditionFragments); + nextConditionIndex = pair.left(); + blocks.add(pair.right()); + } + + if (!noneConditionFragment.isEmpty()) + blocks.add(generateNoneConditionBlock(sortedNoneConditionFragment, nextConditionIndex)); + + this.blocks = blocks; + this.cassandraCommitCL = cassandraCommitCL; + this.preserveTimestamps = preserveTimestamps; + } + + private List mergeSortedFragments(List left, List right) + { + List fragments = new ArrayList<>(left.size() + right.size()); + int l = 0; + int r = 0; + + while (l < left.size() && r < right.size()) + { + if ((Fragment.compareKeys(left.get(l), right.get(r))) <= 0) + fragments.add(left.get(l++)); + else + fragments.add(right.get(r++)); + } + + while (l < left.size()) + fragments.add(left.get(l++)); + + while (r < right.size()) + fragments.add(right.get(r++)); + + return fragments; + } + + private Pair generateConditionBlock(TxnCondition[] conditions, List> sortedFragmentConditionIndexPair) + { + // Maps the index of each TxnCondition to it's list of fragmentIds + List> conditionsIndexToFragmentId = new ArrayList<>(conditions.length); + for (int i = 0; i < conditions.length; i++) + conditionsIndexToFragmentId.add(new ArrayList<>()); + + BlockFragment[] blockFragments = new BlockFragment[sortedFragmentConditionIndexPair.size()]; + ConditionalBlock[] conditionalBlocks = new ConditionalBlock[conditions.length]; + + int blockFragmentIdx = 0; + for (Pair pair : sortedFragmentConditionIndexPair) + { + Fragment fragment = pair.left; + int conditionsIndex = pair.right; + blockFragments[blockFragmentIdx] = new BlockFragment(blockFragmentIdx, fragment.key, Fragment.FragmentSerializer.serialize(fragment, tables, Version.LATEST)); + conditionsIndexToFragmentId.get(conditionsIndex).add(blockFragmentIdx); + blockFragmentIdx++; + } + + int conditionalBlocksIdx = 0; + + for (int i = 0; i < conditions.length; i++) + { + int[] fragmentIds = conditionsIndexToFragmentId.get(i).stream().mapToInt(x -> x).toArray(); + SerializedTxnCondition serializedCondition = new SerializedTxnCondition(conditions[i], tables); + conditionalBlocks[conditionalBlocksIdx] = new ConditionalBlock(conditionalBlocksIdx, serializedCondition, fragmentIds); + conditionalBlocksIdx++; + } + + return Pair.create(conditionalBlocksIdx, new Block(blockFragments, conditionalBlocks)); + } + + private Block generateNoneConditionBlock(List fragments, int conditionalBlockIndex) + { + BlockFragment[] blockFragments = new BlockFragment[fragments.size()]; + int[] fragmentIds = new int[fragments.size()]; + for (int i = 0 ; i < fragments.size() ; ++i) + { + blockFragments[i] = new BlockFragment(i, fragments.get(i).key, Fragment.FragmentSerializer.serialize(fragments.get(i), tables, Version.LATEST)); + fragmentIds[i] = i; + } + + SerializedTxnCondition serializedCondition = new SerializedTxnCondition(TxnCondition.none(), tables); + return new Block(blockFragments, new ConditionalBlock[] { new ConditionalBlock(conditionalBlockIndex, serializedCondition, fragmentIds) }); + } + private TxnUpdate(TableMetadatas tables, Keys keys, List blocks, ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) { this.tables = tables; From 4fa5ba0c2d0f157257eef5dbd18f4691d7b95e9f Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 2 Jul 2026 15:20:34 -0700 Subject: [PATCH 02/16] initial tests --- .../test/accord/AccordCQLTestBase.java | 244 ++++++++++++++++++ 1 file changed, 244 insertions(+) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index f38b84ede287..7898d44f40d5 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -301,6 +301,34 @@ public void testAcceptTransactionWithSameStaticColumnUpdatedWithInClause() throw }); } + @Test + public void testRejectTransactionWithUpdatesToSamePrimaryKeyInTrailingUpdate() throws Throwable + { + test(cluster -> { + try + { + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedAccordTableName + " (k, c, v) VALUES (?, ?, ?)"), ConsistencyLevel.ALL, 1, 1, 2); + String txn = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1 AND c = 1);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1 AND c = 2;\n" + + " ELSE \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 2 WHERE k = 1 AND c = 1;\n" + + " END IF \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 5 WHERE k = 1 AND c = 1;\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(txn, ConsistencyLevel.SERIAL); + fail("Expected exception"); + } + catch (Throwable t) + { + assertEquals(InvalidRequestException.class.getName(), t.getClass().getName()); + assertEquals(TransactionStatement.DUPLICATE_KEYS_IN_SAME_TRANSACTION_MESSAGE, t.getMessage()); + } + }); + } + @Test public void testCounterCreateTableTransactionalModeFails() throws Exception { @@ -3604,4 +3632,220 @@ public void userSeesInvalidRejection() throws Exception .hasMessage("Attempted to set an element on a list which is null"); }); } + + @Test + public void testElseIf() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (2, 3);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " LET k2 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k2.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " ELSE IF k2.v = 3 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 129 WHERE k = 1;\n" + + " ELSE \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 15 WHERE k = 1;\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 129); + }); + } + + @Test + public void testNoConditionsMatch() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k1.v != 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 2); + }); + } + + @Test + public void testRepeatedConditionChoosesFirstOneThatMatches() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " IF k1.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k1.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 10); + }); + } + + @Test + public void testChooseFirstConditionThatMatches() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k1.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " ELSE \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 15 WHERE k = 1;\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 127); + }); + } + + @Test + public void testElseIFWithTrailingUpdate() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (2, 3);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " LET k2 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k2.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " ELSE IF k2.v = 3 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 129 WHERE k = 1;\n" + + " ELSE \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 15 WHERE k = 1;\n" + + " END IF\n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 78 WHERE k = 2;\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 129); + + read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2;\n" + + "COMMIT TRANSACTION"; + + result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(2, 78); + }); + } + + @Test + public void testIfBlockWIthMultipleUpdateStatements() throws Throwable + { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String insert = "BEGIN TRANSACTION\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + + " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (2, 3);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + + String query = "BEGIN TRANSACTION\n" + + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + + " LET k2 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2);\n" + + " IF k1.v > 5 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + + " ELSE IF k2.v = 2 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + + " ELSE IF k2.v = 3 THEN \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 129 WHERE k = 1;\n" + + " ELSE \n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 15 WHERE k = 1;\n" + + " END IF\n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 78 WHERE k = 2;\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + + String read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + "COMMIT TRANSACTION"; + + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(1, 129); + + read = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2;\n" + + "COMMIT TRANSACTION"; + + result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); + assertThat(result).hasSize(1).contains(2, 78); + }); + } } From 9f66a8a6018ca3d37df656b14c58c1467d79a248 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 14:05:53 -0700 Subject: [PATCH 03/16] fix --- .../service/accord/txn/TxnUpdate.java | 22 +++++-------------- 1 file changed, 6 insertions(+), 16 deletions(-) diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index 7be96fb60ff7..17dddf10af8a 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -296,7 +296,7 @@ public long serializedSize(Block t, TableMetadatasAndKeys p) // Must ensure the following invariants when constructing a Block // (1) Every fragment in fragments is sorted by key order - // (2) conditionalBlocks occurs in the same order as the IF/ELSE IF/ELSE statements as the order of the array + // (2) conditionalBlocks occurs in the same order as the IF/ELSE IF/ELSE statements because the order of the array // determines the order in which the conditions get evaluated // (3) Within each ConditionalBlock, fragmentIds is in sorted order Block(BlockFragment[] fragments, ConditionalBlock[] conditionalBlocks) @@ -378,26 +378,16 @@ public Block select(Keys keys) ConditionalBlock[] outConditions; { - List collect = null; + List collect = new ArrayList<>(conditionalBlocks.length - 1); for (int i = 0 ; i < conditionalBlocks.length ; ++i) { ConditionalBlock cb = conditionalBlocks[i]; int[] cbOutFragmentIds = SortedArrays.linearIntersection(cb.fragmentIds, 0, cb.fragmentIds.length, outFragmentIds, 0, count, cachedInts()); - //noinspection ArrayEquality - if (cbOutFragmentIds != cb.fragmentIds) // when arrays are equal the cb.fragmentIds gets returned unchanged, so can do a pointer check to detect a change - { - if (collect == null) - { - collect = new ArrayList<>(conditionalBlocks.length - 1); - for (int j = 0 ; j < i ; ++j) //TODO (review): why do we include the previous blocks that "should" have empty fragments, but we provide them without empty fragments? - collect.add(conditionalBlocks[j]); - } - if (cbOutFragmentIds.length > 0) - collect.add(new ConditionalBlock(cb.id, cb.condition, cbOutFragmentIds)); - } + + if (cbOutFragmentIds.length > 0) + collect.add(new ConditionalBlock(cb.id, cb.condition, cbOutFragmentIds)); } - if (collect == null) outConditions = conditionalBlocks; - else if (collect.isEmpty()) outConditions = NO_CONDITIONAL_BLOCKS; + if (collect.isEmpty()) outConditions = NO_CONDITIONAL_BLOCKS; else outConditions = collect.toArray(ConditionalBlock[]::new); } From 82a0770dc7d8152b15ad35d39a08a3b9599914bb Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 14:09:25 -0700 Subject: [PATCH 04/16] add gitmodule --- .gitmodules | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitmodules b/.gitmodules index 616dacf610a7..073209d22a72 100644 --- a/.gitmodules +++ b/.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/alanwang67/cassandra-accord.git + branch = linearIntersectionBug From 42087cbb375e98df9cc880757a635e988c3491bc Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 16:11:23 -0700 Subject: [PATCH 05/16] update test --- .../service/accord/txn/TxnUpdateTest.java | 64 ++++++++++++++++++- 1 file changed, 62 insertions(+), 2 deletions(-) diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java index 6daad3627fc4..fc19812b6fbe 100644 --- a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java +++ b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java @@ -22,7 +22,12 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.NavigableSet; +import java.util.Set; +import java.util.TreeSet; +import java.util.stream.Collectors; import java.util.stream.IntStream; import org.junit.Test; @@ -35,6 +40,7 @@ import accord.utils.SortedArrays; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.cql3.ast.Conditional; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.partitions.PartitionUpdate; @@ -187,6 +193,37 @@ public void merge() }); } + @Test + public void select() + { + qt().check(rs -> { + Gen blockGen = block(); + Block block = blockGen.next(rs); + List allKeys = new ArrayList<>(); + for (TxnUpdate.BlockFragment fragment : block.fragments) + allKeys.add(fragment.key); + + if (!allKeys.isEmpty()) + { + // Get a random subset of the keys + Collections.shuffle(allKeys, rs.asJdkRandom()); + NavigableSet subListKey = new TreeSet<>(allKeys.subList(0, rs.nextInt(0, allKeys.size()))); + + Block selectedBlock = block.select(new Keys(subListKey)); + Set seenFragmentIds = new HashSet<>(); + for (ConditionalBlock conditionalBlock : selectedBlock.conditionalBlocks) + Arrays.stream(conditionalBlock.fragmentIds).forEach(seenFragmentIds::add); + + // Ensure that every fragment is referenced in a conditionalBlock's fragmentIds + for (TxnUpdate.BlockFragment fragment : selectedBlock.fragments) + assertThat(seenFragmentIds.remove(fragment.id)).isTrue(); + + // Ensure that there is a corresponding fragment for every id in conditionalBlock's fragmentIds + assertThat(seenFragmentIds.isEmpty()).isTrue(); + } + }); + } + private static Gen fragment(List tables) { return rs -> { @@ -223,8 +260,8 @@ private static Gen block() // can't have a empty block Gen bytesGen = Gens.arrays(ByteBuffer.class, TxnUpdateTest.bytesGen.filter(ByteBuffer::hasRemaining)) .ofSizeBetween(0, 10); - var conditionGen = Gens.arrays(ConditionalBlock.class, conditionalBlock()).ofSizeBetween(1, 10); Gen keyGen = (Gen) (Gen) AccordGenerators.keys(Murmur3Partitioner.instance); + Gen serializedTxnConditionGen = serializedTxnCondition(); return rs -> { ByteBuffer[] bbs = bytesGen.next(rs); Key[] keys = IntStream.range(0, bbs.length).mapToObj(i -> keyGen.next(rs)).toArray(Key[]::new); @@ -234,7 +271,30 @@ private static Gen block() TxnUpdate.BlockFragment[] fragments = new TxnUpdate.BlockFragment[bbs.length]; for (int i = 0 ; i < fragments.length ; ++i) fragments[i] = new TxnUpdate.BlockFragment(ids[i], (PartitionKey) keys[i], bbs[i]); - return new Block(fragments, conditionGen.next(rs)); + + List conditionalBlocks = new ArrayList<>(); + List fragmentIds = Arrays.stream(ids).boxed().collect(Collectors.toList()); + List sublist = new ArrayList<>(); + + boolean createNewSublist = false; + while (!fragmentIds.isEmpty()) + { + if (createNewSublist) + { + conditionalBlocks.add(new ConditionalBlock(rs.nextInt(-1, Integer.MAX_VALUE) + 1, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); + sublist = new ArrayList<>(); + } + + int index = rs.nextInt(0, fragmentIds.size()); + sublist.add(fragmentIds.get(index)); + fragmentIds.remove(index); + createNewSublist = rs.nextBoolean(); + } + + if (!sublist.isEmpty()) + conditionalBlocks.add(new ConditionalBlock(rs.nextInt(-1, Integer.MAX_VALUE) + 1, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); + + return new Block(fragments, conditionalBlocks.toArray(new ConditionalBlock[0])); }; } } \ No newline at end of file From bdde49b1ecb7c08b1b7c3028762e747b799234fa Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 16:17:06 -0700 Subject: [PATCH 06/16] update comment --- .../cassandra/cql3/statements/TransactionStatement.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 8c7df6dc53cd..f9994ce5f322 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -390,7 +390,6 @@ List> createWriteFragments(ClientState state, QueryOptio { List> groupedWriteFragments = new ArrayList<>(groupedUpdates.size()); - // This is an over approximation of all the possible seen columns HashMap totalSeenColumns = new HashMap<>(); int idx = 0; @@ -406,8 +405,10 @@ List> createWriteFragments(ClientState state, QueryOptio List writeFragments = modification.getTxnWriteFragment(idx, state, options, keyCollector); fragments.addAll(writeFragments); // TODO: When adding support for consecutive IF statements, we need to revisit CASSANDRA-21136, currently we perform this check per branch since only one branch can be executed - // In the case where we have a trailing update, we use an overapproximation of all the seen columns we - // have accumulated within each branch to perform the validation + // In the case we have a trailing update, we use an overapproximation of the total seen columns we + // have accumulated within each branch to perform the validation. This is because we do not know for certain + // at runtime which branch we will enter. This can produce false positives, but will never let us + // perform an unsafe query. if (groupedUpdatesIdx == groupedUpdates.size() - 1 && groupedUpdates.size() > groupedConditions.size()) validateOnlyModifyPrimaryKeyColumnPairOnce(totalSeenColumns, modification, writeFragments); else From f203e46d072a1259b00ff046b654135635c07222 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 16:20:05 -0700 Subject: [PATCH 07/16] update test --- .../test/accord/AccordCQLTestBase.java | 88 +++++++------------ 1 file changed, 30 insertions(+), 58 deletions(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index 7898d44f40d5..e5b204706fb2 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -87,6 +87,7 @@ import org.apache.cassandra.exceptions.WriteTimeoutException; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.AccordTestUtils; import org.apache.cassandra.service.consensus.TransactionalMode; @@ -3700,37 +3701,7 @@ public void testNoConditionsMatch() throws Throwable } @Test - public void testRepeatedConditionChoosesFirstOneThatMatches() throws Throwable - { - test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { - String insert = "BEGIN TRANSACTION\n" + - " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + - "COMMIT TRANSACTION"; - - cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); - - String query = "BEGIN TRANSACTION\n" + - " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + - " IF k1.v = 2 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + - " ELSE IF k1.v = 2 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + - " END IF\n" + - "COMMIT TRANSACTION"; - - cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); - - String read = "BEGIN TRANSACTION\n" + - " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + - "COMMIT TRANSACTION"; - - SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); - assertThat(result).hasSize(1).contains(1, 10); - }); - } - - @Test - public void testChooseFirstConditionThatMatches() throws Throwable + public void testMultipleMatchingConditions() throws Throwable { test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { String insert = "BEGIN TRANSACTION\n" + @@ -3762,7 +3733,7 @@ public void testChooseFirstConditionThatMatches() throws Throwable } @Test - public void testElseIFWithTrailingUpdate() throws Throwable + public void testTrailingUpdateIsAlwaysApplied() throws Throwable { test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { String insert = "BEGIN TRANSACTION\n" + @@ -3806,46 +3777,47 @@ public void testElseIFWithTrailingUpdate() throws Throwable } @Test - public void testIfBlockWIthMultipleUpdateStatements() throws Throwable + public void testUpdateTwoShardsWithinIfElseStatement() throws Throwable { - test("CREATE TABLE " + qualifiedAccordTableName + " (k int PRIMARY KEY, v int) WITH " + transactionalMode.asCqlParam(), cluster -> { + String keyspace = "ks" + System.currentTimeMillis(); + String currentTable = keyspace + ".tbl"; + List ddls = Arrays.asList("DROP KEYSPACE IF EXISTS " + keyspace + ";", + "CREATE KEYSPACE " + keyspace + " WITH REPLICATION={'class':'SimpleStrategy', 'replication_factor': 1}", + "CREATE TABLE " + currentTable + " (k blob PRIMARY KEY, v int) WITH transactional_mode='" + transactionalMode + "'"); + List keys = tokensToKeys(tokens()); + + test(ddls, cluster -> { String insert = "BEGIN TRANSACTION\n" + - " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (1, 2);\n" + - " INSERT INTO " + qualifiedAccordTableName + " (k, v) VALUES (2, 3);\n" + + " INSERT INTO " + currentTable + " (k, v) VALUES (?, 2);\n" + + " INSERT INTO " + currentTable + " (k, v) VALUES (?, 3);\n" + "COMMIT TRANSACTION"; - cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL); + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL, keys.get(0), keys.get(1)); String query = "BEGIN TRANSACTION\n" + - " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1);\n" + - " LET k2 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2);\n" + - " IF k1.v > 5 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1;\n" + - " ELSE IF k2.v = 2 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 127 WHERE k = 1;\n" + - " ELSE IF k2.v = 3 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 129 WHERE k = 1;\n" + + " LET k1 = (SELECT * FROM " + currentTable + " WHERE k = ?);\n" + + " LET k2 = (SELECT * FROM " + currentTable + " WHERE k = ?);\n" + + " IF k1.v = 2 THEN \n" + + " UPDATE " + currentTable + " SET v = 6 WHERE k = ?;\n" + + " UPDATE " + currentTable + " SET v = 5 WHERE k = ?;\n" + " ELSE \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 15 WHERE k = 1;\n" + + " UPDATE " + currentTable + " SET v = 15 WHERE k = ?;\n" + + " UPDATE " + currentTable + " SET v = 11 WHERE k = ?;\n" + " END IF\n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 78 WHERE k = 2;\n" + "COMMIT TRANSACTION"; - cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL); + cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.SERIAL, keys.get(0), keys.get(1), keys.get(0), keys.get(1), keys.get(0), keys.get(1)); String read = "BEGIN TRANSACTION\n" + - " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1;\n" + + " SELECT * FROM " + currentTable + " WHERE k = ?; \n" + "COMMIT TRANSACTION"; - SimpleQueryResult result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); - assertThat(result).hasSize(1).contains(1, 129); - - read = "BEGIN TRANSACTION\n" + - " SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 2;\n" + - "COMMIT TRANSACTION"; - - result = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL); - assertThat(result).hasSize(1).contains(2, 78); + int[] result = new int[] { 6, 5 }; + for (int i = 0; i < keys.size(); i++) + { + SimpleQueryResult qr = cluster.coordinator(1).executeWithResult(read, ConsistencyLevel.SERIAL, keys.get(i)); + assertThat(qr).hasSize(1).contains(keys.get(i), result[i]); + } }); } } From 8a44ab60001cb19a115a17904f7d9c0444c7e47c Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Tue, 7 Jul 2026 16:21:59 -0700 Subject: [PATCH 08/16] remove import --- .../cassandra/distributed/test/accord/AccordCQLTestBase.java | 1 - 1 file changed, 1 deletion(-) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index e5b204706fb2..d353122b11e1 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -87,7 +87,6 @@ import org.apache.cassandra.exceptions.WriteTimeoutException; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.schema.SchemaConstants; -import org.apache.cassandra.service.StorageService; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.AccordTestUtils; import org.apache.cassandra.service.consensus.TransactionalMode; From fa14de9a9f1747ec1e2ee2ccb055491115e744f5 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 9 Jul 2026 14:26:07 -0700 Subject: [PATCH 09/16] refactor --- .../cql3/statements/TransactionStatement.java | 18 +- .../service/accord/txn/TxnUpdate.java | 157 ++++++++++-------- 2 files changed, 100 insertions(+), 75 deletions(-) diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index f9994ce5f322..82019123a7ae 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -551,9 +551,9 @@ public Txn createTxn(ClientState state, QueryOptions options) { Int2ObjectHashMap autoReads = new Int2ObjectHashMap<>(); + List preTransformedBlocks = new ArrayList<>(); TxnCondition[] conditions = new TxnCondition[groupedConditions.size()]; - // Groups each fragment with an index corresponding to the TxnCondition it is associated with - List> conditionFragments = new ArrayList<>(); + List> fragmentConditionIndexPair = new ArrayList<>(); List> groupedFragments = createWriteFragments(state, options, autoReads, keyCollector); int idx = 0; @@ -562,16 +562,20 @@ public Txn createTxn(ClientState state, QueryOptions options) conditions[idx] = createCondition(groupedConditions.get(idx), options); List correspondingFragments = groupedFragments.get(idx); for (TxnWrite.Fragment fragment : correspondingFragments) - conditionFragments.add(Pair.create(fragment, idx)); + fragmentConditionIndexPair.add(Pair.create(fragment, idx)); idx++; } - List noneConditionFragments = new ArrayList<>(); + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(conditions, fragmentConditionIndexPair, null)); + + List noneConditions = new ArrayList<>(); if (idx < groupedFragments.size()) - noneConditionFragments.addAll(groupedFragments.get(idx)); + noneConditions.addAll(groupedFragments.get(idx)); + + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(new TxnCondition[]{ TxnCondition.none() }, null, noneConditions)); List reads = createNamedReads(options, autoReads, keyCollector); - if (conditionFragments.isEmpty() && noneConditionFragments.isEmpty()) // ModificationStatement yield no Mutation (DELETE WHERE pk=0 AND c < 0 AND c > 0 -- matches no keys; so has no mutation) + if (fragmentConditionIndexPair.isEmpty() && noneConditions.isEmpty()) // ModificationStatement yield no Mutation (DELETE WHERE pk=0 AND c < 0 AND c > 0 -- matches no keys; so has no mutation) { // cleanup memory keyCollector.clear(); @@ -581,7 +585,7 @@ public Txn createTxn(ClientState state, QueryOptions options) ConsistencyLevel commitCL = consistencyLevelForAccordCommit(cm, tables, keyCollector, options.getConsistency()); Keys keys = keyCollector.build(); - AccordUpdate update = new TxnUpdate(tables, conditions, conditionFragments, noneConditionFragments, commitCL, PreserveTimestamp.no); + AccordUpdate update = new TxnUpdate(tables, preTransformedBlocks, commitCL, PreserveTimestamp.no); TxnRead read = createTxnRead(tables, reads, null, Domain.Key); return new Txn.InMemory(keys, read, TxnQuery.ALL, update, new TableMetadatasAndKeys(tables, keys)); } diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index 17dddf10af8a..d10075aca56e 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -26,7 +26,6 @@ import java.util.List; import java.util.Objects; import java.util.function.Function; -import java.util.stream.Collectors; import javax.annotation.Nullable; @@ -254,7 +253,7 @@ public long estimatedSizeOnHeap() } } - static class Block + public static class Block { private static final long EMPTY_SIZE = ObjectSizes.measure(new Block(null, null)); public static final ParameterisedUnversionedSerializer serializer = new ParameterisedUnversionedSerializer<>() @@ -572,103 +571,125 @@ public TxnUpdate(TableMetadatas tables, List fragments, TxnCondition c this.preserveTimestamps = preserveTimestamps; } - - public TxnUpdate(TableMetadatas tables, TxnCondition[] conditions, List> conditionFragments, List noneConditionFragment, @Nullable ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) + public TxnUpdate(TableMetadatas tables, List preTransformedBlocks, @Nullable ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) { requireArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL)); this.tables = tables; - - List> sortedConditionFragments = conditionFragments.stream().sorted((l, r) -> Fragment.compareKeys(l.left(), r.left())).collect(Collectors.toList()); - List sortedNoneConditionFragment = noneConditionFragment.stream().sorted(Fragment::compareKeys).collect(Collectors.toList()); - List sortedAllFragments = mergeSortedFragments(sortedConditionFragments.stream().map(Pair::left).collect(Collectors.toList()), sortedNoneConditionFragment); - - this.keys = Keys.of(sortedAllFragments, fragment -> fragment.key); + this.keys = sortedFragmentKeys(preTransformedBlocks); List blocks = new ArrayList<>(); int nextConditionIndex = 0; - if (conditions.length != 0) + for (PreTransformedBlock preTransformedBlock : preTransformedBlocks) { - Pair pair = generateConditionBlock(conditions, sortedConditionFragments); + Pair pair = preTransformedBlock.generateBlock(nextConditionIndex, tables); nextConditionIndex = pair.left(); blocks.add(pair.right()); } - if (!noneConditionFragment.isEmpty()) - blocks.add(generateNoneConditionBlock(sortedNoneConditionFragment, nextConditionIndex)); - this.blocks = blocks; this.cassandraCommitCL = cassandraCommitCL; this.preserveTimestamps = preserveTimestamps; } - private List mergeSortedFragments(List left, List right) + public Keys sortedFragmentKeys(List preTransformedBlocks) { - List fragments = new ArrayList<>(left.size() + right.size()); - int l = 0; - int r = 0; + List keys = new ArrayList<>(); + for (PreTransformedBlock preTransformedBlock : preTransformedBlocks) + keys.addAll(preTransformedBlock.getKeys()); - while (l < left.size() && r < right.size()) - { - if ((Fragment.compareKeys(left.get(l), right.get(r))) <= 0) - fragments.add(left.get(l++)); - else - fragments.add(right.get(r++)); - } - - while (l < left.size()) - fragments.add(left.get(l++)); - - while (r < right.size()) - fragments.add(right.get(r++)); - - return fragments; + keys.sort(RoutableKey::compareTo); + return Keys.of(keys, k -> k); } - private Pair generateConditionBlock(TxnCondition[] conditions, List> sortedFragmentConditionIndexPair) + public static class PreTransformedBlock { - // Maps the index of each TxnCondition to it's list of fragmentIds - List> conditionsIndexToFragmentId = new ArrayList<>(conditions.length); - for (int i = 0; i < conditions.length; i++) - conditionsIndexToFragmentId.add(new ArrayList<>()); - - BlockFragment[] blockFragments = new BlockFragment[sortedFragmentConditionIndexPair.size()]; - ConditionalBlock[] conditionalBlocks = new ConditionalBlock[conditions.length]; + public TxnCondition[] conditions; + @Nullable + public List> fragmentConditionIndexPair; + @Nullable + public List noneConditions; - int blockFragmentIdx = 0; - for (Pair pair : sortedFragmentConditionIndexPair) + public PreTransformedBlock(TxnCondition[] conditions, List> fragmentConditionIndexPair, List noneConditions) { - Fragment fragment = pair.left; - int conditionsIndex = pair.right; - blockFragments[blockFragmentIdx] = new BlockFragment(blockFragmentIdx, fragment.key, Fragment.FragmentSerializer.serialize(fragment, tables, Version.LATEST)); - conditionsIndexToFragmentId.get(conditionsIndex).add(blockFragmentIdx); - blockFragmentIdx++; + Invariants.require((fragmentConditionIndexPair == null && noneConditions != null) || (fragmentConditionIndexPair != null && noneConditions == null)); + this.conditions = conditions; + if (fragmentConditionIndexPair != null) + fragmentConditionIndexPair.sort((l, r) -> TxnWrite.Fragment.compareKeys(l.left(), r.left())); + this.fragmentConditionIndexPair = fragmentConditionIndexPair; + if (noneConditions != null) + noneConditions.sort(Fragment::compareKeys); + this.noneConditions = noneConditions; } - int conditionalBlocksIdx = 0; - - for (int i = 0; i < conditions.length; i++) + private boolean isTrailingUpdate() { - int[] fragmentIds = conditionsIndexToFragmentId.get(i).stream().mapToInt(x -> x).toArray(); - SerializedTxnCondition serializedCondition = new SerializedTxnCondition(conditions[i], tables); - conditionalBlocks[conditionalBlocksIdx] = new ConditionalBlock(conditionalBlocksIdx, serializedCondition, fragmentIds); - conditionalBlocksIdx++; + return noneConditions != null; } - return Pair.create(conditionalBlocksIdx, new Block(blockFragments, conditionalBlocks)); - } - - private Block generateNoneConditionBlock(List fragments, int conditionalBlockIndex) - { - BlockFragment[] blockFragments = new BlockFragment[fragments.size()]; - int[] fragmentIds = new int[fragments.size()]; - for (int i = 0 ; i < fragments.size() ; ++i) + public List getKeys() { - blockFragments[i] = new BlockFragment(i, fragments.get(i).key, Fragment.FragmentSerializer.serialize(fragments.get(i), tables, Version.LATEST)); - fragmentIds[i] = i; + List keys = new ArrayList<>(); + if (isTrailingUpdate()) + { + for (int i = 0; i < noneConditions.size(); i++) + keys.add(noneConditions.get(i).key); + } + else { + for (int i = 0; i < fragmentConditionIndexPair.size(); i++) + keys.add(fragmentConditionIndexPair.get(i).left().key); + } + return keys; } - SerializedTxnCondition serializedCondition = new SerializedTxnCondition(TxnCondition.none(), tables); - return new Block(blockFragments, new ConditionalBlock[] { new ConditionalBlock(conditionalBlockIndex, serializedCondition, fragmentIds) }); + public Pair generateBlock(int conditionalBlockIndex, TableMetadatas tables) + { + if (isTrailingUpdate()) + { + BlockFragment[] blockFragments = new BlockFragment[noneConditions.size()]; + int[] fragmentIds = new int[noneConditions.size()]; + for (int i = 0 ; i < noneConditions.size() ; ++i) + { + Fragment fragment = noneConditions.get(i); + blockFragments[i] = new BlockFragment(i, fragment.key, Fragment.FragmentSerializer.serialize(fragment, tables, Version.LATEST)); + fragmentIds[i] = i; + } + + SerializedTxnCondition serializedCondition = new SerializedTxnCondition(TxnCondition.none(), tables); + ConditionalBlock[] conditionalBlock = new ConditionalBlock[] { new ConditionalBlock(conditionalBlockIndex++, serializedCondition, fragmentIds) }; + return Pair.create(conditionalBlockIndex, new Block(blockFragments, conditionalBlock)); + } + else + { + List> conditionsIndexToFragmentId = new ArrayList<>(conditions.length); + for (int i = 0; i < conditions.length; i++) + conditionsIndexToFragmentId.add(new ArrayList<>()); + + BlockFragment[] blockFragments = new BlockFragment[fragmentConditionIndexPair.size()]; + ConditionalBlock[] conditionalBlocks = new ConditionalBlock[conditions.length]; + + int blockFragmentIdx = 0; + for (Pair pair : fragmentConditionIndexPair) + { + Fragment fragment = pair.left; + int conditionsIndex = pair.right; + blockFragments[blockFragmentIdx] = new BlockFragment(blockFragmentIdx, fragment.key, Fragment.FragmentSerializer.serialize(fragment, tables, Version.LATEST)); + conditionsIndexToFragmentId.get(conditionsIndex).add(blockFragmentIdx); + blockFragmentIdx++; + } + + int conditionalBlocksIdx = 0; + + for (int i = 0; i < conditions.length; i++) + { + int[] fragmentIds = conditionsIndexToFragmentId.get(i).stream().mapToInt(x -> x).toArray(); + SerializedTxnCondition serializedCondition = new SerializedTxnCondition(conditions[i], tables); + conditionalBlocks[conditionalBlocksIdx] = new ConditionalBlock(conditionalBlocksIdx, serializedCondition, fragmentIds); + conditionalBlocksIdx++; + } + + return Pair.create(conditionalBlocksIdx, new Block(blockFragments, conditionalBlocks)); + } + } } private TxnUpdate(TableMetadatas tables, Keys keys, List blocks, ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) From 3b931ab5fe80ba7e09b0c74dc66afaf3bb6740f0 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 9 Jul 2026 15:57:23 -0700 Subject: [PATCH 10/16] more tests --- .../service/accord/txn/TxnUpdateTest.java | 156 ++++++++++++++++-- 1 file changed, 143 insertions(+), 13 deletions(-) diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java index fc19812b6fbe..627d8334c5bd 100644 --- a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java +++ b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java @@ -40,7 +40,6 @@ import accord.utils.SortedArrays; import org.apache.cassandra.config.DatabaseDescriptor; -import org.apache.cassandra.cql3.ast.Conditional; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.marshal.BytesType; import org.apache.cassandra.db.partitions.PartitionUpdate; @@ -62,6 +61,7 @@ import org.apache.cassandra.utils.AccordGenerators; import org.apache.cassandra.utils.CassandraGenerators; import org.apache.cassandra.utils.Generators; +import org.apache.cassandra.utils.Pair; import static accord.utils.Property.qt; import static accord.utils.SortedArrays.Search.FAST; @@ -111,6 +111,71 @@ public void blockSerde() }); } + @Test + public void blockInvariants() + { + qt().check(rs -> { + List tables = tablesGen.next(rs); + TableMetadatas metadatas = TableMetadatas.of(tables); + List preTransformedBlocks = new ArrayList<>(); + + // We use TxnCondition.Else.instance because, the invariants that we are checking do not depend on the kind of TxnCondition + TxnCondition[] conditions = IntStream.range(0, rs.nextInt(1, 10)).mapToObj(i -> TxnCondition.Else.instance).toArray(TxnCondition[]::new); + List> fragmentConditionIndexPair = new ArrayList<>(); + List fragments = Gens.lists(fragment(tables)).ofSizeBetween(conditions.length, conditions.length + 15).next(rs); + + // Make sure that all conditions have a randomly assigned corresponding fragment to it + List fragmentToConditionIndex = IntStream.range(0, conditions.length).boxed().collect(Collectors.toList()); + Collections.shuffle(fragmentToConditionIndex); + + for (int i = 0; i < fragments.size(); i++) + { + if (i < fragmentToConditionIndex.size()) + fragmentConditionIndexPair.add(Pair.create(fragments.get(i), fragmentToConditionIndex.get(i))); + else + // After we have ensured that each condition is referenced by 1 fragment, + // assign the rest of the fragments to random conditions + fragmentConditionIndexPair.add(Pair.create(fragments.get(i), rs.nextInt(0, conditions.length))); + } + + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(conditions, fragmentConditionIndexPair, null)); + + TxnCondition[] trailingCondition = new TxnCondition[] { TxnCondition.none() }; + List noneCondition = new ArrayList<>(); + boolean shouldHaveTrailingUpdate = rs.nextBoolean(); + if (shouldHaveTrailingUpdate) + { + noneCondition = Gens.lists(fragment(tables)).ofSizeBetween(conditions.length, conditions.length + 15).next(rs); + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(trailingCondition, null, noneCondition)); + } + + int blockSize = 0; + if (!fragmentConditionIndexPair.isEmpty()) + blockSize++; + if (!noneCondition.isEmpty()) + blockSize++; + + TxnUpdate update = new TxnUpdate(metadatas, preTransformedBlocks, null, PreserveTimestamp.no); + + List blocks = update.blocks; + assertThat(blocks.size()).isEqualTo(blockSize); + + for (int blockIdx = 0; blockIdx < blocks.size(); blockIdx++) + { + Block block = blocks.get(blockIdx); + assertThat(ensureFragmentIdsAreOrdered(block)).isTrue(); + assertThat(ensureBlockFragmentsAreSortedByKey(block)).isTrue(); + assertThat(ensureInjectivityOfFragmentIdsToFragments(block)).isTrue(); + + boolean inTrailingUpdateBlock = shouldHaveTrailingUpdate && (blockIdx == blocks.size() - 1); + if (!inTrailingUpdateBlock) + assertThat(ensureConditionalBlockIdsPreserveOrder(block, conditions, metadatas)).isTrue(); + else + assertThat(ensureConditionalBlockIdsPreserveOrder(block, trailingCondition, metadatas)).isTrue(); + } + }); + } + @Test public void slice() { @@ -210,20 +275,84 @@ public void select() NavigableSet subListKey = new TreeSet<>(allKeys.subList(0, rs.nextInt(0, allKeys.size()))); Block selectedBlock = block.select(new Keys(subListKey)); - Set seenFragmentIds = new HashSet<>(); - for (ConditionalBlock conditionalBlock : selectedBlock.conditionalBlocks) - Arrays.stream(conditionalBlock.fragmentIds).forEach(seenFragmentIds::add); - - // Ensure that every fragment is referenced in a conditionalBlock's fragmentIds - for (TxnUpdate.BlockFragment fragment : selectedBlock.fragments) - assertThat(seenFragmentIds.remove(fragment.id)).isTrue(); - - // Ensure that there is a corresponding fragment for every id in conditionalBlock's fragmentIds - assertThat(seenFragmentIds.isEmpty()).isTrue(); + assertThat(ensureFragmentIdsAreOrdered(selectedBlock)).isTrue(); + assertThat(ensureConditionalBlockIdsPreserveOrder(selectedBlock, block)).isTrue(); + assertThat(ensureBlockFragmentsAreSortedByKey(selectedBlock)).isTrue(); + assertThat(ensureInjectivityOfFragmentIdsToFragments(selectedBlock)).isTrue(); } }); } + private boolean ensureFragmentIdsAreOrdered(Block block) + { + for (ConditionalBlock conditionalBlock : block.conditionalBlocks) + { + for (int i = 1; i < conditionalBlock.fragmentIds.length; i++) + if (conditionalBlock.fragmentIds[i-1] > conditionalBlock.fragmentIds[i]) + return false; + } + + return true; + } + + private boolean ensureConditionalBlockIdsPreserveOrder(Block selectedBlock, Block originalBlock) + { + List originalBlockOrder = Arrays.stream(originalBlock.conditionalBlocks).map(i -> i.id).collect(Collectors.toList()); + List selectedBlockOrder = Arrays.stream(selectedBlock.conditionalBlocks).map(i -> i.id).collect(Collectors.toList()); + + int selectedIndex = 0; + int originalIndex = 0; + while (selectedIndex < selectedBlockOrder.size() && originalIndex < originalBlockOrder.size()) + { + if (!originalBlockOrder.get(originalIndex).equals(selectedBlockOrder.get(selectedIndex))) + originalIndex++; + else + selectedIndex++; + } + + return selectedIndex == selectedBlockOrder.size(); + } + + private boolean ensureConditionalBlockIdsPreserveOrder(Block block, TxnCondition[] conditions, TableMetadatas metadatas) + { + if (block.conditionalBlocks.length != conditions.length) + return false; + + for (int i = 0; i < block.conditionalBlocks.length; i++) + { + TxnCondition txnCondition = block.conditionalBlocks[i].condition.deserialize(metadatas); + if (!txnCondition.equals(conditions[i])) + return false; + } + + return true; + } + + private boolean ensureBlockFragmentsAreSortedByKey(Block block) + { + for (int i = 1; i < block.fragments.length; i++) + if (block.fragments[i-1].key.compareTo(block.fragments[i].key) > 0) + return false; + return true; + } + + private boolean ensureInjectivityOfFragmentIdsToFragments(Block block) + { + Set seenFragmentIds = new HashSet<>(); + for (ConditionalBlock conditionalBlock : block.conditionalBlocks) + { + for (int fragmentId : conditionalBlock.fragmentIds) + if (!seenFragmentIds.add(fragmentId)) + return false; + } + + for (TxnUpdate.BlockFragment fragment : block.fragments) + if (!seenFragmentIds.remove(fragment.id)) + return false; + + return seenFragmentIds.isEmpty(); + } + private static Gen fragment(List tables) { return rs -> { @@ -277,11 +406,12 @@ private static Gen block() List sublist = new ArrayList<>(); boolean createNewSublist = false; + int conditionalBlockIdx = 0; while (!fragmentIds.isEmpty()) { if (createNewSublist) { - conditionalBlocks.add(new ConditionalBlock(rs.nextInt(-1, Integer.MAX_VALUE) + 1, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); + conditionalBlocks.add(new ConditionalBlock(conditionalBlockIdx++, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); sublist = new ArrayList<>(); } @@ -292,7 +422,7 @@ private static Gen block() } if (!sublist.isEmpty()) - conditionalBlocks.add(new ConditionalBlock(rs.nextInt(-1, Integer.MAX_VALUE) + 1, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); + conditionalBlocks.add(new ConditionalBlock(conditionalBlockIdx, serializedTxnConditionGen.next(rs), sublist.stream().sorted().mapToInt(i->i).toArray())); return new Block(fragments, conditionalBlocks.toArray(new ConditionalBlock[0])); }; From 56a2c86732dbc0d5dc307646fe220ff2b538b7a1 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 9 Jul 2026 16:22:26 -0700 Subject: [PATCH 11/16] remove alias in parser --- src/antlr/Parser.g | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g index f3458281f5d0..6585e31cafd3 100644 --- a/src/antlr/Parser.g +++ b/src/antlr/Parser.g @@ -29,8 +29,6 @@ options { // enables parsing txn specific syntax when true protected boolean isParsingTxn = false; - // tracks whether a txn has conditional updates - protected boolean isTxnConditional = false; protected List references; @@ -785,16 +783,14 @@ batchTxnStatement returns [TransactionStatement.Parsed expr] List returning = null; List> conditions = new ArrayList<>(); List> updates = new ArrayList<>(); - List e = new ArrayList<>(); - e.add(new ConditionStatement.Raw(null, ConditionStatement.Kind.ELSE, null)); } : K_BEGIN K_TRANSACTION ( let=letStatement ';' { assignments.add(let); })* ( ( (selectStatement) => s=selectStatement ';' { select = s; }) | ( K_SELECT drs=rowDataReferences ';' { returning = drs; }) )? ( - K_IF c=txnConditions K_THEN { isTxnConditional = true; } u=updateStatements { conditions.add(c); updates.add(u); } + K_IF c=txnConditions K_THEN u=updateStatements { conditions.add(c); updates.add(u); } ( K_ELSE K_IF c=txnConditions K_THEN u=updateStatements { conditions.add(c); updates.add(u); } )* - ( K_ELSE u=updateStatements { conditions.add(e); updates.add(u); } )? + ( K_ELSE u=updateStatements { conditions.add(Collections.singletonList(new ConditionStatement.Raw(null, ConditionStatement.Kind.ELSE, null))); updates.add(u); } )? K_END K_IF )? ( (batchStatementObjective) => u=updateStatements { updates.add(u); } )? From 9ae9105cf7ad97359983437dc31c5da5bac199c4 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Thu, 9 Jul 2026 21:55:45 -0700 Subject: [PATCH 12/16] update test --- .../cassandra/cql3/statements/TransactionStatement.java | 3 ++- .../distributed/test/accord/AccordCQLTestBase.java | 8 ++++---- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index 82019123a7ae..a73a198073bf 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -414,7 +414,8 @@ List> createWriteFragments(ClientState state, QueryOptio else { validateOnlyModifyPrimaryKeyColumnPairOnce(seenColumns, modification, writeFragments); - totalSeenColumns.putAll(seenColumns); + for (Map.Entry entry : seenColumns.entrySet()) + totalSeenColumns.merge(entry.getKey(), entry.getValue(), Columns::mergeTo); } if (modification.allReferenceOperations().stream().anyMatch(ReferenceOperation::requiresRead)) diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java index d353122b11e1..83d508983db9 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -304,16 +304,16 @@ public void testAcceptTransactionWithSameStaticColumnUpdatedWithInClause() throw @Test public void testRejectTransactionWithUpdatesToSamePrimaryKeyInTrailingUpdate() throws Throwable { - test(cluster -> { + test("CREATE TABLE " + qualifiedAccordTableName + " (k int, c int, v int, z int, primary key (k, c)) WITH " + transactionalMode.asCqlParam(), cluster -> { try { - cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedAccordTableName + " (k, c, v) VALUES (?, ?, ?)"), ConsistencyLevel.ALL, 1, 1, 2); + cluster.coordinator(1).execute(wrapInTxn("INSERT INTO " + qualifiedAccordTableName + " (k, c, v, z) VALUES (?, ?, ?, ?)"), ConsistencyLevel.ALL, 1, 1, 1, 1); String txn = "BEGIN TRANSACTION\n" + " LET k1 = (SELECT * FROM " + qualifiedAccordTableName + " WHERE k = 1 AND c = 1);\n" + " IF k1.v > 5 THEN \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 10 WHERE k = 1 AND c = 2;\n" + + " UPDATE " + qualifiedAccordTableName + " SET v = 1 WHERE k = 1 AND c = 1;\n" + " ELSE \n" + - " UPDATE " + qualifiedAccordTableName + " SET v = 2 WHERE k = 1 AND c = 1;\n" + + " UPDATE " + qualifiedAccordTableName + " SET z = 3 WHERE k = 1 AND c = 1;\n" + " END IF \n" + " UPDATE " + qualifiedAccordTableName + " SET v = 5 WHERE k = 1 AND c = 1;\n" + "COMMIT TRANSACTION"; From a712a6d7fca217ddb79a0f697989b8c2ea0e0649 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Fri, 10 Jul 2026 13:29:39 -0700 Subject: [PATCH 13/16] updates --- src/antlr/Parser.g | 2 +- .../cql3/statements/TransactionStatement.java | 69 ++++++++++--------- 2 files changed, 37 insertions(+), 34 deletions(-) diff --git a/src/antlr/Parser.g b/src/antlr/Parser.g index 6585e31cafd3..95786f5e0e53 100644 --- a/src/antlr/Parser.g +++ b/src/antlr/Parser.g @@ -793,7 +793,7 @@ batchTxnStatement returns [TransactionStatement.Parsed expr] ( K_ELSE u=updateStatements { conditions.add(Collections.singletonList(new ConditionStatement.Raw(null, ConditionStatement.Kind.ELSE, null))); updates.add(u); } )? K_END K_IF )? - ( (batchStatementObjective) => u=updateStatements { updates.add(u); } )? + ( (updateStatements) => u=updateStatements { updates.add(u); } )? (K_COMMIT K_TRANSACTION) { $expr = new TransactionStatement.Parsed(assignments, select, returning, updates, conditions, references); diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index a73a198073bf..069caa05627a 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -167,14 +167,16 @@ public NamedSelect(int name, SelectStatement select) private long minEpoch = Epoch.EMPTY.getEpoch(); - // Every index of List corresponds with an index of groupedConditions. - // If the size of groupedUpdates is larger than the size of groupedConditions this means that - // we have updates that are not part of a condition + // For each index of groupedConditions, it's corresponding update is the same + // index in groupedUpdates. We maintain the invariant that the size of + // groupedUpdates is equal to the size of groupedConditions or the size of + // groupedUpdates is one greater than the size of groupedConditions + // in the case of trailing updates. public TransactionStatement(List assignments, NamedSelect returningSelect, List returningReferences, List> groupedUpdates, - List> groupedConditions, // Each List represents the condition in IF, ELSE IF, ELSE + List> groupedConditions, VariableSpecifications bindVariables) { this.assignments = assignments; @@ -390,6 +392,10 @@ List> createWriteFragments(ClientState state, QueryOptio { List> groupedWriteFragments = new ArrayList<>(groupedUpdates.size()); + // When we have a trailing update, we use the total seen columns we + // have accumulated within each branch to perform the validation. + // This is because we do not know for certain at runtime which branch we will enter. + // This can produce false positives, but will never let us perform an unsafe query. HashMap totalSeenColumns = new HashMap<>(); int idx = 0; @@ -397,26 +403,26 @@ List> createWriteFragments(ClientState state, QueryOptio { List updates = groupedUpdates.get(groupedUpdatesIdx); HashMap seenColumns = new HashMap<>(); - List fragments = new ArrayList<>(updates.size()); + for (ModificationStatement modification : updates) { minEpoch = Math.max(minEpoch, modification.metadata().epoch.getEpoch()); List writeFragments = modification.getTxnWriteFragment(idx, state, options, keyCollector); fragments.addAll(writeFragments); + boolean containsTrailingUpdate = groupedUpdates.size() > groupedConditions.size(); + // TODO: When adding support for consecutive IF statements, we need to revisit CASSANDRA-21136, currently we perform this check per branch since only one branch can be executed - // In the case we have a trailing update, we use an overapproximation of the total seen columns we - // have accumulated within each branch to perform the validation. This is because we do not know for certain - // at runtime which branch we will enter. This can produce false positives, but will never let us - // perform an unsafe query. - if (groupedUpdatesIdx == groupedUpdates.size() - 1 && groupedUpdates.size() > groupedConditions.size()) + if (groupedUpdatesIdx == groupedUpdates.size() - 1 && containsTrailingUpdate) validateOnlyModifyPrimaryKeyColumnPairOnce(totalSeenColumns, modification, writeFragments); - else + else if (containsTrailingUpdate) { validateOnlyModifyPrimaryKeyColumnPairOnce(seenColumns, modification, writeFragments); for (Map.Entry entry : seenColumns.entrySet()) totalSeenColumns.merge(entry.getKey(), entry.getValue(), Columns::mergeTo); } + else + validateOnlyModifyPrimaryKeyColumnPairOnce(seenColumns, modification, writeFragments); if (modification.allReferenceOperations().stream().anyMatch(ReferenceOperation::requiresRead)) { @@ -775,29 +781,29 @@ public static class Parsed extends QualifiedStatement.Composite private final List assignments; private final SelectStatement.RawStatement select; private final List returning; - private final List> updates; - private final List> conditions; + private final List> groupedUpdates; + private final List> groupedConditions; private final List dataReferences; public Parsed(List assignments, SelectStatement.RawStatement select, List returning, - List> updates, - List> conditions, + List> groupedUpdates, + List> groupedConditions, List dataReferences) { this.assignments = assignments; this.select = select; this.returning = returning; - this.updates = updates; - this.conditions = conditions != null ? conditions : Collections.emptyList(); + this.groupedUpdates = groupedUpdates; + this.groupedConditions = groupedConditions; this.dataReferences = dataReferences; } @Override protected Iterable getStatements() { - Iterable group = Iterables.concat(assignments, updates.stream().flatMap(Collection::stream).collect(Collectors.toList())); + Iterable group = Iterables.concat(assignments, groupedUpdates.stream().flatMap(Collection::stream).collect(Collectors.toList())); if (select != null) group = Iterables.concat(group, Collections.singleton(select)); return group; @@ -806,8 +812,8 @@ protected Iterable getStatements() @Override public CQLStatement prepare(ClientState state) { - checkFalse(updates.isEmpty() && returning == null && select == null, EMPTY_TRANSACTION_MESSAGE); - Invariants.require(updates.size() == conditions.size() + 1 || updates.size() == conditions.size()); + checkFalse(groupedUpdates.isEmpty() && returning == null && select == null, EMPTY_TRANSACTION_MESSAGE); + Invariants.require(groupedUpdates.size() == groupedConditions.size() + 1 || groupedUpdates.size() == groupedConditions.size()); if (select != null || returning != null) checkTrue(select != null ^ returning != null, "Cannot specify both a full SELECT and a SELECT w/ LET references."); @@ -859,15 +865,14 @@ public CQLStatement prepare(ClientState state) .collect(Collectors.toList()); } - List> preparedUpdates = new ArrayList<>(updates.size()); - for (int updateIdx = 0; updateIdx < updates.size(); updateIdx++) { - List preparedUpdate = new ArrayList<>(updates.get(updateIdx).size()); + List> preparedUpdates = new ArrayList<>(groupedUpdates.size()); + for (List update : groupedUpdates) + { + List preparedUpdate = new ArrayList<>(update.size()); // check for any read-before-write updates - for (int i = 0; i < updates.get(updateIdx).size(); i++) + for (ModificationStatement.Parsed parsed : update) { - ModificationStatement.Parsed parsed = updates.get(updateIdx).get(i); - ModificationStatement prepared = parsed.prepare(state, bindVariables); checkTrue(prepared.metadata().isAccordEnabled(), TRANSACTIONS_DISABLED_ON_TABLE_MESSAGE, prepared.type, prepared.source); checkFalse(prepared.metadata().params.pendingDrop, TRANSACTIONS_DISABLED_ON_TABLE_BEING_DROPPED_MESSAGE, prepared.type, prepared.source); @@ -884,15 +889,13 @@ public CQLStatement prepare(ClientState state) preparedUpdates.add(preparedUpdate); } - List> preparedConditions = new ArrayList<>(conditions.size()); - for (int conditionIdx = 0; conditionIdx < conditions.size(); conditionIdx++) + List> preparedConditions = new ArrayList<>(groupedConditions.size()); + for (List condition : groupedConditions) { - List preparedCondition = new ArrayList<>(conditions.get(conditionIdx).size()); - for (ConditionStatement.Raw condition : conditions.get(conditionIdx)) - { + List preparedCondition = new ArrayList<>(condition.size()); + for (ConditionStatement.Raw raw : condition) // TODO: If we eventually support IF ks.function(ref) THEN, the keyspace will have to be provided here - preparedCondition.add(condition.prepare("[txn]", bindVariables)); - } + preparedCondition.add(raw.prepare("[txn]", bindVariables)); preparedConditions.add(preparedCondition); } From 277300bf8702583173df02c3943f3123ac9181f5 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Fri, 10 Jul 2026 13:30:15 -0700 Subject: [PATCH 14/16] module --- modules/accord | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/accord b/modules/accord index 66cab49f1c9a..4e2d55b96725 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 66cab49f1c9af471795339a5fdf9522e6e73b691 +Subproject commit 4e2d55b96725cd49e44aa831ab83cd98a2eb013b From 89bad72cce41863c6e54a3119823d67955970e08 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Fri, 10 Jul 2026 14:44:19 -0700 Subject: [PATCH 15/16] inlined if --- modules/accord | 2 +- .../apache/cassandra/service/accord/txn/TxnUpdateTest.java | 5 +---- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/modules/accord b/modules/accord index 4e2d55b96725..a111e0f1b42c 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 4e2d55b96725cd49e44aa831ab83cd98a2eb013b +Subproject commit a111e0f1b42c8d0f8fe480dd5b435eec803ea8e8 diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java index 627d8334c5bd..b1da722ae927 100644 --- a/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java +++ b/test/unit/org/apache/cassandra/service/accord/txn/TxnUpdateTest.java @@ -168,10 +168,7 @@ public void blockInvariants() assertThat(ensureInjectivityOfFragmentIdsToFragments(block)).isTrue(); boolean inTrailingUpdateBlock = shouldHaveTrailingUpdate && (blockIdx == blocks.size() - 1); - if (!inTrailingUpdateBlock) - assertThat(ensureConditionalBlockIdsPreserveOrder(block, conditions, metadatas)).isTrue(); - else - assertThat(ensureConditionalBlockIdsPreserveOrder(block, trailingCondition, metadatas)).isTrue(); + assertThat(ensureConditionalBlockIdsPreserveOrder(block, inTrailingUpdateBlock ? trailingCondition : conditions, metadatas)).isTrue(); } }); } From ffb04a396f09b8b8083b50f60482f2a34832a871 Mon Sep 17 00:00:00 2001 From: Alan Wang Date: Mon, 13 Jul 2026 13:46:45 -0700 Subject: [PATCH 16/16] bump module --- modules/accord | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/accord b/modules/accord index a111e0f1b42c..46816be4a630 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit a111e0f1b42c8d0f8fe480dd5b435eec803ea8e8 +Subproject commit 46816be4a6301b83dee0b791b182a1d9fc710256