diff --git a/CHANGES.txt b/CHANGES.txt
index c7da4908cc..a2e5bc6676 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -21,11 +21,15 @@
    (CASSANDRA-2419)
  * JDBC CQL driver exposes getColumn for access to timestamp
  * JDBC ResultSetMetadata properties added to AbstractType
+ * r/m clustertool (CASSANDRA-2607)
+ * add support for presenting row key as a column in CQL result sets 
+   (CASSANDRA-2622)
 
 
 0.8.0-beta2
  * fix NPE compacting index CFs (CASSANDRA-2528)
- * Remove checking all column families on startup for compaction candidates (CASSANDRA-2444)
+ * Remove checking all column families on startup for compaction candidates 
+   (CASSANDRA-2444)
  * validate CQL create keyspace options (CASSANDRA-2525)
  * fix nodetool setcompactionthroughput (CASSANDRA-2550)
  * move	gossip heartbeat back to its own thread (CASSANDRA-2554)
diff --git a/NEWS.txt b/NEWS.txt
index 760ff39fa3..8a4a64e793 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -37,6 +37,13 @@ Features
       will be removed in a future major release once we are satisfied that
       memtable_total_space_in_mb works adequately.
 
+Tools
+-----
+    - stress and py_stress moved from contrib/ to tools/
+    - clustertool was removed (see 
+      https://issues.apache.org/jira/browse/CASSANDRA-2607 for examples
+      of how to script nodetool across the cluster instead)
+
 Other
 -----
     - In the past, sstable2json would write column names and values as
diff --git a/bin/clustertool b/bin/clustertool
deleted file mode 100755
index 0dbfd02d6f..0000000000
--- a/bin/clustertool
+++ /dev/null
@@ -1,57 +0,0 @@
-#!/bin/sh
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements.  See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership.  The ASF licenses this file
-# to you under the Apache License, Version 2.0 (the
-# "License"); you may not use this file except in compliance
-# with the License.  You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-
-if [ "x$CASSANDRA_INCLUDE" = "x" ]; then
-    for include in /usr/share/cassandra/cassandra.in.sh \
-                   /usr/local/share/cassandra/cassandra.in.sh \
-                   /opt/cassandra/cassandra.in.sh \
-                   `dirname $0`/cassandra.in.sh; do
-        if [ -r $include ]; then
-            . $include
-            break
-        fi
-    done
-elif [ -r $CASSANDRA_INCLUDE ]; then
-    . $CASSANDRA_INCLUDE
-fi
-
-# Use JAVA_HOME if set, otherwise look for java in PATH
-if [ -x $JAVA_HOME/bin/java ]; then
-    JAVA=$JAVA_HOME/bin/java
-else
-    JAVA=`which java`
-fi
-
-if [ -z $CASSANDRA_CONF -o -z $CLASSPATH ]; then
-    echo "You must set the CASSANDRA_CONF and CLASSPATH vars" >&2
-    exit 1
-fi
-
-# Special-case path variables.
-case "`uname`" in
-    CYGWIN*) 
-        CLASSPATH=`cygpath -p -w "$CLASSPATH"`
-        CASSANDRA_CONF=`cygpath -p -w "$CASSANDRA_CONF"`
-    ;;
-esac
-
-$JAVA -cp $CLASSPATH \
-        -Dlog4j.configuration=log4j-tools.properties \
-        org.apache.cassandra.tools.ClusterCmd $@
-
-# vi:ai sw=4 ts=4 tw=0 et
diff --git a/bin/clustertool.bat b/bin/clustertool.bat
deleted file mode 100644
index c607f2f789..0000000000
--- a/bin/clustertool.bat
+++ /dev/null
@@ -1,67 +0,0 @@
-@REM
-@REM  Licensed to the Apache Software Foundation (ASF) under one or more
-@REM  contributor license agreements.  See the NOTICE file distributed with
-@REM  this work for additional information regarding copyright ownership.
-@REM  The ASF licenses this file to You under the Apache License, Version 2.0
-@REM  (the "License"); you may not use this file except in compliance with
-@REM  the License.  You may obtain a copy of the License at
-@REM
-@REM      http://www.apache.org/licenses/LICENSE-2.0
-@REM
-@REM  Unless required by applicable law or agreed to in writing, software
-@REM  distributed under the License is distributed on an "AS IS" BASIS,
-@REM  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-@REM  See the License for the specific language governing permissions and
-@REM  limitations under the License.
-
-@echo off
-if "%OS%" == "Windows_NT" setlocal
-
-if NOT DEFINED CASSANDRA_HOME set CASSANDRA_HOME=%~dp0..
-if NOT DEFINED CASSANDRA_CONF set CASSANDRA_CONF=%CASSANDRA_HOME%\conf
-if NOT DEFINED CASSANDRA_MAIN set CASSANDRA_MAIN=org.apache.cassandra.tools.ClusterCmd
-if NOT DEFINED JAVA_HOME goto err
-
-REM ***** JAVA options *****
-set JAVA_OPTS=^
- -Dlog4j.configuration=log4j-tools.properties
-
-REM ***** CLASSPATH library setting *****
-
-REM Ensure that any user defined CLASSPATH variables are not used on startup
-set CLASSPATH=%CASSANDRA_HOME%\conf
-
-REM For each jar in the CASSANDRA_HOME lib directory call append to build the CLASSPATH variable.
-for %%i in (%CASSANDRA_HOME%\lib\*.jar) do call :append %%~fi
-goto okClasspath
-
-:append
-set CLASSPATH=%CLASSPATH%;%1%2
-goto :eof
-
-:okClasspath
-REM Include the build\classes\main directory so it works in development
-set CASSANDRA_CLASSPATH=%CLASSPATH%;%CASSANDRA_HOME%\build\classes\main;%CASSANDRA_CONF%;%CASSANDRA_HOME%\build\classes\thrift
-
-set CASSANDRA_PARAMS=
-set TOOLS_PARAMS=
-
-FOR %%A IN (%*) DO call :appendToolsParams %%A
-
-goto runTool
-
-:appendToolsParams
-set TOOLS_PARAMS=%TOOLS_PARAMS% %1
-goto :eof
-
-:runTool
-"%JAVA_HOME%\bin\java" %JAVA_OPTS% %CASSANDRA_PARAMS% -cp "%CASSANDRA_CLASSPATH%" "%CASSANDRA_MAIN%" %TOOLS_PARAMS%
-goto finally
-
-:err
-echo JAVA_HOME environment variable must be set!
-pause
-
-:finally
-
-ENDLOCAL
diff --git a/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java b/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java
index a6367b5ef0..606bb1bcbd 100644
--- a/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java
+++ b/drivers/java/src/org/apache/cassandra/cql/jdbc/ColumnDecoder.java
@@ -25,6 +25,7 @@ import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.config.ConfigurationException;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.thrift.*;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -33,6 +34,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -80,12 +82,30 @@ class ColumnDecoder
     {
 
         CFMetaData md = metadata.get(String.format("%s.%s", keyspace, columnFamily));
+        try
+        {
+            if (ByteBufferUtil.string(name).equalsIgnoreCase(ByteBufferUtil.string(md.getKeyName())))
+                return AsciiType.instance;
+        }
+        catch (CharacterCodingException e)
+        {
+            // not be the key name
+        }
         return md.comparator;
     }
 
     AbstractType getValueType(String keyspace, String columnFamily, ByteBuffer name)
     {
         CFMetaData md = metadata.get(String.format("%s.%s", keyspace, columnFamily));
+        try
+        {
+            if (ByteBufferUtil.string(name).equalsIgnoreCase(ByteBufferUtil.string(md.getKeyName())))
+                return md.getKeyValidator();
+        }
+        catch (CharacterCodingException e)
+        {
+            // not be the key name
+        }
         ColumnDefinition cd = md.getColumnDefinition(name);
         return cd == null ? md.getDefaultValidator() : cd.getValidator();
     }
