SERVER-11611: re-write user insert paths to not use DataFileMgr, and use Collection directly

7 files changed, 166 insertions, 50 deletions

diff --git a/src/mongo/db/commands/write_commands/batch_executor.cpp b/src/mongo/db/commands/write_commands/batch_executor.cpp
index 404c25977c5..c5e4811d8db 100644
--- a/src/mongo/db/commands/write_commands/batch_executor.cpp
+++ b/src/mongo/db/commands/write_commands/batch_executor.cpp
@@ -42,6 +42,7 @@
 #include "mongo/db/repl/replication_server_status.h"
 #include "mongo/db/stats/counters.h"
 #include "mongo/db/write_concern.h"
+#include "mongo/db/repl/oplog.h"
 #include "mongo/s/collection_metadata.h"
 #include "mongo/s/d_logic.h"
 #include "mongo/s/shard_key_pattern.h"
@@ -273,7 +274,7 @@ namespace mongo {
                                          storageGlobalParams.dbpath, // TODO: better constructor?
                                          false /* don't check version here */);
 
-                    opSuccess = doWrite( ns, itemRef, &childOp, stats, upsertedID, error );
+                    opSuccess = doWrite( ns, ctx, itemRef, &childOp, stats, upsertedID, error );
                 }
                 childOp.done();
                 //itemTimeMicros = childOp.totalTimeMicros();
