SERVER-72393 Add more test cases for external data source

3 files changed, 42 insertions, 22 deletions

diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 9ee92d8fd06..50c6125223e 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -170,6 +170,15 @@ public:
                 _usedExternalDataSources.emplace_back(involvedNamespace,
                                                       externalDataSourcesIter->getDataSources());
             }
+
+            if (auto&& pipeline = _aggregationRequest.getPipeline(); !pipeline.empty()) {
+                // An external data source does not support writes and thus cannot be used as a
+                // target for $merge / $out stages.
+                auto&& lastStage = pipeline.back();
+                uassert(7239302,
+                        "The external data source cannot be used for $merge or $out stage",
+                        !lastStage.hasField("$out"_sd) && !lastStage.hasField("$merge"_sd));
+            }
         }
 
     private:
diff --git a/src/mongo/db/storage/named_pipe_posix.cpp b/src/mongo/db/storage/named_pipe_posix.cpp
index 9e2fb785540..fbc1c0c4d40 100644
--- a/src/mongo/db/storage/named_pipe_posix.cpp
+++ b/src/mongo/db/storage/named_pipe_posix.cpp
@@ -42,12 +42,11 @@
 #include "mongo/logv2/log.h"
 #include "mongo/stdx/thread.h"
 
-#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
-
-
 namespace mongo {
 using namespace fmt::literals;
 
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest
+
 namespace {
 // Removes the named pipe and logs an info message if there's an error. The info message should be
 // fine since this is a test-only implementation.
@@ -90,6 +89,9 @@ int NamedPipeOutput::write(const char* data, int size) {
     uassert(7005011, "Output must have been opened before writing", _ofs.is_open());
     _ofs.write(data, size);
     if (_ofs.fail()) {
+        uasserted(7239300,
+                  "Failed to write to a named pipe, error: {}"_format(
+                      getErrorMessage("write", _pipeAbsolutePath)));
         return -1;
     }
     return size;
@@ -101,6 +103,9 @@ void NamedPipeOutput::close() {
     }
 }
 
+#undef MONGO_LOGV2_DEFAULT_COMPONENT
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
+
 NamedPipeInput::NamedPipeInput(const std::string& pipeRelativePath)
     : _pipeAbsolutePath((externalPipeDir == "" ? kDefaultPipePath : externalPipeDir) +
                         pipeRelativePath),
diff --git a/src/mongo/db/storage/named_pipe_windows.cpp b/src/mongo/db/storage/named_pipe_windows.cpp
index 045090d5697..3c9a0b4ec4d 100644
--- a/src/mongo/db/storage/named_pipe_windows.cpp
+++ b/src/mongo/db/storage/named_pipe_windows.cpp
@@ -39,12 +39,11 @@
 #include "mongo/stdx/thread.h"
 #include "mongo/util/errno_util.h"
 
-#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
-
-
 namespace mongo {
 using namespace fmt::literals;
 
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kTest
+
 // On Windows, 'externalPipeDir' parameter is not supported and so the first argument is ignored and
 // instead, 'kDefultPipePath' is used.
 NamedPipeOutput::NamedPipeOutput(const std::string&, const std::string& pipeRelativePath)
@@ -72,12 +71,16 @@ void NamedPipeOutput::open() {
     if (_isOpen) {
         return;
     }
-    auto res = ConnectNamedPipe(_pipe, nullptr);
-    if (!res) {
-        LOGV2_ERROR(7005007,
-                    "Failed to connect a named pipe",
-                    "error"_attr = getErrorMessage("ConnectNamedPipe", _pipeAbsolutePath));
-        return;
+    bool succeeded = ConnectNamedPipe(_pipe, nullptr);
+    if (!succeeded) {
+        // ERROR_PIPE_CONNECTED means that the client has arrived faster and the connection has been
+        // established. It does not count as an error.
+        if (auto ec = lastSystemError().value(); ec != ERROR_PIPE_CONNECTED) {
+            LOGV2_ERROR(7005007,
+                        "Failed to connect a named pipe",
+                        "error"_attr = getErrorMessage("ConnectNamedPipe", _pipeAbsolutePath));
+            return;
+        }
     }
     _isOpen = true;
 }
@@ -86,16 +89,16 @@ int NamedPipeOutput::write(const char* data, int size) {
     uassert(7005012, "Output must have been opened before writing", _isOpen);
     DWORD nWritten = 0;
     // Write the reply to the pipe.
-    auto res = WriteFile(_pipe,      // handle to pipe
-                         data,       // buffer to write from
-                         size,       // number of bytes to write
-                         &nWritten,  // number of bytes written
-                         nullptr);   // not overlapped I/O
-
-    if (!res || size != nWritten) {
-        LOGV2_ERROR(7005008,
-                    "Failed to write to a named pipe",
-                    "error"_attr = getErrorMessage("write", _pipeAbsolutePath));
+    bool succeeded = WriteFile(_pipe,      // handle to pipe
+                               data,       // buffer to write from
+                               size,       // number of bytes to write
+                               &nWritten,  // number of bytes written
+                               nullptr);   // not overlapped I/O
+
+    if (!succeeded || size != nWritten) {
+        uasserted(7239301,
+                  "Failed to write to a named pipe, error: {}"_format(
+                      getErrorMessage("write", _pipeAbsolutePath)));
         return -1;
     }
 
@@ -114,6 +117,9 @@ void NamedPipeOutput::close() {
     }
 }
 
+#undef MONGO_LOGV2_DEFAULT_COMPONENT
+#define MONGO_LOGV2_DEFAULT_COMPONENT ::mongo::logv2::LogComponent::kStorage
+
 NamedPipeInput::NamedPipeInput(const std::string& pipeRelativePath)
     : _pipeAbsolutePath(kDefaultPipePath + pipeRelativePath),
       _pipe(INVALID_HANDLE_VALUE),