mirror of https://github.com/apache/cassandra
allow specifying a subset of columnfamilies to check for get_key_range.
patch by jbellis; reviewed by Eric Evans for CASSANDRA-152 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@779261 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis
parent
97c022f5ee
commit
bb0d6dfa20
|
|
@ -122,7 +122,7 @@ service Cassandra {
|
|||
oneway void touch(1:string key, 2:bool fData),
|
||||
|
||||
# range query: returns matching keys
|
||||
list<string> get_key_range(1:string tablename, 2:string startWith="", 3:string stopAt="", 4:i32 maxResults=1000) throws (1: InvalidRequestException ire),
|
||||
list<string> get_key_range(1:string tablename, 2:list<string> columnFamilies=[], 3:string startWith="", 4:string stopAt="", 5:i32 maxResults=1000) throws (1: InvalidRequestException ire),
|
||||
|
||||
/////////////////////////////////////////////////////////////////////////////////////
|
||||
// The following are beta APIs being introduced for CLI and/or CQL support. //
|
||||
|
|
|
|||
|
|
@ -52,7 +52,7 @@ public class Cassandra {
|
|||
|
||||
public void touch(String key, boolean fData) throws TException;
|
||||
|
||||
public List<String> get_key_range(String tablename, String startWith, String stopAt, int maxResults) throws InvalidRequestException, TException;
|
||||
public List<String> get_key_range(String tablename, List<String> columnFamilies, String startWith, String stopAt, int maxResults) throws InvalidRequestException, TException;
|
||||
|
||||
public String getStringProperty(String propertyName) throws TException;
|
||||
|
||||
|
|
@ -675,17 +675,18 @@ public class Cassandra {
|
|||
oprot_.getTransport().flush();
|
||||
}
|
||||
|
||||
public List<String> get_key_range(String tablename, String startWith, String stopAt, int maxResults) throws InvalidRequestException, TException
|
||||
public List<String> get_key_range(String tablename, List<String> columnFamilies, String startWith, String stopAt, int maxResults) throws InvalidRequestException, TException
|
||||
{
|
||||
send_get_key_range(tablename, startWith, stopAt, maxResults);
|
||||
send_get_key_range(tablename, columnFamilies, startWith, stopAt, maxResults);
|
||||
return recv_get_key_range();
|
||||
}
|
||||
|
||||
public void send_get_key_range(String tablename, String startWith, String stopAt, int maxResults) throws TException
|
||||
public void send_get_key_range(String tablename, List<String> columnFamilies, String startWith, String stopAt, int maxResults) throws TException
|
||||
{
|
||||
oprot_.writeMessageBegin(new TMessage("get_key_range", TMessageType.CALL, seqid_));
|
||||
get_key_range_args args = new get_key_range_args();
|
||||
args.tablename = tablename;
|
||||
args.columnFamilies = columnFamilies;
|
||||
args.startWith = startWith;
|
||||
args.stopAt = stopAt;
|
||||
args.maxResults = maxResults;
|
||||
|
|
@ -1333,7 +1334,7 @@ public class Cassandra {
|
|||
iprot.readMessageEnd();
|
||||
get_key_range_result result = new get_key_range_result();
|
||||
try {
|
||||
result.success = iface_.get_key_range(args.tablename, args.startWith, args.stopAt, args.maxResults);
|
||||
result.success = iface_.get_key_range(args.tablename, args.columnFamilies, args.startWith, args.stopAt, args.maxResults);
|
||||
} catch (InvalidRequestException ire) {
|
||||
result.ire = ire;
|
||||
} catch (Throwable th) {
|
||||
|
|
@ -12339,18 +12340,21 @@ public class Cassandra {
|
|||
public static class get_key_range_args implements TBase, java.io.Serializable, Cloneable {
|
||||
private static final TStruct STRUCT_DESC = new TStruct("get_key_range_args");
|
||||
private static final TField TABLENAME_FIELD_DESC = new TField("tablename", TType.STRING, (short)1);
|
||||
private static final TField START_WITH_FIELD_DESC = new TField("startWith", TType.STRING, (short)2);
|
||||
private static final TField STOP_AT_FIELD_DESC = new TField("stopAt", TType.STRING, (short)3);
|
||||
private static final TField MAX_RESULTS_FIELD_DESC = new TField("maxResults", TType.I32, (short)4);
|
||||
private static final TField COLUMN_FAMILIES_FIELD_DESC = new TField("columnFamilies", TType.LIST, (short)2);
|
||||
private static final TField START_WITH_FIELD_DESC = new TField("startWith", TType.STRING, (short)3);
|
||||
private static final TField STOP_AT_FIELD_DESC = new TField("stopAt", TType.STRING, (short)4);
|
||||
private static final TField MAX_RESULTS_FIELD_DESC = new TField("maxResults", TType.I32, (short)5);
|
||||
|
||||
public String tablename;
|
||||
public static final int TABLENAME = 1;
|
||||
public List<String> columnFamilies;
|
||||
public static final int COLUMNFAMILIES = 2;
|
||||
public String startWith;
|
||||
public static final int STARTWITH = 2;
|
||||
public static final int STARTWITH = 3;
|
||||
public String stopAt;
|
||||
public static final int STOPAT = 3;
|
||||
public static final int STOPAT = 4;
|
||||
public int maxResults;
|
||||
public static final int MAXRESULTS = 4;
|
||||
public static final int MAXRESULTS = 5;
|
||||
|
||||
private final Isset __isset = new Isset();
|
||||
private static final class Isset implements java.io.Serializable {
|
||||
|
|
@ -12360,6 +12364,9 @@ public class Cassandra {
|
|||
public static final Map<Integer, FieldMetaData> metaDataMap = Collections.unmodifiableMap(new HashMap<Integer, FieldMetaData>() {{
|
||||
put(TABLENAME, new FieldMetaData("tablename", TFieldRequirementType.DEFAULT,
|
||||
new FieldValueMetaData(TType.STRING)));
|
||||
put(COLUMNFAMILIES, new FieldMetaData("columnFamilies", TFieldRequirementType.DEFAULT,
|
||||
new ListMetaData(TType.LIST,
|
||||
new FieldValueMetaData(TType.STRING))));
|
||||
put(STARTWITH, new FieldMetaData("startWith", TFieldRequirementType.DEFAULT,
|
||||
new FieldValueMetaData(TType.STRING)));
|
||||
put(STOPAT, new FieldMetaData("stopAt", TFieldRequirementType.DEFAULT,
|
||||
|
|
@ -12373,6 +12380,8 @@ public class Cassandra {
|
|||
}
|
||||
|
||||
public get_key_range_args() {
|
||||
this.columnFamilies = new ArrayList<String>();
|
||||
|
||||
this.startWith = "";
|
||||
|
||||
this.stopAt = "";
|
||||
|
|
@ -12383,12 +12392,14 @@ public class Cassandra {
|
|||
|
||||
public get_key_range_args(
|
||||
String tablename,
|
||||
List<String> columnFamilies,
|
||||
String startWith,
|
||||
String stopAt,
|
||||
int maxResults)
|
||||
{
|
||||
this();
|
||||
this.tablename = tablename;
|
||||
this.columnFamilies = columnFamilies;
|
||||
this.startWith = startWith;
|
||||
this.stopAt = stopAt;
|
||||
this.maxResults = maxResults;
|
||||
|
|
@ -12402,6 +12413,13 @@ public class Cassandra {
|
|||
if (other.isSetTablename()) {
|
||||
this.tablename = other.tablename;
|
||||
}
|
||||
if (other.isSetColumnFamilies()) {
|
||||
List<String> __this__columnFamilies = new ArrayList<String>();
|
||||
for (String other_element : other.columnFamilies) {
|
||||
__this__columnFamilies.add(other_element);
|
||||
}
|
||||
this.columnFamilies = __this__columnFamilies;
|
||||
}
|
||||
if (other.isSetStartWith()) {
|
||||
this.startWith = other.startWith;
|
||||
}
|
||||
|
|
@ -12440,6 +12458,44 @@ public class Cassandra {
|
|||
}
|
||||
}
|
||||
|
||||
public int getColumnFamiliesSize() {
|
||||
return (this.columnFamilies == null) ? 0 : this.columnFamilies.size();
|
||||
}
|
||||
|
||||
public java.util.Iterator<String> getColumnFamiliesIterator() {
|
||||
return (this.columnFamilies == null) ? null : this.columnFamilies.iterator();
|
||||
}
|
||||
|
||||
public void addToColumnFamilies(String elem) {
|
||||
if (this.columnFamilies == null) {
|
||||
this.columnFamilies = new ArrayList<String>();
|
||||
}
|
||||
this.columnFamilies.add(elem);
|
||||
}
|
||||
|
||||
public List<String> getColumnFamilies() {
|
||||
return this.columnFamilies;
|
||||
}
|
||||
|
||||
public void setColumnFamilies(List<String> columnFamilies) {
|
||||
this.columnFamilies = columnFamilies;
|
||||
}
|
||||
|
||||
public void unsetColumnFamilies() {
|
||||
this.columnFamilies = null;
|
||||
}
|
||||
|
||||
// Returns true if field columnFamilies is set (has been asigned a value) and false otherwise
|
||||
public boolean isSetColumnFamilies() {
|
||||
return this.columnFamilies != null;
|
||||
}
|
||||
|
||||
public void setColumnFamiliesIsSet(boolean value) {
|
||||
if (!value) {
|
||||
this.columnFamilies = null;
|
||||
}
|
||||
}
|
||||
|
||||
public String getStartWith() {
|
||||
return this.startWith;
|
||||
}
|
||||
|
|
@ -12518,6 +12574,14 @@ public class Cassandra {
|
|||
}
|
||||
break;
|
||||
|
||||
case COLUMNFAMILIES:
|
||||
if (value == null) {
|
||||
unsetColumnFamilies();
|
||||
} else {
|
||||
setColumnFamilies((List<String>)value);
|
||||
}
|
||||
break;
|
||||
|
||||
case STARTWITH:
|
||||
if (value == null) {
|
||||
unsetStartWith();
|
||||
|
|
@ -12552,6 +12616,9 @@ public class Cassandra {
|
|||
case TABLENAME:
|
||||
return getTablename();
|
||||
|
||||
case COLUMNFAMILIES:
|
||||
return getColumnFamilies();
|
||||
|
||||
case STARTWITH:
|
||||
return getStartWith();
|
||||
|
||||
|
|
@ -12571,6 +12638,8 @@ public class Cassandra {
|
|||
switch (fieldID) {
|
||||
case TABLENAME:
|
||||
return isSetTablename();
|
||||
case COLUMNFAMILIES:
|
||||
return isSetColumnFamilies();
|
||||
case STARTWITH:
|
||||
return isSetStartWith();
|
||||
case STOPAT:
|
||||
|
|
@ -12604,6 +12673,15 @@ public class Cassandra {
|
|||
return false;
|
||||
}
|
||||
|
||||
boolean this_present_columnFamilies = true && this.isSetColumnFamilies();
|
||||
boolean that_present_columnFamilies = true && that.isSetColumnFamilies();
|
||||
if (this_present_columnFamilies || that_present_columnFamilies) {
|
||||
if (!(this_present_columnFamilies && that_present_columnFamilies))
|
||||
return false;
|
||||
if (!this.columnFamilies.equals(that.columnFamilies))
|
||||
return false;
|
||||
}
|
||||
|
||||
boolean this_present_startWith = true && this.isSetStartWith();
|
||||
boolean that_present_startWith = true && that.isSetStartWith();
|
||||
if (this_present_startWith || that_present_startWith) {
|
||||
|
|
@ -12657,6 +12735,23 @@ public class Cassandra {
|
|||
TProtocolUtil.skip(iprot, field.type);
|
||||
}
|
||||
break;
|
||||
case COLUMNFAMILIES:
|
||||
if (field.type == TType.LIST) {
|
||||
{
|
||||
TList _list67 = iprot.readListBegin();
|
||||
this.columnFamilies = new ArrayList<String>(_list67.size);
|
||||
for (int _i68 = 0; _i68 < _list67.size; ++_i68)
|
||||
{
|
||||
String _elem69;
|
||||
_elem69 = iprot.readString();
|
||||
this.columnFamilies.add(_elem69);
|
||||
}
|
||||
iprot.readListEnd();
|
||||
}
|
||||
} else {
|
||||
TProtocolUtil.skip(iprot, field.type);
|
||||
}
|
||||
break;
|
||||
case STARTWITH:
|
||||
if (field.type == TType.STRING) {
|
||||
this.startWith = iprot.readString();
|
||||
|
|
@ -12701,6 +12796,17 @@ public class Cassandra {
|
|||
oprot.writeString(this.tablename);
|
||||
oprot.writeFieldEnd();
|
||||
}
|
||||
if (this.columnFamilies != null) {
|
||||
oprot.writeFieldBegin(COLUMN_FAMILIES_FIELD_DESC);
|
||||
{
|
||||
oprot.writeListBegin(new TList(TType.STRING, this.columnFamilies.size()));
|
||||
for (String _iter70 : this.columnFamilies) {
|
||||
oprot.writeString(_iter70);
|
||||
}
|
||||
oprot.writeListEnd();
|
||||
}
|
||||
oprot.writeFieldEnd();
|
||||
}
|
||||
if (this.startWith != null) {
|
||||
oprot.writeFieldBegin(START_WITH_FIELD_DESC);
|
||||
oprot.writeString(this.startWith);
|
||||
|
|
@ -12731,6 +12837,14 @@ public class Cassandra {
|
|||
}
|
||||
first = false;
|
||||
if (!first) sb.append(", ");
|
||||
sb.append("columnFamilies:");
|
||||
if (this.columnFamilies == null) {
|
||||
sb.append("null");
|
||||
} else {
|
||||
sb.append(this.columnFamilies);
|
||||
}
|
||||
first = false;
|
||||
if (!first) sb.append(", ");
|
||||
sb.append("startWith:");
|
||||
if (this.startWith == null) {
|
||||
sb.append("null");
|
||||
|
|
@ -12982,13 +13096,13 @@ public class Cassandra {
|
|||
case SUCCESS:
|
||||
if (field.type == TType.LIST) {
|
||||
{
|
||||
TList _list67 = iprot.readListBegin();
|
||||
this.success = new ArrayList<String>(_list67.size);
|
||||
for (int _i68 = 0; _i68 < _list67.size; ++_i68)
|
||||
TList _list71 = iprot.readListBegin();
|
||||
this.success = new ArrayList<String>(_list71.size);
|
||||
for (int _i72 = 0; _i72 < _list71.size; ++_i72)
|
||||
{
|
||||
String _elem69;
|
||||
_elem69 = iprot.readString();
|
||||
this.success.add(_elem69);
|
||||
String _elem73;
|
||||
_elem73 = iprot.readString();
|
||||
this.success.add(_elem73);
|
||||
}
|
||||
iprot.readListEnd();
|
||||
}
|
||||
|
|
@ -13024,8 +13138,8 @@ public class Cassandra {
|
|||
oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
|
||||
{
|
||||
oprot.writeListBegin(new TList(TType.STRING, this.success.size()));
|
||||
for (String _iter70 : this.success) {
|
||||
oprot.writeString(_iter70);
|
||||
for (String _iter74 : this.success) {
|
||||
oprot.writeString(_iter74);
|
||||
}
|
||||
oprot.writeListEnd();
|
||||
}
|
||||
|
|
@ -13832,13 +13946,13 @@ public class Cassandra {
|
|||
case SUCCESS:
|
||||
if (field.type == TType.LIST) {
|
||||
{
|
||||
TList _list71 = iprot.readListBegin();
|
||||
this.success = new ArrayList<String>(_list71.size);
|
||||
for (int _i72 = 0; _i72 < _list71.size; ++_i72)
|
||||
TList _list75 = iprot.readListBegin();
|
||||
this.success = new ArrayList<String>(_list75.size);
|
||||
for (int _i76 = 0; _i76 < _list75.size; ++_i76)
|
||||
{
|
||||
String _elem73;
|
||||
_elem73 = iprot.readString();
|
||||
this.success.add(_elem73);
|
||||
String _elem77;
|
||||
_elem77 = iprot.readString();
|
||||
this.success.add(_elem77);
|
||||
}
|
||||
iprot.readListEnd();
|
||||
}
|
||||
|
|
@ -13866,8 +13980,8 @@ public class Cassandra {
|
|||
oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
|
||||
{
|
||||
oprot.writeListBegin(new TList(TType.STRING, this.success.size()));
|
||||
for (String _iter74 : this.success) {
|
||||
oprot.writeString(_iter74);
|
||||
for (String _iter78 : this.success) {
|
||||
oprot.writeString(_iter78);
|
||||
}
|
||||
oprot.writeListEnd();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,10 @@ import java.io.DataOutputStream;
|
|||
import java.io.IOException;
|
||||
import java.io.DataInputStream;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
import java.util.Collections;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
|
||||
import org.apache.cassandra.io.ICompactSerializer;
|
||||
import org.apache.cassandra.io.DataOutputBuffer;
|
||||
|
|
@ -34,13 +38,15 @@ public class RangeCommand
|
|||
private static RangeCommandSerializer serializer = new RangeCommandSerializer();
|
||||
|
||||
public final String table;
|
||||
public final List<String> columnFamilyNames;
|
||||
public final String startWith;
|
||||
public final String stopAt;
|
||||
public final int maxResults;
|
||||
|
||||
public RangeCommand(String table, String startWith, String stopAt, int maxResults)
|
||||
public RangeCommand(String table, List<String> columnFamilyNames, String startWith, String stopAt, int maxResults)
|
||||
{
|
||||
this.table = table;
|
||||
this.columnFamilyNames = Collections.unmodifiableList(columnFamilyNames);
|
||||
this.startWith = startWith;
|
||||
this.stopAt = stopAt;
|
||||
this.maxResults = maxResults;
|
||||
|
|
@ -68,6 +74,7 @@ public class RangeCommand
|
|||
{
|
||||
return "RangeCommand(" +
|
||||
"table='" + table + '\'' +
|
||||
", columnFamilyNames=[" + StringUtils.join(columnFamilyNames, ", ") + "]" +
|
||||
", startWith='" + startWith + '\'' +
|
||||
", stopAt='" + stopAt + '\'' +
|
||||
", maxResults=" + maxResults +
|
||||
|
|
@ -79,6 +86,11 @@ class RangeCommandSerializer implements ICompactSerializer<RangeCommand>
|
|||
{
|
||||
public void serialize(RangeCommand command, DataOutputStream dos) throws IOException
|
||||
{
|
||||
dos.writeInt(command.columnFamilyNames.size());
|
||||
for (String cfName : command.columnFamilyNames)
|
||||
{
|
||||
dos.writeUTF(cfName);
|
||||
}
|
||||
dos.writeUTF(command.table);
|
||||
dos.writeUTF(command.startWith);
|
||||
dos.writeUTF(command.stopAt);
|
||||
|
|
@ -87,6 +99,11 @@ class RangeCommandSerializer implements ICompactSerializer<RangeCommand>
|
|||
|
||||
public RangeCommand deserialize(DataInputStream dis) throws IOException
|
||||
{
|
||||
return new RangeCommand(dis.readUTF(), dis.readUTF(), dis.readUTF(), dis.readInt());
|
||||
String[] cfNames = new String[dis.readInt()];
|
||||
for (int i = 0; i < cfNames.length; i++)
|
||||
{
|
||||
cfNames[i] = dis.readUTF();
|
||||
}
|
||||
return new RangeCommand(dis.readUTF(), Arrays.asList(cfNames), dis.readUTF(), dis.readUTF(), dis.readInt());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,6 @@ import java.io.DataInputStream;
|
|||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.io.File;
|
||||
import java.io.FileNotFoundException;
|
||||
import java.util.concurrent.locks.Lock;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
|
@ -49,7 +48,6 @@ import org.apache.cassandra.net.MessagingService;
|
|||
import org.apache.cassandra.net.io.IStreamComplete;
|
||||
import org.apache.cassandra.net.io.StreamContextManager;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.service.CassandraServer;
|
||||
import org.apache.cassandra.utils.*;
|
||||
import org.apache.log4j.Logger;
|
||||
|
||||
|
|
@ -915,35 +913,43 @@ public class Table
|
|||
* @param maxResults
|
||||
* @return list of keys between startWith and stopAt
|
||||
*/
|
||||
public List<String> getKeyRange(final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException
|
||||
public List<String> getKeyRange(Collection<String> columnFamilyNames, final String startWith, final String stopAt, int maxResults)
|
||||
throws IOException, ExecutionException, InterruptedException
|
||||
{
|
||||
// TODO we need a better way to keep compactions from stomping on reads than One Big Lock per CF.
|
||||
for (String cfName : getApplicationColumnFamilies())
|
||||
if (columnFamilyNames.isEmpty())
|
||||
{
|
||||
columnFamilyNames = getApplicationColumnFamilies();
|
||||
}
|
||||
|
||||
for (String cfName : columnFamilyNames)
|
||||
{
|
||||
getColumnFamilyStore(cfName).getReadLock().lock();
|
||||
}
|
||||
try
|
||||
{
|
||||
return getKeyRangeUnsafe(startWith, stopAt, maxResults);
|
||||
return getKeyRangeUnsafe(columnFamilyNames, startWith, stopAt, maxResults);
|
||||
}
|
||||
finally
|
||||
{
|
||||
for (String cfName : getApplicationColumnFamilies())
|
||||
for (String cfName : columnFamilyNames)
|
||||
{
|
||||
getColumnFamilyStore(cfName).getReadLock().unlock();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private List<String> getKeyRangeUnsafe(final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException
|
||||
private List<String> getKeyRangeUnsafe(final Collection<String> columnFamilyNames, final String startWith, final String stopAt, int maxResults) throws IOException, ExecutionException, InterruptedException
|
||||
{
|
||||
assert !columnFamilyNames.isEmpty(); // checked by the 'safe' method
|
||||
|
||||
// (OPP key decoration is a no-op so using the "decorated" comparator against raw keys is fine)
|
||||
final Comparator<String> comparator = StorageService.getPartitioner().getDecoratedKeyComparator();
|
||||
|
||||
// create a CollatedIterator that will return unique keys from different sources
|
||||
// (current memtable, historical memtables, and SSTables) in the correct order.
|
||||
List<Iterator<String>> iterators = new ArrayList<Iterator<String>>();
|
||||
for (String cfName : getApplicationColumnFamilies())
|
||||
for (String cfName : columnFamilyNames)
|
||||
{
|
||||
ColumnFamilyStore cfs = getColumnFamilyStore(cfName);
|
||||
|
||||
|
|
@ -1002,7 +1008,7 @@ public class Table
|
|||
}
|
||||
// make sure there is actually non-tombstone content associated w/ this key
|
||||
// TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable)
|
||||
for (String cfName : getApplicationColumnFamilies())
|
||||
for (String cfName : columnFamilyNames)
|
||||
{
|
||||
ColumnFamilyStore cfs = getColumnFamilyStore(cfName);
|
||||
ColumnFamily cf = cfs.getColumnFamily(current, cfName, new IdentityFilter(), Integer.MAX_VALUE);
|
||||
|
|
|
|||
|
|
@ -30,21 +30,7 @@ import org.apache.cassandra.config.CFMetaData;
|
|||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.cql.common.CqlResult;
|
||||
import org.apache.cassandra.cql.driver.CqlDriver;
|
||||
import org.apache.cassandra.db.ColumnFamily;
|
||||
import org.apache.cassandra.db.ColumnReadCommand;
|
||||
import org.apache.cassandra.db.ColumnsSinceReadCommand;
|
||||
import org.apache.cassandra.db.SliceByNamesReadCommand;
|
||||
import org.apache.cassandra.db.SliceByRangeReadCommand;
|
||||
import org.apache.cassandra.db.SliceFromReadCommand;
|
||||
import org.apache.cassandra.db.SliceReadCommand;
|
||||
import org.apache.cassandra.db.IColumn;
|
||||
import org.apache.cassandra.db.Row;
|
||||
import org.apache.cassandra.db.RowMutation;
|
||||
import org.apache.cassandra.db.ReadCommand;
|
||||
import org.apache.cassandra.db.Table;
|
||||
import org.apache.cassandra.db.ColumnFamilyNotDefinedException;
|
||||
import org.apache.cassandra.db.TableNotDefinedException;
|
||||
import org.apache.cassandra.db.RangeCommand;
|
||||
import org.apache.cassandra.db.*;
|
||||
import org.apache.cassandra.utils.LogUtil;
|
||||
import org.apache.cassandra.dht.OrderPreservingPartitioner;
|
||||
import org.apache.thrift.TException;
|
||||
|
|
@ -83,16 +69,21 @@ public class CassandraServer implements Cassandra.Iface
|
|||
storageService.start();
|
||||
}
|
||||
|
||||
private void validateCommand(String key, String tablename, String... columnFamilyNames) throws InvalidRequestException
|
||||
private void validateKeyCommand(String key, String tablename, String... columnFamilyNames) throws InvalidRequestException
|
||||
{
|
||||
if (key.isEmpty())
|
||||
{
|
||||
throw new InvalidRequestException("Key may not be empty");
|
||||
}
|
||||
if ( !DatabaseDescriptor.getTables().contains(tablename) )
|
||||
{
|
||||
throw new TableNotDefinedException("Table " + tablename + " does not exist in this schema.");
|
||||
}
|
||||
validateCommand(tablename, columnFamilyNames);
|
||||
}
|
||||
|
||||
private void validateCommand(String tablename, String... columnFamilyNames) throws TableNotDefinedException, ColumnFamilyNotDefinedException
|
||||
{
|
||||
if (!DatabaseDescriptor.getTables().contains(tablename))
|
||||
{
|
||||
throw new TableNotDefinedException("Table " + tablename + " does not exist in this schema.");
|
||||
}
|
||||
Table table = Table.open(tablename);
|
||||
for (String cfName : columnFamilyNames)
|
||||
{
|
||||
|
|
@ -101,12 +92,12 @@ public class CassandraServer implements Cassandra.Iface
|
|||
throw new ColumnFamilyNotDefinedException("Column Family " + cfName + " is invalid.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected ColumnFamily readColumnFamily(ReadCommand command) throws InvalidRequestException
|
||||
}
|
||||
|
||||
protected ColumnFamily readColumnFamily(ReadCommand command) throws InvalidRequestException
|
||||
{
|
||||
String cfName = command.getColumnFamilyName();
|
||||
validateCommand(command.key, command.table, cfName);
|
||||
validateKeyCommand(command.key, command.table, cfName);
|
||||
|
||||
Row row;
|
||||
try
|
||||
|
|
@ -321,7 +312,7 @@ public class CassandraServer implements Cassandra.Iface
|
|||
RowMutation rm = new RowMutation(tablename, key.trim());
|
||||
rm.add(columnFamily_column, cellData, timestamp);
|
||||
Set<String> cfNames = rm.columnFamilyNames();
|
||||
validateCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
|
||||
doInsert(block, rm);
|
||||
}
|
||||
|
|
@ -331,7 +322,7 @@ public class CassandraServer implements Cassandra.Iface
|
|||
logger.debug("batch_insert");
|
||||
RowMutation rm = RowMutation.getRowMutation(batchMutation);
|
||||
Set<String> cfNames = rm.columnFamilyNames();
|
||||
validateCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
|
||||
doInsert(block, rm);
|
||||
}
|
||||
|
|
@ -343,7 +334,7 @@ public class CassandraServer implements Cassandra.Iface
|
|||
RowMutation rm = new RowMutation(tablename, key.trim());
|
||||
rm.delete(columnFamily_column, timestamp);
|
||||
Set<String> cfNames = rm.columnFamilyNames();
|
||||
validateCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
doInsert(block, rm);
|
||||
}
|
||||
|
||||
|
|
@ -433,7 +424,7 @@ public class CassandraServer implements Cassandra.Iface
|
|||
logger.debug("batch_insert_SuperColumn");
|
||||
RowMutation rm = RowMutation.getRowMutation(batchMutationSuper);
|
||||
Set<String> cfNames = rm.columnFamilyNames();
|
||||
validateCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
validateKeyCommand(rm.key(), rm.table(), cfNames.toArray(new String[cfNames.size()]));
|
||||
doInsert(block, rm);
|
||||
}
|
||||
|
||||
|
|
@ -522,15 +513,20 @@ public class CassandraServer implements Cassandra.Iface
|
|||
return result;
|
||||
}
|
||||
|
||||
public List<String> get_key_range(String tablename, String startWith, String stopAt, int maxResults) throws InvalidRequestException
|
||||
public List<String> get_key_range(String tablename, List<String> columnFamilies, String startWith, String stopAt, int maxResults) throws InvalidRequestException
|
||||
{
|
||||
logger.debug("get_key_range");
|
||||
validateCommand(tablename, columnFamilies.toArray(new String[columnFamilies.size()]));
|
||||
if (!(StorageService.getPartitioner() instanceof OrderPreservingPartitioner))
|
||||
{
|
||||
throw new InvalidRequestException("range queries may only be performed against an order-preserving partitioner");
|
||||
}
|
||||
if (maxResults <= 0)
|
||||
{
|
||||
throw new InvalidRequestException("maxResults must be positive");
|
||||
}
|
||||
|
||||
return StorageProxy.getKeyRange(new RangeCommand(tablename, startWith, stopAt, maxResults));
|
||||
return StorageProxy.getKeyRange(new RangeCommand(tablename, columnFamilies, startWith, stopAt, maxResults));
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ public class RangeVerbHandler implements IVerbHandler
|
|||
{
|
||||
RangeCommand command = RangeCommand.read(message);
|
||||
Table table = Table.open(command.table);
|
||||
keys = table.getKeyRange(command.startWith, command.stopAt, command.maxResults);
|
||||
keys = table.getKeyRange(command.columnFamilyNames, command.startWith, command.stopAt, command.maxResults);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
|
|
|
|||
|
|
@ -1,3 +1,6 @@
|
|||
# to run a single test, run from trunk/:
|
||||
# PYTHONPATH=test nosetests --tests=system.test_server:TestMutations.test_empty_range
|
||||
|
||||
import os, sys, time
|
||||
|
||||
from . import client, root, CassandraTester
|
||||
|
|
@ -91,12 +94,13 @@ class TestMutations(CassandraTester):
|
|||
_insert_batch(True)
|
||||
_verify_batch()
|
||||
|
||||
def test_bad_gets(self):
|
||||
def test_bad_calls(self):
|
||||
_expect_exception(lambda: client.get_column('Table1', 'key1', 'Standard1'), InvalidRequestException)
|
||||
_expect_exception(lambda: client.get_column('Table1', 'key1', 'Standard1:x:y'), InvalidRequestException)
|
||||
_expect_exception(lambda: client.get_column('Table1', 'key1', 'Super1'), InvalidRequestException)
|
||||
_expect_exception(lambda: client.get_column('Table1', 'key1', 'Super1:x'), InvalidRequestException)
|
||||
_expect_exception(lambda: client.get_column('Table1', 'key1', 'Super1:x:y:z'), InvalidRequestException)
|
||||
_expect_exception(lambda: client.get_key_range('Table1', ['S'], '', '', 1000), InvalidRequestException)
|
||||
|
||||
def test_batch_insert_super(self):
|
||||
cfmap = {'Super1': _SUPER_COLUMNS,
|
||||
|
|
@ -234,20 +238,23 @@ class TestMutations(CassandraTester):
|
|||
|
||||
|
||||
def test_empty_range(self):
|
||||
assert client.get_key_range('Table1', '', '', 1000) == []
|
||||
assert client.get_key_range('Table1', [], '', '', 1000) == []
|
||||
assert client.get_key_range('Table1', ['Standard1'], '', '', 1000) == []
|
||||
_insert_simple()
|
||||
assert client.get_key_range('Table1', ['Super1'], '', '', 1000) == []
|
||||
|
||||
def test_range_with_remove(self):
|
||||
_insert_simple()
|
||||
assert client.get_key_range('Table1', 'key1', '', 1000) == ['key1']
|
||||
assert client.get_key_range('Table1', [], 'key1', '', 1000) == ['key1']
|
||||
|
||||
client.remove('Table1', 'key1', 'Standard1:c1', 1, True)
|
||||
client.remove('Table1', 'key1', 'Standard1:c2', 1, True)
|
||||
assert client.get_key_range('Table1', '', '', 1000) == []
|
||||
assert client.get_key_range('Table1', [], '', '', 1000) == []
|
||||
|
||||
def test_range_collation(self):
|
||||
for key in ['-a', '-b', 'a', 'b'] + [str(i) for i in xrange(100)]:
|
||||
client.insert('Table1', key, 'Standard1:' + key, 'v', 0, True)
|
||||
L = client.get_key_range('Table1', '', '', 1000)
|
||||
L = client.get_key_range('Table1', ['Standard1'], '', '', 1000)
|
||||
# note the collated ordering rather than ascii
|
||||
assert L == ['0', '1', '10', '11', '12', '13', '14', '15', '16', '17', '18', '19', '2', '20', '21', '22', '23', '24', '25', '26', '27','28', '29', '3', '30', '31', '32', '33', '34', '35', '36', '37', '38', '39', '4', '40', '41', '42', '43', '44', '45', '46', '47', '48', '49', '5', '50', '51', '52', '53', '54', '55', '56', '57', '58', '59', '6', '60', '61', '62', '63', '64', '65', '66', '67', '68', '69', '7', '70', '71', '72', '73', '74', '75', '76', '77', '78', '79', '8', '80', '81', '82', '83', '84', '85', '86', '87', '88', '89', '9', '90', '91', '92', '93', '94', '95', '96', '97', '98', '99', 'a', '-a', 'b', '-b'], L
|
||||
|
||||
|
|
@ -255,14 +262,14 @@ class TestMutations(CassandraTester):
|
|||
for key in ['-a', '-b', 'a', 'b'] + [str(i) for i in xrange(100)]:
|
||||
client.insert('Table1', key, 'Standard1:' + key, 'v', 0, True)
|
||||
|
||||
L = client.get_key_range('Table1', 'a', '', 1000)
|
||||
L = client.get_key_range('Table1', [], 'a', '', 1000)
|
||||
assert L == ['a', '-a', 'b', '-b'], L
|
||||
|
||||
L = client.get_key_range('Table1', '', '15', 1000)
|
||||
L = client.get_key_range('Table1', [], '', '15', 1000)
|
||||
assert L == ['0', '1', '10', '11', '12', '13', '14', '15'], L
|
||||
|
||||
L = client.get_key_range('Table1', '50', '51', 1000)
|
||||
L = client.get_key_range('Table1', [], '50', '51', 1000)
|
||||
assert L == ['50', '51'], L
|
||||
|
||||
L = client.get_key_range('Table1', '1', '', 10)
|
||||
L = client.get_key_range('Table1', [], '1', '', 10)
|
||||
assert L == ['1', '10', '11', '12', '13', '14', '15', '16', '17', '18'], L
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import java.util.concurrent.ExecutionException;
|
|||
import java.util.concurrent.Future;
|
||||
import java.util.Set;
|
||||
import java.util.HashSet;
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
|
|
@ -49,7 +50,7 @@ public class CompactionsTest
|
|||
inserted.add(key);
|
||||
}
|
||||
store.forceBlockingFlush();
|
||||
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
|
||||
assertEquals(table.getKeyRange(Arrays.asList("Standard1"), "", "", 10000).size(), inserted.size());
|
||||
}
|
||||
while (true)
|
||||
{
|
||||
|
|
@ -61,6 +62,6 @@ public class CompactionsTest
|
|||
{
|
||||
store.doCompaction(store.getSSTableFilenames().size());
|
||||
}
|
||||
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
|
||||
assertEquals(table.getKeyRange(Arrays.asList("Standard1"), "", "", 10000).size(), inserted.size());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import java.util.concurrent.ExecutionException;
|
|||
import java.util.concurrent.Future;
|
||||
import java.util.Set;
|
||||
import java.util.HashSet;
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.Test;
|
||||
|
||||
|
|
@ -43,12 +44,12 @@ public class OneCompactionTest
|
|||
rm.apply();
|
||||
inserted.add(key);
|
||||
store.forceBlockingFlush();
|
||||
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
|
||||
assertEquals(table.getKeyRange(Arrays.asList(columnFamilyName), "", "", 10000).size(), inserted.size());
|
||||
}
|
||||
Future<Integer> ft = MinorCompactionManager.instance().submit(store, 2);
|
||||
ft.get();
|
||||
assertEquals(1, store.getSSTableFilenames().size());
|
||||
assertEquals(table.getKeyRange("", "", 10000).size(), inserted.size());
|
||||
assertEquals(table.getKeyRange(Arrays.asList(columnFamilyName), "", "", 10000).size(), inserted.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
Loading…
Reference in New Issue