diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index 4a35bc326b..303c0c9d8e 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -56,17 +56,9 @@
   <Keyspaces>
     <Keyspace Name="Keyspace1">
       <!--
-       ~ The fraction 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.
-      -->
-      <KeysCachedFraction>0.01</KeysCachedFraction>
-      <!-- 
+       ~ ColumnFamily definitions have one required attribute (Name)
+       ~ and several optional ones.
+       ~
        ~ The CompareWith attribute tells Cassandra how to sort the columns
        ~ for slicing operations.  The default is BytesType, which is a
        ~ straightforward lexical comparison of the bytes in each column.
@@ -90,8 +82,25 @@
        ~
        ~ 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
+       ~ 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.
+       ~
+       ~ The optional RowsCachedFraction attribute specifies
+       ~ The fraction of rows per sstable whose entire contents we cache in
+       ~ memory.  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.
       -->
-      <ColumnFamily CompareWith="BytesType" Name="Standard1"/>
+      <ColumnFamily CompareWith="BytesType" 
+                    Name="Standard1" 
+                    RowsCachedFraction="0.1"
+                    KeysCachedFraction="0"/>
       <ColumnFamily CompareWith="UTF8Type" Name="Standard2"/>
       <ColumnFamily CompareWith="TimeUUIDType" Name="StandardByUUID1"/>
       <ColumnFamily ColumnType="Super"
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index aa15ee236e..4d4a71120c 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -26,6 +26,7 @@ import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.locator.IEndPointSnitch;
 import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.io.util.FileUtils;
+import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.XMLUtils;
 import org.apache.log4j.Logger;
 import org.w3c.dom.Node;
@@ -85,7 +86,9 @@ public class DatabaseDescriptor
     private static Set<String> applicationColumnFamilies_ = new HashSet<String>();
     private static int bmtThreshold_ = 256;
 
-    private static Map<String, Double> tableKeysCachedFractions_;
+    private static Map<Pair<String, String>, Double> tableKeysCachedFractions_ = new HashMap<Pair<String, String>, Double>();
+    private static Map<Pair<String, String>, Double> tableRowsCachedFractions_ = new HashMap<Pair<String, String>, Double>();
+
     /*
      * A map from table names to the set of column families for the table and the
      * corresponding meta data for that column family.
@@ -435,7 +438,6 @@ public class DatabaseDescriptor
                 CommitLog.setSegmentSize(Integer.parseInt(value) * 1024 * 1024);
 
             tableToCFMetaDataMap_ = new HashMap<String, Map<String, CFMetaData>>();
-            tableKeysCachedFractions_ = new HashMap<String, Double>();
 
             /* See which replica placement strategy to use */
             String replicaPlacementStrategyClassName = xmlUtils.getNodeValue("/Storage/ReplicaPlacementStrategy");
@@ -472,17 +474,6 @@ public class DatabaseDescriptor
                 tables_.add(tName);
                 tableToCFMetaDataMap_.put(tName, new HashMap<String, CFMetaData>());
 
-                String xqlCacheSize = "/Storage/Keyspaces/Keyspace[@Name='" + tName + "']/KeysCachedFraction";
-                value = xmlUtils.getNodeValue(xqlCacheSize);
-                if (value == null)
-                {
-                    tableKeysCachedFractions_.put(tName, 0.01);
-                }
-                else
-                {
-                    tableKeysCachedFractions_.put(tName, Double.valueOf(value));
-                }
-
                 String xqlTable = "/Storage/Keyspaces/Keyspace[@Name='" + tName + "']/";
                 NodeList columnFamilies = xmlUtils.getRequestedNodeList(xqlTable + "ColumnFamily");
 
@@ -527,6 +518,16 @@ public class DatabaseDescriptor
                         throw new ConfigurationException("CompareSubcolumnsWith is only a valid attribute on super columnfamilies (not regular columnfamily " + cfName + ")");
                     }
 
+                    if ((value = XMLUtils.getAttributeValue(columnFamily, "KeysCachedFraction")) != null)
+                    {
+                        tableKeysCachedFractions_.put(Pair.create(tName, cfName), Double.valueOf(value));
+                    }
+                    
+                    if ((value = XMLUtils.getAttributeValue(columnFamily, "RowsCachedFraction")) != null)
+                    {
+                        tableRowsCachedFractions_.put(Pair.create(tName, cfName), Double.valueOf(value));
+                    }
+
                     // Parse out user-specified logical names for the various dimensions
                     // of a the column family from the config.
                     String cfComment = xmlUtils.getNodeValue(xqlCF + "Comment");
@@ -567,7 +568,6 @@ public class DatabaseDescriptor
             systemMetadata.put(data.cfName, data);
 
             tableToCFMetaDataMap_.put(Table.SYSTEM_TABLE, systemMetadata);
