You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ma...@apache.org on 2023/01/25 20:33:42 UTC
[cassandra] 24/27: Support CAS and serial read on Accord
This is an automated email from the ASF dual-hosted git repository.
maedhroz pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit c4ad93fc95c9401c87488e76a9307480ff52450d
Author: Ariel Weisberg <aw...@apple.com>
AuthorDate: Fri Dec 2 14:05:06 2022 -0500
Support CAS and serial read on Accord
patch by Ariel Weisberg; Reviewed by Caleb Rackliffe for CASSANDRA-18100
---
src/java/org/apache/cassandra/config/Config.java | 29 +-
.../cassandra/config/DatabaseDescriptor.java | 6 +-
src/java/org/apache/cassandra/cql3/Operator.java | 67 ++-
.../cassandra/cql3/conditions/ColumnCondition.java | 420 +++++++++++++++--
.../cassandra/cql3/statements/CQL3CasRequest.java | 131 +++++-
.../cql3/statements/TransactionStatement.java | 6 +-
.../org/apache/cassandra/db/marshal/ListType.java | 4 +-
.../org/apache/cassandra/db/marshal/MapType.java | 4 +-
.../org/apache/cassandra/db/marshal/SetType.java | 9 +-
.../org/apache/cassandra/service/CASRequest.java | 13 +-
.../org/apache/cassandra/service/StorageProxy.java | 57 ++-
.../service/accord/AccordPartialCommand.java | 4 +-
.../service/accord/AccordSerializers.java | 92 +++-
.../cassandra/service/accord/AccordService.java | 22 +-
.../serializers/BeginInvalidationSerializers.java | 8 +-
.../accord/serializers/CheckStatusSerializers.java | 16 +-
.../accord/serializers/CommitSerializers.java | 8 +-
.../accord/serializers/PreacceptSerializers.java | 5 +-
.../accord/serializers/RecoverySerializers.java | 11 +-
.../cassandra/service/accord/txn/TxnCondition.java | 76 ++-
.../cassandra/service/accord/txn/TxnDataName.java | 29 +-
.../cassandra/service/accord/txn/TxnNamedRead.java | 2 +-
.../cassandra/service/accord/txn/TxnQuery.java | 49 +-
.../cassandra/service/accord/txn/TxnRead.java | 14 +-
.../cassandra/service/accord/txn/TxnReference.java | 5 +
.../cassandra/service/accord/txn/TxnUpdate.java | 27 +-
.../service/paxos/PaxosPrepareRefresh.java | 4 +-
.../cassandra/service/paxos/PaxosRepair.java | 29 +-
.../apache/cassandra/service/paxos/PaxosState.java | 27 +-
.../org/apache/cassandra/utils/ByteBufferUtil.java | 43 +-
.../cassandra/utils/CollectionSerializers.java | 5 +
.../apache/cassandra/utils/NullableSerializer.java | 4 +-
.../distributed/test/accord/AccordCQLTest.java | 59 ++-
.../distributed/test/accord/AccordTestBase.java | 58 ++-
.../cql3/conditions/ColumnConditionTest.java | 511 ++++++++++++++++++---
35 files changed, 1565 insertions(+), 289 deletions(-)
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 0de550d033..0a8699bf58 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -21,19 +21,17 @@ import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.util.Collections;
import java.util.HashMap;
-import java.util.LinkedHashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.function.Supplier;
-
import javax.annotation.Nullable;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -1067,6 +1065,8 @@ public class Config
public volatile boolean sstable_read_rate_persistence_enabled = false;
+ public LegacyPaxosStrategy legacy_paxos_strategy = LegacyPaxosStrategy.migration;
+
public volatile int max_top_size_partition_count = 10;
public volatile int max_top_tombstone_partition_count = 10;
public volatile DataStorageSpec.LongBytesBound min_tracked_partition_size = new DataStorageSpec.LongBytesBound("1MiB");
@@ -1162,6 +1162,29 @@ public class Config
exception
}
+ /*
+ * How to pick a consensus protocol for CAS
+ * and serial read operations. Transaction statements
+ * will always run on Accord. Legacy in this context includes PaxosV2.
+ */
+ public enum LegacyPaxosStrategy
+ {
+ /*
+ * Allow both Accord and PaxosV1/V2 to run on the same cluster
+ * Some keys and ranges might be running on Accord if they
+ * have been migrated and the rest will run on Paxos until
+ * they are migrated.
+ */
+ migration,
+
+ /*
+ * Everything will be run on Accord. Useful for new deployments
+ * that don't want to accidentally start using legacy Paxos
+ * requiring migration to Accord.
+ */
+ accord
+ }
+
private static final Set<String> SENSITIVE_KEYS = new HashSet<String>() {{
add("client_encryption_options");
add("server_encryption_options");
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 3b7def0546..4bb4639b25 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -43,7 +43,6 @@ import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
-
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
@@ -2863,6 +2862,11 @@ public class DatabaseDescriptor
return conf.paxos_topology_repair_strict_each_quorum;
}
+ public static Config.LegacyPaxosStrategy getLegacyPaxosStrategy()
+ {
+ return conf.legacy_paxos_strategy;
+ }
+
public static void setNativeTransportMaxRequestDataInFlightPerIpInBytes(long maxRequestDataInFlightInBytes)
{
if (maxRequestDataInFlightInBytes == -1)
diff --git a/src/java/org/apache/cassandra/cql3/Operator.java b/src/java/org/apache/cassandra/cql3/Operator.java
index bcb5f63be6..625e53ee83 100644
--- a/src/java/org/apache/cassandra/cql3/Operator.java
+++ b/src/java/org/apache/cassandra/cql3/Operator.java
@@ -24,11 +24,18 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.stream.Stream;
import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.serializers.ListSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.primitives.Ints.checkedCast;
+import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
+
public enum Operator
{
EQ(0)
@@ -260,6 +267,25 @@ public enum Operator
}
};
+ private static final Operator[] idToOperatorMapping;
+
+ static
+ {
+ Operator[] operators = values();
+ int maxId = Stream.of(operators)
+ .map(Operator::getValue)
+ .max(Integer::compareTo)
+ .get();
+
+ idToOperatorMapping = new Operator[maxId + 1];
+ for (Operator operator : operators)
+ {
+ if (null != idToOperatorMapping[operator.b])
+ throw new IllegalStateException("Duplicate Operator id " + operator.b);
+ idToOperatorMapping[operator.b] = operator;
+ }
+ }
+
/**
* The binary representation of this <code>Enum</code> value.
*/
@@ -285,6 +311,17 @@ public enum Operator
output.writeInt(b);
}
+ /**
+ * Write the serialized version of this <code>Operator</code> to the specified output.
+ *
+ * @param output the output to write to
+ * @throws IOException if an I/O problem occurs while writing to the specified output
+ */
+ public void writeToUnsignedVInt(DataOutputPlus output) throws IOException
+ {
+ output.writeUnsignedVInt(b);
+ }
+
public int getValue()
{
return b;
@@ -299,12 +336,27 @@ public enum Operator
*/
public static Operator readFrom(DataInput input) throws IOException
{
- int b = input.readInt();
- for (Operator operator : values())
- if (operator.b == b)
- return operator;
+ return fromBinary(input.readInt());
+ }
- throw new IOException(String.format("Cannot resolve Relation.Type from binary representation: %s", b));
+ /**
+ * Deserializes a <code>Operator</code> instance from the specified input.
+ *
+ * @param input the input to read from
+ * @return the <code>Operator</code> instance deserialized
+ * @throws IOException if a problem occurs while deserializing the <code>Type</code> instance.
+ */
+ public static Operator readFromUnsignedVInt(DataInputPlus input) throws IOException
+ {
+ return fromBinary(checkedCast(input.readUnsignedVInt()));
+ }
+
+ private static Operator fromBinary(int b) throws IOException
+ {
+ checkArgument(b > -1, "b must be > -1 to be a valid Operator id");
+ if (b > idToOperatorMapping.length)
+ throw new IOException(String.format("Cannot resolve Operator from binary representation: %s", b));
+ return idToOperatorMapping[b];
}
/**
@@ -358,4 +410,9 @@ public enum Operator
{
return this == CONTAINS_KEY;
}
+
+ public long sizeAsUnsignedVInt()
+ {
+ return sizeofUnsignedVInt(b);
+ }
}
diff --git a/src/java/org/apache/cassandra/cql3/conditions/ColumnCondition.java b/src/java/org/apache/cassandra/cql3/conditions/ColumnCondition.java
index 9d9506ff53..7af4635b2a 100644
--- a/src/java/org/apache/cassandra/cql3/conditions/ColumnCondition.java
+++ b/src/java/org/apache/cassandra/cql3/conditions/ColumnCondition.java
@@ -17,24 +17,74 @@
*/
package org.apache.cassandra.cql3.conditions;
+import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.*;
-
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedSet;
+import java.util.function.Supplier;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+import com.google.common.base.Suppliers;
import com.google.common.collect.Iterators;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
-import org.apache.cassandra.cql3.*;
+import org.apache.cassandra.cql3.AbstractMarker;
+import org.apache.cassandra.cql3.ColumnSpecification;
+import org.apache.cassandra.cql3.Constants;
+import org.apache.cassandra.cql3.FieldIdentifier;
+import org.apache.cassandra.cql3.Lists;
+import org.apache.cassandra.cql3.Maps;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.Sets;
+import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.cql3.Term.Terminal;
+import org.apache.cassandra.cql3.Terms;
+import org.apache.cassandra.cql3.UserTypes;
+import org.apache.cassandra.cql3.VariableSpecifications;
import org.apache.cassandra.cql3.functions.Function;
-import org.apache.cassandra.db.rows.*;
-import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.ByteBufferAccessor;
+import org.apache.cassandra.db.marshal.CollectionType;
+import org.apache.cassandra.db.marshal.CounterColumnType;
+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.UserType;
+import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.CellPath;
+import org.apache.cassandra.db.rows.ComplexColumnData;
+import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
-import org.apache.commons.lang3.builder.ToStringBuilder;
-import org.apache.commons.lang3.builder.ToStringStyle;
-import static org.apache.cassandra.cql3.statements.RequestValidations.*;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.primitives.Ints.checkedCast;
+import static java.util.stream.Collectors.toList;
+import static org.apache.cassandra.cql3.Constants.UNSET_VALUE;
+import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
+import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
+import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
+import static org.apache.cassandra.service.accord.AccordSerializers.columnMetadataSerializer;
+import static org.apache.cassandra.service.accord.AccordSerializers.deserializeCqlCollectionAsTerm;
+import static org.apache.cassandra.utils.ByteBufferUtil.UNSET_BYTE_BUFFER;
+import static org.apache.cassandra.utils.ByteBufferUtil.byteBufferSerializer;
+import static org.apache.cassandra.utils.ByteBufferUtil.nullableByteBufferSerializer;
+import static org.apache.cassandra.utils.CollectionSerializers.deserializeList;
+import static org.apache.cassandra.utils.CollectionSerializers.readCollectionSize;
+import static org.apache.cassandra.utils.CollectionSerializers.serializeList;
+import static org.apache.cassandra.utils.CollectionSerializers.serializedListSize;
/**
* A CQL3 condition on the value of a column or collection element. For example, "UPDATE .. IF a = 0".
@@ -73,7 +123,7 @@ public abstract class ColumnCondition
terms.collectMarkerSpecification(boundNames);
}
- public abstract ColumnCondition.Bound bind(QueryOptions options);
+ public abstract Bound bind(QueryOptions options);
protected final List<ByteBuffer> bindAndGetTerms(QueryOptions options)
{
@@ -107,7 +157,7 @@ public abstract class ColumnCondition
{
T value = values.get(i);
// The value can be ByteBuffer or Constants.Value so we need to check the 2 type of UNSET
- if (value != ByteBufferUtil.UNSET_BYTE_BUFFER && value != Constants.UNSET_VALUE)
+ if (value != UNSET_BYTE_BUFFER && value != UNSET_VALUE)
filtered.add(value);
}
return filtered;
@@ -129,7 +179,7 @@ public abstract class ColumnCondition
return new MultiCellCollectionBound(column, operator, bindTerms(options));
if (column.type.isUDT() && column.type.isMultiCell())
- return new MultiCellUdtBound(column, operator, bindAndGetTerms(options), options.getProtocolVersion());
+ return new MultiCellUdtBound(column, operator, bindAndGetTerms(options));
return new SimpleBound(column, operator, bindAndGetTerms(options));
}
@@ -210,13 +260,57 @@ public abstract class ColumnCondition
return new UDTFieldCondition(column, udtField, op, terms);
}
+ enum BoundKind
+ {
+ ELEMENT_ACCESS(0, ElementAccessBound.serializer),
+ MULTI_CELL_COLLECTION(1, MultiCellCollectionBound.serializer),
+ MULTI_CELL_UDT(2, MultiCellUdtBound.serializer),
+ SIMPLE(3, SimpleBound.serializer),
+ UDT_FIELD_ACCESS(4, UDTFieldAccessBound.serializer);
+
+ final int id;
+
+ final BoundSerializer serializer;
+
+ BoundKind(int id, BoundSerializer serializer)
+ {
+ this.id = id;
+ this.serializer = serializer;
+ }
+
+ @SuppressWarnings("rawtypes")
+ static BoundSerializer serializer(int id)
+ {
+ switch(id)
+ {
+ case 0:
+ return ElementAccessBound.serializer;
+ case 1:
+ return MultiCellCollectionBound.serializer;
+ case 2:
+ return MultiCellUdtBound.serializer;
+ case 3:
+ return SimpleBound.serializer;
+ case 4:
+ return UDTFieldAccessBound.serializer;
+ default:
+ throw new AssertionError("Shouldn't have an enum with no serializer");
+ }
+ }
+ }
+
public static abstract class Bound
{
+ @Nonnull
public final ColumnMetadata column;
+
+ @Nonnull
public final Operator comparisonOperator;
protected Bound(ColumnMetadata column, Operator operator)
{
+ checkNotNull(column);
+ checkNotNull(operator);
this.column = column;
// If the operator is an IN we want to compare the value using an EQ.
this.comparisonOperator = operator.isIN() ? Operator.EQ : operator;
@@ -235,7 +329,7 @@ public abstract class ColumnCondition
/** Returns true if the operator is satisfied (i.e. "otherValue operator value == true"), false otherwise. */
public static boolean compareWithOperator(Operator operator, AbstractType<?> type, ByteBuffer value, ByteBuffer otherValue)
{
- if (value == ByteBufferUtil.UNSET_BYTE_BUFFER)
+ if (value == UNSET_BYTE_BUFFER)
throw invalidRequest("Invalid 'unset' value in condition");
if (value == null)
@@ -257,6 +351,55 @@ public abstract class ColumnCondition
}
return operator.isSatisfiedBy(type, otherValue, value);
}
+
+ void checkForUnsetValues(List<ByteBuffer> values)
+ {
+ for (ByteBuffer buffer : values)
+ if (buffer == UNSET_BYTE_BUFFER)
+ throw invalidRequest("Invalid 'unset' value in condition");
+ }
+ protected abstract BoundKind kind();
+
+ public static final IVersionedSerializer<Bound> serializer = new IVersionedSerializer<Bound>()
+ {
+ @Override
+ @SuppressWarnings("unchecked")
+ public void serialize(Bound bound, DataOutputPlus out, int version) throws IOException
+ {
+ columnMetadataSerializer.serialize(bound.column, out, version);
+ bound.comparisonOperator.writeToUnsignedVInt(out);
+ BoundKind kind = bound.kind();
+ out.writeUnsignedVInt(kind.ordinal());
+ kind.serializer.serialize(bound, out, version);
+ }
+
+ @Override
+ public Bound deserialize(DataInputPlus in, int version) throws IOException
+ {
+ ColumnMetadata column = columnMetadataSerializer.deserialize(in, version);
+ Operator comparisonOperator = Operator.readFromUnsignedVInt(in);
+ int boundKind = checkedCast(in.readUnsignedVInt());
+ return BoundKind.serializer(boundKind).deserialize(in, version, column, comparisonOperator);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public long serializedSize(Bound bound, int version)
+ {
+ BoundKind kind = bound.kind();
+ return columnMetadataSerializer.serializedSize(bound.column, version)
+ + bound.comparisonOperator.sizeAsUnsignedVInt()
+ + sizeofUnsignedVInt(kind.ordinal())
+ + kind.serializer.serializedSize(bound, version);
+ }
+ };
+ }
+
+ private interface BoundSerializer<T extends Bound>
+ {
+ void serialize(T bound, DataOutputPlus out, int version) throws IOException;
+ Bound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException;
+ long serializedSize(T condition, int version);
}
protected static final Cell<?> getCell(Row row, ColumnMetadata column)
@@ -317,6 +460,7 @@ public abstract class ColumnCondition
private SimpleBound(ColumnMetadata column, Operator operator, List<ByteBuffer> values)
{
super(column, operator);
+ checkForUnsetValues(values);
this.values = values;
}
@@ -341,6 +485,34 @@ public abstract class ColumnCondition
}
return false;
}
+
+ @Override
+ protected BoundKind kind()
+ {
+ return BoundKind.SIMPLE;
+ }
+
+ private static final BoundSerializer<SimpleBound> serializer = new BoundSerializer<SimpleBound>()
+ {
+ @Override
+ public void serialize(SimpleBound bound, DataOutputPlus out, int version) throws IOException
+ {
+ serializeList(bound.values, out, version, nullableByteBufferSerializer);
+ }
+
+ @Override
+ public SimpleBound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException
+ {
+ List<ByteBuffer> values = deserializeList(in, version, nullableByteBufferSerializer);
+ return new SimpleBound(column, operator, values);
+ }
+
+ @Override
+ public long serializedSize(SimpleBound bound, int version)
+ {
+ return serializedListSize(bound.values, version, nullableByteBufferSerializer);
+ }
+ };
}
/**
@@ -351,11 +523,13 @@ public abstract class ColumnCondition
/**
* The collection element
*/
+ @Nonnull
private final ByteBuffer collectionElement;
/**
* The conditions values.
*/
+ @Nonnull
private final List<ByteBuffer> values;
public ElementAccessBound(ColumnMetadata column,
@@ -364,6 +538,11 @@ public abstract class ColumnCondition
List<ByteBuffer> values)
{
super(column, operator);
+ checkForUnsetValues(values);
+
+ if (collectionElement == null)
+ throw invalidRequest("Invalid null value for %s element access",
+ column.type instanceof MapType ? "map" : "list");
this.collectionElement = collectionElement;
this.values = values;
@@ -374,9 +553,6 @@ public abstract class ColumnCondition
{
boolean isMap = column.type instanceof MapType;
- if (collectionElement == null)
- throw invalidRequest("Invalid null value for %s element access", isMap ? "map" : "list");
-
if (isMap)
{
MapType<?, ?> mapType = (MapType<?, ?>) column.type;
@@ -445,6 +621,36 @@ public abstract class ColumnCondition
checkFalse(idx < 0, "Invalid negative list index %d", idx);
return idx;
}
+
+ @Override
+ protected BoundKind kind()
+ {
+ return BoundKind.ELEMENT_ACCESS;
+ }
+
+ private static final BoundSerializer<ElementAccessBound> serializer = new BoundSerializer<ElementAccessBound>()
+ {
+ @Override
+ public void serialize(ElementAccessBound bound, DataOutputPlus out, int version) throws IOException
+ {
+ byteBufferSerializer.serialize(bound.collectionElement, out, version);
+ serializeList(bound.values, out, version, nullableByteBufferSerializer);
+ }
+
+ @Override
+ public ElementAccessBound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException
+ {
+ ByteBuffer collectionElement = byteBufferSerializer.deserialize(in, version);
+ List<ByteBuffer> values = deserializeList(in, version, nullableByteBufferSerializer);
+ return new ElementAccessBound(column, collectionElement, operator, values);
+ }
+
+ @Override
+ public long serializedSize(ElementAccessBound bound , int version)
+ {
+ return byteBufferSerializer.serializedSize(bound.collectionElement, version) + serializedListSize(bound.values, version, nullableByteBufferSerializer);
+ }
+ };
}
/**
@@ -452,26 +658,47 @@ public abstract class ColumnCondition
*/
public static final class MultiCellCollectionBound extends Bound
{
- private final List<Term.Terminal> values;
+ @Nonnull
+ private final List<Terminal> values;
- public MultiCellCollectionBound(ColumnMetadata column, Operator operator, List<Term.Terminal> values)
+ /**
+ * Avoid recomputing the serialized terminals for serialize and serializedSize
+ */
+ @Nullable
+ private Supplier<List<ByteBuffer>> serializedValues;
+
+ public MultiCellCollectionBound(ColumnMetadata column, Operator operator, List<Terminal> values)
{
super(column, operator);
+ checkNotNull(values);
assert column.type.isMultiCell();
this.values = values;
}
+ public Supplier<List<ByteBuffer>> serializedValues()
+ {
+ if (serializedValues != null)
+ return serializedValues;
+
+ serializedValues = Suppliers.memoize(() ->
+ values.stream()
+ .map(v -> v == null ? null : v.get(ProtocolVersion.CURRENT))
+ .collect(toList()));
+
+ return serializedValues;
+ }
+
public boolean appliesTo(Row row)
{
return appliesTo(column, comparisonOperator, values, row);
}
- public static boolean appliesTo(ColumnMetadata column, Operator operator, List<Term.Terminal> values, Row row)
+ public static boolean appliesTo(ColumnMetadata column, Operator operator, List<Terminal> values, Row row)
{
CollectionType<?> type = (CollectionType<?>) column.type;
// copy iterator contents so that we can properly reuse them for each comparison with an IN value
- for (Term.Terminal value : values)
+ for (Terminal value : values)
{
Iterator<Cell<?>> iter = getCells(row, column);
if (value == null)
@@ -495,7 +722,7 @@ public abstract class ColumnCondition
return false;
}
- private static boolean valueAppliesTo(CollectionType<?> type, Iterator<Cell<?>> iter, Term.Terminal value, Operator operator)
+ private static boolean valueAppliesTo(CollectionType<?> type, Iterator<Cell<?>> iter, Terminal value, Operator operator)
{
if (value == null)
return !iter.hasNext();
@@ -507,10 +734,10 @@ public abstract class ColumnCondition
{
case LIST:
List<ByteBuffer> valueList = ((Lists.Value) value).elements;
- return listAppliesTo((ListType<?>)type, iter, valueList, operator);
+ return setOrListAppliesTo(((ListType<?>)type).getElementsType(), iter, valueList.iterator(), operator, false);
case SET:
- Set<ByteBuffer> valueSet = ((Sets.Value) value).elements;
- return setAppliesTo((SetType<?>)type, iter, valueSet, operator);
+ SortedSet<ByteBuffer> valueSet = ((Sets.Value) value).elements;
+ return setOrListAppliesTo(((SetType<?>)type).getElementsType(), iter, valueSet.iterator(), operator, true);
case MAP:
Map<ByteBuffer, ByteBuffer> valueMap = ((Maps.Value) value).map;
return mapAppliesTo((MapType<?, ?>)type, iter, valueMap, operator);
@@ -527,7 +754,10 @@ public abstract class ColumnCondition
// for lists we use the cell value; for sets we use the cell name
ByteBuffer cellValue = isSet ? iter.next().path().get(0) : iter.next().buffer();
- int comparison = type.compare(cellValue, conditionIter.next());
+ ByteBuffer conditionValue = conditionIter.next();
+ if (conditionValue == null)
+ conditionValue = ByteBufferUtil.EMPTY_BYTE_BUFFER;
+ int comparison = type.compare(cellValue, conditionValue);
if (comparison != 0)
return evaluateComparisonWithOperator(comparison, operator);
}
@@ -539,18 +769,6 @@ public abstract class ColumnCondition
return operator == Operator.EQ || operator == Operator.LTE || operator == Operator.GTE;
}
- private static boolean listAppliesTo(ListType<?> type, Iterator<Cell<?>> iter, List<ByteBuffer> elements, Operator operator)
- {
- return setOrListAppliesTo(type.getElementsType(), iter, elements.iterator(), operator, false);
- }
-
- private static boolean setAppliesTo(SetType<?> type, Iterator<Cell<?>> iter, Set<ByteBuffer> elements, Operator operator)
- {
- ArrayList<ByteBuffer> sortedElements = new ArrayList<>(elements);
- Collections.sort(sortedElements, type.getElementsType());
- return setOrListAppliesTo(type.getElementsType(), iter, sortedElements.iterator(), operator, true);
- }
-
private static boolean mapAppliesTo(MapType<?, ?> type, Iterator<Cell<?>> iter, Map<ByteBuffer, ByteBuffer> elements, Operator operator)
{
Iterator<Map.Entry<ByteBuffer, ByteBuffer>> conditionIter = elements.entrySet().iterator();
@@ -562,13 +780,17 @@ public abstract class ColumnCondition
Map.Entry<ByteBuffer, ByteBuffer> conditionEntry = conditionIter.next();
Cell<?> c = iter.next();
+ ByteBuffer conditionEntryKey = conditionEntry.getKey();
+
// compare the keys
- int comparison = type.getKeysType().compare(c.path().get(0), conditionEntry.getKey());
+ int comparison = type.getKeysType().compare(c.path().get(0), conditionEntryKey);
if (comparison != 0)
return evaluateComparisonWithOperator(comparison, operator);
+ ByteBuffer conditionEntryValue = conditionEntry.getValue();
+
// compare the values
- comparison = type.getValuesType().compare(c.buffer(), conditionEntry.getValue());
+ comparison = type.getValuesType().compare(c.buffer(), conditionEntryValue);
if (comparison != 0)
return evaluateComparisonWithOperator(comparison, operator);
}
@@ -579,6 +801,45 @@ public abstract class ColumnCondition
// they're equal
return operator == Operator.EQ || operator == Operator.LTE || operator == Operator.GTE;
}
+
+ @Override
+ protected BoundKind kind()
+ {
+ return BoundKind.MULTI_CELL_COLLECTION;
+ }
+
+ private static final BoundSerializer<MultiCellCollectionBound> serializer = new BoundSerializer<MultiCellCollectionBound>()
+ {
+ @Override
+ public void serialize(MultiCellCollectionBound bound, DataOutputPlus out, int version) throws IOException
+ {
+ serializeList(bound.serializedValues().get(), out, version, nullableByteBufferSerializer);
+ }
+
+ @Override
+ public MultiCellCollectionBound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException
+ {
+ int numTerminals = readCollectionSize(in, version);
+ List<Terminal> terminals = new ArrayList<>(numTerminals);
+ if (operator.isContains() || operator.isContainsKey())
+ {
+ for (int i = 0; i < numTerminals; i++)
+ terminals.add(new Constants.Value(nullableByteBufferSerializer.deserialize(in, version)));
+ }
+ else
+ {
+ for (int i = 0; i < numTerminals; i++)
+ terminals.add(deserializeCqlCollectionAsTerm(nullableByteBufferSerializer.deserialize(in, version), column.type, ProtocolVersion.CURRENT));
+ }
+ return new MultiCellCollectionBound(column, operator, terminals);
+ }
+
+ @Override
+ public long serializedSize(MultiCellCollectionBound bound, int version)
+ {
+ return serializedListSize(bound.serializedValues().get(), version, nullableByteBufferSerializer);
+ }
+ };
}
private static boolean containsAppliesTo(CollectionType<?> type, Iterator<Cell<?>> iter, ByteBuffer value, Operator operator)
@@ -622,17 +883,22 @@ public abstract class ColumnCondition
/**
* The UDT field.
*/
+ @Nonnull
private final FieldIdentifier field;
/**
* The conditions values.
*/
+ @Nonnull
private final List<ByteBuffer> values;
private UDTFieldAccessBound(ColumnMetadata column, FieldIdentifier field, Operator operator, List<ByteBuffer> values)
{
super(column, operator);
- assert column.type.isUDT() && field != null;
+ checkNotNull(field);
+ checkNotNull(values);
+ checkForUnsetValues(values);
+ assert column.type.isUDT();
this.field = field;
this.values = values;
}
@@ -677,6 +943,37 @@ public abstract class ColumnCondition
{
return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
}
+
+ @Override
+ protected BoundKind kind()
+ {
+ return BoundKind.UDT_FIELD_ACCESS;
+ }
+
+ private static final BoundSerializer<UDTFieldAccessBound> serializer = new BoundSerializer<UDTFieldAccessBound>()
+ {
+ @Override
+ public void serialize(UDTFieldAccessBound bound, DataOutputPlus out, int version) throws IOException
+ {
+ nullableByteBufferSerializer.serialize(bound.field.bytes, out, version);
+ serializeList(bound.values, out, version, nullableByteBufferSerializer);
+ }
+
+ @Override
+ public UDTFieldAccessBound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException
+ {
+ FieldIdentifier field = new FieldIdentifier(nullableByteBufferSerializer.deserialize(in, version));
+ List<ByteBuffer> values = deserializeList(in, version, nullableByteBufferSerializer);
+ return new UDTFieldAccessBound(column, field, operator, values);
+ }
+
+ @Override
+ public long serializedSize(UDTFieldAccessBound bound, int version)
+ {
+
+ return nullableByteBufferSerializer.serializedSize(bound.field.bytes, version) + serializedListSize(bound.values, version, nullableByteBufferSerializer);
+ }
+ };
}
/**
@@ -687,19 +984,16 @@ public abstract class ColumnCondition
/**
* The conditions values.
*/
+ @Nonnull
private final List<ByteBuffer> values;
- /**
- * The protocol version
- */
- private final ProtocolVersion protocolVersion;
-
- public MultiCellUdtBound(ColumnMetadata column, Operator op, List<ByteBuffer> values, ProtocolVersion protocolVersion)
+ public MultiCellUdtBound(ColumnMetadata column, Operator op, List<ByteBuffer> values)
{
super(column, op);
+ checkNotNull(values);
+ checkForUnsetValues(values);
assert column.type.isMultiCell();
this.values = values;
- this.protocolVersion = protocolVersion;
}
@Override
@@ -708,11 +1002,12 @@ public abstract class ColumnCondition
return isSatisfiedBy(rowValue(row));
}
- private final ByteBuffer rowValue(Row row)
+ private ByteBuffer rowValue(Row row)
{
UserType userType = (UserType) column.type;
Iterator<Cell<?>> iter = getCells(row, column);
- return iter.hasNext() ? userType.serializeForNativeProtocol(iter, protocolVersion) : null;
+ // User type doesn't use the protocol version so passing in null
+ return iter.hasNext() ? userType.serializeForNativeProtocol(iter, null) : null;
}
private boolean isSatisfiedBy(ByteBuffer rowValue)
@@ -730,6 +1025,35 @@ public abstract class ColumnCondition
{
return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
}
+
+ @Override
+ public BoundKind kind()
+ {
+ return BoundKind.MULTI_CELL_UDT;
+ }
+
+ private static final BoundSerializer<MultiCellUdtBound> serializer = new BoundSerializer<MultiCellUdtBound>()
+ {
+ @Override
+ public void serialize(MultiCellUdtBound bound, DataOutputPlus out, int version) throws IOException
+ {
+ serializeList(bound.values, out, version, nullableByteBufferSerializer);
+ }
+
+ @Override
+ public MultiCellUdtBound deserialize(DataInputPlus in, int version, ColumnMetadata column, Operator operator) throws IOException
+ {
+ List<ByteBuffer> values = deserializeList(in, version, nullableByteBufferSerializer);
+ // Does decode actually do what we want?
+ return new MultiCellUdtBound(column, operator, values);
+ }
+
+ @Override
+ public long serializedSize(MultiCellUdtBound bound, int version)
+ {
+ return serializedListSize(bound.values, version, nullableByteBufferSerializer);
+ }
+ };
}
public static class Raw
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
index 02931005a7..1bc79c468b 100644
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -18,30 +18,61 @@
package org.apache.cassandra.cql3.statements;
import java.nio.ByteBuffer;
-import java.util.*;
-
-import com.google.common.collect.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.commons.lang3.builder.ToStringStyle;
-import org.apache.cassandra.db.marshal.TimeUUIDType;
-import org.apache.cassandra.index.IndexRegistry;
-import org.apache.cassandra.schema.TableMetadata;
-import org.apache.cassandra.cql3.*;
+import accord.api.Update;
+import accord.primitives.Txn;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.UpdateParameters;
import org.apache.cassandra.cql3.conditions.ColumnCondition;
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.filter.*;
-import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.Columns;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.RegularAndStaticColumns;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
+import org.apache.cassandra.db.Slice;
+import org.apache.cassandra.db.Slices;
+import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
+import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
+import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.filter.DataLimits;
+import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.marshal.TimeUUIDType;
import org.apache.cassandra.db.partitions.FilteredPartition;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.index.IndexRegistry;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.CASRequest;
import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.service.accord.txn.TxnCondition;
+import org.apache.cassandra.service.accord.txn.TxnData;
+import org.apache.cassandra.service.accord.txn.TxnDataName;
+import org.apache.cassandra.service.accord.txn.TxnQuery;
+import org.apache.cassandra.service.accord.txn.TxnRead;
+import org.apache.cassandra.service.accord.txn.TxnReference;
+import org.apache.cassandra.service.accord.txn.TxnUpdate;
+import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.TimeUUID;
-import org.apache.commons.lang3.builder.ToStringBuilder;
-import org.apache.commons.lang3.builder.ToStringStyle;
+import static com.google.common.base.Preconditions.checkState;
+import static org.apache.cassandra.service.accord.txn.TxnDataName.Kind.USER;
/**
* Processed CAS conditions and update on potentially multiple rows of the same partition.
@@ -343,6 +374,8 @@ public class CQL3CasRequest implements CASRequest
}
public abstract boolean appliesTo(FilteredPartition current) throws InvalidRequestException;
+
+ public abstract TxnCondition asTxnCondition();
}
private static class NotExistCondition extends RowCondition
@@ -356,6 +389,14 @@ public class CQL3CasRequest implements CASRequest
{
return current.getRow(clustering) == null;
}
+
+ @Override
+ public TxnCondition asTxnCondition()
+ {
+ TxnDataName txnDataName = new TxnDataName(USER, clustering, TxnRead.SERIAL_READ_NAME);
+ TxnReference txnReference = new TxnReference(txnDataName, null);
+ return new TxnCondition.Exists(txnReference, TxnCondition.Kind.IS_NULL);
+ }
}
private static class ExistCondition extends RowCondition
@@ -369,6 +410,14 @@ public class CQL3CasRequest implements CASRequest
{
return current.getRow(clustering) != null;
}
+
+ @Override
+ public TxnCondition asTxnCondition()
+ {
+ TxnDataName txnDataName = new TxnDataName(USER, clustering, TxnRead.SERIAL_READ_NAME);
+ TxnReference txnReference = new TxnReference(txnDataName, null);
+ return new TxnCondition.Exists(txnReference, TxnCondition.Kind.IS_NOT_NULL);
+ }
}
private static class ColumnsConditions extends RowCondition
@@ -399,6 +448,12 @@ public class CQL3CasRequest implements CASRequest
}
return true;
}
+
+ @Override
+ public TxnCondition asTxnCondition()
+ {
+ return new TxnCondition.ColumnConditionsAdapter(clustering, conditions.values());
+ }
}
@Override
@@ -406,4 +461,56 @@ public class CQL3CasRequest implements CASRequest
{
return ToStringBuilder.reflectionToString(this, ToStringStyle.SHORT_PREFIX_STYLE);
}
+
+ @Override
+ public Txn toAccordTxn(ClientState clientState, int nowInSecs)
+ {
+ SinglePartitionReadCommand readCommand = readCommand(nowInSecs);
+ Update update = createUpdate(clientState);
+ // In a CAS request only one key is supported and writes
+ // can't be dependent on any data that is read (only conditions)
+ // so the only relevant keys are the read key
+ TxnRead read = TxnRead.createSerialRead(readCommand);
+ return new Txn.InMemory(read.keys(), read, TxnQuery.CONDITION, update);
+ }
+
+ private Update createUpdate(ClientState clientState)
+ {
+ return new TxnUpdate(createWriteFragments(clientState), createCondition());
+ }
+
+ private TxnCondition createCondition()
+ {
+ List<TxnCondition> txnConditions = new ArrayList<>(conditions.size() + (staticConditions == null ? 0 : 1));
+ if (staticConditions != null)
+ {
+ txnConditions.add(staticConditions.asTxnCondition());
+ }
+ for (RowCondition condition : conditions.values())
+ txnConditions.add(condition.asTxnCondition());
+ // CAS forbids empty conditions
+ checkState(!txnConditions.isEmpty());
+ return conditions.size() == 1 ? txnConditions.get(0) : new TxnCondition.BooleanGroup(TxnCondition.Kind.AND, txnConditions);
+ }
+
+ private List<TxnWrite.Fragment> createWriteFragments(ClientState state)
+ {
+ List<TxnWrite.Fragment> fragments = new ArrayList<>();
+ int idx = 0;
+ for (RowUpdate update : updates)
+ {
+ ModificationStatement modification = update.stmt;
+ QueryOptions options = update.options;
+ TxnWrite.Fragment fragment = modification.getTxnWriteFragment(idx++, state, options);
+ fragments.add(fragment);
+ }
+ return fragments;
+ }
+
+ @Override
+ public RowIterator toCasResult(TxnData txnData)
+ {
+ FilteredPartition partition = txnData.get(TxnRead.SERIAL_READ);
+ return partition != null ? partition.rowIterator() : null;
+ }
}
diff --git a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
index e4893154b2..ee254beb50 100644
--- a/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/TransactionStatement.java
@@ -30,13 +30,11 @@ import java.util.SortedSet;
import java.util.TreeSet;
import java.util.function.Consumer;
import java.util.stream.Collectors;
-
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,9 +50,9 @@ import org.apache.cassandra.cql3.ResultSet;
import org.apache.cassandra.cql3.VariableSpecifications;
import org.apache.cassandra.cql3.selection.ResultSetBuilder;
import org.apache.cassandra.cql3.selection.Selection;
-import org.apache.cassandra.cql3.transactions.RowDataReference;
import org.apache.cassandra.cql3.transactions.ConditionStatement;
import org.apache.cassandra.cql3.transactions.ReferenceOperation;
+import org.apache.cassandra.cql3.transactions.RowDataReference;
import org.apache.cassandra.cql3.transactions.SelectReferenceSource;
import org.apache.cassandra.db.ReadQuery;
import org.apache.cassandra.db.SinglePartitionReadCommand;
@@ -330,7 +328,7 @@ public class TransactionStatement implements CQLStatement
if (returningSelect != null)
checkAtMostOneRowSpecified(state.getClientState(), options, returningSelect.select, INCOMPLETE_PRIMARY_KEY_SELECT_MESSAGE);
- TxnData data = AccordService.instance().coordinate(createTxn(state.getClientState(), options), options);
+ TxnData data = AccordService.instance().coordinate(createTxn(state.getClientState(), options), options.getConsistency());
if (returningSelect != null)
{
diff --git a/src/java/org/apache/cassandra/db/marshal/ListType.java b/src/java/org/apache/cassandra/db/marshal/ListType.java
index e985f862bd..2ad54aa421 100644
--- a/src/java/org/apache/cassandra/db/marshal/ListType.java
+++ b/src/java/org/apache/cassandra/db/marshal/ListType.java
@@ -33,8 +33,8 @@ import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.SyntaxException;
import org.apache.cassandra.serializers.ListSerializer;
import org.apache.cassandra.serializers.MarshalException;
-import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.bytecomparable.ByteComparable.Version;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
@@ -124,7 +124,7 @@ public class ListType<T> extends CollectionType<List<T>>
}
@Override
- public AbstractType<?> freeze()
+ public ListType<T> freeze()
{
if (isMultiCell)
return getInstance(this.elements, false);
diff --git a/src/java/org/apache/cassandra/db/marshal/MapType.java b/src/java/org/apache/cassandra/db/marshal/MapType.java
index 835e9597cd..87d7c4eba3 100644
--- a/src/java/org/apache/cassandra/db/marshal/MapType.java
+++ b/src/java/org/apache/cassandra/db/marshal/MapType.java
@@ -37,11 +37,11 @@ import org.apache.cassandra.serializers.CollectionSerializer;
import org.apache.cassandra.serializers.MapSerializer;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteComparable.Version;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse;
-import org.apache.cassandra.utils.Pair;
public class MapType<K, V> extends CollectionType<Map<K, V>>
{
@@ -147,7 +147,7 @@ public class MapType<K, V> extends CollectionType<Map<K, V>>
}
@Override
- public AbstractType<?> freeze()
+ public MapType<K, V> freeze()
{
if (isMultiCell)
return getInstance(this.keys, this.values, false);
diff --git a/src/java/org/apache/cassandra/db/marshal/SetType.java b/src/java/org/apache/cassandra/db/marshal/SetType.java
index e8d6413c7b..c893942f28 100644
--- a/src/java/org/apache/cassandra/db/marshal/SetType.java
+++ b/src/java/org/apache/cassandra/db/marshal/SetType.java
@@ -18,7 +18,12 @@
package org.apache.cassandra.db.marshal;
import java.nio.ByteBuffer;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
@@ -115,7 +120,7 @@ public class SetType<T> extends CollectionType<Set<T>>
}
@Override
- public AbstractType<?> freeze()
+ public SetType<T> freeze()
{
if (isMultiCell)
return getInstance(this.elements, false);
diff --git a/src/java/org/apache/cassandra/service/CASRequest.java b/src/java/org/apache/cassandra/service/CASRequest.java
index 19966c883c..d9847c12af 100644
--- a/src/java/org/apache/cassandra/service/CASRequest.java
+++ b/src/java/org/apache/cassandra/service/CASRequest.java
@@ -17,10 +17,13 @@
*/
package org.apache.cassandra.service;
+import accord.primitives.Txn;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.partitions.FilteredPartition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.service.accord.txn.TxnData;
import org.apache.cassandra.service.paxos.Ballot;
/**
@@ -31,17 +34,21 @@ public interface CASRequest
/**
* The command to use to fetch the value to compare for the CAS.
*/
- public SinglePartitionReadCommand readCommand(int nowInSec);
+ SinglePartitionReadCommand readCommand(int nowInSec);
/**
* Returns whether the provided CF, that represents the values fetched using the
* readFilter(), match the CAS conditions this object stands for.
*/
- public boolean appliesTo(FilteredPartition current) throws InvalidRequestException;
+ boolean appliesTo(FilteredPartition current) throws InvalidRequestException;
/**
* The updates to perform of a CAS success. The values fetched using the readFilter()
* are passed as argument.
*/
- public PartitionUpdate makeUpdates(FilteredPartition current, ClientState clientState, Ballot ballot) throws InvalidRequestException;
+ PartitionUpdate makeUpdates(FilteredPartition current, ClientState clientState, Ballot ballot) throws InvalidRequestException;
+
+ Txn toAccordTxn(ClientState clientState, int nowInSecs);
+
+ RowIterator toCasResult(TxnData data);
}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 49f3b84274..4f85776625 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -47,6 +47,7 @@ import com.google.common.util.concurrent.Uninterruptibles;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.primitives.Txn;
import org.apache.cassandra.batchlog.Batch;
import org.apache.cassandra.batchlog.BatchlogManager;
import org.apache.cassandra.concurrent.DebuggableTask.RunnableDebuggableTask;
@@ -122,6 +123,10 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.txn.TxnData;
+import org.apache.cassandra.service.accord.txn.TxnQuery;
+import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.service.paxos.Ballot;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.service.paxos.ContentionStrategy;
@@ -145,12 +150,10 @@ import org.apache.cassandra.utils.TimeUUID;
import org.apache.cassandra.utils.concurrent.CountDownLatch;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
+import static com.google.common.collect.Iterables.concat;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
-
-import static com.google.common.collect.Iterables.concat;
-import static org.apache.commons.lang3.StringUtils.join;
-
+import static org.apache.cassandra.config.Config.LegacyPaxosStrategy.accord;
import static org.apache.cassandra.db.ConsistencyLevel.SERIAL;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casReadMetrics;
import static org.apache.cassandra.metrics.ClientRequestsMetricsHolder.casWriteMetrics;
@@ -178,6 +181,7 @@ import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID;
import static org.apache.cassandra.utils.concurrent.CountDownLatch.newCountDownLatch;
+import static org.apache.commons.lang3.StringUtils.join;
public class StorageProxy implements StorageProxyMBean
{
@@ -315,12 +319,20 @@ public class StorageProxy implements StorageProxyMBean
{
denylistMetrics.incrementWritesRejected();
throw new InvalidRequestException(String.format("Unable to CAS write to denylisted partition [0x%s] in %s/%s",
- key.toString(), keyspaceName, cfName));
+ key, keyspaceName, cfName));
}
- return Paxos.useV2()
- ? Paxos.cas(key, request, consistencyForPaxos, consistencyForCommit, clientState)
- : legacyCas(keyspaceName, cfName, key, request, consistencyForPaxos, consistencyForCommit, clientState, nowInSeconds, queryStartNanoTime);
+ if (DatabaseDescriptor.getLegacyPaxosStrategy() == accord)
+ {
+ TxnData data = AccordService.instance().coordinate(request.toAccordTxn(clientState, nowInSeconds), consistencyForPaxos);
+ return request.toCasResult(data);
+ }
+ else
+ {
+ return Paxos.useV2()
+ ? Paxos.cas(key, request, consistencyForPaxos, consistencyForCommit, clientState)
+ : legacyCas(keyspaceName, cfName, key, request, consistencyForPaxos, consistencyForCommit, clientState, nowInSeconds, queryStartNanoTime);
+ }
}
public static RowIterator legacyCas(String keyspaceName,
@@ -343,7 +355,7 @@ public class StorageProxy implements StorageProxyMBean
{
// read the current values and check they validate the conditions
Tracing.trace("Reading existing values for CAS precondition");
- SinglePartitionReadCommand readCommand = (SinglePartitionReadCommand) request.readCommand(nowInSeconds);
+ SinglePartitionReadCommand readCommand = request.readCommand(nowInSeconds);
ConsistencyLevel readConsistency = consistencyForPaxos == ConsistencyLevel.LOCAL_SERIAL ? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM;
FilteredPartition current;
@@ -1856,7 +1868,7 @@ public class StorageProxy implements StorageProxyMBean
}
return consistencyLevel.isSerialConsistency()
- ? readWithPaxos(group, consistencyLevel, queryStartNanoTime)
+ ? readWithConsensus(group, consistencyLevel, queryStartNanoTime)
: readRegular(group, consistencyLevel, queryStartNanoTime);
}
@@ -1870,12 +1882,29 @@ public class StorageProxy implements StorageProxyMBean
return !StorageService.instance.isBootstrapMode();
}
- private static PartitionIterator readWithPaxos(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, long queryStartNanoTime)
+ private static PartitionIterator readWithConsensus(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, long queryStartNanoTime)
throws InvalidRequestException, UnavailableException, ReadFailureException, ReadTimeoutException
{
- return Paxos.useV2()
- ? Paxos.read(group, consistencyLevel)
- : legacyReadWithPaxos(group, consistencyLevel, queryStartNanoTime);
+ if (DatabaseDescriptor.getLegacyPaxosStrategy() == accord)
+ {
+ return readWithAccord(group, consistencyLevel);
+ }
+ else
+ {
+ return Paxos.useV2()
+ ? Paxos.read(group, consistencyLevel)
+ : legacyReadWithPaxos(group, consistencyLevel, queryStartNanoTime);
+ }
+ }
+
+ private static PartitionIterator readWithAccord(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel)
+ {
+ if (group.queries.size() > 1)
+ throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one partition at a time");
+ TxnRead read = TxnRead.createSerialRead(group.queries.get(0));
+ Txn txn = new Txn.InMemory(read.keys(), read, TxnQuery.ALL);
+ TxnData data = AccordService.instance().coordinate(txn, consistencyLevel);
+ return PartitionIterators.singletonIterator(data.get(TxnRead.SERIAL_READ).rowIterator());
}
private static PartitionIterator legacyReadWithPaxos(SinglePartitionReadCommand.Group group, ConsistencyLevel consistencyLevel, long queryStartNanoTime)
diff --git a/src/java/org/apache/cassandra/service/accord/AccordPartialCommand.java b/src/java/org/apache/cassandra/service/accord/AccordPartialCommand.java
index b3ef93d89b..2f708c53b4 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordPartialCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordPartialCommand.java
@@ -42,7 +42,7 @@ import static org.apache.cassandra.utils.CollectionSerializers.serializeCollecti
import static org.apache.cassandra.utils.CollectionSerializers.serializedCollectionSize;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class AccordPartialCommand extends CommandsForKey.TxnIdWithExecuteAt
{
@@ -169,7 +169,7 @@ public class AccordPartialCommand extends CommandsForKey.TxnIdWithExecuteAt
{
int size = Math.toIntExact(AccordSerializerVersion.serializer.serializedSize(version));
size += CommandSerializers.txnId.serializedSize();
- size += serializedSizeNullable(command.executeAt(), version.msgVersion, CommandSerializers.timestamp);
+ size += serializedNullableSize(command.executeAt(), version.msgVersion, CommandSerializers.timestamp);
size += CommandSerializers.status.serializedSize(command.status(), version.msgVersion);
size += serializedCollectionSize(command.deps, version.msgVersion, CommandSerializers.txnId);
return size;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordSerializers.java
index dc48d4bae7..7d6a3fe212 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordSerializers.java
@@ -20,17 +20,21 @@ package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.List;
import org.apache.cassandra.cql3.Lists;
import org.apache.cassandra.cql3.Maps;
import org.apache.cassandra.cql3.Sets;
import org.apache.cassandra.cql3.Term;
-import org.apache.cassandra.db.TypeSizes;
+import org.apache.cassandra.db.ArrayClustering;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.ClusteringPrefix;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
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.ValueAccessor;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -47,6 +51,7 @@ import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
import static com.google.common.primitives.Ints.checkedCast;
+import static org.apache.cassandra.db.TypeSizes.sizeof;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
import static org.apache.cassandra.db.marshal.CollectionType.Kind.LIST;
import static org.apache.cassandra.db.marshal.CollectionType.Kind.MAP;
@@ -70,6 +75,14 @@ public class AccordSerializers
}
}
+ public static <T> ByteBuffer[] serialize(List<T> items, IVersionedSerializer<T> serializer)
+ {
+ ByteBuffer[] result = new ByteBuffer[items.size()];
+ for (int i = 0, mi = items.size(); i < mi; i++)
+ result[i] = serialize(items.get(i), serializer);
+ return result;
+ }
+
public static <T> T deserialize(ByteBuffer bytes, IVersionedSerializer<T> serializer)
{
try (DataInputBuffer in = new DataInputBuffer(bytes, true))
@@ -135,7 +148,7 @@ public class AccordSerializers
String table = in.readUTF();
ByteBuffer name = ByteBufferUtil.readWithShortLength(in);
TableMetadata metadata = Schema.instance.getTableMetadata(keyspace, table);
-
+
// TODO: Can the metadata be null if the table has been dropped?
return metadata.getColumn(name);
}
@@ -144,8 +157,8 @@ public class AccordSerializers
public long serializedSize(ColumnMetadata column, int version)
{
long size = 0;
- size += TypeSizes.sizeof(column.ksName);
- size += TypeSizes.sizeof(column.cfName);
+ size += sizeof(column.ksName);
+ size += sizeof(column.cfName);
size += ByteBufferUtil.serializedSizeWithShortLength(column.name.bytes);
return size;
}
@@ -171,4 +184,73 @@ public class AccordSerializers
return metadata.id.serializedSize();
}
};
-}
+
+ public static final IVersionedSerializer<Clustering<?>> clusteringSerializer = new IVersionedSerializer<Clustering<?>>()
+ {
+ @Override
+ public void serialize(Clustering<?> clustering, DataOutputPlus out, int version) throws IOException
+ {
+ doSerialize(clustering, out);
+ }
+
+ public <V> void doSerialize(Clustering<V> clustering, DataOutputPlus out) throws IOException
+ {
+ if (clustering.kind() == ClusteringPrefix.Kind.STATIC_CLUSTERING)
+ {
+ out.writeBoolean(true);
+ }
+ else
+ {
+ out.writeBoolean(false);
+ out.writeUnsignedVInt(clustering.size());
+ ValueAccessor<V> accessor = clustering.accessor();
+ for (int i = 0; i < clustering.size(); i++)
+ {
+ accessor.writeWithVIntLength(clustering.get(i), out);
+ }
+ }
+ }
+
+ @Override
+ public Clustering<?> deserialize(DataInputPlus in, int version) throws IOException
+ {
+ Clustering<?> clustering;
+ if (in.readBoolean())
+ {
+ clustering = Clustering.STATIC_CLUSTERING;
+ }
+ else
+ {
+ int numComponents = checkedCast(in.readUnsignedVInt());
+ byte[][] components = new byte[numComponents][];
+ for (int ci = 0; ci < numComponents; ci++)
+ {
+ int componentLength = checkedCast(in.readUnsignedVInt());
+ components[ci] = new byte[componentLength];
+ in.readFully(components[ci]);
+ }
+ clustering = new ArrayClustering(components);
+ }
+ return clustering;
+ }
+
+ @Override
+ public long serializedSize(Clustering<?> clustering, int version)
+ {
+ return computeSerializedSize(clustering);
+ }
+
+ private <V> long computeSerializedSize(Clustering<V> clustering)
+ {
+ int size = sizeof(true) + sizeofUnsignedVInt(clustering.size());
+ ValueAccessor<V> accessor = clustering.accessor();
+ for (int i = 0; i < clustering.size(); i++)
+ {
+ int valueSize = accessor.size(clustering.get(i));
+ size += valueSize;
+ size += sizeofUnsignedVInt(valueSize);
+ }
+ return size;
+ }
+ };
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index 0e99689d5a..0135c741c3 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -18,9 +18,9 @@
package org.apache.cassandra.service.accord;
-import java.util.concurrent.ExecutionException;
import java.util.Arrays;
import java.util.Random;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -34,11 +34,11 @@ import accord.local.Node;
import accord.local.ShardDistributor.EvenSplit;
import accord.messages.Request;
import accord.primitives.Txn;
+import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.exceptions.ReadTimeoutException;
-import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.service.accord.api.AccordAgent;
@@ -116,7 +116,11 @@ public class AccordService implements Shutdownable
return TimeUnit.MILLISECONDS.toMicros(Clock.Global.currentTimeMillis());
}
- public TxnData coordinate(Txn txn, QueryOptions options)
+ /**
+ * Consistency level is just echoed back in timeouts, in the future it may be used for interoperability
+ * with non-Accord operations.
+ */
+ public TxnData coordinate(Txn txn, ConsistencyLevel consistencyLevel)
{
try
{
@@ -128,7 +132,7 @@ public class AccordService implements Shutdownable
{
Throwable cause = e.getCause();
if (cause instanceof Timeout)
- throw throwTimeout(txn, options);
+ throw throwTimeout(txn, consistencyLevel);
throw new RuntimeException(cause);
}
catch (InterruptedException e)
@@ -137,14 +141,14 @@ public class AccordService implements Shutdownable
}
catch (TimeoutException e)
{
- throw throwTimeout(txn, options);
+ throw throwTimeout(txn, consistencyLevel);
}
}
- private static RuntimeException throwTimeout(Txn txn, QueryOptions options)
+ private static RuntimeException throwTimeout(Txn txn, ConsistencyLevel consistencyLevel)
{
- throw txn.isWrite() ? new WriteTimeoutException(WriteType.TRANSACTION, options.getConsistency(), 0, 0)
- : new ReadTimeoutException(options.getConsistency(), 0, 0, false);
+ throw txn.isWrite() ? new WriteTimeoutException(WriteType.TRANSACTION, consistencyLevel, 0, 0)
+ : new ReadTimeoutException(consistencyLevel, 0, 0, false);
}
@VisibleForTesting
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
index e793b14eec..5568b9b01e 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
@@ -33,7 +33,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class BeginInvalidationSerializers
{
@@ -92,12 +92,12 @@ public class BeginInvalidationSerializers
@Override
public long serializedSize(InvalidateReply reply, int version)
{
- return serializedSizeNullable(reply.supersededBy, version, CommandSerializers.ballot)
+ return serializedNullableSize(reply.supersededBy, version, CommandSerializers.ballot)
+ CommandSerializers.ballot.serializedSize(reply.accepted, version)
+ CommandSerializers.status.serializedSize(reply.status, version)
+ TypeSizes.BOOL_SIZE
- + serializedSizeNullable(reply.route, version, KeySerializers.route)
- + serializedSizeNullable(reply.homeKey, version, KeySerializers.routingKey);
+ + serializedNullableSize(reply.route, version, KeySerializers.route)
+ + serializedNullableSize(reply.homeKey, version, KeySerializers.routingKey);
}
};
}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
index 0741c1e662..ce8ef96b48 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
@@ -47,7 +47,7 @@ import org.apache.cassandra.service.accord.txn.TxnData;
import static accord.messages.CheckStatus.SerializationSupport.createOk;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class CheckStatusSerializers
{
@@ -166,20 +166,20 @@ public class CheckStatusSerializers
size += CommandSerializers.saveStatus.serializedSize(ok.saveStatus, version);
size += CommandSerializers.ballot.serializedSize(ok.promised, version);
size += CommandSerializers.ballot.serializedSize(ok.accepted, version);
- size += serializedSizeNullable(ok.executeAt, version, CommandSerializers.timestamp);
+ size += serializedNullableSize(ok.executeAt, version, CommandSerializers.timestamp);
size += TypeSizes.BOOL_SIZE;
size += CommandSerializers.durability.serializedSize(ok.durability, version);
- size += serializedSizeNullable(ok.homeKey, version, KeySerializers.routingKey);
- size += serializedSizeNullable(ok.route, version, KeySerializers.route);
+ size += serializedNullableSize(ok.homeKey, version, KeySerializers.routingKey);
+ size += serializedNullableSize(ok.route, version, KeySerializers.route);
if (!(reply instanceof CheckStatusOkFull))
return size;
CheckStatusOkFull okFull = (CheckStatusOkFull) ok;
- size += serializedSizeNullable(okFull.partialTxn, version, CommandSerializers.partialTxn);
- size += serializedSizeNullable(okFull.committedDeps, version, DepsSerializer.partialDeps);
- size += serializedSizeNullable(okFull.writes, version, CommandSerializers.writes);
- size += serializedSizeNullable((TxnData) okFull.result, version, TxnData.serializer);
+ size += serializedNullableSize(okFull.partialTxn, version, CommandSerializers.partialTxn);
+ size += serializedNullableSize(okFull.committedDeps, version, DepsSerializer.partialDeps);
+ size += serializedNullableSize(okFull.writes, version, CommandSerializers.writes);
+ size += serializedNullableSize((TxnData) okFull.result, version, TxnData.serializer);
return size;
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
index b49b898804..ed251016dd 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/CommitSerializers.java
@@ -32,7 +32,7 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class CommitSerializers
{
@@ -64,10 +64,10 @@ public class CommitSerializers
public long serializedBodySize(Commit msg, int version)
{
return CommandSerializers.timestamp.serializedSize(msg.executeAt, version)
- + serializedSizeNullable(msg.partialTxn, version, CommandSerializers.partialTxn)
+ + serializedNullableSize(msg.partialTxn, version, CommandSerializers.partialTxn)
+ DepsSerializer.partialDeps.serializedSize(msg.partialDeps, version)
- + serializedSizeNullable(msg.route, version, KeySerializers.fullRoute)
- + serializedSizeNullable(msg.read, version, ReadDataSerializers.request);
+ + serializedNullableSize(msg.route, version, KeySerializers.fullRoute)
+ + serializedNullableSize(msg.read, version, ReadDataSerializers.request);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
index 251b281621..cce1942a07 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/PreacceptSerializers.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
-
import javax.annotation.Nullable;
import accord.messages.PreAccept;
@@ -37,7 +36,7 @@ import org.apache.cassandra.service.accord.serializers.TxnRequestSerializer.With
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class PreacceptSerializers
{
@@ -67,7 +66,7 @@ public class PreacceptSerializers
public long serializedBodySize(PreAccept msg, int version)
{
return CommandSerializers.partialTxn.serializedSize(msg.partialTxn, version)
- + serializedSizeNullable(msg.route, version, KeySerializers.fullRoute)
+ + serializedNullableSize(msg.route, version, KeySerializers.fullRoute)
+ TypeSizes.sizeofUnsignedVInt(msg.maxEpoch - msg.minEpoch);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
index e9907911a9..fbd00f637e 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
-
import javax.annotation.Nullable;
import accord.api.Result;
@@ -45,7 +44,7 @@ import org.apache.cassandra.service.accord.txn.TxnData;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class RecoverySerializers
{
@@ -73,7 +72,7 @@ public class RecoverySerializers
{
return CommandSerializers.partialTxn.serializedSize(recover.partialTxn, version)
+ CommandSerializers.ballot.serializedSize(recover.ballot, version)
- + serializedSizeNullable(recover.route, version, KeySerializers.fullRoute);
+ + serializedNullableSize(recover.route, version, KeySerializers.fullRoute);
}
};
@@ -149,13 +148,13 @@ public class RecoverySerializers
long size = CommandSerializers.txnId.serializedSize(recoverOk.txnId, version);
size += CommandSerializers.status.serializedSize(recoverOk.status, version);
size += CommandSerializers.ballot.serializedSize(recoverOk.accepted, version);
- size += serializedSizeNullable(recoverOk.executeAt, version, CommandSerializers.timestamp);
+ size += serializedNullableSize(recoverOk.executeAt, version, CommandSerializers.timestamp);
size += DepsSerializer.partialDeps.serializedSize(recoverOk.deps, version);
size += DepsSerializer.deps.serializedSize(recoverOk.earlierCommittedWitness, version);
size += DepsSerializer.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version);
size += TypeSizes.sizeof(recoverOk.rejectsFastPath);
- size += serializedSizeNullable(recoverOk.writes, version, CommandSerializers.writes);
- size += serializedSizeNullable((TxnData) recoverOk.result, version, TxnData.serializer);
+ size += serializedNullableSize(recoverOk.writes, version, CommandSerializers.writes);
+ size += serializedNullableSize((TxnData) recoverOk.result, version, TxnData.serializer);
return size;
}
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 1dd42831f6..fb2a372068 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnCondition.java
@@ -20,10 +20,13 @@ package org.apache.cassandra.service.accord.txn;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Set;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableSet;
@@ -32,6 +35,8 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.cql3.Term;
import org.apache.cassandra.cql3.conditions.ColumnCondition;
+import org.apache.cassandra.cql3.conditions.ColumnCondition.Bound;
+import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
@@ -47,12 +52,18 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.transport.ProtocolVersion;
import org.apache.cassandra.utils.ByteBufferUtil;
+import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.primitives.Ints.checkedCast;
+import static org.apache.cassandra.service.accord.AccordSerializers.clusteringSerializer;
import static org.apache.cassandra.service.accord.AccordSerializers.deserializeCqlCollectionAsTerm;
+import static org.apache.cassandra.service.accord.txn.TxnRead.SERIAL_READ;
import static org.apache.cassandra.utils.CollectionSerializers.deserializeList;
+import static org.apache.cassandra.utils.CollectionSerializers.serializeCollection;
import static org.apache.cassandra.utils.CollectionSerializers.serializeList;
+import static org.apache.cassandra.utils.CollectionSerializers.serializedCollectionSize;
import static org.apache.cassandra.utils.CollectionSerializers.serializedListSize;
+
public abstract class TxnCondition
{
private interface ConditionSerializer<T extends TxnCondition>
@@ -74,9 +85,12 @@ public abstract class TxnCondition
GREATER_THAN(">", Operator.GT),
GREATER_THAN_OR_EQUAL(">=", Operator.GTE),
LESS_THAN("<", Operator.LT),
- LESS_THAN_OR_EQUAL("<=", Operator.LTE);
+ LESS_THAN_OR_EQUAL("<=", Operator.LTE),
+ COLUMN_CONDITIONS("COLUMN_CONDITIONS", null);
+ @Nonnull
private final String symbol;
+ @Nullable
private final Operator operator;
Kind(String symbol, Operator operator)
@@ -105,6 +119,8 @@ public abstract class TxnCondition
return BooleanGroup.serializer;
case NONE:
return None.serializer;
+ case COLUMN_CONDITIONS:
+ return ColumnConditionsAdapter.serializer;
default:
throw new IllegalArgumentException("No serializer exists for kind " + this);
}
@@ -301,6 +317,62 @@ public abstract class TxnCondition
};
}
+ public static class ColumnConditionsAdapter extends TxnCondition {
+ @Nonnull
+ public final Collection<Bound> bounds;
+
+ @Nonnull
+ public final Clustering<?> clustering;
+
+ public ColumnConditionsAdapter(Clustering<?> clustering, Collection<Bound> bounds)
+ {
+ super(Kind.COLUMN_CONDITIONS);
+ checkNotNull(bounds);
+ checkNotNull(clustering);
+ this.bounds = bounds;
+ this.clustering = clustering;
+ }
+
+ @Override
+ public boolean applies(@Nonnull TxnData data)
+ {
+ checkNotNull(data);
+ FilteredPartition partition = data.get(SERIAL_READ);
+ Row row = partition.getRow(clustering);
+ for (Bound bound : bounds)
+ {
+ if (!bound.appliesTo(row))
+ return false;
+ }
+ return true;
+ }
+
+ private static final ConditionSerializer<ColumnConditionsAdapter> serializer = new ConditionSerializer<ColumnConditionsAdapter>()
+ {
+ @Override
+ public void serialize(ColumnConditionsAdapter condition, DataOutputPlus out, int version) throws IOException
+ {
+ clusteringSerializer.serialize(condition.clustering, out, version);
+ serializeCollection(condition.bounds, out, version, Bound.serializer);
+ }
+
+ @Override
+ public ColumnConditionsAdapter deserialize(DataInputPlus in, int version, Kind ignored) throws IOException
+ {
+ Clustering<?> clustering = clusteringSerializer.deserialize(in, version);
+ List<Bound> bounds = deserializeList(in, version, Bound.serializer);
+ return new ColumnConditionsAdapter(clustering, bounds);
+ }
+
+ @Override
+ public long serializedSize(ColumnConditionsAdapter condition, int version)
+ {
+ return clusteringSerializer.serializedSize(condition.clustering, version)
+ + serializedCollectionSize(condition.bounds, version, Bound.serializer);
+ }
+ };
+ }
+
public static class Value extends TxnCondition
{
private static final Set<Kind> KINDS = ImmutableSet.of(Kind.EQUAL, Kind.NOT_EQUAL,
@@ -380,7 +452,7 @@ public abstract class TxnCondition
return ColumnCondition.Bound.compareWithOperator(kind.operator, reference.getFieldSelectionType(), value, cell.buffer());
}
- return new ColumnCondition.MultiCellUdtBound(column, kind.operator, Collections.singletonList(value), version).appliesTo(row);
+ return new ColumnCondition.MultiCellUdtBound(column, kind.operator, Collections.singletonList(value)).appliesTo(row);
}
throw new UnsupportedOperationException("Unsupported complex type: " + type);
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnDataName.java b/src/java/org/apache/cassandra/service/accord/txn/TxnDataName.java
index de672cad2e..224df3230f 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnDataName.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnDataName.java
@@ -24,7 +24,10 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.io.IVersionedSerializer;
@@ -34,8 +37,13 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.ObjectSizes;
+import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.primitives.Ints.checkedCast;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
+import static org.apache.cassandra.service.accord.AccordSerializers.clusteringSerializer;
+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 TxnDataName implements Comparable<TxnDataName>
{
@@ -71,13 +79,27 @@ public class TxnDataName implements Comparable<TxnDataName>
}
}
+ @Nonnull
private final Kind kind;
+
+ @Nonnull
private final String[] parts;
- public TxnDataName(Kind kind, String... parts)
+ @Nullable
+ private final Clustering<?> clustering;
+
+ public TxnDataName(@Nonnull Kind kind, @Nonnull String... parts)
+ {
+ this(kind, null, parts);
+ }
+
+ public TxnDataName(@Nonnull Kind kind, @Nullable Clustering<?> clustering, @Nonnull String... parts)
{
+ checkNotNull(kind);
+ checkNotNull(parts);
this.kind = kind;
this.parts = parts;
+ this.clustering = clustering;
}
public static TxnDataName user(String name)
@@ -207,6 +229,7 @@ public class TxnDataName implements Comparable<TxnDataName>
out.writeUnsignedVInt(t.parts.length);
for (String part : t.parts)
out.writeUTF(part);
+ serializeNullable(t.clustering, out, version, clusteringSerializer);
}
@Override
@@ -217,7 +240,8 @@ public class TxnDataName implements Comparable<TxnDataName>
String[] parts = new String[length];
for (int i = 0; i < length; i++)
parts[i] = in.readUTF();
- return new TxnDataName(kind, parts);
+ Clustering<?> clustering = deserializeNullable(in, version, clusteringSerializer);
+ return new TxnDataName(kind, clustering, parts);
}
@Override
@@ -226,6 +250,7 @@ public class TxnDataName implements Comparable<TxnDataName>
int size = Byte.BYTES + sizeofUnsignedVInt(t.parts.length);
for (String part : t.parts)
size += TypeSizes.sizeof(part);
+ size += serializedNullableSize(t.clustering, version, clusteringSerializer);
return size;
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
index fe98121f3f..8eda32ad82 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
@@ -101,7 +101,7 @@ public class TxnNamedRead extends AbstractSerialized<ReadCommand>
return "TxnNamedRead{name='" + name + '\'' + ", key=" + key + ", update=" + get() + '}';
}
- public TxnDataName name()
+ public TxnDataName txnDataName()
{
return name;
}
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 d00da39735..29b28606ba 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java
@@ -19,7 +19,6 @@
package org.apache.cassandra.service.accord.txn;
import java.io.IOException;
-
import javax.annotation.Nullable;
import com.google.common.base.Preconditions;
@@ -36,10 +35,18 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.utils.ObjectSizes;
+import static com.google.common.base.Preconditions.checkNotNull;
+
public abstract class TxnQuery implements Query
{
public static final TxnQuery ALL = new TxnQuery()
{
+ @Override
+ protected byte type()
+ {
+ return 1;
+ }
+
@Override
public Result compute(TxnId txnId, Data data, @Nullable Read read, @Nullable Update update)
{
@@ -49,6 +56,12 @@ public abstract class TxnQuery implements Query
public static final TxnQuery NONE = new TxnQuery()
{
+ @Override
+ protected byte type()
+ {
+ return 2;
+ }
+
@Override
public Result compute(TxnId txnId, Data data, @Nullable Read read, @Nullable Update update)
{
@@ -56,10 +69,37 @@ public abstract class TxnQuery implements Query
}
};
+ public static final TxnQuery CONDITION = new TxnQuery()
+ {
+ @Override
+ protected byte type()
+ {
+ return 3;
+ }
+
+ @Override
+ public Result compute(TxnId txnId, Data data, @Nullable Read read, Update update)
+ {
+ checkNotNull(txnId, "txnId should not be null");
+ checkNotNull(data, "data should not be null");
+ checkNotNull(update, "update should not be null");
+ TxnUpdate txnUpdate = (TxnUpdate)update;
+ boolean conditionCheck = txnUpdate.checkCondition(data);
+ // If the condition applied an empty result indicates success
+ if (conditionCheck)
+ return new TxnData();
+ else
+ // If it failed to apply the partition contents (if present) are returned and it indicates failure
+ return (TxnData)data;
+ }
+ };
+
private static final long SIZE = ObjectSizes.measure(ALL);
private TxnQuery() {}
+ abstract protected byte type();
+
public long estimatedSizeOnHeap()
{
return SIZE;
@@ -70,8 +110,8 @@ public abstract class TxnQuery implements Query
@Override
public void serialize(TxnQuery query, DataOutputPlus out, int version) throws IOException
{
- Preconditions.checkArgument(query == null || query == ALL || query == NONE);
- out.writeByte(query == null ? 0 : query == ALL ? 1 : 2);
+ Preconditions.checkArgument(query == null || query == ALL || query == NONE || query == CONDITION);
+ out.writeByte(query == null ? 0 : query.type());
}
@Override
@@ -83,13 +123,14 @@ public abstract class TxnQuery implements Query
case 0: return null;
case 1: return ALL;
case 2: return NONE;
+ case 3: return CONDITION;
}
}
@Override
public long serializedSize(TxnQuery query, int version)
{
- Preconditions.checkArgument(query == null || query == ALL || query == NONE);
+ Preconditions.checkArgument(query == null || query == ALL || query == NONE || query == CONDITION);
return TypeSizes.sizeof((byte)2);
}
};
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
index 94ad0693f1..1b42635021 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
@@ -24,6 +24,8 @@ import java.util.Collections;
import java.util.List;
import java.util.function.BiConsumer;
+import com.google.common.collect.ImmutableList;
+
import accord.api.Data;
import accord.api.DataStore;
import accord.api.Key;
@@ -33,6 +35,7 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@@ -51,6 +54,9 @@ import static org.apache.cassandra.utils.Simulate.With.MONITORS;
public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
{
+ // There is only potentially one partition in a CAS and SERIAL/LOCAL_SERIAL read
+ public static final String SERIAL_READ_NAME = "SERIAL_READ";
+ public static final TxnDataName SERIAL_READ = TxnDataName.user(SERIAL_READ_NAME);
private static final long EMPTY_SIZE = ObjectSizes.measure(new TxnRead(new TxnNamedRead[0], null));
private final Keys txnKeys;
@@ -67,6 +73,12 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
this.txnKeys = txnKeys;
}
+ public static TxnRead createSerialRead(SinglePartitionReadCommand readCommand)
+ {
+ TxnNamedRead read = new TxnNamedRead(SERIAL_READ, readCommand);
+ return new TxnRead(ImmutableList.of(read), Keys.of(read.key()));
+ }
+
public long estimatedSizeOnHeap()
{
long size = EMPTY_SIZE;
@@ -78,7 +90,7 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
@Override
int compareNonKeyFields(TxnNamedRead left, TxnNamedRead right)
{
- return left.name().compareTo(right.name());
+ return left.txnDataName().compareTo(right.txnDataName());
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnReference.java b/src/java/org/apache/cassandra/service/accord/txn/TxnReference.java
index 5e9c497140..6871b7261d 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnReference.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnReference.java
@@ -57,6 +57,11 @@ public class TxnReference
this.path = path;
}
+ public TxnReference(TxnDataName tuple, ColumnMetadata column)
+ {
+ this(tuple, column, null);
+ }
+
@Override
public boolean equals(Object o)
{
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 0eb3e781aa..025217d1e4 100644
--- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
+++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
@@ -51,10 +51,10 @@ import static java.lang.Math.toIntExact;
import static org.apache.cassandra.service.accord.AccordSerializers.serialize;
import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
+import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
-import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
public class TxnUpdate implements Update
{
@@ -64,6 +64,9 @@ public class TxnUpdate implements Update
private final ByteBuffer[] fragments;
private final ByteBuffer condition;
+ // Memoize computation of condition
+ private Boolean conditionResult;
+
public TxnUpdate(List<TxnWrite.Fragment> fragments, TxnCondition condition)
{
// TODO: Figure out a way to shove keys into TxnCondition, and have it implement slice/merge.
@@ -169,9 +172,7 @@ public class TxnUpdate implements Update
@Override
public Write apply(Data data)
{
- TxnCondition condition = AccordSerializers.deserialize(this.condition, TxnCondition.serializer);
-
- if (!condition.applies((TxnData) data))
+ if (!checkCondition(data))
return TxnWrite.EMPTY;
List<TxnWrite.Fragment> fragments = deserialize(this.fragments, TxnWrite.Fragment.serializer);
@@ -192,7 +193,7 @@ public class TxnUpdate implements Update
{
KeySerializers.keys.serialize(update.keys, out, version);
writeWithVIntLength(update.condition, out);
- serializeArray(update.fragments, out, version, ByteBufferUtil.vintSerializer);
+ serializeArray(update.fragments, out, version, ByteBufferUtil.byteBufferSerializer);
}
@Override
@@ -200,7 +201,7 @@ public class TxnUpdate implements Update
{
Keys keys = KeySerializers.keys.deserialize(in, version);
ByteBuffer condition = readWithVIntLength(in);
- ByteBuffer[] fragments = deserializeArray(in, version, ByteBufferUtil.vintSerializer, ByteBuffer[]::new);
+ ByteBuffer[] fragments = deserializeArray(in, version, ByteBufferUtil.byteBufferSerializer, ByteBuffer[]::new);
return new TxnUpdate(keys, fragments, condition);
}
@@ -209,7 +210,8 @@ public class TxnUpdate implements Update
{
long size = KeySerializers.keys.serializedSize(update.keys, version);
size += serializedSizeWithVIntLength(update.condition);
- size += serializedArraySize(update.fragments, version, ByteBufferUtil.vintSerializer);
+ size += serializedArraySize(update.fragments, version, ByteBufferUtil.byteBufferSerializer);
+ assert(ByteBufferUtil.serialized(this, update, version).remaining() == size);
return size;
}
};
@@ -288,4 +290,15 @@ public class TxnUpdate implements Update
result.addAll(deserialize(bytes, serializer));
return result;
}
+
+ // maybeCheckCondition? checkConditionMemoized?
+ public boolean checkCondition(Data data)
+ {
+ // Assert data that was memoized is same as data that is provided?
+ if (conditionResult != null)
+ return conditionResult;
+ TxnCondition condition = AccordSerializers.deserialize(this.condition, TxnCondition.serializer);
+ conditionResult = condition.applies((TxnData) data);
+ return conditionResult;
+ }
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java b/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
index a8812ded69..dcbd8be587 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosPrepareRefresh.java
@@ -46,7 +46,7 @@ import static org.apache.cassandra.service.paxos.PaxosRequestCallback.shouldExec
import static org.apache.cassandra.utils.FBUtilities.getBroadcastAddressAndPort;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
/**
* Nodes that have promised in response to our prepare, may be missing the latestCommit, meaning we cannot be sure the
@@ -237,7 +237,7 @@ public class PaxosPrepareRefresh implements RequestCallbackWithFailure<PaxosPrep
public long serializedSize(Response response, int version)
{
- return serializedSizeNullable(response.isSupersededBy, version, Ballot.Serializer.instance);
+ return serializedNullableSize(response.isSupersededBy, version, Ballot.Serializer.instance);
}
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java b/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
index ab757cbdc5..94fa9dc3a4 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosRepair.java
@@ -20,7 +20,12 @@ package org.apache.cassandra.service.paxos;
import java.io.IOException;
-import java.util.*;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
@@ -63,20 +68,30 @@ import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MonotonicClock;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static org.apache.cassandra.concurrent.ExecutorFactory.Global.executorFactory;
import static org.apache.cassandra.config.CassandraRelevantProperties.PAXOS_REPAIR_RETRY_TIMEOUT_IN_MS;
import static org.apache.cassandra.exceptions.RequestFailureReason.UNKNOWN;
import static org.apache.cassandra.net.Verb.PAXOS2_REPAIR_REQ;
-import static java.util.concurrent.TimeUnit.NANOSECONDS;
-import static org.apache.cassandra.service.paxos.Commit.*;
+import static org.apache.cassandra.service.paxos.Commit.Accepted;
+import static org.apache.cassandra.service.paxos.Commit.Committed;
+import static org.apache.cassandra.service.paxos.Commit.Proposal;
+import static org.apache.cassandra.service.paxos.Commit.isAfter;
+import static org.apache.cassandra.service.paxos.Commit.latest;
+import static org.apache.cassandra.service.paxos.Commit.timestampsClash;
import static org.apache.cassandra.service.paxos.ContentionStrategy.Type.REPAIR;
import static org.apache.cassandra.service.paxos.ContentionStrategy.waitUntilForContention;
-import static org.apache.cassandra.service.paxos.Paxos.*;
-import static org.apache.cassandra.service.paxos.PaxosPrepare.*;
+import static org.apache.cassandra.service.paxos.Paxos.Participants;
+import static org.apache.cassandra.service.paxos.Paxos.isInRangeAndShouldProcess;
+import static org.apache.cassandra.service.paxos.Paxos.staleBallotNewerThan;
+import static org.apache.cassandra.service.paxos.PaxosPrepare.FoundIncompleteAccepted;
+import static org.apache.cassandra.service.paxos.PaxosPrepare.FoundIncompleteCommitted;
+import static org.apache.cassandra.service.paxos.PaxosPrepare.Status;
+import static org.apache.cassandra.service.paxos.PaxosPrepare.prepareWithBallot;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
-import static org.apache.cassandra.utils.NullableSerializer.serializedSizeNullable;
+import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
/**
* Facility to finish any in-progress paxos transaction, and ensure that a quorum of nodes agree on the most recent operation.
@@ -629,7 +644,7 @@ public class PaxosRepair extends AbstractPaxosRepair
public long serializedSize(Response response, int version)
{
return Ballot.sizeInBytes()
- + serializedSizeNullable(response.acceptedButNotCommitted, version, Accepted.serializer)
+ + serializedNullableSize(response.acceptedButNotCommitted, version, Accepted.serializer)
+ Committed.serializer.serializedSize(response.committed, version);
}
}
diff --git a/src/java/org/apache/cassandra/service/paxos/PaxosState.java b/src/java/org/apache/cassandra/service/paxos/PaxosState.java
index e802cd070f..314796f2a8 100644
--- a/src/java/org/apache/cassandra/service/paxos/PaxosState.java
+++ b/src/java/org/apache/cassandra/service/paxos/PaxosState.java
@@ -26,7 +26,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.BiConsumer;
import java.util.function.Function;
-
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -37,12 +36,18 @@ import com.google.common.primitives.Ints;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.cassandra.concurrent.ImmediateExecutor;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.metrics.PaxosMetrics;
-import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Directories;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.SystemKeyspace;
+import org.apache.cassandra.db.WriteType;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.exceptions.RequestTimeoutException;
import org.apache.cassandra.exceptions.WriteTimeoutException;
+import org.apache.cassandra.metrics.PaxosMetrics;
+import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.paxos.uncommitted.PaxosBallotTracker;
import org.apache.cassandra.service.paxos.uncommitted.PaxosStateTracker;
import org.apache.cassandra.service.paxos.uncommitted.PaxosUncommittedTracker;
@@ -50,15 +55,23 @@ import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.Nemesis;
import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.config.Config.PaxosStatePurging.gc_grace;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
-import static org.apache.cassandra.service.paxos.Commit.*;
-import static org.apache.cassandra.service.paxos.PaxosState.MaybePromise.Outcome.*;
+import static org.apache.cassandra.service.paxos.Commit.Accepted;
import static org.apache.cassandra.service.paxos.Commit.Accepted.latestAccepted;
+import static org.apache.cassandra.service.paxos.Commit.AcceptedWithTTL;
+import static org.apache.cassandra.service.paxos.Commit.Agreed;
+import static org.apache.cassandra.service.paxos.Commit.Committed;
import static org.apache.cassandra.service.paxos.Commit.Committed.latestCommitted;
+import static org.apache.cassandra.service.paxos.Commit.CommittedWithTTL;
+import static org.apache.cassandra.service.paxos.Commit.Proposal;
import static org.apache.cassandra.service.paxos.Commit.isAfter;
+import static org.apache.cassandra.service.paxos.Commit.latest;
+import static org.apache.cassandra.service.paxos.PaxosState.MaybePromise.Outcome.PERMIT_READ;
+import static org.apache.cassandra.service.paxos.PaxosState.MaybePromise.Outcome.PROMISE;
+import static org.apache.cassandra.service.paxos.PaxosState.MaybePromise.Outcome.REJECT;
+import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* We save to memory the result of each operation before persisting to disk, however each operation that performs
diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
index a2102646bb..71b0fb5db5 100644
--- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
+++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
@@ -23,7 +23,9 @@ package org.apache.cassandra.utils;
* afterward, and ensure the tests still pass.
*/
-import java.io.*;
+import java.io.DataInput;
+import java.io.IOException;
+import java.io.InputStream;
import java.net.InetAddress;
import java.nio.ByteBuffer;
import java.nio.charset.CharacterCodingException;
@@ -31,7 +33,6 @@ import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Date;
-import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -44,14 +45,14 @@ import net.nicoulaj.compilecommand.annotations.Inline;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.BooleanType;
import org.apache.cassandra.db.marshal.BytesType;
-import org.apache.cassandra.db.marshal.DateType;
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.TimestampType;
import org.apache.cassandra.io.IVersionedSerializer;
-import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.compress.BufferType;
+import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.FileUtils;
@@ -359,6 +360,17 @@ public class ByteBufferUtil
out.writeUnsignedVInt32(bytes.remaining());
out.write(bytes);
}
+ public static void writeWithVIntLengthAndNull(ByteBuffer bytes, DataOutputPlus out) throws IOException
+ {
+ if (bytes == null)
+ {
+ out.writeVInt(-1);
+ return;
+ }
+
+ out.writeVInt(bytes.remaining());
+ out.write(bytes);
+ }
public static void writeWithShortLength(ByteBuffer buffer, DataOutputPlus out) throws IOException
{
@@ -389,12 +401,6 @@ public class ByteBufferUtil
return ByteBufferUtil.read(in, length);
}
- public static int serializedSizeWithLength(ByteBuffer buffer)
- {
- int size = buffer.remaining();
- return TypeSizes.sizeof(size) + size;
- }
-
public static int serializedSizeWithVIntLength(ByteBuffer buffer)
{
int size = buffer.remaining();
@@ -928,7 +934,20 @@ public class ByteBufferUtil
return true;
}
- public static final IVersionedSerializer<ByteBuffer> vintSerializer = new IVersionedSerializer<ByteBuffer>()
+ public static <T> ByteBuffer serialized(IVersionedSerializer<T> serializer, T value, int version)
+ {
+ try (DataOutputBuffer dob = new DataOutputBuffer())
+ {
+ serializer.serialize(value, dob, version);
+ return dob.buffer();
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static final IVersionedSerializer<ByteBuffer> byteBufferSerializer = new IVersionedSerializer<ByteBuffer>()
{
@Override
public void serialize(ByteBuffer bytes, DataOutputPlus out, int version) throws IOException
@@ -948,4 +967,6 @@ public class ByteBufferUtil
return serializedSizeWithVIntLength(bytes);
}
};
+
+ public static final IVersionedSerializer<ByteBuffer> nullableByteBufferSerializer = NullableSerializer.wrap(byteBufferSerializer);
}
diff --git a/src/java/org/apache/cassandra/utils/CollectionSerializers.java b/src/java/org/apache/cassandra/utils/CollectionSerializers.java
index 2e92aee386..93fa02a04d 100644
--- a/src/java/org/apache/cassandra/utils/CollectionSerializers.java
+++ b/src/java/org/apache/cassandra/utils/CollectionSerializers.java
@@ -134,6 +134,11 @@ public class CollectionSerializers
return i -> i == 0 ? Collections.emptyList() : new ArrayList<>(i);
}
+ public static int readCollectionSize(DataInputPlus in, int version) throws IOException
+ {
+ return checkedCast(in.readUnsignedVInt());
+ }
+
/*
* Private to push auto-complete to the convenience methods
* Feel free to make public if there is a weird collection you want to use
diff --git a/src/java/org/apache/cassandra/utils/NullableSerializer.java b/src/java/org/apache/cassandra/utils/NullableSerializer.java
index a286fe5765..7d834be995 100644
--- a/src/java/org/apache/cassandra/utils/NullableSerializer.java
+++ b/src/java/org/apache/cassandra/utils/NullableSerializer.java
@@ -39,7 +39,7 @@ public class NullableSerializer
return in.readBoolean() ? serializer.deserialize(in, version) : null;
}
- public static <T> long serializedSizeNullable(T value, int version, IVersionedSerializer<T> serializer)
+ public static <T> long serializedNullableSize(T value, int version, IVersionedSerializer<T> serializer)
{
return value != null
? TypeSizes.sizeof(true) + serializer.serializedSize(value, version)
@@ -61,7 +61,7 @@ public class NullableSerializer
public long serializedSize(T t, int version)
{
- return serializedSizeNullable(t, version, wrap);
+ return serializedNullableSize(t, version, wrap);
}
};
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java
index aa3a19f95b..3f8edba622 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java
@@ -31,9 +31,15 @@ import com.google.common.base.Splitter;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
+import org.junit.BeforeClass;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import accord.primitives.Unseekables;
import accord.topology.Topologies;
+import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.functions.types.utils.Bytes;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.ListType;
@@ -41,27 +47,20 @@ 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.dht.Murmur3Partitioner;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.distributed.api.QueryResults;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTestUtils;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.assertj.core.api.Assertions;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.cassandra.cql3.CQLTester;
-import org.apache.cassandra.distributed.api.ConsistencyLevel;
-import org.apache.cassandra.distributed.api.SimpleQueryResult;
-import org.apache.cassandra.service.accord.AccordService;
import static org.apache.cassandra.cql3.CQLTester.row;
+import static org.apache.cassandra.distributed.util.QueryResultUtil.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.apache.cassandra.distributed.util.QueryResultUtil.assertThat;
-
public class AccordCQLTest extends AccordTestBase
{
private static final Logger logger = LoggerFactory.getLogger(AccordCQLTest.class);
@@ -324,7 +323,7 @@ public class AccordCQLTest extends AccordTestBase
cluster ->
{
cluster.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, " + lhs + ");", ConsistencyLevel.ALL);
-
+
String query = "BEGIN TRANSACTION\n" +
" LET row1 = (SELECT v FROM " + currentTable + " WHERE k = ? LIMIT 1);\n" +
" SELECT row1.v;\n" +
@@ -333,7 +332,7 @@ public class AccordCQLTest extends AccordTestBase
" END IF\n" +
"COMMIT TRANSACTION";
assertRowEqualsWithPreemptedRetry(cluster, new Object[] { lhs }, query, 0, rhs, 1, 0, 1);
-
+
String check = "BEGIN TRANSACTION\n" +
" SELECT * FROM " + currentTable + " WHERE k = ? AND c = ?;\n" +
"COMMIT TRANSACTION";
@@ -1533,7 +1532,7 @@ public class AccordCQLTest extends AccordTestBase
{
test(cluster -> {
cluster.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, 0);", ConsistencyLevel.ALL);
-
+
String cql = "BEGIN TRANSACTION\n" +
" LET row1 = (SELECT * FROM " + currentTable + " WHERE k = 0 LIMIT ?);\n" +
" SELECT row1.v;\n" +
@@ -2333,7 +2332,7 @@ public class AccordCQLTest extends AccordTestBase
{
test(cluster -> {
cluster.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, 0)", ConsistencyLevel.ALL);
-
+
String cql = "BEGIN TRANSACTION\n" +
" LET a = (SELECT * FROM " + currentTable + " WHERE k=0 AND c=0);\n" +
" IF a IS NOT NULL THEN\n" +
@@ -2351,7 +2350,7 @@ public class AccordCQLTest extends AccordTestBase
{
test(cluster -> {
cluster.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (0, 0, 0)", ConsistencyLevel.ALL);
-
+
String cql = "BEGIN TRANSACTION\n" +
" LET a = (SELECT * FROM " + currentTable + " WHERE k=0 AND c=0);\n" +
" IF a IS NOT NULL THEN\n" +
@@ -2361,4 +2360,30 @@ public class AccordCQLTest extends AccordTestBase
assertEmptyWithPreemptedRetry(cluster, cql);
});
}
+
+ @Test
+ public void testCASAndSerialRead() throws Exception
+ {
+ test("CREATE TABLE " + currentTable + " (id int, c int, v int, s int static, PRIMARY KEY ((id), c));",
+ cluster -> {
+ ICoordinator coordinator = cluster.coordinator(1);
+ int startingAccordCoordinateCount = getAccordCoordinateCount();
+ coordinator.execute("INSERT INTO " + currentTable + " (id, c, v, s) VALUES (1, 2, 3, 5);", ConsistencyLevel.ALL);
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 3, 5);
+ assertRowEquals(cluster, new Object[]{true}, "UPDATE " + currentTable + " SET v = 4 WHERE id = 1 AND c = 2 IF v = 3");
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 4, 5);
+ assertRowEquals(cluster, new Object[]{ false, 4 }, "UPDATE " + currentTable + " SET v = 4 WHERE id = 1 AND c = 2 IF v = 3");
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 4, 5);
+
+ // Test working with a static column
+ assertRowEquals(cluster, new Object[]{ false, 5 }, "UPDATE " + currentTable + " SET v = 5 WHERE id = 1 AND c = 2 IF s = 4");
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 4, 5);
+ assertRowEquals(cluster, new Object[]{true}, "UPDATE " + currentTable + " SET v = 5 WHERE id = 1 AND c = 2 IF s = 5");
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 5, 5);
+ assertRowEquals(cluster, new Object[]{true}, "UPDATE " + currentTable + " SET s = 6 WHERE id = 1 IF s = 5");
+ assertRowSerial(cluster, "SELECT id, c, v, s FROM " + currentTable + " WHERE id = 1 AND c = 2", 1, 2, 5, 6);
+ // Make sure all the consensus using queries actually were run on Accord
+ assertEquals( 11, getAccordCoordinateCount() - startingAccordCoordinateCount);
+ });
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
index 38cca646fc..f039e0c340 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordTestBase.java
@@ -25,15 +25,6 @@ import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.atomic.AtomicInteger;
-import accord.primitives.Txn;
-import net.bytebuddy.ByteBuddy;
-import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
-import net.bytebuddy.implementation.MethodDelegation;
-import net.bytebuddy.implementation.bind.annotation.SuperCall;
-import net.bytebuddy.implementation.bind.annotation.This;
-import org.apache.cassandra.cql3.transactions.ReferenceValue;
-import org.apache.cassandra.distributed.api.QueryResults;
-import org.apache.cassandra.distributed.util.QueryResultUtil;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
@@ -41,20 +32,30 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.coordinate.Preempted;
+import accord.primitives.Txn;
+import net.bytebuddy.ByteBuddy;
+import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
+import net.bytebuddy.implementation.MethodDelegation;
+import net.bytebuddy.implementation.bind.annotation.SuperCall;
+import net.bytebuddy.implementation.bind.annotation.This;
import org.apache.cassandra.cql3.statements.ModificationStatement;
import org.apache.cassandra.cql3.statements.TransactionStatement;
+import org.apache.cassandra.cql3.transactions.ReferenceValue;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.Feature;
+import org.apache.cassandra.distributed.api.QueryResults;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.test.TestBaseImpl;
+import org.apache.cassandra.distributed.util.QueryResultUtil;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTestUtils;
+import org.apache.cassandra.service.accord.txn.TxnData;
import org.apache.cassandra.utils.AssertionUtils;
import org.apache.cassandra.utils.FailingConsumer;
-import org.assertj.core.util.Arrays;
import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
import static org.junit.Assert.assertArrayEquals;
public abstract class AccordTestBase extends TestBaseImpl
@@ -85,10 +86,10 @@ public abstract class AccordTestBase extends TestBaseImpl
currentTable = KEYSPACE + ".tbl" + COUNTER.getAndIncrement();
}
- protected static void assertRow(Cluster cluster, String query, int k, int c, int v)
+ protected static void assertRowSerial(Cluster cluster, String query, int k, int c, int v, int s)
{
- Object[][] result = cluster.coordinator(1).execute(query, ConsistencyLevel.QUORUM);
- assertArrayEquals(new Object[]{new Object[] {k, c, v}}, result);
+ Object[][] result = cluster.coordinator(1).execute(query, ConsistencyLevel.SERIAL);
+ assertArrayEquals(new Object[]{new Object[] {k, c, v, s}}, result);
}
protected void test(String tableDDL, FailingConsumer<Cluster> fn) throws Exception
@@ -120,14 +121,20 @@ public abstract class AccordTestBase extends TestBaseImpl
test("CREATE TABLE " + currentTable + " (k int, c int, v int, primary key (k, c))", fn);
}
+ protected int getAccordCoordinateCount()
+ {
+ return SHARED_CLUSTER.get(1).callOnInstance(() -> BBAccordCoordinateCountHelper.count.get());
+ }
+
private static Cluster createCluster() throws IOException
{
// need to up the timeout else tests get flaky
// disable vnode for now, but should enable before trunk
return init(Cluster.build(2)
.withoutVNodes()
- .withConfig(c -> c.with(Feature.NETWORK).set("write_request_timeout", "10s").set("transaction_timeout", "15s"))
+ .withConfig(c -> c.with(Feature.NETWORK).set("write_request_timeout", "10s").set("transaction_timeout", "15s").set("legacy_paxos_strategy", "accord"))
.withInstanceInitializer(EnforceUpdateDoesNotPerformRead::install)
+ .withInstanceInitializer(BBAccordCoordinateCountHelper::install)
.start());
}
@@ -242,6 +249,27 @@ public abstract class AccordTestBase extends TestBaseImpl
return map;
}
}
-
+
+ public static class BBAccordCoordinateCountHelper
+ {
+ static AtomicInteger count = new AtomicInteger();
+ static void install(ClassLoader cl, int nodeNumber)
+ {
+ if (nodeNumber != 1)
+ return;
+ new ByteBuddy().rebase(AccordService.class)
+ .method(named("coordinate").and(takesArguments(2)))
+ .intercept(MethodDelegation.to(BBAccordCoordinateCountHelper.class))
+ .make()
+ .load(cl, ClassLoadingStrategy.Default.INJECTION);
+ }
+
+ public static TxnData coordinate(Txn txn, @SuperCall Callable<TxnData> actual) throws Exception
+ {
+ count.incrementAndGet();
+ return actual.call();
+ }
+ }
+
protected abstract Logger logger();
}
diff --git a/test/unit/org/apache/cassandra/cql3/conditions/ColumnConditionTest.java b/test/unit/org/apache/cassandra/cql3/conditions/ColumnConditionTest.java
index dd40cf81db..870345a17e 100644
--- a/test/unit/org/apache/cassandra/cql3/conditions/ColumnConditionTest.java
+++ b/test/unit/org/apache/cassandra/cql3/conditions/ColumnConditionTest.java
@@ -17,162 +17,422 @@
*/
package org.apache.cassandra.cql3.conditions;
+import java.io.IOException;
import java.nio.ByteBuffer;
-import java.util.*;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.SortedSet;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import org.junit.Assert;
import org.junit.Test;
-import org.apache.cassandra.cql3.*;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.Constants;
+import org.apache.cassandra.cql3.Lists;
+import org.apache.cassandra.cql3.Maps;
+import org.apache.cassandra.cql3.Operator;
+import org.apache.cassandra.cql3.QueryOptions;
+import org.apache.cassandra.cql3.Sets;
+import org.apache.cassandra.cql3.Term;
+import org.apache.cassandra.cql3.Terms;
+import org.apache.cassandra.cql3.UserTypes;
import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.marshal.AbstractType;
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.rows.*;
+import org.apache.cassandra.db.marshal.UserType;
+import org.apache.cassandra.db.rows.BTreeRow;
+import org.apache.cassandra.db.rows.BufferCell;
+import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.CellPath;
+import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.io.IVersionedSerializer;
+import org.apache.cassandra.io.util.DataInputBuffer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.serializers.ListSerializer;
+import org.apache.cassandra.serializers.MapSerializer;
+import org.apache.cassandra.serializers.SetSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.TimeUUID;
+import static org.apache.cassandra.cql3.Operator.CONTAINS;
+import static org.apache.cassandra.cql3.Operator.CONTAINS_KEY;
+import static org.apache.cassandra.cql3.Operator.EQ;
+import static org.apache.cassandra.cql3.Operator.GT;
+import static org.apache.cassandra.cql3.Operator.GTE;
+import static org.apache.cassandra.cql3.Operator.LT;
+import static org.apache.cassandra.cql3.Operator.LTE;
+import static org.apache.cassandra.cql3.Operator.NEQ;
+import static org.apache.cassandra.transport.ProtocolVersion.CURRENT;
+import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-import static org.apache.cassandra.cql3.Operator.*;
-import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
-
-public class ColumnConditionTest
+public class ColumnConditionTest extends CQLTester
{
public static final ByteBuffer ZERO = Int32Type.instance.fromString("0");
public static final ByteBuffer ONE = Int32Type.instance.fromString("1");
public static final ByteBuffer TWO = Int32Type.instance.fromString("2");
- private static Row newRow(ColumnMetadata definition, ByteBuffer value)
+ private static final ListType<Integer> listType = ListType.getInstance(Int32Type.instance, true);
+ private static final MapType<Integer, Integer> mapType = MapType.getInstance(Int32Type.instance, Int32Type.instance, true);
+ private static final SetType<Integer> setType = SetType.getInstance(Int32Type.instance, true);
+
+ private Row newRow(ColumnMetadata definition, ByteBuffer value)
{
- BufferCell cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, value, null);
+ BufferCell cell;
+ if (definition.type.isUDT() )
+ {
+ if (definition.type.isMultiCell()) {
+ CellPath cellPath = udtType.cellPathForField(udtType.fieldName(0));
+ cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, value, cellPath);
+ }
+ else
+ {
+ ByteBuffer udtValue = UserType.buildValue(value, TWO);
+ cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, udtValue, null);
+ }
+ }
+ else
+ {
+ cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, value, null);
+ }
return BTreeRow.singleCellRow(Clustering.EMPTY, cell);
}
private static Row newRow(ColumnMetadata definition, List<ByteBuffer> values)
{
- Row.Builder builder = BTreeRow.sortedBuilder();
- builder.newRow(Clustering.EMPTY);
- long now = System.currentTimeMillis();
- if (values != null)
+ AbstractType<?> type = definition.type;
+ if (type.isFrozenCollection())
+ {
+ ByteBuffer cellValue = ListSerializer.pack(values, values.size(), CURRENT);
+ Cell<ByteBuffer> cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, cellValue, null);
+ return BTreeRow.singleCellRow(Clustering.EMPTY, cell);
+ }
+ else
{
- for (int i = 0, m = values.size(); i < m; i++)
+ Row.Builder builder = BTreeRow.sortedBuilder();
+ builder.newRow(Clustering.EMPTY);
+ long now = System.currentTimeMillis();
+ if (values != null)
{
- TimeUUID uuid = TimeUUID.Generator.atUnixMillis(now, i);
- ByteBuffer key = uuid.toBytes();
- ByteBuffer value = values.get(i);
- BufferCell cell = new BufferCell(definition,
- 0L,
- Cell.NO_TTL,
- Cell.NO_DELETION_TIME,
- value,
- CellPath.create(key));
- builder.addCell(cell);
+ for (int i = 0, m = values.size(); i < m; i++)
+ {
+ BufferCell cell;
+ if (type.isUDT())
+ {
+ cell = new BufferCell(definition,
+ 0L,
+ Cell.NO_TTL,
+ Cell.NO_DELETION_TIME,
+ values.get(i),
+ ((UserType)type).cellPathForField(((UserType) type).fieldName(i)));
+ }
+ else
+ {
+ TimeUUID uuid = TimeUUID.Generator.atUnixMillis(now, i);
+ ByteBuffer key = uuid.toBytes();
+ ByteBuffer value = values.get(i);
+ cell = new BufferCell(definition,
+ 0L,
+ Cell.NO_TTL,
+ Cell.NO_DELETION_TIME,
+ value,
+ CellPath.create(key));
+ }
+ builder.addCell(cell);
+ }
}
+ return builder.build();
}
- return builder.build();
}
private static Row newRow(ColumnMetadata definition, SortedSet<ByteBuffer> values)
{
- Row.Builder builder = BTreeRow.sortedBuilder();
- builder.newRow(Clustering.EMPTY);
- if (values != null)
+ if (definition.type.isFrozenCollection())
{
- for (ByteBuffer value : values)
+ ByteBuffer cellValue = SetSerializer.pack(values, values.size(), CURRENT);
+ Cell<ByteBuffer> cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, cellValue, null);
+ return BTreeRow.singleCellRow(Clustering.EMPTY, cell);
+ }
+ else
+ {
+ Row.Builder builder = BTreeRow.sortedBuilder();
+ builder.newRow(Clustering.EMPTY);
+ if (values != null)
{
- BufferCell cell = new BufferCell(definition,
- 0L,
- Cell.NO_TTL,
- Cell.NO_DELETION_TIME,
- ByteBufferUtil.EMPTY_BYTE_BUFFER,
- CellPath.create(value));
- builder.addCell(cell);
+ for (ByteBuffer value : values)
+ {
+ BufferCell cell = new BufferCell(definition,
+ 0L,
+ Cell.NO_TTL,
+ Cell.NO_DELETION_TIME,
+ ByteBufferUtil.EMPTY_BYTE_BUFFER,
+ CellPath.create(value));
+ builder.addCell(cell);
+ }
}
+ return builder.build();
}
- return builder.build();
}
- private static Row newRow(ColumnMetadata definition, Map<ByteBuffer, ByteBuffer> values)
+ private static Row newRow(ColumnMetadata definition, SortedMap<ByteBuffer, ByteBuffer> values)
{
- Row.Builder builder = BTreeRow.sortedBuilder();
- builder.newRow(Clustering.EMPTY);
- if (values != null)
+ if (definition.type.isFrozenCollection())
+ {
+ List<ByteBuffer> packableValues = values.entrySet().stream().flatMap(entry -> Stream.of(entry.getKey(), entry.getValue())).collect(Collectors.toList());
+ ByteBuffer cellValue = MapSerializer.pack(packableValues, values.size(), CURRENT);
+ Cell<ByteBuffer> cell = new BufferCell(definition, 0L, Cell.NO_TTL, Cell.NO_DELETION_TIME, cellValue, null);
+ return BTreeRow.singleCellRow(Clustering.EMPTY, cell);
+ }
+ else
{
- for (Map.Entry<ByteBuffer, ByteBuffer> entry : values.entrySet())
+ Row.Builder builder = BTreeRow.sortedBuilder();
+ builder.newRow(Clustering.EMPTY);
+ if (values != null)
{
- BufferCell cell = new BufferCell(definition,
- 0L,
- Cell.NO_TTL,
- Cell.NO_DELETION_TIME,
- entry.getValue(),
- CellPath.create(entry.getKey()));
- builder.addCell(cell);
+ for (Map.Entry<ByteBuffer, ByteBuffer> entry : values.entrySet())
+ {
+ BufferCell cell = new BufferCell(definition,
+ 0L,
+ Cell.NO_TTL,
+ Cell.NO_DELETION_TIME,
+ entry.getValue(),
+ CellPath.create(entry.getKey()));
+ builder.addCell(cell);
+ }
}
+ return builder.build();
+ }
+ }
+
+ private static boolean testRoundtripped(ColumnCondition.Bound bound, Row row)
+ {
+ DataOutputBuffer dab = new DataOutputBuffer();
+ IVersionedSerializer<ColumnCondition.Bound> serializer = ColumnCondition.Bound.serializer;
+ int version = MessagingService.current_version;
+ try
+ {
+ serializer.serialize(bound, dab, version);
+ assertEquals(serializer.serializedSize(bound, version), dab.position());
+ ColumnCondition.Bound deserializedBound = serializer.deserialize(new DataInputBuffer(dab.buffer(), false), version);
+ boolean originalResult = bound.appliesTo(row);
+ assertEquals(originalResult, deserializedBound.appliesTo(row));
+ return originalResult;
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException(e);
}
- return builder.build();
}
- private static boolean conditionApplies(ByteBuffer rowValue, Operator op, ByteBuffer conditionValue)
+ private boolean conditionApplies(ByteBuffer rowValue, Operator op, ByteBuffer conditionValue)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", Int32Type.instance);
+ AbstractType<?> columnType = Int32Type.instance;
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, false), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Constants.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ boolean regularColumnResult = testRoundtripped(bound, newRow(definition, rowValue));
+ // Every simple bound test is also a valid test of the UDT access path
+ boolean conditionUDTApplies = conditionUDTApplies(rowValue, op, conditionValue);
+ assertEquals(regularColumnResult, conditionUDTApplies);
+ return regularColumnResult;
}
- private static boolean conditionApplies(List<ByteBuffer> rowValue, Operator op, List<ByteBuffer> conditionValue)
+ private boolean conditionApplies(List<ByteBuffer> rowValue, Operator op, List<ByteBuffer> conditionValue)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", ListType.getInstance(Int32Type.instance, true));
+ boolean nonFrozenResult = conditionApplies(rowValue, op, conditionValue, listType);
+ boolean frozenResult = conditionApplies(rowValue, op, conditionValue, listType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
+ }
+
+ private boolean conditionApplies(List<ByteBuffer> rowValue, Operator op, List<ByteBuffer> conditionValue, ListType<Integer> columnType)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Lists.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ private boolean conditionContainsApplies(List<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
+ {
+ boolean nonFrozenResult = conditionContainsApplies(rowValue, op, conditionValue, listType);
+ boolean frozenResult = conditionContainsApplies(rowValue, op, conditionValue, listType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return frozenResult;
}
- private static boolean conditionContainsApplies(List<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
+ private boolean conditionContainsApplies(List<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue, ListType<Integer> columnType)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", ListType.getInstance(Int32Type.instance, true));
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Constants.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
}
- private static boolean conditionContainsApplies(Map<ByteBuffer, ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
+ private boolean conditionContainsApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", MapType.getInstance(Int32Type.instance, Int32Type.instance, true));
+ boolean nonFrozenResult = conditionContainsApplies(rowValue, op, conditionValue, mapType);
+ boolean frozenResult = conditionContainsApplies(rowValue, op, conditionValue, mapType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return frozenResult;
+ }
+
+ private boolean conditionContainsApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue, MapType<Integer, Integer> columnType)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Constants.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ private boolean conditionApplies(SortedSet<ByteBuffer> rowValue, Operator op, SortedSet<ByteBuffer> conditionValue)
+ {
+ boolean nonFrozenResult = conditionApplies(rowValue, op, conditionValue, setType);
+ boolean frozenResult = conditionApplies(rowValue, op, conditionValue, setType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
}
- private static boolean conditionApplies(SortedSet<ByteBuffer> rowValue, Operator op, SortedSet<ByteBuffer> conditionValue)
+ private boolean conditionApplies(SortedSet<ByteBuffer> rowValue, Operator op, SortedSet<ByteBuffer> conditionValue, SetType<Integer> columnType)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", SetType.getInstance(Int32Type.instance, true));
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Sets.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
}
- private static boolean conditionContainsApplies(SortedSet<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
+ private boolean conditionContainsApplies(SortedSet<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", SetType.getInstance(Int32Type.instance, true));
+ boolean nonFrozenResult = conditionContainsApplies(rowValue, op, conditionValue, setType);
+ boolean frozenResult = conditionContainsApplies(rowValue, op, conditionValue, setType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return frozenResult;
+ }
+
+ private boolean conditionContainsApplies(SortedSet<ByteBuffer> rowValue, Operator op, ByteBuffer conditionValue, SetType<Integer> columnType)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Constants.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ private boolean conditionApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, Operator op, SortedMap<ByteBuffer, ByteBuffer> conditionValue)
+ {
+ boolean nonFrozenResult = conditionApplies(rowValue, op, conditionValue, mapType);
+ boolean frozenResult = conditionApplies(rowValue, op, conditionValue, mapType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
}
- private static boolean conditionApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, Operator op, SortedMap<ByteBuffer, ByteBuffer> conditionValue)
+ private boolean conditionApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, Operator op, SortedMap<ByteBuffer, ByteBuffer> conditionValue, MapType<Integer, Integer> columnType)
{
- ColumnMetadata definition = ColumnMetadata.regularColumn("ks", "cf", "c", MapType.getInstance(Int32Type.instance, Int32Type.instance, true));
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(new Maps.Value(conditionValue)));
ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
- return bound.appliesTo(newRow(definition, rowValue));
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ private boolean conditionElementApplies(List<ByteBuffer> values, ByteBuffer elementIndex, Operator op, ByteBuffer elementValue)
+ {
+ boolean nonFrozenResult = conditionElementApplies(values, elementIndex, op, elementValue, listType);
+ boolean frozenResult = conditionElementApplies(values, elementIndex, op, elementValue, listType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
+ }
+
+ private boolean conditionElementApplies(List<ByteBuffer> values, ByteBuffer elementIndex, Operator op, ByteBuffer elementValue, ListType<Integer> columnType)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
+ ColumnCondition condition = ColumnCondition.condition(definition, new Constants.Value(elementIndex), op, Terms.of(new Constants.Value(elementValue)));
+ ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
+ return testRoundtripped(bound, newRow(definition, values));
+ }
+
+ private boolean conditionElementApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, ByteBuffer elementKey, Operator op, ByteBuffer elementValue)
+ {
+ boolean nonFrozenResult = conditionElementApplies(rowValue, elementKey, op, elementValue, mapType);
+ boolean frozenResult = conditionElementApplies(rowValue, elementKey, op, elementValue, mapType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
+ }
+
+ private boolean conditionElementApplies(SortedMap<ByteBuffer, ByteBuffer> rowValue, ByteBuffer elementKey, Operator op, ByteBuffer elementValue, MapType<Integer, Integer> columnType)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(columnType, columnType.isFrozenCollection()), "c", columnType);
+ ColumnCondition condition = ColumnCondition.condition(definition, new Constants.Value(elementKey), op, Terms.of(new Constants.Value(elementValue)));
+ ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ private boolean conditionUDTApplies(ByteBuffer fieldValue, Operator op, ByteBuffer elementValue)
+ {
+ boolean nonFrozenResult = conditionUDTApplies(fieldValue, op, elementValue, udtType);
+ boolean frozenResult = conditionUDTApplies(fieldValue, op, elementValue, udtType.freeze());
+ assertEquals(nonFrozenResult, frozenResult);
+ return nonFrozenResult;
+ }
+
+ private boolean conditionUDTApplies(ByteBuffer fieldValue, Operator op, ByteBuffer elementValue, UserType type)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(type, !type.isMultiCell()), "c", type);
+ ColumnCondition condition = ColumnCondition.condition(definition, type.fieldName(0), op, Terms.of(new Constants.Value(elementValue)));
+ ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
+ return testRoundtripped(bound, newRow(definition, fieldValue));
+ }
+
+ private boolean conditionUDTApplies(List<ByteBuffer> rowValue, Operator op, List<ByteBuffer> conditionValue)
+ {
+ ColumnMetadata definition = ColumnMetadata.regularColumn(KEYSPACE, maybeCreateTable(udtType, false), "c", udtType);
+ Term term = conditionValue == null ? Constants.NULL_VALUE : new UserTypes.Value(udtType, conditionValue.toArray(new ByteBuffer[0]));
+ ColumnCondition condition = ColumnCondition.condition(definition, op, Terms.of(term));
+ ColumnCondition.Bound bound = condition.bind(QueryOptions.DEFAULT);
+ return testRoundtripped(bound, newRow(definition, rowValue));
+ }
+
+ @Override
+ public void beforeTest() throws Throwable
+ {
+ super.beforeTest();
+ String typeName = createType("CREATE TYPE %s (a int, b int);");
+ udtType = Schema.instance.getKeyspaceMetadata(KEYSPACE).types.get(ByteBufferUtil.bytes(typeName)).get();
+ }
+ @Override
+ public void afterTest() throws Throwable
+ {
+ super.afterTest();
+ typeToTable.clear();
+ udtType = null;
+ }
+
+ // Is this useful enough to have in CQL tester?
+ private UserType udtType;
+ private final Map<Pair<AbstractType<?>, Boolean>, String> typeToTable = new HashMap<>();
+
+ private String maybeCreateTable(AbstractType<?> columnType, boolean frozen)
+ {
+ String columnTypeCQL = columnType.asCQL3Type().toString();
+ String maybeFrozenColumnTypeCQL = frozen ? columnTypeCQL : "frozen<%s>".format(columnTypeCQL);
+ return typeToTable.computeIfAbsent(Pair.create(columnType, frozen), type -> createTable("CREATE TABLE %s (id int primary key, c " + maybeFrozenColumnTypeCQL + ")"));
}
@FunctionalInterface
@@ -194,7 +454,7 @@ public class ColumnConditionTest
}
@Test
- public void testSimpleBoundIsSatisfiedByValue() throws InvalidRequestException
+ public void testSimpleAndUDTBoundIsSatisfiedByValue() throws InvalidRequestException
{
// EQ
assertTrue(conditionApplies(ONE, EQ, ONE));
@@ -205,7 +465,7 @@ public class ColumnConditionTest
assertTrue(conditionApplies(EMPTY_BYTE_BUFFER, EQ, EMPTY_BYTE_BUFFER));
assertFalse(conditionApplies(ONE, EQ, null));
assertFalse(conditionApplies(null, EQ, ONE));
- assertTrue(conditionApplies((ByteBuffer) null, EQ, (ByteBuffer) null));
+ assertTrue(conditionApplies(null, EQ, (ByteBuffer) null));
// NEQ
assertFalse(conditionApplies(ONE, NEQ, ONE));
@@ -216,7 +476,7 @@ public class ColumnConditionTest
assertFalse(conditionApplies(EMPTY_BYTE_BUFFER, NEQ, EMPTY_BYTE_BUFFER));
assertTrue(conditionApplies(ONE, NEQ, null));
assertTrue(conditionApplies(null, NEQ, ONE));
- assertFalse(conditionApplies((ByteBuffer) null, NEQ, (ByteBuffer) null));
+ assertFalse(conditionApplies(null, NEQ, (ByteBuffer) null));
// LT
assertFalse(conditionApplies(ONE, LT, ONE));
@@ -359,6 +619,18 @@ public class ColumnConditionTest
assertFalse(conditionContainsApplies(list(ZERO, ONE, TWO), CONTAINS, ByteBufferUtil.EMPTY_BYTE_BUFFER));
assertFalse(conditionContainsApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), CONTAINS, ONE));
assertTrue(conditionContainsApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), CONTAINS, ByteBufferUtil.EMPTY_BYTE_BUFFER));
+
+ //ELEMENT
+ assertTrue(conditionElementApplies(list(ZERO, ONE, TWO), ZERO, EQ, ZERO));
+ assertTrue(conditionElementApplies(list(ZERO, ONE, TWO), ONE, EQ, ONE));
+ assertTrue(conditionElementApplies(list(ZERO, ONE, TWO), TWO, EQ, TWO));
+
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), ZERO, EQ, ONE));
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), ZERO, EQ, TWO));
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), ONE, EQ, ZERO));
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), ONE, EQ, TWO));
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), TWO, EQ, ZERO));
+ assertFalse(conditionElementApplies(list(ZERO, ONE, TWO), TWO, EQ, ONE));
}
private static SortedSet<ByteBuffer> set(ByteBuffer... values)
@@ -611,5 +883,96 @@ public class ColumnConditionTest
assertTrue(conditionContainsApplies(map(ONE, ByteBufferUtil.EMPTY_BYTE_BUFFER), CONTAINS_KEY, ONE));
assertFalse(conditionContainsApplies(map(ONE, ByteBufferUtil.EMPTY_BYTE_BUFFER), CONTAINS_KEY, ByteBufferUtil.EMPTY_BYTE_BUFFER));
+ // Element access
+ assertTrue(conditionElementApplies(map(ZERO, ONE), ZERO, EQ, ONE));
+ assertFalse(conditionElementApplies(map(ZERO, ONE), ZERO, EQ, TWO));
+ assertFalse(conditionElementApplies(map(ZERO, ONE), ONE, EQ, TWO));
+ assertFalse(conditionElementApplies(map(), ZERO, EQ, TWO));
+ }
+
+ @Test
+ public void testMultiCellUDTBound() throws InvalidRequestException
+ {
+ // EQ
+ assertTrue(conditionUDTApplies(list(ONE), EQ, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), EQ, list(ZERO)));
+ assertFalse(conditionUDTApplies(list(ZERO), EQ, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE, ONE), EQ, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), EQ, list(ONE, ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), EQ, list()));
+
+ assertFalse(conditionUDTApplies(list(ONE), EQ, null));
+ assertFalse(conditionUDTApplies(null, EQ, list(ONE)));
+ assertTrue(conditionUDTApplies(null, EQ, (List<ByteBuffer>) null));
+
+ assertFalse(conditionUDTApplies(list(ONE), EQ, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), EQ, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), EQ, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+
+ // NEQ
+ assertFalse(conditionUDTApplies(list(ONE), NEQ, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), NEQ, list(ZERO)));
+ assertTrue(conditionUDTApplies(list(ZERO), NEQ, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE, ONE), NEQ, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), NEQ, list(ONE, ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), NEQ, list()));
+ assertTrue(conditionUDTApplies(list(), NEQ, list(ONE)));
+
+ assertTrue(conditionUDTApplies(list(ONE), NEQ, null));
+ assertTrue(conditionUDTApplies(null, NEQ, list(ONE)));
+ assertFalse(conditionUDTApplies(null, NEQ, (List<ByteBuffer>) null));
+
+ assertTrue(conditionUDTApplies(list(ONE), NEQ, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), NEQ, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), NEQ, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+
+ // LT
+ assertFalse(conditionUDTApplies(list(ONE), LT, list(ONE)));
+ assertFalse(conditionUDTApplies(list(), LT, list()));
+ assertFalse(conditionUDTApplies(list(ONE), LT, list(ZERO)));
+ assertTrue(conditionUDTApplies(list(ZERO), LT, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE, ONE), LT, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), LT, list(ONE, ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), LT, list()));
+
+ assertFalse(conditionUDTApplies(list(ONE), LT, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), LT, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), LT, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+
+ // LTE
+ assertTrue(conditionUDTApplies(list(ONE), LTE, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), LTE, list(ZERO)));
+ assertTrue(conditionUDTApplies(list(ZERO), LTE, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE, ONE), LTE, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), LTE, list(ONE, ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), LTE, list()));
+
+ assertFalse(conditionUDTApplies(list(ONE), LTE, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), LTE, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), LTE, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+
+ // GT
+ assertFalse(conditionUDTApplies(list(ONE), GT, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), GT, list(ZERO)));
+ assertFalse(conditionUDTApplies(list(ZERO), GT, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE, ONE), GT, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), GT, list(ONE, ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), GT, list()));
+
+ assertTrue(conditionUDTApplies(list(ONE), GT, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), GT, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), GT, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+
+ // GTE
+ assertTrue(conditionUDTApplies(list(ONE), GTE, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), GTE, list(ZERO)));
+ assertFalse(conditionUDTApplies(list(ZERO), GTE, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ONE, ONE), GTE, list(ONE)));
+ assertFalse(conditionUDTApplies(list(ONE), GTE, list(ONE, ONE)));
+ assertTrue(conditionUDTApplies(list(ONE), GTE, list()));
+
+ assertTrue(conditionUDTApplies(list(ONE), GTE, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
+ assertFalse(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), GTE, list(ONE)));
+ assertTrue(conditionUDTApplies(list(ByteBufferUtil.EMPTY_BYTE_BUFFER), GTE, list(ByteBufferUtil.EMPTY_BYTE_BUFFER)));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org