From bfd73beaf1554370eb23897a403991b3e2467dd1 Mon Sep 17 00:00:00 2001 From: Sylvain Lebresne Date: Thu, 25 Jul 2013 10:25:31 +0200 Subject: [PATCH] DC-local CAS patch by slebresne; reviewed by jbellis for CASSANDRA-5797 --- CHANGES.txt | 1 + doc/native_protocol_v2.spec | 58 ++-- interface/cassandra.thrift | 13 +- .../thrift/AuthenticationRequest.java | 4 +- .../apache/cassandra/thrift/Cassandra.java | 306 +++++++++++++----- .../org/apache/cassandra/thrift/CfDef.java | 8 +- .../apache/cassandra/thrift/ColumnDef.java | 4 +- .../cassandra/thrift/ConsistencyLevel.java | 5 +- .../apache/cassandra/thrift/CqlMetadata.java | 8 +- .../org/apache/cassandra/thrift/KsDef.java | 4 +- .../apache/cassandra/thrift/TriggerDef.java | 4 +- src/java/org/apache/cassandra/auth/Auth.java | 8 +- .../cassandra/auth/CassandraAuthorizer.java | 11 +- .../cassandra/auth/PasswordAuthenticator.java | 9 +- .../apache/cassandra/cql3/CQLStatement.java | 11 +- .../apache/cassandra/cql3/QueryOptions.java | 237 ++++++++++++++ .../apache/cassandra/cql3/QueryProcessor.java | 21 +- .../statements/AuthenticationStatement.java | 3 +- .../statements/AuthorizationStatement.java | 3 +- .../cql3/statements/BatchStatement.java | 6 +- .../statements/ModificationStatement.java | 18 +- .../statements/SchemaAlteringStatement.java | 3 +- .../cql3/statements/SelectStatement.java | 11 +- .../cql3/statements/TruncateStatement.java | 2 +- .../cql3/statements/UseStatement.java | 3 +- .../apache/cassandra/db/ConsistencyLevel.java | 25 +- .../cassandra/service/StorageProxy.java | 64 +++- .../cassandra/thrift/CassandraServer.java | 22 +- .../apache/cassandra/transport/Client.java | 6 +- .../cassandra/transport/SimpleClient.java | 6 +- .../transport/messages/ExecuteMessage.java | 144 +++------ .../transport/messages/QueryMessage.java | 144 ++------- 32 files changed, 719 insertions(+), 453 deletions(-) create mode 100644 src/java/org/apache/cassandra/cql3/QueryOptions.java diff --git a/CHANGES.txt b/CHANGES.txt index 0fee57e2ce..94b52360a6 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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 diff --git a/doc/native_protocol_v2.spec b/doc/native_protocol_v2.spec index 9b6d52c67d..649d493233 100644 --- a/doc/native_protocol_v2.spec +++ b/doc/native_protocol_v2.spec @@ -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 where and are [string]. @@ -280,29 +282,37 @@ Table of Contents 4.1.4. QUERY Performs a CQL query. The body of the message must be: - [][...][] + + where is a [long string] representing the query and + must be + [...][][][] where: + - is the [consistency] level for the operation. - 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, 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] followed by [bytes] + 0x01: Values. In that case, a [short] followed by [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, 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, should be present. 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). - - the query, [long string]. - - is the [consistency] level for the operation. + 0x05: With serial consistency. If present, should be + present. 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: - ....[][] - where: - - is the prepared query ID. It's the [short bytes] returned as a - response to a PREPARE message. - - is a [short] indicating the number of following values. - - ... are the [bytes] to use for bound variables in the - prepared query. - - is the [consistency] level for the operation. - - 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, 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, should be present. - 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, ...). + + where is the prepared query ID. It's the [short bytes] returned as a + response to a PREPARE message. As for , 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 diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift index e3d3374b33..3f294eb545 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -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 expected, 4:list 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), /** diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/AuthenticationRequest.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/AuthenticationRequest.java index 3f7f024261..5697987b84 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/AuthenticationRequest.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/AuthenticationRequest.java @@ -383,7 +383,7 @@ public class AuthenticationRequest implements org.apache.thrift.TBase(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(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java index 0460bd7667..e495373e56 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java @@ -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 expected, List updates, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException; + public CASResult cas(ByteBuffer key, String column_family, List expected, List 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 resultHandler) throws org.apache.thrift.TException; - public void cas(ByteBuffer key, String column_family, List expected, List updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; + public void cas(ByteBuffer key, String column_family, List expected, List updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; public void remove(ByteBuffer key, ColumnPath column_path, long timestamp, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException; @@ -908,20 +918,21 @@ public class Cassandra { return; } - public CASResult cas(ByteBuffer key, String column_family, List expected, List updates, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, org.apache.thrift.TException + public CASResult cas(ByteBuffer key, String column_family, List expected, List 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 expected, List updates, ConsistencyLevel consistency_level) throws org.apache.thrift.TException + public void send_cas(ByteBuffer key, String column_family, List expected, List 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 expected, List updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException { + public void cas(ByteBuffer key, String column_family, List expected, List updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback 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 expected; private List updates; - private ConsistencyLevel consistency_level; - public cas_call(ByteBuffer key, String column_family, List expected, List updates, ConsistencyLevel consistency_level, org.apache.thrift.async.AsyncMethodCallback 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 expected, List updates, ConsistencyLevel serial_consistency_level, ConsistencyLevel commit_consistency_level, org.apache.thrift.async.AsyncMethodCallback 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>(2*_map232.size); for (int _i233 = 0; _i233 < _map232.size; ++_i233) { - ByteBuffer _key234; // optional + ByteBuffer _key234; // required List _val235; // required _key234 = iprot.readBinary(); { @@ -11731,7 +11745,7 @@ public class Cassandra { struct.success = new HashMap>(2*_map243.size); for (int _i244 = 0; _i244 < _map243.size; ++_i244) { - ByteBuffer _key245; // optional + ByteBuffer _key245; // required List _val246; // required _key245 = iprot.readBinary(); { @@ -13082,7 +13096,7 @@ public class Cassandra { struct.success = new HashMap(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(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, SchemeFactory> schemes = new HashMap, 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 byName = new HashMap(); @@ -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 expected, List 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>>(2*_map308.size); for (int _i309 = 0; _i309 < _map308.size; ++_i309) { - ByteBuffer _key310; // optional + ByteBuffer _key310; // required Map> _val311; // required _key310 = iprot.readBinary(); { @@ -24581,7 +24715,7 @@ public class Cassandra { _val311 = new HashMap>(2*_map312.size); for (int _i313 = 0; _i313 < _map312.size; ++_i313) { - String _key314; // optional + String _key314; // required List _val315; // required _key314 = iprot.readString(); { @@ -24714,7 +24848,7 @@ public class Cassandra { struct.mutation_map = new HashMap>>(2*_map325.size); for (int _i326 = 0; _i326 < _map325.size; ++_i326) { - ByteBuffer _key327; // optional + ByteBuffer _key327; // required Map> _val328; // required _key327 = iprot.readBinary(); { @@ -24722,7 +24856,7 @@ public class Cassandra { _val328 = new HashMap>(2*_map329.size); for (int _i330 = 0; _i330 < _map329.size; ++_i330) { - String _key331; // optional + String _key331; // required List _val332; // required _key331 = iprot.readString(); { @@ -25777,7 +25911,7 @@ public class Cassandra { struct.mutation_map = new HashMap>>(2*_map336.size); for (int _i337 = 0; _i337 < _map336.size; ++_i337) { - ByteBuffer _key338; // optional + ByteBuffer _key338; // required Map> _val339; // required _key338 = iprot.readBinary(); { @@ -25785,7 +25919,7 @@ public class Cassandra { _val339 = new HashMap>(2*_map340.size); for (int _i341 = 0; _i341 < _map340.size; ++_i341) { - String _key342; // optional + String _key342; // required List _val343; // required _key342 = iprot.readString(); { @@ -25918,7 +26052,7 @@ public class Cassandra { struct.mutation_map = new HashMap>>(2*_map353.size); for (int _i354 = 0; _i354 < _map353.size; ++_i354) { - ByteBuffer _key355; // optional + ByteBuffer _key355; // required Map> _val356; // required _key355 = iprot.readBinary(); { @@ -25926,7 +26060,7 @@ public class Cassandra { _val356 = new HashMap>(2*_map357.size); for (int _i358 = 0; _i358 < _map357.size; ++_i358) { - String _key359; // optional + String _key359; // required List _val360; // required _key359 = iprot.readString(); { @@ -28121,7 +28255,7 @@ public class Cassandra { struct.success = new HashMap>(2*_map364.size); for (int _i365 = 0; _i365 < _map364.size; ++_i365) { - String _key366; // optional + String _key366; // required List _val367; // required _key366 = iprot.readString(); { @@ -28249,7 +28383,7 @@ public class Cassandra { struct.success = new HashMap>(2*_map375.size); for (int _i376 = 0; _i376 < _map375.size; ++_i376) { - String _key377; // optional + String _key377; // required List _val378; // required _key377 = iprot.readString(); { @@ -31796,7 +31930,7 @@ public class Cassandra { struct.success = new HashMap(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(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java index e742cc5d42..7cd42f2996 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/CfDef.java @@ -3674,7 +3674,7 @@ public class CfDef implements org.apache.thrift.TBase, jav struct.compaction_strategy_options = new HashMap(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, jav struct.compression_options = new HashMap(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, jav struct.compaction_strategy_options = new HashMap(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, jav struct.compression_options = new HashMap(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/ColumnDef.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/ColumnDef.java index 1557c3a1c3..35ec3b7f08 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/ColumnDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/ColumnDef.java @@ -771,7 +771,7 @@ public class ColumnDef implements org.apache.thrift.TBase(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(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/ConsistencyLevel.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/ConsistencyLevel.java index 96b6308fd4..ff7d51eda8 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/ConsistencyLevel.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/ConsistencyLevel.java @@ -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; } diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/CqlMetadata.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/CqlMetadata.java index 7a54c300c9..e37d200c24 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/CqlMetadata.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/CqlMetadata.java @@ -662,7 +662,7 @@ public class CqlMetadata implements org.apache.thrift.TBase(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(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(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(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java index 679f270bb2..dbc4d13afe 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/KsDef.java @@ -841,7 +841,7 @@ public class KsDef implements org.apache.thrift.TBase, jav struct.strategy_options = new HashMap(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, jav struct.strategy_options = new HashMap(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(); diff --git a/interface/thrift/gen-java/org/apache/cassandra/thrift/TriggerDef.java b/interface/thrift/gen-java/org/apache/cassandra/thrift/TriggerDef.java index 398281b177..0d31d1b725 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/TriggerDef.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/TriggerDef.java @@ -478,7 +478,7 @@ public class TriggerDef implements org.apache.thrift.TBase(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(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(); diff --git a/src/java/org/apache/cassandra/auth/Auth.java b/src/java/org/apache/cassandra/auth/Auth.java index 53debce6d2..38da260a10 100644 --- a/src/java/org/apache/cassandra/auth/Auth.java +++ b/src/java/org/apache/cassandra/auth/Auth.java @@ -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) diff --git a/src/java/org/apache/cassandra/auth/CassandraAuthorizer.java b/src/java/org/apache/cassandra/auth/CassandraAuthorizer.java index d07c2c8826..dae81987f9 100644 --- a/src/java/org/apache/cassandra/auth/CassandraAuthorizer.java +++ b/src/java/org/apache/cassandra/auth/CassandraAuthorizer.java @@ -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) diff --git a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java index 87539f27a4..089114270f 100644 --- a/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java +++ b/src/java/org/apache/cassandra/auth/PasswordAuthenticator.java @@ -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) diff --git a/src/java/org/apache/cassandra/cql3/CQLStatement.java b/src/java/org/apache/cassandra/cql3/CQLStatement.java index 2f4dd65319..d123868df7 100644 --- a/src/java/org/apache/cassandra/cql3/CQLStatement.java +++ b/src/java/org/apache/cassandra/cql3/CQLStatement.java @@ -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 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. diff --git a/src/java/org/apache/cassandra/cql3/QueryOptions.java b/src/java/org/apache/cassandra/cql3/QueryOptions.java new file mode 100644 index 0000000000..662d463b2b --- /dev/null +++ b/src/java/org/apache/cassandra/cql3/QueryOptions.java @@ -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.emptyList()); + + public static final CBCodec codec = new Codec(); + + private final ConsistencyLevel consistency; + private final List values; + private final boolean skipMetadata; + + private final SpecificOptions options; + + public QueryOptions(ConsistencyLevel consistency, List values) + { + this(consistency, values, false, SpecificOptions.DEFAULT); + } + + public QueryOptions(ConsistencyLevel consistency, + List values, + boolean skipMetadata, + int pageSize, + PagingState pagingState, + ConsistencyLevel serialConsistency) + { + this(consistency, values, skipMetadata, new SpecificOptions(pageSize, pagingState, serialConsistency)); + } + + private QueryOptions(ConsistencyLevel consistency, List values, boolean skipMetadata, SpecificOptions options) + { + this.consistency = consistency; + this.values = values; + this.skipMetadata = skipMetadata; + this.options = options; + } + + public ConsistencyLevel getConsistency() + { + return consistency; + } + + public List 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 + { + private static enum Flag + { + // The order of that enum matters!! + VALUES, + SKIP_METADATA, + PAGE_SIZE, + PAGING_STATE, + SERIAL_CONSISTENCY; + + public static EnumSet deserialize(int flags) + { + EnumSet 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 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 flags = Flag.deserialize((int)body.readByte()); + + List values = Collections.emptyList(); + if (flags.contains(Flag.VALUES)) + { + int paramCount = body.readUnsignedShort(); + if (paramCount > 0) + { + values = new ArrayList(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 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(); + } + } +} diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java index 91657fb3b9..f2f71e5882 100644 --- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java @@ -94,30 +94,30 @@ public class QueryProcessor } } - private static ResultMessage processStatement(CQLStatement statement, ConsistencyLevel cl, QueryState queryState, List 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.emptyList(), cl, queryState, -1, null); + return process(queryString, queryState, new QueryOptions(cl, Collections.emptyList())); } - public static ResultMessage process(String queryString, List 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.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 variables, int pageSize, PagingState pageState) + public static ResultMessage processPrepared(CQLStatement statement, QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException { + List 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> variables) diff --git a/src/java/org/apache/cassandra/cql3/statements/AuthenticationStatement.java b/src/java/org/apache/cassandra/cql3/statements/AuthenticationStatement.java index eecc73a307..d5a515ddc7 100644 --- a/src/java/org/apache/cassandra/cql3/statements/AuthenticationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/AuthenticationStatement.java @@ -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 variables, int pageSize, PagingState pagingState) + public ResultMessage execute(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException { return execute(state.getClientState()); diff --git a/src/java/org/apache/cassandra/cql3/statements/AuthorizationStatement.java b/src/java/org/apache/cassandra/cql3/statements/AuthorizationStatement.java index cb3baa3977..15ac8256c9 100644 --- a/src/java/org/apache/cassandra/cql3/statements/AuthorizationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/AuthorizationStatement.java @@ -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 variables, int pageSize, PagingState pagingState) + public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException, RequestExecutionException { return execute(state.getClientState()); diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java index 90dc76474a..2c661998f6 100644 --- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java @@ -134,12 +134,12 @@ public class BatchStatement implements CQLStatement } } - public ResultMessage execute(ConsistencyLevel cl, QueryState queryState, List 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; } diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java index 298715463b..0aa188d84c 100644 --- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java @@ -334,32 +334,34 @@ public abstract class ModificationStatement implements CQLStatement return ifNotExists || (columnConditions != null && !columnConditions.isEmpty()); } - public ResultMessage execute(ConsistencyLevel cl, QueryState queryState, List 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 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 variables) + public ResultMessage executeWithCondition(QueryState queryState, QueryOptions options) throws RequestExecutionException, RequestValidationException { + List variables = options.getValues(); List 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)); } diff --git a/src/java/org/apache/cassandra/cql3/statements/SchemaAlteringStatement.java b/src/java/org/apache/cassandra/cql3/statements/SchemaAlteringStatement.java index f92d8b3951..e9e828c3d1 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SchemaAlteringStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SchemaAlteringStatement.java @@ -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 variables, int pageSize, PagingState pagingState) throws RequestValidationException + public ResultMessage execute(QueryState state, QueryOptions options) throws RequestValidationException { announceMigration(); String tableName = cfName == null || columnFamily() == null ? "" : columnFamily(); diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java index 2f9627b975..8b9f7df6dc 100644 --- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java @@ -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 variables, int pageSize, PagingState pagingState) throws RequestExecutionException, RequestValidationException + public ResultMessage.Rows execute(QueryState state, QueryOptions options) throws RequestExecutionException, RequestValidationException { + ConsistencyLevel cl = options.getConsistency(); + List 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); diff --git a/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java b/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java index c66608b186..318995edbd 100644 --- a/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java @@ -55,7 +55,7 @@ public class TruncateStatement extends CFStatement implements CQLStatement ThriftValidation.validateColumnFamily(keyspace(), columnFamily()); } - public ResultMessage execute(ConsistencyLevel cl, QueryState state, List variables, int pageSize, PagingState pagingState) throws InvalidRequestException, TruncateException + public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException, TruncateException { try { diff --git a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java index c2e3c342b1..ab42fed8aa 100644 --- a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java +++ b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java @@ -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 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); diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java index d642d080a4..956ab58289 100644 --- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java +++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java @@ -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"); } diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 443d6b8305..9b7da6de8e 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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, Integer> p = getPaxosParticipants(keyspaceName, key); + Pair, Integer> p = getPaxosParticipants(keyspaceName, key, consistencyForPaxos); List 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, Integer> getPaxosParticipants(String keyspaceName, ByteBuffer key) throws UnavailableException + private static Predicate sameDCPredicateFor(final String dc) + { + final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch(); + return new Predicate() + { + public boolean apply(InetAddress host) + { + return dc.equals(snitch.getDatacenter(host)); + } + }; + } + + private static Pair, Integer> getPaxosParticipants(String keyspaceName, ByteBuffer key, ConsistencyLevel consistencyForPaxos) throws UnavailableException { Token tk = StorageService.getPartitioner().getToken(key); List naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName, tk); Collection 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 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 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 liveEndpoints, int requiredParticipants) + private static UUID beginAndRepairPaxos(long start, ByteBuffer key, CFMetaData metadata, List 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 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, Integer> p = getPaxosParticipants(command.ksName, command.key); + Pair, Integer> p = getPaxosParticipants(command.ksName, command.key, consistency_level); List 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); diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index 15ab9d0b9c..e9473d9452 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -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 expected, List updates, ConsistencyLevel consistency_level) + public CASResult cas(ByteBuffer key, + String column_family, + List expected, + List 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) { diff --git a/src/java/org/apache/cassandra/transport/Client.java b/src/java/org/apache/cassandra/transport/Client.java index 12c46ddcc9..f0b700c9c4 100644 --- a/src/java/org/apache/cassandra/transport/Client.java +++ b/src/java/org/apache/cassandra/transport/Client.java @@ -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.emptyList(), pageSize); + return new QueryMessage(query, new QueryOptions(ConsistencyLevel.ONE, Collections.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) { diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java index df4f8118f4..cfe1bab543 100644 --- a/src/java/org/apache/cassandra/transport/SimpleClient.java +++ b/src/java/org/apache/cassandra/transport/SimpleClient.java @@ -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 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 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; } diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java index f83df9d43a..c2974263a7 100644 --- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java @@ -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 deserialize(int flags) - { - EnumSet 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 flags) - { - int i = 0; - for (Flag flag : flags) - i |= 1 << flag.ordinal(); - return i; - } - } - public static final Message.Codec codec = new Message.Codec() { public ExecuteMessage decode(ChannelBuffer body, int version) { byte[] id = CBUtil.readBytes(body); - int count = body.readUnsignedShort(); - List values = new ArrayList(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 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 values = new ArrayList(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 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 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 values, ConsistencyLevel consistency, int resultPageSize) + public ExecuteMessage(byte[] statementId, List 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 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 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(); } } diff --git a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java index 9e8050c0f8..2b2583e2be 100644 --- a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java +++ b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java @@ -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 deserialize(int flags) - { - EnumSet 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 flags) - { - int i = 0; - for (Flag flag : flags) - i |= 1 << flag.ordinal(); - return i; - } - } - public static final Message.Codec codec = new Message.Codec() { public QueryMessage decode(ChannelBuffer body, int version) { String query = CBUtil.readLongString(body); - ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body); - - int resultPageSize = -1; - List values = Collections.emptyList(); - boolean skipMetadata = false; - PagingState pagingState = null; - if (version >= 2) + if (version == 1) { - EnumSet 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.emptyList() : new ArrayList(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 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 values; - public final boolean skipMetadata; - public final PagingState pagingState; + public final QueryOptions options; public QueryMessage(String query, ConsistencyLevel consistency) { - this(query, consistency, Collections.emptyList(), -1); + this(query, new QueryOptions(consistency, Collections.emptyList())); } - public QueryMessage(String query, ConsistencyLevel consistency, List values, int resultPageSize) - { - this(query, consistency, values, resultPageSize, false, null); - } - - public QueryMessage(String query, ConsistencyLevel consistency, List 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 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)