diff --git a/CHANGES.txt b/CHANGES.txt index 96dd94ae54..db75758e31 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -16,6 +16,7 @@ dev * add support for GT/E, LT/E in subordinate index clauses (CASSANDRA-1401) * cfId counter got out of sync when CFs were added (CASSANDRA-1403) * less chatty schema updates (CASSANDRA-1389) + * add optional per-connection read-your-writes consistency (CASSANDRA-876) 0.7-beta1 diff --git a/NEWS.txt b/NEWS.txt index b60b11c14c..7951c859f2 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -17,6 +17,7 @@ Features - `truncate` thrift method allows clearing an entire ColumnFamily at once - Hadoop OutputFormat support - Up to 8x faster reads from row cache + - Optional per-connection read-your-writes consistency - A new ByteOrderedPartitioner supports bytes keys with arbitrary content, and orders keys by their byte value. This should be used in new deployments instead of OrderPreservingPartitioner. diff --git a/interface/cassandra.thrift b/interface/cassandra.thrift index edc129fc9c..0be712fdc6 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -385,6 +385,13 @@ service Cassandra { # set keyspace void set_keyspace(1: required string keyspace) throws (1:InvalidRequestException ire), + # enable session consistency + # call again to start a new session + void enable_session_consistency(), + + # disable a previously enabled session + void disable_session_consistency(), + # retrieval methods /** 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 5d24ffaae1..84b7830e56 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java @@ -54,6 +54,10 @@ public class Cassandra { public void set_keyspace(String keyspace) throws InvalidRequestException, TException; + public void enable_session_consistency() throws TException; + + public void disable_session_consistency() throws TException; + /** * Get the Column or SuperColumn at the given column_path. If no value is present, NotFoundException is thrown. (This is * the only method that can throw an exception under non-failure conditions.) @@ -287,6 +291,10 @@ public class Cassandra { public void set_keyspace(String keyspace, AsyncMethodCallback resultHandler) throws TException; + public void enable_session_consistency(AsyncMethodCallback resultHandler) throws TException; + + public void disable_session_consistency(AsyncMethodCallback resultHandler) throws TException; + public void get(byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level, AsyncMethodCallback resultHandler) throws TException; public void get_slice(byte[] key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level, AsyncMethodCallback resultHandler) throws TException; @@ -454,6 +462,70 @@ public class Cassandra { return; } + public void enable_session_consistency() throws TException + { + send_enable_session_consistency(); + recv_enable_session_consistency(); + } + + public void send_enable_session_consistency() throws TException + { + oprot_.writeMessageBegin(new TMessage("enable_session_consistency", TMessageType.CALL, ++seqid_)); + enable_session_consistency_args args = new enable_session_consistency_args(); + args.write(oprot_); + oprot_.writeMessageEnd(); + oprot_.getTransport().flush(); + } + + public void recv_enable_session_consistency() throws TException + { + TMessage msg = iprot_.readMessageBegin(); + if (msg.type == TMessageType.EXCEPTION) { + TApplicationException x = TApplicationException.read(iprot_); + iprot_.readMessageEnd(); + throw x; + } + if (msg.seqid != seqid_) { + throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, "enable_session_consistency failed: out of sequence response"); + } + enable_session_consistency_result result = new enable_session_consistency_result(); + result.read(iprot_); + iprot_.readMessageEnd(); + return; + } + + public void disable_session_consistency() throws TException + { + send_disable_session_consistency(); + recv_disable_session_consistency(); + } + + public void send_disable_session_consistency() throws TException + { + oprot_.writeMessageBegin(new TMessage("disable_session_consistency", TMessageType.CALL, ++seqid_)); + disable_session_consistency_args args = new disable_session_consistency_args(); + args.write(oprot_); + oprot_.writeMessageEnd(); + oprot_.getTransport().flush(); + } + + public void recv_disable_session_consistency() throws TException + { + TMessage msg = iprot_.readMessageBegin(); + if (msg.type == TMessageType.EXCEPTION) { + TApplicationException x = TApplicationException.read(iprot_); + iprot_.readMessageEnd(); + throw x; + } + if (msg.seqid != seqid_) { + throw new TApplicationException(TApplicationException.BAD_SEQUENCE_ID, "disable_session_consistency failed: out of sequence response"); + } + disable_session_consistency_result result = new disable_session_consistency_result(); + result.read(iprot_); + iprot_.readMessageEnd(); + return; + } + public ColumnOrSuperColumn get(byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level) throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException { send_get(key, column_path, consistency_level); @@ -1577,6 +1649,62 @@ public class Cassandra { } } + public void enable_session_consistency(AsyncMethodCallback resultHandler) throws TException { + checkReady(); + enable_session_consistency_call method_call = new enable_session_consistency_call(resultHandler, this, protocolFactory, transport); + manager.call(method_call); + } + + public static class enable_session_consistency_call extends TAsyncMethodCall { + public enable_session_consistency_call(AsyncMethodCallback resultHandler, TAsyncClient client, TProtocolFactory protocolFactory, TNonblockingTransport transport) throws TException { + super(client, protocolFactory, transport, resultHandler, false); + } + + public void write_args(TProtocol prot) throws TException { + prot.writeMessageBegin(new TMessage("enable_session_consistency", TMessageType.CALL, 0)); + enable_session_consistency_args args = new enable_session_consistency_args(); + args.write(prot); + prot.writeMessageEnd(); + } + + public void getResult() throws TException { + if (getState() != State.RESPONSE_READ) { + throw new IllegalStateException("Method call not finished!"); + } + TMemoryInputTransport memoryTransport = new TMemoryInputTransport(getFrameBuffer().array()); + TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport); + (new Client(prot)).recv_enable_session_consistency(); + } + } + + public void disable_session_consistency(AsyncMethodCallback resultHandler) throws TException { + checkReady(); + disable_session_consistency_call method_call = new disable_session_consistency_call(resultHandler, this, protocolFactory, transport); + manager.call(method_call); + } + + public static class disable_session_consistency_call extends TAsyncMethodCall { + public disable_session_consistency_call(AsyncMethodCallback resultHandler, TAsyncClient client, TProtocolFactory protocolFactory, TNonblockingTransport transport) throws TException { + super(client, protocolFactory, transport, resultHandler, false); + } + + public void write_args(TProtocol prot) throws TException { + prot.writeMessageBegin(new TMessage("disable_session_consistency", TMessageType.CALL, 0)); + disable_session_consistency_args args = new disable_session_consistency_args(); + args.write(prot); + prot.writeMessageEnd(); + } + + public void getResult() throws TException { + if (getState() != State.RESPONSE_READ) { + throw new IllegalStateException("Method call not finished!"); + } + TMemoryInputTransport memoryTransport = new TMemoryInputTransport(getFrameBuffer().array()); + TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport); + (new Client(prot)).recv_disable_session_consistency(); + } + } + public void get(byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level, AsyncMethodCallback resultHandler) throws TException { checkReady(); get_call method_call = new get_call(key, column_path, consistency_level, resultHandler, this, protocolFactory, transport); @@ -2448,6 +2576,8 @@ public class Cassandra { iface_ = iface; processMap_.put("login", new login()); processMap_.put("set_keyspace", new set_keyspace()); + processMap_.put("enable_session_consistency", new enable_session_consistency()); + processMap_.put("disable_session_consistency", new disable_session_consistency()); processMap_.put("get", new get()); processMap_.put("get_slice", new get_slice()); processMap_.put("get_count", new get_count()); @@ -2578,6 +2708,58 @@ public class Cassandra { } + private class enable_session_consistency implements ProcessFunction { + public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException + { + enable_session_consistency_args args = new enable_session_consistency_args(); + try { + args.read(iprot); + } catch (TProtocolException e) { + iprot.readMessageEnd(); + TApplicationException x = new TApplicationException(TApplicationException.PROTOCOL_ERROR, e.getMessage()); + oprot.writeMessageBegin(new TMessage("enable_session_consistency", TMessageType.EXCEPTION, seqid)); + x.write(oprot); + oprot.writeMessageEnd(); + oprot.getTransport().flush(); + return; + } + iprot.readMessageEnd(); + enable_session_consistency_result result = new enable_session_consistency_result(); + iface_.enable_session_consistency(); + oprot.writeMessageBegin(new TMessage("enable_session_consistency", TMessageType.REPLY, seqid)); + result.write(oprot); + oprot.writeMessageEnd(); + oprot.getTransport().flush(); + } + + } + + private class disable_session_consistency implements ProcessFunction { + public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException + { + disable_session_consistency_args args = new disable_session_consistency_args(); + try { + args.read(iprot); + } catch (TProtocolException e) { + iprot.readMessageEnd(); + TApplicationException x = new TApplicationException(TApplicationException.PROTOCOL_ERROR, e.getMessage()); + oprot.writeMessageBegin(new TMessage("disable_session_consistency", TMessageType.EXCEPTION, seqid)); + x.write(oprot); + oprot.writeMessageEnd(); + oprot.getTransport().flush(); + return; + } + iprot.readMessageEnd(); + disable_session_consistency_result result = new disable_session_consistency_result(); + iface_.disable_session_consistency(); + oprot.writeMessageBegin(new TMessage("disable_session_consistency", TMessageType.REPLY, seqid)); + result.write(oprot); + oprot.writeMessageEnd(); + oprot.getTransport().flush(); + } + + } + private class get implements ProcessFunction { public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException { @@ -4847,6 +5029,768 @@ public class Cassandra { } + public static class enable_session_consistency_args implements TBase, java.io.Serializable, Cloneable { + private static final TStruct STRUCT_DESC = new TStruct("enable_session_consistency_args"); + + + + /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ + public enum _Fields implements TFieldIdEnum { +; + + private static final Map byName = new HashMap(); + + static { + for (_Fields field : EnumSet.allOf(_Fields.class)) { + byName.put(field.getFieldName(), field); + } + } + + /** + * Find the _Fields constant that matches fieldId, or null if its not found. + */ + public static _Fields findByThriftId(int fieldId) { + switch(fieldId) { + default: + return null; + } + } + + /** + * Find the _Fields constant that matches fieldId, throwing an exception + * if it is not found. + */ + public static _Fields findByThriftIdOrThrow(int fieldId) { + _Fields fields = findByThriftId(fieldId); + if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!"); + return fields; + } + + /** + * Find the _Fields constant that matches name, or null if its not found. + */ + public static _Fields findByName(String name) { + return byName.get(name); + } + + private final short _thriftId; + private final String _fieldName; + + _Fields(short thriftId, String fieldName) { + _thriftId = thriftId; + _fieldName = fieldName; + } + + public short getThriftFieldId() { + return _thriftId; + } + + public String getFieldName() { + return _fieldName; + } + } + public static final Map<_Fields, FieldMetaData> metaDataMap; + static { + Map<_Fields, FieldMetaData> tmpMap = new EnumMap<_Fields, FieldMetaData>(_Fields.class); + metaDataMap = Collections.unmodifiableMap(tmpMap); + FieldMetaData.addStructMetaDataMap(enable_session_consistency_args.class, metaDataMap); + } + + public enable_session_consistency_args() { + } + + /** + * Performs a deep copy on other. + */ + public enable_session_consistency_args(enable_session_consistency_args other) { + } + + public enable_session_consistency_args deepCopy() { + return new enable_session_consistency_args(this); + } + + @Deprecated + public enable_session_consistency_args clone() { + return new enable_session_consistency_args(this); + } + + public void setFieldValue(_Fields field, Object value) { + switch (field) { + } + } + + public void setFieldValue(int fieldID, Object value) { + setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value); + } + + public Object getFieldValue(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public Object getFieldValue(int fieldId) { + return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId)); + } + + /** Returns true if field corresponding to fieldID is set (has been asigned a value) and false otherwise */ + public boolean isSet(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public boolean isSet(int fieldID) { + return isSet(_Fields.findByThriftIdOrThrow(fieldID)); + } + + @Override + public boolean equals(Object that) { + if (that == null) + return false; + if (that instanceof enable_session_consistency_args) + return this.equals((enable_session_consistency_args)that); + return false; + } + + public boolean equals(enable_session_consistency_args that) { + if (that == null) + return false; + + return true; + } + + @Override + public int hashCode() { + return 0; + } + + public int compareTo(enable_session_consistency_args other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + enable_session_consistency_args typedOther = (enable_session_consistency_args)other; + + return 0; + } + + public void read(TProtocol iprot) throws TException { + TField field; + iprot.readStructBegin(); + while (true) + { + field = iprot.readFieldBegin(); + if (field.type == TType.STOP) { + break; + } + switch (field.id) { + default: + TProtocolUtil.skip(iprot, field.type); + } + iprot.readFieldEnd(); + } + iprot.readStructEnd(); + + // check for required fields of primitive type, which can't be checked in the validate method + validate(); + } + + public void write(TProtocol oprot) throws TException { + validate(); + + oprot.writeStructBegin(STRUCT_DESC); + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("enable_session_consistency_args("); + boolean first = true; + + sb.append(")"); + return sb.toString(); + } + + public void validate() throws TException { + // check for required fields + } + + } + + public static class enable_session_consistency_result implements TBase, java.io.Serializable, Cloneable { + private static final TStruct STRUCT_DESC = new TStruct("enable_session_consistency_result"); + + + + /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ + public enum _Fields implements TFieldIdEnum { +; + + private static final Map byName = new HashMap(); + + static { + for (_Fields field : EnumSet.allOf(_Fields.class)) { + byName.put(field.getFieldName(), field); + } + } + + /** + * Find the _Fields constant that matches fieldId, or null if its not found. + */ + public static _Fields findByThriftId(int fieldId) { + switch(fieldId) { + default: + return null; + } + } + + /** + * Find the _Fields constant that matches fieldId, throwing an exception + * if it is not found. + */ + public static _Fields findByThriftIdOrThrow(int fieldId) { + _Fields fields = findByThriftId(fieldId); + if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!"); + return fields; + } + + /** + * Find the _Fields constant that matches name, or null if its not found. + */ + public static _Fields findByName(String name) { + return byName.get(name); + } + + private final short _thriftId; + private final String _fieldName; + + _Fields(short thriftId, String fieldName) { + _thriftId = thriftId; + _fieldName = fieldName; + } + + public short getThriftFieldId() { + return _thriftId; + } + + public String getFieldName() { + return _fieldName; + } + } + public static final Map<_Fields, FieldMetaData> metaDataMap; + static { + Map<_Fields, FieldMetaData> tmpMap = new EnumMap<_Fields, FieldMetaData>(_Fields.class); + metaDataMap = Collections.unmodifiableMap(tmpMap); + FieldMetaData.addStructMetaDataMap(enable_session_consistency_result.class, metaDataMap); + } + + public enable_session_consistency_result() { + } + + /** + * Performs a deep copy on other. + */ + public enable_session_consistency_result(enable_session_consistency_result other) { + } + + public enable_session_consistency_result deepCopy() { + return new enable_session_consistency_result(this); + } + + @Deprecated + public enable_session_consistency_result clone() { + return new enable_session_consistency_result(this); + } + + public void setFieldValue(_Fields field, Object value) { + switch (field) { + } + } + + public void setFieldValue(int fieldID, Object value) { + setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value); + } + + public Object getFieldValue(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public Object getFieldValue(int fieldId) { + return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId)); + } + + /** Returns true if field corresponding to fieldID is set (has been asigned a value) and false otherwise */ + public boolean isSet(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public boolean isSet(int fieldID) { + return isSet(_Fields.findByThriftIdOrThrow(fieldID)); + } + + @Override + public boolean equals(Object that) { + if (that == null) + return false; + if (that instanceof enable_session_consistency_result) + return this.equals((enable_session_consistency_result)that); + return false; + } + + public boolean equals(enable_session_consistency_result that) { + if (that == null) + return false; + + return true; + } + + @Override + public int hashCode() { + return 0; + } + + public int compareTo(enable_session_consistency_result other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + enable_session_consistency_result typedOther = (enable_session_consistency_result)other; + + return 0; + } + + public void read(TProtocol iprot) throws TException { + TField field; + iprot.readStructBegin(); + while (true) + { + field = iprot.readFieldBegin(); + if (field.type == TType.STOP) { + break; + } + switch (field.id) { + default: + TProtocolUtil.skip(iprot, field.type); + } + iprot.readFieldEnd(); + } + iprot.readStructEnd(); + + // check for required fields of primitive type, which can't be checked in the validate method + validate(); + } + + public void write(TProtocol oprot) throws TException { + oprot.writeStructBegin(STRUCT_DESC); + + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("enable_session_consistency_result("); + boolean first = true; + + sb.append(")"); + return sb.toString(); + } + + public void validate() throws TException { + // check for required fields + } + + } + + public static class disable_session_consistency_args implements TBase, java.io.Serializable, Cloneable { + private static final TStruct STRUCT_DESC = new TStruct("disable_session_consistency_args"); + + + + /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ + public enum _Fields implements TFieldIdEnum { +; + + private static final Map byName = new HashMap(); + + static { + for (_Fields field : EnumSet.allOf(_Fields.class)) { + byName.put(field.getFieldName(), field); + } + } + + /** + * Find the _Fields constant that matches fieldId, or null if its not found. + */ + public static _Fields findByThriftId(int fieldId) { + switch(fieldId) { + default: + return null; + } + } + + /** + * Find the _Fields constant that matches fieldId, throwing an exception + * if it is not found. + */ + public static _Fields findByThriftIdOrThrow(int fieldId) { + _Fields fields = findByThriftId(fieldId); + if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!"); + return fields; + } + + /** + * Find the _Fields constant that matches name, or null if its not found. + */ + public static _Fields findByName(String name) { + return byName.get(name); + } + + private final short _thriftId; + private final String _fieldName; + + _Fields(short thriftId, String fieldName) { + _thriftId = thriftId; + _fieldName = fieldName; + } + + public short getThriftFieldId() { + return _thriftId; + } + + public String getFieldName() { + return _fieldName; + } + } + public static final Map<_Fields, FieldMetaData> metaDataMap; + static { + Map<_Fields, FieldMetaData> tmpMap = new EnumMap<_Fields, FieldMetaData>(_Fields.class); + metaDataMap = Collections.unmodifiableMap(tmpMap); + FieldMetaData.addStructMetaDataMap(disable_session_consistency_args.class, metaDataMap); + } + + public disable_session_consistency_args() { + } + + /** + * Performs a deep copy on other. + */ + public disable_session_consistency_args(disable_session_consistency_args other) { + } + + public disable_session_consistency_args deepCopy() { + return new disable_session_consistency_args(this); + } + + @Deprecated + public disable_session_consistency_args clone() { + return new disable_session_consistency_args(this); + } + + public void setFieldValue(_Fields field, Object value) { + switch (field) { + } + } + + public void setFieldValue(int fieldID, Object value) { + setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value); + } + + public Object getFieldValue(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public Object getFieldValue(int fieldId) { + return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId)); + } + + /** Returns true if field corresponding to fieldID is set (has been asigned a value) and false otherwise */ + public boolean isSet(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public boolean isSet(int fieldID) { + return isSet(_Fields.findByThriftIdOrThrow(fieldID)); + } + + @Override + public boolean equals(Object that) { + if (that == null) + return false; + if (that instanceof disable_session_consistency_args) + return this.equals((disable_session_consistency_args)that); + return false; + } + + public boolean equals(disable_session_consistency_args that) { + if (that == null) + return false; + + return true; + } + + @Override + public int hashCode() { + return 0; + } + + public int compareTo(disable_session_consistency_args other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + disable_session_consistency_args typedOther = (disable_session_consistency_args)other; + + return 0; + } + + public void read(TProtocol iprot) throws TException { + TField field; + iprot.readStructBegin(); + while (true) + { + field = iprot.readFieldBegin(); + if (field.type == TType.STOP) { + break; + } + switch (field.id) { + default: + TProtocolUtil.skip(iprot, field.type); + } + iprot.readFieldEnd(); + } + iprot.readStructEnd(); + + // check for required fields of primitive type, which can't be checked in the validate method + validate(); + } + + public void write(TProtocol oprot) throws TException { + validate(); + + oprot.writeStructBegin(STRUCT_DESC); + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("disable_session_consistency_args("); + boolean first = true; + + sb.append(")"); + return sb.toString(); + } + + public void validate() throws TException { + // check for required fields + } + + } + + public static class disable_session_consistency_result implements TBase, java.io.Serializable, Cloneable { + private static final TStruct STRUCT_DESC = new TStruct("disable_session_consistency_result"); + + + + /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */ + public enum _Fields implements TFieldIdEnum { +; + + private static final Map byName = new HashMap(); + + static { + for (_Fields field : EnumSet.allOf(_Fields.class)) { + byName.put(field.getFieldName(), field); + } + } + + /** + * Find the _Fields constant that matches fieldId, or null if its not found. + */ + public static _Fields findByThriftId(int fieldId) { + switch(fieldId) { + default: + return null; + } + } + + /** + * Find the _Fields constant that matches fieldId, throwing an exception + * if it is not found. + */ + public static _Fields findByThriftIdOrThrow(int fieldId) { + _Fields fields = findByThriftId(fieldId); + if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!"); + return fields; + } + + /** + * Find the _Fields constant that matches name, or null if its not found. + */ + public static _Fields findByName(String name) { + return byName.get(name); + } + + private final short _thriftId; + private final String _fieldName; + + _Fields(short thriftId, String fieldName) { + _thriftId = thriftId; + _fieldName = fieldName; + } + + public short getThriftFieldId() { + return _thriftId; + } + + public String getFieldName() { + return _fieldName; + } + } + public static final Map<_Fields, FieldMetaData> metaDataMap; + static { + Map<_Fields, FieldMetaData> tmpMap = new EnumMap<_Fields, FieldMetaData>(_Fields.class); + metaDataMap = Collections.unmodifiableMap(tmpMap); + FieldMetaData.addStructMetaDataMap(disable_session_consistency_result.class, metaDataMap); + } + + public disable_session_consistency_result() { + } + + /** + * Performs a deep copy on other. + */ + public disable_session_consistency_result(disable_session_consistency_result other) { + } + + public disable_session_consistency_result deepCopy() { + return new disable_session_consistency_result(this); + } + + @Deprecated + public disable_session_consistency_result clone() { + return new disable_session_consistency_result(this); + } + + public void setFieldValue(_Fields field, Object value) { + switch (field) { + } + } + + public void setFieldValue(int fieldID, Object value) { + setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value); + } + + public Object getFieldValue(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public Object getFieldValue(int fieldId) { + return getFieldValue(_Fields.findByThriftIdOrThrow(fieldId)); + } + + /** Returns true if field corresponding to fieldID is set (has been asigned a value) and false otherwise */ + public boolean isSet(_Fields field) { + switch (field) { + } + throw new IllegalStateException(); + } + + public boolean isSet(int fieldID) { + return isSet(_Fields.findByThriftIdOrThrow(fieldID)); + } + + @Override + public boolean equals(Object that) { + if (that == null) + return false; + if (that instanceof disable_session_consistency_result) + return this.equals((disable_session_consistency_result)that); + return false; + } + + public boolean equals(disable_session_consistency_result that) { + if (that == null) + return false; + + return true; + } + + @Override + public int hashCode() { + return 0; + } + + public int compareTo(disable_session_consistency_result other) { + if (!getClass().equals(other.getClass())) { + return getClass().getName().compareTo(other.getClass().getName()); + } + + int lastComparison = 0; + disable_session_consistency_result typedOther = (disable_session_consistency_result)other; + + return 0; + } + + public void read(TProtocol iprot) throws TException { + TField field; + iprot.readStructBegin(); + while (true) + { + field = iprot.readFieldBegin(); + if (field.type == TType.STOP) { + break; + } + switch (field.id) { + default: + TProtocolUtil.skip(iprot, field.type); + } + iprot.readFieldEnd(); + } + iprot.readStructEnd(); + + // check for required fields of primitive type, which can't be checked in the validate method + validate(); + } + + public void write(TProtocol oprot) throws TException { + oprot.writeStructBegin(STRUCT_DESC); + + oprot.writeFieldStop(); + oprot.writeStructEnd(); + } + + @Override + public String toString() { + StringBuilder sb = new StringBuilder("disable_session_consistency_result("); + boolean first = true; + + sb.append(")"); + return sb.toString(); + } + + public void validate() throws TException { + // check for required fields + } + + } + public static class get_args implements TBase, java.io.Serializable, Cloneable { private static final TStruct STRUCT_DESC = new TStruct("get_args"); diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index db0891ed98..115d1e8166 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -108,12 +108,7 @@ public class Memtable implements Comparable, IFlushable isFrozen = true; } - /** - * Should only be called by ColumnFamilyStore.apply. NOT a public API. - * (CFS handles locking to avoid submitting an op - * to a flushing memtable. Any other way is unsafe.) - */ - void put(DecoratedKey key, ColumnFamily columnFamily) + public void put(DecoratedKey key, ColumnFamily columnFamily) { assert !isFrozen; // not 100% foolproof but hell, it's an assert resolve(key, columnFamily); diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index 9158b9d494..fd4d150a80 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -25,10 +25,12 @@ import java.io.IOException; import java.util.HashMap; import java.util.Map; +import org.apache.cassandra.db.columniterator.IColumnIterator; import org.apache.cassandra.io.ICompactSerializer; import org.apache.cassandra.net.Message; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.db.marshal.AbstractType; +import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.db.filter.QueryPath; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.concurrent.StageManager; @@ -86,12 +88,41 @@ public abstract class ReadCommand public abstract ReadCommand copy(); - public abstract Row getRow(Table table) throws IOException; + public Row getRow(Table table) throws IOException + { + return table.getRow(getQueryFilter()); + } - protected AbstractType getComparator() + protected abstract QueryFilter getQueryFilter(); + + protected AbstractType getComparator() { return ColumnFamily.getComparatorFor(table, getColumnFamilyName(), queryPath.superColumnName); } + + public Row mergeRowWithMemtables(Table table, Row row, Map map) + { + QueryFilter filter = getQueryFilter(); + ColumnFamilyStore cfStore = table.getColumnFamilyStore(filter.getColumnFamilyName()); + Memtable memtable = map.get(cfStore); + + final ColumnFamily memtableCf = ColumnFamily.create(cfStore.metadata); + IColumnIterator iter = filter.getMemtableColumnIterator(memtable, cfStore.getComparator()); + filter.collectCollatedColumns(memtableCf, iter, (int) (System.currentTimeMillis() / 1000) - cfStore.metadata.gcGraceSeconds); + try + { + iter.close(); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + + if (row.cf == null) + return new Row(filter.key, memtableCf); + row.cf.resolve(memtableCf); + return new Row(filter.key, row.cf); + } } class ReadCommandSerializer implements ICompactSerializer diff --git a/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java b/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java index 27f5e297c7..f1b0f5ece8 100644 --- a/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java +++ b/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java @@ -52,13 +52,6 @@ public class SliceByNamesReadCommand extends ReadCommand return readCommand; } - @Override - public Row getRow(Table table) throws IOException - { - DecoratedKey dk = StorageService.getPartitioner().decorateKey(key); - return table.getRow(QueryFilter.getNamesFilter(dk, queryPath, columnNames)); - } - @Override public String toString() { @@ -70,6 +63,11 @@ public class SliceByNamesReadCommand extends ReadCommand ')'; } + protected QueryFilter getQueryFilter() + { + DecoratedKey dk = StorageService.getPartitioner().decorateKey(key); + return QueryFilter.getNamesFilter(dk, queryPath, columnNames); + } } class SliceByNamesReadCommandSerializer extends ReadCommandSerializer diff --git a/src/java/org/apache/cassandra/db/SliceFromReadCommand.java b/src/java/org/apache/cassandra/db/SliceFromReadCommand.java index 62a229f5b2..bbfce341aa 100644 --- a/src/java/org/apache/cassandra/db/SliceFromReadCommand.java +++ b/src/java/org/apache/cassandra/db/SliceFromReadCommand.java @@ -65,10 +65,10 @@ public class SliceFromReadCommand extends ReadCommand } @Override - public Row getRow(Table table) throws IOException + protected QueryFilter getQueryFilter() { DecoratedKey dk = StorageService.getPartitioner().decorateKey(key); - return table.getRow(QueryFilter.getSliceFilter(dk, queryPath, start, finish, bitmasks, reversed, count)); + return QueryFilter.getSliceFilter(dk, queryPath, start, finish, bitmasks, reversed, count); } @Override diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 0cabc0ea66..113a55d2df 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -31,6 +31,7 @@ import com.google.common.base.Function; import com.google.common.collect.Iterables; import org.apache.cassandra.config.*; +import org.apache.cassandra.db.columniterator.IColumnIterator; import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.dht.LocalToken; import org.apache.cassandra.io.sstable.SSTableDeletingReference; @@ -469,6 +470,28 @@ public class Table memtablesToFlush.put(cfs, memtableToFlush); } + public void applyToMemtable(RowMutation rm, Map memtables) + { + for (ColumnFamily columnFamily : rm.getColumnFamilies()) + { + ColumnFamilyStore cfs = columnFamilyStores.get(columnFamily.id()); + if (cfs == null) + { + logger.error("Attempting to mutate non-existant column family " + columnFamily.id()); + continue; + } + + DecoratedKey key = StorageService.getPartitioner().decorateKey(rm.key()); + Memtable memtable = memtables.get(cfs); + if (memtable == null) + { + memtable = new Memtable(cfs, null); + memtables.put(cfs, memtable); + } + memtable.put(key, columnFamily); + } + } + public List> flush() throws IOException { List> futures = new ArrayList>(); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index c428e16aa7..31b5a01a48 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -28,7 +28,6 @@ import java.util.concurrent.*; import javax.management.MBeanServer; import javax.management.ObjectName; -import com.google.common.collect.AbstractIterator; import com.google.common.collect.Multimap; import org.apache.commons.lang.ArrayUtils; import org.apache.commons.lang.StringUtils; @@ -51,6 +50,7 @@ import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.LatencyTracker; +import org.apache.cassandra.utils.Pair; import org.apache.cassandra.utils.WrappedRunnable; import org.apache.cassandra.db.filter.QueryFilter; @@ -64,6 +64,8 @@ public class StorageProxy implements StorageProxyMBean private static final LatencyTracker rangeStats = new LatencyTracker(); private static final LatencyTracker writeStats = new LatencyTracker(); + private static ThreadLocal> sessionWrites = new ThreadLocal>(); + private StorageProxy() {} static { @@ -94,7 +96,7 @@ public class StorageProxy implements StorageProxyMBean RowMutation mostRecentRowMutation = null; StorageService ss = StorageService.instance; - + try { for (RowMutation rm : mutations) @@ -161,6 +163,14 @@ public class StorageProxy implements StorageProxyMBean { responseHandler.get(); } + if (sessionWrites.get() != null) + { + for (RowMutation rm : mutations) + { + // no need to apply locally-written mutations to the session + Table.open(rm.getTable()).applyToMemtable(rm, sessionWrites.get()); + } + } } catch (IOException e) { @@ -230,7 +240,7 @@ public class StorageProxy implements StorageProxyMBean // send off all the commands asynchronously List> localFutures = null; - List remoteResults = null; + List> remoteResults = null; for (ReadCommand command: commands) { InetAddress endPoint = StorageService.instance.findSuitableEndpoint(command.table, command.key); @@ -247,13 +257,13 @@ public class StorageProxy implements StorageProxyMBean else { if (remoteResults == null) - remoteResults = new ArrayList(); + remoteResults = new ArrayList>(); Message message = command.makeReadMessage(); if (logger.isDebugEnabled()) logger.debug("weakread reading " + command + " from " + message.getMessageId() + "@" + endPoint); if (randomlyReadRepair(command)) message.setHeader(ReadCommand.DO_REPAIR, ReadCommand.DO_REPAIR.getBytes()); - remoteResults.add(MessagingService.instance.sendRR(message, endPoint)); + remoteResults.add(new Pair(MessagingService.instance.sendRR(message, endPoint), command)); } } @@ -276,14 +286,17 @@ public class StorageProxy implements StorageProxyMBean } if (remoteResults != null) { - for (IAsyncResult iar: remoteResults) + for (Pair iar: remoteResults) { byte[] body; - body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + body = iar.left.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); ByteArrayInputStream bufIn = new ByteArrayInputStream(body); ReadResponse response = ReadResponse.serializer().deserialize(new DataInputStream(bufIn)); - if (response.row() != null) - rows.add(response.row()); + Row row = response.row(); + if (sessionWrites.get() != null) + row = iar.right.mergeRowWithMemtables(Table.open(iar.right.table), row, sessionWrites.get()); + if (row != null) + rows.add(row); } } @@ -351,6 +364,10 @@ public class StorageProxy implements StorageProxyMBean { long startTime2 = System.currentTimeMillis(); row = quorumResponseHandler.get(); + if (sessionWrites.get() != null) + { + row = command.mergeRowWithMemtables(Table.open(command.table), row, sessionWrites.get()); + } if (row != null) rows.add(row); @@ -790,4 +807,14 @@ public class StorageProxy implements StorageProxyMBean { return !Gossiper.instance.getUnreachableMembers().isEmpty(); } + + public static void enableSessionConsistency() + { + sessionWrites.set(new HashMap()); + } + + public static void disableSessionConsistency() + { + sessionWrites.remove(); + } } diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java index b386407209..54a5d026ad 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -681,6 +681,7 @@ public class CassandraServer implements Cassandra.Iface { keySpace.remove(); loginDone.remove(); + StorageProxy.disableSessionConsistency(); if (logger.isDebugEnabled()) logger.debug("logout complete"); @@ -1014,5 +1015,15 @@ public class CassandraServer implements Cassandra.Iface return StorageProxy.checkSchemaAgreement(); } + public void enable_session_consistency() throws TException + { + StorageProxy.enableSessionConsistency(); + } + + public void disable_session_consistency() throws TException + { + StorageProxy.disableSessionConsistency(); + } + // main method moved to CassandraDaemon }