diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java
index 6f209ff87d..bb39f8d3fe 100644
--- a/src/java/org/apache/cassandra/db/BinaryMemtable.java
+++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java
@@ -160,7 +160,7 @@ public class BinaryMemtable
                 bf.add(key);
             }
         }
-        ssTable.closeRename(bf);
+        ssTable.close(bf);
         cfStore.storeLocation( ssTable.getDataFileLocation(), bf );
         columnFamilies_.clear();       
     }
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 31cd970a4a..8eb0359f4a 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -25,7 +25,6 @@ import javax.management.MBeanServer;
 import javax.management.ObjectName;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -46,7 +45,6 @@ import org.apache.cassandra.utils.*;
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.collections.IteratorUtils;
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.comparators.ReverseComparator;
 
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
@@ -994,7 +992,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
                     SSTable.storeBloomFilter(newfile, compactedBloomFilters.get(0));
                 }
             }
-            SSTable.delete(file);
+            SSTable.open(file, null).delete();
         }
         finally
         {
@@ -1183,7 +1181,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
 
         if (ssTableRange != null)
         {
-            ssTableRange.closeRename(compactedRangeBloomFilter);
+            ssTableRange.close(compactedRangeBloomFilter);
             if (fileList != null)
             {
                 fileList.add(ssTableRange.getDataFileLocation());
@@ -1360,7 +1358,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
         if (ssTable != null)
         {
             // TODO if all the keys were the same nothing will be done here
-            ssTable.closeRename(compactedBloomFilter);
+            ssTable.close(compactedBloomFilter);
             newfile = ssTable.getDataFileLocation();
         }
         lock_.writeLock().lock();
@@ -1379,7 +1377,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
             }
             for (String file : files)
             {
-                SSTable.delete(file);
+                SSTable.open(file, null).delete();
             }
         }
         finally
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index 7218497b3d..6078239b80 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -282,7 +282,7 @@ public class Memtable implements Comparable<Memtable>
                 bf.add(key);
             }
         }
-        ssTable.closeRename(bf);
+        ssTable.close(bf);
         cfStore.onMemtableFlush(cLogCtx);
         cfStore.storeLocation( ssTable.getDataFileLocation(), bf );
         buffer.close();
diff --git a/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java b/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java
index 0f2e6086b9..38d66273e4 100644
--- a/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java
+++ b/src/java/org/apache/cassandra/io/BufferedRandomAccessFile.java
@@ -155,6 +155,7 @@ public final class BufferedRandomAccessFile extends RandomAccessFile
     public void close() throws IOException
     {
         this.flush();
+        this.buff_ = null;
         super.close();
     }
     
diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java
index 2e7b156a74..705f6a28a3 100644
--- a/src/java/org/apache/cassandra/io/SSTable.java
+++ b/src/java/org/apache/cassandra/io/SSTable.java
@@ -48,8 +48,6 @@ import org.apache.cassandra.utils.LogUtil;
 public class SSTable
 {
     private static Logger logger_ = Logger.getLogger(SSTable.class);
-    /* use this as a monitor to lock when loading index. */
-    private static Object indexLoadLock_ = new Object();
     /* Every 128th key is an index. */
     private static final int indexInterval_ = 128;
     /* Required extension for temporary files created during compactions. */
@@ -61,40 +59,14 @@ public class SSTable
      */
     private static Map<String, BloomFilter> bfs_ = new Hashtable<String, BloomFilter>();
 
+    private static FileSSTableMap openedFiles = new FileSSTableMap();
+
 
     public static int indexInterval()
     {
         return indexInterval_;
     }
 
-    /*
-     * Maintains a list of KeyPosition objects per SSTable file loaded.
-     * We do this so that we don't read the index file into memory multiple
-     * times.
-     *
-     * The positions here refer to positions _in the index file_, not in
-     * the sstable file directly.  So looking up a data location is a two
-     * step process: use this map to find the location of the original index
-     * entry, then read the data file at the location given by the index.
-    */
-    static IndexMap indexMetadataMap_ = new IndexMap();
-
-    /**
-     * This method deletes both the specified data file
-     * and the associated index file
-     *
-     * @param dataFile - data file associated with the SSTable
-     */
-    public static void delete(String dataFile) throws IOException
-    {
-        /* remove the cached index table from memory */
-        indexMetadataMap_.remove(dataFile);
-
-        deleteWithConfirm(new File(dataFile));
-        deleteWithConfirm(new File(indexFilename(dataFile)));
-        deleteWithConfirm(new File(filterFilename(dataFile)));
-    }
-
     private static void deleteWithConfirm(File file) throws IOException
     {
         assert file.exists() : "attempted to delete non-existing file " + file.getName();
@@ -104,19 +76,18 @@ public class SSTable
         }
     }
 
