diff --git a/CHANGES.txt b/CHANGES.txt
index 70281be06d..fd96002aab 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -37,7 +37,7 @@
    and few other places responsible for work with SSTable files (CASSANDRA-3040)
  * Stop reading from sstables once we know we have the most recent columns,
    for query-by-name requests (CASSANDRA-2498)
-
+ * Add query-by-column mode to stress.java (CASSANDRA-3064)
 
 0.8.5
  * fix NPE when encryption_options is unspecified (CASSANDRA-3007)
diff --git a/tools/stress/src/org/apache/cassandra/stress/Session.java b/tools/stress/src/org/apache/cassandra/stress/Session.java
index 1508379b25..df0305b5b6 100644
--- a/tools/stress/src/org/apache/cassandra/stress/Session.java
+++ b/tools/stress/src/org/apache/cassandra/stress/Session.java
@@ -20,10 +20,14 @@ package org.apache.cassandra.stress;
 import java.io.*;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.cassandra.config.ConfigurationException;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.TypeParser;
 import org.apache.commons.cli.*;
 
 import org.apache.cassandra.db.ColumnFamilyType;
@@ -41,6 +45,9 @@ public class Session implements Serializable
     // command line options
     public static final Options availableOptions = new Options();
 
+    public static final String DEFAULT_COMPARATOR = "AsciiType";
+    public static final String DEFAULT_VALIDATOR  = "BytesType";
+
     public final AtomicInteger operations;
     public final AtomicInteger keys;
     public final AtomicLong    latency;
@@ -78,6 +85,7 @@ public class Session implements Serializable
         availableOptions.addOption("V",  "average-size-values",  false,  "Generate column values of average rather than specific size");
         availableOptions.addOption("T",  "send-to",              true,   "Send this as a request to the stress daemon at specified address.");
         availableOptions.addOption("I",  "compression",          false,  "Use sstable compression when creating schema");
+        availableOptions.addOption("Q",  "query-names",          true,   "Comma-separated list of column names to retrieve from each row.");
     }
 
     private int numKeys          = 1000 * 1000;
@@ -109,6 +117,9 @@ public class Session implements Serializable
     private String replicationStrategy = "org.apache.cassandra.locator.SimpleStrategy";
     private Map<String, String> replicationStrategyOptions = new HashMap<String, String>();
 
+    // if we know exactly column names that we want to read (set by -Q option)
+    public final List<ByteBuffer> columnNames;
+
     public final boolean averageSizeValues;
 
     // required by Gaussian distribution.
@@ -275,11 +286,30 @@ public class Session implements Serializable
             {
                 throw new RuntimeException(e);
             }
+
+            if (cmd.hasOption("Q"))
+            {
+                AbstractType comparator = TypeParser.parse(DEFAULT_COMPARATOR);
+
+                String[] names = StringUtils.split(cmd.getOptionValue("Q"), ",");
+                columnNames = new ArrayList<ByteBuffer>(names.length);
+
+                for (String columnName : names)
+                    columnNames.add(comparator.fromString(columnName));
+            }
+            else
+            {
+                columnNames = null;
+            }
         }
         catch (ParseException e)
         {
             throw new IllegalArgumentException(e.getMessage(), e);
         }
+        catch (ConfigurationException e)
+        {
+            throw new IllegalStateException(e.getMessage(), e);
+        }
 
         mean  = numDifferentKeys / 2;
         sigma = numDifferentKeys * STDev;
@@ -417,8 +447,11 @@ public class Session implements Serializable
 
         // column family for standard columns
         CfDef standardCfDef = new CfDef("Keyspace1", "Standard1");
-        System.out.println("Compression = " + compression);
-        standardCfDef.setComparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
+
+        standardCfDef.setComparator_type(DEFAULT_COMPARATOR)
+                     .setDefault_validation_class(DEFAULT_VALIDATOR)
+                     .setCompression(compression);
+
         if (indexType != null)
         {
             ColumnDef standardColumn = new ColumnDef(ByteBufferUtil.bytes("C1"), "BytesType");
@@ -428,7 +461,10 @@ public class Session implements Serializable
 
         // column family with super columns
         CfDef superCfDef = new CfDef("Keyspace1", "Super1").setColumn_type("Super");
-        superCfDef.setComparator_type("AsciiType").setSubcomparator_type("AsciiType").setDefault_validation_class("BytesType").setCompression(compression);
+        superCfDef.setComparator_type(DEFAULT_COMPARATOR)
+                  .setSubcomparator_type(DEFAULT_COMPARATOR)
+                  .setDefault_validation_class(DEFAULT_VALIDATOR)
+                  .setCompression(compression);
 
         // column family for standard counters
         CfDef counterCfDef = new CfDef("Keyspace1", "Counter1").setDefault_validation_class("CounterColumnType").setReplicate_on_write(replicateOnWrite).setCompression(compression);
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java b/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java
index 6dcd4cd6fb..b5a8781953 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/Reader.java
@@ -37,16 +37,13 @@ public class Reader extends Operation
 
     public void run(Cassandra.Client client) throws IOException
     {
-        SliceRange sliceRange = new SliceRange();
-
-        // start/finish
-        sliceRange.setStart(new byte[] {}).setFinish(new byte[] {});
-
-        // reversed/count
-        sliceRange.setReversed(false).setCount(session.getColumnsPerKey());
-
         // initialize SlicePredicate with existing SliceRange
-        SlicePredicate predicate = new SlicePredicate().setSlice_range(sliceRange);
+        SlicePredicate predicate = new SlicePredicate();
+
+        if (session.columnNames == null)
+            predicate.setSlice_range(getSliceRange());
+        else // see CASSANDRA-3064 about why this is useful
+            predicate.setColumn_names(session.columnNames);
 
         if (session.getColumnFamilyType() == ColumnFamilyType.Super)
         {
@@ -150,4 +147,12 @@ public class Reader extends Operation
         session.latency.getAndAdd(System.currentTimeMillis() - start);
     }
 
+    private SliceRange getSliceRange()
+    {
+        return new SliceRange()
+                    .setStart(new byte[] {})
+                    .setFinish(new byte[] {})
+                    .setReversed(false)
+                    .setCount(session.getColumnsPerKey());
+    }
 }