SERVER-63079 Avoid using projection parser in $setWindowFields execution

(cherry picked from commit 3688d67613ff68795b42425d9a53a24a76940b19)

7 files changed, 60 insertions, 34 deletions

diff --git a/jstests/aggregation/sources/fill/fill_and_densify.js b/jstests/aggregation/sources/fill/fill_and_densify.js
index 3ab41fd8683..1eda061ea3b 100644
--- a/jstests/aggregation/sources/fill/fill_and_densify.js
+++ b/jstests/aggregation/sources/fill/fill_and_densify.js
@@ -100,28 +100,17 @@ pipeline = [
 result = coll.aggregate(pipeline).toArray();
 
 expected = [
-    {"val": 0, "toFill": 1, "part": 1},
-    {"part": 1, "val": 1, "toFill": 1},
-    {"part": 1, "val": 2, "toFill": 1},
-    {"part": 1, "val": 3, "toFill": 1},
-    {"part": 1, "val": 4, "toFill": 1},
-    {"val": 5, "toFill": 3, "part": 1},
-    {"part": 1, "val": 6, "toFill": 3},
-    {"part": 1, "val": 7, "toFill": 3},
-    {"part": 1, "val": 8, "toFill": 3},
-    {"part": 1, "val": 9, "toFill": 3},
-    {"val": 10, "toFill": 5, "part": 1},
-    {"part": 2, "val": 0},
-    {"part": 2, "val": 1},
-    {"part": 2, "val": 2},
-    {"val": 3, "toFill": 10, "part": 2},
-    {"part": 2, "val": 4, "toFill": 10},
-    {"part": 2, "val": 5, "toFill": 10},
-    {"val": 6, "toFill": 13, "part": 2},
-    {"part": 2, "val": 7, "toFill": 13},
-    {"part": 2, "val": 8, "toFill": 13},
-    {"val": 9, "toFill": 16, "part": 2},
-    {"part": 2, "val": 10, "toFill": 16}
+    {"val": 0, "toFill": 1, "part": 1},    {"part": 1, "val": 1, "toFill": 1},
+    {"part": 1, "val": 2, "toFill": 1},    {"part": 1, "val": 3, "toFill": 1},
+    {"part": 1, "val": 4, "toFill": 1},    {"val": 5, "toFill": 3, "part": 1},
+    {"part": 1, "val": 6, "toFill": 3},    {"part": 1, "val": 7, "toFill": 3},
+    {"part": 1, "val": 8, "toFill": 3},    {"part": 1, "val": 9, "toFill": 3},
+    {"val": 10, "toFill": 5, "part": 1},   {"part": 2, "val": 0, "toFill": null},
+    {"part": 2, "val": 1, "toFill": null}, {"part": 2, "val": 2, "toFill": null},
+    {"val": 3, "toFill": 10, "part": 2},   {"part": 2, "val": 4, "toFill": 10},
+    {"part": 2, "val": 5, "toFill": 10},   {"val": 6, "toFill": 13, "part": 2},
+    {"part": 2, "val": 7, "toFill": 13},   {"part": 2, "val": 8, "toFill": 13},
+    {"val": 9, "toFill": 16, "part": 2},   {"part": 2, "val": 10, "toFill": 16}
 ];
 assertArrayEq({actual: result, expected: expected});
 
diff --git a/jstests/aggregation/sources/setWindowFields/locf.js b/jstests/aggregation/sources/setWindowFields/locf.js
index b6a7d7960b6..dcc8ddb6d4c 100644
--- a/jstests/aggregation/sources/setWindowFields/locf.js
+++ b/jstests/aggregation/sources/setWindowFields/locf.js
@@ -48,7 +48,6 @@ let result = coll.aggregate([{
                      }
                  }])
                  .toArray();
