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

Merge branch cassandra-2.1 into trunk

This commit is contained in:
Pavel Yaskevich 2014-03-13 14:21:49 -07:00
parent c0aa7467e5 4c22b165cb
commit 67feb71eef
52 changed files with 506 additions and 441 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
CHANGES.txt
View File

@ -20,7 +20,17 @@
* Fix ClassCastException for compact table with composites (CASSANDRA-6738)
* Fix potentially repairing with wrong nodes (CASSANDRA-6808)
* Change caching option syntax (CASSANDRA-6745)
* Fix stress to do proper counter reads (CASSANDRA-6835)
Merged from 2.0:
* Fix leaking validator FH in StreamWriter (CASSANDRA-6832)
* fix nodetool getsstables for blob PK (CASSANDRA-6803)
* Fix saving triggers to schema (CASSANDRA-6789)
* Fix trigger mutations when base mutation list is immutable (CASSANDRA-6790)
* Fix accounting in FileCacheService to allow re-using RAR (CASSANDRA-6838)
* Fix static counter columns (CASSANDRA-6827)
* Restore expiring->deleted (cell) compaction optimization (CASSANDRA-6844)
* Fix CompactionManager.needsCleanup (CASSANDRA-6845)
* Correctly compare BooleanType values other than 0 and 1 (CASSANDRA-6779)
* Avoid race-prone second "scrub" of system keyspace (CASSANDRA-6797)
* Pool CqlRecordWriter clients by inetaddress rather than Range
(CASSANDRA-6665)
@ -1258,6 +1268,7 @@ Merged from 1.0:
1.1.1
* add populate_io_cache_on_flush option (CASSANDRA-2635)
* allow larger cache capacities than 2GB (CASSANDRA-4150)
* add getsstables command to nodetool (CASSANDRA-4199)
* apply parent CF compaction settings to secondary index CFs (CASSANDRA-4280)

BIN
lib/cassandra-driver-internal-only-1.0.2.post.zip
View File

Binary file not shown.

6
pylib/cqlshlib/cql3handling.py
View File

