SERVER-35893: Update $out to accept new syntax

17 files changed, 546 insertions, 53 deletions

diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index fbdc846e838..813a9bd05a1 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -211,6 +211,7 @@ env.CppUnitTest(
         'document_source_match_test.cpp',
         'document_source_merge_cursors_test.cpp',
         'document_source_mock_test.cpp',
+        'document_source_out_test.cpp',
         'document_source_project_test.cpp',
         'document_source_redact_test.cpp',
         'document_source_replace_root_test.cpp',
@@ -486,7 +487,9 @@ env.CppUnitTest(
 env.Library(
     target='document_sources_idl',
     source=[
-        env.Idlc('document_sources.idl')[0],
+        env.Idlc('document_source_change_stream.idl')[0],
+        env.Idlc('document_source_list_sessions.idl')[0],
+        env.Idlc('document_source_out.idl')[0],
         'resume_token.cpp',
     ],
     LIBDEPS=[
diff --git a/src/mongo/db/pipeline/document_source_change_stream.h b/src/mongo/db/pipeline/document_source_change_stream.h
index 73b09453c3d..802a717294e 100644
--- a/src/mongo/db/pipeline/document_source_change_stream.h
+++ b/src/mongo/db/pipeline/document_source_change_stream.h
@@ -29,9 +29,9 @@
 #pragma once
 
 #include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/document_source_change_stream_gen.h"
 #include "mongo/db/pipeline/document_source_match.h"
 #include "mongo/db/pipeline/document_source_single_document_transformation.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
 #include "mongo/db/pipeline/field_path.h"
 #include "mongo/db/pipeline/resume_token.h"
 
diff --git a/src/mongo/db/pipeline/document_sources.idl b/src/mongo/db/pipeline/document_source_change_stream.idl
index 0180e47db36..51df24e8cda 100644
--- a/src/mongo/db/pipeline/document_sources.idl
+++ b/src/mongo/db/pipeline/document_source_change_stream.idl
@@ -1,4 +1,4 @@
-# Copyright (C) 2017 MongoDB Inc.
+# Copyright (C) 2018 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,
@@ -24,7 +24,7 @@
 # exception statement from all source files in the program, then also delete
 # it in the license file.
 
-# Document source pipeline stage IDL file
+# Document source change stream stage IDL file
 
 global:
    cpp_namespace: "mongo"
@@ -105,21 +105,3 @@ structs:
               default: false
               description: A flag indicating whether the stream should report all changes that occur
                            on the deployment, aside from those on internal databases or collections.
-
-  ListSessionsUser:
-    description: "A struct representing a $listSessions/$listLocalSessions User"
-    strict: true
-    fields:
-      user: string
-      db: string
-
-  ListSessionsSpec:
-    description: "$listSessions and $listLocalSessions pipeline spec"
-    strict: true
-    fields:
-      allUsers:
-        type: bool
-        default: false
-      users:
-        type: array<ListSessionsUser>
-        optional: true
diff --git a/src/mongo/db/pipeline/document_source_change_stream_transform.h b/src/mongo/db/pipeline/document_source_change_stream_transform.h
index 1825152f887..543a03aedd5 100644
--- a/src/mongo/db/pipeline/document_source_change_stream_transform.h
+++ b/src/mongo/db/pipeline/document_source_change_stream_transform.h
@@ -30,8 +30,8 @@
 
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/document_source_change_stream.h"
+#include "mongo/db/pipeline/document_source_change_stream_gen.h"
 #include "mongo/db/pipeline/document_source_match.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
 #include "mongo/db/pipeline/field_path.h"
 
 namespace mongo {
diff --git a/src/mongo/db/pipeline/document_source_check_resume_token.h b/src/mongo/db/pipeline/document_source_check_resume_token.h
index 430f3c3a6ce..68f756b4519 100644
--- a/src/mongo/db/pipeline/document_source_check_resume_token.h
+++ b/src/mongo/db/pipeline/document_source_check_resume_token.h
@@ -31,8 +31,8 @@
 #include "mongo/db/pipeline/change_stream_constants.h"
 #include "mongo/db/pipeline/document_source.h"
 #include "mongo/db/pipeline/document_source_change_stream.h"
+#include "mongo/db/pipeline/document_source_change_stream_gen.h"
 #include "mongo/db/pipeline/document_source_sort.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
 #include "mongo/db/pipeline/resume_token.h"
 
 namespace mongo {
diff --git a/src/mongo/db/pipeline/document_source_list_local_cursors.cpp b/src/mongo/db/pipeline/document_source_list_local_cursors.cpp
index 7367eebe366..72e53d7b6dd 100644
--- a/src/mongo/db/pipeline/document_source_list_local_cursors.cpp
+++ b/src/mongo/db/pipeline/document_source_list_local_cursors.cpp
@@ -34,7 +34,6 @@
 #include "mongo/db/auth/user_name.h"
 #include "mongo/db/commands/test_commands_enabled.h"
 #include "mongo/db/logical_session_id_helpers.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
 
 namespace mongo {
 
diff --git a/src/mongo/db/pipeline/document_source_list_local_cursors.h b/src/mongo/db/pipeline/document_source_list_local_cursors.h
index 5adec58c981..1725cc70f44 100644
--- a/src/mongo/db/pipeline/document_source_list_local_cursors.h
+++ b/src/mongo/db/pipeline/document_source_list_local_cursors.h
@@ -34,7 +34,6 @@
 #include "mongo/bson/bsonobj.h"
 #include "mongo/db/generic_cursor.h"
 #include "mongo/db/pipeline/document_source.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
 #include "mongo/db/pipeline/lite_parsed_document_source.h"
 
 namespace mongo {
diff --git a/src/mongo/db/pipeline/document_source_list_local_sessions.cpp b/src/mongo/db/pipeline/document_source_list_local_sessions.cpp
index af4d454d208..d0735c0c91d 100644
--- a/src/mongo/db/pipeline/document_source_list_local_sessions.cpp
+++ b/src/mongo/db/pipeline/document_source_list_local_sessions.cpp
@@ -32,7 +32,7 @@
 #include "mongo/db/auth/user_name.h"
 #include "mongo/db/logical_session_id_helpers.h"
 #include "mongo/db/pipeline/document_source_list_local_sessions.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
+#include "mongo/db/pipeline/document_source_list_sessions_gen.h"
 
 namespace mongo {
 
diff --git a/src/mongo/db/pipeline/document_source_list_local_sessions.h b/src/mongo/db/pipeline/document_source_list_local_sessions.h
index f4e71597f98..c2885978cab 100644
--- a/src/mongo/db/pipeline/document_source_list_local_sessions.h
+++ b/src/mongo/db/pipeline/document_source_list_local_sessions.h
@@ -35,7 +35,7 @@
 #include "mongo/crypto/sha256_block.h"
 #include "mongo/db/logical_session_cache.h"
 #include "mongo/db/pipeline/document_source.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
+#include "mongo/db/pipeline/document_source_list_sessions_gen.h"
 #include "mongo/db/pipeline/lite_parsed_document_source.h"
 
 namespace mongo {
diff --git a/src/mongo/db/pipeline/document_source_list_sessions.cpp b/src/mongo/db/pipeline/document_source_list_sessions.cpp
index 7d538ee1f3a..2ae6021a034 100644
--- a/src/mongo/db/pipeline/document_source_list_sessions.cpp
+++ b/src/mongo/db/pipeline/document_source_list_sessions.cpp
@@ -33,7 +33,7 @@
 #include "mongo/db/matcher/expression_parser.h"
 #include "mongo/db/matcher/extensions_callback_noop.h"
 #include "mongo/db/pipeline/document_source_list_sessions.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
+#include "mongo/db/pipeline/document_source_list_sessions_gen.h"
 
 namespace mongo {
 
diff --git a/src/mongo/db/pipeline/document_source_list_sessions.idl b/src/mongo/db/pipeline/document_source_list_sessions.idl
new file mode 100644
index 00000000000..83e185d8a30
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_list_sessions.idl
@@ -0,0 +1,53 @@
+# Copyright (C) 2018 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.
+
+# Document source list sessions stage IDL file
+
+global:
+   cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+structs:
+
+  ListSessionsUser:
+    description: "A struct representing a $listSessions/$listLocalSessions User"
+    strict: true
+    fields:
+      user: string
+      db: string
+
+  ListSessionsSpec:
+    description: "$listSessions and $listLocalSessions pipeline spec"
+    strict: true
+    fields:
+      allUsers:
+        type: bool
+        default: false
+      users:
+        type: array<ListSessionsUser>
+        optional: true
diff --git a/src/mongo/db/pipeline/document_source_out.cpp b/src/mongo/db/pipeline/document_source_out.cpp
index 26d0f14f401..f4a049b074d 100644
--- a/src/mongo/db/pipeline/document_source_out.cpp
+++ b/src/mongo/db/pipeline/document_source_out.cpp
@@ -28,9 +28,9 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/db/pipeline/document_source_out.h"
-
 #include "mongo/db/ops/write_ops.h"
+#include "mongo/db/pipeline/document_source_out.h"
+#include "mongo/db/pipeline/document_source_out_gen.h"
 #include "mongo/stdx/memory.h"
 #include "mongo/util/destructor_guard.h"
 
@@ -51,12 +51,27 @@ DocumentSourceOut::~DocumentSourceOut() {
 
 std::unique_ptr<LiteParsedDocumentSourceForeignCollections> DocumentSourceOut::liteParse(
     const AggregationRequest& request, const BSONElement& spec) {
+
     uassert(ErrorCodes::TypeMismatch,
-            str::stream() << "$out stage requires a string argument, but found "
+            str::stream() << "$out stage requires a string or object argument, but found "
                           << typeName(spec.type()),
-            spec.type() == BSONType::String);
+            spec.type() == BSONType::String || spec.type() == BSONType::Object);
+
+    NamespaceString targetNss;
+    if (spec.type() == BSONType::String) {
+        targetNss = NamespaceString(request.getNamespaceString().db(), spec.valueStringData());
+    } else if (spec.type() == BSONType::Object) {
+        auto outSpec =
+            DocumentSourceOutSpec::parse(IDLParserErrorContext("$out"), spec.embeddedObject());
+
+        if (auto targetDb = outSpec.getTargetDb()) {
+            targetNss = NamespaceString(*targetDb, outSpec.getTargetCollection());
+        } else {
+            targetNss =
+                NamespaceString(request.getNamespaceString().db(), outSpec.getTargetCollection());
+        }
+    }
 
-    NamespaceString targetNss(request.getNamespaceString().db(), spec.valueStringData());
     uassert(ErrorCodes::InvalidNamespace,
             str::stream() << "Invalid $out target namespace, " << targetNss.ns(),
             targetNss.isValid());
@@ -211,37 +226,79 @@ DocumentSource::GetNextResult DocumentSourceOut::getNext() {
 }
 
 DocumentSourceOut::DocumentSourceOut(const NamespaceString& outputNs,
-                                     const intrusive_ptr<ExpressionContext>& pExpCtx)
-    : DocumentSource(pExpCtx),
+                                     const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                     WriteModeEnum mode,
+                                     bool dropTarget,
+                                     boost::optional<Document> uniqueKey)
+    : DocumentSource(expCtx),
       _done(false),
       _tempNs(""),  // Filled in during getNext().
-      _outputNs(outputNs) {}
+      _outputNs(outputNs),
+      _mode(mode),
+      _dropTarget(dropTarget),
+      _uniqueKey(uniqueKey) {}
 
 intrusive_ptr<DocumentSource> DocumentSourceOut::createFromBson(
-    BSONElement elem, const intrusive_ptr<ExpressionContext>& pExpCtx) {
-    uassert(16990,
-            str::stream() << "$out only supports a string argument, not " << typeName(elem.type()),
-            elem.type() == String);
+    BSONElement elem, const intrusive_ptr<ExpressionContext>& expCtx) {
 
     uassert(ErrorCodes::OperationNotSupportedInTransaction,
             "$out cannot be used in a transaction",
-            !pExpCtx->inMultiDocumentTransaction);
+            !expCtx->inMultiDocumentTransaction);
 
-    auto readConcernLevel = repl::ReadConcernArgs::get(pExpCtx->opCtx).getLevel();
+    auto readConcernLevel = repl::ReadConcernArgs::get(expCtx->opCtx).getLevel();
     uassert(ErrorCodes::InvalidOptions,
             "$out cannot be used with a 'majority' read concern level",
             readConcernLevel != repl::ReadConcernLevel::kMajorityReadConcern);
 
-    NamespaceString outputNs(pExpCtx->ns.db().toString() + '.' + elem.str());
-    uassert(17385, "Can't $out to special collection: " + elem.str(), !outputNs.isSpecial());
-    return new DocumentSourceOut(outputNs, pExpCtx);
+    bool dropTarget = true;
+    auto mode = WriteModeEnum::kModeInsert;
+    boost::optional<Document> uniqueKey;
+    NamespaceString outputNs;
+    if (elem.type() == BSONType::String) {
+        outputNs = NamespaceString(expCtx->ns.db().toString() + '.' + elem.str());
+    } else if (elem.type() == BSONType::Object) {
+        auto spec =
+            DocumentSourceOutSpec::parse(IDLParserErrorContext("$out"), elem.embeddedObject());
+
+        dropTarget = spec.getDropTarget();
+        mode = spec.getMode();
+        uassert(ErrorCodes::InvalidOptions,
+                "$out is currently supported only with dropTarget: true and mode: insert.",
+                dropTarget && mode == WriteModeEnum::kModeInsert);
+
+        if (auto uniqueKeyDoc = spec.getUniqueKey()) {
+            uniqueKey = Document{{uniqueKeyDoc.get()}};
+        }
+
+        // Retrieve the target database from the user command, otherwise use the namespace from the
+        // expression context.
+        if (auto targetDb = spec.getTargetDb()) {
+            outputNs = NamespaceString(*targetDb, spec.getTargetCollection());
+        } else {
+            outputNs = NamespaceString(expCtx->ns.db(), spec.getTargetCollection());
+        }
+
+    } else {
+        uasserted(16990,
+                  str::stream() << "$out only supports a string or object argument, not "
+                                << typeName(elem.type()));
+    }
+
+    uassert(17385, "Can't $out to special collection: " + outputNs.coll(), !outputNs.isSpecial());
+
+    return new DocumentSourceOut(outputNs, expCtx, mode, dropTarget, uniqueKey);
 }
 
 Value DocumentSourceOut::serialize(boost::optional<ExplainOptions::Verbosity> explain) const {
-    massert(
-        17000, "$out shouldn't have different db than input", _outputNs.db() == pExpCtx->ns.db());
-
-    return Value(DOC(getSourceName() << _outputNs.coll()));
+    MutableDocument serialized(
+        Document{{DocumentSourceOutSpec::kTargetCollectionFieldName, _outputNs.coll()},
+                 {DocumentSourceOutSpec::kDropTargetFieldName, _dropTarget},
+                 {DocumentSourceOutSpec::kTargetDbFieldName, _outputNs.db()},
+                 {DocumentSourceOutSpec::kModeFieldName, WriteMode_serializer(_mode)}});
+    if (_uniqueKey) {
+        serialized[DocumentSourceOutSpec::kUniqueKeyFieldName] = Value(_uniqueKey.get());
+    }
+    return Value(Document{{getSourceName(), serialized.freeze()}});
 }
 
 DepsTracker::State DocumentSourceOut::getDependencies(DepsTracker* deps) const {
diff --git a/src/mongo/db/pipeline/document_source_out.h b/src/mongo/db/pipeline/document_source_out.h
index 6945c674267..41ed8c0a9d7 100644
--- a/src/mongo/db/pipeline/document_source_out.h
+++ b/src/mongo/db/pipeline/document_source_out.h
@@ -29,6 +29,7 @@
 #pragma once
 
 #include "mongo/db/pipeline/document_source.h"
+#include "mongo/db/pipeline/document_source_out_gen.h"
 
 namespace mongo {
 
@@ -80,7 +81,10 @@ public:
 
 private:
     DocumentSourceOut(const NamespaceString& outputNs,
-                      const boost::intrusive_ptr<ExpressionContext>& pExpCtx);
+                      const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                      WriteModeEnum mode,
+                      bool dropTarget,
+                      boost::optional<Document> uniqueKey);
 
     /**
      * Sets '_tempNs' to a unique temporary namespace, makes sure the output collection isn't
@@ -107,6 +111,10 @@ private:
 
     NamespaceString _tempNs;          // output goes here as it is being processed.
     const NamespaceString _outputNs;  // output will go here after all data is processed.
+
+    WriteModeEnum _mode;
+    bool _dropTarget;
+    boost::optional<Document> _uniqueKey;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/document_source_out.idl b/src/mongo/db/pipeline/document_source_out.idl
new file mode 100644
index 00000000000..1cb5122b005
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_out.idl
@@ -0,0 +1,77 @@
+# Copyright (C) 2018 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.
+
+# Document source out stage IDL file
+
+global:
+    cpp_namespace: "mongo"
+
+imports:
+    - "mongo/idl/basic_types.idl"
+
+enums:
+    WriteMode:
+        description: "Possible merge mode values."
+        type: string
+        values:
+            kModeInsert: "insert"
+            kModeReplace: "replace"
+
+structs:
+    DocumentSourceOutSpec:
+        description: A document used to specify the $out stage of an aggregation pipeline.
+        strict: true
+        fields:
+            to:
+                cpp_name: targetCollection
+                type: string
+                description: Name of the target collection. 
+
+            db:
+                cpp_name: targetDb
+                type: string
+                optional: true
+                description: Name of the target database, defaults to the database of the
+                             aggregation.
+
+            mode:
+                cpp_name: mode
+                type: WriteMode
+                description: The merge mode for the output operation.
+
+            uniqueKey:
+                cpp_name: uniqueKey
+                type: object
+                optional: true
+                description: Document of fields representing the unique key.
+
+            dropTarget:
+                cpp_name: dropTarget
+                type: bool
+                default: false
+                description: If true, the 'to' collection is atomically dropped and replaced with 
+                             the results of the aggregation.
+              
\ No newline at end of file
diff --git a/src/mongo/db/pipeline/document_source_out_test.cpp b/src/mongo/db/pipeline/document_source_out_test.cpp
new file mode 100644
index 00000000000..a919173cda1
--- /dev/null
+++ b/src/mongo/db/pipeline/document_source_out_test.cpp
@@ -0,0 +1,315 @@
+/**
+ * Copyright 2018 (c) 10gen 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/platform/basic.h"
+
+#include <boost/intrusive_ptr.hpp>
+
+#include "mongo/db/pipeline/aggregation_context_fixture.h"
+#include "mongo/db/pipeline/document.h"
+#include "mongo/db/pipeline/document_source_out.h"
+#include "mongo/db/pipeline/document_value_test_util.h"
+
+namespace mongo {
+namespace {
+
+using boost::intrusive_ptr;
+
+class DocumentSourceOutTest : public AggregationContextFixture {
+public:
+    intrusive_ptr<DocumentSource> createOutStage(BSONObj spec) {
+        auto specElem = spec.firstElement();
+        return DocumentSourceOut::createFromBson(specElem, getExpCtx());
+    }
+};
+
+TEST_F(DocumentSourceOutTest, FailsToParseIncorrectType) {
+    BSONObj spec = BSON("$out" << 1);
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 16990);
+
+    spec = BSON("$out" << BSONArray());
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 16990);
+}
+
+TEST_F(DocumentSourceOutTest, AcceptsStringArgument) {
+    BSONObj spec = BSON("$out"
+                        << "some_collection");
+    auto docSource = createOutStage(spec);
+    auto outStage = dynamic_cast<DocumentSourceOut*>(docSource.get());
+    ASSERT_EQ(outStage->getOutputNs().coll(), "some_collection");
+}
+
+TEST_F(DocumentSourceOutTest, SerializeDefaultsModeInsertAndDropTargetTrue) {
+    BSONObj spec = BSON("$out"
+                        << "some_collection");
+    auto docSource = createOutStage(spec);
+    auto outStage = dynamic_cast<DocumentSourceOut*>(docSource.get());
+    auto serialized = outStage->serialize().getDocument();
+    ASSERT_EQ(serialized["$out"][DocumentSourceOutSpec::kDropTargetFieldName].getBool(), true);
+    ASSERT_EQ(serialized["$out"][DocumentSourceOutSpec::kModeFieldName].getStringData(),
+              "insert"_sd);
+
+    // Make sure we can reparse the serialized BSON.
+    auto reparsedDocSource = createOutStage(serialized.toBson());
+    auto reparsedOut = dynamic_cast<DocumentSourceOut*>(reparsedDocSource.get());
+    auto reSerialized = reparsedOut->serialize().getDocument();
+    ASSERT_EQ(reSerialized["$out"][DocumentSourceOutSpec::kDropTargetFieldName].getBool(), true);
+    ASSERT_EQ(reSerialized["$out"][DocumentSourceOutSpec::kModeFieldName].getStringData(),
+              "insert"_sd);
+}
+
+TEST_F(DocumentSourceOutTest, SerializeUniqueKeyOnlyIfSpecified) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "target"
+                                       << "mode"
+                                       << "insert"
+                                       << "dropTarget"
+                                       << true
+                                       << "uniqueKey"
+                                       << BSON("_id" << 1 << "shardKey" << 1)));
+    auto docSource = createOutStage(spec);
+    auto outStage = dynamic_cast<DocumentSourceOut*>(docSource.get());
+    auto serialized = outStage->serialize().getDocument();
+    ASSERT_EQ(serialized["$out"][DocumentSourceOutSpec::kDropTargetFieldName].getBool(), true);
+    ASSERT_EQ(serialized["$out"][DocumentSourceOutSpec::kModeFieldName].getStringData(),
+              "insert"_sd);
+    ASSERT_DOCUMENT_EQ(serialized["$out"][DocumentSourceOutSpec::kUniqueKeyFieldName].getDocument(),
+                       (Document{{"_id", 1}, {"shardKey", 1}}));
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfToIsNotString) {
+    BSONObj spec = BSON("$out" << BSONObj());
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 40414);
+
+    spec = BSON("$out" << BSON("to" << 1));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to" << BSON("a" << 1)));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfToIsNotAValidUserCollection) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "$test"
+                                       << "mode"
+                                       << "insert"
+                                       << "dropTarget"
+                                       << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 17385);
+
+    spec = BSON("$out" << BSON("to"
+                               << "system.views"
+                               << "mode"
+                               << "insert"
+                               << "dropTarget"
+                               << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 17385);
+
+    spec = BSON("$out" << BSON("to"
+                               << ".test."
+                               << "mode"
+                               << "insert"
+                               << "dropTarget"
+                               << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::InvalidNamespace);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfDropTargetIsNotBoolean) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "insert"
+                                       << "dropTarget"
+                                       << "invalid"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "dropTarget"
+                               << BSONArray()));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "dropTarget"
+                               << 1));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfDbIsNotString) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "insert"
+                                       << "db"
+                                       << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "db"
+                               << BSONArray()));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "db"
+                               << BSON(""
+                                       << "test")));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfDbIsNotAValidDatabaseName) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "insert"
+                                       << "dropTarget"
+                                       << true
+                                       << "db"
+                                       << "$invalid"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, 17385);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "dropTarget"
+                               << true
+                               << "db"
+                               << ".test"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::InvalidNamespace);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfModeIsNotString) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << BSONArray()));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << BSON(""
+                                       << "insert")));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfModeIsUnsupportedString) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "not_insert"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::BadValue);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "merge"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::BadValue);
+}
+
+TEST_F(DocumentSourceOutTest, FailsToParseIfUniqueKeyIsNotAnObject) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "insert"
+                                       << "uniqueKey"
+                                       << 1));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "uniqueKey"
+                               << BSONArray()));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+
+    spec = BSON("$out" << BSON("to"
+                               << "test"
+                               << "mode"
+                               << "insert"
+                               << "uniqueKey"
+                               << "_id"));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::TypeMismatch);
+}
+
+TEST_F(DocumentSourceOutTest, CorrectlyUsesTargetDbIfSpecified) {
+    const auto targetDb = "someOtherDb"_sd;
+    const auto targetColl = "test"_sd;
+    BSONObj spec = BSON("$out" << BSON("to" << targetColl << "mode"
+                                            << "insert"
+                                            << "dropTarget"
+                                            << true
+                                            << "db"
+                                            << targetDb));
+
+    auto docSource = createOutStage(spec);
+    auto outStage = dynamic_cast<DocumentSourceOut*>(docSource.get());
+    ASSERT_EQ(outStage->getOutputNs().db(), targetDb);
+    ASSERT_EQ(outStage->getOutputNs().coll(), targetColl);
+}
+
+TEST_F(DocumentSourceOutTest, DropTargetMustBeTrue) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "insert"
+                                       << "dropTarget"
+                                       << false));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::InvalidOptions);
+}
+
+TEST_F(DocumentSourceOutTest, ModeMustBeInsert) {
+    BSONObj spec = BSON("$out" << BSON("to"
+                                       << "test"
+                                       << "mode"
+                                       << "replace"
+                                       << "dropTarget"
+                                       << true));
+    ASSERT_THROWS_CODE(createOutStage(spec), AssertionException, ErrorCodes::InvalidOptions);
+}
+
+}  // namespace
+}  // namespace mongo
diff --git a/src/mongo/db/pipeline/pipeline_test.cpp b/src/mongo/db/pipeline/pipeline_test.cpp
index 6df9e9fb41f..6f49c17ca97 100644
--- a/src/mongo/db/pipeline/pipeline_test.cpp
+++ b/src/mongo/db/pipeline/pipeline_test.cpp
@@ -2119,7 +2119,7 @@ class Out : public needsPrimaryShardMergerBase {
         return "[]";
     }
     string mergePipeJson() {
-        return "[{$out: 'outColl'}]";
+        return "[{$out: {to: 'outColl', dropTarget: true, db: 'a', mode: 'insert'}}]";
     }
 };
 
diff --git a/src/mongo/db/pipeline/resume_token.cpp b/src/mongo/db/pipeline/resume_token.cpp
index 32b688c179a..77643c51f34 100644
--- a/src/mongo/db/pipeline/resume_token.cpp
+++ b/src/mongo/db/pipeline/resume_token.cpp
@@ -35,7 +35,7 @@
 
 #include "mongo/bson/bsonmisc.h"
 #include "mongo/bson/bsonobjbuilder.h"
-#include "mongo/db/pipeline/document_sources_gen.h"
+#include "mongo/db/pipeline/document_source_change_stream_gen.h"
 #include "mongo/db/pipeline/value_comparator.h"
 #include "mongo/db/storage/key_string.h"
 #include "mongo/util/hex.h"