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

optional CQL query support 

Patch by eevans; reviewed by Pavel Yaskevich for CASSANDRA-2268


git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1211521 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Eric Evans 2011-12-07 16:36:19 +00:00
parent c38831b1c6
commit c0adddd077
10 changed files with 688 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -79,6 +79,7 @@ public class Session implements Serializable
availableOptions.addOption("i", "progress-interval", true, "Progress Report Interval (seconds), default:10");
availableOptions.addOption("g", "keys-per-call", true, "Number of keys to get_range_slices or multiget per call, default:1000");
availableOptions.addOption("l", "replication-factor", true, "Replication Factor to use when creating needed column families, default:1");
availableOptions.addOption("L", "enable-cql", false, "Perform queries using CQL (Cassandra Query Language).");
availableOptions.addOption("e", "consistency-level", true, "Consistency Level to use (ONE, QUORUM, LOCAL_QUORUM, EACH_QUORUM, ALL, ANY), default:ONE");
availableOptions.addOption("x", "create-index", true, "Type of index to create on needed column families (KEYS)");
availableOptions.addOption("R", "replication-strategy", true, "Replication strategy to use (only on insert if keyspace does not exist), default:org.apache.cassandra.locator.SimpleStrategy");
@ -112,6 +113,7 @@ public class Session implements Serializable
private int keysPerCall = 1000;
private boolean replicateOnWrite = true;
private boolean ignoreErrors = false;
private boolean enable_cql = false;
private final String outFileName;
@ -260,6 +262,9 @@ public class Session implements Serializable
else if (replicationStrategy.endsWith("SimpleStrategy"))
replicationStrategyOptions.put("replication_factor", "1");
if (cmd.hasOption("L"))
enable_cql = true;
if (cmd.hasOption("O"))
{
String[] pairs = StringUtils.split(cmd.getOptionValue("O"), ',');
@ -490,6 +495,11 @@ public class Session implements Serializable
return sigma;
}
public boolean isCQL()
{
return enable_cql;
}
/**
* Create Keyspace1 with Standard1 and Super1 column families
*/

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

@ -225,25 +225,25 @@ public class StressAction extends Thread
switch (client.getOperation())
{
case READ:
return new Reader(client, index);
return client.isCQL() ? new CqlReader(client, index) : new Reader(client, index);
case COUNTER_GET:
return new CounterGetter(client, index);
return client.isCQL() ? new CqlCounterGetter(client, index) : new CounterGetter(client, index);
case INSERT:
return new Inserter(client, index);
return client.isCQL() ? new CqlInserter(client, index) : new Inserter(client, index);
case COUNTER_ADD:
return new CounterAdder(client, index);
return client.isCQL() ? new CqlCounterAdder(client, index) : new CounterAdder(client, index);
case RANGE_SLICE:
return new RangeSlicer(client, index);
return client.isCQL() ? new CqlRangeSlicer(client, index) : new RangeSlicer(client, index);
case INDEXED_RANGE_SLICE:
return new IndexedRangeSlicer(client, index);
return client.isCQL() ? new CqlIndexedRangeSlicer(client, index) : new IndexedRangeSlicer(client, index);
case MULTI_GET:
return new MultiGetter(client, index);
return client.isCQL() ? new CqlMultiGetter(client, index) : new MultiGetter(client, index);
}
throw new UnsupportedOperationException();

95
tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java Normal file
View File