diff --git a/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java b/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java
index f5301abf56..f3f748fd7b 100644
--- a/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java
+++ b/drivers/java/test/org/apache/cassandra/cql/JdbcDriverTest.java
@@ -33,6 +33,7 @@ import org.junit.BeforeClass;
 import org.junit.Test;
 
 import org.apache.cassandra.db.marshal.*;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
 
 import static junit.framework.Assert.assertEquals;
@@ -120,6 +121,12 @@ public class JdbcDriverTest extends EmbeddedServiceBase
         expectedMetaData(md, 2, BigInteger.class.getName(), "JdbcInteger", "Keyspace1", "2", Types.BIGINT, IntegerType.class.getSimpleName(), true, false);
         expectedMetaData(md, 3, String.class.getName(), "JdbcInteger", "Keyspace1", "42", Types.VARCHAR, UTF8Type.class.getSimpleName(), false, true);
         
+        rs = stmt.executeQuery("select key, 1, 2, 42 from JdbcInteger where key='" + key + "'");
+        assert rs.next();
+        assert Arrays.equals(rs.getBytes("key"), FBUtilities.hexToBytes(key));
+        assert rs.getObject("1").equals(new BigInteger("36893488147419103232"));
+        assert rs.getString("42").equals("fortytwofortytwo") : rs.getString("42");
+
         stmt.executeUpdate("update JdbcUtf8 set a='aa', b='bb', fortytwo='4242' where key='" + key + "'");
         rs = stmt.executeQuery("select a, b, fortytwo from JdbcUtf8 where key='" + key + "'");
         assert rs.next();
diff --git a/drivers/py/cql/__init__.py b/drivers/py/cql/__init__.py
index b5b58bc6ed..de0087699b 100644
--- a/drivers/py/cql/__init__.py
+++ b/drivers/py/cql/__init__.py
@@ -45,8 +45,6 @@ apilevel = 1.0
 threadsafety = 1 # Threads may share the module, but not connections/cursors.
 paramstyle = 'named'
 
-ROW_KEY = "Row Key"
-
 # TODO: Pull connections out of a pool instead.
 def connect(host, port=9160, keyspace='system', user=None, password=None):
     return connection.Connection(host, port, keyspace, user, password)
diff --git a/drivers/py/cql/cursor.py b/drivers/py/cql/cursor.py
index 9771c67a99..8858d8d3ea 100644
--- a/drivers/py/cql/cursor.py
+++ b/drivers/py/cql/cursor.py
@@ -92,16 +92,14 @@ class Cursor:
             return results
 
         def column_families(cf_defs):
-            cfresults = {}
-            if cf_defs:
-                for cf in cf_defs:
-                    cfresults[cf.name] = {"comparator": cf.comparator_type}
-                    cfresults[cf.name]["default_validation_class"] = \
-                             cf.default_validation_class
-                    cfresults[cf.name]["key_validation_class"] = \
-                             cf.key_validation_class
-                    cfresults[cf.name]["columns"] = columns(cf.column_metadata)
-            return cfresults
+            d = {}
+            for cf in cf_defs:
+                d[cf.name] = {'comparator': cf.comparator_type,
+                              'default_validation_class': cf.default_validation_class,
+                              'key_validation_class': cf.key_validation_class,
+                              'columns': columns(cf.column_metadata),
+                              'key_alias': cf.key_alias}
+            return d
 
         schema = {}
         client = self.parent_connection.client
diff --git a/drivers/py/cql/decoders.py b/drivers/py/cql/decoders.py
index 9d7cad1add..f91e2b2486 100644
--- a/drivers/py/cql/decoders.py
+++ b/drivers/py/cql/decoders.py
@@ -32,45 +32,44 @@ class SchemaDecoder(object):
 
     def __comparator_for(self, keyspace, column_family):
         cfam = self.__get_column_family_def(keyspace, column_family)
-        if cfam and "comparator" in cfam:
+        if "comparator" in cfam:
             return cfam["comparator"]
         return None
 
     def __validator_for(self, keyspace, column_family, name):
         cfam = self.__get_column_family_def(keyspace, column_family)
-        if cfam:
-            if name in cfam["columns"]:
-                return cfam["columns"][name]
-            return cfam["default_validation_class"]
-        return None
+        if name in cfam["columns"]:
+            return cfam["columns"][name]
+        return cfam["default_validation_class"]
 
     def __keytype_for(self, keyspace, column_family):
         cfam = self.__get_column_family_def(keyspace, column_family)
-        if cfam and "key_validation_class" in cfam:
+        if "key_validation_class" in cfam:
             return cfam["key_validation_class"]
         return None
 
     def decode_description(self, keyspace, column_family, row):
-        key_type = self.__keytype_for(keyspace, column_family)
-        description = [(cql.ROW_KEY, key_type, None, None, None, None, None, False)]
+        description = []
         comparator = self.__comparator_for(keyspace, column_family)
         unmarshal = unmarshallers.get(comparator, unmarshal_noop)
         for column in row.columns:
-            description.append((unmarshal(column.name), comparator, None, None, None, None, True))
-
+            if column.name == self.__get_column_family_def(keyspace, column_family)['key_alias']:
+                description.append((column.name, 'text', None, None, None, None, True))
+            else:
+                description.append((unmarshal(column.name), comparator, None, None, None, None, True))
         return description
 
     def decode_row(self, keyspace, column_family, row):
-        key_type = self.__keytype_for(keyspace, column_family)
-        key = unmarshallers.get(key_type, unmarshal_noop)(row.key)
         comparator = self.__comparator_for(keyspace, column_family)
         unmarshal = unmarshallers.get(comparator, unmarshal_noop)
-        values = [key]
+        values = []
         for column in row.columns:
-            validator = self.__validator_for(keyspace, column_family, column.name)
             if column.value is None:
                 values.append(None)
+                continue
+            if column.name == self.__get_column_family_def(keyspace, column_family)['key_alias']:
+                validator = self.__keytype_for(keyspace, column_family)
             else:
-                values.append(unmarshallers.get(validator, unmarshal_noop)(column.value))
-
+                validator = self.__validator_for(keyspace, column_family, column.name)
+            values.append(unmarshallers.get(validator, unmarshal_noop)(column.value))
         return values
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index d1f8684954..3042b3da9e 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -78,6 +78,7 @@ public final class CFMetaData
     public static final CFMetaData SchemaCf = newSystemMetadata(Migration.SCHEMA_CF, 3, "current state of the schema", UTF8Type.instance, null, DEFAULT_SYSTEM_MEMTABLE_THROUGHPUT_IN_MB);
     public static final CFMetaData IndexCf = newSystemMetadata(SystemTable.INDEX_CF, 5, "indexes that have been completed", UTF8Type.instance, null, DEFAULT_SYSTEM_MEMTABLE_THROUGHPUT_IN_MB);
     public static final CFMetaData NodeIdCf = newSystemMetadata(SystemTable.NODE_ID_CF, 6, "nodeId and their metadata", TimeUUIDType.instance, null, DEFAULT_SYSTEM_MEMTABLE_THROUGHPUT_IN_MB);