@@ -339,6 +340,7 @@ namespace mongo {
     }
 
     bool WriteBatchExecutor::doWrite( const string& ns,
+                                      Client::Context& ctx,
                                       const BatchItemRef& itemRef,
                                       CurOp* currentOp,
                                       WriteStats* stats,
@@ -398,6 +400,7 @@ namespace mongo {
 
             // Insert
             return doInsert( ns,
+                             ctx,
                              request.getInsertRequest()->getDocumentsAt( index ),
                              currentOp,
                              stats,
@@ -409,6 +412,7 @@ namespace mongo {
 
             // Update
             return doUpdate( ns,
+                             ctx,
                              *request.getUpdateRequest()->getUpdatesAt( index ),
                              currentOp,
                              stats,
@@ -421,6 +425,7 @@ namespace mongo {
 
             // Delete
             return doDelete( ns,
+                             ctx,
                              *request.getDeleteRequest()->getDeletesAt( index ),
                              currentOp,
                              stats,
@@ -429,6 +434,7 @@ namespace mongo {
     }
 
     bool WriteBatchExecutor::doInsert( const string& ns,
+                                       Client::Context& ctx,
                                        const BSONObj& insertOp,
                                        CurOp* currentOp,
                                        WriteStats* stats,
@@ -439,16 +445,74 @@ namespace mongo {
 
         opDebug.op = dbInsert;
 
+        StringData collectionName = nsToCollectionSubstring( ns );
+
+        if ( collectionName == "system.indexes" ) {
+            try {
+                const BSONElement& e = insertOp["ns"];
+                if ( e.type() != String ) {
+                    error->setErrCode( ErrorCodes::BadValue );
+                    error->setErrMessage( "tried to create an index without specifying namespace" );
+                    return false;
+                }
+
+                string targetNS = e.String();
+
+                Collection* collection = ctx.db()->getCollection( targetNS );
+                if ( !collection ) {
+                    // implicitly create
+                    collection = ctx.db()->createCollection( targetNS );
+                    if ( !collection ) {
+                        error->setErrMessage( "could not create collection" );
+                        error->setErrCode( ErrorCodes::InternalError );
+                        return false;
+                    }
+                }
+                Status status = collection->getIndexCatalog()->createIndex( insertOp, true );
+                if ( status.code() == ErrorCodes::IndexAlreadyExists )
+                    return true;
+                if ( !status.isOK() ) {
+                    error->setErrMessage( status.toString() );
+                    error->setErrCode( status.code() );
+                    return false;
+                }
+                logOp( "i", ns.c_str(), insertOp );
+                _le->nObjects = 1; // TODO Replace after implementing LastError::recordInsert().
+                opDebug.ninserted = 1;
+                stats->numInserted++;
+                return true;
+            }
+            catch ( const UserException& ex ) {
+                opDebug.exceptionInfo = ex.getInfo();
+                toBatchedError( ex, error );
+                return false;
+            }
+        }
+
         try {
-            // TODO Should call insertWithObjMod directly instead of checkAndInsert?  Note that
-            // checkAndInsert will use mayInterrupt=false, so index builds initiated here won't
-            // be interruptible.
-            BSONObj doc = insertOp; // b/c we're const going in
-            checkAndInsert( ns.c_str(), doc );
+            Collection* collection = ctx.db()->getCollection( ns );
+            if ( !collection ) {
+                // implicitly create
+                collection = ctx.db()->createCollection( ns );
+                if ( !collection ) {
+                    error->setErrMessage( "could not create collection" );
+                    error->setErrCode( ErrorCodes::InternalError );
+                    return false;
+                }
+            }
+
+            StatusWith<DiskLoc> status = collection->insertDocument( insertOp, true );
+            logOp( "i", ns.c_str(), insertOp );
             getDur().commitIfNeeded();
+            if ( !status.isOK() ) {
+                error->setErrMessage( status.getStatus().toString() );
+                error->setErrCode( status.getStatus().code() );
+                return false;
+            }
             _le->nObjects = 1; // TODO Replace after implementing LastError::recordInsert().
             opDebug.ninserted = 1;
             stats->numInserted++;
+            return true;
         }
         catch ( const UserException& ex ) {
             opDebug.exceptionInfo = ex.getInfo();
@@ -460,6 +524,7 @@ namespace mongo {
     }
 
     bool WriteBatchExecutor::doUpdate( const string& ns,
+                                       Client::Context& ctx,
                                        const BatchedUpdateDocument& updateOp,
                                        CurOp* currentOp,
                                        WriteStats* stats,
@@ -527,6 +592,7 @@ namespace mongo {
     }
 
     bool WriteBatchExecutor::doDelete( const string& ns,
+                                       Client::Context& ctx,
                                        const BatchedDeleteDocument& deleteOp,
                                        CurOp* currentOp,
                                        WriteStats* stats,
diff --git a/src/mongo/db/commands/write_commands/batch_executor.h b/src/mongo/db/commands/write_commands/batch_executor.h
index 5c9a2bea7fc..c066e35ae6d 100644
--- a/src/mongo/db/commands/write_commands/batch_executor.h
+++ b/src/mongo/db/commands/write_commands/batch_executor.h
@@ -31,6 +31,7 @@
 #include <string>
 
 #include "mongo/base/disallow_copying.h"
+#include "mongo/db/client.h"
 #include "mongo/s/write_ops/batched_command_request.h"
 #include "mongo/s/write_ops/batched_command_response.h"
 #include "mongo/s/write_ops/batched_delete_document.h"
@@ -40,7 +41,6 @@
 namespace mongo {
 
     class BSONObjBuilder;
-    class Client;
     class CurOp;
     class OpCounters;
     class OpDebug;
@@ -102,6 +102,7 @@ namespace mongo {
         //
 
         bool doWrite( const string& ns,
+                      Client::Context& ctx,
                       const BatchItemRef& itemRef,
                       CurOp* currentOp,
                       WriteStats* stats,
@@ -109,12 +110,14 @@ namespace mongo {
                       WriteErrorDetail* error );
 
         bool doInsert( const std::string& ns,
+                       Client::Context& ctx,
                        const BSONObj& insertOp,
                        CurOp* currentOp,
                        WriteStats* stats,
                        WriteErrorDetail* error );
 
         bool doUpdate( const std::string& ns,
+                       Client::Context& ctx,
                        const BatchedUpdateDocument& updateOp,
                        CurOp* currentOp,
                        WriteStats* stats,
@@ -122,6 +125,7 @@ namespace mongo {
                        WriteErrorDetail* error );
 
         bool doDelete( const std::string& ns,
+                       Client::Context& ctx,
                        const BatchedDeleteDocument& deleteOp,
                        CurOp* currentOp,
                        WriteStats* stats,
diff --git a/src/mongo/db/commands/write_commands/write_commands.cpp b/src/mongo/db/commands/write_commands/write_commands.cpp
index c5e1c268547..a00eb4f3879 100644
--- a/src/mongo/db/commands/write_commands/write_commands.cpp
+++ b/src/mongo/db/commands/write_commands/write_commands.cpp
@@ -34,6 +34,7 @@
 #include "mongo/db/curop.h"
 #include "mongo/db/json.h"
 #include "mongo/db/lasterror.h"
+#include "mongo/db/ops/insert.h"
 #include "mongo/db/server_parameters.h"
 #include "mongo/db/stats/counters.h"
 
@@ -129,6 +130,21 @@ namespace mongo {
         if ( cc().curop() )
             cc().curop()->setNS( nss.ns() );
 
+        if ( request.getBatchType() == BatchedCommandRequest::BatchType_Insert ) {
+            // check all docs
+            BatchedInsertRequest* insertRequest = request.getInsertRequest();
+            vector<BSONObj>& docsToInsert = insertRequest->getDocuments();
+            for ( size_t i = 0; i < docsToInsert.size(); i++ ) {
+                StatusWith<BSONObj> fixed = fixDocumentForInsert( docsToInsert[i] );
+                if ( !fixed.isOK() ) {
+                    return appendCommandStatus( result, fixed.getStatus() );
+                }
+                if ( fixed.getValue().isEmpty() )
+                    continue;
+                docsToInsert[i] = fixed.getValue();
+            }
+        }
+
         BSONObj defaultWriteConcern;
         // This is really bad - it's only safe because we leak the defaults by overriding them with
         // new defaults and because we never reset to an empty default.
diff --git a/src/mongo/db/instance.cpp b/src/mongo/db/instance.cpp
index 756f812698c..5c98f815efa 100644
--- a/src/mongo/db/instance.cpp
+++ b/src/mongo/db/instance.cpp
@@ -63,6 +63,7 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/ops/count.h"
 #include "mongo/db/ops/delete.h"
+#include "mongo/db/ops/insert.h"
 #include "mongo/db/ops/update.h"
 #include "mongo/db/ops/update_lifecycle_impl.h"
 #include "mongo/db/ops/update_driver.h"
@@ -839,47 +840,44 @@ namespace mongo {
         return ok;
     }
 
-    void checkAndInsert(const char *ns, /*modifies*/BSONObj& js) {
-        uassert( 10059 , "object to insert too large", js.objsize() <= BSONObjMaxUserSize);
-        {
-            BSONObjIterator i( js );
-            while ( i.more() ) {
-                BSONElement e = i.next();
-
-                // No '$' prefixed field names allowed.
-                // NOTE: We only check top level (scanning deep would be too expensive).
-                uassert( 13511,
-                         str::stream() << "Document can't have $ prefixed field names: "
-                                       << e.fieldName(),
-                         e.fieldName()[0] != '$' );
-
-                // check no regexp for _id (SERVER-9502)
-                // also, disallow undefined and arrays
-                if (str::equals(e.fieldName(), "_id")) {
-                    uassert(16824, "can't use a regex for _id", e.type() != RegEx);
-                    uassert(17150, "can't use undefined for _id", e.type() != Undefined);
-                    uassert(17151, "can't use an array for _id", e.type() != Array);
-                }
+    void checkAndInsert(Client::Context& ctx, const char *ns, /*modifies*/BSONObj& js) {
+        if ( nsToCollectionSubstring( ns ) == "system.indexes" ) {
+            string targetNS = js["ns"].String();
+
+            Collection* collection = ctx.db()->getCollection( targetNS );
+            if ( !collection ) {
+                // implicitly create
+                collection = ctx.db()->createCollection( targetNS );
+                verify( collection );
             }
+
+            Status status = collection->getIndexCatalog()->createIndex( js, true );
+            if ( status.code() == ErrorCodes::IndexAlreadyExists )
+                return;
+            uassertStatusOK( status );
+            logOp( "i", ns, js );
+            return;
+        }
+
+        StatusWith<BSONObj> fixed = fixDocumentForInsert( js );
+        uassertStatusOK( fixed.getStatus() );
+
+        Collection* collection = ctx.db()->getCollection( ns );
+        if ( !collection ) {
+            collection = ctx.db()->createCollection( ns );
+            verify( collection );
         }
 
-        theDataFileMgr.insertWithObjMod(ns,
-                                        // May be modified in the call to add an _id field.
-                                        js,
-                                        // Only permit interrupting an (index build) insert if the
-                                        // insert comes from a socket client request rather than a
-                                        // parent operation using the client interface.  The parent
-                                        // operation might not support interrupts.
-                                        cc().curop()->parent() == NULL,
-                                        false);
+        StatusWith<DiskLoc> status = collection->insertDocument( js, true );
+        uassertStatusOK( status.getStatus() );
         logOp("i", ns, js);
     }
 
-    NOINLINE_DECL void insertMulti(bool keepGoing, const char *ns, vector<BSONObj>& objs, CurOp& op) {
+    NOINLINE_DECL void insertMulti(Client::Context& ctx, bool keepGoing, const char *ns, vector<BSONObj>& objs, CurOp& op) {
         size_t i;
         for (i=0; i<objs.size(); i++){
             try {
-                checkAndInsert(ns, objs[i]);
+                checkAndInsert(ctx, ns, objs[i]);
                 getDur().commitIfNeeded();
             } catch (const UserException&) {
                 if (!keepGoing || i == objs.size()-1){
@@ -933,9 +931,9 @@ namespace mongo {
                 
                 if (multi.size() > 1) {
                     const bool keepGoing = d.reservedField() & InsertOption_ContinueOnError;
-                    insertMulti(keepGoing, ns, multi, op);
+                    insertMulti(ctx, keepGoing, ns, multi, op);
                 } else {
-                    checkAndInsert(ns, multi[0]);
+                    checkAndInsert(ctx, ns, multi[0]);
                     globalOpCounters.incInsertInWriteLock(1);
                     op.debug().ninserted = 1;
                 }
diff --git a/src/mongo/db/instance.h b/src/mongo/db/instance.h
index 43f0674ce86..356ffd65ade 100644
--- a/src/mongo/db/instance.h
+++ b/src/mongo/db/instance.h
@@ -137,6 +137,4 @@ namespace mongo {
 
     void exitCleanly( ExitCode code );
 
-    void checkAndInsert(const char *ns, BSONObj& js);
-
 } // namespace mongo
diff --git a/src/mongo/db/ops/insert.cpp b/src/mongo/db/ops/insert.cpp
index 01ddaa4c1e9..63dc1acb44b 100644
--- a/src/mongo/db/ops/insert.cpp
+++ b/src/mongo/db/ops/insert.cpp
@@ -29,25 +29,59 @@
  */
 
 #include "mongo/db/ops/insert.h"
+#include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
 
-    BSONObj fixDocumentForInsert( const BSONObj& doc ) {
+    using namespace mongoutils;
+
+    StatusWith<BSONObj> fixDocumentForInsert( const BSONObj& doc ) {
+        if ( doc.objsize() > BSONObjMaxUserSize )
+            return StatusWith<BSONObj>( ErrorCodes::BadValue, "object to insert too large" );
+
         bool firstElementIsId = doc.firstElement().fieldNameStringData() == "_id";
         bool hasTimestampToFix = false;
         {
             BSONObjIterator i( doc );
-            for( int j = 0; i.more() && j < 2; ++j ) {
+            while ( i.more() ) {
                 BSONElement e = i.next();
-                if ( e.type() == Timestamp ) {
+
+                if ( e.type() == Timestamp && e.timestampValue() == 0 ) {
                     hasTimestampToFix = true;
                     break;
                 }
+
+                const char* fieldName = e.fieldName();
+
+                if ( fieldName[0] == '$' ) {
+                    return StatusWith<BSONObj>( ErrorCodes::BadValue,
+                                                str::stream()
+                                                << "Document can't have $ prefixed field names: "
+                                                << e.fieldName() );
+                }
+
+                // check no regexp for _id (SERVER-9502)
+                // also, disallow undefined and arrays
+                if ( str::equals( fieldName, "_id") ) {
+                    if ( e.type() == RegEx ) {
+                        return StatusWith<BSONObj>( ErrorCodes::BadValue,
+                                                    "can't use a regex for _id" );
+                    }
+                    if ( e.type() == Undefined ) {
+                        return StatusWith<BSONObj>( ErrorCodes::BadValue,
+                                                    "can't use a undefined for _id" );
+                    }
+                    if ( e.type() == Array ) {
+                        return StatusWith<BSONObj>( ErrorCodes::BadValue,
+                                                    "can't use a array for _id" );
+                    }
+                }
+
             }
         }
 
         if ( firstElementIsId && !hasTimestampToFix )
-            return BSONObj();
+            return StatusWith<BSONObj>( BSONObj() );
 
         bool hadId = firstElementIsId;
 
@@ -76,7 +110,7 @@ namespace mongo {
             if ( hadId && e.fieldNameStringData() == "_id" ) {
                 // no-op
             }
-            else if ( pos <= 1 && e.type() == Timestamp && e.timestampValue() == 0 ) {
+            else if ( e.type() == Timestamp && e.timestampValue() == 0 ) {
                 mutex::scoped_lock lk(OpTime::m);
                 b.append( e.fieldName(), OpTime::now(lk) );
             }
@@ -85,7 +119,7 @@ namespace mongo {
             }
             pos++;
         }
-        return b.obj();
+        return StatusWith<BSONObj>( b.obj() );
     }
 
 }
diff --git a/src/mongo/db/ops/insert.h b/src/mongo/db/ops/insert.h
index 56f93c46ffb..831a259ff1d 100644
--- a/src/mongo/db/ops/insert.h
+++ b/src/mongo/db/ops/insert.h
@@ -36,7 +36,7 @@ namespace mongo {
      * if doc is ok, then return is BSONObj()
      * otherwise, BSONObj is what should be inserted instead
      */
-    BSONObj fixDocumentForInsert( const BSONObj& doc );
+    StatusWith<BSONObj> fixDocumentForInsert( const BSONObj& doc );
 
 
 }