-            tableKeysCachedFractions_.put(Table.SYSTEM_TABLE, 0.0);
 
             /* Load the seeds for node contact points */
             String[] seeds = xmlUtils.getNodeValues("/Storage/Seeds/Seed");
@@ -953,9 +953,16 @@ public class DatabaseDescriptor
         return tableToCFMetaDataMap_;
     }
 
-    public static double getKeysCachedFraction(String tableName)
+    public static double getKeysCachedFraction(String tableName, String columnFamilyName)
     {
-        return tableKeysCachedFractions_.get(tableName);
+        Double v = tableKeysCachedFractions_.get(Pair.create(tableName, columnFamilyName));
+        return v == null ? 0.01 : v;
+    }
+
+    public static double getRowsCachedFraction(String tableName, String columnFamilyName)
+    {
+        Double v = tableRowsCachedFractions_.get(Pair.create(tableName, columnFamilyName));
+        return v == null ? 0.01 : v;
     }
 
     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 b413e0b499..15418bf5ba 100644
--- a/src/java/org/apache/cassandra/db/BinaryMemtable.java
+++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java
@@ -149,7 +149,7 @@ public class BinaryMemtable implements IFlushable<DecoratedKey>
             assert bytes.length > 0;
             writer.append(key, bytes);
         }
-        SSTableReader sstable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table_));
+        SSTableReader sstable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table_, cfName_));
         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 c04328c685..f8ed75fd9f 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -185,9 +185,12 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         }
         ssTables_ = new SSTableTracker(sstables);
 
-        int cacheSize = (int)(0.2 * SSTableReader.estimatedKeys(columnFamilyName));
-        logger_.info("cache size for " + columnFamilyName + " is " + cacheSize);
-        rowCache = new InstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", cacheSize);
+        double v = DatabaseDescriptor.getRowsCachedFraction(table, columnFamilyName);
+        int cacheSize = (int)(v * SSTableReader.estimatedKeys(columnFamilyName));
+        if (logger_.isDebugEnabled())
+            logger_.debug("cache size for " + columnFamilyName + " is " + cacheSize);
+        if (cacheSize > 0)
+            rowCache = new InstrumentedCache<String, ColumnFamily>(table, columnFamilyName + "RowCache", cacheSize);
     }
 
     public static ColumnFamilyStore createColumnFamilyStore(String table, String columnFamily) throws IOException
