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 diff --git a/modules/accord b/modules/accord index 66cab49f1c9a..46816be4a630 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 66cab49f1c9af471795339a5fdf9522e6e73b691 +Subproject commit 46816be4a6301b83dee0b791b182a1d9fc710256 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..95786f5e0e53 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; @@ -783,20 +781,33 @@ 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<>(); } : 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 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(Collections.singletonList(new ConditionStatement.Raw(null, ConditionStatement.Kind.ELSE, null))); updates.add(u); } )? + K_END K_IF + )? + ( (updateStatements) => 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..069caa05627a 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,34 @@ 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(); + // 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 updates, - List conditions, + List> groupedUpdates, + List> groupedConditions, 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 +343,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 +388,56 @@ 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()); + + // 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<>(); - 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<>(); + List fragments = new ArrayList<>(updates.size()); - if (modification.allReferenceOperations().stream().anyMatch(ReferenceOperation::requiresRead)) + 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); + 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 + if (groupedUpdatesIdx == groupedUpdates.size() - 1 && containsTrailingUpdate) + validateOnlyModifyPrimaryKeyColumnPairOnce(totalSeenColumns, modification, writeFragments); + 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); - idx++; + 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++; + } + groupedWriteFragments.add(fragments); } - return fragments; + + return groupedWriteFragments; } private static void validateOnlyModifyPrimaryKeyColumnPairOnce(HashMap seenColumns, @@ -520,9 +557,32 @@ public Txn createTxn(ClientState state, QueryOptions options) else { Int2ObjectHashMap autoReads = new Int2ObjectHashMap<>(); - List writeFragments = createWriteFragments(state, options, autoReads, keyCollector); + + List preTransformedBlocks = new ArrayList<>(); + TxnCondition[] conditions = new TxnCondition[groupedConditions.size()]; + List> fragmentConditionIndexPair = 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) + fragmentConditionIndexPair.add(Pair.create(fragment, idx)); + idx++; + } + + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(conditions, fragmentConditionIndexPair, null)); + + List noneConditions = new ArrayList<>(); + if (idx < groupedFragments.size()) + noneConditions.addAll(groupedFragments.get(idx)); + + preTransformedBlocks.add(new TxnUpdate.PreTransformedBlock(new TxnCondition[]{ TxnCondition.none() }, null, noneConditions)); + 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 (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(); @@ -531,7 +591,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, 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)); } @@ -720,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); + 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; @@ -751,8 +812,8 @@ protected Iterable getStatements() @Override public CQLStatement prepare(ClientState state) { - checkFalse(updates.isEmpty() && returning == null && select == null, EMPTY_TRANSACTION_MESSAGE); - + 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."); @@ -804,30 +865,40 @@ 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++) + List> preparedUpdates = new ArrayList<>(groupedUpdates.size()); + for (List update : groupedUpdates) { - ModificationStatement.Parsed parsed = updates.get(i); + List preparedUpdate = new ArrayList<>(update.size()); - 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); + // check for any read-before-write updates + for (ModificationStatement.Parsed parsed : update) + { + 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); + preparedUpdate.add(prepared); + } - preparedUpdates.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<>(groupedConditions.size()); + for (List condition : groupedConditions) + { + 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(raw.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..d10075aca56e 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -253,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<>() @@ -293,6 +293,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 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) { this.fragments = fragments; @@ -372,26 +377,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); } @@ -576,6 +571,127 @@ public TxnUpdate(TableMetadatas tables, List fragments, TxnCondition c this.preserveTimestamps = 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; + this.keys = sortedFragmentKeys(preTransformedBlocks); + + List blocks = new ArrayList<>(); + int nextConditionIndex = 0; + for (PreTransformedBlock preTransformedBlock : preTransformedBlocks) + { + Pair pair = preTransformedBlock.generateBlock(nextConditionIndex, tables); + nextConditionIndex = pair.left(); + blocks.add(pair.right()); + } + + this.blocks = blocks; + this.cassandraCommitCL = cassandraCommitCL; + this.preserveTimestamps = preserveTimestamps; + } + + public Keys sortedFragmentKeys(List preTransformedBlocks) + { + List keys = new ArrayList<>(); + for (PreTransformedBlock preTransformedBlock : preTransformedBlocks) + keys.addAll(preTransformedBlock.getKeys()); + + keys.sort(RoutableKey::compareTo); + return Keys.of(keys, k -> k); + } + + public static class PreTransformedBlock + { + public TxnCondition[] conditions; + @Nullable + public List> fragmentConditionIndexPair; + @Nullable + public List noneConditions; + + public PreTransformedBlock(TxnCondition[] conditions, List> fragmentConditionIndexPair, List noneConditions) + { + 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; + } + + private boolean isTrailingUpdate() + { + return noneConditions != null; + } + + public List getKeys() + { + 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; + } + + 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) { this.tables = tables; 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..83d508983db9 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("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, 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 = 1 WHERE k = 1 AND c = 1;\n" + + " ELSE \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"; + + 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,191 @@ 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 testMultipleMatchingConditions() 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 testTrailingUpdateIsAlwaysApplied() 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 testUpdateTwoShardsWithinIfElseStatement() throws Throwable + { + 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 " + currentTable + " (k, v) VALUES (?, 2);\n" + + " INSERT INTO " + currentTable + " (k, v) VALUES (?, 3);\n" + + "COMMIT TRANSACTION"; + + cluster.coordinator(1).executeWithResult(insert, ConsistencyLevel.SERIAL, keys.get(0), keys.get(1)); + + String query = "BEGIN TRANSACTION\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 " + currentTable + " SET v = 15 WHERE k = ?;\n" + + " UPDATE " + currentTable + " SET v = 11 WHERE k = ?;\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + + 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 " + currentTable + " WHERE k = ?; \n" + + "COMMIT TRANSACTION"; + + 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]); + } + }); + } } 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..b1da722ae927 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; @@ -56,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; @@ -105,6 +111,68 @@ 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); + assertThat(ensureConditionalBlockIdsPreserveOrder(block, inTrailingUpdateBlock ? trailingCondition : conditions, metadatas)).isTrue(); + } + }); + } + @Test public void slice() { @@ -187,6 +255,101 @@ 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)); + 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 -> { @@ -223,8 +386,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 +397,31 @@ 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; + int conditionalBlockIdx = 0; + while (!fragmentIds.isEmpty()) + { + if (createNewSublist) + { + conditionalBlocks.add(new ConditionalBlock(conditionalBlockIdx++, 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(conditionalBlockIdx, 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