@ -419,6 +419,8 @@ def cf_prop_val_completer(ctxt, cass):
return ["{'sstable_compression': '"]
if this_opt == 'compaction':
return ["{'class': '"]
if this_opt == 'caching':
return ["{'keys': '"]
if any(this_opt == opt[0] for opt in CqlRuleSet.obsolete_cf_options):
return ["'<obsolete_option>'"]
if this_opt in ('read_repair_chance', 'bloom_filter_fp_chance',
@ -472,9 +474,9 @@ def cf_prop_val_mapval_completer(ctxt, cass):
return [Hint('<option_value>')]
elif opt == 'caching':
if key == 'rows_per_partition':
return [Hint('ALL'), Hint('NONE'), Hint('#rows_per_partition')]
return ["'ALL'", "'NONE'", Hint('#rows_per_partition')]
elif key == 'keys':
return [Hint('ALL'), Hint('NONE')]
return ["'ALL'", "'NONE'"]
return ()
def cf_prop_val_mapender_completer(ctxt, cass):

2
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -1809,7 +1809,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
public List<String> getSSTablesForKey(String key)
{
DecoratedKey dk = new DecoratedKey(partitioner.getToken(ByteBuffer.wrap(key.getBytes())), ByteBuffer.wrap(key.getBytes()));
DecoratedKey dk = partitioner.decorateKey(metadata.getKeyValidator().fromString(key));
ViewFragment view = markReferenced(dk);
try
{

5
src/java/org/apache/cassandra/db/composites/AbstractComposite.java
View File

@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import org.apache.cassandra.db.filter.ColumnSlice;
import org.apache.cassandra.db.marshal.AbstractCompositeType;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.utils.ByteBufferUtil;
public abstract class AbstractComposite implements Composite
{
@ -75,12 +76,12 @@ public abstract class AbstractComposite implements Composite
// See org.apache.cassandra.db.marshal.CompositeType for details.
ByteBuffer result = ByteBuffer.allocate(dataSize() + 3 * size() + (isStatic() ? 2 : 0));
if (isStatic())
AbstractCompositeType.putShortLength(result, CompositeType.STATIC_MARKER);
ByteBufferUtil.writeShortLength(result, CompositeType.STATIC_MARKER);
for (int i = 0; i < size(); i++)
{
ByteBuffer bb = get(i);
AbstractCompositeType.putShortLength(result, bb.remaining());
ByteBufferUtil.writeShortLength(result, bb.remaining());
result.put(bb.duplicate());
result.put((byte)0);
}

63
src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java
View File

@ -17,15 +17,16 @@
*/
package org.apache.cassandra.db.marshal;
import org.apache.cassandra.serializers.TypeSerializer;
import org.apache.cassandra.serializers.BytesSerializer;
import org.apache.cassandra.serializers.MarshalException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import org.apache.cassandra.serializers.TypeSerializer;
import org.apache.cassandra.serializers.BytesSerializer;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.utils.ByteBufferUtil;
/**
* A class avoiding class duplication between CompositeType and
* DynamicCompositeType.
@ -34,44 +35,6 @@ import java.util.List;
*/
public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
{
// changes bb position
public static int getShortLength(ByteBuffer bb)
{
int length = (bb.get() & 0xFF) << 8;
return length | (bb.get() & 0xFF);
}
// Doesn't change bb position
protected static int getShortLength(ByteBuffer bb, int position)
{
int length = (bb.get(position) & 0xFF) << 8;
return length | (bb.get(position + 1) & 0xFF);
}
// changes bb position
public static void putShortLength(ByteBuffer bb, int length)
{
bb.put((byte) ((length >> 8) & 0xFF));
bb.put((byte) (length & 0xFF));
}
// changes bb position
public static ByteBuffer getBytes(ByteBuffer bb, int length)
{
ByteBuffer copy = bb.duplicate();
copy.limit(copy.position() + length);
bb.position(bb.position() + length);
return copy;
}
// changes bb position
public static ByteBuffer getWithShortLength(ByteBuffer bb)
{
int length = getShortLength(bb);
return getBytes(bb, length);
}
public int compare(ByteBuffer o1, ByteBuffer o2)
{
if (o1 == null || !o1.hasRemaining())
@ -95,8 +58,8 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
{
AbstractType<?> comparator = getComparator(i, bb1, bb2);
ByteBuffer value1 = getWithShortLength(bb1);
ByteBuffer value2 = getWithShortLength(bb2);
ByteBuffer value1 = ByteBufferUtil.readBytesWithShortLength(bb1);
ByteBuffer value2 = ByteBufferUtil.readBytesWithShortLength(bb2);
int cmp = comparator.compareCollectionMembers(value1, value2, previous);
if (cmp != 0)
@ -135,7 +98,7 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
while (bb.remaining() > 0)
{
getComparator(i++, bb);
l.add(getWithShortLength(bb));
l.add(ByteBufferUtil.readBytesWithShortLength(bb));
bb.get(); // skip end-of-component
}
return l.toArray(new ByteBuffer[l.size()]);
@ -164,7 +127,7 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
while (bb.remaining() > 0)
{
AbstractType comparator = getComparator(i, bb);
ByteBuffer value = getWithShortLength(bb);
ByteBuffer value = ByteBufferUtil.readBytesWithShortLength(bb);
list.add( new CompositeComponent(comparator,value) );
@ -237,7 +200,7 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
sb.append(":");
AbstractType<?> comparator = getAndAppendComparator(i, bb, sb);
ByteBuffer value = getWithShortLength(bb);
ByteBuffer value = ByteBufferUtil.readBytesWithShortLength(bb);
sb.append(escape(comparator.getString(value)));
@ -290,7 +253,7 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
for (ByteBuffer component : components)
{
comparators.get(i).serializeComparator(bb);
putShortLength(bb, component.remaining());
ByteBufferUtil.writeShortLength(bb, component.remaining());
bb.put(component); // it's ok to consume component as we won't use it anymore
bb.put((byte)0);
++i;
@ -318,11 +281,11 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
if (bb.remaining() < 2)
throw new MarshalException("Not enough bytes to read value size of component " + i);
int length = getShortLength(bb);
int length = ByteBufferUtil.readShortLength(bb);
if (bb.remaining() < length)
throw new MarshalException("Not enough bytes to read value of component " + i);
ByteBuffer value = getBytes(bb, length);
ByteBuffer value = ByteBufferUtil.readBytes(bb, length);
comparator.validateCollectionMember(value, previous);

6
src/java/org/apache/cassandra/db/marshal/CollectionType.java
View File

@ -146,12 +146,6 @@ public abstract class CollectionType<T> extends AbstractType<T>
return pack(buffers, elements, size);
}
protected static int getUnsignedShort(ByteBuffer bb)
{
int length = (bb.get() & 0xFF) << 8;
return length | (bb.get() & 0xFF);
}
public CQL3Type asCQL3Type()
{
return new CQL3Type.Collection(this);

12
src/java/org/apache/cassandra/db/marshal/CompositeType.java
View File

@ -89,11 +89,11 @@ public class CompositeType extends AbstractCompositeType
if (bb.remaining() < 2)
return false;
int header = getShortLength(bb, bb.position());
int header = ByteBufferUtil.getShortLength(bb, bb.position());
if ((header & 0xFFFF) != STATIC_MARKER)
return false;
getShortLength(bb); // Skip header
ByteBufferUtil.readShortLength(bb); // Skip header
return true;
}
@ -179,7 +179,7 @@ public class CompositeType extends AbstractCompositeType
int i = 0;
while (bb.remaining() > 0)
{
l[i++] = getWithShortLength(bb);
l[i++] = ByteBufferUtil.readBytesWithShortLength(bb);
bb.get(); // skip end-of-component
}
return i == l.length ? l : Arrays.copyOfRange(l, 0, i);
@ -193,7 +193,7 @@ public class CompositeType extends AbstractCompositeType
int i = 0;
while (bb.remaining() > 0)
{
ByteBuffer c = getWithShortLength(bb);
ByteBuffer c = ByteBufferUtil.readBytesWithShortLength(bb);
if (i == idx)
return c;
@ -212,7 +212,7 @@ public class CompositeType extends AbstractCompositeType
public static boolean isStaticName(ByteBuffer bb)
{
return bb.remaining() >= 2 && (getShortLength(bb, bb.position()) & 0xFFFF) == STATIC_MARKER;
return bb.remaining() >= 2 && (ByteBufferUtil.getShortLength(bb, bb.position()) & 0xFFFF) == STATIC_MARKER;
}
@Override
@ -324,7 +324,7 @@ public class CompositeType extends AbstractCompositeType
ByteBuffer out = ByteBuffer.allocate(totalLength);
for (ByteBuffer bb : buffers)
{
putShortLength(out, bb.remaining());
ByteBufferUtil.writeShortLength(out, bb.remaining());
out.put(bb.duplicate());
out.put((byte) 0);
}

14
src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java
View File

@ -90,10 +90,10 @@ public class DynamicCompositeType extends AbstractCompositeType
{
try
{
int header = getShortLength(bb);
int header = ByteBufferUtil.readShortLength(bb);
if ((header & 0x8000) == 0)
{
String name = ByteBufferUtil.string(getBytes(bb, header));
String name = ByteBufferUtil.string(ByteBufferUtil.readBytes(bb, header));
return TypeParser.parse(name);
}
else
@ -152,10 +152,10 @@ public class DynamicCompositeType extends AbstractCompositeType
{
try
{
int header = getShortLength(bb);
int header = ByteBufferUtil.readShortLength(bb);
if ((header & 0x8000) == 0)
{
String name = ByteBufferUtil.string(getBytes(bb, header));
String name = ByteBufferUtil.string(ByteBufferUtil.readBytes(bb, header));
sb.append(name).append("@");
return TypeParser.parse(name);
}
@ -189,13 +189,13 @@ public class DynamicCompositeType extends AbstractCompositeType
AbstractType<?> comparator = null;
if (bb.remaining() < 2)
throw new MarshalException("Not enough bytes to header of the comparator part of component " + i);
int header = getShortLength(bb);
int header = ByteBufferUtil.readShortLength(bb);
if ((header & 0x8000) == 0)
{
if (bb.remaining() < header)
throw new MarshalException("Not enough bytes to read comparator name of component " + i);
ByteBuffer value = getBytes(bb, header);
ByteBuffer value = ByteBufferUtil.readBytes(bb, header);
String valueStr = null;
try
{
@ -325,7 +325,7 @@ public class DynamicCompositeType extends AbstractCompositeType
header = 0x8000 | (((byte)comparatorName.charAt(0)) & 0xFF);
else
header = comparatorName.length();
putShortLength(bb, header);
ByteBufferUtil.writeShortLength(bb, header);
if (!isAlias)
bb.put(ByteBufferUtil.bytes(comparatorName));

6
src/java/org/apache/cassandra/serializers/CollectionSerializer.java
View File

@ -48,10 +48,4 @@ public abstract class CollectionSerializer<T> implements TypeSerializer<T>
size += 2 + bb.remaining();
return pack(buffers, elements, size);
}
protected static int getUnsignedShort(ByteBuffer bb)
{
int length = (bb.get() & 0xFF) << 8;
return length | (bb.get() & 0xFF);
}
}

9
src/java/org/apache/cassandra/serializers/ListSerializer.java
View File

@ -22,6 +22,8 @@ import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.utils.ByteBufferUtil;
public class ListSerializer<T> extends CollectionSerializer<List<T>>
{
// interning instances
@ -50,14 +52,11 @@ public class ListSerializer<T> extends CollectionSerializer<List<T>>
try
{
ByteBuffer input = bytes.duplicate();
int n = getUnsignedShort(input);
int n = ByteBufferUtil.readShortLength(input);
List<T> l = new ArrayList<T>(n);
for (int i = 0; i < n; i++)
{
int s = getUnsignedShort(input);
byte[] data = new byte[s];
input.get(data);
ByteBuffer databb = ByteBuffer.wrap(data);
ByteBuffer databb = ByteBufferUtil.readBytesWithShortLength(input);
elements.validate(databb);
l.add(elements.deserialize(databb));
}

17
src/java/org/apache/cassandra/serializers/MapSerializer.java
View File

@ -18,12 +18,13 @@
package org.apache.cassandra.serializers;
import org.apache.cassandra.utils.Pair;
import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.Pair;
public class MapSerializer<K, V> extends CollectionSerializer<Map<K, V>>
{
// interning instances
@ -55,20 +56,14 @@ public class MapSerializer<K, V> extends CollectionSerializer<Map<K, V>>
try
{
ByteBuffer input = bytes.duplicate();
int n = getUnsignedShort(input);
int n = ByteBufferUtil.readShortLength(input);
Map<K, V> m = new LinkedHashMap<K, V>(n);
for (int i = 0; i < n; i++)
{
int sk = getUnsignedShort(input);
byte[] datak = new byte[sk];
input.get(datak);
ByteBuffer kbb = ByteBuffer.wrap(datak);
ByteBuffer kbb = ByteBufferUtil.readBytesWithShortLength(input);
keys.validate(kbb);
int sv = getUnsignedShort(input);
byte[] datav = new byte[sv];
input.get(datav);
ByteBuffer vbb = ByteBuffer.wrap(datav);
ByteBuffer vbb = ByteBufferUtil.readBytesWithShortLength(input);
values.validate(vbb);
m.put(keys.deserialize(kbb), values.deserialize(vbb));

9
src/java/org/apache/cassandra/serializers/SetSerializer.java
View File

@ -22,6 +22,8 @@ import java.nio.BufferUnderflowException;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.utils.ByteBufferUtil;
public class SetSerializer<T> extends CollectionSerializer<Set<T>>
{
// interning instances
@ -50,14 +52,11 @@ public class SetSerializer<T> extends CollectionSerializer<Set<T>>
try
{
ByteBuffer input = bytes.duplicate();
int n = getUnsignedShort(input);
int n = ByteBufferUtil.readShortLength(input);
Set<T> l = new LinkedHashSet<T>(n);
for (int i = 0; i < n; i++)
{
int s = getUnsignedShort(input);
byte[] data = new byte[s];
input.get(data);
ByteBuffer databb = ByteBuffer.wrap(data);
ByteBuffer databb = ByteBufferUtil.readBytesWithShortLength(input);
elements.validate(databb);
l.add(elements.deserialize(databb));
}

8
src/java/org/apache/cassandra/streaming/StreamWriter.java
View File

@ -72,10 +72,9 @@ public class StreamWriter
{
long totalSize = totalSize();
RandomAccessReader file = sstable.openDataReader();
ChecksumValidator validator = null;
if (new File(sstable.descriptor.filenameFor(Component.CRC)).exists())
validator = DataIntegrityMetadata.checksumValidator(sstable.descriptor);
ChecksumValidator validator = new File(sstable.descriptor.filenameFor(Component.CRC)).exists()
? DataIntegrityMetadata.checksumValidator(sstable.descriptor)
: null;
transferBuffer = validator == null ? new byte[DEFAULT_CHUNK_SIZE] : new byte[validator.chunkSize];
// setting up data compression stream
@ -115,6 +114,7 @@ public class StreamWriter
{
// no matter what happens close file
FileUtils.closeQuietly(file);
FileUtils.closeQuietly(validator);
}
// release reference only when completed successfully

37
src/java/org/apache/cassandra/utils/ByteBufferUtil.java
View File

@ -558,4 +558,41 @@ public class ByteBufferUtil
{
return buf.capacity() > buf.remaining() ? ByteBuffer.wrap(getArray(buf)) : buf;
}
// Doesn't change bb position
public static int getShortLength(ByteBuffer bb, int position)
{
int length = (bb.get(position) & 0xFF) << 8;
return length | (bb.get(position + 1) & 0xFF);
}
// changes bb position
public static int readShortLength(ByteBuffer bb)
{
int length = (bb.get() & 0xFF) << 8;
return length | (bb.get() & 0xFF);
}
// changes bb position
public static void writeShortLength(ByteBuffer bb, int length)
{
bb.put((byte) ((length >> 8) & 0xFF));
bb.put((byte) (length & 0xFF));
}
// changes bb position
public static ByteBuffer readBytes(ByteBuffer bb, int length)
{
ByteBuffer copy = bb.duplicate();
copy.limit(copy.position() + length);
bb.position(bb.position() + length);
return copy;
}
// changes bb position
public static ByteBuffer readBytesWithShortLength(ByteBuffer bb)
{
int length = readShortLength(bb);
return readBytes(bb, length);
}
}

2
src/java/org/apache/cassandra/utils/FBUtilities.java
View File

@ -370,7 +370,7 @@ public class FBUtilities
in = FBUtilities.class.getClassLoader().getResourceAsStream("org/apache/cassandra/config/version.properties");
if (in == null)
{
return "Unknown";
return System.getProperty("cassandra.releaseVersion", "Unknown");
}
Properties props = new Properties();
props.load(in);

109
tools/stress/src/org/apache/cassandra/stress/Operation.java
View File

@ -19,27 +19,16 @@ package org.apache.cassandra.stress;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumMap;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.cassandra.stress.generatedata.Distribution;
import org.apache.cassandra.stress.generatedata.KeyGen;
import org.apache.cassandra.stress.generatedata.RowGen;
import org.apache.cassandra.stress.operations.CqlCounterAdder;
import org.apache.cassandra.stress.operations.CqlCounterGetter;
import org.apache.cassandra.stress.operations.CqlIndexedRangeSlicer;
import org.apache.cassandra.stress.operations.CqlInserter;
import org.apache.cassandra.stress.operations.CqlMultiGetter;
import org.apache.cassandra.stress.operations.CqlRangeSlicer;
import org.apache.cassandra.stress.operations.CqlReader;
import org.apache.cassandra.stress.operations.ThriftCounterAdder;
import org.apache.cassandra.stress.operations.ThriftCounterGetter;
import org.apache.cassandra.stress.operations.ThriftIndexedRangeSlicer;
import org.apache.cassandra.stress.operations.ThriftInserter;
import org.apache.cassandra.stress.operations.ThriftMultiGetter;
import org.apache.cassandra.stress.operations.ThriftRangeSlicer;
import org.apache.cassandra.stress.operations.ThriftReader;
import org.apache.cassandra.stress.settings.Command;
import org.apache.cassandra.stress.settings.CqlVersion;
import org.apache.cassandra.stress.settings.SettingsCommandMixed;
@ -49,6 +38,8 @@ import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -79,6 +70,7 @@ public abstract class Operation
public final Command type;
public final KeyGen keyGen;
public final RowGen rowGen;
public final Distribution counteradd;
public final List<ColumnParent> columnParents;
public final StressMetrics metrics;
public final SettingsCommandMixed.CommandSelector commandSelector;
@ -99,19 +91,12 @@ public abstract class Operation
commandSelector = null;
substates = null;
}
counteradd = settings.command.add.get();
this.settings = settings;
this.keyGen = settings.keys.newKeyGen();
this.rowGen = settings.columns.newRowGen();
this.metrics = metrics;
if (!settings.columns.useSuperColumns)
columnParents = Collections.singletonList(new ColumnParent(settings.schema.columnFamily));
else
{
ColumnParent[] cp = new ColumnParent[settings.columns.superColumns];
for (int i = 0 ; i < cp.length ; i++)
cp[i] = new ColumnParent("Super1").setSuper_column(ByteBufferUtil.bytes("S" + i));
columnParents = Arrays.asList(cp);
}
this.columnParents = columnParents(type, settings);
}
private State(Command type, State copy)
@ -120,13 +105,29 @@ public abstract class Operation
this.timer = copy.timer;
this.rowGen = copy.rowGen;
this.keyGen = copy.keyGen;
this.columnParents = copy.columnParents;
this.columnParents = columnParents(type, copy.settings);
this.metrics = copy.metrics;
this.settings = copy.settings;
this.counteradd = copy.counteradd;
this.substates = null;
this.commandSelector = null;
}
private List<ColumnParent> columnParents(Command type, StressSettings settings)
{
if (!settings.columns.useSuperColumns)
return Collections.singletonList(new ColumnParent(type.table));
else
{
ColumnParent[] cp = new ColumnParent[settings.columns.superColumns];
for (int i = 0 ; i < cp.length ; i++)
cp[i] = new ColumnParent(type.supertable).setSuper_column(ByteBufferUtil.bytes("S" + i));
return Arrays.asList(cp);
}
}
public boolean isCql3()
{
return settings.mode.cqlVersion == CqlVersion.CQL3;
@ -168,6 +169,53 @@ public abstract class Operation
return state.rowGen.generate(index, key);
}
private int sliceStart(int count)
{
if (count == state.settings.columns.maxColumnsPerKey)
return 0;
return 1 + ThreadLocalRandom.current().nextInt(state.settings.columns.maxColumnsPerKey - count);
}
protected SlicePredicate slicePredicate()
{
final SlicePredicate predicate = new SlicePredicate();
if (state.settings.columns.slice)
{
int count = state.rowGen.count(index);
int start = sliceStart(count);
predicate.setSlice_range(new SliceRange()
.setStart(state.settings.columns.names.get(start))
.setFinish(new byte[] {})
.setReversed(false)
.setCount(count)
);
}
else
predicate.setColumn_names(randomNames());
return predicate;
}
protected List<ByteBuffer> randomNames()
{
int count = state.rowGen.count(index);
List<ByteBuffer> src = state.settings.columns.names;
if (count == src.size())
return src;
ThreadLocalRandom rnd = ThreadLocalRandom.current();
List<ByteBuffer> r = new ArrayList<>();
int c = 0, o = 0;
while (c < count && count + o < src.size())
{
int leeway = src.size() - (count + o);
int spreadover = count - c;
o += Math.round(rnd.nextDouble() * (leeway / (double) spreadover));
r.add(src.get(o + c++));
}
while (c < count)
r.add(src.get(o + c++));
return r;
}
/**
* Run operation
* @param client Cassandra Thrift client connection
@ -209,10 +257,11 @@ public abstract class Operation
if (!success)
{
error(String.format("Operation [%d] x%d key %s %s%n",
error(String.format("Operation [%d] x%d key %s (0x%s) %s%n",
index,
tries,
run.key(),
ByteBufferUtil.bytesToHex(ByteBufferUtil.bytes(run.key())),
(exceptionMessage == null)
? "Data returned was not validated"
: "Error executing: " + exceptionMessage));
@ -235,14 +284,4 @@ public abstract class Operation
System.err.println(message);
}
public static ByteBuffer getColumnNameBytes(int i)
{
return ByteBufferUtil.bytes("C" + i);
}
public static String getColumnName(int i)
{
return "C" + i;
}
}

4
tools/stress/src/org/apache/cassandra/stress/StressAction.java
View File

@ -87,7 +87,7 @@ public class StressAction implements Runnable
warmup(subtype, command);
return;
case MULTI:
int keysAtOnce = ((SettingsCommandMulti) command).keysAtOnce;
int keysAtOnce = command.keysAtOnce;
iterations = Math.min(50000, (int) Math.ceil(500000d / keysAtOnce));
break;
default:
@ -298,6 +298,8 @@ public class StressAction implements Runnable
case SIMPLE_NATIVE:
op.run(sclient);
break;
case THRIFT:
case THRIFT_SMART:
default:
op.run(tclient);
}

1
tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java
View File

@ -46,6 +46,7 @@ public abstract class RowGen
// these byte[] may be re-used
abstract List<ByteBuffer> getColumns(long operationIndex);
abstract public int count(long operationIndex);
abstract public boolean isDeterministic();

7
tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java
View File

@ -51,6 +51,8 @@ public class RowGenDistributedSize extends RowGen
this.sizeDistribution = sizeDistribution;
ret = new ByteBuffer[(int) countDistribution.maxValue()];
sizes = new int[ret.length];
// TODO: should keep it deterministic in event that count distribution is not, but size and dataGen are, so that
// we simply need to generate the correct selection of columns
this.isDeterministic = dataGen.isDeterministic() && countDistribution.maxValue() == countDistribution.minValue()
&& sizeDistribution.minValue() == sizeDistribution.maxValue();
}
@ -100,6 +102,11 @@ public class RowGenDistributedSize extends RowGen
return Arrays.asList(ret).subList(0, count);
}
public int count(long operationIndex)
{
return (int) countDistribution.next();
}
@Override
public boolean isDeterministic()
{

21
tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java
View File

@ -22,7 +22,7 @@ package org.apache.cassandra.stress.operations;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.ArrayList;
import java.util.List;
public class CqlCounterAdder extends CqlOperation<Integer>
@ -35,11 +35,7 @@ public class CqlCounterAdder extends CqlOperation<Integer>
@Override
protected String buildQuery()
{
String counterCF = "Counter3";
StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(counterCF));
query.append(" SET ");
StringBuilder query = new StringBuilder("UPDATE \"Counter3\" SET ");
// TODO : increment distribution subset of columns
for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++)
@ -47,20 +43,25 @@ public class CqlCounterAdder extends CqlOperation<Integer>
if (i > 0)
query.append(",");
query.append('C').append(i).append("=C").append(i).append("+1");
String name = state.settings.columns.namestrs.get(i);
query.append(name).append("=").append(name).append("+?");
}
query.append(" WHERE KEY=?");
return query.toString();
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
return Collections.singletonList(ByteBuffer.wrap(key));
final List<Object> list = new ArrayList<>();
for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++)
list.add(state.counteradd.next());
list.add(ByteBuffer.wrap(key));
return list;
}
@Override
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
return new CqlRunOpAlwaysSucceed(client, query, queryId, params, keyid, key, 1);
}

13
tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java
View File

@ -34,19 +34,24 @@ public class CqlCounterGetter extends CqlOperation<Integer>
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
return Collections.singletonList(ByteBuffer.wrap(key));
return Collections.<Object>singletonList(ByteBuffer.wrap(key));
}
@Override
protected String buildQuery()
{
return "SELECT * FROM \"Counter3\" USING CONSISTENCY " + state.settings.command.consistencyLevel + " WHERE KEY=?";
StringBuilder query = new StringBuilder("SELECT *");
// TODO: obey slice/noslice option (instead of always slicing)
query.append(" FROM ").append(wrapInQuotes(state.type.table));
return query.append(" WHERE KEY=?").toString();
}
@Override
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
return new CqlRunOpTestNonEmpty(client, query, queryId, params, keyid, key);
}

18
tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java
View File

@ -26,7 +26,6 @@ import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.List;
import org.apache.cassandra.stress.settings.SettingsCommandMulti;
import org.apache.cassandra.utils.FBUtilities;
public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
@ -40,7 +39,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
throw new UnsupportedOperationException();
}
@ -48,10 +47,11 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
@Override
protected String buildQuery()
{
StringBuilder query = new StringBuilder("SELECT * FROM \"Standard1\"");
final String columnName = getColumnName(1);
query.append(" WHERE ").append(columnName).append("=?")
.append(" AND KEY > ? LIMIT ").append(((SettingsCommandMulti)state.settings.command).keysAtOnce);
final String indexColumn = (state.settings.columns.namestrs.get(1));
StringBuilder query = new StringBuilder("SELECT * FROM ");
query.append(wrapInQuotes(state.type.table));
query.append(" WHERE ").append(indexColumn).append("=?")
.append(" AND KEY > ? LIMIT ").append(state.settings.command.keysAtOnce);
return query.toString();
}
@ -65,7 +65,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
int rowCount;
do
{
List<ByteBuffer> params = Arrays.asList(value, ByteBuffer.wrap(minKey));
List<Object> params = Arrays.<Object>asList(value, ByteBuffer.wrap(minKey));
CqlRunOp<byte[][]> op = run(client, params, value, new String(value.array()));
byte[][] keys = op.result;
rowCount = keys.length;
@ -77,7 +77,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
private final class IndexedRangeSliceRunOp extends CqlRunOpFetchKeys
{
protected IndexedRangeSliceRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected IndexedRangeSliceRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
super(client, query, queryId, params, keyid, key);
}
@ -90,7 +90,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
}
@Override
protected CqlRunOp<byte[][]> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<byte[][]> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
return new IndexedRangeSliceRunOp(client, query, queryId, params, keyid, key);
}

10
tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java
View File

@ -33,12 +33,14 @@ public class CqlInserter extends CqlOperation<Integer>
public CqlInserter(State state, long idx)
{
super(state, idx);
if (state.settings.columns.useTimeUUIDComparator)
throw new IllegalStateException("Cannot use TimeUUID Comparator with CQL");
}
@Override
protected String buildQuery()
{
StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(state.settings.schema.columnFamily));
StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(state.type.table));
query.append(" SET ");
@ -66,9 +68,9 @@ public class CqlInserter extends CqlOperation<Integer>
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
final ArrayList<ByteBuffer> queryParams = new ArrayList<>();
final ArrayList<Object> queryParams = new ArrayList<>();
final List<ByteBuffer> values = generateColumnValues(ByteBuffer.wrap(key));
queryParams.addAll(values);
queryParams.add(ByteBuffer.wrap(key));
@ -76,7 +78,7 @@ public class CqlInserter extends CqlOperation<Integer>
}
@Override
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
return new CqlRunOpAlwaysSucceed(client, query, queryId, params, keyid, key, 1);
}

68
tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.stress.operations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@ -44,9 +45,9 @@ import org.apache.thrift.TException;
public abstract class CqlOperation<V> extends Operation
{
protected abstract List<ByteBuffer> getQueryParameters(byte[] key);
protected abstract List<Object> getQueryParameters(byte[] key);
protected abstract String buildQuery();
protected abstract CqlRunOp<V> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String id, ByteBuffer key);
protected abstract CqlRunOp<V> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String id, ByteBuffer key);
public CqlOperation(State state, long idx)
{
@ -55,9 +56,11 @@ public abstract class CqlOperation<V> extends Operation
throw new IllegalStateException("Super columns are not implemented for CQL");
if (state.settings.columns.variableColumnCount)
throw new IllegalStateException("Variable column counts are not implemented for CQL");
if (state.settings.columns.useTimeUUIDComparator)
throw new IllegalStateException("Cannot use TimeUUID Comparator with CQL");
}
protected CqlRunOp<V> run(final ClientWrapper client, final List<ByteBuffer> queryParams, final ByteBuffer key, final String keyid) throws IOException
protected CqlRunOp<V> run(final ClientWrapper client, final List<Object> queryParams, final ByteBuffer key, final String keyid) throws IOException
{
final CqlRunOp<V> op;
if (state.settings.mode.style == ConnectionStyle.CQL_PREPARED)
@ -99,7 +102,7 @@ public abstract class CqlOperation<V> extends Operation
protected void run(final ClientWrapper client) throws IOException
{
final byte[] key = getKey().array();
final List<ByteBuffer> queryParams = getQueryParameters(key);
final List<Object> queryParams = getQueryParameters(key);
run(client, queryParams, ByteBuffer.wrap(key), new String(key));
}
@ -111,7 +114,7 @@ public abstract class CqlOperation<V> extends Operation
final int keyCount;
protected CqlRunOpAlwaysSucceed(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String id, ByteBuffer key, int keyCount)
protected CqlRunOpAlwaysSucceed(ClientWrapper client, String query, Object queryId, List<Object> params, String id, ByteBuffer key, int keyCount)
{
super(client, query, queryId, RowCountHandler.INSTANCE, params, id, key);
this.keyCount = keyCount;
@ -134,7 +137,7 @@ public abstract class CqlOperation<V> extends Operation
protected final class CqlRunOpTestNonEmpty extends CqlRunOp<Integer>
{
protected CqlRunOpTestNonEmpty(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String id, ByteBuffer key)
protected CqlRunOpTestNonEmpty(ClientWrapper client, String query, Object queryId, List<Object> params, String id, ByteBuffer key)
{
super(client, query, queryId, RowCountHandler.INSTANCE, params, id, key);
}
@ -156,7 +159,7 @@ public abstract class CqlOperation<V> extends Operation
protected abstract class CqlRunOpFetchKeys extends CqlRunOp<byte[][]>
{
protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String id, ByteBuffer key)
protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List<Object> params, String id, ByteBuffer key)
{
super(client, query, queryId, KeysHandler.INSTANCE, params, id, key);
}
@ -175,7 +178,7 @@ public abstract class CqlOperation<V> extends Operation
final List<List<ByteBuffer>> expect;
// a null value for an item in expect means we just check the row is present
protected CqlRunOpMatchResults(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String id, ByteBuffer key, List<List<ByteBuffer>> expect)
protected CqlRunOpMatchResults(ClientWrapper client, String query, Object queryId, List<Object> params, String id, ByteBuffer key, List<List<ByteBuffer>> expect)
{
super(client, query, queryId, RowsHandler.INSTANCE, params, id, key);
this.expect = expect;
@ -209,13 +212,13 @@ public abstract class CqlOperation<V> extends Operation
final ClientWrapper client;
final String query;
final Object queryId;
final List<ByteBuffer> params;
final List<Object> params;
final String id;
final ByteBuffer key;
final ResultHandler<V> handler;
V result;
private CqlRunOp(ClientWrapper client, String query, Object queryId, ResultHandler<V> handler, List<ByteBuffer> params, String id, ByteBuffer key)
private CqlRunOp(ClientWrapper client, String query, Object queryId, ResultHandler<V> handler, List<Object> params, String id, ByteBuffer key)
{
this.client = client;
this.query = query;
@ -284,8 +287,8 @@ public abstract class CqlOperation<V> extends Operation
protected interface ClientWrapper
{
Object createPreparedStatement(String cqlQuery) throws TException;
<V> V execute(Object preparedStatementId, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler) throws TException;
<V> V execute(String query, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler) throws TException;
<V> V execute(Object preparedStatementId, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler) throws TException;
<V> V execute(String query, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler) throws TException;
}
private final class JavaDriverWrapper implements ClientWrapper
@ -297,14 +300,14 @@ public abstract class CqlOperation<V> extends Operation
}
@Override
public <V> V execute(String query, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler)
public <V> V execute(String query, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler)
{
String formattedQuery = formatCqlQuery(query, queryParams);
return handler.javaDriverHandler().apply(client.execute(formattedQuery, ThriftConversion.fromThrift(state.settings.command.consistencyLevel)));
}
@Override
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler)
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler)
{
return handler.javaDriverHandler().apply(
client.executePrepared(
@ -329,19 +332,19 @@ public abstract class CqlOperation<V> extends Operation
}
@Override
public <V> V execute(String query, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler)
public <V> V execute(String query, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler)
{
String formattedQuery = formatCqlQuery(query, queryParams);
return handler.thriftHandler().apply(client.execute(formattedQuery, ThriftConversion.fromThrift(state.settings.command.consistencyLevel)));
}
@Override
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler)
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler)
{
return handler.thriftHandler().apply(
client.executePrepared(
(byte[]) preparedStatementId,
queryParams,
toByteBufferParams(queryParams),
ThriftConversion.fromThrift(state.settings.command.consistencyLevel)));
}
@ -362,7 +365,7 @@ public abstract class CqlOperation<V> extends Operation
}
@Override
public <V> V execute(String query, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler) throws TException
public <V> V execute(String query, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler) throws TException
{
String formattedQuery = formatCqlQuery(query, queryParams);
return handler.simpleNativeHandler().apply(
@ -371,11 +374,11 @@ public abstract class CqlOperation<V> extends Operation
}
@Override
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<ByteBuffer> queryParams, ResultHandler<V> handler) throws TException
public <V> V execute(Object preparedStatementId, ByteBuffer key, List<Object> queryParams, ResultHandler<V> handler) throws TException
{
Integer id = (Integer) preparedStatementId;
return handler.simpleNativeHandler().apply(
client.execute_prepared_cql3_query(id, key, queryParams, state.settings.command.consistencyLevel)
client.execute_prepared_cql3_query(id, key, toByteBufferParams(queryParams), state.settings.command.consistencyLevel)
);
}
@ -608,7 +611,7 @@ public abstract class CqlOperation<V> extends Operation
* @param parms sequence of string query parameters
* @return formatted CQL query string
*/
private static String formatCqlQuery(String query, List<ByteBuffer> parms)
private static String formatCqlQuery(String query, List<Object> parms)
{
int marker, position = 0;
StringBuilder result = new StringBuilder();
@ -616,10 +619,15 @@ public abstract class CqlOperation<V> extends Operation
if (-1 == (marker = query.indexOf('?')) || parms.size() == 0)
return query;
for (ByteBuffer parm : parms)
for (Object parm : parms)
{
result.append(query.substring(position, marker));
result.append(getUnQuotedCqlBlob(parm));
if (parm instanceof ByteBuffer)
result.append(getUnQuotedCqlBlob((ByteBuffer) parm));
else if (parm instanceof Long)
result.append(parm.toString());
else throw new AssertionError();
position = marker + 1;
if (-1 == (marker = query.indexOf('?', position + 1)))
@ -632,6 +640,20 @@ public abstract class CqlOperation<V> extends Operation
return result.toString();
}
private static List<ByteBuffer> toByteBufferParams(List<Object> params)
{
List<ByteBuffer> r = new ArrayList<>();
for (Object param : params)
{
if (param instanceof ByteBuffer)
r.add((ByteBuffer) param);
else if (param instanceof Long)
r.add(ByteBufferUtil.bytes((Long) param));
else throw new AssertionError();
}
return r;
}
protected String wrapInQuotes(String string)
{
return "\"" + string + "\"";

8
tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java
View File

@ -33,19 +33,19 @@ public class CqlRangeSlicer extends CqlOperation<Integer>
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
return Collections.singletonList(ByteBuffer.wrap(key));
return Collections.<Object>singletonList(ByteBuffer.wrap(key));
}
@Override
protected String buildQuery()
{
return "SELECT FIRST " + state.settings.columns.maxColumnsPerKey + " ''..'' FROM " + state.settings.schema.columnFamily + " WHERE KEY > ?";
return "SELECT FIRST " + state.settings.columns.maxColumnsPerKey + " ''..'' FROM " + wrapInQuotes(state.type.table) + " WHERE KEY > ?";
}
@Override
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
return new CqlRunOpTestNonEmpty(client, query, queryId, params, keyid, key);
}

12
tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java
View File

@ -40,7 +40,7 @@ public class CqlReader extends CqlOperation<ByteBuffer[][]>
{
StringBuilder query = new StringBuilder("SELECT ");
if (state.settings.columns.names == null)
if (state.settings.columns.slice)
{
query.append("*");
}
@ -54,28 +54,28 @@ public class CqlReader extends CqlOperation<ByteBuffer[][]>
}
}
query.append(" FROM ").append(wrapInQuotes(state.settings.schema.columnFamily));
query.append(" FROM ").append(wrapInQuotes(state.type.table));
query.append(" WHERE KEY=?");
return query.toString();
}
@Override
protected List<ByteBuffer> getQueryParameters(byte[] key)
protected List<Object> getQueryParameters(byte[] key)
{
if (state.settings.columns.names != null)
{
final List<ByteBuffer> queryParams = new ArrayList<>();
final List<Object> queryParams = new ArrayList<>();
for (ByteBuffer name : state.settings.columns.names)
queryParams.add(name);
queryParams.add(ByteBuffer.wrap(key));
return queryParams;
}
return Collections.singletonList(ByteBuffer.wrap(key));
return Collections.<Object>singletonList(ByteBuffer.wrap(key));
}
@Override
protected CqlRunOp<ByteBuffer[][]> buildRunOp(ClientWrapper client, String query, Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
protected CqlRunOp<ByteBuffer[][]> buildRunOp(ClientWrapper client, String query, Object queryId, List<Object> params, String keyid, ByteBuffer key)
{
List<ByteBuffer> expectRow = state.rowGen.isDeterministic() ? generateColumnValues(key) : null;
return new CqlRunOpMatchResults(client, query, queryId, params, keyid, key, Arrays.asList(expectRow));

12
tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java
View File

@ -23,6 +23,8 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.util.ThriftClient;
@ -33,15 +35,13 @@ public class ThriftCounterAdder extends Operation
public ThriftCounterAdder(State state, long index)
{
super(state, index);
if (state.settings.columns.variableColumnCount)
throw new IllegalStateException("Variable column counts not supported for counters");
}
public void run(final ThriftClient client) throws IOException
{
List<CounterColumn> columns = new ArrayList<>();
for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++)
columns.add(new CounterColumn(getColumnNameBytes(i), 1L));
for (ByteBuffer name : randomNames())
columns.add(new CounterColumn(name, state.counteradd.next()));
Map<String, List<Mutation>> row;
if (state.settings.columns.useSuperColumns)
@ -53,7 +53,7 @@ public class ThriftCounterAdder extends Operation
ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setCounter_super_column(csc);
mutations.add(new Mutation().setColumn_or_supercolumn(cosc));
}
row = Collections.singletonMap("SuperCounter1", mutations);
row = Collections.singletonMap(state.type.supertable, mutations);
}
else
{
@ -63,7 +63,7 @@ public class ThriftCounterAdder extends Operation
ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setCounter_column(c);
mutations.add(new Mutation().setColumn_or_supercolumn(cosc));
}
row = Collections.singletonMap("Counter1", mutations);
row = Collections.singletonMap(state.type.table, mutations);
}
final ByteBuffer key = getKey();

15
tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterGetter.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.stress.operations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.util.ThriftClient;
@ -31,20 +32,11 @@ public class ThriftCounterGetter extends Operation
public ThriftCounterGetter(State state, long index)
{
super(state, index);
if (state.settings.columns.variableColumnCount)
throw new IllegalStateException("Variable column counts not supported for counters");
}
public void run(final ThriftClient client) throws IOException
{
SliceRange sliceRange = new SliceRange();
// start/finish
sliceRange.setStart(new byte[] {}).setFinish(new byte[] {});
// reversed/count
sliceRange.setReversed(false).setCount(state.settings.columns.maxColumnsPerKey);
// initialize SlicePredicate with existing SliceRange
final SlicePredicate predicate = new SlicePredicate().setSlice_range(sliceRange);
final SlicePredicate predicate = slicePredicate();
final ByteBuffer key = getKey();
for (final ColumnParent parent : state.columnParents)
{
@ -54,7 +46,8 @@ public class ThriftCounterGetter extends Operation
@Override
public boolean run() throws Exception
{
return client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel).size() != 0;
List<?> r = client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel);
return r != null && r.size() > 0;
}
@Override

5
tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java
View File

@ -23,7 +23,6 @@ import java.util.Arrays;
import java.util.List;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.settings.SettingsCommandMulti;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.thrift.*;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -52,7 +51,7 @@ public class ThriftIndexedRangeSlicer extends Operation
final List<ByteBuffer> columns = generateColumnValues(getKey());
final ColumnParent parent = state.columnParents.get(0);
final ByteBuffer columnName = getColumnNameBytes(1);
final ByteBuffer columnName = state.settings.columns.names.get(1);
final ByteBuffer value = columns.get(1); // only C1 column is indexed
IndexExpression expression = new IndexExpression(columnName, IndexOperator.EQ, value);
@ -64,7 +63,7 @@ public class ThriftIndexedRangeSlicer extends Operation
final boolean first = minKey.length == 0;
final IndexClause clause = new IndexClause(Arrays.asList(expression),
ByteBuffer.wrap(minKey),
((SettingsCommandMulti) state.settings.command).keysAtOnce);
state.settings.command.keysAtOnce);
timeWithRetry(new RunOp()
{

6
tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java
View File

@ -53,7 +53,7 @@ public final class ThriftInserter extends Operation
ColumnOrSuperColumn column = new ColumnOrSuperColumn().setColumn(c);
mutations.add(new Mutation().setColumn_or_supercolumn(column));
}
row = Collections.singletonMap(state.settings.schema.columnFamily, mutations);
row = Collections.singletonMap(state.type.table, mutations);
}
else
{
@ -64,7 +64,7 @@ public final class ThriftInserter extends Operation
final ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setSuper_column(s);
mutations.add(new Mutation().setColumn_or_supercolumn(cosc));
}
row = Collections.singletonMap("Super1", mutations);
row = Collections.singletonMap(state.settings.command.type.supertable, mutations);
}
final Map<ByteBuffer, Map<String, List<Mutation>>> record = Collections.singletonMap(key, row);
@ -104,7 +104,7 @@ public final class ThriftInserter extends Operation
// TODO : consider randomly allocating column names in case where have fewer than max columns
// but need to think about implications for indexes / indexed range slicer / other knock on effects
for (int i = 0 ; i < values.size() ; i++)
columns.add(new Column(getColumnNameBytes(i)));
columns.add(new Column(state.settings.columns.names.get(i)));
for (int i = 0 ; i < values.size() ; i++)
columns.get(i)

3
tools/stress/src/org/apache/cassandra/stress/operations/ThriftMultiGetter.java
View File

@ -22,7 +22,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.settings.SettingsCommandMulti;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.SlicePredicate;
@ -50,7 +49,7 @@ public final class ThriftMultiGetter extends Operation
)
);
final List<ByteBuffer> keys = getKeys(((SettingsCommandMulti) state.settings.command).keysAtOnce);
final List<ByteBuffer> keys = getKeys(state.settings.command.keysAtOnce);
for (final ColumnParent parent : state.columnParents)
{

3
tools/stress/src/org/apache/cassandra/stress/operations/ThriftRangeSlicer.java
View File

@ -21,7 +21,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.settings.SettingsCommandMulti;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.KeyRange;
@ -55,7 +54,7 @@ public final class ThriftRangeSlicer extends Operation
new KeyRange(state.settings.columns.maxColumnsPerKey)
.setStart_key(start)
.setEnd_key(ByteBufferUtil.EMPTY_BYTE_BUFFER)
.setCount(((SettingsCommandMulti)state.settings.command).keysAtOnce);
.setCount(state.settings.command.keysAtOnce);
for (final ColumnParent parent : state.columnParents)
{

16
tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java
View File

@ -19,7 +19,6 @@ package org.apache.cassandra.stress.operations;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import org.apache.cassandra.stress.Operation;
@ -27,7 +26,6 @@ import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.SuperColumn;
public final class ThriftReader extends Operation
@ -40,17 +38,7 @@ public final class ThriftReader extends Operation
public void run(final ThriftClient client) throws IOException
{
final SlicePredicate predicate = new SlicePredicate();
if (state.settings.columns.names == null)
predicate.setSlice_range(new SliceRange()
.setStart(new byte[] {})
.setFinish(new byte[] {})
.setReversed(false)
.setCount(state.settings.columns.maxColumnsPerKey)
);
else // see CASSANDRA-3064 about why this is useful
predicate.setColumn_names(state.settings.columns.names);
final SlicePredicate predicate = slicePredicate();
final ByteBuffer key = getKey();
final List<ByteBuffer> expect = state.rowGen.isDeterministic() ? generateColumnValues(key) : null;
for (final ColumnParent parent : state.columnParents)
@ -63,6 +51,8 @@ public final class ThriftReader extends Operation
List<ColumnOrSuperColumn> row = client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel);
if (expect == null)
return !row.isEmpty();
if (row == null)
return false;
if (!state.settings.columns.useSuperColumns)
{
if (row.size() != expect.size())

72
tools/stress/src/org/apache/cassandra/stress/settings/Command.java
View File

@ -27,54 +27,46 @@ import java.util.Map;
public enum Command
{
READ(false,
SettingsCommand.helpPrinter("read"),
READ(false, "Standard1", "Super1",
"Multiple concurrent reads - the cluster must first be populated by a write test",
CommandCategory.BASIC
),
WRITE(true,
SettingsCommand.helpPrinter("write"),
WRITE(true, "Standard1", "Super1",
"insert",
"Multiple concurrent writes against the cluster",
CommandCategory.BASIC
),
MIXED(true,
SettingsCommandMixed.helpPrinter(),
MIXED(true, null, null,
"Interleaving of any basic commands, with configurable ratio and distribution - the cluster must first be populated by a write test",
CommandCategory.MIXED
),
RANGESLICE(false,
SettingsCommandMulti.helpPrinter("range_slice"),
RANGESLICE(false, "Standard1", "Super1",
"Range slice queries - the cluster must first be populated by a write test",
CommandCategory.MULTI
),
IRANGESLICE(false,
SettingsCommandMulti.helpPrinter("indexed_range_slice"),
IRANGESLICE(false, "Standard1", "Super1",
"Range slice queries through a secondary index. The cluster must first be populated by a write test, with indexing enabled.",
CommandCategory.MULTI
CommandCategory.BASIC
),
READMULTI(false,
SettingsCommandMulti.helpPrinter("readmulti"),
READMULTI(false, "Standard1", "Super1",
"multi_read",
"Multiple concurrent reads fetching multiple rows at once. The cluster must first be populated by a write test.",
CommandCategory.MULTI
),
COUNTERWRITE(true,
SettingsCommand.helpPrinter("counteradd"),
COUNTERWRITE(true, "Counter1", "SuperCounter1",
"counter_add",
"Multiple concurrent updates of counters.",
CommandCategory.BASIC
),
COUNTERREAD(false,
SettingsCommand.helpPrinter("counterread"),
COUNTERREAD(false, "Counter1", "SuperCounter1",
"counter_get",
"Multiple concurrent reads of counters. The cluster must first be populated by a counterwrite test.",
CommandCategory.BASIC
),
HELP(false, SettingsMisc.helpHelpPrinter(), "-?", "Print help for a command or option", null),
PRINT(false, SettingsMisc.printHelpPrinter(), "Inspect the output of a distribution definition", null),
LEGACY(false, Legacy.helpPrinter(), "Legacy support mode", null)
HELP(false, null, null, "-?", "Print help for a command or option", null),
PRINT(false, null, null, "Inspect the output of a distribution definition", null),
LEGACY(false, null, null, "Legacy support mode", null)
;
@ -100,23 +92,49 @@ public enum Command
public final CommandCategory category;
public final String extraName;
public final String description;
public final Runnable helpPrinter;
public final String table;
public final String supertable;
Command(boolean updates, Runnable helpPrinter, String description, CommandCategory category)
Command(boolean updates, String table, String supertable, String description, CommandCategory category)
{
this(updates, helpPrinter, null, description, category);
this(updates, table, supertable, null, description, category);
}
Command(boolean updates, Runnable helpPrinter, String extra, String description, CommandCategory category)
Command(boolean updates, String table, String supertable, String extra, String description, CommandCategory category)
{
this.table = table;
this.supertable = supertable;
this.updates = updates;
this.category = category;
this.helpPrinter = helpPrinter;
this.extraName = extra;
this.description = description;
}
public void printHelp()
{
helpPrinter.run();
helpPrinter().run();
}
}
public final Runnable helpPrinter()
{
switch (this)
{
case PRINT:
return SettingsMisc.printHelpPrinter();
case HELP:
return SettingsMisc.helpHelpPrinter();
case LEGACY:
return Legacy.helpPrinter();
}
switch (category)
{
case BASIC:
case MULTI:
return SettingsCommand.helpPrinter(this);
case MIXED:
return SettingsCommandMixed.helpPrinter();
}
throw new AssertionError();
}
}

1
tools/stress/src/org/apache/cassandra/stress/settings/Option.java
View File

@ -31,6 +31,7 @@ abstract class Option
abstract String shortDisplay();
abstract String longDisplay();
abstract List<String> multiLineDisplay();
abstract boolean setByUser();
public int hashCode()
{

5
tools/stress/src/org/apache/cassandra/stress/settings/OptionDataGen.java
View File

@ -83,6 +83,11 @@ class OptionDataGen extends Option
return factory != null || defaultFactory != null;
}
public boolean setByUser()
{
return factory != null;
}
@Override
public String shortDisplay()
{

13
tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java
View File

@ -43,11 +43,13 @@ class OptionDistribution extends Option
final String prefix;
private String spec;
private final String defaultSpec;
private final String description;
public OptionDistribution(String prefix, String defaultSpec)
public OptionDistribution(String prefix, String defaultSpec, String description)
{
this.prefix = prefix;
this.defaultSpec = defaultSpec;
this.description = description;
}
@Override
@ -88,7 +90,7 @@ class OptionDistribution extends Option
public String longDisplay()
{
return shortDisplay() + ": Specify a mathematical distribution";
return shortDisplay() + ": " + description;
}
@Override
@ -105,10 +107,15 @@ class OptionDistribution extends Option
);
}
boolean setByUser()
{
return spec != null;
}
@Override
public String shortDisplay()
{
return prefix + "DIST(?)";
return (defaultSpec != null ? "[" : "") + prefix + "DIST(?)" + (defaultSpec != null ? "]" : "");
}
private static final Map<String, Impl> LOOKUP;

39
tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java
View File

@ -100,7 +100,7 @@ abstract class OptionMulti extends Option
StringBuilder sb = new StringBuilder();
sb.append(name);
sb.append("(");
for (Option option : options())
for (Option option : delegate.options())
{
sb.append(option);
sb.append(",");
@ -112,7 +112,7 @@ abstract class OptionMulti extends Option
@Override
public String shortDisplay()
{
return name + "(?)";
return (happy() ? "[" : "") + name + "(?)" + (happy() ? "]" : "");
}
@Override
@ -121,7 +121,7 @@ abstract class OptionMulti extends Option
StringBuilder sb = new StringBuilder();
sb.append(name);
sb.append("(");
for (Option opt : options())
for (Option opt : delegate.options())
{
sb.append(opt.shortDisplay());
}
@ -181,6 +181,37 @@ abstract class OptionMulti extends Option
{
return Collections.emptyList();
}
};
boolean setByUser()
{
return !options.isEmpty();
}
}
List<Option> optionsSetByUser()
{
List<Option> r = new ArrayList<>();
for (Option option : delegate.options())
if (option.setByUser())
r.add(option);
return r;
}
List<Option> defaultOptions()
{
List<Option> r = new ArrayList<>();
for (Option option : delegate.options())
if (!option.setByUser() && option.happy())
r.add(option);
return r;
}
boolean setByUser()
{
for (Option option : delegate.options())
if (option.setByUser())
return true;
return false;
}
}

2
tools/stress/src/org/apache/cassandra/stress/settings/OptionReplication.java
View File

@ -77,7 +77,7 @@ class OptionReplication extends OptionMulti
{
Class<?> clazz = Class.forName(fullname);
if (!AbstractReplicationStrategy.class.isAssignableFrom(clazz))
throw new RuntimeException();
throw new IllegalArgumentException(clazz + " is not a replication strategy");
strategy = fullname;
break;
} catch (Exception _)

2
tools/stress/src/org/apache/cassandra/stress/settings/OptionSimple.java
View File

@ -33,7 +33,7 @@ import com.google.common.base.Function;
class OptionSimple extends Option
{
private final String displayPrefix;
final String displayPrefix;
private final Pattern matchPrefix;
private final String defaultValue;
private final Function<String, String> valueAdapter;

29
tools/stress/src/org/apache/cassandra/stress/settings/SettingsColumn.java
View File

@ -30,6 +30,7 @@ import java.util.Map;
import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.stress.generatedata.*;
import org.apache.cassandra.utils.ByteBufferUtil;
/**
* For parsing column options
@ -39,12 +40,13 @@ public class SettingsColumn implements Serializable
public final int maxColumnsPerKey;
public final List<ByteBuffer> names;
public final List<String> namestrs;
public final String comparator;
public final boolean useTimeUUIDComparator;
public final int superColumns;
public final boolean useSuperColumns;
public final boolean variableColumnCount;
public final boolean slice;
private final DistributionFactory sizeDistribution;
private final DistributionFactory countDistribution;
private final DataGenFactory dataGenFactory;
@ -95,11 +97,12 @@ public class SettingsColumn implements Serializable
comparator = TypeParser.parse(this.comparator);
} catch (Exception e)
{
throw new IllegalStateException(e);
throw new IllegalArgumentException(this.comparator + " is not a valid type");
}
final String[] names = name.name.value().split(",");
this.names = new ArrayList<>(names.length);
this.namestrs = Arrays.asList(names);
for (String columnName : names)
this.names.add(comparator.fromString(columnName));
@ -117,10 +120,21 @@ public class SettingsColumn implements Serializable
else
{
this.countDistribution = count.count.get();
this.names = null;
ByteBuffer[] names = new ByteBuffer[(int) countDistribution.get().maxValue()];
String[] namestrs = new String[(int) countDistribution.get().maxValue()];
for (int i = 0 ; i < names.length ; i++)
{
names[i] = ByteBufferUtil.bytes("C" + i);
namestrs[i] = "C" + i;
}
this.names = Arrays.asList(names);
this.namestrs = Arrays.asList(namestrs);
}
maxColumnsPerKey = (int) countDistribution.get().maxValue();
variableColumnCount = countDistribution.get().minValue() < maxColumnsPerKey;
// TODO: should warn that we always slice for useTimeUUIDComparator?
slice = options.slice.setByUser() || useTimeUUIDComparator;
// TODO: with useTimeUUIDCOmparator, should we still try to select a random start for reads if possible?
}
public RowGen newRowGen()
@ -134,7 +148,8 @@ public class SettingsColumn implements Serializable
{
final OptionSimple superColumns = new OptionSimple("super=", "[0-9]+", "0", "Number of super columns to use (no super columns used if not specified)", false);
final OptionSimple comparator = new OptionSimple("comparator=", "TimeUUIDType|AsciiType|UTF8Type", "AsciiType", "Column Comparator to use", false);
final OptionDistribution size = new OptionDistribution("size=", "FIXED(34)");
final OptionSimple slice = new OptionSimple("slice", "", null, "If set, range slices will be used for reads, otherwise a names query will be", false);
final OptionDistribution size = new OptionDistribution("size=", "FIXED(34)", "Cell size distribution");
final OptionDataGen generator = new OptionDataGen("data=", "REPEAT(50)");
}
@ -145,18 +160,18 @@ public class SettingsColumn implements Serializable
@Override
public List<? extends Option> options()
{
return Arrays.asList(name, superColumns, comparator, size, generator);
return Arrays.asList(name, slice, superColumns, comparator, size, generator);
}
}
private static final class CountOptions extends Options
{
final OptionDistribution count = new OptionDistribution("n=", "FIXED(5)");
final OptionDistribution count = new OptionDistribution("n=", "FIXED(5)", "Cell count distribution, per operation");
@Override
public List<? extends Option> options()
{
return Arrays.asList(count, superColumns, comparator, size, generator);
return Arrays.asList(count, slice, superColumns, comparator, size, generator);
}
}

30
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommand.java
View File

@ -26,6 +26,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import org.apache.cassandra.stress.generatedata.DistributionFactory;
import org.apache.cassandra.thrift.ConsistencyLevel;
// Generic command settings - common to read/write/etc
@ -40,6 +41,8 @@ public class SettingsCommand implements Serializable
public final double targetUncertainty;
public final int minimumUncertaintyMeasurements;
public final int maximumUncertaintyMeasurements;
public final DistributionFactory add;
public final int keysAtOnce;
public SettingsCommand(Command type, GroupedOptions options)
{
@ -55,6 +58,8 @@ public class SettingsCommand implements Serializable
this.tries = Math.max(1, Integer.parseInt(options.retries.value()) + 1);
this.ignoreErrors = options.ignoreErrors.setByUser();
this.consistencyLevel = ConsistencyLevel.valueOf(options.consistencyLevel.value().toUpperCase());
this.keysAtOnce = Integer.parseInt(options.atOnce.value());
this.add = options.add.get();
if (count != null)
{
this.count = Long.parseLong(count.count.value());
@ -78,31 +83,29 @@ public class SettingsCommand implements Serializable
final OptionSimple retries = new OptionSimple("tries=", "[0-9]+", "9", "Number of tries to perform for each operation before failing", false);
final OptionSimple ignoreErrors = new OptionSimple("ignore_errors", "", null, "Do not print/log errors", false);
final OptionSimple consistencyLevel = new OptionSimple("cl=", "ONE|QUORUM|LOCAL_QUORUM|EACH_QUORUM|ALL|ANY", "ONE", "Consistency level to use", false);
final OptionDistribution add = new OptionDistribution("add=", "fixed(1)", "Distribution of value of counter increments");
final OptionSimple atOnce = new OptionSimple("at-once=", "[0-9]+", "1000", "Number of keys per operation for multiget", false);
}
static class Count extends Options
{
final OptionSimple count = new OptionSimple("n=", "[0-9]+", null, "Number of operations to perform", true);
@Override
public List<? extends Option> options()
{
return Arrays.asList(count, retries, ignoreErrors, consistencyLevel);
return Arrays.asList(count, retries, ignoreErrors, consistencyLevel, add, atOnce);
}
}
static class Uncertainty extends Options
{
final OptionSimple uncertainty = new OptionSimple("err<", "0\\.[0-9]+", "0.02", "Run until the standard error of the mean is below this fraction", false);
final OptionSimple minMeasurements = new OptionSimple("n>", "[0-9]+", "30", "Run at least this many iterations before accepting uncertainty convergence", false);
final OptionSimple maxMeasurements = new OptionSimple("n<", "[0-9]+", "200", "Run at most this many iterations before accepting uncertainty convergence", false);
@Override
public List<? extends Option> options()
{
return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, retries, ignoreErrors, consistencyLevel);
return Arrays.asList(uncertainty, minMeasurements, maxMeasurements, retries, ignoreErrors, consistencyLevel, add, atOnce);
}
}
@ -120,9 +123,8 @@ public class SettingsCommand implements Serializable
switch (cmd.category)
{
case BASIC:
return build(cmd, params);
case MULTI:
return SettingsCommandMulti.build(cmd, params);
return build(cmd, params);
case MIXED:
return SettingsCommandMixed.build(params);
}
@ -153,18 +155,6 @@ public class SettingsCommand implements Serializable
GroupedOptions.printOptions(System.out, type.toLowerCase(), new Uncertainty(), new Count());
}
static Runnable helpPrinter(final String type)
{
return new Runnable()
{
@Override
public void run()
{
printHelp(type);
}
};
}
static Runnable helpPrinter(final Command type)
{
return new Runnable()

46
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMixed.java
View File

@ -26,11 +26,12 @@ import java.util.List;
import org.apache.cassandra.stress.generatedata.Distribution;
import org.apache.cassandra.stress.generatedata.DistributionFactory;
import org.apache.commons.math3.distribution.EnumeratedDistribution;
import org.apache.commons.math3.util.Pair;
// Settings unique to the mixed command type
public class SettingsCommandMixed extends SettingsCommandMulti
public class SettingsCommandMixed extends SettingsCommand
{
// Ratios for selecting commands - index for each Command, NaN indicates the command is not requested
@ -41,21 +42,8 @@ public class SettingsCommandMixed extends SettingsCommandMulti
{
super(Command.MIXED, options.parent);
OptionSimple[] ratiosIn = options.probabilities.ratios;
List<Pair<Command, Double>> ratiosOut = new ArrayList<>();
for (int i = 0 ; i < ratiosIn.length ; i++)
{
if (ratiosIn[i] != null && ratiosIn[i].present())
{
double d = Double.parseDouble(ratiosIn[i].value());
if (d > 0)
ratiosOut.add(new Pair<>(Command.values()[i], d));
}
}
ratios = ratiosOut;
clustering = options.clustering.get();
ratios = options.probabilities.ratios();
if (ratios.size() == 0)
throw new IllegalArgumentException("Must specify at least one command with a non-zero ratio");
}
@ -144,16 +132,30 @@ public class SettingsCommandMixed extends SettingsCommandMulti
{
return grouping;
}
List<Pair<Command, Double>> ratios()
{
List<? extends Option> ratiosIn = setByUser() ? optionsSetByUser() : defaultOptions();
List<Pair<Command, Double>> ratiosOut = new ArrayList<>();
for (Option opt : ratiosIn)
{
OptionSimple ratioIn = (OptionSimple) opt;
Command command = Command.get(ratioIn.displayPrefix.substring(0, ratioIn.displayPrefix.length() - 1));
double d = Double.parseDouble(ratioIn.value());
ratiosOut.add(new Pair<>(command, d));
}
return ratiosOut;
}
}
static final class Options extends GroupedOptions
{
final SettingsCommandMulti.Options parent;
protected Options(SettingsCommandMulti.Options parent)
final SettingsCommand.Options parent;
protected Options(SettingsCommand.Options parent)
{
this.parent = parent;
}
final OptionDistribution clustering = new OptionDistribution("clustering=", "GAUSSIAN(1..10)");
final OptionDistribution clustering = new OptionDistribution("clustering=", "GAUSSIAN(1..10)", "Distribution clustering runs of operations of the same kind");
final Probabilities probabilities = new Probabilities();
@Override
@ -173,8 +175,8 @@ public class SettingsCommandMixed extends SettingsCommandMulti
public static SettingsCommandMixed build(String[] params)
{
GroupedOptions options = GroupedOptions.select(params,
new Options(new SettingsCommandMulti.Options(new Uncertainty())),
new Options(new SettingsCommandMulti.Options(new Count())));
new Options(new SettingsCommand.Uncertainty()),
new Options(new SettingsCommand.Count()));
if (options == null)
{
printHelp();
@ -187,8 +189,8 @@ public class SettingsCommandMixed extends SettingsCommandMulti
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "mixed",
new Options(new SettingsCommandMulti.Options(new Uncertainty())),
new Options(new SettingsCommandMulti.Options(new Count())));
new Options(new SettingsCommand.Uncertainty()),
new Options(new SettingsCommand.Count()));
}
public static Runnable helpPrinter()

90
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandMulti.java
View File

@ -1,90 +0,0 @@
package org.apache.cassandra.stress.settings;
/*
*
* 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.
*
*/
import java.util.ArrayList;
import java.util.List;
// Settings common to commands that operate over multiple keys at once
public class SettingsCommandMulti extends SettingsCommand
{
public final int keysAtOnce;
public SettingsCommandMulti(Command type, Options options)
{
super(type, options.parent);
this.keysAtOnce = Integer.parseInt(options.maxKeys.value());
}
// Option Declarations
static final class Options extends GroupedOptions
{
final GroupedOptions parent;
Options(GroupedOptions parent)
{
this.parent = parent;
}
final OptionSimple maxKeys = new OptionSimple("at-once=", "[0-9]+", "1000", "Number of keys per operation", false);
@Override
public List<? extends Option> options()
{
final List<Option> options = new ArrayList<>();
options.add(maxKeys);
options.addAll(parent.options());
return options;
}
}
// CLI Utility Methods
public static SettingsCommand build(Command type, String[] params)
{
GroupedOptions options = GroupedOptions.select(params, new Options(new Uncertainty()), new Options(new Count()));
if (options == null)
{
printHelp(type);
System.out.println("Invalid " + type + " options provided, see output for valid options");
System.exit(1);
}
return new SettingsCommandMulti(type, (Options) options);
}
public static void printHelp(Command type)
{
GroupedOptions.printOptions(System.out, type.toString().toLowerCase(), new Options(new Uncertainty()), new Options(new Count()));
}
public static Runnable helpPrinter(final Command type)
{
return new Runnable()
{
@Override
public void run()
{
printHelp(type);
}
};
}
}

2
tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
View File

@ -63,7 +63,7 @@ public class SettingsKey implements Serializable
public DistributionOptions(String defaultLimit)
{
dist = new OptionDistribution("dist=", "GAUSSIAN(1.." + defaultLimit + ")");
dist = new OptionDistribution("dist=", "GAUSSIAN(1.." + defaultLimit + ")", "Keys are selected from this distribution");
}
@Override

4
tools/stress/src/org/apache/cassandra/stress/settings/SettingsMisc.java
View File

@ -43,7 +43,7 @@ public class SettingsMisc implements Serializable
static final class PrintDistribution extends GroupedOptions
{
final OptionDistribution dist = new OptionDistribution("dist=", null);
final OptionDistribution dist = new OptionDistribution("dist=", null, "A mathematical distribution");
@Override
public List<? extends Option> options()
@ -180,7 +180,7 @@ public class SettingsMisc implements Serializable
@Override
public List<? extends Option> options()
{
return Arrays.asList(new OptionDistribution("dist=", null));
return Arrays.asList(new OptionDistribution("dist=", null, "A mathematical distribution"));
}
});
}

31
tools/stress/src/org/apache/cassandra/stress/settings/SettingsMode.java
View File

@ -42,7 +42,7 @@ public class SettingsMode implements Serializable
{
cqlVersion = CqlVersion.CQL3;
Cql3Options opts = (Cql3Options) options;
api = opts.useNative.setByUser() ? ConnectionAPI.JAVA_DRIVER_NATIVE : ConnectionAPI.THRIFT;
api = opts.mode().displayPrefix.equals("native") ? ConnectionAPI.JAVA_DRIVER_NATIVE : ConnectionAPI.THRIFT;
style = opts.usePrepared.setByUser() ? ConnectionStyle.CQL_PREPARED : ConnectionStyle.CQL;
compression = ProtocolOptions.Compression.valueOf(opts.useCompression.value().toUpperCase()).name();
}
@ -73,21 +73,40 @@ public class SettingsMode implements Serializable
// Option Declarations
private static final class Cql3Options extends GroupedOptions
private static final class Cql3NativeOptions extends Cql3Options
{
final OptionSimple mode = new OptionSimple("native", "", null, "", true);
OptionSimple mode()
{
return mode;
}
}
private static final class Cql3ThriftOptions extends Cql3Options
{
final OptionSimple mode = new OptionSimple("thrift", "", null, "", true);
OptionSimple mode()
{
return mode;
}
}
private static abstract class Cql3Options extends GroupedOptions
{
final OptionSimple api = new OptionSimple("cql3", "", null, "", true);
final OptionSimple useNative = new OptionSimple("native", "", null, "", false);
final OptionSimple usePrepared = new OptionSimple("prepared", "", null, "", false);
final OptionSimple useCompression = new OptionSimple("compression=", "none|lz4|snappy", "none", "", false);
final OptionSimple port = new OptionSimple("port=", "[0-9]+", "9046", "", false);
abstract OptionSimple mode();
@Override
public List<? extends Option> options()
{
return Arrays.asList(useNative, usePrepared, api, useCompression, port);
return Arrays.asList(mode(), usePrepared, api, useCompression, port);
}
}
private static final class Cql3SimpleNativeOptions extends GroupedOptions
{
final OptionSimple api = new OptionSimple("cql3", "", null, "", true);
@ -126,7 +145,7 @@ public class SettingsMode implements Serializable
return new SettingsMode(opts);
}
GroupedOptions options = GroupedOptions.select(params, new ThriftOptions(), new Cql3Options(), new Cql3SimpleNativeOptions());
GroupedOptions options = GroupedOptions.select(params, new ThriftOptions(), new Cql3NativeOptions(), new Cql3SimpleNativeOptions());
if (options == null)
{
printHelp();
@ -138,7 +157,7 @@ public class SettingsMode implements Serializable
public static void printHelp()
{
GroupedOptions.printOptions(System.out, "-mode", new ThriftOptions(), new Cql3Options(), new Cql3SimpleNativeOptions());
GroupedOptions.printOptions(System.out, "-mode", new ThriftOptions(), new Cql3NativeOptions(), new Cql3SimpleNativeOptions());
}
public static Runnable helpPrinter()

7
tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java
View File

@ -45,7 +45,6 @@ public class SettingsSchema implements Serializable
private final String compactionStrategy;
private final Map<String, String> compactionStrategyOptions;
public final String keyspace;
public final String columnFamily;
public SettingsSchema(Options options)
{
@ -60,7 +59,6 @@ public class SettingsSchema implements Serializable
compactionStrategy = options.compaction.getStrategy();
compactionStrategyOptions = options.compaction.getOptions();
keyspace = options.keyspace.value();
columnFamily = options.columnFamily.value();
}
public void createKeySpaces(StressSettings settings)
@ -77,7 +75,7 @@ public class SettingsSchema implements Serializable
KsDef ksdef = new KsDef();
// column family for standard columns
CfDef standardCfDef = new CfDef(keyspace, columnFamily);
CfDef standardCfDef = new CfDef(keyspace, "Standard1");
Map<String, String> compressionOptions = new HashMap<String, String>();
if (compression != null)
compressionOptions.put("sstable_compression", compression);
@ -202,14 +200,13 @@ public class SettingsSchema implements Serializable
final OptionCompaction compaction = new OptionCompaction();
final OptionSimple index = new OptionSimple("index=", "KEYS|CUSTOM|COMPOSITES", null, "Type of index to create on needed column families (KEYS)", false);
final OptionSimple keyspace = new OptionSimple("keyspace=", ".*", "Keyspace1", "The keyspace name to use", false);
final OptionSimple columnFamily = new OptionSimple("columnfamily=", ".*", "Standard1", "The column family name to use", false);
final OptionSimple noReplicateOnWrite = new OptionSimple("no-replicate-on-write", "", null, "Set replicate_on_write to false for counters. Only counter add with CL=ONE will work", false);
final OptionSimple compression = new OptionSimple("compression=", ".*", null, "Specify the compression to use for sstable, default:no compression", false);
@Override
public List<? extends Option> options()
{
return Arrays.asList(replication, index, keyspace, columnFamily, compaction, noReplicateOnWrite, compression);
return Arrays.asList(replication, index, keyspace, compaction, noReplicateOnWrite, compression);
}
}

2
tools/stress/src/org/apache/cassandra/stress/settings/SettingsTransport.java
View File

@ -47,7 +47,7 @@ public class SettingsTransport implements Serializable
{
Class<?> clazz = Class.forName(fqFactoryClass);
if (!ITransportFactory.class.isAssignableFrom(clazz))
throw new ClassCastException();
throw new IllegalArgumentException(clazz + " is not a valid transport factory");
// check we can instantiate it
clazz.newInstance();
}

30
tools/stress/src/org/apache/cassandra/stress/settings/StressSettings.java
View File

@ -172,12 +172,21 @@ public class StressSettings implements Serializable
public static StressSettings parse(String[] args)
{
final Map<String, String[]> clArgs = parseMap(args);
if (clArgs.containsKey("legacy"))
return Legacy.build(Arrays.copyOfRange(args, 1, args.length));
if (SettingsMisc.maybeDoSpecial(clArgs))
try
{
final Map<String, String[]> clArgs = parseMap(args);
if (clArgs.containsKey("legacy"))
return Legacy.build(Arrays.copyOfRange(args, 1, args.length));
if (SettingsMisc.maybeDoSpecial(clArgs))
System.exit(1);
return get(clArgs);
}
catch (IllegalArgumentException e)
{
System.out.println(e.getMessage());
System.exit(1);
return get(clArgs);
throw new AssertionError();
}
}
public static StressSettings get(Map<String, String[]> clArgs)
@ -231,17 +240,24 @@ public class StressSettings implements Serializable
if (i == 0 || args[i].startsWith("-"))
{
if (i > 0)
r.put(key, params.toArray(new String[0]));
putParam(key, params.toArray(new String[0]), r);
key = args[i].toLowerCase();
params.clear();
}
else
params.add(args[i]);
}
r.put(key, params.toArray(new String[0]));
putParam(key, params.toArray(new String[0]), r);
return r;
}
private static void putParam(String key, String[] args, Map<String, String[]> clArgs)
{
String[] prev = clArgs.put(key, args);
if (prev != null)
throw new IllegalArgumentException(key + " is defined multiple times. Each option/command can be specified at most once.");
}
public static void printHelp()
{
SettingsMisc.printHelp();

4
tools/stress/src/org/apache/cassandra/stress/util/JavaDriverClient.java
View File

@ -103,11 +103,11 @@ public class JavaDriverClient
return getSession().execute(stmt);
}
public ResultSet executePrepared(PreparedStatement stmt, List<ByteBuffer> queryParams, org.apache.cassandra.db.ConsistencyLevel consistency)
public ResultSet executePrepared(PreparedStatement stmt, List<Object> queryParams, org.apache.cassandra.db.ConsistencyLevel consistency)
{
stmt.setConsistencyLevel(from(consistency));
BoundStatement bstmt = stmt.bind((Object[]) queryParams.toArray(new ByteBuffer[queryParams.size()]));
BoundStatement bstmt = stmt.bind((Object[]) queryParams.toArray(new Object[queryParams.size()]));
return getSession().execute(bstmt);
}