diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index 6c1f2a1d42..7494ceb97b 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -83,25 +83,22 @@
        ~ An optional `Comment` attribute may be used to attach additional
        ~ human-readable information about the column family to its definition.
        ~ 
-       ~ The optional KeysCachedFraction attribute specifies
-       ~ The fraction of keys per sstable whose locations we keep in
+       ~ The optional KeysCached attribute specifies
+       ~ the number of keys per sstable whose locations we keep in
        ~ memory in "mostly LRU" order.  (JUST the key locations, NOT any
-       ~ column values.) The amount of memory used by the default setting of 
-       ~ 0.01 is comparable to the amount used by the internal per-sstable key
-       ~ index. Consider increasing this if you have fewer, wider rows.
-       ~ Set to 0 to disable entirely.
+       ~ column values.) Specify a fraction (value less than 1), a percentage
+       ~ (ending in a % sign) or an absolute number of keys to cache. 
        ~
        ~ The optional RowsCached attribute specifies the number of rows
-       ~ whose entire contents we cache in memory, either as a fixed number
-       ~ of rows or as a percent of rows in the ColumnFamily.  
-       ~ Do not use this on ColumnFamilies with large rows, or
-       ~ ColumnFamilies with high write:read ratios.  As with key caching,
-       ~ valid values are from 0 to 1.  The default 0 disables it entirely.
+       ~ whose entire contents we cache in memory. Do not use this on
+       ~ ColumnFamilies with large rows, or ColumnFamilies with high write:read
+       ~ ratios. Specify a fraction (value less than 1), a percentage (ending in
+       ~ a % sign) or an absolute number of rows to cache. 
       -->
       <ColumnFamily CompareWith="BytesType" 
                     Name="Standard1" 
                     RowsCached="10%"
-                    KeysCachedFraction="0"/>
+                    KeysCached="0"/>
       <ColumnFamily CompareWith="UTF8Type" Name="Standard2"/>
       <ColumnFamily CompareWith="TimeUUIDType" Name="StandardByUUID1"/>
       <ColumnFamily ColumnType="Super"
@@ -109,7 +106,7 @@
                     CompareSubcolumnsWith="UTF8Type"
                     Name="Super1"
                     RowsCached="1000"
-                    KeysCachedFraction="0"
+                    KeysCached="50%"
                     Comment="A column family with supercolumns, whose column and subcolumn names are UTF8 strings"/>
 
       <!--
diff --git a/src/java/org/apache/cassandra/config/CFMetaData.java b/src/java/org/apache/cassandra/config/CFMetaData.java
index 778b7bb5e4..aea45b522a 100644
--- a/src/java/org/apache/cassandra/config/CFMetaData.java
+++ b/src/java/org/apache/cassandra/config/CFMetaData.java
@@ -29,6 +29,9 @@ import java.io.InputStream;
 
 public final class CFMetaData
 {
+    public final static double DEFAULT_KEY_CACHE_SIZE = 0.1;
+    public final static double DEFAULT_ROW_CACHE_SIZE = 0.0;
+
     public final String tableName;            // name of table which has this column family
     public final String cfName;               // name of the column family
     public final String columnType;           // type: super, standard, etc.
@@ -36,9 +39,9 @@ public final class CFMetaData
     public final AbstractType subcolumnComparator; // like comparator, for supercolumns
     public final String comment; // for humans only
     public final double rowCacheSize; // default 0
-    public final double keysCachedFraction; // default 0.01
+    public final double keyCacheSize; // default 0.01
 
-    CFMetaData(String tableName, String cfName, String columnType, AbstractType comparator, AbstractType subcolumnComparator, String comment, double rowCacheSize, double keysCachedFraction)
+    CFMetaData(String tableName, String cfName, String columnType, AbstractType comparator, AbstractType subcolumnComparator, String comment, double rowCacheSize, double keyCacheSize)
     {
         this.tableName = tableName;
         this.cfName = cfName;
@@ -47,7 +50,7 @@ public final class CFMetaData
         this.subcolumnComparator = subcolumnComparator;
         this.comment = comment;
         this.rowCacheSize = rowCacheSize;
-        this.keysCachedFraction = keysCachedFraction;
+        this.keyCacheSize = keyCacheSize;
     }
 
     // a quick and dirty pretty printer for describing the column family...
@@ -73,7 +76,7 @@ public final class CFMetaData
         if (cfm.comment != null)
             dout.writeUTF(cfm.comment);
         dout.writeDouble(cfm.rowCacheSize);
-        dout.writeDouble(cfm.keysCachedFraction);
+        dout.writeDouble(cfm.keyCacheSize);
         dout.close();
         return bout.toByteArray();
     }
@@ -105,8 +108,8 @@ public final class CFMetaData
         }
         String comment = din.readBoolean() ? din.readUTF() : null;
         double rowCacheSize = din.readDouble();
-        double keysCachedFraction = din.readDouble();
-        return new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keysCachedFraction);
+        double keyCacheSize = din.readDouble();
+        return new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keyCacheSize);
     }
 
     public boolean equals(Object obj)
@@ -121,6 +124,6 @@ public final class CFMetaData
                 && FBUtilities.equals(other.subcolumnComparator, subcolumnComparator)
                 && FBUtilities.equals(other.comment, comment)
                 && other.rowCacheSize == rowCacheSize
