cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From xe...@apache.org
Subject git commit: Improvements and fixes to cassandra/stress Patch by Benedict; reviewed by Pavel Yaskevich for CASSANDRA-6691
Date Thu, 13 Feb 2014 01:32:53 GMT
Updated Branches:
  refs/heads/trunk f66b9eb27 -> 10b617364


Improvements and fixes to cassandra/stress
Patch by Benedict; reviewed by Pavel Yaskevich for CASSANDRA-6691


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/10b61736
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/10b61736
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/10b61736

Branch: refs/heads/trunk
Commit: 10b617364e9f639358f82c70056e31533a1ec11c
Parents: f66b9eb
Author: belliottsmith <github@sub.laerad.com>
Authored: Wed Feb 12 11:36:35 2014 +0000
Committer: Pavel Yaskevich <xedin@apache.org>
Committed: Wed Feb 12 17:25:07 2014 -0800

----------------------------------------------------------------------
 .../org/apache/cassandra/stress/Operation.java  |  74 ++++++++++---
 .../apache/cassandra/stress/StressAction.java   |  13 ++-
 .../cassandra/stress/generatedata/DataGen.java  |   6 +-
 .../stress/generatedata/DataGenBytesRandom.java |   2 +-
 .../stress/generatedata/DataGenHex.java         |   2 +-
 .../generatedata/DataGenStringDictionary.java   |   6 +-
 .../generatedata/DataGenStringRepeats.java      |  16 +--
 .../cassandra/stress/generatedata/KeyGen.java   |   2 +-
 .../cassandra/stress/generatedata/RowGen.java   |   4 +-
 .../operations/CqlIndexedRangeSlicer.java       |   2 +-
 .../stress/operations/CqlInserter.java          |   2 +-
 .../stress/operations/CqlOperation.java         | 111 +++++++++++++++++++
 .../cassandra/stress/operations/CqlReader.java  |   8 +-
 .../stress/operations/ThriftCounterAdder.java   |   2 +-
 .../operations/ThriftIndexedRangeSlicer.java    |   2 +-
 .../stress/operations/ThriftInserter.java       |   6 +-
 .../stress/operations/ThriftReader.java         |  30 ++++-
 .../cassandra/stress/settings/SettingsKey.java  |   7 +-
 18 files changed, 244 insertions(+), 51 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/Operation.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/Operation.java b/tools/stress/src/org/apache/cassandra/stress/Operation.java
index fa7a453..4519b19 100644
--- a/tools/stress/src/org/apache/cassandra/stress/Operation.java
+++ b/tools/stress/src/org/apache/cassandra/stress/Operation.java
@@ -21,10 +21,25 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.EnumMap;
 import java.util.List;
 
 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;
@@ -66,7 +81,8 @@ public abstract class Operation
         public final RowGen rowGen;
         public final List<ColumnParent> columnParents;
         public final StressMetrics metrics;
-        public final SettingsCommandMixed.CommandSelector readWriteSelector;
+        public final SettingsCommandMixed.CommandSelector commandSelector;
+        private final EnumMap<Command, State> substates;
         private Object cqlCache;
 
         public State(Command type, StressSettings settings, StressMetrics metrics)
@@ -74,9 +90,15 @@ public abstract class Operation
             this.type = type;
             this.timer = metrics.getTiming().newTimer();
             if (type == Command.MIXED)
-                readWriteSelector = ((SettingsCommandMixed) settings.command).selector();
+            {
+                commandSelector = ((SettingsCommandMixed) settings.command).selector();
+                substates = new EnumMap<>(Command.class);
+            }
             else
-                readWriteSelector = null;
+            {
+                commandSelector = null;
+                substates = null;
+            }
             this.settings = settings;
             this.keyGen = settings.keys.newKeyGen();
             this.rowGen = settings.columns.newRowGen();
@@ -91,6 +113,20 @@ public abstract class Operation
                 columnParents = Arrays.asList(cp);
             }
         }