+    private static final ByteBuffer DEFAULT_KEY_NAME = ByteBufferUtil.bytes("KEY");
 
     /**
      * @return A calculated memtable throughput size for this machine.
@@ -504,9 +505,9 @@ public final class CFMetaData
         return rowCacheProvider;
     }
 
-    public ByteBuffer getKeyAlias()
+    public ByteBuffer getKeyName()
     {
-        return keyAlias;
+        return keyAlias == null ? DEFAULT_KEY_NAME : keyAlias;
     }
 
     public Map<ByteBuffer, ColumnDefinition> getColumn_metadata()
@@ -640,7 +641,6 @@ public final class CFMetaData
 
         validateMinMaxCompactionThresholds(cf_def);
         validateMemtableSettings(cf_def);
-        validateAliasCompares(cf_def);
 
         CFMetaData newCFMD = new CFMetaData(cf_def.keyspace,
                                             cf_def.name,
@@ -785,7 +785,7 @@ public final class CFMetaData
         def.setMemtable_throughput_in_mb(cfm.memtableThroughputInMb);
         def.setMemtable_operations_in_millions(cfm.memtableOperationsInMillions);
         def.setMerge_shards_chance(cfm.mergeShardsChance);
-        def.setKey_alias(cfm.keyAlias);
+        def.setKey_alias(cfm.getKeyName());
         List<org.apache.cassandra.thrift.ColumnDef> column_meta = new ArrayList< org.apache.cassandra.thrift.ColumnDef>(cfm.column_metadata.size());
         for (ColumnDefinition cd : cfm.column_metadata.values())
         {
@@ -970,13 +970,6 @@ public final class CFMetaData
             DatabaseDescriptor.validateMemtableOperations(cf_def.memtable_operations_in_millions);
     }
 
-    public static void validateAliasCompares(org.apache.cassandra.thrift.CfDef cf_def) throws ConfigurationException
-    {
-        AbstractType comparator = DatabaseDescriptor.getComparator(cf_def.comparator_type);
-        if (cf_def.key_alias != null)
-            comparator.validate(cf_def.key_alias);
-    }
-
     public static void validateAliasCompares(org.apache.cassandra.db.migration.avro.CfDef cf_def) throws ConfigurationException
     {
         AbstractType comparator = DatabaseDescriptor.getComparator(cf_def.comparator_type);
diff --git a/src/java/org/apache/cassandra/cql/Cql.g b/src/java/org/apache/cassandra/cql/Cql.g
index e9bd5dbae4..9f6ab26638 100644
--- a/src/java/org/apache/cassandra/cql/Cql.g
+++ b/src/java/org/apache/cassandra/cql/Cql.g
@@ -169,8 +169,8 @@ selectExpression returns [SelectExpression expr]
       ( K_REVERSED { reversed = true; } )?
       ( first=term { $expr = new SelectExpression(first, count, reversed); }
             (',' next=term { $expr.and(next); })*
-      | start=term RANGEOP finish=term { $expr = new SelectExpression(start, finish, count, reversed); }
-      | '\*' { $expr = new SelectExpression(new Term(), new Term(), count, reversed); }
+      | start=term RANGEOP finish=term { $expr = new SelectExpression(start, finish, count, reversed, false); }
+      | '\*' { $expr = new SelectExpression(new Term(), new Term(), count, reversed, true); }
       )
     ;
 
@@ -396,7 +396,7 @@ comparatorType
     ;
 
 term returns [Term item]
-    : ( t=STRING_LITERAL | t=INTEGER | t=UUID | t=IDENT ) { $item = new Term($t.text, $t.type); }
+    : ( t=K_KEY | t=STRING_LITERAL | t=INTEGER | t=UUID | t=IDENT ) { $item = new Term($t.text, $t.type); }
     ;
 
 termList returns [List<Term> items]
diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java
index 49ce14b594..09b842fac3 100644
--- a/src/java/org/apache/cassandra/cql/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java
@@ -23,12 +23,15 @@ package org.apache.cassandra.cql;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
 import java.util.*;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
 
+import com.google.common.base.Predicates;
+import com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,25 +39,15 @@ import org.antlr.runtime.*;
 import org.apache.cassandra.auth.Permission;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.concurrent.StageManager;
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.config.ColumnDefinition;
-import org.apache.cassandra.config.ConfigurationException;
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.config.KSMetaData;
+import org.apache.cassandra.config.*;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.QueryPath;
-import org.apache.cassandra.db.migration.AddColumnFamily;
-import org.apache.cassandra.db.migration.AddKeyspace;
-import org.apache.cassandra.db.migration.DropColumnFamily;
-import org.apache.cassandra.db.migration.DropKeyspace;
-import org.apache.cassandra.db.migration.Migration;
-import org.apache.cassandra.db.migration.UpdateColumnFamily;
-import org.apache.cassandra.db.migration.avro.CfDef;
 import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.MarshalException;
-import org.apache.cassandra.dht.AbstractBounds;
-import org.apache.cassandra.dht.Bounds;
-import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.db.migration.*;
+import org.apache.cassandra.db.migration.avro.CfDef;
+import org.apache.cassandra.dht.*;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.StorageProxy;
@@ -64,9 +57,6 @@ import org.apache.cassandra.thrift.*;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
 
-import com.google.common.base.Predicates;
-import com.google.common.collect.Maps;
-
 import static org.apache.cassandra.thrift.ThriftValidation.validateColumnFamily;
 
 public class QueryProcessor
@@ -77,19 +67,13 @@ public class QueryProcessor
     throws InvalidRequestException, TimedOutException, UnavailableException
     {
         QueryPath queryPath = new QueryPath(select.getColumnFamily());
-        AbstractType<?> comparator = select.getComparator(keyspace);
-        List<ReadCommand> commands = new ArrayList<ReadCommand>();
-
-        assert select.getKeys().size() == 1;
         CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
+        List<ReadCommand> commands = new ArrayList<ReadCommand>();
 
         // ...of a list of column names
         if (!select.isColumnRange())
         {
-            Collection<ByteBuffer> columnNames = new ArrayList<ByteBuffer>();
-            for (Term column : select.getColumnNames())
-                columnNames.add(column.getByteBuffer(comparator));
-
+            Collection<ByteBuffer> columnNames = getColumnNames(select, metadata);
             validateColumnNames(columnNames);
 
             for (Term rawKey: select.getKeys())
@@ -103,6 +87,7 @@ public class QueryProcessor
         // ...a range (slice) of column names
         else
         {
+            AbstractType<?> comparator = select.getComparator(keyspace);
             ByteBuffer start = select.getColumnStart().getByteBuffer(comparator);
             ByteBuffer finish = select.getColumnFinish().getByteBuffer(comparator);
 
@@ -136,6 +121,19 @@ public class QueryProcessor
         }
     }
 
+    private static List<ByteBuffer> getColumnNames(SelectStatement select, CFMetaData metadata) throws InvalidRequestException
+    {
+        String keyString = getKeyString(metadata);
+        List<ByteBuffer> columnNames = new ArrayList<ByteBuffer>();
+        for (Term column : select.getColumnNames())
+        {
+            // skip the key for the slice op; we'll add it to the resultset in extractThriftColumns
+            if (!column.getText().equalsIgnoreCase(keyString))
+                columnNames.add(column.getByteBuffer(metadata.comparator));
+        }
+        return columnNames;
+    }
+
     private static List<org.apache.cassandra.db.Row> multiRangeSlice(String keyspace, SelectStatement select)
     throws TimedOutException, UnavailableException, InvalidRequestException
     {
@@ -153,12 +151,19 @@ public class QueryProcessor
                                 ? select.getKeyFinish().getByteBuffer(keyType)
                                 : (new Term()).getByteBuffer();
 
-        AbstractBounds bounds = new Bounds(p.getToken(startKey), p.getToken(finishKey));
+        Token startToken = p.getToken(startKey), finishToken = p.getToken(finishKey);
+        if (startToken.compareTo(finishToken) > 0 && !finishToken.equals(p.getMinimumToken()))
+        {
+            if (p instanceof RandomPartitioner)
+                throw new InvalidRequestException("Start key's md5 sorts after end key's md5. This is not allowed; you probably should not specify end key at all, under RandomPartitioner");
+            else
+                throw new InvalidRequestException("Start key must sort before (or equal to) finish key in your partitioner!");
+        }
+        AbstractBounds bounds = new Bounds(startToken, finishToken);
         
         CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
-        AbstractType<?> comparator = metadata.getComparatorFor(null);
         // XXX: Our use of Thrift structs internally makes me Sad. :(
-        SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, comparator);
+        SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, metadata);
         validateSlicePredicate(metadata, thriftSlicePredicate);
 
         int limit = select.isKeyRange() && select.getKeyStart() != null
@@ -210,16 +215,15 @@ public class QueryProcessor
     throws TimedOutException, UnavailableException, InvalidRequestException
     {
         CFMetaData metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
-        AbstractType<?> comparator = metadata.getComparatorFor(null);
         // XXX: Our use of Thrift structs internally (still) makes me Sad. :~(
-        SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, comparator);
+        SlicePredicate thriftSlicePredicate = slicePredicateFromSelect(select, metadata);
         validateSlicePredicate(metadata, thriftSlicePredicate);
         
         List<IndexExpression> expressions = new ArrayList<IndexExpression>();
         for (Relation columnRelation : select.getColumnRelations())
         {
             // Left and right side of relational expression encoded according to comparator/validator.
-            ByteBuffer entity = columnRelation.getEntity().getByteBuffer(comparator);
+            ByteBuffer entity = columnRelation.getEntity().getByteBuffer(metadata.comparator);
             ByteBuffer value = columnRelation.getValue().getByteBuffer(select.getValueValidator(keyspace, entity));
             
             expressions.add(new IndexExpression(entity,
@@ -286,7 +290,7 @@ public class QueryProcessor
         }
     }
     
-    private static SlicePredicate slicePredicateFromSelect(SelectStatement select, AbstractType<?> comparator)
+    private static SlicePredicate slicePredicateFromSelect(SelectStatement select, CFMetaData metadata)
     throws InvalidRequestException
     {
         SlicePredicate thriftSlicePredicate = new SlicePredicate();
@@ -294,18 +298,15 @@ public class QueryProcessor
         if (select.isColumnRange() || select.getColumnNames().size() == 0)
         {
             SliceRange sliceRange = new SliceRange();
-            sliceRange.start = select.getColumnStart().getByteBuffer(comparator);
-            sliceRange.finish = select.getColumnFinish().getByteBuffer(comparator);
+            sliceRange.start = select.getColumnStart().getByteBuffer(metadata.comparator);
+            sliceRange.finish = select.getColumnFinish().getByteBuffer(metadata.comparator);
             sliceRange.reversed = select.isColumnsReversed();
             sliceRange.count = select.getColumnsLimit();
             thriftSlicePredicate.slice_range = sliceRange;
         }
         else
         {
-            List<ByteBuffer> columnNames = new ArrayList<ByteBuffer>();
-            for (Term column : select.getColumnNames())
-                columnNames.add(column.getByteBuffer(comparator));
-            thriftSlicePredicate.column_names = columnNames;
+            thriftSlicePredicate.column_names = getColumnNames(select, metadata);
         }
         
         return thriftSlicePredicate;
@@ -480,19 +481,17 @@ public class QueryProcessor
         // Some statements won't have (or don't need) a keyspace (think USE, or CREATE).
         if (StatementType.requiresKeyspace.contains(statement.type))
             keyspace = clientState.getKeyspace();
-        
+
         CqlResult result = new CqlResult();
         
         logger.debug("CQL statement type: {}", statement.type.toString());
         CFMetaData metadata;
-        AbstractType<?> comparator;
         switch (statement.type)
         {
             case SELECT:
                 SelectStatement select = (SelectStatement)statement.statement;
                 clientState.hasColumnFamilyAccess(select.getColumnFamily(), Permission.READ);
                 metadata = validateColumnFamily(keyspace, select.getColumnFamily(), false);
-                comparator = metadata.getComparatorFor(null);
                 validateSelect(keyspace, select);
                 
                 List<org.apache.cassandra.db.Row> rows;
@@ -529,7 +528,7 @@ public class QueryProcessor
                 
                 List<CqlRow> cqlRows = new ArrayList<CqlRow>();
                 result.type = CqlResultType.ROWS;
-                
+
                 // Create the result set
                 for (org.apache.cassandra.db.Row row : rows)
                 {
@@ -537,7 +536,7 @@ public class QueryProcessor
                     if (row.cf == null)
                         continue;
 
-                    List<Column> thriftColumns = extractThriftColumns(select, comparator, row);
+                    List<Column> thriftColumns = extractThriftColumns(select, metadata, row);
                     // Create a new row, add the columns to it, and then add it to the list of rows
                     CqlRow cqlRow = new CqlRow();
                     cqlRow.key = row.key.key;
@@ -616,7 +615,6 @@ public class QueryProcessor
             
             case DELETE:
                 DeleteStatement delete = (DeleteStatement)statement.statement;
-
                 try
                 {
                     StorageProxy.mutate(delete.prepareRowMutations(keyspace, clientState), delete.getConsistencyLevel());
@@ -794,11 +792,17 @@ public class QueryProcessor
         return null;    // We should never get here.
     }
 
-    private static List<Column> extractThriftColumns(SelectStatement select, AbstractType<?> comparator, Row row)
+    private static List<Column> extractThriftColumns(SelectStatement select, CFMetaData metadata, Row row)
     {
         List<Column> thriftColumns = new ArrayList<Column>();
         if (select.isColumnRange())
         {
+            if (select.isWildcard())
+            {
+                // prepend key
+                thriftColumns.add(new Column(metadata.getKeyName()).setValue(row.key.key).setTimestamp(-1));
+            }
+
             // preserve comparator order
             for (IColumn c : row.cf.getSortedColumns())
             {
@@ -809,13 +813,23 @@ public class QueryProcessor
         }
         else
         {
+            String keyString = getKeyString(metadata);
+
             // order columns in the order they were asked for
             for (Term term : select.getColumnNames())
             {
+                if (term.getText().equalsIgnoreCase(keyString))
+                {
+                    // preserve case of key as it was requested
+                    ByteBuffer requestedKey = ByteBufferUtil.bytes(term.getText());
+                    thriftColumns.add(new Column(requestedKey).setValue(row.key.key).setTimestamp(-1));
+                    continue;
+                }
+
                 ByteBuffer name;
                 try
                 {
-                    name = term.getByteBuffer(comparator);
+                    name = term.getByteBuffer(metadata.comparator);
                 }
                 catch (InvalidRequestException e)
                 {
@@ -831,6 +845,20 @@ public class QueryProcessor
         return thriftColumns;
     }
 
+    private static String getKeyString(CFMetaData metadata)
+    {
+        String keyString;
+        try
+        {
+            keyString = ByteBufferUtil.string(metadata.getKeyName());
+        }
+        catch (CharacterCodingException e)
+        {
+            throw new AssertionError(e);
+        }
+        return keyString;
+    }
+
     private static CQLStatement getStatement(String queryStr) throws InvalidRequestException, RecognitionException
     {
         // Lexer and parser
diff --git a/src/java/org/apache/cassandra/cql/SelectExpression.java b/src/java/org/apache/cassandra/cql/SelectExpression.java
index 562d27d820..f2d8623aa3 100644
--- a/src/java/org/apache/cassandra/cql/SelectExpression.java
+++ b/src/java/org/apache/cassandra/cql/SelectExpression.java
@@ -37,6 +37,7 @@ public class SelectExpression
     
     private int numColumns = MAX_COLUMNS_DEFAULT;
     private boolean reverseColumns = false;
+    private final boolean wildcard;
     private Term start, finish;
     private List<Term> columns;
     
@@ -48,12 +49,13 @@ public class SelectExpression
      * @param count the number of columns to limit the results to
      * @param reverse true to reverse column order
      */