-                && other.keysCachedFraction == keysCachedFraction;
+                && other.keyCacheSize == keyCacheSize;
     }
 }
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 657792d8c1..fe61579d38 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -26,10 +26,10 @@ import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.db.marshal.BytesType;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.locator.EndPointSnitch;
 import org.apache.cassandra.locator.IEndPointSnitch;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.XMLUtils;
 import org.apache.log4j.Logger;
 import org.w3c.dom.Node;
@@ -457,8 +457,8 @@ public class DatabaseDescriptor
                                                                             new UTF8Type(),
                                                                             null,
                                                                             "persistent metadata for the local node",
-                                                                            0d,
-                                                                            0.01d));
+                                                                            0.0,
+                                                                            0.01));
 
             systemMeta.cfMetaData.put(HintedHandOffManager.HINTS_CF, new CFMetaData(Table.SYSTEM_TABLE,
                                                                                     HintedHandOffManager.HINTS_CF,
@@ -466,8 +466,8 @@ public class DatabaseDescriptor
                                                                                     new UTF8Type(),
                                                                                     new BytesType(),
                                                                                     "hinted handoff data",
-                                                                                    0d,
-                                                                                    0.01d));
+                                                                                    0.0,
+                                                                                    0.01));
 
             /* Load the seeds for node contact points */
             String[] seeds = xmlUtils.getNodeValues("/Storage/Seeds/Seed");
@@ -642,23 +642,22 @@ public class DatabaseDescriptor
                         throw new ConfigurationException("CompareSubcolumnsWith is only a valid attribute on super columnfamilies (not regular columnfamily " + cfName + ")");
                     }
 
-                    double keysCachedFraction = 0.01d;
+                    double keyCacheSize = CFMetaData.DEFAULT_KEY_CACHE_SIZE;
                     if ((value = XMLUtils.getAttributeValue(columnFamily, "KeysCachedFraction")) != null)
                     {
-                        keysCachedFraction = Double.valueOf(value);
+                        keyCacheSize = Double.valueOf(value);
+                        // TODO: KeysCachedFraction deprecated: remove in 1.0
+                        logger_.warn("KeysCachedFraction is deprecated: use KeysCached instead.");
+                    }
+                    if ((value = XMLUtils.getAttributeValue(columnFamily, "KeysCached")) != null)
+                    {
+                        keyCacheSize = FBUtilities.parseDoubleOrPercent(value);
                     }
 
-                    double rowCacheSize = 0;
+                    double rowCacheSize = CFMetaData.DEFAULT_ROW_CACHE_SIZE;
                     if ((value = XMLUtils.getAttributeValue(columnFamily, "RowsCached")) != null)
                     {
-                        if (value.endsWith("%"))
-                        {
-                            rowCacheSize = Double.valueOf(value.substring(0, value.length() - 1)) / 100;
-                        }
-                        else
-                        {
-                            rowCacheSize = Double.valueOf(value);
-                        }
+                        rowCacheSize = FBUtilities.parseDoubleOrPercent(value);
                     }
 
                     // Parse out user-specified logical names for the various dimensions
@@ -666,7 +665,7 @@ public class DatabaseDescriptor
                     String comment = xmlUtils.getNodeValue(xqlCF + "Comment");
 
                     // insert it into the table dictionary.
-                    meta.cfMetaData.put(cfName, new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keysCachedFraction));
+                    meta.cfMetaData.put(cfName, new CFMetaData(tableName, cfName, columnType, comparator, subcolumnComparator, comment, rowCacheSize, keyCacheSize));
                 }
 
                 tables_.put(meta.name, meta);
@@ -1085,20 +1084,24 @@ public class DatabaseDescriptor
         return getCFMetaData(tableName, cfName).subcolumnComparator;
     }
 
-    public static double getKeysCachedFraction(String tableName, String columnFamilyName)
+    /**
+     * @return The absolute number of keys that should be cached per table.
+     */
+    public static int getKeysCachedFor(String tableName, String columnFamilyName, long expectedKeys)
     {
         CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
-        if (cfm == null)
-            return 0.01d;
-        return cfm.keysCachedFraction;
+        double v = (cfm == null) ? CFMetaData.DEFAULT_KEY_CACHE_SIZE : cfm.keyCacheSize;
+        return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedKeys), Integer.MAX_VALUE);
     }
 
-    public static double getRowsCachedFraction(String tableName, String columnFamilyName)
+    /**
+     * @return The absolute number of rows that should be cached for the columnfamily.
+     */
+    public static int getRowsCachedFor(String tableName, String columnFamilyName, long expectedRows)
     {
         CFMetaData cfm = getCFMetaData(tableName, columnFamilyName);
-        if (cfm == null)
-            return 0.01d;
-        return cfm.rowCacheSize;
+        double v = (cfm == null) ? CFMetaData.DEFAULT_ROW_CACHE_SIZE : cfm.rowCacheSize;
+        return (int)Math.min(FBUtilities.absoluteFromFraction(v, expectedRows), Integer.MAX_VALUE);
     }
 
     private static class ConfigurationException extends Exception
diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java
index b6d6c33f99..6858478b7c 100644
--- a/src/java/org/apache/cassandra/db/BinaryMemtable.java
+++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java
@@ -128,7 +128,7 @@ public class BinaryMemtable implements IFlushable
             assert bytes.length > 0;
             writer.append(key, bytes);
         }