+    // todo can we refactor to take list of sstables?
     public static int getApproximateKeyCount(List<String> dataFiles)
     {
         int count = 0;
 
-        for (String dataFile : dataFiles)
+        for (String dataFileName : dataFiles)
         {
-            List<KeyPosition> index = indexMetadataMap_.get(dataFile);
-            if (index != null)
-            {
-                int indexKeyCount = index.size();
-                count = count + (indexKeyCount + 1) * indexInterval_;
-                logger_.debug("index size for bloom filter calc for file  : " + dataFile + "   : " + count);
-            }
+            SSTable sstable = openedFiles.get(dataFileName);
+            assert sstable != null;
+            int indexKeyCount = sstable.getIndexPositions().size();
+            count = count + (indexKeyCount + 1) * indexInterval_;
+            logger_.debug("index size for bloom filter calc for file  : " + dataFileName + "   : " + count);
         }
 
         return count;
@@ -127,21 +98,17 @@ public class SSTable
      */
     public static List<String> getIndexedKeys()
     {
-        Set<String> indexFiles = indexMetadataMap_.keySet();
-        List<KeyPosition> positions = new ArrayList<KeyPosition>();
-
-        for (String indexFile : indexFiles)
-        {
-            positions.addAll(indexMetadataMap_.get(indexFile));
-        }
-
         List<String> indexedKeys = new ArrayList<String>();
-        for (KeyPosition kp : positions)
-        {
-            indexedKeys.add(kp.key);
-        }
 
+        for (SSTable sstable : openedFiles.values())
+        {
+            for (KeyPosition kp : sstable.getIndexPositions())
+            {
+                indexedKeys.add(kp.key);
+            }
+        }
         Collections.sort(indexedKeys);
+
         return indexedKeys;
     }
 
@@ -204,17 +171,25 @@ public class SSTable
     private BufferedRandomAccessFile indexRAF_;
     private String lastWrittenKey_;
     private IPartitioner partitioner_;
+    List<KeyPosition> indexPositions_;
+
 
     public static synchronized SSTable open(String dataFileName, IPartitioner partitioner) throws IOException
     {
-        SSTable sstable = new SSTable(dataFileName, partitioner);
-        sstable.dataWriter_.close(); // todo this is dumb
-        if (indexMetadataMap_.get(dataFileName) == null)
+        SSTable sstable = openedFiles.get(dataFileName);
+        if (sstable == null)
         {
+            assert partitioner != null;
+            sstable = new SSTable(dataFileName, partitioner);
+            sstable.dataWriter_.close(); // todo this is dumb
+            sstable.indexRAF_.close();
+
             long start = System.currentTimeMillis();
             sstable.loadIndexFile();
             sstable.loadBloomFilter();
             logger_.debug("INDEX LOAD TIME for "  + dataFileName + ": " + (System.currentTimeMillis() - start) + " ms.");
+
+            openedFiles.put(dataFileName, sstable);
         }
         return sstable;
     }
@@ -242,6 +217,11 @@ public class SSTable
         return parts[0];
     }
 