-    public SelectExpression(Term start, Term finish, int count, boolean reverse)
+    public SelectExpression(Term start, Term finish, int count, boolean reverse, boolean wildcard)
     {
         this.start = start;
         this.finish = finish;
         numColumns = count;
         reverseColumns = reverse;
+        this.wildcard = wildcard;
     }
     
     /**
@@ -65,6 +67,7 @@ public class SelectExpression
      */
     public SelectExpression(Term first, int count, boolean reverse)
     {
+        wildcard = false;
         columns = new ArrayList<Term>();
         columns.add(first);
         numColumns = count;
@@ -125,4 +128,9 @@ public class SelectExpression
     {
         return columns;
     }
+
+    public boolean isWildcard()
+    {
+        return wildcard;
+    }
 }
diff --git a/src/java/org/apache/cassandra/cql/SelectStatement.java b/src/java/org/apache/cassandra/cql/SelectStatement.java
index b5958067d8..dacba989df 100644
--- a/src/java/org/apache/cassandra/cql/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql/SelectStatement.java
@@ -81,6 +81,11 @@ public class SelectStatement
     {
         return expression.isColumnRange();
     }
+
+    public boolean isWildcard()
+    {
+        return expression.isWildcard();
+    }
     
     public List<Term> getColumnNames()
     {
diff --git a/src/java/org/apache/cassandra/thrift/ThriftValidation.java b/src/java/org/apache/cassandra/thrift/ThriftValidation.java
index f200eca054..35e042b095 100644
--- a/src/java/org/apache/cassandra/thrift/ThriftValidation.java
+++ b/src/java/org/apache/cassandra/thrift/ThriftValidation.java
@@ -26,6 +26,7 @@ import java.util.*;
 import org.apache.cassandra.config.*;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.AsciiType;
 import org.apache.cassandra.db.marshal.MarshalException;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.RandomPartitioner;
@@ -515,6 +516,22 @@ public class ThriftValidation
     {
         try
         {
+            if (cf_def.key_alias != null)
+            {
+                if (!cf_def.key_alias.hasRemaining())
+                    throw new InvalidRequestException("key_alias may not be empty");
+                try
+                {
+                    // it's hard to use a key in a select statement if we can't type it.
+                    // for now let's keep it simple and require ascii.
+                    AsciiType.instance.validate(cf_def.key_alias);
+                }
+                catch (MarshalException e)
+                {
+                    throw new InvalidRequestException("Key aliases must be ascii");
+                }
+            }
+
             ColumnFamilyType cfType = ColumnFamilyType.create(cf_def.column_type);
             if (cfType == null)
                 throw new InvalidRequestException("invalid column type " + cf_def.column_type);
diff --git a/src/java/org/apache/cassandra/tools/ClusterCmd.java b/src/java/org/apache/cassandra/tools/ClusterCmd.java
deleted file mode 100644
index b4db79188f..0000000000
--- a/src/java/org/apache/cassandra/tools/ClusterCmd.java
+++ /dev/null
@@ -1,272 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.cassandra.tools;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.util.List;
-
-import org.apache.commons.cli.*;
-
-/**
- * JMX cluster wide operations for Cassandra.
- */
-public class ClusterCmd {
-    private static final String HOST_OPT_LONG = "host";
-    private static final String HOST_OPT_SHORT = "h";
-    private static final String PORT_OPT_LONG = "port";
-    private static final String PORT_OPT_SHORT = "p";
-    private static final int defaultPort = 7199;
-    private static Options options = null;
-    private CommandLine cmd = null;
-    private NodeProbe probe;
-
-    static
-    {
-        options = new Options();
-        Option optHost = new Option(HOST_OPT_SHORT, HOST_OPT_LONG, true, "node hostname or ip address");
-        optHost.setRequired(true);
-        options.addOption(optHost);
-        options.addOption(PORT_OPT_SHORT, PORT_OPT_LONG, true, "remote jmx agent port number (defaults to " + defaultPort + ")");
-    }
-
-    /**
-     * Creates a ClusterProbe using command-line arguments.
-     *
-     * @param cmdArgs list of arguments passed on the command line
-     * @throws ParseException for missing required, or unrecognized options
-     * @throws IOException on connection failures
-     */
-    private ClusterCmd(String[] cmdArgs) throws ParseException, IOException, InterruptedException
-    {
-        parseArgs(cmdArgs);
-        String host = cmd.getOptionValue(HOST_OPT_SHORT);
-
-        String portNum = cmd.getOptionValue(PORT_OPT_SHORT);
-        int port;
-        if (portNum != null)
-        {
-            try
-            {
-                port = Integer.parseInt(portNum);
-            }
-            catch (NumberFormatException e)
-            {
-                throw new ParseException("Port must be a number");
-            }
-        }
-        else
-        {
-            port = defaultPort;
-        }
-
-        probe = new NodeProbe(host, port);
-    }
-
-    /**
-     * Creates a ClusterProbe using the specified JMX host and port.
-     *
-     * @param host hostname or IP address of the JMX agent
-     * @param port TCP port of the remote JMX agent
-     * @throws IOException on connection failures
-     */
-    public ClusterCmd(String host, int port) throws IOException, InterruptedException
-    {
-        probe = new NodeProbe(host, port);
-    }
-
-    /**
-     * Creates a ClusterProbe using the specified JMX host and default port.
-     *
-     * @param host hostname or IP address of the JMX agent
-     * @throws IOException on connection failures
-     */
-    public ClusterCmd(String host) throws IOException, InterruptedException
-    {
-        this(host, defaultPort);
-    }
-
-    /**
-     * Parse the supplied command line arguments.
-     *
-     * @param args arguments passed on the command line
-     * @throws ParseException for missing required, or unrecognized options
-     */
-    private void parseArgs(String[] args) throws ParseException
-    {
-        CommandLineParser parser = new PosixParser();
-        cmd = parser.parse(options, args);
-    }
-
-    /**
-     * Retrieve any non-option arguments passed on the command line.
-     *
-     * @return non-option command args
-     */
-    private String[] getArgs()
-    {
-        return cmd.getArgs();
-    }
-
-    /**
-     * Prints usage information to stdout.
-     */
-    private static void printUsage()
-    {
-        HelpFormatter hf = new HelpFormatter();
-        String header = String.format(
-                "%nAvailable commands: get_endpoints [keyspace] [key], global_snapshot [name], clear_global_snapshot," +
-                "truncate <keyspace> <cfname>");
-        String usage = String.format("java %s -host <arg> <command>%n", ClusterCmd.class.getName());
-        hf.printHelp(usage, "", options, header);
-    }
-    
-    public void close() throws IOException
-    {
-        probe.close(); 
-    }
-
-    public void printEndpoints(String keyspace, String key)
-    {
-        List<InetAddress> endpoints = probe.getEndpoints(keyspace, key);
-        System.out.println(String.format("%-17s: %s", "Key", key));
-        System.out.println(String.format("%-17s: %s", "Endpoints", endpoints));
-    }
-
-     /**
-     * Take a snapshot of all tables on all (live) nodes in the cluster
-     *
-     * @param snapshotName name of the snapshot
-     */
-    public void takeGlobalSnapshot(String snapshotName) throws IOException, InterruptedException
-    {
-
-        for (String liveNode : probe.getLiveNodes())
-        {
-            try
-            {
-                NodeProbe hostProbe = new NodeProbe(liveNode, probe.port);
-                hostProbe.takeSnapshot(snapshotName);
-                System.out.println(liveNode + " snapshot taken in directory: " + snapshotName);
-            }
-            catch (IOException e)
-            {
-                System.out.println(liveNode + " snapshot FAILED: " + e.getMessage());
-            }
-        }
-    }
-
-    /**
-     * Remove all the existing snapshots from all (live) nodes in the cluster
-     */
-    public void clearGlobalSnapshot(String tag) throws IOException, InterruptedException
-    {
-        for (String liveNode : probe.getLiveNodes())
-        {
-            try
-            {
-                NodeProbe hostProbe = new NodeProbe(liveNode, probe.port);
-                hostProbe.clearSnapshot(tag);
-                System.out.println(liveNode + " snapshot cleared");
-            }
-            catch (IOException e)
-            {
-                System.out.println(liveNode + " snapshot clear FAILED: " + e.getMessage());
-            }
-        }
-    }
-
-    public void truncate(String tableName, String cfName)
-    {
-        probe.truncate(tableName, cfName);
-    }
-
-    /**
-     * @param args
-     */
-    public static void main(String[] args) throws IOException, InterruptedException
-    {
-        ClusterCmd clusterCmd = null;
-        try
-        {
-            clusterCmd = new ClusterCmd(args);
-        }
-        catch (ParseException pe)
-        {
-            System.err.println(pe.getMessage());
-            ClusterCmd.printUsage();
-            System.exit(1);
-        }
-        catch (IOException ioe)
-        {
-            System.err.println("Error connecting to remote JMX agent!");
-            ioe.printStackTrace();
-            System.exit(3);
-        }
-
-        if (clusterCmd.getArgs().length < 1)
-        {
-            System.err.println("Missing argument for command.");
-            ClusterCmd.printUsage();
-            System.exit(1);
-        }
-
-        // Execute the requested command.
-        String[] arguments = clusterCmd.getArgs();
-        String cmdName = arguments[0];
-        if (cmdName.equals("get_endpoints"))
-        {
-            if (arguments.length <= 2)
-            {
-                System.err.println("missing keyspace and/or key argument");
-                ClusterCmd.printUsage();
-                System.exit(1);
-            }
-            clusterCmd.printEndpoints(arguments[1], arguments[2]);
-        }
-        else if (cmdName.equals("global_snapshot"))
-        {
-            String snapshotName = arguments.length > 1 ? arguments[1] : new Long(System.currentTimeMillis()).toString();
-            clusterCmd.takeGlobalSnapshot(snapshotName);
-        }
-        else if (cmdName.equals("clear_global_snapshot"))
-        {
-            String snapshotName = arguments.length > 1 ? arguments[1] : "";
-            clusterCmd.clearGlobalSnapshot(snapshotName);
-        }
-        else if (cmdName.equals("truncate"))
-        {
-            if (arguments.length != 3)
-            {
-                System.err.println("truncate requires <keyspace> and <columnfamily> arguments");
-            }
-            String tableName = arguments[1];
-            String cfName = arguments[2];
-            clusterCmd.truncate(tableName, cfName);
-        }
-        else
-        {
-            System.err.println("Unrecognized command: " + cmdName + ".");
-            ClusterCmd.printUsage();
-            System.exit(1);
-        }
-
-        System.exit(0);
-    }
-    
-}
diff --git a/test/system/test_cql.py b/test/system/test_cql.py
index db8d984048..9c326a1152 100644
--- a/test/system/test_cql.py
+++ b/test/system/test_cql.py
@@ -127,11 +127,11 @@ class TestCql(ThriftTester):
     def test_select_simple(self):
         "single-row named column queries"
         cursor = init()
-        cursor.execute("SELECT 'ca1' FROM StandardString1 WHERE KEY='ka'")
+        cursor.execute("SELECT KEY, ca1 FROM StandardString1 WHERE KEY='ka'")
         r = cursor.fetchone()
         d = cursor.description
 
-        assert d[0][0] == cql.ROW_KEY
+        assert d[0][0] == 'KEY'
         assert r[0] == 'ka'
 
         assert d[1][0] == 'ca1'
@@ -144,10 +144,10 @@ class TestCql(ThriftTester):
         """)
 
         d = cursor.description
-        assert ['Row Key', 'ca1', 'col', 'cd1'] == [col_dscptn[0] for col_dscptn in d], d
+        assert ['ca1', 'col', 'cd1'] == [col_dscptn[0] for col_dscptn in d], d
         row = cursor.fetchone()
         # check that the column that didn't exist in the row comes back as null
-        assert ['kd', None, 'val', 'vd1'] == row, row
+        assert [None, 'val', 'vd1'] == row, row
 
     def test_select_row_range(self):
         "retrieve a range of rows with columns"
@@ -219,51 +219,39 @@ class TestCql(ThriftTester):
         "column slice tests"
         cursor = init()
 
-        # all columns
+        # * includes row key, explicit slice does not
         cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'ka';")
-        r = cursor.fetchone()
-        assert len(r) == 3
+        row = cursor.fetchone()
+        assert ['ka', 'va1', 'val'] == row, row
+
         cursor.execute("SELECT ''..'' FROM StandardString1 WHERE KEY = 'ka';")
-        r = cursor.fetchone()
-        assert len(r) == 3
+        row = cursor.fetchone()
+        assert ['va1', 'val'] == row, row
 
         # column subsets
         cursor.execute("SELECT 1..3 FROM StandardLongA WHERE KEY = 'aa';")
         assert cursor.rowcount == 1
-        r = cursor.fetchone()
-        assert r[0] == "aa"
-        assert r[1] == "1"
-        assert r[2] == "2"
-        assert r[3] == "3"
+        row = cursor.fetchone()
+        assert ['1', '2', '3'] == row, row
         
-        cursor.execute("SELECT 10..30 FROM StandardIntegerA WHERE KEY='k1'")
-        assert cursor.rowcount == 1
-        r = cursor.fetchone()
-        assert r[0] == "k1"
-        assert r[1] == "a"
-        assert r[2] == "b"
-        assert r[3] == "c"
+        cursor.execute("""
+            SELECT key,20,40 FROM StandardIntegerA
+            WHERE KEY > 'k1' AND KEY < 'k7' LIMIT 5
+        """)
+        row = cursor.fetchone()
+        assert ['k2', 'f', 'h'] == row, row
 
         # range of columns (slice) by row with FIRST
-        cursor.execute("""
-            SELECT FIRST 1 1..3 FROM StandardLongA WHERE KEY = 'aa';
-        """)
+        cursor.execute("SELECT FIRST 1 1..3 FROM StandardLongA WHERE KEY = 'aa'")
         assert cursor.rowcount == 1
-        r = cursor.fetchone()
-        assert len(r) == 2
-        assert r[0] == "aa"
-        assert r[1] == "1"
+        row = cursor.fetchone()
+        assert ['1'] == row, row
 
         # range of columns (slice) by row reversed
-        cursor.execute("""
-            SELECT FIRST 2 REVERSED 3..1 FROM StandardLongA WHERE KEY = 'aa';
-        """)
+        cursor.execute("SELECT FIRST 2 REVERSED 3..1 FROM StandardLongA WHERE KEY = 'aa'")
         assert cursor.rowcount == 1, "%d != 1" % cursor.rowcount
-        r = cursor.fetchone()
-        assert len(r) == 3
-        assert r[0] == 'aa'
-        assert r[1] == "3"
-        assert r[2] == "2"
+        row = cursor.fetchone()
+        assert ['3', '2'] == row, row
 
     def test_select_range_with_single_column_results(self):
         "range should not fail when keys were not set"
@@ -277,7 +265,7 @@ class TestCql(ThriftTester):
         """)
 
         cursor.execute("""
-          SELECT name FROM StandardString2
+          SELECT KEY, name FROM StandardString2
         """)
 
         assert cursor.rowcount == 3, "expected 3 results, got %d" % cursor.rowcount
@@ -298,7 +286,7 @@ class TestCql(ThriftTester):
         "indexed scan where column equals value"
         cursor = init()
         cursor.execute("""
-            SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
+            SELECT KEY, birthdate FROM IndexedA WHERE birthdate = 100
         """)
         assert cursor.rowcount == 2
 
@@ -314,19 +302,19 @@ class TestCql(ThriftTester):
         "indexed scan where a column is greater than a value"
         cursor = init()
         cursor.execute("""
-            SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
-                    AND 'unindexed' > 200
+            SELECT KEY, 'birthdate' FROM IndexedA 
+            WHERE 'birthdate' = 100 AND 'unindexed' > 200
         """)
         assert cursor.rowcount == 1
-        r = cursor.fetchone()
-        assert r[0] == "asmith"
+        row = cursor.fetchone()
+        assert row[0] == "asmith", row
 
     def test_index_scan_with_start_key(self):
         "indexed scan with a starting key"
         cursor = init()
         cursor.execute("""
-            SELECT 'birthdate' FROM IndexedA WHERE 'birthdate' = 100
-                    AND KEY >= 'asmithZ'
+            SELECT KEY, 'birthdate' FROM IndexedA 
+            WHERE 'birthdate' = 100 AND KEY >= 'asmithZ'
         """)
         assert cursor.rowcount == 1
         r = cursor.fetchone()
@@ -335,7 +323,7 @@ class TestCql(ThriftTester):
     def test_no_where_clause(self):
         "empty where clause (range query w/o start key)"
         cursor = init()
-        cursor.execute("SELECT 'col' FROM StandardString1 LIMIT 3")
+        cursor.execute("SELECT KEY, 'col' FROM StandardString1 LIMIT 3")
         assert cursor.rowcount == 3
         rows = cursor.fetchmany(3)
         assert rows[0][0] == "ka"
@@ -369,7 +357,8 @@ class TestCql(ThriftTester):
         cursor.execute("""
             SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
         """)
-        assert ['Row Key', 'cd1', 'col'] == [col_d[0] for col_d in cursor.description]
+        desc = [col_d[0] for col_d in cursor.description]
+        assert ['cd1', 'col'] == desc, desc
 
         cursor.execute("""
             DELETE 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
@@ -377,31 +366,31 @@ class TestCql(ThriftTester):
         cursor.execute("""
             SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
         """)
-        r = cursor.fetchone()
-        assert ['kd', None, None] == r, r
+        row = cursor.fetchone()
+        assert [None, None] == row, row
 
     def test_delete_columns_multi_rows(self):
         "delete columns from multiple rows"
         cursor = init()
 
+        # verify rows exist initially
         cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
-        r = cursor.fetchone()
-        assert ['kc', 'val'] == r, r
-
+        row = cursor.fetchone()
+        assert ['val'] == row, row
         cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
-        r = cursor.fetchone()
-        assert ['kd', 'val'] == r, r
+        row = cursor.fetchone()
+        assert ['val'] == row, row
 
+        # delete and verify data is gone
         cursor.execute("""
             DELETE 'col' FROM StandardString1 WHERE KEY IN ('kc', 'kd')
         """)
         cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kc'")
-        r = cursor.fetchone()
-        assert ['kc', None] == r, r
-
+        row = cursor.fetchone()
+        assert [None] == row, row
         cursor.execute("SELECT 'col' FROM StandardString1 WHERE KEY = 'kd'")
         r = cursor.fetchone()
-        assert ['kd', None] == r, r
+        assert [None] == r, r
 
     def test_delete_rows(self):
         "delete entire rows"
@@ -409,13 +398,13 @@ class TestCql(ThriftTester):
         cursor.execute("""
             SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
         """)
-        assert ['Row Key', 'cd1', 'col'] == [col_d[0] for col_d in cursor.description]
+        assert ['cd1', 'col'] == [col_d[0] for col_d in cursor.description]
         cursor.execute("DELETE FROM StandardString1 WHERE KEY = 'kd'")
         cursor.execute("""
             SELECT 'cd1', 'col' FROM StandardString1 WHERE KEY = 'kd'
         """)
-        r = cursor.fetchone()
-        assert ['kd', None, None] == r, r
+        row = cursor.fetchone()
+        assert [None, None] == row, row
 
     def test_create_keyspace(self):
         "create a new keyspace"
@@ -571,7 +560,7 @@ class TestCql(ThriftTester):
             SELECT '%s' FROM StandardTimeUUID WHERE KEY = 'uuidtest'
         """ % str(timeuuid))
         d = cursor.description
-        assert d[1][0] == timeuuid, "%s, %s" % (str(d[1][0]), str(timeuuid))
+        assert d[0][0] == timeuuid, "%s, %s" % (str(d[1][0]), str(timeuuid))
 
         # Tests a node-side conversion from bigint to UUID.
         ms = uuid1bytes_to_millis(uuid.uuid1().bytes)
@@ -583,7 +572,7 @@ class TestCql(ThriftTester):
             SELECT 'id' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
         """)
         r = cursor.fetchone()
-        assert uuid1bytes_to_millis(r[1].bytes) == ms
+        assert uuid1bytes_to_millis(r[0].bytes) == ms
 
         # Tests a node-side conversion from ISO8601 to UUID.
         cursor.execute("""
@@ -596,7 +585,7 @@ class TestCql(ThriftTester):
         """)
         # 2011-01-31 17:00:00-0000 == 1296493200000ms
         r = cursor.fetchone()
-        ms = uuid1bytes_to_millis(r[1].bytes)
+        ms = uuid1bytes_to_millis(r[0].bytes)
         assert ms == 1296493200000, \
                 "%d != 1296493200000 (2011-01-31 17:00:00-0000)" % ms
 
@@ -610,7 +599,7 @@ class TestCql(ThriftTester):
             SELECT 'id3' FROM StandardTimeUUIDValues WHERE KEY = 'uuidtest'
         """)
         r = cursor.fetchone()
-        ms = uuid1bytes_to_millis(r[1].bytes)
+        ms = uuid1bytes_to_millis(r[0].bytes)
         assert ((time.time() * 1e3) - ms) < 100, \
             "new timeuuid not within 100ms of now (UPDATE vs. SELECT)"
 
@@ -624,7 +613,7 @@ class TestCql(ThriftTester):
             SELECT :start..:finish FROM StandardTimeUUID WHERE KEY = slicetest
             """, dict(start=uuid_range[0], finish=uuid_range[len(uuid_range)-1]))
         d = cursor.description
-        for (i, col_d) in enumerate(d[1:]):
+        for (i, col_d) in enumerate(d):
             assert uuid_range[i] == col_d[0]
 
 
@@ -638,7 +627,7 @@ class TestCql(ThriftTester):
         cursor.execute("SELECT :name FROM StandardUUID WHERE KEY = 'uuidtest'",
                        dict(name=uid))
         d = cursor.description
-        assert d[1][0] == uid, "expected %s, got %s (%s)" % \
+        assert d[0][0] == uid, "expected %s, got %s (%s)" % \
                 (uid.bytes.encode('hex'), str(d[1][0]).encode('hex'), d[1][1])
 
         # TODO: slices of uuids from cf w/ LexicalUUIDType comparator
@@ -654,18 +643,19 @@ class TestCql(ThriftTester):
 
         cursor.execute("SELECT * FROM StandardUtf82 WHERE KEY = k1")
         d = cursor.description
+        assert d[0][0] == 'KEY', d[0][0]
         assert d[1][0] == u"¢", d[1][0]
         assert d[2][0] == u"©", d[2][0]
         assert d[3][0] == u"®", d[3][0]
         assert d[4][0] == u"¿", d[4][0]
 
         cursor.execute("SELECT :start..'' FROM StandardUtf82 WHERE KEY = k1", dict(start="©"))
-        r = cursor.fetchone()
-        assert len(r) == 4
+        row = cursor.fetchone()
+        assert len(row) == 3, row
         d = cursor.description
-        assert d[1][0] == u"©"
-        assert d[2][0] == u"®"
-        assert d[3][0] == u"¿"
+        assert d[0][0] == u"©"
+        assert d[1][0] == u"®"
+        assert d[2][0] == u"¿"
 
     def test_read_write_negative_numerics(self):
         "reading and writing negative numeric values"
@@ -678,11 +668,11 @@ class TestCql(ThriftTester):
             cursor.execute("SELECT :start..:finish FROM :cf WHERE KEY = negatives;",
                            dict(start=-10, finish=-1, cf=cf))
             r = cursor.fetchone()
-            assert len(r) == 11, \
+            assert len(r) == 10, \
                 "returned %d columns, expected %d" % (len(r) - 1, 10)
             d = cursor.description
-            assert d[1][0] == -10
-            assert d[10][0] == -1
+            assert d[0][0] == -10
+            assert d[9][0] == -1
 
     def test_escaped_quotes(self):
         "reading and writing strings w/ escaped quotes"
@@ -697,17 +687,17 @@ class TestCql(ThriftTester):
                        """, dict(key="test_escaped_quotes"))
         assert cursor.rowcount == 1
         r = cursor.fetchone()
