natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Return current values on failed CAS 

patch by slebresne; reviewed by jbellis for CASSANDRA-5619
This commit is contained in:
Sylvain Lebresne 2013-06-21 09:47:50 +02:00
parent 56d2296adf
commit 0ba5cf12b9
20 changed files with 613 additions and 471 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -9,7 +9,7 @@
* Removed compatibility with pre-1.2.5 sstables and network messages
(CASSANDRA-5511)
* removed PBSPredictor (CASSANDRA-5455)
* CAS support (CASSANDRA-5062, 5441, 5442, 5443)
* CAS support (CASSANDRA-5062, 5441, 5442, 5443, 5619)
* Leveled compaction performs size-tiered compactions in L0
(CASSANDRA-5371, 5439)
* Add yaml network topology snitch for mixed ec2/other envs (CASSANDRA-5339)

13
interface/cassandra.thrift
View File

@ -646,12 +646,15 @@ service Cassandra {
/**
* Atomic compare and set
*
* The returned list of columns will be null if the cas succeed. Otherwise, it will contain the current
* values for the columns in {@param expected}.
*/
bool cas(1:required binary key,
2:required string column_family,
3:list<Column> expected,
4:list<Column> updates,
5:required ConsistencyLevel consistency_level=ConsistencyLevel.QUORUM)
list<Column> cas(1:required binary key,
2:required string column_family,
3:list<Column> expected,
4:list<Column> updates,
5:required ConsistencyLevel consistency_level=ConsistencyLevel.QUORUM)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
/**

718
interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
View File

File diff suppressed because it is too large Load Diff

4
interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java
View File

@ -3601,7 +3601,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.column_metadata = new ArrayList<ColumnDef>(_list92.size);
for (int _i93 = 0; _i93 < _list92.size; ++_i93)
{
ColumnDef _elem94; // optional
ColumnDef _elem94; // required
_elem94 = new ColumnDef();
_elem94.read(iprot);
struct.column_metadata.add(_elem94);
@ -4482,7 +4482,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.column_metadata = new ArrayList<ColumnDef>(_list121.size);
for (int _i122 = 0; _i122 < _list121.size; ++_i122)
{
ColumnDef _elem123; // optional
ColumnDef _elem123; // required
_elem123 = new ColumnDef();
_elem123.read(iprot);
struct.column_metadata.add(_elem123);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/CounterSuperColumn.java
View File

@ -479,7 +479,7 @@ public class CounterSuperColumn implements org.apache.thrift.TBase<CounterSuperC
struct.columns = new ArrayList<CounterColumn>(_list8.size);
for (int _i9 = 0; _i9 < _list8.size; ++_i9)
{
CounterColumn _elem10; // optional
CounterColumn _elem10; // required
_elem10 = new CounterColumn();
_elem10.read(iprot);
struct.columns.add(_elem10);
@ -560,7 +560,7 @@ public class CounterSuperColumn implements org.apache.thrift.TBase<CounterSuperC
struct.columns = new ArrayList<CounterColumn>(_list13.size);
for (int _i14 = 0; _i14 < _list13.size; ++_i14)
{
CounterColumn _elem15; // optional
CounterColumn _elem15; // required
_elem15 = new CounterColumn();
_elem15.read(iprot);
struct.columns.add(_elem15);

8
interface/thrift/gen-java/org/apache/cassandra/thrift/CqlPreparedResult.java
View File

@ -653,7 +653,7 @@ public class CqlPreparedResult implements org.apache.thrift.TBase<CqlPreparedRes
struct.variable_types = new ArrayList<String>(_list194.size);
for (int _i195 = 0; _i195 < _list194.size; ++_i195)
{
String _elem196; // optional
String _elem196; // required
_elem196 = iprot.readString();
struct.variable_types.add(_elem196);
}
@ -671,7 +671,7 @@ public class CqlPreparedResult implements org.apache.thrift.TBase<CqlPreparedRes
struct.variable_names = new ArrayList<String>(_list197.size);
for (int _i198 = 0; _i198 < _list197.size; ++_i198)
{
String _elem199; // optional
String _elem199; // required
_elem199 = iprot.readString();
struct.variable_names.add(_elem199);
}
@ -798,7 +798,7 @@ public class CqlPreparedResult implements org.apache.thrift.TBase<CqlPreparedRes
struct.variable_types = new ArrayList<String>(_list204.size);
for (int _i205 = 0; _i205 < _list204.size; ++_i205)
{
String _elem206; // optional
String _elem206; // required
_elem206 = iprot.readString();
struct.variable_types.add(_elem206);
}
@ -811,7 +811,7 @@ public class CqlPreparedResult implements org.apache.thrift.TBase<CqlPreparedRes
struct.variable_names = new ArrayList<String>(_list207.size);
for (int _i208 = 0; _i208 < _list207.size; ++_i208)
{
String _elem209; // optional
String _elem209; // required
_elem209 = iprot.readString();
struct.variable_names.add(_elem209);
}

4
interface/thrift/gen-java/org/apache/cassandra/thrift/CqlResult.java
View File

@ -648,7 +648,7 @@ public class CqlResult implements org.apache.thrift.TBase<CqlResult, CqlResult._
struct.rows = new ArrayList<CqlRow>(_list186.size);
for (int _i187 = 0; _i187 < _list186.size; ++_i187)
{
CqlRow _elem188; // optional
CqlRow _elem188; // required
_elem188 = new CqlRow();
_elem188.read(iprot);
struct.rows.add(_elem188);
@ -781,7 +781,7 @@ public class CqlResult implements org.apache.thrift.TBase<CqlResult, CqlResult._
struct.rows = new ArrayList<CqlRow>(_list191.size);
for (int _i192 = 0; _i192 < _list191.size; ++_i192)
{
CqlRow _elem193; // optional
CqlRow _elem193; // required
_elem193 = new CqlRow();
_elem193.read(iprot);
struct.rows.add(_elem193);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/CqlRow.java
View File

@ -482,7 +482,7 @@ public class CqlRow implements org.apache.thrift.TBase<CqlRow, CqlRow._Fields>,
struct.columns = new ArrayList<Column>(_list158.size);
for (int _i159 = 0; _i159 < _list158.size; ++_i159)
{
Column _elem160; // optional
Column _elem160; // required
_elem160 = new Column();
_elem160.read(iprot);
struct.columns.add(_elem160);
@ -563,7 +563,7 @@ public class CqlRow implements org.apache.thrift.TBase<CqlRow, CqlRow._Fields>,
struct.columns = new ArrayList<Column>(_list163.size);
for (int _i164 = 0; _i164 < _list163.size; ++_i164)
{
Column _elem165; // optional
Column _elem165; // required
_elem165 = new Column();
_elem165.read(iprot);
struct.columns.add(_elem165);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/IndexClause.java
View File

@ -559,7 +559,7 @@ public class IndexClause implements org.apache.thrift.TBase<IndexClause, IndexCl
struct.expressions = new ArrayList<IndexExpression>(_list24.size);
for (int _i25 = 0; _i25 < _list24.size; ++_i25)
{
IndexExpression _elem26; // optional
IndexExpression _elem26; // required
_elem26 = new IndexExpression();
_elem26.read(iprot);
struct.expressions.add(_elem26);
@ -661,7 +661,7 @@ public class IndexClause implements org.apache.thrift.TBase<IndexClause, IndexCl
struct.expressions = new ArrayList<IndexExpression>(_list29.size);
for (int _i30 = 0; _i30 < _list29.size; ++_i30)
{
IndexExpression _elem31; // optional
IndexExpression _elem31; // required
_elem31 = new IndexExpression();
_elem31.read(iprot);
struct.expressions.add(_elem31);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/KeyRange.java
View File

@ -848,7 +848,7 @@ public class KeyRange implements org.apache.thrift.TBase<KeyRange, KeyRange._Fie
struct.row_filter = new ArrayList<IndexExpression>(_list32.size);
for (int _i33 = 0; _i33 < _list32.size; ++_i33)
{
IndexExpression _elem34; // optional
IndexExpression _elem34; // required
_elem34 = new IndexExpression();
_elem34.read(iprot);
struct.row_filter.add(_elem34);
@ -1017,7 +1017,7 @@ public class KeyRange implements org.apache.thrift.TBase<KeyRange, KeyRange._Fie
struct.row_filter = new ArrayList<IndexExpression>(_list37.size);
for (int _i38 = 0; _i38 < _list37.size; ++_i38)
{
IndexExpression _elem39; // optional
IndexExpression _elem39; // required
_elem39 = new IndexExpression();
_elem39.read(iprot);
struct.row_filter.add(_elem39);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/KeySlice.java
View File

@ -486,7 +486,7 @@ public class KeySlice implements org.apache.thrift.TBase<KeySlice, KeySlice._Fie
struct.columns = new ArrayList<ColumnOrSuperColumn>(_list40.size);
for (int _i41 = 0; _i41 < _list40.size; ++_i41)
{
ColumnOrSuperColumn _elem42; // optional
ColumnOrSuperColumn _elem42; // required
_elem42 = new ColumnOrSuperColumn();
_elem42.read(iprot);
struct.columns.add(_elem42);
@ -567,7 +567,7 @@ public class KeySlice implements org.apache.thrift.TBase<KeySlice, KeySlice._Fie
struct.columns = new ArrayList<ColumnOrSuperColumn>(_list45.size);
for (int _i46 = 0; _i46 < _list45.size; ++_i46)
{
ColumnOrSuperColumn _elem47; // optional
ColumnOrSuperColumn _elem47; // required
_elem47 = new ColumnOrSuperColumn();
_elem47.read(iprot);
struct.columns.add(_elem47);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java
View File

@ -869,7 +869,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
struct.cf_defs = new ArrayList<CfDef>(_list144.size);
for (int _i145 = 0; _i145 < _list144.size; ++_i145)
{
CfDef _elem146; // optional
CfDef _elem146; // required
_elem146 = new CfDef();
_elem146.read(iprot);
struct.cf_defs.add(_elem146);
@ -1018,7 +1018,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
struct.cf_defs = new ArrayList<CfDef>(_list151.size);
for (int _i152 = 0; _i152 < _list151.size; ++_i152)
{
CfDef _elem153; // optional
CfDef _elem153; // required
_elem153 = new CfDef();
_elem153.read(iprot);
struct.cf_defs.add(_elem153);

4
interface/thrift/gen-java/org/apache/cassandra/thrift/SlicePredicate.java
View File

@ -467,7 +467,7 @@ public class SlicePredicate implements org.apache.thrift.TBase<SlicePredicate, S
struct.column_names = new ArrayList<ByteBuffer>(_list16.size);
for (int _i17 = 0; _i17 < _list16.size; ++_i17)
{
ByteBuffer _elem18; // optional
ByteBuffer _elem18; // required
_elem18 = iprot.readBinary();
struct.column_names.add(_elem18);
}
@ -572,7 +572,7 @@ public class SlicePredicate implements org.apache.thrift.TBase<SlicePredicate, S
struct.column_names = new ArrayList<ByteBuffer>(_list21.size);
for (int _i22 = 0; _i22 < _list21.size; ++_i22)
{
ByteBuffer _elem23; // optional
ByteBuffer _elem23; // required
_elem23 = iprot.readBinary();
struct.column_names.add(_elem23);
}

4
interface/thrift/gen-java/org/apache/cassandra/thrift/SuperColumn.java
View File

@ -485,7 +485,7 @@ public class SuperColumn implements org.apache.thrift.TBase<SuperColumn, SuperCo
struct.columns = new ArrayList<Column>(_list0.size);
for (int _i1 = 0; _i1 < _list0.size; ++_i1)
{
Column _elem2; // optional
Column _elem2; // required
_elem2 = new Column();
_elem2.read(iprot);
struct.columns.add(_elem2);
@ -566,7 +566,7 @@ public class SuperColumn implements org.apache.thrift.TBase<SuperColumn, SuperCo
struct.columns = new ArrayList<Column>(_list5.size);
for (int _i6 = 0; _i6 < _list5.size; ++_i6)
{
Column _elem7; // optional
Column _elem7; // required
_elem7 = new Column();
_elem7.read(iprot);
struct.columns.add(_elem7);

12
interface/thrift/gen-java/org/apache/cassandra/thrift/TokenRange.java
View File

@ -774,7 +774,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.endpoints = new ArrayList<String>(_list48.size);
for (int _i49 = 0; _i49 < _list48.size; ++_i49)
{
String _elem50; // optional
String _elem50; // required
_elem50 = iprot.readString();
struct.endpoints.add(_elem50);
}
@ -792,7 +792,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.rpc_endpoints = new ArrayList<String>(_list51.size);
for (int _i52 = 0; _i52 < _list51.size; ++_i52)
{
String _elem53; // optional
String _elem53; // required
_elem53 = iprot.readString();
struct.rpc_endpoints.add(_elem53);
}
@ -810,7 +810,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.endpoint_details = new ArrayList<EndpointDetails>(_list54.size);
for (int _i55 = 0; _i55 < _list54.size; ++_i55)
{
EndpointDetails _elem56; // optional
EndpointDetails _elem56; // required
_elem56 = new EndpointDetails();
_elem56.read(iprot);
struct.endpoint_details.add(_elem56);
@ -953,7 +953,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.endpoints = new ArrayList<String>(_list63.size);
for (int _i64 = 0; _i64 < _list63.size; ++_i64)
{
String _elem65; // optional
String _elem65; // required
_elem65 = iprot.readString();
struct.endpoints.add(_elem65);
}
@ -966,7 +966,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.rpc_endpoints = new ArrayList<String>(_list66.size);
for (int _i67 = 0; _i67 < _list66.size; ++_i67)
{
String _elem68; // optional
String _elem68; // required
_elem68 = iprot.readString();
struct.rpc_endpoints.add(_elem68);
}
@ -979,7 +979,7 @@ public class TokenRange implements org.apache.thrift.TBase<TokenRange, TokenRang
struct.endpoint_details = new ArrayList<EndpointDetails>(_list69.size);
for (int _i70 = 0; _i70 < _list69.size; ++_i70)
{
EndpointDetails _elem71; // optional
EndpointDetails _elem71; // required
_elem71 = new EndpointDetails();
_elem71.read(iprot);
struct.endpoint_details.add(_elem71);

31
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -372,11 +372,34 @@ public abstract class ModificationStatement implements CQLStatement
ColumnFamily updates = updateForKey(key, clusteringPrefix, params);
ColumnFamily expected = buildConditions(key, clusteringPrefix, params);
boolean result = StorageProxy.cas(keyspace(), columnFamily(), key, expected, updates, cl);
ColumnFamily result = StorageProxy.cas(keyspace(), columnFamily(), key, clusteringPrefix, expected, updates, cl);
return result == null
? new ResultMessage.Void()
: new ResultMessage.Rows(buildCasFailureResultSet(key, result));
}
ResultSet.Metadata metadata = new ResultSet.Metadata(Collections.singletonList(new ColumnSpecification(keyspace(), columnFamily(), RESULT_COLUMN, BooleanType.instance)));
List<List<ByteBuffer>> newRows = Collections.singletonList(Collections.singletonList(BooleanType.instance.decompose(result)));
return new ResultMessage.Rows(new ResultSet(metadata, newRows));
private ResultSet buildCasFailureResultSet(ByteBuffer key, ColumnFamily cf) throws InvalidRequestException
{
CFDefinition cfDef = cfm.getCfDef();
Selection selection;
if (ifNotExists)
{
selection = Selection.wildcard(cfDef);
}
else
{
List<CFDefinition.Name> names = new ArrayList<CFDefinition.Name>(columnConditions.size());
for (Operation condition : columnConditions)
names.add(cfDef.get(condition.columnName));
selection = Selection.forColumns(names);
}
long now = System.currentTimeMillis();
Selection.ResultSetBuilder builder = selection.resultSetBuilder(now);
SelectStatement.forSelection(cfDef, selection).processColumnFamily(key, cf, Collections.<ByteBuffer>emptyList(), Integer.MAX_VALUE, now, builder);
return builder.build();
}
public ResultMessage executeInternal(QueryState queryState) throws RequestValidationException, RequestExecutionException

192
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -76,6 +76,9 @@ public class SelectStatement implements CQLStatement
private Map<CFDefinition.Name, Integer> orderingIndexes;
// Used by forSelection below
private static final Parameters defaultParameters = new Parameters(Collections.<ColumnIdentifier, Boolean>emptyMap(), false, null, false);
private static enum Bound
{
START(0), END(1);
@ -104,6 +107,14 @@ public class SelectStatement implements CQLStatement
this.limit = limit;
}
// Creates a simple select based on the given selection.
// Note that the results select statement should not be used for actual queries, but only for processing already
// queried data through processColumnFamily.
static SelectStatement forSelection(CFDefinition cfDef, Selection selection)
{
return new SelectStatement(cfDef, 0, defaultParameters, selection, null);
}
public int getBoundsTerms()
{
return boundTerms;
@ -618,6 +629,9 @@ public class SelectStatement implements CQLStatement
private Iterable<Column> columnsInOrder(final ColumnFamily cf, final List<ByteBuffer> variables) throws InvalidRequestException
{
if (columnRestrictions.length == 0)
return cf.getSortedColumns();
// If the restriction for the last column alias is an IN, respect
// requested order
Restriction last = columnRestrictions[columnRestrictions.length - 1];
@ -665,92 +679,7 @@ public class SelectStatement implements CQLStatement
if (row.cf == null)
continue;
ByteBuffer[] keyComponents = cfDef.hasCompositeKey
? ((CompositeType)cfDef.cfm.getKeyValidator()).split(row.key.key)
: new ByteBuffer[]{ row.key.key };
if (cfDef.isCompact)
{
// One cqlRow per column
for (Column c : columnsInOrder(row.cf, variables))
{
if (c.isMarkedForDelete(now))
continue;
ByteBuffer[] components = null;
if (cfDef.isComposite)
{
components = ((CompositeType)cfDef.cfm.comparator).split(c.name());
}
else if (sliceRestriction != null)
{
// For dynamic CF, the column could be out of the requested bounds, filter here
if (!sliceRestriction.isInclusive(Bound.START) && c.name().equals(sliceRestriction.bound(Bound.START).bindAndGet(variables)))
continue;
if (!sliceRestriction.isInclusive(Bound.END) && c.name().equals(sliceRestriction.bound(Bound.END).bindAndGet(variables)))
continue;
}
result.newRow();
// Respect selection order
for (CFDefinition.Name name : selection.getColumnsList())
{
switch (name.kind)
{
case KEY_ALIAS:
result.add(keyComponents[name.position]);
break;
case COLUMN_ALIAS:
ByteBuffer val = cfDef.isComposite
? (name.position < components.length ? components[name.position] : null)
: c.name();
result.add(val);
break;
case VALUE_ALIAS:
result.add(c);
break;
case COLUMN_METADATA:
// This should not happen for compact CF
throw new AssertionError();
default:
throw new AssertionError();
}
}
}
}
else if (cfDef.isComposite)
{
// Sparse case: group column in cqlRow when composite prefix is equal
CompositeType composite = (CompositeType)cfDef.cfm.comparator;
ColumnGroupMap.Builder builder = new ColumnGroupMap.Builder(composite, cfDef.hasCollections, now);
for (Column c : row.cf)
{
if (c.isMarkedForDelete(now))
continue;
builder.add(c);
}
for (ColumnGroupMap group : builder.groups())
handleGroup(selection, result, keyComponents, group);
}
else
{
if (row.cf.hasOnlyTombstones(now))
continue;
// Static case: One cqlRow for all columns
result.newRow();
for (CFDefinition.Name name : selection.getColumnsList())
{
if (name.kind == CFDefinition.Name.Kind.KEY_ALIAS)
result.add(keyComponents[name.position]);
else
result.add(row.cf.getColumn(name.name.key));
}
}
processColumnFamily(row.key.key, row.cf, variables, limit, now, result);
}
ResultSet cqlRows = result.build();
@ -766,6 +695,97 @@ public class SelectStatement implements CQLStatement
return cqlRows;
}
// Used by ModificationStatement for CAS operations
void processColumnFamily(ByteBuffer key, ColumnFamily cf, List<ByteBuffer> variables, int limit, long now, Selection.ResultSetBuilder result) throws InvalidRequestException
{
ByteBuffer[] keyComponents = cfDef.hasCompositeKey
? ((CompositeType)cfDef.cfm.getKeyValidator()).split(key)
: new ByteBuffer[]{ key };
if (cfDef.isCompact)
{
// One cqlRow per column
for (Column c : columnsInOrder(cf, variables))
{
if (c.isMarkedForDelete(now))
continue;
ByteBuffer[] components = null;
if (cfDef.isComposite)
{
components = ((CompositeType)cfDef.cfm.comparator).split(c.name());
}
else if (sliceRestriction != null)
{
// For dynamic CF, the column could be out of the requested bounds, filter here
if (!sliceRestriction.isInclusive(Bound.START) && c.name().equals(sliceRestriction.bound(Bound.START).bindAndGet(variables)))
continue;
if (!sliceRestriction.isInclusive(Bound.END) && c.name().equals(sliceRestriction.bound(Bound.END).bindAndGet(variables)))
continue;
}
result.newRow();
// Respect selection order
for (CFDefinition.Name name : selection.getColumnsList())
{
switch (name.kind)
{
case KEY_ALIAS:
result.add(keyComponents[name.position]);
break;
case COLUMN_ALIAS:
ByteBuffer val = cfDef.isComposite
? (name.position < components.length ? components[name.position] : null)
: c.name();
result.add(val);
break;
case VALUE_ALIAS:
result.add(c);
break;
case COLUMN_METADATA:
// This should not happen for compact CF
throw new AssertionError();
default:
throw new AssertionError();
}
}
}
}
else if (cfDef.isComposite)
{
// Sparse case: group column in cqlRow when composite prefix is equal
CompositeType composite = (CompositeType)cfDef.cfm.comparator;
ColumnGroupMap.Builder builder = new ColumnGroupMap.Builder(composite, cfDef.hasCollections, now);
for (Column c : cf)
{
if (c.isMarkedForDelete(now))
continue;
builder.add(c);
}
for (ColumnGroupMap group : builder.groups())
handleGroup(selection, result, keyComponents, group);
}
else
{
if (cf.hasOnlyTombstones(now))
return;
// Static case: One cqlRow for all columns
result.newRow();
for (CFDefinition.Name name : selection.getColumnsList())
{
if (name.kind == CFDefinition.Name.Kind.KEY_ALIAS)
result.add(keyComponents[name.position]);
else
result.add(cf.getColumn(name.name.key));
}
}
}
/**
* Orders results when multiple keys are selected (using IN)
*/

5
src/java/org/apache/cassandra/cql3/statements/Selection.java
View File

@ -57,6 +57,11 @@ public abstract class Selection
return new SimpleSelection(all);
}
public static Selection forColumns(List<CFDefinition.Name> columnsList)
{
return new SimpleSelection(columnsList);
}
private static boolean isUsingFunction(List<RawSelector> rawSelectors)
{
for (RawSelector rawSelector : rawSelectors)

41
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -40,6 +40,7 @@ import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.ColumnNameBuilder;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.Table;
import org.apache.cassandra.db.filter.ColumnSlice;
@ -159,7 +160,7 @@ public class StorageProxy implements StorageProxyMBean
/**
* Apply @param updates if and only if the current values in the row for @param key
* match the ones given by @param old. The algorithm is "raw" Paxos: that is, Paxos
* match the ones given by @param expected. The algorithm is "raw" Paxos: that is, Paxos
* minus leader election -- any node in the cluster may propose changes for any row,
* which (that is, the row) is the unit of values being proposed, not single columns.
*
@ -188,9 +189,21 @@ public class StorageProxy implements StorageProxyMBean
* values) between the prepare and accept phases. This gives us a slightly longer window for another
* coordinator to come along and trump our own promise with a newer one but is otherwise safe.
*
* @return true if the operation succeeds in updating the row
* @param table the table for the CAS
* @param cfName the column family for the CAS
* @param key the row key for the row to CAS
* @param prefix a column name prefix that selects the CQL3 row to check if {@code expected} is null. If {@code expected}
* is not null, this is ignored. If {@code expected} is null and this is null, the full row existing is checked (by querying
* the first live column of the row).
* @param expected the expected column values. This can be null to check for existence (see {@code prefix}).
* @param updates the value to insert if {@code expected matches the current values}.
* @param consistencyLevel the consistency for the operation.
*
* @return null if the operation succeeds in updating the row, or the current values for the columns contained in
* expected (since, if the CAS doesn't succeed, it means the current value do not match the one in expected). If
* expected == null and the CAS is unsuccessfull, the first live column of the CF is returned.
*/
public static boolean cas(String table, String cfName, ByteBuffer key, ColumnFamily expected, ColumnFamily updates, ConsistencyLevel consistencyLevel)
public static ColumnFamily cas(String table, String cfName, ByteBuffer key, ColumnNameBuilder prefix, ColumnFamily expected, ColumnFamily updates, ConsistencyLevel consistencyLevel)
throws UnavailableException, IsBootstrappingException, ReadTimeoutException, WriteTimeoutException, InvalidRequestException
{
consistencyLevel.validateForCas(table);
@ -213,18 +226,24 @@ public class StorageProxy implements StorageProxyMBean
// read the current value and compare with expected
Tracing.trace("Reading existing values for CAS precondition");
long timestamp = System.currentTimeMillis();
IDiskAtomFilter filter = expected == null
? new SliceQueryFilter(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1)
: new NamesQueryFilter(ImmutableSortedSet.copyOf(expected.getColumnNames()));
ReadCommand readCommand = filter instanceof SliceQueryFilter
? new SliceFromReadCommand(table, key, cfName, timestamp, (SliceQueryFilter) filter)
: new SliceByNamesReadCommand(table, key, cfName, timestamp, (NamesQueryFilter) filter);
ReadCommand readCommand;
if (expected == null)
{
SliceQueryFilter filter = prefix == null
? new SliceQueryFilter(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1)
: new SliceQueryFilter(prefix.build(), prefix.buildAsEndOfRange(), false, 1, prefix.componentCount());
readCommand = new SliceFromReadCommand(table, key, cfName, timestamp, filter);
}
else
{
readCommand = new SliceByNamesReadCommand(table, key, cfName, timestamp, new NamesQueryFilter(ImmutableSortedSet.copyOf(expected.getColumnNames())));
}
List<Row> rows = read(Arrays.asList(readCommand), ConsistencyLevel.QUORUM);
ColumnFamily current = rows.get(0).cf;
if (!casApplies(expected, current))
{
Tracing.trace("CAS precondition {} does not match current values {}", expected, current);
return false;
return current;
}
// finish the paxos round w/ the desired updates
@ -238,7 +257,7 @@ public class StorageProxy implements StorageProxyMBean
else
commitPaxos(proposal, consistencyLevel);
Tracing.trace("CAS successful");
return true;
return null;
}
Tracing.trace("Paxos proposal not accepted (pre-empted by a higher ballot)");

22
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -160,6 +160,19 @@ public class CassandraServer implements Cassandra.Iface
return thrift_column;
}
private List<Column> thriftifyColumnsAsColumns(Collection<org.apache.cassandra.db.Column> columns, long now)
{
List<Column> thriftColumns = new ArrayList<Column>(columns.size());
for (org.apache.cassandra.db.Column column : columns)
{
if (column.isMarkedForDelete(now))
continue;
thriftColumns.add(thriftifySubColumn(column));
}
return thriftColumns;
}
private CounterColumn thriftifySubCounter(org.apache.cassandra.db.Column column)
{
assert column instanceof org.apache.cassandra.db.CounterColumn;
@ -690,7 +703,7 @@ public class CassandraServer implements Cassandra.Iface
}
}
public boolean cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level)
public List<Column> cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (startSessionIfRequested())
@ -746,7 +759,10 @@ public class CassandraServer implements Cassandra.Iface
}
schedule(DatabaseDescriptor.getWriteRpcTimeout());
return StorageProxy.cas(cState.getKeyspace(), column_family, key, cfExpected, cfUpdates, ThriftConversion.fromThrift(consistency_level));
ColumnFamily result = StorageProxy.cas(cState.getKeyspace(), column_family, key, null, cfExpected, cfUpdates, ThriftConversion.fromThrift(consistency_level));
return result == null
? null
: thriftifyColumnsAsColumns(result.getSortedColumns(), System.currentTimeMillis());
}
catch (RequestTimeoutException e)
{
@ -759,7 +775,7 @@ public class CassandraServer implements Cassandra.Iface
catch (RequestExecutionException e)
{
ThriftConversion.rethrow(e);
return false; // makes javac happy -- it can't tell that rethrow always throws
return null; // makes javac happy -- it can't tell that rethrow always throws
}
finally
{