-        SSTableReader sstable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(cfs.getTable().name, cfs.getColumnFamilyName()));
+        SSTableReader sstable = writer.closeAndOpenReader();
         logger.info("Completed flushing " + writer.getFilename());
         return sstable;
     }
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 3f5f40d476..1b7e444507 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -18,51 +18,45 @@
 
 package org.apache.cassandra.db;
 
+import java.io.Closeable;
 import java.io.File;
 import java.io.IOException;
-import java.io.Closeable;
 import java.lang.management.ManagementFactory;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.locks.Condition;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
 
-import com.google.common.collect.AbstractIterator;
-import com.google.common.collect.Iterables;
-import org.apache.cassandra.cache.IAggregatableCacheProvider;
-import org.apache.cassandra.cache.InstrumentedCache;
-import org.apache.cassandra.cache.JMXAggregatingCache;
-import org.apache.cassandra.cache.JMXInstrumentedCache;
 import org.apache.log4j.Logger;
+import org.apache.commons.collections.IteratorUtils;
 
+import com.google.common.base.Predicate;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Iterators;
+import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
+import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.db.commitlog.CommitLogSegment;
+import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.marshal.AbstractType;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Bounds;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.io.*;
+import org.apache.cassandra.io.SSTable;
+import org.apache.cassandra.io.SSTableReader;
+import org.apache.cassandra.io.SSTableScanner;
+import org.apache.cassandra.io.SSTableTracker;
 import org.apache.cassandra.io.util.FileUtils;
-
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.thrift.SliceRange;
 import org.apache.cassandra.utils.*;
-import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
-import org.apache.cassandra.concurrent.NamedThreadFactory;
-import org.apache.cassandra.db.filter.*;
-import org.apache.cassandra.db.marshal.AbstractType;
-
-import org.apache.commons.collections.IteratorUtils;
-
-import com.google.common.collect.Iterators;
-import com.google.common.base.Predicate;
 
 public class ColumnFamilyStore implements ColumnFamilyStoreMBean
 {
@@ -115,8 +109,6 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     /* active memtable associated with this ColumnFamilyStore. */
     private Memtable memtable_;
 
-    private final JMXInstrumentedCache<String, ColumnFamily> rowCache;
-
     // TODO binarymemtable ops are not threadsafe (do they need to be?)
     private AtomicReference<BinaryMemtable> binaryMemtable_;
 
@@ -190,51 +182,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             }
             sstables.add(sstable);
         }
-        ssTables_ = new SSTableTracker();
+        ssTables_ = new SSTableTracker(table, columnFamilyName);
         ssTables_.add(sstables);
-
-        double v = DatabaseDescriptor.getRowsCachedFraction(table, columnFamilyName);
-        int cacheSize;
-        if (0 < v && v < 1)
-            cacheSize = Math.max(1, (int)(v * ssTables_.estimatedKeys()));
-        else
-            cacheSize = (int)v;
-        if (logger_.isDebugEnabled())
-            logger_.debug("row cache capacity for " + columnFamilyName + " is " + cacheSize);
-        rowCache = new JMXInstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", cacheSize);
-
-        // we don't need to keep a reference to the key cache aggregator, just create it so it registers itself w/ JMX
-        new JMXAggregatingCache(new Iterable<IAggregatableCacheProvider>()
-        {
-            public Iterator<IAggregatableCacheProvider> iterator()
-            {
-                final Iterator<SSTableReader> iter = ssTables_.iterator();
-                return new AbstractIterator<IAggregatableCacheProvider>()
-                {
-                    @Override
-                    protected IAggregatableCacheProvider computeNext()
-                    {
-                        if (!iter.hasNext())
-                            return endOfData();
-
-                        return new IAggregatableCacheProvider()
-                        {
-                            SSTableReader sstable = iter.next();
-
-                            public InstrumentedCache getCache()
-                            {
-                                return sstable.getKeyCache();
-                            }
-
-                            public long getObjectCount()
-                            {
-                                return sstable.getIndexPositions().size() * SSTableReader.indexInterval();
-                            }
-                        };
-                    }
-                };
-            }
-        }, table, columnFamilyName + "KeyCache");
     }
 
     public static ColumnFamilyStore createColumnFamilyStore(String table, String columnFamily) throws IOException
@@ -726,12 +675,12 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     private ColumnFamily cacheRow(String key) throws IOException
     {
         ColumnFamily cached;
-        if ((cached = rowCache.get(key)) == null)
+        if ((cached = ssTables_.getRowCache().get(key)) == null)
         {
             cached = getTopLevelColumns(new IdentityQueryFilter(key, new QueryPath(columnFamily_)), Integer.MIN_VALUE);
             if (cached == null)
                 return null;
-            rowCache.put(key, cached);
+            ssTables_.getRowCache().put(key, cached);
         }
         return cached;
     }
