SERVER-13635: add a key/value -> storage engine interface to simplify other work

15 files changed, 1677 insertions, 7 deletions

diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.cpp b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
index 0259efb197b..a1048a4d401 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.cpp
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.cpp
@@ -124,7 +124,7 @@ namespace mongo {
 
     int BSONCollectionCatalogEntry::MetaData::findIndexOffset( const StringData& name ) const {
         for ( unsigned i = 0; i < indexes.size(); i++ )
-            if ( indexes[i].spec["name"].String() == name )
+            if ( indexes[i].name() == name )
                 return i;
         return -1;
     }
@@ -140,6 +140,17 @@ namespace mongo {
         return true;
     }
 
+    void BSONCollectionCatalogEntry::MetaData::rename( const StringData& toNS ) {
+        ns = toNS.toString();
+        for ( size_t i = 0; i < indexes.size(); i++ ) {
+            BSONObj spec = indexes[i].spec;
+            BSONObjBuilder b;
+            b.append( "ns", toNS );
+            b.appendElementsUnique( spec );
+            indexes[i].spec = b.obj();
+        }
+    }
+
     BSONObj BSONCollectionCatalogEntry::MetaData::toBSON() const {
         BSONObjBuilder b;
         b.append( "ns", ns );
diff --git a/src/mongo/db/storage/bson_collection_catalog_entry.h b/src/mongo/db/storage/bson_collection_catalog_entry.h
index 9b0645d6849..5af770ae4d6 100644
--- a/src/mongo/db/storage/bson_collection_catalog_entry.h
+++ b/src/mongo/db/storage/bson_collection_catalog_entry.h
@@ -77,6 +77,8 @@ namespace mongo {
 
             void updateTTLSetting( long long newExpireSeconds );
 
+            std::string name() const { return spec["name"].String(); }
+
             BSONObj spec;
             bool ready;
             DiskLoc head;
@@ -95,6 +97,8 @@ namespace mongo {
              */
             bool eraseIndex( const StringData& name );
 
+            void rename( const StringData& toNS );
+
             std::string ns;
             CollectionOptions options;
             std::vector<IndexMetaData> indexes;
diff --git a/src/mongo/db/storage/kv/SConscript b/src/mongo/db/storage/kv/SConscript
new file mode 100644
index 00000000000..039e3e9049e
--- /dev/null
+++ b/src/mongo/db/storage/kv/SConscript
@@ -0,0 +1,32 @@
+Import("env")
+
+env.Library(
+    target='kv_engine_core',
+    source=[
+        'kv_catalog.cpp',
+        ],
+    LIBDEPS=[
+        '$BUILD_DIR/mongo/db/storage/bson_collection_catalog_entry',
+        ]
+    )
+
+env.Library(
+    target='kv_engine',
+    source=[
+        'kv_collection_catalog_entry.cpp',
+        'kv_database_catalog_entry.cpp',
+        'kv_storage_engine.cpp',
+        ],
+    LIBDEPS=[
+        'kv_engine_core',
+        ]
+    )
+
+
+env.Library(
+    target='kv_engine_test_harness',
+    source=[
+        'kv_engine_test_harness.cpp',
+        ],
+    LIBDEPS=['kv_engine_core']
+    )
diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
new file mode 100644
index 00000000000..86b757e9163
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -0,0 +1,248 @@
+// kv_catalog.cpp
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/kv/kv_catalog.h"
+
+#include <stdlib.h>
+
+#include "mongo/db/storage/record_store.h"
+#include "mongo/platform/random.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+
+    KVCatalog::KVCatalog( RecordStore* rs )
+        : _rs( rs ) {
+        boost::scoped_ptr<SecureRandom> r( SecureRandom::create() );
+        _rand = r->nextInt64();
+    }
+
+    KVCatalog::~KVCatalog() {
+        _rs = NULL;
+    }
+
+    void KVCatalog::init( OperationContext* opCtx ) {
+        scoped_ptr<RecordIterator> it( _rs->getIterator( opCtx ) );
+        while ( !it->isEOF()  ) {
+            DiskLoc loc = it->getNext();
+            RecordData data = it->dataFor( loc );
+            BSONObj obj( data.data() );
+
+            // no locking needed since can only be one
+            string ns = obj["ns"].String();
+            string ident = obj["ident"].String();
+            _idents[ns] = Entry( ident, loc );
+        }
+    }
+
+    void KVCatalog::getAllCollections( std::vector<std::string>* out ) const {
+        boost::mutex::scoped_lock lk( _identsLock );
+        for ( NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it ) {
+            out->push_back( it->first );
+        }
+    }
+
+    Status KVCatalog::newCollection( OperationContext* opCtx,
+                                     const StringData& ns,
+                                     const CollectionOptions& options ) {
+        std::stringstream ss;
+        ss << ns << "-" << _rand << "-" << _next.fetchAndAdd( 1 );
+        string ident = ss.str();
+
+        boost::mutex::scoped_lock lk( _identsLock );
+        Entry& old = _idents[ns.toString()];
+        if ( !old.ident.empty() ) {
+            return Status( ErrorCodes::NamespaceExists, "collection already exists" );
+        }
+
+        BSONObj obj;
+        {
+            BSONObjBuilder b;
+            b.append( "ns", ns );
+            b.append( "ident", ident );
+            BSONCollectionCatalogEntry::MetaData md;
+            md.ns = ns.toString();
+            md.options = options;
+            b.append( "md", md.toBSON() );
+            obj = b.obj();
+        }
+        StatusWith<DiskLoc> res = _rs->insertRecord( opCtx, obj.objdata(), obj.objsize(), false );
+        if ( !res.isOK() )
+            return res.getStatus();
+
+        old = Entry( ident, res.getValue() );
+        LOG(1) << "stored meta data for " << ns << " @ " << res.getValue();
+        return Status::OK();
+    }
+
+    std::string KVCatalog::getCollectionIdent( const StringData& ns ) const {
+        boost::mutex::scoped_lock lk( _identsLock );
+        NSToIdentMap::const_iterator it = _idents.find( ns.toString() );
+        invariant( it != _idents.end() );
+        return it->second.ident;
+    }
+
+    std::string KVCatalog::getIndexIdent( OperationContext* opCtx,
+                                          const StringData& ns,
+                                          const StringData& idxName ) const {
+        DiskLoc loc;
+        BSONObj obj = _findEntry( opCtx, ns, &loc );
+        BSONObj idxIdent = obj["idxIdent"].Obj();
+        return idxIdent[idxName].String();
+    }
+
+    BSONObj KVCatalog::_findEntry( OperationContext* opCtx,
+                                   const StringData& ns,
+                                   DiskLoc* out ) const {
+        {
+            boost::mutex::scoped_lock lk( _identsLock );
+            NSToIdentMap::const_iterator it = _idents.find( ns.toString() );
+            invariant( it != _idents.end() );
+            *out = it->second.storedLoc;
+        }
+        LOG(1) << "looking up metadata for: " << ns << " @ " << *out;
+        RecordData data = _rs->dataFor( opCtx, *out );
+        return data.toBson().getOwned();
+    }
+
+    const BSONCollectionCatalogEntry::MetaData KVCatalog::getMetaData( OperationContext* opCtx,
+                                                                       const StringData& ns ) {
+        DiskLoc loc;
+        BSONObj obj = _findEntry( opCtx, ns, &loc );
+        LOG(1) << " got: " << obj;
+        BSONCollectionCatalogEntry::MetaData md;
+        if ( obj["md"].isABSONObj() )
+            md.parse( obj["md"].Obj() );
+        return md;
+    }
+
+    void KVCatalog::putMetaData( OperationContext* opCtx,
+                                 const StringData& ns,
+                                 BSONCollectionCatalogEntry::MetaData& md ) {
+        DiskLoc loc;
+        BSONObj obj = _findEntry( opCtx, ns, &loc );
+
+        {
+            // rebuilt doc
+            BSONObjBuilder b;
+            b.append( "md", md.toBSON() );
+
+            BSONObjBuilder newIdentMap;
+            BSONObj oldIdentMap;
+            if ( obj["idxIdent"].isABSONObj() )
+                oldIdentMap = obj["idxIdent"].Obj();
+
+            // fix ident map
+            for ( size_t i = 0; i < md.indexes.size(); i++ ) {
+                string name = md.indexes[i].name();
+                BSONElement e = oldIdentMap[name];
+                if ( e.type() == String ) {
+                    newIdentMap.append( e );
+                    continue;
+                }
+                // missing, create new
+                std::stringstream ss;
+                ss << getCollectionIdent( ns ) << '$' << name
+                   << '-' << _rand << '-' << _next.fetchAndAdd( 1 );
+                newIdentMap.append( name, ss.str() );
+            }
+            b.append( "idxIdent", newIdentMap.obj() );
+
+            // add whatever is left
+            b.appendElementsUnique( obj );
+            obj = b.obj();
+        }
+
+        StatusWith<DiskLoc> status = _rs->updateRecord( opCtx,
+                                                        loc,
+                                                        obj.objdata(),
+                                                        obj.objsize(),
+                                                        false,
+                                                        NULL );
+        fassert( 28521, status.getStatus() );
+        invariant( status.getValue() == loc );
+    }
+
+    Status KVCatalog::renameCollection( OperationContext* opCtx,
+                                        const StringData& fromNS,
+                                        const StringData& toNS,
+                                        bool stayTemp ) {
+        DiskLoc loc;
+        BSONObj old = _findEntry( opCtx, fromNS, &loc ).getOwned();
+        {
+            BSONObjBuilder b;
+
+            b.append( "ns", toNS );
+
+            BSONCollectionCatalogEntry::MetaData md;
+            md.parse( old["md"].Obj() );
+            md.rename( toNS );
+            if ( !stayTemp )
+                md.options.temp = false;
+            b.append( "md", md.toBSON() );
+
+            b.appendElementsUnique( old );
+
+            BSONObj obj = b.obj();
+            StatusWith<DiskLoc> status = _rs->updateRecord( opCtx,
+                                                            loc,
+                                                            obj.objdata(),
+                                                            obj.objsize(),
+                                                            false,
+                                                            NULL );
+            fassert( 28522, status.getStatus() );
+            invariant( status.getValue() == loc );
+        }
+
+
+        boost::mutex::scoped_lock lk( _identsLock );
+        _idents.erase( fromNS.toString() );
+        _idents[toNS.toString()] = Entry( old["ident"].String(), loc );
+
+        return Status::OK();
+    }
+
+    Status KVCatalog::dropCollection( OperationContext* opCtx,
+                                      const StringData& ns ) {
+        boost::mutex::scoped_lock lk( _identsLock );
+        Entry old = _idents[ns.toString()];
+        if ( old.ident.empty() ) {
+            return Status( ErrorCodes::NamespaceNotFound, "collection not found" );
+        }
+
+        LOG(1) << "deleting metadat for " << ns << " @ " << old.storedLoc;
+        _rs->deleteRecord( opCtx, old.storedLoc );
+        _idents.erase( ns.toString() );
+
+        return Status::OK();
+    }
+
+
+}
diff --git a/src/mongo/db/storage/kv/kv_catalog.h b/src/mongo/db/storage/kv/kv_catalog.h
new file mode 100644
index 00000000000..a5e348d53b9
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_catalog.h
@@ -0,0 +1,109 @@
+// kv_catalog.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <map>
+#include <string>
+
+#include <boost/scoped_ptr.hpp>
+#include <boost/thread/mutex.hpp>
+
+#include "mongo/base/string_data.h"
+#include "mongo/db/catalog/collection_options.h"
+#include "mongo/db/diskloc.h"
+#include "mongo/db/storage/bson_collection_catalog_entry.h"
+
+namespace mongo {
+
+    class OperationContext;
+    class RecordStore;
+
+    class KVCatalog {
+    public:
+        /**
+         * @param rs - does NOT take ownership
+         */
+        KVCatalog( RecordStore* rs );
+        ~KVCatalog();
+
+        void init( OperationContext* opCtx );
+
+        void getAllCollections( std::vector<std::string>* out ) const;
+
+        /**
+         * @return error or ident for instance
+         */
+        Status newCollection( OperationContext* opCtx,
+                              const StringData& ns,
+                              const CollectionOptions& options );
+
+        std::string getCollectionIdent( const StringData& ns ) const;
+
+        std::string getIndexIdent( OperationContext* opCtx,
+                                   const StringData& ns,
+                                   const StringData& idName ) const;
+
+        const BSONCollectionCatalogEntry::MetaData getMetaData( OperationContext* opCtx,
+                                                                const StringData& ns );
+        void putMetaData( OperationContext* opCtx,
+                          const StringData& ns,
+                          BSONCollectionCatalogEntry::MetaData& md );
+
+        Status renameCollection( OperationContext* opCtx,
+                                 const StringData& fromNS,
+                                 const StringData& toNS,
+                                 bool stayTemp );
+
+        Status dropCollection( OperationContext* opCtx,
+                               const StringData& ns );
+    private:
+
+        BSONObj _findEntry( OperationContext* opCtx,
+                            const StringData& ns,
+                            DiskLoc* out ) const;
+
+        RecordStore* _rs; // not owned
+        int64_t _rand;
+        AtomicUInt64 _next;
+
+        struct Entry {
+            Entry(){}
+            Entry( std::string i, DiskLoc l )
+                : ident(i), storedLoc( l ) {}
+            std::string ident;
+            DiskLoc storedLoc;
+        };
+        typedef std::map<std::string,Entry> NSToIdentMap;
+        NSToIdentMap _idents;
+        mutable boost::mutex _identsLock;
+    };
+
+}
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
new file mode 100644
index 00000000000..31d481a7c91
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.cpp
@@ -0,0 +1,123 @@
+// kv_collection_catalog_entry.cpp
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
+
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+
+namespace mongo {
+
+    KVCollectionCatalogEntry::KVCollectionCatalogEntry( KVEngine* engine,
+                                                        KVCatalog* catalog,
+                                                        const StringData& ns,
+                                                        const StringData& ident,
+                                                        RecordStore* rs)
+        : BSONCollectionCatalogEntry( ns ),
+          _engine( engine ),
+          _catalog( catalog ),
+          _ident( ident.toString() ),
+          _recrodStore( rs ) {
+    }
+
+    KVCollectionCatalogEntry::~KVCollectionCatalogEntry() {
+    }
+
+    bool KVCollectionCatalogEntry::setIndexIsMultikey(OperationContext* txn,
+                                                      const StringData& indexName,
+                                                      bool multikey ) {
+        MetaData md = _getMetaData(txn);
+
+        int offset = md.findIndexOffset( indexName );
+        invariant( offset >= 0 );
+        if ( md.indexes[offset].multikey == multikey )
+            return false;
+        md.indexes[offset].multikey = multikey;
+        _catalog->putMetaData( txn, ns().toString(), md );
+        return true;
+    }
+
+    void KVCollectionCatalogEntry::setIndexHead( OperationContext* txn,
+                                                 const StringData& indexName,
+                                                 const DiskLoc& newHead ) {
+        MetaData md = _getMetaData( txn );
+        int offset = md.findIndexOffset( indexName );
+        invariant( offset >= 0 );
+        md.indexes[offset].head = newHead;
+        _catalog->putMetaData( txn,  ns().toString(), md );
+    }
+
+    Status KVCollectionCatalogEntry::removeIndex( OperationContext* txn,
+                                                  const StringData& indexName ) {
+        string ident = _catalog->getIndexIdent( txn, ns().ns(), indexName );
+
+        MetaData md = _getMetaData( txn );
+        md.eraseIndex( indexName );
+        _catalog->putMetaData( txn, ns().toString(), md );
+
+        return _engine->dropSortedDataInterface( txn, ident );
+    }
+
+    Status KVCollectionCatalogEntry::prepareForIndexBuild( OperationContext* txn,
+                                                           const IndexDescriptor* spec ) {
+        MetaData md = _getMetaData( txn );
+        md.indexes.push_back( IndexMetaData( spec->infoObj(), false, DiskLoc(), false ) );
+        _catalog->putMetaData( txn, ns().toString(), md );
+
+        string ident = _catalog->getIndexIdent( txn, ns().ns(), spec->indexName() );
+
+        return _engine->createSortedDataInterface( txn, ident, spec );
+    }
+
+    void KVCollectionCatalogEntry::indexBuildSuccess( OperationContext* txn,
+                                                      const StringData& indexName ) {
+        MetaData md = _getMetaData( txn );
+        int offset = md.findIndexOffset( indexName );
+        invariant( offset >= 0 );
+        md.indexes[offset].ready = true;
+        _catalog->putMetaData( txn, ns().toString(), md );
+    }
+
+    void KVCollectionCatalogEntry::updateTTLSetting( OperationContext* txn,
+                                                     const StringData& idxName,
+                                                     long long newExpireSeconds ) {
+        MetaData md = _getMetaData( txn );
+        int offset = md.findIndexOffset( idxName );
+        invariant( offset >= 0 );
+        md.indexes[offset].updateTTLSetting( newExpireSeconds );
+        _catalog->putMetaData( txn, ns().toString(), md );
+    }
+
+    BSONCollectionCatalogEntry::MetaData KVCollectionCatalogEntry::_getMetaData( OperationContext* txn ) const {
+        return _catalog->getMetaData( txn, ns().toString() );
+    }
+
+}
diff --git a/src/mongo/db/storage/kv/kv_collection_catalog_entry.h b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
new file mode 100644
index 00000000000..ccccc1fe5fa
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_collection_catalog_entry.h
@@ -0,0 +1,91 @@
+// kv_collection_catalog_entry.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/catalog/collection_catalog_entry.h"
+#include "mongo/db/storage/bson_collection_catalog_entry.h"
+#include "mongo/db/storage/record_store.h"
+
+namespace mongo {
+
+    class KVCatalog;
+    class KVEngine;
+
+    class KVCollectionCatalogEntry : public BSONCollectionCatalogEntry {
+    public:
+        KVCollectionCatalogEntry( KVEngine* engine,
+                                  KVCatalog* catalog,
+                                  const StringData& ns,
+                                  const StringData& ident,
+                                  RecordStore* rs );
+
+        virtual ~KVCollectionCatalogEntry();
+
+        virtual int getMaxAllowedIndexes() const { return 64; };
+
+        virtual bool setIndexIsMultikey(OperationContext* txn,
+                                        const StringData& indexName,
+                                        bool multikey = true);
+
+        virtual void setIndexHead( OperationContext* txn,
+                                   const StringData& indexName,
+                                   const DiskLoc& newHead );
+
+        virtual Status removeIndex( OperationContext* txn,
+                                    const StringData& indexName );
+
+        virtual Status prepareForIndexBuild( OperationContext* txn,
+                                             const IndexDescriptor* spec );
+
+        virtual void indexBuildSuccess( OperationContext* txn,
+                                        const StringData& indexName );
+
+        /* Updates the expireAfterSeconds field of the given index to the value in newExpireSecs.
+         * The specified index must already contain an expireAfterSeconds field, and the value in
+         * that field and newExpireSecs must both be numeric.
+         */
+        virtual void updateTTLSetting( OperationContext* txn,
+                                       const StringData& idxName,
+                                       long long newExpireSeconds );
+
+        RecordStore* getRecordStore() { return _recrodStore.get(); }
+
+    protected:
+        virtual MetaData _getMetaData( OperationContext* txn ) const;
+
+    private:
+        KVEngine* _engine; // not owned
+        KVCatalog* _catalog; // not owned
+        std::string _ident;
+        boost::scoped_ptr<RecordStore> _recrodStore; // owned
+    };
+
+}
diff --git a/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp b/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
new file mode 100644
index 00000000000..29431ccc2e8
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_database_catalog_entry.cpp
@@ -0,0 +1,266 @@
+// kv_database_catalog_entry.cpp
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/kv/kv_database_catalog_entry.h"
+
+#include "mongo/db/catalog/index_catalog_entry.h"
+#include "mongo/db/index/2d_access_method.h"
+#include "mongo/db/index/btree_access_method.h"
+#include "mongo/db/index/fts_access_method.h"
+#include "mongo/db/index/hash_access_method.h"
+#include "mongo/db/index/haystack_access_method.h"
+#include "mongo/db/index/index_access_method.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/index/s2_access_method.h"
+#include "mongo/db/storage/kv/kv_collection_catalog_entry.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/db/storage/kv/kv_storage_engine.h"
+#include "mongo/util/assert_util.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+    KVDatabaseCatalogEntry::KVDatabaseCatalogEntry( const StringData& db, KVStorageEngine* engine )
+        : DatabaseCatalogEntry( db ), _engine( engine ) {
+
+    }
+
+    KVDatabaseCatalogEntry::~KVDatabaseCatalogEntry() {
+        for ( CollectionMap::const_iterator it = _collections.begin(); it != _collections.end(); ++it ) {
+            delete it->second;
+        }
+        _collections.clear();
+    }
+
+    bool KVDatabaseCatalogEntry::exists() const {
+        return !isEmpty();
+    }
+
+    bool KVDatabaseCatalogEntry::isEmpty() const {
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        return _collections.empty();
+    }
+
+    void KVDatabaseCatalogEntry::appendExtraStats( OperationContext* opCtx,
+                                                   BSONObjBuilder* out,
+                                                   double scale ) const {
+        // todo
+    }
+
+    bool KVDatabaseCatalogEntry::currentFilesCompatible( OperationContext* opCtx ) const {
+        // todo
+        return true;
+    }
+
+    void KVDatabaseCatalogEntry::getCollectionNamespaces( std::list<std::string>* out ) const {
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        for ( CollectionMap::const_iterator it = _collections.begin(); it != _collections.end(); ++it ) {
+            out->push_back( it->first );
+        }
+    }
+
+    CollectionCatalogEntry* KVDatabaseCatalogEntry::getCollectionCatalogEntry( OperationContext* txn,
+                                                                               const StringData& ns ) const {
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        CollectionMap::const_iterator it = _collections.find( ns.toString() );
+        if ( it == _collections.end() )
+            return NULL;
+        return it->second;
+    }
+
+    RecordStore* KVDatabaseCatalogEntry::getRecordStore( OperationContext* txn,
+                                                         const StringData& ns ) {
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        CollectionMap::const_iterator it = _collections.find( ns.toString() );
+        if ( it == _collections.end() )
+            return NULL;
+        return it->second->getRecordStore();
+    }
+
+    IndexAccessMethod* KVDatabaseCatalogEntry::getIndex( OperationContext* txn,
+                                                         const CollectionCatalogEntry* collection,
+                                                         IndexCatalogEntry* index ) {
+        IndexDescriptor* desc = index->descriptor();
+
+        const string& type = desc->getAccessMethodName();
+
+        string ident = _engine->getCatalog()->getIndexIdent( txn,
+                                                             collection->ns().ns(),
+                                                             desc->indexName() );
+
+        SortedDataInterface* sdi =
+            _engine->getEngine()->getSortedDataInterface( txn, ident, desc );
+
+        if ("" == type)
+            return new BtreeAccessMethod( index, sdi );
+
+        if (IndexNames::HASHED == type)
+            return new HashAccessMethod( index, sdi );
+
+        if (IndexNames::GEO_2DSPHERE == type)
+            return new S2AccessMethod( index, sdi );
+
+        if (IndexNames::TEXT == type)
+            return new FTSAccessMethod( index, sdi );
+
+        if (IndexNames::GEO_HAYSTACK == type)
+            return new HaystackAccessMethod( index, sdi );
+
+        if (IndexNames::GEO_2D == type)
+            return new TwoDAccessMethod( index, sdi );
+
+        log() << "Can't find index for keyPattern " << desc->keyPattern();
+        invariant( false );
+    }
+
+    Status KVDatabaseCatalogEntry::createCollection( OperationContext* txn,
+                                                     const StringData& ns,
+                                                     const CollectionOptions& options,
+                                                     bool allocateDefaultSpace ) {
+        // we assume there is a logical lock on the collection name above
+
+        {
+            boost::mutex::scoped_lock lk( _collectionsLock );
+            if ( _collections[ns.toString()] ) {
+                return Status( ErrorCodes::NamespaceExists,
+                               "collection already exists" );
+            }
+        }
+
+        // need to create it
+        Status status = _engine->getCatalog()->newCollection( txn, ns, options );
+        if ( !status.isOK() )
+            return status;
+
+        string ident = _engine->getCatalog()->getCollectionIdent( ns );
+
+        status = _engine->getEngine()->createRecordStore( txn, ident, options );
+        if ( !status.isOK() )
+            return status;
+
+        RecordStore* rs = _engine->getEngine()->getRecordStore( txn, ns, ident, options );
+        invariant( rs );
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        _collections[ns.toString()] =
+            new KVCollectionCatalogEntry( _engine->getEngine(), _engine->getCatalog(),
+                                          ns, ident, rs );
+
+        return Status::OK();
+    }
+
+    void KVDatabaseCatalogEntry::initCollection( OperationContext* opCtx,
+                                                 const std::string& ns ) {
+        string ident = _engine->getCatalog()->getCollectionIdent( ns );
+        BSONCollectionCatalogEntry::MetaData md = _engine->getCatalog()->getMetaData( opCtx, ns );
+
+        RecordStore* rs = _engine->getEngine()->getRecordStore( opCtx, ns, ident, md.options );
+        invariant( rs );
+
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        invariant( !_collections[ns] );
+        _collections[ns] = new KVCollectionCatalogEntry( _engine->getEngine(),
+                                                         _engine->getCatalog(),
+                                                         ns,
+                                                         ident,
+                                                         rs );
+    }
+
+    Status KVDatabaseCatalogEntry::renameCollection( OperationContext* txn,
+                                                     const StringData& fromNS,
+                                                     const StringData& toNS,
+                                                     bool stayTemp ) {
+        {
+            boost::mutex::scoped_lock lk( _collectionsLock );
+            CollectionMap::const_iterator it = _collections.find( fromNS.toString() );
+            if ( it == _collections.end() )
+                return Status( ErrorCodes::NamespaceNotFound, "rename cannot find collection" );
+            KVCollectionCatalogEntry* fromEntry = it->second;
+
+            it = _collections.find( toNS.toString() );
+            if ( it != _collections.end() )
+                return Status( ErrorCodes::NamespaceExists, "for rename to already exists" );
+
+            _collections.erase( fromNS.toString() );
+            delete fromEntry;
+        }
+
+        Status status = _engine->getCatalog()->renameCollection( txn, fromNS, toNS, stayTemp );
+        if ( !status.isOK() )
+            return status;
+
+        string ident = _engine->getCatalog()->getCollectionIdent( toNS );
+        BSONCollectionCatalogEntry::MetaData md = _engine->getCatalog()->getMetaData( txn, toNS );
+        RecordStore* rs = _engine->getEngine()->getRecordStore( txn, toNS, ident, md.options );
+
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        _collections[toNS.toString()] =
+            new KVCollectionCatalogEntry( _engine->getEngine(), _engine->getCatalog(),
+                                          toNS, ident, rs );
+
+        return Status::OK();
+    }
+
+    Status KVDatabaseCatalogEntry::dropCollection( OperationContext* opCtx,
+                                                   const StringData& ns ) {
+        KVCollectionCatalogEntry* entry;
+        {
+            boost::mutex::scoped_lock lk( _collectionsLock );
+            CollectionMap::const_iterator it = _collections.find( ns.toString() );
+            if ( it == _collections.end() )
+                return Status( ErrorCodes::NamespaceNotFound, "cannnot find collection to drop" );
+            entry = it->second;
+        }
+
+        invariant( entry->getTotalIndexCount( opCtx ) == entry->getCompletedIndexCount( opCtx ) );
+        {
+            std::vector<std::string> indexNames;
+            entry->getAllIndexes( opCtx, &indexNames );
+            for ( size_t i = 0; i < indexNames.size(); i++ ) {
+                entry->removeIndex( opCtx, indexNames[i] );
+            }
+        }
+        invariant( entry->getTotalIndexCount( opCtx ) == 0 );
+
+        string ident = _engine->getCatalog()->getCollectionIdent( ns );
+
+        Status status = _engine->getEngine()->dropRecordStore( opCtx, ident );
+        if ( !status.isOK() )
+            return status;
+
+        status = _engine->getCatalog()->dropCollection( opCtx, ns );
+        if ( !status.isOK() )
+            return status;
+
+        boost::mutex::scoped_lock lk( _collectionsLock );
+        _collections.erase( ns.toString() );
+
+        return Status::OK();
+    }
+
+}
diff --git a/src/mongo/db/storage/kv/kv_database_catalog_entry.h b/src/mongo/db/storage/kv/kv_database_catalog_entry.h
new file mode 100644
index 00000000000..d8af6aec048
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_database_catalog_entry.h
@@ -0,0 +1,101 @@
+// kv_database_catalog_entry.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <map>
+#include <string>
+
+#include <boost/scoped_ptr.hpp>
+#include <boost/thread/mutex.hpp>
+
+#include "mongo/db/catalog/database_catalog_entry.h"
+
+namespace mongo {
+
+    class KVCollectionCatalogEntry;
+    class KVStorageEngine;
+
+    class KVDatabaseCatalogEntry : public DatabaseCatalogEntry {
+    public:
+        KVDatabaseCatalogEntry( const StringData& db, KVStorageEngine* engine );
+        virtual ~KVDatabaseCatalogEntry();
+
+        virtual bool exists() const;
+        virtual bool isEmpty() const;
+
+        virtual void appendExtraStats( OperationContext* opCtx,
+                                       BSONObjBuilder* out,
+                                       double scale ) const;
+
+        virtual bool isOlderThan24( OperationContext* opCtx ) const { return false; }
+        virtual void markIndexSafe24AndUp( OperationContext* opCtx ) {}
+
+        virtual bool currentFilesCompatible( OperationContext* opCtx ) const;
+
+        virtual void getCollectionNamespaces( std::list<std::string>* out ) const;
+
+        virtual CollectionCatalogEntry* getCollectionCatalogEntry( OperationContext* txn,
+                                                                   const StringData& ns ) const;
+
+        virtual RecordStore* getRecordStore( OperationContext* txn,
+                                             const StringData& ns );
+
+        virtual IndexAccessMethod* getIndex( OperationContext* txn,
+                                             const CollectionCatalogEntry* collection,
+                                             IndexCatalogEntry* index );
+
+        virtual Status createCollection( OperationContext* txn,
+                                         const StringData& ns,
+                                         const CollectionOptions& options,
+                                         bool allocateDefaultSpace );
+
+        virtual Status renameCollection( OperationContext* txn,
+                                         const StringData& fromNS,
+                                         const StringData& toNS,
+                                         bool stayTemp );
+
+        virtual Status dropCollection( OperationContext* opCtx,
+                                       const StringData& ns );
+
+        // --------------
+
+        void initCollection( OperationContext* opCtx,
+                             const std::string& ns );
+
+    private:
+        KVStorageEngine* _engine; // not owned here
+        bool _used;
+
+        typedef std::map<std::string,KVCollectionCatalogEntry*> CollectionMap;
+        CollectionMap _collections;
+        mutable boost::mutex _collectionsLock;
+    };
+}
diff --git a/src/mongo/db/storage/kv/kv_engine.h b/src/mongo/db/storage/kv/kv_engine.h
new file mode 100644
index 00000000000..eb3368e7c7a
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_engine.h
@@ -0,0 +1,89 @@
+// kv_engine.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/base/status.h"
+#include "mongo/base/string_data.h"
+#include "mongo/db/catalog/collection_options.h"
+
+namespace mongo {
+
+    class IndexDescriptor;
+    class OperationContext;
+    class RecordStore;
+    class RecoveryUnit;
+    class SortedDataInterface;
+
+    class KVEngine {
+    public:
+
+        virtual ~KVEngine() {}
+
+        virtual RecoveryUnit* newRecoveryUnit() = 0;
+
+        // ---------
+
+        /**
+         * @param ident Ident is a one time use string. It is used for this instance
+         *              and never again.
+         */
+        virtual Status createRecordStore( OperationContext* opCtx,
+                                          const StringData& ident,
+                                          const CollectionOptions& options ) = 0;
+
+        /**
+         * Caller takes ownership
+         * Having multiple out for the same ns is a rules violation;
+         * Calling on a non-created ident is invalid and may crash.
+         */
+        virtual RecordStore* getRecordStore( OperationContext* opCtx,
+                                             const StringData& ns,
+                                             const StringData& ident,
+                                             const CollectionOptions& options ) = 0;
+
+        virtual Status dropRecordStore( OperationContext* opCtx,
+                                        const StringData& ident ) = 0;
+
+        // --------
+
+        virtual Status createSortedDataInterface( OperationContext* opCtx,
+                                                  const StringData& ident,
+                                                  const IndexDescriptor* desc ) = 0;
+
+        virtual SortedDataInterface* getSortedDataInterface( OperationContext* opCtx,
+                                                             const StringData& ident,
+                                                             const IndexDescriptor* desc ) = 0;
+
+        virtual Status dropSortedDataInterface( OperationContext* opCtx,
+                                                const StringData& ident ) = 0;
+    };
+
+}
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.cpp b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
new file mode 100644
index 00000000000..9483b6839c5
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.cpp
@@ -0,0 +1,267 @@
+// kv_engine_test_harness.cpp
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/kv/kv_engine_test_harness.h"
+
+#include "mongo/db/operation_context_noop.h"
+#include "mongo/db/index/index_descriptor.h"
+#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/db/storage/record_store.h"
+#include "mongo/db/storage/sorted_data_interface.h"
+#include "mongo/unittest/unittest.h"
+
+namespace mongo {
+
+    namespace {
+        class MyOperationContext : public OperationContextNoop {
+        public:
+            MyOperationContext( KVEngine* engine )
+                : OperationContextNoop( engine->newRecoveryUnit() ) {
+            }
+        };
+    }
+
+    TEST( KVEngineTestHarness, SimpleRS1 ) {
+        scoped_ptr<KVHarnessHelper> helper( KVHarnessHelper::create() );
+        KVEngine* engine = helper->getEngine();
+        ASSERT( engine );
+
+        string ns = "a.b";
+        scoped_ptr<RecordStore> rs;
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_OK( engine->createRecordStore( &opCtx, ns, CollectionOptions() ) );
+            rs.reset( engine->getRecordStore( &opCtx, ns, ns, CollectionOptions() ) );
+            ASSERT( rs );
+        }
+
+
+        DiskLoc loc;
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            StatusWith<DiskLoc> res = rs->insertRecord( &opCtx, "abc", 4, false );
+            ASSERT_OK( res.getStatus() );
+            loc = res.getValue();
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_EQUALS( string("abc"), rs->dataFor( &opCtx, loc ).data() );
+        }
+
+    }
+
+    TEST( KVEngineTestHarness, Restart1 ) {
+        scoped_ptr<KVHarnessHelper> helper( KVHarnessHelper::create() );
+        KVEngine* engine = helper->getEngine();
+        ASSERT( engine );
+
+        string ns = "a.b";
+        scoped_ptr<RecordStore> rs;
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_OK( engine->createRecordStore( &opCtx, ns, CollectionOptions() ) );
+            rs.reset( engine->getRecordStore( &opCtx, ns, ns, CollectionOptions() ) );
+            ASSERT( rs );
+        }
+
+
+        DiskLoc loc;
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            StatusWith<DiskLoc> res = rs->insertRecord( &opCtx, "abc", 4, false );
+            ASSERT_OK( res.getStatus() );
+            loc = res.getValue();
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_EQUALS( string("abc"), rs->dataFor( &opCtx, loc ).data() );
+        }
+
+        engine = helper->restartEngine();
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_EQUALS( string("abc"), rs->dataFor( &opCtx, loc ).data() );
+        }
+
+    }
+
+
+    TEST( KVEngineTestHarness, SimpleSorted1 ) {
+        scoped_ptr<KVHarnessHelper> helper( KVHarnessHelper::create() );
+        KVEngine* engine = helper->getEngine();
+        ASSERT( engine );
+
+        string ident = "abc";
+        IndexDescriptor desc( NULL, "", BSON( "key" << BSON( "a" << 1 ) ) );
+        scoped_ptr<SortedDataInterface> sorted;
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_OK( engine->createSortedDataInterface( &opCtx, ident, &desc ) );
+            sorted.reset( engine->getSortedDataInterface( &opCtx, ident, &desc ) );
+            ASSERT( sorted );
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            ASSERT_OK( sorted->insert( &opCtx, BSON( "" << 5 ), DiskLoc( 6, 4 ), true ) );
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_EQUALS( 1, sorted->numEntries( &opCtx ) );
+        }
+
+    }
+
+    TEST( KVCatalogTest, Coll1 ) {
+        scoped_ptr<KVHarnessHelper> helper( KVHarnessHelper::create() );
+        KVEngine* engine = helper->getEngine();
+
+        scoped_ptr<RecordStore> rs;
+        scoped_ptr<KVCatalog> catalog;
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            ASSERT_OK( engine->createRecordStore( &opCtx, "catalog", CollectionOptions() ) );
+            rs.reset( engine->getRecordStore( &opCtx, "catalog", "catalog", CollectionOptions() ) );
+            catalog.reset( new KVCatalog( rs.get() ) );
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            ASSERT_OK( catalog->newCollection( &opCtx, "a.b", CollectionOptions() ) );
+            ASSERT_NOT_EQUALS( "a.b", catalog->getCollectionIdent( "a.b" ) );
+            uow.commit();
+        }
+
+        string ident = catalog->getCollectionIdent( "a.b" );
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            catalog.reset( new KVCatalog( rs.get() ) );
+            catalog->init( &opCtx );
+            uow.commit();
+        }
+        ASSERT_EQUALS( ident, catalog->getCollectionIdent( "a.b" ) );
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            catalog->dropCollection( &opCtx, "a.b" );
+            catalog->newCollection( &opCtx, "a.b", CollectionOptions() );
+            uow.commit();
+        }
+        ASSERT_NOT_EQUALS( ident, catalog->getCollectionIdent( "a.b" ) );
+    }
+
+
+    TEST( KVCatalogTest, Idx1 ) {
+        scoped_ptr<KVHarnessHelper> helper( KVHarnessHelper::create() );
+        KVEngine* engine = helper->getEngine();
+
+        scoped_ptr<RecordStore> rs;
+        scoped_ptr<KVCatalog> catalog;
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            ASSERT_OK( engine->createRecordStore( &opCtx, "catalog", CollectionOptions() ) );
+            rs.reset( engine->getRecordStore( &opCtx, "catalog", "catalog", CollectionOptions() ) );
+            catalog.reset( new KVCatalog( rs.get() ) );
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+            ASSERT_OK( catalog->newCollection( &opCtx, "a.b", CollectionOptions() ) );
+            ASSERT_NOT_EQUALS( "a.b", catalog->getCollectionIdent( "a.b" ) );
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+
+            BSONCollectionCatalogEntry::MetaData md;
+            md.ns ="a.b";
+            md.indexes.push_back( BSONCollectionCatalogEntry::IndexMetaData( BSON( "name" << "foo" ),
+                                                                             false,
+                                                                             DiskLoc(),
+                                                                             false ) );
+            catalog->putMetaData( &opCtx, "a.b", md );
+            uow.commit();
+        }
+
+        string idxIndent;
+        {
+            MyOperationContext opCtx( engine );
+            idxIndent = catalog->getIndexIdent( &opCtx, "a.b", "foo" );
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_EQUALS( idxIndent, catalog->getIndexIdent( &opCtx, "a.b", "foo" ) );
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            WriteUnitOfWork uow( &opCtx );
+
+            BSONCollectionCatalogEntry::MetaData md;
+            md.ns ="a.b";
+            catalog->putMetaData( &opCtx, "a.b", md ); // remove index
+            md.indexes.push_back( BSONCollectionCatalogEntry::IndexMetaData( BSON( "name" << "foo" ),
+                                                                             false,
+                                                                             DiskLoc(),
+                                                                             false ) );
+            catalog->putMetaData( &opCtx, "a.b", md );
+            uow.commit();
+        }
+
+        {
+            MyOperationContext opCtx( engine );
+            ASSERT_NOT_EQUALS( idxIndent, catalog->getIndexIdent( &opCtx, "a.b", "foo" ) );
+        }
+
+    }
+
+}
diff --git a/src/mongo/db/storage/kv/kv_engine_test_harness.h b/src/mongo/db/storage/kv/kv_engine_test_harness.h
new file mode 100644
index 00000000000..fd828681cc7
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_engine_test_harness.h
@@ -0,0 +1,47 @@
+// kv_engine_test_harness.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include "mongo/db/storage/kv/kv_engine.h"
+
+namespace mongo {
+    class KVHarnessHelper {
+    public:
+        virtual ~KVHarnessHelper(){}
+
+        // returns same thing for entire life
+        virtual KVEngine* getEngine() = 0;
+
+        virtual KVEngine* restartEngine() = 0;
+
+        static KVHarnessHelper* create();
+    };
+}
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.cpp b/src/mongo/db/storage/kv/kv_storage_engine.cpp
new file mode 100644
index 00000000000..ace4483faac
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_storage_engine.cpp
@@ -0,0 +1,180 @@
+// kv_storage_engine.cpp
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#include "mongo/db/storage/kv/kv_storage_engine.h"
+
+#include "mongo/db/operation_context_noop.h"
+#include "mongo/db/storage/kv/kv_database_catalog_entry.h"
+#include "mongo/db/storage/kv/kv_engine.h"
+#include "mongo/util/log.h"
+
+namespace mongo {
+
+    namespace {
+        std::string catalogInfo = "_mdb_catalog";
+    }
+
+    KVStorageEngine::KVStorageEngine( KVEngine* engine )
+        : _engine( engine ), _initialized( false ) {
+    }
+
+    void KVStorageEngine::cleanShutdown(OperationContext* txn) {
+
+        for ( DBMap::const_iterator it = _dbs.begin(); it != _dbs.end(); ++it ) {
+            delete it->second;
+        }
+        _dbs.clear();
+
+        _catalog.reset( NULL );
+        _catalogRecordStore.reset( NULL );
+
+        _engine.reset( NULL );
+    }
+
+    KVStorageEngine::~KVStorageEngine() {
+    }
+
+    void KVStorageEngine::finishInit() {
+        if ( _initialized )
+            return;
+
+        OperationContextNoop opCtx( _engine->newRecoveryUnit() );
+        WriteUnitOfWork uow( &opCtx );
+
+        Status status = _engine->createRecordStore( &opCtx, catalogInfo, CollectionOptions() );
+        fassert( 28520, status );
+
+        _catalogRecordStore.reset( _engine->getRecordStore( &opCtx,
+                                                            catalogInfo,
+                                                            catalogInfo,
+                                                            CollectionOptions() ) );
+        _catalog.reset( new KVCatalog( _catalogRecordStore.get() ) );
+        _catalog->init( &opCtx );
+
+        std::vector<std::string> collections;
+        _catalog->getAllCollections( &collections );
+
+        for ( size_t i = 0; i < collections.size(); i++ ) {
+            std::string coll = collections[i];
+            NamespaceString nss( coll );
+            string dbName = nss.db().toString();
+
+            KVDatabaseCatalogEntry*& db = _dbs[dbName];
+            if ( !db ) {
+                db = new KVDatabaseCatalogEntry( dbName, this );
+            }
+            db->initCollection( &opCtx, coll );
+        }
+
+        uow.commit();
+
+        _initialized = true;
+    }
+
+    RecoveryUnit* KVStorageEngine::newRecoveryUnit( OperationContext* opCtx ) {
+        invariant( _initialized );
+        if ( !_engine ) {
+            // shutdown
+            return NULL;
+        }
+        return _engine->newRecoveryUnit();
+    }
+
+    void KVStorageEngine::listDatabases( std::vector<std::string>* out ) const {
+        invariant( _initialized );
+        boost::mutex::scoped_lock lk( _dbsLock );
+        for ( DBMap::const_iterator it = _dbs.begin(); it != _dbs.end(); ++it ) {
+            if ( it->second->isEmpty() )
+                continue;
+            out->push_back( it->first );
+        }
+    }
+
+    DatabaseCatalogEntry* KVStorageEngine::getDatabaseCatalogEntry( OperationContext* opCtx,
+                                                                    const StringData& dbName ) {
+        invariant( _initialized );
+        boost::mutex::scoped_lock lk( _dbsLock );
+        KVDatabaseCatalogEntry*& db = _dbs[dbName.toString()];
+        if ( !db ) {
+            db = new KVDatabaseCatalogEntry( dbName, this );
+        }
+        return db;
+    }
+
+    Status KVStorageEngine::closeDatabase( OperationContext* txn, const StringData& db ) {
+        invariant( _initialized );
+        // todo: do I have to suppor this?
+        return Status::OK();
+    }
+
+    Status KVStorageEngine::dropDatabase( OperationContext* txn, const StringData& db ) {
+        invariant( _initialized );
+
+        KVDatabaseCatalogEntry* entry;
+        {
+            boost::mutex::scoped_lock lk( _dbsLock );
+            DBMap::const_iterator it = _dbs.find( db.toString() );
+            if ( it == _dbs.end() )
+                return Status( ErrorCodes::NamespaceNotFound, "db not found to drop" );
+            entry = it->second;
+        }
+
+        std::list<std::string> toDrop;
+        entry->getCollectionNamespaces( &toDrop );
+
+        for ( std::list<std::string>::iterator it = toDrop.begin(); it != toDrop.end(); ++it ) {
+            string coll = *it;
+            entry->dropCollection( txn, coll );
+        }
+        toDrop.clear();
+        entry->getCollectionNamespaces( &toDrop );
+        invariant( toDrop.empty() );
+
+        {
+            boost::mutex::scoped_lock lk( _dbsLock );
+            _dbs.erase( db.toString() );
+        }
+        return Status::OK();
+    }
+
+    int KVStorageEngine::flushAllFiles( bool sync ) {
+        // todo: do I have to support this?
+        return 0;
+    }
+
+    Status KVStorageEngine::repairDatabase( OperationContext* txn,
+                                            const std::string& dbName,
+                                            bool preserveClonedFilesOnFailure,
+                                            bool backupOriginalFiles ) {
+        // todo: do I have to support this?
+        return Status::OK();
+    }
+
+}
diff --git a/src/mongo/db/storage/kv/kv_storage_engine.h b/src/mongo/db/storage/kv/kv_storage_engine.h
new file mode 100644
index 00000000000..b1dba3a537c
--- /dev/null
+++ b/src/mongo/db/storage/kv/kv_storage_engine.h
@@ -0,0 +1,101 @@
+// kv_storage_engine.h
+
+/**
+ *    Copyright (C) 2014 MongoDB Inc.
+ *
+ *    This program is free software: you can redistribute it and/or  modify
+ *    it under the terms of the GNU Affero General Public License, version 3,
+ *    as published by the Free Software Foundation.
+ *
+ *    This program is distributed in the hope that it will be useful,
+ *    but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *    GNU Affero General Public License for more details.
+ *
+ *    You should have received a copy of the GNU Affero General Public License
+ *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ *    As a special exception, the copyright holders give permission to link the
+ *    code of portions of this program with the OpenSSL library under certain
+ *    conditions as described in each individual source file and distribute
+ *    linked combinations including the program with the OpenSSL library. You
+ *    must comply with the GNU Affero General Public License in all respects for
+ *    all of the code used other than as permitted herein. If you modify file(s)
+ *    with this exception, you may extend this exception to your version of the
+ *    file(s), but you are not obligated to do so. If you do not wish to do so,
+ *    delete this exception statement from your version. If you delete this
+ *    exception statement from all source files in the program, then also delete
+ *    it in the license file.
+ */
+
+#pragma once
+
+#include <map>
+#include <string>
+
+#include <boost/scoped_ptr.hpp>
+#include <boost/thread/mutex.hpp>
+
+#include "mongo/db/storage/kv/kv_catalog.h"
+#include "mongo/db/storage/record_store.h"
+#include "mongo/db/storage/storage_engine.h"
+
+namespace mongo {
+
+    class KVCatalog;
+    class KVEngine;
+    class KVDatabaseCatalogEntry;
+
+    class KVStorageEngine : public StorageEngine {
+    public:
+        /**
+         * @param engine - owneership passes to me
+         */
+        KVStorageEngine( KVEngine* engine );
+        virtual ~KVStorageEngine();
+
+        virtual void finishInit();
+
+        virtual RecoveryUnit* newRecoveryUnit( OperationContext* opCtx );
+
+        virtual void listDatabases( std::vector<std::string>* out ) const;
+
+        virtual DatabaseCatalogEntry* getDatabaseCatalogEntry( OperationContext* opCtx,
+                                                               const StringData& db );
+
+        virtual bool supportsDocLocking() const { return true; }
+
+        virtual Status closeDatabase( OperationContext* txn, const StringData& db );
+
+        virtual Status dropDatabase( OperationContext* txn, const StringData& db );
+
+        virtual int flushAllFiles( bool sync );
+
+        virtual Status repairDatabase( OperationContext* txn,
+                                       const std::string& dbName,
+                                       bool preserveClonedFilesOnFailure = false,
+                                       bool backupOriginalFiles = false );
+
+        virtual void cleanShutdown(OperationContext* txn);
+
+        // ------ kv ------
+
+        KVEngine* getEngine() { return _engine.get(); }
+        const KVEngine* getEngine() const { return _engine.get(); }
+
+        KVCatalog* getCatalog() { return _catalog.get(); }
+        const KVCatalog* getCatalog() const { return _catalog.get(); }
+
+    private:
+        boost::scoped_ptr<KVEngine> _engine;
+        bool _initialized;
+
+        boost::scoped_ptr<RecordStore> _catalogRecordStore;
+        boost::scoped_ptr<KVCatalog> _catalog;
+
+        typedef std::map<std::string,KVDatabaseCatalogEntry*> DBMap;
+        DBMap _dbs;
+        mutable boost::mutex _dbsLock;
+    };
+
+}
diff --git a/src/mongo/db/storage/sorted_data_interface_test_harness.cpp b/src/mongo/db/storage/sorted_data_interface_test_harness.cpp
index b493d1bc728..e4de9eb5e78 100644
--- a/src/mongo/db/storage/sorted_data_interface_test_harness.cpp
+++ b/src/mongo/db/storage/sorted_data_interface_test_harness.cpp
@@ -493,40 +493,41 @@ namespace mongo {
             }
         }
 
