diff --git a/CHANGES.txt b/CHANGES.txt index 9cac450533..ac47a545f3 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.1 + * Accord: write rejections would be returned to users as server errors rather than INVALID and TxnReferenceOperation didn't handle all collections prperly (CASSANDRA-21061) * Use byte[] directly in QueryOptions instead of ByteBuffer and convert them to ArrayCell instead of BufferCell to reduce allocations (CASSANDRA-20166) * Log queries scanning too many SSTables per read (CASSANDRA-21048) * Extend nodetool verify to (optionally) validate SAI files (CASSANDRA-20949) diff --git a/modules/accord b/modules/accord index 6bae51f6a4..f6b0a6998f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 6bae51f6a4bd560a82840fa0809cd1d630696cc0 +Subproject commit f6b0a6998faca767e6951976097dec704c306b0e diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java index 080a705ffa..9cc2b94c66 100644 --- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java +++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java @@ -74,6 +74,7 @@ import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.service.accord.txn.TxnReference; import org.apache.cassandra.service.accord.txn.TxnResult; import org.apache.cassandra.service.accord.txn.TxnUpdate; +import org.apache.cassandra.service.accord.txn.TxnValidationRejection; import org.apache.cassandra.service.accord.txn.TxnWrite; import org.apache.cassandra.service.paxos.Ballot; import org.apache.cassandra.tcm.ClusterMetadata; @@ -574,6 +575,7 @@ public class CQL3CasRequest implements CASRequest { if (txnResult.kind() == retry_new_protocol) return RETRY_NEW_PROTOCOL; + TxnValidationRejection.maybeThrow(txnResult); TxnData txnData = (TxnData)txnResult; TxnDataKeyValue partition = (TxnDataKeyValue)txnData.get(txnDataName(CAS_READ)); return casResult(partition != null ? partition.rowIterator(false) : null); diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java index caff467cd7..97482d994d 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java @@ -85,6 +85,7 @@ import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.service.accord.txn.TxnReference; import org.apache.cassandra.service.accord.txn.TxnResult; import org.apache.cassandra.service.accord.txn.TxnUpdate; +import org.apache.cassandra.service.accord.txn.TxnValidationRejection; import org.apache.cassandra.service.accord.txn.TxnWrite; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode; @@ -250,7 +251,7 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement, SinglePartitionReadQuery.Group selectQuery = (SinglePartitionReadQuery.Group) select.getQuery(options, 0); if (selectQuery.queries.size() != 1) - throw new IllegalArgumentException("Within a transaction, SELECT statements must select a single partition; found " + selectQuery.queries.size() + " partitions"); + throw invalidRequest("Within a transaction, SELECT statements must select a single partition; found " + selectQuery.queries.size() + " partitions"); SinglePartitionReadCommand command = Iterables.getOnlyElement(selectQuery.queries); return new TxnNamedRead(namedSelect.name, keyCollector.collect(command.metadata(), command.partitionKey()), command, keyCollector.tables); @@ -560,6 +561,8 @@ public class TransactionStatement implements CQLStatement.CompositeCQLStatement, TxnResult txnResult = AccordService.instance().coordinate(minEpoch, txn, options.getConsistency(), requestTime); if (txnResult.kind() == retry_new_protocol) throw new InvalidRequestException(UNSUPPORTED_MIGRATION); + TxnValidationRejection.maybeThrow(txnResult); + TxnData data = (TxnData)txnResult; if (returningSelect != null) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 40b3d1fe82..b1ac0bfffa 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -148,6 +148,7 @@ import org.apache.cassandra.service.accord.txn.TxnQuery; import org.apache.cassandra.service.accord.txn.TxnRangeReadResult; import org.apache.cassandra.service.accord.txn.TxnRead; import org.apache.cassandra.service.accord.txn.TxnResult; +import org.apache.cassandra.service.accord.txn.TxnValidationRejection; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.service.consensus.UnsupportedTransactionConsistencyLevel; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationMutationHelper.SplitConsumer; @@ -1316,13 +1317,16 @@ public class StorageProxy implements StorageProxyMBean { if (accordResult != null) { - TxnResult.Kind kind = accordResult.awaitAndGet().kind(); + TxnResult result = accordResult.awaitAndGet(); + TxnResult.Kind kind = result.kind(); if (kind == retry_new_protocol && failure == null) { Tracing.trace("Accord returned retry new protocol"); logger.debug("Retrying mutations on different system because some mutations were misrouted according to Accord"); continue; } + TxnValidationRejection.maybeThrow(result); + Tracing.trace("Successfully wrote Accord mutations"); } } @@ -1549,9 +1553,11 @@ public class StorageProxy implements StorageProxyMBean // the batch log. if (accordResult != null) { - TxnResult.Kind kind = accordResult.awaitAndGet().kind(); + TxnResult result = accordResult.awaitAndGet(); + TxnResult.Kind kind = result.kind(); if (kind == retry_new_protocol && failure == null) continue; + TxnValidationRejection.maybeThrow(result); Tracing.trace("Successfully wrote Accord mutations"); cleanup.ackMutation(); } diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnDataValue.java b/src/java/org/apache/cassandra/service/accord/txn/TxnDataValue.java index 3fc9c08198..cacafcd616 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnDataValue.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnDataValue.java @@ -42,6 +42,7 @@ public interface TxnDataValue { key(0), range(1); + int id; Kind(int id) @@ -49,14 +50,15 @@ public interface TxnDataValue this.id = id; } - public TxnDataValueSerializer serializer() + @SuppressWarnings("unchecked") + public IVersionedSerializer serializer() { switch (this) { case key: - return TxnDataKeyValue.serializer; + return (IVersionedSerializer) TxnDataKeyValue.serializer; case range: - return TxnDataRangeValue.serializer; + return (IVersionedSerializer) TxnDataRangeValue.serializer; default: throw new IllegalStateException("Unrecognized kind " + this); } @@ -71,7 +73,7 @@ public interface TxnDataValue long estimatedSizeOnHeap(); - IVersionedSerializer serializer = new IVersionedSerializer() + IVersionedSerializer serializer = new IVersionedSerializer<>() { @SuppressWarnings("unchecked") @Override diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java b/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java index e380b87fe6..191518e34a 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java @@ -154,6 +154,9 @@ public abstract class TxnQuery implements Query @Override public Result compute(TxnId txnId, Timestamp executeAt, Seekables keys, @Nullable Data data, @Nullable Read read, @Nullable Update update) { + TxnValidationRejection rejection = TxnUpdate.validationRejection(update); + if (rejection != null) return rejection; + // Skip the migration checks in the base class for empty transactions, we don't // want/need the RetryWithNewProtocolResult return new TxnData(); @@ -207,6 +210,9 @@ public abstract class TxnQuery implements Query @Override public Result compute(TxnId txnId, Timestamp executeAt, Seekables keys, @Nullable Data data, @Nullable Read read, @Nullable Update update) { + TxnValidationRejection rejection = TxnUpdate.validationRejection(update); + if (rejection != null) return rejection; + // TODO (required): This is not the cluster metadata of the current transaction ClusterMetadata clusterMetadata = ClusterMetadata.current(); checkState(clusterMetadata.epoch.getEpoch() >= executeAt.epoch(), "TCM epoch %d is < executeAt epoch %d", clusterMetadata.epoch.getEpoch(), executeAt.epoch()); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnReferenceOperation.java b/src/java/org/apache/cassandra/service/accord/txn/TxnReferenceOperation.java index c453f612b2..f997008210 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnReferenceOperation.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnReferenceOperation.java @@ -24,6 +24,10 @@ import java.util.HashMap; import java.util.Map; import java.util.Objects; +import javax.annotation.Nullable; + +import com.google.common.annotations.VisibleForTesting; + import org.apache.cassandra.cql3.FieldIdentifier; import org.apache.cassandra.cql3.Operation; import org.apache.cassandra.cql3.UpdateParameters; @@ -35,8 +39,10 @@ import org.apache.cassandra.cql3.terms.Sets; import org.apache.cassandra.cql3.terms.Term; import org.apache.cassandra.cql3.terms.UserTypes; import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.marshal.CollectionType; +import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.MapType; import org.apache.cassandra.db.marshal.SetType; @@ -58,25 +64,29 @@ import static org.apache.cassandra.service.accord.AccordSerializers.columnMetada public class TxnReferenceOperation { private static final Map, Kind> operationKindMap = initOperationKindMap(); - - private static Map, Kind> initOperationKindMap() + + @VisibleForTesting + static Map, Kind> initOperationKindMap() { Map, Kind> temp = new HashMap<>(); - temp.put(Sets.Adder.class, Kind.SetAdder); temp.put(Constants.Adder.class, Kind.ConstantAdder); - temp.put(Lists.Appender.class, Kind.ListAppender); - temp.put(Sets.Discarder.class, Kind.SetDiscarder); - temp.put(Lists.Discarder.class, Kind.ListDiscarder); - temp.put(Lists.Prepender.class, Kind.ListPrepender); - temp.put(Maps.Putter.class, Kind.MapPutter); - temp.put(Lists.Setter.class, Kind.ListSetter); - temp.put(Sets.Setter.class, Kind.SetSetter); - temp.put(Maps.Setter.class, Kind.MapSetter); - temp.put(UserTypes.Setter.class, Kind.UserTypeSetter); temp.put(Constants.Setter.class, Kind.ConstantSetter); temp.put(Constants.Substracter.class, Kind.ConstantSubtracter); - temp.put(Maps.SetterByKey.class, Kind.MapSetterByKey); + temp.put(Lists.Appender.class, Kind.ListAppender); + temp.put(Lists.Discarder.class, Kind.ListDiscarder); + temp.put(Lists.DiscarderByIndex.class, Kind.ListDiscarderByIndex); + temp.put(Lists.Prepender.class, Kind.ListPrepender); + temp.put(Lists.Setter.class, Kind.ListSetter); temp.put(Lists.SetterByIndex.class, Kind.ListSetterByIndex); + temp.put(Maps.DiscarderByKey.class, Kind.MapDiscarderByKey); + temp.put(Maps.Putter.class, Kind.MapPutter); + temp.put(Maps.Setter.class, Kind.MapSetter); + temp.put(Maps.SetterByKey.class, Kind.MapSetterByKey); + temp.put(Sets.Adder.class, Kind.SetAdder); + temp.put(Sets.Discarder.class, Kind.SetDiscarder); + temp.put(Sets.ElementDiscarder.class, Kind.SetElementDiscarder); + temp.put(Sets.Setter.class, Kind.SetSetter); + temp.put(UserTypes.Setter.class, Kind.UserTypeSetter); temp.put(UserTypes.SetterByField.class, Kind.UserTypeSetterByField); return temp; } @@ -103,7 +113,10 @@ public class TxnReferenceOperation ConstantSubtracter((byte) 13, (column, keyOrIndex, field, value) -> new Constants.Substracter(column, value)), MapSetterByKey((byte) 14, (column, keyOrIndex, field, value) -> new Maps.SetterByKey(column, keyOrIndex, value)), ListSetterByIndex((byte) 15, (column, keyOrIndex, field, value) -> new Lists.SetterByIndex(column, keyOrIndex, value)), - UserTypeSetterByField((byte) 16, (column, keyOrIndex, field, value) -> new UserTypes.SetterByField(column, field, value)); + UserTypeSetterByField((byte) 16, (column, keyOrIndex, field, value) -> new UserTypes.SetterByField(column, field, value)), + ListDiscarderByIndex((byte) 17, (column, keyOrIndex, field, value) -> new Lists.DiscarderByIndex(column, value)), + MapDiscarderByKey((byte) 18, (column, keyOrIndex, field, value) -> new Maps.DiscarderByKey(column, value)), + SetElementDiscarder((byte) 19, (column, keyOrIndex, field, value) -> new Sets.ElementDiscarder(column, value)); private final byte id; private final ToOperation toOperation; @@ -154,18 +167,20 @@ public class TxnReferenceOperation private final Kind kind; private final ColumnMetadata receiver; - private final TableMetadata table; - private final ByteBuffer key; - private final ByteBuffer field; + public final TableMetadata table; + private final @Nullable ByteBuffer keyOrIndex; + private final @Nullable ByteBuffer field; private final TxnReferenceValue value; + private final @Nullable AbstractType keyOrIndexType; private final AbstractType valueType; - public TxnReferenceOperation(Kind kind, ColumnMetadata receiver, TableMetadata table, ByteBuffer key, ByteBuffer field, TxnReferenceValue value) + public TxnReferenceOperation(Kind kind, ColumnMetadata receiver, TableMetadata table, + @Nullable ByteBuffer keyOrIndex, @Nullable ByteBuffer field, TxnReferenceValue value) { this.kind = kind; this.receiver = receiver; this.table = table; - this.key = key; + this.keyOrIndex = keyOrIndex; this.field = field; // We don't expect operators on clustering keys, but unwrap just in case. @@ -175,20 +190,36 @@ public class TxnReferenceOperation { // The value for a map subtraction is actually a set (see Operation.Substraction) this.valueType = SetType.getInstance(((MapType) receiverType).getKeysType(), true); + this.keyOrIndexType = null; + } + else if (kind == Kind.MapDiscarderByKey || kind == Kind.SetElementDiscarder) + { + CollectionType ct = (CollectionType) receiverType; + this.keyOrIndexType = null; + this.valueType = ct.nameComparator(); } else if (kind == Kind.MapSetterByKey || kind == Kind.ListSetterByIndex) { - this.valueType = ((CollectionType) receiverType).valueComparator(); + CollectionType ct = (CollectionType) receiverType; + this.keyOrIndexType = ct.nameComparator(); + this.valueType = ct.valueComparator(); + } + else if (kind == Kind.ListDiscarderByIndex) + { + this.valueType = Int32Type.instance; + this.keyOrIndexType = null; } else if (kind == Kind.UserTypeSetterByField) { UserType userType = (UserType) receiverType; CellPath fieldPath = userType.cellPathForField(new FieldIdentifier(field)); this.valueType = userType.fieldType(fieldPath); + this.keyOrIndexType = null; } else { this.valueType = receiverType; + this.keyOrIndexType = null; } this.value = value; @@ -202,29 +233,37 @@ public class TxnReferenceOperation TxnReferenceOperation that = (TxnReferenceOperation) o; return Objects.equals(receiver, that.receiver) && kind == that.kind - && Objects.equals(key, that.key) + && Objects.equals(keyOrIndex, that.keyOrIndex) && Objects.equals(field, that.field) && Objects.equals(value, that.value); } + @Override + public int hashCode() + { + return Objects.hash(receiver, kind, keyOrIndex, field, value); + } + + + @Override + public String toString() + { + return "TxnReferenceOperation{" + + "kind=" + kind + + ", receiver=" + receiver + + ", table=" + table + + ", key=" + keyOrIndex + + ", field=" + field + + ", value=" + value + + '}'; + } + public void collect(TableMetadatas.Collector collector) { collector.add(table); value.collect(collector); } - @Override - public int hashCode() - { - return Objects.hash(receiver, kind, key, field, value); - } - - @Override - public String toString() - { - return receiver + " = " + value; - } - public ColumnMetadata receiver() { return receiver; @@ -236,11 +275,12 @@ public class TxnReferenceOperation operation.execute(key, up); } - private Operation toOperation(TxnData data) + @VisibleForTesting + Operation toOperation(TxnData data) { FieldIdentifier fieldIdentifier = field == null ? null : new FieldIdentifier(field); Term valueTerm = toTerm(data, valueType); - Term keyorIndexTerm = key == null ? null : toTerm(key, valueType); + Term keyorIndexTerm = keyOrIndex == null ? null : toTerm(keyOrIndex, keyOrIndexType); return kind.toOperation(receiver, keyorIndexTerm, fieldIdentifier, valueTerm); } @@ -274,9 +314,9 @@ public class TxnReferenceOperation columnMetadataSerializer.serialize(operation.receiver, operation.table, out); TxnReferenceValue.serializer.serialize(operation.value, tables, out); - out.writeBoolean(operation.key != null); - if (operation.key != null) - ByteBufferUtil.writeWithVIntLength(operation.key, out); + out.writeBoolean(operation.keyOrIndex != null); + if (operation.keyOrIndex != null) + ByteBufferUtil.writeWithVIntLength(operation.keyOrIndex, out); out.writeBoolean(operation.field != null); if (operation.field != null) @@ -303,9 +343,11 @@ public class TxnReferenceOperation size += columnMetadataSerializer.serializedSize(operation.receiver, operation.table); size += TxnReferenceValue.serializer.serializedSize(operation.value, tables); - if (operation.key != null) - size += ByteBufferUtil.serializedSizeWithVIntLength(operation.key); + size += TypeSizes.sizeof(operation.keyOrIndex != null); + if (operation.keyOrIndex != null) + size += ByteBufferUtil.serializedSizeWithVIntLength(operation.keyOrIndex); + size += TypeSizes.sizeof(operation.field != null); if (operation.field != null) size += ByteBufferUtil.serializedSizeWithVIntLength(operation.field); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnResult.java b/src/java/org/apache/cassandra/service/accord/txn/TxnResult.java index 2b5af7c08b..b3bf2b3fc8 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnResult.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnResult.java @@ -26,7 +26,8 @@ public interface TxnResult extends Result { txn_data(0), retry_new_protocol(1), - range_read(2); + range_read(2), + validation_rejection(3); int id; 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 d4b549a61d..0d3def4fd0 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -45,6 +45,7 @@ import accord.utils.SortedArrays; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.exceptions.RequestValidationException; import org.apache.cassandra.io.ParameterisedUnversionedSerializer; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.util.DataInputBuffer; @@ -85,7 +86,7 @@ import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize; -public class TxnUpdate extends AccordUpdate +public final class TxnUpdate extends AccordUpdate { static class ConditionalBlock { @@ -547,6 +548,9 @@ public class TxnUpdate extends AccordUpdate // Memoize computation of condition private Boolean anyConditionResult; + @Nullable + private transient volatile Object validationException = this; + public TxnUpdate(TableMetadatas tables, List fragments, TxnCondition condition, @Nullable ConsistencyLevel cassandraCommitCL, PreserveTimestamp preserveTimestamps) { requireArgument(cassandraCommitCL == null || IAccordService.SUPPORTED_COMMIT_CONSISTENCY_LEVELS.contains(cassandraCommitCL)); @@ -584,6 +588,25 @@ public class TxnUpdate extends AccordUpdate return new TxnUpdate(TableMetadatas.none(), Keys.EMPTY, Collections.emptyList(), null, PreserveTimestamp.no); } + public static TxnValidationRejection validationRejection(@Nullable Update update) + { + if (update != null && update.getClass() == TxnUpdate.class) + { + RequestValidationException e = ((TxnUpdate) update).validationRejection(); + if (e != null) return new TxnValidationRejection(e); + } + return null; + } + + @Nullable + private RequestValidationException validationRejection() + { + Object snapshot = validationException; + if (snapshot == this) + throw Invariants.illegalState("Attempted to check for validation exception before .apply was called"); + return snapshot == null ? null : (RequestValidationException) snapshot; + } + @Override public long estimatedSizeOnHeap() { @@ -677,7 +700,18 @@ public class TxnUpdate extends AccordUpdate ClusterMetadata cm = ClusterMetadata.current(); checkState(cm.epoch.getEpoch() >= executeAt.epoch(), "TCM epoch %d is < executeAt epoch %d", cm.epoch.getEpoch(), executeAt.epoch()); - Pair, SimpleBitSet> pair = processCondition(executeAt, data); + Pair, SimpleBitSet> pair = null; + try + { + validationException = null; + pair = processCondition(executeAt, data); + } + catch (RequestValidationException e) + { + // the update isn't allowed + validationException = e; + } + if (pair == null) return new TxnWrite(TableMetadatas.none(), Collections.emptyList(), SimpleBitSets.allUnset(numConditionalBlocks())); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnValidationRejection.java b/src/java/org/apache/cassandra/service/accord/txn/TxnValidationRejection.java new file mode 100644 index 0000000000..aca6fbfda3 --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnValidationRejection.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.accord.txn; + +import java.util.Objects; + +import org.apache.cassandra.exceptions.RequestValidationException; + +import static org.apache.cassandra.service.accord.txn.TxnResult.Kind.validation_rejection; + +public class TxnValidationRejection implements TxnResult +{ + private final RequestValidationException validationException; + + public TxnValidationRejection(RequestValidationException validationException) + { + this.validationException = Objects.requireNonNull(validationException); + } + + public static void maybeThrow(TxnResult txnResult) + { + if (txnResult.kind() == validation_rejection) + throw ((TxnValidationRejection) txnResult).validationException; + } + + @Override + public Kind kind() + { + return Kind.validation_rejection; + } + + @Override + public long estimatedSizeOnHeap() + { + return 0; + } +} 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 bdee5765ac..f3c7fceede 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTestBase.java @@ -3416,4 +3416,20 @@ public abstract class AccordCQLTestBase extends AccordTestBase } }); } + + @Test + public void userSeesInvalidRejection() throws Exception + { + var expectedType = AssertionUtils.isInstanceof(InvalidRequestException.class); + test("CREATE TABLE " + qualifiedAccordTableName + "(k int PRIMARY KEY, l list) WITH " + transactionalMode.asCqlParam(), cluster -> { + String cql = "UPDATE " + qualifiedAccordTableName + " SET l[0] = 42 WHERE k=42"; + Assertions.assertThatThrownBy(() -> cluster.coordinator(1).execute(cql, QUORUM)) + .is(expectedType) + .hasMessage("Attempted to set an element on a list which is null"); + + Assertions.assertThatThrownBy(() -> cluster.coordinator(1).execute(wrapInTxn(cql), QUORUM)) + .is(expectedType) + .hasMessage("Attempted to set an element on a list which is null"); + }); + } } diff --git a/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java b/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java index 383109573c..1c9c361c4b 100644 --- a/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java +++ b/test/harry/main/org/apache/cassandra/harry/model/BytesPartitionState.java @@ -348,7 +348,7 @@ public class BytesPartitionState sb.append(partition); sb.append(", clustering="); if (clustering == null) sb.append("null"); - else appendValues(sb, factory.clusteringColumns, clustering); + else appendValues(sb, clustering == Clustering.STATIC_CLUSTERING ? List.of() : factory.clusteringColumns, clustering); sb.append(')'); return sb.toString(); } diff --git a/test/harry/main/org/apache/cassandra/harry/model/PartitionState.java b/test/harry/main/org/apache/cassandra/harry/model/PartitionState.java index 57ec2ad4af..6c426e8fcc 100644 --- a/test/harry/main/org/apache/cassandra/harry/model/PartitionState.java +++ b/test/harry/main/org/apache/cassandra/harry/model/PartitionState.java @@ -410,6 +410,12 @@ public class PartitionState implements Iterable ", lts(" + StringUtils.toString(lts) + ")"; } } + + @Override + public String toString() + { + return toString(partitionState.valueGenerators); + } } public static long[] arr(int length, long fill) diff --git a/test/unit/org/apache/cassandra/cql3/ast/CollectionAccess.java b/test/unit/org/apache/cassandra/cql3/ast/CollectionAccess.java index d43d7bf34f..fca80da516 100644 --- a/test/unit/org/apache/cassandra/cql3/ast/CollectionAccess.java +++ b/test/unit/org/apache/cassandra/cql3/ast/CollectionAccess.java @@ -35,6 +35,18 @@ public class CollectionAccess implements ReferenceExpression this.type = type; } + @Override + public ReferenceExpression visit(Visitor v) + { + var u = v.visit(this); + if (u != this) return u; + var col = column.visit(v); + var val = element.visit(v); + if (col != column || val != element) + return new CollectionAccess(col, val, type); + return this; + } + @Override public void toCQL(StringBuilder sb, CQLFormatter formatter) { diff --git a/test/unit/org/apache/cassandra/service/accord/txn/TxnReferenceOperationTest.java b/test/unit/org/apache/cassandra/service/accord/txn/TxnReferenceOperationTest.java new file mode 100644 index 0000000000..5515221815 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/txn/TxnReferenceOperationTest.java @@ -0,0 +1,289 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.service.accord.txn; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Set; + +import javax.annotation.Nullable; + +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Sets; +import org.junit.Test; + +import accord.utils.Gen; +import org.apache.cassandra.cql3.ColumnIdentifier; +import org.apache.cassandra.cql3.FieldIdentifier; +import org.apache.cassandra.cql3.Operation; +import org.apache.cassandra.cql3.terms.Constants; +import org.apache.cassandra.cql3.terms.UserTypes; +import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.marshal.CollectionType; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.ListType; +import org.apache.cassandra.db.marshal.MapType; +import org.apache.cassandra.db.marshal.SetType; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.db.marshal.UserType; +import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.io.Serializers; +import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.schema.ColumnMetadata; +import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.accord.serializers.TableMetadatas; +import org.apache.cassandra.utils.AbstractTypeGenerators; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.Generators; +import org.reflections.Reflections; +import org.reflections.scanners.Scanners; +import org.reflections.util.ConfigurationBuilder; + +import static accord.utils.Property.qt; + +public class TxnReferenceOperationTest +{ + private static final String KS = "ks"; + private static final TxnData EMPTY = new TxnData(); + + @Test + public void coverage() + { + Reflections reflections = new Reflections(new ConfigurationBuilder() + .forPackage("org.apache.cassandra") + .setScanners(Scanners.SubTypes) + .setExpandSuperTypes(true) + .setParallel(true)); + var subTypes = reflections.getSubTypesOf(Operation.class); + var knownTypes = TxnReferenceOperation.initOperationKindMap().keySet(); + // these types do not have a way to define a reference, as they are column level: + Set> safeToExclude = ImmutableSet.of(Constants.Deleter.class, // DELETE foo + UserTypes.DeleterByField.class // DELETE foo.bar + ); + + StringBuilder sb = null; + for (var klass : Sets.difference(subTypes, Sets.union(knownTypes, safeToExclude))) + { + if (sb == null) + sb = new StringBuilder(); + sb.append(klass.getCanonicalName()).append('\n'); + } + if (sb != null) + throw new AssertionError(sb.toString()); + } + + @Test + public void serde() + { + @SuppressWarnings({ "resource", "IOResourceOpenedButNotSafelyClosed" }) DataOutputBuffer output = new DataOutputBuffer(); + qt().withExamples(10_000).forAll(gen()).check(txnOp -> { + Serializers.testSerde(output, TxnReferenceOperation.serializer, txnOp, TableMetadatas.of(txnOp.table)); + + txnOp.toOperation(EMPTY); + }); + } + + private enum Group + { + Setter, SetterByIndex, SetterByKey, SetterByField, + Adder, Subtracter, + Appender, Putter, Discarder, Prepender, + } + + private static Gen valueGen(AbstractType type) + { + return Generators.toGen(AbstractTypeGenerators.getTypeSupport(type) + .bytesGen()) + .map(TxnReferenceValue.Constant::new); + } + + private static Gen gen() + { + return rs -> { + TxnReferenceOperation.Kind kind; + ColumnMetadata receiver; + TableMetadata table; + @Nullable ByteBuffer keyOrIndex = null; + @Nullable ByteBuffer field = null; + TxnReferenceValue value; + Group group = rs.pick(Group.values()); + switch (group) + { + case Prepender: + { + kind = TxnReferenceOperation.Kind.ListPrepender; + ListType type = ListType.getInstance(Int32Type.instance, true); + value = valueGen(type).next(rs); + + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + } + break; + case Discarder: + { + CollectionType type = (CollectionType) Generators.toGen(AbstractTypeGenerators.builder() + .withMaxDepth(1) + .withTypeKinds(AbstractTypeGenerators.TypeKind.LIST, + AbstractTypeGenerators.TypeKind.SET, + AbstractTypeGenerators.TypeKind.MAP) + .build()) + .next(rs); + kind = type instanceof ListType + ? TxnReferenceOperation.Kind.ListDiscarder + : TxnReferenceOperation.Kind.SetDiscarder; + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + if (kind == TxnReferenceOperation.Kind.ListDiscarder && rs.nextBoolean()) + { + kind = TxnReferenceOperation.Kind.ListDiscarderByIndex; + value = new TxnReferenceValue.Constant(Int32Type.instance.decompose(42)); + } + else if (type instanceof MapType && rs.nextBoolean()) + { + kind = TxnReferenceOperation.Kind.MapDiscarderByKey; + var keyType = ((MapType) type).getKeysType(); + value = valueGen(keyType).next(rs); + } + else if (type instanceof SetType && rs.nextBoolean()) + { + kind = TxnReferenceOperation.Kind.SetElementDiscarder; + var elementType = ((SetType) type).getElementsType(); + value = valueGen(elementType).next(rs); + } + else + { + CollectionType discardType = type instanceof MapType + ? SetType.getInstance(((MapType) type).getKeysType(), true) + : type; + value = valueGen(discardType).next(rs); + } + } + break; + case Adder: + case Subtracter: + { + if (group == Group.Adder && rs.nextBoolean()) + { + // this is similar to ListAppend + var type = SetType.getInstance(Int32Type.instance, true); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type).next(rs); + kind = TxnReferenceOperation.Kind.SetAdder; + } + else + { + var type = Int32Type.instance; + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type).next(rs); + kind = group == Group.Adder ? TxnReferenceOperation.Kind.ConstantAdder : TxnReferenceOperation.Kind.ConstantSubtracter; + } + } + break; + case Setter: + { + var type = Generators.toGen(AbstractTypeGenerators.builder() + .withoutUnsafeEquality() + .withMaxDepth(1) + .build()) + .next(rs); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type).next(rs); + if (type instanceof ListType) + kind = TxnReferenceOperation.Kind.ListSetter; + else if (type instanceof SetType) + kind = TxnReferenceOperation.Kind.SetSetter; + else if (type instanceof MapType) + kind = TxnReferenceOperation.Kind.MapSetter; + else if (type instanceof UserType) + kind = TxnReferenceOperation.Kind.UserTypeSetter; + else + kind = TxnReferenceOperation.Kind.ConstantSetter; + } + break; + case SetterByIndex: + { + ListType type = ListType.getInstance(UTF8Type.instance, true); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type.getElementsType()).next(rs); + kind = TxnReferenceOperation.Kind.ListSetterByIndex;// x[?] = ? + keyOrIndex = Int32Type.instance.decompose(42); + } + break; + case Appender: + { + ListType type = ListType.getInstance(UTF8Type.instance, true); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type).next(rs); + kind = TxnReferenceOperation.Kind.ListAppender; + } + break; + case SetterByKey: + { + MapType type = MapType.getInstance(Int32Type.instance, UTF8Type.instance, true); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type.getValuesType()).next(rs); + kind = TxnReferenceOperation.Kind.MapSetterByKey; + keyOrIndex = Int32Type.instance.decompose(42); + } + break; + case Putter: // x += {foo: bar, baz: biz} -- basically Appender but for Map! + { + MapType type = MapType.getInstance(Int32Type.instance, UTF8Type.instance, true); + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(type).next(rs); + kind = TxnReferenceOperation.Kind.MapPutter; + } + break; + case SetterByField: + { + UserType type = new UserType(KS, ByteBufferUtil.bytes("udt"), + List.of(FieldIdentifier.forUnquoted("f1")), + List.of(UTF8Type.instance), + true); + kind = TxnReferenceOperation.Kind.UserTypeSetterByField; + table = table(type); + receiver = table.getColumn(ColumnIdentifier.getInterned("col", true)); + value = valueGen(UTF8Type.instance).next(rs); + field = FieldIdentifier.forUnquoted("f1").bytes; + } + break; + + default: + throw new UnsupportedOperationException(); + } + return new TxnReferenceOperation(kind, receiver, table, keyOrIndex, field, value); + }; + } + + private static TableMetadata table(AbstractType type) + { + return TableMetadata.builder(KS, "tbl") + .partitioner(Murmur3Partitioner.instance) + .addPartitionKeyColumn("pk", Int32Type.instance) + .addRegularColumn("col", type) + .build(); + } +}