@@ -750,7 +699,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
         {
             if (filter.path.superColumnName == null)
             {
-                if (rowCache.getCapacity() == 0)
+                if (ssTables_.getRowCache().getCapacity() == 0)
                     return removeDeleted(getTopLevelColumns(filter, gcBefore), gcBefore);
 
                 ColumnFamily cached = cacheRow(filter.key);
@@ -763,7 +712,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
             // we are querying subcolumns of a supercolumn: fetch the supercolumn with NQF, then filter in-memory.
             ColumnFamily cf;
             SuperColumn sc;
-            if (rowCache.getCapacity() == 0)
+            if (ssTables_.getRowCache().getCapacity() == 0)
             {
                 QueryFilter nameFilter = new NamesQueryFilter(filter.key, new QueryPath(columnFamily_), filter.path.superColumnName);
                 cf = getTopLevelColumns(nameFilter, gcBefore);
@@ -1100,13 +1049,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
     /** raw cached row -- does not fetch the row if it is not present.  not counted in cache statistics.  */
     public ColumnFamily getRawCachedRow(String key)
     {
-        return rowCache.getCapacity() == 0 ? null : rowCache.getInternal(key);
+        return ssTables_.getRowCache().getCapacity() == 0 ? null : ssTables_.getRowCache().getInternal(key);
     }
 
     void invalidateCachedRow(String key)
     {
-        if (rowCache != null)
-            rowCache.remove(key);
+        if (ssTables_.getRowCache() != null)
+            ssTables_.getRowCache().remove(key);
     }
 
     public void forceMajorCompaction()
diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java
index cb88a63758..9ffef73c34 100644
--- a/src/java/org/apache/cassandra/db/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/CompactionManager.java
@@ -310,7 +310,7 @@ public class CompactionManager implements CompactionManagerMBean
             ci.close();
         }
 
-        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()));
+        SSTableReader ssTable = writer.closeAndOpenReader();
         cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable));
         submitMinorIfNeeded(cfs);
 
@@ -384,7 +384,7 @@ public class CompactionManager implements CompactionManagerMBean
 
         if (writer != null)
         {
-            results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()), target != null));
+            results.add(writer.closeAndOpenReader());
             String format = "AntiCompacted to %s.  %d/%d bytes for %d keys.  Time: %dms.";
             long dTime = System.currentTimeMillis() - startTime;
             logger.info(String.format(format, writer.getFilename(), SSTable.getTotalBytes(sstables), results.get(0).length(), totalkeysWritten, dTime));
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index cd4166bc8f..eafa6eb711 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -158,7 +158,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable
             writer.append(entry.getKey(), buffer);
         }
 
-        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(getTableName(), cfs.getColumnFamilyName()));
+        SSTableReader ssTable = writer.closeAndOpenReader();
         logger.info("Completed flushing " + ssTable.getFilename());
         return ssTable;
     }
diff --git a/src/java/org/apache/cassandra/io/SSTableReader.java b/src/java/org/apache/cassandra/io/SSTableReader.java
index 2904310faf..7a33500060 100644
--- a/src/java/org/apache/cassandra/io/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/SSTableReader.java
@@ -31,8 +31,8 @@ import org.apache.commons.lang.StringUtils;
 
 import org.apache.cassandra.cache.InstrumentedCache;
 import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.utils.BloomFilter;
