SERVER-36723 Push $limit beneath DocumentSourceCursor into the PlanStage layer.

In addition towards working towards the general goal of
doing as much query execution as possible with a PlanStage
tree, this should have a positive performance impact for
certain agg pipelines. Previously, a pipeline with a
$project (or a $project-like stage such as $addFields)
followed by a $limit might have applied this limit only
after a full batch of data was loaded by
DocumentSourceCursor. After this change, the limit will take
effect prior to DocumentSourceCursor batching, and thus may
reduce the amount of data processed by the query.

1 files changed, 33 insertions, 5 deletions

diff --git a/jstests/aggregation/bugs/skip_limit_overflow.js b/jstests/aggregation/bugs/skip_limit_overflow.js
index 50e665b178f..597518d069b 100644
--- a/jstests/aggregation/bugs/skip_limit_overflow.js
+++ b/jstests/aggregation/bugs/skip_limit_overflow.js
@@ -14,6 +14,7 @@ load("jstests/libs/analyze_plan.js");  // For 'aggPlanHasStages' and other expla
 
 const coll = db.server39788;
 coll.drop();
+assert.commandWorked(db.runCommand({create: coll.getName()}));
 
 function testPipeline(pipeline, expectedResult, optimizedAwayStages) {
     const explainOutput = coll.explain().aggregate(pipeline);
@@ -75,27 +76,54 @@ testPipeline([{$sort: {x: -1}}, {$skip: 1}, {$limit: NumberLong("922337203685477
 // absorb the limit and skip stages.
 // Note that we cannot specify limit == 0, so we expect an error in this case.
 testPipeline([{$sort: {x: -1}}, {$skip: 0}, {$limit: NumberLong("9223372036854775807")}],
-             {"$cursor.limit": [NumberLong("9223372036854775807")]},
+             {"$sort.limit": [NumberLong("9223372036854775807")]},
              ["$skip", "$limit"]);
 
 // Case where limit + skip do not overflow. One value is MAX_LONG - 1 and another one is 1.
 // Should be able to absorb the limit stage.
 testPipeline([{$sort: {x: -1}}, {$skip: NumberLong("9223372036854775806")}, {$limit: 1}],
              {
-                 "$cursor.limit": [NumberLong("9223372036854775807")],
+                 "$sort.limit": [NumberLong("9223372036854775807")],
                  "$skip": [NumberLong("9223372036854775806")]
              },
              ["$limit"]);
 testPipeline([{$sort: {x: -1}}, {$skip: 1}, {$limit: NumberLong("9223372036854775806")}],
-             {"$cursor.limit": [NumberLong("9223372036854775807")], "$skip": [NumberLong(1)]},
+             {"$sort.limit": [NumberLong("9223372036854775807")], "$skip": [NumberLong(1)]},
              ["$limit"]);
 
+// Case where the first $limit can be pushed down, but the second overflows and thus remains in
+// place.
+testPipeline(
+    [
+        {$sort: {x: -1}},
+        {$skip: NumberLong("9223372036854775800")},
+        {$limit: 7},
+        {$skip: 10},
+        {$limit: 1}
+    ],
+    {"$sort.limit": [NumberLong("9223372036854775807")], "$limit": [NumberLong(1)]});
+
+// Case with multiple $limit and $skip stages where the second $limit ends up being the smallest.
+// There is no overflow in this case.
+testPipeline(
+    [
+        {$sort: {x: -1}},
+        {$skip: NumberLong("9223372036854775800")},
+        {$limit: 7},
+        {$skip: 3},
+        {$limit: 1}
+    ],
+    {
+        "$sort.limit": [NumberLong("9223372036854775804")],
+        "$skip": [NumberLong("9223372036854775803")]
+    });
+
 // Case where limit + skip do not overflow. Both values are < MAX_LONG.
 testPipeline([{$sort: {x: -1}}, {$skip: 674761616283}, {$limit: 35361718}],
-             {"$cursor.limit": [NumberLong(674796978001)], "$skip": [NumberLong(674761616283)]},
+             {"$sort.limit": [NumberLong(674796978001)], "$skip": [NumberLong(674761616283)]},
              ["$limit"]);
 testPipeline([{$sort: {x: -1}}, {$skip: 35361718}, {$limit: 674761616283}],
-             {"$cursor.limit": [NumberLong(674796978001)], "$skip": [NumberLong(35361718)]},
+             {"$sort.limit": [NumberLong(674796978001)], "$skip": [NumberLong(35361718)]},
              ["$limit"]);
 
 // Case where where overflow of limit + skip + skip prevents limit stage from being absorbed.