From d76bbca3e2f3572ad83daa8de3db40560148de9e Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 27 Jul 2009 21:44:47 +0000 Subject: [PATCH] add ConsistencyLevel Thrift enum to avoid client having to compute quorum size. patch by Sammy Yu; reviewed by jbellis for CASSANDRA-300 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@798298 13f79535-47bb-0310-9956-ffa450edef68 --- interface/cassandra.thrift | 16 +- .../apache/cassandra/service/Cassandra.java | 428 ++++++++++-------- .../cassandra/service/ConsistencyLevel.java | 29 ++ .../org/apache/cassandra/cli/CliClient.java | 2 +- .../cassandra/service/CassandraServer.java | 23 +- .../cassandra/service/StorageProxy.java | 21 +- test/system/test_server.py | 72 +-- 7 files changed, 349 insertions(+), 242 deletions(-) create mode 100644 interface/gen-java/org/apache/cassandra/service/ConsistencyLevel.java diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift index ecd976e5d7..58679e1c40 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -64,7 +64,6 @@ struct CqlResult { 3: ResultSet result_set, } - # # Exceptions # @@ -89,6 +88,13 @@ exception UnavailableException { # service api # +enum ConsistencyLevel { + ZERO = 0, + ONE = 1, + QUORUM = 2, + ALL = 3, +} + struct ColumnParent { 3: string column_family, 4: optional binary super_column, @@ -125,13 +131,13 @@ service Cassandra { i32 get_column_count(1:string table, 2:string key, 3:ColumnParent column_parent) throws (1: InvalidRequestException ire), - void insert(1:string table, 2:string key, 3:ColumnPath column_path, 4:binary value, 5:i64 timestamp, 6:i32 block_for=0) + void insert(1:string table, 2:string key, 3:ColumnPath column_path, 4:binary value, 5:i64 timestamp, 6:ConsistencyLevel consistency_level=0) throws (1: InvalidRequestException ire, 2: UnavailableException ue), - void batch_insert(1:string table, 2:BatchMutation batch_mutation, 3:i32 block_for=0) + void batch_insert(1:string table, 2:BatchMutation batch_mutation, 3:ConsistencyLevel consistency_level=0) throws (1: InvalidRequestException ire, 2: UnavailableException ue), - void remove(1:string table, 2:string key, 3:ColumnPathOrParent column_path_or_parent, 4:i64 timestamp, 5:i32 block_for=0) + void remove(1:string table, 2:string key, 3:ColumnPathOrParent column_path_or_parent, 4:i64 timestamp, 5:ConsistencyLevel consistency_level=0) throws (1: InvalidRequestException ire, 2: UnavailableException ue), list get_slice_super(1:string table, 2:string key, 3:string column_family, 4:binary start, 5:binary finish, 6:bool is_ascending, 7:i32 count=100) @@ -143,7 +149,7 @@ service Cassandra { SuperColumn get_super_column(1:string table, 2:string key, 3:SuperColumnPath super_column_path) throws (1: InvalidRequestException ire, 2: NotFoundException nfe), - void batch_insert_super_column(1:string table, 2:BatchMutationSuper batch_mutation_super, 3:i32 block_for=0) + void batch_insert_super_column(1:string table, 2:BatchMutationSuper batch_mutation_super, 3:ConsistencyLevel consistency_level=0) throws (1: InvalidRequestException ire, 2: UnavailableException ue), # range query: returns matching keys diff --git a/interface/gen-java/org/apache/cassandra/service/Cassandra.java b/interface/gen-java/org/apache/cassandra/service/Cassandra.java index c9f74e425b..8b55d06291 100644 --- a/interface/gen-java/org/apache/cassandra/service/Cassandra.java +++ b/interface/gen-java/org/apache/cassandra/service/Cassandra.java @@ -30,11 +30,11 @@ public class Cassandra { public int get_column_count(String table, String key, ColumnParent column_parent) throws InvalidRequestException, TException; - public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int block_for) throws InvalidRequestException, UnavailableException, TException; + public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException, TException; - public void batch_insert(String table, BatchMutation batch_mutation, int block_for) throws InvalidRequestException, UnavailableException, TException; + public void batch_insert(String table, BatchMutation batch_mutation, int consistency_level) throws InvalidRequestException, UnavailableException, TException; - public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int block_for) throws InvalidRequestException, UnavailableException, TException; + public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException, TException; public List get_slice_super(String table, String key, String column_family, byte[] start, byte[] finish, boolean is_ascending, int count) throws InvalidRequestException, TException; @@ -42,7 +42,7 @@ public class Cassandra { public SuperColumn get_super_column(String table, String key, SuperColumnPath super_column_path) throws InvalidRequestException, NotFoundException, TException; - public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int block_for) throws InvalidRequestException, UnavailableException, TException; + public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int consistency_level) throws InvalidRequestException, UnavailableException, TException; public List get_key_range(String table, String column_family, String start, String finish, int count) throws InvalidRequestException, TException; @@ -249,13 +249,13 @@ public class Cassandra { throw new TApplicationException(TApplicationException.MISSING_RESULT, "get_column_count failed: unknown result"); } - public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int block_for) throws InvalidRequestException, UnavailableException, TException + public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException, TException { - send_insert(table, key, column_path, value, timestamp, block_for); + send_insert(table, key, column_path, value, timestamp, consistency_level); recv_insert(); } - public void send_insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int block_for) throws TException + public void send_insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int consistency_level) throws TException { oprot_.writeMessageBegin(new TMessage("insert", TMessageType.CALL, seqid_)); insert_args args = new insert_args(); @@ -264,7 +264,7 @@ public class Cassandra { args.column_path = column_path; args.value = value; args.timestamp = timestamp; - args.block_for = block_for; + args.consistency_level = consistency_level; args.write(oprot_); oprot_.writeMessageEnd(); oprot_.getTransport().flush(); @@ -290,19 +290,19 @@ public class Cassandra { return; } - public void batch_insert(String table, BatchMutation batch_mutation, int block_for) throws InvalidRequestException, UnavailableException, TException + public void batch_insert(String table, BatchMutation batch_mutation, int consistency_level) throws InvalidRequestException, UnavailableException, TException { - send_batch_insert(table, batch_mutation, block_for); + send_batch_insert(table, batch_mutation, consistency_level); recv_batch_insert(); } - public void send_batch_insert(String table, BatchMutation batch_mutation, int block_for) throws TException + public void send_batch_insert(String table, BatchMutation batch_mutation, int consistency_level) throws TException { oprot_.writeMessageBegin(new TMessage("batch_insert", TMessageType.CALL, seqid_)); batch_insert_args args = new batch_insert_args(); args.table = table; args.batch_mutation = batch_mutation; - args.block_for = block_for; + args.consistency_level = consistency_level; args.write(oprot_); oprot_.writeMessageEnd(); oprot_.getTransport().flush(); @@ -328,13 +328,13 @@ public class Cassandra { return; } - public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int block_for) throws InvalidRequestException, UnavailableException, TException + public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException, TException { - send_remove(table, key, column_path_or_parent, timestamp, block_for); + send_remove(table, key, column_path_or_parent, timestamp, consistency_level); recv_remove(); } - public void send_remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int block_for) throws TException + public void send_remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int consistency_level) throws TException { oprot_.writeMessageBegin(new TMessage("remove", TMessageType.CALL, seqid_)); remove_args args = new remove_args(); @@ -342,7 +342,7 @@ public class Cassandra { args.key = key; args.column_path_or_parent = column_path_or_parent; args.timestamp = timestamp; - args.block_for = block_for; + args.consistency_level = consistency_level; args.write(oprot_); oprot_.writeMessageEnd(); oprot_.getTransport().flush(); @@ -490,19 +490,19 @@ public class Cassandra { throw new TApplicationException(TApplicationException.MISSING_RESULT, "get_super_column failed: unknown result"); } - public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int block_for) throws InvalidRequestException, UnavailableException, TException + public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int consistency_level) throws InvalidRequestException, UnavailableException, TException { - send_batch_insert_super_column(table, batch_mutation_super, block_for); + send_batch_insert_super_column(table, batch_mutation_super, consistency_level); recv_batch_insert_super_column(); } - public void send_batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int block_for) throws TException + public void send_batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int consistency_level) throws TException { oprot_.writeMessageBegin(new TMessage("batch_insert_super_column", TMessageType.CALL, seqid_)); batch_insert_super_column_args args = new batch_insert_super_column_args(); args.table = table; args.batch_mutation_super = batch_mutation_super; - args.block_for = block_for; + args.consistency_level = consistency_level; args.write(oprot_); oprot_.writeMessageEnd(); oprot_.getTransport().flush(); @@ -879,7 +879,7 @@ public class Cassandra { iprot.readMessageEnd(); insert_result result = new insert_result(); try { - iface_.insert(args.table, args.key, args.column_path, args.value, args.timestamp, args.block_for); + iface_.insert(args.table, args.key, args.column_path, args.value, args.timestamp, args.consistency_level); } catch (InvalidRequestException ire) { result.ire = ire; } catch (UnavailableException ue) { @@ -909,7 +909,7 @@ public class Cassandra { iprot.readMessageEnd(); batch_insert_result result = new batch_insert_result(); try { - iface_.batch_insert(args.table, args.batch_mutation, args.block_for); + iface_.batch_insert(args.table, args.batch_mutation, args.consistency_level); } catch (InvalidRequestException ire) { result.ire = ire; } catch (UnavailableException ue) { @@ -939,7 +939,7 @@ public class Cassandra { iprot.readMessageEnd(); remove_result result = new remove_result(); try { - iface_.remove(args.table, args.key, args.column_path_or_parent, args.timestamp, args.block_for); + iface_.remove(args.table, args.key, args.column_path_or_parent, args.timestamp, args.consistency_level); } catch (InvalidRequestException ire) { result.ire = ire; } catch (UnavailableException ue) { @@ -1055,7 +1055,7 @@ public class Cassandra { iprot.readMessageEnd(); batch_insert_super_column_result result = new batch_insert_super_column_result(); try { - iface_.batch_insert_super_column(args.table, args.batch_mutation_super, args.block_for); + iface_.batch_insert_super_column(args.table, args.batch_mutation_super, args.consistency_level); } catch (InvalidRequestException ire) { result.ire = ire; } catch (UnavailableException ue) { @@ -4396,7 +4396,7 @@ public class Cassandra { private static final TField COLUMN_PATH_FIELD_DESC = new TField("column_path", TType.STRUCT, (short)3); private static final TField VALUE_FIELD_DESC = new TField("value", TType.STRING, (short)4); private static final TField TIMESTAMP_FIELD_DESC = new TField("timestamp", TType.I64, (short)5); - private static final TField BLOCK_FOR_FIELD_DESC = new TField("block_for", TType.I32, (short)6); + private static final TField CONSISTENCY_LEVEL_FIELD_DESC = new TField("consistency_level", TType.I32, (short)6); public String table; public static final int TABLE = 1; @@ -4408,13 +4408,13 @@ public class Cassandra { public static final int VALUE = 4; public long timestamp; public static final int TIMESTAMP = 5; - public int block_for; - public static final int BLOCK_FOR = 6; + public int consistency_level; + public static final int CONSISTENCY_LEVEL = 6; private final Isset __isset = new Isset(); private static final class Isset implements java.io.Serializable { public boolean timestamp = false; - public boolean block_for = false; + public boolean consistency_level = false; } public static final Map metaDataMap = Collections.unmodifiableMap(new HashMap() {{ @@ -4428,7 +4428,7 @@ public class Cassandra { new FieldValueMetaData(TType.STRING))); put(TIMESTAMP, new FieldMetaData("timestamp", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I64))); - put(BLOCK_FOR, new FieldMetaData("block_for", TFieldRequirementType.DEFAULT, + put(CONSISTENCY_LEVEL, new FieldMetaData("consistency_level", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I32))); }}); @@ -4437,7 +4437,7 @@ public class Cassandra { } public insert_args() { - this.block_for = 0; + this.consistency_level = 0; } @@ -4447,7 +4447,7 @@ public class Cassandra { ColumnPath column_path, byte[] value, long timestamp, - int block_for) + int consistency_level) { this(); this.table = table; @@ -4456,8 +4456,8 @@ public class Cassandra { this.value = value; this.timestamp = timestamp; this.__isset.timestamp = true; - this.block_for = block_for; - this.__isset.block_for = true; + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } /** @@ -4479,8 +4479,8 @@ public class Cassandra { } __isset.timestamp = other.__isset.timestamp; this.timestamp = other.timestamp; - __isset.block_for = other.__isset.block_for; - this.block_for = other.block_for; + __isset.consistency_level = other.__isset.consistency_level; + this.consistency_level = other.consistency_level; } @Override @@ -4602,26 +4602,26 @@ public class Cassandra { this.__isset.timestamp = value; } - public int getBlock_for() { - return this.block_for; + public int getConsistency_level() { + return this.consistency_level; } - public void setBlock_for(int block_for) { - this.block_for = block_for; - this.__isset.block_for = true; + public void setConsistency_level(int consistency_level) { + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } - public void unsetBlock_for() { - this.__isset.block_for = false; + public void unsetConsistency_level() { + this.__isset.consistency_level = false; } - // Returns true if field block_for is set (has been asigned a value) and false otherwise - public boolean isSetBlock_for() { - return this.__isset.block_for; + // Returns true if field consistency_level is set (has been asigned a value) and false otherwise + public boolean isSetConsistency_level() { + return this.__isset.consistency_level; } - public void setBlock_forIsSet(boolean value) { - this.__isset.block_for = value; + public void setConsistency_levelIsSet(boolean value) { + this.__isset.consistency_level = value; } public void setFieldValue(int fieldID, Object value) { @@ -4666,11 +4666,11 @@ public class Cassandra { } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (value == null) { - unsetBlock_for(); + unsetConsistency_level(); } else { - setBlock_for((Integer)value); + setConsistency_level((Integer)value); } break; @@ -4696,8 +4696,8 @@ public class Cassandra { case TIMESTAMP: return new Long(getTimestamp()); - case BLOCK_FOR: - return new Integer(getBlock_for()); + case CONSISTENCY_LEVEL: + return getConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); @@ -4717,8 +4717,8 @@ public class Cassandra { return isSetValue(); case TIMESTAMP: return isSetTimestamp(); - case BLOCK_FOR: - return isSetBlock_for(); + case CONSISTENCY_LEVEL: + return isSetConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); } @@ -4782,12 +4782,12 @@ public class Cassandra { return false; } - boolean this_present_block_for = true; - boolean that_present_block_for = true; - if (this_present_block_for || that_present_block_for) { - if (!(this_present_block_for && that_present_block_for)) + boolean this_present_consistency_level = true; + boolean that_present_consistency_level = true; + if (this_present_consistency_level || that_present_consistency_level) { + if (!(this_present_consistency_level && that_present_consistency_level)) return false; - if (this.block_for != that.block_for) + if (this.consistency_level != that.consistency_level) return false; } @@ -4847,10 +4847,10 @@ public class Cassandra { TProtocolUtil.skip(iprot, field.type); } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (field.type == TType.I32) { - this.block_for = iprot.readI32(); - this.__isset.block_for = true; + this.consistency_level = iprot.readI32(); + this.__isset.consistency_level = true; } else { TProtocolUtil.skip(iprot, field.type); } @@ -4895,8 +4895,8 @@ public class Cassandra { oprot.writeFieldBegin(TIMESTAMP_FIELD_DESC); oprot.writeI64(this.timestamp); oprot.writeFieldEnd(); - oprot.writeFieldBegin(BLOCK_FOR_FIELD_DESC); - oprot.writeI32(this.block_for); + oprot.writeFieldBegin(CONSISTENCY_LEVEL_FIELD_DESC); + oprot.writeI32(this.consistency_level); oprot.writeFieldEnd(); oprot.writeFieldStop(); oprot.writeStructEnd(); @@ -4948,8 +4948,16 @@ public class Cassandra { sb.append(this.timestamp); first = false; if (!first) sb.append(", "); - sb.append("block_for:"); - sb.append(this.block_for); + sb.append("consistency_level:"); + String consistency_level_name = ConsistencyLevel.VALUES_TO_NAMES.get(this.consistency_level); + if (consistency_level_name != null) { + sb.append(consistency_level_name); + sb.append(" ("); + } + sb.append(this.consistency_level); + if (consistency_level_name != null) { + sb.append(")"); + } first = false; sb.append(")"); return sb.toString(); @@ -4958,6 +4966,9 @@ public class Cassandra { public void validate() throws TException { // check for required fields // check that fields of type enum have valid values + if (isSetConsistency_level() && !ConsistencyLevel.VALID_VALUES.contains(consistency_level)){ + throw new TProtocolException("The field 'consistency_level' has been assigned the invalid value " + consistency_level); + } } } @@ -5240,18 +5251,18 @@ public class Cassandra { private static final TStruct STRUCT_DESC = new TStruct("batch_insert_args"); private static final TField TABLE_FIELD_DESC = new TField("table", TType.STRING, (short)1); private static final TField BATCH_MUTATION_FIELD_DESC = new TField("batch_mutation", TType.STRUCT, (short)2); - private static final TField BLOCK_FOR_FIELD_DESC = new TField("block_for", TType.I32, (short)3); + private static final TField CONSISTENCY_LEVEL_FIELD_DESC = new TField("consistency_level", TType.I32, (short)3); public String table; public static final int TABLE = 1; public BatchMutation batch_mutation; public static final int BATCH_MUTATION = 2; - public int block_for; - public static final int BLOCK_FOR = 3; + public int consistency_level; + public static final int CONSISTENCY_LEVEL = 3; private final Isset __isset = new Isset(); private static final class Isset implements java.io.Serializable { - public boolean block_for = false; + public boolean consistency_level = false; } public static final Map metaDataMap = Collections.unmodifiableMap(new HashMap() {{ @@ -5259,7 +5270,7 @@ public class Cassandra { new FieldValueMetaData(TType.STRING))); put(BATCH_MUTATION, new FieldMetaData("batch_mutation", TFieldRequirementType.DEFAULT, new StructMetaData(TType.STRUCT, BatchMutation.class))); - put(BLOCK_FOR, new FieldMetaData("block_for", TFieldRequirementType.DEFAULT, + put(CONSISTENCY_LEVEL, new FieldMetaData("consistency_level", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I32))); }}); @@ -5268,20 +5279,20 @@ public class Cassandra { } public batch_insert_args() { - this.block_for = 0; + this.consistency_level = 0; } public batch_insert_args( String table, BatchMutation batch_mutation, - int block_for) + int consistency_level) { this(); this.table = table; this.batch_mutation = batch_mutation; - this.block_for = block_for; - this.__isset.block_for = true; + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } /** @@ -5294,8 +5305,8 @@ public class Cassandra { if (other.isSetBatch_mutation()) { this.batch_mutation = new BatchMutation(other.batch_mutation); } - __isset.block_for = other.__isset.block_for; - this.block_for = other.block_for; + __isset.consistency_level = other.__isset.consistency_level; + this.consistency_level = other.consistency_level; } @Override @@ -5349,26 +5360,26 @@ public class Cassandra { } } - public int getBlock_for() { - return this.block_for; + public int getConsistency_level() { + return this.consistency_level; } - public void setBlock_for(int block_for) { - this.block_for = block_for; - this.__isset.block_for = true; + public void setConsistency_level(int consistency_level) { + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } - public void unsetBlock_for() { - this.__isset.block_for = false; + public void unsetConsistency_level() { + this.__isset.consistency_level = false; } - // Returns true if field block_for is set (has been asigned a value) and false otherwise - public boolean isSetBlock_for() { - return this.__isset.block_for; + // Returns true if field consistency_level is set (has been asigned a value) and false otherwise + public boolean isSetConsistency_level() { + return this.__isset.consistency_level; } - public void setBlock_forIsSet(boolean value) { - this.__isset.block_for = value; + public void setConsistency_levelIsSet(boolean value) { + this.__isset.consistency_level = value; } public void setFieldValue(int fieldID, Object value) { @@ -5389,11 +5400,11 @@ public class Cassandra { } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (value == null) { - unsetBlock_for(); + unsetConsistency_level(); } else { - setBlock_for((Integer)value); + setConsistency_level((Integer)value); } break; @@ -5410,8 +5421,8 @@ public class Cassandra { case BATCH_MUTATION: return getBatch_mutation(); - case BLOCK_FOR: - return new Integer(getBlock_for()); + case CONSISTENCY_LEVEL: + return getConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); @@ -5425,8 +5436,8 @@ public class Cassandra { return isSetTable(); case BATCH_MUTATION: return isSetBatch_mutation(); - case BLOCK_FOR: - return isSetBlock_for(); + case CONSISTENCY_LEVEL: + return isSetConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); } @@ -5463,12 +5474,12 @@ public class Cassandra { return false; } - boolean this_present_block_for = true; - boolean that_present_block_for = true; - if (this_present_block_for || that_present_block_for) { - if (!(this_present_block_for && that_present_block_for)) + boolean this_present_consistency_level = true; + boolean that_present_consistency_level = true; + if (this_present_consistency_level || that_present_consistency_level) { + if (!(this_present_consistency_level && that_present_consistency_level)) return false; - if (this.block_for != that.block_for) + if (this.consistency_level != that.consistency_level) return false; } @@ -5506,10 +5517,10 @@ public class Cassandra { TProtocolUtil.skip(iprot, field.type); } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (field.type == TType.I32) { - this.block_for = iprot.readI32(); - this.__isset.block_for = true; + this.consistency_level = iprot.readI32(); + this.__isset.consistency_level = true; } else { TProtocolUtil.skip(iprot, field.type); } @@ -5541,8 +5552,8 @@ public class Cassandra { this.batch_mutation.write(oprot); oprot.writeFieldEnd(); } - oprot.writeFieldBegin(BLOCK_FOR_FIELD_DESC); - oprot.writeI32(this.block_for); + oprot.writeFieldBegin(CONSISTENCY_LEVEL_FIELD_DESC); + oprot.writeI32(this.consistency_level); oprot.writeFieldEnd(); oprot.writeFieldStop(); oprot.writeStructEnd(); @@ -5569,8 +5580,16 @@ public class Cassandra { } first = false; if (!first) sb.append(", "); - sb.append("block_for:"); - sb.append(this.block_for); + sb.append("consistency_level:"); + String consistency_level_name = ConsistencyLevel.VALUES_TO_NAMES.get(this.consistency_level); + if (consistency_level_name != null) { + sb.append(consistency_level_name); + sb.append(" ("); + } + sb.append(this.consistency_level); + if (consistency_level_name != null) { + sb.append(")"); + } first = false; sb.append(")"); return sb.toString(); @@ -5579,6 +5598,9 @@ public class Cassandra { public void validate() throws TException { // check for required fields // check that fields of type enum have valid values + if (isSetConsistency_level() && !ConsistencyLevel.VALID_VALUES.contains(consistency_level)){ + throw new TProtocolException("The field 'consistency_level' has been assigned the invalid value " + consistency_level); + } } } @@ -5863,7 +5885,7 @@ public class Cassandra { private static final TField KEY_FIELD_DESC = new TField("key", TType.STRING, (short)2); private static final TField COLUMN_PATH_OR_PARENT_FIELD_DESC = new TField("column_path_or_parent", TType.STRUCT, (short)3); private static final TField TIMESTAMP_FIELD_DESC = new TField("timestamp", TType.I64, (short)4); - private static final TField BLOCK_FOR_FIELD_DESC = new TField("block_for", TType.I32, (short)5); + private static final TField CONSISTENCY_LEVEL_FIELD_DESC = new TField("consistency_level", TType.I32, (short)5); public String table; public static final int TABLE = 1; @@ -5873,13 +5895,13 @@ public class Cassandra { public static final int COLUMN_PATH_OR_PARENT = 3; public long timestamp; public static final int TIMESTAMP = 4; - public int block_for; - public static final int BLOCK_FOR = 5; + public int consistency_level; + public static final int CONSISTENCY_LEVEL = 5; private final Isset __isset = new Isset(); private static final class Isset implements java.io.Serializable { public boolean timestamp = false; - public boolean block_for = false; + public boolean consistency_level = false; } public static final Map metaDataMap = Collections.unmodifiableMap(new HashMap() {{ @@ -5891,7 +5913,7 @@ public class Cassandra { new StructMetaData(TType.STRUCT, ColumnPathOrParent.class))); put(TIMESTAMP, new FieldMetaData("timestamp", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I64))); - put(BLOCK_FOR, new FieldMetaData("block_for", TFieldRequirementType.DEFAULT, + put(CONSISTENCY_LEVEL, new FieldMetaData("consistency_level", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I32))); }}); @@ -5900,7 +5922,7 @@ public class Cassandra { } public remove_args() { - this.block_for = 0; + this.consistency_level = 0; } @@ -5909,7 +5931,7 @@ public class Cassandra { String key, ColumnPathOrParent column_path_or_parent, long timestamp, - int block_for) + int consistency_level) { this(); this.table = table; @@ -5917,8 +5939,8 @@ public class Cassandra { this.column_path_or_parent = column_path_or_parent; this.timestamp = timestamp; this.__isset.timestamp = true; - this.block_for = block_for; - this.__isset.block_for = true; + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } /** @@ -5936,8 +5958,8 @@ public class Cassandra { } __isset.timestamp = other.__isset.timestamp; this.timestamp = other.timestamp; - __isset.block_for = other.__isset.block_for; - this.block_for = other.block_for; + __isset.consistency_level = other.__isset.consistency_level; + this.consistency_level = other.consistency_level; } @Override @@ -6036,26 +6058,26 @@ public class Cassandra { this.__isset.timestamp = value; } - public int getBlock_for() { - return this.block_for; + public int getConsistency_level() { + return this.consistency_level; } - public void setBlock_for(int block_for) { - this.block_for = block_for; - this.__isset.block_for = true; + public void setConsistency_level(int consistency_level) { + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } - public void unsetBlock_for() { - this.__isset.block_for = false; + public void unsetConsistency_level() { + this.__isset.consistency_level = false; } - // Returns true if field block_for is set (has been asigned a value) and false otherwise - public boolean isSetBlock_for() { - return this.__isset.block_for; + // Returns true if field consistency_level is set (has been asigned a value) and false otherwise + public boolean isSetConsistency_level() { + return this.__isset.consistency_level; } - public void setBlock_forIsSet(boolean value) { - this.__isset.block_for = value; + public void setConsistency_levelIsSet(boolean value) { + this.__isset.consistency_level = value; } public void setFieldValue(int fieldID, Object value) { @@ -6092,11 +6114,11 @@ public class Cassandra { } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (value == null) { - unsetBlock_for(); + unsetConsistency_level(); } else { - setBlock_for((Integer)value); + setConsistency_level((Integer)value); } break; @@ -6119,8 +6141,8 @@ public class Cassandra { case TIMESTAMP: return new Long(getTimestamp()); - case BLOCK_FOR: - return new Integer(getBlock_for()); + case CONSISTENCY_LEVEL: + return getConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); @@ -6138,8 +6160,8 @@ public class Cassandra { return isSetColumn_path_or_parent(); case TIMESTAMP: return isSetTimestamp(); - case BLOCK_FOR: - return isSetBlock_for(); + case CONSISTENCY_LEVEL: + return isSetConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); } @@ -6194,12 +6216,12 @@ public class Cassandra { return false; } - boolean this_present_block_for = true; - boolean that_present_block_for = true; - if (this_present_block_for || that_present_block_for) { - if (!(this_present_block_for && that_present_block_for)) + boolean this_present_consistency_level = true; + boolean that_present_consistency_level = true; + if (this_present_consistency_level || that_present_consistency_level) { + if (!(this_present_consistency_level && that_present_consistency_level)) return false; - if (this.block_for != that.block_for) + if (this.consistency_level != that.consistency_level) return false; } @@ -6252,10 +6274,10 @@ public class Cassandra { TProtocolUtil.skip(iprot, field.type); } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (field.type == TType.I32) { - this.block_for = iprot.readI32(); - this.__isset.block_for = true; + this.consistency_level = iprot.readI32(); + this.__isset.consistency_level = true; } else { TProtocolUtil.skip(iprot, field.type); } @@ -6295,8 +6317,8 @@ public class Cassandra { oprot.writeFieldBegin(TIMESTAMP_FIELD_DESC); oprot.writeI64(this.timestamp); oprot.writeFieldEnd(); - oprot.writeFieldBegin(BLOCK_FOR_FIELD_DESC); - oprot.writeI32(this.block_for); + oprot.writeFieldBegin(CONSISTENCY_LEVEL_FIELD_DESC); + oprot.writeI32(this.consistency_level); oprot.writeFieldEnd(); oprot.writeFieldStop(); oprot.writeStructEnd(); @@ -6335,8 +6357,16 @@ public class Cassandra { sb.append(this.timestamp); first = false; if (!first) sb.append(", "); - sb.append("block_for:"); - sb.append(this.block_for); + sb.append("consistency_level:"); + String consistency_level_name = ConsistencyLevel.VALUES_TO_NAMES.get(this.consistency_level); + if (consistency_level_name != null) { + sb.append(consistency_level_name); + sb.append(" ("); + } + sb.append(this.consistency_level); + if (consistency_level_name != null) { + sb.append(")"); + } first = false; sb.append(")"); return sb.toString(); @@ -6345,6 +6375,9 @@ public class Cassandra { public void validate() throws TException { // check for required fields // check that fields of type enum have valid values + if (isSetConsistency_level() && !ConsistencyLevel.VALID_VALUES.contains(consistency_level)){ + throw new TProtocolException("The field 'consistency_level' has been assigned the invalid value " + consistency_level); + } } } @@ -9061,18 +9094,18 @@ public class Cassandra { private static final TStruct STRUCT_DESC = new TStruct("batch_insert_super_column_args"); private static final TField TABLE_FIELD_DESC = new TField("table", TType.STRING, (short)1); private static final TField BATCH_MUTATION_SUPER_FIELD_DESC = new TField("batch_mutation_super", TType.STRUCT, (short)2); - private static final TField BLOCK_FOR_FIELD_DESC = new TField("block_for", TType.I32, (short)3); + private static final TField CONSISTENCY_LEVEL_FIELD_DESC = new TField("consistency_level", TType.I32, (short)3); public String table; public static final int TABLE = 1; public BatchMutationSuper batch_mutation_super; public static final int BATCH_MUTATION_SUPER = 2; - public int block_for; - public static final int BLOCK_FOR = 3; + public int consistency_level; + public static final int CONSISTENCY_LEVEL = 3; private final Isset __isset = new Isset(); private static final class Isset implements java.io.Serializable { - public boolean block_for = false; + public boolean consistency_level = false; } public static final Map metaDataMap = Collections.unmodifiableMap(new HashMap() {{ @@ -9080,7 +9113,7 @@ public class Cassandra { new FieldValueMetaData(TType.STRING))); put(BATCH_MUTATION_SUPER, new FieldMetaData("batch_mutation_super", TFieldRequirementType.DEFAULT, new StructMetaData(TType.STRUCT, BatchMutationSuper.class))); - put(BLOCK_FOR, new FieldMetaData("block_for", TFieldRequirementType.DEFAULT, + put(CONSISTENCY_LEVEL, new FieldMetaData("consistency_level", TFieldRequirementType.DEFAULT, new FieldValueMetaData(TType.I32))); }}); @@ -9089,20 +9122,20 @@ public class Cassandra { } public batch_insert_super_column_args() { - this.block_for = 0; + this.consistency_level = 0; } public batch_insert_super_column_args( String table, BatchMutationSuper batch_mutation_super, - int block_for) + int consistency_level) { this(); this.table = table; this.batch_mutation_super = batch_mutation_super; - this.block_for = block_for; - this.__isset.block_for = true; + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } /** @@ -9115,8 +9148,8 @@ public class Cassandra { if (other.isSetBatch_mutation_super()) { this.batch_mutation_super = new BatchMutationSuper(other.batch_mutation_super); } - __isset.block_for = other.__isset.block_for; - this.block_for = other.block_for; + __isset.consistency_level = other.__isset.consistency_level; + this.consistency_level = other.consistency_level; } @Override @@ -9170,26 +9203,26 @@ public class Cassandra { } } - public int getBlock_for() { - return this.block_for; + public int getConsistency_level() { + return this.consistency_level; } - public void setBlock_for(int block_for) { - this.block_for = block_for; - this.__isset.block_for = true; + public void setConsistency_level(int consistency_level) { + this.consistency_level = consistency_level; + this.__isset.consistency_level = true; } - public void unsetBlock_for() { - this.__isset.block_for = false; + public void unsetConsistency_level() { + this.__isset.consistency_level = false; } - // Returns true if field block_for is set (has been asigned a value) and false otherwise - public boolean isSetBlock_for() { - return this.__isset.block_for; + // Returns true if field consistency_level is set (has been asigned a value) and false otherwise + public boolean isSetConsistency_level() { + return this.__isset.consistency_level; } - public void setBlock_forIsSet(boolean value) { - this.__isset.block_for = value; + public void setConsistency_levelIsSet(boolean value) { + this.__isset.consistency_level = value; } public void setFieldValue(int fieldID, Object value) { @@ -9210,11 +9243,11 @@ public class Cassandra { } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (value == null) { - unsetBlock_for(); + unsetConsistency_level(); } else { - setBlock_for((Integer)value); + setConsistency_level((Integer)value); } break; @@ -9231,8 +9264,8 @@ public class Cassandra { case BATCH_MUTATION_SUPER: return getBatch_mutation_super(); - case BLOCK_FOR: - return new Integer(getBlock_for()); + case CONSISTENCY_LEVEL: + return getConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); @@ -9246,8 +9279,8 @@ public class Cassandra { return isSetTable(); case BATCH_MUTATION_SUPER: return isSetBatch_mutation_super(); - case BLOCK_FOR: - return isSetBlock_for(); + case CONSISTENCY_LEVEL: + return isSetConsistency_level(); default: throw new IllegalArgumentException("Field " + fieldID + " doesn't exist!"); } @@ -9284,12 +9317,12 @@ public class Cassandra { return false; } - boolean this_present_block_for = true; - boolean that_present_block_for = true; - if (this_present_block_for || that_present_block_for) { - if (!(this_present_block_for && that_present_block_for)) + boolean this_present_consistency_level = true; + boolean that_present_consistency_level = true; + if (this_present_consistency_level || that_present_consistency_level) { + if (!(this_present_consistency_level && that_present_consistency_level)) return false; - if (this.block_for != that.block_for) + if (this.consistency_level != that.consistency_level) return false; } @@ -9327,10 +9360,10 @@ public class Cassandra { TProtocolUtil.skip(iprot, field.type); } break; - case BLOCK_FOR: + case CONSISTENCY_LEVEL: if (field.type == TType.I32) { - this.block_for = iprot.readI32(); - this.__isset.block_for = true; + this.consistency_level = iprot.readI32(); + this.__isset.consistency_level = true; } else { TProtocolUtil.skip(iprot, field.type); } @@ -9362,8 +9395,8 @@ public class Cassandra { this.batch_mutation_super.write(oprot); oprot.writeFieldEnd(); } - oprot.writeFieldBegin(BLOCK_FOR_FIELD_DESC); - oprot.writeI32(this.block_for); + oprot.writeFieldBegin(CONSISTENCY_LEVEL_FIELD_DESC); + oprot.writeI32(this.consistency_level); oprot.writeFieldEnd(); oprot.writeFieldStop(); oprot.writeStructEnd(); @@ -9390,8 +9423,16 @@ public class Cassandra { } first = false; if (!first) sb.append(", "); - sb.append("block_for:"); - sb.append(this.block_for); + sb.append("consistency_level:"); + String consistency_level_name = ConsistencyLevel.VALUES_TO_NAMES.get(this.consistency_level); + if (consistency_level_name != null) { + sb.append(consistency_level_name); + sb.append(" ("); + } + sb.append(this.consistency_level); + if (consistency_level_name != null) { + sb.append(")"); + } first = false; sb.append(")"); return sb.toString(); @@ -9400,6 +9441,9 @@ public class Cassandra { public void validate() throws TException { // check for required fields // check that fields of type enum have valid values + if (isSetConsistency_level() && !ConsistencyLevel.VALID_VALUES.contains(consistency_level)){ + throw new TProtocolException("The field 'consistency_level' has been assigned the invalid value " + consistency_level); + } } } diff --git a/interface/gen-java/org/apache/cassandra/service/ConsistencyLevel.java b/interface/gen-java/org/apache/cassandra/service/ConsistencyLevel.java new file mode 100644 index 0000000000..0ce14bd32f --- /dev/null +++ b/interface/gen-java/org/apache/cassandra/service/ConsistencyLevel.java @@ -0,0 +1,29 @@ +/** + * Autogenerated by Thrift + * + * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING + */ +package org.apache.cassandra.service; + + +import java.util.Set; +import java.util.HashSet; +import java.util.Collections; +import org.apache.thrift.IntRangeSet; +import java.util.Map; +import java.util.HashMap; + +public class ConsistencyLevel { + public static final int ZERO = 0; + public static final int ONE = 1; + public static final int QUORUM = 2; + public static final int ALL = 3; + + public static final IntRangeSet VALID_VALUES = new IntRangeSet(ZERO, ONE, QUORUM, ALL); + public static final Map VALUES_TO_NAMES = new HashMap() {{ + put(ZERO, "ZERO"); + put(ONE, "ONE"); + put(QUORUM, "QUORUM"); + put(ALL, "ALL"); + }}; +} diff --git a/src/java/org/apache/cassandra/cli/CliClient.java b/src/java/org/apache/cassandra/cli/CliClient.java index ef39395881..492003d049 100644 --- a/src/java/org/apache/cassandra/cli/CliClient.java +++ b/src/java/org/apache/cassandra/cli/CliClient.java @@ -197,7 +197,7 @@ public class CliClient try { thriftClient_.insert(tableName, key, new ColumnPath(columnFamily, null, columnName.getBytes("UTF-8")), - value.getBytes(), System.currentTimeMillis(), 1); + value.getBytes(), System.currentTimeMillis(), ConsistencyLevel.ONE); } catch (UnsupportedEncodingException e) { diff --git a/src/java/org/apache/cassandra/service/CassandraServer.java b/src/java/org/apache/cassandra/service/CassandraServer.java index 8f0500982e..90d113b1eb 100644 --- a/src/java/org/apache/cassandra/service/CassandraServer.java +++ b/src/java/org/apache/cassandra/service/CassandraServer.java @@ -35,6 +35,7 @@ import org.apache.cassandra.cql.driver.CqlDriver; import org.apache.cassandra.db.*; import org.apache.cassandra.db.marshal.MarshalException; import org.apache.cassandra.db.filter.QueryPath; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.LogUtil; import org.apache.cassandra.dht.OrderPreservingPartitioner; import org.apache.thrift.TException; @@ -248,7 +249,7 @@ public class CassandraServer implements Cassandra.Iface return columns.size(); } - public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int block_for) + public void insert(String table, String key, ColumnPath column_path, byte[] value, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException { logger.debug("insert"); @@ -264,10 +265,10 @@ public class CassandraServer implements Cassandra.Iface { throw new InvalidRequestException(e.getMessage()); } - doInsert(block_for, rm); + doInsert(consistency_level, rm); } - public void batch_insert(String table, BatchMutation batch_mutation, int block_for) + public void batch_insert(String table, BatchMutation batch_mutation, int consistency_level) throws InvalidRequestException, UnavailableException { logger.debug("batch_insert"); @@ -275,10 +276,10 @@ public class CassandraServer implements Cassandra.Iface Set cfNames = rm.columnFamilyNames(); ThriftValidation.validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()])); - doInsert(block_for, rm); + doInsert(consistency_level, rm); } - public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int block_for) + public void remove(String table, String key, ColumnPathOrParent column_path_or_parent, long timestamp, int consistency_level) throws InvalidRequestException, UnavailableException { logger.debug("remove"); @@ -287,14 +288,14 @@ public class CassandraServer implements Cassandra.Iface RowMutation rm = new RowMutation(table, key.trim()); rm.delete(new QueryPath(column_path_or_parent), timestamp); - doInsert(block_for, rm); + doInsert(consistency_level, rm); } - private void doInsert(int block, RowMutation rm) throws UnavailableException + private void doInsert(int consistency_level, RowMutation rm) throws UnavailableException { - if (block > 0) + if (consistency_level != ConsistencyLevel.ZERO) { - StorageProxy.insertBlocking(rm,block); + StorageProxy.insertBlocking(rm, consistency_level); } else { @@ -391,7 +392,7 @@ public class CassandraServer implements Cassandra.Iface return new SuperColumn(column.name(), thriftifyColumns(column.getSubColumns())); } - public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int block_for) + public void batch_insert_super_column(String table, BatchMutationSuper batch_mutation_super, int consistency_level) throws InvalidRequestException, UnavailableException { logger.debug("batch_insert_SuperColumn"); @@ -399,7 +400,7 @@ public class CassandraServer implements Cassandra.Iface Set cfNames = rm.columnFamilyNames(); ThriftValidation.validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()])); - doInsert(block_for, rm); + doInsert(consistency_level, rm); } public String get_string_property(String propertyName) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index 617f17e641..d2b62dcffa 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -150,7 +150,7 @@ public class StorageProxy implements StorageProxyMBean } } - public static void insertBlocking(RowMutation rm, int blockFor) throws UnavailableException + public static void insertBlocking(RowMutation rm, int consistency_level) throws UnavailableException { long startTime = System.currentTimeMillis(); Message message = null; @@ -169,6 +169,23 @@ public class StorageProxy implements StorageProxyMBean { throw new UnavailableException(); } + int blockFor; + if (consistency_level == ConsistencyLevel.ONE) + { + blockFor = 1; + } + else if (consistency_level == ConsistencyLevel.QUORUM) + { + blockFor = (DatabaseDescriptor.getReplicationFactor() >> 1) + 1; + } + else if (consistency_level == ConsistencyLevel.ALL) + { + blockFor = DatabaseDescriptor.getReplicationFactor(); + } + else + { + throw new UnsupportedOperationException("invalid consistency level " + consistency_level); + } QuorumResponseHandler quorumResponseHandler = new QuorumResponseHandler(blockFor, new WriteResponseResolver()); logger.debug("insertBlocking writing key " + rm.key() + " to " + message.getMessageId() + "@[" + StringUtils.join(endpoints, ", ") + "]"); @@ -189,7 +206,7 @@ public class StorageProxy implements StorageProxyMBean public static void insertBlocking(RowMutation rm) throws UnavailableException { - insertBlocking(rm, (DatabaseDescriptor.getReplicationFactor() >> 1) + 1); + insertBlocking(rm, ConsistencyLevel.QUORUM); } private static Map constructMessages(Map readMessages) throws IOException diff --git a/test/system/test_server.py b/test/system/test_server.py index 65954964d2..5f2232d2eb 100644 --- a/test/system/test_server.py +++ b/test/system/test_server.py @@ -35,13 +35,22 @@ _SUPER_COLUMNS = [SuperColumn(name='sc1', columns=[Column(_i64(4), 'value4', 0)] Column(_i64(6), 'value6', 0)])] def _insert_simple(block=True): - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, block) - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c2'), 'value2', 0, block) + if block: + consistencyLevel = ConsistencyLevel.ONE + else: + consistencyLevel = ConsistencyLevel.ZERO + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, consistencyLevel) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c2'), 'value2', 0, consistencyLevel) def _insert_batch(block): cfmap = {'Standard1': _SIMPLE_COLUMNS, 'Standard2': _SIMPLE_COLUMNS} - client.batch_insert('Table1', BatchMutation(key='key1', cfmap=cfmap), block) + if block: + consistencyLevel = ConsistencyLevel.ONE + else: + consistencyLevel = ConsistencyLevel.ZERO + + client.batch_insert('Table1', BatchMutation(key='key1', cfmap=cfmap), consistencyLevel) def _verify_batch(): _verify_simple() @@ -54,15 +63,15 @@ def _verify_simple(): assert L == _SIMPLE_COLUMNS, L def _insert_super(): - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc1', _i64(4)), 'value4', 0, False) - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, False) - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(6)), 'value6', 0, False) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc1', _i64(4)), 'value4', 0, ConsistencyLevel.ZERO) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, ConsistencyLevel.ZERO) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(6)), 'value6', 0, ConsistencyLevel.ZERO) time.sleep(0.1) def _insert_range(): - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, True) - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c2'), 'value2', 0, True) - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c3'), 'value3', 0, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, ConsistencyLevel.ONE) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c2'), 'value2', 0, ConsistencyLevel.ONE) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c3'), 'value3', 0, ConsistencyLevel.ONE) time.sleep(0.1) def _verify_range(): @@ -157,7 +166,7 @@ class TestMutations(CassandraTester): _verify_batch() def test_bad_calls(self): - _expect_exception(lambda: client.insert('Table1', 'key1', ColumnPath('Standard1', 'x', 'y'), 'value', 0, True), InvalidRequestException) + _expect_exception(lambda: client.insert('Table1', 'key1', ColumnPath('Standard1', 'x', 'y'), 'value', 0, ConsistencyLevel.ONE), InvalidRequestException) _expect_exception(lambda: client.get_column('Table1', 'key1', ColumnPath('Standard1')), InvalidRequestException) _expect_exception(lambda: client.get_column('Table1', 'key1', ColumnPath('Standard1', 'x', 'y')), InvalidRequestException) @@ -168,7 +177,7 @@ class TestMutations(CassandraTester): def test_batch_insert_super(self): cfmap = {'Super1': _SUPER_COLUMNS, 'Super2': _SUPER_COLUMNS} - client.batch_insert_super_column('Table1', BatchMutation(key='key1', cfmap=cfmap), False) + client.batch_insert_super_column('Table1', BatchMutation(key='key1', cfmap=cfmap), ConsistencyLevel.ZERO) time.sleep(0.1) _verify_super('Super1') _verify_super('Super2') @@ -176,13 +185,13 @@ class TestMutations(CassandraTester): def test_batch_insert_super_blocking(self): cfmap = {'Super1': _SUPER_COLUMNS, 'Super2': _SUPER_COLUMNS} - client.batch_insert_super_column('Table1', BatchMutation(key='key1', cfmap=cfmap), True) + client.batch_insert_super_column('Table1', BatchMutation(key='key1', cfmap=cfmap), ConsistencyLevel.ONE) _verify_super('Super1') _verify_super('Super2') def test_cf_remove_column(self): _insert_simple() - client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c1'), 1, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c1'), 1, ConsistencyLevel.ONE) _expect_missing(lambda: client.get_column('Table1', 'key1', ColumnPath('Standard1', column='c1'))) assert client.get_column('Table1', 'key1', ColumnPath('Standard1', column='c2')) \ == Column('c2', 'value2', 0) @@ -190,17 +199,17 @@ class TestMutations(CassandraTester): == [Column('c2', 'value2', 0)] # New insert, make sure it shows up post-remove: - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c3'), 'value3', 0, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c3'), 'value3', 0, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == \ [Column('c2', 'value2', 0), Column('c3', 'value3', 0)] # Test resurrection. First, re-insert the value w/ older timestamp, # and make sure it stays removed - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == \ [Column('c2', 'value2', 0), Column('c3', 'value3', 0)] # Next, w/ a newer timestamp; it should come back: - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 2, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 2, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == \ [Column('c1', 'value1', 2), Column('c2', 'value2', 0), Column('c3', 'value3', 0)] @@ -210,16 +219,16 @@ class TestMutations(CassandraTester): _insert_super() # Remove the key1:Standard1 cf: - client.remove('Table1', 'key1', ColumnPathOrParent('Standard1'), 3, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Standard1'), 3, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == [] _verify_super() # Test resurrection. First, re-insert a value w/ older timestamp, # and make sure it stays removed: - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 0, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == [] # Next, w/ a newer timestamp; it should come back: - client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 4, True) + client.insert('Table1', 'key1', ColumnPath('Standard1', column='c1'), 'value1', 4, ConsistencyLevel.ONE) assert client.get_slice('Table1', 'key1', ColumnParent('Standard1'), '', '', True, 1000) == \ [Column('c1', 'value1', 4)] @@ -229,7 +238,7 @@ class TestMutations(CassandraTester): _insert_super() # Make sure remove clears out what it's supposed to, and _only_ that: - client.remove('Table1', 'key1', ColumnPathOrParent('Super1', 'sc2', _i64(5)), 5, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Super1', 'sc2', _i64(5)), 5, ConsistencyLevel.ONE) _expect_missing(lambda: client.get_column('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)))) assert client.get_slice_super('Table1', 'key1', 'Super1', '', '', True, 1000) == \ [SuperColumn(name='sc1', columns=[Column(_i64(4), 'value4', 0)]), @@ -237,7 +246,7 @@ class TestMutations(CassandraTester): _verify_simple() # New insert, make sure it shows up post-remove: - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(7)), 'value7', 0, True) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(7)), 'value7', 0, ConsistencyLevel.ONE) scs = [SuperColumn(name='sc1', columns=[Column(_i64(4), 'value4', 0)]), SuperColumn(name='sc2', @@ -248,12 +257,13 @@ class TestMutations(CassandraTester): # Test resurrection. First, re-insert the value w/ older timestamp, # and make sure it stays removed: - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, True) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, ConsistencyLevel.ONE) + actual = client.get_slice_super('Table1', 'key1', 'Super1', '', '', True, 1000) assert actual == scs, actual # Next, w/ a newer timestamp; it should come back - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 6, True) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 6, ConsistencyLevel.ONE) actual = client.get_slice_super('Table1', 'key1', 'Super1', '', '', True, 1000) assert actual == \ [SuperColumn(name='sc1', columns=[Column(_i64(4), 'value4', 0)]), @@ -266,7 +276,7 @@ class TestMutations(CassandraTester): _insert_super() # Make sure remove clears out what it's supposed to, and _only_ that: - client.remove('Table1', 'key1', ColumnPathOrParent('Super1', 'sc2'), 5, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Super1', 'sc2'), 5, ConsistencyLevel.ONE) _expect_missing(lambda: client.get_column('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)))) actual = client.get_slice('Table1', 'key1', ColumnParent('Super1', 'sc2'), '', '', True, 1000) assert actual == [], actual @@ -277,12 +287,12 @@ class TestMutations(CassandraTester): # Test resurrection. First, re-insert the value w/ older timestamp, # and make sure it stays removed: - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, True) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 0, ConsistencyLevel.ONE) actual = client.get_slice_super('Table1', 'key1', 'Super1', '', '', True, 1000) assert actual == scs, actual # Next, w/ a newer timestamp; it should come back - client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 6, True) + client.insert('Table1', 'key1', ColumnPath('Super1', 'sc2', _i64(5)), 'value5', 6, ConsistencyLevel.ONE) actual = client.get_slice_super('Table1', 'key1', 'Super1', '', '', True, 1000) assert actual == \ [SuperColumn(name='sc1', columns=[Column(_i64(4), 'value4', 0)]), @@ -298,8 +308,8 @@ class TestMutations(CassandraTester): _insert_simple() assert client.get_key_range('Table1', 'Standard1', 'key1', '', 1000) == ['key1'] - client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c1'), 1, True) - client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c2'), 1, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c1'), 1, ConsistencyLevel.ONE) + client.remove('Table1', 'key1', ColumnPathOrParent('Standard1', column='c2'), 1, ConsistencyLevel.ONE) actual = client.get_key_range('Table1', 'Standard1', '', '', 1000) assert actual == [], actual @@ -307,20 +317,20 @@ class TestMutations(CassandraTester): _insert_simple() assert client.get_key_range('Table1', 'Standard1', 'key1', '', 1000) == ['key1'] - client.remove('Table1', 'key1', ColumnPathOrParent('Standard1'), 1, True) + client.remove('Table1', 'key1', ColumnPathOrParent('Standard1'), 1, ConsistencyLevel.ONE) actual = client.get_key_range('Table1', 'Standard1', '', '', 1000) assert actual == [], actual def test_range_collation(self): for key in ['-a', '-b', 'a', 'b'] + [str(i) for i in xrange(100)]: - client.insert('Table1', key, ColumnPath('Standard1', column=key), 'v', 0, True) + client.insert('Table1', key, ColumnPath('Standard1', column=key), 'v', 0, ConsistencyLevel.ONE) L = client.get_key_range('Table1', 'Standard1', '', '', 1000) # note the collated ordering rather than ascii assert L == ['0', '1', '10', '11', '12', '13', '14', '15', '16', '17', '18', '19', '2', '20', '21', '22', '23', '24', '25', '26', '27','28', '29', '3', '30', '31', '32', '33', '34', '35', '36', '37', '38', '39', '4', '40', '41', '42', '43', '44', '45', '46', '47', '48', '49', '5', '50', '51', '52', '53', '54', '55', '56', '57', '58', '59', '6', '60', '61', '62', '63', '64', '65', '66', '67', '68', '69', '7', '70', '71', '72', '73', '74', '75', '76', '77', '78', '79', '8', '80', '81', '82', '83', '84', '85', '86', '87', '88', '89', '9', '90', '91', '92', '93', '94', '95', '96', '97', '98', '99', 'a', '-a', 'b', '-b'], L def test_range_partial(self): for key in ['-a', '-b', 'a', 'b'] + [str(i) for i in xrange(100)]: - client.insert('Table1', key, ColumnPath('Standard1', column=key), 'v', 0, True) + client.insert('Table1', key, ColumnPath('Standard1', column=key), 'v', 0, ConsistencyLevel.ONE) L = client.get_key_range('Table1', 'Standard1', 'a', '', 1000) assert L == ['a', '-a', 'b', '-b'], L