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 

Conflicts:
	tools/stress/src/org/apache/cassandra/stress/generate/PartitionIterator.java
	tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java
This commit is contained in:
Benedict Elliott Smith 2015-05-04 18:03:43 +01:00
parent 47964b766c 3bee990ca2
commit ed08402999
15 changed files with 581 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -98,6 +98,7 @@
* fix nodetool names that reference column families (CASSANDRA-8872)
2.1.6
* cassandra-stress supports validation operations over user profiles (CASSANDRA-8773)
* Add support for rate limiting log messages (CASSANDRA-9029)
* Log the partition key with tombstone warnings (CASSANDRA-8561)
* Reduce runWithCompactionsDisabled poll interval to 1ms (CASSANDRA-9271)

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

@ -104,10 +104,7 @@ public abstract class Operation
if (seed == null)
break;
if (spec.useRatio == null)
success = partitionCache.get(i).reset(seed, spec.targetCount, isWrite());
else
success = partitionCache.get(i).reset(seed, spec.useRatio.next(), isWrite());
success = reset(seed, partitionCache.get(i));
}
}
partitionCount = i;
@ -119,6 +116,14 @@ public abstract class Operation
return !partitions.isEmpty();
}
protected boolean reset(Seed seed, PartitionIterator iterator)
{
if (spec.useRatio == null)
return iterator.reset(seed, spec.targetCount, isWrite());
else
return iterator.reset(seed, spec.useRatio.next(), isWrite());
}
public boolean isWrite()
{
return false;

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

@ -88,7 +88,9 @@ public class StressAction implements Runnable
// warmup - do 50k iterations; by default hotspot compiles methods after 10k invocations
PrintStream warmupOutput = new PrintStream(new OutputStream() { @Override public void write(int b) throws IOException { } } );
int iterations = 50000 * settings.node.nodes.size();
int threads = 20;
int threads = 100;
if (iterations > settings.command.count && settings.command.count > 0)
return;
if (settings.rate.maxThreads > 0)
threads = Math.min(threads, settings.rate.maxThreads);

25
tools/stress/src/org/apache/cassandra/stress/StressProfile.java
View File

@ -43,6 +43,7 @@ import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.generate.values.*;
import org.apache.cassandra.stress.operations.userdefined.SchemaInsert;
import org.apache.cassandra.stress.operations.userdefined.SchemaQuery;
import org.apache.cassandra.stress.operations.userdefined.ValidatingSchemaQuery;
import org.apache.cassandra.stress.settings.*;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
@ -75,6 +76,7 @@ public class StressProfile implements Serializable
transient volatile RatioDistributionFactory selectchance;
transient volatile PreparedStatement insertStatement;
transient volatile Integer thriftInsertId;
transient volatile List<ValidatingSchemaQuery.Factory> validationFactories;
transient volatile Map<String, SchemaQuery.ArgSelect> argSelects;
transient volatile Map<String, PreparedStatement> queryStatements;
@ -265,12 +267,11 @@ public class StressProfile implements Serializable
}
}
// TODO validation
name = name.toLowerCase();
if (!queryStatements.containsKey(name))
throw new IllegalArgumentException("No query defined with name " + name);
return new SchemaQuery(timer, settings, generator, seeds, thriftQueryIds.get(name), queryStatements.get(name),
ThriftConversion.fromThrift(settings.command.consistencyLevel), ValidationType.NOT_FAIL, argSelects.get(name));
ThriftConversion.fromThrift(settings.command.consistencyLevel), argSelects.get(name));
}
public SchemaInsert getInsert(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
@ -388,6 +389,26 @@ public class StressProfile implements Serializable
return new SchemaInsert(timer, settings, generator, seedManager, partitions.get(), selectchance.get(), thriftInsertId, insertStatement, ThriftConversion.fromThrift(settings.command.consistencyLevel), batchType);
}
public List<ValidatingSchemaQuery> getValidate(Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings)
{
if (validationFactories == null)
{
synchronized (this)
{
if (validationFactories == null)
{
maybeLoadSchemaInfo(settings);
validationFactories = ValidatingSchemaQuery.create(tableMetaData, settings);
}
}
}
List<ValidatingSchemaQuery> queries = new ArrayList<>();
for (ValidatingSchemaQuery.Factory factory : validationFactories)
queries.add(factory.create(timer, settings, generator, seedManager, ThriftConversion.fromThrift(settings.command.consistencyLevel)));
return queries;
}
private static <E> E select(E first, String key, String defValue, Map<String, String> map, Function<String, E> builder)
{
String val = map.remove(key);

9
tools/stress/src/org/apache/cassandra/stress/generate/PartitionGenerator.java
View File

@ -101,4 +101,13 @@ public class PartitionGenerator
return clusteringComponents.get(c).type.decompose(v);
return valueComponents.get(c - clusteringComponents.size()).type.decompose(v);
}
public Object convert(int c, ByteBuffer v)
{
if (c < 0)
return partitionKey.get(-1-c).type.compose(v);
if (c < clusteringComponents.size())
return clusteringComponents.get(c).type.compose(v);
return valueComponents.get(c - clusteringComponents.size()).type.compose(v);
}
}

