From c0adddd077c20e57231b87ce9f681b3d8b1d3415 Mon Sep 17 00:00:00 2001 From: Eric Evans Date: Wed, 7 Dec 2011 16:36:19 +0000 Subject: [PATCH] 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 --- .../org/apache/cassandra/stress/Session.java | 10 ++ .../apache/cassandra/stress/StressAction.java | 14 +- .../stress/operations/CqlCounterAdder.java | 95 ++++++++++++ .../stress/operations/CqlCounterGetter.java | 93 ++++++++++++ .../operations/CqlIndexedRangeSlicer.java | 136 ++++++++++++++++++ .../stress/operations/CqlInserter.java | 100 +++++++++++++ .../stress/operations/CqlMultiGetter.java | 41 ++++++ .../stress/operations/CqlRangeSlicer.java | 91 ++++++++++++ .../stress/operations/CqlReader.java | 104 ++++++++++++++ .../cassandra/stress/util/Operation.java | 11 ++ 10 files changed, 688 insertions(+), 7 deletions(-) create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlMultiGetter.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java create mode 100644 tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java diff --git a/tools/stress/src/org/apache/cassandra/stress/Session.java b/tools/stress/src/org/apache/cassandra/stress/Session.java index 052025e991..5455e6761c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Session.java +++ b/tools/stress/src/org/apache/cassandra/stress/Session.java @@ -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 */ diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java index 571d498f17..8d6a0f58a5 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java @@ -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(); diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java new file mode 100644 index 0000000000..049bc1eac5 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java @@ -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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java new file mode 100644 index 0000000000..f78e177ea4 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java @@ -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); + } + +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java new file mode 100644 index 0000000000..ed4ee9f2c1 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java @@ -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 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 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; + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java new file mode 100644 index 0000000000..264f4b114a --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java @@ -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 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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlMultiGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlMultiGetter.java new file mode 100644 index 0000000000..9b516355bb --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlMultiGetter.java @@ -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"); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java new file mode 100644 index 0000000000..8dcecedfd2 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java @@ -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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java new file mode 100644 index 0000000000..6337c29430 --- /dev/null +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java @@ -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); + } +} diff --git a/tools/stress/src/org/apache/cassandra/stress/util/Operation.java b/tools/stress/src/org/apache/cassandra/stress/util/Operation.java index dd5128f5df..4e08909e54 100644 --- a/tools/stress/src/org/apache/cassandra/stress/util/Operation.java +++ b/tools/stress/src/org/apache/cassandra/stress/util/Operation.java @@ -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)); + } }