-
 let expected = [
     {_id: 0, val: null},
     {_id: 1, val: 0},
@@ -72,7 +71,7 @@ result = coll.aggregate([{
              .toArray();
 
 expected = [
-    {_id: 0, val: null},
+    {_id: 0, val: null, newVal: null},
     {_id: 1, val: 0, newVal: 0},
     {_id: 2, val: 2, newVal: 2},
     {_id: 3, val: null, newVal: 2},
@@ -110,7 +109,7 @@ expected = [
 ];
 assertArrayEq({actual: result, expected: expected});
 
-// Values stay missing if all values are missing.
+// Defaults to null even with missing values.
 collection = [
     {_id: 1},
     {_id: 2},
@@ -119,7 +118,12 @@ collection = [
 ];
 coll.drop();
 assert.commandWorked(coll.insert(collection));
-expected = collection;
+expected = [
+    {_id: 1, val: null},
+    {_id: 2, val: null},
+    {_id: 3, val: null},
+    {_id: 4, val: null},
+];
 
 result = coll.aggregate([{
                  $setWindowFields: {
@@ -139,7 +143,6 @@ collection = [
 ];
 coll.drop();
 assert.commandWorked(coll.insert(collection));
-expected = collection;
 
 result = coll.aggregate([{
                  $setWindowFields: {
diff --git a/jstests/aggregation/sources/setWindowFields/parse.js b/jstests/aggregation/sources/setWindowFields/parse.js
index 63b92e952e0..d660329fdcf 100644
--- a/jstests/aggregation/sources/setWindowFields/parse.js
+++ b/jstests/aggregation/sources/setWindowFields/parse.js
@@ -250,4 +250,13 @@ err = assert.commandFailedWithCode(
     }),
     ErrorCodes.FailedToParse);
 assert.includes(err.errmsg, 'Unrecognized window function, $summ');
+
+// Test that an empty object is a valid projected field.
+assert.commandWorked(coll.insert({}));
+assert.commandWorked(run({$setWindowFields: {output: {v: {$max: {mergeObjects: {}}}}}}));
+
+// However conflicting field paths is always an error.
+err = assert.commandFailedWithCode(
+    run({$setWindowFields: {output: {a: {$sum: 1}, 'a.b': {$sum: 1}}}}), 6307900);
+assert.includes(err.errmsg, 'specification contains two conflicting paths');
 })();
diff --git a/src/mongo/db/exec/add_fields_projection_executor.h b/src/mongo/db/exec/add_fields_projection_executor.h
index 39fe73946ab..c2c8e8481e3 100644
--- a/src/mongo/db/exec/add_fields_projection_executor.h
+++ b/src/mongo/db/exec/add_fields_projection_executor.h
@@ -57,6 +57,13 @@ public:
                               ProjectionPolicies::ComputedFieldsPolicy::kAllowComputedFields}),
           _root(new InclusionNode(_policies)) {}
 
+    AddFieldsProjectionExecutor(const boost::intrusive_ptr<ExpressionContext>& expCtx,
+                                std::unique_ptr<InclusionNode> root)
+        : ProjectionExecutor(expCtx,
+                             {ProjectionPolicies::DefaultIdPolicy::kIncludeId,
+                              ProjectionPolicies::ArrayRecursionPolicy::kRecurseNestedArrays,
+                              ProjectionPolicies::ComputedFieldsPolicy::kAllowComputedFields}),
+          _root(std::move(root)) {}
     /**
      * Creates the data needed to perform an AddFields.
      * Verifies that there are no conflicting paths in the specification.
diff --git a/src/mongo/db/pipeline/accumulator_for_window_functions.h b/src/mongo/db/pipeline/accumulator_for_window_functions.h
index 2ac3edbdca4..8541b7942c0 100644
--- a/src/mongo/db/pipeline/accumulator_for_window_functions.h
+++ b/src/mongo/db/pipeline/accumulator_for_window_functions.h
@@ -186,7 +186,7 @@ public:
     static boost::intrusive_ptr<AccumulatorState> create(ExpressionContext* expCtx);
 
 private:
-    Value _lastNonNull;
+    Value _lastNonNull{BSONNULL};
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/pipeline/accumulator_locf.cpp b/src/mongo/db/pipeline/accumulator_locf.cpp
index 9d7749e6a5a..9041ab97a86 100644
--- a/src/mongo/db/pipeline/accumulator_locf.cpp
+++ b/src/mongo/db/pipeline/accumulator_locf.cpp
@@ -64,7 +64,7 @@ Value AccumulatorLocf::getValue(bool toBeMerged) {
 }
 
 void AccumulatorLocf::reset() {
-    _lastNonNull = Value();
+    _lastNonNull = Value(BSONNULL);
     _memUsageBytes = sizeof(*this) + _lastNonNull.getApproximateSize();
 }
 
diff --git a/src/mongo/db/pipeline/document_source_set_window_fields.cpp b/src/mongo/db/pipeline/document_source_set_window_fields.cpp
index a02c693f391..64b98126a4b 100644
--- a/src/mongo/db/pipeline/document_source_set_window_fields.cpp
+++ b/src/mongo/db/pipeline/document_source_set_window_fields.cpp
@@ -30,12 +30,14 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/db/exec/add_fields_projection_executor.h"
+#include "mongo/db/field_ref_set.h"
 #include "mongo/db/matcher/expression_algo.h"
 #include "mongo/db/pipeline/document_source_add_fields.h"
 #include "mongo/db/pipeline/document_source_project.h"
 #include "mongo/db/pipeline/document_source_set_window_fields.h"
 #include "mongo/db/pipeline/document_source_set_window_fields_gen.h"
 #include "mongo/db/pipeline/document_source_sort.h"
+#include "mongo/db/pipeline/expression.h"
 #include "mongo/db/pipeline/lite_parsed_document_source.h"
 #include "mongo/db/query/query_knobs_gen.h"
 #include "mongo/db/query/sort_pattern.h"
@@ -110,8 +112,19 @@ list<intrusive_ptr<DocumentSource>> document_source_set_window_fields::createFro
         sortBy.emplace(*sortSpec, expCtx);
     }
 
+    // Verify that the computed fields are valid and do not conflict with each other.
+    FieldRefSet fieldSet;
+    std::vector<FieldRef> backingRefs;
+
     std::vector<WindowFunctionStatement> outputFields;
-    for (auto&& outputElem : spec.getOutput()) {
+    const auto& output = spec.getOutput();
+    backingRefs.reserve(output.nFields());
+    for (auto&& outputElem : output) {
+        backingRefs.push_back(FieldRef(outputElem.fieldNameStringData()));
+        const FieldRef* conflict;
+        uassert(6307900,
+                "$setWindowFields 'output' specification contains two conflicting paths",
+                fieldSet.insert(&backingRefs.back(), &conflict));
         outputFields.push_back(WindowFunctionStatement::parse(outputElem, sortBy, expCtx.get()));
     }
 
@@ -458,12 +471,15 @@ DocumentSource::GetNextResult DocumentSourceInternalSetWindowFields::doGetNext()
     }
 
     // Populate the output document with the result from each window function.
-    MutableDocument addFieldsSpec;
+    auto projSpec = std::make_unique<projection_executor::InclusionNode>(
+        ProjectionPolicies{ProjectionPolicies::DefaultIdPolicy::kIncludeId});
     for (auto&& [fieldName, function] : _executableOutputs) {
         try {
             // If we hit a uassert while evaluating expressions on user data, delete the temporary
             // table before aborting the operation.
-            addFieldsSpec.addField(fieldName, function->getNext());
+            projSpec->addExpressionForPath(
+                FieldPath(fieldName),
+                ExpressionConstant::create(pExpCtx.get(), function->getNext()));
         } catch (const DBException&) {
             _iterator.finalize();
             throw;
@@ -506,8 +522,10 @@ DocumentSource::GetNextResult DocumentSourceInternalSetWindowFields::doGetNext()
             _iterator.finalize();
             break;
     }
-    auto projExec = projection_executor::AddFieldsProjectionExecutor::create(
-        pExpCtx, addFieldsSpec.freeze().toBson());
+
+    // Avoid using the factory 'create' on the executor since we don't want to re-parse.
+    auto projExec = std::make_unique<projection_executor::AddFieldsProjectionExecutor>(
+        pExpCtx, std::move(projSpec));
 
     return projExec->applyProjection(*curDoc);
 }