-        assert len(r) == 2, "wrong number of results"
+        assert len(r) == 1, "wrong number of results"
         d = cursor.description
-        assert d[1][0] == "x\'and\'y"
-
+        assert d[0][0] == "x'and'y"
+        
     def test_typed_keys(self):
         "using typed keys"
         cursor = init()
         cursor.execute("SELECT * FROM StandardString1 WHERE KEY = :key", dict(key="ka"))
-        r = cursor.fetchone()
-        assert isinstance(r[0], unicode), \
-            "wrong key-type returned, expected unicode, got %s" % type(r[0])
+        row = cursor.fetchone()
+        assert isinstance(row[0], unicode), \
+            "wrong key-type returned, expected unicode, got %s" % type(row[0])
 
         # FIXME: The above is woefully inadequate, but the test config uses
         # CollatingOrderPreservingPartitioner which only supports UTF8.
@@ -753,11 +743,9 @@ class TestCql(ThriftTester):
 
         assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
         colnames = [col_d[0] for col_d in cursor.description]
-        assert colnames[1] == "some_name", \
-               "unrecognized name '%s'" % colnames[1]
-        r = cursor.fetchone()
-        assert r[1] == "some_value", \
-               "unrecognized value '%s'" % r[1]
+        assert ['some_name'] == colnames, colnames
+        row = cursor.fetchone()
+        assert ['some_value'] == row, row
 
     def test_batch_with_mixed_statements(self):
         "handle BATCH with INSERT/UPDATE/DELETE statements mixed in it"
