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

per-connection read-your-writes "session" consistency. patch by Brian Palmer; reviewed by jbellis for CASSANDRA-876 

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@986781 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2010-08-18 16:40:42 +00:00
parent 66d32c8e28
commit 35ee5347f8
11 changed files with 1064 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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

1
NEWS.txt
View File

@ -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.

7
interface/cassandra.thrift
View File

@ -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
/**

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

@ -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<AsyncClient.set_keyspace_call> resultHandler) throws TException;
public void enable_session_consistency(AsyncMethodCallback<AsyncClient.enable_session_consistency_call> resultHandler) throws TException;
public void disable_session_consistency(AsyncMethodCallback<AsyncClient.disable_session_consistency_call> resultHandler) throws TException;
public void get(byte[] key, ColumnPath column_path, ConsistencyLevel consistency_level, AsyncMethodCallback<AsyncClient.get_call> resultHandler) throws TException;
public void get_slice(byte[] key, ColumnParent column_parent, SlicePredicate predicate, ConsistencyLevel consistency_level, AsyncMethodCallback<AsyncClient.get_slice_call> 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<enable_session_consistency_call> 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<enable_session_consistency_call> 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<disable_session_consistency_call> 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<disable_session_consistency_call> 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<get_call> 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<enable_session_consistency_args, enable_session_consistency_args._Fields>, 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<String, _Fields> byName = new HashMap<String, _Fields>();
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 <i>other</i>.
*/
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<enable_session_consistency_result, enable_session_consistency_result._Fields>, 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<String, _Fields> byName = new HashMap<String, _Fields>();
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 <i>other</i>.
*/
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<disable_session_consistency_args, disable_session_consistency_args._Fields>, 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<String, _Fields> byName = new HashMap<String, _Fields>();
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 <i>other</i>.
*/
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<disable_session_consistency_result, disable_session_consistency_result._Fields>, 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<String, _Fields> byName = new HashMap<String, _Fields>();
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 <i>other</i>.
*/
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<get_args, get_args._Fields>, java.io.Serializable, Cloneable {
private static final TStruct STRUCT_DESC = new TStruct("get_args");

7
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -108,12 +108,7 @@ public class Memtable implements Comparable<Memtable>, 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);

35
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -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<ColumnFamilyStore, Memtable> 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<ReadCommand>

12
src/java/org/apache/cassandra/db/SliceByNamesReadCommand.java
View File

@ -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

4
src/java/org/apache/cassandra/db/SliceFromReadCommand.java
View File

@ -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

23
src/java/org/apache/cassandra/db/Table.java
View File

@ -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<ColumnFamilyStore, Memtable> 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<Future<?>> flush() throws IOException
{
List<Future<?>> futures = new ArrayList<Future<?>>();

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

@ -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<Map<ColumnFamilyStore, Memtable>> sessionWrites = new ThreadLocal<Map<ColumnFamilyStore,Memtable>>();
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<Future<Object>> localFutures = null;
List<IAsyncResult> remoteResults = null;
List<Pair<IAsyncResult,ReadCommand>> 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<IAsyncResult>();
remoteResults = new ArrayList<Pair<IAsyncResult, ReadCommand>>();
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<IAsyncResult,ReadCommand>(MessagingService.instance.sendRR(message, endPoint), command));
}
}
@ -276,14 +286,17 @@ public class StorageProxy implements StorageProxyMBean
}
if (remoteResults != null)
{
for (IAsyncResult iar: remoteResults)
for (Pair<IAsyncResult,ReadCommand> 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<ColumnFamilyStore, Memtable>());
}
public static void disableSessionConsistency()
{
sessionWrites.remove();
}
}

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

@ -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
}