diff --git a/CHANGES.txt b/CHANGES.txt index db75758e31..96dd94ae54 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -16,7 +16,6 @@ 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 7951c859f2..b60b11c14c 100644 --- a/NEWS.txt +++ b/NEWS.txt @@ -17,7 +17,6 @@ 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 0be712fdc6..edc129fc9c 100644 --- a/interface/cassandra.thrift +++ b/interface/cassandra.thrift @@ -385,13 +385,6 @@ 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 84b7830e56..5d24ffaae1 100644 --- a/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java +++ b/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java @@ -54,10 +54,6 @@ 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.) @@ -291,10 +287,6 @@ 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; @@ -462,70 +454,6 @@ 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); @@ -1649,62 +1577,6 @@ 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); @@ -2576,8 +2448,6 @@ 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()); @@ -2708,58 +2578,6 @@ 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 { @@ -5029,768 +4847,6 @@ 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 115d1e8166..db0891ed98 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -108,7 +108,12 @@ public class Memtable implements Comparable, IFlushable isFrozen = true; } - public void put(DecoratedKey key, ColumnFamily columnFamily) + /** + * 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) { 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 fd4d150a80..9158b9d494 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -25,12 +25,10 @@ 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; @@ -88,41 +86,12 @@ public abstract class ReadCommand public abstract ReadCommand copy(); - public Row getRow(Table table) throws IOException - { - return table.getRow(getQueryFilter()); - } + public abstract Row getRow(Table table) throws IOException; - protected abstract QueryFilter getQueryFilter(); - - protected AbstractType getComparator() + 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 f1b0f5ece8..27f5e297c7 100644 --- a/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java +++ b/src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java @@ -52,6 +52,13 @@ 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() { @@ -63,11 +70,6 @@ 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 bbfce341aa..62a229f5b2 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 - protected QueryFilter getQueryFilter() + public Row getRow(Table table) throws IOException { DecoratedKey dk = StorageService.getPartitioner().decorateKey(key); - return QueryFilter.getSliceFilter(dk, queryPath, start, finish, bitmasks, reversed, count); + return table.getRow(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 113a55d2df..0cabc0ea66 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -31,7 +31,6 @@ 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; @@ -470,28 +469,6 @@ 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 31b5a01a48..c428e16aa7 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -28,6 +28,7 @@ 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; @@ -50,7 +51,6 @@ 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,8 +64,6 @@ 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 { @@ -96,7 +94,7 @@ public class StorageProxy implements StorageProxyMBean RowMutation mostRecentRowMutation = null; StorageService ss = StorageService.instance; - + try { for (RowMutation rm : mutations) @@ -163,14 +161,6 @@ 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) { @@ -240,7 +230,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); @@ -257,13 +247,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(new Pair(MessagingService.instance.sendRR(message, endPoint), command)); + remoteResults.add(MessagingService.instance.sendRR(message, endPoint)); } } @@ -286,17 +276,14 @@ public class StorageProxy implements StorageProxyMBean } if (remoteResults != null) { - for (Pair iar: remoteResults) + for (IAsyncResult iar: remoteResults) { byte[] body; - body = iar.left.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); + body = iar.get(DatabaseDescriptor.getRpcTimeout(), TimeUnit.MILLISECONDS); ByteArrayInputStream bufIn = new ByteArrayInputStream(body); ReadResponse response = ReadResponse.serializer().deserialize(new DataInputStream(bufIn)); - 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); + if (response.row() != null) + rows.add(response.row()); } } @@ -364,10 +351,6 @@ 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); @@ -807,14 +790,4 @@ 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 54a5d026ad..b386407209 100644 --- a/src/java/org/apache/cassandra/thrift/CassandraServer.java +++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java @@ -681,7 +681,6 @@ public class CassandraServer implements Cassandra.Iface { keySpace.remove(); loginDone.remove(); - StorageProxy.disableSessionConsistency(); if (logger.isDebugEnabled()) logger.debug("logout complete"); @@ -1015,15 +1014,5 @@ 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 }