From a16c4fe0a5037ceb4b7b5d88efad6416a762e345 Mon Sep 17 00:00:00 2001 From: T Jake Luciani Date: Thu, 14 May 2015 12:12:50 -0400 Subject: [PATCH] Remove thrift schema creation when native driver is used Patch by tjake; reviewed by Aleksey Yeschenko for CASSANDRA-9374 --- CHANGES.txt | 1 + .../predefined/CqlCounterAdder.java | 4 +- .../stress/settings/SettingsSchema.java | 179 +++++++++++++++--- 3 files changed, 158 insertions(+), 26 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 6aa3059e04..17e70df133 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 2.2 + * Avoid thrift schema creation when native driver is used in stress tool (CASSANDRA-9374) * Populate TokenMetadata early during startup (CASSANDRA-9317) * Make Functions.declared thread-safe * Add client warnings to native protocol v4 (CASSANDRA-8930) 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 9bb0d87021..bb8135cb69 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 @@ -46,7 +46,7 @@ public class CqlCounterAdder extends CqlOperation @Override protected String buildQuery() { - StringBuilder query = new StringBuilder("UPDATE \"Counter3\" SET "); + StringBuilder query = new StringBuilder("UPDATE counter1 SET "); // TODO : increment distribution subset of columns for (int i = 0; i < settings.columns.maxColumnsPerKey; i++) @@ -54,7 +54,7 @@ public class CqlCounterAdder extends CqlOperation if (i > 0) query.append(","); - String name = settings.columns.namestrs.get(i); + String name = wrapInQuotes(settings.columns.namestrs.get(i)); query.append(name).append("=").append(name).append("+?"); } query.append(" WHERE KEY=?"); diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java index 54f27bc2ea..5c437a9cb2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsSchema.java @@ -22,9 +22,13 @@ package org.apache.cassandra.stress.settings; import java.io.Serializable; import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; import java.util.*; +import com.datastax.driver.core.exceptions.AlreadyExistsException; +import org.apache.cassandra.stress.util.JavaDriverClient; import org.apache.cassandra.thrift.*; +import org.apache.cassandra.thrift.ConsistencyLevel; import org.apache.cassandra.utils.ByteBufferUtil; public class SettingsSchema implements Serializable @@ -56,9 +60,159 @@ public class SettingsSchema implements Serializable public void createKeySpaces(StressSettings settings) { - createKeySpacesThrift(settings); + if (settings.mode.api != ConnectionAPI.JAVA_DRIVER_NATIVE) + { + createKeySpacesThrift(settings); + } + else + { + createKeySpacesNative(settings); + } } + /** + * Create Keyspace with Standard and Super/Counter column families + */ + public void createKeySpacesNative(StressSettings settings) + { + + JavaDriverClient client = settings.getJavaDriverClient(false); + + try + { + //Keyspace + client.execute(createKeyspaceStatementCQL3(), org.apache.cassandra.db.ConsistencyLevel.LOCAL_ONE); + + client.execute("USE \""+keyspace+"\"", org.apache.cassandra.db.ConsistencyLevel.LOCAL_ONE); + + //Add standard1 and counter1 + client.execute(createStandard1StatementCQL3(settings), org.apache.cassandra.db.ConsistencyLevel.LOCAL_ONE); + client.execute(createCounter1StatementCQL3(settings), org.apache.cassandra.db.ConsistencyLevel.LOCAL_ONE); + + System.out.println(String.format("Created keyspaces. Sleeping %ss for propagation.", settings.node.nodes.size())); + Thread.sleep(settings.node.nodes.size() * 1000L); // seconds + } + catch (AlreadyExistsException e) + { + //Ok. + } + catch (Exception e) + { + throw new RuntimeException("Encountered exception creating schema", e); + } + } + + String createKeyspaceStatementCQL3() + { + StringBuilder b = new StringBuilder(); + + //Create Keyspace + b.append("CREATE KEYSPACE IF NOT EXISTS \"") + .append(keyspace) + .append("\" WITH replication = {'class': '") + .append(replicationStrategy) + .append("'"); + + if (replicationStrategyOptions.isEmpty()) + { + b.append(", 'replication_factor': '1'}"); + } + else + { + for(Map.Entry entry : replicationStrategyOptions.entrySet()) + { + b.append(", '").append(entry.getKey()).append("' : '").append(entry.getValue()).append("'"); + } + + b.append("}"); + } + + b.append(" AND durable_writes = true;\n"); + + return b.toString(); + } + + String createStandard1StatementCQL3(StressSettings settings) + { + + StringBuilder b = new StringBuilder(); + + b.append("CREATE TABLE IF NOT EXISTS ") + .append("standard1 (key blob PRIMARY KEY "); + + try + { + for (ByteBuffer name : settings.columns.names) + b.append("\n, \"").append(ByteBufferUtil.string(name)).append("\" blob"); + } + catch (CharacterCodingException e) + { + throw new RuntimeException(e); + } + + //Compression + b.append(") WITH COMPACT STORAGE AND compression = {"); + if (compression != null) + b.append("'sstable_compression' : '").append(compression).append("'"); + + b.append("}"); + + //Compaction + if (compactionStrategy != null) + { + b.append(" AND compaction = { 'class' : '").append(compactionStrategy).append("'"); + + for (Map.Entry entry : compactionStrategyOptions.entrySet()) + b.append(", '").append(entry.getKey()).append("' : '").append(entry.getValue()).append("'"); + + b.append("}"); + } + + b.append(";\n"); + + return b.toString(); + } + + String createCounter1StatementCQL3(StressSettings settings) + { + + StringBuilder b = new StringBuilder(); + + b.append("CREATE TABLE IF NOT EXISTS ") + .append("counter1 (key blob PRIMARY KEY,"); + + try + { + for (ByteBuffer name : settings.columns.names) + b.append("\n, \"").append(ByteBufferUtil.string(name)).append("\" counter"); + } + catch (CharacterCodingException e) + { + throw new RuntimeException(e); + } + + //Compression + b.append(") WITH COMPACT STORAGE AND compression = {"); + if (compression != null) + b.append("'sstable_compression' : '").append(compression).append("'"); + + b.append("}"); + + //Compaction + if (compactionStrategy != null) + { + b.append(" AND compaction = { 'class' : '").append(compactionStrategy).append("'"); + + for (Map.Entry entry : compactionStrategyOptions.entrySet()) + b.append(", '").append(entry.getKey()).append("' : '").append(entry.getValue()).append("'"); + + b.append("}"); + } + + b.append(";\n"); + + return b.toString(); + } /** * Create Keyspace with Standard and Super/Counter column families @@ -113,16 +267,7 @@ public class SettingsSchema implements Serializable try { client.system_add_keyspace(ksdef); - - /* CQL3 counter cf */ - client.set_cql_version("3.0.0"); // just to create counter cf for cql3 - client.set_keyspace(keyspace); - client.execute_cql3_query(createCounterCFStatementForCQL3(settings), Compression.NONE, ConsistencyLevel.ONE); - - if (settings.mode.cqlVersion.isCql()) - client.set_cql_version(settings.mode.cqlVersion.connectVersion); - /* end */ System.out.println(String.format("Created keyspaces. Sleeping %ss for propagation.", settings.node.nodes.size())); Thread.sleep(settings.node.nodes.size() * 1000L); // seconds @@ -137,20 +282,6 @@ public class SettingsSchema implements Serializable } } - private ByteBuffer createCounterCFStatementForCQL3(StressSettings options) - { - StringBuilder counter3 = new StringBuilder("CREATE TABLE \"Counter3\" (KEY blob PRIMARY KEY, "); - - for (int i = 0; i < options.columns.maxColumnsPerKey; i++) - { - counter3.append("c").append(i).append(" counter"); - if (i != options.columns.maxColumnsPerKey - 1) - counter3.append(", "); - } - counter3.append(");"); - - return ByteBufferUtil.bytes(counter3.toString()); - } // Option Declarations