+import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.*;
@@ -42,8 +42,6 @@ import org.apache.cassandra.io.util.FileDataInput;
 import org.apache.cassandra.io.util.MappedFileDataInput;
 
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
-import com.google.common.base.Predicate;
-import com.google.common.base.Predicates;
 
 /**
  * SSTableReaders are open()ed by Table.onStart; after that they are created by SSTableWriter.renameAndOpen.
@@ -116,12 +114,10 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
 
     public static SSTableReader open(String dataFileName) throws IOException
     {
-        return open(dataFileName,
-                    StorageService.getPartitioner(),
-                    DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
+        return open(dataFileName, StorageService.getPartitioner());
     }
 
-    public static SSTableReader open(String dataFileName, IPartitioner partitioner, double keysCacheFraction) throws IOException
+    public static SSTableReader open(String dataFileName, IPartitioner partitioner) throws IOException
     {
         assert partitioner != null;
 
@@ -130,10 +126,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         logger.info("Sampling index for " + dataFileName);
         sstable.loadIndexFile();
         sstable.loadBloomFilter();
-        if (keysCacheFraction > 0)
-        {
-            sstable.keyCache = createKeyCache((int)((sstable.getIndexPositions().size() + 1) * INDEX_INTERVAL * keysCacheFraction));
-        }
+
         if (logger.isDebugEnabled())
             logger.debug("INDEX LOAD TIME for "  + dataFileName + ": " + (System.currentTimeMillis() - start) + " ms.");
 
@@ -145,23 +138,15 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
     private final MappedByteBuffer[] indexBuffers;
     private final MappedByteBuffer[] buffers;
 
-
-    public static InstrumentedCache<DecoratedKey, PositionSize> createKeyCache(int size)
-    {
-        return new InstrumentedCache<DecoratedKey, PositionSize>(size);
-    }
-
-    private InstrumentedCache<DecoratedKey, PositionSize> keyCache;
+    private InstrumentedCache<Pair<Descriptor,DecoratedKey>,PositionSize> keyCache;
 
     SSTableReader(String filename,
                   IPartitioner partitioner,
                   List<KeyPosition> indexPositions, Map<KeyPosition, PositionSize> spannedIndexDataPositions,
-                  BloomFilter bloomFilter,
-                  InstrumentedCache<DecoratedKey, PositionSize> keyCache)
-            throws IOException
+                  BloomFilter bloomFilter)
+    throws IOException
     {
         super(filename, partitioner);
-        assert keyCache != null;
 
         if (DatabaseDescriptor.getIndexAccessMode() == DatabaseDescriptor.DiskAccessMode.mmap)
         {
@@ -201,13 +186,13 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         this.indexPositions = indexPositions;
         this.spannedIndexDataPositions = spannedIndexDataPositions;
         this.bf = bloomFilter;
-        this.keyCache = keyCache;
     }
 
-    public void addFinalizingReference(SSTableTracker tracker)
+    public void setTrackedBy(SSTableTracker tracker)
     {
         phantomReference = new SSTableDeletingReference(tracker, this, finalizerQueue);
         finalizers.add(phantomReference);
+        keyCache = tracker.getKeyCache();
     }
 
     private static MappedByteBuffer mmap(String filename, long start, int size) throws IOException
@@ -234,7 +219,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
 
     private SSTableReader(String filename, IPartitioner partitioner) throws IOException
     {
-        this(filename, partitioner, null, null, null, SSTableReader.createKeyCache(0));
+        this(filename, partitioner, null, null, null);
     }
 
     public List<KeyPosition> getIndexPositions()
@@ -242,6 +227,11 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         return indexPositions;
     }
 
+    public long estimatedKeys()
+    {
+        return indexPositions.size() * INDEX_INTERVAL;
+    }
+
     void loadBloomFilter() throws IOException
     {
         DataInputStream stream = new DataInputStream(new FileInputStream(filterFilename()));
@@ -329,21 +319,29 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
      */
     public PositionSize getPosition(DecoratedKey decoratedKey) throws IOException
     {
+        // first, check bloom filter
         if (!bf.isPresent(partitioner.convertToDiskFormat(decoratedKey)))
             return null;
-        if (keyCache.getCapacity() > 0)
+
+        // next, the key cache
+        Pair<Descriptor, DecoratedKey> unifiedKey = new Pair<Descriptor, DecoratedKey>(desc, decoratedKey);
+        if (keyCache != null && keyCache.getCapacity() > 0)
         {
-            PositionSize cachedPosition = keyCache.get(decoratedKey);
+            PositionSize cachedPosition = keyCache.get(unifiedKey);
             if (cachedPosition != null)
             {
                 return cachedPosition;
             }
         }
+
+        // next, see if the sampled index says it's impossible for the key to be present
         KeyPosition sampledPosition = getIndexScanPosition(decoratedKey);
         if (sampledPosition == null)
         {
             return null;
         }
+
+        // handle exact sampled index hit
         if (spannedIndexDataPositions != null)
         {
             PositionSize info = spannedIndexDataPositions.get(sampledPosition);
@@ -351,6 +349,7 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
                 return info;
         }
 
+        // scan the on-disk index, starting at the nearest sampled position
         long p = sampledPosition.position;
         FileDataInput input;
         if (indexBuffers == null)
@@ -392,8 +391,8 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
                     {
                         info = new PositionSize(position, length() - position);
                     }
-                    if (keyCache.getCapacity() > 0)
-                        keyCache.put(decoratedKey, info);
+                    if (keyCache != null && keyCache.getCapacity() > 0)
+                        keyCache.put(unifiedKey, info);
                     return info;
                 }
                 if (v > 0)
@@ -511,11 +510,6 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         return ColumnFamily.create(getTableName(), getColumnFamilyName());
     }
 
-    public InstrumentedCache<DecoratedKey, PositionSize> getKeyCache()
-    {
-        return keyCache;
-    }
-
     public ICompactSerializer2<IColumn> getColumnSerializer()
     {
         return DatabaseDescriptor.getColumnFamilyType(getTableName(), getColumnFamilyName()).equals("Standard")
diff --git a/src/java/org/apache/cassandra/io/SSTableTracker.java b/src/java/org/apache/cassandra/io/SSTableTracker.java
index 4e20980588..ac5fbcb981 100644
--- a/src/java/org/apache/cassandra/io/SSTableTracker.java
+++ b/src/java/org/apache/cassandra/io/SSTableTracker.java
@@ -25,16 +25,35 @@ import java.util.*;
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.cassandra.cache.JMXInstrumentedCache;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.utils.Pair;
+
+import org.apache.log4j.Logger;
 
 public class SSTableTracker implements Iterable<SSTableReader>
 {
+    private static final Logger logger = Logger.getLogger(SSTableTracker.class);
+
     private volatile Set<SSTableReader> sstables;
     private final AtomicLong liveSize = new AtomicLong();
     private final AtomicLong totalSize = new AtomicLong();
 
-    public SSTableTracker()
+    private final String ksname;
+    private final String cfname;
+
+    private final JMXInstrumentedCache<Pair<SSTable.Descriptor,DecoratedKey>,SSTable.PositionSize> keyCache;
+    private final JMXInstrumentedCache<String, ColumnFamily> rowCache;
+
+    public SSTableTracker(String ksname, String cfname)
     {
-        this.sstables = Collections.<SSTableReader>emptySet();
+        this.ksname = ksname;
+        this.cfname = cfname;
+        sstables = Collections.emptySet();
+        keyCache = new JMXInstrumentedCache<Pair<SSTable.Descriptor,DecoratedKey>,SSTable.PositionSize>(ksname, cfname + "KeyCache", 0);
+        rowCache = new JMXInstrumentedCache<String, ColumnFamily>(ksname, cfname + "RowCache", 0);
     }
 
     public synchronized void replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements) throws IOException
@@ -48,7 +67,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
             long size = sstable.bytesOnDisk();
             liveSize.addAndGet(size);
             totalSize.addAndGet(size);
-            sstable.addFinalizingReference(this);
+            sstable.setTrackedBy(this);
         }
 
         for (SSTableReader sstable : oldSSTables)
