diff --git a/CHANGES.txt b/CHANGES.txt index 5ce2f337b3..99ae52dd29 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -20,7 +20,17 @@ * Fix ClassCastException for compact table with composites (CASSANDRA-6738) * Fix potentially repairing with wrong nodes (CASSANDRA-6808) * Change caching option syntax (CASSANDRA-6745) + * Fix stress to do proper counter reads (CASSANDRA-6835) Merged from 2.0: + * Fix leaking validator FH in StreamWriter (CASSANDRA-6832) + * fix nodetool getsstables for blob PK (CASSANDRA-6803) + * Fix saving triggers to schema (CASSANDRA-6789) + * Fix trigger mutations when base mutation list is immutable (CASSANDRA-6790) + * Fix accounting in FileCacheService to allow re-using RAR (CASSANDRA-6838) + * Fix static counter columns (CASSANDRA-6827) + * Restore expiring->deleted (cell) compaction optimization (CASSANDRA-6844) + * Fix CompactionManager.needsCleanup (CASSANDRA-6845) + * Correctly compare BooleanType values other than 0 and 1 (CASSANDRA-6779) * Avoid race-prone second "scrub" of system keyspace (CASSANDRA-6797) * Pool CqlRecordWriter clients by inetaddress rather than Range (CASSANDRA-6665) @@ -1258,6 +1268,7 @@ Merged from 1.0: 1.1.1 + * add populate_io_cache_on_flush option (CASSANDRA-2635) * allow larger cache capacities than 2GB (CASSANDRA-4150) * add getsstables command to nodetool (CASSANDRA-4199) * apply parent CF compaction settings to secondary index CFs (CASSANDRA-4280) diff --git a/lib/cassandra-driver-internal-only-1.0.2.post.zip b/lib/cassandra-driver-internal-only-1.0.2.post.zip index 9f6af56b8c..7ccd5f79af 100644 Binary files a/lib/cassandra-driver-internal-only-1.0.2.post.zip and b/lib/cassandra-driver-internal-only-1.0.2.post.zip differ diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index ae03cde85d..8e9f987f94 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -419,6 +419,8 @@ def cf_prop_val_completer(ctxt, cass): return ["{'sstable_compression': '"] if this_opt == 'compaction': return ["{'class': '"] + if this_opt == 'caching': + return ["{'keys': '"] if any(this_opt == opt[0] for opt in CqlRuleSet.obsolete_cf_options): return ["''"] if this_opt in ('read_repair_chance', 'bloom_filter_fp_chance', @@ -472,9 +474,9 @@ def cf_prop_val_mapval_completer(ctxt, cass): return [Hint('')] elif opt == 'caching': if key == 'rows_per_partition': - return [Hint('ALL'), Hint('NONE'), Hint('#rows_per_partition')] + return ["'ALL'", "'NONE'", Hint('#rows_per_partition')] elif key == 'keys': - return [Hint('ALL'), Hint('NONE')] + return ["'ALL'", "'NONE'"] return () def cf_prop_val_mapender_completer(ctxt, cass): diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 34aa5f54e9..40632b2ceb 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1809,7 +1809,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public List getSSTablesForKey(String key) { - DecoratedKey dk = new DecoratedKey(partitioner.getToken(ByteBuffer.wrap(key.getBytes())), ByteBuffer.wrap(key.getBytes())); + DecoratedKey dk = partitioner.decorateKey(metadata.getKeyValidator().fromString(key)); ViewFragment view = markReferenced(dk); try { diff --git a/src/java/org/apache/cassandra/db/composites/AbstractComposite.java b/src/java/org/apache/cassandra/db/composites/AbstractComposite.java index fbff930721..97417676cd 100644 --- a/src/java/org/apache/cassandra/db/composites/AbstractComposite.java +++ b/src/java/org/apache/cassandra/db/composites/AbstractComposite.java @@ -22,6 +22,7 @@ import java.nio.ByteBuffer; import org.apache.cassandra.db.filter.ColumnSlice; import org.apache.cassandra.db.marshal.AbstractCompositeType; import org.apache.cassandra.db.marshal.CompositeType; +import org.apache.cassandra.utils.ByteBufferUtil; public abstract class AbstractComposite implements Composite { @@ -75,12 +76,12 @@ public abstract class AbstractComposite implements Composite // See org.apache.cassandra.db.marshal.CompositeType for details. ByteBuffer result = ByteBuffer.allocate(dataSize() + 3 * size() + (isStatic() ? 2 : 0)); if (isStatic()) - AbstractCompositeType.putShortLength(result, CompositeType.STATIC_MARKER); + ByteBufferUtil.writeShortLength(result, CompositeType.STATIC_MARKER); for (int i = 0; i < size(); i++) { ByteBuffer bb = get(i); - AbstractCompositeType.putShortLength(result, bb.remaining()); + ByteBufferUtil.writeShortLength(result, bb.remaining()); result.put(bb.duplicate()); result.put((byte)0); } diff --git a/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java b/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java index 236abc7046..8f3aec4849 100644 --- a/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java @@ -17,15 +17,16 @@ */ package org.apache.cassandra.db.marshal; -import org.apache.cassandra.serializers.TypeSerializer; -import org.apache.cassandra.serializers.BytesSerializer; -import org.apache.cassandra.serializers.MarshalException; - import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.serializers.BytesSerializer; +import org.apache.cassandra.serializers.MarshalException; +import org.apache.cassandra.utils.ByteBufferUtil; + /** * A class avoiding class duplication between CompositeType and * DynamicCompositeType. @@ -34,44 +35,6 @@ import java.util.List; */ public abstract class AbstractCompositeType extends AbstractType { - - // changes bb position - public static int getShortLength(ByteBuffer bb) - { - int length = (bb.get() & 0xFF) << 8; - return length | (bb.get() & 0xFF); - } - - // Doesn't change bb position - protected static int getShortLength(ByteBuffer bb, int position) - { - int length = (bb.get(position) & 0xFF) << 8; - return length | (bb.get(position + 1) & 0xFF); - } - - // changes bb position - public static void putShortLength(ByteBuffer bb, int length) - { - bb.put((byte) ((length >> 8) & 0xFF)); - bb.put((byte) (length & 0xFF)); - } - - // changes bb position - public static ByteBuffer getBytes(ByteBuffer bb, int length) - { - ByteBuffer copy = bb.duplicate(); - copy.limit(copy.position() + length); - bb.position(bb.position() + length); - return copy; - } - - // changes bb position - public static ByteBuffer getWithShortLength(ByteBuffer bb) - { - int length = getShortLength(bb); - return getBytes(bb, length); - } - public int compare(ByteBuffer o1, ByteBuffer o2) { if (o1 == null || !o1.hasRemaining()) @@ -95,8 +58,8 @@ public abstract class AbstractCompositeType extends AbstractType { AbstractType comparator = getComparator(i, bb1, bb2); - ByteBuffer value1 = getWithShortLength(bb1); - ByteBuffer value2 = getWithShortLength(bb2); + ByteBuffer value1 = ByteBufferUtil.readBytesWithShortLength(bb1); + ByteBuffer value2 = ByteBufferUtil.readBytesWithShortLength(bb2); int cmp = comparator.compareCollectionMembers(value1, value2, previous); if (cmp != 0) @@ -135,7 +98,7 @@ public abstract class AbstractCompositeType extends AbstractType while (bb.remaining() > 0) { getComparator(i++, bb); - l.add(getWithShortLength(bb)); + l.add(ByteBufferUtil.readBytesWithShortLength(bb)); bb.get(); // skip end-of-component } return l.toArray(new ByteBuffer[l.size()]); @@ -164,7 +127,7 @@ public abstract class AbstractCompositeType extends AbstractType while (bb.remaining() > 0) { AbstractType comparator = getComparator(i, bb); - ByteBuffer value = getWithShortLength(bb); + ByteBuffer value = ByteBufferUtil.readBytesWithShortLength(bb); list.add( new CompositeComponent(comparator,value) ); @@ -237,7 +200,7 @@ public abstract class AbstractCompositeType extends AbstractType sb.append(":"); AbstractType comparator = getAndAppendComparator(i, bb, sb); - ByteBuffer value = getWithShortLength(bb); + ByteBuffer value = ByteBufferUtil.readBytesWithShortLength(bb); sb.append(escape(comparator.getString(value))); @@ -290,7 +253,7 @@ public abstract class AbstractCompositeType extends AbstractType for (ByteBuffer component : components) { comparators.get(i).serializeComparator(bb); - putShortLength(bb, component.remaining()); + ByteBufferUtil.writeShortLength(bb, component.remaining()); bb.put(component); // it's ok to consume component as we won't use it anymore bb.put((byte)0); ++i; @@ -318,11 +281,11 @@ public abstract class AbstractCompositeType extends AbstractType if (bb.remaining() < 2) throw new MarshalException("Not enough bytes to read value size of component " + i); - int length = getShortLength(bb); + int length = ByteBufferUtil.readShortLength(bb); if (bb.remaining() < length) throw new MarshalException("Not enough bytes to read value of component " + i); - ByteBuffer value = getBytes(bb, length); + ByteBuffer value = ByteBufferUtil.readBytes(bb, length); comparator.validateCollectionMember(value, previous); diff --git a/src/java/org/apache/cassandra/db/marshal/CollectionType.java b/src/java/org/apache/cassandra/db/marshal/CollectionType.java index fe672e402d..02d01ff803 100644 --- a/src/java/org/apache/cassandra/db/marshal/CollectionType.java +++ b/src/java/org/apache/cassandra/db/marshal/CollectionType.java @@ -146,12 +146,6 @@ public abstract class CollectionType extends AbstractType return pack(buffers, elements, size); } - protected static int getUnsignedShort(ByteBuffer bb) - { - int length = (bb.get() & 0xFF) << 8; - return length | (bb.get() & 0xFF); - } - public CQL3Type asCQL3Type() { return new CQL3Type.Collection(this); diff --git a/src/java/org/apache/cassandra/db/marshal/CompositeType.java b/src/java/org/apache/cassandra/db/marshal/CompositeType.java index af1f3ebda5..5797af4b8e 100644 --- a/src/java/org/apache/cassandra/db/marshal/CompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/CompositeType.java @@ -89,11 +89,11 @@ public class CompositeType extends AbstractCompositeType if (bb.remaining() < 2) return false; - int header = getShortLength(bb, bb.position()); + int header = ByteBufferUtil.getShortLength(bb, bb.position()); if ((header & 0xFFFF) != STATIC_MARKER) return false; - getShortLength(bb); // Skip header + ByteBufferUtil.readShortLength(bb); // Skip header return true; } @@ -179,7 +179,7 @@ public class CompositeType extends AbstractCompositeType int i = 0; while (bb.remaining() > 0) { - l[i++] = getWithShortLength(bb); + l[i++] = ByteBufferUtil.readBytesWithShortLength(bb); bb.get(); // skip end-of-component } return i == l.length ? l : Arrays.copyOfRange(l, 0, i); @@ -193,7 +193,7 @@ public class CompositeType extends AbstractCompositeType int i = 0; while (bb.remaining() > 0) { - ByteBuffer c = getWithShortLength(bb); + ByteBuffer c = ByteBufferUtil.readBytesWithShortLength(bb); if (i == idx) return c; @@ -212,7 +212,7 @@ public class CompositeType extends AbstractCompositeType public static boolean isStaticName(ByteBuffer bb) { - return bb.remaining() >= 2 && (getShortLength(bb, bb.position()) & 0xFFFF) == STATIC_MARKER; + return bb.remaining() >= 2 && (ByteBufferUtil.getShortLength(bb, bb.position()) & 0xFFFF) == STATIC_MARKER; } @Override @@ -324,7 +324,7 @@ public class CompositeType extends AbstractCompositeType ByteBuffer out = ByteBuffer.allocate(totalLength); for (ByteBuffer bb : buffers) { - putShortLength(out, bb.remaining()); + ByteBufferUtil.writeShortLength(out, bb.remaining()); out.put(bb.duplicate()); out.put((byte) 0); } diff --git a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java index 7f30fde8e4..8311e7e4b6 100644 --- a/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java +++ b/src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java @@ -90,10 +90,10 @@ public class DynamicCompositeType extends AbstractCompositeType { try { - int header = getShortLength(bb); + int header = ByteBufferUtil.readShortLength(bb); if ((header & 0x8000) == 0) { - String name = ByteBufferUtil.string(getBytes(bb, header)); + String name = ByteBufferUtil.string(ByteBufferUtil.readBytes(bb, header)); return TypeParser.parse(name); } else @@ -152,10 +152,10 @@ public class DynamicCompositeType extends AbstractCompositeType { try { - int header = getShortLength(bb); + int header = ByteBufferUtil.readShortLength(bb); if ((header & 0x8000) == 0) { - String name = ByteBufferUtil.string(getBytes(bb, header)); + String name = ByteBufferUtil.string(ByteBufferUtil.readBytes(bb, header)); sb.append(name).append("@"); return TypeParser.parse(name); } @@ -189,13 +189,13 @@ public class DynamicCompositeType extends AbstractCompositeType AbstractType comparator = null; if (bb.remaining() < 2) throw new MarshalException("Not enough bytes to header of the comparator part of component " + i); - int header = getShortLength(bb); + int header = ByteBufferUtil.readShortLength(bb); if ((header & 0x8000) == 0) { if (bb.remaining() < header) throw new MarshalException("Not enough bytes to read comparator name of component " + i); - ByteBuffer value = getBytes(bb, header); + ByteBuffer value = ByteBufferUtil.readBytes(bb, header); String valueStr = null; try { @@ -325,7 +325,7 @@ public class DynamicCompositeType extends AbstractCompositeType header = 0x8000 | (((byte)comparatorName.charAt(0)) & 0xFF); else header = comparatorName.length(); - putShortLength(bb, header); + ByteBufferUtil.writeShortLength(bb, header); if (!isAlias) bb.put(ByteBufferUtil.bytes(comparatorName)); diff --git a/src/java/org/apache/cassandra/serializers/CollectionSerializer.java b/src/java/org/apache/cassandra/serializers/CollectionSerializer.java index 9d4e4a4355..83a391d3f7 100644 --- a/src/java/org/apache/cassandra/serializers/CollectionSerializer.java +++ b/src/java/org/apache/cassandra/serializers/CollectionSerializer.java @@ -48,10 +48,4 @@ public abstract class CollectionSerializer implements TypeSerializer size += 2 + bb.remaining(); return pack(buffers, elements, size); } - - protected static int getUnsignedShort(ByteBuffer bb) - { - int length = (bb.get() & 0xFF) << 8; - return length | (bb.get() & 0xFF); - } } diff --git a/src/java/org/apache/cassandra/serializers/ListSerializer.java b/src/java/org/apache/cassandra/serializers/ListSerializer.java index 02726180cd..59f25d2eb8 100644 --- a/src/java/org/apache/cassandra/serializers/ListSerializer.java +++ b/src/java/org/apache/cassandra/serializers/ListSerializer.java @@ -22,6 +22,8 @@ import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.utils.ByteBufferUtil; + public class ListSerializer extends CollectionSerializer> { // interning instances @@ -50,14 +52,11 @@ public class ListSerializer extends CollectionSerializer> try { ByteBuffer input = bytes.duplicate(); - int n = getUnsignedShort(input); + int n = ByteBufferUtil.readShortLength(input); List l = new ArrayList(n); for (int i = 0; i < n; i++) { - int s = getUnsignedShort(input); - byte[] data = new byte[s]; - input.get(data); - ByteBuffer databb = ByteBuffer.wrap(data); + ByteBuffer databb = ByteBufferUtil.readBytesWithShortLength(input); elements.validate(databb); l.add(elements.deserialize(databb)); } diff --git a/src/java/org/apache/cassandra/serializers/MapSerializer.java b/src/java/org/apache/cassandra/serializers/MapSerializer.java index f04de6da5e..f79d07f8f5 100644 --- a/src/java/org/apache/cassandra/serializers/MapSerializer.java +++ b/src/java/org/apache/cassandra/serializers/MapSerializer.java @@ -18,12 +18,13 @@ package org.apache.cassandra.serializers; -import org.apache.cassandra.utils.Pair; - import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.Pair; + public class MapSerializer extends CollectionSerializer> { // interning instances @@ -55,20 +56,14 @@ public class MapSerializer extends CollectionSerializer> try { ByteBuffer input = bytes.duplicate(); - int n = getUnsignedShort(input); + int n = ByteBufferUtil.readShortLength(input); Map m = new LinkedHashMap(n); for (int i = 0; i < n; i++) { - int sk = getUnsignedShort(input); - byte[] datak = new byte[sk]; - input.get(datak); - ByteBuffer kbb = ByteBuffer.wrap(datak); + ByteBuffer kbb = ByteBufferUtil.readBytesWithShortLength(input); keys.validate(kbb); - int sv = getUnsignedShort(input); - byte[] datav = new byte[sv]; - input.get(datav); - ByteBuffer vbb = ByteBuffer.wrap(datav); + ByteBuffer vbb = ByteBufferUtil.readBytesWithShortLength(input); values.validate(vbb); m.put(keys.deserialize(kbb), values.deserialize(vbb)); diff --git a/src/java/org/apache/cassandra/serializers/SetSerializer.java b/src/java/org/apache/cassandra/serializers/SetSerializer.java index d424a11eac..d6d7062156 100644 --- a/src/java/org/apache/cassandra/serializers/SetSerializer.java +++ b/src/java/org/apache/cassandra/serializers/SetSerializer.java @@ -22,6 +22,8 @@ import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.util.*; +import org.apache.cassandra.utils.ByteBufferUtil; + public class SetSerializer extends CollectionSerializer> { // interning instances @@ -50,14 +52,11 @@ public class SetSerializer extends CollectionSerializer> try { ByteBuffer input = bytes.duplicate(); - int n = getUnsignedShort(input); + int n = ByteBufferUtil.readShortLength(input); Set l = new LinkedHashSet(n); for (int i = 0; i < n; i++) { - int s = getUnsignedShort(input); - byte[] data = new byte[s]; - input.get(data); - ByteBuffer databb = ByteBuffer.wrap(data); + ByteBuffer databb = ByteBufferUtil.readBytesWithShortLength(input); elements.validate(databb); l.add(elements.deserialize(databb)); } diff --git a/src/java/org/apache/cassandra/streaming/StreamWriter.java b/src/java/org/apache/cassandra/streaming/StreamWriter.java index a84d2f4810..cb69d0b5f7 100644 --- a/src/java/org/apache/cassandra/streaming/StreamWriter.java +++ b/src/java/org/apache/cassandra/streaming/StreamWriter.java @@ -72,10 +72,9 @@ public class StreamWriter { long totalSize = totalSize(); RandomAccessReader file = sstable.openDataReader(); - ChecksumValidator validator = null; - if (new File(sstable.descriptor.filenameFor(Component.CRC)).exists()) - validator = DataIntegrityMetadata.checksumValidator(sstable.descriptor); - + ChecksumValidator validator = new File(sstable.descriptor.filenameFor(Component.CRC)).exists() + ? DataIntegrityMetadata.checksumValidator(sstable.descriptor) + : null; transferBuffer = validator == null ? new byte[DEFAULT_CHUNK_SIZE] : new byte[validator.chunkSize]; // setting up data compression stream @@ -115,6 +114,7 @@ public class StreamWriter { // no matter what happens close file FileUtils.closeQuietly(file); + FileUtils.closeQuietly(validator); } // release reference only when completed successfully diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 20abaee4bc..0d1b1416c5 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java @@ -558,4 +558,41 @@ public class ByteBufferUtil { return buf.capacity() > buf.remaining() ? ByteBuffer.wrap(getArray(buf)) : buf; } + + // Doesn't change bb position + public static int getShortLength(ByteBuffer bb, int position) + { + int length = (bb.get(position) & 0xFF) << 8; + return length | (bb.get(position + 1) & 0xFF); + } + + // changes bb position + public static int readShortLength(ByteBuffer bb) + { + int length = (bb.get() & 0xFF) << 8; + return length | (bb.get() & 0xFF); + } + + // changes bb position + public static void writeShortLength(ByteBuffer bb, int length) + { + bb.put((byte) ((length >> 8) & 0xFF)); + bb.put((byte) (length & 0xFF)); + } + + // changes bb position + public static ByteBuffer readBytes(ByteBuffer bb, int length) + { + ByteBuffer copy = bb.duplicate(); + copy.limit(copy.position() + length); + bb.position(bb.position() + length); + return copy; + } + + // changes bb position + public static ByteBuffer readBytesWithShortLength(ByteBuffer bb) + { + int length = readShortLength(bb); + return readBytes(bb, length); + } } diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java index 0a94cc0736..7b574e2690 100644 --- a/src/java/org/apache/cassandra/utils/FBUtilities.java +++ b/src/java/org/apache/cassandra/utils/FBUtilities.java @@ -370,7 +370,7 @@ public class FBUtilities in = FBUtilities.class.getClassLoader().getResourceAsStream("org/apache/cassandra/config/version.properties"); if (in == null) { - return "Unknown"; + return System.getProperty("cassandra.releaseVersion", "Unknown"); } Properties props = new Properties(); props.load(in); diff --git a/tools/stress/src/org/apache/cassandra/stress/Operation.java b/tools/stress/src/org/apache/cassandra/stress/Operation.java index 198f7fecb0..2ed6fed5c9 100644 --- a/tools/stress/src/org/apache/cassandra/stress/Operation.java +++ b/tools/stress/src/org/apache/cassandra/stress/Operation.java @@ -19,27 +19,16 @@ package org.apache.cassandra.stress; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.EnumMap; import java.util.List; +import java.util.concurrent.ThreadLocalRandom; +import org.apache.cassandra.stress.generatedata.Distribution; import org.apache.cassandra.stress.generatedata.KeyGen; import org.apache.cassandra.stress.generatedata.RowGen; -import org.apache.cassandra.stress.operations.CqlCounterAdder; -import org.apache.cassandra.stress.operations.CqlCounterGetter; -import org.apache.cassandra.stress.operations.CqlIndexedRangeSlicer; -import org.apache.cassandra.stress.operations.CqlInserter; -import org.apache.cassandra.stress.operations.CqlMultiGetter; -import org.apache.cassandra.stress.operations.CqlRangeSlicer; -import org.apache.cassandra.stress.operations.CqlReader; -import org.apache.cassandra.stress.operations.ThriftCounterAdder; -import org.apache.cassandra.stress.operations.ThriftCounterGetter; -import org.apache.cassandra.stress.operations.ThriftIndexedRangeSlicer; -import org.apache.cassandra.stress.operations.ThriftInserter; -import org.apache.cassandra.stress.operations.ThriftMultiGetter; -import org.apache.cassandra.stress.operations.ThriftRangeSlicer; -import org.apache.cassandra.stress.operations.ThriftReader; import org.apache.cassandra.stress.settings.Command; import org.apache.cassandra.stress.settings.CqlVersion; import org.apache.cassandra.stress.settings.SettingsCommandMixed; @@ -49,6 +38,8 @@ import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.stress.util.Timer; import org.apache.cassandra.thrift.ColumnParent; import org.apache.cassandra.thrift.InvalidRequestException; +import org.apache.cassandra.thrift.SlicePredicate; +import org.apache.cassandra.thrift.SliceRange; import org.apache.cassandra.transport.SimpleClient; import org.apache.cassandra.utils.ByteBufferUtil; @@ -79,6 +70,7 @@ public abstract class Operation public final Command type; public final KeyGen keyGen; public final RowGen rowGen; + public final Distribution counteradd; public final List columnParents; public final StressMetrics metrics; public final SettingsCommandMixed.CommandSelector commandSelector; @@ -99,19 +91,12 @@ public abstract class Operation commandSelector = null; substates = null; } + counteradd = settings.command.add.get(); this.settings = settings; this.keyGen = settings.keys.newKeyGen(); this.rowGen = settings.columns.newRowGen(); this.metrics = metrics; - if (!settings.columns.useSuperColumns) - columnParents = Collections.singletonList(new ColumnParent(settings.schema.columnFamily)); - else - { - ColumnParent[] cp = new ColumnParent[settings.columns.superColumns]; - for (int i = 0 ; i < cp.length ; i++) - cp[i] = new ColumnParent("Super1").setSuper_column(ByteBufferUtil.bytes("S" + i)); - columnParents = Arrays.asList(cp); - } + this.columnParents = columnParents(type, settings); } private State(Command type, State copy) @@ -120,13 +105,29 @@ public abstract class Operation this.timer = copy.timer; this.rowGen = copy.rowGen; this.keyGen = copy.keyGen; - this.columnParents = copy.columnParents; + this.columnParents = columnParents(type, copy.settings); this.metrics = copy.metrics; this.settings = copy.settings; + this.counteradd = copy.counteradd; this.substates = null; this.commandSelector = null; } + private List columnParents(Command type, StressSettings settings) + { + if (!settings.columns.useSuperColumns) + return Collections.singletonList(new ColumnParent(type.table)); + else + { + ColumnParent[] cp = new ColumnParent[settings.columns.superColumns]; + for (int i = 0 ; i < cp.length ; i++) + cp[i] = new ColumnParent(type.supertable).setSuper_column(ByteBufferUtil.bytes("S" + i)); + return Arrays.asList(cp); + } + } + + + public boolean isCql3() { return settings.mode.cqlVersion == CqlVersion.CQL3; @@ -168,6 +169,53 @@ public abstract class Operation return state.rowGen.generate(index, key); } + private int sliceStart(int count) + { + if (count == state.settings.columns.maxColumnsPerKey) + return 0; + return 1 + ThreadLocalRandom.current().nextInt(state.settings.columns.maxColumnsPerKey - count); + } + + protected SlicePredicate slicePredicate() + { + final SlicePredicate predicate = new SlicePredicate(); + if (state.settings.columns.slice) + { + int count = state.rowGen.count(index); + int start = sliceStart(count); + predicate.setSlice_range(new SliceRange() + .setStart(state.settings.columns.names.get(start)) + .setFinish(new byte[] {}) + .setReversed(false) + .setCount(count) + ); + } + else + predicate.setColumn_names(randomNames()); + return predicate; + } + + protected List randomNames() + { + int count = state.rowGen.count(index); + List src = state.settings.columns.names; + if (count == src.size()) + return src; + ThreadLocalRandom rnd = ThreadLocalRandom.current(); + List r = new ArrayList<>(); + int c = 0, o = 0; + while (c < count && count + o < src.size()) + { + int leeway = src.size() - (count + o); + int spreadover = count - c; + o += Math.round(rnd.nextDouble() * (leeway / (double) spreadover)); + r.add(src.get(o + c++)); + } + while (c < count) + r.add(src.get(o + c++)); + return r; + } + /** * Run operation * @param client Cassandra Thrift client connection @@ -209,10 +257,11 @@ public abstract class Operation if (!success) { - error(String.format("Operation [%d] x%d key %s %s%n", + error(String.format("Operation [%d] x%d key %s (0x%s) %s%n", index, tries, run.key(), + ByteBufferUtil.bytesToHex(ByteBufferUtil.bytes(run.key())), (exceptionMessage == null) ? "Data returned was not validated" : "Error executing: " + exceptionMessage)); @@ -235,14 +284,4 @@ public abstract class Operation System.err.println(message); } - public static ByteBuffer getColumnNameBytes(int i) - { - return ByteBufferUtil.bytes("C" + i); - } - - public static String getColumnName(int i) - { - return "C" + i; - } - } diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java index 94824ec0db..e7cdd0bb79 100644 --- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java +++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java @@ -87,7 +87,7 @@ public class StressAction implements Runnable warmup(subtype, command); return; case MULTI: - int keysAtOnce = ((SettingsCommandMulti) command).keysAtOnce; + int keysAtOnce = command.keysAtOnce; iterations = Math.min(50000, (int) Math.ceil(500000d / keysAtOnce)); break; default: @@ -298,6 +298,8 @@ public class StressAction implements Runnable case SIMPLE_NATIVE: op.run(sclient); break; + case THRIFT: + case THRIFT_SMART: default: op.run(tclient); } diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java index cb0dc1c742..9c6ca43ee7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java +++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java @@ -46,6 +46,7 @@ public abstract class RowGen // these byte[] may be re-used abstract List getColumns(long operationIndex); + abstract public int count(long operationIndex); abstract public boolean isDeterministic(); diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java index eecbc7ee0d..fffad2f05a 100644 --- a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java +++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGenDistributedSize.java @@ -51,6 +51,8 @@ public class RowGenDistributedSize extends RowGen this.sizeDistribution = sizeDistribution; ret = new ByteBuffer[(int) countDistribution.maxValue()]; sizes = new int[ret.length]; + // TODO: should keep it deterministic in event that count distribution is not, but size and dataGen are, so that + // we simply need to generate the correct selection of columns this.isDeterministic = dataGen.isDeterministic() && countDistribution.maxValue() == countDistribution.minValue() && sizeDistribution.minValue() == sizeDistribution.maxValue(); } @@ -100,6 +102,11 @@ public class RowGenDistributedSize extends RowGen return Arrays.asList(ret).subList(0, count); } + public int count(long operationIndex) + { + return (int) countDistribution.next(); + } + @Override public boolean isDeterministic() { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java index 910b6edcde..4e333a4f6c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterAdder.java @@ -22,7 +22,7 @@ package org.apache.cassandra.stress.operations; import java.nio.ByteBuffer; -import java.util.Collections; +import java.util.ArrayList; import java.util.List; public class CqlCounterAdder extends CqlOperation @@ -35,11 +35,7 @@ public class CqlCounterAdder extends CqlOperation @Override protected String buildQuery() { - String counterCF = "Counter3"; - - StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(counterCF)); - - query.append(" SET "); + StringBuilder query = new StringBuilder("UPDATE \"Counter3\" SET "); // TODO : increment distribution subset of columns for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++) @@ -47,20 +43,25 @@ public class CqlCounterAdder extends CqlOperation if (i > 0) query.append(","); - query.append('C').append(i).append("=C").append(i).append("+1"); + String name = state.settings.columns.namestrs.get(i); + query.append(name).append("=").append(name).append("+?"); } query.append(" WHERE KEY=?"); return query.toString(); } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { - return Collections.singletonList(ByteBuffer.wrap(key)); + final List list = new ArrayList<>(); + for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++) + list.add(state.counteradd.next()); + list.add(ByteBuffer.wrap(key)); + return list; } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { return new CqlRunOpAlwaysSucceed(client, query, queryId, params, keyid, key, 1); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java index 6186667528..28e6a1293d 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlCounterGetter.java @@ -34,19 +34,24 @@ public class CqlCounterGetter extends CqlOperation } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { - return Collections.singletonList(ByteBuffer.wrap(key)); + return Collections.singletonList(ByteBuffer.wrap(key)); } @Override protected String buildQuery() { - return "SELECT * FROM \"Counter3\" USING CONSISTENCY " + state.settings.command.consistencyLevel + " WHERE KEY=?"; + StringBuilder query = new StringBuilder("SELECT *"); + + // TODO: obey slice/noslice option (instead of always slicing) + query.append(" FROM ").append(wrapInQuotes(state.type.table)); + + return query.append(" WHERE KEY=?").toString(); } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { return new CqlRunOpTestNonEmpty(client, query, queryId, params, keyid, key); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java index 25af04a4c6..6febe26903 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java @@ -26,7 +26,6 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; -import org.apache.cassandra.stress.settings.SettingsCommandMulti; import org.apache.cassandra.utils.FBUtilities; public class CqlIndexedRangeSlicer extends CqlOperation @@ -40,7 +39,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { throw new UnsupportedOperationException(); } @@ -48,10 +47,11 @@ public class CqlIndexedRangeSlicer extends CqlOperation @Override protected String buildQuery() { - StringBuilder query = new StringBuilder("SELECT * FROM \"Standard1\""); - final String columnName = getColumnName(1); - query.append(" WHERE ").append(columnName).append("=?") - .append(" AND KEY > ? LIMIT ").append(((SettingsCommandMulti)state.settings.command).keysAtOnce); + final String indexColumn = (state.settings.columns.namestrs.get(1)); + StringBuilder query = new StringBuilder("SELECT * FROM "); + query.append(wrapInQuotes(state.type.table)); + query.append(" WHERE ").append(indexColumn).append("=?") + .append(" AND KEY > ? LIMIT ").append(state.settings.command.keysAtOnce); return query.toString(); } @@ -65,7 +65,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation int rowCount; do { - List params = Arrays.asList(value, ByteBuffer.wrap(minKey)); + List params = Arrays.asList(value, ByteBuffer.wrap(minKey)); CqlRunOp op = run(client, params, value, new String(value.array())); byte[][] keys = op.result; rowCount = keys.length; @@ -77,7 +77,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation private final class IndexedRangeSliceRunOp extends CqlRunOpFetchKeys { - protected IndexedRangeSliceRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected IndexedRangeSliceRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { super(client, query, queryId, params, keyid, key); } @@ -90,7 +90,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { return new IndexedRangeSliceRunOp(client, query, queryId, params, keyid, key); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java index 1f8987d920..45e375b32c 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java @@ -33,12 +33,14 @@ public class CqlInserter extends CqlOperation public CqlInserter(State state, long idx) { super(state, idx); + if (state.settings.columns.useTimeUUIDComparator) + throw new IllegalStateException("Cannot use TimeUUID Comparator with CQL"); } @Override protected String buildQuery() { - StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(state.settings.schema.columnFamily)); + StringBuilder query = new StringBuilder("UPDATE ").append(wrapInQuotes(state.type.table)); query.append(" SET "); @@ -66,9 +68,9 @@ public class CqlInserter extends CqlOperation } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { - final ArrayList queryParams = new ArrayList<>(); + final ArrayList queryParams = new ArrayList<>(); final List values = generateColumnValues(ByteBuffer.wrap(key)); queryParams.addAll(values); queryParams.add(ByteBuffer.wrap(key)); @@ -76,7 +78,7 @@ public class CqlInserter extends CqlOperation } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { return new CqlRunOpAlwaysSucceed(client, query, queryId, params, keyid, key, 1); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java index 8674cc03e8..6da145ebcc 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java @@ -20,6 +20,7 @@ package org.apache.cassandra.stress.operations; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -44,9 +45,9 @@ import org.apache.thrift.TException; public abstract class CqlOperation extends Operation { - protected abstract List getQueryParameters(byte[] key); + protected abstract List getQueryParameters(byte[] key); protected abstract String buildQuery(); - protected abstract CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key); + protected abstract CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key); public CqlOperation(State state, long idx) { @@ -55,9 +56,11 @@ public abstract class CqlOperation extends Operation throw new IllegalStateException("Super columns are not implemented for CQL"); if (state.settings.columns.variableColumnCount) throw new IllegalStateException("Variable column counts are not implemented for CQL"); + if (state.settings.columns.useTimeUUIDComparator) + throw new IllegalStateException("Cannot use TimeUUID Comparator with CQL"); } - protected CqlRunOp run(final ClientWrapper client, final List queryParams, final ByteBuffer key, final String keyid) throws IOException + protected CqlRunOp run(final ClientWrapper client, final List queryParams, final ByteBuffer key, final String keyid) throws IOException { final CqlRunOp op; if (state.settings.mode.style == ConnectionStyle.CQL_PREPARED) @@ -99,7 +102,7 @@ public abstract class CqlOperation extends Operation protected void run(final ClientWrapper client) throws IOException { final byte[] key = getKey().array(); - final List queryParams = getQueryParameters(key); + final List queryParams = getQueryParameters(key); run(client, queryParams, ByteBuffer.wrap(key), new String(key)); } @@ -111,7 +114,7 @@ public abstract class CqlOperation extends Operation final int keyCount; - protected CqlRunOpAlwaysSucceed(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key, int keyCount) + protected CqlRunOpAlwaysSucceed(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key, int keyCount) { super(client, query, queryId, RowCountHandler.INSTANCE, params, id, key); this.keyCount = keyCount; @@ -134,7 +137,7 @@ public abstract class CqlOperation extends Operation protected final class CqlRunOpTestNonEmpty extends CqlRunOp { - protected CqlRunOpTestNonEmpty(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key) + protected CqlRunOpTestNonEmpty(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key) { super(client, query, queryId, RowCountHandler.INSTANCE, params, id, key); } @@ -156,7 +159,7 @@ public abstract class CqlOperation extends Operation protected abstract class CqlRunOpFetchKeys extends CqlRunOp { - protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key) + protected CqlRunOpFetchKeys(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key) { super(client, query, queryId, KeysHandler.INSTANCE, params, id, key); } @@ -175,7 +178,7 @@ public abstract class CqlOperation extends Operation 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, String id, ByteBuffer key, List> expect) + protected CqlRunOpMatchResults(ClientWrapper client, String query, Object queryId, List params, String id, ByteBuffer key, List> expect) { super(client, query, queryId, RowsHandler.INSTANCE, params, id, key); this.expect = expect; @@ -209,13 +212,13 @@ public abstract class CqlOperation extends Operation final ClientWrapper client; final String query; final Object queryId; - final List params; + final List params; final String id; final ByteBuffer key; final ResultHandler handler; V result; - private CqlRunOp(ClientWrapper client, String query, Object queryId, ResultHandler handler, List params, String id, ByteBuffer key) + private CqlRunOp(ClientWrapper client, String query, Object queryId, ResultHandler handler, List params, String id, ByteBuffer key) { this.client = client; this.query = query; @@ -284,8 +287,8 @@ public abstract class CqlOperation extends Operation protected interface ClientWrapper { Object createPreparedStatement(String cqlQuery) throws TException; - V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) throws TException; - V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) throws TException; + V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) throws TException; + V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) throws TException; } private final class JavaDriverWrapper implements ClientWrapper @@ -297,14 +300,14 @@ public abstract class CqlOperation extends Operation } @Override - public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) { String formattedQuery = formatCqlQuery(query, queryParams); return handler.javaDriverHandler().apply(client.execute(formattedQuery, ThriftConversion.fromThrift(state.settings.command.consistencyLevel))); } @Override - public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) { return handler.javaDriverHandler().apply( client.executePrepared( @@ -329,19 +332,19 @@ public abstract class CqlOperation extends Operation } @Override - public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) { String formattedQuery = formatCqlQuery(query, queryParams); return handler.thriftHandler().apply(client.execute(formattedQuery, ThriftConversion.fromThrift(state.settings.command.consistencyLevel))); } @Override - public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) + public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) { return handler.thriftHandler().apply( client.executePrepared( (byte[]) preparedStatementId, - queryParams, + toByteBufferParams(queryParams), ThriftConversion.fromThrift(state.settings.command.consistencyLevel))); } @@ -362,7 +365,7 @@ public abstract class CqlOperation extends Operation } @Override - public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) throws TException + public V execute(String query, ByteBuffer key, List queryParams, ResultHandler handler) throws TException { String formattedQuery = formatCqlQuery(query, queryParams); return handler.simpleNativeHandler().apply( @@ -371,11 +374,11 @@ public abstract class CqlOperation extends Operation } @Override - public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) throws TException + public V execute(Object preparedStatementId, ByteBuffer key, List queryParams, ResultHandler handler) throws TException { Integer id = (Integer) preparedStatementId; return handler.simpleNativeHandler().apply( - client.execute_prepared_cql3_query(id, key, queryParams, state.settings.command.consistencyLevel) + client.execute_prepared_cql3_query(id, key, toByteBufferParams(queryParams), state.settings.command.consistencyLevel) ); } @@ -608,7 +611,7 @@ public abstract class CqlOperation extends Operation * @param parms sequence of string query parameters * @return formatted CQL query string */ - private static String formatCqlQuery(String query, List parms) + private static String formatCqlQuery(String query, List parms) { int marker, position = 0; StringBuilder result = new StringBuilder(); @@ -616,10 +619,15 @@ public abstract class CqlOperation extends Operation if (-1 == (marker = query.indexOf('?')) || parms.size() == 0) return query; - for (ByteBuffer parm : parms) + for (Object parm : parms) { result.append(query.substring(position, marker)); - result.append(getUnQuotedCqlBlob(parm)); + + if (parm instanceof ByteBuffer) + result.append(getUnQuotedCqlBlob((ByteBuffer) parm)); + else if (parm instanceof Long) + result.append(parm.toString()); + else throw new AssertionError(); position = marker + 1; if (-1 == (marker = query.indexOf('?', position + 1))) @@ -632,6 +640,20 @@ public abstract class CqlOperation extends Operation return result.toString(); } + private static List toByteBufferParams(List params) + { + List r = new ArrayList<>(); + for (Object param : params) + { + if (param instanceof ByteBuffer) + r.add((ByteBuffer) param); + else if (param instanceof Long) + r.add(ByteBufferUtil.bytes((Long) param)); + else throw new AssertionError(); + } + return r; + } + protected String wrapInQuotes(String string) { return "\"" + string + "\""; diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java index cce47fc603..8b6d6fc239 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlRangeSlicer.java @@ -33,19 +33,19 @@ public class CqlRangeSlicer extends CqlOperation } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { - return Collections.singletonList(ByteBuffer.wrap(key)); + return Collections.singletonList(ByteBuffer.wrap(key)); } @Override protected String buildQuery() { - return "SELECT FIRST " + state.settings.columns.maxColumnsPerKey + " ''..'' FROM " + state.settings.schema.columnFamily + " WHERE KEY > ?"; + return "SELECT FIRST " + state.settings.columns.maxColumnsPerKey + " ''..'' FROM " + wrapInQuotes(state.type.table) + " WHERE KEY > ?"; } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { return new CqlRunOpTestNonEmpty(client, query, queryId, params, keyid, key); } diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java index aa949d427a..c9d88703b7 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java @@ -40,7 +40,7 @@ public class CqlReader extends CqlOperation { StringBuilder query = new StringBuilder("SELECT "); - if (state.settings.columns.names == null) + if (state.settings.columns.slice) { query.append("*"); } @@ -54,28 +54,28 @@ public class CqlReader extends CqlOperation } } - query.append(" FROM ").append(wrapInQuotes(state.settings.schema.columnFamily)); + query.append(" FROM ").append(wrapInQuotes(state.type.table)); query.append(" WHERE KEY=?"); return query.toString(); } @Override - protected List getQueryParameters(byte[] key) + protected List getQueryParameters(byte[] key) { if (state.settings.columns.names != null) { - final List queryParams = new ArrayList<>(); + final List queryParams = new ArrayList<>(); for (ByteBuffer name : state.settings.columns.names) queryParams.add(name); queryParams.add(ByteBuffer.wrap(key)); return queryParams; } - return Collections.singletonList(ByteBuffer.wrap(key)); + return Collections.singletonList(ByteBuffer.wrap(key)); } @Override - protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) + protected CqlRunOp buildRunOp(ClientWrapper client, String query, Object queryId, List params, String keyid, ByteBuffer key) { List expectRow = state.rowGen.isDeterministic() ? generateColumnValues(key) : null; return new CqlRunOpMatchResults(client, query, queryId, params, keyid, key, Arrays.asList(expectRow)); diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java index 26695a6fee..9bfe4406a3 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java @@ -23,6 +23,8 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Random; +import java.util.concurrent.ThreadLocalRandom; import org.apache.cassandra.stress.Operation; import org.apache.cassandra.stress.util.ThriftClient; @@ -33,15 +35,13 @@ public class ThriftCounterAdder extends Operation public ThriftCounterAdder(State state, long index) { super(state, index); - if (state.settings.columns.variableColumnCount) - throw new IllegalStateException("Variable column counts not supported for counters"); } public void run(final ThriftClient client) throws IOException { List columns = new ArrayList<>(); - for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++) - columns.add(new CounterColumn(getColumnNameBytes(i), 1L)); + for (ByteBuffer name : randomNames()) + columns.add(new CounterColumn(name, state.counteradd.next())); Map> row; if (state.settings.columns.useSuperColumns) @@ -53,7 +53,7 @@ public class ThriftCounterAdder extends Operation ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setCounter_super_column(csc); mutations.add(new Mutation().setColumn_or_supercolumn(cosc)); } - row = Collections.singletonMap("SuperCounter1", mutations); + row = Collections.singletonMap(state.type.supertable, mutations); } else { @@ -63,7 +63,7 @@ public class ThriftCounterAdder extends Operation ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setCounter_column(c); mutations.add(new Mutation().setColumn_or_supercolumn(cosc)); } - row = Collections.singletonMap("Counter1", mutations); + row = Collections.singletonMap(state.type.table, mutations); } final ByteBuffer key = getKey(); diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterGetter.java index 8567edd01d..6e36a28e48 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterGetter.java @@ -19,6 +19,7 @@ package org.apache.cassandra.stress.operations; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.List; import org.apache.cassandra.stress.Operation; import org.apache.cassandra.stress.util.ThriftClient; @@ -31,20 +32,11 @@ public class ThriftCounterGetter extends Operation public ThriftCounterGetter(State state, long index) { super(state, index); - if (state.settings.columns.variableColumnCount) - throw new IllegalStateException("Variable column counts not supported for counters"); } public void run(final ThriftClient client) throws IOException { - SliceRange sliceRange = new SliceRange(); - // start/finish - sliceRange.setStart(new byte[] {}).setFinish(new byte[] {}); - // reversed/count - sliceRange.setReversed(false).setCount(state.settings.columns.maxColumnsPerKey); - // initialize SlicePredicate with existing SliceRange - final SlicePredicate predicate = new SlicePredicate().setSlice_range(sliceRange); - + final SlicePredicate predicate = slicePredicate(); final ByteBuffer key = getKey(); for (final ColumnParent parent : state.columnParents) { @@ -54,7 +46,8 @@ public class ThriftCounterGetter extends Operation @Override public boolean run() throws Exception { - return client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel).size() != 0; + List r = client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel); + return r != null && r.size() > 0; } @Override diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java index 6eab209c78..8c8ec31c18 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java @@ -23,7 +23,6 @@ import java.util.Arrays; import java.util.List; import org.apache.cassandra.stress.Operation; -import org.apache.cassandra.stress.settings.SettingsCommandMulti; import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.thrift.*; import org.apache.cassandra.utils.ByteBufferUtil; @@ -52,7 +51,7 @@ public class ThriftIndexedRangeSlicer extends Operation final List columns = generateColumnValues(getKey()); final ColumnParent parent = state.columnParents.get(0); - final ByteBuffer columnName = getColumnNameBytes(1); + final ByteBuffer columnName = state.settings.columns.names.get(1); final ByteBuffer value = columns.get(1); // only C1 column is indexed IndexExpression expression = new IndexExpression(columnName, IndexOperator.EQ, value); @@ -64,7 +63,7 @@ public class ThriftIndexedRangeSlicer extends Operation final boolean first = minKey.length == 0; final IndexClause clause = new IndexClause(Arrays.asList(expression), ByteBuffer.wrap(minKey), - ((SettingsCommandMulti) state.settings.command).keysAtOnce); + state.settings.command.keysAtOnce); timeWithRetry(new RunOp() { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java index b107f261ed..7077a95049 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java @@ -53,7 +53,7 @@ public final class ThriftInserter extends Operation ColumnOrSuperColumn column = new ColumnOrSuperColumn().setColumn(c); mutations.add(new Mutation().setColumn_or_supercolumn(column)); } - row = Collections.singletonMap(state.settings.schema.columnFamily, mutations); + row = Collections.singletonMap(state.type.table, mutations); } else { @@ -64,7 +64,7 @@ public final class ThriftInserter extends Operation final ColumnOrSuperColumn cosc = new ColumnOrSuperColumn().setSuper_column(s); mutations.add(new Mutation().setColumn_or_supercolumn(cosc)); } - row = Collections.singletonMap("Super1", mutations); + row = Collections.singletonMap(state.settings.command.type.supertable, mutations); } final Map>> record = Collections.singletonMap(key, row); @@ -104,7 +104,7 @@ public final class ThriftInserter extends Operation // TODO : consider randomly allocating column names in case where have fewer than max columns // but need to think about implications for indexes / indexed range slicer / other knock on effects for (int i = 0 ; i < values.size() ; i++) - columns.add(new Column(getColumnNameBytes(i))); + columns.add(new Column(state.settings.columns.names.get(i))); for (int i = 0 ; i < values.size() ; i++) columns.get(i) diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftMultiGetter.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftMultiGetter.java index 01c7325000..d8e0117e98 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftMultiGetter.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftMultiGetter.java @@ -22,7 +22,6 @@ import java.nio.ByteBuffer; import java.util.List; import org.apache.cassandra.stress.Operation; -import org.apache.cassandra.stress.settings.SettingsCommandMulti; import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.thrift.ColumnParent; import org.apache.cassandra.thrift.SlicePredicate; @@ -50,7 +49,7 @@ public final class ThriftMultiGetter extends Operation ) ); - final List keys = getKeys(((SettingsCommandMulti) state.settings.command).keysAtOnce); + final List keys = getKeys(state.settings.command.keysAtOnce); for (final ColumnParent parent : state.columnParents) { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftRangeSlicer.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftRangeSlicer.java index ce6c8cd4ca..021c4e842e 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftRangeSlicer.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftRangeSlicer.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import org.apache.cassandra.stress.Operation; -import org.apache.cassandra.stress.settings.SettingsCommandMulti; import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.thrift.ColumnParent; import org.apache.cassandra.thrift.KeyRange; @@ -55,7 +54,7 @@ public final class ThriftRangeSlicer extends Operation new KeyRange(state.settings.columns.maxColumnsPerKey) .setStart_key(start) .setEnd_key(ByteBufferUtil.EMPTY_BYTE_BUFFER) - .setCount(((SettingsCommandMulti)state.settings.command).keysAtOnce); + .setCount(state.settings.command.keysAtOnce); for (final ColumnParent parent : state.columnParents) { diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java index c50843ff8e..dccf4696bc 100644 --- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java +++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java @@ -19,7 +19,6 @@ package org.apache.cassandra.stress.operations; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.ArrayList; import java.util.List; import org.apache.cassandra.stress.Operation; @@ -27,7 +26,6 @@ import org.apache.cassandra.stress.util.ThriftClient; import org.apache.cassandra.thrift.ColumnOrSuperColumn; import org.apache.cassandra.thrift.ColumnParent; import org.apache.cassandra.thrift.SlicePredicate; -import org.apache.cassandra.thrift.SliceRange; import org.apache.cassandra.thrift.SuperColumn; public final class ThriftReader extends Operation @@ -40,17 +38,7 @@ public final class ThriftReader extends Operation public void run(final ThriftClient client) throws IOException { - final SlicePredicate predicate = new SlicePredicate(); - if (state.settings.columns.names == null) - predicate.setSlice_range(new SliceRange() - .setStart(new byte[] {}) - .setFinish(new byte[] {}) - .setReversed(false) - .setCount(state.settings.columns.maxColumnsPerKey) - ); - else // see CASSANDRA-3064 about why this is useful - predicate.setColumn_names(state.settings.columns.names); - + final SlicePredicate predicate = slicePredicate(); final ByteBuffer key = getKey(); final List expect = state.rowGen.isDeterministic() ? generateColumnValues(key) : null; for (final ColumnParent parent : state.columnParents) @@ -63,6 +51,8 @@ public final class ThriftReader extends Operation List row = client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel); if (expect == null) return !row.isEmpty(); + if (row == null) + return false; if (!state.settings.columns.useSuperColumns) { if (row.size() != expect.size()) diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/Command.java b/tools/stress/src/org/apache/cassandra/stress/settings/Command.java index 60b65f7b8c..d0350ad972 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/Command.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/Command.java @@ -27,54 +27,46 @@ import java.util.Map; public enum Command { - READ(false, - SettingsCommand.helpPrinter("read"), + READ(false, "Standard1", "Super1", "Multiple concurrent reads - the cluster must first be populated by a write test", CommandCategory.BASIC ), - WRITE(true, - SettingsCommand.helpPrinter("write"), + WRITE(true, "Standard1", "Super1", "insert", "Multiple concurrent writes against the cluster", CommandCategory.BASIC ), - MIXED(true, - SettingsCommandMixed.helpPrinter(), + MIXED(true, null, null, "Interleaving of any basic commands, with configurable ratio and distribution - the cluster must first be populated by a write test", CommandCategory.MIXED ), - RANGESLICE(false, - SettingsCommandMulti.helpPrinter("range_slice"), + RANGESLICE(false, "Standard1", "Super1", "Range slice queries - the cluster must first be populated by a write test", CommandCategory.MULTI ), - IRANGESLICE(false, - SettingsCommandMulti.helpPrinter("indexed_range_slice"), + IRANGESLICE(false, "Standard1", "Super1", "Range slice queries through a secondary index. The cluster must first be populated by a write test, with indexing enabled.", - CommandCategory.MULTI + CommandCategory.BASIC ), - READMULTI(false, - SettingsCommandMulti.helpPrinter("readmulti"), + READMULTI(false, "Standard1", "Super1", "multi_read", "Multiple concurrent reads fetching multiple rows at once. The cluster must first be populated by a write test.", CommandCategory.MULTI ), - COUNTERWRITE(true, - SettingsCommand.helpPrinter("counteradd"), + COUNTERWRITE(true, "Counter1", "SuperCounter1", "counter_add", "Multiple concurrent updates of counters.", CommandCategory.BASIC ), - COUNTERREAD(false, - SettingsCommand.helpPrinter("counterread"), + COUNTERREAD(false, "Counter1", "SuperCounter1", "counter_get", "Multiple concurrent reads of counters. The cluster must first be populated by a counterwrite test.", CommandCategory.BASIC ), - HELP(false, SettingsMisc.helpHelpPrinter(), "-?", "Print help for a command or option", null), - PRINT(false, SettingsMisc.printHelpPrinter(), "Inspect the output of a distribution definition", null), - LEGACY(false, Legacy.helpPrinter(), "Legacy support mode", null) + HELP(false, null, null, "-?", "Print help for a command or option", null), + PRINT(false, null, null, "Inspect the output of a distribution definition", null), + LEGACY(false, null, null, "Legacy support mode", null) ; @@ -100,23 +92,49 @@ public enum Command public final CommandCategory category; public final String extraName; public final String description; - public final Runnable helpPrinter; + public final String table; + public final String supertable; - Command(boolean updates, Runnable helpPrinter, String description, CommandCategory category) + Command(boolean updates, String table, String supertable, String description, CommandCategory category) { - this(updates, helpPrinter, null, description, category); + this(updates, table, supertable, null, description, category); } - Command(boolean updates, Runnable helpPrinter, String extra, String description, CommandCategory category) + + Command(boolean updates, String table, String supertable, String extra, String description, CommandCategory category) { + this.table = table; + this.supertable = supertable; this.updates = updates; this.category = category; - this.helpPrinter = helpPrinter; this.extraName = extra; this.description = description; } + public void printHelp() { - helpPrinter.run(); + helpPrinter().run(); } -} + public final Runnable helpPrinter() + { + switch (this) + { + case PRINT: + return SettingsMisc.printHelpPrinter(); + case HELP: + return SettingsMisc.helpHelpPrinter(); + case LEGACY: + return Legacy.helpPrinter(); + } + switch (category) + { + case BASIC: + case MULTI: + return SettingsCommand.helpPrinter(this); + case MIXED: + return SettingsCommandMixed.helpPrinter(); + } + throw new AssertionError(); + } + +} \ No newline at end of file diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/Option.java b/tools/stress/src/org/apache/cassandra/stress/settings/Option.java index bc663f56dc..a9e669ce45 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/Option.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/Option.java @@ -31,6 +31,7 @@ abstract class Option abstract String shortDisplay(); abstract String longDisplay(); abstract List multiLineDisplay(); + abstract boolean setByUser(); public int hashCode() { diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDataGen.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDataGen.java index f8ced72e06..bde2b104d2 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDataGen.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDataGen.java @@ -83,6 +83,11 @@ class OptionDataGen extends Option return factory != null || defaultFactory != null; } + public boolean setByUser() + { + return factory != null; + } + @Override public String shortDisplay() { diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java index feaf017ff9..b84bbc2c87 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionDistribution.java @@ -43,11 +43,13 @@ class OptionDistribution extends Option final String prefix; private String spec; private final String defaultSpec; + private final String description; - public OptionDistribution(String prefix, String defaultSpec) + public OptionDistribution(String prefix, String defaultSpec, String description) { this.prefix = prefix; this.defaultSpec = defaultSpec; + this.description = description; } @Override @@ -88,7 +90,7 @@ class OptionDistribution extends Option public String longDisplay() { - return shortDisplay() + ": Specify a mathematical distribution"; + return shortDisplay() + ": " + description; } @Override @@ -105,10 +107,15 @@ class OptionDistribution extends Option ); } + boolean setByUser() + { + return spec != null; + } + @Override public String shortDisplay() { - return prefix + "DIST(?)"; + return (defaultSpec != null ? "[" : "") + prefix + "DIST(?)" + (defaultSpec != null ? "]" : ""); } private static final Map LOOKUP; diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java index 7074dc6a69..60faad87f1 100644 --- a/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java +++ b/tools/stress/src/org/apache/cassandra/stress/settings/OptionMulti.java @@ -100,7 +100,7 @@ abstract class OptionMulti extends Option StringBuilder sb = new StringBuilder(); sb.append(name); sb.append("("); - for (Option option : options()) + for (Option option : delegate.options()) { sb.append(option); sb.append(","); @@ -112,7 +112,7 @@ abstract class OptionMulti extends Option @Override public String shortDisplay() { - return name + "(?)"; + return (happy() ? "[" : "") + name + "(?)" + (happy() ? "]" : ""); } @Override @@ -121,7 +121,7 @@ abstract class OptionMulti extends Option StringBuilder sb = new StringBuilder(); sb.append(name); sb.append("("); - for (Option opt : options()) + for (Option opt : delegate.options()) { sb.append(opt.shortDisplay()); } @@ -181,6 +181,37 @@ abstract class OptionMulti extends Option { return Collections.emptyList(); } - }; + + boolean setByUser() + { + return !options.isEmpty(); + } + } + + List