165
tools/stress/src/org/apache/cassandra/stress/generate/PartitionIterator.java
View File

@ -36,9 +36,12 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
import com.google.common.collect.Iterables;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.stress.generate.values.Generator;
import org.apache.cassandra.utils.Pair;
// a partition is re-used to reduce garbage generation, as is its internal RowIterator
// TODO: we should batch the generation of clustering components so we can bound the time and size necessary to
@ -49,8 +52,11 @@ import org.apache.cassandra.stress.generate.values.Generator;
public abstract class PartitionIterator implements Iterator<Row>
{
abstract boolean reset(double useChance, int targetCount, boolean isWrite);
abstract boolean reset(double useChance, int targetCount, boolean isWrite, PartitionGenerator.Order order);
// picks random (inclusive) bounds to iterate, and returns them
public abstract Pair<Row, Row> resetToBounds(Seed seed, int clusteringComponentDepth);
PartitionGenerator.Order order;
long idseed;
Seed seed;
@ -77,7 +83,7 @@ public abstract class PartitionIterator implements Iterator<Row>
this.row = new Row(partitionKey, new Object[generator.clusteringComponents.size() + generator.valueComponents.size()]);
}
private void setSeed(Seed seed)
void setSeed(Seed seed)
{
long idseed = 0;
for (int i = 0 ; i < partitionKey.length ; i++)
@ -97,13 +103,15 @@ public abstract class PartitionIterator implements Iterator<Row>
public boolean reset(Seed seed, double useChance, boolean isWrite)
{
setSeed(seed);
return reset(useChance, 0, isWrite);
this.order = generator.order;
return reset(useChance, 0, isWrite, PartitionIterator.this.order);
}
public boolean reset(Seed seed, int targetCount, boolean isWrite)
{
setSeed(seed);
return reset(Double.NaN, targetCount, isWrite);
this.order = generator.order;
return reset(Double.NaN, targetCount, isWrite, PartitionIterator.this.order);
}
static class SingleRowIterator extends PartitionIterator
@ -116,7 +124,15 @@ public abstract class PartitionIterator implements Iterator<Row>
super(generator, seedManager);
}
boolean reset(double useChance, int targetCount, boolean isWrite)
public Pair<Row, Row> resetToBounds(Seed seed, int clusteringComponentDepth)
{
assert clusteringComponentDepth == 0;
setSeed(seed);
reset(1d, 1, false, PartitionGenerator.Order.SORTED);
return Pair.create(new Row(partitionKey), new Row(partitionKey));
}
boolean reset(double useChance, int targetCount, boolean isWrite, PartitionGenerator.Order order)
{
done = false;
this.isWrite = isWrite;
@ -200,16 +216,11 @@ public abstract class PartitionIterator implements Iterator<Row>
*
* @return true if there is data to return, false otherwise
*/
boolean reset(double useChance, int targetCount, boolean isWrite)
boolean reset(double useChance, int targetCount, boolean isWrite, PartitionGenerator.Order order)
{
this.isWrite = isWrite;
if (this.useChance < 1d)
{
// we clear our prior roll-modifiers if the use chance was previously less-than zero
Arrays.fill(rollmodifier, 1d);
Arrays.fill(chancemodifier, 1d);
}
this.order = order;
// set the seed for the first clustering component
generator.clusteringComponents.get(0).setSeed(idseed);
@ -228,7 +239,7 @@ public abstract class PartitionIterator implements Iterator<Row>
if (Double.isNaN(useChance))
useChance = Math.max(0d, Math.min(1d, targetCount / (double) expectedRowCount));
this.useChance = useChance;
setUseChance(useChance);
while (true)
{
@ -238,8 +249,7 @@ public abstract class PartitionIterator implements Iterator<Row>
for (Queue<?> q : clusteringComponents)
q.clear();
clusteringSeeds[0] = idseed;
fill(clusteringComponents[0], firstComponentCount, generator.clusteringComponents.get(0));
fill(0);
if (!isWrite)
{
@ -248,8 +258,7 @@ public abstract class PartitionIterator implements Iterator<Row>
return true;
}
int count = Math.max(1, expectedRowCount / seed.visits);
int count = seed.visits == 1 ? 1 + (int) generator.maxRowCount : Math.max(1, expectedRowCount / seed.visits);
position = seed.moveForwards(count);
isFirstWrite = position == 0;
setLastRow(position + count - 1);
@ -265,6 +274,44 @@ public abstract class PartitionIterator implements Iterator<Row>
}
}
void setUseChance(double useChance)
{
if (this.useChance < 1d)
{
// we clear our prior roll-modifiers if the use chance was previously less-than zero
Arrays.fill(rollmodifier, 1d);
Arrays.fill(chancemodifier, 1d);
}
this.useChance = useChance;
}
public Pair<Row, Row> resetToBounds(Seed seed, int clusteringComponentDepth)
{
setSeed(seed);
setUseChance(1d);
if (clusteringComponentDepth == 0)
{
reset(1d, -1, false, PartitionGenerator.Order.SORTED);
return Pair.create(new Row(partitionKey), new Row(partitionKey));
}
this.order = PartitionGenerator.Order.SORTED;
assert clusteringComponentDepth <= clusteringComponents.length;
for (Queue<?> q : clusteringComponents)
q.clear();
fill(0);
Pair<int[], Object[]> bound1 = randomBound(clusteringComponentDepth);
Pair<int[], Object[]> bound2 = randomBound(clusteringComponentDepth);
if (compare(bound1.left, bound2.left) > 0) { Pair<int[], Object[]> tmp = bound1; bound1 = bound2; bound2 = tmp;}
Arrays.fill(lastRow, 0);
System.arraycopy(bound2.left, 0, lastRow, 0, bound2.left.length);
Arrays.fill(currentRow, 0);
System.arraycopy(bound1.left, 0, currentRow, 0, bound1.left.length);
seekToCurrentRow();
return Pair.create(new Row(partitionKey, bound1.right), new Row(partitionKey, bound2.right));
}
// returns expected row count
private int setNoLastRow(int firstComponentCount)
{
@ -295,12 +342,39 @@ public abstract class PartitionIterator implements Iterator<Row>
// OR if that row does not exist, it is the last row prior to it
private int compareToLastRow(int depth)
{
int prev = 0;
for (int i = 0 ; i <= depth ; i++)
{
int p = currentRow[i], l = lastRow[i], r = clusteringComponents[i].size();
if ((p == l) || (r == 1))
continue;
return p - l;
if (prev < 0)
{
// if we're behind our last position in theory, and have known more items to visit in practice
// we're definitely behind our last row
if (r > 1)
return -1;
// otherwise move forwards to see if we might have more to visit
}
else if (p > l)
{
// prev must be == 0, so if p > l, we're after our last row
return 1;
}
else if (p == l)
{
// if we're equal to our last row up to our current depth, then we need to loop and look forwards
}
else if (r == 1)
{
// if this is our last item in practice, store if we're behind our theoretical position
// and move forwards; if every remaining practical item is 1, we're at the last row
// otherwise we're before it
prev = p - l;
}
else
{
// p < l, and r > 1, so we're definitely not at the end
return -1;
}
}
return 0;
}
@ -329,6 +403,14 @@ public abstract class PartitionIterator implements Iterator<Row>
}
}
private static int compare(int[] l, int[] r)
{
for (int i = 0 ; i < l.length ; i++)
if (l[i] != r[i])
return Integer.compare(l[i], r[i]);
return 0;
}
static enum State
{
END_OF_PARTITION, AFTER_LIMIT, SUCCESS;
@ -348,9 +430,12 @@ public abstract class PartitionIterator implements Iterator<Row>
clusteringComponents[0].addFirst(this);
return setHasNext(advance(0, true));
}
decompose(scalar, this.currentRow);
return seekToCurrentRow();
}
private State seekToCurrentRow()
{
int[] position = this.currentRow;
decompose(scalar, position);
for (int i = 0 ; i < position.length ; i++)
{
if (i != 0)
@ -398,7 +483,7 @@ public abstract class PartitionIterator implements Iterator<Row>
// normal method for moving the iterator forward; maintains the row object, and delegates to advance(int)
// to move the iterator to the next item
void advance()
Row advance()
{
// we are always at the leaf level when this method is invoked
// so we calculate the seed for generating the row by combining the seed that generated the clustering components
@ -406,16 +491,18 @@ public abstract class PartitionIterator implements Iterator<Row>
long parentSeed = clusteringSeeds[depth];
long rowSeed = seed(clusteringComponents[depth].peek(), generator.clusteringComponents.get(depth).type, parentSeed);
Row result = row.copy();
// and then fill the row with the _non-clustering_ values for the position we _were_ at, as this is what we'll deliver
for (int i = clusteringSeeds.length ; i < row.row.length ; i++)
{
Generator gen = generator.valueComponents.get(i - clusteringSeeds.length);
gen.setSeed(rowSeed);
row.row[i] = gen.generate();
result.row[i] = gen.generate();
}
// then we advance the leaf level
setHasNext(advance(depth, false));
return result;
}
private boolean advance(int depth, boolean first)
@ -480,15 +567,33 @@ public abstract class PartitionIterator implements Iterator<Row>
}
}
private Pair<int[], Object[]> randomBound(int clusteringComponentDepth)
{
ThreadLocalRandom rnd = ThreadLocalRandom.current();
int[] position = new int[clusteringComponentDepth];
Object[] bound = new Object[clusteringComponentDepth];
position[0] = rnd.nextInt(clusteringComponents[0].size());
bound[0] = Iterables.get(clusteringComponents[0], position[0]);
for (int d = 1 ; d < clusteringComponentDepth ; d++)
{
fill(d);
position[d] = rnd.nextInt(clusteringComponents[d].size());
bound[d] = Iterables.get(clusteringComponents[d], position[d]);
}
for (int d = 1 ; d < clusteringComponentDepth ; d++)
clusteringComponents[d].clear();
return Pair.create(position, bound);
}
// generate the clustering components for the provided depth; requires preceding components
// to have been generated and their seeds populated into clusteringSeeds
void fill(int depth)
{
long seed = clusteringSeeds[depth - 1];
long seed = depth == 0 ? idseed : clusteringSeeds[depth - 1];
Generator gen = generator.clusteringComponents.get(depth);
gen.setSeed(seed);
clusteringSeeds[depth] = seed(clusteringComponents[depth - 1].peek(), generator.clusteringComponents.get(depth - 1).type, seed);
fill(clusteringComponents[depth], (int) gen.clusteringDistribution.next(), gen);
clusteringSeeds[depth] = seed(clusteringComponents[depth].peek(), generator.clusteringComponents.get(depth).type, seed);
}
// generate the clustering components into the queue
@ -500,7 +605,7 @@ public abstract class PartitionIterator implements Iterator<Row>
return;
}
switch (this.generator.order)
switch (order)
{
case SORTED:
if (Comparable.class.isAssignableFrom(generator.clazz))
@ -510,7 +615,7 @@ public abstract class PartitionIterator implements Iterator<Row>
tosort.add(generator.generate());
Collections.sort((List<Comparable>) (List<?>) tosort);
for (int i = 0 ; i < count ; i++)
if (i == 0 || ((Comparable) tosort.get(i - 1)).compareTo(i) < 0)
if (i == 0 || ((Comparable) tosort.get(i - 1)).compareTo(tosort.get(i)) < 0)
queue.add(tosort.get(i));
break;
}
@ -555,8 +660,7 @@ public abstract class PartitionIterator implements Iterator<Row>
{
if (!hasNext())
throw new NoSuchElementException();
advance();
return row;
return advance();
}
public boolean finishedPartition()
@ -645,5 +749,4 @@ public abstract class PartitionIterator implements Iterator<Row>
{
return generator.partitionKey.get(0).type.decompose(partitionKey[0]);
}
}