@@ -60,6 +79,7 @@ public class SSTableTracker implements Iterable<SSTableReader>
         }
 
         sstables = Collections.unmodifiableSet(sstablesNew);
+        updateCacheSizes();
     }
 
     public synchronized void add(Iterable<SSTableReader> sstables)
@@ -80,6 +100,31 @@ public class SSTableTracker implements Iterable<SSTableReader>
         replace(compacted, Collections.<SSTableReader>emptyList());
     }
 
+    /**
+     * Resizes the key and row caches based on the current key estimate.
+     */
+    public synchronized void updateCacheSizes()
+    {
+        long keys = estimatedKeys();
+        
+        int keyCacheSize = DatabaseDescriptor.getKeysCachedFor(ksname, cfname, keys);
+        if (keyCacheSize != keyCache.getCapacity())
+        {
+            // update cache size for the new key volume
+            if (logger.isDebugEnabled())
+                logger.debug("key cache capacity for " + cfname + " is " + keyCacheSize);
+            keyCache.setCapacity(keyCacheSize);
+        }
+
+        int rowCacheSize = DatabaseDescriptor.getRowsCachedFor(ksname, cfname, keys);
+        if (rowCacheSize != rowCache.getCapacity())
+        {   
+            if (logger.isDebugEnabled())
+                logger.debug("row cache capacity for " + cfname + " is " + rowCacheSize);
+            rowCache.setCapacity(rowCacheSize);
+        }
+    }
+
     // the modifiers create new, unmodifiable objects each time; the volatile fences the assignment
     // so we don't need any further synchronization for the common case here
     public Set<SSTableReader> getSSTables()
@@ -102,12 +147,17 @@ public class SSTableTracker implements Iterable<SSTableReader>
         sstables = Collections.emptySet();
     }
 
+    public JMXInstrumentedCache<String, ColumnFamily> getRowCache()
+    {
+        return rowCache;
+    }
+
     public long estimatedKeys()
     {
         long n = 0;
         for (SSTableReader sstable : this)
         {
-            n += sstable.getIndexPositions().size() * SSTableReader.INDEX_INTERVAL;
+            n += sstable.estimatedKeys();
         }
         return n;
     }
@@ -126,5 +176,10 @@ public class SSTableTracker implements Iterable<SSTableReader>
     {
         totalSize.addAndGet(-size);
     }
+
+    public JMXInstrumentedCache<Pair<SSTable.Descriptor, DecoratedKey>, SSTable.PositionSize> getKeyCache()
+    {
+        return keyCache;
+    }
 }
 
diff --git a/src/java/org/apache/cassandra/io/SSTableWriter.java b/src/java/org/apache/cassandra/io/SSTableWriter.java
index 1c3e74397d..d835c18c18 100644
--- a/src/java/org/apache/cassandra/io/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/SSTableWriter.java
@@ -21,22 +21,23 @@ package org.apache.cassandra.io;
  */
 
 
-import java.io.*;
+import java.io.DataOutputStream;
+import java.io.FileOutputStream;
+import java.io.IOError;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 
 import org.apache.log4j.Logger;
 
-import org.apache.cassandra.cache.InstrumentedCache;
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.io.util.BufferedRandomAccessFile;
+import org.apache.cassandra.io.util.DataOutputBuffer;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.BloomFilter;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.config.DatabaseDescriptor;
-import org.apache.cassandra.io.util.BufferedRandomAccessFile;
-import org.apache.cassandra.io.util.DataOutputBuffer;
-import com.reardencommerce.kernel.collections.shared.evictable.ConcurrentLinkedHashMap;
 
 public class SSTableWriter extends SSTable
 {
@@ -128,16 +129,16 @@ public class SSTableWriter extends SSTable
         afterAppend(decoratedKey, currentPosition, value.length);
     }
 
-    public SSTableReader closeAndOpenReader(double cacheFraction) throws IOException
+    public SSTableReader closeAndOpenReader() throws IOException
     {
-        return closeAndOpenReader(cacheFraction, false);
+        return closeAndOpenReader(false);
     }
 
     /**
      * Renames temporary SSTable files to valid data, index, and bloom filter files.
      * The SSTableWriter object will no longer be valid.
      */
-    public SSTableReader closeAndOpenReader(double cacheFraction, boolean temporary) throws IOException
+    public SSTableReader closeAndOpenReader(boolean temporary) throws IOException
     {
         // bloom filter
         FileOutputStream fos = new FileOutputStream(filterFilename());
@@ -162,8 +163,7 @@ public class SSTableWriter extends SSTable
             newpath = rename(newpath); // important to do this last since index & filter file names are derived from it
         }
 
-        InstrumentedCache<DecoratedKey, PositionSize> keyCache = SSTableReader.createKeyCache((int)(cacheFraction * keysWritten));
-        return new SSTableReader(newpath, partitioner, indexPositions, spannedIndexDataPositions, bf, keyCache);
+        return new SSTableReader(newpath, partitioner, indexPositions, spannedIndexDataPositions, bf);
     }
 
     static String rename(String tmpFilename)