+    public List<KeyPosition> getIndexPositions()
+    {
+        return indexPositions_;
+    }
+
     private void loadBloomFilter() throws IOException
     {
         assert bfs_.get(dataFile_) == null;
@@ -252,7 +232,7 @@ public class SSTable
     private void loadIndexFile() throws IOException
     {
         BufferedRandomAccessFile input = new BufferedRandomAccessFile(indexFilename(), "r");
-        ArrayList<KeyPosition> indexEntries = new ArrayList<KeyPosition>();
+        indexPositions_ = new ArrayList<KeyPosition>();
 
         int i = 0;
         long indexSize = input.length();
@@ -267,11 +247,9 @@ public class SSTable
             input.readLong();
             if (i++ % indexInterval_ == 0)
             {
-                indexEntries.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
+                indexPositions_.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
             }
         }
-
-        indexMetadataMap_.put(dataFile_, indexEntries);
     }
 
     private static String indexFilename(String dataFile)
@@ -339,13 +317,11 @@ public class SSTable
 
         if (keysWritten++ % indexInterval_ != 0)
             return;
-        List<KeyPosition> indexEntries = SSTable.indexMetadataMap_.get(dataFile_);
-        if (indexEntries == null)
+        if (indexPositions_ == null)
         {
-            indexEntries = new ArrayList<KeyPosition>();
-            SSTable.indexMetadataMap_.put(dataFile_, indexEntries);
+            indexPositions_ = new ArrayList<KeyPosition>();
         }
-        indexEntries.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
+        indexPositions_.add(new KeyPosition(decoratedKey, indexPosition, partitioner_));
         logger_.trace("wrote index of " + decoratedKey + " at " + indexPosition);
     }
 
@@ -367,7 +343,7 @@ public class SSTable
     /** get the position in the index file to start scanning to find the given key (at most indexInterval keys away) */
     private static long getIndexScanPosition(String decoratedKey, IFileReader dataReader, IPartitioner partitioner)
     {
-        List<KeyPosition> positions = indexMetadataMap_.get(dataReader.getFileName());
+        List<KeyPosition> positions = openedFiles.get(dataReader.getFileName()).getIndexPositions();
         assert positions != null && positions.size() > 0;
         int index = Collections.binarySearch(positions, new KeyPosition(decoratedKey, -1, partitioner));
         if (index < 0)
@@ -509,6 +485,16 @@ public class SSTable
         return next(clientKey, columnFamilyName, cnNames);
     }
 
+    private static String rename(String tmpFilename)
+    {
+        String filename = tmpFilename.replace("-" + temporaryFile_, "");
+        new File(tmpFilename).renameTo(new File(filename));
+        return filename;
+    }
+
+    /**
+     * Renames temporary SSTable files to valid data, index, and bloom filter files
+     */
     public void close(BloomFilter bf) throws IOException
     {
         // bloom filter
@@ -525,29 +511,12 @@ public class SSTable
 
         // main data
         dataWriter_.close(); // calls force
-    }
-    
-    private static String rename(String tmpFilename)
-    {
-        String filename = tmpFilename.replace("-" + temporaryFile_, "");
-        new File(tmpFilename).renameTo(new File(filename));
-        return filename;
-    }
 
-    /**
-     * Renames temporary SSTable files to valid data, index, and bloom filter files
-     */
-    public void closeRename(BloomFilter bf) throws IOException
-    {
-    	close(bf);
-
-        String oldDataFileName = dataFile_;
         rename(indexFilename());
         rename(filterFilename());
         dataFile_ = rename(dataFile_); // important to do this last since index & filter file names are derived from it
 
-    	List<KeyPosition> positions = SSTable.indexMetadataMap_.remove(oldDataFileName);
-    	SSTable.indexMetadataMap_.put(dataFile_, positions);
+        openedFiles.put(dataFile_, this);
     }
 
     /**
@@ -570,11 +539,29 @@ public class SSTable
         }
     }
 
+    public void delete() throws IOException
+    {
+        deleteWithConfirm(new File(dataFile_));
+        deleteWithConfirm(new File(indexFilename(dataFile_)));
+        deleteWithConfirm(new File(filterFilename(dataFile_)));
+    }
+
     /** obviously only for testing */
     public static void forceBloomFilterFailures(String filename)
     {
         SSTable.bfs_.put(filename, BloomFilter.alwaysMatchingBloomFilter());
     }