15
tools/stress/src/org/apache/cassandra/stress/generate/Row.java
View File

@ -23,9 +23,16 @@ package org.apache.cassandra.stress.generate;
public class Row
{
private static final Object[] EMPTY_ROW_DATA = new Object[0];
final Object[] partitionKey;
final Object[] row;
public final Object[] partitionKey;
public final Object[] row;
public Row(Object[] partitionKey)
{
this.partitionKey = partitionKey;
this.row = EMPTY_ROW_DATA;
}
public Row(Object[] partitionKey, Object[] row)
{
@ -40,4 +47,8 @@ public class Row
return row[column];
}
public Row copy()
{
return new Row(partitionKey.clone(), row.clone());
}
}

23
tools/stress/src/org/apache/cassandra/stress/operations/SampledOpDistributionFactory.java
View File

@ -21,11 +21,9 @@ package org.apache.cassandra.stress.operations;
*/
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.*;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.util.Timing;
import org.apache.commons.math3.distribution.EnumeratedDistribution;
import org.apache.commons.math3.util.Pair;
@ -46,7 +44,7 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
this.clustering = clustering;
}
protected abstract Operation get(Timer timer, PartitionGenerator generator, T key);
protected abstract List<? extends Operation> get(Timer timer, PartitionGenerator generator, T key);
protected abstract PartitionGenerator newGenerator();
public OpDistribution get(Timing timing, int sampleCount)
@ -54,8 +52,11 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
PartitionGenerator generator = newGenerator();
List<Pair<Operation, Double>> operations = new ArrayList<>();
for (Map.Entry<T, Double> ratio : ratios.entrySet())
operations.add(new Pair<>(get(timing.newTimer(ratio.getKey().toString(), sampleCount), generator, ratio.getKey()),
ratio.getValue()));
{
List<? extends Operation> ops = get(timing.newTimer(ratio.getKey().toString(), sampleCount), generator, ratio.getKey());
for (Operation op : ops)
operations.add(new Pair<>(op, ratio.getValue() / ops.size()));
}
return new SampledOpDistribution(new EnumeratedDistribution<>(operations), clustering.get());
}
@ -76,7 +77,13 @@ public abstract class SampledOpDistributionFactory<T> implements OpDistributionF
{
public OpDistribution get(Timing timing, int sampleCount)
{
return new FixedOpDistribution(SampledOpDistributionFactory.this.get(timing.newTimer(ratio.getKey().toString(), sampleCount), newGenerator(), ratio.getKey()));
List<? extends Operation> ops = SampledOpDistributionFactory.this.get(timing.newTimer(ratio.getKey().toString(), sampleCount), newGenerator(), ratio.getKey());
if (ops.size() == 1)
return new FixedOpDistribution(ops.get(0));
List<Pair<Operation, Double>> ratios = new ArrayList<>();
for (Operation op : ops)
ratios.add(new Pair<>(op, 1d / ops.size()));
return new SampledOpDistribution(new EnumeratedDistribution<Operation>(ratios), new DistributionFixed(1));
}
public String desc()

14
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaInsert.java
View File

@ -32,7 +32,6 @@ import com.datastax.driver.core.Statement;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
@ -44,7 +43,7 @@ public class SchemaInsert extends SchemaStatement
public SchemaInsert(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, Distribution batchSize, RatioDistribution useRatio, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, BatchStatement.Type batchType)
{
super(timer, settings, new DataSpec(generator, seedManager, batchSize, useRatio), statement, thriftId, cl, ValidationType.NOT_FAIL);
super(timer, settings, new DataSpec(generator, seedManager, batchSize, useRatio), statement, thriftId, cl);
this.batchType = batchType;
}
@ -85,14 +84,7 @@ public class SchemaInsert extends SchemaStatement
stmt = batch;
}
try
{
validate(client.getSession().execute(stmt));
}
catch (ClassCastException e)
{
e.printStackTrace();
}
client.getSession().execute(stmt);
}
return true;
}
@ -113,7 +105,7 @@ public class SchemaInsert extends SchemaStatement
{
while (iterator.hasNext())
{
validate(client.execute_prepared_cql3_query(thriftId, iterator.getToken(), thriftRowArgs(iterator.next()), settings.command.consistencyLevel));
client.execute_prepared_cql3_query(thriftId, iterator.getToken(), thriftRowArgs(iterator.next()), settings.command.consistencyLevel);
rowCount += 1;
}
}