@@ -186,8 +186,7 @@ public class SSTableWriter extends SSTable
         SSTableWriter.rename(filterFilename(dataFileName));
         dataFileName = SSTableWriter.rename(dataFileName);
         return SSTableReader.open(dataFileName,
-                                  StorageService.getPartitioner(),
-                                  DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
+                                  StorageService.getPartitioner());
     }
 
 }
diff --git a/src/java/org/apache/cassandra/tools/SSTableImport.java b/src/java/org/apache/cassandra/tools/SSTableImport.java
index 9d2970ba24..779fafda15 100644
--- a/src/java/org/apache/cassandra/tools/SSTableImport.java
+++ b/src/java/org/apache/cassandra/tools/SSTableImport.java
@@ -161,7 +161,7 @@ public class SSTableImport
                 cfamily.clear();
             }
             
-            writer.closeAndOpenReader(0);
+            writer.closeAndOpenReader();
         }
         catch (ClassCastException cce)
         {
@@ -209,4 +209,4 @@ public class SSTableImport
         System.exit(0);
     }
 
-}
\ No newline at end of file
+}
diff --git a/src/java/org/apache/cassandra/utils/FBUtilities.java b/src/java/org/apache/cassandra/utils/FBUtilities.java
index 898ad380c7..6280ddcfdb 100644
--- a/src/java/org/apache/cassandra/utils/FBUtilities.java
+++ b/src/java/org/apache/cassandra/utils/FBUtilities.java
@@ -63,6 +63,21 @@ public class FBUtilities
         return result.toArray( new String[0] );
     }
 
+    /**
+     * Parses a string representing either a fraction, absolute value or percentage.
+     */
+    public static double parseDoubleOrPercent(String value)
+    {
+        if (value.endsWith("%"))
+        {
+            return Double.valueOf(value.substring(0, value.length() - 1)) / 100;
+        }
+        else
+        {
+            return Double.valueOf(value);
+        }
+    }
+
     public static InetAddress getLocalAddress()
     {
         if (localInetAddress_ == null)
@@ -79,6 +94,27 @@ public class FBUtilities
         return localInetAddress_;
     }
 
