You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jb...@apache.org on 2010/05/10 18:37:29 UTC
svn commit: r942817 - in /cassandra/trunk: ./ interface/
interface/thrift/gen-java/org/apache/cassandra/thrift/
src/java/org/apache/cassandra/db/ src/java/org/apache/cassandra/io/sstable/
src/java/org/apache/cassandra/service/ src/java/org/apache/cassa...
Author: jbellis
Date: Mon May 10 16:37:28 2010
New Revision: 942817
URL: http://svn.apache.org/viewvc?rev=942817&view=rev
Log:
implement truncate. patch by jbellis and Ran Tavory for CASSANDRA-531
Added:
cassandra/trunk/src/java/org/apache/cassandra/db/TruncateResponse.java
cassandra/trunk/src/java/org/apache/cassandra/db/TruncateVerbHandler.java
cassandra/trunk/src/java/org/apache/cassandra/db/Truncation.java
cassandra/trunk/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java
Modified:
cassandra/trunk/CHANGES.txt
cassandra/trunk/NEWS.txt
cassandra/trunk/interface/cassandra.thrift
cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java
cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
cassandra/trunk/src/java/org/apache/cassandra/io/sstable/RowIndexedReader.java
cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java
cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java
cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
cassandra/trunk/src/java/org/apache/cassandra/tools/ClusterCmd.java
cassandra/trunk/src/java/org/apache/cassandra/tools/NodeCmd.java
cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java
cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java
cassandra/trunk/test/system/test_thrift_server.py
cassandra/trunk/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
cassandra/trunk/test/unit/org/apache/cassandra/tools/SSTableExportTest.java
Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Mon May 10 16:37:28 2010
@@ -21,6 +21,7 @@ dev
* add CRC to commitlog header (CASSANDRA-999)
* removed multiget thrift method (CASSANDRA-739)
* removed deprecated batch_insert and get_range_slice methods (CASSANDRA-1065)
+ * add truncate thrift method (CASSANDRA-531)
0.6.2
Modified: cassandra/trunk/NEWS.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/NEWS.txt?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/NEWS.txt (original)
+++ cassandra/trunk/NEWS.txt Mon May 10 16:37:28 2010
@@ -9,6 +9,7 @@ Features
UTF-8 encoded strings, but RandomPartitioner no longer expects strings.
- A new ByteOrderedPartitioner supports bytes keys with arbitrary content,
and orders keys by their byte value.
+ - Truncate thrift method allows clearing an entire ColumnFamily at once
Configuraton
------------
Modified: cassandra/trunk/interface/cassandra.thrift
URL: http://svn.apache.org/viewvc/cassandra/trunk/interface/cassandra.thrift?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/interface/cassandra.thrift (original)
+++ cassandra/trunk/interface/cassandra.thrift Mon May 10 16:37:28 2010
@@ -46,7 +46,7 @@ namespace rb CassandraThrift
# for every edit that doesn't result in a change to major/minor.
#
# See the Semantic Versioning Specification (SemVer) http://semver.org.
-const string VERSION = "6.0.0"
+const string VERSION = "6.1.0"
#
# data structures
@@ -421,6 +421,18 @@ service Cassandra {
2:required ConsistencyLevel consistency_level=ONE)
throws (1:InvalidRequestException ire, 2:UnavailableException ue, 3:TimedOutException te),
+ /**
+ Truncate will mark and entire column family as deleted.
+ From the user's perspective a successful call to truncate will result complete data deletion from cfname.
+ Internally, however, disk space will not be immediatily released, as with all deletes in cassandra, this one
+ only marks the data as deleted.
+ The operation succeeds only if all hosts in the cluster at available and will throw an UnavailableException if
+ some hosts are down.
+ */
+ void truncate(1:required string keyspace,
+ 2:required string cfname)
+ throws (1: InvalidRequestException ire, 2: UnavailableException ue),
+
// Meta-APIs -- APIs to get information about the node or cluster,
// rather than user data. The nodeprobe program provides usage examples.
@@ -473,4 +485,5 @@ service Cassandra {
void system_rename_keyspace(1:required string old_name, 2:required string new_name)
throws (1:InvalidRequestException ire),
+
}
Modified: cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java (original)
+++ cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Cassandra.java Mon May 10 16:37:28 2010
@@ -149,6 +149,19 @@ public class Cassandra {
public void batch_mutate(Map<byte[],Map<String,List<Mutation>>> mutation_map, ConsistencyLevel consistency_level) throws InvalidRequestException, UnavailableException, TimedOutException, TException;
/**
+ * Truncate will mark and entire column family as deleted.
+ * From the user's perspective a successful call to truncate will result complete data deletion from cfname.
+ * Internally, however, disk space will not be immediatily released, as with all deletes in cassandra, this one
+ * only marks the data as deleted.
+ * The operation succeeds only if all hosts in the cluster at available and will throw an UnavailableException if
+ * some hosts are down.
+ *
+ * @param keyspace
+ * @param cfname
+ */
+ public void truncate(String keyspace, String cfname) throws InvalidRequestException, UnavailableException, TException;
+
+ /**
* list the defined keyspaces in this cluster
*/
public Set<String> describe_keyspaces() throws TException;
@@ -707,6 +720,43 @@ public class Cassandra {
return;
}
+ public void truncate(String keyspace, String cfname) throws InvalidRequestException, UnavailableException, TException
+ {
+ send_truncate(keyspace, cfname);
+ recv_truncate();
+ }
+
+ public void send_truncate(String keyspace, String cfname) throws TException
+ {
+ oprot_.writeMessageBegin(new TMessage("truncate", TMessageType.CALL, seqid_));
+ truncate_args args = new truncate_args();
+ args.keyspace = keyspace;
+ args.cfname = cfname;
+ args.write(oprot_);
+ oprot_.writeMessageEnd();
+ oprot_.getTransport().flush();
+ }
+
+ public void recv_truncate() throws InvalidRequestException, UnavailableException, TException
+ {
+ TMessage msg = iprot_.readMessageBegin();
+ if (msg.type == TMessageType.EXCEPTION) {
+ TApplicationException x = TApplicationException.read(iprot_);
+ iprot_.readMessageEnd();
+ throw x;
+ }
+ truncate_result result = new truncate_result();
+ result.read(iprot_);
+ iprot_.readMessageEnd();
+ if (result.ire != null) {
+ throw result.ire;
+ }
+ if (result.ue != null) {
+ throw result.ue;
+ }
+ return;
+ }
+
public Set<String> describe_keyspaces() throws TException
{
send_describe_keyspaces();
@@ -1126,6 +1176,7 @@ public class Cassandra {
processMap_.put("insert", new insert());
processMap_.put("remove", new remove());
processMap_.put("batch_mutate", new batch_mutate());
+ processMap_.put("truncate", new truncate());
processMap_.put("describe_keyspaces", new describe_keyspaces());
processMap_.put("describe_cluster_name", new describe_cluster_name());
processMap_.put("describe_version", new describe_version());
@@ -1624,6 +1675,46 @@ public class Cassandra {
}
+ private class truncate implements ProcessFunction {
+ public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException
+ {
+ truncate_args args = new truncate_args();
+ try {
+ args.read(iprot);
+ } catch (TProtocolException e) {
+ iprot.readMessageEnd();
+ TApplicationException x = new TApplicationException(TApplicationException.PROTOCOL_ERROR, e.getMessage());
+ oprot.writeMessageBegin(new TMessage("truncate", TMessageType.EXCEPTION, seqid));
+ x.write(oprot);
+ oprot.writeMessageEnd();
+ oprot.getTransport().flush();
+ return;
+ }
+ iprot.readMessageEnd();
+ truncate_result result = new truncate_result();
+ try {
+ iface_.truncate(args.keyspace, args.cfname);
+ } catch (InvalidRequestException ire) {
+ result.ire = ire;
+ } catch (UnavailableException ue) {
+ result.ue = ue;
+ } catch (Throwable th) {
+ LOGGER.error("Internal error processing truncate", th);
+ TApplicationException x = new TApplicationException(TApplicationException.INTERNAL_ERROR, "Internal error processing truncate");
+ oprot.writeMessageBegin(new TMessage("truncate", TMessageType.EXCEPTION, seqid));
+ x.write(oprot);
+ oprot.writeMessageEnd();
+ oprot.getTransport().flush();
+ return;
+ }
+ oprot.writeMessageBegin(new TMessage("truncate", TMessageType.REPLY, seqid));
+ result.write(oprot);
+ oprot.writeMessageEnd();
+ oprot.getTransport().flush();
+ }
+
+ }
+
private class describe_keyspaces implements ProcessFunction {
public void process(int seqid, TProtocol iprot, TProtocol oprot) throws TException
{
@@ -13170,6 +13261,744 @@ public class Cassandra {
}
+ public static class truncate_args implements TBase<truncate_args._Fields>, java.io.Serializable, Cloneable, Comparable<truncate_args> {
+ private static final TStruct STRUCT_DESC = new TStruct("truncate_args");
+
+ private static final TField KEYSPACE_FIELD_DESC = new TField("keyspace", TType.STRING, (short)1);
+ private static final TField CFNAME_FIELD_DESC = new TField("cfname", TType.STRING, (short)2);
+
+ public String keyspace;
+ public String cfname;
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements TFieldIdEnum {
+ KEYSPACE((short)1, "keyspace"),
+ CFNAME((short)2, "cfname");
+
+ private static final Map<Integer, _Fields> byId = new HashMap<Integer, _Fields>();
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byId.put((int)field._thriftId, field);
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ return byId.get(fieldId);
+ }
+
+ /**
+ * 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;
+ }
+ }
+
+ // isset id assignments
+
+ public static final Map<_Fields, FieldMetaData> metaDataMap = Collections.unmodifiableMap(new EnumMap<_Fields, FieldMetaData>(_Fields.class) {{
+ put(_Fields.KEYSPACE, new FieldMetaData("keyspace", TFieldRequirementType.REQUIRED,
+ new FieldValueMetaData(TType.STRING)));
+ put(_Fields.CFNAME, new FieldMetaData("cfname", TFieldRequirementType.REQUIRED,
+ new FieldValueMetaData(TType.STRING)));
+ }});
+
+ static {
+ FieldMetaData.addStructMetaDataMap(truncate_args.class, metaDataMap);
+ }
+
+ public truncate_args() {
+ }
+
+ public truncate_args(
+ String keyspace,
+ String cfname)
+ {
+ this();
+ this.keyspace = keyspace;
+ this.cfname = cfname;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public truncate_args(truncate_args other) {
+ if (other.isSetKeyspace()) {
+ this.keyspace = other.keyspace;
+ }
+ if (other.isSetCfname()) {
+ this.cfname = other.cfname;
+ }
+ }
+
+ public truncate_args deepCopy() {
+ return new truncate_args(this);
+ }
+
+ @Deprecated
+ public truncate_args clone() {
+ return new truncate_args(this);
+ }
+
+ public String getKeyspace() {
+ return this.keyspace;
+ }
+
+ public truncate_args setKeyspace(String keyspace) {
+ this.keyspace = keyspace;
+ return this;
+ }
+
+ public void unsetKeyspace() {
+ this.keyspace = null;
+ }
+
+ /** Returns true if field keyspace is set (has been asigned a value) and false otherwise */
+ public boolean isSetKeyspace() {
+ return this.keyspace != null;
+ }
+
+ public void setKeyspaceIsSet(boolean value) {
+ if (!value) {
+ this.keyspace = null;
+ }
+ }
+
+ public String getCfname() {
+ return this.cfname;
+ }
+
+ public truncate_args setCfname(String cfname) {
+ this.cfname = cfname;
+ return this;
+ }
+
+ public void unsetCfname() {
+ this.cfname = null;
+ }
+
+ /** Returns true if field cfname is set (has been asigned a value) and false otherwise */
+ public boolean isSetCfname() {
+ return this.cfname != null;
+ }
+
+ public void setCfnameIsSet(boolean value) {
+ if (!value) {
+ this.cfname = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case KEYSPACE:
+ if (value == null) {
+ unsetKeyspace();
+ } else {
+ setKeyspace((String)value);
+ }
+ break;
+
+ case CFNAME:
+ if (value == null) {
+ unsetCfname();
+ } else {
+ setCfname((String)value);
+ }
+ break;
+
+ }
+ }
+
+ public void setFieldValue(int fieldID, Object value) {
+ setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value);
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case KEYSPACE:
+ return getKeyspace();
+
+ case CFNAME:
+ return getCfname();
+
+ }
+ 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) {
+ case KEYSPACE:
+ return isSetKeyspace();
+ case CFNAME:
+ return isSetCfname();
+ }
+ 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 truncate_args)
+ return this.equals((truncate_args)that);
+ return false;
+ }
+
+ public boolean equals(truncate_args that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_keyspace = true && this.isSetKeyspace();
+ boolean that_present_keyspace = true && that.isSetKeyspace();
+ if (this_present_keyspace || that_present_keyspace) {
+ if (!(this_present_keyspace && that_present_keyspace))
+ return false;
+ if (!this.keyspace.equals(that.keyspace))
+ return false;
+ }
+
+ boolean this_present_cfname = true && this.isSetCfname();
+ boolean that_present_cfname = true && that.isSetCfname();
+ if (this_present_cfname || that_present_cfname) {
+ if (!(this_present_cfname && that_present_cfname))
+ return false;
+ if (!this.cfname.equals(that.cfname))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ public int compareTo(truncate_args other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ truncate_args typedOther = (truncate_args)other;
+
+ lastComparison = Boolean.valueOf(isSetKeyspace()).compareTo(typedOther.isSetKeyspace());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetKeyspace()) { lastComparison = TBaseHelper.compareTo(keyspace, typedOther.keyspace);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(isSetCfname()).compareTo(typedOther.isSetCfname());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetCfname()) { lastComparison = TBaseHelper.compareTo(cfname, typedOther.cfname);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ 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) {
+ case 1: // KEYSPACE
+ if (field.type == TType.STRING) {
+ this.keyspace = iprot.readString();
+ } else {
+ TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ case 2: // CFNAME
+ if (field.type == TType.STRING) {
+ this.cfname = iprot.readString();
+ } else {
+ TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ 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);
+ if (this.keyspace != null) {
+ oprot.writeFieldBegin(KEYSPACE_FIELD_DESC);
+ oprot.writeString(this.keyspace);
+ oprot.writeFieldEnd();
+ }
+ if (this.cfname != null) {
+ oprot.writeFieldBegin(CFNAME_FIELD_DESC);
+ oprot.writeString(this.cfname);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("truncate_args(");
+ boolean first = true;
+
+ sb.append("keyspace:");
+ if (this.keyspace == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.keyspace);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("cfname:");
+ if (this.cfname == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.cfname);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws TException {
+ // check for required fields
+ if (keyspace == null) {
+ throw new TProtocolException("Required field 'keyspace' was not present! Struct: " + toString());
+ }
+ if (cfname == null) {
+ throw new TProtocolException("Required field 'cfname' was not present! Struct: " + toString());
+ }
+ }
+
+ }
+
+ public static class truncate_result implements TBase<truncate_result._Fields>, java.io.Serializable, Cloneable, Comparable<truncate_result> {
+ private static final TStruct STRUCT_DESC = new TStruct("truncate_result");
+
+ private static final TField IRE_FIELD_DESC = new TField("ire", TType.STRUCT, (short)1);
+ private static final TField UE_FIELD_DESC = new TField("ue", TType.STRUCT, (short)2);
+
+ public InvalidRequestException ire;
+ public UnavailableException ue;
+
+ /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
+ public enum _Fields implements TFieldIdEnum {
+ IRE((short)1, "ire"),
+ UE((short)2, "ue");
+
+ private static final Map<Integer, _Fields> byId = new HashMap<Integer, _Fields>();
+ private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
+
+ static {
+ for (_Fields field : EnumSet.allOf(_Fields.class)) {
+ byId.put((int)field._thriftId, field);
+ byName.put(field.getFieldName(), field);
+ }
+ }
+
+ /**
+ * Find the _Fields constant that matches fieldId, or null if its not found.
+ */
+ public static _Fields findByThriftId(int fieldId) {
+ return byId.get(fieldId);
+ }
+
+ /**
+ * 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;
+ }
+ }
+
+ // isset id assignments
+
+ public static final Map<_Fields, FieldMetaData> metaDataMap = Collections.unmodifiableMap(new EnumMap<_Fields, FieldMetaData>(_Fields.class) {{
+ put(_Fields.IRE, new FieldMetaData("ire", TFieldRequirementType.DEFAULT,
+ new FieldValueMetaData(TType.STRUCT)));
+ put(_Fields.UE, new FieldMetaData("ue", TFieldRequirementType.DEFAULT,
+ new FieldValueMetaData(TType.STRUCT)));
+ }});
+
+ static {
+ FieldMetaData.addStructMetaDataMap(truncate_result.class, metaDataMap);
+ }
+
+ public truncate_result() {
+ }
+
+ public truncate_result(
+ InvalidRequestException ire,
+ UnavailableException ue)
+ {
+ this();
+ this.ire = ire;
+ this.ue = ue;
+ }
+
+ /**
+ * Performs a deep copy on <i>other</i>.
+ */
+ public truncate_result(truncate_result other) {
+ if (other.isSetIre()) {
+ this.ire = new InvalidRequestException(other.ire);
+ }
+ if (other.isSetUe()) {
+ this.ue = new UnavailableException(other.ue);
+ }
+ }
+
+ public truncate_result deepCopy() {
+ return new truncate_result(this);
+ }
+
+ @Deprecated
+ public truncate_result clone() {
+ return new truncate_result(this);
+ }
+
+ public InvalidRequestException getIre() {
+ return this.ire;
+ }
+
+ public truncate_result setIre(InvalidRequestException ire) {
+ this.ire = ire;
+ return this;
+ }
+
+ public void unsetIre() {
+ this.ire = null;
+ }
+
+ /** Returns true if field ire is set (has been asigned a value) and false otherwise */
+ public boolean isSetIre() {
+ return this.ire != null;
+ }
+
+ public void setIreIsSet(boolean value) {
+ if (!value) {
+ this.ire = null;
+ }
+ }
+
+ public UnavailableException getUe() {
+ return this.ue;
+ }
+
+ public truncate_result setUe(UnavailableException ue) {
+ this.ue = ue;
+ return this;
+ }
+
+ public void unsetUe() {
+ this.ue = null;
+ }
+
+ /** Returns true if field ue is set (has been asigned a value) and false otherwise */
+ public boolean isSetUe() {
+ return this.ue != null;
+ }
+
+ public void setUeIsSet(boolean value) {
+ if (!value) {
+ this.ue = null;
+ }
+ }
+
+ public void setFieldValue(_Fields field, Object value) {
+ switch (field) {
+ case IRE:
+ if (value == null) {
+ unsetIre();
+ } else {
+ setIre((InvalidRequestException)value);
+ }
+ break;
+
+ case UE:
+ if (value == null) {
+ unsetUe();
+ } else {
+ setUe((UnavailableException)value);
+ }
+ break;
+
+ }
+ }
+
+ public void setFieldValue(int fieldID, Object value) {
+ setFieldValue(_Fields.findByThriftIdOrThrow(fieldID), value);
+ }
+
+ public Object getFieldValue(_Fields field) {
+ switch (field) {
+ case IRE:
+ return getIre();
+
+ case UE:
+ return getUe();
+
+ }
+ 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) {
+ case IRE:
+ return isSetIre();
+ case UE:
+ return isSetUe();
+ }
+ 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 truncate_result)
+ return this.equals((truncate_result)that);
+ return false;
+ }
+
+ public boolean equals(truncate_result that) {
+ if (that == null)
+ return false;
+
+ boolean this_present_ire = true && this.isSetIre();
+ boolean that_present_ire = true && that.isSetIre();
+ if (this_present_ire || that_present_ire) {
+ if (!(this_present_ire && that_present_ire))
+ return false;
+ if (!this.ire.equals(that.ire))
+ return false;
+ }
+
+ boolean this_present_ue = true && this.isSetUe();
+ boolean that_present_ue = true && that.isSetUe();
+ if (this_present_ue || that_present_ue) {
+ if (!(this_present_ue && that_present_ue))
+ return false;
+ if (!this.ue.equals(that.ue))
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return 0;
+ }
+
+ public int compareTo(truncate_result other) {
+ if (!getClass().equals(other.getClass())) {
+ return getClass().getName().compareTo(other.getClass().getName());
+ }
+
+ int lastComparison = 0;
+ truncate_result typedOther = (truncate_result)other;
+
+ lastComparison = Boolean.valueOf(isSetIre()).compareTo(typedOther.isSetIre());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetIre()) { lastComparison = TBaseHelper.compareTo(ire, typedOther.ire);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ lastComparison = Boolean.valueOf(isSetUe()).compareTo(typedOther.isSetUe());
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ if (isSetUe()) { lastComparison = TBaseHelper.compareTo(ue, typedOther.ue);
+ if (lastComparison != 0) {
+ return lastComparison;
+ }
+ }
+ 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) {
+ case 1: // IRE
+ if (field.type == TType.STRUCT) {
+ this.ire = new InvalidRequestException();
+ this.ire.read(iprot);
+ } else {
+ TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ case 2: // UE
+ if (field.type == TType.STRUCT) {
+ this.ue = new UnavailableException();
+ this.ue.read(iprot);
+ } else {
+ TProtocolUtil.skip(iprot, field.type);
+ }
+ break;
+ 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);
+
+ if (this.isSetIre()) {
+ oprot.writeFieldBegin(IRE_FIELD_DESC);
+ this.ire.write(oprot);
+ oprot.writeFieldEnd();
+ } else if (this.isSetUe()) {
+ oprot.writeFieldBegin(UE_FIELD_DESC);
+ this.ue.write(oprot);
+ oprot.writeFieldEnd();
+ }
+ oprot.writeFieldStop();
+ oprot.writeStructEnd();
+ }
+
+ @Override
+ public String toString() {
+ StringBuilder sb = new StringBuilder("truncate_result(");
+ boolean first = true;
+
+ sb.append("ire:");
+ if (this.ire == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.ire);
+ }
+ first = false;
+ if (!first) sb.append(", ");
+ sb.append("ue:");
+ if (this.ue == null) {
+ sb.append("null");
+ } else {
+ sb.append(this.ue);
+ }
+ first = false;
+ sb.append(")");
+ return sb.toString();
+ }
+
+ public void validate() throws TException {
+ // check for required fields
+ }
+
+ }
+
public static class describe_keyspaces_args implements TBase<describe_keyspaces_args._Fields>, java.io.Serializable, Cloneable, Comparable<describe_keyspaces_args> {
private static final TStruct STRUCT_DESC = new TStruct("describe_keyspaces_args");
Modified: cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java (original)
+++ cassandra/trunk/interface/thrift/gen-java/org/apache/cassandra/thrift/Constants.java Mon May 10 16:37:28 2010
@@ -42,6 +42,6 @@ import org.slf4j.LoggerFactory;
public class Constants {
- public static final String VERSION = "6.0.0";
+ public static final String VERSION = "6.1.0";
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Mon May 10 16:37:28 2010
@@ -1113,4 +1113,48 @@ public class ColumnFamilyStore implement
{
return memtablesPendingFlush;
}
+
+ /**
+ * Truncate practically deletes the entire column family's data
+ * @return a Future to the delete operation. Call the future's get() to make
+ * sure the column family has been deleted
+ */
+ public Future<?> truncate() throws IOException
+ {
+ // snapshot will also flush, but we want to truncate the most possible, and anything in a flush written
+ // after truncateAt won't be truncated.
+ try
+ {
+ forceBlockingFlush();
+ }
+ catch (Exception e)
+ {
+ throw new RuntimeException(e);
+ }
+
+ final long truncatedAt = System.currentTimeMillis();
+ snapshot(Table.getTimestampedSnapshotName("before-truncate"));
+
+ Runnable runnable = new WrappedRunnable()
+ {
+ public void runMayThrow() throws InterruptedException, IOException
+ {
+ // putting markCompacted on the commitlogUpdater thread ensures it will run
+ // after any compactions that were in progress when truncate was called, are finished
+ List<SSTableReader> truncatedSSTables = new ArrayList<SSTableReader>();
+ for (SSTableReader sstable : ssTables_.getSSTables())
+ {
+ if (!sstable.newSince(truncatedAt))
+ truncatedSSTables.add(sstable);
+ }
+ markCompacted(truncatedSSTables);
+
+ // Invalidate row cache
+ invalidateRowCache();
+ }
+ };
+
+ return commitLogUpdater_.submit(runnable);
+ }
+
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java Mon May 10 16:37:28 2010
@@ -365,7 +365,7 @@ public class CompactionManager implement
ci.close();
}
- SSTableReader ssTable = writer.closeAndOpenReader();
+ SSTableReader ssTable = writer.closeAndOpenReader(getMaxDataAge(sstables));
cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable));
submitMinorIfNeeded(cfs);
@@ -375,6 +375,17 @@ public class CompactionManager implement
return sstables.size();
}
+ private static long getMaxDataAge(Collection<SSTableReader> sstables)
+ {
+ long max = 0;
+ for (SSTableReader sstable : sstables)
+ {
+ if (sstable.maxDataAge > max)
+ max = sstable.maxDataAge;
+ }
+ return max;
+ }
+
/**
* This function is used to do the anti compaction process , it spits out the file which has keys that belong to a given range
* If the target is not specified it spits out the file as a compacted file with the unecessary ranges wiped out.
@@ -439,7 +450,7 @@ public class CompactionManager implement
if (writer != null)
{
- results.add(writer.closeAndOpenReader());
+ results.add(writer.closeAndOpenReader(getMaxDataAge(sstables)));
String format = "AntiCompacted to %s. %d/%d bytes for %d keys. Time: %dms.";
long dTime = System.currentTimeMillis() - startTime;
logger.info(String.format(format, writer.getFilename(), SSTable.getTotalBytes(sstables), results.get(0).length(), totalkeysWritten, dTime));
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Table.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Table.java Mon May 10 16:37:28 2010
@@ -435,4 +435,21 @@ public class Table
};
return Iterables.transform(DatabaseDescriptor.getTables(), transformer);
}
+
+ /**
+ * Performs a synchronous truncate operation, effectively deleting all data
+ * from the column family cfname
+ * @param cfname
+ * @throws IOException
+ * @throws ExecutionException
+ * @throws InterruptedException
+ */
+ public void truncate(String cfname) throws InterruptedException, ExecutionException, IOException
+ {
+ logger.debug("Truncating...");
+ ColumnFamilyStore cfs = getColumnFamilyStore(cfname);
+ // truncate, blocking
+ cfs.truncate().get();
+ logger.debug("Truncation done.");
+ }
}
Added: cassandra/trunk/src/java/org/apache/cassandra/db/TruncateResponse.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/TruncateResponse.java?rev=942817&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/TruncateResponse.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/TruncateResponse.java Mon May 10 16:37:28 2010
@@ -0,0 +1,83 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.utils.FBUtilities;
+
+
+/**
+ * This message is sent back the truncate operation and basically specifies if
+ * the truncate succeeded.
+ *
+ * @author rantav@gmail.com
+ */
+public class TruncateResponse
+{
+ private static TruncateResponseSerializer serializer_ = new TruncateResponseSerializer();
+
+ public static TruncateResponseSerializer serializer()
+ {
+ return serializer_;
+ }
+
+ public final String keyspace;
+ public final String columnFamily;
+ public final boolean success;
+
+
+ public static Message makeTruncateResponseMessage(Message original, TruncateResponse truncateResponseMessage)
+ throws IOException
+ {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(bos);
+ TruncateResponse.serializer().serialize(truncateResponseMessage, dos);
+ return original.getReply(FBUtilities.getLocalAddress(), bos.toByteArray());
+ }
+
+ public TruncateResponse(String keyspace, String columnFamily, boolean success) {
+ this.keyspace = keyspace;
+ this.columnFamily = columnFamily;
+ this.success = success;
+ }
+
+ public static class TruncateResponseSerializer implements ICompactSerializer<TruncateResponse>
+ {
+ public void serialize(TruncateResponse tr, DataOutputStream dos) throws IOException
+ {
+ dos.writeUTF(tr.keyspace);
+ dos.writeUTF(tr.columnFamily);
+ dos.writeBoolean(tr.success);
+ }
+
+ public TruncateResponse deserialize(DataInputStream dis) throws IOException
+ {
+ String keyspace = dis.readUTF();
+ String columnFamily = dis.readUTF();
+ boolean success = dis.readBoolean();
+ return new TruncateResponse(keyspace, columnFamily, success);
+ }
+ }
+}
Added: cassandra/trunk/src/java/org/apache/cassandra/db/TruncateVerbHandler.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/TruncateVerbHandler.java?rev=942817&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/TruncateVerbHandler.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/TruncateVerbHandler.java Mon May 10 16:37:28 2010
@@ -0,0 +1,105 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.cassandra.net.IVerbHandler;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Handles the TRUNCATE verb
+ *
+ * @author rantav@gmail.com
+ *
+ */
+public class TruncateVerbHandler implements IVerbHandler
+{
+ private static Logger logger = LoggerFactory.getLogger(TruncateVerbHandler.class);
+
+ public void doVerb(Message message)
+ {
+ byte[] bytes = message.getMessageBody();
+ ByteArrayInputStream buffer = new ByteArrayInputStream(bytes);
+
+ try
+ {
+ Truncation t = Truncation.serializer().deserialize(new DataInputStream(buffer));
+ logger.debug("Applying {}", t);
+
+ try
+ {
+ Table.open(t.keyspace).truncate(t.columnFamily);
+ }
+ catch (IOException e)
+ {
+ logger.error("Error in truncation", e);
+ respondError(t, message);
+ throw e;
+ }
+ catch (InterruptedException e)
+ {
+ logger.error("Error in truncation", e);
+ respondError(t, message);
+ throw e;
+ }
+ catch (ExecutionException e)
+ {
+ logger.error("Error in truncation", e);
+ respondError(t, message);
+ throw e;
+ }
+ logger.debug("Truncate operation succeeded at this host");
+
+ TruncateResponse response = new TruncateResponse(t.keyspace, t.columnFamily, true);
+ Message responseMessage = TruncateResponse.makeTruncateResponseMessage(message, response);
+ logger.debug("{} applied. Sending response to {}@{} ",
+ new Object[]{t, message.getMessageId(), message.getFrom()});
+ MessagingService.instance.sendOneWay(responseMessage, message.getFrom());
+ }
+ catch (IOException e)
+ {
+ logger.error("Error in truncation", e);
+ throw new RuntimeException("Error in truncation", e);
+ }
+ catch (InterruptedException e)
+ {
+ logger.error("Error in truncation", e);
+ throw new RuntimeException("Error in truncation", e);
+ }
+ catch (ExecutionException e)
+ {
+ logger.error("Error in truncation", e);
+ throw new RuntimeException("Error in truncation", e);
+ }
+ }
+
+ private static void respondError(Truncation t, Message truncateRequestMessage) throws IOException
+ {
+ TruncateResponse response = new TruncateResponse(t.keyspace, t.columnFamily, false);
+ Message responseMessage = TruncateResponse.makeTruncateResponseMessage(truncateRequestMessage, response);
+ MessagingService.instance.sendOneWay(responseMessage, truncateRequestMessage.getFrom());
+ }
+}
Added: cassandra/trunk/src/java/org/apache/cassandra/db/Truncation.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Truncation.java?rev=942817&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Truncation.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Truncation.java Mon May 10 16:37:28 2010
@@ -0,0 +1,108 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
+
+/**
+ * A truncate operation descriptor
+ *
+ * @author rantav@gmail.com
+ *
+ */
+public class Truncation
+{
+ private static ICompactSerializer<Truncation> serializer;
+
+ public final String keyspace;
+ public final String columnFamily;
+
+ static
+ {
+ serializer = new TruncationSerializer();
+ }
+
+ public static ICompactSerializer<Truncation> serializer()
+ {
+ return serializer;
+ }
+
+ public Truncation(String keyspace, String columnFamily)
+ {
+ this.keyspace = keyspace;
+ this.columnFamily = columnFamily;
+ }
+
+ /**
+ * This is equivalent to calling commit. Applies the changes to
+ * to the table that is obtained by calling Table.open().
+ */
+ public void apply() throws IOException
+ {
+ Table.open(keyspace).getColumnFamilyStore(columnFamily).truncate();
+ }
+
+ public Message makeTruncationMessage() throws IOException
+ {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ DataOutputStream dos = new DataOutputStream(bos);
+ serializer().serialize(this, dos);
+ return new Message(FBUtilities.getLocalAddress(), StageManager.MUTATION_STAGE, StorageService.Verb.TRUNCATE,
+ bos.toByteArray());
+ }
+
+
+ public DataOutputBuffer getSerializedBuffer() throws IOException
+ {
+ DataOutputBuffer buffer = new DataOutputBuffer();
+ Truncation.serializer().serialize(this, buffer);
+ return buffer;
+ }
+
+ public String toString()
+ {
+ return "Truncation(" + "keyspace='" + keyspace + '\'' + ", cf='" + columnFamily + "\')";
+ }
+}
+
+class TruncationSerializer implements ICompactSerializer<Truncation>
+{
+ public void serialize(Truncation t, DataOutputStream dos) throws IOException
+ {
+ dos.writeUTF(t.keyspace);
+ dos.writeUTF(t.columnFamily);
+ }
+
+ public Truncation deserialize(DataInputStream dis) throws IOException
+ {
+ String keyspace = dis.readUTF();
+ String columnFamily = dis.readUTF();
+ return new Truncation(keyspace, columnFamily);
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/RowIndexedReader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/RowIndexedReader.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/RowIndexedReader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/RowIndexedReader.java Mon May 10 16:37:28 2010
@@ -70,10 +70,11 @@ class RowIndexedReader extends SSTableRe
RowIndexedReader(Descriptor desc,
IPartitioner partitioner,
IndexSummary indexSummary,
- BloomFilter bloomFilter)
+ BloomFilter bloomFilter,
+ long maxDataAge)
throws IOException
{
- super(desc, partitioner);
+ super(desc, partitioner, maxDataAge);
if (DatabaseDescriptor.getIndexAccessMode() == Config.DiskAccessMode.mmap)
{
@@ -116,7 +117,7 @@ class RowIndexedReader extends SSTableRe
RowIndexedReader(Descriptor desc, IPartitioner partitioner) throws IOException
{
- this(desc, partitioner, null, null);
+ this(desc, partitioner, null, null, System.currentTimeMillis());
}
public static RowIndexedReader open(Descriptor desc, IPartitioner partitioner) throws IOException
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableReader.java Mon May 10 16:37:28 2010
@@ -82,6 +82,20 @@ public abstract class SSTableReader exte
new Thread(runnable, "SSTABLE-DELETER").start();
}};
+ /**
+ * maxDataAge is a timestamp in local server time (e.g. System.currentTimeMilli) which represents an uppper bound
+ * to the newest piece of data stored in the sstable. In other words, this sstable does not contain items created
+ * later than maxDataAge.
+ *
+ * The field is not serialized to disk, so relying on it for more than what truncate does is not advised.
+ *
+ * When a new sstable is flushed, maxDataAge is set to the time of creation.
+ * When a sstable is created from compaction, maxDataAge is set to max of all merged tables.
+ *
+ * The age is in milliseconds since epoc and is local to this host.
+ */
+ public final long maxDataAge;
+
public static int indexInterval()
{
return IndexSummary.INDEX_INTERVAL;
@@ -166,9 +180,10 @@ public abstract class SSTableReader exte
}
}
- protected SSTableReader(Descriptor desc, IPartitioner partitioner)
+ protected SSTableReader(Descriptor desc, IPartitioner partitioner, long maxDataAge)
{
super(desc, partitioner);
+ this.maxDataAge = maxDataAge;
}
private volatile SSTableDeletingReference phantomReference;
@@ -266,4 +281,15 @@ public abstract class SSTableReader exte
? Column.serializer()
: SuperColumn.serializer(getColumnComparator());
}
+
+ /**
+ * Tests if the sstable contains data newer than the given age param (in localhost currentMilli time).
+ * This works in conjunction with maxDataAge which is an upper bound on the create of data in this sstable.
+ * @param age The age to compare the maxDataAre of this sstable. Measured in millisec since epoc on this host
+ * @return True iff this sstable contains data that's newer than the given age parameter.
+ */
+ public boolean newSince(long age)
+ {
+ return maxDataAge > age;
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableTracker.java Mon May 10 16:37:28 2010
@@ -20,7 +20,6 @@
package org.apache.cassandra.io.sstable;
import java.util.*;
-import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.cache.JMXInstrumentedCache;
@@ -69,11 +68,13 @@ public class SSTableTracker implements I
sstable.setTrackedBy(this);
}
+ long maxDataAge = -1;
for (SSTableReader sstable : oldSSTables)
{
boolean removed = sstablesNew.remove(sstable);
assert removed;
sstable.markCompacted();
+ maxDataAge = Math.max(maxDataAge, sstable.maxDataAge);
liveSize.addAndGet(-sstable.bytesOnDisk());
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/io/sstable/SSTableWriter.java Mon May 10 16:37:28 2010
@@ -130,6 +130,11 @@ public class SSTableWriter extends SSTab
public SSTableReader closeAndOpenReader() throws IOException
{
+ return closeAndOpenReader(System.currentTimeMillis());
+ }
+
+ public SSTableReader closeAndOpenReader(long maxDataAge) throws IOException
+ {
// bloom filter
FileOutputStream fos = new FileOutputStream(filterFilename());
DataOutputStream stream = new DataOutputStream(fos);
@@ -151,7 +156,7 @@ public class SSTableWriter extends SSTab
rename(getFilename());
indexSummary.complete();
- return new RowIndexedReader(newdesc, partitioner, indexSummary, bf);
+ return new RowIndexedReader(newdesc, partitioner, indexSummary, bf, maxDataAge);
}
static String rename(String tmpFilename)
@@ -180,5 +185,4 @@ public class SSTableWriter extends SSTab
dataFileName = SSTableWriter.rename(dataFileName);
return SSTableReader.open(dataFileName);
}
-
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageProxy.java Mon May 10 16:37:28 2010
@@ -22,29 +22,39 @@ import java.io.DataInputStream;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+
import javax.management.MBeanServer;
import javax.management.ObjectName;
-import org.apache.cassandra.config.CFMetaData;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.commons.lang.ArrayUtils;
-import org.apache.commons.lang.StringUtils;
-
-import com.google.common.collect.AbstractIterator;
-import com.google.common.collect.Multimap;
import org.apache.cassandra.concurrent.StageManager;
+import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.RangeSliceCommand;
+import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.ReadResponse;
+import org.apache.cassandra.db.Row;
+import org.apache.cassandra.db.RowMutation;
+import org.apache.cassandra.db.Table;
+import org.apache.cassandra.db.Truncation;
import org.apache.cassandra.dht.AbstractBounds;
-import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.locator.AbstractReplicationStrategy;
import org.apache.cassandra.locator.TokenMetadata;
import org.apache.cassandra.net.IAsyncResult;
@@ -56,6 +66,13 @@ import org.apache.cassandra.utils.FBUtil
import org.apache.cassandra.utils.LatencyTracker;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.WrappedRunnable;
+import org.apache.commons.lang.ArrayUtils;
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.AbstractIterator;
+import com.google.common.collect.Multimap;
public class StorageProxy implements StorageProxyMBean
@@ -770,4 +787,76 @@ public class StorageProxy implements Sto
return row;
}
}
+
+ /**
+ * Performs the truncate operatoin, which effectively deletes all data from
+ * the column family cfname
+ * @param keyspace
+ * @param cfname
+ * @throws UnavailableException If some of the hosts in the ring are down.
+ * @throws TimeoutException
+ * @throws IOException
+ */
+ public static void truncateBlocking(String keyspace, String cfname) throws UnavailableException, TimeoutException, IOException
+ {
+ logger.debug("Starting a blocking truncate operation on keyspace {}, CF ", keyspace, cfname);
+ if (isAnyHostDown())
+ {
+ logger.info("Cannot perform truncate, some hosts are down");
+ // Since the truncate operation is so aggressive and is typically only
+ // invoked by an admin, for simplicity we require that all nodes are up
+ // to perform the operation.
+ throw new UnavailableException();
+ }
+
+ Set<InetAddress> allEndpoints = Gossiper.instance.getLiveMembers();
+ int blockFor = allEndpoints.size();
+ allEndpoints.remove(FBUtilities.getLocalAddress());
+ final TruncateResponseHandler responseHandler = new TruncateResponseHandler(blockFor, keyspace);
+
+ // Send out the truncate calls and track the responses with the callbacks.
+ logger.debug("Starting to send truncate messages to hosts {}", allEndpoints);
+ truncateLocal(keyspace, cfname, responseHandler);
+ truncateRemotes(keyspace, cfname, allEndpoints, responseHandler);
+
+ // Wait for all
+ logger.debug("Sent all truncate messages, now waiting for {} responses", blockFor);
+ responseHandler.get();
+ logger.debug("truncate done");
+ }
+
+ private static void truncateRemotes(String keyspace, String cfname, Set<InetAddress> hosts,
+ TruncateResponseHandler responseHandler) throws IOException
+ {
+ Truncation truncation = new Truncation(keyspace, cfname);
+ Message message = truncation.makeTruncationMessage();
+ MessagingService.instance.sendRR(message, hosts.toArray(new InetAddress[]{}), responseHandler);
+ }
+
+ /**
+ * Asks the gossiper if there are any nodes that are currently down.
+ * @return true if the gossiper thinks all nodes are up.
+ */
+ private static boolean isAnyHostDown()
+ {
+ return !Gossiper.instance.getUnreachableMembers().isEmpty();
+ }
+
+ private static void truncateLocal(final String keyspace, final String cfname,
+ final TruncateResponseHandler responseHandler)
+ {
+ logger.debug("truncating locally (keyspace: {}, CF: {})", keyspace, cfname);
+ Runnable runnable = new WrappedRunnable()
+ {
+ public void runMayThrow() throws IOException, InterruptedException, ExecutionException
+ {
+ // truncate, blocking
+ Table.open(keyspace).truncate(cfname);
+ responseHandler.localResponse();
+ logger.debug("Finished truncating local (keyspace: {}, CF: {})", keyspace, cfname);
+ }
+ };
+ // TODO(ran): Is this correct? Should the truncate operation operate on the row mutation stage?
+ StageManager.getStage(StageManager.MUTATION_STAGE).execute(runnable);
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageService.java Mon May 10 16:37:28 2010
@@ -29,6 +29,7 @@ import java.util.concurrent.ExecutorServ
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.net.InetAddress;
import javax.management.*;
@@ -110,6 +111,7 @@ public class StorageService implements I
GOSSIP_DIGEST_ACK2,
DEFINITIONS_ANNOUNCE,
DEFINITIONS_UPDATE_RESPONSE,
+ TRUNCATE,
;
// remember to add new verbs at the end, since we serialize by ordinal
}
@@ -234,6 +236,7 @@ public class StorageService implements I
MessagingService.instance.registerVerbHandlers(Verb.DEFINITIONS_ANNOUNCE, new DefinitionsAnnounceVerbHandler());
MessagingService.instance.registerVerbHandlers(Verb.DEFINITIONS_UPDATE_RESPONSE, new DefinitionsUpdateResponseVerbHandler());
+ MessagingService.instance.registerVerbHandlers(Verb.TRUNCATE, new TruncateVerbHandler());
replicationStrategies = new HashMap<String, AbstractReplicationStrategy>();
for (String table : DatabaseDescriptor.getNonSystemTables())
@@ -1655,4 +1658,10 @@ public class StorageService implements I
tokenMetadata_ = tmd;
return old;
}
+
+ @Override
+ public void truncate(String keyspace, String columnFamily) throws UnavailableException, TimeoutException, IOException
+ {
+ StorageProxy.truncateBlocking(keyspace, columnFamily);
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/StorageServiceMBean.java Mon May 10 16:37:28 2010
@@ -19,15 +19,16 @@
package org.apache.cassandra.service;
import java.io.IOException;
-import java.net.UnknownHostException;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeoutException;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.thrift.UnavailableException;
+
import java.net.InetAddress;
@@ -176,4 +177,18 @@ public interface StorageServiceMBean
* @todo: deprecate in 0.7+1, remove in 0.7+2.
*/
public void loadSchemaFromYAML() throws ConfigurationException, IOException;
+
+ /**
+ * Truncates (deletes) the given columnFamily from the provided keyspace.
+ * Calling truncate results in actual deletion of all data in the cluster
+ * under the given columnFamily and it will fail unless all hosts are up.
+ * All data in the given column family will be deleted, but its definition
+ * will not be affected.
+ *
+ * @param keyspace The keyspace to delete from
+ * @param columnFamily The column family to delete data from.
+ *
+ * @throws UnavailableException if some of the hosts in the ring are down.
+ */
+ public void truncate(String keyspace, String columnFamily) throws UnavailableException, TimeoutException, IOException;
}
Added: cassandra/trunk/src/java/org/apache/cassandra/service/TruncateResponseHandler.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/TruncateResponseHandler.java?rev=942817&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/TruncateResponseHandler.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/TruncateResponseHandler.java Mon May 10 16:37:28 2010
@@ -0,0 +1,103 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.service;
+
+import java.util.Collection;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.net.IAsyncCallback;
+import org.apache.cassandra.net.Message;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.utils.SimpleCondition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TruncateResponseHandler implements IAsyncCallback
+{
+ protected static final Logger logger = LoggerFactory.getLogger(TruncateResponseHandler.class);
+ protected final SimpleCondition condition = new SimpleCondition();
+ private final int responseCount;
+ protected final Collection<Message> responses;
+ protected AtomicInteger localResponses = new AtomicInteger(0);
+ private final long startTime;
+
+ public TruncateResponseHandler(int responseCount, String table)
+ {
+ // at most one node per range can bootstrap at a time, and these will be added to the write until
+ // bootstrap finishes (at which point we no longer need to write to the old ones).
+ assert 1 <= responseCount: "invalid response count " + responseCount;
+
+ this.responseCount = responseCount;
+ responses = new LinkedBlockingQueue<Message>();
+ startTime = System.currentTimeMillis();
+ }
+
+ public void get() throws TimeoutException
+ {
+ try
+ {
+ long timeout = DatabaseDescriptor.getRpcTimeout() - (System.currentTimeMillis() - startTime);
+ boolean success;
+ try
+ {
+ success = condition.await(timeout, TimeUnit.MILLISECONDS);
+ }
+ catch (InterruptedException ex)
+ {
+ throw new AssertionError(ex);
+ }
+
+ if (!success)
+ {
+ throw new TimeoutException("Operation timed out - received only " + responses.size() + localResponses + " responses");
+ }
+ }
+ finally
+ {
+ for (Message response : responses)
+ {
+ MessagingService.removeRegisteredCallback(response.getMessageId());
+ }
+ }
+ }
+
+ public void response(Message message)
+ {
+ responses.add(message);
+ maybeSignal();
+ }
+
+ public void localResponse()
+ {
+ localResponses.addAndGet(1);
+ maybeSignal();
+ }
+
+ private void maybeSignal()
+ {
+ if (responses.size() + localResponses.get() >= responseCount)
+ {
+ condition.signal();
+ }
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java Mon May 10 16:37:28 2010
@@ -824,6 +824,25 @@ public class CassandraServer implements
}
}
+ @Override
+ public void truncate(String keyspace, String cfname) throws InvalidRequestException, UnavailableException, TException
+ {
+ logger.debug("truncating {} in {}", cfname, keyspace);
+ checkKeyspaceAndLoginAuthorized(AccessLevel.FULL);
+ try
+ {
+ StorageProxy.truncateBlocking(keyspace, cfname);
+ }
+ catch (TimeoutException e)
+ {
+ throw (UnavailableException) new UnavailableException().initCause(e);
+ }
+ catch (IOException e)
+ {
+ throw (UnavailableException) new UnavailableException().initCause(e);
+ }
+ }
+
public void set_keyspace(String keyspace) throws InvalidRequestException, TException {
if (DatabaseDescriptor.getTableDefinition(keyspace) == null)
{
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/ClusterCmd.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/ClusterCmd.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/ClusterCmd.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/ClusterCmd.java Mon May 10 16:37:28 2010
@@ -18,9 +18,10 @@
package org.apache.cassandra.tools;
import java.io.IOException;
+import java.net.InetAddress;
import java.util.List;
import java.util.Set;
-import java.net.InetAddress;
+
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
@@ -139,7 +140,8 @@ public class ClusterCmd {
{
HelpFormatter hf = new HelpFormatter();
String header = String.format(
- "%nAvailable commands: get_endpoints [keyspace] [key], global_snapshot [name], clear_global_snapshot");
+ "%nAvailable commands: get_endpoints [keyspace] [key], global_snapshot [name], clear_global_snapshot," +
+ "truncate <keyspace> <cfname>");
String usage = String.format("java %s -host <arg> <command>%n", ClusterCmd.class.getName());
hf.printHelp(usage, "", options, header);
}
@@ -219,6 +221,11 @@ public class ClusterCmd {
}
}
+ public void truncate(String tableName, String cfName)
+ {
+ probe.truncate(tableName, cfName);
+ }
+
/**
* @param args
*/
@@ -275,6 +282,16 @@ public class ClusterCmd {
{
clusterCmd.clearGlobalSnapshot();
}
+ else if (cmdName.equals("truncate"))
+ {
+ if (arguments.length != 3)
+ {
+ System.err.println("truncate requires <keyspace> and <columnfamily> arguments");
+ }
+ String tableName = arguments[1];
+ String cfName = arguments[2];
+ clusterCmd.truncate(tableName, cfName);
+ }
else
{
System.err.println("Unrecognized command: " + cmdName + ".");
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/NodeCmd.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/NodeCmd.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/NodeCmd.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/NodeCmd.java Mon May 10 16:37:28 2010
@@ -79,7 +79,7 @@ public class NodeCmd {
"%nAvailable commands: ring, info, cleanup, compact, cfstats, snapshot [snapshotname], clearsnapshot, " +
"tpstats, flush, drain, repair, decommission, move, loadbalance, removetoken, " +
"setcachecapacity <keyspace> <cfname> <keycachecapacity> <rowcachecapacity>, " +
- "getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold])" +
+ "getcompactionthreshold, setcompactionthreshold [minthreshold] ([maxthreshold]), " +
"streams [host]");
String usage = String.format("java %s --host <arg> <command>%n", NodeCmd.class.getName());
hf.printHelp(usage, "", options, header);
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/NodeProbe.java Mon May 10 16:37:28 2010
@@ -34,6 +34,7 @@ import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeoutException;
import javax.management.*;
import javax.management.remote.JMXConnector;
@@ -49,6 +50,8 @@ import org.apache.cassandra.dht.Range;
import org.apache.cassandra.service.StorageServiceMBean;
import org.apache.cassandra.streaming.StreamingService;
import org.apache.cassandra.streaming.StreamingServiceMBean;
+import org.apache.cassandra.thrift.UnavailableException;
+
import static org.apache.cassandra.utils.FBUtilities.UTF8;
/**
@@ -437,6 +440,26 @@ public class NodeProbe
{
return ssProxy.getOperationMode();
}
+
+ public void truncate(String tableName, String cfName)
+ {
+ try
+ {
+ ssProxy.truncate(tableName, cfName);
+ }
+ catch (UnavailableException e)
+ {
+ throw new RuntimeException("Error while executing truncate", e);
+ }
+ catch (TimeoutException e)
+ {
+ throw new RuntimeException("Error while executing truncate", e);
+ }
+ catch (IOException e)
+ {
+ throw new RuntimeException("Error while executing truncate", e);
+ }
+ }
}
class ColumnFamilyStoreMBeanIterator implements Iterator<Map.Entry<String, ColumnFamilyStoreMBean>>
Modified: cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/tools/SSTableImport.java Mon May 10 16:37:28 2010
@@ -30,7 +30,7 @@ import org.apache.cassandra.db.filter.Qu
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.sstable.SSTableWriter;
-import org.apache.cassandra.utils.FBUtilities;
+
import static org.apache.cassandra.utils.FBUtilities.hexToBytes;
import org.apache.commons.cli.*;
import org.json.simple.JSONArray;
Modified: cassandra/trunk/test/system/test_thrift_server.py
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/system/test_thrift_server.py?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/test/system/test_thrift_server.py (original)
+++ cassandra/trunk/test/system/test_thrift_server.py Mon May 10 16:37:28 2010
@@ -1143,3 +1143,18 @@ class TestMutations(ThriftTester):
client.remove('key1', ColumnPath('Standard1', column='cttl5'), 1, ConsistencyLevel.ONE)
_expect_missing(lambda: client.get('key1', ColumnPath('Standard1', column='ctt5'), ConsistencyLevel.ONE))
+class TestTruncate(ThriftTester):
+ def test_truncate(self):
+ _set_keyspace('Keyspace1')
+
+ _insert_simple()
+ _insert_super()
+
+ # truncate Standard1
+ client.truncate('Keyspace1', 'Standard1')
+ assert _big_slice('key1', ColumnParent('Standard1')) == []
+
+ # truncate Super1
+ client.truncate('Keyspace1', 'Super1')
+ assert _big_slice('key1', ColumnParent('Super1')) == []
+ assert _big_slice('key1', ColumnParent('Super1', 'sc1')) == []
Added: cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java?rev=942817&view=auto
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java (added)
+++ cassandra/trunk/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java Mon May 10 16:37:28 2010
@@ -0,0 +1,97 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied. See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.apache.cassandra.db;
+
+import static org.apache.cassandra.Util.column;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.cassandra.CleanupHelper;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.db.commitlog.CommitLog;
+import org.apache.cassandra.db.filter.QueryFilter;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.junit.Test;
+
+/**
+ * Test for the truncate operation.
+ *
+ * @author Ran Tavory (rantav@gmail.com)
+ *
+ */
+public class RecoveryManagerTruncateTest extends CleanupHelper
+{
+ @Test
+ public void testTruncate() throws IOException, ExecutionException, InterruptedException
+ {
+ Table table = Table.open("Keyspace1");
+ ColumnFamilyStore cfs = table.getColumnFamilyStore("Standard1");
+
+ RowMutation rm;
+ ColumnFamily cf;
+
+ // trucate clears memtable
+ rm = new RowMutation("Keyspace1", "keymulti".getBytes());
+ cf = ColumnFamily.create("Keyspace1", "Standard1");
+ cf.addColumn(column("col1", "val1", 1L));
+ rm.add(cf);
+ rm.apply();
+
+ // Make sure data was written
+ assertNotNull(getFromTable(table, "Standard1", "keymulti", "col1"));
+
+ // and now truncate it
+ cfs.truncate().get();
+ CommitLog.recover();
+
+ // and validate truncation.
+ assertNull(getFromTable(table, "Standard1", "keymulti", "col1"));
+
+ // truncate clears sstable
+ rm = new RowMutation("Keyspace1", "keymulti".getBytes());
+ cf = ColumnFamily.create("Keyspace1", "Standard1");
+ cf.addColumn(column("col1", "val1", 1L));
+ rm.add(cf);
+ rm.apply();
+ cfs.forceBlockingFlush();
+ cfs.truncate().get();
+ CommitLog.recover();
+ assertNull(getFromTable(table, "Standard1", "keymulti", "col1"));
+ }
+
+ private IColumn getFromTable(Table table, String cfName, String keyName, String columnName)
+ {
+ ColumnFamily cf;
+ ColumnFamilyStore cfStore = table.getColumnFamilyStore(cfName);
+ if (cfStore == null)
+ {
+ return null;
+ }
+ cf = cfStore.getColumnFamily(QueryFilter.getNamesFilter(
+ Util.dk(keyName), new QueryPath(cfName), columnName.getBytes()));
+ if (cf == null)
+ {
+ return null;
+ }
+ return cf.getColumn(columnName.getBytes());
+ }
+}
Modified: cassandra/trunk/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/io/sstable/SSTableUtils.java Mon May 10 16:37:28 2010
@@ -24,7 +24,6 @@ import java.io.IOException;
import java.util.*;
-import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.*;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.io.util.DataOutputBuffer;
Modified: cassandra/trunk/test/unit/org/apache/cassandra/tools/SSTableExportTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/tools/SSTableExportTest.java?rev=942817&r1=942816&r2=942817&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/tools/SSTableExportTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/tools/SSTableExportTest.java Mon May 10 16:37:28 2010
@@ -27,7 +27,6 @@ import java.util.Arrays;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamily;
-import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.dht.IPartitioner;