SERVER-57000 Fix handling of correlated pipeline with facet

jstests/aggregation/sources/lookup/lookup_non_correlated_prefix.js

@@ -77,4 +77,22 @@ cursor.toArray().forEach(user => {
     assert.eq(1, joinedDocs.length);
     assert.eq(user['_id'], joinedDocs[0].owner);
 });
+
+// SERVER-57000: Test handling of lack of correlation (addFields with empty set of columns)
+assert.doesNotThrow(() => testColl.aggregate([
+    {
+        $lookup: {
+            as: 'items_check',
+            from: joinColl.getName(),
+            pipeline: [
+                {$addFields: {}},
+                {
+                    $facet: {
+                        all: [{$match: {}}],
+                    },
+                },
+            ],
+        },
+    },
+]));
 })();

src/mongo/db/pipeline/document_source_sequential_document_cache.cpp

@@ -120,12 +120,15 @@ Pipeline::SourceContainer::iterator DocumentSourceSequentialDocumentCache::doOpt
     // Iterate through the pipeline stages until we find one which cannot be cached.
     // A stage cannot be cached if it either: 1. depends on a variable defined in this scope, or
     // 2. generates random numbers.
+    DocumentSource* lastPtr = nullptr;
     for (; prefixSplit != container->end(); ++prefixSplit) {
         (*prefixSplit)->getDependencies(&deps);
 
         if (deps.hasVariableReferenceTo(varIDs) || deps.needRandomGenerator) {
             break;
         }
+
+        lastPtr = prefixSplit->get();
     }
 
     // The 'prefixSplit' iterator is now pointing to the first stage of the correlated suffix. If
@@ -138,6 +141,8 @@ Pipeline::SourceContainer::iterator DocumentSourceSequentialDocumentCache::doOpt
 
     // If the cache has been populated and is serving results, remove the non-correlated prefix.
     if (_cache->isServing()) {
+        // Need to dispose last stage to be removed.
+        lastPtr->dispose();
         container->erase(container->begin(), prefixSplit);
     }
 

src/mongo/db/pipeline/document_source_single_document_transformation.cpp

@@ -58,6 +58,10 @@ const char* DocumentSourceSingleDocumentTransformation::getSourceName() const {
 }
 
 DocumentSource::GetNextResult DocumentSourceSingleDocumentTransformation::doGetNext() {
+    if (!_parsedTransform) {
+        return DocumentSource::GetNextResult::makeEOF();
+    }
+
     // Get the next input document.
     auto input = pSource->getNext();
     if (!input.isAdvanced()) {
@@ -69,7 +73,9 @@ DocumentSource::GetNextResult DocumentSourceSingleDocumentTransformation::doGetN
 }
 
 intrusive_ptr<DocumentSource> DocumentSourceSingleDocumentTransformation::optimize() {
-    _parsedTransform->optimize();
+    if (_parsedTransform) {
+        _parsedTransform->optimize();
+    }
     return this;
 }