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

Add query-by-column mode to stress.java 

patch by Pavel Yaskevich; reviewed by Jonathan Ellis for CASSANDRA-3064

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1160305 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Pavel Yaskevich 2011-08-22 15:31:59 +00:00
parent 791a78f92c
commit 477bb5f34b
3 changed files with 54 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGES.txt
View File

@ -37,7 +37,7 @@
and few other places responsible for work with SSTable files (CASSANDRA-3040)
* Stop reading from sstables once we know we have the most recent columns,
for query-by-name requests (CASSANDRA-2498)
* Add query-by-column mode to stress.java (CASSANDRA-3064)
0.8.5
* fix NPE when encryption_options is unspecified (CASSANDRA-3007)

42
tools/stress/src/org/apache/cassandra/stress/Session.java
View File

@ -20,10 +20,14 @@ package org.apache.cassandra.stress;
import java.io.*;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;
import org.apache.commons.cli.*;
import org.apache.cassandra.db.ColumnFamilyType;
@ -41,6 +45,9 @@ public class Session implements Serializable
// command line options
public static final Options availableOptions = new Options();
public static final String DEFAULT_COMPARATOR = "AsciiType";
public static final String DEFAULT_VALIDATOR = "BytesType";
public final AtomicInteger operations;
public final AtomicInteger keys;
public final AtomicLong latency;
@ -78,6 +85,7 @@ public class Session implements Serializable
availableOptions.addOption("V", "average-size-values", false, "Generate column values of average rather than specific size");
availableOptions.addOption("T", "send-to", true, "Send this as a request to the stress daemon at specified address.");
availableOptions.addOption("I", "compression", false, "Use sstable compression when creating schema");
availableOptions.addOption("Q", "query-names", true, "Comma-separated list of column names to retrieve from each row.");
}
private int numKeys = 1000 * 1000;
@ -109,6 +117,9 @@ public class Session implements Serializable
private String replicationStrategy = "org.apache.cassandra.locator.SimpleStrategy";
private Map<String, String> replicationStrategyOptions = new HashMap<String, String>();
// if we know exactly column names that we want to read (set by -Q option)
public final List<ByteBuffer> columnNames;
public final boolean averageSizeValues;
// required by Gaussian distribution.
@ -275,11 +286,30 @@ public class Session implements Serializable
{
throw new RuntimeException(e);
}
if (cmd.hasOption("Q"))
{
AbstractType comparator = TypeParser.parse(DEFAULT_COMPARATOR);
String[] names = StringUtils.split(cmd.getOptionValue("Q"), ",");
columnNames = new ArrayList<ByteBuffer>(names.length);
for (String columnName : names)
columnNames.add(comparator.fromString(columnName));
}
else
{
columnNames = null;
}
}
catch (ParseException e)
{
throw new IllegalArgumentException(e.getMessage(), e);
}
catch (ConfigurationException e)
{
throw new IllegalStateException(e.getMessage(), e);
}
mean = numDifferentKeys / 2;
sigma = numDifferentKeys * STDev;
@ -417,8 +447,11 @@ public class Session implements Serializable
// column family for standard columns
CfDef standardCfDef = new CfDef("Keyspace1", "Standard1");
System.out.println("Compression = " + compression);
standardCfDef.setComparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
standardCfDef.setComparator_type(DEFAULT_COMPARATOR)
.setDefault_validation_class(DEFAULT_VALIDATOR)
.setCompression(compression);
if (indexType != null)
{
ColumnDef standardColumn = new ColumnDef(ByteBufferUtil.bytes("C1"), "BytesType");
@ -428,7 +461,10 @@ public class Session implements Serializable
// column family with super columns
CfDef superCfDef = new CfDef("Keyspace1", "Super1").setColumn_type("Super");
superCfDef.setComparator_type("AsciiType").setSubcomparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
superCfDef.setComparator_type(DEFAULT_COMPARATOR)
.setSubcomparator_type(DEFAULT_COMPARATOR)
.setDefault_validation_class(DEFAULT_VALIDATOR)
.setCompression(compression);
// column family for standard counters
CfDef counterCfDef = new CfDef("Keyspace1", "Counter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setCompression(compression);

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

@ -37,16 +37,13 @@ public class Reader extends Operation
public void run(Cassandra.Client client) throws IOException
{
SliceRange sliceRange = new SliceRange();
// start/finish
sliceRange.setStart(new byte[] {}).setFinish(new byte[] {});
// reversed/count
sliceRange.setReversed(false).setCount(session.getColumnsPerKey());
// initialize SlicePredicate with existing SliceRange
SlicePredicate predicate = new SlicePredicate().setSlice_range(sliceRange);
SlicePredicate predicate = new SlicePredicate();
if (session.columnNames == null)
predicate.setSlice_range(getSliceRange());
else // see CASSANDRA-3064 about why this is useful
predicate.setColumn_names(session.columnNames);
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
{
@ -150,4 +147,12 @@ public class Reader extends Operation
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
private SliceRange getSliceRange()
{
return new SliceRange()
.setStart(new byte[] {})
.setFinish(new byte[] {})
.setReversed(false)
.setCount(session.getColumnsPerKey());
}
}