+
+    static void reopenUnsafe() throws IOException // testing only
+    {
+        Collection<SSTable> sstables = new ArrayList<SSTable>(openedFiles.values());
+        openedFiles.clear();
+        bfs_.clear();
+        for (SSTable sstable : sstables)
+        {
+            SSTable.open(sstable.dataFile_, sstable.partitioner_);
+        }
+    }
 }
 
 /**
@@ -608,19 +595,15 @@ class KeyPosition implements Comparable<KeyPosition>
     }
 }
 
-/**
- * wraps a Map to ensure that all filenames used as keys are cannonicalized.
- * (Note that cannonical paths are cached by the JDK so the performance hit is negligible.)
- */
-class IndexMap
+class FileSSTableMap
 {
-    private final Hashtable<String, List<KeyPosition>> hashtable = new Hashtable<String, List<KeyPosition>>();
+    private final HashMap<String, SSTable> map = new HashMap<String, SSTable>();
 
-    private String cannonicalize(String filename)
+    public SSTable get(String filename)
     {
         try
         {
-            return new File(filename).getCanonicalPath();
+            return map.get(new File(filename).getCanonicalPath());
         }
         catch (IOException e)
         {
@@ -628,28 +611,25 @@ class IndexMap
         }
     }
 
-    public List<KeyPosition> get(String filename)
+    public SSTable put(String filename, SSTable value)
     {
-        return hashtable.get(cannonicalize(filename));
+        try
+        {
+            return map.put(new File(filename).getCanonicalPath(), value);
+        }
+        catch (IOException e)
+        {
+            throw new RuntimeException(e);
+        }
     }
 
-    public List<KeyPosition> put(String filename, List<KeyPosition> value)
+    public Collection<SSTable> values()
     {
-        return hashtable.put(cannonicalize(filename), value);
+        return map.values();
     }
 
     public void clear()
     {
-        hashtable.clear();
+        map.clear();
     }
-
-    public Set<String> keySet()
-    {
-        return hashtable.keySet();
-    }
-
-    public List<KeyPosition> remove(String filename)
-    {
-        return hashtable.remove(cannonicalize(filename));
-    }
-}
+}
\ No newline at end of file
diff --git a/src/java/org/apache/cassandra/loader/Loader.java b/src/java/org/apache/cassandra/loader/Loader.java
index 01679b1921..a7ec4a2d18 100644
--- a/src/java/org/apache/cassandra/loader/Loader.java
+++ b/src/java/org/apache/cassandra/loader/Loader.java
@@ -39,7 +39,6 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.RowMutation;
 import org.apache.cassandra.db.Table;
 import org.apache.cassandra.io.SSTable;
-import org.apache.cassandra.locator.EndPointSnitch;
 import org.apache.cassandra.net.EndPoint;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.LogUtil;
@@ -48,7 +47,6 @@ import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.Token;
 import org.apache.lucene.analysis.TokenStream;
 import org.apache.lucene.analysis.standard.StandardAnalyzer;
-import org.apache.cassandra.utils.*;
 
 
 /**
@@ -336,7 +334,7 @@ public class Loader
             */
 
             // TODO Hmm need to double check here
-            SSTable.delete(ssTables.get(0));
+            SSTable.open(ssTables.get(0), null).delete();
             logger_.info("Finished all the requisite clean up ...");
         }
     }
diff --git a/src/java/org/apache/cassandra/loader/PreLoad.java b/src/java/org/apache/cassandra/loader/PreLoad.java
index e4e0bbe1f9..3612aa5d8c 100644
--- a/src/java/org/apache/cassandra/loader/PreLoad.java
+++ b/src/java/org/apache/cassandra/loader/PreLoad.java
@@ -19,7 +19,6 @@
 package org.apache.cassandra.loader;
 
 import java.io.BufferedReader;