@ -0,0 +1,95 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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 org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import static com.google.common.base.Charsets.UTF_8;
public class CqlCounterAdder extends Operation
{
public CqlCounterAdder(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
StringBuilder query = new StringBuilder(
"UPDATE Counter1 USING CONSISTENCY " + session.getConsistencyLevel().toString() + " SET ");
for (int i = 0; i < session.getColumnsPerKey(); i++)
{
if (i > 0)
query.append(",");
query.append('C').append(i).append("=C").append(i).append("+1");
}
String key = String.format("%0" + session.getTotalKeysLength() + "d", index);
query.append( " WHERE KEY=").append(getQuotedCqlBlob(key.getBytes(UTF_8)));
long start = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
client.execute_cql_query(ByteBuffer.wrap(query.toString().getBytes()), Compression.NONE);
success = true;
}
catch (Exception e)
{
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error incrementing key %s %s%n",
index,
session.getRetryTimes(),
key,
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
session.operations.getAndIncrement();
session.keys.getAndIncrement();
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
}

93
tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java Normal file
View File

@ -0,0 +1,93 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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 org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlResultType;
public class CqlCounterGetter extends Operation
{
public CqlCounterGetter(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
byte[] key = generateKey();
String hexKey = getQuotedCqlBlob(key);
StringBuilder query = new StringBuilder("SELECT FIRST ").append(session.getColumnsPerKey())
.append(" ''..'' FROM Counter1 USING CONSISTENCY ").append(session.getConsistencyLevel().toString())
.append(" WHERE KEY=").append(hexKey);
long start = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
CqlResult result = client.execute_cql_query(ByteBuffer.wrap(query.toString().getBytes()),
Compression.NONE);
assert result.type.equals(CqlResultType.ROWS) : "expected ROWS result type";
assert result.rows.size() == 0 : "expected exactly one row";
success = (result.rows.get(0).columns.size() != 0);
}
catch (Exception e)
{
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error reading counter key %s %s%n",
index,
session.getRetryTimes(),
new String(key),
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
session.operations.getAndIncrement();
session.keys.getAndIncrement();
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
}

136
tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java Normal file
View File

@ -0,0 +1,136 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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.List;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlResultType;
import org.apache.cassandra.thrift.CqlRow;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.utils.Hex.bytesToHex;;
public class CqlIndexedRangeSlicer extends Operation
{
private static List<ByteBuffer> values = null;
private static String clauseFragment = "KEY > '%s' LIMIT %d";
public CqlIndexedRangeSlicer(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
if (values == null)
values = generateValues();
String format = "%0" + session.getTotalKeysLength() + "d";
String startOffset = String.format(format, 0);
StringBuilder query = new StringBuilder("SELECT FIRST ").append(session.getColumnsPerKey())
.append(" ''..'' FROM Standard1 USING CONSISTENCY ").append(session.getConsistencyLevel())
.append(" WHERE C1 = ").append(getQuotedCqlBlob(values.get(1).array())).append(" AND ");
int expectedPerValue = session.getNumKeys() / values.size(), received = 0;
while (received < expectedPerValue)
{
long start = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
CqlResult results = null;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
ByteBuffer queryBytes = ByteBuffer.wrap(makeQuery(query, startOffset).getBytes());
results = client.execute_cql_query(queryBytes, Compression.NONE);
success = (results.rows.size() != 0);
}
catch (Exception e)
{
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error executing indexed range query with offset %s %s%n",
index,
session.getRetryTimes(),
startOffset,
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
received += results.rows.size();
// convert max key found back to an integer, and increment it
startOffset = String.format(format, (1 + getMaxKey(results.rows)));
session.operations.getAndIncrement();
session.keys.getAndAdd(results.rows.size());
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
}
private String makeQuery(StringBuilder base, String startOffset)
{
return base.toString() + String.format(clauseFragment, bytesToHex(startOffset.getBytes()), session.getKeysPerCall());
}
/**
* Get maximum key from CqlRow list
* @param rows list of the CqlRow objects
* @return maximum key value of the list
*/
private int getMaxKey(List<CqlRow> rows)
{
int maxKey = ByteBufferUtil.toInt(rows.get(0).key);
for (CqlRow row : rows)
{
int currentKey = ByteBufferUtil.toInt(row.key);
if (currentKey > maxKey)
maxKey = currentKey;
}
return maxKey;
}
}

100
tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java Normal file
View File

@ -0,0 +1,100 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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.List;
import org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
public class CqlInserter extends Operation
{
private static List<ByteBuffer> values;
public CqlInserter(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
if (values == null)
values = generateValues();
StringBuilder query = new StringBuilder("UPDATE Standard1 USING CONSISTENCY ")
.append(session.getConsistencyLevel().toString()).append(" SET ");
for (int i = 0; i < session.getColumnsPerKey(); i++)
{
if (i > 0)
query.append(',');
query.append('C').append(i).append('=');
query.append(getQuotedCqlBlob(values.get(i % values.size()).array()));
}
String key = String.format("%0" + session.getTotalKeysLength() + "d", index);
query.append(" WHERE KEY=").append(getQuotedCqlBlob(key));
long start = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
client.execute_cql_query(ByteBuffer.wrap(query.toString().getBytes()), Compression.NONE);
success = true;
}
catch (Exception e)
{
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error inserting key %s %s%n",
index,
session.getRetryTimes(),
key,
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
session.operations.getAndIncrement();
session.keys.getAndIncrement();
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
}

41
tools/stress/src/org/apache/cassandra/stress/operations/CqlMultiGetter.java Normal file
View File

@ -0,0 +1,41 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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 org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
public class CqlMultiGetter extends Operation
{
public CqlMultiGetter(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
throw new RuntimeException("Multiget is not implemented for CQL");
}
}

91
tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java Normal file
View File

@ -0,0 +1,91 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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 org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlResultType;
public class CqlRangeSlicer extends Operation
{
public CqlRangeSlicer(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
String key = String.format("%0" + session.getTotalKeysLength() + "d", index);
StringBuilder query = new StringBuilder("SELECT FIRST ").append(session.getColumnsPerKey())
.append(" ''..'' FROM Standard1 USING CONSISTENCY ").append(session.getConsistencyLevel().toString())
.append(" WHERE KEY > ").append(getQuotedCqlBlob(key));
long startTime = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
int rowCount = 0;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
CqlResult result = client.execute_cql_query(ByteBuffer.wrap(query.toString().getBytes()),
Compression.NONE);
rowCount = result.rows.size();
success = (rowCount != 0);
}
catch (Exception e)
{
System.err.println(e);
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error executing range slice with offset %s %s%n",
index,
session.getRetryTimes(),
key,
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
session.operations.getAndIncrement();
session.keys.getAndAdd(rowCount);
session.latency.getAndAdd(System.currentTimeMillis() - startTime);
}
}

104
tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java Normal file
View File

@ -0,0 +1,104 @@
package org.apache.cassandra.stress.operations;
/*
*
* 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 org.apache.cassandra.db.ColumnFamilyType;
import org.apache.cassandra.stress.Session;
import org.apache.cassandra.stress.util.Operation;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Compression;
import org.apache.cassandra.thrift.CqlResult;
import org.apache.cassandra.thrift.CqlResultType;
public class CqlReader extends Operation
{
public CqlReader(Session client, int idx)
{
super(client, idx);
}
public void run(Cassandra.Client client) throws IOException
{
if (session.getColumnFamilyType() == ColumnFamilyType.Super)
throw new RuntimeException("Super columns are not implemented for CQL");
StringBuilder query = new StringBuilder("SELECT ");
if (session.columnNames == null)
{
query.append("FIRST ").append(session.getColumnsPerKey()).append(" ''..''");
}
else
{
for (int i = 0; i < session.columnNames.size(); i++)
{
if (i > 0)
query.append(",");
query.append('\'').append(new String(session.columnNames.get(i).array())).append('\'');
}
}
byte[] key = generateKey();
query.append(" FROM Standard1 USING CONSISTENCY ").append(session.getConsistencyLevel().toString());
query.append(" WHERE KEY=").append(getQuotedCqlBlob(key));
long start = System.currentTimeMillis();
boolean success = false;
String exceptionMessage = null;
for (int t = 0; t < session.getRetryTimes(); t++)
{
if (success)
break;
try
{
CqlResult result = client.execute_cql_query(ByteBuffer.wrap(query.toString().getBytes()),
Compression.NONE);
success = (result.rows.get(0).columns.size() != 0);
}
catch (Exception e)
{
exceptionMessage = getExceptionMessage(e);
success = false;
}
}
if (!success)
{
error(String.format("Operation [%d] retried %d times - error reading key %s %s%n",
index,
session.getRetryTimes(),
new String(key),
(exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
}
session.operations.getAndIncrement();
session.keys.getAndIncrement();
session.latency.getAndAdd(System.currentTimeMillis() - start);
}
}

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

@ -35,6 +35,7 @@ import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.cassandra.utils.Hex;
public abstract class Operation
{
@ -224,4 +225,14 @@ public abstract class Operation
else
System.err.println(message);
}
protected String getQuotedCqlBlob(String term)
{
return getQuotedCqlBlob(term.getBytes());
}
protected String getQuotedCqlBlob(byte[] term)
{
return String.format("'%s'", Hex.bytesToHex(term));
}
}