+
+        private State(Command type, State copy)
+        {
+            this.type = type;
+            this.timer = copy.timer;
+            this.rowGen = copy.rowGen;
+            this.keyGen = copy.keyGen;
+            this.columnParents = copy.columnParents;
+            this.metrics = copy.metrics;
+            this.settings = copy.settings;
+            this.substates = null;
+            this.commandSelector = null;
+        }
+
         public boolean isCql3()
         {
             return settings.mode.cqlVersion == CqlVersion.CQL3;
@@ -107,6 +143,18 @@ public abstract class Operation
         {
             cqlCache = val;
         }
+
+        public State substate(Command command)
+        {
+            assert type == Command.MIXED;
+            State substate = substates.get(command);
+            if (substate == null)
+            {
+                substates.put(command, substate = new State(command, this));
+            }
+            return substate;
+        }
+
     }
 
     protected ByteBuffer getKey()
@@ -119,9 +167,9 @@ public abstract class Operation
         return state.keyGen.getKeys(count, index);
     }
 
-    protected List<ByteBuffer> generateColumnValues()
+    protected List<ByteBuffer> generateColumnValues(ByteBuffer key)
     {
-        return state.rowGen.generate(index);
+        return state.rowGen.generate(index, key);
     }
 
     /**
@@ -146,20 +194,18 @@ public abstract class Operation
         boolean success = false;
         String exceptionMessage = null;
 
-        for (int t = 0; t < state.settings.command.tries; t++)
+        int tries = 0;
+        for (; tries < state.settings.command.tries; tries++)
         {
-            if (success)
-                break;
-
             try
             {
                 success = run.run();
+                break;
             }
             catch (Exception e)
             {
                 System.err.println(e);
                 exceptionMessage = getExceptionMessage(e);
-                success = false;
             }
         }
 
@@ -167,11 +213,13 @@ public abstract class Operation
 
         if (!success)
         {
-            error(String.format("Operation [%d] retried %d times - error executing for key
%s %s%n",
+            error(String.format("Operation [%d] x%d key %s %s%n",
                     index,
-                    state.settings.command.tries,
+                    tries,
                     run.key(),
-                    (exceptionMessage == null) ? "" : "(" + exceptionMessage + ")"));
+                    (exceptionMessage == null)
+                        ? "Data returned was not validated"
+                        : "Error executing: " + exceptionMessage));
         }
 
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/StressAction.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/StressAction.java b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
index 0312093..94824ec 100644
--- a/tools/stress/src/org/apache/cassandra/stress/StressAction.java
+++ b/tools/stress/src/org/apache/cassandra/stress/StressAction.java
@@ -93,6 +93,10 @@ public class StressAction implements Runnable
             default:
                 throw new IllegalStateException();
         }
+
+        // we need to warm up all the nodes in the cluster ideally, but we may not be the
only stress instance;
+        // so warm up all the nodes we're speaking to only.
+        iterations *= settings.node.nodes.size();
         output.println(String.format("Warming up %s with %d iterations...", type, iterations));
         run(type, 20, iterations, warmupOutput);
     }
@@ -151,9 +155,7 @@ public class StressAction implements Runnable
 
     private boolean hasAverageImprovement(List<StressMetrics> results, int count, double
minImprovement)
     {
-        if (results.size() < count + 1)
-            return true;
-        return averageImprovement(results, count) >= minImprovement;
+        return results.size() < count + 1 || averageImprovement(results, count) >=
minImprovement;
     }
 
     private double averageImprovement(List<StressMetrics> results, int count)
@@ -385,7 +387,7 @@ public class StressAction implements Runnable
             int batchSize = (int) (operations / (1 << 19));
             if (batchSize < 20)
                 batchSize = 20;
-            ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<Work>(
+            ArrayBlockingQueue<Work> work = new ArrayBlockingQueue<>(
                     (int) ((operations / batchSize)
                   + (operations % batchSize == 0 ? 0 : 1))
             );
@@ -533,7 +535,8 @@ public class StressAction implements Runnable
                 }
 
             case MIXED:
-                return createOperation(state.readWriteSelector.next(), state, index);
+                Command subcommand = state.commandSelector.next();
+                return createOperation(subcommand, state.substate(subcommand), index);
 
         }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGen.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGen.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGen.java
index 4c22005..c441b7e 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGen.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGen.java
@@ -6,13 +6,13 @@ import java.util.List;
 public abstract class DataGen
 {
 
-    public abstract void generate(ByteBuffer fill, long offset);
+    public abstract void generate(ByteBuffer fill, long index, ByteBuffer seed);
     public abstract boolean isDeterministic();
 
-    public void generate(List<ByteBuffer> fills, long offset)
+    public void generate(List<ByteBuffer> fills, long index, ByteBuffer seed)
     {
         for (ByteBuffer fill : fills)
-            generate(fill, offset++);
+            generate(fill, index++, seed);
     }
 
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenBytesRandom.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenBytesRandom.java
b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenBytesRandom.java
index 3906f93..cce438d 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenBytesRandom.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenBytesRandom.java
@@ -9,7 +9,7 @@ public class DataGenBytesRandom extends DataGen
     private final Random rnd = new Random();
 
     @Override
-    public void generate(ByteBuffer fill, long offset)
+    public void generate(ByteBuffer fill, long index, ByteBuffer seed)
     {
         fill.clear();
         rnd.nextBytes(fill.array());

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenHex.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenHex.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenHex.java
index 50d49dd..b71d3e9 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenHex.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenHex.java
@@ -8,7 +8,7 @@ public abstract class DataGenHex extends DataGen
     abstract long next(long operationIndex);
 
     @Override
-    public final void generate(ByteBuffer fill, long operationIndex)
+    public final void generate(ByteBuffer fill, long operationIndex, ByteBuffer seed)
     {
         fill.clear();
         fillKeyStringBytes(next(operationIndex), fill.array());

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
index e581232..7733ed6 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringDictionary.java
@@ -25,13 +25,13 @@ public class DataGenStringDictionary extends DataGen
     }
 
     @Override
-    public void generate(ByteBuffer fill, long index)
+    public void generate(ByteBuffer fill, long index, ByteBuffer seed)
     {
         fill(fill, 0);
     }
 
     @Override
-    public void generate(List<ByteBuffer> fills, long index)
+    public void generate(List<ByteBuffer> fills, long index, ByteBuffer seed)
     {
         for (int i = 0 ; i < fills.size() ; i++)
             fill(fills.get(0), i);
@@ -55,7 +55,7 @@ public class DataGenStringDictionary extends DataGen
     @Override
     public boolean isDeterministic()
     {
-        return true;
+        return false;
     }
 
     public static DataGenFactory getFactory(File file) throws IOException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
index 47091f7..4c5bb89 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/DataGenStringRepeats.java
@@ -25,32 +25,32 @@ public class DataGenStringRepeats extends DataGen
     }
 
     @Override
-    public void generate(ByteBuffer fill, long index)
+    public void generate(ByteBuffer fill, long index, ByteBuffer seed)
     {
-        fill(fill, index, 0);
+        fill(fill, index, 0, seed);
     }
 
     @Override
-    public void generate(List<ByteBuffer> fills, long index)
+    public void generate(List<ByteBuffer> fills, long index, ByteBuffer seed)
     {
         for (int i = 0 ; i < fills.size() ; i++)
         {
-            fill(fills.get(i), index, i);
+            fill(fills.get(i), index, i, seed);
         }
     }
 
-    private void fill(ByteBuffer fill, long index, int column)
+    private void fill(ByteBuffer fill, long index, int column, ByteBuffer seed)
     {
         fill.clear();
         byte[] trg = fill.array();
-        byte[] src = getData(index, column);
+        byte[] src = getData(index, column, seed);
         for (int j = 0 ; j < trg.length ; j += src.length)
             System.arraycopy(src, 0, trg, j, Math.min(src.length, trg.length - j));
     }
 
-    private byte[] getData(long index, int column)
+    private byte[] getData(long index, int column, ByteBuffer seed)
     {
-        final long key = (column * repeatFrequency) + (index % repeatFrequency);
+        final long key = (column * repeatFrequency) + ((seed == null ? index : Math.abs(seed.hashCode()))
% repeatFrequency);
         byte[] r = cache.get(key);
         if (r != null)
             return r;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/KeyGen.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/generatedata/KeyGen.java b/tools/stress/src/org/apache/cassandra/stress/generatedata/KeyGen.java
index cdd6d39..36dc31d 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/KeyGen.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/KeyGen.java
@@ -21,7 +21,7 @@ public class KeyGen
     {
         while (keyBuffers.size() < n)
             keyBuffers.add(ByteBuffer.wrap(new byte[keySize]));
-        dataGen.generate(keyBuffers, index);
+        dataGen.generate(keyBuffers, index, null);
         return keyBuffers;
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java
----------------------------------------------------------------------
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 869fbc7..3174177 100644
--- a/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java
+++ b/tools/stress/src/org/apache/cassandra/stress/generatedata/RowGen.java
@@ -16,10 +16,10 @@ public abstract class RowGen
         this.dataGen = dataGenerator;
     }
 
-    public List<ByteBuffer> generate(long operationIndex)
+    public List<ByteBuffer> generate(long operationIndex, ByteBuffer key)
     {
         List<ByteBuffer> fill = getColumns(operationIndex);
-        dataGen.generate(fill, operationIndex);
+        dataGen.generate(fill, operationIndex, key);
         return fill;
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java
----------------------------------------------------------------------
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 748bf30..ff43322 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlIndexedRangeSlicer.java
@@ -70,7 +70,7 @@ public class CqlIndexedRangeSlicer extends CqlOperation<byte[][]>
     protected void run(CqlOperation.ClientWrapper client) throws IOException
     {
         acceptNoResults = false;
-        final List<ByteBuffer> columns = generateColumnValues();
+        final List<ByteBuffer> columns = generateColumnValues(getKey());
         final ByteBuffer value = columns.get(1); // only C1 column is indexed
         byte[] minKey = new byte[0];
         int rowCount;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java
----------------------------------------------------------------------
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 82f00aa..8d964f5 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlInserter.java
@@ -72,7 +72,7 @@ public class CqlInserter extends CqlOperation<Integer>
     protected List<ByteBuffer> getQueryParameters(byte[] key)
     {
         final ArrayList<ByteBuffer> queryParams = new ArrayList<>();
-        final List<ByteBuffer> values = generateColumnValues();
+        final List<ByteBuffer> values = generateColumnValues(ByteBuffer.wrap(key));
         queryParams.addAll(values);
         queryParams.add(ByteBuffer.wrap(key));
         return queryParams;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/CqlOperation.java
----------------------------------------------------------------------
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 744e7f6..78dd461 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.Arrays;
 import java.util.List;
 
 import com.datastax.driver.core.PreparedStatement;
@@ -33,6 +34,7 @@ import org.apache.cassandra.stress.util.JavaDriverClient;
 import org.apache.cassandra.stress.util.ThriftClient;
 import org.apache.cassandra.thrift.Compression;
 import org.apache.cassandra.thrift.CqlResult;
+import org.apache.cassandra.thrift.CqlRow;
 import org.apache.cassandra.thrift.ThriftConversion;
 import org.apache.cassandra.transport.SimpleClient;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -167,6 +169,35 @@ public abstract class CqlOperation<V> extends Operation
 
     }
 
+    protected final class CqlRunOpMatchResults extends CqlRunOp<ByteBuffer[][]>
+    {
+
+        final List<List<ByteBuffer>> 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<ByteBuffer> params, String id, ByteBuffer key, List<List<ByteBuffer>>
expect)
+        {
+            super(client, query, queryId, RowsHandler.INSTANCE, params, id, key);
+            this.expect = expect;
+        }
+
+        @Override
+        public int keyCount()
+        {
+            return result == null ? 0 : result.length;
+        }
+
+        public boolean validate(ByteBuffer[][] result)
+        {
+            if (result.length != expect.size())
+                return false;
+            for (int i = 0 ; i < result.length ; i++)
+                if (!expect.get(i).equals(Arrays.asList(result[i])))
+                    return false;
+            return true;
+        }
+    }
+
     // Cql
     protected abstract class CqlRunOp<V> implements RunOp
     {
@@ -451,6 +482,86 @@ public abstract class CqlOperation<V> extends Operation
     }
 
     // Processes results from each client into an array of all key bytes returned
+    protected static final class RowsHandler implements ResultHandler<ByteBuffer[][]>
+    {
+        static final RowsHandler INSTANCE = new RowsHandler();
+
+        @Override
+        public Function<ResultSet, ByteBuffer[][]> javaDriverHandler()
+        {
+            return new Function<ResultSet, ByteBuffer[][]>()
+            {
+
+                @Override
+                public ByteBuffer[][] apply(ResultSet result)
+                {
+                    if (result == null)
+                        return new ByteBuffer[0][];
+                    List<Row> rows = result.all();
+
+                    ByteBuffer[][] r = new ByteBuffer[rows.size()][];
+                    for (int i = 0 ; i < r.length ; i++)
+                    {
+                        Row row = rows.get(i);
+                        r[i] = new ByteBuffer[row.getColumnDefinitions().size() - 1];
+                        for (int j = 1 ; j < row.getColumnDefinitions().size() ; j++)
+                            r[i][j - 1] = row.getBytes(j);
+                    }
+                    return r;
+                }
+            };
+        }
+
+        @Override
+        public Function<ResultMessage, ByteBuffer[][]> thriftHandler()
+        {
+            return new Function<ResultMessage, ByteBuffer[][]>()
+            {
+
+                @Override
+                public ByteBuffer[][] apply(ResultMessage result)
+                {
+                    if (!(result instanceof ResultMessage.Rows))
+                        return new ByteBuffer[0][];
+
+                    ResultMessage.Rows rows = ((ResultMessage.Rows) result);
+                    ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
+                    for (int i = 0 ; i < r.length ; i++)
+                    {
+                        List<ByteBuffer> row = rows.result.rows.get(i);
+                        r[i] = new ByteBuffer[row.size()];
+                        for (int j = 0 ; j < row.size() ; j++)
+                            r[i][j] = row.get(j);
+                    }
+                    return r;
+                }
+            };
+        }
+
+        @Override
+        public Function<CqlResult, ByteBuffer[][]> simpleNativeHandler()
+        {
+            return new Function<CqlResult, ByteBuffer[][]>()
+            {
+
+                @Override
+                public ByteBuffer[][] apply(CqlResult result)
+                {
+                    ByteBuffer[][] r = new ByteBuffer[result.getRows().size()][];
+                    for (int i = 0 ; i < r.length ; i++)
+                    {
+                        CqlRow row = result.getRows().get(i);
+                        r[i] = new ByteBuffer[row.getColumns().size()];
+                        for (int j = 0 ; j < r[i].length ; j++)
+                            r[i][j] = ByteBuffer.wrap(row.getColumns().get(j).getValue());
+                    }
+                    return r;
+                }
+            };
+        }
+
+    }
+    // Processes results from each client into an array of all key bytes returned
     protected static final class KeysHandler implements ResultHandler<byte[][]>
     {
         static final KeysHandler INSTANCE = new KeysHandler();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java
----------------------------------------------------------------------
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 749a482..44da43f 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/CqlReader.java
@@ -23,10 +23,11 @@ package org.apache.cassandra.stress.operations;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 
-public class CqlReader extends CqlOperation<Integer>
+public class CqlReader extends CqlOperation<ByteBuffer[][]>
 {
 
     public CqlReader(State state, long idx)
@@ -79,9 +80,10 @@ public class CqlReader extends CqlOperation<Integer>
     }
 
     @Override
-    protected CqlRunOp<Integer> buildRunOp(ClientWrapper client, String query, Object
queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
+    protected CqlRunOp<ByteBuffer[][]> buildRunOp(ClientWrapper client, String query,
Object queryId, List<ByteBuffer> params, String keyid, ByteBuffer key)
     {
-        return new CqlRunOpTestNonEmpty(client, query, queryId, params, keyid, key);
+        List<ByteBuffer> expectRow = state.rowGen.isDeterministic() ? generateColumnValues(key)
: null;
+        return new CqlRunOpMatchResults(client, query, queryId, params, keyid, key, Arrays.asList(expectRow));
     }
 
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java
----------------------------------------------------------------------
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 b1657b2..26695a6 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftCounterAdder.java
@@ -39,7 +39,7 @@ public class ThriftCounterAdder extends Operation
 
     public void run(final ThriftClient client) throws IOException
     {
-        List<CounterColumn> columns = new ArrayList<CounterColumn>();
+        List<CounterColumn> columns = new ArrayList<>();
         for (int i = 0; i < state.settings.columns.maxColumnsPerKey; i++)
             columns.add(new CounterColumn(getColumnNameBytes(i), 1L));
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java
----------------------------------------------------------------------
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 c6b1b03..6eab209 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftIndexedRangeSlicer.java
@@ -49,7 +49,7 @@ public class ThriftIndexedRangeSlicer extends Operation
                 .setSlice_range(new SliceRange(ByteBufferUtil.EMPTY_BYTE_BUFFER,
                         ByteBufferUtil.EMPTY_BYTE_BUFFER,
                         false, state.settings.columns.maxColumnsPerKey));
-        final List<ByteBuffer> columns = generateColumnValues();
+        final List<ByteBuffer> columns = generateColumnValues(getKey());
         final ColumnParent parent = state.columnParents.get(0);
 
         final ByteBuffer columnName = getColumnNameBytes(1);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java
----------------------------------------------------------------------
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 c5f8051..b107f26 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftInserter.java
@@ -42,7 +42,7 @@ public final class ThriftInserter extends Operation
     public void run(final ThriftClient client) throws IOException
     {
         final ByteBuffer key = getKey();
-        final List<Column> columns = generateColumns();
+        final List<Column> columns = generateColumns(key);
 
         Map<String, List<Mutation>> row;
         if (!state.settings.columns.useSuperColumns)
@@ -92,9 +92,9 @@ public final class ThriftInserter extends Operation
         });
     }
 
-    protected List<Column> generateColumns()
+    protected List<Column> generateColumns(ByteBuffer key)
     {
-        final List<ByteBuffer> values = generateColumnValues();
+        final List<ByteBuffer> values = generateColumnValues(key);
         final List<Column> columns = new ArrayList<>(values.size());
 
         if (state.settings.columns.useTimeUUIDComparator)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java
----------------------------------------------------------------------
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 a8605e8..c50843f 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/ThriftReader.java
@@ -19,12 +19,16 @@ 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;
 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
 {
@@ -48,6 +52,7 @@ public final class ThriftReader extends Operation
             predicate.setColumn_names(state.settings.columns.names);
 
         final ByteBuffer key = getKey();
+        final List<ByteBuffer> expect = state.rowGen.isDeterministic() ? generateColumnValues(key)
: null;
         for (final ColumnParent parent : state.columnParents)
         {
             timeWithRetry(new RunOp()
@@ -55,7 +60,30 @@ public final class ThriftReader extends Operation
                 @Override
                 public boolean run() throws Exception
                 {
-                    return client.get_slice(key, parent, predicate, state.settings.command.consistencyLevel).size()
!= 0;
+                    List<ColumnOrSuperColumn> row = client.get_slice(key, parent, predicate,
state.settings.command.consistencyLevel);
+                    if (expect == null)
+                        return !row.isEmpty();
+                    if (!state.settings.columns.useSuperColumns)
+                    {
+                        if (row.size() != expect.size())
+                            return false;
+                        for (int i = 0 ; i < row.size() ; i++)
+                            if (!row.get(i).getColumn().bufferForValue().equals(expect.get(i)))
+                                return false;
+                    }
+                    else
+                    {
+                        for (ColumnOrSuperColumn col : row)
+                        {
+                            SuperColumn superColumn = col.getSuper_column();
+                            if (superColumn.getColumns().size() != expect.size())
+                                return false;
+                            for (int i = 0 ; i < expect.size() ; i++)
+                                if (!superColumn.getColumns().get(i).bufferForValue().equals(expect.get(i)))
+                                    return false;
+                        }
+                    }
+                    return true;
                 }
 
                 @Override

http://git-wip-us.apache.org/repos/asf/cassandra/blob/10b61736/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
----------------------------------------------------------------------
diff --git a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
index 6cef0bf..e2647bb 100644
--- a/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
+++ b/tools/stress/src/org/apache/cassandra/stress/settings/SettingsKey.java
@@ -73,9 +73,10 @@ public class SettingsKey implements Serializable
 
     public KeyGen newKeyGen()
     {
-        if (range != null)
-            return new KeyGen(new DataGenHexFromOpIndex(range[0], range[1]), keySize);
-        return new KeyGen(new DataGenHexFromDistribution(distribution.get()), keySize);
+        return new KeyGen(range == null
+                            ? new DataGenHexFromDistribution(distribution.get())
+                            : new DataGenHexFromOpIndex(range[0], range[1]),
+                          keySize);
     }
 
     // CLI Utility Methods


Mime
View raw message