diff --git a/CHANGES.txt b/CHANGES.txt index 684d003d80..5ecd19d9b2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -76,6 +76,7 @@ Merged from 1.2: * Optimize Cell liveness checks and clean up Cell (CASSANDRA-7119) * Support consistent range movements (CASSANDRA-2434) Merged from 2.0: + * Starting threads in OutboundTcpConnectionPool constructor causes race conditions (CASSANDRA-7177) * Allow overriding cassandra-rackdc.properties file (CASSANDRA-7072) * Set JMX RMI port to 7199 (CASSANDRA-7087) * Use LOCAL_QUORUM for data reads at LOCAL_SERIAL (CASSANDRA-6939) diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index a7d06dfee0..e1266cb101 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -117,6 +117,8 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache { if (timestampSet && statement.isTimestampSet()) throw new InvalidRequestException("Timestamp must be set either on BATCH or individual statements"); + + statement.validate(state); } } diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 7f8b6781e2..23f7cfe243 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -23,18 +23,23 @@ import java.util.*; import com.google.common.base.Function; import com.google.common.collect.Iterables; import org.github.jamm.MemoryMeter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.auth.Permission; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.ColumnDefinition; import org.apache.cassandra.cql3.*; import org.apache.cassandra.db.*; -import org.apache.cassandra.db.composites.CBuilder; -import org.apache.cassandra.db.composites.Composite; import org.apache.cassandra.db.filter.ColumnSlice; +import org.apache.cassandra.db.filter.IDiskAtomFilter; import org.apache.cassandra.db.filter.SliceQueryFilter; +import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.db.marshal.UTF8Type; +import org.apache.cassandra.db.marshal.ListType; import org.apache.cassandra.db.marshal.BooleanType; import org.apache.cassandra.exceptions.*; +import org.apache.cassandra.service.CASConditions; import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.QueryState; import org.apache.cassandra.service.StorageProxy; @@ -49,16 +54,21 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF { private static final ColumnIdentifier CAS_RESULT_COLUMN = new ColumnIdentifier("[applied]", false); + private static final Logger logger = LoggerFactory.getLogger(ModificationStatement.class); + + private static boolean loggedCounterTTL = false; + private static boolean loggedCounterTimestamp = false; + public static enum StatementType { INSERT, UPDATE, DELETE } public final StatementType type; - private final int boundTerms; public final CFMetaData cfm; public final Attributes attrs; private final Map processedKeys = new HashMap(); private final List columnOperations = new ArrayList(); + private int boundTerms; // Separating normal and static conditions makes things somewhat easier private List columnConditions; private List staticConditions; @@ -70,18 +80,17 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF private boolean setsStaticColumns; private boolean setsRegularColumns; - private final Function getColumnForCondition = new Function() + private final Function getColumnForCondition = new Function() { - public ColumnDefinition apply(ColumnCondition cond) + public ColumnIdentifier apply(ColumnCondition cond) { - return cond.column; + return cond.column.name; } }; - public ModificationStatement(StatementType type, int boundTerms, CFMetaData cfm, Attributes attrs) + public ModificationStatement(StatementType type, CFMetaData cfm, Attributes attrs) { this.type = type; - this.boundTerms = boundTerms; this.cfm = cfm; this.attrs = attrs; } @@ -97,7 +106,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF } public abstract boolean requireFullClusteringKey(); - public abstract void addUpdateForKey(ColumnFamily updates, ByteBuffer key, Composite prefix, UpdateParameters params) throws InvalidRequestException; + public abstract void addUpdateForKey(ColumnFamily updates, ByteBuffer key, ColumnNameBuilder builder, UpdateParameters params) throws InvalidRequestException; public int getBoundTerms() { @@ -116,12 +125,12 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF public boolean isCounter() { - return cfm.isCounter(); + return cfm.getDefaultValidator().isCommutative(); } - public long getTimestamp(long now, QueryOptions options) throws InvalidRequestException + public long getTimestamp(long now, List variables) throws InvalidRequestException { - return attrs.getTimestamp(now, options); + return attrs.getTimestamp(now, variables); } public boolean isTimestampSet() @@ -129,9 +138,9 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return attrs.isTimestampSet(); } - public int getTimeToLive(QueryOptions options) throws InvalidRequestException + public int getTimeToLive(List variables) throws InvalidRequestException { - return attrs.getTimeToLive(options); + return attrs.getTimeToLive(variables); } public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException @@ -146,18 +155,31 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF public void validate(ClientState state) throws InvalidRequestException { if (hasConditions() && attrs.isTimestampSet()) - throw new InvalidRequestException("Cannot provide custom timestamp for conditional updates"); + throw new InvalidRequestException("Cannot provide custom timestamp for conditional update"); - if (isCounter() && attrs.isTimestampSet()) - throw new InvalidRequestException("Cannot provide custom timestamp for counter updates"); + if (isCounter()) + { + if (attrs.isTimestampSet() && !loggedCounterTimestamp) + { + logger.warn("Detected use of 'USING TIMESTAMP' in a counter UPDATE. This is invalid " + + "because counters do not use timestamps, and the timestamp has been ignored. " + + "Such queries will be rejected in Cassandra 2.1+ - please fix your queries before then."); + loggedCounterTimestamp = true; + } - if (isCounter() && attrs.isTimeToLiveSet()) - throw new InvalidRequestException("Cannot provide custom TTL for counter updates"); + if (attrs.isTimeToLiveSet() && !loggedCounterTTL) + { + logger.warn("Detected use of 'USING TTL' in a counter UPDATE. This is invalid " + + "because counter tables do not support TTL, and the TTL value has been ignored. " + + "Such queries will be rejected in Cassandra 2.1+ - please fix your queries before then."); + loggedCounterTTL = true; + } + } } public void addOperation(Operation op) { - if (op.column.isStatic()) + if (op.isStatic(cfm)) setsStaticColumns = true; else setsRegularColumns = true; @@ -169,19 +191,19 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return columnOperations; } - public Iterable getColumnsWithConditions() + public Iterable getColumnsWithConditions() { if (ifNotExists || ifExists) return null; - return Iterables.concat(columnConditions == null ? Collections.emptyList() : Iterables.transform(columnConditions, getColumnForCondition), - staticConditions == null ? Collections.emptyList() : Iterables.transform(staticConditions, getColumnForCondition)); + return Iterables.concat(columnConditions == null ? Collections.emptyList() : Iterables.transform(columnConditions, getColumnForCondition), + staticConditions == null ? Collections.emptyList() : Iterables.transform(staticConditions, getColumnForCondition)); } public void addCondition(ColumnCondition cond) throws InvalidRequestException { List conds = null; - if (cond.column.isStatic()) + if (cond.column.kind == CFDefinition.Name.Kind.STATIC) { setsStaticColumns = true; if (staticConditions == null) @@ -218,44 +240,45 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return ifExists; } - private void addKeyValues(ColumnDefinition def, Restriction values) throws InvalidRequestException + private void addKeyValues(CFDefinition.Name name, Restriction values) throws InvalidRequestException { - if (def.kind == ColumnDefinition.Kind.CLUSTERING_COLUMN) + if (name.kind == CFDefinition.Name.Kind.COLUMN_ALIAS) hasNoClusteringColumns = false; - if (processedKeys.put(def.name, values) != null) - throw new InvalidRequestException(String.format("Multiple definitions found for PRIMARY KEY part %s", def.name)); + if (processedKeys.put(name.name, values) != null) + throw new InvalidRequestException(String.format("Multiple definitions found for PRIMARY KEY part %s", name.name)); } - public void addKeyValue(ColumnDefinition def, Term value) throws InvalidRequestException + public void addKeyValue(CFDefinition.Name name, Term value) throws InvalidRequestException { - addKeyValues(def, new Restriction.EQ(value, false)); + addKeyValues(name, new Restriction.EQ(value, false)); } public void processWhereClause(List whereClause, VariableSpecifications names) throws InvalidRequestException { + CFDefinition cfDef = cfm.getCfDef(); for (Relation rel : whereClause) { - ColumnDefinition def = cfm.getColumnDefinition(rel.getEntity()); - if (def == null) + CFDefinition.Name name = cfDef.get(rel.getEntity()); + if (name == null) throw new InvalidRequestException(String.format("Unknown key identifier %s", rel.getEntity())); - switch (def.kind) + switch (name.kind) { - case PARTITION_KEY: - case CLUSTERING_COLUMN: + case KEY_ALIAS: + case COLUMN_ALIAS: Restriction restriction; if (rel.operator() == Relation.Type.EQ) { - Term t = rel.getValue().prepare(keyspace(), def); + Term t = rel.getValue().prepare(name); t.collectMarkerSpecification(names); restriction = new Restriction.EQ(t, false); } - else if (def.kind == ColumnDefinition.Kind.PARTITION_KEY && rel.operator() == Relation.Type.IN) + else if (name.kind == CFDefinition.Name.Kind.KEY_ALIAS && rel.operator() == Relation.Type.IN) { if (rel.getValue() != null) { - Term t = rel.getValue().prepare(keyspace(), def); + Term t = rel.getValue().prepare(name); t.collectMarkerSpecification(names); restriction = Restriction.IN.create(t); } @@ -264,7 +287,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF List values = new ArrayList(rel.getInValues().size()); for (Term.Raw raw : rel.getInValues()) { - Term t = raw.prepare(keyspace(), def); + Term t = raw.prepare(name); t.collectMarkerSpecification(names); values.add(t); } @@ -273,37 +296,40 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF } else { - throw new InvalidRequestException(String.format("Invalid operator %s for PRIMARY KEY part %s", rel.operator(), def.name)); + throw new InvalidRequestException(String.format("Invalid operator %s for PRIMARY KEY part %s", rel.operator(), name)); } - addKeyValues(def, restriction); + addKeyValues(name, restriction); break; - default: - throw new InvalidRequestException(String.format("Non PRIMARY KEY %s found in where clause", def.name)); + case VALUE_ALIAS: + case COLUMN_METADATA: + case STATIC: + throw new InvalidRequestException(String.format("Non PRIMARY KEY %s found in where clause", name)); } } } - public List buildPartitionKeyNames(QueryOptions options) + public List buildPartitionKeyNames(List variables) throws InvalidRequestException { - CBuilder keyBuilder = cfm.getKeyValidatorAsCType().builder(); + CFDefinition cfDef = cfm.getCfDef(); + ColumnNameBuilder keyBuilder = cfDef.getKeyNameBuilder(); List keys = new ArrayList(); - for (ColumnDefinition def : cfm.partitionKeyColumns()) + for (CFDefinition.Name name : cfDef.partitionKeys()) { - Restriction r = processedKeys.get(def.name); + Restriction r = processedKeys.get(name.name); if (r == null) - throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", def.name)); + throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", name)); - List values = r.values(options); + List values = r.values(variables); if (keyBuilder.remainingCount() == 1) { for (ByteBuffer val : values) { if (val == null) - throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", def.name)); - ByteBuffer key = keyBuilder.buildWith(val).toByteBuffer(); + throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", name)); + ByteBuffer key = keyBuilder.copy().add(val).build(); ThriftValidation.validateKey(cfm, key); keys.add(key); } @@ -314,14 +340,14 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF throw new InvalidRequestException("IN is only supported on the last column of the partition key"); ByteBuffer val = values.get(0); if (val == null) - throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", def.name)); + throw new InvalidRequestException(String.format("Invalid null value for partition key part %s", name)); keyBuilder.add(val); } } return keys; } - public Composite createClusteringPrefix(QueryOptions options) + public ColumnNameBuilder createClusteringPrefixBuilder(List variables) throws InvalidRequestException { // If the only updated/deleted columns are static, then we don't need clustering columns. @@ -338,83 +364,96 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF { // If we set no non-static columns, then it's fine not to have clustering columns if (hasNoClusteringColumns) - return cfm.comparator.staticPrefix(); + return cfm.getStaticColumnNameBuilder(); // If we do have clustering columns however, then either it's an INSERT and the query is valid // but we still need to build a proper prefix, or it's not an INSERT, and then we want to reject // (see above) if (type != StatementType.INSERT) { - for (ColumnDefinition def : cfm.clusteringColumns()) - if (processedKeys.get(def.name) != null) - throw new InvalidRequestException(String.format("Invalid restriction on clustering column %s since the %s statement modifies only static columns", def.name, type)); + for (CFDefinition.Name name : cfm.getCfDef().clusteringColumns()) + if (processedKeys.get(name.name) != null) + throw new InvalidRequestException(String.format("Invalid restriction on clustering column %s since the %s statement modifies only static columns", name.name, type)); // we should get there as it contradicts hasNoClusteringColumns == false throw new AssertionError(); } } - return createClusteringPrefixBuilderInternal(options); + return createClusteringPrefixBuilderInternal(variables); } - private Composite createClusteringPrefixBuilderInternal(QueryOptions options) + private ColumnNameBuilder updatePrefixFor(ByteBuffer name, ColumnNameBuilder prefix) + { + return isStatic(name) ? cfm.getStaticColumnNameBuilder() : prefix; + } + + public boolean isStatic(ByteBuffer name) + { + ColumnDefinition def = cfm.getColumnDefinition(name); + return def != null && def.type == ColumnDefinition.Type.STATIC; + } + + private ColumnNameBuilder createClusteringPrefixBuilderInternal(List variables) throws InvalidRequestException { - CBuilder builder = cfm.comparator.prefixBuilder(); - ColumnDefinition firstEmptyKey = null; - for (ColumnDefinition def : cfm.clusteringColumns()) + CFDefinition cfDef = cfm.getCfDef(); + ColumnNameBuilder builder = cfDef.getColumnNameBuilder(); + CFDefinition.Name firstEmptyKey = null; + for (CFDefinition.Name name : cfDef.clusteringColumns()) { - Restriction r = processedKeys.get(def.name); + Restriction r = processedKeys.get(name.name); if (r == null) { - firstEmptyKey = def; - if (requireFullClusteringKey() && !cfm.comparator.isDense() && cfm.comparator.isCompound()) - throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", def.name)); + firstEmptyKey = name; + if (requireFullClusteringKey() && cfDef.isComposite && !cfDef.isCompact) + throw new InvalidRequestException(String.format("Missing mandatory PRIMARY KEY part %s", name)); } else if (firstEmptyKey != null) { - throw new InvalidRequestException(String.format("Missing PRIMARY KEY part %s since %s is set", firstEmptyKey.name, def.name)); + throw new InvalidRequestException(String.format("Missing PRIMARY KEY part %s since %s is set", firstEmptyKey.name, name.name)); } else { - List values = r.values(options); + List values = r.values(variables); assert values.size() == 1; // We only allow IN for row keys so far ByteBuffer val = values.get(0); if (val == null) - throw new InvalidRequestException(String.format("Invalid null value for clustering key part %s", def.name)); + throw new InvalidRequestException(String.format("Invalid null value for clustering key part %s", name)); builder.add(val); } } - return builder.build(); + return builder; } - protected ColumnDefinition getFirstEmptyKey() + protected CFDefinition.Name getFirstEmptyKey() { - for (ColumnDefinition def : cfm.clusteringColumns()) + for (CFDefinition.Name name : cfm.getCfDef().clusteringColumns()) { - if (processedKeys.get(def.name) == null) - return def; + if (processedKeys.get(name.name) == null) + return name; } return null; } - protected Map readRequiredRows(Collection partitionKeys, Composite clusteringPrefix, boolean local, ConsistencyLevel cl) + protected Map readRequiredRows(Collection partitionKeys, ColumnNameBuilder clusteringPrefix, boolean local, ConsistencyLevel cl) throws RequestExecutionException, RequestValidationException { // Lists SET operation incurs a read. - boolean requiresRead = false; + Set toRead = null; for (Operation op : columnOperations) { if (op.requiresRead()) { - requiresRead = true; - break; + if (toRead == null) + toRead = new TreeSet(UTF8Type.instance); + toRead.add(op.columnName.key); } } - return requiresRead ? readRows(partitionKeys, clusteringPrefix, cfm, local, cl) : null; + return toRead == null ? null : readRows(partitionKeys, clusteringPrefix, toRead, (CompositeType)cfm.comparator, local, cl); } - protected Map readRows(Collection partitionKeys, Composite rowPrefix, CFMetaData cfm, boolean local, ConsistencyLevel cl) + private Map readRows(Collection partitionKeys, ColumnNameBuilder clusteringPrefix, Set toRead, CompositeType composite, boolean local, ConsistencyLevel cl) throws RequestExecutionException, RequestValidationException { try @@ -426,7 +465,16 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF throw new InvalidRequestException(String.format("Write operation require a read but consistency %s is not supported on reads", cl)); } - ColumnSlice[] slices = new ColumnSlice[]{ rowPrefix.slice() }; + ColumnSlice[] slices = new ColumnSlice[toRead.size()]; + int i = 0; + for (ByteBuffer name : toRead) + { + ColumnNameBuilder prefix = updatePrefixFor(name, clusteringPrefix); + ByteBuffer start = prefix.copy().add(name).build(); + ByteBuffer finish = prefix.copy().add(name).buildAsEndOfRange(); + slices[i++] = new ColumnSlice(start, finish); + } + List commands = new ArrayList(partitionKeys.size()); long now = System.currentTimeMillis(); for (ByteBuffer key : partitionKeys) @@ -440,19 +488,20 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF ? SelectStatement.readLocally(keyspace(), commands) : StorageProxy.read(commands, cl); - Map map = new HashMap(); + Map map = new HashMap(); for (Row row : rows) { - if (row.cf == null || row.cf.isEmpty()) + if (row.cf == null || row.cf.getColumnCount() == 0) continue; - Iterator iter = cfm.comparator.CQL3RowBuilder(cfm, now).group(row.cf.getSortedColumns().iterator()); - if (iter.hasNext()) - { - map.put(row.key.getKey(), iter.next()); - // We can only update one CQ3Row per partition key at a time (we don't allow IN for clustering key) - assert !iter.hasNext(); - } + ColumnGroupMap.Builder groupBuilder = new ColumnGroupMap.Builder(composite, true, now); + for (Column column : row.cf) + groupBuilder.add(column); + + List groups = groupBuilder.groups(); + assert groups.isEmpty() || groups.size() == 1; + if (!groups.isEmpty()) + map.put(row.key.key, groups.get(0)); } return map; } @@ -488,7 +537,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF else cl.validateForWrite(cfm.ksName); - Collection mutations = getMutations(options, false, options.getTimestamp(queryState)); + Collection mutations = getMutations(options.getValues(), false, cl, queryState.getTimestamp()); if (!mutations.isEmpty()) StorageProxy.mutateWithTriggers(mutations, cl, false); @@ -498,18 +547,18 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF public ResultMessage executeWithCondition(QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException { - List keys = buildPartitionKeyNames(options); + List variables = options.getValues(); + List keys = buildPartitionKeyNames(variables); // We don't support IN for CAS operation so far if (keys.size() > 1) throw new InvalidRequestException("IN on the partition key is not supported with conditional updates"); ByteBuffer key = keys.get(0); - long now = options.getTimestamp(queryState); - CQL3CasConditions conditions = new CQL3CasConditions(cfm, now); - Composite prefix = createClusteringPrefix(options); - ColumnFamily updates = ArrayBackedSortedColumns.factory.create(cfm); - addUpdatesAndConditions(key, prefix, updates, conditions, options, getTimestamp(now, options)); + CQL3CasConditions conditions = new CQL3CasConditions(cfm, queryState.getTimestamp()); + ColumnNameBuilder prefix = createClusteringPrefixBuilder(variables); + ColumnFamily updates = UnsortedColumns.factory.create(cfm); + addUpdatesAndConditions(key, prefix, updates, conditions, variables, getTimestamp(queryState.getTimestamp(), variables)); ColumnFamily result = StorageProxy.cas(keyspace(), columnFamily(), @@ -521,16 +570,16 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return new ResultMessage.Rows(buildCasResultSet(key, result)); } - public void addUpdatesAndConditions(ByteBuffer key, Composite clusteringPrefix, ColumnFamily updates, CQL3CasConditions conditions, QueryOptions options, long now) + public void addUpdatesAndConditions(ByteBuffer key, ColumnNameBuilder clusteringPrefix, ColumnFamily updates, CQL3CasConditions conditions, List variables, long now) throws InvalidRequestException { - UpdateParameters updParams = new UpdateParameters(cfm, options, now, getTimeToLive(options), null); + UpdateParameters updParams = new UpdateParameters(cfm, variables, now, getTimeToLive(variables), null); addUpdateForKey(updates, key, clusteringPrefix, updParams); if (ifNotExists) { // If we use ifNotExists, if the statement applies to any non static columns, then the condition is on the row of the non-static - // columns and the prefix should be the clusteringPrefix. But if only static columns are set, then the ifNotExists apply to the existence + // columns and the prefix should be the rowPrefix. But if only static columns are set, then the ifNotExists apply to the existence // of any static columns and we should use the prefix for the "static part" of the partition. conditions.addNotExist(clusteringPrefix); } @@ -541,9 +590,9 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF else { if (columnConditions != null) - conditions.addConditions(clusteringPrefix, columnConditions, options); + conditions.addConditions(clusteringPrefix, columnConditions, variables); if (staticConditions != null) - conditions.addConditions(cfm.comparator.staticPrefix(), staticConditions, options); + conditions.addConditions(cfm.getStaticColumnNameBuilder(), staticConditions, variables); } } @@ -552,7 +601,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return buildCasResultSet(keyspace(), key, columnFamily(), cf, getColumnsWithConditions(), false); } - public static ResultSet buildCasResultSet(String ksName, ByteBuffer key, String cfName, ColumnFamily cf, Iterable columnsWithConditions, boolean isBatch) + public static ResultSet buildCasResultSet(String ksName, ByteBuffer key, String cfName, ColumnFamily cf, Iterable columnsWithConditions, boolean isBatch) throws InvalidRequestException { boolean success = cf == null; @@ -588,33 +637,34 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF return new ResultSet(new ResultSet.Metadata(specs), rows); } - private static ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf, Iterable columnsWithConditions, boolean isBatch) + private static ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf, Iterable columnsWithConditions, boolean isBatch) throws InvalidRequestException { - CFMetaData cfm = cf.metadata(); + CFDefinition cfDef = cf.metadata().getCfDef(); + Selection selection; if (columnsWithConditions == null) { - selection = Selection.wildcard(cfm); + selection = Selection.wildcard(cfDef); } else { - List defs = new ArrayList<>(); + List names = new ArrayList(); // Adding the partition key for batches to disambiguate if the conditions span multipe rows (we don't add them outside // of batches for compatibility sakes). if (isBatch) { - defs.addAll(cfm.partitionKeyColumns()); - defs.addAll(cfm.clusteringColumns()); + names.addAll(cfDef.partitionKeys()); + names.addAll(cfDef.clusteringColumns()); } - for (ColumnDefinition def : columnsWithConditions) - defs.add(def); - selection = Selection.forColumns(defs); + for (ColumnIdentifier id : columnsWithConditions) + names.add(cfDef.get(id)); + selection = Selection.forColumns(names); } long now = System.currentTimeMillis(); Selection.ResultSetBuilder builder = selection.resultSetBuilder(now); - SelectStatement.forSelection(cfm, selection).processColumnFamily(key, cf, QueryOptions.DEFAULT, now, builder); + SelectStatement.forSelection(cfDef, selection).processColumnFamily(key, cf, Collections.emptyList(), now, builder); return builder.build(); } @@ -624,19 +674,15 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF if (hasConditions()) throw new UnsupportedOperationException(); - for (IMutation mutation : getMutations(QueryOptions.DEFAULT, true, queryState.getTimestamp())) - { - // We don't use counters internally. - assert mutation instanceof Mutation; - ((Mutation) mutation).apply(); - } + for (IMutation mutation : getMutations(Collections.emptyList(), true, null, queryState.getTimestamp())) + mutation.apply(); return null; } /** * Convert statement into a list of mutations to apply on the server * - * @param options value for prepared statement markers + * @param variables value for prepared statement markers * @param local if true, any requests (for collections) performed by getMutation should be done locally only. * @param cl the consistency to use for the potential reads involved in generating the mutations (for lists set/delete operations) * @param now the current timestamp in microseconds to use if no timestamp is user provided. @@ -644,36 +690,37 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF * @return list of the mutations * @throws InvalidRequestException on invalid requests */ - private Collection getMutations(QueryOptions options, boolean local, long now) + private Collection getMutations(List variables, boolean local, ConsistencyLevel cl, long now) throws RequestExecutionException, RequestValidationException { - List keys = buildPartitionKeyNames(options); - Composite clusteringPrefix = createClusteringPrefix(options); + List keys = buildPartitionKeyNames(variables); + ColumnNameBuilder clusteringPrefix = createClusteringPrefixBuilder(variables); - UpdateParameters params = makeUpdateParameters(keys, clusteringPrefix, options, local, now); + UpdateParameters params = makeUpdateParameters(keys, clusteringPrefix, variables, local, cl, now); Collection mutations = new ArrayList(); for (ByteBuffer key: keys) { ThriftValidation.validateKey(cfm, key); - ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfm); + ColumnFamily cf = UnsortedColumns.factory.create(cfm); addUpdateForKey(cf, key, clusteringPrefix, params); - Mutation mut = new Mutation(cfm.ksName, key, cf); - mutations.add(isCounter() ? new CounterMutation(mut, options.getConsistency()) : mut); + RowMutation rm = new RowMutation(cfm.ksName, key, cf); + mutations.add(isCounter() ? new CounterMutation(rm, cl) : rm); } return mutations; } public UpdateParameters makeUpdateParameters(Collection keys, - Composite prefix, - QueryOptions options, + ColumnNameBuilder prefix, + List variables, boolean local, + ConsistencyLevel cl, long now) throws RequestExecutionException, RequestValidationException { // Some lists operation requires reading - Map rows = readRequiredRows(keys, prefix, local, options.getConsistency()); - return new UpdateParameters(cfm, options, getTimestamp(now, options), getTimeToLive(options), rows); + Map rows = readRequiredRows(keys, prefix, local, cl); + return new UpdateParameters(cfm, variables, getTimestamp(now, variables), getTimeToLive(variables), rows); } public static abstract class Parsed extends CFStatement @@ -702,11 +749,16 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF public ModificationStatement prepare(VariableSpecifications boundNames) throws InvalidRequestException { CFMetaData metadata = ThriftValidation.validateColumnFamily(keyspace(), columnFamily()); + CFDefinition cfDef = metadata.getCfDef(); + + // The collected count in the beginning of preparation. + // Will start at non-zero for statements nested inside a BatchStatement (the second and the further ones). + int collected = boundNames.getCollectedCount(); Attributes preparedAttributes = attrs.prepare(keyspace(), columnFamily()); preparedAttributes.collectMarkerSpecification(boundNames); - ModificationStatement stmt = prepareInternal(metadata, boundNames, preparedAttributes); + ModificationStatement stmt = prepareInternal(cfDef, boundNames, preparedAttributes); if (ifNotExists || ifExists || !conditions.isEmpty()) { @@ -714,7 +766,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF throw new InvalidRequestException("Conditional updates are not supported on counter tables"); if (attrs.timestamp != null) - throw new InvalidRequestException("Cannot provide custom timestamp for conditional updates"); + throw new InvalidRequestException("Cannot provide custom timestamp for conditional update"); if (ifNotExists) { @@ -734,28 +786,32 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF { for (Pair entry : conditions) { - ColumnDefinition def = metadata.getColumnDefinition(entry.left); - if (def == null) + CFDefinition.Name name = cfDef.get(entry.left); + if (name == null) throw new InvalidRequestException(String.format("Unknown identifier %s", entry.left)); - ColumnCondition condition = entry.right.prepare(keyspace(), def); + ColumnCondition condition = entry.right.prepare(name); condition.collectMarkerSpecification(boundNames); - switch (def.kind) + switch (name.kind) { - case PARTITION_KEY: - case CLUSTERING_COLUMN: + case KEY_ALIAS: + case COLUMN_ALIAS: throw new InvalidRequestException(String.format("PRIMARY KEY part %s found in SET part", entry.left)); - default: + case VALUE_ALIAS: + case COLUMN_METADATA: + case STATIC: stmt.addCondition(condition); break; } } } } + + stmt.boundTerms = boundNames.getCollectedCount() - collected; return stmt; } - protected abstract ModificationStatement prepareInternal(CFMetaData cfm, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException; + protected abstract ModificationStatement prepareInternal(CFDefinition cfDef, VariableSpecifications boundNames, Attributes attrs) throws InvalidRequestException; } } diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java index 240a362105..e5db1d7c45 100644 --- a/src/java/org/apache/cassandra/net/MessagingService.java +++ b/src/java/org/apache/cassandra/net/MessagingService.java @@ -512,11 +512,11 @@ public final class MessagingService implements MessagingServiceMBean cp = new OutboundTcpConnectionPool(to); OutboundTcpConnectionPool existingPool = connectionManagers.putIfAbsent(to, cp); if (existingPool != null) - { - cp.close(); cp = existingPool; - } + else + cp.start(); } + cp.waitForStarted(); return cp; } diff --git a/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java b/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java index 81168c6873..c45fc530a4 100644 --- a/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java +++ b/src/java/org/apache/cassandra/net/OutboundTcpConnectionPool.java @@ -22,6 +22,8 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.Socket; import java.nio.channels.SocketChannel; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.cassandra.concurrent.Stage; import org.apache.cassandra.config.Config; @@ -36,6 +38,7 @@ public class OutboundTcpConnectionPool { // pointer for the real Address. private final InetAddress id; + private final CountDownLatch started; public final OutboundTcpConnection cmdCon; public final OutboundTcpConnection ackCon; // pointer to the reseted Address. @@ -46,13 +49,10 @@ public class OutboundTcpConnectionPool { id = remoteEp; resetedEndpoint = SystemKeyspace.getPreferredIP(remoteEp); + started = new CountDownLatch(1); cmdCon = new OutboundTcpConnection(this); - cmdCon.start(); ackCon = new OutboundTcpConnection(this); - ackCon.start(); - - metrics = new ConnectionMetrics(id, this); } /** @@ -167,14 +167,45 @@ public class OutboundTcpConnectionPool } return true; } + + public void start() + { + cmdCon.start(); + ackCon.start(); - public void close() + metrics = new ConnectionMetrics(id, this); + + started.countDown(); + } + + public void waitForStarted() + { + if (started.getCount() == 0) + return; + + boolean error = false; + try + { + if (!started.await(1, TimeUnit.MINUTES)) + error = true; + } + catch (InterruptedException e) + { + Thread.currentThread().interrupt(); + error = true; + } + if (error) + throw new IllegalStateException(String.format("Connections to %s are not started!", id.getHostAddress())); + } + + public void close() { // these null guards are simply for tests if (ackCon != null) ackCon.closeSocket(true); if (cmdCon != null) cmdCon.closeSocket(true); + metrics.release(); } }