@@ -778,34 +766,28 @@ class TestCql(ThriftTester):
 
         assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
         colnames = [col_d[0] for col_d in cursor.description]
-        assert colnames[1] == "bName", \
-               "unrecognized name '%s'" % colnames[1]
+        assert ['bName'] == colnames, colnames
         r = cursor.fetchone()
-        assert r[1] == "bVal", \
-               "unrecognized value '%s'" % r[1]
+        assert ['bVal'] == r, r
 
         cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key",
                        dict(name="bCol2", key="bKey3"))
 
         assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
         colnames = [col_d[0] for col_d in cursor.description]
-        assert colnames[1] == "bCol2", \
-               "unrecognized name '%s'" % colnames[1]
+        assert ['bCol2'] == colnames, colnames
         # was deleted by DELETE statement
         r = cursor.fetchone()
-        assert r[1] == None, \
-               "unrecognized value '%s'" % r[1]
+        assert [None] == r, r
 
         cursor.execute("SELECT :name FROM StandardString1 WHERE KEY = :key",
                        dict(name="bCol1", key="bKey2"))
 
         assert cursor.rowcount == 1, "expected 1 result, got %d" % cursor.rowcount
         colnames = [col_d[0] for col_d in cursor.description]
-        assert colnames[1] == "bCol1", \
-               "unrecognized name '%s'" % colnames[1]
+        assert ['bCol1'] == colnames, colnames
         r = cursor.fetchone()
-        assert r[1] == "bVal", \
-               "unrecognized value '%s'" % r[1]
+        assert ['bVal'] == r, r
 
         # using all 3 types of statements allowed in batch to test timestamp
         cursor.execute("""
@@ -929,5 +911,3 @@ class TestCql(ThriftTester):
         r = cursor.fetchone()
         assert r[1] == "name here", \
                "unrecognized value '%s'" % r[1]
-
-