SERVER-107881: BACKPORT-25613: [v8.2] Always perform LPP validation in ClusterAggregate::runAggregate(...) (#39224)

GitOrigin-RevId: 19a039191df510a699ca18fb6dc2c0d7658d6174

jstests/with_mongot/e2e/hybridSearch/rank_fusion_failure_assertions.js

@@ -75,6 +75,12 @@ assert.commandFailedWithCode(
     }]),
     10170100);
 
+// Check that LPP validation catches that $rankFusion is not the first stage. This test may help
+// expose discrepancies across sharding topologies.
+assert.commandFailedWithCode(
+    runPipeline([{$limit: 10}, {$rankFusion: {input: {pipelines: {nested: [{$sort: {_id: 1}}]}}}}]),
+    10170100);
+
 // Check that $score is not an allowed stage in $rankFusion.
 assert.commandFailedWithCode(
     runPipeline([{

src/mongo/s/query/planner/cluster_aggregate.cpp

@@ -567,6 +567,24 @@ Status _parseQueryStatsAndReturnEmptyResult(
     boost::optional<ExplainOptions::Verbosity> verbosity,
     BSONObjBuilder* result) {
 
+    // By forcing the validation checks to be done explicitly, instead of indirectly via a callback
+    // function (runAggregateImpl) in runAggregate(...) that gets passed to
+    // router.routeWithRoutingContext(...), this code ensures that the router always performs
+    // lite parsed pipeline validation. This is critical for $rankFusion and $scoreFusion because
+    // both stages are fully desugared by the time they are sent to the shards (meaning they don't
+    // contain $rankFusion/$scoreFusion DocumentSources) so the lite parsed pipeline validation
+    // performed on the shards will NOT catch any validation errors. Without this explicit check,
+    // it's possible for the router.routeWithRoutingContext(...) to error early before the callback
+    // function, runAggregateImpl(...), is executed. The catch clause catches the error and
+    // execution continues to pipeline parsing and so on. Thus, lite parsed pipeline validation
+    // never happens on the sharding node for single shard/sharded cluster with unsharded collection
+    // topologies.
+    try {
+        performValidationChecks(opCtx, request, liteParsedPipeline);
+    } catch (const DBException& ex) {
+        return ex.toStatus();
+    }
+
     const auto hasChangeStream = liteParsedPipeline.hasChangeStream();
     const auto shouldDoFLERewrite = request.getEncryptionInformation().has_value();
     const auto requiresCollationForParsingUnshardedAggregate =