-import java.io.File;
 import java.io.FileInputStream;
 import java.io.InputStreamReader;
 import java.util.List;
@@ -117,7 +116,7 @@ public class PreLoad
              * Do the cleanup necessary. Delete all commit logs and
              * the SSTables and reset the load state in the StorageService.
             */
-            SSTable.delete(ssTables.get(0));
+            SSTable.open(ssTables.get(0), null).delete();
         }
         logger_.info("Finished all the requisite clean up ...");
     }
diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java
index 7455d680d4..3db14c6c07 100644
--- a/test/unit/org/apache/cassandra/io/SSTableTest.java
+++ b/test/unit/org/apache/cassandra/io/SSTableTest.java
@@ -33,11 +33,10 @@ public class SSTableTest extends CleanupHelper
 {
     @Test
     public void testSingleWrite() throws IOException {
-        File f = File.createTempFile("sstable", "");
-        SSTable ssTable;
+        File f = File.createTempFile("sstable", "-" + SSTable.temporaryFile_);
 
         // write test data
-        ssTable = new SSTable(f.getParent(), f.getName(), new OrderPreservingPartitioner());
+        SSTable ssTable = new SSTable(f.getParent(), f.getName(), new OrderPreservingPartitioner());
         BloomFilter bf = new BloomFilter(1000, 8);
         Random random = new Random();
         byte[] bytes = new byte[1024];
@@ -49,14 +48,14 @@ public class SSTableTest extends CleanupHelper
         ssTable.close(bf);
 
         // verify
-        verifySingle(f, bytes, key);
-        SSTable.indexMetadataMap_.clear(); // force reloading the index
-        verifySingle(f, bytes, key);
+        verifySingle(ssTable.dataFile_, bytes, key);
+        SSTable.reopenUnsafe(); // force reloading the index
+        verifySingle(ssTable.dataFile_, bytes, key);
     }
 
-    private void verifySingle(File f, byte[] bytes, String key) throws IOException
+    private void verifySingle(String filename, byte[] bytes, String key) throws IOException
     {
-        SSTable ssTable = SSTable.open(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
+        SSTable ssTable = SSTable.open(filename, new OrderPreservingPartitioner());
         FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
         fs.seekTo(key);
         int size = fs.getBufIn().readInt();
@@ -67,8 +66,7 @@ public class SSTableTest extends CleanupHelper
 
     @Test
     public void testManyWrites() throws IOException {
-        File f = File.createTempFile("sstable", "");
-        SSTable ssTable;
+        File f = File.createTempFile("sstable", "-" + SSTable.temporaryFile_);
 
         TreeMap<String, byte[]> map = new TreeMap<String,byte[]>();
         for ( int i = 100; i < 1000; ++i )
@@ -77,7 +75,7 @@ public class SSTableTest extends CleanupHelper
         }
 
         // write
-        ssTable = new SSTable(f.getParent(), f.getName(), new OrderPreservingPartitioner());
+        SSTable ssTable = new SSTable(f.getParent(), f.getName(), new OrderPreservingPartitioner());
         BloomFilter bf = new BloomFilter(1000, 8);
         for (String key: map.navigableKeySet())
         {
@@ -86,16 +84,16 @@ public class SSTableTest extends CleanupHelper
         ssTable.close(bf);
 
         // verify
-        verifyMany(f, map);
-        SSTable.indexMetadataMap_.clear(); // force reloading the index
-        verifyMany(f, map);
+        verifyMany(ssTable.dataFile_, map);
+        SSTable.reopenUnsafe(); // force reloading the index
+        verifyMany(ssTable.dataFile_, map);
     }
 
-    private void verifyMany(File f, TreeMap<String, byte[]> map) throws IOException
+    private void verifyMany(String filename, TreeMap<String, byte[]> map) throws IOException
     {
         List<String> keys = new ArrayList(map.keySet());
         Collections.shuffle(keys);
-        SSTable ssTable = SSTable.open(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
+        SSTable ssTable = SSTable.open(filename, new OrderPreservingPartitioner());
         FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
         for (String key : keys)
         {