-        scoped_ptr<SortedDataInterface::Cursor> cursor( sorted->newCursor( NULL, 1 ) );
+        scoped_ptr<OperationContext> opCtx(harnessHelper->newOperationContext());
+        scoped_ptr<SortedDataInterface::Cursor> cursor( sorted->newCursor( opCtx.get(), 1 ) );
         ASSERT( !cursor->locate( BSON( "" << 5 ), DiskLoc(0,0) ) );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 5 ), cursor->getKey() );
         cursor->advance();
         ASSERT_EQUALS( BSON( "" << 7 ), cursor->getKey() );
 
-        cursor.reset( sorted->newCursor( NULL, -1 ) );
+        cursor.reset( sorted->newCursor( opCtx.get(), -1 ) );
         ASSERT( !cursor->locate( BSON( "" << 5 ), DiskLoc(0,0) ) );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 4 ), cursor->getKey() );
 
-        cursor.reset( sorted->newCursor( NULL, -1 ) );
+        cursor.reset( sorted->newCursor( opCtx.get(), -1 ) );
         ASSERT( !cursor->locate( BSON( "" << 5 ), maxDiskLoc ) );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 5 ), cursor->getKey() );
         cursor->advance();
         ASSERT_EQUALS( BSON( "" << 4 ), cursor->getKey() );
 
-        cursor.reset( sorted->newCursor( NULL, -1 ) );
+        cursor.reset( sorted->newCursor( opCtx.get(), -1 ) );
         ASSERT( !cursor->locate( BSON( "" << 5 ), minDiskLoc ) );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 4 ), cursor->getKey() );
         cursor->advance();
         ASSERT_EQUALS( BSON( "" << 3 ), cursor->getKey() );
 
-        cursor.reset( sorted->newCursor( NULL, -1 ) );
+        cursor.reset( sorted->newCursor( opCtx.get(), -1 ) );
         cursor->locate( BSON( "" << 6 ), maxDiskLoc );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 5 ), cursor->getKey() );
         cursor->advance();
         ASSERT_EQUALS( BSON( "" << 4 ), cursor->getKey() );
 
-        cursor.reset( sorted->newCursor( NULL, -1 ) );
+        cursor.reset( sorted->newCursor( opCtx.get(), -1 ) );
         cursor->locate( BSON( "" << 500 ), maxDiskLoc );
         ASSERT( !cursor->isEOF()  );
         ASSERT_EQUALS( BSON( "" << 9 ), cursor->getKey() );