+    /**
+     * @param fractOrAbs A double that may represent a fraction or absolute value.
+     * @param total If fractionOrAbs is a fraction, the total to take the fraction from
+     * @return An absolute value which may be larger than the total.
+     */
+    public static long absoluteFromFraction(double fractOrAbs, long total)
+    {
+        if (fractOrAbs < 0)
+            throw new UnsupportedOperationException("unexpected negative value " + fractOrAbs);
+
+        if (0 < fractOrAbs && fractOrAbs < 1)
+        {
+            // fraction
+            return Math.max(1, (long)(fractOrAbs * total));
+        }
+
+        // absolute
+        assert fractOrAbs >= 1 || fractOrAbs == 0;
+        return (long)fractOrAbs;
+    }
+
     /**
      * Given two bit arrays represented as BigIntegers, containing the given
      * number of significant bits, calculate a midpoint.
diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml
index 292f85632d..c105e67edf 100644
--- a/test/conf/storage-conf.xml
+++ b/test/conf/storage-conf.xml
@@ -40,11 +40,11 @@
    <MemtableOperationsInMillions>0.00002</MemtableOperationsInMillions> <!-- 20 -->
    <Keyspaces>
      <Keyspace Name = "Keyspace1">
-       <ColumnFamily Name="Standard1" RowsCached="10%" KeysCachedFraction="0"/>
+       <ColumnFamily Name="Standard1" RowsCached="10%" KeysCached="0"/>
        <ColumnFamily Name="Standard2"/>
        <ColumnFamily CompareWith="LongType" Name="StandardLong1"/>
        <ColumnFamily CompareWith="LongType" Name="StandardLong2"/>
-       <ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super1" RowsCached="1000" KeysCachedFraction="0"/>
+       <ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super1" RowsCached="1000" KeysCached="0"/>
        <ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super2"/>
        <ColumnFamily ColumnType="Super" CompareSubcolumnsWith="LongType" Name="Super3"/>
        <ColumnFamily ColumnType="Super" CompareSubcolumnsWith="UTF8Type" Name="Super4"/>
diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java
index 5750b55a92..ca7322415e 100644
--- a/test/unit/org/apache/cassandra/io/SSTableTest.java
+++ b/test/unit/org/apache/cassandra/io/SSTableTest.java
@@ -43,7 +43,7 @@ public class SSTableTest extends CleanupHelper
 
         TreeMap<String, byte[]> map = new TreeMap<String,byte[]>();
         map.put(key, bytes);
-        SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "singlewrite", map);
+        SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard1", map);
 
         // verify
         verifySingle(ssTable, bytes, key);
@@ -71,7 +71,7 @@ public class SSTableTest extends CleanupHelper
         }
 
         // write
-        SSTableReader ssTable = SSTableUtils.writeRawSSTable("table", "manywrites", map);
+        SSTableReader ssTable = SSTableUtils.writeRawSSTable("Keyspace1", "Standard2", map);
 
         // verify
         verifyMany(ssTable, map);
diff --git a/test/unit/org/apache/cassandra/io/SSTableUtils.java b/test/unit/org/apache/cassandra/io/SSTableUtils.java
index 6bf7830f26..0089d9b914 100644
--- a/test/unit/org/apache/cassandra/io/SSTableUtils.java
+++ b/test/unit/org/apache/cassandra/io/SSTableUtils.java
@@ -104,6 +104,6 @@ public class SSTableUtils
                           entry.getValue());
         new File(writer.indexFilename()).deleteOnExit();
         new File(writer.filterFilename()).deleteOnExit();
-        return writer.closeAndOpenReader(1.0);
+        return writer.closeAndOpenReader();
     }
 }
diff --git a/test/unit/org/apache/cassandra/tools/SSTableExportTest.java b/test/unit/org/apache/cassandra/tools/SSTableExportTest.java
index 4f067deadb..ef139d1138 100644
--- a/test/unit/org/apache/cassandra/tools/SSTableExportTest.java
+++ b/test/unit/org/apache/cassandra/tools/SSTableExportTest.java
@@ -28,7 +28,6 @@ import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.filter.NamesQueryFilter;
 import org.apache.cassandra.db.filter.QueryPath;
 import org.apache.cassandra.dht.IPartitioner;
-import org.apache.cassandra.io.SSTableAccessor;
 import org.apache.cassandra.io.SSTableReader;
 import org.apache.cassandra.io.SSTableWriter;
 import org.apache.cassandra.io.util.DataOutputBuffer;
@@ -67,7 +66,7 @@ public class SSTableExportTest
         dob.reset();
         cfamily.clear();
      
-        writer.closeAndOpenReader(0);
+        writer.closeAndOpenReader();
         
         // Enumerate and verify
         File temp = File.createTempFile("Standard1", ".txt");
@@ -105,7 +104,7 @@ public class SSTableExportTest
         dob.reset();
         cfamily.clear();
      
-        SSTableReader reader = writer.closeAndOpenReader(0);
+        SSTableReader reader = writer.closeAndOpenReader();
         
         // Export to JSON and verify
         File tempJson = File.createTempFile("Standard1", ".json");
@@ -145,7 +144,7 @@ public class SSTableExportTest
         dob.reset();
         cfamily.clear();
      
-        SSTableReader reader = writer.closeAndOpenReader(0);
+        SSTableReader reader = writer.closeAndOpenReader();
         
         // Export to JSON and verify
         File tempJson = File.createTempFile("Super4", ".json");
@@ -178,7 +177,7 @@ public class SSTableExportTest
         dob.reset();
         cfamily.clear();
         
-        SSTableReader reader = writer.closeAndOpenReader(0);
+        SSTableReader reader = writer.closeAndOpenReader();
         
         // Export to JSON and verify
         File tempJson = File.createTempFile("Standard1", ".json");
@@ -188,7 +187,7 @@ public class SSTableExportTest
         File tempSS2 = tempSSTableFile("Keyspace1", "Standard1");
         SSTableImport.importJson(tempJson.getPath(), "Keyspace1", "Standard1", tempSS2.getPath());        
         
-        reader = SSTableAccessor.getSSTableReader(tempSS2.getPath(), DatabaseDescriptor.getPartitioner());
+        reader = SSTableReader.open(tempSS2.getPath(), DatabaseDescriptor.getPartitioner());
         NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Standard1", null, null), "name".getBytes());
         ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
         assertTrue(cf != null);
diff --git a/test/unit/org/apache/cassandra/tools/SSTableImportTest.java b/test/unit/org/apache/cassandra/tools/SSTableImportTest.java
index 10f3585f16..fe200349c1 100644
--- a/test/unit/org/apache/cassandra/tools/SSTableImportTest.java
+++ b/test/unit/org/apache/cassandra/tools/SSTableImportTest.java
@@ -26,7 +26,6 @@ import org.apache.cassandra.db.ColumnFamily;
 import org.apache.cassandra.db.IColumn;
 import org.apache.cassandra.db.filter.NamesQueryFilter;
 import org.apache.cassandra.db.filter.QueryPath;
-import org.apache.cassandra.io.SSTableAccessor;
 import org.apache.cassandra.io.SSTableReader;
 import static org.apache.cassandra.utils.FBUtilities.hexToBytes;
 import static org.apache.cassandra.io.SSTableUtils.tempSSTableFile;
@@ -44,7 +43,7 @@ public class SSTableImportTest
         SSTableImport.importJson(jsonUrl, "Keyspace1", "Standard1", tempSS.getPath());
 
         // Verify results
-        SSTableReader reader = SSTableAccessor.getSSTableReader(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
+        SSTableReader reader = SSTableReader.open(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
         NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Standard1", null, null), "colAA".getBytes());
         ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
         assert Arrays.equals(cf.getColumn("colAA".getBytes()).value(), hexToBytes("76616c4141"));
@@ -58,7 +57,7 @@ public class SSTableImportTest
         SSTableImport.importJson(jsonUrl, "Keyspace1", "Super4", tempSS.getPath());
         
         // Verify results
-        SSTableReader reader = SSTableAccessor.getSSTableReader(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
+        SSTableReader reader = SSTableReader.open(tempSS.getPath(), DatabaseDescriptor.getPartitioner());
         NamesQueryFilter qf = new NamesQueryFilter("rowA", new QueryPath("Super4", null, null), "superA".getBytes());
         ColumnFamily cf = qf.getSSTableColumnIterator(reader).getColumnFamily();
         IColumn superCol = cf.getColumn("superA".getBytes());