diff --git a/CHANGES.txt b/CHANGES.txt index 671be30550..21d6f9e745 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -18,6 +18,7 @@ * Remove deprecated code in Cassandra 1.x and 2.x (CASSANDRA-18959) * ClientRequestSize metrics should not treat CONTAINS restrictions as being equality-based (CASSANDRA-18896) Merged from 4.0: + * Fix cassandra-stress in simplenative mode with prepared statements (CASSANDRA-18744) * Fix filtering system ks sstables for relocation on startup (CASSANDRA-18963) * Remove completed coordinator sessions (CASSANDRA-18903) Merged from 3.0: diff --git a/test/unit/org/apache/cassandra/tools/cassandrastress/CassandrastressTest.java b/test/unit/org/apache/cassandra/tools/cassandrastress/CassandrastressTest.java index 8d0711a3d7..9569fbdb07 100644 --- a/test/unit/org/apache/cassandra/tools/cassandrastress/CassandrastressTest.java +++ b/test/unit/org/apache/cassandra/tools/cassandrastress/CassandrastressTest.java @@ -70,6 +70,11 @@ public class CassandrastressTest extends CQLTester String hostNameAndPort = String.format("localhost:%s", nativePort); invokeAndAssertCleanExit(baseArgs, "-node", hostNameAndPort); + + invokeAndAssertCleanExit(baseArgs, "-mode", "simplenative", "prepared"); + invokeAndAssertCleanExit(baseArgs, "-mode", "simplenative"); + invokeAndAssertCleanExit(baseArgs, "-mode"); + invokeAndAssertCleanExit(baseArgs, "-mode", "unprepared"); } void invokeAndAssertCleanExit(String[] baseArgs, String ... extraArgs) diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java index 2874b4e411..a1e90d0a6a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterAdder.java @@ -72,9 +72,9 @@ public class CqlCounterAdder extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key) + protected CqlRunOp buildRunOp(QueryExecutor queryExecutor, List params, ByteBuffer key) { - return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1); + return new CqlRunOpAlwaysSucceed(queryExecutor, params, key, 1); } public boolean isWrite() diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java index eb908b0138..5ba02b3100 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlCounterGetter.java @@ -52,9 +52,9 @@ public class CqlCounterGetter extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key) + protected CqlRunOp buildRunOp(QueryExecutor queryExecutor, List params, ByteBuffer key) { - return new CqlRunOpTestNonEmpty(client, query, queryId, params, key); + return new CqlRunOpTestNonEmpty(queryExecutor, params, key); } } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java index 3fda6c2c9d..255cf75086 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlInserter.java @@ -71,9 +71,9 @@ public class CqlInserter extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key) + protected CqlRunOp buildRunOp(QueryExecutor queryExecutor, List params, ByteBuffer key) { - return new CqlRunOpAlwaysSucceed(client, query, queryId, params, key, 1); + return new CqlRunOpAlwaysSucceed(queryExecutor, params, key, 1); } public boolean isWrite() diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java index c89a1d13fc..5d662b9eb8 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java @@ -48,7 +48,7 @@ public abstract class CqlOperation extends PredefinedOperation protected abstract List getQueryParameters(byte[] key); protected abstract String buildQuery(); - protected abstract CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key); + protected abstract CqlRunOp buildRunOp(QueryExecutor queryExecutor, List params, ByteBuffer key); public CqlOperation(Command type, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings) { @@ -57,46 +57,18 @@ public abstract class CqlOperation extends PredefinedOperation throw new IllegalStateException("Variable column counts are not implemented for CQL"); } - protected CqlRunOp run(final ClientWrapper client, final List queryParams, final ByteBuffer key) throws IOException + protected CqlRunOp run(final QueryExecutor queryExecutor, final List queryParams, final ByteBuffer key) throws IOException { - final CqlRunOp op; - if (settings.mode.style == ConnectionStyle.CQL_PREPARED) - { - final Object id; - Object idobj = getCqlCache(); - if (idobj == null) - { - id = client.createPreparedStatement(buildQuery()); - storeCqlCache(id); - } - else - { - id = idobj; - } - - op = buildRunOp(client, null, id, queryParams, key); - } - else - { - final String query; - Object qobj = getCqlCache(); - if (qobj == null) - storeCqlCache(query = buildQuery()); - else - query = qobj.toString(); - - op = buildRunOp(client, query, null, queryParams, key); - } - + final CqlRunOp op = buildRunOp(queryExecutor, queryParams, key); timeWithRetry(op); return op; } - protected void run(final ClientWrapper client) throws IOException + protected void run(final QueryExecutor queryExecutor) throws IOException { final byte[] key = getKey().array(); final List queryParams = getQueryParameters(key); - run(client, queryParams, ByteBuffer.wrap(key)); + run(queryExecutor, queryParams, ByteBuffer.wrap(key)); } // Classes to process Cql results @@ -107,9 +79,9 @@ public abstract class CqlOperation extends PredefinedOperation final int keyCount; - protected CqlRunOpAlwaysSucceed(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key, int keyCount) + protected CqlRunOpAlwaysSucceed(QueryExecutor queryExecutor, List params, ByteBuffer key, int keyCount) { - super(client, query, queryId, RowCountHandler.INSTANCE, params, key); + super(queryExecutor, RowCountHandler.INSTANCE, params, key); this.keyCount = keyCount; } @@ -136,9 +108,9 @@ public abstract class CqlOperation extends PredefinedOperation protected final class CqlRunOpTestNonEmpty extends CqlRunOp { - protected CqlRunOpTestNonEmpty(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key) + protected CqlRunOpTestNonEmpty(QueryExecutor queryExecutor, List params, ByteBuffer key) { - super(client, query, queryId, RowCountHandler.INSTANCE, params, key); + super(queryExecutor, RowCountHandler.INSTANCE, params, key); } @Override @@ -166,9 +138,9 @@ public abstract class CqlOperation extends PredefinedOperation final List> expect; // a null value for an item in expect means we just check the row is present - protected CqlRunOpMatchResults(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key, List> expect) + protected CqlRunOpMatchResults(QueryExecutor queryExecutor, List params, ByteBuffer key, List> expect) { - super(client, query, queryId, RowsHandler.INSTANCE, params, key); + super(queryExecutor, RowsHandler.INSTANCE, params, key); this.expect = expect; } @@ -202,19 +174,15 @@ public abstract class CqlOperation extends PredefinedOperation protected abstract class CqlRunOp implements RunOp { - final ClientWrapper client; - final String query; - final Object queryId; + final QueryExecutor queryExecutor; final List params; final ByteBuffer key; final ResultHandler handler; V result; - private CqlRunOp(ClientWrapper client, String query, Object queryId, ResultHandler handler, List params, ByteBuffer key) + private CqlRunOp(QueryExecutor queryExecutor, ResultHandler handler, List params, ByteBuffer key) { - this.client = client; - this.query = query; - this.queryId = queryId; + this.queryExecutor = queryExecutor; this.handler = handler; this.params = params; this.key = key; @@ -223,9 +191,7 @@ public abstract class CqlOperation extends PredefinedOperation @Override public boolean run() throws Exception { - return queryId != null - ? validate(result = client.execute(queryId, key, params, handler)) - : validate(result = client.execute(query, key, params, handler)); + return validate(result = queryExecutor.execute(key, params, handler)); } public abstract boolean validate(V result); @@ -239,93 +205,144 @@ public abstract class CqlOperation extends PredefinedOperation @Override public void run(SimpleClient client) throws IOException { - run(wrap(client)); + run(new SimpleClientQueryExecutor(client, buildQuery())); } @Override public void run(JavaDriverClient client) throws IOException { - run(wrap(client)); + run(new JavaDriverQueryExecutor(client, buildQuery())); } - public ClientWrapper wrap(JavaDriverClient client) + protected abstract class QueryExecutor { - return new JavaDriverWrapper(client); + private final String query; + private PS preparedStatement; + + public QueryExecutor(String query) + { + this.query = query; + } + + private boolean isPrepared() + { + return settings.mode.style == ConnectionStyle.CQL_PREPARED; + } + + abstract protected PS createPreparedStatement(String query); + + private PS getPreparedStatement() + { + if (preparedStatement == null) + { + preparedStatement = createPreparedStatement(this.query); + } + return preparedStatement; + } + + /** + * Constructs a CQL query string by replacing instances of the character + * '?', with the corresponding parameter. + * + * @param query base query string to format + * @param parms sequence of string query parameters + * @return formatted CQL query string + */ + private String formatCqlQuery(String query, List parms) + { + int marker, position = 0; + StringBuilder result = new StringBuilder(); + + if (-1 == (marker = query.indexOf('?')) || parms.size() == 0) + return query; + + for (Object parm : parms) + { + result.append(query.substring(position, marker)); + + if (parm instanceof ByteBuffer) + result.append(getUnQuotedCqlBlob((ByteBuffer) parm)); + else if (parm instanceof Long) + result.append(parm); + else throw new AssertionError(); + + position = marker + 1; + if (-1 == (marker = query.indexOf('?', position + 1))) + break; + } + + if (position < query.length()) + result.append(query.substring(position)); + + return result.toString(); + } + + V execute(ByteBuffer key, List queryParams, ResultHandler handler) + { + return isPrepared() + ? execute(getPreparedStatement(), key, queryParams, handler) + : execute(formatCqlQuery(this.query, queryParams), key, handler); + } + + abstract V execute(PS preparedStatement, ByteBuffer key, List queryParams, ResultHandler handler); + abstract V execute(String formattedQuery, ByteBuffer key, ResultHandler handler); } - public ClientWrapper wrap(SimpleClient client) - { - return new SimpleClientWrapper(client); - } - - protected interface ClientWrapper - { - Object createPreparedStatement(String cqlQuery); - V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler); - V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler); - } - - private final class JavaDriverWrapper implements ClientWrapper + private final class JavaDriverQueryExecutor extends QueryExecutor { final JavaDriverClient client; - private JavaDriverWrapper(JavaDriverClient client) + private JavaDriverQueryExecutor(JavaDriverClient client, String query) { + super(query); this.client = client; } - @Override - public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) + protected V execute(String formattedQuery, ByteBuffer key, ResultHandler handler) { - String formattedQuery = formatCqlQuery(query, queryParams); return handler.javaDriverHandler().apply(client.execute(formattedQuery, settings.command.consistencyLevel)); } - @Override - public V execute(Object preparedStatement, ByteBuffer key, List queryParams, ResultHandler handler) + protected V execute(PreparedStatement preparedStatement, ByteBuffer key, List queryParams, ResultHandler handler) { return handler.javaDriverHandler().apply( client.executePrepared( - (PreparedStatement) preparedStatement, + preparedStatement, queryParams, settings.command.consistencyLevel)); } - @Override - public Object createPreparedStatement(String cqlQuery) + public PreparedStatement createPreparedStatement(String cqlQuery) { return client.prepare(cqlQuery); } } - private final class SimpleClientWrapper implements ClientWrapper + private final class SimpleClientQueryExecutor extends QueryExecutor { final SimpleClient client; - private SimpleClientWrapper(SimpleClient client) + private SimpleClientQueryExecutor(SimpleClient client, String query) { + super(query); this.client = client; } - @Override - public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(String formattedQuery, ByteBuffer key, ResultHandler handler) { - String formattedQuery = formatCqlQuery(query, queryParams); return handler.simpleClientHandler().apply(client.execute(formattedQuery, settings.command.consistencyLevel)); } - @Override - public V execute(Object preparedStatement, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(ResultMessage.Prepared preparedStatement, ByteBuffer key, List queryParams, ResultHandler handler) { return handler.simpleClientHandler().apply( client.executePrepared( - (ResultMessage.Prepared) preparedStatement, + preparedStatement, toByteBufferParams(queryParams), settings.command.consistencyLevel)); } - @Override - public Object createPreparedStatement(String cqlQuery) + public ResultMessage.Prepared createPreparedStatement(String cqlQuery) { - return client.prepare(cqlQuery).statementId.bytes; + return client.prepare(cqlQuery); } } @@ -481,43 +498,6 @@ public abstract class CqlOperation extends PredefinedOperation return "0x" + ByteBufferUtil.bytesToHex(term); } - /** - * Constructs a CQL query string by replacing instances of the character - * '?', with the corresponding parameter. - * - * @param query base query string to format - * @param parms sequence of string query parameters - * @return formatted CQL query string - */ - private static String formatCqlQuery(String query, List parms) - { - int marker, position = 0; - StringBuilder result = new StringBuilder(); - - if (-1 == (marker = query.indexOf('?')) || parms.size() == 0) - return query; - - for (Object parm : parms) - { - result.append(query.substring(position, marker)); - - if (parm instanceof ByteBuffer) - result.append(getUnQuotedCqlBlob((ByteBuffer) parm)); - else if (parm instanceof Long) - result.append(parm); - else throw new AssertionError(); - - position = marker + 1; - if (-1 == (marker = query.indexOf('?', position + 1))) - break; - } - - if (position < query.length()) - result.append(query.substring(position)); - - return result.toString(); - } - private static List toByteBufferParams(List params) { List r = new ArrayList<>(); diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java index 61c6553612..8884919237 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlReader.java @@ -71,10 +71,10 @@ public class CqlReader extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, ByteBuffer key) + protected CqlRunOp buildRunOp(QueryExecutor queryExecutor, List params, ByteBuffer key) { List expectRow = getColumnValues(); - return new CqlRunOpMatchResults(client, query, queryId, params, key, Arrays.asList(expectRow)); + return new CqlRunOpMatchResults(queryExecutor, params, key, Arrays.asList(expectRow)); } } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java index 220b574fcf..bf969ad46b 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/PredefinedOperation.java @@ -36,7 +36,6 @@ public abstract class PredefinedOperation extends PartitionOperation public static final byte[] EMPTY_BYTE_ARRAY = {}; public final Command type; private final Distribution columnCount; - private Object cqlCache; public PredefinedOperation(Command type, Timer timer, PartitionGenerator generator, SeedManager seedManager, StressSettings settings) { @@ -50,16 +49,6 @@ public abstract class PredefinedOperation extends PartitionOperation return new DataSpec(generator, seedManager, new DistributionFixed(1), rowPopulationCount, 1); } - - public Object getCqlCache() - { - return cqlCache; - } - public void storeCqlCache(Object val) - { - cqlCache = val; - } - protected ByteBuffer getKey() { return (ByteBuffer) partitions.get(0).getPartitionKey(0);