@@ -1259,7 +1262,13 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
     public void invalidate(String key)
     {
-        rowCache.remove(key);
+        if (rowCache != null)
+            rowCache.remove(key);
+    }
+
+    public boolean isRowCacheEnabled()
+    {
+        return rowCache != null;
     }
 
     /**
diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java
index b4a7b3c9d6..1afd3f82aa 100644
--- a/src/java/org/apache/cassandra/db/CompactionManager.java
+++ b/src/java/org/apache/cassandra/db/CompactionManager.java
@@ -306,7 +306,7 @@ public class CompactionManager implements CompactionManagerMBean
             ci.close();
         }
 
-        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name));
+        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()));
         cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable));
         gcAfterRpcTimeout();
         submitMinorIfNeeded(cfs);
@@ -386,7 +386,7 @@ public class CompactionManager implements CompactionManagerMBean
 
         if (writer != null)
         {
-            results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name)));
+            results.add(writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName())));
             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 9bf2752885..fd21598015 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -167,7 +167,7 @@ public class Memtable implements Comparable<Memtable>, IFlushable<DecoratedKey>
             writer.append(key, buffer);
         }
 
-        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table));
+        SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table, columnfamilyName));
         logger.info("Completed flushing " + ssTable.getFilename());
         return ssTable;
     }
diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java
index bf78fc26d3..8763e13e34 100644
--- a/src/java/org/apache/cassandra/db/Table.java
+++ b/src/java/org/apache/cassandra/db/Table.java
@@ -407,6 +407,7 @@ public class Table
         HashMap<ColumnFamilyStore,Memtable> memtablesToFlush = new HashMap<ColumnFamilyStore, Memtable>(2);
 
         // write the mutation to the commitlog and memtables
+        boolean invalidateRequired = false;
         flusherLock.readLock().lock();
         try
         {
@@ -417,6 +418,7 @@ public class Table
             {
                 Memtable memtableToFlush;
                 ColumnFamilyStore cfStore = columnFamilyStores.get(columnFamily.name());
+                invalidateRequired |= cfStore.isRowCacheEnabled();
                 if ((memtableToFlush=cfStore.apply(mutation.key(), columnFamily)) != null)
                     memtablesToFlush.put(cfStore, memtableToFlush);
             }
@@ -427,10 +429,13 @@ public class Table
         }
 
         // invalidate cache.  2nd loop over CFs here to avoid prolonging the lock section unnecessarily.
-        for (ColumnFamily cf : mutation.getColumnFamilies())
+        if (invalidateRequired)
         {
-            ColumnFamilyStore cfs = columnFamilyStores.get(cf.name());
-            cfs.invalidate(mutation.key());
+            for (ColumnFamily cf : mutation.getColumnFamilies())
+            {
+                ColumnFamilyStore cfs = columnFamilyStores.get(cf.name());
+                cfs.invalidate(mutation.key());
+            }
         }
 
         // flush memtables that got filled up.  usually mTF will be empty and this will be a no-op
diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java
index 53d6c6443e..5130af17c9 100644
--- a/src/java/org/apache/cassandra/io/SSTable.java
+++ b/src/java/org/apache/cassandra/io/SSTable.java
@@ -67,11 +67,16 @@ public abstract class SSTable
     public SSTable(String filename, IPartitioner partitioner)
     {
         assert filename.endsWith("-Data.db");
-        columnFamilyName = new File(filename).getName().split("-")[0];
+        columnFamilyName = parseColumnFamilyName(filename);
         this.path = filename;
         this.partitioner = partitioner;
     }
 
+    protected static String parseColumnFamilyName(String filename)
+    {
+        return new File(filename).getName().split("-")[0];
+    }
+
     protected static String indexFilename(String dataFile)
     {
         String[] parts = dataFile.split("-");
diff --git a/src/java/org/apache/cassandra/io/SSTableReader.java b/src/java/org/apache/cassandra/io/SSTableReader.java
index 4c415b7a36..af48aa8bbd 100644
--- a/src/java/org/apache/cassandra/io/SSTableReader.java
+++ b/src/java/org/apache/cassandra/io/SSTableReader.java
@@ -173,10 +173,12 @@ 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)));
+        return open(dataFileName,
+                    StorageService.getPartitioner(),
+                    DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
     }
 
-    public static SSTableReader open(String dataFileName, IPartitioner partitioner, double cacheFraction) throws IOException
+    public static SSTableReader open(String dataFileName, IPartitioner partitioner, double keysCacheFraction) throws IOException
     {
         assert partitioner != null;
         assert openedFiles.get(dataFileName) == null;
@@ -185,9 +187,9 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         SSTableReader sstable = new SSTableReader(dataFileName, partitioner);
         sstable.loadIndexFile();
         sstable.loadBloomFilter();
-        if (cacheFraction > 0)
+        if (keysCacheFraction > 0)
         {
-            sstable.keyCache = createKeyCache((int)((sstable.getIndexPositions().size() + 1) * INDEX_INTERVAL * cacheFraction));
+            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.");
diff --git a/src/java/org/apache/cassandra/io/SSTableWriter.java b/src/java/org/apache/cassandra/io/SSTableWriter.java
index b416b66cad..c34b8c1e03 100644
--- a/src/java/org/apache/cassandra/io/SSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/SSTableWriter.java
@@ -176,7 +176,9 @@ public class SSTableWriter extends SSTable
         SSTableWriter.rename(indexFilename(dataFileName));
         SSTableWriter.rename(filterFilename(dataFileName));
         dataFileName = SSTableWriter.rename(dataFileName);
-        return SSTableReader.open(dataFileName, StorageService.getPartitioner(), DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName)));
+        return SSTableReader.open(dataFileName,
+                                  StorageService.getPartitioner(),
+                                  DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName), parseColumnFamilyName(dataFileName)));
     }
 
 }
diff --git a/src/java/org/apache/cassandra/utils/Pair.java b/src/java/org/apache/cassandra/utils/Pair.java
index f606c5ed23..1d7f2d1375 100644
--- a/src/java/org/apache/cassandra/utils/Pair.java
+++ b/src/java/org/apache/cassandra/utils/Pair.java
@@ -53,4 +53,9 @@ public class Pair<T1, T2>
     {
         return "(" + left + "," + right + ")";
     }
+
+    public static <X, Y> Pair<X, Y> create(X x, Y y)
+    {
+        return new Pair<X, Y>(x, y);
+    }
 }
diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml
index 5d9a797392..3b24c18823 100644
--- a/test/conf/storage-conf.xml
+++ b/test/conf/storage-conf.xml
@@ -44,7 +44,7 @@
    <MemtableOperationsInMillions>0.00002</MemtableOperationsInMillions> <!-- 20 -->
    <Keyspaces>
      <Keyspace Name = "Keyspace1">
-       <ColumnFamily Name="Standard1"/>
+       <ColumnFamily Name="Standard1" RowsCachedFraction="0.1" KeysCachedFraction="0"/>
        <ColumnFamily Name="Standard2"/>
        <ColumnFamily CompareWith="LongType" Name="StandardLong1"/>
        <ColumnFamily CompareWith="LongType" Name="StandardLong2"/>