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

DC-local CAS 

patch by slebresne; reviewed by jbellis for CASSANDRA-5797
This commit is contained in:
Sylvain Lebresne 2013-07-25 10:25:31 +02:00
parent afe4d555bb
commit bfd73beaf1
32 changed files with 719 additions and 453 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -4,6 +4,7 @@
* Better validation when accessing CQL3 table from thrift (CASSANDRA-5138)
* Fix assertion error during repair (CASSANDRA-5801)
* Fix range tombstone bug (CASSANDRA-5805)
* DC-local CAS (CASSANDRA-5797)
2.0.0-beta2

58
doc/native_protocol_v2.spec
View File

@ -218,6 +218,8 @@ Table of Contents
0x0005 ALL
0x0006 LOCAL_QUORUM
0x0007 EACH_QUORUM
0x0008 SERIAL
0x0009 LOCAL_SERIAL
[string map] A [short] n, followed by n pair <k><v> where <k> and <v>
are [string].
@ -280,29 +282,37 @@ Table of Contents
4.1.4. QUERY
Performs a CQL query. The body of the message must be:
<query><consistency><flags>[<result_page_size>][<n><value_1>...<value_n>][<paging_state>]
<query><query_parameters>
where <query> is a [long string] representing the query and
<query_parameters> must be
<consistency><flags>[<n><value_1>...<value_n>][<result_page_size>][<paging_state>][<serial_consistency>]
where:
- <consistency> is the [consistency] level for the operation.
- <flags> is a [byte] whose bits define the options for this query and
in particular influence what the remainder of the message contains.
A flag is set if the bit corresponding to its `mask` is set. Supported
flags are, given there mask:
0x01: Page_size. In that case, <result_page_size> is an [int]
controlling the desired page size of the result (in CQL3 rows).
See the section on paging (Section 7) for more details.
0x02: Values. In that case, a [short] <n> followed by <n> [bytes]
0x01: Values. In that case, a [short] <n> followed by <n> [bytes]
values are provided. Those value are used for bound variables in
the query.
0x04: Skip_metadata. If present, the Result Set returned as a response
0x02: Skip_metadata. If present, the Result Set returned as a response
to that query (if any) will have the NO_METADATA flag (see
Section 4.2.5.2).
0x03: Page_size. In that case, <result_page_size> is an [int]
controlling the desired page size of the result (in CQL3 rows).
See the section on paging (Section 7) for more details.
0x04: With_paging_state. If present, <paging_state> should be present.
<paging_state> is a [bytes] value that should have been returned
in a result set (Section 4.2.5.2). If provided, the query will be
executed but starting from a given paging state. This also to
continue paging on a different node from the one it has been
started (See Section 7 for more details).
- <query> the query, [long string].
- <consistency> is the [consistency] level for the operation.
0x05: With serial consistency. If present, <serial_consistency> should be
present. <serial_consistency> is the [consistency] level for the
serial phase of conditional updates. That consitency can only be
either SERIAL or LOCAL_SERIAL and if not present, it defaults to
SERIAL. This option will be ignored for anything else that a
conditional update/insert.
Note that the consistency is ignored by some queries (USE, CREATE, ALTER,
TRUNCATE, ...).
@ -323,36 +333,14 @@ Table of Contents
4.1.6. EXECUTE
Executes a prepared query. The body of the message must be:
<id><n><value_1>....<value_n><consistency><flags>[<result_page_size>][<paging_state>]
where:
- <id> is the prepared query ID. It's the [short bytes] returned as a
response to a PREPARE message.
- <n> is a [short] indicating the number of following values.
- <value_1>...<value_n> are the [bytes] to use for bound variables in the
prepared query.
- <consistency> is the [consistency] level for the operation.
- <flags> is a [byte] whose bits define the options for this query and
in particular influence what the remainder of the message contains.
A flag is set if the bit corresponding to its `mask` is set. Supported
flags are, given there mask:
0x01: Page size. In that case, <result_page_size> is an [int]
controlling the desired page size of the result (in CQL3 rows).
See the section on paging (Section 7) for more details.
0x02: Skip metadata. If present, the Result Set returned as a response
to that query (if any) will have the NO_METADATA flag (see
Section 4.2.5.2).
0x03: With_paging_state. If present, <paging_state> should be present.
<paging_state> is a [bytes] value that should have been returned
in a result set (Section 4.2.5.2). If provided, the query will be
executed but starting from a given paging state. This also to
continue paging on a different node from the one it has been
started (See Section 7 for more details).
Note that the consistency is ignored by some (prepared) queries (USE, CREATE,
ALTER, TRUNCATE, ...).
<id><query_parameters>
where <id> is the prepared query ID. It's the [short bytes] returned as a
response to a PREPARE message. As for <query_parameters>, it has the exact
same definition than in QUERY (see Section 4.1.4).
The response from the server will be a RESULT message.
4.1.7. BATCH
Allows executing a list of queries (prepared or not) as a batch (note that

13
interface/cassandra.thrift
View File

@ -236,6 +236,7 @@ enum ConsistencyLevel {
TWO = 7,
THREE = 8,
SERIAL = 9,
LOCAL_SERIAL = 10,
}
/**
@ -664,12 +665,22 @@ service Cassandra {
* If the cas is successfull, the success boolean in CASResult will be true and there will be no current_values.
* Otherwise, success will be false and current_values will contain the current values for the columns in
* expected (that, by definition of compare-and-set, will differ from the values in expected).
*
* A cas operation takes 2 consistency level. The first one, serial_consistency_level, simply indicates the
* level of serialization required. This can be either ConsistencyLevel.SERIAL or ConsistencyLevel.LOCAL_SERIAL.
* The second one, commit_consistency_level, defines the consistency level for the commit phase of the cas. This
* is a more traditional consistency level (the same CL than for traditional writes are accepted) that impact
* the visibility for reads of the operation. For instance, if commit_consistency_level is QUORUM, then it is
* guaranteed that a followup QUORUM read will see the cas write (if that one was successful obviously). If
* commit_consistency_level is ANY, you will need to use a SERIAL/LOCAL_SERIAL read to be guaranteed to see
* the write.
*/
CASResult 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)
5:required ConsistencyLevel serial_consistency_level=ConsistencyLevel.SERIAL,
6:required ConsistencyLevel commit_consistency_level=ConsistencyLevel.QUORUM)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
/**

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

@ -383,7 +383,7 @@ public class AuthenticationRequest implements org.apache.thrift.TBase<Authentica
struct.credentials = new HashMap<String,String>(2*_map80.size);
for (int _i81 = 0; _i81 < _map80.size; ++_i81)
{
String _key82; // optional
String _key82; // required
String _val83; // required
_key82 = iprot.readString();
_val83 = iprot.readString();
@ -459,7 +459,7 @@ public class AuthenticationRequest implements org.apache.thrift.TBase<Authentica
struct.credentials = new HashMap<String,String>(2*_map86.size);
for (int _i87 = 0; _i87 < _map86.size; ++_i87)
{
String _key88; // optional
String _key88; // required
String _val89; // required
_key88 = iprot.readString();
_val89 = iprot.readString();

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

@ -170,13 +170,23 @@ public class Cassandra {
* Otherwise, success will be false and current_values will contain the current values for the columns in
* expected (that, by definition of compare-and-set, will differ from the values in expected).
*
* A cas operation takes 2 consistency level. The first one, serial_consistency_level, simply indicates the
* level of serialization required. This can be either ConsistencyLevel.SERIAL or ConsistencyLevel.LOCAL_SERIAL.
* The second one, commit_consistency_level, defines the consistency level for the commit phase of the cas. This
* is a more traditional consistency level (the same CL than for traditional writes are accepted) that impact
* the visibility for reads of the operation. For instance, if commit_consistency_level is QUORUM, then it is
* guaranteed that a followup QUORUM read will see the cas write (if that one was successful obviously). If
* commit_consistency_level is ANY, you will need to use a SERIAL/LOCAL_SERIAL read to be guaranteed to see
* the write.
*
* @param key
* @param column_family
* @param expected
* @param updates
* @param consistency_level
* @param serial_consistency_level
* @param commit_consistency_level
*/
public CASResult cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException;
public CASResult cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException;
/**
* Remove data from the row specified by key at the granularity specified by column_path, and the given timestamp. Note
@ -429,7 +439,7 @@ public class Cassandra {
public void add(ByteBuffer key, ColumnParent column_parent, CounterColumn column, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.add_call> resultHandler) throws org.apache.thrift.TException;
public void cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.cas_call> resultHandler) throws org.apache.thrift.TException;
public void cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.cas_call> resultHandler) throws org.apache.thrift.TException;
public void remove(ByteBuffer key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback<AsyncClient.remove_call> resultHandler) throws org.apache.thrift.TException;
@ -908,20 +918,21 @@ public class Cassandra {
return;
}
public CASResult cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException
public CASResult cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException
{
send_cas(key, column_family, expected, updates, consistency_level);
send_cas(key, column_family, expected, updates, serial_consistency_level, commit_consistency_level);
return recv_cas();
}
public void send_cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level) throws org.apache.thrift.TException
public void send_cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level) throws org.apache.thrift.TException
{
cas_args args = new cas_args();
args.setKey(key);
args.setColumn_family(column_family);
args.setExpected(expected);
args.setUpdates(updates);
args.setConsistency_level(consistency_level);
args.setSerial_consistency_level(serial_consistency_level);
args.setCommit_consistency_level(commit_consistency_level);
sendBase("cas", args);
}
@ -2280,9 +2291,9 @@ public class Cassandra {
}
}
public void cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback<cas_call> resultHandler) throws org.apache.thrift.TException {
public void cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback<cas_call> resultHandler) throws org.apache.thrift.TException {
checkReady();
cas_call method_call = new cas_call(key, column_family, expected, updates, consistency_level, resultHandler, this, ___protocolFactory, ___transport);
cas_call method_call = new cas_call(key, column_family, expected, updates, serial_consistency_level, commit_consistency_level, resultHandler, this, ___protocolFactory, ___transport);
this.___currentMethod = method_call;
___manager.call(method_call);
}
@ -2292,14 +2303,16 @@ public class Cassandra {
private String column_family;
private List<Column> expected;
private List<Column> updates;
private ConsistencyLevel consistency_level;
public cas_call(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback<cas_call> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
private ConsistencyLevel serial_consistency_level;
private ConsistencyLevel commit_consistency_level;
public cas_call(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback<cas_call> resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
super(client, protocolFactory, transport, resultHandler, false);
this.key = key;
this.column_family = column_family;
this.expected = expected;
this.updates = updates;
this.consistency_level = consistency_level;
this.serial_consistency_level = serial_consistency_level;
this.commit_consistency_level = commit_consistency_level;
}
public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
@ -2309,7 +2322,8 @@ public class Cassandra {
args.setColumn_family(column_family);
args.setExpected(expected);
args.setUpdates(updates);
args.setConsistency_level(consistency_level);
args.setSerial_consistency_level(serial_consistency_level);
args.setCommit_consistency_level(commit_consistency_level);
args.write(prot);
prot.writeMessageEnd();
}
@ -3731,7 +3745,7 @@ public class Cassandra {
public cas_result getResult(I iface, cas_args args) throws org.apache.thrift.TException {
cas_result result = new cas_result();
try {
result.success = iface.cas(args.key, args.column_family, args.expected, args.updates, args.consistency_level);
result.success = iface.cas(args.key, args.column_family, args.expected, args.updates, args.serial_consistency_level, args.commit_consistency_level);
} catch (InvalidRequestException ire) {
result.ire = ire;
} catch (UnavailableException ue) {
@ -11562,7 +11576,7 @@ public class Cassandra {
struct.success = new HashMap<ByteBuffer,List<ColumnOrSuperColumn>>(2*_map232.size);
for (int _i233 = 0; _i233 < _map232.size; ++_i233)
{
ByteBuffer _key234; // optional
ByteBuffer _key234; // required
List<ColumnOrSuperColumn> _val235; // required
_key234 = iprot.readBinary();
{
@ -11731,7 +11745,7 @@ public class Cassandra {
struct.success = new HashMap<ByteBuffer,List<ColumnOrSuperColumn>>(2*_map243.size);
for (int _i244 = 0; _i244 < _map243.size; ++_i244)
{
ByteBuffer _key245; // optional
ByteBuffer _key245; // required
List<ColumnOrSuperColumn> _val246; // required
_key245 = iprot.readBinary();
{
@ -13082,7 +13096,7 @@ public class Cassandra {
struct.success = new HashMap<ByteBuffer,Integer>(2*_map258.size);
for (int _i259 = 0; _i259 < _map258.size; ++_i259)
{
ByteBuffer _key260; // optional
ByteBuffer _key260; // required
int _val261; // required
_key260 = iprot.readBinary();
_val261 = iprot.readI32();
@ -13227,7 +13241,7 @@ public class Cassandra {
struct.success = new HashMap<ByteBuffer,Integer>(2*_map264.size);
for (int _i265 = 0; _i265 < _map264.size; ++_i265)
{
ByteBuffer _key266; // optional
ByteBuffer _key266; // required
int _val267; // required
_key266 = iprot.readBinary();
_val267 = iprot.readI32();
@ -20093,7 +20107,8 @@ public class Cassandra {
private static final org.apache.thrift.protocol.TField COLUMN_FAMILY_FIELD_DESC = new org.apache.thrift.protocol.TField("column_family", org.apache.thrift.protocol.TType.STRING, (short)2);
private static final org.apache.thrift.protocol.TField EXPECTED_FIELD_DESC = new org.apache.thrift.protocol.TField("expected", org.apache.thrift.protocol.TType.LIST, (short)3);
private static final org.apache.thrift.protocol.TField UPDATES_FIELD_DESC = new org.apache.thrift.protocol.TField("updates", org.apache.thrift.protocol.TType.LIST, (short)4);
private static final org.apache.thrift.protocol.TField CONSISTENCY_LEVEL_FIELD_DESC = new org.apache.thrift.protocol.TField("consistency_level", org.apache.thrift.protocol.TType.I32, (short)5);
private static final org.apache.thrift.protocol.TField SERIAL_CONSISTENCY_LEVEL_FIELD_DESC = new org.apache.thrift.protocol.TField("serial_consistency_level", org.apache.thrift.protocol.TType.I32, (short)5);
private static final org.apache.thrift.protocol.TField COMMIT_CONSISTENCY_LEVEL_FIELD_DESC = new org.apache.thrift.protocol.TField("commit_consistency_level", org.apache.thrift.protocol.TType.I32, (short)6);
private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
static {
@ -20109,7 +20124,12 @@ public class Cassandra {
*
* @see ConsistencyLevel
*/
public ConsistencyLevel consistency_level; // required
public ConsistencyLevel serial_consistency_level; // required
/**
*
* @see ConsistencyLevel
*/
public ConsistencyLevel commit_consistency_level; // required
/** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
public enum _Fields implements org.apache.thrift.TFieldIdEnum {
@ -20121,7 +20141,12 @@ public class Cassandra {
*
* @see ConsistencyLevel
*/
CONSISTENCY_LEVEL((short)5, "consistency_level");
SERIAL_CONSISTENCY_LEVEL((short)5, "serial_consistency_level"),
/**
*
* @see ConsistencyLevel
*/
COMMIT_CONSISTENCY_LEVEL((short)6, "commit_consistency_level");
private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
@ -20144,8 +20169,10 @@ public class Cassandra {
return EXPECTED;
case 4: // UPDATES
return UPDATES;
case 5: // CONSISTENCY_LEVEL
return CONSISTENCY_LEVEL;
case 5: // SERIAL_CONSISTENCY_LEVEL
return SERIAL_CONSISTENCY_LEVEL;
case 6: // COMMIT_CONSISTENCY_LEVEL
return COMMIT_CONSISTENCY_LEVEL;
default:
return null;
}
@ -20199,14 +20226,18 @@ public class Cassandra {
tmpMap.put(_Fields.UPDATES, new org.apache.thrift.meta_data.FieldMetaData("updates", org.apache.thrift.TFieldRequirementType.DEFAULT,
new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST,
new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, Column.class))));
tmpMap.put(_Fields.CONSISTENCY_LEVEL, new org.apache.thrift.meta_data.FieldMetaData("consistency_level", org.apache.thrift.TFieldRequirementType.REQUIRED,
tmpMap.put(_Fields.SERIAL_CONSISTENCY_LEVEL, new org.apache.thrift.meta_data.FieldMetaData("serial_consistency_level", org.apache.thrift.TFieldRequirementType.REQUIRED,
new org.apache.thrift.meta_data.EnumMetaData(org.apache.thrift.protocol.TType.ENUM, ConsistencyLevel.class)));
tmpMap.put(_Fields.COMMIT_CONSISTENCY_LEVEL, new org.apache.thrift.meta_data.FieldMetaData("commit_consistency_level", org.apache.thrift.TFieldRequirementType.REQUIRED,
new org.apache.thrift.meta_data.EnumMetaData(org.apache.thrift.protocol.TType.ENUM, ConsistencyLevel.class)));
metaDataMap = Collections.unmodifiableMap(tmpMap);
org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(cas_args.class, metaDataMap);
}
public cas_args() {
this.consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.QUORUM;
this.serial_consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.SERIAL;
this.commit_consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.QUORUM;
}
@ -20215,14 +20246,16 @@ public class Cassandra {
String column_family,
List<Column> expected,
List<Column> updates,
ConsistencyLevel consistency_level)
ConsistencyLevel serial_consistency_level,
ConsistencyLevel commit_consistency_level)
{
this();
this.key = key;
this.column_family = column_family;
this.expected = expected;
this.updates = updates;
this.consistency_level = consistency_level;
this.serial_consistency_level = serial_consistency_level;
this.commit_consistency_level = commit_consistency_level;
}
/**
@ -20250,8 +20283,11 @@ public class Cassandra {
}
this.updates = __this__updates;
}
if (other.isSetConsistency_level()) {
this.consistency_level = other.consistency_level;
if (other.isSetSerial_consistency_level()) {
this.serial_consistency_level = other.serial_consistency_level;
}
if (other.isSetCommit_consistency_level()) {
this.commit_consistency_level = other.commit_consistency_level;
}
}
@ -20265,7 +20301,9 @@ public class Cassandra {
this.column_family = null;
this.expected = null;
this.updates = null;
this.consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.QUORUM;
this.serial_consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.SERIAL;
this.commit_consistency_level = org.apache.cassandra.thrift.ConsistencyLevel.QUORUM;
}
@ -20409,31 +20447,63 @@ public class Cassandra {
*
* @see ConsistencyLevel
*/
public ConsistencyLevel getConsistency_level() {
return this.consistency_level;
public ConsistencyLevel getSerial_consistency_level() {
return this.serial_consistency_level;
}
/**
*
* @see ConsistencyLevel
*/
public cas_args setConsistency_level(ConsistencyLevel consistency_level) {
this.consistency_level = consistency_level;
public cas_args setSerial_consistency_level(ConsistencyLevel serial_consistency_level) {
this.serial_consistency_level = serial_consistency_level;
return this;
}
public void unsetConsistency_level() {
this.consistency_level = null;
public void unsetSerial_consistency_level() {
this.serial_consistency_level = null;
}
/** Returns true if field consistency_level is set (has been assigned a value) and false otherwise */
public boolean isSetConsistency_level() {
return this.consistency_level != null;
/** Returns true if field serial_consistency_level is set (has been assigned a value) and false otherwise */
public boolean isSetSerial_consistency_level() {
return this.serial_consistency_level != null;
}
public void setConsistency_levelIsSet(boolean value) {
public void setSerial_consistency_levelIsSet(boolean value) {
if (!value) {
this.consistency_level = null;
this.serial_consistency_level = null;
}
}
/**
*
* @see ConsistencyLevel
*/
public ConsistencyLevel getCommit_consistency_level() {
return this.commit_consistency_level;
}
/**
*
* @see ConsistencyLevel
*/
public cas_args setCommit_consistency_level(ConsistencyLevel commit_consistency_level) {
this.commit_consistency_level = commit_consistency_level;
return this;
}
public void unsetCommit_consistency_level() {
this.commit_consistency_level = null;
}
/** Returns true if field commit_consistency_level is set (has been assigned a value) and false otherwise */
public boolean isSetCommit_consistency_level() {
return this.commit_consistency_level != null;
}
public void setCommit_consistency_levelIsSet(boolean value) {
if (!value) {
this.commit_consistency_level = null;
}
}
@ -20471,11 +20541,19 @@ public class Cassandra {
}
break;
case CONSISTENCY_LEVEL:
case SERIAL_CONSISTENCY_LEVEL:
if (value == null) {
unsetConsistency_level();
unsetSerial_consistency_level();
} else {
setConsistency_level((ConsistencyLevel)value);
setSerial_consistency_level((ConsistencyLevel)value);
}
break;
case COMMIT_CONSISTENCY_LEVEL:
if (value == null) {
unsetCommit_consistency_level();
} else {
setCommit_consistency_level((ConsistencyLevel)value);
}
break;
@ -20496,8 +20574,11 @@ public class Cassandra {
case UPDATES:
return getUpdates();
case CONSISTENCY_LEVEL:
return getConsistency_level();
case SERIAL_CONSISTENCY_LEVEL:
return getSerial_consistency_level();
case COMMIT_CONSISTENCY_LEVEL:
return getCommit_consistency_level();
}
throw new IllegalStateException();
@ -20518,8 +20599,10 @@ public class Cassandra {
return isSetExpected();
case UPDATES:
return isSetUpdates();
case CONSISTENCY_LEVEL:
return isSetConsistency_level();
case SERIAL_CONSISTENCY_LEVEL:
return isSetSerial_consistency_level();
case COMMIT_CONSISTENCY_LEVEL:
return isSetCommit_consistency_level();
}
throw new IllegalStateException();
}
@ -20573,12 +20656,21 @@ public class Cassandra {
return false;
}
boolean this_present_consistency_level = true && this.isSetConsistency_level();
boolean that_present_consistency_level = true && that.isSetConsistency_level();
if (this_present_consistency_level || that_present_consistency_level) {
if (!(this_present_consistency_level && that_present_consistency_level))
boolean this_present_serial_consistency_level = true && this.isSetSerial_consistency_level();
boolean that_present_serial_consistency_level = true && that.isSetSerial_consistency_level();
if (this_present_serial_consistency_level || that_present_serial_consistency_level) {
if (!(this_present_serial_consistency_level && that_present_serial_consistency_level))
return false;
if (!this.consistency_level.equals(that.consistency_level))
if (!this.serial_consistency_level.equals(that.serial_consistency_level))
return false;
}
boolean this_present_commit_consistency_level = true && this.isSetCommit_consistency_level();
boolean that_present_commit_consistency_level = true && that.isSetCommit_consistency_level();
if (this_present_commit_consistency_level || that_present_commit_consistency_level) {
if (!(this_present_commit_consistency_level && that_present_commit_consistency_level))
return false;
if (!this.commit_consistency_level.equals(that.commit_consistency_level))
return false;
}
@ -20609,10 +20701,15 @@ public class Cassandra {
if (present_updates)
builder.append(updates);
boolean present_consistency_level = true && (isSetConsistency_level());
builder.append(present_consistency_level);
if (present_consistency_level)
builder.append(consistency_level.getValue());
boolean present_serial_consistency_level = true && (isSetSerial_consistency_level());
builder.append(present_serial_consistency_level);
if (present_serial_consistency_level)
builder.append(serial_consistency_level.getValue());
boolean present_commit_consistency_level = true && (isSetCommit_consistency_level());
builder.append(present_commit_consistency_level);
if (present_commit_consistency_level)
builder.append(commit_consistency_level.getValue());
return builder.toHashCode();
}
@ -20665,12 +20762,22 @@ public class Cassandra {
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetConsistency_level()).compareTo(typedOther.isSetConsistency_level());
lastComparison = Boolean.valueOf(isSetSerial_consistency_level()).compareTo(typedOther.isSetSerial_consistency_level());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetConsistency_level()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.consistency_level, typedOther.consistency_level);
if (isSetSerial_consistency_level()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.serial_consistency_level, typedOther.serial_consistency_level);
if (lastComparison != 0) {
return lastComparison;
}
}
lastComparison = Boolean.valueOf(isSetCommit_consistency_level()).compareTo(typedOther.isSetCommit_consistency_level());
if (lastComparison != 0) {
return lastComparison;
}
if (isSetCommit_consistency_level()) {
lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.commit_consistency_level, typedOther.commit_consistency_level);
if (lastComparison != 0) {
return lastComparison;
}
@ -20727,11 +20834,19 @@ public class Cassandra {
}
first = false;
if (!first) sb.append(", ");
sb.append("consistency_level:");
if (this.consistency_level == null) {
sb.append("serial_consistency_level:");
if (this.serial_consistency_level == null) {
sb.append("null");
} else {
sb.append(this.consistency_level);
sb.append(this.serial_consistency_level);
}
first = false;
if (!first) sb.append(", ");
sb.append("commit_consistency_level:");
if (this.commit_consistency_level == null) {
sb.append("null");
} else {
sb.append(this.commit_consistency_level);
}
first = false;
sb.append(")");
@ -20746,8 +20861,11 @@ public class Cassandra {
if (column_family == null) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'column_family' was not present! Struct: " + toString());
}
if (consistency_level == null) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'consistency_level' was not present! Struct: " + toString());
if (serial_consistency_level == null) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'serial_consistency_level' was not present! Struct: " + toString());
}
if (commit_consistency_level == null) {
throw new org.apache.thrift.protocol.TProtocolException("Required field 'commit_consistency_level' was not present! Struct: " + toString());
}
// check for sub-struct validity
}
@ -20840,10 +20958,18 @@ public class Cassandra {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 5: // CONSISTENCY_LEVEL
case 5: // SERIAL_CONSISTENCY_LEVEL
if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
struct.consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setConsistency_levelIsSet(true);
struct.serial_consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setSerial_consistency_levelIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
break;
case 6: // COMMIT_CONSISTENCY_LEVEL
if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
struct.commit_consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setCommit_consistency_levelIsSet(true);
} else {
org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
}
@ -20897,9 +21023,14 @@ public class Cassandra {
}
oprot.writeFieldEnd();
}
if (struct.consistency_level != null) {
oprot.writeFieldBegin(CONSISTENCY_LEVEL_FIELD_DESC);
oprot.writeI32(struct.consistency_level.getValue());
if (struct.serial_consistency_level != null) {
oprot.writeFieldBegin(SERIAL_CONSISTENCY_LEVEL_FIELD_DESC);
oprot.writeI32(struct.serial_consistency_level.getValue());
oprot.writeFieldEnd();
}
if (struct.commit_consistency_level != null) {
oprot.writeFieldBegin(COMMIT_CONSISTENCY_LEVEL_FIELD_DESC);
oprot.writeI32(struct.commit_consistency_level.getValue());
oprot.writeFieldEnd();
}
oprot.writeFieldStop();
@ -20921,7 +21052,8 @@ public class Cassandra {
TTupleProtocol oprot = (TTupleProtocol) prot;
oprot.writeBinary(struct.key);
oprot.writeString(struct.column_family);
oprot.writeI32(struct.consistency_level.getValue());
oprot.writeI32(struct.serial_consistency_level.getValue());
oprot.writeI32(struct.commit_consistency_level.getValue());
BitSet optionals = new BitSet();
if (struct.isSetExpected()) {
optionals.set(0);
@ -20957,8 +21089,10 @@ public class Cassandra {
struct.setKeyIsSet(true);
struct.column_family = iprot.readString();
struct.setColumn_familyIsSet(true);
struct.consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setConsistency_levelIsSet(true);
struct.serial_consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setSerial_consistency_levelIsSet(true);
struct.commit_consistency_level = ConsistencyLevel.findByValue(iprot.readI32());
struct.setCommit_consistency_levelIsSet(true);
BitSet incoming = iprot.readBitSet(2);
if (incoming.get(0)) {
{
@ -24573,7 +24707,7 @@ public class Cassandra {
struct.mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>(2*_map308.size);
for (int _i309 = 0; _i309 < _map308.size; ++_i309)
{
ByteBuffer _key310; // optional
ByteBuffer _key310; // required
Map<String,List<Mutation>> _val311; // required
_key310 = iprot.readBinary();
{
@ -24581,7 +24715,7 @@ public class Cassandra {
_val311 = new HashMap<String,List<Mutation>>(2*_map312.size);
for (int _i313 = 0; _i313 < _map312.size; ++_i313)
{
String _key314; // optional
String _key314; // required
List<Mutation> _val315; // required
_key314 = iprot.readString();
{
@ -24714,7 +24848,7 @@ public class Cassandra {
struct.mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>(2*_map325.size);
for (int _i326 = 0; _i326 < _map325.size; ++_i326)
{
ByteBuffer _key327; // optional
ByteBuffer _key327; // required
Map<String,List<Mutation>> _val328; // required
_key327 = iprot.readBinary();
{
@ -24722,7 +24856,7 @@ public class Cassandra {
_val328 = new HashMap<String,List<Mutation>>(2*_map329.size);
for (int _i330 = 0; _i330 < _map329.size; ++_i330)
{
String _key331; // optional
String _key331; // required
List<Mutation> _val332; // required
_key331 = iprot.readString();
{
@ -25777,7 +25911,7 @@ public class Cassandra {
struct.mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>(2*_map336.size);
for (int _i337 = 0; _i337 < _map336.size; ++_i337)
{
ByteBuffer _key338; // optional
ByteBuffer _key338; // required
Map<String,List<Mutation>> _val339; // required
_key338 = iprot.readBinary();
{
@ -25785,7 +25919,7 @@ public class Cassandra {
_val339 = new HashMap<String,List<Mutation>>(2*_map340.size);
for (int _i341 = 0; _i341 < _map340.size; ++_i341)
{
String _key342; // optional
String _key342; // required
List<Mutation> _val343; // required
_key342 = iprot.readString();
{
@ -25918,7 +26052,7 @@ public class Cassandra {
struct.mutation_map = new HashMap<ByteBuffer,Map<String,List<Mutation>>>(2*_map353.size);
for (int _i354 = 0; _i354 < _map353.size; ++_i354)
{
ByteBuffer _key355; // optional
ByteBuffer _key355; // required
Map<String,List<Mutation>> _val356; // required
_key355 = iprot.readBinary();
{
@ -25926,7 +26060,7 @@ public class Cassandra {
_val356 = new HashMap<String,List<Mutation>>(2*_map357.size);
for (int _i358 = 0; _i358 < _map357.size; ++_i358)
{
String _key359; // optional
String _key359; // required
List<Mutation> _val360; // required
_key359 = iprot.readString();
{
@ -28121,7 +28255,7 @@ public class Cassandra {
struct.success = new HashMap<String,List<String>>(2*_map364.size);
for (int _i365 = 0; _i365 < _map364.size; ++_i365)
{
String _key366; // optional
String _key366; // required
List<String> _val367; // required
_key366 = iprot.readString();
{
@ -28249,7 +28383,7 @@ public class Cassandra {
struct.success = new HashMap<String,List<String>>(2*_map375.size);
for (int _i376 = 0; _i376 < _map375.size; ++_i376)
{
String _key377; // optional
String _key377; // required
List<String> _val378; // required
_key377 = iprot.readString();
{
@ -31796,7 +31930,7 @@ public class Cassandra {
struct.success = new HashMap<String,String>(2*_map398.size);
for (int _i399 = 0; _i399 < _map398.size; ++_i399)
{
String _key400; // optional
String _key400; // required
String _val401; // required
_key400 = iprot.readString();
_val401 = iprot.readString();
@ -31901,7 +32035,7 @@ public class Cassandra {
struct.success = new HashMap<String,String>(2*_map404.size);
for (int _i405 = 0; _i405 < _map404.size; ++_i405)
{
String _key406; // optional
String _key406; // required
String _val407; // required
_key406 = iprot.readString();
_val407 = iprot.readString();

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

@ -3674,7 +3674,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.compaction_strategy_options = new HashMap<String,String>(2*_map113.size);
for (int _i114 = 0; _i114 < _map113.size; ++_i114)
{
String _key115; // optional
String _key115; // required
String _val116; // required
_key115 = iprot.readString();
_val116 = iprot.readString();
@ -3694,7 +3694,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.compression_options = new HashMap<String,String>(2*_map117.size);
for (int _i118 = 0; _i118 < _map117.size; ++_i118)
{
String _key119; // optional
String _key119; // required
String _val120; // required
_key119 = iprot.readString();
_val120 = iprot.readString();
@ -4484,7 +4484,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.compaction_strategy_options = new HashMap<String,String>(2*_map135.size);
for (int _i136 = 0; _i136 < _map135.size; ++_i136)
{
String _key137; // optional
String _key137; // required
String _val138; // required
_key137 = iprot.readString();
_val138 = iprot.readString();
@ -4499,7 +4499,7 @@ public class CfDef implements org.apache.thrift.TBase<CfDef, CfDef._Fields>, jav
struct.compression_options = new HashMap<String,String>(2*_map139.size);
for (int _i140 = 0; _i140 < _map139.size; ++_i140)
{
String _key141; // optional
String _key141; // required
String _val142; // required
_key141 = iprot.readString();
_val142 = iprot.readString();

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

@ -771,7 +771,7 @@ public class ColumnDef implements org.apache.thrift.TBase<ColumnDef, ColumnDef._
struct.index_options = new HashMap<String,String>(2*_map90.size);
for (int _i91 = 0; _i91 < _map90.size; ++_i91)
{
String _key92; // optional
String _key92; // required
String _val93; // required
_key92 = iprot.readString();
_val93 = iprot.readString();
@ -908,7 +908,7 @@ public class ColumnDef implements org.apache.thrift.TBase<ColumnDef, ColumnDef._
struct.index_options = new HashMap<String,String>(2*_map96.size);
for (int _i97 = 0; _i97 < _map96.size; ++_i97)
{
String _key98; // optional
String _key98; // required
String _val99; // required
_key98 = iprot.readString();
_val99 = iprot.readString();

5
interface/thrift/gen-java/org/apache/cassandra/thrift/ConsistencyLevel.java
View File

@ -83,7 +83,8 @@ public enum ConsistencyLevel implements org.apache.thrift.TEnum {
ANY(6),
TWO(7),
THREE(8),
SERIAL(9);
SERIAL(9),
LOCAL_SERIAL(10);
private final int value;
@ -122,6 +123,8 @@ public enum ConsistencyLevel implements org.apache.thrift.TEnum {
return THREE;
case 9:
return SERIAL;
case 10:
return LOCAL_SERIAL;
default:
return null;
}

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

@ -662,7 +662,7 @@ public class CqlMetadata implements org.apache.thrift.TBase<CqlMetadata, CqlMeta
struct.name_types = new HashMap<ByteBuffer,String>(2*_map172.size);
for (int _i173 = 0; _i173 < _map172.size; ++_i173)
{
ByteBuffer _key174; // optional
ByteBuffer _key174; // required
String _val175; // required
_key174 = iprot.readBinary();
_val175 = iprot.readString();
@ -682,7 +682,7 @@ public class CqlMetadata implements org.apache.thrift.TBase<CqlMetadata, CqlMeta
struct.value_types = new HashMap<ByteBuffer,String>(2*_map176.size);
for (int _i177 = 0; _i177 < _map176.size; ++_i177)
{
ByteBuffer _key178; // optional
ByteBuffer _key178; // required
String _val179; // required
_key178 = iprot.readBinary();
_val179 = iprot.readString();
@ -807,7 +807,7 @@ public class CqlMetadata implements org.apache.thrift.TBase<CqlMetadata, CqlMeta
struct.name_types = new HashMap<ByteBuffer,String>(2*_map184.size);
for (int _i185 = 0; _i185 < _map184.size; ++_i185)
{
ByteBuffer _key186; // optional
ByteBuffer _key186; // required
String _val187; // required
_key186 = iprot.readBinary();
_val187 = iprot.readString();
@ -820,7 +820,7 @@ public class CqlMetadata implements org.apache.thrift.TBase<CqlMetadata, CqlMeta
struct.value_types = new HashMap<ByteBuffer,String>(2*_map188.size);
for (int _i189 = 0; _i189 < _map188.size; ++_i189)
{
ByteBuffer _key190; // optional
ByteBuffer _key190; // required
String _val191; // required
_key190 = iprot.readBinary();
_val191 = iprot.readString();

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

@ -841,7 +841,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
struct.strategy_options = new HashMap<String,String>(2*_map146.size);
for (int _i147 = 0; _i147 < _map146.size; ++_i147)
{
String _key148; // optional
String _key148; // required
String _val149; // required
_key148 = iprot.readString();
_val149 = iprot.readString();
@ -1032,7 +1032,7 @@ public class KsDef implements org.apache.thrift.TBase<KsDef, KsDef._Fields>, jav
struct.strategy_options = new HashMap<String,String>(2*_map160.size);
for (int _i161 = 0; _i161 < _map160.size; ++_i161)
{
String _key162; // optional
String _key162; // required
String _val163; // required
_key162 = iprot.readString();
_val163 = iprot.readString();

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

@ -478,7 +478,7 @@ public class TriggerDef implements org.apache.thrift.TBase<TriggerDef, TriggerDe
struct.options = new HashMap<String,String>(2*_map100.size);
for (int _i101 = 0; _i101 < _map100.size; ++_i101)
{
String _key102; // optional
String _key102; // required
String _val103; // required
_key102 = iprot.readString();
_val103 = iprot.readString();
@ -562,7 +562,7 @@ public class TriggerDef implements org.apache.thrift.TBase<TriggerDef, TriggerDe
struct.options = new HashMap<String,String>(2*_map106.size);
for (int _i107 = 0; _i107 < _map106.size; ++_i107)
{
String _key108; // optional
String _key108; // required
String _val109; // required
_key108 = iprot.readString();
_val109 = iprot.readString();

8
src/java/org/apache/cassandra/auth/Auth.java
View File

@ -30,6 +30,7 @@ import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.RequestExecutionException;
@ -230,11 +231,8 @@ public class Auth
{
try
{
ResultMessage.Rows rows = selectUserStatement.execute(consistencyForUser(username),
new QueryState(new ClientState(true)),
Lists.newArrayList(ByteBufferUtil.bytes(username)),
-1,
null);
ResultMessage.Rows rows = selectUserStatement.execute(new QueryState(new ClientState(true)),
new QueryOptions(consistencyForUser(username), Lists.newArrayList(ByteBufferUtil.bytes(username))));
return new UntypedResultSet(rows.result);
}
catch (RequestValidationException e)

11
src/java/org/apache/cassandra/auth/CassandraAuthorizer.java
View File

@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.marshal.UTF8Type;
@ -71,12 +72,10 @@ public class CassandraAuthorizer implements IAuthorizer
UntypedResultSet result;
try
{
ResultMessage.Rows rows = authorizeStatement.execute(ConsistencyLevel.ONE,
new QueryState(new ClientState(true)),
Lists.newArrayList(ByteBufferUtil.bytes(user.getName()),
ByteBufferUtil.bytes(resource.getName())),
-1,
null);
ResultMessage.Rows rows = authorizeStatement.execute(new QueryState(new ClientState(true)),
new QueryOptions(ConsistencyLevel.ONE,
Lists.newArrayList(ByteBufferUtil.bytes(user.getName()),
ByteBufferUtil.bytes(resource.getName()))));
result = new UntypedResultSet(rows.result);
}
catch (RequestValidationException e)

9
src/java/org/apache/cassandra/auth/PasswordAuthenticator.java
View File

@ -34,6 +34,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.*;
@ -106,11 +107,9 @@ public class PasswordAuthenticator implements ISaslAwareAuthenticator
UntypedResultSet result;
try
{
ResultMessage.Rows rows = authenticateStatement.execute(consistencyForUser(username),
new QueryState(new ClientState(true)),
Lists.newArrayList(ByteBufferUtil.bytes(username)),
-1,
null);
ResultMessage.Rows rows = authenticateStatement.execute(new QueryState(new ClientState(true)),
new QueryOptions(consistencyForUser(username),
Lists.newArrayList(ByteBufferUtil.bytes(username))));
result = new UntypedResultSet(rows.result);
}
catch (RequestValidationException e)

11
src/java/org/apache/cassandra/cql3/CQLStatement.java
View File

@ -24,7 +24,6 @@ import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.pager.PagingState;
import org.apache.cassandra.exceptions.*;
public interface CQLStatement
@ -52,16 +51,10 @@ public interface CQLStatement
/**
* Execute the statement and return the resulting result or null if there is no result.
*
* @param cl the consistency level for the query
* @param state the current query state
* @param variables the values for bounded variables. The implementation
* can assume that each bound term have a corresponding value.
* @param pageSize the initial page size for the result set potentially returned. A negative value
* means no paging needs to be done. Statements that do not return result sets can ignore this value.
* @param pageState the paging state for paged query. All statement except Select should ignore
* that value.
* @param options options for this query (consistency, variables, pageSize, ...)
*/
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pageState) throws RequestValidationException, RequestExecutionException;
public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException, RequestExecutionException;
/**
* Variante of execute used for internal query against the system tables, and thus only query the local node.

237
src/java/org/apache/cassandra/cql3/QueryOptions.java Normal file
View File

@ -0,0 +1,237 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.cql3;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import org.jboss.netty.buffer.ChannelBuffer;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.service.pager.PagingState;
import org.apache.cassandra.transport.CBCodec;
import org.apache.cassandra.transport.CBUtil;
/**
* Options for a query.
*/
public class QueryOptions
{
public static final QueryOptions DEFAULT = new QueryOptions(ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList());
public static final CBCodec<QueryOptions> codec = new Codec();
private final ConsistencyLevel consistency;
private final List<ByteBuffer> values;
private final boolean skipMetadata;
private final SpecificOptions options;
public QueryOptions(ConsistencyLevel consistency, List<ByteBuffer> values)
{
this(consistency, values, false, SpecificOptions.DEFAULT);
}
public QueryOptions(ConsistencyLevel consistency,
List<ByteBuffer> values,
boolean skipMetadata,
int pageSize,
PagingState pagingState,
ConsistencyLevel serialConsistency)
{
this(consistency, values, skipMetadata, new SpecificOptions(pageSize, pagingState, serialConsistency));
}
private QueryOptions(ConsistencyLevel consistency, List<ByteBuffer> values, boolean skipMetadata, SpecificOptions options)
{
this.consistency = consistency;
this.values = values;
this.skipMetadata = skipMetadata;
this.options = options;
}
public ConsistencyLevel getConsistency()
{
return consistency;
}
public List<ByteBuffer> getValues()
{
return values;
}
public boolean skipMetadata()
{
return skipMetadata;
}
/**
* The pageSize for this query. Will be <= 0 if not relevant for the query.
*/
public int getPageSize()
{
return options.pageSize;
}
/**
* The paging state for this query, or null if not relevant.
*/
public PagingState getPagingState()
{
return options.state;
}
/**
* Serial consistency for conditional updates.
*/
public ConsistencyLevel getSerialConsistency()
{
return options.serialConsistency;
}
// Options that are likely to not be present in most queries
private static class SpecificOptions
{
private static final SpecificOptions DEFAULT = new SpecificOptions(-1, null, null);
private final int pageSize;
private final PagingState state;
private final ConsistencyLevel serialConsistency;
private SpecificOptions(int pageSize, PagingState state, ConsistencyLevel serialConsistency)
{
this.pageSize = pageSize;
this.state = state;
this.serialConsistency = serialConsistency == null ? ConsistencyLevel.SERIAL : serialConsistency;
}
}
private static class Codec implements CBCodec<QueryOptions>
{
private static enum Flag
{
// The order of that enum matters!!
VALUES,
SKIP_METADATA,
PAGE_SIZE,
PAGING_STATE,
SERIAL_CONSISTENCY;
public static EnumSet<Flag> deserialize(int flags)
{
EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
Flag[] values = Flag.values();
for (int n = 0; n < values.length; n++)
{
if ((flags & (1 << n)) != 0)
set.add(values[n]);
}
return set;
}
public static int serialize(EnumSet<Flag> flags)
{
int i = 0;
for (Flag flag : flags)
i |= 1 << flag.ordinal();
return i;
}
}
public QueryOptions decode(ChannelBuffer body, int version)
{
assert version >= 2;
ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
List<ByteBuffer> values = Collections.emptyList();
if (flags.contains(Flag.VALUES))
{
int paramCount = body.readUnsignedShort();
if (paramCount > 0)
{
values = new ArrayList<ByteBuffer>(paramCount);
for (int i = 0; i < paramCount; i++)
values.add(CBUtil.readValue(body));
}
}
boolean skipMetadata = flags.contains(Flag.SKIP_METADATA);
flags.remove(Flag.VALUES);
flags.remove(Flag.SKIP_METADATA);
SpecificOptions options = SpecificOptions.DEFAULT;
if (!flags.isEmpty())
{
int pageSize = flags.contains(Flag.PAGE_SIZE) ? body.readInt() : -1;
PagingState pagingState = flags.contains(Flag.PAGING_STATE) ? PagingState.deserialize(CBUtil.readValue(body)) : null;
ConsistencyLevel serialConsistency = flags.contains(Flag.SERIAL_CONSISTENCY) ? CBUtil.readConsistencyLevel(body) : ConsistencyLevel.SERIAL;
options = new SpecificOptions(pageSize, pagingState, serialConsistency);
}
return new QueryOptions(consistency, values, skipMetadata, options);
}
public ChannelBuffer encode(QueryOptions options, int version)
{
assert version >= 2;
int nbBuff = 2;
EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
if (options.getValues().size() > 0)
{
flags.add(Flag.VALUES);
nbBuff++;
}
if (options.skipMetadata)
flags.add(Flag.SKIP_METADATA);
if (options.getPageSize() >= 0)
{
flags.add(Flag.PAGE_SIZE);
nbBuff++;
}
if (options.getSerialConsistency() != ConsistencyLevel.SERIAL)
{
flags.add(Flag.SERIAL_CONSISTENCY);
nbBuff++;
}
CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, options.values.size() + (flags.contains(Flag.PAGING_STATE) ? 1 : 0));
builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
builder.add(CBUtil.consistencyLevelToCB(options.getConsistency()));
if (flags.contains(Flag.VALUES))
{
builder.add(CBUtil.shortToCB(options.getValues().size()));
for (ByteBuffer value : options.getValues())
builder.addValue(value);
}
if (flags.contains(Flag.PAGE_SIZE))
builder.add(CBUtil.intToCB(options.getPageSize()));
if (flags.contains(Flag.PAGING_STATE))
builder.addValue(options.getPagingState().serialize());
if (flags.contains(Flag.SERIAL_CONSISTENCY))
builder.add(CBUtil.consistencyLevelToCB(options.getSerialConsistency()));
return builder.build();
}
}
}

21
src/java/org/apache/cassandra/cql3/QueryProcessor.java
View File

@ -94,30 +94,30 @@ public class QueryProcessor
}
}
private static ResultMessage processStatement(CQLStatement statement, ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables, int pageSize, PagingState pageState)
private static ResultMessage processStatement(CQLStatement statement, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
logger.trace("Process {} @CL.{}", statement, cl);
logger.trace("Process {} @CL.{}", statement, options.getConsistency());
ClientState clientState = queryState.getClientState();
statement.checkAccess(clientState);
statement.validate(clientState);
ResultMessage result = statement.execute(cl, queryState, variables, pageSize, pageState);
ResultMessage result = statement.execute(queryState, options);
return result == null ? new ResultMessage.Void() : result;
}
public static ResultMessage process(String queryString, ConsistencyLevel cl, QueryState queryState)
throws RequestExecutionException, RequestValidationException
{
return process(queryString, Collections.<ByteBuffer>emptyList(), cl, queryState, -1, null);
return process(queryString, queryState, new QueryOptions(cl, Collections.<ByteBuffer>emptyList()));
}
public static ResultMessage process(String queryString, List<ByteBuffer> variables, ConsistencyLevel cl, QueryState queryState, int pageSize, PagingState pageState)
public static ResultMessage process(String queryString, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
CQLStatement prepared = getStatement(queryString, queryState.getClientState()).statement;
if (prepared.getBoundsTerms() != variables.size())
if (prepared.getBoundsTerms() != options.getValues().size())
throw new InvalidRequestException("Invalid amount of bind variables");
return processStatement(prepared, cl, queryState, variables, pageSize, pageState);
return processStatement(prepared, queryState, options);
}
public static CQLStatement parseStatement(String queryStr, QueryState queryState) throws RequestValidationException
@ -130,7 +130,7 @@ public class QueryProcessor
try
{
QueryState state = new QueryState(new ClientState(true));
ResultMessage result = process(query, cl, state);
ResultMessage result = process(query, state, new QueryOptions(cl, Collections.<ByteBuffer>emptyList()));
if (result instanceof ResultMessage.Rows)
return new UntypedResultSet(((ResultMessage.Rows)result).result);
else
@ -216,9 +216,10 @@ public class QueryProcessor
}
}
public static ResultMessage processPrepared(CQLStatement statement, ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables, int pageSize, PagingState pageState)
public static ResultMessage processPrepared(CQLStatement statement, QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
List<ByteBuffer> variables = options.getValues();
// Check to see if there are any bound variables to verify
if (!(variables.isEmpty() && (statement.getBoundsTerms() == 0)))
{
@ -234,7 +235,7 @@ public class QueryProcessor
logger.trace("[{}] '{}'", i+1, variables.get(i));
}
return processStatement(statement, cl, queryState, variables, pageSize, pageState);
return processStatement(statement, queryState, options);
}
public static ResultMessage processBatch(BatchStatement batch, ConsistencyLevel cl, QueryState queryState, List<List<ByteBuffer>> variables)

3
src/java/org/apache/cassandra/cql3/statements/AuthenticationStatement.java
View File

@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.ClientState;
@ -41,7 +42,7 @@ public abstract class AuthenticationStatement extends ParsedStatement implements
return 0;
}
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState)
public ResultMessage execute(QueryState state, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
return execute(state.getClientState());

3
src/java/org/apache/cassandra/cql3/statements/AuthorizationStatement.java
View File

@ -22,6 +22,7 @@ import java.util.List;
import org.apache.cassandra.auth.DataResource;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.ClientState;
@ -42,7 +43,7 @@ public abstract class AuthorizationStatement extends ParsedStatement implements
return 0;
}
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState)
public ResultMessage execute(QueryState state, QueryOptions options)
throws RequestValidationException, RequestExecutionException
{
return execute(state.getClientState());

6
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -134,12 +134,12 @@ public class BatchStatement implements CQLStatement
}
}
public ResultMessage execute(ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables, int pageSize, PagingState pagingState) throws RequestExecutionException, RequestValidationException
public ResultMessage execute(QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException
{
if (cl == null)
if (options.getConsistency() == null)
throw new InvalidRequestException("Invalid empty consistency level");
execute(getMutations(variables, false, cl, queryState.getTimestamp()), cl);
execute(getMutations(options.getValues(), false, options.getConsistency(), queryState.getTimestamp()), options.getConsistency());
return null;
}

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

@ -334,32 +334,34 @@ public abstract class ModificationStatement implements CQLStatement
return ifNotExists || (columnConditions != null && !columnConditions.isEmpty());
}
public ResultMessage execute(ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables, int pageSize, PagingState pagingState)
public ResultMessage execute(QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
if (cl == null)
if (options.getConsistency() == null)
throw new InvalidRequestException("Invalid empty consistency level");
return hasConditions()
? executeWithCondition(cl, queryState, variables)
: executeWithoutCondition(cl, queryState, variables);
? executeWithCondition(queryState, options)
: executeWithoutCondition(queryState, options);
}
private ResultMessage executeWithoutCondition(ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables)
private ResultMessage executeWithoutCondition(QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
ConsistencyLevel cl = options.getConsistency();
if (isCounter())
cl.validateCounterForWrite(cfm);
else
cl.validateForWrite(cfm.ksName);
StorageProxy.mutateWithTriggers(getMutations(variables, false, cl, queryState.getTimestamp(), false), cl, false);
StorageProxy.mutateWithTriggers(getMutations(options.getValues(), false, cl, queryState.getTimestamp(), false), cl, false);
return null;
}
public ResultMessage executeWithCondition(ConsistencyLevel cl, QueryState queryState, List<ByteBuffer> variables)
public ResultMessage executeWithCondition(QueryState queryState, QueryOptions options)
throws RequestExecutionException, RequestValidationException
{
List<ByteBuffer> variables = options.getValues();
List<ByteBuffer> keys = buildPartitionKeyNames(variables);
// We don't support IN for CAS operation so far
if (keys.size() > 1)
@ -374,7 +376,7 @@ public abstract class ModificationStatement implements CQLStatement
ColumnFamily updates = updateForKey(key, clusteringPrefix, params);
ColumnFamily expected = buildConditions(key, clusteringPrefix, params);
ColumnFamily result = StorageProxy.cas(keyspace(), columnFamily(), key, clusteringPrefix, expected, updates, cl);
ColumnFamily result = StorageProxy.cas(keyspace(), columnFamily(), key, clusteringPrefix, expected, updates, options.getConsistency(), options.getSerialConsistency());
return new ResultMessage.Rows(buildCasResultSet(key, result));
}

3
src/java/org/apache/cassandra/cql3/statements/SchemaAlteringStatement.java
View File

@ -22,6 +22,7 @@ import java.util.List;
import org.apache.cassandra.cql3.CFName;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.ClientState;
@ -65,7 +66,7 @@ public abstract class SchemaAlteringStatement extends CFStatement implements CQL
public abstract void announceMigration() throws RequestValidationException;
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState) throws RequestValidationException
public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException
{
announceMigration();
String tableName = cfName == null || columnFamily() == null ? "" : columnFamily();

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

@ -138,8 +138,10 @@ public class SelectStatement implements CQLStatement
// Nothing to do, all validation has been done by RawStatement.prepare()
}
public ResultMessage.Rows execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState) throws RequestExecutionException, RequestValidationException
public ResultMessage.Rows execute(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException
{
ConsistencyLevel cl = options.getConsistency();
List<ByteBuffer> variables = options.getValues();
if (cl == null)
throw new InvalidRequestException("Invalid empty consistency level");
@ -158,18 +160,19 @@ public class SelectStatement implements CQLStatement
command = commands == null ? null : new Pageable.ReadCommands(commands);
}
int pageSize = options.getPageSize();
// A count query will never be paged for the user, but we always page it internally to avoid OOM.
// If we user provided a pageSize we'll use that to page internally (because why not), otherwise we use our default
if (parameters.isCount && pageSize < 0)
if (parameters.isCount && pageSize <= 0)
pageSize = DEFAULT_COUNT_PAGE_SIZE;
if (pageSize < 0 || command == null || !QueryPagers.mayNeedPaging(command, pageSize))
if (pageSize <= 0 || command == null || !QueryPagers.mayNeedPaging(command, pageSize))
{
return execute(command, cl, variables, limit, now);
}
else
{
QueryPager pager = QueryPagers.pager(command, cl, pagingState);
QueryPager pager = QueryPagers.pager(command, cl, options.getPagingState());
if (parameters.isCount)
return pageCountQuery(pager, variables, pageSize, now);

2
src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java
View File

@ -55,7 +55,7 @@ public class TruncateStatement extends CFStatement implements CQLStatement
ThriftValidation.validateColumnFamily(keyspace(), columnFamily());
}
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState) throws InvalidRequestException, TruncateException
public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException, TruncateException
{
try
{

3
src/java/org/apache/cassandra/cql3/statements/UseStatement.java
View File

@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.UnauthorizedException;
@ -52,7 +53,7 @@ public class UseStatement extends ParsedStatement implements CQLStatement
{
}
public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer> variables, int pageSize, PagingState pagingState) throws InvalidRequestException
public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException
{
state.getClientState().setKeyspace(keyspace);
return new ResultMessage.SetKeyspace(keyspace);

25
src/java/org/apache/cassandra/db/ConsistencyLevel.java
View File

@ -48,7 +48,8 @@ public enum ConsistencyLevel
ALL (5),
LOCAL_QUORUM(6),
EACH_QUORUM (7),
SERIAL (8);
SERIAL (8),
LOCAL_SERIAL(9);
private static final Logger logger = LoggerFactory.getLogger(ConsistencyLevel.class);
@ -277,11 +278,13 @@ public enum ConsistencyLevel
requireNetworkTopologyStrategy(keyspaceName);
break;
case SERIAL:
case LOCAL_SERIAL:
throw new InvalidRequestException("You must use conditional updates for serializable writes");
}
}
public void validateForCas(String keyspaceName) throws InvalidRequestException
// This is the same than validateForWrite really, but we include a slightly different error message for SERIAL/LOCAL_SERIAL
public void validateForCasCommit(String keyspaceName) throws InvalidRequestException
{
switch (this)
{
@ -289,11 +292,23 @@ public enum ConsistencyLevel
case EACH_QUORUM:
requireNetworkTopologyStrategy(keyspaceName);
break;
case ANY:
throw new InvalidRequestException("ANY is not supported with CAS. Use SERIAL if you mean, make sure it is accepted but I don't care how many replicas commit it for non-SERIAL reads");
case SERIAL:
case LOCAL_SERIAL:
throw new InvalidRequestException(this + " is not supported as conditional update commit consistency. Use ANY if you mean \"make sure it is accepted but I don't care how many replicas commit it for non-SERIAL reads\"");
}
}
public void validateForCas() throws InvalidRequestException
{
if (!isSerialConsistency())
throw new InvalidRequestException("Invalid consistency for conditional update. Must be one of SERIAL or LOCAL_SERIAL");
}
public boolean isSerialConsistency()
{
return this == SERIAL || this == LOCAL_SERIAL;
}
public void validateCounterForWrite(CFMetaData metadata) throws InvalidRequestException
{
if (this == ConsistencyLevel.ANY)
@ -304,7 +319,7 @@ public enum ConsistencyLevel
{
throw new InvalidRequestException("cannot achieve CL > CL.ONE without replicate_on_write on columnfamily " + metadata.cfName);
}
else if (this == ConsistencyLevel.SERIAL)
else if (isSerialConsistency())
{
throw new InvalidRequestException("Counter operations are inherently non-serializable");
}

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

@ -29,6 +29,7 @@ import javax.management.MBeanServer;
import javax.management.ObjectName;
import com.google.common.base.Function;
import com.google.common.base.Predicate;
import com.google.common.collect.*;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.lang.StringUtils;
@ -195,16 +196,25 @@ public class StorageProxy implements StorageProxyMBean
* 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.
* @param consistencyForPaxos the consistency for the paxos prepare and propose round. This can only be either SERIAL or LOCAL_SERIAL.
* @param consistencyForCommit the consistency for write done during the commit phase. This can be anything, except SERIAL or LOCAL_SERIAL.
*
* @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 ColumnFamily cas(String keyspaceName, String cfName, ByteBuffer key, ColumnNameBuilder prefix, ColumnFamily expected, ColumnFamily updates, ConsistencyLevel consistencyLevel)
public static ColumnFamily cas(String keyspaceName,
String cfName,
ByteBuffer key,
ColumnNameBuilder prefix,
ColumnFamily expected,
ColumnFamily updates,
ConsistencyLevel consistencyForPaxos,
ConsistencyLevel consistencyForCommit)
throws UnavailableException, IsBootstrappingException, ReadTimeoutException, WriteTimeoutException, InvalidRequestException
{
consistencyLevel.validateForCas(keyspaceName);
consistencyForPaxos.validateForCas();
consistencyForCommit.validateForCasCommit(keyspaceName);
CFMetaData metadata = Schema.instance.getCFMetaData(keyspaceName, cfName);
@ -213,11 +223,11 @@ public class StorageProxy implements StorageProxyMBean
while (System.nanoTime() - start < timeout)
{
// for simplicity, we'll do a single liveness check at the start of each attempt
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(keyspaceName, key);
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(keyspaceName, key, consistencyForPaxos);
List<InetAddress> liveEndpoints = p.left;
int requiredParticipants = p.right;
UUID ballot = beginAndRepairPaxos(start, key, metadata, liveEndpoints, requiredParticipants);
UUID ballot = beginAndRepairPaxos(start, key, metadata, liveEndpoints, requiredParticipants, consistencyForPaxos);
// read the current value and compare with expected
Tracing.trace("Reading existing values for CAS precondition");
@ -250,10 +260,10 @@ public class StorageProxy implements StorageProxyMBean
Tracing.trace("CAS precondition is met; proposing client-requested updates for {}", ballot);
if (proposePaxos(proposal, liveEndpoints, requiredParticipants))
{
if (consistencyLevel == ConsistencyLevel.SERIAL)
if (consistencyForCommit == ConsistencyLevel.ANY)
sendCommit(proposal, liveEndpoints);
else
commitPaxos(proposal, consistencyLevel);
commitPaxos(proposal, consistencyForCommit);
Tracing.trace("CAS successful");
return null;
}
@ -263,7 +273,7 @@ public class StorageProxy implements StorageProxyMBean
// continue to retry
}
throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, -1, -1);
throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, -1, -1);
}
private static boolean hasLiveColumns(ColumnFamily cf, long now)
@ -301,15 +311,35 @@ public class StorageProxy implements StorageProxyMBean
return true;
}
private static Pair<List<InetAddress>, Integer> getPaxosParticipants(String keyspaceName, ByteBuffer key) throws UnavailableException
private static Predicate<InetAddress> sameDCPredicateFor(final String dc)
{
final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
return new Predicate<InetAddress>()
{
public boolean apply(InetAddress host)
{
return dc.equals(snitch.getDatacenter(host));
}
};
}
private static Pair<List<InetAddress>, Integer> getPaxosParticipants(String keyspaceName, ByteBuffer key, ConsistencyLevel consistencyForPaxos) throws UnavailableException
{
Token tk = StorageService.getPartitioner().getToken(key);
List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk);
Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk, keyspaceName);
if (consistencyForPaxos == ConsistencyLevel.LOCAL_SERIAL)
{
// Restrict naturalEndpoints and pendingEndpoints to node in the local DC only
String localDc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
Predicate<InetAddress> isLocalDc = sameDCPredicateFor(localDc);
naturalEndpoints = ImmutableList.copyOf(Iterables.filter(naturalEndpoints, isLocalDc));
pendingEndpoints = ImmutableList.copyOf(Iterables.filter(pendingEndpoints, isLocalDc));
}
int requiredParticipants = pendingEndpoints.size() + 1 + naturalEndpoints.size() / 2; // See CASSANDRA-833
List<InetAddress> liveEndpoints = ImmutableList.copyOf(Iterables.filter(Iterables.concat(naturalEndpoints, pendingEndpoints), IAsyncCallback.isAlive));
if (liveEndpoints.size() < requiredParticipants)
throw new UnavailableException(ConsistencyLevel.SERIAL, requiredParticipants, liveEndpoints.size());
throw new UnavailableException(consistencyForPaxos, requiredParticipants, liveEndpoints.size());
return Pair.create(liveEndpoints, requiredParticipants);
}
@ -319,7 +349,7 @@ public class StorageProxy implements StorageProxyMBean
* @return the Paxos ballot promised by the replicas if no in-progress requests were seen and a quorum of
* nodes have seen the mostRecentCommit. Otherwise, return null.
*/
private static UUID beginAndRepairPaxos(long start, ByteBuffer key, CFMetaData metadata, List<InetAddress> liveEndpoints, int requiredParticipants)
private static UUID beginAndRepairPaxos(long start, ByteBuffer key, CFMetaData metadata, List<InetAddress> liveEndpoints, int requiredParticipants, ConsistencyLevel consistencyForPaxos)
throws WriteTimeoutException
{
long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getCasContentionTimeout());
@ -376,7 +406,7 @@ public class StorageProxy implements StorageProxyMBean
return ballot;
}
throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, -1, -1);
throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, -1, -1);
}
/**
@ -1079,26 +1109,26 @@ public class StorageProxy implements StorageProxyMBean
List<Row> rows = null;
try
{
if (consistency_level == ConsistencyLevel.SERIAL)
if (consistency_level.isSerialConsistency())
{
// make sure any in-progress paxos writes are done (i.e., committed to a majority of replicas), before performing a quorum read
if (commands.size() > 1)
throw new InvalidRequestException("SERIAL consistency may only be requested for one row at a time");
throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may only be requested for one row at a time");
ReadCommand command = commands.get(0);
CFMetaData metadata = Schema.instance.getCFMetaData(command.ksName, command.cfName);
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(command.ksName, command.key);
Pair<List<InetAddress>, Integer> p = getPaxosParticipants(command.ksName, command.key, consistency_level);
List<InetAddress> liveEndpoints = p.left;
int requiredParticipants = p.right;
// does the work of applying in-progress writes; throws UAE or timeout if it can't
try
{
beginAndRepairPaxos(start, command.key, metadata, liveEndpoints, requiredParticipants);
beginAndRepairPaxos(start, command.key, metadata, liveEndpoints, requiredParticipants, consistency_level);
}
catch (WriteTimeoutException e)
{
throw new ReadTimeoutException(ConsistencyLevel.SERIAL, -1, -1, false);
throw new ReadTimeoutException(consistency_level, -1, -1, false);
}
rows = fetchRows(commands, ConsistencyLevel.QUORUM);

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

@ -43,6 +43,7 @@ import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql.CQLStatement;
import org.apache.cassandra.cql.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.IDiskAtomFilter;
@ -691,7 +692,12 @@ public class CassandraServer implements Cassandra.Iface
}
}
public CASResult cas(ByteBuffer key, String column_family, List<Column> expected, List<Column> updates, ConsistencyLevel consistency_level)
public CASResult cas(ByteBuffer key,
String column_family,
List<Column> expected,
List<Column> updates,
ConsistencyLevel serial_consistency_level,
ConsistencyLevel commit_consistency_level)
throws InvalidRequestException, UnavailableException, TimedOutException
{
if (startSessionIfRequested())
@ -747,7 +753,14 @@ public class CassandraServer implements Cassandra.Iface
}
schedule(DatabaseDescriptor.getWriteRpcTimeout());
ColumnFamily result = StorageProxy.cas(cState.getKeyspace(), column_family, key, null, cfExpected, cfUpdates, ThriftConversion.fromThrift(consistency_level));
ColumnFamily result = StorageProxy.cas(cState.getKeyspace(),
column_family,
key,
null,
cfExpected,
cfUpdates,
ThriftConversion.fromThrift(serial_consistency_level),
ThriftConversion.fromThrift(commit_consistency_level));
return result == null
? new CASResult(true)
: new CASResult(false).setCurrent_values(thriftifyColumnsAsColumns(result.getSortedColumns(), System.currentTimeMillis()));
@ -2024,11 +2037,8 @@ public class CassandraServer implements Cassandra.Iface
logger.trace("Retrieved prepared statement #{} with {} bind markers", itemId, statement.getBoundsTerms());
return org.apache.cassandra.cql3.QueryProcessor.processPrepared(statement,
ThriftConversion.fromThrift(cLevel),
cState.getQueryState(),
bindVariables,
-1,
null).toThriftResult();
new QueryOptions(ThriftConversion.fromThrift(cLevel), bindVariables)).toThriftResult();
}
catch (RequestExecutionException e)
{

6
src/java/org/apache/cassandra/transport/Client.java
View File

@ -32,11 +32,13 @@ import java.util.Map;
import com.google.common.base.Splitter;
import org.apache.cassandra.auth.IAuthenticator;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.transport.messages.*;
import org.apache.cassandra.utils.Hex;
import org.apache.cassandra.utils.MD5Digest;
import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
@ -126,7 +128,7 @@ public class Client extends SimpleClient
return null;
}
}
return new QueryMessage(query, ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList(), pageSize);
return new QueryMessage(query, new QueryOptions(ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList(), false, pageSize, null, null));
}
else if (msgType.equals("PREPARE"))
{
@ -154,7 +156,7 @@ public class Client extends SimpleClient
}
values.add(bb);
}
return new ExecuteMessage(id, values, ConsistencyLevel.ONE, -1);
return new ExecuteMessage(MD5Digest.wrap(id), new QueryOptions(ConsistencyLevel.ONE, values));
}
catch (Exception e)
{

6
src/java/org/apache/cassandra/transport/SimpleClient.java
View File

@ -32,6 +32,7 @@ import javax.net.ssl.SSLEngine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.security.SSLFactory;
import org.apache.cassandra.transport.messages.CredentialsMessage;
@ -41,6 +42,7 @@ import org.apache.cassandra.transport.messages.PrepareMessage;
import org.apache.cassandra.transport.messages.QueryMessage;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.transport.messages.StartupMessage;
import org.apache.cassandra.utils.MD5Digest;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
@ -158,7 +160,7 @@ public class SimpleClient
public ResultMessage execute(String query, List<ByteBuffer> values, ConsistencyLevel consistencyLevel)
{
Message.Response msg = execute(new QueryMessage(query, consistencyLevel, values, -1));
Message.Response msg = execute(new QueryMessage(query, new QueryOptions(consistencyLevel, values)));
assert msg instanceof ResultMessage;
return (ResultMessage)msg;
}
@ -172,7 +174,7 @@ public class SimpleClient
public ResultMessage executePrepared(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel consistency)
{
Message.Response msg = execute(new ExecuteMessage(statementId, values, consistency, -1));
Message.Response msg = execute(new ExecuteMessage(MD5Digest.wrap(statementId), new QueryOptions(consistency, values)));
assert msg instanceof ResultMessage;
return (ResultMessage)msg;
}

144
src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
View File

@ -26,9 +26,11 @@ import java.util.UUID;
import com.google.common.collect.ImmutableMap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.apache.cassandra.cql3.CQLStatement;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.PreparedQueryNotFoundException;
import org.apache.cassandra.service.QueryState;
@ -40,131 +42,65 @@ import org.apache.cassandra.utils.UUIDGen;
public class ExecuteMessage extends Message.Request
{
public static enum Flag
{
// The order of that enum matters!!
PAGE_SIZE,
SKIP_METADATA,
PAGING_STATE;
public static EnumSet<Flag> deserialize(int flags)
{
EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
Flag[] values = Flag.values();
for (int n = 0; n < values.length; n++)
{
if ((flags & (1 << n)) != 0)
set.add(values[n]);
}
return set;
}
public static int serialize(EnumSet<Flag> flags)
{
int i = 0;
for (Flag flag : flags)
i |= 1 << flag.ordinal();
return i;
}
}
public static final Message.Codec<ExecuteMessage> codec = new Message.Codec<ExecuteMessage>()
{
public ExecuteMessage decode(ChannelBuffer body, int version)
{
byte[] id = CBUtil.readBytes(body);
int count = body.readUnsignedShort();
List<ByteBuffer> values = new ArrayList<ByteBuffer>(count);
for (int i = 0; i < count; i++)
values.add(CBUtil.readValue(body));
ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
int resultPageSize = -1;
boolean skipMetadata = false;
PagingState pagingState = null;
if (version >= 2)
if (version == 1)
{
EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
if (flags.contains(Flag.PAGE_SIZE))
resultPageSize = body.readInt();
skipMetadata = flags.contains(Flag.SKIP_METADATA);
if (flags.contains(Flag.PAGING_STATE))
pagingState = PagingState.deserialize(CBUtil.readValue(body));
int count = body.readUnsignedShort();
List<ByteBuffer> values = new ArrayList<ByteBuffer>(count);
for (int i = 0; i < count; i++)
values.add(CBUtil.readValue(body));
ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
return new ExecuteMessage(id, values, consistency);
}
else
{
return new ExecuteMessage(MD5Digest.wrap(id), QueryOptions.codec.decode(body, version));
}
return new ExecuteMessage(MD5Digest.wrap(id), values, consistency, resultPageSize, skipMetadata, pagingState);
}
public ChannelBuffer encode(ExecuteMessage msg, int version)
{
// We have:
// - statementId
// - Number of values
// - The values
// - options
int vs = msg.values.size();
EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
if (msg.resultPageSize >= 0)
flags.add(Flag.PAGE_SIZE);
if (msg.skipMetadata)
flags.add(Flag.SKIP_METADATA);
if (msg.pagingState != null)
flags.add(Flag.PAGING_STATE);
assert flags.isEmpty() || version >= 2;
int nbBuff = 3;
if (version >= 2)
ChannelBuffer idBuffer = CBUtil.bytesToCB(msg.statementId.bytes);
ChannelBuffer optBuffer;
if (version == 1)
{
nbBuff++; // the flags themselves
if (flags.contains(Flag.PAGE_SIZE))
nbBuff++;
CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(2, 0, msg.options.getValues().size());
builder.add(CBUtil.shortToCB(msg.options.getValues().size()));
// Values
for (ByteBuffer value : msg.options.getValues())
builder.addValue(value);
builder.add(CBUtil.consistencyLevelToCB(msg.options.getConsistency()));
optBuffer = builder.build();
}
CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, vs + (flags.contains(Flag.PAGING_STATE) ? 1 : 0));
builder.add(CBUtil.bytesToCB(msg.statementId.bytes));
builder.add(CBUtil.shortToCB(vs));
// Values
for (ByteBuffer value : msg.values)
builder.addValue(value);
builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
if (version >= 2)
else
{
builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
if (flags.contains(Flag.PAGE_SIZE))
builder.add(CBUtil.intToCB(msg.resultPageSize));
if (flags.contains(Flag.PAGING_STATE))
builder.addValue(msg.pagingState == null ? null : msg.pagingState.serialize());
optBuffer = QueryOptions.codec.encode(msg.options, version);
}
return builder.build();
return ChannelBuffers.wrappedBuffer(idBuffer, optBuffer);
}
};
public final MD5Digest statementId;
public final List<ByteBuffer> values;
public final ConsistencyLevel consistency;
public final int resultPageSize;
public final boolean skipMetadata;
public final PagingState pagingState;
public final QueryOptions options;
public ExecuteMessage(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel consistency, int resultPageSize)
public ExecuteMessage(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel consistency)
{
this(MD5Digest.wrap(statementId), values, consistency, resultPageSize, false, null);
this(MD5Digest.wrap(statementId), new QueryOptions(consistency, values));
}
public ExecuteMessage(MD5Digest statementId, List<ByteBuffer> values, ConsistencyLevel consistency, int resultPageSize, boolean skipMetadata, PagingState pagingState)
public ExecuteMessage(MD5Digest statementId, QueryOptions options)
{
super(Message.Type.EXECUTE);
this.statementId = statementId;
this.values = values;
this.consistency = consistency;
this.resultPageSize = resultPageSize;
this.skipMetadata = skipMetadata;
this.pagingState = pagingState;
this.options = options;
}
public ChannelBuffer encode(int version)
@ -181,7 +117,7 @@ public class ExecuteMessage extends Message.Request
if (statement == null)
throw new PreparedQueryNotFoundException(statementId);
if (resultPageSize == 0)
if (options.getPageSize() == 0)
throw new ProtocolException("The page size cannot be 0");
UUID tracingId = null;
@ -196,15 +132,15 @@ public class ExecuteMessage extends Message.Request
state.createTracingSession();
ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
if (resultPageSize > 0)
builder.put("page_size", Integer.toString(resultPageSize));
if (options.getPageSize() > 0)
builder.put("page_size", Integer.toString(options.getPageSize()));
// TODO we don't have [typed] access to CQL bind variables here. CASSANDRA-4560 is open to add support.
Tracing.instance.begin("Execute CQL3 prepared query", builder.build());
}
Message.Response response = QueryProcessor.processPrepared(statement, consistency, state, values, resultPageSize, pagingState);
if (skipMetadata && response instanceof ResultMessage.Rows)
Message.Response response = QueryProcessor.processPrepared(statement, state, options);
if (options.skipMetadata() && response instanceof ResultMessage.Rows)
((ResultMessage.Rows)response).result.metadata.setSkipMetadata();
if (tracingId != null)
@ -225,6 +161,6 @@ public class ExecuteMessage extends Message.Request
@Override
public String toString()
{
return "EXECUTE " + statementId + " with " + values.size() + " values at consistency " + consistency;
return "EXECUTE " + statementId + " with " + options.getValues().size() + " values at consistency " + options.getConsistency();
}
}

144
src/java/org/apache/cassandra/transport/messages/QueryMessage.java
View File

@ -26,8 +26,10 @@ import java.util.UUID;
import com.google.common.collect.ImmutableMap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.QueryState;
@ -41,147 +43,43 @@ import org.apache.cassandra.utils.UUIDGen;
*/
public class QueryMessage extends Message.Request
{
public static enum Flag
{
// The order of that enum matters!!
PAGE_SIZE,
VALUES,
SKIP_METADATA,
PAGING_STATE;
public static EnumSet<Flag> deserialize(int flags)
{
EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
Flag[] values = Flag.values();
for (int n = 0; n < values.length; n++)
{
if ((flags & (1 << n)) != 0)
set.add(values[n]);
}
return set;
}
public static int serialize(EnumSet<Flag> flags)
{
int i = 0;
for (Flag flag : flags)
i |= 1 << flag.ordinal();
return i;
}
}
public static final Message.Codec<QueryMessage> codec = new Message.Codec<QueryMessage>()
{
public QueryMessage decode(ChannelBuffer body, int version)
{
String query = CBUtil.readLongString(body);
ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
int resultPageSize = -1;
List<ByteBuffer> values = Collections.emptyList();
boolean skipMetadata = false;
PagingState pagingState = null;
if (version >= 2)
if (version == 1)
{
EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
if (flags.contains(Flag.PAGE_SIZE))
resultPageSize = body.readInt();
if (flags.contains(Flag.VALUES))
{
int paramCount = body.readUnsignedShort();
values = paramCount == 0 ? Collections.<ByteBuffer>emptyList() : new ArrayList<ByteBuffer>(paramCount);
for (int i = 0; i < paramCount; i++)
values.add(CBUtil.readValue(body));
}
skipMetadata = flags.contains(Flag.SKIP_METADATA);
if (flags.contains(Flag.PAGING_STATE))
pagingState = PagingState.deserialize(CBUtil.readValue(body));
ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
return new QueryMessage(query, consistency);
}
else
{
return new QueryMessage(query, QueryOptions.codec.decode(body, version));
}
return new QueryMessage(query, consistency, values, resultPageSize, skipMetadata, pagingState);
}
public ChannelBuffer encode(QueryMessage msg, int version)
{
// We have:
// - query
// - options
// * optional:
// - Number of values
// - The values
int vs = msg.values.size();
EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
if (msg.resultPageSize >= 0)
flags.add(Flag.PAGE_SIZE);
if (vs > 0)
flags.add(Flag.VALUES);
if (msg.skipMetadata)
flags.add(Flag.SKIP_METADATA);
if (msg.pagingState != null)
flags.add(Flag.PAGING_STATE);
assert flags.isEmpty() || version >= 2 : "Version 1 of the protocol supports no option after the consistency level";
int nbBuff = 2;
if (version >= 2)
{
nbBuff++; // the flags themselves
if (flags.contains(Flag.PAGE_SIZE))
nbBuff++;
if (flags.contains(Flag.VALUES))
nbBuff++;
}
CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, vs + (flags.contains(Flag.PAGING_STATE) ? 1 : 0));
builder.add(CBUtil.longStringToCB(msg.query));
builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
if (version >= 2)
{
builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
if (flags.contains(Flag.PAGE_SIZE))
builder.add(CBUtil.intToCB(msg.resultPageSize));
if (flags.contains(Flag.VALUES))
{
builder.add(CBUtil.shortToCB(vs));
for (ByteBuffer value : msg.values)
builder.addValue(value);
}
if (flags.contains(Flag.PAGING_STATE))
builder.addValue(msg.pagingState == null ? null : msg.pagingState.serialize());
}
return builder.build();
return ChannelBuffers.wrappedBuffer(CBUtil.longStringToCB(msg.query),
(version == 1 ? CBUtil.consistencyLevelToCB(msg.options.getConsistency())
: QueryOptions.codec.encode(msg.options, version)));
}
};
public final String query;
public final ConsistencyLevel consistency;
public final int resultPageSize;
public final List<ByteBuffer> values;
public final boolean skipMetadata;
public final PagingState pagingState;
public final QueryOptions options;
public QueryMessage(String query, ConsistencyLevel consistency)
{
this(query, consistency, Collections.<ByteBuffer>emptyList(), -1);
this(query, new QueryOptions(consistency, Collections.<ByteBuffer>emptyList()));
}
public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer> values, int resultPageSize)
{
this(query, consistency, values, resultPageSize, false, null);
}
public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer> values, int resultPageSize, boolean skipMetadata, PagingState pagingState)
public QueryMessage(String query, QueryOptions options)
{
super(Type.QUERY);
this.query = query;
this.consistency = consistency;
this.resultPageSize = resultPageSize;
this.values = values;
this.skipMetadata = skipMetadata;
this.pagingState = pagingState;
this.options = options;
}
public ChannelBuffer encode(int version)
@ -193,7 +91,7 @@ public class QueryMessage extends Message.Request
{
try
{
if (resultPageSize == 0)
if (options.getPageSize() == 0)
throw new ProtocolException("The page size cannot be 0");
UUID tracingId = null;
@ -209,14 +107,14 @@ public class QueryMessage extends Message.Request
ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
builder.put("query", query);
if (resultPageSize > 0)
builder.put("page_size", Integer.toString(resultPageSize));
if (options.getPageSize() > 0)
builder.put("page_size", Integer.toString(options.getPageSize()));
Tracing.instance.begin("Execute CQL3 query", builder.build());
}
Message.Response response = QueryProcessor.process(query, values, consistency, state, resultPageSize, pagingState);
if (skipMetadata && response instanceof ResultMessage.Rows)
Message.Response response = QueryProcessor.process(query, state, options);
if (options.skipMetadata() && response instanceof ResultMessage.Rows)
((ResultMessage.Rows)response).result.metadata.setSkipMetadata();
if (tracingId != null)