7
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaQuery.java
View File

@ -33,7 +33,6 @@ import com.datastax.driver.core.ResultSet;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
@ -52,9 +51,9 @@ public class SchemaQuery extends SchemaStatement
final Object[][] randomBuffer;
final Random random = new Random();
public SchemaQuery(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, ValidationType validationType, ArgSelect argSelect)
public SchemaQuery(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, Integer thriftId, PreparedStatement statement, ConsistencyLevel cl, ArgSelect argSelect)
{
super(timer, settings, new DataSpec(generator, seedManager, new DistributionFixed(1), argSelect == ArgSelect.MULTIROW ? statement.getVariables().size() : 1), statement, thriftId, cl, validationType);
super(timer, settings, new DataSpec(generator, seedManager, new DistributionFixed(1), argSelect == ArgSelect.MULTIROW ? statement.getVariables().size() : 1), statement, thriftId, cl);
this.argSelect = argSelect;
randomBuffer = new Object[argumentIndex.length][argumentIndex.length];
}
@ -71,7 +70,6 @@ public class SchemaQuery extends SchemaStatement
public boolean run() throws Exception
{
ResultSet rs = client.getSession().execute(bindArgs());
validate(rs);
rowCount = rs.all().size();
partitionCount = Math.min(1, rowCount);
return true;
@ -90,7 +88,6 @@ public class SchemaQuery extends SchemaStatement
public boolean run() throws Exception
{
CqlResult rs = client.execute_prepared_cql3_query(thriftId, partitions.get(0).getToken(), thriftArgs(), ThriftConversion.toThrift(cl));
validate(rs);
rowCount = rs.getRowsSize();
partitionCount = Math.min(1, rowCount);
return true;

35
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/SchemaStatement.java
View File

@ -34,7 +34,6 @@ import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.settings.ValidationType;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.CqlResult;
@ -46,18 +45,16 @@ public abstract class SchemaStatement extends Operation
final PreparedStatement statement;
final Integer thriftId;
final ConsistencyLevel cl;
final ValidationType validationType;
final int[] argumentIndex;
final Object[] bindBuffer;
public SchemaStatement(Timer timer, StressSettings settings, DataSpec spec,
PreparedStatement statement, Integer thriftId, ConsistencyLevel cl, ValidationType validationType)
PreparedStatement statement, Integer thriftId, ConsistencyLevel cl)
{
super(timer, settings, spec);
this.statement = statement;
this.thriftId = thriftId;
this.cl = cl;
this.validationType = validationType;
argumentIndex = new int[statement.getVariables().size()];
bindBuffer = new Object[argumentIndex.length];
int i = 0;
@ -86,36 +83,6 @@ public abstract class SchemaStatement extends Operation
return args;
}
void validate(ResultSet rs)
{
switch (validationType)
{
case NOT_FAIL:
return;
case NON_ZERO:
if (rs.all().size() == 0)
throw new IllegalStateException("Expected non-zero results");
break;
default:
throw new IllegalStateException("Unsupported validation type");
}
}
void validate(CqlResult rs)
{
switch (validationType)
{
case NOT_FAIL:
return;
case NON_ZERO:
if (rs.getRowsSize() == 0)
throw new IllegalStateException("Expected non-zero results");
break;
default:
throw new IllegalStateException("Unsupported validation type");
}
}
@Override
public void run(SimpleClient client) throws IOException
{

359
tools/stress/src/org/apache/cassandra/stress/operations/userdefined/ValidatingSchemaQuery.java Normal file
View File

@ -0,0 +1,359 @@
package org.apache.cassandra.stress.operations.userdefined;
/*
*
* 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.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import com.datastax.driver.core.*;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.*;
import org.apache.cassandra.stress.generate.Row;
import org.apache.cassandra.stress.settings.StressSettings;
import org.apache.cassandra.stress.util.JavaDriverClient;
import org.apache.cassandra.stress.util.ThriftClient;
import org.apache.cassandra.stress.util.Timer;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlRow;
import org.apache.cassandra.thrift.ThriftConversion;
import org.apache.cassandra.transport.SimpleClient;
import org.apache.cassandra.utils.Pair;
import org.apache.thrift.TException;
public class ValidatingSchemaQuery extends Operation
{
final Random random = new Random();
private Pair<Row, Row> bounds;
final int clusteringComponents;
final ValidatingStatement[] statements;
final ConsistencyLevel cl;
final int[] argumentIndex;
final Object[] bindBuffer;
@Override
public void run(SimpleClient client) throws IOException
{
throw new UnsupportedOperationException();
}
private ValidatingSchemaQuery(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, ValidatingStatement[] statements, ConsistencyLevel cl, int clusteringComponents)
{
super(timer, settings, new DataSpec(generator, seedManager, new DistributionFixed(1), 1));
this.statements = statements;
this.cl = cl;
argumentIndex = new int[statements[0].statement.getVariables().size()];
bindBuffer = new Object[argumentIndex.length];
int i = 0;
for (ColumnDefinitions.Definition definition : statements[0].statement.getVariables())
argumentIndex[i++] = spec.partitionGenerator.indexOf(definition.getName());
for (ValidatingStatement statement : statements)
statement.statement.setConsistencyLevel(JavaDriverClient.from(cl));
this.clusteringComponents = clusteringComponents;
}
protected boolean reset(Seed seed, PartitionIterator iterator)
{
bounds = iterator.resetToBounds(seed, clusteringComponents);
return true;
}
abstract class Runner implements RunOp
{
int partitionCount;
int rowCount;
final PartitionIterator iter;
final int statementIndex;
protected Runner(PartitionIterator iter)
{
this.iter = iter;
statementIndex = ThreadLocalRandom.current().nextInt(statements.length);
}
@Override
public int partitionCount()
{
return partitionCount;
}
@Override
public int rowCount()
{
return rowCount;
}
}
private class JavaDriverRun extends Runner
{
final JavaDriverClient client;
private JavaDriverRun(JavaDriverClient client, PartitionIterator iter)
{
super(iter);
this.client = client;
}
public boolean run() throws Exception
{
ResultSet rs = client.getSession().execute(bind(statementIndex));
int[] valueIndex = new int[rs.getColumnDefinitions().size()];
{
int i = 0;
for (ColumnDefinitions.Definition definition : rs.getColumnDefinitions())
valueIndex[i++] = spec.partitionGenerator.indexOf(definition.getName());
}
List<Object[]> prev1 = new ArrayList<>();
List<Object[]> prev2 = new ArrayList<>();
rowCount = 0;
Iterator<com.datastax.driver.core.Row> results = rs.iterator();
if (!statements[statementIndex].inclusiveStart && iter.hasNext())
iter.next();
while (iter.hasNext())
{
Row expectedRow = iter.next();
if (!statements[statementIndex].inclusiveEnd && !iter.hasNext())
break;
if (!results.hasNext())
return false;
rowCount++;
com.datastax.driver.core.Row actualRow = results.next();
Object[] vs1 = new Object[actualRow.getColumnDefinitions().size()];
Object[] vs2 = vs1.clone();
for (int i = 0 ; i < actualRow.getColumnDefinitions().size() ; i++)
{
Object expectedValue = expectedRow.get(valueIndex[i]);
Object actualValue = spec.partitionGenerator.convert(valueIndex[i], actualRow.getBytesUnsafe(i));
vs1[i] = expectedValue;
vs2[i] = actualValue;
if (!expectedValue.equals(actualValue))
return false;
}
prev1.add(vs1);
prev2.add(vs2);
}
partitionCount = Math.min(1, rowCount);
if (!rs.isExhausted())
return false;
return true;
}
}
private class ThriftRun extends Runner
{
final ThriftClient client;
private ThriftRun(ThriftClient client, PartitionIterator iter)
{
super(iter);
this.client = client;
}
public boolean run() throws Exception
{
CqlResult rs = client.execute_prepared_cql3_query(statements[statementIndex].thriftId, partitions.get(0).getToken(), thriftArgs(), ThriftConversion.toThrift(cl));
int[] valueIndex = new int[rs.getSchema().name_types.size()];
for (int i = 0 ; i < valueIndex.length ; i++)
valueIndex[i] = spec.partitionGenerator.indexOf(rs.fieldForId(i).getFieldName());
int r = 0;
if (!statements[statementIndex].inclusiveStart && iter.hasNext())
iter.next();
while (iter.hasNext())
{
Row expectedRow = iter.next();
if (!statements[statementIndex].inclusiveEnd && !iter.hasNext())
break;
if (r == rs.num)
return false;
rowCount++;
CqlRow actualRow = rs.getRows().get(r++);
for (int i = 0 ; i < actualRow.getColumnsSize() ; i++)
{
ByteBuffer expectedValue = spec.partitionGenerator.convert(valueIndex[i], expectedRow.get(valueIndex[i]));
ByteBuffer actualValue = actualRow.getColumns().get(i).value;
if (!expectedValue.equals(actualValue))
return false;
}
}
assert r == rs.num;
partitionCount = Math.min(1, rowCount);
return true;
}
}
BoundStatement bind(int statementIndex)
{
int pkc = bounds.left.partitionKey.length;
System.arraycopy(bounds.left.partitionKey, 0, bindBuffer, 0, pkc);
int ccc = bounds.left.row.length;
System.arraycopy(bounds.left.row, 0, bindBuffer, pkc, ccc);
System.arraycopy(bounds.right.row, 0, bindBuffer, pkc + ccc, ccc);
return statements[statementIndex].statement.bind(bindBuffer);
}
List<ByteBuffer> thriftArgs()
{
List<ByteBuffer> args = new ArrayList<>();
int pkc = bounds.left.partitionKey.length;
for (int i = 0 ; i < pkc ; i++)
args.add(spec.partitionGenerator.convert(-i, bounds.left.partitionKey[i]));
int ccc = bounds.left.row.length;
for (int i = 0 ; i < ccc ; i++)
args.add(spec.partitionGenerator.convert(i, bounds.left.get(i)));
for (int i = 0 ; i < ccc ; i++)
args.add(spec.partitionGenerator.convert(i, bounds.right.get(i)));
return args;
}
@Override
public void run(JavaDriverClient client) throws IOException
{
timeWithRetry(new JavaDriverRun(client, partitions.get(0)));
}
@Override
public void run(ThriftClient client) throws IOException
{
timeWithRetry(new ThriftRun(client, partitions.get(0)));
}
public static class Factory
{
final ValidatingStatement[] statements;
final int clusteringComponents;
public Factory(ValidatingStatement[] statements, int clusteringComponents)
{
this.statements = statements;
this.clusteringComponents = clusteringComponents;
}
public ValidatingSchemaQuery create(Timer timer, StressSettings settings, PartitionGenerator generator, SeedManager seedManager, ConsistencyLevel cl)
{
return new ValidatingSchemaQuery(timer, settings, generator, seedManager, statements, cl, clusteringComponents);
}
}
public static List<Factory> create(TableMetadata metadata, StressSettings settings)
{
List<Factory> factories = new ArrayList<>();
StringBuilder sb = new StringBuilder();
boolean first = true;
sb.append("SELECT * FROM ");
sb.append(metadata.getName());
sb.append(" WHERE");
for (ColumnMetadata pk : metadata.getPartitionKey())
{
sb.append(first ? " " : " AND ");
sb.append(pk.getName());
sb.append(" = ?");
first = false;
}
String base = sb.toString();
factories.add(new Factory(new ValidatingStatement[] { prepare(settings, base, true, true) }, 0));
int maxDepth = metadata.getClusteringColumns().size() - 1;
for (int depth = 0 ; depth <= maxDepth ; depth++)
{
StringBuilder cc = new StringBuilder();
StringBuilder arg = new StringBuilder();
cc.append("("); arg.append("(");
for (int d = 0 ; d <= depth ; d++)
{
if (d > 0) { cc.append(","); arg.append(","); }
cc.append(metadata.getClusteringColumns().get(d).getName());
arg.append("?");
}
cc.append(")"); arg.append(")");
ValidatingStatement[] statements = new ValidatingStatement[depth < maxDepth ? 1 : 4];
int i = 0;
for (boolean incLb : depth < maxDepth ? new boolean[] { true } : new boolean[] { true, false } )
{
for (boolean incUb : depth < maxDepth ? new boolean[] { false } : new boolean[] { true, false } )
{
String lb = incLb ? ">=" : ">";
String ub = incUb ? "<=" : "<";
sb.setLength(0);
sb.append(base);
sb.append(" AND ");
sb.append(cc);
sb.append(lb);
sb.append(arg);
sb.append(" AND ");
sb.append(cc);
sb.append(ub);
sb.append(arg);
statements[i++] = prepare(settings, sb.toString(), incLb, incUb);
}
}
factories.add(new Factory(statements, depth + 1));
}
return factories;
}
private static class ValidatingStatement
{
final PreparedStatement statement;
final Integer thriftId;
final boolean inclusiveStart;
final boolean inclusiveEnd;
private ValidatingStatement(PreparedStatement statement, Integer thriftId, boolean inclusiveStart, boolean inclusiveEnd)
{
this.statement = statement;
this.thriftId = thriftId;
this.inclusiveStart = inclusiveStart;
this.inclusiveEnd = inclusiveEnd;
}
}
private static ValidatingStatement prepare(StressSettings settings, String cql, boolean incLb, boolean incUb)
{
JavaDriverClient jclient = settings.getJavaDriverClient();
ThriftClient tclient = settings.getThriftClient();
PreparedStatement statement = jclient.prepare(cql);
try
{
Integer thriftId = tclient.prepare_cql3_query(cql, Compression.NONE);
return new ValidatingStatement(statement, thriftId, incLb, incUb);
}
catch (TException e)
{
throw new RuntimeException(e);
}
}
}

9
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandPreDefinedMixed.java
View File

@ -21,10 +21,7 @@ package org.apache.cassandra.stress.settings;
*/
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.*;
import org.apache.cassandra.stress.Operation;
import org.apache.cassandra.stress.generate.DistributionFactory;
@ -58,9 +55,9 @@ public class SettingsCommandPreDefinedMixed extends SettingsCommandPreDefined
final SeedManager seeds = new SeedManager(settings);
return new SampledOpDistributionFactory<Command>(ratios, clustering)
{
protected Operation get(Timer timer, PartitionGenerator generator, Command key)
protected List<? extends Operation> get(Timer timer, PartitionGenerator generator, Command key)
{
return PredefinedOperation.operation(key, timer, generator, seeds, settings, add);
return Collections.singletonList(PredefinedOperation.operation(key, timer, generator, seeds, settings, add));
}
protected PartitionGenerator newGenerator()

9
tools/stress/src/org/apache/cassandra/stress/settings/SettingsCommandUser.java
View File

@ -24,6 +24,7 @@ package org.apache.cassandra.stress.settings;
import java.io.File;
import java.net.URI;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@ -75,11 +76,13 @@ public class SettingsCommandUser extends SettingsCommand
final SeedManager seeds = new SeedManager(settings);
return new SampledOpDistributionFactory<String>(ratios, clustering)
{
protected Operation get(Timer timer, PartitionGenerator generator, String key)
protected List<? extends Operation> get(Timer timer, PartitionGenerator generator, String key)
{
if (key.equalsIgnoreCase("insert"))
return profile.getInsert(timer, generator, seeds, settings);
return profile.getQuery(key, timer, generator, seeds, settings);
return Collections.singletonList(profile.getInsert(timer, generator, seeds, settings));
if (key.equalsIgnoreCase("validate"))
return profile.getValidate(timer, generator, seeds, settings);
return Collections.singletonList(profile.getQuery(key, timer, generator, seeds, settings));
}
protected PartitionGenerator newGenerator()

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

@ -1,29 +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.
*
*/
public enum ValidationType
{
NOT_FAIL, NON_ZERO, SUBSET, EQUAL
}